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

import pytest
from django.contrib.auth import get_user_model
from django.db import close_old_connections, connection
from django.urls import reverse

from finance.balance_adjustments.services import create_adjustment, void_adjustment
from finance.models import (
    FinanceSettings,
    PayrollAuditEvent,
    PayrollBalanceAdjustment,
    PayrollInputLine,
    PayrollPeriod,
    PayrollRun,
    PayrollStatement,
    PayrollStatementLine,
    PayrollWorkRecord,
)
from finance.payroll.balances import opening_balances
from finance.payroll.exceptions import PayrollOperationError, PayrollPermissionDenied
from finance.payroll.reporting import build_payroll_period_table
from finance.payroll.services import calculate_period
from tests.finance.test_payroll_admin_api import grant, make_approved_period_inputs

pytestmark = pytest.mark.django_db


def url(name, **kwargs):
    return reverse(f"api:v1:finance-payroll:{name}", kwargs=kwargs or None)


@pytest.fixture
def setup(user_factory, auth_client_factory):
    manager = user_factory()
    employee = user_factory()
    grant(manager, "manage_payroll_inputs", "view_all_payroll", "calculate_payroll")
    periods = {}
    for month in (8, 9, 10, 11):
        periods[month] = PayrollPeriod.objects.create(
            code=f"2026-{month:02}",
            name=f"Месяц {month}",
            date_from=date(2026, month, 1),
            date_to=date(2026, month, calendar.monthrange(2026, month)[1]),
            status="closed" if month < 10 else "open",
            created_by=manager,
        )
    payload = {
        "employee_id": employee.pk,
        "effective_period_id": periods[10].pk,
        "source_period_id": periods[8].pk,
        "direction": "company_owes",
        "amount": "1234.56",
        "request_id": str(uuid.uuid4()),
    }
    return manager, employee, periods, payload, auth_client_factory(manager)


def create(setup, **overrides):
    _, _, _, payload, client = setup
    return client.post(
        url("admin-balance-adjustment-create"), {**payload, **overrides}, format="json"
    )


@pytest.mark.parametrize(
    "direction,contribution",
    [("company_owes", "1234.56"), ("employee_owes", "-1234.56")],
)
def test_effective_from_october_once_and_no_payroll_inputs(
    setup, direction, contribution
):
    manager, employee, periods, _, client = setup
    frozen = list(
        PayrollPeriod.objects.filter(pk__in=[periods[8].pk, periods[9].pk]).values()
    )
    result = create(setup, direction=direction)
    assert result.status_code == 201, result.content
    adjustment = PayrollBalanceAdjustment.objects.get()
    assert adjustment.reason == ""
    assert adjustment.created_by == manager
    assert not PayrollInputLine.objects.exists()
    assert not PayrollStatement.objects.exists()
    assert (
        PayrollAuditEvent.objects.filter(
            action="payroll.balance_adjustment_created"
        ).count()
        == 1
    )
    for month in (8, 9):
        assert opening_balances(periods[month], [employee.pk]) == {}
        ledger = client.get(url("admin-period-carryovers", pk=periods[month].pk)).json()
        assert ledger["entries"] == []
    for month in (10, 11):
        assert opening_balances(periods[month], [employee.pk]) == {
            employee.pk: Decimal(contribution)
        }
        ledger = client.get(url("admin-period-carryovers", pk=periods[month].pk)).json()
        assert ledger["summary"]["balance"] == contribution
        assert len(ledger["entries"]) == 1
        entry = ledger["entries"][0]
        assert entry["entry_type"] == "manual_adjustment"
        assert entry["can_void"] is True
        assert entry["source_period"]["id"] == periods[8].pk
    assert frozen == list(
        PayrollPeriod.objects.filter(pk__in=[periods[8].pk, periods[9].pk]).values()
    )


def test_idempotency_and_conflicting_retries(setup):
    assert create(setup).status_code == 201
    assert create(setup).status_code == 200
    result = create(setup, amount="99")
    assert result.status_code == 409, result.content
    assert result.json()["code"] == "BALANCE_ADJUSTMENT_REQUEST_CONFLICT"
    assert PayrollBalanceAdjustment.objects.count() == 1


