from __future__ import annotations

import hashlib
import json
from dataclasses import asdict
from pathlib import Path
from typing import Any

import pytest

from pingcap_datapipelines.netsuite_revenue_admission import AdmittedRow
from pingcap_datapipelines.netsuite_revenue_candidate import CandidateManifest, CandidateRow, LocalCandidateStore
from pingcap_datapipelines.netsuite_revenue_config import load_revenue_sync_config
from pingcap_datapipelines.netsuite_revenue_nexuscrm import (
    APIWritePlan,
    NexusCRMRevenueContractError,
    build_write_plan,
    describe_sha256,
    validate_target_contract,
    verify_active_target_parity,
    write_records,
)
from pingcap_datapipelines.netsuite_revenue_sync import RevenueSyncGateError, run_revenue_sync
from pingcap_datapipelines.nexuscrm_client import UpsertResult

CONFIG_PATH = Path("configs/jobs/netsuite_revenue_to_nexuscrm.yaml")


def _config():
    return load_revenue_sync_config(CONFIG_PATH)


def _candidate(config, *, source_key: str = "7429932:100:200", account_id: str = "A-001") -> CandidateRow:
    payload = {mapping.field: f"value:{mapping.field}" for mapping in config.mappings}
    payload[config.account_id_mapping.field] = account_id
    payload[next(mapping.field for mapping in config.mappings if mapping.header == "Transaction Amount")] = "12.50"
    payload[next(mapping.field for mapping in config.mappings if mapping.header == "Amount")] = "10.00"
    payload_hash = hashlib.sha256(json.dumps(payload, sort_keys=True).encode()).hexdigest()
    return CandidateRow(
        source_key=source_key,
        source_identity={"transactionInternalId": "100", "lineUniqueKey": "200", "accountingBookId": "1"},
        payload=payload,
        payload_hash=payload_hash,
        raw_account_id=account_id,
        display_account_id=account_id,
        admission_evidence_available=True,
    )


def _describe(config) -> dict[str, Any]:
    fields: list[dict[str, Any]] = []

    def add(name: str, field_type: str, *, nillable: bool = False, unique: bool = False) -> None:
        fields.append(
            {
                "name": name,
                "type": field_type,
                "nillable": nillable,
                "unique": unique,
                "createable": True,
                "updateable": True,
            }
        )

    add(config.target["name_field"], "string")
    for mapping in config.mappings:
        field_type = {
            "text": "string",
            "currency": "currency",
            "number": "double",
            "date": "date",
            "datetime": "datetime",
        }[mapping.kind]
        add(mapping.field, field_type)
    add(config.target["external_key_field"], "string", unique=True)
    add(config.target["account"]["relation_field"], "reference")
    for control, name in config.target["controls"].items():
        field_type = "datetime" if control.endswith("_at") else "boolean" if control == "is_source_active" else "string"
        add(name, field_type, nillable=control == "accounting_book_id")
    return {"name": config.target["object"], "fields": fields}


def test_describe_requires_the_51_field_successor_contract() -> None:
    config = _config()
    describe = _describe(config)
    validate_target_contract(config, describe)

    assert len(describe["fields"]) == 51
    relation = next(field for field in describe["fields"] if field["name"] == "CRM_Account__c")
    relation["nillable"] = True
    with pytest.raises(NexusCRMRevenueContractError, match="metadata-required"):
        validate_target_contract(config, describe)


def test_payloads_are_api_whitelisted_and_preserve_first_seen_on_update() -> None:
    config = _config()
    candidate = _candidate(config)
    admitted = AdmittedRow(candidate, "001-account")
    new_plan = build_write_plan(
        config,
        [admitted],
        existing_by_source_key={},
        run_id="run-1",
        posting_period_id="245",
        seen_at="2026-08-20T00:00:00+00:00",
    )
    new_payload = new_plan.records[0]
    controls = config.target["controls"]
    assert new_payload[config.target["account"]["relation_field"]] == "001-account"
    assert new_payload[controls["first_seen_run_id"]] == "run-1"
    assert new_payload[controls["is_source_active"]] is True
    assert not set(new_payload) & set(config.target["protected_system_fields"])

    unchanged_plan = build_write_plan(
        config,
        [admitted],
        existing_by_source_key={
            candidate.source_key: {
                controls["payload_hash"]: candidate.payload_hash,
                config.target["account"]["relation_field"]: "001-account",
                config.target["active_flag_field"]: True,
            }
        },
        run_id="run-2",
        posting_period_id="245",
        seen_at="2026-08-20T01:00:00+00:00",
    )
    unchanged = unchanged_plan.records[0]
    assert unchanged_plan.last_seen_only == 1
    assert controls["first_seen_run_id"] not in unchanged
    assert config.target["account"]["relation_field"] not in unchanged
    assert not any(mapping.field in unchanged for mapping in config.mappings)


