/**
 * The notification-transport interview responder.
 *
 * Watches `interview.required.*` (and the config/ensure/aspect families), sends
 * the question to a route, and publishes the operator's reply back to the
 * waiting deploy via `bus.reply`.
 *
 * QUERIES the events table rather than using `bus.watch`, for two reasons —
 * neither of which is that watch cannot see other processes. (It can: each
 * watch runs a 250ms polling loop alongside its in-process callback.)
 *
 *   1. A watch seeds its high-water mark to MAX(id) at registration, so it
 *      only ever sees events emitted AFTER it starts. A deploy that raised its
 *      question thirty seconds before this poll fires would be invisible —
 *      and that is the normal case, not the edge one.
 *   2. `celilo alerts poll` is a one-shot process. A watch needs the process
 *      to stay alive to be worth anything; this one exits immediately.
 *
 * So "what is still unanswered" has to be a question asked of the table, not a
 * subscription to the future.
 *
 * The "ask" takes minutes on someone's phone rather than milliseconds on a
 * TTY. That is fine: the interview flow has no timeout by design ("the deploy
 * waits indefinitely for a responder"), which is what makes a phone a viable
 * answering device at all.
 *
 * Additive, never exclusive: a terminal responder wins when stdin is a TTY, so
 * a deploy started from a laptop is answered there rather than appearing to
 * hang while waiting for a text.
 */

import { type Bus, defineEvents, openBus } from '@celilo/event-bus';
import { and, eq, gt, isNull } from 'drizzle-orm';
import type { DbClient } from '../../db/client';
import { type Route, notificationDeliveries } from '../../db/schema';
import {
  composeInterviewBody,
  decideInterviewDelivery,
  describeQuestion,
} from './interview-responder';
import type { NotificationTransport } from './notifier';
import { consumeDelivery, mintDelivery } from './tokens';

const NO_SCHEMAS = defineEvents({});

/**
 * How long a question must go unanswered before it is worth someone's phone.
 *
 * An attached responder — a terminal, or the wire bridge in `api-serve` —
 * answers within seconds. Paging a phone about a question somebody is already
 * looking at is noise, and worse, both responders would reply to the same
 * question. Waiting lets the responder that is actually present win, and only
 * escalates to a messaging transport when nobody is there.
 *
 * This is the same rule as "a terminal wins when stdin is a TTY" (D12), applied
 * to responders this process cannot see.
 */
const ASK_AFTER_MS = 90_000;

/** Families a deploy can block on. `secret.*` is watched so it can be REFUSED. */
const INTERVIEW_PATTERNS = [
  'interview.required.*.*',
  'config.required.*.*',
  'ensure.required.*.*',
  'aspect.required.*.*',
  'secret.required.*.*',
] as const;

export interface NotificationResponderOptions {
  db: DbClient;
  busDbPath: string;
  /** Ack-capable routes, in the order questions should be offered. */
  routes: Route[];
  transportFor(route: Route): NotificationTransport;
  /** True when a terminal responder is available (D12). */
  hasTty(): boolean;
  now(): Date;
  ttlMs: number;
  /**
   * Grace before a question is escalated to a messaging transport, so a
   * responder that is actually attached answers first. Defaults to 90s.
   */
  askAfterMs?: number;
  /** Called when a question is declined, so the reason is not lost. */
  onDeclined?(eventType: string, reason: string): void;
  /** Called when a question could not be DELIVERED — never silent. */
  onSendFailed?(eventType: string, error: string): void;
}

export interface AskReport {
  /** Questions delivered to someone who can answer them. */
  asked: number;
  /**
   * Questions whose delivery FAILED. Counted separately and never silently:
   * a deploy blocked on a question nobody received looks identical to a deploy
   * that is merely slow.
   */
  failed: number;
}

export interface NotificationResponderHandle {
  stop(): void;
  /** Deliver any unanswered questions. */
  poll(): Promise<AskReport>;
  /**
   * Publish an answer against a waiting question.
   *
   * Takes the bus event id from the DELIVERY rather than from memory, so a
   * reply that arrives in a later process still lands — the CLI runs one
   * command per invocation, and an in-memory pending list would not survive
   * the gap between asking and being answered.
   */
  answer(eventId: string, value: string): void;
}

interface PendingRow {
  id: number;
  type: string;
  payload: string;
}

