cdc-streaming-ingestion

What this skill does

# CDC / Streaming Ingestion

## When to Go Real-Time

Scheduled ingestion (see `ingestion-orchestration`) suffices when freshness tolerance is minutes-to-hours. Switch to streaming when:
- Users expect < 10s from source write to retrievable.
- Source is high-throughput (> 100 writes/sec).
- Deletes must propagate quickly (compliance).
- Downstream consumers beyond RAG also need the stream.

## Architecture

```
[Postgres]  --WAL-->  [Debezium]  -->  [Kafka topic: docs.changes]
                                               |
                                 +-------------+-------------+
                                 |                           |
                           [embed worker]              [other consumers]
                                 |
                           [Vector DB upsert]
                                 |
                         [Kafka topic: docs.indexed]   (offset log for observability)
```

Separation of concerns:
- CDC produces change events.
- Kafka is the durable, replayable event log.
- Stream processor embeds and upserts.
- Vector DB is the sink.

## CDC Source: Postgres + Debezium

Postgres logical replication exposes WAL changes. Debezium Connect consumes it and publishes to Kafka.

### Postgres setup

```sql
ALTER SYSTEM SET wal_level = 'logical';
SELECT pg_reload_conf();

CREATE PUBLICATION rag_pub FOR TABLE documents, articles;

CREATE USER debezium WITH REPLICATION LOGIN PASSWORD '***';
GRANT SELECT ON documents, articles TO debezium;

SELECT pg_create_logical_replication_slot('debezium_rag', 'pgoutput');
```

### Debezium connector config

```json
{
  "name": "postgres-rag",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "pg.internal",
    "database.port": "5432",
    "database.user": "debezium",
    "database.password": "${secret}",
    "database.dbname": "app",
    "plugin.name": "pgoutput",
    "slot.name": "debezium_rag",
    "publication.name": "rag_pub",
    "topic.prefix": "rag",
    "table.include.list": "public.documents,public.articles",
    "tombstones.on.delete": "true",
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://schema-registry:8081",
    "heartbeat.interval.ms": "30000",
    "heartbeat.action.query": "INSERT INTO debezium_heartbeat (ts) VALUES (now()) ON CONFLICT DO NOTHING"
  }
}
```

Key points:
- `pgoutput` plugin comes with Postgres 10+; no separate install.
- `tombstones.on.delete=true` sends a null-value record after a delete for log-compacted downstream topics.
- Heartbeats keep the replication slot active on low-traffic tables.
- Avro + schema registry for schema evolution discipline.

Event shape (Debezium `op` codes: `c` create, `u` update, `d` delete, `r` snapshot-read):

```json
{
  "before": { "id": 42, "title": "Old title", ... },
  "after":  { "id": 42, "title": "New title", ... },
  "source": { "lsn": 987654321, "txId": 445, "ts_ms": 1744726800000 },
  "op": "u",
  "ts_ms": 1744726800012
}
```

## Stream Processor: Python + Kafka

For moderate throughput, a Python consumer with `aiokafka` is enough.

```python
import asyncio, json
from aiokafka import AIOKafkaConsumer, AIOKafkaProducer
from openai import AsyncOpenAI
from qdrant_client import AsyncQdrantClient
from qdrant_client.models import PointStruct

oai = AsyncOpenAI()
qdr = AsyncQdrantClient(url="http://qdrant:6333")

async def process():
    consumer = AIOKafkaConsumer(
        "rag.public.documents",
        bootstrap_servers="kafka:9092",
        group_id="rag-embedder",
        enable_auto_commit=False,           # manual commit after successful upsert
        auto_offset_reset="earliest",
    )
    producer = AIOKafkaProducer(bootstrap_servers="kafka:9092",
                                 enable_idempotence=True, acks="all")
    await consumer.start(); await producer.start()

    try:
        async for msg in consumer:
            if msg.value is None:
                # Tombstone -> delete from vector DB
                doc_id = json.loads(msg.key.decode())["id"]
                await qdr.delete(collection_name="kb",
                                 points_selector={"filter": {"must": [
                                     {"key": "doc_id", "match": {"value": doc_id}}
                                 ]}})
                await consumer.commit(); continue

            evt = json.loads(msg.value.decode())
            op = evt["op"]
            doc = evt["after"] if op in ("c", "u", "r") else evt["before"]
            doc_id = doc["id"]

            if op == "d":
                await qdr.delete(collection_name="kb",
                                 points_selector={"filter": {"must": [
                                     {"key": "doc_id", "match": {"value": doc_id}}
                                 ]}})
            else:
                chunks = chunk_doc(doc)
                texts = [c["text"] for c in chunks]
                emb = await oai.embeddings.create(model="text-embedding-3-small", input=texts)
                points = [
                    PointStruct(
                        id=f"{doc_id}-{i}",
                        vector=emb.data[i].embedding,
                        payload={"doc_id": doc_id, "chunk_index": i,
                                 "text": chunks[i]["text"], "lsn": evt["source"]["lsn"]},
                    )
                    for i in range(len(chunks))
                ]
                # Delete old chunks then upsert new — handles chunk count change
                await qdr.delete(collection_name="kb",
                                 points_selector={"filter": {"must": [
                                     {"key": "doc_id", "match": {"value": doc_id}}
                                 ]}})
                await qdr.upsert(collection_name="kb", points=points)

            await producer.send("rag.indexed", key=msg.key,
                                value=json.dumps({"doc_id": doc_id, "op": op}).encode())
            await consumer.commit()
    finally:
        await consumer.stop(); await producer.stop()

asyncio.run(process())
```

- `enable_auto_commit=False` with manual commit after upsert => at-least-once delivery.
- Delete-then-upsert handles chunk count changes on update.
- Emit to `rag.indexed` topic for observability.

## Exactly-Once Semantics

Kafka exactly-once (EOS) requires transactional writes. The vector DB is usually not transactional with Kafka, so you cannot get true EOS end-to-end. Alternative: make upserts idempotent.

```python
# Use stable point IDs: f"{doc_id}-{chunk_index}"
# Use upsert (not insert) in the vector DB.
# Include LSN in payload; on reprocess, skip if stored LSN >= event LSN.

async def idempotent_upsert(point: PointStruct):
    existing = await qdr.retrieve(collection_name="kb", ids=[point.id], with_payload=True)
    if existing and existing[0].payload.get("lsn", 0) >= point.payload["lsn"]:
        return  # already applied a newer or equal version
    await qdr.upsert(collection_name="kb", points=[point])
```

Result: at-least-once delivery + idempotent sink = effectively exactly-once.

## Late-Arriving Updates

CDC can reorder across partitions. Pin all events of a given document to the same partition (by primary key) so updates arrive in order per document.

Debezium does this by default when `message.key` is the table primary key. Verify:

```bash
kafka-console-consumer --topic rag.public.documents --property print.key=true
# Key: {"id":42} for all events about document 42
```

Cross-document ordering is not guaranteed; it doesn't matter for embedding.

## Tombstones (Deletes)

Debezium produces two messages on delete:
1. The delete event (`op=d`, before populated, after null).
2. A tombstone (null value) for log compaction.

Handle both in the consumer:
- On `op=d`: delete from vector DB.
- On null value (tombstone): also delete (