diff --git a/mobile/lib/features/channels/channel_messages_provider.dart b/mobile/lib/features/channels/channel_messages_provider.dart index 5d1964f814..fb910f069a 100644 --- a/mobile/lib/features/channels/channel_messages_provider.dart +++ b/mobile/lib/features/channels/channel_messages_provider.dart @@ -3,6 +3,7 @@ import 'package:hooks_riverpod/hooks_riverpod.dart'; import '../../shared/relay/relay.dart'; import 'channel_management_provider.dart'; +import 'pending_local_messages_provider.dart'; import 'channel_window.dart'; import 'thread_replies_provider.dart'; @@ -87,6 +88,7 @@ class ChannelMessagesNotifier extends Notifier>> { final history = await _fetchNewestHistory(session); if (!_isCurrentInit(initVersion)) return; + _confirmLocalMessages(history.map((event) => event.id)); final existing = state.value ?? const []; final existingIds = existing.map((event) => event.id).toSet(); @@ -159,7 +161,13 @@ class ChannelMessagesNotifier extends Notifier>> { }, ); - void _handleLiveEvent(NostrEvent event) { + void _handleLiveEvent(NostrEvent event, {bool authoritative = true}) { + // Reply ownership and its thread-local overlay must transition together. + // The authoritative thread query performs both confirmations after it + // contains the reply; a live echo only triggers that query below. + if (authoritative && event.threadReference.parentId == null) { + _confirmLocalMessages([event.id]); + } if (_usingChannelWindow) { _handleWindowLiveEvent(event); } else { @@ -223,19 +231,108 @@ class ChannelMessagesNotifier extends Notifier>> { return true; } + void _confirmLocalMessages(Iterable eventIds) { + ref + .read(pendingLocalMessagesProvider(channelId).notifier) + .confirm(eventIds); + } + static bool _isMembershipEvent(String content) { return content.contains('member_joined') || content.contains('member_left') || content.contains('member_removed'); } + /// Adds a just-signed outgoing message before the relay acknowledges it. + /// The live relay echo is deduplicated by event id. + void addLocalMessage(NostrEvent event) { + ref.read(pendingLocalMessagesProvider(channelId).notifier).add(event); + final thread = event.threadReference; + if (thread.parentId != null) { + final rootId = thread.rootId; + if (rootId == null) { + throw StateError('Reply ${event.id} has a parent but no thread root.'); + } + ref + .read( + threadLocalRepliesProvider( + ThreadRepliesArgs(channelId: channelId, rootId: rootId), + ).notifier, + ) + .add(event); + return; + } + + final isTimelineRow = EventKind.channelTimelineContentKinds.contains( + event.kind, + ); + if (!_usingChannelWindow && isTimelineRow) { + _windowStore = mergeLiveChannelWindowEvent( + _windowStore, + event, + isTimelineRow: true, + ); + } + _handleLiveEvent(event, authoritative: false); + } + + /// Releases rollback ownership after the publish future succeeds. The + /// optimistic row (and any thread overlay) remains visible until relay data + /// replaces it, because OK and EVENT delivery are unordered. + void completeLocalMessage(String eventId) { + _confirmLocalMessages([eventId]); + } + + /// Rolls back a local message when its publish is rejected or times out. + void removeLocalMessage(String eventId) { + final pending = ref + .read(pendingLocalMessagesProvider(channelId).notifier) + .take(eventId); + if (pending == null) return; + + final thread = pending.threadReference; + if (thread.parentId != null) { + final rootId = thread.rootId; + if (rootId == null) { + throw StateError('Reply $eventId has a parent but no thread root.'); + } + ref + .read( + threadLocalRepliesProvider( + ThreadRepliesArgs(channelId: channelId, rootId: rootId), + ).notifier, + ) + .remove(eventId); + return; + } + + final nextOverlay = _windowStore.liveOverlay + .where((event) => event.id != eventId) + .toList(); + if (nextOverlay.length != _windowStore.liveOverlay.length) { + _windowStore = ChannelWindowStore( + pages: _windowStore.pages, + liveOverlay: nextOverlay, + liveAux: _windowStore.liveAux, + ); + } + + final current = state.value ?? _lastKnownMessages ?? const []; + final next = current.where((event) => event.id != eventId).toList(); + _lastKnownMessages = next; + state = AsyncData(next); + } + static List _mergeEvent( List current, NostrEvent incoming, ) { if (current.any((e) => e.id == incoming.id)) return current; final updated = [...current, incoming]; - updated.sort((a, b) => a.createdAt.compareTo(b.createdAt)); + updated.sort((a, b) { + final createdAt = a.createdAt.compareTo(b.createdAt); + return createdAt != 0 ? createdAt : a.id.compareTo(b.id); + }); return updated; } diff --git a/mobile/lib/features/channels/pending_local_messages_provider.dart b/mobile/lib/features/channels/pending_local_messages_provider.dart new file mode 100644 index 0000000000..064b862022 --- /dev/null +++ b/mobile/lib/features/channels/pending_local_messages_provider.dart @@ -0,0 +1,42 @@ +import 'package:hooks_riverpod/hooks_riverpod.dart'; + +import '../../shared/relay/relay.dart'; + +/// Signed local messages whose publish has not yet been corroborated by an +/// authoritative relay EVENT or query result. +class PendingLocalMessagesNotifier extends Notifier> { + final String channelId; + + PendingLocalMessagesNotifier(this.channelId); + + @override + Map build() => const {}; + + void add(NostrEvent event) { + state = {...state, event.id: event}; + } + + NostrEvent? take(String eventId) { + final event = state[eventId]; + if (event == null) return null; + final next = {...state}..remove(eventId); + state = next; + return event; + } + + void confirm(Iterable eventIds) { + final confirmed = eventIds.toSet(); + if (!state.keys.any(confirmed.contains)) return; + state = { + for (final entry in state.entries) + if (!confirmed.contains(entry.key)) entry.key: entry.value, + }; + } +} + +final pendingLocalMessagesProvider = + NotifierProvider.family< + PendingLocalMessagesNotifier, + Map, + String + >(PendingLocalMessagesNotifier.new); diff --git a/mobile/lib/features/channels/send_message_provider.dart b/mobile/lib/features/channels/send_message_provider.dart index 1771b05f12..3659bda4bf 100644 --- a/mobile/lib/features/channels/send_message_provider.dart +++ b/mobile/lib/features/channels/send_message_provider.dart @@ -4,6 +4,7 @@ import '../../shared/relay/relay.dart'; import '../channels/channel_management_provider.dart'; import '../profile/user_cache_provider.dart'; import '../profile/user_profile.dart'; +import 'channel_messages_provider.dart'; /// Sends messages by signing an event with the user's nsec and publishing it /// over the relay's NIP-42-authenticated WebSocket session. @@ -11,15 +12,25 @@ class SendMessage { final SignedEventRelay _signedEventRelay; final Future> Function(String channelId) _fetchMembers; final Map Function() _readUserCache; + final void Function(String channelId, NostrEvent event) _addLocalMessage; + final void Function(String channelId, String eventId) _completeLocalMessage; + final void Function(String channelId, String eventId) _removeLocalMessage; SendMessage({ required SignedEventRelay signedEventRelay, required Future> Function(String channelId) fetchMembers, required Map Function() readUserCache, + required void Function(String channelId, NostrEvent event) addLocalMessage, + required void Function(String channelId, String eventId) + completeLocalMessage, + required void Function(String channelId, String eventId) removeLocalMessage, }) : _signedEventRelay = signedEventRelay, _fetchMembers = fetchMembers, - _readUserCache = readUserCache; + _readUserCache = readUserCache, + _addLocalMessage = addLocalMessage, + _completeLocalMessage = completeLocalMessage, + _removeLocalMessage = removeLocalMessage; /// Send a text message to a channel. /// @@ -58,11 +69,24 @@ class SendMessage { ...mediaTags, ]; - await _signedEventRelay.submit( - kind: EventKind.streamMessage, - content: content, - tags: tags, - ); + NostrEvent? localMessage; + try { + await _signedEventRelay.submit( + kind: EventKind.streamMessage, + content: content, + tags: tags, + onSigned: (event) { + localMessage = event; + _addLocalMessage(channelId, event); + }, + ); + final event = localMessage; + if (event != null) _completeLocalMessage(channelId, event.id); + } catch (_) { + final event = localMessage; + if (event != null) _removeLocalMessage(channelId, event.id); + rethrow; + } } /// Resolve @mentions to pubkeys, scoped to channel members. @@ -146,5 +170,14 @@ final sendMessageProvider = Provider((ref) { fetchMembers: (channelId) => ref.read(channelMembersProvider(channelId).future), readUserCache: () => ref.read(userCacheProvider), + addLocalMessage: (channelId, event) => ref + .read(channelMessagesProvider(channelId).notifier) + .addLocalMessage(event), + completeLocalMessage: (channelId, eventId) => ref + .read(channelMessagesProvider(channelId).notifier) + .completeLocalMessage(eventId), + removeLocalMessage: (channelId, eventId) => ref + .read(channelMessagesProvider(channelId).notifier) + .removeLocalMessage(eventId), ); }); diff --git a/mobile/lib/features/channels/thread_detail_page.dart b/mobile/lib/features/channels/thread_detail_page.dart index 94e95b8fe5..a81de36e04 100644 --- a/mobile/lib/features/channels/thread_detail_page.dart +++ b/mobile/lib/features/channels/thread_detail_page.dart @@ -54,7 +54,7 @@ class ThreadDetailPage extends HookConsumerWidget { @override Widget build(BuildContext context, WidgetRef ref) { final repliesState = ref.watch( - threadRepliesProvider( + threadRepliesWithLocalProvider( ThreadRepliesArgs(channelId: channelId, rootId: threadHead.id), ), ); diff --git a/mobile/lib/features/channels/thread_replies_provider.dart b/mobile/lib/features/channels/thread_replies_provider.dart index 0fa2b15bd1..6904ba8c5e 100644 --- a/mobile/lib/features/channels/thread_replies_provider.dart +++ b/mobile/lib/features/channels/thread_replies_provider.dart @@ -1,6 +1,9 @@ +import 'dart:async'; + import 'package:hooks_riverpod/hooks_riverpod.dart'; import '../../shared/relay/relay.dart'; +import 'pending_local_messages_provider.dart'; class ThreadRepliesArgs { final String channelId; @@ -64,3 +67,77 @@ NostrFilter _threadRepliesFilter( }, ); } + +class ThreadLocalRepliesNotifier extends Notifier> { + final ThreadRepliesArgs args; + + ThreadLocalRepliesNotifier(this.args); + + @override + List build() => const []; + + void add(NostrEvent event) { + state = _mergeReplies(state, [event]); + } + + void remove(String eventId) { + state = state.where((event) => event.id != eventId).toList(); + } + + void confirm(Set eventIds) { + if (!state.any((event) => eventIds.contains(event.id))) return; + state = state.where((event) => !eventIds.contains(event.id)).toList(); + } +} + +final threadLocalRepliesProvider = + NotifierProvider.family< + ThreadLocalRepliesNotifier, + List, + ThreadRepliesArgs + >(ThreadLocalRepliesNotifier.new); + +/// Relay-backed replies merged with signed local replies that are still +/// waiting for acknowledgement. +final threadRepliesWithLocalProvider = + Provider.family>, ThreadRepliesArgs>(( + ref, + args, + ) { + final relayReplies = ref.watch(threadRepliesProvider(args)); + final localReplies = ref.watch(threadLocalRepliesProvider(args)); + final authoritative = relayReplies.value; + if (authoritative != null && localReplies.isNotEmpty) { + final authoritativeIds = authoritative.map((event) => event.id).toSet(); + if (localReplies.any((event) => authoritativeIds.contains(event.id))) { + Future.microtask(() { + ref + .read(threadLocalRepliesProvider(args).notifier) + .confirm(authoritativeIds); + ref + .read(pendingLocalMessagesProvider(args.channelId).notifier) + .confirm(authoritativeIds); + }); + } + } + if (localReplies.isEmpty) return relayReplies; + return relayReplies.when( + data: (events) => AsyncData(_mergeReplies(events, localReplies)), + loading: () => AsyncData(localReplies), + error: (error, stackTrace) => AsyncData(localReplies), + ); + }); + +List _mergeReplies( + Iterable first, + Iterable second, +) { + final byId = {}; + for (final event in [...first, ...second]) { + byId[event.id] = event; + } + return byId.values.toList()..sort((a, b) { + final createdAt = a.createdAt.compareTo(b.createdAt); + return createdAt != 0 ? createdAt : a.id.compareTo(b.id); + }); +} diff --git a/mobile/lib/shared/relay/signed_event_relay.dart b/mobile/lib/shared/relay/signed_event_relay.dart index 7300095223..a739b76594 100644 --- a/mobile/lib/shared/relay/signed_event_relay.dart +++ b/mobile/lib/shared/relay/signed_event_relay.dart @@ -31,6 +31,7 @@ class SignedEventRelay { required String content, required List> tags, int? createdAt, + void Function(NostrEvent event)? onSigned, }) async { final nsec = _nsec; if (nsec == null || nsec.isEmpty) { @@ -52,6 +53,7 @@ class SignedEventRelay { ); final nostrEvent = NostrEvent.fromJson(event.toMap()); + onSigned?.call(nostrEvent); return _session.publish(nostrEvent); } } diff --git a/mobile/test/features/channels/channel_messages_provider_test.dart b/mobile/test/features/channels/channel_messages_provider_test.dart index 8885b55e16..6620c8c506 100644 --- a/mobile/test/features/channels/channel_messages_provider_test.dart +++ b/mobile/test/features/channels/channel_messages_provider_test.dart @@ -5,6 +5,8 @@ import 'dart:convert'; import 'package:flutter_test/flutter_test.dart'; 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/shared/relay/relay.dart'; void main() { @@ -130,6 +132,369 @@ void main() { }, ); + test( + 'adds and rolls back a local message in the websocket timeline', + () async { + final relaySession = _RecordingRelaySessionNotifier(); + final container = _buildContainer(relaySession); + addTearDown(container.dispose); + + container.read(channelMessagesProvider(_channelId)); + await relaySession.subscribed; + final notifier = container.read( + channelMessagesProvider(_channelId).notifier, + ); + + notifier.addLocalMessage(_event(id: 'local', createdAt: 20)); + expect( + container + .read(channelMessagesProvider(_channelId)) + .value + ?.map((event) => event.id), + ['local'], + ); + + relaySession.completeHistory([_event(id: 'history', createdAt: 10)]); + await _pumpEventQueue(); + + // The initial history merge must retain a local row even if the relay's + // history snapshot was taken before that outgoing event was durable. + expect( + container + .read(channelMessagesProvider(_channelId)) + .value + ?.map((event) => event.id), + ['history', 'local'], + ); + + notifier.removeLocalMessage('local'); + expect( + container + .read(channelMessagesProvider(_channelId)) + .value + ?.map((event) => event.id), + ['history'], + ); + }, + ); + + test( + 'legacy websocket echo retires ownership without duplicating the row', + () async { + final relaySession = _RecordingRelaySessionNotifier(); + final container = _buildContainer(relaySession); + addTearDown(container.dispose); + + container.read(channelMessagesProvider(_channelId)); + await relaySession.subscribed; + final notifier = container.read( + channelMessagesProvider(_channelId).notifier, + ); + final local = _event(id: 'local', createdAt: 20); + notifier.addLocalMessage(local); + + relaySession.emit(local); + await _pumpEventQueue(); + + expect(container.read(pendingLocalMessagesProvider(_channelId)), isEmpty); + expect( + container + .read(channelMessagesProvider(_channelId)) + .value + ?.map((event) => event.id), + ['local'], + ); + }, + ); + + test('adds and rolls back a local message in the channel window', () async { + final relaySession = _RecordingRelaySessionNotifier( + queryResults: [ + [_event(id: 'history', createdAt: 10), _bounds()], + ], + ); + final container = _buildContainer(relaySession); + addTearDown(container.dispose); + + container.read(channelMessagesProvider(_channelId)); + await relaySession.subscribed; + await _pumpEventQueue(); + final notifier = container.read( + channelMessagesProvider(_channelId).notifier, + ); + + notifier.addLocalMessage(_event(id: 'local', createdAt: 20)); + expect( + container + .read(channelMessagesProvider(_channelId)) + .value + ?.map((event) => event.id), + ['history', 'local'], + ); + + notifier.removeLocalMessage('local'); + expect( + container + .read(channelMessagesProvider(_channelId)) + .value + ?.map((event) => event.id), + ['history'], + ); + }); + + test('reconnect hydration cannot retain a rolled-back local row', () async { + final relaySession = _RecordingRelaySessionNotifier( + queryResults: [ + [_event(id: 'history', createdAt: 10), _bounds()], + [_event(id: 'history', createdAt: 10), _bounds()], + ], + ); + final container = _buildContainer(relaySession); + addTearDown(container.dispose); + + container.read(channelMessagesProvider(_channelId)); + await relaySession.subscribed; + await _pumpEventQueue(); + final notifier = container.read( + channelMessagesProvider(_channelId).notifier, + ); + notifier.addLocalMessage(_event(id: 'local', createdAt: 20)); + + relaySession.setConnected(false); + await _pumpEventQueue(); + relaySession.setConnected(true); + await _pumpEventQueue(); + expect( + container + .read(channelMessagesProvider(_channelId)) + .value + ?.map((event) => event.id), + ['history', 'local'], + ); + + notifier.removeLocalMessage('local'); + expect( + container + .read(channelMessagesProvider(_channelId)) + .value + ?.map((event) => event.id), + ['history'], + ); + }); + + test( + 'thread replies are inserted, deduped, and rolled back locally', + () async { + final relaySession = _RecordingRelaySessionNotifier( + queryResults: [ + [_event(id: 'history', createdAt: 10), _bounds()], + [], + [ + _event( + id: 'reply', + createdAt: 20, + extraTags: const [ + ['e', 'root', '', 'reply'], + ], + ), + ], + ], + ); + final container = _buildContainer(relaySession); + addTearDown(container.dispose); + + container.read(channelMessagesProvider(_channelId)); + await relaySession.subscribed; + await _pumpEventQueue(); + const args = ThreadRepliesArgs(channelId: _channelId, rootId: 'root'); + container.read(threadRepliesWithLocalProvider(args)); + await _pumpEventQueue(); + final notifier = container.read( + channelMessagesProvider(_channelId).notifier, + ); + final reply = _event( + id: 'reply', + createdAt: 20, + extraTags: const [ + ['e', 'root', '', 'reply'], + ], + ); + + notifier.addLocalMessage(reply); + expect( + container + .read(threadRepliesWithLocalProvider(args)) + .value + ?.map((event) => event.id), + ['reply'], + ); + expect( + container + .read(channelMessagesProvider(_channelId)) + .value + ?.map((event) => event.id), + ['history'], + ); + + relaySession.emit(reply); + await container.read(threadRepliesProvider(args).future); + container.read(threadRepliesWithLocalProvider(args)); + await _pumpEventQueue(); + expect( + container + .read(threadRepliesWithLocalProvider(args)) + .value + ?.map((event) => event.id), + ['reply'], + ); + expect(container.read(threadLocalRepliesProvider(args)), isEmpty); + expect(container.read(pendingLocalMessagesProvider(_channelId)), isEmpty); + + final rejected = _event( + id: 'rejected', + createdAt: 21, + extraTags: const [ + ['e', 'root', '', 'reply'], + ], + ); + notifier.addLocalMessage(rejected); + notifier.removeLocalMessage('rejected'); + expect( + container + .read(threadRepliesWithLocalProvider(args)) + .value + ?.map((event) => event.id), + ['reply'], + ); + }, + ); + + test( + 'thread live echo keeps ownership until the authoritative refetch succeeds', + () async { + final relaySession = _RecordingRelaySessionNotifier( + queryResults: [ + [_event(id: 'history', createdAt: 10), _bounds()], + [], + Exception('thread refetch failed'), + ], + ); + final container = _buildContainer(relaySession); + addTearDown(container.dispose); + + container.read(channelMessagesProvider(_channelId)); + await relaySession.subscribed; + await _pumpEventQueue(); + const args = ThreadRepliesArgs(channelId: _channelId, rootId: 'root'); + container.read(threadRepliesWithLocalProvider(args)); + await _pumpEventQueue(); + final notifier = container.read( + channelMessagesProvider(_channelId).notifier, + ); + final reply = _event( + id: 'reply', + createdAt: 20, + extraTags: const [ + ['e', 'root', '', 'reply'], + ], + ); + notifier.addLocalMessage(reply); + + relaySession.emit(reply); + await _pumpEventQueue(); + + expect(container.read(pendingLocalMessagesProvider(_channelId)).keys, [ + 'reply', + ]); + expect( + container + .read(threadLocalRepliesProvider(args)) + .map((event) => event.id), + ['reply'], + ); + expect( + container + .read(threadRepliesWithLocalProvider(args)) + .value + ?.map((event) => event.id), + ['reply'], + ); + }, + ); + + test( + 'successful never-echoed send releases ownership but keeps its row across reconnect', + () async { + final relaySession = _RecordingRelaySessionNotifier( + queryResults: [ + [_event(id: 'history', createdAt: 10), _bounds()], + [_event(id: 'history', createdAt: 10), _bounds()], + ], + ); + final container = _buildContainer(relaySession); + addTearDown(container.dispose); + + container.read(channelMessagesProvider(_channelId)); + await relaySession.subscribed; + await _pumpEventQueue(); + final notifier = container.read( + channelMessagesProvider(_channelId).notifier, + ); + notifier.addLocalMessage(_event(id: 'local', createdAt: 20)); + notifier.completeLocalMessage('local'); + + expect(container.read(pendingLocalMessagesProvider(_channelId)), isEmpty); + relaySession.setConnected(false); + await _pumpEventQueue(); + relaySession.setConnected(true); + await _pumpEventQueue(); + + expect(container.read(pendingLocalMessagesProvider(_channelId)), isEmpty); + expect( + container + .read(channelMessagesProvider(_channelId)) + .value + ?.map((event) => event.id), + ['history', 'local'], + ); + }, + ); + + test( + 'window dedupes echoes and orders rapid equal-time local sends', + () async { + final relaySession = _RecordingRelaySessionNotifier( + queryResults: [ + [_event(id: 'history', createdAt: 10), _bounds()], + ], + ); + final container = _buildContainer(relaySession); + addTearDown(container.dispose); + + container.read(channelMessagesProvider(_channelId)); + await relaySession.subscribed; + await _pumpEventQueue(); + final notifier = container.read( + channelMessagesProvider(_channelId).notifier, + ); + notifier.addLocalMessage(_event(id: 'z-local', createdAt: 20)); + notifier.addLocalMessage(_event(id: 'a-local', createdAt: 20)); + relaySession.emit(_event(id: 'z-local', createdAt: 20)); + await _pumpEventQueue(); + + expect(container.read(pendingLocalMessagesProvider(_channelId)).keys, [ + 'a-local', + ]); + expect( + container + .read(channelMessagesProvider(_channelId)) + .value + ?.map((event) => event.id), + ['history', 'z-local', 'a-local'], + ); + }, + ); + test('window pagination failures return false without exhausting', () async { final relaySession = _RecordingRelaySessionNotifier( queryResults: [ @@ -178,14 +543,19 @@ ProviderContainer _buildContainer(_RecordingRelaySessionNotifier relaySession) { ); } -NostrEvent _event({required String id, required int createdAt}) { +NostrEvent _event({ + required String id, + required int createdAt, + List> extraTags = const [], +}) { return NostrEvent( id: id, pubkey: 'alice', createdAt: createdAt, kind: EventKind.streamMessageV2, - tags: const [ + tags: [ ['h', _channelId], + ...extraTags, ], content: id, sig: 'sig', @@ -243,6 +613,12 @@ class _RecordingRelaySessionNotifier extends RelaySessionNotifier { @override SessionState build() => const SessionState(status: SessionStatus.connected); + void setConnected(bool connected) { + state = SessionState( + status: connected ? SessionStatus.connected : SessionStatus.disconnected, + ); + } + @override Future> queryRelay( List filters, { diff --git a/mobile/test/features/channels/read_state/read_state_manager_test.dart b/mobile/test/features/channels/read_state/read_state_manager_test.dart index e4240a7c14..953d2ff10d 100644 --- a/mobile/test/features/channels/read_state/read_state_manager_test.dart +++ b/mobile/test/features/channels/read_state/read_state_manager_test.dart @@ -181,6 +181,7 @@ class _FakeSignedEventRelay implements SignedEventRelay { required String content, required List> tags, int? createdAt, + void Function(NostrEvent event)? onSigned, }) async { submitted.complete(_SubmittedEvent(kind: kind, tags: tags)); return _stubAckEvent(); @@ -199,6 +200,7 @@ class _UnsupportedKindSignedEventRelay implements SignedEventRelay { required String content, required List> tags, int? createdAt, + void Function(NostrEvent event)? onSigned, }) async { submitCount++; throw Exception('restricted: unknown event kind'); @@ -217,6 +219,7 @@ class _MissingScopeSignedEventRelay implements SignedEventRelay { required String content, required List> tags, int? createdAt, + void Function(NostrEvent event)? onSigned, }) async { submitCount++; throw Exception('missing users:write'); diff --git a/mobile/test/features/channels/send_message_provider_test.dart b/mobile/test/features/channels/send_message_provider_test.dart new file mode 100644 index 0000000000..f91ce87b2b --- /dev/null +++ b/mobile/test/features/channels/send_message_provider_test.dart @@ -0,0 +1,96 @@ +import 'dart:async'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:nostr/nostr.dart' as nostr; +import 'package:buzz/features/channels/send_message_provider.dart'; +import 'package:buzz/shared/relay/relay.dart'; + +void main() { + test( + 'adds the signed message locally before relay acknowledgement', + () async { + final session = _PendingPublishRelaySession(); + final localMessages = []; + final removedIds = []; + final completedIds = []; + final send = SendMessage( + signedEventRelay: SignedEventRelay( + session: session, + nsec: nostr.Keys.generate().nsec, + ), + fetchMembers: (_) async => const [], + readUserCache: () => const {}, + addLocalMessage: (_, event) => localMessages.add(event), + completeLocalMessage: (_, eventId) => completedIds.add(eventId), + removeLocalMessage: (_, eventId) => removedIds.add(eventId), + ); + + final result = send(channelId: _channelId, content: 'hello'); + await session.published; + + expect(localMessages, hasLength(1)); + expect(localMessages.single.id, session.event.id); + expect(localMessages.single.content, 'hello'); + expect(localMessages.single.channelId, _channelId); + expect(removedIds, isEmpty); + + session.accept(); + await result; + expect(completedIds, [localMessages.single.id]); + expect(removedIds, isEmpty); + }, + ); + + test('rolls back the signed local message when publish fails', () async { + final session = _PendingPublishRelaySession(); + final localMessages = []; + final completedIds = []; + final removedIds = []; + final send = SendMessage( + signedEventRelay: SignedEventRelay( + session: session, + nsec: nostr.Keys.generate().nsec, + ), + fetchMembers: (_) async => const [], + readUserCache: () => const {}, + addLocalMessage: (_, event) => localMessages.add(event), + completeLocalMessage: (_, eventId) => completedIds.add(eventId), + removeLocalMessage: (_, eventId) => removedIds.add(eventId), + ); + + final result = send(channelId: _channelId, content: 'hello'); + await session.published; + session.reject(); + + await expectLater(result, throwsException); + expect(completedIds, isEmpty); + expect(removedIds, [localMessages.single.id]); + }); +} + +const _channelId = '11111111-1111-4111-8111-111111111111'; + +class _PendingPublishRelaySession extends RelaySessionNotifier { + final Completer _result = Completer(); + final Completer _published = Completer(); + late NostrEvent event; + + Future get published => _published.future; + + @override + SessionState build() => const SessionState(status: SessionStatus.connected); + + @override + Future publish( + NostrEvent event, { + Duration timeout = const Duration(seconds: 8), + }) { + this.event = event; + _published.complete(); + return _result.future; + } + + void accept() => _result.complete(event); + + void reject() => _result.completeError(Exception('relay rejected event')); +}