Per-database ongoing burden (estimate for a 5-engineer team):
  Backup configuration:          2 hours/week monitoring + incident response
  Monitoring dashboard:          1 hour/week reviewing metrics
  Failover runbook:              4 hours to write, 2 hours/quarter to drill
  On-call expertise:             1–2 engineers must be debuggable at 3 AM
  Security patching:             2–4 hours per major version upgrade
  Connection pool management:    Separate pool per service × database
  Schema migration tooling:      Separate migration framework per database
  ─────────────────────────────────────────────────────────────────────
  Total: ~1–2 engineer-days/month per additional database

┌─────────────────────────────────────────────────────────────────┐
│ Postgres (Primary Source of Truth)                              │
│ All writes originate here. ACID transactions. FK constraints.   │
│ Extensions: pgvector, JSONB, PostGIS as the first tier          │
└──────────────┬──────────────┬──────────────┬────────────────────┘
               │              │              │
         (CDC)  │        (CDC)  │        (CDN) │
               ▼              ▼              ▼
┌─────────────────┐  ┌─────────────────┐  ┌─────────────────────┐
│ Redis           │  │ Elasticsearch   │  │ S3 / Object Store   │
│ Cache, session, │  │ Full-text       │  │ Media, exports,     │
│ rate limiting   │  │ search, logs    │  │ backups             │
└─────────────────┘  └─────────────────┘  └─────────────────────┘

// ❌ Broken: dual-write — writes to two stores in the same code path
async function createProduct(product: Product) {
  // Step 1: write to Postgres
  await pg.query('INSERT INTO products (id, name, price) VALUES ($1, $2, $3)',
    [product.id, product.name, product.price])

  // Step 2: write to Elasticsearch — what happens if this fails?
  await elasticsearch.index({ index: 'products', id: product.id, body: product })
  // If Elasticsearch is down:
  //   - Postgres has the product → source of truth has it
  //   - Elasticsearch doesn't → search returns nothing
  //   - Silent inconsistency — no error raised to the user
  //   - Data is now permanently out of sync unless you manually reconcile
}

# Debezium Postgres connector configuration
connector.class: io.debezium.connector.postgresql.PostgresConnector
database.hostname: postgres-primary.internal
database.port: 5432
database.user: debezium_user
database.password: ${SECRET:db-password}  # never inline credentials
database.dbname: production
plugin.name: pgoutput   # Postgres logical replication plugin

# Capture all tables in the public schema
table.include.list: public.products, public.users, public.orders

# Outbox pattern: route through outbox table for exactly-once semantics
transforms: outbox
transforms.outbox.type: io.debezium.transforms.outbox.EventRouter
transforms.outbox.route.by.field: aggregate_type

// Kafka consumer — Elasticsearch sync
const consumer = kafka.consumer({ groupId: 'elasticsearch-sync' })
await consumer.subscribe({ topic: 'postgres.public.products' })

await consumer.run({
  eachMessage: async ({ message }) => {
    const event = JSON.parse(message.value!.toString())

    if (event.op === 'c' || event.op === 'u') {  // create or update
      await elasticsearch.index({
        index: 'products',
        id: event.after.id,
        body: {
          name: event.after.name,
          price: event.after.price,
          category: event.after.category,
          // Mask PII before indexing into secondary stores
          // — Debezium transform handles this at the connector layer
        }
      })
    } else if (event.op === 'd') {  // delete
      await elasticsearch.delete({ index: 'products', id: event.before.id })
    }
  }
})

// Problem: user updates their profile → Postgres committed → Elasticsearch still has old data
// User immediately loads their profile page → reads from Elasticsearch → sees old name

// ❌ Broken: reading from Elasticsearch immediately after a write
await pg.query('UPDATE users SET name = $1 WHERE id = $2', ['Alice Smith', userId])
const profile = await elasticsearch.get({ index: 'users', id: userId })
// profile.name is still 'Alice' — CDC lag is 50–500ms

// ✅ Correct option A: read-your-own-writes from Postgres for immediate reads
await pg.query('UPDATE users SET name = $1 WHERE id = $2', ['Alice Smith', userId])
// For the first render after a write, always read from the source of truth
const profile = await pg.query('SELECT * FROM users WHERE id = $1', [userId])
// Subsequent reads (other users, other pages) can use Elasticsearch

// ✅ Correct option B: return the updated object directly from the write response
// No secondary read needed — the application already has the new data
return { ...existingProfile, name: 'Alice Smith' }  // optimistic local update

// ❌ Broken: TTL-only expiry — all keys expire simultaneously after a deploy
// At t=0: deploy invalidates cache for all 10,000 products
// At t=1: 10,000 concurrent requests hit Postgres simultaneously — connection exhaustion

// ✅ Correct: probabilistic early expiry (PER — prevents stampede)
async function getProduct(id: string): Promise<Product> {
  const cached = await redis.get(`product:${id}`)
  if (cached) {
    const { value, expiresAt } = JSON.parse(cached)
    const timeLeft = expiresAt - Date.now()
    const beta = 1.0  // tuning parameter

    // Probabilistic early refresh: start refreshing before expiry
    // Probability increases as expiry approaches — only ONE request triggers refresh
    if (timeLeft > 0 && -beta * Math.log(Math.random()) * 200 < timeLeft) {
      return value  // cache hit — return immediately
    }
    // Falls through to DB fetch and cache refresh
  }

  const product = await pg.query('SELECT * FROM products WHERE id = $1', [id])
  await redis.set(`product:${id}`,
    JSON.stringify({ value: product, expiresAt: Date.now() + 300_000 }),
    { EX: 300 }
  )
  return product
}

Data gravity: the larger your dataset in a given store, the more expensive migration becomes.

Cost of migrating 100M MongoDB documents to PostgreSQL:
  Step 1: Schema design — 1–2 weeks (new normalized schema)
  Step 2: Transformation script — 1 week (denormalized → normalized)
  Step 3: Batch migration — 3–7 days (100M rows at ~5K rows/sec with minimal prod impact)
  Step 4: Dual-write period — 2–4 weeks (run both stores, validate consistency)
  Step 5: Cutover and decommission — 1 week
  ──────────────────────────────────────────────────────────────
  Total: 7–14 weeks of engineering time, 4–8 weeks of dual-write infra cost

# ❌ Broken: Debezium credentials in connector config (readable in logs)
database.password: my-plain-text-password

# ✅ Correct: reference secrets from a secrets manager
database.password: ${file:/opt/kafka/secrets/db-password.properties:password}
# Or use Kafka Connect Secret Registry (Confluent) / AWS Secrets Manager provider

// PII in CDC events — mask at the Debezium Single Message Transform layer
// Never allow raw PII to flow into Kafka topics readable by all consumers
{
  "transforms": "maskPII",
  "transforms.maskPII.type": "org.apache.kafka.connect.transforms.MaskField$Value",
  "transforms.maskPII.fields": "email,phone,ssn",
  "transforms.maskPII.replacement": "***REDACTED***"
}
// Only the GDPR-authorized consumers (erasure service) receive un-masked data via dedicated topic