Signal Pipeline · Planning

Context

Signal Pipeline เป็นโครงสร้างพื้นฐาน A-owned · อยู่ก่อน runtime ทุกตัว · ตอบสนอง CLAUDE.md Layer B (Event Bus) · Layer C (Data Lake) · Layer D (Pipelines) · Layer E (Knowledge Store). Feeds Dashboard (metrics) · Cases (AI analytics + multi-source intake) · Alerting (breach detect) · Wallet (transactions WORM) · City Health (vital signs).

CLAUDE.md §Kafka 19 topicsRetention · compression · partitions · key schemas

CLAUDE.md Layer B/C/D/EBronze/Silver/Gold · dbt models · embeddings

docs/kb/data/city_health_contract.json5 panels · 14 metrics · a_owned_dependency declared

docs/kb/data/dashboard_contract.json17 dashboards · primary_metrics consumed from Gold

docs/kb/data/multi_source_intake.jsonAI case intake from external signals

docs/kb/data/ai_issue_analytics.json12-stage pipeline consumes signals

A-owned Runtime Boundary

B owns (read-only for A)

city_health_contract metric list + a_owned_dependency declarations
dashboard_contract metric names
multi_source_intake source_origins
ai_issue_analytics pipeline stage spec

A owns (this runtime)

Kafka cluster (Confluent Cloud) + 19 topics config
Producer services (booking-svc, payment-svc, behavior-svc, location-svc, ugc-svc)
Consumer groups (data-lake-ingester, alerting, kpi-snapshot, rag-embed)
dbt project (Bronze→Silver→Gold models)
Airflow DAGs (6: ETL, NLP, Features, KPI, Reports, Backfill)
TimescaleDB schemas + hypertables
pgvector schemas + HNSW indexes
Neo4j schema + Cypher loaders
PII hashing (SHA-256) at producer before publish
GPS fuzz ±50m at producer
k-anonymity ≥5 filter at consumer before materialize
DLQ pattern (19 auto-paired .dlq topics)

Scope

Kafka 19 topics per CLAUDE.md · partitions · retention · compression · key schemas · DLQ auto-pair
Producer libraries: Python + Node · PII hash + GPS fuzz before publish · schema-registry compliance
Consumer groups: data-lake-ingester (→ Iceberg Bronze) · alerting (→ anomaly detect) · kpi-snapshot · rag-embed · case-ai-pipeline
Bronze → Silver → Gold dbt models: silver freshness ≤15min · gold ≤60min · tests enforced (fail blocks promote)
TimescaleDB hypertables: metric fact tables for Dashboard · time-series for SLA measures
pgvector HNSW: m=16 ef_construction=64 · UGC/reviews/KB embeddings
Neo4j schema: actor/service/case/signal nodes · relationships from KB + runtime events
Airflow DAGs: dag_bronze_to_silver (15min) · dag_silver_to_gold (1hr) · dag_kpi_snapshot (daily 06:00) · dag_nlp_embedding · dag_case_ai_pipeline (07:00) · dag_feature_engineering (6hr)
PDPA enforcement: producer-side hash + fuzz · consumer-side k-anon · retention per topic (payment 7yr · location 24h · clickstream 3d)
Monitoring: Grafana · consumer lag · DLQ depth · dbt test pass rate · anomaly alerts

ER Diagram layered architecture

Data flow · producer → Kafka → lake → warehouse → runtime consumers

Field Mapping — Kafka Topics 19 topics per CLAUDE.md

Topic	Parts	Retention	Producer (A)	Consumer Target	Binding	Notes
ptt.booking.events	12	30d	booking-svc	Bronze · Cases (signal)	mapped-not-bound	key: user_id · snappy
ptt.payment.transactions	24	7yr	payment-svc	WORM S3 · Wallet	mapped-not-bound	compress=none · Object Lock
ptt.payment.wallet-events	12	90d	wallet-svc	Silver · user_wallet_summary	mapped-not-bound	—
ptt.affiliate.clicks	8	30d	affiliate-svc	Bronze · attribution	mapped-not-bound	key: partner_id
ptt.affiliate.conversions	8	365d	affiliate-svc	Gold · affiliate_mart	mapped-not-bound	WHT 3% calc
ptt.behavior.clickstream	48	3d	behavior-svc	Bronze · session_analytics	mapped-not-bound	lz4 compression
ptt.behavior.search-events	24	14d	behavior-svc	Silver · search_intent	mapped-not-bound	—
ptt.location.pings	48	24h	location-svc (fuzz ±50m)	Silver · h3_heatmap (k≥5)	placeholder	PDPA: consent_location=true required
ptt.location.poi-visits	12	90d	location-svc	Gold · visit_frequency	placeholder	k-anon ≥5 before materialize
ptt.ugc.content-created	8	∞ compact	ugc-svc	rag-embed → pgvector	mapped-not-bound	key: poi_id
ptt.reviews.submitted	8	∞ compact	reviews-svc	rag-embed + sentiment	mapped-not-bound	—
ptt.knowledge.ingest-requests	4	7d	ingest-svc	dag_notebooklm_ingestion	placeholder	PDF/Audio/Video → Whisper
ptt.ai.llm-interactions	8	90d	ai-svc	Silver · RAGAS eval	mapped-not-bound	Faithfulness log
ptt.external.signals	4	30d	external-signals	Bronze · multi_source_intake	mapped-not-bound	key: location_id
ptt.metrics.kpi-snapshots	4	365d	kpi-emitter	TimescaleDB · dashboard feed	mapped-not-bound	daily dag_kpi_snapshot
ptt.feedback.nps	4	365d	feedback-svc	Silver · nps_trend	mapped-not-bound	—
ptt.audit.trail	8	7yr	audit-emitter	WORM · compress=none	placeholder	Admin plane audit chain
ptt.notifications.outbound	6	7d	notifier-svc	PagerDuty/Slack/LINE	placeholder	alerting_model escalation
ptt.booking.status-changes	12	7d	booking-svc	Silver · booking_lifecycle	mapped-not-bound	key: booking_id

