EventHub

Stakeholders

External Recipient: Backend team (internal)

Product Owner: [internal Name/Team]

Technical Owner: Data Engineering

Purpose

Data products consumed by the Backend team, delivered as Pub/Sub events and file exports derived from dbt models. EventHub covers both real-time delta-based events and scheduled file deliveries.

File Deliveries

CIP Verified Users

Daily export of users eligible for CIP verification, delivered as a CSV to Azure Blob Storage.

What it delivers: A list of user_ids for users who have user_state = 'Active' and kyc_cip_match_status = '3 out of 3'.


dbt model	`rpt_backend_cip_eligibility`
DAG	`backend_cip_verified_users_azure`
Schedule	`50 7 * * *` (07:50 UTC daily)
Destination	Azure Blob Storage — account: `prodpspfunding`, container: `pspfunding-cip-verifications`
Filename	`cip_verified_users_{YYYYMMDD}.csv`
Repos	dt-airflow-dags (DAG), dt-reporting (build & upload)

Pipeline:

DAG runs dbt build on rpt_backend_cip_eligibility (BigQuery, federated target)
dt-reporting exports the table as a full CSV report
File is uploaded to Azure Blob Storage

Restricted Users Export

Daily export of users to be restricted for an unverified address, delivered as a CSV to Azure Blob Storage and scheduled for restriction by the Backend team.

What it delivers: A list of user_ids for active US app users (holding a primary checking account) who have not verified their address, are not already under an ongoing address restriction, and passed KYC more than 15 days ago.


dbt model	`rpt_backend_address_restrict_access_export`
DAG	`internal_address_restricted_users_azure`
Schedule	`0 2 * * *` (02:00 UTC daily)
Destination	Azure Blob Storage — account: `prodkycdocuments`, container: `uploadsschedulerestriction`
Filename	`users_to_restrict_{YYYYMMDD}.csv`
Repos	dt-airflow-dags (DAG), dt-reporting (build & upload)

Pipeline:

DAG runs dbt build on rpt_backend_address_restrict_access_export (BigQuery, federated target)
dt-reporting exports the table as a full CSV report
File is uploaded to Azure Blob Storage

Events

Naming conventions and contents of the events have been defined as per the requirements set by the consumers (Backend team).


DAG	`internal_events_pubsub`
Schedule	Triggered by `dbt-structure-daily` DAG completion Airflow DAG
Weekend Delivery	[Yes/No]
dbt schema	`reverse_etl`
dbt folder	`models/reverse_etl`
Repos	dt-airflow-dags (DAG), dt-bq2pubsub (publish)

RemittanceRateComparisonUpdatedEvent is published from the fetch_fx_rates DAG instead, so it ships with each 15-minute competitor scrape rather than on the daily schedule.

Data

Each event has three dbt models following the pattern: internal_<event_name>, internal_<event_name>_delta, and internal_<event_name>_previous_run.

Snapshot events are the exception — no delta or previous-run model, and every run republishes the full table. RemittanceRateComparisonUpdatedEvent is the only one today.

PspRiskScoreUpdatedEvent

One row per user whose PSP risk score changed.


Event name	`Data.PspRiskScoreUpdatedEvent`
dbt models	`internal_psp_risk_score_updated_event`, `_delta`, `_previous_run`
Payload	See below

{
  "advisorId": null,
  "eventCreated": "2025-10-20T14:59:30.836894+00:00",
  "eventData": {
    "risk_groups": "Medium Risk",
    "risk_score": 0.6,
    "total_points": 8
  },
  "eventId": "033f7f66-2616-4296-a547-2fe7b45d7f5e",
  "eventName": "Data.PspRiskScoreUpdatedEvent",
  "externalReferenceId": null,
  "publishedTime": "2025-10-20T14:59:30.836894+00:00",
  "userId": "3b40a2d3-936e-46a4-bfbd-288ff8d4e826"
}

CrdeUserModelUpdatedEvent

One row per user whose credit/risk decision model changed.


Event name	`Data.CrdeUserModelUpdatedEvent`
dbt models	`internal_crde_user_model_updated_event`, `_delta`, `_previous_run`
Payload	See below

{
  "advisorId": null,
  "eventCreated": "2025-10-20T14:57:41.962859+00:00",
  "eventData": {
    "aml_sar_status": false,
    "avg_money_in_180": 0,
    "num_disputed_transactions_180": 0,
    "service_score": 8,
    "sum_disputed_amount_180": 0,
    "total_money_loss_180": 0,
    "user_credit_segment": "New_To_Platform",
    "user_id": "b956d91d-34da-4ded-8065-aed4bbd472cf",
    "user_state": "Active"
  },
  "eventId": "03f8725d-1841-4e11-a46b-bc89c6a274d3",
  "eventName": "Data.CrdeUserModelUpdatedEvent",
  "externalReferenceId": null,
  "publishedTime": "2025-10-20T14:57:41.962859+00:00",
  "userId": "b956d91d-34da-4ded-8065-aed4bbd472cf"
}

CreditEligibilityDataUpdatedEvent

One row per user whose credit eligibility data changed.


Event name	`Data.CreditEligibilityDataUpdatedEvent`
dbt models	`internal_credit_eligibility_data_updated_event`, `_delta`, `_previous_run`
Payload	See below

{
  "advisorId": null,
  "eventCreated": "2025-10-20T14:58:17.289648+00:00",
  "eventData": {
    "aml_sar_status": false,
    "total_money_loss_180": 0,
    "user_id": "2a57397e-6a45-48f5-ab2f-9e585725c095"
  },
  "eventId": "7ba99b71-f598-4522-be9d-50772f5bf7f4",
  "eventName": "Data.CreditEligibilityDataUpdatedEvent",
  "externalReferenceId": null,
  "publishedTime": "2025-10-20T14:58:17.289648+00:00",
  "userId": "2a57397e-6a45-48f5-ab2f-9e585725c095"
}

