naco-api/internal/server/analytics.go

package server

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"math"
	"sort"
	"strings"
	"time"

	"github.com/modelcontextprotocol/go-sdk/mcp"
	log "github.com/nacorid/logger"
	"github.com/nacorid/naco-api/internal/clustering"
	"github.com/nacorid/naco-api/internal/store"
	"github.com/nacorid/naco-api/internal/utils"
)

const (
	defaultAnalyticsLimit   = 50
	maxAnalyticsLimit       = 200
	defaultAnalyticsDays    = 30
	maxTopicCount           = 10
	minTopicCount           = 3
	sentimentBucketDuration = 24 * time.Hour
)

// ---- Analytics request/response helpers ----

func parseAnalyticsTimeRange(startStr, endStr string) (time.Time, time.Time, error) {
	now := time.Now().UTC()
	end := now
	start := now.AddDate(0, 0, -defaultAnalyticsDays)

	if endStr != "" {
		t, err := time.Parse(time.RFC3339, endStr)
		if err != nil {
			return time.Time{}, time.Time{}, fmt.Errorf("invalid endTime %q: %w", endStr, err)
		}
		end = t
	}
	if startStr != "" {
		t, err := time.Parse(time.RFC3339, startStr)
		if err != nil {
			return time.Time{}, time.Time{}, fmt.Errorf("invalid startTime %q: %w", startStr, err)
		}
		start = t
	}
	if start.After(end) {
		start, end = end, start
	}
	return start, end, nil
}

func hashTimeRange(start, end time.Time, limit int) string {
	data := fmt.Sprintf("%s|%s|%d", start.Format(time.RFC3339), end.Format(time.RFC3339), limit)
	h := sha256.Sum256([]byte(data))
	return hex.EncodeToString(h[:])
}

func (s *Server) getAnalyticsCache(ctx context.Context, author, platform, toolName, timeRangeHash string) (json.RawMessage, bool) {
	result, err := s.db.GetAnalyticsCache(ctx, store.GetAnalyticsCacheParams{
		UserID:        author,
		Platform:      platform,
		ToolName:      toolName,
		TimeRangeHash: timeRangeHash,
	})
	if err != nil {
		return nil, false
	}
	return result, true
}

func (s *Server) setAnalyticsCache(ctx context.Context, author, platform, toolName, timeRangeHash string, result any, ttl time.Duration) error {
	data, err := json.Marshal(result)
	if err != nil {
		return err
	}
	expiresAt := time.Now().Add(ttl)
	return s.db.UpsertAnalyticsCache(ctx, store.UpsertAnalyticsCacheParams{
		UserID:        author,
		Platform:      platform,
		ToolName:      toolName,
		TimeRangeHash: timeRangeHash,
		Result:        data,
		ExpiresAt:     expiresAt,
	})
}

// ---- Main analytics entry points ----

func (s *Server) getEngagementMetrics(ctx context.Context, request *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
	var args AnalyticsRequest
	if err := json.Unmarshal(request.Params.Arguments, &args); err != nil {
		return nil, fmt.Errorf("invalid arguments: %w", err)
	}
	return s.computeAnalytics(ctx, args, "engagementMetrics", s.computeEngagementMetrics)
}

func (s *Server) getPostingPatterns(ctx context.Context, request *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
	var args AnalyticsRequest
	if err := json.Unmarshal(request.Params.Arguments, &args); err != nil {
		return nil, fmt.Errorf("invalid arguments: %w", err)
	}
	return s.computeAnalytics(ctx, args, "postingPatterns", s.computePostingPatterns)
}

func (s *Server) getTopicModeling(ctx context.Context, request *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
	var args AnalyticsRequest
	if err := json.Unmarshal(request.Params.Arguments, &args); err != nil {
		return nil, fmt.Errorf("invalid arguments: %w", err)
	}
	return s.computeAnalytics(ctx, args, "topicModeling", s.computeTopicModeling)
}

