/**
 * The remote-ops broker, tested through the REAL asking half (tasks 5.1/5.2,
 * design D12).
 *
 * Every bridged case below runs the actual `@celilo/capabilities` primitive
 * with `CELILO_HOOK_REMOTE_SOCKET` set — in a SEPARATE process
 * (`test-fixtures/remote-bridge-probe.ts`), because that is the only topology
 * that exists: the bridge blocks its caller in spawnSync while celilo's event
 * loop answers the socket, so asking and answering can never share a process.
 * The round trip therefore exercises what a hook exercises: the primitive's
 * bridge detection, the spawned client, the socket, the Zod parse, the
 * policy, and the broker-side primitive — with the one seam a unit test must
 * inject, the runner that would otherwise run ssh.
 */

import { afterEach, beforeEach, describe, expect, test } from 'bun:test';
import { existsSync, mkdirSync, mkdtempSync, realpathSync, rmSync, writeFileSync } from 'node:fs';
import { connect } from 'node:net';
import { tmpdir } from 'node:os';
import { join, resolve } from 'node:path';
import { type MockRunner, type RunResult, createMockRunner } from '@celilo/capabilities';
import { createCapturingLogger } from './logger';
import { type RemoteAccessPolicy, type RemoteBroker, startRemoteBroker } from './remote-broker';

const REMOTE_SOCKET_ENV = 'CELILO_HOOK_REMOTE_SOCKET';
const PROBE = resolve(__dirname, 'test-fixtures/remote-bridge-probe.ts');

const ALLOW_ALL: RemoteAccessPolicy = {
  moduleId: 'caddy',
  checkTarget: () => ({ allowed: true }),
};

const DENY_ALL: RemoteAccessPolicy = {
  moduleId: 'caddy',
  checkTarget: () => ({ allowed: false, message: 'policy says no' }),
};

/**
 * The shape of the real policy's first rule (`services/remote-access.ts`):
 * a request on the module's own credential is allowed, everything else is
 * refused. Used to prove the broker computes and forwards
 * `hasOwnCredential` — the decision itself is the policy's.
 */
const OWN_CREDENTIAL_ONLY: RemoteAccessPolicy = {
  moduleId: 'generic-cpanel-hosting-provider',
  checkTarget: (_target, hasOwnCredential) =>
    hasOwnCredential ? { allowed: true } : { allowed: false, message: 'fleet key refused' },
};

