feat: 数据库层方言可插拔(SQLite → 高斯/MySQL/达梦易扩展)

.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

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

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

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"

backend/app/api/channels.py

@@ -1,14 +1,13 @@
 from __future__ import annotations
 
-import json
 import logging
 import secrets
 import threading
 import time
 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)