func (s *Server) getSentimentTrends(ctx context.Context, request *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
	var args AnalyticsRequest
	if err := json.Unmarshal(request.Params.Arguments, &args); err != nil {
		return nil, fmt.Errorf("invalid arguments: %w", err)
	}
	return s.computeAnalytics(ctx, args, "sentimentTrends", s.computeSentimentTrends)
}

func (s *Server) getEngagementDrivers(ctx context.Context, request *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
	var args AnalyticsRequest
	if err := json.Unmarshal(request.Params.Arguments, &args); err != nil {
		return nil, fmt.Errorf("invalid arguments: %w", err)
	}
	return s.computeAnalytics(ctx, args, "engagementDrivers", s.computeEngagementDrivers)
}

type analyticsComputer func(ctx context.Context, args AnalyticsRequest, posts []clustering.Item) (any, error)

func (s *Server) computeAnalytics(ctx context.Context, args AnalyticsRequest, toolName string, compute analyticsComputer) (*mcp.CallToolResult, error) {
	if args.Author == "" {
		return nil, fmt.Errorf("author must be a valid did or handle")
	}

	// Normalize args
	limit := args.Limit
	if limit <= 0 {
		limit = defaultAnalyticsLimit
	}
	if limit > maxAnalyticsLimit {
		limit = maxAnalyticsLimit
	}

	start, end, err := parseAnalyticsTimeRange(args.StartTime, args.EndTime)
	if err != nil {
		return nil, fmt.Errorf("invalid time range: %w", err)
	}

	var platforms []string
	if args.Platform != "" {
		platforms = []string{args.Platform}
	} else {
		// Twitter is included by default only when the API is enabled: the API
		// costs it incurs are passed on to the caller through the upto pricing
		// of these tools (see registerTools), so aggregating over it does not
		// run the server at a loss. When no bearer token is configured there is
		// no Twitter to aggregate over.
		platforms = []string{"Bluesky", "ActivityPub", "Nostr"}
		if s.twitterEnabled {
			platforms = append(platforms, "Twitter")
		}
	}

	timeRangeHash := hashTimeRange(start, end, limit)

	// The analytics result cache is keyed by the scope of the computation: a
	// single platform when one is requested, or the "all" sentinel when the
	// result aggregates across all platforms. This keeps aggregate results from
	// being mixed with single-platform results.
	cachePlatform := "all"
	if args.Platform != "" {
		cachePlatform = args.Platform
	}

	if cached, ok := s.getAnalyticsCache(ctx, args.Author, cachePlatform, toolName, timeRangeHash); ok {
		var cachedResult json.RawMessage
		_ = json.Unmarshal(cached, &cachedResult)
		return s.toToolResult(cachedResult)
	}

	// Fetch posts from all platforms
	var allPosts []clustering.Item
	twitterPosts := 0
	for _, p := range platforms {
		plat, ok := platformFromString(p)
		if !ok {
			return nil, fmt.Errorf("unsupported platform %q", p)
		}
		posts, err := s.fetchAnalyticsPosts(ctx, plat, args.Author, start, end, limit)
		if err != nil {
			log.WarnWithContext(ctx, "failed to fetch posts for analytics", "platform", p, "error", err)
			continue
		}
		allPosts = append(allPosts, posts...)
		if plat == Twitter {
			twitterPosts += len(posts)
		}
	}

	if len(allPosts) == 0 {
		return nil, fmt.Errorf("no posts found for author %s in time range", args.Author)
	}

	// Compute
	result, err := compute(ctx, args, allPosts)
	if err != nil {
		return nil, err
	}
	result = setAnalyticsTwitterPostCount(result, twitterPosts)

	// Cache the result under the matching scope key.
	ttl := s.getCacheTTL(toolName)
	_ = s.setAnalyticsCache(ctx, args.Author, cachePlatform, toolName, timeRangeHash, result, ttl)

	return s.toToolResult(result)
}

