feat(analytics): internal refetch + uncached request mode (poll 1/4)

packages/appkit-ui/src/react/hooks/__tests__/analytics-request-store.test.ts

@@ -0,0 +1,276 @@
+import { afterEach, beforeEach, describe, expect, test, vi } from "vitest";
+
+let capturedCallbacks: {
+  onMessage?: (msg: { data: string }) => void;
+  onError?: (err: Error) => void;
+  signal?: AbortSignal;
+} = {};
+
+const mockFetchArrow = vi.fn();
+
+// Mock connectSSE to capture calls and track uncached requests.
+const mockConnectSSE = vi.fn((args: any): unknown => {
+  capturedCallbacks = {
+    onMessage: args?.onMessage,
+    onError: args?.onError,
+    signal: args?.signal,
+  };
+  return () => {};
+});
+
+const mockProcessArrowBuffer = vi.fn();
+
+vi.mock("@/js", () => ({
+  connectSSE: (...args: unknown[]) => mockConnectSSE(...(args as [any])),
+  ArrowClient: {
+    fetchArrow: (...args: unknown[]) => mockFetchArrow(...args),
+    processArrowBuffer: (...args: unknown[]) => mockProcessArrowBuffer(...args),
+  },
+}));
+
+import {
+  getSnapshot,
+  refetch,
+  resetAnalyticsRequestStore,
+  retain,
+  start,
+  subscribe,
+} from "../analytics-request-store";
+
+const JSON_OPTS = {
+  url: "/api/analytics/query/q",
+  payload: JSON.stringify({ parameters: null, format: "JSON_ARRAY" }),
+  format: "JSON_ARRAY",
+};
+
+describe("analytics-request-store", () => {
+  beforeEach(() => {
+    vi.clearAllMocks();
+    capturedCallbacks = {};
+    resetAnalyticsRequestStore();
+  });
+
+  afterEach(() => {
+    vi.unstubAllGlobals();
+  });
+
+  describe("retain and start", () => {
+    test("retain with autoStart:false does not start the request", () => {
+      retain("k", JSON_OPTS, false);
+      expect(mockConnectSSE).not.toHaveBeenCalled();
+    });
+
+    test("start on an autoStart:false entry begins the request", () => {
+      retain("k", JSON_OPTS, false);
+      expect(mockConnectSSE).not.toHaveBeenCalled();
+
+      start("k");
+      expect(mockConnectSSE).toHaveBeenCalledTimes(1);
+    });
+
+    test("retain with default autoStart:true starts the request immediately", () => {
+      retain("k", JSON_OPTS);
+      expect(mockConnectSSE).toHaveBeenCalledTimes(1);
+    });
+  });
+
+  describe("refetch", () => {
+    test("refetch aborts in-flight run and restarts with same options", () => {
+      retain("k", JSON_OPTS);
+      const firstSignal = capturedCallbacks.signal;
+
+      // refetch should call connectSSE a second time.
+      refetch("k");
+
+      expect(mockConnectSSE).toHaveBeenCalledTimes(2);
+
+      // The second call is a fresh start with a distinct signal.
+      const secondCall = mockConnectSSE.mock.calls[1];
+      const secondSignal = secondCall[0].signal;
+
+      expect(secondSignal).not.toBe(firstSignal);
+    });
+
+    test("refetch marks the key for uncached execution on next start", () => {
+      const release = retain("k", JSON_OPTS, false);
+      expect(mockConnectSSE).not.toHaveBeenCalled();
+
+      // Mark for uncached.
+      refetch("k");
+
+      // start() is called; the entry should be re-invoked with skipCache set.
+      // Check that the payload now includes skipCache:true.
+      expect(mockConnectSSE).toHaveBeenCalledTimes(1);
+
+      const call = mockConnectSSE.mock.calls[0];
+      const payload = call[0].payload;
+      const parsed = JSON.parse(payload);
+      expect(parsed.skipCache).toBe(true);
+
+      release();
+    });
+
+    test("skipCache flag is included in payload for SSE requests", () => {
+      const sseOpts = {
+        url: "/api/analytics/query/test",
+        payload: JSON.stringify({ parameters: { x: 1 }, format: "JSON_ARRAY" }),
+        format: "JSON_ARRAY",
+        skipCache: true,
+      };
+
+      retain("ssekey", sseOpts);
+
+      expect(mockConnectSSE).toHaveBeenCalledTimes(1);
+      const call = mockConnectSSE.mock.calls[0];
+      const payload = call[0].payload;
+      const parsed = JSON.parse(payload);
+
+      expect(parsed).toEqual({
+        parameters: { x: 1 },
+        format: "JSON_ARRAY",
+        skipCache: true,
+      });
+    });
+
+    test("skipCache flag is included in payload for ARROW_STREAM requests", async () => {
+      const fakeTable = { numRows: 1, schema: { fields: [] } };
+      const fakeBytes = new Uint8Array([1, 2, 3]);
+      const fetchMock = vi.fn().mockResolvedValue({
+        ok: true,
+        arrayBuffer: async () => fakeBytes.buffer,
+        headers: { get: () => null },
+      });
+      vi.stubGlobal("fetch", fetchMock);
+      mockProcessArrowBuffer.mockResolvedValueOnce(fakeTable);
+
+      const arrowOpts = {
+        url: "/api/analytics/query/arrow_test",
+        payload: JSON.stringify({ parameters: null, format: "ARROW_STREAM" }),
+        format: "ARROW_STREAM",
+        skipCache: true,
+      };
+
+      retain("arrowkey", arrowOpts);
+
+      // Wait for the arrow fetch to be called.
+      await new Promise((r) => setTimeout(r, 10));
+
+      expect(fetchMock).toHaveBeenCalledTimes(1);
+      const [, init] = fetchMock.mock.calls[0];
+      const payload = JSON.parse(init.body);
+
+      expect(payload).toEqual({
+        parameters: null,
+        format: "ARROW_STREAM",
+        skipCache: true,
+      });
+    });
+
+    test("uncached mark is consumed on next start and does not persist", () => {
+      const release1 = retain("k", JSON_OPTS, false);
+
+      // Mark for uncached.
+      refetch("k");
+      expect(mockConnectSSE).toHaveBeenCalledTimes(1);
+
+      let payload = JSON.parse(mockConnectSSE.mock.calls[0][0].payload);
+      expect(payload.skipCache).toBe(true);
+
+      release1();
+
+      // Re-retain the same key without refetch; the uncached mark should not persist.
+      resetAnalyticsRequestStore();
+      vi.clearAllMocks();
+
+      const release2 = retain("k", JSON_OPTS);
+      expect(mockConnectSSE).toHaveBeenCalledTimes(1);
+
+      payload = JSON.parse(mockConnectSSE.mock.calls[0][0].payload);
+      expect(payload.skipCache).toBeUndefined();
+
+      release2();
+    });
+  });
+
+  describe("retain with autoStart:false and refetch composition", () => {
+    test("retain(autoStart:false) + refetch defers start but executes with uncached on refetch", () => {
+      const release = retain("deferred", JSON_OPTS, false);
+
+      // No request yet.
+      expect(mockConnectSSE).not.toHaveBeenCalled();
+
+      // refetch marks for uncached and starts.
+      refetch("deferred");
+
+      expect(mockConnectSSE).toHaveBeenCalledTimes(1);
+      const payload = JSON.parse(mockConnectSSE.mock.calls[0][0].payload);
+      expect(payload.skipCache).toBe(true);
+
+      release();
+    });
+
+    test("multiple refetch calls re-invoke the runner each time", () => {
+      retain("multi", JSON_OPTS, false);
+
+      refetch("multi");
+      expect(mockConnectSSE).toHaveBeenCalledTimes(1);
+
+      refetch("multi");
+      expect(mockConnectSSE).toHaveBeenCalledTimes(2);
+
+      refetch("multi");
+      expect(mockConnectSSE).toHaveBeenCalledTimes(3);
+
+      // All three should have skipCache.
+      for (let i = 0; i < 3; i++) {
+        const payload = JSON.parse(mockConnectSSE.mock.calls[i][0].payload);
+        expect(payload.skipCache).toBe(true);
+      }
+    });
+  });
+
+  describe("snapshot subscribers", () => {
+    test("refetch triggers updates to subscribers via snapshot", () => {
+      const release = retain("snap", JSON_OPTS, false);
+      const listener = vi.fn();
+      subscribe("snap", listener);
+
+      refetch("snap");
+
+      // Refetch should have triggered a start → loading snapshot notification.
+      expect(listener).toHaveBeenCalled();
+      expect(getSnapshot("snap").loading).toBe(true);
+
+      release();
+    });
+  });
+
+  describe("uncached mark does not leak from a no-op refetch", () => {
+    test("refetch on a key with no live entry does not skipCache a later retain", () => {
+      // No entry exists for "ghost": store.start no-ops, so the runner never
+      // runs and never consumes the mark. It must not stick to the next retain.
+      refetch("ghost");
+      expect(mockConnectSSE).not.toHaveBeenCalled();
+
+      const release = retain("ghost", JSON_OPTS);
+      expect(mockConnectSSE).toHaveBeenCalledTimes(1);
+
+      const payload = JSON.parse(mockConnectSSE.mock.calls[0][0].payload);
+      expect(payload.skipCache).toBeUndefined();
+
+      release();
+    });
+
+    test("resetAnalyticsRequestStore clears any pending uncached mark", () => {
+      refetch("ghost");
+      resetAnalyticsRequestStore();
+      vi.clearAllMocks();
+
+      const release = retain("ghost", JSON_OPTS);
+      const payload = JSON.parse(mockConnectSSE.mock.calls[0][0].payload);
+      expect(payload.skipCache).toBeUndefined();
+
+      release();
+    });
+  });
+});

