From 547de61aa5fed28003c2d79e1720dfaaeae297c7 Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Fri, 30 Oct 2026 20:34:41 -0400 Subject: [PATCH] fix(mobile): show channel posts before a burst of join notices (#8798) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary When many people join a channel at once, the mobile app could open that channel and show only a "… joined" line, with none of the channel's posts. You had to scroll up before any messages appeared. With this change, the posts always show first, and the join lines appear among them in the right places. The cause: the app loads the newest 50 timeline rows when a channel opens, and join notices counted toward those 50. A burst of 50 or more joins filled the whole first page. The app combines a run of joins into one line only when it draws the screen, so the user saw a single "Guest 10 joined … and 48 others" line. The app loads older rows only after the user scrolls. This is the mobile port of the desktop app's fix, https://github.com/block/buzz-app/pull/1983, which has already merged. It needs no relay change. - History pages no longer include join notices, so the first page is all posts. - The app reads join notices with a separate read, newest first, 500 per read. It reads down to the oldest post loaded so far, and continues after each older page loads. Like desktop, it stops after 2,000 notices. - Each notice read also reads the reactions on those notices, and any removals of those reactions. A page brings these back for its own rows, but notices are no longer on the pages, so they need their own read. These reads name the channel in `#h`, so a system channel, or an unlisted channel opened by a non-member, still returns them (NIP-CL). Deletions are also read without `#h`, because a deletion can be stored with no channel. Each read names at most 500 target ids, so it stays under the relay's 1 MiB request limit. A notice's reactions show even when they sort below the oldest loaded post. - Posts show as soon as their page loads. The notices follow and are merged into the timeline, so they do not replace rows that arrived live in the meantime. - A notice that is older than the oldest loaded post waits until its page loads. This prevents a join from showing at the top of the timeline with a gap of missing posts below it. - If a notice read fails, the posts still show. The next page load tries the notices again; if a page loaded while the read was running, the read runs once more right away. - The same applies to every system notice (joins, topic changes, channel created or archived), as they share one event kind. ### Related issue AIDA-2328. Desktop fix (merged): https://github.com/block/buzz-app/pull/1983. ### Testing Before (left, older build) and after (right, this branch) on the Android emulator, against a local relay. #general and #4things each have 4 posts followed by 60 joins: https://github.com/user-attachments/assets/53b80942-d8c3-4604-8b95-cc5ca7ab6946 ![Before and after: the old build shows only a joins line; the fix shows the 4 posts above the joins line](https://github.com/user-attachments/assets/58e121b9-855c-4844-a73a-c71b54252db2) - New `channel_messages_provider_test.dart` cases run against the real channel provider, with only relay responses faked: - A burst of 500+ joins does not hold posts back. If the first notice read fails, the posts still show. A later page read retries the notices and continues the descent. - Notices in the same second as the oldest loaded post follow the post's position within that second. - The descent stops at 2,000 notices. - A notice keeps its reactions, and a removed reaction stays removed. Reactions are read past the relay's 1,000-event page. The fake relay applies the NIP-CL rule, so a read without `#h` finds no channel reactions. - A page that loads while a notice read fails still gets its notices. - `just mobile-check` and the full `just mobile-test` pass at `7a38763af`. Not yet tested on a physical device, or on iOS. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Signed-off-by: Logan Johnson Co-authored-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz> Co-authored-by: Claude Opus 5.5 --- .../channels/channel_messages_provider.dart | 37 +- .../channel_messages_provider/notices.dart | 199 +++++++ .../channel_messages_provider/pagination.dart | 5 +- .../lib/features/channels/channel_window.dart | 62 +- mobile/lib/shared/relay/nostr_filters.dart | 56 ++ mobile/lib/shared/relay/nostr_models.dart | 14 +- .../channel_messages_journey_test.dart | 6 +- .../channel_messages_provider_test.dart | 536 +++++++++++++++++- 8 files changed, 896 insertions(+), 19 deletions(-) create mode 100644 mobile/lib/features/channels/channel_messages_provider/notices.dart diff --git a/mobile/lib/features/channels/channel_messages_provider.dart b/mobile/lib/features/channels/channel_messages_provider.dart index f4009f060a..a7cc6521bc 100644 --- a/mobile/lib/features/channels/channel_messages_provider.dart +++ b/mobile/lib/features/channels/channel_messages_provider.dart @@ -14,6 +14,7 @@ import 'timeline_message.dart'; part 'channel_messages_provider/thread_summaries.dart'; part 'channel_messages_provider/pagination.dart'; +part 'channel_messages_provider/notices.dart'; // Channel history keeps partial reply evidence; full threads stay route-scoped. const _maxCachedRepliesPerRoot = 256; @@ -43,6 +44,22 @@ class ChannelMessagesNotifier extends Notifier>> { bool _initialWindowQueryInFlight = false; int _initVersion = 0; ChannelWindowStore _windowStore = const ChannelWindowStore.empty(); + + /// System notices read apart from the window pages (see `_fillNotices`). + final Map _notices = {}; + + /// Reactions to [_notices] and deletions of those reactions. + final Map _noticeAux = {}; + + /// Where the next older notice read starts: null before the first read. + ChannelPageCursor? _noticeCursor; + bool _noticesReachedStart = false; + + /// The init a notice read is running for, so only one runs per window. + int? _noticeFillVersion; + + /// Whether a page loaded while a notice read ran, so the read runs again. + bool _noticeFillAgain = false; final Set _liveSummaryRootsDuringInitialWindowQuery = {}; final Map _deepLinkEvents = {}; final Set _retainedDeepLinkEventIds = {}; @@ -170,6 +187,11 @@ class ChannelMessagesNotifier extends Notifier>> { _reachedOldest = false; _windowStore = const ChannelWindowStore.empty(); + _notices.clear(); + _noticeAux.clear(); + _noticeCursor = null; + _noticesReachedStart = false; + _noticeFillAgain = false; // A reconnect window can lag a confirmed thread query and contains only // top-level rows. Keep known replies as summary evidence, including their // deletion markers so a reset cannot resurrect a deleted reply. @@ -238,6 +260,7 @@ class ChannelMessagesNotifier extends Notifier>> { _reconcileRetainedAggregates(historyVersion); state = AsyncData(merged); _summaryRefreshes.resume(); + unawaited(_fillNotices()); } catch (e, st) { if (!_isCurrentInit(initVersion)) return; final fallbackMessages = state.value ?? _lastKnownMessages; @@ -283,7 +306,7 @@ class ChannelMessagesNotifier extends Notifier>> { _usingChannelWindow = true; _reconcilePageSummaries(page, pageVersion); _reachedOldest = !channelWindowHasMore(_windowStore); - return flattenChannelWindowEvents(_windowStore); + return _windowEvents(); } catch (error) { _initialWindowQueryInFlight = false; _liveSummaryRootsDuringInitialWindowQuery.clear(); @@ -310,7 +333,7 @@ class ChannelMessagesNotifier extends Notifier>> { } NostrFilter _channelWindowFilter(ChannelPageCursor? cursor) => NostrFilter( - kinds: EventKind.channelTimelineContentKinds, + kinds: EventKind.channelWindowPageKinds, tags: { '#h': [channelId], }, @@ -404,7 +427,7 @@ class ChannelMessagesNotifier extends Notifier>> { if (!_mergeWindowEventIntoStore(event, summaryVersion: summaryVersion)) { return; } - final windowEvents = flattenChannelWindowEvents(_windowStore); + final windowEvents = _windowEvents(); // Flattening already orders the window. Only merge and sort again when // there are retained deep-link events to include. final flattened = _deepLinkEvents.isEmpty @@ -783,7 +806,7 @@ class ChannelMessagesNotifier extends Notifier>> { } } if (_usingChannelWindow) { - events = _withDeepLinkEvents(flattenChannelWindowEvents(_windowStore)); + events = _withDeepLinkEvents(_windowEvents()); } events = _boundEventReplies(events); _lastKnownMessages = events; @@ -896,6 +919,12 @@ class ChannelMessagesNotifier extends Notifier>> { bool _isCurrentInit(int initVersion) => initVersion == _initVersion; + List _windowEvents() => flattenChannelWindowEvents( + _windowStore, + notices: _notices.values, + noticeAux: _noticeAux.values, + ); + /// Stops live work for a chat that stayed hidden; resume catches it up. void _detachLive() { _liveDetachTimer = null; diff --git a/mobile/lib/features/channels/channel_messages_provider/notices.dart b/mobile/lib/features/channels/channel_messages_provider/notices.dart new file mode 100644 index 0000000000..0740ac7269 --- /dev/null +++ b/mobile/lib/features/channels/channel_messages_provider/notices.dart @@ -0,0 +1,199 @@ +part of '../channel_messages_provider.dart'; + +/// System notices a window reads below its head. Past it, older pages +/// show without their notices. +const _maxNotices = 2000; + +/// Pages one auxiliary step reads before it stops, as the relay's own +/// `include_aux` does. +const _maxAuxPages = 64; + +const _deletionKinds = [EventKind.deletion, EventKind.nip29DeleteEvent]; + +/// Target ids one auxiliary filter names, which keeps each request well +/// under the relay's 1 MiB body limit. +const _auxTargetBatch = 500; + +/// Where the read after the full page [events] starts: its oldest event in +/// relay order (by time, then the greatest id). Throws when the page did not +/// move below [cursor]. +ChannelPageCursor _nextCursor( + List events, + ChannelPageCursor? cursor, + String what, +) { + final last = events.reduce( + (a, b) => + a.createdAt < b.createdAt || + (a.createdAt == b.createdAt && a.id.compareTo(b.id) > 0) + ? a + : b, + ); + if (cursor != null && !isOlderThanChannelCursor(last, cursor)) { + throw Exception('$what did not advance.'); + } + return ChannelPageCursor(createdAt: last.createdAt, eventId: last.id); +} + +extension _ChannelNotices on ChannelMessagesNotifier { + /// Reads the window's system notices newest first, down to its oldest + /// loaded row. Pages leave notices out (see + /// [EventKind.channelWindowPageKinds]), so the messages show first and the + /// notices between them follow. Each page read calls this again to continue + /// the descent, and a call during a read runs it once more after; a failed + /// read leaves notices out until then. + Future _fillNotices() async { + final initVersion = _initVersion; + if (!_usingChannelWindow) return; + if (_noticeFillVersion == initVersion) { + _noticeFillAgain = true; + return; + } + _noticeFillVersion = initVersion; + _noticeFillAgain = false; + final session = _summarySession; + try { + while (_isCurrentInit(initVersion) && + !_noticesReachedStart && + _notices.length < _maxNotices) { + final cursor = _noticeCursor; + final floor = channelWindowFloor(_windowStore); + // Notices below the oldest loaded row wait for the next page. + if (cursor != null && + floor != null && + !isChannelCursorOlder(floor, cursor)) { + break; + } + final events = await session.queryRelay([ + NostrFilters.channelNotices( + channelId, + until: cursor?.createdAt, + beforeId: cursor?.eventId, + ), + ]); + if (!_isCurrentInit(initVersion)) return; + if (events.any( + (event) => + event.kind != EventKind.systemMessage || + event.channelId != channelId, + )) { + throw Exception('System notices are outside their channel.'); + } + // Pages carry their rows' reactions; notices read apart need their + // own. Read them before the notices count as read, so a failure + // reads the whole page again. + final aux = await _readNoticeAux(session, events); + if (!_isCurrentInit(initVersion)) return; + if (events.length < NostrFilters.channelNoticePageSize) { + _noticesReachedStart = true; + } else { + _noticeCursor = _nextCursor(events, cursor, 'System notices'); + } + for (final event in events) { + _notices[event.id] = event; + } + for (final event in aux) { + _noticeAux[event.id] = event; + } + // Add to what shows rather than replace it: the timeline can hold + // rows the window store does not, such as live rows from before the + // first page. + final current = + _historyState.value ?? _lastKnownMessages ?? const []; + final ids = current.map((event) => event.id).toSet(); + final added = [ + ...channelWindowVisibleNotices(_windowStore, events), + ...aux, + ].where((event) => ids.add(event.id)).toList(); + if (added.isEmpty) continue; + final merged = [...current, ...added] + ..sort(compareChannelTimelineEventsChronologically); + _lastKnownMessages = merged; + _historyState = AsyncData(merged); + } + } catch (error) { + debugPrint( + '[ChannelMessagesNotifier] system notices failed for $channelId: $error', + ); + } finally { + if (_noticeFillVersion == initVersion) { + _noticeFillVersion = null; + // A page that loaded during a failed read still needs its notices. + if (_noticeFillAgain && _isCurrentInit(initVersion)) { + unawaited(_fillNotices()); + } + } + } + } + + /// The reactions to [notices] and the deletions of those reactions, in the + /// two steps a page's `include_aux` takes. + Future> _readNoticeAux( + RelaySessionNotifier session, + List notices, + ) async { + if (notices.isEmpty) return const []; + final aux = await _readTargeting(session, [ + for (final event in notices) event.id, + ], NostrFilters.channelAuxKinds); + final reactions = [ + for (final event in aux) + if (event.kind == EventKind.reaction) event.id, + ]; + if (reactions.isEmpty) return aux; + final removed = await _readTargeting(session, reactions, _deletionKinds); + return [...aux, ...removed]; + } + + /// The events of [kinds] in this channel that target [eventIds], and the + /// deletions that target them with no channel, [_auxTargetBatch] targets + /// per read. See [NostrFilters.channelAux]. + Future> _readTargeting( + RelaySessionNotifier session, + List eventIds, + List kinds, + ) async { + final byId = {}; + for (var start = 0; start < eventIds.length; start += _auxTargetBatch) { + final batch = eventIds.skip(start).take(_auxTargetBatch).toList(); + final inChannel = await _readAllAux( + session, + batch, + kinds, + channelId: channelId, + ); + final deletions = await _readAllAux(session, batch, _deletionKinds); + for (final event in [...inChannel, ...deletions]) { + byId[event.id] = event; + } + } + return byId.values.toList(); + } + + /// Every [NostrFilters.channelAux] event of [kinds] for [eventIds], page + /// by page. + Future> _readAllAux( + RelaySessionNotifier session, + List eventIds, + List kinds, { + String? channelId, + }) async { + final events = []; + ChannelPageCursor? cursor; + for (var page = 0; page < _maxAuxPages; page++) { + final read = await session.queryRelay([ + NostrFilters.channelAux( + eventIds, + channelId: channelId, + kinds: kinds, + until: cursor?.createdAt, + beforeId: cursor?.eventId, + ), + ]); + events.addAll(read); + if (read.length < NostrFilters.channelAuxLimit) break; + cursor = _nextCursor(read, cursor, 'Reactions'); + } + return events; + } +} diff --git a/mobile/lib/features/channels/channel_messages_provider/pagination.dart b/mobile/lib/features/channels/channel_messages_provider/pagination.dart index 692fda5ea3..18b54c7523 100644 --- a/mobile/lib/features/channels/channel_messages_provider/pagination.dart +++ b/mobile/lib/features/channels/channel_messages_provider/pagination.dart @@ -17,12 +17,11 @@ extension _ChannelHistoryPagination on ChannelMessagesNotifier { _windowStore = appendOlderChannelWindow(_windowStore, page); _reconcilePageSummaries(page, pageVersion); _reachedOldest = !channelWindowHasMore(_windowStore); - final flattened = _withDeepLinkEvents( - flattenChannelWindowEvents(_windowStore), - ); + final flattened = _withDeepLinkEvents(_windowEvents()); _lastKnownMessages = flattened; _reconcileRetainedSummaryPayloads(page.rows.map((row) => row.event)); _historyState = AsyncData(flattened); + unawaited(_fillNotices()); return page.rows.isNotEmpty || page.aux.isNotEmpty; } catch (error) { debugPrint( diff --git a/mobile/lib/features/channels/channel_window.dart b/mobile/lib/features/channels/channel_window.dart index 96c5eda549..d842a51f92 100644 --- a/mobile/lib/features/channels/channel_window.dart +++ b/mobile/lib/features/channels/channel_window.dart @@ -360,8 +360,20 @@ ChannelWindowStore mergeLiveChannelWindowEvent( ); } -List flattenChannelWindowEvents(ChannelWindowStore store) { - final byId = {}; +/// Flattens a window into timeline order. [notices] are system notices +/// read apart from the pages (see [channelWindowVisibleNotices]), and +/// [noticeAux] their reactions and deletions. Like a page's aux, those show +/// whatever their own position: they apply to their targets. +List flattenChannelWindowEvents( + ChannelWindowStore store, { + Iterable notices = const [], + Iterable noticeAux = const [], +}) { + final byId = { + for (final event in channelWindowVisibleNotices(store, notices)) + event.id: event, + for (final event in noticeAux) event.id: event, + }; for (final page in store.pages) { for (final row in page.rows) { byId[row.event.id] = row.event; @@ -380,6 +392,30 @@ List flattenChannelWindowEvents(ChannelWindowStore store) { ..sort(compareChannelTimelineEventsChronologically); } +/// The [notices] a window shows: while older pages remain, those older than +/// its oldest loaded row stay out, as the rows between them are not loaded yet. +Iterable channelWindowVisibleNotices( + ChannelWindowStore store, + Iterable notices, +) { + final floor = channelWindowFloor(store); + return floor == null + ? notices + : notices.where((event) => !isOlderThanChannelCursor(event, floor)); +} + +/// The position of the oldest row a window has loaded while older pages +/// remain, or null once the window reaches the channel's start. +ChannelPageCursor? channelWindowFloor(ChannelWindowStore store) { + if (!channelWindowHasMore(store)) return null; + for (final page in store.pages.reversed) { + if (page.rows.isEmpty) continue; + final oldest = page.rows.last.event; + return ChannelPageCursor(createdAt: oldest.createdAt, eventId: oldest.id); + } + return null; +} + bool channelWindowHasMore(ChannelWindowStore store) => store.pages.isNotEmpty && store.pages.last.hasMore; @@ -443,7 +479,7 @@ void _assertValidPage(ChannelWindowPage page) { throw Exception('Duplicate channel row ${event.id}.'); } final startCursor = page.startCursor; - if (startCursor != null && !_isStrictlyOlder(event, startCursor)) { + if (startCursor != null && !isOlderThanChannelCursor(event, startCursor)) { throw Exception( 'Channel row ${event.id} is outside its cursor interval.', ); @@ -455,10 +491,22 @@ void _assertValidPage(ChannelWindowPage page) { } } -bool _isStrictlyOlder(NostrEvent event, ChannelPageCursor cursor) => - event.createdAt < cursor.createdAt || - (event.createdAt == cursor.createdAt && - event.id.compareTo(cursor.eventId) > 0); +/// Whether [event] comes after [cursor] in relay order: time descending, then +/// id ascending at ties. +bool isOlderThanChannelCursor(NostrEvent event, ChannelPageCursor cursor) => + isChannelCursorOlder( + ChannelPageCursor(createdAt: event.createdAt, eventId: event.id), + cursor, + ); + +/// Whether [position] comes after [cursor] in relay order. +bool isChannelCursorOlder( + ChannelPageCursor position, + ChannelPageCursor cursor, +) => + position.createdAt < cursor.createdAt || + (position.createdAt == cursor.createdAt && + position.eventId.compareTo(cursor.eventId) > 0); int _compareRelayOrder(NostrEvent left, NostrEvent right) { if (left.createdAt != right.createdAt) { diff --git a/mobile/lib/shared/relay/nostr_filters.dart b/mobile/lib/shared/relay/nostr_filters.dart index e65e54bba7..8af7f940d2 100644 --- a/mobile/lib/shared/relay/nostr_filters.dart +++ b/mobile/lib/shared/relay/nostr_filters.dart @@ -58,6 +58,62 @@ abstract final class NostrFilters { extensions: {'before_id': ?beforeId}, ); + /// A channel's system notices (kind:40099), newest first, below the + /// `until` + `before_id` cursor the channel-window pages use. + static NostrFilter channelNotices( + String channelId, { + int? until, + String? beforeId, + }) => NostrFilter( + kinds: const [EventKind.systemMessage], + tags: { + '#h': [channelId], + }, + limit: channelNoticePageSize, + until: until, + extensions: {'before_id': ?beforeId}, + ); + + /// Notices one [channelNotices] read asks for. + static const channelNoticePageSize = 500; + + /// Reactions and deletions that target [eventIds]: the auxiliary events a + /// channel-window page returns with its rows (`include_aux`), newest first, + /// below the `until` + `before_id` cursor. Call again with the reaction ids, + /// asking only for [EventKind.deletion] and [EventKind.nip29DeleteEvent], + /// to read removed reactions. + /// + /// With [channelId], the read names the channel in `#h`: a system channel, + /// or an unlisted one for a non-member, shows its events only to a filter + /// that names it (NIP-CL). Without it, the read also finds deletions stored + /// with no channel. + static NostrFilter channelAux( + List eventIds, { + String? channelId, + List kinds = channelAuxKinds, + int? until, + String? beforeId, + }) => NostrFilter( + kinds: kinds, + tags: { + '#e': eventIds, + if (channelId != null) '#h': [channelId], + }, + limit: channelAuxLimit, + until: until, + extensions: {'before_id': ?beforeId}, + ); + + /// The kinds [channelAux] reads by default. + static const channelAuxKinds = [ + EventKind.deletion, + EventKind.reaction, + EventKind.nip29DeleteEvent, + ]; + + /// Auxiliary events one [channelAux] read asks for. + static const channelAuxLimit = 1000; + /// Huddle lifecycle state for one parent channel, independent of timeline paging. static NostrFilter huddleLifecycle(String channelId) => NostrFilter( kinds: [EventKind.huddleStarted, EventKind.huddleEnded], diff --git a/mobile/lib/shared/relay/nostr_models.dart b/mobile/lib/shared/relay/nostr_models.dart index 83ecda9a36..19e657fe31 100644 --- a/mobile/lib/shared/relay/nostr_models.dart +++ b/mobile/lib/shared/relay/nostr_models.dart @@ -83,12 +83,14 @@ abstract final class EventKind { streamMessageEdit, ]; - /// Visible content kinds requested by the NIP-CW channel-window path. - static const channelTimelineContentKinds = [ + /// Row kinds a NIP-CW channel-window page reads. System notices (40099) + /// are read apart by [NostrFilters.channelNotices]: a page counts every row + /// kind against its limit, so a burst of joins would otherwise fill the + /// first page and hold the messages back. + static const channelWindowPageKinds = [ streamMessage, streamMessageV2, streamMessageDiff, - systemMessage, jobRequest, jobAccepted, jobProgress, @@ -98,6 +100,12 @@ abstract final class EventKind { huddleStarted, huddleEnded, ]; + + /// Visible content kinds of the channel timeline. + static const channelTimelineContentKinds = [ + ...channelWindowPageKinds, + systemMessage, + ]; } /// A Nostr event as defined by NIP-01. diff --git a/mobile/test/features/channels/channel_messages_journey_test.dart b/mobile/test/features/channels/channel_messages_journey_test.dart index cb0bac6439..b0009c1c83 100644 --- a/mobile/test/features/channels/channel_messages_journey_test.dart +++ b/mobile/test/features/channels/channel_messages_journey_test.dart @@ -70,7 +70,11 @@ void main() { admitted.complete(); settle(); expect(timeline(), ['committed as it opened', 'first', 'second']); - expect(relay.queries, hasLength(1), reason: 'read the channel window'); + expect( + relay.queries, + hasLength(2), + reason: 'read the channel window, then its system notices', + ); expect( relay.reqs.where((req) => (req[1] as String).startsWith('h-')), isEmpty, diff --git a/mobile/test/features/channels/channel_messages_provider_test.dart b/mobile/test/features/channels/channel_messages_provider_test.dart index f0a7d1a08b..2a97f3f6ff 100644 --- a/mobile/test/features/channels/channel_messages_provider_test.dart +++ b/mobile/test/features/channels/channel_messages_provider_test.dart @@ -240,7 +240,7 @@ void main() { expect(relaySession.liveFilters.single.limit, 200); expect( relaySession.queryFilters.first.kinds, - EventKind.channelTimelineContentKinds, + EventKind.channelWindowPageKinds, ); expect(relaySession.queryFilters.first.tags['#h'], [_channelId]); expect(relaySession.queryFilters.first.extensions['top_level'], isTrue); @@ -354,6 +354,456 @@ void main() { }, ); + test('a burst of joins does not hold channel messages back', () async { + const id = _paddedId; + const notice = _joinNotice; + String cursorKey(int createdAt, String eventId) => + '${_channelId.toLowerCase()}:$createdAt:$eventId'; + // A full notice page, all newer than every message. + final burst = [ + for (var i = 0; i < NostrFilters.channelNoticePageSize; i++) + notice(id('b${i.toString().padLeft(3, '0')}'), 600 - i), + ]; + final relaySession = _RecordingRelaySessionNotifier( + queryResults: [ + [ + _event(id: id('a100'), createdAt: 100), + _event(id: id('a50'), createdAt: 50), + _bounds(hasMore: true, cursorCreatedAt: 50, cursorId: id('a50')), + ], + [ + _event(id: id('a30'), createdAt: 30), + _bounds( + hasMore: true, + cursorCreatedAt: 30, + cursorId: id('a30'), + dTag: cursorKey(50, id('a50')), + ), + ], + [ + _event(id: id('a10'), createdAt: 10), + _bounds(dTag: cursorKey(30, id('a30'))), + ], + ], + ); + relaySession.noticeResults.addAll([ + Exception('notices unavailable'), + burst, + // Also full: it ends below the oldest loaded message. + [ + notice(id('n70'), 70), + for (var i = 0; i < NostrFilters.channelNoticePageSize - 2; i++) + notice(id('c${i.toString().padLeft(3, '0')}'), 25), + notice(id('n20'), 20), + ], + ]); + final container = _buildContainer(relaySession); + addTearDown(container.dispose); + List shown() => container + .read(channelMessagesProvider(_channelId)) + .value! + .map((event) => event.id) + .toList(); + + container.read(channelMessagesProvider(_channelId)); + await relaySession.subscribed; + await _pumpEventQueue(); + + // The page leaves notices to their own read, so it is all messages. + final page = relaySession.queryFilters.single; + expect(page.kinds, isNot(contains(EventKind.systemMessage))); + expect(page.limit, 50); + // A failed notice read leaves the messages showing. + expect(relaySession.noticeFilters, hasLength(1)); + expect( + relaySession.noticeFilters.single.limit, + NostrFilters.channelNoticePageSize, + ); + expect(relaySession.noticeFilters.single.authors, isNull); + expect(container.read(channelMessagesProvider(_channelId)).hasError, false); + expect(shown(), [id('a50'), id('a100')]); + + // The next page read retries the notices from the newest, and reads + // them down to the oldest loaded message. + final notifier = container.read( + channelMessagesProvider(_channelId).notifier, + ); + expect(await notifier.fetchOlder(), isTrue); + await _pumpEventQueue(); + expect(relaySession.noticeFilters, hasLength(3)); + expect(relaySession.noticeFilters[1].until, isNull); + expect(relaySession.noticeFilters[2].until, 101); + expect(relaySession.noticeFilters[2].extensions['before_id'], id('b499')); + expect(relaySession.noticeFilters[2].tags['#h'], [_channelId]); + // Notices below the oldest loaded message wait for its page. + expect(shown(), [ + id('a30'), + id('a50'), + id('n70'), + id('a100'), + ...burst.reversed.map((event) => event.id), + ]); + + // The last page shows the rest, and the descent reads on to the start. + expect(await notifier.fetchOlder(), isTrue); + await _pumpEventQueue(); + expect(relaySession.noticeFilters, hasLength(4)); + expect(relaySession.noticeFilters[3].until, 20); + expect(shown().take(2), [id('a10'), id('n20')]); + expect(shown(), hasLength(4 + 2 * NostrFilters.channelNoticePageSize)); + }); + + test('same-second notices follow the oldest loaded message', () async { + const id = _paddedId; + const notice = _joinNotice; + final relaySession = _RecordingRelaySessionNotifier( + queryResults: [ + [ + _event(id: id('m500'), createdAt: 50), + _bounds(hasMore: true, cursorCreatedAt: 50, cursorId: id('m500')), + ], + [ + _event(id: id('a10'), createdAt: 10), + _bounds(dTag: '${_channelId.toLowerCase()}:50:${id('m500')}'), + ], + ], + ); + final newer = [ + for (var i = 0; i < NostrFilters.channelNoticePageSize - 1; i++) + notice(id('m${i.toString().padLeft(3, '0')}'), 50), + ]; + relaySession.noticeResults.addAll([ + // Full, and ends in the same second as the oldest message, before it. + [ + for (var i = 0; i < NostrFilters.channelNoticePageSize - 1; i++) + notice(id('b${i.toString().padLeft(3, '0')}'), 600 - i), + notice(id('l000'), 50), + ], + // Full, and ends in that second after it. + [...newer, notice(id('n000'), 50)], + ]); + final container = _buildContainer(relaySession); + addTearDown(container.dispose); + List shown() => container + .read(channelMessagesProvider(_channelId)) + .value! + .map((event) => event.id) + .toList(); + + container.read(channelMessagesProvider(_channelId)); + await relaySession.subscribed; + await _pumpEventQueue(); + + // The descent continues past a same-second cursor before the message, + // and stops at one after it. + expect(relaySession.noticeFilters, hasLength(2)); + expect(relaySession.noticeFilters[1].until, 50); + expect(relaySession.noticeFilters[1].extensions['before_id'], id('l000')); + expect(shown(), contains(id('l000'))); + expect(shown(), containsAll(newer.map((event) => event.id))); + expect(shown(), isNot(contains(id('n000')))); + + final notifier = container.read( + channelMessagesProvider(_channelId).notifier, + ); + expect(await notifier.fetchOlder(), isTrue); + await _pumpEventQueue(); + expect(relaySession.noticeFilters, hasLength(3)); + expect(relaySession.noticeFilters[2].extensions['before_id'], id('n000')); + expect(shown(), contains(id('n000'))); + }); + + test('notices read apart keep their reactions', () async { + NostrEvent aux( + String id, + int kind, + String target, { + String content = '', + bool inChannel = true, + }) => NostrEvent( + id: id, + pubkey: 'bob', + createdAt: 30, + kind: kind, + tags: [ + if (inChannel) ['h', _channelId], + ['e', target], + ], + content: content, + sig: 'sig', + ); + final relaySession = _RecordingRelaySessionNotifier( + queryResults: [ + [_event(id: 'message', createdAt: 10), _bounds()], + ], + ); + relaySession.noticeResults.add([_joinNotice('joined', 20)]); + relaySession.auxEvents.addAll([ + aux('kept', EventKind.reaction, 'joined', content: '👍'), + aux('removed', EventKind.reaction, 'joined', content: '🎉'), + aux('unreacted', EventKind.reaction, 'joined', content: '👀'), + aux('removal', EventKind.deletion, 'removed'), + aux('channel-less', EventKind.deletion, 'unreacted', inChannel: false), + ]); + final container = _buildContainer(relaySession); + addTearDown(container.dispose); + + container.read(channelMessagesProvider(_channelId)); + await relaySession.subscribed; + await _pumpEventQueue(); + + // The reactions the page would have carried, then their deletions: each + // read in the channel, and for deletions with no channel. + final reads = relaySession.auxFilters; + expect( + [for (final read in reads) read.tags['#h']], + [ + [_channelId], + null, + [_channelId], + null, + ], + ); + expect(reads[0].tags['#e'], ['joined']); + expect( + reads[2].tags['#e'], + unorderedEquals(['kept', 'removed', 'unreacted']), + ); + for (final read in reads.skip(1)) { + expect(read.kinds, isNot(contains(EventKind.reaction))); + } + final joined = formatTimeline( + container.read(channelMessagesProvider(_channelId)).value!, + ).singleWhere((message) => message.id == 'joined'); + expect(joined.reactions.map((reaction) => reaction.emoji), ['👍']); + }); + + test('notice reactions are read past one page', () async { + final reactions = [ + for (var i = 0; i <= NostrFilters.channelAuxLimit; i++) + NostrEvent( + id: _paddedId('r${i.toString().padLeft(4, '0')}'), + // The last one is the oldest, and it is mine. + pubkey: i == NostrFilters.channelAuxLimit ? 'me' : 'user-$i', + createdAt: 2000 - i, + kind: EventKind.reaction, + tags: const [ + ['h', _channelId], + ['e', 'joined'], + ], + content: '👍', + sig: 'sig', + ), + ]; + final relaySession = _RecordingRelaySessionNotifier( + queryResults: [ + [_event(id: 'message', createdAt: 10), _bounds()], + ], + ); + relaySession.noticeResults.add([_joinNotice('joined', 20)]); + relaySession.auxEvents.addAll(reactions); + final container = _buildContainer(relaySession); + addTearDown(container.dispose); + + container.read(channelMessagesProvider(_channelId)); + await relaySession.subscribed; + await _pumpEventQueue(); + + final reactionReads = relaySession.auxFilters + .where((filter) => filter.kinds.contains(EventKind.reaction)) + .toList(); + expect(reactionReads, hasLength(2)); + expect(reactionReads[1].until, reactions[999].createdAt); + expect(reactionReads[1].extensions['before_id'], reactions[999].id); + final joined = formatTimeline( + container.read(channelMessagesProvider(_channelId)).value!, + currentPubkey: 'me', + ).singleWhere((message) => message.id == 'joined'); + expect(joined.reactions.single.count, reactions.length); + expect(joined.reactions.single.currentUserReactionId, reactions.last.id); + }); + + test('a notice keeps a reaction from below the oldest loaded row', () async { + const id = _paddedId; + // The reaction shares the oldest row's second but sorts after it, so it + // is below the floor while its notice is above it. + final relaySession = _RecordingRelaySessionNotifier( + queryResults: [ + [ + _event(id: id('88'), createdAt: 50), + _bounds(hasMore: true, cursorCreatedAt: 50, cursorId: id('88')), + ], + ], + ); + relaySession.noticeResults.add([_joinNotice(id('22'), 50)]); + relaySession.auxEvents.add( + NostrEvent( + id: id('cc'), + pubkey: 'bob', + createdAt: 50, + kind: EventKind.reaction, + tags: [ + const ['h', _channelId], + ['e', id('22')], + ], + content: '👍', + sig: 'sig', + ), + ); + final container = _buildContainer(relaySession); + addTearDown(container.dispose); + + container.read(channelMessagesProvider(_channelId)); + await relaySession.subscribed; + await _pumpEventQueue(); + + final notice = formatTimeline( + container.read(channelMessagesProvider(_channelId)).value!, + ).singleWhere((message) => message.id == id('22')); + expect(notice.reactions.map((reaction) => reaction.emoji), ['👍']); + }); + + test('notice aux reads name at most 500 targets each', () async { + final reactions = [ + for (var i = 0; i <= 500; i++) + NostrEvent( + id: _paddedId('r${i.toString().padLeft(4, '0')}'), + pubkey: 'user-$i', + createdAt: 2000 - i, + kind: EventKind.reaction, + tags: const [ + ['h', _channelId], + ['e', 'joined'], + ], + content: '👍', + sig: 'sig', + ), + ]; + final relaySession = _RecordingRelaySessionNotifier( + queryResults: [ + [_event(id: 'message', createdAt: 10), _bounds()], + ], + ); + relaySession.noticeResults.add([_joinNotice('joined', 20)]); + relaySession.auxEvents.addAll([ + ...reactions, + // Removes the reaction only the second batch names. + NostrEvent( + id: 'removal', + pubkey: 'user-500', + createdAt: 3000, + kind: EventKind.deletion, + tags: [ + const ['h', _channelId], + ['e', reactions.last.id], + ], + content: '', + sig: 'sig', + ), + ]); + final container = _buildContainer(relaySession); + addTearDown(container.dispose); + + container.read(channelMessagesProvider(_channelId)); + await relaySession.subscribed; + await _pumpEventQueue(); + + final deletionReads = relaySession.auxFilters.skip(2).toList(); + expect(deletionReads, hasLength(4)); + for (final read in deletionReads) { + expect(read.tags['#e']!.length, lessThanOrEqualTo(500)); + } + expect( + deletionReads.expand((read) => read.tags['#e']!).toSet(), + reactions.map((event) => event.id).toSet(), + ); + final joined = formatTimeline( + container.read(channelMessagesProvider(_channelId)).value!, + ).singleWhere((message) => message.id == 'joined'); + expect(joined.reactions.single.count, 500); + }); + + test( + 'a page that loads during a failed notice read gets its notices', + () async { + const id = _paddedId; + final relaySession = _RecordingRelaySessionNotifier( + queryResults: [ + [ + _event(id: id('a100'), createdAt: 100), + _bounds(hasMore: true, cursorCreatedAt: 100, cursorId: id('a100')), + ], + [ + _event(id: id('a10'), createdAt: 10), + _bounds(dTag: '${_channelId.toLowerCase()}:100:${id('a100')}'), + ], + ], + ); + final firstRead = Completer>(); + relaySession.noticeResults.addAll([ + firstRead.future, + [_joinNotice(id('n50'), 50)], + ]); + final container = _buildContainer(relaySession); + addTearDown(container.dispose); + + container.read(channelMessagesProvider(_channelId)); + await relaySession.subscribed; + await _pumpEventQueue(); + expect(relaySession.noticeFilters, hasLength(1)); + + final notifier = container.read( + channelMessagesProvider(_channelId).notifier, + ); + expect(await notifier.fetchOlder(), isTrue); + await _pumpEventQueue(); + expect(relaySession.noticeFilters, hasLength(1)); + + firstRead.completeError(Exception('notices unavailable')); + await _pumpEventQueue(); + expect(relaySession.noticeFilters, hasLength(2)); + expect( + container + .read(channelMessagesProvider(_channelId)) + .value! + .map((event) => event.id), + [id('a10'), id('n50'), id('a100')], + ); + }, + ); + + test('the notice descent stops at its cap', () async { + const id = _paddedId; + const pages = 5; + final relaySession = _RecordingRelaySessionNotifier( + queryResults: [ + [_event(id: id('a10'), createdAt: 10), _bounds()], + ], + ); + relaySession.noticeResults.addAll([ + for (var page = 0; page < pages; page++) + [ + for (var i = 0; i < NostrFilters.channelNoticePageSize; i++) + _joinNotice( + id('p$page-${i.toString().padLeft(3, '0')}'), + 10000 - page * NostrFilters.channelNoticePageSize - i, + ), + ], + ]); + final container = _buildContainer(relaySession); + addTearDown(container.dispose); + + container.read(channelMessagesProvider(_channelId)); + await relaySession.subscribed; + await _pumpEventQueue(); + + expect(relaySession.noticeFilters, hasLength(4)); + expect( + container.read(channelMessagesProvider(_channelId)).value, + hasLength(1 + 4 * NostrFilters.channelNoticePageSize), + ); + }); + test('retains historical and live Huddle end events in the window', () async { final relaySession = _RecordingRelaySessionNotifier( queryResults: [ @@ -5624,6 +6074,24 @@ NostrEvent _summary({ ); } +String _paddedId(String label) => label.padLeft(64, '0'); + +NostrEvent _joinNotice(String id, int createdAt) => NostrEvent( + id: id, + pubkey: 'relay', + createdAt: createdAt, + kind: EventKind.systemMessage, + tags: const [ + ['h', _channelId], + ], + content: jsonEncode({ + 'type': 'member_joined', + 'actor': 'alice', + 'target': id, + }), + sig: 'sig', +); + NostrEvent _bounds({ bool hasMore = false, int? cursorCreatedAt, @@ -5666,6 +6134,17 @@ class _RecordingRelaySessionNotifier extends RelaySessionNotifier { final List liveFilters = []; final List historyFilters = []; final List queryFilters = []; + + /// System notice reads, kept apart from [queryFilters]. They answer + /// from [noticeResults] in order, or empty once it runs out. + final List noticeFilters = []; + final Queue noticeResults = Queue(); + + /// Notice reaction and deletion reads ([NostrFilters.channelAux]), kept + /// apart from [queryFilters]. + /// They answer from [auxEvents] the way the relay does. + final List auxFilters = []; + final List auxEvents = []; final List _listeners = []; final Completer _subscribed = Completer(); final Completer> _history = Completer>(); @@ -5694,6 +6173,61 @@ class _RecordingRelaySessionNotifier extends RelaySessionNotifier { List filters, { Duration timeout = const Duration(seconds: 8), }) async { + if (filters.length == 1 && + filters.single.kinds.length == 1 && + filters.single.kinds.single == EventKind.systemMessage) { + noticeFilters.add(filters.single); + if (noticeResults.isEmpty) return const []; + final result = noticeResults.removeFirst(); + if (result is Exception) throw result; + if (result is Future>) return await result; + return (result as List).toList(); + } + final targets = filters.length == 1 ? filters.single.tags['#e'] : null; + if (targets != null && + filters.single.limit == NostrFilters.channelAuxLimit && + filters.single.kinds.every( + const { + EventKind.deletion, + EventKind.reaction, + EventKind.nip29DeleteEvent, + }.contains, + )) { + final filter = filters.single; + auxFilters.add(filter); + // As for a system channel (NIP-CL): a read sees the channel's events + // only when it names the channel in `#h`, and sees events with no + // channel only when it does not. + final named = filter.tags['#h']; + final until = filter.until; + final beforeId = filter.extensions['before_id'] as String?; + // Newest first, below the cursor, up to the limit, as the relay reads. + final matched = + [ + for (final event in auxEvents) + if (filter.kinds.contains(event.kind) && + (named == null + ? event.channelId == null + : named.contains(event.channelId)) && + (until == null || + event.createdAt < until || + (event.createdAt == until && + beforeId != null && + event.id.compareTo(beforeId) > 0)) && + event.tags.any( + (tag) => + tag.length >= 2 && + tag[0] == 'e' && + targets.contains(tag[1]), + )) + event, + ]..sort( + (a, b) => b.createdAt != a.createdAt + ? b.createdAt.compareTo(a.createdAt) + : a.id.compareTo(b.id), + ); + return matched.take(filter.limit).toList(); + } operations.add('query'); queryFilters.addAll(filters); if (_queryResults.isEmpty) throw Exception('unsupported');