/**
 * The reactive atoms — the server-safe half of the reactivity bridge. This
 * module carries no `'use client'` and imports no React: an `atom`/`derive` value
 * is pure Effect-backed data (a value you can read, write, subscribe to, and
 * `yield*`), so a *service* that holds one stays universal. The React binding
 * (`observe`, `useAtom`, and the in-render reader) lives in the sibling
 * `react/reactivity.tsx`, tagged `'use client'`; only *reading a signal in a
 * component* needs React, not owning one in a service.
 */
import * as Effect from 'effect/Effect';
import * as Equal from 'effect/Equal';
import * as Schema from 'effect/Schema';
import type * as Scope from 'effect/Scope';
import * as Stream from 'effect/Stream';
import { AtomRef } from 'effect/unstable/reactivity';
import { devConnected, emitDev } from '#infrastructure/devtools-core.ts';
import {
  AtomTypeId,
  isAtom,
  query,
  type Atom,
  type ReadableAtom,
  type Suspensable,
  type Volatile,
} from '#domain/protocol.ts';

/** Read an atom inside an `observe`/`<Observe>` selector, subscribing to it. */
export type Read = <A>(atom: ReadableAtom<A>) => A;

/** The single-shot iterator that makes an atom `yield*`able: read + subscribe. */
const yieldSelf = <A>(self: ReadableAtom<A>): Iterator<ReadableAtom<A>, A> => {
  let yielded = false;
  return {
    next(sent?: unknown): IteratorResult<ReadableAtom<A>, A> {
      if (yielded) {
        return { done: true, value: sent as A };
      }
      yielded = true;
      return { done: false, value: self };
    },
  };
};

/**
 * Batching: coalesce a burst of writes into a single notification wave. While a
 * `batch` runs, every atom/derived listener that would fire is collected instead
 * of called; when the outermost batch ends, each is called exactly once. So a
 * derived value that reads two atoms both changed in one batch recomputes once,
 * and a component reading them re-renders once, not twice. Batches nest — only
 * the outermost flush notifies.
 */
let batchDepth = 0;
const pending = new Set<() => void>();

const flush = (): void => {
  // Snapshot + clear before running: a listener may schedule further work, but
  // at depth 0 that fires synchronously and must not mutate the set we iterate.
  const wave = Array.from(pending);
  pending.clear();
  for (const listener of wave) {
    listener();
  }
};

/** Notify a listener set, deferring into the current batch if one is open. */
const notify = (listeners: Set<() => void>): void => {
  if (batchDepth > 0) {
    for (const listener of listeners) {
      pending.add(listener);
    }
  } else {
    // Snapshot: a listener may subscribe/unsubscribe while being notified.
    for (const listener of Array.from(listeners)) {
      listener();
    }
  }
};

/** Dev mode, resolved defensively (bundlers replace this to `false` in prod). */
const isDevMode = (): boolean =>
  typeof process !== 'undefined' && process.env?.['NODE_ENV'] !== 'production';

const isThenable = (u: unknown): boolean =>
  typeof u === 'object' && u !== null && typeof (u as { then?: unknown }).then === 'function';

/**
 * Run `writes` as one atomic notification wave: atoms may be `set` many times
 * inside, but subscribers (and the derived atoms and components downstream) are
 * notified once, after it returns. Returns whatever `writes` returns.
 *
 * ```ts
 * batch(() => {
 *   first.set('Ada');
 *   last.set('Lovelace');
 * }); // one re-render, not two
 * ```
 *
 * **Synchronous-only, single-flight.** `batchDepth` is module-global mutable
 * state, so an `await` inside `writes` would leave the batch *open* across the
 * suspension point — coalescing every unrelated write until it resumes, and on a
 * server interleaving with other requests' writes. So `writes` must be
 * synchronous: do the async work first, then `batch` the synchronous writes it
 * produced. In development this is enforced — a `writes` that returns a `Promise`
 * (an `async` function, or an awaited value) throws, rather than silently
 * corrupting the batch depth.
 */
export const batch = <A>(writes: () => A): A => {
  batchDepth += 1;
  let result: A;
  try {
    result = writes();
  } finally {
    batchDepth -= 1;
    if (batchDepth === 0) {
      flush();
    }
  }
  if (isDevMode() && isThenable(result)) {
    throw new Error(
      'effract: batch(writes) is synchronous-only, but its callback returned a Promise (an `async` ' +
        'function, or an awaited value). Batching coalesces a burst of *synchronous* writes into one ' +
        'notification wave; an `await` inside would leave the batch open across suspension points and ' +
        'interleave with unrelated writes (on a server, across requests). Do the async work first, then ' +
        'batch(...) the synchronous writes.',
    );
  }
  return result;
};

