'use client';

/**
 * The suspensable store — the runtime-level query cache that backs `query`,
 * `suspend`, and (as the one cancellation standard) a raw `yield* asyncEffect`.
 *
 * It solves three problems at once:
 *
 *   1. **Dedup across render attempts.** A component that suspends *before it
 *      first commits* is re-rendered with fresh refs, so a per-instance cache
 *      can't keep its in-flight promise stable. This store lives outside the
 *      render lifecycle; every attempt of the same logical query finds the same
 *      entry and runs the effect once.
 *   2. **Cross-component dedup by key (#40).** A *keyed* query is deduped
 *      runtime-wide: two components under one `mount` issuing the identical
 *      `query(fetchUser(1), 1)` share one fetch, regardless of component type or
 *      yield position. A *keyless* `suspend` / raw async has no user identity, so
 *      it stays load-once-by-position. Keys are compared with Effect `Hash`/`Equal`
 *      (structural, property-order-insensitive in v4), so `{a:1,b:2}` and
 *      `{b:2,a:1}` hit one entry and distinct keys never collide.
 *   3. **Live cache: invalidation, refetch, staleTime (#41–#43).** Each entry is
 *      observable (a version + listeners). A read subscribes the component to its
 *      entry, so `invalidate(tag)`, a manual `refetch`, or a background
 *      revalidation (past `staleTime`) re-renders exactly the components reading
 *      that data. A read then `use`s the entry's current promise — suspending on
 *      first load, returning the (possibly stale) value otherwise, or throwing a
 *      failure for `.catch`.
 *
 * Entries are reference-counted by *committed* component instances (claims); an
 * entry with no claims is interrupted when superseded, or swept a short while
 * after it settles as a backstop, so the store can't grow without bound.
 */
import { useSyncExternalStore } from 'react';
import * as Equal from 'effect/Equal';
import * as Hash from 'effect/Hash';
import type { AnyEffect } from '#domain/protocol.ts';
import type {
  Executor,
  SuspendRequest,
  SuspensableResolver,
  Suspender,
} from '#application/ports.ts';

type Status = 'pending' | 'success' | 'error';

interface Entry {
  /** A unique id, the token the claim bookkeeping tracks this entry by. */
  readonly id: string;
  /** The `(runtimeScope, componentScope, index)` slot — its supersede group. */
  readonly slot: string;
  /** The dedup bucket: runtime-wide for a keyed query, positional for a keyless one. */
  readonly bucket: string;
  /** The runtime this entry belongs to (for tag invalidation scoping). */
  readonly runtimeScope: string;
  /** The user key, kept for the `Equal.equals` confirmation within a bucket. */
  readonly key: unknown;
  /** Invalidation tags for this entry (empty for keyless / untagged). */
  readonly tags: ReadonlyArray<unknown>;
  /** Freshness window in ms; `0` = no automatic revalidation. */
  readonly staleTime: number;
  /**
   * How to (re)run the work. An effect-backed query can refetch; a raw-async
   * fiber cannot (there is no effect to re-run), so `refetch` is `null` for it. A
   * server-seeded entry starts `null` and adopts the client query's effect the
   * first time it is read (see `attachRunner`), so refetch/staleTime then work.
   */
  refetch: ((background: boolean) => void) | null;
  /** The promise React's `use` consumes — resolved to the current value, or pending. */
  promise: Promise<unknown>;
  /** Cancel the in-flight work: abort a query's fiber, or interrupt a raw fiber. */
  cancel: () => void;
  status: Status;
  value: unknown;
  fetchedAt: number;
  /**
   * Monotonic per-entry fetch generation. Every (re)fetch captures the value it
   * bumped to; when it settles it writes back only if the entry is still on that
   * generation. A later fetch (manual refetch, invalidation, staleTime reload)
   * supersedes an earlier one, so an out-of-order response can't clobber newer data.
   */
  fetchId: number;
  /** True while a background revalidation is running (so we don't stack them). */
  revalidating: boolean;
  /** Bumps on every state change; the read subscribes to it. */
  version: number;
  readonly listeners: Set<() => void>;
  readonly subscribe: (listener: () => void) => () => void;
  claims: number;
  sweep: ReturnType<typeof setTimeout> | null;
}

/** How long an unclaimed entry lingers after settling before it is swept. */
const SWEEP_MS = 10_000;

