mirror of
https://github.com/block/buzz.git
synced 2026-08-18 06:50:31 +02:00
Fix monotonic read-state merges (#884)
Signed-off-by: Wes <wesbillman@users.noreply.github.com> Co-authored-by: Pinky <44b8e82baa6e0e254e0208d68f335c283c94e7b78dd1fa10d5a49d3f13dd0435@sprout-oss.stage.blox.sqprod.co>
This commit is contained in:
Wes
@@ -7,10 +7,7 @@ type AppShellContextValue = {
|
||||
channelId: string,
|
||||
readAt: string | null | undefined,
|
||||
) => void;
|
||||
markChannelUnread: (
|
||||
channelId: string,
|
||||
lastMessageAt: string | null | undefined,
|
||||
) => void;
|
||||
markChannelUnread: (channelId: string) => void;
|
||||
openChannelManagement: () => void;
|
||||
// NIP-RS read marker for a channel as a unix-seconds timestamp, or null
|
||||
// when unknown. Backed by the single AppShell-mounted ReadStateManager so
|
||||
|
||||
@@ -0,0 +1,55 @@
|
||||
import assert from "node:assert/strict";
|
||||
import test from "node:test";
|
||||
|
||||
import { applyRemoteContextTimestamp } from "./readStateManager.ts";
|
||||
|
||||
test("applyRemoteContextTimestamp ignores older remote read markers from newer sync events", () => {
|
||||
const effectiveState = new Map([["channel-1", 200]]);
|
||||
const contextSourceCreatedAt = new Map([["channel-1", 10]]);
|
||||
|
||||
const result = applyRemoteContextTimestamp({
|
||||
effectiveState,
|
||||
contextSourceCreatedAt,
|
||||
contextId: "channel-1",
|
||||
timestamp: 100,
|
||||
eventCreatedAt: 11,
|
||||
});
|
||||
|
||||
assert.equal(result, "unchanged");
|
||||
assert.equal(effectiveState.get("channel-1"), 200);
|
||||
assert.equal(contextSourceCreatedAt.get("channel-1"), 11);
|
||||
});
|
||||
|
||||
test("applyRemoteContextTimestamp advances to newer remote read markers", () => {
|
||||
const effectiveState = new Map([["channel-1", 100]]);
|
||||
const contextSourceCreatedAt = new Map([["channel-1", 10]]);
|
||||
|
||||
const result = applyRemoteContextTimestamp({
|
||||
effectiveState,
|
||||
contextSourceCreatedAt,
|
||||
contextId: "channel-1",
|
||||
timestamp: 200,
|
||||
eventCreatedAt: 11,
|
||||
});
|
||||
|
||||
assert.equal(result, "advanced");
|
||||
assert.equal(effectiveState.get("channel-1"), 200);
|
||||
assert.equal(contextSourceCreatedAt.get("channel-1"), 11);
|
||||
});
|
||||
|
||||
test("applyRemoteContextTimestamp keeps read markers monotonic even if sync events arrive out of order", () => {
|
||||
const effectiveState = new Map([["channel-1", 100]]);
|
||||
const contextSourceCreatedAt = new Map([["channel-1", 11]]);
|
||||
|
||||
const result = applyRemoteContextTimestamp({
|
||||
effectiveState,
|
||||
contextSourceCreatedAt,
|
||||
contextId: "channel-1",
|
||||
timestamp: 200,
|
||||
eventCreatedAt: 10,
|
||||
});
|
||||
|
||||
assert.equal(result, "advanced");
|
||||
assert.equal(effectiveState.get("channel-1"), 200);
|
||||
assert.equal(contextSourceCreatedAt.get("channel-1"), 11);
|
||||
});
|
||||
@@ -47,6 +47,49 @@ function slotIdKey(pubkey: string): string {
|
||||
return `${SLOT_ID_KEY_PREFIX}:${pubkey}`;
|
||||
}
|
||||
|
||||
export type ApplyRemoteContextResult = "unchanged" | "advanced";
|
||||
|
||||
function resolveRemoteContextTimestamp(args: {
|
||||
current: number;
|
||||
timestamp: number;
|
||||
}): { next: number; result: ApplyRemoteContextResult } {
|
||||
const next = Math.max(args.current, args.timestamp);
|
||||
return {
|
||||
next,
|
||||
result: next === args.current ? "unchanged" : "advanced",
|
||||
};
|
||||
}
|
||||
|
||||
export function applyRemoteContextTimestamp(args: {
|
||||
effectiveState: Map<string, number>;
|
||||
contextSourceCreatedAt: Map<string, number>;
|
||||
contextId: string;
|
||||
timestamp: number;
|
||||
eventCreatedAt: number;
|
||||
}): ApplyRemoteContextResult {
|
||||
const {
|
||||
effectiveState,
|
||||
contextSourceCreatedAt,
|
||||
contextId,
|
||||
timestamp,
|
||||
eventCreatedAt,
|
||||
} = args;
|
||||
const sourceCreatedAt = contextSourceCreatedAt.get(contextId) ?? 0;
|
||||
const current = effectiveState.get(contextId) ?? 0;
|
||||
const { next, result } = resolveRemoteContextTimestamp({
|
||||
current,
|
||||
timestamp,
|
||||
});
|
||||
|
||||
if (result === "advanced") {
|
||||
effectiveState.set(contextId, next);
|
||||
}
|
||||
if (eventCreatedAt > sourceCreatedAt) {
|
||||
contextSourceCreatedAt.set(contextId, eventCreatedAt);
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
export class ReadStateManager {
|
||||
private pubkey: string;
|
||||
private relayClient: RelayClient;
|
||||
@@ -60,9 +103,7 @@ export class ReadStateManager {
|
||||
private unsubscribeLive: (() => void) | null = null;
|
||||
private initialized = false;
|
||||
private maxFetchedCreatedAt = 0;
|
||||
private forcedContexts = new Set<string>();
|
||||
private contextSourceCreatedAt = new Map<string, number>();
|
||||
private pendingSyncedRollbacks = new Set<string>();
|
||||
private pendingSyncedAdvances = new Set<string>();
|
||||
private destroyed = false;
|
||||
|
||||
@@ -101,7 +142,6 @@ export class ReadStateManager {
|
||||
}
|
||||
|
||||
markContextRead(contextId: string, unixTimestamp: number): void {
|
||||
this.forcedContexts.delete(contextId);
|
||||
this.advanceContext(contextId, unixTimestamp, { publishable: true });
|
||||
this.contextSourceCreatedAt.set(
|
||||
contextId,
|
||||
@@ -113,16 +153,6 @@ export class ReadStateManager {
|
||||
this.advanceContext(contextId, unixTimestamp, { publishable: false });
|
||||
}
|
||||
|
||||
markContextUnread(contextId: string, lastMessageUnix: number): void {
|
||||
const rollbackTo = lastMessageUnix - 1;
|
||||
this.effectiveState.set(contextId, rollbackTo);
|
||||
this.publishableContextIds.add(contextId);
|
||||
this.forcedContexts.add(contextId);
|
||||
this.persistLocalState();
|
||||
this.notifyListeners();
|
||||
this.schedulePublish();
|
||||
}
|
||||
|
||||
private advanceContext(
|
||||
contextId: string,
|
||||
unixTimestamp: number,
|
||||
@@ -241,16 +271,17 @@ export class ReadStateManager {
|
||||
}
|
||||
|
||||
for (const [ctx, ts] of Object.entries(blob.contexts)) {
|
||||
if (this.forcedContexts.has(ctx)) continue;
|
||||
const sourceCreatedAt = this.contextSourceCreatedAt.get(ctx) ?? 0;
|
||||
const current = this.effectiveState.get(ctx) ?? 0;
|
||||
if (event.created_at > sourceCreatedAt) {
|
||||
this.effectiveState.set(ctx, ts);
|
||||
this.contextSourceCreatedAt.set(ctx, event.created_at);
|
||||
} else if (event.created_at === sourceCreatedAt && ts !== current) {
|
||||
this.effectiveState.set(ctx, ts);
|
||||
const result = applyRemoteContextTimestamp({
|
||||
effectiveState: this.effectiveState,
|
||||
contextSourceCreatedAt: this.contextSourceCreatedAt,
|
||||
contextId: ctx,
|
||||
timestamp: ts,
|
||||
eventCreatedAt: event.created_at,
|
||||
});
|
||||
if (result !== "unchanged") {
|
||||
this.pendingSyncedAdvances.add(ctx);
|
||||
this.publishableContextIds.add(ctx);
|
||||
}
|
||||
this.publishableContextIds.add(ctx);
|
||||
}
|
||||
|
||||
if (blob.client_id === this.clientId) {
|
||||
@@ -361,27 +392,15 @@ export class ReadStateManager {
|
||||
|
||||
let anyAdvanced = false;
|
||||
for (const [ctx, ts] of Object.entries(blob.contexts)) {
|
||||
if (this.forcedContexts.has(ctx)) continue;
|
||||
const sourceCreatedAt = this.contextSourceCreatedAt.get(ctx) ?? 0;
|
||||
const current = this.effectiveState.get(ctx) ?? 0;
|
||||
if (event.created_at > sourceCreatedAt) {
|
||||
if (this.effectiveState.get(ctx) !== ts) {
|
||||
if (ts < current && current > 0) {
|
||||
this.pendingSyncedRollbacks.add(ctx);
|
||||
this.pendingSyncedAdvances.delete(ctx);
|
||||
console.debug(
|
||||
`[ReadStateManager] synced rollback ctx=${ctx.substring(0, 12)}… from=${current} to=${ts}`,
|
||||
);
|
||||
} else if (ts > current) {
|
||||
this.pendingSyncedAdvances.add(ctx);
|
||||
this.pendingSyncedRollbacks.delete(ctx);
|
||||
}
|
||||
this.effectiveState.set(ctx, ts);
|
||||
anyAdvanced = true;
|
||||
}
|
||||
this.contextSourceCreatedAt.set(ctx, event.created_at);
|
||||
} else if (event.created_at === sourceCreatedAt && ts !== current) {
|
||||
this.effectiveState.set(ctx, ts);
|
||||
const result = applyRemoteContextTimestamp({
|
||||
effectiveState: this.effectiveState,
|
||||
contextSourceCreatedAt: this.contextSourceCreatedAt,
|
||||
contextId: ctx,
|
||||
timestamp: ts,
|
||||
eventCreatedAt: event.created_at,
|
||||
});
|
||||
if (result === "advanced") {
|
||||
this.pendingSyncedAdvances.add(ctx);
|
||||
anyAdvanced = true;
|
||||
}
|
||||
if (!this.publishableContextIds.has(ctx)) {
|
||||
@@ -467,7 +486,6 @@ export class ReadStateManager {
|
||||
}
|
||||
}
|
||||
this.lastPublishedContexts = contexts;
|
||||
this.forcedContexts.clear();
|
||||
this.maxFetchedCreatedAt = Math.max(
|
||||
this.maxFetchedCreatedAt,
|
||||
event.created_at,
|
||||
@@ -544,12 +562,6 @@ export class ReadStateManager {
|
||||
);
|
||||
}
|
||||
|
||||
drainSyncedRollbacks(): ReadonlySet<string> {
|
||||
const drained = this.pendingSyncedRollbacks;
|
||||
this.pendingSyncedRollbacks = new Set<string>();
|
||||
return drained;
|
||||
}
|
||||
|
||||
drainSyncedAdvances(): ReadonlySet<string> {
|
||||
const drained = this.pendingSyncedAdvances;
|
||||
this.pendingSyncedAdvances = new Set<string>();
|
||||
|
||||
@@ -4,8 +4,6 @@ import type { RelayClient } from "@/shared/api/relayClientSession";
|
||||
|
||||
const noopGetTimestamp = () => null;
|
||||
const noopMarkRead = () => {};
|
||||
const noopMarkUnread = () => {};
|
||||
const noopDrainRollbacks = (): ReadonlySet<string> => new Set<string>();
|
||||
const noopDrainAdvances = (): ReadonlySet<string> => new Set<string>();
|
||||
|
||||
/**
|
||||
@@ -66,13 +64,6 @@ export function useReadState(
|
||||
[],
|
||||
);
|
||||
|
||||
const markContextUnread = React.useCallback(
|
||||
(contextId: string, lastMessageUnix: number): void => {
|
||||
managerRef.current?.markContextUnread(contextId, lastMessageUnix);
|
||||
},
|
||||
[],
|
||||
);
|
||||
|
||||
const seedContextRead = React.useCallback(
|
||||
(contextId: string, unixTimestamp: number): void => {
|
||||
managerRef.current?.seedContextRead(contextId, unixTimestamp);
|
||||
@@ -80,10 +71,6 @@ export function useReadState(
|
||||
[],
|
||||
);
|
||||
|
||||
const drainSyncedRollbacks = React.useCallback((): ReadonlySet<string> => {
|
||||
return managerRef.current?.drainSyncedRollbacks() ?? new Set<string>();
|
||||
}, []);
|
||||
|
||||
const drainSyncedAdvances = React.useCallback((): ReadonlySet<string> => {
|
||||
return managerRef.current?.drainSyncedAdvances() ?? new Set<string>();
|
||||
}, []);
|
||||
@@ -97,9 +84,7 @@ export function useReadState(
|
||||
getEffectiveTimestamp: noopGetTimestamp,
|
||||
isReady: false,
|
||||
markContextRead: noopMarkRead,
|
||||
markContextUnread: noopMarkUnread,
|
||||
seedContextRead: noopMarkRead,
|
||||
drainSyncedRollbacks: noopDrainRollbacks,
|
||||
drainSyncedAdvances: noopDrainAdvances,
|
||||
readStateVersion: 0,
|
||||
};
|
||||
@@ -109,9 +94,7 @@ export function useReadState(
|
||||
getEffectiveTimestamp,
|
||||
isReady,
|
||||
markContextRead,
|
||||
markContextUnread,
|
||||
seedContextRead,
|
||||
drainSyncedRollbacks,
|
||||
drainSyncedAdvances,
|
||||
readStateVersion,
|
||||
};
|
||||
|
||||
@@ -34,7 +34,6 @@ import {
|
||||
} from "@/features/messages/lib/formatTimelineMessages";
|
||||
import { buildThreadPanelData } from "@/features/messages/lib/threadPanel";
|
||||
import { imetaMediaFromTags } from "@/features/messages/lib/imetaMediaMarkdown";
|
||||
import type { TimelineMessage } from "@/features/messages/types";
|
||||
import { useFetchOlderMessages } from "@/features/messages/useFetchOlderMessages";
|
||||
import { useLoadMissingAncestors } from "@/features/messages/useLoadMissingAncestors";
|
||||
import { useChannelTyping } from "@/features/messages/useChannelTyping";
|
||||
@@ -334,14 +333,10 @@ export function ChannelScreen({
|
||||
: undefined,
|
||||
[activeChannel, handleToggleReaction],
|
||||
);
|
||||
const handleMarkUnread = React.useCallback(
|
||||
(message: TimelineMessage) => {
|
||||
if (!activeChannelId) return;
|
||||
const messageIso = new Date(message.createdAt * 1_000).toISOString();
|
||||
markChannelUnread(activeChannelId, messageIso);
|
||||
},
|
||||
[activeChannelId, markChannelUnread],
|
||||
);
|
||||
const handleMarkUnread = React.useCallback(() => {
|
||||
if (!activeChannelId) return;
|
||||
markChannelUnread(activeChannelId);
|
||||
}, [activeChannelId, markChannelUnread]);
|
||||
const {
|
||||
channelAgentSessionAgents,
|
||||
closeAgentSession: handleCloseAgentSession,
|
||||
|
||||
@@ -246,8 +246,6 @@ export function useUnreadChannels(
|
||||
getEffectiveTimestamp,
|
||||
isReady: isReadStateReady,
|
||||
markContextRead,
|
||||
markContextUnread,
|
||||
drainSyncedRollbacks,
|
||||
drainSyncedAdvances,
|
||||
readStateVersion,
|
||||
} = useReadState(pubkey, relayClient);
|
||||
@@ -268,35 +266,24 @@ export function useUnreadChannels(
|
||||
channelsRef.current = channels;
|
||||
|
||||
// Channels manually marked unread this session (e.g., right-click → "mark
|
||||
// unread"). The NIP-RS rollback (markContextUnread) is the cross-device
|
||||
// mechanism; this in-session flag is what makes the badge appear *now* in
|
||||
// the case where we don't yet have an observed latest timestamp to compare
|
||||
// against. Cleared when the user opens the channel.
|
||||
// unread"). Because NIP-RS read markers are monotonic, this in-session flag
|
||||
// is what makes the badge appear *now* without lowering synced read state.
|
||||
// Cleared when the user opens the channel.
|
||||
const forcedUnreadRef = React.useRef(new Set<string>());
|
||||
|
||||
// When a synced event rolls back a read marker (cross-device mark-as-unread),
|
||||
// merge into forcedUnreadRef so the badge appears immediately without waiting
|
||||
// for a catch-up REQ that already ran with the old (higher) marker.
|
||||
// When a synced event advances a read marker (cross-device mark-as-read),
|
||||
// remove from forcedUnreadRef so the dot clears immediately.
|
||||
// biome-ignore lint/correctness/useExhaustiveDependencies: readStateVersion is the intentional drain trigger
|
||||
React.useEffect(() => {
|
||||
const rolled = drainSyncedRollbacks();
|
||||
const advanced = drainSyncedAdvances();
|
||||
let anyNew = false;
|
||||
for (const channelId of rolled) {
|
||||
if (!forcedUnreadRef.current.has(channelId)) {
|
||||
forcedUnreadRef.current.add(channelId);
|
||||
anyNew = true;
|
||||
}
|
||||
}
|
||||
for (const channelId of advanced) {
|
||||
if (forcedUnreadRef.current.delete(channelId)) {
|
||||
anyNew = true;
|
||||
}
|
||||
}
|
||||
if (anyNew) bumpLatestVersion();
|
||||
}, [readStateVersion, drainSyncedRollbacks, drainSyncedAdvances]);
|
||||
}, [readStateVersion, drainSyncedAdvances]);
|
||||
|
||||
// Root event IDs of threads where the current user has replied at least once.
|
||||
// Used to determine if thread replies should trigger unread notifications.
|
||||
@@ -375,28 +362,14 @@ export function useUnreadChannels(
|
||||
);
|
||||
|
||||
// Manually mark a channel unread (e.g., right-click → "mark unread"). Sets
|
||||
// the in-session forced flag so the sidebar badge appears immediately, and
|
||||
// rolls the NIP-RS read marker back so the unread state syncs across
|
||||
// devices. The forced flag is cleared in markChannelRead when the user
|
||||
// opens the channel. If lastMessageAt is unknown we still set the forced
|
||||
// flag, but skip the NIP-RS rollback — without a target timestamp we have
|
||||
// nothing honest to publish.
|
||||
const markChannelUnread = React.useCallback(
|
||||
(channelId: string, lastMessageAt: string | null | undefined) => {
|
||||
if (!forcedUnreadRef.current.has(channelId)) {
|
||||
forcedUnreadRef.current.add(channelId);
|
||||
bumpLatestVersion();
|
||||
}
|
||||
const unixSeconds =
|
||||
toUnixSeconds(lastMessageAt) ??
|
||||
latestByChannelRef.current.get(channelId) ??
|
||||
null;
|
||||
if (unixSeconds !== null) {
|
||||
markContextUnread(channelId, unixSeconds);
|
||||
}
|
||||
},
|
||||
[markContextUnread],
|
||||
);
|
||||
// the in-session forced flag so the sidebar badge appears immediately. NIP-RS
|
||||
// read markers are monotonic, so we do not publish a lower timestamp.
|
||||
const markChannelUnread = React.useCallback((channelId: string) => {
|
||||
if (!forcedUnreadRef.current.has(channelId)) {
|
||||
forcedUnreadRef.current.add(channelId);
|
||||
bumpLatestVersion();
|
||||
}
|
||||
}, []);
|
||||
|
||||
// Mark the active channel as read when it changes or new messages arrive.
|
||||
// Honours the caller's contract that a null activeReadAt suppresses
|
||||
|
||||
@@ -16,11 +16,8 @@ type UseHomeInboxReadStateOptions = {
|
||||
channelId: string,
|
||||
readAt: string | null | undefined,
|
||||
) => void;
|
||||
/** Roll the NIP-RS read marker back to the given ISO timestamp. */
|
||||
markChannelUnread: (
|
||||
channelId: string,
|
||||
lastMessageAt: string | null | undefined,
|
||||
) => void;
|
||||
/** Mark a channel unread locally for the current session. */
|
||||
markChannelUnread: (channelId: string) => void;
|
||||
/** Local fallback: mark a non-channel item done. */
|
||||
markDoneLocal: (id: string) => void;
|
||||
/** Local fallback: undo a non-channel item done. */
|
||||
@@ -35,9 +32,8 @@ type UseHomeInboxReadStateOptions = {
|
||||
* "Mark as read/unread" actions on channel-backed items are routed through
|
||||
* `markChannelRead`/`markChannelUnread` so the sidebar, home badge, and any
|
||||
* other surfaces consuming the same ReadStateManager stay in lockstep.
|
||||
* Caveat: marking an older item unread rolls the *entire* channel marker
|
||||
* back to that item, so newer events in that channel become unread too —
|
||||
* that matches NIP-RS's channel-level granularity.
|
||||
* Caveat: NIP-RS channel read markers are monotonic, so marking an older item
|
||||
* unread is an in-session local affordance rather than synced state.
|
||||
*/
|
||||
export function useHomeInboxReadState({
|
||||
items,
|
||||
@@ -94,10 +90,7 @@ export function useHomeInboxReadState({
|
||||
const item = itemById.get(itemId);
|
||||
const channelId = item?.item.channelId ?? null;
|
||||
if (item && channelId) {
|
||||
markChannelUnread(
|
||||
channelId,
|
||||
new Date(item.latestActivityAt * 1_000).toISOString(),
|
||||
);
|
||||
markChannelUnread(channelId);
|
||||
return;
|
||||
}
|
||||
undoDoneLocal(itemId);
|
||||
|
||||
@@ -119,10 +119,7 @@ type AppSidebarProps = {
|
||||
onOpenBrowseChannels: () => void;
|
||||
onOpenBrowseForums: () => void;
|
||||
onHideDm: (channelId: string) => void;
|
||||
onMarkChannelUnread: (
|
||||
channelId: string,
|
||||
lastMessageAt: string | null | undefined,
|
||||
) => void;
|
||||
onMarkChannelUnread: (channelId: string) => void;
|
||||
onMarkChannelRead: (
|
||||
channelId: string,
|
||||
lastMessageAt: string | null | undefined,
|
||||
|
||||
@@ -142,10 +142,7 @@ export function ChannelContextMenuItems({
|
||||
channelId: string,
|
||||
lastMessageAt: string | null | undefined,
|
||||
) => void;
|
||||
onMarkChannelUnread?: (
|
||||
channelId: string,
|
||||
lastMessageAt: string | null | undefined,
|
||||
) => void;
|
||||
onMarkChannelUnread?: (channelId: string) => void;
|
||||
onMuteChannel?: (channelId: string) => void;
|
||||
onUnmuteChannel?: (channelId: string) => void;
|
||||
onStarChannel?: (channelId: string) => void;
|
||||
@@ -182,9 +179,7 @@ export function ChannelContextMenuItems({
|
||||
Mark as read
|
||||
</ContextMenuItem>
|
||||
) : !hasUnread && onMarkChannelUnread ? (
|
||||
<ContextMenuItem
|
||||
onClick={() => onMarkChannelUnread(channel.id, channel.lastMessageAt)}
|
||||
>
|
||||
<ContextMenuItem onClick={() => onMarkChannelUnread(channel.id)}>
|
||||
<CircleDot className="h-4 w-4" />
|
||||
Mark unread
|
||||
</ContextMenuItem>
|
||||
@@ -340,10 +335,7 @@ export function ChannelGroupSection({
|
||||
channelId: string,
|
||||
lastMessageAt: string | null | undefined,
|
||||
) => void;
|
||||
onMarkChannelUnread: (
|
||||
channelId: string,
|
||||
lastMessageAt: string | null | undefined,
|
||||
) => void;
|
||||
onMarkChannelUnread: (channelId: string) => void;
|
||||
onSelectChannel: (channelId: string) => void;
|
||||
onToggleCollapsed: () => void;
|
||||
selectedChannelId: string | null;
|
||||
@@ -517,10 +509,7 @@ export function CustomChannelSection({
|
||||
channelId: string,
|
||||
lastMessageAt: string | null | undefined,
|
||||
) => void;
|
||||
onMarkChannelUnread: (
|
||||
channelId: string,
|
||||
lastMessageAt: string | null | undefined,
|
||||
) => void;
|
||||
onMarkChannelUnread: (channelId: string) => void;
|
||||
onMarkSectionRead: () => void;
|
||||
onAssignChannel: (channelId: string, sectionId: string) => void;
|
||||
onUnassignChannel: (channelId: string) => void;
|
||||
|
||||
@@ -253,10 +253,7 @@ export function SidebarSection({
|
||||
channelId: string,
|
||||
lastMessageAt: string | null | undefined,
|
||||
) => void;
|
||||
onMarkChannelUnread?: (
|
||||
channelId: string,
|
||||
lastMessageAt: string | null | undefined,
|
||||
) => void;
|
||||
onMarkChannelUnread?: (channelId: string) => void;
|
||||
onSelectChannel: (channelId: string) => void;
|
||||
onToggleCollapsed?: () => void;
|
||||
mutedChannelIds?: ReadonlySet<string>;
|
||||
|
||||
@@ -206,7 +206,7 @@ test("mark-as-unread via context menu shows dot badge", async ({ page }) => {
|
||||
await waitForBadgeState(page, { state: "dot" });
|
||||
});
|
||||
|
||||
test("synced mark-as-unread from another device shows dot, synced mark-as-read clears it", async ({
|
||||
test("remote read-state rollback is ignored while local mark-unread still shows dot", async ({
|
||||
page,
|
||||
}) => {
|
||||
await page.goto("/");
|
||||
@@ -271,8 +271,8 @@ test("synced mark-as-unread from another device shows dot, synced mark-as-read c
|
||||
},
|
||||
);
|
||||
|
||||
// Step 2: rollback — read timestamp drops (another device marks unread).
|
||||
// createdAt must be strictly greater than step 1 to pass LWW gate.
|
||||
// Step 2: a remote rollback carries an older read timestamp in a newer
|
||||
// event. NIP-RS read markers are monotonic, so this must be ignored.
|
||||
await page.evaluate(
|
||||
({ clientId, slotId, channelId, ts, createdAt }) => {
|
||||
(
|
||||
@@ -300,10 +300,16 @@ test("synced mark-as-unread from another device shows dot, synced mark-as-read c
|
||||
},
|
||||
);
|
||||
|
||||
// The unread dot should appear.
|
||||
await expect(page.getByTestId("channel-unread-random")).toBeVisible();
|
||||
await expect(page.getByTestId("channel-unread-random")).toHaveCount(0);
|
||||
|
||||
// Step 3: advance — read timestamp moves forward (device marks read).
|
||||
// Local mark-unread remains an in-session affordance and should still show
|
||||
// the dot immediately without publishing a lower read timestamp.
|
||||
await page.getByTestId("channel-random").click({ button: "right" });
|
||||
await page.getByText("Mark unread").click();
|
||||
await expect(page.getByTestId("channel-unread-random")).toBeVisible();
|
||||
await waitForBadgeState(page, { state: "dot" });
|
||||
|
||||
// Step 3: remote advance clears the local forced-unread dot.
|
||||
await page.evaluate(
|
||||
({ clientId, slotId, channelId, ts, createdAt }) => {
|
||||
(
|
||||
@@ -326,11 +332,10 @@ test("synced mark-as-unread from another device shows dot, synced mark-as-read c
|
||||
clientId: REMOTE_CLIENT_ID,
|
||||
slotId: REMOTE_SLOT_ID,
|
||||
channelId: RANDOM_CHANNEL_ID,
|
||||
ts: now,
|
||||
ts: now + 10,
|
||||
createdAt: now + 10,
|
||||
},
|
||||
);
|
||||
|
||||
// The unread dot should disappear.
|
||||
await expect(page.getByTestId("channel-unread-random")).toHaveCount(0);
|
||||
});
|
||||
|
||||
@@ -37,7 +37,7 @@ enum _QuickAction { createChannel, createForum, newDm }
|
||||
const double _kBannerHeight = 24.0;
|
||||
|
||||
bool _isUnread(Channel channel, ReadStateState readState) {
|
||||
if (readState.syncedForcedChannelIds.contains(channel.id)) {
|
||||
if (readState.locallyForcedChannelIds.contains(channel.id)) {
|
||||
return true;
|
||||
}
|
||||
|
||||
@@ -1119,7 +1119,7 @@ class _ChannelTile extends ConsumerWidget {
|
||||
} else {
|
||||
ref
|
||||
.read(readStateProvider.notifier)
|
||||
.markContextUnread(channel.id, ts);
|
||||
.markContextUnread(channel.id);
|
||||
}
|
||||
}
|
||||
},
|
||||
|
||||
@@ -39,6 +39,8 @@ class ReadStateCrypto {
|
||||
nip44Decrypt(conversationKey, ciphertext);
|
||||
}
|
||||
|
||||
enum _ApplyRemoteContextResult { unchanged, advanced }
|
||||
|
||||
class ReadStateManager {
|
||||
final String pubkey;
|
||||
final ReadStateCrypto _crypto;
|
||||
@@ -63,9 +65,7 @@ class ReadStateManager {
|
||||
Completer<void>? _publishCompleter;
|
||||
bool _remoteUnsupported = false;
|
||||
int _maxFetchedCreatedAt = 0;
|
||||
final Set<String> _forcedContextIds = {};
|
||||
final Map<String, int> _contextSourceCreatedAt = {};
|
||||
final Set<String> _pendingSyncedRollbacks = {};
|
||||
final Set<String> _pendingSyncedAdvances = {};
|
||||
|
||||
ReadStateManager({
|
||||
@@ -116,7 +116,6 @@ class ReadStateManager {
|
||||
}
|
||||
|
||||
void markContextRead(String contextId, int unixTimestamp) {
|
||||
_forcedContextIds.remove(contextId);
|
||||
_advanceContext(contextId, unixTimestamp, publishable: true);
|
||||
_contextSourceCreatedAt[contextId] = max(
|
||||
currentUnixSeconds(),
|
||||
@@ -124,17 +123,6 @@ class ReadStateManager {
|
||||
);
|
||||
}
|
||||
|
||||
void markContextUnread(String contextId, int lastMessageTimestamp) {
|
||||
if (_disposed || lastMessageTimestamp <= 0) return;
|
||||
final rollbackTo = lastMessageTimestamp - 1;
|
||||
_effectiveState[contextId] = rollbackTo;
|
||||
_publishableContextIds.add(contextId);
|
||||
_forcedContextIds.add(contextId);
|
||||
_persistLocalState();
|
||||
_onChanged();
|
||||
_schedulePublish();
|
||||
}
|
||||
|
||||
void seedContextRead(String contextId, int unixTimestamp) {
|
||||
_advanceContext(contextId, unixTimestamp, publishable: false);
|
||||
}
|
||||
@@ -255,17 +243,15 @@ class ReadStateManager {
|
||||
}
|
||||
|
||||
for (final entry in decoded.blob.contexts.entries) {
|
||||
if (_forcedContextIds.contains(entry.key)) continue;
|
||||
final sourceCreatedAt = _contextSourceCreatedAt[entry.key] ?? 0;
|
||||
final current = _effectiveState[entry.key] ?? 0;
|
||||
if (event.createdAt > sourceCreatedAt) {
|
||||
_effectiveState[entry.key] = entry.value;
|
||||
_contextSourceCreatedAt[entry.key] = event.createdAt;
|
||||
} else if (event.createdAt == sourceCreatedAt &&
|
||||
entry.value != current) {
|
||||
_effectiveState[entry.key] = entry.value;
|
||||
final result = _applyRemoteContextTimestamp(
|
||||
contextId: entry.key,
|
||||
timestamp: entry.value,
|
||||
eventCreatedAt: event.createdAt,
|
||||
);
|
||||
if (result == _ApplyRemoteContextResult.advanced) {
|
||||
_pendingSyncedAdvances.add(entry.key);
|
||||
_publishableContextIds.add(entry.key);
|
||||
}
|
||||
_publishableContextIds.add(entry.key);
|
||||
}
|
||||
|
||||
if (decoded.blob.clientId == _clientId &&
|
||||
@@ -329,27 +315,13 @@ class ReadStateManager {
|
||||
|
||||
var changed = false;
|
||||
for (final entry in decoded.blob.contexts.entries) {
|
||||
if (_forcedContextIds.contains(entry.key)) continue;
|
||||
final sourceCreatedAt = _contextSourceCreatedAt[entry.key] ?? 0;
|
||||
final current = _effectiveState[entry.key] ?? 0;
|
||||
if (event.createdAt > sourceCreatedAt) {
|
||||
if (_effectiveState[entry.key] != entry.value) {
|
||||
if (entry.value < current && current > 0) {
|
||||
_pendingSyncedRollbacks.add(entry.key);
|
||||
_pendingSyncedAdvances.remove(entry.key);
|
||||
debugPrint(
|
||||
'[ReadStateManager] synced rollback ctx=${entry.key.substring(0, min(12, entry.key.length))}… from=$current to=${entry.value}',
|
||||
);
|
||||
} else if (entry.value > current) {
|
||||
_pendingSyncedAdvances.add(entry.key);
|
||||
_pendingSyncedRollbacks.remove(entry.key);
|
||||
}
|
||||
_effectiveState[entry.key] = entry.value;
|
||||
changed = true;
|
||||
}
|
||||
_contextSourceCreatedAt[entry.key] = event.createdAt;
|
||||
} else if (event.createdAt == sourceCreatedAt && entry.value != current) {
|
||||
_effectiveState[entry.key] = entry.value;
|
||||
final result = _applyRemoteContextTimestamp(
|
||||
contextId: entry.key,
|
||||
timestamp: entry.value,
|
||||
eventCreatedAt: event.createdAt,
|
||||
);
|
||||
if (result == _ApplyRemoteContextResult.advanced) {
|
||||
_pendingSyncedAdvances.add(entry.key);
|
||||
changed = true;
|
||||
}
|
||||
if (_publishableContextIds.add(entry.key)) {
|
||||
@@ -375,6 +347,27 @@ class ReadStateManager {
|
||||
}
|
||||
}
|
||||
|
||||
_ApplyRemoteContextResult _applyRemoteContextTimestamp({
|
||||
required String contextId,
|
||||
required int timestamp,
|
||||
required int eventCreatedAt,
|
||||
}) {
|
||||
final sourceCreatedAt = _contextSourceCreatedAt[contextId] ?? 0;
|
||||
final current = _effectiveState[contextId] ?? 0;
|
||||
final next = max(current, timestamp);
|
||||
final result = next == current
|
||||
? _ApplyRemoteContextResult.unchanged
|
||||
: _ApplyRemoteContextResult.advanced;
|
||||
|
||||
if (result == _ApplyRemoteContextResult.advanced) {
|
||||
_effectiveState[contextId] = next;
|
||||
}
|
||||
if (eventCreatedAt > sourceCreatedAt) {
|
||||
_contextSourceCreatedAt[contextId] = eventCreatedAt;
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
void _schedulePublish() {
|
||||
if (!_remoteEnabled || _remoteUnsupported || _disposed) return;
|
||||
|
||||
@@ -427,7 +420,6 @@ class ReadStateManager {
|
||||
}
|
||||
}
|
||||
_lastPublishedContexts = contexts;
|
||||
_forcedContextIds.clear();
|
||||
_maxFetchedCreatedAt = max(_maxFetchedCreatedAt, createdAt);
|
||||
_persistLocalState();
|
||||
} catch (error) {
|
||||
@@ -487,12 +479,6 @@ class ReadStateManager {
|
||||
return true;
|
||||
}
|
||||
|
||||
Set<String> drainSyncedRollbacks() {
|
||||
final drained = Set<String>.from(_pendingSyncedRollbacks);
|
||||
_pendingSyncedRollbacks.clear();
|
||||
return drained;
|
||||
}
|
||||
|
||||
Set<String> drainSyncedAdvances() {
|
||||
final drained = Set<String>.from(_pendingSyncedAdvances);
|
||||
_pendingSyncedAdvances.clear();
|
||||
@@ -517,7 +503,6 @@ class ReadStateManager {
|
||||
_publishableContextIds
|
||||
..clear()
|
||||
..addAll(stored.publishableContextIds);
|
||||
_forcedContextIds.clear();
|
||||
_contextSourceCreatedAt
|
||||
..clear()
|
||||
..addAll(stored.sourceCreatedAt);
|
||||
|
||||
@@ -13,14 +13,14 @@ class ReadStateState {
|
||||
final String? pubkey;
|
||||
final Map<String, int> contexts;
|
||||
final int version;
|
||||
final Set<String> syncedForcedChannelIds;
|
||||
final Set<String> locallyForcedChannelIds;
|
||||
|
||||
const ReadStateState({
|
||||
required this.isReady,
|
||||
required this.pubkey,
|
||||
required this.contexts,
|
||||
required this.version,
|
||||
this.syncedForcedChannelIds = const {},
|
||||
this.locallyForcedChannelIds = const {},
|
||||
});
|
||||
|
||||
const ReadStateState.inert()
|
||||
@@ -28,7 +28,7 @@ class ReadStateState {
|
||||
pubkey = null,
|
||||
contexts = const {},
|
||||
version = 0,
|
||||
syncedForcedChannelIds = const {};
|
||||
locallyForcedChannelIds = const {};
|
||||
|
||||
int? effectiveTimestamp(String contextId) => contexts[contextId];
|
||||
|
||||
@@ -43,7 +43,7 @@ class ReadStateState {
|
||||
pubkey: pubkey,
|
||||
contexts: Map.unmodifiable({...contexts, contextId: timestamp}),
|
||||
version: version + 1,
|
||||
syncedForcedChannelIds: syncedForcedChannelIds,
|
||||
locallyForcedChannelIds: locallyForcedChannelIds,
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -51,14 +51,14 @@ class ReadStateState {
|
||||
class ReadStateNotifier extends Notifier<ReadStateState> {
|
||||
ReadStateManager? _manager;
|
||||
bool _isInitialized = false;
|
||||
final Set<String> _syncedForcedChannelIds = {};
|
||||
final Set<String> _locallyForcedChannelIds = {};
|
||||
|
||||
@override
|
||||
ReadStateState build() {
|
||||
_manager?.dispose(flushPending: false);
|
||||
_manager = null;
|
||||
_isInitialized = false;
|
||||
_syncedForcedChannelIds.clear();
|
||||
_locallyForcedChannelIds.clear();
|
||||
|
||||
final relayConfig = ref.watch(relayConfigProvider);
|
||||
ref.watch(relaySessionProvider);
|
||||
@@ -131,12 +131,19 @@ class ReadStateNotifier extends Notifier<ReadStateState> {
|
||||
}
|
||||
|
||||
void markContextRead(String contextId, int unixTimestamp) {
|
||||
_syncedForcedChannelIds.remove(contextId);
|
||||
_locallyForcedChannelIds.remove(contextId);
|
||||
_manager?.markContextRead(contextId, unixTimestamp);
|
||||
}
|
||||
|
||||
void markContextUnread(String contextId, int lastMessageTimestamp) {
|
||||
_manager?.markContextUnread(contextId, lastMessageTimestamp);
|
||||
void markContextUnread(String contextId) {
|
||||
final manager = _manager;
|
||||
if (manager == null) return;
|
||||
_locallyForcedChannelIds.add(contextId);
|
||||
state = _stateFromManager(
|
||||
manager,
|
||||
isReady: _isInitialized,
|
||||
previousVersion: state.version,
|
||||
);
|
||||
}
|
||||
|
||||
void seedContextRead(String contextId, int unixTimestamp) {
|
||||
@@ -145,10 +152,8 @@ class ReadStateNotifier extends Notifier<ReadStateState> {
|
||||
|
||||
void _emitManagerState(ReadStateManager manager) {
|
||||
if (_manager != manager) return;
|
||||
final rollbacks = manager.drainSyncedRollbacks();
|
||||
final advances = manager.drainSyncedAdvances();
|
||||
_syncedForcedChannelIds.addAll(rollbacks);
|
||||
_syncedForcedChannelIds.removeAll(advances);
|
||||
_locallyForcedChannelIds.removeAll(advances);
|
||||
state = _stateFromManager(
|
||||
manager,
|
||||
isReady: _isInitialized,
|
||||
@@ -166,8 +171,8 @@ class ReadStateNotifier extends Notifier<ReadStateState> {
|
||||
pubkey: manager.pubkey,
|
||||
contexts: manager.effectiveContexts,
|
||||
version: (previousVersion ?? 0) + 1,
|
||||
syncedForcedChannelIds: Set.unmodifiable(
|
||||
Set<String>.from(_syncedForcedChannelIds),
|
||||
locallyForcedChannelIds: Set.unmodifiable(
|
||||
Set<String>.from(_locallyForcedChannelIds),
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
@@ -33,15 +33,15 @@ final unreadBadgeProvider = Provider<UnreadBadgeState>((ref) {
|
||||
for (final channel in channels) {
|
||||
if (!channel.isMember || channel.isArchived) continue;
|
||||
|
||||
final isSyncedForced = readState.syncedForcedChannelIds.contains(
|
||||
final isLocallyForced = readState.locallyForcedChannelIds.contains(
|
||||
channel.id,
|
||||
);
|
||||
final lastMessageAt = dateTimeToUnixSeconds(channel.lastMessageAt);
|
||||
if (lastMessageAt == null && !isSyncedForced) continue;
|
||||
if (lastMessageAt == null && !isLocallyForced) continue;
|
||||
|
||||
final readAt = readState.effectiveTimestamp(channel.id);
|
||||
final isUnread =
|
||||
isSyncedForced ||
|
||||
isLocallyForced ||
|
||||
readAt == null ||
|
||||
(lastMessageAt != null && lastMessageAt > readAt);
|
||||
if (!isUnread) continue;
|
||||
|
||||
@@ -332,6 +332,7 @@ class _FakeReadStateNotifier extends ReadStateNotifier {
|
||||
pubkey: state.pubkey,
|
||||
contexts: state.contexts,
|
||||
version: state.version + 1,
|
||||
locallyForcedChannelIds: state.locallyForcedChannelIds,
|
||||
);
|
||||
}
|
||||
|
||||
|
||||
@@ -1,8 +1,10 @@
|
||||
import 'dart:async';
|
||||
import 'dart:convert';
|
||||
|
||||
import 'package:flutter_test/flutter_test.dart';
|
||||
import 'package:nostr/nostr.dart' as nostr;
|
||||
import 'package:shared_preferences/shared_preferences.dart';
|
||||
import 'package:sprout_mobile/features/channels/read_state/read_state_format.dart';
|
||||
import 'package:sprout_mobile/features/channels/read_state/read_state_manager.dart';
|
||||
import 'package:sprout_mobile/shared/relay/relay.dart';
|
||||
|
||||
@@ -104,6 +106,49 @@ void main() {
|
||||
expect(manager.getEffectiveTimestamp('channel-2'), 43);
|
||||
},
|
||||
);
|
||||
|
||||
test('remote read-state rollback is ignored', () async {
|
||||
SharedPreferences.setMockInitialValues({});
|
||||
final prefs = await SharedPreferences.getInstance();
|
||||
final keychain = nostr.Keys.generate();
|
||||
final crypto = ReadStateCrypto.tryCreate(
|
||||
nsec: keychain.nsec,
|
||||
pubkey: keychain.public,
|
||||
);
|
||||
final relay = _FakeRelaySession();
|
||||
final manager = ReadStateManager(
|
||||
pubkey: keychain.public,
|
||||
prefs: prefs,
|
||||
crypto: crypto!,
|
||||
relaySession: relay,
|
||||
signedEventRelay: _FakeSignedEventRelay(),
|
||||
remoteEnabled: true,
|
||||
onChanged: () {},
|
||||
);
|
||||
|
||||
relay.historyEvents = [
|
||||
_readStateEvent(
|
||||
pubkey: keychain.public,
|
||||
crypto: crypto,
|
||||
clientId: 'remote-client',
|
||||
slotId: 'remote-slot',
|
||||
contexts: {'channel-1': 100},
|
||||
createdAt: 100,
|
||||
),
|
||||
_readStateEvent(
|
||||
pubkey: keychain.public,
|
||||
crypto: crypto,
|
||||
clientId: 'remote-client',
|
||||
slotId: 'remote-slot',
|
||||
contexts: {'channel-1': 50},
|
||||
createdAt: 110,
|
||||
),
|
||||
];
|
||||
|
||||
await manager.initialize();
|
||||
|
||||
expect(manager.getEffectiveTimestamp('channel-1'), 100);
|
||||
});
|
||||
}
|
||||
|
||||
class _SubmittedEvent {
|
||||
@@ -177,3 +222,43 @@ class _MissingScopeSignedEventRelay implements SignedEventRelay {
|
||||
throw Exception('missing users:write');
|
||||
}
|
||||
}
|
||||
|
||||
NostrEvent _readStateEvent({
|
||||
required String pubkey,
|
||||
required ReadStateCrypto crypto,
|
||||
required String clientId,
|
||||
required String slotId,
|
||||
required Map<String, int> contexts,
|
||||
required int createdAt,
|
||||
}) {
|
||||
final blob = ReadStateBlob(clientId: clientId, contexts: contexts);
|
||||
return NostrEvent(
|
||||
id: 'event-$clientId-$createdAt',
|
||||
pubkey: pubkey,
|
||||
createdAt: createdAt,
|
||||
kind: EventKind.readState,
|
||||
tags: [
|
||||
['d', '$readStateDTagPrefix$slotId'],
|
||||
['t', 'read-state'],
|
||||
],
|
||||
content: crypto.encrypt(jsonEncode(blob.toJson())),
|
||||
sig: 'sig',
|
||||
);
|
||||
}
|
||||
|
||||
class _FakeRelaySession extends RelaySessionNotifier {
|
||||
List<NostrEvent> historyEvents = [];
|
||||
|
||||
@override
|
||||
Future<List<NostrEvent>> fetchHistory(
|
||||
NostrFilter filter, {
|
||||
Duration timeout = const Duration(seconds: 8),
|
||||
}) async => historyEvents;
|
||||
|
||||
@override
|
||||
Future<void Function()> subscribe(
|
||||
NostrFilter filter,
|
||||
void Function(NostrEvent) onEvent, {
|
||||
void Function(String message)? onClosed,
|
||||
}) async => () {};
|
||||
}
|
||||
|
||||
@@ -62,6 +62,7 @@ void main() {
|
||||
ProviderContainer buildContainer({
|
||||
required List<Channel> channels,
|
||||
Map<String, int> readContexts = const {},
|
||||
Set<String> locallyForcedChannelIds = const {},
|
||||
bool readStateReady = true,
|
||||
Map<String, int> highPriorityMap = const {},
|
||||
}) {
|
||||
@@ -80,6 +81,7 @@ void main() {
|
||||
pubkey: 'me',
|
||||
contexts: readContexts,
|
||||
version: 1,
|
||||
locallyForcedChannelIds: locallyForcedChannelIds,
|
||||
),
|
||||
),
|
||||
),
|
||||
@@ -259,6 +261,23 @@ void main() {
|
||||
},
|
||||
);
|
||||
|
||||
test(
|
||||
'locally forced channel counts unread without publishing rollback',
|
||||
() async {
|
||||
final container = buildContainer(
|
||||
channels: [makeChannel(id: 'ch-a', lastMessageAtSeconds: t20)],
|
||||
readContexts: {'ch-a': t30},
|
||||
locallyForcedChannelIds: {'ch-a'},
|
||||
);
|
||||
addTearDown(container.dispose);
|
||||
|
||||
await container.read(channelsProvider.future);
|
||||
final badge = container.read(unreadBadgeProvider);
|
||||
expect(badge.highPriorityCount, 0);
|
||||
expect(badge.generalUnreadCount, 1);
|
||||
},
|
||||
);
|
||||
|
||||
test('channelsProvider in loading state returns (0, 0)', () {
|
||||
// The provider returns const UnreadBadgeState() while channels are loading.
|
||||
// We intentionally do NOT await the future here — the channels notifier
|
||||
|
||||
Reference in New Issue
Block a user