kenn-io--agentsview/internal/db/trends.go

package db

import (
	"context"
	"fmt"
	"sort"
	"strings"
	"time"
)

const (
	MaxTrendTerms        = 12
	MaxTrendTermVariants = 8
)

type TrendTermInput struct {
	Term     string   `json:"term"`
	Variants []string `json:"variants"`
	Matchers []string `json:"-"`
}

type TrendBucket struct {
	Date         string `json:"date"`
	MessageCount int    `json:"message_count"`
}

type TrendPoint struct {
	Date  string `json:"date"`
	Count int    `json:"count"`
}

type TrendSeries struct {
	Term     string       `json:"term"`
	Variants []string     `json:"variants"`
	Total    int          `json:"total"`
	Points   []TrendPoint `json:"points"`
}

type TrendsTermsResponse struct {
	Granularity  string        `json:"granularity"`
	From         string        `json:"from"`
	To           string        `json:"to"`
	MessageCount int           `json:"message_count"`
	Buckets      []TrendBucket `json:"buckets"`
	Series       []TrendSeries `json:"series"`
}

func (db *DB) GetTrendsTerms(
	ctx context.Context,
	f AnalyticsFilter,
	terms []TrendTermInput,
	granularity string,
) (TrendsTermsResponse, error) {
	if granularity == "" {
		granularity = "week"
	}
	loc := f.location()
	buckets := TrendBucketRange(f.From, f.To, granularity)
	bucketIndex := trendBucketIndex(buckets)
	counts := make([][]int, len(terms))
	for i := range counts {
		counts[i] = make([]int, len(buckets))
	}
	messageCounts := make([]int, len(buckets))

	sessionFilter := f
	sessionFilter.From = ""
	sessionFilter.To = ""
	sessionFilter.DayOfWeek = nil
	sessionFilter.Hour = nil
	sessionFilter.Model = ""
	where, args := sessionFilter.buildWhereWithDate("", false, "s.id")
	flt := f.messageScopeFilter()
	modelFiltering := len(flt.Models) > 0
	query := `SELECT m.session_id, m.ordinal, m.role, m.is_system,
			COALESCE(m.model, ''), m.content, COALESCE(m.timestamp, ''),
			COALESCE(s.started_at, ''), s.created_at
		FROM sessions s
		JOIN messages m ON m.session_id = s.id
		WHERE ` + where + `
			AND m.role IN ('user', 'assistant')
			AND m.is_system = 0
			AND ` + SystemPrefixSQL("m.content", "m.role") + `
		ORDER BY m.session_id, m.ordinal`

	rows, err := db.getReader().QueryContext(ctx, query, args...)
	if err != nil {
		return TrendsTermsResponse{}, fmt.Errorf("querying trends terms: %w", err)
	}
	defer rows.Close()

	type trendRow struct {
		sessionID string
		role      string
		isSystem  bool
		model     string
		content   string
		msgTS     string
		startedAt string
		createdAt string
	}
	processRow := func(row trendRow) {
		msgTime, ok := trendMessageLocalTime(row.msgTS, row.startedAt, row.createdAt, loc)
		if !ok {
			return
		}
		msgDate := msgTime.Format("2006-01-02")
		if !inDateRange(msgDate, f.From, f.To) {
			return
		}
		bucketDate := trendBucketDate(msgTime, loc, granularity)
		bucket, ok := bucketIndex[bucketDate]
		if !ok {
			return
		}
		messageCounts[bucket]++
		for i, term := range terms {
			count := countTrendOccurrences(row.content, term)
			if count > 0 {
				counts[i][bucket] += count
			}
		}
	}
	rowStartedAt := make(map[string]string)
	rowCreatedAt := make(map[string]string)
	emit := func(m ScopedMessage) {
		processRow(trendRow{
			sessionID: m.SessionID,
			role:      m.Role,
			isSystem:  m.IsSystem,
			content:   m.Content,
			msgTS:     m.Timestamp,
			startedAt: rowStartedAt[m.SessionID],
			createdAt: rowCreatedAt[m.SessionID],
		})
	}
	reducer := NewScopeReducer(flt, emit)

	for rows.Next() {
		var row trendRow
		var ordinal int
		if err := rows.Scan(
			&row.sessionID, &ordinal, &row.role, &row.isSystem,
			&row.model, &row.content, &row.msgTS, &row.startedAt,
			&row.createdAt,
		); err != nil {
			return TrendsTermsResponse{}, fmt.Errorf("scanning trends term row: %w", err)
		}
		if !modelFiltering {
			msgTime, ok := trendMessageLocalTime(row.msgTS, row.startedAt, row.createdAt, loc)
			if ok && flt.MatchesDayHour(msgTime, true) {
				processRow(row)
			}
			continue
		}
		rowStartedAt[row.sessionID] = row.startedAt
		rowCreatedAt[row.sessionID] = row.createdAt
		msgTime, has := trendMessageLocalTime(row.msgTS, row.startedAt, row.createdAt, loc)
		if err := reducer.Push(MessageInput{
			SessionID:    row.sessionID,
			Ordinal:      ordinal,
			Role:         row.role,
			Model:        row.model,
			IsSystem:     row.isSystem,
			Timestamp:    row.msgTS,
			LocalTime:    msgTime,
			HasLocalTime: has,
			Content:      row.content,
		}); err != nil {
			return TrendsTermsResponse{}, err
		}
	}
	if err := rows.Err(); err != nil {
		return TrendsTermsResponse{}, fmt.Errorf("iterating trends term rows: %w", err)
	}

	return BuildTrendsTermsResponse(
		f.From, f.To, granularity, buckets, terms, counts, messageCounts,
	), nil
}