/** id → entry, for release/evict by the token the claims track. */
const store = new Map<string, Entry>();
/** Dedup bucket → the entries that hashed there. */
const buckets = new Map<string, Set<Entry>>();
/** `${runtimeScope} ${componentScope} ${index}` → the entries in that slot. */
const slots = new Map<string, Set<Entry>>();
/** The set of entry ids each committed component instance currently holds. */
const claims = new Map<string, ReadonlySet<string>>();

/** A stable-per-handle scope id, so two RECs never share a keyless namespace. */
let scopeSeq = 0;
const scopes = new WeakMap<object, string>();
export const scopeOf = (handle: object): string => {
  let scope = scopes.get(handle);
  if (scope === undefined) {
    scope = `s${scopeSeq++}`;
    scopes.set(handle, scope);
  }
  return scope;
};

/**
 * A stable-per-runtime scope id. Keyed queries dedupe *within* one runtime — so
 * two components under the same `mount` sharing a key share one fetch (#40) — but
 * never across runtimes (a second mount, or a per-request server runtime, stays
 * isolated). Keyed by the `ManagedRuntime` object itself, so code *outside* the
 * render (the `/ssr` `renderToStream`, which builds the runtime and then collects
 * this scope's settled queries) resolves the same scope as the components inside.
 */
let runtimeSeq = 0;
const runtimeScopes = new WeakMap<object, string>();
export const runtimeScopeOf = (runtime: object): string => {
  let scope = runtimeScopes.get(runtime);
  if (scope === undefined) {
    scope = `r${runtimeSeq++}`;
    runtimeScopes.set(runtime, scope);
  }
  return scope;
};

/** A per-instance id, minted lazily and kept in a ref by the adapter. */
let instanceSeq = 0;
export const nextInstanceId = (): string => `i${instanceSeq++}`;

/** A fresh entry id — the claim bookkeeping's stable token for an entry. */
let entrySeq = 0;

const slotKey = (runtimeScope: string, componentScope: string, index: number): string =>
  `${runtimeScope} ${componentScope} ${index}`;

const bucketKeyFor = (runtimeScope: string, slot: string, key: unknown): string =>
  key === undefined ? `${runtimeScope} p ${slot}` : `${runtimeScope} k ${Hash.hash(key)}`;

const notify = (entry: Entry): void => {
  entry.version += 1;
  for (const listener of Array.from(entry.listeners)) {
    listener();
  }
};

const clearSweep = (entry: Entry): void => {
  if (entry.sweep !== null) {
    clearTimeout(entry.sweep);
    entry.sweep = null;
  }
};

/** Drop every index of an entry and interrupt its fiber. */
const evict = (entry: Entry): void => {
  if (store.get(entry.id) !== entry) {
    return;
  }
  clearSweep(entry);
  store.delete(entry.id);
  buckets.get(entry.bucket)?.delete(entry);
  slots.get(entry.slot)?.delete(entry);
  entry.cancel();
};

const releaseId = (id: string): void => {
  const entry = store.get(id);
  if (entry === undefined) {
    return;
  }
  entry.claims -= 1;
  if (entry.claims <= 0) {
    evict(entry);
  }
};

const interruptSupersededSiblings = (slot: string, current: Entry): void => {
  const group = slots.get(slot);
  if (group === undefined) {
    return;
  }
  for (const sibling of Array.from(group)) {
    if (sibling !== current && sibling.claims === 0) {
      evict(sibling);
    }
  }
};

/**
 * A settled promise carrying React's thenable protocol (`status`/`value` or
 * `reason`), so a `use()` of it resolves *synchronously*. Without the stamp a
 * fresh `Promise.resolve(value)` reads as unknown-status: React suspends again
 * on a value it already has — an extra render round on every settle, a dev
 * "uncached promise" warning, and (after a refetch/invalidation) the reveal
 * waits out React's ~300ms fallback throttle.
 */
const fulfilledPromise = (value: unknown): Promise<unknown> => {
  const promise = Promise.resolve(value) as Promise<unknown> & {
    status?: 'fulfilled';
    value?: unknown;
  };
  promise.status = 'fulfilled';
  promise.value = value;
  return promise;
};

const rejectedPromise = (error: unknown): Promise<unknown> => {
  const promise = Promise.reject(error) as Promise<unknown> & {
    status?: 'rejected';
    reason?: unknown;
  };
  promise.status = 'rejected';
  promise.reason = error;
  promise.catch(() => {}); // observed here; `use` still throws it for render
  return promise;
};

