from datetime import UTC, datetime, timedelta
from uuid import uuid4

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import select

from adcloud_server.eventing.catalog import EVENTS
from adcloud_server.eventing.tables import contributions, receipts
from adcloud_server.eventing.worker import FeatureWorker, feature_values
from adcloud_server.main import create_app
from adcloud_server.settings import Settings

AUTH = {"Authorization": "Bearer " + "service-" + "x" * 48}
CONFIG = {
    "standard_event": "commerce.cart.add",
    "description": "加购数码商品",
    "rules": [
        {
            "conditions": [
                {
                    "field": "properties.category_id",
                    "operator": "eq",
                    "value": "digital",
                }
            ],
            "effects": [{"feature": "intent.purchase.digital", "value": 1}],
        }
    ],
}


@pytest.fixture(params=["sqlite", "postgres"])
def client(request, tmp_path):
    import os

    url = "sqlite:///" + str(tmp_path / "events.db")
    if request.param == "postgres":
        url = os.getenv("ADCLOUD_TEST_POSTGRES_URL")
        if not url:
            pytest.skip("No PostgreSQL integration URL configured")
    settings = Settings(
        database_url=url,
        tenant_id="registry-" + uuid4().hex,
        identity_secret="identity-" + "x" * 48,
        service_token="service-" + "x" * 48,
        cookie_secure=False,
    )
    with TestClient(create_app(settings)) as test_client:
        yield test_client


def app(client, app_id="shop"):
    response = client.post(
        "/v1/registry/apps", headers=AUTH, json={"app_id": app_id, "name": "测试商店"}
    )
    assert response.status_code == 201, response.text
    return response.json()


def path(application, name="add_cart", version=1):
    return f"/v1/registry/apps/{application['app_id']}/events/{name}/versions/{version}"


def manage(application):
    return {"X-Adcloud-Management-Key": application["management_key"]}


def register(client, application, config=None, name="add_cart"):
    response = client.put(
        path(application, name), json=config or CONFIG, headers=manage(application)
    )
    assert response.status_code == 200, response.text
    response = client.post(
        path(application, name) + "/publish", headers=manage(application)
    )
    assert response.status_code == 200, response.text
    return response.json()


def event(**kwargs):
    return {
        "app_id": "shop",
        "name": "add_cart",
        "event_version": 1,
        "event_id": str(uuid4()),
        "occurred_at": datetime.now(UTC).isoformat(),
        "properties": {"product_id": "camera", "quantity": 1, "category_id": "digital"},
        **kwargs,
    }


def send(client, application, body=None, server=False):
    return client.post(
        "/v1/events",
        headers={
            "X-Adcloud-Key": application["server_key" if server else "collection_key"]
        },
        json=body or event(),
    )


def worker(client):
    return FeatureWorker(client.app.state.identity_service)


def test_registration_preview_requires_category_and_scope(client):
    application = app(client)
    other = app(client, "other")
    assert (
        client.put(path(application), json=CONFIG, headers=manage(other)).status_code
        == 401
    )
    saved = client.put(
        path(application), json=CONFIG, headers=manage(application)
    ).json()
    assert saved["draft"]["required_fields"] == ["category_id"]
    assert send(client, application).status_code == 409
    preview = client.post(
        path(application) + "/preview",
        headers=manage(application),
        json={"properties": event()["properties"]},
    )
    assert preview.json()["effects"][0]["value"] == 1
    bad = event()
    bad["properties"].pop("category_id")
    assert (
        client.post(
            path(application) + "/preview",
            headers=manage(application),
            json={"properties": bad["properties"]},
        ).status_code
        == 422
    )
    register(client, application)
    assert send(client, application, bad).status_code == 422
    assert send(client, application, event(name="unregistered")).status_code == 404
    assert client.post("/v1/events", json={"name": "page.view"}).status_code == 422


def test_receipt_and_worker_are_idempotent_and_exclude_tokens(client):
    application = app(client)
    register(client, application)
    body = event()
    first = send(client, application, body)
    assert first.status_code == 200, first.text
    assert send(client, application, body).json()["duplicate"] is True
    assert worker(client).flush() == 1
    assert worker(client).flush() == 0
    profile = client.get("/v1/users/" + first.json()["master_id"], headers=AUTH).json()
    assert profile["features"] == {"intent.purchase.digital": 1}
    with client.app.state.identity_service.db.transaction() as conn:
        rows = (
            conn.execute(
                select(receipts).where(
                    receipts.c.tenant_id == client.app.state.settings.tenant_id
                )
            )
            .mappings()
            .all()
        )
        assert len(rows) == 1 and "anonymous_token" not in rows[0]["payload"]


