import assert from 'node:assert/strict';
import {randomUUID} from 'node:crypto';
import {EventEmitter} from 'node:events';
import {createRequire} from 'node:module';
import {fileURLToPath} from 'node:url';
import test from 'node:test';
import {build} from 'esbuild';

const root = fileURLToPath(new URL('../', import.meta.url));
const compiled = await build({
    stdin: {contents: `export * from './bot/forward/stream.ts'; export * from './bot/forward/manager.ts';
        export * from './utils/forward_protocol.ts';`, resolveDir: root, loader: 'ts'},
    bundle: true, platform: 'node', format: 'cjs', packages: 'external', write: false,
    plugins: [{name: 'isolated-registry', setup(build) {
        build.onResolve({filter: /^\.\.\/instance\/index\.ts$/}, () => ({path: 'registry', namespace: 'mock'}));
        build.onLoad({filter: /.*/, namespace: 'mock'}, () => ({loader: 'js', contents:
            'export const get_bot_context = () => undefined; export const observe_bot_context = () => () => {};'}));
    }}],
});
const module = {exports: {}};
new Function('require', 'module', 'exports', compiled.outputFiles[0].text)(createRequire(import.meta.url), module, module.exports);
const {BotStream, NodeForwardManager, validate_rpc} = module.exports;
const turn = () => new Promise(resolve => setImmediate(resolve));
const deferred = () => {
    let resolve, reject;
    const promise = new Promise((yes, no) => { resolve = yes; reject = no; });
    return {promise, resolve, reject};
};
const fails = (action, code = 'INVALID_REQUEST') => assert.throws(action, error => error.code === code);
const rejects = (promise, code) => assert.rejects(promise, error => error.code === code);
const request = (method, payload, scope = randomUUID(), deadline_at = Date.now() + 1000) => ({
    request_id: randomUUID(), correlation_id: randomUUID(), method, payload, forward_session_id: scope, deadline_at,
});

function fixture(t) {
    let now = 100;
    const calls = [];
    const run_id = randomUUID();
    const snapshot = {run_id, workflow_id: 'gather', status: 'running', steps: 1, active_nodes: ['node'],
        started_at: 1, ended_at: null, stop_reason: null, error: null,
        trigger: {event: 'custom', args: [false, 0, null, {nested: [false, 0, null]}]}, trigger_available: true,
        branches: [{trigger: {}, vars: {value: 1}, outputs: {}, inputs: {}, last_result: null, retry_counts: {}}]};
    const manager = {
        start(id, trigger) { calls.push(['start', id, trigger]); return {status: 'started', run_id}; },
        async stop(id) { calls.push(['stop', id]); },
        get_snapshot(id) { calls.push(['snapshot', id]); return id === run_id ? snapshot : undefined; },
        get_snapshots() { calls.push(['snapshots']); return [snapshot]; },
    };
    const context = Object.assign(new EventEmitter(), {bot_status: 'online', bot_instance: {chat() {}}, workflow_manager: manager});
    let current = context;
    const observers = new Set();
    const source = {get: () => current, observe(_id, listener) { observers.add(listener); return () => observers.delete(listener); }};
    const stream = new BotStream(1, {source, now: () => now});
    t.after(() => stream.dispose());
    return {stream, source, context, manager, calls, snapshot, run_id,
        set(context) { current = context; for (const listener of observers) listener(context); },
        time(value) { now = value; },
    };
}

test('workflow protocol accepts exact start, stop and optional snapshot payloads', () => {
    for (const [method, payload] of [
        ['workflow.start', {workflow_id: 'gather-1_test'}],
        ['workflow.start', {workflow_id: 'x'.repeat(100), trigger: {nested: {items: [1, true, null, 'value']}}}],
        ['workflow.stop', {run_id: randomUUID()}],
        ['workflow.snapshot', {}],
        ['workflow.snapshot', {run_id: randomUUID()}],
    ]) {
        const valid = request(method, payload);
        assert.equal(validate_rpc(valid), valid);
        fails(() => validate_rpc({...valid, payload: {...payload, extra: true}}));
    }
    for (const workflow_id of ['', '../x', 'a/b', 'a.b', 'a b', 'x'.repeat(101), '__proto__', 'prototype', 'constructor', 1])
        fails(() => validate_rpc(request('workflow.start', {workflow_id})));
    for (const method of ['workflow.stop', 'workflow.snapshot'])
        for (const run_id of ['', 'not-a-uuid', null, undefined, 1]) fails(() => validate_rpc(request(method, {run_id})));
    fails(() => validate_rpc(request('workflow.start', {})));
    fails(() => validate_rpc(request('workflow.stop', {})));
    fails(() => validate_rpc(request('workflow.delete', {})));
});

