// The standing audit: an independent periodic reconcile, not a per-action log,
// because the failure mode is a no-event blind spot. A mis-registered resource
// or a stray account-wide token leaks nothing until an agent happens to touch
// it. This reconciles what Cloudflare has against what the registry records and
// against what the per-account secrets files still carry.

export type StrayKind = "unattributed-d1" | "unattributed-r2" | "account-wide-token";

export interface Stray {
  kind: StrayKind;
  detail: string;
}

// Keys that must never remain in a sub-account's cloudflare.env after
// remediation. Each is an account-wide capability.
const ACCOUNT_WIDE_KEYS = ["CLOUDFLARE_API_TOKEN", "CF_PAGES_D1_TOKEN", "CF_PAGES_TOKEN"];

/** CONCURRENCY CONTRACT — read before implementing these callbacks.
 *
 *  reconcileStorage invokes all five through ONE Promise.all. They overlap,
 *  including the two registeredNames calls, which run at the same time as each
 *  other. So no callback may share, with any other callback or with a second
 *  invocation of itself, a resource that admits one operation at a time.
 *
 *  The resource that has actually caused this is a Neo4j session: it rejects a
 *  query issued while one of its own is in flight. A caller that opened one
 *  session in the enclosing scope and used it in registeredNames made the
 *  second read reject, which left the standing storage audit reporting nothing
 *  for 802 consecutive ticks on maxy-code and 295 on sitedesk-code (Task 1813).
 *  Open the session inside the callback, one per invocation, and close it
 *  there; driver sessions come from a pool, so this is cheap.
 *
 *  The contract is stated here rather than on reconcileStorage because this
 *  interface is what a caller writing an implementation is looking at, and the
 *  natural thing to write is the thing that breaks. */
export interface AuditDeps {
  cfD1(): Promise<string[]>;
  cfR2(): Promise<string[]>;
  /** Called twice, concurrently, once per kind. See the contract above. */
  registeredNames(kind: "d1" | "r2"): Promise<string[]>;
  accountSecretsFiles(): Promise<Array<{ accountId: string; path: string; contents: string }>>;
}

export async function reconcileStorage(deps: AuditDeps): Promise<{ strays: Stray[] }> {
  const strays: Stray[] = [];

  const [cfD1, cfR2, regD1, regR2, secretsFiles] = await Promise.all([
    deps.cfD1(),
    deps.cfR2(),
    deps.registeredNames("d1"),
    deps.registeredNames("r2"),
    deps.accountSecretsFiles(),
  ]);

  const regD1Set = new Set(regD1);
  for (const name of cfD1) {
    if (!regD1Set.has(name)) strays.push({ kind: "unattributed-d1", detail: name });
  }
  const regR2Set = new Set(regR2);
  for (const name of cfR2) {
    if (!regR2Set.has(name)) strays.push({ kind: "unattributed-r2", detail: name });
  }

  for (const file of secretsFiles) {
    for (const line of file.contents.split("\n")) {
      const key = line.match(/^\s*([A-Za-z_][A-Za-z0-9_]*)=/)?.[1];
      if (key && ACCOUNT_WIDE_KEYS.includes(key)) {
        strays.push({ kind: "account-wide-token", detail: `${file.accountId}:${key}` });
      }
    }
  }

  return { strays };
}

// --- Pages ownership reconcile (Task 1728) --------------------------------
//
// Same no-event argument as above, one class further on. Publishing is gated on
// an ownership record, and both ways that record can be wrong emit nothing and
// never reproduce on demand:
//
//   orphan  — a live Pages project with no owner recorded. Nobody may publish
//             it, and nothing says so until a customer asks for a change and the
//             deploy denies "unregistered". Every project predating this task is
//             an orphan, so on day one this count is the adoption worklist.
//   phantom — an ownership record naming a project Cloudflare does not have.
//             Publishing it fails at wrangler, having already passed the
//             ownership check, so the record lies until someone tries.
//
// Names are reported, not just counted: a count tells an operator something is
// wrong, the name tells them what to adopt.

