diff --git a/.forgejo/workflows/test-endpoints.yaml b/.forgejo/workflows/test-endpoints.yaml index 1d93e50..89c7d27 100644 --- a/.forgejo/workflows/test-endpoints.yaml +++ b/.forgejo/workflows/test-endpoints.yaml @@ -23,7 +23,6 @@ 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 }} @@ -47,15 +46,11 @@ 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 - # 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 + run: go mod download - name: Verify Code Formatting run: | @@ -66,18 +61,19 @@ 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 @@ -90,32 +86,13 @@ 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 - - - 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)" + 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: Start Test Server run: | @@ -129,7 +106,6 @@ jobs: - name: Run Endpoint Tests env: RPC_URL: ${{ env.RPC_URL }} - FUNDING_FAUCET_URL: ${{ secrets.FUNDING_FAUCET_URL }} run: | ./test-endpoints @@ -139,4 +115,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 \ No newline at end of file + docker compose -f docker-compose.test.yml down -v || true diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000..a33550a --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,94 @@ +# 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 1a49e5f..63dd964 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", "") or getattr(item, "type", "") != "http" + if "api.naco.li" in getattr(item, "resource", "") ] print(f'First service found: {getattr(services.items[0], "resource", "N/A")}') diff --git a/cdpScripts/rescueFunds.py b/cdpScripts/rescueFunds.py new file mode 100644 index 0000000..18ef5fd --- /dev/null +++ b/cdpScripts/rescueFunds.py @@ -0,0 +1,25 @@ +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 new file mode 100644 index 0000000..724285c --- /dev/null +++ b/internal/server/actors.go @@ -0,0 +1,76 @@ +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 e34bc98..ea0faf7 100644 --- a/internal/server/analytics.go +++ b/internal/server/analytics.go @@ -11,7 +11,6 @@ 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" @@ -276,56 +275,19 @@ func (s *Server) fetchPostsInTimeForAnalytics(ctx context.Context, platform plat if err != nil { return nil, err } - 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) - } - } + items = blueskyFeedViewsToAnalyticsItems(res) case ActivityPub: res, err := s.apClient.GetPostsInTimeframe(ctx, author, start, end, limit) if err != nil { return nil, err } - 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), - }) - } - } + items = apPostsToAnalyticsItems(res) case Nostr: res, err := s.nostrClient.GetPostsInTimeframe(ctx, author, start, end, limit) if err != nil { return nil, err } - for _, p := range res { - if p.Content != "" { - items = append(items, clustering.Item{ - ID: p.ID, - Text: p.Content, - Timestamp: p.CreatedAt, - }) - } - } + items = nostrPostsToAnalyticsItems(res) case Twitter: // Twitter API calls are expensive (paid). Reuse persisted post data and // only call the API for time ranges that are not yet covered by the @@ -453,22 +415,7 @@ func (s *Server) fetchTwitterFromAPI(ctx context.Context, author string, start, if err != nil { return nil, err } - 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 + return twitterPostsToAnalyticsItems(res), nil } // mergeAnalyticsPosts combines cached posts in [start, end] with newly fetched diff --git a/internal/server/cache.go b/internal/server/cache.go index 10e656f..a9b5869 100644 --- a/internal/server/cache.go +++ b/internal/server/cache.go @@ -45,6 +45,24 @@ 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 new file mode 100644 index 0000000..6dcb2bf --- /dev/null +++ b/internal/server/platform.go @@ -0,0 +1,44 @@ +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 new file mode 100644 index 0000000..eed3144 --- /dev/null +++ b/internal/server/posts.go @@ -0,0 +1,144 @@ +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 new file mode 100644 index 0000000..181c9fc --- /dev/null +++ b/internal/server/posts_convert.go @@ -0,0 +1,252 @@ +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 new file mode 100644 index 0000000..55caf69 --- /dev/null +++ b/internal/server/prewarm.go @@ -0,0 +1,101 @@ +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 f2d1011..7e3d2a8 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -2,52 +2,45 @@ 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 { - 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 + // 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 x402Wallet string activeNetworks []NetworkInfo - enrichSem chan struct{} analyticsTTLEngagement time.Duration analyticsTTLPatterns time.Duration analyticsTTLTopics time.Duration @@ -56,95 +49,9 @@ 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 { - 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) - } - } + toolchain, metrics := buildMiddleware() + srv := newX402Server(ctx, db, cfg, toolchain, metrics) + activeNetworks := resolveActiveNetworks(ctx, cfg.ActiveNetworks) s := &Server{ listenAddress: cfg.ListenAddress, @@ -171,6 +78,12 @@ 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 @@ -183,1632 +96,7 @@ 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 new file mode 100644 index 0000000..679492a --- /dev/null +++ b/internal/server/server_init.go @@ -0,0 +1,118 @@ +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 new file mode 100644 index 0000000..3f8e9b0 --- /dev/null +++ b/internal/server/settlement.go @@ -0,0 +1,78 @@ +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 new file mode 100644 index 0000000..5f9e217 --- /dev/null +++ b/internal/server/style.go @@ -0,0 +1,336 @@ +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 new file mode 100644 index 0000000..9f495c9 --- /dev/null +++ b/internal/server/tools.go @@ -0,0 +1,454 @@ +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 c73d1a0..c78abc5 100644 --- a/testClient/main.go +++ b/testClient/main.go @@ -1,22 +1,15 @@ 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" @@ -25,14 +18,7 @@ import ( evmsigners "github.com/x402-foundation/x402/go/v2/signers/evm" ) -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"}]` +const USDCContractAddress = "0x036CbD53842c5426634e7929541eC2318f3dCF7e" type LoggingRoundTripper struct { Proxied http.RoundTripper @@ -103,172 +89,15 @@ 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)