import test from 'node:test';
import assert from 'node:assert/strict';
import { mkdtemp, writeFile, readFile, rm } from 'node:fs/promises';
import { join } from 'node:path';
import { tmpdir } from 'node:os';
import { setTimeout as delay } from 'node:timers/promises';
import { createServer } from 'node:net';
import * as orgBus from '../../cli/src/org-bus.mjs';
import { busCaller } from '../src/bus.mjs';
import { consoleBus } from '../src/console-bus.mjs';
import { organizationConsole } from '../src/organizations.mjs';
import { readWorkspace, workspaceHTTP, parseWorkspaceQuery } from '../src/workspace-read.mjs';

const session = { provider: 'local-dev', iss: 'local-dev', sub: 'local-dev-owner' };
const goodOptions = overrides => ({
  bus: { tasks: async () => [], agents: async () => [], inbox: async () => [], 'thread.claims': async () => [], usage: async () => ({}), limits: async () => ({}), decisions: async () => [], ...overrides?.bus },
  chat: { projects: async () => [{ id: 'p', name: 'Healthy' }], threads: async () => [], ...overrides?.chat },
});
const member = (id, overrides) => ({ id, doc: {}, status: async () => ({ healthy: true }), options: () => goodOptions(overrides) });
const orgs = (...members) => ({ members: async () => members });
const response = () => ({ writeHead(status) { this.status = status; }, end(body) { this.body = JSON.parse(body); } });

for (const [name, overrides] of [
  ['null thread', { chat: { threads: async () => [null] } }],
  ['null inbox option', { bus: { inbox: async () => [{ id: 'd', options: [null] }] } }],
  ['invalid project name', { chat: { projects: async () => [{ id: 'bad', name: 42 }] } }],
  ['non-array tasks', { bus: { tasks: async () => ({ tasks: [] }) } }],
  ['invalid usage total', { bus: { usage: async () => ({ total: 42 }) } }],
]) {
  test(`malformed ${name} is isolated from a healthy organization`, async () => {
    const res = response();
    await workspaceHTTP({ method: 'GET' }, res, { organizations: orgs(member('good'), member('bad', overrides)), session, owner: true });
    assert.equal(res.status, 200);
    assert.deepEqual(res.body.projects.map(p => p.org), ['good']);
    assert.deepEqual(res.body.summary, { read: 1, of: 2 });
    const bad = res.body.orgs.find(o => o.org === 'bad');
    assert.equal(bad.state, 'invalid');
    assert.equal(bad.error, 'invalid-response');
    assert.ok(Number.isFinite(Date.parse(bad.at)));
  });
}

test('options construction failure is isolated from a healthy organization', async () => {
  const bad = { ...member('bad'), options() { throw new Error('private detail'); } };
  const res = response();
  await workspaceHTTP({ method: 'GET' }, res, { organizations: orgs(member('good'), bad), session, owner: true });
  assert.equal(res.status, 200);
  assert.deepEqual(res.body.projects.map(p => p.org), ['good']);
  assert.deepEqual(res.body.summary, { read: 1, of: 2 });
  assert.equal(res.body.orgs[1].state, 'invalid');
  assert.equal(res.body.orgs[1].error, 'invalid-response');
  assert.ok(Number.isFinite(Date.parse(res.body.orgs[1].at)));
});

