/**
 * `createPostgresAuditStore` — implements the SDK's `AuditStore` contract
 * against the existing `intent_audit` table.
 *
 * Reuses `rowToRecord` from `replay.ts` (single source of truth for the
 * row→AuditRecord conversion) and translates the SDK's filter shape into
 * parameterized SQL.
 *
 * **Pagination is keyset, not offset.** The SDK's `cursor: string` is
 * opaque — we encode `(recorded_at, intent_hash)` of the last seen row
 * as base64url(JSON). Subsequent pages add a strict-less-than predicate
 * against that tuple. Result: O(log N) page latency at any depth — page
 * 100,000 of a 10M-row table runs in the same time as page 2. Offset
 * pagination would scan linearly.
 *
 * **Tiebreaker direction matches the primary sort.** ORDER BY
 * `recorded_at DESC, intent_hash DESC` and the cursor predicate uses
 * tuple < — both DESC. Mismatching directions causes row skipping
 * during millisecond-burst inserts (webhook fan-out is the canonical
 * trigger).
 */

import {
  verifyAuditRecord,
  type AuditRecord,
  type AuditRecordVerification,
} from "@adjudicate/core";
import type {
  AuditQuery,
  AuditQueryResult,
  AuditStore,
} from "@adjudicate/admin-sdk";
import type { PostgresReader } from "./pg-reader.js";
import type { IntentAuditRow } from "./postgres-sink.js";
import { normalizeTimestamptz } from "./pg-types.js";
import { rowToRecord } from "./replay.js";

const SELECT_COLUMNS = `
  intent_hash, session_id, kind, principal, taint, decision_kind,
  refusal_kind, refusal_code, decision_basis, resource_version,
  envelope_jsonb, decision_jsonb, recorded_at, duration_ms,
  partition_month, record_version, plan_jsonb, nonce, supersedes_jsonb,
  kernel_identity_jsonb, policy_version, kernel_version, audit_hash,
  signature_jsonb, metadata_jsonb, prev_audit_hash,
  authority_snapshot_jsonb, aggregate_snapshot_jsonb
`.trim();

/**
 * Thrown when a non-empty `cursor` fails to decode (malformed / truncated /
 * tampered). Silently restarting from page 1 — the previous behavior — hands
 * the caller a `nextCursor` for the SAME first page, an infinite loop for any
 * client that retries. The tRPC layer should map this to a `BAD_REQUEST`.
 */
export class InvalidCursorError extends Error {
  constructor(message = "Cursor is malformed or has been tampered with.") {
    super(message);
    this.name = "InvalidCursorError";
  }
}

interface CursorPayload {
  readonly at: string;
  readonly hash: string;
}

export function encodeCursor(p: CursorPayload): string {
  return Buffer.from(JSON.stringify(p), "utf-8").toString("base64url");
}

export function decodeCursor(s: string): CursorPayload | null {
  try {
    const json = Buffer.from(s, "base64url").toString("utf-8");
    const p = JSON.parse(json) as Partial<CursorPayload>;
    if (typeof p.at === "string" && typeof p.hash === "string") {
      return { at: p.at, hash: p.hash };
    }
    return null;
  } catch {
    return null;
  }
}

interface SqlFragment {
  readonly clauses: readonly string[];
  readonly params: readonly unknown[];
}

/**
 * Builds the WHERE clause set from the SDK's filter shape. Parameter
 * indices are monotonic ($1, $2, ...). Each provided filter contributes
 * one clause; absent filters contribute none. AND-composed.
 */
