/*
 * Event subscriptions — stop relying on someone CHOOSING to tell you.
 *
 * The bus is almost entirely direct messaging: an agent learns something
 * happened because another agent decided to say so. Every miss this week was a
 * missing NOTIFICATION rather than a missing capability — two mergeable PRs sat
 * 17 hours because nobody told the coordinator to gate, the console's trigger
 * ran zero times because nothing woke it, and `stall_check` runs only when a
 * human types it.
 */
import { createHash, randomUUID } from "node:crypto";
import { existsSync, readFileSync, writeFileSync, mkdirSync } from "node:fs";
import path from "node:path";
import { z } from "zod";
import { ROOT } from "../store.js";
import { EVENT_KINDS, EVENT_KIND_IDS, type RecordEvent, type SubKind } from "./event-kinds.js";

const subsFile = () => path.join(ROOT, "subscriptions.json");

// Re-exported so every existing importer keeps its import path. The vocabulary
// itself is defined beside the emitters (record-events.ts) — Task 9.2: a kind
// that can be subscribed to but never emitted is unsatisfiable BY CONSTRUCTION
// only if the enum cannot be widened without an emitter.
export { EVENT_KINDS, EVENT_KIND_IDS };
export type { RecordEvent, SubKind };
export type Subscription = {
  id: string;
  agentId: string;
  kind: SubKind;
  target: string;
  createdAt: number;
  /**
   * null until an event OF THIS KIND was evaluated against this subscription.
   *
   * NOT "the machinery ran" — see lastScannedAt. Those were one field, and the
   * collapse is the defect: `evaluate()` only touches subscriptions whose kind
   * matches the event, so an `item` subscriber sat at `never evaluated` for a
   * week while `pr` events flowed past it. Its health said the wiring was dead;
   * the wiring was fine and nothing of its kind had happened.
   */
  lastEvaluatedAt: number | null;
  /**
   * null until a scan RAN with this subscription registered — regardless of
   * kind. A scan that ran is evidence the wiring works; it says nothing about
   * whether your kind fired, which is the other field.
   *
   * Optional on disk: subscriptions written before this existed have no value,
   * and absent must read as UNKNOWN rather than as "never scanned". Fabricating
   * a scan nobody observed is the failure this whole phase is about.
   */
  lastScannedAt?: number | null;
  lastEventAt: number | null;
  /** Idempotency keys already delivered, for 6.4. */
  delivered: string[];
};

export function readSubs(): Subscription[] {
  const f = subsFile();
  if (!existsSync(f)) return [];
  try {
    return (JSON.parse(readFileSync(f, "utf8")).subscriptions ?? []) as Subscription[];
  } catch {
    return [];
  }
}

function writeSubs(subs: Subscription[]): void {
  mkdirSync(ROOT, { recursive: true });
  writeFileSync(subsFile(), `${JSON.stringify({ subscriptions: subs }, null, 2)}\n`);
}

/**
 * 6.3 — A SUBSCRIPTION THAT NEVER FIRES MUST BE DISTINGUISHABLE FROM ONE NEVER
 * REGISTERED, so never-evaluated is an ERROR rather than a quiet zero.
 *
 * We hit the absence of this rule twice in two days: the console's standing
 * trigger and `stall_check`'s clock both looked healthy while never running. A
 * subscription with no last-evaluated mark has produced no evidence of
 * anything, and "no events" is the same output a broken subscription gives.
 */
/**
 * ⟨q-2b7d9f04⟩ — CAPABILITY, NOT LIVENESS. The `item` subscription read
 * `health: ok — no events yet` for three days while the scanner's grammar
 * could not produce its kind from any commit: the scan clock was fresh, the
 * emitter list had an entry, and the field answered the easier question. The
 * probe is registered by record-events.ts (which owns the grammar) and asks
 * "can this kind be produced at all?"; a kind that cannot is an ERROR, however
 * recently the scanner ran. Absent probe → no claim either way.
 */
