import assert from "node:assert/strict";
import { createApp } from "../../src/core/app";
import type { Db, Stmt } from "../../src/core/db";
import { clientVia, provisionWith } from "./harness";

/**
 * Concurrent-writer durability, stated as invariants before it is run.
 *
 * I1 — exactly one claim per canonical hash. N writers submitting byte-identical
 *      content converge on one record id; a duplicate is permanent damage, and a
 *      read-then-write check alone cannot prevent one.
 * I2 — zero lost observations. Every submission either creates its record or
 *      replays the record that won, and every distinct payload survives.
 * I3 — no partial write survives a failed batch. A write whose batch fails leaves
 *      no observation, no FTS projection, no history, no event, no outbox work,
 *      and no idempotency receipt behind.
 *
 * The same assertions run on bun:sqlite and on D1, because the claim is about the
 * shared core's write path, which expresses every atomic write as one batch
 * (`src/core/db.ts`). Dedup lives inside that batch as a unique index, so the
 * read-then-write window in `appendObservation` and `consolidate` is a fast path,
 * not the safety mechanism — that is what these invariants exist to prove.
 */

const ORIGIN = "http://durability.test";

/** Writers per payload. Six keeps a D1 round trip honest without a long lane. */
export const WRITERS = 6;
/** Distinct payloads in the same burst, so overlap is not the only case. */
export const PAYLOADS = 4;

export interface DurabilityReport {
  runtime: string;
  writers: number;
  payloads: number;
  /**
   * Batches rejected by a canonical unique index during this run.
   *
   * Instrumentation, not an assertion: it records whether the read-then-write
   * window actually opened on this runtime. Zero means the burst serialised and
   * only the fast path was exercised — a weaker result that must be reported as
   * such rather than read as "no race exists".
   */
  canonical_collisions: number;
  /** Concurrent observation submissions issued in one burst. */
  observation_requests: number;
  observation_created: number;
  observation_replayed: number;
  observation_failed: number;
  observation_rows: number;
  observation_canonical_hashes: number;
  claim_requests: number;
  claim_created: number;
  claim_replayed: number;
  claim_failed: number;
  claim_rows: number;
  /** Rows left behind by the deliberately failed batch. Must be zero. */
  partial_rows: number;
}

type Client = Pick<ReturnType<typeof clientVia>, "call">;

const clientFor = (db: Db, runtime: string): Client =>
  clientVia(createApp({ db, runtime, revision: "durability" }), ORIGIN);

/** Counts the batches a canonical unique index rejected, without changing behaviour. */
export function countingCanonicalCollisions(db: Db): { db: Db; collisions: () => number } {
  let collisions = 0;
  return {
    collisions: () => collisions,
    db: {
      all: <Row>(sql: string, params: Parameters<Db["all"]>[1] = []) => db.all<Row>(sql, params),
      exec: (sql: string) => db.exec(sql),
      async batch(statements: Stmt[]) {
        try {
          await db.batch(statements);
        } catch (error) {
          if (error instanceof Error && /UNIQUE.*canonical_hash/iu.test(error.message)) collisions += 1;
          throw error;
        }
      },
    },
  };
}

const observationBody = (runtime: string, index: number) => ({
  subject_id: "subject_durability",
  kind: "tool_result",
  content: `Durability probe ${index} on ${runtime}.`,
  // `source.id` is what makes the write canonically identifiable; without it the
  // core assigns no canonical hash and every submission is a distinct record.
  source: { type: "tool", ref: "durability", id: `durability-${index}` },
  trust: "verified",
});

function assertAccepted(label: string, responses: { status: number; body: any }[]) {
  const rejected = responses.filter((response) => response.status !== 200 && response.status !== 201);
  assert.equal(
    rejected.length,
    0,
    `${label}: ${rejected.length} concurrent writes were rejected: ${
      JSON.stringify(rejected.slice(0, 3).map((response) => [response.status, response.body?.error]))
    }`,
  );
}

async function counters(db: Db) {
  const [row] = await db.all<Record<string, number>>(
    `SELECT (SELECT COUNT(*) FROM observations) AS observations,
            (SELECT COUNT(*) FROM observations_fts) AS observations_fts,
            (SELECT COUNT(*) FROM record_history) AS record_history,
            (SELECT COUNT(*) FROM events) AS events,
            (SELECT COUNT(*) FROM index_outbox) AS index_outbox,
            (SELECT COUNT(*) FROM idempotency_v3) AS idempotency_v3`,
  );
  return Object.fromEntries(Object.entries(row!).map(([key, value]) => [key, Number(value)]));
}

