"""Matching engine tests: bilateral ordering, cross-day windows, ambiguity,
same-company transfers, personal transit mappings, manual decisions, locking,
idempotency, concurrency, Decimal precision and B-44 eligibility."""

from __future__ import annotations

from decimal import Decimal
from pathlib import Path
import sqlite3
import tempfile
import threading
import unittest

from bank_importer import auth, matching, master_data, personal_transit
from bank_importer.db import connect, migrate, utc_now


class MatchingBase(unittest.TestCase):
    def setUp(self) -> None:
        self.temp_dir = tempfile.TemporaryDirectory()
        self.addCleanup(self.temp_dir.cleanup)
        root = Path(self.temp_dir.name)
        self.db_path = root / "app.db"
        self.connection = connect(self.db_path)
        self.addCleanup(self.connection.close)
        migrate(self.connection)
        self.admin = self._admin()
        self.company_a = self._company("甲公司")
        self.company_b = self._company("乙公司")
        self.company_c = self._company("丙公司")
        self.account_a = self._approved_account(self.company_a, "6222000000000001")
        self.account_b = self._approved_account(self.company_b, "6222000000000002")
        self.account_a2 = self._approved_account(self.company_a, "6222000000000003")
        self.account_b2 = self._approved_account(self.company_b, "6222000000000004")

    def _admin(self):
        auth.create_user(self.connection, "admin-u", "AdminPass123", "admin")
        return self.connection.execute(
            "SELECT * FROM users WHERE username = 'admin-u'"
        ).fetchone()

    def _company(self, name: str) -> int:
        with self.connection:
            cursor = self.connection.execute(
                "INSERT INTO companies (name, created_at, updated_at) VALUES (?, ?, ?)",
                (name, utc_now(), utc_now()),
            )
        return int(cursor.lastrowid)

    def _approved_account(self, company_id: int, number: str, start: str = "2026-01-01"):
        account = master_data.submit_bank_account(
            self.connection, company_id=company_id, bank_name="中信银行",
            account_type="基本户", account_number=number, start_date=start,
            actor=None,
        )
        return master_data.review_bank_account(
            self.connection, account["id"], "approve", None, self.admin,
            effective_from=start,
        )

    def add_row(
        self,
        company_id: int,
        *,
        own_account: str,
        own_name: str = "测试公司",
        cp_account: str | None = None,
        cp_name: str | None = None,
        income: str = "0",
        expense: str = "0",
        at: str = "2026-01-05T10:00:00",
        currency: str = "CNY",
        reference: str | None = None,
        summary: str | None = None,
        sheet: str = "流水",
    ) -> int:
        with self.connection:
            cursor = self.connection.execute(
                """
                INSERT INTO source_files (sha256, original_filename, size_bytes, storage_path, created_at)
                VALUES (?, '测试.xlsx', 1, 'data/files/测试.xlsx', ?)
                """,
                (utc_now(), utc_now()),
            )
            source_file_id = int(cursor.lastrowid)
            cursor = self.connection.execute(
                """
                INSERT INTO import_batches (source_file_id, status, company_id, created_at, updated_at)
                VALUES (?, 'parsing', ?, ?, ?)
                """,
                (source_file_id, company_id, utc_now(), utc_now()),
            )
            batch_id = int(cursor.lastrowid)
            cursor = self.connection.execute(
                """
                INSERT INTO sheet_batches (
                    import_batch_id, sheet_name, bank_name, template_id, template_version,
                    header_row, own_account, own_name, period_start, period_end,
                    transaction_count, warnings, created_at
                ) VALUES (?, ?, '测试银行', 'test-v1', 1, 1, NULL, NULL, NULL, NULL, 1, '[]', ?)
                """,
                (batch_id, sheet, utc_now()),
            )
            sheet_batch_id = int(cursor.lastrowid)
            self.connection.execute(
                """
                INSERT INTO sheet_reviews (
                    import_batch_id, sheet_name, outcome, sheet_batch_id,
                    review_status, created_at
                ) VALUES (?, ?, 'parsed', ?, 'confirmed', ?)
                """,
                (batch_id, sheet, sheet_batch_id, utc_now()),
            )
            cursor = self.connection.execute(
                """
                INSERT INTO source_rows (
                    sheet_batch_id, source_row, transaction_at, income, expense, balance,
                    own_account, own_name, counterparty_account, counterparty_name,
                    counterparty_bank, summary, purpose, reference, currency, created_at
                ) VALUES (?, 1, ?, ?, ?, NULL, ?, ?, ?, ?, NULL, ?, NULL, ?, ?, ?)
                """,
                (
                    sheet_batch_id, at, income, expense, own_account, own_name,
                    cp_account, cp_name, summary, reference, currency, utc_now(),
                ),
            )
        return int(cursor.lastrowid)

    def current(self, row_id: int):
        return self.connection.execute(
            """
            SELECT d.classification, d.pairing, d.amount, d.currency, d.effective_at,
                   d.mode, d.locked, d.revision
            FROM transfer_observation_claims c
            JOIN transfer_match_decisions d ON d.id = c.decision_id
            WHERE c.source_row_id = ?
            """,
            (row_id,),
        ).fetchone()

    def _event_of(self, row_id: int) -> int:
        return self.connection.execute(
            "SELECT event_id FROM transfer_observation_claims WHERE source_row_id = ?",
            (row_id,),
        ).fetchone()["event_id"]

    def eligible(self) -> list[sqlite3.Row]:
        return matching.eligible_intercompany_events(self.connection)

    def event_count(self) -> int:
        return self.connection.execute(
            "SELECT COUNT(*) AS n FROM canonical_transfer_events"
        ).fetchone()["n"]

    def decision_count(self) -> int:
        return self.connection.execute(
            "SELECT COUNT(*) AS n FROM transfer_match_decisions"
        ).fetchone()["n"]


