node_server/tests/forward_manager.test.mjs

import assert from 'node:assert/strict';
import {EventEmitter} from 'node:events';
import {createRequire} from 'node:module';
import {randomUUID} from 'node:crypto';
import {fileURLToPath} from 'node:url';
import {readFile} from 'node:fs/promises';
import vm from 'node:vm';
import test from 'node:test';
import {build, transformSync} from 'esbuild';

const require = createRequire(import.meta.url);
const root = fileURLToPath(new URL('../', import.meta.url));
const built = await build({
    stdin: {contents: `export * from './bot/forward/manager.ts';
        export {handler} from './event_handler/bot/forward_bot.ts';`, resolveDir: root, loader: 'ts'},
    bundle: true, platform: 'node', format: 'cjs',
    packages: 'external', write: false,
    plugins: [{name: 'stream-stub', setup(build) {
        build.onResolve({filter: /^\.\/stream\.ts$/}, () => ({path: 'stream', namespace: 'mock'}));
        build.onLoad({filter: /.*/, namespace: 'mock'}, () => ({contents: 'export class BotStream {}', loader: 'js'}));
    }}],
});
const module = {exports: {}};
vm.runInNewContext(built.outputFiles[0].text, {module, exports: module.exports, require,
    Buffer, URL, setTimeout, clearTimeout, console});
const {NodeForwardManager} = module.exports;

function command(overrides = {}) {
    return {forward_id: 0, forward_session_id: randomUUID(), node_session_id: 'node-session', node_id: 1, bot_id: 2,
        url: 'http://127.0.0.1:1234', path: '/api/forward_server', namespace: '/', token: randomUUID(),
        expires_at: Date.now() + 30_000, deadline_at: Date.now() + 500, ...overrides};
}

function setup() {
    const sockets = [];
    const streams = [];
    const files = [];
    const hooks = new Map();
    const initial = command();
    let active = initial;
    const manager = new NodeForwardManager(1, 'node-session', {
        socket_factory(url, options) {
            const socket = new EventEmitter();
            socket.sent = [];
            socket.url = url;
            socket.options = options;
            socket.auth = options.auth;
            socket.connected = false;
            socket.volatile = {emit: (...args) => socket.sent.push(args)};
            socket.io = {engine: {writeBuffer: []}};
            socket.connect = () => {
                socket.connected = true;
                queueMicrotask(() => socket.emit('connect'));
            };
            socket.disconnect = () => {
                socket.connected = false;
                socket.emit('disconnect');
            };
            socket.timeout = () => ({emit(event, payload, ack) {
                socket.sent.push([event, payload]);
                const respond = data => ack(null, {status: 'success', data});
                if (hooks.has(event)) return hooks.get(event)({socket, payload, respond, ack});
                if (event === 'forward_node_ready') return queueMicrotask(() => respond({node_ready: true}));
                const {forward_id, forward_session_id, node_session_id, node_id, bot_id, expires_at} = active;
                queueMicrotask(() => respond({forward_id, forward_session_id, node_session_id, node_id, bot_id,
                    role: 'node', node_ready: event === 'reauth', expires_at}));
            }});
            sockets.push(socket);
            return socket;
        },
        stream_factory(bot_id) {
            const listeners = new Set();
            const stream = {bot_id, disposed: false, chats: 0,
                snapshot: () => ({bot_id}), logs: () => ({records: []}),
                get_context_info: () => [{name: 'health', i18n: {lang: 'en', name: 'Health', description: ''}, value: 20}],
                chat: () => { stream.chats++; return {accepted: true}; },
                subscribe: listener => { listeners.add(listener); return () => listeners.delete(listener); },
                resync() {}, dispose() { stream.disposed = true; },
                event: event => { for (const listener of listeners) listener(event); },
                get listeners() { return listeners.size; },
            };
            streams.push(stream);
            return stream;
        },
        file_factory(bot_id) {
            const service = {bot_id, disposed: false,
                list: async () => ({listing_id: randomUUID(), expires_at: Date.now() + 300_000, partial: false, children: []}),
                read: async ({id, offset = 0}) => ({id, name: 'empty', encoding: 'base64', offset,
                    next_offset: offset, total_bytes: offset, eof: true, data: ''}),
                dispose() { service.disposed = true; },
            };
            files.push(service);
            return service;
        },
    });
    return {manager, sockets, streams, files, hooks, command: active,
        ensure(value = initial) { active = value; return manager.ensure(value); }};
}

