From 62a23325c7ae9735697c1f843c04f6d9d7ced367 Mon Sep 17 00:00:00 2001 From: chengqian Date: Fri, 18 Sep 2026 15:24:09 +0800 Subject: [PATCH] =?UTF-8?q?improve:=E5=AF=BC=E5=87=BA=E6=8E=A5=E5=8F=A3?= =?UTF-8?q?=E6=8A=A4=E6=A0=8F=E4=B8=8E=E4=B8=A4=E7=AB=AF=E6=8E=A7=E5=88=B6?= =?UTF-8?q?=E5=8F=B0=E6=9F=A5=E8=AF=A2=E6=80=A7=E8=83=BD=E4=BC=98=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 导出统一行数上限：xlsx 超限在拉取阶段截停并报错，CSV 到限截断并在文件末尾写入可见标记 - 全部导出接口 OFFSET 循环翻页改为 keyset 游标（用量日志/计费记录/成员/渠道/钱包流水） - 用量日志导出强制时间窗（开始时间必填、跨度≤31天），校验下沉 common 供管理后台与租户端共用 - 管理后台仪表盘趋势/租户排行/模型分布/小时费用聚合加 5 分钟双层缓存 - 交易流水 Username 筛选改 EXISTS 子查询解除对 JOIN 别名的依赖，COUNT 与列表拆分免背联表 - 管理后台成员列表 COUNT 免联表（筛选条件只引用主表列） - 租户端用量日志列表/统计/导出共用筛选构造器，COUNT 与 SUM 全程剥离联表，钱包流水 COUNT 拆分 --- internal/logic/admin/channel.go | 19 +- internal/logic/admin/dashboard.go | 232 +++++++++++++-------- internal/logic/admin/member.go | 22 +- internal/logic/billing/transaction.go | 6 +- internal/logic/common/validation.go | 33 +++ internal/logic/tenant/billing.go | 278 +++++++++++++------------- internal/logic/tenant/member.go | 10 +- internal/utility/export/export.go | 47 ++++- 8 files changed, 405 insertions(+), 242 deletions(-) diff --git a/internal/logic/admin/channel.go b/internal/logic/admin/channel.go index a98de1c..7366efd 100644 --- a/internal/logic/admin/channel.go +++ b/internal/logic/admin/channel.go @@ -1064,7 +1064,14 @@ func (s *sAdmin) ExportChannels(ctx context.Context, req *v1.ChannelExportReq) ( } return nil, export.GenericExport(ctx, config, func(yield func(map[string]any) bool) { - offset := 0 + // keyset 翻页：导出按优先级倒序展示，游标取 (priority, id) 复合键 —— + // priority 大量重复，必须带 id 作 tiebreaker，否则同优先级的行会被跳过或重复。 + // PostgreSQL 行值比较 (a, b) < (?, ?) 按字典序逐列比较，与排序方向一致。 + var ( + hasCursor bool + cursorPriority int + cursorID int64 + ) for { query := dao.ChnChannels.Ctx(ctx). LeftJoin("chn_health_scores h ON chn_channels.id = h.channel_id") @@ -1087,6 +1094,9 @@ func (s *sAdmin) ExportChannels(ctx context.Context, req *v1.ChannelExportReq) ( // 与 ListChannels 保持一致：EXISTS 相关子查询避免 JOIN 重复行 query = query.Where("EXISTS (SELECT 1 FROM chn_abilities WHERE chn_abilities.channel_id = chn_channels.id AND (chn_abilities.model_name LIKE ? OR chn_abilities.upstream_model LIKE ?))", "%"+req.Model+"%", "%"+req.Model+"%") } + if hasCursor { + query = query.Where("(chn_channels.priority, chn_channels.id) < (?, ?)", cursorPriority, cursorID) + } var batch []struct { ID int64 `json:"id"` Name string `json:"name"` @@ -1097,8 +1107,8 @@ func (s *sAdmin) ExportChannels(ctx context.Context, req *v1.ChannelExportReq) ( CreatedAt *gtime.Time `json:"created_at"` HealthScore *float64 `json:"health_score"` } - if err := query.Fields(channelFields).OrderDesc("chn_channels.priority").Limit(1000).Offset(offset).Scan(&batch); err != nil { - g.Log().Errorf(ctx, "ExportChannels: query batch at offset %d failed: %v", offset, err) + if err := query.Fields(channelFields).OrderDesc("chn_channels.priority").OrderDesc("chn_channels.id").Limit(1000).Scan(&batch); err != nil { + g.Log().Errorf(ctx, "ExportChannels: query batch failed (cursor priority=%d id=%d): %v", cursorPriority, cursorID, err) return } for _, ch := range batch { @@ -1122,11 +1132,12 @@ func (s *sAdmin) ExportChannels(ctx context.Context, req *v1.ChannelExportReq) ( }) { return } + cursorPriority, cursorID = ch.Priority, ch.ID } if len(batch) < 1000 { break } - offset += 1000 + hasCursor = true } }) } diff --git a/internal/logic/admin/dashboard.go b/internal/logic/admin/dashboard.go index d1ea6bf..c7b66ad 100644 --- a/internal/logic/admin/dashboard.go +++ b/internal/logic/admin/dashboard.go @@ -20,6 +20,27 @@ import ( "github.com/qianfree/team-api/internal/utility/export" ) +// dashboardCache 仪表盘聚合接口的短缓存（L1 内存 + L2 Redis）。 +// 趋势 / 租户排行 / 模型分布 / 小时费用这几个接口都是对 bil_usage_logs +// 的多天窗口聚合扫描，且数据本身不要求秒级新鲜 —— 5 分钟缓存能消除 +// 仪表盘反复打开 / 多管理员同时在线造成的绝大部分重复扫表。 +var dashboardCache = common.NewCache("admin:dashboard", 5*time.Minute) + +// cachedDashboard 按 key 读仪表盘缓存，miss 时回源并写缓存。 +// key 必须编入全部查询参数（如天数 / TopN），避免不同参数共享一份结果。 +func cachedDashboard[T any](ctx context.Context, key string, load func() (*T, error)) (*T, error) { + var cached T + if dashboardCache.GetJSON(ctx, key, &cached) { + return &cached, nil + } + res, err := load() + if err != nil { + return nil, err + } + dashboardCache.Set(ctx, key, res) + return res, nil +} + // GetDashboardStats 获取管理后台仪表盘统计 func (s *sAdmin) GetDashboardStats(ctx context.Context, req *v1.AdminDashboardReq) (*v1.AdminDashboardRes, error) { today := time.Now().Format("2006-01-02") @@ -133,25 +154,27 @@ func (s *sAdmin) GetDashboardTrends(ctx context.Context, req *v1.AdminDashboardT days = 30 } - startDate := time.Now().AddDate(0, 0, -days).Format("2006-01-02") + return cachedDashboard(ctx, fmt.Sprintf("trends:%d", days), func() (*v1.AdminDashboardTrendsRes, error) { + startDate := time.Now().AddDate(0, 0, -days).Format("2006-01-02") - result, err := g.DB().Ctx(ctx).Raw(` - SELECT - DATE(created_at) as date, - COUNT(*) as requests, - COUNT(DISTINCT tenant_id) as active_tenants, - COALESCE(SUM(total_cost), 0) as revenue - FROM bil_usage_logs - WHERE created_at >= ? - GROUP BY DATE(created_at) - ORDER BY date ASC - `, startDate+" 00:00:00").All() - if err != nil { - return nil, err - } + result, err := g.DB().Ctx(ctx).Raw(` + SELECT + DATE(created_at) as date, + COUNT(*) as requests, + COUNT(DISTINCT tenant_id) as active_tenants, + COALESCE(SUM(total_cost), 0) as revenue + FROM bil_usage_logs + WHERE created_at >= ? + GROUP BY DATE(created_at) + ORDER BY date ASC + `, startDate+" 00:00:00").All() + if err != nil { + return nil, err + } - records := result.List() - return &v1.AdminDashboardTrendsRes{List: records}, nil + records := result.List() + return &v1.AdminDashboardTrendsRes{List: records}, nil + }) } // GetTopTenants returns the top 10 tenants by revenue. @@ -161,28 +184,30 @@ func (s *sAdmin) GetTopTenants(ctx context.Context, req *v1.AdminDashboardTopTen days = 30 } - startDate := time.Now().AddDate(0, 0, -days).Format("2006-01-02") + return cachedDashboard(ctx, fmt.Sprintf("top-tenants:%d", days), func() (*v1.AdminDashboardTopTenantsRes, error) { + startDate := time.Now().AddDate(0, 0, -days).Format("2006-01-02") - result, err := g.DB().Ctx(ctx).Raw(` - SELECT - t.id as tenant_id, - t.name as tenant_name, - COALESCE(SUM(ul.total_cost), 0) as total_cost, - COUNT(*) as requests, - COUNT(DISTINCT ul.user_id) as active_members - FROM bil_usage_logs ul - JOIN tnt_tenants t ON t.id = ul.tenant_id - WHERE ul.created_at >= ? - GROUP BY t.id, t.name - ORDER BY total_cost DESC - LIMIT 10 - `, startDate+" 00:00:00").All() - if err != nil { - return nil, err - } + result, err := g.DB().Ctx(ctx).Raw(` + SELECT + t.id as tenant_id, + t.name as tenant_name, + COALESCE(SUM(ul.total_cost), 0) as total_cost, + COUNT(*) as requests, + COUNT(DISTINCT ul.user_id) as active_members + FROM bil_usage_logs ul + JOIN tnt_tenants t ON t.id = ul.tenant_id + WHERE ul.created_at >= ? + GROUP BY t.id, t.name + ORDER BY total_cost DESC + LIMIT 10 + `, startDate+" 00:00:00").All() + if err != nil { + return nil, err + } - records := result.List() - return &v1.AdminDashboardTopTenantsRes{List: records}, nil + records := result.List() + return &v1.AdminDashboardTopTenantsRes{List: records}, nil + }) } // GetModelDistribution returns the model usage distribution. @@ -192,27 +217,29 @@ func (s *sAdmin) GetModelDistribution(ctx context.Context, req *v1.AdminDashboar days = 30 } - startDate := time.Now().AddDate(0, 0, -days).Format("2006-01-02") + return cachedDashboard(ctx, fmt.Sprintf("model-dist:%d", days), func() (*v1.AdminDashboardModelDistributionRes, error) { + startDate := time.Now().AddDate(0, 0, -days).Format("2006-01-02") - result, err := g.DB().Ctx(ctx).Raw(` - SELECT - model_name, - COUNT(*) as requests, - COALESCE(SUM(input_tokens), 0) as input_tokens, - COALESCE(SUM(output_tokens), 0) as output_tokens, - COALESCE(SUM(total_cost), 0) as total_cost - FROM bil_usage_logs - WHERE created_at >= ? - GROUP BY model_name - ORDER BY total_cost DESC - LIMIT 20 - `, startDate+" 00:00:00").All() - if err != nil { - return nil, err - } + result, err := g.DB().Ctx(ctx).Raw(` + SELECT + model_name, + COUNT(*) as requests, + COALESCE(SUM(input_tokens), 0) as input_tokens, + COALESCE(SUM(output_tokens), 0) as output_tokens, + COALESCE(SUM(total_cost), 0) as total_cost + FROM bil_usage_logs + WHERE created_at >= ? + GROUP BY model_name + ORDER BY total_cost DESC + LIMIT 20 + `, startDate+" 00:00:00").All() + if err != nil { + return nil, err + } - records := result.List() - return &v1.AdminDashboardModelDistributionRes{List: records}, nil + records := result.List() + return &v1.AdminDashboardModelDistributionRes{List: records}, nil + }) } // GetModelHourlyCost 模型费用按小时堆叠统计，用于仪表盘堆叠柱状图： @@ -227,6 +254,13 @@ func (s *sAdmin) GetModelHourlyCost(ctx context.Context, req *v1.AdminDashboardM topN = 8 } + return cachedDashboard(ctx, fmt.Sprintf("model-hourly:%d:%d", hours, topN), func() (*v1.AdminDashboardModelHourlyRes, error) { + return s.queryModelHourlyCost(ctx, hours, topN) + }) +} + +// queryModelHourlyCost 实际执行小时费用聚合（GetModelHourlyCost 的缓存回源路径）。 +func (s *sAdmin) queryModelHourlyCost(ctx context.Context, hours, topN int) (*v1.AdminDashboardModelHourlyRes, error) { // 步骤1：取费用最高的 TopN 个模型名 topResult, err := g.DB().Ctx(ctx).Raw(` SELECT model_name @@ -1179,6 +1213,15 @@ func (s *sAdmin) GetAllTransactions(ctx context.Context, req *v1.AdminTransactio EndDate: req.EndDate, }) + // COUNT 与列表拆开执行：计数不需要租户/用户名称列，筛选条件也不依赖 JOIN + //（Username 过滤在 BuildTransactionQuery 内是 EXISTS 子查询），大表 + // bil_transactions 的计数因此免背 2 个联表。不用 ScanAndCount —— 它的 + // COUNT 会携带模型上的全部 JOIN（gf v2.10.2 源码确认：Clone 保留 join）。 + total, err := query.Count() + if err != nil { + return nil, err + } + type transactionRow struct { Id int64 `json:"id"` TenantId int64 `json:"tenant_id"` @@ -1194,19 +1237,16 @@ func (s *sAdmin) GetAllTransactions(ctx context.Context, req *v1.AdminTransactio CreatedAt *gtime.Time `json:"created_at"` } - var records []*transactionRow - var total int - err := query.Fields("bil_transactions.id, bil_transactions.tenant_id, COALESCE(tn.name, '') AS tenant_name, bil_transactions.type, bil_transactions.amount, bil_transactions.balance_after, bil_transactions.description, bil_transactions.user_id, COALESCE(tu.username, '') AS username, bil_transactions.request_id, bil_transactions.model_name, bil_transactions.created_at"). - LeftJoin("tnt_users tu", "bil_transactions.user_id = tu.id AND bil_transactions.tenant_id = tu.tenant_id"). - LeftJoin("tnt_tenants tn", "bil_transactions.tenant_id = tn.id"). - OrderDesc("bil_transactions.created_at"). - Page(page, pageSize). - ScanAndCount(&records, &total, false) - if err != nil { - return nil, err - } - if records == nil { - records = make([]*transactionRow, 0) + records := make([]*transactionRow, 0) + if total > 0 { + if err := query.Fields("bil_transactions.id, bil_transactions.tenant_id, COALESCE(tn.name, '') AS tenant_name, bil_transactions.type, bil_transactions.amount, bil_transactions.balance_after, bil_transactions.description, bil_transactions.user_id, COALESCE(tu.username, '') AS username, bil_transactions.request_id, bil_transactions.model_name, bil_transactions.created_at"). + LeftJoin("tnt_users tu", "bil_transactions.user_id = tu.id AND bil_transactions.tenant_id = tu.tenant_id"). + LeftJoin("tnt_tenants tn", "bil_transactions.tenant_id = tn.id"). + OrderDesc("bil_transactions.created_at"). + Page(page, pageSize). + Scan(&records); err != nil { + return nil, err + } } list := make([]*v1.AdminTransactionItem, 0, len(records)) @@ -1279,6 +1319,9 @@ func (s *sAdmin) ExportUsageLogs(ctx context.Context, req *v1.AdminUsageLogExpor if err := validateUsageLogDateRange(req.StartDate, req.EndDate); err != nil { return nil, err } + if err := common.ValidateExportTimeWindow(req.StartDate, req.EndDate, common.UsageLogExportMaxWindowDays); err != nil { + return nil, err + } columns := []export.Column{ {Field: "id", Header: "ID"}, @@ -1305,14 +1348,29 @@ func (s *sAdmin) ExportUsageLogs(ctx context.Context, req *v1.AdminUsageLogExpor selectFields := "u.id, COALESCE(tn.name, '') AS tenant_name, COALESCE(t.username, '') AS username, u.model_name, u.request_type, u.input_tokens, u.output_tokens, u.total_cost, u.status, u.created_at" return nil, export.GenericExport(ctx, config, func(yield func(map[string]any) bool) { - offset := 0 + // keyset（游标）翻页替代 OFFSET：OFFSET 每翻一批都要重新扫过并丢弃前面的 + // 所有行，深翻页代价线性上涨；游标条件让每批从上一批断点直接续读。 + // 游标取 id（bigserial 单调递增，PK 索引可直接定位）：排序从 created_at DESC + // 改为 id DESC，两者在追加写表上近似同序（同为「新的在前」），却避免了 + // 时间戳做游标时亚秒精度在驱动往返中丢失导致的跳行/重行问题。 + var cursorID int64 for { - sql := "SELECT " + selectFields + " FROM " + fromClause + where + " ORDER BY u.created_at DESC LIMIT ? OFFSET ?" - // 复制一份再追加分页参数，避免污染循环外构建的共用筛选参数 - batchArgs := append(append([]any{}, filterArgs...), 1000, offset) + sql := "SELECT " + selectFields + " FROM " + fromClause + where + batchArgs := append([]any{}, filterArgs...) + if cursorID > 0 { + // 时间窗必填保证 where 非空，这里安全起见仍处理空 where 的情况 + if where == "" { + sql += " WHERE u.id < ?" + } else { + sql += " AND u.id < ?" + } + batchArgs = append(batchArgs, cursorID) + } + sql += " ORDER BY u.id DESC LIMIT ?" + batchArgs = append(batchArgs, 1000) result, err := g.DB().Ctx(ctx).Query(ctx, sql, batchArgs...) if err != nil { - g.Log().Errorf(ctx, "ExportUsageLogs: query batch at offset %d failed: %v", offset, err) + g.Log().Errorf(ctx, "ExportUsageLogs: query batch failed (cursor id=%d): %v", cursorID, err) return } for _, row := range result { @@ -1334,11 +1392,11 @@ func (s *sAdmin) ExportUsageLogs(ctx context.Context, req *v1.AdminUsageLogExpor }) { return } + cursorID = row["id"].Int64() } if len(result) < 1000 { break } - offset += 1000 } }) } @@ -1364,11 +1422,17 @@ func (s *sAdmin) ExportBillingRecords(ctx context.Context, req *v1.AdminBillingR Columns: columns, } - fetchRecords := func(offset, limit int) ([]map[string]any, error) { + // fetchRecords 取一批记录并返回本批末行 id 作为下一批游标（0 表示无数据） + fetchRecords := func(cursorID int64, limit int) ([]map[string]any, int64, error) { query := dao.BilRecords.Ctx(ctx) if req.TenantID > 0 { query = query.Where("tenant_id", req.TenantID) } + // keyset 翻页：id 游标替代 OFFSET，避免深翻页时重复扫弃前面的行； + // 排序取 id DESC（追加写表上与 created_at DESC 近似同序） + if cursorID > 0 { + query = query.Where("id < ?", cursorID) + } var records []struct { Id int64 `json:"id"` TenantId int64 `json:"tenant_id"` @@ -1382,8 +1446,8 @@ func (s *sAdmin) ExportBillingRecords(ctx context.Context, req *v1.AdminBillingR CreatedAt *gtime.Time `json:"created_at"` } if err := query.Fields("id, tenant_id, user_id, channel_id, model_name, input_tokens, output_tokens, total_cost, status, created_at"). - OrderDesc("created_at").Limit(limit).Offset(offset).Scan(&records); err != nil { - return nil, err + OrderDesc("id").Limit(limit).Scan(&records); err != nil { + return nil, 0, err } // Batch resolve names @@ -1434,6 +1498,7 @@ func (s *sAdmin) ExportBillingRecords(ctx context.Context, req *v1.AdminBillingR } data := make([]map[string]any, len(records)) + var lastID int64 for i, r := range records { data[i] = map[string]any{ "id": r.Id, @@ -1447,16 +1512,17 @@ func (s *sAdmin) ExportBillingRecords(ctx context.Context, req *v1.AdminBillingR "status": r.Status, "created_at": r.CreatedAt.String(), } + lastID = r.Id } - return data, nil + return data, lastID, nil } return nil, export.GenericExport(ctx, config, func(yield func(map[string]any) bool) { - offset := 0 + var cursorID int64 for { - batch, err := fetchRecords(offset, 1000) + batch, lastID, err := fetchRecords(cursorID, 1000) if err != nil { - g.Log().Errorf(ctx, "ExportBillingRecords: query batch at offset %d failed: %v", offset, err) + g.Log().Errorf(ctx, "ExportBillingRecords: query batch failed (cursor id=%d): %v", cursorID, err) return } for _, row := range batch { @@ -1467,7 +1533,7 @@ func (s *sAdmin) ExportBillingRecords(ctx context.Context, req *v1.AdminBillingR if len(batch) < 1000 { break } - offset += 1000 + cursorID = lastID } }) } diff --git a/internal/logic/admin/member.go b/internal/logic/admin/member.go index 63eca45..8983148 100644 --- a/internal/logic/admin/member.go +++ b/internal/logic/admin/member.go @@ -130,17 +130,15 @@ func (s *sAdmin) CreateMember(ctx context.Context, req *v1.AdminMemberCreateReq) func (s *sAdmin) ListAllMembers(ctx context.Context, req *v1.AdminMemberListReq) (*v1.AdminMemberListRes, error) { page, pageSize := common.NormalizePagination(req.Page, req.PageSize) - m := dao.TntUsers.Ctx(ctx). - LeftJoin("tnt_tenants t ON tnt_users.tenant_id = t.id") - m = buildMemberFilters(m, req.Keyword, req.Status, req.Role, req.TenantID) - - total, err := m.Count() + // COUNT 不需要租户名称列，且 buildMemberFilters 的条件只引用 tnt_users.* —— + // 计数免联表（联表目标是 tnt_tenants 主键，1:1 不影响行数） + total, err := buildMemberFilters(dao.TntUsers.Ctx(ctx), req.Keyword, req.Status, req.Role, req.TenantID).Count() if err != nil { return nil, err } - // Rebuild model for data query (Count() modifies internal state) - m = dao.TntUsers.Ctx(ctx). + // 列表查询才需要 JOIN 补租户名称/代码 + m := dao.TntUsers.Ctx(ctx). LeftJoin("tnt_tenants t ON tnt_users.tenant_id = t.id") m = buildMemberFilters(m, req.Keyword, req.Status, req.Role, req.TenantID) @@ -393,12 +391,16 @@ func (s *sAdmin) ExportMembers(ctx context.Context, req *v1.AdminMemberExportReq } return nil, export.GenericExport(ctx, config, func(yield func(map[string]any) bool) { - offset := 0 + // keyset 翻页：id 游标替代 OFFSET，避免深翻页时重复扫弃前面的行（排序本就是 id DESC） + var cursorID int64 for { m := buildMemberFilters( dao.TntUsers.Ctx(ctx).LeftJoin("tnt_tenants t ON tnt_users.tenant_id = t.id"), req.Keyword, req.Status, req.Role, req.TenantID, ) + if cursorID > 0 { + m = m.Where("tnt_users.id < ?", cursorID) + } var batch []struct { Id int64 `json:"id"` TenantName string `json:"tenant_name"` @@ -410,7 +412,7 @@ func (s *sAdmin) ExportMembers(ctx context.Context, req *v1.AdminMemberExportReq LastLoginAt *gtime.Time `json:"last_login_at"` CreatedAt *gtime.Time `json:"created_at"` } - if err := m.Fields(memberFields).OrderDesc("tnt_users.id").Limit(1000).Offset(offset).Scan(&batch); err != nil { + if err := m.Fields(memberFields).OrderDesc("tnt_users.id").Limit(1000).Scan(&batch); err != nil { return } for _, m := range batch { @@ -427,11 +429,11 @@ func (s *sAdmin) ExportMembers(ctx context.Context, req *v1.AdminMemberExportReq }) { return } + cursorID = m.Id } if len(batch) < 1000 { break } - offset += 1000 } }) } diff --git a/internal/logic/billing/transaction.go b/internal/logic/billing/transaction.go index 73f2b38..0b8b360 100644 --- a/internal/logic/billing/transaction.go +++ b/internal/logic/billing/transaction.go @@ -36,7 +36,11 @@ func BuildTransactionQuery(ctx context.Context, params TransactionQueryParams) * query = query.Where("bil_transactions.created_at <= ?", params.EndDate+" 23:59:59") } if params.Username != "" { - query = query.Where("tu.username LIKE ?", "%"+params.Username+"%") + // 用 EXISTS 子查询而非引用调用方 JOIN 的别名（tu.username）： + // ① 筛选条件因此不依赖任何 JOIN，调用方的 COUNT / 聚合路径可以完全不联表； + // ② 避免把 tnt_users 拖成驱动表（EXISTS 内是主键等值探测，代价低）。 + // 语义与 LEFT JOIN 后过滤一致：user_id 无对应用户的行同样被过滤掉。 + query = query.Where("EXISTS (SELECT 1 FROM tnt_users ut WHERE ut.id = bil_transactions.user_id AND ut.tenant_id = bil_transactions.tenant_id AND ut.username LIKE ?)", "%"+params.Username+"%") } if params.ModelName != "" { query = query.Where("bil_transactions.model_name LIKE ?", "%"+params.ModelName+"%") diff --git a/internal/logic/common/validation.go b/internal/logic/common/validation.go index 3c43059..c5b500c 100644 --- a/internal/logic/common/validation.go +++ b/internal/logic/common/validation.go @@ -2,12 +2,14 @@ package common import ( "context" + "fmt" "strings" "time" "unicode" "github.com/gogf/gf/v2/errors/gcode" "github.com/gogf/gf/v2/errors/gerror" + "github.com/gogf/gf/v2/os/gtime" "github.com/qianfree/team-api/internal/consts" ) @@ -111,6 +113,37 @@ func EndOfRange(value string) string { return value } +// UsageLogExportMaxWindowDays 用量日志类导出的时间窗上限（天），管理后台与租户控制台共用。 +const UsageLogExportMaxWindowDays = 31 + +// ValidateExportTimeWindow 校验导出必须限定时间窗：开始时间必填，结束时间缺省按 +// 「现在」计（与前端「截止留空 = 到现在」的口径一致），窗口不得超过 maxDays 天。 +// +// 列表接口天然有分页 LIMIT 兜底，导出没有 —— 不限窗口等于允许一次请求扫全量大表。 +// 更长跨度的取数应按窗口分多次导出。 +func ValidateExportTimeWindow(start, end string, maxDays int) error { + if start == "" { + return NewBadRequestError(fmt.Sprintf("导出必须指定开始时间，且时间跨度不超过 %d 天", maxDays)) + } + startAt := gtime.NewFromStr(StartOfRange(start)) + if startAt == nil { + return NewBadRequestError("开始时间格式不正确") + } + endAt := gtime.Now() + if end != "" { + if endAt = gtime.NewFromStr(EndOfRange(end)); endAt == nil { + return NewBadRequestError("结束时间格式不正确") + } + } + if endAt.Before(startAt) { + return NewBadRequestError("结束时间不能早于开始时间") + } + if endAt.Sub(startAt) > time.Duration(maxDays)*24*time.Hour { + return NewBadRequestError(fmt.Sprintf("导出时间跨度不能超过 %d 天，请缩小范围后分批导出", maxDays)) + } + return nil +} + // ValidateForbiddenWords 校验名称是否包含系统禁用词 // 从系统配置 register_forbidden_words 读取逗号分隔的禁用词列表，对 value 做大小写不敏感的包含检查 func ValidateForbiddenWords(ctx context.Context, value, fieldName string) error { diff --git a/internal/logic/tenant/billing.go b/internal/logic/tenant/billing.go index a4d6991..7a53f51 100644 --- a/internal/logic/tenant/billing.go +++ b/internal/logic/tenant/billing.go @@ -116,19 +116,24 @@ func (s *sTenant) WalletTransactions(ctx context.Context, req *v1.TenantWalletTr CreatedAt *gtime.Time `json:"created_at"` } - var records []transactionRow - var err error - var total int - err = query.Fields("bil_transactions.id, bil_transactions.type, bil_transactions.amount, bil_transactions.balance_after, bil_transactions.frozen_after, bil_transactions.related_id, bil_transactions.related_type, bil_transactions.description, bil_transactions.user_id, COALESCE(tu.username, '') AS username, bil_transactions.request_id, bil_transactions.model_name, bil_transactions.project_id, bil_transactions.api_key_id, bil_transactions.task_id, bil_transactions.created_at"). - LeftJoin("tnt_users tu", "bil_transactions.user_id = tu.id AND bil_transactions.tenant_id = tu.tenant_id"). - OrderDesc("bil_transactions.created_at"). - Page(page, pageSize). - ScanAndCount(&records, &total, false) + // COUNT 与列表拆开执行：计数不需要用户名称列，筛选条件也不依赖 JOIN + //（Username 过滤在 BuildTransactionQuery 内是 EXISTS 子查询），大表 + // bil_transactions 的计数免背联表。不用 ScanAndCount —— 它的 COUNT + // 会携带模型上的全部 JOIN。 + total, err := query.Count() if err != nil { return nil, err } - if records == nil { - records = []transactionRow{} + + records := []transactionRow{} + if total > 0 { + if err := query.Fields("bil_transactions.id, bil_transactions.type, bil_transactions.amount, bil_transactions.balance_after, bil_transactions.frozen_after, bil_transactions.related_id, bil_transactions.related_type, bil_transactions.description, bil_transactions.user_id, COALESCE(tu.username, '') AS username, bil_transactions.request_id, bil_transactions.model_name, bil_transactions.project_id, bil_transactions.api_key_id, bil_transactions.task_id, bil_transactions.created_at"). + LeftJoin("tnt_users tu", "bil_transactions.user_id = tu.id AND bil_transactions.tenant_id = tu.tenant_id"). + OrderDesc("bil_transactions.created_at"). + Page(page, pageSize). + Scan(&records); err != nil { + return nil, err + } } list := make([]map[string]any, 0, len(records)) @@ -161,6 +166,63 @@ func (s *sTenant) WalletTransactions(ctx context.Context, req *v1.TenantWalletTr }, nil } +// tenantUsageLogFilter 用量日志查询的筛选参数（列表 / 统计 / 导出三个入口共用）。 +type tenantUsageLogFilter struct { + TenantID int64 + UserID int64 + Role string + Username string + Model string + Status string + RequestType int + StartDate string + EndDate string +} + +// buildTenantUsageLogFilter 构建用量日志查询的 WHERE 条件与绑定参数（不含 WHERE 前缀； +// tenant_id 恒在条件里，结果必非空）。三个入口共用，保证筛选口径一致。 +// +// 条件只引用主表别名 u，不依赖任何 JOIN —— 计数与统计路径因此可以完全不联表， +// 只有需要展示名称的列表 / 导出才挂 LEFT JOIN。租户隔离（u.tenant_id）与 +// member 角色只见自己（u.user_id）在此统一强制，调用方不得绕过。 +func buildTenantUsageLogFilter(f tenantUsageLogFilter) (where string, args []any) { + conditions := []string{"u.tenant_id = ?"} + args = append(args, f.TenantID) + + // member 角色只能查看自己的用量日志 + if f.Role == "member" { + conditions = append(conditions, "u.user_id = ?") + args = append(args, f.UserID) + } else if f.Username != "" { + // EXISTS 子查询替代对 JOIN 别名 t.username 的引用：条件不再依赖联表， + // 且避免把 tnt_users 拖成驱动表（EXISTS 内是主键等值探测，代价低）。 + // 语义与 LEFT JOIN 后过滤一致：用户不存在的行同样被过滤掉。 + conditions = append(conditions, "EXISTS (SELECT 1 FROM tnt_users ut WHERE ut.id = u.user_id AND ut.tenant_id = u.tenant_id AND ut.username LIKE ?)") + args = append(args, "%"+f.Username+"%") + } + if f.Model != "" { + conditions = append(conditions, "u.model_name = ?") + args = append(args, f.Model) + } + if f.Status != "" { + conditions = append(conditions, "u.status = ?") + args = append(args, f.Status) + } + if f.RequestType > 0 { + conditions = append(conditions, "u.request_type = ?") + args = append(args, f.RequestType) + } + if f.StartDate != "" { + conditions = append(conditions, "u.created_at >= ?") + args = append(args, common.StartOfRange(f.StartDate)) + } + if f.EndDate != "" { + conditions = append(conditions, "u.created_at <= ?") + args = append(args, common.EndOfRange(f.EndDate)) + } + return strings.Join(conditions, " AND "), args +} + // UsageLogs 获取租户用量日志 func (s *sTenant) UsageLogs(ctx context.Context, req *v1.TenantUsageLogsReq) (*v1.TenantUsageLogsRes, error) { if err := common.ValidateDateTimeParam(req.StartDate, "开始时间"); err != nil { @@ -175,46 +237,25 @@ func (s *sTenant) UsageLogs(ctx context.Context, req *v1.TenantUsageLogsReq) (*v role := middleware.GetUserRole(ctx) page, pageSize := common.NormalizePagination(req.Page, req.PageSize) - var conditions []string - var args []any + where, args := buildTenantUsageLogFilter(tenantUsageLogFilter{ + TenantID: tenantID, + UserID: userID, + Role: role, + Username: req.Username, + Model: req.Model, + Status: req.Status, + RequestType: req.RequestType, + StartDate: req.StartDate, + EndDate: req.EndDate, + }) - conditions = append(conditions, "u.tenant_id = ?") - args = append(args, tenantID) - - // member 角色只能查看自己的用量日志 - if role == "member" { - conditions = append(conditions, "u.user_id = ?") - args = append(args, userID) - } else if req.Username != "" { - conditions = append(conditions, "t.username LIKE ?") - args = append(args, "%"+req.Username+"%") - } - if req.Model != "" { - conditions = append(conditions, "u.model_name = ?") - args = append(args, req.Model) - } - if req.Status != "" { - conditions = append(conditions, "u.status = ?") - args = append(args, req.Status) - } - if req.RequestType > 0 { - conditions = append(conditions, "u.request_type = ?") - args = append(args, req.RequestType) - } - if req.StartDate != "" { - conditions = append(conditions, "u.created_at >= ?") - args = append(args, common.StartOfRange(req.StartDate)) - } - if req.EndDate != "" { - conditions = append(conditions, "u.created_at <= ?") - args = append(args, common.EndOfRange(req.EndDate)) - } - - where := strings.Join(conditions, " AND ") // 列表查询不 JOIN mdl_models（model_display_name 仅导出使用） fromClause := "bil_usage_logs u LEFT JOIN tnt_users t ON u.user_id = t.id AND u.tenant_id = t.tenant_id LEFT JOIN tnt_projects p ON u.project_id = p.id LEFT JOIN api_keys ak ON u.api_key_id = ak.id" - countSQL := "SELECT COUNT(*) AS total FROM " + fromClause + " WHERE " + where + // 计数不展示任何名称，且筛选条件只引用 u.*（Username 过滤是 EXISTS 子查询）， + // 3 个展示用 LEFT JOIN 整体剥离 —— bil_usage_logs 是分区大表，列表每翻一页 + // 都要重算一次计数，这里是本接口最重的一次查询。 + countSQL := "SELECT COUNT(*) AS total FROM bil_usage_logs u WHERE " + where countResult, err := g.DB().Ctx(ctx).Query(ctx, countSQL, args...) if err != nil { return nil, err @@ -288,53 +329,26 @@ func (s *sTenant) UsageLogsSummary(ctx context.Context, req *v1.TenantUsageLogsS userID := middleware.GetUserID(ctx) role := middleware.GetUserRole(ctx) - var conditions []string - var args []any - - conditions = append(conditions, "u.tenant_id = ?") - args = append(args, tenantID) - - // member 角色只能查看自己的用量日志 - if role == "member" { - conditions = append(conditions, "u.user_id = ?") - args = append(args, userID) - } else if req.Username != "" { - conditions = append(conditions, "t.username LIKE ?") - args = append(args, "%"+req.Username+"%") - } - if req.Model != "" { - conditions = append(conditions, "u.model_name = ?") - args = append(args, req.Model) - } - if req.Status != "" { - conditions = append(conditions, "u.status = ?") - args = append(args, req.Status) - } - if req.RequestType > 0 { - conditions = append(conditions, "u.request_type = ?") - args = append(args, req.RequestType) - } - if req.StartDate != "" { - conditions = append(conditions, "u.created_at >= ?") - args = append(args, common.StartOfRange(req.StartDate)) - } - if req.EndDate != "" { - conditions = append(conditions, "u.created_at <= ?") - args = append(args, common.EndOfRange(req.EndDate)) - } - - // 统计聚合不展示名称，仅在按用户名筛选时才 JOIN 用户表，避免大表 SUM 背负无关 JOIN - fromClause := "bil_usage_logs u" - if role != "member" && req.Username != "" { - fromClause += " LEFT JOIN tnt_users t ON u.user_id = t.id AND u.tenant_id = t.tenant_id" - } + where, args := buildTenantUsageLogFilter(tenantUsageLogFilter{ + TenantID: tenantID, + UserID: userID, + Role: role, + Username: req.Username, + Model: req.Model, + Status: req.Status, + RequestType: req.RequestType, + StartDate: req.StartDate, + EndDate: req.EndDate, + }) + // 统计聚合不展示名称，Username 过滤已由 EXISTS 子查询承担，全程无需联表： + // 大表 SUM 不再背负任何 join。 summarySQL := `SELECT COALESCE(SUM(COALESCE(NULLIF(u.actual_cost, 0), u.total_cost)), 0) AS total_cost, COALESCE(SUM(u.output_tokens), 0) AS total_output_tokens, COALESCE(SUM(u.input_tokens), 0) AS total_input_tokens, COALESCE(SUM(u.cache_read_tokens), 0) AS total_cache_read - FROM ` + fromClause + ` WHERE ` + strings.Join(conditions, " AND ") + FROM bil_usage_logs u WHERE ` + where summaryResult, err := g.DB().Ctx(ctx).Query(ctx, summarySQL, args...) if err != nil { @@ -379,6 +393,10 @@ func (s *sTenant) ExportUsageLogs(ctx context.Context, req *v1.TenantUsageLogsEx if err := common.ValidateDateTimeParam(req.EndDate, "结束时间"); err != nil { return nil, err } + // 导出护栏：时间窗必填且不超上限（与管理后台导出同一规则） + if err := common.ValidateExportTimeWindow(req.StartDate, req.EndDate, common.UsageLogExportMaxWindowDays); err != nil { + return nil, err + } tenantID := middleware.GetTenantID(ctx) userID := middleware.GetUserID(ctx) @@ -403,58 +421,39 @@ func (s *sTenant) ExportUsageLogs(ctx context.Context, req *v1.TenantUsageLogsEx Columns: columns, } - var conditions []string - var args []any + where, args := buildTenantUsageLogFilter(tenantUsageLogFilter{ + TenantID: tenantID, + UserID: userID, + Role: role, + Username: req.Username, + Model: req.Model, + Status: req.Status, + RequestType: req.RequestType, + StartDate: req.StartDate, + EndDate: req.EndDate, + }) - conditions = append(conditions, "u.tenant_id = ?") - args = append(args, tenantID) - - // member 角色只能导出自己的用量日志 - if role == "member" { - conditions = append(conditions, "u.user_id = ?") - args = append(args, userID) - } else if req.Username != "" { - conditions = append(conditions, "t.username LIKE ?") - args = append(args, "%"+req.Username+"%") - } - if req.Model != "" { - conditions = append(conditions, "u.model_name = ?") - args = append(args, req.Model) - } - if req.Status != "" { - conditions = append(conditions, "u.status = ?") - args = append(args, req.Status) - } - if req.RequestType > 0 { - conditions = append(conditions, "u.request_type = ?") - args = append(args, req.RequestType) - } - if req.StartDate != "" { - conditions = append(conditions, "u.created_at >= ?") - args = append(args, common.StartOfRange(req.StartDate)) - } - if req.EndDate != "" { - conditions = append(conditions, "u.created_at <= ?") - args = append(args, common.EndOfRange(req.EndDate)) - } - - where := strings.Join(conditions, " AND ") fromClause := "bil_usage_logs u LEFT JOIN tnt_users t ON u.user_id = t.id AND u.tenant_id = t.tenant_id LEFT JOIN tnt_projects p ON u.project_id = p.id LEFT JOIN api_keys ak ON u.api_key_id = ak.id LEFT JOIN mdl_models mdl ON u.model_name = mdl.model_id" return nil, export.GenericExport(ctx, config, func(yield func(map[string]any) bool) { - offset := 0 + // keyset（游标）翻页替代 OFFSET：OFFSET 每翻一批都要重新扫过并丢弃前面的 + // 所有行，深翻页代价线性上涨；id 游标（bigserial 单调递增，走 PK 索引） + // 让每批从上一批断点续读。排序 created_at DESC → id DESC，追加写表近似同序， + // 且避免时间戳做游标时亚秒精度在驱动往返中丢失导致的跳行/重行。 + var cursorID int64 for { - dataSQL := fmt.Sprintf( - `SELECT u.id, COALESCE(t.username, '') AS username, COALESCE(mdl.model_name, '') AS model_display_name, u.model_name, u.request_type, - u.input_tokens, u.output_tokens, u.total_cost, u.status, u.created_at - FROM %s WHERE %s ORDER BY u.created_at DESC LIMIT 1000 OFFSET ?`, - fromClause, where, - ) - exportArgs := make([]any, len(args)+1) - copy(exportArgs, args) - exportArgs[len(args)] = offset + dataSQL := `SELECT u.id, COALESCE(t.username, '') AS username, COALESCE(mdl.model_name, '') AS model_display_name, u.model_name, u.request_type, + u.input_tokens, u.output_tokens, u.total_cost, u.status, u.created_at + FROM ` + fromClause + ` WHERE ` + where + exportArgs := append([]any{}, args...) + if cursorID > 0 { + dataSQL += " AND u.id < ?" + exportArgs = append(exportArgs, cursorID) + } + dataSQL += " ORDER BY u.id DESC LIMIT 1000" result, err := g.DB().Ctx(ctx).Query(ctx, dataSQL, exportArgs...) if err != nil { + g.Log().Errorf(ctx, "ExportUsageLogs: query batch failed (cursor id=%d): %v", cursorID, err) return } for _, row := range result { @@ -475,11 +474,11 @@ func (s *sTenant) ExportUsageLogs(ctx context.Context, req *v1.TenantUsageLogsEx if !yield(m) { return } + cursorID = row["id"].Int64() } if len(result) < 1000 { break } - offset += 1000 } }) } @@ -509,7 +508,9 @@ func (s *sTenant) ExportWalletTransactions(ctx context.Context, req *v1.TenantWa } return nil, export.GenericExport(ctx, config, func(yield func(map[string]any) bool) { - offset := 0 + // keyset 翻页：id 游标替代 OFFSET，避免深翻页时重复扫弃前面的行； + // 排序取 id DESC（追加写表上与 created_at DESC 近似同序） + var cursorID int64 for { type transactionRow struct { Id int64 `json:"id"` @@ -544,11 +545,14 @@ func (s *sTenant) ExportWalletTransactions(ctx context.Context, req *v1.TenantWa if req.ModelName != "" { q = q.Where("model_name LIKE ?", "%"+req.ModelName+"%") } + if cursorID > 0 { + q = q.Where("id < ?", cursorID) + } var records []transactionRow err := q.Fields("id, type, amount, balance_after, user_id, request_id, model_name, description, created_at"). - OrderDesc("created_at"). - Limit(1000).Offset(offset). + OrderDesc("id"). + Limit(1000). Scan(&records) if err != nil { return @@ -564,11 +568,11 @@ func (s *sTenant) ExportWalletTransactions(ctx context.Context, req *v1.TenantWa }) { return } + cursorID = r.Id } if len(records) < 1000 { break } - offset += 1000 } }) } diff --git a/internal/logic/tenant/member.go b/internal/logic/tenant/member.go index 40c07bb..e69a8b7 100644 --- a/internal/logic/tenant/member.go +++ b/internal/logic/tenant/member.go @@ -1163,7 +1163,8 @@ func (s *sTenant) ExportMembers(ctx context.Context, req *v1.TenantMemberExportR } return nil, export.GenericExport(ctx, config, func(yield func(map[string]any) bool) { - offset := 0 + // keyset 翻页：id 游标替代 OFFSET，避免深翻页时重复扫弃前面的行（排序本就是 id DESC） + var cursorID int64 for { model := dao.TntUsers.Ctx(ctx). Where("tenant_id", tenantID) @@ -1174,6 +1175,9 @@ func (s *sTenant) ExportMembers(ctx context.Context, req *v1.TenantMemberExportR if req.Role != "" { model = model.Where("role", req.Role) } + if cursorID > 0 { + model = model.Where("id < ?", cursorID) + } var users []struct { Id int64 `json:"id" orm:"id"` @@ -1188,7 +1192,7 @@ func (s *sTenant) ExportMembers(ctx context.Context, req *v1.TenantMemberExportR QuotaPeriod string `json:"quota_period" orm:"quota_period"` UpdatedAt string `json:"updated_at" orm:"updated_at"` } - err := model.OrderDesc("id").Limit(1000).Offset(offset).Scan(&users) + err := model.OrderDesc("id").Limit(1000).Scan(&users) if err = common.IgnoreScanNoRows(err); err != nil { return } @@ -1265,11 +1269,11 @@ func (s *sTenant) ExportMembers(ctx context.Context, req *v1.TenantMemberExportR }) { return } + cursorID = u.Id } if len(users) < 1000 { break } - offset += 1000 } }) } diff --git a/internal/utility/export/export.go b/internal/utility/export/export.go index 658a837..7a314dd 100644 --- a/internal/utility/export/export.go +++ b/internal/utility/export/export.go @@ -2,7 +2,9 @@ package export import ( "context" + "fmt" + "github.com/gogf/gf/v2/errors/gerror" "github.com/gogf/gf/v2/frame/g" ) @@ -31,20 +33,57 @@ func (c Config) GetMaxRows() int { // GenericExport handles format normalization and xlsx/csv branching. // queryFn yields data rows — same callback as StreamCSV's queryFn. -// For xlsx: collects all rows from queryFn, then calls WriteExcel. -// For csv: passes queryFn directly to StreamCSV for streaming. +// For xlsx: collects rows from queryFn (capped), then calls WriteExcel. +// For csv: passes queryFn to StreamCSV for streaming (capped). +// +// 行数上限在这里统一强制执行（对拉取过程截停，而不是拉完再检查）： +// 导出的数据源是「无总量保证」的分批循环查询，上限是防止一次导出请求 +// 拖着数据库扫全量大表的最后一道护栏。 +// - xlsx：达到上限直接返回错误（此时还未写出任何响应，能干净地报错）， +// 并且不再继续拉取 —— 旧实现是全部拉进内存后才检查，护栏形同虚设； +// - csv：边查边流式写出，表头和数据已经发给客户端，超限时无法再改报错误， +// 只能停止拉取并在文件末尾追加一行可见的截断标记，避免使用者 +// 误把部分数据当作全量。 func GenericExport(ctx context.Context, config Config, queryFn func(yield func(map[string]any) bool)) error { r := g.RequestFromCtx(ctx) config.Format = detectFormat(config.Format) + maxRows := config.GetMaxRows() if config.Format == "xlsx" { - var data []map[string]any + var ( + data []map[string]any + overflow bool + ) queryFn(func(row map[string]any) bool { + if len(data) >= maxRows { + overflow = true + return false + } data = append(data, row) return true }) + if overflow { + return gerror.Newf("数据量超过 %d 行，请缩小筛选范围后重试，或改用 CSV 格式导出", maxRows) + } return WriteExcel(r, config, data) } - return StreamCSV(r, config, queryFn) + truncated := false + capped := func(yield func(map[string]any) bool) { + rowCount := 0 + queryFn(func(row map[string]any) bool { + if rowCount >= maxRows { + truncated = true + return false + } + rowCount++ + return yield(row) + }) + if truncated && len(config.Columns) > 0 { + yield(map[string]any{ + config.Columns[0].Field: fmt.Sprintf("注意：已达到导出上限 %d 行，结果被截断，请缩小时间范围或筛选条件后分批导出", maxRows), + }) + } + } + return StreamCSV(r, config, capped) }