/** Same concurrency contract as AuditDeps above: reconcilePages invokes both
 *  callbacks through one Promise.all, so neither may share a single-operation
 *  resource with the other.
 *
 *  That registeredProjects happens to issue exactly one query today is not the
 *  property that makes a shared session safe here, and relying on it would put
 *  the 1813 defect one added read away. */
export interface PagesAuditDeps {
  cfProjects(): Promise<string[]>;
  registeredProjects(): Promise<string[]>;
}

export interface PagesReconcile {
  projects: number;
  registered: number;
  orphans: string[];
  phantoms: string[];
}

export async function reconcilePages(deps: PagesAuditDeps): Promise<PagesReconcile> {
  // Not Promise.all with a catch that swallows one side: if either read fails
  // the reconcile is meaningless, and reporting "orphan=<everything>" off a
  // failed registry read would be worse than reporting nothing. Let it throw to
  // the caller, which logs it as an audit error.
  const [cf, registered] = await Promise.all([deps.cfProjects(), deps.registeredProjects()]);
  const cfSet = new Set(cf);
  const regSet = new Set(registered);
  return {
    projects: cfSet.size,
    registered: regSet.size,
    orphans: [...cfSet].filter((name) => !regSet.has(name)),
    phantoms: [...regSet].filter((name) => !cfSet.has(name)),
  };
}

// --- Data portal reconcile (Task 1704) -------------------------------------
//
// Third instance of the same no-event argument, one level down from the two
// above: those reconcile ownership records, this reconciles the objects
// themselves against the manifest rows that make them ingestible.
//
//   orphan object — an object in R2 with no manifest row. The upload returned
//                   200, the file is stored, and ingestion will never see it.
//   phantom row   — a manifest row whose object is absent. Ingestion will pull
//                   a missing file, and nothing says so until it runs.
//
// Keys are returned rather than only counted so the arithmetic is testable and
// a future caller has them, but the standing audit logs COUNTS ONLY: a key here
// is `<ownerId>/<filename>`, a real person's filename, and the brand journal is
// not where those belong. That is the same ownership discipline the in-process
// audit path owes for holding the account-wide credential and bypassing
// objectGate entirely.

export interface DataPortalObject {
  key: string;
}

export interface DataPortalRow {
  objectKey: string;
  ingested: number;
  uploadedAt: string;
  /** Where the bytes landed on the device, set only after a verified write
   *  (Task 1910). Its presence IS the proof the move happened, which is what
   *  separates a healthy moved row from a phantom. Absent on a portal whose
   *  hand-run ALTER has not happened yet, which reads as "never moved". */
  devicePath?: string;
  /** 1 when the client has withdrawn an already-moved file and the device has
   *  not yet removed it. */
  deleted?: number;
}

export interface DataPortalReconcile {
  /** DISTINCT keys, which is not the same quantity as the length of the list
   *  the read returned. The caller logs the lengths — "how many did I see" —
   *  and these are "how many are distinct". They agree on a healthy store
   *  (manifest.objectKey is UNIQUE and R2 keys are unique by construction), and
   *  giving them one name would conflate two facts that diverge exactly when
   *  the store is unhealthy, which is the case this audit is for. */
  distinctObjects: number;
  distinctRows: number;
  orphanObjects: string[];
  phantomRows: string[];
  /** Null when nothing is uningested, or when no uningested row carries a
   *  parseable timestamp. Rendered `na` by the caller. */
  oldestUningestedHrs: number | null;
  /** Every row at `ingested = 0`, including rows whose `uploadedAt` does not
   *  parse. Deliberately NOT the size of the age sample below: that sample
   *  excludes unparseable timestamps, so a backlog made entirely of them would
   *  report a null age beside a zero count and read as an empty queue. */
  uningestedRows: number;
  /** Objects still in R2 whose row says the bytes already reached the device
   *  (Task 1910). The move deletes the object after the claim commits, so a
   *  survivor is a leak. It emits nothing at request time — the pull has
   *  already returned and no later request touches the key — which is why it
   *  is counted here rather than logged somewhere. */
  unmovedObjects: string[];
  /** Withdrawals the device has not carried out yet. Healthy is 0; a number
   *  that persists across cycles is a withdrawal that will never complete. */
  deletePendingRows: number;
  /** Age of the oldest pending withdrawal, so a stuck one is dateable rather
   *  than merely present. Null when none are pending. */
  oldestDeletePendingHrs: number | null;
  /** Files routed into an exposed folder that the published index does not
   *  carry. The client uploaded into a folder, the file landed, and the tree
   *  still does not show it.
   *
   *  NULL when the index was not supplied, never 0: an unread index is "not
   *  measured", and rendering it as zero would let a failed read present as a
   *  clean bill of health. */
  unpublishedRouted: number | null;
}