// setAnalyticsTwitterPostCount records how many posts in an analytics result
// were sourced from the Twitter API, so upto-scheme settlement can pass the
// realized Twitter costs on to the caller. Unknown result types are returned
// unchanged.
func setAnalyticsTwitterPostCount(result any, count int) any {
	switch v := result.(type) {
	case EngagementMetricsResponse:
		v.TwitterPostCount = count
		return v
	case PostingPatternsResponse:
		v.TwitterPostCount = count
		return v
	case TopicModelingResponse:
		v.TwitterPostCount = count
		return v
	case SentimentTrendsResponse:
		v.TwitterPostCount = count
		return v
	case EngagementDriversResponse:
		v.TwitterPostCount = count
		return v
	}
	return result
}

func (s *Server) fetchAnalyticsPosts(ctx context.Context, platform platform, author string, start, end time.Time, limit int) ([]clustering.Item, error) {
	posts, err := s.fetchPostsInTimeForAnalytics(ctx, platform, author, start, end, limit)
	if err != nil {
		return nil, err
	}
	return posts, nil
}

func (s *Server) getCachedAnalyticsPosts(ctx context.Context, author, platform string, start, end time.Time, limit int) ([]clustering.Item, error) {
	rows, err := s.db.GetAnalyticsPostsInRange(ctx, store.GetAnalyticsPostsInRangeParams{
		UserID:          author,
		Platform:        platform,
		PostTimestamp:   start,
		PostTimestamp_2: end,
		Limit:           int32(limit),
	})
	if err != nil {
		return nil, err
	}
	items := make([]clustering.Item, 0, len(rows))
	for _, r := range rows {
		items = append(items, clustering.Item{
			ID:        r.PostID,
			Text:      r.PostText,
			Timestamp: r.PostTimestamp,
			Likes:     r.Likes,
			Reposts:   r.Reposts,
			Replies:   r.Replies,
			Quotes:    r.Quotes,
		})
	}
	return items, nil
}

func (s *Server) persistAnalyticsPosts(ctx context.Context, author, platform string, posts []clustering.Item) error {
	for _, p := range posts {
		if p.ID == "" {
			continue
		}
		if err := s.db.UpsertAnalyticsPosts(ctx, store.UpsertAnalyticsPostsParams{
			UserID:        author,
			Platform:      platform,
			PostID:        p.ID,
			PostText:      p.Text,
			PostTimestamp: p.Timestamp,
			Likes:         p.Likes,
			Reposts:       p.Reposts,
			Replies:       p.Replies,
			Quotes:        p.Quotes,
		}); err != nil {
			return err
		}
	}
	return nil
}

func (s *Server) fetchPostsInTimeForAnalytics(ctx context.Context, platform platform, author string, start, end time.Time, limit int) ([]clustering.Item, error) {
	var items []clustering.Item
	switch platform {
	case Bluesky:
		res, err := s.bskyClient.GetPostsInTimeframe(ctx, author, start, end, limit)
		if err != nil {
			return nil, err
		}
		items = blueskyFeedViewsToAnalyticsItems(res)
	case ActivityPub:
		res, err := s.apClient.GetPostsInTimeframe(ctx, author, start, end, limit)
		if err != nil {
			return nil, err
		}
		items = apPostsToAnalyticsItems(res)
	case Nostr:
		res, err := s.nostrClient.GetPostsInTimeframe(ctx, author, start, end, limit)
		if err != nil {
			return nil, err
		}
		items = nostrPostsToAnalyticsItems(res)
	case Twitter:
		// Twitter API calls are expensive (paid). Reuse persisted post data and
		// only call the API for time ranges that are not yet covered by the
		// cache, using a coverage watermark to guarantee broad coverage.
		fetched, err := s.fetchTwitterAnalyticsPosts(ctx, author, start, end, limit)
		if err != nil {
			return nil, err
		}
		items = append(items, fetched...)
	}
	return items, nil
}

