ng-eventually/packages/sdk/test/watch-shape.test.ts

/**
 * watch-shape.test.ts — behavioural tests for `watchShape` (src/watch-shape.ts),
 * against a STATEFUL fake `ng` with a CONTROLLABLE `doc_subscribe`.
 *
 * The fake emulates just enough of the broker:
 *   - `doc_create` mints monotonic doc NURIs.
 *   - `sparql_update` parses the shim account writes + the per-entity index
 *     `contains` append + arbitrary anchored triple writes into an in-memory quad
 *     store (same tolerant parser shape as store-registry.test / read-model.test).
 *   - `sparql_query` answers the shim account SELECT, the scope-index `contains`
 *     SELECT, and the anchored per-doc `?s ?p ?o` read (readUnion) — each scoped to
 *     the anchor graph.
 *   - `doc_subscribe` models the platform push order TabInfo→State: on subscribe it
 *     records the callback and fires a `TabInfo` immediately, but the sync BARRIER
 *     `State` is fired only when the TEST releases it (`fireState`) — so we can
 *     assert isPending BEFORE the barrier and isSuccess AFTER. A later write to a
 *     subscribed doc fires a `Patch` push (reactivity).
 *
 * These prove the four distinctions the surface exists for:
 *   (a) isPending at first, isSuccess after the first State (barrier);
 *   (b) isSuccess + data:[] on a synced-but-EMPTY scope (the key distinction);
 *   (c) a write then push → data updates (reactivity, no polling);
 *   (d) timed-out → isSuccess (best-effort), NOT isError.
 */

import { describe, it, expect, mock, beforeEach, afterEach, afterAll } from "bun:test";
import { watchShape } from "../src/surface/watch-shape";
import { configure } from "../src/index";
import { configureStoreRegistry, setCurrentUser } from "../src/shared-wallet/bootstrap";
import { resetCaps, resetConfig, resetStoreRegistry } from "../src/shared-wallet/bootstrap";
import { resetRegistryCache, createEntityDoc } from "../src/shared-wallet/account-registry";
import { resetOpenedRepos, setOpenTimeoutForTests, getSyncState } from "../src/emulated-verifier/open-repo";

const TYPE = "http://www.w3.org/1999/02/22-rdf-syntax-ns#type";
const FP = "http://festipod.org/";
const SESSION = { sessionId: "sid-ws", privateStoreId: "PRIV-WS" };

interface Quad { g: string; s: string; p: string; o: string }

/** Reverse of escapeLiteral: single left-to-right pass over `\x`. */
function unescapeLiteral(s: string): string {
  let out = "";
  for (let i = 0; i < s.length; i++) {
    if (s[i] === "\\" && i + 1 < s.length) {
      const next = s[++i];
      out += next === "n" ? "\n" : next === "r" ? "\r" : next === "t" ? "\t" : next!;
    } else out += s[i];
  }
  return out;
}

interface SubRec { nuri: string; cb: (r: unknown) => void }

/**
 * The stateful fake with a controllable doc_subscribe. `holdState: true` means a
 * fresh subscription does NOT auto-fire its `State` — the test fires it via
 * `fireState(nuri)`. `holdState: false` (default) auto-fires `State` on subscribe
 * (synced immediately), which is the convenient mode for the reactivity/empty cases.
 */
