OnBudget/test/features/notification_parsing/application/transfer_pairing_pipeline_test.dart

import 'dart:async';
import 'dart:convert';

import 'package:drift/drift.dart' hide isNull, isNotNull;
import 'package:drift/native.dart';
import 'package:flutter_riverpod/flutter_riverpod.dart';
import 'package:flutter_test/flutter_test.dart';
import 'package:http/http.dart' as http;
import 'package:http/testing.dart';
import 'package:new_budget/src/core/database/app_database.dart';
import 'package:new_budget/src/core/database/converters/enum_converters.dart';
import 'package:new_budget/src/core/providers/database_provider.dart';
import 'package:new_budget/src/features/notification_parsing/application/ai_providers.dart';
import 'package:new_budget/src/features/notification_parsing/application/inbox_controller.dart';
import 'package:new_budget/src/features/notification_parsing/application/notification_parsing_providers.dart';
import 'package:new_budget/src/features/notification_parsing/application/parsing_pipeline.dart';
import 'package:new_budget/src/features/notification_parsing/application/parsing_settings_controller.dart';
import 'package:new_budget/src/features/notification_parsing/application/parsing_worker.dart';
import 'package:new_budget/src/features/notification_parsing/data/deepseek/deepseek_client.dart';
import 'package:new_budget/src/features/notification_parsing/data/parser/ai_parser.dart';
import 'package:new_budget/src/features/notification_parsing/data/parser/draft_codec.dart';
import 'package:new_budget/src/features/notification_parsing/data/parser/transfer_pair_matcher.dart';
import 'package:new_budget/src/features/notification_parsing/domain/entities/raw_message.dart';
import 'package:new_budget/src/features/notification_parsing/domain/enums.dart';
import 'package:new_budget/src/features/notification_parsing/domain/repositories/raw_messages_repository.dart';
import 'package:new_budget/src/features/transactions/application/transaction_providers.dart';

/// Сквозные тесты склейки переводов (transfer pairing): waitingPair → sweep →
/// merged-карточка / доклейка / релиз по таймауту / расклейка. AI замокан:
/// по телу сообщения возвращает transfer_out («Вы перевели …») либо
/// transfer_in («Пополнение …»).

const _userId = 'u1';
const _accA = 'acc-a';
const _accB = 'acc-b';
const _bankA = 'ru.sberbankmobile';
const _bankB = 'com.idamob.tinkoff.android';

const _outBody = 'Вы перевели 5000 ₽ на счёт в другом банке';
const _inBody = 'Пополнение 5000 ₽ переводом из банка';
const _amountMinor = 500000;

AiParser _pairAiParser() {
  final mock = MockClient((req) async {
    // Матчим ПОЛНЫЙ текст уведомления: system-промпт сам содержит слова
    // «перевели»/«Пополнение» (правила выбора kind), по подстроке нельзя.
    final isOut = req.body.contains(_outBody);
    final content = jsonEncode({
      'type': isOut ? 'expense' : 'income',
      'kind': isOut ? 'transfer_out' : 'transfer_in',
      'amount': 5000,
      'currency': 'RUB',
    });
    return http.Response(
      jsonEncode({
        'choices': [
          {
            'message': {'content': content},
          }
        ],
        'usage': {'total_tokens': 42},
      }),
      200,
      headers: {'content-type': 'application/json'},
    );
  });
  return AiParser(DeepSeekClient(client: mock, apiKey: 'k'));
}

Future<void> _seed(AppDatabase db) async {
  await db.usersDao.insertUser(
    UsersTableCompanion.insert(id: _userId, name: 'Test'),
  );
  await db.accountsDao.insertAccount(
    AccountsTableCompanion.insert(id: _accA, userId: _userId, name: 'A'),
  );
  await db.accountsDao.insertAccount(
    AccountsTableCompanion.insert(id: _accB, userId: _userId, name: 'B'),
  );
  await db.sourceAppsDao.insert(SourceAppsTableCompanion.insert(
    id: 'src-a',
    userId: _userId,
    packageName: _bankA,
  ));
  await db.sourceAppsDao.insert(SourceAppsTableCompanion.insert(
    id: 'src-b',
    userId: _userId,
    packageName: _bankB,
  ));
}

void _activateWorker(ProviderContainer c) {
  c.read(parsingWorkerProvider(_userId));
}

