IKAR/lib/backend/modules/outbox_queue.dart

import 'dart:async';

import '../../core/protocol/packet.dart';
import 'messages.dart';
import 'outgoing_text.dart';

/// The user removed or changed an item after a flush loaded its snapshot.
class OutboxCancelled implements Exception {
  const OutboxCancelled();
}

class OutboxQueue {
  OutboxQueue({
    required this.activeAccount,
    required this.isOnline,
    required this.sessionEpoch,
    required this.load,
    required this.save,
    required this.updatePending,
    required this.replace,
    required this.send,
    required this.onChanged,
    required this.onError,
    DateTime Function()? now,
    this.retryBase = const Duration(seconds: 2),
  }) : now = now ?? DateTime.now;

  final Future<int?> Function() activeAccount;
  final bool Function() isOnline;
  final int Function() sessionEpoch;
  final Future<List<CachedMessage>> Function(int) load;
  final Future<void> Function(CachedMessage) save;
  final Future<bool> Function(CachedMessage) updatePending;
  final Future<void> Function(CachedMessage, CachedMessage) replace;
  final Future<String> Function(CachedMessage, int) send;
  final void Function(CachedMessage, CachedMessage) onChanged;
  final void Function(Object, StackTrace) onError;
  final DateTime Function() now;
  final Duration retryBase;
  final Map<String, CachedMessage> _receipts = {};
  Timer? _timer;
  Future<void>? _running;
  bool _wakeAgain = false;
  bool _disposed = false;

  Future<void> enqueue(
    CachedMessage message, {
    Future<void> Function(CachedMessage)? persist,
  }) async {
    if (OutgoingText.metadata(message) == null) {
      throw ArgumentError('Only prepared outgoing text may be queued');
    }
    await (persist ?? save)(message);
  }

  Future<void> flush() {
    if (_disposed) return Future.value();
    final running = _running;
    if (running != null) {
      _wakeAgain = true;
      return running;
    }
    _timer?.cancel();
    return _running = _flush().whenComplete(() {
      _running = null;
      if (_wakeAgain && !_disposed) {
        _wakeAgain = false;
        unawaited(flush());
      }
    });
  }

  Future<bool> _validSession(int accountId, int epoch) async {
    final active = await activeAccount();
    return !_disposed &&
        active == accountId &&
        isOnline() &&
        sessionEpoch() == epoch;
  }

  Future<void> _flush() async {
    if (!isOnline()) return;
    final epoch = sessionEpoch();
    try {
      final accountId = await activeAccount();
      if (accountId == null || !await _validSession(accountId, epoch)) return;
      final rows = await load(accountId);
      final blockedChats = <int>{};
      for (final pending in rows) {
        if (!await _validSession(accountId, epoch)) break;
        if (pending.accountId != accountId ||
            pending.deleted ||
            blockedChats.contains(pending.chatId)) {
          continue;
        }
        final meta = OutgoingText.metadata(pending);
        if (meta == null || pending.text == null || pending.text!.isEmpty) {
          final failed = pending.copyWith(status: 'error');
          if (!await updatePending(failed)) continue;
          if (await _validSession(accountId, epoch)) _notify(pending, failed);
          continue;
        }
        final retryAt = meta['retryAt'] as int? ?? 0;
        final remaining = retryAt - now().millisecondsSinceEpoch;
        if (remaining > 0) {
          blockedChats.add(pending.chatId);
          _schedule(Duration(milliseconds: remaining));
          continue;
        }
        final receiptKey =
            '${pending.accountId}/${pending.chatId}/${pending.id}';
        try {
          var sent = _receipts[receiptKey];
          if (sent == null) {
            final id = await send(pending, meta['cid'] as int);
            if (id.isEmpty) throw StateError('Missing server message ID');
            sent = CachedMessage(
              id: id,
              accountId: pending.accountId,
              chatId: pending.chatId,
              senderId: pending.senderId,
              text: pending.text,
              time: pending.time,
              status: 'sent',
              payload: pending.payload,
              sealedText: pending.sealedText,
              e2ee: pending.e2ee,
            );
            _receipts[receiptKey] = sent;
          }
          await replace(pending, sent);
          _receipts.remove(receiptKey);
          if (await _validSession(accountId, epoch)) _notify(pending, sent);
        } catch (error, stack) {
          if (error is OutboxCancelled) {
            _receipts.remove(receiptKey);
            continue;
          }
          onError(error, stack);
          if (!await _validSession(accountId, epoch)) return;
          if (isPermanentSendFailure(error)) {
            final failed = pending.withSendFailure(error);
            if (!await updatePending(failed)) continue;
            if (await _validSession(accountId, epoch)) _notify(pending, failed);
            continue;
          }
          final attempts = ((meta['attempts'] as int? ?? 0) + 1).clamp(1, 30);
          final delay = Duration(
            milliseconds:
                (retryBase.inMilliseconds * (1 << (attempts - 1).clamp(0, 5)))
                    .clamp(1, 60000),
          );
          final queued = pending.copyWith(
            status: 'pending',
            payload: {
              ...?pending.payload,
              '_outbox': {
                ...meta,
                'attempts': attempts,
                'retryAt': now().add(delay).millisecondsSinceEpoch,
              },
            },
          );
          if (!await updatePending(queued)) continue;
          if (await _validSession(accountId, epoch)) _notify(pending, queued);
          blockedChats.add(pending.chatId);
          _schedule(delay);
        }
      }
    } catch (error, stack) {
      onError(error, stack);
      _schedule(retryBase);
    }
  }

  void _notify(CachedMessage before, CachedMessage after) {
    try {
      onChanged(before, after);
    } catch (error, stack) {
      onError(error, stack);
    }
  }

  DateTime? _scheduledAt;

  void _schedule(Duration delay) {
    if (_disposed) return;
    final at = now().add(delay);
    if (_timer?.isActive == true && _scheduledAt!.isBefore(at)) return;
    _timer?.cancel();
    _scheduledAt = at;
    _timer = Timer(delay, () => unawaited(flush()));
  }

  void dispose() {
    _disposed = true;
    _timer?.cancel();
  }
}