"""6J inbound CRM events: verify, attribute, normalize, dedupe, loop protection."""
from __future__ import annotations

import base64
import hashlib
import hmac
from uuid import uuid4

import pytest

from src.domain.connectors.crm_events import (
    DEFERRED_CRM_EVENT_TYPES,
    FROZEN_CRM_EVENT_TYPES,
    REJECTED_CRM_EVENT_TYPES,
    CRMAttributedConnection,
    CRMEventError,
    InMemoryCRMEventDedupe,
    InMemoryCRMOutboundFingerprints,
    RecordingCRMEventSink,
    consumer_event_payload,
    ingest_verified_crm_inbound,
    record_crm_outbound_mutation,
    verify_hubspot_webhook_v3,
)
from src.domain.connectors.crm_resolve import CRMConnectionSnapshot
from src.domain.connectors.invoke_capabilities import (
    CAPABILITY_CONVERSION_PATTERN,
    CRM_FAMILY_PLAN,
    CRM_INBOUND_EVENTS_PHASE_6J,
    INVOKE_CAPABILITIES,
)
from src.domain.connectors.public_gateway import PHASE_6_STATUS, PUBLIC_INVOCABLE_CAPABILITIES
from src.domain.integration_events.catalog import STANDARD_EVENTS


def _conn(provider: str, account: str, **config) -> CRMAttributedConnection:
    cfg = dict(config)
    if provider == "hubspot":
        cfg.setdefault("portal_id", account)
    else:
        cfg.setdefault("org_id", account)
    return CRMAttributedConnection(
        project_id=uuid4(),
        workspace_id=uuid4(),
        connection=CRMConnectionSnapshot(
            connection_id=uuid4(),
            connector_key=provider,
            status="active",
            has_secrets=True,
            permission_mode="allow",
            execution_domain="C",
            config=cfg,
        ),
    )


def test_6j_acceptance_matrix() -> None:
    assert PHASE_6_STATUS == "6p_complete"
    assert CAPABILITY_CONVERSION_PATTERN["phase_6_status"] == "6p_complete"
    assert CAPABILITY_CONVERSION_PATTERN["next_program_slice"] == (
        "9_app_block_modernization"
    )
    assert CRM_FAMILY_PLAN["next_slice"] == "forms_first_class_product_domain"
    assert CRM_FAMILY_PLAN["inbound_is_not_outbound"] is True
    assert CRM_INBOUND_EVENTS_PHASE_6J["hydrate_via_get"] is False
    assert CRM_INBOUND_EVENTS_PHASE_6J["raw_fallback"] is False
    assert CRM_INBOUND_EVENTS_PHASE_6J["salesforce_transport"] == "cdc_not_hubspot_webhook"
    assert CRM_INBOUND_EVENTS_PHASE_6J["delete"] == "DEFER"
    assert CRM_INBOUND_EVENTS_PHASE_6J["association_changed"] == "DEFER"
    assert "crm.object.changed" not in FROZEN_CRM_EVENT_TYPES
    assert "crm.contact.deleted" in DEFERRED_CRM_EVENT_TYPES
    assert "crm.object.changed" in REJECTED_CRM_EVENT_TYPES
    for event_type in FROZEN_CRM_EVENT_TYPES:
        assert event_type not in INVOKE_CAPABILITIES
        assert event_type not in PUBLIC_INVOCABLE_CAPABILITIES
    aliases = {row["alias"] for row in STANDARD_EVENTS}
    for event_type in FROZEN_CRM_EVENT_TYPES:
        assert event_type in aliases


def test_hubspot_signature_valid_and_rejected() -> None:
    secret = "hubspot-secret"
    method = "POST"
    url = "https://cloud.example/v1/cloud/webhooks/hubspot"
    body = '[{"eventId":1,"subscriptionType":"contact.creation","objectId":9,"portalId":111}]'
    ts = "1710000000000"
    expected = base64.b64encode(
        hmac.new(secret.encode(), f"{method}{url}{body}{ts}".encode(), hashlib.sha256).digest()
    ).decode()
    verify_hubspot_webhook_v3(
        client_secret=secret,
        method=method,
        url=url,
        raw_body=body,
        signature=expected,
        timestamp=ts,
        now_ms=1710000000000,
    )
    with pytest.raises(CRMEventError) as bad:
        verify_hubspot_webhook_v3(
            client_secret=secret,
            method=method,
            url=url,
            raw_body=body,
            signature="nope",
            timestamp=ts,
            now_ms=1710000000000,
        )
    assert bad.value.code == "crm_inbound_unverified"
    with pytest.raises(CRMEventError) as stale:
        verify_hubspot_webhook_v3(
            client_secret=secret,
            method=method,
            url=url,
            raw_body=body,
            signature=expected,
            timestamp=ts,
            now_ms=1710000000000 + 6 * 60 * 1000,
        )
    assert stale.value.code == "crm_inbound_unverified"