@pytest.mark.django_db(transaction=True)
def test_concurrent_retries_are_exactly_once_on_postgres(setup):
    if connection.vendor != "postgresql":
        pytest.skip("PostgreSQL row-lock integration test")
    manager, employee, periods, payload, _ = setup
    FinanceSettings.get_solo()
    barrier = Barrier(2)

    def record():
        close_old_connections()
        try:
            actor = get_user_model().objects.get(pk=manager.pk)
            person = get_user_model().objects.get(pk=employee.pk)
            effective = PayrollPeriod.objects.get(pk=periods[10].pk)
            source = PayrollPeriod.objects.get(pk=periods[8].pk)
            barrier.wait(timeout=15)
            adjustment, created = create_adjustment(
                actor=actor,
                employee=person,
                effective_period=effective,
                source_period=source,
                direction="company_owes",
                amount=Decimal("1234.56"),
                reason="",
                request_id=uuid.UUID(payload["request_id"]),
            )
            return adjustment.pk, created
        finally:
            close_old_connections()

    with ThreadPoolExecutor(max_workers=2) as pool:
        first = pool.submit(record)
        second = pool.submit(record)
        results = [first.result(timeout=30), second.result(timeout=30)]
    assert results[0][0] == results[1][0]
    assert sorted(result[1] for result in results) == [False, True]
    assert PayrollBalanceAdjustment.objects.count() == 1
    assert (
        PayrollAuditEvent.objects.filter(
            action="payroll.balance_adjustment_created"
        ).count()
        == 1
    )


def test_void_keeps_history_and_cancels_only_balance(setup):
    manager, employee, periods, _, client = setup
    assert create(setup).status_code == 201
    adjustment = PayrollBalanceAdjustment.objects.get()
    result = client.post(
        url("admin-balance-adjustment-void", pk=adjustment.pk), {}, format="json"
    )
    assert result.status_code == 200, result.content
    assert (
        client.post(
            url("admin-balance-adjustment-void", pk=adjustment.pk), {}, format="json"
        ).status_code
        == 200
    )
    adjustment.refresh_from_db()
    assert adjustment.voided_by == manager
    assert adjustment.voided_at
    assert not opening_balances(periods[10], [employee.pk])
    entry = client.get(url("admin-period-carryovers", pk=periods[10].pk)).json()[
        "entries"
    ][0]
    assert entry["status"] == "voided"
    assert entry["contribution"] == "0.00"
    assert entry["can_void"] is False
    assert entry["voided_by"]["id"] == manager.pk
    assert (
        PayrollAuditEvent.objects.filter(
            action="payroll.balance_adjustment_voided"
        ).count()
        == 1
    )
    assert not PayrollInputLine.objects.exists()


@pytest.mark.parametrize(
    "status", ["calculated", "review", "approved", "published", "closed"]
)
def test_calculated_and_later_frozen_periods_block_create_and_void(setup, status):
    _, employee, periods, _, client = setup
    assert create(setup).status_code == 201
    adjustment = PayrollBalanceAdjustment.objects.get()
    PayrollPeriod.objects.filter(pk=periods[11].pk).update(status=status)
    assert create(setup, request_id=str(uuid.uuid4())).status_code == 409
    result = client.post(
        url("admin-balance-adjustment-void", pk=adjustment.pk), {}, format="json"
    )
    assert result.status_code == 409
    assert opening_balances(periods[10], [employee.pk])[employee.pk] == Decimal(
        "1234.56"
    )
    entry = client.get(url("admin-period-carryovers", pk=periods[10].pk)).json()[
        "entries"
    ][0]
    assert entry["can_void"] is False


def test_closed_effective_period_rejected(setup):
    assert create(setup, effective_period_id=setup[2][9].pk).status_code == 409
    assert not PayrollBalanceAdjustment.objects.exists()


def test_future_effective_period_does_not_change_current_balance(setup):
    _, employee, periods, _, client = setup
    assert create(setup, effective_period_id=periods[11].pk).status_code == 201
    assert opening_balances(periods[10], [employee.pk]) == {}
    assert (
        client.get(url("admin-period-carryovers", pk=periods[10].pk)).json()["entries"]
        == []
    )
    assert opening_balances(periods[11], [employee.pk])[employee.pk] == Decimal(
        "1234.56"
    )


