Veskslop-t/poc/src/core-client.ts

/*
 * 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;
    timer: ReturnType<typeof setTimeout>;
}

const REQUEST_TIMEOUT_MS = Number(process.env.VESKSLOP_REQUEST_TIMEOUT_MS ?? 30000);

export class CoreClient {
    private nextId = 1;
    private readonly pending = new Map<number, Pending>();
    private readonly listeners = new Set<CoreEventListener>();
    private readonly closeListeners = new Set<() => void>();

    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()) {
                clearTimeout(request.timer);
                request.reject(new CoreError("DISCONNECTED", "core disconnected"));
            }
            this.pending.clear();

            for (const listener of this.closeListeners) listener();
        });

        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);
        clearTimeout(request.timer);

        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) => {
            const timer = setTimeout(() => {
                if (this.pending.delete(id)) {
                    reject(new CoreError("TIMEOUT", `the core did not answer ${action}`));
                }
            }, REQUEST_TIMEOUT_MS);

            this.pending.set(id, {
                resolve: value => {
                    clearTimeout(timer);
                    resolve(value);
                },
                reject: error => {
                    clearTimeout(timer);
                    reject(error);
                },
                timer
            });

            this.socket.write(JSON.stringify({ id, action, params }) + "\n");
        });
    }

    onEvent(listener: CoreEventListener) {
        this.listeners.add(listener);
        return () => this.listeners.delete(listener);
    }

    onClose(listener: () => void) {
        this.closeListeners.add(listener);
        return () => this.closeListeners.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) {
        return this.request<{ messageId?: string }>("message.send", { channelId, content });
    }

    async reactionAdd(channelId: string, messageId: string, emoji: string) {
        return this.request("reaction.add", { channelId, messageId, emoji });
    }

    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 });
    }

    async fetchMessages(channelId: string, limit: number, options: { before?: string; after?: string } = {}) {
        return this.request("message.fetch", { channelId, limit, ...options });
    }

    async voiceGet(userId?: string, guildId?: string) {
        return this.request("voice.get", {
            ...(userId ? { userId } : {}),
            ...(guildId ? { guildId } : {})
        });
    }

    async voiceOverview(guildId: string) {
        return this.request("voice.overview", { guildId });
    }

    async voiceJoin(channelId: string, mute = true) {
        return this.request("voice.join", { channelId, mute });
    }

    async voiceLeave() {
        return this.request("voice.leave");
    }

    async voiceMove(channelId: string) {
        return this.request("voice.move", { channelId });
    }

    async voiceUsers(userIds: string[]) {
        return this.request("voice.users", { userIds });
    }

    async voiceSpeaking() {
        return this.request("voice.speaking", {});
    }

    async userStatus(userIds: string[]) {
        return this.request("user.status", { userIds });
    }

    async userVolume(userIds: string[]) {
        return this.request("user.volume", { userIds });
    }

    async userSetVolume(userId: string, volume: number) {
        return this.request("user.setVolume", { userId, volume });
    }

    async voiceDevices() {
        return this.request("voice.devices");
    }

    /** Read/turn off Discord's mic processing gates (noise suppression, AGC, echo). */
    async voiceProcessing(
        settings: {
            noiseSuppression?: boolean;
            echoCancellation?: boolean;
            automaticGainControl?: boolean;
            noiseCancellation?: boolean;
            inputVolume?: number;
        } = {}
    ) {
        return this.request("voice.processing", settings);
    }

    async voiceSetDevices(inputDeviceId?: string, outputDeviceId?: string) {
        return this.request("voice.setDevices", {
            ...(inputDeviceId ? { inputDeviceId } : {}),
            ...(outputDeviceId ? { outputDeviceId } : {})
        });
    }

    async userSearch(query: string, guildId?: string, limit?: number) {
        return this.request("user.search", {
            query,
            ...(guildId ? { guildId } : {}),
            ...(limit ? { limit } : {})
        });
    }

    async dmList(limit?: number) {
        return this.request("dm.list", {
            ...(limit ? { limit } : {})
        });
    }

    close() {
        this.socket.destroy();
    }
}