diff --git a/AGENTS.md b/AGENTS.md index c6d34ce7b7..2af3dd4867 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -88,6 +88,9 @@ For ad-hoc tool invocations, use `mise x -- ...` rather than assuming `go`, to authorize and assemble its public operations. Capture a durable boundary, wait for the serving projections through it, and fail the catch-up instead of publishing stale state at a newer cursor. +- Realtime transition metadata needed to update another projection must come + from the immutable signal, not depend on the triggering row still appearing + in a rebuilt current-state page; another client may already have removed it. - Treat projected authorization loss as a persistent privacy boundary. Purge every copied content-bearing or room-sensitive mirror, reject older async responses, and reopen the resource only after an explicit positive grant. diff --git a/apps/docs-website/src/content/docs/reference/connectrpc-api/realtime.mdx b/apps/docs-website/src/content/docs/reference/connectrpc-api/realtime.mdx index 142ed1f9f0..56ca4fbc42 100644 --- a/apps/docs-website/src/content/docs/reference/connectrpc-api/realtime.mdx +++ b/apps/docs-website/src/content/docs/reference/connectrpc-api/realtime.mdx @@ -250,6 +250,8 @@ enclosing replacement remains the canonical notification state. | `action` | [`RealtimeProjectionNotificationAction`](#chatto-realtime-v1-RealtimeProjectionNotificationAction) | No field description provided. | | `notification_id` | `string` | No field description provided. | | `silent` | `bool` | True when a created notification must not produce an alert. | +| `room_id` | `string` | Exact followed-thread target of a created reply or mention, when present. This remains available even if the notification was concurrently dismissed. | +| `thread_root_event_id` | `string` | No field description provided. | diff --git a/apps/docs-website/src/generated/connectrpc-api/realtime.raw.mdx b/apps/docs-website/src/generated/connectrpc-api/realtime.raw.mdx index 1d4daa8343..d3619c9aa6 100644 --- a/apps/docs-website/src/generated/connectrpc-api/realtime.raw.mdx +++ b/apps/docs-website/src/generated/connectrpc-api/realtime.raw.mdx @@ -279,6 +279,8 @@ enclosing replacement remains the canonical notification state. | `action` | [`RealtimeProjectionNotificationAction`](#chatto-realtime-v1-RealtimeProjectionNotificationAction) | No field description provided. | | `notification_id` | `string` | No field description provided. | | `silent` | `bool` | True when a created notification must not produce an alert. | +| `room_id` | `string` | Exact followed-thread target of a created reply or mention, when present. This remains available even if the notification was concurrently dismissed. | +| `thread_root_event_id` | `string` | No field description provided. | diff --git a/apps/frontend/src/lib/api-client-tests/viewer.spec.ts b/apps/frontend/src/lib/api-client-tests/viewer.spec.ts index 284f400fc6..3b0d44222a 100644 --- a/apps/frontend/src/lib/api-client-tests/viewer.spec.ts +++ b/apps/frontend/src/lib/api-client-tests/viewer.spec.ts @@ -136,6 +136,7 @@ describe('getCurrentUserViaConnect', () => { hasVerifiedEmail: true }, capabilities: { + hasUnreadFollowedThreads: true, grants: [ { capability: 'admin.view', granted: true }, { capability: 'dm.start', granted: true }, @@ -188,6 +189,7 @@ describe('getCurrentUserViaConnect', () => { canAdminViewSystem: true, canAdminViewAudit: true, canManageUserPermissions: true, + hasUnreadFollowedThreads: true, serverNotificationPreference: { level: NotificationLevel.AllMessages, effectiveLevel: NotificationLevel.AllMessages diff --git a/apps/frontend/src/lib/api-client/viewer.ts b/apps/frontend/src/lib/api-client/viewer.ts index dda9759091..342ff2b361 100644 --- a/apps/frontend/src/lib/api-client/viewer.ts +++ b/apps/frontend/src/lib/api-client/viewer.ts @@ -62,6 +62,7 @@ export type ViewerState = ViewerCapabilities & { roomNotificationPreferences: RoomNotificationPreference[]; viewerPermissions: Record; viewerHasUnreadRooms: boolean; + hasUnreadFollowedThreads: boolean; }; const capabilityKeys = { @@ -136,6 +137,7 @@ export function viewerResponseToState(response: GetViewerResponse): ViewerState canManageUserPermissions: can(capabilityKeys.manageUserPermissions), viewerPermissions, viewerHasUnreadRooms: response.viewerState?.hasUnreadRooms ?? false, + hasUnreadFollowedThreads: response.capabilities?.hasUnreadFollowedThreads ?? false, serverNotificationPreference: { level: apiNotificationLevel(response.serverNotificationPreference?.level), effectiveLevel: apiNotificationLevel(response.serverNotificationPreference?.effectiveLevel) diff --git a/apps/frontend/src/lib/components/chat/Chrome.svelte b/apps/frontend/src/lib/components/chat/Chrome.svelte index 6bfbbd186e..a621cf676a 100644 --- a/apps/frontend/src/lib/components/chat/Chrome.svelte +++ b/apps/frontend/src/lib/components/chat/Chrome.svelte @@ -242,7 +242,10 @@ {m['chat.overview.title']()} - +
diff --git a/apps/frontend/src/lib/components/chat/MyThreadsNavItem.svelte b/apps/frontend/src/lib/components/chat/MyThreadsNavItem.svelte index 50f58388cb..25f95004bb 100644 --- a/apps/frontend/src/lib/components/chat/MyThreadsNavItem.svelte +++ b/apps/frontend/src/lib/components/chat/MyThreadsNavItem.svelte @@ -2,18 +2,10 @@ import { resolve } from '$app/paths'; import { serverIdToSegment } from '$lib/navigation'; import { getActiveServer } from '$lib/state/activeServer.svelte'; - import { serverRegistry } from '$lib/state/server/registry.svelte'; - import { notificationTarget } from '$lib/state/server/notifications.svelte'; import UnreadDot from '$lib/ui/UnreadDot.svelte'; import * as m from '$lib/i18n/messages'; - let { active }: { active: boolean } = $props(); - - const notificationStore = serverRegistry.getStore(getActiveServer()).notifications; - - const hasUnread = $derived( - notificationStore.notifications.some((n) => notificationTarget(n).threadRootId !== null) - ); + let { active, hasUnread }: { active: boolean; hasUnread: boolean } = $props(); ({ + resolve: (path: string, params: Record) => + path.replace('[serverId]', params.serverId) +})); + +vi.mock('$lib/navigation', () => ({ + serverIdToSegment: (serverId: string) => serverId +})); + +vi.mock('$lib/state/activeServer.svelte', () => ({ + getActiveServer: () => 'server-1' +})); + +describe('MyThreadsNavItem', () => { + it('renders only the exact unread-followed-thread state', async () => { + const rendered = render(MyThreadsNavItem, { + // A pending notification for an unfollowed thread still supplies false. + props: { active: false, hasUnread: false } + }); + + expect(rendered.container.querySelector('[data-testid="my-threads-unread-dot"]')).toBeNull(); + + await rendered.rerender({ active: false, hasUnread: true }); + await expect + .element( + rendered.container.querySelector('[data-testid="my-threads-unread-dot"]') + ) + .toBeInTheDocument(); + + await rendered.rerender({ active: false, hasUnread: false }); + expect(rendered.container.querySelector('[data-testid="my-threads-unread-dot"]')).toBeNull(); + }); +}); diff --git a/apps/frontend/src/lib/state/server/projection.svelte.spec.ts b/apps/frontend/src/lib/state/server/projection.svelte.spec.ts index 687c413113..e64bfe89e2 100644 --- a/apps/frontend/src/lib/state/server/projection.svelte.spec.ts +++ b/apps/frontend/src/lib/state/server/projection.svelte.spec.ts @@ -20,17 +20,20 @@ import { RoomTimelineEvent, RoomTimelinePage } from '@chatto/api-types/api/v1/room_timeline_pb'; -import { Room } from '@chatto/api-types/api/v1/rooms_pb'; +import { Room, RoomSummary } from '@chatto/api-types/api/v1/rooms_pb'; import { User } from '@chatto/api-types/api/v1/users_pb'; import { ActiveCall, CallParticipant } from '@chatto/api-types/api/v1/voice_calls_pb'; import { ListNotificationsResponse, NotificationItem, + ReplyNotification, RoomNotificationCount } from '@chatto/api-types/api/v1/notifications_pb'; import { RealtimeProjectionEvent, RealtimeProjectionActiveCallsReplace, + RealtimeProjectionNotificationAction, + RealtimeProjectionNotificationChange, RealtimeProjectionOperation, RealtimeProjectionPresencesReplace, RealtimeProjectionThreadViewerState, @@ -124,6 +127,7 @@ describe('ServerProjectionStore', () => { expect(viewerState()?.isFollowing).toBe(true); expect(viewerState()?.hasUnread).toBe(true); expect(store.threadViewerStates.get('R1\u0000ROOT')?.hasUnread).toBe(true); + expect(store.hasThreadViewerStatesSnapshot).toBe(true); store.apply( event( @@ -136,6 +140,162 @@ describe('ServerProjectionStore', () => { expect(viewerState()?.isFollowing).toBe(false); expect(viewerState()?.hasUnread).toBe(false); expect(store.threadViewerStates.size).toBe(0); + expect(store.hasThreadViewerStatesSnapshot).toBe(true); + }); + + it('invalidates thread viewer-state snapshot authority on reset', () => { + const store = new ServerProjectionStore(); + store.apply( + event( + operation({ + case: 'threadViewerStatesReplace', + value: new RealtimeProjectionThreadViewerStatesReplace() + }), + operation({ case: 'reset', value: new RealtimeProjectionReset() }) + ) + ); + + expect(store.hasThreadViewerStatesSnapshot).toBe(false); + }); + + it('purges followed-thread state when room access is revoked or removed', () => { + const store = new ServerProjectionStore(); + const room = (isMember: boolean) => + operation({ + case: 'roomUpsert', + value: new RealtimeProjectionRoom({ + room: new RoomWithViewerState({ + room: new Room({ id: 'R1' }), + viewerState: new RoomViewerState({ isMember }) + }) + }) + }); + const unreadThread = operation({ + case: 'threadViewerStatesReplace', + value: new RealtimeProjectionThreadViewerStatesReplace({ + states: [ + new RealtimeProjectionThreadViewerState({ + roomId: 'R1', + threadRootEventId: 'ROOT', + viewerState: new ThreadViewerState({ isFollowing: true, hasUnread: true }) + }) + ] + }) + }); + + store.apply(event(room(true), unreadThread)); + expect(store.hasUnreadFollowedThreads()).toBe(true); + + store.apply( + event( + operation({ + case: 'roomViewerStateReplace', + value: new RealtimeProjectionRoomViewerStateReplace({ + roomId: 'R1', + viewerState: new RoomViewerState({ isMember: false }) + }) + }) + ) + ); + expect(store.threadViewerStates.size).toBe(0); + expect(store.hasUnreadFollowedThreads()).toBe(false); + + store.apply(event(room(true), unreadThread)); + expect(store.hasUnreadFollowedThreads()).toBe(true); + + store.apply( + event( + operation({ + case: 'roomRemove', + value: new RealtimeProjectionRoomRemove({ roomId: 'R1' }) + }) + ) + ); + expect(store.threadViewerStates.size).toBe(0); + expect(store.hasUnreadFollowedThreads()).toBe(false); + }); + + it('marks only an already-followed thread unread from a created reply notification', () => { + const store = new ServerProjectionStore(); + const room = operation({ + case: 'roomUpsert', + value: new RealtimeProjectionRoom({ + room: new RoomWithViewerState({ + room: new Room({ id: 'R1' }), + viewerState: new RoomViewerState({ isMember: true }) + }) + }) + }); + const threadStates = (isFollowing: boolean) => + operation({ + case: 'threadViewerStatesReplace', + value: new RealtimeProjectionThreadViewerStatesReplace({ + states: [ + new RealtimeProjectionThreadViewerState({ + roomId: 'R1', + threadRootEventId: 'ROOT', + viewerState: new ThreadViewerState({ isFollowing, hasUnread: false }) + }) + ] + }) + }); + const createdReply = operation({ + case: 'notificationsReplace', + value: new RealtimeProjectionNotificationsReplace({ + page: new ListNotificationsResponse({ + notifications: [ + new NotificationItem({ + id: 'N1', + kind: { + case: 'reply', + value: new ReplyNotification({ + room: new RoomSummary({ id: 'R1' }), + threadRootEventId: 'ROOT' + }) + } + }) + ] + }), + change: new RealtimeProjectionNotificationChange({ + action: RealtimeProjectionNotificationAction.CREATED, + notificationId: 'N1' + }) + }) + }); + const createdAfterConcurrentDismissal = operation({ + case: 'notificationsReplace', + value: new RealtimeProjectionNotificationsReplace({ + page: new ListNotificationsResponse(), + change: new RealtimeProjectionNotificationChange({ + action: RealtimeProjectionNotificationAction.CREATED, + notificationId: 'N1', + roomId: 'R1', + threadRootEventId: 'ROOT' + }) + }) + }); + + store.apply(event(room, threadStates(false), createdReply)); + expect(store.hasUnreadFollowedThreads()).toBe(false); + + store.apply(event(threadStates(true), createdAfterConcurrentDismissal)); + expect(store.hasUnreadFollowedThreads()).toBe(true); + + store.apply( + event( + operation({ + case: 'notificationsReplace', + value: new RealtimeProjectionNotificationsReplace({ + page: new ListNotificationsResponse(), + change: new RealtimeProjectionNotificationChange({ + action: RealtimeProjectionNotificationAction.DISMISSED, + notificationId: 'N1' + }) + }) + }) + ) + ); + expect(store.hasUnreadFollowedThreads()).toBe(true); }); it('reconciles complete transient presence without changing user profiles', () => { diff --git a/apps/frontend/src/lib/state/server/projection.svelte.ts b/apps/frontend/src/lib/state/server/projection.svelte.ts index 28e6ba1ce8..d44a30fd23 100644 --- a/apps/frontend/src/lib/state/server/projection.svelte.ts +++ b/apps/frontend/src/lib/state/server/projection.svelte.ts @@ -8,9 +8,13 @@ import type { ServerPublicProfile } from '@chatto/api-types/api/v1/server_pb'; import type { GetViewerResponse } from '@chatto/api-types/api/v1/viewer_pb'; import type { ListNotificationsResponse } from '@chatto/api-types/api/v1/notifications_pb'; import type { ActiveCall } from '@chatto/api-types/api/v1/voice_calls_pb'; -import { RealtimeProjectionRoom } from '@chatto/api-types/realtime/v1/realtime_pb'; +import { + RealtimeProjectionNotificationAction, + RealtimeProjectionRoom +} from '@chatto/api-types/realtime/v1/realtime_pb'; import type { RealtimeProjectionEvent, + RealtimeProjectionNotificationsReplace, RealtimeProjectionServerState } from '@chatto/api-types/realtime/v1/realtime_pb'; @@ -26,6 +30,8 @@ export class ServerProjectionStore { activeCalls = $state.raw([]); /** Complete current followed-thread viewer state, keyed by room and root ID. */ threadViewerStates = new SvelteMap(); + /** Whether the complete thread viewer-state replacement has been received. */ + hasThreadViewerStatesSnapshot = false; timelines = new SvelteMap(); private timelineEventCursors = new SvelteMap>(); private revokedRoomIds = new SvelteSet(); @@ -92,7 +98,9 @@ export class ServerProjectionStore { this.timelines.delete(roomId); this.timelineEventCursors.delete(roomId); this.removeActiveCallRoom(roomId); - } else if (room.room?.viewerState?.isMember === true) this.revokedRoomIds.delete(roomId); + this.removeThreadViewerStatesForRoom(roomId); + } else if (room.room?.viewerState?.isMember === true) + this.revokedRoomIds.delete(roomId); } break; } @@ -102,6 +110,7 @@ export class ServerProjectionStore { this.timelines.delete(operation.operation.value.roomId); this.timelineEventCursors.delete(operation.operation.value.roomId); this.removeActiveCallRoom(operation.operation.value.roomId); + this.removeThreadViewerStatesForRoom(operation.operation.value.roomId); break; case 'roomGroupsReplace': this.roomGroups = [...operation.operation.value.groups]; @@ -132,6 +141,7 @@ export class ServerProjectionStore { case 'notificationsReplace': { const replacement = operation.operation.value; this.notifications = replacement.page ?? null; + this.markCreatedNotificationThreadUnread(replacement); const counts = Object.fromEntries( replacement.roomCounts.map((count) => [count.roomId, count.totalCount]) ); @@ -170,6 +180,7 @@ export class ServerProjectionStore { this.timelines.delete(replacement.roomId); this.timelineEventCursors.delete(replacement.roomId); this.removeActiveCallRoom(replacement.roomId); + this.removeThreadViewerStatesForRoom(replacement.roomId); } else if (replacement.viewerState?.isMember === true) { this.revokedRoomIds.delete(replacement.roomId); } @@ -191,6 +202,7 @@ export class ServerProjectionStore { } break; case 'threadViewerStatesReplace': { + this.hasThreadViewerStatesSnapshot = true; this.threadViewerStates.clear(); for (const state of operation.operation.value.states) { this.threadViewerStates.set( @@ -268,11 +280,67 @@ export class ServerProjectionStore { this.notifications = null; this.activeCalls = []; this.threadViewerStates.clear(); + this.hasThreadViewerStatesSnapshot = false; this.timelines.clear(); this.timelineEventCursors.clear(); this.revokedRoomIds.clear(); } + /** Whether an included accessible room contains a followed thread with unread replies. */ + hasUnreadFollowedThreads(includeRoom: (roomId: string) => boolean = () => true): boolean { + for (const [key, state] of this.threadViewerStates) { + if (!state.isFollowing || !state.hasUnread) continue; + const roomId = key.slice(0, key.indexOf('\u0000')); + if (this.rooms.get(roomId)?.room?.viewerState?.isMember === true && includeRoom(roomId)) + return true; + } + return false; + } + + private markCreatedNotificationThreadUnread( + replacement: RealtimeProjectionNotificationsReplace + ): void { + const change = replacement.change; + if ( + !this.hasThreadViewerStatesSnapshot || + change?.action !== RealtimeProjectionNotificationAction.CREATED + ) + return; + let roomId = change.roomId; + let threadRootEventId = change.threadRootEventId; + if (!roomId || !threadRootEventId) { + const notification = replacement.page?.notifications.find( + (candidate) => candidate.id === change.notificationId + ); + if (!notification) return; + switch (notification.kind.case) { + case 'mention': + roomId = notification.kind.value.room?.id ?? ''; + threadRootEventId = notification.kind.value.threadRootEventId ?? ''; + break; + case 'reply': + roomId = notification.kind.value.room?.id ?? ''; + threadRootEventId = notification.kind.value.threadRootEventId ?? ''; + break; + } + } + if (!roomId || !threadRootEventId) return; + + const key = `${roomId}\u0000${threadRootEventId}`; + const current = this.threadViewerStates.get(key); + if (!current?.isFollowing) return; + const next = current.clone(); + next.hasUnread = true; + this.threadViewerStates.set(key, next); + } + + private removeThreadViewerStatesForRoom(roomId: string): void { + const prefix = `${roomId}\u0000`; + for (const key of this.threadViewerStates.keys()) { + if (key.startsWith(prefix)) this.threadViewerStates.delete(key); + } + } + /** * Purge every canonical copy of profile data for an account removed from the * server directory. Stable user IDs remain on historical facts, but no diff --git a/apps/frontend/src/lib/state/server/rooms.svelte.spec.ts b/apps/frontend/src/lib/state/server/rooms.svelte.spec.ts index 8ba95d5500..a48c9456d6 100644 --- a/apps/frontend/src/lib/state/server/rooms.svelte.spec.ts +++ b/apps/frontend/src/lib/state/server/rooms.svelte.spec.ts @@ -95,6 +95,7 @@ function makeViewer(overrides: Partial = {}): ViewerState { roomNotificationPreferences: [], viewerPermissions: {}, viewerHasUnreadRooms: false, + hasUnreadFollowedThreads: false, ...overrides }; } @@ -253,7 +254,8 @@ describe('RoomsStore - refresh', () => { level: NotificationLevel.AllMessages, effectiveLevel: NotificationLevel.AllMessages } - ] + ], + hasUnreadFollowedThreads: true }) ) }); @@ -261,6 +263,7 @@ describe('RoomsStore - refresh', () => { await store.refresh(); expect(store.currentUserId).toBe('U2'); + expect(store.hasUnreadFollowedThreads).toBe(true); expect(notificationLevels.getServerPreference()).toEqual({ level: NotificationLevel.Muted, effectiveLevel: NotificationLevel.Muted @@ -271,6 +274,41 @@ describe('RoomsStore - refresh', () => { }); }); + it('replaces followed-thread unread state from viewer projection state', () => { + const store = makeStore(); + + store.replaceProjection(makeViewer({ hasUnreadFollowedThreads: true }), [], []); + expect(store.hasUnreadFollowedThreads).toBe(true); + + store.replaceProjection(makeViewer({ hasUnreadFollowedThreads: false }), [], []); + expect(store.hasUnreadFollowedThreads).toBe(false); + }); + + it('does not let a delayed refresh overwrite newer realtime thread unread state', async () => { + let resolveViewer!: (viewer: ViewerState) => void; + const store = makeStore({ + viewerStateLoader: vi.fn( + () => new Promise((resolve) => (resolveViewer = resolve)) + ) + }); + + const refresh = store.refresh(); + store.setHasUnreadFollowedThreads(true); + resolveViewer(makeViewer({ hasUnreadFollowedThreads: false })); + await refresh; + + expect(store.hasUnreadFollowedThreads).toBe(true); + }); + + it('clears followed-thread unread state during projection reset', () => { + const store = makeStore(); + store.setHasUnreadFollowedThreads(true); + + store.resetProjectionState(); + + expect(store.hasUnreadFollowedThreads).toBe(false); + }); + it('discards out-of-order responses', async () => { let resolveFirstRooms!: (value: DirectoryRoomSummary[]) => void; let resolveSecondRooms!: (value: DirectoryRoomSummary[]) => void; diff --git a/apps/frontend/src/lib/state/server/rooms.svelte.ts b/apps/frontend/src/lib/state/server/rooms.svelte.ts index da00367a42..9a13c041ae 100644 --- a/apps/frontend/src/lib/state/server/rooms.svelte.ts +++ b/apps/frontend/src/lib/state/server/rooms.svelte.ts @@ -202,6 +202,7 @@ export class RoomsStore { rooms = $state([]); roomGroups = $state(null); isInitialLoading = $state(true); + hasUnreadFollowedThreads = $state(false); // The viewer's user ID, captured from the same sidebar bootstrap query that // produced DM `room.members`. Use this in preference to a global auth // context when filtering self out of DM labels and avatars. @@ -209,6 +210,7 @@ export class RoomsStore { private loadId = 0; private notificationCountsLoadId = 0; + private unreadFollowedThreadsRevision = 0; constructor( private readonly roomDirectoryAPI: RoomDirectoryAPI, @@ -226,6 +228,7 @@ export class RoomsStore { async refresh(): Promise { const thisLoad = ++this.loadId; const unreadSnapshotRevision = this.roomUnread.captureSnapshotRevision(); + const unreadFollowedThreadsRevision = this.unreadFollowedThreadsRevision; const [viewer, rooms, roomGroups] = await Promise.all([ this.viewerStateLoader(), this.roomDirectoryAPI.listRooms(RoomDirectoryScope.ALL), @@ -234,6 +237,9 @@ export class RoomsStore { if (this.loadId !== thisLoad) return; this.currentUserId = viewer.user.id; + if (this.unreadFollowedThreadsRevision === unreadFollowedThreadsRevision) { + this.setHasUnreadFollowedThreads(viewer.hasUnreadFollowedThreads); + } this.notificationLevels.setServerPreference( viewer.serverNotificationPreference.level, viewer.serverNotificationPreference.effectiveLevel @@ -299,6 +305,7 @@ export class RoomsStore { ): void { this.loadId++; this.currentUserId = viewer.user.id; + this.setHasUnreadFollowedThreads(viewer.hasUnreadFollowedThreads); this.notificationLevels.setServerPreference( viewer.serverNotificationPreference.level, viewer.serverNotificationPreference.effectiveLevel @@ -335,9 +342,21 @@ export class RoomsStore { this.rooms = []; this.roomGroups = []; this.currentUserId = null; + this.setHasUnreadFollowedThreads(false); this.isInitialLoading = true; } + /** Replace unread state and invalidate older asynchronous viewer snapshots. */ + setHasUnreadFollowedThreads(hasUnread: boolean): void { + this.unreadFollowedThreadsRevision++; + this.hasUnreadFollowedThreads = hasUnread; + } + + /** Whether room-level or inherited server preferences suppress unread UI. */ + isRoomMuted(roomId: string): boolean { + return this.notificationLevels.isRoomMuted(roomId); + } + private roomListItem(room: DirectoryRoomSummary, members: UserAvatarUserView[]): RoomsListItem { return { id: room.id, diff --git a/apps/frontend/src/lib/state/server/store.svelte.spec.ts b/apps/frontend/src/lib/state/server/store.svelte.spec.ts index 22b2afcfd6..f8a1372bf0 100644 --- a/apps/frontend/src/lib/state/server/store.svelte.spec.ts +++ b/apps/frontend/src/lib/state/server/store.svelte.spec.ts @@ -9,12 +9,14 @@ import { ServerRuntimeConfig } from '@chatto/api-types/api/v1/server_state_pb'; import { ActiveCall, CallParticipant } from '@chatto/api-types/api/v1/voice_calls_pb'; import { User } from '@chatto/api-types/api/v1/users_pb'; import { DirectoryMember } from '@chatto/api-types/api/v1/member_directory_pb'; -import { Message, MessageAttachment } from '@chatto/api-types/api/v1/message_types_pb'; -import { Room } from '@chatto/api-types/api/v1/rooms_pb'; import { - RoomViewerState, - RoomWithViewerState -} from '@chatto/api-types/api/v1/room_directory_pb'; + Message, + MessageAttachment, + ThreadViewerState +} from '@chatto/api-types/api/v1/message_types_pb'; +import { ListNotificationsResponse } from '@chatto/api-types/api/v1/notifications_pb'; +import { Room } from '@chatto/api-types/api/v1/rooms_pb'; +import { RoomViewerState, RoomWithViewerState } from '@chatto/api-types/api/v1/room_directory_pb'; import { RoomMessagePosted, RoomTimelineEvent, @@ -23,14 +25,18 @@ import { import { RealtimeProjectionEvent, RealtimeProjectionActiveCallsReplace, + RealtimeProjectionNotificationsReplace, RealtimeProjectionOperation, RealtimeProjectionRoomActivity, + RealtimeProjectionRoomRemove, RealtimeProjectionRoomViewerStateReplace, RealtimeProjectionReactionChange, RealtimeProjectionRoomTimelineEventRemove, RealtimeProjectionRoomTimelineEventUpsert, RealtimeProjectionRoomTimelineReplace, RealtimeProjectionServerState, + RealtimeProjectionThreadViewerState, + RealtimeProjectionThreadViewerStatesReplace, RealtimeProjectionReset, RealtimeProjectionRoom, RealtimeProjectionUserRemove @@ -138,6 +144,7 @@ const { soundMocks, apiMocks } = vi.hoisted(() => ({ canAdminViewSystem: false, canAdminViewAudit: false, canManageUserPermissions: false, + hasUnreadFollowedThreads: false, serverNotificationPreference: { level: 'DEFAULT', effectiveLevel: 'NORMAL' @@ -469,6 +476,7 @@ beforeEach(() => { canAdminViewSystem: false, canAdminViewAudit: false, canManageUserPermissions: false, + hasUnreadFollowedThreads: false, serverNotificationPreference: { level: 'DEFAULT', effectiveLevel: 'NORMAL' @@ -622,6 +630,7 @@ describe('ServerStateStore live server updates', () => { ); store.rooms.rooms = [{ id: 'R1' } as never]; store.rooms.roomGroups = [{ id: 'G1' } as never]; + store.rooms.hasUnreadFollowedThreads = true; store.rooms.isInitialLoading = false; store.roomDirectory.allRooms = [{ id: 'R1' } as never]; store.roomDirectory.isLoading = false; @@ -652,6 +661,7 @@ describe('ServerStateStore live server updates', () => { expect(store.serverInfo.livekitUrl).toBeNull(); expect(store.rooms.rooms).toEqual([]); expect(store.rooms.roomGroups).toEqual([]); + expect(store.rooms.hasUnreadFollowedThreads).toBe(false); expect(store.rooms.isInitialLoading).toBe(true); expect(store.roomDirectory.allRooms).toEqual([]); expect(store.roomDirectory.isLoading).toBe(true); @@ -687,6 +697,122 @@ describe('ServerStateStore live server updates', () => { expect(store.activeCallRooms.has('R2')).toBe(true); }); + it('derives My Threads unread state from complete projection replacements', () => { + const fake = new FakeServerConnection([]); + const store = makeStore(fake); + eventBusManager.startBus(registered.id, fake as unknown as ServerConnection); + flushSync(); + const bus = eventBusManager.getBus(registered.id)!; + const dispatch = (...operations: RealtimeProjectionOperation[]) => { + for (const handler of bus.projectionHandlers) { + handler(new RealtimeProjectionEvent({ operations })); + } + }; + const threadStates = (isFollowing: boolean, hasUnread: boolean) => + new RealtimeProjectionOperation({ + operation: { + case: 'threadViewerStatesReplace', + value: new RealtimeProjectionThreadViewerStatesReplace({ + states: [ + new RealtimeProjectionThreadViewerState({ + roomId: 'R1', + threadRootEventId: 'ROOT', + viewerState: new ThreadViewerState({ isFollowing, hasUnread }) + }) + ] + }) + } + }); + const roomState = (isMember: boolean) => + new RealtimeProjectionOperation({ + operation: { + case: 'roomUpsert', + value: new RealtimeProjectionRoom({ + room: new RoomWithViewerState({ + room: new Room({ id: 'R1' }), + viewerState: new RoomViewerState({ isMember }) + }) + }) + } + }); + + // Initial projection: an unread unfollowed thread is outside My Threads. + dispatch(roomState(true), threadStates(false, true)); + expect(store.rooms.hasUnreadFollowedThreads).toBe(false); + + // Follow, read-marker, and unfollow replacements are exact latest-value state. + dispatch(roomState(true), threadStates(true, true)); + expect(store.rooms.hasUnreadFollowedThreads).toBe(true); + + // Dismissing its notification does not make the followed thread read. + dispatch( + new RealtimeProjectionOperation({ + operation: { + case: 'notificationsReplace', + value: new RealtimeProjectionNotificationsReplace({ + page: new ListNotificationsResponse() + }) + } + }) + ); + expect(store.rooms.hasUnreadFollowedThreads).toBe(true); + + dispatch(threadStates(true, false)); + expect(store.rooms.hasUnreadFollowedThreads).toBe(false); + + dispatch(threadStates(false, true)); + expect(store.rooms.hasUnreadFollowedThreads).toBe(false); + + // A reconnect reset clears stale state; catch-up establishes it again. + dispatch(threadStates(true, true)); + expect(store.rooms.hasUnreadFollowedThreads).toBe(true); + + dispatch( + new RealtimeProjectionOperation({ + operation: { case: 'reset', value: new RealtimeProjectionReset() } + }) + ); + expect(store.rooms.hasUnreadFollowedThreads).toBe(false); + + dispatch(roomState(true), threadStates(true, true)); + expect(store.rooms.hasUnreadFollowedThreads).toBe(true); + + // Muted rooms keep their thread state but do not contribute sidebar unread. + store.notificationLevels.setRoomPreference('R1', 'MUTED' as never, 'MUTED' as never); + dispatch(threadStates(true, true)); + expect(store.rooms.hasUnreadFollowedThreads).toBe(false); + store.notificationLevels.setRoomPreference('R1', 'NORMAL' as never, 'NORMAL' as never); + dispatch(threadStates(true, true)); + expect(store.rooms.hasUnreadFollowedThreads).toBe(true); + + // Access loss clears cached state even before a complete replacement arrives. + dispatch( + new RealtimeProjectionOperation({ + operation: { + case: 'roomViewerStateReplace', + value: new RealtimeProjectionRoomViewerStateReplace({ + roomId: 'R1', + viewerState: new RoomViewerState({ isMember: false }) + }) + } + }) + ); + expect(store.rooms.hasUnreadFollowedThreads).toBe(false); + + dispatch(roomState(true), threadStates(true, true)); + expect(store.rooms.hasUnreadFollowedThreads).toBe(true); + + dispatch( + new RealtimeProjectionOperation({ + operation: { + case: 'roomRemove', + value: new RealtimeProjectionRoomRemove({ roomId: 'R1' }) + } + }) + ); + expect(store.rooms.hasUnreadFollowedThreads).toBe(false); + }); + it('purges removed users from navigation and retained render stores', () => { const fake = new FakeServerConnection([]); const store = makeStore(fake); diff --git a/apps/frontend/src/lib/state/server/store.svelte.ts b/apps/frontend/src/lib/state/server/store.svelte.ts index 4227f41520..e5ef13753f 100644 --- a/apps/frontend/src/lib/state/server/store.svelte.ts +++ b/apps/frontend/src/lib/state/server/store.svelte.ts @@ -256,6 +256,7 @@ export class ServerStateStore { delete this.#threadMessageRefCounts[key]; } } + this.synchronizeUnreadFollowedThreads(); } /** Reacquire only mounted stores that were previously scrubbed for access loss. */ @@ -456,6 +457,7 @@ export class ServerStateStore { break; } case 'threadViewerStatesReplace': { + this.synchronizeUnreadFollowedThreads(); for (const [roomId, page] of this.projection.timelines) { for (const projectedEvent of page.events) { if ( @@ -551,9 +553,17 @@ export class ServerStateStore { notificationCountsByRoomId, messageHistoryByRoomId ); + this.synchronizeUnreadFollowedThreads(); this.roomDirectory.replaceProjection(rooms); } + private synchronizeUnreadFollowedThreads(): void { + if (!this.projection.hasThreadViewerStatesSnapshot) return; + this.rooms.setHasUnreadFollowedThreads( + this.projection.hasUnreadFollowedThreads((roomId) => !this.rooms.isRoomMuted(roomId)) + ); + } + /** Clear every mirror whose authority was invalidated by a reset frame. */ private resetProjectionMirrors(): void { clearUserSummaryCache(this.serverId); diff --git a/cli/internal/core/threads.go b/cli/internal/core/threads.go index 9f9f6dd7ee..ba4a0a4d99 100644 --- a/cli/internal/core/threads.go +++ b/cli/internal/core/threads.go @@ -933,6 +933,14 @@ func (c *ChattoCore) HasUnreadFollowedThreads(ctx context.Context, userID string continue } for _, thread := range threads { + level, err := c.GetEffectiveNotificationLevel(ctx, userID, thread.RoomID) + if err != nil { + c.logger.Warn("Failed to resolve followed thread notification level", "room_id", thread.RoomID, "error", err) + continue + } + if level == corev1.NotificationLevel_NOTIFICATION_LEVEL_MUTED { + continue + } if c.followedThreadHasUnread(ctx, userID, thread) { return true, nil } diff --git a/cli/internal/core/threads_test.go b/cli/internal/core/threads_test.go index 0dc07f42f1..1f17fe1c08 100644 --- a/cli/internal/core/threads_test.go +++ b/cli/internal/core/threads_test.go @@ -1954,6 +1954,85 @@ func TestChattoCore_SetThreadLastReadEventIDDoesNotRegress(t *testing.T) { } } +func TestChattoCore_HasUnreadFollowedThreadsExcludesMutedRooms(t *testing.T) { + chattoCore, _ := setupTestCore(t) + ctx := testContext(t) + + owner, err := chattoCore.CreateUser(ctx, SystemActorID, "unread-thread-owner", "Unread Thread Owner", "password123") + if err != nil { + t.Fatalf("CreateUser owner: %v", err) + } + viewer, err := chattoCore.CreateUser(ctx, SystemActorID, "unread-thread-viewer", "Unread Thread Viewer", "password123") + if err != nil { + t.Fatalf("CreateUser viewer: %v", err) + } + author, err := chattoCore.CreateUser(ctx, SystemActorID, "unread-thread-author", "Unread Thread Author", "password123") + if err != nil { + t.Fatalf("CreateUser author: %v", err) + } + room, err := chattoCore.CreateRoom(ctx, owner.Id, KindChannel, "", "unread-thread-room", "") + if err != nil { + t.Fatalf("CreateRoom: %v", err) + } + for _, userID := range []string{owner.Id, viewer.Id, author.Id} { + if _, err := chattoCore.JoinRoom(ctx, userID, KindChannel, userID, room.Id); err != nil { + t.Fatalf("JoinRoom %q: %v", userID, err) + } + } + root, err := chattoCore.PostMessage(ctx, KindChannel, room.Id, owner.Id, "thread root", nil, "", "", nil, false) + if err != nil { + t.Fatalf("PostMessage root: %v", err) + } + if _, err := chattoCore.PostMessage(ctx, KindChannel, room.Id, author.Id, "first reply", nil, root.Id, "", nil, false); err != nil { + t.Fatalf("PostMessage first reply: %v", err) + } + if err := chattoCore.FollowThread(ctx, KindChannel, viewer.Id, room.Id, root.Id); err != nil { + t.Fatalf("FollowThread: %v", err) + } + time.Sleep(time.Millisecond) + if _, err := chattoCore.PostMessage(ctx, KindChannel, room.Id, author.Id, "unread reply", nil, root.Id, "", nil, false); err != nil { + t.Fatalf("PostMessage reply: %v", err) + } + if following, err := chattoCore.IsFollowingThread(ctx, KindChannel, viewer.Id, room.Id, root.Id); err != nil || !following { + t.Fatalf("IsFollowingThread = %v, %v", following, err) + } + if _, err := chattoCore.GetThreadMetadata(ctx, KindChannel, room.Id, root.Id); err != nil { + t.Fatalf("GetThreadMetadata: %v", err) + } + if member, err := chattoCore.RoomMembershipExists(ctx, KindChannel, viewer.Id, room.Id); err != nil || !member { + t.Fatalf("RoomMembershipExists = %v, %v", member, err) + } + + hasUnread, err := chattoCore.HasUnreadFollowedThreads(ctx, viewer.Id, []string{LegacyServerSpaceID}) + if err != nil { + t.Fatalf("HasUnreadFollowedThreads before mute: %v", err) + } + if !hasUnread { + threads, listErr := chattoCore.ListFollowedThreads(ctx, viewer.Id, []string{LegacyServerSpaceID}) + t.Fatalf("unmuted followed thread should be unread; threads=%+v listErr=%v", threads, listErr) + } + if err := chattoCore.SetRoomNotificationLevel(ctx, viewer.Id, room.Id, corev1.NotificationLevel_NOTIFICATION_LEVEL_MUTED); err != nil { + t.Fatalf("SetRoomNotificationLevel muted: %v", err) + } + hasUnread, err = chattoCore.HasUnreadFollowedThreads(ctx, viewer.Id, []string{LegacyServerSpaceID}) + if err != nil { + t.Fatalf("HasUnreadFollowedThreads while muted: %v", err) + } + if hasUnread { + t.Fatal("muted followed thread should not contribute unread state") + } + if err := chattoCore.SetRoomNotificationLevel(ctx, viewer.Id, room.Id, corev1.NotificationLevel_NOTIFICATION_LEVEL_UNSPECIFIED); err != nil { + t.Fatalf("SetRoomNotificationLevel unmuted: %v", err) + } + hasUnread, err = chattoCore.HasUnreadFollowedThreads(ctx, viewer.Id, []string{LegacyServerSpaceID}) + if err != nil { + t.Fatalf("HasUnreadFollowedThreads after unmute: %v", err) + } + if !hasUnread { + t.Fatal("unmuting should reveal the unread followed thread") + } +} + func eventIDsForTest(events []*RoomEvent) []string { ids := make([]string, 0, len(events)) for _, event := range events { diff --git a/cli/internal/http_server/realtime_projection.go b/cli/internal/http_server/realtime_projection.go index 3021f6daa5..cdd47dcbe5 100644 --- a/cli/internal/http_server/realtime_projection.go +++ b/cli/internal/http_server/realtime_projection.go @@ -204,6 +204,16 @@ func (s *HTTPServer) realtimeProjectionFrameForEventWithRooms(ctx context.Contex _, ok := retainedRooms[roomID] return ok } + appendThreadViewerStates := func() error { + threadStates, err := s.connectAPI.BuildRealtimeProjectionThreadViewerStates(ctx, viewerID) + if err != nil { + return err + } + appendOperation(&realtimev1.RealtimeProjectionOperation{Operation: &realtimev1.RealtimeProjectionOperation_ThreadViewerStatesReplace{ + ThreadViewerStatesReplace: realtimeProjectionThreadViewerStates(threadStates), + }}) + return nil + } if evt == nil { live := event.LiveEvent() if live == nil { @@ -251,33 +261,32 @@ func (s *HTTPServer) realtimeProjectionFrameForEventWithRooms(ctx context.Contex return nil, false, err } appendOperation(&realtimev1.RealtimeProjectionOperation{Operation: &realtimev1.RealtimeProjectionOperation_ViewerUpsert{ViewerUpsert: viewer}}) + // Muting hides thread unread state while unmuting may reveal replies + // accumulated during the mute. Preference changes are rare and + // user-scoped, so replace the complete set here rather than on each + // room reply. + if err := appendThreadViewerStates(); err != nil { + return nil, false, err + } case *corev1.LiveEvent_NotificationCreated: notifications, err := s.connectAPI.BuildRealtimeProjectionNotifications(ctx, viewerID) if err != nil { return nil, false, err } replacement := realtimeProjectionNotifications(notifications) - replacement.Change = &realtimev1.RealtimeProjectionNotificationChange{ + change := &realtimev1.RealtimeProjectionNotificationChange{ Action: realtimev1.RealtimeProjectionNotificationAction_REALTIME_PROJECTION_NOTIFICATION_ACTION_CREATED, NotificationId: payload.NotificationCreated.GetNotificationId(), Silent: payload.NotificationCreated.GetSilent(), } + if roomID, threadRootID := s.realtimeNotificationThreadTarget(ctx, payload.NotificationCreated); threadRootID != "" { + change.RoomId = roomID + change.ThreadRootEventId = threadRootID + } + replacement.Change = change appendOperation(&realtimev1.RealtimeProjectionOperation{Operation: &realtimev1.RealtimeProjectionOperation_NotificationsReplace{ NotificationsReplace: replacement, }}) - // Reply notifications are also the live signal that a followed - // thread became unread. Replace the complete latest-value set so - // unretained rooms and the My Threads view converge without a - // ConnectRPC refresh. - if payload.NotificationCreated.GetInReplyToId() != "" { - threadStates, err := s.connectAPI.BuildRealtimeProjectionThreadViewerStates(ctx, viewerID) - if err != nil { - return nil, false, err - } - appendOperation(&realtimev1.RealtimeProjectionOperation{Operation: &realtimev1.RealtimeProjectionOperation_ThreadViewerStatesReplace{ - ThreadViewerStatesReplace: realtimeProjectionThreadViewerStates(threadStates), - }}) - } case *corev1.LiveEvent_NotificationDismissed: notifications, err := s.connectAPI.BuildRealtimeProjectionNotifications(ctx, viewerID) if err != nil { @@ -311,13 +320,9 @@ func (s *HTTPServer) realtimeProjectionFrameForEventWithRooms(ctx context.Contex }}) case *corev1.LiveEvent_ThreadFollowChanged: thread := payload.ThreadFollowChanged - threadStates, err := s.connectAPI.BuildRealtimeProjectionThreadViewerStates(ctx, viewerID) - if err != nil { + if err := appendThreadViewerStates(); err != nil { return nil, false, err } - appendOperation(&realtimev1.RealtimeProjectionOperation{Operation: &realtimev1.RealtimeProjectionOperation_ThreadViewerStatesReplace{ - ThreadViewerStatesReplace: realtimeProjectionThreadViewerStates(threadStates), - }}) if retainsTimeline(thread.GetRoomId()) { timelineEvent, includes, eventCursor, err := s.connectAPI.BuildRealtimeProjectionTimelineEvent(ctx, viewerID, thread.GetRoomId(), thread.GetThreadRootEventId()) if err != nil { @@ -487,7 +492,8 @@ func (s *HTTPServer) realtimeProjectionFrameForEventWithRooms(ctx context.Contex if err := appendRoomViewerState(roomID); err != nil { return nil, false, err } - if payload.MessagePosted.GetInThread() == "" { + threadRootID := payload.MessagePosted.GetInThread() + if threadRootID == "" { appendOperation(&realtimev1.RealtimeProjectionOperation{Operation: &realtimev1.RealtimeProjectionOperation_RoomActivity{ RoomActivity: &realtimev1.RealtimeProjectionRoomActivity{RoomId: roomID}, }}) @@ -498,8 +504,8 @@ func (s *HTTPServer) realtimeProjectionFrameForEventWithRooms(ctx context.Contex // Deliver the reply before the authoritative root summary. Existing // reducers optimistically increment a root when ingesting a reply; the // following root upsert then converges that count instead of doubling it. - if rootID := payload.MessagePosted.GetInThread(); rootID != "" { - if err := appendTimeline(roomID, rootID, nil); err != nil { + if threadRootID != "" { + if err := appendTimeline(roomID, threadRootID, nil); err != nil { return nil, false, err } } @@ -616,7 +622,8 @@ func (s *HTTPServer) realtimeProjectionFrameForEventWithRooms(ctx context.Contex }}) case *corev1.Event_RoomUnarchived: roomID := payload.RoomUnarchived.GetRoomId() - if err := appendRoom(roomID); err != nil { + room, err := appendRoomResult(roomID) + if err != nil { return nil, false, err } if err := appendRoomTimelineIfMember(roomID); err != nil { @@ -625,6 +632,11 @@ func (s *HTTPServer) realtimeProjectionFrameForEventWithRooms(ctx context.Contex if err := appendSourceTimeline(roomID); err != nil { return nil, false, err } + if room != nil && room.Room.GetViewerState().GetIsMember() { + if err := appendThreadViewerStates(); err != nil { + return nil, false, fmt.Errorf("assemble thread viewer states after room unarchive: %w", err) + } + } case *corev1.Event_RoomUniversalChanged: roomID := payload.RoomUniversalChanged.GetRoomId() room, err := appendRoomResult(roomID) @@ -642,6 +654,9 @@ func (s *HTTPServer) realtimeProjectionFrameForEventWithRooms(ctx context.Contex if err := appendRoomTimeline(roomID); err != nil { return nil, false, err } + if err := appendThreadViewerStates(); err != nil { + return nil, false, fmt.Errorf("assemble thread viewer states after universal room access gain: %w", err) + } } else { // A universal-membership revocation must remove already-decrypted // timeline state in the same ordered projection event as metadata. @@ -659,6 +674,9 @@ func (s *HTTPServer) realtimeProjectionFrameForEventWithRooms(ctx context.Contex if err := appendRoomTimeline(roomID); err != nil { return nil, false, err } + if err := appendThreadViewerStates(); err != nil { + return nil, false, fmt.Errorf("assemble thread viewer states after room join: %w", err) + } } if err := appendSourceTimeline(roomID); err != nil { return nil, false, err @@ -686,6 +704,9 @@ func (s *HTTPServer) realtimeProjectionFrameForEventWithRooms(ctx context.Contex if err := appendRoomTimeline(roomID); err != nil { return nil, false, err } + if err := appendThreadViewerStates(); err != nil { + return nil, false, fmt.Errorf("assemble thread viewer states after room membership add: %w", err) + } } case *corev1.Event_RoomMemberRemoved: roomID := payload.RoomMemberRemoved.GetRoomId() @@ -779,6 +800,24 @@ func realtimeProjectionRoom(room *connectapi.RealtimeProjectionRoom) *realtimev1 } } +// realtimeNotificationThreadTarget resolves the immutable message target from +// the creation signal, rather than relying on the notification still being +// pending when a slower socket maps the signal. +func (s *HTTPServer) realtimeNotificationThreadTarget(ctx context.Context, created *corev1.NotificationCreatedEvent) (string, string) { + if created == nil || created.GetRoomId() == "" || created.GetEventId() == "" { + return "", "" + } + event, err := s.core.GetRoomEventByEventID(ctx, core.KindChannel, created.GetRoomId(), created.GetEventId()) + if err != nil || event == nil { + return "", "" + } + threadRootID := event.GetMessagePosted().GetInThread() + if threadRootID == "" { + return "", "" + } + return created.GetRoomId(), threadRootID +} + func realtimeProjectionNotifications(notifications *connectapi.RealtimeProjectionNotifications) *realtimev1.RealtimeProjectionNotificationsReplace { if notifications == nil { return &realtimev1.RealtimeProjectionNotificationsReplace{} diff --git a/cli/internal/http_server/realtime_test.go b/cli/internal/http_server/realtime_test.go index 514e000d90..53b659fc1d 100644 --- a/cli/internal/http_server/realtime_test.go +++ b/cli/internal/http_server/realtime_test.go @@ -1367,12 +1367,25 @@ func TestRealtimeWebSocketAdvancesPastRetainedUnarchiveForNonMember(t *testing.T if err := env.core.LeaveRoom(env.ctx, viewer.Id, core.KindChannel, viewer.Id, room.Id); err != nil { t.Fatalf("LeaveRoom: %v", err) } - left := waitRealtimeRoomUpsert(t, conn, 5*time.Second, func(upsert *realtimev1.RealtimeProjectionRoom) bool { - return upsert.GetRoom().GetRoom().GetId() == room.Id && !upsert.GetRoom().GetViewerState().GetIsMember() + left := waitRealtimeProjectionEvent(t, conn, 5*time.Second, func(projection *realtimev1.RealtimeProjectionEvent) bool { + for _, operation := range projection.GetOperations() { + upsert := operation.GetRoomUpsert() + if upsert.GetRoom().GetRoom().GetId() == room.Id && !upsert.GetRoom().GetViewerState().GetIsMember() { + return true + } + } + return false }) if left == nil { t.Fatal("viewer did not receive non-member room state after leaving") } + var replacedThreadStates bool + for _, operation := range left.GetOperations() { + replacedThreadStates = replacedThreadStates || operation.GetThreadViewerStatesReplace() != nil + } + if replacedThreadStates { + t.Fatal("room access loss performed an unrelated exhaustive thread-state replacement") + } if _, err := env.core.ArchiveRoom(env.ctx, owner.Id, core.KindChannel, room.Id); err != nil { t.Fatalf("ArchiveRoom: %v", err) } @@ -1404,6 +1417,106 @@ func TestRealtimeWebSocketAdvancesPastRetainedUnarchiveForNonMember(t *testing.T } } +func TestRealtimeWebSocketRestoresFollowedThreadStateAfterRoomAccessGain(t *testing.T) { + env := setupWebSocketTestServer(t) + owner, err := env.core.CreateUser(env.ctx, core.SystemActorID, "rt-thread-restore-owner", "RT Thread Restore Owner", "password123") + if err != nil { + t.Fatalf("CreateUser owner: %v", err) + } + viewer, err := env.core.CreateUser(env.ctx, core.SystemActorID, "rt-thread-restore-viewer", "RT Thread Restore Viewer", "password123") + if err != nil { + t.Fatalf("CreateUser viewer: %v", err) + } + author, err := env.core.CreateUser(env.ctx, core.SystemActorID, "rt-thread-restore-author", "RT Thread Restore Author", "password123") + if err != nil { + t.Fatalf("CreateUser author: %v", err) + } + room, err := env.core.CreateRoom(env.ctx, owner.Id, core.KindChannel, "", "rt-thread-restore-room", "") + if err != nil { + t.Fatalf("CreateRoom: %v", err) + } + for _, userID := range []string{owner.Id, viewer.Id, author.Id} { + if _, err := env.core.JoinRoom(env.ctx, userID, core.KindChannel, userID, room.Id); err != nil { + t.Fatalf("JoinRoom %q: %v", userID, err) + } + } + root, err := env.core.PostMessage(env.ctx, core.KindChannel, room.Id, owner.Id, "thread restore root", nil, "", "", nil, false) + if err != nil { + t.Fatalf("PostMessage root: %v", err) + } + if err := env.core.FollowThread(env.ctx, core.KindChannel, viewer.Id, room.Id, root.Id); err != nil { + t.Fatalf("FollowThread: %v", err) + } + if _, err := env.core.PostMessage(env.ctx, core.KindChannel, room.Id, author.Id, "thread restore unread reply", nil, root.Id, "", nil, false); err != nil { + t.Fatalf("PostMessage reply: %v", err) + } + token, err := env.core.CreateAuthToken(env.ctx, viewer.Id) + if err != nil { + t.Fatalf("CreateAuthToken: %v", err) + } + conn := env.connectRealtime(t) + subscribeRealtime(t, conn, token) + + waitForLoss := func() { + t.Helper() + if projection := waitRealtimeProjectionEvent(t, conn, 5*time.Second, func(projection *realtimev1.RealtimeProjectionEvent) bool { + for _, operation := range projection.GetOperations() { + projectedRoom := operation.GetRoomUpsert().GetRoom() + if projectedRoom.GetRoom().GetId() == room.Id && !projectedRoom.GetViewerState().GetIsMember() { + return true + } + } + return false + }); projection == nil { + t.Fatal("room access loss did not reach viewer") + } + } + waitForRestoration := func(name string) { + t.Helper() + projection := waitRealtimeProjectionEvent(t, conn, 5*time.Second, func(projection *realtimev1.RealtimeProjectionEvent) bool { + var restoredRoom, restoredThread bool + for _, operation := range projection.GetOperations() { + projectedRoom := operation.GetRoomUpsert().GetRoom() + restoredRoom = restoredRoom || (projectedRoom.GetRoom().GetId() == room.Id && projectedRoom.GetViewerState().GetIsMember()) + for _, state := range operation.GetThreadViewerStatesReplace().GetStates() { + restoredThread = restoredThread || (state.GetRoomId() == room.Id && state.GetThreadRootEventId() == root.Id && state.GetViewerState().GetIsFollowing() && state.GetViewerState().GetHasUnread()) + } + } + return restoredRoom && restoredThread + }) + if projection == nil { + t.Fatalf("%s did not restore followed-thread viewer state", name) + } + } + + if err := env.core.LeaveRoom(env.ctx, viewer.Id, core.KindChannel, viewer.Id, room.Id); err != nil { + t.Fatalf("LeaveRoom before self join: %v", err) + } + waitForLoss() + if _, err := env.core.JoinRoom(env.ctx, viewer.Id, core.KindChannel, viewer.Id, room.Id); err != nil { + t.Fatalf("self JoinRoom: %v", err) + } + waitForRestoration("self join") + + if err := env.core.LeaveRoom(env.ctx, viewer.Id, core.KindChannel, viewer.Id, room.Id); err != nil { + t.Fatalf("LeaveRoom before member add: %v", err) + } + waitForLoss() + if _, err := env.core.AddMember(env.ctx, owner.Id, core.KindChannel, room.Id, viewer.Id); err != nil { + t.Fatalf("AddMember: %v", err) + } + waitForRestoration("member add") + + if err := env.core.LeaveRoom(env.ctx, viewer.Id, core.KindChannel, viewer.Id, room.Id); err != nil { + t.Fatalf("LeaveRoom before universal restore: %v", err) + } + waitForLoss() + if _, err := env.core.SetRoomUniversal(env.ctx, owner.Id, core.KindChannel, room.Id, true); err != nil { + t.Fatalf("SetRoomUniversal: %v", err) + } + waitForRestoration("universal access") +} + func TestRealtimeProjectionNotificationChangesReplaceStateAndCarryLiveTransitions(t *testing.T) { env := setupWebSocketTestServer(t) viewer, err := env.core.CreateUser(env.ctx, core.SystemActorID, "rt-notification-viewer", "RT Notification Viewer", "password123") @@ -1419,30 +1532,38 @@ func TestRealtimeProjectionNotificationChangesReplaceStateAndCarryLiveTransition t.Fatalf("CreateRoom: %v", err) } for _, userID := range []string{viewer.Id, author.Id} { - if _, err := env.core.JoinRoom(env.ctx, viewer.Id, core.KindChannel, userID, room.Id); err != nil { + if _, err := env.core.JoinRoom(env.ctx, userID, core.KindChannel, userID, room.Id); err != nil { t.Fatalf("JoinRoom %q: %v", userID, err) } } - message, err := env.core.PostMessage(env.ctx, core.KindChannel, room.Id, author.Id, "notify me", nil, "", "", nil, false) + root, err := env.core.PostMessage(env.ctx, core.KindChannel, room.Id, author.Id, "notification thread root", nil, "", "", nil, false) if err != nil { - t.Fatalf("PostMessage: %v", err) + t.Fatalf("PostMessage root: %v", err) + } + reply, err := env.core.PostMessage(env.ctx, core.KindChannel, room.Id, author.Id, "notification thread reply", nil, root.Id, "", nil, false) + if err != nil { + t.Fatalf("PostMessage reply: %v", err) } - if err := env.core.FollowThread(env.ctx, core.KindChannel, viewer.Id, room.Id, message.Id); err != nil { + if err := env.core.FollowThread(env.ctx, core.KindChannel, viewer.Id, room.Id, root.Id); err != nil { t.Fatalf("FollowThread: %v", err) } notification, err := env.core.CreateNotification(env.ctx, viewer.Id, author.Id, &corev1.Notification{ Notification: &corev1.Notification_Reply{Reply: &corev1.ReplyNotification{ - RoomId: room.Id, EventId: message.Id, InReplyToId: message.Id, InThread: message.Id, + RoomId: room.Id, EventId: reply.Id, InReplyToId: root.Id, InThread: root.Id, }}, }) if err != nil { t.Fatalf("CreateNotification: %v", err) } + dismissed, err := env.core.DismissNotification(env.ctx, viewer.Id, notification.Id) + if err != nil || !dismissed { + t.Fatalf("DismissNotification before creation mapping: dismissed=%v err=%v", dismissed, err) + } frame, handled, err := env.httpServer.realtimeProjectionFrameForEvent(env.ctx, viewer.Id, core.NewLiveEventEnvelope(&corev1.LiveEvent{ Id: "notification-created-1", ActorId: author.Id, Event: &corev1.LiveEvent_NotificationCreated{NotificationCreated: &corev1.NotificationCreatedEvent{ - NotificationId: notification.Id, RoomId: room.Id, EventId: message.Id, InReplyToId: message.Id, Silent: true, + NotificationId: notification.Id, RoomId: room.Id, EventId: reply.Id, InReplyToId: root.Id, Silent: true, }}, })) if err != nil { @@ -1452,22 +1573,17 @@ func TestRealtimeProjectionNotificationChangesReplaceStateAndCarryLiveTransition t.Fatal("notification-created event was not handled as a projection mutation") } replacement := frame.GetProjectionEvent().GetOperations()[0].GetNotificationsReplace() - if replacement == nil || len(replacement.GetPage().GetNotifications()) != 1 { - t.Fatalf("notification replacement = %+v, want authoritative one-row page", replacement) + if replacement == nil || len(replacement.GetPage().GetNotifications()) != 0 { + t.Fatalf("notification replacement = %+v, want authoritative empty page after concurrent dismissal", replacement) } change := replacement.GetChange() - if change.GetAction() != realtimev1.RealtimeProjectionNotificationAction_REALTIME_PROJECTION_NOTIFICATION_ACTION_CREATED || change.GetNotificationId() != notification.Id || !change.GetSilent() { + if change.GetAction() != realtimev1.RealtimeProjectionNotificationAction_REALTIME_PROJECTION_NOTIFICATION_ACTION_CREATED || change.GetNotificationId() != notification.Id || !change.GetSilent() || change.GetRoomId() != room.Id || change.GetThreadRootEventId() != root.Id { t.Fatalf("notification transition = %+v", change) } - threadStates := frame.GetProjectionEvent().GetOperations()[1].GetThreadViewerStatesReplace() - if threadStates == nil || len(threadStates.GetStates()) != 1 || threadStates.GetStates()[0].GetThreadRootEventId() != message.Id { - t.Fatalf("reply notification thread-state replacement = %+v, want followed thread %q", threadStates, message.Id) + if operations := frame.GetProjectionEvent().GetOperations(); len(operations) != 1 { + t.Fatalf("notification-created operations = %d, want only notification replacement", len(operations)) } - dismissed, err := env.core.DismissNotification(env.ctx, viewer.Id, notification.Id) - if err != nil || !dismissed { - t.Fatalf("DismissNotification: dismissed=%v err=%v", dismissed, err) - } dismissFrame, handled, err := env.httpServer.realtimeProjectionFrameForEvent(env.ctx, viewer.Id, core.NewLiveEventEnvelope(&corev1.LiveEvent{ Id: "notification-dismissed-1", ActorId: viewer.Id, Event: &corev1.LiveEvent_NotificationDismissed{NotificationDismissed: &corev1.NotificationDismissedEvent{ @@ -1490,7 +1606,7 @@ func TestRealtimeProjectionNotificationChangesReplaceStateAndCarryLiveTransition } } -func TestRealtimeProjectionNotificationLevelChangedReplacesViewer(t *testing.T) { +func TestRealtimeProjectionNotificationLevelChangedReplacesViewerAndThreadState(t *testing.T) { env := setupWebSocketTestServer(t) viewer, err := env.core.CreateUser(env.ctx, core.SystemActorID, "rt-notification-level-viewer", "RT Notification Level Viewer", "password123") if err != nil { @@ -1506,6 +1622,9 @@ func TestRealtimeProjectionNotificationLevelChangedReplacesViewer(t *testing.T) if !handled || frame.GetProjectionEvent().GetOperations()[0].GetViewerUpsert() == nil { t.Fatalf("notification-level projection = %+v, handled=%v; want viewer_upsert", frame, handled) } + if operations := frame.GetProjectionEvent().GetOperations(); len(operations) != 2 || operations[1].GetThreadViewerStatesReplace() == nil { + t.Fatalf("notification-level operations = %+v, want viewer and thread-state replacements", operations) + } } func TestRealtimeProjectionThreadFollowReplacesStateForUnretainedRoom(t *testing.T) { @@ -1660,9 +1779,31 @@ func TestRealtimeThreadReadMarkerPublishesProjectionUpdate(t *testing.T) { if _, err := env.core.SetThreadLastReadEventID(env.ctx, core.KindChannel, viewer.Id, room.Id, root.Id, reply.Id); err != nil { t.Fatalf("SetThreadLastReadEventID: %v", err) } - upsert := waitRealtimeTimelineUpsert(t, conn, 5*time.Second, func(upsert *realtimev1.RealtimeProjectionRoomTimelineEventUpsert) bool { - return upsert.GetRoomId() == room.Id && upsert.GetEvent().GetId() == root.Id + projection := waitRealtimeProjectionEvent(t, conn, 5*time.Second, func(projection *realtimev1.RealtimeProjectionEvent) bool { + for _, operation := range projection.GetOperations() { + if upsert := operation.GetRoomTimelineEventUpsert(); upsert.GetRoomId() == room.Id && upsert.GetEvent().GetId() == root.Id { + return true + } + } + return false }) + if projection == nil { + t.Fatal("did not receive thread read-marker projection update") + } + var upsert *realtimev1.RealtimeProjectionRoomTimelineEventUpsert + var replacement *realtimev1.RealtimeProjectionThreadViewerStatesReplace + for _, operation := range projection.GetOperations() { + if candidate := operation.GetRoomTimelineEventUpsert(); candidate.GetRoomId() == room.Id && candidate.GetEvent().GetId() == root.Id { + upsert = candidate + } + if candidate := operation.GetThreadViewerStatesReplace(); candidate != nil { + replacement = candidate + } + } + states := replacement.GetStates() + if len(states) != 1 || states[0].GetRoomId() != room.Id || states[0].GetThreadRootEventId() != root.Id || !states[0].GetViewerState().GetIsFollowing() || states[0].GetViewerState().GetHasUnread() { + t.Fatalf("thread viewer-state replacement after marker advance = %+v, want one followed and read thread", states) + } thread := upsert.GetEvent().GetMessagePosted().GetMessage().GetThread() if !thread.GetViewerState().GetIsFollowing() || thread.GetViewerState().GetHasUnread() { t.Fatalf("thread viewer state after marker advance = %+v, want following and read", thread.GetViewerState()) @@ -1835,41 +1976,126 @@ func TestRealtimeWebSocketConvergesDirectoryRoomsAndAdministrativeMembership(t * func TestRealtimeWebSocketThreadReplyUpdatesRootSummary(t *testing.T) { env := setupWebSocketTestServer(t) - user, err := env.core.CreateUser(env.ctx, core.SystemActorID, "rt-thread-member", "RT Thread Member", "password123") + viewer, err := env.core.CreateUser(env.ctx, core.SystemActorID, "rt-thread-member", "RT Thread Member", "password123") if err != nil { - t.Fatalf("CreateUser: %v", err) + t.Fatalf("CreateUser viewer: %v", err) } - room, err := env.core.CreateRoom(env.ctx, user.Id, core.KindChannel, "", "rt-thread-room", "") + author, err := env.core.CreateUser(env.ctx, core.SystemActorID, "rt-thread-author", "RT Thread Author", "password123") + if err != nil { + t.Fatalf("CreateUser author: %v", err) + } + room, err := env.core.CreateRoom(env.ctx, viewer.Id, core.KindChannel, "", "rt-thread-room", "") if err != nil { t.Fatalf("CreateRoom: %v", err) } - if _, err := env.core.JoinRoom(env.ctx, user.Id, core.KindChannel, user.Id, room.Id); err != nil { - t.Fatalf("JoinRoom: %v", err) + for _, userID := range []string{viewer.Id, author.Id} { + if _, err := env.core.JoinRoom(env.ctx, userID, core.KindChannel, userID, room.Id); err != nil { + t.Fatalf("JoinRoom %q: %v", userID, err) + } } - root, err := env.core.PostMessage(env.ctx, core.KindChannel, room.Id, user.Id, "root", nil, "", "", nil, false) + root, err := env.core.PostMessage(env.ctx, core.KindChannel, room.Id, author.Id, "root", nil, "", "", nil, false) if err != nil { t.Fatalf("PostMessage root: %v", err) } - token, err := env.core.CreateAuthToken(env.ctx, user.Id) + token, err := env.core.CreateAuthToken(env.ctx, viewer.Id) if err != nil { t.Fatalf("CreateAuthToken: %v", err) } conn := env.connectRealtime(t) subscribeRealtime(t, conn, token, room.Id) - reply, err := env.core.PostMessage(env.ctx, core.KindChannel, room.Id, user.Id, "reply", nil, root.Id, root.Id, nil, false) + reply, err := env.core.PostMessage(env.ctx, core.KindChannel, room.Id, author.Id, "reply", nil, root.Id, root.Id, nil, false) if err != nil { t.Fatalf("PostMessage reply: %v", err) } - upsert := waitRealtimeTimelineUpsert(t, conn, 5*time.Second, func(upsert *realtimev1.RealtimeProjectionRoomTimelineEventUpsert) bool { - return upsert.GetEvent().GetId() == root.Id + projection := waitRealtimeProjectionEvent(t, conn, 5*time.Second, func(projection *realtimev1.RealtimeProjectionEvent) bool { + if projection.GetId() != reply.Id { + return false + } + for _, operation := range projection.GetOperations() { + if operation.GetRoomTimelineEventUpsert().GetEvent().GetId() == root.Id { + return true + } + } + return false }) - if upsert == nil { + if projection == nil { t.Fatal("did not receive root summary upsert") } - if got := upsert.GetEvent().GetMessagePosted().GetMessage().GetThread().GetReplyCount(); got != 1 { - t.Fatalf("root reply count = %d, want 1 (reply %q)", got, reply.Id) + for _, operation := range projection.GetOperations() { + if operation.GetThreadViewerStatesReplace() != nil { + t.Fatal("unfollowed member reply performed a complete thread-state replacement") + } + if upsert := operation.GetRoomTimelineEventUpsert(); upsert.GetEvent().GetId() == root.Id { + if got := upsert.GetEvent().GetMessagePosted().GetMessage().GetThread().GetReplyCount(); got != 1 { + t.Fatalf("root reply count = %d, want 1 (reply %q)", got, reply.Id) + } + } + } +} + +func TestRealtimeWebSocketMutedFollowedThreadReplyDoesNotReplaceViewerState(t *testing.T) { + env := setupWebSocketTestServer(t) + viewer, err := env.core.CreateUser(env.ctx, core.SystemActorID, "rt-muted-thread-viewer", "RT Muted Thread Viewer", "password123") + if err != nil { + t.Fatalf("CreateUser viewer: %v", err) + } + author, err := env.core.CreateUser(env.ctx, core.SystemActorID, "rt-muted-thread-author", "RT Muted Thread Author", "password123") + if err != nil { + t.Fatalf("CreateUser author: %v", err) + } + room, err := env.core.CreateRoom(env.ctx, viewer.Id, core.KindChannel, "", "rt-muted-thread-room", "") + if err != nil { + t.Fatalf("CreateRoom: %v", err) + } + for _, userID := range []string{viewer.Id, author.Id} { + if _, err := env.core.JoinRoom(env.ctx, userID, core.KindChannel, userID, room.Id); err != nil { + t.Fatalf("JoinRoom %q: %v", userID, err) + } + } + root, err := env.core.PostMessage(env.ctx, core.KindChannel, room.Id, viewer.Id, "muted thread root", nil, "", "", nil, false) + if err != nil { + t.Fatalf("PostMessage root: %v", err) + } + if err := env.core.FollowThread(env.ctx, core.KindChannel, viewer.Id, room.Id, root.Id); err != nil { + t.Fatalf("FollowThread: %v", err) + } + if err := env.core.SetRoomNotificationLevel(env.ctx, viewer.Id, room.Id, corev1.NotificationLevel_NOTIFICATION_LEVEL_MUTED); err != nil { + t.Fatalf("SetRoomNotificationLevel: %v", err) + } + token, err := env.core.CreateAuthToken(env.ctx, viewer.Id) + if err != nil { + t.Fatalf("CreateAuthToken: %v", err) + } + conn := env.connectRealtime(t) + subscribeRealtime(t, conn, token) + + if _, err := env.core.PostMessage(env.ctx, core.KindChannel, room.Id, author.Id, "muted thread reply", nil, root.Id, "", nil, false); err != nil { + t.Fatalf("PostMessage reply: %v", err) + } + projection := waitRealtimeProjectionEvent(t, conn, 5*time.Second, func(projection *realtimev1.RealtimeProjectionEvent) bool { + for _, operation := range projection.GetOperations() { + if operation.GetRoomViewerStateReplace().GetRoomId() == room.Id { + return true + } + } + return false + }) + if projection == nil { + t.Fatal("muted thread reply did not produce its ordinary room-state projection") + } + for _, operation := range projection.GetOperations() { + if operation.GetThreadViewerStatesReplace() != nil { + t.Fatal("muted thread reply performed a complete thread-state replacement") + } + } + notifications, err := env.core.GetNotifications(env.ctx, viewer.Id) + if err != nil { + t.Fatalf("GetNotifications: %v", err) + } + if len(notifications) != 0 { + t.Fatalf("muted thread notifications = %d, want 0", len(notifications)) } } @@ -2445,6 +2671,8 @@ func TestRealtimeWebSocketResumesAssetAndHiddenEchoGapThenContinuesLive(t *testi if caughtUpCursor == resumeCursor { t.Fatal("caught_up cursor did not advance across durable replay gap") } + // Catch-up reconciles the complete latest-value thread state once, without + // performing an exhaustive replacement for every replayed reply. if replyUpserts != 1 || echoRemovals != 2 || assetUpserts != 3 || notificationReconciliations != 1 || presenceReconciliations != 1 || viewerReconciliations != 1 || roomViewerReconciliations == 0 || threadViewerReconciliations != 1 { t.Fatalf("replay reply/echo/asset/notifications/presence/viewer/room-viewer/thread-viewer = %d/%d/%d/%d/%d/%d/%d/%d, want 1/2/3/1/1/1/>0/1", replyUpserts, echoRemovals, assetUpserts, notificationReconciliations, presenceReconciliations, viewerReconciliations, roomViewerReconciliations, threadViewerReconciliations) } diff --git a/cli/internal/pb/chatto/realtime/v1/realtime.pb.go b/cli/internal/pb/chatto/realtime/v1/realtime.pb.go index 2670f5d39e..504c07d881 100644 --- a/cli/internal/pb/chatto/realtime/v1/realtime.pb.go +++ b/cli/internal/pb/chatto/realtime/v1/realtime.pb.go @@ -2061,9 +2061,13 @@ type RealtimeProjectionNotificationChange struct { Action RealtimeProjectionNotificationAction `protobuf:"varint,1,opt,name=action,proto3,enum=chatto.realtime.v1.RealtimeProjectionNotificationAction" json:"action,omitempty"` NotificationId string `protobuf:"bytes,2,opt,name=notification_id,json=notificationId,proto3" json:"notification_id,omitempty"` // True when a created notification must not produce an alert. - Silent bool `protobuf:"varint,3,opt,name=silent,proto3" json:"silent,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + Silent bool `protobuf:"varint,3,opt,name=silent,proto3" json:"silent,omitempty"` + // Exact followed-thread target of a created reply or mention, when present. + // This remains available even if the notification was concurrently dismissed. + RoomId string `protobuf:"bytes,4,opt,name=room_id,json=roomId,proto3" json:"room_id,omitempty"` + ThreadRootEventId string `protobuf:"bytes,5,opt,name=thread_root_event_id,json=threadRootEventId,proto3" json:"thread_root_event_id,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *RealtimeProjectionNotificationChange) Reset() { @@ -2117,6 +2121,20 @@ func (x *RealtimeProjectionNotificationChange) GetSilent() bool { return false } +func (x *RealtimeProjectionNotificationChange) GetRoomId() string { + if x != nil { + return x.RoomId + } + return "" +} + +func (x *RealtimeProjectionNotificationChange) GetThreadRootEventId() string { + if x != nil { + return x.ThreadRootEventId + } + return "" +} + // Lightweight current viewer state for one projected room. type RealtimeProjectionRoomViewerStateReplace struct { state protoimpl.MessageState `protogen:"open.v1"` @@ -3218,11 +3236,13 @@ const file_chatto_realtime_v1_realtime_proto_rawDesc = "" + "\vroom_counts\x18\x02 \x03(\v2$.chatto.api.v1.RoomNotificationCountR\n" + "roomCounts\x12U\n" + "\x06change\x18\x03 \x01(\v28.chatto.realtime.v1.RealtimeProjectionNotificationChangeH\x00R\x06change\x88\x01\x01B\t\n" + - "\a_change\"\xb9\x01\n" + + "\a_change\"\x83\x02\n" + "$RealtimeProjectionNotificationChange\x12P\n" + "\x06action\x18\x01 \x01(\x0e28.chatto.realtime.v1.RealtimeProjectionNotificationActionR\x06action\x12'\n" + "\x0fnotification_id\x18\x02 \x01(\tR\x0enotificationId\x12\x16\n" + - "\x06silent\x18\x03 \x01(\bR\x06silent\"\x86\x01\n" + + "\x06silent\x18\x03 \x01(\bR\x06silent\x12\x17\n" + + "\aroom_id\x18\x04 \x01(\tR\x06roomId\x12/\n" + + "\x14thread_root_event_id\x18\x05 \x01(\tR\x11threadRootEventId\"\x86\x01\n" + "(RealtimeProjectionRoomViewerStateReplace\x12\x17\n" + "\aroom_id\x18\x01 \x01(\tR\x06roomId\x12A\n" + "\fviewer_state\x18\x02 \x01(\v2\x1e.chatto.api.v1.RoomViewerStateR\vviewerState\"W\n" + diff --git a/docs/architecture/realtime-delivery.md b/docs/architecture/realtime-delivery.md index d9f4582bf8..8d83567b39 100644 --- a/docs/architecture/realtime-delivery.md +++ b/docs/architecture/realtime-delivery.md @@ -176,20 +176,26 @@ IDs; a compacted reset includes only those room windows. Effective membership changes are authoritative timeline boundaries. When a universal room stops granting membership, live mapping pairs its current room state with an empty replacement for any retained timeline plus authoritative -active-call and notification replacements; loss of room -visibility uses `room_remove`, which has the same eviction effect. The browser -also scrubs canonical rows, mounted room stores, open thread stores, optimistic -state, call and notification mirrors, and in-flight reads as soon as projected -membership becomes false. It also disconnects local call media for that room -without issuing a redundant leave command. The privacy fence stays closed until an explicit -positive membership operation arrives, so delayed pagination, previews, -read-your-writes responses, and timeline replacements cannot restore plaintext. +active-call and notification replacements; loss of room visibility uses +`room_remove`, which has the same eviction effect. The browser purges cached +thread viewer states for the room directly, avoiding an unrelated exhaustive +thread-state read on the privacy-critical loss event. + +The browser also scrubs canonical rows, mounted room stores, open thread stores, +optimistic state, call and notification mirrors, cached thread viewer states, +and in-flight reads as soon as projected membership becomes false. It also +disconnects local call media for that room without issuing a redundant leave +command. The privacy fence stays closed until an explicit positive membership +operation arrives, so delayed pagination, previews, read-your-writes responses, +and timeline replacements cannot restore plaintext. The browser keeps only the non-plaintext retained-room intent. If membership later returns, the server rematerialises the current window only for that -retained room; never-requested rooms remain lazy. A disconnected client whose -gap contains an authorization-sensitive revocation receives a compacted reset -instead of incremental replay. +retained room and replaces the viewer's complete followed-thread state so +durable follows purged at the privacy boundary become visible again. +Never-requested rooms remain lazy. A disconnected client whose gap contains an +authorization-sensitive revocation receives a compacted reset instead of +incremental replay. The browser advertises a room as retained only after applying its timeline replacement. Desired rooms with lost or unavailable hydration responses remain @@ -286,10 +292,18 @@ list for existing viewers. Message facts carry lightweight replacements of the affected room summary and viewer state alongside timeline mutations. Root messages also carry a content-free `room_activity` operation, allowing unretained DMs to reorder -without exposing or materialising their message. Notification counts converge -through notification signals and the finite resume replacement. Message -delivery does not reassemble or retransmit complete channel membership. Echo -tombstone upserts explicitly distinguish +without exposing or materialising their message. A created reply or mention +notification marks its exact already-followed thread unread in the browser's +latest-value set. This keeps live reply work bounded and naturally excludes +muted rooms, which do not create notifications or sidebar unread state. +The created transition carries that target independently of the pending page, +so a concurrent dismissal on another device cannot erase the unread signal. + +Follow/read mutations, notification-level changes, access restoration, and the +finite resume replacement still send authoritative complete thread state. +Notification counts converge through notification signals and the finite +resume replacement. Message delivery does not reassemble or retransmit complete +channel membership. Echo tombstone upserts explicitly distinguish canonical-reply deletion from direct echo removal. Room-read signals emit a `RoomViewerStateReplace` for the affected room and a diff --git a/packages/api-types/src/chatto/realtime/v1/realtime_pb.ts b/packages/api-types/src/chatto/realtime/v1/realtime_pb.ts index 21e7e4a70f..343a713401 100644 --- a/packages/api-types/src/chatto/realtime/v1/realtime_pb.ts +++ b/packages/api-types/src/chatto/realtime/v1/realtime_pb.ts @@ -1604,6 +1604,19 @@ export class RealtimeProjectionNotificationChange extends Message) { super(); proto3.util.initPartial(data, this); @@ -1615,6 +1628,8 @@ export class RealtimeProjectionNotificationChange extends Message): RealtimeProjectionNotificationChange { diff --git a/proto/chatto/realtime/v1/realtime.proto b/proto/chatto/realtime/v1/realtime.proto index 2ae2eb6df0..c3e2192216 100644 --- a/proto/chatto/realtime/v1/realtime.proto +++ b/proto/chatto/realtime/v1/realtime.proto @@ -348,6 +348,10 @@ message RealtimeProjectionNotificationChange { string notification_id = 2; // True when a created notification must not produce an alert. bool silent = 3; + // Exact followed-thread target of a created reply or mention, when present. + // This remains available even if the notification was concurrently dismissed. + string room_id = 4; + string thread_root_event_id = 5; } // Kind of live notification transition.