diff --git a/api/admin/admin.go b/api/admin/admin.go index c360e6a..444604c 100644 --- a/api/admin/admin.go +++ b/api/admin/admin.go @@ -282,4 +282,6 @@ type IAdminV1 interface { UsageLogCleanupCreate(ctx context.Context, req *v1.UsageLogCleanupCreateReq) (res *v1.UsageLogCleanupCreateRes, err error) UsageLogCleanupList(ctx context.Context, req *v1.UsageLogCleanupListReq) (res *v1.UsageLogCleanupListRes, err error) UsageLogCleanupCancel(ctx context.Context, req *v1.UsageLogCleanupCancelReq) (res *v1.UsageLogCleanupCancelRes, err error) + AdminWorkbenchSummary(ctx context.Context, req *v1.AdminWorkbenchSummaryReq) (res *v1.AdminWorkbenchSummaryRes, err error) + AdminWorkbenchBadge(ctx context.Context, req *v1.AdminWorkbenchBadgeReq) (res *v1.AdminWorkbenchBadgeRes, err error) } diff --git a/api/admin/v1/workbench.go b/api/admin/v1/workbench.go new file mode 100644 index 0000000..5755530 --- /dev/null +++ b/api/admin/v1/workbench.go @@ -0,0 +1,109 @@ +package v1 + +import ( + "github.com/gogf/gf/v2/frame/g" +) + +// ============================================================ +// 工作台（运营收件箱） +// +// 与仪表盘（/dashboard）的分工：仪表盘回答「现在怎么样」，工作台回答「我该做什么」。 +// 只有满足「有人要处理 + 不处理会恶化 + 能被清空归零」三条的事才进工作台， +// 纯统计一律留在仪表盘。 +// ============================================================ + +// === 严重度 / 域 常量（与前端约定，勿随意改动字面量）=== +const ( + WorkbenchSeverityP0 = "p0" // 紧急：正在影响服务可用性或资金安全 + WorkbenchSeverityP1 = "p1" // 今日：当天要处理 + WorkbenchSeverityP2 = "p2" // 本周：可排期 + + WorkbenchDomainAvailability = "availability" // 服务可用性 + WorkbenchDomainMoney = "money" // 资金安全 + WorkbenchDomainCustomer = "customer" // 客户运营 + WorkbenchDomainSystem = "system" // 系统安全 +) + +// WorkbenchItem 一条待办。 +// +// Key 是稳定可重算的业务标识（如 model_no_channel:claude-sonnet-4）， +// 供前端做列表 key 与去重 —— 同一异常在两次收集之间必须算出同一个 Key， +// 所以禁止把时间戳、随机数或行 ID 拼进 Key（除非该行本身就是待办对象）。 +type WorkbenchItem struct { + Key string `json:"key" dc:"待办唯一标识（稳定可重算，用于静音匹配）"` + Severity string `json:"severity" dc:"严重度：p0=紧急 / p1=今日 / p2=本周"` + Domain string `json:"domain" dc:"所属域：availability / money / customer / system"` + Title string `json:"title" dc:"标题（一句话说清是什么问题）"` + Desc string `json:"desc" dc:"补充说明（影响面、判断依据）"` + // OccurredAt 为空表示该待办是聚合计数（如「5 条反馈待处理」）而非单点事件。 + OccurredAt string `json:"occurred_at" dc:"发生时间（RFC3339），聚合类待办为空"` + ActionText string `json:"action_text" dc:"跳转按钮文案"` + ActionRoute string `json:"action_route" dc:"跳转目标前端路由名"` + ActionQuery map[string]string `json:"action_query" dc:"跳转时附带的查询参数（自动带上筛选条件）"` +} + +// WorkbenchMetric 首屏关键数字。 +type WorkbenchMetric struct { + Key string `json:"key" dc:"指标标识"` + Label string `json:"label" dc:"展示名"` + Value float64 `json:"value" dc:"数值"` + // Unit 决定前端格式化方式：money=本位币金额 / percent=百分比 / count=计数 + Unit string `json:"unit" dc:"单位类型：money / percent / count"` + // Growth 为 nil 表示该指标没有同比口径（如「余额告警租户数」） + Growth *float64 `json:"growth" dc:"环比变化百分比；null 表示无对比口径"` + Sub string `json:"sub" dc:"副标题（说明对比口径或统计窗口）"` +} + +// WorkbenchDomainStat 分域计数。 +type WorkbenchDomainStat struct { + Domain string `json:"domain"` + Total int `json:"total" dc:"该域待办总数"` + Urgent int `json:"urgent" dc:"该域 P0 数量"` +} + +// WorkbenchModelAvail 模型可用渠道速览。 +type WorkbenchModelAvail struct { + Model string `json:"model"` + Available int `json:"available" dc:"熔断均为 CLOSED 的可用渠道数"` + Total int `json:"total" dc:"承载该模型的 active 渠道数"` +} + +// WorkbenchBreaker 正在熔断的渠道×模型。 +type WorkbenchBreaker struct { + ChannelId int64 `json:"channel_id"` + ChannelName string `json:"channel_name"` + Model string `json:"model"` + ChannelLevel bool `json:"channel_level" dc:"true=渠道级熔断（该渠道所有模型受影响）；false=仅该模型被隔离"` + HalfOpen bool `json:"half_open" dc:"是否处于半开探测态"` +} + +// === 工作台汇总 === + +type AdminWorkbenchSummaryReq struct { + g.Meta `path:"/workbench/summary" method:"get" mime:"json" tags:"管理后台-工作台" summary:"工作台汇总"` +} + +type AdminWorkbenchSummaryRes struct { + Metrics []WorkbenchMetric `json:"metrics" dc:"首屏关键数字"` + Items []WorkbenchItem `json:"items" dc:"待办列表（已按 p0→p1→p2 排序，已剔除静音项）"` + Domains []WorkbenchDomainStat `json:"domains" dc:"分域计数"` + Models []WorkbenchModelAvail `json:"models" dc:"模型可用渠道速览"` + Breakers []WorkbenchBreaker `json:"breakers" dc:"正在熔断的渠道×模型"` + GeneratedAt string `json:"generated_at" dc:"数据生成时间（RFC3339），命中缓存时为缓存写入时刻"` +} + +// === 菜单红点 === + +// AdminWorkbenchBadgeReq 轻量计数接口，供左侧菜单红点轮询。 +// 与 summary 共用同一份缓存，不会额外压库。 +type AdminWorkbenchBadgeReq struct { + g.Meta `path:"/workbench/badges" method:"get" mime:"json" tags:"管理后台-工作台" summary:"菜单红点计数"` +} + +type AdminWorkbenchBadgeRes struct { + Total int `json:"total" dc:"待办总数"` + Urgent int `json:"urgent" dc:"P0 数量"` + Domains map[string]int `json:"domains" dc:"各域待办数，键为域标识"` + // Menus 前端菜单路由名 → 待办数，直接挂角标，前端不需要自己映射域到菜单。 + Menus map[string]int `json:"menus" dc:"菜单路由名 → 待办数"` +} diff --git a/internal/controller/admin/admin_v1_workbench.go b/internal/controller/admin/admin_v1_workbench.go new file mode 100644 index 0000000..a444679 --- /dev/null +++ b/internal/controller/admin/admin_v1_workbench.go @@ -0,0 +1,15 @@ +package admin + +import ( + "context" + + "github.com/qianfree/team-api/api/admin/v1" + "github.com/qianfree/team-api/internal/service" +) + +func (c *ControllerV1) AdminWorkbenchSummary(ctx context.Context, req *v1.AdminWorkbenchSummaryReq) (res *v1.AdminWorkbenchSummaryRes, err error) { + return service.Admin().GetWorkbenchSummary(ctx, req) +} +func (c *ControllerV1) AdminWorkbenchBadge(ctx context.Context, req *v1.AdminWorkbenchBadgeReq) (res *v1.AdminWorkbenchBadgeRes, err error) { + return service.Admin().GetWorkbenchBadges(ctx, req) +} diff --git a/internal/dispatchadapter/catalog.go b/internal/dispatchadapter/catalog.go index 26386d3..e1eeb94 100644 --- a/internal/dispatchadapter/catalog.go +++ b/internal/dispatchadapter/catalog.go @@ -216,6 +216,59 @@ func (c *Catalog) ChannelRuntimeStates() map[int64]ChannelRuntimeState { return out } +// ModelAvail 单个模型的渠道可用性视图（工作台「零可用渠道」检测用）。 +// Total 为目录快照中承载该模型的 active 渠道数（disabled / 无 enabled 能力的渠道不在快照里）， +// Available 为其中渠道级与模型级熔断均为 CLOSED 的渠道数。 +// Available=0 且 Total>0 即「配了渠道但一条都用不了」——最容易被大盘总量掩盖的静默故障。 +type ModelAvail struct { + Available int + Total int + Breaking []ModelBreak +} + +// ModelBreak 一条正在熔断的渠道×模型记录。 +type ModelBreak struct { + ChannelID int64 + ChannelName string + Model string + // ChannelLevel 为 true 表示渠道级熔断（该渠道所有模型都受影响）， + // false 表示仅该模型被隔离（模型级熔断）。 + ChannelLevel bool + HalfOpen bool // 处于半开探测态 +} + +// ModelAvailability 返回目录快照中每个模型的渠道可用性与正在熔断的渠道×模型列表。 +// 目录未初始化时返回 nil，调用方按「无数据」处理，不要当成故障。 +// 并发安全：current 为 atomic.Pointer，Load() 取到的快照不可变，无锁。 +func (c *Catalog) ModelAvailability() map[string]ModelAvail { + idx := c.current.Load() + if idx == nil { + return nil + } + out := make(map[string]ModelAvail, len(idx.byModel)) + for model, chans := range idx.byModel { + av := ModelAvail{Total: len(chans)} + for _, ch := range chans { + chOpen := ch.Breaker != dispatch.BreakerClosed + mdOpen := ch.ModelBreaker != dispatch.BreakerClosed + if !chOpen && !mdOpen { + av.Available++ + continue + } + av.Breaking = append(av.Breaking, ModelBreak{ + ChannelID: ch.ID, + ChannelName: ch.Name, + Model: model, + ChannelLevel: chOpen, + HalfOpen: (chOpen && ch.Breaker == dispatch.BreakerHalfOpen) || + (!chOpen && ch.ModelBreaker == dispatch.BreakerHalfOpen), + }) + } + out[model] = av + } + return out +} + // Invalidate 主动触发一次刷新（渠道/能力/Key 管理操作后调用；跨实例走 pub/sub）。 func (c *Catalog) Invalidate() { select { diff --git a/internal/logic/admin/workbench.go b/internal/logic/admin/workbench.go new file mode 100644 index 0000000..32ff739 --- /dev/null +++ b/internal/logic/admin/workbench.go @@ -0,0 +1,266 @@ +package admin + +import ( + "context" + "encoding/json" + "sort" + "time" + + "github.com/gogf/gf/v2/frame/g" + "github.com/gogf/gf/v2/os/gtime" + + v1 "github.com/qianfree/team-api/api/admin/v1" + "github.com/qianfree/team-api/internal/logic/common" +) + +// ============================================================ +// 工作台（运营收件箱） +// +// 核心约定：待办**不落库**，每次都从各业务源表实时派生。 +// 因此「归零」是自动的 —— 工单被回复、告警被确认、渠道 Key 换好之后， +// 下一轮收集自然就查不到它了，不需要任何人来点「完成」，也不存在 +// 「源表已修好但工作台还挂着」的残留。待办的唯一出路就是去源头解决。 +// +// 收集器约定： +// 1. 每个 collector 独立失败不影响其他项 —— 任何一个查询报错只记 warning 并返回空， +// 绝不让整个工作台因为某张表出问题而白屏（工作台本身就是故障时最该能打开的页面）。 +// 2. 不做全表实时 count：高频写入表（bil_usage_logs / aud_* ）一律限定时间窗 + 走 +// 既有索引，整份结果进 Redis 缓存 30s，菜单红点与页面共用同一份缓存。 +// ============================================================ + +const ( + // workbenchCacheKey 整份收集结果的缓存键。菜单红点轮询与页面刷新共用， + // 避免每个管理员每 30s 各打一遍库。 + workbenchCacheKey = "workbench:summary:v1" + workbenchCacheTTL = 30 * time.Second + + // 各项判定阈值。选值理由见对应 collector 注释。 + wbFrozenStaleHours = 2 // 冻结超过 2 小时未释放视为异常 + wbOrderUnfulfilledMin = 15 // 已支付超过 15 分钟仍未履约视为卡单 + wbOAuthExpiryHours = 24 // OAuth 令牌 24 小时内过期即预警 + wbHealthDropPoints = 30 // 健康分 1 小时内跌幅超过 30 分视为骤降 + wbCostSpikeMultiple = 8 // 单笔消费超过近期均值 8 倍视为异常 + wbEmailFailThreshold = 5 // 1 小时内邮件失败超过 5 封视为链路异常 + wbPlanExpiringDays = 7 // 套餐 7 天内到期 + wbTicketSLAHours = 2 // 工单首次响应 SLA + wbWalletDriftRatio = 0.02 // Redis↔DB 余额偏差超过 2% 且绝对额可观时告警 +) + +// wbItem 内部待办结构：比 API 的 v1.WorkbenchItem 多一个 perm 字段。 +// 工作台跨域聚合，必须按权限逐条过滤 —— 只有 support:view 的客服不该在工作台 +// 看到钱包偏差和订单金额，否则等于绕过 RBAC 泄露财务数据。 +type wbItem struct { + v1.WorkbenchItem + Perm string `json:"perm"` +} + +// wbCollected 一次收集的完整结果（进缓存的就是它）。 +type wbCollected struct { + Items []wbItem `json:"items"` + Models []v1.WorkbenchModelAvail `json:"models"` + Breakers []v1.WorkbenchBreaker `json:"breakers"` + Metrics []v1.WorkbenchMetric `json:"metrics"` + GeneratedAt string `json:"generated_at"` +} + +// 域 → 前端菜单路由名，供菜单红点直接挂角标。 +var wbDomainMenus = map[string][]string{ + v1.WorkbenchDomainAvailability: {"AdminChannels"}, + v1.WorkbenchDomainMoney: {"AdminOrders", "AdminTransactions"}, + v1.WorkbenchDomainCustomer: {"AdminTickets", "AdminFeedback"}, + v1.WorkbenchDomainSystem: {"AdminAlertEvents", "AdminErrorLogs", "AdminCronJobs"}, +} + +// ============================================================ +// 对外接口 +// ============================================================ + +// GetWorkbenchSummary 工作台汇总。 +func (s *sAdmin) GetWorkbenchSummary(ctx context.Context, _ *v1.AdminWorkbenchSummaryReq) (*v1.AdminWorkbenchSummaryRes, error) { + col := s.collectWorkbenchCached(ctx) + + items := s.filterWorkbenchItems(ctx, col.Items) + + // 分域计数基于过滤后的结果 —— 计数必须与用户实际看得到的列表一致， + // 否则会出现「显示 3 条但列表只有 1 条」的诡异体验。 + domainStats := make([]v1.WorkbenchDomainStat, 0, 4) + for _, d := range []string{ + v1.WorkbenchDomainAvailability, v1.WorkbenchDomainMoney, + v1.WorkbenchDomainCustomer, v1.WorkbenchDomainSystem, + } { + st := v1.WorkbenchDomainStat{Domain: d} + for _, it := range items { + if it.Domain != d { + continue + } + st.Total++ + if it.Severity == v1.WorkbenchSeverityP0 { + st.Urgent++ + } + } + domainStats = append(domainStats, st) + } + + out := make([]v1.WorkbenchItem, 0, len(items)) + for _, it := range items { + out = append(out, it.WorkbenchItem) + } + + res := &v1.AdminWorkbenchSummaryRes{ + Metrics: col.Metrics, + Items: out, + Domains: domainStats, + Models: col.Models, + Breakers: col.Breakers, + GeneratedAt: col.GeneratedAt, + } + if res.Metrics == nil { + res.Metrics = []v1.WorkbenchMetric{} + } + if res.Models == nil { + res.Models = []v1.WorkbenchModelAvail{} + } + if res.Breakers == nil { + res.Breakers = []v1.WorkbenchBreaker{} + } + return res, nil +} + +// GetWorkbenchBadges 菜单红点计数（与 summary 共用缓存，不额外压库）。 +func (s *sAdmin) GetWorkbenchBadges(ctx context.Context, _ *v1.AdminWorkbenchBadgeReq) (*v1.AdminWorkbenchBadgeRes, error) { + col := s.collectWorkbenchCached(ctx) + items := s.filterWorkbenchItems(ctx, col.Items) + + res := &v1.AdminWorkbenchBadgeRes{ + Domains: map[string]int{}, + Menus: map[string]int{}, + } + for _, it := range items { + res.Total++ + if it.Severity == v1.WorkbenchSeverityP0 { + res.Urgent++ + } + res.Domains[it.Domain]++ + } + for domain, menus := range wbDomainMenus { + n := res.Domains[domain] + if n == 0 { + continue + } + for _, m := range menus { + res.Menus[m] += n + } + } + return res, nil +} + +// ============================================================ +// 收集与过滤 +// ============================================================ + +// collectWorkbenchCached 读缓存，未命中则重新收集。 +// 缓存失败一律降级为直接收集 —— 工作台不能因为 Redis 抖动就打不开。 +func (s *sAdmin) collectWorkbenchCached(ctx context.Context) *wbCollected { + if v, err := g.Redis().Get(ctx, workbenchCacheKey); err == nil && !v.IsEmpty() { + var col wbCollected + if json.Unmarshal(v.Bytes(), &col) == nil { + return &col + } + } + + col := s.collectWorkbench(ctx) + + if b, err := json.Marshal(col); err == nil { + if _, err := g.Redis().Do(ctx, "SET", workbenchCacheKey, b, "EX", int(workbenchCacheTTL.Seconds())); err != nil { + g.Log().Warningf(ctx, "workbench: 写缓存失败: %v", err) + } + } + return col +} + +// filterWorkbenchItems 按当前管理员权限逐条过滤，再按严重度排序。 +func (s *sAdmin) filterWorkbenchItems(ctx context.Context, items []wbItem) []wbItem { + userID := common.GetCtxUserID(ctx) + role, _ := ctx.Value("role").(string) + + // 权限判定结果按 perm 记忆化：一次汇总最多 20 条待办但只有 ~10 种权限点， + // 逐条查库会把一次页面刷新放大成几十次 SQL。 + permCache := make(map[string]bool, 12) + allowed := func(perm string) bool { + if perm == "" { + return true + } + if v, ok := permCache[perm]; ok { + return v + } + v := HasPermission(ctx, userID, role, perm) + permCache[perm] = v + return v + } + + out := make([]wbItem, 0, len(items)) + for _, it := range items { + if !allowed(it.Perm) { + continue + } + out = append(out, it) + } + + sevRank := map[string]int{ + v1.WorkbenchSeverityP0: 0, + v1.WorkbenchSeverityP1: 1, + v1.WorkbenchSeverityP2: 2, + } + sort.SliceStable(out, func(i, j int) bool { + ri, rj := sevRank[out[i].Severity], sevRank[out[j].Severity] + if ri != rj { + return ri < rj + } + // 同 severity 下新发生的排前面；聚合类（OccurredAt 为空）排最后 + if out[i].OccurredAt != out[j].OccurredAt { + if out[i].OccurredAt == "" { + return false + } + if out[j].OccurredAt == "" { + return true + } + return out[i].OccurredAt > out[j].OccurredAt + } + return out[i].Key < out[j].Key + }) + return out +} + +// collectWorkbench 依次跑各 collector。单个失败不影响整体。 +func (s *sAdmin) collectWorkbench(ctx context.Context) *wbCollected { + col := &wbCollected{GeneratedAt: gtime.Now().Format(time.RFC3339)} + + models, breakers, availItems := s.wbCollectAvailability(ctx) + col.Models = models + col.Breakers = breakers + + col.Items = append(col.Items, availItems...) + col.Items = append(col.Items, s.wbCollectMoney(ctx)...) + col.Items = append(col.Items, s.wbCollectCustomer(ctx)...) + col.Items = append(col.Items, s.wbCollectSystem(ctx)...) + + col.Metrics = s.wbCollectMetrics(ctx) + return col +} + +// wbNew 构造一条待办。 +func wbNew(key, sev, domain, perm, title, desc, actionText, route string, query map[string]string, at *gtime.Time) wbItem { + it := wbItem{Perm: perm} + it.Key = key + it.Severity = sev + it.Domain = domain + it.Title = title + it.Desc = desc + it.ActionText = actionText + it.ActionRoute = route + it.ActionQuery = query + if at != nil { + it.OccurredAt = at.Format(time.RFC3339) + } + return it +} diff --git a/internal/logic/admin/workbench_collectors.go b/internal/logic/admin/workbench_collectors.go new file mode 100644 index 0000000..187489c --- /dev/null +++ b/internal/logic/admin/workbench_collectors.go @@ -0,0 +1,989 @@ +package admin + +import ( + "context" + "fmt" + "strings" + "time" + + "github.com/gogf/gf/v2/frame/g" + "github.com/gogf/gf/v2/os/gtime" + + v1 "github.com/qianfree/team-api/api/admin/v1" + "github.com/qianfree/team-api/internal/dispatchadapter" + "github.com/qianfree/team-api/internal/logic/billing" +) + +// ============================================================ +// 服务可用性（channel:view） +// ============================================================ + +// wbCollectAvailability 渠道与模型可用性。 +// 返回模型可用渠道速览、正在熔断列表，以及派生出的待办。 +func (s *sAdmin) wbCollectAvailability(ctx context.Context) ([]v1.WorkbenchModelAvail, []v1.WorkbenchBreaker, []wbItem) { + items := make([]wbItem, 0, 8) + + // --- 目录快照：模型 → 可用/总渠道数 + 熔断明细 --- + var avail map[string]dispatchadapter.ModelAvail + if cat := dispatchadapter.CatalogInstance(); cat != nil { + avail = cat.ModelAvailability() + } + + models := make([]v1.WorkbenchModelAvail, 0, len(avail)) + breakers := make([]v1.WorkbenchBreaker, 0, 8) + for model, av := range avail { + models = append(models, v1.WorkbenchModelAvail{ + Model: model, Available: av.Available, Total: av.Total, + }) + for _, b := range av.Breaking { + breakers = append(breakers, v1.WorkbenchBreaker{ + ChannelId: b.ChannelID, + ChannelName: b.ChannelName, + Model: b.Model, + ChannelLevel: b.ChannelLevel, + HalfOpen: b.HalfOpen, + }) + } + + // P0：配了渠道但一条都用不了。 + // 这是最容易被大盘掩盖的静默故障 —— 单个模型全挂时，平台总请求量可能只掉几个 + // 百分点，成功率曲线也被其他健康模型摊薄，但该模型的客户已经 100% 失败了。 + if av.Total > 0 && av.Available == 0 { + items = append(items, wbNew( + "model_no_channel:"+model, + v1.WorkbenchSeverityP0, v1.WorkbenchDomainAvailability, "channel:view", + fmt.Sprintf("模型 %s 零可用渠道", model), + fmt.Sprintf("承载该模型的 %d 个渠道全部处于熔断或不可用状态，该模型请求当前必然失败", av.Total), + "查看渠道", "AdminChannels", map[string]string{"status": "active"}, nil, + )) + } + } + // 排序保证响应稳定：map 遍历顺序随机会让前端列表每次刷新都跳动 + sortModelAvail(models) + sortBreakers(breakers) + + // P1：渠道级熔断（该渠道所有模型受影响，但还没到零可用） + chLevel := map[int64]string{} + for _, b := range breakers { + if b.ChannelLevel && !b.HalfOpen { + chLevel[b.ChannelId] = b.ChannelName + } + } + for id, name := range chLevel { + items = append(items, wbNew( + fmt.Sprintf("channel_breaker:%d", id), + v1.WorkbenchSeverityP1, v1.WorkbenchDomainAvailability, "channel:view", + fmt.Sprintf("渠道 %s 已熔断", name), + "该渠道全部模型已被调度器摘除，容量下降，恢复前流量会压到其余渠道", + "查看渠道", "AdminChannelDetail", map[string]string{"id": fmt.Sprint(id)}, nil, + )) + } + + items = append(items, s.wbChannelKeyIssues(ctx)...) + items = append(items, s.wbOAuthExpiring(ctx)...) + items = append(items, s.wbHealthDrop(ctx)...) + + return models, breakers, items +} + +// wbChannelKeyIssues 渠道 Key 失效 / 额度耗尽。 +// Key 挂掉时渠道整体仍是 active，容量却已悄悄下降 —— 列表页看不出来。 +func (s *sAdmin) wbChannelKeyIssues(ctx context.Context) []wbItem { + rows, err := g.DB().Ctx(ctx).Query(ctx, + `SELECT k.id, k.channel_id, k.name, k.status, k.last_error, c.name AS channel_name, + (SELECT COUNT(*) FROM chn_channel_keys k2 WHERE k2.channel_id = k.channel_id) AS total_keys + FROM chn_channel_keys k + JOIN chn_channels c ON c.id = k.channel_id + WHERE k.status <> 'active' AND c.status = 'active' + ORDER BY k.updated_at DESC + LIMIT 20`) + if err != nil { + g.Log().Warningf(ctx, "workbench: 查询异常渠道 Key 失败: %v", err) + return nil + } + + items := make([]wbItem, 0, len(rows)) + for _, r := range rows { + status := r["status"].String() + label := "已禁用" + if status == "exhausted" { + label = "额度耗尽" + } + total := r["total_keys"].Int() + desc := fmt.Sprintf("渠道「%s」的 Key「%s」%s", r["channel_name"].String(), r["name"].String(), label) + if total > 1 { + desc += fmt.Sprintf("，该渠道可用 Key %d/%d，容量下降", total-1, total) + } else { + desc += "，该渠道已无可用 Key" + } + if e := strings.TrimSpace(r["last_error"].String()); e != "" { + desc += "；最后错误：" + truncate(e, 120) + } + + items = append(items, wbNew( + fmt.Sprintf("channel_key_invalid:%d", r["id"].Int64()), + v1.WorkbenchSeverityP1, v1.WorkbenchDomainAvailability, "channel:view", + fmt.Sprintf("%s Key %s", r["channel_name"].String(), label), + desc, + "查看渠道", "AdminChannelDetail", + map[string]string{"id": r["channel_id"].String()}, nil, + )) + } + return items +} + +// wbOAuthExpiring OAuth 令牌临期。 +// 极其隐蔽：过期那一刻整条渠道直接不可用，且没有任何渐进信号。 +func (s *sAdmin) wbOAuthExpiring(ctx context.Context) []wbItem { + rows, err := g.DB().Ctx(ctx).Query(ctx, + `SELECT k.id, k.channel_id, k.name, k.token_expires_at, c.name AS channel_name + FROM chn_channel_keys k + JOIN chn_channels c ON c.id = k.channel_id + WHERE k.key_type = 'oauth' + AND c.status = 'active' + AND k.token_expires_at IS NOT NULL + AND k.token_expires_at <= ? + ORDER BY k.token_expires_at ASC + LIMIT 20`, gtime.Now().Add(wbOAuthExpiryHours*time.Hour)) + if err != nil { + g.Log().Warningf(ctx, "workbench: 查询 OAuth 临期令牌失败: %v", err) + return nil + } + + items := make([]wbItem, 0, len(rows)) + for _, r := range rows { + expAt := r["token_expires_at"].GTime() + expired := expAt != nil && expAt.Before(gtime.Now()) + + sev := v1.WorkbenchSeverityP1 + title := fmt.Sprintf("%s OAuth 令牌即将过期", r["channel_name"].String()) + desc := fmt.Sprintf("Key「%s」将于 %s 过期，过期后该渠道立即不可用", + r["name"].String(), fmtTime(expAt)) + if expired { + // 已经过期就不是预警而是故障了 + sev = v1.WorkbenchSeverityP0 + title = fmt.Sprintf("%s OAuth 令牌已过期", r["channel_name"].String()) + desc = fmt.Sprintf("Key「%s」已于 %s 过期，该渠道当前不可用", + r["name"].String(), fmtTime(expAt)) + } + + items = append(items, wbNew( + fmt.Sprintf("channel_oauth_expiring:%d", r["id"].Int64()), + sev, v1.WorkbenchDomainAvailability, "channel:view", + title, desc, + "重新授权", "AdminChannelDetail", + map[string]string{"id": r["channel_id"].String()}, nil, + )) + } + return items +} + +// wbHealthDrop 健康分骤降。 +// 判据是「相对自己 1 小时前的跌幅」而非绝对分值 —— 一个长期 60 分的渠道是已知状态， +// 而 92 分跌到 61 分是正在发生的事故，绝对阈值分不出这两者。 +func (s *sAdmin) wbHealthDrop(ctx context.Context) []wbItem { + rows, err := g.DB().Ctx(ctx).Query(ctx, + `WITH latest AS ( + SELECT DISTINCT ON (channel_id) channel_id, health_score, snapshot_at + FROM chn_health_snapshots + WHERE snapshot_at >= NOW() - INTERVAL '15 minutes' + ORDER BY channel_id, snapshot_at DESC + ), baseline AS ( + SELECT DISTINCT ON (channel_id) channel_id, health_score + FROM chn_health_snapshots + WHERE snapshot_at BETWEEN NOW() - INTERVAL '75 minutes' AND NOW() - INTERVAL '45 minutes' + ORDER BY channel_id, snapshot_at DESC + ) + SELECT l.channel_id, c.name AS channel_name, + ROUND(l.health_score)::int AS now_score, + ROUND(b.health_score)::int AS prev_score, + l.snapshot_at + FROM latest l + JOIN baseline b ON b.channel_id = l.channel_id + JOIN chn_channels c ON c.id = l.channel_id + WHERE c.status = 'active' + AND b.health_score - l.health_score >= ? + ORDER BY (b.health_score - l.health_score) DESC + LIMIT 10`, wbHealthDropPoints) + if err != nil { + g.Log().Warningf(ctx, "workbench: 查询健康分骤降失败: %v", err) + return nil + } + + items := make([]wbItem, 0, len(rows)) + for _, r := range rows { + items = append(items, wbNew( + fmt.Sprintf("channel_health_drop:%d", r["channel_id"].Int64()), + v1.WorkbenchSeverityP1, v1.WorkbenchDomainAvailability, "channel:view", + fmt.Sprintf("%s 健康分骤降", r["channel_name"].String()), + fmt.Sprintf("1 小时内从 %d 分跌至 %d 分，调度权重已自动下调，尚未触发熔断", + r["prev_score"].Int(), r["now_score"].Int()), + "查看健康趋势", "AdminChannelDetail", + map[string]string{"id": r["channel_id"].String()}, + r["snapshot_at"].GTime(), + )) + } + return items +} + +// ============================================================ +// 资金安全（billing:view / order:view） +// ============================================================ + +func (s *sAdmin) wbCollectMoney(ctx context.Context) []wbItem { + items := make([]wbItem, 0, 8) + items = append(items, s.wbWalletDrift(ctx)...) + items = append(items, s.wbFrozenStale(ctx)...) + items = append(items, s.wbOrderUnfulfilled(ctx)...) + items = append(items, s.wbCostSpike(ctx)...) + items = append(items, s.wbLowBalanceTenants(ctx)...) + return items +} + +// wbWalletDrift 钱包 Redis↔DB 物化偏差。 +// +// 按架构约定 Redis 是余额唯一实时权威，DB 只是每 5s 物化一次的滞后副本。 +// 正常情况下二者差异很小；持续放大的偏差意味着物化器停摆或 Lua 与账本不一致， +// 此时 DB 余额不可用于对账，且问题会随时间累积 —— 属于必须立刻知道的一类。 +func (s *sAdmin) wbWalletDrift(ctx context.Context) []wbItem { + // 只抽查有余额的活跃钱包，避免全表逐个读 Redis + rows, err := g.DB().Ctx(ctx).Query(ctx, + `SELECT w.tenant_id, w.balance, t.name AS tenant_name + FROM bil_wallets w + JOIN tnt_tenants t ON t.id = w.tenant_id + WHERE w.balance > 0 AND t.status IN ('active', 'trial') + ORDER BY w.balance DESC + LIMIT 30`) + if err != nil { + g.Log().Warningf(ctx, "workbench: 查询钱包列表失败: %v", err) + return nil + } + + var ( + worstTenant string + worstDrift float64 + driftCount int + ) + for _, r := range rows { + tenantID := r["tenant_id"].Int64() + dbBalance := r["balance"].Float64() + + info, err := billing.GetWallet(ctx, tenantID) + if err != nil || info == nil { + continue + } + drift := info.Balance - dbBalance + if drift < 0 { + drift = -drift + } + // 相对阈值 + 绝对下限：小额账户的分币级差异不值得报警 + if dbBalance > 0 && drift/dbBalance > wbWalletDriftRatio && drift > 1 { + driftCount++ + if drift > worstDrift { + worstDrift = drift + worstTenant = r["tenant_name"].String() + } + } + } + + if driftCount == 0 { + return nil + } + return []wbItem{wbNew( + "wallet_drift", + v1.WorkbenchSeverityP0, v1.WorkbenchDomainMoney, "billing:view", + "钱包 Redis 与 DB 物化余额偏差", + fmt.Sprintf("抽查发现 %d 个钱包偏差超阈值，最大偏差 %.2f（租户「%s」）。"+ + "DB 余额由物化器每 5 秒覆盖，持续偏差说明物化器可能停摆，当前 DB 余额不可用于对账", + driftCount, worstDrift, worstTenant), + "查看定时任务", "AdminCronJobs", nil, nil, + )} +} + +// wbFrozenStale 长时间未释放的冻结。 +// 客户余额被凭空锁住，是客服工单的高发源，且往往是异步任务未回收的信号。 +func (s *sAdmin) wbFrozenStale(ctx context.Context) []wbItem { + rows, err := g.DB().Ctx(ctx).Query(ctx, + `SELECT w.tenant_id, t.name AS tenant_name + FROM bil_wallets w + JOIN tnt_tenants t ON t.id = w.tenant_id + WHERE w.frozen_balance > 0 + ORDER BY w.frozen_balance DESC + LIMIT 20`) + if err != nil { + g.Log().Warningf(ctx, "workbench: 查询冻结钱包失败: %v", err) + return nil + } + + items := make([]wbItem, 0, 4) + for _, r := range rows { + tenantID := r["tenant_id"].Int64() + frozen, err := billing.GetFrozenItems(ctx, tenantID) + if err != nil { + continue + } + var ( + staleCount int + staleTotal float64 + ) + // FrozenItem.CreatedAt 是 Unix 秒（预扣 Lua 写入），与既有冻结明细接口同口径 + staleBefore := time.Now().Add(-wbFrozenStaleHours * time.Hour).Unix() + for _, f := range frozen { + if f.CreatedAt > 0 && f.CreatedAt <= staleBefore { + staleCount++ + staleTotal += f.Amount + } + } + if staleCount == 0 { + continue + } + items = append(items, wbNew( + fmt.Sprintf("frozen_stale:%d", tenantID), + v1.WorkbenchSeverityP1, v1.WorkbenchDomainMoney, "billing:view", + fmt.Sprintf("租户「%s」%d 笔冻结超时未释放", r["tenant_name"].String(), staleCount), + fmt.Sprintf("合计 %.6f 被锁定超过 %d 小时，客户可用余额受影响，通常是异步任务未回收", + staleTotal, wbFrozenStaleHours), + "查看冻结明细", "AdminTenantDetail", + map[string]string{"id": fmt.Sprint(tenantID), "tab": "wallet"}, nil, + )) + } + return items +} + +// wbOrderUnfulfilled 已支付未履约的卡单。真金白银，客户会追。 +func (s *sAdmin) wbOrderUnfulfilled(ctx context.Context) []wbItem { + rows, err := g.DB().Ctx(ctx).Query(ctx, + `SELECT o.id, o.order_no, o.final_amount, o.paid_at, o.order_type, t.name AS tenant_name + FROM ord_orders o + LEFT JOIN tnt_tenants t ON t.id = o.tenant_id + WHERE o.status = 'paid' + AND o.fulfilled_at IS NULL + AND o.paid_at IS NOT NULL + AND o.paid_at <= ? + ORDER BY o.paid_at ASC + LIMIT 20`, gtime.Now().Add(-wbOrderUnfulfilledMin*time.Minute)) + if err != nil { + g.Log().Warningf(ctx, "workbench: 查询卡单失败: %v", err) + return nil + } + + items := make([]wbItem, 0, len(rows)) + for _, r := range rows { + paidAt := r["paid_at"].GTime() + items = append(items, wbNew( + "order_unfulfilled:"+r["order_no"].String(), + v1.WorkbenchSeverityP1, v1.WorkbenchDomainMoney, "order:view", + fmt.Sprintf("订单 %s 已支付未履约", r["order_no"].String()), + fmt.Sprintf("租户「%s」支付 ¥%.2f 于 %s，至今未履约，客户可能已在催", + r["tenant_name"].String(), r["final_amount"].Float64(), fmtTime(paidAt)), + "处理订单", "AdminOrders", + map[string]string{"order_no": r["order_no"].String()}, paidAt, + )) + } + return items +} + +// wbCostSpike 单笔消费异常。 +// 用途有三：防刷、防客户被恶意调用、防定价配置写错一位（乘数多个 0 的杀伤力最大）。 +// 判据是「相对该租户自身近 7 天均值的倍数」，而非全局绝对值 —— 大客户的正常单笔 +// 对小客户就是天价，全局阈值必然两头不讨好。 +func (s *sAdmin) wbCostSpike(ctx context.Context) []wbItem { + rows, err := g.DB().Ctx(ctx).Query(ctx, + `WITH recent AS ( + SELECT tenant_id, id, total_cost, model_name, created_at + FROM bil_usage_logs + WHERE created_at >= NOW() - INTERVAL '2 hours' AND total_cost > 0 + ), baseline AS ( + SELECT tenant_id, AVG(total_cost) AS avg_cost + FROM bil_usage_logs + WHERE created_at >= NOW() - INTERVAL '7 days' AND total_cost > 0 + GROUP BY tenant_id + HAVING COUNT(*) >= 50 + ) + SELECT r.id, r.tenant_id, r.total_cost, r.model_name, r.created_at, + b.avg_cost, t.name AS tenant_name + FROM recent r + JOIN baseline b ON b.tenant_id = r.tenant_id + LEFT JOIN tnt_tenants t ON t.id = r.tenant_id + WHERE r.total_cost >= b.avg_cost * ? + ORDER BY r.total_cost DESC + LIMIT 5`, wbCostSpikeMultiple) + if err != nil { + g.Log().Warningf(ctx, "workbench: 查询消费尖刺失败: %v", err) + return nil + } + + items := make([]wbItem, 0, len(rows)) + for _, r := range rows { + cost := r["total_cost"].Float64() + avg := r["avg_cost"].Float64() + mult := 0.0 + if avg > 0 { + mult = cost / avg + } + items = append(items, wbNew( + fmt.Sprintf("usage_cost_spike:%d", r["id"].Int64()), + v1.WorkbenchSeverityP1, v1.WorkbenchDomainMoney, "billing:view", + fmt.Sprintf("租户「%s」单笔消费异常", r["tenant_name"].String()), + fmt.Sprintf("模型 %s 单笔 %.6f，为该租户近 7 天均值的 %.1f 倍。"+ + "需确认是正常业务、刷量，还是定价配置写错", + r["model_name"].String(), cost, mult), + "查看用量日志", "AdminUsageLogs", + map[string]string{"tenant_id": r["tenant_id"].String()}, r["created_at"].GTime(), + )) + } + return items +} + +// wbLowBalanceTenants 余额低于预警阈值的租户。 +// 一个信号两个用途：流失预警 + 催充窗口。聚合成一条，不逐租户刷屏。 +func (s *sAdmin) wbLowBalanceTenants(ctx context.Context) []wbItem { + rows, err := g.DB().Ctx(ctx).Query(ctx, + `SELECT COUNT(*) AS cnt, COALESCE(SUM(w.balance - w.frozen_balance), 0) AS available + FROM bil_wallets w + JOIN tnt_tenants t ON t.id = w.tenant_id + WHERE t.status IN ('active', 'trial') + AND w.warning_threshold > 0 + AND (w.balance - w.frozen_balance) < w.warning_threshold`) + if err != nil || len(rows) == 0 { + if err != nil { + g.Log().Warningf(ctx, "workbench: 查询低余额租户失败: %v", err) + } + return nil + } + + cnt := rows[0]["cnt"].Int() + if cnt == 0 { + return nil + } + return []wbItem{wbNew( + "tenant_low_balance", + v1.WorkbenchSeverityP2, v1.WorkbenchDomainCustomer, "billing:view", + fmt.Sprintf("%d 个租户余额低于预警阈值", cnt), + fmt.Sprintf("合计可用余额 %.6f，既是服务中断风险，也是催充与流失干预的窗口", + rows[0]["available"].Float64()), + "查看租户", "AdminTenants", nil, nil, + )} +} + +// ============================================================ +// 客户运营（support:view / plan:view） +// ============================================================ + +func (s *sAdmin) wbCollectCustomer(ctx context.Context) []wbItem { + items := make([]wbItem, 0, 8) + items = append(items, s.wbTicketIssues(ctx)...) + items = append(items, s.wbFeedbackPending(ctx)...) + items = append(items, s.wbPlanExpiring(ctx)...) + return items +} + +// wbTicketIssues 工单：只报「超 SLA 未首响」和「未分配」两类。 +// 刻意不报工单总数 —— 那是个永远不为零的数字，放进待办只会制造永久红点。 +func (s *sAdmin) wbTicketIssues(ctx context.Context) []wbItem { + items := make([]wbItem, 0, 4) + + // 超 SLA 未首次响应：pending 且创建超过 SLA 时长 + rows, err := g.DB().Ctx(ctx).Query(ctx, + `SELECT k.id, k.title, k.urgency, k.created_at, t.name AS tenant_name + FROM spt_tickets k + LEFT JOIN tnt_tenants t ON t.id = k.tenant_id + WHERE k.status = 'pending' + AND k.created_at <= ? + ORDER BY k.created_at ASC + LIMIT 10`, gtime.Now().Add(-wbTicketSLAHours*time.Hour)) + if err != nil { + g.Log().Warningf(ctx, "workbench: 查询超时工单失败: %v", err) + } else { + for _, r := range rows { + createdAt := r["created_at"].GTime() + sev := v1.WorkbenchSeverityP1 + if u := r["urgency"].String(); u == "urgent" { + sev = v1.WorkbenchSeverityP0 + } + items = append(items, wbNew( + fmt.Sprintf("ticket_sla_breach:%d", r["id"].Int64()), + sev, v1.WorkbenchDomainCustomer, "support:view", + fmt.Sprintf("工单「%s」超 SLA 未首次响应", truncate(r["title"].String(), 40)), + fmt.Sprintf("提交方「%s」，已等待 %s（SLA %d 小时）", + r["tenant_name"].String(), humanizeSince(createdAt), wbTicketSLAHours), + "去回复", "AdminTickets", + map[string]string{"id": r["id"].String()}, createdAt, + )) + } + } + + // 未分配工单：聚合成一条 + cntRows, err := g.DB().Ctx(ctx).Query(ctx, + `SELECT COUNT(*) AS cnt FROM spt_tickets + WHERE status IN ('pending', 'reopened') + AND (assigned_admin_id IS NULL OR assigned_admin_id = 0)`) + if err == nil && len(cntRows) > 0 { + if n := cntRows[0]["cnt"].Int(); n > 0 { + items = append(items, wbNew( + "ticket_unassigned", + v1.WorkbenchSeverityP2, v1.WorkbenchDomainCustomer, "support:view", + fmt.Sprintf("%d 个工单未分配处理人", n), + "无人认领的工单容易在交接中被漏掉，建议指派到人", + "去分配", "AdminTickets", nil, nil, + )) + } + } + return items +} + +// wbFeedbackPending 待处理反馈（聚合一条 + 最长等待时间）。 +func (s *sAdmin) wbFeedbackPending(ctx context.Context) []wbItem { + rows, err := g.DB().Ctx(ctx).Query(ctx, + `SELECT COUNT(*) AS cnt, MIN(created_at) AS oldest + FROM spt_feedbacks WHERE status = 'pending'`) + if err != nil || len(rows) == 0 { + if err != nil { + g.Log().Warningf(ctx, "workbench: 查询待处理反馈失败: %v", err) + } + return nil + } + cnt := rows[0]["cnt"].Int() + if cnt == 0 { + return nil + } + oldest := rows[0]["oldest"].GTime() + + return []wbItem{wbNew( + "feedback_pending", + v1.WorkbenchSeverityP2, v1.WorkbenchDomainCustomer, "support:view", + fmt.Sprintf("%d 条用户反馈待处理", cnt), + fmt.Sprintf("最早 1 条已等待 %s", humanizeSince(oldest)), + "去处理", "AdminFeedback", nil, nil, + )} +} + +// wbPlanExpiring 套餐即将到期且未开启自动续费 —— 续费转化窗口。 +func (s *sAdmin) wbPlanExpiring(ctx context.Context) []wbItem { + rows, err := g.DB().Ctx(ctx).Query(ctx, + `SELECT COUNT(*) AS cnt + FROM pln_tenant_plans + WHERE status = 'active' + AND auto_renew = false + AND end_at IS NOT NULL + AND end_at BETWEEN NOW() AND ?`, gtime.Now().AddDate(0, 0, wbPlanExpiringDays)) + if err != nil || len(rows) == 0 { + if err != nil { + g.Log().Warningf(ctx, "workbench: 查询到期套餐失败: %v", err) + } + return nil + } + cnt := rows[0]["cnt"].Int() + if cnt == 0 { + return nil + } + return []wbItem{wbNew( + "plan_expiring", + v1.WorkbenchSeverityP2, v1.WorkbenchDomainCustomer, "plan:view", + fmt.Sprintf("%d 个租户套餐 %d 天内到期", cnt, wbPlanExpiringDays), + "均未开启自动续费，到期即停服，建议提前触达", + "查看租户", "AdminTenants", nil, nil, + )} +} + +// ============================================================ +// 系统与安全（system:view / monitor:view / audit:view） +// ============================================================ + +func (s *sAdmin) wbCollectSystem(ctx context.Context) []wbItem { + items := make([]wbItem, 0, 8) + items = append(items, s.wbCronStalled(ctx)...) + items = append(items, s.wbEmailFailures(ctx)...) + items = append(items, s.wbAlertUnacked(ctx)...) + items = append(items, s.wbErrorLogsUnresolved(ctx)...) + items = append(items, s.wbNewIPLogin(ctx)...) + return items +} + +// wbCronStalled 定时任务失败或停摆。 +// 资金相关任务（预扣清扫、钱包物化）挂掉会直接导致资损，且完全无声。 +func (s *sAdmin) wbCronStalled(ctx context.Context) []wbItem { + rows, err := g.DB().Ctx(ctx).Query(ctx, + `SELECT job_name, last_status, last_started_at, last_error_message + FROM sys_cron_jobs + WHERE last_status = 'failed' + OR last_started_at IS NULL + OR last_started_at <= NOW() - INTERVAL '1 hour' + ORDER BY last_started_at ASC NULLS FIRST + LIMIT 10`) + if err != nil { + g.Log().Warningf(ctx, "workbench: 查询定时任务状态失败: %v", err) + return nil + } + + items := make([]wbItem, 0, len(rows)) + for _, r := range rows { + name := r["job_name"].String() + startedAt := r["last_started_at"].GTime() + failed := r["last_status"].String() == "failed" + + // 资金相关任务提级为 P0：这些停摆会累积资损，不能等到明天 + sev := v1.WorkbenchSeverityP1 + if isMoneyCriticalJob(name) { + sev = v1.WorkbenchSeverityP0 + } + + desc := "" + if failed { + desc = "最近一次执行失败" + if e := strings.TrimSpace(r["last_error_message"].String()); e != "" { + desc += "：" + truncate(e, 150) + } + } else if startedAt == nil { + desc = "自服务启动以来从未执行过，调度器可能未正常注册该任务" + } else { + desc = fmt.Sprintf("已 %s 未执行，任务可能已停摆", humanizeSince(startedAt)) + } + if isMoneyCriticalJob(name) { + desc += "。该任务涉及资金结算，停摆会持续累积差错" + } + + items = append(items, wbNew( + "cron_stalled:"+name, + sev, v1.WorkbenchDomainSystem, "system:view", + fmt.Sprintf("定时任务 %s 异常", name), + desc, + "查看定时任务", "AdminCronJobs", nil, startedAt, + )) + } + return items +} + +// isMoneyCriticalJob 判断任务停摆是否会造成资损。 +func isMoneyCriticalJob(name string) bool { + n := strings.ToLower(name) + for _, kw := range []string{"prededuct", "wallet", "materiali", "settle", "billing", "refund"} { + if strings.Contains(n, kw) { + return true + } + } + return false +} + +// wbEmailFailures 邮件发送失败。 +// 隐蔽度极高：验证码发不出去等于新用户注册与找回密码链路整条中断， +// 而受影响的人根本没有渠道告诉你。 +func (s *sAdmin) wbEmailFailures(ctx context.Context) []wbItem { + rows, err := g.DB().Ctx(ctx).Query(ctx, + `SELECT COUNT(*) AS cnt, + COUNT(*) FILTER (WHERE template_code ILIKE '%verify%' OR template_code ILIKE '%code%') AS verify_cnt, + MAX(error_message) AS sample_error + FROM ntf_send_log + WHERE channel = 'email' + AND status = 'failed' + AND created_at >= NOW() - INTERVAL '1 hour'`) + if err != nil || len(rows) == 0 { + if err != nil { + g.Log().Warningf(ctx, "workbench: 查询邮件失败记录失败: %v", err) + } + return nil + } + + cnt := rows[0]["cnt"].Int() + if cnt < wbEmailFailThreshold { + return nil + } + + desc := fmt.Sprintf("近 1 小时内 %d 封邮件发送失败", cnt) + if vc := rows[0]["verify_cnt"].Int(); vc > 0 { + desc += fmt.Sprintf("，其中 %d 封为验证码类 —— 新用户注册与找回密码链路已实际中断", vc) + } + if e := strings.TrimSpace(rows[0]["sample_error"].String()); e != "" { + desc += "；样本错误：" + truncate(e, 120) + } + + return []wbItem{wbNew( + "email_failures", + v1.WorkbenchSeverityP1, v1.WorkbenchDomainSystem, "system:view", + "邮件发送连续失败", + desc, + "查看发送记录", "AdminEmailLogs", map[string]string{"status": "failed"}, nil, + )} +} + +// wbAlertUnacked 未确认告警（聚合一条）。 +func (s *sAdmin) wbAlertUnacked(ctx context.Context) []wbItem { + rows, err := g.DB().Ctx(ctx).Query(ctx, + `SELECT COUNT(*) AS cnt, + COUNT(*) FILTER (WHERE level = 'critical') AS critical_cnt, + MIN(created_at) AS oldest + FROM ops_alert_events + WHERE status = 'firing'`) + if err != nil || len(rows) == 0 { + if err != nil { + g.Log().Warningf(ctx, "workbench: 查询未确认告警失败: %v", err) + } + return nil + } + cnt := rows[0]["cnt"].Int() + if cnt == 0 { + return nil + } + + criticalCnt := rows[0]["critical_cnt"].Int() + sev := v1.WorkbenchSeverityP2 + if criticalCnt > 0 { + sev = v1.WorkbenchSeverityP1 + } + desc := fmt.Sprintf("最早 1 条触发于 %s", humanizeSince(rows[0]["oldest"].GTime())) + if criticalCnt > 0 { + desc = fmt.Sprintf("其中 %d 条为严重级别；", criticalCnt) + desc + } + + return []wbItem{wbNew( + "alert_unack", + sev, v1.WorkbenchDomainSystem, "monitor:view", + fmt.Sprintf("%d 条告警事件未确认", cnt), + desc, + "去确认", "AdminAlertEvents", map[string]string{"status": "firing"}, nil, + )} +} + +// wbErrorLogsUnresolved 未处理系统错误日志（聚合一条）。 +func (s *sAdmin) wbErrorLogsUnresolved(ctx context.Context) []wbItem { + rows, err := g.DB().Ctx(ctx).Query(ctx, + `SELECT COUNT(*) AS cnt, COUNT(DISTINCT error_message) AS distinct_cnt + FROM sys_error_logs + WHERE resolved = false + AND created_at >= NOW() - INTERVAL '7 days'`) + if err != nil || len(rows) == 0 { + if err != nil { + g.Log().Warningf(ctx, "workbench: 查询未处理错误日志失败: %v", err) + } + return nil + } + cnt := rows[0]["cnt"].Int() + if cnt == 0 { + return nil + } + + desc := "近 7 天内产生" + if d := rows[0]["distinct_cnt"].Int(); d > 0 && cnt > d { + desc += fmt.Sprintf("，去重后仅 %d 种，建议按类型合并排查", d) + } + + return []wbItem{wbNew( + "error_log_unresolved", + v1.WorkbenchSeverityP2, v1.WorkbenchDomainSystem, "monitor:view", + fmt.Sprintf("%d 条系统错误日志未处理", cnt), + desc, + "查看错误日志", "AdminErrorLogs", nil, nil, + )} +} + +// wbNewIPLogin 管理员从新 IP 登录成功。 +// 管理后台账号是整个平台的最高权限，异地登录必须让人看见并确认。 +func (s *sAdmin) wbNewIPLogin(ctx context.Context) []wbItem { + rows, err := g.DB().Ctx(ctx).Query(ctx, + `SELECT h.id, h.user_id, h.ip_address, h.location, h.created_at, u.username + FROM aud_login_history h + LEFT JOIN sys_admin_users u ON u.id = h.user_id + WHERE h.user_type = 'admin' + AND h.success = true + AND h.is_new_device = true + AND h.created_at >= NOW() - INTERVAL '24 hours' + ORDER BY h.created_at DESC + LIMIT 5`) + if err != nil { + g.Log().Warningf(ctx, "workbench: 查询异常登录失败: %v", err) + return nil + } + + items := make([]wbItem, 0, len(rows)) + for _, r := range rows { + createdAt := r["created_at"].GTime() + loc := strings.TrimSpace(r["location"].String()) + if loc == "" { + loc = "未知归属地" + } + items = append(items, wbNew( + fmt.Sprintf("admin_new_ip_login:%d", r["id"].Int64()), + v1.WorkbenchSeverityP1, v1.WorkbenchDomainSystem, "audit:view", + fmt.Sprintf("管理员 %s 从新设备登录", r["username"].String()), + fmt.Sprintf("IP %s（%s），此前无该设备记录，请确认是否本人操作", + maskIP(r["ip_address"].String()), loc), + "查看登录历史", "AdminLoginHistory", nil, createdAt, + )) + } + return items +} + +// ============================================================ +// 首屏关键数字 +// ============================================================ + +// wbCollectMetrics 首屏三个数字 + 待办计数由上层补。 +// +// 收入对比刻意用「昨日同期」而非「昨日全天」：全天对比在每天早晨永远是 -80%， +// 是个看了就要在脑子里做一次除法的假信号。 +func (s *sAdmin) wbCollectMetrics(ctx context.Context) []v1.WorkbenchMetric { + metrics := make([]v1.WorkbenchMetric, 0, 3) + + // --- 今日收入 vs 昨日同期 --- + var todayCost, yesterdayCost float64 + rows, err := g.DB().Ctx(ctx).Query(ctx, + `SELECT + COALESCE(SUM(total_cost) FILTER (WHERE created_at >= date_trunc('day', NOW())), 0) AS today_cost, + COALESCE(SUM(total_cost) FILTER ( + WHERE created_at >= date_trunc('day', NOW()) - INTERVAL '1 day' + AND created_at < date_trunc('day', NOW()) - INTERVAL '1 day' + (NOW() - date_trunc('day', NOW())) + ), 0) AS yesterday_cost + FROM bil_usage_logs + WHERE created_at >= date_trunc('day', NOW()) - INTERVAL '1 day'`) + if err != nil { + g.Log().Warningf(ctx, "workbench: 查询收入指标失败: %v", err) + } else if len(rows) > 0 { + todayCost = rows[0]["today_cost"].Float64() + yesterdayCost = rows[0]["yesterday_cost"].Float64() + } + + revenueMetric := v1.WorkbenchMetric{ + Key: "today_revenue", Label: "今日收入", Value: todayCost, + Unit: "money", Sub: "较昨日同期", + } + if yesterdayCost > 0 { + growth := (todayCost - yesterdayCost) / yesterdayCost * 100 + revenueMetric.Growth = &growth + } + metrics = append(metrics, revenueMetric) + + // --- 实时错误率（近 5 分钟）--- + // 刻意不用「今日成功率」：那个数被一整天的历史稀释，正在发生的故障根本顶不动它。 + var errRate float64 + rows, err = g.DB().Ctx(ctx).Query(ctx, + `SELECT COUNT(*) AS total, + COUNT(*) FILTER (WHERE status <> 'success') AS errors + FROM bil_usage_logs + WHERE created_at >= NOW() - INTERVAL '5 minutes'`) + if err != nil { + g.Log().Warningf(ctx, "workbench: 查询实时错误率失败: %v", err) + } else if len(rows) > 0 { + if total := rows[0]["total"].Float64(); total > 0 { + errRate = rows[0]["errors"].Float64() / total * 100 + } + } + metrics = append(metrics, v1.WorkbenchMetric{ + Key: "error_rate_5m", Label: "实时错误率", Value: round2(errRate), + Unit: "percent", Sub: "近 5 分钟", + }) + + // --- 余额告警租户数 --- + var lowBalance float64 + rows, err = g.DB().Ctx(ctx).Query(ctx, + `SELECT COUNT(*) AS cnt + FROM bil_wallets w + JOIN tnt_tenants t ON t.id = w.tenant_id + WHERE t.status IN ('active', 'trial') + AND w.warning_threshold > 0 + AND (w.balance - w.frozen_balance) < w.warning_threshold`) + if err != nil { + g.Log().Warningf(ctx, "workbench: 查询余额告警租户失败: %v", err) + } else if len(rows) > 0 { + lowBalance = rows[0]["cnt"].Float64() + } + metrics = append(metrics, v1.WorkbenchMetric{ + Key: "low_balance_tenants", Label: "余额告警租户", Value: lowBalance, + Unit: "count", Sub: "低于预警阈值", + }) + + return metrics +} + +// ============================================================ +// 小工具 +// ============================================================ + +func truncate(s string, n int) string { + r := []rune(s) + if len(r) <= n { + return s + } + return string(r[:n]) + "…" +} + +func round2(f float64) float64 { + return float64(int64(f*100+0.5)) / 100 +} + +func fmtTime(t *gtime.Time) string { + if t == nil { + return "未知时间" + } + return t.Format("Y-m-d H:i") +} + +// humanizeSince 把时间差写成人话。工作台上「3 小时 12 分」比时间戳更能驱动行动。 +func humanizeSince(t *gtime.Time) string { + if t == nil { + return "未知时长" + } + d := time.Since(t.Time) + if d < time.Minute { + return "不到 1 分钟" + } + if d < time.Hour { + return fmt.Sprintf("%d 分钟", int(d.Minutes())) + } + if d < 24*time.Hour { + h := int(d.Hours()) + m := int(d.Minutes()) - h*60 + if m == 0 { + return fmt.Sprintf("%d 小时", h) + } + return fmt.Sprintf("%d 小时 %d 分", h, m) + } + return fmt.Sprintf("%d 天", int(d.Hours()/24)) +} + +// maskIP 打码 IP 中间两段，兼顾可辨识与最小暴露。 +func maskIP(ip string) string { + parts := strings.Split(ip, ".") + if len(parts) != 4 { + return ip + } + return parts[0] + ".**.**." + parts[3] +} + +func sortModelAvail(list []v1.WorkbenchModelAvail) { + sortSlice(len(list), func(i, j int) bool { + // 零可用的排最前，其次按缺口比例，最后按名字保证确定性 + ai := list[i].Available == 0 && list[i].Total > 0 + aj := list[j].Available == 0 && list[j].Total > 0 + if ai != aj { + return ai + } + gi := list[i].Total - list[i].Available + gj := list[j].Total - list[j].Available + if gi != gj { + return gi > gj + } + return list[i].Model < list[j].Model + }, func(i, j int) { list[i], list[j] = list[j], list[i] }) +} + +func sortBreakers(list []v1.WorkbenchBreaker) { + sortSlice(len(list), func(i, j int) bool { + if list[i].ChannelLevel != list[j].ChannelLevel { + return list[i].ChannelLevel + } + if list[i].ChannelName != list[j].ChannelName { + return list[i].ChannelName < list[j].ChannelName + } + return list[i].Model < list[j].Model + }, func(i, j int) { list[i], list[j] = list[j], list[i] }) +} + +// sortSlice 简易插入排序：工作台的列表规模都在几十条以内， +// 不值得为此引入反射版 sort.Slice 的开销。 +func sortSlice(n int, less func(i, j int) bool, swap func(i, j int)) { + for i := 1; i < n; i++ { + for j := i; j > 0 && less(j, j-1); j-- { + swap(j, j-1) + } + } +} diff --git a/internal/middleware/rbac.go b/internal/middleware/rbac.go index 32222bf..357494b 100644 --- a/internal/middleware/rbac.go +++ b/internal/middleware/rbac.go @@ -306,6 +306,11 @@ var adminPermissionRules = []permissionRule{ // ── dashboard and async task management ── {method: "GET", prefix: "/api/admin/dashboard", perm: "dashboard:view"}, + // 工作台读接口只要求 dashboard:view：跨域条目由 logic 层按 channel:view / billing:view / + // support:view 等逐条过滤，进得来不等于看得到，权限收敛在数据侧而非路由侧。 + // 工作台没有写接口 —— 待办由源表实时派生，唯一的处置方式是去源头解决。 + {method: "GET", path: "/api/admin/workbench/summary", perm: "dashboard:view"}, + {method: "GET", path: "/api/admin/workbench/badges", perm: "dashboard:view"}, {method: "GET", path: "/api/admin/tasks", perm: "task:view"}, {method: "GET", prefix: "/api/admin/tasks/", perm: "task:view"}, {method: "POST", prefix: "/api/admin/tasks/", suffix: "/cancel", perm: "task:edit"}, diff --git a/internal/service/admin.go b/internal/service/admin.go index a3f16c4..2bd912f 100644 --- a/internal/service/admin.go +++ b/internal/service/admin.go @@ -547,6 +547,10 @@ type ( UsageLogCleanupCreate(ctx context.Context, req *v1.UsageLogCleanupCreateReq) (*v1.UsageLogCleanupCreateRes, error) UsageLogCleanupList(ctx context.Context, req *v1.UsageLogCleanupListReq) (*v1.UsageLogCleanupListRes, error) UsageLogCleanupCancel(ctx context.Context, req *v1.UsageLogCleanupCancelReq) (*v1.UsageLogCleanupCancelRes, error) + // GetWorkbenchSummary 工作台汇总。 + GetWorkbenchSummary(ctx context.Context, _ *v1.AdminWorkbenchSummaryReq) (*v1.AdminWorkbenchSummaryRes, error) + // GetWorkbenchBadges 菜单红点计数（与 summary 共用缓存，不额外压库）。 + GetWorkbenchBadges(ctx context.Context, _ *v1.AdminWorkbenchBadgeReq) (*v1.AdminWorkbenchBadgeRes, error) } ) diff --git a/web/admin/src/router/admin.ts b/web/admin/src/router/admin.ts index 9bc3974..76e9dc8 100644 --- a/web/admin/src/router/admin.ts +++ b/web/admin/src/router/admin.ts @@ -24,6 +24,12 @@ const adminRoutes: RouteRecordRaw[] = [ component: () => import('@/views/DashboardPage.vue'), meta: { title: '仪表盘' }, }, + { + path: 'workbench', + name: 'AdminWorkbench', + component: () => import('@/views/WorkbenchPage.vue'), + meta: { title: '工作台' }, + }, { path: 'profile', name: 'AdminProfile', diff --git a/web/admin/src/views/AdminLayout.vue b/web/admin/src/views/AdminLayout.vue index 5ab32d1..8680230 100644 --- a/web/admin/src/views/AdminLayout.vue +++ b/web/admin/src/views/AdminLayout.vue @@ -9,6 +9,7 @@ import RouteErrorBoundary from '@/components/RouteErrorBoundary.vue' import { useWatermark } from '@/composables/useWatermark' import { useSiteName } from '@/composables/useSiteName' import { + IconThunderbolt, IconDashboard, IconUserGroup, IconApps, @@ -305,6 +306,7 @@ const menuGroups = [ label: '数据看板', icon: IconDashboard, items: [ + { name: 'AdminWorkbench', label: '工作台', icon: IconThunderbolt }, { name: 'AdminDashboard', label: '仪表盘', icon: IconDashboard }, { name: 'AdminRealtimeMonitor', label: '实时监控', icon: IconCommand }, ], @@ -453,6 +455,34 @@ function handleClickOutside(e: MouseEvent) { } } +// --- 工作台菜单红点 --- +// 工作台是聚合视图，红点才是常驻发现入口：管理员不会每天主动点开工作台， +// 但会看见菜单上的角标。后端 badges 与 summary 共用同一份 30s Redis 缓存， +// 60s 轮询不会额外压库。 +const workbenchBadges = ref>({}) +const workbenchUrgent = ref(0) +const workbenchTotal = ref(0) +let badgeTimer: ReturnType | null = null + +// 菜单项角标：工作台本身显示待办总数，其余菜单显示后端映射到该路由的待办数 +function badgeOf(name: string): number { + if (name === 'AdminWorkbench') return workbenchTotal.value + return workbenchBadges.value[name] || 0 +} + +async function fetchWorkbenchBadges() { + try { + const res = await request.get('/admin/workbench/badges') + const data = res.data?.data + if (!data) return + workbenchBadges.value = data.menus || {} + workbenchUrgent.value = data.urgent || 0 + workbenchTotal.value = data.total || 0 + } catch { + // 静默：红点拉取失败不该打扰用户，也不该阻塞布局 + } +} + onMounted(() => { document.addEventListener('click', handleClickOutside) updateMobile() @@ -462,6 +492,9 @@ onMounted(() => { checkUpdate() fetchUpdateStatus() // 取后端实时版本，用于准确的"是否可更新"判断 + fetchWorkbenchBadges() + badgeTimer = setInterval(fetchWorkbenchBadges, 60000) + axios.get('/api/settings/public').then((res) => { const settings = res.data?.data?.settings if (settings) { @@ -478,6 +511,7 @@ onUnmounted(() => { document.removeEventListener('click', handleClickOutside) window.removeEventListener('resize', updateMobile) clearTimeout(popupHideTimer!) + if (badgeTimer) clearInterval(badgeTimer) stopPolling() unmountWatermark() }) @@ -524,6 +558,11 @@ onUnmounted(() => { > {{ item.label }} + {{ badgeOf(item.name) > 99 ? '99+' : badgeOf(item.name) }} @@ -590,6 +629,11 @@ onUnmounted(() => { > {{ item.label }} + {{ badgeOf(item.name) > 99 ? '99+' : badgeOf(item.name) }} @@ -950,6 +994,28 @@ onUnmounted(() => { text-overflow: ellipsis; } +/* 待办角标：默认中性灰，仅在工作台有 P0 时转红。 + 全部标红会让红点迅速失去意义 —— 红色必须留给「现在就得看」。 */ +.admin-sidebar__badge { + margin-left: auto; + flex-shrink: 0; + min-width: 18px; + height: 18px; + padding: 0 5px; + line-height: 18px; + text-align: center; + border-radius: 9px; + font-size: 11px; + font-weight: 600; + background: var(--color-fill-3); + color: var(--color-text-2); +} + +.admin-sidebar__badge--urgent { + background: #f53f3f; + color: #fff; +} + /* ===== Collapsed Sidebar Overrides ===== */ .admin-sidebar--collapsed .admin-sidebar__group-header { justify-content: center; diff --git a/web/admin/src/views/WorkbenchPage.vue b/web/admin/src/views/WorkbenchPage.vue new file mode 100644 index 0000000..40c0b61 --- /dev/null +++ b/web/admin/src/views/WorkbenchPage.vue @@ -0,0 +1,818 @@ + + + + +