processQueue method

Implementation

Future<int> processQueue([FutureOr<dynamic> Function(QueuedMutation)? customHandler]) async {
  if (_isProcessing || _queue.isEmpty) return 0;
  _isProcessing = true;

  int syncedCount = 0;
  final toRemove = <QueuedMutation>[];

  logger.info('OfflineMutationQueue: Replaying ${_queue.length} pending mutations...');

  for (final mutation in _queue) {
    final executor = customHandler != null
        ? ((_) => customHandler(mutation))
        : _executors[mutation.mutationType];

    if (executor == null) {
      logger.warn('OfflineMutationQueue: No executor registered for [${mutation.mutationType}]. Skipping.');
      continue;
    }

    try {
      await executor(mutation.payload);
      toRemove.add(mutation);
      syncedCount++;
      logger.debug('OfflineMutationQueue: Mutation [${mutation.id}] successfully replayed.');
    } catch (err) {
      logger.warn('OfflineMutationQueue: Mutation [${mutation.id}] failed during replay: $err');
      mutation.retryCount++;

      // Apply Conflict Policy
      switch (mutation.conflictPolicy) {
        case ConflictPolicy.clientWins:
          break;

        case ConflictPolicy.serverWins:
          logger.info('OfflineMutationQueue: ConflictPolicy.serverWins -> Discarding client mutation [${mutation.id}]');
          toRemove.add(mutation);
          BloomData.invalidateQueries([mutation.mutationType]);
          break;

        case ConflictPolicy.custom:
          final resolver = _resolvers[mutation.mutationType];
          if (resolver != null) {
            final resolvedPayload = await resolver(mutation.payload, err);
            if (resolvedPayload != null) {
              mutation.payload = resolvedPayload;
              try {
                await executor(mutation.payload);
                toRemove.add(mutation);
                syncedCount++;
              } catch (_) {}
            } else {
              toRemove.add(mutation);
            }
          }
          break;
      }
    }
  }

  _queue.removeWhere((item) => toRemove.contains(item));
  await persist();

  _isProcessing = false;
  logger.info('OfflineMutationQueue: Replay finished. Successfully synced: $syncedCount, Remaining: ${_queue.length}');
  return syncedCount;
}