/** Sequence for the internal, unlabeled cells (derived atoms, machinery). */
let atomSeq = 0;

/**
 * Debug names for atoms, so devtools can label them in the render trace and the
 * dependency graph — a WeakMap, not a field, so the public atom shape is
 * unchanged. An unregistered atom (or a non-atom) reads as `'atom'`.
 */
const atomNames = new WeakMap<object, string>();
export const atomNameOf = (atom: object): string => atomNames.get(atom) ?? 'atom';

/**
 * The unlabeled runtime constructor, for *internal* cells only (a derived atom's
 * backing store, a stream drain). The public `atom(label, initial)` requires a
 * label — every user-visible cell is named; this keeps the `atomSeq` fallback
 * for the machinery that has no user-facing name.
 */
const makeAtom = <A>(initial: A, name?: string): Atom<A> => {
  const ref = AtomRef.make(initial);
  const listeners = new Set<() => void>();
  const id = `atom${atomSeq++}`;
  const label = name ?? id;
  const write = (next: A, action: 'set' | 'update'): void => {
    if (Equal.equals(next, ref.value)) {
      return; // no-op write — nothing downstream needs to hear about it
    }
    ref.set(next);
    // Dev instrumentation — a null check in production; the event is built only
    // when a devtools bridge / inspector is connected.
    if (devConnected()) {
      emitDev({ kind: 'atom:write', id, name: label, value: next, action });
    }
    notify(listeners);
  };
  const self: Atom<A> = {
    [AtomTypeId]: true,
    get value() {
      return ref.value;
    },
    set: (value) => write(value, 'set'),
    update: (update) => write(update(ref.value), 'update'),
    subscribe: (listener) => {
      listeners.add(listener);
      return () => {
        listeners.delete(listener);
      };
    },
    derive: (f) => deriveReadonly((read) => f(read(self))),
    [Symbol.iterator]: () => yieldSelf(self),
  };
  atomNames.set(self, label);
  return self;
};

/**
 * Create a writable reactive atom — **label first**, exactly like
 * `Context.Service<X>()('X', …)` names a service. The label is captured as a
 * literal type and is the cell's one canonical name: it is what `RendersOn`
 * reports, what the devtools render trace and dependency graph display, and
 * what the dev render ledger attributes a re-render to. Read the atom with
 * `yield*` in a REC or `.value` imperatively (in an event handler or a service
 * method); write it with `.set` / `.update`. Backed by Effect's `AtomRef` for
 * storage, so its value is reachable from anywhere an Effect runs; writes that
 * don't change the value (by `Equal`) notify no one, and writes made inside
 * {@link batch} are coalesced.
 *
 * ```ts
 * const count = atom('count', 0); // Atom<number, never, 'count'>
 * ```
 */
const atomImpl = <const L extends string, A>(label: L, initial: A): Atom<A, never, L> =>
  // `L` is phantom — the runtime cell carries its label in the names WeakMap,
  // never as a field — so narrowing the wide default is a pure type refinement.
  makeAtom(initial, label) as Atom<A, never, L>;

/**
 * Bridge an Effect `Stream` into a reactive atom (server-safe) — **born
 * {@link Volatile}**. A stream emits at its source's own frequency, so every
 * subscriber would re-render at that frequency; the returned atom therefore
 * carries the volatility obligation `V`, and `mount` refuses the tree until the
 * read is deferred (`defer(atom)`) or the reading REC acknowledges the heat
 * (`.hot()`). Label first, like `atom(label, initial)`.
 *
 * Seeds the atom with `initial`, then drains every emission into it on a fiber
 * **forked into the current scope** — so it stops (and is interrupted) exactly
 * when that scope closes. Used in a service's `make`, the fiber is tied to the
 * runtime's lifetime: live data (a websocket, a poll, a `PubSub`) becomes an
 * ordinary atom every component reads with `yield*`, re-rendering per emission.
 * Latest-wins: the atom always holds the most recent value. A stream error ends
 * the drain on its fiber (handle it in the stream with `Stream.catchAll` if you
 * need a fallback).
 *
 * ```ts
 * class Ticker extends Context.Service<Ticker>()('Ticker', {
 *   make: Effect.gen(function* () {
 *     const price = yield* atom.fromStream('price', priceStream, 0); // Atom<number, Volatile, 'price'>
 *     return { price };
 *   }),
 * }) { static layer = Layer.effect(Ticker, Ticker.make); }
 * ```
 */
