import type {
  ColineActorRef,
  ColineCollection,
  ColineRecordBatchOperation,
  ColineRecordEnvelope,
  ColineRecordQuery,
} from "../capabilities/types";
import {
  canAccessRecord,
  canVisitorAccess,
  DEFAULT_RECORD_ACCESS,
  sameActor,
} from "../capabilities/record-access";
import {
  aggregateRecords,
  compareRecords,
  matchesRecordWhere,
  normalizeRecordSort,
} from "../capabilities/record-query";
import type { CollectionAccess } from "../manifest/collections";
import type { TestWorkspaceState } from "./fake-capabilities";

type Row = ColineRecordEnvelope<unknown> & { data: Record<string, unknown> };

const MAX_PAGE = 1000;
const MAX_BATCH = 500;
const MAX_BULK = 5000;

const TEAM_ONLY: CollectionAccess = { read: "team", create: "team", update: "team", delete: "team" };

export interface FakeCollectionOptions {
  destructiveAllowed?: boolean;
  /** Records created here belong to this actor (a site tool's visitor). */
  createdBy?: ColineActorRef;
  /** The collection's declared visitor rules; used when `actor` is a visitor. */
  rules?: CollectionAccess;
  /** The collection's zod schema; visitor writes must match it, as on Coline. */
  schema?: { safeParse(data: unknown): { success: boolean; error?: { message: string } } };
}

/**
 * The in-memory collection, with the same access semantics as the server:
 * members and app logic use the per-record ACL, visitors use the
 * collection's declared rules and only the operations a site page may call.
 */
