feat(liveQuery): server function producers are declared live()

.changeset/live-query-implies-live.md

@@ -0,0 +1,5 @@
+---
+"@solidjs/router": patch
+---
+
+`liveQuery(fn, key)` now declares a server function `live()` itself, so `liveQuery(fn, key)` gets the transport's live lifecycle (live address, `Last-Event-ID` resume, response-lifetime deaths, `Retry-After`) without wrapping by hand. An explicit `live(fn)` passes through untouched. Hydration takeover forwards the served value to the first connection, so an unchanged value costs nothing on the wire.

src/data/liveQuery.ts

@@ -2,10 +2,10 @@ import { createSignal, type Signal } from "solid-js";
 import {
   getRequestEvent,
   getServerFunctionMetadata,
-  getServerFunctionRPC,
   isServer,
   isServerFunction
 } from "@solidjs/web";
+import { live } from "@solidjs/web/server-functions";
 import { getIntent } from "../routing.js";
 import { hashKey, matchKey, registerFlightDataHook, registerRevalidateHook } from "./query.js";
 
@@ -17,9 +17,16 @@ const PRELOAD_TIMEOUT = 5000;
 // computations meeting a branded iterable apply live SSR policy (document
 // face takes the first value; hydration re-runs the compute after adoption
 // to reconnect). liveQuery writes it on both faces itself — the server-side
-// resolved iterable and the client-side multicast iterable — so a producer
-// needs no separate live() declaration to be live here.
+// resolved iterable and the client-side multicast iterable. A server function
+// producer is declared live() by liveQuery (below); a plain producer needs no
+// declaration at all.
 const LIVE_SOURCE = Symbol.for("solid.LiveSource");
+// Stamped by hydration's takeover on the iterable the compute returned (the
+// subscriber iterable below): the value the page was served with. Forwarded
+// onto the channel's first live() iterable, which yields it first and names
+// it as its position, so a takeover whose value is unchanged costs nothing
+// on the wire.
+const LIVE_RESUME_FROM = Symbol.for("solid.LiveResumeFrom");
 
 export type LiveQueryStatus = "idle" | "connecting" | "connected" | "reconnecting" | "closed";
 
@@ -47,6 +54,8 @@ type Channel = {
   /** keep the record (latest value) past teardown — request-scoped channels
    * replay it to later consumers of the same render for value consistency */
   retain: boolean;
+  /** the hydrated value the first connection resumes from (see LIVE_RESUME_FROM) */
+  resume?: any;
 };
 
 const channelMap = new Map<string, Channel>();
@@ -96,6 +105,10 @@ async function connect(ch: Channel) {
           };
         } catch {}
       }
+      if (ch.resume !== undefined && result !== null && typeof result === "object") {
+        result[LIVE_RESUME_FROM] = ch.resume;
+        ch.resume = undefined;
+      }
       const it =
         result !== null && typeof result === "object" && result[Symbol.asyncIterator]
           ? result[Symbol.asyncIterator]()
