/**
 * The inbound poller — reading replies back off a transport.
 *
 * celilo-mgr polls the transport; the transport never calls in. That direction
 * matters: an inbound webhook would need public HTTPS, which means caddy, DNS
 * and a certificate — so the ack path would depend on the very infrastructure
 * being paged about. Polling needs none of it.
 *
 * Runs on a short interval (seconds, not the five-minute sweep) because "I
 * texted ack and nothing happened" is the whole experience being bought here.
 * It is an ordinary interval inside an already-supervised process, not a new
 * bus primitive.
 */

import type { InboundMessage, NotificationCapability } from '@celilo/capabilities';
import { and, eq, gt, isNull } from 'drizzle-orm';
import type { DbClient } from '../../db/client';
import { type Route, modules, notificationDeliveries, people, routes } from '../../db/schema';
import { loadCapabilityFunctions } from '../../hooks/capability-loader';
import { createCapturingLogger } from '../../hooks/logger';
import { type AckResult, acknowledgeAlert } from './ack';
import { interpretInbound } from './inbound';
import type { NotificationTransport } from './notifier';
import { composeAckBroadcastBody } from './notifier';
import { consumeDelivery } from './tokens';

/**
 * What one attempt to read a transport produced.
 *
 * `unidirectional` and `failed` used to collapse into a bare null, and that
 * cost a week: a transport whose read call was being REFUSED reported the same
 * "0 message(s)" as a transport nobody had replied on. An operator could not
 * tell "my reply never arrived" from "celilo cannot read this transport at
 * all", and neither could anyone debugging it.
 */
export type TransportReceive =
  | { status: 'received'; messages: InboundMessage[]; cursor: string | null }
  | { status: 'unidirectional' }
  | { status: 'failed'; error: string };

/** A transport that could not be read, and why. */
export interface TransportFailure {
  transportModuleId: string;
  error: string;
}

/**
 * The outcome of one attempt to read a transport, as recorded for later.
 *
 * Written on EVERY attempt, including failures — which is the whole point. The
 * only per-transport state celilo persisted before this was the cursor, and a
 * failed read produces no cursor, so `writeCursor` returned early and nothing
 * was written. The store could not REPRESENT a failure, so an absence of
 * recorded failures was never evidence there had been none: a transport that
 * had not been readable for a week looked identical to one nobody had replied
 * on (#501).
 */
export interface TransportReadRecord {
  /** When the attempt happened, ISO-8601. */
  at: string;
  outcome: 'received' | 'unidirectional' | 'failed';
  /** Present only when `failed`. */
  error?: string;
  /** How many messages the read returned. Zero is not the same as failure. */
  messages: number;
  /**
   * When a read last SUCCEEDED, carried forward across failures.
   *
   * This is the field that answers the question the old state could not: a
   * transport reporting "0 messages" for a week and one that has not been
   * readable for a week are the same picture until you can see this.
   */
  lastSuccessAt?: string;
}

/** A message that was read but not acted on, and why. */
export interface UnheardMessage {
  senderAddress: string;
  reason:
    | 'unknown_sender'
    | 'unknown_token'
    | 'wrong_sender'
    | 'ambiguous'
    | 'unrecognised'
    /** A deploy question was named, but with no value to answer it with. */
    | 'needs_value'
    /** The token was valid, but the alert it names no longer exists. */
    | 'stale_target';
}

export interface InboundPollDeps {
  /** Receive from one transport module. */
  receiveFrom(transportModuleId: string, cursor: string | null): Promise<TransportReceive>;
  /** Persisted receive cursor per transport. */
  readCursor(transportModuleId: string): string | null;
  writeCursor(transportModuleId: string, cursor: string | null): void;
  /**
   * Record what one read attempt produced. Called for EVERY attempt — a failed
   * read must leave a trace, or "we could not read this transport" stays
   * indistinguishable from "nobody replied". Optional so a caller that does not
   * care (tests, one-off invocations) need not supply it.
   */
  recordRead?(transportModuleId: string, record: TransportReadRecord): void;
  now(): Date;
  /**
   * Transport for a route, so an ack can be broadcast to everyone else paged.
   * Optional: an installation with no send path still acks correctly, it just
   * cannot tell the others.
   */
  transportFor?(route: Route): NotificationTransport;
  /**
   * Publish an interview answer against the waiting bus question. Optional:
   * an installation with no responder attached still acks alerts, it just
   * cannot answer deploy questions.
   */
  answerInterview?(eventId: string, value: string): void;
}

