# -*- coding: utf-8 -*-
"""M11b-2d-A 测试夹具层：**函数式 API 直连真源**的运行时事务引擎（仅测试使用）。

本文件的历史与本轮收敛（响应 QC #2 硬门禁）
==========================================================
M11b-2a「世界同步真源迁移」把写入实现迁到 **函数式 API**
``pbl_runtime_ext.tx_write.apply_runtime_event`` / ``read_states``
（见 ``modules/pbl_runtime_ext/pbl_runtime_ext/tx_write.py:344`` / ``:466``）。

本文件在 M11b-2c 时的形态是「类式垫片」：定义 ``RuntimeStateEventWriter`` 类，
持有 ``conn_factory`` / ``dialect`` / ``close_connection``，再把方法调用转发到
``world_sync/pbl_runtime_tx.write_event_with_state``。那是**第二条夹具路径**——
它把测试钉在「旧类式调用形态 + world_sync 本地同步实现」上，既没验证现行真源
``apply_runtime_event``，又与真源返回键（``action`` / ``states[]`` / ``stmt_count``
/ ``latency_ms``）不一致，垫片只能靠 ``_augment`` 补假键（``state_inserted`` 等）
维持旧断言，实测 22 failed（``Cannot operate on a closed database.`` /
``KeyError: 'expected_version'``）。

**本轮已删除该类别垫片**，本文件只保留一条夹具路径：

  1. ``RuntimeEngine``：以 **真实 sqlite3** 为存储的事务引擎。表结构（列名/类型/
     NOT NULL/DEFAULT/唯一索引）**从 ``pbl_runtime_ext/models/*.json`` 读取生成**
     ——不另立第二份 schema 定义，避免列名再次漂移（这正是 28 failed 的根因之一：
     旧夹具里手写 DDL 用了 ``event_id`` / ``state`` 等真源不存在的列）。
  2. ``install(engine)`` / ``uninstall()``：把 ``rtx_db.transaction`` /
     ``rtx_db.dbname`` / ``rtx_db.q_all`` 换成 sqlite 引擎实现，使
     ``tx_write.apply_runtime_event`` / ``read_states`` **走自身真实代码路径**
     （幂等预检 → INSERT 事件 → UPSERT 状态 → 快照 → COMMIT → 提交后广播）。
     与 ``modules/pbl_runtime_ext/scripts/m11b_selftest.py::install_fake`` 同一
     思路，差别只在存储换成真 sqlite（可验证真实 ROLLBACK 与 DB 唯一索引兜底）。
  3. ``apply(**kw)`` / ``read(**kw)``：函数式 API 的同步入口（``asyncio.run``）。
     **不做任何返回键增补**——测试直接断言真源返回键。
  4. ``percentile``：转发到 ``world_sync.pbl_runtime_sql.percentile``（真源唯一实现）。

边界（QC #3 允许路径）：只改测试夹具层，**不改业务实现、不改 models/*.json 表定义、
不改任何 load_path.py RBAC 注册**。本文件不含任何生产能力：只有被 pytest 夹具调用时
才安装引擎补丁，``uninstall()`` 逐样还原。
"""

import asyncio
import json
import os
import sqlite3
import sys
import threading
import time

HERE = os.path.dirname(os.path.abspath(__file__))
WS_ROOT = os.path.dirname(HERE)                       # modules/world_sync
RTX_ROOT = os.path.join(os.path.dirname(WS_ROOT), "pbl_runtime_ext")   # modules/pbl_runtime_ext

for _p in (WS_ROOT, RTX_ROOT):
    if _p not in sys.path:
        sys.path.insert(0, _p)

# --- 真源导入（被测对象本体，不是副本）---------------------------------
from pbl_runtime_ext import broadcast as _bcast        # noqa: E402
from pbl_runtime_ext import m11b_config as _cfg        # noqa: E402
from pbl_runtime_ext import rtx_db as _rtx_db          # noqa: E402
from pbl_runtime_ext import tx_write as _tx_write      # noqa: E402
from world_sync import pbl_runtime_sql as ps           # noqa: E402

# 旧夹具里以名字引用的常量：统一取真源，避免第二份字面量
SQLITE = ps.DIALECT_SQLITE
MYSQL = ps.DIALECT_MYSQL
percentile = ps.percentile                       # 真源唯一实现（world_sync.pbl_runtime_sql）
apply_runtime_event = _tx_write.apply_runtime_event   # 现行函数式真源 API
read_states = _tx_write.read_states
build_row = _tx_write.build_row
assert_append_only = _tx_write.assert_append_only
RtxError = _rtx_db.RtxError

