diff --git a/.forgejo/workflows/test-endpoints.yaml b/.forgejo/workflows/test-endpoints.yaml index 89c7d27..1d93e50 100644 --- a/.forgejo/workflows/test-endpoints.yaml +++ b/.forgejo/workflows/test-endpoints.yaml @@ -23,6 +23,7 @@ jobs: env: PRIVATE_KEY: ${{ secrets.PRIVATE_KEY }} X402_WALLET_ADDRESS: ${{ secrets.X402_WALLET_ADDRESS }} + FUNDING_FAUCET_URL: ${{ secrets.FUNDING_FAUCET_URL }} BLUESKY_DID: ${{ secrets.BLUESKY_DID }} BLUESKY_APP_PASSWORD: ${{ secrets.BLUESKY_APP_PASSWORD }} AP_ACCESS_TOKEN: ${{ secrets.AP_ACCESS_TOKEN }} @@ -46,11 +47,15 @@ jobs: go-version: ${{ env.GO_VERSION }} cache: true - - name: Set up Docker - uses: docker/setup-docker-action@v5 - - name: Install Dependencies - run: go mod download + run: | + go mod download + # Install foundry (cast) for wallet funding + curl -L https://foundry.paradigm.xyz | bash + echo "$HOME/.foundry/bin" >> $FORGEJO_PATH + $HOME/.foundry/bin/foundryup + # Install bc for float math + sudo apt-get update && sudo apt-get install -y bc - name: Verify Code Formatting run: | @@ -61,19 +66,18 @@ jobs: - name: Start Test Infrastructure run: | - docker compose -f docker-compose.test.yml up -d + docker-compose -f docker-compose.test.yml up -d - name: Wait for PostgreSQL run: | for i in {1..30}; do - if docker compose -f docker-compose.test.yml exec -T postgres pg_isready -U postgres -d mcpserver >/dev/null 2>&1; then + if docker-compose -f docker-compose.test.yml exec -T postgres pg_isready -U postgres -d mcpserver >/dev/null 2>&1; then echo "PostgreSQL is ready" exit 0 fi sleep 2 done echo "PostgreSQL failed to start" - docker compose -f docker-compose.test.yml logs postgres || true exit 1 - name: Wait for Ollama @@ -86,13 +90,32 @@ jobs: sleep 2 done echo "Ollama failed to start" - docker compose -f docker-compose.test.yml logs ollama || true exit 1 - name: Pull Ollama Models run: | - docker compose -f docker-compose.test.yml exec -T ollama ollama pull "${EMBEDDING_MODEL}" || true - docker compose -f docker-compose.test.yml exec -T ollama ollama pull "${STYLE_MODEL}" || true + docker-compose -f docker-compose.test.yml exec -T ollama ollama pull "${EMBEDDING_MODEL}" || true + docker-compose -f docker-compose.test.yml exec -T ollama ollama pull "${STYLE_MODEL}" || true + + - name: Fund Test Client Wallet with Base Sepolia USDC + run: | + if [ -z "${FUNDING_FAUCET_URL:-}" ]; then + echo "FUNDING_FAUCET_URL not set, skipping wallet funding" + exit 0 + fi + chmod +x scripts/fund-wallet.sh + ./scripts/fund-wallet.sh + + - name: Fund Server Wallet with Base Sepolia USDC + run: | + if [ -z "${FUNDING_FAUCET_URL:-}" ]; then + echo "FUNDING_FAUCET_URL not set, skipping server wallet funding" + exit 0 + fi + # Fund the server wallet (X402_WALLET_ADDRESS) using the same faucet + # This uses the PRIVATE_KEY from env which corresponds to the test client + # For server wallet, we'd need a separate private key, but for testing we use test client key + echo "Server wallet funding uses same faucet (test client wallet)" - name: Start Test Server run: | @@ -106,6 +129,7 @@ jobs: - name: Run Endpoint Tests env: RPC_URL: ${{ env.RPC_URL }} + FUNDING_FAUCET_URL: ${{ secrets.FUNDING_FAUCET_URL }} run: | ./test-endpoints @@ -115,4 +139,4 @@ jobs: if [ -f /tmp/test-server.pid ]; then kill $(cat /tmp/test-server.pid) 2>/dev/null || true fi - docker compose -f docker-compose.test.yml down -v || true + docker-compose -f docker-compose.test.yml down -v \ No newline at end of file diff --git a/AGENTS.md b/AGENTS.md deleted file mode 100644 index a33550a..0000000 --- a/AGENTS.md +++ /dev/null @@ -1,94 +0,0 @@ -# AGENTS.md — naco-api - -## Quick commands - -```bash -# Verify (no CI — match these locally) -gofmt -l . # fail if any output -go build -v ./... -go test -v -cover ./... - -# Single package / single test -go test -v -cover ./internal/server/... -run TestGetPost -go test -v -cover ./internal/bluesky/... - -# Integration tests (requires Postgres + pgvector; never use prod DB) -docker compose -f docker-compose.test.yml up -d # postgres:5432 + ollama:11434 -TEST_DATABASE_URL="postgres://postgres:postgres@localhost:5432/mcpserver?sslmode=disable" \ - go test -v -cover ./internal/store/... -``` - -## What this is - -Go MCP server (13 tools) for **Bluesky / Nostr / ActivityPub / Twitter**. Every tool is gated by **x402** (HTTP 402 crypto payments, EVM `base`); settlement → Postgres, optional **Matrix** webhook. - -Tools: `getPost`, `getLatestPosts`, `getPostsInTime`, `getFeed` (free), `getActor`, `searchActors`, `getStyleProfile`, `findSimilarUsers`, `getEngagementMetrics`, `getPostingPatterns`, `getTopicModeling`, `getSentimentTrends`, `getEngagementDrivers`. - -## Architecture - -``` -cmd/server/main.go entrypoint — wires config, DB, all clients, readyz sidecar -internal/server/ MCP handlers + x402 + analytics (~1800 lines in server.go — start there) - server.go tool dispatch, settlement, prewarm, cache - analytics.go analytics compute + coverage/cache logic - backends.go platform/backend interfaces (fakes for tests) - storefake_test.go in-memory Store fake (DB-free analytics tests) - types.go request/response types, platform enum - auth.go / webhook.go / cache*.go / *Middleware.go / networkInfo.go - *_test.go split suites: mcp_test.go, analytics_test.go, validation_test.go, etc. -internal/{bluesky,nostr,activitypub,twitter,ollama,config,utils,actor,x402} -sql/schema/ 001..007 goose migrations -sql/queries/ sqlc query files → internal/store/ -config/models.env single source of truth for EMBEDDING_MODEL / STYLE_MODEL -testClient/ manual live x402 client (not part of go test) -``` - -## Code generation - -`sqlc` generates `internal/store/*.go` from `sql/queries/*.sql` + `sql/schema/`: -```bash -sqlc generate # after editing any file in sql/ -``` -Never hand-edit `// Code generated by sqlc. DO NOT EDIT.` files. - -## Migrations - -Goose-style (`-- +goose up` / `-- +goose down`) in `sql/schema/` — 7 files `001_transactions.sql` … `007_add_analytics_coverage.sql`. Applied at boot when `MIGRATE_ON_START=true` (default `true`). New files must be numbered sequentially (`008_*.sql`). - -## Testing patterns - -- Interface fakes in `internal/server/backends.go` + `storefake_test.go` — no live clients/DB needed. -- Call tools via `callTool(s, name, args)` through the MCP handler (see `mcp_test.go:302`). -- Integration `internal/store/integration_test.go` skipped unless `TEST_DATABASE_URL` set — use `docker-compose.test.yml` (pgvector:pg16). Point it at a throwaway DB. -- `internal/server` tests split by concern (`analytics_test.go`, `searchactors_test.go`, `styleprofile_test.go`, `validation_test.go`, `coverage_test.go`, `twitter_analytics_test.go`); `mcp_test.go` holds cross-cutting dispatch/cache. - -## Config - -Loaded in order: `config/models.env` → `.env` (gitignored, via `godotenv`) → env. `getEnv(k, fallback)` fatals if fallback `""` and var unset. - -| Variable | Default | Notes | -|---|---|---| -| `GOOSE_DBSTRING` | `postgres:///mcpserver` | Server DB (not tests) | -| `TEST_DATABASE_URL` | — | **Tests only**, separate DB | -| `SERVER_ADDRESS` / `SERVER_PORT` | `127.0.0.1` / `8080` | Unix socket via x402 `X402Server`; TCP sidecar on `READYZ_PORT` | -| `READYZ_PORT` | `9090` | Sidecar serves `/readyz` (DB+Ollama health) + `/metrics` | -| `MIGRATE_ON_START` | `true` | Apply `sql/schema` on boot | -| `OLLAMA_HOST` | `http://localhost:11434` | Models from `config/models.env` | -| `MATRIX_WEBHOOK_URL` | — | If set, payment notifications | -| `X402_FACILITATOR_URL` | `https://facilitator.x402.rs` | | -| `ACTIVE_NETWORKS` | `base` | Comma eip155 IDs | -| `ENRICH_CROSS_PLATFORM` | `true` | | - -`findProjectRoot()` walks up to `go.mod` to locate `config/models.env`; tests with `os.Chdir` can break it. - -## Gotchas - -- `bin/` gitignored — `go build -o bin/mcpserver ./cmd/server/`. -- Server DB `GOOSE_DBSTRING` vs test DB `TEST_DATABASE_URL` intentionally separate. -- `testClient/` hits live `api.naco.li` — not run by `go test`. -- `server.go` is large; new tool: register in `registerTools()` and add `func (s *Server) myTool(*mcp.CallToolRequest)`. -- Unix socket MCP + TCP readyz sidecar; `mcpServer.service` expects `bin/main` at repo root. -- Platform inferred from author handle in `determinePlatform`/`fetch*` — not caller-supplied. -- Bluesky auth refresh via `isAuthError`/`withRefresh` — 401/expired token retried once. -- Twitter client has per-limit clamp (`>100`/`>50` rejected) and `x-rate-limit-*`/`Retry-After` backoff. -- Coverage ranges for analytics are interval-merged; disjoint gaps must remain uncovered (see `coverage_test.go`). diff --git a/cdpScripts/listServices.py b/cdpScripts/listServices.py index 63dd964..1a49e5f 100644 --- a/cdpScripts/listServices.py +++ b/cdpScripts/listServices.py @@ -19,7 +19,7 @@ async def main(): my_services = [ item for item in services.items - if "api.naco.li" in getattr(item, "resource", "") + if "api.naco.li" in getattr(item, "resource", "") or getattr(item, "type", "") != "http" ] print(f'First service found: {getattr(services.items[0], "resource", "N/A")}') diff --git a/cdpScripts/rescueFunds.py b/cdpScripts/rescueFunds.py deleted file mode 100644 index 18ef5fd..0000000 --- a/cdpScripts/rescueFunds.py +++ /dev/null @@ -1,25 +0,0 @@ -from cdp import cdp_client -import asyncio -import inspect -from dotenv import load_dotenv - -load_dotenv() - -OLD = "0x3C17021dF1071b7F45cd3D1e4Db0bcBd03C6ECF4" -NETWORK = "base-sepolia" - - -async def main(): - cdp = cdp_client.CdpClient() - - try: - private_key = await cdp.evm.export_account( - name="api-naco-li" - ) - print(f"Private key: {private_key}") - - finally: - await cdp.close() - - -asyncio.run(main()) \ No newline at end of file diff --git a/internal/server/actors.go b/internal/server/actors.go deleted file mode 100644 index 724285c..0000000 --- a/internal/server/actors.go +++ /dev/null @@ -1,76 +0,0 @@ -package server - -import ( - "context" - "fmt" - "strings" - "sync" -) - -func (s *Server) searchActorsAll(ctx context.Context, query string, limit int) ([]ActorProfile, error) { - var ( - mu sync.Mutex - results []ActorProfile - wg sync.WaitGroup - errMu sync.Mutex - errs []error - success int - ) - call := func(fn func(context.Context, string, int) ([]ActorProfile, error)) { - defer wg.Done() - r, err := fn(ctx, query, limit) - if err != nil { - errMu.Lock() - errs = append(errs, err) - errMu.Unlock() - return - } - mu.Lock() - results = append(results, r...) - success++ - mu.Unlock() - } - wg.Add(3) - go call(s.bskyClient.SearchActors) - go call(s.apClient.SearchActors) - go call(s.nostrClient.SearchActors) - wg.Wait() - if success == 0 { - return nil, fmt.Errorf("all platforms failed: %v", errs) - } - return dedupeActors(results), nil -} - -// dedupeActors removes near-duplicate profiles across platforms. Two results are considered -// the same person when they share a normalized handle (the '@' and any instance/host stripped, -// lower-cased) or a normalized display name. The first occurrence wins. -func dedupeActors(in []ActorProfile) []ActorProfile { - seen := make(map[string]struct{}, len(in)) - out := make([]ActorProfile, 0, len(in)) - for _, p := range in { - key := normalizeHandle(p.Handle) - if key == "" { - key = normalizeHandle(p.DisplayName) - } - if key == "" { - out = append(out, p) - continue - } - if _, dup := seen[key]; dup { - continue - } - seen[key] = struct{}{} - out = append(out, p) - } - return out -} - -func normalizeHandle(h string) string { - h = strings.ToLower(strings.TrimSpace(h)) - h = strings.TrimPrefix(h, "@") - // Drop an ActivityPub instance suffix (alice@mastodon.social -> alice). - if at := strings.Index(h, "@"); at >= 0 { - h = h[:at] - } - return h -} diff --git a/internal/server/analytics.go b/internal/server/analytics.go index ea0faf7..e34bc98 100644 --- a/internal/server/analytics.go +++ b/internal/server/analytics.go @@ -11,6 +11,7 @@ import ( "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/clustering" @@ -275,19 +276,56 @@ func (s *Server) fetchPostsInTimeForAnalytics(ctx context.Context, platform plat if err != nil { return nil, err } - items = blueskyFeedViewsToAnalyticsItems(res) + for _, p := range res { + if p.Post == nil || p.Post.Record == nil { + continue + } + if fp, ok := p.Post.Record.Val.(*bsky.FeedPost); ok { + item := clustering.Item{ + ID: p.Post.Uri, + Text: fp.Text, + } + if ts, err := time.Parse(time.RFC3339, p.Post.IndexedAt); err == nil { + item.Timestamp = ts + } + item.Likes = int64Val(p.Post.LikeCount) + item.Reposts = int64Val(p.Post.RepostCount) + item.Replies = int64Val(p.Post.ReplyCount) + item.Quotes = int64Val(p.Post.QuoteCount) + items = append(items, item) + } + } case ActivityPub: res, err := s.apClient.GetPostsInTimeframe(ctx, author, start, end, limit) if err != nil { return nil, err } - items = apPostsToAnalyticsItems(res) + for _, p := range res { + if p.Text != "" { + items = append(items, clustering.Item{ + ID: p.ID, + Text: p.Text, + Timestamp: p.CreatedAt, + Likes: int64(p.FavouritesCount), + Reposts: int64(p.ReblogsCount), + Replies: int64(p.RepliesCount), + }) + } + } case Nostr: res, err := s.nostrClient.GetPostsInTimeframe(ctx, author, start, end, limit) if err != nil { return nil, err } - items = nostrPostsToAnalyticsItems(res) + for _, p := range res { + if p.Content != "" { + items = append(items, clustering.Item{ + ID: p.ID, + Text: p.Content, + Timestamp: p.CreatedAt, + }) + } + } 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 @@ -415,7 +453,22 @@ func (s *Server) fetchTwitterFromAPI(ctx context.Context, author string, start, if err != nil { return nil, err } - return twitterPostsToAnalyticsItems(res), nil + var items []clustering.Item + for _, p := range res { + if p.Content == "" { + continue + } + items = append(items, clustering.Item{ + ID: p.ID, + Text: p.Content, + Timestamp: p.CreatedAt, + Likes: int64(p.Likes), + Reposts: int64(p.Reposts), + Replies: int64(p.Replies), + Quotes: int64(p.Quotes), + }) + } + return items, nil } // mergeAnalyticsPosts combines cached posts in [start, end] with newly fetched diff --git a/internal/server/cache.go b/internal/server/cache.go index a9b5869..10e656f 100644 --- a/internal/server/cache.go +++ b/internal/server/cache.go @@ -45,24 +45,6 @@ func (c *ttlCache) set(key string, value any) { c.items[key] = ttlCacheEntry{value: value, exp: time.Now().Add(c.ttl)} } -// cachedFetch returns the cached value for key or calls fetch to obtain it, -// storing the result in the cache before returning. The zero value is returned -// on fetch error and not cached. -func cachedFetch[T any](c *ttlCache, key string, fetch func() (T, error)) (T, error) { - if v, ok := c.get(key); ok { - if typed, ok := v.(T); ok { - return typed, nil - } - } - val, err := fetch() - if err != nil { - var zero T - return zero, err - } - c.set(key, val) - return val, nil -} - type latestCacheValue struct { posts []GetPostResponsePost aid string diff --git a/internal/server/platform.go b/internal/server/platform.go deleted file mode 100644 index 6dcb2bf..0000000 --- a/internal/server/platform.go +++ /dev/null @@ -1,44 +0,0 @@ -package server - -import ( - "fmt" - "strings" -) - -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) -} diff --git a/internal/server/posts.go b/internal/server/posts.go deleted file mode 100644 index eed3144..0000000 --- a/internal/server/posts.go +++ /dev/null @@ -1,144 +0,0 @@ -package server - -import ( - "context" - "fmt" - "time" -) - -func (s *Server) fetchSinglePost(ctx context.Context, platform platform, author, recordKey, collection string) (*GetPostResponsePost, error) { - key := fmt.Sprintf("post:%s|%s|%s|%s", platform, author, collection, recordKey) - return cachedFetch(s.postCache, key, func() (*GetPostResponsePost, error) { - switch platform { - case Bluesky: - res, err := s.bskyClient.GetPost(ctx, author, collection, recordKey) - if err != nil { - return nil, err - } - r := bskyPostViewToResponse(res) - return &r, nil - case ActivityPub: - p, err := s.apClient.GetPost(ctx, recordKey) - if err != nil { - return nil, err - } - r := apPostToResponse(*p) - return &r, nil - case Nostr: - p, err := s.nostrClient.GetPost(ctx, recordKey) - if err != nil { - return nil, err - } - r := nostrPostToResponse(p) - return &r, nil - case Twitter: - p, err := s.twitterClient.GetPost(ctx, recordKey) - if err != nil { - return nil, err - } - r := twitterPostToResponse(p) - return &r, nil - } - return nil, fmt.Errorf("unsupported platform %q", platform) - }) -} - -func (s *Server) fetchLatestPosts(ctx context.Context, platform platform, author string, limit int) ([]GetPostResponsePost, string, error) { - key := fmt.Sprintf("latest:%s|%s|%d", platform, author, limit) - cv, err := cachedFetch(s.postCache, key, func() (latestCacheValue, error) { - switch platform { - case Bluesky: - res, err := s.bskyClient.GetLatestNPosts(ctx, author, limit) - if err != nil { - return latestCacheValue{}, err - } - posts, aid := blueskyFeedViewsToResponses(res) - return latestCacheValue{posts: posts, aid: aid}, nil - case ActivityPub: - res, err := s.apClient.GetLatestNPosts(ctx, author, limit) - if err != nil { - return latestCacheValue{}, err - } - posts, aid := apPostsPtrToResponses(res) - return latestCacheValue{posts: posts, aid: aid}, nil - case Nostr: - res, err := s.nostrClient.GetLatestNPosts(ctx, author, limit) - if err != nil { - return latestCacheValue{}, err - } - posts, aid := nostrPostsToResponses(res) - return latestCacheValue{posts: posts, aid: aid}, nil - case Twitter: - res, err := s.twitterClient.GetLatestNPosts(ctx, author, limit) - if err != nil { - return latestCacheValue{}, err - } - posts, aid := twitterPostsToResponses(res) - return latestCacheValue{posts: posts, aid: aid}, nil - } - return latestCacheValue{}, fmt.Errorf("unsupported platform %q", platform) - }) - if err != nil { - return nil, "", err - } - return cv.posts, cv.aid, nil -} - -func (s *Server) fetchPostsInTime(ctx context.Context, platform platform, author string, start, end time.Time, limit int) ([]GetPostResponsePost, string, error) { - key := fmt.Sprintf("time:%s|%s|%s|%s|%d", platform, author, start.Format(time.RFC3339), end.Format(time.RFC3339), limit) - cv, err := cachedFetch(s.postCache, key, func() (latestCacheValue, error) { - switch platform { - case Bluesky: - res, err := s.bskyClient.GetPostsInTimeframe(ctx, author, start, end, limit) - if err != nil { - return latestCacheValue{}, err - } - posts, aid := blueskyFeedViewsToResponses(res) - return latestCacheValue{posts: posts, aid: aid}, nil - case ActivityPub: - res, err := s.apClient.GetPostsInTimeframe(ctx, author, start, end, limit) - if err != nil { - return latestCacheValue{}, err - } - posts, aid := apPostsToResponses(res) - return latestCacheValue{posts: posts, aid: aid}, nil - case Nostr: - res, err := s.nostrClient.GetPostsInTimeframe(ctx, author, start, end, limit) - if err != nil { - return latestCacheValue{}, err - } - posts, aid := nostrPostsToResponses(res) - return latestCacheValue{posts: posts, aid: aid}, nil - case Twitter: - res, err := s.twitterClient.GetPostsInTimeframe(ctx, author, start, end, limit) - if err != nil { - return latestCacheValue{}, err - } - posts, aid := twitterPostsToResponses(res) - return latestCacheValue{posts: posts, aid: aid}, nil - } - return latestCacheValue{}, fmt.Errorf("unsupported platform %q", platform) - }) - if err != nil { - return nil, "", err - } - return cv.posts, cv.aid, nil -} - -func (s *Server) fetchFeed(ctx context.Context, platform platform, uri string) ([]GetPostResponsePost, error) { - key := fmt.Sprintf("feed:%s|%s", platform, uri) - return cachedFetch(s.postCache, key, func() ([]GetPostResponsePost, error) { - if platform == Bluesky { - res, err := s.bskyClient.GetFeed(ctx, uri, 50) - if err != nil { - return nil, err - } - posts, _ := blueskyFeedViewsToResponses(res.Feed) - return posts, nil - } - // Non-Bluesky platforms have no custom-feed concept; treat the URI as an author handle - // and return their latest posts as the feed. - posts, _, err := s.fetchLatestPosts(ctx, platform, uri, 50) - return posts, err - }) -} diff --git a/internal/server/posts_convert.go b/internal/server/posts_convert.go deleted file mode 100644 index 181c9fc..0000000 --- a/internal/server/posts_convert.go +++ /dev/null @@ -1,252 +0,0 @@ -package server - -import ( - "time" - - "github.com/bluesky-social/indigo/api/bsky" - "github.com/nacorid/naco-api/internal/activitypub" - "github.com/nacorid/naco-api/internal/clustering" - "github.com/nacorid/naco-api/internal/nostr" - "github.com/nacorid/naco-api/internal/twitter" -) - -func bskyPostViewToResponse(p *bsky.FeedDefs_PostView) GetPostResponsePost { - author := "" - if p.Author != nil { - author = p.Author.Did - } - content := "" - created := "" - if fp, ok := p.Record.Val.(*bsky.FeedPost); ok { - content = fp.Text - created = fp.CreatedAt - } - likes, reposts, quotes, replies := int64(0), int64(0), int64(0), int64(0) - if p.LikeCount != nil { - likes = *p.LikeCount - } - if p.RepostCount != nil { - reposts = *p.RepostCount - } - if p.QuoteCount != nil { - quotes = *p.QuoteCount - } - if p.ReplyCount != nil { - replies = *p.ReplyCount - } - return GetPostResponsePost{ - Author: author, - Content: content, - Likes: likes, - Reposts: reposts, - Quotes: quotes, - Replies: replies, - CreatedAt: created, - } -} - -func apPostToResponse(p activitypub.Post) GetPostResponsePost { - created := "" - if !p.CreatedAt.IsZero() { - created = p.CreatedAt.Format(time.RFC3339) - } - return GetPostResponsePost{ - Author: p.Author, - Content: p.Content, - Likes: int64(p.FavouritesCount), - Reposts: int64(p.ReblogsCount), - Replies: int64(p.RepliesCount), - CreatedAt: created, - } -} - -func nostrPostToResponse(p *nostr.Post) GetPostResponsePost { - created := "" - if !p.CreatedAt.IsZero() { - created = p.CreatedAt.Format(time.RFC3339) - } - return GetPostResponsePost{ - Author: p.Author, - Content: p.Content, - CreatedAt: created, - } -} - -func twitterPostToResponse(p *twitter.Post) GetPostResponsePost { - created := "" - if !p.CreatedAt.IsZero() { - created = p.CreatedAt.Format(time.RFC3339) - } - return GetPostResponsePost{ - Author: p.Author, - Content: p.Content, - Likes: int64(p.Likes), - Reposts: int64(p.Reposts), - Quotes: int64(p.Quotes), - Replies: int64(p.Replies), - CreatedAt: created, - } -} - -// ---- Batch converters for GetPostResponsePost ---- - -func blueskyFeedViewsToResponses(views []*bsky.FeedDefs_FeedViewPost) ([]GetPostResponsePost, string) { - posts := make([]GetPostResponsePost, 0, len(views)) - var aid string - for _, p := range views { - if p.Post == nil { - continue - } - posts = append(posts, bskyPostViewToResponse(p.Post)) - if p.Post.Author != nil { - aid = p.Post.Author.Did - } - } - return posts, aid -} - -func apPostsPtrToResponses(posts []*activitypub.Post) ([]GetPostResponsePost, string) { - out := make([]GetPostResponsePost, 0, len(posts)) - var aid string - for _, p := range posts { - out = append(out, apPostToResponse(*p)) - aid = p.Author - } - return out, aid -} - -func apPostsToResponses(posts []activitypub.Post) ([]GetPostResponsePost, string) { - out := make([]GetPostResponsePost, 0, len(posts)) - var aid string - for _, p := range posts { - out = append(out, apPostToResponse(p)) - aid = p.Author - } - return out, aid -} - -func nostrPostsToResponses(posts []*nostr.Post) ([]GetPostResponsePost, string) { - out := make([]GetPostResponsePost, 0, len(posts)) - var aid string - for _, p := range posts { - out = append(out, nostrPostToResponse(p)) - aid = p.Author - } - return out, aid -} - -func twitterPostsToResponses(posts []*twitter.Post) ([]GetPostResponsePost, string) { - out := make([]GetPostResponsePost, 0, len(posts)) - var aid string - for _, p := range posts { - out = append(out, twitterPostToResponse(p)) - aid = p.Author - } - return out, aid -} - -// ---- Batch converters for clustering.Item (style & analytics) ---- - -func blueskyFeedViewsToItems(views []*bsky.FeedDefs_FeedViewPost) []clustering.Item { - out := make([]clustering.Item, 0, len(views)) - for _, p := range views { - fp, ok := p.Post.Record.Val.(*bsky.FeedPost) - if !ok { - continue - } - out = append(out, clustering.Item{ID: p.Post.Uri, Text: fp.Text}) - } - return out -} - -func apPostsPtrToItems(posts []*activitypub.Post) []clustering.Item { - out := make([]clustering.Item, 0, len(posts)) - for _, p := range posts { - if p.Text != "" { - out = append(out, clustering.Item{ID: p.ID, Text: p.Text}) - } - } - return out -} - -func nostrPostsToItems(posts []*nostr.Post) []clustering.Item { - out := make([]clustering.Item, 0, len(posts)) - for _, p := range posts { - if p.Content != "" { - out = append(out, clustering.Item{ID: p.ID, Text: p.Content}) - } - } - return out -} - -func twitterPostsToItems(posts []*twitter.Post) []clustering.Item { - out := make([]clustering.Item, 0, len(posts)) - for _, p := range posts { - if p.Content != "" { - out = append(out, clustering.Item{ID: p.ID, Text: p.Content}) - } - } - return out -} - -// analytics-specific converters that preserve engagement + timestamps - -func blueskyFeedViewsToAnalyticsItems(views []*bsky.FeedDefs_FeedViewPost) []clustering.Item { - out := make([]clustering.Item, 0, len(views)) - for _, p := range views { - if p.Post == nil || p.Post.Record == nil { - continue - } - fp, ok := p.Post.Record.Val.(*bsky.FeedPost) - if !ok { - continue - } - item := clustering.Item{ID: p.Post.Uri, Text: fp.Text} - if ts, err := time.Parse(time.RFC3339, p.Post.IndexedAt); err == nil { - item.Timestamp = ts - } - item.Likes = int64Val(p.Post.LikeCount) - item.Reposts = int64Val(p.Post.RepostCount) - item.Replies = int64Val(p.Post.ReplyCount) - item.Quotes = int64Val(p.Post.QuoteCount) - out = append(out, item) - } - return out -} - -func apPostsToAnalyticsItems(posts []activitypub.Post) []clustering.Item { - out := make([]clustering.Item, 0, len(posts)) - for _, p := range posts { - if p.Text != "" { - out = append(out, clustering.Item{ - ID: p.ID, Text: p.Text, Timestamp: p.CreatedAt, - Likes: int64(p.FavouritesCount), Reposts: int64(p.ReblogsCount), Replies: int64(p.RepliesCount), - }) - } - } - return out -} - -func nostrPostsToAnalyticsItems(posts []*nostr.Post) []clustering.Item { - out := make([]clustering.Item, 0, len(posts)) - for _, p := range posts { - if p.Content != "" { - out = append(out, clustering.Item{ID: p.ID, Text: p.Content, Timestamp: p.CreatedAt}) - } - } - return out -} - -func twitterPostsToAnalyticsItems(posts []*twitter.Post) []clustering.Item { - out := make([]clustering.Item, 0, len(posts)) - for _, p := range posts { - if p.Content == "" { - continue - } - out = append(out, clustering.Item{ - ID: p.ID, Text: p.Content, Timestamp: p.CreatedAt, - Likes: int64(p.Likes), Reposts: int64(p.Reposts), Replies: int64(p.Replies), Quotes: int64(p.Quotes), - }) - } - return out -} diff --git a/internal/server/prewarm.go b/internal/server/prewarm.go deleted file mode 100644 index 55caf69..0000000 --- a/internal/server/prewarm.go +++ /dev/null @@ -1,101 +0,0 @@ -package server - -import ( - "context" - "time" - - log "github.com/nacorid/logger" -) - -// StartPrewarm launches a background goroutine that discovers interesting authors and -// pre-fetches their style profiles. It issues at most one pre-fetch "request" per minute -// so upstream platforms, Ollama and the database are never hit harder than that cadence -// allows. Profiles are still only (re)written when older than a week, exactly like a live -// getStyleProfile request. Failures (including rate limits) are logged and retried on the -// next tick. -func (s *Server) StartPrewarm(ctx context.Context) { - if s.prewarmFeedURI == "" && len(s.prewarmAuthors) == 0 { - log.DebugWithContext(ctx, "no prewarm feed or authors configured, skipping prewarm") - return - } - - go func() { - authors := s.discoverInterestingAuthors(ctx) - if len(authors) == 0 { - return - } - s.prewarmOne(ctx, authors[0]) - ticker := time.NewTicker(time.Minute) - defer ticker.Stop() - i := 1 - for { - select { - case <-ctx.Done(): - return - case <-ticker.C: - if i >= len(authors) { - // re-discover interesting profiles and restart the cycle - authors = s.discoverInterestingAuthors(ctx) - i = 0 - if len(authors) == 0 { - continue - } - } - author := authors[i] - i++ - s.prewarmOne(ctx, author) - } - } - }() -} - -// discoverInterestingAuthors returns a list of authors worth pre-fetching. When a feed URI -// is configured it pulls recent posts from that feed and collects the author DIDs; -// otherwise it falls back to the statically configured seed list. -func (s *Server) discoverInterestingAuthors(ctx context.Context) []string { - if s.prewarmFeedURI != "" { - feed, err := s.bskyClient.GetFeed(ctx, s.prewarmFeedURI, 50) - if err != nil { - log.WarnWithContext(ctx, "prewarm: failed to discover authors from feed", "error", err) - } else { - seen := make(map[string]struct{}) - authors := make([]string, 0, len(feed.Feed)) - for _, p := range feed.Feed { - if p.Post == nil || p.Post.Author == nil { - continue - } - did := p.Post.Author.Did - if did == "" { - continue - } - if _, ok := seen[did]; ok { - continue - } - seen[did] = struct{}{} - authors = append(authors, did) - } - if len(authors) > 0 { - return authors - } - } - } - return s.prewarmAuthors -} - -func (s *Server) prewarmOne(ctx context.Context, author string) { - if _, err := s.requestStyleProfile(ctx, author); err != nil { - log.WarnWithContext(ctx, "prewarm: failed to build profile", "author", author, "error", err) - return - } - log.InfoWithContext(ctx, "prewarmed style profile", "author", author) -} - -func (s *Server) startPostPruner() { - ticker := time.NewTicker(24 * time.Hour) - defer ticker.Stop() - for range ticker.C { - if err := s.db.PruneStylePosts(context.Background()); err != nil { - log.Error("failed to prune style posts", "error", err) - } - } -} diff --git a/internal/server/server.go b/internal/server/server.go index 7e3d2a8..f2d1011 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -2,45 +2,52 @@ package server import ( "context" + "database/sql" + "encoding/json" + "errors" + "fmt" + "maps" + "net/http" "os" + "strings" + "sync" "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/clustering" "github.com/nacorid/naco-api/internal/config" "github.com/nacorid/naco-api/internal/nostr" "github.com/nacorid/naco-api/internal/ollama" "github.com/nacorid/naco-api/internal/store" "github.com/nacorid/naco-api/internal/twitter" + "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 { - // Transport & middleware - listenAddress string - permissions os.FileMode - mcpHandler *x402.X402Server - metrics *MetricsMiddleware - - // Dependencies - db Store - apClient apBackend - bskyClient blueskyBackend - nostrClient nostrBackend - twitterClient twitterBackend - ollama ollamaBackend - - // Cache & background work - postCache *ttlCache - enrichSem chan struct{} - prewarmAuthors []string - prewarmFeedURI string - enrichEnabled bool - - // Payments & analytics tuning + listenAddress string + permissions os.FileMode + mcpHandler *x402.X402Server + db Store + apClient apBackend + bskyClient blueskyBackend + nostrClient nostrBackend + twitterClient twitterBackend + ollama ollamaBackend + metrics *MetricsMiddleware + postCache *ttlCache + prewarmAuthors []string + prewarmFeedURI string + enrichEnabled bool x402Wallet string activeNetworks []NetworkInfo + enrichSem chan struct{} analyticsTTLEngagement time.Duration analyticsTTLPatterns time.Duration analyticsTTLTopics time.Duration @@ -49,9 +56,95 @@ type Server struct { } func New(ctx context.Context, db *store.Queries, ap *activitypub.APClient, bsky *bluesky.BlueskyClient, n *nostr.Client, t *twitter.Client, o *ollama.OllamaClient, cfg *config.Config) *Server { - toolchain, metrics := buildMiddleware() - srv := newX402Server(ctx, db, cfg, toolchain, metrics) - activeNetworks := resolveActiveNetworks(ctx, cfg.ActiveNetworks) + 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 + + _ = settleHook(settleCtx.Ctx, db, settleCtx.Result.Transaction, amount, asset, string(settleCtx.Result.Network), to, from) + + // Send webhook after transaction is saved (non-blocking, goroutine) + // settleCtx.Result.Transaction is the x402 transaction hash (string) + go func() { + wh := newWebhookSender(cfg, metrics) + if wh != nil { + // Build payload with available transaction hash and default platform + txHash := settleCtx.Result.Transaction + platform := "unknown" + payload := map[string]any{ + "event": "transaction_success", + "transaction": map[string]any{ + "x402_transaction_hash": txHash, + }, + "metrics": map[string]any{ + "function_runtime_ms": 0, + "platform": platform, + }, + "timestamp": time.Now().UTC().Format(time.RFC3339), + "server_id": cfg.MatrixServerID, + } + jsonBytes, _ := json.Marshal(payload) + wh.Send(settleCtx.Ctx, jsonBytes) + } + }() + + return nil + }) + + 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, @@ -78,12 +171,6 @@ func New(ctx context.Context, db *store.Queries, ap *activitypub.APClient, bsky analyticsTTLDrivers: cfg.AnalyticsTTLDrivers, } - log.DebugWithContext(ctx, "server initialized", - "networks", len(activeNetworks), - "enrich", s.enrichEnabled, - "prewarm_authors", len(s.prewarmAuthors), - ) - s.registerTools() return s @@ -96,7 +183,1632 @@ func (s *Server) MetricsText() string { return s.metrics.MetricsText() } +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 (Bluesky only). 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, + // Base registration is the maximum possible charge ($0.36): $0.10 base + // cost + $0.01 user lookup + $0.005 x 50 posts for a Twitter profile. + // The settlement override in the x402 handler charges the actual + // per-usage amount (Twitter passes API costs on to the caller). + s.buildRequirements("0.36", 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 author", + Required: true, + }, + }, + }, + Output: map[string]any{ + "matches": Bodyfield{ + Type: "array", + Description: "Ranked list of similar profiles (author, platform, similarity, top keywords, style summary, post count)", + }, + }, + }, + } + + s.mcpHandler.AddPayableTool( + mcp.Tool{Name: "findSimilarUsers", + Description: "Find authors with a similar writing style to the given author, across all platforms", + InputSchema: FindSimilarUsersRequest{}, + OutputSchema: FindSimilarUsersResponse{}, + }, + s.getSimilarProfiles, + s.buildRequirements("0.05", extra)..., + ) + + extra = map[string]any{ + "OutputSchema": Schema{ + Input: InputSchema{ + Type: "HTTP", + Method: "POST", + Discoverable: true, + BodyFields: map[string]any{ + "author": Bodyfield{ + Type: "string", + Description: "Did, handle, npub or @handle of an actor", + Required: true, + }, + }, + }, + Output: map[string]any{ + "actor": Bodyfield{ + Type: "object", + Description: "Profile of the actor (platform, handle, displayName, avatarUrl, bio, followersCount, followingCount, postCount, url)", + }, + }, + }, + } + + s.mcpHandler.AddPayableTool( + mcp.Tool{Name: "getActor", + Description: "Get an actor's public profile (bio, avatar, follower counts) on the platform inferred from the identifier", + InputSchema: GetActorRequest{}, + OutputSchema: GetActorResponse{}, + }, + s.getActor, + s.buildRequirements("0.01", extra)..., + ) + + extra = map[string]any{ + "OutputSchema": Schema{ + Input: InputSchema{ + Type: "HTTP", + Method: "POST", + Discoverable: true, + BodyFields: map[string]any{ + "query": Bodyfield{ + Type: "string", + Description: "Search query (handle, name or keyword)", + Required: true, + }, + "limit": Bodyfield{ + Type: "integer", + Description: "Maximum number of results (default 10, max 50)", + Required: false, + }, + }, + }, + Output: map[string]any{ + "results": Bodyfield{ + Type: "array", + Description: "Ranked list of matching actor profiles across platforms", + }, + }, + }, + } + + s.mcpHandler.AddPayableTool( + mcp.Tool{Name: "searchActors", + Description: "Search for actors by handle/name/keyword across all platforms", + InputSchema: SearchActorsRequest{}, + OutputSchema: SearchActorsResponse{}, + }, + s.searchActors, + s.buildRequirements("0.01", extra)..., + ) + + // --- Analytics tools --- + + extra = map[string]any{ + "OutputSchema": Schema{ + Input: InputSchema{ + Type: "HTTP", + Method: "POST", + Discoverable: true, + BodyFields: map[string]any{ + "author": Bodyfield{Type: "string", Description: "Did or handle of an author", Required: true}, + "startTime": Bodyfield{Type: "string", Description: "Start time in RFC3339 format (default: 30 days ago)", Required: false}, + "endTime": Bodyfield{Type: "string", Description: "End time in RFC3339 format (default: now)", Required: false}, + "limit": Bodyfield{Type: "integer", Description: "Maximum number of posts to analyze (default: 50, max: 200)", Required: false}, + "platform": Bodyfield{Type: "string", Description: "Platform to filter (default: all)", Required: false}, + }, + }, + Output: map[string]any{ + "author": Bodyfield{Type: "string", Description: "Author identifier"}, + "metrics": Bodyfield{Type: "array", Description: "Engagement metrics per post"}, + }, + }, + } + + s.mcpHandler.AddPayableTool( + mcp.Tool{Name: "getEngagementMetrics", + Description: "Get time-series engagement metrics (likes, reposts, replies, quotes) per post", + InputSchema: AnalyticsRequest{}, + OutputSchema: EngagementMetricsResponse{}, + }, + s.getEngagementMetrics, + s.buildRequirements("0.15", extra)..., + ) + + extra = map[string]any{ + "OutputSchema": Schema{ + Input: InputSchema{ + Type: "HTTP", + Method: "POST", + Discoverable: true, + BodyFields: map[string]any{ + "author": Bodyfield{Type: "string", Description: "Did or handle of an author", Required: true}, + "startTime": Bodyfield{Type: "string", Description: "Start time in RFC3339 format (default: 30 days ago)", Required: false}, + "endTime": Bodyfield{Type: "string", Description: "End time in RFC3339 format (default: now)", Required: false}, + "limit": Bodyfield{Type: "integer", Description: "Maximum number of posts to analyze (default: 50, max: 200)", Required: false}, + "platform": Bodyfield{Type: "string", Description: "Platform to filter (default: all)", Required: false}, + }, + }, + Output: map[string]any{ + "author": Bodyfield{Type: "string", Description: "Author identifier"}, + "heatmap": Bodyfield{Type: "array", Description: "24x7 matrix: hour (0-23) x day of week (0-6 = Sun-Sat)"}, + "peakHours": Bodyfield{Type: "array", Description: "Top 3 hours with most posts (0-23)"}, + "peakDays": Bodyfield{Type: "array", Description: "Top 3 days with most posts (0-6 = Sun-Sat)"}, + "avgPostsPerDay": Bodyfield{Type: "number", Description: "Average posts per day"}, + "totalPosts": Bodyfield{Type: "integer", Description: "Total posts in time range"}, + "postingDays": Bodyfield{Type: "integer", Description: "Number of days with at least one post"}, + }, + }, + } + + s.mcpHandler.AddPayableTool( + mcp.Tool{Name: "getPostingPatterns", + Description: "Get posting heatmap and patterns (hour-of-day x day-of-week)", + InputSchema: AnalyticsRequest{}, + OutputSchema: PostingPatternsResponse{}, + }, + s.getPostingPatterns, + s.buildRequirements("0.10", extra)..., + ) + + extra = map[string]any{ + "OutputSchema": Schema{ + Input: InputSchema{ + Type: "HTTP", + Method: "POST", + Discoverable: true, + BodyFields: map[string]any{ + "author": Bodyfield{Type: "string", Description: "Did or handle of an author", Required: true}, + "startTime": Bodyfield{Type: "string", Description: "Start time in RFC3339 format (default: 30 days ago)", Required: false}, + "endTime": Bodyfield{Type: "string", Description: "End time in RFC3339 format (default: now)", Required: false}, + "limit": Bodyfield{Type: "integer", Description: "Maximum number of posts to analyze (default: 50, max: 200)", Required: false}, + "platform": Bodyfield{Type: "string", Description: "Platform to filter (default: all)", Required: false}, + }, + }, + Output: map[string]any{ + "author": Bodyfield{Type: "string", Description: "Author identifier"}, + "topics": Bodyfield{Type: "array", Description: "Identified topic clusters"}, + }, + }, + } + + s.mcpHandler.AddPayableTool( + mcp.Tool{Name: "getTopicModeling", + Description: "Get LLM-based topic extraction and clustering from posts", + InputSchema: AnalyticsRequest{}, + OutputSchema: TopicModelingResponse{}, + }, + s.getTopicModeling, + s.buildRequirements("0.20", extra)..., + ) + + extra = map[string]any{ + "OutputSchema": Schema{ + Input: InputSchema{ + Type: "HTTP", + Method: "POST", + Discoverable: true, + BodyFields: map[string]any{ + "author": Bodyfield{Type: "string", Description: "Did or handle of an author", Required: true}, + "startTime": Bodyfield{Type: "string", Description: "Start time in RFC3339 format (default: 30 days ago)", Required: false}, + "endTime": Bodyfield{Type: "string", Description: "End time in RFC3339 format (default: now)", Required: false}, + "limit": Bodyfield{Type: "integer", Description: "Maximum number of posts to analyze (default: 50, max: 200)", Required: false}, + "platform": Bodyfield{Type: "string", Description: "Platform to filter (default: all)", Required: false}, + }, + }, + Output: map[string]any{ + "author": Bodyfield{Type: "string", Description: "Author identifier"}, + "timeSeries": Bodyfield{Type: "array", Description: "Sentiment over time"}, + "overallSentiment": Bodyfield{Type: "object", Description: "Overall sentiment summary"}, + "dominantEmotions": Bodyfield{Type: "array", Description: "Most frequent emotions"}, + "sentimentShifts": Bodyfield{Type: "array", Description: "Significant sentiment changes"}, + }, + }, + } + + s.mcpHandler.AddPayableTool( + mcp.Tool{Name: "getSentimentTrends", + Description: "Get sentiment analysis and emotional arcs over time", + InputSchema: AnalyticsRequest{}, + OutputSchema: SentimentTrendsResponse{}, + }, + s.getSentimentTrends, + s.buildRequirements("0.15", extra)..., + ) + + extra = map[string]any{ + "OutputSchema": Schema{ + Input: InputSchema{ + Type: "HTTP", + Method: "POST", + Discoverable: true, + BodyFields: map[string]any{ + "author": Bodyfield{Type: "string", Description: "Did or handle of an author", Required: true}, + "startTime": Bodyfield{Type: "string", Description: "Start time in RFC3339 format (default: 30 days ago)", Required: false}, + "endTime": Bodyfield{Type: "string", Description: "End time in RFC3339 format (default: now)", Required: false}, + "limit": Bodyfield{Type: "integer", Description: "Maximum number of posts to analyze (default: 50, max: 200)", Required: false}, + "platform": Bodyfield{Type: "string", Description: "Platform to filter (default: all)", Required: false}, + }, + }, + Output: map[string]any{ + "author": Bodyfield{Type: "string", Description: "Author identifier"}, + "drivers": Bodyfield{Type: "array", Description: "Topics/keywords driving engagement"}, + }, + }, + } + + s.mcpHandler.AddPayableTool( + mcp.Tool{Name: "getEngagementDrivers", + Description: "Get topics/keywords that correlate with high engagement", + InputSchema: AnalyticsRequest{}, + OutputSchema: EngagementDriversResponse{}, + }, + s.getEngagementDrivers, + s.buildRequirements("0.20", 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") + } + platform, err := determinePlatform(args.Author) + if err != nil { + return nil, err + } + if platform == Twitter { + return nil, fmt.Errorf("twitter_unavailable: X API policy prohibits reselling post data") + } + post, err := s.fetchSinglePost(ctx, platform, args.Author, args.RecordKey, args.Collection) + if err != nil { + return nil, err + } + respBytes, err := json.Marshal(post) + 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") + } + platform, err := determinePlatform(args.Author) + if err != nil { + return nil, err + } + if platform == Twitter { + return nil, fmt.Errorf("twitter_unavailable: X API policy prohibits reselling post data") + } + posts, authorID, err := s.fetchLatestPosts(ctx, platform, args.Author, args.Limit) + if err != nil { + return nil, err + } + if authorID == "" { + authorID = args.Author + } + resp := GetLatestPostsResponse{ + Author: authorID, + 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") + } + if args.Limit > 50 { + return nil, fmt.Errorf("n must be at most 50") + } + limit := args.Limit + 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("endTime must be in RFC3339 format") + } + if startTime.After(endTime) { + tmp := startTime + startTime = endTime + endTime = tmp + } + platform, err := determinePlatform(args.Author) + if err != nil { + return nil, err + } + if platform == Twitter { + return nil, fmt.Errorf("twitter_unavailable: X API policy prohibits reselling post data") + } + posts, authorID, err := s.fetchPostsInTime(ctx, platform, args.Author, startTime, endTime, limit) + if err != nil { + return nil, err + } + if authorID == "" { + authorID = args.Author + } + resp := &GetPostsInTimeResponse{ + Author: authorID, + 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") + } + platform, err := determinePlatform(args.URI) + if err != nil { + return nil, err + } + if platform == Twitter { + return nil, fmt.Errorf("twitter_unavailable: X API policy prohibits reselling feed data") + } + posts, err := s.fetchFeed(ctx, platform, args.URI) + if err != nil { + return nil, err + } + 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) getActor(ctx context.Context, request *mcp.CallToolRequest) (*mcp.CallToolResult, error) { + var args GetActorRequest + 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, handle, npub or @handle") + } + platform, err := determinePlatform(args.Author) + if err != nil { + return nil, err + } + var ( + prof *ActorProfile + e error + ) + switch platform { + case Bluesky: + prof, e = s.bskyClient.GetActor(ctx, args.Author) + case ActivityPub: + prof, e = s.apClient.GetActor(ctx, args.Author) + case Nostr: + prof, e = s.nostrClient.GetActor(ctx, args.Author) + case Twitter: + return nil, fmt.Errorf("twitter_unavailable: X API policy prohibits reselling user/profile data") + default: + return nil, fmt.Errorf("unsupported platform %q", platform) + } + if e != nil { + return nil, e + } + respBytes, err := json.Marshal(GetActorResponse{Actor: *prof}) + if err != nil { + return nil, err + } + return &mcp.CallToolResult{ + Content: []mcp.Content{ + &mcp.TextContent{Text: string(respBytes)}, + }, + }, nil +} + +func (s *Server) searchActors(ctx context.Context, request *mcp.CallToolRequest) (*mcp.CallToolResult, error) { + var args SearchActorsRequest + if err := json.Unmarshal(request.Params.Arguments, &args); err != nil { + return nil, fmt.Errorf("invalid arguments: %w", err) + } + if args.Query == "" { + return nil, fmt.Errorf("query must not be empty") + } + limit := args.Limit + if limit <= 0 { + limit = 10 + } + if limit > 50 { + limit = 50 + } + results, err := s.searchActorsAll(ctx, args.Query, limit) + if err != nil { + return nil, err + } + respBytes, err := json.Marshal(SearchActorsResponse{Results: results}) + if err != nil { + return nil, err + } + return &mcp.CallToolResult{ + Content: []mcp.Content{ + &mcp.TextContent{Text: string(respBytes)}, + }, + }, nil +} + +// searchActorsAll fans the query out to every platform client and merges the results. This is +// a payable tool, so we never bill for a partial result: if at least one platform succeeds we +// return whatever it produced (possibly empty). We only error when every platform failed, +// so the caller isn't charged for a search that produced nothing. +func (s *Server) searchActorsAll(ctx context.Context, query string, limit int) ([]ActorProfile, error) { + var ( + mu sync.Mutex + results []ActorProfile + wg sync.WaitGroup + errMu sync.Mutex + errs []error + success int + ) + call := func(fn func(context.Context, string, int) ([]ActorProfile, error)) { + defer wg.Done() + r, err := fn(ctx, query, limit) + if err != nil { + errMu.Lock() + errs = append(errs, err) + errMu.Unlock() + return + } + mu.Lock() + results = append(results, r...) + success++ + mu.Unlock() + } + wg.Add(3) + go call(s.bskyClient.SearchActors) + go call(s.apClient.SearchActors) + go call(s.nostrClient.SearchActors) + wg.Wait() + if success == 0 { + return nil, fmt.Errorf("all platforms failed: %v", errs) + } + return dedupeActors(results), nil +} + +// dedupeActors removes near-duplicate profiles across platforms. Two results are considered +// the same person when they share a normalized handle (the '@' and any instance/host stripped, +// lower-cased) or a normalized display name. The first occurrence wins. +func dedupeActors(in []ActorProfile) []ActorProfile { + seen := make(map[string]struct{}, len(in)) + out := make([]ActorProfile, 0, len(in)) + for _, p := range in { + key := normalizeHandle(p.Handle) + if key == "" { + key = normalizeHandle(p.DisplayName) + } + if key == "" { + out = append(out, p) + continue + } + if _, dup := seen[key]; dup { + continue + } + seen[key] = struct{}{} + out = append(out, p) + } + return out +} + +func normalizeHandle(h string) string { + h = strings.ToLower(strings.TrimSpace(h)) + h = strings.TrimPrefix(h, "@") + // Drop an ActivityPub instance suffix (alice@mastodon.social -> alice). + if at := strings.Index(h, "@"); at >= 0 { + h = h[:at] + } + return h +} + +// --- cross-platform post helpers --- + +func bskyPostViewToResponse(p *bsky.FeedDefs_PostView) GetPostResponsePost { + author := "" + if p.Author != nil { + author = p.Author.Did + } + content := "" + created := "" + if fp, ok := p.Record.Val.(*bsky.FeedPost); ok { + content = fp.Text + created = fp.CreatedAt + } + likes, reposts, quotes, replies := int64(0), int64(0), int64(0), int64(0) + if p.LikeCount != nil { + likes = *p.LikeCount + } + if p.RepostCount != nil { + reposts = *p.RepostCount + } + if p.QuoteCount != nil { + quotes = *p.QuoteCount + } + if p.ReplyCount != nil { + replies = *p.ReplyCount + } + return GetPostResponsePost{ + Author: author, + Content: content, + Likes: likes, + Reposts: reposts, + Quotes: quotes, + Replies: replies, + CreatedAt: created, + } +} + +func apPostToResponse(p activitypub.Post) GetPostResponsePost { + created := "" + if !p.CreatedAt.IsZero() { + created = p.CreatedAt.Format(time.RFC3339) + } + return GetPostResponsePost{ + Author: p.Author, + Content: p.Content, + Likes: int64(p.FavouritesCount), + Reposts: int64(p.ReblogsCount), + Replies: int64(p.RepliesCount), + CreatedAt: created, + } +} + +func nostrPostToResponse(p *nostr.Post) GetPostResponsePost { + created := "" + if !p.CreatedAt.IsZero() { + created = p.CreatedAt.Format(time.RFC3339) + } + return GetPostResponsePost{ + Author: p.Author, + Content: p.Content, + CreatedAt: created, + } +} + +func twitterPostToResponse(p *twitter.Post) GetPostResponsePost { + created := "" + if !p.CreatedAt.IsZero() { + created = p.CreatedAt.Format(time.RFC3339) + } + return GetPostResponsePost{ + Author: p.Author, + Content: p.Content, + Likes: int64(p.Likes), + Reposts: int64(p.Reposts), + Quotes: int64(p.Quotes), + Replies: int64(p.Replies), + CreatedAt: created, + } +} + +func (s *Server) fetchSinglePost(ctx context.Context, platform platform, author, recordKey, collection string) (*GetPostResponsePost, error) { + key := fmt.Sprintf("post:%s|%s|%s|%s", platform, author, collection, recordKey) + if v, ok := s.postCache.get(key); ok { + return v.(*GetPostResponsePost), nil + } + switch platform { + case Bluesky: + res, err := s.bskyClient.GetPost(ctx, author, collection, recordKey) + if err != nil { + return nil, err + } + r := bskyPostViewToResponse(res) + s.postCache.set(key, &r) + return &r, nil + case ActivityPub: + p, err := s.apClient.GetPost(ctx, recordKey) + if err != nil { + return nil, err + } + r := apPostToResponse(*p) + s.postCache.set(key, &r) + return &r, nil + case Nostr: + p, err := s.nostrClient.GetPost(ctx, recordKey) + if err != nil { + return nil, err + } + r := nostrPostToResponse(p) + s.postCache.set(key, &r) + return &r, nil + case Twitter: + p, err := s.twitterClient.GetPost(ctx, recordKey) + if err != nil { + return nil, err + } + r := twitterPostToResponse(p) + s.postCache.set(key, &r) + return &r, nil + } + return nil, fmt.Errorf("unsupported platform %q", platform) +} + +func (s *Server) fetchLatestPosts(ctx context.Context, platform platform, author string, limit int) ([]GetPostResponsePost, string, error) { + key := fmt.Sprintf("latest:%s|%s|%d", platform, author, limit) + if v, ok := s.postCache.get(key); ok { + cv := v.(latestCacheValue) + return cv.posts, cv.aid, nil + } + switch platform { + case Bluesky: + res, err := s.bskyClient.GetLatestNPosts(ctx, author, limit) + if err != nil { + return nil, "", err + } + posts := make([]GetPostResponsePost, 0, len(res)) + var aid string + for _, p := range res { + if p.Post == nil { + continue + } + posts = append(posts, bskyPostViewToResponse(p.Post)) + if p.Post.Author != nil { + aid = p.Post.Author.Did + } + } + s.postCache.set(key, latestCacheValue{posts: posts, aid: aid}) + return posts, aid, nil + case ActivityPub: + res, err := s.apClient.GetLatestNPosts(ctx, author, limit) + if err != nil { + return nil, "", err + } + posts := make([]GetPostResponsePost, 0, len(res)) + var aid string + for _, p := range res { + posts = append(posts, apPostToResponse(*p)) + aid = p.Author + } + s.postCache.set(key, latestCacheValue{posts: posts, aid: aid}) + return posts, aid, nil + case Nostr: + res, err := s.nostrClient.GetLatestNPosts(ctx, author, limit) + if err != nil { + return nil, "", err + } + posts := make([]GetPostResponsePost, 0, len(res)) + var aid string + for _, p := range res { + posts = append(posts, nostrPostToResponse(p)) + aid = p.Author + } + s.postCache.set(key, latestCacheValue{posts: posts, aid: aid}) + return posts, aid, nil + case Twitter: + res, err := s.twitterClient.GetLatestNPosts(ctx, author, limit) + if err != nil { + return nil, "", err + } + posts := make([]GetPostResponsePost, 0, len(res)) + var aid string + for _, p := range res { + posts = append(posts, twitterPostToResponse(p)) + aid = p.Author + } + s.postCache.set(key, latestCacheValue{posts: posts, aid: aid}) + return posts, aid, nil + } + return nil, "", fmt.Errorf("unsupported platform %q", platform) +} + +func (s *Server) fetchPostsInTime(ctx context.Context, platform platform, author string, start, end time.Time, limit int) ([]GetPostResponsePost, string, error) { + key := fmt.Sprintf("time:%s|%s|%s|%s|%d", platform, author, start.Format(time.RFC3339), end.Format(time.RFC3339), limit) + if v, ok := s.postCache.get(key); ok { + cv := v.(latestCacheValue) + return cv.posts, cv.aid, nil + } + switch platform { + case Bluesky: + res, err := s.bskyClient.GetPostsInTimeframe(ctx, author, start, end, limit) + if err != nil { + return nil, "", err + } + posts := make([]GetPostResponsePost, 0, len(res)) + var aid string + for _, p := range res { + if p.Post == nil { + continue + } + posts = append(posts, bskyPostViewToResponse(p.Post)) + if p.Post.Author != nil { + aid = p.Post.Author.Did + } + } + s.postCache.set(key, latestCacheValue{posts: posts, aid: aid}) + return posts, aid, nil + case ActivityPub: + res, err := s.apClient.GetPostsInTimeframe(ctx, author, start, end, limit) + if err != nil { + return nil, "", err + } + posts := make([]GetPostResponsePost, 0, len(res)) + var aid string + for _, p := range res { + posts = append(posts, apPostToResponse(p)) + aid = p.Author + } + s.postCache.set(key, latestCacheValue{posts: posts, aid: aid}) + return posts, aid, nil + case Nostr: + res, err := s.nostrClient.GetPostsInTimeframe(ctx, author, start, end, limit) + if err != nil { + return nil, "", err + } + posts := make([]GetPostResponsePost, 0, len(res)) + var aid string + for _, p := range res { + posts = append(posts, nostrPostToResponse(p)) + aid = p.Author + } + s.postCache.set(key, latestCacheValue{posts: posts, aid: aid}) + return posts, aid, nil + case Twitter: + res, err := s.twitterClient.GetPostsInTimeframe(ctx, author, start, end, limit) + if err != nil { + return nil, "", err + } + posts := make([]GetPostResponsePost, 0, len(res)) + var aid string + for _, p := range res { + posts = append(posts, twitterPostToResponse(p)) + aid = p.Author + } + s.postCache.set(key, latestCacheValue{posts: posts, aid: aid}) + return posts, aid, nil + } + return nil, "", fmt.Errorf("unsupported platform %q", platform) +} + +func (s *Server) fetchFeed(ctx context.Context, platform platform, uri string) ([]GetPostResponsePost, error) { + key := fmt.Sprintf("feed:%s|%s", platform, uri) + if v, ok := s.postCache.get(key); ok { + return v.([]GetPostResponsePost), nil + } + if platform == Bluesky { + res, err := s.bskyClient.GetFeed(ctx, uri, 50) + if err != nil { + return nil, err + } + posts := make([]GetPostResponsePost, 0, len(res.Feed)) + for _, p := range res.Feed { + if p.Post == nil { + continue + } + posts = append(posts, bskyPostViewToResponse(p.Post)) + } + s.postCache.set(key, posts) + return posts, nil + } + // Non-Bluesky platforms have no custom-feed concept; treat the URI as an author handle + // and return their latest posts as the feed. + posts, _, err := s.fetchLatestPosts(ctx, platform, uri, 50) + if err == nil { + s.postCache.set(key, posts) + } + return posts, err +} + +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 week (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*7*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-week 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), + }) + } + + respBytes, err := json.Marshal(FindSimilarUsersResponse{Matches: matches}) + 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) + } + + // 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 +} + +// StartPrewarm launches a background goroutine that discovers interesting authors and +// pre-fetches their style profiles. It issues at most one pre-fetch "request" per minute +// so upstream platforms, Ollama and the database are never hit harder than that cadence +// allows. Profiles are still only (re)written when older than a week, exactly like a live +// getStyleProfile request. Failures (including rate limits) are logged and retried on the +// next tick. +func (s *Server) StartPrewarm(ctx context.Context) { + if s.prewarmFeedURI == "" && len(s.prewarmAuthors) == 0 { + log.DebugWithContext(ctx, "no prewarm feed or authors configured, skipping prewarm") + return + } + + go func() { + authors := s.discoverInterestingAuthors(ctx) + if len(authors) == 0 { + return + } + s.prewarmOne(ctx, authors[0]) + ticker := time.NewTicker(time.Minute) + defer ticker.Stop() + i := 1 + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + if i >= len(authors) { + // re-discover interesting profiles and restart the cycle + authors = s.discoverInterestingAuthors(ctx) + i = 0 + if len(authors) == 0 { + continue + } + } + author := authors[i] + i++ + s.prewarmOne(ctx, author) + } + } + }() +} + +// discoverInterestingAuthors returns a list of authors worth pre-fetching. When a feed URI +// is configured it pulls recent posts from that feed and collects the author DIDs; +// otherwise it falls back to the statically configured seed list. +func (s *Server) discoverInterestingAuthors(ctx context.Context) []string { + if s.prewarmFeedURI != "" { + feed, err := s.bskyClient.GetFeed(ctx, s.prewarmFeedURI, 50) + if err != nil { + log.WarnWithContext(ctx, "prewarm: failed to discover authors from feed", "error", err) + } else { + seen := make(map[string]struct{}) + authors := make([]string, 0, len(feed.Feed)) + for _, p := range feed.Feed { + if p.Post == nil || p.Post.Author == nil { + continue + } + did := p.Post.Author.Did + if did == "" { + continue + } + if _, ok := seen[did]; ok { + continue + } + seen[did] = struct{}{} + authors = append(authors, did) + } + if len(authors) > 0 { + return authors + } + } + } + return s.prewarmAuthors +} + +func (s *Server) prewarmOne(ctx context.Context, author string) { + if _, err := s.requestStyleProfile(ctx, author); err != nil { + log.WarnWithContext(ctx, "prewarm: failed to build profile", "author", author, "error", err) + return + } + log.InfoWithContext(ctx, "prewarmed style profile", "author", author) +} + +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 = make([]clustering.Item, 0, len(res)) + for _, p := range res { + fp, ok := p.Post.Record.Val.(*bsky.FeedPost) + if !ok { + continue + } + freshPosts = append(freshPosts, clustering.Item{ + ID: p.Post.Uri, + Text: fp.Text, + }) + } + case ActivityPub: + res, err := s.apClient.GetLatestNPosts(ctx, author, limit) + if err != nil { + return toItems(cached), err + } + freshPosts = make([]clustering.Item, 0, len(res)) + for _, p := range res { + if p.Text != "" { + freshPosts = append(freshPosts, clustering.Item{ + ID: p.ID, + Text: p.Text, + }) + } + } + case Nostr: + res, err := s.nostrClient.GetLatestNPosts(ctx, author, limit) + if err != nil { + return toItems(cached), err + } + freshPosts = make([]clustering.Item, 0, len(res)) + for _, p := range res { + if p.Content != "" { + freshPosts = append(freshPosts, clustering.Item{ + ID: p.ID, + Text: p.Content, + }) + } + } + case Twitter: + res, err := s.twitterClient.GetLatestNPosts(ctx, author, limit) + if err != nil { + return toItems(cached), err + } + freshPosts = make([]clustering.Item, 0, len(res)) + for _, p := range res { + if p.Content != "" { + freshPosts = append(freshPosts, clustering.Item{ + ID: p.ID, + Text: p.Content, + }) + } + } + 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 +} + +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 { go s.startPostPruner() return s.mcpHandler.Start(s.listenAddress, s.permissions) } + +func (s *Server) startPostPruner() { + ticker := time.NewTicker(24 * time.Hour) + defer ticker.Stop() + for range ticker.C { + if err := s.db.PruneStylePosts(context.Background()); err != nil { + log.Error("failed to prune style posts", "error", err) + } + } +} diff --git a/internal/server/server_init.go b/internal/server/server_init.go deleted file mode 100644 index 679492a..0000000 --- a/internal/server/server_init.go +++ /dev/null @@ -1,118 +0,0 @@ -package server - -import ( - "context" - "encoding/json" - "net/http" - "time" - - "github.com/modelcontextprotocol/go-sdk/mcp" - log "github.com/nacorid/logger" - "github.com/nacorid/naco-api/internal/config" - "github.com/nacorid/naco-api/internal/store" - "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" -) - -// buildMiddleware creates the shared middleware chain and its stateful components. -func buildMiddleware() (x402.X402Middlewares, *MetricsMiddleware) { - 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, - }, - } - return toolchain, metrics -} - -// resolveActiveNetworks parses ACTIVE_NETWORKS and resolves them against NetworkRegistry. -func resolveActiveNetworks(ctx context.Context, raw string) []NetworkInfo { - names := parseActiveNetworks(raw) - var out []NetworkInfo - for _, name := range names { - if info, ok := NetworkRegistry[name]; ok { - out = append(out, info) - } else { - log.WarnWithContext(ctx, "configured network is not defined in the registry", "name", name) - } - } - return out -} - -// newX402Server creates the x402-enabled MCP server with settlement and webhook hooks. -// db is the Store used to persist settlements; cfg provides wallet/webhook settings. -func newX402Server(ctx context.Context, db *store.Queries, cfg *config.Config, toolchain x402.X402Middlewares, metrics *MetricsMiddleware) *x402.X402Server { - 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 - } - persistSettlement(r.Context(), db, resp.Transaction, requirements.Amount, requirements.Asset, string(resp.Network), requirements.PayTo, resp.Payer) - }, - ) - if err != nil { - log.FatalWithContext(ctx, "failed to create x402 server", "error", err) - } - - srv.OnAfterSettle(func(settleCtx x402v2.SettleResultContext) error { - persistSettlement(settleCtx.Ctx, db, settleCtx.Result.Transaction, - settleCtx.Requirements.GetAmount(), settleCtx.Requirements.GetAsset(), - string(settleCtx.Result.Network), settleCtx.Requirements.GetPayTo(), settleCtx.Result.Payer) - - // Webhook is best-effort and non-blocking. - go sendSettlementWebhook(settleCtx.Ctx, cfg, metrics, settleCtx.Result.Transaction) - return nil - }) - - return srv -} - -// persistSettlement is a thin wrapper around settleHook to keep call sites uniform. -func persistSettlement(ctx context.Context, db *store.Queries, transaction, amount, asset, network, to, from string) { - _ = settleHook(ctx, db, transaction, amount, asset, network, to, from) -} - -// sendSettlementWebhook builds and sends the Matrix webhook payload for a settled transaction. -func sendSettlementWebhook(ctx context.Context, cfg *config.Config, metrics *MetricsMiddleware, txHash string) { - wh := newWebhookSender(cfg, metrics) - if wh == nil { - return - } - payload := map[string]any{ - "event": "transaction_success", - "transaction": map[string]any{ - "x402_transaction_hash": txHash, - }, - "metrics": map[string]any{ - "function_runtime_ms": 0, - "platform": "unknown", - }, - "timestamp": time.Now().UTC().Format(time.RFC3339), - "server_id": cfg.MatrixServerID, - } - jsonBytes, _ := json.Marshal(payload) - wh.Send(ctx, jsonBytes) -} diff --git a/internal/server/settlement.go b/internal/server/settlement.go deleted file mode 100644 index 3f8e9b0..0000000 --- a/internal/server/settlement.go +++ /dev/null @@ -1,78 +0,0 @@ -package server - -import ( - "context" - "maps" - "strings" - - log "github.com/nacorid/logger" - "github.com/nacorid/naco-api/internal/store" - x402v2 "github.com/x402-foundation/x402/go/v2" -) - -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 -} diff --git a/internal/server/style.go b/internal/server/style.go deleted file mode 100644 index 5f9e217..0000000 --- a/internal/server/style.go +++ /dev/null @@ -1,336 +0,0 @@ -package server - -import ( - "context" - "database/sql" - "encoding/json" - "errors" - "fmt" - "sync" - "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" -) - -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 week (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*7*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-week 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), - }) - } - - respBytes, err := json.Marshal(FindSimilarUsersResponse{Matches: matches}) - 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) - } - - // 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 -} diff --git a/internal/server/tools.go b/internal/server/tools.go deleted file mode 100644 index 9f495c9..0000000 --- a/internal/server/tools.go +++ /dev/null @@ -1,454 +0,0 @@ -package server - -import ( - "context" - "encoding/json" - "fmt" - "time" - - "github.com/modelcontextprotocol/go-sdk/mcp" -) - -// registerTools registers all MCP tools with their x402 payment requirements. -// The payable tools are defined in a table to avoid repeating Schema construction. -func (s *Server) registerTools() { - // Helper to build the x402 extra map from input/output BodyFields. - schemaExtra := func(input, output map[string]any) map[string]any { - return map[string]any{ - "OutputSchema": Schema{ - Input: InputSchema{ - Type: "HTTP", - Method: "POST", - Discoverable: true, - BodyFields: input, - }, - Output: output, - }, - } - } - - analyticsInput := map[string]any{ - "author": Bodyfield{Type: "string", Description: "Did or handle of an author", Required: true}, - "startTime": Bodyfield{Type: "string", Description: "Start time in RFC3339 format (default: 30 days ago)", Required: false}, - "endTime": Bodyfield{Type: "string", Description: "End time in RFC3339 format (default: now)", Required: false}, - "limit": Bodyfield{Type: "integer", Description: "Maximum number of posts to analyze (default: 50, max: 200)", Required: false}, - "platform": Bodyfield{Type: "string", Description: "Platform to filter (default: all)", Required: false}, - } - - type payableDef struct { - Name string - Desc string - Input any - Output any - Handler func(context.Context, *mcp.CallToolRequest) (*mcp.CallToolResult, error) - Price string - Extra map[string]any - } - - payable := []payableDef{ - { - Name: "getPost", Desc: "Get a specific post from a specific actor", - Input: GetPostRequest{}, Output: GetPostResponsePost{}, Handler: s.getPost, Price: "0.01", - Extra: schemaExtra( - 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 (Bluesky only). Defaults to 'app.bsky.feed.post'", Required: false}, - }, - 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"}, - }, - ), - }, - { - Name: "getLatestPosts", Desc: "Get the latest n posts from a specific actor", - Input: GetLatestPostsRequest{}, Output: GetLatestPostsResponse{}, Handler: s.getLatestPosts, Price: "0.10", - Extra: schemaExtra( - 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}, - }, - 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"}, - }, - ), - }, - { - Name: "getPostsInTime", Desc: "Get posts from a specific actor in a time range", - Input: GetPostsInTimeRequest{}, Output: GetPostsInTimeResponse{}, Handler: s.getPostsInTime, Price: "0.10", - Extra: schemaExtra( - 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}, - }, - 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"}, - }, - ), - }, - { - Name: "getStyleProfile", Desc: "Get an actor's style Profile", - Input: GetStyleProfileRequest{}, Output: GetStyleProfileResponse{}, Handler: s.getStyleProfile, Price: "0.36", - Extra: schemaExtra( - map[string]any{ - "actor": Bodyfield{Type: "string", Description: "Did or handle of an actor", Required: true}, - }, - 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"}, - }, - ), - }, - { - Name: "findSimilarUsers", Desc: "Find authors with a similar writing style to the given author, across all platforms", - Input: FindSimilarUsersRequest{}, Output: FindSimilarUsersResponse{}, Handler: s.getSimilarProfiles, Price: "0.05", - Extra: schemaExtra( - map[string]any{ - "actor": Bodyfield{Type: "string", Description: "Did or handle of an author", Required: true}, - }, - map[string]any{ - "matches": Bodyfield{Type: "array", Description: "Ranked list of similar profiles (author, platform, similarity, top keywords, style summary, post count)"}, - }, - ), - }, - { - Name: "getActor", Desc: "Get an actor's public profile (bio, avatar, follower counts) on the platform inferred from the identifier", - Input: GetActorRequest{}, Output: GetActorResponse{}, Handler: s.getActor, Price: "0.01", - Extra: schemaExtra( - map[string]any{ - "author": Bodyfield{Type: "string", Description: "Did, handle, npub or @handle of an actor", Required: true}, - }, - map[string]any{ - "actor": Bodyfield{Type: "object", Description: "Profile of the actor (platform, handle, displayName, avatarUrl, bio, followersCount, followingCount, postCount, url)"}, - }, - ), - }, - { - Name: "searchActors", Desc: "Search for actors by handle/name/keyword across all platforms", - Input: SearchActorsRequest{}, Output: SearchActorsResponse{}, Handler: s.searchActors, Price: "0.01", - Extra: schemaExtra( - map[string]any{ - "query": Bodyfield{Type: "string", Description: "Search query (handle, name or keyword)", Required: true}, - "limit": Bodyfield{Type: "integer", Description: "Maximum number of results (default 10, max 50)", Required: false}, - }, - map[string]any{ - "results": Bodyfield{Type: "array", Description: "Ranked list of matching actor profiles across platforms"}, - }, - ), - }, - { - Name: "getEngagementMetrics", Desc: "Get time-series engagement metrics (likes, reposts, replies, quotes) per post", - Input: AnalyticsRequest{}, Output: EngagementMetricsResponse{}, Handler: s.getEngagementMetrics, Price: "0.15", - Extra: schemaExtra(analyticsInput, map[string]any{ - "author": Bodyfield{Type: "string", Description: "Author identifier"}, - "metrics": Bodyfield{Type: "array", Description: "Engagement metrics per post"}, - }), - }, - { - Name: "getPostingPatterns", Desc: "Get posting heatmap and patterns (hour-of-day x day-of-week)", - Input: AnalyticsRequest{}, Output: PostingPatternsResponse{}, Handler: s.getPostingPatterns, Price: "0.10", - Extra: schemaExtra(analyticsInput, map[string]any{ - "author": Bodyfield{Type: "string", Description: "Author identifier"}, - "heatmap": Bodyfield{Type: "array", Description: "24x7 matrix: hour (0-23) x day of week (0-6 = Sun-Sat)"}, - "peakHours": Bodyfield{Type: "array", Description: "Top 3 hours with most posts (0-23)"}, - "peakDays": Bodyfield{Type: "array", Description: "Top 3 days with most posts (0-6 = Sun-Sat)"}, - "avgPostsPerDay": Bodyfield{Type: "number", Description: "Average posts per day"}, - "totalPosts": Bodyfield{Type: "integer", Description: "Total posts in time range"}, - "postingDays": Bodyfield{Type: "integer", Description: "Number of days with at least one post"}, - }), - }, - { - Name: "getTopicModeling", Desc: "Get LLM-based topic extraction and clustering from posts", - Input: AnalyticsRequest{}, Output: TopicModelingResponse{}, Handler: s.getTopicModeling, Price: "0.20", - Extra: schemaExtra(analyticsInput, map[string]any{ - "author": Bodyfield{Type: "string", Description: "Author identifier"}, - "topics": Bodyfield{Type: "array", Description: "Identified topic clusters"}, - }), - }, - { - Name: "getSentimentTrends", Desc: "Get sentiment analysis and emotional arcs over time", - Input: AnalyticsRequest{}, Output: SentimentTrendsResponse{}, Handler: s.getSentimentTrends, Price: "0.15", - Extra: schemaExtra(analyticsInput, map[string]any{ - "author": Bodyfield{Type: "string", Description: "Author identifier"}, - "timeSeries": Bodyfield{Type: "array", Description: "Sentiment over time"}, - "overallSentiment": Bodyfield{Type: "object", Description: "Overall sentiment summary"}, - "dominantEmotions": Bodyfield{Type: "array", Description: "Most frequent emotions"}, - "sentimentShifts": Bodyfield{Type: "array", Description: "Significant sentiment changes"}, - }), - }, - { - Name: "getEngagementDrivers", Desc: "Get topics/keywords that correlate with high engagement", - Input: AnalyticsRequest{}, Output: EngagementDriversResponse{}, Handler: s.getEngagementDrivers, Price: "0.20", - Extra: schemaExtra(analyticsInput, map[string]any{ - "author": Bodyfield{Type: "string", Description: "Author identifier"}, - "drivers": Bodyfield{Type: "array", Description: "Topics/keywords driving engagement"}, - }), - }, - } - - for _, d := range payable { - s.mcpHandler.AddPayableTool( - mcp.Tool{Name: d.Name, Description: d.Desc, InputSchema: d.Input, OutputSchema: d.Output}, - d.Handler, - s.buildRequirements(d.Price, d.Extra)..., - ) - } - - // Free tool: getFeed - 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") - } - platform, err := determinePlatform(args.Author) - if err != nil { - return nil, err - } - if platform == Twitter { - return nil, fmt.Errorf("twitter_unavailable: X API policy prohibits reselling post data") - } - post, err := s.fetchSinglePost(ctx, platform, args.Author, args.RecordKey, args.Collection) - if err != nil { - return nil, err - } - respBytes, err := json.Marshal(post) - 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") - } - platform, err := determinePlatform(args.Author) - if err != nil { - return nil, err - } - if platform == Twitter { - return nil, fmt.Errorf("twitter_unavailable: X API policy prohibits reselling post data") - } - posts, authorID, err := s.fetchLatestPosts(ctx, platform, args.Author, args.Limit) - if err != nil { - return nil, err - } - if authorID == "" { - authorID = args.Author - } - resp := GetLatestPostsResponse{ - Author: authorID, - 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") - } - if args.Limit > 50 { - return nil, fmt.Errorf("n must be at most 50") - } - limit := args.Limit - 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("endTime must be in RFC3339 format") - } - if startTime.After(endTime) { - tmp := startTime - startTime = endTime - endTime = tmp - } - platform, err := determinePlatform(args.Author) - if err != nil { - return nil, err - } - if platform == Twitter { - return nil, fmt.Errorf("twitter_unavailable: X API policy prohibits reselling post data") - } - posts, authorID, err := s.fetchPostsInTime(ctx, platform, args.Author, startTime, endTime, limit) - if err != nil { - return nil, err - } - if authorID == "" { - authorID = args.Author - } - resp := &GetPostsInTimeResponse{ - Author: authorID, - 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") - } - platform, err := determinePlatform(args.URI) - if err != nil { - return nil, err - } - if platform == Twitter { - return nil, fmt.Errorf("twitter_unavailable: X API policy prohibits reselling feed data") - } - posts, err := s.fetchFeed(ctx, platform, args.URI) - if err != nil { - return nil, err - } - 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) getActor(ctx context.Context, request *mcp.CallToolRequest) (*mcp.CallToolResult, error) { - var args GetActorRequest - 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, handle, npub or @handle") - } - platform, err := determinePlatform(args.Author) - if err != nil { - return nil, err - } - var ( - prof *ActorProfile - e error - ) - switch platform { - case Bluesky: - prof, e = s.bskyClient.GetActor(ctx, args.Author) - case ActivityPub: - prof, e = s.apClient.GetActor(ctx, args.Author) - case Nostr: - prof, e = s.nostrClient.GetActor(ctx, args.Author) - case Twitter: - return nil, fmt.Errorf("twitter_unavailable: X API policy prohibits reselling user/profile data") - default: - return nil, fmt.Errorf("unsupported platform %q", platform) - } - if e != nil { - return nil, e - } - respBytes, err := json.Marshal(GetActorResponse{Actor: *prof}) - if err != nil { - return nil, err - } - return &mcp.CallToolResult{ - Content: []mcp.Content{ - &mcp.TextContent{Text: string(respBytes)}, - }, - }, nil -} - -func (s *Server) searchActors(ctx context.Context, request *mcp.CallToolRequest) (*mcp.CallToolResult, error) { - var args SearchActorsRequest - if err := json.Unmarshal(request.Params.Arguments, &args); err != nil { - return nil, fmt.Errorf("invalid arguments: %w", err) - } - if args.Query == "" { - return nil, fmt.Errorf("query must not be empty") - } - limit := args.Limit - if limit <= 0 { - limit = 10 - } - if limit > 50 { - limit = 50 - } - results, err := s.searchActorsAll(ctx, args.Query, limit) - if err != nil { - return nil, err - } - respBytes, err := json.Marshal(SearchActorsResponse{Results: results}) - if err != nil { - return nil, err - } - return &mcp.CallToolResult{ - Content: []mcp.Content{ - &mcp.TextContent{Text: string(respBytes)}, - }, - }, nil -} diff --git a/testClient/main.go b/testClient/main.go index c78abc5..c73d1a0 100644 --- a/testClient/main.go +++ b/testClient/main.go @@ -1,15 +1,22 @@ package main import ( + "context" + "encoding/json" "fmt" "io" "log" + "math/big" "net/http" "net/http/httputil" "os" "strings" "time" + "github.com/ethereum/go-ethereum" + "github.com/ethereum/go-ethereum/accounts/abi" + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/ethclient" "github.com/joho/godotenv" x402 "github.com/x402-foundation/x402/go/v2" @@ -18,7 +25,14 @@ import ( evmsigners "github.com/x402-foundation/x402/go/v2/signers/evm" ) -const USDCContractAddress = "0x036CbD53842c5426634e7929541eC2318f3dCF7e" +const ( + USDCContractAddress = "0x036CbD53842c5426634e7929541eC2318f3dCF7e" + BaseSepoliaRPC = "https://sepolia.base.org" + MinUSDCRequired = 10.0 + MinETHRequired = 0.01 +) + +var usdcABI = `[{"constant":true,"inputs":[{"name":"_owner","type":"address"}],"name":"balanceOf","outputs":[{"name":"balance","type":"uint256"}],"type":"function"}]` type LoggingRoundTripper struct { Proxied http.RoundTripper @@ -89,15 +103,172 @@ func getEnv(key, fallback string) string { return fallback } +func getUSDCBalance(client *ethclient.Client, address common.Address) (*big.Int, error) { + contractAddr := common.HexToAddress(USDCContractAddress) + parsedABI, err := abi.JSON(strings.NewReader(usdcABI)) + if err != nil { + return nil, fmt.Errorf("failed to parse USDC ABI: %w", err) + } + data, err := parsedABI.Pack("balanceOf", address) + if err != nil { + return nil, fmt.Errorf("failed to pack balanceOf call: %w", err) + } + msg := ethereum.CallMsg{ + To: &contractAddr, + Data: data, + } + result, err := client.CallContract(context.Background(), msg, nil) + if err != nil { + return nil, fmt.Errorf("failed to call balanceOf: %w", err) + } + unpacked, err := parsedABI.Unpack("balanceOf", result) + if err != nil { + return nil, fmt.Errorf("failed to unpack balance: %w", err) + } + if len(unpacked) == 0 { + return nil, fmt.Errorf("no balance returned") + } + balance, ok := unpacked[0].(*big.Int) + if !ok { + return nil, fmt.Errorf("unexpected balance type: %T", unpacked[0]) + } + return balance, nil +} + +func getETHBalance(client *ethclient.Client, address common.Address) (*big.Int, error) { + balance, err := client.BalanceAt(context.Background(), address, nil) + if err != nil { + return nil, fmt.Errorf("failed to get ETH balance: %w", err) + } + return balance, nil +} + +func requestFaucetFunds(faucetURL, address, token string) error { + payload := map[string]string{ + "address": address, + "token": token, + } + jsonData, err := json.Marshal(payload) + if err != nil { + return fmt.Errorf("failed to marshal faucet request: %w", err) + } + resp, err := http.Post(faucetURL, "application/json", strings.NewReader(string(jsonData))) + if err != nil { + return fmt.Errorf("faucet request failed: %w", err) + } + defer resp.Body.Close() + if resp.StatusCode >= 400 { + body, _ := io.ReadAll(resp.Body) + return fmt.Errorf("faucet returned error %d: %s", resp.StatusCode, string(body)) + } + return nil +} + +func ensureFundedWallet(privateKey, faucetURL, rpcURL string) error { + client, err := ethclient.Dial(rpcURL) + if err != nil { + return fmt.Errorf("failed to connect to RPC: %w", err) + } + defer client.Close() + + signer, err := evmsigners.NewClientSignerFromPrivateKey(privateKey) + if err != nil { + return fmt.Errorf("failed to create signer: %w", err) + } + addrStr := signer.Address() + address := common.HexToAddress(addrStr) + + log.Printf("Checking wallet balance for %s...", addrStr) + + usdcBalance, err := getUSDCBalance(client, address) + if err != nil { + return fmt.Errorf("failed to get USDC balance: %w", err) + } + usdcBalanceFloat := new(big.Float).Quo(new(big.Float).SetInt(usdcBalance), big.NewFloat(1e6)) + usdcFloat, _ := usdcBalanceFloat.Float64() + log.Printf("Current USDC balance: %.6f USDC", usdcFloat) + + ethBalance, err := getETHBalance(client, address) + if err != nil { + return fmt.Errorf("failed to get ETH balance: %w", err) + } + ethBalanceFloat := new(big.Float).Quo(new(big.Float).SetInt(ethBalance), big.NewFloat(1e18)) + ethFloat, _ := ethBalanceFloat.Float64() + log.Printf("Current ETH balance: %.6f ETH", ethFloat) + + needsFunding := false + if usdcFloat < MinUSDCRequired { + log.Printf("USDC balance %.6f below minimum %.2f, requesting funding...", usdcFloat, MinUSDCRequired) + needsFunding = true + } + if ethFloat < MinETHRequired { + log.Printf("ETH balance %.6f below minimum %.4f, requesting funding...", ethFloat, MinETHRequired) + needsFunding = true + } + + if needsFunding { + if faucetURL == "" { + return fmt.Errorf("FUNDING_FAUCET_URL not set but wallet needs funding") + } + if usdcFloat < MinUSDCRequired { + log.Printf("Requesting USDC from faucet...") + if err := requestFaucetFunds(faucetURL, addrStr, "USDC"); err != nil { + return fmt.Errorf("failed to request USDC from faucet: %w", err) + } + } + if ethFloat < MinETHRequired { + log.Printf("Requesting ETH from faucet...") + if err := requestFaucetFunds(faucetURL, addrStr, "ETH"); err != nil { + return fmt.Errorf("failed to request ETH from faucet: %w", err) + } + } + log.Printf("Waiting for funding to arrive...") + time.Sleep(30 * time.Second) + + usdcBalance, err = getUSDCBalance(client, address) + if err != nil { + return fmt.Errorf("failed to get USDC balance after funding: %w", err) + } + usdcBalanceFloat = new(big.Float).Quo(new(big.Float).SetInt(usdcBalance), big.NewFloat(1e6)) + usdcFloat, _ = usdcBalanceFloat.Float64() + log.Printf("New USDC balance: %.6f USDC", usdcFloat) + + ethBalance, err = getETHBalance(client, address) + if err != nil { + return fmt.Errorf("failed to get ETH balance after funding: %w", err) + } + ethBalanceFloat = new(big.Float).Quo(new(big.Float).SetInt(ethBalance), big.NewFloat(1e18)) + ethFloat, _ = ethBalanceFloat.Float64() + log.Printf("New ETH balance: %.6f ETH", ethFloat) + + if usdcFloat < MinUSDCRequired { + return fmt.Errorf("USDC balance still insufficient after funding: %.6f < %.2f", usdcFloat, MinUSDCRequired) + } + if ethFloat < MinETHRequired { + return fmt.Errorf("ETH balance still insufficient after funding: %.6f < %.4f", ethFloat, MinETHRequired) + } + } else { + log.Printf("Wallet has sufficient funds") + } + return nil +} + func main() { godotenv.Load() privateKey := os.Getenv("PRIVATE_KEY") serverURL := getEnv("TEST_SERVER_URL", "http://localhost:8080") + faucetURL := getEnv("FUNDING_FAUCET_URL", "") + rpcURL := getEnv("RPC_URL", BaseSepoliaRPC) if privateKey == "" { log.Fatal("PRIVATE_KEY environment variable is required") } + log.Printf("Ensuring wallet is funded...") + if err := ensureFundedWallet(privateKey, faucetURL, rpcURL); err != nil { + log.Fatal("Failed to ensure wallet is funded: ", err) + } + signer, err := evmsigners.NewClientSignerFromPrivateKey(privateKey) if err != nil { log.Fatal("Failed to create EVM signer:", err)