import type { Id, Json, OperationIntent } from '../canvas-core/types.ts';
import { fault, readMap } from '../canvas-core/values.ts';
import { createLedger, type FrozenRequest, type LedgerSnapshot, type Outcome } from './ledger.ts';
import { createDeque, createMinHeap } from './structures.ts';

export type OperationStatus = { readonly outcome: Outcome; readonly error?: string; readonly retryAfterMs?: number; readonly canonical?: Json; readonly blocked?: boolean };
export type CompiledRequest = { readonly kind?: FrozenRequest['kind']; readonly ownerCanvasId?: Id; readonly runId?: Id; readonly protocolVersion?: number; readonly payload: Json; readonly canvasIds: readonly Id[]; readonly expectedVersions: Readonly<Record<Id, number>> };
export type Clock = { now(): number; setTimer(callback: () => void, delay: number): unknown; clearTimer(handle: unknown): void };
const browserClock: Clock = { now: () => Date.now(), setTimer: (fn, delay) => setTimeout(fn, delay), clearTimer: handle => clearTimeout(handle as ReturnType<typeof setTimeout>) };
type Ledger = ReturnType<typeof createLedger>;
export type SchedulerOptions = {
  admit?: (intent:OperationIntent)=>boolean;
  getLedger(): LedgerSnapshot;
  mutate(reduce: (ledger: Ledger) => void): void;
  persist(): Promise<void>;
  createId(): Id;
  participants(intent: OperationIntent): readonly Id[];
  compile(intent: OperationIntent, requestId: Id): CompiledRequest;
  batchLimit?: (intents: readonly OperationIntent[]) => number;
  elide?: (intents: readonly OperationIntent[]) => readonly Id[];
  compileBatch?: (intents: readonly OperationIntent[], requestId: Id) => CompiledRequest;
  submit(request: FrozenRequest, signal: AbortSignal): Promise<OperationStatus>;
  query(request: FrozenRequest, signal: AbortSignal): Promise<OperationStatus>;
  supportsRequest?: (request: FrozenRequest) => boolean;
  /** Atomically reconcile the receipt/base/working document and the ledger outcome. */
  complete(request: FrozenRequest, status: OperationStatus): { readonly blocked: true } | void;
  onError(error: unknown): void;
  clock?: Clock;
  concurrency?: number;
  retryDelay?: (attempt: number) => number;
};

