feat(sync): device-to-device inventory sync, end to end

Wires the encrypted sync transport into a working replication between one
identity's own devices, reusing the CRDT-ready model (HLC + tombstones +
LWW) the data model was built for.

- SocialSession gains a 5th interface, sync, over the shared connection,
  built with a per-install deviceId + the app's inventory namespace.
- DeviceIdStore: a stable random per-install id in the keystore — distinct
  from the seed-derived CRDT node id, which is IDENTICAL across a user's
  devices (so it couldn't tell them apart / they'd clobber each other's
  replaceable record).
- InventorySnapshotIO (implemented by ExportImportService): build/apply the
  inventory as the SAME interchange JSON as backups but UNSEALED — the sync
  transport does the encryption, so it must not double-seal. Apply = the
  existing idempotent LWW importInventory.
- SyncService: on (re)connect publishes this device's snapshot + subscribes
  to the others'; a local edit (debounced, off tableUpdates) republishes;
  incoming snapshots merge idempotently; skips its own echo; a byte check
  skips re-publishing an unchanged snapshot so the ping-pong converges.
  Foreground only, like the inbox.
- DI/Bootstrap/switchSocialAccount wire and lifecycle it per identity.

Tests: DeviceIdStore, ExportImportService snapshot round-trip (unsealed +
idempotent), SyncService.handleRemote (applies others', skips own).

Follow-ups: deltas vs full-snapshot pushes; background sync; conflict UI.
This commit is contained in:
vjrj 2026-07-10 23:45:40 +02:00
parent 182c334883
commit 225880fc64
10 changed files with 363 additions and 12 deletions

View file

@ -22,6 +22,7 @@ import 'services/social_account_store.dart';
import 'services/social_connection.dart'; import 'services/social_connection.dart';
import 'services/social_service.dart'; import 'services/social_service.dart';
import 'services/social_settings.dart'; import 'services/social_settings.dart';
import 'services/sync_service.dart';
import 'services/trust_referents.dart'; import 'services/trust_referents.dart';
import 'services/wot_settings.dart'; import 'services/wot_settings.dart';
import 'ui/theme.dart'; import 'ui/theme.dart';
@ -65,9 +66,12 @@ class _BootstrapState extends State<Bootstrap> {
// Ask for notification permission and set up the OS channel (no-op on // Ask for notification permission and set up the OS channel (no-op on
// unsupported platforms). Done before the inbox starts listening. // unsupported platforms). Done before the inbox starts listening.
if (notifications != null) await notifications.initialize(); if (notifications != null) await notifications.initialize();
// Subscribe the inbox listener BEFORE the shared connection starts final sync =
getIt.isRegistered<SyncService>() ? getIt<SyncService>() : null;
// Subscribe the inbox + sync listeners BEFORE the shared connection starts
// connecting, so the first session is caught; then bring the connection up. // connecting, so the first session is caught; then bring the connection up.
inbox?.start(); inbox?.start();
sync?.start();
connection?.start(); connection?.start();
return TaneApp( return TaneApp(

View file

@ -19,6 +19,7 @@ import '../security/secret_store.dart';
import '../security/secure_key_store.dart'; import '../security/secure_key_store.dart';
import '../services/auto_backup_service.dart'; import '../services/auto_backup_service.dart';
import '../services/auto_backup_store.dart'; import '../services/auto_backup_store.dart';
import '../services/device_id_store.dart';
import '../services/export_import_service.dart'; import '../services/export_import_service.dart';
import '../services/file_picker_file_service.dart'; import '../services/file_picker_file_service.dart';
import '../services/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_connection.dart';
import '../services/social_service.dart'; import '../services/social_service.dart';
import '../services/social_settings.dart'; import '../services/social_settings.dart';
import '../services/sync_service.dart';
import '../services/trust_referents.dart'; import '../services/trust_referents.dart';
import '../services/unread_service.dart'; import '../services/unread_service.dart';
import '../services/wot_settings.dart'; import '../services/wot_settings.dart';
@ -88,6 +90,10 @@ Future<void> configureDependencies() async {
final activeAccount = await accounts.active(); final activeAccount = await accounts.active();
final scope = socialAccountScope(activeAccount); 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( final database = AppDatabase(
openEncryptedExecutor(await _databaseFile(), dbKeyHex), openEncryptedExecutor(await _databaseFile(), dbKeyHex),
); );
@ -104,8 +110,11 @@ Future<void> configureDependencies() async {
// profile stay hidden) rather than blanking the whole app. // profile stay hidden) rather than blanking the whole app.
SocialService? socialService; SocialService? socialService;
try { try {
socialService = socialService = await SocialService.fromRootSeedHex(
await SocialService.fromRootSeedHex(rootSeedHex, account: activeAccount); rootSeedHex,
account: activeAccount,
deviceId: deviceId,
);
} catch (e, s) { } catch (e, s) {
debugPrint('Social identity derivation failed; inventory-only mode: $e\n$s'); debugPrint('Social identity derivation failed; inventory-only mode: $e\n$s');
} }
@ -224,6 +233,16 @@ Future<void> configureDependencies() async {
unread: getIt<UnreadService>(), unread: getIt<UnreadService>(),
notifications: getIt<NotificationService>(), notifications: getIt<NotificationService>(),
), ),
)
// Replicates the inventory across this identity's own devices over the
// shared connection. Started in `Bootstrap`.
..registerSingleton<SyncService>(
SyncService(
connection: connection,
io: getIt<ExportImportService>(),
localChanges: database.tableUpdates().map<void>((_) {}),
selfDeviceId: deviceId,
),
); );
} }
@ -244,8 +263,14 @@ Future<void> switchSocialAccount(int account) async {
await getIt<SocialAccountStore>().setActive(account); await getIt<SocialAccountStore>().setActive(account);
final scope = socialAccountScope(account); final scope = socialAccountScope(account);
final rootSeedHex = await getIt<SecureKeyStore>().rootSeedHex(); final rootSeedHex = await getIt<SecureKeyStore>().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<SyncService>()) {
await getIt<SyncService>().stop();
await getIt.unregister<SyncService>();
}
if (getIt.isRegistered<InboxService>()) { if (getIt.isRegistered<InboxService>()) {
await getIt<InboxService>().stop(); await getIt<InboxService>().stop();
await getIt.unregister<InboxService>(); await getIt.unregister<InboxService>();
@ -275,8 +300,11 @@ Future<void> switchSocialAccount(int account) async {
..registerSingleton<UnreadService>( ..registerSingleton<UnreadService>(
UnreadService(getIt<MessageStore>(), secretStore)); UnreadService(getIt<MessageStore>(), secretStore));
final social = final social = await SocialService.fromRootSeedHex(
await SocialService.fromRootSeedHex(rootSeedHex, account: account); rootSeedHex,
account: account,
deviceId: deviceId,
);
final connection = final connection =
SocialConnection(social: social, settings: getIt<SocialSettings>()); SocialConnection(social: social, settings: getIt<SocialSettings>());
final inbox = InboxService( final inbox = InboxService(
@ -287,12 +315,20 @@ Future<void> switchSocialAccount(int account) async {
unread: getIt<UnreadService>(), unread: getIt<UnreadService>(),
notifications: getIt<NotificationService>(), notifications: getIt<NotificationService>(),
); );
final sync = SyncService(
connection: connection,
io: getIt<ExportImportService>(),
localChanges: getIt<AppDatabase>().tableUpdates().map<void>((_) {}),
selfDeviceId: deviceId,
);
getIt getIt
..registerSingleton<SocialService>(social) ..registerSingleton<SocialService>(social)
..registerSingleton<SocialConnection>(connection) ..registerSingleton<SocialConnection>(connection)
..registerSingleton<InboxService>(inbox); ..registerSingleton<InboxService>(inbox)
// Subscribe the listener BEFORE the connection starts connecting. ..registerSingleton<SyncService>(sync);
// Subscribe the listeners BEFORE the connection starts connecting.
inbox.start(); inbox.start();
sync.start();
connection.start(); connection.start();
} }

View file

@ -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<String>? _cached;
/// This install's device id, generated and persisted on first use.
Future<String> deviceId() => _cached ??= _readOrCreate();
Future<String> _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();
}

View file

@ -9,6 +9,7 @@ import '../data/export_import/inventory_snapshot.dart';
import '../data/variety_repository.dart'; import '../data/variety_repository.dart';
import '../security/secure_key_store.dart'; import '../security/secure_key_store.dart';
import 'file_service.dart'; import 'file_service.dart';
import 'inventory_snapshot_io.dart';
/// A picked backup file awaiting restore kept in memory so asking the user /// A picked backup file awaiting restore kept in memory so asking the user
/// for a recovery code never forces a second file dialog. /// 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; /// 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 /// backup-and-recovery.md): a copy that lands on someone's cloud folder
/// reveals nothing. The printed recovery code opens them on any device. /// reveals nothing. The printed recovery code opens them on any device.
class ExportImportService { class ExportImportService implements InventorySnapshotIO {
ExportImportService({ ExportImportService({
required VarietyRepository repository, required VarietyRepository repository,
required FileService files, required FileService files,
@ -49,6 +50,18 @@ class ExportImportService {
/// The backup file extension. The content is the sealed interchange JSON. /// The backup file extension. The content is the sealed interchange JSON.
static const backupExtension = 'tanemaki'; 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<List<int>> buildSnapshot() async =>
utf8.encode(_jsonCodec.encode(await _repository.exportInventory()));
/// Merges a raw interchange snapshot from another device (LWW, idempotent).
@override
Future<void> applySnapshot(List<int> bytes) async {
await _repository.importInventory(_jsonCodec.decode(utf8.decode(bytes)));
}
/// Builds the full inventory as a sealed (encrypted) backup the same 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 /// [exportBackup] writes to a file. Shared with the automatic backup so both
/// paths seal identically. /// paths seal identically.

View file

@ -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<List<int>> 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<void> applySnapshot(List<int> bytes);
}

View file

@ -9,14 +9,23 @@ import 'package:commons_core/commons_core.dart';
/// social sessions on demand. Local-first: constructing this is cheap and /// social sessions on demand. Local-first: constructing this is cheap and
/// offline; nothing connects to a relay until [openSession] is called, so the /// offline; nothing connects to a relay until [openSession] is called, so the
/// app runs fully without network and the social layer only enriches. /// 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 { class SocialService {
SocialService({ SocialService({
required this.identity, required this.identity,
this.account = 0, this.account = 0,
this.rootSeedHex = '', this.rootSeedHex = '',
this.deviceId = '',
List<String> relays = const [], List<String> relays = const [],
}) : relays = List.unmodifiable(relays); }) : 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 /// The derived Nostr identity (secp256k1). Same key across reinstalls that
/// restore the same seed (for a given [account]). /// restore the same seed (for a given [account]).
final NostrIdentity identity; final NostrIdentity identity;
@ -43,6 +52,7 @@ class SocialService {
static Future<SocialService> fromRootSeedHex( static Future<SocialService> fromRootSeedHex(
String rootSeedHex, { String rootSeedHex, {
int account = 0, int account = 0,
String deviceId = '',
List<String> relays = const [], List<String> relays = const [],
}) async { }) async {
final identity = await NostrKeyDerivation.deriveFromSeed( final identity = await NostrKeyDerivation.deriveFromSeed(
@ -53,6 +63,7 @@ class SocialService {
identity: identity, identity: identity,
account: account, account: account,
rootSeedHex: rootSeedHex, rootSeedHex: rootSeedHex,
deviceId: deviceId,
relays: relays, relays: relays,
); );
} }
@ -74,18 +85,23 @@ class SocialService {
/// skipped; throws only when none are reachable. Caller closes it. /// skipped; throws only when none are reachable. Caller closes it.
Future<SocialSession> openSession(List<String> relayUrls) async { Future<SocialSession> openSession(List<String> relayUrls) async {
final pool = await RelayPool.connect(relayUrls, identity: identity); 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 relay channel with the three transports on top the
/// "one channel, three interfaces" shape, at the app layer. /// "one channel, three interfaces" shape, at the app layer.
class SocialSession { class SocialSession {
SocialSession(this._channel) SocialSession(this._channel, {String deviceId = ''})
: offers = NostrOfferTransport(_channel), : offers = NostrOfferTransport(_channel),
messages = NostrMessageTransport(_channel), messages = NostrMessageTransport(_channel),
trust = NostrTrustTransport(_channel), trust = NostrTrustTransport(_channel),
profile = NostrProfileTransport(_channel); profile = NostrProfileTransport(_channel),
sync = NostrSyncTransport(
_channel,
namespace: kInventorySyncNamespace,
deviceId: deviceId,
);
final NostrChannel _channel; final NostrChannel _channel;
@ -94,6 +110,9 @@ class SocialSession {
final TrustTransport trust; final TrustTransport trust;
final ProfileTransport profile; final ProfileTransport profile;
/// Device-to-device inventory sync for THIS identity's own devices.
final SyncTransport sync;
Future<void> close() => _channel.close(); Future<void> close() => _channel.close();
} }

View file

@ -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<void> 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<void> _localChanges;
final String _selfDeviceId;
final Duration _debounce;
StreamSubscription<SocialSession?>? _sessionsSub;
StreamSubscription<SyncSnapshot>? _snapSub;
StreamSubscription<void>? _changesSub;
Timer? _debounceTimer;
SocialSession? _session;
List<int>? _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<void> 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<void> 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<int> a, List<int>? 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<void> stop() async {
_started = false;
_debounceTimer?.cancel();
await _changesSub?.cancel();
_changesSub = null;
await _sessionsSub?.cancel();
_sessionsSub = null;
await _snapSub?.cancel();
_snapSub = null;
_session = null;
}
}

View file

@ -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));
});
}

View file

@ -141,6 +141,27 @@ void main() {
expect(summary.inserted, 1); 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 { test('legacy plain exports still restore (no seal, direct parse)', () async {
final repoA = newTestRepository(dbA); final repoA = newTestRepository(dbA);
await repoA.addQuickVariety(label: 'Old export'); await repoA.addQuickVariety(label: 'Old export');

View file

@ -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<int>>[];
List<int> snapshot = const [1, 2, 3];
@override
Future<List<int>> buildSnapshot() async => snapshot;
@override
Future<void> applySnapshot(List<int> bytes) async => applied.add(bytes);
}
void main() {
const seedHex =
'000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f';
Future<SyncService> 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<int> 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();
});
}