/**
 * The hook broker.
 *
 * celilo keeps the database, the master key and the live capability objects.
 * The hook runs in its own process and reaches them only by asking. This is
 * the answering half; `hook-runner.ts` is the asking half.
 *
 * **It does not know what a capability is** (design D2). At handshake it sends
 * a shape descriptor built by walking the object `loadCapabilityFunctions`
 * already returns, and thereafter it dispatches `call` frames against that
 * same object by name. That works because every hook-facing capability method
 * is already `(request: JSON) => Promise<JSON>` — measured: twelve
 * capabilities, thirty-seven methods, no callbacks, no streams, no handles —
 * and because `wrapWithLogging` already treats a capability as an opaque table
 * of async methods keyed by name. This is that walk with a process in the
 * middle.
 *
 * Execution function (Rule 10.1) — owns the socket and performs the calls.
 */

import { mkdtempSync, rmSync } from 'node:fs';
import { type Server, type Socket, createServer } from 'node:net';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import {
  type BufferedStoreOp,
  type CapabilityShape,
  type ChildFrame,
  HOOK_PROTOCOL_VERSION,
  type HookError,
  createLineReader,
  encodeFrame,
  parseChildFrame,
  serializeError,
  versionMismatch,
} from './hook-protocol';
import type { HookStores } from './hook-store';
import type { HookLogger } from './types';

/** The answering half's view of one store frame. */
type StoreCallFrame = Extract<ChildFrame, { type: 'store' }>;

/**
 * A lazy provider of the module's two hook state stores.
 *
 * Lazy because building the stores reads the module row and its manifest, and
 * because the secrets store touches the master key only when an operation
 * actually needs it. A hook that never calls `context.secrets` should pay
 * none of that.
 */
export type HookStoresProvider = () => Promise<HookStores>;

/**
 * How long to wait after the child exits for its last frame to arrive. Short,
 * because by this point the writer is already gone and the bytes are either in
 * the buffer or they are never coming.
 */
const DRAIN_GRACE_MS = 1_000;

/** What the hook returned, or what it threw. */
export type HookOutcome =
  | { ok: true; outputs: Record<string, unknown> }
  | { ok: false; error: HookError };

export interface BrokerOptions {
  /** The live capability map, exactly as `loadCapabilityFunctions` built it. */
  capabilities: Record<string, unknown>;
  /** The serialisable half of `HookContext` — no `logger`, no `capabilities`. */
  context: Record<string, unknown>;
  /** Absolute path of the hook script, sent to the child rather than argv'd. */
  scriptPath: string;
  /** Where the hook's own log lines go. */
  logger: HookLogger;
  /** Called on every frame, so the idle timer measures the whole channel. */
  onActivity: () => void;
  /**
   * The module's hook-owned-state stores (hook-owned-state task 3.5). Absent
   * means this run has none wired, and a `store` frame from the hook is
   * REFUSED with an error naming the gap — a write that vanishes silently is
   * the exact defect the accessor exists to remove.
   */
  stores?: HookStoresProvider;
}

export interface Broker {
  /** Unix socket path, handed to the child in its environment. */
  socketPath: string;
  /**
   * Resolves once the child's connection has closed, so its last frame has
   * certainly been delivered.
   *
   * The child writes its terminal frame and then exits, and process exit is
   * what the executor waits on — but the two are different channels. The exit
   * can be observed while the frame is still sitting in the kernel buffer,
   * unread, which would look exactly like a hook that exited without
   * returning a result. Bounded, so a socket that never closes cannot hang
   * celilo waiting for a process that has already gone.
   */
  drained(): Promise<void>;
  /**
   * The hook's terminal frame, or `undefined` if it never sent one — which is
   * what a crash, a `process.exit` or a kill looks like from here. Read after
   * the child has exited, so there is no race to lose.
   */
  outcome(): HookOutcome | undefined;
  /** Every protocol-level complaint, for the error message when one matters. */
  faults(): string[];
  /** Refuse further calls. Idempotent. Called on kill and again on cleanup. */
  stop(): void;
  /** Close the socket and remove its directory. */
  close(): void;
}

