From 5f8a27d0a559052b2f763b60cc623736f1d0f1a0 Mon Sep 17 00:00:00 2001 From: suplex-deal Date: Wed, 21 Oct 2026 18:40:06 +0000 Subject: [PATCH] feat: stream realtime account updates --- docs/realtime-protocol.md | 25 ++ public/flags.json | 5 +- src/app/cover/page.tsx | 115 +++--- src/app/dashboard/policies/[id]/page.tsx | 4 +- src/app/page.tsx | 1 + src/app/provide/page.tsx | 42 ++- src/components/NotificationCenter.tsx | 21 +- src/components/layout/Navbar.tsx | 157 +++++++++ src/hooks/useClaims.ts | 106 ++++++ src/hooks/useHolderPolicies.ts | 51 +++ src/hooks/useLockupStatus.ts | 32 ++ src/hooks/useNotifications.ts | 130 ++++--- src/hooks/useOptimisticReconciliation.ts | 79 +++++ src/hooks/useOracleStatus.ts | 90 +++++ src/hooks/useRecentClaims.ts | 71 ++++ src/lib/api/claims.ts | 4 +- src/lib/api/client.ts | 12 +- src/lib/api/notifications.ts | 5 + src/lib/api/policies.ts | 4 +- src/lib/api/pool.ts | 8 +- src/lib/flags/flagsConfig.ts | 3 + src/lib/realtime/claims.test.ts | 47 +++ src/lib/realtime/claims.ts | 84 +++++ src/lib/realtime/client.test.ts | 198 +++++++++++ src/lib/realtime/client.ts | 330 ++++++++++++++++-- src/lib/realtime/notifications.test.ts | 35 ++ src/lib/realtime/notifications.ts | 53 +++ .../realtime/optimisticReconciliation.test.ts | 62 ++++ src/lib/realtime/optimisticReconciliation.ts | 102 ++++++ src/lib/store/useTransactionStore.ts | 2 + 30 files changed, 1735 insertions(+), 143 deletions(-) create mode 100644 docs/realtime-protocol.md create mode 100644 src/hooks/useOptimisticReconciliation.ts create mode 100644 src/lib/api/notifications.ts create mode 100644 src/lib/realtime/claims.test.ts create mode 100644 src/lib/realtime/claims.ts create mode 100644 src/lib/realtime/client.test.ts create mode 100644 src/lib/realtime/notifications.test.ts create mode 100644 src/lib/realtime/notifications.ts create mode 100644 src/lib/realtime/optimisticReconciliation.test.ts create mode 100644 src/lib/realtime/optimisticReconciliation.ts diff --git a/docs/realtime-protocol.md b/docs/realtime-protocol.md new file mode 100644 index 0000000..55d388d --- /dev/null +++ b/docs/realtime-protocol.md @@ -0,0 +1,25 @@ +# Realtime protocol + +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. + +## 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": "" }`. The response is `{ "token": "", "expiresAt": 1792600000 }`, where `expiresAt` is Unix seconds (Unix milliseconds are also accepted). The client sends `{ "type": "authenticate", "wallet": "", "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. + +## 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`. + +## 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`. diff --git a/public/flags.json b/public/flags.json index f8832b6..520b93b 100644 --- a/public/flags.json +++ b/public/flags.json @@ -1,5 +1,8 @@ { "flags": { - "realtime-oracle": false + "realtime-oracle": false, + "realtime-claims-settlement": false, + "realtime-notifications": false, + "realtime-transactions": false } } diff --git a/src/app/cover/page.tsx b/src/app/cover/page.tsx index a7a0cfd..dadb35d 100644 --- a/src/app/cover/page.tsx +++ b/src/app/cover/page.tsx @@ -29,6 +29,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"; @@ -80,6 +81,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(1); const [selectedType, setSelectedType] = useState(0); @@ -156,6 +158,7 @@ export default function CoverPage() { const expiryMs = Date.now() + durationDays * 86400000; const expiryDate = ; + 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; @@ -186,9 +189,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, @@ -203,11 +206,19 @@ export default function CoverPage() { } async function purchaseCartItem(item: CartItem, holder: string, networkPassphrase: string): Promise { + 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; @@ -239,26 +250,43 @@ 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; } 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) }); + let rejected = false; + let policyCorrection: Partial | undefined; + const releaseReconciliation = reconciliation.register({ + correlationId: txId, kind: "buy", + onConfirm: (authoritative) => { + if (authoritative?.policy && typeof authoritative.policy === "object") { + policyCorrection = authoritative.policy as Partial; + 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" }); @@ -274,25 +302,30 @@ 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 }); - setSubmission({ status: "success", result, demo: false, txHash }); + 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."); + 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, @@ -305,6 +338,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 }); @@ -315,6 +349,11 @@ export default function CoverPage() {
} /> {reviewDialog} + {reconciliation.notices.map((notice) => ( +
+ {notice.message} +
+ ))}
@@ -399,36 +438,15 @@ export default function CoverPage() { Step 1 of 4: Select Coverage Type
- {typesError && ( - -

Couldn't load coverage types: {typesError}

- - )} - - {typesLoading && ( -
- {Array.from({ length: 5 }).map((_, i -
- - {typesError && ( - -

Couldn't load coverage types: {typesError}

- - )} - - {typesLoading && ( -
- {Array.from({ length: 5 }).map((_, i) => ( - - ))} -
- } - errorRender={(err) => ( + + {Array.from({ length: 5 }).map((_, i) => )} + } + errorRender={(err: string) => (

Couldn't load coverage types: {err}

 - )} - > + )}> {coverageTypes && (
)} -
- setDurationDays(Number(e.target.value))} + className="w-full accent-pm-violet" aria-label="Coverage duration in days" /> +
1 day365 days
+ +
+ + +
+ + )} + {/* Right: Quote panel */}
{!ct ? ( diff --git a/src/app/dashboard/policies/[id]/page.tsx b/src/app/dashboard/policies/[id]/page.tsx index 032d7da..f69e8b7 100644 --- a/src/app/dashboard/policies/[id]/page.tsx +++ b/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() {

Policy details

+ {claimsResult.isStale &&

Claim updates are temporarily disconnected. Showing the latest REST data.

}
Policy ID
diff --git a/src/app/page.tsx b/src/app/page.tsx index 38753ab..e3f310f 100644 --- a/src/app/page.tsx +++ b/src/app/page.tsx @@ -355,6 +355,7 @@ export default function Home() { ⚠ Showing example data — the Refract API isn't reachable from this environment.

)} + {recentClaims.isStale &&

Live payouts are temporarily disconnected. Showing the latest REST data.

}
{(recentClaims.data ?? []).map((claim) => ( diff --git a/src/app/provide/page.tsx b/src/app/provide/page.tsx index afa3051..49d81ac 100644 --- a/src/app/provide/page.tsx +++ b/src/app/provide/page.tsx @@ -15,6 +15,7 @@ import { signAndSubmit } from "@/lib/wallet/signAndSubmit"; import { truncateAddress } from "@/lib/wallet/WalletProvider"; import { useNetworkGuard } from "@/hooks/useNetworkGuard"; import { useTransactionStore } from "@/lib/store/useTransactionStore"; +import { useOptimisticReconciliation } from "@/hooks/useOptimisticReconciliation"; // Illustrative allocation breakdown by coverage category — the backend // doesn't currently expose a per-category pool split, so this is presented @@ -40,6 +41,7 @@ export default function ProvidePage() { const { isCorrectNetwork, expectedNetwork } = useNetworkGuard(); const addTransaction = useTransactionStore((s) => s.addTransaction); const updateTransaction = useTransactionStore((s) => s.updateTransaction); + const reconciliation = useOptimisticReconciliation(); const { data: pool, loading: poolLoading, isFixture: poolIsFixture, dataSource: poolSource } = usePoolStats(); const { data: position } = useUserPoolPosition(wallet.status === "connected" ? wallet.address : null); const { lockupExpiresAt } = useLockupStatus(wallet.status === "connected" ? wallet.address : null); @@ -116,37 +118,57 @@ export default function ProvidePage() { setSubmission({ status: "submitting" }); const txId = addTransaction({ type: tab === "deposit" ? "provide" : "withdraw", amount: parsed }); + let actionCorrection: Record | undefined; + let rejected = false; + const releaseReconciliation = reconciliation.register({ + correlationId: txId, kind: tab === "deposit" ? "provide" : "withdraw", + onConfirm: (authoritative) => { + actionCorrection = authoritative; + setSubmission((current) => { + if (current.status !== "success" || !authoritative) return current; + return current.kind === "deposit" ? { ...current, result: { ...current.result, ...authoritative } as ProvideCapitalResponse } : { ...current, result: { ...current.result, ...authoritative } as WithdrawCapitalResponse }; + }); + 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 displayed balance update was reverted.` }); }, + }); try { if (tab === "deposit") { - const result = await provideCapital(wallet.address, toStroops(parsed)); + const result = await provideCapital(wallet.address, toStroops(parsed).toString(), txId); setSubmission({ status: "signing" }); updateTransaction(txId, { status: "signing" }); if (!wallet.networkPassphrase) { throw new Error("Wallet network isn't available — reconnect and try again"); } const txHash = await signAndSubmit(result.txXdr, wallet.address, wallet.networkPassphrase, "provide"); - setSubmission({ status: "success", kind: "deposit", result, demo: false, txHash }); + if (rejected) throw new Error("The transaction was rejected by the server."); + setSubmission({ status: "success", kind: "deposit", result: actionCorrection ? { ...result, ...actionCorrection } as ProvideCapitalResponse : result, demo: false, txHash }); updateTransaction(txId, { status: "confirmed", txHash }); } else { - const result = await withdrawCapital(wallet.address, toStroops(parsed / sharePrice)); + const result = await withdrawCapital(wallet.address, toStroops(parsed / sharePrice).toString(), txId); setSubmission({ status: "signing" }); updateTransaction(txId, { status: "signing" }); if (!wallet.networkPassphrase) { throw new Error("Wallet network isn't available — reconnect and try again"); } const txHash = await signAndSubmit(result.txXdr, wallet.address, wallet.networkPassphrase, "withdraw"); - setSubmission({ status: "success", kind: "withdraw", result, demo: false, txHash }); + if (rejected) throw new Error("The transaction was rejected by the server."); + setSubmission({ status: "success", kind: "withdraw", result: actionCorrection ? { ...result, ...actionCorrection } as WithdrawCapitalResponse : result, demo: false, txHash }); updateTransaction(txId, { status: "confirmed", txHash }); } } catch (err) { if (err instanceof ApiUnreachableError) { + releaseReconciliation(); // Backend unreachable in this environment — simulate locally, // clearly labeled, so the flow can still be exercised end-to-end. if (tab === "deposit") { const demoResult: ProvideCapitalResponse = { provider: wallet.address, - amountUsdc: toStroops(parsed), - sharesOut: toStroops(parsed / sharePrice), + amountUsdc: toStroops(parsed).toString(), + sharesOut: toStroops(parsed / sharePrice).toString(), sharePrice, txXdr: "DEMO_MODE — backend unreachable, no transaction was built", message: "Simulated locally: the Refract API is not running in this environment.", @@ -155,8 +177,8 @@ export default function ProvidePage() { } else { const demoResult: WithdrawCapitalResponse = { provider: wallet.address, - sharesIn: toStroops(parsed / sharePrice), - usdcOut: toStroops(parsed), + sharesIn: toStroops(parsed / sharePrice).toString(), + usdcOut: toStroops(parsed).toString(), sharePrice, txXdr: "DEMO_MODE — backend unreachable, no transaction was built", }; @@ -165,6 +187,7 @@ export default function ProvidePage() { updateTransaction(txId, { status: "confirmed", demo: true }); return; } + releaseReconciliation(); const message = err instanceof Error ? err.message : "Something went wrong"; setSubmission({ status: "error", message }); updateTransaction(txId, { status: "failed", error: message }); @@ -174,6 +197,9 @@ export default function ProvidePage() { return (
} /> + {reconciliation.notices.map((notice) => ( +
{notice.message}
+ ))}
diff --git a/src/components/NotificationCenter.tsx b/src/components/NotificationCenter.tsx index f5a5658..ccfd4de 100644 --- a/src/components/NotificationCenter.tsx +++ b/src/components/NotificationCenter.tsx @@ -4,6 +4,7 @@ import { useEffect, useRef, useState } from "react"; import Link from "next/link"; import { useNotifications } from "@/hooks/useNotifications"; import type { RefractNotification } from "@/lib/notifications"; +import { useReducedMotion } from "@/hooks/usePrefersReducedMotion"; const SEVERITY_ICONS: Record = { info: "ℹ", @@ -12,7 +13,10 @@ const SEVERITY_ICONS: Record = { critical: "⚡", }; -const SEVERITY_COLORS: Record = { +const SEVERITY_COLORS: Record< + RefractNotification["severity"], + { bg: string; text: string; border: string } +> = { info: { bg: "bg-pm-violet/10", text: "text-pm-violet", border: "border-pm-violet/20" }, warning: { bg: "bg-pm-amber/10", text: "text-pm-amber", border: "border-pm-amber/20" }, success: { bg: "bg-pm-green/10", text: "text-pm-green", border: "border-pm-green/20" }, @@ -21,7 +25,8 @@ const SEVERITY_COLORS: Record(null); const buttonRef = useRef(null); @@ -83,7 +88,8 @@ export function NotificationCenter() {
@@ -94,6 +100,11 @@ export function NotificationCenter() { )}
+ {isStale && ( +

+ Live notifications are disconnected. Showing account data from REST. +

+ )} {unreadCount > 0 && ( + {right} +
+ + + +
+ + {menuOpen && ( +
+
+ {NAV_LINKS.map((item) => { + const active = pathname === item.href; + return ( + setMenuOpen(false)} + className={cn( + "rounded-md px-3 py-2.5 text-sm font-medium", + active ? "bg-pm-violet/10 text-pm-violet" : "text-pm-text/70" + )} + > + {item.label} + + ); + })} +
+
{right}
+
+ )} + + ); +} diff --git a/src/hooks/useClaims.ts b/src/hooks/useClaims.ts index e69de29..1715035 100644 --- a/src/hooks/useClaims.ts +++ b/src/hooks/useClaims.ts @@ -0,0 +1,106 @@ +"use client"; + +import { useEffect, useState } from "react"; +import { useWallet } from "@/lib/wallet/WalletProvider"; +import { fetchHolderClaims, type ClaimRecord } from "@/lib/api/claims"; +import type { Policy } from "@/lib/api/policies"; +import { fixtureClaimsForHolder } from "@/lib/fixtures/claims"; +import { useFeatureFlag } from "@/lib/flags/useFeatureFlag"; +import { + mergeClaims, + normalizeClaimFrame, + normalizeRestClaim, + type StreamClaim, +} from "@/lib/realtime/claims"; +import { realtimeClient, type RealtimeState } from "@/lib/realtime/client"; + +export type ClaimsResult = ClaimRecord[] & { + data: ClaimRecord[]; + loading: boolean; + error: string | null; + isFixture: boolean; + isStale: boolean; +}; + +export function useClaims(address?: string | null, policies?: Policy[] | null): ClaimsResult { + const wallet = useWallet(); + const holder = + address === undefined ? (wallet.status === "connected" ? wallet.address : null) : address; + const enabled = useFeatureFlag("realtime-claims-settlement"); + const [restRows, setRestRows] = useState([]); + const [pushedRows, setPushedRows] = useState([]); + const [loading, setLoading] = useState(Boolean(holder)); + const [error, setError] = useState(null); + const [isFixture, setIsFixture] = useState(false); + const [isStale, setIsStale] = useState(false); + + useEffect(() => { + let active = true; + const controller = new AbortController(); + setRestRows([]); + setPushedRows([]); + if (!holder) { + setLoading(false); + setError(null); + setIsFixture(false); + return () => controller.abort(); + } + setLoading(true); + setError(null); + const load = async () => { + let cursor: string | undefined; + const seenCursors = new Set(); + const rows: ClaimRecord[] = []; + try { + do { + if (cursor && seenCursors.has(cursor)) + throw new Error("Claims pagination cursor repeated"); + if (cursor) seenCursors.add(cursor); + const page = await fetchHolderClaims(holder, { cursor, limit: 50 }, controller.signal); + rows.push(...page.items.map(normalizeRestClaim)); + cursor = page.hasMore ? page.nextCursor : undefined; + if (page.hasMore && !cursor) + throw new Error("Claims pagination response omitted nextCursor"); + } while (cursor); + if (active) { + setRestRows(rows); + setIsFixture(false); + setLoading(false); + } + } catch (cause) { + if (!active) return; + setRestRows(fixtureClaimsForHolder(holder, policies ?? []).map(normalizeRestClaim)); + setIsFixture(true); + setError(cause instanceof Error ? cause.message : "Unable to load claims"); + setLoading(false); + } + }; + void load(); + return () => { + active = false; + controller.abort(); + }; + }, [holder, policies]); + + useEffect(() => { + realtimeClient.setIdentity(holder); + if (!enabled || !holder) { + setIsStale(false); + return; + } + return realtimeClient.subscribe( + "claims.settlement", + (payload) => { + const claim = normalizeClaimFrame(payload); + if (!claim || (claim.holder && claim.holder !== holder)) return; + setPushedRows((current) => mergeClaims(current, [claim])); + }, + { authenticated: true, onState: (state: RealtimeState) => setIsStale(state !== "live") } + ); + }, [enabled, holder]); + + const rows = mergeClaims(restRows, pushedRows) as ClaimRecord[]; + const result = rows as ClaimsResult; + Object.assign(result, { data: rows, loading, error, isFixture, isStale }); + return result; +} diff --git a/src/hooks/useHolderPolicies.ts b/src/hooks/useHolderPolicies.ts index e69de29..c8cb460 100644 --- a/src/hooks/useHolderPolicies.ts +++ b/src/hooks/useHolderPolicies.ts @@ -0,0 +1,51 @@ +"use client"; + +import { useEffect, useState } from "react"; +import { fetchHolderPolicies, type Policy } from "@/lib/api/policies"; +import { fixtureHolderPolicies } from "@/lib/fixtures/policies"; +import { useWallet } from "@/lib/wallet/WalletProvider"; + +export function useHolderPolicies(address?: string | null) { + const wallet = useWallet(); + const holder = + address === undefined ? (wallet.status === "connected" ? wallet.address : null) : address; + const [data, setData] = useState(null); + const [loading, setLoading] = useState(Boolean(holder)); + const [error, setError] = useState(null); + const [isFixture, setIsFixture] = useState(false); + const [isStale] = useState(false); + + useEffect(() => { + let active = true; + const controller = new AbortController(); + setData(null); + if (!holder) { + setLoading(false); + setError(null); + setIsFixture(false); + return () => controller.abort(); + } + setLoading(true); + fetchHolderPolicies(holder, undefined, controller.signal) + .then((response) => { + if (!active) return; + setData(response.items); + setIsFixture(false); + setError(null); + setLoading(false); + }) + .catch((cause) => { + if (!active) return; + setData(fixtureHolderPolicies(holder)); + setIsFixture(true); + setError(cause instanceof Error ? cause.message : "Unable to load policies"); + setLoading(false); + }); + return () => { + active = false; + controller.abort(); + }; + }, [holder]); + + return { data, loading, error, isFixture, isStale }; +} diff --git a/src/hooks/useLockupStatus.ts b/src/hooks/useLockupStatus.ts index e69de29..f15109c 100644 --- a/src/hooks/useLockupStatus.ts +++ b/src/hooks/useLockupStatus.ts @@ -0,0 +1,32 @@ +"use client"; + +import { useEffect, useState } from "react"; +import { fetchLockupStatus } from "@/lib/api/pool"; + +export function useLockupStatus(address: string | null) { + const [state, setState] = useState<{ lockupExpiresAt: number | null; loading: boolean }>({ + lockupExpiresAt: null, + loading: false, + }); + useEffect(() => { + if (!address) { + setState({ lockupExpiresAt: null, loading: false }); + return; + } + const controller = new AbortController(); + setState((previous) => ({ ...previous, loading: true })); + fetchLockupStatus(address, controller.signal) + .then(({ lockupExpiresAt }) => { + if (!controller.signal.aborted) + setState({ + lockupExpiresAt: lockupExpiresAt ? Number(lockupExpiresAt) : null, + loading: false, + }); + }) + .catch(() => { + if (!controller.signal.aborted) setState({ lockupExpiresAt: null, loading: false }); + }); + return () => controller.abort(); + }, [address]); + return state; +} diff --git a/src/hooks/useNotifications.ts b/src/hooks/useNotifications.ts index e8fe606..d632ebe 100644 --- a/src/hooks/useNotifications.ts +++ b/src/hooks/useNotifications.ts @@ -1,81 +1,131 @@ "use client"; -import { useEffect, useMemo, useState } from "react"; +import { useCallback, useEffect, useMemo, useState } from "react"; import { useWallet } from "@/lib/wallet/WalletProvider"; import { useHolderPolicies } from "./useHolderPolicies"; import { useClaims } from "./useClaims"; import { useLockupStatus } from "./useLockupStatus"; import { useOracleStatus } from "./useOracleStatus"; -import { - deriveNotifications, - type RefractNotification, -} from "@/lib/notifications"; +import { deriveNotifications, type RefractNotification } from "@/lib/notifications"; +import { useFeatureFlag } from "@/lib/flags/useFeatureFlag"; +import { realtimeClient, type RealtimeState } from "@/lib/realtime/client"; +import { markNotificationRead } from "@/lib/api/notifications"; +import { mergeNotifications, normalizeNotification } from "@/lib/realtime/notifications"; const STORAGE_KEY = "refract_read_notifications_v1"; - export function useNotifications() { const wallet = useWallet(); const address = wallet.status === "connected" ? wallet.address : null; - + const enabled = useFeatureFlag("realtime-notifications"); const { data: policies } = useHolderPolicies(address); const claims = useClaims(address, policies); const { lockupExpiresAt } = useLockupStatus(address); const { data: oracleReadings } = useOracleStatus(); - const [readIds, setReadIds] = useState>(new Set()); + const [pushed, setPushed] = useState([]); + const [isStale, setIsStale] = useState(false); useEffect(() => { + setPushed([]); + if (!address) { + setReadIds(new Set()); + return; + } try { - const stored = localStorage.getItem(STORAGE_KEY); - if (stored) { - setReadIds(new Set(JSON.parse(stored))); - } + const stored = localStorage.getItem(`${STORAGE_KEY}:${address}`); + setReadIds(new Set(stored ? (JSON.parse(stored) as string[]) : [])); } catch { - // Ignore localStorage errors + setReadIds(new Set()); } - }, []); + }, [address]); - const saveReadIds = (newSet: Set) => { - setReadIds(newSet); - try { - localStorage.setItem(STORAGE_KEY, JSON.stringify(Array.from(newSet))); - } catch { - // Ignore + useEffect(() => { + realtimeClient.setIdentity(address); + if (!enabled || !address) { + setIsStale(false); + return; } - }; + return realtimeClient.subscribe( + "notifications", + (payload) => { + const notification = normalizeNotification(payload); + if (!notification) return; + setPushed((current) => mergeNotifications(current, notification)); + }, + { authenticated: true, onState: (state: RealtimeState) => setIsStale(state !== "live") } + ); + }, [enabled, address]); + + const saveReadIds = useCallback( + (next: Set) => { + setReadIds(next); + if (!address) return; + try { + localStorage.setItem(`${STORAGE_KEY}:${address}`, JSON.stringify([...next])); + } catch { + /* storage is optional */ + } + }, + [address] + ); const notifications = useMemo(() => { - return deriveNotifications({ + const derived = deriveNotifications({ policies, claims, lockupExpiresAt, oracleReadings, readIds, }); - }, [policies, claims, lockupExpiresAt, oracleReadings, readIds]); - - const unreadCount = useMemo(() => { - return notifications.filter((n) => !n.read).length; - }, [notifications]); + const byId = new Map(derived.map((item) => [item.id, item])); + for (const item of pushed) + byId.set(item.id, { ...item, read: readIds.has(item.id) || item.read }); + return [...byId.values()].sort((a, b) => b.timestamp - a.timestamp); + }, [policies, claims, lockupExpiresAt, oracleReadings, readIds, pushed]); - const markAsRead = (id: string) => { - const updated = new Set(readIds); - updated.add(id); - saveReadIds(updated); - }; + const markAsRead = useCallback( + (id: string) => { + const wasRead = readIds.has(id); + if (!wasRead) saveReadIds(new Set([...readIds, id])); + void markNotificationRead(id).catch(() => { + if (!wasRead) + setReadIds((current) => { + const next = new Set(current); + next.delete(id); + try { + if (address) + localStorage.setItem(`${STORAGE_KEY}:${address}`, JSON.stringify([...next])); + } catch { + /* optional */ + } + return next; + }); + }); + }, + [address, readIds, saveReadIds] + ); - const markAllAsRead = () => { - const updated = new Set(readIds); - for (const n of notifications) { - updated.add(n.id); - } - saveReadIds(updated); - }; + const markAllAsRead = useCallback(() => { + const ids = notifications.map((item) => item.id); + const previous = readIds; + const next = new Set([...readIds, ...ids]); + saveReadIds(next); + void Promise.all(ids.map((id) => markNotificationRead(id))).catch(() => { + setReadIds(previous); + if (address) + try { + localStorage.setItem(`${STORAGE_KEY}:${address}`, JSON.stringify([...previous])); + } catch { + /* optional */ + } + }); + }, [address, notifications, readIds, saveReadIds]); return { notifications, - unreadCount, + unreadCount: notifications.filter((n) => !n.read).length, markAsRead, markAllAsRead, + isStale, }; } diff --git a/src/hooks/useOptimisticReconciliation.ts b/src/hooks/useOptimisticReconciliation.ts new file mode 100644 index 0000000..253b3a8 --- /dev/null +++ b/src/hooks/useOptimisticReconciliation.ts @@ -0,0 +1,79 @@ +"use client"; + +import { useCallback, useEffect, useState } from "react"; +import { useWallet } from "@/lib/wallet/WalletProvider"; +import { realtimeClient } from "@/lib/realtime/client"; +import { + parseOptimisticEvent, + registerOptimisticAction, + resolveOptimisticAction, + type OptimisticActionKind, +} from "@/lib/realtime/optimisticReconciliation"; +import { useFeatureFlag } from "@/lib/flags/useFeatureFlag"; +import { apiRequest } from "@/lib/api/client"; + +export function useOptimisticReconciliation() { + const wallet = useWallet(); + const enabled = useFeatureFlag("realtime-transactions"); + const [notices, setNotices] = useState>([]); + + useEffect(() => { + const identity = wallet.status === "connected" ? wallet.address : null; + realtimeClient.setIdentity(identity); + if (!enabled || !identity) return; + return realtimeClient.subscribe( + "transactions.confirmation", + (payload) => { + const event = parseOptimisticEvent(payload); + if (!event) return; + resolveOptimisticAction(event); + }, + { authenticated: true } + ); + }, [enabled, wallet.status, wallet.address]); + + const register = useCallback( + (input: { + correlationId: string; + kind: OptimisticActionKind; + onConfirm: (authoritative?: Record) => void; + onRollback: (reason: string) => void; + }) => + registerOptimisticAction({ + ...input, + onRollback: (reason) => { + setNotices((current) => [ + { id: input.correlationId, message: `${reason} The displayed update was reverted.` }, + ...current, + ]); + input.onRollback(reason); + }, + onUnverified: () => + setNotices((current) => [ + { + id: input.correlationId, + message: + "This transaction is still awaiting confirmation. The account view will continue to use REST data.", + }, + ...current, + ]), + verify: async () => { + const result = await apiRequest<{ status: "confirmed" | "rejected" | "pending" }>( + "/transactions/verify", + { + method: "POST", + body: { correlationId: input.correlationId }, + } + ); + return result.status; + }, + }), + [] + ); + + const dismiss = useCallback( + (id: string) => setNotices((current) => current.filter((item) => item.id !== id)), + [] + ); + return { register, notices, dismiss }; +} diff --git a/src/hooks/useOracleStatus.ts b/src/hooks/useOracleStatus.ts index e69de29..98be27c 100644 --- a/src/hooks/useOracleStatus.ts +++ b/src/hooks/useOracleStatus.ts @@ -0,0 +1,90 @@ +"use client"; + +import { useCallback, useEffect, useState } from "react"; +import { fetchOracleStatus, type OracleReading } from "@/lib/api/oracle"; +import { FIXTURE_ORACLE_READINGS } from "@/lib/fixtures/oracle"; +import { useFeatureFlag } from "@/lib/flags/useFeatureFlag"; +import { realtimeClient, type RealtimeState } from "@/lib/realtime/client"; + +export function useOracleStatus() { + const enabled = useFeatureFlag("realtime-oracle"); + const [data, setData] = useState([]); + const [loading, setLoading] = useState(true); + const [isFixture, setIsFixture] = useState(false); + const [isStale, setIsStale] = useState(false); + const [lastUpdated, setLastUpdated] = useState(null); + const [isRefreshing, setIsRefreshing] = useState(false); + + const refresh = useCallback(async () => { + setIsRefreshing(true); + try { + const result = await fetchOracleStatus(); + setData(result.readings); + setIsFixture(false); + setLastUpdated(new Date()); + } catch { + setData(FIXTURE_ORACLE_READINGS); + setIsFixture(true); + setLastUpdated(new Date()); + } finally { + setLoading(false); + setIsRefreshing(false); + } + }, []); + + useEffect(() => { + let active = true; + const controller = new AbortController(); + const load = () => + fetchOracleStatus(controller.signal) + .then((result) => { + if (!active) return; + setData(result.readings); + setIsFixture(false); + setLastUpdated(new Date()); + setLoading(false); + }) + .catch(() => { + if (!active) return; + setData(FIXTURE_ORACLE_READINGS); + setIsFixture(true); + setLastUpdated(new Date()); + setLoading(false); + }); + void load(); + const interval = setInterval(load, 60_000); + return () => { + active = false; + controller.abort(); + clearInterval(interval); + }; + }, []); + + useEffect(() => { + if (!enabled) { + setIsStale(false); + return; + } + return realtimeClient.subscribe( + "oracle_update", + (payload) => { + const value = + payload && typeof payload === "object" && "reading" in payload + ? (payload as { reading: unknown }).reading + : payload; + if (!value || typeof value !== "object") return; + const reading = value as OracleReading; + if (!reading.coverageType) return; + setData((current) => [ + reading, + ...current.filter((item) => item.coverageType !== reading.coverageType), + ]); + setIsFixture(false); + setLastUpdated(new Date()); + }, + { onState: (state: RealtimeState) => setIsStale(state !== "live") } + ); + }, [enabled]); + + return { data, loading, isFixture, isStale, lastUpdated, refresh, isRefreshing }; +} diff --git a/src/hooks/useRecentClaims.ts b/src/hooks/useRecentClaims.ts index e69de29..554fcc7 100644 --- a/src/hooks/useRecentClaims.ts +++ b/src/hooks/useRecentClaims.ts @@ -0,0 +1,71 @@ +"use client"; + +import { useEffect, useState } from "react"; +import { fetchRecentClaims } from "@/lib/api/claims"; +import type { ClaimRecord } from "@/lib/api/claims"; +import { FIXTURE_RECENT_CLAIMS } from "@/lib/fixtures/recentClaims"; +import { useFeatureFlag } from "@/lib/flags/useFeatureFlag"; +import { + mergeClaims, + normalizeClaimFrame, + normalizeRestClaim, + type StreamClaim, +} from "@/lib/realtime/claims"; +import { realtimeClient, type RealtimeState } from "@/lib/realtime/client"; + +export function useRecentClaims() { + const enabled = useFeatureFlag("realtime-claims-settlement"); + const [restRows, setRestRows] = useState([]); + const [pushedRows, setPushedRows] = useState([]); + const [loading, setLoading] = useState(true); + const [isFixture, setIsFixture] = useState(false); + const [isStale, setIsStale] = useState(false); + + useEffect(() => { + let active = true; + const controller = new AbortController(); + setLoading(true); + fetchRecentClaims({ limit: 20 }, controller.signal) + .then((page) => { + if (!active) return; + setRestRows(page.items.map(normalizeRestClaim)); + setIsFixture(false); + setLoading(false); + }) + .catch(() => { + if (!active) return; + setRestRows(FIXTURE_RECENT_CLAIMS.map(normalizeRestClaim)); + setIsFixture(true); + setLoading(false); + }); + return () => { + active = false; + controller.abort(); + }; + }, []); + + useEffect(() => { + if (!enabled) { + setIsStale(false); + return; + } + return realtimeClient.subscribe( + "claims.settlement", + (payload) => { + const claim = normalizeClaimFrame(payload); + if ( + !claim || + (claim.status && ["rejected", "failed", "cancelled"].includes(claim.status.toLowerCase())) + ) + return; + setPushedRows((current) => mergeClaims(current, [claim])); + }, + { onState: (state: RealtimeState) => setIsStale(state !== "live") } + ); + }, [enabled]); + + // A page arriving after a push replaces the REST page source; merging here + // keeps the cursor buffer untouched and removes any event/page duplicate. + const data = mergeClaims(restRows, pushedRows, 20) as ClaimRecord[]; + return { data, loading, isFixture, isStale }; +} diff --git a/src/lib/api/claims.ts b/src/lib/api/claims.ts index ddbdd2f..b55a381 100644 --- a/src/lib/api/claims.ts +++ b/src/lib/api/claims.ts @@ -47,7 +47,7 @@ export function fetchHolderClaims( params: PaginationParams = {}, signal?: AbortSignal, ): Promise> { - return apiRequest(`/claims/holder/${address}${buildPaginationQuery(params)}`, { signal }).then( + return apiRequest<{ claims?: ClaimRecord[]; items?: ClaimRecord[]; nextCursor?: string; hasMore?: boolean }>(`/claims/holder/${address}${buildPaginationQuery(params)}`, { signal }).then( normalizeClaimsResponse, ); } @@ -56,7 +56,7 @@ export function fetchRecentClaims( params: PaginationParams = {}, signal?: AbortSignal, ): Promise> { - return apiRequest(`/claims/recent${buildPaginationQuery(params)}`, { signal }).then( + return apiRequest<{ claims?: ClaimRecord[]; items?: ClaimRecord[]; nextCursor?: string; hasMore?: boolean }>(`/claims/recent${buildPaginationQuery(params)}`, { signal }).then( normalizeClaimsResponse, ); } diff --git a/src/lib/api/client.ts b/src/lib/api/client.ts index 832ddd0..821a8a2 100644 --- a/src/lib/api/client.ts +++ b/src/lib/api/client.ts @@ -10,7 +10,11 @@ export const API_BASE_URL = process.env.NEXT_PUBLIC_API_URL ?? "http://localhost:4001/api/v1"; export class ApiError extends Error { - constructor(message: string, public status: number, public details?: unknown) { + constructor( + message: string, + public status: number, + public details?: unknown + ) { super(message); this.name = "ApiError"; } @@ -28,6 +32,7 @@ interface RequestOptions { method?: "GET" | "POST"; body?: unknown; signal?: AbortSignal; + idempotencyKey?: string; } export async function apiRequest(path: string, opts: RequestOptions = {}): Promise { @@ -35,7 +40,10 @@ export async function apiRequest(path: string, opts: RequestOptions = {}): Pr try { res = await fetch(`${API_BASE_URL}${path}`, { method: opts.method ?? "GET", - headers: opts.body ? { "Content-Type": "application/json" } : undefined, + headers: { + ...(opts.body ? { "Content-Type": "application/json" } : {}), + ...(opts.idempotencyKey ? { "Idempotency-Key": opts.idempotencyKey } : {}), + }, body: opts.body ? JSON.stringify(opts.body) : undefined, signal: opts.signal, }); diff --git a/src/lib/api/notifications.ts b/src/lib/api/notifications.ts new file mode 100644 index 0000000..f286b57 --- /dev/null +++ b/src/lib/api/notifications.ts @@ -0,0 +1,5 @@ +import { apiRequest } from "./client"; + +export function markNotificationRead(id: string): Promise { + return apiRequest(`/notifications/${encodeURIComponent(id)}/read`, { method: "POST" }); +} diff --git a/src/lib/api/policies.ts b/src/lib/api/policies.ts index 04e35ca..9b4ef9a 100644 --- a/src/lib/api/policies.ts +++ b/src/lib/api/policies.ts @@ -106,6 +106,6 @@ export async function fetchHolderPolicies( return toPaginated(raw, "policies"); } -export function buyPolicy(params: BuyPolicyParams): Promise { - return apiRequest("/policies/buy", { method: "POST", body: params }); +export function buyPolicy(params: BuyPolicyParams, correlationId?: string): Promise { + return apiRequest("/policies/buy", { method: "POST", body: params, idempotencyKey: correlationId }); } diff --git a/src/lib/api/pool.ts b/src/lib/api/pool.ts index bafd82d..44e91ec 100644 --- a/src/lib/api/pool.ts +++ b/src/lib/api/pool.ts @@ -71,10 +71,10 @@ export function fetchPoolAllocation(signal?: AbortSignal): Promise { - return apiRequest("/pool/provide", { method: "POST", body: { provider, amount } }); +export function provideCapital(provider: string, amount: string, correlationId?: string): Promise { + return apiRequest("/pool/provide", { method: "POST", body: { provider, amount }, idempotencyKey: correlationId }); } -export function withdrawCapital(provider: string, shares: string): Promise { - return apiRequest("/pool/withdraw", { method: "POST", body: { provider, shares } }); +export function withdrawCapital(provider: string, shares: string, correlationId?: string): Promise { + return apiRequest("/pool/withdraw", { method: "POST", body: { provider, shares }, idempotencyKey: correlationId }); } diff --git a/src/lib/flags/flagsConfig.ts b/src/lib/flags/flagsConfig.ts index 2ecb795..67d61a5 100644 --- a/src/lib/flags/flagsConfig.ts +++ b/src/lib/flags/flagsConfig.ts @@ -5,6 +5,9 @@ export interface FlagsDocument { /** Safe default if the remote flags source is unreachable — new/risky features default off. */ export const DEFAULT_FLAGS: Record = { "realtime-oracle": false, + "realtime-claims-settlement": false, + "realtime-notifications": false, + "realtime-transactions": false, }; export const FLAGS_ENDPOINT = "/flags.json"; diff --git a/src/lib/realtime/claims.test.ts b/src/lib/realtime/claims.test.ts new file mode 100644 index 0000000..015766f --- /dev/null +++ b/src/lib/realtime/claims.test.ts @@ -0,0 +1,47 @@ +import { describe, expect, it } from "vitest"; +import { mergeClaims, normalizeClaimFrame, type StreamClaim } from "./claims"; + +const claim = (fields: Partial = {}): StreamClaim => ({ + policyId: "policy-1", + holder: "wallet", + coverageType: 0, + triggered: true, + payout: "100", + reason: "settled", + processedAt: 10, + status: "settled", + ...fields, +}); + +describe("claim settlement merge", () => { + it("normalizes envelope frames and prefers the contract claim id", () => { + const normalized = normalizeClaimFrame({ + dedupKey: "d1", + claim: { id: "c1", policy_id: "p1", processed_at: 12, payout: 5 }, + }); + expect(normalized).toMatchObject({ id: "c1", policyId: "p1", processedAt: 12, dedupKey: "d1" }); + }); + + it("inserts newest pushed rows and deduplicates replay against REST pages", () => { + const rows = mergeClaims( + [claim({ id: "old", processedAt: 1 })], + [claim({ id: "new", processedAt: 3 }), claim({ id: "new", processedAt: 3 })] + ); + expect(rows.map((item) => item.id)).toEqual(["new", "old"]); + }); + + it("updates a known pending claim and ignores a regressive replay after terminal settlement", () => { + const pending = claim({ status: "pending", triggered: false, processedAt: 2 }); + const settled = claim({ status: "settled", processedAt: 4, payout: "200" }); + expect(mergeClaims([pending], [settled])[0]?.payout).toBe("200"); + expect( + mergeClaims([settled], [claim({ status: "pending", triggered: false, processedAt: 3 })])[0] + ?.status + ).toBe("settled"); + }); + + it("reconciles a pushed row when a later REST page returns it", () => { + const pushed = claim({ id: "claim-1", processedAt: 20 }); + expect(mergeClaims([pushed], [pushed])).toHaveLength(1); + }); +}); diff --git a/src/lib/realtime/claims.ts b/src/lib/realtime/claims.ts new file mode 100644 index 0000000..cfe643d --- /dev/null +++ b/src/lib/realtime/claims.ts @@ -0,0 +1,84 @@ +import type { ClaimRecord } from "@/lib/api/claims"; + +export type StreamClaim = ClaimRecord & { + id?: string; + claimId?: string; + status?: string; + dedupKey?: string; +}; + +function normalizeClaim(value: unknown, fallbackStatus: string): StreamClaim | null { + const candidate = value; + if (!candidate || typeof candidate !== "object") return null; + const row = candidate as Record; + const policyId = String(row.policyId ?? row.policy_id ?? ""); + if (!policyId) return null; + const processedAt = Number( + row.processedAt ?? row.processed_at ?? row.settledAt ?? row.timestamp ?? Date.now() / 1000 + ); + const payout = row.payout == null ? "0" : String(row.payout); + return { + policyId, + holder: String(row.holder ?? row.wallet ?? ""), + coverageType: Number(row.coverageType ?? row.coverage_type ?? 0), + triggered: row.triggered !== false, + payout, + reason: String(row.reason ?? "Claim settled"), + processedAt: Number.isFinite(processedAt) ? processedAt : Date.now() / 1000, + settlementTxHash: + typeof (row.settlementTxHash ?? row.settlement_tx_hash) === "string" + ? String(row.settlementTxHash ?? row.settlement_tx_hash) + : undefined, + id: typeof (row.claimId ?? row.id) === "string" ? String(row.claimId ?? row.id) : undefined, + claimId: typeof row.claimId === "string" ? row.claimId : undefined, + status: typeof row.status === "string" ? row.status : fallbackStatus, + dedupKey: typeof row.dedupKey === "string" ? row.dedupKey : undefined, + }; +} + +export function normalizeRestClaim(value: ClaimRecord): StreamClaim { + return normalizeClaim(value, value.triggered ? "settled" : "pending") ?? value; +} + +export function normalizeClaimFrame(value: unknown): StreamClaim | null { + const outer = value && typeof value === "object" ? (value as Record) : null; + const candidate = outer?.claim && typeof outer.claim === "object" ? outer.claim : value; + const normalized = normalizeClaim(candidate, "settled"); + if (!normalized) return null; + if (typeof outer?.dedupKey === "string") normalized.dedupKey = outer.dedupKey; + return normalized; +} + +export function claimKey(claim: StreamClaim): string { + return claim.dedupKey ?? claim.claimId ?? claim.id ?? claim.settlementTxHash ?? claim.policyId; +} + +const terminal = new Set(["settled", "paid", "rejected", "failed", "cancelled"]); + +/** Reconcile pushed rows with REST rows while keeping the REST order and page buffer. */ +export function mergeClaims( + rest: StreamClaim[], + pushed: StreamClaim[], + limit = Number.POSITIVE_INFINITY +): StreamClaim[] { + const merged = new Map(); + for (const claim of rest) merged.set(claimKey(claim), claim); + for (const claim of pushed) { + const key = claimKey(claim); + const existing = merged.get(key); + if (existing) { + const existingTerminal = terminal.has( + String(existing.status ?? (existing.triggered ? "settled" : "pending")).toLowerCase() + ); + const incomingTerminal = terminal.has( + String(claim.status ?? (claim.triggered ? "settled" : "pending")).toLowerCase() + ); + const older = Number(claim.processedAt) < Number(existing.processedAt); + if ((!existingTerminal || incomingTerminal) && (!older || !existingTerminal)) + merged.set(key, { ...existing, ...claim }); + } else merged.set(key, claim); + } + return [...merged.values()] + .sort((a, b) => Number(b.processedAt) - Number(a.processedAt)) + .slice(0, limit); +} diff --git a/src/lib/realtime/client.test.ts b/src/lib/realtime/client.test.ts new file mode 100644 index 0000000..ce12014 --- /dev/null +++ b/src/lib/realtime/client.test.ts @@ -0,0 +1,198 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { realtimeClient } from "./client"; +import { ApiError } from "@/lib/api/client"; + +class MockSocket { + static OPEN = 1; + readyState = 0; + url: string; + sent: string[] = []; + listeners = new Map void>>(); + constructor(url: string) { + this.url = url; + sockets.push(this); + } + addEventListener(type: string, listener: (event: any) => void) { + this.listeners.set(type, [...(this.listeners.get(type) ?? []), listener]); + } + send(data: string) { + this.sent.push(data); + } + close(code = 1000) { + this.readyState = 3; + this.emit("close", { code }); + } + emit(type: string, event: any = {}) { + if (type === "open") this.readyState = 1; + this.listeners.get(type)?.forEach((listener) => listener(event)); + } +} + +let sockets: MockSocket[] = []; +const originalWebSocket = globalThis.WebSocket; + +describe("authenticated realtime client", () => { + afterEach(() => { + realtimeClient.disconnect(); + realtimeClient.setIdentity(null); + vi.useRealTimers(); + sockets = []; + globalThis.WebSocket = originalWebSocket; + }); + + it("sends credentials only in the handshake and subscribes after authentication", async () => { + sockets = []; + globalThis.WebSocket = MockSocket as unknown as typeof WebSocket; + realtimeClient.setTokenProvider(async () => ({ + token: "secret-token", + expiresAt: Date.now() + 60_000, + })); + realtimeClient.setIdentity("G-WALLET"); + const handler = vi.fn(); + const unsubscribe = realtimeClient.subscribe("notifications", handler, { authenticated: true }); + await Promise.resolve(); + const socket = sockets[0]!; + socket.emit("open"); + const authFrame = JSON.parse(socket.sent[0]!); + expect(authFrame).toMatchObject({ + type: "authenticate", + wallet: "G-WALLET", + token: "secret-token", + }); + expect(socket.url).not.toContain("secret-token"); + socket.emit("message", { data: JSON.stringify({ type: "authenticated" }) }); + expect(socket.sent.map((frame) => JSON.parse(frame).topic).filter(Boolean)).toEqual([ + "notifications", + ]); + socket.emit("message", { + data: JSON.stringify({ topic: "notifications", payload: { id: "n1" } }), + }); + expect(handler).toHaveBeenCalledWith({ id: "n1" }); + unsubscribe(); + }); + + it("refreshes before expiry without resubscribing", async () => { + vi.useFakeTimers(); + sockets = []; + globalThis.WebSocket = MockSocket as unknown as typeof WebSocket; + const provider = vi.fn(async () => ({ + token: `token-${provider.mock.calls.length}`, + expiresAt: Date.now() + 35_000, + })); + realtimeClient.setTokenProvider(provider); + realtimeClient.setIdentity("G-WALLET"); + const unsubscribe = realtimeClient.subscribe("notifications", vi.fn(), { authenticated: true }); + await Promise.resolve(); + const socket = sockets[0]!; + socket.emit("open"); + socket.emit("message", { data: JSON.stringify({ type: "authenticated" }) }); + await vi.advanceTimersByTimeAsync(5_000); + expect(provider).toHaveBeenCalledTimes(2); + expect(socket.sent.filter((frame) => JSON.parse(frame).type === "authenticate")).toHaveLength( + 2 + ); + expect(socket.sent.filter((frame) => JSON.parse(frame).type === "subscribe")).toHaveLength(1); + unsubscribe(); + }); + + it("reauthenticates and restores a topic exactly once after reconnect", async () => { + vi.useFakeTimers(); + globalThis.WebSocket = MockSocket as unknown as typeof WebSocket; + realtimeClient.setTokenProvider(async () => ({ + token: "fresh", + expiresAt: Date.now() + 60_000, + })); + realtimeClient.setIdentity("G-WALLET"); + const unsubscribe = realtimeClient.subscribe("notifications", vi.fn(), { authenticated: true }); + await Promise.resolve(); + const first = sockets[0]!; + first.emit("open"); + first.emit("message", { data: JSON.stringify({ type: "authenticated" }) }); + first.close(1006); + await vi.advanceTimersByTimeAsync(1_000); + await Promise.resolve(); + const second = sockets[1]!; + second.emit("open"); + second.emit("message", { data: JSON.stringify({ type: "authenticated" }) }); + expect(second.sent.filter((frame) => JSON.parse(frame).type === "authenticate")).toHaveLength( + 1 + ); + expect(second.sent.filter((frame) => JSON.parse(frame).type === "subscribe")).toHaveLength(1); + unsubscribe(); + }); + + it("marks auth failures as terminal without retrying", async () => { + vi.useFakeTimers(); + globalThis.WebSocket = MockSocket as unknown as typeof WebSocket; + const states: string[] = []; + realtimeClient.setTokenProvider(async () => { + throw new ApiError("denied", 403); + }); + realtimeClient.setIdentity("G-WALLET"); + const unsubscribePrivate = realtimeClient.subscribe("notifications", vi.fn(), { + authenticated: true, + onState: (state) => states.push(state), + }); + const unsubscribePublic = realtimeClient.subscribe("claims.settlement", vi.fn()); + await Promise.resolve(); + expect(states).toContain("auth_failed"); + sockets[0]!.emit("open"); + expect(sockets[0]!.sent.some((frame) => JSON.parse(frame).topic === "claims.settlement")).toBe( + true + ); + await vi.advanceTimersByTimeAsync(60_000); + expect(sockets).toHaveLength(1); + unsubscribePrivate(); + unsubscribePublic(); + }); + + it("discards a stale token request when the account changes", async () => { + globalThis.WebSocket = MockSocket as unknown as typeof WebSocket; + let releaseOld!: (value: { token: string; expiresAt: number }) => void; + realtimeClient.setTokenProvider((wallet) => + wallet === "G-OLD" + ? new Promise((resolve) => { + releaseOld = resolve; + }) + : Promise.resolve({ token: "new-wallet-token", expiresAt: Date.now() + 60_000 }) + ); + realtimeClient.setIdentity("G-OLD"); + const unsubscribe = realtimeClient.subscribe("notifications", vi.fn(), { authenticated: true }); + realtimeClient.setIdentity("G-NEW"); + releaseOld({ token: "old-wallet-token", expiresAt: Date.now() + 60_000 }); + await Promise.resolve(); + await Promise.resolve(); + expect(sockets).toHaveLength(1); + sockets[0]!.emit("open"); + expect(JSON.parse(sockets[0]!.sent[0]!).wallet).toBe("G-NEW"); + expect(sockets[0]!.sent[0]).not.toContain("old-wallet-token"); + unsubscribe(); + }); + + it("keeps public topics alive after an authenticated topic is denied", async () => { + vi.useFakeTimers(); + globalThis.WebSocket = MockSocket as unknown as typeof WebSocket; + const provider = vi.fn(async () => ({ + token: "private-token", + expiresAt: Date.now() + 60_000, + })); + realtimeClient.setTokenProvider(provider); + realtimeClient.setIdentity("G-WALLET"); + const unsubscribePrivate = realtimeClient.subscribe("notifications", vi.fn(), { + authenticated: true, + }); + const unsubscribePublic = realtimeClient.subscribe("claims.settlement", vi.fn()); + await Promise.resolve(); + sockets[0]!.emit("open"); + sockets[0]!.emit("message", { data: JSON.stringify({ type: "auth_error", status: 403 }) }); + await vi.advanceTimersByTimeAsync(1_000); + await Promise.resolve(); + sockets[1]!.emit("open"); + expect(provider).toHaveBeenCalledOnce(); + expect(sockets[1]!.sent.map((frame) => JSON.parse(frame).topic).filter(Boolean)).toEqual([ + "claims.settlement", + ]); + unsubscribePrivate(); + unsubscribePublic(); + }); +}); diff --git a/src/lib/realtime/client.ts b/src/lib/realtime/client.ts index 8be8d3c..e692447 100644 --- a/src/lib/realtime/client.ts +++ b/src/lib/realtime/client.ts @@ -1,84 +1,344 @@ -import { API_BASE_URL } from "@/lib/api/client"; +import { API_BASE_URL, ApiError } from "@/lib/api/client"; +export type RealtimeState = "connecting" | "live" | "stale" | "auth_failed"; type Handler = (payload: unknown) => void; +type StateHandler = (state: RealtimeState) => void; +type TokenProvider = (wallet: string) => Promise<{ token: string; expiresAt: number }>; const MAX_BACKOFF_MS = 30_000; const BASE_BACKOFF_MS = 1_000; +const REFRESH_SKEW_MS = 30_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. - */ +async function fetchWalletToken(wallet: string) { + const response = await fetch(`${API_BASE_URL}/realtime/token`, { + method: "POST", + credentials: "include", + cache: "no-store", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ wallet }), + }); + if (!response.ok) throw new ApiError("Realtime authentication failed", response.status); + const result = (await response.json()) as { token: string; expiresAt: number }; + if (!result.token || !Number.isFinite(result.expiresAt)) + throw new Error("Invalid realtime token response"); + return result; +} + +interface Subscription { + handlers: Set; + stateHandlers: Set; + authenticated: boolean; + state: RealtimeState; +} + +/** Reconnecting topic client. Credentials are only sent in WebSocket frames. */ class RealtimeClient { private socket: WebSocket | null = null; - private handlers = new Map>(); + private subscriptions = new Map(); private attempt = 0; private closedByUser = false; private reconnectTimer: ReturnType | null = null; + private refreshTimer: ReturnType | null = null; + private authRetryTimer: ReturnType | null = null; + private identity: string | null = null; + private identityGeneration = 0; + private tokenProvider: TokenProvider = fetchWalletToken; + private authenticated = false; + private opening = false; + private subscribedTopics = new Set(); + + setTokenProvider(provider: TokenProvider): void { + this.tokenProvider = provider; + } + + setIdentity(wallet: string | null): void { + if (wallet === this.identity) return; + this.identity = wallet; + this.identityGeneration += 1; + this.clearRefreshTimer(); + if (this.authRetryTimer) clearTimeout(this.authRetryTimer); + this.authRetryTimer = null; + this.authenticated = false; + this.socket?.close(4001, "identity changed"); + this.socket = null; + for (const [topic, sub] of this.subscriptions) { + if (sub.authenticated) this.setState(topic, wallet ? "connecting" : "stale"); + } + if (this.subscriptions.size) this.connect(); + } connect(): void { - if (typeof WebSocket === "undefined" || this.socket) return; + if (typeof WebSocket === "undefined" || this.socket || this.opening) return; this.closedByUser = false; this.open(); } - private open(): void { + private async open(): Promise { + if (this.socket || this.opening || this.closedByUser) return; + this.opening = true; + const generation = this.identityGeneration; + const walletIdentity = this.identity; + const needsAuth = [...this.subscriptions.values()].some( + (sub) => sub.authenticated && sub.state !== "auth_failed" + ); + let token: { token: string; expiresAt: number } | null = null; + if (needsAuth) { + if (!walletIdentity) { + this.setAllAuthState("stale"); + } else + try { + token = await this.tokenProvider(walletIdentity); + } catch (error) { + if (error instanceof ApiError && (error.status === 401 || error.status === 403)) { + this.setAllAuthState("auth_failed"); + } else { + this.setAllAuthState("stale"); + this.scheduleAuthRetry(); + } + } + if ( + token && + (token.expiresAt < 10_000_000_000 ? token.expiresAt * 1000 : token.expiresAt) <= Date.now() + ) { + token = null; + this.setAllAuthState("auth_failed"); + } + } + if (generation !== this.identityGeneration || walletIdentity !== this.identity) { + this.opening = false; + this.open(); + return; + } + if (!token && needsAuth && !this.hasPublicTopics()) { + this.opening = false; + return; + } + if (this.closedByUser) { + this.opening = false; + return; + } const wsUrl = API_BASE_URL.replace(/^http/, "ws").replace(/\/api\/v1$/, "/ws"); try { this.socket = new WebSocket(wsUrl); } catch { + this.opening = false; this.scheduleReconnect(); return; } - - this.socket.addEventListener("open", () => { + const socket = this.socket; + socket.addEventListener("open", () => { + if (this.socket !== socket || generation !== this.identityGeneration) { + socket.close(); + return; + } + this.opening = false; this.attempt = 0; - }); - 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. + if (token || !this.hasRetryableAuth()) { + if (this.authRetryTimer) clearTimeout(this.authRetryTimer); + this.authRetryTimer = null; } + if (token) { + socket.send( + JSON.stringify({ type: "authenticate", wallet: walletIdentity, token: token.token }) + ); + this.scheduleRefresh(token.expiresAt); + } else this.subscribeActiveTopics(); }); - this.socket.addEventListener("close", () => { - this.socket = null; - if (!this.closedByUser) this.scheduleReconnect(); - }); - this.socket.addEventListener("error", () => { - this.socket?.close(); + socket.addEventListener("message", (event) => this.receive(event.data, socket)); + socket.addEventListener("close", (event) => { + const wasCurrent = this.socket === socket; + if (wasCurrent) this.socket = null; + this.subscribedTopics.clear(); + if (!wasCurrent) return; + this.authenticated = false; + this.subscribedTopics.clear(); + this.clearRefreshTimer(); + const authFailure = event.code === 4401 || event.code === 4403; + for (const [topic, sub] of this.subscriptions) { + this.setState(topic, authFailure && sub.authenticated ? "auth_failed" : "stale"); + } + const hasPublicTopics = [...this.subscriptions.values()].some((sub) => !sub.authenticated); + if (!this.closedByUser && (!authFailure || hasPublicTopics)) this.scheduleReconnect(); }); + socket.addEventListener("error", () => socket.close()); + } + + private receive(raw: unknown, socket: WebSocket): void { + try { + const frame = JSON.parse(raw as string) as { + type?: string; + topic?: string; + payload?: unknown; + status?: number; + expiresAt?: number; + }; + if (frame.type === "authenticated") { + this.authenticated = true; + this.subscribeActiveTopics(); + } else if ( + frame.type === "auth_error" || + (frame.type === "error" && (frame.status === 401 || frame.status === 403)) + ) { + this.authenticated = false; + this.setAllAuthState("auth_failed"); + socket.close(4401, "authentication failed"); + } else if (frame.type === "token_expiring") { + void this.refreshToken(socket); + } else if (frame.topic) { + this.subscriptions.get(frame.topic)?.handlers.forEach((handler) => handler(frame.payload)); + } + } catch { + /* malformed frames are ignored */ + } + } + + private subscribeActiveTopics(): void { + for (const [topic, sub] of this.subscriptions) { + if ((!sub.authenticated || this.authenticated) && !this.subscribedTopics.has(topic)) { + this.socket?.send(JSON.stringify({ type: "subscribe", topic })); + this.subscribedTopics.add(topic); + this.setState(topic, "live"); + } + } + } + + private async refreshToken(socket: WebSocket): Promise { + if (!this.identity || socket !== this.socket) return; + try { + const token = await this.tokenProvider(this.identity); + if (socket !== this.socket) return; + socket.send( + JSON.stringify({ + type: "authenticate", + wallet: this.identity, + token: token.token, + refresh: true, + }) + ); + this.scheduleRefresh(token.expiresAt); + } catch (error) { + if (error instanceof ApiError && (error.status === 401 || error.status === 403)) { + this.setAllAuthState("auth_failed"); + socket.close(4401, "authentication failed"); + } else socket.close(); + } + } + + private scheduleRefresh(expiresAt: number): void { + this.clearRefreshTimer(); + const expiryMs = expiresAt < 10_000_000_000 ? expiresAt * 1000 : expiresAt; + const delay = Math.max(1_000, expiryMs - Date.now() - REFRESH_SKEW_MS); + this.refreshTimer = setTimeout(() => { + if (this.socket) void this.refreshToken(this.socket); + }, delay); + } + + private clearRefreshTimer(): void { + if (this.refreshTimer) clearTimeout(this.refreshTimer); + this.refreshTimer = null; + } + + private scheduleAuthRetry(): void { + if (this.authRetryTimer || this.closedByUser) return; + const delay = Math.min(BASE_BACKOFF_MS * 2 ** this.attempt, MAX_BACKOFF_MS); + this.attempt += 1; + this.authRetryTimer = setTimeout(() => { + this.authRetryTimer = null; + if (!this.identity) return; + if (this.socket) this.socket.close(); + else this.open(); + }, delay); } private scheduleReconnect(): void { - if (this.reconnectTimer) return; + if (this.reconnectTimer || this.closedByUser) return; const delay = Math.min(BASE_BACKOFF_MS * 2 ** this.attempt, MAX_BACKOFF_MS); this.attempt += 1; + for (const [topic, sub] of this.subscriptions) { + if (sub.state !== "auth_failed") this.setState(topic, "stale"); + } this.reconnectTimer = setTimeout(() => { this.reconnectTimer = null; - if (!this.closedByUser) this.open(); + 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(); + private setState(topic: string, state: RealtimeState): void { + const sub = this.subscriptions.get(topic); + if (!sub || sub.state === state) return; + sub.state = state; + sub.stateHandlers.forEach((handler) => handler(state)); + } + private setAllAuthState(state: RealtimeState): void { + for (const [topic, sub] of this.subscriptions) + if (sub.authenticated) this.setState(topic, state); + } + + private hasRetryableAuth(): boolean { + return [...this.subscriptions.values()].some( + (sub) => sub.authenticated && sub.state !== "auth_failed" + ); + } + + private hasPublicTopics(): boolean { + return [...this.subscriptions.values()].some((sub) => !sub.authenticated); + } + subscribe( + topic: string, + handler: Handler, + options: { authenticated?: boolean; onState?: StateHandler } = {} + ): () => void { + let sub = this.subscriptions.get(topic); + if (!sub) { + sub = { + handlers: new Set(), + stateHandlers: new Set(), + authenticated: options.authenticated ?? false, + state: "stale", + }; + this.subscriptions.set(topic, sub); + } + const requiresNewAuth = Boolean(options.authenticated) && !sub.authenticated; + sub.authenticated = sub.authenticated || Boolean(options.authenticated); + sub.handlers.add(handler); + if (options.onState) { + sub.stateHandlers.add(options.onState); + options.onState(sub.state); + } + if (requiresNewAuth && this.socket) { + this.socket.close(4001, "authentication required"); + this.socket = null; + } + this.connect(); + if ( + this.socket?.readyState === WebSocket.OPEN && + (!sub.authenticated || this.authenticated) && + !this.subscribedTopics.has(topic) + ) { + this.socket.send(JSON.stringify({ type: "subscribe", topic })); + this.subscribedTopics.add(topic); + this.setState(topic, "live"); + } return () => { - this.handlers.get(topic)?.delete(handler); + const current = this.subscriptions.get(topic); + current?.handlers.delete(handler); + if (options.onState) current?.stateHandlers.delete(options.onState); + if (current && current.handlers.size === 0) { + this.socket?.send(JSON.stringify({ type: "unsubscribe", topic })); + this.subscribedTopics.delete(topic); + this.subscriptions.delete(topic); + } }; } disconnect(): void { this.closedByUser = true; if (this.reconnectTimer) clearTimeout(this.reconnectTimer); + this.reconnectTimer = null; + if (this.authRetryTimer) clearTimeout(this.authRetryTimer); + this.authRetryTimer = null; + this.clearRefreshTimer(); this.socket?.close(); this.socket = null; } diff --git a/src/lib/realtime/notifications.test.ts b/src/lib/realtime/notifications.test.ts new file mode 100644 index 0000000..0c3402f --- /dev/null +++ b/src/lib/realtime/notifications.test.ts @@ -0,0 +1,35 @@ +import { describe, expect, it } from "vitest"; +import { mergeNotifications, normalizeNotification } from "./notifications"; + +const item = (id: string, timestamp: number, title = id) => ({ + id, + type: "claim_payout" as const, + title, + message: "Claim settled", + timestamp, + read: false, + severity: "success" as const, +}); + +describe("notification stream merge", () => { + it("normalizes notification envelopes", () => { + expect(normalizeNotification({ notification: item("n1", 5) })).toMatchObject({ + id: "n1", + timestamp: 5, + }); + }); + + it("orders inserts, deduplicates replay, and keeps newer values on old replay", () => { + let rows = mergeNotifications([], item("new", 10)); + rows = mergeNotifications(rows, item("old", 5)); + rows = mergeNotifications(rows, item("new", 10, "updated")); + rows = mergeNotifications(rows, item("new", 2, "replay")); + expect(rows.map((entry) => entry.id)).toEqual(["new", "old"]); + expect(rows[0]?.title).toBe("updated"); + }); + + it("preserves an item's read state when a later event amends it", () => { + const read = { ...item("n1", 5), read: true }; + expect(mergeNotifications([read], item("n1", 6, "Amended"))[0]?.read).toBe(true); + }); +}); diff --git a/src/lib/realtime/notifications.ts b/src/lib/realtime/notifications.ts new file mode 100644 index 0000000..fe8d7d7 --- /dev/null +++ b/src/lib/realtime/notifications.ts @@ -0,0 +1,53 @@ +import type { RefractNotification } from "@/lib/notifications"; + +const notificationTypes = new Set([ + "policy_expiring", + "claim_payout", + "lockup_expired", + "lockup_soon", + "oracle_trigger", +]); + +export function normalizeNotification(value: unknown): RefractNotification | null { + const outer = value && typeof value === "object" ? (value as Record) : null; + const item = + outer?.notification && typeof outer.notification === "object" + ? (outer.notification as Record) + : outer; + if ( + !item || + typeof item.id !== "string" || + typeof item.type !== "string" || + !notificationTypes.has(item.type as RefractNotification["type"]) + ) + return null; + const timestamp = Number(item.timestamp ?? item.createdAt ?? Date.now()); + const severity = ["info", "warning", "success", "critical"].includes(String(item.severity)) + ? (item.severity as RefractNotification["severity"]) + : "info"; + return { + id: item.id, + type: item.type as RefractNotification["type"], + title: String(item.title ?? "Refract update"), + message: String(item.message ?? "There is a new update on your account."), + timestamp: Number.isFinite(timestamp) ? timestamp : Date.now(), + read: item.read === true, + href: typeof item.href === "string" ? item.href : undefined, + severity, + }; +} + +export function mergeNotifications( + current: RefractNotification[], + incoming: RefractNotification, + limit = 100 +): RefractNotification[] { + const previous = current.find((item) => item.id === incoming.id); + const item = + previous && previous.timestamp > incoming.timestamp + ? previous + : { ...previous, ...incoming, read: Boolean(previous?.read || incoming.read) }; + return [item, ...current.filter((candidate) => candidate.id !== item.id)] + .sort((a, b) => b.timestamp - a.timestamp) + .slice(0, limit); +} diff --git a/src/lib/realtime/optimisticReconciliation.test.ts b/src/lib/realtime/optimisticReconciliation.test.ts new file mode 100644 index 0000000..25bb341 --- /dev/null +++ b/src/lib/realtime/optimisticReconciliation.test.ts @@ -0,0 +1,62 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { + parseOptimisticEvent, + registerOptimisticAction, + resolveOptimisticAction, +} from "./optimisticReconciliation"; + +describe("optimistic action reconciliation", () => { + afterEach(() => vi.useRealTimers()); + + it("confirms or amends an action once and ignores replayed confirmations", () => { + const onConfirm = vi.fn(); + registerOptimisticAction({ + correlationId: "confirm-1", + kind: "buy", + onConfirm, + onRollback: vi.fn(), + verify: vi.fn(), + }); + const event = parseOptimisticEvent({ + correlationId: "confirm-1", + outcome: "amended", + authoritative: { amount: 12 }, + }); + expect(event && resolveOptimisticAction(event)).toBe(true); + expect(event && resolveOptimisticAction(event)).toBe(false); + expect(onConfirm).toHaveBeenCalledOnce(); + expect(onConfirm).toHaveBeenCalledWith({ amount: 12 }); + }); + + it("rolls back rejected actions idempotently", () => { + const onRollback = vi.fn(); + registerOptimisticAction({ + correlationId: "reject-1", + kind: "provide", + onConfirm: vi.fn(), + onRollback, + verify: vi.fn(), + }); + const event = { correlationId: "reject-1", outcome: "rejected" as const, reason: "capacity" }; + expect(resolveOptimisticAction(event)).toBe(true); + expect(resolveOptimisticAction(event)).toBe(false); + expect(onRollback).toHaveBeenCalledOnce(); + expect(onRollback).toHaveBeenCalledWith("capacity"); + }); + + it("uses REST verification after the pending timeout", async () => { + vi.useFakeTimers(); + const verify = vi.fn().mockResolvedValue("rejected"); + const rollback = vi.fn(); + registerOptimisticAction({ + correlationId: "timeout-1", + kind: "withdraw", + onConfirm: vi.fn(), + onRollback: rollback, + verify, + }); + await vi.advanceTimersByTimeAsync(90_000); + expect(verify).toHaveBeenCalledOnce(); + expect(rollback).toHaveBeenCalledOnce(); + }); +}); diff --git a/src/lib/realtime/optimisticReconciliation.ts b/src/lib/realtime/optimisticReconciliation.ts new file mode 100644 index 0000000..d4dfa9b --- /dev/null +++ b/src/lib/realtime/optimisticReconciliation.ts @@ -0,0 +1,102 @@ +export type OptimisticActionKind = "buy" | "provide" | "withdraw"; +export interface OptimisticEvent { + correlationId: string; + outcome: "confirmed" | "amended" | "rejected"; + reason?: string; + authoritative?: Record; +} +interface PendingAction { + kind: OptimisticActionKind; + onConfirm: (authoritative?: Record) => void; + onRollback: (reason: string) => void; + onUnverified?: () => void; + verify: () => Promise<"confirmed" | "rejected" | "pending">; + timer: ReturnType; +} + +const pending = new Map(); +const resolved = new Map(); +const TIMEOUT_MS = 90_000; +const RESOLVED_TTL_MS = 10 * 60_000; + +function rememberResolved(id: string) { + resolved.set(id, Date.now() + RESOLVED_TTL_MS); + for (const [key, expiry] of resolved) if (expiry < Date.now()) resolved.delete(key); +} + +export function registerOptimisticAction(input: { + correlationId: string; + kind: OptimisticActionKind; + onConfirm: (authoritative?: Record) => void; + onRollback: (reason: string) => void; + onUnverified?: () => void; + verify: () => Promise<"confirmed" | "rejected" | "pending">; +}): () => void { + const previous = pending.get(input.correlationId); + if (previous) clearTimeout(previous.timer); + const timer = setTimeout(async () => { + const current = pending.get(input.correlationId); + if (!current) return; + try { + const result = await current.verify(); + if (result === "confirmed") + resolveOptimisticAction({ correlationId: input.correlationId, outcome: "confirmed" }); + else if (result === "rejected") + resolveOptimisticAction({ + correlationId: input.correlationId, + outcome: "rejected", + reason: "The server rejected this action.", + }); + else { + pending.delete(input.correlationId); + rememberResolved(input.correlationId); + current.onUnverified?.(); + } + } catch { + pending.delete(input.correlationId); + rememberResolved(input.correlationId); + current.onUnverified?.(); + } + }, TIMEOUT_MS); + pending.set(input.correlationId, { ...input, timer }); + return () => { + const current = pending.get(input.correlationId); + if (current?.timer === timer) { + clearTimeout(timer); + pending.delete(input.correlationId); + } + }; +} + +export function resolveOptimisticAction(event: OptimisticEvent): boolean { + if (!event.correlationId || resolved.has(event.correlationId)) return false; + const action = pending.get(event.correlationId); + if (!action) return false; + clearTimeout(action.timer); + pending.delete(event.correlationId); + rememberResolved(event.correlationId); + if (event.outcome === "rejected") + action.onRollback(event.reason || "The action could not be confirmed."); + else action.onConfirm(event.authoritative); + return true; +} + +export function parseOptimisticEvent(value: unknown): OptimisticEvent | null { + const outer = value && typeof value === "object" ? (value as Record) : null; + const record = + outer?.event && typeof outer.event === "object" + ? (outer.event as Record) + : outer; + if (!record || typeof record.correlationId !== "string") return null; + const outcome = record.outcome ?? record.status; + if (outcome !== "confirmed" && outcome !== "amended" && outcome !== "rejected") return null; + return { + correlationId: record.correlationId, + outcome, + reason: typeof record.reason === "string" ? record.reason : undefined, + authoritative: + record.authoritative && typeof record.authoritative === "object" + ? (record.authoritative as Record) + : undefined, + }; +} diff --git a/src/lib/store/useTransactionStore.ts b/src/lib/store/useTransactionStore.ts index e2ea2a0..59b623f 100644 --- a/src/lib/store/useTransactionStore.ts +++ b/src/lib/store/useTransactionStore.ts @@ -13,6 +13,8 @@ export interface TransactionEntry { timestamp: number; txHash?: string; error?: string; + /** Authoritative fields returned by the backend/realtime confirmation. */ + authoritative?: Record; /** Simulated client-side because the API was unreachable. */ demo?: boolean; }