feat(chat): app-wide inbox listener so messages arrive without opening the chat
Reception only ran inside an open ChatScreen for a known peer, so a first message from a new peer was invisible: nothing in local storage -> nothing in the inbox list -> you never opened the chat -> you never subscribed. Add InboxService: one long-lived NIP-17 inbox subscription for the whole app (foreground), persisting every incoming message to MessageStore and firing a 'changes' signal the inbox list live-reloads on. Reconnects when the network returns; degrades to nothing offline. Started from main when a social identity exists. Make MessageStore.append idempotent (dedup by sender+timestamp+text) and serialized behind a write lock — the global listener and an open chat's own subscription now write the same conversation concurrently and relays re-deliver stored gift wraps on every resubscribe. Tests for both. Known trade-offs (follow-ups): foreground-only (no push yet); each of InboxService/ChatScreen/MarketScreen opens its own RelayPool (a shared connection is a later optimization).
This commit is contained in:
parent
e852b569ce
commit
73ee98206f
8 changed files with 280 additions and 4 deletions
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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<void> 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>(socialService);
|
||||
getIt
|
||||
..registerSingleton<SocialService>(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>(
|
||||
InboxService(
|
||||
social: socialService,
|
||||
settings: getIt<SocialSettings>(),
|
||||
store: getIt<MessageStore>(),
|
||||
profileCache: getIt<ProfileCache>(),
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
// Registered LAST: the "fully wired" sentinel the guard above checks. Anything
|
||||
|
|
|
|||
|
|
@ -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<void> main() async {
|
|||
final savedLocale = await getIt<LocaleStore>().saved();
|
||||
if (savedLocale != null) LocaleSettings.setLocaleSync(savedLocale);
|
||||
final onboarding = getIt<OnboardingStore>();
|
||||
final inbox =
|
||||
getIt.isRegistered<InboxService>() ? getIt<InboxService>() : 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<void> main() async {
|
|||
messageStore: getIt<MessageStore>(),
|
||||
profileStore: getIt<ProfileStore>(),
|
||||
profileCache: getIt<ProfileCache>(),
|
||||
inbox: inbox,
|
||||
showIntro: !await onboarding.introSeen(),
|
||||
autoBackup: getIt.isRegistered<AutoBackupService>()
|
||||
? getIt<AutoBackupService>()
|
||||
|
|
|
|||
140
apps/app_seeds/lib/services/inbox_service.dart
Normal file
140
apps/app_seeds/lib/services/inbox_service.dart
Normal file
|
|
@ -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<void>.broadcast();
|
||||
SocialSession? _session;
|
||||
StreamSubscription<PrivateMessage>? _inboxSub;
|
||||
StreamSubscription<List<ConnectivityResult>>? _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<void> 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<void> 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<void> _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<void> 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<void> _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<void> 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();
|
||||
}
|
||||
}
|
||||
|
|
@ -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<void> _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<void> 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<bool> 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<bool> _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).
|
||||
|
|
|
|||
|
|
@ -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<ChatListScreen> createState() => _ChatListScreenState();
|
||||
}
|
||||
|
|
@ -33,11 +38,20 @@ class ChatListScreen extends StatefulWidget {
|
|||
class _ChatListScreenState extends State<ChatListScreen> {
|
||||
List<ChatSummary>? _items;
|
||||
final Map<String, String> _names = {};
|
||||
StreamSubscription<void>? _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<void> _load() async {
|
||||
|
|
|
|||
55
apps/app_seeds/test/services/inbox_service_test.dart
Normal file
55
apps/app_seeds/test/services/inbox_service_test.dart
Normal file
|
|
@ -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 = <void>[];
|
||||
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<void>.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();
|
||||
});
|
||||
}
|
||||
|
|
@ -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));
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue