feat: TenantRuntime/Registry——每租户懒建 engine(WAL/pool2)+文件加载器+站点解析

Co-Authored-By: Claude Code <noreply@anthropic.com>

app/engine/models.py

@@ -2,7 +2,7 @@ import enum
 import uuid
 from datetime import datetime, timezone
 
-from sqlalchemy import String, Text, Boolean, Integer, ForeignKey, JSON, Enum as SAEnum
+from sqlalchemy import String, Text, Boolean, Integer, ForeignKey, JSON, Enum as SAEnum, text
 from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
 
 
@@ -159,3 +159,27 @@ class LeadPushLog(Base):
         default=lambda: datetime.now(timezone.utc),
         onupdate=lambda: datetime.now(timezone.utc),
     )
+
+
+async def ensure_schema(engine) -> None:
+    """create_all + 旧库加列迁移(自 app/main.py lifespan/_migrate 提取,逻辑逐字等价)。
+
+    jc 上线库(conversations 无 page_token / evaluations 无 correction_count)
+    经此升级;新库 create_all 已含全部列,加列分支空跑。TenantRegistry 懒建
+    每租户库时复用(Phase 1);生产装配见 app/main.py lifespan。
+    """
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+        result = await conn.execute(text("PRAGMA table_info(conversations)"))
+        columns = {row[1] for row in result}
+        if "page_token" not in columns:
+            await conn.execute(text(
+                "ALTER TABLE conversations ADD COLUMN page_token VARCHAR(64)"
+            ))
+
+        result = await conn.execute(text("PRAGMA table_info(evaluations)"))
+        columns = {row[1] for row in result}
+        if "correction_count" not in columns:
+            await conn.execute(text(
+                "ALTER TABLE evaluations ADD COLUMN correction_count INTEGER DEFAULT 0"
+            ))

app/main.py

@@ -2,11 +2,10 @@ from contextlib import asynccontextmanager
 
 from fastapi import FastAPI
 from fastapi.staticfiles import StaticFiles
-from sqlalchemy import text
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
 
 from app.engine.config import settings, ensure_dirs, STATIC_DIR
-from app.engine.models import Base
+from app.engine.models import ensure_schema
 
 
 # 后台推送任务的独立连接写结果可能撞上 SMTP 窗口内持有的写锁,默认 5s
