perf(ops-report): eliminate logs full-table scans from the report path

controller/ops_report.go

@@ -370,10 +370,11 @@ func buildOpsAggs(includeDisabled bool) (map[int]*opsUserAgg, error) {
 }
 
 // getOpsAggs returns the cached day-independent cohort aggregates, recomputing
-// them at most once per opsReportCacheTTL per node. This is the expensive part
-// of the report: GetOpsUserLogStats scans the whole logs history for the plg
-// cohort no matter which day range is selected, so caching it here keeps a
-// 7/30/60/90 switch from re-running that scan each time.
+// them at most once per opsReportCacheTTL per node. The user-log aggregates
+// come from the ops_user_log_stats table (background-incremented) rather than
+// a live scan of the whole logs history, so a days switch never re-runs a
+// multi-minute logs scan; the cache here still avoids rebuilding the in-memory
+// funnel rollups on every 7/30/60/90 switch.
 func getOpsAggs(includeDisabled bool) (map[int]*opsUserAgg, time.Time, error) {
 	opsAggsMutex.Lock()
 	defer opsAggsMutex.Unlock()

controller/ops_report_stripe.go

@@ -464,13 +464,15 @@ func buildOpsStripeReport(days int) (*opsStripeReport, error) {
 		sort.Strings(out)
 		return out
 	}
-	// last request IP (any log row, playground included) as an identity hint
+	// last request IP within the report window (any log row, playground
+	// included) as an identity hint; the time bound keeps the logs lookup
+	// off the oldest history
 	personIds := make([]int, 0, len(persons))
 	for _, a := range persons {
 		personIds = append(personIds, a.row.UserId)
 	}
 	ipByUser := map[int]string{}
-	if ips, err := model.GetOpsUsersLastIP(personIds); err == nil {
+	if ips, err := model.GetOpsUsersLastIP(personIds, startTs); err == nil {
 		for _, r := range ips {
 			ipByUser[r.UserId] = r.Ip
 		}

main.go

@@ -128,6 +128,9 @@ func main() {
 	// 数据看板
 	go model.UpdateQuotaData()
 
+	// ops 日报的用户日志预聚合(仅在 master 节点跑)
+	go model.StartOpsUserLogStatsSyncTask()
+
 	if os.Getenv("CHANNEL_UPDATE_FREQUENCY") != "" {
 		frequency, err := strconv.Atoi(os.Getenv("CHANNEL_UPDATE_FREQUENCY"))
 		if err != nil {

model/main.go

@@ -929,7 +929,7 @@ func hasActiveRecallMigrationLeases(nowUnix int64) (bool, error) {
 
 func migrateLOGDB() error {
 	var err error
-	if err = LOG_DB.AutoMigrate(&Log{}, &CompanyLogSchema{}, &LogRequestSample{}, &TaskAcceptedAccountingLogLedger{}); err != nil {
+	if err = LOG_DB.AutoMigrate(&Log{}, &CompanyLogSchema{}, &LogRequestSample{}, &TaskAcceptedAccountingLogLedger{}, &OpsUserLogStatsRow{}, &OpsUserLogStatsMeta{}); err != nil {
 		return err
 	}
 	return nil

model/ops_report.go

@@ -144,7 +144,31 @@ func chunkInts(ids []int, size int) [][]int {
 }
 
 // GetOpsUserLogStats returns per-user playground/API-key usage aggregates.
+// Prefer the ops_user_log_stats aggregate table (maintained incrementally in
+// the background) so the report never scans the whole logs history; fall back
+// to the direct logs scan only until the first backfill has populated the
+// table (e.g. right after a fresh deploy).
 func GetOpsUserLogStats(userIds []int) ([]*OpsUserLogStats, error) {
+	if opsUserLogStatsReady() {
+		var all []*OpsUserLogStats
+		for _, chunk := range chunkInts(userIds, opsReportChunkSize) {
+			var batch []*OpsUserLogStats
+			if err := LOG_DB.Table("ops_user_log_stats").
+				Select("user_id, first_playground_at, playground_count, first_api_key_at, api_key_count, last_request_at").
+				Where("user_id IN ?", chunk).
+				Scan(&batch).Error; err != nil {
+				return nil, err
+			}
+			all = append(all, batch...)
+		}
+		return all, nil
+	}
+	return getOpsUserLogStatsFromLogs(userIds)
+}
+
+// getOpsUserLogStatsFromLogs is the original direct logs scan, kept as the
+// fallback until the aggregate table is populated.
+func getOpsUserLogStatsFromLogs(userIds []int) ([]*OpsUserLogStats, error) {
 	var all []*OpsUserLogStats
 	for _, chunk := range chunkInts(userIds, opsReportChunkSize) {
 		var batch []*OpsUserLogStats
@@ -171,6 +195,12 @@ func GetOpsUserLogStats(userIds []int) ([]*OpsUserLogStats, error) {
 // holds n+1 ascending UTC epoch boundaries — the real report-timezone midnights
 // for the n days — so buckets stay correct across DST transitions. day_ts
 // values are the per-day start epochs.
+//
+// Data now comes from quota_data (hourly per-user-per-model rollups, ~500
+// rows/day) instead of raw logs: a 30-day window covers nearly the whole logs
+// table, so the optimizer full-scans ~45M rows there (measured 100s+ on prod).
+// Trade-off (same as GetOpsAllKeyDailyUsage): quota_data counts all consumption
+// including playground, not only token_id>0 API-key calls.
 func GetOpsKeyDailyUsage(userIds []int, dayStarts []int64) ([]*OpsKeyDaily, error) {
 	if len(dayStarts) < 2 {
 		return nil, nil
@@ -183,12 +213,12 @@ func GetOpsKeyDailyUsage(userIds []int, dayStarts []int64) ([]*OpsKeyDaily, erro
 		sql := fmt.Sprintf(`
 			SELECT user_id,
 			       %s AS day_ts,
-			       COUNT(*) AS req_count,
+			       COALESCE(SUM(count), 0) AS req_count,
 			       COALESCE(SUM(quota), 0) AS quota
-			FROM logs%s
-			WHERE type = ? AND %s AND created_at >= ? AND user_id IN ?
-			GROUP BY user_id, %s`, dayExpr, logsForceIndexHint(), opsExternalAPIKeyLogPredicate, dayExpr)
-		if err := LOG_DB.Raw(sql, LogTypeConsume, startTs, chunk).Scan(&batch).Error; err != nil {
+			FROM quota_data
+			WHERE created_at >= ? AND user_id IN ?
+			GROUP BY user_id, %s`, dayExpr, dayExpr)
+		if err := DB.Raw(sql, startTs, chunk).Scan(&batch).Error; err != nil {
 			return nil, err
 		}
 		all = append(all, batch...)
@@ -332,19 +362,21 @@ type OpsUserLastIP struct {
 	Ip     string `json:"ip"`
 }
 
-// GetOpsUsersLastIP returns the most recent non-empty request IP per user.
-// One indexed MAX(id) pass plus one primary-key lookup; used for the full plg
-// user set (~thousands) by the ops report region funnel.
-func GetOpsUsersLastIP(userIds []int) ([]*OpsUserLastIP, error) {
+// GetOpsUsersLastIP returns the most recent non-empty request IP per user,
+// restricted to logs at or after since (the report window start). One indexed
+// MAX(id) pass plus one primary-key lookup; used for the full plg user set
+// (~thousands) by the ops report region funnel. The time bound keeps the MAX
+// pass off the oldest logs, which otherwise grows without bound.
+func GetOpsUsersLastIP(userIds []int, since int64) ([]*OpsUserLastIP, error) {
 	if len(userIds) == 0 {
 		return nil, nil
 	}
 	var maxIds []int64
 	sql := fmt.Sprintf(`
 		SELECT MAX(id) FROM logs%s
-		WHERE user_id IN ? AND ip <> ''
+		WHERE user_id IN ? AND ip <> '' AND created_at >= ?
 		GROUP BY user_id`, logsForceIndexHint())
-	if err := LOG_DB.Raw(sql, userIds).Scan(&maxIds).Error; err != nil {
+	if err := LOG_DB.Raw(sql, userIds, since).Scan(&maxIds).Error; err != nil {
 		return nil, err
 	}
 	if len(maxIds) == 0 {

model/ops_user_log_stats.go

@@ -0,0 +1,230 @@
+package model
+
+import (
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/QuantumNous/new-api/common"
+
+	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
+)
+
+// OpsUserLogStatsRow is the pre-aggregated per-user consume-log statistics that
+// back GetOpsUserLogStats. The alternative — scanning the whole logs history
+// per user on every report rebuild — full-scans a ~45M-row table on prod
+// (measured 100s+), so the report now reads this small table instead.
+//
+// The table lives in LOG_DB next to logs (a separate database when
+// LOG_SQL_DSN is set) and is maintained by StartOpsUserLogStatsSyncTask:
+// an incremental pass aggregates every consume log newer than the stored
+// cursor and upserts per-user rows. The first run backfills from the whole
+// logs history (one-time, in the background, never on the request path).
+//
+// Semantics mirror GetOpsUserLogStats exactly:
+//   - playground = token_name LIKE 'playground%' (auto-fired onboarding call)
+//   - api key    = token_id > 0 AND NOT playground (opsExternalAPIKeyLogPredicate)
+//
+// Deleting old logs (DeleteOldLog) does not retroactively shrink these rows:
+// the cursor only moves forward, and first_* values were captured from the
+// earliest matching log seen so far.
+
+type OpsUserLogStatsRow struct {
+	UserId            int   `gorm:"column:user_id;primaryKey;autoIncrement:false"`
+	FirstPlaygroundAt int64 `gorm:"column:first_playground_at;default:0"`
+	PlaygroundCount   int   `gorm:"column:playground_count;default:0"`
+	FirstApiKeyAt     int64 `gorm:"column:first_api_key_at;default:0"`
+	ApiKeyCount       int   `gorm:"column:api_key_count;default:0"`
+	LastRequestAt     int64 `gorm:"column:last_request_at;default:0"`
+	UpdatedAt         int64 `gorm:"column:updated_at;default:0"`
+}
+
+func (OpsUserLogStatsRow) TableName() string {
+	return "ops_user_log_stats"
+}
+
+// OpsUserLogStatsMeta is the single-row sync cursor for the aggregation task:
+// LastLogId is the highest logs.id already folded into ops_user_log_stats.
+// Backfilled flips true only after the first full pass has caught up with the
+// log tail — until then the report must keep using the slow direct scan rather
+// than reading a half-populated table.
+type OpsUserLogStatsMeta struct {
+	Id         int   `gorm:"primaryKey;autoIncrement:false"` // always 1
+	LastLogId  int64 `gorm:"column:last_log_id;default:0"`
+	Backfilled bool  `gorm:"column:backfilled;default:false"`
+	UpdatedAt  int64 `gorm:"column:updated_at;default:0"`
+}
+
+func (OpsUserLogStatsMeta) TableName() string {
+	return "ops_user_log_stats_meta"
+}
+
+const (
+	opsUserLogStatsSyncBatch = 50000
+	opsUserLogStatsSyncEvery = 5 * time.Minute
+)
+
+// getOpsUserLogStatsMeta reads the single cursor row, creating it on first use.
+func getOpsUserLogStatsMeta() (*OpsUserLogStatsMeta, error) {
+	var meta OpsUserLogStatsMeta
+	err := LOG_DB.Where("id = 1").First(&meta).Error
+	if err == nil {
+		return &meta, nil
+	}
+	if err != gorm.ErrRecordNotFound {
+		return nil, err
+	}
+	meta = OpsUserLogStatsMeta{Id: 1}
+	if err := LOG_DB.Create(&meta).Error; err != nil {
+		return nil, err
+	}
+	return &meta, nil
+}
+
+// opsUserLogStatsReady reports whether the aggregate table is fully usable:
+// the cursor has advanced past the first backfill (a full pass that caught up
+// with the log tail). Single-row primary-key read, cheap.
+func opsUserLogStatsReady() bool {
+	meta, err := getOpsUserLogStatsMeta()
+	return err == nil && meta.Backfilled
+}
+
+// SyncOpsUserLogStats runs one incremental aggregation pass: consume logs with
+// id > cursor are aggregated per user and upserted into ops_user_log_stats,
+// then the cursor advances. The first pass (cursor == 0) backfills the whole
+// logs history in batches. Safe on multi-node: only the master node runs the
+// task, and per-user upserts are idempotent by primary key.
+func SyncOpsUserLogStats() error {
+	meta, err := getOpsUserLogStatsMeta()
+	if err != nil {
+		return err
+	}
+	cursor := meta.LastLogId
+	reachedTail := false
+	for {
+		var logs []*Log
+		if err := LOG_DB.Select("id", "user_id", "created_at", "token_name", "token_id").
+			Where("type = ? AND id > ?", LogTypeConsume, cursor).
+			Order("id").
+			Limit(opsUserLogStatsSyncBatch).
+			Find(&logs).Error; err != nil {
+			return err
+		}
+		if len(logs) == 0 {
+			// No rows newer than the cursor: the pass reached the log tail.
+			reachedTail = true
+			break
+		}
+		agg := map[int]*OpsUserLogStatsRow{}
+		for _, l := range logs {
+			row, ok := agg[l.UserId]
+			if !ok {
+				row = &OpsUserLogStatsRow{UserId: l.UserId}
+				agg[l.UserId] = row
+			}
+			if strings.HasPrefix(l.TokenName, "playground") {
+				row.PlaygroundCount++
+				if row.FirstPlaygroundAt == 0 || l.CreatedAt < row.FirstPlaygroundAt {
+					row.FirstPlaygroundAt = l.CreatedAt
+				}
+			} else if l.TokenId > 0 {
+				row.ApiKeyCount++
+				if row.FirstApiKeyAt == 0 || l.CreatedAt < row.FirstApiKeyAt {
+					row.FirstApiKeyAt = l.CreatedAt
+				}
+			}
+			if l.CreatedAt > row.LastRequestAt {
+				row.LastRequestAt = l.CreatedAt
+			}
+		}
+
+		// Merge with the rows already in the table (the same user can appear
+		// across batches): counts accumulate, first_* keep the earliest
+		// non-zero value, last_request_at takes the max. Then overwrite the
+		// whole row so the upsert is portable across SQLite/MySQL/PostgreSQL.
+		ids := make([]int, 0, len(agg))
+		for uid := range agg {
+			ids = append(ids, uid)
+		}
+		var existing []OpsUserLogStatsRow
+		if err := LOG_DB.Where("user_id IN ?", ids).Find(&existing).Error; err != nil {
+			return err
+		}
+		byId := map[int]*OpsUserLogStatsRow{}
+		for i := range existing {
+			byId[existing[i].UserId] = &existing[i]
+		}
+		now := common.GetTimestamp()
+		rows := make([]OpsUserLogStatsRow, 0, len(agg))
+		for _, row := range agg {
+			if old, ok := byId[row.UserId]; ok {
+				row.PlaygroundCount += old.PlaygroundCount
+				row.ApiKeyCount += old.ApiKeyCount
+				if old.FirstPlaygroundAt > 0 && (row.FirstPlaygroundAt == 0 || old.FirstPlaygroundAt < row.FirstPlaygroundAt) {
+					row.FirstPlaygroundAt = old.FirstPlaygroundAt
+				}
+				if old.FirstApiKeyAt > 0 && (row.FirstApiKeyAt == 0 || old.FirstApiKeyAt < row.FirstApiKeyAt) {
+					row.FirstApiKeyAt = old.FirstApiKeyAt
+				}
+				if old.LastRequestAt > row.LastRequestAt {
+					row.LastRequestAt = old.LastRequestAt
+				}
+			}
+			row.UpdatedAt = now
+			rows = append(rows, *row)
+		}
+		if err := LOG_DB.Clauses(clause.OnConflict{
+			Columns: []clause.Column{{Name: "user_id"}},
+			DoUpdates: clause.AssignmentColumns([]string{
+				"first_playground_at", "playground_count", "first_api_key_at",
+				"api_key_count", "last_request_at", "updated_at",
+			}),
+		}).Create(&rows).Error; err != nil {
+			return err
+		}
+		cursor = int64(logs[len(logs)-1].Id)
+		// Persist the cursor after every batch so a crash resumes from here.
+		if err := LOG_DB.Model(&OpsUserLogStatsMeta{}).Where("id = 1").
+			Updates(map[string]interface{}{"last_log_id": cursor, "updated_at": now}).Error; err != nil {
+			return err
+		}
+		if len(logs) < opsUserLogStatsSyncBatch {
+			// A short final batch means we drained the tail in this pass.
+			reachedTail = true
+			break
+		}
+	}
+	// Only a pass that actually reached the log tail completes the backfill;
+	// a partial first pass must not mark the table ready.
+	if reachedTail && !meta.Backfilled {
+		if err := LOG_DB.Model(&OpsUserLogStatsMeta{}).Where("id = 1").
+			Update("backfilled", true).Error; err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+var opsUserLogStatsTaskOnce sync.Once
+
+// StartOpsUserLogStatsSyncTask runs the incremental aggregation every
+// opsUserLogStatsSyncEvery on the master node (single writer per Rule 11; the
+// report reads are multi-node safe because the table is shared and upserts are
+// idempotent). The first pass is a full backfill, so the aggregate table
+// becomes available within one batch cycle of startup.
+func StartOpsUserLogStatsSyncTask() {
+	if !common.IsMasterNode {
+		return
+	}
+	opsUserLogStatsTaskOnce.Do(func() {
+		go func() {
+			for {
+				if err := SyncOpsUserLogStats(); err != nil {
+					common.SysError("ops user log stats sync: " + err.Error())
+				}
+				time.Sleep(opsUserLogStatsSyncEvery)
+			}
+		}()
+	})
+}