def test_retry_of_already_voided_adjustment_after_closing_is_read_only(setup):
    _, _, periods, _, client = setup
    assert create(setup).status_code == 201
    adjustment = PayrollBalanceAdjustment.objects.get()
    endpoint = url("admin-balance-adjustment-void", pk=adjustment.pk)
    assert client.post(endpoint, {}, format="json").status_code == 200
    PayrollPeriod.objects.filter(pk=periods[10].pk).update(status="closed")
    assert client.post(endpoint, {}, format="json").status_code == 200
    assert (
        PayrollAuditEvent.objects.filter(
            action="payroll.balance_adjustment_voided"
        ).count()
        == 1
    )


@pytest.mark.parametrize(
    "overrides",
    [
        {"amount": "0"},
        {"amount": "-1"},
        {"amount": "1.001"},
        {"direction": "unknown"},
        {"request_id": "invalid"},
    ],
)
def test_invalid_input(setup, overrides):
    assert create(setup, **overrides).status_code == 400
    assert not PayrollBalanceAdjustment.objects.exists()


def test_source_is_reference_only_and_must_be_earlier_same_currency(setup):
    periods = setup[2]
    assert create(setup, source_period_id=periods[10].pk).status_code == 409
    PayrollPeriod.objects.filter(pk=periods[8].pk).update(currency="USD")
    assert create(setup).status_code == 409
    assert create(setup, source_period_id=None).status_code == 201


def test_inactive_allowed_hidden_forbidden(setup):
    employee = setup[1]
    employee.is_active = False
    employee.save(update_fields=["is_active"])
    assert create(setup).status_code == 201
    employee.is_hidden_in_ui = True
    employee.save(update_fields=["is_hidden_in_ui"])
    assert create(setup, request_id=str(uuid.uuid4())).status_code == 400


def test_permissions_and_management_allowlist(setup, user_factory, auth_client_factory):
    manager, employee, periods, payload, client = setup
    outsider = user_factory()
    grant(outsider, "view_all_payroll")
    outsider_client = auth_client_factory(outsider)
    assert (
        outsider_client.post(
            url("admin-balance-adjustment-create"), payload, format="json"
        ).status_code
        == 403
    )
    with pytest.raises(PayrollPermissionDenied):
        create_adjustment(
            actor=outsider, employee=employee, effective_period=periods[10]
        )
    settings = FinanceSettings.get_solo()
    settings.management_users.add(outsider)
    assert create(setup).status_code == 403
    with pytest.raises(PayrollPermissionDenied):
        create_adjustment(
            actor=manager, employee=employee, effective_period=periods[10]
        )


def test_preview_and_calculation_change_only_balance_preserve_september_snapshot(
    setup, user_factory
):
    manager, employee, periods, _, client = setup
    approver = user_factory()
    september = periods[9]
    september.status = "open"
    september.save(update_fields=["status"])
    make_approved_period_inputs(
        period=september, employee=employee, creator=manager, approver=approver
    )
    september_run = calculate_period(september.pk, actor=manager)
    september.refresh_from_db()
    september.status = "closed"
    september.save(update_fields=["status"])
    snapshot_before = list(PayrollStatement.objects.filter(run=september_run).values())
    work = PayrollWorkRecord.objects.get(period=september, employee=employee)
    work.pk = None
    work.period = periods[10]
    work.save()
    before = next(
        row
        for row in build_payroll_period_table(periods[10])["rows"]
        if row["employee"]["id"] == employee.pk
    )
    assert create(setup).status_code == 201
    after = next(
        row
        for row in build_payroll_period_table(periods[10])["rows"]
        if row["employee"]["id"] == employee.pk
    )
    assert Decimal(after["opening_balance"]) - Decimal(
        before["opening_balance"]
    ) == Decimal("1234.56")
    assert Decimal(after["closing_balance"]) - Decimal(
        before["closing_balance"]
    ) == Decimal("1234.56")
    for key in (
        "gross_total",
        "deduction_total",
        "net_pay",
        "payment_total",
        "payable",
    ):
        assert before[key] == after[key], key
    run = calculate_period(periods[10].pk, actor=manager)
    statement = PayrollStatement.objects.get(run=run, employee=employee)
    assert statement.opening_balance == Decimal(after["opening_balance"])
    assert statement.closing_balance == Decimal(after["closing_balance"])
    assert statement.payable == Decimal(before["payable"])
    assert snapshot_before == list(
        PayrollStatement.objects.filter(run=september_run).values()
    )
    assert not PayrollInputLine.objects.exists()
    adjustment = PayrollBalanceAdjustment.objects.get()
    assert (
        client.post(
            url("admin-balance-adjustment-void", pk=adjustment.pk), {}, format="json"
        ).status_code
        == 409
    )
    with pytest.raises(PayrollOperationError):
        void_adjustment(adjustment.pk, actor=manager)