EVENT_TABLE = _cfg.EVENT_TABLE
STATE_TABLE = _cfg.STATE_TABLE
SNAPSHOT_TABLE = _cfg.SNAPSHOT_TABLE
PROTECTED_BASE_TABLES = _cfg.PROTECTED_BASE_TABLES
ERR_TENANT_MISSING = _cfg.ERR_TENANT_MISSING
ERR_BAD_PARAM = _cfg.ERR_BAD_PARAM
ERR_STATE_CONFLICT = _cfg.ERR_STATE_CONFLICT
ERR_APPEND_ONLY = _cfg.ERR_APPEND_ONLY
ERR_BROADCAST_AFTER_COMMIT = _cfg.ERR_BROADCAST_AFTER_COMMIT
# build_row 的三道门禁码（真源 tx_write.py 内以字面量抛出）
ERR_UNKNOWN_COLUMN = "PBL-RTX-0003"
ERR_MISSING_NOT_NULL = "PBL-RTX-0004"
ERR_IMMUTABLE_COLUMN = "PBL-RTX-0006"

_MODELS_DIR = os.path.join(RTX_ROOT, "models")


# ---------------------------------------------------------------- schema
def _sql_type(t):
    """抽象类型 → sqlite 存储类（sqlite 动态类型，映射只为可读性/亲和性）。"""
    t = str(t or "").lower()
    if t in ("long", "int", "bigint", "smallint", "short"):
        return "INTEGER"
    if t in ("float", "double", "ddouble", "decimal"):
        return "REAL"
    if t in ("text", "json"):
        return "TEXT"
    if t in ("date", "datetime", "time", "timestamp"):
        return "TEXT"
    return "TEXT"


def _not_null(fd):
    if fd.get("null") is False:
        return True
    return str(fd.get("nullable") or "").strip().lower() in ("no", "false", "0")


def _default_clause(fd):
    dv = fd.get("default")
    if dv is None:
        return ""
    s = str(dv).strip()
    if s.upper() in ("CURRENT_TIMESTAMP", "NULL", "NONE", ""):
        return ""
    if isinstance(dv, bool):
        return " DEFAULT %d" % int(dv)
    if isinstance(dv, (int, float)):
        return " DEFAULT %s" % dv
    return " DEFAULT '%s'" % s.replace("'", "''")


def model_fields(table):
    """表列定义（权威源 = pbl_runtime_ext/models/{table}.json）。"""
    path = os.path.join(_MODELS_DIR, "%s.json" % table)
    with open(path, "r", encoding="utf-8") as fh:
        return json.load(fh)


def ddl_statements(table):
    """按 models/*.json 生成 sqlite DDL（列/NOT NULL/DEFAULT/唯一索引）。"""
    spec = model_fields(table)
    cols = []
    for fd in spec.get("fields") or []:
        name = fd.get("name")
        if not name:
            continue
        if fd.get("auto_increment"):
            cols.append('"%s" INTEGER PRIMARY KEY AUTOINCREMENT' % name)
            continue
        d = '"%s" %s' % (name, _sql_type(fd.get("type")))
        if _not_null(fd):
            d += " NOT NULL"
        d += _default_clause(fd)
        cols.append(d)
    out = ['CREATE TABLE IF NOT EXISTS "%s" (\n  %s\n)' % (table, ",\n  ".join(cols))]
    for idx in spec.get("indexes") or []:
        fields = idx.get("idxfields") or []
        if not fields:
            continue
        kind = "UNIQUE INDEX" if str(idx.get("idxtype")) == "unique" else "INDEX"
        cols_sql = ", ".join('"%s"' % f for f in fields)
        out.append('CREATE %s IF NOT EXISTS "%s" ON "%s" (%s)'
                   % (kind, idx.get("name") or ("%s_%d" % (table, len(out))), table, cols_sql))
    return out


def _to_qmark(sql):
    """真源 SQL 用 sqlor 的 ``%s`` 占位；sqlite 用 ``?``。"""
    return sql.replace("%s", "?")