@@ -179,7 +192,8 @@ function openChannel(
   key: string,
   call: () => any,
   report: boolean,
-  retain: boolean
+  retain: boolean,
+  resume?: any
 ): Channel {
   let ch = map.get(key);
   if (!ch) {
@@ -196,7 +210,8 @@ function openChannel(
       close: () => {},
       call,
       report,
-      retain
+      retain,
+      resume
     };
     map.set(key, ch);
     connect(ch);
@@ -208,11 +223,11 @@ function openChannel(
 // channel lazily (first pull connects — a hydration trace or speculative
 // call opens nothing), replay the latest value immediately, then follow
 // with latest-wins delivery.
-function subscriberIterable(open: () => Channel) {
-  return {
+function subscriberIterable(open: (resume?: any) => Channel) {
+  const iterable: any = {
     [LIVE_SOURCE]: true,
     [Symbol.asyncIterator]() {
-      const ch = open();
+      const ch = open(iterable[LIVE_RESUME_FROM]);
       ch.count++;
       let seen = 0;
       let released = false;
@@ -248,6 +263,7 @@ function subscriberIterable(open: () => Channel) {
       };
     }
   };
+  return iterable;
 }
 
 function reconnectChannel(ch: Channel) {
@@ -334,7 +350,8 @@ export type LiveFunction<T extends (...args: any) => any> = T extends (...args:
  *   definite rejection (4xx: the transport stamps HTTP statuses onto
  *   failures) ends the channel and surfaces the error to consumers, as does
  *   a first-connect failure.
- * - Server functions are declared GET at creation (like `query`). Live keys
+ * - Server functions are declared `live()` (which implies GET) at creation,
+ *   so the connection gets the transport's live lifecycle. Live keys
  *   are neither collected for single-flight nor swept after a mutation: the
  *   stream is the freshness mechanism, and the mutation reaches a live
  *   query through its producer. (The flight-data hook below still adopts a
@@ -359,13 +376,11 @@ export type LiveFunction<T extends (...args: any) => any> = T extends (...args:
  */
 export function liveQuery<T extends (...args: any) => any>(fn: T, name: string): LiveFunction<T> {
   hookRevalidate();
-  // liveQuery implies GET, exactly as query does: reads belong on the GET
-  // transport (cacheable URLs, no single-flight enveloping). An explicit
-  // GET(fn) or live(GET(fn)) declaration passes through untouched.
-  if (isServerFunction(fn) && !getServerFunctionMetadata(fn)?.method) {
-    const rpc = getServerFunctionRPC();
-    if (rpc) fn = rpc.GET(fn) as unknown as T;
-  }
+  // liveQuery implies live (which implies GET): live() brings the wire
+  // lifecycle — the live address, Last-Event-ID resume, response-lifetime
+  // deaths, Retry-After — so the channel multicasts a live connection. An
+  // explicit live() declaration passes through untouched.
+  if (isServerFunction(fn) && !getServerFunctionMetadata(fn)?.live) fn = live(fn) as unknown as T;
   const liveFn = ((...args: Parameters<T>) => {
     const key = name + hashKey(args);
     if (isServer) {
@@ -395,7 +410,8 @@ export function liveQuery<T extends (...args: any) => any>(fn: T, name: string):
         openChannel(channels, key, () => fn(...(args as any)), false, true)
       );
     }
-    const open = () => openChannel(channelMap, key, () => fn(...(args as any)), true, false);
+    const open = (resume?: any) =>
+      openChannel(channelMap, key, () => fn(...(args as any)), true, false, resume);
     // Preload warms the channel, exactly as calling a query in a preload
     // function warms its cache: connect now, hold a temporary consumer slot
     // so refcounting doesn't close it before navigation renders. When the

test/data/live-query.spec.ts

@@ -1,5 +1,10 @@
 import { vi } from "vitest";
 import { flush } from "solid-js";
+import {
+  createServerReference,
+  getServerFunctionMetadata,
+  live
+} from "@solidjs/web/server-functions";
 import { liveQuery } from "../../src/data/liveQuery.js";
 import { deliverFlightData, revalidate } from "../../src/data/query.js";
 
@@ -563,3 +568,83 @@ describe("liveQuery through solid's reactive consumption", () => {
     });
   });
 });
+
+describe("liveQuery over a server function", () => {
+  // vitest's browser conditions load the client entry, whose
+  // createServerReference takes an id (the types resolve the server ABI).
+  const reference = (id: string) => (createServerReference as unknown as (id: string) => any)(id);
+
+  async function withFetch(run: (seen: string[]) => Promise<void>) {
+    const seen: string[] = [];
+    const originalFetch = globalThis.fetch;
+    globalThis.fetch = (async (url: any) => {
+      seen.push(String(url));
+      // BodyFormat.String: a one-value answer
+      return new Response("v1", {
+        headers: { "X-Server-Function-Format": "1", "Content-Type": "text/plain" }
+      });
+    }) as typeof fetch;
+    try {
+      await run(seen);
+    } finally {
+      globalThis.fetch = originalFetch;
+    }
+  }
+
+  test("implies live(): the channel connects at the live address", async () => {
+    const ref = reference("lq-sf-0");
+    await withFetch(async seen => {
+      const lq = liveQuery(ref, uniqueName());
+      const it = lq()[Symbol.asyncIterator]();
+      expect((await it.next()).value).toBe("v1");
+      await it.return!();
+      expect(seen).toHaveLength(1);
+      expect(seen[0]).toContain("/_server/live/lq-sf-0");
+    });
+    // the declaration is the wrapper's; the reference itself is untouched
+    expect(getServerFunctionMetadata(ref)?.live).toBeUndefined();
+  });
+
+  test("an explicit live() declaration passes through untouched", async () => {
+    await withFetch(async seen => {
+      const lq = liveQuery(live(reference("lq-sf-1")), uniqueName());
+      const it = lq()[Symbol.asyncIterator]();
+      expect((await it.next()).value).toBe("v1");
+      await it.return!();
+      expect(seen).toHaveLength(1);
+      expect(seen[0]).toContain("/_server/live/lq-sf-1");
+    });
+  });
+});
+
+describe("hydration takeover resume", () => {
+  const RESUME = Symbol.for("solid.LiveResumeFrom");
+
+  test("the hydrated value is forwarded onto the first connection's iterable", async () => {
+    const stamped: any[] = [];
+    const fn = () => {
+      const source: any = {
+        [Symbol.asyncIterator]() {
+          stamped.push(source[RESUME]);
+          let sent = false;
+          return {
+            next: async () =>
+              sent
+                ? new Promise<never>(() => {})
+                : ((sent = true), { done: false, value: "fresh" }),
+            return: async () => ({ done: true, value: undefined })
+          };
+        }
+      };
+      return source;
+    };
+    const lq = liveQuery(fn, uniqueName());
+    const iterable: any = lq();
+    // what solid's takeover run does to the iterable its compute returned
+    iterable[RESUME] = "served";
+    const it = iterable[Symbol.asyncIterator]();
+    expect((await it.next()).value).toBe("fresh");
+    expect(stamped).toEqual(["served"]);
+    await it.return!();
+  });
+});