"""Real PostgreSQL competing writers, with independent thread connections."""

from concurrent.futures import ThreadPoolExecutor
from datetime import date
from decimal import Decimal
from threading import Barrier

import pytest
from django.db import close_old_connections, connection, connections

from finance.models import AuditEvent, Transaction, Transfer
from finance.services import Actor, DomainError, create_account, match_transfer, record_transaction

pytestmark = pytest.mark.django_db(transaction=True)


def test_competing_matches_cannot_reuse_one_outgoing_leg():
    assert connection.vendor == "postgresql"
    actor = Actor(kind="AGENT", identifier="synthetic-concurrency-test")
    bank = create_account(name="Synthetic Outgoing", kind="BANK", actor=actor)
    other = create_account(name="Synthetic Incoming", kind="BANK", actor=actor)
    common = dict(kind="TRANSFER", posting_date=date(2026, 1, 15), actor=actor)
    outgoing = record_transaction(account_id=bank.pk, amount=Decimal("-25"), **common)
    incoming = [record_transaction(account_id=other.pk, amount=Decimal("25"), **common) for _ in range(2)]
    barrier = Barrier(2)
    audit_ids = set(AuditEvent.objects.values_list("pk", flat=True))

    def attempt(incoming_id):
        close_old_connections()
        try:
            # Opening connections before the barrier ensures two real sessions.
            with connection.cursor() as cursor:
                cursor.execute("SELECT pg_backend_pid()")
                backend_pid = cursor.fetchone()[0]
            barrier.wait(timeout=15)
            try:
                result = match_transfer(outgoing_id=outgoing.pk, incoming_id=incoming_id, actor=actor)
                return backend_pid, "success", result.pk
            except DomainError as error:
                return backend_pid, "failure", error.code
        finally:
            connections.close_all()

    with ThreadPoolExecutor(max_workers=2) as pool:
        results = list(pool.map(attempt, [row.pk for row in incoming]))

    assert len({result[0] for result in results}) == 2
    assert sorted(result[1] for result in results) == ["failure", "success"]
    assert all(result[2] for result in results)
    assert Transfer.objects.count() == 1
    paired = Transfer.objects.get()
    assert paired.outgoing_id == outgoing.pk
    assert paired.incoming_id in {row.pk for row in incoming}
    assert Transaction.objects.count() == 3
    outcomes = list(AuditEvent.objects.exclude(pk__in=audit_ids).values_list("outcome", flat=True))
    assert sorted(outcomes) == ["FAILURE", "SUCCESS"]