test('lifecycle updates use reliable bot_event even while volatile logs are saturated', async t => {
    const f = setup(); t.after(() => f.manager.stop());
    await f.ensure();
    const socket = f.sockets[0], stream = f.streams[0], received = [];
    socket.on('bot_event', event => received.push(event));
    socket.volatile = {emit() {}};
    const base = {bot_id: 2, context_generation: randomUUID(), stream_id: randomUUID(), emitted_at: Date.now()};
    for (let seq = 1; seq <= 160; seq++) stream.event({...base, seq, name: 'log', payload: {records: []}});
    for (const [index, status] of ['online', 'disconnecting', 'offline'].entries()) {
        stream.event({...base, seq: 161 + index, name: 'status_update', payload: {status, context_exists: true}});
    }
    assert.deepEqual(received.map(event => event.payload.status), ['online', 'disconnecting', 'offline']);
    assert.ok(received.every(event => event.forward_session_id === f.command.forward_session_id));
    stream.event({...base, seq: 165, name: 'message', payload: {text: 'not a lifecycle event'}});
    assert.equal(received.length, 3);
    socket.io.engine.writeBuffer.length = 64;
    stream.event({...base, seq: 166, name: 'status_update', payload: {status: 'offline', context_exists: true}});
    assert.equal(socket.connected, false);
    assert.equal(stream.listeners, 0);
})

const tick = () => new Promise(resolve => setImmediate(resolve));
const code = response => response.data.code;

test('file RPCs work without a bot, clamp ID expiry and invalidate services on transport replacement', async t => {
    const app = setup(); t.after(() => app.manager.stop());
    await app.ensure();
    const rpc = (socket, method, payload = {}, deadline_at = Date.now() + 1000) => new Promise(resolve =>
        socket.emit('relay_request', {request_id: randomUUID(), correlation_id: randomUUID(), method, payload,
            forward_session_id: app.command.forward_session_id, deadline_at}, resolve));
    const first = app.sockets[0];
    const listed = await rpc(first, 'get_data_file_list');
    assert.equal(listed.status, 'success');
    assert.equal(listed.data.expires_at, app.command.expires_at);
    assert.equal(app.files.length, 1);
    assert.equal(app.files[0].bot_id, app.command.bot_id);
    const id = randomUUID();
    assert.equal((await rpc(first, 'get_file_data', {id})).data.id, id);
    assert.equal(code(await rpc(first, 'get_file_data', {id, length: 262145})), 'INVALID_REQUEST');
    assert.equal(code(await rpc(first, 'get_file_data', {id, offset: -1})), 'INVALID_REQUEST');
    let release;
    app.files[0].read = () => new Promise(resolve => {release = resolve;});
    const pending = rpc(first, 'get_file_data', {id});
    first.disconnect();
    assert.equal(app.files[0].disposed, true);
    release({id});
    assert.equal(code(await pending), 'SESSION_STALE');
    assert.equal((await app.ensure({...app.command, token: randomUUID()})).status, 'success');
    assert.equal((await rpc(app.sockets[1], 'get_data_file_list')).status, 'success');
    assert.equal(app.files.length, 2);
    let late;
    app.files[1].list = () => new Promise(resolve => {late = resolve;});
    const timed = rpc(app.sockets[1], 'get_data_file_list', {}, Date.now() + 10);
    await new Promise(resolve => setTimeout(resolve, 15));
    late({listing_id: randomUUID(), expires_at: Date.now() + 1000, partial: false, children: []});
    assert.equal(code(await timed), 'TIMEOUT');
    app.manager.stop();
    assert.equal(app.files[1].disposed, true);
});

