Compare commits

..

28 Commits

Author SHA1 Message Date
姚章浩 0fc73c4152 ci: 接入提交规范闸门与自动编译流水线(本机Runner+微信通知)
build / build (push) Successful in 27s
2026-09-28 18:35:51 +08:00
姚章浩 18c81232ce fix: keep only setting command entries 2026-08-14 12:35:33 +08:00
姚章浩 19ed4b50df feat: add guided administrator management 2026-08-14 12:12:33 +08:00
姚章浩 9f65c3eef0 docs: update AI table binding example 2026-08-14 11:46:37 +08:00
姚章浩 50379b3b87 fix: parse AI table iframe query links 2026-08-14 11:40:28 +08:00
姚章浩 3f9ff359b9 feat: accept trusted AI table link tails 2026-08-14 11:20:15 +08:00
姚章浩 a3defdd1c0 fix: report AI table binding validation failures 2026-08-14 11:11:07 +08:00
姚章浩 b4b68d8b21 fix: accept AI table link cards in settings 2026-08-14 10:46:51 +08:00
姚章浩 365f1b8c7d fix: detect missing-product coverage responses 2026-08-14 10:02:58 +08:00
姚章浩 725db721a8 fix: refresh table dispatchers after binding activation 2026-08-13 18:39:15 +08:00
姚章浩 5d68b22636 feat: add operational component status dashboard 2026-08-13 18:33:52 +08:00
姚章浩 6f24743be2 feat: verify and activate support group binding 2026-08-13 18:22:43 +08:00
姚章浩 4dbc20ee74 fix: validate card business responses and trim help 2026-08-13 15:10:09 +08:00
姚章浩 e5be9ee780 fix: stop enqueueing disabled external table exports 2026-08-13 14:49:44 +08:00
姚章浩 0f30f8cec8 fix: route help and repeated-slash settings locally 2026-08-13 14:39:56 +08:00
姚章浩 44d9398c78 fix: keep setup available without active DingTalk app 2026-08-13 13:55:25 +08:00
姚章浩 b207ee4089 feat: switch DingTalk applications with rollback 2026-08-13 13:36:26 +08:00
姚章浩 0fc64ba77d feat: validate and activate AI table bindings 2026-08-13 13:14:06 +08:00
姚章浩 a7432aed49 feat: wire settings admin into DobotAI runtime 2026-08-13 12:59:20 +08:00
姚章浩 5036071ac2 feat: keep table binding messages out of QA 2026-08-13 12:55:00 +08:00
姚章浩 2d9490f424 feat: apply per-user answer language 2026-08-13 12:50:07 +08:00
姚章浩 462ee1f3ff feat: add admin settings control flow 2026-08-13 12:45:08 +08:00
姚章浩 8d2d4b469b feat: add secure settings foundation 2026-08-13 12:38:01 +08:00
姚章浩 9363d10e44 feat: publish outbound answer images to private OSS 2026-08-13 12:34:39 +08:00
姚章浩 74c9af9944 fix: acknowledge DingTalk callbacks before async processing 2026-08-13 12:34:16 +08:00
姚章浩 5c35a49486 fix: recreate stale WeKnora sessions once 2026-08-13 12:34:05 +08:00
姚章浩 357e9bad5c docs: plan DobotAI settings administration 2026-08-13 12:33:55 +08:00
姚章浩 2c0ed7cc67 feat: expose runtime build version 2026-08-13 10:44:16 +08:00
93 changed files with 7453 additions and 133 deletions
+12
View File
@@ -35,6 +35,18 @@ DINGTALK_FAQ_TABLE_SHEET_ID=
DINGTALK_FAQ_TABLE_VIEW_ID=
DINGTALK_FAQ_TABLE_OPERATOR_ID=
# Optional PDF-answer image egress for DingTalk cards. When any value is set,
# all five values are required. DobotAI downloads only authenticated resource://
# images from WEKNORA_BASE_URL /files, writes content-hashed copies under
# dobotai/im-images/, and emits private OSS GET URLs valid for 2 hours.
# Keep the bucket private and configure an OSS lifecycle rule to delete
# dobotai/im-images/ objects after 1 day. Never commit real credentials.
DOBOTAI_OUTBOUND_IMAGE_OSS_ENDPOINT=
DOBOTAI_OUTBOUND_IMAGE_OSS_REGION=
DOBOTAI_OUTBOUND_IMAGE_OSS_ACCESS_KEY_ID=
DOBOTAI_OUTBOUND_IMAGE_OSS_ACCESS_KEY_SECRET=
DOBOTAI_OUTBOUND_IMAGE_OSS_BUCKET=
# Optional support group for uncovered-answer notifications. This is the
# group's openConversationId, not a webhook. Leave empty to record coverage
# decisions and pending support cases without sending group notifications.
+104
View File
@@ -0,0 +1,104 @@
#!/usr/bin/env python3
"""提交信息规范校验(CI 侧权威闸门)。
规则与本机 ~/.githooks/commit-msg 完全一致,防止 --no-verify 绕过。
放行:Merge / Revert " / fixup! / squash! / amend! 开头的自动生成提交。
"""
import json
import os
import re
import subprocess
import sys
TYPES = "feat|fix|docs|style|refactor|perf|test|build|ci|chore|revert"
RE = re.compile(r"^(?:" + TYPES + r")(?:\([a-zA-Z0-9._/-]+\))?!?: .{3,}$")
ALLOW = re.compile(r'^(?:Merge |Revert "|fixup! |squash! |amend! )')
HELP = """
\u274c 提交信息不合规,已拦截。
规范 : <类型>(<范围>): <简短描述>
类型 : feat 新功能 | fix 修缺陷 | docs 文档 | style 格式(不改逻辑)
refactor 重构 | perf 性能 | test 测试 | build 构建/依赖
ci CI 配置 | chore 杂务 | revert 回滚
示例 : feat(asset): 新增按歌曲生成应援序列
fix(parser): 修复表格解析越界导致的 500
docs: 补充 OSS 短签名接口说明
要点 : 类型小写、冒号后一个空格、描述 \u22653 字符
标题 \u226472 字符、结尾不加句号
破坏性变更类型后加 !,正文写 BREAKING CHANGE:
详见 : docs/提交规范.md
"""
def collect():
"""优先用 Gitea 事件负载里的本次推送提交,取不到则退回 git log。"""
msgs = []
ev = os.environ.get("GITHUB_EVENT_PATH") or ""
if ev and os.path.exists(ev):
try:
with open(ev, encoding="utf-8") as fh:
d = json.load(fh)
for c in (d.get("commits") or []):
m = (c.get("message") or "").strip()
if m:
msgs.append(m)
except Exception as e:
print(" 事件负载解析失败(%s),回退到 git log" % e)
if not msgs:
out = subprocess.run(["git", "log", "-n", "5", "--format=%B%x00"],
capture_output=True, text=True).stdout
msgs = [x.strip() for x in out.split("\x00") if x.strip()]
return msgs
def subject_of(m):
for line in m.splitlines():
s = line.strip()
if not s or s.startswith("#"):
continue
return s
return ""
def judge(subj):
if not subj:
return "空提交信息"
if ALLOW.match(subj):
return ""
if len(subj) > 72:
return "标题 %d 字符 > 72" % len(subj)
if not RE.match(subj):
return "不符合 <类型>(<范围>): <描述>"
if subj.endswith(".") or subj.endswith("\u3002"):
return "标题结尾不要加句号"
return ""
def main():
msgs = collect()
bad = []
print("待校验提交 %d 条:" % len(msgs))
for m in msgs:
subj = subject_of(m)
why = judge(subj)
if why:
bad.append((subj, why))
print(" \u274c [%s] %s" % (why, subj))
else:
print(" \u2705 %s" % subj)
try:
with open("/tmp/ci-gate-status", "w") as fh:
fh.write("failure" if bad else "success")
except Exception:
pass
if bad:
print(HELP)
return 1
print("\u2705 提交信息全部合规")
return 0
if __name__ == "__main__":
sys.exit(main())
+82
View File
@@ -0,0 +1,82 @@
#!/usr/bin/env python3
"""把 CI 结果上报给主机侧 ci-bridge(bridge 负责落报告页 + 推微信)。"""
import json
import os
import subprocess
import urllib.request
def rd(path, default=""):
try:
with open(path, encoding="utf-8") as fh:
return fh.read().strip()
except Exception:
return default
def git(*args):
try:
return subprocess.run(["git"] + list(args), capture_output=True,
text=True).stdout.strip()
except Exception:
return ""
def main():
repo = os.environ.get("GITHUB_REPOSITORY", "unknown/unknown")
sha = rd("/tmp/ci-sha") or git("rev-parse", "HEAD")
branch = os.environ.get("GITHUB_REF_NAME") or \
os.environ.get("GITHUB_REF", "").replace("refs/heads/", "")
gate = rd("/tmp/ci-gate-status", "skipped")
build = rd("/tmp/ci-build-status", "failure")
status = "success" if (gate == "success" and build == "success") else "failure"
log = rd("/tmp/ci-build-tail")
if gate != "success":
log = "提交信息规范校验未通过\n" + log
try:
count = int(rd("/tmp/ci-commits-count", "1") or 1)
except Exception:
count = 1
rep = {
"repo": repo,
"sha": sha,
"branch": branch,
"actor": os.environ.get("GITHUB_ACTOR", ""),
"status": status,
"commit_subject": git("log", "-1", "--format=%s"),
"commit_count": count,
"duration_s": rd("/tmp/ci-duration", "?"),
"preview_url": rd("/tmp/ci-preview-url"),
"log": log[-3500:],
"steps": [
{"name": "提交信息规范校验", "status": gate},
{"name": "编译与自检", "status": build},
],
}
if rep["preview_url"]:
print("\U0001F310 网页预览: %s" % rep["preview_url"])
print("结果: %s | 构建: %s | 提交闸门: %s | 耗时: %ss"
% (status, build, gate, rep["duration_s"]))
base = os.environ.get("CI_BRIDGE_RESOLVED", "").rstrip("/")
if not base:
print("跳过上报:未找到可达的 ci-bridge")
return 0
data = json.dumps(rep, ensure_ascii=False).encode("utf-8")
req = urllib.request.Request(
base + "/api/ci/report", data=data, method="POST",
headers={"Content-Type": "application/json",
"X-CI-Token": os.environ.get("BRIDGE_TOKEN", "")})
try:
with urllib.request.urlopen(req, timeout=30) as resp:
print("\u2705 已上报 ci-bridge:", resp.read().decode()[:300])
except Exception as e:
print("\u274c 上报 ci-bridge 失败:", e)
return 0
if __name__ == "__main__":
raise SystemExit(main())
+156
View File
@@ -0,0 +1,156 @@
name: build
# 提交 -> 本机 Runner 编译 -> 微信通知 + 网页预览
# 实测结论(勿随意改动):
# 1. 不用任何第三方 action(作业容器不联网拉 action,全用镜像自带工具)
# 2. 必须自己 clone —— Runner 不会自动检出代码,工作区初始为空
# 3. act 以 `bash -e -o pipefail` 执行每步 -> 禁止 `cmd | head` 这种会 SIGPIPE 提前退出的管道
# 4. 工具链来自主机只读挂载 /opt/citools(go/uv),缓存写 /opt/citools-cache
# 5. bridge 从容器内走 docker 网关 172.17.0.1:8899(实测可达)
# 6. 校验/上报逻辑放在 .gitea/scripts/ 下的脚本里,YAML 只负责调用(避免 YAML 里嵌深缩进代码)
on:
push:
branches:
- "**"
workflow_dispatch:
env:
BRIDGE_URL: ${{ vars.CI_BRIDGE_URL }}
BRIDGE_TOKEN: ${{ secrets.CI_BRIDGE_TOKEN }}
GIT_TOKEN: ${{ secrets.GITHUB_TOKEN }}
concurrency:
group: build-${{ github.ref }}
cancel-in-progress: true
jobs:
build:
runs-on: ubuntu-latest
timeout-minutes: 40
steps:
- name: 检出代码
run: |
set -uo pipefail
SRV="${GITHUB_SERVER_URL:-http://192.168.50.223:3000}"
HP="${SRV#*://}"
BR="${GITHUB_REF_NAME:-${GITHUB_REF#refs/heads/}}"
cd "${GITHUB_WORKSPACE:-$(pwd)}"
rm -rf ./* .git 2>/dev/null || true
git init -q .
git remote add origin "http://x-access-token:${GIT_TOKEN}@${HP}/${GITHUB_REPOSITORY}.git"
git fetch -q --depth 50 origin "$BR"
git checkout -q -B "$BR" FETCH_HEAD
git rev-parse HEAD > /tmp/ci-sha
date +%s > /tmp/ci-start
: > /tmp/ci-preview-url
echo skipped > /tmp/ci-build-status
git rev-list --count FETCH_HEAD > /tmp/ci-commits-count 2>/dev/null || echo 1 > /tmp/ci-commits-count
echo "已检出 ${GITHUB_REPOSITORY} 分支 ${BR}"
git log -1 --format=' %h %an %s'
- name: 提交信息规范校验
run: python3 .gitea/scripts/ci_gate.py
- name: 编译与自检
run: |
set +e; set +o pipefail
LOG=/tmp/ci-build.log; : > "$LOG"
run() { echo "-- $* --"; "$@" 2>&1 | tee -a "$LOG"; return "${PIPESTATUS[0]}"; }
runin(){ local d="$1"; shift; echo "-- ($d) $* --"; ( cd "$d" && "$@" ) 2>&1 | tee -a "$LOG"; return "${PIPESTATUS[0]}"; }
export PATH=/opt/citools/go/bin:/opt/citools/bin:$PATH
export GOCACHE=/opt/citools-cache/gocache GOMODCACHE=/opt/citools-cache/gomod UV_CACHE_DIR=/opt/citools-cache/uv
export GOPROXY=https://goproxy.cn,direct GOTOOLCHAIN=local GOFLAGS=-mod=mod GOPATH=/opt/citools-cache/gopath
mkdir -p "$GOCACHE" "$GOMODCACHE" "$UV_CACHE_DIR" "$GOPATH" 2>/dev/null
echo "工具链: $(go version 2>&1) | $(uv --version 2>&1) | $(node -v 2>&1) | $(python3 -V 2>&1)"
fail=0; ran=""
if [ -f go.mod ]; then
ran="$ran go"
run go build ./... || fail=1
run go vet ./... || fail=1
run go test ./... || fail=1
fi
if [ -f pyproject.toml ] || [ -f requirements.txt ] || [ -f setup.py ]; then
ran="$ran python"
if grep -qE '^check:' Makefile 2>/dev/null; then
run make UV=uv check || fail=1
else
run uv sync || fail=1
run uv run pytest -q || fail=1
fi
fi
WEBPKG=""
for p in package.json web/package.json frontend/package.json ui/package.json; do
if [ -f "$p" ]; then WEBPKG="$p"; break; fi
done
if [ -n "$WEBPKG" ]; then
ran="$ran node"
D="$(dirname "$WEBPKG")"
if [ -f "$D/package-lock.json" ]; then
runin "$D" npm ci || fail=1
else
runin "$D" npm install || fail=1
fi
if grep -q '"build"[[:space:]]*:' "$WEBPKG" 2>/dev/null; then
runin "$D" npm run build || fail=1
fi
fi
if [ -z "$ran" ] && [ -f Makefile ]; then
ran="$ran make"
run make || fail=1
fi
if [ -z "$ran" ]; then
echo "未识别到构建入口(go.mod / pyproject.toml / package.json / Makefile 都没有),跳过编译"
fi
if [ "$fail" -eq 0 ]; then echo success > /tmp/ci-build-status; else echo failure > /tmp/ci-build-status; fi
tail -c 4000 "$LOG" > /tmp/ci-build-tail
echo "构建结果: $(cat /tmp/ci-build-status)(识别:$ran)"
[ "$fail" -eq 0 ] || exit 1
- name: 发布网页预览
if: always()
run: |
set +e; set +o pipefail
: > /tmp/ci-preview-url
if [ "$(cat /tmp/ci-build-status 2>/dev/null)" != "success" ]; then
echo "构建未成功,跳过预览发布"; exit 0
fi
WEB=""
for d in web/dist dist build out public web/build frontend/dist; do
if [ -f "$d/index.html" ]; then WEB="$d"; break; fi
done
if [ -z "$WEB" ]; then echo "未发现静态网页产物目录,跳过预览发布"; exit 0; fi
tar -czf /tmp/ci-web.tgz -C "$WEB" . || { echo "打包失败"; exit 0; }
B="${BRIDGE_URL:-}"
for cand in "$B" "http://172.17.0.1:8899" "http://192.168.50.134:8899" "http://100.92.49.78:8899"; do
if [ -n "$cand" ]; then
curl -s -m 5 -o /dev/null "$cand/healthz" && { B="$cand"; break; }
fi
done
SHA="$(cat /tmp/ci-sha)"
echo "发布预览到 $B(产物目录 $WEB)"
curl -sS -m 90 -X POST "$B/api/ci/publish?repo=${GITHUB_REPOSITORY}&sha=${SHA}" -H "X-CI-Token: ${BRIDGE_TOKEN}" -H 'Content-Type: application/gzip' --data-binary @/tmp/ci-web.tgz -o /tmp/ci-pub.json
cat /tmp/ci-pub.json; echo
URL="$(python3 -c "import json;print(json.load(open('/tmp/ci-pub.json')).get('url',''))" 2>/dev/null)"
if [ -n "$URL" ]; then echo "$URL" > /tmp/ci-preview-url; echo "网页预览: $URL"; fi
- name: 通知微信
if: always()
run: |
set +e; set +o pipefail
B="${BRIDGE_URL:-}"
for cand in "$B" "http://172.17.0.1:8899" "http://192.168.50.134:8899" "http://100.92.49.78:8899"; do
if [ -n "$cand" ]; then
curl -s -m 5 -o /dev/null "$cand/healthz" && { B="$cand"; break; }
fi
done
echo "使用 ci-bridge: $B"
S=$(date +%s); T=$(cat /tmp/ci-start 2>/dev/null); [ -n "$T" ] || T=$S
echo $((S - T)) > /tmp/ci-duration
CI_BRIDGE_RESOLVED="$B" python3 .gitea/scripts/ci_report.py
+5
View File
@@ -5,6 +5,11 @@
*.local
coverage.out
# Local deployment, backup, and production diagnostic artifacts
/.deploy/
/backups/
/check-live-dobotai.py
# Editors and OS
.idea/
.vscode/
File diff suppressed because it is too large Load Diff
+12
View File
@@ -15,6 +15,7 @@ Implemented:
- external-message idempotency;
- independent PostgreSQL session/turn storage;
- authenticated internal `POST /v1/qa` ingress endpoint.
- unauthenticated read-only `GET /version` runtime build identity endpoint.
- DingTalk Stream text ingress (official SDK) wired into `cmd/dobotai`.
- v1 knowledge coverage decisions, durable support cases, and optional DingTalk support-group notifications.
@@ -74,6 +75,17 @@ Health check:
curl http://127.0.0.1:8090/healthz
```
Runtime version check:
```bash
curl http://127.0.0.1:8090/version
```
`scripts/deploy-local.sh` builds from its own worktree (or the explicit
`DOBOTAI_SOURCE_ROOT`), injects `version`, `commit_sha`, and `build_time`, and
accepts the deployment only when `/version` reports the exact source commit.
It refuses dirty worktrees so the reported SHA cannot hide uncommitted source.
Internal QA request:
```bash
+81
View File
@@ -0,0 +1,81 @@
package main
import (
"context"
"errors"
"time"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/dingtalkconfig"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/questiontable"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/secretstore"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/settings"
)
type candidateConfigRepository interface {
CandidateDingTalkConfig(context.Context, string) (string, string, error)
ActivateDingTalkCandidate(context.Context, string, string) error
}
type candidateCredentialVerifier interface {
Verify(context.Context, string, string) dingtalkconfig.Result
}
var (
candidateTokenURL = ""
candidateTableBaseURL = ""
)
type dingTalkCandidateActivator struct {
tenant string
repo candidateConfigRepository
secrets *secretstore.Store
verifier candidateCredentialVerifier
bindings map[string]settings.ResourceBinding
restart func()
}
func (a *dingTalkCandidateActivator) ActivateCandidate(ctx context.Context) error {
if a == nil || a.repo == nil || a.secrets == nil || a.verifier == nil || a.restart == nil {
return errors.New("candidate_activation_unavailable")
}
clientID, secretRef, err := a.repo.CandidateDingTalkConfig(ctx, a.tenant)
if err != nil {
return errors.New("candidate_not_found")
}
secret, err := a.secrets.Get(secretRef)
if err != nil {
return errors.New("candidate_secret_unavailable")
}
if result := a.verifier.Verify(ctx, clientID, secret); !result.OK {
return errors.New("candidate_oauth_failed")
}
for _, kind := range []string{"question_table", "faq_table"} {
binding, ok := a.bindings[kind]
if !ok {
continue
}
client, err := questiontable.NewClient(questiontable.Config{ClientID: clientID, Secret: secret, BaseID: binding.BaseID, SheetID: binding.SheetID, OperatorID: binding.OperatorID, TokenURL: candidateTokenURL, BaseURL: candidateTableBaseURL}, nil)
if err != nil {
return errors.New("candidate_table_config_failed")
}
fields, err := client.ReadFields(ctx)
if err != nil {
return errors.New("candidate_table_probe_failed")
}
required := questiontable.QuestionSchema()
if kind == "faq_table" {
required = questiontable.FAQSchema()
}
if _, err = questiontable.MapRequiredFields(fields, required); err != nil {
return errors.New("candidate_table_schema_failed")
}
}
if err := a.repo.ActivateDingTalkCandidate(ctx, a.tenant, secretRef); err != nil {
return errors.New("candidate_database_activation_failed")
}
go func() {
time.Sleep(300 * time.Millisecond)
a.restart()
}()
return nil
}
+75
View File
@@ -0,0 +1,75 @@
package main
import (
"context"
"net/http"
"net/http/httptest"
"testing"
"time"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/dingtalkconfig"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/secretstore"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/settings"
)
type candidateRepoStub struct {
activated bool
ref string
}
func (r *candidateRepoStub) CandidateDingTalkConfig(context.Context, string) (string, string, error) {
return "app", r.ref, nil
}
func (r *candidateRepoStub) ActivateDingTalkCandidate(context.Context, string, string) error {
r.activated = true
return nil
}
type candidateVerifierStub struct{ ok bool }
func (v candidateVerifierStub) Verify(context.Context, string, string) dingtalkconfig.Result {
return dingtalkconfig.Result{OK: v.ok}
}
func TestCandidateActivatorRejectsOAuthWithoutActivation(t *testing.T) {
secrets, _ := secretstore.New(t.TempDir())
ref, _ := secrets.Put("dingtalk", "secret")
repo := &candidateRepoStub{ref: ref}
a := &dingTalkCandidateActivator{tenant: "t", repo: repo, secrets: secrets, verifier: candidateVerifierStub{}, bindings: map[string]settings.ResourceBinding{}, restart: func() {}}
if err := a.ActivateCandidate(context.Background()); err == nil {
t.Fatal("expected error")
}
if repo.activated {
t.Fatal("must not activate")
}
}
func TestCandidateActivatorValidatesSchemasThenActivates(t *testing.T) {
mux := http.NewServeMux()
mux.HandleFunc("/token", func(w http.ResponseWriter, _ *http.Request) {
_, _ = w.Write([]byte(`{"accessToken":"token","expireIn":7200}`))
})
mux.HandleFunc("/v1.0/notable/bases/base/sheets/sheet/fields", func(w http.ResponseWriter, _ *http.Request) {
_, _ = w.Write([]byte(`{"value":[{"id":"q","name":"中文问题","type":"text"},{"id":"tm","name":"提问时间","type":"date"},{"id":"u","name":"提问人","type":"user"},{"id":"o","name":"原始问题","type":"text"},{"id":"a","name":"AI答案","type":"text"},{"id":"p","name":"流程触发人员信息","type":"text"},{"id":"l","name":"语言","type":"text"},{"id":"d","name":"部门","type":"text"},{"id":"r","name":"区域","type":"text"}]}`))
})
server := httptest.NewServer(mux)
defer server.Close()
secrets, _ := secretstore.New(t.TempDir())
ref, _ := secrets.Put("dingtalk", "secret")
repo := &candidateRepoStub{ref: ref}
restarted := make(chan struct{}, 1)
a := &dingTalkCandidateActivator{tenant: "t", repo: repo, secrets: secrets, verifier: candidateVerifierStub{ok: true}, bindings: map[string]settings.ResourceBinding{"question_table": {BaseID: "base", SheetID: "sheet", OperatorID: "u"}}, restart: func() { restarted <- struct{}{} }}
oldTokenURL, oldBaseURL := candidateTokenURL, candidateTableBaseURL
candidateTokenURL, candidateTableBaseURL = server.URL+"/token", server.URL
defer func() { candidateTokenURL, candidateTableBaseURL = oldTokenURL, oldBaseURL }()
if err := a.ActivateCandidate(context.Background()); err != nil {
t.Fatal(err)
}
if !repo.activated {
t.Fatal("expected activation")
}
select {
case <-restarted:
case <-time.After(time.Second):
t.Fatal("restart not requested")
}
}
+4 -1
View File
@@ -2,6 +2,7 @@ package main
import (
"os"
"regexp"
"strings"
"testing"
)
@@ -15,10 +16,12 @@ func TestMainWiresDirectoryResolverIntoDingTalkHandler(t *testing.T) {
for _, required := range []string{
"directory.NewDingTalkLookup",
"directory.NewResolver",
"Directory: directoryResolver",
} {
if !strings.Contains(src, required) {
t.Errorf("main.go missing %q", required)
}
}
if !regexp.MustCompile(`Directory:\s*directoryResolver`).MatchString(src) {
t.Error("main.go does not wire directoryResolver into HandlerOptions.Directory")
}
}
+6 -5
View File
@@ -13,10 +13,11 @@ import (
)
type faqConfig struct {
BaseID string
SheetID string
ViewID string
OperatorID string
BaseID string
SheetID string
ViewID string
OperatorID string
FieldMapping map[string]string
}
func readFAQConfig() (faqConfig, error) {
@@ -46,7 +47,7 @@ func buildFAQDispatcher(repo faqtable.ExportRepository, clientID, clientSecret s
}
client, err := questiontable.NewClient(questiontable.Config{
ClientID: clientID, Secret: clientSecret, BaseID: cfg.BaseID,
SheetID: cfg.SheetID, OperatorID: cfg.OperatorID,
SheetID: cfg.SheetID, OperatorID: cfg.OperatorID, FieldMapping: cfg.FieldMapping,
}, nil)
if err != nil {
return nil, err
+147 -10
View File
@@ -8,9 +8,11 @@ import (
"os"
"os/signal"
"strings"
"sync/atomic"
"syscall"
"time"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/adminweb"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/agentbinding"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/audit"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/channels/dingtalk"
@@ -25,9 +27,24 @@ import (
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/weknora"
)
var (
version = "dev"
commitSHA = "unknown"
buildTime = "unknown"
)
func runtimeBuildInfo() httpapi.BuildInfo {
return httpapi.BuildInfo{
Version: version,
CommitSHA: commitSHA,
BuildTime: buildTime,
}
}
func main() {
ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
defer cancel()
var restartRequested atomic.Bool
storeCtx, storeCancel := context.WithTimeout(ctx, 10*time.Second)
defer storeCancel()
store, err := postgres.Open(storeCtx, os.Getenv("DATABASE_URL"))
@@ -35,6 +52,20 @@ func main() {
log.Fatal(err)
}
defer store.Close()
settingsRuntime, err := buildSettingsRuntime(storeCtx, store, os.Getenv("DOBOTAI_TENANT_KEY"))
if err != nil {
log.Fatal(err)
}
settingsRuntime.bindings.restartDelay = 300 * time.Millisecond
settingsRuntime.bindings.restart = func() { restartRequested.Store(true); cancel() }
address := strings.TrimSpace(os.Getenv("DOBOTAI_LISTEN_ADDR"))
if address == "" {
address = "127.0.0.1:8090"
}
if strings.TrimSpace(settingsRuntime.clientID) == "" || strings.TrimSpace(settingsRuntime.secret) == "" {
runSetupOnly(ctx, cancel, settingsRuntime, address, os.Getenv("DOBOTAI_TENANT_KEY"), &restartRequested)
return
}
client, err := weknora.NewClient(os.Getenv("WEKNORA_BASE_URL"), os.Getenv("WEKNORA_API_KEY"), 120*time.Second)
if err != nil {
log.Fatal(err)
@@ -66,31 +97,41 @@ func main() {
if err != nil {
log.Fatal(err)
}
address := strings.TrimSpace(os.Getenv("DOBOTAI_LISTEN_ADDR"))
if address == "" {
address = "127.0.0.1:8090"
}
auditOptions, err := buildAuditOptions(store, os.Getenv("DOBOTAI_TENANT_KEY"), os.Getenv("DOBOTAI_ADMIN_TOKEN"), address)
if err != nil {
log.Fatal(err)
}
auditOptions.BuildInfo = runtimeBuildInfo()
api, err := httpapi.NewWithOptions(qaService, os.Getenv("DOBOTAI_INGRESS_TOKEN"), auditOptions)
if err != nil {
log.Fatal(err)
}
dingCfg := dingtalk.StreamConfig{
ClientID: os.Getenv("DINGTALK_CLIENT_ID"),
ClientSecret: os.Getenv("DINGTALK_CLIENT_SECRET"),
ClientID: settingsRuntime.clientID,
ClientSecret: settingsRuntime.secret,
TenantKey: os.Getenv("DOBOTAI_TENANT_KEY"),
}
if err := dingCfg.Validate(); err != nil {
log.Fatal(err)
}
activeBindings, err := settingsRuntime.repository.ActiveBindings(storeCtx, dingCfg.TenantKey)
if err != nil {
log.Fatal(err)
}
settingsRuntime.admin.SetCandidateActivator(&dingTalkCandidateActivator{
tenant: dingCfg.TenantKey, repo: settingsRuntime.repository, secrets: settingsRuntime.secrets,
verifier: settingsRuntime.verifier, bindings: activeBindings,
restart: func() { restartRequested.Store(true); cancel() },
})
runtimeReady := make(chan struct{})
faqCfg, err := readFAQConfig()
if err != nil {
log.Fatal(err)
}
if binding, ok := activeBindings["faq_table"]; ok {
faqCfg = faqConfig{BaseID: binding.BaseID, SheetID: binding.SheetID, ViewID: binding.ViewID, OperatorID: binding.OperatorID, FieldMapping: binding.FieldMapping}
}
var faqWorkflowDone chan struct{}
var faqRecordLinks *faqtable.RecordLinkBuilder
if faqCfg.BaseID != "" {
@@ -109,23 +150,39 @@ func main() {
faqWorkflowDone = make(chan struct{})
go func() {
defer close(faqWorkflowDone)
select {
case <-runtimeReady:
case <-ctx.Done():
return
}
runFAQPoller(ctx, faqDispatcher, 5*time.Second)
}()
log.Printf("DingTalk FAQ table dispatcher started")
}
supportConversationID := os.Getenv("DINGTALK_SUPPORT_GROUP_OPEN_CONVERSATION_ID")
if binding, ok := activeBindings["support_group"]; ok {
supportConversationID = binding.ConversationID
}
supportDispatcher, err := buildSupportDispatcher(
store,
dingCfg.ClientID,
dingCfg.ClientSecret,
os.Getenv("DINGTALK_SUPPORT_GROUP_OPEN_CONVERSATION_ID"),
supportConversationID,
faqRecordLinks,
)
if err != nil {
log.Fatalf("DingTalk support notification init failed: %v", err)
}
if supportDispatcher != nil {
go runSupportDispatcher(ctx, supportDispatcher, 5*time.Second)
go func() {
select {
case <-runtimeReady:
case <-ctx.Done():
return
}
runSupportDispatcher(ctx, supportDispatcher, 5*time.Second)
}()
log.Printf("DingTalk support notification dispatcher started")
}
// Question table dispatcher (optional, fail-closed).
@@ -133,6 +190,10 @@ func main() {
if err != nil {
log.Fatal(err)
}
if binding, ok := activeBindings["question_table"]; ok {
qtCfg = questionTableConfig{BaseID: binding.BaseID, SheetID: binding.SheetID, OperatorID: binding.OperatorID, FieldMapping: binding.FieldMapping}
}
store.SetExternalTableExports(qtCfg.BaseID != "", faqCfg.BaseID != "")
var questionTableDone chan struct{}
if qtCfg.BaseID != "" {
qtRepo, err := postgres.NewQuestionTableExportRepository(store)
@@ -146,6 +207,11 @@ func main() {
questionTableDone = make(chan struct{})
go func() {
defer close(questionTableDone)
select {
case <-runtimeReady:
case <-ctx.Done():
return
}
runQuestionTablePoller(ctx, qtDispatcher, 5*time.Second)
}()
log.Printf("DingTalk question table dispatcher started")
@@ -166,6 +232,18 @@ func main() {
if err != nil {
log.Fatalf("DingTalk directory resolver init failed: %v", err)
}
outboundRewriter, err := buildOutboundImageRewriter(outboundImageConfig{
WeKnoraBaseURL: os.Getenv("WEKNORA_BASE_URL"),
WeKnoraAPIKey: os.Getenv("WEKNORA_API_KEY"),
Endpoint: os.Getenv("DOBOTAI_OUTBOUND_IMAGE_OSS_ENDPOINT"),
Region: os.Getenv("DOBOTAI_OUTBOUND_IMAGE_OSS_REGION"),
AccessKeyID: os.Getenv("DOBOTAI_OUTBOUND_IMAGE_OSS_ACCESS_KEY_ID"),
AccessKeySecret: os.Getenv("DOBOTAI_OUTBOUND_IMAGE_OSS_ACCESS_KEY_SECRET"),
Bucket: os.Getenv("DOBOTAI_OUTBOUND_IMAGE_OSS_BUCKET"),
})
if err != nil {
log.Fatalf("DingTalk outbound image init failed: %v", err)
}
feedbackService, err := feedback.NewService(dingCfg.TenantKey, "dingtalk", store)
if err != nil {
log.Fatal(err)
@@ -204,7 +282,14 @@ func main() {
dingtalk.TenantConfig{TenantKey: dingCfg.TenantKey, Channel: "dingtalk"},
qaService,
sender,
dingtalk.HandlerOptions{ImageDownloader: downloader, CardSender: cardSender, Directory: directoryResolver},
dingtalk.HandlerOptions{
ImageDownloader: downloader,
CardSender: cardSender,
Directory: directoryResolver,
OutboundRewriter: outboundRewriter,
SettingsRouter: &dingTalkSettingsRouter{controller: settingsRuntime.controller},
LanguageResolver: settingsRuntime.repository,
},
)
if err != nil {
log.Fatal(err)
@@ -217,9 +302,42 @@ func main() {
log.Fatal(err)
}
defer streamClient.Close()
if err := settingsRuntime.repository.MarkDingTalkRuntimeHealthy(context.Background(), dingCfg.TenantKey, settingsRuntime.secretRef); err != nil {
streamClient.Close()
log.Fatal(err)
}
log.Printf("DingTalk Stream ingress started for tenant %s", dingCfg.TenantKey)
}
server := &http.Server{Addr: address, Handler: api.Handler(), ReadHeaderTimeout: 5 * time.Second, ReadTimeout: 15 * time.Second, WriteTimeout: 130 * time.Second, IdleTimeout: 60 * time.Second}
close(runtimeReady)
_, questionBound := activeBindings["question_table"]
_, faqBound := activeBindings["faq_table"]
_, supportBound := activeBindings["support_group"]
weknoraHealthURL := strings.TrimRight(strings.TrimSpace(os.Getenv("WEKNORA_BASE_URL")), "/")
if weknoraHealthURL != "" {
weknoraHealthURL += "/health"
}
settingsRuntime.admin.SetOperationalStatusProvider(&operationalStatusProvider{
metrics: store,
tenant: dingCfg.TenantKey,
weknoraHealthURL: weknoraHealthURL,
httpClient: &http.Client{Timeout: 3 * time.Second},
now: time.Now,
version: runtimeBuildInfo().Version,
streamRunning: streamClient != nil,
questionRunning: questionTableDone != nil,
faqRunning: faqWorkflowDone != nil,
supportRunning: supportDispatcher != nil,
questionBound: questionBound,
faqBound: faqBound,
supportBound: supportBound,
ossConfigured: outboundOSSConfigured(),
})
rootHandler := http.NewServeMux()
rootHandler.Handle("/", api.Handler())
if strings.TrimSpace(os.Getenv("DOBOTAI_ADMIN_TOKEN")) != "" {
rootHandler.Handle("/admin/", adminweb.New(settingsRuntime.admin, os.Getenv("DOBOTAI_ADMIN_TOKEN")))
}
server := &http.Server{Addr: address, Handler: rootHandler, ReadHeaderTimeout: 5 * time.Second, ReadTimeout: 15 * time.Second, WriteTimeout: 130 * time.Second, IdleTimeout: 60 * time.Second}
go func() {
log.Printf("DobotAI ingress listening on %s", address)
if err := server.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
@@ -247,6 +365,25 @@ func main() {
log.Printf("question table poller shutdown timed out")
}
}
if streamClient != nil {
streamClient.Close()
}
if restartRequested.Load() {
os.Exit(75)
}
}
func outboundOSSConfigured() bool {
for _, key := range []string{
"DOBOTAI_OUTBOUND_IMAGE_OSS_ENDPOINT", "DOBOTAI_OUTBOUND_IMAGE_OSS_REGION",
"DOBOTAI_OUTBOUND_IMAGE_OSS_ACCESS_KEY_ID", "DOBOTAI_OUTBOUND_IMAGE_OSS_ACCESS_KEY_SECRET",
"DOBOTAI_OUTBOUND_IMAGE_OSS_BUCKET",
} {
if strings.TrimSpace(os.Getenv(key)) == "" {
return false
}
}
return true
}
func buildAuditOptions(repo audit.TurnRepository, tenantKey, adminToken, address string) (httpapi.Options, error) {
+43
View File
@@ -21,6 +21,20 @@ func (r *auditRepoStub) GetTurn(context.Context, string, string) (audit.TurnDeta
return audit.TurnDetail{}, audit.ErrNotFound
}
func TestRuntimeBuildInfoUsesLinkerVariables(t *testing.T) {
oldVersion, oldCommitSHA, oldBuildTime := version, commitSHA, buildTime
t.Cleanup(func() {
version, commitSHA, buildTime = oldVersion, oldCommitSHA, oldBuildTime
})
version = "2.1.0"
commitSHA = "abc123def456"
buildTime = "2026-07-27T12:00:00Z"
got := runtimeBuildInfo()
if got.Version != version || got.CommitSHA != commitSHA || got.BuildTime != buildTime {
t.Fatalf("build info=%+v", got)
}
}
func TestBuildAuditOptionsDisabledIsBackwardCompatible(t *testing.T) {
opts, err := buildAuditOptions(nil, "", "", "0.0.0.0:8090")
if err != nil {
@@ -101,3 +115,32 @@ func TestMainWiresOptionalDingTalkCardDebugUser(t *testing.T) {
}
}
}
func TestBuildOutboundImageRewriterConfig(t *testing.T) {
t.Run("disabled", func(t *testing.T) {
rewriter, err := buildOutboundImageRewriter(outboundImageConfig{})
if err != nil || rewriter != nil {
t.Fatalf("rewriter=%v err=%v", rewriter, err)
}
})
t.Run("partial configuration rejected", func(t *testing.T) {
_, err := buildOutboundImageRewriter(outboundImageConfig{Bucket: "private-bucket"})
if err == nil {
t.Fatal("expected configuration error")
}
})
t.Run("complete configuration", func(t *testing.T) {
rewriter, err := buildOutboundImageRewriter(outboundImageConfig{
WeKnoraBaseURL: "http://127.0.0.1:8080",
WeKnoraAPIKey: "api-key",
Endpoint: "https://oss-cn-shenzhen.aliyuncs.com",
Region: "cn-shenzhen",
AccessKeyID: "access-key-id",
AccessKeySecret: "access-key-secret",
Bucket: "private-bucket",
})
if err != nil || rewriter == nil {
t.Fatalf("rewriter=%v err=%v", rewriter, err)
}
})
}
+164
View File
@@ -0,0 +1,164 @@
package main
import (
"context"
"fmt"
"io"
"net/http"
"strings"
"time"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/adminweb"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/store/postgres"
)
type adminComponent = adminweb.Component
type operationalMetrics = postgres.OperationalMetrics
type operationalMetricsSource interface {
Ping(context.Context) error
ReadOperationalMetrics(context.Context, string) (postgres.OperationalMetrics, error)
}
type operationalStatusProvider struct {
metrics operationalMetricsSource
tenant string
weknoraHealthURL string
httpClient *http.Client
now func() time.Time
version string
streamRunning bool
questionRunning bool
faqRunning bool
supportRunning bool
questionBound bool
faqBound bool
supportBound bool
ossConfigured bool
}
func (p *operationalStatusProvider) Components(ctx context.Context) ([]adminweb.Component, error) {
now := time.Now()
if p.now != nil {
now = p.now()
}
components := []adminweb.Component{{Name: "运行版本", State: "ok", Code: "version_ok", Detail: "版本:" + safeVersion(p.version), LastSuccess: now, CheckedAt: now}}
metrics := operationalMetrics{}
if p.metrics == nil || p.metrics.Ping(ctx) != nil {
components = append(components, adminweb.Component{Name: "PostgreSQL", State: "error", Code: "database_unavailable", Detail: "数据库健康检查失败", CheckedAt: now})
} else {
var err error
metrics, err = p.metrics.ReadOperationalMetrics(ctx, p.tenant)
if err != nil {
components = append(components, adminweb.Component{Name: "PostgreSQL", State: "error", Code: "metrics_query_failed", Detail: "数据库可连接,但状态聚合失败", LastSuccess: now, CheckedAt: now})
} else {
components = append(components, adminweb.Component{Name: "PostgreSQL", State: "ok", Code: "database_ok", Detail: "连接与状态聚合正常", LastSuccess: now, CheckedAt: now})
}
}
components = append(components,
booleanComponent("钉钉Stream", p.streamRunning, "stream_running", "stream_stopped", "正式Stream消费者运行中", "正式Stream消费者未运行", now),
p.weknoraComponent(ctx, now),
outboxComponent("问题表Outbox", metrics.QuestionPending, metrics.QuestionFailed, metrics.LastQuestionSuccess, now),
outboxComponent("FAQ Outbox", metrics.FAQPending, metrics.FAQFailed, metrics.LastFAQSuccess, now),
supportComponent(p, metrics, now),
dispatcherComponent("问题表Dispatcher", p.questionBound, p.questionRunning, now),
dispatcherComponent("FAQ Dispatcher", p.faqBound, p.faqRunning, now),
booleanComponent("Private OSS图片", p.ossConfigured, "oss_configured", "oss_unconfigured", "图片出站配置完整", "图片出站未配置", now),
)
return components, nil
}
func (p *operationalStatusProvider) weknoraComponent(ctx context.Context, now time.Time) adminweb.Component {
component := adminweb.Component{Name: "WeKnora", State: "error", Code: "health_probe_failed", Detail: "健康探针失败", CheckedAt: now}
if strings.TrimSpace(p.weknoraHealthURL) == "" {
component.Code = "health_url_missing"
component.Detail = "健康探针地址未配置"
return component
}
client := p.httpClient
if client == nil {
client = &http.Client{Timeout: 3 * time.Second}
}
reqCtx, cancel := context.WithTimeout(ctx, 3*time.Second)
defer cancel()
req, err := http.NewRequestWithContext(reqCtx, http.MethodGet, p.weknoraHealthURL, nil)
if err != nil {
return component
}
resp, err := client.Do(req)
if err != nil {
return component
}
defer resp.Body.Close()
_, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 4<<10))
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
component.Code = "health_http_error"
component.Detail = fmt.Sprintf("健康探针HTTP %d", resp.StatusCode)
return component
}
component.State, component.Code, component.Detail = "ok", "health_ok", "HTTP健康探针正常"
component.LastSuccess = now
return component
}
func outboxComponent(name string, pending, failed int, lastSuccess, now time.Time) adminweb.Component {
component := adminweb.Component{Name: name, State: "ok", Code: "outbox_empty", Detail: "pending=0,failed=0", LastSuccess: lastSuccess, CheckedAt: now}
if failed > 0 {
component.State, component.Code = "error", "outbox_failed"
} else if pending > 0 {
component.State, component.Code = "warning", "outbox_pending"
}
component.Detail = fmt.Sprintf("pending=%d,failed=%d", pending, failed)
return component
}
func supportComponent(p *operationalStatusProvider, metrics operationalMetrics, now time.Time) adminweb.Component {
component := adminweb.Component{Name: "技术支持通知", CheckedAt: now, LastSuccess: metrics.LastSupportSuccess}
switch {
case !p.supportBound:
component.State, component.Code, component.Detail = "disabled", "binding_missing", "技术支持群未绑定"
case !p.faqBound:
component.State, component.Code, component.Detail = "blocked", "faq_binding_required", "已绑定支持群;需先激活FAQ表"
case metrics.SupportFailed > 0:
component.State, component.Code = "error", "notification_failed"
component.Detail = fmt.Sprintf("pending=%d,failed=%d", metrics.SupportPending, metrics.SupportFailed)
case metrics.SupportPending > 0:
component.State, component.Code = "warning", "notification_pending"
component.Detail = fmt.Sprintf("pending=%d,failed=0", metrics.SupportPending)
case p.supportRunning:
component.State, component.Code, component.Detail = "ok", "dispatcher_running", "通知Dispatcher运行中;pending=0,failed=0"
default:
component.State, component.Code, component.Detail = "error", "dispatcher_stopped", "绑定与依赖完整,但Dispatcher未运行"
}
return component
}
func dispatcherComponent(name string, bound, running bool, now time.Time) adminweb.Component {
if !bound {
return adminweb.Component{Name: name, State: "disabled", Code: "binding_missing", Detail: "无active绑定", CheckedAt: now}
}
if !running {
return adminweb.Component{Name: name, State: "error", Code: "dispatcher_stopped", Detail: "已有active绑定,但Dispatcher未运行", CheckedAt: now}
}
return adminweb.Component{Name: name, State: "ok", Code: "dispatcher_running", Detail: "Dispatcher运行中", LastSuccess: now, CheckedAt: now}
}
func booleanComponent(name string, ok bool, successCode, failureCode, successDetail, failureDetail string, now time.Time) adminweb.Component {
if ok {
return adminweb.Component{Name: name, State: "ok", Code: successCode, Detail: successDetail, LastSuccess: now, CheckedAt: now}
}
return adminweb.Component{Name: name, State: "disabled", Code: failureCode, Detail: failureDetail, CheckedAt: now}
}
func safeVersion(value string) string {
value = strings.TrimSpace(value)
if value == "" {
return "unknown"
}
if len(value) > 80 {
return value[:80]
}
return value
}
+105
View File
@@ -0,0 +1,105 @@
package main
import (
"context"
"net/http"
"net/http/httptest"
"strings"
"testing"
"time"
)
func TestOperationalStatusProviderBuildsLiveComponents(t *testing.T) {
health := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
w.WriteHeader(http.StatusOK)
}))
defer health.Close()
checked := time.Date(2026, 8, 13, 10, 0, 0, 0, time.UTC)
provider := &operationalStatusProvider{
metrics: operationalMetricsStub{value: operationalMetrics{
QuestionPending: 2, FAQFailed: 1, SupportPending: 3,
LastQuestionSuccess: checked.Add(-time.Minute),
}},
weknoraHealthURL: health.URL,
httpClient: health.Client(),
now: func() time.Time { return checked },
version: "settings-test",
streamRunning: true,
questionRunning: true,
faqRunning: false,
supportRunning: false,
questionBound: true,
faqBound: false,
supportBound: true,
ossConfigured: true,
}
components, err := provider.Components(context.Background())
if err != nil {
t.Fatal(err)
}
byName := componentMap(components)
assertComponent(t, byName, "运行版本", "ok", "version_ok")
assertComponent(t, byName, "PostgreSQL", "ok", "database_ok")
assertComponent(t, byName, "钉钉Stream", "ok", "stream_running")
assertComponent(t, byName, "WeKnora", "ok", "health_ok")
assertComponent(t, byName, "问题表Outbox", "warning", "outbox_pending")
assertComponent(t, byName, "FAQ Outbox", "error", "outbox_failed")
assertComponent(t, byName, "技术支持通知", "blocked", "faq_binding_required")
assertComponent(t, byName, "问题表Dispatcher", "ok", "dispatcher_running")
assertComponent(t, byName, "FAQ Dispatcher", "disabled", "binding_missing")
assertComponent(t, byName, "Private OSS图片", "ok", "oss_configured")
for _, component := range components {
for _, forbidden := range []string{health.URL, "cid-secret-group", "base-secret", "access-token"} {
if strings.Contains(component.Detail, forbidden) {
t.Fatalf("component %q leaks sensitive identifier %q", component.Name, forbidden)
}
}
}
}
func TestOperationalStatusProviderIsolatesWeKnoraFailure(t *testing.T) {
provider := &operationalStatusProvider{
metrics: operationalMetricsStub{value: operationalMetrics{}},
weknoraHealthURL: "http://127.0.0.1:1/health",
httpClient: &http.Client{Timeout: 50 * time.Millisecond},
now: time.Now,
}
components, err := provider.Components(context.Background())
if err != nil {
t.Fatal(err)
}
byName := componentMap(components)
assertComponent(t, byName, "PostgreSQL", "ok", "database_ok")
assertComponent(t, byName, "WeKnora", "error", "health_probe_failed")
}
type operationalMetricsStub struct {
value operationalMetrics
err error
}
func (s operationalMetricsStub) ReadOperationalMetrics(context.Context, string) (operationalMetrics, error) {
return s.value, s.err
}
func (s operationalMetricsStub) Ping(context.Context) error { return s.err }
func componentMap(values []adminComponent) map[string]adminComponent {
out := make(map[string]adminComponent, len(values))
for _, value := range values {
out[value.Name] = value
}
return out
}
func assertComponent(t *testing.T, values map[string]adminComponent, name, state, code string) {
t.Helper()
value, ok := values[name]
if !ok {
t.Fatalf("missing component %q: %+v", name, values)
}
if value.State != state || value.Code != code {
t.Fatalf("component %q=%+v want state=%q code=%q", name, value, state, code)
}
}
+70
View File
@@ -0,0 +1,70 @@
package main
import (
"errors"
"net/http"
"strings"
"time"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/channels/dingtalk"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/outboundimage"
)
const (
outboundImageMaxCount = 5
outboundImageMaxBytes = 10 << 20
outboundImageURLTTL = 2 * time.Hour
outboundImagePrefix = "dobotai/im-images"
)
type outboundImageConfig struct {
WeKnoraBaseURL string
WeKnoraAPIKey string
Endpoint string
Region string
AccessKeyID string
AccessKeySecret string
Bucket string
}
func buildOutboundImageRewriter(cfg outboundImageConfig) (dingtalk.OutboundContentRewriter, error) {
ossValues := []string{cfg.Endpoint, cfg.Region, cfg.AccessKeyID, cfg.AccessKeySecret, cfg.Bucket}
enabled := false
for _, value := range ossValues {
if strings.TrimSpace(value) != "" {
enabled = true
break
}
}
if !enabled {
return nil, nil
}
for _, value := range append([]string{cfg.WeKnoraBaseURL, cfg.WeKnoraAPIKey}, ossValues...) {
if strings.TrimSpace(value) == "" {
return nil, errors.New("outbound image: complete WeKnora and OSS configuration is required")
}
}
transport := http.DefaultTransport.(*http.Transport).Clone()
transport.Proxy = nil
fetcher, err := outboundimage.NewWeKnoraFetcher(
cfg.WeKnoraBaseURL,
cfg.WeKnoraAPIKey,
&http.Client{Timeout: 30 * time.Second, Transport: transport},
outboundImageMaxBytes,
)
if err != nil {
return nil, err
}
store, err := outboundimage.NewAliyunOSSStore(
cfg.Endpoint,
cfg.Region,
cfg.AccessKeyID,
cfg.AccessKeySecret,
cfg.Bucket,
)
if err != nil {
return nil, err
}
publisher := outboundimage.NewHashPublisher(store, outboundImagePrefix, outboundImageURLTTL)
return outboundimage.NewRewriter(fetcher, publisher, outboundImageMaxCount), nil
}
+10 -8
View File
@@ -19,9 +19,10 @@ type questionTablePollerRunner interface {
// questionTableConfig holds the optional Notable question-table env values.
type questionTableConfig struct {
BaseID string
SheetID string
OperatorID string
BaseID string
SheetID string
OperatorID string
FieldMapping map[string]string
}
// readQuestionTableConfig reads the three dedicated env vars and returns
@@ -61,11 +62,12 @@ func buildQuestionTableDispatcher(repo questiontable.ExportRepository, clientID,
}
client, err := questiontable.NewClient(questiontable.Config{
ClientID: clientID,
Secret: clientSecret,
BaseID: cfg.BaseID,
SheetID: cfg.SheetID,
OperatorID: cfg.OperatorID,
ClientID: clientID,
Secret: clientSecret,
BaseID: cfg.BaseID,
SheetID: cfg.SheetID,
OperatorID: cfg.OperatorID,
FieldMapping: cfg.FieldMapping,
}, nil)
if err != nil {
return nil, err
+133
View File
@@ -0,0 +1,133 @@
package main
import (
"context"
"errors"
"time"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/channels/dingtalk"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/questiontable"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/settings"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/store/postgres"
)
type validatedSettingsRepository struct {
*postgres.SettingsRepository
clientID string
secret string
testSupport func(context.Context, settings.ResourceBinding) error
validateTable func(context.Context, settings.ResourceBinding) (map[string]string, error)
saveBinding func(context.Context, settings.ResourceBinding) error
restart func()
restartDelay time.Duration
}
func (r *validatedSettingsRepository) TestAndActivateSupportGroup(ctx context.Context, binding settings.ResourceBinding) error {
if r.testSupport != nil {
if err := r.testSupport(ctx, binding); err != nil {
return errors.New("support_group_test_failed")
}
} else {
sender, err := dingtalk.NewSupportGroupSender(r.clientID, r.secret, binding.ConversationID, nil)
if err != nil {
return errors.New("support_group_test_invalid_config")
}
if err = sender.SendTest(ctx); err != nil {
return errors.New("support_group_test_failed")
}
}
binding.ValidationCode = "support_test_ok"
binding.Activate = true
var err error
if r.saveBinding != nil {
err = r.saveBinding(ctx, binding)
} else if r.SettingsRepository != nil {
err = r.SettingsRepository.SaveBinding(ctx, binding)
} else {
err = errors.New("support_group_binding_repository_unavailable")
}
if err != nil {
return err
}
if r.restart != nil {
go func() {
if r.restartDelay > 0 {
time.Sleep(r.restartDelay)
}
r.restart()
}()
}
return nil
}
func (r *validatedSettingsRepository) SaveBinding(ctx context.Context, binding settings.ResourceBinding) error {
activate := false
switch binding.Kind {
case "question_table", "faq_table":
var mapping map[string]string
var err error
if r.validateTable != nil {
mapping, err = r.validateTable(ctx, binding)
} else {
mapping, err = r.readAndMapTable(ctx, binding)
}
if err != nil {
return err
}
binding.FieldMapping = mapping
binding.ValidationCode = "schema_ok"
binding.Activate = true
activate = true
case "support_group":
binding.ValidationCode = "pending_test_message"
binding.Activate = false
default:
return errors.New("unsupported_binding_kind")
}
var err error
if r.saveBinding != nil {
err = r.saveBinding(ctx, binding)
} else if r.SettingsRepository != nil {
err = r.SettingsRepository.SaveBinding(ctx, binding)
} else {
err = errors.New("binding_repository_unavailable")
}
if err != nil {
return err
}
if activate {
r.requestRestart()
}
return nil
}
func (r *validatedSettingsRepository) readAndMapTable(ctx context.Context, binding settings.ResourceBinding) (map[string]string, error) {
client, err := questiontable.NewClient(questiontable.Config{
ClientID: r.clientID, Secret: r.secret, BaseID: binding.BaseID,
SheetID: binding.SheetID, OperatorID: binding.OperatorID,
}, nil)
if err != nil {
return nil, errors.New("table_binding_invalid_config")
}
fields, err := client.ReadFields(ctx)
if err != nil {
return nil, err
}
required := questiontable.QuestionSchema()
if binding.Kind == "faq_table" {
required = questiontable.FAQSchema()
}
return questiontable.MapRequiredFields(fields, required)
}
func (r *validatedSettingsRepository) requestRestart() {
if r.restart == nil {
return
}
go func() {
if r.restartDelay > 0 {
time.Sleep(r.restartDelay)
}
r.restart()
}()
}
@@ -0,0 +1,117 @@
package main
import (
"context"
"errors"
"testing"
"time"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/settings"
)
func TestTableBindingRestartsOnlyAfterActiveSave(t *testing.T) {
restarted := make(chan struct{}, 1)
var saved settings.ResourceBinding
repo := &validatedSettingsRepository{
validateTable: func(_ context.Context, binding settings.ResourceBinding) (map[string]string, error) {
return map[string]string{"answer": "field-answer"}, nil
},
saveBinding: func(_ context.Context, binding settings.ResourceBinding) error {
saved = binding
return nil
},
restart: func() { restarted <- struct{}{} },
restartDelay: 0,
}
if err := repo.SaveBinding(context.Background(), settings.ResourceBinding{Kind: "question_table", TenantKey: "tenant", BaseID: "base", SheetID: "sheet", OperatorID: "operator"}); err != nil {
t.Fatal(err)
}
if !saved.Activate || saved.ValidationCode != "schema_ok" || saved.FieldMapping["answer"] != "field-answer" {
t.Fatalf("binding=%+v", saved)
}
select {
case <-restarted:
case <-time.After(50 * time.Millisecond):
t.Fatal("restart not requested")
}
}
func TestTableBindingDoesNotRestartOnValidationOrSaveFailure(t *testing.T) {
for _, tc := range []struct {
name string
validateErr error
saveErr error
}{
{name: "validation failure", validateErr: errors.New("schema failed")},
{name: "save failure", saveErr: errors.New("save failed")},
} {
t.Run(tc.name, func(t *testing.T) {
restarted := make(chan struct{}, 1)
repo := &validatedSettingsRepository{
validateTable: func(context.Context, settings.ResourceBinding) (map[string]string, error) {
return map[string]string{"answer": "field"}, tc.validateErr
},
saveBinding: func(context.Context, settings.ResourceBinding) error { return tc.saveErr },
restart: func() { restarted <- struct{}{} },
}
if err := repo.SaveBinding(context.Background(), settings.ResourceBinding{Kind: "question_table"}); err == nil {
t.Fatal("expected failure")
}
select {
case <-restarted:
t.Fatal("must not restart after failed activation")
case <-time.After(50 * time.Millisecond):
}
})
}
}
func TestSupportGroupActivationRestartsOnlyAfterTestAndSaveSucceed(t *testing.T) {
for _, tc := range []struct {
name string
testErr error
saveErr error
wantSaved bool
wantRestart bool
}{
{name: "test fails", testErr: errors.New("send failed")},
{name: "save fails", saveErr: errors.New("save failed"), wantSaved: true},
{name: "success", wantSaved: true, wantRestart: true},
} {
t.Run(tc.name, func(t *testing.T) {
var saved settings.ResourceBinding
restarted := make(chan struct{}, 1)
repo := &validatedSettingsRepository{
testSupport: func(context.Context, settings.ResourceBinding) error { return tc.testErr },
saveBinding: func(_ context.Context, binding settings.ResourceBinding) error {
saved = binding
return tc.saveErr
},
restart: func() { restarted <- struct{}{} },
restartDelay: 0,
}
err := repo.TestAndActivateSupportGroup(context.Background(), settings.ResourceBinding{
Kind: "support_group", TenantKey: "tenant", ConversationID: "group",
})
if (err != nil) != (tc.testErr != nil || tc.saveErr != nil) {
t.Fatalf("err=%v", err)
}
if got := saved.Kind != ""; got != tc.wantSaved {
t.Fatalf("saved=%v want=%v", got, tc.wantSaved)
}
if tc.wantSaved && (saved.ValidationCode != "support_test_ok" || !saved.Activate) {
t.Fatalf("binding=%+v", saved)
}
select {
case <-restarted:
if !tc.wantRestart {
t.Fatal("unexpected restart")
}
case <-time.After(50 * time.Millisecond):
if tc.wantRestart {
t.Fatal("restart not requested")
}
}
})
}
}
+98
View File
@@ -0,0 +1,98 @@
package main
import (
"context"
"errors"
"os"
"path/filepath"
"strings"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/adminweb"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/channels/dingtalk"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/dingtalkconfig"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/secretstore"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/settings"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/store/postgres"
)
type dingTalkSettingsRouter struct{ controller *settings.Controller }
func (r *dingTalkSettingsRouter) Handle(ctx context.Context, m dingtalk.SettingsMessage) (bool, string, error) {
if r == nil || r.controller == nil {
return false, "", nil
}
if !r.controller.ShouldHandle(ctx, m.TenantKey, m.StaffID, m.Text) {
return false, "", nil
}
out, err := r.controller.Handle(ctx, settings.SettingInput{TenantKey: m.TenantKey, Channel: m.Channel, StaffID: m.StaffID, UnionID: m.UnionID, DisplayName: m.DisplayName, Department: m.Department, Text: m.Text, ConversationType: m.ConversationType, ConversationID: m.ConversationID, ConversationTitle: m.ConversationTitle})
return true, strings.TrimSpace(out), err
}
type settingsRuntime struct {
repository *postgres.SettingsRepository
bindings *validatedSettingsRepository
controller *settings.Controller
admin *adminweb.ControlService
clientID string
secret string
secretRef string
secrets *secretstore.Store
verifier *dingtalkconfig.Verifier
pendingActivation bool
}
func buildSettingsRuntime(ctx context.Context, store *postgres.Store, tenantKey string) (*settingsRuntime, error) {
tenantKey = strings.TrimSpace(tenantKey)
if tenantKey == "" {
return nil, errors.New("DOBOTAI_TENANT_KEY is required")
}
repo, err := postgres.NewSettingsRepository(store)
if err != nil {
return nil, err
}
root := strings.TrimSpace(os.Getenv("DOBOTAI_SECRET_DIR"))
if root == "" {
home, homeErr := os.UserHomeDir()
if homeErr != nil {
return nil, homeErr
}
root = filepath.Join(home, ".config", "dobotai", "secrets")
}
secrets, err := secretstore.New(root)
if err != nil {
return nil, err
}
pendingActivation, err := repo.PrepareDingTalkRuntimeStart(ctx, tenantKey)
if err != nil {
return nil, err
}
clientID, secretRef, _, err := repo.ActiveDingTalkConfig(ctx, tenantKey)
if err != nil {
return nil, err
}
var secret string
if secretRef != "" {
secret, err = secrets.Get(secretRef)
if err != nil {
return nil, errors.New("active DingTalk secret is unavailable")
}
}
if clientID == "" && strings.TrimSpace(os.Getenv("DINGTALK_CLIENT_ID")) != "" && strings.TrimSpace(os.Getenv("DINGTALK_CLIENT_SECRET")) != "" {
clientID, secret = strings.TrimSpace(os.Getenv("DINGTALK_CLIENT_ID")), strings.TrimSpace(os.Getenv("DINGTALK_CLIENT_SECRET"))
secretRef, err = secrets.Put("dingtalk", secret)
if err != nil {
return nil, err
}
if err = repo.SaveDingTalkConfig(ctx, tenantKey, clientID, secretRef, true); err != nil {
_ = secrets.Delete(secretRef)
return nil, err
}
}
verifier := dingtalkconfig.NewVerifier("https://oapi.dingtalk.com/gettoken", nil)
control, err := adminweb.NewControlService(tenantKey, repo, secrets, verifier)
if err != nil {
return nil, err
}
validatedRepo := &validatedSettingsRepository{SettingsRepository: repo, clientID: clientID, secret: secret}
return &settingsRuntime{repository: repo, bindings: validatedRepo, controller: settings.NewController(validatedRepo), admin: control, clientID: clientID, secret: secret, secretRef: secretRef, secrets: secrets, verifier: verifier, pendingActivation: pendingActivation}, nil
}
+65
View File
@@ -0,0 +1,65 @@
package main
import (
"context"
"encoding/json"
"errors"
"log"
"net/http"
"os"
"strings"
"sync/atomic"
"time"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/adminweb"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/settings"
)
func newSetupOnlyHandler(runtime *settingsRuntime, token string) (http.Handler, error) {
if runtime == nil || runtime.admin == nil {
return nil, errors.New("setup runtime unavailable")
}
if strings.TrimSpace(token) == "" {
return nil, errors.New("DOBOTAI_ADMIN_TOKEN is required in setup-only mode")
}
mux := http.NewServeMux()
mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, _ *http.Request) {
w.Header().Set("Content-Type", "application/json")
_ = json.NewEncoder(w).Encode(map[string]string{"status": "ok", "mode": "setup_required"})
})
mux.HandleFunc("GET /version", func(w http.ResponseWriter, _ *http.Request) {
w.Header().Set("Content-Type", "application/json")
_ = json.NewEncoder(w).Encode(runtimeBuildInfo())
})
mux.Handle("/admin/", adminweb.New(runtime.admin, token))
return mux, nil
}
func runSetupOnly(ctx context.Context, cancel context.CancelFunc, runtime *settingsRuntime, address, tenant string, restartRequested *atomic.Bool) {
runtime.admin.SetCandidateActivator(&dingTalkCandidateActivator{
tenant: tenant, repo: runtime.repository, secrets: runtime.secrets, verifier: runtime.verifier,
bindings: map[string]settings.ResourceBinding{},
restart: func() { restartRequested.Store(true); cancel() },
})
handler, err := newSetupOnlyHandler(runtime, os.Getenv("DOBOTAI_ADMIN_TOKEN"))
if err != nil {
log.Fatal(err)
}
server := &http.Server{Addr: address, Handler: handler, ReadHeaderTimeout: 5 * time.Second, ReadTimeout: 15 * time.Second, WriteTimeout: 20 * time.Second, IdleTimeout: 60 * time.Second}
go func() {
log.Printf("DobotAI setup-only listener on %s", address)
if err := server.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
log.Printf("setup HTTP server failed: %v", err)
cancel()
}
}()
<-ctx.Done()
shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 10*time.Second)
defer shutdownCancel()
if err := server.Shutdown(shutdownCtx); err != nil {
log.Printf("setup HTTP shutdown: %v", err)
}
if restartRequested.Load() {
os.Exit(75)
}
}
+102
View File
@@ -0,0 +1,102 @@
package main
import (
"context"
"encoding/json"
"net/http"
"net/http/httptest"
"testing"
"time"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/adminweb"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/dingtalkconfig"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/secretstore"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/settings"
)
type setupRepoStub struct{}
func (setupRepoStub) ActiveDingTalkConfig(context.Context, string) (string, string, bool, error) {
return "", "", false, nil
}
func (setupRepoStub) CandidateDingTalkConfig(context.Context, string) (string, string, error) {
return "", "", nil
}
func (setupRepoStub) SaveDingTalkConfig(context.Context, string, string, string, bool) error {
return nil
}
func (setupRepoStub) MarkDingTalkCandidateFailed(context.Context, string, string, string) error {
return nil
}
func (setupRepoStub) AdministratorCount(context.Context, string) (int, error) { return 0, nil }
func (setupRepoStub) ActiveBindings(context.Context, string) (map[string]settings.ResourceBinding, error) {
return map[string]settings.ResourceBinding{}, nil
}
func (setupRepoStub) CreateBootstrapChallenge(context.Context, string, string, time.Time) error {
return nil
}
func (setupRepoStub) ConfirmBootstrap(context.Context, string) error { return nil }
type setupVerifierStub struct{}
func (setupVerifierStub) Verify(context.Context, string, string) dingtalkconfig.Result {
return dingtalkconfig.Result{}
}
func setupControl(t *testing.T) *adminweb.ControlService {
t.Helper()
secrets, err := secretstore.New(t.TempDir())
if err != nil {
t.Fatal(err)
}
control, err := adminweb.NewControlService("tenant", setupRepoStub{}, secrets, setupVerifierStub{})
if err != nil {
t.Fatal(err)
}
return control
}
func TestSetupOnlyHandlerExposesHealthVersionAndProtectedAdmin(t *testing.T) {
runtime := &settingsRuntime{admin: setupControl(t)}
h, err := newSetupOnlyHandler(runtime, "admin-token")
if err != nil {
t.Fatal(err)
}
req := httptest.NewRequest(http.MethodGet, "/healthz", nil)
resp := httptest.NewRecorder()
h.ServeHTTP(resp, req)
if resp.Code != http.StatusOK {
t.Fatalf("health status=%d", resp.Code)
}
var health map[string]string
if err = json.Unmarshal(resp.Body.Bytes(), &health); err != nil {
t.Fatal(err)
}
if health["status"] != "ok" || health["mode"] != "setup_required" {
t.Fatalf("health=%v", health)
}
resp = httptest.NewRecorder()
h.ServeHTTP(resp, httptest.NewRequest(http.MethodGet, "/version", nil))
if resp.Code != http.StatusOK {
t.Fatalf("version status=%d", resp.Code)
}
resp = httptest.NewRecorder()
h.ServeHTTP(resp, httptest.NewRequest(http.MethodGet, "/admin/settings", nil))
if resp.Code != http.StatusUnauthorized {
t.Fatalf("unauth status=%d", resp.Code)
}
req = httptest.NewRequest(http.MethodGet, "/admin/settings", nil)
req.Header.Set("Authorization", "Bearer admin-token")
resp = httptest.NewRecorder()
h.ServeHTTP(resp, req)
if resp.Code != http.StatusOK {
t.Fatalf("auth status=%d body=%s", resp.Code, resp.Body.String())
}
}
func TestSetupOnlyHandlerRequiresAdminToken(t *testing.T) {
_, err := newSetupOnlyHandler(&settingsRuntime{admin: setupControl(t)}, "")
if err == nil {
t.Fatal("expected token error")
}
}
+1 -1
View File
@@ -25,7 +25,7 @@ func buildSupportDispatcher(repo coverage.NotificationRepository, clientID, clie
return nil, errors.New("support notification repository is required")
}
if recordLinks == nil {
return nil, errors.New("support notification FAQ record links are required")
return nil, nil
}
sender, err := dingtalk.NewSupportGroupSender(clientID, clientSecret, openConversationID, nil, dingtalk.WithFAQRecordLinkBuilder(recordLinks))
+9 -2
View File
@@ -62,6 +62,13 @@ func TestBuildSupportDispatcherDisabledWithoutGroup(t *testing.T) {
}
}
func TestBuildSupportDispatcherBlocksWithoutFAQLinks(t *testing.T) {
dispatcher, err := buildSupportDispatcher(&supportNotificationRepoStub{}, "client", "secret", "group", nil)
if err != nil || dispatcher != nil {
t.Fatalf("dispatcher=%v err=%v, want blocked without process failure", dispatcher, err)
}
}
func TestBuildSupportDispatcherEnabledFailsClosed(t *testing.T) {
repo := &supportNotificationRepoStub{}
links := testFAQRecordLinks(t)
@@ -74,7 +81,6 @@ func TestBuildSupportDispatcherEnabledFailsClosed(t *testing.T) {
{name: "missing repo", clientID: "client", secret: "secret", gid: "group", links: links},
{name: "missing client", repo: repo, secret: "secret", gid: "group", links: links},
{name: "missing secret", repo: repo, clientID: "client", gid: "group", links: links},
{name: "missing FAQ links", repo: repo, clientID: "client", secret: "secret", gid: "group"},
} {
t.Run(tc.name, func(t *testing.T) {
if _, err := buildSupportDispatcher(tc.repo, tc.clientID, tc.secret, tc.gid, tc.links); err == nil {
@@ -163,7 +169,8 @@ func TestMainWiresCoverageSupportNotifications(t *testing.T) {
`os.Getenv("DINGTALK_SUPPORT_GROUP_OPEN_CONVERSATION_ID")`,
`coverage.NewService(store)`,
`qa.WithSucceededTurnObserver`,
`go runSupportDispatcher(ctx, supportDispatcher, 5*time.Second)`,
`case <-runtimeReady:`,
`runSupportDispatcher(ctx, supportDispatcher, 5*time.Second)`,
} {
if !strings.Contains(source, fragment) {
t.Errorf("main support notification wiring missing %q", fragment)
@@ -0,0 +1 @@
DROP TABLE IF EXISTS config_audit_events; DROP TABLE IF EXISTS system_resource_bindings; DROP TABLE IF EXISTS administrators; DROP TABLE IF EXISTS system_config_revisions;
@@ -0,0 +1,19 @@
CREATE TABLE IF NOT EXISTS system_config_revisions (
id TEXT PRIMARY KEY, tenant_key TEXT NOT NULL, kind TEXT NOT NULL, revision INTEGER NOT NULL CHECK(revision>0),
state TEXT NOT NULL CHECK(state IN ('draft','validated','active','retired','failed')), public_config JSONB NOT NULL DEFAULT '{}'::JSONB,
secret_ref TEXT NOT NULL DEFAULT '', validation_status TEXT NOT NULL DEFAULT 'pending', validation_code TEXT NOT NULL DEFAULT '',
validated_at TIMESTAMPTZ, activated_at TIMESTAMPTZ, created_by_staff_id TEXT NOT NULL DEFAULT '', created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
UNIQUE(tenant_key,kind,revision));
CREATE UNIQUE INDEX IF NOT EXISTS uq_system_config_active ON system_config_revisions(tenant_key,kind) WHERE state='active';
CREATE TABLE IF NOT EXISTS administrators (
id TEXT PRIMARY KEY, tenant_key TEXT NOT NULL, staff_id TEXT NOT NULL, union_id TEXT NOT NULL DEFAULT '', display_name TEXT NOT NULL DEFAULT '',
department TEXT NOT NULL DEFAULT '', role TEXT NOT NULL CHECK(role IN ('super_admin','admin')), status TEXT NOT NULL DEFAULT 'active' CHECK(status IN ('active','disabled')),
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), UNIQUE(tenant_key,staff_id));
CREATE TABLE IF NOT EXISTS system_resource_bindings (
id TEXT PRIMARY KEY, tenant_key TEXT NOT NULL, kind TEXT NOT NULL, revision INTEGER NOT NULL CHECK(revision>0), state TEXT NOT NULL CHECK(state IN ('draft','validated','active','retired','failed')),
public_config JSONB NOT NULL DEFAULT '{}'::JSONB, validation_code TEXT NOT NULL DEFAULT '', validated_at TIMESTAMPTZ, activated_at TIMESTAMPTZ,
created_by_staff_id TEXT NOT NULL DEFAULT '', created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), UNIQUE(tenant_key,kind,revision));
CREATE UNIQUE INDEX IF NOT EXISTS uq_resource_binding_active ON system_resource_bindings(tenant_key,kind) WHERE state='active';
CREATE TABLE IF NOT EXISTS config_audit_events (
id TEXT PRIMARY KEY, tenant_key TEXT NOT NULL, actor_staff_id TEXT NOT NULL DEFAULT '', action TEXT NOT NULL, target_kind TEXT NOT NULL,
target_revision INTEGER NOT NULL DEFAULT 0, result_code TEXT NOT NULL, metadata JSONB NOT NULL DEFAULT '{}'::JSONB, created_at TIMESTAMPTZ NOT NULL DEFAULT NOW());
@@ -0,0 +1,25 @@
package migrations_test
import (
"os"
"strings"
"testing"
)
func TestUp012CreatesSettingsWithoutSecrets(t *testing.T) {
b, e := os.ReadFile("000012_system_settings.up.sql")
if e != nil {
t.Fatal(e)
}
s := strings.ToUpper(string(b))
for _, x := range []string{"SYSTEM_CONFIG_REVISIONS", "ADMINISTRATORS", "SYSTEM_RESOURCE_BINDINGS", "CONFIG_AUDIT_EVENTS", "PUBLIC_CONFIG"} {
if !strings.Contains(s, x) {
t.Errorf("missing %s", x)
}
}
for _, x := range []string{"CLIENT_SECRET JSON", "ACCESS_TOKEN JSON", "API_KEY JSON"} {
if strings.Contains(s, x) {
t.Errorf("forbidden %s", x)
}
}
}
@@ -0,0 +1 @@
DROP TABLE IF EXISTS admin_bootstrap_challenges;
@@ -0,0 +1,5 @@
CREATE TABLE IF NOT EXISTS admin_bootstrap_challenges (
id TEXT PRIMARY KEY, tenant_key TEXT NOT NULL, code_hash TEXT NOT NULL, expires_at TIMESTAMPTZ NOT NULL, attempts INTEGER NOT NULL DEFAULT 0 CHECK(attempts>=0), max_attempts INTEGER NOT NULL DEFAULT 5 CHECK(max_attempts BETWEEN 1 AND 10),
candidate_staff_id TEXT NOT NULL DEFAULT '', candidate_union_id TEXT NOT NULL DEFAULT '', candidate_display_name TEXT NOT NULL DEFAULT '', candidate_department TEXT NOT NULL DEFAULT '',
status TEXT NOT NULL CHECK(status IN ('pending','candidate_detected','confirmed','expired','locked')), created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW());
CREATE UNIQUE INDEX IF NOT EXISTS uq_admin_bootstrap_pending ON admin_bootstrap_challenges(tenant_key) WHERE status IN ('pending','candidate_detected');
@@ -0,0 +1,18 @@
package migrations_test
import (
"os"
"strings"
"testing"
)
func TestUp013StoresOnlyBootstrapHash(t *testing.T) {
b, e := os.ReadFile("000013_admin_bootstrap.up.sql")
if e != nil {
t.Fatal(e)
}
s := strings.ToUpper(string(b))
if !strings.Contains(s, "CODE_HASH") || strings.Contains(s, " CODE TEXT") {
t.Fatal("bootstrap must store hash only")
}
}
@@ -0,0 +1 @@
DROP TABLE IF EXISTS setting_sessions;
@@ -0,0 +1,5 @@
CREATE TABLE IF NOT EXISTS setting_sessions (
tenant_key TEXT NOT NULL, staff_id TEXT NOT NULL, conversation_id TEXT NOT NULL, conversation_type TEXT NOT NULL CHECK(conversation_type IN ('direct','group')),
current_step TEXT NOT NULL, pending_action TEXT NOT NULL DEFAULT '', draft_config JSONB NOT NULL DEFAULT '{}'::JSONB, expires_at TIMESTAMPTZ NOT NULL, updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
PRIMARY KEY(tenant_key,staff_id));
CREATE INDEX IF NOT EXISTS idx_setting_sessions_expiry ON setting_sessions(expires_at);
@@ -0,0 +1,20 @@
package migrations_test
import (
"os"
"strings"
"testing"
)
func TestUp014CreatesExpiringSettingSessions(t *testing.T) {
b, e := os.ReadFile("000014_setting_sessions.up.sql")
if e != nil {
t.Fatal(e)
}
s := strings.ToUpper(string(b))
for _, x := range []string{"SETTING_SESSIONS", "EXPIRES_AT", "DRAFT_CONFIG"} {
if !strings.Contains(s, x) {
t.Errorf("missing %s", x)
}
}
}
@@ -0,0 +1 @@
DROP TABLE IF EXISTS user_preferences;
@@ -0,0 +1,3 @@
CREATE TABLE IF NOT EXISTS user_preferences (
tenant_key TEXT NOT NULL, channel TEXT NOT NULL, external_actor_id TEXT NOT NULL, language_mode TEXT NOT NULL DEFAULT 'auto' CHECK(language_mode IN ('auto','fixed')),
locale TEXT NOT NULL DEFAULT 'auto' CHECK(locale IN ('auto','zh-CN','en','ja','ko','es')), updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), PRIMARY KEY(tenant_key,channel,external_actor_id));
@@ -0,0 +1,20 @@
package migrations_test
import (
"os"
"strings"
"testing"
)
func TestUp015ScopesPreferencesByTenantAndActor(t *testing.T) {
b, e := os.ReadFile("000015_user_preferences.up.sql")
if e != nil {
t.Fatal(e)
}
s := strings.ToUpper(string(b))
for _, x := range []string{"USER_PREFERENCES", "TENANT_KEY", "EXTERNAL_ACTOR_ID", "LANGUAGE_MODE", "LOCALE"} {
if !strings.Contains(s, x) {
t.Errorf("missing %s", x)
}
}
}
+2
View File
@@ -3,6 +3,7 @@ module gitea.zheromiku.top/Weknora/dobotai-platform
go 1.25.0
require (
github.com/aliyun/alibabacloud-oss-go-sdk-v2 v1.5.1
github.com/jackc/pgx/v5 v5.9.2
github.com/open-dingtalk/dingtalk-stream-sdk-go v0.9.1
)
@@ -15,4 +16,5 @@ require (
github.com/jackc/puddle/v2 v2.2.2 // indirect
golang.org/x/sync v0.20.0 // indirect
golang.org/x/text v0.37.0 // indirect
golang.org/x/time v0.4.0 // indirect
)
+4
View File
@@ -1,3 +1,5 @@
github.com/aliyun/alibabacloud-oss-go-sdk-v2 v1.5.1 h1:vtiFd0hhPAbyYJjztl0wYUq/PqEGkIlDmVuTIy6zw8Y=
github.com/aliyun/alibabacloud-oss-go-sdk-v2 v1.5.1/go.mod h1:FTzydeQVmR24FI0D6XWUOMKckjXehM/jgMn1xC+DA9M=
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
@@ -26,6 +28,8 @@ golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4=
golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
golang.org/x/text v0.37.0 h1:Cqjiwd9eSg8e0QAkyCaQTNHFIIzWtidPahFWR83rTrc=
golang.org/x/text v0.37.0/go.mod h1:a5sjxXGs9hsn/AJVwuElvCAo9v8QYLzvavO5z2PiM38=
golang.org/x/time v0.4.0 h1:Z81tqI5ddIoXDPvVQ7/7CC9TnLM7ubaFG2qXYd5BbYY=
golang.org/x/time v0.4.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+205
View File
@@ -0,0 +1,205 @@
package adminweb
import (
"context"
"errors"
"strings"
"time"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/dingtalkconfig"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/secretstore"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/settings"
)
type ConfigRepository interface {
ActiveDingTalkConfig(context.Context, string) (string, string, bool, error)
CandidateDingTalkConfig(context.Context, string) (string, string, error)
SaveDingTalkConfig(context.Context, string, string, string, bool) error
MarkDingTalkCandidateFailed(context.Context, string, string, string) error
AdministratorCount(context.Context, string) (int, error)
ActiveBindings(context.Context, string) (map[string]settings.ResourceBinding, error)
CreateBootstrapChallenge(context.Context, string, string, time.Time) error
ConfirmBootstrap(context.Context, string) error
}
type AdministratorRepository interface {
ListAdministrators(context.Context, string) ([]settings.Administrator, error)
DisableAdministrator(context.Context, string, string) error
}
type CandidateActivator interface{ ActivateCandidate(context.Context) error }
type OperationalStatusProvider interface {
Components(context.Context) ([]Component, error)
}
type CredentialVerifier interface {
Verify(context.Context, string, string) dingtalkconfig.Result
}
type ControlService struct {
tenant string
repo ConfigRepository
secrets *secretstore.Store
verifier CredentialVerifier
activator CandidateActivator
status OperationalStatusProvider
}
func NewControlService(tenant string, repo ConfigRepository, secrets *secretstore.Store, verifier CredentialVerifier) (*ControlService, error) {
if strings.TrimSpace(tenant) == "" || repo == nil || secrets == nil || verifier == nil {
return nil, errors.New("admin control dependencies are required")
}
return &ControlService{tenant: strings.TrimSpace(tenant), repo: repo, secrets: secrets, verifier: verifier}, nil
}
func (s *ControlService) Snapshot(ctx context.Context) (Snapshot, error) {
id, ref, verified, err := s.repo.ActiveDingTalkConfig(ctx, s.tenant)
if err != nil {
return Snapshot{}, err
}
admins, err := s.repo.AdministratorCount(ctx, s.tenant)
if err != nil {
return Snapshot{}, err
}
bindings, err := s.repo.ActiveBindings(ctx, s.tenant)
if err != nil {
return Snapshot{}, err
}
administratorViews := []AdministratorView{}
if administratorRepo, ok := s.repo.(AdministratorRepository); ok {
administrators, listErr := administratorRepo.ListAdministrators(ctx, s.tenant)
if listErr != nil {
return Snapshot{}, listErr
}
activeCount := 0
for _, administrator := range administrators {
if administrator.Status == "active" {
activeCount++
}
}
for _, administrator := range administrators {
administratorViews = append(administratorViews, AdministratorView{
StaffID: administrator.StaffID, StaffIDMasked: maskID(administrator.StaffID),
DisplayName: administrator.DisplayName, Department: administrator.Department,
Role: administrator.Role, Status: administrator.Status,
CanDisable: administrator.Status == "active" && activeCount > 1,
})
}
}
state := settings.DetermineState(settings.Snapshot{ClientID: id, SecretConfigured: s.secrets.Configured(ref), CredentialsVerified: verified, AdministratorCount: admins})
now := time.Now()
components := []Component{
{Name: "配置仓库", State: "ok", Code: "repository_ok", Detail: "PostgreSQL配置查询成功", LastSuccess: now, CheckedAt: now},
{Name: "钉钉OAuth", State: status(verified), Code: ternaryCode(verified, "oauth_verified", "oauth_unconfigured"), Detail: "当前应用凭据验证状态", CheckedAt: now},
{Name: "管理员", State: countStatus(admins), Code: ternaryCode(admins > 0, "admin_configured", "admin_missing"), Detail: itoa(admins) + " 位有效管理员", CheckedAt: now},
{Name: "技术支持群绑定", State: bindingStatus(bindings, "support_group"), Code: bindingCode(bindings, "support_group"), Detail: "仅显示是否存在active绑定", CheckedAt: now},
{Name: "问题AI表格绑定", State: bindingStatus(bindings, "question_table"), Code: bindingCode(bindings, "question_table"), Detail: "仅显示是否存在active绑定", CheckedAt: now},
{Name: "FAQ表格绑定", State: bindingStatus(bindings, "faq_table"), Code: bindingCode(bindings, "faq_table"), Detail: "仅显示是否存在active绑定", CheckedAt: now},
}
if s.status != nil {
operational, probeErr := s.status.Components(ctx)
if probeErr != nil {
components = append(components, Component{Name: "运行探针", State: "error", Code: "component_probe_failed", Detail: "运行状态探针失败;未显示原始错误", CheckedAt: time.Now()})
} else {
components = append(components, operational...)
}
}
return Snapshot{SystemState: string(state), ClientIDMasked: maskID(id), SecretConfigured: s.secrets.Configured(ref), Components: components, Administrators: administratorViews}, nil
}
func (s *ControlService) VerifyCredentials(ctx context.Context, id, secret string) Verification {
r := s.verifier.Verify(ctx, id, secret)
return Verification{OK: r.OK, Code: r.Code, CheckedAt: r.CheckedAt}
}
func (s *ControlService) SaveCredentials(ctx context.Context, id, secret string) error {
_, oldRef, _ := s.repo.CandidateDingTalkConfig(ctx, s.tenant)
ref, err := s.secrets.Put("dingtalk", secret)
if err != nil {
return err
}
if err = s.repo.SaveDingTalkConfig(ctx, s.tenant, strings.TrimSpace(id), ref, false); err != nil {
_ = s.secrets.Delete(ref)
return err
}
if oldRef != "" && oldRef != ref {
_ = s.secrets.Delete(oldRef)
}
return nil
}
func (s *ControlService) SetCandidateActivator(activator CandidateActivator) { s.activator = activator }
func (s *ControlService) SetOperationalStatusProvider(provider OperationalStatusProvider) {
s.status = provider
}
func (s *ControlService) ActivateCandidate(ctx context.Context) error {
if s.activator == nil {
return errors.New("candidate activator unavailable")
}
_, ref, err := s.repo.CandidateDingTalkConfig(ctx, s.tenant)
if err != nil {
return err
}
if err = s.activator.ActivateCandidate(ctx); err != nil {
_ = s.repo.MarkDingTalkCandidateFailed(ctx, s.tenant, ref, "activation_failed")
return errors.New("candidate activation failed")
}
return nil
}
func (s *ControlService) CreateBootstrap(ctx context.Context, code string) error {
return s.repo.CreateBootstrapChallenge(ctx, s.tenant, code, time.Now().Add(10*time.Minute))
}
func (s *ControlService) ConfirmBootstrap(ctx context.Context) error {
return s.repo.ConfirmBootstrap(ctx, s.tenant)
}
func (s *ControlService) DisableAdministrator(ctx context.Context, staffID string) error {
staffID = strings.TrimSpace(staffID)
if staffID == "" {
return settings.ErrAdministratorNotFound
}
repo, ok := s.repo.(AdministratorRepository)
if !ok {
return errors.New("administrator management unavailable")
}
return repo.DisableAdministrator(ctx, s.tenant, staffID)
}
func maskID(v string) string {
v = strings.TrimSpace(v)
if len(v) <= 6 {
return v
}
return v[:3] + "***" + v[len(v)-3:]
}
func status(ok bool) string {
if ok {
return "ok"
}
return "unconfigured"
}
func countStatus(n int) string {
if n > 0 {
return "ok"
}
return "unconfigured"
}
func bindingStatus(m map[string]settings.ResourceBinding, k string) string {
if _, ok := m[k]; ok {
return "configured"
}
return "unconfigured"
}
func bindingCode(m map[string]settings.ResourceBinding, k string) string {
if _, ok := m[k]; ok {
return "binding_active"
}
return "binding_missing"
}
func ternaryCode(ok bool, success, failure string) string {
if ok {
return success
}
return failure
}
func itoa(n int) string {
if n == 0 {
return "0"
}
b := make([]byte, 0, 8)
for n > 0 {
b = append([]byte{byte('0' + n%10)}, b...)
n /= 10
}
return string(b)
}
+210
View File
@@ -0,0 +1,210 @@
package adminweb
import (
"context"
"errors"
"testing"
"time"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/dingtalkconfig"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/secretstore"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/settings"
)
type controlRepo struct {
savedVerified bool
savedRef string
candidateRef string
failedCode string
admins []settings.Administrator
disabledStaff string
disableErr error
}
func (*controlRepo) ActiveDingTalkConfig(context.Context, string) (string, string, bool, error) {
return "old", "old-ref", true, nil
}
func (r *controlRepo) CandidateDingTalkConfig(context.Context, string) (string, string, error) {
return "new", r.candidateRef, nil
}
func (r *controlRepo) SaveDingTalkConfig(_ context.Context, _, _, ref string, verified bool) error {
r.savedRef = ref
r.savedVerified = verified
r.candidateRef = ref
return nil
}
func (r *controlRepo) MarkDingTalkCandidateFailed(_ context.Context, _, _, code string) error {
r.failedCode = code
return nil
}
func (r *controlRepo) AdministratorCount(context.Context, string) (int, error) {
if r.admins == nil {
return 1, nil
}
n := 0
for _, admin := range r.admins {
if admin.Status == "active" {
n++
}
}
return n, nil
}
func (r *controlRepo) ListAdministrators(context.Context, string) ([]settings.Administrator, error) {
return r.admins, nil
}
func (r *controlRepo) DisableAdministrator(_ context.Context, _, staffID string) error {
r.disabledStaff = staffID
return r.disableErr
}
func (*controlRepo) ActiveBindings(context.Context, string) (map[string]settings.ResourceBinding, error) {
return map[string]settings.ResourceBinding{}, nil
}
func (*controlRepo) CreateBootstrapChallenge(context.Context, string, string, time.Time) error {
return nil
}
func (*controlRepo) ConfirmBootstrap(context.Context, string) error { return nil }
type verifyOK struct{}
func (verifyOK) Verify(context.Context, string, string) dingtalkconfig.Result {
return dingtalkconfig.Result{OK: true, Code: "oauth_ok"}
}
type activatorStub struct {
err error
calls int
}
func (a *activatorStub) ActivateCandidate(context.Context) error { a.calls++; return a.err }
type operationalStatusStub struct {
components []Component
err error
}
func (s operationalStatusStub) Components(context.Context) ([]Component, error) {
return s.components, s.err
}
func TestSnapshotIncludesOperationalComponents(t *testing.T) {
secrets, err := secretstore.New(t.TempDir())
if err != nil {
t.Fatal(err)
}
service, err := NewControlService("tenant", &controlRepo{}, secrets, verifyOK{})
if err != nil {
t.Fatal(err)
}
checked := time.Unix(100, 0)
service.SetOperationalStatusProvider(operationalStatusStub{components: []Component{{
Name: "WeKnora", State: "ok", Code: "health_ok", CheckedAt: checked,
}}})
snapshot, err := service.Snapshot(context.Background())
if err != nil {
t.Fatal(err)
}
found := false
for _, component := range snapshot.Components {
if component.Name == "WeKnora" && component.Code == "health_ok" && component.CheckedAt.Equal(checked) {
found = true
}
}
if !found {
t.Fatalf("components=%+v", snapshot.Components)
}
}
func TestSnapshotKeepsPageAvailableWhenOperationalProbeFails(t *testing.T) {
secrets, _ := secretstore.New(t.TempDir())
service, err := NewControlService("tenant", &controlRepo{}, secrets, verifyOK{})
if err != nil {
t.Fatal(err)
}
service.SetOperationalStatusProvider(operationalStatusStub{err: errors.New("probe failed")})
snapshot, err := service.Snapshot(context.Background())
if err != nil {
t.Fatal(err)
}
last := snapshot.Components[len(snapshot.Components)-1]
if last.Name != "运行探针" || last.State != "error" || last.Code != "component_probe_failed" {
t.Fatalf("component=%+v", last)
}
}
func TestSaveCredentialsCreatesCandidateNotActive(t *testing.T) {
secrets, err := secretstore.New(t.TempDir())
if err != nil {
t.Fatal(err)
}
repo := &controlRepo{}
service, err := NewControlService("tenant", repo, secrets, verifyOK{})
if err != nil {
t.Fatal(err)
}
if err = service.SaveCredentials(context.Background(), "new", "secret"); err != nil {
t.Fatal(err)
}
if repo.savedVerified {
t.Fatal("candidate must not be active")
}
if repo.savedRef == "" || !secrets.Configured(repo.savedRef) {
t.Fatalf("ref=%q", repo.savedRef)
}
}
func TestActivateCandidateDelegatesAndMarksFailure(t *testing.T) {
secrets, err := secretstore.New(t.TempDir())
if err != nil {
t.Fatal(err)
}
repo := &controlRepo{candidateRef: "candidate-ref"}
activator := &activatorStub{err: errors.New("candidate_start_failed_rolled_back")}
service, err := NewControlService("tenant", repo, secrets, verifyOK{})
if err != nil {
t.Fatal(err)
}
service.SetCandidateActivator(activator)
if err = service.ActivateCandidate(context.Background()); err == nil {
t.Fatal("expected failure")
}
if activator.calls != 1 || repo.failedCode != "activation_failed" {
t.Fatalf("calls=%d code=%q", activator.calls, repo.failedCode)
}
}
func TestSnapshotIncludesAdministratorsAndOnlyAllowsDisableWhenAnotherIsActive(t *testing.T) {
secrets, _ := secretstore.New(t.TempDir())
repo := &controlRepo{admins: []settings.Administrator{
{StaffID: "staff-old", DisplayName: "旧管理员", Department: "技术部", Role: "super_admin", Status: "active"},
{StaffID: "staff-new", DisplayName: "新管理员", Department: "产品部", Role: "super_admin", Status: "active"},
{StaffID: "staff-off", DisplayName: "停用管理员", Role: "admin", Status: "disabled"},
}}
service, err := NewControlService("tenant", repo, secrets, verifyOK{})
if err != nil {
t.Fatal(err)
}
snapshot, err := service.Snapshot(context.Background())
if err != nil {
t.Fatal(err)
}
if len(snapshot.Administrators) != 3 || !snapshot.Administrators[0].CanDisable || !snapshot.Administrators[1].CanDisable || snapshot.Administrators[2].CanDisable {
t.Fatalf("administrators=%+v", snapshot.Administrators)
}
if snapshot.Administrators[0].StaffIDMasked == "staff-old" || snapshot.Administrators[0].StaffIDMasked == "" {
t.Fatalf("staff ID must be masked: %+v", snapshot.Administrators[0])
}
}
func TestControlServiceDelegatesAdministratorDisable(t *testing.T) {
secrets, _ := secretstore.New(t.TempDir())
repo := &controlRepo{}
service, err := NewControlService("tenant", repo, secrets, verifyOK{})
if err != nil {
t.Fatal(err)
}
if err = service.DisableAdministrator(context.Background(), "staff-old"); err != nil {
t.Fatal(err)
}
if repo.disabledStaff != "staff-old" {
t.Fatalf("disabled=%q", repo.disabledStaff)
}
}
+253
View File
@@ -0,0 +1,253 @@
package adminweb
import (
"context"
"crypto/rand"
"crypto/subtle"
"encoding/hex"
"errors"
"fmt"
"html/template"
"net/http"
"strings"
"time"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/settings"
)
type Snapshot struct {
SystemState, ClientIDMasked string
SecretConfigured bool
Components []Component
Administrators []AdministratorView
}
type AdministratorView struct {
StaffID, StaffIDMasked, DisplayName, Department, Role, Status string
CanDisable bool
}
type Component struct {
Name, State, Code, Detail string
LastSuccess, CheckedAt time.Time
}
type Verification struct {
OK bool
Code string
CheckedAt time.Time
}
type Service interface {
Snapshot(context.Context) (Snapshot, error)
SaveCredentials(context.Context, string, string) error
VerifyCredentials(context.Context, string, string) Verification
}
type BootstrapService interface {
CreateBootstrap(context.Context, string) error
ConfirmBootstrap(context.Context) error
}
type CandidateService interface {
ActivateCandidate(context.Context) error
}
type AdministratorService interface {
DisableAdministrator(context.Context, string) error
}
type server struct {
service Service
token string
csrfToken string
mux *http.ServeMux
page *template.Template
}
type pageData struct {
Snapshot
CSRFToken string
}
const settingsPage = `<!doctype html>
<html lang="zh-CN">
<head>
<meta charset="utf-8"><meta name="viewport" content="width=device-width">
<title>DobotAI 设置</title>
<style>
body{font-family:system-ui,sans-serif;background:#f5f8ff;color:#17324d;max-width:1080px;margin:36px auto;padding:0 20px;line-height:1.55}
.card{background:#fff;border:1px solid #dbe7f4;border-radius:14px;padding:20px;margin:14px 0;box-shadow:0 4px 16px rgba(32,77,120,.05)}
input{box-sizing:border-box;padding:10px;width:100%;margin:6px 0 12px;border:1px solid #bfd0e2;border-radius:8px}
button{padding:10px 18px;background:#1677ff;color:#fff;border:0;border-radius:8px;margin:5px 0;cursor:pointer}.danger{background:#c73535}.muted{color:#55708a}.warn{background:#fff7e6;border-left:4px solid #f0a020;padding:10px 12px;border-radius:6px}.meta{color:#55708a;font-size:13px;line-height:1.6}
table{width:100%;border-collapse:collapse;margin-top:12px}th,td{text-align:left;padding:10px;border-bottom:1px solid #e4edf6;vertical-align:middle}.inline{margin:0}.grid{display:grid;grid-template-columns:1fr 1fr;gap:14px}@media(max-width:760px){.grid{grid-template-columns:1fr}table{font-size:13px}}
</style>
</head>
<body>
<h1>DobotAI 配置中心</h1>
<div class="card"><b>系统状态:</b>{{.SystemState}}<br><b>Client ID:</b>{{.ClientIDMasked}}<br><b>Client Secret:</b>{{if .SecretConfigured}}已配置(不可查看){{else}}未配置{{end}}</div>
<div class="card"><h2>配置/切换钉钉应用</h2><p>新凭据先在线验证并保存为候选。激活后受控重启整组钉钉组件,失败自动回滚。</p><form method="post" action="/admin/setup/credentials"><input type="hidden" name="csrf_token" value="{{.CSRFToken}}"><label>Client ID / AppKey</label><input name="client_id" required><label>Client Secret / AppSecret</label><input type="password" name="client_secret" required autocomplete="new-password"><button>验证并保存候选</button></form><form method="post" action="/admin/setup/candidate/activate"><input type="hidden" name="csrf_token" value="{{.CSRFToken}}"><button>激活候选并重启</button></form></div>
<div class="card"><h2>新增/更换管理员</h2><p class="warn"><b>必须先加后停:</b>先确认新管理员已生效,再停用旧管理员。系统禁止停用最后一位有效管理员。</p><ol><li>输入任意6位一次性绑定码并生成,有效期10分钟。</li><li>新管理员私聊机器人发送 <code>/setting 6位码</code>。</li><li>回到本页确认检测到的新管理员。</li><li>在下方确认新管理员状态为active,再停用旧管理员。</li></ol><div class="grid"><form method="post" action="/admin/setup/bootstrap"><input type="hidden" name="csrf_token" value="{{.CSRFToken}}"><label>一次性绑定码</label><input name="code" pattern="[0-9]{6}" placeholder="输入任意6位数字" required><button>生成6位绑定码</button></form><form method="post" action="/admin/setup/bootstrap/confirm"><input type="hidden" name="csrf_token" value="{{.CSRFToken}}"><p class="muted">新管理员私聊验证后再点击:</p><button>确认检测到的新管理员</button></form></div><h3>管理员列表</h3><table><thead><tr><th>姓名</th><th>部门</th><th>Staff ID</th><th>角色</th><th>状态</th><th>操作</th></tr></thead><tbody>{{range .Administrators}}<tr><td>{{if .DisplayName}}{{.DisplayName}}{{else}}—{{end}}</td><td>{{if .Department}}{{.Department}}{{else}}—{{end}}</td><td>{{.StaffIDMasked}}</td><td>{{.Role}}</td><td>{{.Status}}</td><td>{{if .CanDisable}}<form class="inline" method="post" action="/admin/settings/administrators/disable"><input type="hidden" name="csrf_token" value="{{$.CSRFToken}}"><input type="hidden" name="staff_id" value="{{.StaffID}}"><button class="danger" type="submit">停用此管理员</button></form>{{else}}—{{end}}</td></tr>{{else}}<tr><td colspan="6" class="muted">暂无管理员记录</td></tr>{{end}}</tbody></table></div>
<h2>组件检查</h2>{{range .Components}}<div class="card"><b>{{.Name}}</b>:{{.State}}<div class="meta">错误码:{{if .Code}}{{.Code}}{{else}}—{{end}}<br>最后成功:{{timeText .LastSuccess}}<br>最后检查:{{timeText .CheckedAt}}</div><p>{{.Detail}}</p></div>{{end}}
</body></html>`
func New(service Service, token string) http.Handler {
functions := template.FuncMap{"timeText": func(value time.Time) string {
if value.IsZero() {
return "—"
}
return value.UTC().Format("2006-01-02 15:04:05")
}}
csrfBytes := make([]byte, 32)
if _, err := rand.Read(csrfBytes); err != nil {
panic("admin CSRF token unavailable")
}
s := &server{service: service, token: strings.TrimSpace(token), csrfToken: hex.EncodeToString(csrfBytes), mux: http.NewServeMux(), page: template.Must(template.New("settings").Funcs(functions).Parse(settingsPage))}
s.mux.HandleFunc("GET /admin/settings", s.settings)
s.mux.HandleFunc("GET /admin/setup", s.settings)
s.mux.HandleFunc("POST /admin/setup/credentials", s.saveCredentials)
s.mux.HandleFunc("POST /admin/setup/candidate/activate", s.activateCandidate)
s.mux.HandleFunc("POST /admin/setup/bootstrap", s.createBootstrap)
s.mux.HandleFunc("POST /admin/setup/bootstrap/confirm", s.confirmBootstrap)
s.mux.HandleFunc("POST /admin/settings/administrators/disable", s.disableAdministrator)
return s
}
func (s *server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Cache-Control", "no-store")
if !s.authorized(r) {
w.Header().Set("WWW-Authenticate", `Basic realm="DobotAI Admin"`)
http.Error(w, "unauthorized", http.StatusUnauthorized)
return
}
s.mux.ServeHTTP(w, r)
}
func (s *server) authorized(r *http.Request) bool {
if s.token == "" {
return false
}
if _, password, ok := r.BasicAuth(); ok && len(password) == len(s.token) && subtle.ConstantTimeCompare([]byte(password), []byte(s.token)) == 1 {
return true
}
v := strings.TrimPrefix(r.Header.Get("Authorization"), "Bearer ")
return len(v) == len(s.token) && subtle.ConstantTimeCompare([]byte(v), []byte(s.token)) == 1
}
func (s *server) requireCSRFForm(w http.ResponseWriter, r *http.Request, maxBytes int64) bool {
r.Body = http.MaxBytesReader(w, r.Body, maxBytes)
if err := r.ParseForm(); err != nil {
http.Error(w, "invalid form", http.StatusBadRequest)
return false
}
provided := strings.TrimSpace(r.FormValue("csrf_token"))
if len(provided) != len(s.csrfToken) || subtle.ConstantTimeCompare([]byte(provided), []byte(s.csrfToken)) != 1 {
http.Error(w, "forbidden", http.StatusForbidden)
return false
}
return true
}
func (s *server) settings(w http.ResponseWriter, r *http.Request) {
snap, err := s.service.Snapshot(r.Context())
if err != nil {
http.Error(w, "settings unavailable", 500)
return
}
w.Header().Set("Content-Type", "text/html; charset=utf-8")
_ = s.page.Execute(w, pageData{Snapshot: snap, CSRFToken: s.csrfToken})
}
func (s *server) saveCredentials(w http.ResponseWriter, r *http.Request) {
if !s.requireCSRFForm(w, r, 16<<10) {
return
}
id, secret := strings.TrimSpace(r.FormValue("client_id")), strings.TrimSpace(r.FormValue("client_secret"))
result := s.service.VerifyCredentials(r.Context(), id, secret)
if !result.OK {
http.Error(w, fmt.Sprintf("credential verification failed: %s", result.Code), 422)
return
}
if err := s.service.SaveCredentials(r.Context(), id, secret); err != nil {
http.Error(w, "credential save failed", 500)
return
}
http.Redirect(w, r, "/admin/settings", http.StatusSeeOther)
}
func (s *server) activateCandidate(w http.ResponseWriter, r *http.Request) {
if !s.requireCSRFForm(w, r, 4<<10) {
return
}
service, ok := s.service.(CandidateService)
if !ok {
http.Error(w, "candidate unavailable", http.StatusNotImplemented)
return
}
if err := service.ActivateCandidate(r.Context()); err != nil {
http.Error(w, "candidate activation failed", http.StatusConflict)
return
}
w.Header().Set("Content-Type", "text/plain; charset=utf-8")
_, _ = w.Write([]byte("候选已激活,DobotAI正在受控重启。\n"))
}
func (s *server) createBootstrap(w http.ResponseWriter, r *http.Request) {
if !s.requireCSRFForm(w, r, 4<<10) {
return
}
service, ok := s.service.(BootstrapService)
if !ok {
http.Error(w, "bootstrap unavailable", http.StatusNotImplemented)
return
}
code := strings.TrimSpace(r.FormValue("code"))
if !sixDigits(code) {
http.Error(w, "binding code must be 6 digits", 400)
return
}
if err := service.CreateBootstrap(r.Context(), code); err != nil {
http.Error(w, "create binding code failed", 500)
return
}
w.Header().Set("Content-Type", "text/plain; charset=utf-8")
fmt.Fprintf(w, "请在10分钟内私聊机器人发送:/setting %s\n", code)
}
func (s *server) confirmBootstrap(w http.ResponseWriter, r *http.Request) {
if !s.requireCSRFForm(w, r, 4<<10) {
return
}
service, ok := s.service.(BootstrapService)
if !ok {
http.Error(w, "bootstrap unavailable", http.StatusNotImplemented)
return
}
if err := service.ConfirmBootstrap(r.Context()); err != nil {
http.Error(w, "尚未检测到可确认的管理员", http.StatusConflict)
return
}
http.Redirect(w, r, "/admin/settings", http.StatusSeeOther)
}
func (s *server) disableAdministrator(w http.ResponseWriter, r *http.Request) {
if !s.requireCSRFForm(w, r, 4<<10) {
return
}
service, ok := s.service.(AdministratorService)
if !ok {
http.Error(w, "administrator management unavailable", http.StatusNotImplemented)
return
}
staffID := strings.TrimSpace(r.FormValue("staff_id"))
if staffID == "" {
http.Error(w, "administrator not found", http.StatusNotFound)
return
}
err := service.DisableAdministrator(r.Context(), staffID)
switch {
case err == nil:
http.Redirect(w, r, "/admin/settings", http.StatusSeeOther)
case errors.Is(err, settings.ErrLastAdministrator):
http.Error(w, "至少保留一位有效管理员;请先确认新管理员已生效。", http.StatusConflict)
case errors.Is(err, settings.ErrAdministratorAlreadyDisabled):
http.Error(w, "管理员已经停用。", http.StatusConflict)
case errors.Is(err, settings.ErrAdministratorNotFound):
http.Error(w, "administrator not found", http.StatusNotFound)
default:
http.Error(w, "administrator update failed", http.StatusInternalServerError)
}
}
func sixDigits(value string) bool {
if len(value) != 6 {
return false
}
for _, ch := range value {
if ch < '0' || ch > '9' {
return false
}
}
return true
}
+166
View File
@@ -0,0 +1,166 @@
package adminweb
import (
"context"
"net/http"
"net/http/httptest"
"strings"
"testing"
"time"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/settings"
)
type fakeService struct {
snapshot Snapshot
verified bool
saved bool
disabled string
disableErr error
}
func (f *fakeService) Snapshot(context.Context) (Snapshot, error) { return f.snapshot, nil }
func (f *fakeService) SaveCredentials(context.Context, string, string) error {
f.saved = true
return nil
}
func (f *fakeService) VerifyCredentials(context.Context, string, string) Verification {
f.verified = true
return Verification{OK: true, Code: "ok", CheckedAt: time.Unix(1, 0)}
}
func (f *fakeService) DisableAdministrator(_ context.Context, staffID string) error {
f.disabled = staffID
return f.disableErr
}
func csrfTokenFromPage(t *testing.T, h http.Handler) string {
t.Helper()
get := httptest.NewRequest(http.MethodGet, "/admin/settings", nil)
get.Header.Set("Authorization", "Bearer admin-token")
page := httptest.NewRecorder()
h.ServeHTTP(page, get)
marker := `name="csrf_token" value="`
start := strings.Index(page.Body.String(), marker)
if start < 0 {
t.Fatalf("csrf token missing: %s", page.Body.String())
}
start += len(marker)
end := strings.Index(page.Body.String()[start:], `"`)
if end < 0 {
t.Fatal("csrf token terminator missing")
}
return page.Body.String()[start : start+end]
}
func TestDashboardNeverRendersSecrets(t *testing.T) {
svc := &fakeService{snapshot: Snapshot{SystemState: "active", ClientIDMasked: "ding***", SecretConfigured: true}}
h := New(svc, "admin-token")
req := httptest.NewRequest(http.MethodGet, "/admin/settings", nil)
req.Header.Set("Authorization", "Bearer admin-token")
w := httptest.NewRecorder()
h.ServeHTTP(w, req)
if w.Code != 200 {
t.Fatalf("status=%d", w.Code)
}
body := w.Body.String()
if strings.Contains(body, "admin-token") || strings.Contains(strings.ToLower(body), "client secret value") {
t.Fatal("secret leaked")
}
if !strings.Contains(body, "ding***") {
t.Fatal("masked client id missing")
}
}
func TestDashboardRendersOperationalComponentFields(t *testing.T) {
checked := time.Date(2026, 8, 13, 18, 0, 0, 0, time.UTC)
svc := &fakeService{snapshot: Snapshot{Components: []Component{{
Name: "WeKnora", State: "ok", Code: "health_ok", Detail: "HTTP健康探针正常",
LastSuccess: checked.Add(-time.Minute), CheckedAt: checked,
}}}}
h := New(svc, "admin-token")
req := httptest.NewRequest(http.MethodGet, "/admin/settings", nil)
req.Header.Set("Authorization", "Bearer admin-token")
w := httptest.NewRecorder()
h.ServeHTTP(w, req)
body := w.Body.String()
for _, want := range []string{"WeKnora", "health_ok", "HTTP健康探针正常", "2026-08-13 17:59:00", "2026-08-13 18:00:00"} {
if !strings.Contains(body, want) {
t.Fatalf("dashboard missing %q: %s", want, body)
}
}
}
func TestSaveCredentialsRequiresVerification(t *testing.T) {
svc := &fakeService{}
h := New(svc, "admin-token")
csrf := csrfTokenFromPage(t, h)
req := httptest.NewRequest(http.MethodPost, "/admin/setup/credentials", strings.NewReader("client_id=id&client_secret=secret&csrf_token="+csrf))
req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
req.Header.Set("Authorization", "Bearer admin-token")
w := httptest.NewRecorder()
h.ServeHTTP(w, req)
if w.Code != http.StatusSeeOther || !svc.verified || !svc.saved {
t.Fatalf("status=%d verified=%v saved=%v", w.Code, svc.verified, svc.saved)
}
}
func TestAllAdminPOSTRoutesRejectMissingCSRF(t *testing.T) {
svc := &fakeService{}
h := New(svc, "admin-token")
for _, route := range []string{"/admin/setup/credentials", "/admin/setup/candidate/activate", "/admin/setup/bootstrap", "/admin/setup/bootstrap/confirm", "/admin/settings/administrators/disable"} {
req := httptest.NewRequest(http.MethodPost, route, strings.NewReader("client_id=id&client_secret=secret&code=123456&staff_id=staff-old"))
req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
req.Header.Set("Authorization", "Bearer admin-token")
w := httptest.NewRecorder()
h.ServeHTTP(w, req)
if w.Code != http.StatusForbidden {
t.Fatalf("route=%s status=%d body=%q", route, w.Code, w.Body.String())
}
}
if svc.saved || svc.disabled != "" {
t.Fatalf("business service called saved=%v disabled=%q", svc.saved, svc.disabled)
}
}
func TestDashboardRendersAdministratorReplacementWorkflow(t *testing.T) {
svc := &fakeService{snapshot: Snapshot{Administrators: []AdministratorView{{StaffID: "staff-old", StaffIDMasked: "sta***old", DisplayName: "旧管理员", Department: "技术部", Role: "super_admin", Status: "active", CanDisable: true}}}}
h := New(svc, "admin-token")
req := httptest.NewRequest(http.MethodGet, "/admin/settings", nil)
req.Header.Set("Authorization", "Bearer admin-token")
w := httptest.NewRecorder()
h.ServeHTTP(w, req)
for _, want := range []string{"新增/更换管理员", "先加后停", "旧管理员", "技术部", "sta***old", "生成6位绑定码", "停用此管理员"} {
if !strings.Contains(w.Body.String(), want) {
t.Fatalf("dashboard missing %q: %s", want, w.Body.String())
}
}
if strings.Contains(w.Body.String(), ">staff-old<") {
t.Fatalf("raw staff ID must not be rendered as visible text: %s", w.Body.String())
}
}
func TestDisableAdministratorRequiresCSRFAndMapsLastAdminConflict(t *testing.T) {
svc := &fakeService{}
h := New(svc, "admin-token")
post := func(body string) *httptest.ResponseRecorder {
req := httptest.NewRequest(http.MethodPost, "/admin/settings/administrators/disable", strings.NewReader(body))
req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
req.Header.Set("Authorization", "Bearer admin-token")
w := httptest.NewRecorder()
h.ServeHTTP(w, req)
return w
}
if w := post("staff_id=staff-old"); w.Code != http.StatusForbidden || svc.disabled != "" {
t.Fatalf("missing csrf status=%d disabled=%q", w.Code, svc.disabled)
}
csrf := csrfTokenFromPage(t, h)
svc.disableErr = settings.ErrLastAdministrator
w := post("staff_id=staff-old&csrf_token=" + csrf)
if w.Code != http.StatusConflict || !strings.Contains(w.Body.String(), "至少保留一位有效管理员") {
t.Fatalf("last-admin status=%d body=%q", w.Code, w.Body.String())
}
svc.disableErr = nil
w = post("staff_id=staff-old&csrf_token=" + csrf)
if w.Code != http.StatusSeeOther || svc.disabled != "staff-old" {
t.Fatalf("success status=%d disabled=%q body=%q", w.Code, svc.disabled, w.Body.String())
}
}
+24 -5
View File
@@ -305,9 +305,25 @@ func (s *CardSender) postCard(ctx context.Context, body map[string]any, token st
if err != nil {
return err
}
// Discard successful response body
if len(payload) > 0 {
// Body is already bounded by postJSON→readBoundedAPIResponse; just discard
return validateDingTalkBusinessResponse(payload)
}
func validateDingTalkBusinessResponse(payload []byte) error {
if len(bytes.TrimSpace(payload)) == 0 {
return nil
}
var response struct {
Success *bool `json:"success"`
ErrCode any `json:"errcode"`
}
if json.Unmarshal(payload, &response) != nil {
return nil
}
if response.Success != nil && !*response.Success {
return errors.New("dingtalk card: API business failure")
}
if response.ErrCode != nil && fmt.Sprint(response.ErrCode) != "0" {
return errors.New("dingtalk card: API business failure")
}
return nil
}
@@ -326,8 +342,11 @@ func (s *CardSender) finalizeCard(ctx context.Context, outTrackID, content, toke
"isFinalize": isFinalize,
"isError": isError,
}
_, err = s.doJSON(ctx, http.MethodPut, s.streamURL, body, token)
return err
payload, err := s.doJSON(ctx, http.MethodPut, s.streamURL, body, token)
if err != nil {
return err
}
return validateDingTalkBusinessResponse(payload)
}
func (s *CardSender) postJSON(ctx context.Context, endpoint string, value any, token string) ([]byte, error) {
+16 -1
View File
@@ -35,6 +35,7 @@ type cardTestServer struct {
createBodySize int64 // Content-Length for oversized response
streamStatus int
streamBodySize int64
streamResponse any
}
func newCardTestServer() *cardTestServer {
@@ -42,6 +43,7 @@ func newCardTestServer() *cardTestServer {
tokenStatus: http.StatusOK,
createStatus: http.StatusOK,
streamStatus: http.StatusOK,
streamResponse: map[string]any{"result": "ok"},
tokenResponse: map[string]any{"accessToken": "test-access-token", "expireIn": 3600},
createResponse: map[string]any{"result": "ok"},
}
@@ -91,7 +93,7 @@ func newCardTestServer() *cardTestServer {
return
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(map[string]any{"result": "ok"})
json.NewEncoder(w).Encode(ts.streamResponse)
default:
http.NotFound(w, r)
}
@@ -969,6 +971,19 @@ func TestSendFinalCardCreateStreamOnce(t *testing.T) {
// Test: UpdateCard with non-final / error flags
// ---------------------------------------------------------------------------
func TestUpdateCardRejectsHTTP200BusinessFailure(t *testing.T) {
ts := newCardTestServer()
defer ts.Close()
ts.streamResponse = map[string]any{"success": false, "code": "InvalidParameter", "message": "sensitive detail"}
sender, err := newCardSenderForTest("client-id", "client-secret", "card-template-123", ts.Client(), ts.URL()+"/token", ts.URL()+"/create")
if err != nil {
t.Fatal(err)
}
if err := sender.UpdateCard(context.Background(), "track-1", "answer", true, false); err == nil {
t.Fatal("HTTP 200 business failure must be rejected")
}
}
func TestUpdateCardNonFinalAndError(t *testing.T) {
ts := newCardTestServer()
defer ts.Close()
@@ -187,6 +187,54 @@ func (s *SupportGroupSender) Send(ctx context.Context, notif coverage.SupportNot
return nil
}
// SendTest sends a fixed, non-sensitive message used to verify that the
// configured robot can reach the candidate support group. It deliberately
// does not require an FAQ-table binding.
func (s *SupportGroupSender) SendTest(ctx context.Context) error {
msgParam, err := json.Marshal(map[string]string{
"title": "DobotAI 技术支持群绑定测试",
"text": "DobotAI 技术支持群绑定测试成功。",
})
if err != nil {
return errors.New("dingtalk group sender: encode test message failed")
}
body := map[string]any{
"msgKey": "sampleMarkdown",
"msgParam": string(msgParam),
"openConversationId": s.openConversation,
"robotCode": s.clientID,
}
bodyJSON, err := json.Marshal(body)
if err != nil {
return errors.New("dingtalk group sender: encode test request failed")
}
if len(bodyJSON) > maxGroupPayloadBytes {
return fmt.Errorf("dingtalk group sender: payload exceeds %d bytes", maxGroupPayloadBytes)
}
token, err := s.accessToken(ctx)
if err != nil {
return err
}
respPayload, err := s.doJSON(ctx, http.MethodPost, s.groupURL, body, token)
if err != nil {
return err
}
var response struct {
ProcessQueryKey string `json:"processQueryKey"`
ErrCode int `json:"errcode"`
}
if json.Unmarshal(respPayload, &response) != nil {
return errors.New("dingtalk group sender: invalid API response")
}
if response.ErrCode != 0 {
return errors.New("dingtalk group sender: API returned business error")
}
if strings.TrimSpace(response.ProcessQueryKey) == "" {
return errors.New("dingtalk group sender: incomplete API response")
}
return nil
}
// renderMarkdown produces the compact customer-question context shown in the action card.
func (s *SupportGroupSender) renderMarkdown(notif coverage.SupportNotification) (title, text string) {
title = "知识库未覆盖提醒"
@@ -539,6 +539,38 @@ func TestSupportGroupSenderNoPreviousQuestions(t *testing.T) {
}
}
func TestSupportGroupSenderSendTestDoesNotRequireFAQBinding(t *testing.T) {
gs := newGroupTestServer()
defer gs.Close()
sender, err := newSupportGroupSenderForTest(
"appKey", "appSecret", "cid-abc",
gs.server.Client(), gs.URL()+"/token", gs.URL()+"/groupMessages/send",
)
if err != nil {
t.Fatal(err)
}
if err = sender.SendTest(context.Background()); err != nil {
t.Fatalf("SendTest: %v", err)
}
if gs.sendCalls != 1 {
t.Fatalf("sendCalls=%d want=1", gs.sendCalls)
}
if gs.lastSendBody["openConversationId"] != "cid-abc" || gs.lastSendBody["robotCode"] != "appKey" {
t.Fatalf("body=%v", gs.lastSendBody)
}
if gs.lastSendBody["msgKey"] != "sampleMarkdown" {
t.Fatalf("msgKey=%v want=sampleMarkdown", gs.lastSendBody["msgKey"])
}
var params map[string]string
if err = json.Unmarshal([]byte(gs.lastSendBody["msgParam"].(string)), &params); err != nil {
t.Fatal(err)
}
if params["title"] != "DobotAI 技术支持群绑定测试" || !strings.Contains(params["text"], "测试成功") {
t.Fatalf("params=%v", params)
}
}
func TestSupportGroupSenderRejectsMalformedOrIncompleteSuccessResponse(t *testing.T) {
tests := []struct {
name string
+140 -7
View File
@@ -79,6 +79,31 @@ type UserDirectory interface {
ResolveUser(ctx context.Context, tenantKey, staffID, fallbackName string) (directory.Profile, error)
}
type OutboundContentRewriter interface {
Rewrite(ctx context.Context, content string) string
}
type SettingsMessage struct {
TenantKey string
Channel string
StaffID string
UnionID string
DisplayName string
Department string
Text string
ConversationType string
ConversationID string
ConversationTitle string
}
type SettingsRouter interface {
Handle(context.Context, SettingsMessage) (handled bool, reply string, err error)
}
type LanguagePreferenceResolver interface {
GetPreference(context.Context, string, string, string) (string, error)
}
// FinalCardSender sends an AI card as the final reply. Implementations must
// not log err, outTrackID, or content at any level.
type FinalCardSender interface {
@@ -93,9 +118,12 @@ type StreamingCardSender interface {
}
type HandlerOptions struct {
ImageDownloader ImageDownloader
CardSender FinalCardSender
Directory UserDirectory
ImageDownloader ImageDownloader
CardSender FinalCardSender
Directory UserDirectory
OutboundRewriter OutboundContentRewriter
SettingsRouter SettingsRouter
LanguageResolver LanguagePreferenceResolver
}
type Handler struct {
@@ -105,6 +133,9 @@ type Handler struct {
downloader ImageDownloader
card FinalCardSender
directory UserDirectory
outboundRewriter OutboundContentRewriter
settingsRouter SettingsRouter
languageResolver LanguagePreferenceResolver
now func() time.Time
cardUpdateInterval time.Duration
finalDeliveryTimeout time.Duration
@@ -146,6 +177,9 @@ func NewHandlerWithOptions(cfg TenantConfig, qa QAAsker, reply MarkdownSender, o
downloader: opts.ImageDownloader,
card: opts.CardSender,
directory: opts.Directory,
outboundRewriter: opts.OutboundRewriter,
settingsRouter: opts.SettingsRouter,
languageResolver: opts.LanguageResolver,
now: time.Now,
cardUpdateInterval: 500 * time.Millisecond,
finalDeliveryTimeout: defaultHTTPTimeout,
@@ -198,6 +232,24 @@ func (h *Handler) Handle(ctx context.Context, cb BotCallback) error {
triggerPerson = senderName
}
if h.settingsRouter != nil && strings.EqualFold(strings.TrimSpace(cb.Msgtype), msgTypeText) {
handled, settingsReply, err := h.settingsRouter.Handle(ctx, SettingsMessage{
TenantKey: h.cfg.TenantKey, Channel: h.cfg.Channel, StaffID: externalUserID,
UnionID: actorUnionID, DisplayName: senderName, Department: department, Text: strings.TrimSpace(cb.Text),
ConversationType: mapDingTalkConversationType(cb.ConversationType),
ConversationID: strings.TrimSpace(cb.ConversationID), ConversationTitle: strings.TrimSpace(cb.ConversationTitle),
})
if err != nil {
return fmt.Errorf("dingtalk settings: %w", err)
}
if handled {
if strings.TrimSpace(settingsReply) == "" {
return nil
}
return h.sendReply(ctx, webhookURL, settingsReply)
}
}
req := core.QARequest{
TenantKey: h.cfg.TenantKey,
Channel: h.cfg.Channel,
@@ -211,6 +263,11 @@ func (h *Handler) Handle(ctx context.Context, cb BotCallback) error {
MessageType: strings.ToLower(strings.TrimSpace(cb.Msgtype)),
ConversationTitle: strings.TrimSpace(cb.ConversationTitle),
}
if h.languageResolver != nil {
if locale, err := h.languageResolver.GetPreference(ctx, h.cfg.TenantKey, h.cfg.Channel, externalUserID); err == nil {
req.ResponseLocale = strings.TrimSpace(locale)
}
}
switch cb.ConversationType {
case conversationTypeGroup:
if strings.TrimSpace(cb.ConversationID) == "" {
@@ -263,6 +320,13 @@ func (h *Handler) Handle(ctx context.Context, cb BotCallback) error {
return h.handleNonStreaming(ctx, webhookURL, req)
}
func mapDingTalkConversationType(value string) string {
if value == conversationTypeGroup {
return "group"
}
return "direct"
}
func (h *Handler) eventStreamingCapable() (StreamingCardSender, EventQAAsker) {
sc, ok := h.card.(StreamingCardSender)
if !ok {
@@ -290,6 +354,7 @@ func (h *Handler) streamingCapable() (StreamingCardSender, StreamingQAAsker) {
func (h *Handler) handleNonStreaming(ctx context.Context, webhookURL string, req core.QARequest) error {
result, err := h.qa.Ask(ctx, req)
if err != nil {
logQAFailure("non_streaming", err)
if replyErr := h.sendReply(ctx, webhookURL, qaErrorReply); replyErr != nil {
return replyErr
}
@@ -298,13 +363,14 @@ func (h *Handler) handleNonStreaming(ctx context.Context, webhookURL string, req
if result.Replayed && result.DeliveryStatus == core.DeliverySent {
return nil
}
content := h.rewriteOutboundContent(ctx, result.Answer)
if h.card != nil {
cardReq := CardRequest{
ConversationType: req.ConversationType,
ExternalUserID: req.ExternalUserID,
ExternalChatID: req.ExternalChatID,
Content: result.Answer,
Content: content,
}
if outTrackID, cardErr := h.card.SendFinalCard(ctx, cardReq); cardErr == nil {
h.recordDelivery(ctx, result, core.DeliverySent, core.DeliveryMethodCard, outTrackID, "")
@@ -312,7 +378,7 @@ func (h *Handler) handleNonStreaming(ctx context.Context, webhookURL string, req
}
log.Printf("DingTalk card delivery failed, falling back to markdown")
}
if err := h.sendReply(ctx, webhookURL, result.Answer); err != nil {
if err := h.sendReply(ctx, webhookURL, content); err != nil {
h.recordDelivery(ctx, result, core.DeliveryFailed, "", "", core.DeliveryFailureChannelDelivery)
return err
}
@@ -333,6 +399,7 @@ func (h *Handler) handleStreaming(ctx context.Context, webhookURL string, req co
// Fallback: Ask → Markdown only (no SendFinalCard, no AskStream)
result, askErr := h.qa.Ask(ctx, req)
if askErr != nil {
logQAFailure("streaming_fallback", askErr)
if replyErr := h.sendReply(ctx, webhookURL, qaErrorReply); replyErr != nil {
return replyErr
}
@@ -341,7 +408,8 @@ func (h *Handler) handleStreaming(ctx context.Context, webhookURL string, req co
if result.Replayed && result.DeliveryStatus == core.DeliverySent {
return nil
}
if replyErr := h.sendReply(ctx, webhookURL, result.Answer); replyErr != nil {
content := h.rewriteOutboundContent(ctx, result.Answer)
if replyErr := h.sendReply(ctx, webhookURL, content); replyErr != nil {
h.recordDelivery(ctx, result, core.DeliveryFailed, "", "", core.DeliveryFailureChannelDelivery)
return replyErr
}
@@ -369,6 +437,7 @@ func (h *Handler) handleStreaming(ctx context.Context, webhookURL string, req co
result, err := streamingQA.AskStream(ctx, req, observer)
if err != nil || strings.TrimSpace(result.Answer) == "" {
logQAFailure("streaming", errOrEmptyAnswer(err, result.Answer))
return h.finalizeCardOrMarkdown(ctx, webhookURL, streamingCard, outTrackID, core.QAResult{}, qaErrorReply, true)
}
return h.finalizeCardOrMarkdown(ctx, webhookURL, streamingCard, outTrackID, result, result.Answer, false)
@@ -386,12 +455,14 @@ func (h *Handler) handleEventStreaming(ctx context.Context, webhookURL string, r
log.Printf("DingTalk event card start failed, falling back to markdown")
result, askErr := h.qa.Ask(ctx, req)
if askErr != nil {
logQAFailure("event_fallback", askErr)
return h.sendReply(ctx, webhookURL, qaErrorReply)
}
if result.Replayed && result.DeliveryStatus == core.DeliverySent {
return nil
}
if replyErr := h.sendReply(ctx, webhookURL, result.Answer); replyErr != nil {
content := h.rewriteOutboundContent(ctx, result.Answer)
if replyErr := h.sendReply(ctx, webhookURL, content); replyErr != nil {
h.recordDelivery(ctx, result, core.DeliveryFailed, "", "", core.DeliveryFailureChannelDelivery)
return replyErr
}
@@ -421,6 +492,7 @@ func (h *Handler) handleEventStreaming(ctx context.Context, webhookURL string, r
pump.Stop()
if err != nil || strings.TrimSpace(result.Answer) == "" {
logQAFailure("event_streaming", errOrEmptyAnswer(err, result.Answer))
return h.finalizeCardOrMarkdown(ctx, webhookURL, streamingCard, outTrackID, core.QAResult{}, qaErrorReply, true)
}
return h.finalizeCardOrMarkdown(ctx, webhookURL, streamingCard, outTrackID, result, result.Answer, false)
@@ -432,6 +504,10 @@ func (h *Handler) finalizeCardOrMarkdown(ctx context.Context, webhookURL string,
timeout = defaultHTTPTimeout
}
if !isError {
content = h.rewriteOutboundContent(ctx, content)
}
// Final card convergence must still be attempted when the QA context was
// canceled or timed out; otherwise the card can remain "processing".
finalCtx, cancelFinal := context.WithTimeout(context.WithoutCancel(ctx), timeout)
@@ -460,6 +536,19 @@ func (h *Handler) finalizeCardOrMarkdown(ctx context.Context, webhookURL string,
return nil
}
func (h *Handler) rewriteOutboundContent(ctx context.Context, content string) string {
if h.outboundRewriter == nil {
return content
}
timeout := h.finalDeliveryTimeout
if timeout <= 0 {
timeout = defaultHTTPTimeout
}
rewriteCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), timeout)
defer cancel()
return h.outboundRewriter.Rewrite(rewriteCtx, content)
}
func (h *Handler) recordDelivery(ctx context.Context, result core.QAResult, status, method, providerMessageID, failureCode string) {
if strings.TrimSpace(result.TurnID) == "" || strings.TrimSpace(result.AnswerMessageID) == "" {
return
@@ -511,6 +600,50 @@ func renderProgressEvent(event weknora.StreamEvent) string {
return ""
}
func errOrEmptyAnswer(err error, answer string) error {
if err != nil {
return err
}
if strings.TrimSpace(answer) == "" {
return weknora.ErrEmptyAnswer
}
return nil
}
func logQAFailure(mode string, err error) {
log.Printf("DingTalk QA failed: mode=%s stage=%s", mode, qaFailureStage(err))
}
func qaFailureStage(err error) string {
switch {
case err == nil:
return "unknown"
case errors.Is(err, context.Canceled):
return "context_canceled"
case errors.Is(err, context.DeadlineExceeded):
return "context_deadline"
case errors.Is(err, core.ErrTurnInProgress):
return "duplicate_or_in_progress"
case errors.Is(err, weknora.ErrEmptyAnswer):
return "empty_answer"
}
message := strings.ToLower(err.Error())
switch {
case strings.Contains(message, "begin turn"):
return "begin_turn"
case strings.Contains(message, "agent") && strings.Contains(message, "binding"):
return "agent_binding"
case strings.Contains(message, "session"):
return "session"
case strings.Contains(message, "weknora agent chat"):
return "agent_chat"
case strings.Contains(message, "required") || strings.Contains(message, "unsupported"):
return "validation"
default:
return "unknown"
}
}
func imageFailureStage(err error) string {
message := strings.ToLower(err.Error())
switch {
+88 -4
View File
@@ -726,6 +726,46 @@ func TestHandleCardSuccessDirectFieldsCorrectNoMarkdown(t *testing.T) {
}
}
func TestHandleNonStreamingRewritesFinalCardAndMarkdownFallback(t *testing.T) {
resource := "resource://AbCdEfGhIjKlMnOpQrStUv"
signed := "https://img.example/signed.jpg"
answer := "说明\n![图](" + resource + ")"
for _, tc := range []struct {
name string
cardErr error
wantReply bool
}{
{name: "card"},
{name: "markdown fallback", cardErr: errors.New("card failed"), wantReply: true},
} {
t.Run(tc.name, func(t *testing.T) {
qa := &fakeQA{result: core.QAResult{Answer: answer}}
reply := &fakeReply{}
card := &fakeCard{returnID: "track", err: tc.cardErr}
h := newTestHandlerWithCard(t, qa, reply, card)
rewriter := &fakeOutboundRewriter{from: resource, to: signed}
h.outboundRewriter = rewriter
if err := h.Handle(context.Background(), directTextCallback()); err != nil {
t.Fatal(err)
}
if !card.called || reply.called != tc.wantReply {
t.Fatalf("card=%t reply=%t", card.called, reply.called)
}
content := card.lastReq.Content
if tc.wantReply {
content = reply.lastText
}
if strings.Contains(content, resource) || !strings.Contains(content, signed) {
t.Fatalf("outbound content=%q", content)
}
if rewriter.calls != 1 {
t.Fatalf("rewrite calls=%d, want 1", rewriter.calls)
}
})
}
}
func TestHandleCardSuccessGroupChatIDCorrect(t *testing.T) {
qa := &fakeQA{result: core.QAResult{Answer: "group card answer"}}
reply := &fakeReply{}
@@ -940,6 +980,17 @@ type cardUpdate struct {
isError bool
}
type fakeOutboundRewriter struct {
calls int
from string
to string
}
func (f *fakeOutboundRewriter) Rewrite(_ context.Context, content string) string {
f.calls++
return strings.ReplaceAll(content, f.from, f.to)
}
func (f *fakeStreamingCard) SendFinalCard(_ context.Context, req CardRequest) (string, error) {
f.finalCalled = true
f.finalReq = req
@@ -1123,11 +1174,14 @@ func TestHandleStreamingThrottleClockAdvances(t *testing.T) {
func TestHandleStreamingStartFailsFallbackAskMarkdown(t *testing.T) {
card := &fakeStreamingCard{startErr: errors.New("card API unavailable")}
resource := "resource://AbCdEfGhIjKlMnOpQrStUv"
qa := &fakeStreamingQA{
result: core.QAResult{Answer: "**fallback markdown**"},
result: core.QAResult{Answer: "**fallback markdown**\n![图](" + resource + ")"},
}
reply := &fakeReply{}
h := newTestHandlerWithStreaming(t, qa, reply, card)
rewriter := &fakeOutboundRewriter{from: resource, to: "https://img.example/signed.jpg"}
h.outboundRewriter = rewriter
err := h.Handle(context.Background(), directTextCallback())
if err != nil {
@@ -1156,8 +1210,11 @@ func TestHandleStreamingStartFailsFallbackAskMarkdown(t *testing.T) {
if !reply.called {
t.Fatal("SendMarkdown must be called on Start failure")
}
if reply.lastText != "**fallback markdown**" {
t.Fatalf("markdown text = %q, want fallback markdown", reply.lastText)
if strings.Contains(reply.lastText, resource) || !strings.Contains(reply.lastText, "https://img.example/signed.jpg") {
t.Fatalf("markdown text = %q, want rewritten fallback markdown", reply.lastText)
}
if rewriter.calls != 1 {
t.Fatalf("rewrite calls=%d, want 1", rewriter.calls)
}
// No card updates
@@ -2005,5 +2062,32 @@ func TestTriggerPersonSnapshotRegression(t *testing.T) {
t.Fatalf("SenderName = %q, want 张三", qa.lastReq.SenderName)
}
})
}
func TestHandleStreamingRewritesOnlyFinalOutboundImageContent(t *testing.T) {
card := &fakeStreamingCard{startID: "track-image-final"}
resource := "resource://AbCdEfGhIjKlMnOpQrStUv"
answer := "操作说明\n![步骤图](" + resource + ")"
qa := &fakeStreamingQA{chunks: []string{answer}, result: core.QAResult{Answer: answer}}
reply := &fakeReply{}
rewriter := &fakeOutboundRewriter{from: resource, to: "https://img.example/signed.jpg"}
h := newTestHandlerWithStreaming(t, qa, reply, card)
h.outboundRewriter = rewriter
if err := h.Handle(context.Background(), directTextCallback()); err != nil {
t.Fatal(err)
}
if len(card.updates) < 2 {
t.Fatalf("updates=%d", len(card.updates))
}
if !strings.Contains(card.updates[0].content, resource) {
t.Fatalf("streaming intermediate content should remain untouched: %q", card.updates[0].content)
}
final := card.updates[len(card.updates)-1]
if !final.isFinalize || strings.Contains(final.content, resource) || !strings.Contains(final.content, "https://img.example/signed.jpg") {
t.Fatalf("final content=%q finalize=%t", final.content, final.isFinalize)
}
if rewriter.calls != 1 {
t.Fatalf("rewrite calls=%d, want 1", rewriter.calls)
}
}
@@ -0,0 +1,55 @@
package dingtalk
import (
"context"
"testing"
)
type fakeSettingsRouter struct {
called bool
handled bool
reply string
}
func (f *fakeSettingsRouter) Handle(_ context.Context, in SettingsMessage) (bool, string, error) {
f.called = true
if in.StaffID != "staff-123" {
panic("wrong identity")
}
return f.handled, f.reply, nil
}
func TestSettingsCommandBypassesQA(t *testing.T) {
qa := &fakeQA{}
reply := &fakeReply{}
router := &fakeSettingsRouter{handled: true, reply: "settings menu"}
h, err := NewHandlerWithOptions(TenantConfig{TenantKey: "tenant-a", Channel: "dingtalk"}, qa, reply, HandlerOptions{SettingsRouter: router})
if err != nil {
t.Fatal(err)
}
cb := directTextCallback()
cb.Text = "/setting"
if err = h.Handle(context.Background(), cb); err != nil {
t.Fatal(err)
}
if !router.called {
t.Fatal("router not called")
}
if qa.called {
t.Fatal("settings must bypass QA")
}
if !reply.called || reply.lastText != "settings menu" {
t.Fatalf("reply=%+v", reply)
}
}
func TestOrdinaryQuestionStillCallsQA(t *testing.T) {
qa := &fakeQA{}
reply := &fakeReply{}
router := &fakeSettingsRouter{}
h, _ := NewHandlerWithOptions(TenantConfig{TenantKey: "tenant-a", Channel: "dingtalk"}, qa, reply, HandlerOptions{SettingsRouter: router})
if err := h.Handle(context.Background(), directTextCallback()); err != nil {
t.Fatal(err)
}
if !qa.called {
t.Fatal("ordinary question must call QA")
}
}
+143 -16
View File
@@ -5,7 +5,11 @@ import (
"encoding/json"
"errors"
"fmt"
"log"
"net/url"
"strings"
"sync"
"time"
sdkcard "github.com/open-dingtalk/dingtalk-stream-sdk-go/card"
"github.com/open-dingtalk/dingtalk-stream-sdk-go/chatbot"
@@ -88,12 +92,85 @@ func BotCallbackFromSDK(data *chatbot.BotCallbackDataModel) (BotCallback, error)
cb.DownloadCode = rtCode
return cb, nil
}
if msgtype == "link" {
linkURL := extractLinkMessageURL(data.Content)
if linkURL == "" {
linkURL = safeHTTPURL(data.Text.Content)
}
if linkURL != "" {
cb.Msgtype = msgTypeText
cb.Text = linkURL
}
return cb, nil
}
if msgtype == "picture" {
cb.DownloadCode = extractDownloadCode(data.Content)
}
return cb, nil
}
// extractLinkMessageURL extracts an HTTP(S) target from DingTalk link-card
// content. Only known URL and envelope keys are traversed so a card image URL
// (for example picUrl) cannot be mistaken for the message target.
func extractLinkMessageURL(raw interface{}) string {
if raw == nil {
return ""
}
b, err := json.Marshal(raw)
if err != nil {
return ""
}
var value interface{}
if json.Unmarshal(b, &value) != nil {
return ""
}
return findLinkMessageURL(value)
}
func findLinkMessageURL(value interface{}) string {
object, ok := value.(map[string]interface{})
if !ok {
return ""
}
for key, candidate := range object {
switch normalizeLinkField(key) {
case "messageurl", "linkurl", "singletitleurl", "href", "url":
if text, ok := candidate.(string); ok {
if target := safeHTTPURL(text); target != "" {
return target
}
}
}
}
for key, nested := range object {
switch normalizeLinkField(key) {
case "content", "link", "data", "card":
if target := findLinkMessageURL(nested); target != "" {
return target
}
}
}
return ""
}
func normalizeLinkField(value string) string {
value = strings.ToLower(strings.TrimSpace(value))
value = strings.ReplaceAll(value, "_", "")
return strings.ReplaceAll(value, "-", "")
}
func safeHTTPURL(raw string) string {
raw = strings.TrimSpace(raw)
u, err := url.ParseRequestURI(raw)
if err != nil || u.Host == "" {
return ""
}
if !strings.EqualFold(u.Scheme, "https") && !strings.EqualFold(u.Scheme, "http") {
return ""
}
return raw
}
// parseRichText parses the SDK Content field for richText messages.
// It handles three content shapes:
// 1. Direct object: {"richText": [...]}
@@ -222,18 +299,77 @@ func extractDownloadCode(raw interface{}) string {
return strings.TrimSpace(nested.Content.DownloadCode)
}
// StreamClient wraps the official DingTalk Stream SDK client. It wires
// robot callbacks into the existing Handler synchronously (no goroutines,
// no queue). The SDK default auto-reconnect stays enabled while running;
// Close disables it so the connection stays closed.
const callbackDedupTTL = 5 * time.Minute
type chatCallbackHandler interface {
Handle(context.Context, BotCallback) error
}
// newAsyncChatCallbackRouter acknowledges valid DingTalk frames immediately,
// then processes them outside the SDK callback lifetime. DingTalk retries slow
// callbacks, so retaining MsgIDs for the platform's five-minute retry window
// prevents a retry from creating a second streaming card.
func newAsyncChatCallbackRouter(handler chatCallbackHandler) func(context.Context, *chatbot.BotCallbackDataModel) ([]byte, error) {
var mu sync.Mutex
seen := make(map[string]time.Time)
return func(ctx context.Context, data *chatbot.BotCallbackDataModel) ([]byte, error) {
cb, err := BotCallbackFromSDK(data)
if err != nil {
return nil, err
}
sourceType := safeCallbackMessageType(data.Msgtype)
if sourceType == "link" || sourceType == "richtext" {
log.Printf("DingTalk callback normalized: source_msgtype=%s normalized_msgtype=%s text_present=%t", sourceType, safeCallbackMessageType(cb.Msgtype), strings.TrimSpace(cb.Text) != "")
}
messageID := strings.TrimSpace(cb.MsgID)
if messageID == "" {
return nil, errors.New("dingtalk stream: msgId is required")
}
now := time.Now()
mu.Lock()
for id, expiresAt := range seen {
if !expiresAt.After(now) {
delete(seen, id)
}
}
if expiresAt, exists := seen[messageID]; exists && expiresAt.After(now) {
mu.Unlock()
return nil, nil
}
seen[messageID] = now.Add(callbackDedupTTL)
mu.Unlock()
workCtx := context.WithoutCancel(ctx)
go func() {
if err := handler.Handle(workCtx, cb); err != nil {
log.Printf("DingTalk async callback failed: stage=handler")
}
}()
return nil, nil
}
}
func safeCallbackMessageType(raw string) string {
switch strings.ToLower(strings.TrimSpace(raw)) {
case "text", "picture", "richtext", "link":
return strings.ToLower(strings.TrimSpace(raw))
default:
return "other"
}
}
// StreamClient wraps the official DingTalk Stream SDK client. Robot callbacks
// are acknowledged before the potentially slow QA/card work begins. The SDK
// default auto-reconnect stays enabled while running; Close disables it so the
// connection stays closed.
type StreamClient struct {
inner *dtsdk.StreamClient
}
// NewStreamClient validates the config, builds the SDK client with the app
// credential and registers the chatbot callback router. The handler is
// invoked synchronously from the SDK callback; its return value only
// acknowledges the frame, replies always go through the session webhook.
// acknowledged immediately; replies always go through the session webhook.
func NewStreamClient(config StreamConfig, handler *Handler, feedbackHandlers ...*CardFeedbackHandler) (*StreamClient, error) {
if err := config.Validate(); err != nil {
return nil, err
@@ -256,16 +392,7 @@ func NewStreamClient(config StreamConfig, handler *Handler, feedbackHandlers ...
inner := dtsdk.NewStreamClient(
dtsdk.WithAppCredential(cred),
)
inner.RegisterChatBotCallbackRouter(func(ctx context.Context, data *chatbot.BotCallbackDataModel) ([]byte, error) {
cb, err := BotCallbackFromSDK(data)
if err != nil {
return nil, err
}
if err := handler.Handle(ctx, cb); err != nil {
return nil, err
}
return nil, nil
})
inner.RegisterChatBotCallbackRouter(newAsyncChatCallbackRouter(handler))
if len(feedbackHandlers) == 1 {
inner.RegisterCardCallbackRouter(func(ctx context.Context, request *sdkcard.CardRequest) (*sdkcard.CardResponse, error) {
return feedbackHandlers[0].HandleSDK(ctx, request)
+107
View File
@@ -1,12 +1,79 @@
package dingtalk
import (
"context"
"strings"
"sync"
"testing"
"time"
"github.com/open-dingtalk/dingtalk-stream-sdk-go/chatbot"
)
type blockingCallbackHandler struct {
started chan struct{}
release chan struct{}
mu sync.Mutex
calls int
}
func (h *blockingCallbackHandler) Handle(_ context.Context, _ BotCallback) error {
h.mu.Lock()
h.calls++
if h.calls == 1 {
close(h.started)
}
h.mu.Unlock()
<-h.release
return nil
}
func (h *blockingCallbackHandler) callCount() int {
h.mu.Lock()
defer h.mu.Unlock()
return h.calls
}
func TestAsyncChatCallbackAcknowledgesBeforeWorkAndDeduplicatesInflightMessage(t *testing.T) {
handler := &blockingCallbackHandler{started: make(chan struct{}), release: make(chan struct{})}
router := newAsyncChatCallbackRouter(handler)
data := &chatbot.BotCallbackDataModel{
ConversationType: "1",
Msgtype: "text",
MsgId: "same-message",
SenderStaffId: "staff-1",
SessionWebhook: "https://oapi.dingtalk.com/robot/sendBySession?session=test",
Text: chatbot.BotCallbackDataTextModel{Content: "hello"},
}
done := make(chan error, 1)
go func() {
_, err := router(context.Background(), data)
done <- err
}()
select {
case err := <-done:
if err != nil {
t.Fatal(err)
}
case <-time.After(100 * time.Millisecond):
t.Fatal("callback ACK waited for slow QA work")
}
select {
case <-handler.started:
case <-time.After(time.Second):
t.Fatal("async handler did not start")
}
if _, err := router(context.Background(), data); err != nil {
t.Fatal(err)
}
time.Sleep(20 * time.Millisecond)
if got := handler.callCount(); got != 1 {
t.Fatalf("same in-flight MsgID handled %d times, want 1", got)
}
close(handler.release)
}
func TestStreamConfigValidate(t *testing.T) {
t.Parallel()
tests := []struct {
@@ -142,6 +209,46 @@ func TestBotCallbackFromSDK(t *testing.T) {
})
}
func TestBotCallbackFromSDKLinkCardPromotesTrustedURLToText(t *testing.T) {
t.Parallel()
const notableURL = "https://alidocs.dingtalk.com/i/nodes/base?sheetId=sheet&viewId=view"
tests := []struct {
name string
content interface{}
}{
{name: "direct content", content: map[string]interface{}{"title": "FAQ表", "messageUrl": notableURL}},
{name: "nested callback envelope", content: map[string]interface{}{"content": map[string]interface{}{"messageUrl": notableURL}}},
}
for _, tt := range tests {
tt := tt
t.Run(tt.name, func(t *testing.T) {
t.Parallel()
cb, err := BotCallbackFromSDK(&chatbot.BotCallbackDataModel{Msgtype: "link", MsgId: "link-1", Content: tt.content})
if err != nil {
t.Fatal(err)
}
if cb.Msgtype != "text" || cb.Text != notableURL {
t.Fatalf("callback=%+v, want promoted text URL", cb)
}
})
}
}
func TestBotCallbackFromSDKLinkCardRejectsUnsafeURL(t *testing.T) {
t.Parallel()
cb, err := BotCallbackFromSDK(&chatbot.BotCallbackDataModel{
Msgtype: "link",
MsgId: "link-unsafe",
Content: map[string]interface{}{"messageUrl": "javascript:alert(1)"},
})
if err != nil {
t.Fatal(err)
}
if cb.Msgtype != "link" || cb.Text != "" {
t.Fatalf("unsafe link must not be promoted: %+v", cb)
}
}
func TestBotCallbackFromSDKPicture(t *testing.T) {
t.Parallel()
+1
View File
@@ -49,6 +49,7 @@ type QARequest struct {
ActorUnionIDSnapshot string
Department string
Region string
ResponseLocale string
Query string
Images []ImageInput
MessageType string
+1
View File
@@ -16,6 +16,7 @@ var noAnswerPhrases = []string{
"暂无相关",
"暂无已发布的权威知识",
"知识库中没有",
"产品不存在于资料中",
}
// retrievalGapPhrases are only considered a no-answer signal when the answer
+20
View File
@@ -62,6 +62,26 @@ func TestEvaluator_UncoveredExplicitNoAnswer(t *testing.T) {
}
}
func TestEvaluator_ProductionMissingProductPhraseTrumpsReferences(t *testing.T) {
e := NewEvaluator()
turn := core.Turn{
ID: "missing-product-turn",
Status: "succeeded",
Answer: "根据提供的检索信息,无法提供对应手册。原因:产品不存在于资料中;检索资料未提及该产品。建议联系技术支持部门。",
References: []core.Reference{
{ID: "unrelated-1", Score: 0.395940},
{ID: "unrelated-2", Score: 0.311595},
},
}
d, err := e.Evaluate(turn)
if err != nil {
t.Fatal(err)
}
if d.Verdict != VerdictUncovered || d.ReasonCode != ReasonUncoveredExplicitNoAnswer {
t.Fatalf("verdict=%s reason=%s", d.Verdict, d.ReasonCode)
}
}
func TestEvaluator_UncoveredNoReferences(t *testing.T) {
e := NewEvaluator()
turn := core.Turn{
+69
View File
@@ -0,0 +1,69 @@
package dingtalkconfig
import (
"context"
"encoding/json"
"io"
"net/http"
"net/url"
"strings"
"time"
)
type Result struct {
OK bool
Code string
CheckedAt time.Time
}
type Verifier struct {
endpoint string
client *http.Client
}
func NewVerifier(endpoint string, client *http.Client) *Verifier {
if client == nil {
client = &http.Client{Timeout: 10 * time.Second}
}
return &Verifier{endpoint: strings.TrimSpace(endpoint), client: client}
}
func (v *Verifier) Verify(ctx context.Context, id, secret string) Result {
r := Result{Code: "verification_failed", CheckedAt: time.Now().UTC()}
if strings.TrimSpace(id) == "" || strings.TrimSpace(secret) == "" {
r.Code = "credential_incomplete"
return r
}
u, err := url.Parse(v.endpoint)
if err != nil {
r.Code = "endpoint_invalid"
return r
}
q := u.Query()
q.Set("appkey", id)
q.Set("appsecret", secret)
u.RawQuery = q.Encode()
req, err := http.NewRequestWithContext(ctx, http.MethodGet, u.String(), nil)
if err != nil {
return r
}
resp, err := v.client.Do(req)
if err != nil {
r.Code = "network_error"
return r
}
defer resp.Body.Close()
var body struct {
ErrCode int `json:"errcode"`
AccessToken string `json:"access_token"`
}
if json.NewDecoder(io.LimitReader(resp.Body, 1<<20)).Decode(&body) != nil {
r.Code = "invalid_response"
return r
}
if resp.StatusCode < 200 || resp.StatusCode >= 300 || body.ErrCode != 0 || strings.TrimSpace(body.AccessToken) == "" {
r.Code = "credential_rejected"
return r
}
r.OK = true
r.Code = "ok"
return r
}
+35
View File
@@ -0,0 +1,35 @@
package dingtalkconfig
import (
"context"
"net/http"
"net/http/httptest"
"strings"
"testing"
)
func TestVerifierNeverReturnsToken(t *testing.T) {
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write([]byte(`{"errcode":0,"access_token":"secret-token"}`))
}))
defer ts.Close()
v := NewVerifier(ts.URL, ts.Client())
result := v.Verify(context.Background(), "id", "secret")
if !result.OK {
t.Fatalf("result=%+v", result)
}
if strings.Contains(result.Code, "secret-token") {
t.Fatal("token leaked")
}
}
func TestVerifierRejectsBadCredential(t *testing.T) {
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
_, _ = w.Write([]byte(`{"errcode":40014,"errmsg":"bad secret"}`))
}))
defer ts.Close()
result := NewVerifier(ts.URL, ts.Client()).Verify(context.Background(), "id", "wrong")
if result.OK || result.Code != "credential_rejected" {
t.Fatalf("result=%+v", result)
}
}
+26 -1
View File
@@ -25,9 +25,16 @@ type AuditService interface {
GetTurn(context.Context, string) (audit.TurnDetail, error)
}
type BuildInfo struct {
Version string `json:"version"`
CommitSHA string `json:"commit_sha"`
BuildTime string `json:"build_time"`
}
type Options struct {
AuditService AuditService
AdminToken string
BuildInfo BuildInfo
}
type Server struct {
@@ -35,6 +42,7 @@ type Server struct {
ingressToken string
auditService AuditService
adminToken string
buildInfo BuildInfo
mux *http.ServeMux
}
@@ -58,13 +66,25 @@ func NewWithOptions(service QAService, ingressToken string, opts Options) (*Serv
if hasAdminToken && constantTimeTokenEqual(adminToken, ingressToken) {
return nil, errors.New("DOBOTAI_ADMIN_TOKEN must differ from DOBOTAI_INGRESS_TOKEN")
}
buildInfo := opts.BuildInfo
if buildInfo.Version == "" {
buildInfo.Version = "dev"
}
if buildInfo.CommitSHA == "" {
buildInfo.CommitSHA = "unknown"
}
if buildInfo.BuildTime == "" {
buildInfo.BuildTime = "unknown"
}
s := &Server{
service: service, ingressToken: ingressToken,
auditService: opts.AuditService, adminToken: adminToken,
mux: http.NewServeMux(),
buildInfo: buildInfo,
mux: http.NewServeMux(),
}
s.mux.HandleFunc("GET /healthz", s.health)
s.mux.HandleFunc("GET /version", s.version)
s.mux.HandleFunc("POST /v1/qa", s.authorizeToken(s.ingressToken, s.ask))
if hasAuditService {
s.mux.HandleFunc("GET /v1/admin/audit/turns", s.authorizeToken(s.adminToken, s.listAuditTurns))
@@ -102,6 +122,11 @@ func (s *Server) health(w http.ResponseWriter, _ *http.Request) {
writeJSON(w, http.StatusOK, map[string]string{"status": "ok"})
}
func (s *Server) version(w http.ResponseWriter, _ *http.Request) {
w.Header().Set("Cache-Control", "no-store")
writeJSON(w, http.StatusOK, s.buildInfo)
}
func (s *Server) ask(w http.ResponseWriter, r *http.Request) {
r.Body = http.MaxBytesReader(w, r.Body, 1<<20)
defer r.Body.Close()
+40
View File
@@ -25,6 +25,46 @@ func (f *captureQA) Ask(_ context.Context, request core.QARequest) (core.QAResul
return core.QAResult{TurnID: "turn-capture", Answer: "answer"}, nil
}
func TestVersionEndpointReturnsBuildMetadataWithoutAuthentication(t *testing.T) {
server, err := NewWithOptions(fakeQA{}, "ingress-secret", Options{
BuildInfo: BuildInfo{
Version: "2.1.0",
CommitSHA: "abc123def456",
BuildTime: "2026-07-27T12:00:00Z",
},
})
if err != nil {
t.Fatal(err)
}
request := httptest.NewRequest(http.MethodGet, "/version", nil)
result := httptest.NewRecorder()
server.Handler().ServeHTTP(result, request)
if result.Code != http.StatusOK {
t.Fatalf("status=%d body=%s", result.Code, result.Body.String())
}
if got := result.Header().Get("Cache-Control"); got != "no-store" {
t.Fatalf("Cache-Control=%q, want no-store", got)
}
want := `{"version":"2.1.0","commit_sha":"abc123def456","build_time":"2026-07-27T12:00:00Z"}`
if got := strings.TrimSpace(result.Body.String()); got != want {
t.Fatalf("body=%s, want %s", got, want)
}
}
func TestVersionEndpointUsesSafeDefaults(t *testing.T) {
server, err := New(fakeQA{}, "ingress-secret")
if err != nil {
t.Fatal(err)
}
request := httptest.NewRequest(http.MethodGet, "/version", nil)
result := httptest.NewRecorder()
server.Handler().ServeHTTP(result, request)
want := `{"version":"dev","commit_sha":"unknown","build_time":"unknown"}`
if got := strings.TrimSpace(result.Body.String()); result.Code != http.StatusOK || got != want {
t.Fatalf("status=%d body=%s, want %s", result.Code, got, want)
}
}
func TestQAEndpointRequiresTokenAndReturnsResult(t *testing.T) {
server, err := New(fakeQA{}, "ingress-secret")
if err != nil {
+64
View File
@@ -0,0 +1,64 @@
package outboundimage
import (
"bytes"
"context"
"errors"
"strings"
"time"
"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss"
"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss/credentials"
)
type AliyunOSSStore struct {
client *oss.Client
bucket string
}
func NewAliyunOSSStore(endpoint, region, accessKeyID, accessKeySecret, bucket string) (*AliyunOSSStore, error) {
endpoint = strings.TrimSpace(endpoint)
region = strings.TrimSpace(region)
accessKeyID = strings.TrimSpace(accessKeyID)
accessKeySecret = strings.TrimSpace(accessKeySecret)
bucket = strings.TrimSpace(bucket)
if endpoint == "" || region == "" || accessKeyID == "" || accessKeySecret == "" || bucket == "" {
return nil, errors.New("outbound image: OSS endpoint, region, credentials, and bucket are required")
}
provider := credentials.NewStaticCredentialsProvider(accessKeyID, accessKeySecret, "")
config := oss.LoadDefaultConfig().
WithCredentialsProvider(provider).
WithRegion(region).
WithEndpoint(endpoint)
return &AliyunOSSStore{client: oss.NewClient(config), bucket: bucket}, nil
}
func (s *AliyunOSSStore) Put(ctx context.Context, key string, data []byte, contentType string) error {
if s == nil || s.client == nil || strings.TrimSpace(key) == "" || len(data) == 0 {
return errors.New("outbound image: invalid OSS put request")
}
_, err := s.client.PutObject(ctx, &oss.PutObjectRequest{
Bucket: oss.Ptr(s.bucket),
Key: oss.Ptr(key),
Body: bytes.NewReader(data),
ContentType: oss.Ptr(contentType),
})
if err != nil {
return errors.New("outbound image: OSS put failed")
}
return nil
}
func (s *AliyunOSSStore) PresignGet(ctx context.Context, key string, ttl time.Duration) (string, error) {
if s == nil || s.client == nil || strings.TrimSpace(key) == "" || ttl <= 0 {
return "", errors.New("outbound image: invalid OSS presign request")
}
result, err := s.client.Presign(ctx, &oss.GetObjectRequest{
Bucket: oss.Ptr(s.bucket),
Key: oss.Ptr(key),
}, oss.PresignExpires(ttl))
if err != nil {
return "", errors.New("outbound image: OSS presign failed")
}
return result.URL, nil
}
+65
View File
@@ -0,0 +1,65 @@
package outboundimage
import (
"context"
"crypto/sha256"
"encoding/hex"
"errors"
"path"
"strings"
"time"
)
type ObjectStore interface {
Put(ctx context.Context, key string, data []byte, contentType string) error
PresignGet(ctx context.Context, key string, ttl time.Duration) (string, error)
}
type HashPublisher struct {
store ObjectStore
prefix string
ttl time.Duration
}
func NewHashPublisher(store ObjectStore, prefix string, ttl time.Duration) *HashPublisher {
return &HashPublisher{store: store, prefix: strings.Trim(strings.TrimSpace(prefix), "/"), ttl: ttl}
}
func (p *HashPublisher) Publish(ctx context.Context, data []byte, contentType string) (string, error) {
if p == nil || p.store == nil || len(data) == 0 || p.prefix == "" || p.ttl <= 0 {
return "", errors.New("outbound image: invalid publisher configuration")
}
extension := imageExtension(contentType)
if extension == "" {
return "", errors.New("outbound image: unsupported image content type")
}
sum := sha256.Sum256(data)
digest := hex.EncodeToString(sum[:])
key := path.Join(p.prefix, digest[:2], digest+extension)
if err := p.store.Put(ctx, key, data, contentType); err != nil {
return "", errors.New("outbound image: publish object failed")
}
signedURL, err := p.store.PresignGet(ctx, key, p.ttl)
if err != nil {
return "", errors.New("outbound image: sign object URL failed")
}
if !strings.HasPrefix(strings.TrimSpace(signedURL), "https://") {
return "", errors.New("outbound image: signed object URL must use HTTPS")
}
return signedURL, nil
}
func imageExtension(contentType string) string {
switch strings.ToLower(strings.TrimSpace(strings.Split(contentType, ";")[0])) {
case "image/jpeg", "image/jpg":
return ".jpg"
case "image/png":
return ".png"
case "image/webp":
return ".webp"
case "image/gif":
return ".gif"
default:
return ""
}
}
+55
View File
@@ -0,0 +1,55 @@
package outboundimage
import (
"context"
"strings"
"testing"
"time"
)
type fakeObjectStore struct {
key string
data []byte
contentType string
ttl time.Duration
}
func (s *fakeObjectStore) Put(_ context.Context, key string, data []byte, contentType string) error {
s.key = key
s.data = append([]byte(nil), data...)
s.contentType = contentType
return nil
}
func (s *fakeObjectStore) PresignGet(_ context.Context, key string, ttl time.Duration) (string, error) {
s.ttl = ttl
return "https://private-oss.example/" + key + "?signature=short", nil
}
func TestHashPublisherUsesDeterministicTemporaryKeyAndTwoHourHTTPSURL(t *testing.T) {
store := &fakeObjectStore{}
publisher := NewHashPublisher(store, "dobotai/im-images", 2*time.Hour)
first, err := publisher.Publish(context.Background(), []byte("same-image"), "image/jpeg")
if err != nil {
t.Fatal(err)
}
firstKey := store.key
second, err := publisher.Publish(context.Background(), []byte("same-image"), "image/jpeg")
if err != nil {
t.Fatal(err)
}
if first != second || firstKey != store.key {
t.Fatalf("publish is not deterministic: url1=%q url2=%q key1=%q key2=%q", first, second, firstKey, store.key)
}
if !strings.HasPrefix(store.key, "dobotai/im-images/") || !strings.HasSuffix(store.key, ".jpg") {
t.Fatalf("key=%q", store.key)
}
if strings.Contains(store.key, "same-image") {
t.Fatalf("key leaks content: %q", store.key)
}
if store.ttl != 2*time.Hour || !strings.HasPrefix(first, "https://") {
t.Fatalf("ttl=%s url=%q", store.ttl, first)
}
}
+65
View File
@@ -0,0 +1,65 @@
package outboundimage
import (
"context"
"regexp"
"strings"
)
var resourceMarkdownImageRE = regexp.MustCompile(`!\[[^\]]*\]\((resource://[0-9A-Za-z_-]+)\)`)
type Fetcher interface {
Fetch(ctx context.Context, reference string) ([]byte, string, error)
}
type Publisher interface {
Publish(ctx context.Context, data []byte, contentType string) (string, error)
}
type Rewriter struct {
fetcher Fetcher
publisher Publisher
maxImages int
}
func NewRewriter(fetcher Fetcher, publisher Publisher, maxImages int) *Rewriter {
return &Rewriter{fetcher: fetcher, publisher: publisher, maxImages: maxImages}
}
func (r *Rewriter) Rewrite(ctx context.Context, markdown string) string {
if r == nil || r.fetcher == nil || r.publisher == nil || r.maxImages <= 0 {
return resourceMarkdownImageRE.ReplaceAllString(markdown, "")
}
urls := map[string]string{}
attempted := 0
return resourceMarkdownImageRE.ReplaceAllStringFunc(markdown, func(image string) string {
match := resourceMarkdownImageRE.FindStringSubmatch(image)
if len(match) != 2 {
return ""
}
reference := match[1]
if signedURL, ok := urls[reference]; ok {
if signedURL == "" {
return ""
}
return strings.Replace(image, reference, signedURL, 1)
}
if attempted >= r.maxImages {
urls[reference] = ""
return ""
}
attempted++
data, contentType, err := r.fetcher.Fetch(ctx, reference)
if err != nil {
urls[reference] = ""
return ""
}
signedURL, err := r.publisher.Publish(ctx, data, contentType)
if err != nil || !strings.HasPrefix(signedURL, "https://") {
urls[reference] = ""
return ""
}
urls[reference] = signedURL
return strings.Replace(image, reference, signedURL, 1)
})
}
+78
View File
@@ -0,0 +1,78 @@
package outboundimage
import (
"context"
"errors"
"strings"
"testing"
)
type fakeFetcher struct {
calls map[string]int
fail map[string]bool
}
func (f *fakeFetcher) Fetch(_ context.Context, ref string) ([]byte, string, error) {
if f.calls == nil {
f.calls = map[string]int{}
}
f.calls[ref]++
if f.fail[ref] {
return nil, "", errors.New("fetch failed")
}
return []byte("image:" + ref), "image/png", nil
}
type fakePublisher struct {
calls int
}
func (p *fakePublisher) Publish(_ context.Context, data []byte, contentType string) (string, error) {
p.calls++
if contentType != "image/png" || len(data) == 0 {
return "", errors.New("bad image")
}
return "https://img.example/" + string(rune('0'+p.calls)) + ".png?sig=short", nil
}
func TestRewriteMarkdownResourceImagesDeduplicatesAndPreservesOtherURLs(t *testing.T) {
fetcher := &fakeFetcher{}
publisher := &fakePublisher{}
r := NewRewriter(fetcher, publisher, 5)
input := "步骤一\n![图1](resource://AbCdEfGhIjKlMnOpQrStUv)\n重复\n![图1重复](resource://AbCdEfGhIjKlMnOpQrStUv)\n外部\n![官网](https://example.com/a.png)"
got := r.Rewrite(context.Background(), input)
if strings.Contains(got, "resource://") {
t.Fatalf("resource reference remained: %s", got)
}
if strings.Count(got, "https://img.example/1.png?sig=short") != 2 {
t.Fatalf("deduplicated signed URL count mismatch: %s", got)
}
if !strings.Contains(got, "![官网](https://example.com/a.png)") {
t.Fatalf("ordinary HTTPS image changed: %s", got)
}
if fetcher.calls["resource://AbCdEfGhIjKlMnOpQrStUv"] != 1 || publisher.calls != 1 {
t.Fatalf("calls fetch=%v publish=%d", fetcher.calls, publisher.calls)
}
}
func TestRewriteMarkdownResourceImagesDropsFailuresAndImagesBeyondLimit(t *testing.T) {
failed := "resource://FailedRefAbCdEfGhIjKlMn"
fetcher := &fakeFetcher{fail: map[string]bool{failed: true}}
publisher := &fakePublisher{}
r := NewRewriter(fetcher, publisher, 3)
input := "保留正文\n![坏图](" + failed + ")\n![一](resource://FirstRefAbCdEfGhIjKlMnO)\n![二](resource://SecondRefAbCdEfGhIjKlMn)\n![三](resource://ThirdRefAbCdEfGhIjKlMnO)"
got := r.Rewrite(context.Background(), input)
if !strings.Contains(got, "保留正文") {
t.Fatalf("text was lost: %s", got)
}
if strings.Contains(got, "resource://") || strings.Contains(got, "坏图") || strings.Contains(got, "![三]") {
t.Fatalf("broken or over-limit image remained: %s", got)
}
if publisher.calls != 2 {
t.Fatalf("publisher calls=%d, want 2", publisher.calls)
}
}
+80
View File
@@ -0,0 +1,80 @@
package outboundimage
import (
"context"
"errors"
"fmt"
"io"
"net/http"
"net/url"
"strconv"
"strings"
"time"
)
type WeKnoraFetcher struct {
baseURL *url.URL
apiKey string
client *http.Client
maxBytes int64
}
func NewWeKnoraFetcher(baseURL, apiKey string, client *http.Client, maxBytes int64) (*WeKnoraFetcher, error) {
parsed, err := url.Parse(strings.TrimRight(strings.TrimSpace(baseURL), "/"))
if err != nil || parsed.Scheme == "" || parsed.Host == "" {
return nil, errors.New("outbound image: invalid WeKnora base URL")
}
if strings.TrimSpace(apiKey) == "" {
return nil, errors.New("outbound image: WeKnora API key is required")
}
if maxBytes <= 0 {
return nil, errors.New("outbound image: positive max bytes is required")
}
if client == nil {
client = &http.Client{Timeout: 30 * time.Second}
}
return &WeKnoraFetcher{baseURL: parsed, apiKey: apiKey, client: client, maxBytes: maxBytes}, nil
}
func (f *WeKnoraFetcher) Fetch(ctx context.Context, reference string) ([]byte, string, error) {
if !strings.HasPrefix(reference, "resource://") || !resourceMarkdownImageRE.MatchString("![]("+reference+")") {
return nil, "", errors.New("outbound image: unsupported resource reference")
}
u := *f.baseURL
u.Path = "/files"
u.RawPath = ""
query := u.Query()
query.Set("file_path", reference)
u.RawQuery = query.Encode()
req, err := http.NewRequestWithContext(ctx, http.MethodGet, u.String(), nil)
if err != nil {
return nil, "", errors.New("outbound image: create WeKnora request failed")
}
req.Header.Set("X-API-Key", f.apiKey)
resp, err := f.client.Do(req)
if err != nil {
return nil, "", fmt.Errorf("outbound image: WeKnora request failed: %w", err)
}
defer resp.Body.Close()
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
_, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 4096))
return nil, "", fmt.Errorf("outbound image: WeKnora HTTP %d", resp.StatusCode)
}
contentType := strings.TrimSpace(strings.Split(resp.Header.Get("Content-Type"), ";")[0])
if !strings.HasPrefix(contentType, "image/") {
return nil, "", errors.New("outbound image: WeKnora response is not an image")
}
if length := strings.TrimSpace(resp.Header.Get("Content-Length")); length != "" {
if n, convErr := strconv.ParseInt(length, 10, 64); convErr == nil && n > f.maxBytes {
return nil, "", errors.New("outbound image: WeKnora image exceeds size limit")
}
}
data, err := io.ReadAll(io.LimitReader(resp.Body, f.maxBytes+1))
if err != nil {
return nil, "", errors.New("outbound image: read WeKnora response failed")
}
if int64(len(data)) == 0 || int64(len(data)) > f.maxBytes {
return nil, "", errors.New("outbound image: WeKnora image exceeds size limit")
}
return data, contentType, nil
}
@@ -0,0 +1,77 @@
package outboundimage
import (
"context"
"io"
"net/http"
"net/http/httptest"
"testing"
)
func TestWeKnoraFetcherDownloadsAuthenticatedResourceImage(t *testing.T) {
var called bool
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
called = true
if req.URL.Path != "/files" {
t.Fatalf("path=%q", req.URL.Path)
}
if got := req.URL.Query().Get("file_path"); got != "resource://AbCdEfGhIjKlMnOpQrStUv" {
t.Fatalf("file_path=%q", got)
}
if got := req.Header.Get("X-API-Key"); got != "secret-test-key" {
t.Fatalf("api key header=%q", got)
}
w.Header().Set("Content-Type", "image/jpeg")
w.Header().Set("Content-Length", "4")
_, _ = io.WriteString(w, "JPEG")
}))
defer server.Close()
fetcher, err := NewWeKnoraFetcher(server.URL+"/api/v1", "secret-test-key", server.Client(), 10<<20)
if err != nil {
t.Fatal(err)
}
data, contentType, err := fetcher.Fetch(context.Background(), "resource://AbCdEfGhIjKlMnOpQrStUv")
if err != nil {
t.Fatal(err)
}
if !called || string(data) != "JPEG" || contentType != "image/jpeg" {
t.Fatalf("called=%t data=%q type=%q", called, data, contentType)
}
}
func TestWeKnoraFetcherRejectsUnsafeReferenceNonImageAndOversize(t *testing.T) {
t.Run("unsafe reference", func(t *testing.T) {
fetcher, err := NewWeKnoraFetcher("https://weknora.example", "key", http.DefaultClient, 4)
if err != nil {
t.Fatal(err)
}
if _, _, err := fetcher.Fetch(context.Background(), "https://evil.example/a.png"); err == nil {
t.Fatal("expected unsafe reference rejection")
}
})
for _, tc := range []struct {
name string
contentType string
body string
}{
{name: "non image", contentType: "text/plain", body: "TEXT"},
{name: "oversize", contentType: "image/png", body: "12345"},
} {
t.Run(tc.name, func(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
w.Header().Set("Content-Type", tc.contentType)
_, _ = io.WriteString(w, tc.body)
}))
defer server.Close()
fetcher, err := NewWeKnoraFetcher(server.URL, "key", server.Client(), 4)
if err != nil {
t.Fatal(err)
}
if _, _, err := fetcher.Fetch(context.Background(), "resource://AbCdEfGhIjKlMnOpQrStUv"); err == nil {
t.Fatal("expected response rejection")
}
})
}
}
+20
View File
@@ -0,0 +1,20 @@
package qa
import "testing"
func TestLocalizedQueryPreservesOriginalWhenAuto(t *testing.T) {
if got := localizedQuery("原始问题", "auto"); got != "原始问题" {
t.Fatalf("got %q", got)
}
}
func TestLocalizedQueryAddsRequestedLanguageWithoutChangingInput(t *testing.T) {
original := "机械臂怎么复位?"
got := localizedQuery(original, "ja")
if got != "日本語で回答してください。\n\n"+original {
t.Fatalf("got %q", got)
}
if original != "机械臂怎么复位?" {
t.Fatal("original query changed")
}
}
+82 -40
View File
@@ -32,6 +32,10 @@ type eventStreamingWeKnora interface {
AgentChatEvents(ctx context.Context, sessionID string, request weknora.AgentChatRequest, observer weknora.StreamObserver) (weknora.ChatResult, error)
}
type sessionInvalidator interface {
InvalidateSession(ctx context.Context, sessionKey, expectedSessionID string) error
}
// SucceededTurnObserver observes a successfully finished turn. It is called
// synchronously after FinishTurn succeeds and before the QAResult is returned.
// A non-nil error from ObserveSucceededTurn must never change the QAResult or
@@ -184,9 +188,10 @@ func (s *Service) ask(ctx context.Context, request core.QARequest, answerObserve
mapping := core.SessionMapping{SessionKey: sessionKey, TenantKey: request.TenantKey,
Channel: request.Channel, AgentID: agentID, ConversationType: request.ConversationType,
ExternalSubjectID: subjectID, DisplayTitle: displayTitle}
sessionID, err := s.repository.ResolveSession(ctx, mapping, func(createCtx context.Context) (string, error) {
createSession := func(createCtx context.Context) (string, error) {
return s.weknora.CreateSession(createCtx, displayTitle, "Managed by DobotAI; business state remains outside WeKnora")
})
}
sessionID, err := s.repository.ResolveSession(ctx, mapping, createSession)
if err != nil {
turn.Status, turn.ErrorMessage = "failed", err.Error()
_ = s.repository.FinishTurn(ctx, turn)
@@ -197,9 +202,9 @@ func (s *Service) ask(ctx context.Context, request core.QARequest, answerObserve
return s.weknora.UpdateSessionTitle(updateCtx, existingSessionID, title, "Managed by DobotAI; business state remains outside WeKnora")
})
query := request.Query
query := localizedQuery(request.Query, request.ResponseLocale)
if request.ConversationType == core.ConversationGroup && strings.TrimSpace(request.SenderName) != "" {
query = request.SenderName + ": " + request.Query
query = request.SenderName + ": " + query
}
images := make([]weknora.Image, 0, len(request.Images))
for _, image := range request.Images {
@@ -209,46 +214,65 @@ func (s *Service) ask(ctx context.Context, request core.QARequest, answerObserve
Query: query, AgentID: agentID, AgentEnabled: true, Channel: "im", EnableMemory: true,
Images: images,
}
var chat weknora.ChatResult
if eventObserver != nil {
// Prefer AgentChatEvents; fallback to AgentChatStream adaptation; then AgentChat.
if eventStream, ok := s.weknora.(eventStreamingWeKnora); ok {
chat, err = eventStream.AgentChatEvents(ctx, sessionID, chatRequest, eventObserver)
} else if streaming, ok := s.weknora.(streamingWeKnora); ok {
// Adapt AgentChatStream answer chunks to StreamEvent
chat, err = streaming.AgentChatStream(ctx, sessionID, chatRequest, func(full string) {
if answerObserver != nil {
answerObserver(full)
}
eventObserver(weknora.StreamEvent{
ResponseType: "answer",
FullAnswer: full,
runChat := func(targetSessionID string) (weknora.ChatResult, error) {
var result weknora.ChatResult
var runErr error
if eventObserver != nil {
// Prefer AgentChatEvents; fallback to AgentChatStream adaptation; then AgentChat.
if eventStream, ok := s.weknora.(eventStreamingWeKnora); ok {
result, runErr = eventStream.AgentChatEvents(ctx, targetSessionID, chatRequest, eventObserver)
} else if streaming, ok := s.weknora.(streamingWeKnora); ok {
// Adapt AgentChatStream answer chunks to StreamEvent
result, runErr = streaming.AgentChatStream(ctx, targetSessionID, chatRequest, func(full string) {
if answerObserver != nil {
answerObserver(full)
}
eventObserver(weknora.StreamEvent{
ResponseType: "answer",
FullAnswer: full,
})
})
})
} else {
result, runErr = s.weknora.AgentChat(ctx, targetSessionID, chatRequest)
if runErr == nil && strings.TrimSpace(result.Answer) != "" {
if answerObserver != nil {
answerObserver(result.Answer)
}
eventObserver(weknora.StreamEvent{
ResponseType: "answer",
FullAnswer: result.Answer,
Done: true,
})
}
}
} else if answerObserver != nil {
if streaming, ok := s.weknora.(streamingWeKnora); ok {
result, runErr = streaming.AgentChatStream(ctx, targetSessionID, chatRequest, answerObserver)
} else {
result, runErr = s.weknora.AgentChat(ctx, targetSessionID, chatRequest)
if runErr == nil && strings.TrimSpace(result.Answer) != "" {
answerObserver(result.Answer)
}
}
} else {
chat, err = s.weknora.AgentChat(ctx, sessionID, chatRequest)
if err == nil && strings.TrimSpace(chat.Answer) != "" {
if answerObserver != nil {
answerObserver(chat.Answer)
}
eventObserver(weknora.StreamEvent{
ResponseType: "answer",
FullAnswer: chat.Answer,
Done: true,
})
result, runErr = s.weknora.AgentChat(ctx, targetSessionID, chatRequest)
}
return result, runErr
}
chat, err := runChat(sessionID)
if errors.Is(err, weknora.ErrSessionNotFound) {
if invalidator, ok := s.repository.(sessionInvalidator); ok {
if invalidateErr := invalidator.InvalidateSession(ctx, sessionKey, sessionID); invalidateErr != nil {
err = fmt.Errorf("invalidate stale weknora session: %w", invalidateErr)
} else if freshSessionID, resolveErr := s.repository.ResolveSession(ctx, mapping, createSession); resolveErr != nil {
err = fmt.Errorf("recreate weknora session: %w", resolveErr)
} else {
sessionID = freshSessionID
turn.WeKnoraSessionID = freshSessionID
chat, err = runChat(freshSessionID)
}
}
} else if answerObserver != nil {
if streaming, ok := s.weknora.(streamingWeKnora); ok {
chat, err = streaming.AgentChatStream(ctx, sessionID, chatRequest, answerObserver)
} else {
chat, err = s.weknora.AgentChat(ctx, sessionID, chatRequest)
if err == nil && strings.TrimSpace(chat.Answer) != "" {
answerObserver(chat.Answer)
}
}
} else {
chat, err = s.weknora.AgentChat(ctx, sessionID, chatRequest)
}
if err == nil && strings.TrimSpace(chat.Answer) == "" {
err = weknora.ErrEmptyAnswer
@@ -274,6 +298,24 @@ func (s *Service) ask(ctx context.Context, request core.QARequest, answerObserve
WeKnoraRequestID: chat.RequestID, Answer: chat.Answer, References: chat.References}, nil
}
func localizedQuery(query, locale string) string {
query = strings.TrimSpace(query)
switch strings.TrimSpace(locale) {
case "zh-CN":
return "请使用简体中文回答。\n\n" + query
case "en":
return "Answer in English.\n\n" + query
case "ja":
return "日本語で回答してください。\n\n" + query
case "ko":
return "한국어로 답변해 주세요.\n\n" + query
case "es":
return "Responde en español.\n\n" + query
default:
return query
}
}
func normalizeRequestIdentity(request core.QARequest) core.QARequest {
request.TenantKey = strings.TrimSpace(request.TenantKey)
request.Channel = strings.TrimSpace(request.Channel)
+67
View File
@@ -76,6 +76,14 @@ func (r *memoryRepository) SyncSessionTitle(ctx context.Context, m core.SessionM
r.mu.Unlock()
return nil
}
func (r *memoryRepository) InvalidateSession(_ context.Context, sessionKey, expectedSessionID string) error {
r.mu.Lock()
defer r.mu.Unlock()
if r.sessions[sessionKey] == expectedSessionID {
delete(r.sessions, sessionKey)
}
return nil
}
func (r *memoryRepository) FinishTurn(_ context.Context, turn core.Turn) error {
r.mu.Lock()
defer r.mu.Unlock()
@@ -142,6 +150,28 @@ func (f *fakeWeKnora) AgentChat(_ context.Context, sessionID string, request wek
return weknora.ChatResult{RequestID: "request-1", Answer: "answer", References: []core.Reference{{KnowledgeID: "kb-1"}}}, nil
}
type staleSessionWeKnora struct {
createdSessions int
chatSessions []string
}
func (f *staleSessionWeKnora) CreateSession(context.Context, string, string) (string, error) {
f.createdSessions++
return "fresh-session", nil
}
func (f *staleSessionWeKnora) UpdateSessionTitle(context.Context, string, string, string) error {
return nil
}
func (f *staleSessionWeKnora) AgentChat(_ context.Context, sessionID string, _ weknora.AgentChatRequest) (weknora.ChatResult, error) {
f.chatSessions = append(f.chatSessions, sessionID)
if sessionID == "stale-session" {
return weknora.ChatResult{}, weknora.ErrSessionNotFound
}
return weknora.ChatResult{RequestID: "request-1", Answer: "recovered"}, nil
}
type fakeStreamingWeKnora struct {
*fakeWeKnora
updates []string
@@ -785,6 +815,43 @@ func TestAskReturnsGeneratedDeliveryIdentityAndReplay(t *testing.T) {
}
}
func TestAskRecoversStaleWeKnoraSessionOnce(t *testing.T) {
repo := newMemoryRepository()
client := &staleSessionWeKnora{}
service, err := New(repo, fixedResolver{}, client, session.GroupShared)
if err != nil {
t.Fatal(err)
}
req := core.QARequest{
TenantKey: "tenant", Channel: "dingtalk", ConversationType: core.ConversationDirect,
ExternalUserID: "user-1", ExternalMessageID: "message-stale-session", Query: "question",
}
sessionKey, _, err := session.BuildKey(session.Identity{
TenantKey: req.TenantKey, Channel: req.Channel, AgentID: "agent-1",
ConversationType: req.ConversationType, ExternalUserID: req.ExternalUserID,
GroupPolicy: session.GroupShared,
})
if err != nil {
t.Fatal(err)
}
repo.sessions[sessionKey] = "stale-session"
result, err := service.Ask(context.Background(), req)
if err != nil {
t.Fatalf("Ask should recover stale session: %v", err)
}
if result.Answer != "recovered" || result.WeKnoraSessionID != "fresh-session" {
t.Fatalf("unexpected recovered result: %+v", result)
}
if client.createdSessions != 1 || len(client.chatSessions) != 2 ||
client.chatSessions[0] != "stale-session" || client.chatSessions[1] != "fresh-session" {
t.Fatalf("expected one recreate and one retry, created=%d chats=%v", client.createdSessions, client.chatSessions)
}
if repo.sessions[sessionKey] != "fresh-session" {
t.Fatalf("repository mapping was not replaced: %q", repo.sessions[sessionKey])
}
}
func TestRecordDeliveryDelegatesToRepository(t *testing.T) {
repo := newMemoryRepository()
service, err := New(repo, fixedResolver{}, &fakeWeKnora{}, session.GroupShared)
+53 -18
View File
@@ -19,11 +19,12 @@ const defaultBase = "https://api.dingtalk.com"
// Config holds the required parameters for the DingTalk question-table client.
type Config struct {
ClientID string
Secret string
BaseID string
SheetID string
OperatorID string
ClientID string
Secret string
BaseID string
SheetID string
OperatorID string
FieldMapping map[string]string
// TokenURL overrides the token endpoint. Empty → default.
TokenURL string
@@ -54,11 +55,12 @@ var defaultTransport = func() *http.Transport {
// Client is a stdlib-only DingTalk HTTP client that inserts records into a
// Notable v1 base. It owns a mutex-protected in-memory access-token cache.
type Client struct {
clientID string
secret string
baseID string
sheetID string
operatorID string
clientID string
secret string
baseID string
sheetID string
operatorID string
fieldMapping map[string]string
tokenURL string
baseURL string
@@ -113,14 +115,15 @@ func NewClient(cfg Config, httpClient *http.Client) (*Client, error) {
}
return &Client{
clientID: clientID,
secret: secret,
baseID: baseID,
sheetID: sheetID,
operatorID: operatorID,
tokenURL: tokenURL,
baseURL: baseURL,
http: httpClient,
clientID: clientID,
secret: secret,
baseID: baseID,
sheetID: sheetID,
operatorID: operatorID,
fieldMapping: cloneFieldMapping(cfg.FieldMapping),
tokenURL: tokenURL,
baseURL: baseURL,
http: httpClient,
}, nil
}
@@ -161,6 +164,7 @@ func (c *Client) InsertRecord(ctx context.Context, row Row) (string, error) {
// InsertFields sends one already-typed field map to DingTalk Notable v1 and
// returns the record ID. It is shared by other reviewed Notable table schemas.
func (c *Client) InsertFields(ctx context.Context, fields map[string]interface{}) (string, error) {
fields = c.remapFields(fields)
token, err := c.getAccessToken(ctx)
if err != nil {
return "", fmt.Errorf("questiontable token: %w", err)
@@ -213,6 +217,37 @@ func (c *Client) InsertFields(ctx context.Context, fields map[string]interface{}
return ir.Value[0].ID, nil
}
func cloneFieldMapping(in map[string]string) map[string]string {
out := make(map[string]string, len(in))
for key, value := range in {
out[key] = value
}
return out
}
var providerFieldLogical = map[string]string{
"ZaMZF6j": "chinese_question", "SCug9My": "asked_time", "S2XA1uZ": "asker",
"bETcL3r": "original_question", "Y4cpmxU": "answer", "JgY2f1z": "trigger_person",
"DNIV3mQ": "language", "eBc6a6b": "department", "fPNLeVW": "region",
"01ZM8y7": "standard_question", "fj3JQ1C": "original_question", "07YYagS": "language", "Eu7gkKu": "asker",
}
func (c *Client) remapFields(in map[string]interface{}) map[string]interface{} {
if len(c.fieldMapping) == 0 {
return in
}
out := make(map[string]interface{}, len(in))
for providerID, value := range in {
logical := providerFieldLogical[providerID]
if mapped := c.fieldMapping[logical]; logical != "" && mapped != "" {
out[mapped] = value
} else {
out[providerID] = value
}
}
return out
}
// safeDoErr wraps an http.Do error without exposing the raw URL or credentials.
// If the context is already done, returns a wrapped ctx.Err() so errors.Is works.
func safeDoErr(ctx context.Context, err error) error {
+14
View File
@@ -0,0 +1,14 @@
package questiontable
import "testing"
func TestRemapFieldsUsesValidatedLogicalMapping(t *testing.T) {
client := &Client{fieldMapping: map[string]string{"answer": "new-answer", "original_question": "new-question"}}
got := client.remapFields(map[string]interface{}{FieldAnswer: "A", FieldOriginalQuestion: "Q"})
if got["new-answer"] != "A" || got["new-question"] != "Q" {
t.Fatalf("got=%#v", got)
}
if _, ok := got[FieldAnswer]; ok {
t.Fatal("legacy field id must not remain when mapping exists")
}
}
+108
View File
@@ -0,0 +1,108 @@
package questiontable
import (
"context"
"encoding/json"
"errors"
"fmt"
"io"
"net/http"
"net/url"
"strings"
)
type Field struct {
ID string `json:"id"`
Name string `json:"name"`
Type string `json:"type"`
}
type RequiredField struct {
Logical string
Name string
Type string
}
func QuestionSchema() []RequiredField {
return []RequiredField{
{"chinese_question", "中文问题", "text"},
{"asked_time", "提问时间", "date"},
{"asker", "提问人", "user"},
{"original_question", "原始问题", "text"},
{"answer", "AI答案", "text"},
{"trigger_person", "流程触发人员信息", "text"},
{"language", "语言", "text"},
{"department", "部门", "text"},
{"region", "区域", "text"},
}
}
func FAQSchema() []RequiredField {
return []RequiredField{
{"standard_question", "标准问题(必填)", "text"},
{"original_question", "原始问题", "text"},
{"language", "语言", "text"},
{"asker", "提问人", "user"},
}
}
func MapRequiredFields(fields []Field, required []RequiredField) (map[string]string, error) {
out := make(map[string]string, len(required))
for _, want := range required {
nameMatches := make([]Field, 0, 1)
for _, got := range fields {
if strings.TrimSpace(got.Name) == want.Name {
nameMatches = append(nameMatches, got)
}
}
if len(nameMatches) == 0 {
return nil, errors.New("missing_field")
}
if len(nameMatches) > 1 {
return nil, errors.New("duplicate_field")
}
if strings.TrimSpace(nameMatches[0].Type) != want.Type {
return nil, errors.New("field_type_mismatch")
}
if strings.TrimSpace(nameMatches[0].ID) == "" {
return nil, errors.New("missing_field_id")
}
out[want.Logical] = strings.TrimSpace(nameMatches[0].ID)
}
return out, nil
}
func (c *Client) ReadFields(ctx context.Context) ([]Field, error) {
token, err := c.getAccessToken(ctx)
if err != nil {
return nil, errors.New("schema_token_failed")
}
endpoint := fmt.Sprintf("%s/v1.0/notable/bases/%s/sheets/%s/fields?operatorId=%s", c.baseURL, url.PathEscape(c.baseID), url.PathEscape(c.sheetID), url.QueryEscape(c.operatorID))
req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil)
if err != nil {
return nil, errors.New("schema_request_failed")
}
req.Header.Set("x-acs-dingtalk-access-token", token)
resp, err := c.http.Do(req)
if err != nil {
return nil, errors.New("schema_network_failed")
}
defer resp.Body.Close()
raw, err := io.ReadAll(io.LimitReader(resp.Body, 256<<10))
if err != nil {
return nil, errors.New("schema_read_failed")
}
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
return nil, fmt.Errorf("schema_http_%d", resp.StatusCode)
}
var payload struct {
Value []Field `json:"value"`
}
if json.Unmarshal(raw, &payload) != nil {
return nil, errors.New("schema_decode_failed")
}
if len(payload.Value) == 0 {
return nil, errors.New("schema_empty")
}
return payload.Value, nil
}
+77
View File
@@ -0,0 +1,77 @@
package questiontable
import (
"context"
"net/http"
"net/http/httptest"
"strings"
"testing"
)
func TestReadFieldsUsesOperatorAndMapsQuestionSchema(t *testing.T) {
mux := http.NewServeMux()
mux.HandleFunc("/token", func(w http.ResponseWriter, _ *http.Request) {
_, _ = w.Write([]byte(`{"accessToken":"token","expireIn":7200}`))
})
mux.HandleFunc("/v1.0/notable/bases/base/sheets/sheet/fields", func(w http.ResponseWriter, r *http.Request) {
if r.URL.Query().Get("operatorId") != "union" {
t.Fatalf("operator=%q", r.URL.Query().Get("operatorId"))
}
_, _ = w.Write([]byte(`{"value":[
{"id":"q","name":"中文问题","type":"text"},{"id":"time","name":"提问时间","type":"date"},
{"id":"asker","name":"提问人","type":"user"},{"id":"orig","name":"原始问题","type":"text"},
{"id":"answer","name":"AI答案","type":"text"},{"id":"trigger","name":"流程触发人员信息","type":"text"},
{"id":"lang","name":"语言","type":"text"},{"id":"dept","name":"部门","type":"text"},{"id":"region","name":"区域","type":"text"}]}`))
})
server := httptest.NewServer(mux)
defer server.Close()
client, err := NewClient(Config{ClientID: "id", Secret: "secret", BaseID: "base", SheetID: "sheet", OperatorID: "union", TokenURL: server.URL + "/token", BaseURL: server.URL}, server.Client())
if err != nil {
t.Fatal(err)
}
fields, err := client.ReadFields(context.Background())
if err != nil {
t.Fatal(err)
}
mapping, err := MapRequiredFields(fields, QuestionSchema())
if err != nil {
t.Fatal(err)
}
if mapping["answer"] != "answer" || len(mapping) != 9 {
t.Fatalf("mapping=%v", mapping)
}
}
func TestMapRequiredFieldsRejectsMissingDuplicateAndWrongType(t *testing.T) {
required := []RequiredField{{Logical: "question", Name: "问题", Type: "text"}}
cases := []struct {
name string
fields []Field
code string
}{
{"missing", nil, "missing_field"},
{"duplicate", []Field{{ID: "1", Name: "问题", Type: "text"}, {ID: "2", Name: "问题", Type: "text"}}, "duplicate_field"},
{"wrong type", []Field{{ID: "1", Name: "问题", Type: "user"}}, "field_type_mismatch"},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
_, err := MapRequiredFields(tc.fields, required)
if err == nil || !strings.Contains(err.Error(), tc.code) {
t.Fatalf("err=%v", err)
}
})
}
}
func TestFAQSchemaUsesFourCurrentInsertFields(t *testing.T) {
required := FAQSchema()
if len(required) != 4 {
t.Fatalf("len=%d", len(required))
}
want := map[string]string{"standard_question": "text", "original_question": "text", "language": "text", "asker": "user"}
for _, field := range required {
if want[field.Logical] != field.Type {
t.Fatalf("field=%+v", field)
}
}
}
+102
View File
@@ -0,0 +1,102 @@
package runtime
import (
"context"
"errors"
"sync"
"time"
)
type Credentials struct {
ClientID string
Secret string
Ref string
}
type Component interface {
Start(context.Context, Credentials) error
Stop(context.Context) error
}
type Status struct {
State string
ClientID string
LastCode string
UpdatedAt time.Time
}
type Manager struct {
mu sync.Mutex
component Component
active Credentials
status Status
}
func NewManager(component Component) *Manager {
return &Manager{component: component, status: Status{State: "stopped", UpdatedAt: time.Now()}}
}
func (m *Manager) Start(ctx context.Context, credentials Credentials) error {
m.mu.Lock()
defer m.mu.Unlock()
if m.component == nil {
return errors.New("runtime_component_required")
}
if m.status.State == "running" {
return errors.New("runtime_already_running")
}
if err := m.component.Start(ctx, credentials); err != nil {
m.status = Status{State: "error", LastCode: "active_start_failed", UpdatedAt: time.Now()}
return errors.New("active_start_failed")
}
m.active = credentials
m.status = Status{State: "running", ClientID: credentials.ClientID, UpdatedAt: time.Now()}
return nil
}
func (m *Manager) Switch(ctx context.Context, candidate Credentials) error {
m.mu.Lock()
defer m.mu.Unlock()
if m.component == nil {
return errors.New("runtime_component_required")
}
old := m.active
m.status = Status{State: "switching", ClientID: old.ClientID, UpdatedAt: time.Now()}
if err := m.component.Stop(ctx); err != nil {
m.status = Status{State: "running", ClientID: old.ClientID, LastCode: "active_stop_failed", UpdatedAt: time.Now()}
return errors.New("active_stop_failed")
}
if err := m.component.Start(ctx, candidate); err == nil {
m.active = candidate
m.status = Status{State: "running", ClientID: candidate.ClientID, LastCode: "switch_ok", UpdatedAt: time.Now()}
return nil
}
if old.ClientID != "" && m.component.Start(ctx, old) == nil {
m.active = old
m.status = Status{State: "running", ClientID: old.ClientID, LastCode: "candidate_start_failed_rolled_back", UpdatedAt: time.Now()}
return errors.New("candidate_start_failed_rolled_back")
}
m.active = Credentials{}
m.status = Status{State: "error", LastCode: "rollback_failed", UpdatedAt: time.Now()}
return errors.New("rollback_failed")
}
func (m *Manager) Stop(ctx context.Context) error {
m.mu.Lock()
defer m.mu.Unlock()
if m.component == nil || m.status.State == "stopped" {
return nil
}
if err := m.component.Stop(ctx); err != nil {
return errors.New("runtime_stop_failed")
}
m.active = Credentials{}
m.status = Status{State: "stopped", UpdatedAt: time.Now()}
return nil
}
func (m *Manager) Status() Status {
m.mu.Lock()
defer m.mu.Unlock()
return m.status
}
+88
View File
@@ -0,0 +1,88 @@
package runtime
import (
"context"
"errors"
"sync"
"testing"
)
type fakeComponent struct {
mu sync.Mutex
running string
starts []string
stops int
fail map[string]error
}
func (f *fakeComponent) Start(_ context.Context, cfg Credentials) error {
f.mu.Lock()
defer f.mu.Unlock()
if f.running != "" {
return errors.New("already_running")
}
f.starts = append(f.starts, cfg.ClientID)
if err := f.fail[cfg.ClientID]; err != nil {
return err
}
f.running = cfg.ClientID
return nil
}
func (f *fakeComponent) Stop(context.Context) error {
f.mu.Lock()
defer f.mu.Unlock()
f.stops++
f.running = ""
return nil
}
func TestManagerSwitchesWithAtMostOneRuntime(t *testing.T) {
component := &fakeComponent{fail: map[string]error{}}
m := NewManager(component)
if err := m.Start(context.Background(), Credentials{ClientID: "old", Secret: "s1"}); err != nil {
t.Fatal(err)
}
if err := m.Switch(context.Background(), Credentials{ClientID: "new", Secret: "s2"}); err != nil {
t.Fatal(err)
}
if component.running != "new" || component.stops != 1 {
t.Fatalf("running=%q stops=%d", component.running, component.stops)
}
status := m.Status()
if status.ClientID != "new" || status.State != "running" {
t.Fatalf("status=%+v", status)
}
}
func TestManagerRollsBackWhenCandidateFails(t *testing.T) {
component := &fakeComponent{fail: map[string]error{"bad": errors.New("stream_failed")}}
m := NewManager(component)
if err := m.Start(context.Background(), Credentials{ClientID: "old", Secret: "s1"}); err != nil {
t.Fatal(err)
}
if err := m.Switch(context.Background(), Credentials{ClientID: "bad", Secret: "s2"}); err == nil {
t.Fatal("expected failure")
}
if component.running != "old" {
t.Fatalf("running=%q starts=%v", component.running, component.starts)
}
status := m.Status()
if status.State != "running" || status.LastCode != "candidate_start_failed_rolled_back" {
t.Fatalf("status=%+v", status)
}
}
func TestManagerReportsRollbackFailure(t *testing.T) {
component := &fakeComponent{fail: map[string]error{"bad": errors.New("bad"), "old": nil}}
m := NewManager(component)
if err := m.Start(context.Background(), Credentials{ClientID: "old", Secret: "s1"}); err != nil {
t.Fatal(err)
}
component.fail["old"] = errors.New("old_failed")
if err := m.Switch(context.Background(), Credentials{ClientID: "bad", Secret: "s2"}); err == nil {
t.Fatal("expected failure")
}
if m.Status().State != "error" || m.Status().LastCode != "rollback_failed" {
t.Fatalf("status=%+v", m.Status())
}
}
+84
View File
@@ -0,0 +1,84 @@
package secretstore
import (
"crypto/rand"
"encoding/hex"
"errors"
"os"
"path/filepath"
"strings"
)
type Store struct{ root string }
func New(root string) (*Store, error) {
root = strings.TrimSpace(root)
if root == "" {
return nil, errors.New("secret root is required")
}
if err := os.MkdirAll(root, 0700); err != nil {
return nil, err
}
if err := os.Chmod(root, 0700); err != nil {
return nil, err
}
return &Store{root: root}, nil
}
func (s *Store) Put(kind, value string) (string, error) {
kind = strings.TrimSpace(kind)
value = strings.TrimSpace(value)
if kind == "" || value == "" {
return "", errors.New("secret kind and value are required")
}
if strings.ContainsAny(kind, "/\\") || kind == "." || kind == ".." {
return "", errors.New("invalid secret kind")
}
b := make([]byte, 16)
if _, err := rand.Read(b); err != nil {
return "", err
}
ref := kind + "-" + hex.EncodeToString(b)
tmp, err := os.CreateTemp(s.root, ".new-")
if err != nil {
return "", err
}
tmpName := tmp.Name()
defer os.Remove(tmpName)
if err = tmp.Chmod(0600); err == nil {
_, err = tmp.WriteString(value)
}
if err == nil {
err = tmp.Sync()
}
closeErr := tmp.Close()
if err == nil {
err = closeErr
}
if err != nil {
return "", err
}
if err = os.Rename(tmpName, filepath.Join(s.root, ref)); err != nil {
return "", err
}
return ref, nil
}
func (s *Store) Get(ref string) (string, error) {
if ref == "" || filepath.Base(ref) != ref {
return "", errors.New("invalid secret reference")
}
b, err := os.ReadFile(filepath.Join(s.root, ref))
if err != nil {
return "", err
}
return string(b), nil
}
func (s *Store) Configured(ref string) bool {
_, err := os.Stat(filepath.Join(s.root, filepath.Base(ref)))
return ref != "" && filepath.Base(ref) == ref && err == nil
}
func (s *Store) Delete(ref string) error {
if ref == "" || filepath.Base(ref) != ref {
return errors.New("invalid secret reference")
}
return os.Remove(filepath.Join(s.root, ref))
}
+39
View File
@@ -0,0 +1,39 @@
package secretstore
import (
"os"
"path/filepath"
"testing"
)
func TestFileStoreWritesSecretWithPrivatePermissions(t *testing.T) {
root := t.TempDir()
s, err := New(root)
if err != nil {
t.Fatal(err)
}
ref, err := s.Put("dingtalk", "top-secret")
if err != nil {
t.Fatal(err)
}
info, err := os.Stat(filepath.Join(root, ref))
if err != nil {
t.Fatal(err)
}
if info.Mode().Perm() != 0600 {
t.Fatalf("mode=%o", info.Mode().Perm())
}
value, err := s.Get(ref)
if err != nil || value != "top-secret" {
t.Fatalf("value=%q err=%v", value, err)
}
}
func TestFileStoreRejectsTraversalAndBlank(t *testing.T) {
s, _ := New(t.TempDir())
if _, err := s.Put("../bad", "x"); err == nil {
t.Fatal("expected traversal rejection")
}
if _, err := s.Put("ok", ""); err == nil {
t.Fatal("expected blank rejection")
}
}
+17
View File
@@ -0,0 +1,17 @@
package settings
import "errors"
type Administrator struct {
StaffID string
DisplayName string
Department string
Role string
Status string
}
var (
ErrAdministratorNotFound = errors.New("administrator not found")
ErrLastAdministrator = errors.New("last active administrator cannot be disabled")
ErrAdministratorAlreadyDisabled = errors.New("administrator already disabled")
)
+262
View File
@@ -0,0 +1,262 @@
package settings
import (
"context"
"errors"
"fmt"
"strings"
"time"
)
type ResourceBinding struct {
Kind, TenantKey, ConversationID, ConversationTitle string
BaseID, SheetID, ViewID, OperatorID string
CreatedByStaffID string
FieldMapping map[string]string
ValidationCode string
Activate bool
}
type BootstrapCandidate struct {
TenantKey, Code, StaffID, UnionID, DisplayName, Department string
DetectedAt time.Time
}
type SettingSession struct {
TenantKey, StaffID, ConversationID, ConversationType, PendingAction string
ExpiresAt time.Time
}
type SettingInput struct{ TenantKey, Channel, StaffID, UnionID, DisplayName, Department, Text, ConversationType, ConversationID, ConversationTitle string }
type ControllerRepository interface {
IsAdministrator(context.Context, string, string) (bool, error)
SetPreference(context.Context, string, string, string, string) error
SaveBinding(context.Context, ResourceBinding) error
TestAndActivateSupportGroup(context.Context, ResourceBinding) error
DetectBootstrapCandidate(context.Context, BootstrapCandidate) error
GetSettingSession(context.Context, string, string) (SettingSession, bool, error)
SaveSettingSession(context.Context, SettingSession) error
DeleteSettingSession(context.Context, string, string) error
}
type Controller struct {
repo ControllerRepository
now func() time.Time
}
func NewController(repo ControllerRepository) *Controller {
return &Controller{repo: repo, now: time.Now}
}
func (c *Controller) ShouldHandle(ctx context.Context, tenant, staff, text string) bool {
if _, ok := ParseCommand(text); ok {
return true
}
if c == nil || c.repo == nil {
return false
}
s, ok, err := c.repo.GetSettingSession(ctx, tenant, staff)
return err == nil && ok && s.ExpiresAt.After(c.now())
}
func (c *Controller) Handle(ctx context.Context, in SettingInput) (string, error) {
if c == nil || c.repo == nil {
return "", errors.New("settings repository is required")
}
cmd, explicit := ParseCommand(in.Text)
var action string
var args []string
if explicit {
args = cmd.Args
} else {
s, ok, err := c.repo.GetSettingSession(ctx, in.TenantKey, in.StaffID)
if err != nil {
return "", err
}
if !ok || !s.ExpiresAt.After(c.now()) {
return "", nil
}
if strings.TrimSpace(in.Text) == "取消" {
_ = c.repo.DeleteSettingSession(ctx, in.TenantKey, in.StaffID)
return "已取消设置。", nil
}
action = s.PendingAction
args = []string{action, strings.TrimSpace(in.Text)}
}
if explicit && len(args) == 1 && len(args[0]) == 6 && allDigits(args[0]) {
if strings.TrimSpace(in.ConversationType) != "direct" {
return "管理员绑定码只能私聊机器人使用。", nil
}
err := c.repo.DetectBootstrapCandidate(ctx, BootstrapCandidate{TenantKey: in.TenantKey, Code: args[0], StaffID: in.StaffID, UnionID: in.UnionID, DisplayName: in.DisplayName, Department: in.Department, DetectedAt: c.now()})
if err != nil {
return "", err
}
return "身份已验证,请回到后台确认首位管理员。", nil
}
if explicit && len(args) == 0 {
admin, err := c.repo.IsAdministrator(ctx, in.TenantKey, in.StaffID)
if err != nil {
return "", err
}
if admin {
return settingMenu(true, strings.TrimSpace(in.ConversationType) == "direct"), nil
}
return settingMenu(false, false), nil
}
if action == "" {
action = strings.ToLower(args[0])
}
if action == "language" {
if len(args) != 2 || !validLocale(args[1]) {
return "可选语言:auto、zh-CN、en、ja、ko、es。", nil
}
if err := c.repo.SetPreference(ctx, in.TenantKey, defaultChannel(in.Channel), in.StaffID, args[1]); err != nil {
return "", err
}
return "✅ 回答语言已设置为 " + args[1] + "。", nil
}
admin, err := c.repo.IsAdministrator(ctx, in.TenantKey, in.StaffID)
if err != nil {
return "", err
}
if !admin {
return "你没有系统管理权限。", nil
}
switch action {
case "support":
if in.ConversationType != "group" || strings.TrimSpace(in.ConversationID) == "" {
return "请在目标技术支持群中@机器人发送 /setting support。", nil
}
binding := ResourceBinding{Kind: "support_group", TenantKey: in.TenantKey, ConversationID: in.ConversationID, ConversationTitle: in.ConversationTitle, CreatedByStaffID: in.StaffID}
if len(args) == 2 && strings.EqualFold(args[1], "test") {
err = c.repo.TestAndActivateSupportGroup(ctx, binding)
if err == nil {
return "✅ 测试成功,当前群已激活为技术支持群。", nil
}
break
}
err = c.repo.SaveBinding(ctx, binding)
if err == nil {
return "✅ 当前群已保存为技术支持群(待验证)。确认发送测试消息后,请在本群@机器人发送 /setting support test。", nil
}
case "question", "faq":
if len(args) != 2 {
err = c.repo.SaveSettingSession(ctx, SettingSession{TenantKey: in.TenantKey, StaffID: in.StaffID, ConversationID: in.ConversationID, ConversationType: in.ConversationType, PendingAction: action, ExpiresAt: c.now().Add(10 * time.Minute)})
if err != nil {
return "", err
}
return tableBindingInstructions(action), nil
}
link, e := ParseNotableLink(args[1])
if e != nil {
return "AI表格链接无法识别。", nil
}
kind := action + "_table"
err = c.repo.SaveBinding(ctx, ResourceBinding{Kind: kind, TenantKey: in.TenantKey, BaseID: link.BaseID, SheetID: link.SheetID, ViewID: link.ViewID, OperatorID: in.UnionID, CreatedByStaffID: in.StaffID})
if err != nil {
return tableBindingFailureReply(action, err), nil
}
_ = c.repo.DeleteSettingSession(ctx, in.TenantKey, in.StaffID)
return fmt.Sprintf("✅ %sSchema验证通过并已激活,DobotAI将短暂重启加载Dispatcher。", map[string]string{"question": "问题AI表格", "faq": "FAQ表格"}[action]), nil
default:
return "未知设置项。发送 /setting 查看菜单。", nil
}
return "", err
}
func settingMenu(admin, direct bool) string {
menu := "⚙️ 个人设置\n入口:/setting 或 /设置\n" +
"回答语言:/setting language <auto|zh-CN|en|ja|ko|es>\n" +
"- auto:跟随当前提问语言(推荐)\n" +
"- zh-CN:简体中文\n" +
"- en / ja / ko / es:英语 / 日语 / 韩语 / 西班牙语\n" +
"例如:/setting language en"
if !admin {
return menu
}
menu += "\n\n🛠 系统管理(仅管理员)\n" +
"- support:在目标技术支持群@机器人发送 /setting support 保存;再发送 /setting support test,测试成功后激活。\n" +
"- question:私聊发送 /setting question,再在10分钟内发送问题AI表链接;Schema验证通过后激活并短暂重启。\n" +
"- faq:私聊发送 /setting faq,再在10分钟内发送FAQ表链接;Schema验证通过后激活并短暂重启。"
if !direct {
return menu + "\n\n🌐 管理后台与管理员更换:请私聊机器人发送 /setting 查看。"
}
return menu + "\n\n🌐 管理后台(仅PC)\n" +
"1. 通过SSH隧道将本地8090转发到服务器127.0.0.1:8090。\n" +
"2. 浏览器打开:http://127.0.0.1:8090/admin/settings\n" +
"3. Basic Auth用户名任意;密码为服务器DOBOTAI_ADMIN_TOKEN(仅从服务器安全获取,禁止发送到钉钉)。\n" +
"4. 更换管理员:后台生成6位码 → 新管理员私聊 /setting <6位码> → 回后台确认 → 新管理员生效后再停用旧管理员。"
}
func tableBindingFailureReply(action string, err error) string {
label := map[string]string{"question": "问题AI表格", "faq": "FAQ表格"}[action]
if label == "" {
label = "AI表格"
}
code := "table_validation_failed"
guidance := "请检查链接是否来自目标表格中的任意记录,并确认字段名称和类型符合提示要求。"
raw := ""
if err != nil {
raw = strings.ToLower(err.Error())
}
switch {
case strings.Contains(raw, "schema_http_401"), strings.Contains(raw, "schema_http_403"), strings.Contains(raw, "schema_token_failed"):
code = "table_permission_denied"
guidance = "请确认机器人应用已开通钉钉AI表格接口权限,并确认当前管理员对目标表格有访问和编辑权限。"
case strings.Contains(raw, "missing_field"):
code = "table_schema_missing_field"
guidance = "目标表缺少必需字段,请按上一条提示补齐字段后重试。"
case strings.Contains(raw, "duplicate_field"):
code = "table_schema_duplicate_field"
guidance = "目标表存在重名必需字段,请保留唯一字段后重试。"
case strings.Contains(raw, "field_type_mismatch"):
code = "table_schema_field_type_mismatch"
guidance = "目标表的必需字段类型不正确,请按上一条提示调整字段类型后重试。"
case strings.Contains(raw, "schema_network_failed"), strings.Contains(raw, "schema_request_failed"), strings.Contains(raw, "schema_read_failed"):
code = "table_schema_network_failed"
guidance = "钉钉AI表格接口暂时不可用,请稍后重试。"
case strings.Contains(raw, "schema_decode_failed"), strings.Contains(raw, "schema_empty"):
code = "table_schema_response_invalid"
guidance = "未读取到可用字段,请确认链接指向AI表格中的具体数据表。"
}
return fmt.Sprintf("❌ %s验证失败,新配置未激活。\n错误码:%s\n%s\n当前设置仍在10分钟等待期内,可直接重新发送链接卡片;发送“取消”退出。", label, code, guidance)
}
func tableBindingInstructions(action string) string {
if action == "faq" {
return "请在10分钟内直接发送FAQ表中任意一条记录的链接(钉钉显示为链接卡片也可以直接发送)。\n\n" +
"操作前请确认:\n" +
"1. 当前机器人应用具备钉钉AI表格接口权限,当前管理员具备目标表格的访问和编辑权限。\n" +
"2. FAQ表包含以下字段:标准问题(必填)、原始问题、语言、提问人。\n" +
"3. Schema验证成功后会立即激活FAQ写入,并短暂重启DobotAI加载Dispatcher。\n\n" +
"如果完整URL被钉钉自动转换为链接卡片,请只发送 alidocs.dingtalk.com/ 后面的部分。示例:\n" +
"i/nodes/<BaseID>?iframeQuery=entrance%3Ddata%26sheetId%3D<SheetID>%26viewId%3D<ViewID>\n\n" +
"iframeQuery必须完整保留,系统会自动解码其中的sheetId/viewId。\n\n" +
"发送“取消”退出。"
}
return "请在10分钟内直接发送问题AI表格中任意一条记录的链接(钉钉显示为链接卡片也可以直接发送)。\n\n" +
"操作前请确认:\n" +
"1. 当前机器人应用具备钉钉AI表格接口权限,当前管理员具备目标表格的访问和编辑权限。\n" +
"2. 问题表包含:中文问题、提问时间、提问人、原始问题、AI答案、流程触发人员信息、语言、部门、区域。\n" +
"3. Schema验证成功后会立即激活问题表写入,并短暂重启DobotAI加载Dispatcher。\n\n" +
"如果完整URL被钉钉自动转换为链接卡片,请只发送 alidocs.dingtalk.com/ 后面的部分。示例:\n" +
"i/nodes/<BaseID>?iframeQuery=entrance%3Ddata%26sheetId%3D<SheetID>%26viewId%3D<ViewID>\n\n" +
"iframeQuery必须完整保留,系统会自动解码其中的sheetId/viewId。\n\n" +
"发送“取消”退出。"
}
func validLocale(s string) bool {
switch s {
case "auto", "zh-CN", "en", "ja", "ko", "es":
return true
}
return false
}
func defaultChannel(s string) string {
if strings.TrimSpace(s) == "" {
return "dingtalk"
}
return strings.TrimSpace(s)
}
func allDigits(s string) bool {
for _, r := range s {
if r < '0' || r > '9' {
return false
}
}
return s != ""
}
+322
View File
@@ -0,0 +1,322 @@
package settings
import (
"context"
"errors"
"strings"
"testing"
"time"
)
type memoryRepo struct {
admins map[string]bool
prefs map[string]string
bindings map[string]ResourceBinding
candidate BootstrapCandidate
sessions map[string]SettingSession
saveErr error
}
func newMemoryRepo() *memoryRepo {
return &memoryRepo{admins: map[string]bool{}, prefs: map[string]string{}, bindings: map[string]ResourceBinding{}, sessions: map[string]SettingSession{}}
}
func (r *memoryRepo) IsAdministrator(_ context.Context, tenant, staff string) (bool, error) {
return r.admins[tenant+"/"+staff], nil
}
func (r *memoryRepo) SetPreference(_ context.Context, tenant, channel, staff, locale string) error {
r.prefs[tenant+"/"+staff] = locale
return nil
}
func (r *memoryRepo) SaveBinding(_ context.Context, b ResourceBinding) error {
if r.saveErr != nil {
return r.saveErr
}
r.bindings[b.Kind] = b
return nil
}
func (r *memoryRepo) TestAndActivateSupportGroup(_ context.Context, b ResourceBinding) error {
b.Activate = true
b.ValidationCode = "support_test_ok"
r.bindings[b.Kind] = b
return nil
}
func (r *memoryRepo) DetectBootstrapCandidate(_ context.Context, c BootstrapCandidate) error {
r.candidate = c
return nil
}
func (r *memoryRepo) GetSettingSession(_ context.Context, tenant, staff string) (SettingSession, bool, error) {
s, ok := r.sessions[tenant+"/"+staff]
return s, ok, nil
}
func (r *memoryRepo) SaveSettingSession(_ context.Context, s SettingSession) error {
r.sessions[s.TenantKey+"/"+s.StaffID] = s
return nil
}
func (r *memoryRepo) DeleteSettingSession(_ context.Context, tenant, staff string) error {
delete(r.sessions, tenant+"/"+staff)
return nil
}
func TestControllerLanguageForOrdinaryUser(t *testing.T) {
r := newMemoryRepo()
c := NewController(r)
out, err := c.Handle(context.Background(), SettingInput{TenantKey: "t", Channel: "dingtalk", StaffID: "u", Text: "/setting language ja", ConversationType: "direct"})
if err != nil {
t.Fatal(err)
}
if r.prefs["t/u"] != "ja" || out == "" {
t.Fatalf("prefs=%v out=%q", r.prefs, out)
}
}
func TestControllerSettingMenuExplainsLanguageWithoutAdminActionsToOrdinaryUser(t *testing.T) {
r := newMemoryRepo()
c := NewController(r)
out, err := c.Handle(context.Background(), SettingInput{TenantKey: "t", Channel: "dingtalk", StaffID: "u", Text: "/setting", ConversationType: "direct"})
if err != nil {
t.Fatal(err)
}
for _, want := range []string{"个人设置", "/setting language <auto|zh-CN|en|ja|ko|es>", "auto:跟随当前提问语言", "zh-CN:简体中文", "例如:/setting language en"} {
if !strings.Contains(out, want) {
t.Fatalf("reply=%q must contain %q", out, want)
}
}
for _, forbidden := range []string{"系统管理", "/setting support", "/admin/settings", "DOBOTAI_ADMIN_TOKEN"} {
if strings.Contains(out, forbidden) {
t.Fatalf("ordinary-user reply=%q must not contain %q", out, forbidden)
}
}
}
func TestControllerSettingMenuExplainsAdminActionsAndBackendAccess(t *testing.T) {
r := newMemoryRepo()
r.admins["t/a"] = true
c := NewController(r)
out, err := c.Handle(context.Background(), SettingInput{TenantKey: "t", Channel: "dingtalk", StaffID: "a", Text: "/setting", ConversationType: "direct"})
if err != nil {
t.Fatal(err)
}
for _, want := range []string{
"系统管理(仅管理员)",
"/setting support",
"/setting support test",
"/setting question",
"/setting faq",
"Schema验证通过后激活并短暂重启",
"SSH隧道",
"127.0.0.1:8090/admin/settings",
"Basic Auth",
"DOBOTAI_ADMIN_TOKEN",
"新管理员私聊 /setting <6位码>",
"新管理员生效后再停用旧管理员",
} {
if !strings.Contains(out, want) {
t.Fatalf("reply=%q must contain %q", out, want)
}
}
}
func TestControllerRejectsOrdinaryUserSystemBinding(t *testing.T) {
r := newMemoryRepo()
c := NewController(r)
out, err := c.Handle(context.Background(), SettingInput{TenantKey: "t", StaffID: "u", Text: "/setting support", ConversationType: "group", ConversationID: "g"})
if err != nil {
t.Fatal(err)
}
if len(r.bindings) != 0 || out != "你没有系统管理权限。" {
t.Fatalf("bindings=%v out=%q", r.bindings, out)
}
}
func TestControllerBindsAdminSupportGroup(t *testing.T) {
r := newMemoryRepo()
r.admins["t/a"] = true
c := NewController(r)
out, err := c.Handle(context.Background(), SettingInput{TenantKey: "t", StaffID: "a", Text: "/setting support", ConversationType: "group", ConversationID: "g", ConversationTitle: "技术支持"})
if err != nil {
t.Fatal(err)
}
binding := r.bindings["support_group"]
if binding.ConversationID != "g" || binding.Activate {
t.Fatalf("binding=%+v", binding)
}
if !strings.Contains(out, "support test") {
t.Fatalf("reply=%q must explain explicit test command", out)
}
}
func TestControllerExplicitlyTestsAndActivatesSupportGroup(t *testing.T) {
r := newMemoryRepo()
r.admins["t/a"] = true
c := NewController(r)
out, err := c.Handle(context.Background(), SettingInput{TenantKey: "t", StaffID: "a", Text: "/setting support test", ConversationType: "group", ConversationID: "g", ConversationTitle: "技术支持"})
if err != nil {
t.Fatal(err)
}
binding := r.bindings["support_group"]
if binding.ConversationID != "g" || !binding.Activate {
t.Fatalf("binding=%+v", binding)
}
if !strings.Contains(out, "测试成功") || !strings.Contains(out, "已激活") {
t.Fatalf("reply=%q", out)
}
}
func TestControllerOnlyHandlesSettingEntrypoints(t *testing.T) {
r := newMemoryRepo()
c := NewController(r)
if c.ShouldHandle(context.Background(), "t", "u", "/help") {
t.Fatal("help must not enter the settings controller")
}
out, err := c.Handle(context.Background(), SettingInput{TenantKey: "t", StaffID: "u", Text: "/help", ConversationType: "direct"})
if err != nil {
t.Fatal(err)
}
if out != "" {
t.Fatalf("help reply=%q must be empty", out)
}
}
func TestControllerChineseSettingAliasRunsSubcommands(t *testing.T) {
r := newMemoryRepo()
c := NewController(r)
if !c.ShouldHandle(context.Background(), "t", "u", "/设置 language ja") {
t.Fatal("Chinese setting alias must be handled")
}
out, err := c.Handle(context.Background(), SettingInput{TenantKey: "t", Channel: "dingtalk", StaffID: "u", Text: "/设置 language ja", ConversationType: "direct"})
if err != nil {
t.Fatal(err)
}
if r.prefs["t/u"] != "ja" || !strings.Contains(out, "ja") {
t.Fatalf("pref=%q reply=%q", r.prefs["t/u"], out)
}
}
func TestControllerParsesAdminQuestionTable(t *testing.T) {
r := newMemoryRepo()
r.admins["t/a"] = true
c := NewController(r)
_, err := c.Handle(context.Background(), SettingInput{TenantKey: "t", StaffID: "a", Text: "/setting question https://alidocs.dingtalk.com/i/nodes/base?sheetId=sheet&viewId=view", ConversationType: "direct"})
if err != nil {
t.Fatal(err)
}
if r.bindings["question_table"].BaseID != "base" {
t.Fatalf("binding=%+v", r.bindings)
}
}
func TestControllerAcceptsTableLinkInNextMessage(t *testing.T) {
r := newMemoryRepo()
r.admins["t/a"] = true
c := NewController(r)
out, err := c.Handle(context.Background(), SettingInput{TenantKey: "t", Channel: "dingtalk", StaffID: "a", Text: "/setting question", ConversationType: "direct"})
if err != nil || !strings.Contains(out, "链接") {
t.Fatalf("out=%q err=%v", out, err)
}
if !c.ShouldHandle(context.Background(), "t", "a", "https://alidocs.dingtalk.com/i/nodes/base?sheetId=sheet&viewId=view") {
t.Fatal("active setting session must intercept next message")
}
_, err = c.Handle(context.Background(), SettingInput{TenantKey: "t", Channel: "dingtalk", StaffID: "a", UnionID: "union", Text: "https://alidocs.dingtalk.com/i/nodes/base?sheetId=sheet&viewId=view", ConversationType: "direct"})
if err != nil {
t.Fatal(err)
}
if r.bindings["question_table"].BaseID != "base" {
t.Fatalf("binding=%+v", r.bindings)
}
if _, ok := r.sessions["t/a"]; ok {
t.Fatal("setting session must be cleared")
}
}
func TestControllerFAQBindingPromptExplainsPermissionsSchemaActivationAndLinkCards(t *testing.T) {
r := newMemoryRepo()
r.admins["t/a"] = true
c := NewController(r)
out, err := c.Handle(context.Background(), SettingInput{TenantKey: "t", Channel: "dingtalk", StaffID: "a", Text: "/setting faq", ConversationType: "direct"})
if err != nil {
t.Fatal(err)
}
for _, want := range []string{
"10分钟",
"链接卡片",
"机器人应用",
"管理员",
"标准问题(必填)",
"原始问题",
"语言",
"提问人",
"立即激活FAQ写入",
"短暂重启DobotAI",
"只发送 alidocs.dingtalk.com/ 后面的部分",
"i/nodes/<BaseID>?iframeQuery=entrance%3Ddata%26sheetId%3D<SheetID>%26viewId%3D<ViewID>",
"完整保留",
"取消",
} {
if !strings.Contains(out, want) {
t.Fatalf("reply=%q must contain %q", out, want)
}
}
if strings.Contains(out, "i/nodes/<BaseID>?sheetId=<SheetID>") {
t.Fatalf("reply still contains obsolete direct-query example: %q", out)
}
}
func TestControllerReportsFAQPermissionFailureAndKeepsWaitingSession(t *testing.T) {
r := newMemoryRepo()
r.admins["t/a"] = true
c := NewController(r)
if _, err := c.Handle(context.Background(), SettingInput{TenantKey: "t", StaffID: "a", Text: "/setting faq", ConversationType: "direct"}); err != nil {
t.Fatal(err)
}
r.saveErr = errors.New("schema_http_403")
out, err := c.Handle(context.Background(), SettingInput{
TenantKey: "t", StaffID: "a", UnionID: "union", ConversationType: "direct",
Text: "https://alidocs.dingtalk.com/i/nodes/base?sheetId=sheet&viewId=view",
})
if err != nil {
t.Fatalf("validation failure must become an actionable reply: %v", err)
}
for _, want := range []string{"table_permission_denied", "未激活", "机器人应用", "管理员", "重新发送链接卡片", "取消"} {
if !strings.Contains(out, want) {
t.Fatalf("reply=%q must contain %q", out, want)
}
}
if _, ok := r.sessions["t/a"]; !ok {
t.Fatal("failed validation must keep the setting session for retry")
}
if _, ok := r.bindings["faq_table"]; ok {
t.Fatal("failed validation must not save a binding")
}
}
func TestTableBindingFailureReplyUsesStableCodes(t *testing.T) {
tests := []struct {
err error
code string
}{
{errors.New("schema_http_403"), "table_permission_denied"},
{errors.New("schema_token_failed"), "table_permission_denied"},
{errors.New("missing_field"), "table_schema_missing_field"},
{errors.New("duplicate_field"), "table_schema_duplicate_field"},
{errors.New("field_type_mismatch"), "table_schema_field_type_mismatch"},
{errors.New("schema_network_failed"), "table_schema_network_failed"},
{errors.New("something private happened"), "table_validation_failed"},
}
for _, tt := range tests {
out := tableBindingFailureReply("faq", tt.err)
if !strings.Contains(out, tt.code) {
t.Fatalf("error=%q reply=%q want code=%q", tt.err, out, tt.code)
}
if tt.code == "table_validation_failed" && strings.Contains(out, tt.err.Error()) {
t.Fatalf("reply leaked raw unknown error: %q", out)
}
}
}
func TestControllerDetectsBootstrapCandidate(t *testing.T) {
r := newMemoryRepo()
c := NewController(r)
c.now = func() time.Time { return time.Unix(100, 0) }
_, err := c.Handle(context.Background(), SettingInput{TenantKey: "t", StaffID: "a", UnionID: "union", DisplayName: "Alice", Text: "/setting 123456", ConversationType: "direct"})
if err != nil {
t.Fatal(err)
}
if r.candidate.Code != "123456" || r.candidate.StaffID != "a" {
t.Fatalf("candidate=%+v", r.candidate)
}
}
+181
View File
@@ -0,0 +1,181 @@
package settings
import (
"context"
"errors"
"net/url"
"path"
"strings"
)
type SystemState string
const (
StateUninitialized SystemState = "uninitialized"
StateCredentialsSaved SystemState = "credentials_saved"
StateDingTalkVerified SystemState = "dingtalk_verified"
StateWaitingAdmin SystemState = "waiting_first_admin"
StateActive SystemState = "active"
StateCredentialError SystemState = "credential_error"
)
type Snapshot struct {
ClientID string
SecretConfigured bool
CredentialsVerified bool
AdministratorCount int
}
func DetermineState(s Snapshot) SystemState {
id := strings.TrimSpace(s.ClientID) != ""
if !id && !s.SecretConfigured {
return StateUninitialized
}
if !id || !s.SecretConfigured {
return StateCredentialError
}
if !s.CredentialsVerified {
return StateCredentialsSaved
}
if s.AdministratorCount < 1 {
return StateWaitingAdmin
}
return StateActive
}
type Command struct {
Name string
Args []string
}
func ParseCommand(raw string) (Command, bool) {
fields := strings.Fields(strings.TrimSpace(raw))
if len(fields) == 0 {
return Command{}, false
}
name := strings.ToLower(fields[0])
switch name {
case "/setting", "/设置":
default:
return Command{}, false
}
return Command{Name: name, Args: fields[1:]}, true
}
type NotableLink struct{ BaseID, SheetID, ViewID string }
func ParseNotableLink(raw string) (NotableLink, error) {
if len(raw) > 4096 {
return NotableLink{}, errors.New("AI Table link is too long")
}
input := strings.TrimSpace(raw)
u, err := url.Parse(input)
if err != nil {
return NotableLink{}, errors.New("invalid AI Table link")
}
if u.Scheme == "" {
if u.Host != "" || u.User != nil || strings.Contains(input, "\\") {
return NotableLink{}, errors.New("invalid AI Table relative path")
}
relativePath := "/" + strings.TrimPrefix(u.Path, "/")
if !strings.HasPrefix(relativePath, "/i/nodes/") || path.Clean(relativePath) != relativePath {
return NotableLink{}, errors.New("invalid AI Table relative path")
}
u = &url.URL{Scheme: "https", Host: "alidocs.dingtalk.com", Path: relativePath, RawQuery: u.RawQuery}
}
if u.Scheme != "https" {
return NotableLink{}, errors.New("invalid AI Table link")
}
host := strings.ToLower(u.Hostname())
if host != "alidocs.dingtalk.com" && host != "docs.dingtalk.com" {
return NotableLink{}, errors.New("untrusted AI Table host")
}
base := path.Base(strings.TrimRight(u.Path, "/"))
q := u.Query()
sheetID, err := uniqueNotableQueryValue(q, "sheetId")
if err != nil {
return NotableLink{}, err
}
viewID, err := uniqueNotableQueryValue(q, "viewId")
if err != nil {
return NotableLink{}, err
}
iframeQuery, err := uniqueNotableQueryValue(q, "iframeQuery")
if err != nil {
return NotableLink{}, err
}
if iframeQuery != "" {
nested, parseErr := url.ParseQuery(iframeQuery)
if parseErr != nil {
return NotableLink{}, errors.New("invalid AI Table iframeQuery")
}
nestedSheetID, valueErr := uniqueNotableQueryValue(nested, "sheetId")
if valueErr != nil {
return NotableLink{}, valueErr
}
nestedViewID, valueErr := uniqueNotableQueryValue(nested, "viewId")
if valueErr != nil {
return NotableLink{}, valueErr
}
if sheetID != "" && nestedSheetID != "" && sheetID != nestedSheetID {
return NotableLink{}, errors.New("conflicting AI Table sheet IDs")
}
if viewID != "" && nestedViewID != "" && viewID != nestedViewID {
return NotableLink{}, errors.New("conflicting AI Table view IDs")
}
if sheetID == "" {
sheetID = nestedSheetID
}
if viewID == "" {
viewID = nestedViewID
}
}
result := NotableLink{BaseID: base, SheetID: sheetID, ViewID: viewID}
if result.BaseID == "" || result.BaseID == "." || result.BaseID == ".." || result.SheetID == "" {
return NotableLink{}, errors.New("AI Table link is missing base or sheet")
}
return result, nil
}
func uniqueNotableQueryValue(values url.Values, key string) (string, error) {
items := values[key]
if len(items) > 1 {
return "", errors.New("conflicting AI Table query values")
}
if len(items) == 0 {
return "", nil
}
return strings.TrimSpace(items[0]), nil
}
type Runtime interface {
Start(context.Context, string) error
Stop(context.Context) error
}
type Switcher struct {
runtime Runtime
active string
}
func NewSwitcher(r Runtime, active string) *Switcher { return &Switcher{runtime: r, active: active} }
func (s *Switcher) Switch(ctx context.Context, candidate string) error {
if s == nil || s.runtime == nil {
return errors.New("runtime is required")
}
candidate = strings.TrimSpace(candidate)
if candidate == "" {
return errors.New("candidate is required")
}
old := s.active
if err := s.runtime.Stop(ctx); err != nil {
return err
}
if err := s.runtime.Start(ctx, candidate); err != nil {
if old != "" {
_ = s.runtime.Start(ctx, old)
}
return err
}
s.active = candidate
return nil
}
+133
View File
@@ -0,0 +1,133 @@
package settings
import (
"context"
"errors"
"strings"
"testing"
)
func TestDetermineState(t *testing.T) {
cases := []struct {
name string
in Snapshot
want SystemState
}{
{"empty", Snapshot{}, StateUninitialized},
{"partial", Snapshot{ClientID: "id"}, StateCredentialError},
{"saved", Snapshot{ClientID: "id", SecretConfigured: true}, StateCredentialsSaved},
{"verified", Snapshot{ClientID: "id", SecretConfigured: true, CredentialsVerified: true}, StateWaitingAdmin},
{"active", Snapshot{ClientID: "id", SecretConfigured: true, CredentialsVerified: true, AdministratorCount: 1}, StateActive},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
if got := DetermineState(tc.in); got != tc.want {
t.Fatalf("got %s want %s", got, tc.want)
}
})
}
}
func TestParseCommandStrict(t *testing.T) {
yes := []string{"/setting", "/设置", "/setting 123456", "/设置 123456", "/setting language ja", "/设置 language ja", "/setting support", "/设置 question https://alidocs.dingtalk.com/i/nodes/abc"}
for _, raw := range yes {
if c, ok := ParseCommand(raw); !ok || c.Name == "" {
t.Fatalf("must match %q", raw)
}
}
no := []string{"/help", "//help", "/settings", "//setting", "如何设置机器人", "abc /setting", "请解释/setting", "/settiing"}
for _, raw := range no {
if _, ok := ParseCommand(raw); ok {
t.Fatalf("must not match %q", raw)
}
}
}
func TestParseNotableLink(t *testing.T) {
got, err := ParseNotableLink("https://alidocs.dingtalk.com/i/nodes/base123?sheetId=sheet456&viewId=view789&recordId=row")
if err != nil {
t.Fatal(err)
}
if got.BaseID != "base123" || got.SheetID != "sheet456" || got.ViewID != "view789" {
t.Fatalf("got %+v", got)
}
if _, err = ParseNotableLink("http://evil.example/x?sheetId=s"); err == nil {
t.Fatal("expected rejection")
}
}
func TestParseNotableLinkAcceptsTrustedHostRelativeTail(t *testing.T) {
for _, raw := range []string{
"i/nodes/base123?sheetId=sheet456&viewId=view789&recordId=row",
"/i/nodes/base123?sheetId=sheet456&viewId=view789&recordId=row",
} {
got, err := ParseNotableLink(raw)
if err != nil {
t.Fatalf("ParseNotableLink(%q): %v", raw, err)
}
if got.BaseID != "base123" || got.SheetID != "sheet456" || got.ViewID != "view789" {
t.Fatalf("ParseNotableLink(%q)=%+v", raw, got)
}
}
}
func TestParseNotableLinkDecodesSheetAndViewFromIframeQuery(t *testing.T) {
got, err := ParseNotableLink("i/nodes/base123?iframeQuery=entrance%3Ddata%26sheetId%3Dsheet456%26viewId%3Dview789")
if err != nil {
t.Fatal(err)
}
if got.BaseID != "base123" || got.SheetID != "sheet456" || got.ViewID != "view789" {
t.Fatalf("got=%+v", got)
}
}
func TestParseNotableLinkRejectsConflictingIframeQueryIDs(t *testing.T) {
_, err := ParseNotableLink("i/nodes/base123?sheetId=sheet-top&iframeQuery=sheetId%3Dsheet-nested%26viewId%3Dview789")
if err == nil || !strings.Contains(err.Error(), "conflicting") {
t.Fatalf("err=%v", err)
}
}
func TestParseNotableLinkRejectsUnsafeOrAmbiguousRelativeTail(t *testing.T) {
for _, raw := range []string{
"//evil.example/i/nodes/base?sheetId=sheet",
"evil.example/i/nodes/base?sheetId=sheet",
"../i/nodes/base?sheetId=sheet",
"i/nodes/../base?sheetId=sheet",
"i/nodes/base",
"https://evil.example/i/nodes/base?sheetId=sheet",
} {
if _, err := ParseNotableLink(raw); err == nil {
t.Fatalf("ParseNotableLink(%q) must reject", raw)
}
}
}
type fakeRuntime struct {
active string
starts []string
fail string
}
func (f *fakeRuntime) Start(_ context.Context, id string) error {
f.starts = append(f.starts, id)
if id == f.fail {
return errors.New("boom")
}
f.active = id
return nil
}
func (f *fakeRuntime) Stop(_ context.Context) error { f.active = ""; return nil }
func TestSwitcherRollsBack(t *testing.T) {
f := &fakeRuntime{active: "old", fail: "new"}
s := NewSwitcher(f, "old")
if err := s.Switch(context.Background(), "new"); err == nil {
t.Fatal("expected error")
}
if f.active != "old" {
t.Fatalf("active=%q", f.active)
}
if len(f.starts) != 2 || f.starts[0] != "new" || f.starts[1] != "old" {
t.Fatalf("starts=%v", f.starts)
}
}
+10 -3
View File
@@ -49,10 +49,10 @@ func (s *Store) SaveCoverageDecision(ctx context.Context, d coverage.Decision) e
return fmt.Errorf("coverage case id: %w", err)
}
_, err = tx.Exec(ctx, `INSERT INTO support_cases
(id, tenant_key, channel, turn_id, coverage_decision_id, status)
VALUES($1,$2,$3,$4,$5,$6)
(id, tenant_key, channel, turn_id, coverage_decision_id, status, faq_export_status)
VALUES($1,$2,$3,$4,$5,$6,$7)
ON CONFLICT (tenant_key, channel, turn_id) DO NOTHING`,
caseID, d.TenantKey, d.Channel, d.TurnID, persistedDecisionID, "pending")
caseID, d.TenantKey, d.Channel, d.TurnID, persistedDecisionID, "pending", faqExportStatus(s.faqExportEnabled))
if err != nil {
return fmt.Errorf("coverage insert support case: %w", err)
}
@@ -64,6 +64,13 @@ func (s *Store) SaveCoverageDecision(ctx context.Context, d coverage.Decision) e
return nil
}
func faqExportStatus(enabled bool) string {
if enabled {
return "pending"
}
return "not_applicable"
}
// newUUID produces a 32-char hex id.
func newUUID() (string, error) {
var b [16]byte
@@ -127,6 +127,15 @@ func TestCoverageRepoInsertsSupportCaseOnlyForUncovered(t *testing.T) {
}
}
func TestCoverageRepoFAQExportStatusFollowsRuntimeFlag(t *testing.T) {
if got := faqExportStatus(false); got != "not_applicable" {
t.Fatalf("disabled status=%q", got)
}
if got := faqExportStatus(true); got != "pending" {
t.Fatalf("enabled status=%q", got)
}
}
func TestCoverageRepoUsesSingleTransaction(t *testing.T) {
src := readCoverageSource(t)
// Must use BeginTx
@@ -0,0 +1,54 @@
package postgres
import (
"context"
"time"
)
const operationalStatusSQL = `
SELECT
(SELECT COUNT(*) FROM question_table_exports WHERE tenant_key = $1 AND status IN ('pending','exporting')),
(SELECT COUNT(*) FROM question_table_exports WHERE tenant_key = $1 AND status = 'export_failed'),
(SELECT MAX(exported_at) FROM question_table_exports WHERE tenant_key = $1 AND status = 'exported'),
(SELECT COUNT(*) FROM support_cases WHERE tenant_key = $1 AND faq_export_status IN ('pending','exporting')),
(SELECT COUNT(*) FROM support_cases WHERE tenant_key = $1 AND faq_export_status = 'export_failed'),
(SELECT MAX(faq_exported_at) FROM support_cases WHERE tenant_key = $1 AND faq_export_status = 'exported'),
(SELECT COUNT(*) FROM support_cases WHERE tenant_key = $1 AND status IN ('pending','notifying') AND faq_export_status = 'exported'),
(SELECT COUNT(*) FROM support_cases WHERE tenant_key = $1 AND status = 'notification_failed'),
(SELECT MAX(notified_at) FROM support_cases WHERE tenant_key = $1 AND status = 'notified')`
type OperationalMetrics struct {
QuestionPending, QuestionFailed int
FAQPending, FAQFailed int
SupportPending, SupportFailed int
LastQuestionSuccess time.Time
LastFAQSuccess time.Time
LastSupportSuccess time.Time
}
func (s *Store) Ping(ctx context.Context) error {
if s == nil || s.pool == nil {
return context.Canceled
}
return s.pool.Ping(ctx)
}
func (s *Store) ReadOperationalMetrics(ctx context.Context, tenant string) (OperationalMetrics, error) {
var value OperationalMetrics
var questionSuccess, faqSuccess, supportSuccess *time.Time
err := s.pool.QueryRow(ctx, operationalStatusSQL, tenant).Scan(
&value.QuestionPending, &value.QuestionFailed, &questionSuccess,
&value.FAQPending, &value.FAQFailed, &faqSuccess,
&value.SupportPending, &value.SupportFailed, &supportSuccess,
)
if questionSuccess != nil {
value.LastQuestionSuccess = *questionSuccess
}
if faqSuccess != nil {
value.LastFAQSuccess = *faqSuccess
}
if supportSuccess != nil {
value.LastSupportSuccess = *supportSuccess
}
return value, err
}
@@ -0,0 +1,29 @@
package postgres
import (
"strings"
"testing"
)
func TestOperationalStatusSQLIsTenantScopedAndReadOnly(t *testing.T) {
upper := strings.ToUpper(operationalStatusSQL)
for _, want := range []string{
"QUESTION_TABLE_EXPORTS",
"SUPPORT_CASES",
"TENANT_KEY = $1",
"FAQ_EXPORT_STATUS",
"NOTIFICATION_FAILED",
"MAX(EXPORTED_AT)",
"MAX(FAQ_EXPORTED_AT)",
"MAX(NOTIFIED_AT)",
} {
if !strings.Contains(upper, want) {
t.Errorf("status SQL missing %q", want)
}
}
for _, forbidden := range []string{"INSERT ", "UPDATE ", "DELETE ", "FOR UPDATE"} {
if strings.Contains(upper, forbidden) {
t.Errorf("status SQL must be read-only; found %q", forbidden)
}
}
}
@@ -59,9 +59,12 @@ func TestShouldEnqueueQuestionTableExportOnlyForDingTalkSucceededAnswers(t *test
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
if got := shouldEnqueueQuestionTableExport(tc.turn); got != tc.want {
if got := shouldEnqueueQuestionTableExport(tc.turn, true); got != tc.want {
t.Fatalf("shouldEnqueueQuestionTableExport()=%v, want %v", got, tc.want)
}
if got := shouldEnqueueQuestionTableExport(tc.turn, false); got {
t.Fatal("disabled question-table export must not enqueue")
}
})
}
}
@@ -188,8 +191,8 @@ func TestFinishTurnOutboxGuardedByShouldCreateAnswerMessage(t *testing.T) {
if !strings.Contains(guardBlock, "question_table_exports") {
t.Error("question_table_exports INSERT must be inside the shouldCreateAnswerMessage guard block")
}
if !strings.Contains(guardBlock, "if shouldEnqueueQuestionTableExport(turn) {") {
t.Error("question_table_exports INSERT must be guarded to DingTalk turns explicitly")
if !strings.Contains(guardBlock, "if shouldEnqueueQuestionTableExport(turn, s.questionTableExportEnabled) {") {
t.Error("question_table_exports INSERT must require both runtime enablement and a DingTalk turn")
}
// The guard block must be the one that contains the messages INSERT
+412
View File
@@ -0,0 +1,412 @@
package postgres
import (
"context"
"crypto/rand"
"crypto/sha256"
"crypto/subtle"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"strings"
"time"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/settings"
"github.com/jackc/pgx/v5"
)
type SettingsRepository struct{ store *Store }
func NewSettingsRepository(s *Store) (*SettingsRepository, error) {
if s == nil || s.pool == nil {
return nil, errors.New("settings repository: store is required")
}
return &SettingsRepository{store: s}, nil
}
func (r *SettingsRepository) IsAdministrator(ctx context.Context, tenant, staff string) (bool, error) {
var ok bool
err := r.store.pool.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM administrators WHERE tenant_key=$1 AND staff_id=$2 AND status='active')`, strings.TrimSpace(tenant), strings.TrimSpace(staff)).Scan(&ok)
return ok, err
}
func (r *SettingsRepository) SetPreference(ctx context.Context, tenant, channel, staff, locale string) error {
mode := "fixed"
if locale == "auto" {
mode = "auto"
}
_, err := r.store.pool.Exec(ctx, `INSERT INTO user_preferences(tenant_key,channel,external_actor_id,language_mode,locale) VALUES($1,$2,$3,$4,$5) ON CONFLICT(tenant_key,channel,external_actor_id) DO UPDATE SET language_mode=EXCLUDED.language_mode,locale=EXCLUDED.locale,updated_at=NOW()`, tenant, channel, staff, mode, locale)
return err
}
func (r *SettingsRepository) GetPreference(ctx context.Context, tenant, channel, staff string) (string, error) {
var locale string
err := r.store.pool.QueryRow(ctx, `SELECT locale FROM user_preferences WHERE tenant_key=$1 AND channel=$2 AND external_actor_id=$3`, tenant, channel, staff).Scan(&locale)
if errors.Is(err, pgx.ErrNoRows) {
return "auto", nil
}
return locale, err
}
func (r *SettingsRepository) GetSettingSession(ctx context.Context, tenant, staff string) (settings.SettingSession, bool, error) {
var session settings.SettingSession
err := r.store.pool.QueryRow(ctx, `SELECT tenant_key,staff_id,conversation_id,conversation_type,pending_action,expires_at FROM setting_sessions WHERE tenant_key=$1 AND staff_id=$2 AND expires_at>NOW()`, tenant, staff).Scan(&session.TenantKey, &session.StaffID, &session.ConversationID, &session.ConversationType, &session.PendingAction, &session.ExpiresAt)
if errors.Is(err, pgx.ErrNoRows) {
return settings.SettingSession{}, false, nil
}
return session, err == nil, err
}
func (r *SettingsRepository) SaveSettingSession(ctx context.Context, session settings.SettingSession) error {
_, err := r.store.pool.Exec(ctx, `INSERT INTO setting_sessions(tenant_key,staff_id,conversation_id,conversation_type,current_step,pending_action,draft_config,expires_at) VALUES($1,$2,$3,$4,'waiting_input',$5,'{}'::JSONB,$6) ON CONFLICT(tenant_key,staff_id) DO UPDATE SET conversation_id=EXCLUDED.conversation_id,conversation_type=EXCLUDED.conversation_type,current_step='waiting_input',pending_action=EXCLUDED.pending_action,draft_config='{}'::JSONB,expires_at=EXCLUDED.expires_at,updated_at=NOW()`, session.TenantKey, session.StaffID, session.ConversationID, session.ConversationType, session.PendingAction, session.ExpiresAt)
return err
}
func (r *SettingsRepository) DeleteSettingSession(ctx context.Context, tenant, staff string) error {
_, err := r.store.pool.Exec(ctx, `DELETE FROM setting_sessions WHERE tenant_key=$1 AND staff_id=$2`, tenant, staff)
return err
}
func (r *SettingsRepository) SaveBinding(ctx context.Context, b settings.ResourceBinding) error {
payload, err := json.Marshal(map[string]interface{}{"conversation_id": b.ConversationID, "conversation_title": b.ConversationTitle, "base_id": b.BaseID, "sheet_id": b.SheetID, "view_id": b.ViewID, "operator_id": b.OperatorID, "field_mapping": b.FieldMapping})
if err != nil {
return err
}
tx, err := r.store.pool.Begin(ctx)
if err != nil {
return err
}
defer tx.Rollback(ctx)
var rev int
if err = tx.QueryRow(ctx, `SELECT COALESCE(MAX(revision),0)+1 FROM system_resource_bindings WHERE tenant_key=$1 AND kind=$2`, b.TenantKey, b.Kind).Scan(&rev); err != nil {
return err
}
state := "validated"
var validatedAt, activatedAt interface{}
if b.Activate {
state = "active"
validatedAt, activatedAt = time.Now(), time.Now()
if _, err = tx.Exec(ctx, `UPDATE system_resource_bindings SET state='retired' WHERE tenant_key=$1 AND kind=$2 AND state='active'`, b.TenantKey, b.Kind); err != nil {
return err
}
}
validationCode := strings.TrimSpace(b.ValidationCode)
if validationCode == "" {
validationCode = "pending_external_probe"
}
_, err = tx.Exec(ctx, `INSERT INTO system_resource_bindings(id,tenant_key,kind,revision,state,public_config,validation_code,validated_at,activated_at,created_by_staff_id) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10)`, newSettingsID(), b.TenantKey, b.Kind, rev, state, payload, validationCode, validatedAt, activatedAt, b.CreatedByStaffID)
if err != nil {
return err
}
return tx.Commit(ctx)
}
func (r *SettingsRepository) DetectBootstrapCandidate(ctx context.Context, c settings.BootstrapCandidate) error {
hash := sha256.Sum256([]byte(strings.TrimSpace(c.Code)))
var id, stored string
var attempts, max int
err := r.store.pool.QueryRow(ctx, `SELECT id,code_hash,attempts,max_attempts FROM admin_bootstrap_challenges WHERE tenant_key=$1 AND status='pending' AND expires_at>NOW() ORDER BY created_at DESC LIMIT 1`, c.TenantKey).Scan(&id, &stored, &attempts, &max)
if err != nil {
return err
}
if attempts >= max {
return errors.New("bootstrap challenge locked")
}
provided := hex.EncodeToString(hash[:])
if len(provided) != len(stored) || subtle.ConstantTimeCompare([]byte(provided), []byte(stored)) != 1 {
_, _ = r.store.pool.Exec(ctx, `UPDATE admin_bootstrap_challenges SET attempts=attempts+1,status=CASE WHEN attempts+1>=max_attempts THEN 'locked' ELSE status END,updated_at=NOW() WHERE id=$1`, id)
return errors.New("invalid bootstrap code")
}
_, err = r.store.pool.Exec(ctx, `UPDATE admin_bootstrap_challenges SET candidate_staff_id=$2,candidate_union_id=$3,candidate_display_name=$4,candidate_department=$5,status='candidate_detected',updated_at=NOW() WHERE id=$1`, id, c.StaffID, c.UnionID, c.DisplayName, c.Department)
return err
}
func (r *SettingsRepository) CreateBootstrapChallenge(ctx context.Context, tenant, code string, expires time.Time) error {
sum := sha256.Sum256([]byte(code))
tx, err := r.store.pool.Begin(ctx)
if err != nil {
return err
}
defer tx.Rollback(ctx)
if _, err = tx.Exec(ctx, `UPDATE admin_bootstrap_challenges SET status='expired',updated_at=NOW() WHERE tenant_key=$1 AND status IN ('pending','candidate_detected')`, tenant); err != nil {
return err
}
_, err = tx.Exec(ctx, `INSERT INTO admin_bootstrap_challenges(id,tenant_key,code_hash,expires_at,status) VALUES($1,$2,$3,$4,'pending')`, newSettingsID(), tenant, hex.EncodeToString(sum[:]), expires)
if err != nil {
return err
}
return tx.Commit(ctx)
}
func (r *SettingsRepository) ConfirmBootstrap(ctx context.Context, tenant string) error {
tx, err := r.store.pool.Begin(ctx)
if err != nil {
return err
}
defer tx.Rollback(ctx)
var challengeID, staff, union, name, dept string
err = tx.QueryRow(ctx, `SELECT id,candidate_staff_id,candidate_union_id,candidate_display_name,candidate_department FROM admin_bootstrap_challenges WHERE tenant_key=$1 AND status='candidate_detected' AND expires_at>NOW() FOR UPDATE`, tenant).Scan(&challengeID, &staff, &union, &name, &dept)
if err != nil {
return err
}
if strings.TrimSpace(staff) == "" {
return errors.New("bootstrap candidate missing")
}
_, err = tx.Exec(ctx, `INSERT INTO administrators(id,tenant_key,staff_id,union_id,display_name,department,role,status) VALUES($1,$2,$3,$4,$5,$6,'super_admin','active') ON CONFLICT(tenant_key,staff_id) DO UPDATE SET status='active',role='super_admin',updated_at=NOW()`, newSettingsID(), tenant, staff, union, name, dept)
if err != nil {
return err
}
if _, err = tx.Exec(ctx, `UPDATE admin_bootstrap_challenges SET status='confirmed',updated_at=NOW() WHERE id=$1`, challengeID); err != nil {
return err
}
return tx.Commit(ctx)
}
func (r *SettingsRepository) AdministratorCount(ctx context.Context, tenant string) (int, error) {
var n int
err := r.store.pool.QueryRow(ctx, `SELECT COUNT(*) FROM administrators WHERE tenant_key=$1 AND status='active'`, tenant).Scan(&n)
return n, err
}
func (r *SettingsRepository) ListAdministrators(ctx context.Context, tenant string) ([]settings.Administrator, error) {
rows, err := r.store.pool.Query(ctx, `SELECT staff_id,display_name,department,role,status FROM administrators WHERE tenant_key=$1 ORDER BY CASE status WHEN 'active' THEN 0 ELSE 1 END,created_at`, strings.TrimSpace(tenant))
if err != nil {
return nil, err
}
defer rows.Close()
administrators := []settings.Administrator{}
for rows.Next() {
var administrator settings.Administrator
if err = rows.Scan(&administrator.StaffID, &administrator.DisplayName, &administrator.Department, &administrator.Role, &administrator.Status); err != nil {
return nil, err
}
administrators = append(administrators, administrator)
}
return administrators, rows.Err()
}
func (r *SettingsRepository) DisableAdministrator(ctx context.Context, tenant, staffID string) error {
tenant, staffID = strings.TrimSpace(tenant), strings.TrimSpace(staffID)
if tenant == "" || staffID == "" {
return settings.ErrAdministratorNotFound
}
tx, err := r.store.pool.Begin(ctx)
if err != nil {
return err
}
defer tx.Rollback(ctx)
rows, err := tx.Query(ctx, `SELECT staff_id,status FROM administrators WHERE tenant_key=$1 FOR UPDATE`, tenant)
if err != nil {
return err
}
activeCount, targetStatus, found := 0, "", false
for rows.Next() {
var currentStaffID, status string
if err = rows.Scan(&currentStaffID, &status); err != nil {
rows.Close()
return err
}
if status == "active" {
activeCount++
}
if currentStaffID == staffID {
targetStatus, found = status, true
}
}
rows.Close()
if err = rows.Err(); err != nil {
return err
}
if !found {
return settings.ErrAdministratorNotFound
}
if err = canDisableAdministrator(targetStatus, activeCount); err != nil {
return err
}
result, err := tx.Exec(ctx, `UPDATE administrators SET status='disabled',updated_at=NOW() WHERE tenant_key=$1 AND staff_id=$2 AND status='active'`, tenant, staffID)
if err != nil {
return err
}
if result.RowsAffected() != 1 {
return settings.ErrAdministratorNotFound
}
if _, err = tx.Exec(ctx, `INSERT INTO config_audit_events(id,tenant_key,actor_staff_id,action,target_kind,target_revision,result_code,metadata) VALUES($1,$2,'','administrator.disable','administrator',0,'administrator_disabled',jsonb_build_object('target_staff_id',$3,'source','admin_web'))`, newSettingsID(), tenant, staffID); err != nil {
return err
}
return tx.Commit(ctx)
}
func canDisableAdministrator(status string, activeCount int) error {
if status != "active" {
return settings.ErrAdministratorAlreadyDisabled
}
if activeCount <= 1 {
return settings.ErrLastAdministrator
}
return nil
}
func (r *SettingsRepository) SaveDingTalkConfig(ctx context.Context, tenant, clientID, secretRef string, verified bool) error {
payload, _ := json.Marshal(map[string]string{"client_id": clientID})
state := "validated"
validation := "ok"
if verified {
state = "active"
}
tx, err := r.store.pool.Begin(ctx)
if err != nil {
return err
}
defer tx.Rollback(ctx)
var rev int
if err = tx.QueryRow(ctx, `SELECT COALESCE(MAX(revision),0)+1 FROM system_config_revisions WHERE tenant_key=$1 AND kind='dingtalk_app'`, tenant).Scan(&rev); err != nil {
return err
}
if state == "active" {
if _, err = tx.Exec(ctx, `UPDATE system_config_revisions SET state='retired' WHERE tenant_key=$1 AND kind='dingtalk_app' AND state='active'`, tenant); err != nil {
return err
}
}
_, err = tx.Exec(ctx, `INSERT INTO system_config_revisions(id,tenant_key,kind,revision,state,public_config,secret_ref,validation_status,validation_code,validated_at,activated_at) VALUES($1,$2,'dingtalk_app',$3,$4,$5,$6,$7,$7,CASE WHEN $7='ok' THEN NOW() END,CASE WHEN $7='ok' THEN NOW() END)`, newSettingsID(), tenant, rev, state, payload, secretRef, validation)
if err != nil {
return err
}
return tx.Commit(ctx)
}
func (r *SettingsRepository) CandidateDingTalkConfig(ctx context.Context, tenant string) (clientID, secretRef string, err error) {
var payload []byte
err = r.store.pool.QueryRow(ctx, `SELECT public_config,secret_ref FROM system_config_revisions WHERE tenant_key=$1 AND kind='dingtalk_app' AND state='validated' AND validation_status='ok' ORDER BY revision DESC LIMIT 1`, tenant).Scan(&payload, &secretRef)
if err != nil {
return
}
var value map[string]string
if json.Unmarshal(payload, &value) != nil {
return "", "", errors.New("decode dingtalk candidate")
}
return value["client_id"], secretRef, nil
}
func (r *SettingsRepository) ActivateDingTalkCandidate(ctx context.Context, tenant, secretRef string) error {
tx, err := r.store.pool.Begin(ctx)
if err != nil {
return err
}
defer tx.Rollback(ctx)
var candidateID string
err = tx.QueryRow(ctx, `SELECT id FROM system_config_revisions WHERE tenant_key=$1 AND kind='dingtalk_app' AND state='validated' AND secret_ref=$2 AND validation_status='ok' FOR UPDATE`, tenant, secretRef).Scan(&candidateID)
if err != nil {
return err
}
if _, err = tx.Exec(ctx, `UPDATE system_config_revisions SET state='retired' WHERE tenant_key=$1 AND kind='dingtalk_app' AND state='active'`, tenant); err != nil {
return err
}
if _, err = tx.Exec(ctx, `UPDATE system_config_revisions SET state='active',validation_code='pending_runtime',activated_at=NOW() WHERE id=$1`, candidateID); err != nil {
return err
}
return tx.Commit(ctx)
}
func (r *SettingsRepository) PrepareDingTalkRuntimeStart(ctx context.Context, tenant string) (bool, error) {
tx, err := r.store.pool.Begin(ctx)
if err != nil {
return false, err
}
defer tx.Rollback(ctx)
var activeID, activeRef, code string
err = tx.QueryRow(ctx, `SELECT id,secret_ref,validation_code FROM system_config_revisions WHERE tenant_key=$1 AND kind='dingtalk_app' AND state='active' FOR UPDATE`, tenant).Scan(&activeID, &activeRef, &code)
if errors.Is(err, pgx.ErrNoRows) {
return false, tx.Commit(ctx)
}
if err != nil {
return false, err
}
switch code {
case "pending_runtime":
if _, err = tx.Exec(ctx, `UPDATE system_config_revisions SET validation_code='runtime_starting' WHERE id=$1`, activeID); err != nil {
return false, err
}
return true, tx.Commit(ctx)
case "runtime_starting":
var oldID string
err = tx.QueryRow(ctx, `SELECT id FROM system_config_revisions WHERE tenant_key=$1 AND kind='dingtalk_app' AND state='retired' AND validation_status='ok' ORDER BY revision DESC LIMIT 1 FOR UPDATE`, tenant).Scan(&oldID)
if err != nil && !errors.Is(err, pgx.ErrNoRows) {
return false, err
}
if _, err = tx.Exec(ctx, `UPDATE system_config_revisions SET state='failed',validation_status='failed',validation_code='runtime_start_failed' WHERE id=$1`, activeID); err != nil {
return false, err
}
if oldID != "" {
if _, err = tx.Exec(ctx, `UPDATE system_config_revisions SET state='active',validation_code='rollback_restored',activated_at=NOW() WHERE id=$1`, oldID); err != nil {
return false, err
}
}
return false, tx.Commit(ctx)
default:
_ = activeRef
return false, tx.Commit(ctx)
}
}
func (r *SettingsRepository) MarkDingTalkRuntimeHealthy(ctx context.Context, tenant, secretRef string) error {
_, err := r.store.pool.Exec(ctx, `UPDATE system_config_revisions SET validation_code='runtime_ok' WHERE tenant_key=$1 AND kind='dingtalk_app' AND state='active' AND secret_ref=$2`, tenant, secretRef)
return err
}
func (r *SettingsRepository) RollbackPendingDingTalkActivation(ctx context.Context, tenant, failedRef string) error {
tx, err := r.store.pool.Begin(ctx)
if err != nil {
return err
}
defer tx.Rollback(ctx)
var currentID string
if err = tx.QueryRow(ctx, `SELECT id FROM system_config_revisions WHERE tenant_key=$1 AND kind='dingtalk_app' AND state='active' AND secret_ref=$2 AND validation_code='pending_runtime' FOR UPDATE`, tenant, failedRef).Scan(&currentID); err != nil {
return err
}
var oldID string
if err = tx.QueryRow(ctx, `SELECT id FROM system_config_revisions WHERE tenant_key=$1 AND kind='dingtalk_app' AND state='retired' AND validation_status='ok' ORDER BY revision DESC LIMIT 1 FOR UPDATE`, tenant).Scan(&oldID); err != nil {
return err
}
if _, err = tx.Exec(ctx, `UPDATE system_config_revisions SET state='failed',validation_status='failed',validation_code='runtime_start_failed' WHERE id=$1`, currentID); err != nil {
return err
}
if _, err = tx.Exec(ctx, `UPDATE system_config_revisions SET state='active',validation_code='rollback_restored',activated_at=NOW() WHERE id=$1`, oldID); err != nil {
return err
}
return tx.Commit(ctx)
}
func (r *SettingsRepository) MarkDingTalkCandidateFailed(ctx context.Context, tenant, secretRef, code string) error {
_, err := r.store.pool.Exec(ctx, `UPDATE system_config_revisions SET state='failed',validation_status='failed',validation_code=$3 WHERE tenant_key=$1 AND kind='dingtalk_app' AND state='validated' AND secret_ref=$2`, tenant, secretRef, code)
return err
}
func (r *SettingsRepository) ActiveDingTalkConfig(ctx context.Context, tenant string) (clientID, secretRef string, verified bool, err error) {
var payload []byte
var status string
err = r.store.pool.QueryRow(ctx, `SELECT public_config,secret_ref,validation_status FROM system_config_revisions WHERE tenant_key=$1 AND kind='dingtalk_app' AND state='active' ORDER BY revision DESC LIMIT 1`, tenant).Scan(&payload, &secretRef, &status)
if errors.Is(err, pgx.ErrNoRows) {
err = nil
return
}
if err != nil {
return
}
var v map[string]string
if json.Unmarshal(payload, &v) != nil {
err = fmt.Errorf("decode dingtalk config")
return
}
clientID = v["client_id"]
verified = status == "ok"
return
}
func (r *SettingsRepository) ActiveBindings(ctx context.Context, tenant string) (map[string]settings.ResourceBinding, error) {
rows, err := r.store.pool.Query(ctx, `SELECT kind,public_config FROM system_resource_bindings WHERE tenant_key=$1 AND state='active'`, tenant)
if err != nil {
return nil, err
}
defer rows.Close()
out := map[string]settings.ResourceBinding{}
for rows.Next() {
var kind string
var raw []byte
if err = rows.Scan(&kind, &raw); err != nil {
return nil, err
}
var v struct {
ConversationID string `json:"conversation_id"`
ConversationTitle string `json:"conversation_title"`
BaseID string `json:"base_id"`
SheetID string `json:"sheet_id"`
ViewID string `json:"view_id"`
OperatorID string `json:"operator_id"`
FieldMapping map[string]string `json:"field_mapping"`
}
if json.Unmarshal(raw, &v) != nil {
continue
}
out[kind] = settings.ResourceBinding{Kind: kind, TenantKey: tenant, ConversationID: v.ConversationID, ConversationTitle: v.ConversationTitle, BaseID: v.BaseID, SheetID: v.SheetID, ViewID: v.ViewID, OperatorID: v.OperatorID, FieldMapping: v.FieldMapping, Activate: true}
}
return out, rows.Err()
}
func newSettingsID() string { b := make([]byte, 16); _, _ = rand.Read(b); return hex.EncodeToString(b) }
@@ -0,0 +1,28 @@
package postgres
import (
"errors"
"testing"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/settings"
)
func TestCanDisableAdministratorProtectsLastActiveAdministrator(t *testing.T) {
for _, tc := range []struct {
name string
status string
activeCount int
wantErr error
}{
{name: "two active", status: "active", activeCount: 2},
{name: "last active", status: "active", activeCount: 1, wantErr: settings.ErrLastAdministrator},
{name: "already disabled", status: "disabled", activeCount: 2, wantErr: settings.ErrAdministratorAlreadyDisabled},
} {
t.Run(tc.name, func(t *testing.T) {
err := canDisableAdministrator(tc.status, tc.activeCount)
if !errors.Is(err, tc.wantErr) {
t.Fatalf("err=%v want=%v", err, tc.wantErr)
}
})
}
}
+21 -4
View File
@@ -13,7 +13,19 @@ import (
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/core"
)
type Store struct{ pool *pgxpool.Pool }
type Store struct {
pool *pgxpool.Pool
questionTableExportEnabled bool
faqExportEnabled bool
}
func (s *Store) SetExternalTableExports(questionEnabled, faqEnabled bool) {
if s == nil {
return
}
s.questionTableExportEnabled = questionEnabled
s.faqExportEnabled = faqEnabled
}
func Open(ctx context.Context, databaseURL string) (*Store, error) {
if databaseURL == "" {
@@ -176,6 +188,11 @@ func (s *Store) SyncSessionTitle(ctx context.Context, mapping core.SessionMappin
return tx.Commit(ctx)
}
func (s *Store) InvalidateSession(ctx context.Context, sessionKey, expectedSessionID string) error {
_, err := s.pool.Exec(ctx, `DELETE FROM session_mappings WHERE session_key=$1 AND weknora_session_id=$2`, sessionKey, expectedSessionID)
return err
}
func (s *Store) FinishTurn(ctx context.Context, turn core.Turn) error {
references, err := json.Marshal(turn.References)
if err != nil {
@@ -208,7 +225,7 @@ func (s *Store) FinishTurn(ctx context.Context, turn core.Turn) error {
}
answerMessageID = turn.AnswerMessageID
if shouldEnqueueQuestionTableExport(turn) {
if shouldEnqueueQuestionTableExport(turn, s.questionTableExportEnabled) {
exportID, err := newUUID()
if err != nil {
return err
@@ -361,8 +378,8 @@ func shouldCreateAnswerMessage(turn core.Turn) bool {
strings.TrimSpace(turn.AnswerMessageID) != ""
}
func shouldEnqueueQuestionTableExport(turn core.Turn) bool {
return shouldCreateAnswerMessage(turn) && strings.EqualFold(strings.TrimSpace(turn.Channel), "dingtalk")
func shouldEnqueueQuestionTableExport(turn core.Turn, enabled bool) bool {
return enabled && shouldCreateAnswerMessage(turn) && strings.EqualFold(strings.TrimSpace(turn.Channel), "dingtalk")
}
func validateDeliveryUpdate(update core.DeliveryUpdate) error {
+20
View File
@@ -18,6 +18,7 @@ import (
var ErrIncompleteStream = errors.New("weknora stream ended without a terminal event")
var ErrEmptyAnswer = errors.New("weknora stream completed with an empty answer")
var ErrSessionNotFound = errors.New("weknora session not found")
// AnswerObserver is called with the accumulated full answer content after each
// answer chunk. The observer is purely a notification; its return is ignored.
@@ -153,11 +154,30 @@ func (c *Client) agentChat(ctx context.Context, sessionID string, request AgentC
defer resp.Body.Close()
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
message, _ := io.ReadAll(io.LimitReader(resp.Body, 32<<10))
if isSessionNotFound(resp.StatusCode, message) {
return ChatResult{}, fmt.Errorf("%w: HTTP %d", ErrSessionNotFound, resp.StatusCode)
}
return ChatResult{}, fmt.Errorf("weknora agent chat returned HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(message)))
}
return parseSSEWithEventObserver(resp.Body, answerObserver, eventObserver)
}
func isSessionNotFound(status int, body []byte) bool {
if status != http.StatusNotFound {
return false
}
var response struct {
Error struct {
Code int `json:"code"`
Message string `json:"message"`
} `json:"error"`
}
if json.Unmarshal(body, &response) != nil {
return false
}
return response.Error.Code == 1003 && strings.EqualFold(strings.TrimSpace(response.Error.Message), "Session not found")
}
func (c *Client) doJSON(ctx context.Context, method, path string, input, output any) error {
body, err := json.Marshal(input)
if err != nil {
+17
View File
@@ -11,6 +11,23 @@ import (
"time"
)
func TestClientClassifiesMissingSession(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusNotFound)
_, _ = w.Write([]byte(`{"error":{"code":1003,"message":"Session not found"},"success":false}`))
}))
defer server.Close()
client, err := NewClient(server.URL, "test-key", 5*time.Second)
if err != nil {
t.Fatal(err)
}
_, err = client.AgentChat(context.Background(), "stale-session", AgentChatRequest{Query: "q", AgentID: "agent-1"})
if !errors.Is(err, ErrSessionNotFound) {
t.Fatalf("expected ErrSessionNotFound, got %v", err)
}
}
func TestClientCreateSessionAndAgentChat(t *testing.T) {
mux := http.NewServeMux()
mux.HandleFunc("/api/v1/sessions", func(w http.ResponseWriter, r *http.Request) {
+20 -4
View File
@@ -1,18 +1,30 @@
#!/usr/bin/env bash
set -euo pipefail
root="/home/zheromiku/dobotai-platform"
script_dir="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
root="${DOBOTAI_SOURCE_ROOT:-$(cd "$script_dir/.." && pwd)}"
go_bin="${GO_BIN:-/home/zheromiku/.local/share/go1.26.5/bin/go}"
bin_dir="$root/bin"
current="$bin_dir/dobotai"
next="$bin_dir/dobotai.new"
previous="$bin_dir/dobotai.prev"
service="dobotai.service"
version_url="http://127.0.0.1:8090/version"
if [[ -n "$(git -C "$root" status --porcelain)" ]]; then
echo "DobotAI source worktree is dirty; refusing unverifiable build" >&2
exit 1
fi
commit_sha="$(git -C "$root" rev-parse HEAD)"
version="${DOBOTAI_VERSION:-dev}"
build_time="$(date -u '+%Y-%m-%dT%H:%M:%SZ')"
ldflags="-X main.version=$version -X main.commitSHA=$commit_sha -X main.buildTime=$build_time"
mkdir -p "$bin_dir"
cd "$root"
"$go_bin" build -trimpath -o "$next" ./cmd/dobotai
"$go_bin" build -trimpath -ldflags "$ldflags" -o "$next" ./cmd/dobotai
chmod 0755 "$next"
if [[ -f "$current" ]]; then
@@ -24,8 +36,12 @@ systemctl --user restart "$service"
for _ in $(seq 1 30); do
if curl --noproxy '*' --fail --silent --max-time 2 http://127.0.0.1:8090/healthz >/dev/null; then
echo "DOBOTAI_DEPLOY_OK"
exit 0
deployed_sha="$(curl --noproxy '*' --fail --silent --max-time 2 "$version_url" | python3 -c 'import json,sys; print(json.load(sys.stdin).get("commit_sha", ""))' 2>/dev/null || true)"
if [[ "$deployed_sha" == "$commit_sha" ]]; then
echo "DOBOTAI_DEPLOY_OK"
echo "DOBOTAI_COMMIT_SHA=$commit_sha"
exit 0
fi
fi
sleep 1
done
+32
View File
@@ -0,0 +1,32 @@
package scripts
import (
"os"
"strings"
"testing"
)
func TestDeployLocalBuildsItsSelectedSourceAndVerifiesVersion(t *testing.T) {
data, err := os.ReadFile("deploy-local.sh")
if err != nil {
t.Fatal(err)
}
source := string(data)
if strings.Contains(source, `root="/home/zheromiku/dobotai-platform"`) {
t.Fatal("deploy script must not hard-code the production checkout")
}
for _, fragment := range []string{
`BASH_SOURCE[0]`,
`DOBOTAI_SOURCE_ROOT`,
`git -C "$root" status --porcelain`,
`git -C "$root" rev-parse HEAD`,
`main.version`,
`main.commitSHA`,
`main.buildTime`,
`http://127.0.0.1:8090/version`,
} {
if !strings.Contains(source, fragment) {
t.Errorf("deploy script missing %q", fragment)
}
}
}