naco-api/internal/server/style.go

package server

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"sync"
	"time"

	"strings"

	"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"
)

func (s *Server) getStyleProfile(ctx context.Context, request *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
	var args GetStyleProfileRequest
	if err := json.Unmarshal(request.Params.Arguments, &args); err != nil {
		return nil, fmt.Errorf("invalid arguments: %w", err)
	}
	if args.Author == "" {
		return nil, fmt.Errorf("author must be a valid did or handle")
	}

	profile, err := s.requestStyleProfile(ctx, args.Author)
	if err != nil {
		return nil, err
	}

	// Best-effort: also look for this author on every other platform in the background so
	// future requests can be served from cache. Failures are logged and never affect the
	// response. Uses a detached context so the work outlives the request.
	if s.enrichEnabled {
		if src, ok := platformFromString(profile.SourcePlatform); ok {
			go s.enrichCrossPlatform(context.WithoutCancel(ctx), src, args.Author)
		}
	}

	profileBytes, err := json.Marshal(profile)
	if err != nil {
		return nil, err
	}

	return &mcp.CallToolResult{
		Content: []mcp.Content{
			&mcp.TextContent{Text: string(profileBytes)},
		},
	}, nil
}

// requestStyleProfile returns a style profile for author. It computes and persists the
// profile only when a request is made AND the cached profile is older than a month (or
// missing); otherwise it serves the cached copy. The prewarmer calls this same path, so a
// pre-fetch is treated as a system-initiated request and obeys the identical update policy.
func (s *Server) requestStyleProfile(ctx context.Context, author string) (*GetStyleProfileResponse, error) {
	platform, err := determinePlatform(author)
	if err != nil {
		return nil, err
	}
	return s.requestStyleProfileForced(ctx, platform, author)
}

// requestStyleProfileForced is the core of requestStyleProfile for an explicitly chosen
// platform. It resolves the author id, serves a fresh cached profile, or builds (and
// persists) a new one when the cache is stale/missing.
func (s *Server) requestStyleProfileForced(ctx context.Context, platform platform, author string) (*GetStyleProfileResponse, error) {
	var authorID string
	switch platform {
	case Bluesky:
		did, rerr := s.bskyClient.ResolveUser(ctx, author)
		if rerr != nil {
			return nil, rerr
		}
		authorID = did.String()
	case ActivityPub, Nostr, Twitter:
		authorID = author
	default:
		return nil, fmt.Errorf("platform %q is not supported yet", platform)
	}

	styleProfile, err := s.db.GetStyleProfile(ctx, store.GetStyleProfileParams{
		UserID:   authorID,
		Platform: platform.String(),
	})
	if err == nil {
		isRecent := styleProfile.ModifiedAt.Valid && time.Since(styleProfile.ModifiedAt.Time) < 24*30*time.Hour
		if isRecent {
			embedding, perr := utils.ParseEmbedding(styleProfile.Embedding)
			if perr != nil {
				return nil, perr
			}
			return &GetStyleProfileResponse{
				Author:           styleProfile.UserID,
				PersonaEmbedding: embedding,
				TopKeywords:      styleProfile.TopKeywords,
				PostCount:        int(styleProfile.PostCount),
				StyleSummary:     styleProfile.StyleSummary,
				SourcePlatform:   platform.String(),
			}, nil
		}
	} else if !errors.Is(err, sql.ErrNoRows) {
		return nil, fmt.Errorf("failed to load cached style profile: %w", err)
	}

	return s.buildStyleProfile(ctx, platform, authorID, author)
}

