diff --git a/mobile/lib/features/channels/channel_messages_provider/notices.dart b/mobile/lib/features/channels/channel_messages_provider/notices.dart index 0740ac7269..1afecdc3d9 100644 --- a/mobile/lib/features/channels/channel_messages_provider/notices.dart +++ b/mobile/lib/features/channels/channel_messages_provider/notices.dart @@ -64,30 +64,52 @@ extension _ChannelNotices on ChannelMessagesNotifier { !isChannelCursorOlder(floor, cursor)) { break; } - final events = await session.queryRelay([ + final page = await session.queryRelay([ NostrFilters.channelNotices( channelId, + since: floor?.createdAt, until: cursor?.createdAt, beforeId: cursor?.eventId, ), ]); if (!_isCurrentInit(initVersion)) return; - if (events.any( + if (page.any( (event) => event.kind != EventKind.systemMessage || event.channelId != channelId, )) { throw Exception('System notices are outside their channel.'); } + // Short bounded ranges can leave a partial page's worth of room. + // Count unique additions, keep the newest, and exclude overflow + // before fetching auxiliary events. Cached notices and their aux + // already come from _windowEvents when an older page reveals them. + final fresh = + { + for (final event in page) + if (!_notices.containsKey(event.id)) event.id: event, + }.values.toList()..sort( + (a, b) => compareChannelTimelineEventsChronologically(b, a), + ); + final events = fresh.take(_maxNotices - _notices.length).toList(); // 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; + if (page.length < NostrFilters.channelNoticePageSize) { + if (floor == null) { + _noticesReachedStart = true; + } else { + // Only this range is exhausted. Resume below its message floor + // when older pages arrive, even if it was empty. Same-second + // notices below the floor are retained and may be read again; + // merging by id keeps them unique. Use the captured floor: a + // page may have extended the window while this read was in flight. + _noticeCursor = floor; + } } else { - _noticeCursor = _nextCursor(events, cursor, 'System notices'); + _noticeCursor = _nextCursor(page, cursor, 'System notices'); } for (final event in events) { _notices[event.id] = event; diff --git a/mobile/lib/shared/relay/nostr_filters.dart b/mobile/lib/shared/relay/nostr_filters.dart index 3f97060b80..c640fc4cb5 100644 --- a/mobile/lib/shared/relay/nostr_filters.dart +++ b/mobile/lib/shared/relay/nostr_filters.dart @@ -50,9 +50,11 @@ abstract final class NostrFilters { ); /// A channel's system notices (kind:40099), newest first, below the - /// `until` + `before_id` cursor the channel-window pages use. + /// `until` + `before_id` cursor the channel-window pages use. [since] + /// includes the oldest loaded row's whole second, including id ties. static NostrFilter channelNotices( String channelId, { + int? since, int? until, String? beforeId, }) => NostrFilter( @@ -61,6 +63,7 @@ abstract final class NostrFilters { '#h': [channelId], }, limit: channelNoticePageSize, + since: since, until: until, extensions: {'before_id': ?beforeId}, ); diff --git a/mobile/test/features/channels/channel_messages_provider_test.dart b/mobile/test/features/channels/channel_messages_provider_test.dart index 6bcd1aa2d9..5aa51a689a 100644 --- a/mobile/test/features/channels/channel_messages_provider_test.dart +++ b/mobile/test/features/channels/channel_messages_provider_test.dart @@ -514,6 +514,8 @@ void main() { }); test('notices read apart keep their reactions', () async { + const id = _paddedId; + final joinedId = id('20'); NostrEvent aux( String id, int kind, @@ -534,14 +536,24 @@ void main() { ); final relaySession = _RecordingRelaySessionNotifier( queryResults: [ - [_event(id: 'message', createdAt: 10), _bounds()], + [ + _event(id: id('10'), createdAt: 50), + _bounds(hasMore: true, cursorCreatedAt: 50, cursorId: id('10')), + ], + [ + _event(id: id('30'), createdAt: 50), + _bounds(dTag: '${_channelId.toLowerCase()}:50:${id('10')}'), + ], ], ); - relaySession.noticeResults.add([_joinNotice('joined', 20)]); + relaySession.noticeResults.addAll([ + [_joinNotice(joinedId, 50)], + [_joinNotice(joinedId, 50)], + ]); relaySession.auxEvents.addAll([ - aux('kept', EventKind.reaction, 'joined', content: '👍'), - aux('removed', EventKind.reaction, 'joined', content: '🎉'), - aux('unreacted', EventKind.reaction, 'joined', content: '👀'), + aux('kept', EventKind.reaction, joinedId, content: '👍'), + aux('removed', EventKind.reaction, joinedId, content: '🎉'), + aux('unreacted', EventKind.reaction, joinedId, content: '👀'), aux('removal', EventKind.deletion, 'removed'), aux('channel-less', EventKind.deletion, 'unreacted', inChannel: false), ]); @@ -564,7 +576,7 @@ void main() { null, ], ); - expect(reads[0].tags['#e'], ['joined']); + expect(reads[0].tags['#e'], [joinedId]); expect( reads[2].tags['#e'], unorderedEquals(['kept', 'removed', 'unreacted']), @@ -572,10 +584,43 @@ void main() { for (final read in reads.skip(1)) { expect(read.kinds, isNot(contains(EventKind.reaction))); } + // The notice is cached below the current same-second message floor. + expect( + container + .read(channelMessagesProvider(_channelId)) + .requireValue + .where((event) => event.id == joinedId), + isEmpty, + ); + relaySession.emit(aux('live', EventKind.reaction, joinedId, content: '🔥')); + final notifier = container.read( + channelMessagesProvider(_channelId).notifier, + ); + expect(await notifier.fetchOlder(), isTrue); + await _pumpEventQueue(); + expect( + reads, + hasLength(4), + reason: 'cached notices need no auxiliary reread', + ); final joined = formatTimeline( container.read(channelMessagesProvider(_channelId)).value!, - ).singleWhere((message) => message.id == 'joined'); - expect(joined.reactions.map((reaction) => reaction.emoji), ['👍']); + ).singleWhere((message) => message.id == joinedId); + expect( + joined.reactions.map((reaction) => reaction.emoji), + unorderedEquals(['👍', '🔥']), + ); + relaySession.emit(aux('live-removal', EventKind.deletion, 'kept')); + await _pumpEventQueue(); + expect( + formatTimeline( + container.read(channelMessagesProvider(_channelId)).requireValue, + ) + .singleWhere((message) => message.id == joinedId) + .reactions + .map((reaction) => reaction.emoji), + ['🔥'], + ); }); test('notice reactions are read past one page', () async { @@ -723,54 +768,102 @@ void main() { expect(joined.reactions.single.count, 500); }); - test( - 'a page that loads during a failed notice read gets its notices', - () async { - const id = _paddedId; + for (final rowId in [_paddedId('50'), 'z-row']) { + test('an empty notice range stops at its message floor: $rowId', () async { 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')}'), + _event(id: rowId, createdAt: 50), + _bounds( + hasMore: true, + cursorCreatedAt: 50, + cursorId: _paddedId('50'), + ), ], ], ); - final firstRead = Completer>(); relaySession.noticeResults.addAll([ - firstRead.future, - [_joinNotice(id('n50'), 50)], + [], + // Bound a broken implementation too, so a repeated empty-range read + // fails the assertion instead of exhausting the test process's heap. + Exception('unexpected repeated notice range'), ]); 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')], + .requireValue + .map((e) => e.id), + [rowId], ); - }, - ); + }); + } + + for (final fails in [false, true]) { + test( + 'a page that loads during a notice read gets its notices: fails=$fails', + () 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)); + + if (fails) { + firstRead.completeError(Exception('notices unavailable')); + } else { + firstRead.complete([]); + } + await _pumpEventQueue(); + expect(relaySession.noticeFilters, hasLength(2)); + expect(relaySession.noticeFilters.first.since, 100); + expect(relaySession.noticeFilters.last.since, isNull); + 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; @@ -804,6 +897,97 @@ void main() { ); }); + for (final overlaps in [false, true]) { + test( + 'mixed notice ranges respect the unique cap: overlaps=$overlaps', + () async { + const id = _paddedId; + final shortPages = [ + for (var page = 0; page < 4; page++) + [ + for (var i = 0; i < 499; i++) + _joinNotice( + id('p$page-${i.toString().padLeft(3, '0')}'), + page == 3 ? 2000 : 6000 - page * 1000 - i, + ), + ], + ]; + final tail = [ + for (var i = 0; i < (overlaps ? 498 : 500); i++) + _joinNotice(id('tail-${i.toString().padLeft(3, '0')}'), 1999 - i), + ]; + final relaySession = _RecordingRelaySessionNotifier( + queryResults: [ + for (var page = 0; page < 6; page++) + [ + _event(id: id('a$page'), createdAt: 5000 - page * 1000), + _bounds( + hasMore: page < 5, + cursorCreatedAt: 5000 - page * 1000, + cursorId: id('a$page'), + dTag: page == 0 + ? null + : '${_channelId.toLowerCase()}:${6000 - page * 1000}:${id('a${page - 1}')}', + ), + ], + ], + ); + relaySession.noticeResults.addAll([ + ...shortPages, + [if (overlaps) ...shortPages.last.take(2), ...tail], + ]); + final container = _buildContainer(relaySession); + addTearDown(container.dispose); + container.read(channelMessagesProvider(_channelId)); + await relaySession.subscribed; + await _pumpEventQueue(); + final notifier = container.read( + channelMessagesProvider(_channelId).notifier, + ); + for (var page = 1; page < 5; page++) { + expect(await notifier.fetchOlder(), isTrue); + await _pumpEventQueue(); + } + final retained = container + .read(channelMessagesProvider(_channelId)) + .requireValue + .where((event) => event.kind == EventKind.systemMessage) + .toList(); + expect(retained, hasLength(2000)); + expect(retained.map((event) => event.id).toSet(), hasLength(2000)); + expect( + retained.map((event) => event.id), + containsAll(tail.take(4).map((e) => e.id)), + ); + final discarded = tail.skip(4).map((e) => e.id).toSet(); + expect( + retained.map((event) => event.id), + isNot(anyElement(isIn(discarded))), + ); + expect( + relaySession.auxFilters.expand( + (filter) => filter.tags['#e'] ?? const [], + ), + isNot(anyElement(isIn(discarded))), + reason: + 'excess notices must not fetch reactions or deletion metadata', + ); + final reads = relaySession.noticeFilters.length; + expect(await notifier.fetchOlder(), isTrue); + await _pumpEventQueue(); + expect(relaySession.noticeFilters, hasLength(reads)); + expect( + container + .read(channelMessagesProvider(_channelId)) + .requireValue + .map((event) => event.id), + contains(id('a5')), + reason: 'reaching the notice cap must not stop message paging', + ); + }, + ); + } + test('retains historical and live Huddle end events in the window', () async { final relaySession = _RecordingRelaySessionNotifier( queryResults: [ diff --git a/mobile/test/features/channels/channel_notices_journey_test.dart b/mobile/test/features/channels/channel_notices_journey_test.dart new file mode 100644 index 0000000000..192b5d4c04 --- /dev/null +++ b/mobile/test/features/channels/channel_notices_journey_test.dart @@ -0,0 +1,281 @@ +import 'dart:convert'; + +import 'package:buzz/features/channels/channel.dart'; +import 'package:buzz/features/channels/channel_detail_page.dart'; +import 'package:buzz/features/channels/channel_messages_provider.dart'; +import 'package:buzz/shared/relay/relay.dart'; +import 'package:buzz/shared/theme/theme.dart'; +import 'package:crypto/crypto.dart'; +import 'package:fake_async/fake_async.dart'; +import 'package:flutter/material.dart'; +import 'package:flutter_test/flutter_test.dart'; +import 'package:hooks_riverpod/hooks_riverpod.dart'; +import 'package:shared_preferences/shared_preferences.dart'; + +import '../../helpers/fake_relay.dart'; + +const _channelId = '11111111-2222-4333-8444-555555555555'; + +// History fixtures cross the fake network boundary; signing hundreds of +// fixtures adds no paging coverage. Auth and window bounds still use FakeRelay. +NostrEvent _message(FakeRelay relay, int second) => NostrEvent( + id: '8${second.toRadixString(16).padLeft(63, '0')}', + pubkey: relay.pubkey, + kind: EventKind.streamMessage, + content: 'Message $second', + createdAt: second, + sig: '0' * 128, + tags: const [ + ['h', _channelId], + ], +); + +NostrEvent _notice(FakeRelay relay, int second, String topic) => NostrEvent( + id: sha256.convert(utf8.encode('$second:$topic')).toString(), + pubkey: relay.pubkey, + kind: EventKind.systemMessage, + content: jsonEncode({'type': 'topic_changed', 'topic': topic}), + createdAt: second, + sig: '0' * 128, + tags: const [ + ['h', _channelId], + ], +); + +Iterable> _noticeReads(FakeRelay relay) => relay.queries + .expand((filters) => filters) + .where((filter) => (filter['kinds'] as List).singleOrNull == 40099); + +void main() { + // The real notifier and session issue HTTP filters; only the network is + // fake. The fixture applies since, the composite cursor and the limit, so + // removing a production bound or skipping a range changes the result. + for (final count in [0, 1, 500, 501]) { + test('notice ranges page through $count boundary-second notices', () { + fakeAsync((clock) { + final relay = FakeRelay(); + final messages = [for (var i = 1; i <= 151; i++) _message(relay, i)]; + final boundary = [ + for (var i = 0; i < count; i++) _notice(relay, 102, 'Boundary $i'), + ]; + final burst = [ + if (count == 501) + for (var i = 0; i < 501; i++) _notice(relay, 103, 'Burst $i'), + ]; + final middle = _notice(relay, 75, 'Middle'); + final oldest = _notice(relay, 0, 'Before the first message'); + relay.seed([...messages, ...burst, ...boundary, middle, oldest]); + final container = ProviderContainer(overrides: relay.overrides); + addTearDown(container.dispose); + final subscription = container.listen( + channelMessagesProvider(_channelId), + (_, _) {}, + ); + addTearDown(subscription.close); + void settle() => clock.elapse(const Duration(milliseconds: 50)); + List shown() => subscription.read().requireValue; + void checkIds(Iterable expected) { + expect( + shown().map((event) => event.id), + unorderedEquals(expected.map((event) => event.id)), + ); + expect( + shown().map((event) => event.id).toSet(), + hasLength(shown().length), + ); + } + + final notifier = container.read( + channelMessagesProvider(_channelId).notifier, + ); + void older() { + var completed = false; + notifier.fetchOlder().then((loaded) { + expect(loaded, isTrue); + completed = true; + }); + settle(); + expect(completed, isTrue); + } + + settle(); + expect(_noticeReads(relay).first['since'], 102); + if (burst.isNotEmpty) { + expect(_noticeReads(relay), hasLength(2)); + expect(_noticeReads(relay).last['since'], 102); + } + checkIds([ + ...messages.skip(101), + ...burst, + ...boundary.where( + (notice) => notice.id.compareTo(messages[101].id) <= 0, + ), + ]); + + // A short OR empty bounded response must not finish all history. + // The previous boundary's hidden same-second notices now show too. + older(); + checkIds([...messages.skip(51), ...burst, ...boundary, middle]); + expect(_noticeReads(relay).last['since'], 52); + + // This range has no notices. It still advances, without rereading + // the newest range or abandoning the notices before message history. + older(); + checkIds([...messages.skip(1), ...burst, ...boundary, middle]); + expect(_noticeReads(relay).last['since'], 2); + older(); + expect(_noticeReads(relay).last['since'], isNull); + checkIds([...messages, ...burst, ...boundary, middle, oldest]); + expect(notifier.reachedOldest, isTrue); + final reads = relay.queries.length; + notifier.fetchOlder().then((loaded) => expect(loaded, isFalse)); + settle(); + expect(relay.queries, hasLength(reads)); + }); + }); + } + + test('same-second message pages reveal cached notices without a gap', () { + fakeAsync((clock) { + final relay = FakeRelay(); + final messages = [ + for (var i = 1; i <= 120; i++) + NostrEvent.fromJson({ + ..._message(relay, i).toJson(), + 'created_at': 50, + 'id': '${i.toRadixString(16).padLeft(2, '0')}${'0' * 62}', + }), + ]; + final notice = NostrEvent.fromJson({ + ..._notice(relay, 50, 'Between equal-second messages').toJson(), + // Between the 50th (0x32) and 51st (0x33) message in relay order. + 'id': '328${'0' * 61}', + }); + final beforeMessages = _notice(relay, 0, 'Before all messages'); + relay.seed([...messages, notice, beforeMessages]); + final container = ProviderContainer(overrides: relay.overrides); + addTearDown(container.dispose); + final subscription = container.listen( + channelMessagesProvider(_channelId), + (_, _) {}, + ); + addTearDown(subscription.close); + void settle() => clock.elapse(const Duration(milliseconds: 50)); + Iterable ids() => + subscription.read().requireValue.map((e) => e.id); + settle(); + expect(ids(), hasLength(50)); + expect(ids(), isNot(contains(notice.id))); + expect(_noticeReads(relay).single['since'], 50); + + final notifier = container.read( + channelMessagesProvider(_channelId).notifier, + ); + notifier.fetchOlder().then((loaded) => expect(loaded, isTrue)); + settle(); + expect(ids(), hasLength(101)); + expect(ids().where((id) => id == notice.id), hasLength(1)); + expect(_noticeReads(relay), hasLength(2)); + expect(_noticeReads(relay).last['since'], 50); + notifier.fetchOlder().then((loaded) => expect(loaded, isTrue)); + settle(); + expect(ids(), hasLength(122)); + expect(ids(), contains(beforeMessages.id)); + expect(_noticeReads(relay).last['since'], isNull); + }); + }); + + for (final hasMessage in [false, true]) { + test('channel start retains earlier notices: message=$hasMessage', () { + fakeAsync((clock) { + final relay = FakeRelay(); + final notice = _notice(relay, 0, 'First activity'); + relay.seed([if (hasMessage) _message(relay, 10), notice]); + final container = ProviderContainer(overrides: relay.overrides); + addTearDown(container.dispose); + final subscription = container.listen( + channelMessagesProvider(_channelId), + (_, _) {}, + ); + addTearDown(subscription.close); + clock.elapse(const Duration(milliseconds: 50)); + expect( + subscription.read().requireValue.map((event) => event.id), + contains(notice.id), + ); + expect(_noticeReads(relay).single['since'], isNull); + }); + }); + } + + testWidgets('scrolling back recovers notices after a failed read and ' + 'interleaves them with older messages', (tester) async { + SharedPreferences.setMockInitialValues({}); + final prefs = await SharedPreferences.getInstance(); + tester.view.physicalSize = const Size(390, 844); + tester.view.devicePixelRatio = 1; + addTearDown(tester.view.reset); + final relay = FakeRelay(); + final notice = _notice(relay, 75, 'Older topic'); + relay.seed([for (var i = 1; i <= 151; i++) _message(relay, i), notice]); + relay.failQuery( + (filter) => (filter['kinds'] as List).singleOrNull == 40099, + ); + await tester.pumpWidget( + ProviderScope( + overrides: [ + ...relay.overrides, + savedPrefsProvider.overrideWithValue(prefs), + appLifecycleProvider.overrideWith(_Resumed.new), + ], + child: MaterialApp( + theme: AppTheme.light(), + home: ChannelDetailPage( + channel: Channel( + id: _channelId, + name: 'general', + description: 'General discussion', + channelType: 'stream', + visibility: 'open', + createdBy: relay.pubkey, + createdAt: DateTime(2025), + memberCount: 1, + isMember: true, + ), + ), + ), + ), + ); + await tester.pumpAndSettle(); + expect(find.text('Message 151'), findsOneWidget); + expect(find.textContaining('Older topic'), findsNothing); + + // Drive the actual timeline's older-page trigger with scroll gestures. + for ( + var i = 0; + i < 30 && find.textContaining('Older topic').evaluate().isEmpty; + i++ + ) { + await tester.drag(find.byType(Scrollable).first, const Offset(0, 650)); + await tester.pumpAndSettle(); + } + expect(find.textContaining('Older topic'), findsOneWidget); + final container = ProviderScope.containerOf( + tester.element(find.byType(ChannelDetailPage)), + ); + final events = container + .read(channelMessagesProvider(_channelId)) + .requireValue; + final position = events.indexWhere((event) => event.id == notice.id); + expect(position, greaterThan(0)); + expect(events[position - 1].createdAt, lessThanOrEqualTo(75)); + expect(events[position + 1].createdAt, greaterThanOrEqualTo(75)); + expect(events.where((event) => event.id == notice.id), hasLength(1)); + expect(_noticeReads(relay).map((filter) => filter['since']), contains(52)); + }); +} + +class _Resumed extends AppLifecycleNotifier { + @override + AppLifecycleState build() => AppLifecycleState.resumed; +}