test('ready ACK waits for ready confirmation, with one RPC handler and log observer across renewals', async t => {
    const app = setup();
    t.after(() => app.manager.stop());
    let release;
    app.hooks.set('forward_node_ready', ({respond}) => { release = respond; });
    let completed = false;
    const result = app.ensure().then(value => { completed = true; return value; });
    await tick();
    assert.equal(completed, false);
    assert.equal(app.sockets[0].listenerCount('relay_request'), 1);
    assert.equal(app.streams[0].listeners, 1);
    release({node_ready: true});
    assert.equal((await result).status, 'success');
    app.hooks.delete('forward_node_ready');
    const next = {...app.command, token: randomUUID()};
    assert.equal((await app.ensure(next)).status, 'success');
    assert.equal(app.sockets.length, 1);
    assert.equal(app.streams[0].listeners, 1);
    assert.equal(app.sockets[0].sent.filter(([event]) => event === 'reauth').length, 1);
});

test('absolute deadline cancels stalled readiness, closes channel and ignores late ready ACK', async t => {
    const app = setup();
    t.after(() => app.manager.stop());
    let release;
    app.hooks.set('forward_node_ready', ({respond}) => { release = respond; });
    const request = {...app.command, deadline_at: Date.now() + 35};
    const response = await app.ensure(request);
    assert.equal(code(response), 'TIMEOUT');
    assert.equal(app.sockets[0].connected, false);
    assert.equal(app.sockets[0].listenerCount('relay_request'), 0);
    assert.equal(app.streams[0].disposed, true);
    release({node_ready: true});
    await tick();
    assert.equal(app.sockets[0].connected, false);
});

test('session scope and whitelist validation reject stale node, wrong bot scope, and extra fields', async t => {
    const app = setup();
    t.after(() => app.manager.stop());
    assert.equal(code(await app.ensure({...app.command, node_session_id: 'old'})), 'SESSION_STALE');
    assert.equal(code(await app.ensure({...app.command, node_id: 3})), 'FORBIDDEN');
    assert.equal(code(await app.ensure({...app.command, extra: true})), 'INVALID_REQUEST');
    assert.equal((await app.ensure()).status, 'success');
    assert.equal(code(await app.ensure({...app.command, bot_id: 3})), 'SESSION_STALE');
    assert.equal(app.sockets.length, 1);
});

test('serial ensure ignores expired queued work without closing a healthy current operation', async t => {
    const app = setup();
    t.after(() => app.manager.stop());
    let release;
    app.hooks.set('forward_node_ready', ({respond}) => { release = respond; });
    const first = app.ensure();
    await tick();
    const second = app.ensure({...app.command, token: randomUUID(), deadline_at: Date.now() + 20});
    assert.equal(code(await second), 'TIMEOUT');
    assert.equal(app.sockets[0].connected, true);
    release({node_ready: true});
    assert.equal((await first).status, 'success');
    await tick();
    assert.equal(app.sockets.length, 1);
    assert.equal(app.sockets[0].connected, true);
});

test('closed session cannot revive, old callbacks cannot execute chat, and manager stop removes observers', async () => {
    const app = setup();
    assert.equal((await app.ensure()).status, 'success');
    const socket = app.sockets[0];
    const stale_rpc = socket.listeners('relay_request')[0];
    socket.emit('forward_session_closed', {forward_session_id: app.command.forward_session_id,
        reason: 'revoked', code: 'CANCELLED'});
    assert.equal(app.streams[0].disposed, true);
    assert.equal(socket.connected, false);
    assert.equal(code(await app.ensure({...app.command, token: randomUUID()})), 'SESSION_STALE');
    let response;
    stale_rpc({request_id: randomUUID(), correlation_id: randomUUID(), method: 'bot.chat', payload: {message: 'hello'},
        forward_session_id: app.command.forward_session_id, deadline_at: Date.now() + 1000}, value => { response = value; });
    assert.equal(code(response), 'SESSION_STALE');
    assert.equal(app.streams[0].chats, 0);
    app.manager.stop();
    assert.equal(code(await app.ensure(command())), 'SESSION_STALE');
});

