package console import ( "fmt" "strings" "sync/atomic" "time" "yunyan/comm" "yunyan/lego/sys/postgres" "yunyan/pb" "yunyan/utils" "gorm.io/gorm/clause" ) // 统计落库与 dashboard 聚合查询的数据模型。console 为 stats_global_day 的唯一写库方: // 消费各业务部署 analyze 推送的【绝对快照】(comm.StatSnapshot),直接以绝对值覆盖 upsert // (见 stat.go.onMessage)。console 自身不再维护 Redis 累加/HLL——去重人数由业务侧算好随快照带来。 // statLoc 统计日界时区,由 statComp Init 经 setStatLoc 写入(默认 Asia/Shanghai)。 // 与各 analyze 端共用同一口径,保证 0 点切日一致;未初始化时回退固定 UTC+8。 var statLoc atomic.Value // *time.Location func setStatLoc(loc *time.Location) { statLoc.Store(loc) } func statLocation() *time.Location { if v := statLoc.Load(); v != nil { return v.(*time.Location) } return time.FixedZone("UTC+8", 8*3600) } // statDayNumber 把时间转成 YYYYMMDD 形式的统计日(按固定统计时区,与 analyze 口径一致)。 func statDayNumber(t time.Time) uint32 { return utils.StatDayNumber(t, statLocation()) } // ensureStatTable 在 console 公共库(AdminDB/supabase)建好 stats_global_day。 func ensureStatTable() error { if err := postgres.CreateTable(comm.TableStatsGlobalDay, &pb.StatsGlobalDay{}); err != nil { return err } // postgres CreateTable 对已存在表跳过 AutoMigrate,补可能漂移缺失的列(幂等)。 // qr_active_count 为后加字段,老表常缺,缺失会让所有 SUM 查询(statSumSelect)整体报错。 if res := postgres.Exec("ALTER TABLE " + comm.TableStatsGlobalDay + " ADD COLUMN IF NOT EXISTS qr_active_count bigint DEFAULT 0"); res.Error != nil { return res.Error } return nil } // verifyStatTablePK 只读自检 stats_global_day 的复合主键是否含四个维度列 // (app_id, product_id, region, stat_day)。OnConflict 的「唯一写入」依赖此主键: // 同一复合键的重复写会走 UPDATE 而非 INSERT,故主键缺一不可。 // 仅返回结果供调用方告警,不自动改主键(避免在已有重复行时 ALTER 失败)。 func verifyStatTablePK() (ok bool, cols string) { postgres.Table(comm.TableStatsGlobalDay).Raw( `SELECT COALESCE(string_agg(a.attname, ','), '') FROM pg_index i JOIN pg_attribute a ON a.attrelid = i.indrelid AND a.attnum = ANY(i.indkey) WHERE i.indrelid = ?::regclass AND i.indisprimary`, comm.TableStatsGlobalDay).Scan(&cols) ok = strings.Contains(cols, "app_id") && strings.Contains(cols, "product_id") && strings.Contains(cols, "region") && strings.Contains(cols, "stat_day") return } // upsertStatDay 以「绝对值」覆盖写入一行(OnConflict UpdateAll)。Redis 侧持有的就是当日累计 // 绝对值,故覆盖式写入幂等、可重复执行。 func upsertStatDay(row *pb.StatsGlobalDay) error { return postgres.Table(comm.TableStatsGlobalDay). Clauses(clause.OnConflict{UpdateAll: true}). Create(row).Error } // upsertStatDayBatch 批量覆盖写入多行(一条多行 INSERT ... ON CONFLICT DO UPDATE)。 // 历史分析一次产出成百上千行,逐行 upsert 会被远程库(Supabase) RTT 放大成 N×往返; // 批量后压成 ⌈N/batchSize⌉ 次往返。语义与 upsertStatDay 一致(绝对值覆盖、幂等)。 func upsertStatDayBatch(rows []*pb.StatsGlobalDay) error { if len(rows) == 0 { return nil } return postgres.Table(comm.TableStatsGlobalDay). Clauses(clause.OnConflict{UpdateAll: true}). CreateInBatches(rows, 200).Error } // ============================ dashboard 聚合查询 ============================ // // 直接查 console 公共库的 stats_global_day(console 消费各部署快照后按 app×product×region×day 落行), // 按 stat_day 区间 SUM 聚合,支持按 app/region/product 下钻。 // // 注意:*_user_count 三列是「当日去重」值,跨日/跨区域 SUM 得到的是「人日」之和而非 // 唯一人数(HLL 不可跨行精确合并);次数/金额/时长类 SUM 则为精确累计。 // statsQueryReq 聚合查询请求。指针字段为 nil 表示该维度不过滤(全量); // 用指针而非零值,是因为 product_id=0、空区域都是合法取值,不能拿零值当“全部”。 // app_id 为应用名称、region 为区域代码(与 stats_global_day 存储一致)。 type statsQueryReq struct { StartDay uint32 `json:"start_day"` // 起始统计日 YYYYMMDD(含),0 不限 EndDay uint32 `json:"end_day"` // 结束统计日 YYYYMMDD(含),0 不限 AppId *string `json:"app_id"` // 应用名称过滤(单值),nil 全部 Region *string `json:"region"` // 区域代码过滤(单值),nil 全部 ProductId *uint32 `json:"product_id"` // 产品过滤(单值),nil 全部 // 多值 IN 过滤:代理"全部(我的)"用,传绑定列表,落 app_id IN (...) 等。与单值并存(同维度二选一即可)。 Apps []string `json:"apps"` // 应用名称列表 Regions []string `json:"regions"` // 区域代码列表 ProductIds []uint32 `json:"product_ids"` // 产品 id 列表 } // StatMetrics stats_global_day 的全部可累加指标(聚合结果,列名与表一致)。 type StatMetrics struct { LoginUserCount int64 `json:"login_user_count"` LoginCount int64 `json:"login_count"` NewUserCount int64 `json:"new_user_count"` ActiveUserCount int64 `json:"active_user_count"` ActiveDeviceCount int64 `json:"active_device_count"` BindDeviceCount int64 `json:"bind_device_count"` QrActiveCount int64 `json:"qr_active_count"` OrderCreateCount int64 `json:"order_create_count"` OrderPaidCount int64 `json:"order_paid_count"` OrderFailedCount int64 `json:"order_failed_count"` OrderCreateAmount int64 `json:"order_create_amount"` OrderPaidAmount int64 `json:"order_paid_amount"` PayUserCount int64 `json:"pay_user_count"` GrantVipDay int64 `json:"grant_vip_day"` GrantAiIntegral int64 `json:"grant_ai_integral"` GrantTradeSecond int64 `json:"grant_trade_second"` GrantMeetSecond int64 `json:"grant_meet_second"` AiChatCount int64 `json:"ai_chat_count"` AiUpToken int64 `json:"ai_up_token"` AiDownToken int64 `json:"ai_down_token"` TradeCount int64 `json:"trade_count"` TradeSecond int64 `json:"trade_second"` TradeWords int64 `json:"trade_words"` MeetCount int64 `json:"meet_count"` MeetSecond int64 `json:"meet_second"` } // StatTrendRow 按日聚合的一行(趋势图用)。 type StatTrendRow struct { StatDay uint32 `json:"stat_day"` StatMetrics } // statMetricCols 与 StatMetrics 字段一一对应的列名(SUM 用)。 var statMetricCols = []string{ "login_user_count", "login_count", "new_user_count", "active_user_count", "active_device_count", "bind_device_count", "qr_active_count", "order_create_count", "order_paid_count", "order_failed_count", "order_create_amount", "order_paid_amount", "pay_user_count", "grant_vip_day", "grant_ai_integral", "grant_trade_second", "grant_meet_second", "ai_chat_count", "ai_up_token", "ai_down_token", "trade_count", "trade_second", "trade_words", "meet_count", "meet_second", } // statSumSelect 拼出 "extra..., COALESCE(SUM(col),0) AS col, ..." 的 select 子句。 // COALESCE 兜底空结果集的 SUM(NULL),避免扫描 NULL 进 int64 报错。 func statSumSelect(extra ...string) string { parts := make([]string, 0, len(extra)+len(statMetricCols)) parts = append(parts, extra...) for _, c := range statMetricCols { parts = append(parts, fmt.Sprintf("COALESCE(SUM(%s),0) AS %s", c, c)) } return strings.Join(parts, ", ") } // statWhere 按请求拼过滤条件。 func statWhere(req *statsQueryReq) (string, []interface{}) { conds := make([]string, 0, 5) args := make([]interface{}, 0, 5) if req.StartDay > 0 { conds = append(conds, "stat_day >= ?") args = append(args, req.StartDay) } if req.EndDay > 0 { conds = append(conds, "stat_day <= ?") args = append(args, req.EndDay) } if req.AppId != nil { conds = append(conds, "app_id = ?") args = append(args, *req.AppId) } if req.Region != nil { conds = append(conds, "region = ?") args = append(args, *req.Region) } if req.ProductId != nil { conds = append(conds, "product_id = ?") args = append(args, *req.ProductId) } // 多值 IN 过滤(代理"全部(我的)"):gorm 对 "col IN ?" + 切片会展开为 IN (v1,v2,...)。 if len(req.Apps) > 0 { conds = append(conds, "app_id IN ?") args = append(args, req.Apps) } if len(req.Regions) > 0 { conds = append(conds, "region IN ?") args = append(args, req.Regions) } if len(req.ProductIds) > 0 { conds = append(conds, "product_id IN ?") args = append(args, req.ProductIds) } return strings.Join(conds, " AND "), args } // andWhere 把两个 WHERE 片段用 AND 合并(任一为空则取另一个),并按顺序拼接参数。 func andWhere(w1 string, a1 []interface{}, w2 string, a2 []interface{}) (string, []interface{}) { switch { case w1 == "" && w2 == "": return "", nil case w1 == "": return w2, a2 case w2 == "": return w1, a1 default: return "(" + w1 + ") AND (" + w2 + ")", append(append([]interface{}{}, a1...), a2...) } } // queryStatSummary 区间总量(全维度 SUM 成一行)。scopeWhere/scopeArgs 为账号作用域附加条件(与下钻 AND)。 func queryStatSummary(req *statsQueryReq, scopeWhere string, scopeArgs []interface{}) (*StatMetrics, error) { out := &StatMetrics{} where, args := statWhere(req) where, args = andWhere(where, args, scopeWhere, scopeArgs) tx := postgres.Table(comm.TableStatsGlobalDay).Select(statSumSelect()) if where != "" { tx = tx.Where(where, args...) } err := tx.Scan(out).Error return out, err } // queryStatTrend 按 stat_day 分组的逐日序列(升序)。 func queryStatTrend(req *statsQueryReq, scopeWhere string, scopeArgs []interface{}) ([]*StatTrendRow, error) { rows := make([]*StatTrendRow, 0) where, args := statWhere(req) where, args = andWhere(where, args, scopeWhere, scopeArgs) tx := postgres.Table(comm.TableStatsGlobalDay). Select(statSumSelect("stat_day")). Group("stat_day").Order("stat_day ASC") if where != "" { tx = tx.Where(where, args...) } err := tx.Scan(&rows).Error return rows, err }