/** Attach the settle/sweep observers to a fresh promise on an entry. */
const observeSettlement = (entry: Entry, promise: Promise<unknown>): void => {
  const onSettled = (): void => {
    if (entry.claims === 0 && store.get(entry.id) === entry) {
      entry.sweep = setTimeout(() => {
        if (entry.claims === 0 && store.get(entry.id) === entry) {
          evict(entry);
        }
      }, SWEEP_MS);
      (entry.sweep as { unref?: () => void }).unref?.();
    }
  };
  promise.then(onSettled, onSettled);
};

/** Find the entry in a bucket whose key is `Equal`-equal — else undefined. */
const findByEqual = (entries: Set<Entry>, key: unknown): Entry | undefined => {
  for (const entry of entries) {
    if (Equal.equals(entry.key, key)) {
      return entry;
    }
  }
  return undefined;
};

/**
 * Build the resolver a single render pass uses. It records into `used` every
 * entry id it touches, so the adapter can reconcile this instance's claims once
 * the render commits. `onSuspend` fires when a read is about to suspend on a
 * pending entry — React's `use` then throws its (non-thenable) suspension
 * sentinel, and the flag is how the adapter knows to re-throw it rather than
 * mistake it for a failure (a `.catchDefect` must never absorb a suspension).
 * `onResolve` (dev only) reports which entry a suspending yield hit and the
 * entry's current version — the render ledger compares it across renders to
 * attribute a re-render to a query settle/refetch/invalidation; fired before
 * the read may suspend, so even a suspending render records its entry.
 */
export const makeSuspensableResolver = (
  runtimeScope: string,
  componentScope: string,
  executor: Executor,
  suspender: Suspender,
  used: Set<string>,
  onSuspend?: () => void,
  onResolve?: (index: number, entryId: string, version: number, key: unknown) => void,
): SuspensableResolver => ({
  resolve: (request: SuspendRequest, index: number): unknown => {
    const slot = slotKey(runtimeScope, componentScope, index);
    const bucket = bucketKeyFor(runtimeScope, slot, request.key);
    let entriesInBucket = buckets.get(bucket);
    let entry = entriesInBucket ? findByEqual(entriesInBucket, request.key) : undefined;
    if (entry === undefined) {
      entry = createEntry(runtimeScope, slot, bucket, request, executor);
      if (entriesInBucket === undefined) {
        entriesInBucket = new Set();
        buckets.set(bucket, entriesInBucket);
      }
      entriesInBucket.add(entry);
      let group = slots.get(slot);
      if (group === undefined) {
        group = new Set();
        slots.set(slot, group);
      }
      group.add(entry);
    } else if ('fiber' in request) {
      // Cache hit for a raw async yield: this render attempt spun up its own probe
      // fiber (the effect already ran once), but a prior attempt's fiber is the one
      // we hold. Interrupt the redundant one so it doesn't run on.
      executor.interrupt(request.fiber);
    } else if (entry.refetch === null) {
      // A server-seeded entry, read for the first time by the client query that
      // matches its key — adopt that query's effect so refetch/staleTime work.
      attachRunner(entry, request.effect, executor);
    }
    interruptSupersededSiblings(slot, entry);
    used.add(entry.id);
    onResolve?.(index, entry.id, entry.version, entry.key);

    // Hook 1: subscribe this render to the entry, so invalidate / refetch / a
    // background revalidation re-renders it. `entry.subscribe` is stable per entry.
    const resolved = entry;
    useSyncExternalStore(
      resolved.subscribe,
      () => resolved.version,
      () => resolved.version,
    );

    // Stale-while-revalidate: a fresh value past its staleTime is served now, with
    // a background refetch kicked off (once).
    if (
      resolved.status === 'success' &&
      resolved.staleTime > 0 &&
      !resolved.revalidating &&
      resolved.refetch !== null &&
      now() - resolved.fetchedAt > resolved.staleTime
    ) {
      resolved.revalidating = true;
      resolved.refetch(true);
    }

    // Hook 2: use the entry's current promise — suspend, return, or throw for `.catch`.
    if (resolved.status === 'pending') {
      onSuspend?.();
    }
    return suspender.use(resolved.promise);
  },
});

/** Wall-clock now, isolated so tests can drive it via fake timers. */
const now = (): number => Date.now();

