diff --git a/apps/app_seeds/lib/app.dart b/apps/app_seeds/lib/app.dart index c05f58d..88fc885 100644 --- a/apps/app_seeds/lib/app.dart +++ b/apps/app_seeds/lib/app.dart @@ -9,6 +9,7 @@ import 'data/variety_repository.dart'; import 'i18n/strings.g.dart'; import 'services/auto_backup_service.dart'; import 'services/coarse_location.dart'; +import 'services/inbox_service.dart'; import 'services/message_store.dart'; import 'services/offer_outbox.dart'; import 'services/onboarding_store.dart'; @@ -47,6 +48,7 @@ class TaneApp extends StatelessWidget { this.messageStore, this.profileStore, this.profileCache, + this.inbox, this.showIntro = false, this.autoBackup, super.key, @@ -61,6 +63,7 @@ class TaneApp extends StatelessWidget { messageStore, profileStore, profileCache, + inbox, ); final VarietyRepository repository; @@ -86,6 +89,9 @@ class TaneApp extends StatelessWidget { /// Optional cache of peers' published display names. final ProfileCache? profileCache; + + /// App-wide inbox listener; drives the messages list's live refresh. + final InboxService? inbox; final bool showIntro; /// Drives silent periodic backups off the app lifecycle. Null in widget tests @@ -104,6 +110,7 @@ class TaneApp extends StatelessWidget { MessageStore? messageStore, ProfileStore? profileStore, ProfileCache? profileCache, + InboxService? inbox, ) { return GoRouter( initialLocation: showIntro ? '/intro' : '/', @@ -131,6 +138,7 @@ class TaneApp extends StatelessWidget { social: social, settings: socialSettings, profileCache: profileCache, + inbox: inbox, ), ), if (social != null && socialSettings != null && profileStore != null) diff --git a/apps/app_seeds/lib/di/injector.dart b/apps/app_seeds/lib/di/injector.dart index 79537e2..c5edfa5 100644 --- a/apps/app_seeds/lib/di/injector.dart +++ b/apps/app_seeds/lib/di/injector.dart @@ -21,6 +21,7 @@ import '../services/auto_backup_store.dart'; import '../services/export_import_service.dart'; import '../services/file_picker_file_service.dart'; import '../services/file_service.dart'; +import '../services/inbox_service.dart'; import '../services/locale_store.dart'; import '../services/ocr/label_text_extractor.dart'; import '../services/ocr/ocr_language.dart'; @@ -167,7 +168,18 @@ Future configureDependencies() async { // Optional: absent only if the derivation above failed — then the app runs // inventory-only, by design (market/chat/profile hidden). if (socialService != null) { - getIt.registerSingleton(socialService); + getIt + ..registerSingleton(socialService) + // App-wide inbox listener so private messages arrive (and land in the + // inbox list) even when the specific chat isn't open. Started in `main`. + ..registerSingleton( + InboxService( + social: socialService, + settings: getIt(), + store: getIt(), + profileCache: getIt(), + ), + ); } // Registered LAST: the "fully wired" sentinel the guard above checks. Anything diff --git a/apps/app_seeds/lib/main.dart b/apps/app_seeds/lib/main.dart index 2f82f29..15c9b2b 100644 --- a/apps/app_seeds/lib/main.dart +++ b/apps/app_seeds/lib/main.dart @@ -1,3 +1,5 @@ +import 'dart:async'; + import 'package:flutter/widgets.dart'; import 'app.dart'; @@ -7,6 +9,7 @@ import 'di/injector.dart'; import 'i18n/strings.g.dart'; import 'services/auto_backup_service.dart'; import 'services/coarse_location.dart'; +import 'services/inbox_service.dart'; import 'services/locale_store.dart'; import 'services/message_store.dart'; import 'services/offer_outbox.dart'; @@ -26,6 +29,11 @@ Future main() async { final savedLocale = await getIt().saved(); if (savedLocale != null) LocaleSettings.setLocaleSync(savedLocale); final onboarding = getIt(); + final inbox = + getIt.isRegistered() ? getIt() : null; + // Start listening for incoming private messages app-wide (foreground), so + // they arrive and populate the inbox list even before their chat is opened. + if (inbox != null) unawaited(inbox.start()); runApp( TranslationProvider( child: TaneApp( @@ -41,6 +49,7 @@ Future main() async { messageStore: getIt(), profileStore: getIt(), profileCache: getIt(), + inbox: inbox, showIntro: !await onboarding.introSeen(), autoBackup: getIt.isRegistered() ? getIt() diff --git a/apps/app_seeds/lib/services/inbox_service.dart b/apps/app_seeds/lib/services/inbox_service.dart new file mode 100644 index 0000000..5d61ca0 --- /dev/null +++ b/apps/app_seeds/lib/services/inbox_service.dart @@ -0,0 +1,140 @@ +import 'dart:async'; + +import 'package:commons_core/commons_core.dart'; +import 'package:connectivity_plus/connectivity_plus.dart'; +import 'package:flutter/foundation.dart'; + +import 'message_store.dart'; +import 'profile_cache.dart'; +import 'social_service.dart'; +import 'social_settings.dart'; + +/// App-wide inbox listener for private messages (NIP-17). +/// +/// Without this, a message only arrived while its specific chat screen was +/// open — so a first message from a new peer was invisible (nothing in local +/// storage, so nothing in the inbox list, so you never opened the chat, so you +/// never subscribed). This keeps ONE long-lived subscription for the whole app +/// while online: it persists every incoming message to the [MessageStore] and +/// fires [changes] so the inbox list refreshes. Reconnects when the network +/// returns; degrades to nothing offline (local-first). +/// +/// Foreground only — background/push delivery is a later concern. +class InboxService { + InboxService({ + required SocialService social, + required SocialSettings settings, + required MessageStore store, + ProfileCache? profileCache, + }) : _social = social, + _settings = settings, + _store = store, + _profileCache = profileCache; + + final SocialService _social; + final SocialSettings _settings; + final MessageStore _store; + final ProfileCache? _profileCache; + + final _changes = StreamController.broadcast(); + SocialSession? _session; + StreamSubscription? _inboxSub; + StreamSubscription>? _connSub; + bool _started = false; + bool _connecting = false; + + /// Fires (with no payload) after a NEW message is persisted — the inbox list + /// listens to this to reload. Broadcast, so several screens can listen. + Stream get changes => _changes.stream; + + /// Begins listening. Idempotent. Connects now if online, and (re)connects + /// whenever the network returns. Safe to call once at app start. + Future start() async { + if (_started) return; + _started = true; + try { + _connSub = Connectivity().onConnectivityChanged.listen((results) { + final offline = + results.isEmpty || results.every((r) => r == ConnectivityResult.none); + if (offline) { + _dropSession(); + } else { + unawaited(_connect()); + } + }); + } catch (_) { + // Platform without connectivity support — just try once below. + } + await _connect(); + } + + /// Opens a session and subscribes to the inbox, if not already connected. + /// Guarded so overlapping connectivity events can't open two sessions. + Future _connect() async { + if (_session != null || _connecting) return; // already (being) connected + _connecting = true; + try { + final relays = await _settings.relayUrls(); + if (relays.isEmpty) return; // offline / unconfigured + final session = await _social.openSession(relays); + if (!_started) { + await session.close(); // stopped while connecting + return; + } + _session = session; + _inboxSub = session.messages.inbox().listen( + ingest, + onError: (_) => _dropSession(), // relay dropped — retry on next change + ); + } catch (_) { + // No relay reachable — a later connectivity change retries. + } finally { + _connecting = false; + } + } + + /// Persists one incoming [message] (deduped) and, when it's new, fires + /// [changes] and best-effort caches the sender's display name. Separated from + /// the transport so it can be unit-tested without a relay. + @visibleForTesting + Future ingest(PrivateMessage message) async { + final stored = await _store.append(message.fromPubkey, message); + if (!stored) return; // a re-delivered duplicate + if (!_changes.isClosed) _changes.add(null); + await _cacheName(message.fromPubkey); + } + + Future _cacheName(String peerPubkey) async { + final cache = _profileCache; + final session = _session; + if (cache == null || session == null) return; + if (await cache.name(peerPubkey) != null) return; // already known + try { + final profile = await session.profile.fetch(peerPubkey); + if (profile != null && profile.name.isNotEmpty) { + await cache.setName(peerPubkey, profile.name); + if (!_changes.isClosed) _changes.add(null); // re-render with the name + } + } catch (_) { + // best effort — the short key shows meanwhile + } + } + + void _dropSession() { + unawaited(_inboxSub?.cancel()); + _inboxSub = null; + unawaited(_session?.close()); + _session = null; + } + + Future stop() async { + _started = false; + await _connSub?.cancel(); + _connSub = null; + await _inboxSub?.cancel(); + _inboxSub = null; + await _session?.close(); + _session = null; + if (!_changes.isClosed) await _changes.close(); + } +} diff --git a/apps/app_seeds/lib/services/message_store.dart b/apps/app_seeds/lib/services/message_store.dart index 3ffd67f..e798969 100644 --- a/apps/app_seeds/lib/services/message_store.dart +++ b/apps/app_seeds/lib/services/message_store.dart @@ -1,3 +1,4 @@ +import 'dart:async'; import 'dart:convert'; import 'package:commons_core/commons_core.dart'; @@ -25,6 +26,13 @@ class MessageStore { MessageStore(this._store); final SecretStore _store; + + /// Serializes [append]'s read-modify-write. Several sources now write the same + /// conversation concurrently (the app-wide inbox listener and an open chat's + /// own subscription), so without this two near-simultaneous messages could + /// read the same history and the second write would clobber the first. + Future _writeTail = Future.value(); + static const _prefix = 'tane.social.chat.'; /// Index of peers we have a conversation with (the keystore is key/value with @@ -53,9 +61,25 @@ class MessageStore { } /// Appends [message] to the conversation with [peerPubkey] (trimming to the - /// cap) and records the peer in the conversation index. - Future append(String peerPubkey, PrivateMessage message) async { - final next = [...await history(peerPubkey), message]; + /// cap) and records the peer in the conversation index. Idempotent: a relay + /// re-delivers stored gift wraps on every (re)subscribe, so an identical + /// message (same sender, timestamp and text) is dropped instead of duplicated. + /// Returns true only when the message was newly stored. + Future append(String peerPubkey, PrivateMessage message) { + // Chain onto the write tail so concurrent appends run one at a time. + final result = _writeTail.then((_) => _appendLocked(peerPubkey, message)); + _writeTail = result.then((_) {}, onError: (_) {}); + return result; + } + + Future _appendLocked(String peerPubkey, PrivateMessage message) async { + final existing = await history(peerPubkey); + final isDup = existing.any((m) => + m.fromPubkey == message.fromPubkey && + m.text == message.text && + m.at.millisecondsSinceEpoch == message.at.millisecondsSinceEpoch); + if (isDup) return false; + final next = [...existing, message]; final capped = next.length > _cap ? next.sublist(next.length - _cap) : next; await _store.write( @@ -70,6 +94,7 @@ class MessageStore { ]), ); await _rememberPeer(peerPubkey); + return true; } /// Conversations, most-recently-active first (for the messages inbox). diff --git a/apps/app_seeds/lib/ui/chat_list_screen.dart b/apps/app_seeds/lib/ui/chat_list_screen.dart index 9d0d320..a3535c7 100644 --- a/apps/app_seeds/lib/ui/chat_list_screen.dart +++ b/apps/app_seeds/lib/ui/chat_list_screen.dart @@ -4,6 +4,7 @@ import 'package:flutter/material.dart'; import 'package:go_router/go_router.dart'; import '../i18n/strings.g.dart'; +import '../services/inbox_service.dart'; import '../services/message_store.dart'; import '../services/profile_cache.dart'; import '../services/social_service.dart'; @@ -18,6 +19,7 @@ class ChatListScreen extends StatefulWidget { this.social, this.settings, this.profileCache, + this.inbox, super.key, }); @@ -26,6 +28,9 @@ class ChatListScreen extends StatefulWidget { final SocialSettings? settings; final ProfileCache? profileCache; + /// App-wide inbox listener; the list reloads whenever it reports a change. + final InboxService? inbox; + @override State createState() => _ChatListScreenState(); } @@ -33,11 +38,20 @@ class ChatListScreen extends StatefulWidget { class _ChatListScreenState extends State { List? _items; final Map _names = {}; + StreamSubscription? _changesSub; @override void initState() { super.initState(); _load(); + // Live-refresh when the app-wide inbox listener persists a new message. + _changesSub = widget.inbox?.changes.listen((_) => _load()); + } + + @override + void dispose() { + _changesSub?.cancel(); + super.dispose(); } Future _load() async { diff --git a/apps/app_seeds/test/services/inbox_service_test.dart b/apps/app_seeds/test/services/inbox_service_test.dart new file mode 100644 index 0000000..2c85de0 --- /dev/null +++ b/apps/app_seeds/test/services/inbox_service_test.dart @@ -0,0 +1,55 @@ +import 'package:commons_core/commons_core.dart'; +import 'package:flutter_test/flutter_test.dart'; +import 'package:tane/services/inbox_service.dart'; +import 'package:tane/services/message_store.dart'; +import 'package:tane/services/social_service.dart'; +import 'package:tane/services/social_settings.dart'; + +import '../support/test_support.dart'; + +/// The app-wide inbox listener persists incoming messages and announces changes, +/// so the inbox list refreshes even when the specific chat isn't open. Driven +/// through the [InboxService.ingest] seam so no relay/network is involved. +void main() { + const seedHex = + '000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f'; + + late MessageStore store; + late InboxService inbox; + + setUp(() async { + store = MessageStore(InMemorySecretStore()); + inbox = InboxService( + social: await SocialService.fromRootSeedHex(seedHex), + settings: SocialSettings(InMemorySecretStore()), + store: store, + ); + }); + + tearDown(() => inbox.stop()); + + PrivateMessage msg(String from, String text, int atMs) => PrivateMessage( + fromPubkey: from, + text: text, + at: DateTime.fromMillisecondsSinceEpoch(atMs)); + + test('an incoming message is persisted into its conversation', () async { + await inbox.ingest(msg('alice', 'got seeds?', 1000)); + final convos = await store.conversations(); + expect(convos.single.peerPubkey, 'alice'); + expect(convos.single.lastText, 'got seeds?'); + }); + + test('a new message announces a change; a duplicate stays silent', () async { + final changes = []; + final sub = inbox.changes.listen(changes.add); + + await inbox.ingest(msg('alice', 'hola', 1000)); + await inbox.ingest(msg('alice', 'hola', 1000)); // relay re-delivery + await Future.delayed(Duration.zero); // let the broadcast flush + + expect(changes, hasLength(1)); // only the first, new one fired + expect(await store.history('alice'), hasLength(1)); + await sub.cancel(); + }); +} diff --git a/apps/app_seeds/test/services/message_store_test.dart b/apps/app_seeds/test/services/message_store_test.dart index 7bd742b..138ee46 100644 --- a/apps/app_seeds/test/services/message_store_test.dart +++ b/apps/app_seeds/test/services/message_store_test.dart @@ -46,6 +46,19 @@ void main() { expect(convos.last.lastText, 'back A'); }); + test('append is idempotent: a re-delivered message is not duplicated', + () async { + // Same sender + timestamp + text = the same gift wrap redelivered by a + // relay on resubscribe. It must not pile up. + expect(await store.append('peer', msg('peer', 'hola', 1000)), isTrue); + expect(await store.append('peer', msg('peer', 'hola', 1000)), isFalse); + expect(await store.history('peer'), hasLength(1)); + + // A genuinely different message (later timestamp) still lands. + expect(await store.append('peer', msg('peer', 'hola', 2000)), isTrue); + expect(await store.history('peer'), hasLength(2)); + }); + test('history is capped to the most recent 200', () async { for (var i = 0; i < 210; i++) { await store.append('peer', msg('me', 'm$i', i));