class BilateralOrderTests(MatchingBase):
    def test_a_outgoing_then_b_incoming_pairs_once(self) -> None:
        row_a = self.add_row(
            self.company_a, own_account="6222000000000001",
            cp_account="6222000000000002", expense="100.00", at="2026-01-05T10:00:00",
        )
        matching.reconcile_rows(self.connection, [row_a])
        self.assertEqual("internal_single", self.current(row_a)["classification"])
        # A single internal observation must not enter the B-44 balance yet.
        self.assertEqual([], self.eligible())

        row_b = self.add_row(
            self.company_b, own_account="6222000000000002",
            cp_account="6222000000000001", income="100.00", at="2026-01-05T11:00:00",
        )
        matching.reconcile_rows(self.connection, [row_b])
        for row_id in (row_a, row_b):
            self.assertEqual("intercompany", self.current(row_id)["classification"])
            self.assertEqual("paired", self.current(row_id)["pairing"])
        eligible = self.eligible()
        self.assertEqual(1, len(eligible))
        self.assertEqual(Decimal("100.00"), Decimal(eligible[0]["amount"]))
        self.assertEqual("CNY", eligible[0]["currency"])
        self.assertEqual(self.company_a, eligible[0]["payer_company_id"])
        self.assertEqual(self.company_b, eligible[0]["payee_company_id"])
        self.assertEqual("paired", eligible[0]["pairing"])
        # One event, one paired decision for both rows.
        self.assertEqual(1, self.event_count())

    def test_b_incoming_then_a_outgoing_same_result(self) -> None:
        row_b = self.add_row(
            self.company_b, own_account="6222000000000002",
            cp_account="6222000000000001", income="100.00", at="2026-01-05T11:00:00",
        )
        matching.reconcile_rows(self.connection, [row_b])
        row_a = self.add_row(
            self.company_a, own_account="6222000000000001",
            cp_account="6222000000000002", expense="100.00", at="2026-01-05T10:00:00",
        )
        matching.reconcile_rows(self.connection, [row_a])
        eligible = self.eligible()
        self.assertEqual(1, len(eligible))
        self.assertEqual("100.00", eligible[0]["amount"])
        # Economic date is the payer's outgoing posting time, import-order free.
        self.assertEqual("2026-01-05T10:00:00", eligible[0]["effective_at"])
        self.assertEqual(self.company_a, eligible[0]["payer_company_id"])

    def test_same_batch_both_sides_pairs_once(self) -> None:
        row_a = self.add_row(
            self.company_a, own_account="6222000000000001",
            cp_account="6222000000000002", expense="100.00",
        )
        row_b = self.add_row(
            self.company_b, own_account="6222000000000002",
            cp_account="6222000000000001", income="100.00",
        )
        matching.reconcile_rows(self.connection, [row_b, row_a])
        eligible = self.eligible()
        self.assertEqual(1, len(eligible))
        self.assertEqual("paired", self.current(row_a)["pairing"])
        # One event, one auto decision carrying both observations.
        self.assertEqual(1, self.event_count())
        self.assertEqual(1, self.decision_count())

    def test_repeated_reconcile_is_idempotent(self) -> None:
        row_a = self.add_row(
            self.company_a, own_account="6222000000000001",
            cp_account="6222000000000002", expense="100.00",
        )
        row_b = self.add_row(
            self.company_b, own_account="6222000000000002",
            cp_account="6222000000000001", income="100.00",
        )
        matching.reconcile_rows(self.connection, [row_a, row_b])
        before_decisions = self.decision_count()
        stats = matching.reconcile_rows(self.connection, [row_a, row_b])
        self.assertEqual(self.decision_count(), before_decisions)
        self.assertEqual(2, stats["unchanged"])
        self.assertEqual(0, stats["created_events"])
        self.assertEqual(0, stats["updated_events"])


