Wire validated core bridge, migrate daemon to it, verify live

poc/src/core-client.ts

@@ -0,0 +1,137 @@
+/*
+ * Veskslop daemon — client for the validated core bridge.
+ *
+ * Speaks newline-delimited JSON over the core socket. A request gets one
+ * response line. Events arrive as { type: "event" } lines after subscribe.
+ */
+
+import net from "node:net";
+
+export const coreSocketPath =
+    process.env.VESKSLOP_SOCKET ??
+    (process.env.XDG_RUNTIME_DIR ? `${process.env.XDG_RUNTIME_DIR}/vesktop-puppet.sock` : "/tmp/vesktop-puppet.sock");
+
+export class CoreError extends Error {
+    constructor(
+        public readonly code: string,
+        message: string
+    ) {
+        super(message);
+    }
+}
+
+export type CoreEventListener = (event: any) => void;
+
+interface Pending {
+    resolve: (value: any) => void;
+    reject: (error: Error) => void;
+}
+
+export class CoreClient {
+    private nextId = 1;
+    private readonly pending = new Map<number, Pending>();
+    private readonly listeners = new Set<CoreEventListener>();
+
+    private constructor(private readonly socket: net.Socket) {
+        socket.setEncoding("utf-8");
+
+        let buffer = "";
+        socket.on("data", (chunk: string) => {
+            buffer += chunk;
+
+            let newline: number;
+            while ((newline = buffer.indexOf("\n")) !== -1) {
+                const line = buffer.slice(0, newline);
+                buffer = buffer.slice(newline + 1);
+                if (line) this.onLine(line);
+            }
+        });
+
+        socket.on("close", () => {
+            for (const request of this.pending.values()) {
+                request.reject(new CoreError("DISCONNECTED", "core disconnected"));
+            }
+            this.pending.clear();
+        });
+
+        socket.on("error", () => socket.destroy());
+    }
+
+    static connect(path = coreSocketPath): Promise<CoreClient> {
+        return new Promise((resolve, reject) => {
+            const socket = net.connect(path);
+            const client = new CoreClient(socket);
+
+            socket.once("connect", async () => {
+                try {
+                    await client.request("events.subscribe");
+                    resolve(client);
+                } catch (error) {
+                    reject(error);
+                }
+            });
+            socket.once("error", reject);
+        });
+    }
+
+    private onLine(line: string) {
+        let message: any;
+        try {
+            message = JSON.parse(line);
+        } catch {
+            return;
+        }
+
+        if (message.type === "event") {
+            for (const listener of this.listeners) listener(message.event);
+            return;
+        }
+
+        const request = this.pending.get(message.id);
+        if (!request) return;
+        this.pending.delete(message.id);
+
+        if (message.ok) request.resolve(message.result);
+        else request.reject(new CoreError(message.error?.code ?? "UNKNOWN", message.error?.message ?? "core error"));
+    }
+
+    request<T = any>(action: string, params?: Record<string, unknown>): Promise<T> {
+        const id = this.nextId++;
+
+        return new Promise<T>((resolve, reject) => {
+            this.pending.set(id, { resolve, reject });
+            this.socket.write(JSON.stringify({ id, action, params }) + "\n");
+        });
+    }
+
+    onEvent(listener: CoreEventListener) {
+        this.listeners.add(listener);
+        return () => this.listeners.delete(listener);
+    }
+
+    /* Convenience actions */
+
+    async getState() {
+        return this.request("state.get");
+    }
+
+    async getSelfId(): Promise<string | undefined> {
+        return (await this.getState())?.user?.id;
+    }
+
+    async sendMessage(channelId: string, content: string) {
+        await this.request("message.send", { channelId, content });
+    }
+
+    async sendDm(userId: string, content: string) {
+        return this.request("dm.send", { userId, content });
+    }
+
+    async readMessages(channelId: string, limit: number) {
+        return this.request("message.read", { channelId, limit });
+    }
+
+    close() {
+        this.socket.destroy();
+    }
+}