@pytest.mark.parametrize(
    "direction,expected", [("company_owes", "1234.56"), ("employee_owes", "-1234.56")]
)
def test_inactive_employee_with_only_manual_balance_is_visible_without_salary(
    setup, user_factory, direction, expected
):
    manager, employee, periods, _, client = setup
    # An old approved salary rate must not turn a balance-only row into salary.
    make_approved_period_inputs(
        period=periods[8], employee=employee, creator=manager, approver=user_factory()
    )
    employee.is_active = False
    employee.save(update_fields=["is_active"])
    assert create(setup, direction=direction).status_code == 201
    for month in (10, 11):
        table = client.get(url("admin-period-table", pk=periods[month].pk))
        assert table.status_code == 200, table.content
        row = next(
            r for r in table.json()["rows"] if r["employee"]["id"] == employee.pk
        )
        assert row["status"] == "balance_only"
        assert row["opening_balance"] == expected
        assert row["closing_balance"] == expected
        assert row["rate_amount"] is None
        assert row["attendance_alert"] is None
        for key in (
            "gross_total",
            "deduction_total",
            "net_pay",
            "payment_total",
            "payable",
        ):
            assert Decimal(row[key]) == 0, key
        assert len(row["manual_balance_adjustments"]) == int(month == 10)
    assert not PayrollStatement.objects.exists()
    assert not PayrollInputLine.objects.exists()


def test_current_offsetting_adjustments_are_visible_but_settled_balance_is_not_carried(
    setup,
):
    employee, periods = setup[1:3]
    employee.is_active = False
    employee.save(update_fields=["is_active"])
    assert create(setup).status_code == 201
    assert (
        create(
            setup, direction="employee_owes", request_id=str(uuid.uuid4())
        ).status_code
        == 201
    )
    october = build_payroll_period_table(periods[10])
    row = next(r for r in october["rows"] if r["employee"]["id"] == employee.pk)
    assert row["status"] == "balance_only" and Decimal(row["closing_balance"]) == 0
    assert len(row["manual_balance_adjustments"]) == 2
    assert employee.pk not in {
        r["employee"]["id"] for r in build_payroll_period_table(periods[11])["rows"]
    }


def test_later_adjustment_is_reference_only_in_original_closed_period(
    setup, user_factory
):
    manager, employee, periods, _, client = setup
    source = periods[8]
    source.status = "open"
    source.save(update_fields=["status"])
    make_approved_period_inputs(
        period=source, employee=employee, creator=manager, approver=user_factory()
    )
    calculate_period(source.pk, actor=manager)
    source.refresh_from_db()
    source.status = "closed"
    source.save(update_fields=["status"])
    before_table = client.get(url("admin-period-table", pk=source.pk)).json()
    tracked_models = (
        PayrollPeriod,
        PayrollRun,
        PayrollStatement,
        PayrollStatementLine,
        PayrollInputLine,
    )
    before = {
        model: list(model.objects.order_by("pk").values()) for model in tracked_models
    }
    assert (
        create(setup, direction="employee_owes", reason="Доплата за август").status_code
        == 201
    )
    adjustment = PayrollBalanceAdjustment.objects.get()
    for _ in range(2):
        response = client.get(url("admin-period-table", pk=source.pk))
        assert response.status_code == 200, response.content
        after_table = response.json()
        reference = after_table["later_balance_adjustments"][0]
        assert reference["id"] == adjustment.pk
        assert reference["employee"]["id"] == employee.pk
        assert reference["effective_period"]["id"] == periods[10].pk
        assert reference["source_period_id"] == source.pk
        assert reference["contribution"] == "-1234.56"
        assert reference["reason"] == "Доплата за август"
        assert reference["is_voided"] is False
        assert after_table["rows"] == before_table["rows"]
        assert after_table["summary"] == before_table["summary"]
    for model in tracked_models:
        assert list(model.objects.order_by("pk").values()) == before[model], model