class DateWindowTests(MatchingBase):
    def test_m1_same_reference_within_3_days(self) -> None:
        row_a = self.add_row(
            self.company_a, own_account="6222000000000001",
            cp_account="6222000000000002", expense="100.00",
            at="2026-01-01T09:00:00", reference="R1001",
        )
        row_b = self.add_row(
            self.company_b, own_account="6222000000000002",
            cp_account="6222000000000001", income="100.00",
            at="2026-01-04T15:00:00", reference="R1001",
        )
        matching.reconcile_rows(self.connection, [row_a, row_b])
        self.assertEqual("matched", matching.exposed_status(self.current(row_a)))
        candidate = self.connection.execute(
            "SELECT rule_tier FROM transfer_match_candidates WHERE accepted = 1"
        ).fetchone()
        self.assertEqual("M1", candidate["rule_tier"])

    def test_m2_exact_mirror_same_day_without_reference(self) -> None:
        row_a = self.add_row(
            self.company_a, own_account="6222000000000001",
            cp_account="6222000000000002", expense="100.00",
        )
        row_b = self.add_row(
            self.company_b, own_account="6222000000000002",
            cp_account="6222000000000001", income="100.00",
        )
        matching.reconcile_rows(self.connection, [row_a, row_b])
        self.assertEqual("matched", matching.exposed_status(self.current(row_a)))
        candidate = self.connection.execute(
            "SELECT rule_tier FROM transfer_match_candidates WHERE accepted = 1"
        ).fetchone()
        self.assertEqual("M2", candidate["rule_tier"])

    def test_over_window_goes_to_review(self) -> None:
        row_a = self.add_row(
            self.company_a, own_account="6222000000000001",
            cp_account="6222000000000002", expense="100.00",
            at="2026-01-01T09:00:00", reference="R1001",
        )
        row_b = self.add_row(
            self.company_b, own_account="6222000000000002",
            cp_account="6222000000000001", income="100.00",
            at="2026-01-05T15:00:00", reference="R1001",
        )
        matching.reconcile_rows(self.connection, [row_a, row_b])
        self.assertEqual("needs_review", matching.exposed_status(self.current(row_a)))
        self.assertEqual([], self.eligible())

    def test_reference_conflict_goes_to_review(self) -> None:
        row_a = self.add_row(
            self.company_a, own_account="6222000000000001",
            cp_account="6222000000000002", expense="100.00",
            reference="R-A",
        )
        row_b = self.add_row(
            self.company_b, own_account="6222000000000002",
            cp_account="6222000000000001", income="100.00",
            reference="R-B",
        )
        matching.reconcile_rows(self.connection, [row_a, row_b])
        self.assertEqual("needs_review", matching.exposed_status(self.current(row_a)))


class AmbiguityTests(MatchingBase):
    def test_alias_across_companies_goes_to_review(self) -> None:
        # The same observed number is an account alias of TWO companies; the
        # counterparty cannot be uniquely decided, so it must go to review.
        company_c_account = self._approved_account(self.company_c, "6222000000000005")
        master_data.add_alias(
            self.connection, self.account_b["id"], "account", "770077007700",
            actor=self.admin,
        )
        master_data.add_alias(
            self.connection, company_c_account["id"], "account", "770077007700",
            actor=self.admin,
        )
        row_a = self.add_row(
            self.company_a, own_account="6222000000000001",
            cp_account="770077007700", expense="100.00",
        )
        matching.reconcile_rows(self.connection, [row_a])
        self.assertEqual("needs_review", matching.exposed_status(self.current(row_a)))
        self.assertEqual([], self.eligible())

    def test_two_same_tier_candidates_go_to_review(self) -> None:
        # Two identical incoming observations mirror the outgoing row exactly;
        # no deterministic rule may pick one, so the pair goes to review.
        row_a = self.add_row(
            self.company_a, own_account="6222000000000001",
            cp_account="6222000000000002", expense="100.00",
        )
        row_b1 = self.add_row(
            self.company_b, own_account="6222000000000002",
            cp_account="6222000000000001", income="100.00",
        )
        row_b2 = self.add_row(
            self.company_b, own_account="6222000000000002",
            cp_account="6222000000000001", income="100.00",
        )
        matching.reconcile_rows(self.connection, [row_a, row_b1, row_b2])
        # The ambiguous side stays in review; the two identical observations
        # may not be auto-grabbed by a processing-order tie-break.
        self.assertEqual("needs_review", matching.exposed_status(self.current(row_a)))
        self.assertEqual("internal_single", matching.exposed_status(self.current(row_b1)))
        self.assertEqual("internal_single", matching.exposed_status(self.current(row_b2)))
        self.assertEqual([], self.eligible())

    def test_different_amounts_never_match(self) -> None:
        row_a = self.add_row(
            self.company_a, own_account="6222000000000001",
            cp_account="6222000000000002", expense="100.00",
        )
        row_b = self.add_row(
            self.company_b, own_account="6222000000000002",
            cp_account="6222000000000001", income="100.01",
        )
        matching.reconcile_rows(self.connection, [row_a, row_b])
        self.assertEqual("internal_single", matching.exposed_status(self.current(row_a)))
        self.assertEqual("internal_single", matching.exposed_status(self.current(row_b)))
        self.assertEqual([], self.eligible())