export function createFakeCollection<T>(
  state: TestWorkspaceState,
  name: string,
  actor: ColineActorRef,
  installId: string,
  options: FakeCollectionOptions = {},
): ColineCollection<T> {
  if (!state.records.has(name)) state.records.set(name, new Map());
  const records = state.records.get(name)! as Map<string, Row>;
  const visitor = actor.kind === "visitor" ? actor : null;
  const rules = options.rules ?? TEAM_ONLY;
  const creator = options.createdBy ?? actor;
  const fail = (code: string, message: string): never => {
    throw Object.assign(new Error(message), { code });
  };
  const visitorOnly = (operation: string): never =>
    fail("unavailable_to_visitors", `Site visitors cannot call "${operation}".`);
  const assertShape = (data: unknown) => {
    if (!visitor || !options.schema) return;
    const result = options.schema.safeParse(data);
    if (!result.success) fail("record_invalid", `This record doesn't match the collection's schema. ${result.error?.message ?? ""}`);
  };
  const visible = (row: Row) =>
    visitor
      ? canVisitorAccess(rules.read, visitor, rules.read === "owner" ? row.createdBy : null) &&
        (rules.read !== "owner" || sameActor(row.createdBy, visitor))
      : canAccessRecord(row.access, row.createdBy, actor, "read");
  const wrap = (row: Row) => structuredClone(row) as unknown as ColineRecordEnvelope<T>;
  const changed = () => {
    state.recordRevisions.set(name, (state.recordRevisions.get(name) ?? 0) + 1);
    for (const listener of state.realtimeListeners.get("coline.records.changed") ?? [])
      listener({ event: "coline.records.changed", data: null });
  };
  function writable(id: string, expectedVersion?: number, action: "update" | "delete" | "manage" = "update") {
    const row = records.get(id);
    if (!row || !visible(row)) return fail("record_not_found", "Record not found.");
    const permitted = visitor
      ? action !== "manage" && canVisitorAccess(rules[action], visitor, row.createdBy)
      : canAccessRecord(row.access, row.createdBy, actor, action === "manage" ? "manage" : "write");
    if (!permitted) return fail("record_forbidden", "You cannot change this record.");
    if (expectedVersion !== undefined && row.version !== expectedVersion)
      return fail("record_conflict", "This record changed. Refresh before saving again.");
    return row;
  }
  function matching(where: unknown) {
    return [...records.values()].filter((row) => visible(row) && matchesRecordWhere(row.data, where));
  }
  function query(input?: ColineRecordQuery<T>) {
    const rows = matching(input?.where);
    const sorts = normalizeRecordSort(input);
    if (sorts.length > 3) return fail("invalid_query", "Sort by at most three fields.");
    rows.sort((a, b) => compareRecords(a, b, sorts));
    const offset = Number(input?.cursor ?? 0),
      limit = input?.limit ?? 50;
    if (!Number.isSafeInteger(offset) || offset < 0)
      return fail("invalid_cursor", "Invalid collection cursor.");
    if (!Number.isInteger(limit) || limit < 1 || limit > MAX_PAGE)
      return fail("invalid_query", `Record limit must be 1–${MAX_PAGE}.`);
    return {
      records: rows.slice(offset, offset + limit).map(wrap),
      nextCursor: rows.length > offset + limit ? String(offset + limit) : null,
    };
  }
  function insertRow(data: unknown, insertOptions?: { id?: string | undefined; access?: Row["access"] | undefined }) {
    if (visitor) {
      if (!canVisitorAccess(rules.create, visitor, null))
        return fail("record_forbidden", "Visitors cannot add records here.");
      if (insertOptions?.id || insertOptions?.access)
        return fail("unavailable_to_visitors", "Visitors cannot choose record ids or access.");
      assertShape(data);
    }
    const id = insertOptions?.id ?? state.nextRecordId();
    const existing = records.get(id);
    if (existing) {
      if (
        !sameActor(existing.createdBy, creator) ||
        JSON.stringify(existing.data) !== JSON.stringify(data) ||
        JSON.stringify(existing.access) !==
          JSON.stringify(insertOptions?.access ?? DEFAULT_RECORD_ACCESS)
      )
        return fail("record_identity_conflict", "Record identity already in use.");
      return existing;
    }
    const now = new Date().toISOString();
    const row: Row = {
      id,
      appId: "app_test",
      installId,
      collection: name,
      createdBy: structuredClone(creator),
      createdAt: now,
      updatedAt: now,
      version: 1,
      access: structuredClone(insertOptions?.access ?? DEFAULT_RECORD_ACCESS),
      data: structuredClone(data) as Record<string, unknown>,
    };
    records.set(id, row);
    return row;
  }
  function patchRow(row: Row, patch: unknown) {
    row.data = { ...row.data, ...(structuredClone(patch) as Record<string, unknown>) };
    row.version++;
    row.updatedAt = new Date().toISOString();
  }
  function atomically<R>(work: () => R): R {
    const snapshot = new Map([...records].map(([id, row]) => [id, structuredClone(row)]));
    try {
      const result = work();
      changed();
      return result;
    } catch (error) {
      records.clear();
      for (const [id, row] of snapshot) records.set(id, row);
      throw error;
    }
  }
  function applyBatch(operations: Array<ColineRecordBatchOperation<T>>) {
    if (operations.length > MAX_BATCH) return fail("batch_too_large", `Batches are limited to ${MAX_BATCH} operations.`);
    if (operations.some((operation) => operation.op === "delete") && options.destructiveAllowed === false)
      return fail("effect_ceiling_exceeded", "Batches with deletes need a destructive effect.");
    return atomically(() =>
      operations.map((operation) => {
        if (operation.op === "insert") return wrap(insertRow(operation.data, operation));
        if (operation.op === "update") {
          const row = writable(operation.id, operation.expectedVersion);
          patchRow(row, operation.data);
          return wrap(row);
        }
        writable(operation.id, operation.expectedVersion, "delete");
        records.delete(operation.id);
        return null;
      }),
    );
  }
  return {
    insert: async (data, insertOptions) => {
      const row = insertRow(data, insertOptions);
      changed();
      return wrap(row);
    },
    get: async (id) => {
      const row = records.get(id);
      return row && visible(row) ? wrap(row) : null;
    },
    update: async (id, patch, writeOptions) => {
      const row = writable(id, writeOptions?.expectedVersion);
      assertShape({ ...(row.data as Record<string, unknown>), ...(patch as Record<string, unknown>) });
      patchRow(row, patch);
      changed();
      return wrap(row);
    },
    delete: async (id, writeOptions) => {
      writable(id, writeOptions?.expectedVersion, "delete");
      records.delete(id);
      changed();
    },
    setAccess: async (id, access, writeOptions) => {
      if (visitor) return visitorOnly("storage.records.setAccess");
      const row = writable(id, writeOptions?.expectedVersion, "manage");
      row.access = structuredClone(access);
      row.version++;
      row.updatedAt = new Date().toISOString();
      changed();
      return wrap(row);
    },
    query: async (input) => query(input),
    sync: async (input, revision) => {
      const current = state.recordRevisions.get(name) ?? 0;
      return {
        revision: current,
        ...(revision === current ? { records: null, nextCursor: null } : query(input)),
      };
    },
    count: async (where) => ({ count: matching(where).length }),
    aggregate: async (input) => {
      if (visitor) return visitorOnly("storage.records.aggregate");
      if ((input.groupBy?.length ?? 0) > 3) return fail("invalid_query", "Group by at most three fields.");
      if (Object.keys(input.metrics).length === 0) return fail("invalid_query", "Name at least one metric.");
      return aggregateRecords(matching(input.where) as Array<{ data: T }>, input);
    },
    batch: async (operations) =>
      visitor ? visitorOnly("storage.records.batch") : { results: applyBatch(operations) },
    insertMany: async (rows) =>
      visitor
        ? visitorOnly("storage.records.insertMany")
        : {
            records: applyBatch(rows.map((row) => ({ op: "insert" as const, ...row }))) as Array<
              ColineRecordEnvelope<T>
            >,
          },
    updateMany: async (input) =>
      visitor
        ? visitorOnly("storage.records.updateMany")
        : atomically(() => {
            const rows = matching(input.where).filter((row) =>
              canAccessRecord(row.access, row.createdBy, actor, "write"),
            );
            if (rows.length > MAX_BULK) return fail("bulk_too_large", `Bulk updates are limited to ${MAX_BULK} records.`);
            for (const row of rows) patchRow(row, input.data);
            return { count: rows.length };
          }),
    deleteMany: async (input) =>
      visitor
        ? visitorOnly("storage.records.deleteMany")
        : atomically(() => {
            const rows = matching(input.where).filter((row) =>
              canAccessRecord(row.access, row.createdBy, actor, "write"),
            );
            if (rows.length > MAX_BULK) return fail("bulk_too_large", `Bulk deletes are limited to ${MAX_BULK} records.`);
            for (const row of rows) records.delete(row.id);
            return { count: rows.length };
          }),
  };
}