/**
 * Walk a capability map into the descriptor the child rebuilds proxies from.
 *
 * Function-valued keys become `methods`; everything else is copied into `data`,
 * which is where `stampProvider` puts `providerModuleId` — a hook reads it to
 * name the provider that could not supply what it asked for.
 *
 * Own string keys only, the way `wrapWithLogging` walks. Every capability the
 * loader builds is a plain object (a spread, an `Object.assign`, or
 * `Object.create` plus own properties), so a prototype walk would find nothing
 * a hook can call today — and if that ever stopped being true, auto-logging
 * would break in the same breath, which is a louder signal than this would be.
 * Symbol keys cannot cross JSON and are dropped.
 *
 * Planning function (Rule 10.4) — pure, so the descriptor is testable without
 * a socket.
 */
export function capabilityShape(
  capabilities: Record<string, unknown>,
): Record<string, CapabilityShape> {
  const shape: Record<string, CapabilityShape> = {};

  for (const [name, capability] of Object.entries(capabilities)) {
    if (!capability || typeof capability !== 'object') continue;

    const methods: string[] = [];
    const data: Record<string, unknown> = {};

    for (const key of Reflect.ownKeys(capability)) {
      if (typeof key !== 'string') continue;
      const value = (capability as Record<string, unknown>)[key];
      if (typeof value === 'function') {
        methods.push(key);
      } else if (isJsonSafe(value)) {
        data[key] = value;
      }
    }

    shape[name] = { methods, data };
  }

  return shape;
}

/** Everything `JSON.stringify` round-trips without inventing or losing a value. */
function isJsonSafe(value: unknown): boolean {
  if (value === null) return true;
  const type = typeof value;
  if (type === 'string' || type === 'boolean') return true;
  if (type === 'number') return Number.isFinite(value as number);
  if (type !== 'object') return false;
  try {
    return JSON.parse(JSON.stringify(value)) !== undefined;
  } catch {
    return false;
  }
}

/**
 * Bind a Unix socket in a fresh per-run directory and answer one hook.
 *
 * The socket, not stdout: seventeen module script files spawn subprocesses,
 * and a grandchild writing to fd 1 would corrupt the frame stream (design D3).
 * The directory is short-named because `sun_path` is 104 bytes on macOS and
 * the platform temp directory already spends half of it.
 */