def test_server_only_events_and_business_deduplication(client):
    application = app(client)
    register(
        client,
        application,
        {
            "standard_event": "commerce.payment.success",
            "description": "支付成功",
            "rules": [
                {"effects": [{"feature": "commerce.purchase.count", "value": 1}]}
            ],
        },
        "paid",
    )
    body = event(
        name="paid",
        properties={
            "order_id": "o1",
            "payment_id": "p1",
            "amount": 100,
            "currency": "CNY",
        },
    )
    assert send(client, application, body).status_code == 401
    assert send(client, application, body, server=True).status_code == 200
    retry = {**body, "event_id": str(uuid4())}
    assert send(client, application, retry, server=True).json()["duplicate"] is True
    assert worker(client).flush() == 1
    retry["properties"] = {**body["properties"], "amount": 200}
    assert send(client, application, retry, server=True).status_code == 409


def test_frozen_rule_revisions_and_rollback(client):
    application = app(client)
    register(client, application)
    first = send(client, application).json()
    updated = {
        **CONFIG,
        "rules": [
            {
                "conditions": CONFIG["rules"][0]["conditions"],
                "effects": [{"feature": "intent.purchase.digital", "value": 2}],
            }
        ],
    }
    assert (
        client.put(
            path(application), json=updated, headers=manage(application)
        ).status_code
        == 200
    )
    assert (
        client.post(path(application) + "/publish", headers=manage(application)).json()[
            "rule_revision"
        ]
        == 2
    )
    second = send(client, application).json()
    worker(client).flush()
    assert (
        client.get("/v1/users/" + first["master_id"], headers=AUTH).json()["features"][
            "intent.purchase.digital"
        ]
        == 3
    )
    assert first["rule_revision"] == 1 and second["rule_revision"] == 2
    assert (
        client.post(
            path(application) + "/rollback",
            json={"revision": 1},
            headers=manage(application),
        ).status_code
        == 200
    )
    assert send(client, application).json()["rule_revision"] == 1
    assert (
        client.post(
            path(application) + "/disable", headers=manage(application)
        ).status_code
        == 200
    )
    assert send(client, application).status_code == 409


def test_stitch_sums_features_across_anonymous_nodes(client):
    application = app(client)
    register(client, application)
    first = send(client, application).json()
    worker(client).flush()
    client.post(
        "/v1/identity/stitch",
        headers=AUTH,
        json={"identifiers": [{"kind": "email", "value": "alice@example.com"}]},
    )
    client.post("/v1/identity/anonymous?rotate=true")
    second = send(client, application).json()
    worker(client).flush()
    assert first["anonymous_id"] != second["anonymous_id"]
    merged = client.post(
        "/v1/identity/stitch",
        headers=AUTH,
        json={"identifiers": [{"kind": "email", "value": "alice@example.com"}]},
    ).json()
    assert client.get("/v1/users/" + merged["master_id"], headers=AUTH).json()[
        "features"
    ] == {"intent.purchase.digital": 2}
    assert worker(client).flush() == 0


def test_transaction_failure_keeps_task_without_partial_effects(client, monkeypatch):
    application = app(client)
    register(client, application)
    send(client, application)
    identity = client.app.state.identity_service
    original = identity._save_profile

    def fail(*args):
        raise RuntimeError("simulated failure")

    monkeypatch.setattr(identity, "_save_profile", fail)
    with pytest.raises(RuntimeError):
        worker(client).flush()
    with identity.db.transaction() as conn:
        assert (
            conn.execute(
                select(contributions).where(
                    contributions.c.tenant_id == identity.tenant
                )
            ).first()
            is None
        )
        assert (
            conn.execute(
                select(receipts.c.processed).where(
                    receipts.c.tenant_id == identity.tenant
                )
            ).scalar_one()
            == 0
        )
    monkeypatch.setattr(identity, "_save_profile", original)
    assert worker(client).flush() == 1


