/**
 * Unix-socket server test — saga tb4593.3 (PROTOCOL §7.1, §9.1 hello-on-connect).
 *
 * Connects a REAL `net.createConnection` client to a `SocketServer` over a tmp socket
 * path. Hermetic: tmp dir, no network, no daemon, no agent. Verifies the framing reuse
 * (LF-only JSONL, U+2028/U+2029-safe) and the hello-on-connect handshake against the
 * `HelloSchema` shape.
 *
 * Coverage:
 *   - start() binds; a client connects; `hello` arrives first (§9.1) + validates schema.
 *   - onConnect fires with a Connection; `send()` writes JSONL to the client.
 *   - `messages()` yields inbound JSONL records; a multi-line chunk splits correctly.
 *   - a partial record split across chunks reassembles (streaming JsonlDecoder).
 *   - a U+2028 inside a JSON string survives intact (the readline bug, §2.1).
 *   - a malformed JSON line yields a record with `parseError` (no stream abort, §8).
 *   - stale-socket removal at boot (a leftover file at the path doesn't block bind).
 *   - stop() unlinks the socket + destroys connections; clients observe EOF.
 *   - connections() async iterator yields accepted connections.
 */
import { describe, expect, it, beforeEach, afterEach } from 'bun:test';
import { mkdtempSync, rmSync, writeFileSync, existsSync } from 'node:fs';
import { join } from 'node:path';
import { tmpdir } from 'node:os';
import { createConnection } from 'node:net';
import { Value } from '@sinclair/typebox/value';
import { HelloSchema } from '@/protocol/lifecycle';
import { encodeJsonl, splitJsonl } from '@/protocol/framing/jsonl';import { SocketServer, type Connection } from '@/daemon/socket-server';

let dir: string;
let sockPath: string;

beforeEach(() => {
  dir = mkdtempSync(join(tmpdir(), 'shard-sock-'));
  sockPath = join(dir, 'daemon.sock');
});

afterEach(() => {
  rmSync(dir, { recursive: true, force: true });
});

/** Promise-resolved on the next `length` LF-delimited records from a socket. */
function readN(socket: NodeJS.ReadableStream, n: number): Promise<string[]> {
  return new Promise((resolve) => {
    const lines: string[] = [];
    let buf = '';
    const onData = (chunk: Buffer) => {
      buf += chunk.toString('utf8');
      while (buf.includes('\n')) {
        const nl = buf.indexOf('\n');
        const line = buf.slice(0, nl);
        if (line.length > 0) lines.push(line);
        buf = buf.slice(nl + 1);
      }
      if (lines.length >= n) {
        socket.off('data', onData);
        resolve(lines);
      }
    };
    socket.on('data', onData);
  });
}

