diff --git a/mobile/lib/features/channels/channel_messages_provider.dart b/mobile/lib/features/channels/channel_messages_provider.dart index fbcbca8956..e2cab7ddc8 100644 --- a/mobile/lib/features/channels/channel_messages_provider.dart +++ b/mobile/lib/features/channels/channel_messages_provider.dart @@ -254,7 +254,11 @@ class ChannelMessagesNotifier extends Notifier>> { ), ); } - if (!_isBroadcastReply(event)) return false; + // Replies are kept in the store rather than dropped here, matching + // desktop: the main timeline filters them out at render + // (`buildMainTimelineEntries`), and their parent's "N replies" row needs + // them as the local half of the summary merge when the relay's + // best-effort recount is delayed, lost, or older than this reply. } // Thread summaries are neither a timeline row nor an aux event, but they are // how the root's "N replies" row learns a reply landed — a reply itself @@ -495,12 +499,6 @@ class ChannelMessagesNotifier extends Notifier>> { } } -bool _isBroadcastReply(NostrEvent event) { - return event.tags.any( - (tag) => tag.length >= 2 && tag[0] == 'broadcast' && tag[1] == '1', - ); -} - int _currentUnixSeconds() => DateTime.now().millisecondsSinceEpoch ~/ 1000; final channelMessagesProvider = diff --git a/mobile/lib/features/channels/timeline_message.dart b/mobile/lib/features/channels/timeline_message.dart index c8fd96491b..411902ea86 100644 --- a/mobile/lib/features/channels/timeline_message.dart +++ b/mobile/lib/features/channels/timeline_message.dart @@ -509,13 +509,9 @@ List buildMainTimelineEntries( List messages, { Map? relaySummaries, }) { - // Index direct children by parentId. - final childrenByParent = >{}; - for (final msg in messages) { - final pid = msg.parentId; - if (pid == null) continue; - childrenByParent.putIfAbsent(pid, () => []).add(msg); - } + // Index descendant stats by ancestor, so a nested reply updates every summary + // above it and not only the summary of its direct parent. + final descendantStats = _buildDescendantStats(messages); return [ for (final msg in messages) @@ -524,7 +520,7 @@ List buildMainTimelineEntries( message: msg, summary: _buildSummary( msg.id, - childrenByParent, + descendantStats, relaySummaries?[msg.id], ), ), @@ -537,39 +533,176 @@ bool _isBroadcastReply(TimelineMessage message) { ); } +/// Combine what the relay counted with what this client has actually seen. +/// +/// The count and last-reply time follow the desktop's `mergeThreadSummaries` +/// (`desktop/src/features/messages/lib/threadPanel.ts`): the relay recount is +/// authoritative for replies this client never loaded, and the locally observed +/// replies are authoritative for anything that landed after (or alongside) the +/// last recount. Neither source alone is complete, so take the larger count and +/// the later reply time rather than letting one shadow the other. The facepile +/// order is mobile's own (see [_mergeParticipants]) because this file already +/// renders relay participants in the order the relay sent them. +/// +/// Both halves count descendants, not direct replies: the relay's +/// `descendant_count` and the locally assembled [_buildDescendantStats]. A badge +/// on the main timeline stands for the whole thread under that message, so a +/// reply to a reply has to raise it. ThreadSummary? _buildSummary( String messageId, - Map> childrenByParent, + Map descendantStats, ChannelWindowThreadSummary? relaySummary, ) { - if (relaySummary != null && relaySummary.replyCount > 0) { - return ThreadSummary( - threadHeadId: messageId, - replyCount: relaySummary.replyCount, - participantPubkeys: relaySummary.participantPubkeys.take(3).toList(), - lastReplyAt: relaySummary.lastReplyAt, - ); - } + final local = _buildLocalSummary(messageId, descendantStats); + final relay = _buildRelaySummary(messageId, relaySummary); + if (relay == null) return local; + if (local == null) return relay; - final replies = childrenByParent[messageId]; - if (replies == null || replies.isEmpty) return null; + return ThreadSummary( + threadHeadId: messageId, + replyCount: local.replyCount > relay.replyCount + ? local.replyCount + : relay.replyCount, + // Relay participants first: they describe the whole thread, including + // replies this client never loaded, so a recount's facepile keeps rendering + // as it does today. Locally seen repliers (newest first, matching the relay + // order this file already renders) only fill the remaining slots. + participantPubkeys: _mergeParticipants( + relay.participantPubkeys, + local.participantPubkeys.reversed, + ), + lastReplyAt: _laterOf(local.lastReplyAt, relay.lastReplyAt), + ); +} - // Up to 3 most recent unique participants (walk backwards). - final seen = {}; - final participants = []; - for (var i = replies.length - 1; i >= 0 && participants.length < 3; i--) { - final pk = replies[i].pubkey.toLowerCase(); - if (seen.add(pk)) participants.add(pk); +/// Summary assembled from the replies present in the loaded timeline. +/// +/// Counts every loaded descendant, not only direct children, because the root +/// badge in the main timeline stands for the whole thread. This mirrors the +/// desktop's `buildSummaryForDirectReplies`, which reads the same descendant +/// stats and reverses the newest-first participants to oldest-first. +ThreadSummary? _buildLocalSummary( + String messageId, + Map descendantStats, +) { + final stats = descendantStats[messageId]; + if (stats == null || stats.descendantCount == 0) return null; + + return ThreadSummary( + threadHeadId: messageId, + replyCount: stats.descendantCount, + participantPubkeys: stats.recentParticipantsNewestFirst.reversed.toList(), + lastReplyAt: stats.lastReplyAt, + ); +} + +/// Descendant count, last reply time, and recent participants for every message +/// that has at least one loaded descendant. +/// +/// Mirrors the desktop's `buildDescendantStatsByMessageId` +/// (`desktop/src/features/messages/lib/threadPanel.ts`): each message is +/// attributed to every ancestor on its parent chain, so a reply nested under a +/// reply still counts towards the root it belongs to. Messages are visited +/// newest first so the capped participant list keeps the most recent repliers. +Map _buildDescendantStats( + List messages, +) { + final messageById = { + for (final msg in messages) msg.id: msg, + }; + final statsByMessageId = { + for (final msg in messages) msg.id: _DescendantStats(), + }; + + // Oldest first, keeping the original order for messages sharing a timestamp, + // then walked in reverse so participants are collected newest first. + final ordered = List.generate(messages.length, (index) => index) + ..sort((left, right) { + final byCreatedAt = messages[left].createdAt.compareTo( + messages[right].createdAt, + ); + return byCreatedAt != 0 ? byCreatedAt : left.compareTo(right); + }); + + for (var i = ordered.length - 1; i >= 0; i--) { + final message = messages[ordered[i]]; + final participant = message.pubkey.toLowerCase(); + + // Cap the walk so a malformed parent chain (a cycle, for instance) cannot + // spin forever. + var ancestorId = message.parentId; + var hops = 0; + final maxHops = messages.length + 1; + + while (ancestorId != null && hops < maxHops) { + final ancestorStats = statsByMessageId[ancestorId]; + if (ancestorStats == null) break; + + ancestorStats.descendantCount += 1; + ancestorStats.lastReplyAt = _laterOf( + ancestorStats.lastReplyAt, + message.createdAt, + ); + if (ancestorStats.recentParticipantsNewestFirst.length < 3 && + !ancestorStats.recentParticipantsNewestFirst.contains(participant)) { + ancestorStats.recentParticipantsNewestFirst.add(participant); + } + + ancestorId = messageById[ancestorId]?.parentId; + hops += 1; + } } + return statsByMessageId; +} + +/// Mutable accumulator for [_buildDescendantStats]. +class _DescendantStats { + int descendantCount = 0; + int? lastReplyAt; + final List recentParticipantsNewestFirst = []; +} + +/// Summary from the relay's recount, or null when it reports no replies. +ThreadSummary? _buildRelaySummary( + String messageId, + ChannelWindowThreadSummary? relaySummary, +) { + if (relaySummary == null || relaySummary.descendantCount <= 0) return null; return ThreadSummary( threadHeadId: messageId, - replyCount: replies.length, - participantPubkeys: participants.reversed.toList(), - lastReplyAt: replies.last.createdAt, + replyCount: relaySummary.descendantCount, + participantPubkeys: relaySummary.participantPubkeys.take(3).toList(), + lastReplyAt: relaySummary.lastReplyAt, ); } +int? _laterOf(int? left, int? right) { + if (left == null) return right; + if (right == null) return left; + return left > right ? left : right; +} + +/// Up to 3 unique pubkeys, [primary] first. +/// +/// [secondary] is expected newest-first so that a capped facepile keeps the +/// most recent participants rather than the oldest ones. Uniqueness is +/// case-insensitive because the locally assembled half lowercases pubkeys while +/// the relay half is passed through as received. +List _mergeParticipants( + Iterable primary, + Iterable secondary, +) { + final seen = {}; + final merged = []; + for (final pubkey in [...primary, ...secondary]) { + if (!seen.add(pubkey.toLowerCase())) continue; + merged.add(pubkey); + if (merged.length == 3) break; + } + return merged; +} + class _Edit { final String content; final int createdAt; diff --git a/mobile/test/features/channels/channel_messages_provider_test.dart b/mobile/test/features/channels/channel_messages_provider_test.dart index 4f4efdce1f..cbeb00182c 100644 --- a/mobile/test/features/channels/channel_messages_provider_test.dart +++ b/mobile/test/features/channels/channel_messages_provider_test.dart @@ -7,6 +7,7 @@ import 'package:hooks_riverpod/hooks_riverpod.dart'; import 'package:buzz/features/channels/channel_messages_provider.dart'; import 'package:buzz/features/channels/pending_local_messages_provider.dart'; import 'package:buzz/features/channels/thread_replies_provider.dart'; +import 'package:buzz/features/channels/timeline_message.dart'; import 'package:buzz/shared/relay/relay.dart'; void main() { @@ -572,6 +573,106 @@ void main() { }, ); + test( + 'a live reply reaches the store so its parent badge can count it', + () async { + final relaySession = _RecordingRelaySessionNotifier( + queryResults: [ + [_event(id: 'root', createdAt: 10), _bounds()], + ], + ); + final container = _buildContainer(relaySession); + addTearDown(container.dispose); + + container.read(channelMessagesProvider(_channelId)); + await relaySession.subscribed; + await _pumpEventQueue(); + + relaySession.emit( + _event( + id: 'reply', + createdAt: 20, + extraTags: const [ + ['e', 'root', '', 'reply'], + ], + ), + ); + await _pumpEventQueue(); + + // The reply is retained as the local half of the summary merge. It is + // filtered out of the main timeline by `buildMainTimelineEntries`, which + // owns reply visibility. + expect( + container + .read(channelMessagesProvider(_channelId)) + .value + ?.map((event) => event.id), + ['root', 'reply'], + ); + expect( + buildMainTimelineEntries( + formatTimeline( + container.read(channelMessagesProvider(_channelId)).value!, + ), + relaySummaries: container + .read(channelMessagesProvider(_channelId).notifier) + .threadSummaries, + ).map((entry) => entry.message.id), + ['root'], + ); + }, + ); + + test('a reply newer than the relay recount raises the badge', () async { + final relaySession = _RecordingRelaySessionNotifier( + queryResults: [ + [_event(id: 'root', createdAt: 10), _bounds()], + ], + ); + final container = _buildContainer(relaySession); + addTearDown(container.dispose); + + container.read(channelMessagesProvider(_channelId)); + await relaySession.subscribed; + await _pumpEventQueue(); + + relaySession.emit( + _event( + id: 'reply-1', + createdAt: 20, + extraTags: const [ + ['e', 'root', '', 'reply'], + ], + ), + ); + relaySession.emit(_summary(rootId: 'root', replyCount: 1, createdAt: 20)); + // A second reply lands, and its recount is lost or still in flight. + relaySession.emit( + _event( + id: 'reply-2', + createdAt: 21, + extraTags: const [ + ['e', 'root', '', 'reply'], + ], + ), + ); + await _pumpEventQueue(); + + final notifier = container.read( + channelMessagesProvider(_channelId).notifier, + ); + expect(notifier.threadSummaries['root']?.replyCount, 1); + final entries = buildMainTimelineEntries( + formatTimeline( + container.read(channelMessagesProvider(_channelId)).value!, + ), + relaySummaries: notifier.threadSummaries, + ); + expect(entries.single.message.id, 'root'); + expect(entries.single.summary!.replyCount, 2); + expect(entries.single.summary!.lastReplyAt, 21); + }); + test('window pagination failures return false without exhausting', () async { final relaySession = _RecordingRelaySessionNotifier( queryResults: [ diff --git a/mobile/test/features/channels/timeline_message_test.dart b/mobile/test/features/channels/timeline_message_test.dart index c29a12aef4..3dfaf184c1 100644 --- a/mobile/test/features/channels/timeline_message_test.dart +++ b/mobile/test/features/channels/timeline_message_test.dart @@ -1,6 +1,7 @@ import 'dart:convert'; import 'package:flutter_test/flutter_test.dart'; +import 'package:buzz/features/channels/channel_window.dart'; import 'package:buzz/features/channels/timeline_message.dart'; import 'package:buzz/shared/relay/relay.dart'; @@ -826,17 +827,23 @@ void main() { expect(entries[0].summary!.lastReplyAt, 3000); }); - test('summary counts only direct children, not nested replies', () { - final messages = formatTimeline([ - _textMsg(id: 'a', createdAt: 1000), - _replyMsg(id: 'r1', parentId: 'a', createdAt: 2000), - _replyMsg(id: 'r2', parentId: 'r1', rootId: 'a', createdAt: 3000), - ]); - - final entries = buildMainTimelineEntries(messages); - // Only r1 is a direct child of a; r2 is a child of r1. - expect(entries[0].summary!.replyCount, 1); - }); + test( + 'summary counts every loaded descendant, not only direct children', + () { + final messages = formatTimeline([ + _textMsg(id: 'a', createdAt: 1000), + _replyMsg(id: 'r1', parentId: 'a', createdAt: 2000), + _replyMsg(id: 'r2', parentId: 'r1', rootId: 'a', createdAt: 3000), + ]); + + final entries = buildMainTimelineEntries(messages); + // The root badge stands for the whole thread, so the reply to r1 counts + // towards 'a' as well. This matches the relay's `descendant_count` and + // the desktop's `buildDescendantStatsByMessageId`. + expect(entries[0].summary!.replyCount, 2); + expect(entries[0].summary!.lastReplyAt, 3000); + }, + ); test('summary has up to 3 unique participant pubkeys', () { final messages = formatTimeline([ @@ -877,6 +884,262 @@ void main() { }); }); + group('buildMainTimelineEntries relay summary merge', () { + ChannelWindowThreadSummary relaySummary({ + required int replyCount, + int? descendantCount, + int? lastReplyAt, + List participantPubkeys = const ['zoe'], + }) => ChannelWindowThreadSummary( + replyCount: replyCount, + descendantCount: descendantCount ?? replyCount, + lastReplyAt: lastReplyAt, + participantPubkeys: participantPubkeys, + ); + + test('relay recount covers replies this client never loaded', () { + final messages = formatTimeline([_textMsg(id: 'a', createdAt: 1000)]); + + final entries = buildMainTimelineEntries( + messages, + relaySummaries: {'a': relaySummary(replyCount: 4, lastReplyAt: 9000)}, + ); + + expect(entries.single.summary!.replyCount, 4); + expect(entries.single.summary!.lastReplyAt, 9000); + expect(entries.single.summary!.participantPubkeys, ['zoe']); + }); + + test('a reply newer than the recount is added to the badge', () { + final messages = formatTimeline([ + _textMsg(id: 'a', createdAt: 1000), + _replyMsg(id: 'r1', parentId: 'a', pubkey: 'bob', createdAt: 2000), + _replyMsg(id: 'r2', parentId: 'a', pubkey: 'carol', createdAt: 3000), + ]); + + // The relay counted only r1 before r2 landed. + final entries = buildMainTimelineEntries( + messages, + relaySummaries: {'a': relaySummary(replyCount: 1, lastReplyAt: 2000)}, + ); + + expect(entries.single.summary!.replyCount, 2); + expect(entries.single.summary!.lastReplyAt, 3000); + }); + + test('a reply in the same second as the recount still counts', () { + final messages = formatTimeline([ + _textMsg(id: 'a', createdAt: 1000), + _replyMsg(id: 'r1', parentId: 'a', pubkey: 'bob', createdAt: 2000), + _replyMsg(id: 'r2', parentId: 'a', pubkey: 'carol', createdAt: 2000), + ]); + + // Relay timestamps have second precision, so an equal timestamp is no + // proof the recount already included the second reply. + final entries = buildMainTimelineEntries( + messages, + relaySummaries: {'a': relaySummary(replyCount: 1, lastReplyAt: 2000)}, + ); + + expect(entries.single.summary!.replyCount, 2); + }); + + test('a lost recount still leaves a badge from the local reply', () { + final messages = formatTimeline([ + _textMsg(id: 'a', createdAt: 1000), + _replyMsg(id: 'r1', parentId: 'a', pubkey: 'bob', createdAt: 2000), + ]); + + final entries = buildMainTimelineEntries(messages); + + expect(entries.single.summary!.replyCount, 1); + expect(entries.single.summary!.participantPubkeys, ['bob']); + }); + + test('the relay recount wins when it is ahead of the loaded replies', () { + final messages = formatTimeline([ + _textMsg(id: 'a', createdAt: 1000), + _replyMsg(id: 'r1', parentId: 'a', pubkey: 'bob', createdAt: 2000), + ]); + + final entries = buildMainTimelineEntries( + messages, + relaySummaries: {'a': relaySummary(replyCount: 7, lastReplyAt: 8000)}, + ); + + expect(entries.single.summary!.replyCount, 7); + expect(entries.single.summary!.lastReplyAt, 8000); + }); + + test('a zero recount does not erase a locally seen reply', () { + final messages = formatTimeline([ + _textMsg(id: 'a', createdAt: 1000), + _replyMsg(id: 'r1', parentId: 'a', pubkey: 'bob', createdAt: 2000), + ]); + + final entries = buildMainTimelineEntries( + messages, + relaySummaries: {'a': relaySummary(replyCount: 0, lastReplyAt: null)}, + ); + + expect(entries.single.summary!.replyCount, 1); + expect(entries.single.summary!.lastReplyAt, 2000); + }); + + test('a locally seen nested reply raises its root badge', () { + final messages = formatTimeline([ + _textMsg(id: 'a', createdAt: 1000), + _replyMsg(id: 'r1', parentId: 'a', pubkey: 'bob', createdAt: 2000), + _replyMsg( + id: 'r2', + parentId: 'r1', + rootId: 'a', + pubkey: 'carol', + createdAt: 3000, + ), + ]); + + // The recount for 'a' predates r2, so only the locally seen nested reply + // can bring the root badge, time, and facepile up to date. + final entries = buildMainTimelineEntries( + messages, + relaySummaries: { + 'a': relaySummary( + replyCount: 1, + lastReplyAt: 2000, + participantPubkeys: const ['bob'], + ), + }, + ); + + final root = entries.firstWhere((entry) => entry.message.id == 'a'); + expect(root.summary!.replyCount, 2); + expect(root.summary!.lastReplyAt, 3000); + expect(root.summary!.participantPubkeys, ['bob', 'carol']); + }); + + test('the relay half counts descendants, not direct replies', () { + final messages = formatTimeline([_textMsg(id: 'a', createdAt: 1000)]); + + // The relay reports both numbers. A thread of nested replies has a + // `reply_count` far below its `descendant_count`, and the badge stands + // for the whole thread. + final entries = buildMainTimelineEntries( + messages, + relaySummaries: { + 'a': relaySummary( + replyCount: 1, + descendantCount: 5, + lastReplyAt: 9000, + ), + }, + ); + + expect(entries.single.summary!.replyCount, 5); + }); + + test('a nested reply badges the reply it answers', () { + final messages = formatTimeline([ + _textMsg(id: 'a', createdAt: 1000), + _replyMsg( + id: 'r1', + parentId: 'a', + pubkey: 'bob', + createdAt: 2000, + extraTags: const [ + ['broadcast', '1'], + ], + ), + _replyMsg( + id: 'r2', + parentId: 'r1', + rootId: 'a', + pubkey: 'carol', + createdAt: 3000, + ), + ]); + + // The relay keys recounts by the outermost root, so r1 never gets one and + // its badge has to come from the locally seen nested reply. + final entries = buildMainTimelineEntries( + messages, + relaySummaries: {'a': relaySummary(replyCount: 1, lastReplyAt: 2000)}, + ); + + final byId = {for (final entry in entries) entry.message.id: entry}; + expect(byId['a']!.summary!.replyCount, 2); + expect(byId['r1']!.summary!.replyCount, 1); + expect(byId['r1']!.summary!.participantPubkeys, ['carol']); + expect(byId['r1']!.summary!.lastReplyAt, 3000); + }); + + test('merged participants keep relay identities first, capped at 3', () { + final messages = formatTimeline([ + _textMsg(id: 'a', createdAt: 1000), + _replyMsg(id: 'r1', parentId: 'a', pubkey: 'bob', createdAt: 2000), + _replyMsg(id: 'r2', parentId: 'a', pubkey: 'carol', createdAt: 3000), + ]); + + final entries = buildMainTimelineEntries( + messages, + relaySummaries: { + 'a': relaySummary( + replyCount: 1, + lastReplyAt: 2000, + participantPubkeys: const ['zoe', 'yara'], + ), + }, + ); + + expect(entries.single.summary!.participantPubkeys, [ + 'zoe', + 'yara', + 'carol', + ]); + }); + + test('a deleted reply does not hold the badge above the recount', () { + // The relay counts down on delete and re-emits the recount, so taking the + // larger count must not resurrect a deleted reply. `formatTimeline` drops + // it from the local half first, which is what keeps `max` honest here. + final messages = formatTimeline([ + _textMsg(id: 'a', createdAt: 1000), + _replyMsg(id: 'r1', parentId: 'a', pubkey: 'bob', createdAt: 2000), + _replyMsg(id: 'r2', parentId: 'a', pubkey: 'carol', createdAt: 3000), + _deletion(id: 'd1', targets: ['r2']), + ]); + + final entries = buildMainTimelineEntries( + messages, + relaySummaries: {'a': relaySummary(replyCount: 1, lastReplyAt: 2000)}, + ); + + expect(entries.single.summary!.replyCount, 1); + expect(entries.single.summary!.lastReplyAt, 2000); + }); + + test('a participant in both halves is not listed twice', () { + final messages = formatTimeline([ + _textMsg(id: 'a', createdAt: 1000), + _replyMsg(id: 'r1', parentId: 'a', pubkey: 'BOB', createdAt: 2000), + _replyMsg(id: 'r2', parentId: 'a', pubkey: 'carol', createdAt: 3000), + ]); + + final entries = buildMainTimelineEntries( + messages, + relaySummaries: { + 'a': relaySummary( + replyCount: 2, + lastReplyAt: 3000, + participantPubkeys: const ['carol', 'bob'], + ), + }, + ); + + expect(entries.single.summary!.participantPubkeys, ['carol', 'bob']); + }); + }); + group('groupMembershipTimelineEntries', () { List entries(List events) => buildMainTimelineEntries(formatTimeline(events));