A tested app: commit a row to PostgreSQL, publish its event to Kafka, verify through MCP

CREATE TABLE orders (
  id          uuid PRIMARY KEY DEFAULT gen_random_uuid(),
  customer_id text NOT NULL,
  total_cents integer NOT NULL CHECK (total_cents >= 0),
  created_at  timestamptz NOT NULL DEFAULT now()
);

CREATE TABLE outbox (
  event_id     uuid PRIMARY KEY DEFAULT gen_random_uuid(),
  aggregate_id uuid NOT NULL,
  topic        text NOT NULL,
  payload      jsonb NOT NULL,
  created_at   timestamptz NOT NULL DEFAULT now(),
  published_at timestamptz
);
CREATE INDEX outbox_unpublished ON outbox (created_at) WHERE published_at IS NULL;

import os, json, psycopg

with psycopg.connect(os.environ["DATABASE_URL"]) as conn:  # sslmode=verify-full&sslrootcert=ca.pem in the URL
    with conn.transaction():
        row = conn.execute(
            "INSERT INTO orders (customer_id, total_cents) VALUES (%s, %s) RETURNING id",
            ("cust-42", 1999),
        ).fetchone()
        conn.execute(
            "INSERT INTO outbox (aggregate_id, topic, payload) VALUES (%s, %s, %s)",
            (row[0], "orders.created", json.dumps({"order_id": str(row[0]), "total_cents": 1999})),
        )
    print("committed", row[0])

import os, json, psycopg
from kafka import KafkaProducer

producer = KafkaProducer(
    bootstrap_servers=os.environ["KAFKA_BOOTSTRAP"],
    security_protocol="SSL",
    ssl_cafile=os.environ["KAFKA_CA"], ssl_certfile=os.environ["KAFKA_CERT"], ssl_keyfile=os.environ["KAFKA_KEY"],
    acks="all",  # idempotent producer support depends on the client library [TO VERIFY]
)

with psycopg.connect(os.environ["DATABASE_URL"]) as conn:
    rows = conn.execute(
        "SELECT event_id, aggregate_id, topic, payload FROM outbox "
        "WHERE published_at IS NULL ORDER BY created_at LIMIT 100 FOR UPDATE SKIP LOCKED"
    ).fetchall()
    for event_id, agg, topic, payload in rows:
        meta = producer.send(topic, key=str(agg).encode(), value=json.dumps({**payload, "event_id": str(event_id)}).encode()).get(timeout=10)
        print("sent", event_id, meta.partition, meta.offset)
        conn.execute("UPDATE outbox SET published_at = now() WHERE event_id = %s", (event_id,))

import os, json, sys
from kafka import KafkaConsumer

want = sys.argv[1]
c = KafkaConsumer("orders.created", bootstrap_servers=os.environ["KAFKA_BOOTSTRAP"],
    security_protocol="SSL", ssl_cafile=os.environ["KAFKA_CA"], ssl_certfile=os.environ["KAFKA_CERT"],
    ssl_keyfile=os.environ["KAFKA_KEY"], auto_offset_reset="earliest", consumer_timeout_ms=10000)
hits = [m for m in c if json.loads(m.value).get("event_id") == want]
print(f"found {len(hits)} record(s)", [(m.partition, m.offset) for m in hits])
sys.exit(0 if hits else 1)

---
name: pg-commit-kafka-verify
description: Insert an order with an outbox event in PostgreSQL, publish it to Kafka, and verify both.
---
# Steps
1. Read DATABASE_URL, KAFKA_BOOTSTRAP, KAFKA_CA, KAFKA_CERT, KAFKA_KEY from the environment. Never print them.
2. Run `python commit_order.py`; capture the order id.
3. Run `python publish_outbox.py`; capture partition and offset.
4. Using the Aiven MCP server, run read-only SQL: orders row exists; outbox.published_at is not null.
5. Run `python verify_topic.py <event_id>`; it must exit 0.
# Stop conditions
- If step 4 shows an unpublished outbox row, do not retry blindly; report it.
- Do not create or delete Aiven services; ask the user.