From c2ba27dbb442e6741ef7f98cd97907fe353afe6b Mon Sep 17 00:00:00 2001 From: Kyle Zemek Date: Sun, 8 Nov 2026 12:31:27 -0800 Subject: [PATCH] fix(unread): preserve selected message intent Co-authored-by: Codex Signed-off-by: Kyle Zemek --- docs/buzz-help.md | 8 +- docs/unread.md | 9 + src/features/relay/unread.test.ts | 615 ++++++++++++++++++++++++++++-- src/features/relay/unread.ts | 192 +++++++--- 4 files changed, 750 insertions(+), 74 deletions(-) diff --git a/docs/buzz-help.md b/docs/buzz-help.md index 0612f509df..9179534c14 100644 --- a/docs/buzz-help.md +++ b/docs/buzz-help.md @@ -394,8 +394,12 @@ revokes all of them, and a pairing unused for 30 days expires. read: unread threads keep their dots in the conversation and in Activity → Threads, and **New messages** still marks where you left off. A mention, DM or reply to you that arrives after you leave lights the row again, as does any - newer message; **Mark as unread** still shows on it. On another device, rows - with unread mentions, DMs or replies to you light again until you read them. + newer message; **Mark as unread** still shows on it. The channel message + menu's **Mark unread** keeps the selected message and currently loaded reply + subtree unread when you leave and reopen that channel during this window + session. Reading the selected subtree or channel in that window clears the exact mark. It is + not restored after Buzz restarts. On another device, rows with unread + mentions, DMs or replies to you light again until you read them. - **Communities** have their own channels, people, and agents. Switch with the community switcher at the bottom of the dock; in **Personal space**, Messages offers **Choose a community**, which opens it. Switching diff --git a/docs/unread.md b/docs/unread.md index 75687d1a7d..70d4555809 100644 --- a/docs/unread.md +++ b/docs/unread.md @@ -103,6 +103,15 @@ viewport): marked unread keep their intent. These marks stay on this device: frontier-only sync never publishes `msg:` marks, though it still reads them from older clients. +- The channel message menu's **Mark unread** keeps the selected message and its + currently loaded reply subtree as exact unread targets for the current window + session. Leaving and reopening the channel does not clear those IDs: shared + message selectors continue to report them after the sidebar hint is + reconciled. Explicit reads clear only the selected targets their scope and + frontier cover, even after observation eviction; a DM thread read preserves + selections in other threads. Deletion, access retirement, and session disposal + end the intent. The exact target set is not restored as a restart-persistent + mark. - `thread-activity:` acknowledges replies through the newest reply only in that thread. A collapsed newest reply, or one outside the bottom viewport in a branch-ordered thread, cannot earn catch-up; visible rows still read individually. diff --git a/src/features/relay/unread.test.ts b/src/features/relay/unread.test.ts index 9cf76bffc1..cd448910a9 100644 --- a/src/features/relay/unread.test.ts +++ b/src/features/relay/unread.test.ts @@ -179,6 +179,26 @@ function setup( }; } +/** Overflow only the observation cache, not the session's explicit unread intent. */ +function overflowObservations(h: ReturnType, channelId: string) { + // Admitted synthetic rows avoid thousands of signatures for this capacity boundary. + h.emit( + Array.from( + { length: 4096 }, + (_, index) => + ({ + id: `f${index.toString(16).padStart(63, "0")}`, + pubkey: h.alice.pubkey, + sig: "0".repeat(128), + kind: 9, + created_at: 100 + index, + tags: [["h", channelId]], + content: `filler ${index}`, + }) as RelayEvent, + ), + ); +} + it("a read reply that needs no conversation lookup is not reported pending", async () => { const h = setup(); h.grant("room"); @@ -1021,16 +1041,25 @@ it.each([false, true])( }, ); -it("a forged-author deletion does not hide a message", () => { - const h = setup(); - h.grant("room"); - const row = message(h.alice, "room", "retained", 11); - h.emit([ - row, - signed(h.viewer, { kind: 5, content: "", tags: [["e", row.id]] }), - ]); - expect(h.snapshot().observedCount).toBe(1); -}); +it.each([5, 9005])( + "a viewer's kind-%s deletion cannot clear another author's selected unread message", + async (kind) => { + const h = setup(); + h.grant("room"); + const row = message(h.alice, "room", "retained", 11); + h.emit([row]); + const unread = h.session.unread; + await unread.enterChannel("room"); + await unread.markMessageRead("room", row.id); + await unread.markMessageUnread("room", row.id); + h.emit([signed(h.viewer, { kind, content: "", tags: [["e", row.id]] })]); + expect(h.snapshot().observedCount).toBe(1); + expect(unread.attention("room", row.id)).toMatchObject({ + unread: true, + forced: true, + }); + }, +); it("all projections are denied before any revocation subscriber runs; durable intent survives", async () => { const h = setup(); @@ -1578,30 +1607,570 @@ it("forces loaded reply subtrees independently and preserves channel manual inte expect(h.snapshot().manual).toBe("local-only"); // Independent channel action survives. }); -it("leaves only the sidebar hint after channel leave and reconciles it on reopen", async () => { +it("keeps selected messages unread across leave and re-entry for exact selectors", async () => { const h = setup(); h.grant("room"); - const row = message(h.alice, "room", "read", 11); - h.emit([row]); - await h.session.unread.enterChannel("room"); - await h.session.unread.markThrough( - { kind: "message", channelId: "room", messageId: row.id }, - row.id, - ); - await h.session.unread.markMessageUnread("room", row.id); - expect(h.snapshot()).toMatchObject({ + const root = message(h.alice, "room", "read root", 11); + const child = message(h.alice, "room", "read reply", 12, [ + ["e", root.id, "", "reply"], + ]); + const other = message(h.alice, "room", "read other", 13); + h.emit([root, child, other]); + const unread = h.session.unread; + await unread.enterChannel("room"); + await unread.markMessageRead("room", root.id); + await unread.markMessageRead("room", other.id); + await unread.markMessageUnread("room", root.id); + await unread.markMessageUnread("room", other.id); + const later = message(h.alice, "room", "unrelated newer arrival", 20); + h.emit([later]); + + const rootTarget = { + kind: "message" as const, + channelId: "room", + messageId: root.id, + }; + expect(unread.snapshot(rootTarget).manual).toBe("local-only"); + expect(unread.attention("room", root.id)).toMatchObject({ + unread: true, + forced: true, + }); + expect(unread.attention("room", child.id)).toMatchObject({ + unread: true, + forced: true, + }); + expect(unread.attention("room", other.id)).toMatchObject({ + unread: true, + forced: true, + }); + + unread.leaveChannel("room"); + expect(unread.snapshot(rootTarget)).toMatchObject({ observedCount: 1, manual: "local-only", }); - h.session.unread.leaveChannel("room"); expect(h.snapshot()).toMatchObject({ - observedCount: 0, + observedCount: 4, + latestMessage: { id: later.id }, + }); + for (const id of [root.id, child.id, other.id]) + expect(unread.attention("room", id)).toMatchObject({ + unread: true, + forced: true, + }); + expect(unread.attention("room", later.id)).toMatchObject({ + unread: true, + forced: false, + }); + + await unread.enterChannel("room"); + expect(h.snapshot()).toMatchObject({ observedCount: 4 }); + expect(unread.snapshot(rootTarget)).toMatchObject({ + observedCount: 1, manual: "local-only", }); - await h.session.unread.enterChannel("room"); + expect(unread.attention("room", root.id)).toMatchObject({ + unread: true, + forced: true, + }); + + await unread.markMessageRead("room", root.id); + expect(unread.attention("room", root.id)).toMatchObject({ + unread: false, + forced: false, + }); + expect(unread.attention("room", child.id)).toMatchObject({ + unread: false, + forced: false, + }); + expect(unread.attention("room", later.id)).toMatchObject({ + unread: true, + forced: false, + }); + expect(unread.attention("room", other.id)).toMatchObject({ + unread: true, + forced: true, + }); + await unread.markMessageRead("room", other.id); + expect(unread.attention("room", other.id)).toMatchObject({ + unread: false, + forced: false, + }); + expect(unread.attention("room", later.id).forced).toBe(false); + await unread.markMessageRead("room", later.id); expect(h.snapshot()).toMatchObject({ observedCount: 0, manual: "none" }); }); +it("does not retain a selected-message override when its initial save fails", async () => { + const h = setup(); + h.grant("room"); + const row = message(h.alice, "room", "read message", 11); + h.emit([row]); + const unread = h.session.unread; + await unread.enterChannel("room"); + await unread.markMessageRead("room", row.id); + + h.failNextSave(); + await expect(unread.markMessageUnread("room", row.id)).rejects.toThrow( + "Storage update rejected", + ); + + expect(unread.attention("room", row.id)).toMatchObject({ + unread: false, + forced: false, + }); + expect(h.journal()?.localUnread["message-force:room"]).toBeUndefined(); +}); + +it("channel read while away clears retained selected-message intent", async () => { + const h = setup(); + h.grant("room"); + const root = message(h.alice, "room", "selected", 11); + const child = message(h.alice, "room", "reply", 12, [ + ["e", root.id, "", "reply"], + ]); + h.emit([root, child]); + const unread = h.session.unread; + await unread.enterChannel("room"); + await unread.markMessageRead("room", root.id); + await unread.markMessageUnread("room", root.id); + unread.leaveChannel("room"); + + for (const id of [root.id, child.id]) + expect(unread.attention("room", id)).toMatchObject({ + unread: true, + forced: true, + }); + + await unread.markChannelRead("room"); + + for (const id of [root.id, child.id]) + expect(unread.attention("room", id)).toMatchObject({ + unread: false, + forced: false, + }); + expect( + unread.snapshot({ + kind: "message", + channelId: "room", + messageId: root.id, + }), + ).toMatchObject({ manual: "none", observedCount: 0 }); + expect(h.journal()?.localUnread["message-force:room"]).toBeUndefined(); +}); + +it("unread cache clear retires the in-memory selected-message override", async () => { + const h = setup(); + h.grant("room"); + const root = message(h.alice, "room", "read root", 11); + const child = message(h.alice, "room", "read reply", 12, [ + ["e", root.id, "", "reply"], + ]); + h.emit([root, child]); + const unread = h.session.unread; + await unread.enterChannel("room"); + await unread.markMessageRead("room", root.id); + await unread.markMessageUnread("room", root.id); + expect(unread.attention("room", child.id)).toMatchObject({ + unread: true, + forced: true, + }); + + await h.clearCache(); + h.grant("room"); + h.emit([root, child]); + await vi.waitFor(() => expect(h.snapshot().observedCount).toBe(0)); + + for (const id of [root.id, child.id]) + expect(unread.attention("room", id)).toMatchObject({ + unread: false, + forced: false, + }); +}); + +it.each( + (["markMessageRead", "markMessageUnread"] as const).flatMap((action) => + ( + [ + "leave", + "overflow", + "delete-overflow", + "forged-overflow", + "viewer-forged-overflow", + ] as const + ).map((transition) => ({ action, transition })), + ), +)( + "$action keeps its committed result across $transition before storage resolves", + async ({ action, transition }) => { + const h = setup(); + h.grant("room"); + const root = message(h.alice, "room", "read root", 11); + const child = message(h.alice, "room", "read reply", 12, [ + ["e", root.id, "", "reply"], + ]); + h.emit([root, child]); + const unread = h.session.unread; + await unread.enterChannel("room"); + await unread.markMessageRead("room", root.id); + if (action === "markMessageRead") + await unread.markMessageUnread("room", root.id); + + const held = h.holdCommit(); + const marking = unread[action]("room", root.id); + await held.committed; + try { + expect(unread.attention("room", root.id).forced).toBe( + action === "markMessageRead", + ); + unread.leaveChannel("room"); + if ( + transition === "delete-overflow" || + transition === "forged-overflow" || + transition === "viewer-forged-overflow" + ) + h.emit([ + signed( + transition === "delete-overflow" + ? h.alice + : transition === "viewer-forged-overflow" + ? h.viewer + : h.relay, + { + kind: 5, + content: "", + tags: [ + ["e", root.id], + ["e", child.id], + ], + }, + ), + ]); + if (transition !== "leave") { + overflowObservations(h, "room"); + expect(h.snapshot().freshness).toBe("stale"); + } + } finally { + held.release(); + } + await marking; + h.emit([root, child]); + await unread.enterChannel("room"); + + const forced = + action === "markMessageUnread" && transition !== "delete-overflow"; + for (const id of [root.id, child.id]) + expect(unread.attention("room", id)).toMatchObject({ + unread: forced, + forced, + }); + expect( + unread.snapshot({ + kind: "message", + channelId: "room", + messageId: root.id, + }), + ).toMatchObject({ manual: forced ? "local-only" : "none" }); + }, +); + +it.each(["message", "thread", "channel", "timeline"] as const)( + "an explicit %s read retires only its covered retained targets after persistence", + async (kind) => { + const h = setup(); + h.grant("room"); + h.emit([metadata(h.relay, "room", "room", 11, [["t", "stream"]])]); + const root = message(h.alice, "room", "root", 11); + const child = message(h.alice, "room", "reply", 12, [ + ["e", root.id, "", "reply"], + ["p", h.viewer.pubkey], + ]); + const other = message(h.alice, "room", "independent", 14); + h.emit([root, child, other]); + const unread = h.session.unread; + await unread.enterChannel("room"); + await unread.markMessageRead("room", root.id); + await unread.markMessageRead("room", other.id); + await unread.markMessageUnread("room", root.id); + await unread.markMessageUnread("room", other.id); + unread.leaveChannel("room"); + await unread.enterChannel("room"); + const read = () => + kind === "timeline" + ? unread.markTimelineRead("room", [root.id]) + : unread.markThrough( + kind === "message" + ? { kind, channelId: "room", messageId: root.id } + : kind === "thread" + ? { kind, channelId: "room", rootId: root.id } + : { kind, channelId: "room" }, + kind === "thread" ? child.id : root.id, + ); + + h.failNextSave(); + await expect(read()).rejects.toThrow("Storage update rejected"); + for (const id of [root.id, child.id, other.id]) + expect(unread.attention("room", id).forced).toBe(true); + await read(); + expect(unread.attention("room", root.id)).toMatchObject({ + unread: false, + forced: false, + }); + expect(unread.attention("room", child.id)).toMatchObject({ + unread: kind !== "thread", + forced: kind !== "thread", + }); + expect(unread.attention("room", other.id)).toMatchObject({ + unread: true, + forced: true, + }); + }, +); + +it("does not restore selected-message intent after access retires during a committed save", async () => { + const h = setup(); + h.grant("room"); + const row = message(h.alice, "room", "read message", 11); + h.emit([row]); + const unread = h.session.unread; + await unread.enterChannel("room"); + await unread.markMessageRead("room", row.id); + + const held = h.holdCommit(); + const marking = unread.markMessageUnread("room", row.id); + await held.committed; + try { + h.emit([roster(h.relay, "room", [], 20)]); + } finally { + held.release(); + } + await marking; + + expect(unread.attention("room", row.id)).toMatchObject({ + status: "unknown", + unread: false, + forced: false, + }); +}); + +it.each(["revoke-regrant", "overflow"])( + "cancels a selected-message save before commit across %s", + async (transition) => { + const h = setup(); + h.grant("room"); + const row = message(h.alice, "room", "read message", 11); + h.emit([row]); + const unread = h.session.unread; + await unread.enterChannel("room"); + await unread.markMessageRead("room", row.id); + + const held = h.holdSaveStarted(); + const marking = unread.markMessageUnread("room", row.id); + await held.started; + try { + if (transition === "revoke-regrant") { + h.emit([roster(h.relay, "room", [], 20)]); + h.grant("room", 21); + } else { + overflowObservations(h, "room"); + expect(h.snapshot().freshness).toBe("stale"); + } + } finally { + held.release(); + } + await expect(marking).rejects.toThrow("Reading observation expired"); + + expect(h.journal()?.localUnread["message-force:room"]).toBeUndefined(); + h.emit([row]); + await unread.enterChannel("room"); + expect(unread.attention("room", row.id)).toMatchObject({ + unread: false, + forced: false, + }); + }, +); + +it.each([false, true])( + "Mark all read after an observation reset: %s", + async (readAll) => { + const h = setup(); + h.grant("room"); + const marked = message(h.alice, "room", "selected", 11); + h.emit([marked]); + const unread = h.session.unread; + await unread.enterChannel("room"); + await unread.markMessageRead("room", marked.id); + await unread.markMessageUnread("room", marked.id); + unread.leaveChannel("room"); + await unread.enterChannel("room"); + + overflowObservations(h, "room"); + expect(h.snapshot().freshness).toBe("stale"); + + if (readAll) expect(await unread.markAllChannelsRead()).toHaveLength(1); + h.emit([marked]); + expect(unread.attention("room", marked.id)).toMatchObject({ + unread: !readAll, + forced: !readAll, + }); + }, +); + +it.each([ + { scope: "thread", orphan: false, nested: false, evict: true, late: false }, + { scope: "thread", orphan: true, nested: false, evict: true, late: false }, + { scope: "thread", orphan: true, nested: true, evict: false, late: false }, + { scope: "thread", orphan: true, nested: true, evict: true, late: false }, + { scope: "thread", orphan: true, nested: true, evict: true, late: true }, + { scope: "dm", orphan: true, nested: true, evict: false, late: false }, + { scope: "dm", orphan: true, nested: true, evict: true, late: false }, + { scope: "dm", orphan: true, nested: true, evict: true, late: true }, +])( + "a covering $scope read retires a selection (orphan: $orphan, nested: $nested, evicted: $evict, late ancestry: $late)", + async ({ scope, orphan, nested, evict, late }) => { + const h = setup(); + h.grant("room"); + if (scope === "dm") + h.emit([metadata(h.relay, "room", "DM", 11, [["t", "dm"]])]); + const root = message(h.alice, "room", "root", 11); + const parent = message(h.alice, "room", "parent reply", 12, [ + ["e", root.id, "", "reply"], + ]); + const selected = message(h.alice, "room", "selected reply", 13, [ + ["e", nested ? parent.id : root.id, "", "reply"], + ["p", h.viewer.pubkey], + ]); + const newer = message(h.alice, "room", "newer reply", 14, [ + ["e", root.id, "", "reply"], + ["p", h.viewer.pubkey], + ]); + h.emit(orphan ? [selected] : [root, selected]); + const unread = h.session.unread; + await unread.enterChannel("room"); + await unread.markMessageRead("room", selected.id); + await unread.markMessageUnread("room", selected.id); + unread.leaveChannel("room"); + if (nested && !late) h.emit([root, parent]); + if (evict) { + overflowObservations(h, "room"); + expect(h.snapshot().freshness).toBe("stale"); + expect(unread.attention("room", selected.id).status).toBe("unknown"); + } + if (late) h.emit([root, parent]); + h.emit([root, newer]); + + if (scope === "dm") await unread.prepareChannelRead("room", root.id)(); + else + await unread.markThrough( + { kind: "thread", channelId: "room", rootId: root.id }, + newer.id, + ); + expect( + h.journal()?.state.frontiers[ + scope === "dm" ? "room" : `thread:${root.id}` + ], + ).toBe(14); + h.emit(nested ? [parent, selected] : [selected]); + expect(unread.attention("room", selected.id)).toMatchObject({ + unread: false, + forced: false, + }); + }, +); + +it("a scoped DM read preserves an evicted selection in another thread", async () => { + const h = setup(); + h.grant("dm"); + h.emit([metadata(h.relay, "dm", "DM", 11, [["t", "dm"]])]); + const a = message(h.alice, "dm", "thread A", 20); + const selected = message(h.alice, "dm", "selected A reply", 21, [ + ["e", a.id, "", "reply"], + ]); + const b = message(h.viewer, "dm", "thread B", 22); + const newer = message(h.alice, "dm", "unread B reply", 23, [ + ["e", b.id, "", "reply"], + ]); + h.emit([a, selected]); + const unread = h.session.unread; + await unread.enterChannel("dm"); + await unread.markMessageRead("dm", selected.id); + await unread.markMessageUnread("dm", selected.id); + unread.leaveChannel("dm"); + overflowObservations(h, "dm"); + h.emit([b, newer]); + expect(unread.attention("dm", selected.id).status).toBe("unknown"); + expect( + unread.inbox().items.find((row) => row.channelId === "dm"), + ).toMatchObject({ rootId: b.id, threadOnly: true }); + + await unread.prepareChannelRead("dm", b.id)(); + expect(unread.attention("dm", newer.id).unread).toBe(false); + h.emit([a, selected]); + expect(unread.attention("dm", selected.id)).toMatchObject({ + unread: true, + forced: true, + }); + expect(h.journal()?.localUnread["message-force:dm"]).toBeDefined(); + await unread.markChannelRead("dm"); + expect(unread.attention("dm", selected.id)).toMatchObject({ + unread: false, + forced: false, + }); +}); + +it("drops a retained selected-message mark when its message is deleted", async () => { + const h = setup(); + h.grant("room"); + const row = message(h.alice, "room", "read", 11); + h.emit([row]); + const unread = h.session.unread; + await unread.enterChannel("room"); + await unread.markMessageRead("room", row.id); + await unread.markMessageUnread("room", row.id); + unread.leaveChannel("room"); + expect(unread.attention("room", row.id)).toMatchObject({ + unread: true, + forced: true, + }); + + h.emit([signed(h.alice, { kind: 5, content: "", tags: [["e", row.id]] })]); + expect(unread.attention("room", row.id)).toMatchObject({ + status: "ineligible", + unread: false, + forced: false, + }); + expect( + unread.snapshot({ + kind: "message", + channelId: "room", + messageId: row.id, + }), + ).toMatchObject({ manual: "none", observedCount: 0 }); +}); + +it("does not revive a retained selected-message mark after access is restored", async () => { + const h = setup(); + h.grant("room"); + const row = message(h.alice, "room", "read", 11); + h.emit([row]); + const unread = h.session.unread; + await unread.enterChannel("room"); + await unread.markMessageRead("room", row.id); + await unread.markMessageUnread("room", row.id); + unread.leaveChannel("room"); + expect(unread.attention("room", row.id).forced).toBe(true); + + h.emit([roster(h.relay, "room", [], 20)]); + expect(unread.attention("room", row.id).status).toBe("unknown"); + h.grant("room", 21); + h.emit([row]); + await unread.enterChannel("room"); + expect(unread.attention("room", row.id)).toMatchObject({ + unread: false, + forced: false, + }); +}); + it("manual message intent participates in shared counts and is cleared by explicit subtree read", async () => { const h = setup(); h.grant("room"); diff --git a/src/features/relay/unread.ts b/src/features/relay/unread.ts index dec2d29545..079491fdff 100644 --- a/src/features/relay/unread.ts +++ b/src/features/relay/unread.ts @@ -62,7 +62,7 @@ export type MessageAttention = Readonly<{ * attention when the lookup finishes. */ pending?: boolean; unread: boolean; - /** This row is explicitly forced unread during the current channel visit. */ + /** Explicitly forced unread until read, deletion, access retirement, or session end. */ forced: boolean; viewing: boolean; }>; @@ -161,7 +161,7 @@ export interface UnreadCapability { channelId: string, messageId: string, ): Promise; - /** End the channel visit; the device-local sidebar force remains until next open. */ + /** End the visit, retaining exact message targets; the sidebar force ends on next open. */ leaveChannel(channelId: string): void; /** Reconcile the previous visit's sidebar force without clearing independent channel intent. */ enterChannel(channelId: string, active?: boolean): Promise; @@ -321,8 +321,14 @@ export function createUnread({ visible: (messageId: string, measured: boolean) => boolean; } >(); - // Per-visit message overlay; only the sidebar hint is persisted in read state. - const forcedMessages = new Map>(); + // Exact selected-message intent lasts for this host session; read state keeps + // only the channel hint. Keep scope/time even when observation evicts the event. + type ForcedMessage = Readonly<{ + timestamp: number; + rootId: string | undefined; + }>; + const forcedMessages = new Map>(); + const pendingForcedMessages = new Map>(); const entered = new Set(); const openChannels = new Set(); // What each channel's row has seen this session, applied before its marks @@ -433,8 +439,8 @@ export function createUnread({ const byChannel = new Map(); const byId = new Map(); const tombstones = new Set(); - // Ids named by an `e` tag of any retained or fetched event: a deletion, edit - // or reply that arrived before its target. Built with the index. + // Ids named by retained/fetched events or selected-message scope: a deletion, + // edit or reply whose target/ancestry may arrive later. Built with the index. const referenced = new Set(); // Conversation membership: the viewer's own messages, and the parents the // viewer has replied to, each keyed by channel. A conversation is the set of @@ -606,6 +612,29 @@ export function createUnread({ } for (const event of events.values()) if (contentKind(event) && !tombstones.has(event.id)) indexContent(event); + // Retire exact intent only when deletion evidence names it. A bounded + // observation reset is not a read of the selected message. + for (const marks of [forcedMessages, pendingForcedMessages]) { + for (const [channelId, ids] of marks) { + for (const [id, mark] of ids) { + if (tombstones.has(id)) { + ids.delete(id); + continue; + } + referenced.add(id); + if (mark.rootId) referenced.add(mark.rootId); + // A reply-only tag can name an intermediate parent. Keep the verified + // root once ancestry arrives, even if the selection is later evicted. + const row = + structural(id) ?? (mark.rootId && structural(mark.rootId)); + const rootId = + row && channelOf(row) === channelId ? root(row) : undefined; + if (rootId && rootId !== mark.rootId) + ids.set(id, { ...mark, rootId }); + } + if (!ids.size) marks.delete(channelId); + } + } for (const { channelId, evidence } of lookups.values()) { const event = evidence === undefined ? undefined : witnesses.get(evidence); @@ -2176,6 +2205,7 @@ export function createUnread({ continue; const size = new TextEncoder().encode(JSON.stringify(event)).byteLength; if (events.size >= 4096 || bytes + size > 8 * 1024 * 1024) { + indexEvidence(); events.clear(); indexed = false; forget(); @@ -2294,6 +2324,13 @@ export function createUnread({ channelReadGenerations.get(channelId) === generation && allowed(channelId); } + function clearForcedMessages(channelId: string, ids: readonly string[]) { + const forced = forcedMessages.get(channelId); + if (!forced) return; + for (const id of ids) forced.delete(id); + if (!forced.size) forcedMessages.delete(channelId); + publish(new Set([channelId])); + } /** A DM read through retained evidence: its newest message. A reader that * shows only the thread `rootId` stops one second before the earliest * message outside it that the frontier still leaves unread (a top-level @@ -2335,15 +2372,17 @@ export function createUnread({ // so a device clock running ahead also covers arrivals until real time // passes it; automatic reading must never use the clock. A DM thread shown // alone reads as its reader does: a later arrival outside it was not shown. + const thread = rootId && isDm(channelId) ? rootId : undefined; const cut = - rootId && rows.length && isDm(channelId) - ? dmCut(channelId, rootId) + thread && rows.length + ? dmCut(channelId, thread) : rows.reduce( (newest, { event }) => Math.max(newest, event.created_at), clickedAt, ); - const keys = new Set([channelId, messageForceKey(channelId)]); - for (const { event, rootId } of rows) { + const keys = new Set([channelId]); + for (const { event, rootId, threadRootId } of rows) { + if (thread && (rootId ?? threadRootId) !== thread) continue; keys.add(`msg:${event.id}`); if (rootId) keys.add(`thread:${rootId}`); // A retained top-level message establishes its thread's channel even @@ -2352,12 +2391,25 @@ export function createUnread({ } const valid = channelIntentValid(channelId); return async () => { + const forced = forcedMessages.get(channelId); + const ids = [...(forced ?? [])] + .filter(([id, mark]) => { + const row = events.get(id); + return ( + mark.timestamp <= cut && + (!thread || (mark.rootId ?? (row && root(row))) === thread) + ); + }) + .map(([id]) => id); + const clear = new Set(keys); + for (const id of ids) clear.add(`msg:${id}`); + if (!forced || forced.size === ids.length) + clear.add(messageForceKey(channelId)); const result = rows.length || reads.snapshot().capability === "frontier-sync" - ? await reads.read(channelId, cut, valid, true, [...keys]) - : await reads.clearLocalUnread(channelId, [...keys], valid); - if (valid() && forcedMessages.delete(channelId)) - publish(new Set([channelId])); + ? await reads.read(channelId, cut, valid, true, [...clear]) + : await reads.clearLocalUnread(channelId, [...clear], valid); + if (valid()) clearForcedMessages(channelId, ids); return result; }; } @@ -2688,17 +2740,37 @@ export function createUnread({ async markThrough(target, id) { const event = requireMessage(target, id), generation = epoch; - return serialize(target.channelId, () => - reads.read( - targetKey(target), + const current = channelIntentValid(target.channelId); + return serialize(target.channelId, async () => { + const key = targetKey(target); + const forced = forcedMessages.get(target.channelId); + const ids = [...(forced ?? [])] + .filter(([id, mark]) => { + const row = events.get(id); + return ( + mark.timestamp <= event.created_at && + (target.kind === "channel" || + (target.kind === "message" + ? id === target.messageId + : (mark.rootId ?? (row && root(row))) === target.rootId)) + ); + }) + .map(([id]) => id); + const result = await reads.read( + key, event.created_at, () => - !closed && + current() && generation === epoch && requireMessage(target, id) === event, true, - ), - ); + forced?.size === ids.length + ? [key, messageForceKey(target.channelId)] + : [key], + ); + if (current()) clearForcedMessages(target.channelId, ids); + return result; + }); }, async markTimelineRead(channelId, messageIds) { const target = { kind: "channel" as const, channelId }; @@ -2713,16 +2785,19 @@ export function createUnread({ if (!shown.length || shown.some(threadReference)) throw new Error("A thread reply cannot advance the channel frontier"); const generation = epoch; + const current = channelIntentValid(channelId); const valid = () => - !closed && + current() && generation === epoch && shown.every((event, index) => { const id = messageIds[index]; return id !== undefined && requireMessage(target, id) === event; }); - return serialize(channelId, () => { + return serialize(channelId, async () => { indexEvidence(); const state = reads.state(); + const forced = forcedMessages.get(channelId); + const ids = new Set(messageIds); const entries = shown.flatMap((event) => byId.get(event.id) ?? []); const cut = Math.max( ...entries @@ -2734,7 +2809,8 @@ export function createUnread({ .filter( (entry) => (!timeline(entry, false) || - !!reads.localUnread(`msg:${entry.event.id}`)) && + !!reads.localUnread(`msg:${entry.event.id}`) || + forced?.has(entry.event.id)) && isUnread(entry, state, false), ) .map((entry) => ({ @@ -2747,16 +2823,20 @@ export function createUnread({ state.frontiers[`activity:${channelId}`] ?? -1, state.frontiers[channelId] ?? -1, ) >= cut; - return reads.readMessages( + const result = await reads.readMessages( covered ? marks : [ { key: `activity:${channelId}`, timestamp: cut, channelId }, ...marks, ], - undefined, + forced && [...forced.keys()].every((id) => ids.has(id)) + ? messageForceKey(channelId) + : undefined, valid, ); + if (current()) clearForcedMessages(channelId, messageIds); + return result; }); }, async markMessageUnread(channelId, messageId) { @@ -2765,42 +2845,60 @@ export function createUnread({ refuseAgentCoordination(messageId); const generation = epoch; return serialize(channelId, async () => { - const valid = () => - !closed && - generation === epoch && + const current = () => + !closed && generation === epoch && allowed(channelId); + const visitValid = () => + current() && (visits.get(channelId) ?? 0) === visit && - allowed(channelId) && rows.every((row) => events.get(row.id) === row && !deleted(row)); - if (!valid()) throw new Error("Unread channel visit expired"); - const result = await reads.markLocalUnread( - messageForceKey(channelId), - valid, + if (!visitValid()) throw new Error("Unread channel visit expired"); + const targets = new Map( + rows.map((row) => [ + row.id, + { + timestamp: row.created_at, + rootId: root(row) ?? threadReference(row)?.rootId, + }, + ]), ); - // A leave after persistence must not restore a previous visit's overlay. - if (valid()) { - const forced = forcedMessages.get(channelId) ?? new Set(); - for (const row of rows) forced.add(row.id); - forcedMessages.set(channelId, forced); - publish(new Set([channelId])); + // Reconcile deletion/ancestry while storage finishes, without exposing + // unsaved intent or requiring the observation cache to keep its rows. + pendingForcedMessages.set(channelId, targets); + try { + const result = await reads.markLocalUnread( + messageForceKey(channelId), + visitValid, + ); + indexEvidence(); + if (current() && targets.size) { + const forced = + forcedMessages.get(channelId) ?? new Map(); + for (const [id, mark] of targets) forced.set(id, mark); + forcedMessages.set(channelId, forced); + publish(new Set([channelId])); + } + return result; + } finally { + pendingForcedMessages.delete(channelId); } - return result; }); }, async markMessageRead(channelId, messageId) { const visit = visits.get(channelId) ?? 0; const rows = messageSubtree(channelId, messageId); const generation = epoch; + const current = channelIntentValid(channelId); return serialize(channelId, async () => { const valid = () => - !closed && + current() && generation === epoch && (visits.get(channelId) ?? 0) === visit && - allowed(channelId) && rows.every((row) => events.get(row.id) === row && !deleted(row)); if (!valid()) throw new Error("Unread channel visit expired"); const forced = forcedMessages.get(channelId); const ids = new Set(rows.map((row) => row.id)); - const remaining = forced && [...forced].some((id) => !ids.has(id)); + const remaining = + forced && [...forced.keys()].some((id) => !ids.has(id)); const result = await reads.readMessages( rows .filter((row) => row.pubkey !== viewer) @@ -2815,18 +2913,13 @@ export function createUnread({ remaining ? undefined : messageForceKey(channelId), valid, ); - if (valid()) { - for (const row of rows) forced?.delete(row.id); - if (!forced?.size) forcedMessages.delete(channelId); - publish(new Set([channelId])); - } + if (current()) clearForcedMessages(channelId, [...ids]); return result; }); }, leaveChannel(channelId) { openChannels.delete(channelId); visits.set(channelId, (visits.get(channelId) ?? 0) + 1); - forcedMessages.delete(channelId); if (!closed && allowed(channelId)) { indexEvidence(); seeChannel(channelId); @@ -2903,6 +2996,7 @@ export function createUnread({ // Explicit Mark all must consume retained unread intent, not styling. return ( current.manual !== "none" || + !!forcedMessages.get(channel.id)?.size || (byChannel.get(channel.id) ?? []).some( (entry) => isUnread(entry, reads.state(), channel.channelType === "dm") ||