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');