from concurrent.futures import ThreadPoolExecutor
from uuid import uuid4

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

from adcloud_server.database import Database, identifiers
from adcloud_server.identity_models import StitchRequest
from adcloud_server.identity_service import IdentityService
from adcloud_server.main import create_app
from adcloud_server.settings import Settings

SECRET = "identity-test-secret-" + "x" * 32
SERVICE_TOKEN = "trusted-test-token-" + "y" * 32
AUTH = {"Authorization": "Bearer " + SERVICE_TOKEN}


@pytest.fixture
def settings(tmp_path):
    return Settings(
        database_url="sqlite:///" + str(tmp_path / "identity.db"),
        identity_secret=SECRET,
        service_token=SERVICE_TOKEN,
        cookie_secure=False,
        event_registry_required=False,
    )


@pytest.fixture
def client(settings):
    with TestClient(create_app(settings)) as test_client:
        yield test_client


def anonymous(client):
    response = client.post("/v1/identity/anonymous")
    assert response.status_code == 200, response.text
    return response.json()


def stitch(client, identity, token=None, **kwargs):
    body = {"identifiers": identity, **kwargs}
    if token:
        body["anonymous_token"] = token
    response = client.post("/v1/identity/stitch", json=body, headers=AUTH)
    assert response.status_code == 200, response.text
    return response.json()


def email(value="alice@example.com"):
    return {"kind": "email", "value": value}


def profile(client, master):
    response = client.get("/v1/users/" + master, headers=AUTH)
    assert response.status_code == 200, response.text
    return response.json()


def history(client, master):
    response = client.get("/v1/events", params={"master_id": master}, headers=AUTH)
    assert response.status_code == 200, response.text
    return response.json()


def test_first_event_seeds_persistent_cookie_and_reuses_it(client):
    first = client.post(
        "/v1/events", json={"name": "page.view", "properties": {"path": "/shoes"}}
    )
    assert first.status_code == 200
    assert "HttpOnly" in first.headers["set-cookie"]
    assert "Max-Age=31536000" in first.headers["set-cookie"]
    assert "SameSite=lax" in first.headers["set-cookie"]
    assert first.headers["cache-control"] == "no-store"
    second = anonymous(client)
    assert first.json()["anonymous_id"] == second["anonymous_id"]
    assert profile(client, second["master_id"])["event_count"] == 1


def test_cross_device_stitch_inherits_events_sparse_traits_and_tags(client):
    known = anonymous(client)
    stitch(client, [email()], attributes={"preferred_color": "blue"}, tags=["member"])
    client.post("/v1/identity/anonymous?rotate=true")
    visitor = anonymous(client)
    client.patch(
        "/v1/users/" + visitor["master_id"],
        headers=AUTH,
        json={
            "attributes": {
                "interest": "running",
                "missing": None,
                "nested": {"present": 1, "empty": None},
                "preferred_color": "red",
            },
            "tags": ["runner"],
        },
    )
    event = client.post(
        "/v1/events", json={"name": "product.view", "properties": {"sku": "shoe-1"}}
    ).json()
    result = stitch(client, [email()])
    assert result["master_id"] == known["master_id"]
    assert visitor["master_id"] in result["merged_master_ids"]
    assert result["rotated_anonymous_id"] is False
    merged = profile(client, result["master_id"])
    assert merged["attributes"] == {
        "preferred_color": "blue",
        "interest": "running",
        "nested": {"present": 1},
    }
    assert merged["tags"] == ["member", "runner"]
    assert merged["event_count"] == 1
    assert history(client, visitor["master_id"])["items"][0]["id"] == event["event_id"]
    graph = client.get(
        "/v1/identity/" + result["master_id"] + "/graph", headers=AUTH
    ).json()
    assert len(graph["member_ids"]) == 2
    assert len(graph["edges"]) == 1
    assert graph["identifiers"][0]["display"] == "a***@example.com"


def test_transitive_verified_bridge_resolves_every_alias(client):
    anonymous(client)
    a = stitch(client, [email()])
    client.post("/v1/identity/anonymous?rotate=true")
    phone = {"kind": "phone", "value": "+1 (415) 555-2671"}
    b = stitch(client, [phone])
    client.post("/v1/identity/anonymous?rotate=true")
    oauth = {
        "kind": "oauth",
        "namespace": "https://issuer.example/client-a",
        "value": "subject-123",
    }
    c = stitch(client, [oauth])
    joined = stitch(client, [email(), phone, oauth])
    assert joined["master_id"] == a["master_id"]
    for old in [a, b, c]:
        assert profile(client, old["master_id"])["master_id"] == a["master_id"]
    resolved = client.post(
        "/v1/identity/resolve",
        headers=AUTH,
        json={
            "identifiers": [email(), {"kind": "phone", "value": "+14155552671"}, oauth]
        },
    ).json()
    assert resolved == {
        "master_id": a["master_id"],
        "matches": [a["master_id"]],
        "ambiguous": False,
    }