export function buildWhereClauses(q: AuditQuery): SqlFragment {
  const clauses: string[] = [];
  const params: unknown[] = [];
  let i = 1;

  if (q.intentKind !== undefined) {
    clauses.push(`kind = $${i++}`);
    params.push(q.intentKind);
  }
  if (q.decisionKind !== undefined) {
    clauses.push(`decision_kind = $${i++}`);
    params.push(q.decisionKind);
  }
  if (q.refusalCode !== undefined) {
    // refusal_code is NULL except on REFUSE rows (CHECK constraint
    // intent_audit_refusal_pair guarantees this). Filtering by code
    // implicitly narrows to REFUSE.
    clauses.push(`refusal_code = $${i++}`);
    params.push(q.refusalCode);
  }
  if (q.taint !== undefined) {
    clauses.push(`taint = $${i++}`);
    params.push(q.taint);
  }
  if (q.intentHash !== undefined) {
    clauses.push(`intent_hash = $${i++}`);
    params.push(q.intentHash);
  }
  if (q.since !== undefined) {
    clauses.push(`recorded_at >= $${i++}`);
    params.push(q.since);
  }
  if (q.until !== undefined) {
    clauses.push(`recorded_at <= $${i++}`);
    params.push(q.until);
  }

  return { clauses, params };
}

export interface CreatePostgresAuditStoreDeps {
  readonly reader: PostgresReader;
}

export function createPostgresAuditStore(
  deps: CreatePostgresAuditStoreDeps,
): AuditStore {
  return {
    async query(q: AuditQuery): Promise<AuditQueryResult> {
      const { clauses, params } = buildWhereClauses(q);
      const allParams: unknown[] = [...params];
      let i = allParams.length + 1;

      // Keyset pagination — strict-less-than on (recorded_at, intent_hash).
      // Tuple comparison is lexicographic in Postgres, which is exactly
      // what we want: walks the index in DESC order without OFFSET.
      //
      // A non-empty cursor that fails to decode is a hard error (not a silent
      // restart from page 1): returning the first page plus a fresh nextCursor
      // would loop any client retrying with the bad cursor.
      const rawCursor = q.cursor ? decodeCursor(q.cursor) : null;
      if (q.cursor && rawCursor === null) {
        throw new InvalidCursorError();
      }
      const cursor = rawCursor;
      let cursorClause = "";
      if (cursor) {
        cursorClause = `(recorded_at, intent_hash) < ($${i++}, $${i++})`;
        allParams.push(cursor.at, cursor.hash);
      }

      const allClauses = [...clauses, ...(cursorClause ? [cursorClause] : [])];
      const whereClause =
        allClauses.length > 0 ? `WHERE ${allClauses.join(" AND ")}` : "";

      // +1 to detect "is there a next page" without a separate COUNT.
      allParams.push(q.limit + 1);
      const limitParam = i;

      const sql = `
        SELECT ${SELECT_COLUMNS}
        FROM intent_audit
        ${whereClause}
        ORDER BY recorded_at DESC, intent_hash DESC
        LIMIT $${limitParam}
      `.replace(/\s+/g, " ").trim();

      const rawRows = await deps.reader.query<IntentAuditRow>(sql, allParams);

      // Normalize recorded_at to string in case the pg driver returned Date.
      const rows = rawRows.map((row) => ({
        ...row,
        recorded_at: normalizeTimestamptz(row.recorded_at, "intent_audit.recorded_at"),
      }));

      const hasMore = rows.length > q.limit;
      const slice = hasMore ? rows.slice(0, q.limit) : rows;
      const records = slice.map(rowToRecord);

      // 092 — VERIFY-ON-READ. Re-derive each cold-store row's tamper-evident
      // auditHash (and verify the hash-bind signature leg) so a row whose bytes
      // were modified after build, or whose signature is forged, is FLAGGED
      // rather than rendered as authoritative. `verifyAuditRecord` is pure / no
      // I/O (audit.ts), so the cost is bounded per row and the read path stays
      // an O(rows) walk. Verdicts are aligned BY INDEX with `records` (§C: the
      // read only ADDS friction — it never drops or rewrites a row). Asymmetric
      // (ed25519) signatures are opaque to this browser-safe verifier; they stay
      // verified:true on the hash axis until a node-side verifier is wired.
      const verifications: AuditRecordVerification[] = records.map((r) =>
        verifyAuditRecord(r),
      );

      // nextCursor encodes the LAST row in the slice (not the n+1-th
      // sentinel). Operators paginating forward see continuous coverage.
      const nextCursor =
        hasMore && slice.length > 0
          ? encodeCursor({
              at: slice[slice.length - 1]!.recorded_at,
              hash: slice[slice.length - 1]!.intent_hash,
            })
          : undefined;

      return {
        records,
        verifications,
        ...(nextCursor !== undefined ? { nextCursor } : {}),
      };
    },

    async getByIntentHash(
      intentHash: string,
      // 112-T3 — the `AuditStore` contract's host-enforced tenant-isolation
      // injection point. This reference cold-store is SINGLE-TENANT (one
      // `intent_audit` table, no tenant column), so it IGNORES `tenantScope` —
      // accepting the argument keeps the signature contract-compatible so the
      // SDK's `audit.byHash` seam (which now threads `input.tenantScope`) does
      // not silently drop it. A genuinely multi-tenant adopter MUST override
      // this method to add a `WHERE tenant = $2` predicate; ignoring the scope
      // here is safe only because this store holds one tenant's records.
      _tenantScope?: string,
    ): Promise<AuditRecord | null> {
      // ORDER BY recorded_at DESC LIMIT 1 because intent_hash is the
      // partition-aware deduplication key but a hash CAN appear in
      // multiple rows under degenerate replay (two writers race on the
      // same intent). Returning the most recent one is the safe choice.
      const sql = `
        SELECT ${SELECT_COLUMNS}
        FROM intent_audit
        WHERE intent_hash = $1
        ORDER BY recorded_at DESC
        LIMIT 1
      `.replace(/\s+/g, " ").trim();

      const rawRows = await deps.reader.query<IntentAuditRow>(sql, [intentHash]);
      if (rawRows.length === 0) return null;
      const row = {
        ...rawRows[0]!,
        recorded_at: normalizeTimestamptz(rawRows[0]!.recorded_at, "intent_audit.recorded_at"),
      };
      const record = rowToRecord(row);
      // 092 — VERIFY-ON-READ for the single-record path. The `AuditStore`
      // contract returns a bare `AuditRecord`, so the verdict rides as a
      // non-enumerable `verification` slot read by the standalone helper
      // `getVerifiedByIntentHash` below; the wire `byHash` output schema strips
      // it. Single-record consumers (replay via `replayWithIntegrity`, the
      // approval-chain join) independently re-verify, so verification on this
      // path is defense-in-depth rather than the surfacing surface (that is the
      // list `query`'s `verifications`). A hard-tampered/forged row is still
      // RETURNED (forensics need the bytes) — never silently dropped, never
      // rendered as authoritative without its verdict.
      return attachVerification(record, verifyAuditRecord(record));
    },
  };
}

