import { Database } from 'bun:sqlite';
import { hostname } from 'node:os';
import pkg from '../package.json';
import { type CompiledPattern, compilePattern, matchPattern } from './pattern';
import type { AnyEventRegistry, EventTypeOf, PayloadOf, ReplyOf } from './registry';
import { applySchema } from './schema';
import { decodePayload, encodePayload, serializeError } from './serialize';
import type {
  BusEvent,
  BusHealth,
  Delivery,
  DeliveryStatus,
  DispatcherOptions,
  EmitOptions,
  FailedBySubscriber,
  FailingStreak,
  LiveDispatcher,
  QueryOptions,
  RetentionOptions,
  SubscribeOptions,
  Subscriber,
} from './types';

type AnyRegistry = AnyEventRegistry;

export interface OpenBusOptions<R extends AnyRegistry> {
  dbPath: string;
  events: R;
  retention?: RetentionOptions;
  dispatcherId?: string;
}

interface WatchEntry {
  id: number;
  compiled: CompiledPattern;
  callback: (event: BusEvent) => void | Promise<void>;
  /** Event ids this watch has already fired its callback for. */
  seenIds: Set<number>;
  /** High-water mark used by the poll path to limit each scan. */
  lastPolledId: number;
}

interface OnEventEntry {
  callback: (event: BusEvent) => void | Promise<void>;
}

const DEFAULT_RETENTION: Required<RetentionOptions> = {
  eventsRetentionDays: 30,
  failedRetentionDays: 90,
};

// Derived from package.json so it can never drift from the published version
// (the old hardcoded '0.1.0' silently diverged to 0.1.6 and made the doctor's
// dispatcher-currency detail wrong — ISS-0121).
export const BUS_VERSION: string = pkg.version;

/** How long an un-probeable (off-host) heartbeat still counts as live. */
const REMOTE_HEARTBEAT_LIVE_MS = 120_000;

export interface DispatcherHeartbeatRow {
  dispatcher_id: string;
  last_heartbeat: number;
  started_at: number;
  pid: number;
  version: string | null;
}

/** Signal 0 tests existence without delivering. EPERM = alive but not ours. */
export function isProcessAlive(pid: number): boolean {
  try {
    process.kill(pid, 0);
    return true;
  } catch (err) {
    return (err as NodeJS.ErrnoException).code === 'EPERM';
  }
}

/**
 * The Bus is the entry point for everything. Holds the database
 * handle, the event registry, and runtime-only state (transient
 * watches, in-process subscriber callbacks). Methods are split across
 * module files via mixin-style imports to keep this file from getting
 * unwieldy.
 */
export class Bus<R extends AnyRegistry = AnyRegistry> {
  readonly db: Database;
  readonly registry: R;
  readonly retention: Required<RetentionOptions>;
  readonly dispatcherId: string;

  /**
   * Transient watchers. Each watch has BOTH an in-process callback
   * (fired synchronously by fireWatches when an emit happens in this
   * Bus instance) AND a per-watch polling loop (catches emits from
   * OTHER Bus instances against the same DB — SQLite has no
   * cross-process LISTEN/NOTIFY). Per-watch `seenIds` deduplicates
   * the two paths so no event fires its callback twice.
   */
  readonly watches = new Map<number, WatchEntry>();
  /** Active poll intervals keyed by watch id — bus.close() clears them. */
  private watchIntervals = new Map<number, ReturnType<typeof setInterval>>();
  private nextWatchId = 1;

  /** In-process callbacks registered against persistent subscriber rows. */
  readonly onEventHandlers = new Map<string, OnEventEntry>();

  /** Set when runDispatcher is active so close() can stop it cleanly. */
  dispatcherRunning = false;
  dispatcherStop: (() => void) | null = null;

  constructor(opts: OpenBusOptions<R>) {
    this.db = new Database(opts.dbPath);
    applySchema(this.db);
    this.registry = opts.events;
    this.retention = { ...DEFAULT_RETENTION, ...(opts.retention ?? {}) };
    // Per-PROCESS, not per-host. Two dispatchers on one box used to write the
    // same `hostname()` row, so pid/started_at/version flip-flopped between them
    // every tick: the row said whichever process wrote last, the duplicate was
    // invisible without `ps`, and the doctor's started_at currency check (ISS-0121)
    // read a coin flip. One row per process is what makes a duplicate countable
    // (see liveDispatchers).
    this.dispatcherId = opts.dispatcherId ?? `${hostname()}:${process.pid}`;
  }

  // --- emit -----------------------------------------------------------

