diff --git a/desktop/src/features/sidebar/lib/channelMutesSync.ts b/desktop/src/features/sidebar/lib/channelMutesSync.ts index 7ad9fd8620..3966fb6136 100644 --- a/desktop/src/features/sidebar/lib/channelMutesSync.ts +++ b/desktop/src/features/sidebar/lib/channelMutesSync.ts @@ -1,10 +1,3 @@ -import { relayClient } from "@/shared/api/relayClient"; -import { - nip44DecryptFromSelf, - nip44EncryptToSelf, - signRelayEvent, -} from "@/shared/api/tauri"; -import type { RelayEvent } from "@/shared/api/types"; import { KIND_CHANNEL_MUTES } from "@/shared/constants/kinds"; import { clearChannelMutesOutbox, @@ -15,464 +8,81 @@ import { type ChannelMuteStore, } from "./channelMutesStorage"; import { - advanceWatermark, - clampPublishCreatedAt, - readWatermark, - runBootstrap, - type FetchResult, -} from "./sidebarSyncWatermark"; + MergeLaneSyncManager, + type RemoteMergeBlob, +} from "./mergeLaneSyncManager"; -const D_TAG = "channel-mutes"; -const BLOB_TYPE = D_TAG; -const DEBOUNCE_MS = 2_000; +export type RemoteMutes = RemoteMergeBlob; -// Bounded backoff for a retained pending edit whose publish failed transiently -// (timeout / socket error) on an otherwise-healthy socket, so it does not wait -// for a reconnect that may never fire. -const RETRY_BASE_MS = 2_000; -const RETRY_MAX_MS = 30_000; - -export type RemoteMutes = { - store: ChannelMuteStore; - createdAt: number; - eventId: string; -}; - -/** - * Pre-publish decision for the merge lane. `publish` carries the max-merged - * store to write; `retain` keeps the durable pending edit and retries. Retain - * covers two unreadable cases where publishing would blindly overwrite state we - * could not inspect (Carl P1): a pre-publish fetch that THREW (timeout / auth / - * socket — not proof no head exists), and an existing head that failed - * decryption/parsing (a max-merge is only safe once both operands were read). - */ -type MergePublishDecision = - | { kind: "publish"; store: ChannelMuteStore } - | { kind: "retain" }; - -async function decryptAndParse(event: RelayEvent): Promise { - try { - const plaintext = await nip44DecryptFromSelf(event.content); - const store = parseMutePayload(JSON.parse(plaintext)); - if (!store) return null; - return { store, createdAt: event.created_at, eventId: event.id }; - } catch { - return null; - } -} - -export class ChannelMuteSyncManager { - private pubkey: string; - private relayUrl: string; - private debounceTimer: number | null = null; - private retryTimer: number | null = null; - private retryDelayMs = RETRY_BASE_MS; - private lastRemoteCreatedAt: number; - private pendingStore: ChannelMuteStore | null = null; - // Monotonic id for the current pending edit. Every publishMutes() bumps it; - // every scheduled publish/retry captures the value it was queued for. A - // completion (success or no-op) may only clear pending state via - // compare-and-swap on this generation, so an older in-flight publish can - // never erase a newer edit that arrived while it was in flight. - private pendingGeneration = 0; - // Publish cycles are serialized: at most one runs at a time. A newer edit - // queued while a cycle is in flight defers; the in-flight cycle's completion - // re-drives it. Serialization guarantees there is never more than one - // fetch/publish sequence touching shared manager state. - private publishInFlight = false; - private lastPublishedStore: ChannelMuteStore | null = null; - private destroyed = false; - // Per-channel high-water of every `rev` and `updatedAt` this manager has - // observed (bootstrap, live, reconnect, reconcile, pre-publish, cross-window - // storage, and initial persisted state). A click reads both so its minted - // `updatedAt = max(now, maxUpdatedAtSeen)` never regresses below observed - // state (the read-state logical-monotonic idiom), and `rev = maxRevSeen + 1` - // wins the resulting same-second tie. - private highWater = new Map(); - - constructor(pubkey: string, relayUrl: string) { - this.pubkey = pubkey; - this.relayUrl = relayUrl; - this.lastRemoteCreatedAt = readWatermark(pubkey, BLOB_TYPE, relayUrl); - } - - /** - * Ingest a store into the per-channel high-water. Called synchronously before - * any merge is applied to React state, so a click that follows reads a current - * watermark on both dimensions. Monotonic (`Math.max`) and idempotent. - */ - observe(store: ChannelMuteStore): void { - for (const [id, entry] of Object.entries(store.channels)) { - const cur = this.highWater.get(id) ?? { rev: 0, updatedAt: 0 }; - this.highWater.set(id, { - rev: Math.max(cur.rev, entry.rev), - updatedAt: Math.max(cur.updatedAt, entry.updatedAt), - }); - } - } - - maxRevSeen(id: string): number { - return this.highWater.get(id)?.rev ?? 0; - } - - maxUpdatedAtSeen(id: string): number { - return this.highWater.get(id)?.updatedAt ?? 0; - } - - async fetchRemoteMutes(): Promise> { - try { - const events = await relayClient.fetchEvents({ - kinds: [KIND_CHANNEL_MUTES], - authors: [this.pubkey], - "#d": [D_TAG], - limit: 1, - }); - if (events.length === 0 || events[0].pubkey !== this.pubkey) { - return { status: "absent" }; - } - const event = events[0]; - this.recordRemoteHead(event.created_at); - const result = await decryptAndParse(event); - if (!result) { - return { status: "failed", createdAt: event.created_at }; - } - this.observe(result.store); - return { - status: "found", - data: result, - createdAt: result.createdAt, - eventId: result.eventId, - }; - } catch { - return { status: "failed" }; - } - } - - private recordRemoteHead(createdAt: number): void { - if (createdAt > this.lastRemoteCreatedAt) { - this.lastRemoteCreatedAt = createdAt; - } - advanceWatermark(this.pubkey, BLOB_TYPE, this.relayUrl, createdAt); - } - - cancelPendingMutePublish(): void { - if (this.debounceTimer !== null) { - window.clearTimeout(this.debounceTimer); - this.debounceTimer = null; - } - if (this.retryTimer !== null) { - window.clearTimeout(this.retryTimer); - this.retryTimer = null; - } - } - - getPendingMuteStore(): ChannelMuteStore | null { - return this.pendingStore; - } - - publishMutes(store: ChannelMuteStore): void { - this.pendingStore = store; - ++this.pendingGeneration; - // Persist synchronously so a click made <2s before quit/community-switch - // survives teardown and resumes on next mount (durable outbox). This - // window's own key is the only one written — a single unconditional - // setItem, never a shared-key read-modify-write. - writeChannelMutesOutbox(this.pubkey, store, this.relayUrl); - if (this.debounceTimer !== null) { - window.clearTimeout(this.debounceTimer); - } - // A fresh edit supersedes any retry scheduled for the previous generation. - if (this.retryTimer !== null) { - window.clearTimeout(this.retryTimer); - this.retryTimer = null; - } - this.retryDelayMs = RETRY_BASE_MS; - this.debounceTimer = window.setTimeout(() => { - this.debounceTimer = null; - this.startCycle(); - }, DEBOUNCE_MS); - } - - /** - * Serialize publish cycles: at most one runs at a time. A debounce/retry timer - * that fires while a cycle is in flight defers — the in-flight cycle's - * completion re-drives if a pending edit still needs publishing. A newer edit - * queued during a cycle cannot start its own concurrent cycle, so a stale - * generation can never publish after a newer edit exists. - */ - private startCycle(): void { - if (this.destroyed || this.pendingStore === null) return; - if (this.publishInFlight) return; - const store = this.pendingStore; - const gen = this.pendingGeneration; - this.publishInFlight = true; - void this.doPublish(store, gen).finally(() => { - this.publishInFlight = false; - if ( - !this.destroyed && - this.pendingStore !== null && - this.debounceTimer === null && - this.retryTimer === null - ) { - this.startCycle(); - } - }); - } - - private async fetchOwnBlobBeforePublish( - store: ChannelMuteStore, - ): Promise { - let events: RelayEvent[]; - try { - events = await relayClient.fetchEvents({ - kinds: [KIND_CHANNEL_MUTES], - authors: [this.pubkey], - "#d": [D_TAG], - limit: 1, - }); - } catch { - // The fetch itself failed (timeout / auth / socket) — NOT proof that no - // head exists. Publishing the local store here could erase an unseen - // newer head during a transient outage, so retain and retry (Carl P1). - return { kind: "retain" }; - } - // A successful fetch that proves no head exists: publish the local store. - if (events.length === 0 || events[0].pubkey !== this.pubkey) - return { kind: "publish", store }; - const event = events[0]; - // Record the raw head before decrypt on the pre-publish path too. - this.recordRemoteHead(event.created_at); - const remote = await decryptAndParse(event); - // A head exists but could not be read (decrypt fault / malformed / future - // schema). A max-merge is only safe once both operands were actually read, - // so retain rather than publish the local store over an uninspectable head - // (Carl P1). The whole-blob lanes retain here too; the durable outbox and - // bounded retry resume normal resolution once a readable head returns. - if (!remote) return { kind: "retain" }; - this.observe(remote.store); - // Max-merge: the local edit's per-entry winners survive by construction and - // any newer remote entries fold in, so no adopt step is needed. - return { kind: "publish", store: mergeStores(store, remote.store) }; - } - - /** - * After a publish OK, fetch the authoritative retained head and report whether - * it subsumes the store we attempted to write. The relay returns OK for a - * superseded NIP-33 write as a no-op, so two windows racing distinct blobs - * from the same head can both get OK while only one blob is retained (Carl - * P1). Only a retained head that subsumes our store proves our click is - * durable on the relay; otherwise the outbox must be kept and retried. - * - * A failed/absent/undecryptable fetch cannot prove subsumption and returns - * false, so the durable click is retained and retried rather than dropped. - * Records the fetched head and observes it so a subsequent merge/retry folds - * the retained blob in. - */ - private async confirmRetainedHeadSubsumes( - store: ChannelMuteStore, - ): Promise { - try { - const events = await relayClient.fetchEvents({ - kinds: [KIND_CHANNEL_MUTES], - authors: [this.pubkey], - "#d": [D_TAG], - limit: 1, - }); - if (events.length === 0 || events[0].pubkey !== this.pubkey) return false; - const event = events[0]; - this.recordRemoteHead(event.created_at); - const remote = await decryptAndParse(event); - if (!remote) return false; - this.observe(remote.store); - return isMutesStoreSubsumedBy(store, remote.store); - } catch { +function mutesStoresEqual(a: ChannelMuteStore, b: ChannelMuteStore): boolean { + const aKeys = Object.keys(a.channels); + const bKeys = Object.keys(b.channels); + if (aKeys.length !== bKeys.length) return false; + for (const key of aKeys) { + const ae = a.channels[key]; + const be = b.channels[key]; + if ( + !ae || + !be || + ae.muted !== be.muted || + ae.updatedAt !== be.updatedAt || + ae.rev !== be.rev + ) return false; - } } + return true; +} - private isIdenticalToLastPublished(store: ChannelMuteStore): boolean { - if (!this.lastPublishedStore) return false; - const lastKeys = Object.keys(this.lastPublishedStore.channels); - const currentKeys = Object.keys(store.channels); - if (lastKeys.length !== currentKeys.length) return false; - for (const key of currentKeys) { - const last = this.lastPublishedStore.channels[key]; - const current = store.channels[key]; - if ( - !last || - last.muted !== current.muted || - last.updatedAt !== current.updatedAt || - last.rev !== current.rev - ) - return false; - } - return true; - } - - /** - * Clear the in-memory pending edit and this window's own durable outbox key — - * but only if the completing publish still owns the current generation. A - * publish for an older edit that finishes after a newer edit was queued must - * leave the newer edit (and its retry state) untouched. - */ - private discardPending(gen: number): void { - if (gen !== this.pendingGeneration) return; - this.pendingStore = null; - clearChannelMutesOutbox(this.pubkey, this.relayUrl); - } - - /** Schedule a bounded-backoff retry of the retained pending edit. */ - private scheduleRetry(gen: number): void { - if (this.destroyed || this.pendingStore === null) return; - // A newer edit has superseded this one; its own timer owns the retry. - if (gen !== this.pendingGeneration) return; - if (this.retryTimer !== null) return; - const delay = this.retryDelayMs; - this.retryDelayMs = Math.min(this.retryDelayMs * 2, RETRY_MAX_MS); - this.retryTimer = window.setTimeout(() => { - this.retryTimer = null; - this.startCycle(); - }, delay); - } - - private async doPublish(store: ChannelMuteStore, gen: number): Promise { - // A newer edit was queued after this publish was scheduled; it owns the - // pending state and will publish the latest store — abandon this stale run. - if (gen !== this.pendingGeneration) return; - try { - const decision = await this.fetchOwnBlobBeforePublish(store); - // Guard: manager may have been destroyed while fetchOwnBlobBeforePublish - // was awaited (community switch during in-flight fetch). - if (this.destroyed) return; - // A newer edit was queued while we awaited the pre-publish fetch. It owns - // convergence now; the serialized cycle re-drives for it once this run - // unwinds. - if (gen !== this.pendingGeneration) return; - // The pre-publish read failed or the head was unreadable — keep the - // durable pending edit and retry rather than publish over uninspectable - // state (Carl P1). - if (decision.kind === "retain") { - this.scheduleRetry(gen); - return; - } - const merged = decision.store; - if (this.isIdenticalToLastPublished(merged)) { - this.discardPending(gen); - return; - } - const payload = { - version: 1, - channels: merged.channels, - }; - const ciphertext = await nip44EncryptToSelf(JSON.stringify(payload)); - // Clamp inside the relay's future-drift window so a skewed remote head - // can never make us stamp an unbounded future timestamp that wedges every - // subsequent publish; we adopt such a head on the next fetch instead. - const createdAt = clampPublishCreatedAt(this.lastRemoteCreatedAt); - const event = await signRelayEvent({ - kind: KIND_CHANNEL_MUTES, - content: ciphertext, - createdAt, - tags: [ - ["d", D_TAG], - ["t", D_TAG], // relay discoverability; not used in our filters - ], - }); - // Final guard immediately before the network call: a newer edit may have - // been queued during the encrypt/sign await, or the manager destroyed. - if (this.destroyed || gen !== this.pendingGeneration) return; - await relayClient.publishEvent( - event, - "Timed out publishing channel mutes.", - "Failed to publish channel mutes.", - ); - this.recordRemoteHead(event.created_at); - this.observe(merged); - // A publish OK is NOT proof of retention: the relay OKs a superseded - // NIP-33 write as a no-op, so a peer window racing a distinct blob from - // the same head can win retention while ours is silently dropped (Carl - // P1). Fetch the authoritative retained head and clear the durable outbox - // only when it subsumes what we wrote. If it does not (or the fetch - // could not prove it), keep the outbox and retry so the click is never - // lost; the retry's pre-publish max-merge folds the retained blob in. - if (this.destroyed) return; - const confirmed = await this.confirmRetainedHeadSubsumes(merged); - if (this.destroyed) return; - if (!confirmed) { - this.scheduleRetry(gen); - return; - } - // Only claim this store as the published head if it is still the current - // edit; a newer edit queued mid-flight owns lastPublishedStore now. - if (gen === this.pendingGeneration) { - this.lastPublishedStore = merged; - this.retryDelayMs = RETRY_BASE_MS; - } - this.discardPending(gen); - } catch (error) { - if (this.destroyed) return; - // Transient publish failure (timeout / socket error). Keep the pending - // edit and retry with backoff rather than waiting for a reconnect that a - // healthy socket never fires. Max-merge makes a duplicate publish - // idempotent, so a lost-ACK write that the relay actually accepted is - // harmless to re-send. - console.warn("[channelMutesSync] publish failed:", error); - this.scheduleRetry(gen); - } - } - - async subscribeToMutes( - onUpdate: (remote: RemoteMutes) => void, - ): Promise<() => Promise> { - return relayClient.subscribeLive( - { - kinds: [KIND_CHANNEL_MUTES], - authors: [this.pubkey], - "#d": [D_TAG], - limit: 0, +export class ChannelMuteSyncManager extends MergeLaneSyncManager { + constructor(pubkey: string, relayUrl: string) { + super(pubkey, relayUrl, { + kind: KIND_CHANNEL_MUTES, + dTag: "channel-mutes", + logPrefix: "channelMutesSync", + publishTimeoutMsg: "Timed out publishing channel mutes.", + publishErrorMsg: "Failed to publish channel mutes.", + parse: parseMutePayload, + serializePayload: (store) => ({ version: 1, channels: store.channels }), + mergeWithRemote: mergeStores, + isSubsumedBy: isMutesStoreSubsumedBy, + storesEqual: mutesStoresEqual, + observe: (highWater, store) => { + for (const [id, entry] of Object.entries(store.channels)) { + const cur = highWater.get(id) ?? { rev: 0, updatedAt: 0 }; + highWater.set(id, { + rev: Math.max(cur.rev, entry.rev), + updatedAt: Math.max(cur.updatedAt, entry.updatedAt), + }); + } }, - (event: RelayEvent) => { - if (event.pubkey !== this.pubkey) return; - // Record the raw head before decrypt so an undecryptable live event - // still advances the watermark and blocks future seed-publish. - this.recordRemoteHead(event.created_at); - void decryptAndParse(event).then((result) => { - if (result) { - this.observe(result.store); - onUpdate(result); - } - }); - }, - ); - } - - /** - * Fetches the remote blob on first mount, records the remote head, and - * delegates the seed/hold/apply-remote decision to `runBootstrap`. - */ - async bootstrap(localStore: ChannelMuteStore) { - // Seed the high-water from the caller's persisted local store so a click - // before the remote fetch resolves already reflects retained entries. - this.observe(localStore); - const fetchResult = await this.fetchRemoteMutes(); - return runBootstrap({ - fetchResult, - lastHead: this.lastRemoteCreatedAt, - localStore, + writeOutbox: writeChannelMutesOutbox, + clearOutbox: clearChannelMutesOutbox, isLocalNonEmpty: (s) => Object.keys(s.channels).length > 0, - publishFn: (s) => this.publishMutes(s), }); } - destroy(): void { - // Cancel any pending publish and mark this manager as destroyed so any - // in-flight doPublish() calls abort before reaching relayClient. - // Debounce-window changes are NOT lost: publishMutes persisted them to the - // durable outbox synchronously, and the next mount resumes them. Flushing - // here is still avoided — it could publish relay A's state to relay B via - // the shared relayClient singleton. - this.destroyed = true; - this.cancelPendingMutePublish(); - this.pendingStore = null; + /** Publish a mute store, debounced to 2s. */ + publishMutes(store: ChannelMuteStore): void { + this.publish(store); + } + + /** Fetch the current remote head for this pubkey's mutes blob. */ + fetchRemoteMutes() { + return this.fetchRemoteBlob(); + } + + /** Cancel any pending mute publish (debounce or retry timer). */ + cancelPendingMutePublish(): void { + this.cancelPendingPublish(); + } + + /** The currently pending store, or null if nothing is queued. */ + getPendingMuteStore(): ChannelMuteStore | null { + return this.getPendingStore(); + } + + /** Subscribe to live relay events for this pubkey's mutes blob. */ + subscribeToMutes(onUpdate: (remote: RemoteMutes) => void) { + return this.subscribeLive(onUpdate); } } diff --git a/desktop/src/features/sidebar/lib/channelSectionsSync.ts b/desktop/src/features/sidebar/lib/channelSectionsSync.ts index 2315440156..a9818bbb35 100644 --- a/desktop/src/features/sidebar/lib/channelSectionsSync.ts +++ b/desktop/src/features/sidebar/lib/channelSectionsSync.ts @@ -1,10 +1,3 @@ -import { relayClient } from "@/shared/api/relayClient"; -import { - nip44DecryptFromSelf, - nip44EncryptToSelf, - signRelayEvent, -} from "@/shared/api/tauri"; -import type { RelayEvent } from "@/shared/api/types"; import { KIND_CHANNEL_SECTIONS } from "@/shared/constants/kinds"; import { clearChannelSectionsOutbox, @@ -13,768 +6,68 @@ import { type ChannelSection, type ChannelSectionStore, } from "./channelSectionsStorage"; -import { - advanceWatermark, - clampPublishCreatedAt, - readWatermark, - runBootstrap, - type FetchResult, -} from "./sidebarSyncWatermark"; +import { WholeBlobSyncManager, type RemoteBlob } from "./wholeBlobSyncManager"; -const D_TAG = "channel-sections"; -const BLOB_TYPE = D_TAG; -const DEBOUNCE_MS = 2_000; +export type RemoteSections = RemoteBlob; -// Bounded backoff for a retained pending edit whose publish failed transiently -// (timeout / socket error) on an otherwise-healthy socket, so it does not wait -// for a reconnect that may never fire. -const RETRY_BASE_MS = 2_000; -const RETRY_MAX_MS = 30_000; - -export type RemoteSections = { - store: ChannelSectionStore; - createdAt: number; - eventId: string; -}; - -/** - * Outcome of the pre-publish head check. - * - * - `publish` — local edit is at or ahead of the head; publish it. - * - `adopt` — a newer remote head exists; the local edit lost whole-blob - * LWW and must be discarded in favour of the remote store so UI - * and relay converge (see the fix-2 design note). The manager - * hands the remote back to the hook and never publishes. - * - `retain` — a head event exists but could not be decrypted/parsed (future - * schema, transient keychain/decrypt fault, malformed payload). - * The client cannot inspect it to decide adopt-or-publish, so - * overwriting it would be blind data loss (Carl P1). Keep the - * durable pending edit and retry rather than clobber authoritative - * state the client could not read. - */ -type PublishDecision = - | { kind: "publish"; store: ChannelSectionStore } - | { kind: "adopt"; remote: RemoteSections } - | { kind: "retain" }; - -/** - * The canonical remote head as it stood when an edit was queued. The pre-publish - * check compares the fetched head against this frozen baseline — never against - * the mutable in-memory watermark, which a live event observed during the - * debounce window may already have advanced to that same head (silently - * suppressing the adopt). - */ -type PublishBaseline = { createdAt: number; eventId: string }; - -/** - * Outcome of the post-publish retained-head confirmation (Carl P1). - * - * - `confirmed` — the retained head is exactly our event; the write won LWW. - * - `adopt` — a different, readable event is the head; our write was - * superseded (a peer's same-second blob with a lower id won, or - * a strictly-newer head landed). The caller adopts the winner. - * - `retain` — the fetch failed, returned no/foreign head, or the head was - * unreadable; retention cannot be proven, so keep the durable - * edit and retry. - */ -type RetentionConfirmation = - | { kind: "confirmed" } - | { kind: "adopt"; remote: RemoteSections } - | { kind: "retain" }; - -/** - * True when `head` is the canonical winner over the baseline the edit was queued - * against — i.e. the head advanced since the edit began. Canonical order is - * `created_at DESC, id ASC`: a strictly-later head wins, and a same-second head - * wins only with a strictly-lower id. A same-second head is comparable only once - * the baseline id is known (empty id = no prior head seen → not superseded). - */ -function remoteAdvancedSince( - head: RemoteSections, - baseline: PublishBaseline, +function sectionsStoresEqual( + a: ChannelSectionStore, + b: ChannelSectionStore, ): boolean { - if (head.createdAt > baseline.createdAt) return true; - return ( - head.createdAt === baseline.createdAt && - baseline.eventId !== "" && - head.eventId < baseline.eventId - ); -} - -/** - * True when tuple `a` is the canonical winner over `b` (`created_at DESC, - * id ASC`). An empty id means "no head seen yet" and always loses. - */ -function canonicalGreater(a: PublishBaseline, b: PublishBaseline): boolean { - if (a.eventId === "") return false; - if (b.eventId === "") return true; - if (a.createdAt !== b.createdAt) return a.createdAt > b.createdAt; - return a.eventId < b.eventId; -} - -/** The canonical-greater of two head tuples (`created_at DESC, id ASC`). */ -function canonicalMax(a: PublishBaseline, b: PublishBaseline): PublishBaseline { - return canonicalGreater(a, b) ? a : b; -} - -async function decryptAndParse( - event: RelayEvent, -): Promise { - try { - const plaintext = await nip44DecryptFromSelf(event.content); - const store = parseChannelSectionPayload(JSON.parse(plaintext)); - if (!store) return null; - return { store, createdAt: event.created_at, eventId: event.id }; - } catch { - return null; + if (a.sections.length !== b.sections.length) return false; + for (let i = 0; i < a.sections.length; i++) { + const as = a.sections[i] as ChannelSection | undefined; + const bs = b.sections[i] as ChannelSection | undefined; + if ( + !as || + !bs || + as.id !== bs.id || + as.name !== bs.name || + as.icon !== bs.icon || + as.order !== bs.order + ) + return false; } + const aKeys = Object.keys(a.assignments); + const bKeys = Object.keys(b.assignments); + if (aKeys.length !== bKeys.length) return false; + for (const key of aKeys) { + if (a.assignments[key] !== b.assignments[key]) return false; + } + return true; } -export class ChannelSectionSyncManager { - private pubkey: string; - private relayUrl: string; - private debounceTimer: number | null = null; - private retryTimer: number | null = null; - private retryDelayMs = RETRY_BASE_MS; - private lastRemoteCreatedAt: number; - // Canonical best head observed so far (`created_at DESC, id ASC`). Frozen into - // a per-edit baseline at publishSections so the pre-publish check can tell - // whether the head advanced *since the edit was queued*, independent of the - // mutable watermark that a live event during the debounce window may advance. - private lastRemoteHead: PublishBaseline = { createdAt: 0, eventId: "" }; - // The canonical head this pending edit is racing against, frozen when the - // edit was queued (publishSections) and advanced ONLY by our own successful - // publishes. Freezing at queue time is what makes a genuine remote observed - // during the debounce window still adopt-worthy (pass-2): the mutable - // watermark advanced to that remote, but the baseline did not. Folding our - // own published head forward is what stops a newer edit from adopting an - // older generation's own accepted write (pass-3): our prior publish is our - // baseline, not a competing remote. - private publishBaseline: PublishBaseline = { createdAt: 0, eventId: "" }; - private pendingStore: ChannelSectionStore | null = null; - // Monotonic id for the current pending edit. Every publishSections() bumps - // it; every scheduled publish/retry captures the value it was queued for. - // A completion (success, adopt, or no-op) may only clear pending state via - // compare-and-swap on this generation, so an older in-flight publish can - // never erase a newer edit that arrived while it was in flight. - private pendingGeneration = 0; - // Publish cycles are serialized: at most one runs at a time. A newer edit - // queued while a cycle is in flight does NOT start its own concurrent cycle; - // it defers, and the in-flight cycle's completion schedules it. Serialization - // guarantees there is never more than one baseline/fetch/publish sequence - // touching shared manager state, so a stale generation can never sign or - // publish after a newer edit exists. - private publishInFlight = false; - // Event ids we signed and sent to the relay but whose ACK never arrived (the - // publish promise rejected as a timeout/socket error after the frame left). - // The relay MAY have accepted such a write, so if a later cycle's pre-publish - // fetch returns a head whose id is in this map, that head is OUR OWN accepted - // predecessor — fold it forward and publish above it, rather than adopting it - // and erasing the queued edit. An attempt the relay never accepted can never - // surface as the head, so an id match is proof of our own accepted write. - // Maps attempt id → pendingGeneration at the time of signing, so - // foldSupersedingAttemptWinner can require the baseline was poisoned by a - // strictly prior generation (gen < pendingGeneration). - private ambiguousAttemptIds = new Map(); - private lastPublishedStore: ChannelSectionStore | null = null; - private destroyed = false; - // Set by the hook so an adopted remote head (local edit lost whole-blob LWW) - // is written through to React state + localStorage. - private onRemoteAdopted: ((remote: RemoteSections) => void) | null = null; - +export class ChannelSectionSyncManager extends WholeBlobSyncManager { constructor(pubkey: string, relayUrl: string) { - this.pubkey = pubkey; - this.relayUrl = relayUrl; - // Hydrate from localStorage so we never seed-publish if a remote blob has - // been seen in a prior session. - this.lastRemoteCreatedAt = readWatermark(pubkey, BLOB_TYPE, relayUrl); - } - - /** Register the hook's adopt-remote sink (write-through to UI + storage). */ - setOnRemoteAdopted(cb: (remote: RemoteSections) => void): void { - this.onRemoteAdopted = cb; - } - - async fetchRemoteSections(): Promise> { - try { - const events = await relayClient.fetchEvents({ - kinds: [KIND_CHANNEL_SECTIONS], - authors: [this.pubkey], - "#d": [D_TAG], - limit: 1, - }); - if (events.length === 0 || events[0].pubkey !== this.pubkey) { - return { status: "absent" }; - } - const event = events[0]; - // An event exists — record its created_at regardless of whether we can - // decrypt it, so seed-publish is blocked even when the payload is - // unreadable (e.g. wrong key). - this.recordRemoteHead(event.created_at, event.id); - const result = await decryptAndParse(event); - if (!result) { - return { status: "failed", createdAt: event.created_at }; - } - return { - status: "found", - data: result, - createdAt: result.createdAt, - eventId: result.eventId, - }; - } catch { - return { status: "failed" }; - } - } - - /** Update in-memory + persisted watermark and the canonical head tuple. */ - private recordRemoteHead(createdAt: number, eventId: string): void { - if (createdAt > this.lastRemoteCreatedAt) { - this.lastRemoteCreatedAt = createdAt; - } - // Track the canonical-best head (`created_at DESC, id ASC`): a later head - // always wins; a same-second head wins only with a strictly-lower id. This - // mirrors the relay's stored winner so a frozen baseline reflects reality. - if ( - createdAt > this.lastRemoteHead.createdAt || - (createdAt === this.lastRemoteHead.createdAt && - (this.lastRemoteHead.eventId === "" || - eventId < this.lastRemoteHead.eventId)) - ) { - this.lastRemoteHead = { createdAt, eventId }; - } - advanceWatermark(this.pubkey, BLOB_TYPE, this.relayUrl, createdAt); - } - - cancelPendingPublish(): void { - if (this.debounceTimer !== null) { - window.clearTimeout(this.debounceTimer); - this.debounceTimer = null; - } - if (this.retryTimer !== null) { - window.clearTimeout(this.retryTimer); - this.retryTimer = null; - } - } - - getPendingStore(): ChannelSectionStore | null { - return this.pendingStore; - } - - /** - * Re-drive the CURRENT pending edit's publish immediately without opening a - * new generation — used by the reconnect handler. Calling the public - * `publishSections()` there would bump the generation and reset - * `publishBaseline` to the just-fetched head, so a remote that won LWW while - * the edit was pending would be published over instead of adopted (Carl P1). - * Waking the existing generation keeps the frozen baseline, so the debounced - * cycle's pre-publish check still adopts a genuinely-advanced remote. No-op - * when nothing is pending. - */ - retryPendingPublish(): void { - if (this.pendingStore === null) return; - this.cancelPendingPublish(); - this.startCycle(); - } - - /** True while an unpublished local edit is queued (debouncing or retrying). */ - hasPendingEdit(): boolean { - return this.pendingStore !== null; - } - - /** - * When the current `publishBaseline` was frozen against one of our own - * attempted (but non-retained) event ids and `remote` is the same-second - * lower-id peer that won the collision, the baseline is poisoned: the - * next pre-publish check would see the winner as having advanced past our - * phantom attempt and adopt the edit away (Carl P1 / same-second variants). - * Folding the winner in repairs the baseline so the pre-publish check sees - * the true retained head and publishes above it. - * - * Scoped to same-second lower-id only: a strictly-later remote is a genuine - * advance that the pending edit must still adopt (pass-2 invariant). - * - * Requires the baseline attempt to be prior-generation (gen < pendingGeneration): - * if no newer edit was queued, the current generation's own in-flight attempt - * cannot trigger the fold — the loser must adopt (LWW correctness). - * The post-ACK ordering is load-bearing for causal soundness: because A's - * ACK returned before B queued, the peer winner pre-existed B, so preserving - * B is causally justified. If B queues while A's ACK is pending, the peer may - * have been accepted after B and is a genuine remote advance B must adopt — - * B's baseline will not equal A's attempt id in that case, so the fold - * does not fire (correct behaviour, by the causal argument above). - * - * Returns true when the fold was applied (caller may use for branching). - */ - private foldSupersedingAttemptWinner(remote: { - createdAt: number; - eventId: string; - }): boolean { - const attempt = this.ambiguousAttemptIds.get(this.publishBaseline.eventId); - if ( - this.publishBaseline.eventId !== "" && - attempt !== undefined && - attempt < this.pendingGeneration && - remote.createdAt === this.publishBaseline.createdAt && - remote.eventId < this.publishBaseline.eventId - ) { - this.publishBaseline = canonicalMax(this.publishBaseline, { - createdAt: remote.createdAt, - eventId: remote.eventId, - }); - return true; - } - return false; - } - - /** - * Adopt a remote store that superseded a local edit: hand it to the hook for - * write-through, advance the watermark, and drop the losing pending edit — - * including the durable outbox, so the outbox can never replay an edit that - * adopt just decided lost (which would reintroduce divergence). - * - * Compare-and-swap on `gen`: this adopt was decided against the edit queued at - * generation `gen`. If a newer edit arrived while this publish was in flight, - * the generation has moved on and that newer edit is the latest writer — it - * will publish and win LWW — so a stale adopt must not clear its pending state - * or overwrite its optimistic UI. We still advance the watermark (monotonic - * and always safe) so the newer edit stamps above this head. - */ - private adoptRemote(remote: RemoteSections, gen: number): void { - this.recordRemoteHead(remote.createdAt, remote.eventId); - if (gen !== this.pendingGeneration) { - // A newer edit is pending. Its baseline may have been frozen against our - // own attempted (non-retained) event id if it was queued after the ACK - // but before confirmation resolved. Repair any poisoned baseline so the - // newer edit's pre-publish check publishes above the true retained head - // rather than adopting the winner away (Carl P1). - this.foldSupersedingAttemptWinner(remote); - return; - } - this.pendingStore = null; - clearChannelSectionsOutbox(this.pubkey, this.relayUrl); - this.lastPublishedStore = remote.store; - if (this.destroyed) return; - this.onRemoteAdopted?.(remote); - } - - /** - * Clear the in-memory pending edit and this window's own durable outbox key — - * but only if the completing publish still owns the current generation. A - * publish for an older edit that finishes after a newer edit was queued must - * leave the newer edit (and its retry state) untouched. - */ - private discardPending(gen: number): void { - if (gen !== this.pendingGeneration) return; - this.pendingStore = null; - clearChannelSectionsOutbox(this.pubkey, this.relayUrl); - } - - publishSections(store: ChannelSectionStore): boolean { - this.pendingStore = store; - ++this.pendingGeneration; - // Freeze the canonical head this edit is racing against at queue time. The - // pre-publish check compares the fetched head against this baseline, not the - // mutable watermark — a live event applied during the debounce window - // advances the watermark to a new remote head, which would otherwise make - // the pre-publish comparison see equality and fall through to a publish that - // overwrites a remote that became head *after* this edit was queued. The - // baseline only advances via our own successful publishes (see doPublish), - // so a prior generation's own accepted write is folded in rather than - // mistaken for a competing remote. - this.publishBaseline = { ...this.lastRemoteHead }; - // Persist synchronously so an edit made <2s before quit/community-switch - // survives teardown and resumes on next mount (durable outbox). This - // window's own key is the only one written — a single unconditional - // setItem, never a shared-key read-modify-write; `queuedAt` stamps it so - // resume replays only the newest queued blob (whole-blob LWW). The returned - // flag reports whether the intent is now durably held, so a legacy replay - // can gate its consumption marker on a proven transfer. - const durable = writeChannelSectionsOutbox( - this.pubkey, - store, - this.relayUrl, - ); - if (this.debounceTimer !== null) { - window.clearTimeout(this.debounceTimer); - } - // A fresh edit supersedes any retry scheduled for the previous generation. - if (this.retryTimer !== null) { - window.clearTimeout(this.retryTimer); - this.retryTimer = null; - } - this.retryDelayMs = RETRY_BASE_MS; - this.debounceTimer = window.setTimeout(() => { - this.debounceTimer = null; - this.startCycle(); - }, DEBOUNCE_MS); - return durable; - } - - /** - * Serialize publish cycles: at most one runs at a time. A debounce/retry - * timer that fires while a cycle is in flight defers — the in-flight cycle's - * completion re-drives if a pending edit still needs publishing. This kills - * the cross-generation race class by construction: a newer edit queued during - * a cycle cannot start its own concurrent cycle, so there is never more than - * one baseline/fetch/publish sequence competing over shared manager state. - */ - private startCycle(): void { - if (this.destroyed || this.pendingStore === null) return; - if (this.publishInFlight) return; - const store = this.pendingStore; - const gen = this.pendingGeneration; - this.publishInFlight = true; - void this.doPublish(store, gen).finally(() => { - this.publishInFlight = false; - // A newer edit queued during the cycle (or a cycle that ended without - // clearing its pending edit) still needs publishing and has no timer - // pending to drive it — drive the next cycle now that the lane is free. - if ( - !this.destroyed && - this.pendingStore !== null && - this.debounceTimer === null && - this.retryTimer === null - ) { - this.startCycle(); - } - }); - } - - private async fetchOwnBlobBeforePublish( - store: ChannelSectionStore, - ): Promise { - try { - const events = await relayClient.fetchEvents({ - kinds: [KIND_CHANNEL_SECTIONS], - authors: [this.pubkey], - "#d": [D_TAG], - limit: 1, - }); - if (events.length === 0 || events[0].pubkey !== this.pubkey) - return { kind: "publish", store }; - const event = events[0]; - const remote = await decryptAndParse(event); - // Record the head after decrypt attempt so the watermark/head-tuple - // advance even for an undecryptable payload. - this.recordRemoteHead(event.created_at, event.id); - // A head event exists but we could not read it (future schema, transient - // decrypt fault, malformed payload). We cannot decide adopt-or-publish - // against state we cannot inspect, so retain the pending edit and retry - // rather than blindly overwrite authoritative state (Carl P1). - if (!remote) return { kind: "retain" }; - // Sections use whole-blob LWW. Compare the fetched head against the - // baseline frozen when this edit was queued — NOT the live watermark, - // which a passive live event during debounce may already have advanced to - // this same head. If the canonical head advanced since the edit began, the - // local edit lost and is adopted-away rather than republished over it. - if (remoteAdvancedSince(remote, this.publishBaseline)) { - // Unless the advancing head is a prior publish of OURS whose ACK was - // lost: the relay accepted it, but our promise rejected before we could - // fold it forward, so it is our own accepted predecessor — not a - // competing remote. Fold it into the baseline and publish above it so - // the queued edit survives instead of adopting our own stale write away. - if (this.ambiguousAttemptIds.has(remote.eventId)) { - this.publishBaseline = canonicalMax(this.publishBaseline, { - createdAt: remote.createdAt, - eventId: remote.eventId, - }); - return { kind: "publish", store }; - } - // Same-second lower-id peer winner that superseded our own attempted - // head: baseline.eventId is our attempted id (still in - // ambiguousAttemptIds because the adopt path never clears it) and the - // relay retained the peer's lower-id blob instead. This arises when - // confirmation returned retain/failed and the retry pre-publish fetch - // sees the winner directly with the original poisoned baseline still in - // place. Fold the winner in so this edit publishes above the true - // retained head rather than adopting it away (Carl P1). - if (this.foldSupersedingAttemptWinner(remote)) { - return { kind: "publish", store }; - } - return { kind: "adopt", remote }; - } - return { kind: "publish", store }; - } catch { - // The pre-publish fetch itself failed (timeout / auth / socket) — this is - // NOT proof that no head exists. Publishing here would sign above a stale - // watermark and could erase an unseen newer head during a transient - // outage. Retain the durable pending edit and retry rather than overwrite - // state we could not read (Carl P1). - return { kind: "retain" }; - } - } - - /** - * After a publish OK, fetch the authoritative retained head and decide whether - * our event actually became it. The relay OKs a superseded NIP-33 write as a - * no-op (`Duplicate`), so a resolved publish is NOT proof of retention: two - * windows stamping the same `created_at` both get OK while only the lower - * event id is retained (Carl P1). Only an exact event-id match proves our - * whole-blob write won LWW. - * - * - head id === ours → `confirmed` - * - head is our own accepted prior → `retain` (a lost-ACK / same-second - * attempt (ambiguousAttemptIds) lower-id attempt of ours won; the retry - * folds it forward and republishes above) - * - head is a different readable → `adopt` (our blob lost; adopt the - * event winner so UI and relay converge) - * - fetch failed / no head / head → `retain` (retention unprovable or the - * unreadable winner unreadable — keep the durable - * edit and retry) - */ - private async confirmRetainedHead( - ourEventId: string, - ): Promise { - try { - const events = await relayClient.fetchEvents({ - kinds: [KIND_CHANNEL_SECTIONS], - authors: [this.pubkey], - "#d": [D_TAG], - limit: 1, - }); - if (events.length === 0 || events[0].pubkey !== this.pubkey) - return { kind: "retain" }; - const event = events[0]; - this.recordRemoteHead(event.created_at, event.id); - if (event.id === ourEventId) return { kind: "confirmed" }; - // A different event is the head. If it is one of OUR OWN accepted prior - // attempts (a lost-ACK write the relay retained, or a same-second lower-id - // attempt that won over this one), it is not a competing remote — retain - // so the retry's pre-publish folds it forward and publishes above it, - // rather than adopting our own older blob and dropping the current edit. - if (this.ambiguousAttemptIds.has(event.id)) return { kind: "retain" }; - const remote = await decryptAndParse(event); - if (!remote) return { kind: "retain" }; - return { kind: "adopt", remote }; - } catch { - return { kind: "retain" }; - } - } - - private isIdenticalToLastPublished(store: ChannelSectionStore): boolean { - if (!this.lastPublishedStore) return false; - const lastSections = this.lastPublishedStore.sections; - const currentSections = store.sections; - if (lastSections.length !== currentSections.length) return false; - for (let i = 0; i < currentSections.length; i++) { - const last = lastSections[i] as ChannelSection | undefined; - const current = currentSections[i] as ChannelSection; - if ( - !last || - last.id !== current.id || - last.name !== current.name || - last.icon !== current.icon || - last.order !== current.order - ) - return false; - } - const lastAssignKeys = Object.keys(this.lastPublishedStore.assignments); - const currentAssignKeys = Object.keys(store.assignments); - if (lastAssignKeys.length !== currentAssignKeys.length) return false; - for (const key of currentAssignKeys) { - if (this.lastPublishedStore.assignments[key] !== store.assignments[key]) - return false; - } - return true; - } - - /** Schedule a bounded-backoff retry of the retained pending edit. */ - private scheduleRetry(gen: number): void { - if (this.destroyed || this.pendingStore === null) return; - // A newer edit has superseded this one; its own timer owns the retry. - if (gen !== this.pendingGeneration) return; - if (this.retryTimer !== null) return; - const delay = this.retryDelayMs; - this.retryDelayMs = Math.min(this.retryDelayMs * 2, RETRY_MAX_MS); - this.retryTimer = window.setTimeout(() => { - this.retryTimer = null; - this.startCycle(); - }, delay); - } - - private async doPublish( - store: ChannelSectionStore, - gen: number, - ): Promise { - // A newer edit was queued after this publish was scheduled; it owns the - // pending state and will publish the latest store — abandon this stale run. - if (gen !== this.pendingGeneration) return; - try { - const decision = await this.fetchOwnBlobBeforePublish(store); - // Guard: manager may have been destroyed while fetchOwnBlobBeforePublish - // was awaited (community switch during in-flight fetch). If so, abort - // before touching the relay. - if (this.destroyed) return; - // A newer edit was queued while we awaited the pre-publish fetch. It owns - // convergence now; abort so we neither publish this stale store nor adopt - // over the newer pending edit. The serialized cycle re-drives for the - // newer generation once this one unwinds. - if (gen !== this.pendingGeneration) return; - if (decision.kind === "adopt") { - this.adoptRemote(decision.remote, gen); - return; - } - if (decision.kind === "retain") { - // Head exists but is unreadable — keep the durable pending edit and - // retry with backoff rather than overwrite state we could not inspect. - this.scheduleRetry(gen); - return; - } - const merged = decision.store; - if (this.isIdenticalToLastPublished(merged)) { - this.discardPending(gen); - return; - } - const payload = { + super(pubkey, relayUrl, { + kind: KIND_CHANNEL_SECTIONS, + dTag: "channel-sections", + logPrefix: "channelSectionsSync", + parse: parseChannelSectionPayload, + serializePayload: (store) => ({ version: 1, - sections: merged.sections, - assignments: merged.assignments, - }; - const ciphertext = await nip44EncryptToSelf(JSON.stringify(payload)); - // Clamp inside the relay's future-drift window: never manufacture a - // timestamp so far ahead that this or a later publish is rejected for - // drift and wedges. If a skewed remote head sits beyond the window we - // will lose LWW and adopt it on the next pre-publish fetch rather than - // walking past it. - const createdAt = clampPublishCreatedAt(this.lastRemoteCreatedAt); - const event = await signRelayEvent({ - kind: KIND_CHANNEL_SECTIONS, - content: ciphertext, - createdAt, - tags: [ - ["d", D_TAG], - ["t", D_TAG], // relay discoverability; not used in our filters - ], - }); - // Final guard immediately before the network call — sign/encrypt are - // synchronous-ish but cheap; the relay socket may have moved to a - // different community by the time we reach this point, or a newer edit - // may have been queued during the encrypt/sign await (invariant: a stale - // generation never signs/publishes after a newer edit exists). - if (this.destroyed || gen !== this.pendingGeneration) return; - // Record this signed id as an in-flight attempt of unknown fate before we - // send it. If the ACK is lost below, a later cycle that fetches this id as - // the head recognises it as our own accepted write and folds it forward - // rather than adopting it away. - this.ambiguousAttemptIds.set(event.id, gen); - await relayClient.publishEvent( - event, - "Timed out publishing channel sections.", - "Failed to publish channel sections.", - ); - this.recordRemoteHead(event.created_at, event.id); - // A publish OK is NOT proof of retention: the relay OKs a superseded - // NIP-33 write as a no-op (`Duplicate`), so two windows stamping the same - // `created_at` both get OK while only the lower event id is retained (Carl - // P1). Fetch the authoritative head; only an exact id match proves our - // whole-blob write won. Anything else must keep the durable edit — never - // fold a nonexistent head into the baseline, which would poison the next - // edit into adopting the true winner away. - if (this.destroyed) return; - const confirmation = await this.confirmRetainedHead(event.id); - if (this.destroyed) return; - if (confirmation.kind === "adopt") { - // A different readable event is the head — our blob lost the same-second - // collision. Adopt the winner so UI and relay converge (CAS on gen - // inside adoptRemote leaves a newer edit untouched). - this.adoptRemote(confirmation.remote, gen); - return; - } - if (confirmation.kind === "retain") { - // Retention unprovable (fetch failed / no head), or the head is our own - // accepted prior attempt whose ACK was lost. Keep the durable edit and - // retry: the retry's pre-publish folds our own accepted predecessor - // forward (ambiguousAttemptIds) and republishes above it. Leave - // ambiguousAttemptIds intact so that fold still recognises our writes. - this.scheduleRetry(gen); - return; - } - // Confirmed: our event is the retained head. It dominates every prior - // attempt (`created_at DESC, id ASC`), so no earlier ambiguous id can ever - // be the canonical head again — clear the set to keep it bounded. - this.ambiguousAttemptIds.clear(); - // Fold our own accepted head into the pending edit's baseline. This is - // unconditional across generations: even a stale generation's own - // confirmed write must advance the current pending baseline so the newer - // edit's pre-publish check does not mistake OUR prior publish for a - // competing remote and adopt it away (pass-3). Genuine remotes never fold - // in here — they only advance the watermark — so a remote that became head - // during the debounce window still adopts (pass-2). canonicalMax keeps the - // advance monotonic (`created_at DESC, id ASC`). - this.publishBaseline = canonicalMax(this.publishBaseline, { - createdAt: event.created_at, - eventId: event.id, - }); - // Only claim this store as the published head if it is still the current - // edit; a newer edit queued mid-flight owns lastPublishedStore now. - if (gen === this.pendingGeneration) { - this.lastPublishedStore = merged; - this.retryDelayMs = RETRY_BASE_MS; - } - this.discardPending(gen); - } catch (error) { - if (this.destroyed) return; - // Ambiguous outcome: the publish promise rejected (timeout / socket - // error), but the relay may already have accepted the write before the - // ACK was lost. Keep the pending edit and retry with backoff rather than - // waiting for a reconnect that a healthy socket never fires. The attempt - // id stays in ambiguousAttemptIds: if the relay did accept it, a later - // cycle that fetches this id as the head folds it forward as our own - // accepted predecessor (see fetchOwnBlobBeforePublish) instead of - // adopting it away and erasing the queued edit. - console.warn("[channelSectionsSync] publish failed:", error); - this.scheduleRetry(gen); - } - } - - async subscribeToSections( - onUpdate: (remote: RemoteSections) => void, - ): Promise<() => Promise> { - return relayClient.subscribeLive( - { - kinds: [KIND_CHANNEL_SECTIONS], - authors: [this.pubkey], - "#d": [D_TAG], - limit: 0, - }, - (event: RelayEvent) => { - if (event.pubkey !== this.pubkey) return; - // Record the raw head before decrypt so an undecryptable live event - // still advances the watermark and blocks future seed-publish. - this.recordRemoteHead(event.created_at, event.id); - void decryptAndParse(event).then((result) => { - if (result) { - onUpdate(result); - } - }); - }, - ); - } - - /** - * Fetches the remote blob on first mount, records the remote head, and - * delegates the seed/hold/apply-remote decision to `runBootstrap`. - */ - async bootstrap(localStore: ChannelSectionStore) { - const fetchResult = await this.fetchRemoteSections(); - return runBootstrap({ - fetchResult, - lastHead: this.lastRemoteCreatedAt, - localStore, + sections: store.sections, + assignments: store.assignments, + }), + writeOutbox: writeChannelSectionsOutbox, + clearOutbox: clearChannelSectionsOutbox, + storesEqual: sectionsStoresEqual, isLocalNonEmpty: (s) => s.sections.length > 0, - publishFn: (s) => this.publishSections(s), }); } - destroy(): void { - // Cancel any pending publish and mark this manager as destroyed so any - // in-flight doPublish() calls abort before reaching relayClient. - // Debounce-window changes are NOT lost: publishSections persisted them to - // the durable outbox synchronously, and the next mount resumes them. - // Flushing here is still avoided — it could publish relay A's sections to - // relay B via the shared relayClient singleton. - this.destroyed = true; - this.cancelPendingPublish(); - this.pendingStore = null; + /** Publish a sections store, debounced to 2s. Returns whether the intent is durably held. */ + publishSections(store: ChannelSectionStore): boolean { + return this.publish(store); + } + + /** Fetch the current remote head for this pubkey's sections blob. */ + fetchRemoteSections() { + return this.fetchRemoteBlob(); + } + + /** Subscribe to live relay events for this pubkey's sections blob. */ + subscribeToSections(onUpdate: (remote: RemoteSections) => void) { + return this.subscribeLive(onUpdate); } } diff --git a/desktop/src/features/sidebar/lib/channelSortSync.ts b/desktop/src/features/sidebar/lib/channelSortSync.ts index c56783b6e3..740dc41e10 100644 --- a/desktop/src/features/sidebar/lib/channelSortSync.ts +++ b/desktop/src/features/sidebar/lib/channelSortSync.ts @@ -1,10 +1,3 @@ -import { relayClient } from "@/shared/api/relayClient"; -import { - nip44DecryptFromSelf, - nip44EncryptToSelf, - signRelayEvent, -} from "@/shared/api/tauri"; -import type { RelayEvent } from "@/shared/api/types"; import { KIND_CHANNEL_SORT } from "@/shared/constants/kinds"; import { clearChannelSortOutbox, @@ -12,720 +5,50 @@ import { writeChannelSortOutbox, type ChannelSortStore, } from "./channelSortPreference"; -import { - advanceWatermark, - clampPublishCreatedAt, - readWatermark, - runBootstrap, - type FetchResult, -} from "./sidebarSyncWatermark"; +import { WholeBlobSyncManager, type RemoteBlob } from "./wholeBlobSyncManager"; -const D_TAG = "channel-sort"; -const BLOB_TYPE = D_TAG; -const DEBOUNCE_MS = 2_000; +export type RemoteSortPrefs = RemoteBlob; -// Bounded backoff for a retained pending edit whose publish failed transiently -// (timeout / socket error) on an otherwise-healthy socket, so it does not wait -// for a reconnect that may never fire. -const RETRY_BASE_MS = 2_000; -const RETRY_MAX_MS = 30_000; - -export type RemoteSortPrefs = { - store: ChannelSortStore; - createdAt: number; - eventId: string; -}; - -/** - * Outcome of the pre-publish head check. - * - * - `publish` — local edit is at or ahead of the head; publish it. - * - `adopt` — a newer remote head exists; the local edit lost whole-blob LWW - * and must be discarded in favour of the remote store so UI and - * relay converge. The manager hands the remote back to the hook - * and never publishes. - * - `retain` — a head event exists but could not be decrypted/parsed (future - * schema, transient keychain/decrypt fault, malformed payload). - * The client cannot inspect it to decide adopt-or-publish, so - * overwriting it would be blind data loss (Carl P1). Keep the - * durable pending edit and retry rather than clobber authoritative - * state the client could not read. - */ -type PublishDecision = - | { kind: "publish"; store: ChannelSortStore } - | { kind: "adopt"; remote: RemoteSortPrefs } - | { kind: "retain" }; - -/** - * The canonical remote head as it stood when an edit was queued. The pre-publish - * check compares the fetched head against this frozen baseline — never against - * the mutable in-memory watermark, which a live event observed during the - * debounce window may already have advanced to that same head (silently - * suppressing the adopt). - */ -type PublishBaseline = { createdAt: number; eventId: string }; - -/** - * Outcome of the post-publish retained-head confirmation (Carl P1). - * - * - `confirmed` — the retained head is exactly our event; the write won LWW. - * - `adopt` — a different, readable event is the head; our write was - * superseded (a peer's same-second blob with a lower id won, or - * a strictly-newer head landed). The caller adopts the winner. - * - `retain` — the fetch failed, returned no/foreign head, or the head was - * unreadable; retention cannot be proven, so keep the durable - * edit and retry. - */ -type RetentionConfirmation = - | { kind: "confirmed" } - | { kind: "adopt"; remote: RemoteSortPrefs } - | { kind: "retain" }; - -/** - * True when `head` is the canonical winner over the baseline the edit was queued - * against — i.e. the head advanced since the edit began. Canonical order is - * `created_at DESC, id ASC`: a strictly-later head wins, and a same-second head - * wins only with a strictly-lower id. A same-second head is comparable only once - * the baseline id is known (empty id = no prior head seen → not superseded). - */ -function remoteAdvancedSince( - head: RemoteSortPrefs, - baseline: PublishBaseline, -): boolean { - if (head.createdAt > baseline.createdAt) return true; - return ( - head.createdAt === baseline.createdAt && - baseline.eventId !== "" && - head.eventId < baseline.eventId - ); -} - -/** - * True when tuple `a` is the canonical winner over `b` (`created_at DESC, - * id ASC`). An empty id means "no head seen yet" and always loses. - */ -function canonicalGreater(a: PublishBaseline, b: PublishBaseline): boolean { - if (a.eventId === "") return false; - if (b.eventId === "") return true; - if (a.createdAt !== b.createdAt) return a.createdAt > b.createdAt; - return a.eventId < b.eventId; -} - -/** The canonical-greater of two head tuples (`created_at DESC, id ASC`). */ -function canonicalMax(a: PublishBaseline, b: PublishBaseline): PublishBaseline { - return canonicalGreater(a, b) ? a : b; -} - -async function decryptAndParse( - event: RelayEvent, -): Promise { - try { - const plaintext = await nip44DecryptFromSelf(event.content); - const store = parseChannelSortPayload(JSON.parse(plaintext)); - if (!store) return null; - return { store, createdAt: event.created_at, eventId: event.id }; - } catch { - return null; +function sortStoresEqual(a: ChannelSortStore, b: ChannelSortStore): boolean { + const aKeys = Object.keys(a.groups); + const bKeys = Object.keys(b.groups); + if (aKeys.length !== bKeys.length) return false; + for (const key of aKeys) { + if (a.groups[key] !== b.groups[key]) return false; } + return true; } -/** - * Syncs the per-group sidebar sort preferences across clients via encrypted - * NIP-78 app data (kind 30078, d-tag `channel-sort`), following the same - * pattern as channel sections: NIP-44 encrypted-to-self content, debounced - * writes, and whole-blob last-write-wins. The sort map is a compact, - * low-frequency preference blob, so whole-blob LWW (like sections) is - * sufficient — per-key merge (like stars/mutes) would be unnecessary - * complexity here. - * - * The durable lane (outbox + generation/CAS + bounded retry) is mirrored from - * sections so a sort edit made just before quit/community-switch, or one whose - * publish times out, self-heals rather than being silently dropped. - */ -export class ChannelSortSyncManager { - private pubkey: string; - private relayUrl: string; - private debounceTimer: number | null = null; - private retryTimer: number | null = null; - private retryDelayMs = RETRY_BASE_MS; - private lastRemoteCreatedAt: number; - // Canonical best head observed so far (`created_at DESC, id ASC`). Frozen into - // a per-edit baseline at publishSortPrefs so the pre-publish check can tell - // whether the head advanced *since the edit was queued*, independent of the - // mutable watermark that a live event during the debounce window may advance. - private lastRemoteHead: PublishBaseline = { createdAt: 0, eventId: "" }; - // The canonical head this pending edit is racing against, frozen when the edit - // was queued and advanced ONLY by our own successful publishes. - private publishBaseline: PublishBaseline = { createdAt: 0, eventId: "" }; - private pendingStore: ChannelSortStore | null = null; - // Monotonic id for the current pending edit. Every publishSortPrefs() bumps - // it; every scheduled publish/retry captures the value it was queued for. A - // completion (success, adopt, or no-op) may only clear pending state via - // compare-and-swap on this generation, so an older in-flight publish can - // never erase a newer edit that arrived while it was in flight. - private pendingGeneration = 0; - // Publish cycles are serialized: at most one runs at a time. A newer edit - // queued while a cycle is in flight defers; the in-flight cycle's completion - // re-drives it. This kills the cross-generation race class by construction. - private publishInFlight = false; - // Event ids we signed and sent to the relay but whose ACK never arrived. The - // relay MAY have accepted such a write, so if a later cycle's pre-publish - // fetch returns a head whose id is in this map, that head is OUR OWN accepted - // predecessor — fold it forward and publish above it, rather than adopting it - // and erasing the queued edit. - // Maps attempt id → pendingGeneration at the time of signing, so - // foldSupersedingAttemptWinner can require the baseline was poisoned by a - // strictly prior generation (gen < pendingGeneration). - private ambiguousAttemptIds = new Map(); - private lastPublishedStore: ChannelSortStore | null = null; - private destroyed = false; - // Set by the hook so an adopted remote head (local edit lost whole-blob LWW) - // is written through to React state + localStorage. - private onRemoteAdopted: ((remote: RemoteSortPrefs) => void) | null = null; - +export class ChannelSortSyncManager extends WholeBlobSyncManager { constructor(pubkey: string, relayUrl: string) { - this.pubkey = pubkey; - this.relayUrl = relayUrl; - this.lastRemoteCreatedAt = readWatermark(pubkey, BLOB_TYPE, relayUrl); - } - - /** Register the hook's adopt-remote sink (write-through to UI + storage). */ - setOnRemoteAdopted(cb: (remote: RemoteSortPrefs) => void): void { - this.onRemoteAdopted = cb; - } - - async fetchRemoteSortPrefs(): Promise> { - try { - const events = await relayClient.fetchEvents({ - kinds: [KIND_CHANNEL_SORT], - authors: [this.pubkey], - "#d": [D_TAG], - limit: 1, - }); - if (events.length === 0 || events[0].pubkey !== this.pubkey) { - return { status: "absent" }; - } - const event = events[0]; - // An event exists — record its created_at regardless of whether we can - // decrypt it, so seed-publish is blocked even when the payload is - // unreadable (e.g. wrong key). - this.recordRemoteHead(event.created_at, event.id); - const result = await decryptAndParse(event); - if (!result) { - return { status: "failed", createdAt: event.created_at }; - } - return { - status: "found", - data: result, - createdAt: result.createdAt, - eventId: result.eventId, - }; - } catch { - return { status: "failed" }; - } - } - - /** Update in-memory + persisted watermark and the canonical head tuple. */ - private recordRemoteHead(createdAt: number, eventId: string): void { - if (createdAt > this.lastRemoteCreatedAt) { - this.lastRemoteCreatedAt = createdAt; - } - // Track the canonical-best head (`created_at DESC, id ASC`): a later head - // always wins; a same-second head wins only with a strictly-lower id. - if ( - createdAt > this.lastRemoteHead.createdAt || - (createdAt === this.lastRemoteHead.createdAt && - (this.lastRemoteHead.eventId === "" || - eventId < this.lastRemoteHead.eventId)) - ) { - this.lastRemoteHead = { createdAt, eventId }; - } - advanceWatermark(this.pubkey, BLOB_TYPE, this.relayUrl, createdAt); - } - - cancelPendingPublish(): void { - if (this.debounceTimer !== null) { - window.clearTimeout(this.debounceTimer); - this.debounceTimer = null; - } - if (this.retryTimer !== null) { - window.clearTimeout(this.retryTimer); - this.retryTimer = null; - } - } - - getPendingStore(): ChannelSortStore | null { - return this.pendingStore; - } - - /** - * Re-drive the CURRENT pending edit's publish immediately without opening a - * new generation — used by the reconnect handler. Calling the public - * `publishSortPrefs()` there would bump the generation and reset - * `publishBaseline` to the just-fetched head, so a remote that won LWW while - * the edit was pending would be published over instead of adopted (Carl P1). - * Waking the existing generation keeps the frozen baseline, so the debounced - * cycle's pre-publish check still adopts a genuinely-advanced remote. No-op - * when nothing is pending. - */ - retryPendingPublish(): void { - if (this.pendingStore === null) return; - this.cancelPendingPublish(); - this.startCycle(); - } - - /** True while an unpublished local edit is queued (debouncing or retrying). */ - hasPendingEdit(): boolean { - return this.pendingStore !== null; - } - - /** - * When the current `publishBaseline` was frozen against one of our own - * attempted (but non-retained) event ids and `remote` is the same-second - * lower-id peer that won the collision, the baseline is poisoned: the - * next pre-publish check would see the winner as having advanced past our - * phantom attempt and adopt the edit away (Carl P1 / same-second variants). - * Folding the winner in repairs the baseline so the pre-publish check sees - * the true retained head and publishes above it. - * - * Scoped to same-second lower-id only: a strictly-later remote is a genuine - * advance that the pending edit must still adopt (pass-2 invariant). - * - * Requires the baseline attempt to be prior-generation (gen < pendingGeneration): - * if no newer edit was queued, the current generation's own in-flight attempt - * cannot trigger the fold — the loser must adopt (LWW correctness). - * The post-ACK ordering is load-bearing for causal soundness: because A's - * ACK returned before B queued, the peer winner pre-existed B, so preserving - * B is causally justified. If B queues while A's ACK is pending, the peer may - * have been accepted after B and is a genuine remote advance B must adopt — - * B's baseline will not equal A's attempt id in that case, so the fold - * does not fire (correct behaviour, by the causal argument above). - * - * Returns true when the fold was applied (caller may use for branching). - */ - private foldSupersedingAttemptWinner(remote: { - createdAt: number; - eventId: string; - }): boolean { - const attempt = this.ambiguousAttemptIds.get(this.publishBaseline.eventId); - if ( - this.publishBaseline.eventId !== "" && - attempt !== undefined && - attempt < this.pendingGeneration && - remote.createdAt === this.publishBaseline.createdAt && - remote.eventId < this.publishBaseline.eventId - ) { - this.publishBaseline = canonicalMax(this.publishBaseline, { - createdAt: remote.createdAt, - eventId: remote.eventId, - }); - return true; - } - return false; - } - - /** - * Adopt a remote store that superseded a local edit: hand it to the hook for - * write-through, advance the watermark, and drop the losing pending edit — - * including the durable outbox, so the outbox can never replay an edit that - * adopt just decided lost. - * - * Compare-and-swap on `gen`: if a newer edit arrived while this publish was in - * flight, the generation has moved on and that newer edit is the latest writer - * — a stale adopt must not clear its pending state or overwrite its optimistic - * UI. We still advance the watermark (monotonic and always safe). - */ - private adoptRemote(remote: RemoteSortPrefs, gen: number): void { - this.recordRemoteHead(remote.createdAt, remote.eventId); - if (gen !== this.pendingGeneration) { - // A newer edit is pending. Its baseline may have been frozen against our - // own attempted (non-retained) event id if it was queued after the ACK - // but before confirmation resolved. Repair any poisoned baseline so the - // newer edit's pre-publish check publishes above the true retained head - // rather than adopting the winner away (Carl P1). - this.foldSupersedingAttemptWinner(remote); - return; - } - this.pendingStore = null; - clearChannelSortOutbox(this.pubkey, this.relayUrl); - this.lastPublishedStore = remote.store; - if (this.destroyed) return; - this.onRemoteAdopted?.(remote); - } - - /** - * Clear the in-memory pending edit and this window's own durable outbox key — - * but only if the completing publish still owns the current generation. A - * publish for an older edit that finishes after a newer edit was queued must - * leave the newer edit (and its retry state) untouched. - */ - private discardPending(gen: number): void { - if (gen !== this.pendingGeneration) return; - this.pendingStore = null; - clearChannelSortOutbox(this.pubkey, this.relayUrl); - } - - publishSortPrefs(store: ChannelSortStore): boolean { - this.pendingStore = store; - ++this.pendingGeneration; - // Freeze the canonical head this edit is racing against at queue time so a - // live event applied during the debounce window (which advances the mutable - // watermark) cannot make the pre-publish comparison see equality and fall - // through to a publish that overwrites a remote that became head after this - // edit was queued. The baseline only advances via our own successful - // publishes, so a prior generation's own accepted write is folded in rather - // than mistaken for a competing remote. - this.publishBaseline = { ...this.lastRemoteHead }; - // Persist synchronously so an edit made <2s before quit/community-switch - // survives teardown and resumes on next mount (durable outbox). This - // window's own key is the only one written — a single unconditional - // setItem, never a shared-key read-modify-write; `queuedAt` stamps it so - // resume replays only the newest queued blob (whole-blob LWW). The returned - // flag reports whether the intent is now durably held, so a legacy replay - // can gate its consumption marker on a proven transfer. - const durable = writeChannelSortOutbox(this.pubkey, store, this.relayUrl); - if (this.debounceTimer !== null) { - window.clearTimeout(this.debounceTimer); - } - // A fresh edit supersedes any retry scheduled for the previous generation. - if (this.retryTimer !== null) { - window.clearTimeout(this.retryTimer); - this.retryTimer = null; - } - this.retryDelayMs = RETRY_BASE_MS; - this.debounceTimer = window.setTimeout(() => { - this.debounceTimer = null; - this.startCycle(); - }, DEBOUNCE_MS); - return durable; - } - - /** - * Serialize publish cycles: at most one runs at a time. A debounce/retry timer - * that fires while a cycle is in flight defers — the in-flight cycle's - * completion re-drives if a pending edit still needs publishing. - */ - private startCycle(): void { - if (this.destroyed || this.pendingStore === null) return; - if (this.publishInFlight) return; - const store = this.pendingStore; - const gen = this.pendingGeneration; - this.publishInFlight = true; - void this.doPublish(store, gen).finally(() => { - this.publishInFlight = false; - if ( - !this.destroyed && - this.pendingStore !== null && - this.debounceTimer === null && - this.retryTimer === null - ) { - this.startCycle(); - } - }); - } - - private async fetchOwnBlobBeforePublish( - store: ChannelSortStore, - ): Promise { - try { - const events = await relayClient.fetchEvents({ - kinds: [KIND_CHANNEL_SORT], - authors: [this.pubkey], - "#d": [D_TAG], - limit: 1, - }); - if (events.length === 0 || events[0].pubkey !== this.pubkey) - return { kind: "publish", store }; - const event = events[0]; - const remote = await decryptAndParse(event); - // Record the head after decrypt attempt so the watermark/head-tuple - // advance even for an undecryptable payload. - this.recordRemoteHead(event.created_at, event.id); - // A head event exists but we could not read it (future schema, transient - // decrypt fault, malformed payload). We cannot decide adopt-or-publish - // against state we cannot inspect, so retain the pending edit and retry - // rather than blindly overwrite authoritative state (Carl P1). - if (!remote) return { kind: "retain" }; - // Sort prefs use whole-blob LWW. Compare the fetched head against the - // baseline frozen when this edit was queued — NOT the live watermark. If - // the canonical head advanced since the edit began, the local edit lost - // and is adopted-away rather than republished over it. - if (remoteAdvancedSince(remote, this.publishBaseline)) { - // Unless the advancing head is a prior publish of OURS whose ACK was - // lost: it is our own accepted predecessor, not a competing remote. - // Fold it into the baseline and publish above it so the queued edit - // survives instead of adopting our own stale write away. - if (this.ambiguousAttemptIds.has(remote.eventId)) { - this.publishBaseline = canonicalMax(this.publishBaseline, { - createdAt: remote.createdAt, - eventId: remote.eventId, - }); - return { kind: "publish", store }; - } - // Same-second lower-id peer winner that superseded our own attempted - // head: baseline.eventId is our attempted id (still in - // ambiguousAttemptIds because the adopt path never clears it) and the - // relay retained the peer's lower-id blob instead. This arises when - // confirmation returned retain/failed and the retry pre-publish fetch - // sees the winner directly with the original poisoned baseline still in - // place. Fold the winner in so this edit publishes above the true - // retained head rather than adopting it away (Carl P1). - if (this.foldSupersedingAttemptWinner(remote)) { - return { kind: "publish", store }; - } - return { kind: "adopt", remote }; - } - return { kind: "publish", store }; - } catch { - // The pre-publish fetch itself failed (timeout / auth / socket) — this is - // NOT proof that no head exists. Publishing here would sign above a stale - // watermark and could erase an unseen newer head during a transient - // outage. Retain the durable pending edit and retry rather than overwrite - // state we could not read (Carl P1). - return { kind: "retain" }; - } - } - - /** - * After a publish OK, fetch the authoritative retained head and decide whether - * our event actually became it. The relay OKs a superseded NIP-33 write as a - * no-op (`Duplicate`), so a resolved publish is NOT proof of retention: two - * windows stamping the same `created_at` both get OK while only the lower - * event id is retained (Carl P1). Only an exact event-id match proves our - * whole-blob write won LWW. - * - * - head id === ours → `confirmed` - * - head is our own accepted prior → `retain` (a lost-ACK / same-second - * attempt (ambiguousAttemptIds) lower-id attempt of ours won; the retry - * folds it forward and republishes above) - * - head is a different readable → `adopt` (our blob lost; adopt the - * event winner so UI and relay converge) - * - fetch failed / no head / head → `retain` (retention unprovable or the - * unreadable winner unreadable — keep the durable - * edit and retry) - */ - private async confirmRetainedHead( - ourEventId: string, - ): Promise { - try { - const events = await relayClient.fetchEvents({ - kinds: [KIND_CHANNEL_SORT], - authors: [this.pubkey], - "#d": [D_TAG], - limit: 1, - }); - if (events.length === 0 || events[0].pubkey !== this.pubkey) - return { kind: "retain" }; - const event = events[0]; - this.recordRemoteHead(event.created_at, event.id); - if (event.id === ourEventId) return { kind: "confirmed" }; - // A different event is the head. If it is one of OUR OWN accepted prior - // attempts (a lost-ACK write the relay retained, or a same-second lower-id - // attempt that won over this one), it is not a competing remote — retain - // so the retry's pre-publish folds it forward and publishes above it, - // rather than adopting our own older blob and dropping the current edit. - if (this.ambiguousAttemptIds.has(event.id)) return { kind: "retain" }; - const remote = await decryptAndParse(event); - if (!remote) return { kind: "retain" }; - return { kind: "adopt", remote }; - } catch { - return { kind: "retain" }; - } - } - - private isIdenticalToLastPublished(store: ChannelSortStore): boolean { - if (!this.lastPublishedStore) return false; - const lastGroups = this.lastPublishedStore.groups; - const currentGroups = store.groups; - const lastKeys = Object.keys(lastGroups); - const currentKeys = Object.keys(currentGroups); - if (lastKeys.length !== currentKeys.length) return false; - for (const key of currentKeys) { - if (lastGroups[key] !== currentGroups[key]) return false; - } - return true; - } - - /** Schedule a bounded-backoff retry of the retained pending edit. */ - private scheduleRetry(gen: number): void { - if (this.destroyed || this.pendingStore === null) return; - // A newer edit has superseded this one; its own timer owns the retry. - if (gen !== this.pendingGeneration) return; - if (this.retryTimer !== null) return; - const delay = this.retryDelayMs; - this.retryDelayMs = Math.min(this.retryDelayMs * 2, RETRY_MAX_MS); - this.retryTimer = window.setTimeout(() => { - this.retryTimer = null; - this.startCycle(); - }, delay); - } - - private async doPublish(store: ChannelSortStore, gen: number): Promise { - // A newer edit was queued after this publish was scheduled; it owns the - // pending state and will publish the latest store — abandon this stale run. - if (gen !== this.pendingGeneration) return; - try { - const decision = await this.fetchOwnBlobBeforePublish(store); - // Guard: manager may have been destroyed while fetchOwnBlobBeforePublish - // was awaited (community switch during in-flight fetch). - if (this.destroyed) return; - // A newer edit was queued while we awaited the pre-publish fetch. It owns - // convergence now; the serialized cycle re-drives for it once this run - // unwinds. - if (gen !== this.pendingGeneration) return; - if (decision.kind === "adopt") { - this.adoptRemote(decision.remote, gen); - return; - } - if (decision.kind === "retain") { - // Head exists but is unreadable — keep the durable pending edit and - // retry with backoff rather than overwrite state we could not inspect. - this.scheduleRetry(gen); - return; - } - const merged = decision.store; - if (this.isIdenticalToLastPublished(merged)) { - this.discardPending(gen); - return; - } - const payload = { + super(pubkey, relayUrl, { + kind: KIND_CHANNEL_SORT, + dTag: "channel-sort", + logPrefix: "channelSortSync", + parse: parseChannelSortPayload, + serializePayload: (store) => ({ version: 1, - groups: merged.groups, - }; - const ciphertext = await nip44EncryptToSelf(JSON.stringify(payload)); - // Clamp inside the relay's future-drift window so a skewed remote head - // can never make us stamp an unbounded future timestamp that wedges every - // subsequent publish; we adopt such a head on the next fetch instead. - const createdAt = clampPublishCreatedAt(this.lastRemoteCreatedAt); - const event = await signRelayEvent({ - kind: KIND_CHANNEL_SORT, - content: ciphertext, - createdAt, - tags: [ - ["d", D_TAG], - ["t", D_TAG], // relay discoverability; not used in our filters - ], - }); - // Final guard immediately before the network call — a newer edit may have - // been queued during the encrypt/sign await, or the manager destroyed. - if (this.destroyed || gen !== this.pendingGeneration) return; - // Record this signed id as an in-flight attempt of unknown fate before we - // send it. If the ACK is lost below, a later cycle that fetches this id as - // the head recognises it as our own accepted write and folds it forward - // rather than adopting it away. - this.ambiguousAttemptIds.set(event.id, gen); - await relayClient.publishEvent( - event, - "Timed out publishing channel sort preferences.", - "Failed to publish channel sort preferences.", - ); - this.recordRemoteHead(event.created_at, event.id); - // A publish OK is NOT proof of retention: the relay OKs a superseded - // NIP-33 write as a no-op (`Duplicate`), so two windows stamping the same - // `created_at` both get OK while only the lower event id is retained (Carl - // P1). Fetch the authoritative head; only an exact id match proves our - // whole-blob write won. Anything else must keep the durable edit — never - // fold a nonexistent head into the baseline, which would poison the next - // edit into adopting the true winner away. - if (this.destroyed) return; - const confirmation = await this.confirmRetainedHead(event.id); - if (this.destroyed) return; - if (confirmation.kind === "adopt") { - // A different readable event is the head — our blob lost the same-second - // collision. Adopt the winner so UI and relay converge (CAS on gen - // inside adoptRemote leaves a newer edit untouched). - this.adoptRemote(confirmation.remote, gen); - return; - } - if (confirmation.kind === "retain") { - // Retention unprovable (fetch failed / no head), or the head is our own - // accepted prior attempt whose ACK was lost. Keep the durable edit and - // retry: the retry's pre-publish folds our own accepted predecessor - // forward (ambiguousAttemptIds) and republishes above it. Leave - // ambiguousAttemptIds intact so that fold still recognises our writes. - this.scheduleRetry(gen); - return; - } - // Confirmed: our event is the retained head. It dominates every prior - // attempt (`created_at DESC, id ASC`), so no earlier ambiguous id can ever - // be the canonical head again — clear the set to keep it bounded. - this.ambiguousAttemptIds.clear(); - // Fold our own accepted head into the pending edit's baseline — - // unconditional across generations so a newer edit's pre-publish check - // does not mistake OUR prior publish for a competing remote and adopt it - // away. Genuine remotes never fold in here. - this.publishBaseline = canonicalMax(this.publishBaseline, { - createdAt: event.created_at, - eventId: event.id, - }); - // Only claim this store as the published head if it is still the current - // edit; a newer edit queued mid-flight owns lastPublishedStore now. - if (gen === this.pendingGeneration) { - this.lastPublishedStore = merged; - this.retryDelayMs = RETRY_BASE_MS; - } - this.discardPending(gen); - } catch (error) { - if (this.destroyed) return; - // Ambiguous outcome: the publish promise rejected (timeout / socket - // error), but the relay may already have accepted the write before the - // ACK was lost. Keep the pending edit and retry with backoff. The attempt - // id stays in ambiguousAttemptIds: if the relay did accept it, a later - // cycle that fetches this id as the head folds it forward as our own - // accepted predecessor instead of adopting it away. - console.warn("[channelSortSync] publish failed:", error); - this.scheduleRetry(gen); - } - } - - async subscribeToSortPrefs( - onUpdate: (remote: RemoteSortPrefs) => void, - ): Promise<() => Promise> { - return relayClient.subscribeLive( - { - kinds: [KIND_CHANNEL_SORT], - authors: [this.pubkey], - "#d": [D_TAG], - limit: 0, - }, - (event: RelayEvent) => { - if (event.pubkey !== this.pubkey) return; - // Record the raw head before decrypt so an undecryptable live event - // still advances the watermark and blocks future seed-publish. - this.recordRemoteHead(event.created_at, event.id); - void decryptAndParse(event).then((result) => { - if (result) { - onUpdate(result); - } - }); - }, - ); - } - - /** - * Fetches the remote blob on first mount, records the remote head, and - * delegates the seed/hold/apply-remote decision to `runBootstrap`. - */ - async bootstrap(localStore: ChannelSortStore) { - const fetchResult = await this.fetchRemoteSortPrefs(); - return runBootstrap({ - fetchResult, - lastHead: this.lastRemoteCreatedAt, - localStore, + groups: store.groups, + }), + writeOutbox: writeChannelSortOutbox, + clearOutbox: clearChannelSortOutbox, + storesEqual: sortStoresEqual, isLocalNonEmpty: (s) => Object.keys(s.groups).length > 0, - publishFn: (s) => this.publishSortPrefs(s), }); } - destroy(): void { - // Cancel any pending publish and mark this manager as destroyed so any - // in-flight doPublish() calls abort before reaching relayClient. - // Debounce-window changes are NOT lost: publishSortPrefs persisted them to - // the durable outbox synchronously, and the next mount resumes them. - // Flushing here is still avoided — it could publish relay A's sort prefs to - // relay B via the shared relayClient singleton. - this.destroyed = true; - this.cancelPendingPublish(); - this.pendingStore = null; + /** Publish a sort prefs store, debounced to 2s. Returns whether the intent is durably held. */ + publishSortPrefs(store: ChannelSortStore): boolean { + return this.publish(store); + } + + /** Fetch the current remote head for this pubkey's sort-prefs blob. */ + fetchRemoteSortPrefs() { + return this.fetchRemoteBlob(); + } + + /** Subscribe to live relay events for this pubkey's sort-prefs blob. */ + subscribeToSortPrefs(onUpdate: (remote: RemoteSortPrefs) => void) { + return this.subscribeLive(onUpdate); } } diff --git a/desktop/src/features/sidebar/lib/channelStarsSync.ts b/desktop/src/features/sidebar/lib/channelStarsSync.ts index bec5a0eb34..1464a27dcb 100644 --- a/desktop/src/features/sidebar/lib/channelStarsSync.ts +++ b/desktop/src/features/sidebar/lib/channelStarsSync.ts @@ -1,10 +1,3 @@ -import { relayClient } from "@/shared/api/relayClient"; -import { - nip44DecryptFromSelf, - nip44EncryptToSelf, - signRelayEvent, -} from "@/shared/api/tauri"; -import type { RelayEvent } from "@/shared/api/types"; import { KIND_CHANNEL_STARS } from "@/shared/constants/kinds"; import { clearChannelStarsOutbox, @@ -15,464 +8,81 @@ import { type ChannelStarStore, } from "./channelStarsStorage"; import { - advanceWatermark, - clampPublishCreatedAt, - readWatermark, - runBootstrap, - type FetchResult, -} from "./sidebarSyncWatermark"; + MergeLaneSyncManager, + type RemoteMergeBlob, +} from "./mergeLaneSyncManager"; -const D_TAG = "channel-stars"; -const BLOB_TYPE = D_TAG; -const DEBOUNCE_MS = 2_000; +export type RemoteStars = RemoteMergeBlob; -// Bounded backoff for a retained pending edit whose publish failed transiently -// (timeout / socket error) on an otherwise-healthy socket, so it does not wait -// for a reconnect that may never fire. -const RETRY_BASE_MS = 2_000; -const RETRY_MAX_MS = 30_000; - -export type RemoteStars = { - store: ChannelStarStore; - createdAt: number; - eventId: string; -}; - -/** - * Pre-publish decision for the merge lane. `publish` carries the max-merged - * store to write; `retain` keeps the durable pending edit and retries. Retain - * covers two unreadable cases where publishing would blindly overwrite state we - * could not inspect (Carl P1): a pre-publish fetch that THREW (timeout / auth / - * socket — not proof no head exists), and an existing head that failed - * decryption/parsing (a max-merge is only safe once both operands were read). - */ -type MergePublishDecision = - | { kind: "publish"; store: ChannelStarStore } - | { kind: "retain" }; - -async function decryptAndParse(event: RelayEvent): Promise { - try { - const plaintext = await nip44DecryptFromSelf(event.content); - const store = parseStarPayload(JSON.parse(plaintext)); - if (!store) return null; - return { store, createdAt: event.created_at, eventId: event.id }; - } catch { - return null; - } -} - -export class ChannelStarSyncManager { - private pubkey: string; - private relayUrl: string; - private debounceTimer: number | null = null; - private retryTimer: number | null = null; - private retryDelayMs = RETRY_BASE_MS; - private lastRemoteCreatedAt: number; - private pendingStore: ChannelStarStore | null = null; - // Monotonic id for the current pending edit. Every publishStars() bumps it; - // every scheduled publish/retry captures the value it was queued for. A - // completion (success or no-op) may only clear pending state via - // compare-and-swap on this generation, so an older in-flight publish can - // never erase a newer edit that arrived while it was in flight. - private pendingGeneration = 0; - // Publish cycles are serialized: at most one runs at a time. A newer edit - // queued while a cycle is in flight defers; the in-flight cycle's completion - // re-drives it. Serialization guarantees there is never more than one - // fetch/publish sequence touching shared manager state. - private publishInFlight = false; - private lastPublishedStore: ChannelStarStore | null = null; - private destroyed = false; - // Per-channel high-water of every `rev` and `updatedAt` this manager has - // observed (bootstrap, live, reconnect, reconcile, pre-publish, cross-window - // storage, and initial persisted state). A click reads both so its minted - // `updatedAt = max(now, maxUpdatedAtSeen)` never regresses below observed - // state (the read-state logical-monotonic idiom), and `rev = maxRevSeen + 1` - // wins the resulting same-second tie. - private highWater = new Map(); - - constructor(pubkey: string, relayUrl: string) { - this.pubkey = pubkey; - this.relayUrl = relayUrl; - this.lastRemoteCreatedAt = readWatermark(pubkey, BLOB_TYPE, relayUrl); - } - - /** - * Ingest a store into the per-channel high-water. Called synchronously before - * any merge is applied to React state, so a click that follows reads a current - * watermark on both dimensions. Monotonic (`Math.max`) and idempotent. - */ - observe(store: ChannelStarStore): void { - for (const [id, entry] of Object.entries(store.channels)) { - const cur = this.highWater.get(id) ?? { rev: 0, updatedAt: 0 }; - this.highWater.set(id, { - rev: Math.max(cur.rev, entry.rev), - updatedAt: Math.max(cur.updatedAt, entry.updatedAt), - }); - } - } - - maxRevSeen(id: string): number { - return this.highWater.get(id)?.rev ?? 0; - } - - maxUpdatedAtSeen(id: string): number { - return this.highWater.get(id)?.updatedAt ?? 0; - } - - async fetchRemoteStars(): Promise> { - try { - const events = await relayClient.fetchEvents({ - kinds: [KIND_CHANNEL_STARS], - authors: [this.pubkey], - "#d": [D_TAG], - limit: 1, - }); - if (events.length === 0 || events[0].pubkey !== this.pubkey) { - return { status: "absent" }; - } - const event = events[0]; - this.recordRemoteHead(event.created_at); - const result = await decryptAndParse(event); - if (!result) { - return { status: "failed", createdAt: event.created_at }; - } - this.observe(result.store); - return { - status: "found", - data: result, - createdAt: result.createdAt, - eventId: result.eventId, - }; - } catch { - return { status: "failed" }; - } - } - - private recordRemoteHead(createdAt: number): void { - if (createdAt > this.lastRemoteCreatedAt) { - this.lastRemoteCreatedAt = createdAt; - } - advanceWatermark(this.pubkey, BLOB_TYPE, this.relayUrl, createdAt); - } - - cancelPendingStarPublish(): void { - if (this.debounceTimer !== null) { - window.clearTimeout(this.debounceTimer); - this.debounceTimer = null; - } - if (this.retryTimer !== null) { - window.clearTimeout(this.retryTimer); - this.retryTimer = null; - } - } - - getPendingStarStore(): ChannelStarStore | null { - return this.pendingStore; - } - - publishStars(store: ChannelStarStore): void { - this.pendingStore = store; - ++this.pendingGeneration; - // Persist synchronously so a click made <2s before quit/community-switch - // survives teardown and resumes on next mount (durable outbox). This - // window's own key is the only one written — a single unconditional - // setItem, never a shared-key read-modify-write. - writeChannelStarsOutbox(this.pubkey, store, this.relayUrl); - if (this.debounceTimer !== null) { - window.clearTimeout(this.debounceTimer); - } - // A fresh edit supersedes any retry scheduled for the previous generation. - if (this.retryTimer !== null) { - window.clearTimeout(this.retryTimer); - this.retryTimer = null; - } - this.retryDelayMs = RETRY_BASE_MS; - this.debounceTimer = window.setTimeout(() => { - this.debounceTimer = null; - this.startCycle(); - }, DEBOUNCE_MS); - } - - /** - * Serialize publish cycles: at most one runs at a time. A debounce/retry timer - * that fires while a cycle is in flight defers — the in-flight cycle's - * completion re-drives if a pending edit still needs publishing. A newer edit - * queued during a cycle cannot start its own concurrent cycle, so a stale - * generation can never publish after a newer edit exists. - */ - private startCycle(): void { - if (this.destroyed || this.pendingStore === null) return; - if (this.publishInFlight) return; - const store = this.pendingStore; - const gen = this.pendingGeneration; - this.publishInFlight = true; - void this.doPublish(store, gen).finally(() => { - this.publishInFlight = false; - if ( - !this.destroyed && - this.pendingStore !== null && - this.debounceTimer === null && - this.retryTimer === null - ) { - this.startCycle(); - } - }); - } - - private async fetchOwnBlobBeforePublish( - store: ChannelStarStore, - ): Promise { - let events: RelayEvent[]; - try { - events = await relayClient.fetchEvents({ - kinds: [KIND_CHANNEL_STARS], - authors: [this.pubkey], - "#d": [D_TAG], - limit: 1, - }); - } catch { - // The fetch itself failed (timeout / auth / socket) — NOT proof that no - // head exists. Publishing the local store here could erase an unseen - // newer head during a transient outage, so retain and retry (Carl P1). - return { kind: "retain" }; - } - // A successful fetch that proves no head exists: publish the local store. - if (events.length === 0 || events[0].pubkey !== this.pubkey) - return { kind: "publish", store }; - const event = events[0]; - // Record the raw head before decrypt on the pre-publish path too. - this.recordRemoteHead(event.created_at); - const remote = await decryptAndParse(event); - // A head exists but could not be read (decrypt fault / malformed / future - // schema). A max-merge is only safe once both operands were actually read, - // so retain rather than publish the local store over an uninspectable head - // (Carl P1). The whole-blob lanes retain here too; the durable outbox and - // bounded retry resume normal resolution once a readable head returns. - if (!remote) return { kind: "retain" }; - this.observe(remote.store); - // Max-merge: the local edit's per-entry winners survive by construction and - // any newer remote entries fold in, so no adopt step is needed. - return { kind: "publish", store: mergeStores(store, remote.store) }; - } - - /** - * After a publish OK, fetch the authoritative retained head and report whether - * it subsumes the store we attempted to write. The relay returns OK for a - * superseded NIP-33 write as a no-op, so two windows racing distinct blobs - * from the same head can both get OK while only one blob is retained (Carl - * P1). Only a retained head that subsumes our store proves our click is - * durable on the relay; otherwise the outbox must be kept and retried. - * - * A failed/absent/undecryptable fetch cannot prove subsumption and returns - * false, so the durable click is retained and retried rather than dropped. - * Records the fetched head and observes it so a subsequent merge/retry folds - * the retained blob in. - */ - private async confirmRetainedHeadSubsumes( - store: ChannelStarStore, - ): Promise { - try { - const events = await relayClient.fetchEvents({ - kinds: [KIND_CHANNEL_STARS], - authors: [this.pubkey], - "#d": [D_TAG], - limit: 1, - }); - if (events.length === 0 || events[0].pubkey !== this.pubkey) return false; - const event = events[0]; - this.recordRemoteHead(event.created_at); - const remote = await decryptAndParse(event); - if (!remote) return false; - this.observe(remote.store); - return isStarsStoreSubsumedBy(store, remote.store); - } catch { +function starsStoresEqual(a: ChannelStarStore, b: ChannelStarStore): boolean { + const aKeys = Object.keys(a.channels); + const bKeys = Object.keys(b.channels); + if (aKeys.length !== bKeys.length) return false; + for (const key of aKeys) { + const ae = a.channels[key]; + const be = b.channels[key]; + if ( + !ae || + !be || + ae.starred !== be.starred || + ae.updatedAt !== be.updatedAt || + ae.rev !== be.rev + ) return false; - } } + return true; +} - private isIdenticalToLastPublished(store: ChannelStarStore): boolean { - if (!this.lastPublishedStore) return false; - const lastKeys = Object.keys(this.lastPublishedStore.channels); - const currentKeys = Object.keys(store.channels); - if (lastKeys.length !== currentKeys.length) return false; - for (const key of currentKeys) { - const last = this.lastPublishedStore.channels[key]; - const current = store.channels[key]; - if ( - !last || - last.starred !== current.starred || - last.updatedAt !== current.updatedAt || - last.rev !== current.rev - ) - return false; - } - return true; - } - - /** - * Clear the in-memory pending edit and this window's own durable outbox key — - * but only if the completing publish still owns the current generation. A - * publish for an older edit that finishes after a newer edit was queued must - * leave the newer edit (and its retry state) untouched. - */ - private discardPending(gen: number): void { - if (gen !== this.pendingGeneration) return; - this.pendingStore = null; - clearChannelStarsOutbox(this.pubkey, this.relayUrl); - } - - /** Schedule a bounded-backoff retry of the retained pending edit. */ - private scheduleRetry(gen: number): void { - if (this.destroyed || this.pendingStore === null) return; - // A newer edit has superseded this one; its own timer owns the retry. - if (gen !== this.pendingGeneration) return; - if (this.retryTimer !== null) return; - const delay = this.retryDelayMs; - this.retryDelayMs = Math.min(this.retryDelayMs * 2, RETRY_MAX_MS); - this.retryTimer = window.setTimeout(() => { - this.retryTimer = null; - this.startCycle(); - }, delay); - } - - private async doPublish(store: ChannelStarStore, gen: number): Promise { - // A newer edit was queued after this publish was scheduled; it owns the - // pending state and will publish the latest store — abandon this stale run. - if (gen !== this.pendingGeneration) return; - try { - const decision = await this.fetchOwnBlobBeforePublish(store); - // Guard: manager may have been destroyed while fetchOwnBlobBeforePublish - // was awaited (community switch during in-flight fetch). - if (this.destroyed) return; - // A newer edit was queued while we awaited the pre-publish fetch. It owns - // convergence now; the serialized cycle re-drives for it once this run - // unwinds. - if (gen !== this.pendingGeneration) return; - // The pre-publish read failed or the head was unreadable — keep the - // durable pending edit and retry rather than publish over uninspectable - // state (Carl P1). - if (decision.kind === "retain") { - this.scheduleRetry(gen); - return; - } - const merged = decision.store; - if (this.isIdenticalToLastPublished(merged)) { - this.discardPending(gen); - return; - } - const payload = { - version: 1, - channels: merged.channels, - }; - const ciphertext = await nip44EncryptToSelf(JSON.stringify(payload)); - // Clamp inside the relay's future-drift window so a skewed remote head - // can never make us stamp an unbounded future timestamp that wedges every - // subsequent publish; we adopt such a head on the next fetch instead. - const createdAt = clampPublishCreatedAt(this.lastRemoteCreatedAt); - const event = await signRelayEvent({ - kind: KIND_CHANNEL_STARS, - content: ciphertext, - createdAt, - tags: [ - ["d", D_TAG], - ["t", D_TAG], // relay discoverability; not used in our filters - ], - }); - // Final guard immediately before the network call: a newer edit may have - // been queued during the encrypt/sign await, or the manager destroyed. - if (this.destroyed || gen !== this.pendingGeneration) return; - await relayClient.publishEvent( - event, - "Timed out publishing channel stars.", - "Failed to publish channel stars.", - ); - this.recordRemoteHead(event.created_at); - this.observe(merged); - // A publish OK is NOT proof of retention: the relay OKs a superseded - // NIP-33 write as a no-op, so a peer window racing a distinct blob from - // the same head can win retention while ours is silently dropped (Carl - // P1). Fetch the authoritative retained head and clear the durable outbox - // only when it subsumes what we wrote. If it does not (or the fetch - // could not prove it), keep the outbox and retry so the click is never - // lost; the retry's pre-publish max-merge folds the retained blob in. - if (this.destroyed) return; - const confirmed = await this.confirmRetainedHeadSubsumes(merged); - if (this.destroyed) return; - if (!confirmed) { - this.scheduleRetry(gen); - return; - } - // Only claim this store as the published head if it is still the current - // edit; a newer edit queued mid-flight owns lastPublishedStore now. - if (gen === this.pendingGeneration) { - this.lastPublishedStore = merged; - this.retryDelayMs = RETRY_BASE_MS; - } - this.discardPending(gen); - } catch (error) { - if (this.destroyed) return; - // Transient publish failure (timeout / socket error). Keep the pending - // edit and retry with backoff rather than waiting for a reconnect that a - // healthy socket never fires. Max-merge makes a duplicate publish - // idempotent, so a lost-ACK write that the relay actually accepted is - // harmless to re-send. - console.warn("[channelStarsSync] publish failed:", error); - this.scheduleRetry(gen); - } - } - - async subscribeToStars( - onUpdate: (remote: RemoteStars) => void, - ): Promise<() => Promise> { - return relayClient.subscribeLive( - { - kinds: [KIND_CHANNEL_STARS], - authors: [this.pubkey], - "#d": [D_TAG], - limit: 0, +export class ChannelStarSyncManager extends MergeLaneSyncManager { + constructor(pubkey: string, relayUrl: string) { + super(pubkey, relayUrl, { + kind: KIND_CHANNEL_STARS, + dTag: "channel-stars", + logPrefix: "channelStarsSync", + publishTimeoutMsg: "Timed out publishing channel stars.", + publishErrorMsg: "Failed to publish channel stars.", + parse: parseStarPayload, + serializePayload: (store) => ({ version: 1, channels: store.channels }), + mergeWithRemote: mergeStores, + isSubsumedBy: isStarsStoreSubsumedBy, + storesEqual: starsStoresEqual, + observe: (highWater, store) => { + for (const [id, entry] of Object.entries(store.channels)) { + const cur = highWater.get(id) ?? { rev: 0, updatedAt: 0 }; + highWater.set(id, { + rev: Math.max(cur.rev, entry.rev), + updatedAt: Math.max(cur.updatedAt, entry.updatedAt), + }); + } }, - (event: RelayEvent) => { - if (event.pubkey !== this.pubkey) return; - // Record the raw head before decrypt so an undecryptable live event - // still advances the watermark and blocks future seed-publish. - this.recordRemoteHead(event.created_at); - void decryptAndParse(event).then((result) => { - if (result) { - this.observe(result.store); - onUpdate(result); - } - }); - }, - ); - } - - /** - * Fetches the remote blob on first mount, records the remote head, and - * delegates the seed/hold/apply-remote decision to `runBootstrap`. - */ - async bootstrap(localStore: ChannelStarStore) { - // Seed the high-water from the caller's persisted local store so a click - // before the remote fetch resolves already reflects retained entries. - this.observe(localStore); - const fetchResult = await this.fetchRemoteStars(); - return runBootstrap({ - fetchResult, - lastHead: this.lastRemoteCreatedAt, - localStore, + writeOutbox: writeChannelStarsOutbox, + clearOutbox: clearChannelStarsOutbox, isLocalNonEmpty: (s) => Object.keys(s.channels).length > 0, - publishFn: (s) => this.publishStars(s), }); } - destroy(): void { - // Cancel any pending publish and mark this manager as destroyed so any - // in-flight doPublish() calls abort before reaching relayClient. - // Debounce-window changes are NOT lost: publishStars persisted them to the - // durable outbox synchronously, and the next mount resumes them. Flushing - // here is still avoided — it could publish relay A's state to relay B via - // the shared relayClient singleton. - this.destroyed = true; - this.cancelPendingStarPublish(); - this.pendingStore = null; + /** Publish a star store, debounced to 2s. */ + publishStars(store: ChannelStarStore): void { + this.publish(store); + } + + /** Fetch the current remote head for this pubkey's stars blob. */ + fetchRemoteStars() { + return this.fetchRemoteBlob(); + } + + /** Cancel any pending star publish (debounce or retry timer). */ + cancelPendingStarPublish(): void { + this.cancelPendingPublish(); + } + + /** The currently pending store, or null if nothing is queued. */ + getPendingStarStore(): ChannelStarStore | null { + return this.getPendingStore(); + } + + /** Subscribe to live relay events for this pubkey's stars blob. */ + subscribeToStars(onUpdate: (remote: RemoteStars) => void) { + return this.subscribeLive(onUpdate); } } diff --git a/desktop/src/features/sidebar/lib/mergeLaneSyncManager.ts b/desktop/src/features/sidebar/lib/mergeLaneSyncManager.ts new file mode 100644 index 0000000000..24a55ac0cb --- /dev/null +++ b/desktop/src/features/sidebar/lib/mergeLaneSyncManager.ts @@ -0,0 +1,495 @@ +import { relayClient } from "@/shared/api/relayClient"; +import { + nip44DecryptFromSelf, + nip44EncryptToSelf, + signRelayEvent, +} from "@/shared/api/tauri"; +import type { RelayEvent } from "@/shared/api/types"; +import { + advanceWatermark, + clampPublishCreatedAt, + readWatermark, + runBootstrap, + type FetchResult, +} from "./sidebarSyncWatermark"; + +const DEBOUNCE_MS = 2_000; +const RETRY_BASE_MS = 2_000; +const RETRY_MAX_MS = 30_000; + +export type RemoteMergeBlob = { + store: S; + createdAt: number; + eventId: string; +}; + +/** + * Pre-publish decision for the merge lane. `publish` carries the max-merged + * store to write; `retain` keeps the durable pending edit and retries. Retain + * covers two unreadable cases where publishing would blindly overwrite state we + * could not inspect (Carl P1): a pre-publish fetch that THREW (timeout / auth / + * socket — not proof no head exists), and an existing head that failed + * decryption/parsing (a max-merge is only safe once both operands were read). + */ +type MergePublishDecision = + | { kind: "publish"; store: S } + | { kind: "retain" }; + +/** + * Per-lane configuration for the shared merge-lane sync engine. + * + * `S` is the per-lane store type (ChannelStarStore, ChannelMuteStore, …). + * `E` is the per-entry type (the value type of `S`'s channel map). Used for + * the high-water observation logic. + */ +export type MergeLaneConfig = { + /** NIP-78 event kind for this lane. */ + kind: number; + /** NIP-78 `d` tag value used to scope the lane's blob. */ + dTag: string; + /** Human-readable prefix for console.warn messages. */ + logPrefix: string; + /** Publish timeout/failure message. */ + publishTimeoutMsg: string; + /** Publish error message. */ + publishErrorMsg: string; + /** Parse decrypted JSON into a typed store, or null for unreadable payloads. */ + parse: (json: unknown) => S | null; + /** Serialize a store into the JSON object to be encrypted and published. */ + serializePayload: (store: S) => unknown; + /** Max-merge a local pending store with the fetched remote, producing the store to publish. */ + mergeWithRemote: (local: S, remote: S) => S; + /** + * True when `retained` subsumes `attempted` — i.e. the retained relay head + * contains all the per-entry winners from the local write. Used to confirm + * a successful publish without a strict id match (the relay OKs superseded + * NIP-33 writes as no-ops, so two windows can both receive OK while only one + * blob is retained). + */ + isSubsumedBy: (attempted: S, retained: S) => boolean; + /** + * True when `a` is identical to `b` (used to skip no-op publishes against + * the last-published head). + */ + storesEqual: (a: S, b: S) => boolean; + /** Observe a store into the per-channel high-water (called synchronously before React state updates). */ + observe: ( + highWater: Map, + store: S, + ) => void; + /** Persist this window's pending edit to the durable outbox. */ + writeOutbox: (pubkey: string, store: S, relayUrl: string) => void; + /** Clear this window's own outbox key. */ + clearOutbox: (pubkey: string, relayUrl: string) => void; + /** True when the local store is non-empty and eligible for a first-sync seed-publish. */ + isLocalNonEmpty: (store: S) => boolean; +}; + +/** + * Shared merge-lane sync engine used by ChannelStarSyncManager and + * ChannelMuteSyncManager. Both lanes use the same generation-CAS, outbox, + * watermark, pre-publish max-merge, and retained-head subsumption confirmation + * — only lane config (kind, d-tag, store type, merge/subsumption logic, + * outbox functions) differs. + */ +export class MergeLaneSyncManager { + private pubkey: string; + private relayUrl: string; + protected config: MergeLaneConfig; + + private debounceTimer: number | null = null; + private retryTimer: number | null = null; + private retryDelayMs = RETRY_BASE_MS; + private lastRemoteCreatedAt: number; + private pendingStore: S | null = null; + // Monotonic id for the current pending edit. Every publish() bumps it; every + // scheduled publish/retry captures the value it was queued for. A completion + // (success or no-op) may only clear pending state via compare-and-swap on + // this generation, so an older in-flight publish can never erase a newer edit + // that arrived while it was in flight. + private pendingGeneration = 0; + // Publish cycles are serialized: at most one runs at a time. A newer edit + // queued while a cycle is in flight defers; the in-flight cycle's completion + // re-drives it. Serialization guarantees there is never more than one + // fetch/publish sequence touching shared manager state. + private publishInFlight = false; + private lastPublishedStore: S | null = null; + protected destroyed = false; + // Per-channel high-water of every `rev` and `updatedAt` this manager has + // observed (bootstrap, live, reconnect, reconcile, pre-publish, cross-window + // storage, and initial persisted state). A click reads both so its minted + // `updatedAt = max(now, maxUpdatedAtSeen)` never regresses below observed + // state (the read-state logical-monotonic idiom), and `rev = maxRevSeen + 1` + // wins the resulting same-second tie. + protected highWater = new Map(); + + constructor(pubkey: string, relayUrl: string, config: MergeLaneConfig) { + this.pubkey = pubkey; + this.relayUrl = relayUrl; + this.config = config; + this.lastRemoteCreatedAt = readWatermark(pubkey, config.dTag, relayUrl); + } + + /** + * Ingest a store into the per-channel high-water. Called synchronously before + * any merge is applied to React state, so a click that follows reads a current + * watermark on both dimensions. Monotonic (`Math.max`) and idempotent. + */ + observe(store: S): void { + this.config.observe(this.highWater, store); + } + + maxRevSeen(id: string): number { + return this.highWater.get(id)?.rev ?? 0; + } + + maxUpdatedAtSeen(id: string): number { + return this.highWater.get(id)?.updatedAt ?? 0; + } + + private async decryptAndParse( + event: RelayEvent, + ): Promise | null> { + try { + const plaintext = await nip44DecryptFromSelf(event.content); + const store = this.config.parse(JSON.parse(plaintext)); + if (!store) return null; + return { store, createdAt: event.created_at, eventId: event.id }; + } catch { + return null; + } + } + + async fetchRemoteBlob(): Promise>> { + try { + const events = await relayClient.fetchEvents({ + kinds: [this.config.kind], + authors: [this.pubkey], + "#d": [this.config.dTag], + limit: 1, + }); + if (events.length === 0 || events[0].pubkey !== this.pubkey) { + return { status: "absent" }; + } + const event = events[0]; + this.recordRemoteHead(event.created_at); + const result = await this.decryptAndParse(event); + if (!result) { + return { status: "failed", createdAt: event.created_at }; + } + this.observe(result.store); + return { + status: "found", + data: result, + createdAt: result.createdAt, + eventId: result.eventId, + }; + } catch { + return { status: "failed" }; + } + } + + private recordRemoteHead(createdAt: number): void { + if (createdAt > this.lastRemoteCreatedAt) { + this.lastRemoteCreatedAt = createdAt; + } + advanceWatermark(this.pubkey, this.config.dTag, this.relayUrl, createdAt); + } + + cancelPendingPublish(): void { + if (this.debounceTimer !== null) { + window.clearTimeout(this.debounceTimer); + this.debounceTimer = null; + } + if (this.retryTimer !== null) { + window.clearTimeout(this.retryTimer); + this.retryTimer = null; + } + } + + getPendingStore(): S | null { + return this.pendingStore; + } + + publish(store: S): void { + this.pendingStore = store; + ++this.pendingGeneration; + // Persist synchronously so a click made <2s before quit/community-switch + // survives teardown and resumes on next mount (durable outbox). This + // window's own key is the only one written — a single unconditional + // setItem, never a shared-key read-modify-write. + this.config.writeOutbox(this.pubkey, store, this.relayUrl); + if (this.debounceTimer !== null) { + window.clearTimeout(this.debounceTimer); + } + // A fresh edit supersedes any retry scheduled for the previous generation. + if (this.retryTimer !== null) { + window.clearTimeout(this.retryTimer); + this.retryTimer = null; + } + this.retryDelayMs = RETRY_BASE_MS; + this.debounceTimer = window.setTimeout(() => { + this.debounceTimer = null; + this.startCycle(); + }, DEBOUNCE_MS); + } + + /** + * Serialize publish cycles: at most one runs at a time. A debounce/retry + * timer that fires while a cycle is in flight defers — the in-flight cycle's + * completion re-drives if a pending edit still needs publishing. A newer edit + * queued during a cycle cannot start its own concurrent cycle, so a stale + * generation can never publish after a newer edit exists. + */ + private startCycle(): void { + if (this.destroyed || this.pendingStore === null) return; + if (this.publishInFlight) return; + const store = this.pendingStore; + const gen = this.pendingGeneration; + this.publishInFlight = true; + void this.doPublish(store, gen).finally(() => { + this.publishInFlight = false; + if ( + !this.destroyed && + this.pendingStore !== null && + this.debounceTimer === null && + this.retryTimer === null + ) { + this.startCycle(); + } + }); + } + + private async fetchOwnBlobBeforePublish( + store: S, + ): Promise> { + let events: RelayEvent[]; + try { + events = await relayClient.fetchEvents({ + kinds: [this.config.kind], + authors: [this.pubkey], + "#d": [this.config.dTag], + limit: 1, + }); + } catch { + // The fetch itself failed (timeout / auth / socket) — NOT proof that no + // head exists. Publishing the local store here could erase an unseen + // newer head during a transient outage, so retain and retry (Carl P1). + return { kind: "retain" }; + } + // A successful fetch that proves no head exists: publish the local store. + if (events.length === 0 || events[0].pubkey !== this.pubkey) + return { kind: "publish", store }; + const event = events[0]; + // Record the raw head before decrypt on the pre-publish path too. + this.recordRemoteHead(event.created_at); + const remote = await this.decryptAndParse(event); + // A head exists but could not be read (decrypt fault / malformed / future + // schema). A max-merge is only safe once both operands were actually read, + // so retain rather than publish the local store over an uninspectable head + // (Carl P1). The durable outbox and bounded retry resume normal resolution + // once a readable head returns. + if (!remote) return { kind: "retain" }; + this.observe(remote.store); + // Max-merge: the local edit's per-entry winners survive by construction and + // any newer remote entries fold in, so no adopt step is needed. + return { + kind: "publish", + store: this.config.mergeWithRemote(store, remote.store), + }; + } + + /** + * After a publish OK, fetch the authoritative retained head and report whether + * it subsumes the store we attempted to write. The relay returns OK for a + * superseded NIP-33 write as a no-op, so two windows racing distinct blobs + * from the same head can both get OK while only one blob is retained (Carl + * P1). Only a retained head that subsumes our store proves our click is + * durable on the relay; otherwise the outbox must be kept and retried. + */ + private async confirmRetainedHeadSubsumes(store: S): Promise { + try { + const events = await relayClient.fetchEvents({ + kinds: [this.config.kind], + authors: [this.pubkey], + "#d": [this.config.dTag], + limit: 1, + }); + if (events.length === 0 || events[0].pubkey !== this.pubkey) return false; + const event = events[0]; + this.recordRemoteHead(event.created_at); + const remote = await this.decryptAndParse(event); + if (!remote) return false; + this.observe(remote.store); + return this.config.isSubsumedBy(store, remote.store); + } catch { + return false; + } + } + + /** + * Clear the in-memory pending edit and this window's own durable outbox key — + * but only if the completing publish still owns the current generation. + */ + private discardPending(gen: number): void { + if (gen !== this.pendingGeneration) return; + this.pendingStore = null; + this.config.clearOutbox(this.pubkey, this.relayUrl); + } + + /** Schedule a bounded-backoff retry of the retained pending edit. */ + private scheduleRetry(gen: number): void { + if (this.destroyed || this.pendingStore === null) return; + // A newer edit has superseded this one; its own timer owns the retry. + if (gen !== this.pendingGeneration) return; + if (this.retryTimer !== null) return; + const delay = this.retryDelayMs; + this.retryDelayMs = Math.min(this.retryDelayMs * 2, RETRY_MAX_MS); + this.retryTimer = window.setTimeout(() => { + this.retryTimer = null; + this.startCycle(); + }, delay); + } + + private async doPublish(store: S, gen: number): Promise { + // A newer edit was queued after this publish was scheduled; it owns the + // pending state and will publish the latest store — abandon this stale run. + if (gen !== this.pendingGeneration) return; + try { + const decision = await this.fetchOwnBlobBeforePublish(store); + // Guard: manager may have been destroyed while fetchOwnBlobBeforePublish + // was awaited (community switch during in-flight fetch). + if (this.destroyed) return; + // A newer edit was queued while we awaited the pre-publish fetch. It owns + // convergence now; the serialized cycle re-drives for it once this run + // unwinds. + if (gen !== this.pendingGeneration) return; + // The pre-publish read failed or the head was unreadable — keep the + // durable pending edit and retry rather than publish over uninspectable + // state (Carl P1). + if (decision.kind === "retain") { + this.scheduleRetry(gen); + return; + } + const merged = decision.store; + if ( + this.lastPublishedStore !== null && + this.config.storesEqual(merged, this.lastPublishedStore) + ) { + this.discardPending(gen); + return; + } + const ciphertext = await nip44EncryptToSelf( + JSON.stringify(this.config.serializePayload(merged)), + ); + // Clamp inside the relay's future-drift window so a skewed remote head + // can never make us stamp an unbounded future timestamp that wedges every + // subsequent publish; we adopt such a head on the next fetch instead. + const createdAt = clampPublishCreatedAt(this.lastRemoteCreatedAt); + const event = await signRelayEvent({ + kind: this.config.kind, + content: ciphertext, + createdAt, + tags: [ + ["d", this.config.dTag], + ["t", this.config.dTag], // relay discoverability; not used in our filters + ], + }); + // Final guard immediately before the network call: a newer edit may have + // been queued during the encrypt/sign await, or the manager destroyed. + if (this.destroyed || gen !== this.pendingGeneration) return; + await relayClient.publishEvent( + event, + this.config.publishTimeoutMsg, + this.config.publishErrorMsg, + ); + this.recordRemoteHead(event.created_at); + this.observe(merged); + // A publish OK is NOT proof of retention: the relay OKs a superseded + // NIP-33 write as a no-op, so a peer window racing a distinct blob from + // the same head can win retention while ours is silently dropped (Carl + // P1). Fetch the authoritative retained head and clear the durable outbox + // only when it subsumes what we wrote. If it does not (or the fetch could + // not prove it), keep the outbox and retry so the click is never lost; the + // retry's pre-publish max-merge folds the retained blob in. + if (this.destroyed) return; + const confirmed = await this.confirmRetainedHeadSubsumes(merged); + if (this.destroyed) return; + if (!confirmed) { + this.scheduleRetry(gen); + return; + } + // Only claim this store as the published head if it is still the current + // edit; a newer edit queued mid-flight owns lastPublishedStore now. + if (gen === this.pendingGeneration) { + this.lastPublishedStore = merged; + this.retryDelayMs = RETRY_BASE_MS; + } + this.discardPending(gen); + } catch (error) { + if (this.destroyed) return; + // Transient publish failure (timeout / socket error). Keep the pending + // edit and retry with backoff rather than waiting for a reconnect that a + // healthy socket never fires. Max-merge makes a duplicate publish + // idempotent, so a lost-ACK write that the relay actually accepted is + // harmless to re-send. + console.warn(`[${this.config.logPrefix}] publish failed:`, error); + this.scheduleRetry(gen); + } + } + + async subscribeLive( + onUpdate: (remote: RemoteMergeBlob) => void, + ): Promise<() => Promise> { + return relayClient.subscribeLive( + { + kinds: [this.config.kind], + authors: [this.pubkey], + "#d": [this.config.dTag], + limit: 0, + }, + (event: RelayEvent) => { + if (event.pubkey !== this.pubkey) return; + // Record the raw head before decrypt so an undecryptable live event + // still advances the watermark and blocks future seed-publish. + this.recordRemoteHead(event.created_at); + void this.decryptAndParse(event).then((result) => { + if (result) { + this.observe(result.store); + onUpdate(result); + } + }); + }, + ); + } + + /** + * Fetches the remote blob on first mount, records the remote head, and + * delegates the seed/hold/apply-remote decision to `runBootstrap`. + */ + async bootstrap(localStore: S) { + // Seed the high-water from the caller's persisted local store so a click + // before the remote fetch resolves already reflects retained entries. + this.observe(localStore); + const fetchResult = await this.fetchRemoteBlob(); + return runBootstrap({ + fetchResult, + lastHead: this.lastRemoteCreatedAt, + localStore, + isLocalNonEmpty: this.config.isLocalNonEmpty, + publishFn: (s) => this.publish(s), + }); + } + + destroy(): void { + // Cancel any pending publish and mark this manager as destroyed so any + // in-flight doPublish() calls abort before reaching relayClient. + // Debounce-window changes are NOT lost: publish() persisted them to the + // durable outbox synchronously, and the next mount resumes them. Flushing + // here is still avoided — it could publish relay A's state to relay B via + // the shared relayClient singleton. + this.destroyed = true; + this.cancelPendingPublish(); + this.pendingStore = null; + } +} diff --git a/desktop/src/features/sidebar/lib/wholeBlobSyncManager.ts b/desktop/src/features/sidebar/lib/wholeBlobSyncManager.ts new file mode 100644 index 0000000000..22fc409c20 --- /dev/null +++ b/desktop/src/features/sidebar/lib/wholeBlobSyncManager.ts @@ -0,0 +1,797 @@ +import { relayClient } from "@/shared/api/relayClient"; +import { + nip44DecryptFromSelf, + nip44EncryptToSelf, + signRelayEvent, +} from "@/shared/api/tauri"; +import type { RelayEvent } from "@/shared/api/types"; +import { + advanceWatermark, + clampPublishCreatedAt, + readWatermark, + runBootstrap, + type FetchResult, +} from "./sidebarSyncWatermark"; + +const DEBOUNCE_MS = 2_000; + +// Bounded backoff for a retained pending edit whose publish failed transiently +// (timeout / socket error) on an otherwise-healthy socket, so it does not wait +// for a reconnect that may never fire. +const RETRY_BASE_MS = 2_000; +const RETRY_MAX_MS = 30_000; + +export type RemoteBlob = { + store: S; + createdAt: number; + eventId: string; +}; + +/** + * Outcome of the pre-publish head check. + * + * - `publish` — local edit is at or ahead of the head; publish it. + * - `adopt` — a newer remote head exists; the local edit lost whole-blob + * LWW and must be discarded in favour of the remote store so UI + * and relay converge (see the fix-2 design note). The manager + * hands the remote back to the hook and never publishes. + * - `retain` — a head event exists but could not be decrypted/parsed (future + * schema, transient keychain/decrypt fault, malformed payload). + * The client cannot inspect it to decide adopt-or-publish, so + * overwriting it would be blind data loss (Carl P1). Keep the + * durable pending edit and retry rather than clobber authoritative + * state the client could not read. + */ +type PublishDecision = + | { kind: "publish"; store: S } + | { kind: "adopt"; remote: RemoteBlob } + | { kind: "retain" }; + +/** + * The canonical remote head as it stood when an edit was queued. The + * pre-publish check compares the fetched head against this frozen baseline — + * never against the mutable in-memory watermark, which a live event observed + * during the debounce window may already have advanced to that same head + * (silently suppressing the adopt). + */ +type PublishBaseline = { createdAt: number; eventId: string }; + +/** + * Outcome of the post-publish retained-head confirmation (Carl P1). + * + * - `confirmed` — the retained head is exactly our event; the write won LWW. + * - `adopt` — a different, readable event is the head; our write was + * superseded (a peer's same-second blob with a lower id won, or + * a strictly-newer head landed). The caller adopts the winner. + * - `retain` — the fetch failed, returned no/foreign head, or the head was + * unreadable; retention cannot be proven, so keep the durable + * edit and retry. + */ +type RetentionConfirmation = + | { kind: "confirmed" } + | { kind: "adopt"; remote: RemoteBlob } + | { kind: "retain" }; + +/** + * True when `head` is the canonical winner over the baseline the edit was + * queued against — i.e. the head advanced since the edit began. Canonical order + * is `created_at DESC, id ASC`: a strictly-later head wins, and a same-second + * head wins only with a strictly-lower id. A same-second head is comparable + * only once the baseline id is known (empty id = no prior head seen → not + * superseded). + */ +function remoteAdvancedSince( + head: RemoteBlob, + baseline: PublishBaseline, +): boolean { + if (head.createdAt > baseline.createdAt) return true; + return ( + head.createdAt === baseline.createdAt && + baseline.eventId !== "" && + head.eventId < baseline.eventId + ); +} + +/** + * True when tuple `a` is the canonical winner over `b` (`created_at DESC, + * id ASC`). An empty id means "no head seen yet" and always loses. + */ +function canonicalGreater(a: PublishBaseline, b: PublishBaseline): boolean { + if (a.eventId === "") return false; + if (b.eventId === "") return true; + if (a.createdAt !== b.createdAt) return a.createdAt > b.createdAt; + return a.eventId < b.eventId; +} + +/** The canonical-greater of two head tuples (`created_at DESC, id ASC`). */ +function canonicalMax(a: PublishBaseline, b: PublishBaseline): PublishBaseline { + return canonicalGreater(a, b) ? a : b; +} + +/** + * Per-lane configuration injected into the shared whole-blob sync engine. + * + * `S` is the store type (ChannelSectionStore, ChannelSortStore, …). + */ +export type WholeBlobLaneConfig = { + /** NIP-78 event kind for this lane (e.g. KIND_CHANNEL_SECTIONS). */ + kind: number; + /** NIP-78 `d` tag value used to scope the lane's blob. */ + dTag: string; + /** Human-readable prefix for console.warn messages. */ + logPrefix: string; + /** Parse decrypted JSON into a typed store, or null for unreadable payloads. */ + parse: (json: unknown) => S | null; + /** Serialize a store into the JSON object to be encrypted and published. */ + serializePayload: (store: S) => unknown; + /** Persist this window's pending edit to the durable outbox. */ + writeOutbox: (pubkey: string, store: S, relayUrl: string) => boolean; + /** Clear this window's own outbox key. */ + clearOutbox: (pubkey: string, relayUrl: string) => void; + /** True when `store` is semantically identical to `last`. Used to skip no-op publishes. */ + storesEqual: (store: S, last: S) => boolean; + /** True when the local store is non-empty and eligible for a first-sync seed-publish. */ + isLocalNonEmpty: (store: S) => boolean; +}; + +/** + * Shared whole-blob LWW sync engine used by ChannelSectionSyncManager and + * ChannelSortSyncManager. Both lanes use the same generation-CAS, outbox, + * watermark, retained-head confirmation, prior-gen fold, and bounded-retry + * machinery — only lane config (kind, d-tag, store type, parse/serialize, + * outbox functions) differs. + * + * ### Invariants preserved across all passes + * + * **pass-2 (genuine-remote adoption):** A remote observed passively during the + * debounce window still gets adopted because the pending edit's + * `publishBaseline` is frozen at queue time and only advances via our own + * successful publishes — not via passive remote observations. + * + * **pass-3 (own-confirmed fold):** A prior publish of ours that was confirmed + * is folded into `publishBaseline` so a later edit's pre-publish check does not + * mistake our own accepted write for a competing remote and adopt it away. + * + * **pass-4 (conditional catch rollback):** The `claimLegacy` logic lives in the + * storage layer (lane-specific) and is not affected by this shared engine. + * + * **r7 prior-gen fold:** When a same-second lower-id peer wins a LWW collision + * and a newer edit (gen B) was queued after A's ACK but before confirmation + * resolved, `adoptRemote`'s stale-gen path repairs the poisoned baseline via + * `foldSupersedingAttemptWinner` so B publishes above the true retained head + * rather than adopting the winner away. The generation guard requires the + * baseline attempt to be prior-generation (`attempt < pendingGeneration`). + */ +export class WholeBlobSyncManager { + private pubkey: string; + private relayUrl: string; + private config: WholeBlobLaneConfig; + + private debounceTimer: number | null = null; + private retryTimer: number | null = null; + private retryDelayMs = RETRY_BASE_MS; + private lastRemoteCreatedAt: number; + // Canonical best head observed so far (`created_at DESC, id ASC`). Frozen into + // a per-edit baseline at publish time so the pre-publish check can tell + // whether the head advanced *since the edit was queued*, independent of the + // mutable watermark that a live event during the debounce window may advance. + private lastRemoteHead: PublishBaseline = { createdAt: 0, eventId: "" }; + // The canonical head this pending edit is racing against, frozen when the + // edit was queued and advanced ONLY by our own successful publishes. Freezing + // at queue time is what makes a genuine remote observed during the debounce + // window still adopt-worthy (pass-2): the mutable watermark advanced to that + // remote, but the baseline did not. Folding our own published head forward is + // what stops a newer edit from adopting an older generation's own accepted + // write (pass-3): our prior publish is our baseline, not a competing remote. + private publishBaseline: PublishBaseline = { createdAt: 0, eventId: "" }; + private pendingStore: S | null = null; + // Monotonic id for the current pending edit. Every publish() bumps it; every + // scheduled publish/retry captures the value it was queued for. A completion + // (success, adopt, or no-op) may only clear pending state via compare-and-swap + // on this generation, so an older in-flight publish can never erase a newer + // edit that arrived while it was in flight. + private pendingGeneration = 0; + // Publish cycles are serialized: at most one runs at a time. A newer edit + // queued while a cycle is in flight does NOT start its own concurrent cycle; + // it defers, and the in-flight cycle's completion schedules it. Serialization + // guarantees there is never more than one baseline/fetch/publish sequence + // touching shared manager state, so a stale generation can never sign or + // publish after a newer edit exists. + private publishInFlight = false; + // Event ids we signed and sent to the relay but whose ACK never arrived (the + // publish promise rejected as a timeout/socket error after the frame left). + // The relay MAY have accepted such a write, so if a later cycle's pre-publish + // fetch returns a head whose id is in this map, that head is OUR OWN accepted + // predecessor — fold it forward and publish above it, rather than adopting it + // and erasing the queued edit. An attempt the relay never accepted can never + // surface as the head, so an id match is proof of our own accepted write. + // Maps attempt id → pendingGeneration at the time of signing, so + // foldSupersedingAttemptWinner can require the baseline was poisoned by a + // strictly prior generation (gen < pendingGeneration). + private ambiguousAttemptIds = new Map(); + private lastPublishedStore: S | null = null; + protected destroyed = false; + // Set by the hook so an adopted remote head (local edit lost whole-blob LWW) + // is written through to React state + localStorage. + private onRemoteAdopted: ((remote: RemoteBlob) => void) | null = null; + + constructor( + pubkey: string, + relayUrl: string, + config: WholeBlobLaneConfig, + ) { + this.pubkey = pubkey; + this.relayUrl = relayUrl; + this.config = config; + // Hydrate from localStorage so we never seed-publish if a remote blob has + // been seen in a prior session. + this.lastRemoteCreatedAt = readWatermark(pubkey, config.dTag, relayUrl); + } + + /** Register the hook's adopt-remote sink (write-through to UI + storage). */ + setOnRemoteAdopted(cb: (remote: RemoteBlob) => void): void { + this.onRemoteAdopted = cb; + } + + async fetchRemoteBlob(): Promise>> { + try { + const events = await relayClient.fetchEvents({ + kinds: [this.config.kind], + authors: [this.pubkey], + "#d": [this.config.dTag], + limit: 1, + }); + if (events.length === 0 || events[0].pubkey !== this.pubkey) { + return { status: "absent" }; + } + const event = events[0]; + // An event exists — record its created_at regardless of whether we can + // decrypt it, so seed-publish is blocked even when the payload is + // unreadable (e.g. wrong key). + this.recordRemoteHead(event.created_at, event.id); + const result = await this.decryptAndParse(event); + if (!result) { + return { status: "failed", createdAt: event.created_at }; + } + return { + status: "found", + data: result, + createdAt: result.createdAt, + eventId: result.eventId, + }; + } catch { + return { status: "failed" }; + } + } + + private async decryptAndParse( + event: RelayEvent, + ): Promise | null> { + try { + const plaintext = await nip44DecryptFromSelf(event.content); + const store = this.config.parse(JSON.parse(plaintext)); + if (!store) return null; + return { store, createdAt: event.created_at, eventId: event.id }; + } catch { + return null; + } + } + + /** Update in-memory + persisted watermark and the canonical head tuple. */ + private recordRemoteHead(createdAt: number, eventId: string): void { + if (createdAt > this.lastRemoteCreatedAt) { + this.lastRemoteCreatedAt = createdAt; + } + // Track the canonical-best head (`created_at DESC, id ASC`): a later head + // always wins; a same-second head wins only with a strictly-lower id. This + // mirrors the relay's stored winner so a frozen baseline reflects reality. + if ( + createdAt > this.lastRemoteHead.createdAt || + (createdAt === this.lastRemoteHead.createdAt && + (this.lastRemoteHead.eventId === "" || + eventId < this.lastRemoteHead.eventId)) + ) { + this.lastRemoteHead = { createdAt, eventId }; + } + advanceWatermark(this.pubkey, this.config.dTag, this.relayUrl, createdAt); + } + + cancelPendingPublish(): void { + if (this.debounceTimer !== null) { + window.clearTimeout(this.debounceTimer); + this.debounceTimer = null; + } + if (this.retryTimer !== null) { + window.clearTimeout(this.retryTimer); + this.retryTimer = null; + } + } + + getPendingStore(): S | null { + return this.pendingStore; + } + + /** + * Re-drive the CURRENT pending edit's publish immediately without opening a + * new generation — used by the reconnect handler. Calling the public + * `publish()` there would bump the generation and reset `publishBaseline` to + * the just-fetched head, so a remote that won LWW while the edit was pending + * would be published over instead of adopted (Carl P1). Waking the existing + * generation keeps the frozen baseline, so the debounced cycle's pre-publish + * check still adopts a genuinely-advanced remote. No-op when nothing is + * pending. + */ + retryPendingPublish(): void { + if (this.pendingStore === null) return; + this.cancelPendingPublish(); + this.startCycle(); + } + + /** True while an unpublished local edit is queued (debouncing or retrying). */ + hasPendingEdit(): boolean { + return this.pendingStore !== null; + } + + /** + * When the current `publishBaseline` was frozen against one of our own + * attempted (but non-retained) event ids and `remote` is the same-second + * lower-id peer that won the collision, the baseline is poisoned: the next + * pre-publish check would see the winner as having advanced past our phantom + * attempt and adopt the edit away (Carl P1 / same-second variants). Folding + * the winner in repairs the baseline so the pre-publish check sees the true + * retained head and publishes above it. + * + * Scoped to same-second lower-id only: a strictly-later remote is a genuine + * advance that the pending edit must still adopt (pass-2 invariant). + * + * Requires the baseline attempt to be prior-generation (gen < + * pendingGeneration): if no newer edit was queued, the current generation's + * own in-flight attempt cannot trigger the fold — the loser must adopt (LWW + * correctness). The post-ACK ordering is load-bearing for causal soundness: + * because A's ACK returned before B queued, the peer winner pre-existed B, so + * preserving B is causally justified. If B queues while A's ACK is pending, + * the peer may have been accepted after B and is a genuine remote advance B + * must adopt — B's baseline will not equal A's attempt id in that case, so + * the fold does not fire (correct behaviour, by the causal argument above). + * + * Returns true when the fold was applied (caller may use for branching). + */ + private foldSupersedingAttemptWinner(remote: { + createdAt: number; + eventId: string; + }): boolean { + const attempt = this.ambiguousAttemptIds.get(this.publishBaseline.eventId); + if ( + this.publishBaseline.eventId !== "" && + attempt !== undefined && + attempt < this.pendingGeneration && + remote.createdAt === this.publishBaseline.createdAt && + remote.eventId < this.publishBaseline.eventId + ) { + this.publishBaseline = canonicalMax(this.publishBaseline, { + createdAt: remote.createdAt, + eventId: remote.eventId, + }); + return true; + } + return false; + } + + /** + * Adopt a remote store that superseded a local edit: hand it to the hook for + * write-through, advance the watermark, and drop the losing pending edit — + * including the durable outbox, so the outbox can never replay an edit that + * adopt just decided lost (which would reintroduce divergence). + * + * Compare-and-swap on `gen`: this adopt was decided against the edit queued + * at generation `gen`. If a newer edit arrived while this publish was in + * flight, the generation has moved on and that newer edit is the latest writer + * — it will publish and win LWW — so a stale adopt must not clear its pending + * state or overwrite its optimistic UI. We still advance the watermark + * (monotonic and always safe) so the newer edit stamps above this head. + */ + private adoptRemote(remote: RemoteBlob, gen: number): void { + this.recordRemoteHead(remote.createdAt, remote.eventId); + if (gen !== this.pendingGeneration) { + // A newer edit is pending. Its baseline may have been frozen against our + // own attempted (non-retained) event id if it was queued after the ACK + // but before confirmation resolved. Repair any poisoned baseline so the + // newer edit's pre-publish check publishes above the true retained head + // rather than adopting the winner away (Carl P1). + this.foldSupersedingAttemptWinner(remote); + return; + } + this.pendingStore = null; + this.config.clearOutbox(this.pubkey, this.relayUrl); + this.lastPublishedStore = remote.store; + if (this.destroyed) return; + this.onRemoteAdopted?.(remote); + } + + /** + * Clear the in-memory pending edit and this window's own durable outbox key — + * but only if the completing publish still owns the current generation. A + * publish for an older edit that finishes after a newer edit was queued must + * leave the newer edit (and its retry state) untouched. + */ + private discardPending(gen: number): void { + if (gen !== this.pendingGeneration) return; + this.pendingStore = null; + this.config.clearOutbox(this.pubkey, this.relayUrl); + } + + publish(store: S): boolean { + this.pendingStore = store; + ++this.pendingGeneration; + // Freeze the canonical head this edit is racing against at queue time. The + // pre-publish check compares the fetched head against this baseline, not the + // mutable watermark — a live event applied during the debounce window + // advances the watermark to a new remote head, which would otherwise make + // the pre-publish comparison see equality and fall through to a publish that + // overwrites a remote that became head *after* this edit was queued. The + // baseline only advances via our own successful publishes (see doPublish), + // so a prior generation's own accepted write is folded in rather than + // mistaken for a competing remote. + this.publishBaseline = { ...this.lastRemoteHead }; + // Persist synchronously so an edit made <2s before quit/community-switch + // survives teardown and resumes on next mount (durable outbox). This + // window's own key is the only one written — a single unconditional setItem, + // never a shared-key read-modify-write; `queuedAt` stamps it so resume + // replays only the newest queued blob (whole-blob LWW). The returned flag + // reports whether the intent is now durably held, so a legacy replay can + // gate its consumption marker on a proven transfer. + const durable = this.config.writeOutbox(this.pubkey, store, this.relayUrl); + if (this.debounceTimer !== null) { + window.clearTimeout(this.debounceTimer); + } + // A fresh edit supersedes any retry scheduled for the previous generation. + if (this.retryTimer !== null) { + window.clearTimeout(this.retryTimer); + this.retryTimer = null; + } + this.retryDelayMs = RETRY_BASE_MS; + this.debounceTimer = window.setTimeout(() => { + this.debounceTimer = null; + this.startCycle(); + }, DEBOUNCE_MS); + return durable; + } + + /** + * Serialize publish cycles: at most one runs at a time. A debounce/retry + * timer that fires while a cycle is in flight defers — the in-flight cycle's + * completion re-drives if a pending edit still needs publishing. This kills + * the cross-generation race class by construction: a newer edit queued during + * a cycle cannot start its own concurrent cycle, so there is never more than + * one baseline/fetch/publish sequence competing over shared manager state. + */ + private startCycle(): void { + if (this.destroyed || this.pendingStore === null) return; + if (this.publishInFlight) return; + const store = this.pendingStore; + const gen = this.pendingGeneration; + this.publishInFlight = true; + void this.doPublish(store, gen).finally(() => { + this.publishInFlight = false; + // A newer edit queued during the cycle (or a cycle that ended without + // clearing its pending edit) still needs publishing and has no timer + // pending to drive it — drive the next cycle now that the lane is free. + if ( + !this.destroyed && + this.pendingStore !== null && + this.debounceTimer === null && + this.retryTimer === null + ) { + this.startCycle(); + } + }); + } + + private async fetchOwnBlobBeforePublish( + store: S, + ): Promise> { + try { + const events = await relayClient.fetchEvents({ + kinds: [this.config.kind], + authors: [this.pubkey], + "#d": [this.config.dTag], + limit: 1, + }); + if (events.length === 0 || events[0].pubkey !== this.pubkey) + return { kind: "publish", store }; + const event = events[0]; + const remote = await this.decryptAndParse(event); + // Record the head after decrypt attempt so the watermark/head-tuple + // advance even for an undecryptable payload. + this.recordRemoteHead(event.created_at, event.id); + // A head event exists but we could not read it (future schema, transient + // decrypt fault, malformed payload). We cannot decide adopt-or-publish + // against state we cannot inspect, so retain the pending edit and retry + // rather than blindly overwrite authoritative state (Carl P1). + if (!remote) return { kind: "retain" }; + // Whole-blob LWW. Compare the fetched head against the baseline frozen + // when this edit was queued — NOT the live watermark, which a passive live + // event during debounce may already have advanced to this same head. If + // the canonical head advanced since the edit began, the local edit lost + // and is adopted-away rather than republished over it. + if (remoteAdvancedSince(remote, this.publishBaseline)) { + // Unless the advancing head is a prior publish of OURS whose ACK was + // lost: the relay accepted it, but our promise rejected before we could + // fold it forward, so it is our own accepted predecessor — not a + // competing remote. Fold it into the baseline and publish above it so + // the queued edit survives instead of adopting our own stale write away. + if (this.ambiguousAttemptIds.has(remote.eventId)) { + this.publishBaseline = canonicalMax(this.publishBaseline, { + createdAt: remote.createdAt, + eventId: remote.eventId, + }); + return { kind: "publish", store }; + } + // Same-second lower-id peer winner that superseded our own attempted + // head: baseline.eventId is our attempted id (still in + // ambiguousAttemptIds because the adopt path never clears it) and the + // relay retained the peer's lower-id blob instead. This arises when + // confirmation returned retain/failed and the retry pre-publish fetch + // sees the winner directly with the original poisoned baseline still in + // place. Fold the winner in so this edit publishes above the true + // retained head rather than adopting it away (Carl P1). + if (this.foldSupersedingAttemptWinner(remote)) { + return { kind: "publish", store }; + } + return { kind: "adopt", remote }; + } + return { kind: "publish", store }; + } catch { + // The pre-publish fetch itself failed (timeout / auth / socket) — this is + // NOT proof that no head exists. Publishing here would sign above a stale + // watermark and could erase an unseen newer head during a transient + // outage. Retain the durable pending edit and retry rather than overwrite + // state we could not read (Carl P1). + return { kind: "retain" }; + } + } + + /** + * After a publish OK, fetch the authoritative retained head and decide whether + * our event actually became it. The relay OKs a superseded NIP-33 write as a + * no-op (`Duplicate`), so a resolved publish is NOT proof of retention: two + * windows stamping the same `created_at` both get OK while only the lower + * event id is retained (Carl P1). Only an exact event-id match proves our + * whole-blob write won LWW. + * + * - head id === ours → `confirmed` + * - head is our own accepted prior → `retain` (a lost-ACK / same-second + * attempt (ambiguousAttemptIds) lower-id attempt of ours won; the retry + * folds it forward and republishes above) + * - head is a different readable → `adopt` (our blob lost; adopt the + * event winner so UI and relay converge) + * - fetch failed / no head / head → `retain` (retention unprovable or the + * unreadable winner unreadable — keep the durable + * edit and retry) + */ + private async confirmRetainedHead( + ourEventId: string, + ): Promise> { + try { + const events = await relayClient.fetchEvents({ + kinds: [this.config.kind], + authors: [this.pubkey], + "#d": [this.config.dTag], + limit: 1, + }); + if (events.length === 0 || events[0].pubkey !== this.pubkey) + return { kind: "retain" }; + const event = events[0]; + this.recordRemoteHead(event.created_at, event.id); + if (event.id === ourEventId) return { kind: "confirmed" }; + // A different event is the head. If it is one of OUR OWN accepted prior + // attempts (a lost-ACK write the relay retained, or a same-second lower-id + // attempt that won over this one), it is not a competing remote — retain + // so the retry's pre-publish folds it forward and publishes above it, + // rather than adopting our own older blob and dropping the current edit. + if (this.ambiguousAttemptIds.has(event.id)) return { kind: "retain" }; + const remote = await this.decryptAndParse(event); + if (!remote) return { kind: "retain" }; + return { kind: "adopt", remote }; + } catch { + return { kind: "retain" }; + } + } + + /** Schedule a bounded-backoff retry of the retained pending edit. */ + private scheduleRetry(gen: number): void { + if (this.destroyed || this.pendingStore === null) return; + // A newer edit has superseded this one; its own timer owns the retry. + if (gen !== this.pendingGeneration) return; + if (this.retryTimer !== null) return; + const delay = this.retryDelayMs; + this.retryDelayMs = Math.min(this.retryDelayMs * 2, RETRY_MAX_MS); + this.retryTimer = window.setTimeout(() => { + this.retryTimer = null; + this.startCycle(); + }, delay); + } + + private async doPublish(store: S, gen: number): Promise { + // A newer edit was queued after this publish was scheduled; it owns the + // pending state and will publish the latest store — abandon this stale run. + if (gen !== this.pendingGeneration) return; + try { + const decision = await this.fetchOwnBlobBeforePublish(store); + // Guard: manager may have been destroyed while fetchOwnBlobBeforePublish + // was awaited (community switch during in-flight fetch). If so, abort + // before touching the relay. + if (this.destroyed) return; + // A newer edit was queued while we awaited the pre-publish fetch. It owns + // convergence now; abort so we neither publish this stale store nor adopt + // over the newer pending edit. The serialized cycle re-drives for the + // newer generation once this one unwinds. + if (gen !== this.pendingGeneration) return; + if (decision.kind === "adopt") { + this.adoptRemote(decision.remote, gen); + return; + } + if (decision.kind === "retain") { + // Head exists but is unreadable — keep the durable pending edit and + // retry with backoff rather than overwrite state we could not inspect. + this.scheduleRetry(gen); + return; + } + const merged = decision.store; + if ( + this.lastPublishedStore !== null && + this.config.storesEqual(merged, this.lastPublishedStore) + ) { + this.discardPending(gen); + return; + } + const ciphertext = await nip44EncryptToSelf( + JSON.stringify(this.config.serializePayload(merged)), + ); + // Clamp inside the relay's future-drift window: never manufacture a + // timestamp so far ahead that this or a later publish is rejected for + // drift and wedges. If a skewed remote head sits beyond the window we + // will lose LWW and adopt it on the next pre-publish fetch rather than + // walking past it. + const createdAt = clampPublishCreatedAt(this.lastRemoteCreatedAt); + const event = await signRelayEvent({ + kind: this.config.kind, + content: ciphertext, + createdAt, + tags: [ + ["d", this.config.dTag], + ["t", this.config.dTag], // relay discoverability; not used in our filters + ], + }); + // Final guard immediately before the network call — sign/encrypt are + // synchronous-ish but cheap; the relay socket may have moved to a + // different community by the time we reach this point, or a newer edit + // may have been queued during the encrypt/sign await (invariant: a stale + // generation never signs/publishes after a newer edit exists). + if (this.destroyed || gen !== this.pendingGeneration) return; + // Record this signed id as an in-flight attempt of unknown fate before we + // send it. If the ACK is lost below, a later cycle that fetches this id as + // the head recognises it as our own accepted write and folds it forward + // rather than adopting it away. + this.ambiguousAttemptIds.set(event.id, gen); + await relayClient.publishEvent( + event, + `Timed out publishing ${this.config.dTag}.`, + `Failed to publish ${this.config.dTag}.`, + ); + this.recordRemoteHead(event.created_at, event.id); + // A publish OK is NOT proof of retention: the relay OKs a superseded + // NIP-33 write as a no-op (`Duplicate`), so two windows stamping the same + // `created_at` both get OK while only the lower event id is retained (Carl + // P1). Fetch the authoritative head; only an exact id match proves our + // whole-blob write won. Anything else must keep the durable edit — never + // fold a nonexistent head into the baseline, which would poison the next + // edit into adopting the true winner away. + if (this.destroyed) return; + const confirmation = await this.confirmRetainedHead(event.id); + if (this.destroyed) return; + if (confirmation.kind === "adopt") { + // A different readable event is the head — our blob lost the same-second + // collision. Adopt the winner so UI and relay converge (CAS on gen + // inside adoptRemote leaves a newer edit untouched). + this.adoptRemote(confirmation.remote, gen); + return; + } + if (confirmation.kind === "retain") { + // Retention unprovable (fetch failed / no head), or the head is our own + // accepted prior attempt whose ACK was lost. Keep the durable edit and + // retry: the retry's pre-publish folds our own accepted predecessor + // forward (ambiguousAttemptIds) and republishes above it. Leave + // ambiguousAttemptIds intact so that fold still recognises our writes. + this.scheduleRetry(gen); + return; + } + // Confirmed: our event is the retained head. It dominates every prior + // attempt (`created_at DESC, id ASC`), so no earlier ambiguous id can ever + // be the canonical head again — clear the set to keep it bounded. + this.ambiguousAttemptIds.clear(); + // Fold our own accepted head into the pending edit's baseline. This is + // unconditional across generations: even a stale generation's own + // confirmed write must advance the current pending baseline so the newer + // edit's pre-publish check does not mistake OUR prior publish for a + // competing remote and adopt it away (pass-3). Genuine remotes never fold + // in here — they only advance the watermark — so a remote that became head + // during the debounce window still adopts (pass-2). canonicalMax keeps the + // advance monotonic (`created_at DESC, id ASC`). + this.publishBaseline = canonicalMax(this.publishBaseline, { + createdAt: event.created_at, + eventId: event.id, + }); + // Only claim this store as the published head if it is still the current + // edit; a newer edit queued mid-flight owns lastPublishedStore now. + if (gen === this.pendingGeneration) { + this.lastPublishedStore = merged; + this.retryDelayMs = RETRY_BASE_MS; + } + this.discardPending(gen); + } catch (error) { + if (this.destroyed) return; + // Ambiguous outcome: the publish promise rejected (timeout / socket + // error), but the relay may already have accepted the write before the + // ACK was lost. Keep the pending edit and retry with backoff rather than + // waiting for a reconnect that a healthy socket never fires. The attempt + // id stays in ambiguousAttemptIds: if the relay did accept it, a later + // cycle that fetches this id as the head folds it forward as our own + // accepted predecessor (see fetchOwnBlobBeforePublish) instead of + // adopting it away and erasing the queued edit. + console.warn(`[${this.config.logPrefix}] publish failed:`, error); + this.scheduleRetry(gen); + } + } + + async subscribeLive( + onUpdate: (remote: RemoteBlob) => void, + ): Promise<() => Promise> { + return relayClient.subscribeLive( + { + kinds: [this.config.kind], + authors: [this.pubkey], + "#d": [this.config.dTag], + limit: 0, + }, + (event: RelayEvent) => { + if (event.pubkey !== this.pubkey) return; + // Record the raw head before decrypt so an undecryptable live event + // still advances the watermark and blocks future seed-publish. + this.recordRemoteHead(event.created_at, event.id); + void this.decryptAndParse(event).then((result) => { + if (result) { + onUpdate(result); + } + }); + }, + ); + } + + /** + * Fetches the remote blob on first mount, records the remote head, and + * delegates the seed/hold/apply-remote decision to `runBootstrap`. + */ + async bootstrap(localStore: S) { + const fetchResult = await this.fetchRemoteBlob(); + return runBootstrap({ + fetchResult, + lastHead: this.lastRemoteCreatedAt, + localStore, + isLocalNonEmpty: this.config.isLocalNonEmpty, + publishFn: (s) => this.publish(s), + }); + } + + destroy(): void { + // Cancel any pending publish and mark this manager as destroyed so any + // in-flight doPublish() calls abort before reaching relayClient. + // Debounce-window changes are NOT lost: publish() persisted them to the + // durable outbox synchronously, and the next mount resumes them. + // Flushing here is still avoided — it could publish relay A's data to + // relay B via the shared relayClient singleton. + this.destroyed = true; + this.cancelPendingPublish(); + this.pendingStore = null; + } +}