feat: stream realtime account updates

docs/realtime-protocol.md

@@ -1,7 +1,9 @@
-# Realtime protocol (v1)
+# Realtime protocol
 
 WebSocket at `ws(s)://<api-host>/ws` (derived from `NEXT_PUBLIC_API_URL` by `deriveRealtimeUrl`). Types live in `src/lib/realtime/protocol.ts`; the transport is `src/lib/realtime/client.ts`; the dev/CI server is `scripts/mock-realtime-server.mjs`.
 
+The frontend connects to the API origin's `/ws` endpoint. WebSocket frames are JSON. A topic event has the shape `{ "topic": "claims.settlement", "payload": { ... } }`. Realtime delivery is best effort: REST remains authoritative for initial state and paginated history, and clients show REST data while the connection is stale.
+
 ## Server → client frame
 
 ```json
@@ -13,6 +15,14 @@ WebSocket at `ws(s)://<api-host>/ws` (derived from `NEXT_PUBLIC_API_URL` by `der
 - Payloads: `oracle {id, price, updatedAt}`, `claims {id, policyId, status, amount, updatedAt}`, `pool {id, tvl, utilization, updatedAt}`, `notifications {id, kind, message, createdAt}`.
 - `{"op":"pong"}` answers a client ping.
 
+## Connection and authentication
+
+Public subscriptions can be used without a wallet. Before sending any wallet-scoped subscription, the client requests a short-lived credential with `POST /api/v1/realtime/token` and `{ "wallet": "<public address>" }`. The response is `{ "token": "<opaque bearer token>", "expiresAt": 1792600000 }`, where `expiresAt` is Unix seconds (Unix milliseconds are also accepted). The client sends `{ "type": "authenticate", "wallet": "<public address>", "token": "<opaque bearer token>" }` as the first authenticated WebSocket frame. Tokens are never placed in URLs or logged.
+
+The server replies `{ "type": "authenticated" }`, then the client sends one `{ "type": "subscribe", "topic": "..." }` frame for each active topic. On token refresh, the client sends another `authenticate` frame with `refresh: true`; existing subscriptions remain active. On reconnect, the client obtains a fresh token, authenticates, and sends each active subscription once. Identity changes close the old socket and repeat the handshake for the new wallet.
+
+The server can send `{ "type": "token_expiring" }` to request an early refresh. Authentication errors use `{ "type": "auth_error", "status": 401|403, "code": "..." }` or an error frame with the same status. WebSocket close codes 4401 and 4403 also mean authentication failed. The client stops retries for those wallet-scoped topics and reports a degraded state; network failures continue to retry with backoff. Error reporting must omit tokens and full authentication frames.
+
 ## Client → server
 
 `{"op":"subscribe","topic":"oracle","cursor":12}` (cursor optional; the server replays frames with `seq > cursor`), `{"op":"unsubscribe","topic":"oracle"}`, `{"op":"ping"}`.