export async function startBroker(options: BrokerOptions): Promise<Broker> {
  const directory = mkdtempSync(join(tmpdir(), 'celilo-hook-'));
  const socketPath = join(directory, 's');

  let outcome: HookOutcome | undefined;
  let stopped = false;
  const faults: string[] = [];
  let connection: Socket | undefined;
  // Already resolved: no connection means nothing left to deliver.
  let connectionClosed: Promise<void> = Promise.resolve();
  // The module's stores, resolved from the provider on first use.
  let storesCache: HookStores | undefined;

  const server: Server = createServer((socket) => {
    if (connection) {
      // One hook, one connection. A second is either a bug or a hook trying
      // to hold the channel open past its own run.
      faults.push('a second connection to the hook socket was refused');
      socket.destroy();
      return;
    }
    connection = socket;
    connectionClosed = new Promise<void>((resolve) => socket.once('close', () => resolve()));
    socket.setEncoding('utf-8');

    const send = (frame: Parameters<typeof encodeFrame>[0]) => {
      if (!socket.destroyed) socket.write(encodeFrame(frame));
    };

    const feed = createLineReader((line) => {
      options.onActivity();

      const parsed = parseChildFrame(line);
      if (!parsed.ok) {
        faults.push(`malformed frame from the hook: ${parsed.error}`);
        return;
      }

      const frame = parsed.frame;
      switch (frame.type) {
        case 'ready': {
          const mismatch = versionMismatch(frame.protocolVersion, 'the hook runner');
          if (mismatch) {
            faults.push(mismatch);
            socket.destroy();
            return;
          }
          send({
            type: 'context',
            protocolVersion: HOOK_PROTOCOL_VERSION,
            scriptPath: options.scriptPath,
            context: options.context,
          });
          send({ type: 'capabilities', shape: capabilityShape(options.capabilities) });
          return;
        }

        case 'log':
          options.logger[frame.level](frame.message);
          return;

        case 'call':
          void dispatch(frame.id, frame.capability, frame.method, frame.args, send);
          return;

        case 'store':
          void dispatchStore(frame, send);
          return;

        case 'result':
          outcome = { ok: true, outputs: frame.outputs };
          return;

        case 'throw':
          outcome = { ok: false, error: frame.error };
          return;
      }
    });

    socket.on('data', feed);
    socket.on('error', (error) => faults.push(`hook socket error: ${error.message}`));
  });

  async function dispatch(
    id: string,
    capabilityName: string,
    method: string,
    args: unknown[],
    send: (frame: Parameters<typeof encodeFrame>[0]) => void,
  ): Promise<void> {
    if (stopped) {
      // The run is over — killed at its timeout, most likely. Refusing here is
      // what makes the kill real: the whole failure celilo#1003 describes is a
      // hook still registering DNS records after celilo moved on.
      send({
        type: 'throw',
        id,
        error: {
          name: 'Error',
          message: `Hook run has ended; refusing ${capabilityName}.${method}.`,
        },
      });
      return;
    }

    const capability = options.capabilities[capabilityName];
    const fn =
      capability && typeof capability === 'object'
        ? (capability as Record<string, unknown>)[method]
        : undefined;

    if (typeof fn !== 'function') {
      send({
        type: 'throw',
        id,
        error: {
          name: 'TypeError',
          message: `Capability '${capabilityName}' has no method '${method}'.`,
        },
      });
      return;
    }

    try {
      const value = await (fn as (...a: unknown[]) => unknown).apply(capability, args);
      options.onActivity();
      send({ type: 'return', id, value: value === undefined ? null : value });
    } catch (error) {
      options.onActivity();
      send({ type: 'throw', id, error: serializeError(error) });
    }
  }

  await new Promise<void>((resolve, reject) => {
    server.once('error', reject);
    server.listen(socketPath, resolve);
  });

  /**
   * Answer one hook-owned-state store call.
   *
   * The stores resolve on first use and stay cached for the run. Everything
   * that can fail — an undeclared name, a missing module row, a store that was
   * never wired — answers as a `throw` frame rather than crashing the broker,
   * so the hook's own `try`/`catch` sees it the way an in-process throw would
   * behave.
   */
  async function dispatchStore(
    frame: StoreCallFrame,
    send: (frame: Parameters<typeof encodeFrame>[0]) => void,
  ): Promise<void> {
    if (stopped) {
      send({
        type: 'throw',
        id: frame.id,
        error: {
          name: 'Error',
          message: `Hook run has ended; refusing context.${frame.store}.${frame.method}.`,
        },
      });
      return;
    }

    try {
      if (!options.stores) {
        throw new Error(
          `No hook state stores were wired for this run, so context.${frame.store} is unavailable. This is a celilo bug: every hook invocation must receive both stores (hook-owned-state task 3.5).`,
        );
      }
      storesCache ??= await options.stores();
      const backend = storesCache[frame.store];

      let value: unknown;
      switch (frame.method) {
        case 'get':
          value = await backend.get(frame.args[0] as string);
          break;
        case 'set':
          await backend.set(frame.args[0] as string, frame.args[1] as string);
          value = null;
          break;
        case 'delete':
          await backend.delete(frame.args[0] as string);
          value = null;
          break;
        case 'transaction':
          await backend.applyTransaction(frame.args[0] as BufferedStoreOp[]);
          value = null;
          break;
      }
      options.onActivity();
      send({ type: 'return', id: frame.id, value });
    } catch (error) {
      options.onActivity();
      send({ type: 'throw', id: frame.id, error: serializeError(error) });
    }
  }

  return {
    socketPath,
    drained: () =>
      Promise.race([
        connectionClosed,
        new Promise<void>((resolve) => setTimeout(resolve, DRAIN_GRACE_MS).unref()),
      ]),
    outcome: () => outcome,
    faults: () => [...faults],
    stop: () => {
      stopped = true;
    },
    close: () => {
      stopped = true;
      connection?.destroy();
      server.close();
      rmSync(directory, { recursive: true, force: true });
    },
  };
}