  emit<T extends EventTypeOf<R>>(
    type: T,
    payload: PayloadOf<R, T>,
    opts?: EmitOptions,
  ): BusEvent<T, PayloadOf<R, T>> {
    return this.emitRaw(type, payload, opts) as BusEvent<T, PayloadOf<R, T>>;
  }

  /**
   * Emit without compile-time type checking. Used by internal events
   * (`_internal.*`), reply events (`<parent>.reply` validated against
   * the parent's `replies` schema), and the CLI which receives types
   * as strings.
   */
  emitRaw(type: string, payload: unknown, opts?: EmitOptions): BusEvent {
    // Skip validation for:
    //   - internal events (`_internal.*`) — bus-managed
    //   - reply events when their parent declares a `replies` schema
    //     (validated by lookupSchema below)
    //   - empty-registry buses (the CLI / operator path) — there's
    //     nothing to validate against
    const hasSchemas = Object.keys(this.registry.defs).length > 0;
    if (hasSchemas && !type.startsWith('_internal.')) {
      const schema = this.lookupSchema(type);
      if (!schema) {
        throw new Error(`event-bus: event type "${type}" is not registered`);
      }
      schema.parse(payload);
    }

    const now = Date.now();
    const encoded = encodePayload(payload);

    const dedupKey = opts?.dedupKey ?? null;
    if (dedupKey) {
      const existing = this.db
        .query<{ id: number }, [string]>('SELECT id FROM events WHERE dedup_key = ? LIMIT 1')
        .get(dedupKey);
      if (existing) {
        return this.getEvent(existing.id) as BusEvent;
      }
    }

    const result = this.db.run(
      `INSERT INTO events (type, payload, emitted_at, emitted_by, dedup_key, reply_to, reply_for)
       VALUES (?, ?, ?, ?, ?, ?, ?)`,
      [
        type,
        encoded,
        now,
        opts?.emittedBy ?? null,
        dedupKey,
        opts?.replyTo ?? null,
        opts?.replyFor ?? null,
      ],
    );
    const id = Number(result.lastInsertRowid);

    this.materializeDeliveries(id, type);

    const event = this.getEvent(id) as BusEvent;
    this.fireWatches(event);
    return event;
  }

  /**
   * Resolve the schema for an event type. Direct match wins; otherwise,
   * a `<parent>.reply` event is validated against the parent's `replies`
   * schema if declared.
   */
  private lookupSchema(type: string) {
    const direct = this.registry.payloadSchema(type);
    if (direct) return direct;
    if (type.endsWith('.reply')) {
      return this.registry.replySchema(type.slice(0, -'.reply'.length));
    }
    return undefined;
  }

  private materializeDeliveries(eventId: number, type: string): void {
    // SQL pre-filter: subscribers whose like_pattern matches the type.
    // Then JS post-filter to enforce single-segment glob semantics.
    const candidates = this.db
      .query<
        { id: number; pattern: string; like_pattern: string; segment_count: number },
        [string]
      >(
        // ESCAPE '\' lets compiled patterns include `\_` and `\%` so
        // literal segments containing `_` or `%` (e.g. snake_case
        // event names) match correctly. See pattern.ts.
        `SELECT id, pattern, like_pattern, segment_count FROM subscribers WHERE ? LIKE like_pattern ESCAPE '\\'`,
      )
      .all(type);

    if (candidates.length === 0) return;

    const typeSegments = type.split('.').length;
    const insert = this.db.prepare(
      `INSERT INTO deliveries (event_id, subscriber_id, status, attempts, next_attempt_at)
       VALUES (?, ?, 'pending', 0, ?)`,
    );
    const now = Date.now();
    for (const c of candidates) {
      // segment_count: -1 means pattern has `**`, accept any length.
      if (c.segment_count !== -1 && c.segment_count !== typeSegments) continue;
      // Final precise check via compiled regex.
      const compiled = compilePattern(c.pattern);
      if (!matchPattern(compiled, type)) continue;
      insert.run(eventId, c.id, now);
    }
  }

  /**
   * True when at least one subscriber's pattern matches `type`. Same
   * two-phase check as materializeDeliveries (SQL LIKE pre-filter,
   * compiled-regex post-filter) without inserting anything. Used by
   * the timer emitter to skip ticks nobody listens for.
   */
  hasSubscriberMatching(type: string): boolean {
    const candidates = this.db
      .query<{ pattern: string; segment_count: number }, [string]>(
        `SELECT pattern, segment_count FROM subscribers WHERE ? LIKE like_pattern ESCAPE '\\'`,
      )
      .all(type);
    const typeSegments = type.split('.').length;
    for (const c of candidates) {
      if (c.segment_count !== -1 && c.segment_count !== typeSegments) continue;
      if (matchPattern(compilePattern(c.pattern), type)) return true;
    }
    return false;
  }

