import type {
  ColineCapabilities,
  ColineCollection,
  ColineRecordEnvelope,
  ColineRecordQuery,
} from "./types";
import { DEFAULT_RECORD_ACCESS, type ColineRecordAccess } from "./record-access";
import { compareRecords, matchesRecordWhere, normalizeRecordSort } from "./record-query";

export interface LiveCollectionSnapshot<T> {
  records: readonly ColineRecordEnvelope<T>[];
  nextCursor: string | null;
  status: "loading" | "ready" | "syncing" | "error";
  pending: number;
  error: Error | null;
}
export interface LiveCollectionOptions {
  /** Fallback for dropped events and reconnects; minimum 250 ms, default 2000 ms. */
  pollIntervalMs?: number;
  /** Identity for local optimistic envelopes; the server always supplies authoritative identity. */
  actorUserId?: string;
  installId?: string;
}
interface Pending<T> {
  id: string;
  overlay: (rows: Map<string, ColineRecordEnvelope<T>>) => void;
  execute: () => Promise<ColineRecordEnvelope<T> | null>;
  resolve: (value: ColineRecordEnvelope<T> | null) => void;
  reject: (error: unknown) => void;
}
const authorizationCodes = new Set([
  "permission_not_granted",
  "permission_not_declared",
  "app_install_not_found",
  "install_not_active",
  "entitlement_required",
  "app_invocation_token_invalid",
  "app_invocation_token_expired",
  "app_invocation_install_mismatch",
  "app_operation_denied",
  "workspace_access_denied",
  "actor_not_member",
  "forbidden",
  "unauthorized",
  "record_forbidden",
]);
const asError = (cause: unknown) => (cause instanceof Error ? cause : new Error(String(cause)));

