Intercom

Stakeholders

External Recipient: Customer Support / CRM (Intercom)

Product Owner: Growth

Technical Owner: Data Engineering

Purpose

User data is pushed into Intercom so the support team sees up-to-date context on every contact. Two complementary pipelines deliver it:

Pipeline	Repo	Source	Trigger	Cadence
Attribute batch	dt-intercom-batch	BigQuery (`dbt.intercom_user_attributes_export`)	Airflow DAG `intercom_upload_user_attributes`	Daily, after `dbt_structure_daily`
Event stream (Lambda)	dt-intercom-lambda	Kinesis backend events	AWS Lambda (event-driven)	Real-time

The batch keeps slowly-changing user attributes in sync from the data warehouse; the Lambda streams real-time lifecycle events as they happen. Both write to the Intercom Contacts API, upserting by external_id.

Direction of flow

This page covers data flowing to Intercom. For the reverse direction — extracting data from the Intercom API into GCS — see Intercom API.

flowchart LR
    subgraph DW[Data Warehouse]
        EXP[dbt.intercom_user_attributes_export]
    end
    subgraph BE[Backend]
        K[Kinesis event stream]
    end
    EXP -->|Airflow DAG, daily| BATCH[dt-intercom-batch]
    K -->|real-time| LAMBDA[dt-intercom-lambda]
    BATCH -->|POST / PUT contacts| IC[(Intercom)]
    LAMBDA -->|POST / PUT contacts| IC

Hold "Alt" / "Option" to enable pan & zoom

Upsert behaviour (both pipelines)

Every contact is identified by external_id and sent with role: "user". The write logic is shared in spirit:

POST /contacts to create the contact.
On 409 Conflict (contact already exists), extract the Intercom contact ID from the response and PUT /contacts/{id} to update it.
On 404 Not Found (contact archived in Intercom), the write is skipped and treated as success.
429 and 5xx responses are retried with exponential backoff (3 attempts, 5 + 2^attempt seconds).

null/empty values are omitted from the payload — Intercom only receives attributes that have a value.

Attribute Batch

A scheduled job that syncs warehouse-derived user attributes to Intercom.


DAG	`intercom_upload_user_attributes` (no schedule of its own — triggered by `dbt_structure_daily`)
Source table	`dbt.intercom_user_attributes_export`
State table	`reverse_etl.last_intercom_user_attributes_export`
Runtime	Python 3.13, `aiohttp` + `google-cloud-bigquery`
Image	`minoritycr.azurecr.io/datateam/intercom-batch` (run via `KubernetesPodOperator`, `python run.py`)
Repos	dt-airflow-dags (DAG), dt-intercom-batch (job)

Pipeline:

Read dbt.intercom_user_attributes_export and diff it against the state table using a farm_fingerprint(to_json_string(...)) row hash — only new or changed contacts are sent.
Map BigQuery columns to Intercom custom attributes and upsert each contact (writes bounded by asyncio.Semaphore, MAX_CONCURRENCY = 20).
Advance the row hash for successfully-synced contacts (state written in chunks of 10000); failed contacts retry next run.

A --dry-run flag computes the diff and logs a sample payload without writing to Intercom or state.

Datapoints sent

Source columns are mapped to Intercom custom attributes (CUSTOM_ATTRIBUTES_MAPPING in src/intercom_batch/run.py). external_reference_id is sent as the contact's external_id.

BigQuery column	Intercom custom attribute	Notes
`external_reference_id`	(used as `external_id`)	Contact identifier
`user_id`	`Hydra ID`	Internal user ID
`account_country_id`	`Account Country ID`
`locale`	`Locale`
`diaspora`	`Diaspora`
`verification_type`	`Verification Type`	KYC verification type
`rfm_segment`	`RFM Segment`	Recency/Frequency/Monetary segment
`first_calling_country`	`First Calling Country`
`first_remittance_country`	`First Remittance Country`
`calling_country`	`Calling Countries`	Array → hyphen-joined (e.g. `SE-NO`)
`remittance_country`	`Remittance Countries`	Array → hyphen-joined
`mtu_country`	`Mtu Countries`	Array → hyphen-joined
`primary_card_status`	`Primary Card Status`
`primary_card_type`	`Primary Card Type`
`card_created_at`	`Card Created Time`	ISO 8601
`card_shipped_at`	`Card Shipped Time`	ISO 8601
`user_signup_at`	`User Signup Time`	ISO 8601
`card_shipped_last_30_days`	`Card Shipped Last 30 days`	Boolean
`direct_deposit_last_30_days`	`Direct Deposit Last 30 days`	Boolean
`has_mtu_30_days`	`Has MTU 30 Days`	Boolean
`has_remitted_30_days`	`Has Remitted 30 Days`	Boolean
`has_referred`	`Has Referred`	Boolean
`is_referred`	`Is Referred`	Boolean
`has_received_mpay`	`Has Received MPay`	Boolean
`has_sent_mpay`	`Has Sent MPay`	Boolean
`is_advisor`	`Is Advisor`	Boolean
`is_user_blocked`	`Is User Blocked`	Boolean

