naco-api/internal/server/server.go

package server

import (
	"context"
	"encoding/json"
	"fmt"
	"maps"
	"net/http"
	"os"
	"strings"
	"time"

	"github.com/bluesky-social/indigo/api/bsky"
	"github.com/modelcontextprotocol/go-sdk/mcp"
	log "github.com/nacorid/logger"
	"github.com/nacorid/naco-api/internal/activitypub"
	"github.com/nacorid/naco-api/internal/bluesky"
	"github.com/nacorid/naco-api/internal/config"
	"github.com/nacorid/naco-api/internal/ollama"
	"github.com/nacorid/naco-api/internal/store"
	"github.com/nacorid/naco-api/internal/utils"
	"github.com/nacorid/naco-api/internal/x402"
	x402v2 "github.com/x402-foundation/x402/go/v2"
	x402http "github.com/x402-foundation/x402/go/v2/http/nethttp"
)

type Server struct {
	listenAddress  string
	permissions    os.FileMode
	mcpHandler     *x402.X402Server
	db             *store.Queries
	apClient       *activitypub.APClient
	bskyClient     *bluesky.BlueskyClient
	ollama         *ollama.OllamaClient
	x402Wallet     string
	activeNetworks []NetworkInfo
}

func New(ctx context.Context, db *store.Queries, ap *activitypub.APClient, bsky *bluesky.BlueskyClient, o *ollama.OllamaClient, cfg *config.Config) *Server {
	rateLimits := NewRateLimitMiddleware(float64(1.0/10.0), 2, []string{"getFeed"})
	metrics := NewMetricsHook()
	logging := NewLoggingMiddleware()
	toolChain := x402.X402Middlewares{
		MCPMiddleware: []mcp.Middleware{
			logging.OnCall,
			metrics.OnCall,
			rateLimits.OnCall,
		},
		HTTPMiddleware: []func(http.Handler) http.Handler{
			logging.OnCallHTTP,
			metrics.OnCallHTTP,
			rateLimits.OnCallHTTP,
		},
	}
	auth, err := NewCDPAuth(cfg.ApiKey, cfg.ApiSecret)
	if err != nil {
		log.FatalWithContext(ctx, "failed to create CDP auth", "error", err)
	}

	srv, err := x402.NewX402Server(
		"x402 enabled Bluesky lookingglass for autonomous Agents",
		"0.2.0",
		cfg.X402FacilitatorURL,
		auth,
		toolChain,
		GlobalLoggingMiddleware,
		func(w http.ResponseWriter, r *http.Request, resp *x402v2.SettleResponse) {
			requirements, ok := x402http.RequirementsFromContext(r.Context())
			if !ok {
				log.WarnWithContext(r.Context(), "failed to extract payment requirements from context during settlement hook")
				return
			}
			amount := requirements.Amount
			asset := requirements.Asset
			to := requirements.PayTo
			_ = settleHook(r.Context(), db, resp.Transaction, amount, asset, string(resp.Network), to, resp.Payer)
		},
	)
	if err != nil {
		log.FatalWithContext(ctx, "failed to create x402 server", "error", err)
	}

	srv.OnAfterSettle(func(settleCtx x402v2.SettleResultContext) error {
		amount := settleCtx.Requirements.GetAmount()
		asset := settleCtx.Requirements.GetAsset()
		to := settleCtx.Requirements.GetPayTo()
		from := settleCtx.Result.Payer

		return settleHook(settleCtx.Ctx, db, settleCtx.Result.Transaction, amount, asset, string(settleCtx.Result.Network), to, from)
	})

	enabledNetworkNames := parseActiveNetworks(cfg.ActiveNetworks)

	var activeNetworks []NetworkInfo
	for _, name := range enabledNetworkNames {
		if info, exists := NetworkRegistry[name]; exists {
			activeNetworks = append(activeNetworks, info)
		} else {
			log.WarnWithContext(ctx, "configured network is not defined in the registry", "name", name)
		}
	}

	s := &Server{
		listenAddress:  cfg.ListenAddress,
		permissions:    cfg.UnixSocketPermissions,
		mcpHandler:     srv,
		db:             db,
		apClient:       ap,
		bskyClient:     bsky,
		ollama:         o,
		x402Wallet:     cfg.X402WalletAddress,
		activeNetworks: activeNetworks,
	}

	s.registerTools()

	return s
}