# ---------------------------------------------------------------- 引擎
class RuntimeEngine(object):
    """真实 sqlite 事务引擎（测试专用），适配 ``rtx_db.Transaction`` 接口。

    记录三条可断言的轨迹：
      * ``trace``   有序事务轨迹：``BEGIN`` / ``SQL`` / ``COMMIT`` / ``ROLLBACK`` /
                    ``BROADCAST`` —— 用于「广播只在 commit 之后」「BEGIN/COMMIT 配平」
      * ``sql_log`` 实际发出的 SQL（含参数），用于「SET 子句不含不可变列」「无广播/
                    轮询/睡眠语句」这类语句级断言
      * ``broadcasts`` 提交后广播消息（hub.publish 录制）
    """

    def __init__(self, path=":memory:", tables=(EVENT_TABLE, STATE_TABLE, SNAPSHOT_TABLE)):
        self.path = path
        self.tables = tuple(tables)
        self.trace = []
        self.sql_log = []
        self.broadcasts = []
        self.txn_depth = 0
        self._lock = threading.RLock()
        self._mem = None
        if path == ":memory:":
            self._mem = sqlite3.connect(":memory:", isolation_level=None,
                                        check_same_thread=False)
            self._mem.execute("PRAGMA busy_timeout=15000")
            self._create(self._mem)
        else:
            c = sqlite3.connect(path, isolation_level=None)
            c.execute("PRAGMA busy_timeout=15000")
            self._create(c)
            c.close()

    # -- 建表 -------------------------------------------------------------
    def _create(self, conn):
        for t in self.tables:
            for stmt in ddl_statements(t):
                conn.execute(stmt)
        conn.commit()

    # -- 连接 -------------------------------------------------------------
    def admin_conn(self):
        """给测试直接查库用的连接（文件库返回新连接，调用方负责 close）。"""
        if self._mem is not None:
            return self._mem, False
        c = sqlite3.connect(self.path, isolation_level=None)
        c.execute("PRAGMA busy_timeout=15000")
        return c, True

    def query(self, sql, params=()):
        conn, own = self.admin_conn()
        try:
            cur = conn.execute(_to_qmark(sql), tuple(params or ()))
            cols = [d[0] for d in cur.description] if cur.description else []
            return [dict(zip(cols, r)) for r in cur.fetchall()]
        finally:
            if own:
                conn.close()

    def scalar(self, sql, params=(), default=None):
        rows = self.query(sql, params)
        if not rows:
            return default
        vals = list(rows[0].values())
        return vals[0] if vals else default

    def count(self, table):
        return int(self.scalar("SELECT COUNT(*) AS c FROM \"%s\"" % table, (), 0) or 0)

    def seed_state(self, tenant_id, session_id, entity_id, state_json,
                   state_version=0, checksum="", updated_by="seed"):
        """预置状态行（走真实 INSERT，不绕过列契约）。"""
        now = time.strftime("%Y-%m-%d %H:%M:%S")
        conn, own = self.admin_conn()
        try:
            conn.execute("DELETE FROM \"%s\" WHERE tenant_id=? AND session_id=? AND entity_id=?"
                         % STATE_TABLE, (str(tenant_id), str(session_id), str(entity_id)))
            conn.execute(
                "INSERT INTO \"%s\" (tenant_id, session_id, entity_id, state_json,"
                " state_version, checksum, updated_by, created_at, updated_at)"
                " VALUES (?,?,?,?,?,?,?,?,?)" % STATE_TABLE,
                (str(tenant_id), str(session_id), str(entity_id), state_json,
                 int(state_version), checksum, updated_by, now, now))
            conn.commit()
        finally:
            if own:
                conn.close()

    # -- 事务轨迹 ---------------------------------------------------------
    def _note(self, kind, detail=None):
        with self._lock:
            self.trace.append((kind, detail))

    # -- 只读查询（供 rtx_db.q_all 补丁）----------------------------------
    async def q_all(self, sql, params=None, db=None):
        return _rtx_db._norm_rows(self.query(sql, params))


