flush method

Future<OfflineFlushReport> flush({ required Future<String?> credential(String), required Future<OfflineAttempt> send(String, String, Map<String, dynamic>), bool cancelled()?, })

Implementation

Future<OfflineFlushReport> flush({
  required Future<String?> Function(String) credential,
  required Future<OfflineAttempt> Function(
    String,
    String,
    Map<String, dynamic>,
  )
  send,
  bool Function()? cancelled,
}) async {
  if (_flushing) throw StateError('flush active');
  _flushing = true;
  var attempted = 0, accepted = 0, stopped = false;
  final outcomes = <OfflineOutcome>[];
  try {
    final (initial, initialBad) = await _read();
    var expiredChanged = false;
    for (final record in initial) {
      if (record.state != 'expired_local' &&
          nowMillis() - record.createdAtMillis >
              limits.maxAgeSeconds * 1000) {
        record.state = 'expired_local';
        record.reason = null;
        expiredChanged = true;
      }
    }
    if (expiredChanged) await _write(initial, initialBad);
    final ids = initial.where((r) => r.state == 'pending').toList()
      ..sort(
        (a, b) => a.createdAtMillis != b.createdAtMillis
            ? a.createdAtMillis.compareTo(b.createdAtMillis)
            : a.id.compareTo(b.id),
      );
    for (final item in ids) {
      if (cancelled?.call() == true) {
        stopped = true;
        break;
      }
      var (good, bad) = await _read();
      final index = good.indexWhere(
        (r) => r.id == item.id && r.state == 'pending',
      );
      if (index < 0) continue;
      final record = good[index];
      final token = await credential(record.collectionId);
      if (token == null) {
        outcomes.add(OfflineOutcome(record.id, 'credential_unavailable'));
        continue;
      }
      attempted++;
      OfflineAttempt attempt;
      try {
        attempt = await send(
          record.collectionId,
          token,
          jsonDecode(utf8.decode(record.submission)),
        );
      } catch (_) {
        attempt = const OfflineAttempt(0);
      }
      (good, bad) = await _read();
      final fresh = good.indexWhere((r) => r.id == record.id);
      if (fresh < 0) continue;
      if (attempt.status >= 200 &&
          attempt.status < 300 &&
          attempt.accepted &&
          attempt.receiptCollectionId == good[fresh].collectionId &&
          (attempt.responseId?.isNotEmpty ?? false)) {
        good.removeAt(fresh);
        accepted++;
        outcomes.add(OfflineOutcome(record.id, 'accepted'));
      } else {
        good[fresh].attemptCount++;
        final retry =
            [408, 425, 429].contains(attempt.status) ||
            attempt.status >= 500 ||
            attempt.status == 0;
        final reason = {
          400: OfflineReason.invalid,
          409: OfflineReason.conflict,
          401: OfflineReason.unauthorized,
          403: OfflineReason.revoked,
          404: OfflineReason.deleted,
          410: OfflineReason.expired,
        }[attempt.status] ??
            (attempt.status >= 400 && attempt.status < 500 && !retry
                ? OfflineReason.invalid
                : null);
        if (reason != null) {
          good[fresh].state = 'blocked';
          good[fresh].reason = reason;
          outcomes.add(OfflineOutcome(record.id, 'blocked', reason: reason));
        } else {
          outcomes.add(
            OfflineOutcome(
              record.id,
              'retry',
              retryAfterSeconds: retry && attempt.retryAfterSeconds != null
                  ? attempt.retryAfterSeconds!.clamp(0, 86400)
                  : null,
            ),
          );
        }
      }
      await _write(good, bad);
    }
  } finally {
    _flushing = false;
  }
  return OfflineFlushReport(
    attempted,
    accepted,
    stopped,
    outcomes,
    await snapshot(),
  );
}