let capabilityProbe: ((kind: SubKind) => boolean) | null = null;
export function setCapabilityProbe(fn: ((kind: SubKind) => boolean) | null): void {
  capabilityProbe = fn;
}
export const kindCanFire = (kind: SubKind): boolean | null => (capabilityProbe ? capabilityProbe(kind) : null);

export function subscriptionHealth(s: Subscription): { level: "ok" | "error" | "unknown"; detail: string } {
  const iso = (n: number) => new Date(n).toISOString();
  // ⟨q-2b7d9f04⟩ — a kind the grammar cannot produce is not ok, scanned or not.
  if (kindCanFire(s.kind) === false) {
    return {
      level: "error",
      detail:
        `CANNOT FIRE — the scanner's grammar produces no '${s.kind}' event from the kind's own probe, so this subscription is structurally unable to deliver ` +
        `${s.lastScannedAt ? `(scanned ${iso(s.lastScannedAt)}: the clock is live, the capability is not)` : "(and has never been scanned)"}. ` +
        "Never fired and cannot fire are different facts; this is the second.",
    };
  }

  // THREE STATES, BECAUSE THERE ARE THREE FACTS. They were two, and the
  // collapse cost a week: an `item` subscriber read `error — never evaluated`
  // the whole time, because `evaluate()` only touches subscriptions whose kind
  // matches the event and no `item` event had fired. The wiring was fine. The
  // health field said it was dead, and the agent holding it believed the field.
  //
  // Same shape as `alive` swallowing `heartbeatFresh`, and as
  // `stall_clock_status` reporting green while covering nothing: one field
  // answering two questions always answers the easier one.
  if (s.lastScannedAt === undefined && s.lastEvaluatedAt === null) {
    // Written before this field existed AND never evaluated. Absent is UNKNOWN,
    // never "never scanned" — inventing a scan nobody observed is the failure
    // this phase is about, and asserting one never happened is the same error
    // pointed the other way.
    return {
      level: "unknown",
      detail:
        "predates scan tracking and has never been evaluated — whether the machinery has run for it CANNOT be determined from this record. " +
        "It will resolve to ok or error on the next scan; until then this is an absence of evidence, not evidence of absence.",
    };
  }
  if (!s.lastScannedAt) {
    return {
      level: "error",
      detail:
        "NEVER SCANNED — no scan has run with this subscription registered, so it has produced no evidence of being wired to anything. " +
        "This is the state that means broken.",
    };
  }
  if (s.lastEvaluatedAt === null) {
    // THE STATE THAT USED TO READ AS BROKEN. The machinery ran and nothing of
    // this kind happened, which is a healthy idle watch.
    return {
      level: "ok",
      detail:
        `scanned ${iso(s.lastScannedAt)}, never evaluated — the machinery RAN and no '${s.kind}' event has occurred yet. ` +
        "Scanned-and-quiet is healthy; it is not the same as never run, and those shared a field until now.",
    };
  }
  return {
    level: "ok",
    detail: s.lastEventAt
      ? `scanned ${iso(s.lastScannedAt)}, last evaluated ${iso(s.lastEvaluatedAt)}, last event ${iso(s.lastEventAt)}`
      : `scanned ${iso(s.lastScannedAt)}, last evaluated ${iso(s.lastEvaluatedAt)}, no events yet`,
  };
}

/**
 * 6.4 — DELIVERY IS AT-LEAST-ONCE BY DESIGN. Safe for a reader, DOUBLE
 * EXECUTION for an executor: a callback that triggers work must carry a key, or
 * the same merge lands twice. The key is derived from the EVENT, never from the
 * delivery attempt, so a retry produces the same key.
 */
export const eventKey = (kind: SubKind, target: string, ref: string): string =>
  createHash("sha256").update(`${kind}:${target}:${ref}`).digest("hex").slice(0, 16);