class SameCompanyTests(MatchingBase):
    def test_same_company_transfer_excluded_from_intercompany(self) -> None:
        row_a = self.add_row(
            self.company_a, own_account="6222000000000001",
            cp_account="6222000000000003", expense="50.00",
        )
        row_b = self.add_row(
            self.company_a, own_account="6222000000000003",
            cp_account="6222000000000001", income="50.00",
        )
        matching.reconcile_rows(self.connection, [row_a, row_b])
        self.assertEqual("same_company_transfer", matching.exposed_status(self.current(row_a)))
        self.assertEqual("same_company_transfer", matching.exposed_status(self.current(row_b)))
        self.assertEqual([], self.eligible())
        # The cash trail is kept: both rows remain claimed by one event.
        self.assertEqual(1, self.event_count())

    def test_same_company_single_observation_classified_immediately(self) -> None:
        row_a = self.add_row(
            self.company_a, own_account="6222000000000001",
            cp_account="6222000000000003", expense="50.00",
        )
        matching.reconcile_rows(self.connection, [row_a])
        self.assertEqual("same_company_transfer", matching.exposed_status(self.current(row_a)))


class UnresolvedAndExternalTests(MatchingBase):
    def test_unknown_counterparty_stays_unresolved(self) -> None:
        row_a = self.add_row(
            self.company_a, own_account="6222000000000001",
            cp_account="9999999999999999", expense="100.00",
        )
        matching.reconcile_rows(self.connection, [row_a])
        self.assertEqual("unresolved", matching.exposed_status(self.current(row_a)))
        self.assertEqual([], self.eligible())
        unresolved = matching.unresolved_amounts(self.connection, self.company_a)
        self.assertEqual(1, len(unresolved))
        self.assertEqual("100.00", unresolved[0]["amount"])

    def test_internal_single_counts_as_unresolved_amount(self) -> None:
        row_a = self.add_row(
            self.company_a, own_account="6222000000000001",
            cp_account="6222000000000002", expense="100.00",
        )
        matching.reconcile_rows(self.connection, [row_a])
        self.assertEqual("internal_single", matching.exposed_status(self.current(row_a)))
        self.assertEqual([], self.eligible())
        unresolved = matching.unresolved_amounts(self.connection, self.company_a)
        self.assertEqual(1, len(unresolved))
        self.assertEqual("100.00", unresolved[0]["amount"])

    def test_admin_confirm_single_becomes_eligible(self) -> None:
        row_a = self.add_row(
            self.company_a, own_account="6222000000000001",
            cp_account="6222000000000002", expense="100.00",
        )
        matching.reconcile_rows(self.connection, [row_a])
        current = self.current(row_a)
        result = matching.apply_manual_decision(
            self.connection, self._event_of(row_a),
            "assign_participant",
            reason="对方公司函证确认",
            expected_revision=current["revision"],
            request_key="assign-1",
            actor=self.admin,
            participant={"role": "payee", "company_id": self.company_b},
        )
        self.assertEqual("intercompany", result["classification"])
        self.assertTrue(result["locked"])
        eligible = self.eligible()
        self.assertEqual(1, len(eligible))
        self.assertEqual("single", eligible[0]["pairing"])
        # The manual confirmation is skipped by auto reconcile.
        decisions_before = self.decision_count()
        stats = matching.reconcile_rows(self.connection, [row_a])
        self.assertEqual(1, stats["skipped_locked"])
        self.assertEqual(decisions_before, self.decision_count())

    def test_mark_external_requires_admin_and_excludes(self) -> None:
        row_a = self.add_row(
            self.company_a, own_account="6222000000000001",
            cp_account="9999999999999999", expense="100.00",
        )
        matching.reconcile_rows(self.connection, [row_a])
        event_id = self._event_of(row_a)
        result = matching.apply_manual_decision(
            self.connection, event_id, "mark_external",
            reason="经核实为外部供应商付款",
            expected_revision=self.current(row_a)["revision"],
            request_key=None, actor=self.admin,
        )
        self.assertEqual("external", result["classification"])
        self.assertEqual([], self.eligible())

    def test_reverse_frees_claims_for_reconcile(self) -> None:
        row_a = self.add_row(
            self.company_a, own_account="6222000000000001",
            cp_account="6222000000000002", expense="100.00",
        )
        row_b = self.add_row(
            self.company_b, own_account="6222000000000002",
            cp_account="6222000000000001", income="100.00",
        )
        matching.reconcile_rows(self.connection, [row_a, row_b])
        event_id = self._event_of(row_a)
        decisions_before = self.decision_count()
        result = matching.apply_manual_decision(
            self.connection, event_id, "reverse",
            reason="配对依据有误，需要重新匹配",
            expected_revision=self.current(row_a)["revision"],
            request_key="rev-1", actor=self.admin,
        )
        self.assertEqual(decisions_before + 1, self.decision_count())
        # Claims and current pointer are gone; B-44 sees nothing.
        self.assertIsNone(self.connection.execute(
            "SELECT 1 FROM transfer_observation_claims WHERE source_row_id = ?",
            (row_a,),
        ).fetchone())
        self.assertEqual([], self.eligible())
        # A re-run is allowed and re-derives deterministically.
        matching.reconcile_rows(self.connection, [row_a])
        self.assertEqual("matched", matching.exposed_status(self.current(row_a)))