/** Connect a raw client + collect the first record (the hello). */
async function connectAndHello(server: SocketServer): Promise<{
  sock: ReturnType<typeof createConnection>;
  hello: unknown;
}> {
  const sock = createConnection({ path: server.path });
  // Wait for the socket to be writable.
  await new Promise<void>((resolve) => sock.once('connect', () => resolve()));
  const [helloLine] = await readN(sock, 1).then((lines) => [JSON.parse(lines[0]!)]);
  return { sock, hello: helloLine };
}
describe('SocketServer — hello + framing (tb4593.3)', () => {
  it('binds and emits hello on connect matching HelloSchema (§9.1)', async () => {
    const server = new SocketServer({ socketPath: sockPath });
    await server.start();

    const { hello } = await connectAndHello(server);
    expect(Value.Check(HelloSchema, hello)).toBe(true);
    expect((hello as { type: string }).type).toBe('hello');
    expect((hello as { server_version: string }).server_version).toContain('shard');

    await server.stop();
  });

  it('onConnect fires with a Connection after hello is sent', async () => {
    const server = new SocketServer({ socketPath: sockPath });
    await server.start();
    const seen: Connection[] = [];
    server.onConnect((conn) => seen.push(conn));

    await connectAndHello(server);
    await new Promise((r) => setTimeout(r, 10)); // microtask flush

    expect(seen).toHaveLength(1);
    expect(seen[0]?.closed).toBe(false);

    await server.stop();
  });

  it('drops closed clients from live connection tracking', async () => {
    const server = new SocketServer({ socketPath: sockPath });
    await server.start();
    const { sock } = await connectAndHello(server);
    expect(server.liveConnections()).toHaveLength(1);

    const closed = new Promise<void>((resolve) => sock.once('close', () => resolve()));
    sock.destroy();
    await closed;
    await new Promise((resolve) => setTimeout(resolve, 10));
    expect(server.liveConnections()).toHaveLength(0);

    await server.stop();
  });

  it('send() writes a JSONL message to the client', async () => {
    const server = new SocketServer({ socketPath: sockPath });
    await server.start();
    let conn: Connection | undefined;
    server.onConnect((c) => {
      conn = c;
    });

    const { sock } = await connectAndHello(server);
    await new Promise((r) => setTimeout(r, 10));

    await conn!.send({ type: 'status', status: 'idle' });
    const [line] = await readN(sock, 1).then((ls) => [JSON.parse(ls[0]!)]);
    expect(line).toEqual({ type: 'status', status: 'idle' });

    await server.stop();
  });

  it('messages() yields inbound JSONL records (multi-line chunk splits)', async () => {
    const server = new SocketServer({ socketPath: sockPath });
    await server.start();
    let conn: Connection | undefined;
    server.onConnect((c) => {
      conn = c;
    });
    const { sock } = await connectAndHello(server);
    await new Promise((r) => setTimeout(r, 10));

    const iter = conn!.messages();
    // Send two records in one write — JsonlDecoder splits on \n.
    sock.write(encodeJsonl({ type: 'agent.list' }) + encodeJsonl({ type: 'agent.status' }));

    const r1 = (await iter.next()).value;
    const r2 = (await iter.next()).value;
    expect(r1?.value).toEqual({ type: 'agent.list' });
    expect(r2?.value).toEqual({ type: 'agent.status' });

    await server.stop();
  });

  it('reassembles a record split across chunks (streaming decoder)', async () => {
    const server = new SocketServer({ socketPath: sockPath });
    await server.start();
    let conn: Connection | undefined;
    server.onConnect((c) => {
      conn = c;
    });
    const { sock } = await connectAndHello(server);
    await new Promise((r) => setTimeout(r, 10));

    const iter = conn!.messages();
    // Write the first half, then the second half + terminator — one record total.
    const full = encodeJsonl({ type: 'prompt', message: 'hello world' });
    const mid = Math.floor(full.length / 2);
    sock.write(full.slice(0, mid));
    sock.write(full.slice(mid));

    const r = (await iter.next()).value;
    expect(r?.value).toEqual({ type: 'prompt', message: 'hello world' });

    await server.stop();
  });

  it('survives U+2028 inside a JSON string (§2.1 — the readline bug)', async () => {
    const server = new SocketServer({ socketPath: sockPath });
    await server.start();
    let conn: Connection | undefined;
    server.onConnect((c) => {
      conn = c;
    });
    const { sock } = await connectAndHello(server);
    await new Promise((r) => setTimeout(r, 10));

    const iter = conn!.messages();
    const payload = { type: 'prompt', message: 'line\u2028break' }; // U+2028 inside the string
    sock.write(encodeJsonl(payload));

    const r = (await iter.next()).value;
    expect(r?.value).toEqual(payload);
    // `splitJsonl` confirms a single line — readline would have split it into two.
    expect(splitJsonl(encodeJsonl(payload))).toHaveLength(1);

    await server.stop();
  });

  it('yields a record with parseError on malformed JSON (no stream abort)', async () => {
    const server = new SocketServer({ socketPath: sockPath });
    await server.start();
    let conn: Connection | undefined;
    server.onConnect((c) => {
      conn = c;
    });
    const { sock } = await connectAndHello(server);
    await new Promise((r) => setTimeout(r, 10));

    const iter = conn!.messages();
    sock.write('{ not valid json }\n');

    const r = (await iter.next()).value;
    expect(r?.value).toBeUndefined();
    expect(typeof r?.parseError).toBe('string');

    await server.stop();
  });

  it('removes a stale socket file at boot (does not block bind)', async () => {
    // Leave a stale file at the socket path — a crashed-daemon leftover.
    writeFileSync(sockPath, 'stale');
    expect(existsSync(sockPath)).toBe(true);

    const server = new SocketServer({ socketPath: sockPath });
    await server.start(); // unlinks stale + binds
    expect(existsSync(sockPath)).toBe(true); // now a real socket, not the stale file

    const { hello } = await connectAndHello(server);
    expect((hello as { type: string }).type).toBe('hello');

    await server.stop();
  });

  it('refuses to unlink and replace a live daemon socket', async () => {
    const first = new SocketServer({ socketPath: sockPath });
    const second = new SocketServer({ socketPath: sockPath });
    await first.start();

    await expect(second.start()).rejects.toMatchObject({ code: 'EADDRINUSE' });

    // The failed second startup must leave the original pathname and listener intact.
    const { hello } = await connectAndHello(first);
    expect((hello as { type: string }).type).toBe('hello');
    await first.stop();
  });

  it('stop() unlinks the socket file and destroys live connections', async () => {
    const server = new SocketServer({ socketPath: sockPath });
    await server.start();
    const { sock } = await connectAndHello(server);
    await new Promise((r) => setTimeout(r, 10));

    await server.stop();

    // The socket file is gone.
    expect(existsSync(sockPath)).toBe(false);
    // The client observed EOF/closed.
    await new Promise<void>((resolve) => {
      if (sock.destroyed) return resolve();
      sock.once('close', () => resolve());
    });
    expect(sock.destroyed).toBe(true);
  });

  it('connections() async iterator yields accepted connections', async () => {
    const server = new SocketServer({ socketPath: sockPath });
    await server.start();
    const it = server.connections();

    // Connect a client; the iterator should produce it.
    const connectP = new Promise<void>((resolve) => {
      const sock = createConnection({ path: server.path });
      sock.once('connect', () => resolve());
      void sock;
    });
    const connP = it.next();
    await connectP;
    const { value: conn } = await connP;
    expect(conn).toBeDefined();
    expect(conn?.closed).toBe(false);

    await server.stop();
  });
});