"""Inbound CRM event plane. Not capability invocation. Provider fact → verified transport → connection attribution → canonical CRMEvent → dedupe / self-originated suppression → project event sink. Does not call crm.* invoke verbs or automation.workflow.trigger. """ from __future__ import annotations import base64 import hashlib import hmac import json import time from dataclasses import dataclass, field from datetime import UTC, datetime from typing import Any, Literal, Protocol from uuid import UUID, uuid4 from pydantic import BaseModel, ConfigDict, Field, StrictStr from src.domain.connectors.crm_canonical import ( CRM_FIELD_MAP, CRMObjectRef, ) from src.domain.connectors.crm_resolve import CRMConnectionSnapshot from src.domain.connectors.crm_resources import ( pipeline_map_from_config, reverse_alias, stage_map_from_config, ) CRMEventType = Literal[ "crm.contact.created", "crm.contact.updated", "crm.company.created", "crm.company.updated", "crm.deal.created", "crm.deal.updated", ] FROZEN_CRM_EVENT_TYPES: tuple[CRMEventType, ...] = ( "crm.contact.created", "crm.contact.updated", "crm.company.created", "crm.company.updated", "crm.deal.created", "crm.deal.updated", ) DEFERRED_CRM_EVENT_TYPES: tuple[str, ...] = ( "crm.contact.deleted", "crm.company.deleted", "crm.deal.deleted", "crm.contact.restored", "crm.company.restored", "crm.deal.restored", "crm.association.changed", ) REJECTED_CRM_EVENT_TYPES: tuple[str, ...] = ( "crm.object.changed", "crm.lead.created", "crm.lead.updated", ) HUBSPOT_TIMESTAMP_TOLERANCE_MS = 5 * 60 * 1000 HUBSPOT_SELF_CHANGE_SOURCES = frozenset({"INTEGRATION", "AUTOMATION_PLATFORM"}) _PROVIDER_TO_CANONICAL: dict[tuple[str, str, str], str] = {} for _row in CRM_FIELD_MAP: _field = str(_row["field"]).split(".")[-1] _object = str(_row["field"]).split(".")[0] _object_type = { "CRMContact": "contact", "CRMCompany": "company", "CRMDeal": "deal", }[_object] if _row.get("hubspot"): _PROVIDER_TO_CANONICAL[("hubspot", _object_type, str(_row["hubspot"]).lower())] = _field if _row.get("salesforce"): _PROVIDER_TO_CANONICAL[("salesforce", _object_type, str(_row["salesforce"]).lower())] = ( _field ) class CRMEventError(ValueError): def __init__(self, code: str, message: str = "") -> None: self.code = code super().__init__(message or code) class CRMEvent(BaseModel): """Application-facing canonical envelope. No provider wire names or secrets.""" model_config = ConfigDict(extra="forbid") event_id: StrictStr event_type: CRMEventType occurred_at: StrictStr | None = None object_ref: CRMObjectRef changed_fields: list[str] = Field(default_factory=list) object: dict[str, Any] | None = None class CRMEventDispatch(BaseModel): """Internal dispatch record. connection_id is stripped from generic consumers.""" model_config = ConfigDict(extra="forbid") project_id: StrictStr workspace_id: StrictStr | None = None connection_id: StrictStr provider_event_id: StrictStr event: CRMEvent @dataclass(frozen=True) class CRMAttributedConnection: project_id: UUID workspace_id: UUID | None connection: CRMConnectionSnapshot @dataclass class CRMInboundDraft: provider: str provider_event_id: str event_type: str object_type: str | None external_id: str | None occurred_at: str | None provider_fields: dict[str, Any] = field(default_factory=dict) provider_changed: list[str] = field(default_factory=list) provider_account_id: str | None = None app_id: str | None = None change_source: str | None = None change_origin: str | None = None class CRMEventDedupePort(Protocol): def claim(self, key: str) -> bool: """True if this key has not been accepted before.""" def put(self, key: str, dispatch: CRMEventDispatch) -> None: ... def get(self, key: str) -> CRMEventDispatch | None: ... class CRMOutboundFingerprintPort(Protocol): def remember( self, connection_id: str, object_type: str, external_id: str, fields: frozenset[str], ) -> None: ... def is_self_originated( self, connection_id: str, object_type: str, external_id: str, fields: frozenset[str], ) -> bool: ... class CRMEventSink(Protocol): def emit(self, dispatch: CRMEventDispatch) -> None: ... class InMemoryCRMEventDedupe: def __init__(self) -> None: self._seen: set[str] = set() self._payloads: dict[str, CRMEventDispatch] = {} def claim(self, key: str) -> bool: if key in self._seen: return False self._seen.add(key) return True def put(self, key: str, dispatch: CRMEventDispatch) -> None: self._payloads[key] = dispatch def get(self, key: str) -> CRMEventDispatch | None: return self._payloads.get(key) class InMemoryCRMOutboundFingerprints: def __init__(self) -> None: self._rows: dict[tuple[str, str, str], frozenset[str]] = {} def remember( self, connection_id: str, object_type: str, external_id: str, fields: frozenset[str], ) -> None: self._rows[(connection_id, object_type, external_id)] = fields def is_self_originated( self, connection_id: str, object_type: str, external_id: str, fields: frozenset[str], ) -> bool: remembered = self._rows.get((connection_id, object_type, external_id)) if not remembered: return False if not fields: return True return fields <= remembered class RecordingCRMEventSink: def __init__(self) -> None: self.events: list[CRMEventDispatch] = [] def emit(self, dispatch: CRMEventDispatch) -> None: self.events.append(dispatch) DEFAULT_DEDUPE = InMemoryCRMEventDedupe() DEFAULT_FINGERPRINTS = InMemoryCRMOutboundFingerprints() def record_crm_outbound_mutation( *, connection_id: UUID | str, object_type: str, external_id: str | None, fields: set[str] | frozenset[str], fingerprints: CRMOutboundFingerprintPort | None = None, ) -> None: if not external_id: return store = fingerprints or DEFAULT_FINGERPRINTS store.remember(str(connection_id), object_type, external_id, frozenset(fields)) def canonical_field_for_provider( provider: str, object_type: str, provider_field: str ) -> str | None: return _PROVIDER_TO_CANONICAL.get( (provider, object_type, provider_field.strip().lower().lstrip("/")) ) def verify_hubspot_webhook_v3( *, client_secret: str, method: str, url: str, raw_body: bytes | str, signature: str, timestamp: str, now_ms: int | None = None, ) -> None: if not client_secret: raise CRMEventError("crm_inbound_unverified", "missing_hubspot_webhook_secret") if not signature or not timestamp: raise CRMEventError("crm_inbound_unverified", "missing_hubspot_signature") try: ts = int(timestamp) except (TypeError, ValueError) as exc: raise CRMEventError("crm_inbound_unverified", "stale_hubspot_signature") from exc now = now_ms if now_ms is not None else int(time.time() * 1000) if abs(now - ts) > HUBSPOT_TIMESTAMP_TOLERANCE_MS: raise CRMEventError("crm_inbound_unverified", "stale_hubspot_signature") body = raw_body.decode("utf-8") if isinstance(raw_body, bytes) else raw_body base = f"{method}{url}{body}{timestamp}" digest = hmac.new(client_secret.encode(), base.encode(), hashlib.sha256).digest() expected = base64.b64encode(digest).decode() if not hmac.compare_digest(signature, expected): raise CRMEventError("crm_inbound_unverified", "bad_hubspot_signature") def _iso(value: Any) -> str | None: if value is None: return None if isinstance(value, datetime): return value.astimezone(UTC).isoformat().replace("+00:00", "Z") if isinstance(value, (int, float)): ms = float(value) if ms > 1e12: ms = ms / 1000.0 if ms > 1e14 else ms if ms > 1e11: return datetime.fromtimestamp(ms / 1000.0, tz=UTC).isoformat().replace("+00:00", "Z") if ms > 1e9: return datetime.fromtimestamp(ms, tz=UTC).isoformat().replace("+00:00", "Z") text = str(value).strip() return text or None def _as_list(payload: Any) -> list[dict[str, Any]]: if isinstance(payload, list): return [item for item in payload if isinstance(item, dict)] if isinstance(payload, dict): if isinstance(payload.get("events"), list): return [item for item in payload["events"] if isinstance(item, dict)] return [payload] return [] def parse_hubspot_webhook_payload(payload: Any) -> list[CRMInboundDraft]: drafts: list[CRMInboundDraft] = [] for item in _as_list(payload): subscription = str(item.get("subscriptionType") or "") object_type = None action = "changed" if subscription.startswith("contact."): object_type = "contact" elif subscription.startswith("company."): object_type = "company" elif subscription.startswith("deal."): object_type = "deal" if subscription.endswith(".creation") or subscription.endswith(".created"): action = "created" elif subscription.endswith(".deletion") or subscription.endswith(".deleted"): action = "deleted" elif subscription.endswith(".propertyChange"): action = "updated" elif "association" in subscription.lower(): action = "association_changed" event_id = item.get("eventId") if event_id is None: continue portal = item.get("portalId") property_name = item.get("propertyName") property_value = item.get("propertyValue") provider_fields: dict[str, Any] = {} changed: list[str] = [] if isinstance(property_name, str) and property_name: changed.append(property_name) provider_fields[property_name] = property_value event_type = ( f"crm.{object_type}.{action}" if object_type and action in {"created", "updated", "deleted"} else f"hubspot.{subscription or 'unknown'}" ) if action == "association_changed": event_type = "crm.association.changed" drafts.append( CRMInboundDraft( provider="hubspot", provider_event_id=str(event_id), event_type=event_type, object_type=object_type, external_id=None if item.get("objectId") is None else str(item.get("objectId")), occurred_at=_iso(item.get("occurredAt")), provider_fields=provider_fields, provider_changed=changed, provider_account_id=None if portal is None else str(portal), app_id=None if item.get("appId") is None else str(item.get("appId")), change_source=str(item["changeSource"]) if item.get("changeSource") else None, ) ) return drafts def _cdc_payloads(payload: Any) -> list[dict[str, Any]]: if isinstance(payload, list): return [item for item in payload if isinstance(item, dict)] if not isinstance(payload, dict): return [] if "ChangeEventHeader" in payload: return [payload] nested = payload.get("data") if isinstance(nested, dict): inner = nested.get("payload") if isinstance(inner, dict) and "ChangeEventHeader" in inner: return [inner] if isinstance(payload.get("payload"), dict) and "ChangeEventHeader" in payload["payload"]: return [payload["payload"]] if isinstance(payload.get("events"), list): return [item for item in payload["events"] if isinstance(item, dict)] return [payload] def _sf_object_type(entity: str) -> str | None: key = entity.strip().lower() if key in {"contact", "contactchangeevent"}: return "contact" if key in {"account", "accountchangeevent"}: return "company" if key in {"opportunity", "opportunitychangeevent"}: return "deal" if key in {"lead", "leadchangeevent"}: return "lead" return None def _sf_action(change_type: str) -> str: text = change_type.lower() if "undelete" in text: return "restored" if "create" in text: return "created" if "delete" in text: return "deleted" if "update" in text or "gap" in text: return "updated" return "changed" def parse_salesforce_cdc_payload(payload: Any) -> list[CRMInboundDraft]: drafts: list[CRMInboundDraft] = [] for item in _cdc_payloads(payload): header = ( item.get("ChangeEventHeader") if isinstance(item.get("ChangeEventHeader"), dict) else {} ) entity = str( header.get("entityName") or item.get("entityName") or item.get("object_type") or "" ) object_type = _sf_object_type(entity) action = _sf_action( str(header.get("changeType") or item.get("changeType") or item.get("event_type") or "") ) record_ids = header.get("recordIds") or item.get("recordIds") or [] if not isinstance(record_ids, list) or not record_ids: single = item.get("record_id") or item.get("recordId") record_ids = [single] if single else [] transaction_key = str( header.get("transactionKey") or item.get("transactionKey") or item.get("event_id") or item.get("eventId") or "" ) replay_id = None event_wrap = item.get("event") if isinstance(item.get("event"), dict) else None if event_wrap and event_wrap.get("replayId") is not None: replay_id = str(event_wrap["replayId"]) elif item.get("replayId") is not None: replay_id = str(item["replayId"]) changed_raw = header.get("changedFields") or item.get("changedFields") or [] if not isinstance(changed_raw, list): changed_raw = [] provider_changed = [str(name).lstrip("/") for name in changed_raw if name] provider_fields = { key: value for key, value in item.items() if key not in {"ChangeEventHeader", "event", "schema"} and not str(key).startswith("_") } org_id = item.get("organizationId") or header.get("organizationId") or item.get("org_id") change_origin = header.get("changeOrigin") or item.get("changeOrigin") occurred = _iso( header.get("commitTimestamp") or item.get("commitTimestamp") or item.get("occurred_at") ) if not record_ids: continue for record_id in record_ids: if record_id is None or str(record_id) == "": continue rid = str(record_id) provider_event_id = ( f"{replay_id}:{rid}" if replay_id else f"{transaction_key}:{rid}" if transaction_key else "" ) event_type = ( f"crm.{object_type}.{action}" if object_type in {"contact", "company", "deal"} and action in {"created", "updated", "deleted", "restored"} else f"salesforce.{entity or 'unknown'}.{action}" ) if object_type == "lead": event_type = ( f"crm.lead.{action}" if action in {"created", "updated", "deleted"} else event_type ) drafts.append( CRMInboundDraft( provider="salesforce", provider_event_id=provider_event_id, event_type=event_type, object_type=object_type if object_type in {"contact", "company", "deal"} else object_type, external_id=rid, occurred_at=occurred, provider_fields=provider_fields, provider_changed=provider_changed, provider_account_id=None if org_id is None else str(org_id), change_origin=str(change_origin) if change_origin else None, ) ) return drafts def account_id_from_config(provider: str, config: dict[str, Any] | None) -> str | None: cfg = config or {} if provider == "hubspot": value = cfg.get("portal_id") or cfg.get("hub_id") else: value = cfg.get("org_id") if value is None: return None text = str(value).strip() return text or None def attribute_crm_connection( *, provider: str, provider_account_id: str | None, connections: list[CRMAttributedConnection], ) -> CRMAttributedConnection: if not provider_account_id: raise CRMEventError("crm_connection_unmapped") matches = [ row for row in connections if row.connection.connector_key == provider and account_id_from_config(provider, row.connection.config) == str(provider_account_id) and row.connection.execution_domain == "C" ] if not matches: raise CRMEventError("crm_connection_unmapped") active = [row for row in matches if row.connection.status == "active"] chosen = active if not chosen: projects = {row.project_id for row in matches} connection_ids = {row.connection.connection_id for row in matches} if len(projects) == 1 and len(connection_ids) == 1: raise CRMEventError("crm_connection_inactive") raise CRMEventError("crm_connection_unmapped") projects = {row.project_id for row in chosen} connection_ids = {row.connection.connection_id for row in chosen} if len(projects) > 1 or len(connection_ids) > 1: raise CRMEventError("crm_attribution_ambiguous") return chosen[0] def _sparse_object( provider: str, object_type: str, fields: dict[str, Any], config: dict[str, Any] | None, ) -> dict[str, Any]: out: dict[str, Any] = {} stages = stage_map_from_config(config) pipelines = pipeline_map_from_config(config) for provider_key, value in fields.items(): canonical = canonical_field_for_provider(provider, object_type, str(provider_key)) if not canonical: continue if canonical == "stage": alias = reverse_alias(value, stages) if alias: out["stage"] = alias continue if canonical == "pipeline": if provider != "hubspot": continue alias = reverse_alias(value, pipelines) if alias: out["pipeline"] = alias continue if value is None or value == "": continue out[canonical] = value return out def _changed_canonical( provider: str, object_type: str, provider_changed: list[str], config: dict[str, Any] | None, sparse: dict[str, Any], ) -> list[str]: names: list[str] = [] for raw in provider_changed: canonical = canonical_field_for_provider(provider, object_type, raw) if not canonical: continue if canonical == "pipeline" and provider != "hubspot": continue if canonical == "stage" and "stage" not in sparse: continue if canonical == "pipeline" and "pipeline" not in sparse: continue if canonical not in names: names.append(canonical) return names def _hubspot_self_originated(draft: CRMInboundDraft, config: dict[str, Any] | None) -> bool: app_id = str((config or {}).get("app_id") or "").strip() if not app_id or not draft.app_id: return False source = (draft.change_source or "").upper() return draft.app_id == app_id and source in HUBSPOT_SELF_CHANGE_SOURCES def _salesforce_self_originated(draft: CRMInboundDraft, config: dict[str, Any] | None) -> bool: marker = str((config or {}).get("change_origin") or "").strip() if not marker or not draft.change_origin: return False return marker in draft.change_origin @dataclass class CRMInboundAcceptResult: accepted: list[CRMEventDispatch] = field(default_factory=list) duplicates: list[str] = field(default_factory=list) suppressed: list[str] = field(default_factory=list) ignored: list[str] = field(default_factory=list) replays: list[CRMEventDispatch] = field(default_factory=list) def ingest_verified_crm_inbound( *, provider: str, verified: bool, payload: Any, connections: list[CRMAttributedConnection], provider_account_id: str | None = None, claimed_project_id: UUID | None = None, dedupe: CRMEventDedupePort | None = None, fingerprints: CRMOutboundFingerprintPort | None = None, sink: CRMEventSink | None = None, ) -> CRMInboundAcceptResult: if not verified: raise CRMEventError("crm_inbound_unverified") if provider not in {"hubspot", "salesforce"}: raise CRMEventError("crm_connection_unmapped") drafts = ( parse_hubspot_webhook_payload(payload) if provider == "hubspot" else parse_salesforce_cdc_payload(payload) ) account_ids = {d.provider_account_id for d in drafts if d.provider_account_id} account = provider_account_id or ( next(iter(account_ids)) if len(account_ids) == 1 else provider_account_id ) if account_ids and provider_account_id and str(provider_account_id) not in account_ids: raise CRMEventError("crm_connection_unmapped") attributed = attribute_crm_connection( provider=provider, provider_account_id=account, connections=connections, ) if claimed_project_id is not None and claimed_project_id != attributed.project_id: raise CRMEventError("crm_attribution_ambiguous") result = CRMInboundAcceptResult() store = dedupe or DEFAULT_DEDUPE prints = fingerprints or DEFAULT_FINGERPRINTS config = attributed.connection.config or {} for draft in drafts: if not draft.provider_event_id: raise CRMEventError("crm_event_identity_missing") if draft.event_type in REJECTED_CRM_EVENT_TYPES or draft.event_type.startswith("crm.lead."): result.ignored.append(draft.provider_event_id) continue if ( draft.event_type in DEFERRED_CRM_EVENT_TYPES or draft.event_type not in FROZEN_CRM_EVENT_TYPES ): result.ignored.append(draft.provider_event_id) continue if draft.object_type not in {"contact", "company", "deal"} or not draft.external_id: result.ignored.append(draft.provider_event_id) continue dedupe_key = f"{provider}:{attributed.connection.connection_id}:{draft.provider_event_id}" if not store.claim(dedupe_key): result.duplicates.append(draft.provider_event_id) previous = store.get(dedupe_key) if previous is not None: result.replays.append(previous) continue sparse = _sparse_object(provider, draft.object_type, draft.provider_fields, config) changed = _changed_canonical( provider, draft.object_type, draft.provider_changed, config, sparse ) if ( _hubspot_self_originated(draft, config) or _salesforce_self_originated(draft, config) or prints.is_self_originated( str(attributed.connection.connection_id), draft.object_type, draft.external_id, frozenset(changed), ) ): result.suppressed.append(draft.provider_event_id) continue event = CRMEvent( event_id=str(uuid4()), event_type=draft.event_type, # type: ignore[arg-type] occurred_at=draft.occurred_at, object_ref=CRMObjectRef(object_type=draft.object_type, external_id=draft.external_id), changed_fields=changed, object=sparse or None, ) dispatch = CRMEventDispatch( project_id=str(attributed.project_id), workspace_id=str(attributed.workspace_id) if attributed.workspace_id else None, connection_id=str(attributed.connection.connection_id), provider_event_id=draft.provider_event_id, event=event, ) result.accepted.append(dispatch) store.put(dedupe_key, dispatch) if sink is not None: sink.emit(dispatch) return result def consumer_event_payload(dispatch: CRMEventDispatch) -> dict[str, Any]: body = dispatch.event.model_dump() body["project_id"] = dispatch.project_id return json.loads(json.dumps(body))