diff --git a/docs/realtime-protocol.md b/docs/realtime-protocol.md index acc08a7..89a4d64 100644 --- a/docs/realtime-protocol.md +++ b/docs/realtime-protocol.md @@ -1,7 +1,9 @@ -# Realtime protocol (v1) +# Realtime protocol WebSocket at `ws(s):///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):///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": "" }`. 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. + ## 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):///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`. diff --git a/public/flags.json b/public/flags.json index 0f388b6..03577bf 100644 --- a/public/flags.json +++ b/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, diff --git a/src/app/cover/page.tsx b/src/app/cover/page.tsx index e5df608..26b5208 100644 --- a/src/app/cover/page.tsx +++ b/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(1); const [selectedType, setSelectedType] = useState(0); @@ -182,6 +184,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; @@ -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 { + 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,8 +276,8 @@ 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); @@ -274,19 +285,36 @@ export default function CoverPage() { 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 | 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" }); @@ -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() {
} /> {reviewDialog} + {reconciliation.notices.map((notice) => ( +
+ {notice.message} +
+ ))}
@@ -429,19 +468,15 @@ export default function CoverPage() { Step 1 of 4: Select Coverage Type
- {typesError && ( + + {Array.from({ length: 5 }).map((_, i) => )} + } + errorRender={(err: string) => ( -

Couldn't load coverage types: {typesError}

+

Couldn't load coverage types: {err}

 - )} - - {typesLoading && !coverageTypes && ( -
- {Array.from({ length: 5 }).map((_, i) => ( - - ))} -
- )} + )}> {coverageTypes && (
)} + +
- setDurationDays(Number(event.target.value))} - className="w-full accent-violet-500" - /> + setDurationDays(Number(e.target.value))} + className="w-full accent-pm-violet" aria-label="Coverage duration in days" /> +
1 day365 days
+
+
+ + +
+ + )} + + />
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 c3a7341..e69de29 100644 --- a/src/app/page.tsx +++ b/src/app/page.tsx @@ -1,401 +0,0 @@ -"use client"; - -import { useState, useEffect, useRef } from "react"; -import { Navbar, Footer } from "@/components/layout"; -import { Container, Button, Badge, Card, Skeleton } from "@/components/ui"; -import { usePrefersReducedMotion } from "@/hooks/usePrefersReducedMotion"; -import { useOracleStatus } from "@/hooks/useOracleStatus"; -import { useRecentClaims } from "@/hooks/useRecentClaims"; -import { usePoolStats } from "@/hooks/usePoolStats"; -import { useClaimStats } from "@/hooks/useClaimStats"; -import { useProtocolStats } from "@/hooks/useProtocolStats"; -import type { OracleReading } from "@/lib/domain/realtime"; -import type { BadgeTone } from "@/components/ui/Badge"; -import { formatUsd, formatRelativeTime, fromStroops } from "@/lib/format"; -import { truncateAddress } from "@/lib/wallet/WalletProvider"; -import { ACTIVE_NETWORK } from "@/lib/network"; -import { HowItWorks } from "@/components/HowItWorks"; -import { DataFreshness } from "@/components/DataFreshness"; -import { RealtimeBoundary } from "@/components/realtime/RealtimeBoundary"; -import { RealtimeProfiler } from "@/lib/realtime/devProfiler"; - -function Counter({ to, prefix = "", suffix = "", decimals = 0 }: { to: number; prefix?: string; suffix?: string; decimals?: number }) { - const [val, setVal] = useState(0); - const reducedMotion = usePrefersReducedMotion(); - - useEffect(() => { - if (reducedMotion) { - setVal(to); - return; - } - const duration = 2000; - const start = performance.now(); - const from = val; - let frame = 0; - const tick = (now: number) => { - const t = Math.min((now - start) / duration, 1); - const ease = 1 - Math.pow(1 - t, 3); - setVal(from + (to - from) * ease); - if (t < 1) frame = requestAnimationFrame(tick); - }; - frame = requestAnimationFrame(tick); - return () => cancelAnimationFrame(frame); - }, [to, reducedMotion]); - return <>{prefix}{val.toLocaleString("en-US", { minimumFractionDigits: decimals, maximumFractionDigits: decimals })}{suffix}; -} - -const COVERAGE_TYPES = [ - { icon: "🪙", name: "Stablecoin Depeg", desc: "Auto-pays if USDC breaks $0.95 peg", color: "#8b5cf6", risk: "Medium" }, - { icon: "📉", name: "Market Crash", desc: "Covers 30%+ drawdown in 24 hours", color: "#f59e0b", risk: "High" }, - { icon: "🛡️", name: "Liquidation Shield", desc: "Saves your DeFi position from liquidation loss", color: "#10b981", risk: "High" }, - { icon: "🔐", name: "Smart Contract Risk", desc: "Protocol hack or exploit coverage", color: "#ef4444", risk: "Critical" }, - { icon: "✈️", name: "Flight Delay", desc: "Instant payout for 2hr+ flight delays", color: "#06b6d4", risk: "Low" }, -]; - -const COVERAGE_NAME_BY_TYPE = COVERAGE_TYPES.map((c) => c.name); - -const ORACLE_FEED_DISPLAY: Record string }> = { - StablecoinDepeg: { feed: "USDC/USD", icon: "🪙", formatValue: (r) => `$${r.value.toFixed(4)}` }, - MarketCrash: { feed: "Market 24h", icon: "📊", formatValue: (r) => `${r.value >= 0 ? "+" : ""}${r.value.toFixed(2)}%` }, - SmartContractRisk: { feed: "Protocol TVL 24h", icon: "🔐", formatValue: (r) => `${r.value >= 0 ? "+" : ""}${r.value.toFixed(2)}%` }, -}; - -const SEVERITY_BADGE_TONE: Record = { - low: "safe", - medium: "risk", - high: "danger", - triggered: "danger", -}; - -const TRIGGER_DETAILS = [ - "USDC price < $0.95 for 15+ min", - "Market index -30% in 24h", - "Collateral ratio < 85%", - "Protocol TVL drops >50% in 1hr", - "Flight delayed >120 min (AviationStack)", -]; - -export default function Home() { - const [activeType, setActiveType] = useState(0); - const tabRefs = useRef>({}); - const ct = COVERAGE_TYPES[activeType]; - const oracleStatus = useOracleStatus(); - const recentClaims = useRecentClaims(); - const protocolStats = useProtocolStats(); - const statsLoading = protocolStats.loading; - const anyTriggered = oracleStatus.data?.some((r) => r.severity === "triggered") ?? false; - const [secondsAgo, setSecondsAgo] = useState(0); - - useEffect(() => { - if (!oracleStatus.lastUpdated) return; - const interval = setInterval(() => { - const diff = Math.floor((Date.now() - (oracleStatus.lastUpdated?.getTime() ?? Date.now())) / 1000); - setSecondsAgo(Math.max(0, diff)); - }, 1000); - return () => clearInterval(interval); - }, [oracleStatus.lastUpdated]); - - return ( -
- - - - - } - /> - -
- {/* Hero */} - -
-
-
- -

- Parametric insurance -
- - that pays itself. - -

- -

- No claims. No adjusters. No waiting. When an oracle confirms your trigger condition on-chain, Refract - automatically transfers your payout. Trustless coverage for DeFi, markets, and real-world events. -

- -
- - -
- - {/* Stats */} -
-
- -
-
- {[ - { - val: protocolStats.data ? fromStroops(protocolStats.data.tvl) : 0, - prefix: "$", - suffix: "", - label: "Total Value Protected", - dec: 0, - payout: false, - }, - { - val: 0, - prefix: "", - suffix: "", - label: "Cumulative Payouts", - dec: 0, - payout: true, - }, - { - val: protocolStats.data?.activePolicies ?? 0, - prefix: "", - suffix: "", - label: "Policies Active", - dec: 0, - payout: false, - }, - ].map((s) => ( -
-
- {statsLoading ? ( - - ) : ( - - )} -
-
{s.label}
-
- ))} -
-
-
- - {/* Hero right: Live coverage card */} -
-

Live oracle readings are unavailable. Other Refract sections remain available.

}> - -
-
- Live Oracle Status - - -
-
-
-
- -
    - {(oracleStatus.data ?? []).map((reading) => { - const display = ORACLE_FEED_DISPLAY[reading.coverageType]; - if (!display) return null; - return ( -
  • -
    - - {display.feed} -
    -
    - - {display.formatValue(reading)} - - - {reading.severity === "low" ? "Safe" : reading.severity === "triggered" ? "Triggered" : "Elevated"} - -
    -
    • - ); - })} -