const createEntry = (
  runtimeScope: string,
  slot: string,
  bucket: string,
  request: SuspendRequest,
  executor: Executor,
): Entry => {
  const listeners = new Set<() => void>();
  const entry: Entry = {
    id: `e${entrySeq++}`,
    slot,
    bucket,
    runtimeScope,
    key: request.key,
    tags: 'fiber' in request ? [] : (request.tags ?? []),
    staleTime: 'fiber' in request ? 0 : (request.staleTime ?? 0),
    refetch: null, // set below for effect-backed queries
    promise: Promise.resolve(undefined),
    cancel: () => {},
    status: 'pending',
    value: undefined,
    fetchedAt: 0,
    fetchId: 0,
    revalidating: false,
    version: 0,
    listeners,
    subscribe: (listener) => {
      listeners.add(listener);
      return () => listeners.delete(listener);
    },
    claims: 0,
    sweep: null,
  };
  store.set(entry.id, entry);

  if ('fiber' in request) {
    // Raw async: the effect is already running on this fiber — join it (no re-run).
    const fiber = request.fiber;
    entry.promise = executor.join(fiber);
    entry.cancel = () => executor.interrupt(fiber);
    trackOutcome(entry, entry.promise);
    observeSettlement(entry, entry.promise);
    return entry;
  }

  if (request.scoped) {
    // A component-lifetime scoped resource: run against a fresh scope and cancel
    // by closing it (running finalizers on unmount). No refetch — acquire once.
    const { promise, close } = executor.withScope(request.effect);
    entry.promise = promise;
    entry.cancel = close;
    trackOutcome(entry, promise);
    observeSettlement(entry, promise);
    return entry;
  }

  attachRunner(entry, request.effect, executor);
  entry.refetch?.(false); // initial fetch
  return entry;
};

/**
 * Give an entry its (re)fetch routine — reused for the initial run, staleTime
 * revalidation, manual refetch, and invalidation. Also how a server-seeded entry
 * (which starts with no effect) adopts the client query's effect on first read.
 */
const attachRunner = (entry: Entry, effect: AnyEffect, executor: Executor): void => {
  entry.refetch = (background: boolean): void => {
    // Supersede any fetch already in flight: abort its work (no wasted request)
    // and, via the generation bump below, discard its result if it lands late.
    entry.cancel();
    const fetchId = (entry.fetchId += 1);
    const superseded = (): boolean => entry.fetchId !== fetchId;

    const controller = new AbortController();
    const promise = executor.runPromise(effect, controller.signal);
    entry.cancel = () => controller.abort();
    if (!background) {
      // Hard (re)fetch: show pending — the read suspends until it settles.
      entry.status = 'pending';
      entry.promise = promise;
    }
    promise.then(
      (value) => {
        if (superseded()) return; // a newer fetch owns the entry — don't clobber it
        entry.revalidating = false;
        entry.status = 'success';
        entry.value = value;
        entry.fetchedAt = now();
        entry.promise = fulfilledPromise(value);
        notify(entry);
      },
      (error: unknown) => {
        if (superseded()) return; // includes this fetch's own abort when superseded
        entry.revalidating = false;
        // A background revalidation that fails keeps the last good value on screen.
        if (background && entry.status === 'success') {
          notify(entry);
          return;
        }
        entry.status = 'error';
        entry.promise = rejectedPromise(error);
        notify(entry);
      },
    );
    observeSettlement(entry, promise);
  };
};

/** Track a raw-async fiber's outcome onto the entry (no refetch). */
const trackOutcome = (entry: Entry, promise: Promise<unknown>): void => {
  promise.then(
    (value) => {
      entry.status = 'success';
      entry.value = value;
      entry.fetchedAt = now();
    },
    () => {
      entry.status = 'error';
    },
  );
};

export const reconcileClaims = (instanceId: string, used: ReadonlySet<string>): void => {
  const held = claims.get(instanceId) ?? new Set<string>();
  for (const id of used) {
    if (!held.has(id)) {
      const entry = store.get(id);
      if (entry !== undefined) {
        entry.claims += 1;
        clearSweep(entry);
      }
    }
  }
  for (const id of held) {
    if (!used.has(id)) {
      releaseId(id);
    }
  }
  claims.set(instanceId, new Set(used));
};

export const releaseClaims = (instanceId: string): void => {
  const held = claims.get(instanceId);
  if (held === undefined) {
    return;
  }
  for (const id of held) {
    releaseId(id);
  }
  claims.delete(instanceId);
};

// --- the live-cache API: invalidation + manual refetch (#41, #43) ---

/**
 * Refetch every entry under `runtimeScope` whose tags match `matches`. An entry
 * with active subscribers is revalidated in the background (its readers keep
 * showing the current value, then update); an entry with none is evicted, so its
 * next read fetches afresh. Returns how many entries matched.
 */