function makeFake(opts?: { holdState?: boolean }) {
  const quads: Quad[] = [];
  let docCounter = 0;
  const subs: SubRec[] = [];
  const hold = opts?.holdState ?? false;
  // Nuris whose barrier `State` has been released (auto-fire on future subscribe).
  const released = new Set<string>();
  // The shim ANCHOR (private-store-root) is ALWAYS loaded/synced on the real broker
  // (the store repo is bootstrapped at connect), so its barrier `State` is always
  // available. `resolveAccount`/`ensureAccount` now open it (the cold-start heal)
  // before touching the shim — pre-release it here so `holdState` (which gates the
  // per-ENTITY docs the tests control) never blocks the anchor open. This mirrors the
  // real invariant the production heal relies on.
  released.add(`did:ng:${SESSION.privateStoreId}`);
  let releaseEverything = false;

  const doc_create = mock(async () => `did:ng:o:doc${++docCounter}`);

  const sparql_update = mock(async (...a: unknown[]) => {
    const query = a[1] as string;
    const anchor = a[2] as string | undefined;
    const gm = query.match(/GRAPH <([^>]+)>\s*\{([\s\S]*)\}/);
    let g: string;
    let body: string;
    if (gm) {
      g = gm[1]!;
      body = gm[2]!;
    } else {
      if (!anchor) return undefined;
      g = anchor;
      body = query.replace(/^\s*INSERT DATA\s*\{/, "").replace(/\}\s*$/, "");
    }
    const sm = body.match(/<([^>]+)>/);
    if (!sm) return undefined;
    const s = sm[1]!;
    const pairRe = /(?:a|<([^>]+)>)\s+(?:"((?:[^"\\]|\\.)*)"|<([^>]+)>)/g;
    let m: RegExpExecArray | null;
    const after = body.slice(body.indexOf(sm[0]) + sm[0].length);
    while ((m = pairRe.exec(after)) !== null) {
      const p = m[1] ?? "urn:ng-eventually:shim:Account";
      const o = m[2] !== undefined ? unescapeLiteral(m[2]) : (m[3] ?? "");
      quads.push({ g, s, p, o });
      // The doc-shim (named by the write-once pointer triple) is INFRASTRUCTURE, like
      // the store-root: `doc_create` bootstrapped it into the session, so its barrier
      // `State` is immediately available. Pre-release it so `holdState` (which gates the
      // per-ENTITY docs the tests control) never blocks the doc-shim open. The pointer is
      // published BEFORE the doc-shim barrier open (resolveShimDoc first-login order).
      if (p === "urn:ng-eventually:shim:shimDoc") {
        released.add(o);
        for (const sub of subs) if (sub.nuri === o) sub.cb({ V0: { State: {} } });
      }
    }
    // A write to a subscribed doc fires a Patch push (reactivity signal).
    for (const sub of subs) {
      if (sub.nuri === g) sub.cb({ V0: { Patch: {} } });
    }
    return undefined;
  });

  const sparql_query = mock(async (...a: unknown[]) => {
    const query = a[1] as string;
    const anchor = a[3] as string | undefined;
    // Pointer SELECT (store-root -> doc-shim).
    if (query.includes("<urn:ng-eventually:shim:shimDoc>")) {
      const bindings = quads
        .filter((q) => q.g === anchor && q.p === "urn:ng-eventually:shim:shimDoc")
        .map((q) => ({ shimDoc: { value: q.o } }));
      return { results: { bindings } };
    }
    if (query.includes("<urn:ng-eventually:shim:id>")) {
      const subjM = query.match(
        /<([^>]+)>\s+a\s+<urn:ng-eventually:shim:Account>/,
      );
      const onlySubject = subjM ? subjM[1]! : null;
      const bySubject = new Map<string, Record<string, string>>();
      for (const q of quads) {
        if (q.g !== anchor) continue;
        if (onlySubject !== null && q.s !== onlySubject) continue;
        const rec = bySubject.get(q.s) ?? {};
        if (q.p === "urn:ng-eventually:shim:id") rec.id = q.o;
        if (q.p === "urn:ng-eventually:shim:docPublic") rec.docPublic = q.o;
        if (q.p === "urn:ng-eventually:shim:docProtected") rec.docProtected = q.o;
        if (q.p === "urn:ng-eventually:shim:docPrivate") rec.docPrivate = q.o;
        bySubject.set(q.s, rec);
      }
      const bindings = [...bySubject.values()]
        .filter((r) => r.id)
        .map((r) => ({
          id: { value: r.id! },
          docPublic: { value: r.docPublic ?? "" },
          docProtected: { value: r.docProtected ?? "" },
          docPrivate: { value: r.docPrivate ?? "" },
        }));
      return { results: { bindings } };
    }
    if (query.includes("<urn:ng-eventually:shim:inboxCap>")) {
      const bindings = quads
        .filter((q) => q.g === anchor && q.p === "urn:ng-eventually:shim:inboxCap")
        .map((q) => ({ c: { value: q.o } }));
      return { results: { bindings } };
    }
    if (query.includes("<urn:ng-eventually:shim:readCap>")) {
      const bindings = quads
        .filter((q) => q.g === anchor && q.p === "urn:ng-eventually:shim:readCap")
        .map((q) => ({ c: { value: q.o } }));
      return { results: { bindings } };
    }
    if (query.includes("<urn:ng-eventually:shim:contains>")) {
      const bindings = quads
        .filter((q) => q.g === anchor && q.p === "urn:ng-eventually:shim:contains")
        .map((q) => ({ e: { value: q.o } }));
      return { results: { bindings } };
    }
    // Anchored per-doc read (readUnion `SELECT ?s ?p ?o`).
    const bindings = quads
      .filter((q) => q.g === anchor)
      .map((q) => ({ s: { value: q.s }, p: { value: q.p }, o: { value: q.o } }));
    return { results: { bindings } };
  });

  const doc_subscribe = mock(
    async (nuri: string, _sid: string, cb: (r: unknown) => void) => {
      subs.push({ nuri, cb });
      // Platform pushes TabInfo FIRST (never the barrier).
      setTimeout(() => cb({ V0: { TabInfo: {} } }), 0);
      // Fire the barrier State if this fake auto-syncs, or if this nuri was already
      // released (so a doc subscribed AFTER a release still crosses the barrier).
      if (!hold || releaseEverything || released.has(nuri)) {
        setTimeout(() => cb({ V0: { State: {} } }), 0);
      }
      return () => {};
    },
  );

  /** Release the barrier for `nuri` (fire State now + auto-fire for future subs). */
  function fireState(nuri: string): void {
    released.add(nuri);
    for (const sub of subs) if (sub.nuri === nuri) sub.cb({ V0: { State: {} } });
  }
  /** Release the barrier for EVERY doc, present and future. */
  function releaseAll(): void {
    releaseEverything = true;
    for (const sub of subs) sub.cb({ V0: { State: {} } });
  }

  return {
    doc_create,
    sparql_update,
    sparql_query,
    doc_subscribe,
    _quads: quads,
    fireState,
    releaseAll,
    subs,
  };
}