def test_unverified_batch_rejected() -> None:
    hs = _conn("hubspot", "111")
    with pytest.raises(CRMEventError) as err:
        ingest_verified_crm_inbound(
            provider="hubspot",
            verified=False,
            payload=[{"eventId": 1, "subscriptionType": "contact.creation", "objectId": 9, "portalId": 111}],
            connections=[hs],
        )
    assert err.value.code == "crm_inbound_unverified"


def test_attribution_exact_connection_not_first_hubspot() -> None:
    other = _conn("hubspot", "999")
    match = _conn("hubspot", "111")
    sink = RecordingCRMEventSink()
    result = ingest_verified_crm_inbound(
        provider="hubspot",
        verified=True,
        payload=[{
            "eventId": 88,
            "subscriptionType": "contact.creation",
            "objectId": 9,
            "portalId": 111,
        }],
        connections=[other, match],
        sink=sink,
        dedupe=InMemoryCRMEventDedupe(),
    )
    assert len(result.accepted) == 1
    assert result.accepted[0].connection_id == str(match.connection.connection_id)
    assert result.accepted[0].project_id == str(match.project_id)


def test_unmapped_and_ambiguous_org() -> None:
    a = _conn("salesforce", "00Daaa")
    b = _conn("salesforce", "00Daaa")
    with pytest.raises(CRMEventError) as missing:
        ingest_verified_crm_inbound(
            provider="salesforce",
            verified=True,
            payload={"ChangeEventHeader": {
                "entityName": "Contact",
                "changeType": "CREATE",
                "recordIds": ["003aaa"],
                "transactionKey": "tx1",
                "organizationId": "00Dzzz",
            }},
            connections=[a],
            provider_account_id="00Dzzz",
            dedupe=InMemoryCRMEventDedupe(),
        )
    assert missing.value.code == "crm_connection_unmapped"
    with pytest.raises(CRMEventError) as amb:
        ingest_verified_crm_inbound(
            provider="salesforce",
            verified=True,
            payload={"ChangeEventHeader": {
                "entityName": "Contact",
                "changeType": "CREATE",
                "recordIds": ["003aaa"],
                "transactionKey": "tx2",
                "organizationId": "00Daaa",
            }},
            connections=[a, b],
            provider_account_id="00Daaa",
            dedupe=InMemoryCRMEventDedupe(),
        )
    assert amb.value.code == "crm_attribution_ambiguous"