/// Ждёт, пока сообщение [id] не пройдёт [test] (по стриму watchAll).
Future<RawMessage> _waitFor(
  RawMessagesRepository repo,
  String id,
  bool Function(RawMessage) test, {
  Duration timeout = const Duration(seconds: 10),
}) async {
  final completer = Completer<RawMessage>();
  late final StreamSubscription<List<RawMessage>> sub;
  sub = repo.watchAll(_userId).listen((list) {
    for (final m in list) {
      if (m.id == id && test(m)) {
        if (!completer.isCompleted) completer.complete(m);
        return;
      }
    }
  });
  try {
    return await completer.future.timeout(
      timeout,
      onTimeout: () async {
        final m = await repo.findById(id);
        throw TimeoutException(
            'message $id stuck: status=${m?.status}, paired=${m?.pairedWithId}');
      },
    );
  } finally {
    await sub.cancel();
  }
}

void main() {
  late AppDatabase db;
  late ProviderContainer container;
  late RawMessagesRepository repo;

  Future<void> bootstrap() async {
    db = AppDatabase.forTesting(NativeDatabase.memory());
    await _seed(db);
    container = ProviderContainer(overrides: [
      appDatabaseProvider.overrideWithValue(db),
      isOnlineProvider.overrideWith((ref) => Stream<bool>.value(true)),
      aiParserProvider.overrideWith((ref) async => _pairAiParser()),
    ]);
    repo = container.read(rawMessagesRepositoryProvider);
    final settings = container.read(parsingSettingsControllerProvider.notifier);
    await container.read(parsingSettingsControllerProvider.future);
    await settings.setAiConsent(true);
    // Дефолтные счета приложений: bankA → accA, bankB → accB (resolver #2,
    // trusted).
    final sourceApps = container.read(sourceAppsRepositoryProvider);
    await sourceApps.setDefaultAccount('src-a', _accA);
    await sourceApps.setDefaultAccount('src-b', _accB);
  }

  tearDown(() async {
    container.dispose();
    await db.close();
  });

  Future<void> forceDeadlinePast(String id) => (db.update(db.rawMessagesTable)
        ..where((t) => t.id.equals(id)))
      .write(RawMessagesTableCompanion(
        pairDeadline: Value(DateTime.now().subtract(const Duration(minutes: 1))),
      ));

  /// out (bankA) + in (bankB) в окне → склейка воркером.
  Future<(RawMessage out, RawMessage inn)> mergePairFlow() async {
    final t0 = DateTime(2026, 7, 1, 12);
    final out = await repo.insertIncoming(
      userId: _userId,
      packageName: _bankA,
      body: _outBody,
      receivedAt: t0,
    );
    final inn = await repo.insertIncoming(
      userId: _userId,
      packageName: _bankB,
      body: _inBody,
      receivedAt: t0.add(const Duration(minutes: 1)),
    );
    final mergedOut = await _waitFor(
      repo,
      out.id,
      (m) => m.status == RawMessageStatus.inbox && m.pairedWithId == inn.id,
    );
    final pairedIn = await _waitFor(
      repo,
      inn.id,
      (m) => m.status == RawMessageStatus.paired,
    );
    return (mergedOut, pairedIn);
  }

  test('склейка: out+in в окне → primary inbox (merged draft), secondary paired',
      () async {
    await bootstrap();
    _activateWorker(container);

    final (out, inn) = await mergePairFlow();

    final bundle = decodeDraftBundle(out.draftJson)!;
    expect(bundle.pairedRawMessageId, inn.id);
    expect(bundle.draft.type, TransactionType.transfer);
    expect(bundle.draft.accountId, _accA);
    expect(bundle.draft.transferToAccountId, _accB);
    expect(bundle.draft.categoryId, isNull);
    expect(bundle.preMerge?['type'], 'expense',
        reason: 'снапшот исходного типа для расклейки');

    expect(inn.pairedWithId, out.id);
    final innBundle = decodeDraftBundle(inn.draftJson)!;
    expect(innBundle.draft.kind, TxKind.transferIn,
        reason: 'одиночный draft secondary не перезаписывается');
    expect(innBundle.pairedRawMessageId, isNull);

    // Транзакций нет — склеенная пара всегда ждёт подтверждения.
    expect(await db.select(db.transactionsTable).get(), isEmpty);
  });

  test('confirmPair → одна transfer-транзакция, обе половинки applied',
      () async {
    await bootstrap();
    _activateWorker(container);

    final (out, _) = await mergePairFlow();
    final bundle = decodeDraftBundle(out.draftJson)!;

    await container.read(inboxControllerProvider.notifier).confirmPair(
          userId: _userId,
          message: out,
          draft: bundle.draft,
          secondaryId: bundle.pairedRawMessageId!,
          fromAccountId: _accA,
          toAccountId: _accB,
        );

    final txns = await db.select(db.transactionsTable).get();
    expect(txns, hasLength(1));
    expect(txns.single.type, TransactionType.transfer);
    expect(txns.single.accountId, _accA);
    expect(txns.single.transferToAccountId, _accB);
    expect(txns.single.amount, _amountMinor);
    expect(txns.single.rawMessageId, out.id);

    final outAfter = await repo.findById(out.id);
    final innAfter = await repo.findById(bundle.pairedRawMessageId!);
    expect(outAfter!.status, RawMessageStatus.applied);
    expect(innAfter!.status, RawMessageStatus.applied);
    expect(outAfter.transactionId, txns.single.id);
    expect(innAfter.transactionId, txns.single.id);
  });

  test('таймаут: релиз в Inbox одиночным черновиком, draft не затёрт',
      () async {
    await bootstrap();
    _activateWorker(container);

    final out = await repo.insertIncoming(
      userId: _userId,
      packageName: _bankA,
      body: _outBody,
      receivedAt: DateTime(2026, 7, 1, 12),
    );
    final waiting = await _waitFor(
        repo, out.id, (m) => m.status == RawMessageStatus.waitingPair);
    expect(waiting.pairDeadline, isNotNull);

    await forceDeadlinePast(out.id);
    await container.read(parsingPipelineProvider).sweepPairs(_userId);

    final released = await _waitFor(
        repo, out.id, (m) => m.status == RawMessageStatus.inbox);
    final bundle = decodeDraftBundle(released.draftJson)!;
    expect(bundle.draft.kind, TxKind.transferOut);
    expect(bundle.draft.amount, _amountMinor);
    expect(bundle.pairedRawMessageId, isNull);
    expect(released.confidenceAmount, isNotNull,
        reason: 'релиз не должен затирать confidence-оценки');
  });

  test('inbox-контрпартнёр: релизнутая половинка склеивается со второй',
      () async {
    await bootstrap();
    _activateWorker(container);

    final t0 = DateTime(2026, 7, 1, 12);
    final out = await repo.insertIncoming(
      userId: _userId,
      packageName: _bankA,
      body: _outBody,
      receivedAt: t0,
    );
    await _waitFor(repo, out.id, (m) => m.status == RawMessageStatus.waitingPair);
    await forceDeadlinePast(out.id);
    await container.read(parsingPipelineProvider).sweepPairs(_userId);
    await _waitFor(repo, out.id, (m) => m.status == RawMessageStatus.inbox);

    // Вторая половинка приходит в окне receivedAt первой (которая уже в Inbox).
    final inn = await repo.insertIncoming(
      userId: _userId,
      packageName: _bankB,
      body: _inBody,
      receivedAt: t0.add(const Duration(minutes: 4)),
    );

    final mergedOut = await _waitFor(
      repo,
      out.id,
      (m) => m.status == RawMessageStatus.inbox && m.pairedWithId == inn.id,
    );
    final pairedIn = await _waitFor(
        repo, inn.id, (m) => m.status == RawMessageStatus.paired);
    expect(pairedIn.pairedWithId, out.id);
    final bundle = decodeDraftBundle(mergedOut.draftJson)!;
    expect(bundle.draft.type, TransactionType.transfer);
    expect(bundle.draft.transferToAccountId, _accB);
  });

  test('доклейка: applied-контрпартнёр конвертируется в transfer + откат',
      () async {
    await bootstrap();
    final settings = container.read(parsingSettingsControllerProvider.notifier);
    // Первая половинка проходит обычным путём (пейринг выключен) и
    // подтверждается пользователем как обычный расход.
    await settings.setTransferPairingEnabled(false);
    _activateWorker(container);

    final t0 = DateTime(2026, 7, 1, 12);
    final out = await repo.insertIncoming(
      userId: _userId,
      packageName: _bankA,
      body: _outBody,
      receivedAt: t0,
    );
    final outInbox = await _waitFor(
        repo, out.id, (m) => m.status == RawMessageStatus.inbox);
    final outBundle = decodeDraftBundle(outInbox.draftJson)!;
    await container.read(inboxControllerProvider.notifier).confirmOnce(
          userId: _userId,
          message: outInbox,
          draft: outBundle.draft,
          accountId: _accA,
        );
    await _waitFor(repo, out.id,
        (m) => m.status == RawMessageStatus.applied && m.transactionId != null);

    await settings.setTransferPairingEnabled(true);
    final inn = await repo.insertIncoming(
      userId: _userId,
      packageName: _bankB,
      body: _inBody,
      receivedAt: t0.add(const Duration(minutes: 2)),
    );

    final merged = await _waitFor(
      repo,
      inn.id,
      (m) => m.status == RawMessageStatus.applied && m.transactionId != null,
    );
    final txRepo = container.read(transactionRepositoryProvider);
    final tx = (await txRepo.findById(merged.transactionId!))!;
    expect(tx.type, TransactionType.transfer);
    expect(tx.accountId, _accA);
    expect(tx.transferToAccountId, _accB);
    expect(tx.categoryId, isNull);

    final innBundle = decodeDraftBundle(merged.draftJson)!;
    expect(innBundle.mergeUndo, isNotNull);
    expect(innBundle.mergeUndo!['prevType'], 'expense');
    expect(merged.pairedWithId, out.id);

    // Откат доклейки из журнала.
    await container.read(inboxControllerProvider.notifier).unmergeApplied(
          userId: _userId,
          message: merged,
        );
    final restoredTx = (await txRepo.findById(tx.id))!;
    expect(restoredTx.type, TransactionType.expense);
    expect(restoredTx.transferToAccountId, isNull);

    final innAfter = (await repo.findById(inn.id))!;
    expect(innAfter.status, RawMessageStatus.inbox);
    expect(innAfter.transactionId, isNull);
    expect(innAfter.pairedWithId, isNull);
    expect(decodeDraftBundle(innAfter.draftJson)!.mergeUndo, isNull);

    final blocked = await container
        .read(transferPairingBlocklistRepositoryProvider)
        .contains(_userId, pairSignature(out.id, inn.id));
    expect(blocked, isTrue);
  });

  test('расклейка из Inbox: preMerge восстановлен, повторный матч заблокирован',
      () async {
    await bootstrap();
    _activateWorker(container);

    final (out, inn) = await mergePairFlow();

    await container.read(inboxControllerProvider.notifier).unpair(
          userId: _userId,
          message: out,
        );

    final outAfter = (await repo.findById(out.id))!;
    expect(outAfter.status, RawMessageStatus.inbox);
    expect(outAfter.pairedWithId, isNull);
    final outBundle = decodeDraftBundle(outAfter.draftJson)!;
    expect(outBundle.pairedRawMessageId, isNull);
    expect(outBundle.draft.type, TransactionType.expense,
        reason: 'тип восстановлен из preMerge');
    expect(outBundle.draft.transferToAccountId, isNull);

    final innAfter = (await repo.findById(inn.id))!;
    expect(innAfter.status, RawMessageStatus.inbox);
    expect(innAfter.pairedWithId, isNull);
    expect(decodeDraftBundle(innAfter.draftJson)!.draft.kind,
        TxKind.transferIn);

    final blocked = await container
        .read(transferPairingBlocklistRepositoryProvider)
        .contains(_userId, pairSignature(out.id, inn.id));
    expect(blocked, isTrue);

    // Повторный матч заблокирован: возвращаем secondary в ожидание пары —
    // sweep не должен склеить её с primary, стоящим в Inbox.
    await repo.holdForPairing(
      id: inn.id,
      draftJson: innAfter.draftJson!,
      deadline: DateTime.now().add(const Duration(minutes: 5)),
    );
    await container.read(parsingPipelineProvider).sweepPairs(_userId);
    final stillWaiting = (await repo.findById(inn.id))!;
    expect(stillWaiting.status, RawMessageStatus.waitingPair,
        reason: 'blocklist не даёт склеить ту же пару снова');
  });
}