ng-eventually/packages/polyfill/e2e/reactivity-doc-subscribe.ts

/**
 * DECISIVE real-broker determination: does `doc_subscribe` actually PUSH when a
 * subscribed document is written?
 *
 * This is the reactive-layer coverage whose ABSENCE let a reactivity bug ship: the
 * app's whole read-model reactivity rests on `subscribeDoc(nuri, cb)` (the polyfill
 * wrapper over `ng.doc_subscribe`, `src/subscribe.ts`) firing `cb` again on every
 * commit to the doc. Two pushes are load-bearing in production and were reported as
 * NOT firing:
 *   (SELF)  a session's own `sparqlUpdate` to a doc it subscribes to.
 *   (CROSS) another session writes to a doc the first session subscribes to.
 *
 * This runner exercises BOTH against the REAL broker, through the SAME public
 * surface the app uses — `subscribeDoc` (via the harness's `stateProbe*` bridge,
 * which passes the raw `AppResponse` straight through the polyfill wrapper),
 * `docs.docCreate`, and `docs.sparqlUpdate` (`writeTo`). It records EVERY push as a
 * typed event (`{ typeKey: "State" | "Patch" | "TabInfo" | …, elapsedMs }`) so the
 * verdict is the ground truth "did the subscription callback fire again", not a
 * re-read of the document. Each wait is a single event-driven promise+timeout on the
 * push (NO re-read loop) — a timeout is a DEFINITE "did-not-fire", not a flaky miss.
 *
 * Standalone (NOT `bun test`). Run:
 *   bun run e2e/reactivity-doc-subscribe.ts
 *   (or `bun run test:e2e:reactivity` from packages/polyfill)
 *
 * It reuses the exact real-broker plumbing of run.ts / broker.ts: the dedicated lib
 * wallet, the broker iframe, `window.__sdk`. The CROSS case opens a SECOND page on
 * the SAME persistent wallet context — a second concurrent verifier session on one
 * shared wallet (as faithfulReconnect does) — and writes from it.
 */

import type { Frame, Page, BrowserContext } from "playwright";
import {
  buildBundle,
  serveHarness,
  ensureWallet,
  launchWalletContext,
  setupBrokerPage,
} from "./broker";

type Check = { name: string; ok: boolean; detail?: string };
const results: Check[] = [];
function record(name: string, ok: boolean, detail?: string): void {
  results.push({ name, ok, detail });
  console.log(`  [${ok ? "PASS" : "FAIL"}] ${name}${detail ? " — " + detail : ""}`);
}

type Event = { typeKey: string; elapsedMs: number };

// Call a bridge method inside a given iframe.
function sdk<T>(frame: Frame, method: string, ...args: unknown[]): Promise<T> {
  return frame.evaluate(
    ([m, a]) => (window as any).__sdk[m as string](...(a as unknown[])),
    [method, args] as const,
  ) as Promise<T>;
}

/**
 * The decisive wait: resolve TRUE as soon as the probe's recorded push count grows
 * past `base` (the subscription callback fired again), or FALSE on timeout. This is
 * a promise+timeout on the PUSH itself — it polls only the in-memory event counter
 * the `subscribeDoc` callback writes, NEVER re-reads the document. A FALSE here is a
 * definite non-delivery within the window, not a missed re-read.
 */
async function waitForPush(frame: Frame, base: number, timeoutMs: number): Promise<boolean> {
  try {
    await frame.waitForFunction(
      (b) => (window as any).__sdk.stateProbeEvents().length > (b as number),
      base,
      { timeout: timeoutMs },
    );
    return true;
  } catch {
    return false; // timed out → the callback did NOT fire again within the window
  }
}

const seq = (events: Event[]): string =>
  events.length ? events.map((e) => `${e.typeKey}@${e.elapsedMs}ms`).join(" → ") : "(none)";

async function openSession(
  ctx: BrowserContext,
  url: string,
  tag: string,
): Promise<{ page: Page; frame: Frame; sessionId: string }> {
  const page = await ctx.newPage();
  page.on("pageerror", (e) => console.error(`[iframe error:${tag}]`, e.message));
  page.on("console", (m) => {
    const t = m.text();
    // Surface the polyfill's own "doc_subscribe FIRE" diagnostic (subscribe.ts) if
    // access logging happens to be on — an independent confirmation of a push.
    if (m.type() === "error") console.error(`[iframe console:${tag}]`, t);
    else if (t.includes("doc_subscribe FIRE")) console.log(`[${tag}] ${t}`);
  });
  const frame = await setupBrokerPage(page, url);
  await frame.waitForFunction(() => (window as any).__sdk !== undefined, { timeout: 30000 });
  await frame.waitForFunction(() => (window as any).__sdk.status() === "connected", {
    timeout: 60000,
  });
  const info = await sdk<{ session_id: string } | null>(frame, "sessionInfo");
  const sessionId = info?.session_id ?? "(none)";
  console.log(`[session:${tag}] connected — session_id=${sessionId}`);
  return { page, frame, sessionId };
}

