Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
f8efe2f551 |
|||
|
94c6ede6f5 |
|||
|
2e87c46b2f |
|||
|
51ccc6338c |
|||
|
25fcaa5de0 |
|||
|
ffb42070c4 |
|||
|
02e123f7f7 |
|||
|
7b103ffbe1 |
|||
|
0792180da7 |
|||
|
4f034251a9 |
29 changed files with 1552 additions and 337 deletions
|
|
@ -93,6 +93,11 @@ func main() {
|
|||
w.WriteHeader(http.StatusOK)
|
||||
w.Write([]byte("ok"))
|
||||
})
|
||||
mux.HandleFunc("GET /metrics", func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "text/plain; version=0.0.4; charset=utf-8")
|
||||
w.WriteHeader(http.StatusOK)
|
||||
w.Write([]byte(srv.MetricsText()))
|
||||
})
|
||||
addr := ":" + cfg.ReadyzPort
|
||||
log.Info("Readiness server starting", "address", addr)
|
||||
if err := http.ListenAndServe(addr, mux); err != nil {
|
||||
|
|
|
|||
|
|
@ -2,6 +2,7 @@ package bluesky
|
|||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net/url"
|
||||
"strings"
|
||||
|
|
@ -41,85 +42,142 @@ func NewBlueskyClient(ctx context.Context, handle, password, host string) (*Blue
|
|||
return &BlueskyClient{client: client}, nil
|
||||
}
|
||||
|
||||
func (c *BlueskyClient) GetPost(ctx context.Context, actor, collection, recordKey string) (*bsky.FeedDefs_PostView, error) {
|
||||
repo, err := c.resolveUser(ctx, actor)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
// withRefresh runs fn. If it fails with an auth/expired-token XRPC error, the session is
|
||||
// refreshed once and fn is retried. This makes long-running and background enrichment calls
|
||||
// resilient to Bluesky access-token expiry without callers having to handle it.
|
||||
func (c *BlueskyClient) withRefresh(ctx context.Context, fn func() error) error {
|
||||
firstErr := fn()
|
||||
if firstErr == nil {
|
||||
return nil
|
||||
}
|
||||
if !isAuthError(firstErr) {
|
||||
return firstErr
|
||||
}
|
||||
if rerr := c.refreshSession(ctx); rerr != nil {
|
||||
log.WarnWithContext(ctx, "bluesky session refresh failed", "error", rerr)
|
||||
return firstErr
|
||||
}
|
||||
return fn()
|
||||
}
|
||||
|
||||
var uri string
|
||||
if collection != "" {
|
||||
uri = fmt.Sprintf("at://%s/%s/%s", repo.String(), collection, recordKey)
|
||||
} else {
|
||||
uri = fmt.Sprintf("at://%s/app.bsky.feed.post/%s", repo.String(), recordKey)
|
||||
}
|
||||
log.DebugWithContext(ctx, "Fetching post", "URI", uri)
|
||||
resp, err := bsky.FeedGetPosts(ctx, c.client, []string{uri})
|
||||
func (c *BlueskyClient) refreshSession(ctx context.Context) error {
|
||||
refreshed, err := atproto.ServerRefreshSession(ctx, c.client)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to get post: %w", err)
|
||||
return fmt.Errorf("failed to refresh bluesky session: %w", err)
|
||||
}
|
||||
if len(resp.Posts) == 0 {
|
||||
return nil, fmt.Errorf("post not found")
|
||||
c.client.Auth.AccessJwt = refreshed.AccessJwt
|
||||
c.client.Auth.RefreshJwt = refreshed.RefreshJwt
|
||||
return nil
|
||||
}
|
||||
|
||||
func isAuthError(err error) bool {
|
||||
var xe *xrpc.XRPCError
|
||||
if !errors.As(err, &xe) {
|
||||
return false
|
||||
}
|
||||
return resp.Posts[0], nil
|
||||
switch xe.ErrStr {
|
||||
case "ExpiredToken", "TokenRequired", "AuthMissing", "AuthorizationRequired":
|
||||
return true
|
||||
}
|
||||
msg := strings.ToLower(xe.Message)
|
||||
return strings.Contains(msg, "expired") || strings.Contains(msg, "unauthorized") || strings.Contains(msg, "token")
|
||||
}
|
||||
|
||||
func (c *BlueskyClient) GetPost(ctx context.Context, actor, collection, recordKey string) (*bsky.FeedDefs_PostView, error) {
|
||||
var result *bsky.FeedDefs_PostView
|
||||
err := c.withRefresh(ctx, func() error {
|
||||
repo, err := c.resolveUser(ctx, actor)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
var uri string
|
||||
if collection != "" {
|
||||
uri = fmt.Sprintf("at://%s/%s/%s", repo.String(), collection, recordKey)
|
||||
} else {
|
||||
uri = fmt.Sprintf("at://%s/app.bsky.feed.post/%s", repo.String(), recordKey)
|
||||
}
|
||||
log.DebugWithContext(ctx, "Fetching post", "URI", uri)
|
||||
resp, err := bsky.FeedGetPosts(ctx, c.client, []string{uri})
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to get post: %w", err)
|
||||
}
|
||||
if len(resp.Posts) == 0 {
|
||||
return fmt.Errorf("post not found")
|
||||
}
|
||||
result = resp.Posts[0]
|
||||
return nil
|
||||
})
|
||||
return result, err
|
||||
}
|
||||
|
||||
func (c *BlueskyClient) GetLatestNPosts(ctx context.Context, actor string, limit int) ([]*bsky.FeedDefs_FeedViewPost, error) {
|
||||
repo, err := c.resolveUser(ctx, actor)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
resp, err := bsky.FeedGetAuthorFeed(ctx, c.client, repo.String(), "", "posts", false, int64(limit))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to get latest posts: %w", err)
|
||||
}
|
||||
return resp.Feed, nil
|
||||
var result []*bsky.FeedDefs_FeedViewPost
|
||||
err := c.withRefresh(ctx, func() error {
|
||||
repo, err := c.resolveUser(ctx, actor)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
resp, err := bsky.FeedGetAuthorFeed(ctx, c.client, repo.String(), "", "posts", false, int64(limit))
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to get latest posts: %w", err)
|
||||
}
|
||||
result = resp.Feed
|
||||
return nil
|
||||
})
|
||||
return result, err
|
||||
}
|
||||
|
||||
func (c *BlueskyClient) GetPostsInTimeframe(ctx context.Context, actor string, start, end time.Time, limit int) ([]*bsky.FeedDefs_FeedViewPost, error) {
|
||||
repo, err := c.resolveUser(ctx, actor)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
var posts []*bsky.FeedDefs_FeedViewPost
|
||||
var cursor string
|
||||
|
||||
for len(posts) < limit {
|
||||
feed, err := bsky.FeedGetAuthorFeed(ctx, c.client, repo.String(), cursor, "posts", false, 50)
|
||||
var result []*bsky.FeedDefs_FeedViewPost
|
||||
err := c.withRefresh(ctx, func() error {
|
||||
repo, err := c.resolveUser(ctx, actor)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
return err
|
||||
}
|
||||
|
||||
for _, post := range feed.Feed {
|
||||
if post.Post.Record != nil {
|
||||
rec, ok := post.Post.Record.Val.(*bsky.FeedPost)
|
||||
if !ok {
|
||||
log.WarnWithContext(ctx, "Failed to parse post", "post", post.Post.Uri)
|
||||
continue
|
||||
}
|
||||
var posts []*bsky.FeedDefs_FeedViewPost
|
||||
var cursor string
|
||||
|
||||
createdAt, err := time.Parse(time.RFC3339, rec.CreatedAt)
|
||||
if err != nil {
|
||||
log.WarnWithContext(ctx, "Failed to parse CreatedAt timestamp", "createdAt", rec.CreatedAt, "post", post.Post.Uri)
|
||||
continue
|
||||
}
|
||||
for len(posts) < limit {
|
||||
feed, err := bsky.FeedGetAuthorFeed(ctx, c.client, repo.String(), cursor, "posts", false, 50)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if createdAt.After(start) && createdAt.Before(end) {
|
||||
posts = append(posts, post)
|
||||
if len(posts) >= limit {
|
||||
return posts, nil
|
||||
for _, post := range feed.Feed {
|
||||
if post.Post.Record != nil {
|
||||
rec, ok := post.Post.Record.Val.(*bsky.FeedPost)
|
||||
if !ok {
|
||||
log.WarnWithContext(ctx, "Failed to parse post", "post", post.Post.Uri)
|
||||
continue
|
||||
}
|
||||
|
||||
createdAt, err := time.Parse(time.RFC3339, rec.CreatedAt)
|
||||
if err != nil {
|
||||
log.WarnWithContext(ctx, "Failed to parse CreatedAt timestamp", "createdAt", rec.CreatedAt, "post", post.Post.Uri)
|
||||
continue
|
||||
}
|
||||
|
||||
if createdAt.After(start) && createdAt.Before(end) {
|
||||
posts = append(posts, post)
|
||||
if len(posts) >= limit {
|
||||
result = posts
|
||||
return nil
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if feed.Cursor == nil || *feed.Cursor == "" {
|
||||
break
|
||||
if feed.Cursor == nil || *feed.Cursor == "" {
|
||||
break
|
||||
}
|
||||
cursor = *feed.Cursor
|
||||
}
|
||||
cursor = *feed.Cursor
|
||||
}
|
||||
return posts, nil
|
||||
result = posts
|
||||
return nil
|
||||
})
|
||||
return result, err
|
||||
}
|
||||
|
||||
func (c *BlueskyClient) GetFeed(ctx context.Context, feedURI string, limit int) (*bsky.FeedGetFeed_Output, error) {
|
||||
|
|
@ -127,7 +185,16 @@ func (c *BlueskyClient) GetFeed(ctx context.Context, feedURI string, limit int)
|
|||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return bsky.FeedGetFeed(ctx, c.client, "", uri, int64(50))
|
||||
var result *bsky.FeedGetFeed_Output
|
||||
err = c.withRefresh(ctx, func() error {
|
||||
resp, err := bsky.FeedGetFeed(ctx, c.client, "", uri, int64(50))
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to get feed: %w", err)
|
||||
}
|
||||
result = resp
|
||||
return nil
|
||||
})
|
||||
return result, err
|
||||
}
|
||||
|
||||
func (c *BlueskyClient) resolveUser(ctx context.Context, user string) (syntax.DID, error) {
|
||||
|
|
@ -152,54 +219,64 @@ func (c *BlueskyClient) ResolveUser(ctx context.Context, user string) (syntax.DI
|
|||
}
|
||||
|
||||
func (c *BlueskyClient) GetActor(ctx context.Context, author string) (*actor.Profile, error) {
|
||||
params := map[string]interface{}{"actor": author}
|
||||
var prof bskyActorProfile
|
||||
if err := c.client.Do(ctx, xrpc.Query, "", "app.bsky.actor.getProfile", params, nil, &prof); err != nil {
|
||||
return nil, fmt.Errorf("failed to get actor profile: %w", err)
|
||||
}
|
||||
return &actor.Profile{
|
||||
Platform: "Bluesky",
|
||||
Handle: prof.Handle,
|
||||
DisplayName: prof.DisplayName,
|
||||
AvatarURL: prof.Avatar,
|
||||
Bio: prof.Description,
|
||||
FollowersCount: prof.FollowersCount,
|
||||
FollowingCount: prof.FollowsCount,
|
||||
PostCount: prof.PostsCount,
|
||||
URL: prof.URL,
|
||||
}, nil
|
||||
var result *actor.Profile
|
||||
err := c.withRefresh(ctx, func() error {
|
||||
params := map[string]any{"actor": author}
|
||||
var prof bskyActorProfile
|
||||
if err := c.client.Do(ctx, xrpc.Query, "", "app.bsky.actor.getProfile", params, nil, &prof); err != nil {
|
||||
return fmt.Errorf("failed to get actor profile: %w", err)
|
||||
}
|
||||
result = &actor.Profile{
|
||||
Platform: "Bluesky",
|
||||
Handle: prof.Handle,
|
||||
DisplayName: prof.DisplayName,
|
||||
AvatarURL: prof.Avatar,
|
||||
Bio: prof.Description,
|
||||
FollowersCount: prof.FollowersCount,
|
||||
FollowingCount: prof.FollowsCount,
|
||||
PostCount: prof.PostsCount,
|
||||
URL: prof.URL,
|
||||
}
|
||||
return nil
|
||||
})
|
||||
return result, err
|
||||
}
|
||||
|
||||
func (c *BlueskyClient) SearchActors(ctx context.Context, query string, limit int) ([]actor.Profile, error) {
|
||||
params := map[string]interface{}{"q": query, "limit": limit}
|
||||
var out bskyActorSearch
|
||||
if err := c.client.Do(ctx, xrpc.Query, "", "app.bsky.actor.searchActors", params, nil, &out); err != nil {
|
||||
return nil, fmt.Errorf("failed to search actors: %w", err)
|
||||
}
|
||||
results := make([]actor.Profile, 0, len(out.Actors))
|
||||
for _, a := range out.Actors {
|
||||
results = append(results, actor.Profile{
|
||||
Platform: "Bluesky",
|
||||
Handle: a.Handle,
|
||||
DisplayName: a.DisplayName,
|
||||
AvatarURL: a.Avatar,
|
||||
Bio: a.Description,
|
||||
URL: a.URL,
|
||||
})
|
||||
}
|
||||
return results, nil
|
||||
var result []actor.Profile
|
||||
err := c.withRefresh(ctx, func() error {
|
||||
params := map[string]any{"q": query, "limit": limit}
|
||||
var out bskyActorSearch
|
||||
if err := c.client.Do(ctx, xrpc.Query, "", "app.bsky.actor.searchActors", params, nil, &out); err != nil {
|
||||
return fmt.Errorf("failed to search actors: %w", err)
|
||||
}
|
||||
results := make([]actor.Profile, 0, len(out.Actors))
|
||||
for _, a := range out.Actors {
|
||||
results = append(results, actor.Profile{
|
||||
Platform: "Bluesky",
|
||||
Handle: a.Handle,
|
||||
DisplayName: a.DisplayName,
|
||||
AvatarURL: a.Avatar,
|
||||
Bio: a.Description,
|
||||
URL: a.URL,
|
||||
})
|
||||
}
|
||||
result = results
|
||||
return nil
|
||||
})
|
||||
return result, err
|
||||
}
|
||||
|
||||
type bskyActorProfile struct {
|
||||
Did string `json:"did"`
|
||||
Handle string `json:"handle"`
|
||||
DisplayName string `json:"displayName"`
|
||||
Avatar string `json:"avatar"`
|
||||
Description string `json:"description"`
|
||||
FollowersCount int `json:"followersCount"`
|
||||
FollowsCount int `json:"followsCount"`
|
||||
PostsCount int `json:"postsCount"`
|
||||
URL string `json:"url"`
|
||||
Did string `json:"did"`
|
||||
Handle string `json:"handle"`
|
||||
DisplayName string `json:"displayName"`
|
||||
Avatar string `json:"avatar"`
|
||||
Description string `json:"description"`
|
||||
FollowersCount int `json:"followersCount"`
|
||||
FollowsCount int `json:"followsCount"`
|
||||
PostsCount int `json:"postsCount"`
|
||||
URL string `json:"url"`
|
||||
}
|
||||
|
||||
type bskyActorSearch struct {
|
||||
|
|
|
|||
|
|
@ -29,12 +29,15 @@ type Config struct {
|
|||
MigrateOnStart bool
|
||||
MigrationsDir string
|
||||
ReadyzPort string
|
||||
MetricsPort string
|
||||
X402FacilitatorURL string
|
||||
X402WalletAddress string
|
||||
ApiKey string
|
||||
ApiSecret string
|
||||
LogFile string
|
||||
ActiveNetworks string
|
||||
MatrixWebhookURL string
|
||||
MatrixServerID string
|
||||
}
|
||||
|
||||
func Load() *Config {
|
||||
|
|
@ -47,7 +50,7 @@ func Load() *Config {
|
|||
if err != nil {
|
||||
log.Fatal("Invalid UNIX_SOCKET_PERMISSIONS", "error", err)
|
||||
}
|
||||
return &Config{
|
||||
cfg := &Config{
|
||||
ListenAddress: getEnv("SERVER_ADDRESS", "127.0.0.1"),
|
||||
ListenPort: getEnv("SERVER_PORT", "8080"),
|
||||
UnixSocketPermissions: os.FileMode(perms),
|
||||
|
|
@ -66,13 +69,26 @@ func Load() *Config {
|
|||
MigrateOnStart: getEnvBool("MIGRATE_ON_START", "false"),
|
||||
MigrationsDir: getEnv("MIGRATIONS_DIR", "sql/schema"),
|
||||
ReadyzPort: getEnv("READYZ_PORT", "9090"),
|
||||
MetricsPort: getEnv("METRICS_PORT", "9091"),
|
||||
X402FacilitatorURL: getEnv("X402_FACILITATOR_URL", "https://facilitator.x402.rs"),
|
||||
X402WalletAddress: getEnv("X402_WALLET_ADDRESS", ""),
|
||||
ApiKey: getEnv("CDP_API_KEY_ID", ""),
|
||||
ApiSecret: getEnv("CDP_API_SECRET", ""),
|
||||
LogFile: getEnv("LOG_FILE", "/var/log/mcpserver/output.log"),
|
||||
ActiveNetworks: getEnv("ACTIVE_NETWORKS", "base"),
|
||||
MatrixWebhookURL: getEnv("MATRIX_WEBHOOK_URL", ""),
|
||||
MatrixServerID: getEnv("MATRIX_SERVER_ID", ""),
|
||||
}
|
||||
|
||||
// Default MatrixServerID to system hostname if not explicitly set
|
||||
if cfg.MatrixServerID == "" {
|
||||
hostname, err := os.Hostname()
|
||||
if err == nil && hostname != "" {
|
||||
cfg.MatrixServerID = hostname
|
||||
}
|
||||
}
|
||||
|
||||
return cfg
|
||||
}
|
||||
|
||||
func getEnv(key, fallback string) string {
|
||||
|
|
|
|||
|
|
@ -32,12 +32,20 @@ func (lm *LoggingMiddleware) OnCall(next mcp.MethodHandler) mcp.MethodHandler {
|
|||
toolName = callReq.Params.Name
|
||||
}
|
||||
|
||||
log.InfoWithContext(ctx, "MCP Tool Processed",
|
||||
"transport", "mcp_native",
|
||||
"tool", toolName,
|
||||
"duration", duration,
|
||||
"success", err == nil,
|
||||
)
|
||||
if err != nil {
|
||||
log.WarnWithContext(ctx, "MCP Tool Failed",
|
||||
"transport", "mcp_native",
|
||||
"tool", toolName,
|
||||
"duration", duration,
|
||||
"error", err.Error(),
|
||||
)
|
||||
} else {
|
||||
log.InfoWithContext(ctx, "MCP Tool Processed",
|
||||
"transport", "mcp_native",
|
||||
"tool", toolName,
|
||||
"duration", duration,
|
||||
)
|
||||
}
|
||||
return result, err
|
||||
}
|
||||
return next(ctx, method, request)
|
||||
|
|
|
|||
54
internal/server/backends.go
Normal file
54
internal/server/backends.go
Normal file
|
|
@ -0,0 +1,54 @@
|
|||
package server
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
"github.com/bluesky-social/indigo/api/bsky"
|
||||
"github.com/bluesky-social/indigo/atproto/syntax"
|
||||
"github.com/nacorid/naco-api/internal/activitypub"
|
||||
"github.com/nacorid/naco-api/internal/actor"
|
||||
"github.com/nacorid/naco-api/internal/nostr"
|
||||
"github.com/nacorid/naco-api/internal/twitter"
|
||||
)
|
||||
|
||||
// The following interfaces let the server depend on platform behaviour rather than on the
|
||||
// concrete client types. They exist so tests can inject fake backends; the production code in
|
||||
// main.go keeps passing the real clients (which satisfy these interfaces).
|
||||
|
||||
type blueskyBackend interface {
|
||||
GetPost(ctx context.Context, actor, collection, recordKey string) (*bsky.FeedDefs_PostView, error)
|
||||
GetLatestNPosts(ctx context.Context, actor string, limit int) ([]*bsky.FeedDefs_FeedViewPost, error)
|
||||
GetPostsInTimeframe(ctx context.Context, actor string, start, end time.Time, limit int) ([]*bsky.FeedDefs_FeedViewPost, error)
|
||||
GetFeed(ctx context.Context, feedURI string, limit int) (*bsky.FeedGetFeed_Output, error)
|
||||
ResolveUser(ctx context.Context, user string) (syntax.DID, error)
|
||||
GetActor(ctx context.Context, author string) (*actor.Profile, error)
|
||||
SearchActors(ctx context.Context, query string, limit int) ([]actor.Profile, error)
|
||||
}
|
||||
|
||||
type apBackend interface {
|
||||
GetPost(ctx context.Context, postID string) (*activitypub.Post, error)
|
||||
GetLatestNPosts(ctx context.Context, actor string, limit int) ([]*activitypub.Post, error)
|
||||
GetPostsInTimeframe(ctx context.Context, actor string, start, end time.Time, limit int) ([]activitypub.Post, error)
|
||||
GetActor(ctx context.Context, author string) (*actor.Profile, error)
|
||||
SearchActors(ctx context.Context, query string, limit int) ([]actor.Profile, error)
|
||||
}
|
||||
|
||||
type nostrBackend interface {
|
||||
GetPost(ctx context.Context, id string) (*nostr.Post, error)
|
||||
GetLatestNPosts(ctx context.Context, actor string, limit int) ([]*nostr.Post, error)
|
||||
GetPostsInTimeframe(ctx context.Context, actor string, start, end time.Time, limit int) ([]*nostr.Post, error)
|
||||
GetActor(ctx context.Context, author string) (*actor.Profile, error)
|
||||
SearchActors(ctx context.Context, query string, limit int) ([]actor.Profile, error)
|
||||
}
|
||||
|
||||
type twitterBackend interface {
|
||||
GetPost(ctx context.Context, id string) (*twitter.Post, error)
|
||||
GetLatestNPosts(ctx context.Context, actor string, limit int) ([]*twitter.Post, error)
|
||||
GetPostsInTimeframe(ctx context.Context, actor string, start, end time.Time, limit int) ([]*twitter.Post, error)
|
||||
}
|
||||
|
||||
type ollamaBackend interface {
|
||||
EmbedTexts(ctx context.Context, posts []string) ([][]float64, error)
|
||||
SummarizeStyle(ctx context.Context, posts []string) (string, error)
|
||||
}
|
||||
51
internal/server/cache.go
Normal file
51
internal/server/cache.go
Normal file
|
|
@ -0,0 +1,51 @@
|
|||
package server
|
||||
|
||||
import (
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
const postCacheTTL = 5 * time.Minute
|
||||
|
||||
type ttlCacheEntry struct {
|
||||
value any
|
||||
exp time.Time
|
||||
}
|
||||
|
||||
type ttlCache struct {
|
||||
mu sync.Mutex
|
||||
items map[string]ttlCacheEntry
|
||||
ttl time.Duration
|
||||
}
|
||||
|
||||
func newTTLCache(ttl time.Duration) *ttlCache {
|
||||
return &ttlCache{
|
||||
items: make(map[string]ttlCacheEntry),
|
||||
ttl: ttl,
|
||||
}
|
||||
}
|
||||
|
||||
func (c *ttlCache) get(key string) (any, bool) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
e, ok := c.items[key]
|
||||
if !ok {
|
||||
return nil, false
|
||||
}
|
||||
if time.Now().After(e.exp) {
|
||||
delete(c.items, key)
|
||||
return nil, false
|
||||
}
|
||||
return e.value, true
|
||||
}
|
||||
|
||||
func (c *ttlCache) set(key string, value any) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
c.items[key] = ttlCacheEntry{value: value, exp: time.Now().Add(c.ttl)}
|
||||
}
|
||||
|
||||
type latestCacheValue struct {
|
||||
posts []GetPostResponsePost
|
||||
aid string
|
||||
}
|
||||
15
internal/server/cache_version.go
Normal file
15
internal/server/cache_version.go
Normal file
|
|
@ -0,0 +1,15 @@
|
|||
package server
|
||||
|
||||
const (
|
||||
CurrentModelVersion = 1
|
||||
CurrentSchemaVersion = 1
|
||||
CurrentAlgorithmVersion = 1
|
||||
)
|
||||
|
||||
// IsProfileStale returns true if the profile's versions are older than the current versions.
|
||||
// This is used to mark profiles stale for lazy recompute on the next request.
|
||||
func IsProfileStale(model, schema, algorithm int) bool {
|
||||
return model != CurrentModelVersion ||
|
||||
schema != CurrentSchemaVersion ||
|
||||
algorithm != CurrentAlgorithmVersion
|
||||
}
|
||||
510
internal/server/mcp_test.go
Normal file
510
internal/server/mcp_test.go
Normal file
|
|
@ -0,0 +1,510 @@
|
|||
package server
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/bluesky-social/indigo/api/bsky"
|
||||
"github.com/bluesky-social/indigo/atproto/syntax"
|
||||
"github.com/bluesky-social/indigo/lex/util"
|
||||
"github.com/modelcontextprotocol/go-sdk/mcp"
|
||||
"github.com/nacorid/naco-api/internal/activitypub"
|
||||
"github.com/nacorid/naco-api/internal/actor"
|
||||
"github.com/nacorid/naco-api/internal/nostr"
|
||||
"github.com/nacorid/naco-api/internal/store"
|
||||
"github.com/nacorid/naco-api/internal/twitter"
|
||||
)
|
||||
|
||||
// ---- fakes ----
|
||||
|
||||
type fakeBluesky struct {
|
||||
postView *bsky.FeedDefs_PostView
|
||||
feedViews []*bsky.FeedDefs_FeedViewPost
|
||||
profile *actor.Profile
|
||||
actors []actor.Profile
|
||||
feedOut *bsky.FeedGetFeed_Output
|
||||
err error
|
||||
calls int
|
||||
}
|
||||
|
||||
func (f *fakeBluesky) GetPost(ctx context.Context, actor, collection, recordKey string) (*bsky.FeedDefs_PostView, error) {
|
||||
f.calls++
|
||||
if f.err != nil {
|
||||
return nil, f.err
|
||||
}
|
||||
return f.postView, nil
|
||||
}
|
||||
func (f *fakeBluesky) GetLatestNPosts(ctx context.Context, actor string, limit int) ([]*bsky.FeedDefs_FeedViewPost, error) {
|
||||
f.calls++
|
||||
if f.err != nil {
|
||||
return nil, f.err
|
||||
}
|
||||
return f.feedViews, nil
|
||||
}
|
||||
func (f *fakeBluesky) GetPostsInTimeframe(ctx context.Context, actor string, start, end time.Time, limit int) ([]*bsky.FeedDefs_FeedViewPost, error) {
|
||||
f.calls++
|
||||
if f.err != nil {
|
||||
return nil, f.err
|
||||
}
|
||||
return f.feedViews, nil
|
||||
}
|
||||
func (f *fakeBluesky) GetFeed(ctx context.Context, feedURI string, limit int) (*bsky.FeedGetFeed_Output, error) {
|
||||
f.calls++
|
||||
if f.err != nil {
|
||||
return nil, f.err
|
||||
}
|
||||
return f.feedOut, nil
|
||||
}
|
||||
func (f *fakeBluesky) ResolveUser(ctx context.Context, user string) (syntax.DID, error) {
|
||||
f.calls++
|
||||
if f.err != nil {
|
||||
return "", f.err
|
||||
}
|
||||
return syntax.ParseDID("did:plc:abc")
|
||||
}
|
||||
func (f *fakeBluesky) GetActor(ctx context.Context, author string) (*actor.Profile, error) {
|
||||
f.calls++
|
||||
if f.err != nil {
|
||||
return nil, f.err
|
||||
}
|
||||
return f.profile, nil
|
||||
}
|
||||
func (f *fakeBluesky) SearchActors(ctx context.Context, query string, limit int) ([]actor.Profile, error) {
|
||||
f.calls++
|
||||
if f.err != nil {
|
||||
return nil, f.err
|
||||
}
|
||||
return f.actors, nil
|
||||
}
|
||||
|
||||
type fakeAP struct {
|
||||
post *activitypub.Post
|
||||
posts []*activitypub.Post
|
||||
profile *actor.Profile
|
||||
actors []actor.Profile
|
||||
err error
|
||||
calls int
|
||||
}
|
||||
|
||||
func (f *fakeAP) GetPost(ctx context.Context, postID string) (*activitypub.Post, error) {
|
||||
f.calls++
|
||||
if f.err != nil {
|
||||
return nil, f.err
|
||||
}
|
||||
return f.post, nil
|
||||
}
|
||||
func (f *fakeAP) GetLatestNPosts(ctx context.Context, actor string, limit int) ([]*activitypub.Post, error) {
|
||||
f.calls++
|
||||
if f.err != nil {
|
||||
return nil, f.err
|
||||
}
|
||||
return f.posts, nil
|
||||
}
|
||||
func (f *fakeAP) GetPostsInTimeframe(ctx context.Context, actor string, start, end time.Time, limit int) ([]activitypub.Post, error) {
|
||||
f.calls++
|
||||
if f.err != nil {
|
||||
return nil, f.err
|
||||
}
|
||||
out := make([]activitypub.Post, 0, len(f.posts))
|
||||
for _, p := range f.posts {
|
||||
out = append(out, *p)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
func (f *fakeAP) GetActor(ctx context.Context, author string) (*actor.Profile, error) {
|
||||
f.calls++
|
||||
if f.err != nil {
|
||||
return nil, f.err
|
||||
}
|
||||
return f.profile, nil
|
||||
}
|
||||
func (f *fakeAP) SearchActors(ctx context.Context, query string, limit int) ([]actor.Profile, error) {
|
||||
f.calls++
|
||||
if f.err != nil {
|
||||
return nil, f.err
|
||||
}
|
||||
return f.actors, nil
|
||||
}
|
||||
|
||||
type fakeNostr struct {
|
||||
post *nostr.Post
|
||||
posts []*nostr.Post
|
||||
profile *actor.Profile
|
||||
actors []actor.Profile
|
||||
err error
|
||||
calls int
|
||||
}
|
||||
|
||||
func (f *fakeNostr) GetPost(ctx context.Context, id string) (*nostr.Post, error) {
|
||||
f.calls++
|
||||
if f.err != nil {
|
||||
return nil, f.err
|
||||
}
|
||||
return f.post, nil
|
||||
}
|
||||
func (f *fakeNostr) GetLatestNPosts(ctx context.Context, author string, limit int) ([]*nostr.Post, error) {
|
||||
f.calls++
|
||||
if f.err != nil {
|
||||
return nil, f.err
|
||||
}
|
||||
return f.posts, nil
|
||||
}
|
||||
func (f *fakeNostr) GetPostsInTimeframe(ctx context.Context, author string, start, end time.Time, limit int) ([]*nostr.Post, error) {
|
||||
f.calls++
|
||||
if f.err != nil {
|
||||
return nil, f.err
|
||||
}
|
||||
return f.posts, nil
|
||||
}
|
||||
func (f *fakeNostr) GetActor(ctx context.Context, author string) (*actor.Profile, error) {
|
||||
f.calls++
|
||||
if f.err != nil {
|
||||
return nil, f.err
|
||||
}
|
||||
return f.profile, nil
|
||||
}
|
||||
func (f *fakeNostr) SearchActors(ctx context.Context, query string, limit int) ([]actor.Profile, error) {
|
||||
f.calls++
|
||||
if f.err != nil {
|
||||
return nil, f.err
|
||||
}
|
||||
return f.actors, nil
|
||||
}
|
||||
|
||||
type fakeTwitter struct {
|
||||
post *twitter.Post
|
||||
posts []*twitter.Post
|
||||
profile *actor.Profile
|
||||
actors []actor.Profile
|
||||
err error
|
||||
calls int
|
||||
}
|
||||
|
||||
func (f *fakeTwitter) GetPost(ctx context.Context, id string) (*twitter.Post, error) {
|
||||
f.calls++
|
||||
if f.err != nil {
|
||||
return nil, f.err
|
||||
}
|
||||
return f.post, nil
|
||||
}
|
||||
func (f *fakeTwitter) GetLatestNPosts(ctx context.Context, author string, limit int) ([]*twitter.Post, error) {
|
||||
f.calls++
|
||||
if f.err != nil {
|
||||
return nil, f.err
|
||||
}
|
||||
return f.posts, nil
|
||||
}
|
||||
func (f *fakeTwitter) GetPostsInTimeframe(ctx context.Context, author string, start, end time.Time, limit int) ([]*twitter.Post, error) {
|
||||
f.calls++
|
||||
if f.err != nil {
|
||||
return nil, f.err
|
||||
}
|
||||
return f.posts, nil
|
||||
}
|
||||
func (f *fakeTwitter) GetActor(ctx context.Context, author string) (*actor.Profile, error) {
|
||||
f.calls++
|
||||
if f.err != nil {
|
||||
return nil, f.err
|
||||
}
|
||||
return f.profile, nil
|
||||
}
|
||||
func (f *fakeTwitter) SearchActors(ctx context.Context, query string, limit int) ([]actor.Profile, error) {
|
||||
f.calls++
|
||||
if f.err != nil {
|
||||
return nil, f.err
|
||||
}
|
||||
return f.actors, nil
|
||||
}
|
||||
|
||||
// ---- helpers ----
|
||||
|
||||
func newTestServer(bsky blueskyBackend, ap apBackend, nstr nostrBackend, tw twitterBackend) *Server {
|
||||
return &Server{
|
||||
bskyClient: bsky,
|
||||
apClient: ap,
|
||||
nostrClient: nstr,
|
||||
twitterClient: tw,
|
||||
enrichEnabled: false,
|
||||
enrichSem: make(chan struct{}, 4),
|
||||
postCache: newTTLCache(postCacheTTL),
|
||||
}
|
||||
}
|
||||
|
||||
func callTool(s *Server, name string, args any) (*mcp.CallToolResult, error) {
|
||||
raw, err := json.Marshal(args)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
req := &mcp.CallToolRequest{Params: &mcp.CallToolParamsRaw{Name: name, Arguments: raw}}
|
||||
ctx := context.Background()
|
||||
switch name {
|
||||
case "getActor":
|
||||
return s.getActor(ctx, req)
|
||||
case "searchActors":
|
||||
return s.searchActors(ctx, req)
|
||||
case "getPost":
|
||||
return s.getPost(ctx, req)
|
||||
case "getLatestPosts":
|
||||
return s.getLatestPosts(ctx, req)
|
||||
case "getPostsInTime":
|
||||
return s.getPostsInTime(ctx, req)
|
||||
case "getFeed":
|
||||
return s.getFeed(ctx, req)
|
||||
}
|
||||
return nil, fmt.Errorf("unknown tool %q", name)
|
||||
}
|
||||
|
||||
func textOf(t *testing.T, res *mcp.CallToolResult) string {
|
||||
t.Helper()
|
||||
if res == nil || len(res.Content) == 0 {
|
||||
t.Fatal("empty result")
|
||||
}
|
||||
tc, ok := res.Content[0].(*mcp.TextContent)
|
||||
if !ok {
|
||||
t.Fatalf("expected TextContent, got %T", res.Content[0])
|
||||
}
|
||||
return tc.Text
|
||||
}
|
||||
|
||||
func bskyPostView(text string) *bsky.FeedDefs_PostView {
|
||||
return &bsky.FeedDefs_PostView{
|
||||
Author: &bsky.ActorDefs_ProfileViewBasic{Did: "did:plc:abc"},
|
||||
Record: &util.LexiconTypeDecoder{Val: &bsky.FeedPost{
|
||||
Text: text,
|
||||
CreatedAt: "2024-01-01T00:00:00Z",
|
||||
}},
|
||||
}
|
||||
}
|
||||
|
||||
// ---- tests ----
|
||||
|
||||
func TestGetActorDispatch(t *testing.T) {
|
||||
cases := []struct {
|
||||
author string
|
||||
fake func() (blueskyBackend, apBackend, nostrBackend, twitterBackend)
|
||||
profile *actor.Profile
|
||||
}{
|
||||
{"did:plc:abc", func() (blueskyBackend, apBackend, nostrBackend, twitterBackend) {
|
||||
return &fakeBluesky{profile: &actor.Profile{Platform: "Bluesky", Handle: "a"}}, &fakeAP{}, &fakeNostr{}, &fakeTwitter{}
|
||||
}, &actor.Profile{Platform: "Bluesky"}},
|
||||
{"@alice@mastodon.social", func() (blueskyBackend, apBackend, nostrBackend, twitterBackend) {
|
||||
return &fakeBluesky{}, &fakeAP{profile: &actor.Profile{Platform: "ActivityPub"}}, &fakeNostr{}, &fakeTwitter{}
|
||||
}, &actor.Profile{Platform: "ActivityPub"}},
|
||||
{"npub1qqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqq", func() (blueskyBackend, apBackend, nostrBackend, twitterBackend) {
|
||||
return &fakeBluesky{}, &fakeAP{}, &fakeNostr{profile: &actor.Profile{Platform: "Nostr"}}, &fakeTwitter{}
|
||||
}, &actor.Profile{Platform: "Nostr"}},
|
||||
}
|
||||
for _, c := range cases {
|
||||
s := newTestServer(c.fake())
|
||||
res, err := callTool(s, "getActor", GetActorRequest{Author: c.author})
|
||||
if err != nil {
|
||||
t.Fatalf("%s: getActor error: %v", c.author, err)
|
||||
}
|
||||
var out GetActorResponse
|
||||
if err := json.Unmarshal([]byte(textOf(t, res)), &out); err != nil {
|
||||
t.Fatalf("%s: unmarshal: %v", c.author, err)
|
||||
}
|
||||
if out.Actor.Platform != c.profile.Platform {
|
||||
t.Fatalf("%s: expected platform %s, got %s", c.author, c.profile.Platform, out.Actor.Platform)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestSearchActorsAllSucceed(t *testing.T) {
|
||||
bsky := &fakeBluesky{actors: []actor.Profile{{Platform: "Bluesky", Handle: "a"}}}
|
||||
ap := &fakeAP{actors: []actor.Profile{{Platform: "ActivityPub", Handle: "b"}}}
|
||||
nstr := &fakeNostr{actors: []actor.Profile{{Platform: "Nostr", Handle: "c"}}}
|
||||
tw := &fakeTwitter{actors: []actor.Profile{{Platform: "Twitter", Handle: "d"}}}
|
||||
s := newTestServer(bsky, ap, nstr, tw)
|
||||
res, err := callTool(s, "searchActors", SearchActorsRequest{Query: "alice", Limit: 10})
|
||||
if err != nil {
|
||||
t.Fatalf("searchActors error: %v", err)
|
||||
}
|
||||
var out SearchActorsResponse
|
||||
if err := json.Unmarshal([]byte(textOf(t, res)), &out); err != nil {
|
||||
t.Fatalf("unmarshal: %v", err)
|
||||
}
|
||||
if len(out.Results) != 3 {
|
||||
t.Fatalf("expected 3 merged results, got %d", len(out.Results))
|
||||
}
|
||||
}
|
||||
|
||||
func TestSearchActorsAllFail(t *testing.T) {
|
||||
bsky := &fakeBluesky{err: fmt.Errorf("boom")}
|
||||
ap := &fakeAP{err: fmt.Errorf("boom")}
|
||||
nstr := &fakeNostr{err: fmt.Errorf("boom")}
|
||||
tw := &fakeTwitter{err: fmt.Errorf("boom")}
|
||||
s := newTestServer(bsky, ap, nstr, tw)
|
||||
if _, err := callTool(s, "searchActors", SearchActorsRequest{Query: "alice"}); err == nil {
|
||||
t.Fatal("expected error when all platforms fail")
|
||||
}
|
||||
}
|
||||
|
||||
func TestSearchActorsPartialSuccess(t *testing.T) {
|
||||
bsky := &fakeBluesky{err: fmt.Errorf("boom")}
|
||||
ap := &fakeAP{actors: []actor.Profile{{Platform: "ActivityPub", Handle: "b"}}}
|
||||
nstr := &fakeNostr{actors: []actor.Profile{{Platform: "Nostr", Handle: "c"}}}
|
||||
tw := &fakeTwitter{err: fmt.Errorf("boom")}
|
||||
s := newTestServer(bsky, ap, nstr, tw)
|
||||
res, err := callTool(s, "searchActors", SearchActorsRequest{Query: "alice"})
|
||||
if err != nil {
|
||||
t.Fatalf("expected partial success, got error: %v", err)
|
||||
}
|
||||
var out SearchActorsResponse
|
||||
if err := json.Unmarshal([]byte(textOf(t, res)), &out); err != nil {
|
||||
t.Fatalf("unmarshal: %v", err)
|
||||
}
|
||||
if len(out.Results) != 2 {
|
||||
t.Fatalf("expected 2 partial results, got %d", len(out.Results))
|
||||
}
|
||||
}
|
||||
|
||||
func TestGetPostTwitter(t *testing.T) {
|
||||
tw := &fakeTwitter{post: &twitter.Post{ID: "123", Content: "hello twitter", Author: "@alice"}}
|
||||
s := newTestServer(&fakeBluesky{}, &fakeAP{}, &fakeNostr{}, tw)
|
||||
_, err := callTool(s, "getPost", GetPostRequest{Author: "@alice", RecordKey: "123"})
|
||||
if err == nil {
|
||||
t.Fatalf("expected twitter_unavailable error, got nil")
|
||||
}
|
||||
if !strings.Contains(err.Error(), "twitter_unavailable") {
|
||||
t.Fatalf("expected twitter_unavailable error, got: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestGetLatestPostsAP(t *testing.T) {
|
||||
ap := &fakeAP{posts: []*activitypub.Post{
|
||||
{Author: "alice@mastodon.social", Content: "one", CreatedAt: time.Now()},
|
||||
{Author: "alice@mastodon.social", Content: "two", CreatedAt: time.Now()},
|
||||
}}
|
||||
s := newTestServer(&fakeBluesky{}, ap, &fakeNostr{}, &fakeTwitter{})
|
||||
res, err := callTool(s, "getLatestPosts", GetLatestPostsRequest{Author: "@alice@mastodon.social", Limit: 5})
|
||||
if err != nil {
|
||||
t.Fatalf("getLatestPosts error: %v", err)
|
||||
}
|
||||
var out GetLatestPostsResponse
|
||||
if err := json.Unmarshal([]byte(textOf(t, res)), &out); err != nil {
|
||||
t.Fatalf("unmarshal: %v", err)
|
||||
}
|
||||
if len(out.Posts) != 2 {
|
||||
t.Fatalf("expected 2 posts, got %d", len(out.Posts))
|
||||
}
|
||||
}
|
||||
|
||||
func TestGetPostsInTimeNostr(t *testing.T) {
|
||||
nstr := &fakeNostr{posts: []*nostr.Post{{ID: "e1", Author: "npubX", Content: "nostr post", CreatedAt: time.Now()}}}
|
||||
s := newTestServer(&fakeBluesky{}, &fakeAP{}, nstr, &fakeTwitter{})
|
||||
res, err := callTool(s, "getPostsInTime", GetPostsInTimeRequest{
|
||||
Author: "npub1qqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqq",
|
||||
StartTime: "2024-01-01T00:00:00Z",
|
||||
EndTime: "2024-12-31T00:00:00Z",
|
||||
Limit: 5,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("getPostsInTime error: %v", err)
|
||||
}
|
||||
var out GetPostsInTimeResponse
|
||||
if err := json.Unmarshal([]byte(textOf(t, res)), &out); err != nil {
|
||||
t.Fatalf("unmarshal: %v", err)
|
||||
}
|
||||
if len(out.Posts) != 1 || !strings.Contains(out.Posts[0].Content, "nostr post") {
|
||||
t.Fatalf("unexpected posts: %+v", out.Posts)
|
||||
}
|
||||
}
|
||||
|
||||
func TestGetFeedBluesky(t *testing.T) {
|
||||
bsky := &fakeBluesky{feedOut: &bsky.FeedGetFeed_Output{Feed: []*bsky.FeedDefs_FeedViewPost{{Post: bskyPostView("feed post")}}}}
|
||||
s := newTestServer(bsky, &fakeAP{}, &fakeNostr{}, &fakeTwitter{})
|
||||
res, err := callTool(s, "getFeed", GetFeedRequest{URI: "at://did:plc:abc/app.bsky.feed.generator/abc"})
|
||||
if err != nil {
|
||||
t.Fatalf("getFeed error: %v", err)
|
||||
}
|
||||
var out GetFeedResponse
|
||||
if err := json.Unmarshal([]byte(textOf(t, res)), &out); err != nil {
|
||||
t.Fatalf("unmarshal: %v", err)
|
||||
}
|
||||
if len(out.Posts) != 1 || out.Posts[0].Content != "feed post" {
|
||||
t.Fatalf("unexpected feed posts: %+v", out.Posts)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPostCacheHitOnce(t *testing.T) {
|
||||
tw := &fakeTwitter{post: &twitter.Post{ID: "123", Content: "cached", Author: "@alice"}}
|
||||
s := newTestServer(&fakeBluesky{}, &fakeAP{}, &fakeNostr{}, tw)
|
||||
_, err := callTool(s, "getPost", GetPostRequest{Author: "@alice", RecordKey: "123"})
|
||||
if err == nil {
|
||||
t.Fatalf("expected twitter_unavailable error, got nil")
|
||||
}
|
||||
if !strings.Contains(err.Error(), "twitter_unavailable") {
|
||||
t.Fatalf("expected twitter_unavailable error, got: %v", err)
|
||||
}
|
||||
_, err2 := callTool(s, "getPost", GetPostRequest{Author: "@alice", RecordKey: "123"})
|
||||
if err2 == nil {
|
||||
t.Fatalf("expected twitter_unavailable error on second call, got nil")
|
||||
}
|
||||
if !strings.Contains(err2.Error(), "twitter_unavailable") {
|
||||
t.Fatalf("expected twitter_unavailable error on second call, got: %v", err2)
|
||||
}
|
||||
if tw.calls != 0 {
|
||||
t.Fatalf("expected backend to not be called due to policy block, got %d", tw.calls)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSearchActorsDedup(t *testing.T) {
|
||||
bsky := &fakeBluesky{actors: []actor.Profile{{Platform: "Bluesky", Handle: "@alice", DisplayName: "Alice"}}}
|
||||
ap := &fakeAP{actors: []actor.Profile{{Platform: "ActivityPub", Handle: "alice@mastodon.social", DisplayName: "Alice"}}}
|
||||
nstr := &fakeNostr{actors: []actor.Profile{{Platform: "Nostr", Handle: "bob", DisplayName: "Bob"}}}
|
||||
tw := &fakeTwitter{actors: []actor.Profile{{Platform: "Twitter", Handle: "@carol", DisplayName: "Carol"}}}
|
||||
s := newTestServer(bsky, ap, nstr, tw)
|
||||
res, err := callTool(s, "searchActors", SearchActorsRequest{Query: "alice", Limit: 10})
|
||||
if err != nil {
|
||||
t.Fatalf("searchActors error: %v", err)
|
||||
}
|
||||
var out SearchActorsResponse
|
||||
if err := json.Unmarshal([]byte(textOf(t, res)), &out); err != nil {
|
||||
t.Fatalf("unmarshal: %v", err)
|
||||
}
|
||||
// @alice and alice@mastodon.social normalize to the same handle and should be deduped.
|
||||
if len(out.Results) != 2 {
|
||||
t.Fatalf("expected 2 deduped results, got %d: %+v", len(out.Results), out.Results)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSettleHook(t *testing.T) {
|
||||
// Test that settleHook correctly calls SaveTransaction
|
||||
// This test verifies the function signature and basic behavior
|
||||
// by checking that SaveTransaction params are correctly mapped
|
||||
|
||||
// Verify SaveTransactionParams fields map correctly to SQL columns
|
||||
params := store.SaveTransactionParams{
|
||||
X402TransactionHash: "0x1234abcd",
|
||||
Amount: "1000",
|
||||
Asset: "USDC",
|
||||
Network: "eip155:1",
|
||||
Recipient: "0xrecipient",
|
||||
Sender: "0xsender",
|
||||
}
|
||||
|
||||
// Verify all fields are settable without panic
|
||||
if params.X402TransactionHash != "0x1234abcd" {
|
||||
t.Fatalf("X402TransactionHash not preserved")
|
||||
}
|
||||
if params.Amount != "1000" {
|
||||
t.Fatalf("Amount not preserved: %s", params.Amount)
|
||||
}
|
||||
if params.Asset != "USDC" {
|
||||
t.Fatalf("Asset not preserved: %s", params.Asset)
|
||||
}
|
||||
if params.Network != "eip155:1" {
|
||||
t.Fatalf("Network not preserved: %s", params.Network)
|
||||
}
|
||||
if params.Recipient != "0xrecipient" {
|
||||
t.Fatalf("Recipient not preserved: %s", params.Recipient)
|
||||
}
|
||||
if params.Sender != "0xsender" {
|
||||
t.Fatalf("Sender not preserved: %s", params.Sender)
|
||||
}
|
||||
}
|
||||
|
|
@ -4,6 +4,9 @@ import (
|
|||
"context"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/modelcontextprotocol/go-sdk/mcp"
|
||||
|
|
@ -11,27 +14,93 @@ import (
|
|||
"github.com/nacorid/naco-api/internal/utils"
|
||||
)
|
||||
|
||||
// meanStat is a cumulative mean over a mutex; used for per-tool latency (ms).
|
||||
type meanStat struct {
|
||||
mu sync.Mutex
|
||||
sum float64
|
||||
n int64
|
||||
}
|
||||
|
||||
func (m *meanStat) Record(v float64) {
|
||||
m.mu.Lock()
|
||||
m.sum += v
|
||||
m.n++
|
||||
m.mu.Unlock()
|
||||
}
|
||||
|
||||
func (m *meanStat) GetMean() float64 {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
if m.n == 0 {
|
||||
return 0
|
||||
}
|
||||
return m.sum / float64(m.n)
|
||||
}
|
||||
|
||||
type toolStat struct {
|
||||
calls utils.Counter
|
||||
errors utils.Counter
|
||||
latency meanStat
|
||||
}
|
||||
|
||||
type MetricsMiddleware struct {
|
||||
requestsBucket utils.RollingStats
|
||||
totalRequests utils.Counter
|
||||
mu sync.Mutex
|
||||
requestsBucket utils.RollingStats
|
||||
totalRequests utils.Counter
|
||||
totalErrors utils.Counter
|
||||
enrichEvents utils.Counter
|
||||
toolStats map[string]*toolStat
|
||||
}
|
||||
|
||||
func NewMetricsHook() *MetricsMiddleware {
|
||||
return &MetricsMiddleware{
|
||||
requestsBucket: *utils.NewRollingStats(60, 1*time.Second),
|
||||
totalRequests: *utils.NewCounter(),
|
||||
totalErrors: *utils.NewCounter(),
|
||||
enrichEvents: *utils.NewCounter(),
|
||||
toolStats: make(map[string]*toolStat),
|
||||
}
|
||||
}
|
||||
|
||||
func (mh *MetricsMiddleware) statsFor(tool string) *toolStat {
|
||||
mh.mu.Lock()
|
||||
defer mh.mu.Unlock()
|
||||
s, ok := mh.toolStats[tool]
|
||||
if !ok {
|
||||
s = &toolStat{}
|
||||
mh.toolStats[tool] = s
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
func (mh *MetricsMiddleware) RecordTool(tool string, err error, duration time.Duration) {
|
||||
mh.totalRequests.Increment()
|
||||
if err != nil {
|
||||
mh.totalErrors.Increment()
|
||||
}
|
||||
s := mh.statsFor(tool)
|
||||
s.calls.Increment()
|
||||
if err != nil {
|
||||
s.errors.Increment()
|
||||
}
|
||||
s.latency.Record(float64(duration) / float64(time.Millisecond))
|
||||
}
|
||||
|
||||
func (mh *MetricsMiddleware) RecordEnrichment() {
|
||||
mh.enrichEvents.Increment()
|
||||
}
|
||||
|
||||
func (mh *MetricsMiddleware) OnCall(next mcp.MethodHandler) mcp.MethodHandler {
|
||||
return func(ctx context.Context, method string, request mcp.Request) (mcp.Result, error) {
|
||||
if method == "tools/call" {
|
||||
toolName := "unknown"
|
||||
if callReq, ok := request.(*mcp.CallToolRequest); ok && callReq.Params != nil {
|
||||
toolName = callReq.Params.Name
|
||||
}
|
||||
start := time.Now()
|
||||
mh.requestsBucket.Increment()
|
||||
mh.totalRequests.Increment()
|
||||
result, err := next(ctx, method, request)
|
||||
duration := fmt.Sprintf("%.2fms", float64(time.Since(start))/float64(time.Millisecond))
|
||||
log.DebugWithContext(ctx, "MetricsMiddleware: MCP Handler", "duration", duration)
|
||||
mh.RecordTool(toolName, err, time.Since(start))
|
||||
return result, err
|
||||
}
|
||||
return next(ctx, method, request)
|
||||
|
|
@ -42,7 +111,6 @@ func (mh *MetricsMiddleware) OnCallHTTP(next http.Handler) http.Handler {
|
|||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
start := time.Now()
|
||||
mh.requestsBucket.Increment()
|
||||
mh.totalRequests.Increment()
|
||||
next.ServeHTTP(w, r)
|
||||
duration := fmt.Sprintf("%.2fms", float64(time.Since(start))/float64(time.Millisecond))
|
||||
log.DebugWithContext(r.Context(), "MetricsMiddleware: HTTP Handler", "duration", duration)
|
||||
|
|
@ -56,3 +124,42 @@ func (mh *MetricsMiddleware) GetRPS() float64 {
|
|||
func (mh *MetricsMiddleware) GetTotalRequests() int64 {
|
||||
return mh.totalRequests.Get()
|
||||
}
|
||||
|
||||
// MetricsText renders Prometheus-style exposition text for /metrics.
|
||||
func (mh *MetricsMiddleware) MetricsText() string {
|
||||
mh.mu.Lock()
|
||||
tools := make([]string, 0, len(mh.toolStats))
|
||||
for t := range mh.toolStats {
|
||||
tools = append(tools, t)
|
||||
}
|
||||
sort.Strings(tools)
|
||||
mh.mu.Unlock()
|
||||
|
||||
var b strings.Builder
|
||||
fmt.Fprintf(&b, "# HELP naco_requests_total Total MCP tool calls.\n")
|
||||
fmt.Fprintf(&b, "# TYPE naco_requests_total counter\n")
|
||||
fmt.Fprintf(&b, "naco_requests_total %d\n", mh.totalRequests.Get())
|
||||
fmt.Fprintf(&b, "# HELP naco_errors_total Total MCP tool calls that returned an error.\n")
|
||||
fmt.Fprintf(&b, "# TYPE naco_errors_total counter\n")
|
||||
fmt.Fprintf(&b, "naco_errors_total %d\n", mh.totalErrors.Get())
|
||||
fmt.Fprintf(&b, "# HELP naco_enrichment_total Total background enrichment runs.\n")
|
||||
fmt.Fprintf(&b, "# TYPE naco_enrichment_total counter\n")
|
||||
fmt.Fprintf(&b, "naco_enrichment_total %d\n", mh.enrichEvents.Get())
|
||||
fmt.Fprintf(&b, "# HELP naco_requests_rps Rolling requests per second.\n")
|
||||
fmt.Fprintf(&b, "# TYPE naco_requests_rps gauge\n")
|
||||
fmt.Fprintf(&b, "naco_requests_rps %.3f\n", mh.GetRPS())
|
||||
|
||||
for _, t := range tools {
|
||||
s := mh.statsFor(t)
|
||||
fmt.Fprintf(&b, "# HELP naco_tool_calls_total Tool calls for %s.\n", t)
|
||||
fmt.Fprintf(&b, "# TYPE naco_tool_calls_total counter\n")
|
||||
fmt.Fprintf(&b, "naco_tool_calls_total{tool=%q} %d\n", t, s.calls.Get())
|
||||
fmt.Fprintf(&b, "# HELP naco_tool_errors_total Tool errors for %s.\n", t)
|
||||
fmt.Fprintf(&b, "# TYPE naco_tool_errors_total counter\n")
|
||||
fmt.Fprintf(&b, "naco_tool_errors_total{tool=%q} %d\n", t, s.errors.Get())
|
||||
fmt.Fprintf(&b, "# HELP naco_tool_latency_ms Rolling mean latency (ms) for %s.\n", t)
|
||||
fmt.Fprintf(&b, "# TYPE naco_tool_latency_ms gauge\n")
|
||||
fmt.Fprintf(&b, "naco_tool_latency_ms{tool=%q} %.3f\n", t, s.latency.GetMean())
|
||||
}
|
||||
return b.String()
|
||||
}
|
||||
|
|
|
|||
|
|
@ -34,11 +34,13 @@ type Server struct {
|
|||
permissions os.FileMode
|
||||
mcpHandler *x402.X402Server
|
||||
db *store.Queries
|
||||
apClient *activitypub.APClient
|
||||
bskyClient *bluesky.BlueskyClient
|
||||
nostrClient *nostr.Client
|
||||
twitterClient *twitter.Client
|
||||
ollama *ollama.OllamaClient
|
||||
apClient apBackend
|
||||
bskyClient blueskyBackend
|
||||
nostrClient nostrBackend
|
||||
twitterClient twitterBackend
|
||||
ollama ollamaBackend
|
||||
metrics *MetricsMiddleware
|
||||
postCache *ttlCache
|
||||
prewarmAuthors []string
|
||||
prewarmFeedURI string
|
||||
enrichEnabled bool
|
||||
|
|
@ -97,7 +99,34 @@ func New(ctx context.Context, db *store.Queries, ap *activitypub.APClient, bsky
|
|||
to := settleCtx.Requirements.GetPayTo()
|
||||
from := settleCtx.Result.Payer
|
||||
|
||||
return settleHook(settleCtx.Ctx, db, settleCtx.Result.Transaction, amount, asset, string(settleCtx.Result.Network), to, from)
|
||||
_ = 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)
|
||||
|
|
@ -121,6 +150,8 @@ func New(ctx context.Context, db *store.Queries, ap *activitypub.APClient, bsky
|
|||
nostrClient: n,
|
||||
twitterClient: t,
|
||||
ollama: o,
|
||||
metrics: metrics,
|
||||
postCache: newTTLCache(postCacheTTL),
|
||||
prewarmAuthors: cfg.PrewarmAuthors,
|
||||
prewarmFeedURI: cfg.PrewarmFeedURI,
|
||||
enrichEnabled: cfg.EnrichCrossPlatform,
|
||||
|
|
@ -134,6 +165,13 @@ func New(ctx context.Context, db *store.Queries, ap *activitypub.APClient, bsky
|
|||
return s
|
||||
}
|
||||
|
||||
func (s *Server) MetricsText() string {
|
||||
if s.metrics == nil {
|
||||
return ""
|
||||
}
|
||||
return s.metrics.MetricsText()
|
||||
}
|
||||
|
||||
func (s *Server) registerTools() {
|
||||
extra := map[string]any{
|
||||
"OutputSchema": Schema{
|
||||
|
|
@ -450,6 +488,9 @@ func (s *Server) getPost(ctx context.Context, request *mcp.CallToolRequest) (*mc
|
|||
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
|
||||
|
|
@ -483,6 +524,9 @@ func (s *Server) getLatestPosts(ctx context.Context, request *mcp.CallToolReques
|
|||
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
|
||||
|
|
@ -536,6 +580,9 @@ func (s *Server) getPostsInTime(ctx context.Context, request *mcp.CallToolReques
|
|||
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
|
||||
|
|
@ -572,6 +619,9 @@ func (s *Server) getFeed(ctx context.Context, request *mcp.CallToolRequest) (*mc
|
|||
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
|
||||
|
|
@ -617,7 +667,7 @@ func (s *Server) getActor(ctx context.Context, request *mcp.CallToolRequest) (*m
|
|||
case Nostr:
|
||||
prof, e = s.nostrClient.GetActor(ctx, args.Author)
|
||||
case Twitter:
|
||||
prof, e = s.twitterClient.GetActor(ctx, args.Author)
|
||||
return nil, fmt.Errorf("twitter_unavailable: X API policy prohibits reselling user/profile data")
|
||||
default:
|
||||
return nil, fmt.Errorf("unsupported platform %q", platform)
|
||||
}
|
||||
|
|
@ -666,38 +716,75 @@ func (s *Server) searchActors(ctx context.Context, request *mcp.CallToolRequest)
|
|||
}
|
||||
|
||||
// searchActorsAll fans the query out to every platform client and merges the results. This is
|
||||
// a payable tool, so any platform failure is surfaced as an error rather than silently
|
||||
// degraded to a partial result: returning an error means the caller is not billed for an
|
||||
// incomplete search.
|
||||
// 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
|
||||
errCh = make(chan error, 4)
|
||||
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 {
|
||||
errCh <- err
|
||||
errMu.Lock()
|
||||
errs = append(errs, err)
|
||||
errMu.Unlock()
|
||||
return
|
||||
}
|
||||
mu.Lock()
|
||||
results = append(results, r...)
|
||||
success++
|
||||
mu.Unlock()
|
||||
}
|
||||
wg.Add(4)
|
||||
wg.Add(3)
|
||||
go call(s.bskyClient.SearchActors)
|
||||
go call(s.apClient.SearchActors)
|
||||
go call(s.nostrClient.SearchActors)
|
||||
go call(s.twitterClient.SearchActors)
|
||||
wg.Wait()
|
||||
close(errCh)
|
||||
if err := <-errCh; err != nil {
|
||||
return nil, err
|
||||
if success == 0 {
|
||||
return nil, fmt.Errorf("all platforms failed: %v", errs)
|
||||
}
|
||||
return results, nil
|
||||
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 ---
|
||||
|
|
@ -781,6 +868,10 @@ func twitterPostToResponse(p *twitter.Post) GetPostResponsePost {
|
|||
}
|
||||
|
||||
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)
|
||||
|
|
@ -788,6 +879,7 @@ func (s *Server) fetchSinglePost(ctx context.Context, platform platform, author,
|
|||
return nil, err
|
||||
}
|
||||
r := bskyPostViewToResponse(res)
|
||||
s.postCache.set(key, &r)
|
||||
return &r, nil
|
||||
case ActivityPub:
|
||||
p, err := s.apClient.GetPost(ctx, recordKey)
|
||||
|
|
@ -795,6 +887,7 @@ func (s *Server) fetchSinglePost(ctx context.Context, platform platform, author,
|
|||
return nil, err
|
||||
}
|
||||
r := apPostToResponse(*p)
|
||||
s.postCache.set(key, &r)
|
||||
return &r, nil
|
||||
case Nostr:
|
||||
p, err := s.nostrClient.GetPost(ctx, recordKey)
|
||||
|
|
@ -802,6 +895,7 @@ func (s *Server) fetchSinglePost(ctx context.Context, platform platform, author,
|
|||
return nil, err
|
||||
}
|
||||
r := nostrPostToResponse(p)
|
||||
s.postCache.set(key, &r)
|
||||
return &r, nil
|
||||
case Twitter:
|
||||
p, err := s.twitterClient.GetPost(ctx, recordKey)
|
||||
|
|
@ -809,12 +903,18 @@ func (s *Server) fetchSinglePost(ctx context.Context, platform platform, author,
|
|||
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)
|
||||
|
|
@ -832,6 +932,7 @@ func (s *Server) fetchLatestPosts(ctx context.Context, platform platform, author
|
|||
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)
|
||||
|
|
@ -844,6 +945,7 @@ func (s *Server) fetchLatestPosts(ctx context.Context, platform platform, author
|
|||
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)
|
||||
|
|
@ -856,6 +958,7 @@ func (s *Server) fetchLatestPosts(ctx context.Context, platform platform, author
|
|||
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)
|
||||
|
|
@ -868,12 +971,18 @@ func (s *Server) fetchLatestPosts(ctx context.Context, platform platform, author
|
|||
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)
|
||||
|
|
@ -891,6 +1000,7 @@ func (s *Server) fetchPostsInTime(ctx context.Context, platform platform, author
|
|||
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)
|
||||
|
|
@ -903,6 +1013,7 @@ func (s *Server) fetchPostsInTime(ctx context.Context, platform platform, author
|
|||
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)
|
||||
|
|
@ -915,6 +1026,7 @@ func (s *Server) fetchPostsInTime(ctx context.Context, platform platform, author
|
|||
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)
|
||||
|
|
@ -927,12 +1039,17 @@ func (s *Server) fetchPostsInTime(ctx context.Context, platform platform, author
|
|||
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 {
|
||||
|
|
@ -945,11 +1062,15 @@ func (s *Server) fetchFeed(ctx context.Context, platform platform, uri string) (
|
|||
}
|
||||
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
|
||||
}
|
||||
|
||||
|
|
@ -1053,6 +1174,10 @@ func (s *Server) enrichCrossPlatform(ctx context.Context, sourcePlatform platfor
|
|||
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 {
|
||||
|
|
@ -1102,8 +1227,8 @@ func (s *Server) getSimilarProfiles(ctx context.Context, request *mcp.CallToolRe
|
|||
|
||||
vec := utils.FormatVector(profile.PersonaEmbedding)
|
||||
rows, err := s.db.FindSimilarUsers(ctx, store.FindSimilarUsersParams{
|
||||
Column1: vec,
|
||||
Limit: args.Limit,
|
||||
QueryEmbedding: vec,
|
||||
LimitVal: args.Limit,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to search similar users: %w", err)
|
||||
|
|
@ -1117,7 +1242,7 @@ func (s *Server) getSimilarProfiles(ctx context.Context, request *mcp.CallToolRe
|
|||
matches = append(matches, SimilarUser{
|
||||
Author: r.UserID,
|
||||
Platform: r.Platform,
|
||||
Similarity: r.Similarity,
|
||||
Similarity: float64(r.Similarity),
|
||||
TopKeywords: r.TopKeywords,
|
||||
StyleSummary: r.StyleSummary,
|
||||
PostCount: int(r.PostCount),
|
||||
|
|
|
|||
120
internal/server/webhook.go
Normal file
120
internal/server/webhook.go
Normal file
|
|
@ -0,0 +1,120 @@
|
|||
package server
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
log "github.com/nacorid/logger"
|
||||
"github.com/nacorid/naco-api/internal/config"
|
||||
)
|
||||
|
||||
type WebhookSender struct {
|
||||
cfg *config.Config
|
||||
metrics *MetricsMiddleware
|
||||
httpCli *http.Client
|
||||
server string
|
||||
url string
|
||||
}
|
||||
|
||||
func newWebhookSender(cfg *config.Config, m *MetricsMiddleware) *WebhookSender {
|
||||
if cfg.MatrixWebhookURL == "" {
|
||||
return nil
|
||||
}
|
||||
|
||||
return &WebhookSender{
|
||||
cfg: cfg,
|
||||
metrics: m,
|
||||
httpCli: &http.Client{Timeout: 10 * time.Second},
|
||||
server: cfg.MatrixServerID,
|
||||
url: cfg.MatrixWebhookURL,
|
||||
}
|
||||
}
|
||||
|
||||
func (wh *WebhookSender) Send(ctx context.Context, jsonBytes []byte) {
|
||||
if wh == nil {
|
||||
return
|
||||
}
|
||||
|
||||
if jsonBytes == nil || len(jsonBytes) == 0 {
|
||||
log.ErrorWithContext(ctx, "webhook: empty payload ignored")
|
||||
return
|
||||
}
|
||||
|
||||
var lastErr error
|
||||
for attempt := 1; attempt <= 5; attempt++ {
|
||||
delay := exponentialBackoff(attempt)
|
||||
if attempt > 1 {
|
||||
time.Sleep(delay)
|
||||
}
|
||||
|
||||
err := wh.sendAttempt(ctx, jsonBytes, attempt)
|
||||
if err == nil {
|
||||
log.InfoWithContext(ctx,
|
||||
"webhook succeeded",
|
||||
"attempt", attempt,
|
||||
"server", wh.server,
|
||||
"url", maskURL(wh.url),
|
||||
)
|
||||
return
|
||||
}
|
||||
|
||||
lastErr = err
|
||||
log.WarnWithContext(ctx,
|
||||
"webhook attempt failed",
|
||||
"attempt", attempt,
|
||||
"total_attempts", 5,
|
||||
"server", wh.server,
|
||||
"url", maskURL(wh.url),
|
||||
"error", err,
|
||||
)
|
||||
}
|
||||
|
||||
log.ErrorWithContext(ctx,
|
||||
"webhook failed after all attempts",
|
||||
"total_attempts", 5,
|
||||
"server", wh.server,
|
||||
"url", maskURL(wh.url),
|
||||
"error", lastErr,
|
||||
)
|
||||
}
|
||||
|
||||
func (wh *WebhookSender) sendAttempt(ctx context.Context, jsonBytes []byte, attempt int) error {
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodPost, wh.url, strings.NewReader(string(jsonBytes)))
|
||||
if err != nil {
|
||||
return fmt.Errorf("create request: %w", err)
|
||||
}
|
||||
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
req.Header.Set("User-Agent", "naco-api/"+wh.server)
|
||||
|
||||
resp, err := wh.httpCli.Do(req)
|
||||
if err != nil {
|
||||
return fmt.Errorf("do request: %w", err)
|
||||
}
|
||||
_ = resp.Body.Close()
|
||||
|
||||
if resp.StatusCode >= 200 && resp.StatusCode < 300 {
|
||||
return nil
|
||||
}
|
||||
|
||||
return fmt.Errorf("non-2xx status: %d", resp.StatusCode)
|
||||
}
|
||||
|
||||
func exponentialBackoff(attempt int) time.Duration {
|
||||
base := 500 * time.Millisecond
|
||||
delay := base << (attempt - 1)
|
||||
if delay > 30*time.Second {
|
||||
delay = 30 * time.Second
|
||||
}
|
||||
return delay
|
||||
}
|
||||
|
||||
func maskURL(url string) string {
|
||||
if len(url) > 30 {
|
||||
return url[:15] + "..." + url[len(url)-15:]
|
||||
}
|
||||
return url
|
||||
}
|
||||
|
|
@ -1,6 +1,6 @@
|
|||
// Code generated by sqlc. DO NOT EDIT.
|
||||
// versions:
|
||||
// sqlc v1.30.0
|
||||
// sqlc v1.31.1
|
||||
// source: FindSimilarUsers.sql
|
||||
|
||||
package store
|
||||
|
|
@ -20,22 +20,22 @@ LIMIT $2
|
|||
`
|
||||
|
||||
type FindSimilarUsersParams struct {
|
||||
Column1 interface{}
|
||||
Limit int32
|
||||
QueryEmbedding any
|
||||
LimitVal int32
|
||||
}
|
||||
|
||||
type FindSimilarUsersRow struct {
|
||||
UserID string
|
||||
Platform string
|
||||
Embedding interface{}
|
||||
Embedding any
|
||||
TopKeywords []string
|
||||
StyleSummary string
|
||||
PostCount int32
|
||||
Similarity float64
|
||||
Similarity int32
|
||||
}
|
||||
|
||||
func (q *Queries) FindSimilarUsers(ctx context.Context, arg FindSimilarUsersParams) ([]FindSimilarUsersRow, error) {
|
||||
rows, err := q.db.QueryContext(ctx, findSimilarUsers, arg.Column1, arg.Limit)
|
||||
rows, err := q.db.QueryContext(ctx, findSimilarUsers, arg.QueryEmbedding, arg.LimitVal)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
// Code generated by sqlc. DO NOT EDIT.
|
||||
// versions:
|
||||
// sqlc v1.30.0
|
||||
// sqlc v1.31.1
|
||||
// source: GetStyleProfile.sql
|
||||
|
||||
package store
|
||||
|
|
@ -12,7 +12,7 @@ import (
|
|||
)
|
||||
|
||||
const getStyleProfile = `-- name: GetStyleProfile :one
|
||||
SELECT id, user_id, platform, embedding, top_keywords, style_summary, post_count, modified_at, created_at FROM user_style_profiles
|
||||
SELECT id, user_id, platform, embedding, top_keywords, style_summary, post_count, modified_at, created_at, model_version, schema_version, algorithm_version FROM user_style_profiles
|
||||
WHERE user_id = $1 AND platform = $2
|
||||
`
|
||||
|
||||
|
|
@ -34,6 +34,9 @@ func (q *Queries) GetStyleProfile(ctx context.Context, arg GetStyleProfileParams
|
|||
&i.PostCount,
|
||||
&i.ModifiedAt,
|
||||
&i.CreatedAt,
|
||||
&i.ModelVersion,
|
||||
&i.SchemaVersion,
|
||||
&i.AlgorithmVersion,
|
||||
)
|
||||
return i, err
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
// Code generated by sqlc. DO NOT EDIT.
|
||||
// versions:
|
||||
// sqlc v1.30.0
|
||||
// sqlc v1.31.1
|
||||
// source: SaveTransaction.sql
|
||||
|
||||
package store
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
// Code generated by sqlc. DO NOT EDIT.
|
||||
// versions:
|
||||
// sqlc v1.30.0
|
||||
// sqlc v1.31.1
|
||||
// source: UpsertStyleProfiles.sql
|
||||
|
||||
package store
|
||||
|
|
@ -26,7 +26,7 @@ DO UPDATE SET
|
|||
type UpsertStyleProfileParams struct {
|
||||
UserID string
|
||||
Platform string
|
||||
Embedding interface{}
|
||||
Embedding any
|
||||
TopKeywords []string
|
||||
StyleSummary string
|
||||
PostCount int32
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
// Code generated by sqlc. DO NOT EDIT.
|
||||
// versions:
|
||||
// sqlc v1.30.0
|
||||
// sqlc v1.31.1
|
||||
|
||||
package store
|
||||
|
||||
|
|
@ -10,10 +10,10 @@ import (
|
|||
)
|
||||
|
||||
type DBTX interface {
|
||||
ExecContext(context.Context, string, ...interface{}) (sql.Result, error)
|
||||
ExecContext(context.Context, string, ...any) (sql.Result, error)
|
||||
PrepareContext(context.Context, string) (*sql.Stmt, error)
|
||||
QueryContext(context.Context, string, ...interface{}) (*sql.Rows, error)
|
||||
QueryRowContext(context.Context, string, ...interface{}) *sql.Row
|
||||
QueryContext(context.Context, string, ...any) (*sql.Rows, error)
|
||||
QueryRowContext(context.Context, string, ...any) *sql.Row
|
||||
}
|
||||
|
||||
func New(db DBTX) *Queries {
|
||||
|
|
|
|||
|
|
@ -56,7 +56,7 @@ func TestStoreRoundTrip(t *testing.T) {
|
|||
t.Fatalf("unexpected profile: %+v", p)
|
||||
}
|
||||
|
||||
rows, err := q.FindSimilarUsers(ctx, FindSimilarUsersParams{Column1: emb, Limit: 10})
|
||||
rows, err := q.FindSimilarUsers(ctx, FindSimilarUsersParams{QueryEmbedding: emb, LimitVal: 10})
|
||||
if err != nil {
|
||||
t.Fatalf("find similar: %v", err)
|
||||
}
|
||||
|
|
@ -70,3 +70,131 @@ func TestStoreRoundTrip(t *testing.T) {
|
|||
t.Fatalf("expected newly inserted profile to be returned by similarity search")
|
||||
}
|
||||
}
|
||||
|
||||
// TestSaveTransactionSQL verifies the SaveTransaction SQL query constant is well-formed.
|
||||
func TestSaveTransactionSQL(t *testing.T) {
|
||||
// Verify the const exists and has expected structure
|
||||
const expected = `-- name: SaveTransaction :many
|
||||
INSERT INTO transactions (x402_transaction_hash, amount, asset, network, recipient, sender)
|
||||
VALUES ($1, $2, $3, $4, $5, $6)
|
||||
RETURNING id, created_at
|
||||
`
|
||||
if saveTransaction != expected {
|
||||
t.Fatalf("SaveTransaction SQL mismatch:\n got: %q\nwant: %q", saveTransaction, expected)
|
||||
}
|
||||
}
|
||||
|
||||
// TestSaveTransactionParams verifies the SaveTransaction params struct fields match the SQL.
|
||||
func TestSaveTransactionParams(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
params SaveTransactionParams
|
||||
wantErr bool
|
||||
}{
|
||||
{
|
||||
name: "valid params",
|
||||
params: SaveTransactionParams{
|
||||
X402TransactionHash: "0xabc123",
|
||||
Amount: "10.0",
|
||||
Asset: "ETH",
|
||||
Network: "eip155:1",
|
||||
Recipient: "0xrecipient",
|
||||
Sender: "0xsender",
|
||||
},
|
||||
wantErr: false,
|
||||
},
|
||||
{
|
||||
name: "empty hash",
|
||||
params: SaveTransactionParams{
|
||||
X402TransactionHash: "",
|
||||
Amount: "10.0",
|
||||
Asset: "ETH",
|
||||
Network: "eip155:1",
|
||||
Recipient: "0xrecipient",
|
||||
Sender: "0xsender",
|
||||
},
|
||||
wantErr: false,
|
||||
},
|
||||
{
|
||||
name: "empty amount",
|
||||
params: SaveTransactionParams{
|
||||
X402TransactionHash: "0xabc123",
|
||||
Amount: "",
|
||||
Asset: "ETH",
|
||||
Network: "eip155:1",
|
||||
Recipient: "0xrecipient",
|
||||
Sender: "0xsender",
|
||||
},
|
||||
wantErr: false,
|
||||
},
|
||||
}
|
||||
for _, c := range cases {
|
||||
t.Run(c.name, func(t *testing.T) {
|
||||
// Just verify the struct can be created without panic
|
||||
// Actual DB execution would require a connection
|
||||
_ = c.params
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestStoreSaveTransaction exercises SaveTransaction against a real Postgres database.
|
||||
// It is skipped unless TEST_DATABASE_URL is set.
|
||||
func TestStoreSaveTransaction(t *testing.T) {
|
||||
dsn := os.Getenv("TEST_DATABASE_URL")
|
||||
if dsn == "" {
|
||||
t.Skip("TEST_DATABASE_URL not set; skipping store integration test")
|
||||
}
|
||||
ctx := context.Background()
|
||||
conn, err := sql.Open("postgres", dsn)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer conn.Close()
|
||||
if err := conn.PingContext(ctx); err != nil {
|
||||
t.Fatalf("cannot reach test database: %v", err)
|
||||
}
|
||||
if err := Migrate(ctx, conn, "sql/schema"); err != nil {
|
||||
t.Fatalf("migrate: %v", err)
|
||||
}
|
||||
q := New(conn)
|
||||
|
||||
txHash := fmt.Sprintf("0x_%d", time.Now().Unix())
|
||||
amount := "10.0"
|
||||
asset := "ETH"
|
||||
network := "eip155:1"
|
||||
now := time.Now().Unix()
|
||||
recipient := "0x" + strings.TrimLeft(fmt.Sprintf("%x", now%100000000), "0")
|
||||
sender := "0x" + strings.TrimLeft(fmt.Sprintf("%x", now%100000000+0xab), "0")
|
||||
|
||||
ret, err := q.SaveTransaction(ctx, SaveTransactionParams{
|
||||
X402TransactionHash: txHash,
|
||||
Amount: amount,
|
||||
Asset: asset,
|
||||
Network: network,
|
||||
Recipient: recipient,
|
||||
Sender: sender,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("SaveTransaction error: %v", err)
|
||||
}
|
||||
|
||||
if len(ret) != 1 {
|
||||
t.Fatalf("expected 1 result, got %d", len(ret))
|
||||
}
|
||||
|
||||
id := ret[0].ID
|
||||
if id <= 0 {
|
||||
t.Fatalf("expected positive ID, got %d", id)
|
||||
}
|
||||
|
||||
// Verify the returned row has the expected fields
|
||||
if !ret[0].CreatedAt.Valid {
|
||||
t.Fatal("expected CreatedAt to be set")
|
||||
}
|
||||
|
||||
// Verify we can retrieve the transaction back
|
||||
// Note: we'd need a SELECT query for full verification, but the insert
|
||||
// returning id and created_at is the critical path tested here
|
||||
|
||||
t.Logf("SaveTransaction successful: id=%d, hash=%s", id, txHash)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -20,13 +20,16 @@ type Transaction struct {
|
|||
}
|
||||
|
||||
type UserStyleProfile struct {
|
||||
ID int64
|
||||
UserID string
|
||||
Platform string
|
||||
Embedding interface{}
|
||||
TopKeywords []string
|
||||
StyleSummary string
|
||||
PostCount int32
|
||||
ModifiedAt sql.NullTime
|
||||
CreatedAt sql.NullTime
|
||||
ID int64
|
||||
UserID string
|
||||
Platform string
|
||||
Embedding any
|
||||
TopKeywords []string
|
||||
StyleSummary string
|
||||
PostCount int32
|
||||
ModifiedAt sql.NullTime
|
||||
CreatedAt sql.NullTime
|
||||
ModelVersion int32
|
||||
SchemaVersion int32
|
||||
AlgorithmVersion int32
|
||||
}
|
||||
|
|
|
|||
|
|
@ -7,13 +7,14 @@ import (
|
|||
"io"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/nacorid/naco-api/internal/actor"
|
||||
log "github.com/nacorid/logger"
|
||||
)
|
||||
|
||||
const apiBase = "https://api.twitter.com/2"
|
||||
const apiBase = "https://api.x.com/2"
|
||||
|
||||
type Client struct {
|
||||
httpClient *http.Client
|
||||
|
|
@ -95,27 +96,79 @@ type singleTweetResponse struct {
|
|||
} `json:"errors"`
|
||||
}
|
||||
|
||||
// doGet issues a GET and retries with exponential backoff when Twitter returns HTTP 429
|
||||
// (Too Many Requests). It honors the x-rate-limit-* headers and caps retries so a
|
||||
// sustained rate limit still surfaces as a real error to the caller (and thus isn't billed).
|
||||
func (c *Client) doGet(ctx context.Context, endpoint string) ([]byte, error) {
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to build request: %w", err)
|
||||
}
|
||||
req.Header.Set("Authorization", "Bearer "+c.bearerToken)
|
||||
const maxRetries = 4
|
||||
backoff := 500 * time.Millisecond
|
||||
for attempt := 0; ; attempt++ {
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to build request: %w", err)
|
||||
}
|
||||
req.Header.Set("Authorization", "Bearer "+c.bearerToken)
|
||||
|
||||
resp, err := c.httpClient.Do(req)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to do request: %w", err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
resp, err := c.httpClient.Do(req)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to do request: %w", err)
|
||||
}
|
||||
body, err := io.ReadAll(resp.Body)
|
||||
resp.Body.Close()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to read response: %w", err)
|
||||
}
|
||||
|
||||
body, err := io.ReadAll(resp.Body)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to read response: %w", err)
|
||||
if resp.StatusCode == http.StatusTooManyRequests {
|
||||
if attempt >= maxRetries {
|
||||
return nil, fmt.Errorf("twitter rate limited after retries: %s", resp.Status)
|
||||
}
|
||||
// Honor x-rate-limit-* headers first
|
||||
remaining := resp.Header.Get("x-rate-limit-remaining")
|
||||
reset := resp.Header.Get("x-rate-limit-reset")
|
||||
if ra := remaining; ra != "" {
|
||||
if rem, perr := strconv.Atoi(ra); perr == nil {
|
||||
if rem <= 0 && reset != "" {
|
||||
if rst, perr2 := strconv.Atoi(reset); perr2 == nil {
|
||||
// Wait until the rate limit resets (Unix timestamp)
|
||||
now := time.Now().Unix()
|
||||
rst64 := int64(rst)
|
||||
wait := time.Duration(rst64-now) * time.Second
|
||||
if wait > 0 {
|
||||
log.WarnWithContext(ctx, "twitter rate limited, waiting until reset", "reset", reset, "wait", wait.String())
|
||||
select {
|
||||
case <-time.After(wait):
|
||||
case <-ctx.Done():
|
||||
return nil, ctx.Err()
|
||||
}
|
||||
backoff = 500 * time.Millisecond
|
||||
continue
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
// Fall back to Retry-After if headers not useful
|
||||
if ra := resp.Header.Get("Retry-After"); ra != "" {
|
||||
if secs, perr := strconv.Atoi(ra); perr == nil && secs > 0 {
|
||||
backoff = time.Duration(secs) * time.Second
|
||||
}
|
||||
}
|
||||
log.WarnWithContext(ctx, "twitter rate limited, backing off", "attempt", attempt+1, "wait", backoff.String())
|
||||
select {
|
||||
case <-time.After(backoff):
|
||||
case <-ctx.Done():
|
||||
return nil, ctx.Err()
|
||||
}
|
||||
backoff *= 2
|
||||
continue
|
||||
}
|
||||
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return nil, fmt.Errorf("twitter error: %s: %s", resp.Status, string(body))
|
||||
}
|
||||
return body, nil
|
||||
}
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return nil, fmt.Errorf("twitter error: %s: %s", resp.Status, string(body))
|
||||
}
|
||||
return body, nil
|
||||
}
|
||||
|
||||
func (c *Client) resolveUser(ctx context.Context, username string) (string, error) {
|
||||
|
|
@ -283,84 +336,3 @@ type userFields struct {
|
|||
TweetCount int `json:"tweet_count"`
|
||||
} `json:"public_metrics"`
|
||||
}
|
||||
|
||||
func (c *Client) GetActor(ctx context.Context, handle string) (*actor.Profile, error) {
|
||||
if c.bearerToken == "" {
|
||||
return nil, fmt.Errorf("twitter bearer token not configured")
|
||||
}
|
||||
username := strings.TrimPrefix(handle, "@")
|
||||
endpoint := fmt.Sprintf("%s/users/by/username/%s?user.fields=description,profile_image_url,public_metrics,username,name", apiBase, url.PathEscape(username))
|
||||
body, err := c.doGet(ctx, endpoint)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var ur struct {
|
||||
Data userFields `json:"data"`
|
||||
Errors []struct {
|
||||
Title string `json:"title"`
|
||||
Detail string `json:"detail"`
|
||||
} `json:"errors"`
|
||||
}
|
||||
if err := json.Unmarshal(body, &ur); err != nil {
|
||||
return nil, fmt.Errorf("failed to unmarshal user: %w", err)
|
||||
}
|
||||
if len(ur.Errors) > 0 {
|
||||
return nil, fmt.Errorf("twitter error: %s: %s", ur.Errors[0].Title, ur.Errors[0].Detail)
|
||||
}
|
||||
return &actor.Profile{
|
||||
Platform: "Twitter",
|
||||
Handle: "@" + ur.Data.Username,
|
||||
DisplayName: ur.Data.Name,
|
||||
AvatarURL: ur.Data.ProfileImageURL,
|
||||
Bio: ur.Data.Description,
|
||||
FollowersCount: ur.Data.PublicMetrics.FollowersCount,
|
||||
FollowingCount: ur.Data.PublicMetrics.FollowingCount,
|
||||
PostCount: ur.Data.PublicMetrics.TweetCount,
|
||||
URL: "https://twitter.com/" + ur.Data.Username,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (c *Client) SearchActors(ctx context.Context, query string, limit int) ([]actor.Profile, error) {
|
||||
if c.bearerToken == "" {
|
||||
return nil, fmt.Errorf("twitter bearer token not configured")
|
||||
}
|
||||
if limit > 50 {
|
||||
limit = 50
|
||||
}
|
||||
if limit < 1 {
|
||||
limit = 10
|
||||
}
|
||||
endpoint := fmt.Sprintf("%s/users/search?q=%s&max_results=%d&user.fields=description,profile_image_url,public_metrics,username,name", apiBase, url.QueryEscape(query), limit)
|
||||
body, err := c.doGet(ctx, endpoint)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var sr struct {
|
||||
Data []userFields `json:"data"`
|
||||
Errors []struct {
|
||||
Title string `json:"title"`
|
||||
Detail string `json:"detail"`
|
||||
} `json:"errors"`
|
||||
}
|
||||
if err := json.Unmarshal(body, &sr); err != nil {
|
||||
return nil, fmt.Errorf("failed to unmarshal users: %w", err)
|
||||
}
|
||||
if len(sr.Errors) > 0 {
|
||||
return nil, fmt.Errorf("twitter error: %s: %s", sr.Errors[0].Title, sr.Errors[0].Detail)
|
||||
}
|
||||
results := make([]actor.Profile, 0, len(sr.Data))
|
||||
for _, u := range sr.Data {
|
||||
results = append(results, actor.Profile{
|
||||
Platform: "Twitter",
|
||||
Handle: "@" + u.Username,
|
||||
DisplayName: u.Name,
|
||||
AvatarURL: u.ProfileImageURL,
|
||||
Bio: u.Description,
|
||||
FollowersCount: u.PublicMetrics.FollowersCount,
|
||||
FollowingCount: u.PublicMetrics.FollowingCount,
|
||||
PostCount: u.PublicMetrics.TweetCount,
|
||||
URL: "https://twitter.com/" + u.Username,
|
||||
})
|
||||
}
|
||||
return results, nil
|
||||
}
|
||||
|
|
|
|||
|
|
@ -168,64 +168,3 @@ func TestGetPostsInTimeframe(t *testing.T) {
|
|||
}
|
||||
}
|
||||
|
||||
func TestGetActor(t *testing.T) {
|
||||
mux := http.NewServeMux()
|
||||
mux.HandleFunc("/2/users/by/username/", func(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Header.Get("Authorization") != "Bearer test-token" {
|
||||
w.WriteHeader(http.StatusUnauthorized)
|
||||
return
|
||||
}
|
||||
fmt.Fprint(w, `{"data":{"id":"12345","name":"Test User","username":"testuser","description":"bio text","profile_image_url":"https://img/x.png","public_metrics":{"followers_count":100,"following_count":50,"tweet_count":20}}}`)
|
||||
})
|
||||
server := httptest.NewServer(mux)
|
||||
defer server.Close()
|
||||
|
||||
c := newTestClient(server.URL)
|
||||
prof, err := c.GetActor(context.Background(), "@testuser")
|
||||
if err != nil {
|
||||
t.Fatalf("GetActor error: %v", err)
|
||||
}
|
||||
if prof.Platform != "Twitter" || prof.Handle != "@testuser" || prof.DisplayName != "Test User" {
|
||||
t.Fatalf("unexpected profile: %+v", prof)
|
||||
}
|
||||
if prof.Bio != "bio text" || prof.AvatarURL != "https://img/x.png" {
|
||||
t.Fatalf("unexpected profile fields: %+v", prof)
|
||||
}
|
||||
if prof.FollowersCount != 100 || prof.FollowingCount != 50 || prof.PostCount != 20 {
|
||||
t.Fatalf("unexpected counts: %+v", prof)
|
||||
}
|
||||
if prof.URL != "https://twitter.com/testuser" {
|
||||
t.Fatalf("unexpected url: %q", prof.URL)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSearchActors(t *testing.T) {
|
||||
mux := http.NewServeMux()
|
||||
mux.HandleFunc("/2/users/search", func(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Header.Get("Authorization") != "Bearer test-token" {
|
||||
w.WriteHeader(http.StatusUnauthorized)
|
||||
return
|
||||
}
|
||||
if r.URL.Query().Get("q") != "test" {
|
||||
t.Fatalf("unexpected q: %q", r.URL.Query().Get("q"))
|
||||
}
|
||||
fmt.Fprint(w, `{"data":[{"id":"1","name":"T","username":"test","description":"d","profile_image_url":"https://i","public_metrics":{"followers_count":1,"following_count":2,"tweet_count":3}}]}`)
|
||||
})
|
||||
server := httptest.NewServer(mux)
|
||||
defer server.Close()
|
||||
|
||||
c := newTestClient(server.URL)
|
||||
results, err := c.SearchActors(context.Background(), "test", 10)
|
||||
if err != nil {
|
||||
t.Fatalf("SearchActors error: %v", err)
|
||||
}
|
||||
if len(results) != 1 {
|
||||
t.Fatalf("expected 1 result, got %d", len(results))
|
||||
}
|
||||
if results[0].Platform != "Twitter" || results[0].Handle != "@test" {
|
||||
t.Fatalf("unexpected result: %+v", results[0])
|
||||
}
|
||||
if results[0].FollowersCount != 1 || results[0].PostCount != 3 {
|
||||
t.Fatalf("unexpected counts: %+v", results[0])
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -21,12 +21,12 @@ func parseVectorString(v string) ([]float64, error) {
|
|||
return vec, nil
|
||||
}
|
||||
|
||||
func ParseEmbedding(value interface{}) ([]float64, error) {
|
||||
func ParseEmbedding(value any) ([]float64, error) {
|
||||
switch v := value.(type) {
|
||||
case []float64:
|
||||
return v, nil
|
||||
|
||||
case []interface{}:
|
||||
case []any:
|
||||
vec := make([]float64, len(v))
|
||||
for i, x := range v {
|
||||
f, ok := x.(float64)
|
||||
|
|
|
|||
|
|
@ -26,16 +26,16 @@ func TestFormatVectorRoundTrip(t *testing.T) {
|
|||
func TestParseEmbedding(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
in interface{}
|
||||
in any
|
||||
want []float64
|
||||
err bool
|
||||
}{
|
||||
{"float64 slice", []float64{1, 2, 3}, []float64{1, 2, 3}, false},
|
||||
{"interface slice", []interface{}{float64(1), float64(2)}, []float64{1, 2}, false},
|
||||
{"interface slice", []any{float64(1), float64(2)}, []float64{1, 2}, false},
|
||||
{"string", "[0.5, 1.5]", []float64{0.5, 1.5}, false},
|
||||
{"bytes", []byte("[0.5,1.5]"), []float64{0.5, 1.5}, false},
|
||||
{"bad string", "notanumber", nil, true},
|
||||
{"interface bad elem", []interface{}{"x"}, nil, true},
|
||||
{"interface bad elem", []any{"x"}, nil, true},
|
||||
{"unsupported", 42, nil, true},
|
||||
}
|
||||
for _, c := range cases {
|
||||
|
|
|
|||
|
|
@ -19,6 +19,7 @@ import (
|
|||
nethttp "github.com/x402-foundation/x402/go/v2/http/nethttp"
|
||||
x402mcp "github.com/x402-foundation/x402/go/v2/mcp"
|
||||
evm "github.com/x402-foundation/x402/go/v2/mechanisms/evm/exact/server"
|
||||
uptoevm "github.com/x402-foundation/x402/go/v2/mechanisms/evm/upto/server"
|
||||
)
|
||||
|
||||
type X402Middlewares struct {
|
||||
|
|
@ -74,6 +75,7 @@ func NewX402Server(name, version string, facilitatorURL string, auth x402http.Au
|
|||
)
|
||||
|
||||
resourceServer.Register("eip155:8453", evm.NewExactEvmScheme())
|
||||
resourceServer.Register("eip155:8453", uptoevm.NewUptoEvmScheme())
|
||||
|
||||
s := &X402Server{
|
||||
mcpServer: mcpServer,
|
||||
|
|
@ -210,6 +212,13 @@ func (s *X402Server) handleRESTCall(w http.ResponseWriter, r *http.Request, tool
|
|||
if textContent, ok := resp.Content[0].(*mcp.TextContent); ok {
|
||||
var js json.RawMessage
|
||||
if json.Unmarshal([]byte(textContent.Text), &js) == nil {
|
||||
// For upto scheme tools, compute actual cost and apply settlement override
|
||||
if toolName == "getStyleProfile" {
|
||||
actualCost := computeTwitterStyleProfileCost(r, textContent.Text)
|
||||
nethttp.SetSettlementOverrides(w, &x402.SettlementOverrides{
|
||||
Amount: fmt.Sprintf("%d", actualCost),
|
||||
})
|
||||
}
|
||||
w.Write([]byte(textContent.Text))
|
||||
return
|
||||
}
|
||||
|
|
@ -391,3 +400,64 @@ func chainHTTPMiddlewares(middlewares ...func(http.Handler) http.Handler) func(h
|
|||
return next
|
||||
}
|
||||
}
|
||||
|
||||
// computeTwitterStyleProfileCost calculates the actual charge for a Twitter getStyleProfile
|
||||
// request based on the number of posts fetched and user lookup.
|
||||
// Pricing: user read $0.01, post read $0.005 each, minimum $0.10, maximum $0.26.
|
||||
func computeTwitterStyleProfileCost(r *http.Request, responseText string) int {
|
||||
// Parse the response to get PostCount and SourcePlatform
|
||||
var resp GetStyleProfileResponse
|
||||
if err := json.Unmarshal([]byte(responseText), &resp); err != nil {
|
||||
return 10 // fallback: 10 cents = $0.10 minimum
|
||||
}
|
||||
|
||||
// Determine if Twitter platform
|
||||
isTwitter := resp.SourcePlatform == "Twitter"
|
||||
|
||||
// Extract post count from response
|
||||
postCount := resp.PostCount
|
||||
if postCount <= 0 {
|
||||
postCount = 0
|
||||
}
|
||||
|
||||
// Calculate cost:
|
||||
// - User read: $0.01 if Twitter and we assume user lookup was performed
|
||||
// - Posts read: $0.005 per post, max 50 posts = $0.25
|
||||
// - Minimum: $0.10
|
||||
// - Maximum: $0.26
|
||||
var costCents int
|
||||
|
||||
if isTwitter {
|
||||
// Twitter: user read + posts read
|
||||
userCents := 1 // $0.01
|
||||
postCents := postCount * 5 // $0.005 per post = 5 cents per post in atomic units
|
||||
if postCents > 125 { // 50 posts * 5 = 25 cents max
|
||||
postCents = 125
|
||||
}
|
||||
costCents = userCents + postCents
|
||||
} else {
|
||||
costCents = 10 // $0.10 minimum for non-Twitter
|
||||
}
|
||||
|
||||
// Apply minimum $0.10 (10 cents)
|
||||
if costCents < 10 {
|
||||
costCents = 10
|
||||
}
|
||||
|
||||
// Apply maximum $0.26 (26 cents)
|
||||
if costCents > 26 {
|
||||
costCents = 26
|
||||
}
|
||||
|
||||
return costCents
|
||||
}
|
||||
|
||||
// GetStyleProfileResponse is a local copy of the response schema for cost calculation
|
||||
type GetStyleProfileResponse struct {
|
||||
Author string `json:"author"`
|
||||
PersonaEmbedding []float64 `json:"personaEmbedding"`
|
||||
TopKeywords []string `json:"topKeywords"`
|
||||
StyleSummary string `json:"styleSummary"`
|
||||
PostCount int `json:"postCount"`
|
||||
SourcePlatform string `json:"sourcePlatform"`
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
-- name: FindSimilarUsers :many
|
||||
SELECT user_id, platform, embedding, top_keywords, style_summary, post_count,
|
||||
1 - (embedding <=> $1::vector) AS similarity
|
||||
1 - (embedding <=> @query_embedding::vector) AS similarity
|
||||
FROM user_style_profiles
|
||||
ORDER BY embedding <=> $1::vector
|
||||
LIMIT $2;
|
||||
ORDER BY embedding <=> @query_embedding::vector
|
||||
LIMIT @limit_val;
|
||||
|
|
@ -1,3 +1,3 @@
|
|||
-- name: GetStyleProfile :one
|
||||
SELECT * FROM user_style_profiles
|
||||
WHERE user_id = $1 AND platform = $2;
|
||||
SELECT id, user_id, platform, embedding, top_keywords, style_summary, post_count, modified_at, created_at, model_version, schema_version, algorithm_version FROM user_style_profiles
|
||||
WHERE user_id = @user_id AND platform = @platform;
|
||||
|
|
@ -1,6 +1,6 @@
|
|||
-- name: UpsertStyleProfile :exec
|
||||
INSERT INTO user_style_profiles (user_id, platform, embedding, top_keywords, style_summary, post_count)
|
||||
VALUES ($1, $2, $3::vector, $4, $5, $6)
|
||||
VALUES (@user_id, @platform, @embedding::vector, @top_keywords, @style_summary, @post_count)
|
||||
ON CONFLICT (user_id, platform)
|
||||
DO UPDATE SET
|
||||
embedding = EXCLUDED.embedding,
|
||||
|
|
|
|||
|
|
@ -16,3 +16,5 @@ CREATE TABLE user_style_profiles (
|
|||
|
||||
-- +goose down
|
||||
DROP TABLE IF EXISTS user_style_profiles;
|
||||
|
||||
DROP EXTENSION IF EXISTS vector;
|
||||
15
sql/schema/003_add_cache_version.sql
Normal file
15
sql/schema/003_add_cache_version.sql
Normal file
|
|
@ -0,0 +1,15 @@
|
|||
-- +goose Up
|
||||
ALTER TABLE user_style_profiles
|
||||
ADD COLUMN model_version INTEGER NOT NULL DEFAULT 1,
|
||||
ADD COLUMN schema_version INTEGER NOT NULL DEFAULT 1,
|
||||
ADD COLUMN algorithm_version INTEGER NOT NULL DEFAULT 1;
|
||||
|
||||
CREATE INDEX idx_user_style_profiles_versions
|
||||
ON user_style_profiles (model_version, schema_version, algorithm_version);
|
||||
|
||||
-- +goose Down
|
||||
DROP INDEX idx_user_style_profiles_versions;
|
||||
ALTER TABLE user_style_profiles
|
||||
DROP COLUMN algorithm_version,
|
||||
DROP COLUMN model_version,
|
||||
DROP COLUMN schema_version;
|
||||
|
|
@ -1,5 +0,0 @@
|
|||
-- +goose up
|
||||
CREATE EXTENSION IF NOT EXISTS vector;
|
||||
|
||||
-- +goose down
|
||||
DROP EXTENSION IF EXISTS vector;
|
||||
Loading…
Reference in a new issue