def test_shared_browser_switches_accounts_without_merging_them(client):
    a = stitch_after_anonymous(client, email())
    client.post("/v1/events", json={"name": "alice.view"})
    b = stitch(client, [email("bob@example.com")])
    assert b["rotated_anonymous_id"] is True
    assert a["master_id"] != b["master_id"]
    assert profile(client, b["master_id"])["event_count"] == 0
    assert profile(client, a["master_id"])["event_count"] == 1


def stitch_after_anonymous(client, identifier):
    anonymous(client)
    return stitch(client, [identifier])


def test_authenticated_backend_can_add_new_identity_to_current_account(client):
    a = stitch_after_anonymous(client, email())
    result = stitch(
        client,
        [{"kind": "phone", "value": "+14155552671"}],
        account_master_id=a["master_id"],
    )
    assert result["master_id"] == a["master_id"]
    assert result["rotated_anonymous_id"] is False
    assert (
        len(
            client.get(
                "/v1/identity/" + a["master_id"] + "/graph", headers=AUTH
            ).json()["identifiers"]
        )
        == 2
    )


def test_namespace_isolates_third_party_subject_ids(client):
    first = {"kind": "oauth", "namespace": "issuer/client-a", "value": "same-subject"}
    second = {"kind": "oauth", "namespace": "issuer/client-b", "value": "same-subject"}
    a = stitch_after_anonymous(client, first)
    client.post("/v1/identity/anonymous?rotate=true")
    b = stitch(client, [second])
    assert a["master_id"] != b["master_id"]
    resolved = client.post(
        "/v1/identity/resolve", headers=AUTH, json={"identifiers": [first, second]}
    ).json()
    assert resolved["ambiguous"] is True
    assert resolved["master_id"] is None


def test_email_domain_normalization_preserves_case_sensitive_local_parts(client):
    first = stitch_after_anonymous(client, email("Alice@EXAMPLE.com"))
    resolved = client.post(
        "/v1/identity/resolve",
        headers=AUTH,
        json={"identifiers": [email("Alice@example.com")]},
    ).json()
    assert resolved["master_id"] == first["master_id"]
    different = client.post(
        "/v1/identity/resolve",
        headers=AUTH,
        json={"identifiers": [email("alice@example.com")]},
    ).json()
    assert different["master_id"] is None


def test_stitch_requires_trusted_backend_and_signed_anonymous_proof(client):
    anon = anonymous(client)
    response = client.post("/v1/identity/stitch", json={"identifiers": [email()]})
    assert response.status_code == 401
    response = client.post(
        "/v1/identity/stitch",
        headers=AUTH,
        json={"identifiers": [email()], "anonymous_token": anon["anonymous_id"]},
    )
    assert response.status_code == 401
    response = client.get("/v1/users/" + anon["master_id"])
    assert response.status_code == 401
    response = client.post(
        "/v1/identity/stitch",
        headers=AUTH,
        json={"identifiers": [{**email(), "verified": True}]},
    )
    assert response.status_code == 422


@pytest.mark.parametrize(
    "identifier",
    [
        email("not-an-email"),
        {"kind": "phone", "value": "4155552671"},
        {"kind": "phone", "value": "+10000000000"},
        {"kind": "oauth", "value": "subject"},
        {"kind": "maid", "value": "device-id"},
    ],
)
def test_invalid_or_weak_identifiers_cannot_create_edges(client, identifier):
    anon = anonymous(client)
    response = client.post(
        "/v1/identity/stitch", headers=AUTH, json={"identifiers": [identifier]}
    )
    assert response.status_code == 422
    assert (
        client.get("/v1/identity/" + anon["master_id"] + "/graph", headers=AUTH).json()[
            "identifiers"
        ]
        == []
    )


def test_browser_origin_and_credentials(client):
    denied = client.post(
        "/v1/events",
        headers={"Origin": "https://untrusted.example"},
        json={"name": "page.view"},
    )
    assert denied.status_code == 403
    preflight = client.options(
        "/v1/events",
        headers={
            "Origin": "http://localhost:5174",
            "Access-Control-Request-Method": "POST",
        },
    )
    assert preflight.status_code == 200
    assert preflight.headers["access-control-allow-credentials"] == "true"
    assert preflight.headers["access-control-allow-origin"] == "http://localhost:5174"


def test_event_replay_is_idempotent_and_cannot_steal_another_subjects_event(client):
    event_id = str(uuid4())
    body = {"event_id": event_id, "name": "page.view"}
    first = client.post("/v1/events", json=body).json()
    replay = client.post("/v1/events", json=body).json()
    assert replay["duplicate"] is True
    assert profile(client, first["master_id"])["event_count"] == 1
    client.post("/v1/identity/anonymous?rotate=true")
    assert client.post("/v1/events", json=body).status_code == 409
    assert (
        client.post(
            "/v1/events",
            json={"name": "page.view", "occurred_at": "2020-01-01T00:00:00"},
        ).status_code
        == 422
    )