// fetchTwitterAnalyticsPosts returns posts in [start, end] for the author,
// fetching from the Twitter API only the sub-ranges not already covered by the
// persistent post cache, and storing any newly fetched posts for future use.
//
// Coverage is tracked as a set of discrete intervals, so disjoint fetched
// ranges (e.g. [Dec 1, Dec 15] and [Dec 20, Dec 31]) are not merged into one
// continuous window. A later request that lands in the never-fetched gap
// (e.g. [Dec 15, Dec 20]) is therefore fetched from the API rather than being
// wrongly served from an incomplete cache.
func (s *Server) fetchTwitterAnalyticsPosts(ctx context.Context, author string, start, end time.Time, limit int) ([]clustering.Item, error) {
	covered, err := s.getAnalyticsCoverageRanges(ctx, author, "Twitter")
	if err != nil {
		return nil, err
	}

	// If the entire requested range falls within already-covered intervals,
	// serve from the cache without incurring a paid Twitter request.
	if len(uncoveredRanges(start, end, covered)) == 0 {
		return s.getCachedAnalyticsPosts(ctx, author, "Twitter", start, end, limit)
	}

	// Determine the sub-ranges of [start, end] not yet covered, and fetch each
	// from the API. Fetching per uncovered range keeps costs bounded while
	// expanding coverage on every call.
	ranges := uncoveredRanges(start, end, covered)

	var fetched []clustering.Item
	for _, r := range ranges {
		posts, err := s.fetchTwitterFromAPI(ctx, author, r.start, r.end, limit)
		if err != nil {
			return nil, err
		}
		fetched = append(fetched, posts...)
	}

	// Persist newly fetched posts permanently so repeat calls never re-pay.
	if len(fetched) > 0 {
		if err := s.persistAnalyticsPosts(ctx, author, "Twitter", fetched); err != nil {
			log.WarnWithContext(ctx, "failed to persist analytics posts", "platform", "Twitter", "error", err)
		}
	}

	// Record the newly fetched intervals as covered so future calls can reuse
	// them without hitting the paid API.
	for _, r := range ranges {
		if err := s.addAnalyticsCoverageRange(ctx, author, "Twitter", r.start, r.end); err != nil {
			log.WarnWithContext(ctx, "failed to add analytics coverage range", "platform", "Twitter", "range", r, "error", err)
		}
	}

	// Return the full set: posts from cache plus anything newly fetched, merged
	// and de-duplicated by post ID.
	return s.mergeAnalyticsPosts(ctx, author, start, end, limit, fetched)
}

type timeRange struct {
	start, end time.Time
}

// uncoveredRanges returns the sub-ranges of [start, end] that are not covered
// by any of the given intervals. Intervals are assumed sorted by their start,
// and no coverage (empty slice) yields the whole range as uncovered.
func uncoveredRanges(start, end time.Time, covered []timeRange) []timeRange {
	var ranges []timeRange
	cursor := start
	for _, c := range covered {
		if c.end.Before(cursor) || c.end.Equal(cursor) {
			continue
		}
		if c.start.After(end) {
			break
		}
		if c.start.After(cursor) {
			ranges = append(ranges, timeRange{start: cursor, end: c.start})
		}
		if c.end.After(cursor) {
			cursor = c.end
		}
	}
	if cursor.Before(end) {
		ranges = append(ranges, timeRange{start: cursor, end: end})
	}
	return ranges
}

func (s *Server) getAnalyticsCoverageRanges(ctx context.Context, author, platform string) ([]timeRange, error) {
	rows, err := s.db.GetAnalyticsCoverageRanges(ctx, store.GetAnalyticsCoverageRangesParams{
		UserID:   author,
		Platform: platform,
	})
	if err != nil {
		return nil, err
	}
	ranges := make([]timeRange, 0, len(rows))
	for _, r := range rows {
		ranges = append(ranges, timeRange{start: r.CoveredStart, end: r.CoveredEnd})
	}
	return ranges, nil
}

func (s *Server) addAnalyticsCoverageRange(ctx context.Context, author, platform string, start, end time.Time) error {
	return s.db.InsertAnalyticsCoverageRange(ctx, store.InsertAnalyticsCoverageRangeParams{
		UserID:       author,
		Platform:     platform,
		CoveredStart: start,
		CoveredEnd:   end,
	})
}