  // --- subscribers ----------------------------------------------------

  subscribe(opts: SubscribeOptions): Subscriber {
    const compiled = compilePattern(opts.pattern);
    const now = Date.now();
    const existing = this.db
      .query<{ id: number }, [string]>('SELECT id FROM subscribers WHERE name = ? LIMIT 1')
      .get(opts.name);

    if (existing) {
      this.db.run(
        `UPDATE subscribers
           SET pattern = ?, like_pattern = ?, segment_count = ?, handler = ?,
               max_attempts = ?, timeout_ms = ?, registered_at = ?, registered_by = ?
         WHERE id = ?`,
        [
          opts.pattern,
          compiled.likePattern,
          compiled.segmentCount,
          opts.handler,
          opts.maxAttempts ?? 3,
          opts.timeoutMs ?? 60_000,
          now,
          opts.registeredBy ?? null,
          existing.id,
        ],
      );
      return this.requireSubscriber(existing.id);
    }

    const result = this.db.run(
      `INSERT INTO subscribers
         (name, pattern, like_pattern, segment_count, handler, max_attempts, timeout_ms, registered_at, registered_by)
       VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`,
      [
        opts.name,
        opts.pattern,
        compiled.likePattern,
        compiled.segmentCount,
        opts.handler,
        opts.maxAttempts ?? 3,
        opts.timeoutMs ?? 60_000,
        now,
        opts.registeredBy ?? null,
      ],
    );
    return this.requireSubscriber(Number(result.lastInsertRowid));
  }

  unsubscribe(name: string): void {
    this.db.run('DELETE FROM subscribers WHERE name = ?', [name]);
    this.onEventHandlers.delete(name);
  }

  /**
   * Register an in-process callback for an existing persistent
   * subscriber row. The dispatcher prefers this over spawning the
   * subscriber's `handler` command when the bus is in-process.
   */
  onEvent<TPayload = unknown>(
    name: string,
    callback: (event: BusEvent<string, TPayload>) => void | Promise<void>,
  ): void {
    const sub = this.db
      .query<{ id: number }, [string]>('SELECT id FROM subscribers WHERE name = ? LIMIT 1')
      .get(name);
    if (!sub) {
      throw new Error(`event-bus: no subscriber named "${name}" — call subscribe() first`);
    }
    this.onEventHandlers.set(name, {
      callback: callback as (event: BusEvent) => void | Promise<void>,
    });
  }

  // --- watches (transient, in-process + cross-process via polling) ---

  watch(
    pattern: string,
    callback: (event: BusEvent) => void | Promise<void>,
    opts?: { pollIntervalMs?: number },
  ): { close: () => void } {
    const id = this.nextWatchId++;
    const compiled = compilePattern(pattern);

    // Seed the poll high-water mark to the current max event id so we
    // don't replay history when a watch starts. The watch only sees
    // events emitted AFTER it was registered.
    const seedRow = this.db
      .query<{ id: number | null }, []>('SELECT MAX(id) AS id FROM events')
      .get();
    const lastPolledId = seedRow?.id ?? 0;

    const entry: WatchEntry = {
      id,
      compiled,
      callback,
      seenIds: new Set<number>(),
      lastPolledId,
    };
    this.watches.set(id, entry);

    // Cross-process polling: SQLite has no LISTEN/NOTIFY across
    // processes, so a watch in process A can't see events emitted by
    // a Bus in process B without scanning the events table.
    // Default 250ms — interactive responders feel snappy without
    // burning CPU for idle watches. fireWatches handles the
    // in-process case synchronously; both paths share entry.seenIds
    // so no event fires twice.
    const pollIntervalMs = opts?.pollIntervalMs ?? 250;
    const poll = setInterval(() => {
      const rows = this.db
        .query<EventRow, [number]>('SELECT * FROM events WHERE id > ? ORDER BY id ASC LIMIT 100')
        .all(entry.lastPolledId);
      for (const row of rows) {
        entry.lastPolledId = Math.max(entry.lastPolledId, row.id);
        if (entry.seenIds.has(row.id)) continue;
        if (!matchPattern(entry.compiled, row.type)) continue;
        entry.seenIds.add(row.id);
        const event = rowToEvent(row);
        Promise.resolve()
          .then(() => entry.callback(event))
          .catch((err) => {
            console.error(`event-bus: watch handler threw on ${event.type}:`, err);
          });
      }
    }, pollIntervalMs);
    this.watchIntervals.set(id, poll);

    return {
      close: () => {
        this.watches.delete(id);
        const interval = this.watchIntervals.get(id);
        if (interval) clearInterval(interval);
        this.watchIntervals.delete(id);
      },
    };
  }

