"""Competing import commits use independent PostgreSQL connections."""

from concurrent.futures import ThreadPoolExecutor
from threading import Barrier

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

from finance.models import AuditEvent, SourceRecord, Transaction
from finance.services import Actor, DomainError, create_account
from ingestion import services

pytestmark = pytest.mark.django_db(transaction=True)
ACTOR = Actor(kind="AGENT", identifier="synthetic-import-concurrency")
DATA = b"Data,Documento,Historico,Agencia,Lote,Valor\n15/01/2026,1,Synthetic purchase,0,0,-12.50\n"


def reviewed(bank):
    batch = services.create_preview(data=DATA, profile="BANK_CSV_V1", section_mapping={"1": str(bank.pk)}, actor=ACTOR)
    return services.apply_decisions(batch_id=batch.pk, revision=batch.revision, actor=ACTOR,
                                    decisions=[{"locator": batch.items.get().locator, "action": "CREATE"}])


def race(batches):
    assert connection.vendor == "postgresql"
    barrier = Barrier(2)

    def attempt(batch):
        close_old_connections()
        try:
            with connection.cursor() as cursor:
                cursor.execute("SET statement_timeout = '20s'")
                cursor.execute("SELECT pg_backend_pid()")
                backend_pid = cursor.fetchone()[0]
            barrier.wait(timeout=15)
            try:
                result = services.commit_preview(batch_id=batch.pk, revision=batch.revision, actor=ACTOR)
                return backend_pid, "success", result.result
            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, batches))
    assert len({item[0] for item in results}) == 2
    return results


def test_two_commits_of_same_batch_return_same_result_without_duplicate_audit():
    bank = create_account(name="Synthetic concurrent bank", kind="BANK", actor=ACTOR)
    batch = reviewed(bank)
    previous_audits = set(AuditEvent.objects.values_list("pk", flat=True))
    results = race([batch, batch])
    assert [item[1] for item in results] == ["success", "success"]
    assert results[0][2] == results[1][2]
    assert Transaction.objects.count() == SourceRecord.objects.count() == 1
    assert list(AuditEvent.objects.exclude(pk__in=previous_audits).values_list("outcome", flat=True)) == ["SUCCESS"]


def test_identical_batches_have_one_success_and_one_stale_review():
    bank = create_account(name="Synthetic concurrent bank", kind="BANK", actor=ACTOR)
    batches = [reviewed(bank), reviewed(bank)]
    previous_audits = set(AuditEvent.objects.values_list("pk", flat=True))
    results = race(batches)
    assert sorted(item[1] for item in results) == ["failure", "success"]
    assert next(item[2] for item in results if item[1] == "failure") == "STALE_PREVIEW"
    assert Transaction.objects.count() == SourceRecord.objects.count() == 1
    for batch in batches:
        batch.refresh_from_db()
    assert sorted(batch.state for batch in batches) == ["COMMITTED", "PREVIEW"]
    assert sorted(AuditEvent.objects.exclude(pk__in=previous_audits).values_list("outcome", flat=True)) == ["FAILURE", "SUCCESS"]