func (s *Server) fetchTwitterFromAPI(ctx context.Context, author string, start, end time.Time, limit int) ([]clustering.Item, error) {
	res, err := s.twitterClient.GetPostsInTimeframe(ctx, author, start, end, limit)
	if err != nil {
		return nil, err
	}
	return twitterPostsToAnalyticsItems(res), nil
}

// mergeAnalyticsPosts combines cached posts in [start, end] with newly fetched
// posts, de-duplicating by post ID and capping the result at limit. The cached
// query and each fetched sub-range individually honor limit, so the merged set
// can otherwise exceed it when the range is split into multiple fetch calls.
func (s *Server) mergeAnalyticsPosts(ctx context.Context, author string, start, end time.Time, limit int, fetched []clustering.Item) ([]clustering.Item, error) {
	merged := make([]clustering.Item, 0, len(fetched))
	seen := make(map[string]struct{}, len(fetched))
	for _, p := range fetched {
		if len(merged) >= limit {
			break
		}
		if _, ok := seen[p.ID]; ok {
			continue
		}
		seen[p.ID] = struct{}{}
		merged = append(merged, p)
	}

	cached, err := s.getCachedAnalyticsPosts(ctx, author, "Twitter", start, end, limit)
	if err != nil {
		return merged, nil
	}
	for _, p := range cached {
		if len(merged) >= limit {
			break
		}
		if _, ok := seen[p.ID]; ok {
			continue
		}
		seen[p.ID] = struct{}{}
		merged = append(merged, p)
	}
	return merged, nil
}

func (s *Server) toToolResult(result any) (*mcp.CallToolResult, error) {
	data, err := json.Marshal(result)
	if err != nil {
		return nil, err
	}
	return &mcp.CallToolResult{
		Content: []mcp.Content{
			&mcp.TextContent{Text: string(data)},
		},
	}, nil
}

func (s *Server) getCacheTTL(toolName string) time.Duration {
	switch toolName {
	case "postingPatterns":
		return s.analyticsTTLPatterns
	case "topicModeling":
		return s.analyticsTTLTopics
	case "sentimentTrends":
		return s.analyticsTTLSentiment
	case "engagementDrivers":
		return s.analyticsTTLDrivers
	case "engagementMetrics":
		fallthrough
	default:
		return s.analyticsTTLEngagement
	}
}

// ---- Computation implementations ----

func (s *Server) computeEngagementMetrics(ctx context.Context, args AnalyticsRequest, posts []clustering.Item) (any, error) {
	var metrics []EngagementMetricPoint
	for _, post := range posts {
		total := post.Likes + post.Reposts + post.Replies + post.Quotes
		// Rate is a simple relative measure (interactions per post). Without a
		// follower/fan base to normalize against, report total interactions.
		metrics = append(metrics, EngagementMetricPoint{
			PostID:         post.ID,
			Timestamp:      post.Timestamp.Format(time.RFC3339),
			Likes:          post.Likes,
			Reposts:        post.Reposts,
			Replies:        post.Replies,
			Quotes:         post.Quotes,
			EngagementRate: float64(total),
		})
	}

	return EngagementMetricsResponse{
		Author:  args.Author,
		Metrics: metrics,
	}, nil
}

