synchronize method

Future<DatumSyncResult<DatumEntityInterface>> synchronize(String userId, { DatumSyncOptions<DatumEntityInterface>? options, })

A global sync that can coordinate across all managers.

Implementation

Future<DatumSyncResult<DatumEntityInterface>> synchronize(
  String userId, {
  DatumSyncOptions<DatumEntityInterface>? options,
}) async {
  final snapshot = _getSnapshot(userId);
  if (snapshot.status == DatumSyncStatus.syncing) {
    logger.info('[Global] Sync for user $userId skipped: another global sync is already in progress.');
    return DatumSyncResult.skipped(userId, snapshot.pendingOperations);
  }

  final stopwatch = Stopwatch()..start();
  _updateSnapshot(userId, (s) => s.copyWith(status: DatumSyncStatus.syncing));
  for (final observer in globalObservers) {
    observer.onSyncStart();
  }

  var totalSynced = 0;
  var totalFailed = 0;
  var totalConflicts = 0;
  final allPending = <DatumSyncOperation<DatumEntityInterface>>[];

  try {
    final direction = options?.direction ?? config.defaultSyncDirection;

    // Uniform accumulation for every phase/direction. The previous per-branch
    // loops drifted out of sync (e.g. pushThenPull dropped pull failures and
    // no branch counted conflicts resolved during push).
    void accumulate(Iterable<DatumSyncResult<DatumEntityInterface>> results) {
      for (final res in results) {
        totalSynced += res.syncedCount;
        totalFailed += res.failedCount;
        totalConflicts += res.conflictsResolved;
        allPending.addAll(res.pendingOperations);
      }
    }

    switch (direction) {
      case SyncDirection.pushThenPull:
        accumulate(await _pushChanges(userId, options));
        accumulate(await _pullChanges(userId, options));
      case SyncDirection.pullThenPush:
        accumulate(await _pullChanges(userId, options));
        accumulate(await _pushChanges(userId, options));
      case SyncDirection.pushOnly:
        accumulate(await _pushChanges(userId, options));
      case SyncDirection.pullOnly:
        accumulate(await _pullChanges(userId, options));
    }

    final result = DatumSyncResult<DatumEntityInterface>(
      userId: userId,
      duration: stopwatch.elapsed,
      syncedCount: totalSynced,
      failedCount: totalFailed,
      conflictsResolved: totalConflicts,
      pendingOperations: allPending,
    );

    _updateSnapshot(userId, (s) => s.copyWith(status: DatumSyncStatus.completed, lastCompletedAt: DateTime.now()));
    for (final observer in globalObservers) {
      observer.onSyncEnd(result);
    }

    return result;
  } catch (e, stack) {
    logger.error('Synchronization failed for user $userId', stack);
    _updateSnapshot(userId, (s) => s.copyWith(status: DatumSyncStatus.failed, errors: [e]));
    return Future.error(e, stack);
  }
}