งานถูกบันทึกแล้ว แต่ Agent ยังส่งงานต่อไม่ถึง? ออกแบบ Transactional Outbox

"""Educational SQLite transactional-outbox demo. Python 3.12+, stdlib only.

The provider is an in-memory mock. This is not a production delivery system.
"""
from __future__ import annotations

import argparse
import hashlib
import json
import re
import sqlite3
import tempfile
import uuid
from pathlib import Path
from typing import Any


class IdentityConflict(ValueError):
    """A stable logical ID was reused for different content."""


class MockProvider:
    """Same-process idempotency illustration; state vanishes on restart."""

    def __init__(self) -> None:
        self.receipts: dict[str, tuple[str, str]] = {}

    def send(self, event_id: str, payload_hash: str, *, lose_response: bool = False) -> str:
        previous = self.receipts.get(event_id)
        if previous and previous[0] != payload_hash:
            raise IdentityConflict("provider key reused with different payload")
        receipt = previous[1] if previous else f"mock-{uuid.uuid4().hex[:12]}"
        self.receipts[event_id] = (payload_hash, receipt)
        if lose_response:
            raise TimeoutError("mock accepted; response was lost")
        return receipt


def connect(path: str | Path) -> sqlite3.Connection:
    db = sqlite3.connect(path, autocommit=True)
    db.execute("PRAGMA foreign_keys = ON")
    db.executescript("""
        CREATE TABLE IF NOT EXISTS review_requests (
            review_id TEXT PRIMARY KEY,
            payload_hash TEXT NOT NULL,
            body_json TEXT NOT NULL
        );
        CREATE TABLE IF NOT EXISTS outbox_events (
            event_id TEXT PRIMARY KEY,
            review_id TEXT NOT NULL UNIQUE REFERENCES review_requests(review_id),
            payload_hash TEXT NOT NULL,
            body_json TEXT NOT NULL,
            state TEXT NOT NULL CHECK(state IN ('PENDING','UNKNOWN','SENT')),
            provider_receipt TEXT
        );
    """)
    return db


def canonical_payload(payload: dict[str, Any]) -> tuple[str, str]:
    if type(payload) is not dict or set(payload) != {"asset", "decision"}:
        raise ValueError("expected exactly asset and decision")
    if not isinstance(payload["asset"], str) or not re.fullmatch(r"[A-Za-z0-9_-]{1,80}", payload["asset"]):
        raise ValueError("asset must be a bounded synthetic reference")
    if payload["decision"] not in ("approve_edit", "request_changes"):
        raise ValueError("unsupported decision")
    body = json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False)
    return body, hashlib.sha256(body.encode("utf-8")).hexdigest()


def record_review(db: sqlite3.Connection, review_id: str, payload: dict[str, Any], *, fail_after_business: bool = False) -> str:
    """Commit business row and notification intent together; never sends."""
    if not isinstance(review_id, str) or not re.fullmatch(r"[A-Za-z0-9_-]{1,80}", review_id):
        raise ValueError("review_id must be a stable bounded identifier")
    body, digest = canonical_payload(payload)
    event_id = f"review:{review_id}"
    db.execute("BEGIN IMMEDIATE")
    try:
        old = db.execute("SELECT payload_hash FROM review_requests WHERE review_id=?", (review_id,)).fetchone()
        if old:
            if old[0] != digest:
                raise IdentityConflict("review ID already binds different content")
            db.execute("ROLLBACK")
            return event_id
        db.execute("INSERT INTO review_requests VALUES (?, ?, ?)", (review_id, digest, body))
        if fail_after_business:
            raise RuntimeError("injected failure before outbox insert")
        db.execute("""INSERT INTO outbox_events
            (event_id,review_id,payload_hash,body_json,state)
            VALUES (?,?,?,?, 'PENDING')""", (event_id, review_id, digest, body))
        db.execute("COMMIT")
        return event_id
    except BaseException:
        if db.in_transaction:
            db.execute("ROLLBACK")
        raise