Array & date handling

Array columns (calling_country, remittance_country, mtu_country) are joined with -; empty arrays are dropped. Date columns are sent as ISO 8601 strings; NaT/null are dropped.

Event Stream (Lambda)

An event-driven AWS Lambda (Python 3.13) that turns real-time backend events into Intercom contact updates.


Trigger	AWS Kinesis stream of backend events
Dispatch	`eventName` matched against nine known event types (`src/event_validation.py`); unknown events skipped
Concurrency	`asyncio.Semaphore` (`INTERCOM_POST_DATA_CONCURRENCY`, default `5`)
Config	`INTERCOM_API_BASE_URL`, `INTERCOM_API_TOKEN` (required); `INTERCOM_POST_DATA_CONCURRENCY`, `LOG_LEVEL` (optional)
Deploy	Azure DevOps pipeline packages `src/` + dependencies into `intercom_lambda.zip`
Repos	dt-intercom-lambda

Each Kinesis record's data is a base64-encoded JSON payload carrying eventName, eventData, externalReferenceId and userId. externalReferenceId becomes external_id on every event (required). Standard Intercom fields (phone, email, name) are set where available; everything else lands under custom_attributes.

Events and datapoints sent

external_id (from externalReferenceId, required) and custom_attributes.Hydra ID (from userId) are sent on every event and omitted from the table below.

Event	Target	Source
`UserCreatedEvent`	`phone`	`eventData.msisdn` (prefixed `+`)
	`custom_attributes.Locale`	`eventData.locale`
	`custom_attributes.Account Country ID`	`eventData.accountCountry`
	`custom_attributes.User Creation Received`	`"yes"` (constant)
`UserCompleteSignInEvent`	`phone`	`eventData.msisdn` (prefixed `+`)
`AddressChangedEvent`	`custom_attributes.Address`	`eventData.addressLine1`
	`custom_attributes.Address Line 2`	`eventData.addressLine2`
	`custom_attributes.city`	`eventData.city`
	`custom_attributes.State`	`eventData.state`
	`custom_attributes.Zip Code`	`eventData.zipCode`
`KycSuccessfullyVerifiedEvent`	`name`	`eventData.name.firstName` + `familyName`
	`custom_attributes.Address`	`eventData.address.addressLine1`
	`custom_attributes.city`	`eventData.address.city`
	`custom_attributes.State`	`eventData.address.state`
	`custom_attributes.Zip Code`	`eventData.address.zipCode`
	`custom_attributes.country`	`eventData.address.country`
	`custom_attributes.Kyc Passed Date`	`eventCreated`
	`custom_attributes.Kyc Passed`	`"yes"` (constant)
`EmailVerificationCreatedEvent`	`email`	`eventData.email`
`EmailVerificationSuccessEvent`	`email`	`eventData.email`
	`custom_attributes.Email Verified`	`"yes"` (constant)
`LocaleChangedEvent`	`custom_attributes.Locale`	`eventData.locale`
`SubscriptionFeeChargedEvent`	`custom_attributes.Next Membership Renewal Date`	`eventData.NextChargeDate`
	`custom_attributes.Trial Subscription Start Date`	`eventData.ChargedDate` (trial only)
	`custom_attributes.Trial Subscription End Date`	`eventData.NextChargeDate` (trial only)
`ChangeMsisdnResultMinorityEvent` ¹	`phone`	`eventData.NewMsisdn`

¹ Filtered to events where eventData.MsisdnSuccessfullyChanged is true.