test('workflow triggers require bounded plain JSON without accessors, cycles or unsafe keys', () => {
    let reads = 0;
    const getter = Object.defineProperty({}, 'secret', {enumerable: true, get() { reads++; return 1; }});
    const cycle = {}; cycle.self = cycle;
    let deep = {};
    for (let index = 0; index < 66; index++) deep = {next: deep};
    const sparse = []; sparse.length = 2;
    for (const trigger of [null, [], 'text', undefined, new Date(), Object.create({inherited: true}),
        getter, cycle, deep, {sparse}, {value: NaN}, {value: Infinity}, {value: 1n}, {value() {}},
        {value: undefined}, {huge: 'x'.repeat(1_048_576)}, {values: Array(100_001).fill(0)},
        JSON.parse('{"__proto__":1}'), {nested: {constructor: 'unsafe'}}, {[Symbol('key')]: 1},
        Object.defineProperty({}, 'hidden', {value: 1})]) {
        fails(() => validate_rpc(request('workflow.start', {workflow_id: 'test', trigger})));
    }
    assert.equal(reads, 0);
    const trigger = Object.assign(Object.create(null), {safe: {values: [null, 2]}});
    assert.equal(validate_rpc(request('workflow.start', {workflow_id: 'test', trigger})).payload.trigger, trigger);
});

test('stream start preserves the manager result and copies its trigger input and result', t => {
    const app = fixture(t);
    const trigger = {data: {value: 1}};
    const result = {status: 'queued', request_id: randomUUID()};
    app.manager.start = (id, input) => { app.calls.push([id, input]); input.data.value = 2; return result; };
    let checks = 0;
    const actual = app.stream.workflow_start({workflow_id: 'gather', trigger}, 200, () => checks++);
    assert.deepEqual(actual, result);
    assert.notEqual(actual, result);
    assert.equal(trigger.data.value, 1);
    assert.equal(app.calls[0][0], 'gather');
    assert.equal(app.calls[0][1].data.value, 2);
    assert.ok(checks >= 2);
    app.manager.start = () => ({status: 'skipped'});
    assert.deepEqual(app.stream.workflow_start({workflow_id: 'gather'}, 200), {status: 'skipped'});
});

test('stream workflow guards reject invalid, expired, missing, offline and stale starts before manager calls', t => {
    const app = fixture(t);
    fails(() => app.stream.workflow_start({workflow_id: '../x'}, 200));
    for (const deadline of [100, 99, NaN, Infinity, 1.5])
        fails(() => app.stream.workflow_start({workflow_id: 'gather'}, deadline), 'TIMEOUT');
    app.context.bot_status = 'offline';
    fails(() => app.stream.workflow_start({workflow_id: 'gather'}, 200), 'BOT_NOT_ONLINE');
    app.context.bot_status = 'online';
    app.context.bot_instance = null;
    fails(() => app.stream.workflow_start({workflow_id: 'gather'}, 200), 'BOT_NOT_ONLINE');
    app.context.bot_instance = {chat() {}};
    app.context.workflow_manager = null;
    fails(() => app.stream.workflow_start({workflow_id: 'gather'}, 200), 'INTERNAL_ERROR');
    app.context.workflow_manager = app.manager;
    fails(() => app.stream.workflow_start({workflow_id: 'gather'}, 200, () => {
        app.context.workflow_manager = {};
    }), 'CANCELLED');
    app.set(undefined);
    fails(() => app.stream.workflow_start({workflow_id: 'gather'}, 200), 'NO_CONTEXT');
    app.stream.dispose();
    fails(() => app.stream.workflow_start({workflow_id: 'gather'}, 200), 'CANCELLED');
    assert.deepEqual(app.calls, []);
});