func ParseTrendTerms(values []string) ([]TrendTermInput, error) {
	terms := make([]TrendTermInput, 0, min(len(values), MaxTrendTerms))
	for _, value := range values {
		variants := parseTrendTermVariants(value)
		if len(variants) == 0 {
			continue
		}
		if len(variants) > MaxTrendTermVariants {
			return nil, fmt.Errorf("trend terms can have at most %d variants", MaxTrendTermVariants)
		}
		matchers := make([]string, 0, len(variants)*3)
		for _, variant := range variants {
			matchers = append(matchers, variant)
			if plural, ok := simpleTrendPlural(variant); ok {
				matchers = append(matchers, plural)
			}
			matchers = append(matchers, simpleSilentEStemMatchers(variant)...)
		}
		matchers = dedupeCaseFolded(matchers)
		terms = append(terms, TrendTermInput{
			Term:     variants[0],
			Variants: variants,
			Matchers: matchers,
		})
	}
	if len(terms) == 0 {
		return nil, fmt.Errorf("at least one trend term is required")
	}
	if len(terms) > MaxTrendTerms {
		return nil, fmt.Errorf("trend query supports at most %d terms", MaxTrendTerms)
	}
	return terms, nil
}

func parseTrendTermVariants(value string) []string {
	parts := strings.Split(value, "|")
	variants := make([]string, 0, len(parts))
	seen := make(map[string]struct{}, len(parts))
	for _, part := range parts {
		variant := strings.TrimSpace(part)
		if variant == "" {
			continue
		}
		key := strings.ToLower(variant)
		if _, ok := seen[key]; ok {
			continue
		}
		seen[key] = struct{}{}
		variants = append(variants, variant)
	}
	return variants
}

func simpleTrendPlural(value string) (string, bool) {
	if value == "" || strings.ContainsAny(value, " \t\n\r") {
		return "", false
	}
	if strings.HasSuffix(strings.ToLower(value), "s") {
		return "", false
	}
	return value + "s", true
}

func simpleSilentEStemMatchers(value string) []string {
	if len(value) < 3 || !isSingleWordMatcher(value) {
		return nil
	}
	lower := strings.ToLower(value)
	if !strings.HasSuffix(lower, "c") ||
		strings.HasSuffix(lower, "e") ||
		strings.HasSuffix(lower, "s") {
		return nil
	}
	withE := value + "e"
	return []string{
		withE,
		withE + "s",
		withE + "d",
		value + "ing",
	}
}

func dedupeCaseFolded(values []string) []string {
	out := make([]string, 0, len(values))
	seen := make(map[string]struct{}, len(values))
	for _, value := range values {
		key := strings.ToLower(value)
		if _, ok := seen[key]; ok {
			continue
		}
		seen[key] = struct{}{}
		out = append(out, value)
	}
	return out
}

type matchSpan struct {
	start int
	end   int
}

func countTrendOccurrences(text string, term TrendTermInput) int {
	return CountTrendOccurrences(text, term)
}

func CountTrendOccurrences(text string, term TrendTermInput) int {
	spans := make([]matchSpan, 0)
	for _, matcher := range term.Matchers {
		matcher = strings.TrimSpace(matcher)
		if matcher == "" {
			continue
		}
		wordBounded := isSingleWordMatcher(matcher)
		spans = append(spans, collectTrendSpans(text, matcher, wordBounded)...)
	}
	return mergeCountSpans(spans)
}

func collectTrendSpans(text string, matcher string, wordBounded bool) []matchSpan {
	needle := strings.ToLower(matcher)
	haystack := strings.ToLower(text)
	if needle == "" || haystack == "" {
		return nil
	}
	spans := make([]matchSpan, 0)
	for offset := 0; offset < len(haystack); {
		idx := strings.Index(haystack[offset:], needle)
		if idx < 0 {
			break
		}
		start := offset + idx
		end := start + len(needle)
		if !wordBounded || hasWordBoundaries(haystack, start, end) {
			spans = append(spans, matchSpan{start: start, end: end})
		}
		offset = start + 1
	}
	return spans
}

