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