/**
 * CellHandle - Represents a `Cell` in a runtime.
 */

import {
  FabricInstance,
  FabricPrimitive,
  type FabricValue,
  type FabricValuePlus,
  isFabricSpecialObject,
  valueEqual,
} from "@commonfabric/data-model";
import { FabricBytes } from "@commonfabric/data-model/fabric-primitives";
import { DID } from "@commonfabric/identity";
import { type CfcCellLinkRefPayload } from "@commonfabric/runner/cfc";
import type { NativeUiControl } from "@commonfabric/runner/native-ui";
import {
  cfcLabelViewsEqual,
  rebaseCfcLabelView,
} from "@commonfabric/runner/cfc/label-view-core";
import {
  type Cancel,
  isSigilLink,
  type JSONSchema,
  linkRefFrom,
  linkRefPayload,
  linkRefPayloadToString,
  refuseFabricInstance,
  type SigilLink,
} from "@commonfabric/runner/shared";
import { getLogger } from "@commonfabric/utils/logger";
import { isObjectNotArray, isObjectOrArray } from "@commonfabric/utils/types";

import { InitializedRuntimeConnection } from "./client/connection.ts";
import {
  type CellGetRequest,
  type CellPullRequest,
  type CellReadRefusal,
  type CellRef,
  type CellValueResponse,
  type CfcLabelView,
  JSONValue,
  RequestType,
  type SqliteParams,
} from "./protocol/mod.ts";
import { $conn, type RuntimeClient } from "./runtime-client.ts";
import { cellRefToIdentityKey } from "./shared/utils.ts";

// Logger for schema warnings - disabled by default.
// Enable via: globalThis.commonfabric.logger["cell-handle"].disabled = false
const logger = getLogger("cell-handle", { enabled: false });

/** A read's answer, and whether no update to the cell overtook it. */
interface ReadAnswer<T> {
  /** What the read found: the value, or the refusal standing for it. */
  read: CellAnswer<T>;
  /** The worker's raw answer, which other handles on the cell take. */
  response: CellValueResponse;
  /** Whether no update to the cell reached any handle on it meanwhile. */
  authoritative: boolean;
}

/**
 * The operations on one cell, in the order they reach the worker, and the
 * value the latest of them left. Handles that read the cell under different
 * schemas share one queue, so their writes stay in order, but not that value:
 * it is a reading under the schema of `view`, and a handle reading under
 * another takes it as no value.
 */
interface CellOperationQueue {
  tail: Promise<void> | undefined;
  hasValue: boolean;
  value: unknown;
  // `viewKey()` of the handle whose operation left `value`.
  view: string | undefined;
  // Changes when any equivalent handle receives an authoritative update.
  authoritativeGeneration: number;
}

const operationQueues = new WeakMap<
  RuntimeClient,
  Map<string, CellOperationQueue>
>();

/**
 * The clock every handle's write generation is drawn from, so that a read one
 * handle made can tell whether another has mutated its value since.
 */
let writeClock = 0;

export const $onCellUpdate = Symbol("$onCellUpdate");

/**
 * Tells a handle the worker delivered an update of its cell that the
 * connection dropped: a value of `undefined`, which the handle cannot tell
 * from a document not yet loaded, so it keeps the value it holds.
 */
export const $onCellUpdateDropped = Symbol("$onCellUpdateDropped");

/**
 * Delivers to a handle a subscription update the worker refused, which
 * carries the refusal and nothing of the cell.
 */
export const $onCellRefused = Symbol("$onCellRefused");

/**
 * What a handle holds of its cell: nothing yet (`unread`), before the worker
 * has answered a read of it and before anything was written through it; the
 * value its last read returned, or its last write wrote, which is
 * `undefined` for a cell that holds nothing; or the refusal that stands in
 * its place when the worker refused that read.
 *
 * Unread is not empty: a value computed from a handle that has read nothing,
 * such as a count incremented from it, would replace one nobody was shown.
 */
export type CellHandleRead<T> =
  | { readonly unread: true }
  | CellAnswer<T>;

/** What the worker's answer to a read holds: the value, or the refusal. */
type CellAnswer<T> =
  | { readonly value: T | undefined }
  | { readonly refused: CellReadRefusal };

/** What a subscriber is told about the delivery its value callback is for. */
export type CellDelivery = {
  /**
   * Whether the `cfcLabel` handed to the callback is the label the worker
   * read with the value: it is for a delivery the worker made with its label,
   * and is not for a value-only delivery or a local write, which carry the
   * label the handle last received, nor once the handle has left the cell's
   * subscription and joined it again, until the worker delivers a label. Nor
   * is it once the worker has sent an update the connection dropped, one
   * holding nothing, since the handle then holds a value and label the worker
   * has read past; a handle delivering labels tells its subscribers so with
   * what it holds, and tells them again when the worker's next labeled update
   * makes it current, whatever that update holds.
   */
  readonly cfcLabelCurrent: boolean;
};

/** How a subscriber to a handle hears what the handle holds of its cell. */
export type CellSubscribeOptions = {
  /** Whether to hear the cell's display label with each value. */
  includeCfcLabel?: boolean;

  /**
   * Hears each refusal of the handle's read, which the value callback does
   * not: a refusal carries nothing of the cell, so it is never handed to the
   * value callback as a value. It runs at once when the handle's last read
   * was refused, as the value callback does for a value.
   *
   * Every subscriber gives one, so that each says what it shows while its
   * cell is withheld. One that heard nothing would go on showing the last
   * value it was given, which the worker may now refuse.
   */
  onRefused: (refusal: CellReadRefusal) => void;
};

/** How a strict send of an event waits on it. */
export type CellSendOptions = {
  /**
   * Whether to wait for the event's handling as well: the outcome of the run
   * of the stream's handler, which under server execution is the consequence
   * the served run recorded. The send then rejects, with the reason, when
   * that run's commit is refused, as it is for a write whose UI contract the
   * event does not satisfy, and when the run throws or the event is dropped
   * or refused admission. A run that finishes without an error resolves it,
   * whether or not it wrote anything.
   */
  awaitHandling?: boolean;
};

/**
 * Thrown by a read of a handle whose read the worker refused, and the
 * rejection of `sync()`, `pull()` and `initialize()` when the worker refuses
 * the read they make: a read the host could not make never returns as a cell
 * that holds nothing.
 */
export class CellReadRefusedError extends Error {
  readonly #refusal: CellReadRefusal;

  constructor(refusal: CellReadRefusal) {
    super(`The worker refused this read of a cell (${refusal.refusedBy}).`);
    this.name = "CellReadRefusedError";
    this.#refusal = refusal;
  }

  /** What refused the read. */
  get refusal(): CellReadRefusal {
    return this.#refusal;
  }
}

/**
 * A cell's value as the _client_ holds it: what a cell holds, with a
 * `CellHandle` wherever a cell sits. That is the substitution `vnode-types.ts`
 * already makes for the render types -- `Cell` replaced by `CellHandle` --
 * applied to a cell's own value.
 *
 * What a cell holds is a `FabricValue`, so this is `FabricValuePlus` at
 * `CellHandle` rather than something restated: a `FabricBytes` in a cell is a
 * value the client holds like any other, and this stays true of whatever
 * `FabricValue` comes to admit, while a handle may sit at the top or inside
 * any container.
 *
 * The connection's encoding carries this whole domain. The conversion walk
 * that feeds it is narrower: `CellHandle.serialize()` refuses a
 * `FabricInstance`, being a container it cannot descend to find a handle
 * inside. So a value admitted here can still be refused on the way out, and
 * the refusal names which.
 */
export type ClientCellValue = FabricValuePlus<CellHandle<unknown>>;

/**
 * CellHandle provides a cell interface for cells living in a web worker.
 */
export class CellHandle<T = unknown> {
  #rt: RuntimeClient;
  #conn: InitializedRuntimeConnection;
  #ref: CellRef;
  #value: T | undefined;
  #cfcLabel: CfcLabelView | undefined;

  /**
   * Whether `#value` and `#cfcLabel` are the worker's latest read of the
   * cell, the label being the one it read with that value: set by a delivery
   * that carried a label, and cleared by a value that arrived without one, by
   * a local write, by a refusal, by an update the connection dropped, and
   * when the handle leaves the cell's subscription, as its last subscriber
   * leaving and an upgrade to labels do, since it then stops hearing of the
   * label.
   */
  #cfcLabelCurrent = false;

  /** The refusal of this handle's last read, while it stands. */
  #refusal: CellReadRefusal | undefined;

  /**
   * Whether the handle holds nothing yet: no read of it has been answered,
   * and nothing was written through it.
   */
  #unread = true;

  /**
   * Whether any subscriber asked for the CFC label. Sticky: once a label-aware
   * subscription exists on this handle, label changes, and changes in whether
   * the label is current, also fire its callbacks.
   */
  #wantsCfcLabel = false;

  #callbacks = new Map<
    number,
    {
      onValue: (
        value: Readonly<T>,
        cfcLabel: CfcLabelView | undefined,
        delivery: CellDelivery,
      ) => void;
      /** Runs the cleanup the subscriber's last value callback returned. */
      clear: () => void;
      onRefused: (refusal: CellReadRefusal) => void;
    }
  >();
  #nextCallbackId = 0;
  #schemaWarned = false;
  // `#viewKey()`, memoized.
  #view: string | undefined;
  #updateGeneration = 0;

