/**
 * Remote bus responder (`kind: 'daemon'`) — Slice 3.
 *
 * Runs in the `api-serve` parent and bridges the event bus to the wire: watches
 * the generic `interview.required.*` family, forwards each question to the
 * remote client via an injected `ask` round-trip, and replies on the bus with
 * the client's answer. Also answers `responder.probe` so the (non-TTY) command
 * child doesn't fail-fast for lack of a responder.
 *
 * The deploy-specific families (config / secret / ensure / aspect) reuse this
 * same plumbing with per-family adapters — tracked as a follow-up.
 */

import { type Bus, defineEvents, openBus } from '@celilo/event-bus';
import type { InterviewRequiredPayload } from './bus-interview';
import { RESPONDER_PROBE_EVENT } from './responder-probe';

const NO_SCHEMAS = defineEvents({});

/** The normalized question handed to the wire (mirrors InterviewMessage). */
export interface WireInterview {
  id: string;
  /** The question's stable identity, so a client can pre-stage `<scope>.<key>`. */
  scope: string;
  key: string;
  kind: InterviewRequiredPayload['kind'];
  message: string;
  description?: string;
  defaultValue?: string;
  placeholder?: string;
  options?: Array<{ value: string; label: string; hint?: string }>;
  required?: boolean;
}

export interface RemoteResponderOptions {
  /** Path to the shared bus sqlite db (command child + responder share it). */
  busDbPath: string;
  /** Forward a question to the client and resolve with its answer. */
  ask: (interview: WireInterview) => Promise<unknown>;
  /** `emittedBy` audit label on replies. Defaults to `daemon`. */
  emittedBy?: string;
}

export interface RemoteResponderHandle {
  close(): void;
}

export function startRemoteResponder(opts: RemoteResponderOptions): RemoteResponderHandle {
  const me = opts.emittedBy ?? 'daemon';
  const bus: Bus = openBus({ dbPath: opts.busDbPath, events: NO_SCHEMAS });

  const interviewWatch = bus.watch('interview.required.*.*', async (event) => {
    if (event.replyFor !== null) return;

    const payload = event.payload as InterviewRequiredPayload;
    if (!payload || typeof payload.scope !== 'string' || typeof payload.key !== 'string') {
      return;
    }

    let value: unknown;
    try {
      value = await opts.ask({
        id: String(event.id),
        scope: payload.scope,
        key: payload.key,
        kind: payload.kind,
        message: payload.message,
        description: payload.description,
        defaultValue: payload.defaultValue,
        placeholder: payload.placeholder,
        options: payload.options,
        required: payload.required,
      });
    } catch (err) {
      // The client had no way to answer — so we emit NOTHING. A reply of any
      // shape consumes the query (it carries `replyFor: event.id`), destroying
      // a question nobody has answered yet; the asking command then dies with
      // it and no other responder can ever act. Leaving it unanswered parks the
      // command instead, which is what `busInterview`'s `timeoutMs: 0` is for.
      // The client learns it is parked from the `blocked` wire message.
      process.stderr.write(
        `[remote-responder] parked ${event.type} (#${event.id}): ${
          err instanceof Error ? err.message : String(err)
        }\n`,
      );
      return;
    }

    bus.emitRaw(`${event.type}.reply`, { value }, { replyFor: event.id, emittedBy: me });
  });

  // Liveness probe: the non-TTY command child emits `responder.probe` before
  // asking, to confirm someone is listening. Answer with our kind.
  const probeWatch = bus.watch(RESPONDER_PROBE_EVENT, async (event) => {
    if (event.replyFor !== null) return;
    bus.emitRaw(
      `${event.type}.reply`,
      { kind: 'daemon', emittedBy: me },
      { replyFor: event.id, emittedBy: me },
    );
  });

  return {
    close: () => {
      interviewWatch.close();
      probeWatch.close();
      bus.close();
    },
  };
}