/**
 * 6.2 — EVENTS ARE DERIVED FROM THE RECORD, NEVER PARALLEL TO IT.
 *
 * Enforced here rather than promised in a comment: the event's `ref` must
 * already be present in the record document before anything is emitted. An
 * event stream that can say "task X complete" while DONE.md does not is a
 * second source of truth, and record-vs-state divergence is the defect this
 * fleet hit most this week. ADR-003 keeps markdown authoritative, and this must
 * not quietly reopen it.
 *
 * So the ordering is not a convention: emission READS the record, and an event
 * whose cause is not in the record cannot be emitted at all.
 */
export function eventIsDerived(recordText: string, ev: RecordEvent): { ok: true } | { ok: false; error: string } {
  if (!ev.ref) return { ok: false, error: `event for ${ev.kind} ${ev.target} carries no ref — nothing ties it to a record entry` };
  // ⟨q-cbace757⟩ — a multi-PR closer is N citations, checked ONE BY ONE. The
  // joined string ("a#1, a#2, a#3") is how the event is keyed, not how the record
  // is read: a DONE line citing the same three PRs in another arrangement carries
  // every ref and none of the joined form. The guard stays exactly as strict per
  // ref — a citation absent from the record still refuses, and is named.
  const refs = ev.refs?.length ? ev.refs : [ev.ref];
  const missing = refs.filter((r) => !String(recordText).includes(r));
  if (missing.length)
    return {
      ok: false,
      error:
        `refusing to emit ${ev.kind} ${ev.target}: ${missing.length === refs.length && refs.length === 1 ? `its ref ${ev.ref}` : `${missing.length} of its ${refs.length} cited ref(s) — ${missing.join(", ")} —`} is NOT in the record. ` +
        `An event that exists without the record change that caused it is a second source of truth — ` +
        `the stream would claim something the authoritative document does not.`,
    };
  return { ok: true };
}

/** Subscriptions matching an event. Exact target match; no wildcards yet. */
export const matching = (subs: Subscription[], ev: RecordEvent): Subscription[] =>
  subs.filter((s) => s.kind === ev.kind && s.target === ev.target);

export type Delivery = { subscriptionId: string; agentId: string; key: string; status: "delivered" | "duplicate-suppressed" };

/**
 * Evaluate every subscription against one event and return what to deliver.
 *
 * EVALUATION IS RECORDED EVEN WHEN NOTHING MATCHES — that is 6.3's whole point.
 * A subscription only learns it is alive by being evaluated, so the mark is
 * written for every subscription of that kind, not only the ones that fired.
 */
/**
 * Record that a scan RAN, for every live subscription regardless of kind.
 *
 * Called once per scan — including a scan that produced NO events, which is
 * exactly the case that starved the old field: no events of your kind means
 * `evaluate` never touches you, and a subscription that is never touched cannot
 * be told from one that is not wired to anything.
 */
export function markScanned(subs: Subscription[], now: number): Subscription[] {
  return subs.map((s) => ({ ...s, lastScannedAt: now }));
}

export function evaluate(subs: Subscription[], ev: RecordEvent, now: number): { subs: Subscription[]; deliveries: Delivery[] } {
  const key = eventKey(ev.kind, ev.target, ev.ref);
  const deliveries: Delivery[] = [];
  const next = subs.map((s) => {
    // Every subscription observed this scan, whatever its kind.
    const scanned = { ...s, lastScannedAt: now };
    if (s.kind !== ev.kind) return scanned;
    const evaluated = { ...scanned, lastEvaluatedAt: now };
    if (s.target !== ev.target) return evaluated;
    if (s.delivered.includes(key)) {
      deliveries.push({ subscriptionId: s.id, agentId: s.agentId, key, status: "duplicate-suppressed" });
      return evaluated;
    }
    deliveries.push({ subscriptionId: s.id, agentId: s.agentId, key, status: "delivered" });
    return { ...evaluated, lastEventAt: now, delivered: [...s.delivered, key].slice(-200) };
  });
  return { subs: next, deliveries };
}

/* ── verbs ─────────────────────────────────────────────────────────────────── */