def test_created_updated_normalization_and_stage_reverse() -> None:
    hs = _conn(
        "hubspot",
        "111",
        stage_map={"qualified": "appointmentscheduled", "won": "closedwon"},
        pipeline_map={"sales": "default"},
    )
    sf = _conn("salesforce", "00Daaa", stage_map={"qualified": "Qualification", "won": "Closed Won"})
    hs_result = ingest_verified_crm_inbound(
        provider="hubspot",
        verified=True,
        payload=[
            {
                "eventId": 1,
                "subscriptionType": "deal.propertyChange",
                "objectId": 55,
                "portalId": 111,
                "propertyName": "dealstage",
                "propertyValue": "appointmentscheduled",
            },
            {
                "eventId": 2,
                "subscriptionType": "contact.propertyChange",
                "objectId": 9,
                "portalId": 111,
                "propertyName": "email",
                "propertyValue": "a@b.com",
            },
        ],
        connections=[hs],
        dedupe=InMemoryCRMEventDedupe(),
        sink=RecordingCRMEventSink(),
    )
    deal = next(e for e in hs_result.accepted if e.event.event_type == "crm.deal.updated")
    assert deal.event.object_ref.object_type == "deal"
    assert deal.event.changed_fields == ["stage"]
    assert deal.event.object == {"stage": "qualified"}
    contact = next(e for e in hs_result.accepted if e.event.event_type == "crm.contact.updated")
    assert contact.event.changed_fields == ["email"]
    assert "firstname" not in str(contact.event.object)

    sf_result = ingest_verified_crm_inbound(
        provider="salesforce",
        verified=True,
        payload={
            "ChangeEventHeader": {
                "entityName": "Opportunity",
                "changeType": "UPDATE",
                "recordIds": ["006aaa"],
                "changedFields": ["StageName", "Amount"],
                "transactionKey": "tx-opp",
                "organizationId": "00Daaa",
            },
            "StageName": "Qualification",
            "Amount": 1200,
            "Pipeline": "should-not-leak",
        },
        connections=[sf],
        provider_account_id="00Daaa",
        dedupe=InMemoryCRMEventDedupe(),
    )
    opp = sf_result.accepted[0]
    assert opp.event.event_type == "crm.deal.updated"
    assert opp.event.object["stage"] == "qualified"
    assert "pipeline" not in (opp.event.object or {})
    assert "StageName" not in (opp.event.object or {})
    assert "amount" in opp.event.changed_fields
    public = consumer_event_payload(opp)
    assert "connection_id" not in public
    assert public["project_id"] == str(sf.project_id)


def test_unknown_stage_omitted() -> None:
    hs = _conn("hubspot", "111", stage_map={"won": "closedwon"})
    result = ingest_verified_crm_inbound(
        provider="hubspot",
        verified=True,
        payload=[{
            "eventId": 3,
            "subscriptionType": "deal.propertyChange",
            "objectId": 55,
            "portalId": 111,
            "propertyName": "dealstage",
            "propertyValue": "legalreview",
        }],
        connections=[hs],
        dedupe=InMemoryCRMEventDedupe(),
    )
    assert result.accepted[0].event.object is None
    assert result.accepted[0].event.changed_fields == []


def test_salesforce_one_event_per_record_id() -> None:
    sf = _conn("salesforce", "00Daaa")
    result = ingest_verified_crm_inbound(
        provider="salesforce",
        verified=True,
        payload={
            "ChangeEventHeader": {
                "entityName": "Account",
                "changeType": "CREATE",
                "recordIds": ["001aaa", "001bbb"],
                "transactionKey": "tx-acc",
                "organizationId": "00Daaa",
            },
            "Name": "Acme",
        },
        connections=[sf],
        provider_account_id="00Daaa",
        dedupe=InMemoryCRMEventDedupe(),
    )
    assert len(result.accepted) == 2
    assert {e.event.object_ref.external_id for e in result.accepted} == {"001aaa", "001bbb"}
    assert all(e.event.event_type == "crm.company.created" for e in result.accepted)


def test_delete_and_association_deferred() -> None:
    hs = _conn("hubspot", "111")
    result = ingest_verified_crm_inbound(
        provider="hubspot",
        verified=True,
        payload=[
            {"eventId": 10, "subscriptionType": "contact.deletion", "objectId": 1, "portalId": 111},
            {"eventId": 11, "subscriptionType": "contact.associationChange", "objectId": 1, "portalId": 111},
        ],
        connections=[hs],
        dedupe=InMemoryCRMEventDedupe(),
    )
    assert result.accepted == []
    assert result.ignored == ["10", "11"]


def test_dedupe_same_provider_event_once() -> None:
    hs = _conn("hubspot", "111")
    dedupe = InMemoryCRMEventDedupe()
    payload = [{
        "eventId": 44,
        "subscriptionType": "company.creation",
        "objectId": 7,
        "portalId": 111,
    }]
    first = ingest_verified_crm_inbound(
        provider="hubspot", verified=True, payload=payload, connections=[hs], dedupe=dedupe
    )
    second = ingest_verified_crm_inbound(
        provider="hubspot", verified=True, payload=payload, connections=[hs], dedupe=dedupe
    )
    assert len(first.accepted) == 1
    assert second.accepted == []
    assert second.duplicates == ["44"]
    other = ingest_verified_crm_inbound(
        provider="hubspot",
        verified=True,
        payload=[{
            "eventId": 45,
            "subscriptionType": "company.propertyChange",
            "objectId": 7,
            "portalId": 111,
            "propertyName": "name",
            "propertyValue": "Acme",
        }],
        connections=[hs],
        dedupe=dedupe,
    )
    assert len(other.accepted) == 1


