sessBridge.ts

import * as fs from 'fs';
import * as net from 'net';
import * as path from 'path';
import { randomUUID } from 'crypto';
import { JsonLines } from '../framing';
import { object } from '../protocol';
import { BackendEvent, ClientReply } from '../backend';

interface Pending {
    resolve(value: unknown): void;
    reject(error: Error): void;
    timer: NodeJS.Timeout;
}

/** Pure R sess transport for inspection, editor requests, and adopted events. */
export class SessBridge {
    private sess?: net.Socket;
    private sockets = new Set<net.Socket>();
    private servers: net.Server[] = [];
    private pending = new Map<number, Pending>();
    private requests = new Map<
        string,
        { socket: net.Socket; id: unknown; timer: NodeJS.Timeout }
    >();
    private counter = 0;
    private closed = false;
    constructor(
        private directory: string,
        private token: string,
        private emit: (event: BackendEvent) => void,
        private interactiveEvent: (message: Record<string, unknown>) => void,
        private notification: (method: string, params: Record<string, unknown>) => void,
    ) {}

    async start(): Promise<void> {
        await this.listen('sess.sock', (socket) => this.connectSess(socket));
    }
    private async listen(name: string, connect: (socket: net.Socket) => void): Promise<void> {
        if (this.closed) {
            throw new Error('Bridge disposed during startup');
        }
        const endpoint = path.join(this.directory, name);
        const server = net.createServer((socket) => {
            this.sockets.add(socket);
            const timer = setTimeout(() => {
                if (socket !== this.sess) {
                    socket.destroy();
                }
            }, 5000);
            socket.on('close', () => {
                this.sockets.delete(socket);
                clearTimeout(timer);
            });
            socket.on('error', () => socket.destroy());
            connect(socket);
        });
        this.servers.push(server);
        await new Promise<void>((resolve, reject) => {
            server.once('error', reject);
            server.listen(endpoint, resolve);
        });
        fs.chmodSync(endpoint, 0o600);
    }
    private send(socket: net.Socket, message: unknown): void {
        if (!socket.destroyed) {
            socket.write(JSON.stringify(message) + '\n');
        }
    }
    private parse(socket: net.Socket, callback: (message: Record<string, unknown>) => void): void {
        const parser = new JsonLines((message) => {
            if (!this.closed) {
                callback(message);
            }
        });
        socket.on('data', (chunk: Buffer) => {
            try {
                parser.push(chunk);
            } catch {
                socket.destroy();
            }
        });
    }
    private connectSess(socket: net.Socket): void {
        let attached = false;
        this.parse(socket, (message) => {
            if (message.method === 'attach') {
                const params = object(message.params);
                if (
                    params.protocol_version !== 2 ||
                    params.interactive_token !== this.token ||
                    (this.sess && this.sess !== socket)
                ) {
                    socket.destroy();
                    return;
                }
                this.sess = socket;
                attached = true;
                this.emit({
                    type: 'metadata',
                    metadata: {
                        rPid: Number(params.pid),
                        rVersion: String(params.version),
                        runtimeSessionId: String(params.session_id),
                    },
                });
                return;
            }
            if (!attached || this.sess !== socket) {
                return;
            }
            if (typeof message.id === 'number' && !message.method) {
                const pending = this.pending.get(message.id);
                if (!pending) {
                    return;
                }
                clearTimeout(pending.timer);
                this.pending.delete(message.id);
                if (message.error) {
                    pending.reject(new Error(JSON.stringify(message.error)));
                } else {
                    pending.resolve(message.result);
                }
            } else if (message.method === 'interactive_event' && message.id === undefined) {
                this.interactiveEvent(object(message.params));
            } else if (
                typeof message.method === 'string' &&
                message.method &&
                message.id !== undefined
            ) {
                const id = randomUUID();
                const timer = setTimeout(() => {
                    this.requests.delete(id);
                    this.emit({ type: 'clientRequestExpired', id });
                    this.send(socket, {
                        jsonrpc: '2.0',
                        id: message.id,
                        error: {
                            code: -32000,
                            message: 'No controlling editor replied within 30 seconds',
                        },
                    });
                }, 30000);
                this.requests.set(id, { socket, id: message.id, timer });
                this.emit({
                    type: 'clientRequest',
                    id,
                    method: message.method,
                    params: message.params,
                });
            } else if (typeof message.method === 'string' && message.method) {
                this.notification(message.method, object(message.params ?? {}));
            }
        });
        socket.on('close', () => {
            if (this.sess !== socket) {
                return;
            }
            this.sess = undefined;
            this.rejectPending();
            for (const [id, request] of this.requests) {
                clearTimeout(request.timer);
                this.requests.delete(id);
                if (!this.closed) {
                    this.emit({ type: 'clientRequestExpired', id });
                }
            }
            if (!this.closed) {
                this.emit({ type: 'unavailable', message: 'R inspection transport disconnected' });
            }
        });
    }
    request(method: string, params: Record<string, unknown>, timeout = 10000): Promise<unknown> {
        const socket = this.sess;
        if (this.closed || !socket || socket.destroyed) {
            return Promise.reject(new Error('R session is disconnected'));
        }
        return new Promise((resolve, reject) => {
            const id = ++this.counter;
            const timer = setTimeout(() => {
                this.pending.delete(id);
                reject(new Error('R request timed out'));
            }, timeout);
            this.pending.set(id, { resolve, reject, timer });
            this.send(socket, { jsonrpc: '2.0', id, method, params });
        });
    }
    reply(id: string, reply: ClientReply): void {
        const request = this.requests.get(id);
        if (!request) {
            return;
        }
        clearTimeout(request.timer);
        this.requests.delete(id);
        this.send(request.socket, {
            jsonrpc: '2.0',
            id: request.id,
            ...(reply.error
                ? { error: { code: -32000, message: reply.error } }
                : { result: reply.result ?? null }),
        });
    }
    private rejectPending(): void {
        for (const pending of this.pending.values()) {
            clearTimeout(pending.timer);
            pending.reject(new Error('R disconnected'));
        }
        this.pending.clear();
    }
    dispose(): void {
        if (this.closed) {
            return;
        }
        this.closed = true;
        this.rejectPending();
        for (const request of this.requests.values()) {
            clearTimeout(request.timer);
        }
        this.requests.clear();
        for (const socket of this.sockets) {
            socket.destroy();
        }
        for (const server of this.servers) {
            server.close();
        }
    }
}