HubSpot webhooks

Changes made by people in HubSpot — a rep editing a contact, someone clicking an unsubscribe link — need to flow back into Morpheus. HubSpot delivers those as v3 webhook notifications, which an external forwarder wraps into a Crm.HubSpotWebhook message on CrmSyncQueue.

flowchart LR
  HS(["HubSpot portal"]) -->|v3 webhook| FWD(["forwarder"])
  FWD -->|Crm.HubSpotWebhook| Q[["CrmSyncQueue.fifo"]]
  Q --> H["CrmHubSpotWebhookHandler"]
  H --> R["WebhookProcessorRegistry\nfirst match wins"]
  R --> P1["NewsletterOptOutProcessor"]
  P1 --> DB[("target_emails")]

The forwarder is not part of this repository. Morpheus only consumes the wrapped event.

Envelope

One message carries a batch of inner notifications. Inner events arrive in HubSpot’s camelCase and are accepted in either case via AliasChoices.

{
  "event_type": "Crm.HubSpotWebhook",
  "source": "hubspot-forwarder",
  "portalId": 12345678,
  "events": [
    {
      "subscriptionType": "contact.propertyChange",
      "objectId": 987654321,
      "occurredAt": 1713628800000,
      "propertyName": "hs_email_optout_44467216",
      "propertyValue": "true"
    }
  ]
}

Full model: CrmSyncQueue events. events requires at least one element (min_length=1), and objectId accepts a numeric string as well as an integer.

Handler

CrmHubSpotWebhookHandler opens one database session and one CRM client for the whole batch, then dispatches each inner event:

with get_session() as session:
    for inner in event.events:
        _registry.dispatch(
            session=session, client=client,
            subscription_id=subscription_id, inner=inner,
        )

One session for the batch means one commit for the batch. A processor that raises rolls back the whole message — which is why processors are written to log and return rather than raise.

The processor registry

WebhookProcessorRegistry walks its processors in declaration order and calls the first whose applies_to(inner) returns True, then stops. That prevents two processors mutating the same row or double-counting a side effect.

Nothing matching logs at DEBUG and returns — no exception — so an unknown webhook shape cannot poison the queue.

Processor Matches

NewsletterOptOutProcessor

subscription_type == "contact.propertyChange" and property_name matching ^hs_email_optout_(\d+)$ where the captured id equals 44467216

The registry accepts an injected tuple, which is how tests supply a custom chain.

The processor contract

class HubSpotWebhookProcessor(ABC):
    @abstractmethod
    def applies_to(self, inner: HubSpotWebhookInnerEvent) -> bool: ...

    @abstractmethod
    def process(self, *, session: Session, client: HubSpot,
                subscription_id: str, inner: HubSpotWebhookInnerEvent) -> None: ...

Rules, from the registry’s own documentation:

  • Match tightly in applies_to() — on subscription_type, property_name, and any discriminating value. A loose matcher silently swallows events a later processor should have handled.

  • Read from HubSpot with client if you need more than the notification carries; persist with session and the data-layer repositories.

  • Never commit inside a processor. The handler’s get_session() block owns the transaction.

  • Never log full payloads or raw PII.

  • An unsupported shape must log and return, not raise.

  • Order matters — put specific matchers before generic ones when both could match.

NewsletterOptOutProcessor

The one production processor, and a good model for a new one.

_OPTOUT_PROPERTY = re.compile(r"^hs_email_optout_(\d+)$")

def applies_to(self, inner) -> bool:
    if inner.subscription_type != "contact.propertyChange":
        return False
    m = _OPTOUT_PROPERTY.match(inner.property_name or "")
    if not m:
        return False
    return int(m.group(1)) == HUBSPOT_NEWSLETTER_SUBSCRIPTION_TYPE_ID    # 44467216

process() then:

  1. Interprets property_value: "true" → opted out, "false" or missing/blank → resubscribed. HubSpot may omit or clear the value when the contact is no longer opted out, so blank is deliberately treated as false. Anything else logs a warning and returns.

  2. Converts occurred_at (epoch milliseconds) to a UTC datetime.

  3. Reads the contact’s email from the CRM by object_id. A 404 logs a warning and returns; any other ApiException is re-raised.

  4. Parses subscription_id; an invalid UUID logs at ERROR and returns.

  5. Calls TargetEmailsRepository.set_newsletter_opt_out(), which writes newsletterOptOut and newsletterOptOutAt — newsletter only, not the other two opt-out flags — and refuses to move newsletterOptOutAt backwards, so an out-of-order webhook delivery cannot undo a newer state.

  6. No matching target_emails row logs at INFO and returns.

Every failure path is a log-and-return. That is what keeps one bad notification from rolling back the rest of the batch.

Adding a processor

  1. Create a module under lambdas/morpheus_hubspot/events/webhook_processors/ subclassing HubSpotWebhookProcessor.

  2. Implement applies_to() with a tight matcher.

  3. Implement process() — read via client, write via session and the repositories, and do not commit.

  4. Append an instance to the default tuple in WebhookProcessorRegistry.init, in the right order.

  5. Optionally export it from webhook_processors/init.py.

  6. Add a test asserting both a match and a near-miss — the near-miss is what proves the matcher is tight.

There is no registry entry to add and no template change: the Crm.HubSpotWebhook event type is already registered, and processors are internal to it.

Which HubSpot subscriptions to configure

The forwarder decides what HubSpot sends. For the current processor the portal needs a contact.propertyChange subscription covering hs_email_optout_44467216. Adding a processor generally means adding a matching HubSpot subscription too — otherwise the processor never fires.

Testing

cd lambdas && uv run pytest morpheus_hubspot/tests/test_newsletter_opt_out_processor.py -v

Sample payload: lambdas/morpheus_hubspot/events/samples/crm_hubspot_webhook_newsletter_optout.json.