Skip to content
BranchesTags

Repository files navigation

Real-Time Customer 360

Real-Time Customer 360 turns fragmented customer activity into one trusted, continuously updated profile. It ingests change events from operational systems, resolves identities across channels, preserves every historical version, and serves current or point-in-time customer context through a low-latency API.

Use it to power customer support, analytics, segmentation, personalization, compliance audits, and downstream machine-learning features from the same canonical customer record.

What the platform delivers

Capability Product behavior
Real-time ingestion Idempotently consumes Debezium-style create, update, delete, and snapshot events.
Unified identity Connects CRM, billing, order, support, marketing, web, and mobile identifiers to a stable master customer ID.
Golden profiles Applies attribute-level survivorship rules and produces one current profile per customer.
Historical accuracy Reconstructs what was true at a business time using only information known at a specified system time.
Event-time context Enriches web and mobile events with the customer state that applied when the event occurred.
Governed access Hashes identity keys, masks PII by default, quarantines unsafe matches, and exposes authenticated APIs.
Replayable processing Retains an immutable Bronze change log and uses idempotent downstream writes for recovery and backfills.

Customer profile

A profile combines golden attributes, resolved identifiers, and live behavioral rollups:

{
  "master_customer_id": "c5131969-5bfd-51cd-8fce-14f2419e37bb",
  "first_name": "Ada",
  "last_name": "Lovelace",
  "email": "a***@gmail.com",
  "tier": "gold",
  "status": "active",
  "region": "UK",
  "source_systems": ["billing", "crm", "marketing"],
  "metrics": {
    "lifetime_orders": 1,
    "lifetime_revenue": 129.95,
    "open_tickets": 1,
    "event_count": 1
  }
}

Architecture

PostgreSQL / MySQL ── Debezium ─┐
                                ├── Kafka ── Bronze CDC log
Web / mobile events ────────────┘                 │
                                                  ▼
                                      Identity + stream processing
                                         Flink / Spark jobs
                                                  │
                                     ┌────────────┴────────────┐
                                     ▼                         ▼
                              Iceberg Silver             Quarantine/remaps
                              SCD2 + facts                      │
                                     │                         │
                                     └────────────┬────────────┘
                                                  ▼
                                      dbt Gold Customer 360
                                         │              │
                                         ▼              ▼
                                      Redis           Trino
                                         └──────┬───────┘
                                                ▼
                                           Profile API

The repository includes two deployment profiles:

  • Product runtime: a complete Python and SQLite implementation for local deployments, demonstrations, contract testing, and CI. It exercises ingestion, identity resolution, bitemporal history, rollups, enrichment, and every public API.
  • Distributed runtime: source schemas, Debezium connectors, Kafka infrastructure, Iceberg DDL, Spark jobs, Flink SQL, dbt models, and Dagster assets for scaling the same data contracts across production engines.

Read the detailed architecture and data model.

Quick start

Run locally

Requirements: Python 3.10+ and GNU Make.

make install
make demo
make test

make demo initializes the platform, processes a cross-system customer scenario, runs the data quality gates, and prints the unified profile.

Start the API:

make api

The service is available at:

  • API: http://localhost:8000
  • OpenAPI: http://localhost:8000/openapi.json
  • Interactive documentation: http://localhost:8000/docs
  • Health check: http://localhost:8000/health

Run with Docker

Requirements: Docker with Compose v2.

cp .env.example .env
make up
docker compose -f infra/compose/docker-compose.yml exec profile-api c360 seed --reset
curl -H 'X-API-Key: local-development-key' \
  'http://localhost:8000/resolve/email/ada.lovelace@gmail.com'

Stop the service with make down.

Generate live activity

With the API running and seeded:

python generators/generate_events.py --count 100 --interval 0.1

Every generated event passes through the CDC endpoint, identity lookup, event-time enrichment, fact store, and current profile materialization.

API

All product endpoints except /health require an X-API-Key header.

Method Endpoint Description
GET /profile/{master_id} Return the current unified profile.
GET /resolve/{identifier_type}/{value} Resolve an identifier to its master customer ID.
GET /profile/{master_id}/as-of Reconstruct a bitemporal profile.
GET /profile/{master_id}/timeline Return merged customer activity in reverse chronological order.
POST /cdc/events Ingest up to 1,000 CDC events in one request.
GET /health Return service health without authentication.

Resolve an identity

curl -H 'X-API-Key: local-development-key' \
  'http://localhost:8000/resolve/email/ada.lovelace@gmail.com'
{"master_customer_id":"c5131969-5bfd-51cd-8fce-14f2419e37bb"}

Supported identifiers include email, phone, account_number, crm_customer_id, billing_account_id, order_customer_id, support_user_id, lead_id, login_user_id, device_id, and anonymous_id.

Reconstruct a historical profile

curl -G -H 'X-API-Key: local-development-key' \
  --data-urlencode 'business_ts=2025-01-01T15:00:00Z' \
  --data-urlencode 'knowledge_ts=2025-01-01T20:00:00Z' \
  'http://localhost:8000/profile/MASTER_ID/as-of'