test('RPC checks deadlines and deduplicates correlation IDs, while allowing grant request ID reuse', async t => {
    const app = setup();
    t.after(() => app.manager.stop());
    await app.ensure();
    const base = {request_id: randomUUID(), correlation_id: randomUUID(), method: 'bot.chat', payload: {message: 'hello'},
        forward_session_id: app.command.forward_session_id, deadline_at: Date.now() + 1000};
    const rpc = request => new Promise(resolve => app.sockets[0].emit('relay_request', request, resolve));
    assert.equal((await rpc(base)).status, 'success');
    assert.equal(code(await rpc(base)), 'CANCELLED');
    assert.equal((await rpc({...base, correlation_id: randomUUID()})).status, 'success');
    assert.equal(app.streams[0].chats, 2);
    assert.equal(code(await rpc({...base, correlation_id: randomUUID(), deadline_at: Date.now() - 1})), 'TIMEOUT');
    assert.equal(app.streams[0].chats, 2);
    assert.equal(code(await rpc({...base, method: 'bot.subscribe', payload: {events: []}})), 'INVALID_REQUEST');
    const info = await rpc({...base, correlation_id: randomUUID(), method: 'get_context_info', payload: {}});
    assert.equal(info.status, 'success');
    assert.equal(info.data[0].value, 20);
});

test('transport recovery requires a fresh token; ordinary disconnect never reconnects automatically', async t => {
    const app = setup();
    t.after(() => app.manager.stop());
    await app.ensure();
    app.sockets[0].disconnect();
    assert.equal(app.sockets[0].options.reconnection, false);
    assert.equal(code(await app.ensure(app.command)), 'TOKEN_EXPIRED');
    assert.equal(app.sockets.length, 1);
    assert.equal((await app.ensure({...app.command, token: randomUUID()})).status, 'success');
    assert.equal(app.sockets.length, 2);
    assert.equal(app.streams.length, 1);
    assert.equal(app.streams[0].listeners, 1);
});

test('wrong-scope join response closes the channel and sanitized ACK never exposes raw relay details', async t => {
    const app = setup();
    t.after(() => app.manager.stop());
    app.hooks.set('join_room', ({respond}) => respond({node_id: 999, token: 'secret', stack: 'private'}));
    const response = await app.ensure();
    assert.equal(code(response), 'SESSION_STALE');
    assert.equal(app.sockets[0].connected, false);
    assert.doesNotMatch(JSON.stringify(response), /secret|private|stack/);
});

test('many historical closed sessions do not consume the active-session budget', async t => {
    const app = setup();
    t.after(() => app.manager.stop());
    for (let index = 0; index < 505; index++) {
        const request = command({deadline_at: Date.now() + 2000});
        assert.equal((await app.ensure(request)).status, 'success');
        app.sockets.at(-1).emit('forward_session_closed', {forward_session_id: request.forward_session_id,
            reason: 'revoked', code: 'CANCELLED'});
    }
    assert.equal(app.streams.filter(stream => !stream.disposed).length, 0);
    assert.equal((await app.ensure(command())).status, 'success');
});