def dispatch_one(db: sqlite3.Connection, provider: MockProvider, *, lose_response: bool = False) -> str:
    """One dispatcher invocation; UNKNOWN is a stop fence, never auto-replayed."""
    row = db.execute("SELECT event_id,payload_hash,body_json FROM outbox_events WHERE state='PENDING' ORDER BY event_id LIMIT 1").fetchone()
    if row is None:
        return "NOTHING_PENDING"
    event_id, digest, _body = row  # SELECT has ended; no transaction spans provider I/O.
    db.execute("BEGIN IMMEDIATE")
    try:
        changed = db.execute("UPDATE outbox_events SET state='UNKNOWN' WHERE event_id=? AND payload_hash=? AND state='PENDING'", (event_id, digest)).rowcount
        if changed != 1:
            db.execute("ROLLBACK")
            return "NOTHING_PENDING"
        db.execute("COMMIT")  # Persist uncertainty fence before the provider call.
    except BaseException:
        if db.in_transaction:
            db.execute("ROLLBACK")
        raise  # Stop before any provider call; do not infer commit outcome.
    try:
        receipt = provider.send(event_id, digest, lose_response=lose_response)
    except Exception:
        return "UNKNOWN"
    db.execute("BEGIN IMMEDIATE")
    try:
        changed = db.execute("UPDATE outbox_events SET state='SENT',provider_receipt=? WHERE event_id=? AND payload_hash=? AND state='UNKNOWN'", (receipt, event_id, digest)).rowcount
        if changed != 1:
            db.execute("ROLLBACK")
            return "UNKNOWN"
        db.execute("COMMIT")
    except BaseException:
        if db.in_transaction:
            db.execute("ROLLBACK")
        return "UNKNOWN"
    return "SENT"


def demo() -> None:
    with tempfile.TemporaryDirectory(prefix="hm-outbox-") as folder:
        db = connect(Path(folder) / "lesson.sqlite3")
        event_id = record_review(db, "synthetic-review-001", {"asset": "sample-video-7", "decision": "approve_edit"})
        print(f"locally_recorded={event_id}")
        print(f"dispatch={dispatch_one(db, MockProvider())}")
        print(f"state={db.execute('SELECT state FROM outbox_events').fetchone()[0]}")
        db.close()


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--demo", action="store_true", help="write a synthetic row to a temporary database")
    if parser.parse_args().demo:
        demo()
    else:
        parser.print_help()

"""Authored invariant tests. Status: NOT RUN."""
import sqlite3
import tempfile
import unittest
from pathlib import Path

from outbox_mock import IdentityConflict, MockProvider, connect, dispatch_one, record_review