business_ts selects when the customer state was effective. knowledge_ts limits the query to information the platform had received by that time, preventing future information leakage.

Ingest a CDC event

curl -X POST 'http://localhost:8000/cdc/events' \
  -H 'Content-Type: application/json' \
  -H 'X-API-Key: local-development-key' \
  -d '{
    "process": true,
    "events": [{
      "event_id": "crm.customers:42:100",
      "source_system": "crm",
      "entity": "customers",
      "natural_key": "42",
      "operation": "u",
      "business_ts": "2025-01-01T12:00:00Z",
      "system_ts": "2025-01-01T12:00:02Z",
      "after": {
        "customer_id": 42,
        "email": "ada@example.com",
        "tier": "gold"
      },
      "partition": 0,
      "offset": 100
    }]
  }'

Operations follow the Debezium envelope: c create, u update, d delete, and r snapshot read. event_id is the idempotency key. Events are processed in system-time, partition, and offset order.

Identity safety

Identity resolution favors precision over aggressive matching:

  • Emails are lowercased and configurable Gmail dot/plus aliases are canonicalized.
  • Source identifiers are namespaced before matching.
  • Role addresses such as info@ and support@ never create customer links.
  • Phone numbers never merge two profiles without another strong signal.
  • Shared phones and oversized components are quarantined for review.
  • Existing master IDs survive the arrival of new identifiers.
  • Merges emit remap records; splits preserve identity history.
  • Identifier values are stored as salted SHA-256 hashes with masked display values.

See identity resolution for the full rules.

Security and privacy

  • Change C360_API_KEY and C360_IDENTITY_SALT before using persistent data.
  • Profile responses mask email and phone fields by default.
  • Send X-PII-Scope: pii:read to exercise unmasked local responses. Production deployments must derive this scope from authenticated JWT claims or an API gateway, not a caller-controlled header.
  • Store production secrets in a secret manager; never commit .env.
  • Apply encryption at rest to MinIO/S3, database volumes, backups, and transport connections.
  • Follow the privacy-erasure procedure in the runbook.

Configuration

Copy .env.example to .env and override these settings:

Variable Default Purpose
C360_DATABASE_PATH ./var/customer360.db Local product database.
C360_IDENTITY_SALT Development value Salt used for identifier hashing.
C360_API_KEY local-development-key API authentication key.
C360_LOG_LEVEL INFO Application log level.
C360_PROFILE_CACHE_TTL_SECONDS 300 Intended serving-cache lifetime.
KAFKA_BOOTSTRAP_SERVERS localhost:9092 Distributed event backbone.
SCHEMA_REGISTRY_URL http://localhost:8081 Avro schema registry endpoint.

Operations

make seed       # Reset and load the deterministic product scenario
make validate   # Run data-quality and temporal invariants
make test       # Run unit and API tests
make lint       # Run static checks
make check      # Run lint and tests
make up         # Start the containerized API
make down       # Stop containers

The pipeline records every execution in pipeline_runs. Failed or interrupted work can be safely resumed with c360 process; already-ingested events and downstream facts remain idempotent.

For connector recovery, identity incidents, backfills, and privacy workflows, use the operations runbook. Service objectives are documented in SLOs.

Distributed deployment

Start the infrastructure profile with:

export SOURCE_DB_PASSWORD='change-this-local-password'
make up-distributed
connectors/scripts/register.sh

This profile provisions the API, Kafka in KRaft mode, Schema Registry, Kafka Connect, PostgreSQL source systems, MySQL orders, MinIO, an Iceberg REST catalog, and Redis. Connector configurations are registered separately so credentials remain environment-controlled.

The production processing assets are organized under:

Directory Responsibility
connectors/ PostgreSQL/MySQL Debezium source connectors.
source-systems/ OLTP schemas and deterministic CRM seed data.
streaming/ Flink event-time temporal enrichment.
lakehouse/ Iceberg DDL and Spark identity, SCD2, history, and maintenance jobs.
transform/ dbt Gold profile models, tests, seeds, and as-of macro.
orchestration/ Dagster assets and refresh schedule.
quality/ Source contracts and local quality gates.
infra/ Container images and split Compose topology.

The distributed artifacts establish the production contracts and engine boundaries. Packaging the Flink/Spark jobs, installing the Iceberg Kafka Connect sink, production authentication, complete observability, and Kubernetes deployment remain environment-specific hardening work.

Development status

The local product runtime is fully executable and covered by automated tests for:

  • Cross-source identity resolution and stable IDs.
  • Merge, split, role-email, and shared-phone behavior.
  • CDC replay and idempotency.
  • Bitemporal reconstruction without future leakage.
  • Event-time profile enrichment and behavioral rollups.
  • Authenticated API access and PII redaction.
  • Identity, dimension, and materialization quality invariants.

The complete delivery sequence and production acceptance criteria remain in plan.md.

About

No description, website, or topics provided.

Resources

Readme
Activity

Stars

0 stars

Watchers

0 watching

Forks

0 forks
Report repository

Releases

Packages

Contributors

Languages