diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..8ff8507 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,21 @@ +# 多阶段构建：产出可直接跑的静态风格二进制（仍动态链接 libc 即可） +FROM golang:1.23-bookworm AS build +WORKDIR /src +ENV GOPROXY=https://goproxy.cn,direct +COPY go.mod go.sum ./ +RUN go mod download +COPY . . +RUN CGO_ENABLED=0 go build -ldflags="-s -w" -o /out/eightbf-analysis-go ./cmd/server + +FROM debian:bookworm-slim +RUN apt-get update && apt-get install -y --no-install-recommends ca-certificates \ + && rm -rf /var/lib/apt/lists/* \ + && mkdir -p /app /var/log/eightbf-analysis +WORKDIR /app +COPY --from=build /out/eightbf-analysis-go /app/eightbf-analysis-go +COPY schema.sql /app/schema.sql +ENV EBA_SCHEMA_PATH=/app/schema.sql \ + SERVER_PORT=18097 \ + LOG_LEVEL=info +EXPOSE 18097 +ENTRYPOINT ["/app/eightbf-analysis-go"] diff --git a/README.md b/README.md index 166451c..775d1f6 100644 --- a/README.md +++ b/README.md @@ -45,30 +45,38 @@ go build -o bin/eightbf-analysis-go ./cmd/server ## 接口（与 Java 一致） -- `GET /api/v1/analysis/health` +- `GET /api/v1/analysis/health` — 存活 +- `GET /api/v1/analysis/ready` — 就绪（含 MySQL ping，失败 503） - `POST /api/v1/analysis/jobs` - `GET /api/v1/analysis/jobs/{id}` - `GET /api/v1/analysis/jobs?taskId=&latest=` - `POST /api/v1/analysis/jobs/{id}/cancel|resume|plan|inject` - `GET /api/v1/analysis/jobs/{id}/result` -身份头：`X-Account-Id`、`X-Analysis-Role`。 +身份头：`X-Account-Id`、`X-Analysis-Role`。响应头带回 `X-Request-Id`。 ## 目录 ```text -cmd/server 入口 -internal/api HTTP +cmd/server 入口（优雅停机） +internal/api Gin 路由 + 审计日志 + 中间件 +internal/logging zap / 滚动文件 internal/service 编排 / 鉴权投影 / DemoWorker internal/worker LiveWorker / LiveChainRunner / 提示词 internal/deerflow 登录、SSE、模型校验 -internal/repo MySQL +internal/repo sqlx + MySQL internal/domain 作业实体与状态常量 -internal/config 环境变量 -docs/ 设计要求（自 Java 同步）+ env.sample +internal/config Viper 读环境变量 + 启动校验 +deploy/ systemd unit +docs/ 设计要求、运维、env.sample +Dockerfile ``` -## 日志 +## 日志与运维 -遵循设计要求 §11：业务日志带 `jobId=`；禁止打印 token/密码/完整 dialogues。 -当前写 stdout，正式环境用 systemd/Docker 收集或挂滚动文件。 +见 [docs/运维.md](./docs/运维.md)。要点： + +- `LOG_LEVEL` / `LOG_FILE`（滚动，默认保留 14 天） +- 写操作 INFO 审计：`jobId` / `accountId` / `reqId` +- `EBA_STRICT_CONFIG=true` 禁止 live 默认弱口令启动 +- SIGTERM 优雅停机；systemd 见 `deploy/eightbf-analysis-go.service` diff --git a/cmd/server/main.go b/cmd/server/main.go index c06af05..d267b06 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -1,18 +1,27 @@ +// 会商后台进程入口。 +// +// 启动顺序不能颠倒：先日志（后面的失败才看得到）→ 校验配置 → 连库建表 → 挂 Gin → 按模式起一个 Worker → 听信号停机。 +// 边界：live 与 demo 只起一个 Worker。两进程不要同时消费同一库的 queued。 package main import ( - "database/sql" - "log" + "context" "net/http" "os" + "os/signal" "strconv" + "strings" + "syscall" "time" + "github.com/gin-gonic/gin" _ "github.com/go-sql-driver/mysql" + "github.com/jmoiron/sqlx" "git.gozeros.com/wsk/eightbf-analysis-go/internal/api" "git.gozeros.com/wsk/eightbf-analysis-go/internal/config" "git.gozeros.com/wsk/eightbf-analysis-go/internal/deerflow" + "git.gozeros.com/wsk/eightbf-analysis-go/internal/logging" "git.gozeros.com/wsk/eightbf-analysis-go/internal/repo" "git.gozeros.com/wsk/eightbf-analysis-go/internal/service" "git.gozeros.com/wsk/eightbf-analysis-go/internal/worker" @@ -20,59 +29,120 @@ import ( func main() { cfg := config.Load() - log.Printf("eightbf-analysis-go starting port=%d mode=%s auth=%v demo=%v", - cfg.Port, cfg.AnalysisMode, cfg.AuthEnabled, cfg.DemoEnabled) + closeLog := logging.Setup(logging.Options{ + Level: cfg.LogLevel, + File: cfg.LogFile, + MaxMB: cfg.LogMaxMB, + MaxAge: cfg.LogMaxAgeDays, + MaxBak: cfg.LogMaxBackups, + }) + defer closeLog() - db, err := sql.Open("mysql", cfg.DBDSN) + warns, err := cfg.Validate() if err != nil { - log.Fatalf("open db: %v", err) + logging.L.Errorw("配置校验失败", "err", err) + os.Exit(1) + } + for _, w := range warns { + logging.L.Warnw("配置告警", "msg", w) + } + + logging.L.Infow("eightbf-analysis-go 启动", + "version", cfg.Version, + "port", cfg.Port, + "mode", cfg.AnalysisMode, + "auth", cfg.AuthEnabled, + "demo", cfg.DemoEnabled, + "logFile", cfg.LogFile, + ) + + db, err := sqlx.Open("mysql", cfg.DBDSN) + if err != nil { + logging.L.Errorw("打开数据库失败", "err", err) + os.Exit(1) } db.SetMaxOpenConns(20) db.SetMaxIdleConns(5) db.SetConnMaxLifetime(time.Hour) if err := db.Ping(); err != nil { - log.Fatalf("ping db: %v（请检查 EBA_DB_DSN / EBA_DB_*）", err) + logging.L.Errorw("数据库不可达", "err", err) + os.Exit(1) } + defer db.Close() jobRepo := repo.NewJobRepo(db) - if schema, err := os.ReadFile("schema.sql"); err != nil { - log.Printf("未找到 schema.sql，跳过自动建表: %v", err) + if schema, err := os.ReadFile(cfg.SchemaPath); err != nil { + logging.L.Warnw("未找到 schema，跳过自动建表", "path", cfg.SchemaPath, "err", err) } else { jobRepo.ApplySchema(string(schema)) } access := &service.AccessService{AuthEnabled: cfg.AuthEnabled} jobs := &service.JobService{Repo: jobRepo, Access: access} - h := &api.Handler{Jobs: jobs} + h := &api.Handler{ + Jobs: jobs, + DB: db, + Version: cfg.Version, + Started: time.Now(), + } - mux := http.NewServeMux() - h.Routes(mux) + // Release 模式关掉 Gin 自带的逐条访问日志。作业页会轮询 GET，那些行对排障没有帮助。 + if strings.EqualFold(cfg.LogLevel, "debug") { + gin.SetMode(gin.DebugMode) + } else { + gin.SetMode(gin.ReleaseMode) + } + engine := gin.New() + engine.Use(api.RequestID(), api.CORS(), api.Recovery()) + h.Register(engine) if cfg.AnalysisMode == "live" && !cfg.DemoEnabled { auth := deerflow.NewAuth(cfg) df := deerflow.NewClient(cfg, auth) runner := &worker.LiveChainRunner{DF: df, Repo: jobRepo, Coord: cfg.DeerFlowCoordinator} lw := &worker.LiveWorker{ - Repo: jobRepo, - Runner: runner, - PollEvery: time.Duration(cfg.LivePollMS) * time.Millisecond, + Repo: jobRepo, + Runner: runner, + PollEvery: time.Duration(cfg.LivePollMS) * time.Millisecond, + MaxInflight: int32(cfg.LiveMaxInflight), } lw.Start() - log.Printf("LiveWorker 已启动 poll=%dms", cfg.LivePollMS) + logging.L.Infow("LiveWorker 已启动", "pollMs", cfg.LivePollMS, "maxInflight", cfg.LiveMaxInflight) } else if cfg.DemoEnabled || cfg.AnalysisMode == "demo" { dw := &service.DemoWorker{ Repo: jobRepo, Every: time.Duration(cfg.DemoTickMS) * time.Millisecond, } dw.Start() - log.Printf("DemoWorker 已启动 tick=%dms", cfg.DemoTickMS) + logging.L.Infow("DemoWorker 已启动", "tickMs", cfg.DemoTickMS) } else { - log.Printf("警告：未启动任何 Worker（mode=%s demo=%v）", cfg.AnalysisMode, cfg.DemoEnabled) + logging.L.Warnw("未启动任何 Worker", "mode", cfg.AnalysisMode, "demo", cfg.DemoEnabled) } addr := ":" + strconv.Itoa(cfg.Port) - log.Printf("监听 %s （健康检查 GET /api/v1/analysis/health）", addr) - if err := http.ListenAndServe(addr, api.WithCORS(mux)); err != nil { - log.Fatal(err) + srv := &http.Server{ + Addr: addr, + Handler: engine, + ReadHeaderTimeout: 10 * time.Second, } + + go func() { + logging.L.Infow("HTTP 监听中", "addr", addr) + if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed { + logging.L.Errorw("HTTP 服务异常退出", "err", err) + os.Exit(1) + } + }() + + // 优雅停机：停接收新连接，给进行中请求一点时间（Runner 仍可能在后台跑完当前席）。 + stop := make(chan os.Signal, 1) + signal.Notify(stop, syscall.SIGINT, syscall.SIGTERM) + sig := <-stop + logging.L.Infow("收到退出信号，开始优雅停机", "signal", sig.String()) + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + if err := srv.Shutdown(ctx); err != nil { + logging.L.Warnw("优雅停机超时或失败", "err", err) + } + logging.L.Infow("服务已退出") } diff --git a/deploy/eightbf-analysis-go.service b/deploy/eightbf-analysis-go.service new file mode 100644 index 0000000..24b5a95 --- /dev/null +++ b/deploy/eightbf-analysis-go.service @@ -0,0 +1,32 @@ +[Unit] +Description=eightbf-analysis-go (8BF 多岗位会商后台) +After=network-online.target mysql.service +Wants=network-online.target + +[Service] +Type=simple +User=root +WorkingDirectory=/opt/eightbf-analysis-go +EnvironmentFile=-/etc/eightbf-analysis-go.env +ExecStart=/opt/eightbf-analysis-go/bin/eightbf-analysis-go +Restart=on-failure +RestartSec=3 +# 优雅退出：先 SIGTERM，等 20s +TimeoutStopSec=20 +KillMode=mixed + +# 建议在 EnvironmentFile 中设置： +# SERVER_PORT=18097 +# ANALYSIS_MODE=live +# ANALYSIS_AUTH_ENABLED=true +# EBA_STRICT_CONFIG=true +# EBA_DB_DSN=... +# EBA_DF_BASE_URL=... +# EBA_DF_USER=... +# EBA_DF_PASSWORD=... +# LOG_LEVEL=info +# LOG_FILE=/var/log/eightbf-analysis/app-go.log +# EBA_SCHEMA_PATH=/opt/eightbf-analysis-go/schema.sql + +[Install] +WantedBy=multi-user.target diff --git a/docs/env.sample b/docs/env.sample index 7b9de08..0707386 100644 --- a/docs/env.sample +++ b/docs/env.sample @@ -1,24 +1,38 @@ -# eightbf-analysis / eightbf-analysis-go 内网环境变量样例 +# eightbf-analysis-go 环境变量样例（export 或 systemd EnvironmentFile） # 勿把真实密码提交进 Git。 -# MySQL（正式必填） -# Java 可用 JDBC；Go 优先 EBA_DB_DSN，也能解析 EBA_DB_URL(jdbc) -EBA_DB_URL=jdbc:mysql://:3306/eightbf_analysis?useUnicode=true&characterEncoding=utf8&useSSL=false&serverTimezone=Asia/Shanghai&allowPublicKeyRetrieval=true -EBA_DB_USER=eba -EBA_DB_PASSWORD= -# Go 推荐： -# EBA_DB_DSN=eba:xxx@tcp(mysql-host:3306)/eightbf_analysis?parseTime=true&loc=Asia%2FShanghai&charset=utf8mb4&multiStatements=true +# —— 服务 —— +SERVER_PORT=18097 +ANALYSIS_MODE=live +ANALYSIS_AUTH_ENABLED=false +EBA_STRICT_CONFIG=false +EBA_VERSION=0.1.0 +EBA_SCHEMA_PATH=schema.sql -# DeerFlow Gateway（ANALYSIS_MODE=live 时必填） -EBA_DF_BASE_URL=http://:8001 +# —— 日志（设计要求 §11）—— +LOG_LEVEL=info +# LOG_FILE=/var/log/eightbf-analysis/app-go.log +LOG_MAX_MB=100 +LOG_MAX_AGE_DAYS=14 +LOG_MAX_BACKUPS=14 + +# —— MySQL —— +# Go 推荐 DSN；也可 EBA_DB_URL(jdbc) + USER/PASSWORD +EBA_DB_DSN=eba:xxx@tcp(127.0.0.1:3306)/eightbf_analysis?parseTime=true&loc=Asia%2FShanghai&charset=utf8mb4&multiStatements=true +# EBA_DB_URL=jdbc:mysql://127.0.0.1:3306/eightbf_analysis?... +# EBA_DB_USER=eba +# EBA_DB_PASSWORD= + +# —— DeerFlow（live 必填）—— +EBA_DF_BASE_URL=http://127.0.0.1:8001 EBA_DF_USER=admin EBA_DF_PASSWORD= +EBA_DF_COORDINATOR=roundtable-coordinator +EBA_DF_STREAM_IDLE_TIMEOUT_MS=180000 -# 可选：改端口 -# SERVER_PORT=18097 - -# Go 运行模式 -# ANALYSIS_MODE=live -# ANALYSIS_AUTH_ENABLED=false +# —— Worker —— +EBA_WORKER_LIVE_POLL_MS=1000 +EBA_WORKER_LIVE_MAX=2 +EBA_WORKER_TICK_MS=400 # EBA_WORKER_DEMO_ENABLED=false # 假进度：ANALYSIS_MODE=demo diff --git a/docs/对齐说明.md b/docs/对齐说明.md index 5f83acb..f90a6f2 100644 --- a/docs/对齐说明.md +++ b/docs/对齐说明.md @@ -17,7 +17,7 @@ 1. **进程互斥**：勿与 Java 实例同时抢同一 MySQL 的 `queued`（尚未做多实例领取锁）。 2. **schema 启动**：Go 启动读工作目录 `schema.sql`；部署时保证 cwd 或改绝对路径。 3. **SSE 细节**：DeerFlow 帧格式若有变，优先对照 Java `DeerFlowClient` 再改本仓 `internal/deerflow`。 -4. **日志**：先 stdout；滚动文件按设计要求 §11 后续加 lumberjack 即可。 +4. **日志 / 就绪 / 停机**：zap + 可选 `LOG_FILE` 滚动、`/ready`、SIGTERM 优雅停机；HTTP 为 Gin。详见 [运维.md](./运维.md)。 ## 切换建议 diff --git a/docs/运维.md b/docs/运维.md new file mode 100644 index 0000000..cdf3f7e --- /dev/null +++ b/docs/运维.md @@ -0,0 +1,46 @@ +# eightbf-analysis-go 运维要点 + +对齐 [服务端开发设计要求.md](./服务端开发设计要求.md) §9–§11。 + +## 健康检查 + +| 路径 | 含义 | +|------|------| +| `GET /api/v1/analysis/health` | 存活：进程在听即可 | +| `GET /api/v1/analysis/ready` | 就绪：至少 MySQL `Ping` 成功，否则 **503** | + +Nginx / 编排探活建议：liveness → health；readiness → ready。 + +## 日志 + +| 变量 | 说明 | +|------|------| +| `LOG_LEVEL` | debug / info / warn / error，默认 info | +| `LOG_FILE` | 非空则 **stdout + 滚动文件**（zap + lumberjack，控制台键值格式） | +| `LOG_MAX_MB` | 单文件上限 MB，默认 100 | +| `LOG_MAX_AGE_DAYS` | 保留天数，默认 14 | +| `LOG_MAX_BACKUPS` | 备份份数，默认 14 | + +业务键：`jobId=` `requestId=` `accountId=` `reqId=` `status=` `phase=` `model=`。 +不写 token / 密码 / 完整 dialogues。 + +采集：systemd journal 或扫 `LOG_FILE` 目录 → Filebeat/Vector → Loki/ELK。 + +## 配置校验 + +- 启动时 `Validate()`：非法 mode/端口、空 DSN 直接退出。 +- `EBA_STRICT_CONFIG=true`：live 仍用联调默认 DeerFlow 密码则 **拒绝启动**。 +- 非 strict 仅 WARN。 + +## 优雅停机 + +捕获 SIGINT/SIGTERM → `http.Server.Shutdown`（最长 15s）。 +进行中的 LiveRunner 可能仍在跑当前席；取消靠库内 `cancel_flag`。 + +## systemd + +见 `deploy/eightbf-analysis-go.service`，环境文件建议 `/etc/eightbf-analysis-go.env`（勿提交 Git）。 + +## 单活约束 + +与 Java 实例、或第二份 Go 进程 **不要** 同时抢同一 MySQL 的 `queued`（尚无领取互斥）。 diff --git a/go.mod b/go.mod index c6af3d9..2355d2a 100644 --- a/go.mod +++ b/go.mod @@ -2,6 +2,50 @@ module git.gozeros.com/wsk/eightbf-analysis-go go 1.26.5 -require github.com/go-sql-driver/mysql v1.8.1 +require ( + github.com/gin-gonic/gin v1.10.0 + github.com/go-sql-driver/mysql v1.8.1 + github.com/jmoiron/sqlx v1.4.0 + github.com/spf13/viper v1.20.1 + go.uber.org/zap v1.27.0 + gopkg.in/natefinch/lumberjack.v2 v2.2.1 +) -require filippo.io/edwards25519 v1.1.0 // indirect +require ( + filippo.io/edwards25519 v1.1.0 // indirect + github.com/bytedance/sonic v1.11.6 // indirect + github.com/bytedance/sonic/loader v0.1.1 // indirect + github.com/cloudwego/base64x v0.1.4 // indirect + github.com/cloudwego/iasm v0.2.0 // indirect + github.com/fsnotify/fsnotify v1.8.0 // indirect + github.com/gabriel-vasile/mimetype v1.4.3 // indirect + github.com/gin-contrib/sse v0.1.0 // indirect + github.com/go-playground/locales v0.14.1 // indirect + github.com/go-playground/universal-translator v0.18.1 // indirect + github.com/go-playground/validator/v10 v10.20.0 // indirect + github.com/go-viper/mapstructure/v2 v2.2.1 // indirect + github.com/goccy/go-json v0.10.2 // indirect + github.com/json-iterator/go v1.1.12 // indirect + github.com/klauspost/cpuid/v2 v2.2.7 // indirect + github.com/leodido/go-urn v1.4.0 // indirect + github.com/mattn/go-isatty v0.0.20 // indirect + github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect + github.com/modern-go/reflect2 v1.0.2 // indirect + github.com/pelletier/go-toml/v2 v2.2.3 // indirect + github.com/sagikazarmark/locafero v0.7.0 // indirect + github.com/sourcegraph/conc v0.3.0 // indirect + github.com/spf13/afero v1.12.0 // indirect + github.com/spf13/cast v1.7.1 // indirect + github.com/spf13/pflag v1.0.6 // indirect + github.com/subosito/gotenv v1.6.0 // indirect + github.com/twitchyliquid64/golang-asm v0.15.1 // indirect + github.com/ugorji/go/codec v1.2.12 // indirect + go.uber.org/multierr v1.10.0 // indirect + golang.org/x/arch v0.8.0 // indirect + golang.org/x/crypto v0.32.0 // indirect + golang.org/x/net v0.33.0 // indirect + golang.org/x/sys v0.29.0 // indirect + golang.org/x/text v0.21.0 // indirect + google.golang.org/protobuf v1.36.1 // indirect + gopkg.in/yaml.v3 v3.0.1 // indirect +) diff --git a/go.sum b/go.sum index 19dbcec..5a451c4 100644 --- a/go.sum +++ b/go.sum @@ -1,4 +1,130 @@ filippo.io/edwards25519 v1.1.0 h1:FNf4tywRC1HmFuKW5xopWpigGjJKiJSV0Cqo0cJWDaA= filippo.io/edwards25519 v1.1.0/go.mod h1:BxyFTGdWcka3PhytdK4V28tE5sGfRvvvRV7EaN4VDT4= +github.com/bytedance/sonic v1.11.6 h1:oUp34TzMlL+OY1OUWxHqsdkgC/Zfc85zGqw9siXjrc0= +github.com/bytedance/sonic v1.11.6/go.mod h1:LysEHSvpvDySVdC2f87zGWf6CIKJcAvqab1ZaiQtds4= +github.com/bytedance/sonic/loader v0.1.1 h1:c+e5Pt1k/cy5wMveRDyk2X4B9hF4g7an8N3zCYjJFNM= +github.com/bytedance/sonic/loader v0.1.1/go.mod h1:ncP89zfokxS5LZrJxl5z0UJcsk4M4yY2JpfqGeCtNLU= +github.com/cloudwego/base64x v0.1.4 h1:jwCgWpFanWmN8xoIUHa2rtzmkd5J2plF/dnLS6Xd/0Y= +github.com/cloudwego/base64x v0.1.4/go.mod h1:0zlkT4Wn5C6NdauXdJRhSKRlJvmclQ1hhJgA0rcu/8w= +github.com/cloudwego/iasm v0.2.0 h1:1KNIy1I1H9hNNFEEH3DVnI4UujN+1zjpuk6gwHLTssg= +github.com/cloudwego/iasm v0.2.0/go.mod h1:8rXZaNYT2n95jn+zTI1sDr+IgcD2GVs0nlbbQPiEFhY= +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= +github.com/frankban/quicktest v1.14.6 h1:7Xjx+VpznH+oBnejlPUj8oUpdxnVs4f8XU8WnHkI4W8= +github.com/frankban/quicktest v1.14.6/go.mod h1:4ptaffx2x8+WTWXmUCuVU6aPUX1/Mz7zb5vbUoiM6w0= +github.com/fsnotify/fsnotify v1.8.0 h1:dAwr6QBTBZIkG8roQaJjGof0pp0EeF+tNV7YBP3F/8M= +github.com/fsnotify/fsnotify v1.8.0/go.mod h1:8jBTzvmWwFyi3Pb8djgCCO5IBqzKJ/Jwo8TRcHyHii0= +github.com/gabriel-vasile/mimetype v1.4.3 h1:in2uUcidCuFcDKtdcBxlR0rJ1+fsokWf+uqxgUFjbI0= +github.com/gabriel-vasile/mimetype v1.4.3/go.mod h1:d8uq/6HKRL6CGdk+aubisF/M5GcPfT7nKyLpA0lbSSk= +github.com/gin-contrib/sse v0.1.0 h1:Y/yl/+YNO8GZSjAhjMsSuLt29uWRFHdHYUb5lYOV9qE= +github.com/gin-contrib/sse v0.1.0/go.mod h1:RHrZQHXnP2xjPF+u1gW/2HnVO7nvIa9PG3Gm+fLHvGI= +github.com/gin-gonic/gin v1.10.0 h1:nTuyha1TYqgedzytsKYqna+DfLos46nTv2ygFy86HFU= +github.com/gin-gonic/gin v1.10.0/go.mod h1:4PMNQiOhvDRa013RKVbsiNwoyezlm2rm0uX/T7kzp5Y= +github.com/go-playground/assert/v2 v2.2.0 h1:JvknZsQTYeFEAhQwI4qEt9cyV5ONwRHC+lYKSsYSR8s= +github.com/go-playground/assert/v2 v2.2.0/go.mod h1:VDjEfimB/XKnb+ZQfWdccd7VUvScMdVu0Titje2rxJ4= +github.com/go-playground/locales v0.14.1 h1:EWaQ/wswjilfKLTECiXz7Rh+3BjFhfDFKv/oXslEjJA= +github.com/go-playground/locales v0.14.1/go.mod h1:hxrqLVvrK65+Rwrd5Fc6F2O76J/NuW9t0sjnWqG1slY= +github.com/go-playground/universal-translator v0.18.1 h1:Bcnm0ZwsGyWbCzImXv+pAJnYK9S473LQFuzCbDbfSFY= +github.com/go-playground/universal-translator v0.18.1/go.mod h1:xekY+UJKNuX9WP91TpwSH2VMlDf28Uj24BCp08ZFTUY= +github.com/go-playground/validator/v10 v10.20.0 h1:K9ISHbSaI0lyB2eWMPJo+kOS/FBExVwjEviJTixqxL8= +github.com/go-playground/validator/v10 v10.20.0/go.mod h1:dbuPbCMFw/DrkbEynArYaCwl3amGuJotoKCe95atGMM= github.com/go-sql-driver/mysql v1.8.1 h1:LedoTUt/eveggdHS9qUFC1EFSa8bU2+1pZjSRpvNJ1Y= github.com/go-sql-driver/mysql v1.8.1/go.mod h1:wEBSXgmK//2ZFJyE+qWnIsVGmvmEKlqwuVSjsCm7DZg= +github.com/go-viper/mapstructure/v2 v2.2.1 h1:ZAaOCxANMuZx5RCeg0mBdEZk7DZasvvZIxtHqx8aGss= +github.com/go-viper/mapstructure/v2 v2.2.1/go.mod h1:oJDH3BJKyqBA2TXFhDsKDGDTlndYOZ6rGS0BRZIxGhM= +github.com/goccy/go-json v0.10.2 h1:CrxCmQqYDkv1z7lO7Wbh2HN93uovUHgrECaO5ZrCXAU= +github.com/goccy/go-json v0.10.2/go.mod h1:6MelG93GURQebXPDq3khkgXZkazVtN9CRI+MGFi0w8I= +github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI= +github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY= +github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg= +github.com/jmoiron/sqlx v1.4.0 h1:1PLqN7S1UYp5t4SrVVnt4nUVNemrDAtxlulVe+Qgm3o= +github.com/jmoiron/sqlx v1.4.0/go.mod h1:ZrZ7UsYB/weZdl2Bxg6jCRO9c3YHl8r3ahlKmRT4JLY= +github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM= +github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo= +github.com/klauspost/cpuid/v2 v2.0.9/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg= +github.com/klauspost/cpuid/v2 v2.2.7 h1:ZWSB3igEs+d0qvnxR/ZBzXVmxkgt8DdzP6m9pfuVLDM= +github.com/klauspost/cpuid/v2 v2.2.7/go.mod h1:Lcz8mBdAVJIBVzewtcLocK12l3Y+JytZYpaMropDUws= +github.com/knz/go-libedit v1.10.1/go.mod h1:MZTVkCWyz0oBc7JOWP3wNAzd002ZbM/5hgShxwh4x8M= +github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= +github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= +github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= +github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= +github.com/leodido/go-urn v1.4.0 h1:WT9HwE9SGECu3lg4d/dIA+jxlljEa1/ffXKmRjqdmIQ= +github.com/leodido/go-urn v1.4.0/go.mod h1:bvxc+MVxLKB4z00jd1z+Dvzr47oO32F/QSNjSBOlFxI= +github.com/lib/pq v1.10.9 h1:YXG7RB+JIjhP29X+OtkiDnYaXQwpS4JEWq7dtCCRUEw= +github.com/lib/pq v1.10.9/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o= +github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY= +github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y= +github.com/mattn/go-sqlite3 v1.14.22 h1:2gZY6PC6kBnID23Tichd1K+Z0oS6nE/XwU+Vz/5o4kU= +github.com/mattn/go-sqlite3 v1.14.22/go.mod h1:Uh1q+B4BYcTPb+yiD3kU8Ct7aC0hY9fxUwlHK0RXw+Y= +github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= +github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg= +github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= +github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9Gz0M= +github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk= +github.com/pelletier/go-toml/v2 v2.2.3 h1:YmeHyLY8mFWbdkNWwpr+qIL2bEqT0o95WSdkNHvL12M= +github.com/pelletier/go-toml/v2 v2.2.3/go.mod h1:MfCQTFTvCcUyyvvwm1+G6H/jORL20Xlb6rzQu9GuUkc= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/rogpeppe/go-internal v1.9.0 h1:73kH8U+JUqXU8lRuOHeVHaa/SZPifC7BkcraZVejAe8= +github.com/rogpeppe/go-internal v1.9.0/go.mod h1:WtVeX8xhTBvf0smdhujwtBcq4Qrzq/fJaraNFVN+nFs= +github.com/sagikazarmark/locafero v0.7.0 h1:5MqpDsTGNDhY8sGp0Aowyf0qKsPrhewaLSsFaodPcyo= +github.com/sagikazarmark/locafero v0.7.0/go.mod h1:2za3Cg5rMaTMoG/2Ulr9AwtFaIppKXTRYnozin4aB5k= +github.com/sourcegraph/conc v0.3.0 h1:OQTbbt6P72L20UqAkXXuLOj79LfEanQ+YQFNpLA9ySo= +github.com/sourcegraph/conc v0.3.0/go.mod h1:Sdozi7LEKbFPqYX2/J+iBAM6HpqSLTASQIKqDmF7Mt0= +github.com/spf13/afero v1.12.0 h1:UcOPyRBYczmFn6yvphxkn9ZEOY65cpwGKb5mL36mrqs= +github.com/spf13/afero v1.12.0/go.mod h1:ZTlWwG4/ahT8W7T0WQ5uYmjI9duaLQGy3Q2OAl4sk/4= +github.com/spf13/cast v1.7.1 h1:cuNEagBQEHWN1FnbGEjCXL2szYEXqfJPbP2HNUaca9Y= +github.com/spf13/cast v1.7.1/go.mod h1:ancEpBxwJDODSW/UG4rDrAqiKolqNNh2DX3mk86cAdo= +github.com/spf13/pflag v1.0.6 h1:jFzHGLGAlb3ruxLB8MhbI6A8+AQX/2eW4qeyNZXNp2o= +github.com/spf13/pflag v1.0.6/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg= +github.com/spf13/viper v1.20.1 h1:ZMi+z/lvLyPSCoNtFCpqjy0S4kPbirhpTMwl8BkW9X4= +github.com/spf13/viper v1.20.1/go.mod h1:P9Mdzt1zoHIG8m2eZQinpiBjo6kCmZSKBClNNqjJvu4= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw= +github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo= +github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= +github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= +github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= +github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU= +github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4= +github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA= +github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= +github.com/subosito/gotenv v1.6.0 h1:9NlTDc1FTs4qu0DDq7AEtTPNw6SVm7uBMsUCUjABIf8= +github.com/subosito/gotenv v1.6.0/go.mod h1:Dk4QP5c2W3ibzajGcXpNraDfq2IrhjMIvMSWPKKo0FU= +github.com/twitchyliquid64/golang-asm v0.15.1 h1:SU5vSMR7hnwNxj24w34ZyCi/FmDZTkS4MhqMhdFk5YI= +github.com/twitchyliquid64/golang-asm v0.15.1/go.mod h1:a1lVb/DtPvCB8fslRZhAngC2+aY1QWCk3Cedj/Gdt08= +github.com/ugorji/go/codec v1.2.12 h1:9LC83zGrHhuUA9l16C9AHXAqEV/2wBQ4nkvumAE65EE= +github.com/ugorji/go/codec v1.2.12/go.mod h1:UNopzCgEMSXjBc6AOMqYvWC1ktqTAfzJZUZgYf6w6lg= +go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= +go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= +go.uber.org/multierr v1.10.0 h1:S0h4aNzvfcFsC3dRF1jLoaov7oRaKqRGC/pUEJ2yvPQ= +go.uber.org/multierr v1.10.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y= +go.uber.org/zap v1.27.0 h1:aJMhYGrd5QSmlpLMr2MftRKl7t8J8PTZPA732ud/XR8= +go.uber.org/zap v1.27.0/go.mod h1:GB2qFLM7cTU87MWRP2mPIjqfIDnGu+VIO4V/SdhGo2E= +golang.org/x/arch v0.0.0-20210923205945-b76863e36670/go.mod h1:5om86z9Hs0C8fWVUuoMHwpExlXzs5Tkyp9hOrfG7pp8= +golang.org/x/arch v0.8.0 h1:3wRIsP3pM4yUptoR96otTUOXI367OS0+c9eeRi9doIc= +golang.org/x/arch v0.8.0/go.mod h1:FEVrYAQjsQXMVJ1nsMoVVXPZg6p2JE2mx8psSWTDQys= +golang.org/x/crypto v0.32.0 h1:euUpcYgM8WcP71gNpTqQCn6rC2t6ULUPiOzfWaXVVfc= +golang.org/x/crypto v0.32.0/go.mod h1:ZnnJkOaASj8g0AjIduWNlq2NRxL0PlBrbKVyZ6V/Ugc= +golang.org/x/net v0.33.0 h1:74SYHlV8BIgHIFC/LrYkOGIwL19eTYXQ5wc6TBuO36I= +golang.org/x/net v0.33.0/go.mod h1:HXLR5J+9DxmrqMwG9qjGCxZ+zKXxBru04zlTvWlWuN4= +golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.29.0 h1:TPYlXGxvx1MGTn2GiZDhnjPA9wZzZeGKHHmKhHYvgaU= +golang.org/x/sys v0.29.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= +golang.org/x/text v0.21.0 h1:zyQAAkrwaneQ066sspRyJaG9VNi/YJ1NfzcGB3hZ/qo= +golang.org/x/text v0.21.0/go.mod h1:4IBbMaMmOPCJ8SecivzSH54+73PCFmPWxNTLm+vZkEQ= +google.golang.org/protobuf v1.36.1 h1:yBPeRvTftaleIgM3PZ/WBIZ7XM/eEYAaEyCwvyjq/gk= +google.golang.org/protobuf v1.36.1/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 h1:YR8cESwS4TdDjEe65xsg0ogRM/Nc3DYOhEAlW+xobZo= +gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/natefinch/lumberjack.v2 v2.2.1 h1:bBRl1b0OH9s/DuPhuXpNl+VtCaJXFZ5/uEFST95x9zc= +gopkg.in/natefinch/lumberjack.v2 v2.2.1/go.mod h1:YD8tP3GAjkrDg1eZH7EGmyESg/lsYskCTPBJVb9jqSc= +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= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +nullprogram.com/x/optparse v1.0.0/go.mod h1:KdyPE+Igbe0jQUrVfMqDMeJQIJZEuyV7pjYmp6pbG50= +rsc.io/pdf v0.1.1/go.mod h1:n8OzWcQ6Sp37PL01nO98y4iUCRdTGarVfzxY20ICaU4= diff --git a/internal/api/handler.go b/internal/api/handler.go index 2f12ae3..365bb45 100644 --- a/internal/api/handler.go +++ b/internal/api/handler.go @@ -3,153 +3,250 @@ package api import ( "encoding/json" "errors" + "io" "net/http" "strings" + "time" + "github.com/gin-gonic/gin" + "github.com/jmoiron/sqlx" + + "git.gozeros.com/wsk/eightbf-analysis-go/internal/logging" "git.gozeros.com/wsk/eightbf-analysis-go/internal/service" ) // Handler HTTP 入口，路径与 Java AnalysisJobController 对齐。 +// +// 为什么用 Gin：路由、参数、中间件是这层的全部职责，业务判断仍在 JobService。 +// 约定：成功业务体是 {code:0,message,data}；health/ready 是给探活的扁平 JSON，不要套这层外壳。 type Handler struct { - Jobs *service.JobService + Jobs *service.JobService + DB *sqlx.DB + Version string + Started time.Time } -func (h *Handler) Routes(mux *http.ServeMux) { - mux.HandleFunc("GET /api/v1/analysis/health", h.health) - mux.HandleFunc("POST /api/v1/analysis/jobs", h.create) - mux.HandleFunc("GET /api/v1/analysis/jobs/{id}", h.get) - mux.HandleFunc("GET /api/v1/analysis/jobs", h.getByTask) - mux.HandleFunc("POST /api/v1/analysis/jobs/{id}/cancel", h.cancel) - mux.HandleFunc("POST /api/v1/analysis/jobs/{id}/resume", h.resume) - mux.HandleFunc("POST /api/v1/analysis/jobs/{id}/plan", h.patchPlan) - mux.HandleFunc("POST /api/v1/analysis/jobs/{id}/inject", h.inject) - mux.HandleFunc("GET /api/v1/analysis/jobs/{id}/result", h.result) +// Register 挂上 /api/v1/analysis 下的全部路径。方法+路径与 Java 保持一致，前端不用改。 +func (h *Handler) Register(r *gin.Engine) { + g := r.Group("/api/v1/analysis") + g.GET("/health", h.health) + g.GET("/ready", h.ready) + g.POST("/jobs", h.create) + g.GET("/jobs/:id", h.get) + g.GET("/jobs", h.getByTask) + g.POST("/jobs/:id/cancel", h.cancel) + g.POST("/jobs/:id/resume", h.resume) + g.POST("/jobs/:id/plan", h.patchPlan) + g.POST("/jobs/:id/inject", h.inject) + g.GET("/jobs/:id/result", h.result) } -func (h *Handler) caller(r *http.Request) service.Caller { - return service.NewCaller(r.Header.Get("X-Account-Id"), r.Header.Get("X-Analysis-Role")) +func (h *Handler) caller(c *gin.Context) service.Caller { + return service.NewCaller(c.GetHeader("X-Account-Id"), c.GetHeader("X-Analysis-Role")) } -func (h *Handler) health(w http.ResponseWriter, _ *http.Request) { - writeJSON(w, http.StatusOK, map[string]interface{}{ +// health 存活探活：进程在听即可，不查依赖（给进程托管用）。 +func (h *Handler) health(c *gin.Context) { + c.JSON(http.StatusOK, gin.H{ "ok": true, "service": "eightbf-analysis-go", + "version": h.Version, + "uptime": time.Since(h.Started).Round(time.Second).String(), }) } -func (h *Handler) create(w http.ResponseWriter, r *http.Request) { +// ready 就绪：至少能 ping MySQL；失败返回 503，便于摘流量。 +func (h *Handler) ready(c *gin.Context) { + body := gin.H{ + "ok": true, + "service": "eightbf-analysis-go", + "checks": map[string]string{}, + } + checks := body["checks"].(map[string]string) + if h.DB == nil { + checks["db"] = "not_configured" + body["ok"] = false + c.JSON(http.StatusServiceUnavailable, body) + return + } + if err := h.DB.PingContext(c.Request.Context()); err != nil { + checks["db"] = "fail" + body["ok"] = false + body["error"] = err.Error() + logging.L.Errorw("就绪检查失败", "check", "db", "err", err) + c.JSON(http.StatusServiceUnavailable, body) + return + } + checks["db"] = "ok" + c.JSON(http.StatusOK, body) +} + +// create 新建一场会商。空 body 允许（字段都有服务端缺省）；非法 JSON 直接 400，不再当成空请求吞掉。 +func (h *Handler) create(c *gin.Context) { var body service.CreateJobRequest - _ = json.NewDecoder(r.Body).Decode(&body) - job, err := h.Jobs.Create(&body, h.caller(r)) - if err != nil { - h.mapErr(w, err) + if err := bindBody(c, &body); err != nil { + Fail(c, http.StatusBadRequest, 400, "请求体不是合法 JSON") return } - Created(w, job) + caller := h.caller(c) + job, err := h.Jobs.Create(&body, caller) + if err != nil { + h.mapErr(c, err) + return + } + logging.L.Infow("作业创建", + logging.KeyReqID, RequestIDFrom(c), + logging.KeyJobID, job.ID, + logging.KeyRequestID, job.RequestID, + logging.KeyTaskID, job.TaskID, + logging.KeyAccountID, caller.AccountID, + logging.KeyStatus, job.Status, + ) + Created(c, job) } -func (h *Handler) get(w http.ResponseWriter, r *http.Request) { - id := r.PathValue("id") - job, err := h.Jobs.Get(id, h.caller(r)) +func (h *Handler) get(c *gin.Context) { + job, err := h.Jobs.Get(c.Param("id"), h.caller(c)) if err != nil { - h.mapErr(w, err) + h.mapErr(c, err) return } if job == nil { - Fail(w, http.StatusNotFound, 404, "job not found") + Fail(c, http.StatusNotFound, 404, "job not found") return } - OK(w, job) + OK(c, job) } -func (h *Handler) getByTask(w http.ResponseWriter, r *http.Request) { - taskID := strings.TrimSpace(r.URL.Query().Get("taskId")) +// getByTask 用 taskId 找作业。latest=true 只取最近一场；否则先找还在跑的，没有再退回最近一场。 +func (h *Handler) getByTask(c *gin.Context) { + taskID := strings.TrimSpace(c.Query("taskId")) if taskID == "" { - Fail(w, http.StatusBadRequest, 400, "taskId required") + Fail(c, http.StatusBadRequest, 400, "taskId required") return } - latest := r.URL.Query().Get("latest") == "true" - job, err := h.Jobs.GetByTaskID(taskID, latest, h.caller(r)) + latest := c.Query("latest") == "true" + job, err := h.Jobs.GetByTaskID(taskID, latest, h.caller(c)) if err != nil { - h.mapErr(w, err) + h.mapErr(c, err) return } if job == nil { - Fail(w, http.StatusNotFound, 404, "job not found for taskId") + Fail(c, http.StatusNotFound, 404, "job not found for taskId") return } - OK(w, job) + OK(c, job) } -func (h *Handler) cancel(w http.ResponseWriter, r *http.Request) { - job, err := h.Jobs.Cancel(r.PathValue("id"), h.caller(r)) +func (h *Handler) cancel(c *gin.Context) { + caller := h.caller(c) + job, err := h.Jobs.Cancel(c.Param("id"), caller) if err != nil { - h.mapErr(w, err) + h.mapErr(c, err) return } if job == nil { - Fail(w, http.StatusNotFound, 404, "job not found") + Fail(c, http.StatusNotFound, 404, "job not found") return } - OK(w, job) + logging.L.Infow("作业取消", + logging.KeyReqID, RequestIDFrom(c), + logging.KeyJobID, job.ID, + logging.KeyAccountID, caller.AccountID, + logging.KeyStatus, job.Status, + ) + OK(c, job) } -func (h *Handler) resume(w http.ResponseWriter, r *http.Request) { +func (h *Handler) resume(c *gin.Context) { var body service.ResumeJobRequest - _ = json.NewDecoder(r.Body).Decode(&body) - job, err := h.Jobs.Resume(r.PathValue("id"), &body, h.caller(r)) + if err := bindBody(c, &body); err != nil { + Fail(c, http.StatusBadRequest, 400, "请求体不是合法 JSON") + return + } + caller := h.caller(c) + job, err := h.Jobs.Resume(c.Param("id"), &body, caller) if err != nil { - h.mapErr(w, err) + h.mapErr(c, err) return } if job == nil { - Fail(w, http.StatusNotFound, 404, "job not found") + Fail(c, http.StatusNotFound, 404, "job not found") return } - OK(w, job) + logging.L.Infow("作业续跑", + logging.KeyReqID, RequestIDFrom(c), + logging.KeyJobID, job.ID, + logging.KeyAccountID, caller.AccountID, + logging.KeyStatus, job.Status, + logging.KeyPhase, job.Phase, + ) + OK(c, job) } -func (h *Handler) patchPlan(w http.ResponseWriter, r *http.Request) { +func (h *Handler) patchPlan(c *gin.Context) { var body service.PatchJobPlanRequest - _ = json.NewDecoder(r.Body).Decode(&body) - job, err := h.Jobs.PatchPlan(r.PathValue("id"), &body, h.caller(r)) + if err := bindBody(c, &body); err != nil { + Fail(c, http.StatusBadRequest, 400, "请求体不是合法 JSON") + return + } + caller := h.caller(c) + job, err := h.Jobs.PatchPlan(c.Param("id"), &body, caller) if err != nil { - h.mapErr(w, err) + h.mapErr(c, err) return } if job == nil { - Fail(w, http.StatusNotFound, 404, "job not found") + Fail(c, http.StatusNotFound, 404, "job not found") return } - OK(w, job) + logging.L.Infow("作业改计划", + logging.KeyReqID, RequestIDFrom(c), + logging.KeyJobID, job.ID, + logging.KeyAccountID, caller.AccountID, + logging.KeyStatus, job.Status, + logging.KeyModel, job.Model, + ) + OK(c, job) } -func (h *Handler) inject(w http.ResponseWriter, r *http.Request) { +func (h *Handler) inject(c *gin.Context) { var body service.InjectJobRequest - _ = json.NewDecoder(r.Body).Decode(&body) - job, err := h.Jobs.Inject(r.PathValue("id"), &body, h.caller(r)) + if err := bindBody(c, &body); err != nil { + Fail(c, http.StatusBadRequest, 400, "请求体不是合法 JSON") + return + } + caller := h.caller(c) + job, err := h.Jobs.Inject(c.Param("id"), &body, caller) if err != nil { - h.mapErr(w, err) + h.mapErr(c, err) return } if job == nil { - Fail(w, http.StatusNotFound, 404, "job not found") + Fail(c, http.StatusNotFound, 404, "job not found") return } - OK(w, job) + // 不打印 inject 全文（脱敏），只留字数，确认请求到了即可。 + logging.L.Infow("作业注入补充", + logging.KeyReqID, RequestIDFrom(c), + logging.KeyJobID, job.ID, + logging.KeyAccountID, caller.AccountID, + "chars", len([]rune(strings.TrimSpace(body.Text))), + ) + OK(c, job) } -func (h *Handler) result(w http.ResponseWriter, r *http.Request) { - job, err := h.Jobs.Get(r.PathValue("id"), h.caller(r)) +func (h *Handler) result(c *gin.Context) { + job, err := h.Jobs.Get(c.Param("id"), h.caller(c)) if err != nil { - h.mapErr(w, err) + h.mapErr(c, err) return } if job == nil { - Fail(w, http.StatusNotFound, 404, "job not found") + Fail(c, http.StatusNotFound, 404, "job not found") return } - OK(w, map[string]interface{}{ + OK(c, gin.H{ "id": job.ID, "status": job.Status, "result": job.Result, @@ -158,32 +255,48 @@ func (h *Handler) result(w http.ResponseWriter, r *http.Request) { }) } -func (h *Handler) mapErr(w http.ResponseWriter, err error) { +// mapErr 把业务错误翻成前端已经认的 HTTP 状态。未知错误才 500，并打全量 err。 +func (h *Handler) mapErr(c *gin.Context, err error) { var be *service.BizError if errors.As(err, &be) { switch { case errors.Is(be.Kind, service.ErrForbidden): - Fail(w, http.StatusForbidden, 403, be.Msg) + logging.L.Warnw("权限拒绝", + logging.KeyReqID, RequestIDFrom(c), + "path", c.Request.URL.Path, + "msg", be.Msg, + ) + Fail(c, http.StatusForbidden, 403, be.Msg) case errors.Is(be.Kind, service.ErrBadState), errors.Is(be.Kind, service.ErrBadArg): - Fail(w, http.StatusBadRequest, 400, be.Msg) + logging.L.Infow("业务拒绝", + logging.KeyReqID, RequestIDFrom(c), + "path", c.Request.URL.Path, + "msg", be.Msg, + ) + Fail(c, http.StatusBadRequest, 400, be.Msg) default: - Fail(w, http.StatusInternalServerError, 500, be.Msg) + Fail(c, http.StatusInternalServerError, 500, be.Msg) } return } - Fail(w, http.StatusInternalServerError, 500, err.Error()) + logging.L.Errorw("接口内部错误", + logging.KeyReqID, RequestIDFrom(c), + "path", c.Request.URL.Path, + "err", err, + ) + Fail(c, http.StatusInternalServerError, 500, err.Error()) } -// WithCORS 本机联调放开跨域（经反代同源时无影响）。 -func WithCORS(next http.Handler) http.Handler { - return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - w.Header().Set("Access-Control-Allow-Origin", "*") - w.Header().Set("Access-Control-Allow-Methods", "GET, POST, PUT, DELETE, OPTIONS") - w.Header().Set("Access-Control-Allow-Headers", "*") - if r.Method == http.MethodOptions { - w.WriteHeader(http.StatusNoContent) - return - } - next.ServeHTTP(w, r) - }) +// bindBody 读取 JSON。没有 body（或只有空白）视为空对象，和以前「解码失败就当空请求」里合法的那一半对齐。 +// 有内容但不是 JSON 则返回错误，由调用方回 400。 +func bindBody(c *gin.Context, dst interface{}) error { + if c.Request.Body == nil { + return nil + } + dec := json.NewDecoder(c.Request.Body) + err := dec.Decode(dst) + if err == nil || errors.Is(err, io.EOF) { + return nil + } + return err } diff --git a/internal/api/handler_test.go b/internal/api/handler_test.go new file mode 100644 index 0000000..c658ef1 --- /dev/null +++ b/internal/api/handler_test.go @@ -0,0 +1,44 @@ +package api + +import ( + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/gin-gonic/gin" +) + +// 只锁路由外壳：探活不碰库，坏 JSON 在进业务前被拒绝。 +func TestHealthAndRejectsBadJSON(t *testing.T) { + gin.SetMode(gin.TestMode) + r := gin.New() + r.Use(RequestID(), CORS(), Recovery()) + h := &Handler{Version: "test", Started: time.Now()} + h.Register(r) + + req := httptest.NewRequest(http.MethodGet, "/api/v1/analysis/health", nil) + rec := httptest.NewRecorder() + r.ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("health %d %s", rec.Code, rec.Body.String()) + } + if rec.Header().Get("X-Request-Id") == "" { + t.Fatal("响应应带回 X-Request-Id") + } + + req = httptest.NewRequest(http.MethodPost, "/api/v1/analysis/jobs", strings.NewReader("{")) + rec = httptest.NewRecorder() + r.ServeHTTP(rec, req) + if rec.Code != http.StatusBadRequest { + t.Fatalf("bad json %d %s", rec.Code, rec.Body.String()) + } + + req = httptest.NewRequest(http.MethodGet, "/api/v1/analysis/jobs", nil) + rec = httptest.NewRecorder() + r.ServeHTTP(rec, req) + if rec.Code != http.StatusBadRequest { + t.Fatalf("missing taskId %d %s", rec.Code, rec.Body.String()) + } +} diff --git a/internal/api/middleware.go b/internal/api/middleware.go new file mode 100644 index 0000000..4fb787c --- /dev/null +++ b/internal/api/middleware.go @@ -0,0 +1,87 @@ +package api + +import ( + "crypto/rand" + "encoding/hex" + "net/http" + "time" + + "github.com/gin-gonic/gin" + + "git.gozeros.com/wsk/eightbf-analysis-go/internal/logging" +) + +const reqIDKey = "reqId" + +// RequestID 为每个请求分配 reqId（可读头 X-Request-Id），并做慢请求 WARN。 +// +// 约定：调用方已带 X-Request-Id 就沿用，方便和 Nginx / 前端对上同一次操作。 +// 边界：轮询 GET 不打 INFO，只有超过 3 秒才 WARN，避免作业查询把故障日志淹没。 +func RequestID() gin.HandlerFunc { + return func(c *gin.Context) { + rid := c.GetHeader("X-Request-Id") + if rid == "" { + rid = newReqID() + } + c.Header("X-Request-Id", rid) + c.Set(reqIDKey, rid) + start := time.Now() + c.Next() + elapsed := time.Since(start) + if elapsed >= 3*time.Second { + logging.L.Warnw("慢请求", + logging.KeyReqID, rid, + "method", c.Request.Method, + "path", c.Request.URL.Path, + "ms", elapsed.Milliseconds(), + ) + } + } +} + +// RequestIDFrom 取出中间件写入的请求号。 +func RequestIDFrom(c *gin.Context) string { + if c == nil { + return "" + } + v, ok := c.Get(reqIDKey) + if !ok { + return "" + } + s, _ := v.(string) + return s +} + +func newReqID() string { + b := make([]byte, 8) + _, _ = rand.Read(b) + return "req_" + hex.EncodeToString(b) +} + +// CORS 本机联调放开跨域。经 Nginx 同源反代时浏览器不会走这段，多写无副作用。 +func CORS() gin.HandlerFunc { + return func(c *gin.Context) { + h := c.Writer.Header() + h.Set("Access-Control-Allow-Origin", "*") + h.Set("Access-Control-Allow-Methods", "GET, POST, PUT, DELETE, OPTIONS") + h.Set("Access-Control-Allow-Headers", "*") + h.Set("Access-Control-Expose-Headers", "X-Request-Id") + if c.Request.Method == http.MethodOptions { + c.AbortWithStatus(http.StatusNoContent) + return + } + c.Next() + } +} + +// Recovery 接住 handler panic，记一条错误日志并回 500，避免 Gin 默认文本把响应外壳弄乱。 +func Recovery() gin.HandlerFunc { + return gin.CustomRecovery(func(c *gin.Context, rec interface{}) { + logging.L.Errorw("接口 panic", + logging.KeyReqID, RequestIDFrom(c), + "path", c.Request.URL.Path, + "err", rec, + ) + Fail(c, http.StatusInternalServerError, 500, "internal error") + }) +} diff --git a/internal/api/response.go b/internal/api/response.go index ae81152..4162225 100644 --- a/internal/api/response.go +++ b/internal/api/response.go @@ -1,31 +1,28 @@ package api import ( - "encoding/json" "net/http" + + "github.com/gin-gonic/gin" ) // ApiResponse 与 Java 一致：{ code, message, data }；成功 code=0。 +// +// 探活 /health、/ready 不走这个外壳，编排系统要直接读 ok 字段。 type ApiResponse struct { Code int `json:"code"` Message string `json:"message"` Data interface{} `json:"data"` } -func writeJSON(w http.ResponseWriter, status int, body interface{}) { - w.Header().Set("Content-Type", "application/json; charset=utf-8") - w.WriteHeader(status) - _ = json.NewEncoder(w).Encode(body) +func OK(c *gin.Context, data interface{}) { + c.JSON(http.StatusOK, ApiResponse{Code: 0, Message: "ok", Data: data}) } -func OK(w http.ResponseWriter, data interface{}) { - writeJSON(w, http.StatusOK, ApiResponse{Code: 0, Message: "ok", Data: data}) +func Created(c *gin.Context, data interface{}) { + c.JSON(http.StatusCreated, ApiResponse{Code: 0, Message: "ok", Data: data}) } -func Created(w http.ResponseWriter, data interface{}) { - writeJSON(w, http.StatusCreated, ApiResponse{Code: 0, Message: "ok", Data: data}) -} - -func Fail(w http.ResponseWriter, httpStatus, code int, msg string) { - writeJSON(w, httpStatus, ApiResponse{Code: code, Message: msg, Data: nil}) +func Fail(c *gin.Context, httpStatus, code int, msg string) { + c.JSON(httpStatus, ApiResponse{Code: code, Message: msg, Data: nil}) } diff --git a/internal/config/config.go b/internal/config/config.go index ccbbd4a..eb63395 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -1,87 +1,168 @@ +// Package config 从环境变量装载运行配置。 +// +// 为什么用 Viper：Go 服务里读配置的常见入口，以后若要加配置文件不用再改调用方。 +// 约定：当前只认环境变量，键名与 Java / 旧版 Go 完全一致（EBA_*、ANALYSIS_*、LOG_*、SERVER_PORT）。 +// 边界：空字符串当成没配，回落到联调缺省；非法数字同样回落，不在这里把进程打崩（致命项交给 Validate）。 package config import ( - "os" - "strconv" + "fmt" "strings" + + "github.com/spf13/viper" ) // Config 运行配置（环境变量优先，与 Java EBA_* / analysis.* 对齐）。 type Config struct { Port int - // MySQL：优先 EBA_DB_DSN；否则用 HOST/PORT/NAME/USER/PASSWORD 拼，或解析 JDBC URL。 + // DBDSN MySQL 连接串。优先 EBA_DB_DSN；否则用 HOST/PORT/NAME/USER/PASSWORD 拼，或解析 JDBC URL。 DBDSN string - AnalysisMode string // live | demo - AuthEnabled bool - DemoEnabled bool - LivePollMS int - DemoTickMS int + AnalysisMode string // live | demo。demo 时强制走假进度，不再打 DeerFlow。 + AuthEnabled bool // false 时写操作不强制账号头，仅联调。 + DemoEnabled bool // 真模型与假进度二选一；live 下只有显式打开才为 true。 + LivePollMS int + DemoTickMS int + LiveMaxInflight int // 真模型并发上限。小于 1 时 Load 会抬回 2。 DeerFlowBaseURL string DeerFlowUser string DeerFlowPassword string DeerFlowCoordinator string - DeerFlowStreamIdleTimeout int // 毫秒 + DeerFlowStreamIdleTimeout int // 毫秒。流式读空闲超过这个值视为卡住。 + + LogLevel string + LogFile string + LogMaxMB int + LogMaxAgeDays int + LogMaxBackups int + + // SchemaPath 建表脚本路径；空则尝试 ./schema.sql。 + SchemaPath string + + // StrictConfig true 时：live 使用联调默认/空 DeerFlow 口令直接拒绝启动。 + StrictConfig bool + + // Version 构建信息（也可用环境变量 EBA_VERSION；ldflags 可再覆盖二进制）。 + Version string } // Load 从环境变量加载；缺省贴近本机联调。 +// +// ANALYSIS_MODE=demo 时无视 worker 开关，一律假进度。 +// ANALYSIS_MODE=live 且没有显式设置 EBA_WORKER_DEMO_ENABLED 时，强制关掉 demo， +// 避免壳环境里残留的默认值把真模型服务悄悄变成演示。 func Load() Config { + v := newEnv() c := Config{ - Port: envInt("SERVER_PORT", 18097), - AnalysisMode: env("ANALYSIS_MODE", "live"), - AuthEnabled: envBool("ANALYSIS_AUTH_ENABLED", false), - DemoEnabled: envBool("EBA_WORKER_DEMO_ENABLED", false), - LivePollMS: envInt("EBA_WORKER_LIVE_POLL_MS", 1000), - DemoTickMS: envInt("EBA_WORKER_TICK_MS", 400), - DeerFlowBaseURL: strings.TrimRight(env("EBA_DF_BASE_URL", "http://127.0.0.1:8001"), "/"), - DeerFlowUser: env("EBA_DF_USER", "admin"), - DeerFlowPassword: env("EBA_DF_PASSWORD", "123ewq"), - DeerFlowCoordinator: env("EBA_DF_COORDINATOR", "roundtable-coordinator"), - DeerFlowStreamIdleTimeout: envInt("EBA_DF_STREAM_IDLE_TIMEOUT_MS", 180000), + Port: vInt(v, "SERVER_PORT", 18097), + AnalysisMode: strings.ToLower(vString(v, "ANALYSIS_MODE", "live")), + AuthEnabled: vBool(v, "ANALYSIS_AUTH_ENABLED", false), + DemoEnabled: vBool(v, "EBA_WORKER_DEMO_ENABLED", false), + LivePollMS: vInt(v, "EBA_WORKER_LIVE_POLL_MS", 1000), + DemoTickMS: vInt(v, "EBA_WORKER_TICK_MS", 400), + LiveMaxInflight: vInt(v, "EBA_WORKER_LIVE_MAX", 2), + DeerFlowBaseURL: strings.TrimRight(vString(v, "EBA_DF_BASE_URL", "http://127.0.0.1:8001"), "/"), + DeerFlowUser: vString(v, "EBA_DF_USER", "admin"), + DeerFlowPassword: vString(v, "EBA_DF_PASSWORD", "123ewq"), + DeerFlowCoordinator: vString(v, "EBA_DF_COORDINATOR", "roundtable-coordinator"), + DeerFlowStreamIdleTimeout: vInt(v, "EBA_DF_STREAM_IDLE_TIMEOUT_MS", 180000), + LogLevel: vString(v, "LOG_LEVEL", "info"), + LogFile: vString(v, "LOG_FILE", ""), + LogMaxMB: vInt(v, "LOG_MAX_MB", 100), + LogMaxAgeDays: vInt(v, "LOG_MAX_AGE_DAYS", 14), + LogMaxBackups: vInt(v, "LOG_MAX_BACKUPS", 14), + SchemaPath: vString(v, "EBA_SCHEMA_PATH", "schema.sql"), + StrictConfig: vBool(v, "EBA_STRICT_CONFIG", false), + Version: vString(v, "EBA_VERSION", "dev"), } - c.DBDSN = resolveDSN() - // live 默认关掉 demo；显式 ANALYSIS_MODE=demo 时打开假进度 + c.DBDSN = resolveDSN(v) if c.AnalysisMode == "demo" { c.DemoEnabled = true } - if c.AnalysisMode == "live" && !envExplicit("EBA_WORKER_DEMO_ENABLED") { + if c.AnalysisMode == "live" && !v.IsSet("EBA_WORKER_DEMO_ENABLED") { c.DemoEnabled = false } + if c.LiveMaxInflight < 1 { + c.LiveMaxInflight = 2 + } return c } -func resolveDSN() string { - if d := strings.TrimSpace(os.Getenv("EBA_DB_DSN")); d != "" { +// Validate 启动前校验；返回警告列表。strict 时致命错误用 error 返回，调用方应退出进程。 +func (c Config) Validate() (warnings []string, err error) { + switch c.AnalysisMode { + case "live", "demo": + default: + return nil, fmt.Errorf("ANALYSIS_MODE 只能是 live 或 demo，当前=%q", c.AnalysisMode) + } + if c.Port < 1 || c.Port > 65535 { + return nil, fmt.Errorf("SERVER_PORT 非法: %d", c.Port) + } + if c.DBDSN == "" { + return nil, fmt.Errorf("数据库 DSN 为空，请配置 EBA_DB_DSN 或 EBA_DB_*") + } + + weakDF := c.DeerFlowPassword == "123ewq" || c.DeerFlowPassword == "" + weakHint := "正式环境请改 EBA_DF_PASSWORD，勿用联调默认口令" + + // live 声明下就要管 DF 口令（即使误开了 demo 开关也不放过 strict）。 + if c.AnalysisMode == "live" { + if !c.DemoEnabled && c.DeerFlowBaseURL == "" { + return nil, fmt.Errorf("live 模式必须配置 EBA_DF_BASE_URL") + } + if weakDF { + msg := "DeerFlow 仍使用联调默认/空密码：" + weakHint + if c.StrictConfig { + return nil, fmt.Errorf("%s（EBA_STRICT_CONFIG=true）", msg) + } + warnings = append(warnings, msg) + } + } + if !c.AuthEnabled { + warnings = append(warnings, "ANALYSIS_AUTH_ENABLED=false：写操作不强制账号头（仅联调）") + } + return warnings, nil +} + +// newEnv 只绑定进程环境变量，不读配置文件，避免工作目录里偶然的 yaml 盖掉部署注入的 EBA_*。 +func newEnv() *viper.Viper { + v := viper.New() + v.AutomaticEnv() + return v +} + +func resolveDSN(v *viper.Viper) string { + if d := strings.TrimSpace(v.GetString("EBA_DB_DSN")); v.IsSet("EBA_DB_DSN") && d != "" { return ensureMultiStatements(d) } - user := env("EBA_DB_USER", "eba") - pass := env("EBA_DB_PASSWORD", "eba_local_dev") - jdbc := strings.TrimSpace(os.Getenv("EBA_DB_URL")) + user := vString(v, "EBA_DB_USER", "eba") + pass := vString(v, "EBA_DB_PASSWORD", "eba_local_dev") + jdbc := "" + if v.IsSet("EBA_DB_URL") { + jdbc = strings.TrimSpace(v.GetString("EBA_DB_URL")) + } if jdbc != "" { if dsn, ok := jdbcToDSN(jdbc, user, pass); ok { return dsn } } - host := env("EBA_DB_HOST", "127.0.0.1") - port := env("EBA_DB_PORT", "3306") - name := env("EBA_DB_NAME", "eightbf_analysis") + host := vString(v, "EBA_DB_HOST", "127.0.0.1") + port := vString(v, "EBA_DB_PORT", "3306") + name := vString(v, "EBA_DB_NAME", "eightbf_analysis") return user + ":" + pass + "@tcp(" + host + ":" + port + ")/" + name + "?parseTime=true&loc=Asia%2FShanghai&charset=utf8mb4&multiStatements=true" } -// jdbcToDSN 把 Java 的 jdbc:mysql://host:port/db?... 转成 Go driver DSN。 func jdbcToDSN(jdbc, user, pass string) (string, bool) { s := strings.TrimPrefix(jdbc, "jdbc:mysql://") if s == jdbc { return "", false } hostPath := s - query := "" if i := strings.Index(s, "?"); i >= 0 { hostPath = s[:i] - query = s[i+1:] } host := hostPath db := "eightbf_analysis" @@ -90,7 +171,6 @@ func jdbcToDSN(jdbc, user, pass string) (string, bool) { db = hostPath[i+1:] } params := "parseTime=true&loc=Asia%2FShanghai&charset=utf8mb4&multiStatements=true" - _ = query // Java JDBC 专用参数不照搬 return user + ":" + pass + "@tcp(" + host + ")/" + db + "?" + params, true } @@ -104,24 +184,23 @@ func ensureMultiStatements(dsn string) string { return dsn + "?multiStatements=true" } -func env(k, def string) string { - if v := strings.TrimSpace(os.Getenv(k)); v != "" { - return v - } - return def -} - -func envExplicit(k string) bool { - _, ok := os.LookupEnv(k) - return ok -} - -func envBool(k string, def bool) bool { - v := strings.TrimSpace(os.Getenv(k)) - if v == "" { +// vString 未设置或空白都用缺省。Viper 对空环境变量会返回空串，这里与旧版 os.Getenv 行为对齐。 +func vString(v *viper.Viper, key, def string) string { + if !v.IsSet(key) { return def } - switch strings.ToLower(v) { + s := strings.TrimSpace(v.GetString(key)) + if s == "" { + return def + } + return s +} + +func vBool(v *viper.Viper, key string, def bool) bool { + if !v.IsSet(key) { + return def + } + switch strings.ToLower(strings.TrimSpace(v.GetString(key))) { case "1", "true", "yes", "on": return true case "0", "false", "no", "off": @@ -131,13 +210,16 @@ func envBool(k string, def bool) bool { } } -func envInt(k string, def int) int { - v := strings.TrimSpace(os.Getenv(k)) - if v == "" { +func vInt(v *viper.Viper, key string, def int) int { + if !v.IsSet(key) { return def } - n, err := strconv.Atoi(v) - if err != nil { + s := strings.TrimSpace(v.GetString(key)) + if s == "" { + return def + } + var n int + if _, err := fmt.Sscan(s, &n); err != nil { return def } return n diff --git a/internal/config/config_test.go b/internal/config/config_test.go new file mode 100644 index 0000000..6baac50 --- /dev/null +++ b/internal/config/config_test.go @@ -0,0 +1,56 @@ +package config + +import "testing" + +// 这些用例锁住「环境变量怎么解释」，避免换 Viper 后联调缺省或 demo 开关被悄悄改掉。 + +func TestLoadDefaults(t *testing.T) { + t.Setenv("ANALYSIS_MODE", "") + t.Setenv("EBA_WORKER_DEMO_ENABLED", "") + t.Setenv("SERVER_PORT", "") + // 空字符串按「没配」处理，应回到 live 且不跑 demo。 + c := Load() + if c.AnalysisMode != "live" { + t.Fatalf("mode=%s", c.AnalysisMode) + } + if c.DemoEnabled { + t.Fatal("live 且未显式打开 demo 时不应跑假进度") + } + if c.Port != 18097 { + t.Fatalf("port=%d", c.Port) + } +} + +func TestDemoModeForcesWorker(t *testing.T) { + t.Setenv("ANALYSIS_MODE", "demo") + t.Setenv("EBA_WORKER_DEMO_ENABLED", "false") + c := Load() + if !c.DemoEnabled { + t.Fatal("ANALYSIS_MODE=demo 必须开假进度，忽略 worker 开关") + } +} + +func TestLiveExplicitDemo(t *testing.T) { + t.Setenv("ANALYSIS_MODE", "live") + t.Setenv("EBA_WORKER_DEMO_ENABLED", "yes") + c := Load() + if !c.DemoEnabled { + t.Fatal("live 下显式 EBA_WORKER_DEMO_ENABLED=yes 应保留假进度") + } +} + +func TestStrictRejectsWeakPassword(t *testing.T) { + c := Config{ + AnalysisMode: "live", + Port: 18097, + DBDSN: "x", + DeerFlowPassword: "123ewq", + DeerFlowBaseURL: "http://127.0.0.1:8001", + StrictConfig: true, + DemoEnabled: true, + } + _, err := c.Validate() + if err == nil { + t.Fatal("live + strict 即使同时开了 demo，也不能放过默认口令") + } +} diff --git a/internal/deerflow/auth.go b/internal/deerflow/auth.go index efafef0..97055ab 100644 --- a/internal/deerflow/auth.go +++ b/internal/deerflow/auth.go @@ -10,6 +10,7 @@ import ( "time" "git.gozeros.com/wsk/eightbf-analysis-go/internal/config" + "git.gozeros.com/wsk/eightbf-analysis-go/internal/logging" ) // Auth 向 DeerFlow 换 Bearer，过期前复用。 @@ -68,5 +69,7 @@ func (a *Auth) BearerToken() (string, error) { } a.token = parsed.AccessToken a.expiresAt = now.Add(time.Duration(parsed.ExpiresIn) * time.Second) + // 只打过期秒数，不打 token + logging.L.Infow("DeerFlow 登录成功", "expiresInSec", parsed.ExpiresIn) return a.token, nil } diff --git a/internal/deerflow/client.go b/internal/deerflow/client.go index c0e4331..680a827 100644 --- a/internal/deerflow/client.go +++ b/internal/deerflow/client.go @@ -6,12 +6,12 @@ import ( "encoding/json" "fmt" "io" - "log" "net/http" "strings" "time" "git.gozeros.com/wsk/eightbf-analysis-go/internal/config" + "git.gozeros.com/wsk/eightbf-analysis-go/internal/logging" ) // Client 调 DeerFlow：模型校验 / init / run/stream。 @@ -127,7 +127,7 @@ func (c *Client) InitMultiAgent(agentNames []string, mainAgent, model string) (m if len(threadIDs) == 0 { return nil, fmt.Errorf("多智能体初始化未返回 thread_ids") } - log.Printf("DeerFlow init 完成 seats=%d", len(threadIDs)) + logging.L.Infow("DeerFlow init 完成", "seats", len(threadIDs)) return threadIDs, nil } diff --git a/internal/domain/job.go b/internal/domain/job.go index ff89f5e..a6f2930 100644 --- a/internal/domain/job.go +++ b/internal/domain/job.go @@ -4,31 +4,35 @@ import "time" // AnalysisJob 一场会商作业（服务端「研讨录像带」）。 // JSON 字段名与前端 mapJob / Java 实体对齐，少做二次映射。 +// +// plan 里约定的键：seatIndex（下一席）、manualIntervention、pendingInject（用户补充）、awaitReason。 +// access 里约定的键：taskOwnerAccountIds、adminAccountIds、seatAccounts（席位 id → 账号列表）。 +// 对话正文落在表的 chain_json，不是单独的 dialogues 列。 type AnalysisJob struct { - ID string `json:"id"` - RequestID string `json:"requestId,omitempty"` - TaskID string `json:"taskId,omitempty"` - DraftID string `json:"draftId,omitempty"` - DraftKey string `json:"draftKey,omitempty"` - Status string `json:"status"` - Phase string `json:"phase,omitempty"` - Cycle int `json:"cycle"` - ActiveAgentID string `json:"activeAgentId,omitempty"` - ConsensusPercentage int `json:"consensusPercentage"` - PendingClarification string `json:"pendingClarification,omitempty"` - OrchestrationMode string `json:"orchestrationMode,omitempty"` - Model string `json:"model,omitempty"` - Error string `json:"error,omitempty"` - OwnerAccountID string `json:"ownerAccountId,omitempty"` - Agents []map[string]interface{} `json:"agents"` - Dialogues []map[string]interface{} `json:"dialogues"` - ThreadIDs map[string]interface{} `json:"threadIds,omitempty"` - Plan map[string]interface{} `json:"plan"` - Access map[string]interface{} `json:"access"` - Progress map[string]interface{} `json:"progress"` - Result map[string]interface{} `json:"result,omitempty"` - CreatedAt *time.Time `json:"createdAt,omitempty"` - UpdatedAt *time.Time `json:"updatedAt,omitempty"` + ID string `json:"id"` + RequestID string `json:"requestId,omitempty"` + TaskID string `json:"taskId,omitempty"` + DraftID string `json:"draftId,omitempty"` + DraftKey string `json:"draftKey,omitempty"` + Status string `json:"status"` + Phase string `json:"phase,omitempty"` + Cycle int `json:"cycle"` + ActiveAgentID string `json:"activeAgentId,omitempty"` + ConsensusPercentage int `json:"consensusPercentage"` + PendingClarification string `json:"pendingClarification,omitempty"` + OrchestrationMode string `json:"orchestrationMode,omitempty"` + Model string `json:"model,omitempty"` + Error string `json:"error,omitempty"` + OwnerAccountID string `json:"ownerAccountId,omitempty"` + Agents []map[string]interface{} `json:"agents"` + Dialogues []map[string]interface{} `json:"dialogues"` + ThreadIDs map[string]interface{} `json:"threadIds,omitempty"` + Plan map[string]interface{} `json:"plan"` + Access map[string]interface{} `json:"access"` + Progress map[string]interface{} `json:"progress"` + Result map[string]interface{} `json:"result,omitempty"` + CreatedAt *time.Time `json:"createdAt,omitempty"` + UpdatedAt *time.Time `json:"updatedAt,omitempty"` } // EnsurePlan plan 为空时补可写 Map。 diff --git a/internal/logging/logging.go b/internal/logging/logging.go new file mode 100644 index 0000000..f2b8984 --- /dev/null +++ b/internal/logging/logging.go @@ -0,0 +1,99 @@ +// Package logging 提供进程级日志。 +// +// 为什么用 zap：Gin 服务里最常见的结构化日志，字段稳定，方便按 jobId / reqId 检索。 +// 约定：业务代码走 SugaredLogger 的 *w 方法（Infow/Warnw/Errorw），键名用本文件常量，不要自造别名。 +// 边界：不写 token、密码、完整对话。失败详情在调用方截断后再传入。 +package logging + +import ( + "os" + "strings" + + "go.uber.org/zap" + "go.uber.org/zap/zapcore" + "gopkg.in/natefinch/lumberjack.v2" +) + +// Options 日志初始化选项（对齐设计要求 §11）。 +type Options struct { + Level string // debug|info|warn|error + File string // 非空则同时写滚动文件 + MaxMB int + MaxAge int // 天 + MaxBak int +} + +// L 全局日志。Setup 之前是空实现，避免测试或提前打日志时空指针。 +var L = zap.NewNop().Sugar() + +// Setup 配置全局 zap：默认 stdout；LOG_FILE 非空时再加滚动文件。返回关闭函数（刷盘）。 +// +// 用控制台编码器而不是 JSON：现有采集按 `jobId=` 这种键值对 grep，换 JSON 会让运维脚本对不上。 +func Setup(opt Options) (closeFn func()) { + level := parseLevel(opt.Level) + encCfg := zap.NewDevelopmentEncoderConfig() + encCfg.EncodeTime = zapcore.ISO8601TimeEncoder + encCfg.EncodeLevel = zapcore.CapitalLevelEncoder + encCfg.EncodeCaller = zapcore.ShortCallerEncoder + encoder := zapcore.NewConsoleEncoder(encCfg) + + writers := []zapcore.WriteSyncer{zapcore.AddSync(os.Stdout)} + closeFn = func() { _ = L.Sync() } + + if path := strings.TrimSpace(opt.File); path != "" { + if opt.MaxMB <= 0 { + opt.MaxMB = 100 + } + if opt.MaxAge <= 0 { + opt.MaxAge = 14 + } + if opt.MaxBak <= 0 { + opt.MaxBak = 14 + } + lj := &lumberjack.Logger{ + Filename: path, + MaxSize: opt.MaxMB, + MaxAge: opt.MaxAge, + MaxBackups: opt.MaxBak, + LocalTime: true, + Compress: false, + } + writers = append(writers, zapcore.AddSync(lj)) + prev := closeFn + closeFn = func() { + prev() + _ = lj.Close() + } + } + + core := zapcore.NewCore(encoder, zapcore.NewMultiWriteSyncer(writers...), level) + logger := zap.New(core, zap.AddCaller()) + zap.ReplaceGlobals(logger) + L = logger.Sugar() + return closeFn +} + +func parseLevel(s string) zapcore.Level { + switch strings.ToLower(strings.TrimSpace(s)) { + case "debug": + return zapcore.DebugLevel + case "warn", "warning": + return zapcore.WarnLevel + case "error": + return zapcore.ErrorLevel + default: + return zapcore.InfoLevel + } +} + +// 常用键名，方便 grep / 采集。调用方不要改成 job_id 这类别名。 +const ( + KeyJobID = "jobId" + KeyRequestID = "requestId" + KeyTaskID = "taskId" + KeyAccountID = "accountId" + KeyStatus = "status" + KeyPhase = "phase" + KeyModel = "model" + KeyReqID = "reqId" +) diff --git a/internal/repo/job_repo.go b/internal/repo/job_repo.go index d0d4d8d..74a1d93 100644 --- a/internal/repo/job_repo.go +++ b/internal/repo/job_repo.go @@ -5,15 +5,21 @@ import ( "encoding/json" "time" + "github.com/jmoiron/sqlx" + "git.gozeros.com/wsk/eightbf-analysis-go/internal/domain" ) -// JobRepo 会商作业表读写（eba_analysis_job）。大字段整段 JSON 文本存。 +// JobRepo 会商作业表读写（eba_analysis_job）。大字段整段 JSON 文本存，不拆列。 +// +// 为什么用 sqlx：在标准库连接上多了按列名扫描，SELECT * 换列顺序也不会对错字段。 +// 约定：SQL 与 Java 表结构对齐；空字符串写入 NULL，读回来再变成空串。 +// 边界：本仓库没有领取锁。两个进程同时扫 queued 会抢同一行。 type JobRepo struct { - DB *sql.DB + DB *sqlx.DB } -func NewJobRepo(db *sql.DB) *JobRepo { +func NewJobRepo(db *sqlx.DB) *JobRepo { return &JobRepo{DB: db} } @@ -67,51 +73,54 @@ func (r *JobRepo) UpdateProgress(job *domain.AnalysisJob) error { } func (r *JobRepo) FindByID(id string) (*domain.AnalysisJob, error) { - rows, err := r.DB.Query(`SELECT * FROM eba_analysis_job WHERE id=?`, id) + rows, err := r.DB.Queryx(`SELECT * FROM eba_analysis_job WHERE id=?`, id) if err != nil { return nil, err } defer rows.Close() - if !rows.Next() { - return nil, nil + list, err := scanJobs(rows) + if err != nil || len(list) == 0 { + return nil, err } - return scanJob(rows) + return list[0], nil } func (r *JobRepo) FindActiveByTaskID(taskID string) (*domain.AnalysisJob, error) { if taskID == "" { return nil, nil } - rows, err := r.DB.Query(`SELECT * FROM eba_analysis_job WHERE task_id=? AND status IN (?,?,?) + rows, err := r.DB.Queryx(`SELECT * FROM eba_analysis_job WHERE task_id=? AND status IN (?,?,?) ORDER BY created_at DESC LIMIT 1`, taskID, domain.StatusQueued, domain.StatusRunning, domain.StatusAwaitingInput) if err != nil { return nil, err } defer rows.Close() - if !rows.Next() { - return nil, nil + list, err := scanJobs(rows) + if err != nil || len(list) == 0 { + return nil, err } - return scanJob(rows) + return list[0], nil } func (r *JobRepo) FindLatestByTaskID(taskID string) (*domain.AnalysisJob, error) { if taskID == "" { return nil, nil } - rows, err := r.DB.Query(`SELECT * FROM eba_analysis_job WHERE task_id=? ORDER BY created_at DESC LIMIT 1`, taskID) + rows, err := r.DB.Queryx(`SELECT * FROM eba_analysis_job WHERE task_id=? ORDER BY created_at DESC LIMIT 1`, taskID) if err != nil { return nil, err } defer rows.Close() - if !rows.Next() { - return nil, nil + list, err := scanJobs(rows) + if err != nil || len(list) == 0 { + return nil, err } - return scanJob(rows) + return list[0], nil } func (r *JobRepo) FindActiveJobs() ([]*domain.AnalysisJob, error) { - rows, err := r.DB.Query(`SELECT * FROM eba_analysis_job WHERE status IN (?,?,?) ORDER BY created_at ASC`, + rows, err := r.DB.Queryx(`SELECT * FROM eba_analysis_job WHERE status IN (?,?,?) ORDER BY created_at ASC`, domain.StatusQueued, domain.StatusRunning, domain.StatusAwaitingInput) if err != nil { return nil, err @@ -121,7 +130,7 @@ func (r *JobRepo) FindActiveJobs() ([]*domain.AnalysisJob, error) { } func (r *JobRepo) FindQueuedJobs() ([]*domain.AnalysisJob, error) { - rows, err := r.DB.Query(`SELECT * FROM eba_analysis_job WHERE status=? ORDER BY created_at ASC`, domain.StatusQueued) + rows, err := r.DB.Queryx(`SELECT * FROM eba_analysis_job WHERE status=? ORDER BY created_at ASC`, domain.StatusQueued) if err != nil { return nil, err } @@ -151,35 +160,20 @@ func (r *JobRepo) IsCancelFlag(id string) (bool, error) { return v.Valid && v.Int64 == 1, nil } -func scanJobs(rows *sql.Rows) ([]*domain.AnalysisJob, error) { +func scanJobs(rows *sqlx.Rows) ([]*domain.AnalysisJob, error) { var out []*domain.AnalysisJob for rows.Next() { - j, err := scanJob(rows) - if err != nil { + m := map[string]interface{}{} + if err := rows.MapScan(m); err != nil { return nil, err } - out = append(out, j) + out = append(out, jobFromMap(m)) } return out, rows.Err() } -func scanJob(rows *sql.Rows) (*domain.AnalysisJob, error) { - cols, err := rows.Columns() - if err != nil { - return nil, err - } - vals := make([]interface{}, len(cols)) - ptrs := make([]interface{}, len(cols)) - for i := range vals { - ptrs[i] = &vals[i] - } - if err := rows.Scan(ptrs...); err != nil { - return nil, err - } - m := map[string]interface{}{} - for i, c := range cols { - m[c] = vals[i] - } +// jobFromMap 按列名组装作业。对话正文存在 chain_json，这是和 Java 表对齐的历史列名，不是链条配置。 +func jobFromMap(m map[string]interface{}) *domain.AnalysisJob { j := &domain.AnalysisJob{} j.ID = asString(m["id"]) j.RequestID = asString(m["request_id"]) @@ -209,7 +203,7 @@ func scanJob(rows *sql.Rows) (*domain.AnalysisJob, error) { if t := asTime(m["updated_at"]); t != nil { j.UpdatedAt = t } - return j, nil + return j } func mustJSON(v interface{}) interface{} { diff --git a/internal/repo/schema.go b/internal/repo/schema.go index 03a714c..37552ff 100644 --- a/internal/repo/schema.go +++ b/internal/repo/schema.go @@ -1,8 +1,9 @@ package repo import ( - "log" "strings" + + "git.gozeros.com/wsk/eightbf-analysis-go/internal/logging" ) // ApplySchema 按语句拆开执行；重复索引等可忽略错误不阻断启动。 @@ -15,11 +16,18 @@ func (r *JobRepo) ApplySchema(schemaSQL string) { strings.Contains(msg, "already exists") { continue } - log.Printf("schema 语句失败（已跳过）: %v | sql=%.80s…", err, stmt) + logging.L.Warnw("schema 语句失败（已跳过）", "err", err, "sql", trunc(stmt, 80)) } } } +func trunc(s string, n int) string { + if len(s) <= n { + return s + } + return s[:n] + "…" +} + func splitSQL(s string) []string { var out []string var b strings.Builder diff --git a/internal/service/access.go b/internal/service/access.go index b632556..b543cf8 100644 --- a/internal/service/access.go +++ b/internal/service/access.go @@ -33,6 +33,11 @@ type AccessService struct { AuthEnabled bool } +// CanMutate 谁可以续跑、改计划、注入。 +// +// 鉴权关闭时一律放行（联调）。打开后：没账号拒绝；管理员、任务责任人角色、作业拥有者、 +// access.taskOwnerAccountIds / adminAccountIds 里的账号可以改。 +// 岗位责任人只有绑了席位才能改本场，取消权另见 CanCancel。 func (a *AccessService) CanMutate(job *domain.AnalysisJob, c Caller) bool { if !a.AuthEnabled { return true @@ -52,6 +57,8 @@ func (a *AccessService) CanMutate(job *domain.AnalysisJob, c Caller) bool { return c.Role == RolePositionOwner && len(a.boundSeats(job, c.AccountID)) > 0 } +// CanCancel 谁可以取消。岗位责任人默认不能取消，除非他同时是作业拥有者或名单里的任务责任人。 +// 这条比 CanMutate 更严：席位上的人可以插话，但不能把整场停掉。 func (a *AccessService) CanCancel(job *domain.AnalysisJob, c Caller) bool { if c.Role == RolePositionOwner { if c.AccountID == strings.TrimSpace(job.OwnerAccountID) || a.isListedTaskOwner(job, c.AccountID) { @@ -74,6 +81,7 @@ func (a *AccessService) CanCancel(job *domain.AnalysisJob, c Caller) bool { return a.isListedTaskOwner(job, c.AccountID) } +// CanRead 谁可以看。能改或能取消的人可以看；只绑了席位、没有写权限的账号也可以看（随后会被裁剪）。 func (a *AccessService) CanRead(job *domain.AnalysisJob, c Caller) bool { if !a.AuthEnabled { return true diff --git a/internal/service/demo_worker.go b/internal/service/demo_worker.go index ac0f2a3..40817b9 100644 --- a/internal/service/demo_worker.go +++ b/internal/service/demo_worker.go @@ -1,17 +1,17 @@ package service import ( - "log" "strconv" "time" "git.gozeros.com/wsk/eightbf-analysis-go/internal/domain" + "git.gozeros.com/wsk/eightbf-analysis-go/internal/logging" "git.gozeros.com/wsk/eightbf-analysis-go/internal/repo" ) // DemoWorker 假进度推进（仅 ANALYSIS_MODE=demo）。 type DemoWorker struct { - Repo *repo.JobRepo + Repo *repo.JobRepo Every time.Duration } @@ -40,7 +40,7 @@ func (w *DemoWorker) Start() { } for _, job := range list { if err := w.advance(job); err != nil { - log.Printf("demo worker advance failed jobId=%s: %v", job.ID, err) + logging.L.Warnw("演示推进失败", logging.KeyJobID, job.ID, "err", err) job.Status = domain.StatusError job.Phase = "error" job.Error = err.Error() diff --git a/internal/service/job_service.go b/internal/service/job_service.go index bec430a..0b6bfb2 100644 --- a/internal/service/job_service.go +++ b/internal/service/job_service.go @@ -33,7 +33,7 @@ type ResumeJobRequest struct { } type PatchJobPlanRequest struct { - ManualIntervention *bool `json:"manualIntervention"` + ManualIntervention *bool `json:"manualIntervention"` Model *string `json:"model"` } @@ -47,6 +47,11 @@ type JobService struct { Access *AccessService } +// Create 落一场排队作业，不在这里调模型。Worker 下一拍才会领走。 +// +// 约定：没传 manualIntervention 时默认开人工卡点；编排模式缺省 roundtable。 +// 席位为空时塞两个联调桩，避免前端漏传 agents 时直接起不来。 +// 拥有者优先用 body，没有再用请求头里的账号。 func (s *JobService) Create(req *CreateJobRequest, caller Caller) (*domain.AnalysisJob, error) { if req == nil { req = &CreateJobRequest{} @@ -64,19 +69,19 @@ func (s *JobService) Create(req *CreateJobRequest, caller Caller) (*domain.Analy mode = "roundtable" } job := &domain.AnalysisJob{ - ID: "job_" + randHex(8), - RequestID: firstNonEmpty(blankToNull(req.RequestID), "req_"+randHex(6)), - TaskID: blankToNull(req.TaskID), - DraftID: blankToNull(req.DraftID), - DraftKey: blankToNull(req.DraftKey), - Status: domain.StatusQueued, - Phase: "queued", - OrchestrationMode: mode, - Model: strings.TrimSpace(req.Model), - OwnerAccountID: owner, - Agents: normalizeAgents(req.Agents), - Dialogues: []map[string]interface{}{}, - Access: orEmptyMap(req.Access), + ID: "job_" + randHex(8), + RequestID: firstNonEmpty(blankToNull(req.RequestID), "req_"+randHex(6)), + TaskID: blankToNull(req.TaskID), + DraftID: blankToNull(req.DraftID), + DraftKey: blankToNull(req.DraftKey), + Status: domain.StatusQueued, + Phase: "queued", + OrchestrationMode: mode, + Model: strings.TrimSpace(req.Model), + OwnerAccountID: owner, + Agents: normalizeAgents(req.Agents), + Dialogues: []map[string]interface{}{}, + Access: orEmptyMap(req.Access), Plan: map[string]interface{}{ "seatIndex": 0, "manualIntervention": manual, @@ -131,6 +136,8 @@ func (s *JobService) GetByTaskID(taskID string, latestOnly bool, caller Caller) return s.Access.ProjectForCaller(job, caller), nil } +// Cancel 给还在排队、运行、等人中的作业打 cancel_flag，并把状态写成 cancelled。 +// 已经结束的作业再点取消：不报错，原样返回（前端可以重复点）。 func (s *JobService) Cancel(id string, caller Caller) (*domain.AnalysisJob, error) { job, err := s.Repo.FindByID(id) if err != nil || job == nil { @@ -154,6 +161,8 @@ func (s *JobService) Cancel(id string, caller Caller) (*domain.AnalysisJob, erro return s.Access.ProjectForCaller(job, caller), nil } +// Resume 只接受 awaiting_input。回答会写进用户气泡，并拼进 plan.pendingInject，下一轮派活时带上。 +// 然后状态回到 queued，等 Worker 重新领走。其它状态拒绝，避免把跑到一半的场子重置。 func (s *JobService) Resume(id string, body *ResumeJobRequest, caller Caller) (*domain.AnalysisJob, error) { job, err := s.Repo.FindByID(id) if err != nil || job == nil { @@ -195,6 +204,10 @@ func (s *JobService) Resume(id string, body *ResumeJobRequest, caller Caller) (* return s.Access.ProjectForCaller(fresh, caller), nil } +// PatchPlan 改模型或人工干预开关。作业已结束则拒绝。 +// +// 关掉人工干预、且当前正停在 manual_continue 时，顺手放行：清掉等待原因并重新排队。 +// 这样页面上关开关后不用再点一次「继续」。 func (s *JobService) PatchPlan(id string, body *PatchJobPlanRequest, caller Caller) (*domain.AnalysisJob, error) { job, err := s.Repo.FindByID(id) if err != nil || job == nil { @@ -253,6 +266,8 @@ func (s *JobService) PatchPlan(id string, body *PatchJobPlanRequest, caller Call return s.Access.ProjectForCaller(fresh, caller), nil } +// Inject 在会商进行中插入用户补充。空文本拒绝；已结束拒绝。 +// 正文进对话气泡，同时累加到 plan.pendingInject，下一轮才喂给模型，不打断当前席。 func (s *JobService) Inject(id string, body *InjectJobRequest, caller Caller) (*domain.AnalysisJob, error) { job, err := s.Repo.FindByID(id) if err != nil || job == nil { diff --git a/internal/worker/live_runner.go b/internal/worker/live_runner.go index 43c6ef8..6f56808 100644 --- a/internal/worker/live_runner.go +++ b/internal/worker/live_runner.go @@ -2,7 +2,6 @@ package worker import ( "fmt" - "log" "strings" "sync" "time" @@ -10,24 +9,30 @@ import ( "git.gozeros.com/wsk/eightbf-analysis-go/internal/deerflow" "git.gozeros.com/wsk/eightbf-analysis-go/internal/domain" + "git.gozeros.com/wsk/eightbf-analysis-go/internal/logging" "git.gozeros.com/wsk/eightbf-analysis-go/internal/repo" ) // LiveChainRunner 真模型按链条顺序执行（被 LiveWorker 异步调用）。 type LiveChainRunner struct { - DF *deerflow.Client - Repo *repo.JobRepo + DF *deerflow.Client + Repo *repo.JobRepo Coord string - mu sync.Mutex - lastFlush time.Time + mu sync.Mutex + lastFlush time.Time } +// Run 跑完一场真模型会商。调用方已经把状态改成 running，这里不再抢队列。 +// +// 顺序：席位不够直接失败 → 校验模型（禁止网关静默换模）→ 没有 thread 就 init → +// 总控派活、席位发言，人工干预开着则每席结束后暂停 → 收口写成 done。 +// 边界：任一步 panic 记失败，不当成功；取消看库里的 cancel_flag，不靠内存标记。 func (r *LiveChainRunner) Run(job *domain.AnalysisJob) { jobID := job.ID defer func() { if rec := recover(); rec != nil { - log.Printf("真模型会商 panic jobId=%s: %v", jobID, rec) + logging.L.Errorw("真模型会商 panic", logging.KeyJobID, jobID, "err", rec) r.fail(job, fmt.Sprint(rec)) } }() @@ -285,9 +290,15 @@ func (r *LiveChainRunner) Run(job *domain.AnalysisJob) { plan["awaitReason"] = nil job.Plan = plan r.flush(job) - log.Printf("真模型会商完成 jobId=%s", jobID) + logging.L.Infow("真模型会商完成", + logging.KeyJobID, jobID, + logging.KeyStatus, domain.StatusDone, + logging.KeyModel, job.Model, + ) } +// pauseForInput 把作业停在 awaiting_input。seatIndex 记下一席，等人或等确认后再从这里继续。 +// reason 只有两类约定：澄清（用户要补一句）和 manual_continue（人工确认再往下走）。 func (r *LiveChainRunner) pauseForInput(job *domain.AnalysisJob, plan map[string]interface{}, next int, reason, detail string) { plan["seatIndex"] = next plan["awaitReason"] = reason @@ -301,7 +312,11 @@ func (r *LiveChainRunner) pauseForInput(job *domain.AnalysisJob, plan map[string } putProgress(job, job.ConsensusPercentage, msg) r.flush(job) - log.Printf("作业暂停等待输入 jobId=%s reason=%s", job.ID, reason) + logging.L.Infow("作业暂停等待输入", + logging.KeyJobID, job.ID, + logging.KeyStatus, domain.StatusAwaitingInput, + logging.KeyPhase, reason, + ) } func (r *LiveChainRunner) deltaHandler(job *domain.AnalysisJob, bubbleID string) deerflow.TextDeltaHandler { @@ -326,6 +341,8 @@ func (r *LiveChainRunner) flush(job *domain.AnalysisJob) { r.flushLocked(job) } +// flushLocked 把内存里的作业写回表。写之前从库里拉模型、人工开关和尚未消费的注入， +// 这样页面上改设置不会被本轮旧内存盖掉。调用方必须已持有 r.mu。 func (r *LiveChainRunner) flushLocked(job *domain.AnalysisJob) { plan := ensurePlan(job) r.pullManualFromDB(job, plan) @@ -353,6 +370,7 @@ func (r *LiveChainRunner) markCancelled(job *domain.AnalysisJob) { job.Error = "cancelled" putProgress(job, job.ConsensusPercentage, "cancelled") r.flush(job) + logging.L.Infow("真模型会商已取消", logging.KeyJobID, job.ID, logging.KeyStatus, domain.StatusCancelled) } func (r *LiveChainRunner) fail(job *domain.AnalysisJob, message string) { @@ -367,6 +385,17 @@ func (r *LiveChainRunner) fail(job *domain.AnalysisJob, message string) { } putProgress(job, 0, "error") r.flush(job) + // 详情截断，避免把整段模型正文打进日志 + detail := message + if rs := []rune(detail); len(rs) > 200 { + detail = string(rs[:200]) + "…" + } + logging.L.Warnw("真模型会商失败", + logging.KeyJobID, job.ID, + logging.KeyStatus, domain.StatusError, + logging.KeyModel, job.Model, + "detail", detail, + ) } func (r *LiveChainRunner) pullModelFromDB(job *domain.AnalysisJob) { diff --git a/internal/worker/live_worker.go b/internal/worker/live_worker.go index 53cff27..357065f 100644 --- a/internal/worker/live_worker.go +++ b/internal/worker/live_worker.go @@ -1,20 +1,21 @@ package worker import ( - "log" "sync/atomic" "time" "git.gozeros.com/wsk/eightbf-analysis-go/internal/domain" + "git.gozeros.com/wsk/eightbf-analysis-go/internal/logging" "git.gozeros.com/wsk/eightbf-analysis-go/internal/repo" ) -// LiveWorker 定时扫 queued，最多并发 2 场交给 LiveChainRunner。 +// LiveWorker 定时扫 queued，最多并发 MaxInflight 场交给 LiveChainRunner。 type LiveWorker struct { - Repo *repo.JobRepo - Runner *LiveChainRunner - PollEvery time.Duration - inflight int32 + Repo *repo.JobRepo + Runner *LiveChainRunner + PollEvery time.Duration + MaxInflight int32 + inflight int32 } func (w *LiveWorker) Start() { @@ -31,22 +32,26 @@ func (w *LiveWorker) Start() { } func (w *LiveWorker) poll() { - if atomic.LoadInt32(&w.inflight) >= 2 { + max := w.MaxInflight + if max < 1 { + max = 2 + } + if atomic.LoadInt32(&w.inflight) >= max { return } queued, err := w.Repo.FindQueuedJobs() if err != nil { - log.Printf("LiveWorker 扫队列失败: %v", err) + logging.L.Warnw("LiveWorker 扫队列失败", "err", err) return } for _, job := range queued { - if atomic.LoadInt32(&w.inflight) >= 2 { + if atomic.LoadInt32(&w.inflight) >= max { break } job.Status = domain.StatusRunning job.Phase = "claimed" if err := w.Repo.UpdateProgress(job); err != nil { - log.Printf("LiveWorker 占坑失败 jobId=%s: %v", job.ID, err) + logging.L.Warnw("LiveWorker 占坑失败", logging.KeyJobID, job.ID, "err", err) continue } atomic.AddInt32(&w.inflight, 1) @@ -57,7 +62,13 @@ func (w *LiveWorker) poll() { if err != nil || fresh == nil { return } - log.Printf("开始真模型会商 jobId=%s", jobID) + logging.L.Infow("开始真模型会商", + logging.KeyJobID, jobID, + logging.KeyRequestID, fresh.RequestID, + logging.KeyTaskID, fresh.TaskID, + logging.KeyModel, fresh.Model, + logging.KeyStatus, fresh.Status, + ) w.Runner.Run(fresh) }() }