405 lines
11 KiB
Go
405 lines
11 KiB
Go
package logweb
|
||
|
||
import (
|
||
"encoding/json"
|
||
"fmt"
|
||
"io"
|
||
"io/fs"
|
||
"log"
|
||
"net/http"
|
||
"strconv"
|
||
"strings"
|
||
|
||
"xk-hy-transit-go/internal/applog"
|
||
"xk-hy-transit-go/internal/config"
|
||
"xk-hy-transit-go/internal/db"
|
||
"xk-hy-transit-go/internal/hy"
|
||
syncer "xk-hy-transit-go/internal/sync"
|
||
"xk-hy-transit-go/internal/syncgate"
|
||
)
|
||
|
||
// Runner 同步执行器（与 serve 共用）。
|
||
type Runner interface {
|
||
Run(step, anchorDate string) error
|
||
}
|
||
|
||
func asSyncRunner(r Runner) *syncer.Runner {
|
||
if sr, ok := r.(*syncer.Runner); ok {
|
||
return sr
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// Deps Web 控制台依赖。
|
||
type Deps struct {
|
||
LogDir string
|
||
Store *db.Store
|
||
Runner Runner
|
||
Cfg config.Config
|
||
AllowTest bool
|
||
MySQLOK bool
|
||
MySQLErr string
|
||
}
|
||
|
||
// Server HTTP 控制台。
|
||
type Server struct {
|
||
deps Deps
|
||
mux *http.ServeMux
|
||
}
|
||
|
||
// NewServer 创建服务。
|
||
func NewServer(deps Deps) *Server {
|
||
if deps.LogDir == "" {
|
||
deps.LogDir = applog.Dir()
|
||
}
|
||
s := &Server{deps: deps, mux: http.NewServeMux()}
|
||
s.routes()
|
||
return s
|
||
}
|
||
|
||
func (s *Server) routes() {
|
||
sub, _ := fs.Sub(webFS, "web")
|
||
fileServer := http.FileServer(http.FS(sub))
|
||
|
||
s.mux.HandleFunc("/", s.handleIndex)
|
||
s.mux.HandleFunc("/logs", s.handlePage("logs.html"))
|
||
s.mux.HandleFunc("/logs/view", s.handlePage("logs_view.html"))
|
||
s.mux.HandleFunc("/runs", s.handlePage("runs.html"))
|
||
s.mux.HandleFunc("/runs/record", s.handlePage("runs_record.html"))
|
||
s.mux.HandleFunc("/test", s.handlePage("test.html"))
|
||
s.mux.HandleFunc("/supervise-step", s.handleSuperviseStepPage)
|
||
s.mux.HandleFunc("/config", s.handlePage("config.html"))
|
||
s.mux.HandleFunc("/upload", s.handlePage("upload.html"))
|
||
s.mux.HandleFunc("/fileauth", s.handlePage("fileauth.html"))
|
||
|
||
s.mux.HandleFunc("/api/meta", s.handleAPIMeta)
|
||
s.mux.HandleFunc("/api/fileauth/generate", s.handleAPIFileAuthGenerate)
|
||
s.mux.HandleFunc("/api/config/view", s.handleAPIConfigView)
|
||
s.mux.HandleFunc("/api/config/refresh", s.handleAPIConfigRefresh)
|
||
s.mux.HandleFunc("/api/upload/test", s.handleAPIUploadTest)
|
||
s.mux.HandleFunc("/api/dates", s.handleAPIDates)
|
||
s.mux.HandleFunc("/api/log", s.handleAPILog)
|
||
s.mux.HandleFunc("/api/runs", s.handleAPIRuns)
|
||
s.mux.HandleFunc("/api/runs/record", s.handleAPIRecord)
|
||
s.mux.HandleFunc("/api/runs/truncate", s.handleAPIRunsTruncate)
|
||
s.mux.HandleFunc("/api/runs/retry", s.handleAPIRunsRetry)
|
||
s.mux.HandleFunc("/api/test/status", s.handleAPITestStatus)
|
||
s.mux.HandleFunc("/api/test/sync", s.handleAPITestSync)
|
||
s.mux.HandleFunc("/api/supervise/step-test", s.handleAPISuperviseStepTest)
|
||
|
||
s.mux.Handle("/static/", http.StripPrefix("/static/", fileServer))
|
||
}
|
||
|
||
func (s *Server) handlePage(name string) http.HandlerFunc {
|
||
return func(w http.ResponseWriter, r *http.Request) {
|
||
b, err := webFS.ReadFile("web/" + name)
|
||
if err != nil {
|
||
http.NotFound(w, r)
|
||
return
|
||
}
|
||
w.Header().Set("Content-Type", "text/html; charset=utf-8")
|
||
_, _ = w.Write(b)
|
||
}
|
||
}
|
||
|
||
func (s *Server) handleIndex(w http.ResponseWriter, r *http.Request) {
|
||
if r.URL.Path != "/" {
|
||
http.NotFound(w, r)
|
||
return
|
||
}
|
||
s.handlePage("index.html")(w, r)
|
||
}
|
||
|
||
func (s *Server) handleAPIMeta(w http.ResponseWriter, r *http.Request) {
|
||
writeJSON(w, map[string]any{
|
||
"logDir": s.deps.LogDir,
|
||
"mysqlOk": s.deps.MySQLOK,
|
||
"mysqlError": s.deps.MySQLErr,
|
||
"allowTest": s.deps.AllowTest && s.deps.Runner != nil,
|
||
"allowTruncate": s.deps.Store != nil && s.deps.MySQLOK,
|
||
"hasRunner": s.deps.Runner != nil,
|
||
"stepOptions": hy.StepSelectOptions(),
|
||
"testStepOptions": hy.TestStepSelectOptions(),
|
||
"logKindOptions": []map[string]string{{"value": "app", "label": "应用"}, {"value": "pull", "label": "拉取"}, {"value": "push", "label": "推送"}},
|
||
})
|
||
}
|
||
|
||
func (s *Server) handleAPIDates(w http.ResponseWriter, r *http.Request) {
|
||
kind := r.URL.Query().Get("kind")
|
||
dates, err := ListLogDates(s.deps.LogDir, kind)
|
||
if err != nil {
|
||
writeErr(w, 400, err.Error())
|
||
return
|
||
}
|
||
writeJSON(w, dates)
|
||
}
|
||
|
||
func (s *Server) handleAPILog(w http.ResponseWriter, r *http.Request) {
|
||
kind := r.URL.Query().Get("kind")
|
||
date := r.URL.Query().Get("date")
|
||
meta, lines, err := ReadLogFile(s.deps.LogDir, kind, date)
|
||
if err != nil {
|
||
writeErr(w, 404, err.Error())
|
||
return
|
||
}
|
||
writeJSON(w, map[string]any{
|
||
"meta": meta,
|
||
"lines": ParseLogLines(lines),
|
||
})
|
||
}
|
||
|
||
func (s *Server) handleAPIRuns(w http.ResponseWriter, r *http.Request) {
|
||
if s.deps.Store == nil {
|
||
writeErr(w, 503, "MySQL 未连接")
|
||
return
|
||
}
|
||
limit, _ := strconv.Atoi(r.URL.Query().Get("limit"))
|
||
items, err := ListRuns(s.deps.Store, r.URL.Query().Get("anchor_date"), r.URL.Query().Get("step"), limit)
|
||
if err != nil {
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
jobs, _ := s.deps.Store.ListJobs(30, r.URL.Query().Get("anchor_date"), r.URL.Query().Get("step"))
|
||
writeJSON(w, map[string]any{"items": items, "jobs": enrichJobs(jobs)})
|
||
}
|
||
|
||
func (s *Server) handleAPIRecord(w http.ResponseWriter, r *http.Request) {
|
||
if s.deps.Store == nil {
|
||
writeErr(w, 503, "MySQL 未连接")
|
||
return
|
||
}
|
||
id, _ := strconv.ParseInt(r.URL.Query().Get("id"), 10, 64)
|
||
if id <= 0 {
|
||
writeErr(w, 400, "invalid id")
|
||
return
|
||
}
|
||
detail, err := GetRecordDetail(s.deps.Store, id)
|
||
if err != nil {
|
||
writeErr(w, 404, err.Error())
|
||
return
|
||
}
|
||
writeJSON(w, detail)
|
||
}
|
||
|
||
func (s *Server) handleAPIRunsTruncate(w http.ResponseWriter, r *http.Request) {
|
||
if r.Method != http.MethodPost {
|
||
writeErr(w, 405, "POST only")
|
||
return
|
||
}
|
||
if s.deps.Store == nil || !s.deps.MySQLOK {
|
||
writeErr(w, 503, "MySQL 未连接")
|
||
return
|
||
}
|
||
if syncgate.IsRunning() {
|
||
writeErr(w, 409, "已有同步任务在执行中")
|
||
return
|
||
}
|
||
if err := s.deps.Store.TruncateAuditTables(); err != nil {
|
||
applog.Appf("web truncate audit tables failed: %v", err)
|
||
writeErr(w, 500, err.Error())
|
||
return
|
||
}
|
||
applog.Appf("web truncate audit tables ok")
|
||
writeJSON(w, map[string]any{
|
||
"ok": true,
|
||
"tables": []string{"hy_push_log", "hy_push_record", "hy_sync_job"},
|
||
})
|
||
}
|
||
|
||
func (s *Server) handleAPITestStatus(w http.ResponseWriter, r *http.Request) {
|
||
writeJSON(w, enrichTestStatus(syncgate.GetStatus()))
|
||
}
|
||
|
||
func (s *Server) handleAPIConfigView(w http.ResponseWriter, r *http.Request) {
|
||
if r.Method != http.MethodGet {
|
||
writeErr(w, 405, "GET only")
|
||
return
|
||
}
|
||
sr := asSyncRunner(s.deps.Runner)
|
||
if sr == nil {
|
||
writeErr(w, 503, "未配置 Runner（请使用 transit serve）")
|
||
return
|
||
}
|
||
writeJSON(w, BuildConfigView(sr.ConfigSnapshot(), s.deps.Cfg))
|
||
}
|
||
|
||
func (s *Server) handleAPIConfigRefresh(w http.ResponseWriter, r *http.Request) {
|
||
if r.Method != http.MethodPost {
|
||
writeErr(w, 405, "POST only")
|
||
return
|
||
}
|
||
sr := asSyncRunner(s.deps.Runner)
|
||
if sr == nil {
|
||
writeErr(w, 503, "未配置 Runner（请使用 transit serve）")
|
||
return
|
||
}
|
||
if err := sr.ReloadCloudConfig(); err != nil {
|
||
writeErr(w, 502, err.Error())
|
||
return
|
||
}
|
||
writeJSON(w, BuildConfigView(sr.ConfigSnapshot(), s.deps.Cfg))
|
||
}
|
||
|
||
func (s *Server) handleAPIUploadTest(w http.ResponseWriter, r *http.Request) {
|
||
if r.Method != http.MethodPost {
|
||
writeErr(w, 405, "POST only")
|
||
return
|
||
}
|
||
sr := asSyncRunner(s.deps.Runner)
|
||
if sr == nil {
|
||
writeErr(w, 503, "未配置 Runner（请使用 transit serve）")
|
||
return
|
||
}
|
||
if err := r.ParseMultipartForm(32 << 20); err != nil {
|
||
writeErr(w, 400, "invalid multipart")
|
||
return
|
||
}
|
||
file, hdr, err := r.FormFile("file")
|
||
if err != nil {
|
||
writeErr(w, 400, "missing file field")
|
||
return
|
||
}
|
||
defer file.Close()
|
||
pdf, err := io.ReadAll(file)
|
||
if err != nil {
|
||
writeErr(w, 400, err.Error())
|
||
return
|
||
}
|
||
name := hdr.Filename
|
||
if name == "" {
|
||
name = "test.pdf"
|
||
}
|
||
res, err := sr.UploadTestPDF(pdf, name)
|
||
if err != nil {
|
||
writeUploadErr(w, 502, err.Error(), sr.FileUploadURL(), uploadTestPayload(res, sr.FileUploadURL(), false, ""))
|
||
return
|
||
}
|
||
writeJSON(w, uploadTestPayload(res, sr.FileUploadURL(), true, res.FileID))
|
||
}
|
||
|
||
func (s *Server) handleAPIRunsRetry(w http.ResponseWriter, r *http.Request) {
|
||
if r.Method != http.MethodPost {
|
||
writeErr(w, 405, "POST only")
|
||
return
|
||
}
|
||
if s.deps.Store == nil || !s.deps.MySQLOK {
|
||
writeErr(w, 503, "MySQL 未连接")
|
||
return
|
||
}
|
||
sr := asSyncRunner(s.deps.Runner)
|
||
if sr == nil {
|
||
writeErr(w, 503, "未配置 Runner（请使用 transit serve）")
|
||
return
|
||
}
|
||
if syncgate.IsRunning() {
|
||
writeErr(w, 409, "已有同步任务在执行中")
|
||
return
|
||
}
|
||
var body struct {
|
||
ID int64 `json:"id"`
|
||
}
|
||
if err := json.NewDecoder(r.Body).Decode(&body); err != nil || body.ID <= 0 {
|
||
writeErr(w, 400, "body: {\"id\": <hy_push_record.id>}")
|
||
return
|
||
}
|
||
var runErr error
|
||
if !syncgate.TryRun("retry", fmt.Sprintf("record-%d", body.ID), func() error {
|
||
runErr = sr.RetryRecord(body.ID)
|
||
return runErr
|
||
}) {
|
||
writeErr(w, 409, "已有同步任务在执行中")
|
||
return
|
||
}
|
||
if runErr != nil {
|
||
writeErr(w, 502, runErr.Error())
|
||
return
|
||
}
|
||
writeJSON(w, map[string]any{"ok": true, "id": body.ID})
|
||
}
|
||
|
||
func (s *Server) handleAPITestSync(w http.ResponseWriter, r *http.Request) {
|
||
if r.Method != http.MethodPost {
|
||
writeErr(w, 405, "POST only")
|
||
return
|
||
}
|
||
if !s.deps.AllowTest || s.deps.Runner == nil {
|
||
writeErr(w, 403, "测试同步未启用或未配置 Runner（请使用 transit serve）")
|
||
return
|
||
}
|
||
var body struct {
|
||
Step string `json:"step"`
|
||
Date string `json:"date"`
|
||
}
|
||
if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
|
||
writeErr(w, 400, "invalid json")
|
||
return
|
||
}
|
||
step := strings.TrimSpace(body.Step)
|
||
if step == "" {
|
||
step = "all"
|
||
}
|
||
anchor := syncer.ResolveAnchorDate(s.deps.Cfg, strings.TrimSpace(body.Date))
|
||
runner := s.deps.Runner
|
||
if !syncgate.TryRun(step, anchor, func() error {
|
||
applog.RunSeparator(fmt.Sprintf("BEGIN web-test step=%s date=%s", step, anchor))
|
||
applog.Appf("web test sync start step=%s date=%s", step, anchor)
|
||
err := runner.Run(step, anchor)
|
||
if err != nil {
|
||
applog.Appf("web test sync failed step=%s date=%s err=%v", step, anchor, err)
|
||
applog.RunSeparator(fmt.Sprintf("END web-test step=%s date=%s ok=false", step, anchor))
|
||
return err
|
||
}
|
||
applog.Appf("web test sync done step=%s date=%s", step, anchor)
|
||
applog.RunSeparator(fmt.Sprintf("END web-test step=%s date=%s ok=true", step, anchor))
|
||
return nil
|
||
}) {
|
||
writeErr(w, 409, "已有同步任务在执行中")
|
||
return
|
||
}
|
||
writeJSON(w, map[string]any{"ok": true, "step": step, "date": anchor})
|
||
}
|
||
|
||
// Run 启动 HTTP 服务（阻塞）。
|
||
func Run(addr string, deps Deps) error {
|
||
s := NewServer(deps)
|
||
log.Printf("logweb: 控制台 http://%s/", addr)
|
||
return http.ListenAndServe(addr, s.mux)
|
||
}
|
||
|
||
func writeJSON(w http.ResponseWriter, v any) {
|
||
w.Header().Set("Content-Type", "application/json; charset=utf-8")
|
||
enc := json.NewEncoder(w)
|
||
enc.SetIndent("", " ")
|
||
_ = enc.Encode(v)
|
||
}
|
||
|
||
func writeErr(w http.ResponseWriter, code int, msg string) {
|
||
w.Header().Set("Content-Type", "application/json; charset=utf-8")
|
||
w.WriteHeader(code)
|
||
_ = json.NewEncoder(w).Encode(map[string]string{"error": msg})
|
||
}
|
||
|
||
func writeUploadErr(w http.ResponseWriter, code int, detail, uploadURL string, extra map[string]any) {
|
||
w.Header().Set("Content-Type", "application/json; charset=utf-8")
|
||
w.WriteHeader(code)
|
||
summary := detail
|
||
if idx := strings.IndexByte(summary, '\n'); idx >= 0 {
|
||
summary = summary[:idx]
|
||
}
|
||
if len(summary) > 200 {
|
||
summary = summary[:200] + "…"
|
||
}
|
||
body := map[string]any{
|
||
"error": summary,
|
||
"detail": detail,
|
||
"uploadUrl": uploadURL,
|
||
}
|
||
for k, v := range extra {
|
||
if v == nil {
|
||
continue
|
||
}
|
||
body[k] = v
|
||
}
|
||
_ = json.NewEncoder(w).Encode(body)
|
||
}
|