Data Integration And Consistency

Goal: ensure that a POST /payments endpoint can be safely retried and that an event is reliably emitted.

-- Tables
CREATE TABLE payments (
  id UUID PRIMARY KEY,
  external_id TEXT UNIQUE NOT NULL, -- idempotency key from client
  amount_cents INT NOT NULL,
  status TEXT NOT NULL DEFAULT 'pending'
);

CREATE TABLE outbox (
  id UUID PRIMARY KEY,
  aggregate_type TEXT NOT NULL,
  aggregate_id UUID NOT NULL,
  event_type TEXT NOT NULL,
  payload JSONB NOT NULL,
  created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
  delivered_at TIMESTAMPTZ
);

// Pseudocode
function createPayment(req) {
  const key = req.headers['Idempotency-Key'];
  const body = req.body;

  db.transaction(() => {
    // Try insert; on conflict, return existing
    const payment = db.query(`
      INSERT INTO payments (id, external_id, amount_cents, status)
      VALUES (gen_random_uuid(), $1, $2, 'created')
      ON CONFLICT (external_id) DO UPDATE SET amount_cents = EXCLUDED.amount_cents
      RETURNING *;
    `, [key, body.amount_cents]);

    // Write outbox atomically with business change
    db.exec(`
      INSERT INTO outbox (id, aggregate_type, aggregate_id, event_type, payload)
      VALUES (gen_random_uuid(), 'payment', $1, 'PaymentCreated', $2);
    `, [payment.id, JSON.stringify(payment)]);
  });

  return {status: 201};
}

// A worker periodically relays undelivered outbox rows
function outboxRelay() {
  const rows = db.query(
    "SELECT * FROM outbox WHERE delivered_at IS NULL ORDER BY created_at LIMIT 100"
  );
  for (const row of rows) {
    try {
      http.post(EVENT_BUS_URL, row.payload);
      db.exec("UPDATE outbox SET delivered_at = now() WHERE id = $1", [row.id]);
    } catch (e) {
      // leave for retry; use exponential backoff in scheduling
    }
  }
}

Why it works: The business write and the event are in the same DB transaction. Retries are safe due to external_id uniqueness.

Scenario: Create an order and reserve inventory in the same database.

BEGIN;
INSERT INTO orders (id, user_id, total_cents, status)
VALUES (gen_random_uuid(), $1, $2, 'created');

UPDATE inventory SET reserved = reserved + $3
WHERE sku = $4 AND (stock - reserved) >= $3;

-- Ensure that exactly one row was updated
IF NOT FOUND THEN ROLLBACK; RAISE 'Insufficient stock'; END IF;

COMMIT;

Same-DB updates belong in one ACID transaction. If inventory lived in another service, use a saga: mark order as pending, emit an event, and confirm when reservation succeeds.

Goal: Deliver partner webhooks with at-least-once semantics and idempotency.

// Sender keeps retrying with backoff; includes signature and idempotency key
POST /webhook { event_id, type, data }
Headers: X-Idempotency-Key: event_id

// Receiver stores deliveries to dedupe
CREATE TABLE webhook_receipts (
  idempotency_key TEXT PRIMARY KEY,
  received_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

function handleWebhook(req) {
  const key = req.headers['X-Idempotency-Key'];
  db.transaction(() => {
    db.exec(
      "INSERT INTO webhook_receipts (idempotency_key) VALUES ($1) ON CONFLICT DO NOTHING",
      [key]
    );
    // Process business logic only if first time
    const inserted = db.rowCount == 1;
    if (inserted) process(req.body);
  });
  return 200; // 2xx signals success; otherwise sender retries
}

Tip: Use exponential backoff with jitter on the sender. Cap retries and route to a dead-letter log if repeatedly failing.

Goal: Keep a denormalized read table fresh within 5 seconds for fast queries.

-- Events table fed by outbox or bus
CREATE TABLE events (
  id UUID PRIMARY KEY,
  type TEXT NOT NULL,
  payload JSONB NOT NULL,
  created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

-- Read model
CREATE TABLE order_summary (
  order_id UUID PRIMARY KEY,
  total_cents INT,
  item_count INT,
  last_event_at TIMESTAMPTZ
);

-- Projector
function project() {
  const evts = db.query("SELECT * FROM events WHERE id > $lastSeen ORDER BY created_at LIMIT 500");
  for (const e of evts) {
    switch (e.type) {
      case 'OrderCreated':
        db.exec("INSERT INTO order_summary(order_id, total_cents, item_count, last_event_at) VALUES ($1,$2,$3,$4) ON CONFLICT (order_id) DO UPDATE SET total_cents=$2, item_count=$3, last_event_at=$4", [e.payload.id, e.payload.total_cents, e.payload.item_count, e.created_at]);
        break;
      // handle other events
    }
  }
}

Communicate freshness in the API response, e.g., { data, maxLagSeconds: 3 } so clients understand consistency.

Goal: Add new field customer_tier without breaking older clients.

-- Add with default and allow nulls initially
ALTER TABLE customers ADD COLUMN customer_tier TEXT DEFAULT 'standard';

-- Backfill gradually
UPDATE customers SET customer_tier = 'standard' WHERE customer_tier IS NULL;

-- API: treat field as optional; do not remove old fields until clients migrate

Rules: Prefer additive changes, keep old fields for a deprecation window, and version message formats if you must remove or change semantics.

• Duplicate records: Check for missing unique index; review retry logs for the same request ID.
• Out-of-sync read model: Inspect projector offset vs latest event timestamp; look for poison events.
• Webhook storm: Verify the receiver’s 2xx response; ensure idempotency keys are stable across retries.
• Phantom missing links: Run left-join reconciliation queries to identify drift and backfill.

• POST /orders is idempotent using a client-supplied key.
• Order creation and outbox write are in one transaction.
• A relay worker delivers outbox events to an /partner-hook endpoint with exponential backoff and jitter.
• The partner endpoint is idempotent using the event_id as a key.
• Read model order_summary is updated by a projector from events.
• Provide a reconciliation job that detects orders without a corresponding order_summary row.

1. Define schema (orders, outbox, events, order_summary, webhook_receipts).
2. Implement POST /orders with upsert and outbox write.
3. Create the outbox relay with retry policy and dead-letter log.
4. Implement the partner receiver with idempotency dedupe.
5. Add the projector and expose GET /orders/{id}/summary with freshness metadata.
6. Write a reconciliation script to report and fix drift.