const atomFromStream = <const L extends string, A, E, R>(
  label: L,
  stream: Stream.Stream<A, E, R>,
  initial: A,
): Effect.Effect<Atom<A, Volatile, L>, never, R | Scope.Scope> =>
  Effect.gen(function* () {
    // `Volatile` and `L` are phantom: the runtime cell is an ordinary atom (the
    // obligation and the label exist only in the type), so the refinement is a cast.
    const cell = makeAtom(initial, label) as Atom<A, never, L>;
    yield* Effect.forkScoped(
      Stream.runForEach(stream, (value) => Effect.sync(() => cell.set(value))),
    );
    return cell;
  });

/**
 * A writable atom that can cross the wire: it carries a stable `serialKey` and a
 * `Schema` for encode/decode, so its server-computed value can seed the client
 * (see {@link dehydrate} / {@link hydrate}). Everything else is an ordinary atom
 * — including the render channel: `V`/`L` ride along exactly as on {@link Atom},
 * and the serial key doubles as the label `L`.
 */
export interface SerializableAtom<A, out V = never, out L extends string = string> extends Atom<
  A,
  V,
  L
> {
  readonly serialKey: string;
  /** Schema-encode the current value (the shape that crosses the wire). */
  readonly encode: () => unknown;
  /** Schema-decode `encoded` and set it — how a hydrated value seeds the atom. */
  readonly hydrate: (encoded: unknown) => void;
}

/**
 * Every serializable atom, by `serialKey` — the set {@link dehydrate} encodes and
 * {@link hydrate} seeds. Server-side, keep this per request: build a fresh runtime
 * per render (the default `mount` scope) and dehydrate right after driving the
 * tree, so one request's atoms don't mix with another's (a module-level registry
 * is process-global, the same footgun as a module-level atom).
 */
const serialRegistry = new Map<string, SerializableAtom<unknown>>();

/**
 * Server values whose atom has not registered yet. {@link hydrate} runs at the
 * top of the tree, but an atom held by a service is only created the first time
 * a component resolves that service — *later* in the same render pass. Stash the
 * encoded value here by `serialKey` and let registration consume it, so seeding
 * is order-independent. Consume-once: a runtime built long after hydration (a
 * second mount of the same services, a `.catch` retry) keeps the by-then-current
 * client state rather than resurrecting stale server state.
 */
const pendingHydration = new Map<string, unknown>();

/**
 * Create a serializable atom — **key first**, because the key *is* the label:
 * one literal names the cell for `RendersOn`, the devtools trace, *and* the
 * hydration payload it crosses the wire under. Everything else is an ordinary
 * `atom(label, initial)` plus the Schema round-trip.
 */
const serializableAtom = <const L extends string, A, E>(
  key: L,
  initial: A,
  schema: Schema.Codec<A, E, never, never>,
): SerializableAtom<A, never, L> => {
  const base = makeAtom(initial, key);
  const encode = Schema.encodeUnknownSync(schema);
  const decode = Schema.decodeUnknownSync(schema);
  const self: SerializableAtom<A> = Object.assign(base, {
    serialKey: key,
    encode: () => encode(base.value),
    hydrate: (encoded: unknown) => base.set(decode(encoded)),
  });
  serialRegistry.set(key, self as unknown as SerializableAtom<unknown>);
  if (pendingHydration.has(key)) {
    const encoded = pendingHydration.get(key);
    pendingHydration.delete(key);
    self.hydrate(encoded);
  }
  // `L` is phantom (the key already names the runtime cell); pure type refinement.
  return self as SerializableAtom<A, never, L>;
};

/**
 * Snapshot every serializable atom's Schema-encoded value into a plain,
 * JSON-serialisable payload — what the server embeds in the stream so the client
 * can seed its atoms with the same state (no hydration mismatch, no refetch).
 *
 * Internal: `renderToStream` (`@tmonier/effract/ssr`) calls this after the render
 * settles and embeds the payload automatically — user code never does.
 */
export const dehydrate = (): Record<string, unknown> => {
  const out: Record<string, unknown> = {};
  for (const [key, serializable] of serialRegistry) {
    out[key] = serializable.encode();
  }
  return out;
};

/**
 * Seed serializable atoms from a {@link dehydrate} payload — Schema-decoding each
 * value into its atom by `serialKey`. Runs before the first render so the very
 * first read sees the server value. A key whose atom does not exist yet (it lives
 * in a service not resolved until deeper in the tree) is stashed and applied the
 * moment that atom registers.
 *
 * Internal: the client `mount` finds the payload the server embedded and calls
 * this automatically — user code never does.
 */