def test_reference_is_shown_even_without_historical_employee_row_and_tracks_voiding(
    setup,
):
    _, employee, periods, _, client = setup
    employee.is_active = False
    employee.save(update_fields=["is_active"])
    assert create(setup).status_code == 201
    adjustment = PayrollBalanceAdjustment.objects.get()
    source = build_payroll_period_table(periods[8])
    assert employee.pk not in {row["employee"]["id"] for row in source["rows"]}
    assert source["later_balance_adjustments"][0]["employee"]["id"] == employee.pk
    assert (
        build_payroll_period_table(periods[9])["later_balance_adjustments"][0]["id"]
        == adjustment.pk
    )
    assert not build_payroll_period_table(periods[10])["later_balance_adjustments"]
    assert (
        client.post(
            url("admin-balance-adjustment-void", pk=adjustment.pk), {}, format="json"
        ).status_code
        == 200
    )
    after = build_payroll_period_table(periods[8])
    assert after["later_balance_adjustments"][0]["is_voided"] is True
    assert employee.pk not in {
        row["employee"]["id"] for row in build_payroll_period_table(periods[10])["rows"]
    }
    employee.is_hidden_in_ui = True
    employee.save(update_fields=["is_hidden_in_ui"])
    assert not build_payroll_period_table(periods[8])["later_balance_adjustments"]


def test_balance_only_employee_is_not_added_to_calculation_or_payslips(
    setup, user_factory
):
    manager, former, periods, _, client = setup
    former.is_active = False
    former.save(update_fields=["is_active"])
    worker = user_factory()
    make_approved_period_inputs(
        period=periods[10], employee=worker, creator=manager, approver=user_factory()
    )
    assert create(setup).status_code == 201
    run = calculate_period(periods[10].pk, actor=manager)
    assert set(run.statements.values_list("employee_id", flat=True)) == {worker.pk}
    periods[10].refresh_from_db()
    table = client.get(url("admin-period-table", pk=periods[10].pk)).json()
    row = next(r for r in table["rows"] if r["employee"]["id"] == former.pk)
    assert row["status"] == "balance_only"
    assert row["closing_balance"] == "1234.56"
    assert Decimal(row["payable"]) == 0
    assert row["payment_confirmation"] is None


def test_future_notice_covers_source_and_intermediate_months_without_changing_totals(
    setup,
):
    _, employee, periods, _, _ = setup
    july = PayrollPeriod.objects.create(
        code="2026-07",
        date_from=date(2026, 7, 1),
        date_to=date(2026, 7, 31),
        status="closed",
    )
    before = {
        month: build_payroll_period_table(periods[month]) for month in (8, 9, 10, 11)
    }
    frozen = list(PayrollPeriod.objects.order_by("pk").values())
    assert create(setup, effective_period_id=periods[11].pk).status_code == 201
    adjustment = PayrollBalanceAdjustment.objects.get()
    for month in (8, 9, 10):
        after = build_payroll_period_table(periods[month])
        assert after["rows"] == before[month]["rows"]
        assert after["summary"] == before[month]["summary"]
        reference = after["later_balance_adjustments"][0]
        assert reference["id"] == adjustment.pk
        assert reference["source_period"]["id"] == periods[8].pk
        assert reference["effective_period"]["id"] == periods[11].pk
        assert opening_balances(periods[month], [employee.pk]) == {}
    assert not build_payroll_period_table(july)["later_balance_adjustments"]
    assert not build_payroll_period_table(periods[11])["later_balance_adjustments"]
    assert frozen == list(PayrollPeriod.objects.order_by("pk").values())


def test_no_intermediate_notice_when_source_period_was_not_specified(setup):
    assert create(setup, source_period_id=None).status_code == 201
    for month in (8, 9):
        assert not build_payroll_period_table(setup[2][month])[
            "later_balance_adjustments"
        ]