synchronize method

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

Implementation

Future<DatumSyncResult<T>> synchronize(
  String userId, {
  DatumSyncOptions<DatumEntityInterface>? options,
  DatumSyncScope? scope,
}) async {
  _ensureInitialized();

  if (config.excludedSyncUserIds.contains(userId)) {
    _logger.info('Sync for user $userId skipped: user is excluded from sync.');
    return DatumSyncResult.skipped(
      userId,
      await getPendingCount(userId),
      reason: 'User is excluded from sync',
    );
  }

  _logger.info('🔄 [${T.toString()}] Starting sync for user: $userId');

  return _syncRequestStrategy.execute(
    () async {
      if (_isSyncPaused) {
        _logger.info('Sync for user $userId skipped: manager is paused.');
        return DatumSyncResult.skipped(
          userId,
          await getPendingCount(userId),
          reason: 'Sync is paused',
        );
      }

      // Handle user switching logic before proceeding with synchronization.
      if (_syncEngineInstance.lastActiveUserId != null && _syncEngineInstance.lastActiveUserId != userId) {
        if (config.defaultUserSwitchStrategy == UserSwitchStrategy.promptIfUnsyncedData) {
          final oldUserOps = await _queueManager.getPending(
            _syncEngineInstance.lastActiveUserId ?? '',
          );
          if (oldUserOps.isNotEmpty) {
            throw UserSwitchException(
              oldUserId: _syncEngineInstance.lastActiveUserId,
              newUserId: userId,
              message: 'Cannot switch user while unsynced data exists for the previous user.',
            );
          }
        }
        // Other strategies like syncThenSwitch or clearAndFetch would be handled here.
      }

      try {
        // Merge provided options with defaults from config
        final mergedOptions = _mergeSyncOptions(options);

        // Convert options to the correct type if needed.
        // This handles cases where options might be passed with a different generic type from Datum.
        var typedOptions = mergedOptions != null
            ? DatumSyncOptions<T>(
                includeDeletes: mergedOptions.includeDeletes,
                resolveConflicts: mergedOptions.resolveConflicts,
                forceFullSync: mergedOptions.forceFullSync,
                overrideBatchSize: mergedOptions.overrideBatchSize,
                timeout: mergedOptions.timeout,
                direction: mergedOptions.direction,
                conflictResolver: mergedOptions.conflictResolver is DatumConflictResolver<T> ? mergedOptions.conflictResolver as DatumConflictResolver<T> : null,
                query: mergedOptions.query,
              )
            : null;

        // Allow custom sync direction resolution via callback
        final pendingCount = await getPendingCount(userId);
        final currentDirection = typedOptions?.direction ?? config.defaultSyncDirection;

        if (config.syncDirectionResolver != null) {
          final resolvedDirection = config.syncDirectionResolver!(pendingCount, currentDirection);
          if (resolvedDirection != null && resolvedDirection != currentDirection) {
            _logger.debug('Custom sync direction resolver changed direction from $currentDirection to $resolvedDirection for user $userId');
            final optimizedOptions = typedOptions?.copyWith(direction: resolvedDirection) ?? DatumSyncOptions<T>(direction: resolvedDirection);
            typedOptions = optimizedOptions;
          }
        }

        // If no scope is provided but options contain a query, create a scope from the query
        DatumSyncScope? effectiveScope = scope;
        if (effectiveScope == null && typedOptions?.query != null && typedOptions!.query != const DatumQuery()) {
          effectiveScope = DatumSyncScope(query: typedOptions.query);
        }

        // Check if sync should be skipped based on final direction and pending operations
        // Check if sync should be skipped based on final direction and pending operations
        final finalDirection = typedOptions?.direction ?? config.defaultSyncDirection;
        if (finalDirection == SyncDirection.pushOnly && pendingCount == 0) {
          // If the direction is pushOnly and there are no pending operations,
          // we can skip the sync entirely for this manager.
          _logger.info('Push-only sync for user $userId skipped: no pending operations.');
          return DatumSyncResult.skipped(userId, 0);
        }

        DatumSyncResult<T> result;
        List<DatumSyncEvent<T>> events;

        if (config.useIsolateSync) {
          // Capture dependencies into local variables to avoid capturing 'this' in Isolate.run
          final localAdapterCaptured = localAdapter;
          final remoteAdapterCaptured = remoteAdapter;
          final conflictResolverCaptured = _conflictResolver;
          final queueManagerCaptured = _queueManager;
          final conflictDetectorCaptured = _conflictDetector;
          final loggerCaptured = _logger.getWorkerLogger();
          // Sanitize config to remove unsendable callbacks. Note: copyWith
          // cannot clear a field (null means "unchanged"), so this must use
          // the dedicated sanitizer.
          final configCaptured = config.sanitizedForIsolate<T>();
          final connectivityCaptured = _connectivity;
          final isolateHelperCaptured = _isolateHelper;
          final deviceIdCaptured = deviceId;
          final optionsCaptured = typedOptions;
          final scopeCaptured = effectiveScope;

          // Offload the entire sync process to a background isolate.
          // note: This requires Adapters and other dependencies to be sendable.
          // Spawned via the top-level trampoline — see [_spawnSyncIsolate]
          // for why a closure built here would capture `this`.
          (result, events) = await _spawnSyncIsolate<T>(
            userId,
            localAdapterCaptured,
            remoteAdapterCaptured,
            conflictResolverCaptured,
            queueManagerCaptured,
            conflictDetectorCaptured,
            loggerCaptured,
            configCaptured,
            connectivityCaptured,
            isolateHelperCaptured,
            deviceIdCaptured,
            optionsCaptured,
            scopeCaptured,
          );
        } else {
          // Enforce the configured sync timeout (config.syncTimeout /
          // options.timeout) — previously it was configured everywhere but
          // never applied, so a hung remote call blocked syncs forever (the
          // "already syncing" guard then rejected all future syncs too).
          final effectiveTimeout = typedOptions?.timeout ?? config.syncTimeout;
          final engineFuture = Future.sync(() => _syncEngineInstance.synchronize(
                userId,
                options: typedOptions,
                scope: effectiveScope,
              ));
          var timedOut = false;
          (result, events) = await engineFuture.timeout(
            effectiveTimeout,
            onTimeout: () {
              timedOut = true;
              return (DatumSyncResult<T>.cancelled(userId, 0), <DatumSyncEvent<T>>[]);
            },
          );
          if (timedOut) {
            // Silence the still-running engine future (its loops halt once
            // the status leaves `syncing`) and surface a typed timeout.
            unawaited(engineFuture.then((_) {}, onError: (_, __) {}));
            if (!_statusSubject.isClosed) {
              _statusSubject.add(currentStatus.copyWith(
                status: DatumSyncStatus.failed,
                health: const DatumHealth(status: DatumSyncHealth.error),
              ));
            }
            throw DatumException(
              code: DatumExceptionCode.timeout,
              message: 'Synchronization for user $userId timed out after ${effectiveTimeout.inMilliseconds}ms.',
            );
          }
        }

        _processSyncEvents(events);
        // Persist the result of the sync operation.
        if (!result.wasSkipped) {
          await localAdapter.saveLastSyncResult(userId, result);
          // Also update sync metadata in persistence
          final metadata = await localAdapter.getSyncMetadata(userId);
          if (metadata != null) {
            await persistence?.saveSyncMetadata(userId, metadata);
          }
        }
        return result;
      } on Object catch (e, stack) {
        // Delegate to the shared handler: it processes the events carried
        // inside a SyncExceptionWithEvents and rethrows the original error
        // WITH its original stack trace (the previous inline duplicate of
        // this logic dropped the trace via a bare `throw e.originalError`).
        SyncErrorHandler.handleManagerSyncErrorSync<T>(e, stack, const [], _processSyncEvents);
      }
    },
    isSyncInProgress: () => _syncEngineInstance.isSyncing,
    onSkipped: () {
      _logger.info('Sync for user $userId skipped: another sync is in progress.');
      return DatumSyncResult.skipped(
        userId,
        0, // Can't reliably get pending count here without async, so default to 0.
        reason: 'Sync in progress',
      );
    },
  );
}