export const hydrate = (payload: Record<string, unknown>): void => {
  for (const [key, encoded] of Object.entries(payload)) {
    const serializable = serialRegistry.get(key);
    if (serializable !== undefined) {
      serializable.hydrate(encoded);
    } else {
      pendingHydration.set(key, encoded);
    }
  }
};

/**
 * Create a writable reactive atom (`atom('count', 0)` — label first, like a
 * service Tag), plus `atom.fromStream(label, stream, initial)` to bridge an
 * Effect `Stream` into one (born {@link Volatile} — discharge with `defer` or
 * `.hot()`) and `atom.serializable(key, initial, schema)` for state that crosses
 * the wire (the key is the label).
 */
export const atom = Object.assign(atomImpl, {
  fromStream: atomFromStream,
  serializable: serializableAtom,
});

/**
 * A keyed collection of atoms — one lazily-created, memoised atom per key, so
 * per-entity state (a row, a cart line, a todo) is a family lookup rather than
 * one giant atom you slice. `family(key)` returns the same atom for equal keys;
 * `make` builds a fresh one the first time a key is seen. `forget`/`clear` drop
 * cached entries (e.g. when an entity is deleted). Non-primitive keys need a
 * `keyOf` to reduce them to a stable map key.
 *
 * Label first, like `atom(label, initial)`: the label names the *family*, and a
 * member that is itself an atom gets the runtime name `` `label(key)` `` — so
 * the devtools trace attributes a write to `quantities(sku-1)`, not `atom17`.
 *
 * ```ts
 * const quantities = atomFamily('quantities', (id: string) => atom(id, 1));
 * quantities('sku-1').update((n) => n + 1); // independent per id
 * ```
 */
export interface AtomFamily<K, A> {
  (key: K): A;
  /** Drop the cached atom for `key` (the next lookup makes a fresh one). */
  forget(key: K): void;
  /** Drop every cached atom. */
  clear(): void;
}

/**
 * Opt-in reclamation policy for an {@link atomFamily} keyed by unbounded dynamic
 * keys (a per-visited id/route), which would otherwise grow without bound.
 *
 * - `'unreferenced'` — drop an entry once its atom's last subscriber unsubscribes
 *   (a component reading it via `yield*`/`useAtomValue` counts). Requires the
 *   created value to be subscribable (an atom/derived); any other value is never
 *   reclaimed by this policy. Eviction is deferred a microtask so a StrictMode
 *   unsubscribe→resubscribe keeps the entry.
 * - `{ maxSize }` — a bounded LRU: keep at most `maxSize` entries, evicting the
 *   least-recently-accessed when the cap is exceeded.
 */
export type AtomFamilyGc = 'unreferenced' | { readonly maxSize: number };

/** Options for {@link atomFamily}. A bare `keyOf` function is still accepted. */
export interface AtomFamilyOptions<K> {
  /** Reduce a non-primitive key to a stable map key. */
  readonly keyOf?: (key: K) => unknown;
  /** Opt-in reclamation; absent means every entry is kept (strong — the default). */
  readonly gc?: AtomFamilyGc;
}

type Subscribable = { subscribe: (listener: () => void) => () => void };
const isSubscribable = (u: unknown): u is Subscribable =>
  typeof u === 'object' &&
  u !== null &&
  typeof (u as { subscribe?: unknown }).subscribe === 'function';

/**
 * Instrument an atom's `subscribe` in place to reference-count its subscribers,
 * calling `onZero` when the last one unsubscribes. Mutating the atom object (not
 * wrapping it) keeps every access path — direct `.subscribe`, `yield*`, `observe`
 * — pointed at the counting version.
 */
const trackUnreferenced = (value: unknown, onZero: () => void): void => {
  if (!isSubscribable(value)) {
    return;
  }
  const target = value;
  const original = target.subscribe.bind(target);
  let count = 0;
  target.subscribe = (listener) => {
    count += 1;
    const unsubscribe = original(listener);
    let done = false;
    return () => {
      if (done) {
        return;
      }
      done = true;
      count -= 1;
      unsubscribe();
      if (count === 0) {
        // Defer: a StrictMode unsubscribe→resubscribe in the same tick re-arms it.
        queueMicrotask(() => {
          if (count === 0) {
            onZero();
          }
        });
      }
    };
  };
};