  private fireWatches(event: BusEvent): void {
    if (this.watches.size === 0) return;
    for (const w of this.watches.values()) {
      if (w.seenIds.has(event.id)) continue;
      if (!matchPattern(w.compiled, event.type)) continue;
      w.seenIds.add(event.id);
      // Track the in-process emit so the poll's high-water mark stays
      // ahead — otherwise the poll would re-discover this event next
      // tick and seenIds would block the second fire (correct), but
      // the lastPolledId wouldn't advance and the poll's LIMIT 100
      // window would clog with already-seen events. Cheap correctness.
      if (event.id > w.lastPolledId) w.lastPolledId = event.id;
      Promise.resolve()
        .then(() => w.callback(event))
        .catch((err) => {
          console.error(`event-bus: watch handler threw on ${event.type}:`, err);
        });
    }
  }

  // --- query / reply --------------------------------------------------

  async query<T extends EventTypeOf<R>>(
    type: T,
    payload: PayloadOf<R, T>,
    opts?: QueryOptions,
  ): Promise<BusEvent<string, ReplyOf<R, T>>[]> {
    const replyTo = crypto.randomUUID();
    const timeoutMs = opts?.timeoutMs ?? 5_000;
    const pollIntervalMs = opts?.pollIntervalMs ?? 250;
    const expect = opts?.expect ?? 'first';

    const replies: BusEvent[] = [];
    const seenReplyIds = new Set<number>();
    let resolve!: (value: BusEvent[]) => void;
    const done = new Promise<BusEvent[]>((r) => {
      resolve = r;
    });

    const expectedCount =
      expect === 'first' ? 1 : expect === 'all' ? Number.POSITIVE_INFINITY : expect.count;

    const recordReply = (event: BusEvent): void => {
      if (seenReplyIds.has(event.id)) return;
      seenReplyIds.add(event.id);
      replies.push(event);
      if (replies.length >= expectedCount) resolve(replies);
    };

    // Fast path: in-process watch fires synchronously when a responder
    // running in this same Bus instance calls bus.reply.
    const watch = this.watch('**', (event) => {
      if (event.replyFor !== queryEventId) return;
      recordReply(event);
    });

    const queryEvent = this.emitRaw(type, payload, {
      ...opts,
      replyTo,
    });
    const queryEventId = queryEvent.id;

    // Cross-process path: a responder in a different process emits
    // bus.reply, which writes to the same SQLite DB. Polling the
    // events table is the only way for this Bus instance to see it
    // (SQLite has no LISTEN/NOTIFY equivalent across processes).
    const poll = setInterval(() => {
      const rows = this.db
        .query<EventRow, [number]>('SELECT * FROM events WHERE reply_for = ? ORDER BY id ASC')
        .all(queryEventId);
      for (const row of rows) {
        if (seenReplyIds.has(row.id)) continue;
        recordReply(rowToEvent(row));
      }
    }, pollIntervalMs);

    // timeoutMs of 0 means "wait forever" — the celilo interview flow
    // wants this so deploys block until a responder answers, with no
    // arbitrary deadline.
    let timer: ReturnType<typeof setTimeout> | null = null;
    if (timeoutMs > 0) {
      timer = setTimeout(() => resolve(replies), timeoutMs);
    }

    const result = await done;
    if (timer) clearTimeout(timer);
    clearInterval(poll);
    watch.close();
    return result as BusEvent<string, ReplyOf<R, T>>[];
  }

  /**
   * Emit a reply to an earlier query event. The library auto-fills
   * `replyFor` from the original event's id; the caller supplies the
   * reply payload (validated against the registry's `replies` schema
   * for the original event type).
   */
  reply<TQueryType extends EventTypeOf<R>>(
    queryEvent: BusEvent<TQueryType>,
    replyPayload: ReplyOf<R, TQueryType>,
  ): BusEvent {
    const replySchema = this.registry.replySchema(queryEvent.type);
    if (replySchema) replySchema.parse(replyPayload);
    return this.emitRaw(`${queryEvent.type}.reply`, replyPayload, {
      replyFor: queryEvent.id,
    });
  }

  // --- delivery state writes (called by handlers / dispatcher) -------

  markSucceeded(deliveryId: { eventId: number; subscriberId: number }): void {
    const now = Date.now();
    this.db.run(
      `UPDATE deliveries
         SET status = 'succeeded', finished_at = ?, last_error = NULL
       WHERE event_id = ? AND subscriber_id = ?`,
      [now, deliveryId.eventId, deliveryId.subscriberId],
    );
  }

