feat: add channel monitor rollups

Aggregate monitor history into daily rollups and add maintenance helpers for rollup refresh and history cleanup.

model/channel_monitor_daily_rollup.go

@@ -1,5 +1,7 @@
 package model
 
+import "gorm.io/gorm/clause"
+
 type ChannelMonitorDailyRollup struct {
 	Id            int    `json:"id" gorm:"primaryKey;autoIncrement"`
 	MonitorId     int    `json:"monitor_id" gorm:"not null;index;uniqueIndex:idx_channel_monitor_daily_rollup,priority:1"`
@@ -39,3 +41,24 @@ func GetChannelMonitorDailyRollups(monitorId int, modelName string, startDate st
 func GetMonitorDailyRollups(monitorId int, modelName string, startDate string, endDate string) ([]ChannelMonitorDailyRollup, error) {
 	return GetChannelMonitorDailyRollups(monitorId, modelName, startDate, endDate)
 }
+
+func UpsertChannelMonitorDailyRollups(rollups []ChannelMonitorDailyRollup) error {
+	if len(rollups) == 0 {
+		return nil
+	}
+	return DB.Clauses(clause.OnConflict{
+		Columns: []clause.Column{
+			{Name: "monitor_id"},
+			{Name: "channel_id"},
+			{Name: "model_name"},
+			{Name: "date"},
+		},
+		DoUpdates: clause.AssignmentColumns([]string{
+			"total_checks",
+			"pass_count",
+			"degraded_count",
+			"failed_count",
+			"avg_latency_ms",
+		}),
+	}).Create(&rollups).Error
+}

model/channel_monitor_history.go

@@ -59,3 +59,8 @@ func GetChannelMonitorHistories(monitorId int, page int, pageSize int, modelName
 func GetMonitorHistory(monitorId int, page int, pageSize int, modelName string, status string, startTime *time.Time, endTime *time.Time) ([]ChannelMonitorHistory, int64, error) {
 	return GetChannelMonitorHistories(monitorId, page, pageSize, modelName, status, startTime, endTime)
 }
+
+func DeleteChannelMonitorHistoriesBefore(cutoff time.Time) (int64, error) {
+	result := DB.Where("checked_at < ?", cutoff).Delete(&ChannelMonitorHistory{})
+	return result.RowsAffected, result.Error
+}

service/channel_monitor_checker_test.go

@@ -27,7 +27,7 @@ func setupChannelMonitorTestDB(t *testing.T) *gorm.DB {
 	dsn := fmt.Sprintf("file:%s?mode=memory&cache=shared", strings.ReplaceAll(t.Name(), "/", "_"))
 	db, err := gorm.Open(sqlite.Open(dsn), &gorm.Config{})
 	require.NoError(t, err)
-	require.NoError(t, db.AutoMigrate(&model.ChannelMonitor{}, &model.ChannelMonitorHistory{}))
+	require.NoError(t, db.AutoMigrate(&model.ChannelMonitor{}, &model.ChannelMonitorHistory{}, &model.ChannelMonitorDailyRollup{}))
 
 	model.DB = db
 	model.LOG_DB = db

service/channel_monitor_cleanup.go

@@ -0,0 +1,17 @@
+package service
+
+import (
+	"errors"
+	"time"
+
+	"github.com/QuantumNous/new-api/model"
+)
+
+const ChannelMonitorHistoryRetention = 7 * 24 * time.Hour
+
+func CleanOldChannelMonitorHistory(retention time.Duration, now time.Time) (int64, error) {
+	if retention <= 0 {
+		return 0, errors.New("history retention must be positive")
+	}
+	return model.DeleteChannelMonitorHistoriesBefore(now.Add(-retention))
+}

service/channel_monitor_maintenance.go

@@ -0,0 +1,97 @@
+package service
+
+import (
+	"errors"
+	"fmt"
+	"log"
+	"sync"
+	"time"
+)
+
+var (
+	channelMonitorMaintenanceOnce    sync.Once
+	channelMonitorMaintenanceStopMu  sync.Mutex
+	channelMonitorMaintenanceStopCh  chan struct{}
+	channelMonitorMaintenanceStopped bool
+
+	channelMonitorMaintenanceRunMu  sync.Mutex
+	channelMonitorMaintenanceActive bool
+)
+
+func StartChannelMonitorMaintenanceTask() {
+	channelMonitorMaintenanceOnce.Do(func() {
+		channelMonitorMaintenanceStopMu.Lock()
+		defer channelMonitorMaintenanceStopMu.Unlock()
+
+		channelMonitorMaintenanceStopCh = make(chan struct{})
+		channelMonitorMaintenanceStopped = false
+		go runChannelMonitorMaintenanceLoop(channelMonitorMaintenanceStopCh)
+	})
+}
+
+func StopChannelMonitorMaintenanceTask() {
+	channelMonitorMaintenanceStopMu.Lock()
+	defer channelMonitorMaintenanceStopMu.Unlock()
+
+	if channelMonitorMaintenanceStopCh != nil && !channelMonitorMaintenanceStopped {
+		close(channelMonitorMaintenanceStopCh)
+		channelMonitorMaintenanceStopped = true
+	}
+}
+
+func RunChannelMonitorMaintenanceOnce(now time.Time) error {
+	if !markChannelMonitorMaintenanceRunning() {
+		return nil
+	}
+	defer unmarkChannelMonitorMaintenanceRunning()
+
+	var errs []error
+	if _, err := AggregateChannelMonitorDailyRollupsForDate(now.AddDate(0, 0, -1)); err != nil {
+		errs = append(errs, fmt.Errorf("aggregate yesterday rollup: %w", err))
+	}
+	if _, err := AggregateChannelMonitorDailyRollupsForDate(now); err != nil {
+		errs = append(errs, fmt.Errorf("aggregate today rollup: %w", err))
+	}
+	if _, err := CleanOldChannelMonitorHistory(ChannelMonitorHistoryRetention, now); err != nil {
+		errs = append(errs, fmt.Errorf("clean old history: %w", err))
+	}
+	return errors.Join(errs...)
+}
+
+func runChannelMonitorMaintenanceLoop(stopCh <-chan struct{}) {
+	if err := RunChannelMonitorMaintenanceOnce(time.Now()); err != nil {
+		log.Printf("channel monitor maintenance: %v", err)
+	}
+
+	ticker := time.NewTicker(time.Hour)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case now := <-ticker.C:
+			if err := RunChannelMonitorMaintenanceOnce(now); err != nil {
+				log.Printf("channel monitor maintenance: %v", err)
+			}
+		case <-stopCh:
+			return
+		}
+	}
+}
+
+func markChannelMonitorMaintenanceRunning() bool {
+	channelMonitorMaintenanceRunMu.Lock()
+	defer channelMonitorMaintenanceRunMu.Unlock()
+
+	if channelMonitorMaintenanceActive {
+		return false
+	}
+	channelMonitorMaintenanceActive = true
+	return true
+}
+
+func unmarkChannelMonitorMaintenanceRunning() {
+	channelMonitorMaintenanceRunMu.Lock()
+	defer channelMonitorMaintenanceRunMu.Unlock()
+
+	channelMonitorMaintenanceActive = false
+}

service/channel_monitor_rollup.go

@@ -0,0 +1,114 @@
+package service
+
+import (
+	"errors"
+	"sort"
+	"strings"
+	"time"
+
+	"github.com/QuantumNous/new-api/model"
+)
+
+type channelMonitorRollupKey struct {
+	monitorID int
+	channelID int
+	modelName string
+	date      string
+}
+
+type channelMonitorRollupAccumulator struct {
+	totalChecks  int
+	passCount    int
+	degraded     int
+	failed       int
+	latencyTotal int
+}
+
+func AggregateChannelMonitorDailyRollupsForDate(day time.Time) (int, error) {
+	start := time.Date(day.Year(), day.Month(), day.Day(), 0, 0, 0, 0, day.Location())
+	return AggregateChannelMonitorDailyRollups(start, start.AddDate(0, 0, 1))
+}
+
+func AggregateChannelMonitorDailyRollups(start time.Time, end time.Time) (int, error) {
+	if !end.After(start) {
+		return 0, errors.New("rollup end time must be after start time")
+	}
+
+	var histories []model.ChannelMonitorHistory
+	if err := model.DB.
+		Select("monitor_id", "channel_id", "model_name", "status", "latency_ms", "checked_at").
+		Where("checked_at >= ? AND checked_at < ?", start, end).
+		Find(&histories).Error; err != nil {
+		return 0, err
+	}
+	if len(histories) == 0 {
+		return 0, nil
+	}
+
+	rollupsByKey := make(map[channelMonitorRollupKey]*channelMonitorRollupAccumulator)
+	for _, history := range histories {
+		key := channelMonitorRollupKey{
+			monitorID: history.MonitorId,
+			channelID: history.ChannelId,
+			modelName: history.ModelName,
+			date:      history.CheckedAt.In(start.Location()).Format("2006-01-02"),
+		}
+		acc := rollupsByKey[key]
+		if acc == nil {
+			acc = &channelMonitorRollupAccumulator{}
+			rollupsByKey[key] = acc
+		}
+		acc.totalChecks++
+		acc.latencyTotal += history.LatencyMs
+		switch strings.ToLower(history.Status) {
+		case "pass":
+			acc.passCount++
+		case "degraded":
+			acc.degraded++
+		case "failed", "error":
+			acc.failed++
+		}
+	}
+
+	keys := make([]channelMonitorRollupKey, 0, len(rollupsByKey))
+	for key := range rollupsByKey {
+		keys = append(keys, key)
+	}
+	sort.Slice(keys, func(i, j int) bool {
+		if keys[i].date != keys[j].date {
+			return keys[i].date < keys[j].date
+		}
+		if keys[i].monitorID != keys[j].monitorID {
+			return keys[i].monitorID < keys[j].monitorID
+		}
+		if keys[i].channelID != keys[j].channelID {
+			return keys[i].channelID < keys[j].channelID
+		}
+		return keys[i].modelName < keys[j].modelName
+	})
+
+	rollups := make([]model.ChannelMonitorDailyRollup, 0, len(keys))
+	for _, key := range keys {
+		acc := rollupsByKey[key]
+		avgLatency := 0
+		if acc.totalChecks > 0 {
+			avgLatency = acc.latencyTotal / acc.totalChecks
+		}
+		rollups = append(rollups, model.ChannelMonitorDailyRollup{
+			MonitorId:     key.monitorID,
+			ChannelId:     key.channelID,
+			ModelName:     key.modelName,
+			Date:          key.date,
+			TotalChecks:   acc.totalChecks,
+			PassCount:     acc.passCount,
+			DegradedCount: acc.degraded,
+			FailedCount:   acc.failed,
+			AvgLatencyMs:  avgLatency,
+		})
+	}
+
+	if err := model.UpsertChannelMonitorDailyRollups(rollups); err != nil {
+		return 0, err
+	}
+	return len(rollups), nil
+}

service/channel_monitor_rollup_test.go

@@ -0,0 +1,65 @@
+package service
+
+import (
+	"testing"
+	"time"
+
+	"github.com/QuantumNous/new-api/model"
+	"github.com/stretchr/testify/require"
+)
+
+func TestAggregateChannelMonitorDailyRollupsForDateUpsertsIdempotently(t *testing.T) {
+	db := setupChannelMonitorTestDB(t)
+	day := time.Date(2026, 5, 20, 10, 0, 0, 0, time.UTC)
+
+	histories := []model.ChannelMonitorHistory{
+		{MonitorId: 11, ChannelId: 7, ModelName: "gpt-test", Status: "pass", LatencyMs: 100, CheckedAt: day.Add(-9 * time.Hour)},
+		{MonitorId: 11, ChannelId: 7, ModelName: "gpt-test", Status: "degraded", LatencyMs: 300, CheckedAt: day.Add(-8 * time.Hour)},
+		{MonitorId: 11, ChannelId: 7, ModelName: "gpt-test", Status: "failed", LatencyMs: 500, CheckedAt: day.Add(-7 * time.Hour)},
+		{MonitorId: 12, ChannelId: 8, ModelName: "other-model", Status: "pass", LatencyMs: 900, CheckedAt: day.Add(-6 * time.Hour)},
+		{MonitorId: 11, ChannelId: 7, ModelName: "gpt-test", Status: "pass", LatencyMs: 999, CheckedAt: day.Add(15 * time.Hour)},
+	}
+	require.NoError(t, db.Create(&histories).Error)
+
+	written, err := AggregateChannelMonitorDailyRollupsForDate(day)
+	require.NoError(t, err)
+	require.Equal(t, 2, written)
+
+	written, err = AggregateChannelMonitorDailyRollupsForDate(day)
+	require.NoError(t, err)
+	require.Equal(t, 2, written)
+
+	var rollup model.ChannelMonitorDailyRollup
+	require.NoError(t, db.First(&rollup, "monitor_id = ? AND channel_id = ? AND model_name = ? AND date = ?", 11, 7, "gpt-test", "2026-05-20").Error)
+	require.Equal(t, 3, rollup.TotalChecks)
+	require.Equal(t, 1, rollup.PassCount)
+	require.Equal(t, 1, rollup.DegradedCount)
+	require.Equal(t, 1, rollup.FailedCount)
+	require.Equal(t, 300, rollup.AvgLatencyMs)
+
+	var count int64
+	require.NoError(t, db.Model(&model.ChannelMonitorDailyRollup{}).
+		Where("monitor_id = ? AND channel_id = ? AND model_name = ? AND date = ?", 11, 7, "gpt-test", "2026-05-20").
+		Count(&count).Error)
+	require.Equal(t, int64(1), count)
+}
+
+func TestCleanOldChannelMonitorHistoryDeletesRecordsBeforeRetention(t *testing.T) {
+	db := setupChannelMonitorTestDB(t)
+	now := time.Date(2026, 5, 20, 12, 0, 0, 0, time.UTC)
+
+	histories := []model.ChannelMonitorHistory{
+		{MonitorId: 1, ChannelId: 1, ModelName: "old", Status: "pass", CheckedAt: now.AddDate(0, 0, -8)},
+		{MonitorId: 1, ChannelId: 1, ModelName: "recent", Status: "pass", CheckedAt: now.AddDate(0, 0, -6)},
+	}
+	require.NoError(t, db.Create(&histories).Error)
+
+	deleted, err := CleanOldChannelMonitorHistory(7*24*time.Hour, now)
+	require.NoError(t, err)
+	require.Equal(t, int64(1), deleted)
+
+	var remaining []model.ChannelMonitorHistory
+	require.NoError(t, db.Order("id ASC").Find(&remaining).Error)
+	require.Len(t, remaining, 1)
+	require.Equal(t, "recent", remaining[0].ModelName)
+}