@@ -21,6 +31,20 @@ WebSocket at `ws(s)://<api-host>/ws` (derived from `NEXT_PUBLIC_API_URL` by `der
 
 `4401` auth rejected — the client does not reconnect. Any other close triggers capped, jittered exponential reconnect (1s base, 30s cap) that re-subscribes with each topic's last `seq`.
 
+## Topics
+
+`oracle_update` carries an `OracleReading` directly or under `payload.reading`.
+
+`claims.settlement` carries a normalized claim directly or under `payload.claim`. Claim rows contain `claimId` (preferred stable ID), `policyId`, `holder`, `coverageType`, `triggered`, `payout`, `reason`, `processedAt`, optional `settlementTxHash`, and optional `status`. `dedupKey` may be supplied at the envelope or row level. `processedAt` is Unix seconds. A terminal status is `settled`, `paid`, `rejected`, `failed`, or `cancelled`. Clients upsert by `dedupKey`, `claimId`, `id`, settlement transaction hash, then policy ID. A terminal claim cannot regress from an older replay. Pushed rows are held separately from REST cursors and merged with every REST page.
+
+`notifications` carries `{ "notification": { "id", "type", "title", "message", "timestamp", "severity", "href?" } }`, or the notification directly. IDs are stable across replay and REST refetch. Timestamps are Unix milliseconds or ISO-compatible numeric timestamps in milliseconds. The frontend's notification read mutation is `POST /api/v1/notifications/{id}/read`.
+
+`transactions.confirmation` carries `{ "event": { "correlationId", "outcome", "authoritative?", "reason?" } }`, or the event directly. `outcome` is `confirmed`, `amended`, or `rejected`. The client places the correlation ID in the `Idempotency-Key` header for the original buy/provide/withdraw request. A rejected event rolls back the optimistic transaction entry and displays a dismissible notice. Unresolved actions are checked with `POST /api/v1/transactions/verify` and `{ "correlationId" }`; the response status is `confirmed`, `rejected`, or `pending`.
+
 ## Merge rules
 
 `src/lib/realtime/merge.ts` folds events into a REST snapshot: idempotent, version (`ts`) ordered, stale events dropped, unknown ids inserted, removals tombstoned.
+
+## Feature flags
+
+Realtime remains disabled by default. `realtime-oracle`, `realtime-claims-settlement`, `realtime-notifications`, and `realtime-transactions` independently enable their corresponding topic consumers in `public/flags.json`.

public/flags.json

@@ -3,8 +3,10 @@
     "realtime-enabled": false,
     "realtime-oracle": false,
     "realtime-claims": false,
+    "realtime-claims-settlement": false,
     "realtime-pool": false,
     "realtime-notifications": false,
+    "realtime-transactions": false,
     "horizon-stream": false,
     "realtime-quotes": false,
     "realtime-heartbeat": true,

src/app/cover/page.tsx

@@ -30,6 +30,7 @@ import {
 } from "@/lib/cover/batchCheckout";
 import { PreSignConfirmModal } from "@/components/PreSignConfirmModal";
 import { useCoverQueryParams } from "@/hooks/useCoverQueryParams";
+import { useOptimisticReconciliation } from "@/hooks/useOptimisticReconciliation";
 import { parseCoverQueryParams } from "@/lib/coverQueryParams";
 import { PolicyCertificateModal } from "@/components/PolicyCertificate";
 import { validateBuyCoverage } from "@/lib/validation/coverage";
@@ -84,6 +85,7 @@ export default function CoverPage() {
   const updateTransaction = useTransactionStore((s) => s.updateTransaction);
   const { minCoverage: chainMinCoverage, maxCoverage: chainMaxCoverage } = useCoverageBounds();
   const networkFee = useEstimatedFee("buy");
+  const reconciliation = useOptimisticReconciliation();
 
   const [currentStep, setCurrentStep] = useState<number>(1);
   const [selectedType, setSelectedType] = useState<number>(0);
@@ -182,6 +184,7 @@ export default function CoverPage() {
 
   const expiryMs = Date.now() + durationDays * 86400000;
   const expiryDate = <RelativeDate target={expiryMs} />;
+  const expiryLabel = new Date(expiryMs).toLocaleDateString("en-US", { month: "short", day: "numeric", year: "numeric" });
 
   const effectiveMin = Math.max(100, chainMinCoverage ?? 0);
   const effectiveMax = ct ? Math.min(ct.maxCoverage, chainMaxCoverage ?? ct.maxCoverage) : 0;
@@ -212,9 +215,9 @@ export default function CoverPage() {
         id: crypto.randomUUID(),
         params: {
           coverageType: ct.id,
-          coverageAmount: toStroops(parseFloat(coverageAmount)),
+          coverageAmount: toStroops(parseFloat(coverageAmount)).toString(),
           durationDays,
-          ...(isFlightDelay ? { triggerParams: { flightNumber: flightNumber.trim() } } : {}),
+          ...(isFlightDelay ? { triggerParams: { coverageType: 4 as const, params: { flightNumber: flightNumber.trim(), departureDate: new Date().toISOString().slice(0, 10) } } } : {}),
         },
         typeName: ct.name,
         icon: ct.icon,
@@ -229,11 +232,19 @@ export default function CoverPage() {
   }
 
   async function purchaseCartItem(item: CartItem, holder: string, networkPassphrase: string): Promise<PurchaseResult> {
+    const correlationId = crypto.randomUUID();
+    let rejected = false;
+    const release = reconciliation.register({ correlationId, kind: "buy",
+      onConfirm: (authoritative) => updateCartItem(item.id, { status: "confirmed", ...(typeof authoritative?.policyId === "string" ? { policyId: authoritative.policyId } : {}), ...(typeof authoritative?.txHash === "string" ? { txHash: authoritative.txHash } : {}) }),
+      onRollback: (reason) => { rejected = true; updateCartItem(item.id, { status: "failed", error: reason }); },
+    });
     try {
-      const result = await buyPolicy({ ...item.params, holder });
+      const result = await buyPolicy({ ...item.params, holder }, correlationId);
       const txHash = await signAndSubmit(result.txXdr, holder, networkPassphrase);
+      if (rejected) throw new Error("The transaction was rejected by the server.");
       return { policyId: result.policy.id, txHash, demo: false };
     } catch (err) {
+      release?.();
       // Same clearly-labeled demo fallback as the single-purchase flow.
       if (err instanceof ApiUnreachableError) return { policyId: `demo-${crypto.randomUUID()}`, demo: true };
       throw err;
@@ -265,28 +276,45 @@ export default function CoverPage() {
       // A declined/failed connect resolves null and stops here.
       const connected = await wallet.connect();
       if (!connected) return;
-      holder = connected.address;
-      networkPassphrase = connected.networkPassphrase;
+      holder = connected;
+      networkPassphrase = wallet.networkPassphrase;
     }
     setPremiumChanged(false);
     setConfirmOpen(true);
   }
 
   async function submitBuy() {
     setConfirmOpen(false);
-    if (!ct || !wallet.address) return;
+    const holder = wallet.address;
+    if (!ct || !holder) return;
     setSubmittingQuote(true);
 
     setSubmission({ status: "submitting" });
     const txId = addTransaction({ type: "buy", amount: parseFloat(coverageAmount) });
+    let rejected = false;
+    let policyCorrection: Partial<BuyPolicyResponse["policy"]> | undefined;
+    const releaseReconciliation = reconciliation.register({
+      correlationId: txId, kind: "buy",
+      onConfirm: (authoritative) => {
+        if (authoritative?.policy && typeof authoritative.policy === "object") {
+          policyCorrection = authoritative.policy as Partial<BuyPolicyResponse["policy"]>;
+          setSubmission((current) => current.status === "success" ? { ...current, result: { ...current.result, policy: { ...current.result.policy, ...policyCorrection } } } : current);
+        }
+        updateTransaction(txId, { status: "confirmed", authoritative,
+          ...(typeof authoritative?.amount === "number" ? { amount: authoritative.amount } : {}),
+          ...(typeof authoritative?.txHash === "string" ? { txHash: authoritative.txHash } : {}),
+        });
+      },
+      onRollback: (reason) => { rejected = true; updateTransaction(txId, { status: "failed", error: reason }); setSubmission({ status: "error", message: `${reason} The purchase details were reverted.` }); },
+    });
     try {
       const result = await buyPolicy({
         holder,
         coverageType: ct.id,
-        coverageAmount: toStroops(parseFloat(coverageAmount)),
+        coverageAmount: toStroops(parseFloat(coverageAmount)).toString(),
         durationDays,
-        ...(isFlightDelay ? { triggerParams: { flightNumber: flightNumber.trim() } } : {}),
-      });
+        ...(isFlightDelay ? { triggerParams: { coverageType: 4 as const, params: { flightNumber: flightNumber.trim(), departureDate: new Date().toISOString().slice(0, 10) } } } : {}),
+      }, txId);
 
       setSubmission({ status: "signing" });
       updateTransaction(txId, { status: "signing" });
@@ -302,26 +330,31 @@ export default function CoverPage() {
           { label: "Coverage amount", value: formatUsd(parseFloat(coverageAmount)) },
           { label: "Duration", value: `${durationDays} days` },
         ],
-        expectedStroops: toStroops(parseFloat(coverageAmount)),
+        expectedStroops: toStroops(parseFloat(coverageAmount)).toString(),
       });
       if (!approved) {
+        releaseReconciliation();
         setSubmission({ status: "idle" });
         return;
       }
-      const txHash = await signAndSubmit(result.txXdr, wallet.address, wallet.networkPassphrase, { hardware: wallet.hardwareWallet });
+      const networkPassphrase = wallet.networkPassphrase;
+      if (!networkPassphrase) throw new Error("Wallet network isn't available — reconnect and try again");
+      const txHash = await signAndSubmit(result.txXdr, holder, networkPassphrase, { hardware: wallet.hardwareWallet });
+      if (rejected) throw new Error("The transaction was rejected by the server.");
       publishAccountBalanceDelta(wallet.address, "USDC", -fromStroops(result.policy.premium));
-      setSubmission({ status: "success", result, demo: false, txHash });
+      setSubmission({ status: "success", result: policyCorrection ? { ...result, policy: { ...result.policy, ...policyCorrection } } : result, demo: false, txHash });
       updateTransaction(txId, { status: "confirmed", txHash });
     } catch (err) {
       if (err instanceof ApiUnreachableError) {
+        releaseReconciliation();
         const demoResult: BuyPolicyResponse = {
           policy: {
             id: `demo-${crypto.randomUUID()}`,
             holder,
             coverageType: ct.id,
             coverageTypeName: ct.name,
-            coverageAmount: toStroops(parseFloat(coverageAmount)),
-            premium: toStroops(premium),
+            coverageAmount: toStroops(parseFloat(coverageAmount)).toString(),
+            premium: toStroops(premium).toString(),
             durationDays,
             expiresAt: Math.floor(Date.now() / 1000) + durationDays * 86400,
             isActive: true,
@@ -334,6 +367,7 @@ export default function CoverPage() {
         updateTransaction(txId, { status: "confirmed", demo: true });
         return;
       }
+      releaseReconciliation();
       const message = err instanceof Error ? err.message : "Something went wrong buying coverage";
       setSubmission({ status: "error", message });
       updateTransaction(txId, { status: "failed", error: message });
@@ -344,6 +378,11 @@ export default function CoverPage() {
     <div className="min-h-screen bg-pm-bg">
       <Navbar right={<WalletButton />} />
       {reviewDialog}
+      {reconciliation.notices.map((notice) => (
+        <div key={notice.id} role="status" aria-live="polite" className="mx-auto mt-3 flex max-w-5xl items-start justify-between gap-3 rounded-lg border border-pm-red/30 bg-pm-red/10 px-4 py-3 text-sm text-pm-text">
+          <span>{notice.message}</span><button type="button" onClick={() => reconciliation.dismiss(notice.id)} aria-label="Dismiss update">×</button>
+        </div>
+      ))}
 
       <main id="main-content">
         <Container className="py-9 sm:py-10">
@@ -429,19 +468,15 @@ export default function CoverPage() {
                     Step 1 of 4: Select Coverage Type
                   </div>
 
-                  {typesError && (
+                  <AsyncState loading={typesLoading} error={typesError} data={coverageTypes}
+                    loadingRender={<div className="flex flex-col gap-2.5" role="status" aria-label="Loading coverage types">
+                      {Array.from({ length: 5 }).map((_, i) => <Skeleton key={i} height={72} rounded="md" />)}
+                    </div>}
+                    errorRender={(err: string) => (
                     <Card className="border-pm-red/30 !bg-pm-red/[0.04]">
-                      <p className="text-sm text-pm-red">Couldn&apos;t load coverage types: {typesError}</p>
+                      <p className="text-sm text-pm-red">Couldn&apos;t load coverage types: {err}</p>
                     </Card>
-                  )}
-
-                  {typesLoading && !coverageTypes && (
-                    <div className="flex flex-col gap-2.5" role="status" aria-label="Loading coverage types">
-                      {Array.from({ length: 5 }).map((_, i) => (
-                        <Skeleton key={i} height={72} rounded="md" />
-                      ))}
-                    </div>
-                  )}
+                  )}>
                   {coverageTypes && (
                     <div
                       className="flex flex-col gap-2.5"
@@ -520,6 +555,8 @@ export default function CoverPage() {
                       })}
                     </div>
                   )}
+                </AsyncState>
+
                   <div className="mt-6 flex justify-end">
                     <Button
                       type="button"
@@ -700,15 +737,19 @@ export default function CoverPage() {
                         {durationDays} days · Expires {expiryDate}
                       </span>
                     </div>
-                    <input
-                      id="duration"
-                      type="range"
-                      min={1}
-                      max={365}
-                      value={durationDays}
-                      onChange={(event) => setDurationDays(Number(event.target.value))}
-                      className="w-full accent-violet-500"
-                    />
+                    <input id="duration" type="range" min={1} max={365} value={durationDays}
+                      onChange={(e) => setDurationDays(Number(e.target.value))}
+                      className="w-full accent-pm-violet" aria-label="Coverage duration in days" />
+                    <div className="mt-2 flex justify-between text-[10px] text-pm-text/35"><span>1 day</span><span>365 days</span></div>
+                  </div>
+                  <div className="mt-6 flex items-center justify-between border-t border-pm-border pt-4">
+                    <Button type="button" variant="outline" onClick={() => setCurrentStep(2)}>← Back</Button>
+                    <Button type="button" variant="primary" onClick={() => setCurrentStep(4)} disabled={amountInvalid}>Review & Confirm →</Button>
+                  </div>
+                </Card>
+              )}
+            </div>
+   />
                   </div>
                   <div className="mt-6 flex items-center justify-between border-t border-pm-border pt-4">
                     <Button type="button" variant="outline" onClick={() => setCurrentStep(2)}>← Back</Button>

src/app/dashboard/policies/[id]/page.tsx

@@ -33,7 +33,8 @@ export default function PolicyDetailPage() {
 
   const { data: policies, loading: policiesLoading } = useHolderPolicies();
   const { data: readings, loading: oracleLoading, isFixture } = useOracleStatus();
-  const { data: claims, loading: claimsLoading } = useClaims();
+  const claimsResult = useClaims();
+  const { data: claims, loading: claimsLoading } = claimsResult;
 
   const policy = policies?.find((p) => String(p.id) === String(id)) ?? null;
   const coverageType = policy?.coverageType ?? policy?.coverage_type;
@@ -79,6 +80,7 @@ export default function PolicyDetailPage() {
 
           <section className="rounded-lg border border-gray-200 p-6">
             <h2 className="text-lg font-medium">Policy details</h2>
+            {claimsResult.isStale && <p role="status" className="mt-2 text-xs text-amber-700">Claim updates are temporarily disconnected. Showing the latest REST data.</p>}
             <dl className="mt-4 grid grid-cols-1 gap-4 sm:grid-cols-2">
               <div>
                 <dt className="text-sm text-gray-500">Policy ID</dt>