  markFailed(
    deliveryId: { eventId: number; subscriberId: number },
    error: unknown,
    opts?: { retryAfter?: number; abandoned?: boolean },
  ): void {
    const now = Date.now();
    const errStr = serializeError(error);
    if (opts?.abandoned) {
      this.db.run(
        `UPDATE deliveries
           SET status = 'abandoned', finished_at = ?, last_error = ?, next_attempt_at = NULL
         WHERE event_id = ? AND subscriber_id = ?`,
        [now, errStr, deliveryId.eventId, deliveryId.subscriberId],
      );
      return;
    }
    if (typeof opts?.retryAfter === 'number') {
      this.db.run(
        `UPDATE deliveries
           SET status = 'pending', last_error = ?, next_attempt_at = ?, started_at = NULL
         WHERE event_id = ? AND subscriber_id = ?`,
        [errStr, opts.retryAfter, deliveryId.eventId, deliveryId.subscriberId],
      );
      return;
    }
    this.db.run(
      `UPDATE deliveries
         SET status = 'failed', finished_at = ?, last_error = ?
       WHERE event_id = ? AND subscriber_id = ?`,
      [now, errStr, deliveryId.eventId, deliveryId.subscriberId],
    );
  }

  // --- reads ----------------------------------------------------------

  getEvent(id: number): BusEvent | null {
    const row = this.db.query<EventRow, [number]>('SELECT * FROM events WHERE id = ?').get(id);
    return row ? rowToEvent(row) : null;
  }

  /** The row was just written in this transaction; its absence is corruption, not a miss. */
  private requireSubscriber(id: number): Subscriber {
    const subscriber = this.getSubscriber(id);
    if (!subscriber) throw new Error(`subscriber ${id} is missing right after being written`);
    return subscriber;
  }

  getSubscriber(id: number): Subscriber | null {
    const row = this.db
      .query<SubscriberRow, [number]>('SELECT * FROM subscribers WHERE id = ?')
      .get(id);
    return row ? rowToSubscriber(row) : null;
  }

  getSubscriberByName(name: string): Subscriber | null {
    const row = this.db
      .query<SubscriberRow, [string]>('SELECT * FROM subscribers WHERE name = ?')
      .get(name);
    return row ? rowToSubscriber(row) : null;
  }

  getDelivery(eventId: number, subscriberId: number): Delivery | null {
    const row = this.db
      .query<DeliveryRow, [number, number]>(
        'SELECT * FROM deliveries WHERE event_id = ? AND subscriber_id = ?',
      )
      .get(eventId, subscriberId);
    return row ? rowToDelivery(row) : null;
  }

  /**
   * Every delivery materialized for a single event, one row per matching
   * subscriber. Empty when no subscriber matched the event's type (the
   * common "nothing is listening" case). Lets a caller block until an
   * event it emitted has been fully processed — see celilo's deploy wait
   * for the public_web reconcile (ISS-0035).
   */
  deliveriesForEvent(eventId: number): Delivery[] {
    const rows = this.db
      .query<DeliveryRow, [number]>(
        'SELECT * FROM deliveries WHERE event_id = ? ORDER BY subscriber_id ASC',
      )
      .all(eventId);
    return rows.map(rowToDelivery);
  }

  recentEvents(opts?: { limit?: number; type?: string }): BusEvent[] {
    const limit = opts?.limit ?? 50;
    const rows = opts?.type
      ? this.db
          .query<EventRow, [string, number]>(
            'SELECT * FROM events WHERE type = ? ORDER BY id DESC LIMIT ?',
          )
          .all(opts.type, limit)
      : this.db
          .query<EventRow, [number]>('SELECT * FROM events ORDER BY id DESC LIMIT ?')
          .all(limit);
    return rows.map(rowToEvent);
  }

  /** Replies correlated to a query event — empty means still unanswered. */
  repliesFor(eventId: number): BusEvent[] {
    const rows = this.db
      .query<EventRow, [number]>('SELECT * FROM events WHERE reply_for = ? ORDER BY id ASC')
      .all(eventId);
    return rows.map(rowToEvent);
  }

  /**
   * Query events still awaiting a correlated reply — i.e. questions nobody has
   * answered, and therefore commands that may be parked on them.
   *
   * A *query* is any event `bus.query` emitted: it carries a `reply_to`
   * correlation id. Unanswered means no event has `reply_for = <its id>`.
   *
   * This is NOT `pendingDeliveries`, which reads the `deliveries` table
   * (subscriber fan-out) and cannot see an unanswered query at all. Trusting
   * that reporter by its name is how celilo#609 was misdiagnosed: it returned
   * `[]` both before and after, and `[]` reads as knowledge rather than as a
   * category error.
   */
  unansweredQueries(opts?: { limit?: number }): BusEvent[] {
    const rows = this.db
      .query<EventRow, [number]>(
        `SELECT * FROM events e
          WHERE e.reply_to IS NOT NULL
            AND NOT EXISTS (SELECT 1 FROM events r WHERE r.reply_for = e.id)
          ORDER BY e.id DESC LIMIT ?`,
      )
      .all(opts?.limit ?? 50);
    return rows.map(rowToEvent);
  }