def test_feature_window_expiry_and_cap(client):
    application = app(client)
    register(
        client,
        application,
        {
            **CONFIG,
            "rules": [
                {"effects": [{"feature": "intent.purchase.digital", "value": 100}]}
            ],
        },
    )
    result = send(client, application).json()
    send(client, application)
    worker(client).flush()
    identity = client.app.state.identity_service
    with identity.db.transaction() as conn:
        assert feature_values(conn, identity.tenant, result["master_id"]) == {
            "intent.purchase.digital": 100
        }
        assert (
            feature_values(
                conn,
                identity.tenant,
                result["master_id"],
                datetime.now(UTC) + timedelta(days=8),
            )
            == {}
        )


@pytest.mark.parametrize("standard", list(EVENTS))
def test_catalog_required_fields_are_enforced(standard):
    from fastapi import HTTPException

    from adcloud_server.eventing.rules import validate_properties

    with pytest.raises(HTTPException) as error:
        validate_properties(
            {"standard_event": standard, "extra_fields": {}, "required_extra": []}, {}
        )
    assert error.value.status_code == 422


@pytest.mark.parametrize("standard", list(EVENTS))
def test_catalog_accepts_valid_required_and_optional_parameters(standard):
    from adcloud_server.eventing.rules import validate_properties

    def sample(spec, name):
        if name == "currency":
            return "CNY"
        if "enum" in spec:
            return spec["enum"][0]
        if spec["type"] == "string":
            return "example"
        if spec["type"] in ("integer", "number"):
            return spec.get("minimum", 0)
        if spec["type"] == "array":
            return [sample(spec["items"], "item")]
        if spec["type"] == "object":
            return {key: sample(value, key) for key, value in spec["fields"].items()}
        raise AssertionError("Unsupported catalogue field")

    properties = {
        key: sample(spec, key) for key, spec in EVENTS[standard]["fields"].items()
    }
    validate_properties(
        {"standard_event": standard, "extra_fields": {}, "required_extra": []},
        properties,
    )


@pytest.mark.parametrize(
    "operator,value,expected",
    [
        ("gte", 1, 1),
        ("gt", 0, 1),
        ("lte", 1, 1),
        ("lt", 2, 1),
        ("eq", 2, 0),
        ("in", [1, 2], 1),
    ],
)
def test_condition_operators(operator, value, expected):
    from adcloud_server.eventing.rules import evaluate

    config = {
        "rules": [
            {
                "conditions": [
                    {
                        "field": "properties.quantity",
                        "operator": operator,
                        "value": value,
                    }
                ],
                "effects": [{"feature": "intent.purchase.digital", "value": 1}],
            }
        ]
    }
    assert len(evaluate(config, {"quantity": 1})["effects"]) == expected


def test_schema_changes_require_new_version_and_retries_use_original_publication(
    client,
):
    application = app(client)
    register(client, application)
    body = event()
    assert send(client, application, body).status_code == 200
    changed = {**CONFIG, "extra_fields": {"variant_id": "string"}}
    assert (
        client.put(
            path(application), headers=manage(application), json=changed
        ).status_code
        == 409
    )
    assert (
        client.put(
            path(application, version=2), headers=manage(application), json=changed
        ).status_code
        == 200
    )
    client.post(path(application) + "/disable", headers=manage(application))
    assert send(client, application, body).json()["duplicate"] is True
    bad = event(
        properties={"product_id": "camera", "quantity": True, "category_id": "digital"}
    )
    client.post(
        path(application) + "/rollback",
        headers=manage(application),
        json={"revision": 1},
    )
    assert send(client, application, bad).status_code == 422


def test_invalid_mapping_and_unregistered_extensions_are_rejected(client):
    application = app(client)
    for config in (
        {**CONFIG, "extra_fields": {"email": "string"}},
        {**CONFIG, "rules": [{"effects": [{"feature": "unknown", "value": 1}]}]},
        {
            **CONFIG,
            "rules": [
                {
                    "conditions": [
                        {"field": "properties.missing", "operator": "eq", "value": "x"}
                    ],
                    "effects": CONFIG["rules"][0]["effects"],
                }
            ],
        },
    ):
        assert (
            client.put(
                path(application), headers=manage(application), json=config
            ).status_code
            == 422
        )
    register(client, application)
    body = event()
    body["properties"]["unregistered"] = "x"
    assert send(client, application, body).status_code == 422
    with client.app.state.identity_service.db.transaction() as conn:
        assert (
            conn.execute(
                select(receipts).where(
                    receipts.c.tenant_id == client.app.state.settings.tenant_id
                )
            ).first()
            is None
        )