A newsroom’s wire desk cannot wait for tomorrow’s print run to correct a live story. The correction has to go out the moment it’s confirmed, or the wrong version keeps circulating. A morning paper’s fact-checking process is thorough, but it runs once a day, on a schedule, in a batch. Every RAG pipeline covered so far in this track is built like the morning paper: a job chunks, embeds, and indexes an entire corpus overnight, and the index sits frozen until the next run touches it. That works when the underlying documents change slowly. It breaks the moment your source of truth is a live system, a support ticket queue, a CRM, a product catalog, a wiki someone edits mid-incident, where documents change every few seconds and a stale answer is actively worse than a slow one.

The scale that makes this hard: 4,000 document change events per second sustained across CDC streams, webhooks, and native Kafka producers, feeding a 40 million chunk index built from roughly 6 million source documents, with an 8 second p99 budget from “the source system committed the change” to “a query against the index reflects it.” The batch pipeline this system replaces took 6 hours to fully reindex the same 40 million chunks overnight, every night, whether 2% of the corpus changed that day or 40% of it did. A CRM field edit at 2:47pm has no business waiting for a midnight job.

The naive fixes fail for different reasons. Shrinking the batch window from nightly to every 5 minutes still reprocesses at the granularity of “did this document change,” which means one edited sentence triggers a full re-chunk and re-embed of the entire document, plus every other document that happened to change in that window, most of which touched a handful of fields. Writing straight into the vector index on every single event sounds like the obvious fix for latency, but HNSW graphs are not built for high-frequency single-item deletes, and a live production index that never removes stale vectors accumulates zombie chunks that keep surfacing in results long after the source document is gone. We need to solve for chunk-level diffing so we only touch what actually changed, an index structure that accepts writes instantly without requiring an in-place graph delete, and an explicit consistency contract between how fresh the index is and what a query is allowed to return, all at once, under an 8 second p99 budget.

Requirements and Constraints

Functional Requirements

• Ingest document change events from three source types: CDC streams (Debezium on top of a transactional database), webhooks (push notifications from external SaaS tools), and native Kafka producers (internal services emitting events directly)
• Normalize all three source types into one canonical DocumentEvent schema before anything downstream touches them
• Diff an updated document against its last-committed version at the chunk level, so only chunks whose content actually changed get re-embedded
• Support deletes as first-class events, including partial deletes (some chunks of a document removed, not the whole document)
• Write new and updated vectors into the index without blocking on a full index rebuild
• Deduplicate replayed and out-of-order events using a monotonic per-document version, since Kafka and most CDC connectors only guarantee at-least-once delivery
• Expose a per-shard freshness watermark so a query can either wait for a minimum freshness guarantee or accept results annotated as stale
• Periodically compact the write buffer into the durable, query-optimized index in the background, without pausing ingestion

Non-Functional Requirements

• Latency: staleness p99 under 8 seconds from source commit to index-visible, p50 under 2 seconds
• Throughput: 4,000 document events/second sustained, bursting to 12,000/second during CDC snapshot replays or backfills
• Quality: recall@10 on the merged (buffer + durable) index must stay within 2 points of a fully-compacted index; contradiction between “deleted” and “still returned” must be zero
• Cost: under $7 per million chunk-level upserts processed end to end (diff, embed, index write)
• Capacity: write buffer capped at 200,000 vectors per shard before a merge is forced; durable index shards fit under 6 GB RAM each after compaction

Constraints

• Embeddings are produced by a self-hosted bge-large-en-v1.5 model (1024 dimensions, cosine distance); swapping embedding models is a separate migration project, not something this pipeline handles inline
• Out of scope: the initial historical backfill of the corpus (handled by the batch pipeline from build-a-production-rag-pipeline-with-hybrid-bm25-plus-dense-retrieval), cross-region replication, and full-text search replacement
• Kafka is the assumed backbone; connectors are Debezium for CDC sources and a thin HTTP-to-Kafka bridge for webhooks

High-Level Architecture