class PersonalTransitTests(MatchingBase):
    def _submit_mapping(self, company_id: int, number: str, direction: str) -> int:
        mapping = personal_transit.submit_mapping(
            self.connection,
            account_number=number,
            account_name="张个人",
            represented_company_id=company_id,
            allowed_direction=direction,
            effective_from="2026-01-01",
            actor=self.admin,
        )
        approved = personal_transit.review_mapping(
            self.connection, mapping["id"], "approve", None, self.admin,
            effective_from="2026-01-01",
        )
        return approved["id"]

    def test_personal_mapping_resolves_counterparty(self) -> None:
        self._submit_mapping(self.company_b, "880088008800", "incoming")
        # A pays to the personal account -> the counterparty represents B.
        row_a = self.add_row(
            self.company_a, own_account="6222000000000001",
            cp_account="880088008800", expense="100.00",
        )
        matching.reconcile_rows(self.connection, [row_a])
        self.assertEqual("internal_single", matching.exposed_status(self.current(row_a)))
        unresolved = matching.unresolved_amounts(self.connection, self.company_a)
        self.assertEqual(1, len(unresolved))

    def test_personal_mapping_direction_mismatch_unresolved(self) -> None:
        self._submit_mapping(self.company_b, "880088008800", "incoming")
        # An outgoing row to the personal account is NOT covered by an
        # incoming-only mapping.
        row_a = self.add_row(
            self.company_a, own_account="6222000000000001",
            cp_account="880088008800", expense="100.00",
        )
        # reverse direction check: make it incoming via income
        row_b = self.add_row(
            self.company_b, own_account="6222000000000002",
            cp_account="880088008800", income="100.00",
        )
        matching.reconcile_rows(self.connection, [row_a])
        matching.reconcile_rows(self.connection, [row_b])
        self.assertEqual("unresolved", matching.exposed_status(self.current(row_b)))

    def test_unapproved_mapping_never_resolves(self) -> None:
        personal_transit.submit_mapping(
            self.connection,
            account_number="880088008801",
            account_name="李个人",
            represented_company_id=self.company_b,
            allowed_direction="both",
            actor=self.admin,
        )
        row_a = self.add_row(
            self.company_a, own_account="6222000000000001",
            cp_account="880088008801", expense="100.00",
        )
        matching.reconcile_rows(self.connection, [row_a])
        self.assertEqual("unresolved", matching.exposed_status(self.current(row_a)))

    def test_mapping_effective_window_and_resubmit_reuse(self) -> None:
        mapping = personal_transit.submit_mapping(
            self.connection,
            account_number="880088008802",
            account_name="王个人",
            represented_company_id=self.company_b,
            allowed_direction="both",
            effective_from="2026-03-01",
            actor=self.admin,
        )
        personal_transit.review_mapping(
            self.connection, mapping["id"], "approve", None, self.admin,
            effective_from="2026-03-01",
        )
        # Outside the effective window the mapping does not resolve.
        row_before = self.add_row(
            self.company_a, own_account="6222000000000001",
            cp_account="880088008802", expense="100.00", at="2026-02-20T10:00:00",
        )
        matching.reconcile_rows(self.connection, [row_before])
        self.assertEqual("unresolved", matching.exposed_status(self.current(row_before)))
        # Inside the window it does.
        row_inside = self.add_row(
            self.company_a, own_account="6222000000000001",
            cp_account="880088008802", expense="100.00", at="2026-03-05T10:00:00",
        )
        matching.reconcile_rows(self.connection, [row_inside])
        self.assertEqual("internal_single", matching.exposed_status(self.current(row_inside)))

        # Returned mappings reopen on resubmission reusing the same row.
        returned = personal_transit.submit_mapping(
            self.connection,
            account_number="880088008803",
            account_name="赵个人",
            represented_company_id=self.company_b,
            allowed_direction="both",
            actor=self.admin,
        )
        personal_transit.review_mapping(
            self.connection, returned["id"], "return", "资料待补充", self.admin,
        )
        resubmitted = personal_transit.submit_mapping(
            self.connection,
            account_number="8800 8800 8803",
            account_name="赵个人",
            represented_company_id=self.company_b,
            allowed_direction="incoming",
            actor=self.admin,
        )
        self.assertEqual(returned["id"], resubmitted["id"])
        self.assertEqual("pending", resubmitted["status"])