export const atomFamily = <const L extends string, K, A>(
  label: L,
  make: (key: K) => A,
  optionsOrKeyOf?: AtomFamilyOptions<K> | ((key: K) => unknown),
): AtomFamily<K, A> => {
  const options: AtomFamilyOptions<K> =
    typeof optionsOrKeyOf === 'function' ? { keyOf: optionsOrKeyOf } : (optionsOrKeyOf ?? {});
  const keyOf = options.keyOf ?? ((key: K) => key);
  const gc = options.gc;
  const maxSize = typeof gc === 'object' ? gc.maxSize : undefined;
  const cache = new Map<unknown, A>();

  const family = ((key: K): A => {
    const id = keyOf(key);
    if (!cache.has(id)) {
      const value = make(key);
      // Name what we can: a member that is itself an atom is re-labeled
      // `label(key)` for the devtools trace; any other value keeps its own name
      // (the family label still names the collection for future introspection).
      if (isAtom(value)) {
        atomNames.set(value, `${label}(${String(key)})`);
      }
      cache.set(id, value);
      if (gc === 'unreferenced') {
        trackUnreferenced(value, () => {
          cache.delete(id);
        });
      }
      if (maxSize !== undefined) {
        // Just added at the end (most-recent); evict oldest beyond the cap.
        for (const oldest of cache.keys()) {
          if (cache.size <= maxSize) {
            break;
          }
          if (oldest !== id) {
            cache.delete(oldest);
          }
        }
      }
      return value;
    }
    const value = cache.get(id) as A;
    if (maxSize !== undefined) {
      // LRU touch: re-insert to move this id to the most-recent position.
      cache.delete(id);
      cache.set(id, value);
    }
    return value;
  }) as AtomFamily<K, A>;
  family.forget = (key) => {
    cache.delete(keyOf(key));
  };
  family.clear = () => {
    cache.clear();
  };
  return family;
};

/**
 * A standalone tracked computation: it evaluates a selector, remembering which
 * atoms it read, and — while anyone is subscribed — keeps a memoised value in
 * sync by re-running when a tracked atom changes. Read with no subscribers, it
 * computes fresh. The engine behind both {@link derive} and `observe`.
 */
export interface Computation<A> {
  get(): A;
  subscribe(listener: () => void): () => void;
}

export const computation = <A>(selector: (read: Read) => A, name?: string): Computation<A> => {
  const listeners = new Set<() => void>();
  let currentDeps: Array<ReadableAtom<unknown>> = [];
  let unsubscribes: Array<() => void> = [];
  let cached: A;
  let hasCached = false;
  let tracking = false;

  const evaluate = (): { value: A; deps: Array<ReadableAtom<unknown>> } => {
    const deps: Array<ReadableAtom<unknown>> = [];
    const read: Read = (atomRead) => {
      deps.push(atomRead as ReadableAtom<unknown>);
      return atomRead.value;
    };
    return { value: selector(read), deps };
  };

  const sameDeps = (deps: Array<ReadableAtom<unknown>>): boolean =>
    deps.length === currentDeps.length && deps.every((dep, i) => dep === currentDeps[i]);

  const listen = (deps: Array<ReadableAtom<unknown>>): void => {
    for (const unsubscribe of unsubscribes) {
      unsubscribe();
    }
    unsubscribes = deps.map((dep) => dep.subscribe(onChange));
    currentDeps = deps;
    // A named computation is a derived atom — report its source edges for the
    // devtools dependency graph (a no-op unless the inspector is connected).
    if (name !== undefined && devConnected()) {
      emitDev({ kind: 'atom:derive', target: name, sources: deps.map((dep) => atomNameOf(dep)) });
    }
  };

  function onChange(): void {
    const { value, deps } = evaluate();
    // Only re-subscribe when the dependency *set* actually changed. Re-subscribing
    // on every value change would mutate the notifying atom's listener set while
    // it iterates — re-entrant, and for a derived-of-derived chain, an infinite loop.
    if (!sameDeps(deps)) {
      listen(deps);
    }
    const changed = !Equal.equals(value, cached);
    cached = value;
    hasCached = true;
    if (changed) {
      // Snapshot the listeners: one may subscribe/unsubscribe while being notified.
      for (const listener of Array.from(listeners)) {
        listener();
      }
    }
  }

  return {
    // While tracked, `cached` is authoritative. Read with no subscribers, we
    // recompute — but return the *same reference* when the value is `Equal`-equal
    // to the last, so a caller polling `get()` (React's `getSnapshot`) sees an
    // `Object.is`-stable value and does not spin.
    //
    // INVARIANT (relied on by `derive.effect`, pinned by the regression test in
    // `#38`): while tracked, `get()` returns the *same array reference* across
    // renders that don't change a source. `derive.effect` keys its refetch on the
    // snapshot this returns; if a future change returned a fresh array per call,
    // every unrelated re-render would refetch. Keep the identity stable.
    get: () => {
      if (tracking) {
        return cached;
      }
      const next = evaluate().value;
      if (hasCached && Equal.equals(next, cached)) {
        return cached;
      }
      cached = next;
      hasCached = true;
      return cached;
    },
    subscribe: (listener) => {
      if (listeners.size === 0) {
        const first = evaluate();
        cached = first.value;
        hasCached = true;
        tracking = true;
        listen(first.deps);
      }
      listeners.add(listener);
      return () => {
        listeners.delete(listener);
        if (listeners.size === 0) {
          for (const unsubscribe of unsubscribes) {
            unsubscribe();
          }
          unsubscribes = [];
          currentDeps = [];
          tracking = false;
        }
      };
    },
  };
};

