refactor(sections-sync): extract shared WholeBlobSyncManager and MergeLaneSyncManager

The four sidebar sync engines (sections, sort, stars, mutes) shared
nearly all implementation. The two whole-blob lanes (sections + sort)
carried the same generation-CAS, outbox, watermark, retained-head
confirmation, prior-gen fold, and bounded-retry machinery; the two
merge lanes (stars + mutes) differed only in store type and the
merge/subsumption functions.

Extract:
- wholeBlobSyncManager.ts: parameterized by WholeBlobLaneConfig<S>
  (kind, dTag, parse, serializePayload, writeOutbox, clearOutbox,
  storesEqual, isLocalNonEmpty). Preserves all pass-2/pass-3/r7
  invariants. ChannelSectionSyncManager and ChannelSortSyncManager
  are now thin wrappers (~75 lines each) that pass lane config and
  expose their named API methods.

- mergeLaneSyncManager.ts: parameterized by MergeLaneConfig<S>
  (kind, dTag, parse, serializePayload, mergeWithRemote, isSubsumedBy,
  storesEqual, observe, writeOutbox, clearOutbox, isLocalNonEmpty).
  ChannelStarSyncManager and ChannelMuteSyncManager are thin wrappers
  (~90 lines each).

All 315 sidebar lib tests pass at this commit with no behavior change.
Each lane wrapper still exposes its named public API (publishSections,
publishSortPrefs, publishStars, publishMutes, fetchRemoteSections,
subscribeToSections, etc.) so no hook changes are required.

Co-authored-by: Will Pfleger <wpfleger96@gmail.com>
Signed-off-by: Will Pfleger <wpfleger96@gmail.com>
This commit is contained in:
Duncan
2026-08-29 13:47:20 -04:00
parent 7c195d8fbd
commit 553aab4a02
6 changed files with 1509 additions and 2381 deletions

View File

@@ -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<ChannelMuteStore>;
// 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<RemoteMutes | null> {
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<string, { rev: number; updatedAt: number }>();
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<FetchResult<RemoteMutes>> {
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<MergePublishDecision> {
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<boolean> {
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<void> {
// 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<void>> {
return relayClient.subscribeLive(
{
kinds: [KIND_CHANNEL_MUTES],
authors: [this.pubkey],
"#d": [D_TAG],
limit: 0,
export class ChannelMuteSyncManager extends MergeLaneSyncManager<ChannelMuteStore> {
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);
}
}

View File

@@ -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<ChannelSectionStore>;
// 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<RemoteSections | null> {
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<string, number>();
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<ChannelSectionStore> {
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<FetchResult<RemoteSections>> {
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<PublishDecision> {
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<RetentionConfirmation> {
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<void> {
// 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<void>> {
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);
}
}

View File

@@ -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<ChannelSortStore>;
// 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<RemoteSortPrefs | null> {
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<string, number>();
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<ChannelSortStore> {
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<FetchResult<RemoteSortPrefs>> {
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<PublishDecision> {
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<RetentionConfirmation> {
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<void> {
// 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<void>> {
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);
}
}

View File

@@ -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<ChannelStarStore>;
// 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<RemoteStars | null> {
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<string, { rev: number; updatedAt: number }>();
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<FetchResult<RemoteStars>> {
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<MergePublishDecision> {
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<boolean> {
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<void> {
// 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<void>> {
return relayClient.subscribeLive(
{
kinds: [KIND_CHANNEL_STARS],
authors: [this.pubkey],
"#d": [D_TAG],
limit: 0,
export class ChannelStarSyncManager extends MergeLaneSyncManager<ChannelStarStore> {
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);
}
}

View File

@@ -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<S> = {
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<S> =
| { 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<S> = {
/** 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<string, { rev: number; updatedAt: number }>,
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<S> {
private pubkey: string;
private relayUrl: string;
protected config: MergeLaneConfig<S>;
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<string, { rev: number; updatedAt: number }>();
constructor(pubkey: string, relayUrl: string, config: MergeLaneConfig<S>) {
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<RemoteMergeBlob<S> | 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<FetchResult<RemoteMergeBlob<S>>> {
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<MergePublishDecision<S>> {
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<boolean> {
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<void> {
// 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<S>) => void,
): Promise<() => Promise<void>> {
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;
}
}

View File

@@ -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<S> = {
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<S> =
| { kind: "publish"; store: S }
| { kind: "adopt"; remote: RemoteBlob<S> }
| { 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<S> =
| { kind: "confirmed" }
| { kind: "adopt"; remote: RemoteBlob<S> }
| { 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<unknown>,
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<S> = {
/** 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<S> {
private pubkey: string;
private relayUrl: string;
private config: WholeBlobLaneConfig<S>;
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<string, number>();
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<S>) => void) | null = null;
constructor(
pubkey: string,
relayUrl: string,
config: WholeBlobLaneConfig<S>,
) {
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<S>) => void): void {
this.onRemoteAdopted = cb;
}
async fetchRemoteBlob(): Promise<FetchResult<RemoteBlob<S>>> {
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<RemoteBlob<S> | 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<S>, 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<PublishDecision<S>> {
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<RetentionConfirmation<S>> {
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<void> {
// 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<S>) => void,
): Promise<() => Promise<void>> {
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;
}
}