class ResolutionTests(MatchingBase):
    def test_name_alias_never_resolves_alone(self) -> None:
        master_data.add_alias(
            self.connection, self.account_b["id"], "name", "乙公司贸易部",
            actor=self.admin,
        )
        row_a = self.add_row(
            self.company_a, own_account="6222000000000001",
            cp_account=None, cp_name="乙公司贸易部", expense="100.00",
        )
        matching.reconcile_rows(self.connection, [row_a])
        self.assertEqual("unresolved", matching.exposed_status(self.current(row_a)))

    def test_own_account_conflict_with_upload_account_goes_to_review(self) -> None:
        # own_account belongs to company B while the batch is uploaded for A.
        row = self.add_row(
            self.company_a, own_account="6222000000000002",
            cp_account="6222000000000001", expense="100.00",
        )
        # Simulate an upload account pointing at A.
        self.connection.execute(
            """
            UPDATE import_batches SET upload_bank_account_id = ?
            WHERE id = (SELECT b.id FROM import_batches b
                        JOIN sheet_batches s ON s.import_batch_id = b.id
                        JOIN source_rows r ON r.sheet_batch_id = s.id
                        WHERE r.id = ?)
            """,
            (self.account_a["id"], row),
        )
        matching.reconcile_rows(self.connection, [row])
        self.assertEqual("needs_review", matching.exposed_status(self.current(row)))

    def test_upload_account_fallback_resolves_own(self) -> None:
        row = self.add_row(
            self.company_a, own_account=None,
            cp_account="6222000000000002", expense="100.00",
        )
        self.connection.execute(
            """
            UPDATE import_batches SET upload_bank_account_id = ?
            WHERE id = (SELECT b.id FROM import_batches b
                        JOIN sheet_batches s ON s.import_batch_id = b.id
                        JOIN source_rows r ON r.sheet_batch_id = s.id
                        WHERE r.id = ?)
            """,
            (self.account_a["id"], row),
        )
        matching.reconcile_rows(self.connection, [row])
        self.assertEqual("internal_single", matching.exposed_status(self.current(row)))

    def test_alias_mirror_m3_pairs(self) -> None:
        # A's own account has an account alias; B's statement references the
        # alias, so the mirror is proven through the approved alias (M3).
        master_data.add_alias(
            self.connection, self.account_a["id"], "account", "770077007700",
            actor=self.admin,
        )
        row_a = self.add_row(
            self.company_a, own_account="6222000000000001",
            cp_account="6222000000000002", expense="100.00", summary="货款",
        )
        row_b = self.add_row(
            self.company_b, own_account="6222000000000002",
            cp_account="770077007700", income="100.00", summary="货款",
        )
        matching.reconcile_rows(self.connection, [row_a, row_b])
        self.assertEqual("matched", matching.exposed_status(self.current(row_a)))
        candidate = self.connection.execute(
            "SELECT rule_tier FROM transfer_match_candidates WHERE accepted = 1"
        ).fetchone()
        self.assertEqual("M3", candidate["rule_tier"])


class M3EvidenceTests(MatchingBase):
    """M3 requires positive reference or summary evidence; a missing reference
    on either side is never treated as agreement (None == None is not equal)."""

    def _alias_for_a(self) -> None:
        master_data.add_alias(
            self.connection, self.account_a["id"], "account", "770077007700",
            actor=self.admin,
        )

    def test_both_references_empty_and_summaries_different_do_not_pair(self) -> None:
        self._alias_for_a()
        row_a = self.add_row(
            self.company_a, own_account="6222000000000001",
            cp_account="6222000000000002", expense="100.00", summary="货款A",
        )
        row_b = self.add_row(
            self.company_b, own_account="6222000000000002",
            cp_account="770077007700", income="100.00", summary="货款B",
        )
        matching.reconcile_rows(self.connection, [row_a, row_b])
        self.assertEqual("needs_review", matching.exposed_status(self.current(row_a)))
        self.assertEqual("internal_single", matching.exposed_status(self.current(row_b)))
        self.assertEqual([], self.eligible())
        self.assertIsNone(self.connection.execute(
            "SELECT rule_tier FROM transfer_match_candidates WHERE accepted = 1"
        ).fetchone())

    def test_single_sided_reference_is_not_reference_equality(self) -> None:
        self._alias_for_a()
        row_a = self.add_row(
            self.company_a, own_account="6222000000000001",
            cp_account="6222000000000002", expense="100.00", reference="R-ONLY-A",
        )
        row_b = self.add_row(
            self.company_b, own_account="6222000000000002",
            cp_account="770077007700", income="100.00",
        )
        matching.reconcile_rows(self.connection, [row_a, row_b])
        self.assertEqual("needs_review", matching.exposed_status(self.current(row_a)))
        self.assertEqual("internal_single", matching.exposed_status(self.current(row_b)))
        self.assertEqual([], self.eligible())

    def test_both_references_empty_but_summary_equal_still_pairs_m3(self) -> None:
        self._alias_for_a()
        row_a = self.add_row(
            self.company_a, own_account="6222000000000001",
            cp_account="6222000000000002", expense="100.00", summary="货款",
        )
        row_b = self.add_row(
            self.company_b, own_account="6222000000000002",
            cp_account="770077007700", income="100.00", summary="货款",
        )
        matching.reconcile_rows(self.connection, [row_a, row_b])
        self.assertEqual("matched", matching.exposed_status(self.current(row_a)))
        candidate = self.connection.execute(
            "SELECT rule_tier FROM transfer_match_candidates WHERE accepted = 1"
        ).fetchone()
        self.assertEqual("M3", candidate["rule_tier"])