func (s *Server) registerTools() {
	extra := map[string]any{
		"OutputSchema": Schema{
			Input: InputSchema{
				Type:         "HTTP",
				Method:       "POST",
				Discoverable: true,
				BodyFields: map[string]any{
					"actor": Bodyfield{
						Type:        "string",
						Description: "Did or handle of an actor",
						Required:    true,
					},
					"rkey": Bodyfield{
						Type:        "string",
						Description: "Record key of the post",
						Required:    true,
					},
					"collection": Bodyfield{
						Type:        "string",
						Description: "Collection the post belongs to. Defaults to 'app.bsky.feed.post'",
						Required:    false,
					},
				},
			},
			Output: map[string]any{
				"actor": Bodyfield{
					Type:        "string",
					Description: "Did of the actor",
				},
				"post": Bodyfield{
					Type:        "object",
					Description: "The post object. Includes the post content 'content', like 'likes', quote 'quotes', replies 'replies' and repost 'reposts' count",
				},
			},
		},
	}

	s.mcpHandler.AddPayableTool(
		mcp.Tool{Name: "getPost",
			Description:  "Get a specific post from a specific actor",
			InputSchema:  GetPostRequest{},
			OutputSchema: GetPostResponsePost{},
		},
		s.getPost,
		s.buildRequirements("0.01", extra)...,
	)

	extra = map[string]any{
		"OutputSchema": Schema{
			Input: InputSchema{
				Type:         "HTTP",
				Method:       "POST",
				Discoverable: true,
				BodyFields: map[string]any{
					"actor": Bodyfield{
						Type:        "string",
						Description: "Did or handle of an actor",
						Required:    true,
					},
					"n": Bodyfield{
						Type:        "integer",
						Description: "Number of posts to retrieve, maximum 50",
						Required:    true,
					},
				},
			},
			Output: map[string]any{
				"actor": Bodyfield{
					Type:        "string",
					Description: "Did of an actor",
				},
				"posts": Bodyfield{
					Type:        "array",
					Description: "Array of post objects. Each includes the post content 'content', like 'likes', quote 'quotes', replies 'replies' and repost 'reposts' count",
				},
			},
		},
	}

	s.mcpHandler.AddPayableTool(
		mcp.Tool{Name: "getLatestPosts",
			Description:  "Get the latest n posts from a specific actor",
			InputSchema:  GetLatestPostsRequest{},
			OutputSchema: GetLatestPostsResponse{},
		},
		s.getLatestPosts,
		s.buildRequirements("0.10", extra)...,
	)

	extra = map[string]any{
		"OutputSchema": Schema{
			Input: InputSchema{
				Type:         "HTTP",
				Method:       "POST",
				Discoverable: true,
				BodyFields: map[string]any{
					"actor": Bodyfield{
						Type:        "string",
						Description: "Did or handle of an actor",
						Required:    true,
					},
					"start": Bodyfield{
						Type:        "string",
						Description: "Start time in RFC3339 format",
						Required:    true,
					},
					"end": Bodyfield{
						Type:        "string",
						Description: "End time in RFC3339 format",
						Required:    true,
					},
					"n": Bodyfield{
						Type:        "integer",
						Description: "Maximum number of posts to retrieve, maximum 50",
						Required:    true,
					},
				},
			},
			Output: map[string]any{
				"actor": Bodyfield{
					Type:        "string",
					Description: "Did of an actor",
				},
				"posts": Bodyfield{
					Type:        "array",
					Description: "Array of post objects. Each includes the post content 'content', like 'likes', quote 'quotes', replies 'replies' and repost 'reposts' count",
				},
			},
		},
	}

	s.mcpHandler.AddPayableTool(
		mcp.Tool{Name: "getPostsInTime",
			Description:  "Get posts from a specific actor in a time range",
			InputSchema:  GetPostsInTimeRequest{},
			OutputSchema: GetPostsInTimeResponse{},
		},
		s.getPostsInTime,
		s.buildRequirements("0.10", extra)...,
	)

	extra = map[string]any{
		"OutputSchema": Schema{
			Input: InputSchema{
				Type:         "HTTP",
				Method:       "POST",
				Discoverable: true,
				BodyFields: map[string]any{
					"actor": Bodyfield{
						Type:        "string",
						Description: "Did or handle of an actor",
						Required:    true,
					},
				},
			},
			Output: map[string]any{
				"PersonaEmbedding": Bodyfield{
					Type:        "array",
					Description: "Embedding vector representing the actor's persona",
				},
				"TopKeywords": Bodyfield{
					Type:        "array",
					Description: "Top words used by the actor",
				},
				"StyleSummary": Bodyfield{
					Type:        "string",
					Description: "Summary of the actor's writing style",
				},
				"PostCount": Bodyfield{
					Type:        "integer",
					Description: "Number of posts analyzed",
				},
			},
		},
	}

	/*s.mcpHandler.AddPayableTool(
		mcp.Tool{Name: "getStyleProfile",
			Description:  "Get an actor's style Profile",
			InputSchema:  GetStyleProfileRequest{},
			OutputSchema: GetStyleProfileResponse{},
		},
		s.getStyleProfile,
		s.buildRequirements("100000", extra)...,
	)*/

	s.mcpHandler.AddTool(
		mcp.Tool{Name: "getFeed",
			Description:  "Get a feed by its URI",
			InputSchema:  GetFeedRequest{},
			OutputSchema: GetFeedResponse{},
		},
		s.getFeed,
	)
}