function inject(ng: ReturnType<typeof makeFake>) {
  configure({ ng: ng as any, useShape: (() => {}) as any });
  configureStoreRegistry({
    getSession: async () => SESSION,
    normalizeId: (u: string) => u.trim().replace(/^@+/, "").toLowerCase(),
  });
  resetRegistryCache();
  resetOpenedRepos();
  resetCaps();
}

// Insert a triple straight into a doc's graph in the fake store (no push).
function seed(ng: ReturnType<typeof makeFake>, doc: string, p: string, o: string): void {
  ng._quads.push({ g: doc, s: doc, p, o });
}

const tick = () => new Promise((r) => setTimeout(r, 5));

// A minimal SHEX ShapeType pinning rdf:type to `${FP}Event`.
const EventShape = {
  shape: `${FP}EventShape`,
  schema: {
    [`${FP}EventShape`]: {
      iri: `${FP}EventShape`,
      predicates: [{ iri: TYPE, dataTypes: [{ literals: [`${FP}Event`], valType: "iri" }] }],
    },
  },
};

afterEach(() => {
  setCurrentUser(null);
});
afterAll(() => {
  resetConfig();
  resetStoreRegistry();
  resetRegistryCache();
  resetOpenedRepos();
  // The cap registry is process-wide: leaving caps behind would put the possession
  // gate in force for a suite that never declares any.
  resetCaps();
});

