DB更新とイベント配信を分離するtransactional outbox設計｜SQLite・Python・冪等性

DBの部品recordを更新
  ↓
外部APIまたはmessage brokerへeventを送信

API request
  ↓
同一DB transaction
  ├─ 業務tableを更新
  └─ outbox tableへeventをINSERT
  ↓ commit
relayが未配信eventを取得
  ↓
外部systemへ送信
  ↓
配信済みstatusを記録

{
  "event_id": "evt-01K6W3P8Y7",
  "event_type": "asset.updated",
  "schema_version": 1,
  "aggregate_type": "asset",
  "aggregate_id": "asset-042",
  "aggregate_version": 8,
  "occurred_at": "2026-10-07T01:20:00Z",
  "data": {
    "name": "Hydraulic unit B",
    "material": "S45C"
  }
}

CREATE TABLE assets (
    asset_id   TEXT PRIMARY KEY NOT NULL,
    name       TEXT NOT NULL,
    material   TEXT NOT NULL,
    version    INTEGER NOT NULL CHECK (version > 0),
    updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
);

CREATE TABLE outbox_events (
    event_id          TEXT PRIMARY KEY NOT NULL,
    aggregate_type    TEXT NOT NULL,
    aggregate_id      TEXT NOT NULL,
    aggregate_version INTEGER NOT NULL CHECK (aggregate_version > 0),
    event_type        TEXT NOT NULL,
    schema_version    INTEGER NOT NULL CHECK (schema_version > 0),
    payload_json      TEXT NOT NULL,
    status            TEXT NOT NULL DEFAULT 'pending'
        CHECK (status IN ('pending', 'published', 'failed')),
    attempt_count     INTEGER NOT NULL DEFAULT 0,
    available_at      TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
    created_at        TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
    published_at      TEXT,
    last_error        TEXT,
    UNIQUE (aggregate_type, aggregate_id, aggregate_version)
);

CREATE INDEX idx_outbox_delivery
ON outbox_events (status, available_at, created_at);

import json
import sqlite3
from datetime import datetime, timezone


def update_asset(
    conn: sqlite3.Connection,
    *,
    event_id: str,
    asset_id: str,
    expected_version: int,
    name: str,
    material: str,
) -> None:
    new_version = expected_version + 1
    event = {
        "event_id": event_id,
        "event_type": "asset.updated",
        "schema_version": 1,
        "aggregate_type": "asset",
        "aggregate_id": asset_id,
        "aggregate_version": new_version,
        "occurred_at": datetime.now(timezone.utc)
        .isoformat()
        .replace("+00:00", "Z"),
        "data": {"name": name, "material": material},
    }
    payload = json.dumps(
        event,
        ensure_ascii=False,
        separators=(",", ":"),
        allow_nan=False,
    )

    with conn:
        cursor = conn.execute(
            """
            UPDATE assets
            SET name = ?, material = ?, version = ?,
                updated_at = CURRENT_TIMESTAMP
            WHERE asset_id = ? AND version = ?
            """,
            (name, material, new_version, asset_id, expected_version),
        )
        if cursor.rowcount != 1:
            raise ValueError("対象がないか、versionが競合しています")

        conn.execute(
            """
            INSERT INTO outbox_events (
                event_id, aggregate_type, aggregate_id,
                aggregate_version, event_type,
                schema_version, payload_json
            ) VALUES (?, ?, ?, ?, ?, ?, ?)
            """,
            (
                event_id, "asset", asset_id, new_version,
                "asset.updated", 1, payload,
            ),
        )

def relay_once(conn, publish) -> bool:
    row = conn.execute(
        """
        SELECT event_id, payload_json
        FROM outbox_events
        WHERE status = 'pending'
          AND available_at <= CURRENT_TIMESTAMP
        ORDER BY created_at, event_id
        LIMIT 1
        """
    ).fetchone()
    if row is None:
        return False

    event_id, payload = row
    try:
        publish(event_id, payload)
    except Exception as exc:
        with conn:
            conn.execute(
                """
                UPDATE outbox_events
                SET attempt_count = attempt_count + 1,
                    available_at = datetime('now', '+1 minute'),
                    last_error = ?
                WHERE event_id = ? AND status = 'pending'
                """,
                (str(exc)[:500], event_id),
            )
        return False

    with conn:
        conn.execute(
            """
            UPDATE outbox_events
            SET status = 'published',
                attempt_count = attempt_count + 1,
                published_at = CURRENT_TIMESTAMP,
                last_error = NULL
            WHERE event_id = ? AND status = 'pending'
            """,
            (event_id,),
        )
    return True

CREATE TABLE processed_events (
    event_id     TEXT PRIMARY KEY NOT NULL,
    processed_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
);

def consume_once(conn, event: dict) -> bool:
    with conn:
        cursor = conn.execute(
            """
            INSERT INTO processed_events (event_id)
            VALUES (?)
            ON CONFLICT (event_id) DO NOTHING
            """,
            (event["event_id"],),
        )
        if cursor.rowcount == 0:
            return False

        apply_asset_update(conn, event)
    return True