ng-eventually/packages/client/src/surface/watch-shape.ts

/**
 * watch-shape — a REACTIVE, TanStack-`useQuery`-shaped read over one SHEX shape in
 * one logical scope. This is the surface the consuming app will bind (phase B) with
 * `useSyncExternalStore` — the polyfill deliberately exposes an OBSERVABLE, never a
 * React hook (the lib has NO React dependency, same constraint as `subscribe.ts`).
 *
 * ── Why an observable, and why this exact shape ────────────────────────────
 * It anticipates NextGraph's planned `useShape(shape, scope)` upgrade, which will
 * natively distinguish "sync in progress" from "synced but empty". That distinction
 * ALREADY exists lib-internally (`open-repo.ts` `getSyncState`: syncing / synced /
 * timed-out); `watchShape` merely SURFACES it as a `useQuery`-minimal snapshot:
 *   `ShapeQuery<T> = { data: T[]; isPending; isSuccess; isError; error }`.
 * `data` is ALWAYS an array (never `undefined`), so a synced-but-empty scope reads
 * `{ data: [], isPending: false, isSuccess: true }` — the key distinction — while a
 * scope still syncing reads `{ data: [], isPending: true, isSuccess: false }`.
 *
 * ── What the observable OWNS (the whole read pipeline) ─────────────────────
 *   1. Resolve the logical scope → the doc set: the CURRENT wallet's own per-entity
 *      documents for that scope (`storeRegistry.listMyEntityDocs`), and nothing
 *      else. There is no "everything public" to fold in — **you cannot discover,
 *      you can only follow links** (see docs/readcap-and-nuri-model.md §4ter-bis),
 *      and a link reaches you through an inbox or through a document you already
 *      hold. A document whose cap you were given is read by NAMING it
 *      (`readUnion`), not by turning up in a scope you never put it in.
 *   2. Open the docs (`ensureReposOpen`) — this AWAITS the sync BARRIER (first
 *      `State` per doc, `getSyncState` → `synced`, or `timed-out` on the bounded
 *      fallback). `isPending` holds until the barrier is reached for the current
 *      doc set AND the first `readUnion` has rendered.
 *   3. `readUnion(docs)` — the read-model (cap filter already applied inside; we do
 *      NOT double-filter), then FILTER the union by the requested shape's `@type`
 *      (a `readUnion` union spans multiple types; each `watchShape` yields only the
 *      subjects of its shape). Non-domain: the type IRI is read from the SHEX
 *      ShapeType, not from any application concept.
 *
 * ── Reactivity WITHOUT polling (no `setInterval`) ──────────────────────────
 * Reactivity is push-only (rule no-broker-polling). It has TWO sources: document
 * pushes, and the KEYRING — a cap that arrives asynchronously (an inbox delivery
 * absorbed by the consumer's `inbox.watch`) makes documents readable that were not,
 * so `CapRegistry.onChange` re-reads. Without that, a view stays stale until an
 * unrelated push happens to fire. On the document side, `subscribeDoc` on every doc
 * in the current set re-runs `readUnion` on any push. The set is DYNAMIC (creating an
 * entity appends a NURI to the scope-index doc), so we ALSO subscribe to the
 * scope-index document: a push there re-RESOLVES the scope and re-keys the
 * subscribed set. Subscriptions are idempotent — an already-followed
 * doc is not re-subscribed. Everything reuses `subscribe.ts` / `open-repo.ts`; no
 * parallel channel.
 *
 * ── timed-out → isSuccess (best-effort), NOT isError ───────────────────────
 * A doc whose barrier fell back to `timed-out` still counts as "barrier reached"
 * (`isSuccess`): a slow-but-empty wallet must read as empty-success, not error.
 * `isError` fires ONLY on a real thrown exception in the pipeline.
 */

import { getCaps, getCurrentUser } from "../shared-wallet/bootstrap";
import { ensureReposOpen, getSyncState } from "../emulated-verifier/open-repo";
import { readUnion, type UnionSubject } from "./read-model";
import { subscribeDoc, type Unsubscribe } from "./subscribe";
import { listMyEntityDocs, userStoreDoc } from "../shared-wallet/account-registry";
import type { Nuri, Scope } from "../model/types";