describe('remote-ops broker', () => {
  let scratch: string;
  let socketDir: string;
  let stateDir: string;
  let broker: RemoteBroker | undefined;
  let runner: MockRunner;

  beforeEach(() => {
    scratch = mkdtempSync(join(tmpdir(), 'celilo-remote-broker-'));
    // Short-named like the real run directory: sun_path is 104 bytes on macOS.
    socketDir = mkdtempSync(join(tmpdir(), 'celilo-hook-'));
    stateDir = join(scratch, 'state');
    mkdirSync(stateDir, { recursive: true });
    runner = createMockRunner([
      { match: 'ssh', result: { ok: true, stdout: 'remote-ok', stderr: '' } },
    ]);
  });

  afterEach(() => {
    broker?.close();
    broker = undefined;
    rmSync(scratch, { recursive: true, force: true });
    rmSync(socketDir, { recursive: true, force: true });
  });

  async function start(policy: RemoteAccessPolicy | undefined): Promise<void> {
    const { logger } = createCapturingLogger();
    broker = await startRemoteBroker({
      socketDir,
      policy,
      readableRoots: [scratch],
      writableRoots: [stateDir],
      logger,
      onActivity: () => {},
      runner: runner.run,
    });
  }

  /** Run one primitive from a hook-shaped process; return what it returned. */
  async function fromHookProcess<T = RunResult>(instruction: Record<string, unknown>): Promise<T> {
    const proc = Bun.spawn({
      cmd: [process.execPath, PROBE, JSON.stringify(instruction)],
      env: { ...process.env, [REMOTE_SOCKET_ENV]: (broker as RemoteBroker).socketPath },
      stdout: 'pipe',
      stderr: 'pipe',
    });
    const [stdout, stderr] = await Promise.all([
      new Response(proc.stdout).text(),
      new Response(proc.stderr).text(),
    ]);
    const code = await proc.exited;
    if (code !== 0) throw new Error(`probe exited ${code}: ${stderr}`);
    return JSON.parse(stdout) as T;
  }

  test('bridges remoteExec end to end, and the broker builds the ssh line itself', async () => {
    await start(ALLOW_ALL);

    const result = await fromHookProcess({
      op: 'remoteExec',
      target: { ipv4_address: '10.0.10.10' },
      command: 'echo hi',
    });

    expect(result).toEqual({ ok: true, stdout: 'remote-ok', stderr: '' });
    expect(runner.calls).toHaveLength(1);
    expect(runner.calls[0]?.cmd).toContain('root@10.0.10.10');
    expect(runner.calls[0]?.cmd).toContain("'echo hi'");
  });

  test('sugar primitives ride the bridge with no awareness of their own', async () => {
    await start(ALLOW_ALL);

    const result = await fromHookProcess({
      op: 'serviceCtl',
      target: { ipv4_address: '10.0.10.10' },
      unit: 'caddy',
      action: 'restart',
    });

    expect(result.ok).toBe(true);
    expect(runner.calls[0]?.cmd).toContain('systemctl restart caddy');
  });

  test('a policy refusal comes back as the failed RunResult, and nothing runs', async () => {
    await start(DENY_ALL);

    const result = await fromHookProcess({
      op: 'remoteExec',
      target: { ipv4_address: '10.0.30.7' },
      command: 'id -un',
    });

    expect(result.ok).toBe(false);
    expect(result.stderr).toContain('policy says no');
    expect(runner.calls).toHaveLength(0);
  });

  test('no policy means deny, naming the gap (Rule 6.4)', async () => {
    await start(undefined);

    const result = await fromHookProcess({
      op: 'remoteExec',
      target: { ipv4_address: '10.0.10.10' },
      command: 'echo hi',
    });

    expect(result.ok).toBe(false);
    expect(result.stderr).toContain('without a remote-access policy');
    expect(runner.calls).toHaveLength(0);
  });

  test('streamBackup writes only inside the writable roots', async () => {
    await start(ALLOW_ALL);

    const inside = await fromHookProcess({
      op: 'streamBackup',
      target: { ipv4_address: '10.0.10.10' },
      producerCommand: 'cat /etc/x',
      localPath: join(stateDir, 'x'),
    });
    expect(inside.ok).toBe(true);
    // The broker realpaths before it binds the containment decision to the
    // command, so assert on the resolved path (macOS: /var → /private/var).
    expect(runner.calls[0]?.cmd).toContain(join(realpathSync(stateDir), 'x'));

    const outside = await fromHookProcess({
      op: 'streamBackup',
      target: { ipv4_address: '10.0.10.10' },
      producerCommand: 'cat /etc/x',
      localPath: join(scratch, 'not-writable'),
    });
    expect(outside.ok).toBe(false);
    expect(outside.stderr).toContain('writable directories');
    expect(runner.calls).toHaveLength(1);
  });

  test('streamBackup refuses a symlink leaf inside a writable root', async () => {
    // The shell's `>` follows a link, so `state/evil` → anywhere would aim
    // celilo's write outside the roots. Inside the jail the link can be
    // dangling (the target path does not exist THERE) and still resolve on
    // the broker's side — which is exactly why the leaf is refused rather
    // than resolved.
    await start(ALLOW_ALL);
    const { symlinkSync } = await import('node:fs');
    symlinkSync('/var/celilo/master.key', join(stateDir, 'evil'));

    const result = await fromHookProcess({
      op: 'streamBackup',
      target: { ipv4_address: '10.0.10.10' },
      producerCommand: 'cat /etc/x',
      localPath: join(stateDir, 'evil'),
    });

    expect(result.ok).toBe(false);
    expect(result.stderr).toContain('symlink');
    expect(runner.calls).toHaveLength(0);
  });

  test('streamRestore reads only inside the readable roots', async () => {
    await start(ALLOW_ALL);
    const source = join(scratch, 'payload.tar');
    writeFileSync(source, 'bytes');

    const inside = await fromHookProcess({
      op: 'streamRestore',
      target: { ipv4_address: '10.0.10.10' },
      localPath: source,
      consumerCommand: 'tar -xf -',
    });
    expect(inside.ok).toBe(true);

    const outside = await fromHookProcess({
      op: 'streamRestore',
      target: { ipv4_address: '10.0.10.10' },
      localPath: '/etc/passwd',
      consumerCommand: 'cat',
    });
    expect(outside.ok).toBe(false);
    expect(outside.stderr).toContain('readable directories');
  });

  test('an identityFile crosses as content, is materialised for the call, and is removed', async () => {
    // A target carrying its own credential reaches the policy with
    // hasOwnCredential=true, which the real policy's first rule allows.
    await start(OWN_CREDENTIAL_ONLY);
    const keyPath = join(scratch, 'own-key');
    writeFileSync(keyPath, 'not-a-real-key\n');

    const result = await fromHookProcess({
      op: 'remoteExec',
      target: { ipv4_address: '198.51.100.7', user: 'peba-hosting', identityFile: keyPath },
      command: 'echo hi',
    });

    expect(result.ok).toBe(true);
    const cmd = runner.calls[0]?.cmd ?? '';
    const materialized = cmd.match(/-i (\S*celilo-hook-identity-[^ ]*)/)?.[1];
    expect(materialized).toBeDefined();
    expect(materialized).not.toBe(keyPath);
    expect(existsSync(materialized as string)).toBe(false);
  });

  test("installAuthorizedKey is brokered and its password counts as the module's own credential", async () => {
    runner = createMockRunner([
      { match: 'ssh-copy-id', result: { ok: true, stdout: 'installed', stderr: '' } },
    ]);
    // The password IS the credential: the broker reports hasOwnCredential
    // and the policy's own-credential rule is what allows the call.
    await start(OWN_CREDENTIAL_ONLY);

    const result = await fromHookProcess({
      op: 'installAuthorizedKey',
      target: { ipv4_address: '198.51.100.7', user: 'peba-hosting', port: 7822 },
      password: 'hunter2',
    });

    expect(result.ok).toBe(true);
    expect(runner.calls[0]?.cmd).toContain('ssh-copy-id');
    expect(runner.calls[0]?.cmd).toContain('peba-hosting@198.51.100.7');
    // The password rides the child environment, never the command line.
    expect(runner.calls[0]?.cmd).not.toContain('hunter2');
  });

  test('probeHttp asks the broker about the target and honours a refusal', async () => {
    await start(DENY_ALL);

    const refused = await fromHookProcess<{ healthy: boolean; failure?: string; detail: string }>({
      op: 'probeHttpRefused',
      target: { ipv4_address: '10.0.10.10' },
    });

    expect(refused.healthy).toBe(false);
    expect(refused.failure).toBe('refused');
    expect(refused.detail).toContain('policy says no');
  });

  test('probeHttp fetches when the broker allows the target', async () => {
    await start(ALLOW_ALL);

    const result = await fromHookProcess<{ healthy: boolean; detail: string }>({
      op: 'probeHttpAllowed',
      target: { ipv4_address: '10.0.10.10' },
    });

    expect(result).toEqual({ healthy: true, detail: 'GET http://10.0.10.10:80/ → 200' });
  });

  test('opts.env is refused on the asking side before anything crosses', async () => {
    await start(ALLOW_ALL);

    const result = await fromHookProcess({
      op: 'remoteExec',
      target: { ipv4_address: '10.0.10.10' },
      command: 'echo hi',
      opts: { env: { LD_PRELOAD: '/tmp/evil.so' } },
    });

    expect(result.ok).toBe(false);
    expect(result.stderr).toContain('opts.env does not cross');
    expect(runner.calls).toHaveLength(0);
  });

  test('stop() refuses further operations, which is what makes a kill real', async () => {
    await start(ALLOW_ALL);
    broker?.stop();

    const result = await fromHookProcess({
      op: 'remoteExec',
      target: { ipv4_address: '10.0.10.10' },
      command: 'echo hi',
    });

    expect(result.ok).toBe(false);
    expect(result.stderr).toContain('Hook run has ended');
    expect(runner.calls).toHaveLength(0);
  });

  test('a malformed request line is answered, not hung on', async () => {
    await start(ALLOW_ALL);

    const answer = await new Promise<string>((resolvePromise, reject) => {
      const socket = connect((broker as RemoteBroker).socketPath);
      socket.setEncoding('utf-8');
      let buffer = '';
      socket.on('connect', () => socket.write('this is not JSON\n'));
      socket.on('data', (chunk: string) => {
        buffer += chunk;
        if (buffer.includes('\n')) {
          socket.end();
          resolvePromise(buffer);
        }
      });
      socket.on('error', reject);
    });

    const parsed = JSON.parse(answer) as { ok: boolean; stderr: string };
    expect(parsed.ok).toBe(false);
    expect(parsed.stderr).toContain('malformed request');
  });
});