/**
 * Eventual Consistency Testing Infrastructure
 *
 * Verifies that optimistic writes (done by actions) produce the same
 * database state as the ingester would produce when processing PDS records.
 *
 * How it works:
 * 1. Actions do optimistic writes to local DB
 * 2. Actions schedule PDS writes via after() - we capture these
 * 3. After each test, we:
 *    a) Snapshot the optimistic state
 *    b) Run captured after() callbacks (PDS operations execute)
 *    c) Extract what was written to PDS
 *    d) Reset DB to pre-action state
 *    e) Replay PDS operations through the ingester
 *    f) Compare: optimistic state should equal ingester state
 */

import { vi, expect } from "vitest";
import { emit, emitDelete } from "./test-events";
import { getTestDb, walks, completions, trails } from "./test-db";
import { eq } from "drizzle-orm";

// ============================================================================
// Configuration: Tables to track for eventual consistency
// ============================================================================

// All tables that actions can modify - add new tables here
const TRACKED_TABLES = {
  walks: { table: walks, collection: "app.sidetrail.walk" },
  completions: { table: completions, collection: "app.sidetrail.completion" },
  trails: { table: trails, collection: "app.sidetrail.trail" },
} as const;

type TableName = keyof typeof TRACKED_TABLES;

// ============================================================================
// Types
// ============================================================================

type PdsOperation =
  | {
      type: "create";
      collection: string;
      did: string;
      rkey: string;
      record: Record<string, unknown>;
    }
  | {
      type: "update";
      collection: string;
      did: string;
      rkey: string;
      record: Record<string, unknown>;
    }
  | { type: "delete"; collection: string; did: string; rkey: string };

interface CapturedState {
  pdsOps: PdsOperation[];
  afterCallbacks: Array<() => Promise<void>>;
}

// Rows from db.select() - loose type since we just restore them as-is
type Row = Record<string, unknown>;
type Snapshot = {
  walks: Row[];
  completions: Row[];
  trails: Row[];
};

// ============================================================================
// Global State (reset per test)
// ============================================================================

const state: CapturedState = {
  pdsOps: [],
  afterCallbacks: [],
};

const emptySnapshot = (): Snapshot => ({ walks: [], completions: [], trails: [] });

let initialSnapshot: Snapshot = emptySnapshot();

// Track whether we're inside an after() callback
let insideAfterCallback = false;
// Track blocking PDS operations (called outside after())
let blockingPdsOps: PdsOperation[] = [];

// ============================================================================
// Mock Factories
// ============================================================================

/**
 * Creates an after() mock that captures callbacks for later execution
 */
export function createAfterMock() {
  return vi.fn((fn: () => Promise<void>) => {
    state.afterCallbacks.push(fn);
  });
}

/**
 * Creates a mock lex client that captures PDS operations
 */