func (s *Server) computePostingPatterns(ctx context.Context, args AnalyticsRequest, posts []clustering.Item) (any, error) {
	// Heatmap: 24 hours x 7 days
	heatmap := make([][]int, 24)
	for i := range heatmap {
		heatmap[i] = make([]int, 7)
	}
	dayCounts := make(map[int]int)
	hourCounts := make(map[int]int)
	dateSet := make(map[string]bool)

	for _, post := range posts {
		t := post.Timestamp
		if t.IsZero() {
			continue
		}
		hour := t.Hour()
		weekday := int(t.Weekday())
		heatmap[hour][weekday]++
		hourCounts[hour]++
		dayCounts[weekday]++
		dateSet[t.Format("2006-01-02")] = true
	}

	// Find peak hours (top 3)
	type hourCount struct {
		Hour  int
		Count int
	}
	var hours []hourCount
	for h, c := range hourCounts {
		hours = append(hours, hourCount{h, c})
	}
	sort.Slice(hours, func(i, j int) bool { return hours[i].Count > hours[j].Count })
	peakHours := make([]int, 0, 3)
	for i := 0; i < len(hours) && i < 3; i++ {
		peakHours = append(peakHours, hours[i].Hour)
	}

	// Find peak days (top 3)
	type dayCount struct {
		Day   int
		Count int
	}
	var days []dayCount
	for d, c := range dayCounts {
		days = append(days, dayCount{d, c})
	}
	sort.Slice(days, func(i, j int) bool { return days[i].Count > days[j].Count })
	peakDays := make([]int, 0, 3)
	for i := 0; i < len(days) && i < 3; i++ {
		peakDays = append(peakDays, days[i].Day)
	}

	totalPosts := len(posts)
	postingDays := len(dateSet)
	avgPostsPerDay := 0.0
	if postingDays > 0 {
		avgPostsPerDay = float64(totalPosts) / float64(postingDays)
	}

	return PostingPatternsResponse{
		Author:         args.Author,
		Heatmap:        heatmap,
		PeakHours:      peakHours,
		PeakDays:       peakDays,
		AvgPostsPerDay: avgPostsPerDay,
		TotalPosts:     totalPosts,
		PostingDays:    postingDays,
	}, nil
}

func (s *Server) computeTopicModeling(ctx context.Context, args AnalyticsRequest, posts []clustering.Item) (any, error) {
	topics, _, err := s.clusterPosts(ctx, posts)
	if err != nil {
		return nil, err
	}
	return TopicModelingResponse{
		Author: args.Author,
		Topics: topics,
	}, nil
}

// clusterPosts runs topic clustering over the posts and returns both the
// resulting topics and a map from post ID to the label of the topic the post
// belongs to. The membership map lets callers correlate topics with per-post
// attributes (e.g. engagement) without repeating the clustering.
func (s *Server) clusterPosts(ctx context.Context, posts []clustering.Item) ([]TopicCluster, map[string]string, error) {
	if len(posts) < minTopicCount {
		return []TopicCluster{}, map[string]string{}, nil
	}

	items := posts
	embeddings, err := s.ollama.EmbedTexts(ctx, items)
	if err != nil {
		return nil, nil, fmt.Errorf("failed to compute embeddings: %w", err)
	}

	for i := range items {
		items[i].Embedding = embeddings[i]
	}

	topicCount := maxTopicCount
	if len(items) < maxTopicCount {
		topicCount = len(items)
	}
	if topicCount < minTopicCount {
		topicCount = minTopicCount
	}

	results, err := clustering.KMeans(items, topicCount, 50, 42)
	if err != nil {
		return nil, nil, fmt.Errorf("kmeans failed: %w", err)
	}

	membership := make(map[string]string)
	labelByCluster := make(map[int]string)
	var topics []TopicCluster
	for _, cluster := range results {
		if len(cluster.Items) == 0 {
			continue
		}

		var samples []string
		for _, item := range cluster.Items {
			if len(samples) >= 3 {
				break
			}
			samples = append(samples, item.Text)
		}

		// Generate label using Ollama
		label := s.generateTopicLabel(ctx, samples)
		labelByCluster[cluster.Cluster] = label
		keywords := s.extractKeywords(cluster.Items)

		topics = append(topics, TopicCluster{
			TopicID:     cluster.Cluster,
			Label:       label,
			Keywords:    keywords,
			PostCount:   len(cluster.Items),
			Percentage:  float64(len(cluster.Items)) / float64(len(items)) * 100,
			SamplePosts: samples,
		})
	}

	// Map each post to the label of the cluster it belongs to.
	for _, cluster := range results {
		label := labelByCluster[cluster.Cluster]
		for _, item := range cluster.Items {
			if item.ID != "" {
				membership[item.ID] = label
			}
		}
	}

	// Sort by post count descending
	sort.Slice(topics, func(i, j int) bool { return topics[i].PostCount > topics[j].PostCount })

	return topics, membership, nil
}