  pendingDeliveries(opts?: { subscriber?: string; limit?: number }): Delivery[] {
    const limit = opts?.limit ?? 100;
    if (opts?.subscriber) {
      const rows = this.db
        .query<DeliveryRow, [string, number]>(
          `SELECT d.* FROM deliveries d
             JOIN subscribers s ON s.id = d.subscriber_id
            WHERE d.status = 'pending' AND s.name = ?
            ORDER BY d.next_attempt_at ASC LIMIT ?`,
        )
        .all(opts.subscriber, limit);
      return rows.map(rowToDelivery);
    }
    const rows = this.db
      .query<DeliveryRow, [number]>(
        `SELECT * FROM deliveries WHERE status = 'pending' ORDER BY next_attempt_at ASC LIMIT ?`,
      )
      .all(limit);
    return rows.map(rowToDelivery);
  }

  failedDeliveries(opts?: { limit?: number; subscriber?: string }): Delivery[] {
    const limit = opts?.limit ?? 100;
    if (opts?.subscriber) {
      const rows = this.db
        .query<DeliveryRow, [string, number]>(
          `SELECT d.* FROM deliveries d
             JOIN subscribers s ON s.id = d.subscriber_id
            WHERE d.status IN ('failed','abandoned') AND s.name = ?
            ORDER BY d.finished_at DESC LIMIT ?`,
        )
        .all(opts.subscriber, limit);
      return rows.map(rowToDelivery);
    }
    const rows = this.db
      .query<DeliveryRow, [number]>(
        `SELECT * FROM deliveries WHERE status IN ('failed','abandoned') ORDER BY finished_at DESC LIMIT ?`,
      )
      .all(limit);
    return rows.map(rowToDelivery);
  }

  /**
   * How many failed/abandoned deliveries there actually are, per subscriber,
   * newest-first overall.
   *
   * `failedDeliveries().length` is NOT this: it saturates at its LIMIT, so
   * doctor reported a literal `50` on celilo-mgr and it read as a count
   * (celilo#623). Raising the limit is the same bug with a bigger number — ask
   * for a total when you want a total.
   */
  failedDeliveryTotals(): { total: number; bySubscriber: FailedBySubscriber[] } {
    const bySubscriber = this.db
      .query<FailedBySubscriber, []>(
        `SELECT s.name AS subscriber, COUNT(*) AS count, MAX(d.finished_at) AS latestFinishedAt
           FROM deliveries d JOIN subscribers s ON s.id = d.subscriber_id
          WHERE d.status IN ('failed','abandoned')
          GROUP BY s.name ORDER BY count DESC, s.name ASC`,
      )
      .all();
    // Deliveries can outlive their subscriber row, so the join can drop rows a
    // bare COUNT(*) sees. Count the table itself for the total.
    const total =
      this.db
        .query<{ count: number }, []>(
          `SELECT COUNT(*) AS count FROM deliveries WHERE status IN ('failed','abandoned')`,
        )
        .get()?.count ?? 0;
    return { total, bySubscriber };
  }

  /**
   * Each subscriber's unbroken run of failures since it last succeeded.
   *
   * `failedDeliveryTotals` says how much has gone wrong. It cannot say whether
   * anything still goes RIGHT: a subscriber with 400 old failures and a clean
   * week reads the same as one that has failed every tick for a month
   * (celilo#440). This asks the other question, and only finished deliveries
   * count, so a subscriber that is merely busy has no streak.
   *
   * ponytail: a success older than the retention window has been swept, so a
   * streak that long reports `lastSucceededAt: null` and starts at the oldest
   * failure still on record. That understates how long it has been, and never
   * the fact of it.
   */
  failingStreaks(): FailingStreak[] {
    return this.db
      .query<FailingStreak, []>(
        `SELECT s.name AS subscriber,
                COUNT(*) AS failures,
                MIN(d.finished_at) AS since,
                MAX(d.finished_at) AS latest,
                ok.at AS lastSucceededAt
           FROM deliveries d
           JOIN subscribers s ON s.id = d.subscriber_id
           LEFT JOIN (SELECT subscriber_id, MAX(finished_at) AS at
                        FROM deliveries WHERE status = 'succeeded'
                       GROUP BY subscriber_id) ok
             ON ok.subscriber_id = d.subscriber_id
          WHERE d.status IN ('failed','abandoned')
            AND d.finished_at IS NOT NULL
            AND d.finished_at > COALESCE(ok.at, -1)
          GROUP BY s.name
          ORDER BY since ASC, s.name ASC`,
      )
      .all();
  }