export function createCapturingLexClient(currentDid: () => string) {
  let cidCounter = 0;

  const generateCid = () => `bafycapture${++cidCounter}`;

  return {
    get assertDid() {
      return currentDid();
    },

    create: vi.fn(
      async (
        schema: { id?: string; $nsid?: string; nsid?: string; $type?: string },
        record: Record<string, unknown>,
        opts?: { rkey?: string },
      ): Promise<{ uri: string; cid: string }> => {
        const did = currentDid();
        const rkey = opts?.rkey || `gen_${Date.now()}_${Math.random().toString(36).slice(2)}`;
        const collection = schema.$nsid || schema.nsid || schema.id || schema.$type || "";
        const uri = `at://${did}/${collection}/${rkey}`;
        const cid = generateCid();

        const op: PdsOperation = { type: "create", collection, did, rkey, record };
        if (insideAfterCallback) {
          state.pdsOps.push(op);
        } else {
          blockingPdsOps.push(op);
        }
        return { uri, cid };
      },
    ),

    put: vi.fn(
      async (
        schema: { id?: string; $nsid?: string; nsid?: string; $type?: string },
        record: Record<string, unknown>,
        opts: { rkey: string },
      ): Promise<{ uri: string; cid: string }> => {
        const did = currentDid();
        const collection = schema.$nsid || schema.nsid || schema.id || schema.$type || "";
        const uri = `at://${did}/${collection}/${opts.rkey}`;
        const cid = generateCid();

        const op: PdsOperation = { type: "update", collection, did, rkey: opts.rkey, record };
        if (insideAfterCallback) {
          state.pdsOps.push(op);
        } else {
          blockingPdsOps.push(op);
        }
        return { uri, cid };
      },
    ),

    delete: vi.fn(
      async (
        schema: { id?: string; $nsid?: string; nsid?: string; $type?: string },
        opts: { rkey: string },
      ): Promise<void> => {
        const did = currentDid();
        const collection = schema.$nsid || schema.nsid || schema.id || schema.$type || "";

        const op: PdsOperation = { type: "delete", collection, did, rkey: opts.rkey };
        if (insideAfterCallback) {
          state.pdsOps.push(op);
        } else {
          blockingPdsOps.push(op);
        }
      },
    ),

    get: vi.fn(
      async (
        schema: { id?: string; $nsid?: string; nsid?: string; $type?: string },
        opts: { rkey: string },
      ) => {
        const did = currentDid();
        const collection = schema.$nsid || schema.nsid || schema.id || schema.$type || "";
        const uri = `at://${did}/${collection}/${opts.rkey}`;
        const db = getTestDb();

        if (collection === "app.sidetrail.walk") {
          const [row] = await db.select().from(walks).where(eq(walks.uri, uri)).limit(1);
          if (!row) throw new Error(`Walk not found: ${uri}`);
          return { uri, cid: row.cid, value: row.record };
        }

        throw new Error(`Unsupported collection for get: ${collection}`);
      },
    ),

    list: vi.fn(async (schema: { id?: string; $nsid?: string; nsid?: string; $type?: string }) => {
      const did = currentDid();
      const collection = schema.$nsid || schema.nsid || schema.id || schema.$type || "";
      const db = getTestDb();

      if (collection === "app.sidetrail.walk") {
        const rows = await db.select().from(walks).where(eq(walks.authorDid, did));
        return { records: rows.map((r) => ({ uri: r.uri, cid: r.cid, value: r.record })) };
      }

      if (collection === "app.sidetrail.completion") {
        const rows = await db.select().from(completions).where(eq(completions.authorDid, did));
        return { records: rows.map((r) => ({ uri: r.uri, cid: r.cid, value: r.record })) };
      }

      throw new Error(`Unsupported collection for list: ${collection}`);
    }),
  };
}

// ============================================================================
// Generic Snapshot & Restore
// ============================================================================

/**
 * Snapshot all tracked tables
 */
async function snapshotAll(): Promise<Snapshot> {
  const db = getTestDb();
  const snapshot = {} as Snapshot;

  for (const [name, { table }] of Object.entries(TRACKED_TABLES)) {
    snapshot[name as TableName] = (await db.select().from(table)) as Row[];
  }

  return snapshot;
}

/**
 * Restore all tracked tables from snapshot (delete all, re-insert)
 */
async function restoreSnapshot(snapshot: Snapshot): Promise<void> {
  const db = getTestDb();

  for (const [name, { table }] of Object.entries(TRACKED_TABLES)) {
    await db.delete(table);
    const rows = snapshot[name as TableName];
    if (rows.length > 0) {
      // rows came from db.select() so they're valid for insert back into the same table
      type InsertFn = (rows: Row[]) => Promise<unknown>;
      await (db.insert(table).values as unknown as InsertFn)(rows);
    }
  }
}

/**
 * Compute diff between two snapshots
 */
function snapshotDiff(before: Snapshot, after: Snapshot): string[] {
  const changes: string[] = [];

  for (const name of Object.keys(TRACKED_TABLES) as TableName[]) {
    const beforeUris = new Set(before[name].map((r) => r.uri));
    const afterUris = new Set(after[name].map((r) => r.uri));

    // Added
    for (const row of after[name]) {
      if (!beforeUris.has(row.uri)) changes.push(`+${name}:${row.uri}`);
    }
    // Deleted
    for (const row of before[name]) {
      if (!afterUris.has(row.uri)) changes.push(`-${name}:${row.uri}`);
    }
    // Modified
    for (const row of after[name]) {
      if (beforeUris.has(row.uri)) {
        const beforeRow = before[name].find((r) => r.uri === row.uri);
        if (JSON.stringify(beforeRow?.record) !== JSON.stringify(row.record)) {
          changes.push(`~${name}:${row.uri}`);
        }
      }
    }
  }

  return changes;
}