class _Tx(object):
    """sqlite 事务句柄：与 ``rtx_db.Transaction`` 同接口（execute/insert/update/
    select/commit/rollback/stmt_count），**stmt_count 只计 DML**（与真源一致：
    真源 ``execute`` 不计数、``insert``/``update`` 计数）。
    """

    def __init__(self, engine):
        self.e = engine
        self.conn = None
        self.committed = False
        self.rolled_back = False
        self.stmt_count = 0

    async def __aenter__(self):
        self.conn = self.e._mem if self.e._mem is not None else sqlite3.connect(
            self.e.path, isolation_level=None)
        if self.e._mem is None:
            self.conn.execute("PRAGMA busy_timeout=15000")
        with self.e._lock:
            self.conn.execute("BEGIN IMMEDIATE")
            self.e.txn_depth += 1
        self.e._note("BEGIN")
        return self

    async def __aexit__(self, exc_type, exc, tb):
        if exc_type is not None or not self.committed:
            await self.rollback()
        self._close()
        return False

    def _close(self):
        if self.e._mem is None and self.conn is not None:
            try:
                self.conn.close()
            except Exception:  # noqa: BLE001
                pass
        self.conn = None

    def _run(self, sql, params):
        self.e.sql_log.append((" ".join(str(sql).split()), tuple(params or ())))
        cur = self.conn.cursor()
        try:
            cur.execute(_to_qmark(sql), tuple(params or ()))
            cols = [d[0] for d in cur.description] if cur.description else []
            rows = [dict(zip(cols, r)) for r in cur.fetchall()] if cols else []
            return rows
        finally:
            try:
                cur.close()
            except Exception:  # noqa: BLE001
                pass

    # -- DML / 查询 -------------------------------------------------------
    async def execute(self, sql, params=None):
        return self._run(sql, params)

    async def insert(self, table, row):
        row = dict(row)
        if "tenant_id" in row and row.get("tenant_id") in (None, "", 0):
            raise RtxError(ERR_TENANT_MISSING, "%s 写入缺少 tenant_id" % table)
        cols = list(row.keys())
        sql = 'INSERT INTO "%s" (%s) VALUES (%s)' % (
            table, ", ".join('"%s"' % c for c in cols), ", ".join(["?"] * len(cols)))
        self.e.sql_log.append((" ".join(sql.split()), tuple(row[c] for c in cols)))
        cur = self.conn.cursor()
        try:
            cur.execute(sql, tuple(row[c] for c in cols))
            rid = cur.lastrowid
        finally:
            try:
                cur.close()
            except Exception:  # noqa: BLE001
                pass
        self.stmt_count += 1
        return rid

    async def update(self, table, values, where):
        cols = list(dict(values).keys())
        sql = 'UPDATE "%s" SET %s' % (table, ", ".join('"%s" = ?' % c for c in cols))
        self.e.sql_log.append((" ".join(sql.split()), tuple(dict(values)[c] for c in cols)))
        cur = self.conn.cursor()
        try:
            cur.execute(sql, tuple(dict(values)[c] for c in cols))
            n = cur.rowcount
        finally:
            try:
                cur.close()
            except Exception:  # noqa: BLE001
                pass
        self.stmt_count += 1
        return n

    async def select(self, table, where="", fields="*", order="", limit=200):
        sql = 'SELECT %s FROM "%s"' % (fields or "*", table)
        if where:
            sql += " WHERE %s" % where
        if order:
            sql += " ORDER BY %s" % order
        sql += " LIMIT %d" % int(limit or 200)
        return self._run(sql, ())

    # -- 事务控制 ---------------------------------------------------------
    async def commit(self):
        if self.committed or self.rolled_back:
            return
        self.conn.commit()
        self.committed = True
        with self.e._lock:
            self.e.txn_depth -= 1
        self.e._note("COMMIT")

    async def rollback(self):
        if self.committed or self.rolled_back:
            return
        try:
            self.conn.rollback()
        except Exception:  # noqa: BLE001
            pass
        self.rolled_back = True
        with self.e._lock:
            self.e.txn_depth -= 1
        self.e._note("ROLLBACK")


# ---------------------------------------------------------------- 安装 / 还原
_ORIG = {}
_INSTALLED = False


def install(engine):
    """把 ``rtx_db`` 的事务/查询入口换成 sqlite 引擎，并录制提交后广播。"""
    global _INSTALLED
    if _INSTALLED:
        uninstall()
    _ORIG["transaction"] = _rtx_db.transaction
    _ORIG["dbname"] = _rtx_db.dbname
    _ORIG["q_all"] = _rtx_db.q_all
    _rtx_db.dbname = lambda env=None: "m11b2d_test_sqlite"
    _rtx_db.transaction = lambda db=None: _Tx(engine)
    _rtx_db.q_all = engine.q_all

    _bcast.reset_hub()
    hub = _bcast.get_hub()
    original_publish = hub.publish

    def _recording_publish(channel, event):
        engine.broadcasts.append({"channel": channel, "event": event})
        engine._note("BROADCAST", channel)
        return original_publish(channel, event)

    hub.publish = _recording_publish
    _ORIG["hub"] = (hub, original_publish)
    _INSTALLED = True
    return engine


def uninstall():
    global _INSTALLED
    if not _INSTALLED:
        return
    for key in ("transaction", "dbname", "q_all"):
        if key in _ORIG:
            setattr(_rtx_db, key, _ORIG[key])
    _INSTALLED = False
    _ORIG.clear()


def apply(**kwargs):
    """函数式真源 API 的同步入口（不做返回键增补）。"""
    return asyncio.run(apply_runtime_event(**kwargs))


def read(**kwargs):
    return asyncio.run(read_states(**kwargs))


def trace_kinds(engine):
    return [k for k, _d in engine.trace]


def update_statements(engine):
    """引擎实际发出的 UPDATE 语句（大写、压空格），用于不可变列/乐观锁断言。"""
    return [(s, p) for s, p in engine.sql_log if s.upper().startswith("UPDATE ")]


def insert_statements(engine, table):
    return [(s, p) for s, p in engine.sql_log
            if s.upper().startswith('INSERT INTO "%s"' % table.upper())]