  /**
   * Every dispatcher that still exists, newest heartbeat first.
   *
   * Liveness is decided by the PROCESS, not the heartbeat age: a dispatcher tick
   * blocks for as long as its slowest handler runs (a 60s timeout means a 60s-old
   * heartbeat on a perfectly healthy bus), so freshness alone would declare a busy
   * dispatcher dead and wave a second one through. `kill(pid, 0)` is exact for
   * same-host peers, which is the only shape that matters — the bus DB is a local
   * SQLite file. An off-host row can't be probed, so it falls back to heartbeat age.
   */
  liveDispatchers(now: number = Date.now()): LiveDispatcher[] {
    const host = hostname();
    return this.db
      .query<DispatcherHeartbeatRow, []>(
        `SELECT dispatcher_id, last_heartbeat, started_at, pid, version
           FROM dispatcher_heartbeat ORDER BY last_heartbeat DESC`,
      )
      .all()
      .filter((row) => {
        const sameHost = row.dispatcher_id.split(':')[0] === host;
        if (!sameHost) return now - row.last_heartbeat < REMOTE_HEARTBEAT_LIVE_MS;
        return isProcessAlive(row.pid);
      })
      .map((row) => ({
        dispatcherId: row.dispatcher_id,
        pid: row.pid,
        version: row.version,
        lastHeartbeatAgeMs: now - row.last_heartbeat,
        startedAgoMs: now - row.started_at,
      }));
  }

  health(opts?: { staleHandlerMultiplier?: number }): BusHealth {
    const mult = opts?.staleHandlerMultiplier ?? 2;
    const now = Date.now();
    const hb = this.db
      .query<{ last_heartbeat: number }, []>(
        'SELECT last_heartbeat FROM dispatcher_heartbeat ORDER BY last_heartbeat DESC LIMIT 1',
      )
      .get();
    const live = this.liveDispatchers(now);
    const stuck = this.db
      .query<{ count: number }, [number]>(
        `SELECT COUNT(*) AS count FROM deliveries d
            JOIN subscribers s ON s.id = d.subscriber_id
          WHERE d.status = 'running'
            AND d.started_at < ?`,
      )
      .get(now - 60_000 * mult) ?? { count: 0 };
    const oldest = this.db
      .query<{ next_attempt_at: number | null }, []>(
        `SELECT MIN(next_attempt_at) AS next_attempt_at FROM deliveries WHERE status = 'pending'`,
      )
      .get();

    // Anything the dispatcher is mid-flight on. A tick blocks for as long as its
    // slowest handler runs, so a RUNNING delivery is the legitimate reason for a
    // stale heartbeat — its absence is what separates wedged from merely busy.
    const running = this.db
      .query<{ count: number }, []>(
        `SELECT COUNT(*) AS count FROM deliveries WHERE status = 'running'`,
      )
      .get() ?? { count: 0 };

    let status: BusHealth['status'] = 'healthy';
    let remedy: string | undefined;
    let lastHeartbeatAgeMs: number | null = null;
    if (!hb) {
      status = 'no_dispatcher';
    } else {
      lastHeartbeatAgeMs = now - hb.last_heartbeat;
      const stale = lastHeartbeatAgeMs > 30_000;
      // `live` is PID-liveness, so registered-and-alive is a different fact from
      // beating. Reporting a stale-but-alive dispatcher as `no_dispatcher` was
      // self-contradicting against dispatcherCount >= 1 (celilo#1398): the process
      // is PRESENT BUT DEAD, which is `wedged` and needs a restart, not a start.
      if (stale && live.length > 0 && running.count === 0) {
        status = 'wedged';
        remedy = 'celilo events restart-daemon --system';
      } else if (stale && stuck.count > 0) status = 'stuck';
      else if (stale && live.length === 0) status = 'no_dispatcher';
      else if (stuck.count > 0) status = 'recovering';
      // Remaining case — stale, alive, deliveries running — is a busy dispatcher
      // blocked in a slow handler. That is healthy.
    }
    // Only override the healthy case: a stuck/recovering bus has a more urgent
    // headline, and `dispatchers` below reports the duplicate either way.
    if (status === 'healthy' && live.length > 1) status = 'duplicate_dispatcher';

    return {
      status,
      lastHeartbeatAgeMs,
      stuckRunningCount: stuck.count,
      oldestPendingAgeMs: oldest?.next_attempt_at != null ? now - oldest.next_attempt_at : null,
      dispatcherCount: live.length,
      ...(remedy ? { remedy } : {}),
      dispatchers: live,
    };
  }