class OutboxInvariants(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory()
        self.db = connect(Path(self.temp.name) / "test.sqlite3")
        self.provider = MockProvider()

    def tearDown(self):
        self.db.close()
        self.temp.cleanup()

    def test_business_change_and_intent_commit_as_pair(self):
        record_review(self.db, "r1", {"asset": "sample-video-7", "decision": "approve_edit"})
        self.assertEqual(self.db.execute("SELECT count(*) FROM review_requests").fetchone()[0], 1)
        self.assertEqual(self.db.execute("SELECT state FROM outbox_events").fetchone()[0], "PENDING")

    def test_failure_between_inserts_rolls_back_both(self):
        with self.assertRaises(RuntimeError):
            record_review(self.db, "r1", {"asset": "sample-video-7", "decision": "approve_edit"}, fail_after_business=True)
        self.assertEqual(self.db.execute("SELECT count(*) FROM review_requests").fetchone()[0], 0)
        self.assertEqual(self.db.execute("SELECT count(*) FROM outbox_events").fetchone()[0], 0)

    def test_replay_same_identity_is_stable(self):
        first = record_review(self.db, "r1", {"asset": "sample-video-7", "decision": "approve_edit"})
        second = record_review(self.db, "r1", {"asset": "sample-video-7", "decision": "approve_edit"})
        self.assertEqual(first, second)
        self.assertEqual(self.db.execute("SELECT count(*) FROM outbox_events").fetchone()[0], 1)

    def test_conflicting_identity_fails_before_second_send(self):
        record_review(self.db, "r1", {"asset": "sample-video-7", "decision": "approve_edit"})
        with self.assertRaises(IdentityConflict):
            record_review(self.db, "r1", {"asset": "sample-video-7", "decision": "request_changes"})
        self.assertEqual(self.provider.receipts, {})

    def test_definite_acceptance_records_receipt(self):
        record_review(self.db, "r1", {"asset": "sample-video-7", "decision": "approve_edit"})
        self.assertEqual(dispatch_one(self.db, self.provider), "SENT")
        self.assertTrue(self.db.execute("SELECT provider_receipt FROM outbox_events").fetchone()[0])

    def test_accepted_but_lost_response_stays_unknown_and_is_skipped(self):
        record_review(self.db, "r1", {"asset": "sample-video-7", "decision": "approve_edit"})
        self.assertEqual(dispatch_one(self.db, self.provider, lose_response=True), "UNKNOWN")
        self.assertEqual(self.db.execute("SELECT state FROM outbox_events").fetchone()[0], "UNKNOWN")
        before = dict(self.provider.receipts)
        self.assertEqual(dispatch_one(self.db, self.provider), "NOTHING_PENDING")
        self.assertEqual(self.provider.receipts, before)

    def test_sqlite_transaction_does_not_span_provider_call(self):
        class InspectingProvider(MockProvider):
            def send(inner, *args, **kwargs):
                self.assertFalse(self.db.in_transaction)
                return super().send(*args, **kwargs)
        record_review(self.db, "r1", {"asset": "sample-video-7", "decision": "approve_edit"})
        self.assertEqual(dispatch_one(self.db, InspectingProvider()), "SENT")


    def test_unknown_row_change_cannot_report_sent(self):
        class ChangingProvider(MockProvider):
            def send(inner, *args, **kwargs):
                receipt = super().send(*args, **kwargs)
                self.db.execute("UPDATE outbox_events SET payload_hash='changed' WHERE state='UNKNOWN'")
                return receipt
        record_review(self.db, "r1", {"asset": "sample-video-7", "decision": "approve_edit"})
        self.assertEqual(dispatch_one(self.db, ChangingProvider()), "UNKNOWN")
        self.assertIsNone(self.db.execute("SELECT provider_receipt FROM outbox_events").fetchone()[0])

    def test_invalid_shape_stops_before_writes(self):
        with self.assertRaises(ValueError):
            record_review(self.db, "r1", {"asset": "sample-video-7", "decision": "approve_edit", "extra": 1})
        self.assertEqual(self.db.execute("SELECT count(*) FROM review_requests").fetchone()[0], 0)

if __name__ == "__main__":
    unittest.main()

# Transactional Outbox mock

This is an educational, local-only Python 3.12+ example using the standard library. It records a synthetic video-review decision and a notification intent in the same SQLite transaction, then demonstrates a separate dispatcher and an in-memory mock provider. It contains no network client, credentials, external API, scheduler, deployment code, or real review/video data. The sample deliberately has no UNKNOWN reconciliation or reset operation.

## Environment and permissions

- Python 3.12 or newer; no packages to install and no environment variables.
- Running `--demo` creates a temporary SQLite database under the operating system's temporary directory and removes it on exit. Running tests also creates a temporary database. Neither should be run against production data.
- The mock's receipt map exists only in process memory. It does not prove provider idempotency after a process restart.
- No API, hosting, or usage cost is incurred by this mock. A future sandbox or live provider would require separate authorization, credentials, privacy review, provider-specific idempotency/status semantics, and cost/latency measurement.

## Run

From this folder, an owner may choose to run:

```powershell
python code/outbox_mock.py --demo
```

Illustrative output (not observed in this task):

```text
locally_recorded=review:synthetic-review-001
dispatch=SENT
state=SENT
```

To run authored invariants from this run folder:

```powershell
python -m unittest discover -s code -p "test_outbox_mock.py" -v
```

Tests were authored but NOT RUN. Import, compile, database creation, and execution were intentionally not performed while preparing this package.

## Mock, sandbox, live

**Mock** means deterministic local behavior with synthetic data and no network. **Sandbox** means a provider-owned test environment that may still transmit data and create provider-side effects; it needs explicit credentials, permissions, and cleanup rules. **Live** means real provider actions, real data, and possible cost. This package implements only Mock. Do not read its output as evidence for Sandbox or Live delivery guarantees.