from __future__ import annotations

from datetime import UTC, datetime
from pathlib import Path
from typing import Any

import pytest

from pingcap_datapipelines.month_end_arr import EXTERNAL_ID_FIELD, TARGET_FIELDS, RowValidationError
from pingcap_datapipelines.nexuscrm_client import UpsertResult
from pingcap_datapipelines.outbound import OutboundRunError, month_window, run_outbound
from pingcap_datapipelines.outbound_config import OutboundConfig


class FakeCRM:
    def __init__(self, *, fail_at: int | None = None, missing_accounts: set[str] | None = None) -> None:
        self.fail_at = fail_at
        self.missing_accounts = missing_accounts or set()
        self.batches: list[list[dict[str, Any]]] = []

    def describe_object(self, _object_name: str) -> dict[str, Any]:
        fields = [{"name": name, "createable": True, "updateable": True} for name in TARGET_FIELDS]
        next(field for field in fields if field["name"] == "Order_Region__c").update(
            {"type": "picklist", "picklistValues": [{"value": "APAC"}]}
        )
        next(field for field in fields if field["name"] == "Product_Type__c").update(
            {"type": "picklist", "picklistValues": [{"value": "Cloud"}]}
        )
        fields.append(
            {"name": EXTERNAL_ID_FIELD, "unique": True, "createable": True, "updateable": True}
        )
        return {"fields": fields}

    def retrieve_collection(
        self, ids: list[str], *, fields: tuple[str, ...] = ("Id",)
    ) -> list[dict[str, Any] | None]:
        assert fields == ("Id",)
        return [
            None if record_id in self.missing_accounts else {"Id": record_id, "attributes": {"type": "Account"}}
            for record_id in ids
        ]

    def resolve_account_ids(self, external_ids: list[str]) -> dict[str, str]:
        return {
            external_id: f"crm-{external_id}"
            for external_id in external_ids
            if external_id not in self.missing_accounts
        }

    def upsert_collection(
        self, *, object_name: str, external_id_field: str, records: list[dict[str, Any]]
    ) -> list[UpsertResult]:
        assert object_name == "Month_End_ARR__c"
        assert external_id_field == EXTERNAL_ID_FIELD
        assert all(str(record["Account__c"]).startswith("crm-") for record in records)
        self.batches.append(records)
        results = []
        for index, _record in enumerate(records):
            absolute_index = sum(len(batch) for batch in self.batches[:-1]) + index
            if self.fail_at == absolute_index:
                results.append(
                    UpsertResult(
                        id="",
                        success=False,
                        created=None,
                        errors=({"statusCode": "FIELD_INTEGRITY_EXCEPTION", "message": "bad account"},),
                    )
                )
            else:
                results.append(UpsertResult(id=f"a0G-{absolute_index}", success=True, created=True, errors=()))
        return results


class FakeState:
    def __init__(self, prior: dict[str, dict[str, Any]] | None = None) -> None:
        self.prior = prior or {}
        self.attempts: list[dict[str, Any]] = []
        self.finished: list[tuple[str, dict[str, Any]]] = []
        self.released = False

    def ensure_tables(self) -> None:
        pass

    def acquire_lease(self, _job_name: str, _run_id: str, _lease_seconds: int) -> None:
        pass

    def release_lease(self, _job_name: str, _run_id: str) -> None:
        self.released = True

    def start_run(self, _job_name: str, _run_id: str) -> None:
        pass

    def finish_run(
        self,
        _job_name: str,
        _run_id: str,
        *,
        status: str,
        metrics: dict[str, Any],
        error_message: str | None = None,
    ) -> None:
        del error_message
        self.finished.append((status, dict(metrics)))

    def latest_record_states(self, _job_name: str) -> dict[str, dict[str, Any]]:
        return self.prior

    def append_attempts(self, rows: list[dict[str, Any]]) -> None:
        self.attempts.extend(rows)


def test_dry_run_skips_blank_accounts_without_target_access() -> None:
    summary = run_outbound(
        _config(),
        dry_run=True,
        source_rows=[_row("001-a"), _row("")],
        run_id="run-1",
    )

    assert summary["status"] == "validated"
    assert summary["valid_rows"] == 1
    assert summary["skipped_blank_account"] == 1


def test_write_batches_and_a_second_run_skips_unchanged_records() -> None:
    crm = FakeCRM()
    state = FakeState()
    first = run_outbound(
        _config(batch_size=1),
        source_rows=[_row("001-a"), _row("001-b")],
        crm_client=crm,
        state=state,
        run_id="run-1",
    )

    assert first["created_rows"] == 2
    assert [len(batch) for batch in crm.batches] == [1, 1]
    assert state.released is True

    prior = {
        attempt["external_key"]: {
            "source_hash": attempt["source_hash"],
            "status": "succeeded",
            "target_id": attempt["target_id"],
        }
        for attempt in state.attempts
    }
    second_crm = FakeCRM()
    second = run_outbound(
        _config(),
        source_rows=[_row("001-a"), _row("001-b")],
        crm_client=second_crm,
        state=FakeState(prior),
        run_id="run-2",
    )

    assert second["unchanged_rows"] == 2
    assert second["attempted_rows"] == 0
    assert second_crm.batches == []