for (const [name, statusDelay, pluginDelay] of [['in-flight child', 0, 100], ['late channel continuation', 220, 300]]) {
  test(`deadline cancels ${name} and prevents any later child`, async t => {
    const dir = await mkdtemp(join(tmpdir(), 'ws-'));
    t.after(() => rm(dir, { recursive: true, force: true }));
    const cli = join(dir, 'cli.mjs'), log = join(dir, 'log');
    await writeFile(cli, `import {appendFileSync} from 'node:fs';let input='';process.stdin.on('data',x=>input+=x);process.stdin.on('end',()=>{const q=JSON.parse(input);appendFileSync(${JSON.stringify(log)},JSON.stringify({pid:process.pid,verb:q.verb})+'\\n');if(q.verb==='plugins')setTimeout(()=>process.stdout.write(JSON.stringify({plugins:{channel:['webui']}})),${pluginDelay});else if(q.verb==='project.list')setInterval(()=>{},1000);else process.stdout.write('[]');});`);
    const timeoutMs = 500;
    const organizations = organizationConsole({ ctx: {}, listOrgs: async () => ['good', 'slow'].map(id => ({ id, state: 'active' })),
      showOrg: async () => ({ doc: {} }), resolveOrg: async (_, id) => ({ id, state: 'active', root: dir }), orgPaths: () => ({ socket: join(dir, 'unused.sock') }),
      busStatus: async (_, id) => { if (id === 'slow') await delay(statusDelay); return { healthy: true }; },
      options: ({ id, ...bound }) => id === 'good' ? goodOptions() : consoleBus(busCaller({ dataRoot: dir, socket: bound.paths.socket, business: id, cli, timeoutMs, ...bound })) });
    const body = await readWorkspace({ organizations, session, timeoutMs, query: parseWorkspaceQuery('sections=tasks') });
    assert.equal(body.orgs.find(o => o.org === 'slow').state, 'timeout');
    assert.deepEqual(body.summary, { read: 1, of: 2 });
    const rows = async () => (await readFile(log, 'utf8')).trim().split('\n').map(JSON.parse);
    const atReturn = await rows();
    if (!statusDelay) {
      await delay(30);
      for (const { pid } of atReturn) assert.throws(() => process.kill(pid, 0), { code: 'ESRCH' }, `transport ${pid} still alive`);
    }
    await delay(350);
    const later = await rows();
    assert.deepEqual(later, atReturn, 'no transport starts after timeout');
    assert.ok(atReturn.some(r => r.verb === (statusDelay ? 'plugins' : 'project.list')));
    for (const { pid } of later) assert.throws(() => process.kill(pid, 0), { code: 'ESRCH' }, `transport ${pid} still alive`);
  });
}

// The real status probe must release its socket as soon as its org deadline expires.
test('status probe closes its in-flight socket on cancellation', async t => {
  const dir = await mkdtemp(join(tmpdir(), 'sock-')), path = join(dir, 's');
  const server = createServer(socket => { socket.on('data', () => {}); });
  await new Promise(resolve => server.listen(path, resolve));
  t.after(async () => { await new Promise(resolve => server.close(resolve)); await rm(dir, { recursive: true, force: true }); });
  const controller = new AbortController();
  assert.equal(typeof orgBus.probeBusSocket, 'function');
  const connected = new Promise(resolve => server.once('connection', socket => socket.once('close', resolve)));
  const result = orgBus.probeBusSocket(path, { signal: controller.signal, deadline: Date.now() + 100 });
  await delay(30);
  controller.abort();
  assert.equal(await result, false);
  await Promise.race([connected, delay(250).then(() => { throw new Error('status socket still open'); })]);
});

test('workspace deadline reaches the organization status socket', async t => {
  const dir = await mkdtemp(join(tmpdir(), 'sock-')), path = join(dir, 's');
  let closed;
  const socketClosed = new Promise(resolve => { closed = resolve; });
  const server = createServer(socket => { socket.on('data', () => {}); socket.once('close', closed); });
  await new Promise(resolve => server.listen(path, resolve));
  t.after(async () => { await new Promise(resolve => server.close(resolve)); await rm(dir, { recursive: true, force: true }); });
  const organizations = organizationConsole({ ctx: {}, listOrgs: async () => ['good', 'slow'].map(id => ({ id, state: 'active' })),
    showOrg: async () => ({ doc: {} }), resolveOrg: async (_, id) => ({ id, state: 'active', root: dir }), orgPaths: () => ({ socket: path }),
    busStatus: async (_, id, bound) => id === 'good' ? { healthy: true } : { healthy: await orgBus.probeBusSocket(path, bound) },
    options: () => goodOptions() });
  const body = await readWorkspace({ organizations, session, timeoutMs: 150 });
  assert.equal(body.orgs.find(o => o.org === 'slow').state, 'timeout');
  assert.deepEqual(body.summary, { read: 1, of: 2 });
  await Promise.race([socketClosed, delay(250).then(() => { throw new Error('status socket still open'); })]);
});