test('workflow transport and post-start context checks prevent stale responses', t => {
    const app = fixture(t);
    fails(() => app.stream.workflow_start({workflow_id: 'gather'}, 200, () => {
        throw Object.assign(Error('stale'), {code: 'SESSION_STALE'});
    }), 'SESSION_STALE');
    assert.deepEqual(app.calls, []);
    app.manager.start = () => { app.set(Object.assign(new EventEmitter(), app.context)); return {status: 'skipped'}; };
    fails(() => app.stream.workflow_start({workflow_id: 'gather'}, 200), 'CANCELLED');
});

test('snapshots are deeply copied, available offline and return null for an unknown run', t => {
    const app = fixture(t);
    app.context.bot_status = 'offline';
    app.context.bot_instance = null;
    const all = app.stream.workflow_snapshot({}, 200);
    const one = app.stream.workflow_snapshot({run_id: app.run_id}, 200);
    const {branches, trigger, ...summary} = app.snapshot;
    const expected_trigger = structuredClone(trigger);
    assert.deepEqual(all, [summary]);
    assert.equal(Object.hasOwn(all[0], 'trigger'), false);
    assert.equal(Object.hasOwn(all[0], 'branches'), false);
    assert.deepEqual(one, app.snapshot);
    all[0].active_nodes.push('summary mutation');
    one.branches[0].vars.value = 2;
    one.trigger.args[3].nested.push('detail mutation');
    one.active_nodes.push('extra');
    assert.equal(app.snapshot.branches[0].vars.value, 1);
    assert.deepEqual(app.snapshot.trigger, expected_trigger);
    assert.deepEqual(app.snapshot.active_nodes, ['node']);
    assert.equal(app.stream.workflow_snapshot({run_id: randomUUID()}, 200), null);
    delete app.snapshot.trigger;
    assert.equal(app.stream.workflow_snapshot({}, 200)[0].trigger_available, false);
    app.manager.get_snapshots = () => { app.time(201); return []; };
    fails(() => app.stream.workflow_snapshot({}, 200), 'TIMEOUT');
});

test('workflow stop waits for drain, works offline and rejects changed contexts or expired results', async t => {
    for (const change of ['none', 'deadline', 'context', 'instance', 'manager', 'disposed', 'transport']) {
        await t.test(change, async t => {
            const app = fixture(t);
            app.context.bot_status = 'offline';
            const drain = deferred();
            let settled = false;
            let stale = false;
            app.manager.stop = id => { app.calls.push(id); return drain.promise; };
            const result = app.stream.workflow_stop({run_id: app.run_id}, 200, () => {
                if (stale) throw Object.assign(Error('stale'), {code: 'SESSION_STALE'});
            });
            const expected = {deadline: 'TIMEOUT', context: 'CANCELLED', instance: 'CANCELLED', manager: 'CANCELLED',
                disposed: 'CANCELLED', transport: 'SESSION_STALE'}[change];
            const checked = expected ? rejects(result, expected) : result.then(value => assert.deepEqual(value, {stopped: true}));
            void result.then(() => {settled = true;}, () => {settled = true;});
            await turn();
            assert.equal(settled, false);
            assert.deepEqual(app.calls, [app.run_id]);
            if (change === 'deadline') app.time(201);
            if (change === 'context') app.set(undefined);
            if (change === 'instance') app.context.bot_instance = {};
            if (change === 'manager') app.context.workflow_manager = {};
            if (change === 'disposed') app.stream.dispose();
            if (change === 'transport') stale = true;
            drain.resolve();
            await checked;
        });
    }
});

async function relay(t, stream) {
    const command = {forward_id: 1, forward_session_id: randomUUID(), node_session_id: 'session', node_id: 1, bot_id: 1,
        url: 'http://127.0.0.1:1234', path: '/socket.io', namespace: '/', token: 'opaque',
        expires_at: Date.now() + 10_000, deadline_at: Date.now() + 2000};
    const socket = new EventEmitter();
    socket.auth = {};
    socket.io = {engine: {writeBuffer: []}};
    socket.volatile = {emit() {}};
    socket.connect = () => { socket.connected = true; queueMicrotask(() => socket.emit('connect')); };
    socket.disconnect = () => { socket.connected = false; socket.emit('disconnect'); };
    socket.timeout = () => ({emit(event, _payload, ack) {
        const {forward_id, forward_session_id, node_session_id, node_id, bot_id, expires_at} = command;
        const data = event === 'forward_node_ready' ? {node_ready: true}
            : {forward_id, forward_session_id, node_session_id, node_id, bot_id, expires_at, role: 'node', node_ready: false};
        queueMicrotask(() => ack(null, {status: 'success', data}));
    }});
    const manager = new NodeForwardManager(1, 'session', {socket_factory: () => socket, stream_factory: () => stream});
    t.after(() => manager.stop());
    assert.equal((await manager.ensure(command)).status, 'success');
    return {socket, manager, rpc: (method, payload, deadline = Date.now() + 1000) => new Promise(resolve =>
        socket.emit('relay_request', request(method, payload, command.forward_session_id, deadline), resolve))};
}

