"""发件箱（Outbox）模式：业务数据和“待发送消息”写在同一个数据库事务里，由后台程序投递。

只用标准库（sqlite3）演示，直接运行：python outbox_demo.py
真实项目里把 sqlite 换成 MySQL / PostgreSQL，把 FakeBroker 换成 RabbitMQ / Kafka 客户端。
"""
import json
import sqlite3


class FakeBroker:
    """模拟消息队列：fail_times 次之内发送失败，用来演示重试。"""

    def __init__(self, fail_times: int = 0) -> None:
        self.sent: list[dict] = []
        self.fail_times = fail_times

    def publish(self, topic: str, msg: dict) -> None:
        if self.fail_times > 0:
            self.fail_times -= 1
            raise ConnectionError("broker 暂时不可用")
        self.sent.append({"topic": topic, **msg})


def init(db: sqlite3.Connection) -> None:
    db.executescript("""
    CREATE TABLE orders (id INTEGER PRIMARY KEY, user_id INTEGER NOT NULL, amount INTEGER NOT NULL);
    CREATE TABLE outbox (
      id INTEGER PRIMARY KEY AUTOINCREMENT,
      topic TEXT NOT NULL,
      payload TEXT NOT NULL,
      sent_at TEXT,             -- NULL 表示还没发送
      attempts INTEGER NOT NULL DEFAULT 0
    );
    """)


def create_order(db: sqlite3.Connection, user_id: int, amount: int, fail_after_insert: bool = False) -> int:
    with db:  # 一个事务：订单和消息要么都写入，要么都不写入
        cur = db.execute("INSERT INTO orders (user_id, amount) VALUES (?, ?)", (user_id, amount))
        order_id = cur.lastrowid
        db.execute("INSERT INTO outbox (topic, payload) VALUES (?, ?)",
                   ("order.created", json.dumps({"order_id": order_id, "user_id": user_id, "amount": amount})))
        if fail_after_insert:
            raise RuntimeError("模拟业务异常：事务回滚")
    return order_id


def relay(db: sqlite3.Connection, broker: FakeBroker, batch: int = 100) -> int:
    """后台投递：取未发送的消息，发送成功后标记。可能重复发送（至少一次），消费者要幂等。"""
    rows = db.execute("SELECT id, topic, payload FROM outbox WHERE sent_at IS NULL ORDER BY id LIMIT ?",
                      (batch,)).fetchall()
    sent = 0
    for mid, topic, payload in rows:
        try:
            broker.publish(topic, {"message_id": mid, **json.loads(payload)})
        except ConnectionError:
            with db:
                db.execute("UPDATE outbox SET attempts = attempts + 1 WHERE id = ?", (mid,))
            break  # 保持顺序：前面的没发出去，后面的先不发
        with db:
            db.execute("UPDATE outbox SET sent_at = datetime('now') WHERE id = ?", (mid,))
        sent += 1
    return sent


class Consumer:
    """幂等消费者：按 message_id 去重（真实项目里用数据库唯一约束记录已处理的消息）。"""

    def __init__(self) -> None:
        self.processed: set[int] = set()
        self.points: dict[int, int] = {}

    def handle(self, msg: dict) -> None:
        if msg["message_id"] in self.processed:
            return
        self.points[msg["user_id"]] = self.points.get(msg["user_id"], 0) + msg["amount"] // 100
        self.processed.add(msg["message_id"])


def main() -> None:
    db = sqlite3.connect(":memory:")
    init(db)
    create_order(db, user_id=1, amount=19900)
    try:
        create_order(db, user_id=2, amount=500, fail_after_insert=True)
    except RuntimeError:
        pass
    create_order(db, user_id=1, amount=10000)
    assert db.execute("SELECT count(*) FROM orders").fetchone()[0] == 2
    assert db.execute("SELECT count(*) FROM outbox").fetchone()[0] == 2  # 回滚的订单没有留下消息

    broker = FakeBroker(fail_times=1)
    assert relay(db, broker) == 0  # 第一次 broker 不可用
    assert relay(db, broker) == 2  # 重试成功
    assert relay(db, broker) == 0  # 没有待发送的了

    consumer = Consumer()
    for m in broker.sent + broker.sent[:1]:  # 模拟第一条消息被重复投递
        consumer.handle(m)
    assert consumer.points == {1: 199 + 100}, consumer.points
    print("发件箱：订单和消息同事务写入；投递失败可重试；消费者幂等，重复消息不会重复加积分")


if __name__ == "__main__":
    main()