export async function assertConcurrentWriterDurability(
  db: Db,
  runtime: string,
): Promise<DurabilityReport> {
  const principal = await provisionWith(db, { scopes: ["*"] });
  const scope = [principal.orgId, principal.principalId];
  const counted = countingCanonicalCollisions(db);
  // Each writer is its own app instance over the same store, as N agents are.
  const writers = Array.from({ length: WRITERS }, () => clientFor(counted.db, runtime));

  // --- I1 + I2, observations -------------------------------------------------
  const burst = writers.flatMap((client) =>
    Array.from({ length: PAYLOADS }, (_, index) => ({
      index,
      sent: client.call("POST", "/v1/observations", {
        key: principal.key,
        body: observationBody(runtime, index),
      }),
    })));
  const observed = await Promise.all(burst.map(async ({ index, sent }) => ({ index, res: await sent })));
  assertAccepted("observations", observed.map(({ res }) => res));

  for (let index = 0; index < PAYLOADS; index += 1) {
    const group = observed.filter((entry) => entry.index === index);
    const ids = new Set(group.map(({ res }) => res.body.data.observation_id as string));
    assert.equal(
      ids.size,
      1,
      `I1: payload ${index} minted ${ids.size} identities under ${WRITERS} concurrent writers`,
    );
    assert.equal(
      group.filter(({ res }) => res.status === 201).length,
      1,
      `I1: payload ${index} reported more than one creation`,
    );
  }

  const [observationRows] = await db.all<{
    rows: number; hashes: number; fts: number; history: number; events: number;
  }>(
    `SELECT (SELECT COUNT(*) FROM observations WHERE org_id = ? AND actor_id = ?) AS rows,
            (SELECT COUNT(DISTINCT canonical_hash) FROM observations
              WHERE org_id = ? AND actor_id = ?) AS hashes,
            (SELECT COUNT(*) FROM observations_fts f
               JOIN observations o ON o.id = f.observation_id
              WHERE o.org_id = ? AND o.actor_id = ?) AS fts,
            (SELECT COUNT(*) FROM record_history
              WHERE org_id = ? AND record_type = 'observation') AS history,
            (SELECT COUNT(*) FROM events
              WHERE org_id = ? AND kind = 'observation.appended') AS events`,
    [...scope, ...scope, ...scope, principal.orgId, principal.orgId],
  );
  assert.deepEqual(
    Object.fromEntries(Object.entries(observationRows!).map(([key, value]) => [key, Number(value)])),
    { rows: PAYLOADS, hashes: PAYLOADS, fts: PAYLOADS, history: PAYLOADS, events: PAYLOADS },
    `I1/I2: ${WRITERS * PAYLOADS} concurrent submissions must leave exactly ${PAYLOADS} observations and one projection each`,
  );

  // --- I1, claims ------------------------------------------------------------
  const evidenceId = observed[0]!.res.body.data.observation_id as string;
  const consolidation = {
    subject_id: "subject_durability",
    claims: [{
      kind: "procedural",
      statement: "Concurrent writers must converge on one claim per canonical hash.",
      confidence: 0.9,
      sources: [{ observation_id: evidenceId, relation: "supports" }],
    }],
  };
  const consolidated = await Promise.all(writers.map((client) =>
    client.call("POST", "/v1/consolidations", { key: principal.key, body: consolidation })));
  assertAccepted("consolidations", consolidated);
  const claimIds = new Set(consolidated.map((res) => res.body.data.claims[0].claim_id as string));
  assert.equal(claimIds.size, 1, `I1: ${WRITERS} concurrent consolidations minted ${claimIds.size} claims`);

  const [claimRows] = await db.all<{ rows: number; sources: number; fts: number; history: number }>(
    `SELECT (SELECT COUNT(*) FROM claims WHERE org_id = ? AND actor_id = ?) AS rows,
            (SELECT COUNT(*) FROM claim_sources s
               JOIN claims c ON c.id = s.claim_id
              WHERE c.org_id = ? AND c.actor_id = ?) AS sources,
            (SELECT COUNT(*) FROM claims_fts f
               JOIN claims c ON c.id = f.claim_id
              WHERE c.org_id = ? AND c.actor_id = ?) AS fts,
            (SELECT COUNT(*) FROM record_history
              WHERE org_id = ? AND record_type = 'claim') AS history`,
    [...scope, ...scope, ...scope, principal.orgId],
  );
  assert.deepEqual(
    Object.fromEntries(Object.entries(claimRows!).map(([key, value]) => [key, Number(value)])),
    { rows: 1, sources: 1, fts: 1, history: 1 },
    `I1: ${WRITERS} concurrent consolidations must leave exactly one claim and one projection each`,
  );

  // --- I3, no partial write survives a failed batch ---------------------------
  const before = await counters(db);
  let poisoned = false;
  const faulty: Db = {
    all: <Row>(sql: string, params: Parameters<Db["all"]>[1] = []) => db.all<Row>(sql, params),
    exec: (sql: string) => db.exec(sql),
    async batch(statements: Stmt[]) {
      if (!poisoned && statements.some((statement) => /INSERT INTO observations\s/u.test(statement.sql))) {
        poisoned = true;
        // Fail the last statement of a real write batch. Every earlier statement
        // in it must roll back with it, on both drivers.
        await db.batch([
          ...statements,
          { sql: `INSERT INTO titen_durability_absent_table (id) VALUES ('x')` },
        ]);
        return;
      }
      await db.batch(statements);
    },
  };
  const failed = await clientFor(faulty, runtime).call("POST", "/v1/observations", {
    key: principal.key,
    body: observationBody(runtime, PAYLOADS),
  });
  assert.equal(poisoned, true, "I3: the fault was never injected into a real write batch");
  assert.ok(
    failed.status >= 500,
    `I3: a failed write batch must not report success, got ${failed.status}`,
  );
  const after = await counters(db);
  assert.deepEqual(after, before, "I3: a failed batch left rows behind");

  const report: DurabilityReport = {
    runtime,
    writers: WRITERS,
    payloads: PAYLOADS,
    canonical_collisions: counted.collisions(),
    observation_requests: observed.length,
    observation_created: observed.filter(({ res }) => res.status === 201).length,
    observation_replayed: observed.filter(({ res }) => res.status === 200).length,
    observation_failed: 0,
    observation_rows: Number(observationRows!.rows),
    observation_canonical_hashes: Number(observationRows!.hashes),
    claim_requests: consolidated.length,
    claim_created: consolidated.filter((res) => res.status === 201).length,
    claim_replayed: consolidated.filter((res) => res.status === 200).length,
    claim_failed: 0,
    claim_rows: Number(claimRows!.rows),
    partial_rows: Object.entries(after)
      .reduce((total, [key, value]) => total + (value - before[key]!), 0),
  };
  process.stderr.write(`[durability] ${JSON.stringify(report)}\n`);
  return report;
}