  private lastHeartbeatAt(): number | null {
    return (
      this.db
        .query<{ last_heartbeat: number }, []>(
          'SELECT last_heartbeat FROM dispatcher_heartbeat ORDER BY last_heartbeat DESC LIMIT 1',
        )
        .get()?.last_heartbeat ?? null
    );
  }

  /**
   * `health()` with the `wedged` verdict confirmed over time.
   *
   * A single sample cannot separate a dead loop from one starved by box load
   * (celilo#1399): both show a stale heartbeat, a live pid and an advancing age.
   * The difference is whether the heartbeat TIMESTAMP still moves. So when — and
   * only when — the instant verdict is `wedged`, watch for one more beat: a beat
   * during the window means `lagging` (alive, recovers on its own, no remedy),
   * silence means `wedged` stands.
   *
   * Callers that must act on the verdict (abandon a reconcile, tell an operator
   * to restart, restart automatically) use this. `health()` stays synchronous
   * for everything that only reports.
   */
  async healthConfirmed(opts?: {
    windowMs?: number;
    staleHandlerMultiplier?: number;
  }): Promise<BusHealth> {
    const first = this.health(opts);
    if (first.status !== 'wedged') return first;
    const before = this.lastHeartbeatAt();
    await new Promise((r) => setTimeout(r, opts?.windowMs ?? 5_000));
    const after = this.health(opts);
    // Re-read rather than trusting `first`: the beat may have landed and cleared
    // the staleness outright, in which case its own verdict is the honest one.
    if (after.status !== 'wedged') return after;
    const beat = this.lastHeartbeatAt();
    if (beat !== null && before !== null && beat > before) {
      const { remedy: _drop, ...rest } = after;
      return { ...rest, status: 'lagging' };
    }
    return after;
  }

  close(): void {
    if (this.dispatcherStop) {
      this.dispatcherStop();
      this.dispatcherStop = null;
    }
    // Clear every watch's polling interval before closing the DB —
    // otherwise the next tick's poll fires SELECT against a closed
    // bun:sqlite Database and throws "disk I/O error" (rare but real).
    for (const interval of this.watchIntervals.values()) {
      clearInterval(interval);
    }
    this.watchIntervals.clear();
    this.watches.clear();
    this.onEventHandlers.clear();
    this.db.close();
  }
}

export function openBus<R extends AnyRegistry>(opts: OpenBusOptions<R>): Bus<R> {
  return new Bus(opts);
}

// --- row helpers ------------------------------------------------------

interface EventRow {
  id: number;
  type: string;
  payload: string | null;
  emitted_at: number;
  emitted_by: string | null;
  dedup_key: string | null;
  reply_to: string | null;
  reply_for: number | null;
}

interface SubscriberRow {
  id: number;
  name: string;
  pattern: string;
  like_pattern: string;
  segment_count: number;
  handler: string;
  max_attempts: number;
  timeout_ms: number;
  registered_at: number;
  registered_by: string | null;
}

interface DeliveryRow {
  event_id: number;
  subscriber_id: number;
  status: DeliveryStatus;
  attempts: number;
  next_attempt_at: number | null;
  started_at: number | null;
  finished_at: number | null;
  last_error: string | null;
}

function rowToEvent(row: EventRow): BusEvent {
  return {
    id: row.id,
    type: row.type,
    payload: decodePayload(row.payload),
    emittedAt: row.emitted_at,
    emittedBy: row.emitted_by,
    dedupKey: row.dedup_key,
    replyTo: row.reply_to,
    replyFor: row.reply_for,
  };
}

function rowToSubscriber(row: SubscriberRow): Subscriber {
  return {
    id: row.id,
    name: row.name,
    pattern: row.pattern,
    likePattern: row.like_pattern,
    segmentCount: row.segment_count,
    handler: row.handler,
    maxAttempts: row.max_attempts,
    timeoutMs: row.timeout_ms,
    registeredAt: row.registered_at,
    registeredBy: row.registered_by,
  };
}

function rowToDelivery(row: DeliveryRow): Delivery {
  return {
    eventId: row.event_id,
    subscriberId: row.subscriber_id,
    status: row.status,
    attempts: row.attempts,
    nextAttemptAt: row.next_attempt_at,
    startedAt: row.started_at,
    finishedAt: row.finished_at,
    lastError: row.last_error,
  };
}

// re-export DispatcherOptions for the dispatcher module
export type { DispatcherOptions };