def test_relation_drift_and_bad_describe_type_fail_closed() -> None:
    config = _config()
    candidate = _candidate(config)
    with pytest.raises(NexusCRMRevenueContractError, match="differs from exact"):
        build_write_plan(
            config,
            [AdmittedRow(candidate, "001-account")],
            existing_by_source_key={
                candidate.source_key: {
                    config.target["account"]["relation_field"]: "different-account",
                    config.target["active_flag_field"]: True,
                }
            },
            run_id="run-1",
            posting_period_id="245",
        )
    describe = _describe(config)
    exchange = next(field for field in describe["fields"] if field["name"] == "Exchange_Rate__c")
    exchange["type"] = "currency"
    with pytest.raises(NexusCRMRevenueContractError, match="type contract drift"):
        validate_target_contract(config, describe)


def test_write_batches_are_at_most_200_and_failures_leave_an_attempt_ledger(tmp_path: Path) -> None:
    config = _config()

    class Client:
        def __init__(self, fail: bool = False) -> None:
            self.batches: list[list[dict[str, Any]]] = []
            self.fail = fail

        def upsert_collection(self, **kwargs: Any) -> list[UpsertResult]:
            records = kwargs["records"]
            self.batches.append(records)
            return [
                UpsertResult(id="id", success=not (self.fail and index == 0), created=True, errors=())
                for index, _ in enumerate(records)
            ]

    records = tuple(
        {
            config.target["external_key_field"]: f"key-{index}",
            config.target["controls"]["payload_hash"]: "a" * 64,
        }
        for index in range(201)
    )
    store = LocalCandidateStore(tmp_path, run_id="batch")
    client = Client()
    result = write_records(config, client, store, APIWritePlan(records, 201, 0), attempt_prefix="attempt")
    assert [len(batch) for batch in client.batches] == [200, 1]
    assert result.attempted == 201

    failed_store = LocalCandidateStore(tmp_path, run_id="failed")
    with pytest.raises(NexusCRMRevenueContractError, match="rejected"):
        write_records(
            config,
            Client(fail=True),
            failed_store,
            APIWritePlan(records[:1], 1, 0),
            attempt_prefix="attempt",
        )
    ledger = (failed_store.root / "attempt/records-api-attempt-ledger.jsonl").read_text()
    assert '"success":false' in ledger


def test_api_readback_parity_requires_active_exact_account_and_amounts() -> None:
    config = _config()
    candidate = _candidate(config)
    admitted = [AdmittedRow(candidate, "001-account")]
    record = {
        config.target["active_flag_field"]: True,
        config.target["controls"]["payload_hash"]: candidate.payload_hash,
        config.target["account"]["relation_field"]: "001-account",
        "Transaction_Amount__c": "12.50",
        "Base_Amount__c": "10.00",
    }
    result = verify_active_target_parity(
        config, admitted=admitted, existing_by_source_key={candidate.source_key: record}
    )
    assert result["target_active_count"] == 1
    record[config.target["account"]["relation_field"]] = None
    with pytest.raises(NexusCRMRevenueContractError, match="CRM_Account__c parity"):
        verify_active_target_parity(config, admitted=admitted, existing_by_source_key={candidate.source_key: record})


def test_dry_run_never_builds_a_crm_client_and_commit_gates_precede_connection(monkeypatch: pytest.MonkeyPatch) -> None:
    config = _config()
    row = _candidate(config)
    manifest = CandidateManifest(
        run_id="run-1",
        posting_period_id="245",
        source_total=1,
        source_hash="b" * 64,
        page_count=1,
        headers=tuple(mapping.header for mapping in config.mappings),
        control_totals={},
        fixture=False,
        created_at="2026-08-20T00:00:00+00:00",
        object_prefix="local",
    )

    class Store:
        prefix = "local"

    monkeypatch.setattr("pingcap_datapipelines.netsuite_revenue_sync.build_store", lambda *_args, **_kwargs: Store())
    monkeypatch.setattr(
        "pingcap_datapipelines.netsuite_revenue_sync.freeze_source_candidate",
        lambda *_args, **_kwargs: ([row], manifest),
    )
    monkeypatch.setattr(
        "pingcap_datapipelines.netsuite_revenue_sync._build_crm_client",
        lambda *_args: pytest.fail("dry-run attempted CRM authentication"),
    )
    result = run_revenue_sync(
        config,
        run_id="run-1",
        posting_period_id="245",
        commit=False,
        validate_only=False,
        expected_project=None,
        expected_environment=None,
        expected_target_base_url=None,
        expected_api_version=None,
        expected_describe_sha256=None,
        expected_manifest_sha256=None,
        candidate_run_id=None,
    )
    assert result["status"] == "candidate_frozen"

    with pytest.raises(RevenueSyncGateError, match="expected manifest"):
        run_revenue_sync(
            config,
            run_id="run-1",
            posting_period_id="245",
            commit=True,
            validate_only=False,
            expected_project="gcp-digital-bi",
            expected_environment="production",
            expected_target_base_url="https://crm.example",
            expected_api_version="v63.0",
            expected_describe_sha256="a" * 64,
            expected_manifest_sha256=None,
            candidate_run_id="run-1",
        )