func isSingleWordMatcher(matcher string) bool {
	if matcher == "" {
		return false
	}
	for i := 0; i < len(matcher); i++ {
		if !isWordByte(matcher[i]) {
			return false
		}
	}
	return true
}

func hasWordBoundaries(text string, start, end int) bool {
	if start > 0 && isWordByte(text[start-1]) {
		return false
	}
	if end < len(text) && isWordByte(text[end]) {
		return false
	}
	return true
}

func isWordByte(b byte) bool {
	return (b >= 'a' && b <= 'z') ||
		(b >= 'A' && b <= 'Z') ||
		(b >= '0' && b <= '9') ||
		b == '_'
}

func mergeCountSpans(spans []matchSpan) int {
	if len(spans) == 0 {
		return 0
	}
	sort.Slice(spans, func(i, j int) bool {
		if spans[i].start == spans[j].start {
			return spans[i].end < spans[j].end
		}
		return spans[i].start < spans[j].start
	})
	count := 0
	mergedEnd := -1
	for _, span := range spans {
		if span.start >= mergedEnd {
			count++
			mergedEnd = span.end
			continue
		}
		if span.end > mergedEnd {
			mergedEnd = span.end
		}
	}
	return count
}

func trendMessageLocalTime(
	messageTS string,
	startedAt string,
	createdAt string,
	loc *time.Location,
) (time.Time, bool) {
	for _, ts := range []string{messageTS, startedAt, createdAt} {
		if t, ok := localTime(ts, loc); ok {
			return t, true
		}
	}
	return time.Time{}, false
}

func trendBucketDate(t time.Time, loc *time.Location, granularity string) string {
	return TrendBucketDate(t, loc, granularity)
}

func TrendBucketDate(t time.Time, loc *time.Location, granularity string) string {
	local := t.In(loc)
	switch granularity {
	case "week":
		weekday := int(local.Weekday())
		if weekday == 0 {
			weekday = 7
		}
		start := local.AddDate(0, 0, -(weekday - 1))
		return time.Date(
			start.Year(), start.Month(), start.Day(),
			0, 0, 0, 0, loc,
		).Format("2006-01-02")
	case "month":
		return time.Date(
			local.Year(), local.Month(), 1,
			0, 0, 0, 0, loc,
		).Format("2006-01-02")
	default:
		return local.Format("2006-01-02")
	}
}

func TrendBucketRange(from, to, granularity string) []TrendBucket {
	if from == "" || to == "" {
		return nil
	}
	start, err := time.Parse("2006-01-02", from)
	if err != nil {
		return nil
	}
	end, err := time.Parse("2006-01-02", to)
	if err != nil {
		return nil
	}
	startDate := trendBucketDate(start, time.UTC, granularity)
	endDate := trendBucketDate(end, time.UTC, granularity)
	cur, err := time.Parse("2006-01-02", startDate)
	if err != nil {
		return nil
	}
	last, err := time.Parse("2006-01-02", endDate)
	if err != nil {
		return nil
	}
	buckets := make([]TrendBucket, 0)
	for !cur.After(last) {
		buckets = append(buckets, TrendBucket{Date: cur.Format("2006-01-02")})
		switch granularity {
		case "month":
			cur = cur.AddDate(0, 1, 0)
		case "week":
			cur = cur.AddDate(0, 0, 7)
		default:
			cur = cur.AddDate(0, 0, 1)
		}
	}
	return buckets
}

func trendBucketIndex(buckets []TrendBucket) map[string]int {
	index := make(map[string]int, len(buckets))
	for i, bucket := range buckets {
		index[bucket.Date] = i
	}
	return index
}

func BuildTrendsTermsResponse(
	from string,
	to string,
	granularity string,
	buckets []TrendBucket,
	terms []TrendTermInput,
	counts [][]int,
	messageCounts []int,
) TrendsTermsResponse {
	outBuckets := make([]TrendBucket, len(buckets))
	totalMessages := 0
	for i, bucket := range buckets {
		outBuckets[i] = bucket
		if i < len(messageCounts) {
			outBuckets[i].MessageCount = messageCounts[i]
			totalMessages += messageCounts[i]
		}
	}
	series := make([]TrendSeries, len(terms))
	for i, term := range terms {
		points := make([]TrendPoint, len(buckets))
		total := 0
		for j, bucket := range buckets {
			count := 0
			if i < len(counts) && j < len(counts[i]) {
				count = counts[i][j]
			}
			total += count
			points[j] = TrendPoint{
				Date:  bucket.Date,
				Count: count,
			}
		}
		series[i] = TrendSeries{
			Term:     term.Term,
			Variants: append([]string(nil), term.Variants...),
			Total:    total,
			Points:   points,
		}
	}
	return TrendsTermsResponse{
		Granularity:  granularity,
		From:         from,
		To:           to,
		MessageCount: totalMessages,
		Buckets:      outBuckets,
		Series:       series,
	}
}