// enrichCrossPlatform looks for author on every platform other than sourcePlatform, in the
// background. It uses the same identifier on each platform (matching only where that handle
// exists; mismatches fail gracefully and are skipped) and obeys the same 1-month update gate
// as a live request. Concurrency is bounded to the other three platforms.
func (s *Server) enrichCrossPlatform(ctx context.Context, sourcePlatform platform, author string) {
	ctx, cancel := context.WithTimeout(ctx, 5*time.Minute)
	defer cancel()

	if s.metrics != nil {
		s.metrics.RecordEnrichment()
	}

	// Bound concurrent background enrichment so a burst of live requests can't overwhelm
	// Ollama. Block until a slot is free (or the context is cancelled).
	select {
	case s.enrichSem <- struct{}{}:
		defer func() { <-s.enrichSem }()
	case <-ctx.Done():
		return
	}

	var wg sync.WaitGroup
	for _, p := range []platform{Bluesky, ActivityPub, Nostr, Twitter} {
		if p == sourcePlatform {
			continue
		}
		wg.Add(1)
		go func(p platform) {
			defer wg.Done()
			if _, err := s.requestStyleProfileForced(ctx, p, author); err != nil {
				log.DebugWithContext(ctx, "cross-platform enrichment skipped", "platform", p.String(), "author", author, "error", err)
				return
			}
			log.InfoWithContext(ctx, "cross-platform enrichment done", "platform", p.String(), "author", author)
		}(p)
	}
	wg.Wait()
}

func (s *Server) getSimilarProfiles(ctx context.Context, request *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
	var args FindSimilarUsersRequest
	if err := json.Unmarshal(request.Params.Arguments, &args); err != nil {
		return nil, fmt.Errorf("invalid arguments: %w", err)
	}
	if args.Author == "" {
		return nil, fmt.Errorf("author must be a valid did or handle")
	}
	if args.Limit <= 0 {
		args.Limit = 10
	}
	if args.Limit > 50 {
		args.Limit = 50
	}

	profile, err := s.requestStyleProfile(ctx, args.Author)
	if err != nil {
		return nil, err
	}

	vec := utils.FormatVector(profile.PersonaEmbedding)
	rows, err := s.db.FindSimilarUsers(ctx, store.FindSimilarUsersParams{
		QueryEmbedding: vec,
		LimitVal:       args.Limit,
	})
	if err != nil {
		return nil, fmt.Errorf("failed to search similar users: %w", err)
	}

	matches := make([]SimilarUser, 0, len(rows))
	for _, r := range rows {
		if r.UserID == profile.Author {
			continue
		}
		matches = append(matches, SimilarUser{
			Author:       r.UserID,
			Platform:     r.Platform,
			Similarity:   float64(r.Similarity),
			TopKeywords:  r.TopKeywords,
			StyleSummary: r.StyleSummary,
			PostCount:    int(r.PostCount),
		})
	}

	twitterPostCount := 0
	if profile.SourcePlatform == "Twitter" {
		twitterPostCount = profile.PostCount
	}
	respBytes, err := json.Marshal(FindSimilarUsersResponse{
		Matches:          matches,
		TwitterPostCount: twitterPostCount,
	})
	if err != nil {
		return nil, err
	}
	return &mcp.CallToolResult{
		Content: []mcp.Content{
			&mcp.TextContent{Text: string(respBytes)},
		},
	}, nil
}

