diff --git a/.github/workflows/gaussdb-smoke.yml b/.github/workflows/gaussdb-smoke.yml new file mode 100644 index 0000000..3f040bb --- /dev/null +++ b/.github/workflows/gaussdb-smoke.yml @@ -0,0 +1,113 @@ +name: openGauss dialect smoke + +on: + workflow_dispatch: + +jobs: + smoke: + runs-on: ubuntu-latest + timeout-minutes: 30 + steps: + - name: Checkout + uses: actions/checkout@v4 + + - name: Start openGauss + run: | + GAUSS_PASSWORD="Gs$(openssl rand -hex 9)_aB1" + echo "GAUSS_PASSWORD=$GAUSS_PASSWORD" >> "$GITHUB_ENV" + docker run -d --name opengauss \ + -e GS_PASSWORD="$GAUSS_PASSWORD" \ + -p 5432:5432 \ + enmotech/opengauss:5.0.0 + for i in $(seq 1 120); do + if docker exec opengauss su - omm -c "gsql -d postgres -c 'select 1'" >/dev/null 2>&1; then + echo "openGauss ready after ${i}s" + break + fi + sleep 2 + if [ "$i" -eq 120 ]; then + echo "openGauss failed to become ready" + docker logs opengauss + exit 1 + fi + done + docker exec opengauss su - omm -c "gsql -d postgres -c 'select version()'" + + - name: Setup Python + uses: actions/setup-python@v5 + with: + python-version: "3.12" + + - name: Install backend deps + run: | + python -m venv .venv + .venv/bin/pip install -q -e "backend[dev,postgres]" 2>/dev/null || (cd backend && ../.venv/bin/pip install -q -e ".[dev,postgres]" && ../.venv/bin/pip install -q pytest pytest-asyncio ruff httpx) + + - name: create_all + dialect smoke on openGauss + env: + DATABASE_URL: postgresql+psycopg://postgres:${{ env.GAUSS_PASSWORD }}@127.0.0.1:5432/postgres + STAFFDECK_ROLE: all + working-directory: backend + run: | + set -e + ../.venv/bin/python - <<'PY' + from app.db.database import init_db + init_db() + print("create_all on openGauss: OK") + from sqlalchemy import inspect + from app.db import engine + idx = {i["name"] for t in ("sessions", "model_configs") for i in inspect(engine).get_indexes(t)} + assert "uq_sessions_agent_channel_extconv" in idx, "missing session index" + assert "uq_model_configs_tenant_default" in idx, "missing default-model partial index" + print("indexes: OK") + from app.db.dialect import get_dialect + from sqlmodel import Session + d = get_dialect(engine.url.get_backend_name()) + print("dialect:", d.name) + with Session(engine) as lock_session: + assert d.acquire_advisory_lock(lock_session, "staffdeck-connector") + assert d.check_advisory_lock(lock_session, "staffdeck-connector") + print("advisory lock acquire+check: OK") + d.release_advisory_lock(lock_session, "staffdeck-connector") + print("advisory lock release: OK") + with Session(engine) as db: + from app.db.models import User + db.exec(__import__("sqlmodel").select(User)).all() + print("basic query: OK") + PY + + - name: Boot app and API smoke on openGauss + env: + DATABASE_URL: postgresql+psycopg://postgres:${{ env.GAUSS_PASSWORD }}@127.0.0.1:5432/postgres + STAFFDECK_ROLE: all + ULTRARAG_PORT: 5199 + working-directory: backend + run: | + set -e + ../.venv/bin/python -m uvicorn app.main:app --host 127.0.0.1 --port 5199 & + APP_PID=$! + for i in $(seq 1 60); do + if curl -sf http://127.0.0.1:5199/api/health >/dev/null 2>&1; then echo "healthy after ${i}s"; break; fi + sleep 1 + if [ "$i" -eq 60 ]; then echo "app failed to start"; kill $APP_PID; exit 1; fi + done + curl -sf http://127.0.0.1:5199/api/health + TOKEN=$(curl -sf -X POST http://127.0.0.1:5199/api/auth/login \ + -H 'Content-Type: application/json' \ + -d '{"tenant_id":"tenant_demo","username":"admin","password":"admin"}' | python3 -c 'import sys,json;print(json.load(sys.stdin)["token"])') + echo "login: OK" + AGENTS=$(curl -sf "http://127.0.0.1:5199/api/chat/agents?tenant_id=tenant_demo" -H "Authorization: Bearer $TOKEN") + echo "agents: $(echo "$AGENTS" | python3 -c 'import sys,json;print(len(json.load(sys.stdin)))')" + BID=$(curl -sf -X POST http://127.0.0.1:5199/api/enterprise/channels \ + -H "Authorization: Bearer $TOKEN" -H 'Content-Type: application/json' \ + -d '{"tenant_id":"tenant_demo","agent_id":"agent_30b8f623c6fe445b","channel":"wecom"}' | python3 -c 'import sys,json;print(json.load(sys.stdin)["id"])') + echo "binding: $BID" + curl -sf -X POST "http://127.0.0.1:5199/api/enterprise/channels/$BID/wecom/credentials" \ + -H "Authorization: Bearer $TOKEN" -H 'Content-Type: application/json' \ + -d '{"tenant_id":"tenant_demo","bot_id":"smoke_bot","secret":"smoke_secret","corp_id":"smoke_corp"}' -o /dev/null + echo "credentials(JSON config patch): OK" + curl -sf "http://127.0.0.1:5199/api/enterprise/channels/$BID/deliveries/days?tenant_id=tenant_demo" \ + -H "Authorization: Bearer $TOKEN" | python3 -c 'import sys,json;print("day-bucket endpoint:", json.load(sys.stdin)["total_days"], "days")' + curl -sf "http://127.0.0.1:5199/api/enterprise/knowledge-bases?tenant_id=tenant_demo" \ + -H "Authorization: Bearer $TOKEN" | python3 -c 'import sys,json;print("knowledge-bases:", len(json.load(sys.stdin)))' + kill $APP_PID || true diff --git a/README.md b/README.md index 6f753e0..ba9c20b 100644 --- a/README.md +++ b/README.md @@ -240,6 +240,23 @@ Digital employees can serve users directly over IM channels: users chat with emp - Channel credentials (bot tokens/secrets) are stored Fernet-encrypted and never returned by any API; - Binding management is restricted to admins or the binding creator; mounting an employee exposes it to all users of that channel — grant with care. +## Database Backends + +StaffDeck keeps all state in one SQL database and defaults to local SQLite (zero setup). The data layer is dialect-pluggable: business code goes through SQLAlchemy ORM, and the few dialect-specific points (day bucketing, JSON config patching, process/advisory locks, partial indexes) are centralized in `backend/app/db/dialect.py`. + +| Backend | Status | Notes | +| --- | --- | --- | +| SQLite (default) | Supported | File database; process locks are file locks placed next to the DB file. | +| PostgreSQL / openGauss | Experimental | Full feature set via the `postgresql` dialect. Install the optional driver first: `pip install "skill-agent-loop-backend[postgres]"`, then set `DATABASE_URL="postgresql+psycopg://user:pass@host:5432/dbname"` in `backend/.env`. | +| MySQL / Dameng (达梦) | Adapter needed | Register a small dialect adapter (see the `dialect.py` docstring). These engines lack partial unique indexes, so the "one default model per tenant" invariant is enforced in code plus a startup check (no DB-level constraint at runtime); without a native advisory-lock implementation, the connector lock fails loudly at startup instead of silently mis-locking. JSON columns degrade to CLOB via `PortableJSON`; long-text columns are explicitly `Text` (no silent VARCHAR(255) truncation). | + +Current limitations when running on PostgreSQL/openGauss: + +- **Fresh databases only**: `create_all` initializes a new database, but there is no migration path for existing schemas yet (Alembic migrations are a follow-up task). A startup warning is emitted as a reminder. +- **Single-writer startup**: run only one application worker during first boot/upgrades — concurrent `create_all` + demo seeding from multiple workers can race (DuplicateTable / unique conflicts). SQLite serializes this via write locks; PostgreSQL deployments should init with a single worker first. +- `APP_TIMEZONE` only affects PostgreSQL day bucketing; on SQLite, day buckets always use the server's local timezone. +- Run exactly one connector process per database — the single-instance guard uses a session-scoped PostgreSQL advisory lock (held on a dedicated autocommit connection) with periodic liveness checks (channel services degrade and stop if the lock is lost). + ## Project Structure ```text diff --git a/README.zh.md b/README.zh.md index 7a1db24..cae65c5 100644 --- a/README.zh.md +++ b/README.zh.md @@ -240,6 +240,23 @@ curl.exe http://127.0.0.1:5173/api/health - 渠道凭证（bot token / secret)Fernet 加密落库，任何接口不回传明文； - 绑定管理权限：管理员或绑定创建者；员工挂载动作本身即"对该渠道全部用户开放该员工",请按需授权。 +## 数据库后端 + +StaffDeck 的全部状态存放在一个 SQL 数据库中，默认使用本地 SQLite（零配置）。数据访问层为方言可插拔设计：业务代码统一走 SQLAlchemy ORM，少数方言相关点（日分桶、JSON 配置补丁、进程/advisory 锁、部分索引）集中收口在 `backend/app/db/dialect.py`。 + +| 后端 | 状态 | 说明 | +| --- | --- | --- | +| SQLite（默认） | 支持 | 文件库；进程锁为数据库文件旁的文件锁。 | +| PostgreSQL / openGauss | 实验性 | 通过 `postgresql` 方言提供全能力。需先安装可选驱动：`pip install "skill-agent-loop-backend[postgres]"`，再在 `backend/.env` 设置 `DATABASE_URL="postgresql+psycopg://user:pass@host:5432/dbname"`。 | +| MySQL / 达梦 | 需适配器 | 注册一个小方言适配器即可（见 `dialect.py` 模块 docstring)。此类引擎不支持部分唯一索引，"每租户至多一条默认模型"改由代码层维护 + 启动校验兜底（运行期无 DB 级约束）；未实现原生 advisory lock 时，connector 锁会在启动期响亮失败，而不是静默错锁。JSON 列经 `PortableJSON` 降级为 CLOB，长文本列显式 `Text`(MySQL 不会退化为 VARCHAR(255))。 | + +在 PostgreSQL/openGauss 上运行的当前限制： + +- **仅支持全新库**:`create_all` 负责初始化新库，暂无存量 schema 的迁移通路（Alembic 迁移为后续任务），启动时会输出提醒告警； +- **单写者启动**：首次启动/升级只跑一个应用 worker——多 worker 并发 `create_all` + 演示数据播种会撞键（SQLite 靠写锁串行化无此问题）,PG 部署请先用单 worker 完成初始化； +- `APP_TIMEZONE` 只对 PostgreSQL 日分桶生效；SQLite 的日分桶恒为服务器本地时区； +- 每个数据库只允许运行一个 connector 进程——单实例守护使用会话级 PostgreSQL advisory lock（常驻于一条专属 AUTOCOMMIT 连接），并带定期存活校验（锁失效时渠道服务主动降级停止）。 + ## 项目结构 ```text diff --git a/backend/.env.example b/backend/.env.example index 77fdd1c..603a699 100644 --- a/backend/.env.example +++ b/backend/.env.example @@ -1,5 +1,9 @@ APP_NAME="Skill Agent Loop Service" +# 默认 SQLite;PostgreSQL/openGauss 改为 postgresql+psycopg://user:pass@host:5432/dbname +# (需先安装可选驱动:pip install "skill-agent-loop-backend[postgres]") DATABASE_URL="sqlite:///./skill_agent_loop.db" +# 应用时区(如 Asia/Shanghai):仅 PostgreSQL 日分桶换算使用,SQLite 恒为服务器本地时区 +APP_TIMEZONE="" APP_SECRET="change-me-in-development" DEMO_MODEL_BASE_URL="http://localhost:52010/v1" DEMO_MODEL_NAME="qwen3.6-27b" diff --git a/backend/app/api/channels.py b/backend/app/api/channels.py index 5723d7d..c5330a8 100644 --- a/backend/app/api/channels.py +++ b/backend/app/api/channels.py @@ -1,6 +1,5 @@ from __future__ import annotations -import json import logging import secrets import threading @@ -8,7 +7,7 @@ from datetime import timedelta from fastapi import APIRouter, Depends, HTTPException, Query, Response -from sqlalchemy import case, text, update +from sqlalchemy import case, update from sqlalchemy.exc import IntegrityError from sqlmodel import Session, select @@ -61,6 +60,7 @@ ) from app.config import get_settings from app.db import get_session +from app.db.dialect import get_dialect from app.db.models import ( AgentProfile, ChannelBindCode, @@ -95,24 +95,28 @@ def _patch_binding_config_key( key: str, value: object, ) -> None: - """Patch one API-owned config key against the latest JSON value.""" - result = db.exec( - text( - "UPDATE channel_bindings " - "SET config_json = json_set(COALESCE(config_json, '{}'), :path, json(:value)), " - "updated_at = :updated_at " - "WHERE id = :binding_id AND tenant_id = :tenant_id" - ), - params={ - "path": f"$.{key}", - "value": json.dumps(value, ensure_ascii=False), - "updated_at": utc_now(), - "binding_id": binding_id, - "tenant_id": tenant_id, - }, - ) - if result.rowcount != 1: + """Patch one API-owned config key against the latest JSON value. + + ORM 读-改-写(方言助手收口):SELECT ... FOR UPDATE 行锁覆盖读-改-写全程, + 与 connector 侧 _patch_runtime_config 互斥,避免并发补丁互相覆盖(PG 真实 + 加锁;SQLite 由 SQLAlchemy 省略 FOR UPDATE,单行 WAL 写串行兜底)。 + populate_existing 保证读到最新已提交值——它只绕 identity map,调用方仍须 + 保证进入本会话时没有基于旧快照的活跃事务。 + """ + binding = db.exec( + select(ChannelBinding) + .where(ChannelBinding.id == binding_id) + .with_for_update() + .execution_options(populate_existing=True) + ).first() + if not binding or binding.tenant_id != tenant_id: raise HTTPException(status_code=404, detail="渠道绑定不存在") + binding.config_json = get_dialect(db.get_bind().url.get_backend_name()).json_config_set( + binding.config_json, key, value + ) + binding.updated_at = utc_now() + db.add(binding) + SUPPORTED_CHANNELS = {"wechat", "wecom", "feishu"} INGRESS_QUIESCE_TIMEOUT_SECONDS = 5.0 @@ -1088,8 +1092,10 @@ def list_channel_delivery_days( _ensure_binding_manager(db, tenant_id, binding, current_user) from sqlalchemy import func - # 按服务器本地时区的自然日分桶(SQLite date(created_at, 'localtime')) - day_bucket = func.date(ChannelDelivery.created_at, "localtime") + # 按应用时区的自然日分桶(方言助手收口:SQLite=服务器本地 date(localtime)) + day_bucket = get_dialect(db.get_bind().url.get_backend_name()).day_bucket( + ChannelDelivery.created_at + ) day_rows = db.exec( select(day_bucket, func.count()) .where(ChannelDelivery.binding_id == binding.id) diff --git a/backend/app/api/knowledge.py b/backend/app/api/knowledge.py index ee5ac7e..7f28b51 100644 --- a/backend/app/api/knowledge.py +++ b/backend/app/api/knowledge.py @@ -838,67 +838,125 @@ def bucket_read_with_stats(row: KnowledgeBucket, chunk_count: int) -> KnowledgeB return item +_BUCKET_ROW_FIELDS = ( + "id", + "tenant_id", + "knowledge_base_id", + "knowledge_base_version_id", + "document_id", + "bucket_key", + "title", + "summary", + "token_estimate", + "metadata_json", + "created_at", + "updated_at", +) +_CHUNK_ROW_FIELDS = ( + "id", + "tenant_id", + "knowledge_base_id", + "knowledge_base_version_id", + "document_id", + "bucket_id", + "chunk_index", + "content", + "summary", + "source_ref", + "metadata_json", + "created_at", + "updated_at", +) + + +def _model_row_mapping(row: Any, fields: tuple[str, ...]) -> Mapping[str, Any]: + """ORM 对象转与原 CAST AS BLOB 查询同形的 Mapping(下游统一走 _safe_* 读取)。""" + return {field: getattr(row, field) for field in fields} + + def _safe_document_bucket_rows( db: Session, tenant_id: str, document_id: str ) -> list[Mapping[str, Any]]: - return list( - db.execute( - text( - """ - SELECT - id, - tenant_id, - knowledge_base_id, - knowledge_base_version_id, - document_id, - CAST(bucket_key AS BLOB) AS bucket_key, - CAST(title AS BLOB) AS title, - CAST(summary AS BLOB) AS summary, - token_estimate, - CAST(metadata_json AS BLOB) AS metadata_json, - created_at, - updated_at - FROM knowledge_buckets - WHERE tenant_id = :tenant_id AND document_id = :document_id - ORDER BY created_at ASC - """ - ), - {"tenant_id": tenant_id, "document_id": document_id}, + # SQLite:文本列可能混入非 UTF-8 字节,CAST AS BLOB 取原始字节交给 _safe_text 解码; + # 其它后端走 ORM(驱动返回规范 str,无需 BLOB 兜底) + if db.get_bind().url.get_backend_name() == "sqlite": + return list( + db.execute( + text( + """ + SELECT + id, + tenant_id, + knowledge_base_id, + knowledge_base_version_id, + document_id, + CAST(bucket_key AS BLOB) AS bucket_key, + CAST(title AS BLOB) AS title, + CAST(summary AS BLOB) AS summary, + token_estimate, + CAST(metadata_json AS BLOB) AS metadata_json, + created_at, + updated_at + FROM knowledge_buckets + WHERE tenant_id = :tenant_id AND document_id = :document_id + ORDER BY created_at ASC + """ + ), + {"tenant_id": tenant_id, "document_id": document_id}, + ) + .mappings() + .all() ) - .mappings() - .all() - ) + rows = db.exec( + select(KnowledgeBucket) + .where( + KnowledgeBucket.tenant_id == tenant_id, + KnowledgeBucket.document_id == document_id, + ) + .order_by(KnowledgeBucket.created_at.asc()) + ).all() + return [_model_row_mapping(row, _BUCKET_ROW_FIELDS) for row in rows] def _safe_bucket_chunk_rows(db: Session, tenant_id: str, bucket_id: str) -> list[Mapping[str, Any]]: - return list( - db.execute( - text( - """ - SELECT - id, - tenant_id, - knowledge_base_id, - knowledge_base_version_id, - document_id, - bucket_id, - chunk_index, - CAST(content AS BLOB) AS content, - CAST(summary AS BLOB) AS summary, - CAST(source_ref AS BLOB) AS source_ref, - CAST(metadata_json AS BLOB) AS metadata_json, - created_at, - updated_at - FROM knowledge_chunks - WHERE tenant_id = :tenant_id AND bucket_id = :bucket_id - ORDER BY chunk_index ASC - """ - ), - {"tenant_id": tenant_id, "bucket_id": bucket_id}, + if db.get_bind().url.get_backend_name() == "sqlite": + return list( + db.execute( + text( + """ + SELECT + id, + tenant_id, + knowledge_base_id, + knowledge_base_version_id, + document_id, + bucket_id, + chunk_index, + CAST(content AS BLOB) AS content, + CAST(summary AS BLOB) AS summary, + CAST(source_ref AS BLOB) AS source_ref, + CAST(metadata_json AS BLOB) AS metadata_json, + created_at, + updated_at + FROM knowledge_chunks + WHERE tenant_id = :tenant_id AND bucket_id = :bucket_id + ORDER BY chunk_index ASC + """ + ), + {"tenant_id": tenant_id, "bucket_id": bucket_id}, + ) + .mappings() + .all() ) - .mappings() - .all() - ) + rows = db.exec( + select(KnowledgeChunk) + .where( + KnowledgeChunk.tenant_id == tenant_id, + KnowledgeChunk.bucket_id == bucket_id, + ) + .order_by(KnowledgeChunk.chunk_index.asc()) + ).all() + return [_model_row_mapping(row, _CHUNK_ROW_FIELDS) for row in rows] def _bucket_read_mapping_with_stats( diff --git a/backend/app/channels/__init__.py b/backend/app/channels/__init__.py index fa133f8..cd254a9 100644 --- a/backend/app/channels/__init__.py +++ b/backend/app/channels/__init__.py @@ -5,8 +5,6 @@ import threading import time from contextlib import contextmanager -from pathlib import Path -from typing import IO from app.config import get_settings @@ -18,77 +16,118 @@ _feishu_process_manager = None _binding_lifecycle_locks: dict[str, threading.RLock] = {} _binding_lifecycle_locks_guard = threading.Lock() -_connector_lock_file: IO[bytes] | None = None +# connector 单实例锁:统一锁 key;PG 持锁连接由方言常驻(_connector_lock_session 仅作 bind 来源) +_CONNECTOR_LOCK_KEY = "staffdeck-connector" _connector_lock_pid: int | None = None +_connector_lock_session = None _intake_sweep_thread: threading.Thread | None = None +_connector_lock_watchdog_thread: threading.Thread | None = None +# PG advisory lock 存活校验周期(秒):连接被服务端掐断后锁会静默释放,须定期核实 +_CONNECTOR_LOCK_CHECK_SECONDS = 15.0 def _acquire_connector_process_lock() -> bool: - global _connector_lock_file, _connector_lock_pid + """单实例 connector 锁:走方言提供者(PG=advisory lock,其它=数据目录文件锁)。 + + preload+fork 部署下,子进程不得把继承的锁状态当作自己持有。 + """ + global _connector_lock_pid, _connector_lock_session current_pid = os.getpid() - if _connector_lock_file is not None and _connector_lock_pid == current_pid: + if _connector_lock_pid == current_pid: return True - if _connector_lock_file is not None: - # preload 后 fork 的子进程不能把继承句柄当作自己已持有锁。 - _connector_lock_file.close() - _connector_lock_file = None + if _connector_lock_pid is not None: + # fork 子进程:继承的会话/句柄只丢弃引用,不在共享连接上做任何操作 + _connector_lock_session = None _connector_lock_pid = None - from app.db import engine + from sqlmodel import Session - database_path = engine.url.database - if engine.url.get_backend_name() != "sqlite" or not database_path or database_path == ":memory:": - logger.error("渠道服务要求文件 SQLite 进程锁；当前数据库不支持可靠的单实例 Outbox") - return False - lock_path = Path(database_path).resolve().with_name(f"{Path(database_path).name}.connector.lock") - handle = lock_path.open("a+b") - try: - if os.name == "nt": - import msvcrt - - handle.seek(0) - if handle.read(1) == b"": - handle.write(b"0") - handle.flush() - handle.seek(0) - msvcrt.locking(handle.fileno(), msvcrt.LK_NBLCK, 1) - else: - import fcntl - - fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) - except (BlockingIOError, OSError): - handle.close() - return False - _connector_lock_file = handle + from app.db import engine + from app.db.dialect import get_dialect + + dialect = get_dialect(engine.url.get_backend_name()) + if dialect.session_scoped_advisory_lock: + # PG advisory lock 随连接存活:专属连接由方言内部常驻,此会话仅作 bind 来源 + session = Session(engine) + if not dialect.acquire_advisory_lock(session, _CONNECTOR_LOCK_KEY): + session.close() + logger.error("另一 connector 进程已持有该数据库的 advisory lock") + return False + _connector_lock_session = session + else: + # SQLite/其它:数据目录文件锁(锁句柄由方言实例持有,行为不变) + with Session(engine) as session: + if not dialect.acquire_advisory_lock(session, _CONNECTOR_LOCK_KEY): + return False _connector_lock_pid = current_pid return True def _release_connector_process_lock() -> None: - global _connector_lock_file, _connector_lock_pid - handle = _connector_lock_file - if handle is None: + global _connector_lock_pid, _connector_lock_session + if _connector_lock_pid is None: return if _connector_lock_pid != os.getpid(): - handle.close() - _connector_lock_file = None + # fork 子进程:仅丢弃引用,不解父进程的锁 + _connector_lock_session = None _connector_lock_pid = None return - try: - if os.name == "nt": - import msvcrt + from sqlmodel import Session - handle.seek(0) - msvcrt.locking(handle.fileno(), msvcrt.LK_UNLCK, 1) - else: - import fcntl + from app.db import engine + from app.db.dialect import get_dialect - fcntl.flock(handle.fileno(), fcntl.LOCK_UN) + dialect = get_dialect(engine.url.get_backend_name()) + session, _connector_lock_session = _connector_lock_session, None + try: + try: + if session is not None: + dialect.release_advisory_lock(session, _CONNECTOR_LOCK_KEY) + else: + with Session(engine) as fallback_session: + dialect.release_advisory_lock(fallback_session, _CONNECTOR_LOCK_KEY) + except Exception: + # 会话可能已随连接中断死亡(锁实际已被服务端释放),释放失败只记录不抛出 + logger.exception("释放 connector 锁失败(锁可能已随断连释放)") finally: - handle.close() - _connector_lock_file = None + if session is not None: + session.close() _connector_lock_pid = None +def _connector_lock_healthy() -> bool: + """会话级 advisory lock 存活校验;文件锁无静默失效模式,恒为 True。""" + if _connector_lock_session is None: + return True + from app.db import engine + from app.db.dialect import get_dialect + + dialect = get_dialect(engine.url.get_backend_name()) + try: + return dialect.check_advisory_lock(_connector_lock_session, _CONNECTOR_LOCK_KEY) + except Exception: + logger.exception("connector 锁存活校验异常") + return False + + +def _connector_lock_watchdog() -> None: + """PG advisory lock 断连检测:锁静默失效后主动降级(停渠道服务,避免双 connector)。""" + while True: + time.sleep(_CONNECTOR_LOCK_CHECK_SECONDS) + if _connector_lock_pid != os.getpid() or _connector_lock_session is None: + return # 锁已正常释放或本进程不再持有,看门狗退出 + if _connector_lock_healthy(): + continue + logger.error( + "connector advisory lock 已失效(数据库连接中断),渠道服务主动降级停止;" + "恢复后请重启进程重新接管" + ) + try: + stop_channel_services() + except Exception: + logger.exception("渠道服务降级停止失败") + return + + def get_wechat_poll_manager(): global _wechat_poll_manager if _wechat_poll_manager is None: @@ -210,7 +249,7 @@ def restart_binding_ingress(channel: str, binding_id: str, *, wait_seconds: floa def start_channel_services() -> None: - global _intake_sweep_thread + global _intake_sweep_thread, _connector_lock_watchdog_thread if not channel_services_enabled(): logger.info("staffdeck_role=%s,渠道服务不启动", get_settings().staffdeck_role) return @@ -236,6 +275,14 @@ def start_channel_services() -> None: daemon=True, ) _intake_sweep_thread.start() + if _connector_lock_session is not None: + # 会话级 advisory lock(PG):定期核实锁仍持有,断连静默失效时主动降级 + _connector_lock_watchdog_thread = threading.Thread( + target=_connector_lock_watchdog, + name="staffdeck-connector-lock-watchdog", + daemon=True, + ) + _connector_lock_watchdog_thread.start() except Exception: stop_channel_services() raise diff --git a/backend/app/channels/adapters/wechat.py b/backend/app/channels/adapters/wechat.py index 8edd4c8..98190dc 100644 --- a/backend/app/channels/adapters/wechat.py +++ b/backend/app/channels/adapters/wechat.py @@ -1,7 +1,6 @@ from __future__ import annotations import base64 -import json import logging import os import threading @@ -11,7 +10,7 @@ from uuid import uuid4 import httpx -from sqlalchemy import text +from sqlalchemy import update from sqlmodel import Session, select from app.channels.adapters.base import ( @@ -46,6 +45,27 @@ WECHAT_ALLOWED_HOSTS = ("ilinkai.weixin.qq.com",) +# 同进程运行时补丁串行锁:读-改-写不是单条 SQL,避免并发补丁互相覆盖 +# (跨会话/进程的并发写由 SELECT ... FOR UPDATE 行锁互斥,见 _patch_runtime_config) +_runtime_patch_locks: dict[str, threading.Lock] = {} +_runtime_patch_locks_guard = threading.Lock() + + +def _runtime_patch_lock(binding_id: str) -> threading.Lock: + with _runtime_patch_locks_guard: + lock = _runtime_patch_locks.get(binding_id) + if lock is None: + lock = threading.Lock() + _runtime_patch_locks[binding_id] = lock + return lock + + +def _drop_runtime_patch_lock(binding_id: str) -> None: + """binding ingress 停止后清理其补丁锁,避免锁字典随绑定增删无上限累积。""" + with _runtime_patch_locks_guard: + _runtime_patch_locks.pop(binding_id, None) + + def _patch_runtime_config( db_engine, binding_id: str, @@ -57,49 +77,53 @@ def _patch_runtime_config( expected_values: dict[str, Any] | None = None, binding_values: dict[str, Any] | None = None, ) -> bool: - """Atomically patch connector-owned JSON keys without replacing API configuration.""" - config_expr = "COALESCE(config_json, '{}')" - params: dict[str, Any] = {"binding_id": binding_id, "updated_at": utc_now()} - for index, (key, value) in enumerate((set_values or {}).items()): - params[f"set_path_{index}"] = f"$.{key}" - params[f"set_value_{index}"] = json.dumps(value, ensure_ascii=False) - config_expr = ( - f"json_set({config_expr}, :set_path_{index}, json(:set_value_{index}))" - ) - for index, key in enumerate(remove_keys): - params[f"remove_path_{index}"] = f"$.{key}" - config_expr = f"json_remove({config_expr}, :remove_path_{index})" - - assignments = ["updated_at = :updated_at"] - if set_values or remove_keys: - assignments.insert(0, f"config_json = {config_expr}") + """Atomically patch connector-owned JSON keys without replacing API configuration. + + ORM 读-改-写 + SELECT ... FOR UPDATE 行锁 + config_revision CAS:读取即锁定 + 目标行直到提交,与 API 侧补丁(_patch_binding_config_key,同样持 FOR UPDATE) + 互斥,关闭整份 JSON 读-改-写的丢写窗口;PG 真实加锁,SQLite 不支持 FOR UPDATE + 由 SQLAlchemy 自动省略(单行 WAL 写串行 + 同进程补丁锁,行为不变)。 + expected_revision/expected_values 在 Python 侧校验后随写回一并执行。 + """ allowed_binding_values = {"connected", "status"} + column_values: dict[str, Any] = {} for key, value in (binding_values or {}).items(): if key not in allowed_binding_values: raise ValueError(f"unsupported binding runtime field: {key}") - params[f"binding_value_{key}"] = value - assignments.append(f"{key} = :binding_value_{key}") - - predicates = ["id = :binding_id"] - if require_active: - predicates.append("status = 'active'") - if expected_revision is not None: - params["expected_revision"] = expected_revision - predicates.append("config_revision = :expected_revision") - for index, (key, value) in enumerate((expected_values or {}).items()): - params[f"expected_path_{index}"] = f"$.{key}" - params[f"expected_value_{index}"] = value - predicates.append( - f"json_extract(config_json, :expected_path_{index}) = :expected_value_{index}" - ) - - with Session(db_engine) as db: + column_values[key] = value + + with _runtime_patch_lock(binding_id), Session(db_engine) as db: + binding = db.exec( + select(ChannelBinding) + .where(ChannelBinding.id == binding_id) + .with_for_update() + ).first() + if not binding: + return False + if require_active and binding.status != "active": + return False + revision = binding.config_revision + if expected_revision is not None and revision != expected_revision: + return False + config = dict(binding.config_json or {}) + for key, value in (expected_values or {}).items(): + if config.get(key) != value: + return False + for key, value in (set_values or {}).items(): + config[key] = value + for key in remove_keys: + config.pop(key, None) + + values: dict[str, Any] = {"updated_at": utc_now(), **column_values} + if set_values or remove_keys: + values["config_json"] = config result = db.exec( - text( - f"UPDATE channel_bindings SET {', '.join(assignments)} " - f"WHERE {' AND '.join(predicates)}" - ), - params=params, + update(ChannelBinding) + .where( + ChannelBinding.id == binding_id, + ChannelBinding.config_revision == revision, + ) + .values(**values) ) db.commit() return result.rowcount == 1 @@ -547,7 +571,11 @@ def wait_binding_stopped(self, binding_id: str, timeout_seconds: float = 5.0) -> if thread and thread.is_alive(): deadline = time.monotonic() + max(0.0, timeout_seconds) thread.join(timeout=max(0.0, deadline - time.monotonic())) - return not (thread and thread.is_alive()) + stopped = not (thread and thread.is_alive()) + if stopped: + # 线程确认退出后清理其运行时补丁锁,避免锁字典随绑定增删无上限累积 + _drop_runtime_patch_lock(binding_id) + return stopped def running_binding_ids(self) -> set[str]: with self._lock: diff --git a/backend/app/channels/adapters/wecom.py b/backend/app/channels/adapters/wecom.py index 6ee0176..b018556 100644 --- a/backend/app/channels/adapters/wecom.py +++ b/backend/app/channels/adapters/wecom.py @@ -8,7 +8,7 @@ from datetime import timedelta from typing import Any -from sqlalchemy import text, update +from sqlalchemy import update from sqlalchemy.pool import NullPool from sqlmodel import Session, create_engine, select @@ -19,6 +19,7 @@ ) from app.channels.crypto import decrypt_channel_secret from app.db import engine +from app.db.dialect import get_dialect from app.db.models import ChannelBinding, utc_now logger = logging.getLogger(__name__) @@ -409,16 +410,14 @@ def _set_connected( result = db.exec(statement.values(**values)) if result.rowcount == 1: if connected: - # 断开告警标记在重连成功时清除(允许下次再告警) - db.execute( - text( - "UPDATE channel_bindings " - "SET config_json = json_remove(config_json, '$.disconnect_alerted_at'), " - "updated_at = :updated_at " - "WHERE id = :binding_id" - ), - {"binding_id": binding_id, "updated_at": utc_now()}, - ) + # 断开告警标记在重连成功时清除(允许下次再告警);读-改-写,方言中立 + binding = db.get(ChannelBinding, binding_id) + if binding is not None: + binding.config_json = get_dialect( + db.get_bind().url.get_backend_name() + ).json_config_remove(binding.config_json, "disconnect_alerted_at") + binding.updated_at = utc_now() + db.add(binding) db.commit() else: db.rollback() diff --git a/backend/app/channels/feishu_manager.py b/backend/app/channels/feishu_manager.py index 7bc4f61..882474a 100644 --- a/backend/app/channels/feishu_manager.py +++ b/backend/app/channels/feishu_manager.py @@ -3,7 +3,6 @@ import logging import threading import time -from pathlib import Path from typing import Callable from sqlalchemy import update @@ -28,9 +27,12 @@ def __init__( ): self._engine = db_engine or engine database = self._engine.url.database - if self._engine.url.get_backend_name() != "sqlite" or not database or database == ":memory:": - raise RuntimeError("飞书长连接首版仅支持文件 SQLite") - self._database_path = Path(database).expanduser().resolve() + if self._engine.url.get_backend_name() == "sqlite" and ( + not database or database == ":memory:" + ): + raise RuntimeError("飞书长连接子进程无法共享 SQLite 内存数据库") + # 子进程按完整 SQLAlchemy URL 自建引擎(密码不遮蔽) + self._database_url = self._engine.url.render_as_string(hide_password=False) self._supervisor_factory = supervisor_factory self._supervisor: FeishuProcessSupervisor | None = None self._reconcile_seconds = reconcile_seconds @@ -45,7 +47,7 @@ def __init__( def _get_supervisor(self) -> FeishuProcessSupervisor: with self._lock: if self._supervisor is None: - self._supervisor = self._supervisor_factory(database_path=self._database_path) + self._supervisor = self._supervisor_factory(database_url=self._database_url) return self._supervisor def start(self) -> None: diff --git a/backend/app/channels/feishu_process.py b/backend/app/channels/feishu_process.py index b4aee4b..81035c4 100644 --- a/backend/app/channels/feishu_process.py +++ b/backend/app/channels/feishu_process.py @@ -1,5 +1,6 @@ from __future__ import annotations +import logging import multiprocessing import secrets import threading @@ -12,14 +13,19 @@ from pathlib import Path from typing import Any +from sqlalchemy.pool import NullPool +from sqlmodel import Session, create_engine + +from app.db.dialect import get_dialect from feishu_connector_worker import ( - BindingProcessLock, ConnectorChildSpec, PRODUCTION_RUNTIME, - binding_lock_path, + binding_lock_key, connector_child_entry, ) +logger = logging.getLogger(__name__) + class ConnectorState(str, Enum): ABSENT = "absent" @@ -52,6 +58,7 @@ def __init__( self, *, runtime_path: str = PRODUCTION_RUNTIME, + database_url: str | None = None, database_path: Path | None = None, data_dir: Path | None = None, watchdog_seconds: float = 2.5, @@ -68,7 +75,12 @@ def __init__( self._ctx = multiprocessing.get_context("spawn") self._runtime_path = runtime_path root = (data_dir or Path.cwd()).expanduser().resolve() - self._database_path = (database_path or root / "skill_agent_loop.db").expanduser().resolve() + # database_path 为兼容旧参数;子进程统一按完整 SQLAlchemy URL 自建引擎 + if database_url: + self._database_url = database_url + else: + legacy_path = (database_path or root / "skill_agent_loop.db").expanduser().resolve() + self._database_url = f"sqlite:///{legacy_path}" self._watchdog_seconds = watchdog_seconds self._terminate_grace_seconds = terminate_grace_seconds self._max_processes = max_processes @@ -83,7 +95,8 @@ def __init__( self._reserved_process_slots = 0 self._crashes: dict[str, deque[float]] = {} self._backoff_until: dict[str, float] = {} - self._pending_lock_paths: set[Path] = set() + self._pending_lock_bindings: set[str] = set() + self._probe_engine = None self._lock = threading.RLock() self._condition = threading.Condition(self._lock) self._global_stop_lock = threading.Lock() @@ -160,8 +173,7 @@ def _start_binding_core(self, binding_id: str, config_revision: int) -> Connecto def _spawn_reserved_binding( self, binding_id: str, config_revision: int ) -> ConnectorRecord: - lock_path = binding_lock_path(binding_id, self._database_path) - if not self._wait_for_binding_lock(lock_path, self._lock_wait_seconds): + if not self._wait_for_binding_lock(binding_id, self._lock_wait_seconds): raise RuntimeError(f"Feishu connector binding lock is still held: {binding_id}") nonce = secrets.token_hex(16) parent_connection, child_connection = self._ctx.Pipe(duplex=True) @@ -170,8 +182,7 @@ def _spawn_reserved_binding( config_revision=config_revision, child_nonce=nonce, runtime_path=self._runtime_path, - binding_lock_path=str(lock_path), - database_path=str(self._database_path), + database_url=self._database_url, watchdog_seconds=self._watchdog_seconds, ) process = self._ctx.Process( @@ -399,7 +410,7 @@ def stop_binding(self, binding_id: str, timeout: float = 5.0) -> bool: self._records.pop(binding_id, None) self._dispose_record(record) self._condition.notify_all() - return stopped and self._binding_lock_is_free(Path(record.spec.binding_lock_path)) + return stopped and self._binding_lock_is_free(record.spec.binding_id) def replace_binding( self, @@ -420,7 +431,7 @@ def replace_binding( current.state = ConnectorState.STOPPING if not self._stop_records([current], deadline): raise TimeoutError("old Feishu connector did not stop before reconfiguration") - if not self._binding_lock_is_free(Path(current.spec.binding_lock_path)): + if not self._binding_lock_is_free(current.spec.binding_id): raise TimeoutError("old Feishu connector binding lock is still held") with self._condition: self._drain_events(current) @@ -478,15 +489,17 @@ def stop(self, timeout: float = 5.0) -> bool: self._dispose_record(record) self._records.clear() self._condition.notify_all() - lock_paths = self._pending_lock_paths | { - Path(record.spec.binding_lock_path) for record in records + lock_bindings = self._pending_lock_bindings | { + record.spec.binding_id for record in records } - blocked_paths = { - path for path in lock_paths if not self._binding_lock_is_free(path) + blocked_bindings = { + binding_id + for binding_id in lock_bindings + if not self._binding_lock_is_free(binding_id) } - locks_free = not blocked_paths + locks_free = not blocked_bindings with self._condition: - self._pending_lock_paths = blocked_paths + self._pending_lock_bindings = blocked_bindings self._closed = stopped and monitor_stopped and locks_free self._condition.notify_all() return stopped and monitor_stopped and locks_free @@ -517,18 +530,35 @@ def _dispose_record(record: ConnectorRecord) -> None: except ValueError: pass - @staticmethod - def _binding_lock_is_free(path: Path) -> bool: - lock = BindingProcessLock(path) - if not lock.acquire(): + def _lock_probe_engine(self): + # 探测用缓存 engine:NullPool,每次探测开短会话,不常驻连接 + if self._probe_engine is None: + self._probe_engine = create_engine(self._database_url, poolclass=NullPool) + return self._probe_engine + + def _binding_lock_is_free(self, binding_id: str) -> bool: + """方言探测 binding 锁:短会话 try-acquire + release。 + + 探测本身失败(数据库不可达等)按"仍占用"保守处理:宁误拒启动, + 不冒双起 connector 的风险。 + """ + key = binding_lock_key(binding_id) + try: + engine = self._lock_probe_engine() + dialect = get_dialect(engine.url.get_backend_name()) + with Session(engine) as session: + if not dialect.acquire_advisory_lock(session, key): + return False + dialect.release_advisory_lock(session, key) + return True + except Exception: + logger.warning("飞书 binding 锁探测失败 binding=%s", binding_id, exc_info=True) return False - lock.release() - return True - def _wait_for_binding_lock(self, path: Path, timeout: float) -> bool: + def _wait_for_binding_lock(self, binding_id: str, timeout: float) -> bool: deadline = time.monotonic() + max(0.0, timeout) while True: - if self._binding_lock_is_free(path): + if self._binding_lock_is_free(binding_id): return True if time.monotonic() >= deadline: return False diff --git a/backend/app/channels/feishu_runtime.py b/backend/app/channels/feishu_runtime.py index 8288bf2..947cbd4 100644 --- a/backend/app/channels/feishu_runtime.py +++ b/backend/app/channels/feishu_runtime.py @@ -3,7 +3,6 @@ import asyncio import importlib.metadata import json -from pathlib import Path from sqlalchemy.pool import NullPool from sqlmodel import Session, create_engine @@ -124,12 +123,14 @@ def ignore_lifecycle_event(_event) -> None: def run_feishu_runtime(spec, control, watchdog) -> None: if importlib.metadata.version("lark-channel-sdk") != SDK_CONTRACT_VERSION: raise RuntimeError(f"lark-channel-sdk must be exactly {SDK_CONTRACT_VERSION}") - database_path = Path(spec.database_path).expanduser().resolve() - stage_engine = create_engine( - f"sqlite:///{database_path}", - connect_args={"check_same_thread": False, "timeout": 0.5}, - poolclass=NullPool, + # 子进程按完整 SQLAlchemy URL 自建引擎;非 SQLite 不传 check_same_thread + database_url = spec.database_url + connect_args = ( + {"check_same_thread": False, "timeout": 0.5} + if database_url.startswith("sqlite") + else {} ) + stage_engine = create_engine(database_url, connect_args=connect_args, poolclass=NullPool) with Session(stage_engine) as db: binding = db.get(ChannelBinding, spec.binding_id) if ( diff --git a/backend/app/config.py b/backend/app/config.py index 061eac6..5450629 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -8,6 +8,8 @@ class Settings(BaseSettings): app_name: str = "Skill Agent Loop Service" database_url: str = "sqlite:///./skill_agent_loop.db" app_secret: str = "change-me-in-development" + # 应用时区(Postgres 日分桶等按此换算;留空=服务器本地固定偏移;SQLite 恒为服务器本地) + app_timezone: str = "" demo_model_base_url: str = "http://localhost:52010/v1" demo_model_name: str = "qwen3.6-27b" demo_model_api_key: str = "" diff --git a/backend/app/db/database.py b/backend/app/db/database.py index 55dfe5c..88d2a30 100644 --- a/backend/app/db/database.py +++ b/backend/app/db/database.py @@ -2,13 +2,18 @@ from contextlib import contextmanager import hashlib import json +import logging from pathlib import Path from urllib.parse import unquote -from sqlalchemy import Engine, inspect, text -from sqlmodel import Session, SQLModel, create_engine +from sqlalchemy import Engine, func, inspect, text +from sqlalchemy.engine import make_url +from sqlmodel import Session, SQLModel, create_engine, select from app.config import get_settings +from app.db.dialect import get_dialect + +logger = logging.getLogger(__name__) def _normalize_database_url(url: str) -> str: @@ -31,8 +36,9 @@ def _normalize_database_url(url: str) -> str: settings = get_settings() database_url = _normalize_database_url(settings.database_url) -connect_args = {"check_same_thread": False, "timeout": 30} if database_url.startswith("sqlite") else {} -engine: Engine = create_engine(database_url, echo=False, connect_args=connect_args) +# 引擎创建参数由方言提供者收口(SQLite=check_same_thread/timeout,行为不变) +_dialect = get_dialect(make_url(database_url).get_backend_name()) +engine: Engine = create_engine(database_url, echo=False, **_dialect.engine_kwargs(database_url)) _DEFAULT_MODEL_OUTPUT_LIMIT_MIGRATION_ID = "20260712_default_model_output_tokens_8192" _LEGACY_DEFAULT_MODEL_OUTPUT_TOKENS = 2048 @@ -66,9 +72,64 @@ def _normalize_database_url(url: str) -> str: def init_db() -> None: import app.db.models # noqa: F401 + _ensure_driver_available() _configure_sqlite_runtime() SQLModel.metadata.create_all(engine) _migrate_sqlite_skill_schema() + _validate_default_model_invariant() + _warn_backend_constraints() + + +def _ensure_driver_available() -> None: + """PG/高斯驱动为可选依赖:缺失时在启动处给出明确安装指引,而不是连接期栈trace。""" + if engine.url.get_backend_name() != "postgresql": + return + try: + import psycopg # noqa: F401 + except ImportError as exc: + raise RuntimeError( + "PostgreSQL/openGauss 需要可选驱动:pip install 'skill-agent-loop-backend[postgres]'" + ) from exc + + +def _validate_default_model_invariant() -> None: + """不支持部分唯一索引的后端:启动校验"每租户至多一条默认模型"未被破坏。 + + 该类后端不会创建 uq_model_configs_tenant_default(见 models.py 的按方言 DDL), + 约束退化为代码层维护(默认切换在同事务先清后设)+ 此处启动兜底。 + """ + if _dialect.supports_partial_index: + return + from app.db.models import ModelConfig + + with Session(engine) as db: + duplicates = db.exec( + select(ModelConfig.tenant_id) + .where(ModelConfig.is_default == True) # noqa: E712 + .group_by(ModelConfig.tenant_id) + .having(func.count() > 1) + ).all() + if duplicates: + raise RuntimeError( + "当前数据库后端不支持部分唯一索引,且检测到以下租户存在多条默认模型: " + f"{sorted(duplicates)};请人工清理至每租户至多一条后再启动" + ) + + +def _warn_backend_constraints() -> None: + """非 SQLite 后端的显式约束与配置陷阱告警(启动期一次性)。""" + backend = engine.url.get_backend_name() + if backend == "sqlite": + if (get_settings().app_timezone or "").strip(): + logger.warning( + "app_timezone 对 SQLite 不生效(日分桶恒为服务器本地时区),该配置将被忽略" + ) + return + logger.warning( + "数据库后端 %s 当前仅支持全新库初始化(create_all),无存量 schema 迁移通路;" + "版本升级若涉及 models 变更需人工对齐 schema(Alembic 迁移为后续任务)", + backend, + ) def _configure_sqlite_runtime() -> None: diff --git a/backend/app/db/dialect.py b/backend/app/db/dialect.py new file mode 100644 index 0000000..9c6694d --- /dev/null +++ b/backend/app/db/dialect.py @@ -0,0 +1,330 @@ +"""数据库方言可插拔层. + +ORM 优先:能用 SQLAlchemy ORM/标准 SQL 表达的一律走通用实现,无法通用的 +少数方言点收口到本模块的 DatabaseDialect 提供者。新增数据库 = 新增一个小 +适配器并 register_dialect 注册,业务代码零改动: + +- postgresql(含高斯):全能力,即 PostgresDialect(psycopg3 驱动,URL + postgresql+psycopg://);高斯直接复用该适配器。 +- mysql:不预装驱动(按需装 pymysql);不支持部分唯一索引—— + models.py 的 uq_model_configs_tenant_default 只在 sqlite/postgresql 方言 + 创建,其它后端由 init_db 的启动校验兜底;advisory lock 可用 GET_LOCK(key, 0) + 实现,未实现前文件锁会响亮拒绝(见 BaseDialect.acquire_advisory_lock)。 +- dm(达梦):dmPython 驱动(dm+dmPython://),Oracle 系语法;同样不支持 + 部分唯一索引;无原生锁实现前同样响亮拒绝。 + +未注册的后端名回退 GenericDialect:ORM 通用实现开箱可用;适配器只补充各自 +特性(原生 advisory lock、部分索引、日期函数等)。 + +日分桶口径注意:SQLite 为服务器本地自然日(func.date(col,'localtime')); +Postgres 按 app_timezone(缺省=服务器本地固定偏移);Generic 为 +cast(col, Date) 即数据库服务器时区自然日(云上实例常为 UTC),与 SQLite +口径可能不同——跨库迁移数据后按日统计会整体平移,属预期差异。 +""" + +from __future__ import annotations + +import logging +import os +from datetime import datetime, timedelta +from pathlib import Path +from typing import Any, Protocol, runtime_checkable + +from sqlalchemy import Date, cast, func, literal, text +from sqlalchemy.types import Interval + +from app.config import get_settings + +logger = logging.getLogger(__name__) + + +@runtime_checkable +class DatabaseDialect(Protocol): + """数据库方言提供者:引擎参数、日分桶、JSON 配置读改写、advisory lock。""" + + name: str # sqlite / postgresql / mysql / dm / ... + supports_partial_index: bool # 部分唯一索引(WHERE 子句)能力 + session_scoped_advisory_lock: bool # advisory lock 是否随连接存活(方言须常驻专属连接) + + def engine_kwargs(self, url: str) -> dict[str, Any]: ... + + def day_bucket(self, column) -> Any: ... + + def json_config_get(self, config: dict | None, key: str) -> Any: ... + + def json_config_set(self, config: dict | None, key: str, value: Any) -> dict: ... + + def json_config_remove(self, config: dict | None, key: str) -> dict: ... + + def acquire_advisory_lock(self, session, key: str) -> bool: ... + + def release_advisory_lock(self, session, key: str) -> None: ... + + def check_advisory_lock(self, session, key: str) -> bool: ... + + +class BaseDialect: + """通用默认实现:JSON 读改写 = Python 侧读-改-写;日分桶 = cast(col, Date) + (标准 SQL,按数据库服务器时区取日);advisory lock = SQLite 数据库文件旁的 + 文件锁(非 SQLite 文件库响亮拒绝,由具体适配器补原生锁实现)。""" + + name = "generic" + supports_partial_index = False + session_scoped_advisory_lock = False + + def __init__(self, backend_name: str = "generic") -> None: + self.name = backend_name + self._lock_handles: dict[str, tuple[int, Any]] = {} + + def engine_kwargs(self, url: str) -> dict[str, Any]: + return {} + + def day_bucket(self, column) -> Any: + return cast(column, Date) + + def json_config_get(self, config: dict | None, key: str) -> Any: + return dict(config or {}).get(key) + + def json_config_set(self, config: dict | None, key: str, value: Any) -> dict: + patched = dict(config or {}) + patched[key] = value + return patched + + def json_config_remove(self, config: dict | None, key: str) -> dict: + patched = dict(config or {}) + patched.pop(key, None) + return patched + + def acquire_advisory_lock(self, session, key: str) -> bool: + """SQLite 数据库文件旁的文件锁;已持有同 key 锁时重入成功。 + + 只对 SQLite 文件库有效:url.database 对 MySQL/达梦等是库名而非文件路径, + 直接拒绝(响亮失败),避免把库名当路径在 CWD 下生成各进程互不可见的锁文件, + 静默破坏单实例保证。 + + fork 防护:继承自父进程的句柄不算持有(只关闭不解锁——解锁会把父进程 + 的锁一起放掉),随后按本进程身份真实抢锁。 + """ + held = self._lock_handles.get(key) + if held is not None: + held_pid, held_handle = held + if held_pid == os.getpid(): + return True + held_handle.close() + self._lock_handles.pop(key, None) + bind = session.get_bind() + url = getattr(bind, "url", None) + backend_name = url.get_backend_name() if url is not None else "" + database_path = getattr(url, "database", None) + if backend_name != "sqlite" or not database_path or database_path == ":memory:": + logger.error( + "方言 %s 无原生 advisory lock 且数据库非 SQLite 文件库,无法提供进程锁;" + "请为该后端实现原生 advisory lock 适配器", + self.name, + ) + return False + lock_path = ( + Path(database_path).resolve().with_name(f"{Path(database_path).name}.{key}.lock") + ) + handle = lock_path.open("a+b") + try: + if os.name == "nt": + import msvcrt + + handle.seek(0) + if handle.read(1) == b"": + handle.write(b"0") + handle.flush() + handle.seek(0) + msvcrt.locking(handle.fileno(), msvcrt.LK_NBLCK, 1) + else: + import fcntl + + fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + except (BlockingIOError, OSError): + handle.close() + return False + self._lock_handles[key] = (os.getpid(), handle) + return True + + def release_advisory_lock(self, session, key: str) -> None: + held = self._lock_handles.pop(key, None) + if held is None: + return + held_pid, handle = held + if held_pid != os.getpid(): + # fork 子进程:仅关闭继承句柄,不解父进程的锁 + handle.close() + return + try: + if os.name == "nt": + import msvcrt + + handle.seek(0) + msvcrt.locking(handle.fileno(), msvcrt.LK_UNLCK, 1) + else: + import fcntl + + fcntl.flock(handle.fileno(), fcntl.LOCK_UN) + finally: + handle.close() + + def check_advisory_lock(self, session, key: str) -> bool: + """文件锁无静默失效模式(句柄随进程存活):只校验本进程仍持有句柄。""" + held = self._lock_handles.get(key) + return held is not None and held[0] == os.getpid() and not held[1].closed + + +class SQLiteDialect(BaseDialect): + """SQLite:文件锁 + date(localtime) 日分桶(均为现行行为,保持不变)。""" + + supports_partial_index = True + + def __init__(self) -> None: + super().__init__("sqlite") + + def engine_kwargs(self, url: str) -> dict[str, Any]: + return {"connect_args": {"check_same_thread": False, "timeout": 30}} + + def day_bucket(self, column) -> Any: + # 服务器本地时区自然日(既有行为) + return func.date(column, "localtime") + + +class PostgresDialect(BaseDialect): + """PostgreSQL/高斯:psycopg3 驱动(postgresql+psycopg://),全能力。 + + advisory lock 属于**连接**而非会话:持锁期间由本方言常驻一条专属 + `engine.connect()` Connection(AUTOCOMMIT——每条语句即提即放,不留 + idle-in-transaction;Connection 不归池,commit/rollback 不脱钩, + `pg_backend_pid()` 全程稳定)。绝不能用 Session.commit() 代替: + commit 会把连接归还池中,锁静默释放或后续 check 取到别的连接。""" + + supports_partial_index = True + session_scoped_advisory_lock = True + + def __init__(self) -> None: + super().__init__("postgresql") + # key -> (pid, Connection);fork 防护同 BaseDialect 文件锁 + self._pg_lock_conns: dict[str, tuple[int, Any]] = {} + + def engine_kwargs(self, url: str) -> dict[str, Any]: + return { + # 陈旧连接(PG 重启/idle 超时后)首用前先 ping,失效即重连而非报错 + "pool_pre_ping": True, + # 主动回收长闲连接,避开服务端/LB 的 idle 掐断 + "pool_recycle": 1800, + # 后台线程随 binding 数增长,默认 5+10 不够 + "pool_size": 10, + "max_overflow": 20, + } + + def day_bucket(self, column) -> Any: + tz = (get_settings().app_timezone or "").strip() + if tz: + # created_at 为 naive UTC:先按 UTC 还原时刻,再换算到应用时区取自然日 + return cast(func.timezone(tz, func.timezone("UTC", column)), Date) + # 未配置时区:按服务器本地固定偏移换算(与 SQLite localtime 语义对齐) + offset = datetime.now().astimezone().utcoffset() or timedelta(0) + interval = literal(offset, type_=Interval) + return cast(column + interval, Date) + + def _engine_of(self, session): + bind = session.get_bind() + return getattr(bind, "engine", bind) + + def acquire_advisory_lock(self, session, key: str) -> bool: + """以专属 AUTOCOMMIT Connection 取锁并常驻;重入成功,fork 继承不作数。""" + held = self._pg_lock_conns.get(key) + if held is not None: + held_pid, held_conn = held + if held_pid == os.getpid(): + return True + # fork 子进程:继承的连接只丢弃引用,不在共享 socket 上做任何操作 + self._pg_lock_conns.pop(key, None) + # 双 int4 键形式(classid=hashtext(key), objid=0):便于在 pg_locks 中直接校验 + conn = self._engine_of(session).connect().execution_options( + isolation_level="AUTOCOMMIT" + ) + try: + locked = conn.execute( + text("SELECT pg_try_advisory_lock(hashtext(:key), 0)"), + {"key": key}, + ).scalar() + except Exception: + conn.close() + raise + if not locked: + conn.close() + return False + self._pg_lock_conns[key] = (os.getpid(), conn) + return True + + def release_advisory_lock(self, session, key: str) -> None: + held = self._pg_lock_conns.pop(key, None) + if held is None: + return + held_pid, conn = held + if held_pid != os.getpid(): + # fork 子进程:仅丢弃引用,不解父进程的锁 + return + try: + conn.execute(text("SELECT pg_advisory_unlock(hashtext(:key), 0)"), {"key": key}) + finally: + conn.close() + + def check_advisory_lock(self, session, key: str) -> bool: + """校验本进程常驻的那条连接仍持有该 advisory lock。 + + 必须在**同一条专属 Connection** 上查 pg_locks(pid 稳定);连接被服务端 + 掐断后锁已静默释放,查询会抛错——同样判 False。 + """ + held = self._pg_lock_conns.get(key) + if held is None: + return False + held_pid, conn = held + if held_pid != os.getpid(): + return False + try: + row = conn.execute( + text( + "SELECT 1 FROM pg_locks " + "WHERE locktype = 'advisory' AND pid = pg_backend_pid() " + "AND classid = hashtext(:key) AND objid = 0" + ), + {"key": key}, + ).first() + return row is not None + except Exception: + logger.exception("PG advisory lock 存活校验失败 key=%s", key) + return False + + +class GenericDialect(BaseDialect): + """未注册后端的回退:ORM 通用实现;文件锁仅对 SQLite 文件库有效。 + + 扩展新数据库(MySQL/达梦等):继承 BaseDialect 按能力覆写后 + register_dialect("") 注册即可。supports_partial_index=False 的 + 后端不会创建 models.py 的部分唯一索引,由 init_db 启动校验兜底; + 未实现原生 advisory lock 时 BaseDialect 会响亮拒绝而非静默错锁。 + """ + + +_DIALECTS: dict[str, DatabaseDialect] = {} + + +def register_dialect(backend_name: str, dialect: DatabaseDialect) -> None: + """注册方言提供者(backend_name 即 engine.url.get_backend_name())。""" + _DIALECTS[backend_name] = dialect + + +def get_dialect(backend_name: str) -> DatabaseDialect: + """按 SQLAlchemy backend 名取方言提供者;未注册回退 GenericDialect。""" + dialect = _DIALECTS.get(backend_name) + if dialect is None: + dialect = _DIALECTS.setdefault(backend_name, GenericDialect(backend_name)) + return dialect + + +register_dialect("sqlite", SQLiteDialect()) +register_dialect("postgresql", PostgresDialect()) diff --git a/backend/app/db/models.py b/backend/app/db/models.py index df0ba06..7be08ff 100644 --- a/backend/app/db/models.py +++ b/backend/app/db/models.py @@ -1,10 +1,12 @@ from __future__ import annotations +import json from datetime import UTC, datetime from typing import Any, Optional from uuid import uuid4 -from sqlalchemy import Column, Index, Integer, JSON, UniqueConstraint +from sqlalchemy import Column, DDL, Index, Integer, JSON, Text, TypeDecorator, UniqueConstraint, event +from sqlalchemy.dialects.mysql import MEDIUMTEXT from sqlmodel import Field, SQLModel @@ -16,6 +18,34 @@ def new_id(prefix: str) -> str: return f"{prefix}_{uuid4().hex[:16]}" +class PortableJSON(TypeDecorator): + """JSON 列的可移植实现:sqlite/postgresql/mysql 用原生 JSON;Oracle/达梦系 + 无 JSON 类型,降级为 CLOB 存序列化文本(读出自动还原,Python 侧透明)。""" + + impl = JSON + cache_ok = True + + def load_dialect_impl(self, dialect): + if dialect.name in ("oracle", "dm"): + return dialect.type_descriptor(Text()) + return dialect.type_descriptor(JSON()) + + def process_bind_param(self, value, dialect): + if value is not None and dialect.name in ("oracle", "dm"): + return json.dumps(value, ensure_ascii=False) + return value + + def process_result_value(self, value, dialect): + if value is not None and dialect.name in ("oracle", "dm"): + return json.loads(value) + return value + + +# 超过 TEXT(64KB)上限的大字段(如 base64 头像 ~2.8MB):MySQL 用 MEDIUMTEXT, +# 其它方言 TEXT 本身无界(SQLite/PG)或映射 CLOB(Oracle/达梦) +HugeText = Text().with_variant(MEDIUMTEXT(), "mysql") + + class Tenant(SQLModel, table=True): __tablename__ = "tenants" @@ -50,7 +80,7 @@ class UserAvatar(SQLModel, table=True): __tablename__ = "user_avatars" user_id: str = Field(primary_key=True) - data_url: str + data_url: str = Field(sa_column=Column(HugeText)) updated_at: datetime = Field(default_factory=utc_now) @@ -63,9 +93,9 @@ class Skill(SQLModel, table=True): skill_id: str = Field(index=True) version: str = "1.0.0" name: str - business_domain: Optional[str] = None - description: Optional[str] = None - content_json: dict[str, Any] = Field(sa_column=Column(JSON, nullable=False)) + business_domain: Optional[str] = Field(default=None, sa_column=Column(Text)) + description: Optional[str] = Field(default=None, sa_column=Column(Text)) + content_json: dict[str, Any] = Field(sa_column=Column(PortableJSON, nullable=False)) status: str = Field(default="draft", index=True) created_at: datetime = Field(default_factory=utc_now) updated_at: datetime = Field(default_factory=utc_now) @@ -80,9 +110,9 @@ class SkillVersion(SQLModel, table=True): skill_id: str = Field(index=True) version: str = Field(index=True) name: str - business_domain: Optional[str] = None - description: Optional[str] = None - content_json: dict[str, Any] = Field(sa_column=Column(JSON, nullable=False)) + business_domain: Optional[str] = Field(default=None, sa_column=Column(Text)) + description: Optional[str] = Field(default=None, sa_column=Column(Text)) + content_json: dict[str, Any] = Field(sa_column=Column(PortableJSON, nullable=False)) status: str = Field(default="draft", index=True) created_at: datetime = Field(default_factory=utc_now) updated_at: datetime = Field(default_factory=utc_now) @@ -101,10 +131,10 @@ class AgentSkillBranch(SQLModel, table=True): source_skill_id: str = Field(index=True) base_version: str = "1.0.0" head_version: str = "1.0.0" - content_json: dict[str, Any] = Field(sa_column=Column(JSON, nullable=False)) + content_json: dict[str, Any] = Field(sa_column=Column(PortableJSON, nullable=False)) status: str = Field(default="active", index=True) sync_state: str = Field(default="synced", index=True) - metadata_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(JSON)) + metadata_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(PortableJSON)) created_at: datetime = Field(default_factory=utc_now) updated_at: datetime = Field(default_factory=utc_now) @@ -122,10 +152,10 @@ class AgentSkillBranchVersion(SQLModel, table=True): source_skill_id: str = Field(index=True) version: str = Field(index=True) base_version: str = "1.0.0" - content_json: dict[str, Any] = Field(sa_column=Column(JSON, nullable=False)) + content_json: dict[str, Any] = Field(sa_column=Column(PortableJSON, nullable=False)) status: str = Field(default="active", index=True) sync_state: str = Field(default="diverged", index=True) - change_summary: Optional[str] = None + change_summary: Optional[str] = Field(default=None, sa_column=Column(Text)) created_at: datetime = Field(default_factory=utc_now) updated_at: datetime = Field(default_factory=utc_now) @@ -138,14 +168,14 @@ class GeneralSkill(SQLModel, table=True): tenant_id: str = Field(index=True) slug: str = Field(index=True) name: str - description: Optional[str] = None - homepage: Optional[str] = None - skill_markdown: str - skill_files_json: list[dict[str, Any]] = Field(default_factory=list, sa_column=Column(JSON)) - metadata_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(JSON)) + description: Optional[str] = Field(default=None, sa_column=Column(Text)) + homepage: Optional[str] = Field(default=None, sa_column=Column(Text)) + skill_markdown: str = Field(sa_column=Column(Text)) + skill_files_json: list[dict[str, Any]] = Field(default_factory=list, sa_column=Column(PortableJSON)) + metadata_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(PortableJSON)) status: str = Field(default="draft", index=True) - permissions_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(JSON)) - runtime_config_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(JSON)) + permissions_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(PortableJSON)) + runtime_config_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(PortableJSON)) created_at: datetime = Field(default_factory=utc_now) updated_at: datetime = Field(default_factory=utc_now) @@ -157,9 +187,9 @@ class KnowledgeBase(SQLModel, table=True): id: str = Field(default_factory=lambda: new_id("kb"), primary_key=True) tenant_id: str = Field(index=True) name: str - description: Optional[str] = None + description: Optional[str] = Field(default=None, sa_column=Column(Text)) status: str = Field(default="active", index=True) - metadata_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(JSON)) + metadata_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(PortableJSON)) created_at: datetime = Field(default_factory=utc_now) updated_at: datetime = Field(default_factory=utc_now) @@ -175,9 +205,9 @@ class KnowledgeBaseVersion(SQLModel, table=True): knowledge_base_id: str = Field(index=True) version: str = Field(default="1.0.0", index=True) name: str - description: Optional[str] = None + description: Optional[str] = Field(default=None, sa_column=Column(Text)) status: str = Field(default="active", index=True) - metadata_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(JSON)) + metadata_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(PortableJSON)) created_at: datetime = Field(default_factory=utc_now) updated_at: datetime = Field(default_factory=utc_now) @@ -196,7 +226,7 @@ class AgentKnowledgeBranch(SQLModel, table=True): head_version: str = "1.0.0" status: str = Field(default="active", index=True) sync_state: str = Field(default="synced", index=True) - metadata_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(JSON)) + metadata_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(PortableJSON)) created_at: datetime = Field(default_factory=utc_now) updated_at: datetime = Field(default_factory=utc_now) @@ -210,12 +240,12 @@ class KnowledgeDocument(SQLModel, table=True): knowledge_base_version_id: Optional[str] = Field(default=None, index=True) filename: str file_type: str = Field(index=True) - title: Optional[str] = None + title: Optional[str] = Field(default=None, sa_column=Column(Text)) status: str = Field(default="processing", index=True) bucket_count: int = 0 chunk_count: int = 0 - metadata_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(JSON)) - error: Optional[str] = None + metadata_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(PortableJSON)) + error: Optional[str] = Field(default=None, sa_column=Column(Text)) created_at: datetime = Field(default_factory=utc_now) updated_at: datetime = Field(default_factory=utc_now) @@ -229,10 +259,10 @@ class KnowledgeBucket(SQLModel, table=True): knowledge_base_version_id: Optional[str] = Field(default=None, index=True) document_id: str = Field(index=True) bucket_key: str = Field(index=True) - title: str + title: str = Field(sa_column=Column(Text)) summary: str token_estimate: int = 0 - metadata_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(JSON)) + metadata_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(PortableJSON)) created_at: datetime = Field(default_factory=utc_now) updated_at: datetime = Field(default_factory=utc_now) @@ -247,10 +277,10 @@ class KnowledgeChunk(SQLModel, table=True): document_id: str = Field(index=True) bucket_id: str = Field(index=True) chunk_index: int = Field(index=True) - content: str - summary: Optional[str] = None + content: str = Field(sa_column=Column(Text)) + summary: Optional[str] = Field(default=None, sa_column=Column(Text)) source_ref: Optional[str] = None - metadata_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(JSON)) + metadata_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(PortableJSON)) created_at: datetime = Field(default_factory=utc_now) updated_at: datetime = Field(default_factory=utc_now) @@ -273,13 +303,13 @@ class KnowledgeConcept(SQLModel, table=True): document_id: Optional[str] = Field(default=None, index=True) concept_id: str = Field(index=True) concept_type: str = Field(index=True) - title: str - description: Optional[str] = None - content_md: str - frontmatter_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(JSON)) - links_json: list[dict[str, Any]] = Field(default_factory=list, sa_column=Column(JSON)) - citations_json: list[dict[str, Any]] = Field(default_factory=list, sa_column=Column(JSON)) - source_refs_json: list[dict[str, Any]] = Field(default_factory=list, sa_column=Column(JSON)) + title: str = Field(sa_column=Column(Text)) + description: Optional[str] = Field(default=None, sa_column=Column(Text)) + content_md: str = Field(sa_column=Column(Text)) + frontmatter_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(PortableJSON)) + links_json: list[dict[str, Any]] = Field(default_factory=list, sa_column=Column(PortableJSON)) + citations_json: list[dict[str, Any]] = Field(default_factory=list, sa_column=Column(PortableJSON)) + source_refs_json: list[dict[str, Any]] = Field(default_factory=list, sa_column=Column(PortableJSON)) status: str = Field(default="active", index=True) created_at: datetime = Field(default_factory=utc_now) updated_at: datetime = Field(default_factory=utc_now) @@ -295,11 +325,11 @@ class KnowledgeDiscoverySuggestion(SQLModel, table=True): document_id: str = Field(index=True) bucket_id: Optional[str] = Field(default=None, index=True) suggestion_type: str = Field(index=True) - title: str + title: str = Field(sa_column=Column(Text)) status: str = Field(default="pending", index=True) - payload_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(JSON)) - source_refs_json: list[dict[str, Any]] = Field(default_factory=list, sa_column=Column(JSON)) - reason: Optional[str] = None + payload_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(PortableJSON)) + source_refs_json: list[dict[str, Any]] = Field(default_factory=list, sa_column=Column(PortableJSON)) + reason: Optional[str] = Field(default=None, sa_column=Column(Text)) created_at: datetime = Field(default_factory=utc_now) updated_at: datetime = Field(default_factory=utc_now) @@ -316,8 +346,8 @@ class KnowledgeIngestJob(SQLModel, table=True): status: str = Field(default="queued", index=True) stage: str = "queued" progress: float = 0.0 - error: Optional[str] = None - metadata_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(JSON)) + error: Optional[str] = Field(default=None, sa_column=Column(Text)) + metadata_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(PortableJSON)) created_at: datetime = Field(default_factory=utc_now) started_at: Optional[datetime] = None finished_at: Optional[datetime] = None @@ -332,15 +362,15 @@ class ModelConfig(SQLModel, table=True): name: str provider: str = "openai_compatible" api_protocol: str = Field(default="openai_chat_completions", index=True) - base_url: Optional[str] = None - api_key_encrypted: str + base_url: Optional[str] = Field(default=None, sa_column=Column(Text)) + api_key_encrypted: str = Field(sa_column=Column(Text)) model: str temperature: float = 0.2 max_output_tokens: int = 8192 - extra_body_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(JSON)) - protocol_options_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(JSON)) + extra_body_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(PortableJSON)) + protocol_options_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(PortableJSON)) legacy_unmapped_options_json: dict[str, Any] = Field( - default_factory=dict, sa_column=Column(JSON) + default_factory=dict, sa_column=Column(PortableJSON) ) trust_status: str = Field(default="unverified", index=True) verified_at: Optional[datetime] = None @@ -358,11 +388,33 @@ class ModelConfig(SQLModel, table=True): updated_at: datetime = Field(default_factory=utc_now) +# 每租户至多一条默认模型:部分唯一索引只按方言显式创建(SQLite=1 字面量,PG=布尔)。 +# 不把该索引挂在 metadata 上:sqlite_where/postgresql_where 在其它方言下会被静默 +# 丢弃,退化为全量唯一索引(MySQL/达梦下变成"每租户仅一条模型配置");不支持 +# 部分索引的后端由 init_db 的启动校验兜底(见 database.py)。 +event.listen( + ModelConfig.__table__, + "after_create", + DDL( + "CREATE UNIQUE INDEX uq_model_configs_tenant_default " + "ON model_configs (tenant_id) WHERE is_default = 1" + ).execute_if(dialect="sqlite"), +) +event.listen( + ModelConfig.__table__, + "after_create", + DDL( + "CREATE UNIQUE INDEX uq_model_configs_tenant_default " + "ON model_configs (tenant_id) WHERE is_default" + ).execute_if(dialect="postgresql"), +) + + class PersonaConfig(SQLModel, table=True): __tablename__ = "persona_configs" tenant_id: str = Field(primary_key=True) - system_prompt: str + system_prompt: str = Field(sa_column=Column(Text)) created_at: datetime = Field(default_factory=utc_now) updated_at: datetime = Field(default_factory=utc_now) @@ -387,11 +439,11 @@ class AgentProfile(SQLModel, table=True): id: str = Field(default_factory=lambda: new_id("agent"), primary_key=True) tenant_id: str = Field(index=True) name: str - description: Optional[str] = None - persona_prompt: Optional[str] = None + description: Optional[str] = Field(default=None, sa_column=Column(Text)) + persona_prompt: Optional[str] = Field(default=None, sa_column=Column(Text)) is_overall: bool = Field(default=False, index=True) status: str = Field(default="active", index=True) - metadata_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(JSON)) + metadata_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(PortableJSON)) created_at: datetime = Field(default_factory=utc_now) updated_at: datetime = Field(default_factory=utc_now) @@ -406,7 +458,7 @@ class AgentUsage(SQLModel, table=True): tenant_id: str = Field(index=True) user_id: str = Field(index=True) agent_id: str = Field(index=True) - metadata_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(JSON)) + metadata_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(PortableJSON)) created_at: datetime = Field(default_factory=utc_now) updated_at: datetime = Field(default_factory=utc_now) @@ -438,7 +490,7 @@ class AgentResourceBinding(SQLModel, table=True): resource_type: str = Field(index=True) resource_id: str = Field(index=True) status: str = Field(default="active", index=True) - metadata_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(JSON)) + metadata_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(PortableJSON)) created_at: datetime = Field(default_factory=utc_now) updated_at: datetime = Field(default_factory=utc_now) @@ -451,17 +503,17 @@ class Tool(SQLModel, table=True): tenant_id: str = Field(index=True) name: str = Field(index=True) display_name: Optional[str] = None - description: Optional[str] = None + description: Optional[str] = Field(default=None, sa_column=Column(Text)) bucket: str = Field(default="未分桶", index=True) tool_type: str = Field(default="http", index=True) method: str - url: str - headers_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(JSON)) - auth_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(JSON)) - config_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(JSON)) - input_schema: dict[str, Any] = Field(default_factory=dict, sa_column=Column(JSON)) - output_schema: dict[str, Any] = Field(default_factory=dict, sa_column=Column(JSON)) - allowed_skills_json: list[str] = Field(default_factory=list, sa_column=Column(JSON)) + url: str = Field(sa_column=Column(Text)) + headers_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(PortableJSON)) + auth_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(PortableJSON)) + config_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(PortableJSON)) + input_schema: dict[str, Any] = Field(default_factory=dict, sa_column=Column(PortableJSON)) + output_schema: dict[str, Any] = Field(default_factory=dict, sa_column=Column(PortableJSON)) + allowed_skills_json: list[str] = Field(default_factory=list, sa_column=Column(PortableJSON)) mcp_server_id: Optional[str] = Field(default=None, index=True) enabled: bool = True created_at: datetime = Field(default_factory=utc_now) @@ -476,20 +528,20 @@ class MCPServer(SQLModel, table=True): tenant_id: str = Field(index=True) name: str = Field(index=True) display_name: Optional[str] = None - description: Optional[str] = None + description: Optional[str] = Field(default=None, sa_column=Column(Text)) bucket: str = Field(default="MCP 工具", index=True) # 连接方式：stdio / streamable_http / sse / builtin transport: str = Field(default="streamable_http", index=True) # streamable_http / sse 使用 - url: Optional[str] = None - headers_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(JSON)) + url: Optional[str] = Field(default=None, sa_column=Column(Text)) + headers_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(PortableJSON)) # stdio 使用 - command: Optional[str] = None - args_json: list[str] = Field(default_factory=list, sa_column=Column(JSON)) - env_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(JSON)) - cwd: Optional[str] = None + command: Optional[str] = Field(default=None, sa_column=Column(Text)) + args_json: list[str] = Field(default_factory=list, sa_column=Column(PortableJSON)) + env_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(PortableJSON)) + cwd: Optional[str] = Field(default=None, sa_column=Column(Text)) # 最近一次发现的原始工具定义（预览/审计用） - discovered_tools_json: list[dict[str, Any]] = Field(default_factory=list, sa_column=Column(JSON)) + discovered_tools_json: list[dict[str, Any]] = Field(default_factory=list, sa_column=Column(PortableJSON)) last_synced_at: Optional[datetime] = None enabled: bool = True created_at: datetime = Field(default_factory=utc_now) @@ -511,34 +563,46 @@ class MockOrder(SQLModel, table=True): refundable: bool = True total_amount: float = 0.0 currency: str = "CNY" - metadata_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(JSON)) + metadata_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(PortableJSON)) created_at: datetime = Field(default_factory=utc_now) updated_at: datetime = Field(default_factory=utc_now) class ChatSession(SQLModel, table=True): __tablename__ = "sessions" + __table_args__ = ( + # SQLite/PG 唯一索引中 NULL 互不相等,web 会话(channel 为空)不受约束; + # 含 channel_binding_id 以隔离同企业多 Bot(与 SQLite 迁移中的同名索引一致) + Index( + "uq_sessions_agent_channel_extconv", + "agent_id", + "channel", + "channel_binding_id", + "external_conv_id", + unique=True, + ), + ) id: str = Field(primary_key=True) tenant_id: str = Field(index=True) user_id: Optional[str] = Field(default=None, index=True) agent_id: Optional[str] = Field(default=None, index=True) - title: Optional[str] = None + title: Optional[str] = Field(default=None, sa_column=Column(Text)) active_skill_id: Optional[str] = None active_step_id: Optional[str] = None - slots_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(JSON)) - skill_stack_json: list[dict[str, Any]] = Field(default_factory=list, sa_column=Column(JSON)) - pending_tasks_json: list[dict[str, Any]] = Field(default_factory=list, sa_column=Column(JSON)) - resume_after_answer_json: Optional[dict[str, Any]] = Field(default=None, sa_column=Column(JSON)) - awaiting_input_json: Optional[dict[str, Any]] = Field(default=None, sa_column=Column(JSON)) - knowledge_context_json: list[dict[str, Any]] = Field(default_factory=list, sa_column=Column(JSON)) - context_state_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(JSON)) - summary: Optional[str] = None - last_agent_question: Optional[str] = None + slots_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(PortableJSON)) + skill_stack_json: list[dict[str, Any]] = Field(default_factory=list, sa_column=Column(PortableJSON)) + pending_tasks_json: list[dict[str, Any]] = Field(default_factory=list, sa_column=Column(PortableJSON)) + resume_after_answer_json: Optional[dict[str, Any]] = Field(default=None, sa_column=Column(PortableJSON)) + awaiting_input_json: Optional[dict[str, Any]] = Field(default=None, sa_column=Column(PortableJSON)) + knowledge_context_json: list[dict[str, Any]] = Field(default_factory=list, sa_column=Column(PortableJSON)) + context_state_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(PortableJSON)) + summary: Optional[str] = Field(default=None, sa_column=Column(Text)) + last_agent_question: Optional[str] = Field(default=None, sa_column=Column(Text)) status: str = "active" channel: Optional[str] = None external_conv_id: Optional[str] = None - channel_target_json: Optional[dict[str, Any]] = Field(default=None, sa_column=Column(JSON)) + channel_target_json: Optional[dict[str, Any]] = Field(default=None, sa_column=Column(PortableJSON)) # 渠道会话直挂绑定:出站 staging 优先按它直查,不再靠 (agent_id, channel) 反查 channel_binding_id: Optional[str] = None # 渠道外部账号稳定键:绑定删除后仍保留,仅允许同一外部 Bot 精确认领历史会话 @@ -557,9 +621,9 @@ class ChannelBinding(SQLModel, table=True): # pending/active/expired/disabled status: str = Field(default="pending", index=True) # Fernet 加密后的渠道凭证（如微信 bot_token），绝不回传明文 - credentials_enc: Optional[str] = None + credentials_enc: Optional[str] = Field(default=None, sa_column=Column(Text)) # ilink_bot_id、baseurl、get_updates_buf 游标、session_expired、bound_at 等 - config_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(JSON)) + config_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(PortableJSON)) # provider 侧 Bot 的稳定连接键,全部署唯一;pending 绑定激活前允许为空 external_account_key: Optional[str] = Field(default=None, unique=True, index=True) # 身份作用域稳定键:企微为 corp_id,微信为空字符串 @@ -672,7 +736,7 @@ class ChannelInboundEvent(SQLModel, table=True): binding_id: str = Field(index=True) channel: str = Field(index=True) event_id: str - payload_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(JSON)) + payload_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(PortableJSON)) # 入站时的绑定配置代次，仅用于 ingress 代际审计；已落库事件不因后续轮换失效 config_revision: int = Field( default=0, @@ -681,7 +745,7 @@ class ChannelInboundEvent(SQLModel, table=True): # 每条事件不可变的回复目标；异步处理不得读取会话上的可变 target target_json: dict[str, Any] = Field( default_factory=dict, - sa_column=Column(JSON, nullable=False, server_default="{}"), + sa_column=Column(PortableJSON, nullable=False, server_default="{}"), ) # 飞书收到确认 reaction 的远端 ID；最终回复送达后据此异步删除 reaction_id: Optional[str] = Field(default=None, index=True) @@ -689,7 +753,7 @@ class ChannelInboundEvent(SQLModel, table=True): status: str = Field(default="received", index=True) # 创建/接管该事件的进程启动代次；当前代次仍在运行时禁止按墙钟误接管。 processor_run_id: Optional[str] = Field(default=None, index=True) - error: Optional[str] = None + error: Optional[str] = Field(default=None, sa_column=Column(Text)) processed_at: Optional[datetime] = None created_at: datetime = Field(default_factory=utc_now) updated_at: datetime = Field(default_factory=utc_now) @@ -704,17 +768,17 @@ class ChannelDelivery(SQLModel, table=True): session_id: str = Field(index=True) message_id: Optional[str] = Field(default=None, index=True) # 投递目标：to_user_id + context_token - target_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(JSON)) + target_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(PortableJSON)) # reply/error_notice kind: str = Field(default="reply", index=True) - text: str + text: str = Field(sa_column=Column(Text)) # pending/sending/delivered/failed status: str = Field(default="pending", index=True) attempts: int = 0 next_attempt_at: Optional[datetime] = Field(default=None, index=True) # 原子 claim 的抢占时间(守护据此重置卡死投递) sending_since: Optional[datetime] = None - last_error: Optional[str] = None + last_error: Optional[str] = Field(default=None, sa_column=Column(Text)) # 回复类投递 = message_id，天然幂等 idempotency_key: str = Field(unique=True, index=True) # 第一次真正尝试远端发送的时间，用于飞书 UUID 一小时去重窗口 @@ -735,12 +799,12 @@ class HumanHandoffRequest(SQLModel, table=True): assignee_user_id: Optional[str] = Field(default=None, index=True) trigger_skill_id: Optional[str] = Field(default=None, index=True) trigger_step_id: Optional[str] = Field(default=None, index=True) - context_summary: Optional[str] = None - pending_question: Optional[str] = None + context_summary: Optional[str] = Field(default=None, sa_column=Column(Text)) + pending_question: Optional[str] = Field(default=None, sa_column=Column(Text)) status: str = Field(default="pending", index=True) - human_reply: Optional[str] = None - resume_payload_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(JSON)) - metadata_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(JSON)) + human_reply: Optional[str] = Field(default=None, sa_column=Column(Text)) + resume_payload_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(PortableJSON)) + metadata_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(PortableJSON)) created_at: datetime = Field(default_factory=utc_now) updated_at: datetime = Field(default_factory=utc_now) answered_at: Optional[datetime] = None @@ -753,11 +817,11 @@ class ScheduledTask(SQLModel, table=True): tenant_id: str = Field(index=True) agent_id: str = Field(index=True) created_by_user_id: str = Field(index=True) - title: str - prompt: str - description: Optional[str] = None + title: str = Field(sa_column=Column(Text)) + prompt: str = Field(sa_column=Column(Text)) + description: Optional[str] = Field(default=None, sa_column=Column(Text)) schedule_type: str = Field(default="daily", index=True) - schedule_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(JSON)) + schedule_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(PortableJSON)) timezone: str = Field(default="Asia/Shanghai", index=True) rrule: Optional[str] = None status: str = Field(default="active", index=True) @@ -772,7 +836,7 @@ class ScheduledTask(SQLModel, table=True): lease_owner: Optional[str] = Field(default=None, index=True) lease_until: Optional[datetime] = Field(default=None, index=True) source_session_id: Optional[str] = Field(default=None, index=True) - metadata_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(JSON)) + metadata_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(PortableJSON)) created_at: datetime = Field(default_factory=utc_now) updated_at: datetime = Field(default_factory=utc_now) @@ -793,9 +857,9 @@ class ScheduledTaskRun(SQLModel, table=True): status: str = Field(default="queued", index=True) started_at: Optional[datetime] = Field(default=None, index=True) finished_at: Optional[datetime] = Field(default=None, index=True) - result_summary: Optional[str] = None - error: Optional[str] = None - trace_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(JSON)) + result_summary: Optional[str] = Field(default=None, sa_column=Column(Text)) + error: Optional[str] = Field(default=None, sa_column=Column(Text)) + trace_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(PortableJSON)) created_at: datetime = Field(default_factory=utc_now) updated_at: datetime = Field(default_factory=utc_now) @@ -807,8 +871,8 @@ class Message(SQLModel, table=True): tenant_id: str = Field(index=True) session_id: str = Field(index=True) role: str - content: str - metadata_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(JSON)) + content: str = Field(sa_column=Column(Text)) + metadata_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(PortableJSON)) created_at: datetime = Field(default_factory=utc_now) @@ -824,10 +888,10 @@ class MessageFeedback(SQLModel, table=True): rating: str = Field(index=True) analysis_status: str = Field(default="pending", index=True) analysis_bucket: Optional[str] = Field(default=None, index=True) - analysis_reason: Optional[str] = None - analysis_summary: Optional[str] = None + analysis_reason: Optional[str] = Field(default=None, sa_column=Column(Text)) + analysis_summary: Optional[str] = Field(default=None, sa_column=Column(Text)) analysis_confidence: Optional[float] = None - analysis_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(JSON)) + analysis_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(PortableJSON)) analyzed_at: Optional[datetime] = None created_at: datetime = Field(default_factory=utc_now) updated_at: datetime = Field(default_factory=utc_now) @@ -857,7 +921,7 @@ class AgentEvent(SQLModel, table=True): tenant_id: str = Field(index=True) session_id: str = Field(index=True) event_type: str = Field(index=True) - payload_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(JSON)) + payload_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(PortableJSON)) created_at: datetime = Field(default_factory=utc_now) @@ -870,8 +934,8 @@ class MemoryRecord(SQLModel, table=True): username: Optional[str] = Field(default=None, index=True) session_id: Optional[str] = Field(default=None, index=True) kind: str = Field(default="conversation", index=True) - content: str + content: str = Field(sa_column=Column(Text)) importance: float = 0.5 - metadata_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(JSON)) + metadata_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(PortableJSON)) created_at: datetime = Field(default_factory=utc_now) updated_at: datetime = Field(default_factory=utc_now) diff --git a/backend/feishu_connector_worker.py b/backend/feishu_connector_worker.py index e7d3200..bd01e1f 100644 --- a/backend/feishu_connector_worker.py +++ b/backend/feishu_connector_worker.py @@ -39,18 +39,20 @@ def _user_data_dir() -> Path: return Path.home() / ".local" / "share" / "StaffDeck" -def binding_lock_path(binding_id: str, database_path: Path) -> Path: - database = database_path.expanduser().resolve() - database_fingerprint = hashlib.sha256(str(database).encode("utf-8")).hexdigest()[:16] - binding_fingerprint = hashlib.sha256(binding_id.encode("utf-8")).hexdigest()[:16] - return ( - database.parent - / "connector-locks" - / f"feishu-{database_fingerprint}-{binding_fingerprint}.lock" - ) +def binding_lock_key(binding_id: str) -> str: + """binding 级进程锁的统一 key:方言 advisory lock,锁身份 = 数据库作用域 + key。 + + 同一数据库内同一 binding 互斥;不再取 database_url 指纹(指纹随密码轮换 + 漂移,且跨 HOME/容器互不可见)。 + """ + return f"staffdeck-binding-{binding_id}" class BindingProcessLock: + """用户数据目录文件锁:仅作 spec.database_url 为空(单测直构造 spec)时的兜底。 + + 生产路径一律走 _DialectBindingLock(方言 advisory lock,数据库作用域)。""" + def __init__(self, path: Path): self.path = path self._handle = None @@ -97,14 +99,63 @@ def release(self) -> None: self._handle = None +class _DialectBindingLock: + """binding 级方言 advisory lock:engine 与持锁 session 常驻至 release()。 + + PG 锁随连接存活(专属 AUTOCOMMIT 连接由方言内部常驻,进程退出即释放); + SQLite 为数据库文件旁的文件锁,句柄由方言实现管理。engine 用 NullPool: + 持锁期间不需要连接池复用。engine/session 仅作 bind 来源常驻至 release()。 + """ + + def __init__(self, database_url: str, key: str) -> None: + self._database_url = database_url + self._key = key + self._dialect = None + self._session = None + self._engine = None + + def acquire(self) -> bool: + from sqlalchemy.pool import NullPool + from sqlmodel import Session, create_engine + + from app.db.dialect import get_dialect + + self._engine = create_engine(self._database_url, poolclass=NullPool) + self._session = Session(self._engine) + self._dialect = get_dialect(self._engine.url.get_backend_name()) + if self._dialect.acquire_advisory_lock(self._session, self._key): + return True + self._session.close() + self._session = None + self._engine.dispose() + self._engine = None + return False + + def release(self) -> None: + if self._session is not None: + self._dialect.release_advisory_lock(self._session, self._key) + self._session.close() + self._session = None + if self._engine is not None: + self._engine.dispose() + self._engine = None + + +def _fallback_binding_lock(binding_id: str) -> BindingProcessLock: + """无 database_url 时的兜底文件锁(仅单测直构造 spec 场景)。""" + digest = hashlib.sha256(binding_id.encode("utf-8")).hexdigest()[:16] + path = _user_data_dir() / "connector-locks" / f"feishu-binding-{digest}.lock" + return BindingProcessLock(path) + + @dataclass(frozen=True) class ConnectorChildSpec: binding_id: str config_revision: int child_nonce: str runtime_path: str - binding_lock_path: str - database_path: str = "" + # 完整 SQLAlchemy URL:子进程按它自建引擎(非 SQLite 不传 check_same_thread) + database_url: str = "" watchdog_seconds: float = 2.5 @@ -268,7 +319,13 @@ def _load_runtime(path: str) -> Callable[[ConnectorChildSpec, ChildControl, Fram def connector_child_entry(spec: ConnectorChildSpec, connection) -> None: - lock = BindingProcessLock(Path(spec.binding_lock_path)) + # 生产路径:方言 advisory lock(数据库作用域,key 与父进程探测一致); + # spec.database_url 为空(单测直构造 spec)时回退用户数据目录文件锁。 + lock: BindingProcessLock | _DialectBindingLock + if spec.database_url: + lock = _DialectBindingLock(spec.database_url, binding_lock_key(spec.binding_id)) + else: + lock = _fallback_binding_lock(spec.binding_id) if not lock.acquire(): try: connection.send( diff --git a/backend/pyproject.toml b/backend/pyproject.toml index 7b8fb1f..c683ba9 100644 --- a/backend/pyproject.toml +++ b/backend/pyproject.toml @@ -35,6 +35,10 @@ dev = [ "pytest-asyncio>=0.23.0", "ruff>=0.5.0" ] +# PostgreSQL/openGauss 驱动按需安装(默认 SQLite 的桌面端不背 ~20MB 二进制) +postgres = [ + "psycopg[binary]>=3.1" +] packaging = [ "pyinstaller>=6.6.0", "certifi>=2024.2.2" diff --git a/backend/tests/test_channel_connector_lock.py b/backend/tests/test_channel_connector_lock.py new file mode 100644 index 0000000..987ffd0 --- /dev/null +++ b/backend/tests/test_channel_connector_lock.py @@ -0,0 +1,184 @@ +"""connector 单实例锁:方言分发、SQLite 文件锁行为、PG 持锁会话常驻、占用冲突。""" + +import os +from types import SimpleNamespace + +from sqlalchemy.pool import StaticPool +from sqlmodel import Session, create_engine + +import app.channels as channels +from app.db.dialect import SQLiteDialect + + +def _sqlite_engine(tmp_path, name: str = "connector.db"): + return create_engine( + f"sqlite:///{tmp_path / name}", + connect_args={"check_same_thread": False}, + poolclass=StaticPool, + ) + + +def _reset_connector_lock_state() -> None: + channels._connector_lock_pid = None + channels._connector_lock_session = None + + +class _FakeSessionScopedDialect: + """PG 形态的假方言:记录 acquire/release,会话级锁标志位为真。""" + + session_scoped_advisory_lock = True + + def __init__(self, *, acquired: bool = True) -> None: + self.acquired = acquired + self.lock_held = acquired + self.calls: list[tuple[str, str]] = [] + + def acquire_advisory_lock(self, session, key: str) -> bool: + self.calls.append(("acquire", key)) + return self.acquired + + def release_advisory_lock(self, session, key: str) -> None: + self.calls.append(("release", key)) + + def check_advisory_lock(self, session, key: str) -> bool: + self.calls.append(("check", key)) + return self.lock_held + + +def test_connector_lock_sqlite_file_roundtrip(tmp_path, monkeypatch) -> None: + engine = _sqlite_engine(tmp_path) + monkeypatch.setattr("app.db.engine", engine) + _reset_connector_lock_state() + + assert channels._acquire_connector_process_lock() is True + # 重入成功 + assert channels._acquire_connector_process_lock() is True + # 数据目录文件锁已落盘(行为不变,锁文件名带统一 key) + assert (tmp_path / "connector.db.staffdeck-connector.lock").exists() + channels._release_connector_process_lock() + assert channels._connector_lock_pid is None + # 释放后可再获取 + assert channels._acquire_connector_process_lock() is True + channels._release_connector_process_lock() + _reset_connector_lock_state() + + +def test_connector_lock_sqlite_contention_returns_false(tmp_path, monkeypatch) -> None: + engine = _sqlite_engine(tmp_path) + monkeypatch.setattr("app.db.engine", engine) + _reset_connector_lock_state() + + holder = SQLiteDialect() + with Session(engine) as session: + # 另一"进程"(独立方言实例)已持有同一文件锁 + assert holder.acquire_advisory_lock(session, channels._CONNECTOR_LOCK_KEY) is True + assert channels._acquire_connector_process_lock() is False + holder.release_advisory_lock(session, channels._CONNECTOR_LOCK_KEY) + # 释放后可获取 + assert channels._acquire_connector_process_lock() is True + channels._release_connector_process_lock() + _reset_connector_lock_state() + + +def test_connector_lock_pg_resident_session_roundtrip(monkeypatch) -> None: + engine = create_engine("postgresql+psycopg://user:secret@db.internal/staffdeck") + monkeypatch.setattr("app.db.engine", engine) + dialect = _FakeSessionScopedDialect() + monkeypatch.setattr("app.db.dialect.get_dialect", lambda _name: dialect) + _reset_connector_lock_state() + + assert channels._acquire_connector_process_lock() is True + assert dialect.calls == [("acquire", channels._CONNECTOR_LOCK_KEY)] + # PG advisory lock 随连接存活:持锁会话常驻模块级 + assert channels._connector_lock_session is not None + channels._release_connector_process_lock() + assert ("release", channels._CONNECTOR_LOCK_KEY) in dialect.calls + assert channels._connector_lock_session is None + _reset_connector_lock_state() + + +def test_connector_lock_pg_contention_closes_session(monkeypatch) -> None: + engine = create_engine("postgresql+psycopg://user:secret@db.internal/staffdeck") + monkeypatch.setattr("app.db.engine", engine) + dialect = _FakeSessionScopedDialect(acquired=False) + monkeypatch.setattr("app.db.dialect.get_dialect", lambda _name: dialect) + _reset_connector_lock_state() + + assert channels._acquire_connector_process_lock() is False + assert channels._connector_lock_session is None + assert channels._connector_lock_pid is None + _reset_connector_lock_state() + + +# ---------- 会话级锁存活校验与断连降级 ---------- + + +def test_connector_lock_healthy_without_session_is_true() -> None: + """文件锁路径无常驻会话:无静默失效模式,健康检查恒 True。""" + _reset_connector_lock_state() + assert channels._connector_lock_healthy() is True + + +def test_connector_lock_healthy_delegates_to_dialect(monkeypatch) -> None: + engine = create_engine("postgresql+psycopg://user:secret@db.internal/staffdeck") + monkeypatch.setattr("app.db.engine", engine) + dialect = _FakeSessionScopedDialect() + monkeypatch.setattr("app.db.dialect.get_dialect", lambda _name: dialect) + _reset_connector_lock_state() + + channels._connector_lock_session = object() + dialect.lock_held = True + assert channels._connector_lock_healthy() is True + dialect.lock_held = False + assert channels._connector_lock_healthy() is False + _reset_connector_lock_state() + + +def test_connector_lock_watchdog_degrades_on_lock_loss(monkeypatch) -> None: + """锁静默失效:看门狗调用 stop_channel_services 主动降级后退出。""" + _reset_connector_lock_state() + monkeypatch.setattr(channels, "_CONNECTOR_LOCK_CHECK_SECONDS", 0) + channels._connector_lock_pid = os.getpid() + channels._connector_lock_session = object() + monkeypatch.setattr(channels, "_connector_lock_healthy", lambda: False) + calls: list[str] = [] + monkeypatch.setattr( + channels, "stop_channel_services", lambda: calls.append("stop") or True + ) + + channels._connector_lock_watchdog() + assert calls == ["stop"] + _reset_connector_lock_state() + + +def test_connector_lock_watchdog_exits_after_normal_release(monkeypatch) -> None: + """正常释放锁后看门狗自行退出,不触发降级。""" + _reset_connector_lock_state() + monkeypatch.setattr(channels, "_CONNECTOR_LOCK_CHECK_SECONDS", 0) + # pid=None:锁未持有,看门狗第一轮即返回 + calls: list[str] = [] + monkeypatch.setattr( + channels, "stop_channel_services", lambda: calls.append("stop") or True + ) + channels._connector_lock_watchdog() + assert calls == [] + + +def test_release_tolerates_dead_lock_session(monkeypatch) -> None: + """释放路径容错:会话已随断连死亡(release 抛错)时状态仍被清理,不再抛出。""" + engine = create_engine("postgresql+psycopg://user:secret@db.internal/staffdeck") + monkeypatch.setattr("app.db.engine", engine) + + class _DeadReleaseDialect(_FakeSessionScopedDialect): + def release_advisory_lock(self, session, key: str) -> None: + raise RuntimeError("connection is closed") + + dialect = _DeadReleaseDialect() + monkeypatch.setattr("app.db.dialect.get_dialect", lambda _name: dialect) + _reset_connector_lock_state() + channels._connector_lock_pid = os.getpid() + channels._connector_lock_session = SimpleNamespace(close=lambda: None) + + channels._release_connector_process_lock() # 不应抛出 + assert channels._connector_lock_pid is None + assert channels._connector_lock_session is None diff --git a/backend/tests/test_channel_session.py b/backend/tests/test_channel_session.py index d659971..6f42199 100644 --- a/backend/tests/test_channel_session.py +++ b/backend/tests/test_channel_session.py @@ -1,27 +1,20 @@ import pytest from sqlalchemy.exc import IntegrityError from sqlalchemy.pool import StaticPool -from sqlmodel import Session, SQLModel, create_engine, text +from sqlmodel import Session, SQLModel, create_engine from app.channels.service_session import find_or_create_channel_session from app.db.models import ChannelBinding, ChatSession, Tenant, User -def _test_engine(with_unique_index: bool = False): +def _test_engine(): + # 唯一索引 uq_sessions_agent_channel_extconv 已由 models.py 并入 create_all engine = create_engine( "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool, ) SQLModel.metadata.create_all(engine) - if with_unique_index: - with engine.begin() as conn: - conn.execute( - text( - "CREATE UNIQUE INDEX uq_sessions_agent_channel_extconv " - "ON sessions(agent_id, channel, channel_binding_id, external_conv_id)" - ) - ) return engine @@ -62,7 +55,7 @@ def test_create_then_reuse_channel_session() -> None: def test_user_mismatch_archives_legacy_session_and_starts_clean() -> None: - engine = _test_engine(with_unique_index=True) + engine = _test_engine() with Session(engine) as db: binding, old_user = _seed(db) new_user = User( @@ -136,7 +129,7 @@ def test_channel_target_json_roundtrip() -> None: def test_unique_index_rejects_duplicate_anchor() -> None: - engine = _test_engine(with_unique_index=True) + engine = _test_engine() with Session(engine) as db: binding, user = _seed(db) first = find_or_create_channel_session(db, binding, user, "agent_1", "wechat_p2p_wxid_1", "hi") @@ -160,7 +153,7 @@ def test_unique_index_rejects_duplicate_anchor() -> None: def test_web_sessions_coexist_under_unique_index() -> None: - engine = _test_engine(with_unique_index=True) + engine = _test_engine() with Session(engine) as db: _seed(db) # channel 为 NULL 的 web 会话在唯一索引下互不相等,可重复存在 diff --git a/backend/tests/test_channel_wechat.py b/backend/tests/test_channel_wechat.py index 32716ef..1dbbfaa 100644 --- a/backend/tests/test_channel_wechat.py +++ b/backend/tests/test_channel_wechat.py @@ -970,3 +970,123 @@ def handler(request: httpx.Request) -> httpx.Response: random_ids = [payload["msg"]["client_id"] for payload in sent] assert len(set(random_ids)) == 3 assert all(not cid.startswith("staffdeck:msg_1") for cid in random_ids) +# ---------- _patch_runtime_config 读-改-写 + revision CAS ---------- + + +def test_patch_runtime_config_revision_cas_and_expected_values() -> None: + from app.channels.adapters.wechat import _patch_runtime_config + + engine = _test_engine() + with Session(engine) as db: + db.add(Tenant(id="tenant_demo", name="Demo")) + binding = ChannelBinding( + tenant_id="tenant_demo", + agent_id="agent_1", + channel="wechat", + status="active", + config_json={"get_updates_buf": "old", "typing_ticket": "t1"}, + ) + db.add(binding) + db.commit() + db.refresh(binding) + binding_id = binding.id + revision = binding.config_revision + + # stale expected_revision:CAS 拒绝且不落库 + assert ( + _patch_runtime_config( + engine, + binding_id, + set_values={"get_updates_buf": "new"}, + expected_revision=revision + 1, + ) + is False + ) + # JSON 键 CAS 不符:拒绝 + assert ( + _patch_runtime_config( + engine, + binding_id, + remove_keys=("typing_ticket",), + expected_values={"typing_ticket": "other"}, + ) + is False + ) + with Session(engine) as db: + row = db.get(ChannelBinding, binding_id) + assert row.config_json["get_updates_buf"] == "old" + assert row.config_json["typing_ticket"] == "t1" + + # revision 与 expected_values 均匹配:set/remove/列值一并原子落库 + assert ( + _patch_runtime_config( + engine, + binding_id, + set_values={"get_updates_buf": "new"}, + remove_keys=("typing_ticket",), + expected_revision=revision, + expected_values={"typing_ticket": "t1"}, + binding_values={"connected": True}, + ) + is True + ) + with Session(engine) as db: + row = db.get(ChannelBinding, binding_id) + assert row.config_json == {"get_updates_buf": "new"} + assert row.connected is True + + +def test_patch_runtime_config_require_active_and_field_whitelist() -> None: + from app.channels.adapters.wechat import _patch_runtime_config + + engine = _test_engine() + with Session(engine) as db: + db.add(Tenant(id="tenant_demo", name="Demo")) + binding = ChannelBinding( + tenant_id="tenant_demo", + agent_id="agent_1", + channel="wechat", + status="expired", + config_json={}, + ) + db.add(binding) + db.commit() + db.refresh(binding) + binding_id = binding.id + + # 非 active:require_active 拒绝 + assert ( + _patch_runtime_config( + engine, binding_id, set_values={"get_updates_buf": "x"}, require_active=True + ) + is False + ) + # 白名单外列:拒绝 + try: + _patch_runtime_config(engine, binding_id, binding_values={"tenant_id": "evil"}) + except ValueError as error: + assert "unsupported binding runtime field" in str(error) + else: + raise AssertionError("白名单外字段必须拒绝") + # 不存在的 binding:False + assert _patch_runtime_config(engine, "chan_missing", set_values={"a": 1}) is False + + +def test_runtime_patch_lock_dropped_after_thread_stopped() -> None: + """补丁锁清理:wait_binding_stopped 确认线程退出后删除锁字典项,防无上限累积。""" + from app.channels.adapters.wechat import ( + WeChatPollManager, + _runtime_patch_lock, + _runtime_patch_locks, + ) + + engine = _test_engine() + first = _runtime_patch_lock("b1") + assert _runtime_patch_lock("b1") is first + + manager = WeChatPollManager(db_engine=engine, client_factory=lambda binding: None) + # 无线程存活:wait_binding_stopped 直接 True 并清理该 binding 的补丁锁 + assert manager.wait_binding_stopped("b1", 0) is True + assert "b1" not in _runtime_patch_locks + assert _runtime_patch_lock("b1") is not first + _runtime_patch_locks.pop("b1", None) diff --git a/backend/tests/test_db_dialect.py b/backend/tests/test_db_dialect.py new file mode 100644 index 0000000..1957d45 --- /dev/null +++ b/backend/tests/test_db_dialect.py @@ -0,0 +1,504 @@ +"""方言提供者单测:注册表、engine_kwargs、日分桶表达式形态、JSON 读改写、锁。""" + +import os +from types import SimpleNamespace + +from sqlalchemy import Column, DateTime +from sqlalchemy.dialects import postgresql, sqlite +from sqlalchemy.pool import StaticPool +from sqlmodel import Session, SQLModel, create_engine, select +from sqlalchemy import text as sa_text + +from app.db.dialect import ( + GenericDialect, + PostgresDialect, + SQLiteDialect, + get_dialect, + register_dialect, +) +import app.db.dialect as dialect_module + +import pytest + +_COLUMN = Column("created_at", DateTime) + + +@pytest.fixture(autouse=True) +def _restore_dialect_registry(): + """用例可能注册新方言或缓存 Generic 实例:每个用例后还原注册表,防跨测试污染。""" + snapshot = dict(dialect_module._DIALECTS) + yield + dialect_module._DIALECTS.clear() + dialect_module._DIALECTS.update(snapshot) + + +def test_registry_returns_expected_dialects() -> None: + assert get_dialect("sqlite").name == "sqlite" + assert get_dialect("sqlite").supports_partial_index is True + assert get_dialect("postgresql").name == "postgresql" + assert get_dialect("postgresql").supports_partial_index is True + # 未注册后端回退 GenericDialect(名按 backend 名,能力保守) + fallback = get_dialect("mysql") + assert isinstance(fallback, GenericDialect) + assert fallback.name == "mysql" + assert fallback.supports_partial_index is False + + +def test_register_dialect_override() -> None: + custom = GenericDialect("dm") + register_dialect("dm", custom) + assert get_dialect("dm") is custom + + +def test_engine_kwargs_sqlite_matches_legacy_behavior() -> None: + kwargs = get_dialect("sqlite").engine_kwargs("sqlite:///x.db") + assert kwargs == {"connect_args": {"check_same_thread": False, "timeout": 30}} + # PG 连接池加固:pre_ping 防陈旧连接、recycle 防 idle 掐断、池容随 binding 增长 + pg_kwargs = get_dialect("postgresql").engine_kwargs("postgresql+psycopg://h/db") + assert pg_kwargs["pool_pre_ping"] is True + + +def test_day_bucket_sqlite_keeps_localtime_date() -> None: + compiled = str( + SQLiteDialect() + .day_bucket(_COLUMN) + .compile(dialect=sqlite.dialect(), compile_kwargs={"literal_binds": True}) + ) + assert "date(" in compiled and "localtime" in compiled + + +def test_day_bucket_postgres_casts_with_configured_timezone(monkeypatch) -> None: + monkeypatch.setattr( + "app.db.dialect.get_settings", lambda: SimpleNamespace(app_timezone="Asia/Shanghai") + ) + compiled = str( + PostgresDialect() + .day_bucket(_COLUMN) + .compile(dialect=postgresql.dialect(), compile_kwargs={"literal_binds": True}) + ) + assert "timezone('Asia/Shanghai'" in compiled + assert compiled.startswith("CAST(") and compiled.endswith("AS DATE)") + + +def test_day_bucket_postgres_falls_back_to_local_offset(monkeypatch) -> None: + monkeypatch.setattr( + "app.db.dialect.get_settings", lambda: SimpleNamespace(app_timezone="") + ) + compiled = str( + PostgresDialect() + .day_bucket(_COLUMN) + .compile(dialect=postgresql.dialect(), compile_kwargs={"literal_binds": True}) + ) + assert "make_interval" in compiled + assert compiled.startswith("CAST(") and compiled.endswith("AS DATE)") + + +def test_day_bucket_generic_is_standard_cast() -> None: + compiled = str( + GenericDialect("dm") + .day_bucket(_COLUMN) + .compile(dialect=sqlite.dialect(), compile_kwargs={"literal_binds": True}) + ) + assert compiled.startswith("CAST(") and " AS DATE)" in compiled + + +def test_json_config_helpers_read_modify_write() -> None: + dialect = SQLiteDialect() + assert dialect.json_config_get(None, "k") is None + config = dialect.json_config_set({"a": 1}, "b", {"x": True}) + assert config == {"a": 1, "b": {"x": True}} + # 不改动入参原 dict + assert dialect.json_config_get(config, "a") == 1 + removed = dialect.json_config_remove(config, "a") + assert removed == {"b": {"x": True}} + assert dialect.json_config_remove(config, "missing") == config + + +def test_sqlite_file_advisory_lock_excludes_other_holder(tmp_path) -> None: + engine = create_engine( + f"sqlite:///{tmp_path / 'lock.db'}", + connect_args={"check_same_thread": False}, + poolclass=StaticPool, + ) + first, second = SQLiteDialect(), SQLiteDialect() + with Session(engine) as session: + assert first.acquire_advisory_lock(session, "connector") is True + # 同一持有者重入成功 + assert first.acquire_advisory_lock(session, "connector") is True + # 另一持有者抢同一把文件锁失败 + assert second.acquire_advisory_lock(session, "connector") is False + first.release_advisory_lock(session, "connector") + # 释放后可被他人获取 + assert second.acquire_advisory_lock(session, "connector") is True + second.release_advisory_lock(session, "connector") + + +class _StubConn: + """PG 专属连接桩:记录 SQL、AUTOCOMMIT 设置与关闭状态。""" + + def __init__(self, scalar=1, *, first_row=(1,)) -> None: + self._scalar = scalar + self._first_row = first_row + self.calls: list[tuple[str, dict]] = [] + self.isolation: str | None = None + self.closed = False + + def execution_options(self, **kwargs): + self.isolation = kwargs.get("isolation_level") + return self + + def execute(self, statement, params=None): + self.calls.append((str(statement), params)) + return SimpleNamespace( + scalar=lambda: self._scalar, + first=lambda: self._first_row, + ) + + def close(self) -> None: + self.closed = True + + +class _StubEngine: + """connect() 按需产出新 _StubConn 的引擎桩。""" + + def __init__(self) -> None: + self.conns: list[_StubConn] = [] + self.scalar = 1 + self.first_row = (1,) + + def connect(self): + conn = _StubConn(self.scalar, first_row=self.first_row) + self.conns.append(conn) + return conn + + +class _StubSession: + """最小会话桩:仅作 bind 来源(锁实现不得 commit/rollback 它)。""" + + def __init__(self, engine: _StubEngine) -> None: + self._engine = engine + + def get_bind(self): + return self._engine + + +def test_postgres_advisory_lock_uses_dedicated_autocommit_connection() -> None: + """acquire/check/release 全程使用同一条专属 AUTOCOMMIT Connection,不归池。""" + dialect = PostgresDialect() + engine = _StubEngine() + session = _StubSession(engine) + + assert dialect.acquire_advisory_lock(session, "connector") is True + assert len(engine.conns) == 1 + conn = engine.conns[0] + assert conn.isolation == "AUTOCOMMIT" + assert "pg_try_advisory_lock(hashtext(:key), 0)" in conn.calls[0][0] + assert conn.closed is False + + # check 在同一连接上查 pg_locks(pid 稳定),不再从池里取新连接 + assert dialect.check_advisory_lock(session, "connector") is True + assert len(engine.conns) == 1 + sql = conn.calls[1][0] + assert "pg_locks" in sql and "locktype = 'advisory'" in sql + assert "pg_backend_pid()" in sql + + dialect.release_advisory_lock(session, "connector") + assert "pg_advisory_unlock(hashtext(:key), 0)" in conn.calls[2][0] + assert conn.closed is True + + # 释放后再取:新连接 + assert dialect.acquire_advisory_lock(session, "connector") is True + assert len(engine.conns) == 2 + dialect.release_advisory_lock(session, "connector") + + +def test_postgres_advisory_lock_contention_closes_connection() -> None: + """抢锁失败:专属连接立即关闭,不驻留;check 判 False。""" + dialect = PostgresDialect() + engine = _StubEngine() + engine.scalar = 0 + session = _StubSession(engine) + + assert dialect.acquire_advisory_lock(session, "connector") is False + assert engine.conns[0].closed is True + assert dialect.check_advisory_lock(session, "connector") is False + + +def test_postgres_advisory_lock_reentrant_and_fork_guard(monkeypatch) -> None: + """同进程重入不新建连接;fork 子进程视角下继承连接不作数,按新身份真实抢锁。""" + parent = PostgresDialect() + engine = _StubEngine() + session = _StubSession(engine) + assert parent.acquire_advisory_lock(session, "k") is True + assert parent.acquire_advisory_lock(session, "k") is True + assert len(engine.conns) == 1 + + # 模拟另一"进程"(独立方言实例 + 另一进程号):继承状态不作数 + child = PostgresDialect() + child._pg_lock_conns = dict(parent._pg_lock_conns) # fork 继承的是内存副本 + real_pid = os.getpid() + monkeypatch.setattr(os, "getpid", lambda: real_pid + 100000) + assert child.check_advisory_lock(session, "k") is False + assert child.acquire_advisory_lock(session, "k") is True + assert len(engine.conns) == 2 + child.release_advisory_lock(session, "k") + monkeypatch.undo() + # 父进程锁不受子进程动作影响 + assert parent.check_advisory_lock(session, "k") is True + parent.release_advisory_lock(session, "k") + + +def test_postgres_engine_kwargs_pool_hardening() -> None: + kwargs = PostgresDialect().engine_kwargs("postgresql+psycopg://u:p@h/db") + assert kwargs["pool_pre_ping"] is True + assert kwargs["pool_recycle"] > 0 + assert kwargs["pool_size"] >= 10 + + +def test_generic_file_lock_refuses_non_sqlite_backend() -> None: + """MySQL/达梦等:url.database 是库名而非文件路径,文件锁必须响亮拒绝。""" + from sqlalchemy.engine import make_url + + dialect = GenericDialect("mysql") + session = SimpleNamespace( + get_bind=lambda: SimpleNamespace(url=make_url("mysql+pymysql://u:p@db.internal/staffdeck")) + ) + assert dialect.acquire_advisory_lock(session, "staffdeck-connector") is False + + +def test_base_check_advisory_lock_tracks_file_handle(tmp_path) -> None: + """文件锁无静默失效:check 只核实本进程仍持有打开的句柄。""" + engine = create_engine( + f"sqlite:///{tmp_path / 'check.db'}", + connect_args={"check_same_thread": False}, + poolclass=StaticPool, + ) + dialect = SQLiteDialect() + with Session(engine) as session: + assert dialect.check_advisory_lock(session, "k") is False + assert dialect.acquire_advisory_lock(session, "k") is True + assert dialect.check_advisory_lock(session, "k") is True + dialect.release_advisory_lock(session, "k") + assert dialect.check_advisory_lock(session, "k") is False + + +def test_default_model_partial_index_not_in_table_metadata() -> None: + """部分唯一索引不进 metadata:其它方言 create_all 不会静默退化为全量唯一索引。""" + import app.db.models as models # noqa: F401 + + index_names = {index.name for index in models.ModelConfig.__table__.indexes} + assert "uq_model_configs_tenant_default" not in index_names + + +def test_for_update_compile_dialect_behavior() -> None: + """FOR UPDATE:PG 真实加锁,SQLite 省略(编译期即验证,不依赖真实连接)。""" + from sqlalchemy.dialects import postgresql as pg_dialect + from sqlalchemy.dialects import sqlite as sqlite_dialect + + from app.db.models import ChannelBinding + + stmt = select(ChannelBinding).where(ChannelBinding.id == "b1").with_for_update() + assert "FOR UPDATE" in str(stmt.compile(dialect=pg_dialect.dialect())) + assert "FOR UPDATE" not in str(stmt.compile(dialect=sqlite_dialect.dialect())) + + +def test_create_all_contains_channel_session_unique_index() -> None: + import app.db.models # noqa: F401 - 注册全部表模型 + + engine = create_engine( + "sqlite://", + connect_args={"check_same_thread": False}, + poolclass=StaticPool, + ) + SQLModel.metadata.create_all(engine) + with engine.connect() as conn: + session_indexes = { + row[1] for row in conn.execute(sa_text("PRAGMA index_list('sessions')")) + } + assert "uq_sessions_agent_channel_extconv" in session_indexes + index_sql = conn.execute( + sa_text( + "SELECT sql FROM sqlite_master " + "WHERE type='index' AND name='uq_model_configs_tenant_default'" + ) + ).scalar_one() + # 部分唯一索引:仅默认模型一行受约束 + assert "WHERE is_default = 1" in index_sql + + +def test_sqlite_file_lock_fork_child_does_not_inherit(tmp_path, monkeypatch) -> None: + """fork 防护:继承句柄不作数(重抢并登记新进程号);父进程仍持锁时真实抢锁失败。""" + engine = create_engine( + f"sqlite:///{tmp_path / 'fork.db'}", + connect_args={"check_same_thread": False}, + poolclass=StaticPool, + ) + dialect = SQLiteDialect() + sibling = SQLiteDialect() # 模拟父进程仍存活持有的同一把锁 + real_pid = os.getpid() + with Session(engine) as session: + assert dialect.acquire_advisory_lock(session, "k") is True + # 进入"子进程"视角(getpid 被 mock 成另一个进程号):继承句柄被关闭并按 + # 本进程身份重抢(同进程内等价:旧锁已随句柄关闭释放),锁登记到新进程号 + monkeypatch.setattr(os, "getpid", lambda: real_pid + 100000) + assert dialect.acquire_advisory_lock(session, "k") is True + assert dialect._lock_handles["k"][0] == real_pid + 100000 + dialect.release_advisory_lock(session, "k") + # 父进程仍持锁(sibling 句柄打开):子进程身份真实抢锁失败 + assert sibling.acquire_advisory_lock(session, "k") is True + assert dialect.acquire_advisory_lock(session, "k") is False + sibling.release_advisory_lock(session, "k") + # 父释放后可得 + assert dialect.acquire_advisory_lock(session, "k") is True + dialect.release_advisory_lock(session, "k") + + +def _seed_duplicate_default_models(engine) -> None: + from app.db.models import ModelConfig + + # sqlite 上索引已由按方言 DDL 创建,先删掉再插入(模拟无部分索引后端的数据形态) + with engine.begin() as conn: + conn.execute(sa_text("DROP INDEX uq_model_configs_tenant_default")) + with Session(engine) as db: + for name in ("m1", "m2"): + db.add( + ModelConfig( + tenant_id="t1", + name=name, + model="gpt-x", + api_key_encrypted="enc", + is_default=True, + ) + ) + db.commit() + + +def test_validate_default_model_invariant_rejects_duplicates(monkeypatch, tmp_path) -> None: + """无部分索引后端:启动校验发现同租户多条默认模型时响亮拒绝,清理后放行。""" + import app.db.database as database + import app.db.models # noqa: F401 - 注册全部表模型 + from app.db.models import ModelConfig + + engine = create_engine( + f"sqlite:///{tmp_path / 'dup.db'}", + connect_args={"check_same_thread": False}, + poolclass=StaticPool, + ) + SQLModel.metadata.create_all(engine) + _seed_duplicate_default_models(engine) + + monkeypatch.setattr(database, "engine", engine) + monkeypatch.setattr(database, "_dialect", SimpleNamespace(supports_partial_index=False)) + with pytest.raises(RuntimeError, match="多条默认模型"): + database._validate_default_model_invariant() + + # 清理到每租户一条后校验通过 + with Session(engine) as db: + row = db.exec(select(ModelConfig).where(ModelConfig.name == "m2")).one() + row.is_default = False + db.add(row) + db.commit() + database._validate_default_model_invariant() + + +def test_validate_default_model_invariant_skipped_when_partial_index_supported( + monkeypatch, tmp_path +) -> None: + """支持部分索引的后端:约束由 DB 唯一索引保证,启动校验直接跳过。""" + import app.db.database as database + import app.db.models # noqa: F401 + + engine = create_engine( + f"sqlite:///{tmp_path / 'skip.db'}", + connect_args={"check_same_thread": False}, + poolclass=StaticPool, + ) + SQLModel.metadata.create_all(engine) + _seed_duplicate_default_models(engine) # 即使存在脏数据形态也不查库 + + monkeypatch.setattr(database, "engine", engine) + monkeypatch.setattr(database, "_dialect", SimpleNamespace(supports_partial_index=True)) + database._validate_default_model_invariant() + + +# ---------- MySQL/Oracle 可移植性:全表 DDL 编译扫描 + 类型策略 ---------- + + +def test_all_tables_compile_on_mysql_and_oracle() -> None: + """42 张表在 mysql/oracle 方言下 DDL 全部可编译(JSON 列经 PortableJSON 降级)。""" + from sqlalchemy.dialects import mysql as mysql_dialect + from sqlalchemy.dialects import oracle as oracle_dialect + from sqlalchemy.schema import CreateTable + + import app.db.models # noqa: F401 - 注册全部表模型 + + tables = SQLModel.metadata.sorted_tables + assert len(tables) >= 40 + for table in tables: + assert str(CreateTable(table).compile(dialect=mysql_dialect.dialect())) + assert str(CreateTable(table).compile(dialect=oracle_dialect.dialect())) + + +def test_mysql_long_text_columns_are_not_varchar255() -> None: + """MySQL 对无长度 str 列默认 VARCHAR(255):长文本列必须显式 Text(抽验关键列)。""" + from sqlalchemy.dialects import mysql as mysql_dialect + + from app.db.models import ( + AgentProfile, + ChannelBinding, + ChannelDelivery, + KnowledgeChunk, + MemoryRecord, + Message, + ModelConfig, + PersonaConfig, + Tool, + UserAvatar, + ) + + dialect = mysql_dialect.dialect() + long_columns = [ + (Message, "content"), + (KnowledgeChunk, "content"), + (PersonaConfig, "system_prompt"), + (AgentProfile, "persona_prompt"), + (ChannelDelivery, "text"), + (ChannelBinding, "credentials_enc"), + (MemoryRecord, "content"), + (ModelConfig, "api_key_encrypted"), + (Tool, "url"), + (Tool, "description"), + # 头像 base64 ~2.8MB:MySQL 须 MEDIUMTEXT(超 TEXT 64KB 上限) + (UserAvatar, "data_url"), + ] + for table, column in long_columns: + compiled = str(table.__table__.c[column].type.compile(dialect=dialect)) + assert "TEXT" in compiled.upper(), (table.__tablename__, column, compiled) + assert "VARCHAR(255)" not in compiled.upper(), (table.__tablename__, column, compiled) + + +def test_portable_json_roundtrip_and_oracle_fallback() -> None: + """PortableJSON:pg/sqlite/mysql 走原生 JSON;oracle/dm 序列化进 CLOB 并还原。""" + from sqlalchemy.dialects import oracle as oracle_dialect + from sqlalchemy.dialects import postgresql as pg_dialect + + from app.db.models import PortableJSON + + portable = PortableJSON() + oracle = oracle_dialect.dialect() + pg = pg_dialect.dialect() + + value = {"key": ["a", 1], "中文": True} + # oracle:序列化为文本,读出还原 + bound = portable.process_bind_param(value, oracle) + assert isinstance(bound, str) + assert portable.process_result_value(bound, oracle) == value + # pg:原样透传(由原生 JSON 类型处理) + assert portable.process_bind_param(value, pg) is value + assert portable.process_result_value(value, pg) is value + # None 两方言都透传 + assert portable.process_bind_param(None, oracle) is None + assert portable.process_result_value(None, oracle) is None + # oracle 下 DDL 类型为 CLOB,pg 下为 JSON + assert "CLOB" in str(portable.compile(dialect=oracle)).upper() + assert "JSON" in str(portable.compile(dialect=pg)).upper() diff --git a/backend/tests/test_feishu_manager.py b/backend/tests/test_feishu_manager.py index 9b80a00..633685b 100644 --- a/backend/tests/test_feishu_manager.py +++ b/backend/tests/test_feishu_manager.py @@ -11,7 +11,7 @@ class FakeSupervisor: def __init__(self, **kwargs): - self.database_path = kwargs["database_path"] + self.database_url = kwargs["database_url"] self.started = [] self.stopped = [] self.states = {} @@ -302,3 +302,29 @@ def test_feishu_adapter_is_registered() -> None: channels._ensure_adapters_registered() assert get_channel_adapter("feishu").__class__.__name__ == "FeishuAdapter" + + +# ---------- 数据库 URL 透传(P2 方言化) ---------- + + +def test_manager_accepts_non_sqlite_database_url() -> None: + """非 SQLite 不再报"仅支持文件 SQLite":完整 URL(密码不遮蔽)透传给子进程。""" + engine = create_engine("postgresql+psycopg://user:secret@db.internal:5432/staffdeck") + manager = FeishuProcessManager(db_engine=engine, reconcile_seconds=60) + assert ( + manager._database_url + == "postgresql+psycopg://user:secret@db.internal:5432/staffdeck" + ) + + +def test_manager_rejects_memory_sqlite() -> None: + """内存 SQLite 子进程无法共享,仍然拒绝(与方言无关的物理限制)。""" + from sqlalchemy.pool import StaticPool + + engine = create_engine("sqlite://", poolclass=StaticPool) + try: + FeishuProcessManager(db_engine=engine) + except RuntimeError as error: + assert "内存数据库" in str(error) + else: + raise AssertionError("内存 SQLite 必须拒绝(子进程无法共享)") diff --git a/backend/tests/test_feishu_process_spike.py b/backend/tests/test_feishu_process_spike.py index bdb09bc..ec9aec6 100644 --- a/backend/tests/test_feishu_process_spike.py +++ b/backend/tests/test_feishu_process_spike.py @@ -24,7 +24,8 @@ from lark_channel.ws.pb.pbbp2_pb2 import Frame from app.channels.feishu_process import ConnectorState, FeishuProcessSupervisor -from feishu_connector_worker import BindingProcessLock, binding_lock_path +from app.db.dialect import get_dialect +from feishu_connector_worker import binding_lock_key SDK_RUNTIME = "feishu_connector_worker:run_sdk_contract_runtime" IDLE_RUNTIME = "feishu_connector_worker:run_idle_contract_runtime" @@ -32,6 +33,23 @@ UNSTOPPABLE_RUNTIME = "feishu_connector_worker:run_unstoppable_contract_runtime" +def _binding_lock_free(database: Path, binding_id: str) -> bool: + """方言探测 binding 锁是否空闲(与父进程同语义:短会话 try-acquire + release)。""" + from sqlmodel import Session, create_engine + + engine = create_engine(f"sqlite:///{database}") + try: + dialect = get_dialect(engine.url.get_backend_name()) + key = binding_lock_key(binding_id) + with Session(engine) as session: + if not dialect.acquire_advisory_lock(session, key): + return False + dialect.release_advisory_lock(session, key) + return True + finally: + engine.dispose() + + class _LocalFeishuServer: def __init__(self) -> None: self.loop = asyncio.new_event_loop() @@ -837,9 +855,7 @@ def test_child_owned_lock_is_released_after_parent_pipe_closes(tmp_path: Path, m time.sleep(0.01) record.process.join(timeout=1.0) assert not record.process.is_alive() - lock = BindingProcessLock(binding_lock_path("binding-orphan", database)) - assert lock.acquire() - lock.release() + assert _binding_lock_free(database, "binding-orphan") assert supervisor.stop(timeout=2.0) @@ -1156,10 +1172,7 @@ def test_replace_binding_stops_old_generation_before_revision_commit( commit_observations: list[tuple[int | None, bool]] = [] def commit_revision() -> bool: - lock = BindingProcessLock(Path(old.spec.binding_lock_path)) - lock_free = lock.acquire() - if lock_free: - lock.release() + lock_free = _binding_lock_free(tmp_path / "replace.db", "binding-replace") commit_observations.append((old.exit_code, lock_free)) return True @@ -1289,12 +1302,12 @@ def test_stop_does_not_mark_closed_until_binding_locks_are_free( original_lock_check = supervisor._binding_lock_is_free checks = 0 - def first_check_blocked(path: Path) -> bool: + def first_check_blocked(binding_id: str) -> bool: nonlocal checks checks += 1 if checks == 1: return False - return original_lock_check(path) + return original_lock_check(binding_id) monkeypatch.setattr(supervisor, "_binding_lock_is_free", first_check_blocked) assert not supervisor.stop(timeout=2.0) diff --git a/backend/tests/test_model_configs_api.py b/backend/tests/test_model_configs_api.py index abb39cc..e7c8585 100644 --- a/backend/tests/test_model_configs_api.py +++ b/backend/tests/test_model_configs_api.py @@ -176,7 +176,7 @@ def test_switching_default_clears_existing_row_before_setting_new(tmp_path) -> N with _db(tmp_path) as db: db.exec( text( - "CREATE UNIQUE INDEX uq_model_configs_tenant_default " + "CREATE UNIQUE INDEX IF NOT EXISTS uq_model_configs_tenant_default " "ON model_configs(tenant_id) WHERE is_default = 1" ) ) @@ -413,13 +413,7 @@ def test_concurrent_initial_verification_activates_only_one_default(tmp_path, mo connect_args={"check_same_thread": False, "timeout": 30}, ) SQLModel.metadata.create_all(engine) - with engine.begin() as conn: - conn.execute( - text( - "CREATE UNIQUE INDEX uq_model_configs_tenant_default " - "ON model_configs(tenant_id) WHERE is_default = 1" - ) - ) + # 部分唯一索引 uq_model_configs_tenant_default 已由 models.py 并入 create_all with Session(engine) as db: db.add(Tenant(id="tenant_a", name="Tenant A")) for model_id in ("model_a", "model_b"): diff --git a/scripts/smoke_postgres.sh b/scripts/smoke_postgres.sh new file mode 100755 index 0000000..0536697 --- /dev/null +++ b/scripts/smoke_postgres.sh @@ -0,0 +1,169 @@ +#!/usr/bin/env bash +# Postgres/高斯冒烟:DATABASE_URL=postgresql+psycopg://... 下验证 +# create_all、/api/health 与日分桶/知识库/渠道配置三处方言路径。 +# 用法:DATABASE_URL='postgresql+psycopg://user:pass@host:5432/dbname' scripts/smoke_postgres.sh +set -euo pipefail + +ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +PYTHON_BIN="$ROOT_DIR/backend/.venv/bin/python" +[[ -x "$PYTHON_BIN" ]] || PYTHON_BIN="${PYTHON:-python3}" + +if [[ -z "${DATABASE_URL:-}" ]]; then + echo "请先设置 DATABASE_URL=postgresql+psycopg://user:pass@host:5432/dbname" >&2 + exit 1 +fi +if [[ "$DATABASE_URL" != postgresql* ]]; then + echo "DATABASE_URL 必须是 postgresql(+psycopg) 方言: $DATABASE_URL" >&2 + exit 1 +fi + +echo "== 1/4 create_all(schema 初始化) ==" +(cd "$ROOT_DIR/backend" && DATABASE_URL="$DATABASE_URL" "$PYTHON_BIN" - <<'PY' +from app.db.database import init_db + +init_db() +print("create_all ok") +PY +) + +echo "== 2/4 日分桶/知识库/渠道配置方言路径 ==" +(cd "$ROOT_DIR/backend" && DATABASE_URL="$DATABASE_URL" "$PYTHON_BIN" - <<'PY' +"""三处生产方言点冒烟:全部走 ORM/方言助手,失败即抛异常。""" +from sqlmodel import Session, select + +from app.api.channels import _patch_binding_config_key +from app.api.knowledge import _safe_bucket_chunk_rows, _safe_document_bucket_rows +from app.db import engine +from app.db.dialect import get_dialect +from app.db.models import ( + ChannelBinding, + ChannelDelivery, + KnowledgeBucket, + KnowledgeChunk, + utc_now, +) + +backend = engine.url.get_backend_name() +dialect = get_dialect(backend) +print(f"backend={backend} dialect={dialect.name}") + +# advisory lock 闭环:acquire → check → release(真实 PG 验证会话级锁与 pg_locks 校验) +from sqlalchemy import text as sa_text + +with Session(engine) as lock_session: + assert dialect.acquire_advisory_lock(lock_session, "smoke-connector"), "acquire 失败" + assert dialect.check_advisory_lock(lock_session, "smoke-connector"), "check 未持锁" + # 回归:制造并发池流量(其他连接频繁 checkout)后,锁校验必须仍 True—— + # 专属 Connection 不归池,pg_backend_pid 稳定;Session.commit 方案在此会误判 + for _ in range(3): + with engine.connect() as busy_conn: + busy_conn.execute(sa_text("SELECT 1")) + with Session(engine) as busy_session: + busy_session.execute(sa_text("SELECT 1")) + busy_session.commit() + assert dialect.check_advisory_lock(lock_session, "smoke-connector"), "并发 checkout 后锁校验误判" + dialect.release_advisory_lock(lock_session, "smoke-connector") + assert not dialect.check_advisory_lock(lock_session, "smoke-connector"), "release 后仍持锁" +print("advisory lock roundtrip ok") + +# 部分唯一索引存在(PG 方言按方言 DDL 创建,不进 metadata) +from sqlalchemy import inspect as sa_inspect + +index_names = {i["name"] for i in sa_inspect(engine).get_indexes("model_configs")} +assert "uq_model_configs_tenant_default" in index_names, index_names +print("partial index present ok") + +with Session(engine) as db: + binding = ChannelBinding( + tenant_id="smoke_tenant", + agent_id="agent_smoke", + channel="wecom", + status="active", + config_json={"corp_id": "corpSmoke", "bot_id": "bot_smoke"}, + ) + db.add(binding) + db.commit() + db.refresh(binding) + + # 渠道配置补丁:ORM 读-改-写(原 json_set 路径) + _patch_binding_config_key(db, binding.tenant_id, binding.id, "auto_route", False) + db.commit() + db.refresh(binding) + assert binding.config_json.get("auto_route") is False, binding.config_json + assert get_dialect(backend).json_config_get(binding.config_json, "corp_id") == "corpSmoke" + + # 日分桶:方言助手表达式可执行且分组正确 + db.add( + ChannelDelivery( + tenant_id="smoke_tenant", + binding_id=binding.id, + session_id="s_smoke", + kind="reply", + text="smoke", + status="delivered", + next_attempt_at=utc_now(), + idempotency_key="smoke_day_bucket", + ) + ) + bucket = KnowledgeBucket( + tenant_id="smoke_tenant", + knowledge_base_id="kb_smoke", + knowledge_base_version_id="kbv_smoke", + document_id="doc_smoke", + bucket_key="bucket_smoke", + title="冒烟片段", + summary="smoke", + ) + db.add(bucket) + db.flush() + db.add( + KnowledgeChunk( + tenant_id="smoke_tenant", + knowledge_base_id="kb_smoke", + knowledge_base_version_id="kbv_smoke", + document_id="doc_smoke", + bucket_id=bucket.id, + chunk_index=0, + content="smoke chunk", + ) + ) + db.commit() + + day_bucket = dialect.day_bucket(ChannelDelivery.created_at) + day_rows = db.exec( + select(day_bucket) + .where(ChannelDelivery.binding_id == binding.id) + .group_by(day_bucket) + ).all() + assert len(day_rows) == 1, day_rows + + # 知识库安全读(非 SQLite 走 ORM 分支) + bucket_rows = _safe_document_bucket_rows(db, "smoke_tenant", "doc_smoke") + assert len(bucket_rows) == 1 and bucket_rows[0]["title"] == "冒烟片段" + chunk_rows = _safe_bucket_chunk_rows(db, "smoke_tenant", bucket.id) + assert len(chunk_rows) == 1 and chunk_rows[0]["content"] == "smoke chunk" + + # 清理冒烟数据 + db.delete(binding) + db.delete(bucket) + db.commit() +print("dialect paths ok") +PY +) + +echo "== 3/4 应用启动 + /api/health ==" +UVICORN_PORT="${SMOKE_PORT:-58099}" +(cd "$ROOT_DIR/backend" && DATABASE_URL="$DATABASE_URL" \ + "$PYTHON_BIN" -m uvicorn app.main:app --port "$UVICORN_PORT" >/tmp/smoke_pg_uvicorn.log 2>&1 & + echo $! > /tmp/smoke_pg_uvicorn.pid) +trap 'kill "$(cat /tmp/smoke_pg_uvicorn.pid)" 2>/dev/null || true' EXIT +for _ in $(seq 1 30); do + if curl -fsS "http://127.0.0.1:${UVICORN_PORT}/api/health" >/dev/null 2>&1; then + echo "/api/health ok" + break + fi + sleep 1 +done +curl -fsS "http://127.0.0.1:${UVICORN_PORT}/api/health" + +echo "== 4/4 冒烟完成 =="