export interface InboundPollReport {
  transportsPolled: number;
  messagesRead: number;
  acked: number;
  /** Routes told that someone else took the alert. */
  broadcast: number;
  /** Deploy questions answered from a phone. */
  answered: number;
  /** Transports that could not be read at all. Never silent — see above. */
  failures: TransportFailure[];
  /** Messages read and then discarded, with the reason each was discarded. */
  unheard: UnheardMessage[];
}

/** Transports that have at least one route pointing at them. */
export function transportsWithRoutes(db: DbClient): string[] {
  const ids = new Set(
    db
      .select({ id: routes.transportModuleId })
      .from(routes)
      .where(eq(routes.enabled, true))
      .all()
      .map((r) => r.id),
  );
  return [...ids];
}

/**
 * Poll every transport once and apply what came back.
 *
 * At-least-once is fine and exactly-once is not attempted: the cursor is
 * persisted after processing, so a crash mid-batch replays it. Every action
 * below is idempotent — consuming an already-consumed token finds nothing,
 * and acking an already-acked alert re-writes the same row — so a replay is
 * harmless. Trying for exactly-once would add machinery to prevent something
 * that does not hurt.
 */
export async function pollInbound(db: DbClient, deps: InboundPollDeps): Promise<InboundPollReport> {
  const report: InboundPollReport = {
    transportsPolled: 0,
    messagesRead: 0,
    acked: 0,
    broadcast: 0,
    answered: 0,
    failures: [],
    unheard: [],
  };

  for (const transportId of transportsWithRoutes(db)) {
    const received = await deps.receiveFrom(transportId, deps.readCursor(transportId));
    report.transportsPolled++;
    deps.recordRead?.(transportId, {
      at: deps.now().toISOString(),
      outcome: received.status,
      ...(received.status === 'failed' ? { error: received.error } : {}),
      messages: received.status === 'received' ? received.messages.length : 0,
    });
    // One dead transport must not stop the others being read — but it is
    // RECORDED rather than skipped in silence, because "cannot read" and
    // "nothing to read" are the two things an operator most needs to tell
    // apart at 3am.
    if (received.status === 'failed') {
      report.failures.push({ transportModuleId: transportId, error: received.error });
      continue;
    }
    if (received.status === 'unidirectional') continue;

    for (const message of received.messages) {
      report.messagesRead++;

      const outcome = interpretInbound(db, {
        senderAddress: message.senderAddress,
        body: message.body,
        now: deps.now(),
        outstandingForRoute: (routeId) => outstandingAlertDeliveries(db, routeId, deps.now()),
      });

      if (outcome.action === 'ignored' || outcome.action === 'rejected') {
        report.unheard.push({ senderAddress: message.senderAddress, reason: outcome.reason });
        continue;
      }

      // An interview reply IS the answer rather than an acknowledgement —
      // same token table, same sender check, different meaning for the body.
      // The value was extracted by `interpretInbound`, which is the only place
      // that knows the delivery's kind; re-parsing it here is what left the
      // real grammar unreachable for months (#533).
      if (outcome.action === 'answer') {
        if (!deps.answerInterview) {
          // Nothing is attached to publish the answer against, so the question
          // is still unanswered. Say so rather than consuming the token: the
          // operator's next attempt has to be able to work.
          report.unheard.push({ senderAddress: message.senderAddress, reason: 'unrecognised' });
          continue;
        }
        consumeDelivery(db, outcome.delivery.id, deps.now());
        deps.answerInterview(outcome.delivery.targetId, outcome.value);
        report.answered++;
        continue;
      }

      // Consume first: a token is single-use, and a crash after acking but
      // before consuming would let the same token act twice.
      consumeDelivery(db, outcome.delivery.id, deps.now());
      const ack = acknowledgeAlert(
        db,
        outcome.delivery.targetId,
        outcome.route.personId,
        deps.now(),
      );

      // A live token can outlive the alert it names. When the row is gone
      // there is nothing to acknowledge, and saying "1 acked" anyway is a lie
      // the operator has no way to catch — observed live: a reply reported
      // `1 acked` while every alert still read `ackedBy: null`. Count what
      // HAPPENED, never what was attempted.
      if (!ack) {
        report.unheard.push({ senderAddress: message.senderAddress, reason: 'stale_target' });
        continue;
      }

      report.acked++;

      // Everyone else who was paged is still expecting to act. Telling them is
      // the entire point of a per-delivery token — it is what makes the reply
      // identify a PERSON rather than just an alert.
      report.broadcast += await broadcastAck(db, ack, deps);
    }

    deps.writeCursor(transportId, received.cursor);
  }

  return report;
}