func (s *Server) getPost(ctx context.Context, request *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
	var args GetPostRequest
	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.RecordKey == "" {
		return nil, fmt.Errorf("rkey must be a valid record key")
	}
	res, err := s.bskyClient.GetPost(ctx, args.Author, args.Collection, args.RecordKey)
	if err != nil {
		return nil, err
	}
	post, ok := res.Record.Val.(*bsky.FeedPost)
	if !ok {
		return nil, fmt.Errorf("record is not a feed post")
	}
	resp := GetPostResponsePost{
		Author:    res.Author.Did,
		Content:   post.Text,
		Likes:     *res.LikeCount,
		Reposts:   *res.RepostCount,
		Quotes:    *res.QuoteCount,
		Replies:   *res.ReplyCount,
		CreatedAt: post.CreatedAt,
	}

	respBytes, err := json.Marshal(resp)
	if err != nil {
		return nil, err
	}

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

func (s *Server) getLatestPosts(ctx context.Context, request *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
	var args GetLatestPostsRequest
	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 {
		return nil, fmt.Errorf("n must be a positive integer")
	}
	if args.Limit > 50 {
		return nil, fmt.Errorf("n must be at most 50")
	}
	res, err := s.bskyClient.GetLatestNPosts(ctx, args.Author, args.Limit)
	if err != nil {
		return nil, err
	}
	posts := make([]GetPostResponsePost, 0, len(res))
	var author string
	for _, p := range res {
		post, ok := p.Post.Record.Val.(*bsky.FeedPost)
		if !ok {
			continue
		}
		posts = append(posts, GetPostResponsePost{
			Content:   post.Text,
			Likes:     *p.Post.LikeCount,
			Reposts:   *p.Post.RepostCount,
			Quotes:    *p.Post.QuoteCount,
			Replies:   *p.Post.ReplyCount,
			CreatedAt: post.CreatedAt,
		})
		author = p.Post.Author.Did
	}
	if author == "" {
		author = args.Author
	}
	resp := GetLatestPostsResponse{
		Author: author,
		Posts:  posts,
	}

	respBytes, err := json.Marshal(resp)
	if err != nil {
		return nil, err
	}

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

func (s *Server) getPostsInTime(ctx context.Context, request *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
	var args GetPostsInTimeRequest
	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 {
		return nil, fmt.Errorf("n must be a positive integer")
	}
	limit := min(args.Limit, 50)
	startTime, err := time.Parse(time.RFC3339, args.StartTime)
	if err != nil {
		return nil, fmt.Errorf("startTime must be in RFC3339 format")
	}
	endTime, err := time.Parse(time.RFC3339, args.EndTime)
	if err != nil {
		return nil, fmt.Errorf("startTime must be in RFC3339 format")
	}
	if startTime.After(endTime) {
		tmp := startTime
		startTime = endTime
		endTime = tmp
	}
	res, err := s.bskyClient.GetPostsInTimeframe(ctx, args.Author, startTime, endTime, limit)
	if err != nil {
		return nil, err
	}
	posts := make([]GetPostResponsePost, 0, len(res))
	var author string
	for _, p := range res {
		post, ok := p.Post.Record.Val.(*bsky.FeedPost)
		if !ok {
			continue
		}
		posts = append(posts, GetPostResponsePost{
			Content:   post.Text,
			Likes:     *p.Post.LikeCount,
			Reposts:   *p.Post.RepostCount,
			Quotes:    *p.Post.QuoteCount,
			Replies:   *p.Post.ReplyCount,
			CreatedAt: post.CreatedAt,
		})
		author = p.Post.Author.Did
	}
	if author == "" {
		author = args.Author
	}
	resp := &GetPostsInTimeResponse{
		Author: author,
		Posts:  posts,
	}

	respBytes, err := json.Marshal(resp)
	if err != nil {
		return nil, err
	}

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

func (s *Server) getFeed(ctx context.Context, request *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
	var args GetFeedRequest
	if err := json.Unmarshal(request.Params.Arguments, &args); err != nil {
		return nil, fmt.Errorf("invalid arguments: %w", err)
	}
	if args.URI == "" {
		return nil, fmt.Errorf("feedURI must be a valid feed URI")
	}
	res, err := s.bskyClient.GetFeed(ctx, args.URI, 50)
	if err != nil {
		return nil, err
	}

	posts := make([]GetPostResponsePost, 0, len(res.Feed))
	for _, p := range res.Feed {
		post, ok := p.Post.Record.Val.(*bsky.FeedPost)
		if !ok {
			continue
		}
		posts = append(posts, GetPostResponsePost{
			Author:    p.Post.Author.Did,
			Content:   post.Text,
			Likes:     *p.Post.LikeCount,
			Reposts:   *p.Post.RepostCount,
			Quotes:    *p.Post.QuoteCount,
			Replies:   *p.Post.ReplyCount,
			CreatedAt: post.CreatedAt,
		})
	}
	resp := GetFeedResponse{
		URI:   args.URI,
		Posts: posts,
	}

	respBytes, err := json.Marshal(resp)
	if err != nil {
		return nil, err
	}

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

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

	styleProfile, err := s.db.GetStyleProfile(ctx, store.GetStyleProfileParams{
		UserID:   args.Author,
		Platform: "bluesky",
	})
	isRecent := styleProfile.ModifiedAt.Valid && time.Since(styleProfile.ModifiedAt.Time) < 24*7*time.Hour
	if err == nil && isRecent {
		embedding, err := utils.ParseEmbedding(styleProfile.Embedding)
		if err != nil {
			return nil, err
		}
		resp := GetStyleProfileResponse{
			Author:           styleProfile.UserID,
			PersonaEmbedding: embedding,
			TopKeywords:      styleProfile.TopKeywords,
			PostCount:        int(styleProfile.PostCount),
			StyleSummary:     styleProfile.StyleSummary,
		}

		respBytes, err := json.Marshal(resp)
		if err != nil {
			return nil, err
		}

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

	platform, err := determinePlatform(args.Author)
	if err != nil {
		return nil, err
	}
	switch platform {
	case Bluesky:

	}
	res, err := s.bskyClient.GetLatestNPosts(ctx, args.Author, 50)
	if err != nil {
		return nil, err
	}
	posts := make([]string, 0, len(res))
	var author string
	for _, p := range res {
		post, ok := p.Post.Record.Val.(*bsky.FeedPost)
		if !ok {
			continue
		}
		posts = append(posts, post.Text)
		author = p.Post.Author.Did
	}

	embeddings, err := s.ollama.EmbedTexts(ctx, posts)
	if err != nil {
		return nil, err
	}
	personaEmbedding := utils.AverageVector(embeddings)
	topWords := utils.TopWords(posts, 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:       author,
		Platform:     "bluesky",
		Embedding:    personaEmbedding,
		TopKeywords:  topWords,
		StyleSummary: styleSummary,
		PostCount:    int32(len(posts)),
	})
	if err != nil {
		return nil, err
	}

	if author == "" {
		author = args.Author
	}
	profile := &GetStyleProfileResponse{
		Author:           author,
		PersonaEmbedding: personaEmbedding,
		TopKeywords:      topWords,
		StyleSummary:     styleSummary,
		PostCount:        len(posts),
		SourcePlatform:   "bluesky",
	}

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

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

func determinePlatform(author string) (platform, error) {
	if strings.HasPrefix(author, "did:") {
		return Bluesky, nil
	}
	if strings.HasPrefix(author, "npub") {
		return Nostr, nil
	}
	if strings.HasPrefix(author, "@") {
		if strings.Count(author, "@") > 1 {
			return ActivityPub, nil
		}

		if strings.Contains(author, ".") {
			return Bluesky, nil
		}

		return Twitter, nil
	}
	if len(author) == 64 {
		isHex := true
		for _, c := range author {
			if !((c >= '0' && c <= '9') || (c >= 'a' && c <= 'f') || (c >= 'A' && c <= 'F')) {
				isHex = false
				break
			}
		}
		if isHex {
			return Nostr, nil
		}
	}

	if strings.Contains(author, ".") {
		return Bluesky, nil
	}

	return -1, fmt.Errorf("unknown author format: %s", author)
}

func settleHook(ctx context.Context, db *store.Queries, transaction, amount, asset, network, to, from string) error {
	ret, err := db.SaveTransaction(ctx, store.SaveTransactionParams{
		X402TransactionHash: transaction,
		Amount:              amount,
		Asset:               asset,
		Network:             network,
		Recipient:           to,
		Sender:              from,
	})

	var id int64 = 0
	if len(ret) > 0 {
		id = ret[0].ID
	}

	fields := log.Fields{
		"transaction_id":        id,
		"X402_transaction_hash": transaction,
		"amount":                amount,
		"asset":                 asset,
		"network":               network,
		"recipient":             to,
		"sender":                from,
	}

	if err != nil {
		log.WithFields(fields).ErrorWithContext(ctx, "Failed to save transaction", "value", ret, "error", err)
		return err
	}

	log.WithFields(fields).DebugWithContext(ctx, "Saved transaction")
	return nil
}

func (s *Server) buildRequirements(amount string, extra map[string]any) []x402v2.PaymentRequirements {
	requirements := make([]x402v2.PaymentRequirements, 0, len(s.activeNetworks))

	for _, info := range s.activeNetworks {
		mergedExtra := make(map[string]any)
		maps.Copy(mergedExtra, extra)
		maps.Copy(mergedExtra, info.Extra)
		requirements = append(requirements, x402v2.PaymentRequirements{
			Scheme:            "exact",
			Network:           info.Network,
			Asset:             info.Asset,
			Amount:            amount,
			PayTo:             s.x402Wallet,
			MaxTimeoutSeconds: 60,
			Extra:             mergedExtra,
		})
	}

	return requirements
}

func parseActiveNetworks(raw string) []string {
	parts := strings.Split(raw, ",")
	var slugs []string
	for _, part := range parts {
		trimmed := strings.ToLower(strings.TrimSpace(part))
		if trimmed != "" {
			slugs = append(slugs, trimmed)
		}
	}
	return slugs
}

func (s *Server) Start() error {
	return s.mcpHandler.Start(s.listenAddress, s.permissions)
}