/**
 * Symbol slot carrying the verify-on-read verdict alongside a record returned by
 * `getByIntentHash` (092). A Symbol key keeps the `AuditRecord` structurally
 * unchanged for every existing consumer (and is stripped by JSON/zod at the wire
 * boundary), while `getVerifiedByIntentHash` can read the verdict back.
 */
const VERIFICATION_SLOT = Symbol("adjudicate.audit.verification");

function attachVerification(
  record: AuditRecord,
  verification: AuditRecordVerification,
): AuditRecord {
  // Non-enumerable so it never widens the record's JSON / canonical shape — the
  // record hashes and serializes byte-identically to one with no slot.
  Object.defineProperty(record, VERIFICATION_SLOT, {
    value: verification,
    enumerable: false,
    writable: false,
    configurable: false,
  });
  return record;
}

/**
 * Read back the verify-on-read verdict attached by `createPostgresAuditStore`'s
 * `getByIntentHash` (092). Returns `undefined` for a record from a store that
 * does not verify on read (the in-memory reference) or one that crossed a wire
 * boundary (the Symbol slot is non-serializable). Re-verifying directly via
 * `verifyAuditRecord(record)` is always available as the canonical fallback.
 */
export function readVerificationSlot(
  record: AuditRecord,
): AuditRecordVerification | undefined {
  const v = (record as unknown as Record<symbol, unknown>)[VERIFICATION_SLOT];
  return v as AuditRecordVerification | undefined;
}