// Replay reader — reads `intent_audit` rows back into AuditRecord instances
// for the replay harness. Inverse of recordToRow().
//
// Adopters supply a query function that returns rows; this module reconstructs
// AuditRecord objects so the standard `replay()` from @adjudicate/audit can
// re-adjudicate them.

import type {
  AuditPlanSnapshot,
  AuditRecord,
  AuditRecordVersion,
  Decision,
  IntentEnvelope,
  RecordedAggregateSnapshot,
  RecordedAuthoritySnapshot,
  Supersession,
} from "@adjudicate/core";
import type { IntentAuditRow } from "./postgres-sink.js";

export interface AuditQueryFnWindow {
  readonly fromIso: string;
  readonly toIso: string;
  readonly intentKind?: string;
  readonly limit?: number;
}

export interface AuditQueryFn {
  /**
   * Return rows whose `recorded_at` falls within the inclusive window
   * [fromIso, toIso] (APIReviewer-003 boundary convention — both ends
   * inclusive, matching the admin-sdk audit query window). Optional filter by
   * `intentKind`. Limit caps the result set; adopters may stream via repeated
   * calls if needed.
   */
  fetchRows(window: AuditQueryFnWindow): Promise<readonly IntentAuditRow[]>;
}

/**
 * Reconstruct an AuditRecord from a stored row. Inverse of recordToRow().
 *
 * `decision_basis`: the TEXT[] column (`decision_basis`) is a query-optimized
 * projection (written by `recordToRow` as `category:code` strings for SQL-side
 * `WHERE … = ANY(decision_basis)` filtering). The reader reconstructs
 * `AuditRecord.decision_basis` from `decision_jsonb.basis`, NOT the TEXT[]
 * column — the JSONB carries the full structured `DecisionBasis[]` including
 * `detail`. If the TEXT[] and JSONB ever diverge (writer bug / malicious row),
 * this reader silently prefers JSONB. See `recordToRow` for the dual-encoding
 * invariant: TEXT[] must equal `decision.basis.map(b => "${b.category}:${b.code}")`.
 *
 * Version dispatch:
 *   - `record_version` NULL or 1 → v1 row (no plan field, no nonce).
 *   - `record_version = 2` → v2 row. `plan` populated from `plan_jsonb`
 *     when present. `nonce` populated from the row column or the stored
 *     envelope JSON (T8).
 *
 * For replay drift detection on v1 rows, use `legacyV1ToV2(row)` to
 * synthesize a v2 envelope from the historical createdAt — the original
 * `intentHash` does NOT reproduce (different recipe) but the Decision
 * does, so kind/basis comparison is meaningful.
 */
/**
 * Parse a jsonb column value tolerantly. node-postgres returns `jsonb`/`json`
 * columns as ALREADY-PARSED objects by default; a reader that configures a text
 * type parser (or casts `::text`) returns the raw string. Accept BOTH so the
 * reader works with any pool config — a plain `new pg.Pool()` (e.g. the operator
 * console's) returns objects, and `JSON.parse(object)` would throw
 * `"[object Object]" is not valid JSON`. Postgres normalizes jsonb either way,
 * so the reconstructed object is identical and auditHash verification is
 * unaffected.
 */
function parseJsonb(value: unknown): unknown {
  return typeof value === "string" ? JSON.parse(value) : value;
}