describe("watchShape", () => {
  it("(a) isPending at first, then isSuccess after the first State (barrier)", async () => {
    const ng = makeFake({ holdState: true });
    inject(ng);
    setCurrentUser("alice");
    // One protected entity doc for alice, carrying an Event triple.
    const doc = await createEntityDoc("alice", "protected");
    seed(ng, doc, TYPE, `${FP}Event`);
    seed(ng, doc, `${FP}title`, "Alpha");

    const obs = watchShape(EventShape, "protected");
    let notes = 0;
    const unsub = obs.subscribe(() => {
      notes += 1;
    });

    // Before the barrier: pending, no data.
    await tick();
    expect(obs.getSnapshot().isPending).toBe(true);
    expect(obs.getSnapshot().isSuccess).toBe(false);
    expect(obs.getSnapshot().data).toEqual([]);

    // Release the barrier for every opened doc (present + future) → synced.
    ng.releaseAll();
    await tick();
    await tick();
    await tick();

    const snap = obs.getSnapshot();
    expect(snap.isPending).toBe(false);
    expect(snap.isSuccess).toBe(true);
    expect(snap.isError).toBe(false);
    expect(snap.data.length).toBe(1);
    expect(snap.data[0]!.props[`${FP}title`]).toEqual(["Alpha"]);
    expect(notes).toBeGreaterThan(0);
    unsub();
  });

  it("(b) isSuccess + data:[] on a synced-but-EMPTY scope (the key distinction)", async () => {
    const ng = makeFake(); // auto-fires State → synced immediately
    inject(ng);
    setCurrentUser("bob");
    // bob has NO entity docs in this scope — the scope is genuinely empty.

    const obs = watchShape(EventShape, "protected");
    const unsub = obs.subscribe(() => {});
    await tick();
    await tick();

    const snap = obs.getSnapshot();
    expect(snap.isPending).toBe(false);
    expect(snap.isSuccess).toBe(true); // synced, NOT stuck pending
    expect(snap.isError).toBe(false);
    expect(snap.data).toEqual([]); // empty — distinguishable from "still syncing"
    unsub();
  });

  it("(c) a write then push updates data (reactivity, no polling)", async () => {
    const ng = makeFake(); // synced immediately
    inject(ng);
    setCurrentUser("carol");
    const doc = await createEntityDoc("carol", "protected");
    seed(ng, doc, TYPE, `${FP}Event`);
    seed(ng, doc, `${FP}title`, "One");

    const obs = watchShape(EventShape, "protected");
    const unsub = obs.subscribe(() => {});
    await tick();
    await tick();
    expect(obs.getSnapshot().data.length).toBe(1);

    // Write a SECOND event doc + fire the push via a write to the ALREADY-subscribed
    // doc. Because a new doc must appear in the set, write into the scope-INDEX
    // (createEntityDoc appends to it, and the index is subscribed → re-resolve).
    const doc2 = await createEntityDoc("carol", "protected");
    seed(ng, doc2, TYPE, `${FP}Event`);
    seed(ng, doc2, `${FP}title`, "Two");
    // createEntityDoc's index append fired a Patch on the index doc → re-resolve.
    await tick();
    await tick();

    const titles = obs
      .getSnapshot()
      .data.flatMap((s) => s.props[`${FP}title`] ?? [])
      .sort();
    expect(titles).toEqual(["One", "Two"]);
    // No setInterval anywhere — reactivity was push-driven.
    unsub();
  });

  it("(d) timed-out → isSuccess (best-effort), NOT isError", async () => {
    // A doc whose subscription NEVER pushes a `State`: open-repo's bounded fallback
    // fires and marks the nuri "timed-out" (NOT "synced"). We shrink the fallback to
    // a few ms so this is fast, and assert the barrier is genuinely reached via
    // timed-out (getSyncState === "timed-out") and that the snapshot maps that to
    // isSuccess, never isError.
    const ng = makeFake({ holdState: true }); // State is never released
    inject(ng);
    setOpenTimeoutForTests(20); // fallback fires quickly instead of after 8s
    setCurrentUser("dave");
    const doc = await createEntityDoc("dave", "protected");
    seed(ng, doc, TYPE, `${FP}Event`);
    seed(ng, doc, `${FP}title`, "Timed");

    const obs = watchShape(EventShape, "protected");
    const unsub = obs.subscribe(() => {});
    await tick();
    // Before the fallback fires: still pending (subscribed, no State).
    expect(obs.getSnapshot().isPending).toBe(true);

    // Let the bounded fallback elapse → open-repo marks each opened doc timed-out.
    await new Promise((r) => setTimeout(r, 60));
    await tick();
    await tick();

    // The entity doc's barrier resolved via timed-out (never a State).
    expect(getSyncState(doc)).toBe("timed-out");
    const snap = obs.getSnapshot();
    expect(snap.isError).toBe(false);
    expect(snap.isSuccess).toBe(true); // timed-out is best-effort success
    expect(snap.isPending).toBe(false);
    // The data still read (best-effort): the doc's triples resolved.
    expect(snap.data.length).toBe(1);
    unsub();
  });

});