test('lease expiry retires data channel and queue capacity stays bounded under stalled work', async t => {
    const app = setup();
    t.after(() => app.manager.stop());
    const request = {...app.command, expires_at: Date.now() + 40};
    assert.equal((await app.ensure(request)).status, 'success');
    await new Promise(resolve => setTimeout(resolve, 50));
    assert.equal(app.sockets[0].connected, false);
    assert.equal(app.streams[0].disposed, true);
    assert.equal(code(await app.ensure({...request, expires_at: Date.now() + 1000, token: randomUUID()})), 'SESSION_STALE');

    const stalled = setup();
    t.after(() => stalled.manager.stop());
    stalled.hooks.set('forward_node_ready', () => {});
    const first = stalled.ensure();
    await tick();
    const queued = Array.from({length: 15}, () => stalled.ensure({...stalled.command, token: randomUUID()}));
    assert.equal(code(await stalled.ensure({...stalled.command, token: randomUUID()})), 'RATE_LIMITED');
    stalled.manager.stop();
    const replies = await Promise.all([first, ...queued]);
    assert.ok(replies.every(reply => reply.status === 'error'));
    assert.equal(stalled.sockets.length, 1);
});

test('lifecycle events are forwarded while ready and congestion closes the stale transport', async t => {
    const app = setup();
    t.after(() => app.manager.stop());
    await app.ensure();
    const socket = app.sockets[0];
    const stream = app.streams[0];
    let resyncs = 0;
    stream.resync = () => { resyncs++; };
    const event = {name: 'status_update', seq: 1, payload: {status: 'offline', context_exists: false}};
    let lifecycle = 0;
    socket.on('bot_event', () => lifecycle++);
    stream.event(event);
    assert.equal(lifecycle, 1);
    assert.equal(socket.sent.filter(([name]) => name === 'bot_log').length, 0);
    stream.event({name: 'log', seq: 2, payload: {records: []}});
    assert.equal(socket.sent.filter(([name]) => name === 'bot_log').length, 1);
    socket.emit('relay_status', {forward_session_id: app.command.forward_session_id, node_ready: true, reason: 'resync_required'});
    assert.equal(resyncs, 1);
    socket.io.engine.writeBuffer.length = 64;
    stream.event({...event, seq: 2});
    stream.event({name: 'log', seq: 3, payload: {records: []}});
    assert.equal(socket.sent.filter(([name]) => name === 'bot_log').length, 1);
    assert.equal(lifecycle, 1);
    assert.equal(socket.connected, false);
    socket.emit('relay_status', {forward_session_id: app.command.forward_session_id, node_ready: true, reason: 'resync_required'});
    assert.equal(resyncs, 1);
    socket.disconnect();
    stream.event({...event, seq: 3});
    stream.event({name: 'log', seq: 4, payload: {records: []}});
    assert.equal(socket.sent.filter(([name]) => name === 'bot_log').length, 1);
    assert.equal(lifecycle, 1);
    assert.equal(stream.listeners, 0);
});

test('info updates use the data envelope and share push congestion and rate controls with logs', async t => {
    const app = setup(); t.after(() => app.manager.stop());
    let ready;
    app.hooks.set('forward_node_ready', ({respond}) => {ready = respond;});
    const pending = app.ensure(); await tick();
    const stream = app.streams[0]; const socket = app.sockets[0];
    const event = {bot_id: 2, context_generation: randomUUID(), stream_id: randomUUID(), emitted_at: Date.now(),
        seq: 1, name: 'info_update', payload: {name: 'health', value: 18}};
    stream.event(event);
    assert.equal(socket.sent.some(([name]) => name === 'info_update'), false);
    ready({node_ready: true}); await pending;
    stream.event(event);
    const packet = socket.sent.find(([name]) => name === 'info_update')[1];
    assert.deepEqual(Object.keys(packet).sort(), ['bot_id', 'context_generation', 'data', 'emitted_at', 'forward_session_id', 'seq', 'stream_id']);
    assert.equal(packet.data.value, 18);
    assert.equal(packet.forward_session_id, app.command.forward_session_id);
    socket.io.engine.writeBuffer.length = 64; stream.event({...event, seq: 2});
    assert.equal(socket.sent.filter(([name]) => name === 'info_update').length, 1);
    socket.io.engine.writeBuffer.length = 0;
    for (let seq = 3; seq < 203; seq++) stream.event({...event, seq, name: seq % 2 ? 'log' : 'info_update'});
    assert.equal(socket.sent.filter(([name]) => ['info_update', 'bot_log'].includes(name)).length, 149);
    socket.disconnect(); stream.event({...event, seq: 204});
    assert.equal(stream.listeners, 0);
    assert.equal(socket.sent.filter(([name]) => ['info_update', 'bot_log'].includes(name)).length, 149);
});