@@ -16,28 +15,10 @@ engine = create_async_engine(settings.DATABASE_URL, echo=settings.APP_DEBUG,
 async_session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
 
 
-async def _migrate(conn):
-    result = await conn.execute(text("PRAGMA table_info(conversations)"))
-    columns = {row[1] for row in result}
-    if "page_token" not in columns:
-        await conn.execute(text(
-            "ALTER TABLE conversations ADD COLUMN page_token VARCHAR(64)"
-        ))
-
-    result = await conn.execute(text("PRAGMA table_info(evaluations)"))
-    columns = {row[1] for row in result}
-    if "correction_count" not in columns:
-        await conn.execute(text(
-            "ALTER TABLE evaluations ADD COLUMN correction_count INTEGER DEFAULT 0"
-        ))
-
-
 @asynccontextmanager
 async def lifespan(app: FastAPI):
     ensure_dirs()
-    async with engine.begin() as conn:
-        await conn.run_sync(Base.metadata.create_all)
-        await _migrate(conn)
+    await ensure_schema(engine)
     # 询盘推送(设计 8.2): 清扫崩溃遗留的 stale pending → failed
     # 函数内导入: lead_push_service 顶部 from app.main import async_session,
     # 顶层互相导入会循环

app/platform/registry.py

@@ -0,0 +1,236 @@
+"""TenantRegistry——进程内租户注册表(Phase 1 文件加载器)。
+
+数据源:seed_profiles/<slug>/tenant.json(站点/品牌/引擎配置)+ 同目录
+wall.json(并入 snapshot["wall"],compile_wall 编译期显性炸)。Phase 2 换
+platform.db 时本模块的对外接口不变(spec §5.3:后台保存→原子替换条目,
+进行中的会话下一条消息即用新配置)。
+
+关键行为:
+- resolve_site(host):host 小写去端口 → 域名索引 → (TenantRuntime, Site);
+  未知 Host → UnknownTenantError(spec §4:404,不做兜底租户)。
+  resolve(host) 只是它的便捷形式——双 API 一层实现,防两套解析漂移。
+- 懒建:首次 resolve 才建 engine(WAL/pool=2)+ session_factory + ensure_schema,
+  per-slug asyncio.Lock 串行,双检防并发双建;建库失败不留半开 engine。
+- registry_reload(slug):重读磁盘→校验(坏 JSON/墙/域名冲突全在替换前抛,
+  旧条目不动)→原子替换 _by_slug 条目并重建域名索引→dispose 旧 engine
+  (Windows 下不关=库文件句柄泄漏)。
+
+测试/teardown 纪律:凡建过 engine 的 Registry 用完必须 await dispose_all()
+(Windows 文件锁,tmp 目录删不掉)。
+"""
+import asyncio
+import json
+import pathlib
+from dataclasses import dataclass
+
+from sqlalchemy import event
+from sqlalchemy.ext.asyncio import (
+    AsyncEngine,
+    AsyncSession,
+    async_sessionmaker,
+    create_async_engine,
+)
+
+from app.engine.models import ensure_schema
+from app.engine.wall_schema import CompiledWall, compile_wall
+from app.platform.tenant_config import Site, TenantConfig
+
+
+class UnknownTenantError(LookupError):
+    """Host/域名无法映射到任何租户(调用方语义:聊天/公开端点 404)。"""
+
+
+def make_tenant_engine(db_path: pathlib.Path) -> AsyncEngine:
+    """每租户 SQLite engine:pool=2 无溢出(单租户低并发),忙等 30s
+    (对齐 jc 现状:后台推送连接可能撞 SMTP 窗口写锁),打开即 WAL
+    (spec §6.1 硬纪律:读写不互斥,优于靠 timeout 硬扛)。"""
+    engine = create_async_engine(
+        f"sqlite+aiosqlite:///{db_path.as_posix()}",
+        pool_size=2, max_overflow=0,
+        connect_args={"timeout": 30})
+
+    @event.listens_for(engine.sync_engine, "connect")
+    def _wal(dbapi_conn, _):                     # dbapi_conn = aiosqlite 同步适配层
+        dbapi_conn.execute("PRAGMA journal_mode=WAL")
+
+    return engine
+
+
+def _normalize_host(host: str) -> str:
+    """Host 头归一:去空白/小写/去端口/去 FQDN 尾点(真实请求 Host:8018、
+    大写都可能出现);IPv6 字面量 [::1]:8018 取方括号内部分。"""
+    host = (host or "").strip().lower().rstrip(".")
+    if host.startswith("["):
+        return host.split("]", 1)[0].lstrip("[")
+    return host.split(":", 1)[0]
+
+
+@dataclass
+class TenantRuntime:
+    """单租户运行时。config/db_path/compiled_wall 加载期就绪;engine/
+    session_factory 由 Registry 首次 resolve 懒建;manager 由装配线注入
+    (Task 5b),prompts 由 Task 6 注入——本任务不含。"""
+
+    config: TenantConfig
+    db_path: pathlib.Path
+    compiled_wall: CompiledWall
+    engine: AsyncEngine | None = None
+    session_factory: async_sessionmaker | None = None
+
+    def site_by_slug(self, slug: str) -> Site | None:
+        """按 slug 取站点(空域名站点如 es 靠嵌入代码 data-site 接入)。"""
+        for site in self.config.sites:
+            if site.slug == slug:
+                return site
+        return None
+
+
+class TenantRegistry:
+    """tenant slug → TenantRuntime;域名 → (tenant_slug, site_slug) 索引。
+    reload 重建索引而非增量维护——租户数是个位数,整建最简且无漂移。"""
+
+    def __init__(self):
+        self._seed_dir: pathlib.Path | None = None
+        self._data_dir: pathlib.Path | None = None
+        self._by_slug: dict[str, TenantRuntime] = {}
+        self._domain_index: dict[str, tuple[str, str]] = {}
+        self._locks: dict[str, asyncio.Lock] = {}
+
+    # ---- 加载 -----------------------------------------------------------
+
+    @classmethod
+    def load_from_dir(cls, seed_dir, data_dir) -> "TenantRegistry":
+        """seed_profiles/<slug>/tenant.json(+同目录 wall.json)→ Registry。
+
+        加载期显性失败(不流入运行期):目录无任何 tenant.json(seed_dir
+        路径拼错→静默空注册表→全量 404,必须炸)、slug 与目录名不一致、
+        零站点、wall 编译错误、域名重复绑定。
+        """
+        reg = cls()
+        reg._seed_dir = pathlib.Path(seed_dir)
+        reg._data_dir = pathlib.Path(data_dir)
+        tenant_jsons = sorted(reg._seed_dir.glob("*/tenant.json"))
+        if not tenant_jsons:
+            raise ValueError(f"{reg._seed_dir} 下没有任何 <slug>/tenant.json")
+        for tenant_json in tenant_jsons:
+            rt = reg._load_tenant(tenant_json.parent)
+            if rt.config.slug in reg._by_slug:
+                raise ValueError(f"租户 slug 重复: {rt.config.slug}")
+            reg._by_slug[rt.config.slug] = rt
+        reg._domain_index = reg._build_domain_index(reg._by_slug)
+        return reg
+
+    def _load_tenant(self, tenant_dir: pathlib.Path) -> TenantRuntime:
+        slug = tenant_dir.name
+        tenant_json = json.loads(
+            (tenant_dir / "tenant.json").read_text(encoding="utf-8"))
+        if tenant_json.get("slug") != slug:
+            raise ValueError(
+                f"{tenant_dir / 'tenant.json'}: slug 字段 "
+                f"{tenant_json.get('slug')!r} 与目录名 {slug!r} 不一致")
+        sites = tenant_json.get("sites") or []
+        if not sites:
+            raise ValueError(f"{slug}: 每租户至少一个站点(sites 为空)")
+        # wall 并入 snapshot["wall"]:data["wall"] 保留原始词表(后台编辑/
+        # 回滚用),编译产物挂 runtime.compiled_wall(引擎消费,单一编译)
+        wall_cfg = json.loads(
+            (tenant_dir / "wall.json").read_text(encoding="utf-8"))
+        snapshot = dict(tenant_json.get("config") or {})
+        snapshot["wall"] = wall_cfg
+        config = TenantConfig.from_snapshot(
+            slug, tenant_json.get("name", slug), snapshot,
+            sites=sites, status=tenant_json.get("status", "active"))
+        return TenantRuntime(
+            config=config,
+            db_path=self._data_dir / f"{slug}.db",
+            compiled_wall=compile_wall(wall_cfg))
+
+    def _build_domain_index(
+            self, by_slug: dict[str, TenantRuntime]
+    ) -> dict[str, tuple[str, str]]:
+        index: dict[str, tuple[str, str]] = {}
+        for slug, rt in by_slug.items():
+            for site in rt.config.sites:
+                for domain in site.domains:
+                    key = _normalize_host(domain)
+                    if not key:
+                        continue
+                    if index.get(key, (slug, site.slug)) != (slug, site.slug):
+                        raise ValueError(
+                            f"域名 {key} 同时绑定 {index[key][0]} 与 "
+                            f"{slug} 的站点(域名不可重复绑定)")
+                    index[key] = (slug, site.slug)
+        return index
+
+    # ---- 解析(对外 API)--------------------------------------------------
+
+    async def resolve_site(self, host: str) -> tuple[TenantRuntime, Site]:
+        """host → (TenantRuntime, Site)。首次 resolve 懒建该租户 engine。"""
+        mapped = self._domain_index.get(_normalize_host(host))
+        if mapped is None:
+            raise UnknownTenantError(f"未知 Host: {host!r}")
+        tenant_slug, site_slug = mapped
+        rt = self._by_slug[tenant_slug]
+        await self._ensure_ready(rt)
+        site = rt.site_by_slug(site_slug)
+        if site is None:        # 防御:索引与配置同源重建,正常不可达
+            raise UnknownTenantError(
+                f"站点 {site_slug!r} 不存在于租户 {tenant_slug!r}")
+        return rt, site
+
+    async def resolve(self, host: str) -> TenantRuntime:
+        """resolve_site 的便捷形式(双 API 一层实现)。"""
+        return (await self.resolve_site(host))[0]
+
+    # ---- 懒建 / 热加载 / 释放 ---------------------------------------------
+
+    async def _ensure_ready(self, rt: TenantRuntime) -> TenantRuntime:
+        """懒建 engine + session_factory + ensure_schema,per-slug Lock 串行。
+
+        双检:抢到锁后重看 engine——等锁期间前一个协程可能已建好。
+        建库失败 dispose 半开 engine 并保持未就绪,下次 resolve 重试。
+        """
+        if rt.engine is not None:
+            return rt
+        lock = self._locks.setdefault(rt.config.slug, asyncio.Lock())
+        async with lock:
+            if rt.engine is not None:
+                return rt
+            self._data_dir.mkdir(parents=True, exist_ok=True)
+            engine = make_tenant_engine(rt.db_path)
+            try:
+                await ensure_schema(engine)
+            except Exception:
+                await engine.dispose()
+                raise
+            rt.engine = engine
+            rt.session_factory = async_sessionmaker(
+                engine, class_=AsyncSession, expire_on_commit=False)
+        return rt
+
+    async def registry_reload(self, slug: str) -> None:
+        """重读磁盘,原子替换条目;旧 runtime 若已建 engine,dispose 掉。
+
+        先全量校验(重读 tenant.json/wall.json + 候选域名索引)再替换——
+        坏配置抛异常时旧条目原样服务,不留半更新状态。
+        """
+        if slug not in self._by_slug:
+            raise UnknownTenantError(f"租户不存在: {slug!r}")
+        old = self._by_slug[slug]
+        rt = self._load_tenant(self._seed_dir / slug)
+        candidate = dict(self._by_slug)
+        candidate[slug] = rt
+        new_index = self._build_domain_index(candidate)
+        self._by_slug[slug] = rt                 # 原子替换(单事件循环内无 await)
+        self._domain_index = new_index
+        if old.engine is not None:
+            await old.engine.dispose()
+            old.engine = old.session_factory = None
+
+    async def dispose_all(self) -> None:
+        """关闭全部 engine 并复位为未就绪(测试 teardown 用——Windows 文件锁;
+        复位后若再 resolve 会重新懒建,可安全复用)。"""
+        for rt in self._by_slug.values():
+            if rt.engine is not None:
+                await rt.engine.dispose()
+                rt.engine = rt.session_factory = None