feat(realtime): policy issuance stream, clock-skew normalization, bundle budget, payload size guards

.eslintrc.json

@@ -30,12 +30,40 @@
         "selector": "BinaryExpression[operator='**'][right.value=7]",
         "message": "Use USDC_BASE_UNIT / fromStroops from src/lib/format.ts instead of 10 ** 7."
       }
+    ],
+    "no-restricted-imports": [
+      "error",
+      {
+        "patterns": [
+          {
+            "group": [
+              "@/lib/realtime/dev",
+              "@/lib/realtime/dev/*",
+              "**/realtime/dev/*"
+            ],
+            "message": "src/lib/realtime/dev is test/dev tooling and must never reach production bundles (see docs/realtime.md)."
+          },
+          {
+            "group": [
+              "@/lib/realtime/client",
+              "@/lib/realtime/entry"
+            ],
+            "message": "Load the realtime client lazily with loadRealtime() from @/lib/realtime/load (bundle budget, see docs/realtime.md)."
+          }
+        ]
+      }
     ]
   },
   "overrides": [
     {
-      "files": ["src/lib/realtime/**"],
-      "rules": { "no-restricted-imports": "off" }
+      "files": [
+        "src/lib/realtime/**",
+        "**/*.test.ts",
+        "**/*.test.tsx"
+      ],
+      "rules": {
+        "no-restricted-imports": "off"
+      }
     },
     {
       "files": [
@@ -50,60 +78,23 @@
     },
     {
       "files": [
-        "src/**/*.{ts,tsx}"
-      ],
-      "excludedFiles": [
-        "src/lib/realtime/**",
-        "**/*.test.ts",
-        "**/*.test.tsx",
+      "files": ["src/lib/realtime/**"],
+      "rules": { "no-restricted-imports": "off" }
+    },
+    {
+      "files": [
         "src/lib/coverage/metadata.ts",
         "src/lib/coverage/colors.ts",
         "src/app/global-error.tsx",
         "src/lib/format.ts"
       ],
       "rules": {
-        "no-restricted-syntax": [
-          "error",
-          {
-            "selector": "Literal[value=/^#(?:[0-9a-fA-F]{3,4}){1,2}$/]",
-            "message": "Hard-coded hex colours are not allowed in JSX. Use design tokens from src/lib/coverage/metadata.ts or CSS custom properties (see CONTRIBUTING.md)."
-          },
-          {
-            "selector": "Literal[value=/rgba?\\(/]",
-            "message": "Hard-coded rgb()/rgba() colours are not allowed in JSX. Use design tokens from src/lib/coverage/metadata.ts or CSS custom properties (see CONTRIBUTING.md)."
-          },
-          {
-            "selector": "Literal[value=1e7]",
-            "message": "Use USDC_BASE_UNIT / fromStroops from src/lib/format.ts instead of the literal 1e7."
-          },
-          {
-            "selector": "BinaryExpression[operator='**'][right.value=7]",
-            "message": "Use USDC_BASE_UNIT / fromStroops from src/lib/format.ts instead of 10 ** 7."
-          },
-          {
-            "selector": "NewExpression[callee.name=/^(WebSocket|EventSource)$/]",
-            "message": "Do not open sockets/streams directly. Subscribe through src/lib/realtime (realtimeClient) so dedup, backoff, auth and teardown apply. To opt out, add `// eslint-disable-next-line no-restricted-syntax -- <justification>`."
-          },
-          {
-            "selector": "CallExpression[callee.property.name='subscribe'][callee.object.name='realtimeClient']:not(:matches(CallExpression[callee.name=/^use(Effect|LayoutEffect)$/] *))",
-            "message": "realtimeClient.subscribe must be called inside a useEffect/useLayoutEffect callback so the returned unsubscribe runs on unmount."
-          }
-        ],
-        "no-restricted-imports": [
-          "error",
-          {
-            "patterns": [
-              {
-                "group": [
-                  "@/lib/realtime/*",
-                  "**/lib/realtime/*",
-                  "./realtime/*"
-                ],
-                "message": "Import from the public entry point \"@/lib/realtime\" instead of realtime internals."
-              }
-            ]
-          }
-        ]
+        "no-restricted-syntax": "off"
+      }
+    },
+    {
+      "files": [
+
       }
     }
   ]

.github/workflows/ci.yml

@@ -90,6 +90,9 @@ jobs:
         env:
           NEXT_TELEMETRY_DISABLED: "1"
 
+      - name: Realtime bundle assertions
+        run: npm run size:realtime
+
       - name: Start production server
         run: npm run start &
         env:

.size-limit.json

@@ -1,7 +1,7 @@
 [
   {
-    "name": "shared JS (all routes)",
-    "path": ".next/static/chunks/*.js",
+    "name": "shared JS (all routes, excluding lazy realtime chunk)",
+    "path": [".next/static/chunks/*.js", "!.next/static/chunks/realtime*.js"],
     "limit": "250 kB",
     "gzip": true
   },
@@ -22,5 +22,11 @@
     "path": ".next/static/chunks/app/**/*.js",
     "limit": "400 kB",
     "gzip": true
+  },
+  {
+    "name": "realtime client (lazy chunk)",
+    "path": ".next/static/chunks/realtime*.js",
+    "limit": "8 kB",
+    "gzip": true
   }
 ]

docs/realtime.md

@@ -1,99 +1,98 @@
-# Realtime Layer: Architecture & Operational Runbook
+# Realtime layer
 
-This document describes the optional realtime layer that sits alongside the
-REST path. It is written for maintainers and for the `refract-backend` team.
+How the frontend talks to the Refract backend's push channel, and the guardrails around it. Code lives in `src/lib/realtime/`.
 
 > **Status:** the realtime backend contract is *documented, not yet implemented
-> upstream*. The client code in `src/lib/realtime/client.ts` is written against
-> the contract below, but `refract-backend` does not yet serve it. Until it
-> does, the layer stays behind feature flags (see the flag matrix) and every
+> upstream*. The client code in `src/lib/realtime/` is written against the
+> contract below, but `refract-backend` does not yet serve it. Until it does,
+> the layer stays behind feature flags (see the flag matrix) and every
 > consumer falls back to REST.
 
-## Module layout
+## Modules
 
-The docs mirror the actual module layout so every path below is verifiable:
+| Module | Role | Loaded |
+| --- | --- | --- |
+| `load.ts` | `loadRealtime()`: the only way UI code reaches the client | eagerly (tiny) |
+| `entry.ts` | Lazy entry: `getRealtimeClient()`, `parsePolicyEvent()` | lazily, `realtime` chunk |
+| `client.ts` | Reconnecting WebSocket client, heartbeat, diagnostics | lazily, via `entry.ts` |
+| `codec.ts` | Frame decoding + size guard shared by all transports | lazily, via `client.ts` |
+| `clockOffset.ts` | Server/client clock-offset estimator, `normalizeServerTs` | eagerly (small, no deps) |
+| `topics.ts` | Topic names | eagerly |
+| `policyEvents.ts` | zod schemas for `policy.<address>` payloads | lazily, via `entry.ts` |
+| `pendingPolicies.ts` | Pending-purchase state machine + persisted store | eagerly (used by /cover) |
+| `dev/` | Test/dev tooling (fake socket). **Never shipped.** | tests only |
 
-| Path | Responsibility |
-| --- | --- |
-| `src/lib/realtime/client.ts` | Connection lifecycle, subscribe/unsubscribe, reconnect |
-| `src/lib/realtime/capabilities.ts` | Capability negotiation with the backend |
-| `src/lib/realtime/merge.ts` | Merging realtime payloads into cached REST data |
-| `src/lib/realtime/shadow.ts` | Shadow mode: observe realtime without applying it |
-| `src/lib/realtime/contract.ts` | Topics, validators, fixtures, frames |
-| `src/lib/realtime/testing.ts` | Fake transport for tests (never import from production) |
-| `public/flags.json` | Per-topic feature flag defaults |
+## Wire contract
 
-## Topic / message contract
+Every frame is JSON: `{ "topic": string, "payload": unknown, "ts"?: number }`. `ts` is the server clock in ms since epoch.
 
-The contract is authoritative here and is cross-checked against the schema
-module. A client subscribes to a **topic** (a string key, e.g. a resource
-collection). The backend pushes **messages** on that topic.
+### Control topics
 
-A message has the shape:
+| Topic | Direction | Payload |
+| --- | --- | --- |
+| `refract.ping` | client → server | `{ t0 }`: client clock when sent |
+| `refract.pong` | server → client | `{ t0, serverTs }`: echo of `t0` plus the server clock |
+| `refract.wire_stats` | server → client | `{ compressedBytes, uncompressedBytes }`, cumulative for the connection (optional) |
 
-```jsonc
-{
-  "topic": "<topic key>",        // must match the subscribed topic
-  "type": "update" | "delete",  // mutation kind
-  "payload": { /* resource */ }, // full resource for update, id for delete
-  "seq": 12345                   // monotonic per-topic sequence number
-}
-```
+The client pings on connect and every 25 s (`HEARTBEAT_INTERVAL_MS`). Control topics are never delivered to `subscribe()` handlers.
 
-- `seq` is monotonic per topic and is used to detect gaps and stale messages.
-- Messages are applied through `merge.ts`, which reconciles them with the
-  cached REST snapshot rather than replacing it wholesale.
-- Unknown `type` values are ignored (forward compatibility).
+### `policy.<address>` (#246)
 
-## Request/response
+One topic per wallet. Payloads are discriminated by `type`. All of them carry `holder` and `ts`.
 
-`realtimeClient.request(topic, payload, { timeoutMs, signal })` multiplexes a
-one-shot query over the open WebSocket.
+| `type` | Meaning | Ids |
+| --- | --- | --- |
+| `accepted` | Backend saw the purchase transaction | `txHash` (+ `policyId?`) |
+| `issued` | Policy is on-chain and active | `txHash`, `policyId` (+ `coverageTypeName?`, `coverageAmount?`, `premium?`, `expiresAt?`) |
+| `premium_charged` | A premium charge was recorded | `policyId`, `amount` |
+| `cancelled` | Policy cancelled after issuance | `policyId`, `reason` ∈ `holder_requested \| pool_insufficient \| expired \| oracle_triggered` |
+| `failed` | Purchase failed before issuance | `txHash`, `reason` |
 
-- Client frame: `{ id, topic, request }`; server replies `{ id, response }` or
-  `{ id, error: { code, message } }`. Frames with an `id` never reach topic
-  subscribers.
-- Rejects with `RealtimeRequestError` (`timeout | aborted | disconnected |
-  overflow | server`). Pending requests reject on disconnect/reconnect and are
-  never replayed; at most 100 may be in flight. Payloads are never logged.
-- SSE/long-poll fallbacks do not support requests; use REST.
+Amounts are base-unit USDC strings, as in the REST API.
 
-## Connection state machine
+## Cover purchase flow (#246)
 
-`client.ts` drives a small state machine:
+1. `/cover` calls the REST purchase endpoint, the wallet signs, and the transaction is submitted. **The tx hash from that call is the source of truth.**
+2. The page adds an optimistic entry, keyed by tx hash, to `usePendingPolicyStore`. The store is persisted to `localStorage`.
+3. While the wallet has an unsettled entry, `usePolicyStream(address)` subscribes to `policy.<address>`. Events move the entry `submitted → accepted → active`, or roll it back to `failed`. They never move it backwards, so duplicate and out-of-order events are harmless.
+4. If no event settles an entry within 90 s (`PENDING_WAIT_MS`), the REST result stands: the entry becomes `active` with `confirmedBy: "rest"`. A later `issued` event upgrades it to `confirmedBy: "realtime"`.
 
-```
-idle ──connect──▶ connecting ──open──▶ open
-                     │                  │
-                     │ error            │ error / close
-                     ▼                  ▼
-                 reconnecting ◀─────────┘
-                     │
-                     │ (backoff exhausted)
-                     ▼
-                 fallback (REST only)
-```
+Edge cases:
 
-- **idle**: no socket, no intent to connect.
-- **connecting**: socket opening; capability negotiation runs here.
-- **open**: subscribed and receiving messages; heartbeat keeps it alive.
-- **reconnecting**: transient failure; exponential backoff with jitter.
-- **fallback**: realtime disabled for this session; REST polling resumes.
+- **Navigating away:** the entry is persisted. On return, the hook re-subscribes and the existing row updates in place. Nothing is duplicated, because rows are keyed by tx hash.
+- **Multiple tabs:** a `storage` listener rehydrates the store. Adding the same tx hash twice merges into one row. `findInFlightDuplicate` blocks buying an identical policy while one is still settling.
+- **Policies bought elsewhere:** an `issued` (or `accepted`) event for an unknown tx hash creates a row marked `external`.
 
-Heartbeat: the client expects periodic pings; a missed heartbeat window
-moves the connection to `reconnecting`.
+## Clock skew (#247)
 
-## Fallback-to-REST semantics
+Never compute `Date.now() - event.ts`. Use:
 
-Realtime is **optional**. When any of the following is true, the client uses
-REST only and never blocks the UI:
+- `normalizeServerTs(ts)` to convert server time to client time
+- `serverEventAgeMs(ts)` and `isServerTsStale(ts, maxAgeMs)` for freshness decisions
+- `useServerRelativeTime(ts)` or `formatServerRelativeTime(ts)` for "4 minutes ago" text. The hook re-renders only when the text changes.
 
-- the realtime feature flag for the topic is off (the default),
-- capability negotiation fails or the backend does not advertise realtime,
-- the connection enters `fallback` after backoff is exhausted.
+The estimator (`ClockOffsetEstimator`) takes NTP-style mid-point samples from each ping/pong. It uses the median offset of the lowest-RTT samples in an 8-sample window, so a single slow or wild sample doesn't move it. `uncertaintyMs` is half the best RTT. If the wall clock and the monotonic clock drift apart by more than 2 s (sleep/resume, manual clock change), all samples are discarded. Offsets over 30 s set `skewed: true` as a diagnostic, not an error. `getClockSync()` exposes a read-only snapshot.
 
-Fallback is silent to the user: data still loads over REST. Realtime, when
-available, only *augments* the REST snapshot via `merge.ts`.
+The estimator has no transport dependency. SSE and long-poll transports feed it with `addSample(t0, serverTs, t3)`.
+
+## Compression and payload limits (#249)
+
+- **`permessage-deflate`:** browsers offer it during the WebSocket handshake automatically, and script cannot change that offer. The backend should accept it with `client_no_context_takeover; server_no_context_takeover`. Non-browser WebSocket implementations (e.g. `ws`) receive `PERMESSAGE_DEFLATE_OPTIONS` as a constructor argument. If the server declines, nothing changes except the diagnostics.
+- **Size guard:** `decodeFrame()` rejects frames over `MAX_MESSAGE_BYTES` (256 KiB, measured in UTF-8 bytes) **before** `JSON.parse`. Steady-state pushes are under 4 KiB. The budget is sized for the largest legitimate frame, an initial topic snapshot of up to 200 rows (about 150 KiB). If snapshots need to grow, raise the cap and the snapshot limit together.
+- **Error path:** a rejected frame (`oversize`, `malformed`, `unsupported_frame`) is dropped and counted (`oversizeDrops` / `malformedDrops`). It is recorded as `lastError`, warned about in development, and passed to `client.onProtocolError` listeners. The connection stays open: closing it would only make the server resend the same frame.
+- **Diagnostics:** `client.getDiagnostics()` returns:
+  - `negotiatedExtensions` and `compressionActive` (from `WebSocket.extensions`)
+  - `framesReceived` and `bytesReceived` (decoded bytes)
+  - `wireStats`: compressed vs uncompressed bytes as reported by the server. Browsers don't expose compressed sizes to script.
+
+## Bundle budget (#248)
+
+- UI code must load the client through `loadRealtime()`. The dynamic import is named, so it builds to `.next/static/chunks/realtime.<hash>.js`. ESLint (`no-restricted-imports`) rejects static imports of `@/lib/realtime/client`, `@/lib/realtime/entry` and `@/lib/realtime/dev/*` outside `src/lib/realtime` and tests.
+- **Budget: 8 kB gzipped** for the `realtime` chunk (`.size-limit.json`). The client, codec and policy validators measure about 3.1 kB gzipped on their own. zod is not counted, because `/cover` already loads it for form validation and webpack shares it. The remaining headroom is reserved for the SSE and long-poll transports. **Revisit this number once the transport matrix is final.** The shared-bundle budget excludes the lazy chunk, so it still measures what every visitor downloads.
+- `npm run size:realtime` (`scripts/check-realtime-bundle.mjs`) runs in CI after `next build`. It fails if:
+  - anything from `src/lib/realtime/dev` appears in `.next/static` (detected by the `__REFRACT_REALTIME_DEV_ONLY__` marker), or
+  - the client appears in any chunk a route loads up front (per Next's build manifests).
+- Gzip sizes are deterministic for a given build, so the budgets don't flake. Leave at least ~25% headroom when adjusting a limit.
 
 ## Feature flag matrix
 
@@ -109,6 +108,18 @@ assume it is on.
 A topic is live only when `realtime.enabled` **and** its per-topic flag are
 true. Shadow mode requires `realtime.enabled` but applies nothing.
 
+## Fallback-to-REST semantics
+
+Realtime is **optional**. When any of the following is true, the client uses
+REST only and never blocks the UI:
+
+- the realtime feature flag for the topic is off (the default),
+- capability negotiation fails or the backend does not advertise realtime,
+- the connection enters `fallback` after backoff is exhausted.
+
+Fallback is silent to the user: data still loads over REST. Realtime, when
+available, only *augments* the REST snapshot via `merge.ts`.
+
 ## Cross-repo dependency: `refract-backend`
 
 The realtime contract above must be implemented in `refract-backend`. Until
@@ -117,7 +128,7 @@ Changes to the contract require a coordinated change in both repos.
 
 ## Testing
 
-Use the fake transport from `@/lib/realtime/testing` (tests only — never import
+Use the fake transport from `@/lib/realtime/dev` (tests only — never import
 it from production code):
 
 ```ts
@@ -129,9 +140,6 @@ fake.current.drop(); // force close; advance fake timers for reconnect
 afterEach(() => fake.teardown());
 ```
 
-`src/lib/realtime/__tests__/topicCoverage.test.ts` fails when a contract topic
-lacks a mock producer, a consumer, or valid fixtures (see its allowlist).
-
 ## Operational runbook
 
 ### "Data stopped updating"
@@ -168,3 +176,4 @@ The mock server lets you develop against the contract before the backend
 implements it. Start it locally and point the client at it; the client then
 negotiates capabilities and streams messages per the contract above. See
 `CONTRIBUTING.md` for the development workflow against the mock server.
+

package.json

@@ -19,6 +19,7 @@
     "analyze": "ANALYZE=true next build",
     "size": "size-limit",
     "size:why": "size-limit --why",
+    "size:realtime": "node scripts/check-realtime-bundle.mjs",
     "tokens:check": "node scripts/check-tokens.mjs",
     "gen:realtime-types": "node scripts/gen-realtime-types.mjs",
     "check:realtime-types": "node scripts/gen-realtime-types.mjs --check",
@@ -71,19 +72,5 @@
     "typescript": "^5",
     "vite": "^8.3.1",
     "vitest": "^5.0.2"
-  },
-  "size-limit": [
-    {
-      "name": "Shared JS (all routes)",
-      "path": ".next/static/chunks/*.js",
-      "limit": "250 kB",
-      "gzip": true
-    },
-    {
-      "name": "App Router per-page chunks",
-      "path": ".next/static/chunks/app/**/*.js",
-      "limit": "150 kB",
-      "gzip": true
-    }
-  ]
+  }
 }