class EligibleViewTests(MatchingBase):
    """``eligible_intercompany_events``: single-sided intercompany events only
    enter the B-44 output when locked; paired events are unaffected."""

    def _insert_intercompany(self, *, pairing: str, locked: int) -> int:
        now = utc_now()
        with self.connection:
            cursor = self.connection.execute(
                "INSERT INTO canonical_transfer_events (lifecycle, created_at) VALUES ('active', ?)",
                (now,),
            )
            event_id = int(cursor.lastrowid)
            cursor = self.connection.execute(
                """
                INSERT INTO transfer_match_decisions (
                    event_id, revision, classification, pairing, amount, currency,
                    effective_at, mode, rule_version, locked, created_at
                ) VALUES (?, 1, 'intercompany', ?, '100.00', 'CNY',
                          '2026-01-05T10:00:00', 'auto', 'transfer-match-v1', ?, ?)
                """,
                (event_id, pairing, locked, now),
            )
            decision_id = int(cursor.lastrowid)
            for role, company_id in (("payer", self.company_a), ("payee", self.company_b)):
                self.connection.execute(
                    """
                    INSERT INTO transfer_decision_participants (
                        decision_id, role, company_id, bank_account_id,
                        resolve_method, evidence, created_at
                    ) VALUES (?, ?, ?, NULL, 'own_exact', '{}', ?)
                    """,
                    (decision_id, role, company_id, now),
                )
            self.connection.execute(
                "INSERT INTO current_transfer_decisions (event_id, decision_id) VALUES (?, ?)",
                (event_id, decision_id),
            )
        return event_id

    def test_unlocked_single_intercompany_is_excluded(self) -> None:
        self._insert_intercompany(pairing="single", locked=0)
        self.assertEqual([], self.eligible())

    def test_locked_single_intercompany_is_included(self) -> None:
        self._insert_intercompany(pairing="single", locked=1)
        eligible = self.eligible()
        self.assertEqual(1, len(eligible))
        self.assertEqual("single", eligible[0]["pairing"])
        self.assertEqual("100.00", eligible[0]["amount"])

    def test_paired_intercompany_is_included_regardless_of_lock(self) -> None:
        self._insert_intercompany(pairing="paired", locked=0)
        self._insert_intercompany(pairing="paired", locked=1)
        eligible = self.eligible()
        self.assertEqual(2, len(eligible))
        self.assertTrue(all(item["pairing"] == "paired" for item in eligible))


class ManualLinkTests(MatchingBase):
    def test_manual_link_locks_pair(self) -> None:
        row_a = self.add_row(
            self.company_a, own_account="6222000000000001",
            cp_account="6222000000000002", expense="100.00",
        )
        row_b = self.add_row(
            self.company_b, own_account="6222000000000002",
            cp_account="6222000000000001", income="100.00",
        )
        matching.reconcile_rows(self.connection, [row_a])
        matching.reconcile_rows(self.connection, [row_b])
        event_id = self._event_of(row_a)
        current_revision = self.current(row_a)["revision"]
        result = matching.apply_manual_decision(
            self.connection, event_id, "link_rows",
            reason="人工核对后确认是同一笔",
            expected_revision=current_revision,
            request_key="link-1", actor=self.admin,
            source_row_ids=[row_a, row_b],
        )
        self.assertTrue(result["locked"])
        self.assertEqual("paired", result["pairing"])
        # Auto reconcile never touches locked decisions.
        stats = matching.reconcile_rows(self.connection, [row_a, row_b])
        self.assertEqual(2, stats["skipped_locked"])

    def test_manual_decision_requires_reason(self) -> None:
        row_a = self.add_row(
            self.company_a, own_account="6222000000000001",
            cp_account="6222000000000002", expense="100.00",
        )
        matching.reconcile_rows(self.connection, [row_a])
        with self.assertRaises(matching.MatchInputError):
            matching.apply_manual_decision(
                self.connection, self._event_of(row_a), "reverse",
                reason=" ", expected_revision=None, request_key=None,
                actor=self.admin,
            )

    def test_stale_expected_revision_conflicts(self) -> None:
        row_a = self.add_row(
            self.company_a, own_account="6222000000000001",
            cp_account="6222000000000002", expense="100.00",
        )
        matching.reconcile_rows(self.connection, [row_a])
        event_id = self._event_of(row_a)
        with self.assertRaises(matching.MatchConflictError):
            matching.apply_manual_decision(
                self.connection, event_id, "reverse",
                reason="测试", expected_revision=999, request_key=None,
                actor=self.admin,
            )