def test_loop_protection_narrow_not_object_wide() -> None:
    hs = _conn("hubspot", "111", app_id="99", stage_map={"qualified": "appointmentscheduled"})
    prints = InMemoryCRMOutboundFingerprints()
    record_crm_outbound_mutation(
        connection_id=hs.connection.connection_id,
        object_type="deal",
        external_id="55",
        fields={"stage"},
        fingerprints=prints,
    )
    echo = ingest_verified_crm_inbound(
        provider="hubspot",
        verified=True,
        payload=[{
            "eventId": 70,
            "subscriptionType": "deal.propertyChange",
            "objectId": 55,
            "portalId": 111,
            "propertyName": "dealstage",
            "propertyValue": "appointmentscheduled",
        }],
        connections=[hs],
        dedupe=InMemoryCRMEventDedupe(),
        fingerprints=prints,
    )
    assert echo.suppressed == ["70"]
    external = ingest_verified_crm_inbound(
        provider="hubspot",
        verified=True,
        payload=[{
            "eventId": 71,
            "subscriptionType": "deal.propertyChange",
            "objectId": 55,
            "portalId": 111,
            "propertyName": "amount",
            "propertyValue": "500",
        }],
        connections=[hs],
        dedupe=InMemoryCRMEventDedupe(),
        fingerprints=prints,
    )
    assert len(external.accepted) == 1
    app_echo = ingest_verified_crm_inbound(
        provider="hubspot",
        verified=True,
        payload=[{
            "eventId": 72,
            "subscriptionType": "contact.creation",
            "objectId": 3,
            "portalId": 111,
            "appId": 99,
            "changeSource": "INTEGRATION",
        }],
        connections=[hs],
        dedupe=InMemoryCRMEventDedupe(),
        fingerprints=prints,
    )
    assert app_echo.suppressed == ["72"]


def test_same_alias_maps_per_connection_on_inbound() -> None:
    a = _conn("hubspot", "111", stage_map={"qualified": "appointmentscheduled"})
    b = _conn("hubspot", "222", stage_map={"qualified": "123456"})
    first = ingest_verified_crm_inbound(
        provider="hubspot",
        verified=True,
        payload=[{
            "eventId": 1,
            "subscriptionType": "deal.propertyChange",
            "objectId": 1,
            "portalId": 111,
            "propertyName": "dealstage",
            "propertyValue": "appointmentscheduled",
        }],
        connections=[a, b],
        dedupe=InMemoryCRMEventDedupe(),
    )
    second = ingest_verified_crm_inbound(
        provider="hubspot",
        verified=True,
        payload=[{
            "eventId": 2,
            "subscriptionType": "deal.propertyChange",
            "objectId": 1,
            "portalId": 222,
            "propertyName": "dealstage",
            "propertyValue": "123456",
        }],
        connections=[a, b],
        dedupe=InMemoryCRMEventDedupe(),
    )
    assert first.accepted[0].event.object["stage"] == "qualified"
    assert second.accepted[0].event.object["stage"] == "qualified"
    assert first.accepted[0].connection_id != second.accepted[0].connection_id


def test_p_resources_never_attributed() -> None:
    fake_p = CRMAttributedConnection(
        project_id=uuid4(),
        workspace_id=uuid4(),
        connection=CRMConnectionSnapshot(
            connection_id=uuid4(),
            connector_key="hubspot",
            status="active",
            has_secrets=False,
            permission_mode="allow",
            execution_domain="P",
            config={"portal_id": "111"},
        ),
    )
    with pytest.raises(CRMEventError) as err:
        ingest_verified_crm_inbound(
            provider="hubspot",
            verified=True,
            payload=[{
                "eventId": 1,
                "subscriptionType": "contact.creation",
                "objectId": 9,
                "portalId": 111,
            }],
            connections=[fake_p],
            dedupe=InMemoryCRMEventDedupe(),
        )
    assert err.value.code == "crm_connection_unmapped"