diff --git a/desktop/src/features/agents/lib/personaCatalogRelay.test.mjs b/desktop/src/features/agents/lib/personaCatalogRelay.test.mjs index fbaf1f5274..f3727598c0 100644 --- a/desktop/src/features/agents/lib/personaCatalogRelay.test.mjs +++ b/desktop/src/features/agents/lib/personaCatalogRelay.test.mjs @@ -356,7 +356,7 @@ test("test_foreign_entry_with_no_local_copy_stays_unselected", () => { BOB, ); - assert.equal(personas[0].id, "catalog:" + ALICE + ":reviewer"); + assert.equal(personas[0].id, `catalog:${ALICE}:reviewer`); assert.equal(personas[0].isActive, false); }); @@ -377,7 +377,7 @@ test("test_catalog_source_match_is_scoped_to_the_publishing_owner", () => { ALICE, ); - assert.equal(personas[0].id, "catalog:" + BOB + ":reviewer"); + assert.equal(personas[0].id, `catalog:${BOB}:reviewer`); assert.equal(personas[0].isActive, false); }); diff --git a/desktop/src/features/channels/nip-rs-unread-ui.test.mjs b/desktop/src/features/channels/nip-rs-unread-ui.test.mjs index badb6c8816..2ffea2f4f8 100644 --- a/desktop/src/features/channels/nip-rs-unread-ui.test.mjs +++ b/desktop/src/features/channels/nip-rs-unread-ui.test.mjs @@ -706,8 +706,8 @@ test("preReady_unread_witness_c_pendingUnread_visible_during_loading", () => { // committed forced entry; committed forced with no pending is visible. // Exercises the ratified precedence rule via the production computePreReadyUnread helper. const { restore } = makeIsolatedStorage(); - const pubkey = "cc".repeat(32); - pendingOverrideIntentStore.loadForPubkey(pubkey); + const _pubkey = "cc".repeat(32); + pendingOverrideIntentStore.restoreFromStorage(new Map(), 1); pendingOverrideIntentStore.enqueue("ch-unread", "unread"); pendingOverrideIntentStore.enqueue("ch-read", "read"); diff --git a/desktop/src/features/channels/pendingOverrideIntents.ts b/desktop/src/features/channels/pendingOverrideIntents.ts index 56a9128ee1..cdf6145a72 100644 --- a/desktop/src/features/channels/pendingOverrideIntents.ts +++ b/desktop/src/features/channels/pendingOverrideIntents.ts @@ -3,16 +3,19 @@ * * When the full-state load is incomplete at click time, override actions cannot * immediately mutate the register. Instead the user's intent is recorded here - * as a pending entry keyed by (pubkey, channelId). A drain on the next - * complete-load transition replays each intent through the normal action rules - * and surfaces the definitive applied/refused result. + * as a pending entry keyed by channelId. A drain on the next complete-load + * transition replays each intent through the normal action rules and surfaces + * the definitive applied/refused result. * * Design notes * ───────────── - * • **Separate store, deterministic write ordering** (plan ruling 2): - * `forcedUnreadStore` entries (the optimistic local badge) are written - * BEFORE the intent. A crash between the two leaves bold with no queued - * bump — pre-#3966 per-device behaviour, always safe. + * • **In-memory only** (consolidation ruling): + * This store holds in-memory state only. Persistence is exclusively through + * the `buzz.nip-rs.override-state.v2:` blob written by + * `writeStoredReadState()` in readStateStorage.ts. The manager's + * `persistLocalState()` is the single commit point for both the intent queue + * and the register+receipt. The former `buzz.nip-rs.pending-intents.v1` + * key is never written and is not read after migration. * * • **Per-channel, last-intent-wins** (plan ruling 3): * A re-mark while a drain is in flight replaces the queue entry and advances @@ -20,25 +23,15 @@ * and commits zero local effects for the stale intent. * * • **Applied receipt** (plan Amendment C): - * The receipt `{intentGen, op}` is NOT stored here — it lives inside the - * `buzz.nip-rs.override-state.v2:` blob alongside the mutated - * register (`readStateStorage.ts`) so register mutation and receipt are - * never observable separately. This store only holds the intent queue and - * exposes the generation counter. - * - * • **Storage failure** (plan ruling 2): - * If the localStorage write fails, the intent is kept in-memory and - * classified `storage_failed` internally. Optimistic presentation and - * same-session drain still work; a crash before drain degrades to - * per-device behaviour. + * The receipt `{intentGen, op}` lives inside the v2 blob alongside the + * mutated register so register mutation and receipt are never observable + * separately. This store only holds the in-memory intent queue. * * • **Identity swap** (plan phase 1): - * `loadForPubkey` replaces the in-memory map with the new pubkey's persisted - * data. Old pubkey data is NOT wiped from localStorage. + * `restoreFromStorage` replaces the in-memory map with the stored data. + * Old pubkey data is NOT wiped from the v2 blob — the manager owns that. */ -import { isPlainRecord } from "@/features/channels/readState/readStateFormat"; - // ── Types ───────────────────────────────────────────────────────────────────── /** The operation the user intended. */ @@ -53,156 +46,77 @@ export type PendingIntentOp = "unread" | "read"; * `sourceScope` — for `read` intents, the exact source (or undefined = all * sources) whose removal scope was captured at click time. * Used by the drain to apply source cleanup idempotently. + * `readTarget` — for `read` intents, the frontier value captured at click + * time. Drain advances the frontier to this value before the + * C-bump so that an incomplete-load read later applies at the + * correct logical position. */ export type PendingIntent = { gen: number; op: PendingIntentOp; sourceScope?: string; + readTarget?: number; }; -/** In-memory + persisted state for one pubkey. */ -type PendingIntentRecord = { - /** Next generation to assign when a new intent is enqueued. */ - nextGen: number; - /** Per-channel intent. At most one intent per channel at any time. */ - intents: Map; - /** True when the last localStorage write failed. */ - storageFailed: boolean; -}; - -// ── Storage keys ────────────────────────────────────────────────────────────── - -const STORAGE_PREFIX = "buzz.nip-rs.pending-intents.v1"; -const storageKey = (pubkey: string) => `${STORAGE_PREFIX}:${pubkey}`; - -// ── Serialisation ───────────────────────────────────────────────────────────── - -type PersistedIntentEntry = { - gen: number; - op: PendingIntentOp; - sourceScope?: string; -}; - -type PersistedRecord = { - nextGen: number; - intents: Record; -}; - -function serialise(record: PendingIntentRecord): string { - const intents: Record = {}; - for (const [channelId, intent] of record.intents) { - const entry: PersistedIntentEntry = { gen: intent.gen, op: intent.op }; - if (intent.sourceScope !== undefined) - entry.sourceScope = intent.sourceScope; - intents[channelId] = entry; - } - return JSON.stringify({ nextGen: record.nextGen, intents }); -} - -function deserialise(pubkey: string): PendingIntentRecord { - const empty: PendingIntentRecord = { - nextGen: 1, - intents: new Map(), - storageFailed: false, - }; - try { - const raw = localStorage.getItem(storageKey(pubkey)); - if (!raw) return empty; - const parsed: unknown = JSON.parse(raw); - if (!isPlainRecord(parsed)) return empty; - const { nextGen, intents } = parsed as Partial; - if ( - typeof nextGen !== "number" || - !Number.isInteger(nextGen) || - nextGen < 1 - ) - return empty; - if (!isPlainRecord(intents)) return empty; - const map = new Map(); - for (const [channelId, entry] of Object.entries(intents)) { - if (!isPlainRecord(entry)) continue; - const { gen, op, sourceScope } = entry as Partial; - if (typeof gen !== "number" || !Number.isInteger(gen) || gen < 1) - continue; - if (op !== "unread" && op !== "read") continue; - if (sourceScope !== undefined && typeof sourceScope !== "string") - continue; - map.set(channelId, { - gen, - op, - ...(sourceScope !== undefined ? { sourceScope } : {}), - }); - } - return { nextGen, intents: map, storageFailed: false }; - } catch { - return empty; - } -} - -function persist(pubkey: string, record: PendingIntentRecord): void { - try { - localStorage.setItem(storageKey(pubkey), serialise(record)); - record.storageFailed = false; - } catch { - record.storageFailed = true; - } -} - // ── Store ───────────────────────────────────────────────────────────────────── /** - * Pending intent store — one instance per session, keyed by pubkey. + * Pending intent store — one instance per session, pure in-memory. * * All mutations are synchronous and serialised through this object. The store * owner is the single authority for generation ordering (plan ruling 3: * enqueue-during-drain serialises through the store owner). + * + * Persistence is exclusively through the manager's `persistLocalState()` which + * serialises the in-memory map into the v2 blob's `pendingIntents` sub-object. */ export class PendingOverrideIntentStore { - private pubkey: string | null = null; - private record: PendingIntentRecord = { - nextGen: 1, - intents: new Map(), - storageFailed: false, - }; + private _nextGen = 1; + private intents = new Map(); /** - * Load or swap to the given pubkey's persisted intents. - * Called on identity initialisation and identity change. + * Restore in-memory state from deserialized storage (called by manager + * during `hydrateFromLocalStorage`). Replaces any in-memory state. */ - loadForPubkey(pubkey: string): void { - this.pubkey = pubkey; - this.record = deserialise(pubkey); + restoreFromStorage( + intents: ReadonlyMap, + nextGen: number, + ): void { + this.intents = new Map(intents); + this._nextGen = nextGen; } /** * Enqueue a new intent for `channelId`, replacing any existing one. * * Returns the enqueued intent (including the assigned generation). - * The generation is monotonically increasing per-channel (and - * per-pubkey-session), making every intent distinguishable from predecessors. + * The generation is monotonically increasing per-channel-session, making + * every intent distinguishable from predecessors. + * + * Does NOT persist — the caller (drain or manager) calls persistLocalState(). */ enqueue( channelId: string, op: PendingIntentOp, sourceScope?: string, + readTarget?: number, ): PendingIntent { - const gen = this.record.nextGen++; + const gen = this._nextGen++; const intent: PendingIntent = { gen, op, ...(sourceScope !== undefined ? { sourceScope } : {}), + ...(readTarget !== undefined ? { readTarget } : {}), }; - this.record.intents.set(channelId, intent); - if (this.pubkey) persist(this.pubkey, this.record); + this.intents.set(channelId, intent); return intent; } /** * Get the current intent for `channelId`, or undefined if none exists. - * Called from the drain capture step and from hydration. */ get(channelId: string): PendingIntent | undefined { - return this.record.intents.get(channelId); + return this.intents.get(channelId); } /** @@ -210,41 +124,38 @@ export class PendingOverrideIntentStore { * Used by the drain to iterate over all pending work. */ channelIds(): IterableIterator { - return this.record.intents.keys(); + return this.intents.keys(); } /** * Compare-and-delete: atomically verify that the current intent's generation - * still matches `capturedGen`, then delete the intent (and its receipt is - * deleted by the caller via `readStateStorage` in the same transaction). + * still matches `capturedGen`, then delete the intent. * * Returns true when the deletion was performed; false when the generation * no longer matches (a newer intent replaced this one). * - * This is the Amendment A + B compare-and-delete seam. Called after the - * Amendment C gen re-check confirms the generation is current. + * Does NOT persist — caller calls persistLocalState() afterward. */ compareAndDelete(channelId: string, capturedGen: number): boolean { - const current = this.record.intents.get(channelId); + const current = this.intents.get(channelId); if (!current || current.gen !== capturedGen) return false; - this.record.intents.delete(channelId); - if (this.pubkey) persist(this.pubkey, this.record); + this.intents.delete(channelId); return true; } - /** - * True if the last localStorage write failed. Same-session drain still - * works; a crash before drain degrades to per-device behaviour. - */ - get isStorageFailed(): boolean { - return this.record.storageFailed; + /** Current nextGen value — used by readStateStorage serialization. */ + get nextGen(): number { + return this._nextGen; } - /** - * Number of pending intents. Used by tests and diagnostics. - */ + /** Current intents map — used by readStateStorage serialization. */ + get all(): ReadonlyMap { + return this.intents; + } + + /** Number of pending intents. Used by tests and diagnostics. */ get size(): number { - return this.record.intents.size; + return this.intents.size; } } diff --git a/desktop/src/features/channels/readState/readStateDrain.ts b/desktop/src/features/channels/readState/readStateDrain.ts index 241104d651..17626ca3df 100644 --- a/desktop/src/features/channels/readState/readStateDrain.ts +++ b/desktop/src/features/channels/readState/readStateDrain.ts @@ -14,6 +14,7 @@ import { } from "@/features/channels/readState/readStateFormat"; import type { AppliedReceipt } from "@/features/channels/readState/readStateStorage"; import { pendingOverrideIntentStore } from "@/features/channels/pendingOverrideIntents"; +import { toast } from "sonner"; import type { MarkResult } from "@/features/channels/readState/readStateManager"; /** Subset of ReadStateManager state and methods needed for drain operations. */ @@ -33,8 +34,97 @@ export interface DrainContext { schedulePublish(): void; notifyListeners(): void; channelFrontier(channelId: string): number; + markContextRead(channelId: string, unixTimestamp: number): void; tryCandidatePlan(rawCtxId: string, reg: OverrideRegister): boolean; restoreExtraSlotIds(prev: string[]): void; + /** Drain outcome callback — wired by the hook for toast/rollback routing. + * `sourceScope` is set for `read+applied` outcomes; it carries the source + * that was being cleared so the hook can remove exactly that source from the + * forced-unread entry rather than deleting the whole entry. */ + onDrainOutcome: + | (( + channelId: string, + op: string, + status: string, + reason?: string, + sourceScope?: string, + ) => void) + | null; +} + +/** + * Shape of ReadStateManager internals accessed by createDrainContext. + * Listed explicitly so TypeScript validates the access at the factory site. + */ +interface ManagerInternals { + readonly pubkey: string; + readonly destroyed: boolean; + isLoadComplete: boolean; + readonly loadGeneration: number; + readonly overrideRegisters: Map; + readonly appliedReceipts: Map; + readonly publishableContextIds: Set; + readonly extraSlotIds: string[]; + onDrainOutcome: + | (( + channelId: string, + op: string, + status: string, + reason?: string, + sourceScope?: string, + ) => void) + | null; + persistLocalState(): boolean; + schedulePublish(): void; + notifyListeners(): void; + channelFrontier(channelId: string): number; + markContextRead(channelId: string, unixTimestamp: number): void; + tryCandidatePlan(rawCtxId: string, reg: OverrideRegister): boolean; + restoreExtraSlotIds(prev: string[]): void; +} + +/** + * Build a typed DrainContext adapter from the manager's internals. + * This is the single place where the manager exposes its internals to the + * drain layer — no cast required. + */ +export function createDrainContext(mgr: ManagerInternals): DrainContext { + return { + get pubkey() { + return mgr.pubkey; + }, + get destroyed() { + return mgr.destroyed; + }, + get isLoadComplete() { + return mgr.isLoadComplete; + }, + get loadGeneration() { + return mgr.loadGeneration; + }, + get overrideRegisters() { + return mgr.overrideRegisters; + }, + get appliedReceipts() { + return mgr.appliedReceipts; + }, + get publishableContextIds() { + return mgr.publishableContextIds; + }, + get extraSlotIds() { + return mgr.extraSlotIds; + }, + get onDrainOutcome() { + return mgr.onDrainOutcome; + }, + persistLocalState: () => mgr.persistLocalState(), + schedulePublish: () => mgr.schedulePublish(), + notifyListeners: () => mgr.notifyListeners(), + channelFrontier: (id) => mgr.channelFrontier(id), + markContextRead: (id, ts) => mgr.markContextRead(id, ts), + tryCandidatePlan: (id, reg) => mgr.tryCandidatePlan(id, reg), + restoreExtraSlotIds: (prev) => mgr.restoreExtraSlotIds(prev), + }; } /** @@ -42,7 +132,7 @@ export interface DrainContext { * Implements plan phase 4 / Amendments A + B + C. * * Normative drain sequence per intent: - * 1. Capture — snapshot intent (op, gen, sourceScope) and pubkey fence. + * 1. Capture — snapshot intent (op, gen, sourceScope, readTarget) and pubkey fence. * 2. Replay — perform the register action against the complete-load state. * For applied: co-commit register + receipt atomically (Amendment C). * 3. Re-check (Amendment A whole-transaction fence) — compare captured gen @@ -69,6 +159,7 @@ export async function drainPendingIntents( const capturedGen = captured.gen; const capturedOp = captured.op; const capturedSourceScope = captured.sourceScope; + const capturedReadTarget = captured.readTarget; const capturedPubkey = ctx.pubkey; // Pubkey fence: abort if identity changed mid-drain. @@ -97,7 +188,13 @@ export async function drainPendingIntents( replayResult = capturedOp === "unread" ? markChannelUnreadDirect(ctx, channelId, capturedGen, capturedOp) - : markChannelReadDirect(ctx, channelId, capturedGen, capturedOp); + : markChannelReadDirect( + ctx, + channelId, + capturedGen, + capturedOp, + capturedReadTarget, + ); } // Amendment A whole-transaction fence: re-check gen under store owner. @@ -129,25 +226,34 @@ export async function drainPendingIntents( else ctx.appliedReceipts.set(channelId, prevReceipt); if (!wasPublishable) ctx.publishableContextIds.delete(channelId); ctx.restoreExtraSlotIds(prevExtraSlotIds); - console.warn( - `[ReadStateManager] drain: register+receipt commit failed for ${channelId}, intent preserved`, - ); continue; } ctx.schedulePublish(); } - // Step 2: source cleanup (idempotent). - void capturedSourceScope; // reserved for future UI-layer cleanup hook + // Step 2: source cleanup (idempotent) — capturedSourceScope scopes removal. + // The sourceScope is forwarded to the onDrainOutcome callback so the hook + // layer can remove exactly the right source from the forced-unread entry. - // Step 3: toast for genuine refusals. + // Step 3: surface outcomes — toast genuine refusals, route to hook callback. if (replayResult.status === "refused") { const reason = replayResult.reason; if (reason !== "already_inactive" && reason !== "load_incomplete") { - console.warn( - `[ReadStateManager] drain: intent ${capturedOp} for ${channelId} refused: ${reason}`, - ); + // Genuine refusal (uint32_overflow, budget_exhausted, storage_failed): + // show toast and surface to hook for forced-entry rollback. + const msg = toastForDrainRefusal(capturedOp, reason); + if (msg) toast.error(msg); + ctx.onDrainOutcome?.(channelId, capturedOp, "refused", reason); } + // already_inactive: silent definitive success — no badge, no toast. + } else if (replayResult.status === "applied") { + ctx.onDrainOutcome?.( + channelId, + capturedOp, + "applied", + undefined, + capturedSourceScope, + ); } // Step 4: atomic cleanup commit — delete receipt + compare-and-delete intent. @@ -163,6 +269,21 @@ export async function drainPendingIntents( } } +/** Human-readable toast for a genuine drain refusal. Returns null for silent reasons. */ +function toastForDrainRefusal(op: string, reason: string): string | null { + if (reason === "budget_exhausted") + return "Could not mark unread: override budget exhausted."; + if (reason === "uint32_overflow") + return op === "unread" + ? "Could not mark unread: counter limit reached." + : "Could not clear unread override: counter limit reached."; + if (reason === "storage_failed") + return op === "unread" + ? "Could not mark unread: storage write failed." + : "Could not clear unread override: storage write failed."; + return null; +} + /** * Replay a mark-unread directly against the complete-load state (drain path). * Sets ctx.appliedReceipts so the subsequent atomic persistLocalState() call @@ -205,6 +326,9 @@ export function markChannelUnreadDirect( /** * Replay a mark-read directly against the complete-load state (drain path). + * Advances the frontier to `readTarget` (captured at click time) before the + * C-bump so the read lands at the correct logical position even when the load + * completed later. * Sets ctx.appliedReceipts so the subsequent atomic persistLocalState() call * in drainPendingIntents includes the receipt (Amendment C). * Does NOT persist or enqueue — only called from drainPendingIntents. @@ -214,9 +338,14 @@ export function markChannelReadDirect( channelId: string, capturedGen: number, capturedOp: "read", + readTarget?: number, ): MarkResult { if (!ctx.isLoadComplete) return { status: "refused", reason: "load_incomplete" }; + // Advance frontier to readTarget first (spec order: frontier → C-bump). + if (readTarget !== undefined && readTarget > 0) { + ctx.markContextRead(channelId, readTarget); + } const reg = ctx.overrideRegisters.get(channelId); const effectiveFrontier = ctx.channelFrontier(channelId); if (!reg) return { status: "refused", reason: "already_inactive" }; @@ -242,6 +371,7 @@ export function markChannelReadDirect( /** * Public mark-unread: write-ordered forced entry then queue/apply the override. * Queues an intent when load is incomplete; applies immediately otherwise. + * Persists the intent atomically in the v2 blob via persistLocalState(). */ export function markChannelUnread( ctx: DrainContext, @@ -249,6 +379,10 @@ export function markChannelUnread( ): MarkResult { if (!ctx.isLoadComplete) { pendingOverrideIntentStore.enqueue(channelId, "unread"); + if (!ctx.persistLocalState()) { + // Storage failure — intent is in-memory only; session drain still works. + // Optimistic presentation continues (forced entry already written). + } return { status: "queued" }; } const existing = ctx.overrideRegisters.get(channelId); @@ -286,13 +420,27 @@ export function markChannelUnread( /** * Public mark-read: queues an intent when load is incomplete; C-bumps otherwise. + * Captures the current frontier as readTarget when queuing, so the drain can + * advance the frontier to the correct position later. + * Captures sourceScope when provided so the drain can perform exact source cleanup. + * Persists the intent atomically in the v2 blob via persistLocalState(). */ export function markChannelRead( ctx: DrainContext, channelId: string, + sourceScope?: string, ): MarkResult { if (!ctx.isLoadComplete) { - pendingOverrideIntentStore.enqueue(channelId, "read", undefined); + const readTarget = ctx.channelFrontier(channelId); + pendingOverrideIntentStore.enqueue( + channelId, + "read", + sourceScope, + readTarget > 0 ? readTarget : undefined, + ); + if (!ctx.persistLocalState()) { + // Storage failure — intent is in-memory only; session drain still works. + } return { status: "queued" }; } const reg = ctx.overrideRegisters.get(channelId); diff --git a/desktop/src/features/channels/readState/readStateManager.test.mjs b/desktop/src/features/channels/readState/readStateManager.test.mjs index 3d7845d237..2347bf1810 100644 --- a/desktop/src/features/channels/readState/readStateManager.test.mjs +++ b/desktop/src/features/channels/readState/readStateManager.test.mjs @@ -2379,7 +2379,7 @@ test("drain_sameSourceReincarnation_genFenceZeroLocalEffects", async () => { // observable as read. This asserts that never happens. globalThis.window.localStorage = makeLocalStorage(); const pubkey = "d1".repeat(32); - const channelId = "reincarnation-channel-" + "a".repeat(40); + const channelId = `reincarnation-channel-${"a".repeat(40)}`; const fakeRelay = { fetchEvents: async () => [], @@ -2469,7 +2469,7 @@ test("drain_crashWindow1_restartReplaysOnce_extraBumpOnly", async () => { // before step 1) to confirm restart does replay exactly once. globalThis.window.localStorage = makeLocalStorage(); const pubkey = "d2".repeat(32); - const channelId = "crash-window-1-ch-" + "b".repeat(44); + const channelId = `crash-window-1-ch-${"b".repeat(44)}`; const fakeRelay = { fetchEvents: async () => [], @@ -2493,21 +2493,16 @@ test("drain_crashWindow1_restartReplaysOnce_extraBumpOnly", async () => { ); // The intent is now persisted (gen=1). Simulate a crash: destroy manager - // without draining — intent stays in localStorage, no register committed. + // without draining — intent stays in the v2 blob, no register committed. mgr1.destroy(); - // Verify: intent is persisted (survived the "crash"). - const _survivingIntent = pendingOverrideIntentStore.get(channelId); - // Note: pendingOverrideIntentStore is a singleton; loadForPubkey reloads on - // the new manager construction below. - // ── Session 2: restart. Intent present, no receipt → replay fires once ──── const mgr2 = new ReadStateManager(pubkey, fakeRelay); mgr2.hydrateFromLocalStorage(); mgr2.effectiveState.set(channelId, 100); mgr2.isLoadComplete = true; - // Confirm intent survived (reloaded from localStorage by constructor's loadForPubkey). + // Confirm intent survived (reloaded from v2 blob by hydrateFromLocalStorage). const intentAfterRestart = pendingOverrideIntentStore.get(channelId); assert.ok( intentAfterRestart, @@ -2558,11 +2553,10 @@ test("drain_crashWindow2_receiptPresent_restartNoRebump", async () => { // register is unchanged from the original single commit. globalThis.window.localStorage = makeLocalStorage(); const pubkey = "d3".repeat(32); - const channelId = "crash-window-2-ch-" + "c".repeat(44); + const channelId = `crash-window-2-ch-${"c".repeat(44)}`; const ls = globalThis.window.localStorage; const v2Key = `buzz.nip-rs.override-state.v2:${pubkey}`; - const intentKey = `buzz.nip-rs.pending-intents.v1:${pubkey}`; // Craft the crash state directly in localStorage. // Register committed with S=2 (unread applied: max(0,0)+1=1... let's use S=2 for clarity), @@ -2574,13 +2568,8 @@ test("drain_crashWindow2_receiptPresent_restartNoRebump", async () => { JSON.stringify({ r: { [channelId]: { s: sSingleCommit, c: 0, b: 100, f: 100 } }, receipts: { [channelId]: { intentGen: 1, op: "unread" } }, - }), - ); - ls.setItem( - intentKey, - JSON.stringify({ - nextGen: 2, - intents: { [channelId]: { gen: 1, op: "unread" } }, + pi: { [channelId]: { gen: 1, op: "unread" } }, + ng: 2, }), ); @@ -2646,7 +2635,7 @@ test("drain_crashAfterAppliedUnread_newerRemoteRead_noRebump", async () => { // proves application; only cleanup+delete runs, so C=6 stands. globalThis.window.localStorage = makeLocalStorage(); const pubkey = "d4".repeat(32); - const channelId = "amend-c-unread-race-" + "d".repeat(44); + const channelId = `amend-c-unread-race-${"d".repeat(44)}`; const fakeRelay = { fetchEvents: async () => [], @@ -2665,19 +2654,13 @@ test("drain_crashAfterAppliedUnread_newerRemoteRead_noRebump", async () => { // and after restart we merge C=6 into the register. const ls = globalThis.window.localStorage; const v2Key = `buzz.nip-rs.override-state.v2:${pubkey}`; - const intentKey = `buzz.nip-rs.pending-intents.v1:${pubkey}`; ls.setItem( v2Key, JSON.stringify({ r: { [channelId]: { s: 5, c: 3, b: 100, f: 100 } }, receipts: { [channelId]: { intentGen: 1, op: "unread" } }, - }), - ); - ls.setItem( - intentKey, - JSON.stringify({ - nextGen: 2, - intents: { [channelId]: { gen: 1, op: "unread" } }, + pi: { [channelId]: { gen: 1, op: "unread" } }, + ng: 2, }), ); @@ -2730,11 +2713,10 @@ test("drain_crashAfterAppliedRead_newerRemoteUnread_noRebump", async () => { // reverse the new unread. globalThis.window.localStorage = makeLocalStorage(); const pubkey = "d5".repeat(32); - const channelId = "amend-c-read-race-" + "e".repeat(45); + const channelId = `amend-c-read-race-${"e".repeat(45)}`; const ls = globalThis.window.localStorage; const v2Key = `buzz.nip-rs.override-state.v2:${pubkey}`; - const intentKey = `buzz.nip-rs.pending-intents.v1:${pubkey}`; // Crash state: read C-bump committed (S=5, C=6), receipt present, intent alive. ls.setItem( @@ -2742,13 +2724,8 @@ test("drain_crashAfterAppliedRead_newerRemoteUnread_noRebump", async () => { JSON.stringify({ r: { [channelId]: { s: 5, c: 6, b: 100, f: 200 } }, receipts: { [channelId]: { intentGen: 1, op: "read" } }, - }), - ); - ls.setItem( - intentKey, - JSON.stringify({ - nextGen: 2, - intents: { [channelId]: { gen: 1, op: "read" } }, + pi: { [channelId]: { gen: 1, op: "read" } }, + ng: 2, }), ); @@ -2807,7 +2784,7 @@ test("drain_receiptWithoutMatchingIntent_sweptWithoutRegisterEffect", async () = // receipt persists across sessions" invariant. globalThis.window.localStorage = makeLocalStorage(); const pubkey = "d6".repeat(32); - const channelId = "receipt-sweep-ch-" + "f".repeat(46); + const channelId = `receipt-sweep-ch-${"f".repeat(46)}`; const ls = globalThis.window.localStorage; const v2Key = `buzz.nip-rs.override-state.v2:${pubkey}`; @@ -2890,7 +2867,7 @@ test("drain_registerReceiptAtomicity_noStateWhereOnePersistsWithoutOther", async // asserts the combined single-write invariant. globalThis.window.localStorage = makeLocalStorage(); const pubkey = "d7".repeat(32); - const channelId = "atomicity-ch-" + "a".repeat(51); + const channelId = `atomicity-ch-${"a".repeat(51)}`; const ls = globalThis.window.localStorage; let blockV2Writes = false; @@ -3023,3 +3000,810 @@ test("fetchOwnBlobBeforePublish_foreignClientId_rotatesSlotAndUpdatesMetadata", delete globalThis.window.__TAURI_INTERNALS__; } }); + +// ── Test 30: Amendment A biting — gen2 enqueued DURING replay ───────────────── +test("drain_gen2EnqueuedDuringReplay_rollsBackGen1Register", async () => { + // Biting Amendment A: gen2 is enqueued DURING replay of gen1 (not before drain + // starts). The re-check sees gen2 != capturedGen1 and rolls back all mutations. + // + // Injection point: override mgr.tryCandidatePlan to enqueue gen2 inside + // markChannelUnreadDirect's budget check — the register write follows, but + // the subsequent re-check must detect the gen change and roll it back. + globalThis.window.localStorage = makeLocalStorage(); + const pubkey = "30".repeat(32); + const channelId = `amend-a-biting-ch-${"b".repeat(46)}`; + + const fakeRelay = { + fetchEvents: async () => [], + publishEvent: async () => {}, + subscribeFenced: async (_f, _h) => makeFenceHandle({ eose: true }), + subscribeLive: async (_f, _h) => () => {}, + subscribeToReconnects: () => () => {}, + getConnectionGeneration: () => 0, + }; + + const mgr = new ReadStateManager(pubkey, fakeRelay); + mgr.effectiveState.set(channelId, 50); + + // Queue gen1 (unread) during incomplete load. + mgr.isLoadComplete = false; + const q = mgr.markChannelUnread(channelId); + assert.equal(q.status, "queued", "gen1 must be queued"); + const gen1 = pendingOverrideIntentStore.get(channelId).gen; + assert.ok(gen1 >= 1, "gen1 must be a positive generation number"); + + // Override tryCandidatePlan to enqueue gen2 DURING replay of gen1. + const origTry = mgr.tryCandidatePlan.bind(mgr); + let injected = false; + mgr.tryCandidatePlan = (id, reg) => { + const ok = origTry(id, reg); + if (!injected && id === channelId) { + injected = true; + // This fires inside markChannelUnreadDirect, during gen1's replay. + // After tryCandidatePlan returns, the register is written temporarily. + // The Amendment A re-check must still roll it back. + pendingOverrideIntentStore.enqueue(channelId, "unread"); + } + return ok; + }; + + // Complete load and drain gen1. The override fires gen2 mid-replay. + mgr.isLoadComplete = true; + await mgr.drainPendingIntents(mgr.loadGeneration); + + // Gen2 was enqueued during gen1's replay → Amendment A re-check fires. + // Gen1 committed zero local effects: register must still be absent (rolled back). + const reg = mgr.overrideRegisters.get(channelId); + assert.equal( + reg, + undefined, + "register must be rolled back by gen-fence (gen2 was enqueued during gen1 replay)", + ); + + // Gen2 must still be alive in the intent store — it was not consumed. + const gen2Intent = pendingOverrideIntentStore.get(channelId); + assert.ok(gen2Intent, "gen2 intent must still be in the queue"); + assert.ok( + gen2Intent.gen > gen1, + "gen2 must have a higher generation than gen1", + ); + + // Clean up gen2. + pendingOverrideIntentStore.compareAndDelete(channelId, gen2Intent.gen); + mgr.destroy(); +}); + +// ── Test 31: unread→read inversion — read gen replaces unread gen ───────────── +test("drain_unreadToReadInversion_onlyReadApplied", async () => { + // Both inversion directions: unread gen1, then read gen2 replaces it. + // Drain must apply only gen2 (read = C-bump), not gen1 (unread = S-bump). + globalThis.window.localStorage = makeLocalStorage(); + const pubkey = "31".repeat(32); + const channelId = `inversion-ur-ch-${"c".repeat(48)}`; + + const fakeRelay = { + fetchEvents: async () => [], + publishEvent: async () => {}, + subscribeFenced: async (_f, _h) => makeFenceHandle({ eose: true }), + subscribeLive: async (_f, _h) => () => {}, + subscribeToReconnects: () => () => {}, + getConnectionGeneration: () => 0, + }; + + const mgr = new ReadStateManager(pubkey, fakeRelay); + // Pre-seed an existing unread register so C-bump has something to work with. + mgr.effectiveState.set(channelId, 50); + mgr.overrideRegisters.set(channelId, { s: 1, c: 0, b: 50 }); + mgr.publishableContextIds.add(channelId); + mgr.isLoadComplete = false; + + // Step 1: queue unread (gen1). + const r1 = mgr.markChannelUnread(channelId); + assert.equal(r1.status, "queued"); + const gen1 = pendingOverrideIntentStore.get(channelId).gen; + + // Step 2: queue read (gen2) — replaces gen1. + const r2 = mgr.markChannelRead(channelId); + assert.equal(r2.status, "queued"); + const intent = pendingOverrideIntentStore.get(channelId); + assert.equal(intent.op, "read", "current intent must be read (gen2)"); + assert.ok(intent.gen > gen1, "gen2 must exceed gen1"); + + // Complete load and drain. + mgr.isLoadComplete = true; + await mgr.drainPendingIntents(mgr.loadGeneration); + + // Gen2 (read) must have applied: C bumped, S unchanged at 1. + const reg = mgr.overrideRegisters.get(channelId); + assert.ok(reg, "register must exist after read drain"); + assert.equal( + reg.s, + 1, + "S must be unchanged — unread gen1 was replaced, not applied", + ); + assert.ok(reg.c > 0, "C must be bumped by the read drain (gen2)"); + + // Gen2 intent consumed. + assert.equal( + pendingOverrideIntentStore.get(channelId), + undefined, + "gen2 intent must be deleted after drain", + ); + + mgr.destroy(); +}); + +// ── Test 32: read→unread inversion — unread gen replaces read gen ───────────── +test("drain_readToUnreadInversion_onlySBumpApplied", async () => { + // Mirror of Test 31: read gen1 queued, then unread gen2 replaces it. + // Drain applies only gen2 (S-bump), not gen1 (C-bump). + globalThis.window.localStorage = makeLocalStorage(); + const pubkey = "32".repeat(32); + const channelId = `inversion-ru-ch-${"d".repeat(48)}`; + + const fakeRelay = { + fetchEvents: async () => [], + publishEvent: async () => {}, + subscribeFenced: async (_f, _h) => makeFenceHandle({ eose: true }), + subscribeLive: async (_f, _h) => () => {}, + subscribeToReconnects: () => () => {}, + getConnectionGeneration: () => 0, + }; + + const mgr = new ReadStateManager(pubkey, fakeRelay); + mgr.effectiveState.set(channelId, 50); + mgr.isLoadComplete = false; + + // Queue read (gen1) — no existing register, will be already_inactive after drain. + const r1 = mgr.markChannelRead(channelId); + assert.equal(r1.status, "queued"); + + // Queue unread (gen2) — replaces read. + const r2 = mgr.markChannelUnread(channelId); + assert.equal(r2.status, "queued"); + const intent = pendingOverrideIntentStore.get(channelId); + assert.equal(intent.op, "unread", "current intent must be unread (gen2)"); + + // Complete load and drain. + mgr.isLoadComplete = true; + await mgr.drainPendingIntents(mgr.loadGeneration); + + // Gen2 (unread) applied: S bumped. + const reg = mgr.overrideRegisters.get(channelId); + assert.ok(reg, "register must exist after unread drain"); + assert.ok(reg.s > 0, "S must be bumped by unread drain (gen2)"); + assert.equal( + reg.c, + 0, + "C must be zero — read gen1 was replaced, not applied", + ); + + // Consumed. + assert.equal( + pendingOverrideIntentStore.get(channelId), + undefined, + "gen2 intent consumed", + ); + + mgr.destroy(); +}); + +// ── Test 33: compare-delete preserves newer intent ──────────────────────────── +test("drain_compareDeletePreservesNewerIntent", async () => { + // compare-and-delete for gen1 must fail if gen2 was enqueued before the + // compare-delete step. The drain's cleanup commit must not consume gen2. + // + // We inject gen2 via onDrainOutcome (fires just before compare-delete). + globalThis.window.localStorage = makeLocalStorage(); + const pubkey = "33".repeat(32); + const channelId = `cmp-del-newer-ch-${"e".repeat(47)}`; + + const fakeRelay = { + fetchEvents: async () => [], + publishEvent: async () => {}, + subscribeFenced: async (_f, _h) => makeFenceHandle({ eose: true }), + subscribeLive: async (_f, _h) => () => {}, + subscribeToReconnects: () => () => {}, + getConnectionGeneration: () => 0, + }; + + const mgr = new ReadStateManager(pubkey, fakeRelay); + mgr.effectiveState.set(channelId, 50); + mgr.isLoadComplete = false; + + // Queue gen1 (unread). + mgr.markChannelUnread(channelId); + const gen1 = pendingOverrideIntentStore.get(channelId).gen; + + // Wire onDrainOutcome to enqueue gen2 BEFORE compare-and-delete runs. + // The onDrainOutcome fires at step 3 (surface outcomes), before step 4 (cleanup commit). + mgr.onDrainOutcome = (chId, _op, _status) => { + if (chId === channelId) { + // Enqueue gen2 between outcome surfacing and compare-delete. + pendingOverrideIntentStore.enqueue(channelId, "unread"); + } + }; + + mgr.isLoadComplete = true; + await mgr.drainPendingIntents(mgr.loadGeneration); + + // Gen2 must still be in the queue (compare-delete for gen1 preserved gen2). + const remaining = pendingOverrideIntentStore.get(channelId); + assert.ok( + remaining, + "gen2 intent must survive compare-delete targeting gen1", + ); + assert.ok( + remaining.gen > gen1, + "surviving intent must be gen2 (newer than gen1)", + ); + + // Clean up. + pendingOverrideIntentStore.compareAndDelete(channelId, remaining.gen); + mgr.destroy(); +}); + +// ── Test 34: Amendment C via production load — restart + fetchAndMerge ──────── +test("drain_amendmentC_restart_fetchAndMerge_noBumpAfterReceipt", async () => { + // Amendment C via the PRODUCTION load path: after a crash-in-cleanup, the + // next session hydrates from storage and runs fetchAndMerge() (not a direct + // register write). The receipt must prevent a second S-bump. + // + // Thufir's finding: Tests 26/27 directly set isLoadComplete=true and write + // overrideRegisters directly — they bypass the production load path. + // This test uses the actual fetchAndMerge() flow via fakeRelay event delivery. + globalThis.window.localStorage = makeLocalStorage(); + const pubkey = "34".repeat(32); + const channelId = `amend-c-prod-ch-${"f".repeat(48)}`; + + // Craft crash-state in localStorage: unread applied (S=5), receipt intact. + const ls = globalThis.window.localStorage; + const v2Key = `buzz.nip-rs.override-state.v2:${pubkey}`; + ls.setItem( + v2Key, + JSON.stringify({ + r: { [channelId]: { s: 5, c: 3, b: 100, f: 100 } }, + receipts: { [channelId]: { intentGen: 1, op: "unread" } }, + pi: { [channelId]: { gen: 1, op: "unread" } }, + ng: 2, + }), + ); + + // A remote device published C=6 (read). Simulate via a fetched event that, + // when decrypted, delivers C=6 for the channel via the standard merge path. + const remoteBlob = JSON.stringify({ + v: 1, + client_id: "remote-device-client", + contexts: { + [`ov_s:${channelId}`]: 5, + [`ov_c:${channelId}`]: 6, + [`ov_b:${channelId}`]: 100, + [channelId]: 200, // frontier advanced past B → override inactive + }, + }); + // Slot ID must match /^[0-9a-f]{32}$/ to pass isValidReadStateDTag. + const remoteSlotId = "0d".repeat(16); + const blobEvent = { + id: "ab".repeat(32), + pubkey, + created_at: 9999, + kind: 30078, + tags: [ + ["d", `read-state:${remoteSlotId}`], + ["t", "read-state"], + ], + content: "CIPHER", + sig: "s".repeat(128), + }; + + let fetchCallCount = 0; + globalThis.window.__TAURI_INTERNALS__ = { + invoke: async (command) => { + if (command === "nip44_decrypt_from_self") return remoteBlob; + throw new Error(`Unexpected: ${command}`); + }, + }; + + const fakeRelay = { + fetchEvents: async () => { + fetchCallCount++; + return fetchCallCount === 1 ? [blobEvent] : []; + }, + publishEvent: async () => {}, + subscribeToReconnects: () => () => {}, + getConnectionGeneration: () => 0, + subscribeFenced: async (_filter, handler) => { + // Deliver the event synchronously before EOSE. + if (fetchCallCount === 0) { + handler(blobEvent); + } + return makeFenceHandle({ eose: true }); + }, + subscribeLive: async (_f, _h) => () => {}, + }; + + try { + // Session 2: hydrate, then complete load via real fetchAndMerge(). + const mgr = new ReadStateManager(pubkey, fakeRelay); + mgr.hydrateFromLocalStorage(); + + // Receipt must be loaded. + const receipt = mgr.appliedReceipts.get(channelId); + assert.ok(receipt, "receipt must be loaded on restart"); + assert.equal(receipt.intentGen, 1); + + // Run fetchAndMerge via the production path. + await mgr.fetchAndMerge(mgr.loadGeneration); + assert.equal(mgr.isLoadComplete, true, "load must complete"); + + // Drain via production path. + await mgr.drainPendingIntents(mgr.loadGeneration); + + // The alreadyApplied path must have run: no new S-bump. + // C must reflect the merged remote (max(3,6)=6). + const reg = mgr.overrideRegisters.get(channelId); + assert.ok(reg, "register must exist"); + assert.equal( + reg.s, + 5, + "S must NOT be re-bumped by drain (receipt prevents it)", + ); + assert.ok(reg.c >= 6, "C must reflect remote C=6 merge"); + + // Intent consumed. + assert.equal( + pendingOverrideIntentStore.get(channelId), + undefined, + "intent must be consumed after drain", + ); + } finally { + delete globalThis.window.__TAURI_INTERNALS__; + } +}); + +// ── Test 35: atomic deletion — intent + receipt deleted in one persist ───────── +test("drain_cleanupCommit_intentAndReceiptDeletedAtomically", async () => { + // Atomic deletion witness: after successful drain, both the intent AND the + // receipt are absent from the v2 blob in a single commit (same persist call). + // If they were deleted in separate writes, there would be a window where one + // survives without the other. + // + // Verify: after drain, the v2 blob contains neither pi nor receipts entry. + globalThis.window.localStorage = makeLocalStorage(); + const pubkey = "35".repeat(32); + const channelId = `atomic-del-ch-${"g".repeat(50)}`; + + const ls = globalThis.window.localStorage; + const v2Key = `buzz.nip-rs.override-state.v2:${pubkey}`; + + const fakeRelay = { + fetchEvents: async () => [], + publishEvent: async () => {}, + subscribeFenced: async (_f, _h) => makeFenceHandle({ eose: true }), + subscribeLive: async (_f, _h) => () => {}, + subscribeToReconnects: () => () => {}, + getConnectionGeneration: () => 0, + }; + + const mgr = new ReadStateManager(pubkey, fakeRelay); + mgr.effectiveState.set(channelId, 50); + mgr.isLoadComplete = false; + + // Queue the intent. + mgr.markChannelUnread(channelId); + + // Drain: apply the intent (receipt gets co-committed with register). + mgr.isLoadComplete = true; + await mgr.drainPendingIntents(mgr.loadGeneration); + + // After successful drain, check the persisted v2 blob. + const rawAfter = ls.getItem(v2Key); + const blobAfter = rawAfter ? JSON.parse(rawAfter) : {}; + + // Both intent (pi) and receipt (receipts) must be absent — single cleanup commit. + const piEntry = blobAfter?.pi?.[channelId]; + assert.equal( + piEntry, + undefined, + "intent must be deleted from pi in cleanup commit", + ); + + const receiptEntry = blobAfter?.receipts?.[channelId]; + assert.equal( + receiptEntry, + undefined, + "receipt must be deleted from receipts in cleanup commit", + ); + + // The register IS present (drain succeeded, override applied). + const regEntry = blobAfter?.r?.[channelId]; + assert.ok(regEntry, "register must persist after successful drain"); + + mgr.destroy(); +}); + +// ── Test 36: controller — automatic backoff after incomplete verdict ─────────── +test("retryLoad_incompleteVerdictAdvancesRetryAttempt", async () => { + // After an incomplete verdict, retryAttempt increments so the next retry + // will use exponential backoff. Destroy/cancel before the delayed retry fires. + globalThis.window.localStorage = makeLocalStorage(); + const pubkey = "36".repeat(32); + + const fakeRelay = { + fetchEvents: async () => [], + publishEvent: async () => {}, + subscribeToReconnects: () => () => {}, + getConnectionGeneration: () => 0, + subscribeFenced: async () => { + throw new Error("fence timeout"); + }, + subscribeLive: async (_f, _h) => () => {}, + }; + + const mgr = new ReadStateManager(pubkey, fakeRelay); + assert.equal(mgr.retryAttempt, 0, "precondition: retryAttempt is 0"); + + // First retry (immediate — attempt 0 means delayMs=0). + await mgr.retryLoad(); + + // After incomplete verdict, retryAttempt must be incremented. + assert.ok( + mgr.retryAttempt > 0, + "retryAttempt must be > 0 after incomplete verdict", + ); + + // Second retry would use backoff. Destroy before it fires. + mgr.destroy(); + assert.equal(mgr.retryBackoffTimer, null, "timer must be cleared on destroy"); +}); + +// ── Test 37: controller — reconnect during loadInFlight → pendingRetryOnComplete +test("retryLoad_reconnectDuringLoadInFlight_coalescesToPendingRetry", async () => { + // When retryLoad() is called while a load is already in flight, + // pendingRetryOnComplete must be set (not dropped). The pending flag + // triggers a fresh generation after the in-flight load completes. + globalThis.window.localStorage = makeLocalStorage(); + const pubkey = "37".repeat(32); + + let resolveFetch; + const fetchBarrier = new Promise((r) => { + resolveFetch = r; + }); + + const fakeRelay = { + fetchEvents: async () => [], + publishEvent: async () => {}, + subscribeToReconnects: () => () => {}, + getConnectionGeneration: () => 0, + subscribeFenced: async (_f, _h) => { + // Block until barrier released. + await fetchBarrier; + return makeFenceHandle({ eose: true }); + }, + subscribeLive: async (_f, _h) => () => {}, + }; + + const mgr = new ReadStateManager(pubkey, fakeRelay); + + // Start a load that won't complete until we release the barrier. + const loadPromise = mgr.retryLoad(); + // Give the async function time to enter the subscribeFenced await. + await new Promise((r) => setTimeout(r, 0)); + assert.equal( + mgr.loadInFlight, + true, + "loadInFlight must be true while fetch is blocked", + ); + + // Simulate reconnect arriving while load is in flight. + mgr.retryLoad(); // synchronous entry → sets pendingRetryOnComplete + assert.equal( + mgr.pendingRetryOnComplete, + true, + "reconnect during loadInFlight must set pendingRetryOnComplete", + ); + + // Release the barrier so the first retryLoad() can complete. + resolveFetch(); + await loadPromise; + + mgr.destroy(); +}); + +// ── Test 38: controller — stale-completion ordering via gen check ────────────── +test("fetchAndMerge_staleGeneration_doesNotSetLoadComplete", async () => { + // If loadGeneration advances while fetchAndMerge is in flight, the stale + // generation's complete verdict must NOT mark the manager as load-complete. + globalThis.window.localStorage = makeLocalStorage(); + const pubkey = "38".repeat(32); + + let resolveFirst; + const firstBarrier = new Promise((r) => { + resolveFirst = r; + }); + let callCount = 0; + + const fakeRelay = { + fetchEvents: async () => [], + publishEvent: async () => {}, + subscribeToReconnects: () => () => {}, + getConnectionGeneration: () => 0, + subscribeFenced: async (_f, _h) => { + callCount++; + if (callCount === 1) { + // First call: wait for barrier before returning EOSE. + await firstBarrier; + return makeFenceHandle({ eose: true }); + } + return makeFenceHandle({ eose: true }); + }, + subscribeLive: async (_f, _h) => () => {}, + }; + + const mgr = new ReadStateManager(pubkey, fakeRelay); + + // Start gen=1 fetch; it blocks at subscribeFenced. + const gen1 = mgr.loadGeneration; + const firstFetch = mgr.fetchAndMerge(gen1); + await new Promise((r) => setTimeout(r, 0)); + + // Advance the generation by starting a new fetch. + ++mgr.loadGeneration; + const gen2 = mgr.loadGeneration; + await mgr.fetchAndMerge(gen2); + assert.equal(mgr.isLoadComplete, true, "gen2 load must complete"); + + // Now release the blocked gen1 fetch. + resolveFirst(); + await firstFetch; + + // Gen1's verdict should not have mutated isLoadComplete back to stale state. + // (Gen2 left it true; gen1 must not interfere.) + assert.equal( + mgr.isLoadComplete, + true, + "gen1 stale verdict must not corrupt gen2's complete state", + ); + + mgr.destroy(); +}); + +// ── Test 39: controller — destroy cancels backoff timer ─────────────────────── +test("retryLoad_destroyWhileTimerPending_cancelsTimer", async () => { + // destroy() must cancel any pending retryBackoffTimer so no stale retry fires. + globalThis.window.localStorage = makeLocalStorage(); + const pubkey = "39".repeat(32); + + let timerSet = false; + let pendingTimerResolve = null; // captured so we can drain after destroy + const origSetTimeout = globalThis.window.setTimeout; + const origClearTimeout = globalThis.window.clearTimeout; + + // Track active backoff timers by real ID so clearTimeout works correctly. + // When ms > 0 we wrap the fn to capture state, but still use the real timer + // (with a large delay so it doesn't fire during the test) so origClearTimeout + // can cancel it. After destroy() we must resolve the pending await manually + // to let retryLoad's finally-block run and the promise settle. + const activeTimerIds = new Set(); + globalThis.window.setTimeout = (fn, ms) => { + if (ms > 0) { + timerSet = true; + // Schedule with origSetTimeout at a safe delay so clearTimeout works. + const id = origSetTimeout(() => { + activeTimerIds.delete(id); + pendingTimerResolve = null; + fn(); + }, 60_000); // long enough never to fire during test + // Capture resolve so we can drain manually after destroy(). + pendingTimerResolve = fn; + activeTimerIds.add(id); + return id; + } + return origSetTimeout(fn, ms); + }; + globalThis.window.clearTimeout = (id) => { + activeTimerIds.delete(id); + origClearTimeout(id); + }; + + try { + const fakeRelay = { + fetchEvents: async () => [], + publishEvent: async () => {}, + subscribeToReconnects: () => () => {}, + getConnectionGeneration: () => 0, + subscribeFenced: async () => { + throw new Error("lapse"); + }, + subscribeLive: async (_f, _h) => () => {}, + }; + + const mgr = new ReadStateManager(pubkey, fakeRelay); + + // First retry (immediate, attempt=0 → no timer). + await mgr.retryLoad(); + // retryAttempt is now 1; next retry will schedule a timer. + + // Start second retry asynchronously — it will set a timer. + const retryPromise = mgr.retryLoad(); + // Give async fn time to reach the timer await. + await new Promise((r) => origSetTimeout(r, 0)); + + assert.equal(timerSet, true, "a backoff timer must have been scheduled"); + assert.ok( + mgr.retryBackoffTimer !== null, + "retryBackoffTimer must be non-null while waiting", + ); + + // Destroy must cancel the timer. + mgr.destroy(); + assert.equal( + mgr.retryBackoffTimer, + null, + "retryBackoffTimer must be null after destroy", + ); + assert.equal( + activeTimerIds.size, + 0, + "backoff timer must be cleared by destroy()", + ); + + // retryLoad is still awaiting its backoff Promise. Fire the pending resolve + // manually (the real timer was cleared so it won't auto-fire) so the + // finally-block in retryLoad runs and retryPromise settles. Since + // mgr.destroyed===true, retryLoad returns immediately after the resolve. + if (pendingTimerResolve) pendingTimerResolve(); + await retryPromise; + } finally { + globalThis.window.setTimeout = origSetTimeout; + globalThis.window.clearTimeout = origClearTimeout; + } +}); + +// ── Test 40: controller — exactly one drain per complete-load generation ─────── +test("retryLoad_exactlyOneDrainPerCompleteGeneration", async () => { + // A single complete-load transition must trigger exactly one drain. + // Multiple retryLoad() calls for the same complete generation must not + // enqueue a second drain. + globalThis.window.localStorage = makeLocalStorage(); + const pubkey = "40".repeat(32); + const channelId = `one-drain-ch-${"h".repeat(51)}`; + + const fakeRelay = { + fetchEvents: async () => [], + publishEvent: async () => {}, + subscribeToReconnects: () => () => {}, + getConnectionGeneration: () => 0, + subscribeFenced: async (_f, _h) => makeFenceHandle({ eose: true }), + subscribeLive: async (_f, _h) => () => {}, + }; + + const mgr = new ReadStateManager(pubkey, fakeRelay); + mgr.isLoadComplete = false; + + // Queue an intent that will be consumed by the drain. + mgr.markChannelUnread(channelId); + + // First retryLoad → complete → drain fires. + await mgr.retryLoad(); + assert.equal(mgr.isLoadComplete, true, "load must be complete"); + + // Record register state after first drain. + const regAfterFirst = mgr.overrideRegisters.get(channelId); + const sAfterFirst = regAfterFirst?.s ?? 0; + + // Second retryLoad on an already-complete manager: drainScheduled is false + // (already consumed), so a second drain must NOT re-bump the register. + await mgr.retryLoad(); + + const regAfterSecond = mgr.overrideRegisters.get(channelId); + assert.equal( + regAfterSecond?.s ?? 0, + sAfterFirst, + "S must NOT be re-bumped by a second retryLoad (exactly-one-drain invariant)", + ); + + mgr.destroy(); +}); + +// ── Test 41: store/UI — identity swap during drain cancels stale gen ────────── +test("drain_identitySwapDuringDrain_staleGenCancelled", async () => { + // If the manager is destroyed and recreated for a new pubkey, the stale + // generation's drain must be cancelled. Verify via the pubkey fence inside + // drainPendingIntents: pubkeyFence !== ctx.pubkey → break. + globalThis.window.localStorage = makeLocalStorage(); + const pubkeyA = "41".repeat(32); + const channelId = `identity-swap-ch-${"i".repeat(47)}`; + + const fakeRelay = { + fetchEvents: async () => [], + publishEvent: async () => {}, + subscribeFenced: async (_f, _h) => makeFenceHandle({ eose: true }), + subscribeLive: async (_f, _h) => () => {}, + subscribeToReconnects: () => () => {}, + getConnectionGeneration: () => 0, + }; + + const mgrA = new ReadStateManager(pubkeyA, fakeRelay); + mgrA.effectiveState.set(channelId, 50); + mgrA.isLoadComplete = false; + + // Queue intent for pubkeyA. + mgrA.markChannelUnread(channelId); + const intentA = pendingOverrideIntentStore.get(channelId); + assert.ok(intentA, "intent must exist for pubkeyA"); + + // Destroy mgrA before drain (simulates identity swap). + mgrA.destroy(); + + // The drainPendingIntents call with mgrA's context should be cancelled + // because mgrA.destroyed === true. + // Verify by running the drain and asserting the intent is NOT consumed. + // (The drain early-exits on ctx.destroyed.) + await mgrA.drainPendingIntents(mgrA.loadGeneration); + + // The intent must still be alive — drain was no-op because destroyed. + const intentAfter = pendingOverrideIntentStore.get(channelId); + assert.ok(intentAfter, "intent must survive drain when manager is destroyed"); + + // Clean up. + pendingOverrideIntentStore.compareAndDelete(channelId, intentAfter.gen); +}); + +// ── Test 42: deferred refusal — queued unread intent refused by drain ───────── +test("drain_queuedUnread_refused_surfacesOutcome", async () => { + // When a queued unread intent is refused by drain (e.g., budget_exhausted), + // onDrainOutcome must be called with (channelId, "unread", "refused", reason). + globalThis.window.localStorage = makeLocalStorage(); + const pubkey = "42".repeat(32); + const channelId = `deferred-refusal-ch-${"j".repeat(44)}`; + + const fakeRelay = { + fetchEvents: async () => [], + publishEvent: async () => {}, + subscribeFenced: async (_f, _h) => makeFenceHandle({ eose: true }), + subscribeLive: async (_f, _h) => () => {}, + subscribeToReconnects: () => () => {}, + getConnectionGeneration: () => 0, + }; + + const mgr = new ReadStateManager(pubkey, fakeRelay); + mgr.effectiveState.set(channelId, 50); + mgr.isLoadComplete = false; + + // Queue an unread intent with uint32 overflow condition: S is at max. + mgr.overrideRegisters.set(channelId, { s: 0xffffffff, c: 0, b: 50 }); + mgr.publishableContextIds.add(channelId); + const q = mgr.markChannelUnread(channelId); + assert.equal(q.status, "queued", "must queue pre-ready"); + + // Wire outcome callback. + const outcomes = []; + mgr.onDrainOutcome = (chId, op, status, reason) => { + outcomes.push({ chId, op, status, reason }); + }; + + // Complete load and drain. + mgr.isLoadComplete = true; + await mgr.drainPendingIntents(mgr.loadGeneration); + + // Drain must have surfaced the refusal. + assert.equal(outcomes.length, 1, "exactly one outcome must be emitted"); + assert.equal(outcomes[0].chId, channelId); + assert.equal(outcomes[0].op, "unread"); + assert.equal(outcomes[0].status, "refused"); + assert.equal( + outcomes[0].reason, + "uint32_overflow", + "refusal reason must be uint32_overflow", + ); + + // Intent must be cleaned up even after refusal (it's a genuine refusal, not load_incomplete). + assert.equal( + pendingOverrideIntentStore.get(channelId), + undefined, + "intent must be deleted after genuine refusal", + ); + + mgr.destroy(); +}); diff --git a/desktop/src/features/channels/readState/readStateManager.ts b/desktop/src/features/channels/readState/readStateManager.ts index 38a11576f3..cd643e66fb 100644 --- a/desktop/src/features/channels/readState/readStateManager.ts +++ b/desktop/src/features/channels/readState/readStateManager.ts @@ -9,11 +9,9 @@ import { READ_STATE_MAX_SLOTS, MSG_PREFIX, THREAD_PREFIX, - isOverrideKey, encodeOverrideGroup, isOverrideActive, escapeFrontierKey, - unescapeFrontierKey, localExtraSlotIdsKey, type ReadStateBlob, type OverrideRegister, @@ -30,6 +28,12 @@ import { writeStoredReadState, type AppliedReceipt, } from "@/features/channels/readState/readStateStorage"; +import { + splitContextsIntoBudgetedSlots, + trimContextsToBudget, + type SlotSplitResult, + type TrimResult, +} from "@/features/channels/readState/readStateSlotUtils"; import { fencedEnumerationLoad, deduplicateByCoordinate, @@ -40,8 +44,12 @@ import { drainPendingIntents as drainPendingIntentsImpl, markChannelUnread as markChannelUnreadImpl, markChannelRead as markChannelReadImpl, - type DrainContext, + createDrainContext, } from "@/features/channels/readState/readStateDrain"; + +export type { SlotSplitResult, TrimResult }; +export { splitContextsIntoBudgetedSlots, trimContextsToBudget }; + const CLIENT_ID_KEY_PREFIX = "buzz.nip-rs.client-id"; const SLOT_ID_KEY_PREFIX = "buzz.nip-rs.slot-id"; const DEBOUNCE_MS = 5_000; @@ -153,128 +161,14 @@ export function applyRemoteContextTimestamp(args: { return result; } -/** Slot split result. */ -export interface SlotSplitResult { - slots: Array>; - extraSlotIds: string[]; -} - -/** Partition channelEntries across slots; override groups pinned to slot 0. Exported for testing. */ -export function splitContextsIntoBudgetedSlots(args: { - channelEntries: [string, number][]; - threadMsgEntries: [string, number][]; - clientId: string; - initialSlotCount: number; - maxSlots: number; - maxBytes: number; - slotIdGenerator: () => string; -}): SlotSplitResult | null { - const { - channelEntries, - threadMsgEntries, - clientId, - initialSlotCount, - maxSlots, - maxBytes, - slotIdGenerator, - } = args; - - const encoder = new TextEncoder(); - const blobFor = (c: Record) => - JSON.stringify({ v: 1, client_id: clientId, contexts: c }); - - const overrideRawIds = new Set(); - for (const [key] of channelEntries) { - if (isOverrideKey(key)) overrideRawIds.add(key.slice(5)); - } - const pinnedEntries: [string, number][] = []; - const roundRobinEntries: [string, number][] = []; - for (const [key, ts] of channelEntries) { - if (isOverrideKey(key) || overrideRawIds.has(unescapeFrontierKey(key))) { - pinnedEntries.push([key, ts]); - } else { - roundRobinEntries.push([key, ts]); - } - } - let slotCount = initialSlotCount; - const extraSlotIds: string[] = []; - const distribute = (count: number): Array> => { - const slotContexts: Array> = Array.from( - { length: count }, - () => ({}), - ); - for (let i = 0; i < roundRobinEntries.length; i++) { - const [key, ts] = roundRobinEntries[i]; - slotContexts[i % count][key] = ts; - } - for (const [key, ts] of pinnedEntries) slotContexts[0][key] = ts; - return slotContexts; - }; - - let slotContexts = distribute(slotCount); - while ( - slotContexts.some((c) => encoder.encode(blobFor(c)).length > maxBytes) && - slotCount < maxSlots - ) { - extraSlotIds.push(slotIdGenerator()); - slotCount++; - slotContexts = distribute(slotCount); - } - if (slotContexts.some((c) => encoder.encode(blobFor(c)).length > maxBytes)) - return null; - for (const [key, ts] of threadMsgEntries) slotContexts[0][key] = ts; - trimContextsToBudget(slotContexts[0], clientId, maxBytes); - return { slots: slotContexts, extraSlotIds }; -} - -export interface TrimResult { - evicted: number; - fitsAfterTrim: boolean; -} - -/** Trim a contexts map to fit within `maxBytes`. Evicts oldest msg:/thread: entries; channel/ov_* keys never evicted. */ -export function trimContextsToBudget( - contexts: Record, - clientId: string, - maxBytes: number, -): TrimResult { - const encoder = new TextEncoder(); - const blobFor = (c: Record) => - JSON.stringify({ v: 1, client_id: clientId, contexts: c }); - - let currentBytes = encoder.encode(blobFor(contexts)).length; - if (currentBytes <= maxBytes) { - return { evicted: 0, fitsAfterTrim: true }; - } - - const msgEntries: [string, number][] = []; - const threadEntries: [string, number][] = []; - for (const [key, ts] of Object.entries(contexts)) { - if (isOverrideKey(key)) continue; - if (key.startsWith(MSG_PREFIX)) msgEntries.push([key, ts]); - else if (key.startsWith(THREAD_PREFIX)) threadEntries.push([key, ts]); - } - msgEntries.sort((a, b) => a[1] - b[1]); - threadEntries.sort((a, b) => a[1] - b[1]); - const toEvict: string[] = []; - for (const [key, ts] of [...msgEntries, ...threadEntries]) { - if (currentBytes <= maxBytes) break; - currentBytes -= key.length + 3 + String(ts).length + 1; - toEvict.push(key); - } - for (const key of toEvict) delete contexts[key]; - const fitsAfterTrim = encoder.encode(blobFor(contexts)).length <= maxBytes; - return { evicted: toEvict.length, fitsAfterTrim }; -} - export class ReadStateManager { - private pubkey: string; + readonly pubkey: string; private relayClient: RelayClient; - private clientId: string; + readonly clientId: string; private slotId: string; - private extraSlotIds: string[]; - private effectiveState = new Map(); - private publishableContextIds = new Set(); + extraSlotIds: string[]; + readonly effectiveState = new Map(); + readonly publishableContextIds = new Set(); private lastPublishedContexts: Record = {}; private debounceTimer: number | null = null; private listeners = new Set<() => void>(); @@ -283,15 +177,27 @@ export class ReadStateManager { private maxFetchedCreatedAt = 0; private contextSourceCreatedAt = new Map(); private pendingSyncedAdvances = new Set(); - private destroyed = false; + destroyed = false; private parentResolver: ContextParentResolver | null = null; - private overrideRegisters = new Map(); - private isLoadComplete = false; - private appliedReceipts = new Map(); // co-committed with register (Amendment C) - private loadGeneration = 0; // single-flight controller - private loadInFlight = false; - private retryBackoffTimer: number | null = null; - private drainScheduled = false; + readonly overrideRegisters = new Map(); + isLoadComplete = false; + readonly appliedReceipts = new Map(); // co-committed with register (Amendment C) + loadGeneration = 0; // single-flight controller + loadInFlight = false; + retryBackoffTimer: number | null = null; + retryAttempt = 0; // bounded exponential backoff attempt counter + pendingRetryOnComplete = false; // reconnect arrived during loadInFlight + drainScheduled = false; + /** Called after each drain intent is settled; wired by the hook for outcome routing. */ + onDrainOutcome: + | (( + channelId: string, + op: string, + status: string, + reason?: string, + sourceScope?: string, + ) => void) + | null = null; constructor(pubkey: string, relayClient: RelayClient) { this.pubkey = pubkey; @@ -304,7 +210,7 @@ export class ReadStateManager { generateHex(16), ); this.extraSlotIds = loadExtraSlotIds(pubkey); - pendingOverrideIntentStore.loadForPubkey(pubkey); + // pendingOverrideIntentStore is restored from v2 blob in hydrateFromLocalStorage. } async initialize(): Promise { @@ -313,7 +219,7 @@ export class ReadStateManager { this.loadInFlight = true; const gen = ++this.loadGeneration; try { - await this.fetchAndMerge(); + await this.fetchAndMerge(gen); } finally { if (gen === this.loadGeneration) this.loadInFlight = false; } @@ -395,6 +301,7 @@ export class ReadStateManager { this.destroyed = true; this.loadGeneration++; this.loadInFlight = false; + this.pendingRetryOnComplete = false; this.drainScheduled = false; if (this.retryBackoffTimer !== null) { window.clearTimeout(this.retryBackoffTimer); @@ -412,10 +319,10 @@ export class ReadStateManager { this.listeners.clear(); } - private async fetchAndMerge(): Promise { - if (this.destroyed) return; + async fetchAndMerge(gen: number = this.loadGeneration): Promise { + if (this.destroyed || gen !== this.loadGeneration) return; const result = await fencedEnumerationLoad(this.relayClient, this.pubkey); - if (this.destroyed) return; + if (this.destroyed || gen !== this.loadGeneration) return; this.isLoadComplete = result.complete; if (!result.complete) console.warn( @@ -424,47 +331,82 @@ export class ReadStateManager { this.processLoadedParsedEvents(result.events); const merged = mergeParsedEvents(result.events); await this.ingest(merged); + if (this.destroyed || gen !== this.loadGeneration) return; this.persistLocalState(); this.notifyListeners(); } - /** Retry load: generation-fenced, coalesces in-flight loads, resets backoff. */ + /** + * Retry load: generation-fenced, coalesces in-flight loads. + * Implements bounded exponential backoff (1s→2s→4s…capped at 60s). + * A reconnect during loadInFlight sets pendingRetryOnComplete so the next + * complete/incomplete verdict triggers exactly one fresh generation rather + * than dropping the reconnect signal. + */ async retryLoad(): Promise { if (this.destroyed) return; if (this.retryBackoffTimer !== null) { window.clearTimeout(this.retryBackoffTimer); this.retryBackoffTimer = null; } - if (this.loadInFlight) return; + if (this.loadInFlight) { + // Reconnect arrived while a load is in flight — coalesce into a required + // fresh generation rather than dropping the signal. + this.pendingRetryOnComplete = true; + return; + } + const delayMs = + this.retryAttempt === 0 + ? 0 + : Math.min(1_000 * 2 ** (this.retryAttempt - 1), 60_000); + if (delayMs > 0) { + await new Promise((resolve) => { + this.retryBackoffTimer = window.setTimeout(() => { + this.retryBackoffTimer = null; + resolve(); + }, delayMs); + }); + } + if (this.destroyed || this.loadInFlight) return; + this.retryAttempt++; this.loadInFlight = true; const gen = ++this.loadGeneration; try { this.isLoadComplete = false; - await this.fetchAndMerge(); + await this.fetchAndMerge(gen); if (this.destroyed || gen !== this.loadGeneration) return; - if (!this.drainScheduled) { + if (this.isLoadComplete) this.retryAttempt = 0; + if (!this.drainScheduled && this.isLoadComplete) { this.drainScheduled = true; void this.drainPendingIntents(gen); } } finally { if (gen === this.loadGeneration) { this.loadInFlight = false; + if (this.pendingRetryOnComplete && !this.destroyed) { + this.pendingRetryOnComplete = false; + void this.retryLoad(); + } } } } /** Drain pending override intents. See readStateDrain.ts for the full implementation. */ - private async drainPendingIntents(drainGen: number): Promise { - await drainPendingIntentsImpl(this as unknown as DrainContext, drainGen); + async drainPendingIntents(drainGen: number): Promise { + await drainPendingIntentsImpl(createDrainContext(this), drainGen); this.drainScheduled = false; } markChannelUnread(channelId: string): MarkResult { - return markChannelUnreadImpl(this as unknown as DrainContext, channelId); + return markChannelUnreadImpl(createDrainContext(this), channelId); } - markChannelRead(channelId: string): MarkResult { - return markChannelReadImpl(this as unknown as DrainContext, channelId); + markChannelRead(channelId: string, sourceScope?: string): MarkResult { + return markChannelReadImpl( + createDrainContext(this), + channelId, + sourceScope, + ); } /** Shared ingest: accept pre-merged state and return semantic delta (changed contexts, canonical flip). */ @@ -625,7 +567,7 @@ export class ReadStateManager { } } - private schedulePublish(): void { + schedulePublish(): void { if (this.debounceTimer !== null) window.clearTimeout(this.debounceTimer); this.debounceTimer = window.setTimeout(() => { this.debounceTimer = null; @@ -633,7 +575,7 @@ export class ReadStateManager { }, DEBOUNCE_MS); } - private async publish(): Promise { + async publish(): Promise { if (!this.isLoadComplete) return; if (!(await this.fetchOwnBlobBeforePublish())) { console.warn( @@ -710,7 +652,7 @@ export class ReadStateManager { await this.publishOneSlot(slotId, contexts); } - private async deleteExtraSlots(): Promise { + async deleteExtraSlots(): Promise { if (!this.isLoadComplete) return; for (const slotId of this.extraSlotIds) { try { @@ -733,7 +675,7 @@ export class ReadStateManager { saveExtraSlotIds(this.pubkey, []); } - private async fetchOwnBlobBeforePublish(): Promise { + async fetchOwnBlobBeforePublish(): Promise { const dTags = [this.slotId, ...this.extraSlotIds].map( (id) => `${READ_STATE_D_TAG_PREFIX}${id}`, ); @@ -773,7 +715,7 @@ export class ReadStateManager { } /** Effective frontier including parent resolver. */ - private channelFrontier(channelId: string): number { + channelFrontier(channelId: string): number { return ( resolveEffectiveTimestamp({ effectiveState: this.effectiveState, @@ -902,7 +844,7 @@ export class ReadStateManager { : saveExtraSlotIds(this.pubkey, this.extraSlotIds); } - private hydrateFromLocalStorage(): void { + hydrateFromLocalStorage(): void { const stored = readStoredReadState(this.pubkey); for (const [contextId, timestamp] of stored.contexts) this.effectiveState.set(contextId, timestamp); @@ -927,6 +869,11 @@ export class ReadStateManager { ); this.publishableContextIds.add(ctx); // v2 entry is inherently publishable } + // Restore pending intents from v2 blob (consolidation: no separate key). + pendingOverrideIntentStore.restoreFromStorage( + stored.pendingIntents, + stored.nextGen, + ); // Amendment C hydration: load receipts; orphan sweep for receipts with no matching intent. for (const [channelId, receipt] of stored.appliedReceipts) { const intent = pendingOverrideIntentStore.get(channelId); @@ -942,7 +889,7 @@ export class ReadStateManager { } /** Persist local state. Returns false only if the v2 override write failed (commit point). */ - private persistLocalState(): boolean { + persistLocalState(): boolean { const entries = new Map( [...this.overrideRegisters].map(([ctx, r]) => [ ctx, @@ -956,6 +903,8 @@ export class ReadStateManager { this.contextSourceCreatedAt, entries, this.appliedReceipts, + pendingOverrideIntentStore.all, + pendingOverrideIntentStore.nextGen, ); } @@ -965,7 +914,7 @@ export class ReadStateManager { return drained; } - private notifyListeners(): void { + notifyListeners(): void { for (const listener of this.listeners) { try { listener(); diff --git a/desktop/src/features/channels/readState/readStateOverride.test.mjs b/desktop/src/features/channels/readState/readStateOverride.test.mjs index d326b644aa..e33f5639f2 100644 --- a/desktop/src/features/channels/readState/readStateOverride.test.mjs +++ b/desktop/src/features/channels/readState/readStateOverride.test.mjs @@ -729,7 +729,7 @@ test("roundTrip_escapedRawCtx_survivesEncodeAndParse", () => { test("parseContexts_tombstoneFloor_multibyte_exactly256Bytes_accepted", () => { // 'é' × 125 = 250 bytes, + 'a' = 251 bytes suffix → wire key exactly 256 bytes. - const suffix = "\u00e9".repeat(125) + "a"; + const suffix = `${"\u00e9".repeat(125)}a`; const raw = { [suffix]: 0, [`${OV_C_PREFIX}${suffix}`]: 7, @@ -740,7 +740,7 @@ test("parseContexts_tombstoneFloor_multibyte_exactly256Bytes_accepted", () => { test("parseContexts_tombstoneFloor_multibyte_exactly257Bytes_dropped", () => { // 'é' × 125 = 250 bytes, + 'aa' = 252 bytes suffix → wire key exactly 257 bytes. - const suffix = "\u00e9".repeat(125) + "aa"; + const suffix = `${"\u00e9".repeat(125)}aa`; const raw = { [suffix]: 0, [`${OV_C_PREFIX}${suffix}`]: 7, @@ -751,7 +751,7 @@ test("parseContexts_tombstoneFloor_multibyte_exactly257Bytes_dropped", () => { test("parseContexts_liveGroup_multibyte_exactly256Bytes_accepted", () => { // Same suffix construction: 'é' × 125 + 'a' = 251 bytes → ov_s: wire key 256 bytes. - const suffix = "\u00e9".repeat(125) + "a"; + const suffix = `${"\u00e9".repeat(125)}a`; const raw = { [suffix]: 10, [`${OV_S_PREFIX}${suffix}`]: 2, @@ -764,7 +764,7 @@ test("parseContexts_liveGroup_multibyte_exactly256Bytes_accepted", () => { test("parseContexts_liveGroup_multibyte_exactly257Bytes_dropped", () => { // 'é' × 125 + 'aa' = 252 bytes → ov_s: wire key 257 bytes → group dropped. - const suffix = "\u00e9".repeat(125) + "aa"; + const suffix = `${"\u00e9".repeat(125)}aa`; const raw = { [suffix]: 10, [`${OV_S_PREFIX}${suffix}`]: 2, @@ -1020,9 +1020,9 @@ for (const [name, input, expected] of [ // 'g' and 'z' are not hexadecimal. ["nonHexChars_returnsFalse", "read-state:gggggggggggggggggggggggggggggggg", false], // One char short — must be exactly 32. - ["31HexChars_returnsFalse", "read-state:" + "a".repeat(31), false], + ["31HexChars_returnsFalse", `read-state:${"a".repeat(31)}`, false], // One char over — must be exactly 32. - ["33HexChars_returnsFalse", "read-state:" + "a".repeat(33), false], + ["33HexChars_returnsFalse", `read-state:${"a".repeat(33)}`, false], ["missingPrefix_returnsFalse", "a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6", false], ["emptySlot_returnsFalse", "read-state:", false], ["undefined_returnsFalse", undefined, false], diff --git a/desktop/src/features/channels/readState/readStateSlotUtils.ts b/desktop/src/features/channels/readState/readStateSlotUtils.ts new file mode 100644 index 0000000000..ad6bd3f5ff --- /dev/null +++ b/desktop/src/features/channels/readState/readStateSlotUtils.ts @@ -0,0 +1,127 @@ +/** + * Pure slot-splitting and trimming utilities for NIP-RS read-state publication. + * + * Extracted from readStateManager.ts to keep that file within the 1000-line + * size ratchet. Both functions are stateless and depend only on the + * readStateFormat constants and the TextEncoder API. + */ +import { + isOverrideKey, + unescapeFrontierKey, + MSG_PREFIX, + THREAD_PREFIX, +} from "@/features/channels/readState/readStateFormat"; + +/** Slot split result. */ +export interface SlotSplitResult { + slots: Array>; + extraSlotIds: string[]; +} + +export interface TrimResult { + evicted: number; + fitsAfterTrim: boolean; +} + +/** Partition channelEntries across slots; override groups pinned to slot 0. */ +export function splitContextsIntoBudgetedSlots(args: { + channelEntries: [string, number][]; + threadMsgEntries: [string, number][]; + clientId: string; + initialSlotCount: number; + maxSlots: number; + maxBytes: number; + slotIdGenerator: () => string; +}): SlotSplitResult | null { + const { + channelEntries, + threadMsgEntries, + clientId, + initialSlotCount, + maxSlots, + maxBytes, + slotIdGenerator, + } = args; + + const encoder = new TextEncoder(); + const blobFor = (c: Record) => + JSON.stringify({ v: 1, client_id: clientId, contexts: c }); + + const overrideRawIds = new Set(); + for (const [key] of channelEntries) { + if (isOverrideKey(key)) overrideRawIds.add(key.slice(5)); + } + const pinnedEntries: [string, number][] = []; + const roundRobinEntries: [string, number][] = []; + for (const [key, ts] of channelEntries) { + if (isOverrideKey(key) || overrideRawIds.has(unescapeFrontierKey(key))) { + pinnedEntries.push([key, ts]); + } else { + roundRobinEntries.push([key, ts]); + } + } + let slotCount = initialSlotCount; + const extraSlotIds: string[] = []; + const distribute = (count: number): Array> => { + const slotContexts: Array> = Array.from( + { length: count }, + () => ({}), + ); + for (let i = 0; i < roundRobinEntries.length; i++) { + const [key, ts] = roundRobinEntries[i]; + slotContexts[i % count][key] = ts; + } + for (const [key, ts] of pinnedEntries) slotContexts[0][key] = ts; + return slotContexts; + }; + + let slotContexts = distribute(slotCount); + while ( + slotContexts.some((c) => encoder.encode(blobFor(c)).length > maxBytes) && + slotCount < maxSlots + ) { + extraSlotIds.push(slotIdGenerator()); + slotCount++; + slotContexts = distribute(slotCount); + } + if (slotContexts.some((c) => encoder.encode(blobFor(c)).length > maxBytes)) + return null; + for (const [key, ts] of threadMsgEntries) slotContexts[0][key] = ts; + trimContextsToBudget(slotContexts[0], clientId, maxBytes); + return { slots: slotContexts, extraSlotIds }; +} + +/** Trim a contexts map to fit within `maxBytes`. Evicts oldest msg:/thread: entries; channel/ov_* keys never evicted. */ +export function trimContextsToBudget( + contexts: Record, + clientId: string, + maxBytes: number, +): TrimResult { + const encoder = new TextEncoder(); + const blobFor = (c: Record) => + JSON.stringify({ v: 1, client_id: clientId, contexts: c }); + + let currentBytes = encoder.encode(blobFor(contexts)).length; + if (currentBytes <= maxBytes) { + return { evicted: 0, fitsAfterTrim: true }; + } + + const msgEntries: [string, number][] = []; + const threadEntries: [string, number][] = []; + for (const [key, ts] of Object.entries(contexts)) { + if (isOverrideKey(key)) continue; + if (key.startsWith(MSG_PREFIX)) msgEntries.push([key, ts]); + else if (key.startsWith(THREAD_PREFIX)) threadEntries.push([key, ts]); + } + msgEntries.sort((a, b) => a[1] - b[1]); + threadEntries.sort((a, b) => a[1] - b[1]); + const toEvict: string[] = []; + for (const [key, ts] of [...msgEntries, ...threadEntries]) { + if (currentBytes <= maxBytes) break; + currentBytes -= key.length + 3 + String(ts).length + 1; + toEvict.push(key); + } + for (const key of toEvict) delete contexts[key]; + const fitsAfterTrim = encoder.encode(blobFor(contexts)).length <= maxBytes; + return { evicted: toEvict.length, fitsAfterTrim }; +} diff --git a/desktop/src/features/channels/readState/readStateStorage.test.mjs b/desktop/src/features/channels/readState/readStateStorage.test.mjs index e265d8cf2d..65dcddf557 100644 --- a/desktop/src/features/channels/readState/readStateStorage.test.mjs +++ b/desktop/src/features/channels/readState/readStateStorage.test.mjs @@ -99,6 +99,8 @@ test("writeStoredReadState prunes all three keys consistently", () => { ]), new Map(), new Map(), + new Map(), + 1, ); const state = JSON.parse( @@ -132,6 +134,8 @@ test("writeStoredReadState round-trips through readStoredReadState", () => { new Map([["channel-9", nowSeconds]]), new Map([["channel-9", { s: 3, c: 1, b: nowSeconds, f: nowSeconds }]]), new Map(), + new Map(), + 1, ); const stored = readStoredReadState(pubkey); @@ -162,6 +166,8 @@ test("writeStoredReadState_survives_throwing_localStorage_and_returns_false", () new Map([["channel-1", nowSeconds]]), new Map(), new Map(), + new Map(), + 1, ); }); assert.equal( diff --git a/desktop/src/features/channels/readState/readStateStorage.ts b/desktop/src/features/channels/readState/readStateStorage.ts index ed620b7b6e..7b93d12314 100644 --- a/desktop/src/features/channels/readState/readStateStorage.ts +++ b/desktop/src/features/channels/readState/readStateStorage.ts @@ -11,7 +11,10 @@ import { type OverrideRegister, } from "@/features/channels/readState/readStateFormat"; import { setLocalStorageItemWithRecovery } from "@/shared/lib/localStorageQuota"; -import type { PendingIntentOp } from "@/features/channels/pendingOverrideIntents"; +import type { + PendingIntent, + PendingIntentOp, +} from "@/features/channels/pendingOverrideIntents"; export type StoredReadState = { contexts: Map; @@ -19,6 +22,10 @@ export type StoredReadState = { contextSourceCreatedAt: Map; overrideRegisters: Map; appliedReceipts: Map; + /** Pending intents from the v2 blob (consolidated from pending-intents.v1). */ + pendingIntents: Map; + /** Next generation counter for the pending intent queue. */ + nextGen: number; }; /** @@ -124,6 +131,8 @@ export function readStoredReadState(pubkey: string): StoredReadState { contextSourceCreatedAt: readContextSourceCreatedAt(pubkey), overrideRegisters: overrideState.registers, appliedReceipts: overrideState.receipts, + pendingIntents: overrideState.pendingIntents, + nextGen: overrideState.nextGen, }; } @@ -149,11 +158,15 @@ export type StoredOverrideEntry = OverrideRegister & { f: number }; type OverrideStateBlob = { registers: Map; receipts: Map; + pendingIntents: Map; + nextGen: number; }; function readOverrideState(pubkey: string): OverrideStateBlob { const registers = new Map(); const receipts = new Map(); + const pendingIntents = new Map(); + let nextGen = 1; const raw = localStorage.getItem(localOverrideStateKey(pubkey)); if (raw) { try { @@ -162,8 +175,9 @@ function readOverrideState(pubkey: string): OverrideStateBlob { // Registers are stored under the "r" key (or top-level for compat). const registersObj = isPlainRecord(parsed.r) ? parsed.r : parsed; for (const [rawCtx, value] of Object.entries(registersObj)) { - // Skip the reserved "receipts" key. - if (rawCtx === "receipts") continue; + // Skip reserved top-level keys. + if (rawCtx === "receipts" || rawCtx === "pi" || rawCtx === "ng") + continue; if (!isPlainRecord(value)) continue; const { s, c, b, f } = value; if ( @@ -203,20 +217,55 @@ function readOverrideState(pubkey: string): OverrideStateBlob { receipts.set(channelId, { intentGen, op }); } } + // Parse pending intents from the "pi" sub-object. + if (isPlainRecord(parsed.pi)) { + for (const [channelId, entry] of Object.entries(parsed.pi)) { + if (!isPlainRecord(entry)) continue; + const { gen, op, sourceScope, readTarget } = entry as { + gen?: unknown; + op?: unknown; + sourceScope?: unknown; + readTarget?: unknown; + }; + if (typeof gen !== "number" || !Number.isInteger(gen) || gen < 1) + continue; + if (op !== "unread" && op !== "read") continue; + if (sourceScope !== undefined && typeof sourceScope !== "string") + continue; + if (readTarget !== undefined && typeof readTarget !== "number") + continue; + const intent: PendingIntent = { + gen, + op, + ...(sourceScope !== undefined ? { sourceScope } : {}), + ...(readTarget !== undefined ? { readTarget } : {}), + }; + pendingIntents.set(channelId, intent); + } + } + // Parse nextGen. + if ( + typeof parsed.ng === "number" && + Number.isInteger(parsed.ng) && + parsed.ng >= 1 + ) { + nextGen = parsed.ng; + } } } catch { // Corrupt storage — return empty; repopulated on next ingest. } - return { registers, receipts }; + return { registers, receipts, pendingIntents, nextGen }; } // Migration: read from legacy v1 key if v2 key absent. const legacyRaw = localStorage.getItem( localOverrideRegistersKeyLegacy(pubkey), ); - if (!legacyRaw) return { registers, receipts }; + if (!legacyRaw) return { registers, receipts, pendingIntents, nextGen }; try { const parsed = JSON.parse(legacyRaw); - if (!isPlainRecord(parsed)) return { registers, receipts }; + if (!isPlainRecord(parsed)) + return { registers, receipts, pendingIntents, nextGen }; for (const [rawCtx, value] of Object.entries(parsed)) { if (!isPlainRecord(value)) continue; const { s, c, b } = value; @@ -239,7 +288,7 @@ function readOverrideState(pubkey: string): OverrideStateBlob { } catch { // Corrupt legacy storage — ignore. } - return { registers, receipts }; + return { registers, receipts, pendingIntents, nextGen }; } /** @@ -282,6 +331,8 @@ export function writeStoredReadState( contextSourceCreatedAt: ReadonlyMap, overrideRegisters: ReadonlyMap, appliedReceipts: ReadonlyMap, + pendingIntents: ReadonlyMap, + nextGen: number, ): boolean { const pruned = pruneStaleContexts(contexts, Math.floor(Date.now() / 1_000)); @@ -290,17 +341,19 @@ export function writeStoredReadState( state[contextId] = new Date(timestamp * 1_000).toISOString(); } - // Persist override registers atomically with their frontier timestamps and - // applied receipts (Amendment C) in one JSON blob — a single write ensures - // they are never torn. The receipt lives alongside the register it proves - // was applied, so "register committed ↔ receipt committed" is a storage - // invariant. + // Persist override registers, applied receipts, and pending intents atomically + // in one JSON blob — a single write ensures they are never torn. // - // Format: { r: { : {s,c,b,f}, ... }, receipts: { : {intentGen,op}, ... } } + // Format: { + // r: { : {s,c,b,f} }, + // receipts: { : {intentGen, op} }, // omitted when empty + // pi: { : {gen, op, sourceScope?, readTarget?} }, // omitted when empty + // ng: , // next generation counter; omitted when 1 + // } // - // This is the ACTION COMMIT POINT: if ok4 is true the action is durably - // committed; ancillary frontier/cache write failures (ok1-ok3) do not fail - // the action. If ok4 is false, the caller must roll back ALL state. + // This is the ACTION + INTENT COMMIT POINT: if ok4 is true, the action, + // receipt, and pending intent state are all durably committed together. + // If ok4 is false, the caller must roll back ALL state. const registersObj: Record< string, { s: number; c: number; b: number; f: number } @@ -315,11 +368,37 @@ export function writeStoredReadState( for (const [channelId, receipt] of appliedReceipts) { receiptsObj[channelId] = { intentGen: receipt.intentGen, op: receipt.op }; } + const piObj: Record< + string, + { + gen: number; + op: PendingIntentOp; + sourceScope?: string; + readTarget?: number; + } + > = {}; + for (const [channelId, intent] of pendingIntents) { + const entry: { + gen: number; + op: PendingIntentOp; + sourceScope?: string; + readTarget?: number; + } = { + gen: intent.gen, + op: intent.op, + }; + if (intent.sourceScope !== undefined) + entry.sourceScope = intent.sourceScope; + if (intent.readTarget !== undefined) entry.readTarget = intent.readTarget; + piObj[channelId] = entry; + } const ok4 = setLocalStorageItemWithRecovery( localOverrideStateKey(pubkey), JSON.stringify({ r: registersObj, ...(Object.keys(receiptsObj).length > 0 ? { receipts: receiptsObj } : {}), + ...(Object.keys(piObj).length > 0 ? { pi: piObj } : {}), + ...(nextGen > 1 ? { ng: nextGen } : {}), }), ); diff --git a/desktop/src/features/channels/readState/useReadState.ts b/desktop/src/features/channels/readState/useReadState.ts index 9d3d114eab..51a5abd139 100644 --- a/desktop/src/features/channels/readState/useReadState.ts +++ b/desktop/src/features/channels/readState/useReadState.ts @@ -18,6 +18,7 @@ const noopMarkOverride = (): MarkResult => ({ }); const noopGetOverrideLiveness = (): OverrideLiveness | null => null; const noopGetProjection = (): ReadStateProjection | null => null; +const noopSetOnDrainOutcome = (): void => {}; /** * React hook that creates and manages a ReadStateManager instance. @@ -115,14 +116,17 @@ export function useReadState( [], ); - const markChannelRead = React.useCallback((channelId: string): MarkResult => { - return ( - managerRef.current?.markChannelRead(channelId) ?? { - status: "refused", - reason: "load_incomplete", - } - ); - }, []); + const markChannelRead = React.useCallback( + (channelId: string, sourceScope?: string): MarkResult => { + return ( + managerRef.current?.markChannelRead(channelId, sourceScope) ?? { + status: "refused", + reason: "load_incomplete", + } + ); + }, + [], + ); const getOverrideLiveness = React.useCallback( (channelId: string): OverrideLiveness | null => { @@ -135,6 +139,27 @@ export function useReadState( return managerRef.current?.getProjection() ?? null; }, []); + // Wire the drain outcome callback into the current manager. + // Called by useUnreadChannels to route rollback/cleanup through the hook layer. + const setOnDrainOutcome = React.useCallback( + ( + callback: + | (( + channelId: string, + op: string, + status: string, + reason?: string, + sourceScope?: string, + ) => void) + | null, + ): void => { + if (managerRef.current) { + managerRef.current.onDrainOutcome = callback; + } + }, + [], + ); + const isReady = Boolean( pubkey && relayClient && initializedPubkey === pubkey, ); @@ -160,6 +185,7 @@ export function useReadState( markChannelRead: noopMarkOverride, getOverrideLiveness: noopGetOverrideLiveness, getProjection: noopGetProjection, + setOnDrainOutcome: noopSetOnDrainOutcome, }; } @@ -177,5 +203,6 @@ export function useReadState( markChannelRead, getOverrideLiveness, getProjection, + setOnDrainOutcome, }; } diff --git a/desktop/src/features/channels/readStateOverride.ts b/desktop/src/features/channels/readStateOverride.ts index ace260d801..cb22fde328 100644 --- a/desktop/src/features/channels/readStateOverride.ts +++ b/desktop/src/features/channels/readStateOverride.ts @@ -52,7 +52,7 @@ export type OverrideAPIs = { * in a partial view", not "known absent register". */ isReadStateReady: boolean; markChannelUnread: (channelId: string) => MarkResult; - markChannelRead: (channelId: string) => MarkResult; + markChannelRead: (channelId: string, sourceScope?: string) => MarkResult; getOverrideLiveness: (channelId: string) => OverrideLiveness | null; }; @@ -115,10 +115,15 @@ export type OverrideReadOutcome = "overrideCleared" | "overrideStillActive"; * no error toast (representable C-bump was not possible, frontier * already provides the inactive verdict). * c. Other refusal keeping liveness active: show toast, overrideStillActive. + * + * `sourceScope` is forwarded to markChannelRead so the intent can carry the + * exact source being cleared; the drain surfaces it to onDrainOutcome for + * exact hook-layer source cleanup (rather than whole-entry deletion). */ export function applyOverrideRead( channelId: string, apis: OverrideAPIs, + sourceScope?: string, ): OverrideReadOutcome { // Step 1: load complete + no register → known absence, cleared. // Guard: only treat null as "known absent" when the load is complete. @@ -133,7 +138,7 @@ export function applyOverrideRead( // register or frontier mutation. Sources are preserved; the intent drains // on the next complete-load transition and rawUnread suppresses presentation // via the pending-read precedence rule. - const result = apis.markChannelRead(channelId); + const result = apis.markChannelRead(channelId, sourceScope); if (result.status === "queued") { // Intent enqueued — sources preserved, rawUnread handles presentation. @@ -206,7 +211,7 @@ export function useClearChannelUnreadSource( return; } // Last-source removal: route through the mark-read seam. - const outcome = applyOverrideRead(channelId, apis); + const outcome = applyOverrideRead(channelId, apis, source); if (outcome === "overrideCleared") { delete forcedUnreadRef.current[channelId]; if (pubkey) forcedUnreadStore.write(pubkey, forcedUnreadRef.current); diff --git a/desktop/src/features/channels/useUnreadChannels.ts b/desktop/src/features/channels/useUnreadChannels.ts index e9e9027a3e..ff6f23a91c 100644 --- a/desktop/src/features/channels/useUnreadChannels.ts +++ b/desktop/src/features/channels/useUnreadChannels.ts @@ -61,6 +61,7 @@ import { mutedStore, resolveChannelReadMarker, resolveObservedUnreadRootId, + useDrainOutcomeCallback, } from "@/features/channels/useUnreadChannelsHelpers"; export function useUnreadChannels( @@ -82,10 +83,8 @@ export function useUnreadChannels( const normalizedRelayUrl = relayUrlOption ? normalizeRelayUrl(relayUrlOption) : ""; - // Single identity for the in-memory thread-activity buffer — computed once - // per render and used at reset, both writers, and the return fence. The - // helper returns "" when either value is absent, which never matches a valid - // loaded scope, so the fence returns [] until the buffer is seeded. + // Single identity for the in-memory thread-activity buffer — reset/return fence; + // empty string never matches a valid loaded scope, so the fence returns [] until seeded. const currentActivityScope = activityScopeKey( normalizedPubkey, normalizedRelayUrl, @@ -104,6 +103,7 @@ export function useUnreadChannels( markChannelRead: markChannelOverrideRead, getOverrideLiveness, getProjection, + setOnDrainOutcome, } = useReadState(pubkey, relayClient); const overrideApis = React.useMemo( () => ({ @@ -159,17 +159,12 @@ export function useUnreadChannels( }, [readStateVersion, drainSyncedAdvances]); // Root event IDs of threads where the current user has replied at least once. - // Used to determine if thread replies should trigger unread notifications. const participatedRootIdsRef = React.useRef(new Set()); // Root event IDs of top-level messages authored by the current user. - // Used to notify the author when someone replies to their posts. const authoredRootIdsRef = React.useRef(new Set()); - // Root event IDs of threads where an external message @-mentioned the user. - // ORed into the badge gate so a mention recipient who never participated, - // authored, or followed the thread still gets the thread-unread badge. + // Root event IDs of threads @-mentioning the user — ORed into the badge gate. const mentionedRootIdsRef = React.useRef(new Set()); - // Root event IDs of threads the user has explicitly muted. Takes precedence - // over participation, follow, and authorship for notification suppression. + // Root event IDs of threads the user muted — overrides participation/follow. const mutedRootIdsRef = React.useRef(new Set()); // Stable ref for the caller-supplied muted channel IDs. Updated every render @@ -177,12 +172,9 @@ export function useUnreadChannels( const mutedChannelIdsRef = React.useRef>(new Set()); mutedChannelIdsRef.current = mutedChannelIdsOption ?? new Set(); - // Thread reply events that triggered notifications — surfaced in the Home - // activity feed as synthetic FeedItems. + // Thread reply events that triggered notifications — surfaced in the Home activity feed. const threadActivityRef = React.useRef([]); - // Tracks the (pubkey:relayUrl) scope currently loaded into threadActivityRef. - // Writers guard against this before merging so in-flight writes from a prior - // scope cannot corrupt the new one; renders return [] until it matches. + // Scope key for threadActivityRef; guards against cross-scope writes. const threadActivityScopeRef = React.useRef(""); // Tracks which channels we've already issued a catch-up REQ for this @@ -204,6 +196,14 @@ export function useUnreadChannels( 0, ); + // Wire drain outcome callback; returns the pending-unread-snapshots ref. + const pendingUnreadSnapshotsRef = useDrainOutcomeCallback( + setOnDrainOutcome, + forcedUnreadRef, + pubkey, + bumpLatestVersion, + ); + // Persistence layer: hydration, pagehide flush, scope fence, write-through, marker-prune. const observedPersistence = useObservedUnreadPersistence( normalizedPubkey, @@ -313,20 +313,26 @@ export function useUnreadChannels( // Mark channel unread: write forced entry first (write ordering), then apply override. // Refused results undo the forced entry. Returns true when accepted (applied or queued). + // biome-ignore lint/correctness/useExhaustiveDependencies: pendingUnreadSnapshotsRef is a stable ref; .current.set is Map.prototype.set (stable identity) const markChannelUnread = React.useCallback( (channelId: string, source?: ForcedUnreadSource): boolean => { - // Step 1: write the forced entry first (write ordering). + // Snapshot the prior entry before the optimistic write for drain rollback. + const priorEntry = Object.hasOwn(forcedUnreadRef.current, channelId) + ? forcedUnreadRef.current[channelId] + : undefined; + // Write forced entry first (write ordering), then apply override. forcedMarkChannelUnread(channelId, source); - // Step 2: apply override (may enqueue intent or apply immediately). if (!applyOverrideUnread(channelId, overrideApis)) { - // Refused — undo the forced entry. - if (Object.hasOwn(forcedUnreadRef.current, channelId)) { - delete forcedUnreadRef.current[channelId]; - if (pubkey) forcedUnreadStore.write(pubkey, forcedUnreadRef.current); - bumpLatestVersion(); - } + // Refused immediately — restore exact prior entry (or remove if none existed). + if (priorEntry !== undefined) + forcedUnreadRef.current[channelId] = priorEntry; + else delete forcedUnreadRef.current[channelId]; + if (pubkey) forcedUnreadStore.write(pubkey, forcedUnreadRef.current); + bumpLatestVersion(); return false; } + // Store snapshot; drain will call onDrainOutcome to commit or rollback. + pendingUnreadSnapshotsRef.current.set(channelId, priorEntry); return true; }, [forcedMarkChannelUnread, overrideApis, pubkey], @@ -990,10 +996,8 @@ export function useUnreadChannels( markChannelRead, markChannelUnread, clearChannelUnreadSource, - // Exposed so other surfaces (e.g. Home) can project per-item read state - // off the same NIP-RS read marker without instantiating a second - // ReadStateManager. readStateVersion is the invalidation signal callers - // should include in memo deps. + // Exposed for other surfaces (e.g. Home) to project read state off the same + // NIP-RS marker; readStateVersion is the invalidation signal for memo deps. getEffectiveTimestamp, getOwnTimestamp, readStateVersion, diff --git a/desktop/src/features/channels/useUnreadChannelsHelpers.ts b/desktop/src/features/channels/useUnreadChannelsHelpers.ts index 1b8455c992..c376a60436 100644 --- a/desktop/src/features/channels/useUnreadChannelsHelpers.ts +++ b/desktop/src/features/channels/useUnreadChannelsHelpers.ts @@ -3,6 +3,7 @@ * useUnreadChannels to keep that file within the repository size ratchet. * All exports are load-bearing for useUnreadChannels.ts. */ +import * as React from "react"; import type { UseLiveChannelUpdatesOptions } from "@/features/channels/useLiveChannelUpdates"; import { makeRootIdStore } from "@/features/channels/unreadRootIdStore"; import { @@ -13,6 +14,13 @@ import type { RelayClient } from "@/shared/api/relayClientSession"; import type { Channel } from "@/shared/api/types"; import { CHANNEL_MESSAGE_EVENT_KINDS } from "@/shared/constants/kinds"; import { DM_NOTIFIABLE_EVENT_KINDS } from "./isDmNotifiableKind"; +import { + forcedUnreadStore, + removeForcedUnreadSource, + type ForcedUnreadEntry, + type ForcedUnreadMap, + type ForcedUnreadSource, +} from "@/features/channels/forcedUnreadStore"; export type UseUnreadChannelsOptions = UseLiveChannelUpdatesOptions & { pubkey?: string; @@ -87,3 +95,84 @@ export function resolveChannelReadMarker( export function resolveObservedUnreadRootId(tags: string[][]): string | null { return isBroadcastReply(tags) ? null : getThreadReference(tags).rootId; } + +/** + * Hook that builds and wires the drain outcome callback into the ReadStateManager + * via `setOnDrainOutcome`. Creates and owns the pending-unread-snapshots map. + * Returns the snapshots ref for use in markChannelUnread. + * + * Handles: + * - `unread + refused` → restore forced-unread entry to its pre-mark snapshot + * - `unread + applied` → discard snapshot (intent succeeded) + * - `read + applied` → remove forced-unread entry (or specific source), mirroring + * the synchronous overrideCleared path + * + * Genuine refusal toasts are fired in the drain itself and not repeated here. + */ +export function useDrainOutcomeCallback( + setOnDrainOutcome: ( + cb: + | (( + channelId: string, + op: string, + status: string, + reason?: string, + sourceScope?: string, + ) => void) + | null, + ) => void, + forcedUnreadRef: React.MutableRefObject, + pubkey: string | undefined, + bumpLatestVersion: () => void, +): React.MutableRefObject> { + const pendingUnreadSnapshotsRef = React.useRef( + new Map(), + ); + // biome-ignore lint/correctness/useExhaustiveDependencies: stable refs; setOnDrainOutcome is stable + React.useEffect(() => { + setOnDrainOutcome((channelId, op, status, _reason, sourceScope) => { + if (op === "unread" && status === "refused") { + // Roll back the optimistic forced-unread entry to its exact prior state. + const prior = pendingUnreadSnapshotsRef.current.get(channelId); + pendingUnreadSnapshotsRef.current.delete(channelId); + if (prior === undefined) { + if (Object.hasOwn(forcedUnreadRef.current, channelId)) { + delete forcedUnreadRef.current[channelId]; + if (pubkey) + forcedUnreadStore.write(pubkey, forcedUnreadRef.current); + bumpLatestVersion(); + } + } else { + forcedUnreadRef.current[channelId] = prior; + if (pubkey) forcedUnreadStore.write(pubkey, forcedUnreadRef.current); + bumpLatestVersion(); + } + } else if (op === "unread" && status === "applied") { + pendingUnreadSnapshotsRef.current.delete(channelId); + } else if (op === "read" && status === "applied") { + if (Object.hasOwn(forcedUnreadRef.current, channelId)) { + if (sourceScope !== undefined) { + const current = forcedUnreadRef.current[channelId]; + const next = removeForcedUnreadSource( + current, + sourceScope as ForcedUnreadSource, + ); + if (next !== undefined) { + forcedUnreadRef.current[channelId] = next; + } else { + delete forcedUnreadRef.current[channelId]; + } + } else { + delete forcedUnreadRef.current[channelId]; + } + if (pubkey) forcedUnreadStore.write(pubkey, forcedUnreadRef.current); + bumpLatestVersion(); + } + } + }); + return () => { + setOnDrainOutcome(null); + }; + }, [pubkey, setOnDrainOutcome]); + return pendingUnreadSnapshotsRef; +} diff --git a/desktop/src/shared/styles/globals/terminal.css b/desktop/src/shared/styles/globals/terminal.css index 26b5f42ba8..47a5a88925 100644 --- a/desktop/src/shared/styles/globals/terminal.css +++ b/desktop/src/shared/styles/globals/terminal.css @@ -248,7 +248,7 @@ } .buzz-terminal-substrate[data-terminal-visible="false"] { - height: 0 !important; + height: 0; min-height: 0; pointer-events: none; transform: translateY(16px); diff --git a/docs/nips/NIP-RS.md b/docs/nips/NIP-RS.md index 9d0b49b3ec..1abd5d7a07 100644 --- a/docs/nips/NIP-RS.md +++ b/docs/nips/NIP-RS.md @@ -536,13 +536,20 @@ Every action below requires a complete full-state load (see Full-State Load). On - (a) **`applied`** — the action was applied against a complete full-state load; the register was mutated, the canonical form was updated, and publication was scheduled. - (b) **`queued`** — the action was accepted as a pending intent; the register was NOT mutated; a forced local presentation MAY be shown optimistically; the client MUST drain the intent on the next complete-load transition and surface the definitive `applied` or `refused` result then. -- (c) **`refused`** — the action was rejected; neither the register nor local presentation was modified; a reason MUST be provided (see reasons below). +- (c) **`refused`** — the action was rejected; neither the register nor local presentation was modified; a reason MUST be provided (one of the refusal reasons below). + +**Refusal reasons:** + +- `load_incomplete` — the full-state load was not proven complete; no register mutation was performed. +- `already_inactive` — the override register was already inactive (or absent) at the time of the mark-read; no C-bump was needed. This is a definitive success condition: the channel is already cleared. +- `uint32_overflow` — `max(S, C)` has reached the uint32 maximum (4294967295); no representable counter increment exists. +- `budget_exhausted` — the new register would exceed the encoding budget; the client MUST refuse rather than silently drop other channels' overrides. A client that does not implement pending intents MUST report the action as `refused` with reason `load_incomplete` rather than act on a partial view of its own override state. **Mark-unread:** increment S to `max(S, C) + 1`; set B to the current effective frontier value for the context. C is unchanged. If `max(S, C) == 4294967295` (uint32 maximum), the client MUST refuse the mark-unread action and leave the register unchanged; wrapping or resetting to zero is prohibited. -**Mark-read (explicit):** advance the frontier to cover the context as normal; increment C to `max(S, C) + 1`. S and B are unchanged. If `max(S, C) == 4294967295`, no representable counter increment exists; wrapping or resetting to zero is prohibited. The client MUST then complete the action only if the resulting state satisfies `override_active == false` — i.e. the frontier advance alone deactivates the override, or the override was already inactive. Otherwise the counters MUST be left unchanged and the client MUST NOT report the mark-read as `applied` while `override_active` remains true; instead the client MUST report the outcome as `refused` (see reason `already_inactive` and other reasons above). A client MAY report `queued` for a mark-read on an incomplete load per (b) above, but MUST NOT apply the counter mutation until a complete load is available. +**Mark-read (explicit):** advance the frontier to cover the context as normal; increment C to `max(S, C) + 1`. S and B are unchanged. If `max(S, C) == 4294967295`, no representable counter increment exists; wrapping or resetting to zero is prohibited. The client MUST then complete the action only if the resulting state satisfies `override_active == false` — i.e. the frontier advance alone deactivates the override, or the override was already inactive. Otherwise the counters MUST be left unchanged and the client MUST NOT report the mark-read as `applied` while `override_active` remains true; instead the client MUST report the outcome as `refused` with reason `already_inactive`. A client MAY report `queued` for a mark-read on an incomplete load per (b) above, but MUST NOT apply the counter mutation until a complete load is available. **Natural read (frontier advance):** advance the frontier past B. No counter update is needed — the liveness predicate's `F <= B` condition automatically deactivates the override when the frontier dominates the baseline.