/** Queue/heap/claims are indexes only. Every authoritative transition lives in the ledger. */
export function createScheduler(options: SchedulerOptions) {
  const clock = options.clock ?? browserClock, concurrency = options.concurrency ?? 2;
  if (!Number.isSafeInteger(concurrency) || concurrency < 1) throw fault('invalid_concurrency');
  const admissionWaiters=new Set<Id>();
  const ready = createDeque<Id>(), queued = new Set<Id>(), claims = new Map<Id, Id>();
  const waiters = new Map<Id, Set<Id>>(), dependents = new Map<Id, Set<Id>>(), indexed = new Set<Id>();
  const retry = createMinHeap<{ id: Id; dueAt: number; sequence: number; token: number }>((a, b) => a.dueAt - b.dueAt || a.sequence - b.sequence);
  const tokens = new Map<Id, number>(), active = new Map<Id, { controller: AbortController; promise: Promise<void>; kind: FrozenRequest['kind'] }>();
  const holds = new Set<symbol>();
  const ordered: { id: Id; sequence: number }[] = [];
  let scheduledPump: Promise<void> | undefined;
  const laneWaiters = new Map<NonNullable<FrozenRequest['kind']>, Set<Id>>(), deferredRetries = new Map<Id, { id: Id; dueAt: number; sequence: number; token: number }>();
  let timer: unknown, pumpTimer: unknown, enqueueSequence = 0, stopped = true, closed = false, pumping = false;
  const delay = (attempt: number) => Math.max(1, options.retryDelay?.(attempt) ?? Math.min(30_000, 500 * 2 ** Math.min(attempt, 6)));
  const report = (error: unknown) => { try { options.onError(error); } catch { /* Reporting cannot change a receipt. */ } };
  const supported = (request: FrozenRequest) => options.supportsRequest?.(request) ?? request.protocolVersion === 1;
  const canRun = (kind: FrozenRequest['kind']) => active.size < concurrency && (kind !== 'asset.upload' || ![...active.values()].some(task => task.kind === 'asset.upload'));

  function enqueue(id: Id) {
    if (queued.has(id)) return;
    const record = options.getLedger().operations.get(id);
    if (!record || record.phase !== 'pending' || record.blockers.length || record.intent.dependencies.some(dependency => options.getLedger().operations.get(dependency)?.phase !== 'resolved')) {admissionWaiters.delete(id);return;}
    if(options.admit&&!options.admit(record.intent)){admissionWaiters.add(id);return;}
    admissionWaiters.delete(id);queued.add(id); ready.push(id);
  }
  function index(id: Id) {
    const record = options.getLedger().operations.get(id);
    if (!record) return;
    if (!indexed.has(id)) {
      indexed.add(id);
      let lo = 0, hi = ordered.length;
      while (lo < hi) { const mid = (lo + hi) >>> 1; if (ordered[mid].sequence < record.intent.sequence) lo = mid + 1; else hi = mid; }
      ordered.splice(lo, 0, { id, sequence: record.intent.sequence });
      for (const dependency of record.intent.dependencies) {
        const set = dependents.get(dependency) ?? new Set<Id>(); set.add(id); dependents.set(dependency, set);
      }
    }
    enqueue(id);
  }
  /** Only a contiguous mutable document prefix may share a request. Task barriers stay visible. */
  function batch(first: OperationIntent, canvases: Set<Id>): readonly OperationIntent[] {
    const selected = [first], ids = new Set([first.id]);
    if ((!options.compileBatch && !options.elide) || first.operationType !== 'document.change' || first.operationSchemaVersion !== 1) return selected;
    let lo = 0, hi = ordered.length, changeCount = first.changes.length;
    while (lo < hi) { const mid = (lo + hi) >>> 1; if (ordered[mid].sequence <= first.sequence) lo = mid + 1; else hi = mid; }
    const ledger = options.getLedger();
    for (let i = lo; i < ordered.length && selected.length < 64; i++) {
      const record = ledger.operations.get(ordered[i].id);
      if (!record || record.phase !== 'pending' || record.requestId || record.blockers.length || record.intent.operationType !== 'document.change' || record.intent.operationSchemaVersion !== 1) break;
      if (record.intent.dependencies.some(id => !ids.has(id) && ledger.operations.get(id)?.phase !== 'resolved')) break;
      if(options.admit&&!options.admit(record.intent))break;
      const scope = options.participants(record.intent);
      // Independent canvas lanes keep independent failure and retry boundaries.
      if (!scope.length || !scope.some(id => canvases.has(id)) || scope.some(id => claims.has(id)) || changeCount + record.intent.changes.length > 10_000) break;
      selected.push(record.intent); ids.add(record.intent.id); changeCount += record.intent.changes.length;
      for (const id of scope) canvases.add(id);
    }
    return selected;
  }
  function requestPump() {
    if (!options.compileBatch && !options.elide) { pump(); return; }
    if (!scheduledPump) scheduledPump = Promise.resolve().then(() => { scheduledPump = undefined; pump(); });
  }
  function own(request: FrozenRequest) {
    for (const canvas of request.canvasIds) {
      const prior = claims.get(canvas);
      if (prior && prior !== request.id) throw fault('recovery_claim_conflict', [canvas]);
    }
    for (const canvas of request.canvasIds) claims.set(canvas, request.id);
  }
  function release(request: FrozenRequest) {
    for (const canvas of request.canvasIds) if (claims.get(canvas) === request.id) {
      claims.delete(canvas);
      const waiting = waiters.get(canvas); waiters.delete(canvas);
      for (const id of waiting ?? []) enqueue(id);
    }
    for (const id of request.operationIds) for (const dependent of dependents.get(id) ?? []) enqueue(dependent);
  }
  function schedule(id: Id, dueAt: number, action: 'submit' | 'query', persist = true) {
    if (persist) options.mutate(ledger => ledger.scheduleRetry(id, dueAt, action));
    const token = (tokens.get(id) ?? 0) + 1; tokens.set(id, token);
    retry.push({ id, dueAt, sequence: ++enqueueSequence, token }); armTimer();
  }
  function validRetry(item: { id: Id; token: number }) {
    return tokens.get(item.id) === item.token && options.getLedger().requests.get(item.id)?.phase !== 'resolved';
  }
  function armTimer() {
    if (timer !== undefined) { clock.clearTimer(timer); timer = undefined; }
    while (retry.peek() && !validRetry(retry.peek()!)) retry.pop();
    if (stopped || holds.size || active.size >= concurrency || !retry.peek()) return;
    timer = clock.setTimer(() => { timer = undefined; pump(); }, Math.max(0, retry.peek()!.dueAt - clock.now()));
  }
  function launch(id: Id, action: 'submit' | 'query') {
    if (active.has(id) || stopped || holds.size) return;
    const controller = new AbortController();
    // Start in a microtask so the handle exists before any synchronous adapter error.
    const promise = Promise.resolve().then(async () => {
      try {
        if (stopped) return;
        if (action === 'submit') options.mutate(ledger => ledger.markDispatch(id));
        // This barrier includes the exact body/key and the conservative dispatch marker.
        await options.persist();
        if (stopped) return;
        const request = options.getLedger().requests.get(id)!;
        let status: OperationStatus;
        try { status = await (action === 'submit' ? options.submit(request, controller.signal) : options.query(request, controller.signal)); }
        catch { status = { outcome: 'unresolved', error: 'transport_unknown' }; }
        if (stopped) return; // A stale identity/runtime cannot accept a late callback.
        const completion = options.complete(request, status);
        const current = options.getLedger().requests.get(id)!;
        if (current.phase === 'resolved') { tokens.delete(id); await options.persist(); if (!stopped) release(current); }
        else if (completion?.blocked || status.blocked) { await options.persist(); }
        else {
          // A missing receipt is unknown, so the only write retry reuses this frozen request.
          const nextAction = status.outcome === 'accepted' || action === 'submit' ? 'query' : 'submit';
          schedule(id, clock.now() + Math.max(1, status.retryAfterMs ?? delay(current.attempt)), nextAction);
          await options.persist();
        }
      } catch (error) { report(error); } // Storage/reconciliation failure keeps its claims held.
    }).finally(() => {
      active.delete(id);
      if (!stopped) {
        for (const item of deferredRetries.values()) retry.push(item);
        deferredRetries.clear();
        for (const [kind, waiting] of laneWaiters) if (canRun(kind)) { for (const operationId of waiting) enqueue(operationId); laneWaiters.delete(kind); }
        pump();
      }
    });
    active.set(id, { controller, promise, kind: options.getLedger().requests.get(id)?.kind });
  }
  function pump() {
    if (stopped || holds.size || pumping) return;
    pumping = true;
    try {
      while (active.size < concurrency && retry.peek() && retry.peek()!.dueAt <= clock.now()) {
        const item = retry.pop()!;
        if (!validRetry(item)) continue;
        const request = options.getLedger().requests.get(item.id);
        if (request && !active.has(item.id)) {
          if (!canRun(request.kind)) { deferredRetries.set(item.id, item); continue; }
          launch(item.id, request.retryAction ?? 'query');
        }
      }
      let budget = Math.min(ready.size, 64);
      while (active.size < concurrency && budget-- > 0) {
        const id = ready.pop(); if (!id) break;
        queued.delete(id);
        const record = options.getLedger().operations.get(id);
        if (!record || record.phase !== 'pending' || record.requestId || record.blockers.length || record.intent.dependencies.some(dependency => options.getLedger().operations.get(dependency)?.phase !== 'resolved')) continue;
        try {
          if(options.admit&&!options.admit(record.intent)){admissionWaiters.add(id);continue;}
          if (!canRun(record.intent.operationType)) {
            const set = laneWaiters.get(record.intent.operationType) ?? new Set<Id>(); set.add(id); laneWaiters.set(record.intent.operationType, set); continue;
          }
          const canvases = [...new Set(options.participants(record.intent))].sort();
          if (!canvases.length && record.intent.operationType !== 'asset.upload' && record.intent.operationType !== 'run.cancel') throw fault('missing_participants');
          const occupied = canvases.filter(canvas => claims.has(canvas));
          if (occupied.length) {
            for (const canvas of occupied) { const set = waiters.get(canvas) ?? new Set<Id>(); set.add(id); waiters.set(canvas, set); }
            continue;
          }
          const scope = new Set(canvases), selected = batch(record.intent, scope);
          const limit = options.batchLimit?.(selected) ?? selected.length;
          if (!Number.isSafeInteger(limit) || limit < 1 || limit > selected.length) throw fault('invalid_batch_limit');
          const candidates = selected.slice(0, limit);
          const elided = options.elide?.(candidates) ?? [];
          if (elided.length) {
            if (elided.some((id, index) => id !== candidates[index]?.id || options.getLedger().operations.get(id)?.resolution !== 'elided')) throw fault('invalid_elision');
            for (const id of elided) for (const dependent of dependents.get(id) ?? []) enqueue(dependent);
            continue;
          }
          const intents = options.compileBatch ? candidates : [record.intent];
          const participants = [...new Set(intents.flatMap(intent => options.participants(intent)))].sort();
          const requestId = options.createId(), compiled = intents.length > 1 ? options.compileBatch!(intents, requestId) : options.compile(record.intent, requestId);
          if (participants.length !== compiled.canvasIds.length || participants.some(canvas => !compiled.canvasIds.includes(canvas))) throw fault('participant_mismatch');
          options.mutate(ledger => ledger.freeze({ ...compiled, id: requestId, operationIds: intents.map(intent => intent.id) }));
          own(options.getLedger().requests.get(requestId)!); launch(requestId, 'submit');
        } catch (error) { options.mutate(ledger => ledger.block(id, [(error as Error).message || 'compile_failed'])); report(error); }
      }
    } finally {
      pumping = false; armTimer();
      if (!stopped && !holds.size && ready.size && active.size < concurrency && pumpTimer === undefined) pumpTimer = clock.setTimer(() => { pumpTimer = undefined; pump(); }, 0);
    }
  }
  return Object.freeze({
    start() {
      if (closed) throw fault('scheduler_closed');
      if (!stopped) return;
      // Validate the entire recovered claim set before sending anything.
      claims.clear();
      for (const request of options.getLedger().requests.values()) if (request.phase !== 'resolved') own(request);
      stopped = false;
      for (const operation of [...options.getLedger().operations.values()].sort((a, b) => a.intent.sequence - b.intent.sequence)) index(operation.intent.id);
      for (const request of options.getLedger().requests.values()) if (request.phase !== 'resolved') {
        if (!supported(request)) { options.mutate(ledger => ledger.blockRequest(request.id, 'request_interpreter_unavailable')); continue; }
        schedule(request.id, request.retryAt ?? clock.now(), request.retryAction ?? (request.phase === 'frozen' ? 'submit' : 'query'), false);
      }
      pump();
    },
    refreshAdmission(){for(const id of [...admissionWaiters])enqueue(id);if(ready.size)requestPump();},
    notify(operationIds: readonly Id[]) { for (const id of operationIds) index(id); requestPump(); },
    /** Acquire synchronously, then settle(): already admitted work completes with its original identity. */
    hold() {
      if (closed) throw fault('scheduler_closed');
      const token = Symbol(); holds.add(token);
      if (timer !== undefined) clock.clearTimer(timer);
      if (pumpTimer !== undefined) clock.clearTimer(pumpTimer);
      timer = pumpTimer = undefined;
      return () => {
        if (!holds.delete(token) || holds.size || stopped) return;
        pump();
      };
    },
    /** Explicit retry after storage/auth/permission recovery; preserves frozen bodies. */
    async resume() {
      if (stopped) return;
      // A receipt may be in memory after its durable write failed. Keep claims until retry succeeds.
      await options.persist();
      if (stopped) return;
      for (const request of options.getLedger().requests.values()) {
        if (active.has(request.id)) continue;
        if (request.phase === 'resolved') release(request);
        else if (supported(request)) schedule(request.id, request.retryAt ?? clock.now(), request.retryAction ?? 'query', false);
      }
      for (const id of indexed) enqueue(id);
      pump();
    },
    getClaims: () => readMap(claims),
    async settle() {
      do {
        if (scheduledPump) await scheduledPump;
        if (active.size) await Promise.all([...active.values()].map(item => item.promise));
        if (!stopped && !holds.size && pumpTimer !== undefined) { clock.clearTimer(pumpTimer); pumpTimer = undefined; pump(); }
      } while (scheduledPump || active.size || (!stopped && !holds.size && pumpTimer !== undefined));
    },
    stop() { closed = true; stopped = true; if (timer !== undefined) clock.clearTimer(timer); if (pumpTimer !== undefined) clock.clearTimer(pumpTimer); timer = pumpTimer = undefined; for (const item of active.values()) item.controller.abort(); },
  });
}