  /**
   * Monotonic invocation order for local value mutations on this handle,
   * drawn from `writeClock`. Async read and strict-write responses use it to
   * avoid replacing a later optimistic value while their remote operations
   * remain FIFO-serialized, as does a read of another handle on this cell.
   */
  #writeGeneration = 0;

  /**
   * A handle on `cellRef`, holding `initial`, what a read of it already
   * found, or nothing yet when no read has been made.
   */
  constructor(
    worker: RuntimeClient,
    cellRef: CellRef,
    initial: CellHandleRead<T> = { unread: true },
  ) {
    this.#rt = worker;
    this.#conn = worker[$conn]();
    this.#ref = cellRef;
    this.#hold(initial);
  }

  /** Takes `read` as what the handle holds, telling no subscriber. */
  #hold(read: CellHandleRead<T>): void {
    this.#unread = "unread" in read;
    this.#refusal = "refused" in read ? read.refused : undefined;
    this.#value = "value" in read ? read.value : undefined;
  }

  /**
   * Check if this cell has a schema defined. Warns if no schema is set.
   * Warning is disabled by default; enable via:
   * globalThis.commonfabric.logger["cell-handle"].disabled = false
   */
  #requireSchema(method: string): void {
    if (!this.#ref.schema && !this.#ref.keyedFrom && !this.#schemaWarned) {
      this.#schemaWarned = true;
      // Use callback for lazy evaluation - stack trace only generated if logging is enabled
      logger.warn(`no-schema-${method}`, () => {
        const stack = new Error().stack;
        return [
          `${method}() called without schema on cell ${this.#ref.id}:${
            this.#ref.path.join(".")
          }. ` +
          `Please bind a schema using asSchema() or pass a schema to the cell controller's bind() method.\n` +
          `Stack trace:\n${stack}`,
        ];
      });
    }
  }

  runtime(): RuntimeClient {
    return this.#rt;
  }

  ref(): CellRef {
    return this.#ref;
  }

  space(): DID {
    return this.#ref.space;
  }

  /**
   * The FULL schemed id — identical to `ref().id` and safe to use as
   * identity (keys, equality, round-trips). The hash preimage is kind-free,
   * so the URI scheme is the only thing distinguishing a computed doc from
   * a state sibling of the same cause; this accessor never strips it. For
   * the piece-root routing/display form (bare, `of:`-stripped), use
   * `PieceHandle.id()` — see docs/specs/computed-cell-identity.md.
   */
  id(): string {
    return this.#ref.id;
  }

  /**
   * The refusal of this handle's last read, while it stands: the handle then
   * holds nothing of its cell, and {@link get} throws. `undefined` while the
   * handle holds a value, holds nothing because its cell holds nothing, or
   * has not been read.
   */
  get refusal(): CellReadRefusal | undefined {
    return this.#refusal;
  }

  /**
   * What the handle holds of its cell: nothing yet (`{ unread: true }`),
   * before the worker has answered a read of it and before anything was
   * written through it; the value (`{ value }`), which is `undefined` for a
   * cell that holds nothing; or the refusal that stands in its place
   * (`{ refused }`). Unread is not `{ value: undefined }`: a value computed
   * from it would be computed from nothing the worker said. Unlike
   * {@link get}, it does not throw for a refusal, so a caller that shows the
   * cell can show its placeholder instead.
   */
  lastRead(): CellHandleRead<Readonly<T>> {
    if (this.#refusal !== undefined) return { refused: this.#refusal };
    return this.#unread ? { unread: true } : { value: this.#value };
  }

  /**
   * Get the current cached value.
   *
   * @throws {CellReadRefusedError} While the worker's refusal of the
   *   handle's last read stands, since the handle then holds nothing of its
   *   cell, which is not the same as a cell that holds nothing.
   */
  get(): Readonly<T> | undefined {
    this.#requireSchema("get");
    if (this.#refusal !== undefined) {
      throw new CellReadRefusedError(this.#refusal);
    }
    return this.#value !== undefined ? this.#value as Readonly<T> : undefined;
  }

  /**
   * Set the cell's value locally, as well as in the runtime. A failure is
   * logged rather than returned, the refusal of a write through a handle
   * whose read is refused ({@link refusal}) among them.
   */
  async set(value: T): Promise<void> {
    this.#requireSchema("set");
    const refused = this.#writeRefusal();
    if (refused !== undefined) {
      console.error("[CellHandle] Write failed:", refused);
      return;
    }
    // A plain set is a blind last-write-wins overwrite (CellSet).
    const serialized = this.#serializeWrite(value);
    const snapshot = applyValue(
      CellHandle.#serialize(value as ClientCellValue, "sigil"),
      value,
      this,
    ) as T;
    this.#writeGeneration = ++writeClock;
    this.#publishWrite(snapshot);
    await this.#enqueueOperation(async (queue) => {
      // A refusal that arrived while the write waited its turn stands.
      const refusedSince = this.#writeRefusal();
      if (refusedSince !== undefined) {
        console.error("[CellHandle] Write failed:", refusedSince);
        return;
      }
      const updateGeneration = this.#updateGeneration;
      const authoritativeGeneration = queue.authoritativeGeneration;
      await this.#sendWrite(serialized);
      if (
        updateGeneration === this.#updateGeneration &&
        authoritativeGeneration === queue.authoritativeGeneration
      ) {
        this.#holdInQueue(queue, snapshot);
      }
    });
  }

  /**
   * Writes the value `updater` makes of the cell's current one, in the order
   * of the cell's operations through this runtime client: `updater` runs
   * once the reads and writes asked for before this call, through any handle
   * on the same cell and path, have run, and the ones asked for after it
   * wait for its write. A handle reached through `key()` names another path,
   * with an order of its own.
   *
   * `updater` is handed the value those operations left, what this handle
   * holds where they left none, or, where it holds nothing yet, what a read
   * from the worker finds, `undefined` for a cell that holds nothing. A write
   * another handle made, and that has settled since, counts once its update
   * reaches this handle. `updater` runs before `update()` returns when
   * nothing waits and the handle holds a value. Returning the current value
   * writes nothing.
   *
   * What it returns is shown when it runs, unless a write asked for through
   * this handle after this call shows its own, and is sent as {@link set}
   * sends a value: a blind last-write-wins overwrite, whose refusal by the
   * runtime is logged. It is not an atomic read-modify-write: a writer
   * elsewhere can change the cell between the read and the write, and the
   * write replaces that change. Appending to a list is {@link push}, which
   * the worker merges.
   *
   * Not the pattern API's `Cell.update()`, which merges a partial value.
   *
   * Settles once the worker has taken the write, or once `updater` writes
   * nothing. Rejects, writing nothing, while this handle's read is refused
   * ({@link refusal}), when the read it makes is refused, and when `updater`
   * throws.
   */
  update(updater: (current: Readonly<T> | undefined) => T): Promise<void> {
    this.#requireSchema("update");
    const refused = this.#writeRefusal();
    if (refused !== undefined) return Promise.reject(refused);
    // Taken now, as a write's, so that neither an older read's answer nor this
    // update's own shows over a write asked for after this call.
    const writeGeneration = this.#writeGeneration = ++writeClock;
    const startedAt = writeClock;
    return this.#enqueueOperation(async (queue) => {
      let current = this.#base(queue, writeGeneration);
      if (current === undefined) {
        const request: CellPullRequest = {
          type: RequestType.CellPull,
          cell: this.ref(),
          awaitDurability: false,
        };
        const updateGeneration = this.#updateGeneration;
        const answer = await this.#ask(queue, request, updateGeneration);
        const found = this.#takeAnswer(
          request,
          answer,
          writeGeneration,
          updateGeneration,
          startedAt,
        );
        // A refusal or an update that reached the cell while the read was
        // on its way is newer than what the read found.
        current = this.#base(queue, writeGeneration) ?? { value: found };
      }
      let next: T;
      try {
        next = updater(current.value as Readonly<T> | undefined);
      } catch (error) {
        this.#showPassedOver(current.value, writeGeneration);
        throw error;
      }
      if (valuesOrCellsEqual(next, current.value)) {
        this.#showPassedOver(current.value, writeGeneration);
        return;
      }
      const serialized = this.#serializeWrite(next);
      const snapshot = applyValue(
        CellHandle.#serialize(next as ClientCellValue, "sigil"),
        next,
        this,
      ) as T;
      if (writeGeneration === this.#writeGeneration) {
        this.#publishWrite(snapshot);
      }
      const updateGeneration = this.#updateGeneration;
      const authoritativeGeneration = queue.authoritativeGeneration;
      await this.#sendWrite(serialized);
      if (
        updateGeneration === this.#updateGeneration &&
        authoritativeGeneration === queue.authoritativeGeneration
      ) {
        this.#holdInQueue(queue, snapshot);
      }
    });
  }

  /**
   * Helper for {@link update}, which finds the value an update in `queue`'s
   * turn computes from without asking the worker: the one the operations
   * before it left, or what this handle holds, where no write asked for
   * since `writeGeneration` replaced it. `undefined` where neither holds one,
   * and the worker must be asked.
   *
   * @throws {CellReadRefusedError} While this handle's read is refused.
   */
  #base(
    queue: CellOperationQueue,
    writeGeneration: number,
  ): { value: unknown } | undefined {
    const refused = this.#writeRefusal();
    if (refused !== undefined) throw refused;
    const queued = this.#queuedValue(queue);
    if (queued !== undefined) return queued;
    if (!this.#unread && writeGeneration === this.#writeGeneration) {
      return { value: this.#value };
    }
    return undefined;
  }

  /**
   * Helper for {@link update}, which shows `value`, what an update that
   * writes nothing computed from, where it is the latest this handle has
   * been asked for: taking the generation, the update kept an older read's
   * answer from showing it.
   */
  #showPassedOver(value: unknown, writeGeneration: number): void {
    if (writeGeneration !== this.#writeGeneration) return;
    if (this.#unread || !valuesOrCellsEqual(value, this.#value)) {
      this.#publishWrite(value as T);
    }
  }

  /**
   * Set the cell's value and reject when the runtime refuses the write, or
   * when this handle's read is refused ({@link refusal}).
   */
  async setStrict(value: T): Promise<void> {
    this.#requireSchema("setStrict");
    const refused = this.#writeRefusal();
    if (refused !== undefined) throw refused;
    const serialized = this.#serializeWrite(value);
    const snapshot = applyValue(
      CellHandle.#serialize(value as ClientCellValue, "sigil"),
      value,
      this,
    ) as T;
    const writeGeneration = this.#writeGeneration;
    await this.#enqueueOperation(async (queue) => {
      // A refusal that arrived while the write waited its turn stands.
      const refusedSince = this.#writeRefusal();
      if (refusedSince !== undefined) throw refusedSince;
      const updateGeneration = this.#updateGeneration;
      const authoritativeGeneration = queue.authoritativeGeneration;
      await this.#sendStrictWrite(
        snapshot,
        serialized,
        writeGeneration,
        queue,
        authoritativeGeneration,
      );
      // Publication follows handle-local and identity-wide generations, but
      // this shared queue follows committed operation order. A later queued
      // append must start from this replacement even when a still-later
      // mutation suppresses its direct publication on this handle.
      if (
        updateGeneration === this.#updateGeneration &&
        authoritativeGeneration === queue.authoritativeGeneration
      ) {
        this.#holdInQueue(queue, snapshot);
      }
    });
  }

  /**
   * Atomically stores `value` only if the cell has no backing value, then
   * returns the value selected by that transaction. A readable schema fallback
   * does not count as stored. An existing value can return after producer and
   * load readiness when its read dependencies have no pending local writes.
   * Otherwise demand remains active through the runtime-wide commit barrier
   * before choosing a value or storing the default. Concurrent initializers
   * converge on one winner instead of replacing it with a blind write.
   *
   * It goes through a handle whose read is refused: it stores nothing over a
   * value the cell holds, and `value` is the caller's default, computed from
   * nothing the handle read. Its answer is a read of what is stored, which the
   * worker decides, so only an admitted one ends a refusal.
   *
   * @throws {CellReadRefusedError} When the worker refuses the read of the
   *   value that storage already held, which then stands as the handle's
   *   refusal.
   */
  async initialize(value: T): Promise<Readonly<T>> {
    this.#requireSchema("initialize");
    if (value === undefined) {
      throw new TypeError("Cell initialize requires a defined value.");
    }
    const serialized = this.#serializeWrite(value);
    const writeGeneration = this.#writeGeneration = ++writeClock;
    const updateGeneration = this.#updateGeneration;
    const { read, authoritative } = await this.#enqueueOperation(
      async (queue) => {
        const authoritativeGeneration = queue.authoritativeGeneration;
        const response = await this.#conn.request<RequestType.CellInitialize>({
          type: RequestType.CellInitialize,
          cell: this.ref(),
          value: serialized,
        });
        const read = this.#readOf(response);
        const authoritative = updateGeneration === this.#updateGeneration &&
          authoritativeGeneration === queue.authoritativeGeneration;
        if (authoritative) this.#recordInQueue(queue, read);
        return { read, authoritative };
      },
    );
    const latest = writeGeneration === this.#writeGeneration &&
      updateGeneration === this.#updateGeneration &&
      authoritative;
    if ("refused" in read) {
      if (latest) this.#refuse(read.refused);
      throw new CellReadRefusedError(read.refused);
    }
    // What an initialization stores, and what it finds stored, are both
    // defined.
    const current = read.value as T;
    if (latest) this.#publishValue(current);
    return current as Readonly<T>;
  }

  /**
   * The error a write through this handle fails with while its read is
   * refused, or `undefined` when none stands. Such a handle holds nothing of
   * its cell, so a value written through it was made from nothing it could
   * see: a toggle of a value it read as absent, a list with one item removed
   * from a list it read as empty. Writing it would replace a value the host
   * was never shown. A handle reached through `key()` makes reads of its own,
   * and is not held to its parent's refusal.
   */
  #writeRefusal(): CellReadRefusedError | undefined {
    return this.#refusal === undefined
      ? undefined
      : new CellReadRefusedError(this.#refusal);
  }

  #enqueueOperation<R>(
    operation: (queue: CellOperationQueue) => Promise<R>,
  ): Promise<R> {
    let queues = operationQueues.get(this.#rt);
    if (!queues) {
      queues = new Map();
      operationQueues.set(this.#rt, queues);
    }
    const key = cellRefToIdentityKey(this.#ref);
    let queue = queues.get(key);
    if (!queue) {
      queue = {
        tail: undefined,
        hasValue: false,
        value: undefined,
        view: undefined,
        authoritativeGeneration: 0,
      };
      queues.set(key, queue);
    }
    const previous = queue.tail;
    // With nothing ahead of it, the operation starts at once, but only after
    // it holds the queue: an operation asked for while it runs, as a
    // subscriber its publication reaches may ask for one, waits behind it.
    let result: Promise<R>;
    let started: PromiseWithResolvers<R> | undefined;
    if (previous) {
      result = previous.then(() => operation(queue));
    } else {
      started = Promise.withResolvers<R>();
      result = started.promise;
    }
    const tail = result.then(() => {}, () => {});
    queue.tail = tail;
    void tail.then(() => {
      if (queue.tail === tail) {
        queues.delete(key);
        if (queues.size === 0) operationQueues.delete(this.#rt);
      }
    });
    if (started) {
      try {
        started.resolve(operation(queue));
      } catch (error) {
        // Thrown at once, as before the queue was held; the queue goes on.
        started.reject(error);
        throw error;
      }
    }
    return result;
  }

  #serializeWrite(value: T): FabricValue {
    // Serialize before an optimistic local publication, because this can
    // refuse. A value the runtime will never see must not become this handle's
    // cached value or reach its subscribers.
    //
    // `T` is unconstrained, so this says what the write path requires rather
    // than what the class guarantees. Constraining `T` to `ClientCellValue` is
    // the honest fix and is not a small one: the schema-derived types
    // (`ObjectFromProperties<...>`) and the looser `Props` of `vnode-types.ts`
    // do not satisfy it, an interface having no implicit index signature where
    // an identical type alias does.
    //
    // TODO(danfuzz): constrain `T`, once those types are assignable.
    return CellHandle.serialize(value as ClientCellValue);
  }

  #sendWrite(
    serialized: FabricValue,
  ): Promise<void> {
    const cell = this.ref();
    const request = this.#conn.request<RequestType.CellSet>({
      type: RequestType.CellSet,
      cell,
      value: serialized,
    });
    return request.catch((error) => {
      if (!this.#conn.signal.aborted) {
        console.error("[CellHandle] Write failed:", error);
      }
    });
  }

  #sendStrictWrite(
    value: T,
    serialized: FabricValue,
    writeGeneration: number,
    queue: CellOperationQueue,
    authoritativeGeneration: number,
  ): Promise<boolean> {
    const before = this.#value;
    const updateGeneration = this.#updateGeneration;
    return this.#conn.request<RequestType.CellSet>({
      type: RequestType.CellSet,
      cell: this.ref(),
      value: serialized,
      awaitCommit: true,
    }).then(() => {
      const publish = writeGeneration === this.#writeGeneration &&
        updateGeneration === this.#updateGeneration &&
        authoritativeGeneration === queue.authoritativeGeneration &&
        this.#value === before;
      if (publish) this.#publishWrite(value);
      return publish;
    });
  }

  /**
   * Publishes `value`, which a write computed, unless a refusal stands: a
   * refused handle shows nothing of its cell, and a write, which may have
   * been computed from what the worker now refuses, never ends a refusal.
   * Only an admitted read does ({@link #publishValue}).
   */
  #publishWrite(value: T): void {
    if (this.#refusal !== undefined) return;
    this.#publishValue(value);
  }

  /**
   * Publishes `value` to every subscriber. An admitted read publishes what it
   * found through here, which ends a refusal; a write goes through
   * {@link #publishWrite}.
   */
  #publishValue(value: T): void {
    this.#value = value;
    this.#refusal = undefined;
    this.#unread = false;
    // A local update carries the label the handle last received, which the
    // worker did not read with this value.
    this.#cfcLabelCurrent = false;
    for (const { onValue } of this.#callbacks.values()) {
      try {
        onValue(value as Readonly<T>, this.#cfcLabel, {
          cfcLabelCurrent: false,
        });
      } catch (error) {
        console.error("[CellHandle] Callback error:", error);
      }
    }
  }

  /**
   * Drops what the handle holds of its cell for `refusal`, and tells each
   * subscriber that listens for refusals.
   */
  #refuse(refusal: CellReadRefusal): void {
    this.#value = undefined;
    this.#cfcLabel = undefined;
    this.#cfcLabelCurrent = false;
    this.#refusal = refusal;
    this.#unread = false;
    for (const { clear, onRefused } of this.#callbacks.values()) {
      // What a subscriber set up for the value it was shown goes with it.
      clear();
      onRefused(refusal);
    }
  }

  /**
   * Sends `event` to the stream this handle names. It goes through a handle
   * whose read is refused: an event is not computed from what the handle
   * read, and the worker decides what the stream's handler may do with it.
   */
  async send(event: T): Promise<void> {
    const serialized = CellHandle.serialize(event as ClientCellValue);
    await this.#enqueueOperation(() => this.#send(serialized));
  }

  /**
   * Sends `event` to the stream this handle names, and rejects when the
   * runtime refuses to take it. Resolving confirms only that the event was
   * delivered to the runtime, on its way to the stream's handler: a write the
   * handler makes that the runtime refuses, such as one whose UI contract the
   * event does not satisfy, does not reject it. `options.awaitHandling` waits
   * for the handling too, and rejects on its refusal.
   */
  async sendStrict(event: T, options?: CellSendOptions): Promise<void> {
    const serialized = CellHandle.serialize(event as ClientCellValue);
    await this.#enqueueOperation(() =>
      this.#send(serialized, true, options?.awaitHandling === true)
    );
  }

  /**
   * Sends `event` to the stream this handle names as the action `control`
   * names, taken on a control the host draws itself, bound to that trusted
   * surface and action. The runtime stamps the event with `native`
   * provenance for them, replacing any `provenance` field `event` carries,
   * and marks it renderer-trusted, so that it satisfies the UI contract of a
   * write gated on that surface and action, such as a `TrustedActionWrite`,
   * as a reviewed gesture on the pattern's rendered surface does. It is a
   * trusted gesture, as that gesture is, so it meets what `grantSpaceAccess()`
   * and `revokeSpaceAccess()` require of a handler's event. The write, and any
   * change to an access list, is held to every other check.
   *
   * It resolves once the run of the stream's handler has committed, which
   * under server execution is the consequence the served run recorded, and
   * rejects with the reason when the runtime refuses the event, when that
   * run's commit is refused, as it is for a write whose UI contract the event
   * does not satisfy, and when the run throws or the event is dropped or
   * refused admission. A run that finishes without an error resolves it,
   * whether or not it wrote anything; a handler that turns an event down
   * throws, which fails its run and rejects the send. The surface and action are read once, here.
   *
   * This mints trusted events. The host calls it only from the control's
   * real user-input path, with exactly the values the control displayed and
   * the surface and action from the control's own definition, never from an
   * event or a payload; and it keeps it, and the connection carrying it, out
   * of reach of pattern code, loaded content, automation and agent
   * interfaces, URL handlers, and restored state.
   *
   * @throws If `event` is not a record, or `control` names a blank surface or
   *   action.
   */
  async sendReviewed(event: T, control: NativeUiControl): Promise<void> {
    const serialized = CellHandle.serialize(event as ClientCellValue);
    const { surface, action } = control;
    await this.#enqueueOperation(() =>
      this.#conn.request<RequestType.CellSendReviewed>({
        type: RequestType.CellSendReviewed,
        cell: this.ref(),
        event: serialized,
        surface,
        action,
      })
    );
  }

  #send(
    event: FabricValue,
    propagateFailure = false,
    awaitHandling = false,
  ): Promise<void> {
    const request = this.#conn.request<RequestType.CellSend>({
      type: RequestType.CellSend,
      cell: this.ref(),
      event,
      ...(propagateFailure && { awaitCommit: true }),
      ...(awaitHandling && { awaitHandling: true }),
    });
    if (propagateFailure) return request;
    return request.catch((error) => {
      if (!this.#conn.signal.aborted) {
        console.error("[CellHandle] Send failed:", error);
      }
    });
  }

  /**
   * Get a child cell at the specified key.
   * Returns a new CellHandle with an extended path.
   *
   * The child starts with the part of this handle's value at the key, which
   * an admitted read of this cell showed. It does not start in this handle's
   * refusal: the child names a read of its own, which the worker decides on
   * its own, and may admit where it refuses the whole, so the child is
   * unread until that read answers, as it is when this handle is.
   *
   * The child is read under the schema the worker narrows this handle's
   * schema to at the key, as the runtime's own `Cell.key()` is, and under
   * none when neither this handle nor an ancestor it was keyed from has one.
   * Its `ref().schema` stays unset: the narrowing happens in the worker.
   */
  key<K extends keyof T>(valueKey: K): CellHandle<T[K]> {
    const childRef = this.#extendPath(String(valueKey));
    const child = new CellHandle<T[K]>(this.#rt, childRef);
    if (this.#unread || this.#refusal !== undefined) return child;
    // What this handle holds shows what the child holds.
    child.#hold({
      value: this.#value == null
        ? undefined
        : (this.#value as Record<string, unknown>)[String(valueKey)] as T[K],
    });
    return child;
  }

  /**
   * Appends `values` to the array this handle holds. A push publishes an
   * invocation-time local snapshot, but sends only the appended members. The
   * runtime owns the authoritative mergeable append, so an equivalent handle's
   * stale private cache cannot replace committed data.
   */
  push<U>(
    this: CellHandle<U[]>,
    ...values: T extends (infer U)[] ? U[] : never
  ): void {
    this.#pushValuesReportingFailure(values);
  }

  /**
   * Like `push()`, except that it takes the values to append as one list, so
   * the list can be of any length.
   */
  pushAll<U>(
    this: CellHandle<U[]>,
    values: T extends (infer U)[] ? readonly U[] : never,
  ): void {
    this.#pushValuesReportingFailure(values);
  }

  /** Append values and reject when the runtime refuses the mergeable write. */
  pushStrict<U>(
    this: CellHandle<U[]>,
    ...values: T extends (infer U)[] ? U[] : never
  ): Promise<void> {
    return this.#pushValues(values);
  }

  /**
   * Like `pushStrict()`, except that it takes the values to append as one
   * list, so the list can be of any length.
   */
  pushAllStrict<U>(
    this: CellHandle<U[]>,
    values: T extends (infer U)[] ? readonly U[] : never,
  ): Promise<void> {
    return this.#pushValues(values);
  }

  /**
   * Helper for `push()` and `pushAll()`, which appends `values` and logs a
   * refusal instead of returning it, unless the connection has been aborted.
   */
  #pushValuesReportingFailure(values: readonly unknown[]): void {
    void this.#pushValues(values).catch((error) => {
      if (!this.#conn.signal.aborted) {
        console.error("[CellHandle] Push failed:", error);
      }
    });
  }

  #pushValues(values: readonly unknown[]): Promise<void> {
    const refused = this.#writeRefusal();
    if (refused !== undefined) return Promise.reject(refused);
    const serializedValues = values.map((value) =>
      CellHandle.serialize(value as ClientCellValue)
    );
    const snapshots = values.map((value) => {
      const snapshot = CellHandle.#serialize(
        value as ClientCellValue,
        "sigil",
      );
      return applyValue(snapshot, value, this);
    });
    const cached = this.#value;
    const fallback = cached === undefined ? undefined : applyValue(
      CellHandle.#serialize(cached as ClientCellValue, "sigil"),
      cached,
      this,
    );
    const writeGeneration = this.#writeGeneration = ++writeClock;
    const append = (queue: CellOperationQueue): Promise<void> => {
      // A refusal that arrived while the append waited its turn stands, and
      // what it would append to, the value from before it, is refused.
      const refusedSince = this.#writeRefusal();
      if (refusedSince !== undefined) return Promise.reject(refusedSince);
      const authoritativeGeneration = queue.authoritativeGeneration;
      const queued = this.#queuedValue(queue);
      const current = (queued ? queued.value : fallback) as unknown[];
      if (!Array.isArray(current)) {
        throw new Error("push() can only be used on array cells");
      }
      const value = [...current, ...snapshots];
      const updateGeneration = this.#updateGeneration;
      this.#holdInQueue(queue, value);
      if (writeGeneration === this.#writeGeneration) {
        this.#publishWrite(value as unknown as T);
      }
      const rollback = () => {
        if (
          updateGeneration === this.#updateGeneration &&
          authoritativeGeneration === queue.authoritativeGeneration
        ) {
          // Keep the restored base explicit in the shared queue. A later
          // append may have captured this append's optimistic publication as
          // its private fallback before the refusal arrived.
          this.#holdInQueue(queue, current);
          if (writeGeneration === this.#writeGeneration) {
            this.#publishWrite(current as unknown as T);
          }
        }
      };
      let request: Promise<unknown>;
      try {
        request = this.#conn.request<RequestType.CellPush>({
          type: RequestType.CellPush,
          cell: this.ref(),
          values: serializedValues,
          awaitCommit: true,
        });
      } catch (error) {
        rollback();
        return Promise.reject(error);
      }
      return request.then(() => {}, (error) => {
        rollback();
        throw error;
      });
    };
    return this.#enqueueOperation(append);
  }

  /** The cell's current display CFC label, for label-aware subscribers. */
  get cfcLabel(): CfcLabelView | undefined {
    return this.#cfcLabel;
  }

  /**
   * Whether {@link cfcLabel} is the label the worker read with the value the
   * handle holds, as {@link CellDelivery} tells a subscriber.
   */
  get cfcLabelCurrent(): boolean {
    return this.#cfcLabelCurrent;
  }

  /** Whether this handle subscribed asking for reactive CFC-label delivery. */
  get wantsCfcLabel(): boolean {
    return this.#wantsCfcLabel;
  }

  /**
   * Subscribe to cell value changes.
   *
   * At once, the subscriber hears what the handle holds: `callback` is called
   * with the current value (even if undefined), or, where the worker refused
   * the handle's last read, `onRefused` with the refusal and `callback` not
   * at all. A handle that has read nothing yet calls `callback` with
   * `undefined` as well; {@link lastRead} tells that apart from a cell that
   * holds nothing. A read that answers it ({@link sync}, {@link pull})
   * reaches `callback` whatever it holds, as does an update the connection
   * delivers; the connection delivers no update that holds nothing (the
   * worker sends one for a document it has not loaded too), so a handle on a
   * cell that holds nothing stays unread until a read answers it. After that,
   * `callback` is called whenever the value changes, and `onRefused` with
   * each refusal. Once any subscriber has asked for the label, every
   * subscriber of the handle is called again with the value it holds when the
   * label changes, and when the label stops or starts being current, as an
   * update the connection dropped and the worker's next labeled update make
   * it ({@link CellDelivery}). The
   * callback's return value, if a Cancel function, is called before the next
   * value, when a refusal drops the value it was given, and when the
   * subscription is cancelled.
   *
   * A change reaches `callback` through an update, or through what a read
   * finds, whether this handle made it or another on the same cell under the
   * same schema did, unless something newer than the read reached the handle
   * first: a write through it, or an update to the cell, while the read was in
   * flight. Since no update holds nothing, a subscriber that was given a value
   * hears that the cell now holds nothing only from a {@link pull}, as
   * `undefined`: a `sync()` that finds nothing may have found a document not
   * loaded yet, so the handle keeps the value it holds.
   */
  subscribe(
    callback: (
      value: T | undefined,
      cfcLabel?: CfcLabelView | undefined,
      delivery?: CellDelivery,
    ) => Cancel | undefined | void,
    options: CellSubscribeOptions,
  ): Cancel {
    // A caller the compiler did not check can leave the options out. It is
    // told what it owes, rather than meeting a property read of `undefined`.
    if (typeof options?.onRefused !== "function") {
      throw new TypeError(
        "CellHandle.subscribe() requires `options.onRefused`, which says what " +
          "the subscriber shows while the worker refuses the cell's read",
      );
    }
    this.#requireSchema("subscribe");
    // If a label-aware subscription is added AFTER a value-only one already
    // opened the backend subscription, that backend sub carries no label and
    // the connection would dedup this one away. Re-establish it so it delivers
    // labels (the worker recreates its sink with includeCfcLabel). This works
    // when this handle is the sole subscriber of its ref; a value-only handle
    // sharing the exact same ref would keep the backend sub label-less.
    const upgradeToCfcLabel = options.includeCfcLabel === true &&
      !this.#wantsCfcLabel && this.#callbacks.size > 0;
    if (options.includeCfcLabel) {
      this.#wantsCfcLabel = true;
    }
    const callbackId = this.#nextCallbackId++;
    let cleanup: Cancel | undefined | void;

    const clear = () => {
      if (typeof cleanup === "function") {
        try {
          cleanup();
        } catch (error) {
          console.error("[CellHandle] Cleanup error:", error);
        }
      }
      cleanup = undefined;
    };
    const wrappedCallback = (
      value: T | undefined,
      cfcLabel: CfcLabelView | undefined,
      delivery: CellDelivery,
    ) => {
      clear();
      try {
        cleanup = callback(value, cfcLabel, delivery);
      } catch (error) {
        console.error("[CellHandle] Callback error:", error);
      }
    };

    // The connection seeds a handle new to its ref with what another handle
    // of the same ref holds, which reaches this handle's subscribers so far.
    // The subscriber added here hears what the handle holds once, below.
    if (upgradeToCfcLabel) {
      // Tear down the label-less backend sub, then re-open it label-aware.
      // Off the cell's subscription in between, the handle hears of no
      // change, so the label it holds is current only once the worker
      // delivers one again, as when its last subscriber leaves.
      this.#cfcLabelCurrent = false;
      void this.#conn.unsubscribe(this).finally(() => {
        this.#conn.subscribe(this);
      });
    } else {
      this.#conn.subscribe(this);
    }

    const onRefused = options.onRefused;
    this.#callbacks.set(callbackId, {
      onValue: wrappedCallback,
      clear,
      onRefused: (refusal: CellReadRefusal) => {
        try {
          onRefused(refusal);
        } catch (error) {
          console.error("[CellHandle] Callback error:", error);
        }
      },
    });

    // Call immediately with what the handle holds, as Cell does: the value,
    // even when it is undefined, or the refusal standing in its place.
    if (this.#refusal === undefined) {
      wrappedCallback(this.#value, this.#cfcLabel, {
        cfcLabelCurrent: this.#cfcLabelCurrent,
      });
    } else {
      this.#callbacks.get(callbackId)?.onRefused(this.#refusal);
    }

    return () => {
      clear();
      this.#callbacks.delete(callbackId);
      if (this.#callbacks.size === 0) {
        // Off the cell's subscription, the handle hears of no label change,
        // so the one it holds is current only until the worker delivers
        // another.
        this.#cfcLabelCurrent = false;
        this.#conn.unsubscribe(this);
      }
    };
  }

  /**
   * Fetch the current value from the worker.
   * If the value is itself a link, follows it to get the actual value.
   *
   * @throws {CellReadRefusedError} When the worker refuses the read, which
   *   then stands as the handle's refusal and reaches its subscribers'
   *   `onRefused`.
   */
  async sync(): Promise<Readonly<T> | undefined> {
    return await this.#read({ type: RequestType.CellGet, cell: this.ref() });
  }

  /**
   * Demands lazy producers and required loads before fetching reactive state.
   * Writes may remain unconfirmed when this resolves, including for an absent
   * value or an empty object. Pass `awaitDurability: true` to also cross the
   * runtime-wide commit-aware barrier.
   *
   * What it finds reaches the subscribers of this handle, and of every other
   * handle on the same cell under the same schema that has not written since,
   * wherever it differs from what that handle holds, `undefined` included: a
   * pull waits for the loads its read starts, so finding nothing means the
   * cell holds nothing, and a subscriber that was given a value is told so.
   * It reaches none of them when this handle has written since, or an update
   * to the cell has reached any handle on it since, which is newer.
   *
   * @throws {CellReadRefusedError} When the worker refuses the read, as
   *   {@link sync} does.
   */
  async pull(
    options: { awaitDurability?: boolean } = {},
  ): Promise<Readonly<T> | undefined> {
    return await this.#read({
      type: RequestType.CellPull,
      cell: this.ref(),
      awaitDurability: options.awaitDurability ?? false,
    });
  }

  /**
   * Helper for {@link sync} and {@link pull}, which asks the worker for the
   * cell's value with `request` and caches the answer, unless a write or an
   * update this handle took meanwhile is newer. An answer that changes what
   * the handle holds reaches its subscribers, and the answer goes to every
   * other handle subscribed to the same cell under the same schema, except
   * that a `sync()` finding nothing leaves a value the handle holds in place
   * and goes to no other handle. A refused read caches the refusal and
   * rejects.
   */
  async #read(
    request: CellGetRequest | CellPullRequest,
  ): Promise<Readonly<T> | undefined> {
    const writeGeneration = this.#writeGeneration;
    const updateGeneration = this.#updateGeneration;
    const startedAt = writeClock;
    const answer = await this.#enqueueOperation((queue) =>
      this.#ask(queue, request, updateGeneration)
    );
    return this.#takeAnswer(
      request,
      answer,
      writeGeneration,
      updateGeneration,
      startedAt,
    );
  }

  /**
   * Helper for {@link #read} and {@link update}, which asks the worker with
   * `request` in the turn `queue` gives it, and records an answer no update
   * has overtaken as the base the queue's later operations start from.
   */
  async #ask(
    queue: CellOperationQueue,
    request: CellGetRequest | CellPullRequest,
    updateGeneration: number,
  ): Promise<ReadAnswer<T>> {
    const authoritativeGeneration = queue.authoritativeGeneration;
    const response = await this.#conn.request<
      RequestType.CellGet | RequestType.CellPull
    >(request);
    const read = this.#readOf(response);
    const authoritative = updateGeneration === this.#updateGeneration &&
      authoritativeGeneration === queue.authoritativeGeneration;
    if (authoritative) this.#recordInQueue(queue, read);
    return { read, response, authoritative };
  }

  /**
   * Helper for {@link #read} and {@link update}, which takes `answer`, the
   * worker's answer to `request`, as {@link #read} describes: what it finds
   * reaches subscribers where nothing newer has, and a refusal rejects.
   */
  #takeAnswer(
    request: CellGetRequest | CellPullRequest,
    { read, response, authoritative }: ReadAnswer<T>,
    writeGeneration: number,
    updateGeneration: number,
    startedAt: number,
  ): Readonly<T> | undefined {
    const latest = writeGeneration === this.#writeGeneration &&
      updateGeneration === this.#updateGeneration &&
      authoritative;
    if ("refused" in read) {
      // Subscribers hear it, as they hear a refused update.
      if (latest) this.#refuse(read.refused);
      throw new CellReadRefusedError(read.refused);
    }
    if (!latest) return read.value;
    // A pull that finds nothing has waited for the loads its read started, so
    // the cell holds nothing. A `sync()` waits for none, and finds nothing in
    // a document the worker has not loaded yet as well.
    const settled = read.value !== undefined ||
      request.type === RequestType.CellPull;
    // A read that ends a refusal, or answers a handle that held nothing yet,
    // is news to its subscribers even when it finds what the handle held.
    const isNews = this.#refusal !== undefined || this.#unread;
    if (isNews || (settled && !valuesOrCellsEqual(read.value, this.#value))) {
      this.#publishValue(read.value as T);
    }
    if (settled) {
      // The connection drops an update that holds nothing, so another handle
      // on the cell can hold a value this read found gone, whatever this
      // handle held. An update that reached any handle on the cell while the
      // read was in flight moved the queue's authoritative generation, so a
      // read that gets here is no older than what any peer took from one.
      for (const peer of this.#conn.peersOf(this)) {
        peer.#takeRead(response.value, startedAt);
      }
    }
    return read.value;
  }

  /**
   * Helper for {@link #read}, which takes `value`, the raw answer to a read
   * another handle on this cell made, as what this handle holds, telling its
   * subscribers when that changes it, unless this handle has mutated its
   * value since the read was made, at `startedAt` on `writeClock`.
   */
  #takeRead(value: unknown, startedAt: number): void {
    if (this.#writeGeneration > startedAt) return;
    const applied = applyValue(value, this.#value, this) as T;
    const isNews = this.#refusal !== undefined || this.#unread;
    if (isNews || !valuesOrCellsEqual(applied, this.#value)) {
      this.#publishValue(applied);
    }
  }

  /** What `response`, a read's answer, holds: the value, or the refusal. */
  #readOf(response: CellValueResponse): CellAnswer<T> {
    return response.refused !== undefined
      ? { refused: response.refused }
      : { value: CellHandle.deserialize<T>(this, response.value) as T };
  }

  /**
   * Records `read` as the base the queue's later appends start from: a value,
   * or none for a refusal, which holds nothing of the cell.
   */
  #recordInQueue(queue: CellOperationQueue, read: CellAnswer<T>): void {
    if ("refused" in read) {
      queue.value = undefined;
      queue.hasValue = false;
    } else {
      this.#holdInQueue(queue, read.value);
    }
  }

  /**
   * Records `value` as what this handle's operation left in `queue`, as read
   * under this handle's schema.
   */
  #holdInQueue(queue: CellOperationQueue, value: unknown): void {
    queue.value = value;
    queue.hasValue = true;
    queue.view = this.#viewKey();
  }

  /**
   * The value the operations before this one left in `queue`, where they
   * left one read under this handle's schema; `undefined` otherwise.
   */
  #queuedValue(queue: CellOperationQueue): { value: unknown } | undefined {
    return queue.hasValue && queue.view === this.#viewKey()
      ? { value: queue.value }
      : undefined;
  }

  /** What decides the shape a read through this handle returns. */
  #viewKey(): string {
    return this.#view ??= JSON.stringify({
      schema: this.#ref.schema,
      keyedFrom: this.#ref.keyedFrom,
    });
  }

  /**
   * Resolve links in this cell to get the actual cell it points to.
   * Returns a new CellHandle pointing to the resolved cell.
   */
  async resolveAsCell(): Promise<CellHandle<T>> {
    const response = await this.#enqueueOperation(() =>
      this.#conn.request<RequestType.CellResolveAsCell>({
        type: RequestType.CellResolveAsCell,
        cell: this.ref(),
      })
    );

    return new CellHandle<T>(this.#rt, response.cell);
  }

  /**
   * The fields this record cell holds, by name, each as a handle on the
   * field within it, whose reads the worker decides one by one. Nothing a
   * field holds is read to list them. The names come from the record, not its
   * schema. `undefined` where the cell holds no record (nothing at all, a
   * list, or a single value), which is not the empty list of a record that
   * holds no fields.
   *
   * @throws {CellReadRefusedError} When the worker refuses even the list, as
   *   for a record whose own label the viewer may not see.
   * @throws When the worker could not read the record, as when its space
   *   refused the worker access.
   */
  async fields(): Promise<Record<string, CellHandle<unknown>> | undefined> {
    const response = await this.#enqueueOperation(() =>
      this.#conn.request<RequestType.CellFields>({
        type: RequestType.CellFields,
        cell: this.ref(),
      })
    );
    if (response.refused !== undefined) {
      throw new CellReadRefusedError(response.refused);
    }
    if (response.fields === undefined) return undefined;
    const fields: Record<string, CellHandle<unknown>> = {};
    for (const [name, ref] of Object.entries(response.fields)) {
      fields[name] = new CellHandle(this.#rt, ref);
    }
    return fields;
  }

  async getCfcLabel(): Promise<CfcLabelView | undefined> {
    const response = await this.#enqueueOperation(() =>
      this.#conn.request<RequestType.CellGetCfcLabel>({
        type: RequestType.CellGetCfcLabel,
        cell: this.ref(),
      })
    );
    return response.cfcLabel;
  }

  /** Run a read-only query when this handle refers to a SQLite database. */
  async querySqlite<Row = Record<string, unknown>>(
    sql: string,
    params?: ReadonlyArray<ClientCellValue> | Record<string, ClientCellValue>,
  ): Promise<Row[]> {
    const serializedParams = params === undefined
      ? undefined
      : CellHandle.#serializeSqliteParams(params);
    const response = await this.#enqueueOperation(() =>
      this.#conn.request<RequestType.SqliteQuery>({
        type: RequestType.SqliteQuery,
        cell: this.ref(),
        sql,
        ...(serializedParams !== undefined && {
          params: serializedParams,
        }),
      })
    );
    return response.rows.map((row) =>
      Object.fromEntries(
        Object.entries(row).map(([key, value]) => [
          key,
          CellHandle.deserialize(this, value),
        ]),
      ) as Row
    );
  }

  /** Commit a SQL write when this handle refers to a SQLite database. */
  async execSqlite(
    sql: string,
    params?: ReadonlyArray<ClientCellValue> | Record<string, ClientCellValue>,
  ): Promise<void> {
    const serializedParams = params === undefined
      ? undefined
      : CellHandle.#serializeSqliteParams(params);
    await this.#enqueueOperation(() =>
      this.#conn.request<RequestType.SqliteExec>({
        type: RequestType.SqliteExec,
        cell: this.ref(),
        sql,
        ...(serializedParams !== undefined && {
          params: serializedParams,
        }),
      })
    );
  }

  equals(other: unknown): boolean {
    if (this === other) return true;
    if (!isCellHandle(other)) return false;
    return cellRefsEqual(this.ref(), other.ref());
  }

  /**
   * Create a new CellHandle with a different schema. It addresses the same
   * cell, so it starts in the state this handle's last read left: the value,
   * the refusal standing in its place, which refuses its writes as it does
   * this handle's, or nothing yet. Its first read of its own replaces it.
   */
  asSchema<U = unknown>(schema: JSONSchema): CellHandle<U> {
    const { schema: _schema, keyedFrom: _keyedFrom, ...rest } = this.#ref;
    const newCell = new CellHandle(this.#rt, {
      ...rest,
      schema,
    });
    newCell.#value = this.#value;
    newCell.#refusal = this.#refusal;
    newCell.#unread = this.#unread;
    return newCell as CellHandle<U>;
  }

  #extendPath(key: string): CellRef {
    // The child's schema is the worker's to narrow, so the child carries the
    // nearest ancestor schema instead of one of its own.
    const keyedFrom = this.#ref.schema !== undefined
      ? { depth: this.#ref.path.length, schema: this.#ref.schema }
      : this.#ref.keyedFrom;
    return {
      id: this.#ref.id,
      space: this.#ref.space,
      scope: this.#ref.scope,
      path: [...this.#ref.path, key],
      ...(keyedFrom !== undefined && { keyedFrom }),
      ...(this.#ref.cfcLabelView !== undefined && {
        cfcLabelView: rebaseCfcLabelView(this.#ref.cfcLabelView, [key]),
      }),
    };
  }

  /**
   * Returns the sigil link naming this cell.
   *
   * The link is what stands in for a cell wherever a cell itself has no
   * representation, and this is how a caller that has already recognized the
   * value as a handle asks for it -- by name, rather than by reaching for a
   * serialization protocol that happens to yield one. `Cell` is arranged the
   * same way, its `toSigilLinkOrNull()` answering the callers that have
   * recognized a cell and its `toJSON()` answering the protocol.
   *
   * The ref-carried `cfcLabelView` is deliberately not included, as
   * `toWireString()` omits it: what this produces re-enters the worker without
   * passing `getCell()` or `cellRefToSigilLink()`, and a main-thread display
   * copy must not go back in as label state.
   */
  toSigilLink(): SigilLink {
    return linkRefFrom<CfcCellLinkRefPayload>({
      id: this.#ref.id,
      space: this.#ref.space,
      scope: this.#ref.scope,
      path: this.#ref.path,
      ...(this.#ref.schema !== undefined && { schema: this.#ref.schema }),
      ...(this.#ref.overwrite !== undefined &&
        { overwrite: this.#ref.overwrite }),
    });
  }

  /**
   * Returns the same link under the JSON protocol's name, so a handle reads as
   * the cell it names wherever a renderer honors that protocol -- which is what
   * carries one through `JSON.stringify()`. A caller that has recognized the
   * handle asks {@link toSigilLink} instead.
   */
  toJSON(): SigilLink {
    return this.toSigilLink();
  }

  /**
   * Encodes this cell's link to a wire string (the `fcl1:` cell-link form) for
   * transport across a string boundary (e.g. an HTTP body) from which it will
   * be decoded back to a link. Only the plain addressing fields cross the wire;
   * `schema` and the cfc label view are deliberately omitted (see
   * {@link linkRefPayloadToString}).
   */
  toWireString(): string {
    return linkRefPayloadToString({
      id: this.#ref.id,
      space: this.#ref.space,
      ...(this.#ref.scope !== undefined && { scope: this.#ref.scope }),
      path: this.#ref.path,
      ...(this.#ref.overwrite !== undefined &&
        { overwrite: this.#ref.overwrite }),
    });
  }

  /**
   * Handles an update of the cell the connection dropped. The handle keeps
   * the value and label it holds, but the worker has read the cell since, so
   * the label is no longer the one the worker read with the cell's value. A
   * handle delivering labels hands its subscribers what it holds again,
   * saying so: one that took the label as the worker's read, as
   * `observeAuthorship()` does, would otherwise hear nothing more, since no
   * later update need come.
   */
  [$onCellUpdateDropped](): void {
    if (!this.#cfcLabelCurrent) return;
    this.#cfcLabelCurrent = false;
    if (!this.#wantsCfcLabel) return;
    for (const { onValue } of this.#callbacks.values()) {
      onValue(this.#value as Readonly<T>, this.#cfcLabel, {
        cfcLabelCurrent: false,
      });
    }
  }

  /**
   * Handles an update of the cell from the backend, whose raw value may
   * contain `CellRef`s.
   */
  [$onCellUpdate](
    value: unknown,
    labelUpdate?: { cfcLabel: CfcLabelView | undefined; current: boolean },
  ): void {
    this.#updateGeneration++;
    // A value admitted after a refusal, or the first delivered to a handle
    // that held nothing yet, is news to the subscribers, even one equal to
    // what the handle held.
    const isNews = this.#refusal !== undefined || this.#unread;
    this.#refusal = undefined;
    this.#unread = false;
    const applied = applyValue(
      value,
      this.#value,
      this as CellHandle<unknown>,
    ) as T;
    const queue = operationQueues.get(this.#rt)?.get(
      cellRefToIdentityKey(this.#ref),
    );
    if (queue) {
      queue.authoritativeGeneration++;
      this.#holdInQueue(queue, applied);
    }
    const valueChanged = !valuesOrCellsEqual(applied, this.#value);
    // A label-only change (value identical) still fires label-aware
    // subscribers, and so does a change in whether the label is current alone,
    // as when the worker brings back the value and label a dropped update left
    // stale. `labelUpdate` is present only on notifications that carried a
    // label, so a value-only notification never spuriously churns the label.
    const labelChanged = labelUpdate !== undefined && this.#wantsCfcLabel &&
      (!cfcLabelViewsEqual(labelUpdate.cfcLabel, this.#cfcLabel) ||
        labelUpdate.current !== this.#cfcLabelCurrent);
    // The label the handle holds follows every update that carries one,
    // whether or not a subscriber hears of it: a handle whose subscribers
    // never asked for labels still seeds a later handle on the cell with the
    // label it holds, and says how current it is.
    if (labelUpdate !== undefined) {
      this.#cfcLabel = labelUpdate.cfcLabel;
      this.#cfcLabelCurrent = labelUpdate.current;
    } else if (valueChanged) {
      // The value moved on without its label, which the worker read with an
      // earlier one.
      this.#cfcLabelCurrent = false;
    }
    if (!valueChanged && !labelChanged && !isNews) {
      return;
    }

    if (valueChanged) this.#value = applied;
    for (const { onValue } of this.#callbacks.values()) {
      onValue(this.#value as Readonly<T>, this.#cfcLabel, {
        cfcLabelCurrent: this.#cfcLabelCurrent,
      });
    }
  }

  /**
   * Handles a subscription update the worker refused. Each refusal reaches
   * the subscribers that listen for one, since each stands for a change to
   * what the read found, while what the handle holds, nothing of the cell,
   * stays the same.
   */
  [$onCellRefused](refusal: CellReadRefusal): void {
    this.#updateGeneration++;
    const queue = operationQueues.get(this.#rt)?.get(
      cellRefToIdentityKey(this.#ref),
    );
    if (queue) {
      queue.authoritativeGeneration++;
      queue.value = undefined;
      queue.hasValue = false;
    }
    this.#refuse(refusal);
  }

  /**
   * Recursively hydrate any object, converting any sigil links into
   * CellHandle instances. `$alias` records are plain data — they are only
   * meaningful as bindings inside Pattern objects, which the client never
   * interprets. A `FabricPrimitive` comes back as itself.
   *
   * @throws If the value holds a `FabricInstance`, which is a container this
   *   walk cannot descend and so cannot hydrate a link inside.
   */
  static deserialize<T>(
    base: CellHandle<T>,
    value: unknown,
  ): unknown {
    if (
      !value && typeof value === "string" || typeof value === "boolean" ||
      typeof value === "number"
    ) {
      return value;
    }

    if (Array.isArray(value)) {
      return value.map((item) => CellHandle.deserialize(base, item));
    }

    // A `FabricPrimitive` is a leaf, so a walk that stops at it has already
    // done the right thing -- and this goes _before_ the record branch, which
    // rebuilds from enumerable own properties a fabric class does not have and
    // would put `{}` here in place of the value.
    if (value instanceof FabricPrimitive) return value;

    // An instance is a container, reached by its codec contents rather than by
    // property name, so a sigil link can sit inside one where this walk cannot
    // see it -- and a value handed back unhydrated would carry that link where
    // a `CellHandle` belongs.
    //
    // Nothing reaches this today, de facto rather than by construction. The
    // transport does not stop an instance: the envelope's encoding carries one
    // across with its class. What keeps it unreachable is the refusal at each
    // of the other ends of the crossing -- `convertCellsToLinks()` on the way
    // out of the worker, and `serialize()` below on the way in.
    if (value instanceof FabricInstance) {
      refuseFabricInstance(value, "when hydrating a value off the connection");
    }

    if (isObjectOrArray(value)) {
      const reference = parseAsCellRef(
        value as JSONValue | undefined,
        base.ref(),
      );
      if (reference) {
        return new CellHandle(base.#rt, reference);
      }

      const result: Record<string, unknown> = {};
      for (const [key, val] of Object.entries(value)) {
        result[key] = CellHandle.deserialize(base, val);
      }
      return result;
    }

    return value;
  }

  static #serializeSqliteParams(
    params: ReadonlyArray<ClientCellValue> | Record<string, ClientCellValue>,
  ): SqliteParams {
    const serialize = (value: ClientCellValue) =>
      CellHandle.#sqliteFabricValue(value);
    return Array.isArray(params)
      ? { kind: "positional", values: params.map(serialize) }
      : {
        kind: "named",
        entries: Object.entries(params).map(([key, value]) => [
          key,
          serialize(value),
        ]),
      };
  }

  static #sqliteFabricValue(value: ClientCellValue): FabricValue {
    if (isCellHandle(value)) return value.ref();
    if (value instanceof FabricBytes) return value;
    if (isFabricSpecialObject(value)) {
      throw new TypeError(
        `SQLite bind values support \`FabricBytes\` but not ` +
          `\`${value.constructor.name}\`.`,
      );
    }
    if (Array.isArray(value)) {
      return value.map((member) => CellHandle.#sqliteFabricValue(member));
    }
    if (isObjectOrArray(value)) {
      return Object.fromEntries(
        Object.entries(value).map(([key, member]) => [
          key,
          CellHandle.#sqliteFabricValue(member),
        ]),
      );
    }
    return value as FabricValue;
  }

  /**
   * Converts a value the client holds into the one the connection carries,
   * which is the same data with a `CellRef` wherever a `CellHandle` sat.
   *
   * A `FabricPrimitive` crosses as itself, the envelope's encoding carrying it
   * with its class, as it carries a `bigint` and a `symbol`.
   *
   * `CellHandle.deserialize()` is the inverse.
   *
   * @throws If the value holds a `FabricInstance`, which is a container this
   *   walk cannot descend and so cannot convert a handle inside.
   */
  static serialize(value: ClientCellValue): FabricValue {
    return CellHandle.#serialize(value, "ref") as FabricValue;
  }

  /**
   * Serializes `value` for the wire, converting each `CellHandle` to a link
   * in `linkFormat`: a `CellRef` (`ref`) or a hydratable sigil link (`sigil`).
   * An ordinary record that happens to resemble a `CellRef` is walked as data,
   * and a live `CellHandle` is never replaced in place.
   */
  static #serialize(
    value: ClientCellValue,
    linkFormat: "ref" | "sigil" = "ref",
  ): unknown {
    if (isCellHandle(value)) {
      return linkFormat === "sigil"
        ? linkRefFrom<CfcCellLinkRefPayload>(value.ref())
        : value.ref();
    }

    if (Array.isArray(value)) {
      return value.map((element) => CellHandle.#serialize(element, linkFormat));
    }

    // A `FabricPrimitive` crosses whole rather than being walked: it is a
    // leaf, so stopping at it loses nothing. This goes _before_ the record
    // test, since such a value is also a record and that branch would rebuild
    // it from enumerable own properties it is not supposed to have, putting
    // `{}` on the wire in place of a `FabricBytes`.
    if (value instanceof FabricPrimitive) return value;

    // An instance is a container whose contents this walk cannot reach, so a
    // `CellHandle` inside one would cross unconverted -- as a handle, which
    // the wire has no representation for. Refused here rather than downstream:
    // the worker's `mapCellRefsToSigilLinks()` refuses one as well, and a
    // refusal there arrives as an error reply, after this handle has already
    // cached the value and told its subscribers. Refused through the shared
    // helper, as `deserialize()` above already does, so the two walks say the
    // same thing about the same value.
    if (value instanceof FabricInstance) {
      refuseFabricInstance(value, "when sending a value over this connection");
    }

    if (isObjectOrArray(value)) {
      return Object.fromEntries(
        Object.entries(value).map((
          [key, member],
        ) => [key, CellHandle.#serialize(member, linkFormat)]),
      );
    }

    // Every remaining arm is a primitive the encoding carries as itself.
    // `CellHandle<T>` does not constrain `T`, so a caller can arrive here with
    // something `ClientCellValue` does not admit at all; that is the encode's
    // to answer, per the input contract, and not worth a check on the way to
    // it that every correct value would also pay.
    //
    // Cast because `Array.isArray()` does not narrow a `readonly` array out of
    // the union, though it does catch one at run time -- so the arm TypeScript
    // still sees here cannot actually reach it.
    return value as FabricValue;
  }
}

export function isCellHandle<T = unknown>(
  value: unknown,
): value is CellHandle<T> {
  return value instanceof CellHandle;
}

/**
 * Applies `current` to `previous`, returning the result.
 * Notably, this preserves `CellHandle` instances when encountering
 * a `CellRef` referencing the same `CellHandle`.
 */
function applyValue<T>(
  current: unknown,
  previous: unknown,
  base: CellHandle<T>,
): unknown {
  const cellRef = parseAsCellRef(current as JSONValue, base.ref());

  if (cellRef) {
    if (isCellHandle(previous) && cellRefsEqual(cellRef, previous.ref())) {
      return previous;
    }
    return new CellHandle(base.runtime(), cellRef);
  }

  // Currently, `current` will not contain `CellHandle`s,
  // but for completeness.
  if (isCellHandle(current)) {
    if (isCellHandle(previous) && current.equals(previous)) {
      return previous;
    }
    return current;
  }

  // For arrays, recursively apply to each element
  if (Array.isArray(current)) {
    const prevArray = Array.isArray(previous) ? previous : [];
    return current.map((item, index) =>
      applyValue(item, prevArray[index], base)
    );
  }

  // A leaf, carried through whole as `deserialize()` hydrates one: the record
  // branch below would rebuild it from enumerable own properties it does not
  // have.
  if (current instanceof FabricPrimitive) {
    return current;
  }

  // A container this walk cannot descend, so it cannot preserve a handle
  // inside one against the incoming value the way it does for a record.
  // Unreachable for the same reason as in `deserialize()`.
  if (current instanceof FabricInstance) {
    refuseFabricInstance(current, "when applying a delivered value");
  }

  // For plain objects, recursively apply to each property
  if (isObjectOrArray(current)) {
    const prevRecord = (isObjectNotArray(previous))
      ? previous as Record<string, unknown>
      : {};
    const result: Record<string, unknown> = {};
    for (const [key, val] of Object.entries(current)) {
      result[key] = applyValue(val, prevRecord[key], base);
    }
    return result;
  }

  // For primitives, just return current
  return current;
}

function cellRefsEqual(a: CellRef, b: CellRef): boolean {
  if (a.id !== b.id) return false;
  if (a.space !== b.space) return false;
  if ((a.scope ?? "space") !== (b.scope ?? "space")) return false;
  if (a.path.length !== b.path.length) return false;
  for (let i = 0; i < a.path.length; i++) {
    if (a.path[i] !== b.path[i]) return false;
  }
  if (!cfcLabelViewsEqual(a.cfcLabelView, b.cfcLabelView)) return false;
  return true;
}

/**
 * Compares two values a cell can hold, deciding whether a delivered update is
 * a change this handle's subscribers hear about.
 *
 * Distinct from `valueEqual()`, which it calls: that settles a `FabricValue`
 * by content, and this also knows about cells. A `CellHandle` compares by the
 * cell it names rather than by what that cell holds, so two handles on one
 * cell are equal and a handle is equal to nothing else.
 */
function valuesOrCellsEqual(a: unknown, b: unknown): boolean {
  // `Object.is`, not `===`: an unchanged `NaN` leaf must compare equal (else
  // every delivery of a NaN-bearing value re-notifies all subscribers), and a
  // `0` -> `-0` change must compare unequal (else the update is dropped).
  if (Object.is(a, b)) return true;
  if (a == null || b == null) return a === b;
  if (typeof a !== typeof b) return false;
  if (typeof a !== "object") return false;
  // Either side being a handle is enough to ask. A handle holds its state
  // privately, so the record branch below would read `{}` off it and call it
  // equal to anything else without enumerable own keys -- `{}` itself among
  // them, so replacing a handle with a record would be judged a no-change and
  // reach no subscriber. A handle is a reference to a cell and equals only
  // another reference to the same cell.
  if (isCellHandle(a) || isCellHandle(b)) {
    return isCellHandle(a) && isCellHandle(b) && a.equals(b);
  }

  // A `FabricPrimitive` is compared by the data model rather than by this
  // walk, and _before_ the record branch, for the same reason: two
  // `FabricBytes` over different bytes both present as `{}` there and would
  // compare equal. A primitive is a leaf, so comparing its content is the
  // whole comparison.
  //
  // There is no arm for a `FabricInstance`. `applyValue()` is this function's
  // only caller and refuses one before it returns, so neither argument can
  // hold one -- an arm here would be unreachable rather than defensive, and
  // untestable with it.
  if (a instanceof FabricPrimitive || b instanceof FabricPrimitive) {
    return valueEqual(a as FabricValue, b as FabricValue);
  }
  if (Array.isArray(a)) {
    if (!Array.isArray(b)) return false;
    if (a.length !== b.length) return false;
    for (let i = 0; i < a.length; i++) {
      if (!valuesOrCellsEqual(a[i], b[i])) return false;
    }
    return true;
  }
  if (Array.isArray(b)) return false;

  const aKeys = Object.keys(a as object);
  const bKeys = Object.keys(b as object);
  if (aKeys.length !== bKeys.length) return false;

  for (const key of aKeys) {
    if (!(key in (b as object))) return false;
    if (
      !valuesOrCellsEqual(
        (a as Record<string, unknown>)[key],
        (b as Record<string, unknown>)[key],
      )
    ) {
      return false;
    }
  }

  return true;
}

function parseAsCellRef(
  value: JSONValue | undefined,
  from: CellRef,
): CellRef | undefined {
  if (isSigilLink(value)) {
    const linkData = linkRefPayload(value);

    return {
      id: linkData.id ?? from.id,
      space: linkData.space ?? from.space,
      scope: linkData.scope === "space" || linkData.scope === "user" ||
          linkData.scope === "session"
        ? linkData.scope
        : from.scope,
      path: (linkData.path ?? []).map((p) => p.toString()),
      ...(linkData.schema !== undefined && { schema: linkData.schema }),
      ...((linkData as { cfcLabelView?: CfcLabelView }).cfcLabelView !==
          undefined && {
        cfcLabelView: (linkData as { cfcLabelView?: CfcLabelView })
          .cfcLabelView,
      }),
    };
  }
}