neil/silverbullet
1
0
Fork 0
Code Activity

Something works

Browse Source
This commit is contained in:
Zef Hemel
2022-03-09 12:25:42 +01:00
parent 5e34395407
commit 7ae3496749
19 changed files with 587 additions and 387 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/.vscode/settings.json
Vendored
-4
View File
@@ -1,4 +0,0 @@
{
"deno.enable": false,
"deno.unstable": true
}
server/src/server.ts
+247 -128
View File
@@ -5,12 +5,21 @@ import { readdir, readFile, stat, unlink } from "fs/promises";
import path from "path";
import stream from "stream";
import { promisify } from "util";
import { debounce } from "lodash";
import { ChangeSet, Text } from "@codemirror/state";
import { Update } from "@codemirror/collab";
import http from "http";
import { Server } from "socket.io";
import { cursorEffect } from "../../webapp/src/cursorEffect";
function safeRun(fn: () => Promise<void>) {
fn().catch((e) => {
console.error(e);
});
}
const app = express();
const server = http.createServer(app);
const io = new Server(server, {
@@ -82,7 +91,7 @@ class DiskFS {
async writePage(pageName: string, body: any): Promise<PageMeta> {
let localPath = path.join(pagesPath, pageName + ".md");
await pipeline(body, fs.createWriteStream(localPath));
console.log(`Wrote to ${localPath}`);
// console.log(`Wrote to ${localPath}`);
const s = await stat(localPath);
return {
name: pageName,
@@ -105,14 +114,245 @@ class DiskFS {
}
}
import { Socket } from "socket.io";
class Page {
text: Text;
updates: Update[];
sockets: Set<Socket>;
meta: PageMeta;
pending: ((value: any) => void)[] = [];
saveTimer: NodeJS.Timeout | undefined;
constructor(text: string, meta: PageMeta) {
this.updates = [];
this.text = Text.of(text.split("\n"));
this.meta = meta;
this.sockets = new Set<Socket>();
}
}
class RealtimeEditFS extends DiskFS {
openPages = new Map<string, Page>();
disconnectSocket(socket: Socket, pageName: string) {
let page = this.openPages.get(pageName);
if (page) {
page.sockets.delete(socket);
if (page.sockets.size === 0) {
console.log("No more sockets for", pageName, "flushing");
this.flushPageToDisk(pageName, page);
this.openPages.delete(pageName);
}
}
}
flushPageToDisk(name: string, page: Page) {
super
.writePage(name, page.text.sliceString(0))
.then((meta) => {
console.log(`Wrote page ${name} to disk`);
page.meta = meta;
})
.catch((e) => {
console.log(`Could not write ${name} to disk:`, e);
});
}
// Override
async readPage(pageName: string): Promise<{ text: string; meta: PageMeta }> {
let page = this.openPages.get(pageName);
if (page) {
console.log("Serving page from memory", pageName);
return {
text: page.text.sliceString(0),
meta: page.meta,
};
} else {
return super.readPage(pageName);
}
}
async writePage(pageName: string, body: any): Promise<PageMeta> {
let page = this.openPages.get(pageName);
if (page) {
for (let socket of page.sockets) {
socket.emit("reload", pageName);
}
this.openPages.delete(pageName);
}
return super.writePage(pageName, body);
}
constructor(rootPath: string, io: Server) {
super(rootPath);
setInterval(() => {
console.log("Currently open pages:", this.openPages.keys());
}, 10000);
// Disk watcher
fs.watch(
rootPath,
{
recursive: true,
persistent: false,
},
(eventType, filename) => {
safeRun(async () => {
if (path.extname(filename) !== ".md") {
return;
}
let localPath = path.join(rootPath, filename);
let pageName = filename.substring(0, filename.length - 3);
let s = await stat(localPath);
// console.log("Edit in", pageName);
const openPage = this.openPages.get(pageName);
if (openPage) {
if (openPage.meta.lastModified < s.mtime.getTime()) {
console.log("Page changed on disk outside of editor, reloading");
for (let socket of openPage.sockets) {
socket.emit("reload", pageName);
}
this.openPages.delete(pageName);
}
}
});
}
);
io.on("connection", (socket) => {
console.log("Connected", socket.id);
let socketOpenPages = new Set<string>();
function onCall(eventName: string, cb: (...args: any[]) => Promise<any>) {
socket.on(eventName, (reqId: number, ...args) => {
cb(...args).then((result) => {
socket.emit(`${eventName}Resp${reqId}`, result);
});
});
}
onCall("openPage", async (pageName: string) => {
let page = this.openPages.get(pageName);
if (!page) {
try {
let { text, meta } = await super.readPage(pageName);
page = new Page(text, meta);
} catch (e) {
// console.log(`Could not open ${pageName}:`, e);
// Page does not exist, let's create a new one
console.log("Creating new page", pageName);
page = new Page("", { name: pageName, lastModified: 0 });
}
this.openPages.set(pageName, page);
}
page.sockets.add(socket);
socketOpenPages.add(pageName);
console.log("Opened page", pageName);
return [page.updates.length, page.text.toJSON()];
});
socket.on("closePage", (pageName: string) => {
console.log("Closing page", pageName);
this.disconnectSocket(socket, pageName);
socketOpenPages.delete(pageName);
});
onCall(
"pushUpdates",
async (
pageName: string,
version: number,
updates: any[]
): Promise<boolean> => {
let page = this.openPages.get(pageName);
if (!page) {
console.error(
"Received updates for not open page",
pageName,
this.openPages.keys()
);
return;
}
if (version !== page.updates.length) {
console.error("Invalid version", version, page.updates.length);
return false;
} else {
console.log("Applying", updates.length, "updates");
let transformedUpdates = [];
for (let update of updates) {
let changes = ChangeSet.fromJSON(update.changes);
console.log("Got effect", update);
let transformedUpdate = {
changes,
clientID: update.clientID,
effects: update.cursors?.map((c) => {
return cursorEffect.of(c);
}),
};
page.updates.push(transformedUpdate);
transformedUpdates.push(transformedUpdate);
// TODO: save cursors locally as well
page.text = changes.apply(page.text);
}
if (page.saveTimer) {
clearTimeout(page.saveTimer);
}
page.saveTimer = setTimeout(() => {
this.flushPageToDisk(pageName, page);
}, 1000);
while (page.pending.length) {
page.pending.pop()!(transformedUpdates);
}
return true;
}
}
);
onCall(
"pullUpdates",
async (pageName: string, version: number): Promise<Update[]> => {
let page = this.openPages.get(pageName);
// console.log("Pulling updates for", pageName);
if (!page) {
console.error("Fetching updates for not open page");
return [];
}
if (version < page.updates.length) {
return page.updates.slice(version);
} else {
return new Promise((resolve) => {
page.pending.push(resolve);
});
}
}
);
socket.on("disconnect", () => {
console.log("Disconnected", socket.id);
socketOpenPages.forEach((page) => {
this.disconnectSocket(socket, page);
});
});
});
}
}
app.use("/", express.static(distDir));
let fsRouter = express.Router();
let diskFS = new DiskFS(pagesPath);
// let diskFS = new DiskFS(pagesPath);
let filesystem = new RealtimeEditFS(pagesPath, io);
// Page list
fsRouter.route("/").get(async (req, res) => {
res.json(await diskFS.listPages());
res.json(await filesystem.listPages());
});
fsRouter
@@ -121,7 +361,7 @@ fsRouter
let reqPath = req.params[0];
console.log("Getting", reqPath);
try {
let { text, meta } = await diskFS.readPage(reqPath);
let { text, meta } = await filesystem.readPage(reqPath);
res.status(200);
res.header("Last-Modified", "" + meta.lastModified);
res.header("Content-Type", "text/markdown");
@@ -135,7 +375,7 @@ fsRouter
let reqPath = req.params[0];
try {
let meta = await diskFS.writePage(reqPath, req);
let meta = await filesystem.writePage(reqPath, req);
res.status(200);
res.header("Last-Modified", "" + meta.lastModified);
res.send("OK");
@@ -148,7 +388,7 @@ fsRouter
.options(async (req, res) => {
let reqPath = req.params[0];
try {
const meta = await diskFS.getPageMeta(reqPath);
const meta = await filesystem.getPageMeta(reqPath);
res.status(200);
res.header("Last-Modified", "" + meta.lastModified);
res.header("Content-Type", "text/markdown");
@@ -161,7 +401,7 @@ fsRouter
.delete(async (req, res) => {
let reqPath = req.params[0];
try {
await diskFS.deletePage(reqPath);
await filesystem.deletePage(reqPath);
res.status(200);
res.send("OK");
} catch (e) {
@@ -189,127 +429,6 @@ app.get("/*", async (req, res) => {
res.status(200).header("Content-Type", "text/html").send(cachedIndex);
});
import { Socket } from "socket.io";
class Page {
text: Text;
updates: Update[];
sockets: Map<string, Socket>;
meta: PageMeta;
pending: ((value: any) => void)[] = [];
constructor(text: string, meta: PageMeta) {
this.updates = [];
this.text = Text.of(text.split("\n"));
this.meta = meta;
this.sockets = new Map<string, Socket>();
}
}
let openPages = new Map<string, Page>();
io.on("connection", (socket) => {
function disconnectSocket(pageName: string) {
let page = openPages.get(pageName);
if (page) {
page.sockets.delete(socket.id);
if (page.sockets.size === 0) {
console.log("No more sockets for", pageName, "flushing");
openPages.delete(pageName);
}
}
}
console.log("Connected", socket.id);
let socketOpenPages = new Set<string>();
function onCall(eventName: string, cb: (...args: any[]) => Promise<any>) {
socket.on(eventName, (reqId: number, ...args) => {
cb(...args).then((result) => {
socket.emit(`${eventName}Resp${reqId}`, result);
});
});
}
onCall("openPage", async (pageName: string) => {
let page = openPages.get(pageName);
if (!page) {
let { text, meta } = await diskFS.readPage(pageName);
page = new Page(text, meta);
openPages.set(pageName, page);
}
page.sockets.set(socket.id, socket);
socketOpenPages.add(pageName);
console.log("Sending document text");
let enhancedMeta = { ...page.meta, version: page.updates.length };
return [enhancedMeta, page.text.toJSON()];
});
socket.on("closePage", (pageName: string) => {
console.log("Closing page", pageName);
disconnectSocket(pageName);
socketOpenPages.delete(pageName);
});
onCall(
"pushUpdates",
async (
pageName: string,
version: number,
updates: any[]
): Promise<boolean> => {
let page = openPages.get(pageName);
if (!page) {
console.error("Received updates for not open page");
return;
}
if (version !== page.updates.length) {
console.error("Invalid version", version, page.updates.length);
return false;
} else {
console.log("Applying", updates.length, "updates");
for (let update of updates) {
let changes = ChangeSet.fromJSON(update.changes);
page.updates.push({ changes, clientID: update.clientID });
page.text = changes.apply(page.text);
}
while (page.pending.length) {
page.pending.pop()!(updates);
}
return true;
}
}
);
onCall(
"pullUpdates",
async (pageName: string, version: number): Promise<Update[]> => {
let page = openPages.get(pageName);
console.log("Pulling updates for", pageName);
if (!page) {
console.error("Received updates for not open page");
return;
}
console.log(`Let's get real: ${version} < ${page.updates.length}`);
if (version < page.updates.length) {
console.log("Yes");
return page.updates.slice(version);
} else {
console.log("No");
return new Promise((resolve) => {
page.pending.push(resolve);
});
}
}
);
socket.on("disconnect", () => {
console.log("Disconnected", socket.id);
socketOpenPages.forEach(disconnectSocket);
});
});
//sup
server.listen(port, () => {
console.log(`Server istening on port ${port}`);
server/yarn.lock
+10
View File
@@ -670,6 +670,11 @@
"@types/qs" "*"
"@types/serve-static" "*"
"@types/lodash@^4.14.179":
version "4.14.179"
resolved "https://registry.yarnpkg.com/@types/lodash/-/lodash-4.14.179.tgz#490ec3288088c91295780237d2497a3aa9dfb5c5"
integrity sha512-uwc1x90yCKqGcIOAT6DwOSuxnrAbpkdPsUOZtwrXb4D/6wZs+6qG7QnIawDuZWg0sWpxl+ltIKCaLoMlna678w==
"@types/mime@^1":
version "1.3.2"
resolved "https://registry.yarnpkg.com/@types/mime/-/mime-1.3.2.tgz#93e25bf9ee75fe0fd80b594bc4feb0e862111b5a"
@@ -1735,6 +1740,11 @@ lodash.uniq@^4.5.0:
resolved "https://registry.yarnpkg.com/lodash.uniq/-/lodash.uniq-4.5.0.tgz#d0225373aeb652adc1bc82e4945339a842754773"
integrity sha1-0CJTc662Uq3BvILklFM5qEJ1R3M=
lodash@^4.17.21:
version "4.17.21"
resolved "https://registry.yarnpkg.com/lodash/-/lodash-4.17.21.tgz#679591c564c3bffaae8454cf0b3df370c3d6911c"
integrity sha512-v2kDEe57lecTulaDIuNTPy3Ry4gLGJ6Z1O3vE1krgXZNrsQ+LFTGHVxVjcXPs17LhbZVGedAJv8XZ1tvj5FvSg==
lowercase-keys@^1.0.0, lowercase-keys@^1.0.1:
version "1.0.1"
resolved "https://registry.yarnpkg.com/lowercase-keys/-/lowercase-keys-1.0.1.tgz#6f9e30b47084d971a7c820ff15a6c5167b74c26f"