flushQueue method

Future<FlushResult> flushQueue()

Flushes the offline queue to the server.

Returns a FlushResult with counts of synced, denied, and failed ops.

  • synced: successfully sent to server.
  • denied: rejected by security rules — op is removed from queue and local cache is cleaned for that key.
  • failed: network error — op stays in queue for next attempt.

Implementation

Future<FlushResult> flushQueue() async {
  // Wait for the queue to finish loading from disk
  await ready;

  if (_flushing || _pendingQueue.isEmpty) return const FlushResult();
  if (!_wsChannel.isConnected) return const FlushResult();

  _flushing = true;
  var synced = 0;
  var denied = 0;

  try {
    final remaining = <_PendingOp>[];

    for (final op in _pendingQueue) {
      // Skip entries marked as local on disk
      if (_disk.isLocal(op.blockId)) {
        synced++;
        continue;
      }
      try {
        if (op.type == 'put' && op.data != null) {
          if (op.indexEntries != null && op.indexEntries!.isNotEmpty) {
            await _wsChannel.putRangeTagged(op.blockId, op.data!, [], op.indexEntries!);
          } else {
            await _wsChannel.put(op.blockId, op.data!);
          }
        } else if (op.type == 'delete') {
          await _wsChannel.delete(op.blockId);
        }
        synced++;
      } catch (e) {
        if (e is WsDeniedException) {
          // Rules denied this op — clean local cache and discard.
          denied++;
          _sieveRemove(op.blockId);
          await _disk.evict(op.blockId);
          _deniedController.add(op.blockId);
          continue;
        }
        // Network error — keep this and all remaining ops for retry.
        remaining.add(op);
        // Stop trying — next ops likely fail too.
        break;
      }
    }

    // Keep only ops that weren't processed yet (after the failed one)
    if (remaining.isNotEmpty) {
      final failedIdx = _pendingQueue.indexOf(remaining.first);
      if (failedIdx >= 0) {
        _pendingQueue = _pendingQueue.sublist(failedIdx);
      }
    } else {
      _pendingQueue.clear();
    }
  } finally {
    _flushing = false;
    await _persistQueue();
  }

  final result = FlushResult(
    synced: synced,
    denied: denied,
    failed: _pendingQueue.length,
  );
  _flushResultController.add(result);
  return result;
}