/**
 * Attach the responder to the bus.
 *
 * Holds no state of its own. What has been asked is recorded as a delivery row
 * (`kind: 'interview'`, `targetId` = the bus event id), which is also what the
 * inbound path looks a reply up by — so asking and answering can happen in
 * different processes, minutes apart, which is the normal case.
 */
export function startNotificationResponder(
  opts: NotificationResponderOptions,
): NotificationResponderHandle {
  const bus: Bus = openBus({ dbPath: opts.busDbPath, events: NO_SCHEMAS });
  const askAfterMs = opts.askAfterMs ?? ASK_AFTER_MS;

  /**
   * Bus events already asked about, and still awaiting a reply.
   *
   * Read from the deliveries table rather than kept in memory. The CLI runs
   * one command per invocation and the poll runs every few seconds, so an
   * in-memory guard would re-send the same question forever. A delivery that
   * has EXPIRED is deliberately absent here — nobody answered in a day, so
   * asking again is the right move.
   */
  function alreadyAsked(now: Date): Set<string> {
    return new Set(
      opts.db
        .select({ targetId: notificationDeliveries.targetId })
        .from(notificationDeliveries)
        .where(
          and(
            eq(notificationDeliveries.kind, 'interview'),
            isNull(notificationDeliveries.consumedAt),
            gt(notificationDeliveries.expiresAt, now),
          ),
        )
        .all()
        .map((row) => row.targetId),
    );
  }

  /**
   * Questions with no reply yet.
   *
   * `reply_for` on a reply event points at the question, so "unanswered" is
   * simply the absence of such a row. Matching on the family prefixes keeps
   * this from picking up unrelated traffic.
   */
  function unanswered(now: Date): PendingRow[] {
    return bus.db
      .query(
        `SELECT id, type, payload FROM events
          WHERE reply_for IS NULL
            AND emitted_at <= ?
            AND (${INTERVIEW_PATTERNS.map(() => 'type LIKE ?').join(' OR ')})
            AND NOT EXISTS (SELECT 1 FROM events r WHERE r.reply_for = events.id)
          ORDER BY id ASC`,
      )
      .all(
        now.getTime() - askAfterMs,
        ...INTERVIEW_PATTERNS.map((p) => `${p.replace(/\*\.\*$/, '')}%`),
      ) as PendingRow[];
  }

  return {
    stop() {
      bus.close();
    },

    async poll(): Promise<AskReport> {
      const now = opts.now();
      const asked = alreadyAsked(now);
      const report: AskReport = { asked: 0, failed: 0 };
      for (const row of unanswered(now)) {
        if (asked.has(String(row.id))) continue;

        const decision = decideInterviewDelivery({
          eventType: row.type,
          hasTty: opts.hasTty(),
          hasBidirectionalRoute: opts.routes.length > 0,
        });

        if (!decision.deliver) {
          // No delivery row is written for a declined question, so this fires
          // once per poll. That is intentional for `secret.*`: the operator
          // needs to keep seeing why their deploy is stuck.
          opts.onDeclined?.(row.type, decision.reason ?? 'declined');
          continue;
        }

        const route = opts.routes[0];
        const question = describeQuestion(
          row.type,
          JSON.parse(row.payload) as Record<string, unknown>,
        );

        const delivery = mintDelivery(opts.db, {
          kind: 'interview',
          // A BUS event id, not an alerts row — the two share this table and
          // differ only in what targetId points at.
          targetId: String(row.id),
          routeId: route.id,
          now,
          ttlMs: opts.ttlMs,
        });

        try {
          await opts.transportFor(route).send({
            address: route.address,
            body: composeInterviewBody({ ...question, token: delivery.token }),
          });
          report.asked++;
        } catch (error) {
          // The delivery row is what records "this was asked", and it was
          // minted BEFORE the send because the token has to be in the body.
          // A send that failed must not leave that record behind: it would
          // suppress every retry, and the deploy would wait forever on a
          // question nobody ever received.
          consumeDelivery(opts.db, delivery.id, now);
          report.failed++;
          opts.onSendFailed?.(row.type, error instanceof Error ? error.message : String(error));
        }
      }
      return report;
    },

    answer(eventId: string, value: string): void {
      const question = bus.db
        .query('SELECT type FROM events WHERE id = ?')
        .get(Number(eventId)) as { type: string } | null;
      if (!question) return;

      bus.emitRaw(
        `${question.type}.reply`,
        { value },
        { replyFor: Number(eventId), emittedBy: 'celilo-notification-responder' },
      );
    },
  };
}