class DecimalPrecisionTests(MatchingBase):
    def test_equivalent_decimal_strings_match(self) -> None:
        row_a = self.add_row(
            self.company_a, own_account="6222000000000001",
            cp_account="6222000000000002", expense="100.0",
        )
        row_b = self.add_row(
            self.company_b, own_account="6222000000000002",
            cp_account="6222000000000001", income="100.00",
        )
        matching.reconcile_rows(self.connection, [row_a, row_b])
        self.assertEqual("matched", matching.exposed_status(self.current(row_a)))

    def test_cent_difference_never_matches(self) -> None:
        row_a = self.add_row(
            self.company_a, own_account="6222000000000001",
            cp_account="6222000000000002", expense="100.00",
        )
        row_b = self.add_row(
            self.company_b, own_account="6222000000000002",
            cp_account="6222000000000001", income="100.01",
        )
        matching.reconcile_rows(self.connection, [row_a, row_b])
        self.assertEqual("internal_single", matching.exposed_status(self.current(row_a)))

    def test_large_amount_precision(self) -> None:
        amount = "99999999999999999999.99"
        row_a = self.add_row(
            self.company_a, own_account="6222000000000001",
            cp_account="6222000000000002", expense=amount,
        )
        row_b = self.add_row(
            self.company_b, own_account="6222000000000002",
            cp_account="6222000000000001", income=amount,
        )
        matching.reconcile_rows(self.connection, [row_a, row_b])
        eligible = self.eligible()
        self.assertEqual(1, len(eligible))
        self.assertEqual(Decimal(amount), Decimal(eligible[0]["amount"]))

    def test_currency_mismatch_never_matches(self) -> None:
        row_a = self.add_row(
            self.company_a, own_account="6222000000000001",
            cp_account="6222000000000002", expense="100.00", currency="CNY",
        )
        row_b = self.add_row(
            self.company_b, own_account="6222000000000002",
            cp_account="6222000000000001", income="100.00", currency="USD",
        )
        matching.reconcile_rows(self.connection, [row_a, row_b])
        self.assertEqual("internal_single", matching.exposed_status(self.current(row_a)))


class ProjectionRebuildTests(MatchingBase):
    def test_rebuild_matches_current_projection(self) -> None:
        row_a = self.add_row(
            self.company_a, own_account="6222000000000001",
            cp_account="6222000000000002", expense="100.00",
        )
        row_b = self.add_row(
            self.company_b, own_account="6222000000000002",
            cp_account="6222000000000001", income="100.00",
        )
        matching.reconcile_rows(self.connection, [row_a, row_b])
        before = [
            (item["event_id"], item["decision_id"])
            for item in self.connection.execute(
                "SELECT * FROM current_transfer_decisions"
            ).fetchall()
        ]
        claims_before = [
            (item["source_row_id"], item["event_id"], item["decision_id"])
            for item in self.connection.execute(
                "SELECT * FROM transfer_observation_claims"
            ).fetchall()
        ]
        matching.rebuild_current_projection(self.connection)
        after = [
            (item["event_id"], item["decision_id"])
            for item in self.connection.execute(
                "SELECT * FROM current_transfer_decisions"
            ).fetchall()
        ]
        claims_after = [
            (item["source_row_id"], item["event_id"], item["decision_id"])
            for item in self.connection.execute(
                "SELECT * FROM transfer_observation_claims"
            ).fetchall()
        ]
        self.assertEqual(sorted(before), sorted(after))
        self.assertEqual(sorted(claims_before), sorted(claims_after))


class ConcurrentReconcileTests(MatchingBase):
    def test_concurrent_reconcile_creates_one_event(self) -> None:
        row_a = self.add_row(
            self.company_a, own_account="6222000000000001",
            cp_account="6222000000000002", expense="100.00",
        )
        row_b = self.add_row(
            self.company_b, own_account="6222000000000002",
            cp_account="6222000000000001", income="100.00",
        )
        errors: list[Exception] = []

        def run() -> None:
            db = connect(self.db_path)
            try:
                matching.reconcile_rows(db, [row_a, row_b])
            except Exception as exc:  # pragma: no cover
                errors.append(exc)
            finally:
                db.close()

        threads = [threading.Thread(target=run) for _ in range(2)]
        for thread in threads:
            thread.start()
        for thread in threads:
            thread.join()
        self.assertEqual([], errors)
        self.assertEqual(1, self.event_count())
        eligible = self.eligible()
        self.assertEqual(1, len(eligible))
        claims = self.connection.execute(
            "SELECT COUNT(*) AS n FROM transfer_observation_claims"
        ).fetchone()["n"]
        self.assertEqual(2, claims)


if __name__ == "__main__":
    unittest.main()