"""Optional real PostgreSQL integration: set ADCLOUD_TEST_POSTGRES_URL."""

import os
from concurrent.futures import ThreadPoolExecutor
from uuid import uuid4

import pytest
from sqlalchemy import text

from adcloud_server.database import Database
from adcloud_server.identity_models import EventRequest, ProfilePatch, StitchRequest
from adcloud_server.identity_service import IdentityService
from adcloud_server.settings import Settings

pytestmark = pytest.mark.skipif(
    not os.getenv("ADCLOUD_TEST_POSTGRES_URL"),
    reason="No PostgreSQL integration URL configured",
)


@pytest.fixture
def service():
    settings = Settings(
        database_url=os.environ["ADCLOUD_TEST_POSTGRES_URL"],
        tenant_id="test-" + str(uuid4()),
        identity_secret="postgres-test-" + "x" * 48,
        profile_backend="cassandra",
    )
    db = Database(settings.database_url, settings.tenant_id)
    db.initialize()
    try:
        yield IdentityService(settings, db)
    finally:
        db.close()


def test_postgres_jsonb_stitch_and_immediate_history(service):
    known = service.anonymous()
    identity = StitchRequest(
        identifiers=[{"kind": "email", "value": "alice@example.com"}]
    )
    service.stitch(known["anonymous_token"], identity)
    visitor = service.anonymous(rotate=True)
    service.patch_profile(
        visitor["master_id"],
        ProfilePatch(
            attributes={"nested": {"interest": "running"}, "absent": None},
            tags=["runner"],
        ),
    )
    event = service.record_event(
        visitor["anonymous_token"], EventRequest(name="product.view")
    )
    stitched = service.stitch(visitor["anonymous_token"], identity)
    assert stitched["master_id"] == known["master_id"]
    assert service.profile(known["master_id"])["attributes"] == {
        "nested": {"interest": "running"}
    }
    assert service.history(visitor["master_id"])["items"][0]["id"] == event["event_id"]
    with service.db.transaction() as conn:
        result = conn.execute(
            text(
                "SELECT pg_typeof(document)::text FROM identity_profiles WHERE tenant_id=:tenant LIMIT 1"
            ),
            {"tenant": service.tenant},
        ).scalar_one()
        assert result == "jsonb"


def test_postgres_multiple_process_connections_serialize_identity_merges(service):
    second_db = Database(service.settings.database_url, service.tenant)
    other = IdentityService(service.settings, second_db)
    tokens = [
        service.anonymous()["anonymous_token"],
        other.anonymous()["anonymous_token"],
    ]
    instances = [service, other]
    request = StitchRequest(identifiers=[{"kind": "phone", "value": "+14155552671"}])
    try:
        with ThreadPoolExecutor(max_workers=2) as pool:
            results = list(
                pool.map(lambda i: instances[i].stitch(tokens[i], request), [0, 1])
            )
        assert results[0]["master_id"] == results[1]["master_id"]
        assert len(service.graph(results[0]["master_id"])["identifiers"]) == 1
    finally:
        second_db.close()


def test_postgres_failed_transaction_rolls_back_graph_and_outbox(service, monkeypatch):
    visitor = service.anonymous()

    def fail(*args):
        raise OSError("Simulated outbox failure")

    monkeypatch.setattr(service, "_queue", fail)
    with pytest.raises(OSError):
        service.stitch(
            visitor["anonymous_token"],
            StitchRequest(
                identifiers=[{"kind": "email", "value": "alice@example.com"}]
            ),
        )
    assert service.graph(visitor["master_id"])["identifiers"] == []
    assert service.profile(visitor["master_id"])["revision"] == 1