Moving everything into a single repo and build
This commit is contained in:
Zef Hemel
@@ -0,0 +1,96 @@
|
||||
import { test, expect, beforeAll, afterAll, describe } from "@jest/globals";
|
||||
|
||||
import { createServer } from "http";
|
||||
import { io as Client } from "socket.io-client";
|
||||
import { Server } from "socket.io";
|
||||
import { SocketServer } from "./api_server";
|
||||
import * as path from "path";
|
||||
import * as fs from "fs";
|
||||
|
||||
describe("Server test", () => {
|
||||
let io: Server,
|
||||
socketServer: SocketServer,
|
||||
clientSocket: any,
|
||||
reqId = 0;
|
||||
const tmpDir = path.join(__dirname, "test");
|
||||
|
||||
function wsCall(eventName: string, ...args: any[]): Promise<any> {
|
||||
return new Promise((resolve, reject) => {
|
||||
reqId++;
|
||||
clientSocket.once(`${eventName}Resp${reqId}`, (err: any, result: any) => {
|
||||
if (err) {
|
||||
reject(err);
|
||||
} else {
|
||||
resolve(result);
|
||||
}
|
||||
});
|
||||
clientSocket.emit(eventName, reqId, ...args);
|
||||
});
|
||||
}
|
||||
|
||||
beforeAll((done) => {
|
||||
const httpServer = createServer();
|
||||
io = new Server(httpServer);
|
||||
fs.mkdirSync(tmpDir, { recursive: true });
|
||||
fs.writeFileSync(`${tmpDir}/test.md`, "This is a simple test");
|
||||
httpServer.listen(async () => {
|
||||
// @ts-ignore
|
||||
const port = httpServer.address().port;
|
||||
// @ts-ignore
|
||||
clientSocket = new Client(`http://localhost:${port}`);
|
||||
socketServer = new SocketServer(tmpDir, io);
|
||||
clientSocket.on("connect", done);
|
||||
await socketServer.init();
|
||||
});
|
||||
});
|
||||
|
||||
afterAll(() => {
|
||||
io.close();
|
||||
clientSocket.close();
|
||||
socketServer.close();
|
||||
fs.rmSync(tmpDir, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
test("List pages", async () => {
|
||||
let pages = await wsCall("page.listPages");
|
||||
expect(pages.length).toBe(1);
|
||||
await wsCall("page.writePage", "test2.md", "This is another test");
|
||||
let pages2 = await wsCall("page.listPages");
|
||||
expect(pages2.length).toBe(2);
|
||||
await wsCall("page.deletePage", "test2.md");
|
||||
let pages3 = await wsCall("page.listPages");
|
||||
expect(pages3.length).toBe(1);
|
||||
});
|
||||
|
||||
test("Index operations", async () => {
|
||||
await wsCall("index.clearPageIndexForPage", "test");
|
||||
await wsCall("index.set", "test", "testkey", "value");
|
||||
expect(await wsCall("index.get", "test", "testkey")).toBe("value");
|
||||
await wsCall("index.delete", "test", "testkey");
|
||||
expect(await wsCall("index.get", "test", "testkey")).toBe(null);
|
||||
await wsCall("index.set", "test", "unrelated", 10);
|
||||
await wsCall("index.set", "test", "unrelated", 12);
|
||||
await wsCall("index.set", "test2", "complicated", {
|
||||
name: "Bla",
|
||||
age: 123123,
|
||||
});
|
||||
await wsCall("index.set", "test", "complicated", { name: "Bla", age: 100 });
|
||||
await wsCall("index.set", "test", "complicated2", {
|
||||
name: "Bla",
|
||||
age: 101,
|
||||
});
|
||||
expect(await wsCall("index.get", "test", "complicated")).toStrictEqual({
|
||||
name: "Bla",
|
||||
age: 100,
|
||||
});
|
||||
let result = await wsCall("index.scanPrefixForPage", "test", "compli");
|
||||
expect(result.length).toBe(2);
|
||||
let result2 = await wsCall("index.scanPrefixGlobal", "compli");
|
||||
expect(result2.length).toBe(3);
|
||||
await wsCall("index.deletePrefixForPage", "test", "compli");
|
||||
let result3 = await wsCall("index.scanPrefixForPage", "test", "compli");
|
||||
expect(result3.length).toBe(0);
|
||||
let result4 = await wsCall("index.scanPrefixGlobal", "compli");
|
||||
expect(result4.length).toBe(1);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,121 @@
|
||||
import { Server, Socket } from "socket.io";
|
||||
import { Page } from "./types";
|
||||
import * as path from "path";
|
||||
import { IndexApi } from "./index_api";
|
||||
import { PageApi } from "./page_api";
|
||||
import { System } from "../plugbox/runtime";
|
||||
import { createSandbox } from "../plugbox/node_sandbox";
|
||||
import { NuggetHook } from "../webapp/types";
|
||||
import corePlug from "../webapp/generated/core.plug.json";
|
||||
import pageIndexSyscalls from "./syscalls/page_index";
|
||||
|
||||
export class ClientConnection {
|
||||
openPages = new Set<string>();
|
||||
constructor(readonly sock: Socket) {}
|
||||
}
|
||||
|
||||
export interface ApiProvider {
|
||||
init(): Promise<void>;
|
||||
api(): Object;
|
||||
}
|
||||
|
||||
export class SocketServer {
|
||||
private openPages = new Map<string, Page>();
|
||||
private connectedSockets = new Set<Socket>();
|
||||
private apis = new Map<string, ApiProvider>();
|
||||
readonly rootPath: string;
|
||||
private serverSocket: Server;
|
||||
system: System<NuggetHook>;
|
||||
|
||||
constructor(rootPath: string, serverSocket: Server) {
|
||||
this.rootPath = path.resolve(rootPath);
|
||||
this.serverSocket = serverSocket;
|
||||
this.system = new System<NuggetHook>();
|
||||
}
|
||||
|
||||
async registerApi(name: string, apiProvider: ApiProvider) {
|
||||
await apiProvider.init();
|
||||
this.apis.set(name, apiProvider);
|
||||
}
|
||||
|
||||
public async init() {
|
||||
const indexApi = new IndexApi(this.rootPath);
|
||||
await this.registerApi("index", indexApi);
|
||||
this.system.registerSyscalls(pageIndexSyscalls(indexApi.db));
|
||||
await this.registerApi(
|
||||
"page",
|
||||
new PageApi(
|
||||
this.rootPath,
|
||||
this.connectedSockets,
|
||||
this.openPages,
|
||||
this.system
|
||||
)
|
||||
);
|
||||
|
||||
let plug = await this.system.load(
|
||||
"core",
|
||||
corePlug,
|
||||
createSandbox(this.system)
|
||||
);
|
||||
|
||||
this.serverSocket.on("connection", (socket) => {
|
||||
const clientConn = new ClientConnection(socket);
|
||||
|
||||
console.log("Connected", socket.id);
|
||||
this.connectedSockets.add(socket);
|
||||
|
||||
socket.on("disconnect", () => {
|
||||
console.log("Disconnected", socket.id);
|
||||
clientConn.openPages.forEach(disconnectPageSocket);
|
||||
this.connectedSockets.delete(socket);
|
||||
});
|
||||
|
||||
socket.on("page.closePage", (pageName: string) => {
|
||||
console.log("Client closed page", pageName);
|
||||
disconnectPageSocket(pageName);
|
||||
clientConn.openPages.delete(pageName);
|
||||
});
|
||||
|
||||
const onCall = (
|
||||
eventName: string,
|
||||
cb: (...args: any[]) => Promise<any>
|
||||
) => {
|
||||
socket.on(eventName, (reqId: number, ...args) => {
|
||||
cb(...args)
|
||||
.then((result) => {
|
||||
socket.emit(`${eventName}Resp${reqId}`, null, result);
|
||||
})
|
||||
.catch((err) => {
|
||||
socket.emit(`${eventName}Resp${reqId}`, err.message);
|
||||
});
|
||||
});
|
||||
};
|
||||
|
||||
const disconnectPageSocket = (pageName: string) => {
|
||||
let page = this.openPages.get(pageName);
|
||||
if (page) {
|
||||
for (let client of page.clientStates) {
|
||||
if (client.socket === socket) {
|
||||
(this.apis.get("page")! as PageApi).disconnectClient(
|
||||
client,
|
||||
page
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
};
|
||||
for (let [apiName, apiProvider] of this.apis) {
|
||||
Object.entries(apiProvider.api()).forEach(([eventName, cb]) => {
|
||||
onCall(`${apiName}.${eventName}`, (...args: any[]): any => {
|
||||
// @ts-ignore
|
||||
return cb(clientConn, ...args);
|
||||
});
|
||||
});
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
close() {
|
||||
(this.apis.get("index")! as IndexApi).db.destroy();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,91 @@
|
||||
import { readdir, readFile, stat, unlink, writeFile } from "fs/promises";
|
||||
import * as path from "path";
|
||||
import { PageMeta } from "./types";
|
||||
|
||||
export class DiskStorage {
|
||||
rootPath: string;
|
||||
|
||||
constructor(rootPath: string) {
|
||||
this.rootPath = rootPath;
|
||||
}
|
||||
|
||||
async listPages(): Promise<PageMeta[]> {
|
||||
let fileNames: PageMeta[] = [];
|
||||
|
||||
const walkPath = async (dir: string) => {
|
||||
let files = await readdir(dir);
|
||||
for (let file of files) {
|
||||
const fullPath = path.join(dir, file);
|
||||
let s = await stat(fullPath);
|
||||
if (s.isDirectory()) {
|
||||
await walkPath(fullPath);
|
||||
} else {
|
||||
if (path.extname(file) === ".md") {
|
||||
fileNames.push({
|
||||
name: fullPath.substring(
|
||||
this.rootPath.length + 1,
|
||||
fullPath.length - 3
|
||||
),
|
||||
lastModified: s.mtime.getTime(),
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
};
|
||||
await walkPath(this.rootPath);
|
||||
return fileNames;
|
||||
}
|
||||
|
||||
async readPage(pageName: string): Promise<{ text: string; meta: PageMeta }> {
|
||||
const localPath = path.join(this.rootPath, pageName + ".md");
|
||||
try {
|
||||
const s = await stat(localPath);
|
||||
return {
|
||||
text: await readFile(localPath, "utf8"),
|
||||
meta: {
|
||||
name: pageName,
|
||||
lastModified: s.mtime.getTime(),
|
||||
},
|
||||
};
|
||||
} catch (e) {
|
||||
// console.error("Error while writing page", pageName, e);
|
||||
throw Error(`Could not read page ${pageName}`);
|
||||
}
|
||||
}
|
||||
|
||||
async writePage(pageName: string, text: string): Promise<PageMeta> {
|
||||
let localPath = path.join(this.rootPath, pageName + ".md");
|
||||
try {
|
||||
await writeFile(localPath, text);
|
||||
|
||||
// console.log(`Wrote to ${localPath}`);
|
||||
const s = await stat(localPath);
|
||||
return {
|
||||
name: pageName,
|
||||
lastModified: s.mtime.getTime(),
|
||||
};
|
||||
} catch (e) {
|
||||
console.error("Error while writing page", pageName, e);
|
||||
throw Error(`Could not write ${pageName}`);
|
||||
}
|
||||
}
|
||||
|
||||
async getPageMeta(pageName: string): Promise<PageMeta> {
|
||||
let localPath = path.join(this.rootPath, pageName + ".md");
|
||||
try {
|
||||
const s = await stat(localPath);
|
||||
return {
|
||||
name: pageName,
|
||||
lastModified: s.mtime.getTime(),
|
||||
};
|
||||
} catch (e) {
|
||||
console.error("Error while getting page meta", pageName, e);
|
||||
throw Error(`Could not get meta for ${pageName}`);
|
||||
}
|
||||
}
|
||||
|
||||
async deletePage(pageName: string) {
|
||||
let localPath = path.join(this.rootPath, pageName + ".md");
|
||||
await unlink(localPath);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,90 @@
|
||||
import { ApiProvider, ClientConnection } from "./api_server";
|
||||
import knex, { Knex } from "knex";
|
||||
import path from "path";
|
||||
import pageIndexSyscalls from "./syscalls/page_index";
|
||||
|
||||
type IndexItem = {
|
||||
page: string;
|
||||
key: string;
|
||||
value: any;
|
||||
};
|
||||
|
||||
export class IndexApi implements ApiProvider {
|
||||
db: Knex;
|
||||
|
||||
constructor(rootPath: string) {
|
||||
this.db = knex({
|
||||
client: "better-sqlite3",
|
||||
connection: {
|
||||
filename: path.join(rootPath, "data.db"),
|
||||
},
|
||||
useNullAsDefault: true,
|
||||
});
|
||||
}
|
||||
|
||||
async init() {
|
||||
if (!(await this.db.schema.hasTable("page_index"))) {
|
||||
await this.db.schema.createTable("page_index", (table) => {
|
||||
table.string("page");
|
||||
table.string("key");
|
||||
table.text("value");
|
||||
table.primary(["page", "key"]);
|
||||
});
|
||||
console.log("Created table page_index");
|
||||
}
|
||||
}
|
||||
|
||||
api() {
|
||||
const syscalls = pageIndexSyscalls(this.db);
|
||||
return {
|
||||
clearPageIndexForPage: async (
|
||||
clientConn: ClientConnection,
|
||||
page: string
|
||||
) => {
|
||||
return syscalls["indexer.clearPageIndexForPage"](page);
|
||||
},
|
||||
set: async (
|
||||
clientConn: ClientConnection,
|
||||
page: string,
|
||||
key: string,
|
||||
value: any
|
||||
) => {
|
||||
return syscalls["indexer.set"](page, key, value);
|
||||
},
|
||||
get: async (clientConn: ClientConnection, page: string, key: string) => {
|
||||
return syscalls["indexer.get"](page, key);
|
||||
},
|
||||
delete: async (
|
||||
clientConn: ClientConnection,
|
||||
page: string,
|
||||
key: string
|
||||
) => {
|
||||
return syscalls["indexer.delete"](page, key);
|
||||
},
|
||||
scanPrefixForPage: async (
|
||||
clientConn: ClientConnection,
|
||||
page: string,
|
||||
prefix: string
|
||||
) => {
|
||||
return syscalls["indexer.scanPrefixForPage"](page, prefix);
|
||||
},
|
||||
scanPrefixGlobal: async (
|
||||
clientConn: ClientConnection,
|
||||
prefix: string
|
||||
) => {
|
||||
return syscalls["indexer.scanPrefixGlobal"](prefix);
|
||||
},
|
||||
deletePrefixForPage: async (
|
||||
clientConn: ClientConnection,
|
||||
page: string,
|
||||
prefix: string
|
||||
) => {
|
||||
return syscalls["indexer.deletePrefixForPage"](page, prefix);
|
||||
},
|
||||
|
||||
clearPageIndex: async (clientConn: ClientConnection) => {
|
||||
return syscalls["indexer.clearPageIndex"]();
|
||||
},
|
||||
};
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,306 @@
|
||||
import { ClientPageState, Page, PageMeta } from "./types";
|
||||
import { ChangeSet } from "@codemirror/state";
|
||||
import { Update } from "@codemirror/collab";
|
||||
import { ApiProvider, ClientConnection } from "./api_server";
|
||||
import { Socket } from "socket.io";
|
||||
import { DiskStorage } from "./disk_storage";
|
||||
import { safeRun } from "./util";
|
||||
import fs from "fs";
|
||||
import path from "path";
|
||||
import { stat } from "fs/promises";
|
||||
import { Cursor, cursorEffect } from "../webapp/cursorEffect";
|
||||
import { System } from "../plugbox/runtime";
|
||||
import { NuggetHook } from "../webapp/types";
|
||||
|
||||
export class PageApi implements ApiProvider {
|
||||
openPages: Map<string, Page>;
|
||||
pageStore: DiskStorage;
|
||||
rootPath: string;
|
||||
connectedSockets: Set<Socket>;
|
||||
private system: System<NuggetHook>;
|
||||
|
||||
constructor(
|
||||
rootPath: string,
|
||||
connectedSockets: Set<Socket>,
|
||||
openPages: Map<string, Page>,
|
||||
system: System<NuggetHook>
|
||||
) {
|
||||
this.pageStore = new DiskStorage(rootPath);
|
||||
this.rootPath = rootPath;
|
||||
this.openPages = openPages;
|
||||
this.connectedSockets = connectedSockets;
|
||||
this.system = system;
|
||||
}
|
||||
|
||||
async init(): Promise<void> {
|
||||
this.fileWatcher();
|
||||
}
|
||||
|
||||
broadcastCursors(page: Page) {
|
||||
page.clientStates.forEach((client) => {
|
||||
client.socket.emit(
|
||||
"cursorSnapshot",
|
||||
page.name,
|
||||
Object.fromEntries(page.cursors.entries())
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
flushPageToDisk(name: string, page: Page) {
|
||||
safeRun(async () => {
|
||||
let meta = await this.pageStore.writePage(name, page.text.sliceString(0));
|
||||
console.log(`Wrote page ${name} to disk`);
|
||||
page.meta = meta;
|
||||
});
|
||||
}
|
||||
|
||||
disconnectClient(client: ClientPageState, page: Page) {
|
||||
console.log("Disconnecting client");
|
||||
page.clientStates.delete(client);
|
||||
if (page.clientStates.size === 0) {
|
||||
console.log("No more clients for", page.name, "flushing");
|
||||
this.flushPageToDisk(page.name, page);
|
||||
this.openPages.delete(page.name);
|
||||
} else {
|
||||
page.cursors.delete(client.socket.id);
|
||||
this.broadcastCursors(page);
|
||||
}
|
||||
}
|
||||
|
||||
fileWatcher() {
|
||||
fs.watch(
|
||||
this.rootPath,
|
||||
{
|
||||
recursive: true,
|
||||
persistent: false,
|
||||
},
|
||||
(eventType, filename) => {
|
||||
safeRun(async () => {
|
||||
if (!filename.endsWith(".md")) {
|
||||
return;
|
||||
}
|
||||
let localPath = path.join(this.rootPath, filename);
|
||||
let pageName = filename.substring(0, filename.length - 3);
|
||||
// console.log("Edit in", pageName, eventType);
|
||||
let modifiedTime = 0;
|
||||
try {
|
||||
let s = await stat(localPath);
|
||||
modifiedTime = s.mtime.getTime();
|
||||
} catch (e) {
|
||||
// File was deleted
|
||||
console.log("Deleted", pageName);
|
||||
for (let socket of this.connectedSockets) {
|
||||
socket.emit("pageDeleted", pageName);
|
||||
}
|
||||
return;
|
||||
}
|
||||
const openPage = this.openPages.get(pageName);
|
||||
if (openPage) {
|
||||
if (openPage.meta.lastModified < modifiedTime) {
|
||||
console.log("Page changed on disk outside of editor, reloading");
|
||||
this.openPages.delete(pageName);
|
||||
const meta = {
|
||||
name: pageName,
|
||||
lastModified: modifiedTime,
|
||||
} as PageMeta;
|
||||
for (let client of openPage.clientStates) {
|
||||
client.socket.emit("pageChanged", meta);
|
||||
}
|
||||
}
|
||||
}
|
||||
if (eventType === "rename") {
|
||||
// This most likely means a new file was created, let's push new file listings to all connected sockets
|
||||
console.log(
|
||||
"New file created, broadcasting to all connected sockets",
|
||||
pageName
|
||||
);
|
||||
for (let socket of this.connectedSockets) {
|
||||
socket.emit("pageCreated", {
|
||||
name: pageName,
|
||||
lastModified: modifiedTime,
|
||||
} as PageMeta);
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
);
|
||||
}
|
||||
|
||||
api() {
|
||||
return {
|
||||
openPage: async (clientConn: ClientConnection, pageName: string) => {
|
||||
let page = this.openPages.get(pageName);
|
||||
if (!page) {
|
||||
try {
|
||||
let { text, meta } = await this.pageStore.readPage(pageName);
|
||||
page = new Page(pageName, text, meta);
|
||||
} catch (e) {
|
||||
console.log("Creating new page", pageName);
|
||||
page = new Page(pageName, "", { name: pageName, lastModified: 0 });
|
||||
}
|
||||
this.openPages.set(pageName, page);
|
||||
}
|
||||
page.clientStates.add(
|
||||
new ClientPageState(clientConn.sock, page.version)
|
||||
);
|
||||
clientConn.openPages.add(pageName);
|
||||
console.log("Opened page", pageName);
|
||||
this.broadcastCursors(page);
|
||||
return page.toJSON();
|
||||
},
|
||||
pushUpdates: async (
|
||||
clientConn: ClientConnection,
|
||||
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 false;
|
||||
}
|
||||
if (version !== page.version) {
|
||||
console.error("Invalid version", version, page.version);
|
||||
return false;
|
||||
} else {
|
||||
console.log("Applying", updates.length, "updates to", pageName);
|
||||
let transformedUpdates = [];
|
||||
let textChanged = false;
|
||||
for (let update of updates) {
|
||||
let changes = ChangeSet.fromJSON(update.changes);
|
||||
let transformedUpdate = {
|
||||
changes,
|
||||
clientID: update.clientID,
|
||||
effects: update.cursors?.map((c: Cursor) => {
|
||||
page!.cursors.set(c.userId, c);
|
||||
return cursorEffect.of(c);
|
||||
}),
|
||||
};
|
||||
page.updates.push(transformedUpdate);
|
||||
transformedUpdates.push(transformedUpdate);
|
||||
let oldText = page.text;
|
||||
page.text = changes.apply(page.text);
|
||||
if (oldText !== page.text) {
|
||||
textChanged = true;
|
||||
}
|
||||
}
|
||||
console.log(
|
||||
"New version",
|
||||
page.version,
|
||||
"Updates buffered:",
|
||||
page.updates.length
|
||||
);
|
||||
|
||||
if (textChanged) {
|
||||
// Throttle
|
||||
if (!page.saveTimer) {
|
||||
page.saveTimer = setTimeout(() => {
|
||||
if (page) {
|
||||
console.log("Indexing", pageName);
|
||||
|
||||
this.system.dispatchEvent("page:index", {
|
||||
name: pageName,
|
||||
text: page.text.sliceString(0),
|
||||
});
|
||||
this.flushPageToDisk(pageName, page);
|
||||
page.saveTimer = undefined;
|
||||
}
|
||||
}, 1000);
|
||||
}
|
||||
}
|
||||
while (page.pending.length) {
|
||||
page.pending.pop()!(transformedUpdates);
|
||||
}
|
||||
return true;
|
||||
}
|
||||
},
|
||||
|
||||
pullUpdates: async (
|
||||
clientConn: ClientConnection,
|
||||
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 [];
|
||||
}
|
||||
// TODO: Optimize this
|
||||
let oldestVersion = Infinity;
|
||||
page.clientStates.forEach((client) => {
|
||||
oldestVersion = Math.min(client.version, oldestVersion);
|
||||
if (client.socket === clientConn.sock) {
|
||||
client.version = version;
|
||||
}
|
||||
});
|
||||
page.flushUpdates(oldestVersion);
|
||||
if (version < page.version) {
|
||||
return page.updatesSince(version);
|
||||
} else {
|
||||
return new Promise((resolve) => {
|
||||
page!.pending.push(resolve);
|
||||
});
|
||||
}
|
||||
},
|
||||
|
||||
readPage: async (
|
||||
clientConn: ClientConnection,
|
||||
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 this.pageStore.readPage(pageName);
|
||||
}
|
||||
},
|
||||
|
||||
writePage: async (
|
||||
clientConn: ClientConnection,
|
||||
pageName: string,
|
||||
text: string
|
||||
) => {
|
||||
let page = this.openPages.get(pageName);
|
||||
if (page) {
|
||||
for (let client of page.clientStates) {
|
||||
client.socket.emit("reloadPage", pageName);
|
||||
}
|
||||
this.openPages.delete(pageName);
|
||||
}
|
||||
return this.pageStore.writePage(pageName, text);
|
||||
},
|
||||
|
||||
deletePage: async (clientConn: ClientConnection, pageName: string) => {
|
||||
this.openPages.delete(pageName);
|
||||
clientConn.openPages.delete(pageName);
|
||||
// Cascading of this to all connected clients will be handled by file watcher
|
||||
return this.pageStore.deletePage(pageName);
|
||||
},
|
||||
|
||||
listPages: async (clientConn: ClientConnection): Promise<PageMeta[]> => {
|
||||
return this.pageStore.listPages();
|
||||
},
|
||||
|
||||
getPageMeta: async (
|
||||
clientConn: ClientConnection,
|
||||
pageName: string
|
||||
): Promise<PageMeta> => {
|
||||
let page = this.openPages.get(pageName);
|
||||
if (page) {
|
||||
return page.meta;
|
||||
}
|
||||
return this.pageStore.getPageMeta(pageName);
|
||||
},
|
||||
};
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,46 @@
|
||||
import express from "express";
|
||||
import { readFile } from "fs/promises";
|
||||
import http from "http";
|
||||
import { Server } from "socket.io";
|
||||
import { SocketServer } from "./api_server";
|
||||
import yargs from "yargs";
|
||||
import { hideBin } from "yargs/helpers";
|
||||
|
||||
let args = yargs(hideBin(process.argv))
|
||||
.option("debug", {
|
||||
type: "boolean",
|
||||
})
|
||||
.option("port", {
|
||||
type: "number",
|
||||
default: 3000,
|
||||
})
|
||||
.parse();
|
||||
|
||||
const app = express();
|
||||
const server = http.createServer(app);
|
||||
const io = new Server(server, {
|
||||
cors: {
|
||||
methods: "GET,HEAD,PUT,OPTIONS,POST,DELETE",
|
||||
preflightContinue: true,
|
||||
},
|
||||
});
|
||||
|
||||
const port = args.port;
|
||||
const distDir = `${__dirname}/../webapp`;
|
||||
|
||||
app.use("/", express.static(distDir));
|
||||
let socketServer = new SocketServer(args._[0] as string, io);
|
||||
socketServer.init();
|
||||
|
||||
// Fallback, serve index.html
|
||||
let cachedIndex: string | undefined = undefined;
|
||||
app.get("/*", async (req, res) => {
|
||||
if (!cachedIndex) {
|
||||
cachedIndex = await readFile(`${distDir}/index.html`, "utf8");
|
||||
}
|
||||
res.status(200).header("Content-Type", "text/html").send(cachedIndex);
|
||||
});
|
||||
|
||||
server.listen(port, () => {
|
||||
console.log(`Server listening on port ${port}`);
|
||||
});
|
||||
@@ -0,0 +1,83 @@
|
||||
import { Knex } from "knex";
|
||||
|
||||
type IndexItem = {
|
||||
page: string;
|
||||
key: string;
|
||||
value: any;
|
||||
};
|
||||
|
||||
export type KV = {
|
||||
key: string;
|
||||
value: any;
|
||||
};
|
||||
|
||||
export default function (db: Knex) {
|
||||
const apiObj = {
|
||||
"indexer.clearPageIndexForPage": async (page: string) => {
|
||||
await db<IndexItem>("page_index").where({ page }).del();
|
||||
},
|
||||
"indexer.set": async (page: string, key: string, value: any) => {
|
||||
let changed = await db<IndexItem>("page_index")
|
||||
.where({ page, key })
|
||||
.update("value", JSON.stringify(value));
|
||||
if (changed === 0) {
|
||||
await db<IndexItem>("page_index").insert({
|
||||
page,
|
||||
key,
|
||||
value: JSON.stringify(value),
|
||||
});
|
||||
}
|
||||
},
|
||||
"indexer.batchSet": async (page: string, kvs: KV[]) => {
|
||||
for (let { key, value } of kvs) {
|
||||
await apiObj["indexer.set"](page, key, value);
|
||||
}
|
||||
},
|
||||
"indexer.get": async (page: string, key: string) => {
|
||||
let result = await db<IndexItem>("page_index")
|
||||
.where({ page, key })
|
||||
.select("value");
|
||||
if (result.length) {
|
||||
return JSON.parse(result[0].value);
|
||||
} else {
|
||||
return null;
|
||||
}
|
||||
},
|
||||
"indexer.delete": async (page: string, key: string) => {
|
||||
await db<IndexItem>("page_index").where({ page, key }).del();
|
||||
},
|
||||
"indexer.scanPrefixForPage": async (page: string, prefix: string) => {
|
||||
return (
|
||||
await db<IndexItem>("page_index")
|
||||
.where({ page })
|
||||
.andWhereLike("key", `${prefix}%`)
|
||||
.select("page", "key", "value")
|
||||
).map(({ page, key, value }) => ({
|
||||
page,
|
||||
key,
|
||||
value: JSON.parse(value),
|
||||
}));
|
||||
},
|
||||
"indexer.scanPrefixGlobal": async (prefix: string) => {
|
||||
return (
|
||||
await db<IndexItem>("page_index")
|
||||
.andWhereLike("key", `${prefix}%`)
|
||||
.select("page", "key", "value")
|
||||
).map(({ page, key, value }) => ({
|
||||
page,
|
||||
key,
|
||||
value: JSON.parse(value),
|
||||
}));
|
||||
},
|
||||
"indexer.deletePrefixForPage": async (page: string, prefix: string) => {
|
||||
return db<IndexItem>("page_index")
|
||||
.where({ page })
|
||||
.andWhereLike("key", `${prefix}%`)
|
||||
.del();
|
||||
},
|
||||
"indexer.clearPageIndex": async () => {
|
||||
return db<IndexItem>("page_index").del();
|
||||
},
|
||||
};
|
||||
return apiObj;
|
||||
}
|
||||
@@ -0,0 +1,62 @@
|
||||
import { Update } from "@codemirror/collab";
|
||||
import { Text } from "@codemirror/state";
|
||||
import { Socket } from "socket.io";
|
||||
import { Cursor } from "../webapp/cursorEffect";
|
||||
export class ClientPageState {
|
||||
constructor(public socket: Socket, public version: number) {}
|
||||
}
|
||||
|
||||
export type PageMeta = {
|
||||
name: string;
|
||||
lastModified: number;
|
||||
version?: number;
|
||||
};
|
||||
|
||||
export class Page {
|
||||
versionOffset = 0;
|
||||
updates: Update[] = [];
|
||||
cursors = new Map<string, Cursor>();
|
||||
clientStates = new Set<ClientPageState>();
|
||||
|
||||
pending: ((value: any) => void)[] = [];
|
||||
|
||||
text: Text;
|
||||
meta: PageMeta;
|
||||
|
||||
saveTimer: NodeJS.Timeout | undefined;
|
||||
name: string;
|
||||
|
||||
constructor(name: string, text: string, meta: PageMeta) {
|
||||
this.name = name;
|
||||
this.text = Text.of(text.split("\n"));
|
||||
this.meta = meta;
|
||||
}
|
||||
|
||||
updatesSince(version: number): Update[] {
|
||||
return this.updates.slice(version - this.versionOffset);
|
||||
}
|
||||
|
||||
get version(): number {
|
||||
return this.updates.length + this.versionOffset;
|
||||
}
|
||||
|
||||
flushUpdates(version: number) {
|
||||
if (this.versionOffset > version) {
|
||||
throw Error("This should never happen");
|
||||
}
|
||||
if (this.versionOffset === version) {
|
||||
return;
|
||||
}
|
||||
this.updates = this.updates.slice(version - this.versionOffset);
|
||||
this.versionOffset = version;
|
||||
// console.log("Flushed updates, now got", this.updates.length, "updates");
|
||||
}
|
||||
|
||||
toJSON() {
|
||||
return {
|
||||
text: this.text,
|
||||
version: this.version,
|
||||
cursors: Object.fromEntries(this.cursors.entries()),
|
||||
};
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,5 @@
|
||||
export function safeRun(fn: () => Promise<void>) {
|
||||
fn().catch((e) => {
|
||||
console.error(e);
|
||||
});
|
||||
}
|
||||
Reference in New Issue
Block a user