/**
 * Tell every other paged route who took the alert.
 *
 * Best-effort per route: one unreachable phone must not stop the others being
 * told, and the ack itself has already been recorded either way.
 */
async function broadcastAck(db: DbClient, ack: AckResult, deps: InboundPollDeps): Promise<number> {
  if (!deps.transportFor || ack.broadcastTo.length === 0) return 0;

  const acknowledger = ack.alert.ackedBy
    ? db.select().from(people).where(eq(people.id, ack.alert.ackedBy)).get()?.name
    : undefined;

  let sent = 0;
  for (const routeId of ack.broadcastTo) {
    const route = db.select().from(routes).where(eq(routes.id, routeId)).get();
    if (!route) continue;
    try {
      await deps.transportFor(route).send({
        address: route.address,
        body: composeAckBroadcastBody(ack.alert, acknowledger ?? 'Someone'),
      });
      sent++;
    } catch {
      // See above: the ack stands regardless of who could be reached.
    }
  }
  return sent;
}

/**
 * Live alert deliveries outstanding for a route — what a bare `ack` with no
 * token could be referring to.
 */
function outstandingAlertDeliveries(db: DbClient, routeId: string, now: Date) {
  return db
    .select()
    .from(notificationDeliveries)
    .where(
      and(
        eq(notificationDeliveries.routeId, routeId),
        eq(notificationDeliveries.kind, 'alert'),
        isNull(notificationDeliveries.consumedAt),
        gt(notificationDeliveries.expiresAt, now),
      ),
    )
    .all();
}

/**
 * Build the real receive function for a transport.
 *
 * A transport with no `receive` is unidirectional — that is not an error, it
 * just means replies cannot arrive, which the route's `can_ack` already
 * records. Anything else going wrong IS an error and is returned as one.
 *
 * This used to be a bare `catch {}` returning null, on the reasoning that the
 * transport's own health check would report an unreachable daemon. That was
 * wrong twice over: a health check that only proves the daemon answers cannot
 * see a read call being REFUSED by a daemon that is otherwise perfectly
 * healthy, and the swallowed error was the only place the reason existed. The
 * live signal-cli case was exactly that shape — `receive` refused with
 * "Receive command cannot be used if messages are already being received."
 * while every health check passed and every poll reported zero messages.
 */
export function makeReceiver(db: DbClient) {
  return async (transportModuleId: string, cursor: string | null): Promise<TransportReceive> => {
    const module = db.select().from(modules).where(eq(modules.id, transportModuleId)).get();
    if (!module) {
      return { status: 'failed', error: `no module '${transportModuleId}' is installed` };
    }

    try {
      const { logger } = createCapturingLogger();
      const capabilities = await loadCapabilityFunctions(transportModuleId, db, logger);
      const notification = (capabilities as Record<string, unknown>).notification as
        | NotificationCapability
        | undefined;
      if (!notification?.receive) return { status: 'unidirectional' };
      const result = await notification.receive(cursor);
      return { status: 'received', messages: result.messages, cursor: result.cursor };
    } catch (error) {
      return { status: 'failed', error: error instanceof Error ? error.message : String(error) };
    }
  };
}