The system has five components sitting between the source systems and the query path. A Change Event Ingestor normalizes CDC, webhook, and native events into one canonical DocumentEvent and publishes them to a Kafka topic partitioned by a hash of doc_id. A Content Diff Service re-chunks the updated document and compares each chunk’s content hash against the last-committed hash, emitting only the chunk-level operations that actually changed. An Embedding Worker Pool batches those changed chunks and produces vectors, backpressure-aware so a burst doesn’t take down the pipeline. An Incremental Index Writer holds two tiers per shard: a small, mutable, exact buffer that accepts writes instantly, and a set of immutable, compacted HNSW segments that serve the bulk of queries; a background process merges the buffer into a new segment on a schedule. A Consistency Coordinator tracks a commit watermark per shard and answers the question every query eventually has to ask: how fresh is what I’m about to read.

A change lands as a raw event from whichever source produced it: a row update captured by Debezium, an HTTP callback from a SaaS webhook, or a message from an internal service’s own Kafka producer. The Change Event Ingestor normalizes all three into the same DocumentEvent shape and publishes to a canonical topic, partitioned so that every event for a given doc_id always lands on the same partition and, downstream, the same shard. The Content Diff Service consumes that partition, re-chunks the document, and compares each chunk’s hash against what’s stored in the metadata store, emitting a small list of ChunkOp records, upserts and deletes, instead of forwarding the whole document. Those ops fan out: upserts go to the Embedding Worker Pool for vectorization, deletes go straight to the Incremental Index Writer as tombstones. Once a vector is produced, it lands in that shard’s mutable buffer immediately, and the Consistency Coordinator advances the shard’s watermark. A query hits the Query Router, which merges results from the buffer and the compacted segments and, depending on the caller’s freshness requirement, either blocks briefly for the watermark to catch up or returns results annotated with how stale they are.

The Change Event Ingestor

This component’s job: take three structurally different event shapes and produce one canonical DocumentEvent, published in a way that guarantees per-document ordering downstream.

The non-obvious assumption to correct here: it’s tempting to give each source its own downstream pipeline, since a Debezium payload, a webhook body, and a native Kafka message look nothing alike. That multiplies every downstream component by three for no benefit, and worse, it makes the diffing and versioning logic source-aware when it should be source-blind. The fix is normalizing at the edge, immediately, so everything after this component only ever sees one schema.

# Normalizes CDC, webhook, and native Kafka sources into one canonical DocumentEvent
import hashlib
import json
from dataclasses import dataclass, asdict
from confluent_kafka import Producer

@dataclass
class DocumentEvent:
    doc_id: str
    tenant_id: str
    source: str           # "cdc" | "webhook" | "kafka_native"
    source_version: int   # monotonic version from the source system
    event_type: str        # "upsert" | "delete"
    content: str | None
    content_hash: str | None
    occurred_at_ms: int

def normalize_debezium(payload: dict) -> DocumentEvent:
    after = payload.get("after")
    op = payload["op"]  # "c", "u", "d", "r"
    event_type = "delete" if op == "d" else "upsert"
    content = after.get("body") if after else None
    content_hash = hashlib.sha256(content.encode()).hexdigest() if content else None
    return DocumentEvent(
        doc_id=str(payload["before"]["id"] if op == "d" else after["id"]),
        tenant_id=str(after.get("tenant_id", "default")) if after else "default",
        source="cdc",
        source_version=payload["source"]["ts_ms"],
        event_type=event_type,
        content=content,
        content_hash=content_hash,
        occurred_at_ms=payload["source"]["ts_ms"],
    )

class EventPublisher:
    def __init__(self, brokers: str, topic: str, num_partitions: int = 32):
        self.producer = Producer({"bootstrap.servers": brokers, "linger.ms": 5})
        self.topic = topic
        self.num_partitions = num_partitions

    def publish(self, event: DocumentEvent) -> None:
        partition = int(hashlib.md5(event.doc_id.encode()).hexdigest(), 16) % self.num_partitions
        self.producer.produce(
            self.topic,
            key=event.doc_id.encode(),
            value=json.dumps(asdict(event)).encode(),
            partition=partition,
        )
        self.producer.poll(0)

Hashing doc_id to pick the partition, rather than round-robin or key-less publishing, is what guarantees Kafka’s per-partition ordering actually protects a single document’s edit history. Two edits to the same document, seconds apart, always land on the same partition in commit order.