goclaw/internal/http/usage_live_merge.go

package http

import (
	"net/http"
	"sort"
	"time"

	"github.com/nextlevelbuilder/goclaw/internal/store"
)

func (h *UsageHandler) aggregateTimeSeriesWithLive(r *http.Request, q store.SnapshotQuery, now time.Time) usageSummary {
	points, err := h.snapshots.GetTimeSeries(r.Context(), q)
	if err != nil {
		return usageSummary{}
	}
	if from, to, ok := liveUsageWindow(q, now); ok {
		livePoint, err := h.queryLiveTimeSeries(r, from, to, q)
		if err != nil {
			return summarizeTimeSeries(points)
		}
		if livePoint != nil {
			points = mergeSnapshotTimeSeries(points, *livePoint, q.GroupBy)
		}
	}
	return summarizeTimeSeries(points)
}

func summarizeTimeSeries(points []store.SnapshotTimeSeries) usageSummary {
	var s usageSummary
	var totalWeightedDuration int64
	for _, p := range points {
		s.Requests += p.RequestCount
		s.InputTokens += p.InputTokens
		s.OutputTokens += p.OutputTokens
		s.Cost += p.TotalCost
		s.UniqueUsers += p.UniqueUsers
		s.Errors += p.ErrorCount
		s.LLMCalls += p.LLMCallCount
		s.ToolCalls += p.ToolCallCount
		totalWeightedDuration += int64(p.AvgDurationMS) * int64(p.RequestCount)
	}
	if s.Requests > 0 {
		s.AvgDurationMS = int(totalWeightedDuration / int64(s.Requests))
	}
	return s
}

func mergeSnapshotTimeSeries(points []store.SnapshotTimeSeries, live store.SnapshotTimeSeries, groupBy string) []store.SnapshotTimeSeries {
	for i := range points {
		if points[i].BucketTime.Equal(live.BucketTime) {
			if groupBy == "day" {
				addPointToTimeSeries(&points[i], live)
			} else {
				points[i] = live
			}
			return points
		}
	}
	points = append(points, live)
	sort.Slice(points, func(i, j int) bool {
		return points[i].BucketTime.Before(points[j].BucketTime)
	})
	return points
}

func addPointToTimeSeries(dst *store.SnapshotTimeSeries, src store.SnapshotTimeSeries) {
	oldRequests := dst.RequestCount
	weightedDuration := int64(dst.AvgDurationMS)*int64(oldRequests) + int64(src.AvgDurationMS)*int64(src.RequestCount)
	dst.InputTokens += src.InputTokens
	dst.OutputTokens += src.OutputTokens
	dst.CacheReadTokens += src.CacheReadTokens
	dst.CacheCreateTokens += src.CacheCreateTokens
	dst.ThinkingTokens += src.ThinkingTokens
	dst.TotalCost += src.TotalCost
	dst.RequestCount += src.RequestCount
	dst.LLMCallCount += src.LLMCallCount
	dst.ToolCallCount += src.ToolCallCount
	dst.ErrorCount += src.ErrorCount
	dst.UniqueUsers += src.UniqueUsers
	dst.MemoryDocs += src.MemoryDocs
	dst.MemoryChunks += src.MemoryChunks
	dst.KGEntities += src.KGEntities
	dst.KGRelations += src.KGRelations
	if dst.RequestCount > 0 {
		dst.AvgDurationMS = int(weightedDuration / int64(dst.RequestCount))
	}
}

func mergeSnapshotBreakdowns(base, extra []store.SnapshotBreakdown, groupBy string) []store.SnapshotBreakdown {
	if len(extra) == 0 {
		return base
	}
	byKey := make(map[string]int, len(base)+len(extra))
	var result []store.SnapshotBreakdown
	add := func(row store.SnapshotBreakdown) {
		if row.Key == "" && groupBy != "agent" {
			return
		}
		if idx, ok := byKey[row.Key]; ok {
			addBreakdown(&result[idx], row)
			return
		}
		result = append(result, row)
		byKey[row.Key] = len(result) - 1
	}
	for _, row := range base {
		add(row)
	}
	for _, row := range extra {
		add(row)
	}

	sort.Slice(result, func(i, j int) bool {
		if groupBy == "channel" {
			return result[i].RequestCount > result[j].RequestCount
		}
		return result[i].InputTokens > result[j].InputTokens
	})
	return result
}

func addBreakdown(dst *store.SnapshotBreakdown, src store.SnapshotBreakdown) {
	oldRequests := dst.RequestCount
	weightedDuration := int64(dst.AvgDurationMS)*int64(oldRequests) + int64(src.AvgDurationMS)*int64(src.RequestCount)
	dst.InputTokens += src.InputTokens
	dst.OutputTokens += src.OutputTokens
	dst.CacheReadTokens += src.CacheReadTokens
	dst.CacheCreateTokens += src.CacheCreateTokens
	dst.TotalCost += src.TotalCost
	dst.RequestCount += src.RequestCount
	dst.LLMCallCount += src.LLMCallCount
	dst.ToolCallCount += src.ToolCallCount
	dst.ErrorCount += src.ErrorCount
	if dst.RequestCount > 0 {
		dst.AvgDurationMS = int(weightedDuration / int64(dst.RequestCount))
	}
}

func hasTimeSeriesUsage(point store.SnapshotTimeSeries) bool {
	return point.RequestCount > 0 ||
		point.LLMCallCount > 0 ||
		point.ToolCallCount > 0 ||
		point.InputTokens > 0 ||
		point.OutputTokens > 0 ||
		point.CacheReadTokens > 0 ||
		point.CacheCreateTokens > 0 ||
		point.ThinkingTokens > 0 ||
		point.TotalCost > 0 ||
		point.ErrorCount > 0
}