/** A bounded live query: immediate optimistic changes, CAS writes, authoritative replacement on sync. */
export function createLiveCollection<T extends Record<string, unknown>>(
  coline: Pick<ColineCapabilities, "storage" | "realtime">,
  name: string,
  query: ColineRecordQuery<T> = {},
  options: LiveCollectionOptions = {},
) {
  const collection: ColineCollection<T> = coline.storage.collection<T>(name);
  const input = structuredClone(query);
  const listeners = new Set<() => void>();
  let base = new Map<string, ColineRecordEnvelope<T>>();
  let pending: Pending<T>[] = [];
  let revision: number | undefined;
  let connectionEpoch = 0,
    accessEpoch = 0,
    snapshotEpoch = 0;
  let generation = 0,
    running = false,
    disposed = false,
    started = false;
  let syncing: Promise<void> | null = null,
    syncAgain = false;
  let timer: ReturnType<typeof setInterval> | null = null;
  let subscription: { unsubscribe: () => void } | null = null;
  let snapshot: LiveCollectionSnapshot<T> = {
    records: [],
    nextCursor: null,
    status: "loading",
    pending: 0,
    error: null,
  };
  function emit(update: Partial<LiveCollectionSnapshot<T>> = {}) {
    if (disposed) return;
    const rows = new Map(base);
    for (const mutation of pending) mutation.overlay(rows);
    const records = [...rows.values()].filter((row) => matchesRecordWhere(row.data, input.where));
    const sorts = normalizeRecordSort(input);
    records.sort((a, b) => compareRecords(a, b, sorts));
    snapshot = {
      ...snapshot,
      ...update,
      records: records.slice(0, input.limit ?? 50),
      pending: pending.length,
    };
    for (const listener of listeners) listener();
  }
  function cancel(error: unknown, id?: string) {
    const removed = pending.filter((item) => !id || item.id === id);
    pending = pending.filter((item) => id && item.id !== id);
    for (const item of removed) item.reject(error);
  }
  async function refresh(): Promise<void> {
    if (disposed) return;
    if (syncing) {
      syncAgain = true;
      return syncing;
    }
    syncing = (async () => {
      do {
        syncAgain = false;
        const before = generation;
        try {
          const result = await collection.sync(input, revision);
          if (disposed) return;
          if (before !== generation) {
            revision = undefined;
            syncAgain = true;
            continue;
          }
          if (result.records) {
            snapshotEpoch++;
            base = new Map(result.records.map((row) => [row.id, row]));
            emit({ nextCursor: result.nextCursor });
          }
          revision = result.revision;
          emit({ status: "ready", error: null });
        } catch (cause) {
          if (disposed) return;
          const code = (cause as { code?: string })?.code;
          // HTTP routes prefix guard denials; the local bridge uses their raw codes.
          const denial =
            typeof code === "string" ? code.replace(/^APP_CAPABILITY_/i, "").toLowerCase() : null;
          if (denial && authorizationCodes.has(denial)) {
            accessEpoch++;
            base.clear();
            cancel(cause);
            revision = undefined;
          }
          emit({ status: "error", error: asError(cause) });
        }
      } while (syncAgain && !disposed);
    })();
    try {
      await syncing;
    } finally {
      syncing = null;
    }
  }
  async function drain() {
    if (running || disposed) return;
    running = true;
    try {
      while (pending.length && !disposed) {
        const item = pending[0]!;
        try {
          const accessAtStart = accessEpoch,
            snapshotAtStart = snapshotEpoch;
          const result = await item.execute();
          if (disposed) return;
          if (accessAtStart !== accessEpoch) continue;
          generation++;
          if (result) {
            // A newer authorized snapshot may have removed this record (ACL or filter).
            if (snapshotAtStart === snapshotEpoch || base.has(item.id)) base.set(item.id, result);
          } else base.delete(item.id);
          pending = pending.filter((entry) => entry !== item);
          item.resolve(result);
          emit({ error: null });
        } catch (cause) {
          generation++;
          cancel(cause, item.id);
          revision = undefined;
          emit({ error: asError(cause) });
          await refresh();
        }
      }
    } finally {
      running = false;
      if (!disposed && started) {
        revision = undefined;
        void refresh();
      }
    }
  }
  function enqueue(id: string, overlay: Pending<T>["overlay"], execute: Pending<T>["execute"]) {
    if (disposed) return Promise.reject(new Error("Live collection is disposed."));
    const promise = new Promise<ColineRecordEnvelope<T> | null>((resolve, reject) =>
      pending.push({ id, overlay, execute, resolve, reject }),
    );
    emit({ error: null }); // Synchronous: rendering does not wait for the network.
    void drain();
    return promise;
  }
  return {
    getSnapshot: () => snapshot,
    subscribe(listener: () => void) {
      listeners.add(listener);
      return () => {
        listeners.delete(listener);
      };
    },
    async start() {
      if (disposed || started) return;
      started = true;
      const epoch = ++connectionEpoch;
      timer = setInterval(() => void refresh(), Math.max(250, options.pollIntervalMs ?? 2000));
      try {
        void coline.realtime
          .subscribe("coline.records.changed", () => {
            void refresh();
          })
          .then((value) => {
            if (disposed || !started || epoch !== connectionEpoch) value.unsubscribe();
            else subscription = value;
          })
          .catch(() => {
            /* Polling remains authoritative when realtime consent is absent. */
          });
      } catch {
        /* Transports may reject client-only subscriptions synchronously; keep polling. */
      }
      await refresh();
    },
    stop() {
      started = false;
      connectionEpoch++;
      if (timer) clearInterval(timer);
      timer = null;
      subscription?.unsubscribe();
      subscription = null;
    },
    refresh,
    insert(data: T, access: ColineRecordAccess = DEFAULT_RECORD_ACCESS) {
      const id = `rec_${crypto.randomUUID()}`;
      const now = new Date().toISOString();
      const row: ColineRecordEnvelope<T> = {
        id,
        appId: "",
        installId: options.installId ?? "",
        collection: name,
        version: 1,
        access: structuredClone(access),
        createdBy: options.actorUserId
          ? { kind: "user", userId: options.actorUserId }
          : { kind: "app" },
        createdAt: now,
        updatedAt: now,
        data: structuredClone(data),
      };
      return enqueue(
        id,
        (rows) => {
          rows.set(id, row);
        },
        () => collection.insert(data, { id, access }),
      );
    },
    update(id: string, patch: Partial<T>) {
      const copy = structuredClone(patch);
      const expectedVersion = snapshot.records.find((row) => row.id === id)?.version;
      if (expectedVersion === undefined)
        return Promise.reject(new Error("Record is not in this view."));
      return enqueue(
        id,
        (rows) => {
          const row = rows.get(id);
          if (row)
            rows.set(id, { ...row, version: row.version + 1, data: { ...row.data, ...copy } });
        },
        () => collection.update(id, copy, { expectedVersion }),
      );
    },
    delete(id: string) {
      const expectedVersion = snapshot.records.find((row) => row.id === id)?.version;
      if (expectedVersion === undefined)
        return Promise.reject(new Error("Record is not in this view."));
      return enqueue(
        id,
        (rows) => {
          rows.delete(id);
        },
        async () => {
          await collection.delete(id, { expectedVersion });
          return null;
        },
      );
    },
    setAccess(id: string, access: ColineRecordAccess) {
      const expectedVersion = snapshot.records.find((row) => row.id === id)?.version;
      if (expectedVersion === undefined)
        return Promise.reject(new Error("Record is not in this view."));
      return enqueue(
        id,
        (rows) => {
          const row = rows.get(id);
          if (row) rows.set(id, { ...row, version: row.version + 1 });
        },
        () => collection.setAccess(id, access, { expectedVersion }),
      );
    },
    dispose() {
      if (disposed) return;
      disposed = true;
      if (timer) clearInterval(timer);
      subscription?.unsubscribe();
      cancel(new Error("Live collection disposed. A sent write may still finish on the server."));
      listeners.clear();
      base.clear();
    },
  };
}