From a9a786ddfabea27552a6633494c789ffd2ad9a80 Mon Sep 17 00:00:00 2001 From: Wes Date: Fri, 14 Aug 2026 09:09:00 -0600 Subject: [PATCH] fix(mobile): harden startup subscription lifecycle Publish the bounded channel snapshot independently of live EOSE, quarantine terminally closed channel subscriptions until a material invalidator, and extract channel sync and relay support code below the file-size ceiling. Co-authored-by: Carl Signed-off-by: Wes --- .../lib/features/channels/channel_sync.dart | 253 +++++++++++++++ .../features/channels/channels_provider.dart | 290 ++---------------- mobile/lib/shared/relay/relay.dart | 2 + mobile/lib/shared/relay/relay_nip98_auth.dart | 38 +++ mobile/lib/shared/relay/relay_session.dart | 48 +-- .../lib/shared/relay/relay_session_state.dart | 11 + .../channels/channels_provider_test.dart | 51 ++- 7 files changed, 381 insertions(+), 312 deletions(-) create mode 100644 mobile/lib/features/channels/channel_sync.dart create mode 100644 mobile/lib/shared/relay/relay_nip98_auth.dart create mode 100644 mobile/lib/shared/relay/relay_session_state.dart diff --git a/mobile/lib/features/channels/channel_sync.dart b/mobile/lib/features/channels/channel_sync.dart new file mode 100644 index 0000000000..2018380da0 --- /dev/null +++ b/mobile/lib/features/channels/channel_sync.dart @@ -0,0 +1,253 @@ +import 'dart:async'; +import 'dart:convert'; +import 'dart:math'; + +import 'package:flutter/widgets.dart'; + +import '../../shared/relay/relay.dart'; +import '../../shared/utils/string_utils.dart'; +import 'channel.dart'; +import 'unread_badge/should_notify_for_event.dart'; + +const lastMessageQueryBatchSize = 100; +const lastMessageQueryTotalTimeout = Duration(seconds: 8); + +Future> fetchChannelHistoryBatch( + RelaySessionNotifier session, + List filters, { + required String operation, + required bool Function() isCancelled, +}) async { + if (filters.isEmpty || isCancelled()) return const []; + + try { + return await session.queryRelay(filters); + } catch (error) { + debugPrint( + '[ChannelsNotifier] batched $operation failed; ' + 'using bounded websocket fallback: $error', + ); + } + + const fallbackConcurrency = 4; + final events = []; + for (var start = 0; start < filters.length; start += fallbackConcurrency) { + if (isCancelled()) break; + final end = min(start + fallbackConcurrency, filters.length); + final results = await Future.wait( + filters.sublist(start, end).map((filter) async { + try { + return await session.fetchHistory(filter); + } catch (_) { + return const []; + } + }), + ); + if (isCancelled()) break; + for (final result in results) { + events.addAll(result); + } + } + return events; +} + +Future> fetchHiddenDmIds( + RelaySessionNotifier session, + String myPk, +) async { + try { + final events = await session.fetchHistory(NostrFilters.hiddenDms(myPk)); + if (events.isEmpty) return const {}; + NostrEvent latest = events.first; + for (final event in events.skip(1)) { + if (event.createdAt > latest.createdAt) { + latest = event; + } + } + return { + for (final tag in latest.tags) + if (tag.length >= 2 && tag[0] == 'h') tag[1], + }; + } catch (_) { + return const {}; + } +} + +/// Build a [Channel] from a kind:39000 metadata event. +/// +/// [displayNames] maps lowercase participant pubkey → resolved label and is +/// used to populate [Channel.participants] for DMs so [Channel.displayLabel] +/// can render real names instead of the relay-canonical "DM" name. +Channel channelFromMeta( + NostrEvent event, { + required bool isMember, + Map displayNames = const {}, +}) { + final data = ChannelData.fromEvent(event); + final participants = data.channelType == 'dm' + ? [ + for (final pk in data.participantPubkeys) + displayNames[pk.toLowerCase()] ?? shortPubkey(pk), + ] + : const []; + return Channel( + id: data.id, + name: data.name, + channelType: data.channelType, + visibility: data.visibility, + description: data.description, + topic: data.topic, + createdBy: event.pubkey, + createdAt: DateTime.fromMillisecondsSinceEpoch( + event.createdAt * 1000, + isUtc: true, + ), + memberCount: 0, + lastMessageAt: null, + // `archivedAt` doubles as both the archived-state flag and the timestamp. + // The kind:39000 metadata only carries `["archived", "true"]`, not the + // moment of archival, so we stamp the event's `createdAt` — that's when + // the relay republished the metadata, which is the closest signal we have. + archivedAt: data.isArchived + ? DateTime.fromMillisecondsSinceEpoch( + event.createdAt * 1000, + isUtc: true, + ) + : null, + participants: participants, + participantPubkeys: data.participantPubkeys, + isMember: isMember, + ttlSeconds: data.ttlSeconds, + ttlDeadline: data.ttlDeadline, + ); +} + +List> chunkChannelsForLastMessageQuery(List channels) => + [ + for ( + var start = 0; + start < channels.length; + start += lastMessageQueryBatchSize + ) + channels.sublist( + start, + min(start + lastMessageQueryBatchSize, channels.length), + ), + ]; + +List buildChannelLastMessageFilters(List channels) => [ + for (final channel in channels) + NostrFilter( + kinds: EventKind.channelMessageEventKinds, + tags: { + '#h': [channel.id], + }, + limit: channel.isDm ? 1 : 20, + ), +]; + +Map resolveChannelLastMessages( + List channels, + Iterable events, { + required String myPk, + Set mutedChannelIds = const {}, +}) { + final channelsById = {for (final channel in channels) channel.id: channel}; + final result = {}; + for (final event in events) { + final channelId = event.channelId; + final channel = channelId == null ? null : channelsById[channelId]; + if (channel == null) continue; + if (!channel.isDm && + !shouldNotifyForEvent( + event, + myPk, + mutedChannelIds: mutedChannelIds, + channelId: channelId, + )) { + continue; + } + final current = result[channel.id]; + if (current == null || event.createdAt > current) { + result[channel.id] = event.createdAt; + } + } + return result; +} + +typedef TaskDelay = Future Function(Duration duration); +typedef TaskErrorHandler = void Function(Object error); + +Future runPacedTasks( + List Function()> tasks, { + required int maxConcurrent, + required Duration startInterval, + required bool Function() isCancelled, + TaskDelay delay = defaultTaskDelay, + TaskErrorHandler onError = _defaultTaskErrorHandler, +}) async { + if (maxConcurrent < 1) throw ArgumentError.value(maxConcurrent); + var nextTask = 0; + var firstStart = true; + Future startPermit = Future.value(); + + Future acquireStartPermit() async { + if (firstStart) { + firstStart = false; + } else { + startPermit = startPermit.then((_) => delay(startInterval)); + await startPermit; + } + return !isCancelled(); + } + + Future worker() async { + while (true) { + final index = nextTask++; + if (index >= tasks.length) return; + if (!await acquireStartPermit()) return; + try { + await tasks[index](); + } catch (error) { + onError(error); + } + } + } + + await Future.wait([ + for (var i = 0; i < min(maxConcurrent, tasks.length); i++) worker(), + ]); +} + +Future defaultTaskDelay(Duration duration) => + Future.delayed(duration); + +void _defaultTaskErrorHandler(Object error) { + debugPrint('[ChannelsNotifier] paced task failed: $error'); +} + +bool sameStringSet(Set left, Set right) => + left.length == right.length && left.containsAll(right); + +String? observedUnreadRootId(NostrEvent event) => + isBroadcastReply(event) ? null : event.threadReference.rootId; + +bool isBroadcastReply(NostrEvent event) => event.tags.any( + (tag) => tag.length >= 2 && tag[0] == 'broadcast' && tag[1] == '1', +); + +Set readRootIdSet(String? raw) { + if (raw == null || raw.isEmpty) return {}; + try { + final decoded = jsonDecode(raw); + if (decoded is! List) return {}; + return { + for (final value in decoded) + if (value is String) value, + }; + } catch (_) { + return {}; + } +} + +String encodeRootIdSet(Set values) => jsonEncode(values.toList()); diff --git a/mobile/lib/features/channels/channels_provider.dart b/mobile/lib/features/channels/channels_provider.dart index 4ede6c2097..0ec61e4289 100644 --- a/mobile/lib/features/channels/channels_provider.dart +++ b/mobile/lib/features/channels/channels_provider.dart @@ -1,5 +1,4 @@ import 'dart:async'; -import 'dart:convert'; import 'dart:math'; import 'package:flutter/widgets.dart'; @@ -9,6 +8,7 @@ import '../../shared/relay/relay.dart'; import '../../shared/theme/theme_provider.dart'; import '../../shared/utils/string_utils.dart'; import 'channel.dart'; +import 'channel_sync.dart'; import 'channel_management_provider.dart' show ChannelMember, channelDetailsProvider; import 'channel_mutes/channel_mutes_provider.dart'; @@ -20,8 +20,6 @@ import 'unread_badge/should_notify_for_event.dart'; const _channelTypeOrder = {'stream': 0, 'forum': 1, 'dm': 2}; const _unreadCatchUpLimit = 1000; -const _lastMessageQueryBatchSize = 100; -const _lastMessageQueryTotalTimeout = Duration(seconds: 8); const _unreadCatchUpConcurrency = 4; const _unreadCatchUpStartInterval = Duration(milliseconds: 125); const _participatedRootIdsPrefix = 'buzz-thread-participation.v1'; @@ -43,6 +41,7 @@ class ChannelsNotifier extends AsyncNotifier> { final Map _liveSubscriptionsByChannel = {}; final Map> _pendingSubscriptionCancellations = {}; + final Set _terminallyClosedChannelIds = {}; Future _liveSubscriptionQueue = Future.value(); List _desiredLiveChannels = const []; Set _desiredLiveChannelIds = const {}; @@ -73,7 +72,6 @@ class ChannelsNotifier extends AsyncNotifier> { Map get latestObservedByChannel => Map.unmodifiable(_latestObservedByChannel); - Map> get observedUnreadEventsByChannel => Map>.unmodifiable({ @@ -112,6 +110,7 @@ class ChannelsNotifier extends AsyncNotifier> { !connected.isCompleted) { connected.complete(); } else if (previous?.status != SessionStatus.connected) { + _terminallyClosedChannelIds.clear(); unawaited(_backstopRefresh()); } }); @@ -252,11 +251,11 @@ class ChannelsNotifier extends AsyncNotifier> { } } - final hiddenDmIds = await _fetchHiddenDmIds(session, myPk); + final hiddenDmIds = await fetchHiddenDmIds(session, myPk); final channels = []; for (final event in dedupedMetas) { - final channel = _channelFromMeta( + final channel = channelFromMeta( event, isMember: true, displayNames: displayNames, @@ -338,8 +337,7 @@ class ChannelsNotifier extends AsyncNotifier> { // so a slow relay cannot delay live subscription setup once per batch. final queryStopwatch = Stopwatch()..start(); for (final batch in chunkChannelsForLastMessageQuery(activeChannels)) { - final remaining = - _lastMessageQueryTotalTimeout - queryStopwatch.elapsed; + final remaining = lastMessageQueryTotalTimeout - queryStopwatch.elapsed; if (remaining <= Duration.zero) { debugPrint('[ChannelsNotifier] last-message query budget exhausted'); break; @@ -430,11 +428,7 @@ class ChannelsNotifier extends AsyncNotifier> { _applyLiveEventToChannels(channels, event, myPk); } _bootstrapLiveEvents.clear(); - if (channels.length <= _lastMessageQueryBatchSize) { - await bootstrapLiveSync; - } else { - unawaited(bootstrapLiveSync!); - } + unawaited(bootstrapLiveSync!); } return channels; } @@ -473,116 +467,6 @@ class ChannelsNotifier extends AsyncNotifier> { _memberSnapshotsByChannelId = Map.unmodifiable(snapshots); } - Future> _fetchChannelHistoryBatch( - RelaySessionNotifier session, - List filters, { - required String operation, - required bool Function() isCancelled, - }) async { - if (filters.isEmpty || isCancelled()) return const []; - - try { - return await session.queryRelay(filters); - } catch (error) { - debugPrint( - '[ChannelsNotifier] batched $operation failed; ' - 'using bounded websocket fallback: $error', - ); - } - - const fallbackConcurrency = 4; - final events = []; - for (var start = 0; start < filters.length; start += fallbackConcurrency) { - if (isCancelled()) break; - final end = min(start + fallbackConcurrency, filters.length); - final results = await Future.wait( - filters.sublist(start, end).map((filter) async { - try { - return await session.fetchHistory(filter); - } catch (_) { - return const []; - } - }), - ); - if (isCancelled()) break; - for (final result in results) { - events.addAll(result); - } - } - return events; - } - - Future> _fetchHiddenDmIds( - RelaySessionNotifier session, - String myPk, - ) async { - try { - final events = await session.fetchHistory(NostrFilters.hiddenDms(myPk)); - if (events.isEmpty) return const {}; - NostrEvent latest = events.first; - for (final event in events.skip(1)) { - if (event.createdAt > latest.createdAt) { - latest = event; - } - } - return { - for (final tag in latest.tags) - if (tag.length >= 2 && tag[0] == 'h') tag[1], - }; - } catch (_) { - return const {}; - } - } - - /// Build a [Channel] from a kind:39000 metadata event. - /// - /// [displayNames] maps lowercase participant pubkey → resolved label and is - /// used to populate [Channel.participants] for DMs so [Channel.displayLabel] - /// can render real names instead of the relay-canonical "DM" name. - Channel _channelFromMeta( - NostrEvent event, { - required bool isMember, - Map displayNames = const {}, - }) { - final data = ChannelData.fromEvent(event); - final participants = data.channelType == 'dm' - ? [ - for (final pk in data.participantPubkeys) - displayNames[pk.toLowerCase()] ?? shortPubkey(pk), - ] - : const []; - return Channel( - id: data.id, - name: data.name, - channelType: data.channelType, - visibility: data.visibility, - description: data.description, - topic: data.topic, - createdBy: event.pubkey, - createdAt: DateTime.fromMillisecondsSinceEpoch( - event.createdAt * 1000, - isUtc: true, - ), - memberCount: 0, - lastMessageAt: null, - // `archivedAt` doubles as both the archived-state flag and the timestamp. - // The kind:39000 metadata only carries `["archived", "true"]`, not the - // moment of archival, so we stamp the event's `createdAt` — that's when - // the relay republished the metadata, which is the closest signal we have. - archivedAt: data.isArchived - ? DateTime.fromMillisecondsSinceEpoch( - event.createdAt * 1000, - isUtc: true, - ) - : null, - participants: participants, - participantPubkeys: data.participantPubkeys, - isMember: isMember, - ttlSeconds: data.ttlSeconds, - ttlDeadline: data.ttlDeadline, - ); - } - /// Subscribe per-channel to live events (requires `#h` tag for relay /// channel-scoped fan-out). Also starts a 60s WS backstop poll to detect /// newly created channels we don't yet have subscriptions for. @@ -592,7 +476,7 @@ class ChannelsNotifier extends AsyncNotifier> { if (channel.isMember && !channel.isArchived) channel.id, }; final relayBaseUrl = ref.read(relayConfigProvider).baseUrl; - if (!_sameStringSet(_unreadCatchUpChannelIds, channelIds)) { + if (!sameStringSet(_unreadCatchUpChannelIds, channelIds)) { // Stop admitting work from the old channel set while its replacement // live subscriptions are being established. _unreadCatchUpGeneration++; @@ -602,6 +486,9 @@ class ChannelsNotifier extends AsyncNotifier> { _pendingSubscriptionCancellations.remove(entry.key); if (!entry.value.isCompleted) entry.value.complete(); } + if (!sameStringSet(_desiredLiveChannelIds, channelIds)) { + _terminallyClosedChannelIds.clear(); + } _desiredLiveChannels = channels; _desiredLiveChannelIds = channelIds; final subscriptionVersion = ++_subscriptionVersion; @@ -662,7 +549,9 @@ class ChannelsNotifier extends AsyncNotifier> { final newChannelIds = [ for (final channelId in channelIds) - if (!_liveSubscriptionsByChannel.containsKey(channelId)) channelId, + if (!_liveSubscriptionsByChannel.containsKey(channelId) && + !_terminallyClosedChannelIds.contains(channelId)) + channelId, ]; final subscribeTasks = Function()>[ for (final channelId in newChannelIds) @@ -691,6 +580,7 @@ class ChannelsNotifier extends AsyncNotifier> { final current = _liveSubscriptionsByChannel[channelId]; if (current?.attempt != attempt) return; _liveSubscriptionsByChannel.remove(channelId); + _terminallyClosedChannelIds.add(channelId); _unreadCatchUpChannelIds = Set.unmodifiable( _unreadCatchUpChannelIds.difference({channelId}), ); @@ -778,7 +668,7 @@ class ChannelsNotifier extends AsyncNotifier> { for (final channel in channels) if (channel.isMember && !channel.isArchived) channel.id, }; - if (_sameStringSet(_unreadCatchUpChannelIds, catchUpChannelIds)) return; + if (sameStringSet(_unreadCatchUpChannelIds, catchUpChannelIds)) return; _unreadCatchUpChannelIds = Set.unmodifiable(catchUpChannelIds); final catchUpGeneration = ++_unreadCatchUpGeneration; unawaited(_catchUpUnreadEvents(channels, catchUpGeneration)); @@ -830,7 +720,7 @@ class ChannelsNotifier extends AsyncNotifier> { ]; try { - final events = await _fetchChannelHistoryBatch( + final events = await fetchChannelHistoryBatch( session, filters, operation: 'unread catch-up', @@ -963,10 +853,10 @@ class ChannelsNotifier extends AsyncNotifier> { _threadInterestPubkey = normalizedPubkey; try { final prefs = ref.read(savedPrefsProvider); - _participatedRootIds = _readRootIdSet( + _participatedRootIds = readRootIdSet( prefs.getString('$_participatedRootIdsPrefix:$normalizedPubkey'), ); - _authoredRootIds = _readRootIdSet( + _authoredRootIds = readRootIdSet( prefs.getString('$_authoredRootIdsPrefix:$normalizedPubkey'), ); } catch (_) { @@ -989,11 +879,11 @@ class ChannelsNotifier extends AsyncNotifier> { final prefs = ref.read(savedPrefsProvider); prefs.setString( '$_participatedRootIdsPrefix:$normalizedPubkey', - _encodeRootIdSet(_participatedRootIds), + encodeRootIdSet(_participatedRootIds), ); prefs.setString( '$_authoredRootIdsPrefix:$normalizedPubkey', - _encodeRootIdSet(_authoredRootIds), + encodeRootIdSet(_authoredRootIds), ); } catch (_) { // Ignore storage failures; in-memory interest still works this session. @@ -1002,7 +892,7 @@ class ChannelsNotifier extends AsyncNotifier> { void _recordUnreadEvent(Channel channel, NostrEvent event, String myPk) { final isThreadedReply = - event.threadReference.parentId != null && !_isBroadcastReply(event); + event.threadReference.parentId != null && !isBroadcastReply(event); final isHighPriority = channel.isDm || isHighPriorityEvent(event.tags, myPk); recordObservedUnreadEvent( @@ -1011,7 +901,7 @@ class ChannelsNotifier extends AsyncNotifier> { makeObservedUnreadEvent( id: event.id, createdAt: event.createdAt, - rootId: _observedUnreadRootId(event), + rootId: observedUnreadRootId(event), highPriority: isHighPriority, channelType: channel.channelType, isThreadedReply: isThreadedReply, @@ -1071,6 +961,7 @@ class ChannelsNotifier extends AsyncNotifier> { // cached channel list with [] or an error. Wait for `build()` to re-run // when the session transitions to connected. if (sessionState.status != SessionStatus.connected) return; + _terminallyClosedChannelIds.clear(); state = await AsyncValue.guard(() => _fetch(subscribeLive: true)); } @@ -1090,6 +981,7 @@ class ChannelsNotifier extends AsyncNotifier> { subscription.unsubscribe(); } _liveSubscriptionsByChannel.clear(); + _terminallyClosedChannelIds.clear(); _subscriptionRelayBaseUrl = null; _backstopTimer?.cancel(); _backstopTimer = null; @@ -1103,135 +995,5 @@ final channelsProvider = AsyncNotifierProvider>( ); final channelsLiveSubscriptionDelayProvider = Provider( - (_) => _defaultTaskDelay, + (_) => defaultTaskDelay, ); - -List> chunkChannelsForLastMessageQuery(List channels) => - [ - for ( - var start = 0; - start < channels.length; - start += _lastMessageQueryBatchSize - ) - channels.sublist( - start, - min(start + _lastMessageQueryBatchSize, channels.length), - ), - ]; - -List buildChannelLastMessageFilters(List channels) => [ - for (final channel in channels) - NostrFilter( - kinds: EventKind.channelMessageEventKinds, - tags: { - '#h': [channel.id], - }, - limit: channel.isDm ? 1 : 20, - ), -]; - -Map resolveChannelLastMessages( - List channels, - Iterable events, { - required String myPk, - Set mutedChannelIds = const {}, -}) { - final channelsById = {for (final channel in channels) channel.id: channel}; - final result = {}; - for (final event in events) { - final channelId = event.channelId; - final channel = channelId == null ? null : channelsById[channelId]; - if (channel == null) continue; - if (!channel.isDm && - !shouldNotifyForEvent( - event, - myPk, - mutedChannelIds: mutedChannelIds, - channelId: channelId, - )) { - continue; - } - final current = result[channel.id]; - if (current == null || event.createdAt > current) { - result[channel.id] = event.createdAt; - } - } - return result; -} - -typedef TaskDelay = Future Function(Duration duration); -typedef TaskErrorHandler = void Function(Object error); - -Future runPacedTasks( - List Function()> tasks, { - required int maxConcurrent, - required Duration startInterval, - required bool Function() isCancelled, - TaskDelay delay = _defaultTaskDelay, - TaskErrorHandler onError = _defaultTaskErrorHandler, -}) async { - if (maxConcurrent < 1) throw ArgumentError.value(maxConcurrent); - var nextTask = 0; - var firstStart = true; - Future startPermit = Future.value(); - - Future acquireStartPermit() async { - if (firstStart) { - firstStart = false; - } else { - startPermit = startPermit.then((_) => delay(startInterval)); - await startPermit; - } - return !isCancelled(); - } - - Future worker() async { - while (true) { - final index = nextTask++; - if (index >= tasks.length) return; - if (!await acquireStartPermit()) return; - try { - await tasks[index](); - } catch (error) { - onError(error); - } - } - } - - await Future.wait([ - for (var i = 0; i < min(maxConcurrent, tasks.length); i++) worker(), - ]); -} - -Future _defaultTaskDelay(Duration duration) => - Future.delayed(duration); - -void _defaultTaskErrorHandler(Object error) { - debugPrint('[ChannelsNotifier] paced task failed: $error'); -} - -bool _sameStringSet(Set left, Set right) => - left.length == right.length && left.containsAll(right); - -String? _observedUnreadRootId(NostrEvent event) => - _isBroadcastReply(event) ? null : event.threadReference.rootId; - -bool _isBroadcastReply(NostrEvent event) => event.tags.any( - (tag) => tag.length >= 2 && tag[0] == 'broadcast' && tag[1] == '1', -); - -Set _readRootIdSet(String? raw) { - if (raw == null || raw.isEmpty) return {}; - try { - final decoded = jsonDecode(raw); - if (decoded is! List) return {}; - return { - for (final value in decoded) - if (value is String) value, - }; - } catch (_) { - return {}; - } -} - -String _encodeRootIdSet(Set values) => jsonEncode(values.toList()); diff --git a/mobile/lib/shared/relay/relay.dart b/mobile/lib/shared/relay/relay.dart index dd153fc928..083aa8c82d 100644 --- a/mobile/lib/shared/relay/relay.dart +++ b/mobile/lib/shared/relay/relay.dart @@ -9,6 +9,8 @@ export 'relay_closed_policy.dart'; export 'relay_client.dart'; export 'relay_provider.dart'; export 'relay_rate_limit_gate.dart'; +export 'relay_nip98_auth.dart'; export 'relay_session.dart'; +export 'relay_session_state.dart'; export 'relay_socket.dart'; export 'signed_event_relay.dart'; diff --git a/mobile/lib/shared/relay/relay_nip98_auth.dart b/mobile/lib/shared/relay/relay_nip98_auth.dart new file mode 100644 index 0000000000..4dbb13c0c9 --- /dev/null +++ b/mobile/lib/shared/relay/relay_nip98_auth.dart @@ -0,0 +1,38 @@ +import 'dart:convert'; +import 'dart:typed_data'; + +import 'package:nostr/nostr.dart' as nostr; +import 'package:pointycastle/digests/sha256.dart'; +import 'package:uuid/uuid.dart'; + +String buildNip98AuthHeader({ + required String method, + required String url, + required List bodyBytes, + required String? nsec, +}) { + if (nsec == null || nsec.isEmpty) { + throw Exception('Cannot query relay: no signing key available'); + } + final privkeyHex = nostr.Nip19.decode(payload: nsec).data; + if (privkeyHex.isEmpty) { + throw Exception('Invalid nsec'); + } + final payloadHash = SHA256Digest() + .process(Uint8List.fromList(bodyBytes)) + .map((byte) => byte.toRadixString(16).padLeft(2, '0')) + .join(); + final event = nostr.Event.from( + kind: 27235, + content: '', + tags: [ + ['u', url], + ['method', method.toUpperCase()], + ['payload', payloadHash], + ['nonce', const Uuid().v4()], + ], + secretKey: privkeyHex, + verify: false, + ); + return 'Nostr ${base64.encode(utf8.encode(event.toJson()))}'; +} diff --git a/mobile/lib/shared/relay/relay_session.dart b/mobile/lib/shared/relay/relay_session.dart index 028892d72d..5de7e54cea 100644 --- a/mobile/lib/shared/relay/relay_session.dart +++ b/mobile/lib/shared/relay/relay_session.dart @@ -3,9 +3,6 @@ import 'dart:convert'; import 'dart:math'; import 'package:http/http.dart' as http; -import 'package:nostr/nostr.dart' as nostr; -import 'package:pointycastle/digests/sha256.dart'; -import 'package:uuid/uuid.dart'; import 'package:flutter/foundation.dart'; import 'package:hooks_riverpod/hooks_riverpod.dart'; @@ -15,20 +12,13 @@ import 'nostr_models.dart'; import 'relay_client.dart'; import 'relay_closed_policy.dart'; import 'relay_http_query_client.dart'; +import 'relay_nip98_auth.dart'; +export 'relay_nip98_auth.dart'; +import 'relay_session_state.dart'; import 'relay_provider.dart'; import 'relay_rate_limit_gate.dart'; import 'relay_socket.dart'; -enum SessionStatus { disconnected, connecting, connected, reconnecting } - -@immutable -class SessionState { - final SessionStatus status; - final int reconnectAttempt; - - const SessionState({required this.status, this.reconnectAttempt = 0}); -} - class _HistorySubscription { final List events = []; final Completer> completer; @@ -1004,35 +994,3 @@ final relaySessionProvider = NotifierProvider( RelaySessionNotifier.new, ); - -String buildNip98AuthHeader({ - required String method, - required String url, - required List bodyBytes, - required String? nsec, -}) { - if (nsec == null || nsec.isEmpty) { - throw Exception('Cannot query relay: no signing key available'); - } - final privkeyHex = nostr.Nip19.decode(payload: nsec).data; - if (privkeyHex.isEmpty) { - throw Exception('Invalid nsec'); - } - final payloadHash = SHA256Digest() - .process(Uint8List.fromList(bodyBytes)) - .map((byte) => byte.toRadixString(16).padLeft(2, '0')) - .join(); - final event = nostr.Event.from( - kind: 27235, - content: '', - tags: [ - ['u', url], - ['method', method.toUpperCase()], - ['payload', payloadHash], - ['nonce', const Uuid().v4()], - ], - secretKey: privkeyHex, - verify: false, - ); - return 'Nostr ${base64.encode(utf8.encode(event.toJson()))}'; -} diff --git a/mobile/lib/shared/relay/relay_session_state.dart b/mobile/lib/shared/relay/relay_session_state.dart new file mode 100644 index 0000000000..13c8c739eb --- /dev/null +++ b/mobile/lib/shared/relay/relay_session_state.dart @@ -0,0 +1,11 @@ +import 'package:flutter/foundation.dart'; + +enum SessionStatus { disconnected, connecting, connected, reconnecting } + +@immutable +class SessionState { + final SessionStatus status; + final int reconnectAttempt; + + const SessionState({required this.status, this.reconnectAttempt = 0}); +} diff --git a/mobile/test/features/channels/channels_provider_test.dart b/mobile/test/features/channels/channels_provider_test.dart index accfe67242..0d2df968b7 100644 --- a/mobile/test/features/channels/channels_provider_test.dart +++ b/mobile/test/features/channels/channels_provider_test.dart @@ -7,6 +7,7 @@ import 'package:hooks_riverpod/hooks_riverpod.dart'; import 'package:buzz/features/channels/channel.dart'; import 'package:buzz/features/channels/channel_management_provider.dart'; import 'package:buzz/features/channels/channels_provider.dart'; +import 'package:buzz/features/channels/channel_sync.dart'; import 'package:buzz/shared/relay/relay.dart'; /// Tests for [ChannelsNotifier] in the pure-Nostr world. @@ -127,6 +128,29 @@ void main() { ); }); + test( + 'small-account snapshot publishes without waiting for live EOSE', + () async { + final session = _FakeRelaySession( + memberships: [_membership(_channelA, myPk)], + metadata: [_meta(id: _channelA, name: 'general')], + ); + session.pauseNextSubscribe(); + final container = _buildContainer(session: session); + addTearDown(container.dispose); + + final channelFuture = container.read(channelsProvider.future); + await session.nextSubscribeStarted; + + final channels = await channelFuture.timeout(const Duration(seconds: 1)); + expect(channels.single.id, _channelA); + expect(session.activeChannels, isEmpty); + + session.resumePausedSubscribe(); + await _waitUntil(() => session.activeChannels.contains(_channelA)); + }, + ); + test('loaded refresh preserves newer live timestamp', () async { final channelIds = List.generate(101, _generatedChannelId); final session = _FakeRelaySession( @@ -376,6 +400,7 @@ void main() { addTearDown(container.dispose); await container.read(channelsProvider.future); + await _waitUntil(() => session.activeSubscriptionCount == 2); // One subscription per joined, non-archived channel. expect(session.subscribeFilters, hasLength(2)); @@ -426,6 +451,7 @@ void main() { addTearDown(container.dispose); await container.read(channelsProvider.future); + await _waitUntil(() => session.activeSubscriptionCount == 2); final initialSubscribeCount = session.totalSubscribeCount; await container.read(channelsProvider.notifier).refresh(); @@ -497,7 +523,7 @@ void main() { ); test( - 'post-ready terminal close retries subscription and catches up gap', + 'post-ready terminal close stays quarantined until explicit refresh', () async { final session = _FakeRelaySession( memberships: [_membership(_channelA, myPk)], @@ -512,9 +538,14 @@ void main() { session.terminallyClose(_channelA); expect(session.activeChannels, isEmpty); + final subscribeCountAfterClose = session.totalSubscribeCount; final missedEvent = _message(_channelA, createdAt: 50); session.unreadEvents = [missedEvent]; + await Future.delayed(const Duration(milliseconds: 10)); + expect(session.totalSubscribeCount, subscribeCountAfterClose); + expect(session.activeChannels, isEmpty); + await container.read(channelsProvider.notifier).refresh(); await _waitUntil( @@ -958,6 +989,7 @@ void main() { final initial = await container.read(channelsProvider.future); expect(initial.single.name, 'general'); + await _waitUntil(() => session.activeSubscriptionCount == 1); expect(session.subscribeFilters, hasLength(1)); session.setStatus(SessionStatus.reconnecting); @@ -1417,7 +1449,8 @@ class _FakeRelaySession extends RelaySessionNotifier { void Function(NostrEvent) onEvent, { void Function(String message)? onClosed, Future? cancelled, - }) => _subscribeFake(filter, onEvent, onClosed: onClosed); + }) => + _subscribeFake(filter, onEvent, onClosed: onClosed, cancelled: cancelled); @override Future subscribe( @@ -1430,6 +1463,7 @@ class _FakeRelaySession extends RelaySessionNotifier { NostrFilter filter, void Function(NostrEvent) onEvent, { void Function(String message)? onClosed, + Future? cancelled, }) async { totalSubscribeCount++; subscribeFilters.add(filter); @@ -1444,7 +1478,18 @@ class _FakeRelaySession extends RelaySessionNotifier { final paused = _pausedSubscribe; if (paused != null && !_subscribeStarted!.isCompleted) { _subscribeStarted!.complete(); - await paused.future; + if (cancelled == null) { + await paused.future; + } else { + final wasCancelled = await Future.any([ + paused.future.then((_) => false), + cancelled.then((_) => true), + ]); + if (wasCancelled) { + subscribeFilters.remove(filter); + throw const RelaySubscriptionCancelledException(); + } + } _pausedSubscribe = null; _subscribeStarted = null; }