func (s *Server) computeSentimentTrends(ctx context.Context, args AnalyticsRequest, posts []clustering.Item) (any, error) {
	// Bucket posts by day
	bucketMap := make(map[string][]string)
	for _, post := range posts {
		if post.Timestamp.IsZero() {
			continue
		}
		bucket := post.Timestamp.Truncate(sentimentBucketDuration).Format(time.RFC3339)
		bucketMap[bucket] = append(bucketMap[bucket], post.Text)
	}

	var timeSeries []SentimentPoint
	overallPos, overallNeg, overallNeu := 0.0, 0.0, 0.0
	emotionCounts := make(map[string]int)
	var shifts []SentimentShift
	var prevSentiment string
	var prevScores [3]float64

	// Only posts with a timestamp are bucketed and contribute to sentiment, so
	// the overall ratios are normalized by that same count (not the total
	// number of fetched posts, some of which may lack a timestamp).
	timestampedPosts := 0
	for _, post := range posts {
		if !post.Timestamp.IsZero() {
			timestampedPosts++
		}
	}

	// Process buckets in chronological order so sentiment shifts are detected correctly.
	buckets := make([]string, 0, len(bucketMap))
	for b := range bucketMap {
		buckets = append(buckets, b)
	}
	sort.Strings(buckets)

	for _, bucket := range buckets {
		texts := bucketMap[bucket]
		pos, neg, neu, emotions := s.analyzeSentiment(ctx, texts)
		timeSeries = append(timeSeries, SentimentPoint{
			Timestamp: bucket,
			Positive:  pos,
			Negative:  neg,
			Neutral:   neu,
			PostCount: len(texts),
		})

		overallPos += pos * float64(len(texts))
		overallNeg += neg * float64(len(texts))
		overallNeu += neu * float64(len(texts))

		for e, c := range emotions {
			emotionCounts[e] += c
		}

		// Detect sentiment shifts
		dominant := maxSentiment(pos, neg, neu)
		if prevSentiment != "" && dominant != prevSentiment {
			mag := math.Abs(pos-prevScores[0]) + math.Abs(neg-prevScores[1]) + math.Abs(neu-prevScores[2])
			shifts = append(shifts, SentimentShift{
				Timestamp:     bucket,
				FromSentiment: prevSentiment,
				ToSentiment:   dominant,
				Magnitude:     mag,
			})
		}
		prevSentiment = dominant
		prevScores = [3]float64{pos, neg, neu}
	}

	totalPosts := float64(timestampedPosts)
	if totalPosts > 0 {
		overallPos /= totalPosts
		overallNeg /= totalPosts
		overallNeu /= totalPosts
	}

	// Top emotions
	var topEmotions []EmotionScore
	type emoCount struct {
		Emotion string
		Count   int
	}
	var emos []emoCount
	for e, c := range emotionCounts {
		emos = append(emos, emoCount{e, c})
	}
	sort.Slice(emos, func(i, j int) bool { return emos[i].Count > emos[j].Count })
	for i := 0; i < len(emos) && i < 5; i++ {
		topEmotions = append(topEmotions, EmotionScore{
			Emotion: emos[i].Emotion,
			Score:   float64(emos[i].Count) / totalPosts,
		})
	}

	return SentimentTrendsResponse{
		Author:     args.Author,
		TimeSeries: timeSeries,
		OverallSentiment: SentimentSummary{
			Positive: overallPos,
			Negative: overallNeg,
			Neutral:  overallNeu,
		},
		DominantEmotions: topEmotions,
		SentimentShifts:  shifts,
	}, nil
}