export function rowToRecord(row: IntentAuditRow): AuditRecord {
  const envelope = parseJsonb(row.envelope_jsonb) as IntentEnvelope;
  const decision = parseJsonb(row.decision_jsonb) as Decision;
  const version: AuditRecordVersion =
    row.record_version === 5
      ? 5
      : row.record_version === 4
        ? 4
        : row.record_version === 3
          ? 3
          : row.record_version === 2
            ? 2
            : 1;
  const plan: AuditPlanSnapshot | undefined =
    version >= 2 && row.plan_jsonb
      ? (parseJsonb(row.plan_jsonb) as AuditPlanSnapshot)
      : undefined;
  const supersedes: Supersession | undefined =
    version >= 3 && row.supersedes_jsonb
      ? (parseJsonb(row.supersedes_jsonb) as Supersession)
      : undefined;
  // v3+ kernelIdentity and v4+ policyVersion/kernelVersion are part of the
  // v4 auditHash pre-image. They MUST be reconstructed with the same
  // key-presence buildAuditRecord used (omit when absent, never `undefined`-
  // valued) or verifyAuditRecord re-derives a different hash and reports
  // false-positive tampering. auditHash + signature are excluded from the
  // pre-image by verifyAuditRecord, but auditHash is restored so verification
  // has a stored value to compare against.
  const kernelIdentity: NonNullable<AuditRecord["kernelIdentity"]> | undefined =
    version >= 3 && row.kernel_identity_jsonb
      ? (parseJsonb(row.kernel_identity_jsonb) as NonNullable<
          AuditRecord["kernelIdentity"]
        >)
      : undefined;
  const signature: NonNullable<AuditRecord["signature"]> | undefined =
    version >= 4 && row.signature_jsonb
      ? (parseJsonb(row.signature_jsonb) as NonNullable<
          AuditRecord["signature"]
        >)
      : undefined;
  // v5+ metadata is EXCLUDED from the auditHash pre-image, so unlike the v4
  // fields above it does NOT need presence-exact reconstruction for verify —
  // but it is restored so the round-trip is lossless and the console can read it.
  const metadata: NonNullable<AuditRecord["metadata"]> | undefined =
    version >= 5 && row.metadata_jsonb
      ? (parseJsonb(row.metadata_jsonb) as NonNullable<AuditRecord["metadata"]>)
      : undefined;
  // 033/052 read-path completion (093 / 092-F1): authoritySnapshot and
  // aggregateSnapshot are BOTH part of the v4+ auditHash pre-image. They MUST be
  // reconstructed with the SAME key-presence buildAuditRecord used (present when
  // the column is non-NULL, OMITTED otherwise — never an `undefined`-valued key)
  // or verifyAuditRecord re-derives a different hash and 092 verify-on-read
  // FALSELY flags the record tampered. Gated on version >= 4 (the version the
  // auditHash itself appears at); a NULL column omits the field, hash-stable.
  const authoritySnapshot: RecordedAuthoritySnapshot | undefined =
    version >= 4 && row.authority_snapshot_jsonb
      ? (parseJsonb(row.authority_snapshot_jsonb) as RecordedAuthoritySnapshot)
      : undefined;
  const aggregateSnapshot: RecordedAggregateSnapshot | undefined =
    version >= 4 && row.aggregate_snapshot_jsonb
      ? (parseJsonb(row.aggregate_snapshot_jsonb) as RecordedAggregateSnapshot)
      : undefined;
  return {
    version,
    intentHash: row.intent_hash,
    envelope,
    decision,
    decision_basis: decision.basis,
    resourceVersion: row.resource_version ?? undefined,
    at: row.recorded_at,
    durationMs: row.duration_ms,
    ...(plan !== undefined ? { plan } : {}),
    ...(supersedes !== undefined ? { supersedes } : {}),
    ...(kernelIdentity !== undefined ? { kernelIdentity } : {}),
    ...(version >= 4 && row.policy_version != null
      ? { policyVersion: row.policy_version }
      : {}),
    ...(version >= 4 && row.kernel_version != null
      ? { kernelVersion: row.kernel_version }
      : {}),
    // 033/052 — recorded snapshots, IN the auditHash pre-image: place BEFORE
    // auditHash/signature in the field stream and OMIT when absent so the
    // reconstructed record hashes byte-identically to the one buildAuditRecord
    // produced (verify-on-read does not false-tamper).
    ...(authoritySnapshot !== undefined ? { authoritySnapshot } : {}),
    ...(aggregateSnapshot !== undefined ? { aggregateSnapshot } : {}),
    ...(version >= 4 && row.audit_hash != null
      ? { auditHash: row.audit_hash }
      : {}),
    ...(signature !== undefined ? { signature } : {}),
    ...(metadata !== undefined ? { metadata } : {}),
    // 093 — the inter-record chain link, EXCLUDED from the auditHash pre-image
    // (like signature/metadata), so it can ride anywhere in the field stream and
    // omitting it (genesis / pre-093) is hash-stable. Round-tripped so the chain-
    // continuity harness + supersession-chain report can read it.
    ...(row.prev_audit_hash != null
      ? { prevAuditHash: row.prev_audit_hash }
      : {}),
  };
}

/**
 * Read a window of audit rows and return them as AuditRecord[] suitable for
 * `replay()` from @adjudicate/audit.
 */
export async function readAuditWindow(
  query: AuditQueryFn,
  window: AuditQueryFnWindow,
): Promise<AuditRecord[]> {
  const rows = await query.fetchRows(window);
  return rows.map(rowToRecord);
}