packages/appkit-ui/src/react/hooks/analytics-request-store.ts

@@ -34,6 +34,8 @@ interface AnalyticsRequestOptions {
   payload: string;
   /** Response format; selects the transport. */
   format: string;
+  /** @internal Force skip cache on next execution; cleared after each run. */
+  skipCache?: boolean;
 }
 
 /** Immutable per-key request state; mirrors the hook's public result shape. */
@@ -162,15 +164,34 @@ async function fetchArrowDirect(
  * format-appropriate transport, reporting state through `controls.patch`.
  */
 function runAnalyticsRequest(
+  cacheKey: string,
   options: AnalyticsRequestOptions,
 ): RequestRunner<AnalyticsRequestSnapshot> {
   return (controls) => {
     controls.patch(LOADING_SNAPSHOT);
 
+    // Consume any one-shot uncached mark for this key.
+    const shouldSkipCache = options.skipCache || uncachedKeys.has(cacheKey);
+    if (uncachedKeys.has(cacheKey)) {
+      uncachedKeys.delete(cacheKey);
+    }
+
+    // Build the request payload, injecting skipCache when needed.
+    let requestPayload = options.payload;
+    if (shouldSkipCache) {
+      try {
+        const parsed = JSON.parse(options.payload);
+        requestPayload = JSON.stringify({ ...parsed, skipCache: true });
+      } catch {
+        // If payload parsing fails, use the original payload and let it fail downstream.
+        requestPayload = options.payload;
+      }
+    }
+
     // ARROW_STREAM: the server streams raw Arrow IPC bytes back on the query
     // response body (no SSE). Fetch and decode directly.
     if (options.format === "ARROW_STREAM") {
-      void fetchArrowDirect(controls, options);
+      void fetchArrowDirect(controls, { ...options, payload: requestPayload });
       return;
     }
 
@@ -195,7 +216,7 @@ function runAnalyticsRequest(
 
     connectSSE({
       url: options.url,
-      payload: options.payload,
+      payload: requestPayload,
       signal: controls.signal,
       onMessage: (message) =>
         handleAnalyticsSseMessage(message.data, sseContext),
@@ -206,6 +227,13 @@ function runAnalyticsRequest(
 
 const store = createRequestStore<AnalyticsRequestSnapshot>(EMPTY_SNAPSHOT);
 
+/**
+ * Internal tracking of which cache keys should force uncached execution on their
+ * next run. Entries are removed after the run to avoid sticking to subsequent
+ * invocations with the same key.
+ */
+const uncachedKeys = new Set<string>();
+
 /**
  * Register a subscriber for `key`, starting the shared request on first use.
  * Returns a `release` function that must be called on unmount.
@@ -215,12 +243,35 @@ export function retain(
   options: AnalyticsRequestOptions,
   autoStart = true,
 ): () => void {
-  return store.retain(key, runAnalyticsRequest(options), autoStart);
+  return store.retain(key, runAnalyticsRequest(key, options), autoStart);
 }
 
 export const start = store.start;
 export const subscribe = store.subscribe;
 export const getSnapshot = store.getSnapshot;
 
+/**
+ * Internal API: re-run a request with cache bypassed.
+ *
+ * Marks the key to force uncached execution on its next run, then calls
+ * store.start() to abort any in-flight run and restart. The mark is consumed
+ * immediately by the runner, so it doesn't persist across multiple refetch
+ * calls or independent hook instances.
+ * @internal
+ */
+export function refetch(cacheKey: string): void {
+  uncachedKeys.add(cacheKey);
+  store.start(cacheKey);
+  // `store.start` is a synchronous no-op when the key has no live entry (e.g.
+  // the last subscriber released and teardown ran). In that case the runner
+  // never ran and never consumed the mark, so clear it here to avoid leaking
+  // skipCache onto a later `retain` of the same key. When the runner did run
+  // it already deleted the mark synchronously, so this is a harmless no-op.
+  uncachedKeys.delete(cacheKey);
+}
+
 /** Test-only: abort every in-flight request and clear the store. */
-export const resetAnalyticsRequestStore = store.reset;
+export function resetAnalyticsRequestStore(): void {
+  uncachedKeys.clear();
+  store.reset();
+}