test('Node dispatches workflow RPCs through stream guards and retains function RPC callbacks', async t => {
    const app = fixture(t);
    // Relay deadlines use the wall clock while direct stream tests use a fixed clock.
    app.time(Date.now());
    let before = 0;
    app.context.function_manager = {async run(path, name, args, check) {
        check(); before++; assert.deepEqual([path, name, args], ['bot', 'chat', ['hello']]); return true;
    }};
    const channel = await relay(t, app.stream);
    assert.deepEqual(await channel.rpc('workflow.start', {workflow_id: 'gather'}),
        {status: 'success', data: {status: 'started', run_id: app.run_id}});
    assert.deepEqual(await channel.rpc('workflow.snapshot', {run_id: app.run_id}), {status: 'success', data: app.snapshot});
    const {branches, trigger, ...summary} = app.snapshot;
    assert.deepEqual(await channel.rpc('workflow.snapshot', {}), {status: 'success', data: [summary]});
    assert.deepEqual(await channel.rpc('workflow.stop', {run_id: app.run_id}), {status: 'success', data: {stopped: true}});
    assert.deepEqual(await channel.rpc('execute_function', {function: 'bot', name: 'chat', args: ['hello']}),
        {status: 'success', data: true});
    assert.equal(before, 1);
    assert.equal((await channel.rpc('workflow.start', {workflow_id: '../bad'})).data.code, 'INVALID_REQUEST');
    assert.equal((await channel.rpc('workflow.start', {workflow_id: 'gather'}, Date.now() - 1)).data.code, 'TIMEOUT');
});

test('Node returns stop TIMEOUT without waiting for drain and safely contains late rejection', async t => {
    const app = fixture(t);
    const drain = deferred();
    app.manager.stop = () => drain.promise;
    const channel = await relay(t, app.stream);
    const response = await channel.rpc('workflow.stop', {run_id: app.run_id}, Date.now() + 30);
    assert.equal(response.status, 'error');
    assert.equal(response.data.code, 'TIMEOUT');
    drain.reject(Error('SECRET late failure'));
    await turn();
    assert.equal((await channel.rpc('workflow.snapshot', {})).status, 'success');
});

test('Node rejects stopped transport results and sanitizes workflow manager failures', async t => {
    const app = fixture(t);
    const drain = deferred();
    app.manager.stop = () => drain.promise;
    const channel = await relay(t, app.stream);
    app.manager.start = () => { throw Error('SECRET workflow error'); };
    const failed = await channel.rpc('workflow.start', {workflow_id: 'gather'});
    assert.equal(failed.data.code, 'INTERNAL_ERROR');
    assert.equal(JSON.stringify(failed).includes('SECRET'), false);
    const pending = channel.rpc('workflow.stop', {run_id: app.run_id});
    channel.socket.disconnect();
    drain.resolve();
    assert.equal((await pending).data.code, 'SESSION_STALE');
});

test('workflow state updates retain reliable transport emission', async t => {
    const app = fixture(t);
    app.context.info_manager = {};
    const channel = await relay(t, app.stream);
    const reliable = [];
    const volatile = [];
    channel.socket.on('info_update', packet => reliable.push(packet));
    channel.socket.volatile.emit = (event, packet) => volatile.push([event, packet]);
    const update = {name: 'workflow_states', value: [{run_id: app.run_id, status: 'completed'}]};
    app.context.emit('info_update', update);
    assert.equal(reliable.length, 1);
    assert.deepEqual(reliable[0].data, update);
    assert.deepEqual(volatile, []);
});