/** Pure over two lists and an injected clock, so the age arithmetic is
 *  deterministic under test rather than dependent on when the suite runs.
 *
 *  Unlike its two siblings this takes lists rather than callbacks: both sides
 *  are read per account by a caller that must catch each independently, so
 *  there is no fan-out here to carry a concurrency contract. */
export function reconcileDataPortal(
  objects: DataPortalObject[],
  rows: DataPortalRow[],
  nowMs: number,
  publishedPaths?: Set<string>,
): DataPortalReconcile {
  const objectKeys = new Set(objects.map((o) => o.key));
  const rowKeys = new Set(rows.map((r) => r.objectKey));

  // A moved row's bytes are deliberately NOT in R2, so it is neither a phantom
  // nor an orphan — it is the healthy end state of the exchange. Splitting the
  // row set here is what keeps the pre-existing counts meaning what they meant
  // before the move existed.
  const moved = rows.filter((r) => Boolean(r.devicePath));
  const movedKeys = new Set(moved.map((r) => r.objectKey));

  // Date.parse of a truthy non-ISO value is NaN, which slips past a `??` guard
  // and renders as `oldestUningestedHrs=NaN` — the defect the availability
  // loop's snapshotAgeMs already had to fix. Unparseable rows are excluded from
  // the minimum; if that leaves none, the answer is null, never NaN.
  const uningested = rows.filter((r) => r.ingested === 0);
  const uningestedMs = uningested
    .map((r) => Date.parse(r.uploadedAt))
    .filter((ms) => !Number.isNaN(ms));

  // Reduced, never `Math.min(...uningestedMs)`. A spread passes one argument per
  // element and blows the call stack — measured: fine at 125k, RangeError at
  // 130k. One element per uningested row, and an ingestion sweep that has
  // stalled accumulates those without bound. A stalled sweep is one of the three
  // conditions this audit exists to reveal, so the arithmetic must not be what
  // goes dark once the backlog it is watching grows large. The throw would also
  // escape the caller's per-side catches and cost every later account its line.
  const oldestMs = uningestedMs.reduce((min, ms) => (ms < min ? ms : min), Infinity);

  const pendingDeletes = rows.filter((r) => r.deleted === 1);
  const pendingDeleteMs = pendingDeletes
    .map((r) => Date.parse(r.uploadedAt))
    .filter((ms) => !Number.isNaN(ms));
  // Reduced rather than spread, for the same stack-overflow reason as above.
  const oldestPendingMs = pendingDeleteMs.reduce((min, ms) => (ms < min ? ms : min), Infinity);

  return {
    distinctObjects: objectKeys.size,
    distinctRows: rowKeys.size,
    orphanObjects: [...objectKeys].filter((k) => !rowKeys.has(k)),
    // A moved row is excluded: its object is SUPPOSED to be gone, so counting
    // it would manufacture a defect on every healthy exchange.
    phantomRows: [...rowKeys].filter((k) => !objectKeys.has(k) && !movedKeys.has(k)),
    oldestUningestedHrs: uningestedMs.length === 0 ? null : (nowMs - oldestMs) / 3_600_000,
    uningestedRows: uningested.length,
    unmovedObjects: [...movedKeys].filter((k) => objectKeys.has(k)),
    deletePendingRows: pendingDeletes.length,
    oldestDeletePendingHrs:
      pendingDeleteMs.length === 0 ? null : (nowMs - oldestPendingMs) / 3_600_000,
    unpublishedRouted:
      publishedPaths === undefined
        ? null
        : moved.filter(
            (r) =>
              // uploads/ is never published by design, so only a file routed
              // into an exposed folder can be "unpublished".
              !r.devicePath!.startsWith("uploads/") && !publishedPaths.has(r.devicePath!),
          ).length,
  };
}