Service Overview

The CRM (Actian CRM365) is RIO’s system of record, but it is not built for analytics and the RIO application cannot query it directly. This service copies the CRM’s data — on a schedule and incrementally — into three purpose-built stores:

• S3 (bronze) — a durable archive of every table.
• ClickHouse — the analytics warehouse behind dashboards and the Cube semantic layer.
• RDS PostgreSQL — the transactional store the RIO application reads.

Along the way it standardises each table, writes an HMAC-SHA256 snapshot of every PII row into a DynamoDB vault, and emits an audit event at every stage so a run can be traced end to end.

Note that vaulting is an index, not a redaction. vault_pii_batch() writes the token and the lowercased values to DynamoDB and returns the dataframe unchanged, so plaintext PII still reaches S3, ClickHouse and RDS.

The full step-by-step walkthrough is documented as a flow: RIO Ingestion Service — CRM Data Sync Pipeline.

It is not event-driven

Despite publishing eight events, this service subscribes to no RIO event. It is started by a clock:

Trigger	Detail
EventBridge Scheduler cron	cron(0 * * * ? *) in prod, cron(0 0/6 * * ? *) elsewhere → Step Functions → ECS Fargate task
S3 object upload	aws.s3 / Object Created on the default bus, filtered to the product-master bucket and the reference/product_master/ prefix (infrastructure/modules/ecs/crm-data-sync.yaml:368-390)

The second is the only rule in the repo, and it listens to AWS, not to RIO. See AWS Default Event Bus.

Two event families, two envelopes

	Audit family	Hierarchy event
source	rio.platform (dependencies/audit_events.py:36)	rio.glue.crm_sync (dependencies/constants.py:57)
Events	8 (CRM Sync Started, ETL Batch Completed, …)	1 (Hierarchy Updated)
Payload	AuditEventDetail dataclass (audit_events.py:100-136)	business event with actor / context / data
Purpose	Observability and the audit trail	Trigger hierarchy recalculation

Audit event reference

All eight share source rio.platform. Enum definitions at audit_events.py:43-51 (detail-types) and :54-62 (event names).

Event	detail-type	detail.event_name	Emitter
ExternalSignalReceived	External Signal Received	rio.platform.ingestion.external_signal.received	:351
CRMSyncStarted	CRM Sync Started	rio.platform.ingestion.crm_sync.started	:284
CRMSyncCompleted	CRM Sync Completed	rio.platform.ingestion.crm_sync.completed	:305
CRMSyncFailed	CRM Sync Failed	rio.platform.ingestion.crm_sync.failed	:327
ETLBatchStarted	ETL Batch Started	rio.platform.ingestion.etl_batch.started	:372
ETLBatchCompleted	ETL Batch Completed	rio.platform.ingestion.etl_batch.completed	:392
ETLBatchFailed	ETL Batch Failed	rio.platform.ingestion.etl_batch.failed	:414

Data stores

dependencies/table_registry.py maps 29 CRM source tables to their targets. Every table is archived to S3; 16 of them also land in ClickHouse and 8 in RDS. Five are archive-only.

• ClickHouse (16 source tables) — dim_account, dim_campaign, dim_contract, dim_product, dim_territory, fact_bpf_history, fact_entitlement_header, fact_entitlement_line, fact_lead_snapshot, fact_opportunity_product, fact_opportunity_stage_history, fact_quote, fact_quote_approval_history, fact_quote_detail_history, fact_sales_order, fact_salesorder_detail_history.
• RDS Postgres (8 source tables → 5 target tables) — person, tenant_role, opportunity, account_team, deal_team, upserted through staging.temp_* staging tables. Three source tables fan out to two Postgres tables each via rds_destinations.
• DynamoDB — watermark table (how far each table has synced per destination) and the PII vault.

Observability

The task exports OpenTelemetry metrics over OTLP/HTTP when OTEL_ENABLED=true and OTEL_EXPORTER_ENDPOINT is set (both default to on in crm-data-sync.yaml), flushed every 15 seconds. init_metrics() runs at the top of main() and shutdown_metrics() in the closing finally.

Metric	Type	Meaning	Attributes
rio.ingestion.s3_queue_rows_sent	Counter	Rows written to the S3 gold bucket for ClickHouse	table_name, tenant_id
rio.ingestion.clickhouse_rows_received	Counter	Rows confirmed in ClickHouse after S3Queue ingestion	table_name, tenant_id

Comparing the two counters is how a silent S3Queue ingestion failure is caught — the audit events cannot show it, since ETLBatchCompleted always reports zero rows. System-level CPU, memory, disk and network metrics come from SystemMetricsInstrumentor.

A note on this repo’s own docs

events/contracts.md claims source rio.glue.crm_sync and a single detail-type RIO Ingestion Event for the whole audit family. The code uses source rio.platform and eight distinct Title Case detail-types. contracts.md also lists action and status values (sync_started, SUCCESS) that do not match the code enums (started, success). The code is authoritative and is what this catalog records.

Also note the README describes glue-jobs/, but the code actually lives in jobs/crm-data-sync/ and now runs on ECS, not Glue.

Inbound and Outbound Message Flow