Field Mapping — city_health_contract B-declared metric pipeline dependencies

Source: city_health_contract.panels.*.metrics[*]. Every metric carries a_owned_dependency + refresh_cadence_request.

B Metric	A Signal Source	Refresh	Binding	Notes
vital_sign.fever	AI sentiment from reviews + UGC	15min	mapped-not-bound	Threshold <7/10 · verbatim CONCEPT-005
vital_sign.heartbeat	Booking completion rate · silver mart	15min	mapped-not-bound	Threshold <60% · verbatim CONCEPT-005
vital_sign.blood_density	Spend velocity anomaly · Z-score	1hr	mapped-not-bound	Threshold −20% · verbatim
vital_sign.respiration	External AQI API	1hr	placeholder	Threshold AQI >100 · needs API key
vital_sign.pressure	Civic report aggregation	6hr	placeholder	Threshold +30% volume · verbatim
zone_uplift	H3-cell GMV uplift · k-anon ≥5	daily	placeholder	Proposed · Claude-derived formula · requires_human_review
pulse	Real-time activity rate	1min	mapped-not-bound	Proposed · TimescaleDB continuous aggregate

API Sketch pipeline ops + query

GET/api/ops/pipeline/status

Pipeline health overview · silver/gold freshness · DAG status · consumer lag · DLQ depth

// response { "kafka": { "lag_p99_ms": 145, "dlq_depth": { "ptt.booking.events.dlq": 2 } }, "dbt": { "silver_freshness_min": 12, "gold_freshness_min": 48, "last_test_fail": null }, "airflow": { "dag_bronze_to_silver": "success", "dag_silver_to_gold": "running" }, "health": "green" }

GET/api/metrics/query?metric=gmv_daily&range=30d&dims=tenant

Dashboard metric query · hits TimescaleDB hypertable

// response { "metric": "gmv_daily", "unit": "THB", "series": [ { "ts": "2026-04-17", "value": 1520000, "tenant": "pty-zeroth" } ], "freshness_min": 14, "k_anon_applied": true }

POST/api/ops/backfill

Admin-only backfill trigger · ranges date + metric · Airflow DAG

Sequence Flow PDPA-compliant ingest

Producer → Kafka → Consumer · hash + fuzz + k-anon

Dependencies

Upstream (infra)

Confluent Cloud Kafka
GCS + Iceberg
BigQuery + dbt
Cloud Composer 2 (Airflow)
Postgres + TimescaleDB + pgvector
Neo4j · Redis Cluster
Schema Registry

Downstream (enables)

Dashboard Runtime
Cases AI pipeline
Wizard AI suggestions (RAG)
Alerting anomaly detect
City Health Dashboard
Reporting PDF generation

Approval Gates

gate-pdpa-producer PII hash + GPS fuzz verified at every producer · CI check
gate-kanon-consumer k≥5 enforced before materialize · penetration test
gate-retention Retention policy per topic matches CLAUDE.md · annual audit
gate-dbt-tests Silver/Gold tests pass before promote · CI/CD gate
gate-worm-7yr Payment + audit topics use S3 Object Lock · non-deletable
gate-schema-change Breaking schema change requires producer + consumer migration plan + DEC

Risks

PII leak via unhashed producer

High

Schema Registry enforces hash fields · CI test · pentest · alert on raw PII detected in Bronze

k-anon violated on small H3 cell

High

Consumer-side filter k≥5 · pentest · unit test per dbt model · suppression card on UI

Consumer lag causes stale dashboard

Med

Lag alerts · horizontal autoscale consumer group · dashboard shows freshness_min

dbt_test fail but gold materializes anyway

High

Fail blocks promote · hard rule · CI/CD enforces · alert on skip

7yr retention accidentally compressed (data loss)

High

compress=none for payment + audit · S3 Object Lock · annual verify

DLQ grows unbounded

Med

DLQ 14d retention · depth alert · manual replay tooling · RCA on large DLQ

Definition of Done

All 19 Kafka topics created with correct partitions / retention / compression / DLQ
Producer libraries enforce PII hash + GPS fuzz (CI test passes)
Consumer k-anon filter tested ≥5 on all spatial consumers
Bronze/Silver/Gold dbt models pass tests with ≤15min freshness
6 Airflow DAGs scheduled and green for 7 consecutive days
TimescaleDB hypertables + pgvector HNSW + Neo4j schemas deployed
Retention policy verified (7yr WORM for payment + audit)
Grafana dashboards show consumer lag · DLQ · freshness · anomaly
Penetration test on PDPA compliance passed
All 6 approval gates signed off

Deferred

Multi-region replication (single region at launch)
Change-data-capture from legacy systems (future migration layer)
Schema evolution automation (manual migration plan OK at v1)
Tiered storage (hot/cold) — all hot at launch, tier later
Cross-tenant data marketplace (out of platform scope)

Signal Pipeline 09 · Kafka · dbt · Timescale · pgvector · Neo4j · PDPA at producer · k-anon at consumer

Context

A-owned Runtime Boundary

B owns (read-only for A)

A owns (this runtime)

Scope

ER Diagram layered architecture

Field Mapping — Kafka Topics 19 topics per CLAUDE.md

Field Mapping — city_health_contract B-declared metric pipeline dependencies

API Sketch pipeline ops + query

Sequence Flow PDPA-compliant ingest

Dependencies

Upstream (infra)

Downstream (enables)

Approval Gates

Risks

Definition of Done

Deferred