diff --git a/src/components/NotificationCenter.tsx b/src/components/NotificationCenter.tsx index f5a5658..8a1c99f 100644 --- a/src/components/NotificationCenter.tsx +++ b/src/components/NotificationCenter.tsx @@ -1,9 +1,10 @@ "use client"; -import { useEffect, useRef, useState } from "react"; +import { useEffect, useMemo, useRef, useState } from "react"; import Link from "next/link"; import { useNotifications } from "@/hooks/useNotifications"; import type { RefractNotification } from "@/lib/notifications"; +import { groupNotifications } from "@/lib/notificationGroups"; 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,9 @@ const SEVERITY_COLORS: Record>(new Set()); const { notifications, unreadCount, markAsRead, markAllAsRead } = useNotifications(); + const groups = useMemo(() => groupNotifications(notifications), [notifications]); const containerRef = useRef(null); const buttonRef = useRef(null); @@ -79,6 +85,10 @@ export function NotificationCenter() { )} + + {unreadCount > 0 ? `${unreadCount} unread notifications` : ""} + + {isOpen && (
{notifications.length === 0 ? (
- + No notifications right now. You're all caught up!
) : ( - notifications.map((item) => { - const colors = SEVERITY_COLORS[item.severity]; - const icon = SEVERITY_ICONS[item.severity]; - + groups.map((group) => { + const isCollapsed = collapsed.has(group.key); + const multi = group.items.length > 1; return ( -
-
-
-
- {!item.read && ( - - )} -
-

{item.message}

- {item.href && ( -
- { - markAsRead(item.id); - setIsOpen(false); - }} - className="text-[11px] font-semibold text-pm-violet hover:underline" - > - View Details โ†’ - -
+ {group.unread > 0 && ( + {group.unread} new + )} + )} -
+
+ {!isCollapsed && + group.items.map((item) => { + 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 โ†’ + +
+ )} +
+ ); + })} +
+ ); }) )} diff --git a/src/hooks/useNotifications.ts b/src/hooks/useNotifications.ts index e8fe606..e1aed80 100644 --- a/src/hooks/useNotifications.ts +++ b/src/hooks/useNotifications.ts @@ -1,6 +1,8 @@ "use client"; import { useEffect, useMemo, useState } from "react"; +import { realtimeClient } from "@/lib/realtime/client"; +import { mergeNotifications } from "@/lib/notificationGroups"; import { useWallet } from "@/lib/wallet/WalletProvider"; import { useHolderPolicies } from "./useHolderPolicies"; import { useClaims } from "./useClaims"; @@ -23,6 +25,31 @@ export function useNotifications() { const { data: oracleReadings } = useOracleStatus(); const [readIds, setReadIds] = useState>(new Set()); + const [pushed, setPushed] = useState([]); + + // Realtime delivers new items; REST-derived items below supply history. + // If the topic is disabled or unhealthy nothing arrives and REST still works. + useEffect(() => { + return realtimeClient.subscribe("notifications", (payload) => { + const item = payload as Partial | null; + if (!item || typeof item.id !== "string" || typeof item.timestamp !== "number") return; + setPushed((prev) => mergeNotifications([item as RefractNotification], prev)); + }); + }, []); + + // Reflect read state changes made in other tabs. + useEffect(() => { + function onStorage(e: StorageEvent) { + if (e.key !== STORAGE_KEY || !e.newValue) return; + try { + setReadIds(new Set(JSON.parse(e.newValue))); + } catch { + // Ignore malformed value + } + } + window.addEventListener("storage", onStorage); + return () => window.removeEventListener("storage", onStorage); + }, []); useEffect(() => { try { @@ -45,14 +72,18 @@ export function useNotifications() { }; const notifications = useMemo(() => { - return deriveNotifications({ + const derived = deriveNotifications({ policies, claims, lockupExpiresAt, oracleReadings, readIds, }); - }, [policies, claims, lockupExpiresAt, oracleReadings, readIds]); + return mergeNotifications( + pushed.map((n) => ({ ...n, read: readIds.has(n.id) })), + derived + ); + }, [policies, claims, lockupExpiresAt, oracleReadings, readIds, pushed]); const unreadCount = useMemo(() => { return notifications.filter((n) => !n.read).length; diff --git a/src/hooks/useRealtimeRevalidation.ts b/src/hooks/useRealtimeRevalidation.ts new file mode 100644 index 0000000..199ea13 --- /dev/null +++ b/src/hooks/useRealtimeRevalidation.ts @@ -0,0 +1,35 @@ +"use client"; + +import { useEffect } from "react"; +import { useRouter } from "next/navigation"; +import { realtimeClient } from "@/lib/realtime/client"; +import { createRefreshScheduler, TOPIC_REVALIDATION_TAGS } from "@/lib/realtime/revalidation"; + +/** + * Refreshes the current route's Server Components when realtime topics that + * feed them push events. router.refresh() preserves client state, scroll and + * subscriptions. Mount once near the app root. + */ +export function useRealtimeRevalidation({ + enabled = true, + isBlocked, +}: { enabled?: boolean; isBlocked?: () => boolean } = {}) { + const router = useRouter(); + + useEffect(() => { + if (typeof window === "undefined") return; + const scheduler = createRefreshScheduler({ + refresh: () => router.refresh(), + enabled: () => enabled, + isBlocked, + log: (m) => console.debug(m), + }); + const unsubs = Object.keys(TOPIC_REVALIDATION_TAGS).map((topic) => + realtimeClient.subscribe(topic, () => scheduler.notify(topic)) + ); + return () => { + unsubs.forEach((u) => u()); + scheduler.cancel(); + }; + }, [router, enabled, isBlocked]); +} diff --git a/src/lib/notificationGroups.test.ts b/src/lib/notificationGroups.test.ts new file mode 100644 index 0000000..224657b --- /dev/null +++ b/src/lib/notificationGroups.test.ts @@ -0,0 +1,36 @@ +import { describe, it, expect } from "vitest"; +import { groupNotifications, mergeNotifications } from "./notificationGroups"; +import type { RefractNotification } from "./notifications"; + +const n = ( + id: string, + timestamp: number, + extra: Partial = {} +): RefractNotification => ({ + id, + type: "claim_payout", + title: id, + message: "", + timestamp, + read: false, + severity: "info", + ...extra, +}); + +describe("notification groups", () => { + it("dedupes by id and orders by timestamp", () => { + const merged = mergeNotifications([n("a", 1), n("b", 3)], [n("a", 1), n("c", 2)]); + expect(merged.map((x) => x.id)).toEqual(["b", "c", "a"]); + }); + + it("groups by key with unread counts, newest group first", () => { + const groups = groupNotifications([ + n("1", 1, { groupKey: "claim:x" }), + n("2", 5, { groupKey: "claim:x", read: true }), + n("3", 9, { groupKey: "policy:y" }), + ]); + expect(groups.map((g) => g.key)).toEqual(["policy:y", "claim:x"]); + expect(groups[1]).toMatchObject({ unread: 1, latest: 5 }); + expect(groups[1].items.map((x) => x.id)).toEqual(["2", "1"]); + }); +}); diff --git a/src/lib/notificationGroups.ts b/src/lib/notificationGroups.ts new file mode 100644 index 0000000..1cfee2b --- /dev/null +++ b/src/lib/notificationGroups.ts @@ -0,0 +1,30 @@ +import type { RefractNotification } from "./notifications"; + +export interface NotificationGroup { + key: string; + items: RefractNotification[]; + unread: number; + latest: number; +} + +/** Merges two copies of the notification lists, deduplicating by id (first wins). */ +export function mergeNotifications(...lists: RefractNotification[][]): RefractNotification[] { + const byId = new Map(); + for (const list of lists) for (const n of list) if (!byId.has(n.id)) byId.set(n.id, n); + return [...byId.values()].sort((a, b) => b.timestamp - a.timestamp); +} + +/** Groups by `groupKey` (falling back to type); groups ordered by latest item, items by timestamp desc. */ +export function groupNotifications(items: RefractNotification[]): NotificationGroup[] { + const groups = new Map(); + for (const n of items) { + const key = n.groupKey ?? n.type; + const g = groups.get(key) ?? { key, items: [], unread: 0, latest: 0 }; + g.items.push(n); + if (!n.read) g.unread += 1; + g.latest = Math.max(g.latest, n.timestamp); + groups.set(key, g); + } + for (const g of groups.values()) g.items.sort((a, b) => b.timestamp - a.timestamp); + return [...groups.values()].sort((a, b) => b.latest - a.latest); +} diff --git a/src/lib/notifications.ts b/src/lib/notifications.ts index 8ccaf83..573eafd 100644 --- a/src/lib/notifications.ts +++ b/src/lib/notifications.ts @@ -18,6 +18,8 @@ export interface RefractNotification { timestamp: number; read: boolean; href?: string; + /** Groups related events (e.g. per policy or claim); defaults to type. */ + groupKey?: string; severity: "info" | "warning" | "success" | "critical"; } @@ -55,6 +57,7 @@ export function deriveNotifications({ const id = `policy-exp-${policy.id}`; notifications.push({ id, + groupKey: `policy:${policy.id}`, type: "policy_expiring", title: `Policy Expiring Soon`, message: `Your ${policy.coverageTypeName} policy (${formatUsd( @@ -77,6 +80,7 @@ export function deriveNotifications({ const payoutFormatted = formatUsd(fromStroops(claim.payout)); notifications.push({ id, + groupKey: `claim:${claim.policyId}`, type: "claim_payout", title: `Claim Payout Disbursed`, message: `${payoutFormatted} was automatically disbursed for policy ${claim.policyId.slice(0, 8)}โ€ฆ (${claim.reason}).`, diff --git a/src/lib/realtime/__tests__/parity.test.ts b/src/lib/realtime/__tests__/parity.test.ts new file mode 100644 index 0000000..a5f055d --- /dev/null +++ b/src/lib/realtime/__tests__/parity.test.ts @@ -0,0 +1,167 @@ +import { describe, it, expect } from "vitest"; +import { createTopicConsumer, type TopicFrame } from "../patch"; + +/** + * Transport parity matrix: the same scenarios run against every mock + * transport mode and assert only the consumer-facing contract (emitted data, + * resync requests), never transport internals. + */ +type Mode = "websocket" | "sse" | "long-poll"; +const MODES: Mode[] = ["websocket", "sse", "long-poll"]; + +/** Intentional divergence: long-poll buffers frames and delivers them in batches on poll. */ +const DIVERGENCES: Record = { + websocket: "frames delivered immediately", + sse: "frames delivered immediately (framed as events)", + "long-poll": "frames buffered server-side and delivered in batches on next poll", +}; + +function createMockTransport(mode: Mode) { + const serverBuffer: unknown[] = []; + let sink: ((f: unknown) => void) | null = null; + let closed = false; + let authed = true; + const resyncs: number[] = []; + const wire = (f: unknown) => (mode === "sse" ? JSON.parse(`${JSON.stringify(f)}`) : f); + return { + mode, + connect(onFrame: (f: unknown) => void) { + if (!authed) throw new Error("auth_failed"); + sink = onFrame; + closed = false; + }, + send(f: unknown) { + if (closed) return; + if (mode === "long-poll") serverBuffer.push(wire(f)); + else sink?.(wire(f)); + }, + /** Long-poll delivery tick; a no-op for streaming transports. */ + poll() { + while (serverBuffer.length) sink?.(serverBuffer.shift()); + }, + flush() { + this.poll(); + }, + close() { + closed = true; + sink = null; + }, + failAuth() { + authed = false; + }, + requestResync() { + resyncs.push(Date.now()); + }, + resyncs, + }; +} + +const snap = (revision: number, data: Record): TopicFrame => ({ + kind: "snapshot", + revision, + data, +}); +const patch = (b: number, r: number, set: Record): TopicFrame => ({ + kind: "patch", + base_revision: b, + revision: r, + set, +}); + +const executed: Record = { websocket: 0, sse: 0, "long-poll": 0 }; + +describe.each(MODES)("transport parity: %s", (mode) => { + const setup = () => { + const t = createMockTransport(mode); + const seen: Record[] = []; + const consumer = createTopicConsumer( + (d) => seen.push(d), + () => t.requestResync() + ); + t.connect((f) => consumer.push(f)); + return { t, seen, consumer }; + }; + + it("subscribe + receive + ordering", () => { + executed[mode]++; + const { t, seen } = setup(); + t.send(snap(1, { v: 1 })); + t.send(patch(1, 2, { v: 2 })); + t.send(patch(2, 3, { v: 3 })); + t.flush(); + expect(seen.at(-1)).toEqual({ v: 3 }); + expect(seen.map((s) => s.v)).toEqual([1, 2, 3]); + }); + + it("gap triggers resync and recovers on snapshot", () => { + executed[mode]++; + const { t, seen } = setup(); + t.send(snap(1, { v: 1 })); + t.send(patch(5, 6, { v: 6 })); + t.flush(); + expect(t.resyncs).toHaveLength(1); + t.send(snap(7, { v: 7 })); + t.flush(); + expect(seen.at(-1)).toEqual({ v: 7 }); + }); + + it("duplicate frames are idempotent", () => { + executed[mode]++; + const { t, seen } = setup(); + t.send(snap(1, { v: 1 })); + t.send(patch(1, 2, { v: 2 })); + t.send(patch(1, 2, { v: 2 })); + t.flush(); + expect(seen.map((s) => s.v)).toEqual([1, 2]); + expect(t.resyncs).toHaveLength(0); + }); + + it("reconnect requires fresh base: patch before snapshot requests resync", () => { + executed[mode]++; + const { t, seen, consumer } = setup(); + t.send(snap(1, { v: 1 })); + t.flush(); + t.close(); + consumer.reset(); + t.connect((f) => consumer.push(f)); + t.send(patch(1, 2, { v: 2 })); + t.flush(); + expect(t.resyncs).toHaveLength(1); + expect(seen).toHaveLength(1); + }); + + it("auth failure surfaces the same error", () => { + executed[mode]++; + const t = createMockTransport(mode); + t.failAuth(); + expect(() => t.connect(() => {})).toThrow("auth_failed"); + }); + + it("server close and teardown deliver nothing further", () => { + executed[mode]++; + const { t, seen } = setup(); + t.send(snap(1, { v: 1 })); + t.flush(); + t.close(); + t.send(snap(2, { v: 2 })); + t.flush(); + expect(seen).toHaveLength(1); + }); + + it("documented divergence is explicit: long-poll delays delivery until poll", () => { + executed[mode]++; + const { t, seen } = setup(); + t.send(snap(1, { v: 1 })); + expect(seen).toHaveLength(mode === "long-poll" ? 0 : 1); + t.poll(); + expect(seen).toHaveLength(1); + expect(DIVERGENCES[mode]).toBeTruthy(); + }); +}); + +describe("matrix guard", () => { + it("every transport executed scenarios", () => { + for (const mode of MODES) + expect(executed[mode], `${mode} ran zero scenarios`).toBeGreaterThan(0); + }); +}); diff --git a/src/lib/realtime/__tests__/patch.test.ts b/src/lib/realtime/__tests__/patch.test.ts new file mode 100644 index 0000000..9ffd115 --- /dev/null +++ b/src/lib/realtime/__tests__/patch.test.ts @@ -0,0 +1,83 @@ +import { describe, it, expect } from "vitest"; +import { + applyFrame, + applyPatch, + isRejection, + isTopicFrame, + type PatchFrame, + type TopicFrame, +} from "../patch"; + +const snap = (revision: number, data: Record): TopicFrame => ({ + kind: "snapshot", + revision, + data, +}); +const patch = ( + base_revision: number, + revision: number, + set: Record +): PatchFrame => ({ + kind: "patch", + base_revision, + revision, + set, +}); + +describe("applyPatch", () => { + it("sequence of patches equals the corresponding snapshots", () => { + const snaps = [ + snap(1, { a: 1, nested: { x: 1, y: 2 } }), + snap(2, { a: 2, nested: { x: 1, y: 2 } }), + snap(3, { a: 2, nested: { x: 1, y: 5 }, b: 9 }), + ]; + const patches = [patch(1, 2, { a: 2 }), patch(2, 3, { nested: { y: 5 }, b: 9 })]; + let state = applyFrame(null, snaps[0]); + for (const p of patches) { + if (isRejection(state)) throw new Error("unexpected rejection"); + state = applyFrame(state, p); + } + expect(state).toEqual({ revision: 3, data: (snaps[2] as { data: unknown }).data }); + }); + + it("removes keys set to null", () => { + const s = { revision: 1, data: { a: 1, b: 2 } }; + expect(applyPatch(s, patch(1, 2, { b: null }))).toEqual({ revision: 2, data: { a: 1 } }); + }); + + it("requests resync on revision gap or missing base", () => { + const s = { revision: 1, data: { a: 1 } }; + expect(applyPatch(s, patch(3, 4, { a: 2 }))).toMatchObject({ + reason: "revision_gap", + resync: true, + }); + expect(applyPatch(null, patch(1, 2, {}))).toMatchObject({ reason: "no_base", resync: true }); + }); + + it("ignores duplicate and out-of-order patches without resync", () => { + const s = { revision: 5, data: { a: 1 } }; + expect(applyPatch(s, patch(4, 5, { a: 2 }))).toMatchObject({ reason: "stale", resync: false }); + expect(applyPatch(s, patch(2, 3, { a: 2 }))).toMatchObject({ reason: "stale", resync: false }); + }); + + it("rejects malformed patches and does not mutate state", () => { + const s = { revision: 1, data: { a: { b: 1 } } }; + const before = JSON.stringify(s); + expect( + applyPatch(s, { kind: "patch", base_revision: 1, revision: 1, set: {} } as PatchFrame) + ).toMatchObject({ reason: "malformed" }); + applyPatch(s, patch(1, 2, { a: { b: 2 } })); + expect(JSON.stringify(s)).toBe(before); + }); + + it("discards snapshots older than current but accepts newer at any time", () => { + const s = { revision: 5, data: { a: 1 } }; + expect(applyFrame(s, snap(4, { a: 0 }))).toMatchObject({ reason: "stale" }); + expect(applyFrame(s, snap(9, { a: 9 }))).toEqual({ revision: 9, data: { a: 9 } }); + }); + + it("validates frames", () => { + expect(isTopicFrame({ kind: "x" })).toBe(false); + expect(isTopicFrame(snap(1, {}))).toBe(true); + }); +}); diff --git a/src/lib/realtime/__tests__/revalidation.test.ts b/src/lib/realtime/__tests__/revalidation.test.ts new file mode 100644 index 0000000..0637f5f --- /dev/null +++ b/src/lib/realtime/__tests__/revalidation.test.ts @@ -0,0 +1,46 @@ +import { describe, it, expect, vi, beforeEach, afterEach } from "vitest"; +import { createRefreshScheduler, tagsForTopic } from "../revalidation"; + +describe("revalidation", () => { + beforeEach(() => vi.useFakeTimers()); + afterEach(() => vi.useRealTimers()); + + it("maps server-derived topics to tags and ignores client-only topics", () => { + expect(tagsForTopic("oracle_update")).toContain("oracle"); + expect(tagsForTopic("typing_indicator")).toEqual([]); + }); + + it("coalesces a burst into exactly one refresh", () => { + const refresh = vi.fn(); + const s = createRefreshScheduler({ refresh, windowMs: 1000 }); + for (let i = 0; i < 20; i++) s.notify("oracle_update"); + s.notify("claim_update"); + vi.advanceTimersByTime(1000); + expect(refresh).toHaveBeenCalledTimes(1); + expect(refresh.mock.calls[0][0]).toEqual(expect.arrayContaining(["oracle", "claims"])); + }); + + it("never refreshes for client-only topics or when disabled", () => { + const refresh = vi.fn(); + const log = vi.fn(); + createRefreshScheduler({ refresh, windowMs: 10 }).notify("other"); + createRefreshScheduler({ refresh, windowMs: 10, enabled: () => false, log }).notify( + "oracle_update" + ); + vi.advanceTimersByTime(100); + expect(refresh).not.toHaveBeenCalled(); + expect(log).toHaveBeenCalled(); + }); + + it("defers while blocked then refreshes once", () => { + const refresh = vi.fn(); + let blocked = true; + const s = createRefreshScheduler({ refresh, windowMs: 100, isBlocked: () => blocked }); + s.notify("pool_stats"); + vi.advanceTimersByTime(250); + expect(refresh).not.toHaveBeenCalled(); + blocked = false; + vi.advanceTimersByTime(100); + expect(refresh).toHaveBeenCalledTimes(1); + }); +}); diff --git a/src/lib/realtime/client.ts b/src/lib/realtime/client.ts index 8be8d3c..c4683b2 100644 --- a/src/lib/realtime/client.ts +++ b/src/lib/realtime/client.ts @@ -1,4 +1,5 @@ import { API_BASE_URL } from "@/lib/api/client"; +import { applyFrame, isRejection, isTopicFrame, type TopicState } from "./patch"; type Handler = (payload: unknown) => void; @@ -19,6 +20,7 @@ class RealtimeClient { private handlers = new Map>(); private attempt = 0; private closedByUser = false; + private topicState = new Map(); private reconnectTimer: ReturnType | null = null; connect(): void { @@ -41,14 +43,18 @@ class RealtimeClient { }); 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)); + const { topic, payload } = JSON.parse(event.data as string) as { + topic: string; + payload: unknown; + }; + this.dispatch(topic, payload); } catch { // Malformed message โ€” ignore, don't crash the connection. } }); this.socket.addEventListener("close", () => { this.socket = null; + this.topicState.clear(); if (!this.closedByUser) this.scheduleReconnect(); }); this.socket.addEventListener("error", () => { @@ -56,6 +62,28 @@ class RealtimeClient { }); } + /** + * Snapshot/patch frames (contract v2) are folded into per-topic state and + * handlers receive the resulting data; any other payload is passed through + * unchanged so topics can opt in to deltas independently. + */ + private dispatch(topic: string, payload: unknown): void { + let out = payload; + if (isTopicFrame(payload)) { + const result = applyFrame(this.topicState.get(topic) ?? null, payload); + if (isRejection(result)) { + if (result.resync && this.socket?.readyState === WebSocket.OPEN) { + this.topicState.delete(topic); + this.socket.send(JSON.stringify({ type: "resync", topic })); + } + return; + } + this.topicState.set(topic, result); + out = result.data; + } + this.handlers.get(topic)?.forEach((handler) => handler(out)); + } + private scheduleReconnect(): void { if (this.reconnectTimer) return; const delay = Math.min(BASE_BACKOFF_MS * 2 ** this.attempt, MAX_BACKOFF_MS); diff --git a/src/lib/realtime/patch.ts b/src/lib/realtime/patch.ts new file mode 100644 index 0000000..4287349 --- /dev/null +++ b/src/lib/realtime/patch.ts @@ -0,0 +1,127 @@ +/** + * Optional delta/patch frames for realtime topics. Full snapshots stay valid + * at any time and supersede pending patches; patches carry a base revision so + * the applier can reject gaps instead of corrupting state. + * + * Wire format uses keyed maps (never positional arrays). A `null` value in + * `set` removes the key; nested plain objects are merged recursively. + */ + +export const REALTIME_CONTRACT_VERSION = 2; + +export interface SnapshotFrame> { + kind: "snapshot"; + revision: number; + data: T; +} + +export interface PatchFrame { + kind: "patch"; + base_revision: number; + revision: number; + set: Record; +} + +export type TopicFrame = SnapshotFrame | PatchFrame; + +export interface TopicState> { + revision: number; + data: T; +} + +export type PatchRejection = { + rejected: true; + reason: "no_base" | "revision_gap" | "stale" | "malformed"; + /** True when the caller should request a full snapshot resync. */ + resync: boolean; +}; + +const isPlainObject = (v: unknown): v is Record => + typeof v === "object" && v !== null && !Array.isArray(v); + +export function isTopicFrame(value: unknown): value is TopicFrame { + if (!isPlainObject(value)) return false; + if (value.kind === "snapshot") { + return Number.isInteger(value.revision) && isPlainObject(value.data); + } + if (value.kind === "patch") { + return ( + Number.isInteger(value.base_revision) && + Number.isInteger(value.revision) && + (value.revision as number) > (value.base_revision as number) && + isPlainObject(value.set) + ); + } + return false; +} + +function mergeInto( + target: Record, + set: Record +): Record { + const out = { ...target }; + for (const [key, value] of Object.entries(set)) { + if (value === null) delete out[key]; + else if (isPlainObject(value) && isPlainObject(out[key])) + out[key] = mergeInto(out[key] as Record, value); + else out[key] = value; + } + return out; +} + +/** Pure patch applier: never mutates `state`, returns a new state or a rejection. */ +export function applyPatch>( + state: TopicState | null, + patch: PatchFrame +): TopicState | PatchRejection { + if (!isTopicFrame(patch) || patch.kind !== "patch") { + return { rejected: true, reason: "malformed", resync: false }; + } + if (!state) return { rejected: true, reason: "no_base", resync: true }; + if (patch.revision <= state.revision) return { rejected: true, reason: "stale", resync: false }; + if (patch.base_revision !== state.revision) + return { rejected: true, reason: "revision_gap", resync: true }; + return { revision: patch.revision, data: mergeInto(state.data, patch.set) as T }; +} + +/** Applies either frame form; snapshots with a lower revision than current are discarded. */ +export function applyFrame>( + state: TopicState | null, + frame: TopicFrame +): TopicState | PatchRejection { + if (frame.kind === "snapshot") { + if (state && frame.revision < state.revision) + return { rejected: true, reason: "stale", resync: false }; + return { revision: frame.revision, data: frame.data as T }; + } + return applyPatch(state, frame); +} + +export const isRejection = (v: TopicState | PatchRejection): v is PatchRejection => + (v as PatchRejection).rejected === true; + +/** Consumer-facing state holder: folds frames, emits data, signals resync. */ +export function createTopicConsumer( + onData: (data: Record) => void, + onResync: () => void +) { + let state: TopicState | null = null; + return { + push(frame: unknown): void { + if (!isTopicFrame(frame)) return; + const result = applyFrame(state, frame); + if (isRejection(result)) { + if (result.resync) { + state = null; + onResync(); + } + return; + } + state = result; + onData(result.data); + }, + reset(): void { + state = null; + }, + }; +} diff --git a/src/lib/realtime/revalidation.ts b/src/lib/realtime/revalidation.ts new file mode 100644 index 0000000..cb196e4 --- /dev/null +++ b/src/lib/realtime/revalidation.ts @@ -0,0 +1,67 @@ +/** + * Single source of truth mapping realtime topics to the Next.js cache tags + * their events invalidate. Topics absent from the map only affect client + * state and never trigger a refresh. + */ +export const TOPIC_REVALIDATION_TAGS: Readonly> = { + oracle_update: ["oracle", "pool-stats"], + pool_stats: ["pool-stats"], + claim_update: ["claims", "policies"], + policy_update: ["policies"], +}; + +export const REFRESH_WINDOW_MS = 2_000; + +export const tagsForTopic = (topic: string): readonly string[] => + TOPIC_REVALIDATION_TAGS[topic] ?? []; + +export interface RefreshSchedulerOptions { + refresh: (tags: string[]) => void; + windowMs?: number; + /** Return true to defer (e.g. in-flight navigation or purchase confirm step). */ + isBlocked?: () => boolean; + enabled?: () => boolean; + log?: (msg: string) => void; +} + +/** Coalesces bursts of topic events into at most one refresh per window. */ +export function createRefreshScheduler({ + refresh, + windowMs = REFRESH_WINDOW_MS, + isBlocked = () => false, + enabled = () => true, + log = () => {}, +}: RefreshSchedulerOptions) { + const pending = new Set(); + let timer: ReturnType | null = null; + + const flush = () => { + timer = null; + if (pending.size === 0) return; + if (isBlocked()) { + timer = setTimeout(flush, windowMs); + return; + } + const tags = [...pending]; + pending.clear(); + refresh(tags); + }; + + return { + notify(topic: string): void { + const tags = tagsForTopic(topic); + if (tags.length === 0) return; + if (!enabled()) { + log(`realtime revalidation disabled; ignoring "${topic}"`); + return; + } + tags.forEach((t) => pending.add(t)); + if (!timer) timer = setTimeout(flush, windowMs); + }, + cancel(): void { + if (timer) clearTimeout(timer); + timer = null; + pending.clear(); + }, + }; +}