diff --git a/model/channel_monitor_daily_rollup.go b/model/channel_monitor_daily_rollup.go index bb2a2b6d..571f4ce9 100644 --- a/model/channel_monitor_daily_rollup.go +++ b/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 +} diff --git a/model/channel_monitor_history.go b/model/channel_monitor_history.go index 01a993bd..cc66c8c5 100644 --- a/model/channel_monitor_history.go +++ b/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 +} diff --git a/service/channel_monitor_checker_test.go b/service/channel_monitor_checker_test.go index 4f7fd209..ba009424 100644 --- a/service/channel_monitor_checker_test.go +++ b/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 diff --git a/service/channel_monitor_cleanup.go b/service/channel_monitor_cleanup.go new file mode 100644 index 00000000..d81e403f --- /dev/null +++ b/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)) +} diff --git a/service/channel_monitor_maintenance.go b/service/channel_monitor_maintenance.go new file mode 100644 index 00000000..648835db --- /dev/null +++ b/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 +} diff --git a/service/channel_monitor_rollup.go b/service/channel_monitor_rollup.go new file mode 100644 index 00000000..f4a14330 --- /dev/null +++ b/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 +} diff --git a/service/channel_monitor_rollup_test.go b/service/channel_monitor_rollup_test.go new file mode 100644 index 00000000..23979048 --- /dev/null +++ b/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) +}