- -
-
-
- {anyTriggered ? "One or more triggers active" : "No triggers active"} -
- {oracleStatus.lastUpdated && ( - - {secondsAgo === 0 ? "Just updated" : `${secondsAgo}s ago`} - - )} -
-
- {oracleStatus.isFixture - ? "Example data — Refract API unreachable" - : oracleStatus.isStale - ? "Live feed interrupted — showing last known data, refreshing every 60 seconds" - : "Oracles auto-refreshed every 60 seconds"} -
-
- - - - - - {/* Coverage Types */} -
- -
-

- Five coverage categories -

-

- Each policy is backed by real capital in the Refract risk pool and settles automatically via oracle proof. -

-
- -
{ - if (!["ArrowLeft", "ArrowRight", "Home", "End"].includes(e.key)) return; - e.preventDefault(); - let next = activeType; - if (e.key === "ArrowLeft") next = (activeType - 1 + COVERAGE_TYPES.length) % COVERAGE_TYPES.length; - if (e.key === "ArrowRight") next = (activeType + 1) % COVERAGE_TYPES.length; - if (e.key === "Home") next = 0; - if (e.key === "End") next = COVERAGE_TYPES.length - 1; - setActiveType(next); - tabRefs.current[next]?.focus(); - }} - > - {COVERAGE_TYPES.map((t, i) => ( - - ))} -
- - -
-
- -

