tane/apps/app_seeds/lib/services/sync_service.dart

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