UserRiskLevelUpdatedEvent

One row per user whose risk level changed.


Event name	`Data.UserRiskLevelUpdatedEvent`
dbt models	`internal_user_risk_level_updated_event`, `_delta`, `_previous_run`
Payload	See below

{
  "advisorId": null,
  "eventCreated": "2025-10-20T14:58:17.289648+00:00",
  "eventData": {
    "customer_risk_rating": "Medium",
    "diaspora_risk_level": "High",
    "user_id": "2a57397e-6a45-48f5-ab2f-9e585725c095"
  },
  "eventId": "7ba99b71-f598-4522-be9d-50772f5bf7f4",
  "eventName": "Data.UserRiskLevelUpdatedEvent",
  "externalReferenceId": null,
  "publishedTime": "2025-10-20T14:58:17.289648+00:00",
  "userId": "2a57397e-6a45-48f5-ab2f-9e585725c095"
}

CheckRiskLevelUpdatedEvent

One row per user whose check risk level changed.


Event name	`Data.CheckRiskLevelUpdatedEvent`
dbt models	`internal_check_risk_level_updated_event`, `_delta`, `_previous_run`
Payload	See below

{
  "advisorId": null,
  "eventCreated": "2025-10-20T14:58:17.289648+00:00",
  "eventData": {
    "risk_level": "High",
    "user_id": "2a57397e-6a45-48f5-ab2f-9e585725c095"
  },
  "eventId": "7ba99b71-f598-4522-be9d-50772f5bf7f4",
  "eventName": "Data.CheckRiskLevelUpdatedEvent",
  "externalReferenceId": null,
  "publishedTime": "2025-10-20T14:58:17.289648+00:00",
  "userId": "2a57397e-6a45-48f5-ab2f-9e585725c095"
}

RemittanceRateComparisonUpdatedEvent

One row per remittance corridor, holding every competitor quote scraped for that corridor in the last 48 hours. Powers the competitor rate comparison table on the remittance landing pages.


Event name	`Data.RemittanceRateComparisonUpdatedEvent`
dbt models	`internal_remittance_rate_comparison_updated_event` (no `_delta` or `_previous_run` — full snapshot each run)
DAG	`fetch_fx_rates`
Schedule	`/15 * * *`, published after the `dbt_fx` task
Payload	See below

userId is always null — the event is keyed to a corridor, not a user. Only the most recent quote per competitor and payout method is kept. provider_code, provider_name, exchange_rate and fetched_at are always present; pickup_type, fee, promo_exchange_rate and promo_fee are null when the provider does not report them. send_amount is nullable in the schema, though the dbt model currently drops quotes that did not report one.

{
  "advisorId": null,
  "eventCreated": "2026-08-10T09:15:04.512873+00:00",
  "eventData": {
    "source_country_id": "USA",
    "source_currency": "USD",
    "destination_country_id": "MEX",
    "destination_currency": "MXN",
    "quotes": [
      {
        "provider_code": "western_union",
        "provider_name": "Western Union",
        "pickup_type": "cash",
        "send_amount": 500.0,
        "exchange_rate": 16.63,
        "fee": 1.99,
        "promo_exchange_rate": 17.64,
        "promo_fee": 0.0,
        "fetched_at": "2026-08-10T07:45:24.780729+00:00"
      },
      {
        "provider_code": "felix_pago",
        "provider_name": "Felix Pago",
        "pickup_type": null,
        "send_amount": 500.0,
        "exchange_rate": 16.4,
        "fee": 0.0,
        "promo_exchange_rate": null,
        "promo_fee": null,
        "fetched_at": "2026-08-10T07:30:11.780729+00:00"
      }
    ]
  },
  "eventId": "9b1f0a2c-4c7e-4b6f-9a4d-77c2f0e1b5aa",
  "eventName": "Data.RemittanceRateComparisonUpdatedEvent",
  "externalReferenceId": null,
  "publishedTime": "2026-08-10T09:15:04.512873+00:00",
  "userId": null
}

Publishing Flow

graph TB
    Credit[Credit Event<br/>Publisher DAG]

    Credit --> Base[(update<br/>Base Table)]

    Base --> D1[(payload Delta<br/>event 1)]
    Base --> D2[(payload Delta<br/>event 2)]

    D1 --> Get[Get payload<br/>data<br/>from BQ<br/>per event]
    D2 --> Get

    Get --> Create[Create Event<br/>Messages]
    Create --> Validate[Validate<br/>All Messages]

    Validate -->Publish[Publish<br/>to Pub/Sub]

    Publish --> Update[Update<br/>Archive]

    Update --> Success[Complete]

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

Run Daily Models — Daily models run in the morning (Swedish timezone) via the dbt-structure-daily DAG in Airflow.
Trigger Event Publisher — Once daily models complete, they trigger the internal_events_pubsub DAG.
Calculate Base & Delta Tables — The DAG runs dbt models tagged with EventHub: calculates the base tables (internal_<event_name>), then creates delta tables (internal_<event_name>_delta) comparing current fields vs. last run. Delta contains only fields meant for event payloads.
Publish Events — One event is published to Pub/Sub for each row in the delta table.
Update Archive — The DAG updates the previous run table (internal_<event_name>_previous_run) holding the current state of the base table, serving as the benchmark for the next comparison run.

Snapshot events skip steps 3 and 5: fetch_fx_rates builds the base model with dbt build --select tag:FX after each scrape, then publishes one event per row. With no delta comparison, unchanged corridors are republished every run.