Compare commits

..
Author SHA1 Message Date
a59cee1d2a
chore: cleanup manual funding scripts and docs
Some checks failed
Test Every Endpoint on Base Sepolia / test_endpoints (push) Failing after 1s
CI/CD Go Verification / build_and_test (push) Successful in 1m8s
- cdpScripts/listServices.py: fix filter to only api.naco.li resources
  (remove stray type != http check)
- cdpScripts/rescueFunds.py: add CDP export script for api-naco-li
- testClient/main.go: remove faucet/balance/funding logic (manual funding),
  drop go-ethereum abi/ethclient, big, json, context imports; keep
  x402 payment flow only
- AGENTS.md: document quick commands, architecture, codegen, migrations,
  testing patterns and gotchas
2026-09-03 23:59:22 +00:00
a218b8fbf5
refactor(server): split server.go into focused modules (phases 1-4)
Phase 1: mechanical splits
- platform.go: determinePlatform
- settlement.go: settleHook, buildRequirements, parseActiveNetworks
- posts_convert.go: bsky/ap/nostr/twitter single converters
- actors.go: searchActorsAll, dedupeActors, normalizeHandle
- posts.go: fetchSinglePost/fetchLatestPosts/fetchPostsInTime/fetchFeed
- style.go: getStyleProfile, requestStyleProfile*, enrichCrossPlatform,
  getSimilarProfiles, buildStyleProfile, fetchProfilePosts, toItems
- prewarm.go: StartPrewarm, discoverInterestingAuthors, prewarmOne, startPostPruner

Phase 2: deduplicate registerTools
- tools.go: table-driven payableDef + schemaExtra helper, analyticsInput reuse
- server.go 1814 -> 180 lines, registerTools 470 -> 150 lines

Phase 3: unify platform dispatch
- cache.go: generic cachedFetch[T]
- posts_convert.go: batch converters (blueskyFeedViewsToResponses,
  apPosts*ToResponses, nostr/twitter variants) and clustering Item
  variants (blueskyFeedViewsToAnalyticsItems etc.)
- posts.go: use cachedFetch + batch helpers, fetchFeed via helper
- style.go: fetchProfilePosts via *ToItems helpers
- analytics.go: fetchPostsInTimeForAnalytics/fetchTwitterFromAPI via analytics helpers

Phase 4: slim bootstrap
- server_init.go: buildMiddleware, resolveActiveNetworks, newX402Server,
  persistSettlement, sendSettlementWebhook (deduplicate settleHook)
- server.go: New 120 -> 32 lines, grouped Server fields by
  Transport/Dependencies/Cache/Payments, keep struct compat for tests
- verified: gofmt clean, go build/vet pass, go test ./... pass
2026-09-03 23:59:16 +00:00
cd630a4fad
ci: use self-contained docker and remove wallet funding
- runs-on node20 has node for checkout, use docker/setup-docker-action@v5
  for self-contained dockerd (no host dind dependency) instead of
  setup-compose only
- install: drop foundry/cast and bc (manual funding) -> go mod download only
- remove Fund Test/Server Wallet steps, FUNDING_FAUCET_URL env
- docker-compose -> docker compose via setup-docker, add logs on wait failures
2026-09-03 23:59:11 +00:00
17 changed files with 1788 additions and 2008 deletions

View file

@ -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
docker compose -f docker-compose.test.yml down -v || true

94
AGENTS.md Normal file
View file

@ -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`).

View file

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

25
cdpScripts/rescueFunds.py Normal file
View file

@ -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())

76
internal/server/actors.go Normal file
View file

@ -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
}

View file

@ -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

View file

@ -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

View file

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

144
internal/server/posts.go Normal file
View file

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

View file

@ -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
}

101
internal/server/prewarm.go Normal file
View file

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

File diff suppressed because it is too large Load diff

View file

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

View file

@ -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
}

336
internal/server/style.go Normal file
View file

@ -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
}

454
internal/server/tools.go Normal file
View file

@ -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
}

View file

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