def test_profile_patch_omits_nulls_and_deletes_explicitly(client):
    anon = anonymous(client)
    client.patch(
        "/v1/users/" + anon["master_id"],
        headers=AUTH,
        json={"attributes": {"keep": 1, "remove": "later"}},
    )
    result = client.patch(
        "/v1/users/" + anon["master_id"],
        headers=AUTH,
        json={"attributes": {"remove": None}},
    ).json()
    assert result["attributes"] == {"keep": 1}


def test_persistence_survives_restart(settings):
    with TestClient(create_app(settings)) as first:
        anon = anonymous(first)
        first.post("/v1/events", json={"name": "page.view"})
        stitched = stitch(first, [email()])
    with TestClient(create_app(settings)) as second:
        recorded = second.post(
            "/v1/events",
            json={"name": "page.return", "anonymous_token": anon["anonymous_token"]},
        ).json()
        assert recorded["master_id"] == stitched["master_id"]
        assert profile(second, stitched["master_id"])["event_count"] == 2
        service = second.app.state.identity_service
        with service.db.transaction() as conn:
            stored = conn.execute(select(identifiers)).mappings().one()
            assert "alice" not in stored["digest"]
            assert "value" not in stored


def test_tenants_cannot_resolve_or_reuse_each_others_ids(settings):
    other = Settings(
        database_url=settings.database_url,
        tenant_id="other",
        identity_secret=SECRET,
        service_token=SERVICE_TOKEN,
        cookie_secure=False,
        event_registry_required=False,
    )
    with (
        TestClient(create_app(settings)) as first,
        TestClient(create_app(other)) as second,
    ):
        a = stitch_after_anonymous(first, email())
        b = stitch_after_anonymous(second, email())
        assert a["master_id"] != b["master_id"]
        assert (
            second.get("/v1/users/" + a["master_id"], headers=AUTH).status_code == 404
        )
        assert (
            second.post(
                "/v1/events",
                json={"name": "forged", "anonymous_token": a["anonymous_token"]},
            ).status_code
            == 401
        )


def test_parallel_workers_stitch_same_email_into_one_master(settings):
    databases = [Database(settings.database_url, settings.tenant_id) for _ in range(2)]
    for db in databases:
        db.initialize()
    services = [IdentityService(settings, db) for db in databases]
    tokens = [service.anonymous()["anonymous_token"] for service in services]
    request = StitchRequest(identifiers=[email()])
    try:
        with ThreadPoolExecutor(max_workers=2) as pool:
            results = list(
                pool.map(lambda i: services[i].stitch(tokens[i], request), [0, 1])
            )
        assert results[0]["master_id"] == results[1]["master_id"]
        assert len(services[0].graph(results[0]["master_id"])["identifiers"]) == 1
    finally:
        for db in databases:
            db.close()


def test_merge_limit_failure_rolls_back_all_alias_and_profile_changes(client):
    a = stitch_after_anonymous(client, email())
    client.patch(
        "/v1/users/" + a["master_id"],
        headers=AUTH,
        json={"attributes": {"a" + str(i): i for i in range(150)}},
    )
    client.post("/v1/identity/anonymous?rotate=true")
    b = anonymous(client)
    client.patch(
        "/v1/users/" + b["master_id"],
        headers=AUTH,
        json={"attributes": {"b" + str(i): i for i in range(150)}},
    )
    result = client.post(
        "/v1/identity/stitch", headers=AUTH, json={"identifiers": [email()]}
    )
    assert result.status_code == 409
    assert profile(client, b["master_id"])["master_id"] == b["master_id"]
    assert len(profile(client, a["master_id"])["attributes"]) == 150
    assert len(profile(client, b["master_id"])["attributes"]) == 150


def test_missing_configuration_disables_identity_but_keeps_gateway_available(tmp_path):
    with TestClient(
        create_app(Settings(database_url="sqlite:///" + str(tmp_path / "unused.db")))
    ) as client:
        assert client.get("/health").status_code == 200
        assert client.post("/v1/identity/anonymous").status_code == 503
        assert (
            client.post(
                "/v1/identity/stitch", json={"identifiers": [email()]}
            ).status_code
            == 503
        )
        # Advertiser auth initializes storage independently of visitor identity.
        assert (tmp_path / "unused.db").exists()
        assert client.get("/v1/advertiser/me").status_code == 401


def test_secure_cookie_configuration():
    with TestClient(
        create_app(Settings(database_url="sqlite://", identity_secret=SECRET)),
        base_url="https://testserver",
    ) as client:
        result = client.post("/v1/identity/anonymous")
        assert "Secure" in result.headers["set-cookie"]
        assert anonymous(client)["master_id"] == result.json()["master_id"]
    with pytest.raises(ValueError):
        Settings(cookie_secure=False, cookie_samesite="none")
    with pytest.raises(ValueError):
        Settings(profile_backend="unknown")


def test_parallel_workers_initialize_empty_database_safely(settings):
    databases = [Database(settings.database_url, settings.tenant_id) for _ in range(2)]
    try:
        with ThreadPoolExecutor(max_workers=2) as pool:
            list(pool.map(lambda db: db.initialize(), databases))
        assert IdentityService(settings, databases[0]).anonymous()["master_id"]
    finally:
        for db in databases:
            db.close()