def test_partial_record_failure_is_a_failed_run_with_audited_results() -> None:
    state = FakeState()
    with pytest.raises(OutboundRunError) as captured:
        run_outbound(
            _config(),
            source_rows=[_row("001-a"), _row("001-b")],
            crm_client=FakeCRM(fail_at=1),
            state=state,
            run_id="run-1",
        )

    assert captured.value.summary["failed_rows"] == 1
    assert [attempt["status"] for attempt in state.attempts] == ["succeeded", "failed"]
    assert state.finished[-1][0] == "failed"
    assert state.released is True


def test_duplicate_business_key_fails_before_any_write() -> None:
    with pytest.raises(RowValidationError, match="duplicate business key"):
        run_outbound(
            _config(),
            dry_run=True,
            source_rows=[_row("001-a"), _row("001-a")],
        )


def test_nonexistent_account_fails_before_arr_writes() -> None:
    crm = FakeCRM(missing_accounts={"001-b"})
    with pytest.raises(RowValidationError, match="do not resolve"):
        run_outbound(
            _config(),
            source_rows=[_row("001-a"), _row("001-b")],
            crm_client=crm,
            state=FakeState(),
        )

    assert crm.batches == []


def _row(account_id: str) -> dict[str, str]:
    return {
        "region": "APAC",
        "product_family": "TiDB Cloud",
        "product_type": "Cloud",
        "account_id": account_id,
        "revenue_month": month_window(datetime.now(UTC))[0].isoformat(),
        "month_end_arr_amount": "100.00",
    }


def _config(*, batch_size: int = 200) -> OutboundConfig:
    return OutboundConfig(
        path=Path("configs/jobs/month_end_arr_outbound.yaml"),
        data={
            "job": {"name": "month_end_arr_outbound"},
            "source": {"project": "gcp-digital-bi", "location": "US", "query_file": "unused.sql"},
            "target": {
                "base_url": "https://crm.example",
                "object": "Month_End_ARR__c",
                "external_id_field": EXTERNAL_ID_FIELD,
                "required_fields": list(TARGET_FIELDS),
            },
            "state": {"project": "gcp-digital-bi", "dataset": "outbound_control"},
            "execution": {"batch_size": batch_size},
        },
    )


def test_month_window_is_shanghai_calendar_months_across_year_boundary() -> None:
    from datetime import date

    assert month_window(datetime(2026, 12, 31, 16, tzinfo=UTC)) == (date(2026, 12, 1), date(2027, 2, 1))
    assert month_window(datetime(2026, 8, 31, 15, 59, tzinfo=UTC)) == (date(2026, 7, 1), date(2026, 9, 1))


def test_source_outside_window_fails_before_target_access() -> None:
    crm = FakeCRM()
    old = {**_row("a"), "revenue_month": "2026-07-01"}
    with pytest.raises(RowValidationError, match="outside current/previous"):
        run_outbound(_config(), source_rows=[old], crm_client=crm, state=FakeState(),
                     as_of=datetime(2026, 9, 8, tzinfo=UTC))
    assert crm.batches == []


def test_current_month_absent_does_not_fall_back_to_two_latest_source_months() -> None:
    last_month = {**_row("a"), "revenue_month": "2026-08-01"}
    summary = run_outbound(_config(), dry_run=True, source_rows=[last_month],
                           as_of=datetime(2026, 9, 8, tzinfo=UTC))
    assert summary["valid_rows"] == 1
    assert summary["window_start"] == "2026-08-01"
    assert summary["window_end_exclusive"] == "2026-10-01"


def test_source_query_binds_dates_even_for_limited_probes() -> None:
    from datetime import date
    from unittest.mock import Mock

    from pingcap_datapipelines.outbound import _read_source_rows
    from pingcap_datapipelines.outbound_config import load_outbound_config

    client = Mock()
    config = load_outbound_config("configs/jobs/month_end_arr_outbound.yaml")
    _read_source_rows(client, config, limit=20, window_start=date(2026, 8, 1), window_end=date(2026, 10, 1))
    args, kwargs = client.query.call_args
    assert "@window_start" in args[0] and "@window_end" in args[0]
    assert args[0].endswith("LIMIT 20")
    params = {p.name: p.value for p in kwargs["job_config"].query_parameters}
    assert params == {"window_start": date(2026, 8, 1), "window_end": date(2026, 10, 1)}