export const subscribeSchema = {
  agentId: z.string().min(1),
  kind: z.enum(EVENT_KIND_IDS),
  target: z.string().min(1),
};

export async function subscribeTool(args: { agentId: string; kind: SubKind; target: string }) {
  const subs = readSubs();
  const dupe = subs.find((s) => s.agentId === args.agentId && s.kind === args.kind && s.target === args.target);
  if (dupe) return { ok: true as const, subscription: dupe, note: "already subscribed — returning the existing registration rather than a second one" };
  const sub: Subscription = {
    id: randomUUID(),
    agentId: args.agentId,
    kind: args.kind,
    target: args.target,
    createdAt: Date.now(),
    // EXPLICIT null, never left absent. `null` means "we know no scan has run
    // for this"; `undefined` means "this record predates the field and we
    // cannot say". A subscription created now is the first case, and writing it
    // explicitly is what keeps a brand-new registration from being reported as
    // unknowable.
    lastScannedAt: null,
    lastEvaluatedAt: null,
    lastEventAt: null,
    delivered: [],
  };
  writeSubs([...subs, sub]);
  return { ok: true as const, subscription: sub, health: subscriptionHealth(sub) };
}

export const unsubscribeSchema = { agentId: z.string().min(1), id: z.string().min(1) };

export async function unsubscribeTool(args: { agentId: string; id: string }) {
  const subs = readSubs();
  const sub = subs.find((s) => s.id === args.id);
  if (!sub) return { ok: false as const, error: `no subscription '${args.id}'` };
  // Another agent's subscription is not yours to remove: silently dropping
  // someone else's notification is how a miss is manufactured.
  if (sub.agentId !== args.agentId) return { ok: false as const, error: `subscription '${args.id}' belongs to '${sub.agentId}', not '${args.agentId}'` };
  writeSubs(subs.filter((s) => s.id !== args.id));
  return { ok: true as const, removed: sub };
}

export const listSubscriptionsSchema = { agentId: z.string().optional() };

export async function listSubscriptionsTool(args: { agentId?: string }) {
  const all = readSubs();
  const subs = args.agentId ? all.filter((s) => s.agentId === args.agentId) : all;
  const rows = subs.map((s) => ({ ...s, health: subscriptionHealth(s) }));
  const cannotFire = rows.filter((r) => r.health.level === "error" && /^CANNOT FIRE/.test(r.health.detail));
  const neverEvaluated = rows.filter((r) => r.health.level === "error" && !/^CANNOT FIRE/.test(r.health.detail));
  const undetermined = rows.filter((r) => r.health.level === "unknown");
  return {
    ok: neverEvaluated.length === 0 && cannotFire.length === 0,
    // ⟨q-2b7d9f04⟩ — capability per kind, beside each row's liveness.
    capability: Object.fromEntries(EVENT_KIND_IDS.map((k) => [k, kindCanFire(k)])),
    ...(cannotFire.length
      ? { cannotFire: `${cannotFire.length} of ${rows.length} subscription(s) are to a kind the scanner CANNOT PRODUCE — they will never fire, however live the scan clock reads: ${[...new Set(cannotFire.map((r) => r.kind))].join(", ")}.` }
      : {}),
    // Population beside the verdict, always: "no subscriptions" and "none
    // listed for you" are different claims.
    population: { listed: rows.length, total: all.length },
    subscriptions: rows,
    ...(neverEvaluated.length
      ? { error: `${neverEvaluated.length} of ${rows.length} subscription(s) have NEVER BEEN SCANNED — no scan has run with them registered, so they have produced no evidence of being wired to anything.` }
      : {}),
    ...(undetermined.length
      ? {
          undetermined: `${undetermined.length} of ${rows.length} subscription(s) predate scan tracking and cannot be judged yet — they resolve on the next scan. Reported rather than counted as either healthy or broken.`,
        }
      : {}),
  };
}

/** Persist an evaluation. Callers do this after a record write, never before. */
export function commitEvaluation(next: Subscription[]): void {
  writeSubs(next);
}