export const invalidateByTag = (
  runtimeScope: string,
  matches: (tag: unknown) => boolean,
): number => {
  let count = 0;
  for (const entry of Array.from(store.values())) {
    if (entry.runtimeScope !== runtimeScope || entry.tags.length === 0) {
      continue;
    }
    if (!entry.tags.some(matches)) {
      continue;
    }
    count += 1;
    if (entry.listeners.size > 0 && entry.refetch !== null) {
      entry.revalidating = true;
      entry.refetch(true); // background — keep showing current data while it reloads
    } else {
      evict(entry); // inactive — drop it so the next read fetches afresh
    }
  }
  return count;
};

/**
 * Refetch every entry under `runtimeScope` whose key is `Equal`-equal to `key`.
 * A failed or completed query re-runs without a key change or a remount. Returns
 * how many entries matched.
 */
export const refetchByKey = (runtimeScope: string, key: unknown): number => {
  let count = 0;
  for (const entry of Array.from(store.values())) {
    if (entry.runtimeScope !== runtimeScope || entry.refetch === null) {
      continue;
    }
    if (!Equal.equals(entry.key, key)) {
      continue;
    }
    count += 1;
    // Keep a good value visible (background); a failed one refetches hard (suspends).
    entry.refetch(entry.status === 'success');
  }
  return count;
};

/** A serialisable snapshot of a query entry — for the dev inspector (#55). */
export interface QuerySnapshot {
  readonly key: unknown;
  readonly status: Status;
  readonly claims: number;
  readonly tags: ReadonlyArray<unknown>;
  readonly staleTime: number;
}

/** Snapshot every live cache entry (keyed queries only) — the inspector's data source. */
export const inspectStore = (): ReadonlyArray<QuerySnapshot> =>
  Array.from(store.values())
    .filter((entry) => entry.key !== undefined)
    .map((entry) => ({
      key: entry.key,
      status: entry.status,
      claims: entry.claims,
      tags: entry.tags,
      staleTime: entry.staleTime,
    }));

/** A server-resolved query result — the shape that crosses the wire (#49). */
export interface SeededQuery {
  readonly key: unknown;
  readonly value: unknown;
  /** Server fetch time (ms) — the staleness clock starts here. Defaults to now. */
  readonly fetchedAt?: number;
}

/**
 * Snapshot every *keyed, settled* query in one runtime's scope as seedable
 * results — what the server embeds in the stream so the matching client `query`
 * renders without suspending or re-fetching. Keyless `suspend`s have no user
 * identity to match on the client and pending/failed entries have no value worth
 * shipping (the client will fetch), so both are skipped. Keys and values cross
 * the wire as JSON, so both must be JSON-serialisable.
 */
export const collectSeededQueries = (runtimeScope: string): ReadonlyArray<SeededQuery> =>
  Array.from(store.values())
    .filter(
      (entry) =>
        entry.runtimeScope === runtimeScope &&
        entry.key !== undefined &&
        entry.status === 'success',
    )
    .map((entry) => ({ key: entry.key, value: entry.value, fetchedAt: entry.fetchedAt }));

/**
 * Seed the cache with a query result the server already resolved (#49): the
 * client renders it without suspending or re-fetching. The entry starts with no
 * effect — the matching client `query` adopts its effect on first read, so
 * refetch/staleTime work afterward. `fetchedAt` starts at the server fetch time
 * (`when`), so the staleness clock is honest. A key already present is left alone.
 */
export const seedQuery = (
  runtimeScope: string,
  key: unknown,
  value: unknown,
  when: number = now(),
): void => {
  const bucket = bucketKeyFor(runtimeScope, `${runtimeScope} seed`, key);
  const existing = buckets.get(bucket);
  if (existing && findByEqual(existing, key)) {
    return; // already seeded or live
  }
  const listeners = new Set<() => void>();
  const entry: Entry = {
    id: `e${entrySeq++}`,
    slot: `${runtimeScope} seed`,
    bucket,
    runtimeScope,
    key,
    tags: [],
    staleTime: 0,
    refetch: null, // adopted on first client read
    promise: fulfilledPromise(value),
    cancel: () => {},
    status: 'success',
    value,
    fetchedAt: when,
    fetchId: 0,
    revalidating: false,
    version: 0,
    listeners,
    subscribe: (listener) => {
      listeners.add(listener);
      return () => listeners.delete(listener);
    },
    claims: 0,
    sweep: null,
  };
  store.set(entry.id, entry);
  const set = existing ?? new Set<Entry>();
  set.add(entry);
  buckets.set(bucket, set);
  observeSettlement(entry, entry.promise);
};