func (s *Server) computeEngagementDrivers(ctx context.Context, args AnalyticsRequest, posts []clustering.Item) (any, error) {
	// Topic modeling first, getting post->topic membership for correlation.
	topics, membership, err := s.clusterPosts(ctx, posts)
	if err != nil {
		return nil, err
	}

	// Group engagement by topic label. Each post contributes its total
	// engagement (likes + reposts + replies + quotes) to its topic.
	topicEngagement := make(map[string][]float64)
	for _, post := range posts {
		label, ok := membership[post.ID]
		if !ok {
			continue
		}
		total := float64(post.Likes + post.Reposts + post.Replies + post.Quotes)
		topicEngagement[label] = append(topicEngagement[label], total)
	}

	// Baseline is the mean engagement across all posts, for comparison.
	var allEngagement []float64
	for _, post := range posts {
		allEngagement = append(allEngagement, float64(post.Likes+post.Reposts+post.Replies+post.Quotes))
	}
	baseline := mean(allEngagement)

	var drivers []EngagementDriver
	for _, topic := range topics {
		values := topicEngagement[topic.Label]
		topicMean := mean(values)
		correlation := 0.0
		if denom := topicMean + baseline; denom > 0 {
			correlation = (topicMean - baseline) / denom
			if correlation > 1 {
				correlation = 1
			} else if correlation < -1 {
				correlation = -1
			}
		}
		drivers = append(drivers, EngagementDriver{
			Topic:         topic.Label,
			AvgEngagement: topicMean,
			Correlation:   correlation,
			PostCount:     topic.PostCount,
			SamplePosts:   topic.SamplePosts[:min(2, len(topic.SamplePosts))],
		})
	}

	sort.Slice(drivers, func(i, j int) bool {
		return math.Abs(drivers[i].Correlation) > math.Abs(drivers[j].Correlation)
	})

	return EngagementDriversResponse{
		Author:  args.Author,
		Drivers: drivers,
	}, nil
}

func mean(values []float64) float64 {
	if len(values) == 0 {
		return 0
	}
	sum := 0.0
	for _, v := range values {
		sum += v
	}
	return sum / float64(len(values))
}

// ---- Helper functions ----

func (s *Server) generateTopicLabel(ctx context.Context, samples []string) string {
	if len(samples) == 0 {
		return "General"
	}
	prompt := fmt.Sprintf(`Given these sample posts, provide a concise 1-3 word topic label:

Posts:
%s

Label only (no explanation):`, strings.Join(samples, "\n---\n"))

	resp, err := s.ollama.SummarizeStyle(ctx, []clustering.Item{{Text: prompt}})
	if err != nil {
		return "General"
	}
	return strings.TrimSpace(resp)
}

func (s *Server) extractKeywords(items []clustering.Item) []string {
	var allText []string
	for _, item := range items {
		allText = append(allText, item.Text)
	}
	return utils.TopWords(allText, make(map[string]struct{}), 5)
}

func (s *Server) analyzeSentiment(ctx context.Context, texts []string) (pos, neg, neu float64, emotions map[string]int) {
	if len(texts) == 0 {
		return 0, 0, 0, nil
	}

	// Use Ollama for sentiment analysis
	prompt := fmt.Sprintf(`Analyze sentiment and emotions in these texts. Return JSON only:
{
  "positive": 0.0-1.0,
  "negative": 0.0-1.0,
  "neutral": 0.0-1.0,
  "emotions": {"emotion": count}
}

Texts:
%s`, strings.Join(texts, "\n---\n"))

	resp, err := s.ollama.SummarizeStyle(ctx, []clustering.Item{{Text: prompt}})
	if err != nil {
		return 0.33, 0.33, 0.34, nil
	}

	var result struct {
		Positive float64        `json:"positive"`
		Negative float64        `json:"negative"`
		Neutral  float64        `json:"neutral"`
		Emotions map[string]int `json:"emotions"`
	}
	_ = json.Unmarshal([]byte(resp), &result)

	// Normalize
	sum := result.Positive + result.Negative + result.Neutral
	if sum > 0 {
		result.Positive /= sum
		result.Negative /= sum
		result.Neutral /= sum
	}

	return result.Positive, result.Negative, result.Neutral, result.Emotions
}

func maxSentiment(pos, neg, neu float64) string {
	if pos >= neg && pos >= neu {
		return "positive"
	}
	if neg >= pos && neg >= neu {
		return "negative"
	}
	return "neutral"
}

func int64Val(p *int64) int64 {
	if p == nil {
		return 0
	}
	return *p
}