test('plugin lifecycle updates survive a temporarily unwritable transport and retain congestion bounds', async t => {
    const app = setup(); t.after(() => app.manager.stop());
    await app.ensure();
    const socket = app.sockets[0], stream = app.streams[0];
    const received = [];
    socket.on('info_update', update => received.push(update));
    // Match Socket.IO volatile behavior during an outstanding WebSocket write.
    socket.volatile = {emit() {}};
    const event = {bot_id: 2, context_generation: randomUUID(), stream_id: randomUUID(), emitted_at: Date.now(),
        name: 'info_update', payload: {name: 'plugin_states', value: {available: true, plugins: []}}};
    for (const [index, lifecycle] of ['initializing', 'enabled'].entries()) {
        stream.event({...event, seq: index + 1, payload: {...event.payload, value: {available: true,
            plugins: [{name: 'auto_reconnect', path: 'runtime', lifecycle, enabled: lifecycle === 'enabled', states: {attempts: 0}}]}}});
    }
    assert.deepEqual(received.map(update => update.data.value.plugins[0].lifecycle), ['initializing', 'enabled']);
    assert.deepEqual(received.map(update => update.seq), [1, 2]);
    assert.ok(received.every(update => update.forward_session_id === app.command.forward_session_id));
    assert.ok(received.every(update => !('name' in update) && !('payload' in update)));
    stream.event({...event, seq: 3, payload: {name: 'health', value: 20}});
    assert.equal(received.length, 2);
    socket.io.engine.writeBuffer.length = 64;
    stream.event({...event, seq: 4});
    assert.equal(received.length, 2);
    socket.io.engine.writeBuffer.length = 0;
    for (let seq = 5; seq <= 155; seq++) stream.event({...event, seq});
    assert.equal(received.length, 148);
    socket.disconnect(); stream.event({...event, seq: 156});
    assert.equal(received.length, 148);
    assert.equal(stream.listeners, 0);
});

test('forward_bot handler requires an ACK and returns structured authorization errors', async () => {
    const {handler} = module.exports;
    const socket = {connected: true, emit() { assert.fail('Unexpected non-ACK response'); }};
    await handler(socket, command(), undefined);
    const replies = [];
    await handler(socket, command(), reply => replies.push(reply));
    assert.equal(replies.length, 1);
    assert.equal(code(replies[0]), 'UNAUTHORIZED');
    socket.connected = false;
    await handler(socket, command(), reply => replies.push(reply));
    assert.equal(code(replies[1]), 'NODE_OFFLINE');
});

test('dispatcher delivers forward_bot while hosted relay control events remain on their own handlers', async () => {
    const source = await readFile(new URL('../event_handler/init.ts', import.meta.url), 'utf8');
    const dispatch = transformSync(source.slice(source.indexOf('export async function event_handler'))
        .replace('export async function', 'async function'), {loader: 'ts'}).code;
    const calls = [];
    const map = new Map([['forward_bot', (...args) => calls.push(args)]]);
    const context = {handler_map: map, console_print: {log() {}, warning() {}, error() {}}, socket: {}, payload: command()};
    await vm.runInNewContext(`${dispatch}\nevent_handler(socket, 'forward_bot', payload, () => {});`, context);
    assert.equal(calls.length, 1);
    await vm.runInNewContext("event_handler(socket, 'forward_open_session', payload, () => {});", context);
    assert.equal(calls.length, 1);
});