feat(service,host): poll where Neon Realtime is not configured

packages/features/fixtures/json-schema/live-wire-accountInboxLiveAnswerSchema.json

@@ -19,6 +19,9 @@
             "expiresAt": {
               "type": "number",
               "minimum": 0
+            },
+            "rows": {
+              "type": "string"
             }
           },
           "required": [
@@ -40,6 +43,9 @@
             "expiresAt": {
               "type": "number",
               "minimum": 0
+            },
+            "rows": {
+              "type": "string"
             }
           },
           "required": [
@@ -61,6 +67,9 @@
             "expiresAt": {
               "type": "number",
               "minimum": 0
+            },
+            "rows": {
+              "type": "string"
             }
           },
           "required": [
@@ -82,6 +91,9 @@
             "expiresAt": {
               "type": "number",
               "minimum": 0
+            },
+            "rows": {
+              "type": "string"
             }
           },
           "required": [
@@ -103,6 +115,9 @@
             "expiresAt": {
               "type": "number",
               "minimum": 0
+            },
+            "rows": {
+              "type": "string"
             }
           },
           "required": [
@@ -124,6 +139,9 @@
             "expiresAt": {
               "type": "number",
               "minimum": 0
+            },
+            "rows": {
+              "type": "string"
             }
           },
           "required": [
@@ -145,6 +163,9 @@
             "expiresAt": {
               "type": "number",
               "minimum": 0
+            },
+            "rows": {
+              "type": "string"
             }
           },
           "required": [

packages/features/fixtures/json-schema/live-wire-scopeInboxLiveAnswerSchema.json

@@ -19,6 +19,9 @@
             "expiresAt": {
               "type": "number",
               "minimum": 0
+            },
+            "rows": {
+              "type": "string"
             }
           },
           "required": [
@@ -40,6 +43,9 @@
             "expiresAt": {
               "type": "number",
               "minimum": 0
+            },
+            "rows": {
+              "type": "string"
             }
           },
           "required": [
@@ -61,6 +67,9 @@
             "expiresAt": {
               "type": "number",
               "minimum": 0
+            },
+            "rows": {
+              "type": "string"
             }
           },
           "required": [

packages/features/fixtures/json-schema/live-wire-sealedLiveQuerySchema.json

@@ -10,6 +10,9 @@
     "expiresAt": {
       "type": "number",
       "minimum": 0
+    },
+    "rows": {
+      "type": "string"
     }
   },
   "required": [

packages/features/src/live-wire.ts

@@ -44,14 +44,29 @@ import { countedNumber } from "./service-wire.js";
  * statement to select every column under the key the wire reads it by and
  * each instant as epoch milliseconds. Realtime keeps no order, so the
  * reader orders them as the held read does.
+ *
+ * A deployment without Realtime (a Preview, or local dev) answers the same
+ * shape at `LIVE_POLL_URL` instead, each statement already run and its
+ * `rows` inline, and the desktop polls by sealing again.
  */
 
+/** The address a deployment without Realtime answers, which tells the desktop to poll rather than subscribe. */
+export const LIVE_POLL_URL = "luke-poll:";
+
 /** One sealed statement as the service answers it, which Realtime's client subscribes to and renews with. */
 export const sealedLiveQuerySchema = EffectSchema.Struct({
   capability: EffectSchema.String,
   queryFingerprint: EffectSchema.String,
   /** When the seal expires, in epoch milliseconds; the reader renews it before. */
   expiresAt: countedNumber,
+  /**
+   * The statement's rows as they stand, answered only at `LIVE_POLL_URL`,
+   * where nothing subscribes. They travel as JSON text, since a row has no
+   * wire form until its reader decodes it.
+   */
+  rows: EffectSchema.optionalKey(
+    EffectSchema.fromJsonString(EffectSchema.Array(EffectSchema.Unknown)),
+  ),
 });
 
 export type SealedLiveQueryWire = typeof sealedLiveQuerySchema.Type;

packages/host/src/live-follow.test.ts

@@ -0,0 +1,67 @@
+import assert from "node:assert/strict";
+import { it } from "@effect/vitest";
+import { LIVE_POLL_URL, type SealedLiveQueryWire } from "@luke/features/live-wire";
+import { Effect, Fiber, Stream } from "effect";
+import { TestClock } from "effect/testing";
+import { liveRows, openRealtimeClient, type SealedRead } from "./live-follow.js";
+
+/**
+ * A read the service answered at the poll address, where it has no Realtime:
+ * the rows its seal carried are the read's first rows, and each seal after,
+ * a few seconds apart, hands on its rows only where they changed.
+ *
+ * Rows are synthetic; the service is a script of the seals it answers.
+ */
+
+/** A seal as the polling service answers one: already expired, its rows inline. */
+const polled = (rows: readonly unknown[]): SealedRead<"items"> => ({
+  url: LIVE_POLL_URL,
+  queries: {
+    items: {
+      capability: LIVE_POLL_URL,
+      queryFingerprint: LIVE_POLL_URL,
+      expiresAt: 0,
+      rows,
+    } satisfies SealedLiveQueryWire,
+  },
+});
+
+/** Every step the read has to take, taken. */
+const settled = Effect.repeat(Effect.yieldNow, { times: 50 });
+
+it.effect(
+  "a polled read hands on its first rows, then each seal's rows a few seconds apart, skipping a seal whose rows did not change",
+  () =>
+    Effect.gen(function* () {
+      const answers = [[{ id: "a" }], [{ id: "a" }, { id: "b" }], [{ id: "b" }]];
+      const heard: unknown[][] = [];
+      const reading = yield* Effect.forkChild(
+        Stream.runForEach(
+          liveRows(
+            {
+              openLive: openRealtimeClient,
+              reseal: () => Effect.sync(() => polled(answers.shift() ?? [])),
+              now: () => 0,
+            },
+            polled([{ id: "a" }]),
+          ),
+          (rows) => Effect.sync(() => void heard.push([...rows.items])),
+        ),
+      );
+      yield* settled;
+      assert.deepEqual(heard, [[{ id: "a" }]]);
+
+      yield* TestClock.adjust("3 seconds");
+      yield* settled;
+      assert.deepEqual(heard, [[{ id: "a" }]], "an unchanged poll is no change");
+
+      yield* TestClock.adjust("3 seconds");
+      yield* settled;
+      assert.deepEqual(heard, [[{ id: "a" }], [{ id: "a" }, { id: "b" }]]);
+
+      yield* TestClock.adjust("3 seconds");
+      yield* settled;
+      assert.deepEqual(heard.at(-1), [{ id: "b" }]);
+      yield* Fiber.interrupt(reading);
+    }),
+);

packages/host/src/live-follow.ts

@@ -1,4 +1,4 @@
-import type { SealedLiveQueryWire } from "@luke/features/live-wire";
+import { LIVE_POLL_URL, type SealedLiveQueryWire } from "@luke/features/live-wire";
 import { Cause, Data, Duration, Effect, Queue, Stream } from "effect";
 
 /**
@@ -12,12 +12,19 @@ import { Cause, Data, Duration, Effect, Queue, Stream } from "effect";
  * later, as Neon's own bindings renew ("Renew the sealed query"). The sealed
  * statements stay here, where the account's own bearer is; only what the
  * caller builds from the rows leaves.
+ *
+ * Where the service has no Realtime it answers `LIVE_POLL_URL` with each
+ * statement's rows inline, and the read polls instead: the same client
+ * shape hands those rows on (`pollingClient`), and the read is sealed again
+ * every `LIVE_POLL_EVERY` rather than before an expiry.
  */
 
 /** How long before a sealed statement expires it is renewed, as Neon's own bindings renew it. */
 const RENEW_EARLY = Duration.seconds(10);
 /** How soon a renewal whose fetch failed is asked again, as Neon's own bindings ask. */
 const RENEW_RETRY = Duration.seconds(1);
+/** How often a read the service answered at `LIVE_POLL_URL` is sealed again, which is how often it polls. */
+const LIVE_POLL_EVERY = Duration.seconds(3);
 
 /** A subscription that stopped for good: Realtime's client did not load, Realtime said so, a renewal was refused, or a row did not decode. */
 export class LiveBroken extends Data.TaggedError("LiveBroken")<{ readonly reason: string }> {}
@@ -40,13 +47,53 @@ export interface LiveClient {
   close(): void;
 }
 
+/** One snapshot of a part, as Realtime's client reports it. */
+type LiveSnapshot = Parameters<Parameters<LiveSubscription["onChange"]>[0]>[0];
+
 /**
- * Realtime's own client, at the address the service named, loaded only when
+ * The client where the service polls: each part's rows are the ones its
+ * seal carried, handed on as each listener joins and again as each renewal
+ * brings rows that differ from the last, so an unchanged poll is no change.
+ */
+function pollingClient(): LiveClient {
+  return {
+    subscribe: (query) => {
+      let rows = query.rows;
+      let said = JSON.stringify(rows);
+      const listeners = new Set<(snapshot: LiveSnapshot) => void>();
+      return {
+        onChange: (listener) => {
+          listeners.add(listener);
+          listener({ status: "ready", data: rows });
+          return () => void listeners.delete(listener);
+        },
+        renew: (renewed) => {
+          const saying = JSON.stringify(renewed.rows);
+          if (saying !== said) {
+            rows = renewed.rows;
+            said = saying;
+            for (const listener of listeners) listener({ status: "ready", data: rows });
+          }
+          return Promise.resolve();
+        },
+        unsubscribe: () => listeners.clear(),
+      };
+    },
+    close: () => undefined,
+  };
+}
+
+/**
+ * Realtime's own client at the address the service named, loaded only when
  * a read is first followed, so its protocol's validators stay out of the
- * web app's first load.
+ * web app's first load; or the polling one where it named `LIVE_POLL_URL`.
  */
 export const openRealtimeClient = (url: string): Promise<LiveClient> =>
-  import("@neon/realtime/client").then(({ createRealtimeClient }) => createRealtimeClient({ url }));
+  url === LIVE_POLL_URL
+    ? Promise.resolve(pollingClient())
+    : import("@neon/realtime/client").then(({ createRealtimeClient }) =>
+        createRealtimeClient({ url }),
+      );
 
 /** A read sealed for following: where to subscribe, and each part's sealed statement. */
 export interface SealedRead<Part extends string> {
@@ -70,6 +117,12 @@ function earliestExpiry<Part extends string>(read: SealedRead<Part>): number {
   );
 }
 
+/** How long until the read is sealed again: the poll's own beat where it polls, and otherwise just before the earliest seal expires. */
+function untilRenewal<Part extends string>(read: SealedRead<Part>, now: number): Duration.Duration {
+  if (read.url === LIVE_POLL_URL) return LIVE_POLL_EVERY;
+  return Duration.millis(Math.max(0, earliestExpiry(read) - Duration.toMillis(RENEW_EARLY) - now));
+}
+
 /**
  * Every part's rows, whole and undecoded, each time any part changes once
  * every part has its first result, for as long as the subscriptions stand.
@@ -116,10 +169,9 @@ export function liveRows<Part extends string>(
             for (const [, subscription] of standing) subscription.unsubscribe();
           }),
       );
-      let expiresAt = earliestExpiry(first);
+      let latest: SealedRead<Part> = first;
       const renewal = Effect.gen(function* () {
-        const wait = expiresAt - Duration.toMillis(RENEW_EARLY) - dependencies.now();
-        yield* Effect.sleep(Duration.millis(Math.max(0, wait)));
+        yield* Effect.sleep(untilRenewal(latest, dependencies.now()));
         let next = yield* dependencies.reseal();
         while ("failure" in next) {
           yield* Effect.sleep(RENEW_RETRY);
@@ -135,7 +187,7 @@ export function liveRows<Part extends string>(
             }),
           { discard: true },
         );
-        expiresAt = earliestExpiry(renewed);
+        latest = renewed;
       });
       yield* Effect.forkScoped(
         Effect.forever(renewal).pipe(

packages/service/src/http/held-read.ts

@@ -22,7 +22,12 @@ import { LiveWake } from "../live/live-wake.js";
  * way (`lookedUntil`).
  */
 
-/** How often a watched wait looks though nothing told it to: what a change the watch missed waits at most. */
+/**
+ * How often a watched wait looks though nothing told it to: what a change the
+ * watch missed waits at most. Note that it is the only recovery from a
+ * Realtime subscription that was refused or dropped, which the wake never
+ * subscribes again, so it stays even where Realtime is configured.
+ */
 const LIVE_SAFETY_POLL = Duration.seconds(5);
 
 /** How many characters of the digest a cursor carries: far more than any two readings of one plan's rows could share by chance. */