const SELF_TIMEOUT_MS = 10000;
const CROSS_TIMEOUT_MS = 15000;
const STATE_TIMEOUT_MS = 20000;

async function main(): Promise<void> {
  console.log("[reactivity] building SDK page bundle...");
  buildBundle();
  console.log("[reactivity] ensuring dedicated lib wallet...");
  await ensureWallet();
  const { url, close: closeServer } = await serveHarness();
  console.log(`[reactivity] harness served at ${url}`);

  let ctx: BrowserContext | null = null;
  try {
    ctx = await launchWalletContext();

    // ── Session A (the subscriber for both cases) ────────────────────────────
    const A = await openSession(ctx, url, "A");

    // ════════════════════════════════════════════════════════════════════════
    // CASE 1 — SELF: A subscribes to D, then A itself writes to D.
    // ════════════════════════════════════════════════════════════════════════
    console.log("\n── CASE 1: SELF (single session — own write to own subscribed doc) ──");
    {
      const doc = await sdk<string>(A.frame, "docCreate");
      console.log(`  [SELF] created doc D = ${doc}`);
      await sdk(A.frame, "stateProbeSubscribe", doc);

      // Wait for the initial State (the sync barrier). TabInfo may precede it.
      const gotState = await (async () => {
        try {
          await A.frame.waitForFunction(
            () => (window as any).__sdk.stateProbeStateCount() >= 1,
            { timeout: STATE_TIMEOUT_MS },
          );
          return true;
        } catch {
          return false;
        }
      })();
      const afterSubscribe = await sdk<Event[]>(A.frame, "stateProbeEvents");
      console.log(`  [SELF] pushes after subscribe: ${seq(afterSubscribe)}`);
      record(
        "SELF: initial State push arrives on subscribe (baseline sanity)",
        gotState && afterSubscribe.some((e) => e.typeKey === "State"),
        `sequence=${seq(afterSubscribe)}`,
      );

      // Now the decisive write: A's OWN sparqlUpdate to D.
      const preWrite = afterSubscribe.length;
      console.log(`  [SELF] A writes to D (own sparqlUpdate); waiting ≤${SELF_TIMEOUT_MS}ms for a push…`);
      await sdk(A.frame, "writeTo", doc, "self-1");
      const fired = await waitForPush(A.frame, preWrite, SELF_TIMEOUT_MS);

      const afterWrite = await sdk<Event[]>(A.frame, "stateProbeEvents");
      const newEvents = afterWrite.slice(preWrite);
      console.log(`  [SELF] pushes AFTER own write: ${seq(newEvents)}`);
      console.log(`  [SELF] VERDICT: callback ${fired ? "FIRED" : "did NOT fire"} within ${SELF_TIMEOUT_MS}ms`);
      record(
        `SELF: subscription callback fires on the session's OWN write (≤${SELF_TIMEOUT_MS}ms)`,
        fired,
        `newPushes=${seq(newEvents)}`,
      );
      await sdk(A.frame, "stateProbeStop");
    }

    // ════════════════════════════════════════════════════════════════════════
    // CASE 2 — CROSS-SESSION: A subscribes to D2; a SECOND session B (same shared
    // wallet, own concurrent verifier session) writes to D2.
    // ════════════════════════════════════════════════════════════════════════
    console.log("\n── CASE 2: CROSS-SESSION (session B writes to a doc session A subscribes to) ──");
    let B: { page: Page; frame: Frame; sessionId: string } | null = null;
    try {
      B = await openSession(ctx, url, "B");
    } catch (e: any) {
      console.log(`  [CROSS] COULD-NOT-TEST: second concurrent session on the shared wallet failed to open: ${String(e?.message ?? e)}`);
      record(
        "CROSS: second concurrent session opened on the shared wallet",
        false,
        `open failed: ${String(e?.message ?? e)} — see Festipod multibrowser harness as the alternative venue`,
      );
    }

    if (B) {
      // NB: `session_id` is a PER-PAGE local verifier counter (each fresh iframe
      // numbers its first session "1"), so it is NOT a global identifier and cannot
      // be used to prove distinctness. The REAL proof that A and B are two separate
      // verifier sessions is behavioural: B's write reaches A only after a broker
      // round-trip (a delayed Patch), not as an instant same-session echo.
      console.log(
        `  [CROSS] both pages connected — A.session=${A.sessionId} B.session=${B.sessionId} (per-page local counter; distinctness shown by the cross-broker propagation below)`,
      );
      record(
        "CROSS: a second concurrent page/session is open on the same shared wallet",
        true,
        `A=${A.sessionId} B=${B.sessionId} (session_id is a per-page counter, not a global id)`,
      );

      // A creates D2 and subscribes.
      const doc2 = await sdk<string>(A.frame, "docCreate");
      console.log(`  [CROSS] A created doc D2 = ${doc2}`);
      await sdk(A.frame, "stateProbeSubscribe", doc2);
      const gotState2 = await (async () => {
        try {
          await A.frame.waitForFunction(
            () => (window as any).__sdk.stateProbeStateCount() >= 1,
            { timeout: STATE_TIMEOUT_MS },
          );
          return true;
        } catch {
          return false;
        }
      })();
      const afterSub2 = await sdk<Event[]>(A.frame, "stateProbeEvents");
      console.log(`  [CROSS] A pushes after subscribe: ${seq(afterSub2)}`);
      record(
        "CROSS: A receives its initial State on D2 (baseline sanity)",
        gotState2 && afterSub2.some((e) => e.typeKey === "State"),
        `sequence=${seq(afterSub2)}`,
      );

      // B writes to D2. Capture a write failure (e.g. RepoNotFound) explicitly —
      // it would mean B cannot reach A's doc, which is itself a determination.
      const preCross = afterSub2.length;
      let writeThrew: string | null = null;
      // Cross-session writes to a doc created by ANOTHER session can be slow: B must
      // sync/open D2's repo before it can commit. Time it separately so the push
      // latency is reported relative to when B's write actually LANDED, not to
      // subscribe time.
      console.log(`  [CROSS] B writes to D2 from its own session…`);
      const tWriteStart = Date.now();
      try {
        await sdk(B.frame, "writeTo", doc2, "cross-1");
      } catch (e: any) {
        writeThrew = String(e?.message ?? e);
        console.log(`  [CROSS] B's write THREW: ${writeThrew}`);
      }
      const writeMs = Date.now() - tWriteStart;
      record("CROSS: session B's write to D2 did not throw", writeThrew === null, writeThrew ? writeThrew : `landed in ${writeMs}ms`);

      console.log(`  [CROSS] B's write returned in ${writeMs}ms; now waiting ≤${CROSS_TIMEOUT_MS}ms for A's push…`);
      const tWaitStart = Date.now();
      const crossFired = writeThrew ? false : await waitForPush(A.frame, preCross, CROSS_TIMEOUT_MS);
      const pushAfterWriteMs = Date.now() - tWaitStart;
      const afterCross = await sdk<Event[]>(A.frame, "stateProbeEvents");
      const crossNew = afterCross.slice(preCross);
      console.log(`  [CROSS] A pushes AFTER B's write: ${seq(crossNew)}`);
      console.log(
        `  [CROSS] VERDICT: A's callback ${crossFired ? `FIRED (${pushAfterWriteMs}ms after B's write landed)` : "did NOT fire"} within ${CROSS_TIMEOUT_MS}ms${writeThrew ? " (B's write threw first)" : ""}`,
      );
      record(
        `CROSS: A's subscription callback fires on B's write (≤${CROSS_TIMEOUT_MS}ms after B's write landed)`,
        crossFired,
        `newPushes=${seq(crossNew)} (B write took ${writeMs}ms; push ${crossFired ? pushAfterWriteMs + "ms after" : "not seen"})${writeThrew ? ` — B write threw: ${writeThrew}` : ""}`,
      );
      await sdk(A.frame, "stateProbeStop");
    }
  } finally {
    try { if (ctx) await ctx.close(); } catch { /* ignore */ }
    closeServer();
  }

  // ── Determination summary (not a pass/fail gate — this is a probe) ──────────
  console.log("\n══ doc_subscribe delivery determination ══");
  for (const r of results) console.log(`  [${r.ok ? "PASS" : "FAIL"}] ${r.name}${r.detail ? " — " + r.detail : ""}`);
  const self = results.find((r) => r.name.startsWith("SELF: subscription callback fires"));
  const cross = results.find((r) => r.name.startsWith("CROSS: A's subscription callback fires"));
  console.log("\n  SELF  →", self ? (self.ok ? "FIRES" : "DOES-NOT-FIRE") : "could-not-test");
  console.log("  CROSS →", cross ? (cross.ok ? "FIRES" : "DOES-NOT-FIRE") : "could-not-test");
}

main().catch((e) => {
  console.error("[reactivity] fatal:", e);
  process.exit(1);
});