{ct.name}

-

{ct.desc}

-
- = 1 ? "risk" : "safe"}>{ct.risk} Risk - Auto-Settle - No Claims Form -
-
- -
-
-
- {[ - { label: "How it triggers", value: TRIGGER_DETAILS[activeType] }, - { label: "Settlement time", value: "< 60 seconds after oracle confirms" }, - { label: "Oracle source", value: activeType === 4 ? "AviationStack API + Refract relay" : "Band Protocol + Pyth Network" }, - { label: "Dispute process", value: "None — oracle data is final and immutable" }, - ].map((item) => ( -
-
{item.label}
-
{item.value}
-
- ))} -
-
- - -
- - {/* Scroll-driven How it works section */} - - - {/* Recent payouts */} -

Recent payouts

Payout updates are temporarily unavailable.

}> -
- -
-

Recent payouts

- -
-
- {recentClaims.isFixture && ( -

- ⚠ Showing example data — the Refract API isn't reachable from this environment. -

- )} - {recentClaims.isStale &&

Saved payouts — refreshing live data…

} -
- {(recentClaims.data ?? []).map((claim) => ( - -
- {COVERAGE_NAME_BY_TYPE[claim.coverageType] ?? "Unknown"} - {formatRelativeTime(claim.processedAt)} -
-
- {formatUsd(fromStroops(claim.payout))} -
-
- tx: {claim.settlementTxHash ? truncateAddress(claim.settlementTxHash, 3, 3) : "pending"} -
- - ))} -
- -
- -
- -
-
- ); -} diff --git a/src/components/NotificationCenter.tsx b/src/components/NotificationCenter.tsx index 5ad3a1d..e69de29 100644 --- a/src/components/NotificationCenter.tsx +++ b/src/components/NotificationCenter.tsx @@ -1,204 +0,0 @@ -"use client"; - -import { useEffect, useRef, useState } from "react"; -import Link from "next/link"; -import { useNotifications } from "@/hooks/useNotifications"; -import { usePreferencesStore } from "@/lib/store/preferencesStore"; -import type { RefractNotification } from "@/lib/notifications"; -import { VirtualList } from "@/components/ui/VirtualList"; - -/** Above this many items the list is windowed; below it the original markup is unchanged. */ -const VIRTUALIZE_THRESHOLD = 50; -const ROW_HEIGHT = 120; - -const SEVERITY_ICONS: Record = { - info: "ℹ", - warning: "⚠", - success: "✓", - critical: "⚡", -}; - -const SEVERITY_COLORS: Record = { - 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" }, - critical: { bg: "bg-pm-red/10", text: "text-pm-red", border: "border-pm-red/20" }, -}; - -export function NotificationCenter() { - const [isOpen, setIsOpen] = useState(false); - const { notifications, unreadCount, markAsRead, markAllAsRead, isStale } = useNotifications(); - const attentionCues = usePreferencesStore((s) => s.attentionCues); - const setAttentionCues = usePreferencesStore((s) => s.setAttentionCues); - const containerRef = useRef(null); - const buttonRef = useRef(null); - - useEffect(() => { - function handleKeyDown(e: KeyboardEvent) { - if (e.key === "Escape" && isOpen) { - setIsOpen(false); - buttonRef.current?.focus(); - } - } - - function handleClickOutside(e: MouseEvent) { - if (containerRef.current && !containerRef.current.contains(e.target as Node)) { - setIsOpen(false); - } - } - - document.addEventListener("keydown", handleKeyDown); - document.addEventListener("mousedown", handleClickOutside); - return () => { - document.removeEventListener("keydown", handleKeyDown); - document.removeEventListener("mousedown", handleClickOutside); - }; - }, [isOpen]); - - const renderItem = (item: RefractNotification) => { - const colors = SEVERITY_COLORS[item.severity]; - const icon = SEVERITY_ICONS[item.severity]; - - return ( -
-
-
- -

{item.title}

-
- {!item.read && ( - - )} -
-

{item.message}

- {item.href && ( -
- { - markAsRead(item.id); - setIsOpen(false); - }} - className="text-[11px] font-semibold text-pm-violet hover:underline" - > - View Details → - -
- )} -
- ); - }; - - return ( -
- - - {isOpen && ( -
-
-
- Notifications - {isStale && Saved} - {unreadCount > 0 && ( - - {unreadCount} new - - )} -
- {unreadCount > 0 && ( - - )} -
- - - -
- {notifications.length === 0 ? ( -
- - No notifications right now. You're all caught up! -
- ) : ( - notifications.length > VIRTUALIZE_THRESHOLD ? ( - n.id} - renderRow={renderItem} - label="Notifications" - /> - ) : ( - notifications.map(renderItem) - ) - )} -
-
- )} -
- ); -} diff --git a/src/components/layout/Navbar.tsx b/src/components/layout/Navbar.tsx index 5410296..e69de29 100644 --- a/src/components/layout/Navbar.tsx +++ b/src/components/layout/Navbar.tsx @@ -1,28 +0,0 @@ -"use client"; - -import Link from "next/link"; -import type { ReactNode } from "react"; - -export function Navbar({ right }: { right?: ReactNode }) { - return ( -
- -
- ); -} diff --git a/src/hooks/useClaims.ts b/src/hooks/useClaims.ts index 81b896f..e69de29 100644 --- a/src/hooks/useClaims.ts +++ b/src/hooks/useClaims.ts @@ -1,24 +0,0 @@ -"use client"; -import { useCallback, useMemo } from "react"; -import { fetchHolderClaims, fetchRecentClaims, type ClaimRecord } from "@/lib/api/claims"; -import type { Policy } from "@/lib/api/policies"; -import { ACTIVE_NETWORK } from "@/lib/network"; -import { useWallet } from "@/lib/wallet/WalletProvider"; -import { useRealtimeTopic } from "./useRealtimeTopic"; - -export type ClaimsResult = ClaimRecord[] & { data: ClaimRecord[] | null; loading: boolean; isStale: boolean }; - -export function useClaims(addressArg?: string | null, _policies?: Policy[] | null): ClaimsResult { - const wallet = useWallet(); - const address = addressArg === undefined - ? (wallet.status === "connected" ? wallet.address : null) - : addressArg; - const load = useCallback(async (signal: AbortSignal) => address - ? (await fetchHolderClaims(address, {}, signal)).items - : (await fetchRecentClaims({ limit: 50 }, signal)).items, [address]); - const topic = address ? `claims:${ACTIVE_NETWORK.id}:${address}` : "claim_settlement"; - const { data, isLoading, isStale } = useRealtimeTopic(topic, load, address ?? undefined); - return useMemo(() => Object.assign([...(data ?? [])], { - data: data ?? (address ? null : []), loading: isLoading, isStale, - }) as ClaimsResult, [data, address, isLoading, isStale]); -} 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/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 b54837b..e69de29 100644 --- a/src/hooks/useOracleStatus.ts +++ b/src/hooks/useOracleStatus.ts @@ -1,70 +0,0 @@ -"use client"; - -import { useEffect } from "react"; -import { useQuery, useQueryClient } from "@tanstack/react-query"; -import { fetchOracleStatus, type OracleReading } from "@/lib/api/oracle"; -import { FIXTURE_ORACLE_READINGS } from "@/lib/fixtures/oracle"; -import { queryFnWithFixtureFallback, type FixtureResult } from "@/lib/query/fixtureFallback"; -import { realtimeClient } from "@/lib/realtime/client"; - -export function useOracleStatus() { - const client = useQueryClient(); - const query = useQuery({ - queryKey: ["oracleStatus"], - queryFn: async ({ - signal, - }): Promise< - FixtureResult<{ readings: OracleReading[] }> & { source?: "api" | "fixture" | "realtime" } - > => { - const result = await queryFnWithFixtureFallback( - () => fetchOracleStatus(signal), - () => ({ readings: FIXTURE_ORACLE_READINGS }), - { signal, fallbackOnAnyError: true } - ); - return { ...result, source: result.isFixture ? "fixture" : "api" }; - }, - refetchInterval: 60_000, - }); - useEffect( - () => - realtimeClient.subscribe("oracle_update", (payload) => { - const reading = payload as OracleReading; - if (!reading || typeof reading !== "object" || typeof reading.coverageType !== "string") - return; - client.setQueryData( - ["oracleStatus"], - ( - current: - | { data: { readings: OracleReading[] }; isFixture: boolean; source?: string } - | undefined - ) => ({ - data: { - readings: [ - ...(current?.data.readings ?? []).filter( - (item) => item.coverageType !== reading.coverageType - ), - reading, - ], - }, - isFixture: false, - source: "realtime", - }) - ); - }), - [client] - ); - const result = query.data?.data; - return { - data: result?.readings ?? null, - loading: query.isLoading, - isFixture: query.data?.isFixture ?? false, - lastUpdated: query.dataUpdatedAt ? new Date(query.dataUpdatedAt) : undefined, - dataSource: query.data?.isFixture - ? ("fixture" as const) - : query.data?.source === "realtime" - ? ("realtime" as const) - : ("api" as const), - isRefreshing: query.isFetching, - refresh: query.refetch, - }; -} diff --git a/src/hooks/useRecentClaims.ts b/src/hooks/useRecentClaims.ts index f6fdd6d..e69de29 100644 --- a/src/hooks/useRecentClaims.ts +++ b/src/hooks/useRecentClaims.ts @@ -1,53 +0,0 @@ -"use client"; - -import { useEffect } from "react"; -import { useQuery, useQueryClient } from "@tanstack/react-query"; -import { fetchRecentClaims, type ClaimRecord } from "@/lib/api/claims"; -import { FIXTURE_RECENT_CLAIMS } from "@/lib/fixtures/recentClaims"; -import { queryFnWithFixtureFallback } from "@/lib/query/fixtureFallback"; -import { realtimeClient } from "@/lib/realtime/client"; - -export function useRecentClaims() { - const client = useQueryClient(); - const query = useQuery({ - queryKey: ["recentClaims"], - queryFn: ({ signal }) => - queryFnWithFixtureFallback( - () => fetchRecentClaims({ limit: 10 }, signal), - () => ({ items: FIXTURE_RECENT_CLAIMS, hasMore: false }), - { signal, fallbackOnAnyError: true } - ), - refetchInterval: 60_000, - }); - useEffect( - () => - realtimeClient.subscribe("claim_update", (payload) => { - const claim = payload as ClaimRecord; - if (!claim || typeof claim !== "object" || typeof claim.policyId !== "string") return; - client.setQueryData( - ["recentClaims"], - ( - current: - { data: { items: ClaimRecord[]; hasMore: boolean }; isFixture: boolean } | undefined - ) => ({ - data: { - items: [ - claim, - ...(current?.data.items ?? []).filter((item) => item.policyId !== claim.policyId), - ].slice(0, 10), - hasMore: current?.data.hasMore ?? false, - }, - isFixture: false, - }) - ); - }), - [client] - ); - return { - data: query.data?.data?.items ?? [], - loading: query.isLoading, - isFixture: query.data?.isFixture ?? false, - lastUpdated: query.dataUpdatedAt ? new Date(query.dataUpdatedAt) : undefined, - dataSource: query.data?.isFixture ? ("fixture" as const) : ("api" as const), - }; -} diff --git a/src/lib/api/client.ts b/src/lib/api/client.ts index 1d7dd3f..02af82c 100644 --- a/src/lib/api/client.ts +++ b/src/lib/api/client.ts @@ -11,7 +11,11 @@ import { env } from "@/lib/env"; export const API_BASE_URL = env.NEXT_PUBLIC_API_URL; 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"; } @@ -29,6 +33,7 @@ interface RequestOptions { method?: "GET" | "POST"; body?: unknown; signal?: AbortSignal; + idempotencyKey?: string; } export async function apiRequest(path: string, opts: RequestOptions = {}): Promise { @@ -36,7 +41,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 4aec10c..277087b 100644 --- a/src/lib/api/pool.ts +++ b/src/lib/api/pool.ts @@ -73,10 +73,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/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 index c776105..e69de29 100644 --- a/src/lib/realtime/client.test.ts +++ b/src/lib/realtime/client.test.ts @@ -1,32 +0,0 @@ -import { afterEach, describe, expect, it, vi } from "vitest"; -import { RealtimeClient } from "./client"; - -describe("RealtimeClient outbound batching", () => { - afterEach(() => vi.unstubAllGlobals()); - - it("writes twenty topic operations as one frame", () => { - const socket = { readyState: 1, send: vi.fn() }; - vi.stubGlobal("WebSocket", { OPEN: 1 }); - const client = new RealtimeClient() as any; - client.socket = socket; - for (let i = 0; i < 20; i++) client.enqueue(`topic-${i}`, "subscribe"); - client.flush(); - expect(socket.send).toHaveBeenCalledTimes(1); - expect(JSON.parse(socket.send.mock.calls[0]![0]).operations).toHaveLength(20); - }); - - it("elides subscribe then unsubscribe before the queued flush", () => { - const client = new RealtimeClient() as any; - client.enqueue("oracle_update", "subscribe"); - client.enqueue("oracle_update", "unsubscribe"); - expect(client.pending.size).toBe(0); - }); - - it("bounds queued topic operations and warns on overflow", () => { - const client = new RealtimeClient() as any; - const warning = vi.spyOn(console, "warn").mockImplementation(() => {}); - for (let i = 0; i < 501; i++) client.enqueue(`topic-${i}`, "subscribe"); - expect(client.pending.size).toBe(500); - expect(warning).toHaveBeenCalled(); - }); -}); 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; }