let deriveSeq = 0;

/**
 * The internal derived-atom constructor: a read-only atom over a tracked
 * selector. Both the explicit-sources `derive([a, b], f)` (whose selector reads
 * a *fixed* tuple, so the tracking settles into static subscriptions after the
 * first run) and the single-source `.derive(f)` method build on it. It reuses
 * {@link computation}, which also preserves the stable-reference-when-`Equal`
 * `get()` invariant `derive.effect` keys its refetch on (pinned by #38).
 */
const deriveReadonly = <A>(selector: (read: Read) => A): ReadableAtom<A> => {
  const label = `derive${deriveSeq++}`;
  const comp = computation(selector, label);
  const self: ReadableAtom<A> = {
    [AtomTypeId]: true,
    get value() {
      return comp.get();
    },
    subscribe: (listener) => comp.subscribe(listener),
    derive: (f) => deriveReadonly((read) => f(read(self))),
    [Symbol.iterator]: () => yieldSelf(self),
  };
  atomNames.set(self, label);
  return self;
};

/**
 * Any readable source a `derive` may draw from — value, volatility and label
 * erased. The tuple constraint below recovers all three per position.
 */
type AnySource = ReadableAtom<unknown, unknown, string>;

/** The value each source atom in the tuple holds, position by position. */
export type ValuesOfTuple<T extends ReadonlyArray<AnySource>> = {
  readonly [K in keyof T]: T[K] extends ReadableAtom<infer A, unknown, string> ? A : never;
};

/**
 * The union of the sources' volatility — {@link Volatile} if *any* source is
 * volatile, `never` once none is. A derive of a volatile atom stays volatile.
 */
export type VolatilityOfTuple<T extends ReadonlyArray<AnySource>> = {
  [K in keyof T]: T[K] extends ReadableAtom<unknown, infer V, string> ? V : never;
}[number];

/**
 * The union of the sources' label literals — a derived value re-renders its
 * readers exactly when one of its sources changes, so its `RendersOn` set is
 * the union of theirs.
 */
export type LabelsOfTuple<T extends ReadonlyArray<AnySource>> = {
  [K in keyof T]: T[K] extends ReadableAtom<unknown, unknown, infer L> ? L : never;
}[number];

/** Read every source in the tuple through `read`, subscribing to each. */
const readAll = <T extends ReadonlyArray<AnySource>>(sources: T, read: Read): ValuesOfTuple<T> =>
  // The erased-source view is sound: `read` only touches `value`/`subscribe`,
  // and `V`/`L` are phantom.
  sources.map((source) => read(source as ReadableAtom<unknown>)) as ValuesOfTuple<T>;

/**
 * Create a *derived* atom from **explicit sources**: a read-only reactive value
 * over a tuple of atoms, recomputing — and re-rendering its readers — precisely
 * when a source changes. The sources are explicit so the type channel sees them:
 * the value types are inferred position by position, and the result inherits the
 * *union* of the sources' labels and volatility ({@link LabelsOfTuple} /
 * {@link VolatilityOfTuple}) with zero annotation — a derive of a volatile atom
 * stays volatile, and `RendersOn` keeps naming the real sources. Derived atoms
 * compose. Keep derivation here, in the Effect world, not in a component.
 *
 * For the common case of deriving from a *single* atom, prefer the method form
 * {@link ReadableAtom.derive | `atom.derive`} — it hands you the value directly:
 * `const count = items.derive((list) => list.length)`.
 *
 * ```ts
 * const total = derive([price, qty], ([p, q]) => p * q); // ReadableAtom<number, never, 'price' | 'qty'>
 * ```
 *
 * {@link deriveWritable | `derive.writable`} adds a two-way variant;
 * {@link deriveEffect | `derive.effect`} an async, suspending one.
 */