/**
 * The RDF `type` predicate IRI. A SHEX shape pins its class via a triple
 * constraint on this predicate; we filter the read union by it.
 */
const RDF_TYPE = "http://www.w3.org/1999/02/22-rdf-syntax-ns#type";

/**
 * A minimal TanStack-`useQuery`-shaped read snapshot. `data` is ALWAYS an array
 * (never `undefined`). Defaults `T` to {@link UnionSubject} — `watchShape` yields
 * the generic per-subject property bags of the read-model (NO application domain);
 * the app maps them to its own entity types in phase B.
 */
export interface ShapeQuery<T = UnionSubject> {
  /** The subjects of the requested shape/scope. Empty array when none (never undefined). */
  data: T[];
  /** True while the sync barrier for the current doc set is not yet reached OR the
   *  first `readUnion` has not rendered. Mutually exclusive with `isSuccess`. */
  isPending: boolean;
  /** True once the barrier is reached (all docs `synced` OR `timed-out`) AND the
   *  first `readUnion` has rendered. A synced-but-EMPTY scope is `isSuccess` with
   *  `data: []` — the distinction this surface exists for. */
  isSuccess: boolean;
  /** True ONLY on a real thrown exception in the read pipeline (never for `timed-out`). */
  isError: boolean;
  /** The caught error when `isError`, else `undefined`. */
  error: unknown;
}

/** The observable a caller binds with `useSyncExternalStore` (phase B). */
export interface ShapeObservable<T = UnionSubject> {
  /** The current snapshot. STABLE across calls until it actually changes (so
   *  `useSyncExternalStore` does not loop): the same reference is returned until a
   *  state transition produces a new one. */
  getSnapshot(): ShapeQuery<T>;
  /** Register a change listener; returns an unsubscribe. The last listener's
   *  unsubscribe tears down the underlying doc subscriptions. */
  subscribe(onChange: () => void): () => void;
  /** Force a re-resolve + re-read now (e.g. an imperative refresh). Idempotent
   *  w.r.t. subscriptions; never polls. */
  refetch(): void;
}

/**
 * Derive the class IRI(s) a SHEX {@link ShapeType} pins on `rdf:type`, if any.
 * A generated shape constrains its subject's type via a triple constraint on the
 * `rdf:type` predicate whose `literals` carry the class IRI(s). Returns the set of
 * those IRIs, or `null` when the shape pins NO type (then no type-filter is applied
 * and every subject in the doc set flows through). Purely structural — reads only
 * the SHEX schema, no application domain.
 */
function shapeTypeIris(shapeType: unknown): Set<string> | null {
  try {
    const st = shapeType as {
      shape?: string;
      schema?: Record<string, { predicates?: Array<{ iri?: string; dataTypes?: Array<{ literals?: unknown[] }> }> }>;
    };
    const shape = st?.shape && st.schema ? st.schema[st.shape] : undefined;
    const preds = shape?.predicates ?? [];
    const iris = new Set<string>();
    for (const p of preds) {
      if (p?.iri !== RDF_TYPE) continue;
      for (const dt of p.dataTypes ?? []) {
        for (const lit of dt.literals ?? []) {
          if (typeof lit === "string") iris.add(lit);
        }
      }
    }
    return iris.size > 0 ? iris : null;
  } catch {
    return null;
  }
}

/** Whether a subject satisfies the shape's `@type` constraint (or the shape pins none). */
function matchesShape(subject: UnionSubject, typeIris: Set<string> | null): boolean {
  if (!typeIris) return true; // shape pins no rdf:type → accept every subject
  const types = subject.props[RDF_TYPE] ?? [];
  return types.some((t) => typeIris.has(t));
}

