IKAR/lib/backend/modules/outbox.dart

import 'dart:async';
import 'package:shared_preferences/shared_preferences.dart';

import '../../core/crypto/message_decryption_cache.dart';
import '../../core/storage/app_database.dart';
import '../../core/storage/token_storage.dart';
import '../../core/utils/logger.dart';
import '../api.dart';
import 'chats.dart';
import 'messages.dart';
import 'outbox_queue.dart';
import 'outgoing_text.dart';

class OutboxService {
  OutboxService._();

  static final OutboxService instance = OutboxService._();
  OutboxQueue? _queue;
  StreamSubscription<SessionState>? _subscription;
  final Map<String, CachedMessage> _latest = {};

  void init(
    Api api,
    MessagesModule messages, {
    required bool Function() isAuthenticated,
  }) {
    if (_queue != null) {
      unawaited(flush());
      return;
    }
    _queue = OutboxQueue(
      activeAccount: TokenStorage.getActiveAccountId,
      isOnline: () => api.state == SessionState.online && isAuthenticated(),
      sessionEpoch: () => api.sessionEpoch,
      load: (accountId) async => (await AppDatabase.loadPendingMessages(
        accountId,
      )).map(CachedMessage.fromDbRow).toList(),
      save: (message) => AppDatabase.saveMessages([message.toDbRow()]),
      updatePending: (message) => AppDatabase.replacePendingMessage(
        message.toDbRow(),
        pendingId: message.id,
      ),
      replace: (before, after) async {
        if (!await AppDatabase.replacePendingMessage(
          after.toDbRow(),
          pendingId: before.id,
        )) {
          throw const OutboxCancelled();
        }
      },
      send: (message, cid) async {
        final epoch = api.sessionEpoch;
        final preferences = await SharedPreferences.getInstance();
        final stored = await AppDatabase.loadMessage(
          message.accountId,
          message.chatId,
          message.id,
        );
        if (stored == null ||
            stored['deleted'] != 0 ||
            stored['status'] != 'pending') {
          throw const OutboxCancelled();
        }
        void checkSession() {
          if (api.sessionEpoch != epoch ||
              !isAuthenticated() ||
              preferences.getString('active_account_id') !=
                  message.accountId.toString()) {
            throw StateError('Outbox account changed before dispatch');
          }
        }

        checkSession();
        return messages.sendMessage(
          message.accountId,
          message.chatId,
          message.text!,
          clientId: cid,
          replyToMessageId: OutgoingText.replyId(message),
          replySourceChatId: OutgoingText.replyChatId(message),
          elements: OutgoingText.elements(message),
          beforeSend: checkSession,
        );
      },
      onChanged: (before, after) {
        _latest[before.id] = after;
        if (_latest.length > 256) _latest.remove(_latest.keys.first);
        MessageDecryptionCache.instance.adopt(before.id, after.id);
        chats.emitMessageSent(after.chatId, before.id, after);
        unawaited(_updatePreview(after));
      },
      onError: (error, stack) => logger.w('Outbox: $error'),
    );
    _subscription = api.stateStream.listen((state) {
      if (state == SessionState.online) unawaited(flush());
    });
    unawaited(flush());
  }

  Future<CachedMessage> enqueue(
    CachedMessage message, {
    bool createDialog = false,
    String? dialogTitle,
    String? dialogIconUrl,
  }) async {
    final queue = _queue;
    if (queue == null) throw StateError('Outbox is not initialized');
    await queue.enqueue(
      message,
      persist: createDialog
          ? (message) => AppDatabase.saveOutgoingMessage(
              message.toDbRow(),
              createDialog: true,
              dialogTitle: dialogTitle,
              dialogIconUrl: dialogIconUrl,
            )
          : null,
    );
    await _updatePreview(message);
    return _latest.remove(message.id) ?? message;
  }

  Future<void> _updatePreview(CachedMessage message) async {
    try {
      await chats.applyOutgoing(
        message.accountId,
        message.chatId,
        messageId: message.id,
        time: message.time,
        text: message.text ?? '',
        status: message.status!,
        elements: OutgoingText.elements(message),
      );
    } catch (error) {
      logger.w('Outbox preview: $error');
    }
  }

  Future<void> flush() => _queue?.flush() ?? Future.value();

  String resolveMessageId(String id) => _latest[id]?.id ?? id;

  Future<void> dispose() async {
    _queue?.dispose();
    _queue = null;
    await _subscription?.cancel();
    _subscription = null;
  }
}