diff --git a/apps/app_seeds/lib/bootstrap.dart b/apps/app_seeds/lib/bootstrap.dart index f6a68c1..9df9c4e 100644 --- a/apps/app_seeds/lib/bootstrap.dart +++ b/apps/app_seeds/lib/bootstrap.dart @@ -22,6 +22,7 @@ import 'services/social_account_store.dart'; import 'services/social_connection.dart'; import 'services/social_service.dart'; import 'services/social_settings.dart'; +import 'services/sync_service.dart'; import 'services/trust_referents.dart'; import 'services/wot_settings.dart'; import 'ui/theme.dart'; @@ -65,9 +66,12 @@ class _BootstrapState extends State { // Ask for notification permission and set up the OS channel (no-op on // unsupported platforms). Done before the inbox starts listening. if (notifications != null) await notifications.initialize(); - // Subscribe the inbox listener BEFORE the shared connection starts + final sync = + getIt.isRegistered() ? getIt() : null; + // Subscribe the inbox + sync listeners BEFORE the shared connection starts // connecting, so the first session is caught; then bring the connection up. inbox?.start(); + sync?.start(); connection?.start(); return TaneApp( diff --git a/apps/app_seeds/lib/di/injector.dart b/apps/app_seeds/lib/di/injector.dart index bd3108f..04abfad 100644 --- a/apps/app_seeds/lib/di/injector.dart +++ b/apps/app_seeds/lib/di/injector.dart @@ -19,6 +19,7 @@ import '../security/secret_store.dart'; import '../security/secure_key_store.dart'; import '../services/auto_backup_service.dart'; import '../services/auto_backup_store.dart'; +import '../services/device_id_store.dart'; import '../services/export_import_service.dart'; import '../services/file_picker_file_service.dart'; import '../services/file_service.dart'; @@ -40,6 +41,7 @@ import '../services/social_account_store.dart'; import '../services/social_connection.dart'; import '../services/social_service.dart'; import '../services/social_settings.dart'; +import '../services/sync_service.dart'; import '../services/trust_referents.dart'; import '../services/unread_service.dart'; import '../services/wot_settings.dart'; @@ -88,6 +90,10 @@ Future configureDependencies() async { final activeAccount = await accounts.active(); final scope = socialAccountScope(activeAccount); + // Stable per-install id for device-to-device sync (distinct from the + // seed-derived CRDT node id, which is shared across a user's devices). + final deviceId = await DeviceIdStore(secretStore).deviceId(); + final database = AppDatabase( openEncryptedExecutor(await _databaseFile(), dbKeyHex), ); @@ -104,8 +110,11 @@ Future configureDependencies() async { // profile stay hidden) rather than blanking the whole app. SocialService? socialService; try { - socialService = - await SocialService.fromRootSeedHex(rootSeedHex, account: activeAccount); + socialService = await SocialService.fromRootSeedHex( + rootSeedHex, + account: activeAccount, + deviceId: deviceId, + ); } catch (e, s) { debugPrint('Social identity derivation failed; inventory-only mode: $e\n$s'); } @@ -224,6 +233,16 @@ Future configureDependencies() async { unread: getIt(), notifications: getIt(), ), + ) + // Replicates the inventory across this identity's own devices over the + // shared connection. Started in `Bootstrap`. + ..registerSingleton( + SyncService( + connection: connection, + io: getIt(), + localChanges: database.tableUpdates().map((_) {}), + selfDeviceId: deviceId, + ), ); } @@ -244,8 +263,14 @@ Future switchSocialAccount(int account) async { await getIt().setActive(account); final scope = socialAccountScope(account); final rootSeedHex = await getIt().rootSeedHex(); + // Per-install, so it survives an identity switch. + final deviceId = await DeviceIdStore(secretStore).deviceId(); - // Tear down the old identity's listener + shared connection before replacing. + // Tear down the old identity's listener + sync + shared connection first. + if (getIt.isRegistered()) { + await getIt().stop(); + await getIt.unregister(); + } if (getIt.isRegistered()) { await getIt().stop(); await getIt.unregister(); @@ -275,8 +300,11 @@ Future switchSocialAccount(int account) async { ..registerSingleton( UnreadService(getIt(), secretStore)); - final social = - await SocialService.fromRootSeedHex(rootSeedHex, account: account); + final social = await SocialService.fromRootSeedHex( + rootSeedHex, + account: account, + deviceId: deviceId, + ); final connection = SocialConnection(social: social, settings: getIt()); final inbox = InboxService( @@ -287,12 +315,20 @@ Future switchSocialAccount(int account) async { unread: getIt(), notifications: getIt(), ); + final sync = SyncService( + connection: connection, + io: getIt(), + localChanges: getIt().tableUpdates().map((_) {}), + selfDeviceId: deviceId, + ); getIt ..registerSingleton(social) ..registerSingleton(connection) - ..registerSingleton(inbox); - // Subscribe the listener BEFORE the connection starts connecting. + ..registerSingleton(inbox) + ..registerSingleton(sync); + // Subscribe the listeners BEFORE the connection starts connecting. inbox.start(); + sync.start(); connection.start(); } diff --git a/apps/app_seeds/lib/services/device_id_store.dart b/apps/app_seeds/lib/services/device_id_store.dart new file mode 100644 index 0000000..9c2e7f4 --- /dev/null +++ b/apps/app_seeds/lib/services/device_id_store.dart @@ -0,0 +1,34 @@ +import 'dart:math'; + +import '../security/secret_store.dart'; + +/// A stable per-INSTALL device id, kept in the keystore. Distinct from the +/// seed-derived CRDT node id (`rootSeedHex.substring(0,16)`), which is IDENTICAL +/// across a user's devices — so it can't tell them apart. Device-to-device sync +/// needs each device to own a separate record, hence a random per-install id. +class DeviceIdStore { + DeviceIdStore(this._store, {Random? random}) : _random = random ?? Random.secure(); + + final SecretStore _store; + final Random _random; + + static const _key = 'tane.device.id'; + + Future? _cached; + + /// This install's device id, generated and persisted on first use. + Future deviceId() => _cached ??= _readOrCreate(); + + Future _readOrCreate() async { + final existing = await _store.read(_key); + if (existing != null && existing.isNotEmpty) return existing; + final id = _randomHex(8); // 64 bits — plenty to avoid device collisions + await _store.write(_key, id); + return id; + } + + String _randomHex(int bytes) => [ + for (var i = 0; i < bytes; i++) + _random.nextInt(256).toRadixString(16).padLeft(2, '0'), + ].join(); +} diff --git a/apps/app_seeds/lib/services/export_import_service.dart b/apps/app_seeds/lib/services/export_import_service.dart index bda48e7..ef5a450 100644 --- a/apps/app_seeds/lib/services/export_import_service.dart +++ b/apps/app_seeds/lib/services/export_import_service.dart @@ -9,6 +9,7 @@ import '../data/export_import/inventory_snapshot.dart'; import '../data/variety_repository.dart'; import '../security/secure_key_store.dart'; import 'file_service.dart'; +import 'inventory_snapshot_io.dart'; /// A picked backup file awaiting restore — kept in memory so asking the user /// for a recovery code never forces a second file dialog. @@ -27,7 +28,7 @@ class PendingBackup { /// Backups are **sealed** (encrypted under a key derived from the root seed; /// backup-and-recovery.md): a copy that lands on someone's cloud folder /// reveals nothing. The printed recovery code opens them on any device. -class ExportImportService { +class ExportImportService implements InventorySnapshotIO { ExportImportService({ required VarietyRepository repository, required FileService files, @@ -49,6 +50,18 @@ class ExportImportService { /// The backup file extension. The content is the sealed interchange JSON. static const backupExtension = 'tanemaki'; + /// Raw (unsealed) interchange snapshot for device-to-device sync — the same + /// JSON as a backup, but the sync transport (not this) does the encryption. + @override + Future> buildSnapshot() async => + utf8.encode(_jsonCodec.encode(await _repository.exportInventory())); + + /// Merges a raw interchange snapshot from another device (LWW, idempotent). + @override + Future applySnapshot(List bytes) async { + await _repository.importInventory(_jsonCodec.decode(utf8.decode(bytes))); + } + /// Builds the full inventory as a sealed (encrypted) backup — the same bytes /// [exportBackup] writes to a file. Shared with the automatic backup so both /// paths seal identically. diff --git a/apps/app_seeds/lib/services/inventory_snapshot_io.dart b/apps/app_seeds/lib/services/inventory_snapshot_io.dart new file mode 100644 index 0000000..7021bcc --- /dev/null +++ b/apps/app_seeds/lib/services/inventory_snapshot_io.dart @@ -0,0 +1,12 @@ +/// Reads/writes the inventory as a raw (unsealed) interchange snapshot — the +/// same JSON a backup uses, but WITHOUT the file-level sealing, because the sync +/// transport already encrypts on the wire. [SyncService] depends on this +/// interface (not the whole export service) so its logic is fakeable. +abstract interface class InventorySnapshotIO { + /// The full inventory serialized to interchange bytes. + Future> buildSnapshot(); + + /// Merges an incoming snapshot into the local inventory (LWW by HLC, + /// idempotent — re-applying the same or older state is a no-op). + Future applySnapshot(List bytes); +} diff --git a/apps/app_seeds/lib/services/social_service.dart b/apps/app_seeds/lib/services/social_service.dart index db4cfab..17d02dd 100644 --- a/apps/app_seeds/lib/services/social_service.dart +++ b/apps/app_seeds/lib/services/social_service.dart @@ -9,14 +9,23 @@ import 'package:commons_core/commons_core.dart'; /// social sessions on demand. Local-first: constructing this is cheap and /// offline; nothing connects to a relay until [openSession] is called, so the /// app runs fully without network and the social layer only enriches. +/// The app-data namespace this identity's devices sync their inventory under. +/// Lives here (app layer), not in `commons_core`, which stays seed-agnostic. +const kInventorySyncNamespace = 'org.comunes.tane/inventory'; + class SocialService { SocialService({ required this.identity, this.account = 0, this.rootSeedHex = '', + this.deviceId = '', List relays = const [], }) : relays = List.unmodifiable(relays); + /// This install's device id (for device-to-device sync). Empty in tests / + /// when sync isn't wired. + final String deviceId; + /// The derived Nostr identity (secp256k1). Same key across reinstalls that /// restore the same seed (for a given [account]). final NostrIdentity identity; @@ -43,6 +52,7 @@ class SocialService { static Future fromRootSeedHex( String rootSeedHex, { int account = 0, + String deviceId = '', List relays = const [], }) async { final identity = await NostrKeyDerivation.deriveFromSeed( @@ -53,6 +63,7 @@ class SocialService { identity: identity, account: account, rootSeedHex: rootSeedHex, + deviceId: deviceId, relays: relays, ); } @@ -74,18 +85,23 @@ class SocialService { /// skipped; throws only when none are reachable. Caller closes it. Future openSession(List relayUrls) async { final pool = await RelayPool.connect(relayUrls, identity: identity); - return SocialSession(pool); + return SocialSession(pool, deviceId: deviceId); } } /// One relay channel with the three transports on top — the /// "one channel, three interfaces" shape, at the app layer. class SocialSession { - SocialSession(this._channel) + SocialSession(this._channel, {String deviceId = ''}) : offers = NostrOfferTransport(_channel), messages = NostrMessageTransport(_channel), trust = NostrTrustTransport(_channel), - profile = NostrProfileTransport(_channel); + profile = NostrProfileTransport(_channel), + sync = NostrSyncTransport( + _channel, + namespace: kInventorySyncNamespace, + deviceId: deviceId, + ); final NostrChannel _channel; @@ -94,6 +110,9 @@ class SocialSession { final TrustTransport trust; final ProfileTransport profile; + /// Device-to-device inventory sync for THIS identity's own devices. + final SyncTransport sync; + Future close() => _channel.close(); } diff --git a/apps/app_seeds/lib/services/sync_service.dart b/apps/app_seeds/lib/services/sync_service.dart new file mode 100644 index 0000000..e585156 --- /dev/null +++ b/apps/app_seeds/lib/services/sync_service.dart @@ -0,0 +1,122 @@ +import 'dart:async'; + +import 'package:commons_core/commons_core.dart'; +import 'package:flutter/foundation.dart'; + +import 'inventory_snapshot_io.dart'; +import 'social_connection.dart'; +import 'social_service.dart'; + +/// Keeps this identity's inventory replicated across its OWN devices, over the +/// shared [SocialConnection]'s [SyncTransport]. +/// +/// On (re)connect it publishes this device's snapshot and subscribes to the +/// others'; a local edit (debounced) republishes. Incoming snapshots are merged +/// with the existing idempotent LWW importer — so applying one you already have +/// writes nothing and can't echo back into a push loop. A content check skips +/// re-publishing an unchanged snapshot, bounding the ping-pong to convergence. +/// +/// Foreground only, like the inbox — a background sync daemon is a later concern. +class SyncService { + SyncService({ + required SocialConnection connection, + required InventorySnapshotIO io, + required Stream localChanges, + required String selfDeviceId, + Duration debounce = const Duration(seconds: 2), + }) : _connection = connection, + _io = io, + _localChanges = localChanges, + _selfDeviceId = selfDeviceId, + _debounce = debounce; + + final SocialConnection _connection; + final InventorySnapshotIO _io; + final Stream _localChanges; + final String _selfDeviceId; + final Duration _debounce; + + StreamSubscription? _sessionsSub; + StreamSubscription? _snapSub; + StreamSubscription? _changesSub; + Timer? _debounceTimer; + SocialSession? _session; + List? _lastPushed; // the snapshot we last published (for change dedup) + bool _started = false; + + /// Begins syncing. Subscribe to the connection's sessions BEFORE it starts + /// connecting, so the first session is caught too. + void start() { + if (_started) return; + _started = true; + _sessionsSub = _connection.sessions.listen(_onSession); + _changesSub = _localChanges.listen((_) => _schedulePush()); + final current = _connection.current; + if (current != null) _onSession(current); + } + + void _onSession(SocialSession? session) { + if (identical(session, _session) && _snapSub != null) return; + unawaited(_snapSub?.cancel()); + _snapSub = null; + _session = session; + if (session != null) { + _snapSub = session.sync.snapshots().listen(handleRemote, onError: (_) {}); + _lastPushed = null; // a fresh session should re-publish + unawaited(pushLocal()); // publish our latest on (re)connect + } + } + + /// Merges one incoming [snapshot], skipping our own device's echo. A testable + /// seam (no relay). + @visibleForTesting + Future handleRemote(SyncSnapshot snapshot) async { + if (snapshot.deviceId == _selfDeviceId) return; // our own — nothing to do + try { + await _io.applySnapshot(snapshot.data); + } catch (_) { + // A malformed/old snapshot merges to nothing; ignore. + } + } + + void _schedulePush() { + _debounceTimer?.cancel(); + _debounceTimer = Timer(_debounce, () => unawaited(pushLocal())); + } + + /// Publishes this device's current inventory, unless it's byte-identical to + /// what we last published (so a merge that changed nothing doesn't loop). + @visibleForTesting + Future pushLocal() async { + final session = _session; + if (session == null) return; + try { + final bytes = await _io.buildSnapshot(); + if (_sameBytes(bytes, _lastPushed)) return; + await session.sync.pushSnapshot(bytes); + _lastPushed = bytes; + } catch (_) { + // offline / relay refused — the next connect or change retries. + } + } + + static bool _sameBytes(List a, List? b) { + if (b == null || a.length != b.length) return false; + for (var i = 0; i < a.length; i++) { + if (a[i] != b[i]) return false; + } + return true; + } + + Future stop() async { + _started = false; + _debounceTimer?.cancel(); + await _changesSub?.cancel(); + _changesSub = null; + await _sessionsSub?.cancel(); + _sessionsSub = null; + await _snapSub?.cancel(); + _snapSub = null; + _session = null; + } +} diff --git a/apps/app_seeds/test/services/device_id_store_test.dart b/apps/app_seeds/test/services/device_id_store_test.dart new file mode 100644 index 0000000..3ae832b --- /dev/null +++ b/apps/app_seeds/test/services/device_id_store_test.dart @@ -0,0 +1,31 @@ +import 'dart:math'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:tane/services/device_id_store.dart'; + +import '../support/test_support.dart'; + +void main() { + test('generates a 64-bit hex id and persists it', () async { + final secret = InMemorySecretStore(); + final id = await DeviceIdStore(secret).deviceId(); + expect(id, matches(RegExp(r'^[0-9a-f]{16}$'))); + + // A fresh store over the same keystore reads the SAME id (stable install). + final again = await DeviceIdStore(secret).deviceId(); + expect(again, id); + }); + + test('is stable across calls on one instance', () async { + final store = DeviceIdStore(InMemorySecretStore()); + expect(await store.deviceId(), await store.deviceId()); + }); + + test('two fresh installs get different ids', () async { + final a = await DeviceIdStore(InMemorySecretStore(), random: Random(1)) + .deviceId(); + final b = await DeviceIdStore(InMemorySecretStore(), random: Random(2)) + .deviceId(); + expect(a, isNot(b)); + }); +} diff --git a/apps/app_seeds/test/services/export_import_service_test.dart b/apps/app_seeds/test/services/export_import_service_test.dart index d1fefab..c427c88 100644 --- a/apps/app_seeds/test/services/export_import_service_test.dart +++ b/apps/app_seeds/test/services/export_import_service_test.dart @@ -141,6 +141,27 @@ void main() { expect(summary.inserted, 1); }); + test('sync snapshot round-trips UNSEALED between devices, idempotently', + () async { + final (serviceA, _, _) = newService(dbA); + final repoA = newTestRepository(dbA); + await repoA.addQuickVariety(label: 'Sync tomato'); + + final snapshot = await serviceA.buildSnapshot(); + // Unsealed: plain interchange JSON — the sync transport does the encryption, + // so this must NOT be double-sealed. + expect(BackupBox.looksSealed(Uint8List.fromList(snapshot)), isFalse); + expect(utf8.decode(snapshot).contains('Sync tomato'), isTrue); + + final (serviceB, _, _) = newService(dbB); + await serviceB.applySnapshot(snapshot); + expect((await dbB.select(dbB.varieties).get()).single.label, 'Sync tomato'); + + // Idempotent: re-applying the same snapshot changes nothing. + await serviceB.applySnapshot(snapshot); + expect(await dbB.select(dbB.varieties).get(), hasLength(1)); + }); + test('legacy plain exports still restore (no seal, direct parse)', () async { final repoA = newTestRepository(dbA); await repoA.addQuickVariety(label: 'Old export'); diff --git a/apps/app_seeds/test/services/sync_service_test.dart b/apps/app_seeds/test/services/sync_service_test.dart new file mode 100644 index 0000000..e7e7eef --- /dev/null +++ b/apps/app_seeds/test/services/sync_service_test.dart @@ -0,0 +1,59 @@ +import 'package:commons_core/commons_core.dart'; +import 'package:flutter_test/flutter_test.dart'; +import 'package:tane/services/inventory_snapshot_io.dart'; +import 'package:tane/services/social_connection.dart'; +import 'package:tane/services/social_service.dart'; +import 'package:tane/services/social_settings.dart'; +import 'package:tane/services/sync_service.dart'; + +import '../support/test_support.dart'; + +/// Records what it's asked to build/apply — no DB, no relay. +class _FakeIO implements InventorySnapshotIO { + final applied = >[]; + List snapshot = const [1, 2, 3]; + + @override + Future> buildSnapshot() async => snapshot; + + @override + Future applySnapshot(List bytes) async => applied.add(bytes); +} + +void main() { + const seedHex = + '000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f'; + + Future make(_FakeIO io) async => SyncService( + connection: SocialConnection( + social: await SocialService.fromRootSeedHex(seedHex), + settings: SocialSettings(InMemorySecretStore()), + open: (_) async => throw StateError('unused'), + online: const Stream.empty(), + ), + io: io, + localChanges: const Stream.empty(), + selfDeviceId: 'me', + ); + + SyncSnapshot snap(String device, List data) => + SyncSnapshot(deviceId: device, data: data, at: DateTime(2026)); + + test('merges another device\'s snapshot into the local inventory', () async { + final io = _FakeIO(); + final sync = await make(io); + await sync.handleRemote(snap('other', const [9, 9, 9])); + expect(io.applied, [ + [9, 9, 9] + ]); + await sync.stop(); + }); + + test('ignores our own device\'s snapshot (no self-echo)', () async { + final io = _FakeIO(); + final sync = await make(io); + await sync.handleRemote(snap('me', const [1])); + expect(io.applied, isEmpty); + await sync.stop(); + }); +}