func (s *Server) buildStyleProfile(ctx context.Context, platform platform, authorID, author string) (*GetStyleProfileResponse, error) {
	posts, err := s.fetchProfilePosts(ctx, platform, authorID, 50)
	if err != nil {
		return nil, err
	}
	if len(posts) == 0 {
		return nil, fmt.Errorf("no posts found for author %s", author)
	}

	// Drop posts without any text (e.g. Bluesky image/reply-only posts). Embedding them
	// yields empty vectors, which would poison the persona-average and the persisted
	// profile.
	filtered := posts[:0]
	for _, p := range posts {
		if strings.TrimSpace(p.Text) == "" {
			continue
		}
		filtered = append(filtered, p)
	}
	if len(filtered) == 0 {
		return nil, fmt.Errorf("no posts found for author %s", author)
	}
	posts = filtered

	// Delete old posts for non-Twitter platforms before rebuild
	if platform != Twitter {
		if err := s.db.DeleteStylePosts(ctx, store.DeleteStylePostsParams{
			UserID:   authorID,
			Platform: platform.String(),
		}); err != nil {
			log.WarnWithContext(ctx, "failed to delete old style posts", "author", authorID, "platform", platform, "error", err)
		}
	}

	// Extract texts for embedding and keyword analysis
	postTexts := make([]string, len(posts))
	for i, p := range posts {
		postTexts[i] = p.Text
	}

	// Fail fast on any server-side error. These are payable tools: returning a degraded
	// profile here would bill the caller for a deliverable they didn't ask for. The x402
	// layer only settles payment when the handler succeeds, so returning the error means
	// the caller is not charged.
	embeddings, err := s.ollama.EmbedTexts(ctx, posts)
	if err != nil {
		return nil, err
	}
	personaEmbedding := utils.AverageVector(embeddings)
	topWords := utils.TopWords(postTexts, make(map[string]struct{}), 10)
	styleSummary, err := s.ollama.SummarizeStyle(ctx, posts)
	if err != nil {
		return nil, err
	}

	err = s.db.UpsertStyleProfile(ctx, store.UpsertStyleProfileParams{
		UserID:       authorID,
		Platform:     platform.String(),
		Embedding:    utils.FormatVector(personaEmbedding),
		TopKeywords:  topWords,
		StyleSummary: styleSummary,
		PostCount:    int32(len(posts)),
	})
	if err != nil {
		return nil, err
	}

	return &GetStyleProfileResponse{
		Author:           authorID,
		PersonaEmbedding: personaEmbedding,
		TopKeywords:      topWords,
		StyleSummary:     styleSummary,
		PostCount:        len(posts),
		SourcePlatform:   platform.String(),
	}, nil
}

func (s *Server) fetchProfilePosts(ctx context.Context, platform platform, author string, limit int) ([]clustering.Item, error) {
	platformStr := platform.String()

	// 1. Try cache first
	cached, err := s.db.GetStylePosts(ctx, store.GetStylePostsParams{
		UserID:   author,
		Platform: platformStr,
		Limit:    int32(limit),
	})
	if err == nil && len(cached) >= limit {
		posts := make([]clustering.Item, len(cached))
		for i, c := range cached {
			posts[i] = clustering.Item{ID: c.PostID, Text: c.PostText}
		}
		return posts, nil
	}

	// 2. Fetch from platform
	var freshPosts []clustering.Item
	switch platform {
	case Bluesky:
		res, err := s.bskyClient.GetLatestNPosts(ctx, author, limit)
		if err != nil {
			return toItems(cached), err
		}
		freshPosts = blueskyFeedViewsToItems(res)
	case ActivityPub:
		res, err := s.apClient.GetLatestNPosts(ctx, author, limit)
		if err != nil {
			return toItems(cached), err
		}
		freshPosts = apPostsPtrToItems(res)
	case Nostr:
		res, err := s.nostrClient.GetLatestNPosts(ctx, author, limit)
		if err != nil {
			return toItems(cached), err
		}
		freshPosts = nostrPostsToItems(res)
	case Twitter:
		res, err := s.twitterClient.GetLatestNPosts(ctx, author, limit)
		if err != nil {
			return toItems(cached), err
		}
		freshPosts = twitterPostsToItems(res)
	default:
		return nil, fmt.Errorf("platform %q is not supported yet", platform)
	}

	// 3. Upsert to cache (batch)
	for _, post := range freshPosts {
		if err := s.db.UpsertStylePosts(ctx, store.UpsertStylePostsParams{
			UserID:   author,
			Platform: platformStr,
			PostID:   post.ID,
			PostText: post.Text,
		}); err != nil {
			log.WarnWithContext(ctx, "failed to cache style post", "author", author, "platform", platform, "postID", post.ID, "error", err)
		}
	}

	return freshPosts, nil
}

func toItems(rows []store.GetStylePostsRow) []clustering.Item {
	items := make([]clustering.Item, len(rows))
	for i, r := range rows {
		items[i] = clustering.Item{ID: r.PostID, Text: r.PostText}
	}
	return items
}