From a5538edc5e456f6a6334a314085a564465f92c1b Mon Sep 17 00:00:00 2001 From: lucky Ekpa <64299817+Lucky-Ekpa-800@users.noreply.github.com> Date: Wed, 21 Oct 2026 22:57:13 +0000 Subject: [PATCH] feat(realtime): policy issuance stream, clock-skew normalization, bundle budget, payload size guards Closes #246, #247, #248, #249. - #246: `policy.
` topic (accepted/issued/premium_charged/cancelled/failed), persisted pending-purchase store keyed by tx hash, usePolicyStream hook with bounded-wait REST fallback and cross-tab sync, PolicyProgress UI on /cover. - #247: transport-independent ClockOffsetEstimator (min-RTT median, clock-jump invalidation, skew diagnostic), normalizeServerTs helpers, useServerRelativeTime. - #248: lazy `realtime` chunk via loadRealtime(), 8 kB size-limit budget, scripts/check-realtime-bundle.mjs CI assertion, ESLint import restrictions. - #249: shared decodeFrame with a 256 KiB pre-parse size guard, protocol-error path and counters, permessage-deflate options and negotiation diagnostics. Also repairs the truncated JSX in src/app/cover/page.tsx (and moves its hard-coded colours onto design tokens), removes the empty vitest.config.ts that shadowed vitest.config.mts, fixes the invalid ESLint selector that crashed lint, and drops the duplicate size-limit config from package.json that hid .size-limit.json. Co-Authored-By: Claude Opus 5.5 --- .eslintrc.json | 41 +- .github/workflows/ci.yml | 3 + .size-limit.json | 10 +- docs/realtime.md | 89 +++ package.json | 17 +- scripts/check-realtime-bundle.mjs | 82 +++ src/app/cover/page.tsx | 772 ++++++++++++++++------- src/components/cover/PolicyProgress.tsx | 85 +++ src/components/ui/index.ts | 5 + src/hooks/usePolicyStream.test.ts | 144 +++++ src/hooks/usePolicyStream.ts | 105 +++ src/hooks/useServerRelativeTime.ts | 40 ++ src/lib/realtime/client.test.ts | 217 +++++++ src/lib/realtime/client.ts | 297 +++++++-- src/lib/realtime/clockOffset.test.ts | 186 ++++++ src/lib/realtime/clockOffset.ts | 228 +++++++ src/lib/realtime/codec.test.ts | 63 ++ src/lib/realtime/codec.ts | 130 ++++ src/lib/realtime/dev/fakeSocket.ts | 93 +++ src/lib/realtime/entry.ts | 22 + src/lib/realtime/load.ts | 20 + src/lib/realtime/pendingPolicies.test.ts | 148 +++++ src/lib/realtime/pendingPolicies.ts | 267 ++++++++ src/lib/realtime/policyEvents.test.ts | 40 ++ src/lib/realtime/policyEvents.ts | 83 +++ src/lib/realtime/topics.ts | 29 + vitest.config.mts | 2 + vitest.config.ts | 0 28 files changed, 2937 insertions(+), 281 deletions(-) create mode 100644 docs/realtime.md create mode 100644 scripts/check-realtime-bundle.mjs create mode 100644 src/components/cover/PolicyProgress.tsx create mode 100644 src/hooks/usePolicyStream.test.ts create mode 100644 src/hooks/usePolicyStream.ts create mode 100644 src/hooks/useServerRelativeTime.ts create mode 100644 src/lib/realtime/client.test.ts create mode 100644 src/lib/realtime/clockOffset.test.ts create mode 100644 src/lib/realtime/clockOffset.ts create mode 100644 src/lib/realtime/codec.test.ts create mode 100644 src/lib/realtime/codec.ts create mode 100644 src/lib/realtime/dev/fakeSocket.ts create mode 100644 src/lib/realtime/entry.ts create mode 100644 src/lib/realtime/load.ts create mode 100644 src/lib/realtime/pendingPolicies.test.ts create mode 100644 src/lib/realtime/pendingPolicies.ts create mode 100644 src/lib/realtime/policyEvents.test.ts create mode 100644 src/lib/realtime/policyEvents.ts create mode 100644 src/lib/realtime/topics.ts delete mode 100644 vitest.config.ts diff --git a/.eslintrc.json b/.eslintrc.json index 3cbf99e..1992e0a 100644 --- a/.eslintrc.json +++ b/.eslintrc.json @@ -12,21 +12,58 @@ "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]", + "selector": "Literal[value=10000000]", "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." } + ], + "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/coverage/metadata.ts", "src/lib/coverage/colors.ts", "src/app/global-error.tsx", "src/lib/format.ts"], + "files": [ + "src/lib/coverage/metadata.ts", + "src/lib/coverage/colors.ts", + "src/app/global-error.tsx", + "src/lib/format.ts" + ], "rules": { "no-restricted-syntax": "off" } + }, + { + "files": [ + "src/lib/realtime/**", + "**/*.test.ts", + "**/*.test.tsx" + ], + "rules": { + "no-restricted-imports": "off" + } } ] } diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index f273c23..e67d398 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -78,6 +78,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: diff --git a/.size-limit.json b/.size-limit.json index cc36d8a..657a661 100644 --- a/.size-limit.json +++ b/.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 } ] diff --git a/docs/realtime.md b/docs/realtime.md new file mode 100644 index 0000000..ceb6437 --- /dev/null +++ b/docs/realtime.md @@ -0,0 +1,89 @@ +# Realtime layer + +How the frontend talks to the Refract backend's push channel, and the guardrails around it. Code lives in `src/lib/realtime/`. + +## Modules + +| 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.
` 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 | + +## Wire contract + +Every frame is JSON: `{ "topic": string, "payload": unknown, "ts"?: number }`. `ts` is the server clock in ms since epoch. + +### Control topics + +| 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) | + +The client pings on connect and every 25 s (`HEARTBEAT_INTERVAL_MS`). Control topics are never delivered to `subscribe()` handlers. + +### `policy.
` (#246) + +One topic per wallet. Payloads are discriminated by `type`. All of them carry `holder` and `ts`. + +| `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` | + +Amounts are base-unit USDC strings, as in the REST API. + +## Cover purchase flow (#246) + +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.
`. 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"`. + +Edge cases: + +- **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`. + +## Clock skew (#247) + +Never compute `Date.now() - event.ts`. Use: + +- `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 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. + +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..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. diff --git a/package.json b/package.json index 51f59c9..5e1c39d 100644 --- a/package.json +++ b/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" }, "lint-staged": { @@ -66,19 +67,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 - } - ] + } } diff --git a/scripts/check-realtime-bundle.mjs b/scripts/check-realtime-bundle.mjs new file mode 100644 index 0000000..fd7591c --- /dev/null +++ b/scripts/check-realtime-bundle.mjs @@ -0,0 +1,82 @@ +#!/usr/bin/env node +/** + * Production-bundle assertions for the realtime layer (issue #248). + * Run after `next build`; exits non-zero on any violation. + * + * 1. Dev/test tooling (src/lib/realtime/dev) never ships: no file under + * .next/static may contain the dev-only marker. + * 2. The realtime client is lazy: it lives in the `realtime` chunk and in + * no chunk a route loads up front (per .next/app-build-manifest.json and + * .next/build-manifest.json). + * + * Usage: node scripts/check-realtime-bundle.mjs [nextDir] + */ +import { existsSync, readdirSync, readFileSync, statSync } from "node:fs"; +import { join, relative } from "node:path"; + +const NEXT_DIR = process.argv[2] ?? ".next"; +const STATIC_DIR = join(NEXT_DIR, "static"); + +/** Exported from src/lib/realtime/dev/fakeSocket.ts. */ +const DEV_ONLY_MARKER = "__REFRACT_REALTIME_DEV_ONLY__"; +/** A property name that only src/lib/realtime/client.ts contains (minifiers keep it). */ +const CLIENT_MARKER = "clientNoContextTakeover"; + +function walk(dir) { + return readdirSync(dir).flatMap((name) => { + const path = join(dir, name); + return statSync(path).isDirectory() ? walk(path) : [path]; + }); +} + +function readJson(path) { + return existsSync(path) ? JSON.parse(readFileSync(path, "utf8")) : null; +} + +if (!existsSync(STATIC_DIR)) { + console.error(`✗ ${STATIC_DIR} not found — run \`next build\` first.`); + process.exit(1); +} + +const failures = []; +const jsFiles = walk(STATIC_DIR).filter((f) => f.endsWith(".js")); +const contents = new Map(jsFiles.map((f) => [f, readFileSync(f, "utf8")])); + +// 1. Dev tooling must not ship. +for (const [file, code] of contents) { + if (code.includes(DEV_ONLY_MARKER)) failures.push(`dev-only realtime code shipped in ${relative(NEXT_DIR, file)}`); +} + +// 2. The client must be in the lazy chunk and nowhere a route loads eagerly. +const clientChunks = [...contents].filter(([, code]) => code.includes(CLIENT_MARKER)).map(([f]) => f); +const realtimeChunks = clientChunks.filter((f) => /[\\/]realtime[.-][^\\/]*\.js$/.test(f)); +if (clientChunks.length === 0) { + console.warn("! realtime client not found in any chunk (no route uses it yet?)"); +} else if (realtimeChunks.length === 0) { + failures.push( + `realtime client found outside the named \`realtime\` chunk: ${clientChunks.map((f) => relative(NEXT_DIR, f)).join(", ")}. ` + + "Load it with loadRealtime() from src/lib/realtime/load.ts.", + ); +} + +const eager = new Set(); +const appManifest = readJson(join(NEXT_DIR, "app-build-manifest.json")); +for (const files of Object.values(appManifest?.pages ?? {})) files.forEach((f) => eager.add(f)); +const buildManifest = readJson(join(NEXT_DIR, "build-manifest.json")); +for (const key of ["rootMainFiles", "polyfillFiles", "lowPriorityFiles"]) (buildManifest?.[key] ?? []).forEach((f) => eager.add(f)); +for (const files of Object.values(buildManifest?.pages ?? {})) files.forEach((f) => eager.add(f)); + +for (const file of eager) { + const code = contents.get(join(NEXT_DIR, file)); + if (code?.includes(CLIENT_MARKER)) failures.push(`realtime client is in an eagerly loaded chunk: ${file}`); +} + +if (failures.length) { + console.error("✗ realtime bundle check failed:"); + failures.forEach((f) => console.error(` - ${f}`)); + process.exit(1); +} +console.log( + `✓ realtime bundle check passed (${realtimeChunks.map((f) => relative(NEXT_DIR, f)).join(", ") || "no realtime chunk"}; ` + + `${eager.size} eager files scanned, no dev-only code shipped)`, +); diff --git a/src/app/cover/page.tsx b/src/app/cover/page.tsx index a7a0cfd..ec81497 100644 --- a/src/app/cover/page.tsx +++ b/src/app/cover/page.tsx @@ -3,7 +3,22 @@ import { useEffect, useMemo, useRef, useState } from "react"; import { useSearchParams } from "next/navigation"; import { Navbar, Footer } from "@/components/layout"; -import { Container, Card, Badge, Input, Button, Skeleton, SuccessBurst, QuickAmountChips, Tour, TourReplayButton, useTour, type TourStep, AsyncState, RelativeDate } from "@/components/ui"; +import { + Container, + Card, + Badge, + Input, + Button, + Skeleton, + SuccessBurst, + QuickAmountChips, + Tour, + TourReplayButton, + useTour, + type TourStep, + AsyncState, + RelativeDate, +} from "@/components/ui"; import { WalletButton, WrongNetworkBanner, TxErrorMessage } from "@/components/wallet"; import { useWallet } from "@/lib/wallet/WalletProvider"; import { useCoverageTypes } from "@/hooks/useCoverageTypes"; @@ -33,23 +48,53 @@ import { parseCoverQueryParams } from "@/lib/coverQueryParams"; import { PolicyCertificateModal } from "@/components/PolicyCertificate"; import { validateBuyCoverage } from "@/lib/validation/coverage"; import { searchFlights, POPULAR_FLIGHTS, type FlightInfo } from "@/lib/flights"; - -const RISK_TAG_COLORS: Record = { - low: "#10b981", - medium: "#8b5cf6", - high: "#f59e0b", - critical: "#ef4444", +import { usePolicyStream } from "@/hooks/usePolicyStream"; +import { + findInFlightDuplicate, + isTerminal, + usePendingPolicyStore, +} from "@/lib/realtime/pendingPolicies"; +import { PolicyProgress } from "@/components/cover/PolicyProgress"; + +/** Risk colours as design-token classes (text colour + gradient end stop). */ +const RISK_TAG_CLASSES: Record = { + low: { text: "text-pm-green", to: "to-pm-green" }, + medium: { text: "text-pm-violet", to: "to-pm-violet" }, + high: { text: "text-pm-amber", to: "to-pm-amber" }, + critical: { text: "text-pm-red", to: "to-pm-red" }, }; const RISK_HEAT: Record = { low: 20, medium: 45, high: 72, critical: 95 }; -const QUICK_AMOUNTS = [1_000, 5_000, 10_000, 25_000].map((v) => ({ label: `$${v.toLocaleString()}`, value: String(v) })); +const QUICK_AMOUNTS = [1_000, 5_000, 10_000, 25_000].map((v) => ({ + label: `$${v.toLocaleString()}`, + value: String(v), +})); const COVER_TOUR: TourStep[] = [ - { target: "cover-type", title: "Pick a coverage type", content: "Each type pays out automatically when its on-chain trigger fires — no claims form. Use the arrow keys to move between types." }, - { target: "cover-amount", title: "Set your coverage amount", content: "This is the USDC payout you'd receive if the trigger fires. Quick chips fill in common amounts." }, - { target: "cover-duration", title: "Choose a duration", content: "Your premium scales with how long the policy stays active, from 1 day up to a year." }, - { target: "cover-buy", title: "Review and buy", content: "Check the premium, then buy and confirm in your wallet — or add several policies to the cart and check out together." }, + { + target: "cover-type", + title: "Pick a coverage type", + content: + "Each type pays out automatically when its on-chain trigger fires — no claims form. Use the arrow keys to move between types.", + }, + { + target: "cover-amount", + title: "Set your coverage amount", + content: + "This is the USDC payout you'd receive if the trigger fires. Quick chips fill in common amounts.", + }, + { + target: "cover-duration", + title: "Choose a duration", + content: "Your premium scales with how long the policy stays active, from 1 day up to a year.", + }, + { + target: "cover-buy", + title: "Review and buy", + content: + "Check the premium, then buy and confirm in your wallet — or add several policies to the cart and check out together.", + }, ]; const CART_STORAGE_KEY = "refract:cover-cart"; @@ -74,7 +119,12 @@ export default function CoverPage() { const wallet = useWallet(); const { requestReview, reviewDialog } = useTxReview(); const searchParams = useSearchParams(); - const { data: coverageTypes, loading: typesLoading, error: typesError, isFixture } = useCoverageTypes(); + const { + data: coverageTypes, + loading: typesLoading, + error: typesError, + isFixture, + } = useCoverageTypes(); const { isCorrectNetwork, expectedNetwork } = useNetworkGuard(); const addTransaction = useTransactionStore((s) => s.addTransaction); const updateTransaction = useTransactionStore((s) => s.updateTransaction); @@ -147,6 +197,20 @@ export default function CoverPage() { const ct = coverageTypes?.[selectedType]; + // Live policy issuance state (issue #246). The REST purchase gives us the + // tx hash; `policy.
` events move that purchase from submitted → + // accepted → active (or failed). Pending purchases are persisted, so they + // are still tracked after navigating away and back. + const pendingPolicies = usePolicyStream(wallet.address); + const addPendingPolicy = usePendingPolicyStore((s) => s.addOptimistic); + const dismissPendingPolicy = usePendingPolicyStore((s) => s.dismiss); + const currentTxHash = + submission.status === "success" && !submission.demo ? submission.txHash : undefined; + const currentPending = currentTxHash + ? pendingPolicies.find((p) => p.txHash === currentTxHash) + : undefined; + const otherPending = pendingPolicies.filter((p) => p.txHash !== currentTxHash); + const premium = useMemo(() => { if (!ct) return 0; const amount = parseFloat(coverageAmount) || 0; @@ -168,6 +232,16 @@ export default function CoverPage() { const amountInvalid = !validation.isValid && !!validation.errors.coverageAmount; const flightNumberInvalid = isFlightDelay && (!flightNumber || flightNumber.trim().length === 0); + // The same purchase still settling (possibly started in another tab). + let duplicatePending: ReturnType; + if (ct && wallet.address && parseFloat(coverageAmount) > 0) { + duplicatePending = findInFlightDuplicate(pendingPolicies, { + holder: wallet.address, + coverageType: ct.id, + coverageAmount: String(toStroops(parseFloat(coverageAmount))), + }); + } + const filteredFlights = useMemo(() => { return searchFlights(flightQuery); }, [flightQuery]); @@ -186,7 +260,7 @@ export default function CoverPage() { id: crypto.randomUUID(), params: { coverageType: ct.id, - coverageAmount: toStroops(parseFloat(coverageAmount)), + coverageAmount: String(toStroops(parseFloat(coverageAmount))), durationDays, ...(isFlightDelay ? { triggerParams: { flightNumber: flightNumber.trim() } } : {}), }, @@ -202,14 +276,19 @@ export default function CoverPage() { setCart((items) => items.map((item) => (item.id === id ? { ...item, ...patch } : item))); } - async function purchaseCartItem(item: CartItem, holder: string, networkPassphrase: string): Promise { + async function purchaseCartItem( + item: CartItem, + holder: string, + networkPassphrase: string + ): Promise { try { const result = await buyPolicy({ ...item.params, holder }); const txHash = await signAndSubmit(result.txXdr, holder, networkPassphrase); return { policyId: result.policy.id, txHash, demo: false }; } catch (err) { // Same clearly-labeled demo fallback as the single-purchase flow. - if (err instanceof ApiUnreachableError) return { policyId: `demo-${crypto.randomUUID()}`, demo: true }; + if (err instanceof ApiUnreachableError) + return { policyId: `demo-${crypto.randomUUID()}`, demo: true }; throw err; } } @@ -223,7 +302,11 @@ export default function CoverPage() { if (!networkPassphrase) return; setCheckingOut(true); try { - await runBatchCheckout(cart, (item) => purchaseCartItem(item, address, networkPassphrase), updateCartItem); + await runBatchCheckout( + cart, + (item) => purchaseCartItem(item, address, networkPassphrase), + updateCartItem + ); } finally { setCheckingOut(false); } @@ -231,23 +314,21 @@ export default function CoverPage() { async function handleBuy() { if (!ct) return; - if (amountInvalid || flightNumberInvalid || !isCorrectNetwork) return; + if (amountInvalid || flightNumberInvalid || !isCorrectNetwork || duplicatePending) return; let holder = wallet.status === "connected" ? wallet.address : null; - let networkPassphrase = wallet.networkPassphrase; if (!holder) { // Connect, then carry on with the same purchase — no second click. // A declined/failed connect resolves null and stops here. - const connected = await wallet.connect(); - if (!connected) return; - holder = connected.address; - networkPassphrase = connected.networkPassphrase; + holder = await wallet.connect(); + if (!holder) return; } setConfirmOpen(true); } async function submitBuy() { setConfirmOpen(false); - if (!ct || !wallet.address) return; + const holder = wallet.address; + if (!ct || !holder) return; setSubmission({ status: "submitting" }); const txId = addTransaction({ type: "buy", amount: parseFloat(coverageAmount) }); @@ -255,7 +336,7 @@ export default function CoverPage() { const result = await buyPolicy({ holder, coverageType: ct.id, - coverageAmount: toStroops(parseFloat(coverageAmount)), + coverageAmount: String(toStroops(parseFloat(coverageAmount))), durationDays, ...(isFlightDelay ? { triggerParams: { flightNumber: flightNumber.trim() } } : {}), }); @@ -274,13 +355,26 @@ export default function CoverPage() { { label: "Coverage amount", value: formatUsd(parseFloat(coverageAmount)) }, { label: "Duration", value: `${durationDays} days` }, ], - expectedStroops: toStroops(parseFloat(coverageAmount)), + expectedStroops: String(toStroops(parseFloat(coverageAmount))), }); if (!approved) { setSubmission({ status: "idle" }); return; } - const txHash = await signAndSubmit(result.txXdr, wallet.address, wallet.networkPassphrase, { hardware: wallet.hardwareWallet }); + const txHash = await signAndSubmit(result.txXdr, holder, wallet.networkPassphrase, { + hardware: wallet.hardwareWallet, + }); + // The REST call and the submitted tx hash are the source of truth; + // `policy.
` events only move this entry forward from here. + addPendingPolicy({ + txHash, + holder, + policyId: result.policy.id, + coverageType: ct.id, + coverageTypeName: result.policy.coverageTypeName ?? ct.name, + coverageAmount: String(toStroops(parseFloat(coverageAmount))), + premium: String(result.policy.premium ?? toStroops(premium)), + }); setSubmission({ status: "success", result, demo: false, txHash }); updateTransaction(txId, { status: "confirmed", txHash }); } catch (err) { @@ -291,8 +385,8 @@ export default function CoverPage() { holder, coverageType: ct.id, coverageTypeName: ct.name, - coverageAmount: toStroops(parseFloat(coverageAmount)), - premium: toStroops(premium), + coverageAmount: String(toStroops(parseFloat(coverageAmount))), + premium: String(toStroops(premium)), durationDays, expiresAt: Math.floor(Date.now() / 1000) + durationDays * 86400, isActive: true, @@ -348,16 +442,21 @@ export default function CoverPage() { - ); - })} - - )} - - + {active && ( +
+ Auto-settle + No form required + On-chain +
+ )} + + ); + })} + + )} +
{ct.icon} @@ -541,7 +639,9 @@ export default function CoverPage() { {isFlightDelay ? (

- Enter your flight number or search popular commercial routes. The on-chain oracle will monitor flight status and automatically trigger a payout if departure is delayed > 120 minutes. + Enter your flight number or search popular commercial routes. The on-chain + oracle will monitor flight status and automatically trigger a payout if + departure is delayed > 120 minutes.

@@ -555,7 +655,11 @@ export default function CoverPage() { setFlightNumber(val); }} placeholder="e.g. BA249, JFK, British Airways" - error={flightNumberInvalid ? "Please select or enter a valid flight number" : undefined} + error={ + flightNumberInvalid + ? "Please select or enter a valid flight number" + : undefined + } />
@@ -574,16 +678,24 @@ export default function CoverPage() { onClick={() => handleSelectFlight(f)} className={`flex flex-col gap-1 rounded-lg border p-3 text-left transition-all ${ isSelected - ? "border-pm-violet bg-pm-violet/15 shadow-[0_0_15px_rgba(139,92,246,0.15)]" + ? "border-pm-violet bg-pm-violet/15 shadow-lg shadow-pm-violet/15" : "border-white/5 bg-white/[0.02] hover:border-white/15" }`} >
- {f.flightNumber} - {f.departureTime} + + {f.flightNumber} + + + {f.departureTime} + +
+
+ {f.airline} +
+
+ {f.from} → {f.to}
-
{f.airline}
-
{f.from} → {f.to}
); })} @@ -593,11 +705,16 @@ export default function CoverPage() { {selectedFlight && (
- ✓ Flight Selected: {selectedFlight.flightNumber} - Oracle Source: Chainlink FlightStats + + ✓ Flight Selected: {selectedFlight.flightNumber} + + + Oracle Source: Chainlink FlightStats +
- {selectedFlight.airline} · {selectedFlight.from} to {selectedFlight.to} ({selectedFlight.departureTime}) + {selectedFlight.airline} · {selectedFlight.from} to {selectedFlight.to}{" "} + ({selectedFlight.departureTime})
)} @@ -605,21 +722,33 @@ export default function CoverPage() { ) : (
-
Monitoring Condition
+
+ Monitoring Condition +
{ct.trigger}
- Refract contracts autonomously poll decentralized oracle feeds. When trigger criteria are met on-chain, payouts are disbursed directly to your wallet without claim forms. + Refract contracts autonomously poll decentralized oracle feeds. When + trigger criteria are met on-chain, payouts are disbursed directly to your + wallet without claim forms.
-
Resolution Oracle
-
Pyth & Chainlink Feeds
+
+ Resolution Oracle +
+
+ Pyth & Chainlink Feeds +
-
Settlement Speed
-
Sub-minute (~5s on Soroban)
+
+ Settlement Speed +
+
+ Sub-minute (~5s on Soroban) +
@@ -633,7 +762,9 @@ export default function CoverPage() { type="button" variant="primary" onClick={() => setCurrentStep(3)} - disabled={isFlightDelay && (!flightNumber || flightNumber.trim().length === 0)} + disabled={ + isFlightDelay && (!flightNumber || flightNumber.trim().length === 0) + } > Next: Amount & Duration → @@ -675,7 +806,11 @@ export default function CoverPage() { value={flightNumber} onChange={(e) => setFlightNumber(e.target.value.toUpperCase())} placeholder="BA249" - error={flightNumberInvalid ? "Enter the flight number this policy should monitor" : undefined} + error={ + flightNumberInvalid + ? "Enter the flight number this policy should monitor" + : undefined + } />
)} @@ -690,6 +825,74 @@ export default function CoverPage() { setDurationDays(Number(e.target.value))} + aria-valuetext={`${durationDays} days`} + className="pm-slider w-full" + style={{ "--pct": `${(durationDays / 365) * 100}%` } as React.CSSProperties} + /> +
+ 1 day + 1 year +
+ + +
+ + +
+ + )} + + {/* STEP 4: Review & Confirm */} + {currentStep === 4 && ct && ( + +
+ Step 4 of 4: Review & Confirm +
+
+ {[ + { label: "Coverage type", value: ct.name }, + ...(isFlightDelay + ? [{ label: "Flight", value: flightNumber.trim() || "—" }] + : []), + { + label: "Coverage amount", + value: formatUsd(parseFloat(coverageAmount || "0")), + }, + { label: "Duration", value: `${durationDays} days` }, + { label: "Premium", value: formatUsd(premium) }, + ].map((item) => ( +
+
{item.label}
+
{item.value}
+
+ ))} +
+

+ Use the quote panel to buy this policy or add it to your cart. +

+
+ +
+ + )} + {/* Right: Quote panel */}
@@ -699,20 +902,49 @@ export default function CoverPage() { ) : submission.status === "success" ? ( -
- - Coverage purchased -
+ {currentPending?.status === "failed" || currentPending?.status === "cancelled" ? ( +
+ + + {currentPending.status === "failed" + ? "Purchase failed" + : "Policy cancelled"} + +
+ ) : submission.demo || currentPending?.status === "active" ? ( +
+ + + {submission.demo ? "Coverage purchased" : "Coverage active"} + +
+ ) : ( +
+ Purchase submitted +
+ )} + {submission.demo && (

- Demo mode: the Refract API wasn't reachable, so this was simulated client-side — - no real transaction was built or submitted. + Demo mode: the Refract API wasn't reachable, so this was simulated + client-side — no real transaction was built or submitted.

)} + + {currentPending && ( +
+ +
+ )} +
Policy ID
-
{submission.result.policy.id.slice(0, 13)}…
+
+ {submission.result.policy.id.slice(0, 13)}… +
Coverage
@@ -720,31 +952,50 @@ export default function CoverPage() {
Holder
-
{truncateAddress(submission.result.policy.holder)}
+
+ {truncateAddress(submission.result.policy.holder)} +
{submission.txHash && (
Transaction
-
{truncateAddress(submission.txHash)}
+
+ {truncateAddress(submission.txHash)} +
+
+ )} + {currentPending?.premiumCharged && ( +
+
Premium charged
+
+ {formatUsd(fromStroops(currentPending.premiumCharged))} +
)}
+
- + {(submission.demo || currentPending?.status === "active") && ( + + )}
@@ -761,7 +1012,9 @@ export default function CoverPage() { ) : (
- +
{ct.name}
{durationDays}-day policy
@@ -771,24 +1024,26 @@ export default function CoverPage() {
Risk level - + {ct.riskLevel}
{[ - { label: "Coverage amount", value: formatUsd(parseFloat(coverageAmount || "0")) }, + { + label: "Coverage amount", + value: formatUsd(parseFloat(coverageAmount || "0")), + }, { label: "Annual rate", value: `${ct.baseRatePct}%` }, { label: "Duration", value: `${durationDays} days` }, { label: "Expires", value: expiryDate }, @@ -803,13 +1058,19 @@ export default function CoverPage() {
Total premium - {formatUsd(premium)} + + {formatUsd(premium)} + +
+
+ One-time payment · USDC
-
One-time payment · USDC
{networkFee && (
Network fee - {networkFee} est. network fee + + {networkFee} est. network fee +
)}
@@ -820,8 +1081,17 @@ export default function CoverPage() { size="lg" block data-tour-id="cover-buy" - disabled={amountInvalid || flightNumberInvalid || !isCorrectNetwork} - title={!isCorrectNetwork ? `Switch your wallet to ${expectedNetwork} to continue` : undefined} + disabled={ + amountInvalid || + flightNumberInvalid || + !isCorrectNetwork || + !!duplicatePending + } + title={ + !isCorrectNetwork + ? `Switch your wallet to ${expectedNetwork} to continue` + : undefined + } loading={ submission.status === "submitting" || submission.status === "signing" || @@ -836,18 +1106,28 @@ export default function CoverPage() { : !isCorrectNetwork ? `Switch to ${expectedNetwork}` : wallet.status === "connected" - ? "Buy Coverage" - : wallet.status === "connecting" - ? "Connecting…" - : "Connect & Buy Coverage"} + ? "Buy Coverage" + : wallet.status === "connecting" + ? "Connecting…" + : "Connect & Buy Coverage"} {wallet.status !== "connected" && (

- After you connect, your purchase continues automatically — you'll still confirm it in Freighter. + After you connect, your purchase continues automatically — you'll still + confirm it in Freighter.

)} - {submission.status === "error" && } + {duplicatePending && ( +

+ An identical purchase is still being confirmed. Wait for it to settle before + buying again. +

+ )} + + {submission.status === "error" && ( + + )} + )} +
+ + + ))} + + + )} + {cart.length > 0 && (
-

Cart

+

+ Cart +

{cart.filter((i) => i.status === "confirmed").length} of{" "} {cart.filter((i) => i.status !== "skipped").length} confirmed @@ -889,20 +1203,25 @@ export default function CoverPage() { {index + 1}. {item.typeName} - + {CART_STATUS_LABEL[item.status].label}
- {formatUsd(fromStroops(item.params.coverageAmount))} · {item.params.durationDays} days ·{" "} - {formatUsd(item.premium)} premium + {formatUsd(fromStroops(item.params.coverageAmount))} ·{" "} + {item.params.durationDays} days · {formatUsd(item.premium)} premium {item.demo && " · demo"} {item.txHash && ` · tx ${truncateAddress(item.txHash)}`}
- {item.error && item.status === "failed" && } + {item.error && item.status === "failed" && ( + + )} {item.error && item.status === "unconfirmed" && (

- This may still have gone through — check your Dashboard before buying it again. ({item.error}) + This may still have gone through — check your Dashboard before buying it + again. ({item.error})

)} {!checkingOut && item.status !== "processing" && ( @@ -911,7 +1230,11 @@ export default function CoverPage() { @@ -920,7 +1243,11 @@ export default function CoverPage() { @@ -928,7 +1255,9 @@ export default function CoverPage() { @@ -949,7 +1278,9 @@ export default function CoverPage() { {wallet.status !== "connected" ? "Connect to Checkout" : `Checkout ${cart.filter((i) => i.status === "queued").length} ${ - cart.filter((i) => i.status === "queued").length === 1 ? "policy" : "policies" + cart.filter((i) => i.status === "queued").length === 1 + ? "policy" + : "policies" }`} @@ -967,7 +1298,14 @@ export default function CoverPage() { { label: "Action", value: "Buy coverage" }, { label: "Coverage type", value: ct.name }, { label: "Coverage amount", value: formatUsd(parseFloat(coverageAmount || "0")) }, - { label: "Duration", value: <>{durationDays} days · Expires {expiryDate} }, + { + label: "Duration", + value: ( + <> + {durationDays} days · Expires {expiryDate} + + ), + }, ...(isFlightDelay ? [{ label: "Flight", value: flightNumber.trim() }] : []), ]} total={{ label: "Estimated total cost", value: formatUsd(premium) }} diff --git a/src/components/cover/PolicyProgress.tsx b/src/components/cover/PolicyProgress.tsx new file mode 100644 index 0000000..5646f68 --- /dev/null +++ b/src/components/cover/PolicyProgress.tsx @@ -0,0 +1,85 @@ +"use client"; + +import type { PendingPolicy, PendingPolicyStatus } from "@/lib/realtime/pendingPolicies"; +import { useServerRelativeTime } from "@/hooks/useServerRelativeTime"; + +const STEPS: { status: PendingPolicyStatus; label: string }[] = [ + { status: "submitted", label: "Submitted" }, + { status: "accepted", label: "Accepted" }, + { status: "active", label: "Active" }, +]; + +const STEP_INDEX: Record = { + submitted: 0, + accepted: 1, + active: 2, + failed: -1, + cancelled: -1, +}; + +const REASON_LABEL: Record = { + holder_requested: "cancelled by the holder", + pool_insufficient: "pool liquidity was insufficient", + expired: "the policy expired", + oracle_triggered: "the oracle trigger fired", +}; + +function describe(entry: PendingPolicy): string { + switch (entry.status) { + case "submitted": + return "Transaction submitted — waiting for the network to pick it up."; + case "accepted": + return "Accepted by the network — waiting for the policy to be issued."; + case "active": + return entry.confirmedBy === "rest" + ? "Active. Your transaction went through; live confirmation didn't arrive in time, so this is based on the transaction receipt." + : "Active and confirmed on-chain."; + case "failed": + return `The purchase failed${entry.reason ? `: ${entry.reason}` : ""}. You have not been charged for coverage.`; + case "cancelled": + return `Policy cancelled${entry.reason ? ` — ${REASON_LABEL[entry.reason] ?? entry.reason.replace(/_/g, " ")}` : ""}.`; + } +} + +/** Submitted → accepted → active progression for one purchase, driven by `policy.
` events. */ +export function PolicyProgress({ entry }: { entry: PendingPolicy }) { + const updated = useServerRelativeTime(entry.lastEventTs); + const current = STEP_INDEX[entry.status]; + const failed = entry.status === "failed" || entry.status === "cancelled"; + + return ( +
+ {!failed && ( +
    + {STEPS.map((step, i) => { + const done = i <= current; + return ( +
  1. +
    2. + ); + })} +
+ )} +

+ {describe(entry)} + {updated && · updated {updated}} +

+
+ ); +} diff --git a/src/components/ui/index.ts b/src/components/ui/index.ts index 663cd8e..b4139e8 100644 --- a/src/components/ui/index.ts +++ b/src/components/ui/index.ts @@ -9,3 +9,8 @@ export { Container } from "./Container"; export { AddressInput } from "./AddressInput"; export { LiveCountdown } from "./LiveCountdown"; export { RelativeDate } from "./RelativeDate"; +export { AsyncState } from "./AsyncState"; +export { SuccessBurst } from "./SuccessBurst"; +export { QuickAmountChips } from "./QuickAmountChips"; +export { Tour, TourReplayButton, useTour } from "./Tour"; +export type { TourStep } from "./Tour"; diff --git a/src/hooks/usePolicyStream.test.ts b/src/hooks/usePolicyStream.test.ts new file mode 100644 index 0000000..f3076f1 --- /dev/null +++ b/src/hooks/usePolicyStream.test.ts @@ -0,0 +1,144 @@ +// @vitest-environment jsdom +import { createElement, act } from "react"; +import { createRoot, type Root } from "react-dom/client"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { usePolicyStream } from "./usePolicyStream"; +import { PolicyProgress } from "@/components/cover/PolicyProgress"; +import { RealtimeClient } from "@/lib/realtime/client"; +import { FakeWebSocket } from "@/lib/realtime/dev/fakeSocket"; +import { PENDING_WAIT_MS, usePendingPolicyStore } from "@/lib/realtime/pendingPolicies"; +import { parsePolicyEvent } from "@/lib/realtime/policyEvents"; +import { policyTopic } from "@/lib/realtime/topics"; + +(globalThis as { IS_REACT_ACT_ENVIRONMENT?: boolean }).IS_REACT_ACT_ENVIRONMENT = true; + +const HOLDER = "GHOLDERADDRESS"; + +let container: HTMLDivElement; +let root: Root; +let client: RealtimeClient; +const load = () => Promise.resolve({ getRealtimeClient: () => client, parsePolicyEvent }); + +function Harness({ address }: { address: string }) { + const entries = usePolicyStream(address, { load }); + return createElement( + "div", + null, + entries.map((entry) => createElement(PolicyProgress, { key: entry.key, entry })) + ); +} + +async function render(address = HOLDER) { + await act(async () => { + root.render(createElement(Harness, { address })); + }); +} + +const statuses = () => + [...container.querySelectorAll("[data-testid=policy-progress]")].map((el) => + el.getAttribute("data-status") + ); + +async function push(payload: Record) { + await act(async () => { + FakeWebSocket.latest().push(policyTopic(HOLDER), { + holder: HOLDER, + ts: Date.now(), + ...payload, + }); + }); +} + +function submit(txHash = "tx1") { + act(() => { + usePendingPolicyStore + .getState() + .addOptimistic({ txHash, holder: HOLDER, policyId: "p1", coverageTypeName: "Depeg" }); + }); +} + +beforeEach(() => { + FakeWebSocket.reset(); + localStorage.clear(); + usePendingPolicyStore.setState({ entries: [] }); + client = new RealtimeClient({ url: "ws://test", WebSocketImpl: FakeWebSocket as never }); + container = document.createElement("div"); + document.body.appendChild(container); + root = createRoot(container); +}); + +afterEach(() => { + act(() => root.unmount()); + client.disconnect(); + container.remove(); + vi.useRealTimers(); +}); + +describe("usePolicyStream + cover progress (integration)", () => { + it("does not subscribe while nothing is pending", async () => { + await render(); + expect(FakeWebSocket.instances).toHaveLength(0); + }); + + it("renders submitted → accepted → active as events arrive", async () => { + await render(); + submit(); + await act(async () => {}); + FakeWebSocket.latest().open(); + expect(statuses()).toEqual(["submitted"]); + + await push({ type: "accepted", txHash: "tx1" }); + expect(statuses()).toEqual(["accepted"]); + + await push({ type: "issued", txHash: "tx1", policyId: "p1" }); + expect(statuses()).toEqual(["active"]); + expect(container.textContent).toContain("Active and confirmed on-chain"); + }); + + it("rolls back to failed on a failure event", async () => { + await render(); + submit(); + await act(async () => {}); + FakeWebSocket.latest().open(); + await push({ type: "failed", txHash: "tx1", reason: "transaction reverted" }); + expect(statuses()).toEqual(["failed"]); + expect(container.textContent).toContain("transaction reverted"); + }); + + it("ignores malformed payloads and events for another wallet", async () => { + await render(); + submit(); + await act(async () => {}); + FakeWebSocket.latest().open(); + await push({ type: "issued" }); // missing ids + await push({ type: "issued", txHash: "tx1", policyId: "p1", holder: "GSOMEONEELSE" }); + expect(statuses()).toEqual(["submitted"]); + }); + + it("resumes a pending purchase after navigating away and back, without duplicating it", async () => { + submit(); + await render(); + await act(async () => {}); + act(() => root.unmount()); + + // Back on /cover: a fresh mount picks up the persisted pending entry and subscribes. + root = createRoot(container); + await render(); + await act(async () => {}); + FakeWebSocket.latest().open(); + await push({ type: "issued", txHash: "tx1", policyId: "p1" }); + expect(statuses()).toEqual(["active"]); + }); + + it("settles from the REST result when no event arrives in time", async () => { + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout", "Date"] }); + await render(); + submit(); + await act(async () => {}); + await act(async () => { + vi.advanceTimersByTime(PENDING_WAIT_MS + 1); + }); + expect(statuses()).toEqual(["active"]); + expect(container.textContent).toContain("live confirmation didn't arrive in time"); + }); +}); diff --git a/src/hooks/usePolicyStream.ts b/src/hooks/usePolicyStream.ts new file mode 100644 index 0000000..5c626ae --- /dev/null +++ b/src/hooks/usePolicyStream.ts @@ -0,0 +1,105 @@ +import { useEffect, useMemo } from "react"; +import { loadRealtime } from "@/lib/realtime/load"; +import type { PolicyEvent } from "@/lib/realtime/policyEvents"; +import { + isTerminal, + PENDING_POLICIES_STORAGE_KEY, + PENDING_WAIT_MS, + usePendingPolicyStore, + type PendingPolicy, +} from "@/lib/realtime/pendingPolicies"; +import { policyTopic } from "@/lib/realtime/topics"; + +/** The parts of the lazily-loaded realtime module this hook uses. */ +export interface PolicyStreamModule { + getRealtimeClient(): { + subscribe(topic: string, handler: (payload: unknown) => void): () => void; + }; + parsePolicyEvent(payload: unknown): PolicyEvent | null; +} + +export interface UsePolicyStreamOptions { + /** + * Loads the realtime module. Defaults to `loadRealtime()`, so the client + * and the payload validator stay out of the initial bundle. Must be stable. + */ + load?: () => Promise; +} + +/** + * Streams `policy.
` into the pending-policy store (issue #246). + * + * Subscribes while the wallet has at least one unsettled purchase (so on + * submit, or on mount when one is already pending) and unsubscribes once all + * have settled. Each unsettled entry also gets a bounded wait: if no event + * settles it within `PENDING_WAIT_MS`, the REST result stands. + * + * Returns this wallet's entries, newest first. + */ +export function usePolicyStream( + address: string | null | undefined, + options: UsePolicyStreamOptions = {} +): PendingPolicy[] { + const all = usePendingPolicyStore((s) => s.entries); + const applyEvent = usePendingPolicyStore((s) => s.applyEvent); + const settle = usePendingPolicyStore((s) => s.settleViaRest); + const load = options.load ?? loadRealtime; + + const entries = useMemo( + () => (address ? all.filter((p) => p.holder === address) : []), + [all, address] + ); + const unsettled = entries.filter((p) => !isTerminal(p.status)); + const hasUnsettled = unsettled.length > 0; + // Re-arm timers only when the set of unsettled entries changes. + const unsettledKey = unsettled.map((p) => `${p.key}:${p.createdAt}`).join("|"); + + // Live subscription. + useEffect(() => { + if (!address || !hasUnsettled) return; + let unsubscribe: (() => void) | null = null; + let cancelled = false; + + load() + .then(({ getRealtimeClient, parsePolicyEvent }) => { + if (cancelled) return; + unsubscribe = getRealtimeClient().subscribe(policyTopic(address), (payload) => { + const event = parsePolicyEvent(payload); + // Malformed payloads and events for another wallet are ignored. + if (event && event.holder === address) applyEvent(event); + }); + }) + .catch(() => { + // Realtime unavailable: the bounded wait below still settles entries. + }); + + return () => { + cancelled = true; + unsubscribe?.(); + }; + // `load` is expected to be stable (module-level or memoized). + // eslint-disable-next-line react-hooks/exhaustive-deps + }, [address, hasUnsettled, applyEvent]); + + // Bounded wait per unsettled entry. + useEffect(() => { + if (!unsettledKey) return; + const now = Date.now(); + const timers = unsettled.map((p) => + setTimeout(() => settle(p.key), Math.max(0, p.createdAt + PENDING_WAIT_MS - now)) + ); + return () => timers.forEach(clearTimeout); + // eslint-disable-next-line react-hooks/exhaustive-deps + }, [unsettledKey, settle]); + + // Cross-tab sync: another tab bought or settled a policy. + useEffect(() => { + const onStorage = (e: StorageEvent) => { + if (e.key === PENDING_POLICIES_STORAGE_KEY) void usePendingPolicyStore.persist.rehydrate(); + }; + window.addEventListener("storage", onStorage); + return () => window.removeEventListener("storage", onStorage); + }, []); + + return entries; +} diff --git a/src/hooks/useServerRelativeTime.ts b/src/hooks/useServerRelativeTime.ts new file mode 100644 index 0000000..d4d39c7 --- /dev/null +++ b/src/hooks/useServerRelativeTime.ts @@ -0,0 +1,40 @@ +import { useCallback, useSyncExternalStore } from "react"; +import { + defaultClock, + normalizeServerTs, + type ClockOffsetEstimator, +} from "@/lib/realtime/clockOffset"; +import { formatRelativeTime } from "@/lib/format"; + +/** + * "4 minutes ago" for a server timestamp, corrected for device clock skew + * (issue #247). Use this for every realtime-driven relative time. + * + * Re-renders only when the displayed text changes: offset updates and the + * tick both feed a string snapshot, and React skips equal snapshots. + */ +export function formatServerRelativeTime( + serverTs: number, + clock: ClockOffsetEstimator = defaultClock +): string { + return formatRelativeTime(normalizeServerTs(serverTs, clock)); +} + +export function useServerRelativeTime( + serverTs: number | null | undefined, + { tickMs = 15_000, clock = defaultClock }: { tickMs?: number; clock?: ClockOffsetEstimator } = {} +): string | null { + const subscribe = useCallback( + (onChange: () => void) => { + const unsubscribe = clock.subscribe(onChange); + const timer = setInterval(onChange, tickMs); + return () => { + unsubscribe(); + clearInterval(timer); + }; + }, + [clock, tickMs] + ); + const getSnapshot = () => (serverTs == null ? null : formatServerRelativeTime(serverTs, clock)); + return useSyncExternalStore(subscribe, getSnapshot, getSnapshot); +} diff --git a/src/lib/realtime/client.test.ts b/src/lib/realtime/client.test.ts new file mode 100644 index 0000000..05a1874 --- /dev/null +++ b/src/lib/realtime/client.test.ts @@ -0,0 +1,217 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { PERMESSAGE_DEFLATE_OPTIONS, RealtimeClient } from "./client"; +import { ClockOffsetEstimator } from "./clockOffset"; +import { MAX_MESSAGE_BYTES, type RealtimeProtocolError } from "./codec"; +import { FakeWebSocket } from "./dev/fakeSocket"; +import { INTERNAL_TOPICS } from "./topics"; + +function makeClient(opts: { maxMessageBytes?: number; clock?: ClockOffsetEstimator } = {}) { + return new RealtimeClient({ + url: "ws://test/ws", + WebSocketImpl: FakeWebSocket as never, + heartbeatIntervalMs: 1_000, + ...opts, + }); +} + +beforeEach(() => { + FakeWebSocket.reset(); + vi.spyOn(console, "warn").mockImplementation(() => {}); +}); + +afterEach(() => { + vi.restoreAllMocks(); + vi.useRealTimers(); +}); + +describe("RealtimeClient", () => { + it("delivers payloads and envelopes to topic subscribers", () => { + const client = makeClient(); + const handler = vi.fn(); + client.subscribe("oracle_update", handler); + const ws = FakeWebSocket.latest(); + ws.open(); + ws.push("oracle_update", { price: 1 }, 123); + ws.push("other", { price: 2 }); + expect(handler).toHaveBeenCalledTimes(1); + expect(handler).toHaveBeenCalledWith( + { price: 1 }, + { topic: "oracle_update", payload: { price: 1 }, ts: 123 } + ); + }); + + it("stops delivering after unsubscribe", () => { + const client = makeClient(); + const handler = vi.fn(); + const off = client.subscribe("t", handler); + FakeWebSocket.latest().open(); + off(); + FakeWebSocket.latest().push("t", 1); + expect(handler).not.toHaveBeenCalled(); + }); + + it("keeps delivering to other subscribers when one throws", () => { + vi.spyOn(console, "error").mockImplementation(() => {}); + const client = makeClient(); + const good = vi.fn(); + client.subscribe("t", () => { + throw new Error("boom"); + }); + client.subscribe("t", good); + FakeWebSocket.latest().open(); + FakeWebSocket.latest().push("t", 1); + expect(good).toHaveBeenCalled(); + }); + + describe("payload size guard", () => { + it("drops oversize frames, counts them and reports a protocol error", () => { + const client = makeClient({ maxMessageBytes: 64 }); + const handler = vi.fn(); + const onError = vi.fn<(e: RealtimeProtocolError) => void>(); + client.subscribe("t", handler); + client.onProtocolError(onError); + const ws = FakeWebSocket.latest(); + ws.open(); + + ws.push("t", "x".repeat(100)); + expect(handler).not.toHaveBeenCalled(); + expect(client.getDiagnostics().oversizeDrops).toBe(1); + expect(onError).toHaveBeenCalledTimes(1); + expect(onError.mock.calls[0]![0].code).toBe("oversize"); + expect(client.getDiagnostics().lastError?.code).toBe("oversize"); + + // The connection keeps working for normal frames. + ws.push("t", "ok"); + expect(handler).toHaveBeenCalledWith("ok", expect.anything()); + }); + + it("accepts a frame exactly at the limit and rejects one byte over", () => { + const client = makeClient(); + const handler = vi.fn(); + client.subscribe("t", handler); + const ws = FakeWebSocket.latest(); + ws.open(); + const shell = JSON.stringify({ topic: "t", payload: "" }).length; + ws.receiveRaw(JSON.stringify({ topic: "t", payload: "x".repeat(MAX_MESSAGE_BYTES - shell) })); + ws.receiveRaw( + JSON.stringify({ topic: "t", payload: "x".repeat(MAX_MESSAGE_BYTES - shell + 1) }) + ); + expect(handler).toHaveBeenCalledTimes(1); + expect(client.getDiagnostics().oversizeDrops).toBe(1); + }); + + it("counts malformed frames separately", () => { + const client = makeClient(); + client.subscribe("t", vi.fn()); + const ws = FakeWebSocket.latest(); + ws.open(); + ws.receiveRaw("{not json"); + expect(client.getDiagnostics()).toMatchObject({ malformedDrops: 1, oversizeDrops: 0 }); + }); + + it("tracks decoded byte counts", () => { + const client = makeClient(); + client.subscribe("t", vi.fn()); + const ws = FakeWebSocket.latest(); + ws.open(); + const frame = JSON.stringify({ topic: "t", payload: "hello" }); + ws.receiveRaw(frame); + expect(client.getDiagnostics()).toMatchObject({ + framesReceived: 1, + bytesReceived: frame.length, + }); + }); + }); + + describe("permessage-deflate", () => { + it("asks for deflate with no context takeover", () => { + makeClient().connect(); + expect(FakeWebSocket.latest().options).toEqual(PERMESSAGE_DEFLATE_OPTIONS); + }); + + it("reports compression when the server accepts the extension", () => { + const client = makeClient(); + client.connect(); + FakeWebSocket.latest().open( + "permessage-deflate; client_no_context_takeover; server_no_context_takeover" + ); + expect(client.getDiagnostics()).toMatchObject({ + compressionActive: true, + negotiatedExtensions: + "permessage-deflate; client_no_context_takeover; server_no_context_takeover", + }); + }); + + it("keeps working when the server declines compression", () => { + const client = makeClient(); + const handler = vi.fn(); + client.subscribe("t", handler); + const ws = FakeWebSocket.latest(); + ws.open(""); + ws.push("t", 1); + expect(client.getDiagnostics()).toMatchObject({ + compressionActive: false, + negotiatedExtensions: "", + }); + expect(handler).toHaveBeenCalled(); + }); + + it("records server-reported compressed/uncompressed byte counts", () => { + const client = makeClient(); + client.connect(); + const ws = FakeWebSocket.latest(); + ws.open("permessage-deflate"); + ws.push(INTERNAL_TOPICS.WIRE_STATS, { compressedBytes: 1_200, uncompressedBytes: 9_600 }); + expect(client.getDiagnostics().wireStats).toEqual({ + compressedBytes: 1_200, + uncompressedBytes: 9_600, + }); + }); + }); + + describe("heartbeat / clock sync", () => { + it("pings on open and on every heartbeat", () => { + vi.useFakeTimers(); + const client = makeClient(); + client.connect(); + const ws = FakeWebSocket.latest(); + ws.open(); + expect(ws.sent).toHaveLength(1); + expect(ws.sent[0]).toMatchObject({ + topic: INTERNAL_TOPICS.PING, + payload: { t0: expect.any(Number) }, + }); + vi.advanceTimersByTime(2_000); + expect(ws.sent).toHaveLength(3); + client.disconnect(); + vi.advanceTimersByTime(5_000); + expect(ws.sent).toHaveLength(3); + }); + + it("feeds pong round trips into the clock estimator and never exposes internal topics", () => { + const clock = new ClockOffsetEstimator(); + const client = makeClient({ clock }); + const handler = vi.fn(); + client.subscribe(INTERNAL_TOPICS.PONG, handler); + const ws = FakeWebSocket.latest(); + ws.open(); + const t0 = Date.now(); + ws.push(INTERNAL_TOPICS.PONG, { t0, serverTs: t0 + 60_000 }); + expect(clock.getSnapshot().sampleCount).toBe(1); + expect(Math.abs(clock.getSnapshot().offsetMs - 60_000)).toBeLessThan(1_000); + expect(handler).not.toHaveBeenCalled(); + }); + }); + + it("reconnects with backoff after an unexpected close", () => { + vi.useFakeTimers(); + const client = makeClient(); + client.connect(); + FakeWebSocket.latest().open(); + FakeWebSocket.latest().close(); + expect(FakeWebSocket.instances).toHaveLength(1); + vi.advanceTimersByTime(1_000); + expect(FakeWebSocket.instances).toHaveLength(2); + client.disconnect(); + }); +}); diff --git a/src/lib/realtime/client.ts b/src/lib/realtime/client.ts index 8be8d3c..e8dd292 100644 --- a/src/lib/realtime/client.ts +++ b/src/lib/realtime/client.ts @@ -1,61 +1,287 @@ import { API_BASE_URL } from "@/lib/api/client"; +import { ClockOffsetEstimator, defaultClock } from "./clockOffset"; +import { + decodeFrame, + MAX_MESSAGE_BYTES, + RealtimeProtocolError, + type RealtimeEnvelope, +} from "./codec"; +import { INTERNAL_TOPICS, isInternalTopic } from "./topics"; -type Handler = (payload: unknown) => void; +/** + * Reconnecting WebSocket client with a subscribe-by-topic API. + * + * Do not import this module from UI code directly: load it through + * `loadRealtime()` in ./load so it stays out of the initial bundle (#248). + * + * Backend contract (docs/realtime.md): + * - JSON text frames shaped `{ topic, payload, ts? }`. + * - Answers `refract.ping { t0 }` with `refract.pong { t0, serverTs }`. + * - Should accept `permessage-deflate` with no context takeover. + * + * The connection is best-effort: callers keep REST as the source of truth + * for initial state. + */ + +export type TopicHandler = (payload: unknown, envelope: RealtimeEnvelope) => void; +export type ProtocolErrorListener = (error: RealtimeProtocolError) => void; const MAX_BACKOFF_MS = 30_000; const BASE_BACKOFF_MS = 1_000; +/** Keepalive + clock-sync ping interval. */ +export const HEARTBEAT_INTERVAL_MS = 25_000; + /** - * A small reconnecting WebSocket wrapper exposing a subscribe-by-topic API. - * The backend contract this assumes (documented, not yet implemented - * upstream): connect to `${wsUrl}`, receive JSON messages shaped - * `{ topic: string, payload: unknown }`, one topic per push-based update - * type (e.g. "oracle_update"). Callers must treat this connection as - * best-effort — always keep an independent REST fetch as the source of - * truth for initial state. + * `permessage-deflate` parameters we ask for (RFC 7692). Browsers negotiate + * the extension themselves during the HTTP upgrade and give script no way to + * change the offer, so in a browser this is documentation for the backend + * config. Non-browser WebSocket implementations that take an options argument + * (e.g. `ws` in Node tooling and tests) receive it as the third constructor + * argument; browsers ignore extra arguments. + * + * No context takeover keeps per-connection memory flat on both ends, which is + * the conservative choice for mobile clients. */ -class RealtimeClient { +export const PERMESSAGE_DEFLATE_OPTIONS = { + perMessageDeflate: { + clientNoContextTakeover: true, + serverNoContextTakeover: true, + }, +} as const; + +export interface RealtimeDiagnostics { + /** `WebSocket.extensions` from the last handshake ("" if the server declined all extensions). */ + negotiatedExtensions: string; + /** True when the server accepted `permessage-deflate`. */ + compressionActive: boolean; + /** Frames accepted and dispatched (including internal ones). */ + framesReceived: number; + /** Decoded (uncompressed) bytes of accepted frames. */ + bytesReceived: number; + /** Frames dropped for exceeding `maxMessageBytes`. */ + oversizeDrops: number; + /** Frames dropped as malformed or of an unsupported type. */ + malformedDrops: number; + /** + * Server-reported cumulative wire stats (`refract.wire_stats`). Browsers do + * not expose compressed byte counts to script, so this is the only source + * for the compressed size. Null until the server reports it. + */ + wireStats: { compressedBytes: number; uncompressedBytes: number } | null; + /** Last protocol error, for debugging. */ + lastError: { code: RealtimeProtocolError["code"]; message: string; at: number } | null; +} + +type WebSocketCtor = new ( + url: string, + protocols?: string | string[], + options?: unknown +) => WebSocket; + +export interface RealtimeClientOptions { + url?: string; + /** WebSocket implementation. Defaults to the global `WebSocket`. */ + WebSocketImpl?: WebSocketCtor; + /** Inbound frame size ceiling in bytes. Defaults to `MAX_MESSAGE_BYTES`. */ + maxMessageBytes?: number; + /** Clock estimator fed by heartbeat round trips. Defaults to the app-wide one. */ + clock?: ClockOffsetEstimator; + heartbeatIntervalMs?: number; +} + +const freshDiagnostics = (): RealtimeDiagnostics => ({ + negotiatedExtensions: "", + compressionActive: false, + framesReceived: 0, + bytesReceived: 0, + oversizeDrops: 0, + malformedDrops: 0, + wireStats: null, + lastError: null, +}); + +export function defaultRealtimeUrl(): string { + return API_BASE_URL.replace(/^http/, "ws").replace(/\/api\/v1$/, "/ws"); +} + +export class RealtimeClient { private socket: WebSocket | null = null; - private handlers = new Map>(); + private handlers = new Map>(); + private errorListeners = new Set(); private attempt = 0; private closedByUser = false; private reconnectTimer: ReturnType | null = null; + private heartbeatTimer: ReturnType | null = null; + private diagnostics = freshDiagnostics(); + + private readonly url: string; + private readonly WebSocketImpl: WebSocketCtor | undefined; + private readonly maxMessageBytes: number; + private readonly clock: ClockOffsetEstimator; + private readonly heartbeatIntervalMs: number; + + constructor(options: RealtimeClientOptions = {}) { + this.url = options.url ?? defaultRealtimeUrl(); + this.WebSocketImpl = + options.WebSocketImpl ?? + (typeof WebSocket === "undefined" ? undefined : (WebSocket as unknown as WebSocketCtor)); + this.maxMessageBytes = options.maxMessageBytes ?? MAX_MESSAGE_BYTES; + this.clock = options.clock ?? defaultClock; + this.heartbeatIntervalMs = options.heartbeatIntervalMs ?? HEARTBEAT_INTERVAL_MS; + } connect(): void { - if (typeof WebSocket === "undefined" || this.socket) return; + if (!this.WebSocketImpl || this.socket) return; this.closedByUser = false; this.open(); } + subscribe(topic: string, handler: TopicHandler): () => void { + let set = this.handlers.get(topic); + if (!set) this.handlers.set(topic, (set = new Set())); + set.add(handler); + this.connect(); + return () => { + set!.delete(handler); + if (set!.size === 0 && this.handlers.get(topic) === set) this.handlers.delete(topic); + }; + } + + /** + * Error path for rejected frames (oversize, malformed). The frame is + * dropped and counted either way; listeners are for surfacing it. + */ + onProtocolError(listener: ProtocolErrorListener): () => void { + this.errorListeners.add(listener); + return () => { + this.errorListeners.delete(listener); + }; + } + + getDiagnostics(): RealtimeDiagnostics { + return { + ...this.diagnostics, + wireStats: this.diagnostics.wireStats && { ...this.diagnostics.wireStats }, + }; + } + + disconnect(): void { + this.closedByUser = true; + if (this.reconnectTimer) clearTimeout(this.reconnectTimer); + this.reconnectTimer = null; + this.stopHeartbeat(); + this.socket?.close(); + this.socket = null; + } + + /** Handles one raw inbound frame. Public so non-WebSocket transports can reuse it. */ + handleFrame(data: unknown): void { + const result = decodeFrame(data, this.maxMessageBytes); + if (!result.ok) { + this.reportProtocolError(result.error); + return; + } + + this.diagnostics.framesReceived += 1; + this.diagnostics.bytesReceived += result.bytes; + const { envelope } = result; + + if (isInternalTopic(envelope.topic)) { + this.handleInternal(envelope); + return; + } + this.handlers.get(envelope.topic)?.forEach((handler) => { + try { + handler(envelope.payload, envelope); + } catch (err) { + // One faulty subscriber must not break delivery to the others. + if (process.env.NODE_ENV !== "production") console.error("[realtime] handler threw", err); + } + }); + } + private open(): void { - const wsUrl = API_BASE_URL.replace(/^http/, "ws").replace(/\/api\/v1$/, "/ws"); + let socket: WebSocket; try { - this.socket = new WebSocket(wsUrl); + socket = new this.WebSocketImpl!(this.url, undefined, PERMESSAGE_DEFLATE_OPTIONS); } catch { this.scheduleReconnect(); return; } + this.socket = socket; + socket.binaryType = "arraybuffer"; - this.socket.addEventListener("open", () => { + socket.addEventListener("open", () => { this.attempt = 0; + const extensions = socket.extensions ?? ""; + this.diagnostics.negotiatedExtensions = extensions; + // A server that declines compression simply omits it; nothing else changes. + this.diagnostics.compressionActive = /(^|,)\s*permessage-deflate\b/.test(extensions); + this.diagnostics.wireStats = null; + this.startHeartbeat(); }); - this.socket.addEventListener("message", (event) => { - try { - const { topic, payload } = JSON.parse(event.data as string) as { topic: string; payload: unknown }; - this.handlers.get(topic)?.forEach((handler) => handler(payload)); - } catch { - // Malformed message — ignore, don't crash the connection. - } - }); - this.socket.addEventListener("close", () => { - this.socket = null; + + socket.addEventListener("message", (event: MessageEvent) => this.handleFrame(event.data)); + + socket.addEventListener("close", () => { + if (this.socket === socket) this.socket = null; + this.stopHeartbeat(); if (!this.closedByUser) this.scheduleReconnect(); }); - this.socket.addEventListener("error", () => { - this.socket?.close(); + + socket.addEventListener("error", () => { + socket.close(); }); } + private handleInternal({ topic, payload }: RealtimeEnvelope): void { + const p = (payload ?? {}) as Record; + if (topic === INTERNAL_TOPICS.PONG) { + if (typeof p.t0 === "number" && typeof p.serverTs === "number") { + this.clock.addSample(p.t0, p.serverTs, Date.now()); + } + } else if (topic === INTERNAL_TOPICS.WIRE_STATS) { + if (typeof p.compressedBytes === "number" && typeof p.uncompressedBytes === "number") { + this.diagnostics.wireStats = { + compressedBytes: p.compressedBytes, + uncompressedBytes: p.uncompressedBytes, + }; + } + } + } + + private reportProtocolError(error: RealtimeProtocolError): void { + if (error.code === "oversize") this.diagnostics.oversizeDrops += 1; + else this.diagnostics.malformedDrops += 1; + this.diagnostics.lastError = { code: error.code, message: error.message, at: Date.now() }; + if (process.env.NODE_ENV !== "production") + console.warn(`[realtime] dropped frame: ${error.message}`); + this.errorListeners.forEach((l) => l(error)); + } + + private sendPing(): void { + if (!this.socket || this.socket.readyState !== 1 /* OPEN */) return; + try { + this.socket.send( + JSON.stringify({ topic: INTERNAL_TOPICS.PING, payload: { t0: Date.now() } }) + ); + } catch { + // Best effort: the next heartbeat retries. + } + } + + private startHeartbeat(): void { + this.stopHeartbeat(); + this.sendPing(); + this.heartbeatTimer = setInterval(() => this.sendPing(), this.heartbeatIntervalMs); + } + + private stopHeartbeat(): void { + if (this.heartbeatTimer) clearInterval(this.heartbeatTimer); + this.heartbeatTimer = null; + } + private scheduleReconnect(): void { if (this.reconnectTimer) return; const delay = Math.min(BASE_BACKOFF_MS * 2 ** this.attempt, MAX_BACKOFF_MS); @@ -65,23 +291,4 @@ class RealtimeClient { if (!this.closedByUser) this.open(); }, delay); } - - subscribe(topic: string, handler: Handler): () => void { - if (!this.handlers.has(topic)) this.handlers.set(topic, new Set()); - this.handlers.get(topic)!.add(handler); - this.connect(); - - return () => { - this.handlers.get(topic)?.delete(handler); - }; - } - - disconnect(): void { - this.closedByUser = true; - if (this.reconnectTimer) clearTimeout(this.reconnectTimer); - this.socket?.close(); - this.socket = null; - } } - -export const realtimeClient = new RealtimeClient(); diff --git a/src/lib/realtime/clockOffset.test.ts b/src/lib/realtime/clockOffset.test.ts new file mode 100644 index 0000000..6db7b71 --- /dev/null +++ b/src/lib/realtime/clockOffset.test.ts @@ -0,0 +1,186 @@ +import { describe, expect, it, vi } from "vitest"; +import { + ClockOffsetEstimator, + CLOCK_SAMPLE_WINDOW, + isServerTsStale, + serverEventAgeMs, + SKEW_WARN_THRESHOLD_MS, + type ClockSources, +} from "./clockOffset"; +import { formatServerRelativeTime } from "@/hooks/useServerRelativeTime"; + +/** Controllable wall + monotonic clocks. */ +function fakeSources(start = 1_700_000_000_000) { + let wall = start; + let mono = 0; + const sources: ClockSources = { wall: () => wall, mono: () => mono }; + return { + sources, + advance(ms: number) { + wall += ms; + mono += ms; + }, + jumpWall(ms: number) { + wall += ms; + }, + get wall() { + return wall; + }, + }; +} + +/** + * Simulates one ping/pong where the server clock is `trueOffset` ahead and + * the request/response legs take `up`/`down` ms. + */ +function roundTrip( + est: ClockOffsetEstimator, + clk: ReturnType, + trueOffset: number, + up: number, + down: number +) { + const t0 = clk.wall; + clk.advance(up); + const serverTs = clk.wall + trueOffset; + clk.advance(down); + return est.addSample(t0, serverTs, clk.wall); +} + +describe("ClockOffsetEstimator", () => { + it("starts with no estimate", () => { + const est = new ClockOffsetEstimator(fakeSources().sources); + expect(est.getSnapshot()).toMatchObject({ offsetMs: 0, sampleCount: 0, skewed: false }); + expect(est.getSnapshot().uncertaintyMs).toBe(Infinity); + }); + + it("recovers the exact offset with symmetric latency", () => { + const clk = fakeSources(); + const est = new ClockOffsetEstimator(clk.sources); + roundTrip(est, clk, 5_000, 40, 40); + expect(est.getSnapshot()).toMatchObject({ offsetMs: 5_000, uncertaintyMs: 40, sampleCount: 1 }); + }); + + it("stays within ±rtt/2 under asymmetric latency", () => { + const clk = fakeSources(); + const est = new ClockOffsetEstimator(clk.sources); + roundTrip(est, clk, -3_000, 10, 90); // 100 ms rtt, heavily asymmetric + const { offsetMs, uncertaintyMs } = est.getSnapshot(); + expect(Math.abs(offsetMs - -3_000)).toBeLessThanOrEqual(uncertaintyMs); + expect(uncertaintyMs).toBe(50); + }); + + it("ignores a single slow outlier sample", () => { + const clk = fakeSources(); + const est = new ClockOffsetEstimator(clk.sources); + for (let i = 0; i < 4; i++) roundTrip(est, clk, 2_000, 20, 20); + // 4 s RTT, almost all of it on one leg: a naive estimate is off by ~2 s. + roundTrip(est, clk, 2_000, 4_000, 20); + expect(est.getSnapshot().offsetMs).toBe(2_000); + expect(est.getSnapshot().sampleCount).toBe(5); + }); + + it("ignores a single wild low-latency sample (median)", () => { + const clk = fakeSources(); + const est = new ClockOffsetEstimator(clk.sources); + roundTrip(est, clk, 1_000, 20, 20); + roundTrip(est, clk, 1_000, 20, 20); + roundTrip(est, clk, 60_000, 20, 20); + expect(est.getSnapshot().offsetMs).toBe(1_000); + }); + + it("rejects negative and excessive round trips", () => { + const clk = fakeSources(); + const est = new ClockOffsetEstimator(clk.sources); + expect(est.addSample(1_000, 5_000, 900)).toBe(false); + expect(est.addSample(0, 5_000, 60_000)).toBe(false); + expect(est.getSnapshot().sampleCount).toBe(0); + }); + + it("keeps a bounded window", () => { + const clk = fakeSources(); + const est = new ClockOffsetEstimator(clk.sources); + for (let i = 0; i < CLOCK_SAMPLE_WINDOW + 5; i++) roundTrip(est, clk, 0, 10, 10); + expect(est.getSnapshot().sampleCount).toBe(CLOCK_SAMPLE_WINDOW); + }); + + it("invalidates samples when the wall clock jumps (sleep/resume, manual change)", () => { + const clk = fakeSources(); + const est = new ClockOffsetEstimator(clk.sources); + roundTrip(est, clk, 10_000, 20, 20); + roundTrip(est, clk, 10_000, 20, 20); + clk.jumpWall(10_000); // the user fixed their clock; server offset is now ~0 + roundTrip(est, clk, 0, 20, 20); + expect(est.getSnapshot()).toMatchObject({ offsetMs: 0, sampleCount: 1 }); + }); + + it("only notifies subscribers on meaningful changes", () => { + const clk = fakeSources(); + const est = new ClockOffsetEstimator(clk.sources); + const listener = vi.fn(); + est.subscribe(listener); + + roundTrip(est, clk, 1_000, 20, 20); + expect(listener).toHaveBeenCalledTimes(1); + + // Jitter well under the notify threshold: the estimate refines silently. + roundTrip(est, clk, 1_030, 20, 20); + roundTrip(est, clk, 990, 20, 20); + expect(listener).toHaveBeenCalledTimes(1); + expect(est.getSnapshot().sampleCount).toBe(3); + + // A real change (> threshold) notifies once the median moves. + for (let i = 0; i < 5; i++) roundTrip(est, clk, 5_000, 20, 20); + expect(listener.mock.calls.length).toBeGreaterThan(1); + expect(est.getSnapshot().offsetMs).toBe(5_000); + }); + + it("flags large offsets as a diagnostic", () => { + const clk = fakeSources(); + const est = new ClockOffsetEstimator(clk.sources); + roundTrip(est, clk, SKEW_WARN_THRESHOLD_MS + 5_000, 20, 20); + expect(est.getSnapshot().skewed).toBe(true); + }); +}); + +describe("normalized time helpers", () => { + it("renders correct relative times on a device whose clock is 3 hours behind", () => { + const realNow = Date.UTC(2026, 9, 21, 12, 0, 0); + const deviceNow = realNow - 3 * 60 * 60 * 1000; + vi.useFakeTimers(); + vi.setSystemTime(deviceNow); + try { + const clk = fakeSources(deviceNow); + const est = new ClockOffsetEstimator(clk.sources); + roundTrip(est, clk, 3 * 60 * 60 * 1000, 25, 25); + vi.setSystemTime(clk.wall); + + // An event the server stamped 4 s ago (on the server's clock). + const serverTs = clk.wall + 3 * 60 * 60 * 1000 - 4_000; + expect(serverEventAgeMs(serverTs, clk.wall, est)).toBe(4_000); + expect(formatServerRelativeTime(serverTs, est)).toBe("just now"); + expect(formatServerRelativeTime(serverTs - 5 * 60_000, est)).toBe("5 minutes ago"); + + // Without correction a 5-minute-old event looks like it's from the + // future (clamped to "just now"), and a fresh one looks 3 hours ahead. + const uncorrected = new ClockOffsetEstimator(clk.sources); + expect(formatServerRelativeTime(serverTs - 5 * 60_000, uncorrected)).toBe("just now"); + expect(serverEventAgeMs(serverTs, clk.wall, uncorrected)).toBe(0); + expect(isServerTsStale(serverTs, 30_000, clk.wall, est)).toBe(false); + expect(isServerTsStale(serverTs - 60_000, 30_000, clk.wall, est)).toBe(true); + } finally { + vi.useRealTimers(); + } + }); + + it("does not flag fresh data as stale when the device clock is ahead", () => { + const clk = fakeSources(); + const est = new ClockOffsetEstimator(clk.sources); + roundTrip(est, clk, -10 * 60_000, 20, 20); // device 10 minutes fast + const serverTs = clk.wall - 10 * 60_000 - 1_000; + expect(isServerTsStale(serverTs, 30_000, clk.wall, est)).toBe(false); + expect(isServerTsStale(serverTs, 30_000, clk.wall, new ClockOffsetEstimator(clk.sources))).toBe( + true + ); + }); +}); diff --git a/src/lib/realtime/clockOffset.ts b/src/lib/realtime/clockOffset.ts new file mode 100644 index 0000000..21f4955 --- /dev/null +++ b/src/lib/realtime/clockOffset.ts @@ -0,0 +1,228 @@ +/** + * Server/client clock-offset estimation (issue #247). + * + * Realtime events carry server timestamps. A device with a wrong clock would + * render a just-arrived event as "in 3 hours" or flag fresh data as stale, so + * every server timestamp goes through `normalizeServerTs` before it is + * rendered or compared with `Date.now()`. + * + * Each heartbeat ping/pong gives one sample (NTP-style mid-point estimate): + * + * t0 client wall clock when the ping was sent + * serverTs server wall clock when it answered + * t3 client wall clock when the pong arrived + * + * rtt = t3 - t0 + * offset = serverTs - (t0 + t3) / 2 (positive => server is ahead) + * + * The true offset lies within ±rtt/2 of that estimate (latency may be + * asymmetric), so low-RTT samples are the most trustworthy. The estimate is + * the median offset of the lowest-RTT samples in a small sliding window, which + * keeps a single slow or wild sample from moving it. + * + * The estimator does not know about any transport: WebSocket, SSE or + * long-poll paths all feed it through `addSample`. + */ + +/** Samples kept in the sliding window. */ +export const CLOCK_SAMPLE_WINDOW = 8; + +/** Samples slower than this are discarded outright (offset bound too loose). */ +export const MAX_SAMPLE_RTT_MS = 10_000; + +/** + * If the wall clock and the monotonic clock drift apart by more than this + * between two observations, the wall clock jumped (manual change, NTP step, + * laptop sleep/resume). Every sample taken before the jump is invalidated. + */ +export const CLOCK_JUMP_THRESHOLD_MS = 2_000; + +/** + * Offsets larger than this are reported as a diagnostic (`skewed: true`). + * Normalization still corrects for them; this only tells developers and + * support that the user's device clock is noticeably wrong. + */ +export const SKEW_WARN_THRESHOLD_MS = 30_000; + +/** + * Subscribers are only notified when the offset moves by at least this much, + * so UI bound to the offset does not re-render on sub-display-precision jitter. + */ +export const OFFSET_NOTIFY_THRESHOLD_MS = 250; + +export interface ClockSyncSnapshot { + /** Estimated server-minus-client offset in ms. Positive => server clock ahead. */ + readonly offsetMs: number; + /** Half the best sample's RTT: the true offset is within ±this. `Infinity` when there are no samples. */ + readonly uncertaintyMs: number; + /** Valid samples currently in the window. 0 => no estimate yet (offset treated as 0). */ + readonly sampleCount: number; + /** True when |offset| exceeds `SKEW_WARN_THRESHOLD_MS`. Diagnostic only. */ + readonly skewed: boolean; +} + +interface Sample { + offset: number; + rtt: number; +} + +export interface ClockSources { + /** Wall clock, ms since epoch. */ + wall: () => number; + /** Monotonic clock, ms. Unaffected by wall-clock changes. */ + mono: () => number; +} + +const defaultSources: ClockSources = { + wall: () => Date.now(), + mono: () => (typeof performance !== "undefined" ? performance.now() : Date.now()), +}; + +const EMPTY_SNAPSHOT: ClockSyncSnapshot = Object.freeze({ + offsetMs: 0, + uncertaintyMs: Number.POSITIVE_INFINITY, + sampleCount: 0, + skewed: false, +}); + +export class ClockOffsetEstimator { + private samples: Sample[] = []; + private snapshot: ClockSyncSnapshot = EMPTY_SNAPSHOT; + private listeners = new Set<() => void>(); + /** Wall/mono pair from the last observation, used to detect clock jumps. */ + private anchor: { wall: number; mono: number } | null = null; + + constructor(private readonly sources: ClockSources = defaultSources) {} + + /** + * Records one ping/pong round trip. + * Returns false when the sample was rejected (negative or excessive RTT). + */ + addSample(t0: number, serverTs: number, t3: number): boolean { + this.checkForClockJump(); + const rtt = t3 - t0; + if (!Number.isFinite(rtt) || !Number.isFinite(serverTs) || rtt < 0 || rtt > MAX_SAMPLE_RTT_MS) + return false; + + this.samples.push({ offset: serverTs - (t0 + t3) / 2, rtt }); + if (this.samples.length > CLOCK_SAMPLE_WINDOW) this.samples.shift(); + this.recompute(); + return true; + } + + /** + * Detects a wall-clock jump since the last observation and, if one + * happened, drops all samples (they were measured against the old clock). + * Called on every sample; transports may also call it on wake/visibility + * events. + */ + checkForClockJump(): boolean { + const wall = this.sources.wall(); + const mono = this.sources.mono(); + const prev = this.anchor; + this.anchor = { wall, mono }; + if (!prev) return false; + const drift = wall - prev.wall - (mono - prev.mono); + if (Math.abs(drift) <= CLOCK_JUMP_THRESHOLD_MS) return false; + this.reset(); + return true; + } + + /** + * Latest estimate. Subscribers are only notified on meaningful changes, so + * UI should derive a display value (see `useServerRelativeTime`) rather + * than render from this object directly. + */ + getSnapshot = (): ClockSyncSnapshot => this.snapshot; + + /** Subscribe to meaningful estimate changes (see `OFFSET_NOTIFY_THRESHOLD_MS`). */ + subscribe = (listener: () => void): (() => void) => { + this.listeners.add(listener); + return () => { + this.listeners.delete(listener); + }; + }; + + /** Converts a server timestamp (ms since epoch) into client wall-clock time. */ + normalize = (serverTs: number): number => serverTs - this.snapshot.offsetMs; + + /** Discards all samples (e.g. after a detected clock jump). */ + reset(): void { + this.samples = []; + this.publish(EMPTY_SNAPSHOT); + } + + private recompute(): void { + const minRtt = Math.min(...this.samples.map((s) => s.rtt)); + // Only trust samples close to the best observed latency: those have the + // tightest offset bound. Slow samples are the classic outliers. + const trusted = this.samples.filter((s) => s.rtt <= minRtt * 1.5 + 20); + const offsets = trusted.map((s) => s.offset).sort((a, b) => a - b); + const mid = offsets.length >> 1; + const median = offsets.length % 2 ? offsets[mid]! : (offsets[mid - 1]! + offsets[mid]!) / 2; + const offsetMs = Math.round(median); + + const next: ClockSyncSnapshot = Object.freeze({ + offsetMs, + uncertaintyMs: Math.round(minRtt / 2), + sampleCount: this.samples.length, + skewed: Math.abs(offsetMs) > SKEW_WARN_THRESHOLD_MS, + }); + + const prev = this.snapshot; + const meaningful = + prev.sampleCount === 0 || + Math.abs(next.offsetMs - prev.offsetMs) >= OFFSET_NOTIFY_THRESHOLD_MS || + next.skewed !== prev.skewed; + this.publish(next, meaningful); + } + + /** Always stores the latest estimate; only wakes subscribers when it matters. */ + private publish(next: ClockSyncSnapshot, notify = true): void { + if (next === this.snapshot) return; + this.snapshot = next; + if (notify) this.listeners.forEach((l) => l()); + } +} + +/** + * The app-wide estimator. Realtime transports write to it; everything else + * should use the read-only helpers below rather than the instance itself. + */ +export const defaultClock = new ClockOffsetEstimator(); + +/** Read-only view of the current clock-sync estimate. */ +export function getClockSync(): ClockSyncSnapshot { + return defaultClock.getSnapshot(); +} + +/** + * Converts a server timestamp (ms since epoch) into client wall-clock time. + * Use this before rendering a server timestamp as a relative time or + * comparing it with `Date.now()`. + */ +export function normalizeServerTs( + serverTs: number, + clock: ClockOffsetEstimator = defaultClock +): number { + return clock.normalize(serverTs); +} + +/** Age of a server-timestamped event in ms, measured on the client clock. Never negative. */ +export function serverEventAgeMs( + serverTs: number, + now: number = Date.now(), + clock: ClockOffsetEstimator = defaultClock +): number { + return Math.max(0, now - clock.normalize(serverTs)); +} + +/** Freshness check for server-timestamped data, corrected for clock skew. */ +export function isServerTsStale( + serverTs: number, + maxAgeMs: number, + now: number = Date.now(), + clock: ClockOffsetEstimator = defaultClock +): boolean { + return serverEventAgeMs(serverTs, now, clock) > maxAgeMs; +} diff --git a/src/lib/realtime/codec.test.ts b/src/lib/realtime/codec.test.ts new file mode 100644 index 0000000..7e4b0bb --- /dev/null +++ b/src/lib/realtime/codec.test.ts @@ -0,0 +1,63 @@ +import { describe, expect, it } from "vitest"; +import { decodeFrame, MAX_MESSAGE_BYTES, utf8ByteLength } from "./codec"; + +/** A JSON envelope whose serialized form is exactly `bytes` long (ASCII). */ +function frameOfSize(bytes: number): string { + const shell = JSON.stringify({ topic: "t", payload: "" }); + return JSON.stringify({ topic: "t", payload: "x".repeat(bytes - shell.length) }); +} + +describe("decodeFrame", () => { + it("accepts a frame exactly at the limit", () => { + const frame = frameOfSize(MAX_MESSAGE_BYTES); + expect(frame.length).toBe(MAX_MESSAGE_BYTES); + const result = decodeFrame(frame); + expect(result.ok).toBe(true); + if (result.ok) expect(result.bytes).toBe(MAX_MESSAGE_BYTES); + }); + + it("rejects a frame one byte over the limit, before parsing", () => { + const result = decodeFrame(frameOfSize(MAX_MESSAGE_BYTES + 1)); + expect(result.ok).toBe(false); + if (!result.ok) { + expect(result.error.code).toBe("oversize"); + expect(result.error.bytes).toBe(MAX_MESSAGE_BYTES + 1); + } + }); + + it("measures UTF-8 bytes, not string length", () => { + // 100 three-byte characters: 100 code units, 300 bytes. + const frame = JSON.stringify({ topic: "t", payload: "€".repeat(100) }); + expect(decodeFrame(frame, frame.length + 10).ok).toBe(false); + expect(decodeFrame(frame, utf8ByteLength(frame)).ok).toBe(true); + }); + + it("counts surrogate pairs as 4 bytes", () => { + expect(utf8ByteLength("😀")).toBe(4); + expect(utf8ByteLength("aé€😀")).toBe(1 + 2 + 3 + 4); + }); + + it("applies the same limit to binary frames", () => { + const bytes = new TextEncoder().encode(frameOfSize(1_000)); + expect(decodeFrame(bytes.buffer, 999).ok).toBe(false); + const ok = decodeFrame(bytes.buffer, 1_000); + expect(ok.ok && ok.envelope.topic).toBe("t"); + }); + + it("rejects malformed JSON and envelopes without a topic", () => { + expect(decodeFrame("{nope")).toMatchObject({ ok: false, error: { code: "malformed" } }); + expect(decodeFrame('{"payload":1}')).toMatchObject({ ok: false, error: { code: "malformed" } }); + expect(decodeFrame("null")).toMatchObject({ ok: false, error: { code: "malformed" } }); + }); + + it("rejects unsupported frame types", () => { + expect(decodeFrame(42)).toMatchObject({ ok: false, error: { code: "unsupported_frame" } }); + }); + + it("keeps a numeric ts and drops a bad one", () => { + const a = decodeFrame('{"topic":"t","payload":1,"ts":123}'); + expect(a.ok && a.envelope).toEqual({ topic: "t", payload: 1, ts: 123 }); + const b = decodeFrame('{"topic":"t","payload":1,"ts":"soon"}'); + expect(b.ok && b.envelope).toEqual({ topic: "t", payload: 1 }); + }); +}); diff --git a/src/lib/realtime/codec.ts b/src/lib/realtime/codec.ts new file mode 100644 index 0000000..d74c13a --- /dev/null +++ b/src/lib/realtime/codec.ts @@ -0,0 +1,130 @@ +/** + * Shared inbound-frame decoding for every realtime transport (issue #249). + * + * WebSocket, SSE and long-poll paths all hand raw frames to `decodeFrame`, so + * they share one size ceiling and one definition of a malformed message. + */ + +/** + * Maximum accepted inbound message size, in UTF-8 bytes, checked before + * `JSON.parse` runs. + * + * Sizing: steady-state pushes (oracle readings, pool stats, policy events) + * are well under 4 KiB. The largest legitimate frame is an initial topic + * snapshot, which the realtime contract caps at 200 policy/claim rows + * (~150 KiB as JSON). 256 KiB leaves headroom over that snapshot while still + * keeping a hostile frame from stalling the main thread in `JSON.parse`. + * Raise it together with the snapshot cap in docs/realtime.md, never alone. + */ +export const MAX_MESSAGE_BYTES = 256 * 1024; + +/** Envelope every realtime message uses. */ +export interface RealtimeEnvelope { + topic: string; + payload: unknown; + /** Server timestamp (ms since epoch). Normalize with `normalizeServerTs` before use. */ + ts?: number; +} + +export type RealtimeProtocolErrorCode = "oversize" | "malformed" | "unsupported_frame"; + +/** A frame the client refused to process. See docs/realtime.md "Error path". */ +export class RealtimeProtocolError extends Error { + constructor( + public readonly code: RealtimeProtocolErrorCode, + message: string, + /** Size of the rejected frame in bytes, when known. */ + public readonly bytes?: number + ) { + super(message); + this.name = "RealtimeProtocolError"; + } +} + +export type DecodeResult = + | { ok: true; envelope: RealtimeEnvelope; bytes: number } + | { ok: false; error: RealtimeProtocolError }; + +/** UTF-8 byte length of a string without allocating an encoded copy. */ +export function utf8ByteLength(text: string): number { + let bytes = 0; + for (let i = 0; i < text.length; i++) { + const code = text.charCodeAt(i); + if (code < 0x80) bytes += 1; + else if (code < 0x800) bytes += 2; + else if (code >= 0xd800 && code <= 0xdbff && i + 1 < text.length) { + // Surrogate pair: one 4-byte code point. + bytes += 4; + i++; + } else bytes += 3; + } + return bytes; +} + +function oversize(bytes: number, maxBytes: number): DecodeResult { + return { + ok: false, + error: new RealtimeProtocolError( + "oversize", + `Inbound frame of ${bytes} bytes exceeds the ${maxBytes}-byte limit`, + bytes + ), + }; +} + +/** + * Validates and parses one inbound frame. + * + * Accepts text frames and `ArrayBuffer` binary frames (UTF-8 JSON). Frames + * over `maxBytes` are rejected *before* parsing. Never throws. + */ +export function decodeFrame(data: unknown, maxBytes: number = MAX_MESSAGE_BYTES): DecodeResult { + let text: string; + let bytes: number; + + if (typeof data === "string") { + // Every UTF-16 code unit is at least one UTF-8 byte, so a string longer + // than the limit is oversize without scanning it. + if (data.length > maxBytes) return oversize(data.length, maxBytes); + bytes = utf8ByteLength(data); + if (bytes > maxBytes) return oversize(bytes, maxBytes); + text = data; + } else if (data instanceof ArrayBuffer) { + bytes = data.byteLength; + if (bytes > maxBytes) return oversize(bytes, maxBytes); + text = new TextDecoder().decode(data); + } else { + return { + ok: false, + error: new RealtimeProtocolError("unsupported_frame", "Unsupported frame type"), + }; + } + + let parsed: unknown; + try { + parsed = JSON.parse(text); + } catch { + return { + ok: false, + error: new RealtimeProtocolError("malformed", "Frame is not valid JSON", bytes), + }; + } + + if ( + typeof parsed !== "object" || + parsed === null || + typeof (parsed as { topic?: unknown }).topic !== "string" + ) { + return { + ok: false, + error: new RealtimeProtocolError("malformed", "Frame is missing a string `topic`", bytes), + }; + } + + const { topic, payload, ts } = parsed as { topic: string; payload?: unknown; ts?: unknown }; + return { + ok: true, + bytes, + envelope: { topic, payload, ...(typeof ts === "number" && Number.isFinite(ts) ? { ts } : {}) }, + }; +} diff --git a/src/lib/realtime/dev/fakeSocket.ts b/src/lib/realtime/dev/fakeSocket.ts new file mode 100644 index 0000000..bc5f32f --- /dev/null +++ b/src/lib/realtime/dev/fakeSocket.ts @@ -0,0 +1,93 @@ +/** + * In-process fake WebSocket for tests and local tooling. + * + * DEV/TEST ONLY. Nothing under src/lib/realtime/dev may be imported by + * production code: ESLint blocks the import and + * scripts/check-realtime-bundle.mjs fails the build if the marker below shows + * up in `.next/static`. + */ + +export const REALTIME_DEV_ONLY_MARKER = "__REFRACT_REALTIME_DEV_ONLY__"; + +type Listener = (event: { data?: unknown }) => void; + +export class FakeWebSocket { + static readonly CONNECTING = 0; + static readonly OPEN = 1; + static readonly CLOSING = 2; + static readonly CLOSED = 3; + + /** Every socket constructed, newest last. */ + static instances: FakeWebSocket[] = []; + /** Extensions header the fake "server" will accept on the next `open()`. */ + static nextExtensions = ""; + + readonly marker = REALTIME_DEV_ONLY_MARKER; + readyState = FakeWebSocket.CONNECTING; + extensions = ""; + binaryType: "blob" | "arraybuffer" = "blob"; + /** Frames the client sent, parsed. */ + sent: unknown[] = []; + private listeners = new Map>(); + + constructor( + readonly url: string, + readonly protocols?: string | string[], + readonly options?: unknown + ) { + FakeWebSocket.instances.push(this); + } + + static latest(): FakeWebSocket { + const s = FakeWebSocket.instances[FakeWebSocket.instances.length - 1]; + if (!s) throw new Error("No FakeWebSocket has been constructed"); + return s; + } + + static reset(): void { + FakeWebSocket.instances = []; + FakeWebSocket.nextExtensions = ""; + } + + addEventListener(type: string, listener: Listener): void { + if (!this.listeners.has(type)) this.listeners.set(type, new Set()); + this.listeners.get(type)!.add(listener); + } + + removeEventListener(type: string, listener: Listener): void { + this.listeners.get(type)?.delete(listener); + } + + send(data: string): void { + this.sent.push(JSON.parse(data)); + } + + close(): void { + if (this.readyState === FakeWebSocket.CLOSED) return; + this.readyState = FakeWebSocket.CLOSED; + this.emit("close", {}); + } + + // --- "server" side -------------------------------------------------------- + + /** Completes the handshake, optionally accepting extensions. */ + open(extensions = FakeWebSocket.nextExtensions): void { + this.extensions = extensions; + this.readyState = FakeWebSocket.OPEN; + this.emit("open", {}); + } + + /** Delivers a raw frame (string or ArrayBuffer). */ + receiveRaw(data: unknown): void { + this.emit("message", { data }); + } + + /** Delivers a `{ topic, payload, ts? }` envelope as a JSON text frame. */ + push(topic: string, payload: unknown, ts?: number): void { + this.receiveRaw(JSON.stringify(ts === undefined ? { topic, payload } : { topic, payload, ts })); + } + + private emit(type: string, event: { data?: unknown }): void { + this.listeners.get(type)?.forEach((l) => l(event)); + } +} diff --git a/src/lib/realtime/entry.ts b/src/lib/realtime/entry.ts new file mode 100644 index 0000000..9c07ae6 --- /dev/null +++ b/src/lib/realtime/entry.ts @@ -0,0 +1,22 @@ +/** + * The realtime layer's single lazily-loaded entry (issue #248). + * + * Everything that ships the WebSocket client or payload validators to the + * browser is reached through this module, and this module is only reached + * through `loadRealtime()` in ./load (a dynamic import). That keeps the + * realtime code in its own `realtime` chunk instead of the initial bundle. + * + * Never import from ./dev here: dev/test tooling must not reach production. + */ +import { RealtimeClient } from "./client"; + +export { RealtimeClient } from "./client"; +export type { RealtimeDiagnostics, TopicHandler } from "./client"; +export { parsePolicyEvent } from "./policyEvents"; + +let client: RealtimeClient | null = null; + +/** The app-wide realtime client, created on first use. */ +export function getRealtimeClient(): RealtimeClient { + return (client ??= new RealtimeClient()); +} diff --git a/src/lib/realtime/load.ts b/src/lib/realtime/load.ts new file mode 100644 index 0000000..42c99d5 --- /dev/null +++ b/src/lib/realtime/load.ts @@ -0,0 +1,20 @@ +/** + * Dynamic-import boundary for the realtime layer (issue #248). + * + * UI code (hooks, providers) calls `loadRealtime()` instead of importing + * ./client or ./entry statically. The named chunk makes the output + * predictable (`.next/static/chunks/realtime..js`) so `.size-limit.json` + * can budget it and `scripts/check-realtime-bundle.mjs` can find it. + */ +export type RealtimeModule = typeof import("./entry"); + +let pending: Promise | null = null; + +export function loadRealtime(): Promise { + pending ??= import(/* webpackChunkName: "realtime" */ "./entry").catch((err) => { + // Allow a later retry (e.g. after a flaky chunk download). + pending = null; + throw err; + }); + return pending; +} diff --git a/src/lib/realtime/pendingPolicies.test.ts b/src/lib/realtime/pendingPolicies.test.ts new file mode 100644 index 0000000..4246209 --- /dev/null +++ b/src/lib/realtime/pendingPolicies.test.ts @@ -0,0 +1,148 @@ +import { describe, expect, it } from "vitest"; +import { + addOptimistic, + applyPolicyEvent, + findInFlightDuplicate, + settleViaRest, + type PendingPolicy, +} from "./pendingPolicies"; +import type { PolicyEvent } from "./policyEvents"; + +const HOLDER = "GHOLDER"; +const NOW = 1_700_000_000_000; + +const optimistic = (list: PendingPolicy[] = [], txHash = "tx1") => + addOptimistic( + list, + { + txHash, + holder: HOLDER, + policyId: "p1", + coverageType: 1, + coverageTypeName: "Depeg", + coverageAmount: "50000000000", + }, + NOW + ); + +const ev = (e: Record) => + ({ holder: HOLDER, ts: NOW + 1_000, ...e }) as PolicyEvent; + +describe("pending policy transitions", () => { + it("optimistic → accepted → active (confirmed)", () => { + let list = optimistic(); + expect(list[0]).toMatchObject({ key: "tx1", status: "submitted" }); + + list = applyPolicyEvent(list, ev({ type: "accepted", txHash: "tx1" }), NOW + 1); + expect(list[0]!.status).toBe("accepted"); + + list = applyPolicyEvent( + list, + ev({ type: "issued", txHash: "tx1", policyId: "p1", ts: NOW + 2_000 }), + NOW + 2 + ); + expect(list).toHaveLength(1); + expect(list[0]).toMatchObject({ + status: "active", + confirmedBy: "realtime", + lastEventTs: NOW + 2_000, + }); + }); + + it("optimistic → failed rolls the entry back", () => { + let list = optimistic(); + list = applyPolicyEvent( + list, + ev({ type: "failed", txHash: "tx1", reason: "tx_bad_seq" }), + NOW + 1 + ); + expect(list[0]).toMatchObject({ status: "failed", reason: "tx_bad_seq" }); + }); + + it("never moves backwards (duplicates and out-of-order delivery)", () => { + let list = optimistic(); + list = applyPolicyEvent(list, ev({ type: "issued", txHash: "tx1", policyId: "p1" }), NOW + 1); + const settled = list; + list = applyPolicyEvent(list, ev({ type: "accepted", txHash: "tx1" }), NOW + 2); + list = applyPolicyEvent(list, ev({ type: "issued", txHash: "tx1", policyId: "p1" }), NOW + 3); + list = applyPolicyEvent(list, ev({ type: "failed", txHash: "tx1", reason: "late" }), NOW + 4); + expect(list).toBe(settled); + }); + + it("cancellation after issuance", () => { + let list = optimistic(); + list = applyPolicyEvent(list, ev({ type: "issued", txHash: "tx1", policyId: "p1" }), NOW + 1); + list = applyPolicyEvent( + list, + ev({ type: "cancelled", policyId: "p1", reason: "pool_insufficient" }), + NOW + 2 + ); + expect(list[0]).toMatchObject({ status: "cancelled", reason: "pool_insufficient" }); + }); + + it("records premium charges by policy id", () => { + let list = optimistic(); + list = applyPolicyEvent( + list, + ev({ type: "premium_charged", policyId: "p1", amount: "1230000" }), + NOW + 1 + ); + expect(list[0]!.premiumCharged).toBe("1230000"); + }); + + it("does not duplicate a row when the same purchase is added twice (other tab, retry)", () => { + let list = optimistic(); + list = optimistic(list); + expect(list).toHaveLength(1); + }); + + it("merges when the issuance event beats the REST response", () => { + let list = applyPolicyEvent([], ev({ type: "issued", txHash: "tx1", policyId: "p1" }), NOW); + expect(list[0]).toMatchObject({ status: "active", external: true }); + list = optimistic(list); + expect(list).toHaveLength(1); + expect(list[0]).toMatchObject({ status: "active", external: false, coverageTypeName: "Depeg" }); + }); + + it("shows policies issued outside this session", () => { + const list = applyPolicyEvent( + optimistic(), + ev({ type: "issued", txHash: "tx-elsewhere", policyId: "p9" }), + NOW + ); + expect(list).toHaveLength(2); + expect(list[0]).toMatchObject({ key: "tx-elsewhere", status: "active", external: true }); + }); + + it("ignores failure/cancel events for unknown policies", () => { + const list = optimistic(); + expect(applyPolicyEvent(list, ev({ type: "failed", txHash: "nope", reason: "x" }), NOW)).toBe( + list + ); + expect( + applyPolicyEvent(list, ev({ type: "cancelled", policyId: "nope", reason: "expired" }), NOW) + ).toBe(list); + }); + + it("falls back to the REST result after the bounded wait", () => { + let list = optimistic(); + list = settleViaRest(list, "tx1", NOW + 90_000); + expect(list[0]).toMatchObject({ status: "active", confirmedBy: "rest" }); + // A late issuance event upgrades the confirmation source. + list = applyPolicyEvent( + list, + ev({ type: "issued", txHash: "tx1", policyId: "p1" }), + NOW + 91_000 + ); + expect(list[0]).toMatchObject({ status: "active", confirmedBy: "realtime" }); + }); + + it("detects an identical in-flight purchase", () => { + const list = optimistic(); + const params = { holder: HOLDER, coverageType: 1, coverageAmount: "50000000000" }; + expect(findInFlightDuplicate(list, params)?.key).toBe("tx1"); + expect(findInFlightDuplicate(list, { ...params, coverageAmount: "1" })).toBeUndefined(); + const settled = settleViaRest(list, "tx1", NOW); + expect(findInFlightDuplicate(settled, params)).toBeUndefined(); + }); +}); diff --git a/src/lib/realtime/pendingPolicies.ts b/src/lib/realtime/pendingPolicies.ts new file mode 100644 index 0000000..1ebdaef --- /dev/null +++ b/src/lib/realtime/pendingPolicies.ts @@ -0,0 +1,267 @@ +import { create } from "zustand"; +import { createJSONStorage, persist } from "zustand/middleware"; +import type { PolicyEvent } from "./policyEvents"; + +/** + * Pending policy purchases and how realtime events move them along + * (issue #246). + * + * submitted ──accepted──▶ accepted ──issued──▶ active + * │ │ │ + * └──────failed──────────┴──▶ failed └──cancelled──▶ cancelled + * + * The REST purchase call creates the optimistic entry (keyed by its tx hash); + * events only move it forward. If no event arrives within + * `PENDING_WAIT_MS`, the REST result stands and the entry settles as active + * with `confirmedBy: "rest"`. + * + * The list is persisted in localStorage so a pending purchase survives + * navigating away from /cover, and the `storage` event keeps other tabs in + * sync (see `usePolicyStream`). + */ + +export type PendingPolicyStatus = "submitted" | "accepted" | "active" | "failed" | "cancelled"; + +export interface PendingPolicy { + /** Stable identity: the purchase tx hash, or `policy:` for policies we only know by id. */ + key: string; + txHash?: string; + policyId?: string; + holder: string; + status: PendingPolicyStatus; + coverageType?: number; + coverageTypeName?: string; + /** Base-unit USDC string. */ + coverageAmount?: string; + /** Base-unit USDC string. */ + premium?: string; + /** Base-unit USDC string, from the latest `premium_charged` event. */ + premiumCharged?: string; + /** What moved this entry to `active`. */ + confirmedBy?: "realtime" | "rest"; + /** Failure or cancellation reason. */ + reason?: string; + /** True for policies first seen through an event (bought elsewhere). */ + external?: boolean; + /** Client ms when the entry was created. */ + createdAt: number; + /** Client ms of the last change. */ + updatedAt: number; + /** Server ms (`ts`) of the last applied event. Normalize before display. */ + lastEventTs?: number; +} + +/** How long to wait for a realtime event before trusting the REST result. */ +export const PENDING_WAIT_MS = 90_000; +/** Settled entries older than this are pruned. */ +export const SETTLED_RETENTION_MS = 24 * 60 * 60 * 1000; +export const MAX_PENDING_ENTRIES = 20; + +const RANK: Record = { + submitted: 0, + accepted: 1, + active: 2, + failed: 3, + cancelled: 3, +}; + +export function isTerminal(status: PendingPolicyStatus): boolean { + return status === "active" || status === "failed" || status === "cancelled"; +} + +export interface OptimisticPolicyInput { + txHash: string; + holder: string; + policyId?: string; + coverageType?: number; + coverageTypeName?: string; + coverageAmount?: string; + premium?: string; +} + +function findIndex(list: PendingPolicy[], txHash?: string, policyId?: string): number { + if (txHash) { + const i = list.findIndex((p) => p.txHash === txHash); + if (i !== -1) return i; + } + if (policyId) return list.findIndex((p) => p.policyId === policyId); + return -1; +} + +function prune(list: PendingPolicy[], now: number): PendingPolicy[] { + return list + .filter((p) => !isTerminal(p.status) || now - p.updatedAt < SETTLED_RETENTION_MS) + .slice(0, MAX_PENDING_ENTRIES); +} + +/** + * Adds the optimistic entry for a just-submitted purchase. Idempotent: a + * second call for the same tx hash (another tab, a retry, an event that + * arrived first) merges into the existing row instead of duplicating it. + */ +export function addOptimistic( + list: PendingPolicy[], + input: OptimisticPolicyInput, + now: number +): PendingPolicy[] { + const existing = list[findIndex(list, input.txHash, input.policyId)]; + if (existing) { + // Keep what events already established; fill gaps from the REST result. + const known = Object.fromEntries( + Object.entries(existing).filter(([, v]) => v !== undefined) + ) as PendingPolicy; + const merged: PendingPolicy = { + ...input, + ...known, + txHash: input.txHash, + policyId: existing.policyId ?? input.policyId, + external: false, + }; + return list.map((p) => (p === existing ? merged : p)); + } + const entry: PendingPolicy = { + ...input, + key: input.txHash, + status: "submitted", + createdAt: now, + updatedAt: now, + }; + return prune([entry, ...list], now); +} + +/** + * Applies one realtime event. Events never move an entry backwards, so + * duplicates and out-of-order delivery are harmless. + */ +export function applyPolicyEvent( + list: PendingPolicy[], + event: PolicyEvent, + now: number +): PendingPolicy[] { + const prev = list[findIndex(list, "txHash" in event ? event.txHash : undefined, event.policyId)]; + + if (!prev) { + // Policies bought outside this session (another device, a direct contract + // call) still appear. Events about unknown policies that don't create one + // have nothing to update. + if (event.type !== "issued" && event.type !== "accepted") return list; + const created: PendingPolicy = { + key: event.txHash, + txHash: event.txHash, + policyId: event.policyId, + holder: event.holder, + status: event.type === "issued" ? "active" : "accepted", + ...(event.type === "issued" + ? { + confirmedBy: "realtime" as const, + coverageTypeName: event.coverageTypeName, + coverageAmount: event.coverageAmount, + premium: event.premium, + } + : {}), + external: true, + createdAt: now, + updatedAt: now, + lastEventTs: event.ts, + }; + return prune([created, ...list], now); + } + + let next: PendingPolicy = prev; + const advance = (status: PendingPolicyStatus, patch: Partial = {}) => { + if (RANK[status] > RANK[prev.status]) next = { ...prev, ...patch, status }; + }; + + switch (event.type) { + case "accepted": + advance("accepted", { policyId: prev.policyId ?? event.policyId }); + break; + case "issued": + advance("active", { + policyId: event.policyId, + confirmedBy: "realtime", + coverageTypeName: event.coverageTypeName ?? prev.coverageTypeName, + coverageAmount: event.coverageAmount ?? prev.coverageAmount, + premium: event.premium ?? prev.premium, + }); + // A REST-settled entry is already active; record that the chain agreed. + if (prev.status === "active" && prev.confirmedBy === "rest") + next = { ...prev, confirmedBy: "realtime" }; + break; + case "failed": + // Roll back the optimistic entry. Only valid before issuance. + if (prev.status === "submitted" || prev.status === "accepted") + next = { ...prev, status: "failed", reason: event.reason }; + break; + case "cancelled": + if (prev.status !== "failed" && prev.status !== "cancelled") + next = { ...prev, status: "cancelled", reason: event.reason }; + break; + case "premium_charged": + next = { ...prev, premiumCharged: event.amount }; + break; + } + + if (next === prev) return list; + next = { ...next, updatedAt: now, lastEventTs: Math.max(prev.lastEventTs ?? 0, event.ts) }; + return list.map((p) => (p === prev ? next : p)); +} + +/** + * Bounded-wait fallback: no event arrived in time, so trust the REST result + * (the transaction was submitted successfully) and mark the entry active. + */ +export function settleViaRest(list: PendingPolicy[], key: string, now: number): PendingPolicy[] { + return list.map((p) => + p.key === key && (p.status === "submitted" || p.status === "accepted") + ? { ...p, status: "active", confirmedBy: "rest", updatedAt: now } + : p + ); +} + +/** + * An in-flight purchase with the same parameters, if any. Used to warn + * before a duplicate submission (including from another tab). + */ +export function findInFlightDuplicate( + list: PendingPolicy[], + params: { holder: string; coverageType: number; coverageAmount: string } +): PendingPolicy | undefined { + return list.find( + (p) => + !isTerminal(p.status) && + p.holder === params.holder && + p.coverageType === params.coverageType && + p.coverageAmount === params.coverageAmount + ); +} + +export const PENDING_POLICIES_STORAGE_KEY = "refract:pending-policies"; + +interface PendingPolicyStore { + entries: PendingPolicy[]; + addOptimistic: (input: OptimisticPolicyInput) => void; + applyEvent: (event: PolicyEvent) => void; + settleViaRest: (key: string) => void; + dismiss: (key: string) => void; +} + +export const usePendingPolicyStore = create()( + persist( + (set) => ({ + entries: [], + addOptimistic: (input) => + set((s) => ({ entries: addOptimistic(s.entries, input, Date.now()) })), + applyEvent: (event) => + set((s) => ({ entries: applyPolicyEvent(s.entries, event, Date.now()) })), + settleViaRest: (key) => set((s) => ({ entries: settleViaRest(s.entries, key, Date.now()) })), + dismiss: (key) => set((s) => ({ entries: s.entries.filter((p) => p.key !== key) })), + }), + { + name: PENDING_POLICIES_STORAGE_KEY, + version: 1, + storage: createJSONStorage(() => localStorage), + partialize: (s) => ({ entries: s.entries }), + } + ) +); diff --git a/src/lib/realtime/policyEvents.test.ts b/src/lib/realtime/policyEvents.test.ts new file mode 100644 index 0000000..e1288d1 --- /dev/null +++ b/src/lib/realtime/policyEvents.test.ts @@ -0,0 +1,40 @@ +import { describe, expect, it } from "vitest"; +import { parsePolicyEvent } from "./policyEvents"; +import { policyTopic } from "./topics"; + +const HOLDER = "GABC"; + +describe("parsePolicyEvent", () => { + it("parses each event type", () => { + const base = { holder: HOLDER, ts: 1_700_000_000_000 }; + expect(parsePolicyEvent({ ...base, type: "accepted", txHash: "tx1" })?.type).toBe("accepted"); + expect(parsePolicyEvent({ ...base, type: "issued", txHash: "tx1", policyId: "p1" })?.type).toBe( + "issued" + ); + expect( + parsePolicyEvent({ ...base, type: "premium_charged", policyId: "p1", amount: "10" })?.type + ).toBe("premium_charged"); + expect( + parsePolicyEvent({ ...base, type: "cancelled", policyId: "p1", reason: "expired" })?.type + ).toBe("cancelled"); + expect( + parsePolicyEvent({ ...base, type: "failed", txHash: "tx1", reason: "tx_failed" })?.type + ).toBe("failed"); + }); + + it("returns null for malformed payloads", () => { + expect(parsePolicyEvent(null)).toBeNull(); + expect(parsePolicyEvent({ type: "issued", holder: HOLDER, ts: 1 })).toBeNull(); // missing ids + expect(parsePolicyEvent({ type: "teleported", holder: HOLDER, ts: 1, txHash: "x" })).toBeNull(); + expect( + parsePolicyEvent({ type: "cancelled", holder: HOLDER, ts: 1, policyId: "p", reason: "bored" }) + ).toBeNull(); + expect(parsePolicyEvent({ type: "accepted", holder: HOLDER, ts: -5, txHash: "x" })).toBeNull(); + }); +}); + +describe("policyTopic", () => { + it("keys the topic by wallet address", () => { + expect(policyTopic("GXYZ")).toBe("policy.GXYZ"); + }); +}); diff --git a/src/lib/realtime/policyEvents.ts b/src/lib/realtime/policyEvents.ts new file mode 100644 index 0000000..45f41a0 --- /dev/null +++ b/src/lib/realtime/policyEvents.ts @@ -0,0 +1,83 @@ +import { z } from "zod"; + +/** + * Payloads published on `policy.
` (issue #246). + * + * Every event names the policy (`policyId`) and/or the purchase transaction + * (`txHash`) it belongs to. The REST purchase call is the source of truth for + * the transaction hash; these events are state transitions for it. + * + * Amounts are base-unit USDC strings (1e7 per USDC), like the REST API. + * `ts` is the server clock (ms since epoch); normalize it before display. + */ + +const base = { + holder: z.string().min(1), + ts: z.number().int().positive(), +}; + +/** The backend has seen the purchase transaction and is waiting for it to settle. */ +const accepted = z.object({ + ...base, + type: z.literal("accepted"), + txHash: z.string().min(1), + policyId: z.string().min(1).optional(), +}); + +/** The policy exists on-chain and is active. */ +const issued = z.object({ + ...base, + type: z.literal("issued"), + policyId: z.string().min(1), + txHash: z.string().min(1), + coverageTypeName: z.string().optional(), + coverageAmount: z.string().optional(), + premium: z.string().optional(), + /** Unix seconds. */ + expiresAt: z.number().int().positive().optional(), +}); + +/** A premium charge for the policy was recorded on-chain. */ +const premiumCharged = z.object({ + ...base, + type: z.literal("premium_charged"), + policyId: z.string().min(1), + amount: z.string(), + txHash: z.string().min(1).optional(), +}); + +/** The policy was cancelled after issuance. */ +const cancelled = z.object({ + ...base, + type: z.literal("cancelled"), + policyId: z.string().min(1), + reason: z.enum(["holder_requested", "pool_insufficient", "expired", "oracle_triggered"]), + refundAmount: z.string().optional(), + txHash: z.string().min(1).optional(), +}); + +/** The purchase failed before the policy was issued (reverted, expired, rejected). */ +const failed = z.object({ + ...base, + type: z.literal("failed"), + txHash: z.string().min(1), + policyId: z.string().min(1).optional(), + reason: z.string(), +}); + +export const policyEventSchema = z.discriminatedUnion("type", [ + accepted, + issued, + premiumCharged, + cancelled, + failed, +]); + +export type PolicyEvent = z.infer; +export type PolicyEventType = PolicyEvent["type"]; + +/** Validates a `policy.
` payload. Returns null for anything malformed. */ +export function parsePolicyEvent(payload: unknown): PolicyEvent | null { + const result = policyEventSchema.safeParse(payload); + return result.success ? result.data : null; +} diff --git a/src/lib/realtime/topics.ts b/src/lib/realtime/topics.ts new file mode 100644 index 0000000..5338752 --- /dev/null +++ b/src/lib/realtime/topics.ts @@ -0,0 +1,29 @@ +/** + * Realtime topic names. The full contract (payload shapes, who publishes + * what) is documented in docs/realtime.md. + */ + +/** Client/transport control topics. Never delivered to `subscribe` handlers. */ +export const INTERNAL_TOPICS = { + /** Client → server: `{ t0 }` clock-sync / keepalive ping. */ + PING: "refract.ping", + /** Server → client: `{ t0, serverTs }` reply to a ping. */ + PONG: "refract.pong", + /** Server → client: `{ compressedBytes, uncompressedBytes }` cumulative wire stats for this connection. */ + WIRE_STATS: "refract.wire_stats", +} as const; + +const INTERNAL_TOPIC_SET: ReadonlySet = new Set(Object.values(INTERNAL_TOPICS)); + +export function isInternalTopic(topic: string): boolean { + return INTERNAL_TOPIC_SET.has(topic); +} + +/** + * Policy lifecycle events (accepted / issued / premium_charged / cancelled / + * failed) for every policy held by `address`. Payloads: `policyEventSchema` + * in ./policyEvents. + */ +export function policyTopic(address: string): string { + return `policy.${address}`; +} diff --git a/vitest.config.mts b/vitest.config.mts index 3505082..52a04aa 100644 --- a/vitest.config.mts +++ b/vitest.config.mts @@ -3,5 +3,7 @@ import { defineConfig } from "vitest/config"; export default defineConfig({ resolve: { alias: { "@": fileURLToPath(new URL("./src", import.meta.url)) } }, + // tsconfig uses `jsx: preserve` for Next; tests need JSX compiled. + oxc: { jsx: { runtime: "automatic" } }, test: { include: ["src/**/*.test.ts"] }, }); diff --git a/vitest.config.ts b/vitest.config.ts deleted file mode 100644 index e69de29..0000000