"""Outbox/Inbox 的单机模拟：两个数据库，中间交付允许重复；不是消息队列服务。"""
import argparse
import sqlite3
from contextlib import closing

# closing 负责关闭连接，后面的连接上下文负责提交/回滚事务。
from pathlib import Path
from runtime_store import RuntimeStore


def relay(directory: Path, *, lose_ack: bool = False):
    store = RuntimeStore(directory)
    # 消费者独立持久保存已消费事件，重复投递时不重复更新通知计数。
    with closing(sqlite3.connect(directory / "consumer.sqlite")) as consumer, consumer:
        consumer.executescript("""
            CREATE TABLE IF NOT EXISTS inbox (event_id TEXT PRIMARY KEY);
            CREATE TABLE IF NOT EXISTS counters (task_id TEXT PRIMARY KEY, count INTEGER NOT NULL);
        """)
    with store.connect() as source:
        messages = source.execute("SELECT id, task_id FROM outbox WHERE sent=0 ORDER BY rowid").fetchall()
    for event_id, task_id in messages:
        with closing(sqlite3.connect(directory / "consumer.sqlite")) as consumer, consumer:
            consumer.execute("BEGIN IMMEDIATE")
            inserted = consumer.execute("INSERT OR IGNORE INTO inbox VALUES (?)", (event_id,)).rowcount
            if inserted:
                # Inbox 去重记录与本地业务更新在同一消费者事务提交。
                # 若此处换成发邮件等远端调用，仍需单独设计外部幂等和未知结果处理。
                consumer.execute("INSERT INTO counters VALUES (?, 1) ON CONFLICT(task_id) DO UPDATE SET count=count+1",
                                 (task_id,))
        if lose_ack:
            # 消费者已提交，发布者却没记 sent；下一次 relay 将再次交付同一事件。
            raise TimeoutError("教学故障：消费者已提交，发布确认丢失")
        with store.connect() as source:
            source.execute("UPDATE outbox SET sent=1 WHERE id=?", (event_id,))
    with closing(sqlite3.connect(directory / "consumer.sqlite")) as consumer, consumer:
        return consumer.execute("SELECT task_id, count FROM counters ORDER BY task_id").fetchall()


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--data-dir", type=Path, default=Path(".runtime-demo"))
    parser.add_argument("--lose-ack", action="store_true")
    args = parser.parse_args()
    print(relay(args.data_dir, lose_ack=args.lose_ack))