/**
 * Whether the sync BARRIER is reached for the whole doc set. Called only AFTER
 * `ensureReposOpen(docs)` has resolved, so each doc has been requested; the only
 * state that still holds the barrier open is `syncing` (subscribed, first `State`
 * not yet received). `synced` and `timed-out` both count as reached (`timed-out` is
 * best-effort, not an error). `unknown` means the injected `ng` has no
 * `doc_subscribe` (the fake/no-op open path, which has NO barrier semantics) — after
 * a completed open it can only mean that path, so it counts as reached (opened,
 * nothing to wait on). An EMPTY doc set is trivially past the barrier.
 */
function barrierReached(docs: Nuri[]): boolean {
  for (const d of docs) {
    if (getSyncState(d) === "syncing") return false;
  }
  return true;
}

/**
 * Create a reactive, `useQuery`-shaped observable over one SHEX `shapeType` in one
 * logical `scope` (`'public' | 'protected' | 'private'`). See the module header for
 * the full pipeline. The returned observable is inert until its first
 * {@link ShapeObservable.subscribe} (or {@link ShapeObservable.refetch}) — that is
 * what kicks off resolution, opening and the first read; before then `getSnapshot`
 * reports the initial pending snapshot.
 */
export function watchShape<T = UnionSubject>(
  shapeType: unknown,
  scope: Scope,
): ShapeObservable<T> {
  const typeIris = shapeTypeIris(shapeType);

  // The current, STABLE snapshot (same reference until a transition rebuilds it).
  let snapshot: ShapeQuery<UnionSubject> = {
    data: [],
    isPending: true,
    isSuccess: false,
    isError: false,
    error: undefined,
  };

  const listeners = new Set<() => void>();
  let started = false;
  // The docs currently subscribed for change signals, keyed by NURI → unsubscribe.
  // Idempotent: a doc already here is not re-subscribed. Excludes the container
  // (scope-index / discovery-index) subscriptions, held separately.
  const docSubs = new Map<Nuri, Unsubscribe>();
  // Container subscriptions (scope-index doc; discovery-index doc for `public`) —
  // a push here means the doc SET may have changed → re-resolve.
  const containerSubs = new Map<Nuri, Unsubscribe>();
  // Unsubscribe from the held-caps change signal (see the subscription in `start`).
  let capsUnsub: (() => void) | null = null;
  // True while `resolveDocs` runs. Folding a repo link files a cap, which fires the
  // held-caps signal; the resolution in progress already accounts for it, so the
  // signal is ignored during that window instead of restarting the cycle.
  let resolving = false;
  // Monotonic token so a slow in-flight refresh cannot clobber a newer one.
  let refreshToken = 0;

  function emit(): void {
    for (const l of listeners) {
      try {
        l();
      } catch (error) {
        console.error("[watch-shape] listener threw", error);
      }
    }
  }

  function setSnapshot(next: ShapeQuery<UnionSubject>): void {
    snapshot = next;
    emit();
  }

  /** Resolve the logical scope → the current doc set: the CURRENT wallet's own
   *  entity documents for that scope, and nothing else. Tolerant: a resolution
   *  failure yields whatever resolved.
   *
   *  There is no "everything public" to fold in. You cannot discover; you can only
   *  follow links, and a link reaches you through an inbox or through a document
   *  you already hold — never through a shared index. A document someone gave you
   *  the cap for is read by naming it (`readUnion`), not by appearing in
   *  a scope you did not put it in. */
  async function resolveDocs(): Promise<Nuri[]> {
    const user = getCurrentUser();
    const set = new Set<Nuri>();
    resolving = true;
    try {
      if (user) {
        try {
          for (const d of await listMyEntityDocs(user, scope)) set.add(d);
        } catch (error) {
          console.error("[watch-shape] listMyEntityDocs failed", error);
        }
      }
    } finally {
      resolving = false;
    }
    return [...set];
  }

  /** Subscribe to the CONTAINER document (the scope index) so a change to the doc
   *  SET re-resolves. Idempotent per NURI. */
  async function ensureContainerSubs(): Promise<void> {
    const containers: Nuri[] = [];
    const user = getCurrentUser();
    if (user) {
      try {
        containers.push(await userStoreDoc(user, scope));
      } catch (error) {
        console.error("[watch-shape] userStoreDoc failed", error);
      }
    }
    for (const c of containers) {
      if (!c || containerSubs.has(c)) continue;
      // A push on a container doc means the set may have changed → full re-resolve.
      containerSubs.set(c, subscribeDoc(c, () => void refresh()));
    }
  }

  /** Re-key the per-doc change subscriptions to exactly `docs` (idempotent adds,
   *  prune removed). A push on any of these re-reads (data-only, no re-resolve). */
  function syncDocSubs(docs: Nuri[]): void {
    const wanted = new Set(docs.filter(Boolean));
    for (const [nuri, unsub] of docSubs) {
      if (!wanted.has(nuri)) {
        try {
          unsub();
        } catch {
          /* ignore */
        }
        docSubs.delete(nuri);
      }
    }
    for (const nuri of wanted) {
      if (docSubs.has(nuri)) continue;
      docSubs.set(nuri, subscribeDoc(nuri, () => void reread()));
    }
  }

  /** Read (union + shape filter) the CURRENT doc set and publish a snapshot.
   *  Derives isPending/isSuccess from the barrier + whether the read rendered. */
  async function readAndPublish(docs: Nuri[], token: number): Promise<void> {
    let subjects: UnionSubject[];
    try {
      subjects = await readUnion(docs);
    } catch (error) {
      if (token !== refreshToken) return;
      setSnapshot({ data: [], isPending: false, isSuccess: false, isError: true, error });
      return;
    }
    if (token !== refreshToken) return; // superseded by a newer refresh/reread
    const data = subjects.filter((s) => matchesShape(s, typeIris));
    const past = barrierReached(docs);
    setSnapshot({
      data,
      isPending: !past,
      isSuccess: past,
      isError: false,
      error: undefined,
    });
  }

  /** Full cycle: resolve the scope, (re)establish container subs, open the docs
   *  (await the barrier), sync per-doc subs, then read + publish. */
  async function refresh(): Promise<void> {
    const token = ++refreshToken;
    try {
      await ensureContainerSubs();
      const docs = await resolveDocs();
      if (token !== refreshToken) return;
      syncDocSubs(docs);
      // Open/await the barrier (first State per doc, or timed-out). No-op once open.
      await ensureReposOpen(docs);
      if (token !== refreshToken) return;
      await readAndPublish(docs, token);
    } catch (error) {
      if (token !== refreshToken) return;
      setSnapshot({ data: [], isPending: false, isSuccess: false, isError: true, error });
    }
  }

  /** A push on an already-open doc: re-read the CURRENT set only (no re-resolve,
   *  the set is unchanged). Reuses the docs we are subscribed to. */
  async function reread(): Promise<void> {
    const token = ++refreshToken;
    const docs = [...docSubs.keys()];
    await readAndPublish(docs, token);
  }

  function start(): void {
    if (started) return;
    started = true;
    // A cap that arrives ASYNCHRONOUSLY (an inbox deposit absorbed by the
    // consumer's `inbox.watch`) makes documents readable that were not. Without
    // this the view would stay stale until some unrelated push happened to fire —
    // so re-read whenever they change. This is the delivery channel key
    // ROTATION uses too, which is why keeping an access needs no subscription
    // obligation on the consumer's side.
    capsUnsub = getCaps().onChange(() => {
      if (!resolving) void refresh();
    });
    void refresh();
  }

  return {
    getSnapshot(): ShapeQuery<T> {
      return snapshot as unknown as ShapeQuery<T>;
    },
    subscribe(onChange: () => void): () => void {
      listeners.add(onChange);
      start();
      return () => {
        listeners.delete(onChange);
        if (listeners.size === 0) {
          // Last listener gone → tear down the underlying subscriptions. A later
          // subscribe restarts a fresh cycle.
          for (const u of docSubs.values()) {
            try {
              u();
            } catch {
              /* ignore */
            }
          }
          for (const u of containerSubs.values()) {
            try {
              u();
            } catch {
              /* ignore */
            }
          }
          docSubs.clear();
          containerSubs.clear();
          if (capsUnsub) {
            capsUnsub();
            capsUnsub = null;
          }
          started = false;
        }
      };
    },
    refetch(): void {
      start();
      void refresh();
    },
  };
}