new-api/service/channel_monitor_rollup.go

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
}