/**
 * Take initial snapshot before action runs
 */
export async function captureInitialState(): Promise<void> {
  initialSnapshot = await snapshotAll();
}

// ============================================================================
// Verification
// ============================================================================

/** Strip $type from records for comparison (redundant with collection) */
function normalizeRecord(record: unknown): unknown {
  if (!record || typeof record !== "object") return record;
  const { $type, ...rest } = record as Record<string, unknown>;
  return rest;
}

/**
 * Verify eventual consistency after a test.
 * Call this in afterEach() for action tests.
 */
export async function verifyEventualConsistency(): Promise<void> {
  // Fail fast if any blocking PDS operations were detected
  if (blockingPdsOps.length > 0) {
    const ops = blockingPdsOps.map((op) => `${op.type}:${op.collection}/${op.rkey}`);
    throw new Error(
      `Blocking PDS operations detected! These should be wrapped in after(). ` +
        `Operations: ${ops.join(", ")}`,
    );
  }

  const currentState = await snapshotAll();
  const changes = snapshotDiff(initialSnapshot, currentState);
  const hasChanges = changes.length > 0;

  // If no after callbacks were captured but DB changed, that's a bug
  if (state.afterCallbacks.length === 0) {
    if (hasChanges) {
      throw new Error(
        `Action changed DB but no after() callback was scheduled. ` +
          `This means PDS writes are missing! Changes: ${changes.join(", ")}`,
      );
    }
    reset();
    return;
  }

  // 1. Run captured after() callbacks to execute PDS operations
  insideAfterCallback = true;
  for (const callback of state.afterCallbacks) {
    try {
      await callback();
    } catch (err) {
      console.error("[eventual-consistency] after() callback failed:", err);
    }
  }
  insideAfterCallback = false;

  // If no PDS ops after running callbacks, check if DB changed
  if (state.pdsOps.length === 0) {
    const postCallbackState = await snapshotAll();
    const postChanges = snapshotDiff(initialSnapshot, postCallbackState);

    if (postChanges.length > 0) {
      throw new Error(
        `Action changed DB but no PDS operations were captured. ` +
          `This means PDS writes are missing! Changes: ${postChanges.join(", ")}`,
      );
    }

    reset();
    return;
  }

  // 2. Snapshot the optimistic state (what actions wrote)
  const optimisticState = await snapshotAll();

  // 3. Restore DB to initial state (before action)
  await restoreSnapshot(initialSnapshot);

  // 4. Replay PDS operations through the ingester
  for (const op of state.pdsOps) {
    const collection = op.collection as "app.sidetrail.walk" | "app.sidetrail.completion";

    if (op.type === "delete") {
      const uri = `at://${op.did}/${op.collection}/${op.rkey}`;
      await emitDelete(collection, uri);
    } else {
      await emit(collection, op.did, op.record, op.rkey);
    }
  }

  // 5. Snapshot the ingester state
  const ingesterState = await snapshotAll();

  // 6. Compare - optimistic should match ingester (generic over all tables)
  for (const name of Object.keys(TRACKED_TABLES) as TableName[]) {
    const optRows = optimisticState[name];
    const ingRows = ingesterState[name];

    expect(optRows.length).toBe(ingRows.length);

    for (const optRow of optRows) {
      const ingRow = ingRows.find((r) => r.uri === optRow.uri);
      if (!ingRow) {
        throw new Error(
          `Optimistic ${name} ${optRow.uri} not found in ingester state. ` +
            `This likely means the PDS write is missing.`,
        );
      }
      expect(normalizeRecord(optRow.record)).toEqual(normalizeRecord(ingRow.record));
    }
  }

  // 7. Clean up
  reset();
}

/**
 * Reset captured state between tests
 */
export function reset(): void {
  state.pdsOps.length = 0;
  state.afterCallbacks.length = 0;
  initialSnapshot = emptySnapshot();
  blockingPdsOps = [];
  insideAfterCallback = false;
}

/**
 * Get captured PDS operations (for debugging)
 */
export function getCapturedPdsOps(): readonly PdsOperation[] {
  return state.pdsOps;
}
