OnBudget/lib/src/features/notification_parsing/application/parsing_worker.dart

import 'package:riverpod_annotation/riverpod_annotation.dart';

import '../../categories/application/category_providers.dart';
import '../../transactions/application/transactions_controller.dart';
import '../data/parser/account_resolver.dart';
import '../data/parser/confidence_scorer.dart';
import '../data/parser/decision_gate.dart';
import '../data/parser/draft_codec.dart';
import '../data/parser/regex_parser.dart';
import '../data/parser/rule_lookup.dart';
import '../data/parser/rule_suggester.dart';
import '../domain/entities/parse_draft.dart';
import '../domain/entities/raw_message.dart';
import '../domain/entities/rule_candidate.dart';
import '../domain/entities/rule_suggestion.dart';
import '../domain/enums.dart';
import 'notification_parsing_providers.dart';
import 'parsing_settings_controller.dart';

part 'parsing_worker.g.dart';

/// Поток необработанных сообщений — вход воркера.
@riverpod
Stream<List<RawMessage>> pendingMessages(Ref ref, String userId) =>
    ref.watch(rawMessagesRepositoryProvider).watchPending(userId);

/// ParsingWorker (§5): слушает `raw_messages.pending` и прогоняет pipeline
/// regex → resolver → rule_lookup → suggester → scorer → gate по одному.
///
/// Идемпотентен: при возврате сообщения в `pending` перепарсивается.
/// Провайдер `keepAlive` — активируется чтением из HomeScreen.
@Riverpod(keepAlive: true)
class ParsingWorker extends _$ParsingWorker {
  final Set<String> _inFlight = {};

  @override
  void build(String userId) {
    final sub = ref.listen(
      pendingMessagesProvider(userId),
      (_, next) {
        final list = next.value;
        if (list != null && list.isNotEmpty) {
          _drain(userId, list);
        }
      },
      fireImmediately: true,
    );
    ref.onDispose(sub.close);
  }

  Future<void> _drain(String userId, List<RawMessage> pending) async {
    for (final msg in pending) {
      if (_inFlight.contains(msg.id)) continue;
      _inFlight.add(msg.id);
      final repo = ref.read(rawMessagesRepositoryProvider);
      try {
        await repo.incrementParseAttempts(msg.id);
        await _process(userId, msg);
      } catch (e) {
        await repo.updateAfterParse(
          id: msg.id,
          status: RawMessageStatus.failed,
        );
      } finally {
        _inFlight.remove(msg.id);
      }
    }
  }

  Future<void> _process(String userId, RawMessage msg) async {
    final repo = ref.read(rawMessagesRepositoryProvider);

    final settings = await ref.read(parsingSettingsControllerProvider.future);
    if (!settings.enabled) return; // фича выключена — оставляем pending.

    // 1. Regex (Phase 1 — без AI fallback).
    final parsed = const RegexParser().parse(msg);
    if (parsed == null) {
      // Не распознано: баланс/реклама → ignored, иначе → Inbox (вручную).
      if (looksLikeNonTransaction(msg.body)) {
        await repo.updateStatus(msg.id, RawMessageStatus.ignored);
      } else {
        await repo.updateAfterParse(id: msg.id, status: RawMessageStatus.inbox);
      }
      return;
    }

    // 3. Разрешение счёта.
    final resolver =
        AccountResolver(ref.read(accountBindingsRepositoryProvider));
    final resolution = await resolver.resolve(
      userId: userId,
      packageName: msg.packageName,
      cardLast4: parsed.cardLast4,
      phone: parsed.counterpartyPhone,
    );
    var draft = parsed.copyWith(accountId: resolution.accountId);

    // 4. Правила: исключение → ignored; иначе ищем merchantToCategory.
    final rules =
        await ref.read(parseRulesRepositoryProvider).getEnabledByUser(userId);
    if (findIgnoreRule(rules, body: msg.body, merchantRaw: draft.merchantRaw) !=
        null) {
      await repo.updateStatus(msg.id, RawMessageStatus.ignored);
      return;
    }
    final rule =
        findMerchantRule(rules, body: msg.body, merchantRaw: draft.merchantRaw);

    RuleCandidate? candidate;
    if (draft.merchantRaw != null) {
      candidate = await ref.read(ruleCandidatesRepositoryProvider).findByRawValue(
            userId,
            ParseRuleKind.merchantToCategory,
            draft.merchantRaw!,
          );
    }

    // Применяем действие правила к draft.
    if (rule != null) {
      draft = draft.copyWith(
        merchantCanonical: rule.merchantCanonical,
        categoryId: rule.categoryId,
        accountId: draft.accountId ?? rule.accountId,
      );
    }

    // Предложение правила для Inbox (только для незнакомого мерчанта).
    RuleSuggestion? suggestion;
    if (rule == null && draft.merchantRaw != null) {
      suggestion =
          suggestRule(merchantRaw: draft.merchantRaw!, categoryCandidate: candidate);
      if (suggestion.categoryId != null) {
        final cat = await ref
            .read(categoryRepositoryProvider)
            .findById(suggestion.categoryId!);
        suggestion = suggestion.copyWith(categoryName: cat?.name);
      }
    }

    // 6. Confidence.
    final scores = scoreDraft(
      draft: draft,
      message: msg,
      accountScore: resolution.score,
      merchantRule: rule,
      merchantCandidate: candidate,
      categoryCandidate: candidate,
    );

    // 8. Gate.
    final decision = decide(
      merchantRule: rule,
      sanityPassed: true,
      scores: scores,
      strictness: settings.strictness,
      amountMinor: draft.amount,
      hasAccount: draft.accountId != null,
    );

    if (decision == GateDecision.autoApply) {
      await _autoApply(userId, msg, draft, rule!, scores, resolution.bindingId);
    } else {
      await repo.updateAfterParse(
        id: msg.id,
        status: RawMessageStatus.inbox,
        draftJson: encodeDraftBundle(draft, suggestion),
        confidenceAmount: scores.amount,
        confidenceAccount: scores.account,
        confidenceType: scores.type,
        confidenceMerchant: scores.merchant,
        confidenceCategory: scores.category,
      );
    }
  }

  Future<void> _autoApply(
    String userId,
    RawMessage msg,
    ParseDraft draft,
    rule,
    FieldScores scores,
    String? bindingId,
  ) async {
    final repo = ref.read(rawMessagesRepositoryProvider);
    final tx =
        await ref.read(transactionsControllerProvider.notifier).createTransaction(
              userId: userId,
              accountId: draft.accountId!,
              categoryId: draft.categoryId,
              type: draft.type,
              amount: draft.amount,
              date: draft.dateTime ?? msg.receivedAt,
              merchant: draft.merchantCanonical ?? draft.merchantRaw,
              rawMessageId: msg.id,
              autoApplied: true,
              appliedByRuleId: rule.id,
            );

    await repo.updateAfterParse(
      id: msg.id,
      status: RawMessageStatus.applied,
      draftJson: encodeDraftBundle(draft, null),
      confidenceAmount: scores.amount,
      confidenceAccount: scores.account,
      confidenceType: scores.type,
      confidenceMerchant: scores.merchant,
      confidenceCategory: scores.category,
    );
    await repo.linkTransaction(msg.id, tx.id);

    await ref
        .read(parseRulesRepositoryProvider)
        .incrementMatchCount(rule.id, DateTime.now());
    if (bindingId != null) {
      await ref
          .read(accountBindingsRepositoryProvider)
          .incrementMatchCount(bindingId);
    }
  }
}