const deriveImpl = <const T extends ReadonlyArray<AnySource>, B>(
  sources: T,
  f: (values: ValuesOfTuple<T>) => B,
): ReadableAtom<B, VolatilityOfTuple<T>, LabelsOfTuple<T>> =>
  // `V`/`L` are phantom — the runtime atom subscribes to exactly the tuple; the
  // inherited union is a pure type refinement over the wide default.
  deriveReadonly((read) => f(readAll(sources, read))) as unknown as ReadableAtom<
    B,
    VolatilityOfTuple<T>,
    LabelsOfTuple<T>
  >;

/**
 * A *writable* derived atom over **explicit sources**: read like a `derive`
 * (recomputes when a source changes; inherits the sources' labels and
 * volatility), but also `set`/`update`, with the write routed back through
 * `write` to the source atoms. For adapters and two-way projections — a form
 * field over a model, a unit conversion — where the value both derives from and
 * feeds the atoms beneath it.
 *
 * ```ts
 * const fahrenheit = derive.writable(
 *   [celsius],
 *   ([c]) => c * 1.8 + 32,
 *   (f) => celsius.set((f - 32) / 1.8),
 * );
 * ```
 */
const deriveWritable = <const T extends ReadonlyArray<AnySource>, B>(
  sources: T,
  f: (values: ValuesOfTuple<T>) => B,
  write: (value: B) => void,
): Atom<B, VolatilityOfTuple<T>, LabelsOfTuple<T>> => {
  const label = `derive${deriveSeq++}`;
  const comp = computation((read) => f(readAll(sources, read)), label);
  const self: Atom<B> = {
    [AtomTypeId]: true,
    get value() {
      return comp.get();
    },
    set: (value) => write(value),
    update: (update) => write(update(comp.get())),
    subscribe: (listener) => comp.subscribe(listener),
    derive: (g) => deriveReadonly((read) => g(read(self))),
    [Symbol.iterator]: () => yieldSelf(self),
  };
  atomNames.set(self, label);
  // `V`/`L` are phantom; see `deriveImpl`.
  return self as unknown as Atom<B, VolatilityOfTuple<T>, LabelsOfTuple<T>>;
};

/**
 * An *async derived* value over **explicit sources**: `f` receives the sources'
 * values and returns an `Effect`, and the result reads in a REC with `yield*`
 * like any other atom — but it *suspends* while the effect runs, re-runs when a
 * source atom changes (keyed by the source values, so an unchanged source is not
 * refetched), and contributes the same loading obligation `S` (plus the effect's
 * `E`/`R`) that {@link query} does. And like every derive, it inherits the
 * sources' labels and volatility. Async derivation, still expressed as data in
 * the Effect world:
 *
 * ```ts
 * const price = derive.effect([sku], ([s]) => fetchPrice(s)); // suspends; refetches on sku change
 * // in a REC: const p = yield* price;  // .suspense(...) must handle the loading
 * ```
 *
 * It owns no interpreter machinery of its own — yielding it drives, in order, a
 * subscription (so the component re-renders when a source changes) and a keyed
 * `query` (so it suspends and refetches). Both are ordinary yieldables, so the
 * `E`/`R`/`S` channels flow exactly as a hand-written `query` would — and the
 * `V`/`L` render channel flows through the subscription.
 */
export interface AsyncDerived<A, E, R, out V = never, out L extends string = string> {
  [Symbol.iterator](): Iterator<
    Suspensable<A, E, R> | ReadableAtom<ReadonlyArray<unknown>, V, L>,
    A
  >;
}