def test_commit_uses_composite_api_and_api_readback_only(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    config = _config()
    candidate = _candidate(config)
    manifest = CandidateManifest(
        run_id="run-commit",
        posting_period_id="245",
        source_total=1,
        source_hash="c" * 64,
        page_count=1,
        headers=tuple(mapping.header for mapping in config.mappings),
        control_totals={},
        fixture=False,
        created_at="2026-08-20T00:00:00+00:00",
        object_prefix="local",
    )
    store = LocalCandidateStore(tmp_path, run_id="run-commit")
    describe = _describe(config)

    class Client:
        def __init__(self) -> None:
            self.targets: dict[str, dict[str, Any]] = {}
            self.upsert_calls: list[tuple[str, str, list[dict[str, Any]]]] = []

        def describe_object(self, object_name: str) -> dict[str, Any]:
            assert object_name == config.target["object"]
            return describe

        def resolve_active_account_matches(self, external_ids: list[str]) -> dict[str, list[str]]:
            return {value: ["001-account"] if value == "A-001" else [] for value in external_ids}

        def query_records(self, _soql: str) -> list[dict[str, Any]]:
            return list(self.targets.values())

        def upsert_collection(
            self, *, object_name: str, external_id_field: str, records: list[dict[str, Any]]
        ) -> list[UpsertResult]:
            self.upsert_calls.append((object_name, external_id_field, records))
            result: list[UpsertResult] = []
            for record in records:
                key = record[external_id_field]
                created = key not in self.targets
                self.targets[key] = {**self.targets.get(key, {}), **record}
                result.append(UpsertResult(id=f"id-{key}", success=True, created=created, errors=()))
            return result

    client = Client()
    monkeypatch.setenv("NEXUSCRM_REVENUE_BASE_URL", "https://crm.example")
    monkeypatch.setattr("pingcap_datapipelines.netsuite_revenue_sync.build_store", lambda *_args, **_kwargs: store)
    monkeypatch.setattr(
        "pingcap_datapipelines.netsuite_revenue_sync.freeze_source_candidate",
        lambda *_args, **_kwargs: ([candidate], manifest),
    )
    result = run_revenue_sync(
        config,
        run_id="run-commit",
        posting_period_id="245",
        commit=True,
        validate_only=False,
        expected_project="gcp-digital-bi",
        expected_environment="production",
        expected_target_base_url="https://crm.example",
        expected_api_version="v63.0",
        expected_describe_sha256=describe_sha256(describe),
        expected_manifest_sha256=hashlib.sha256(
            json.dumps(asdict(manifest), sort_keys=True, separators=(",", ":"), default=str).encode()
        ).hexdigest(),
        candidate_run_id="run-commit",
        crm_client=client,
    )

    assert result["status"] == "committed"
    assert client.upsert_calls[0][0] == "NetSuite_Revenue_Voucher_Line__c"
    assert client.upsert_calls[0][1] == "NetSuite_Source_Key__c"
    assert list(store.root.rglob("records-api-attempt-ledger.jsonl"))


def test_direct_database_variant_is_absent() -> None:
    root = Path(__file__).parents[1]
    assert not (root / "src/pingcap_datapipelines/nexuscrm_tidb_direct.py").exists()
    assert not (root / "terraform/netsuite_revenue_direct.tf").exists()
    assert not (root / "configs/jobs/netsuite_revenue_to_crm_tidb.yaml").exists()
    revenue_sources = "\n".join(
        (root / "src/pingcap_datapipelines/netsuite_revenue_sync.py").read_text().splitlines()
    )
    assert "TiDB" not in revenue_sources and "pymysql" not in revenue_sources