const deriveEffect = <const T extends ReadonlyArray<AnySource>, A, E, R>(
  sources: T,
  f: (values: ValuesOfTuple<T>) => Effect.Effect<A, E, R>,
): AsyncDerived<A, E, R, VolatilityOfTuple<T>, LabelsOfTuple<T>> => {
  // A stable tracker over exactly the source tuple: its value is the snapshot of
  // the sources' current values, so it recomputes (and, yielded, re-renders its
  // component) only on a real change — and that snapshot is the refetch key.
  // INVARIANT (pinned by #38): `key.value` returns the *same array reference*
  // across renders that don't change a source — `computation`'s `Equal`-guarded
  // `get()` guarantees it — so an unrelated re-render never refetches.
  const key = deriveReadonly<ReadonlyArray<unknown>>((read) =>
    readAll(sources, read),
  ) as unknown as ReadableAtom<ReadonlyArray<unknown>, VolatilityOfTuple<T>, LabelsOfTuple<T>>;
  return {
    [Symbol.iterator]() {
      let step: 0 | 1 | 2 = 0;
      return {
        next(
          sent?: unknown,
        ): IteratorResult<
          | Suspensable<A, E, R>
          | ReadableAtom<ReadonlyArray<unknown>, VolatilityOfTuple<T>, LabelsOfTuple<T>>,
          A
        > {
          if (step === 0) {
            step = 1;
            return { done: false, value: key }; // subscribe: re-render when a source changes
          }
          if (step === 1) {
            step = 2;
            // Suspend on the effect built from the snapshot just read back, keyed
            // by that same snapshot, so it refetches iff a source value changed.
            const snapshot = sent as ValuesOfTuple<T> & ReadonlyArray<unknown>;
            return {
              done: false,
              value: query(f(snapshot), snapshot),
            };
          }
          return { done: true, value: sent as A };
        },
      };
    },
  };
};

/**
 * Create a *derived* atom from *several* atoms — **explicit sources**:
 * `derive([a, b], ([x, y]) => x + y)`. The tuple is the subscription set *and*
 * the type-level fact: labels and volatility flow through with zero annotation.
 * `derive.writable(sources, f, write)` adds a two-way variant; `derive.effect(sources, f)`
 * an async, suspending one. Single source? Prefer the `a.derive(f)` method.
 */
export const derive = Object.assign(deriveImpl, {
  writable: deriveWritable,
  effect: deriveEffect,
});

/**
 * Deferred-read brands. `defer(a)` returns a *wrapper* atom (memoised per
 * source, so the wrapper — and therefore the hook order at a yield site — is
 * stable across renders); the React read side checks the brand and reads the
 * wrapper through `useDeferredValue`. A WeakSet, not a field: the wrapper stays
 * shaped exactly like any other atom.
 */
const deferredBrand = new WeakSet<object>();
const deferredWrappers = new WeakMap<object, ReadableAtom<unknown>>();

/**
 * Is this atom a {@link defer} wrapper (or a derive of one)? The React binding
 * (`react/reactivity.tsx`) uses it to route the read through `useDeferredValue`;
 * on the server a deferred read is a plain read, so the driver never asks.
 */
export const isDeferred = (u: unknown): boolean =>
  typeof u === 'object' && u !== null && deferredBrand.has(u);

/**
 * Discharge a volatile atom **at the read**: `defer(a)` strips the
 * {@link Volatile} obligation (the label stays) by lagging the read behind
 * React's urgent work — on the client the wrapper is read through
 * `useDeferredValue`, so a burst of emissions coalesces instead of storming the
 * tree; on the server it is a plain read (there are no re-renders to schedule).
 * The other discharge is `.hot()` on the REC — the deliberate acknowledgment
 * that a subtree is *meant* to render at source frequency; prefer `defer` when
 * the UI can lag a frame.
 *
 * ```ts
 * const price = yield* defer(ticker.price); // V discharged; label 'price' kept
 * ```
 *
 * Wrappers are memoised per source atom, so `defer(a)` inside a REC body is
 * stable across renders (same identity, same subscription, same hook order).
 * A `.derive` of a deferred atom is itself deferred — the lag survives
 * projection, which is what keeps its `V = never` honest.
 */
export const defer = <A, L extends string>(
  source: ReadableAtom<A, Volatile, L>,
): ReadableAtom<A, never, L> => {
  const cached = deferredWrappers.get(source);
  if (cached !== undefined) {
    return cached as ReadableAtom<A, never, L>;
  }
  const self: ReadableAtom<A> = {
    [AtomTypeId]: true,
    get value() {
      return source.value;
    },
    subscribe: (listener) => source.subscribe(listener),
    derive: (f) => {
      const derived = deriveReadonly((read) => f(read(self)));
      // The projection of a deferred read is read deferred too — otherwise the
      // derived atom would re-render at source frequency under a `V = never` type.
      deferredBrand.add(derived);
      return derived;
    },
    [Symbol.iterator]: () => yieldSelf(self),
  };
  deferredBrand.add(self);
  atomNames.set(self, atomNameOf(source));
  deferredWrappers.set(source, self);
  // `Volatile` in, `never` out is the point; `L` rides along. Both phantom.
  return self as ReadableAtom<A, never, L>;
};