package activitypub import ( "context" "encoding/json" "fmt" "io" "net/http" "net/url" "strings" "sync" "time" "github.com/nacorid/naco-api/internal/actor" ) const maxMastodonPageSize = 40 const maxTimeframePages = 100 const maxResponseBody = 1 << 20 const maxLimit = 200 type APClient struct { client *http.Client host string bearerToken string outbox *outboxClient limiter *hostLimiter } func NewAPClient(host, apikey string) (*APClient, error) { if host == "" || apikey == "" { return nil, fmt.Errorf("host and apikey must not be empty") } return &APClient{ client: &http.Client{Timeout: 30 * time.Second}, host: strings.TrimRight(host, "/"), bearerToken: "Bearer " + apikey, outbox: newOutboxClient(), limiter: newHostLimiter(), }, nil } func (ap *APClient) get(ctx context.Context, endpoint string, target any) error { req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil) if err != nil { return fmt.Errorf("failed to create request: %w", err) } req.Header.Set("Authorization", ap.bearerToken) resp, err := ap.client.Do(req) if err != nil { return err } defer resp.Body.Close() if resp.StatusCode < 200 || resp.StatusCode >= 300 { errBody, _ := io.ReadAll(io.LimitReader(resp.Body, 4096)) return fmt.Errorf("api error (status %d): %s", resp.StatusCode, string(errBody)) } if err := json.NewDecoder(io.LimitReader(resp.Body, maxResponseBody)).Decode(target); err != nil { return fmt.Errorf("failed to decode response: %w", err) } return nil } func (ap *APClient) accountSearch(ctx context.Context, query string) (*Account, error) { endpoint := fmt.Sprintf("%s/api/v1/accounts/search?q=%s&resolve=true&limit=1", ap.host, url.QueryEscape(query)) var accounts []*Account if err := ap.get(ctx, endpoint, &accounts); err != nil { return nil, fmt.Errorf("failed to search accounts: %w", err) } for _, acc := range accounts { if acc != nil { return acc, nil } } return nil, nil } func (ap *APClient) resolveAccount(ctx context.Context, actorHandle string) (*Account, error) { handle := strings.TrimPrefix(actorHandle, "@") acc, err := ap.accountSearch(ctx, handle) if err != nil { return nil, err } if acc != nil { return acc, nil } if !strings.Contains(handle, "@") { var byID Account endpoint := fmt.Sprintf("%s/api/v1/accounts/%s", ap.host, url.PathEscape(handle)) if err := ap.get(ctx, endpoint, &byID); err != nil { return nil, fmt.Errorf("account %q not found by search or id: %w", actorHandle, err) } if byID.ID == "" { return nil, fmt.Errorf("account %q not found", actorHandle) } return &byID, nil } return nil, fmt.Errorf("account %q not found", actorHandle) } func (ap *APClient) GetPost(ctx context.Context, postID string) (*Post, error) { status, err := ap.resolveStatus(ctx, postID) if err != nil { return nil, err } return status.toPost(), nil } func (ap *APClient) resolveStatus(ctx context.Context, postID string) (*internalPost, error) { if strings.HasPrefix(postID, "http://") || strings.HasPrefix(postID, "https://") || strings.HasPrefix(postID, "//") { if status, err := ap.searchStatus(ctx, postID); err == nil && status != nil { return status, nil } note, derr := ap.outbox.resolveNote(ctx, postID) if derr != nil { return nil, fmt.Errorf("status %q not resolvable: %w", postID, derr) } author := extractHandleFromURL(postID) return noteToInternalPost(note, author), nil } endpoint := fmt.Sprintf("%s/api/v1/statuses/%s", ap.host, url.PathEscape(postID)) var status internalPost if err := ap.get(ctx, endpoint, &status); err != nil { return nil, fmt.Errorf("failed to get post: %w", err) } return &status, nil } func (ap *APClient) searchStatus(ctx context.Context, postID string) (*internalPost, error) { endpoint := fmt.Sprintf("%s/api/v2/search?q=%s&resolve=true&type=statuses&limit=1", ap.host, url.QueryEscape(postID)) var res searchResult if err := ap.get(ctx, endpoint, &res); err != nil { return nil, fmt.Errorf("failed to resolve status %q: %w", postID, err) } for _, s := range res.Statuses { if s != nil && (s.URL == postID || s.URI == postID) { return s, nil } } return nil, nil } func (ap *APClient) GetLatestNPosts(ctx context.Context, actorID string, limit int) ([]*Post, error) { if limit <= 0 { return nil, nil } if limit > maxLimit { limit = maxLimit } acc, err := ap.resolveAccount(ctx, actorID) if err != nil { return nil, err } cached, cacheErr := ap.cacheLatestPosts(ctx, acc, limit) if cacheErr != nil { return nil, cacheErr } if len(cached) >= limit { return cachePostsToPosts(cached), nil } if !isRemoteAccount(acc) || acc.URL == "" { return cachePostsToPosts(cached), nil } outboxPosts, apErr := ap.outbox.getLatest(ctx, acc.URL, acc.Acct, limit-len(cached)) if apErr != nil { return nil, fmt.Errorf("outbox fallback failed: %w", apErr) } merged := mergePostsBySource(cached, outboxPosts) if len(outboxPosts) > 0 { urls := make([]string, 0, len(outboxPosts)) for _, p := range outboxPosts { if p != nil && p.ID != "" { urls = append(urls, p.ID) } } go ap.warmGTSAsync(context.WithoutCancel(ctx), urls) } return merged, nil } func (ap *APClient) cacheLatestPosts(ctx context.Context, acc *Account, limit int) ([]*internalPost, error) { var ( internal []*internalPost maxID string ) for len(internal) < limit { if ctx.Err() != nil { return nil, ctx.Err() } pageSize := limit - len(internal) if pageSize > maxMastodonPageSize { pageSize = maxMastodonPageSize } endpoint := fmt.Sprintf("%s/api/v1/accounts/%s/statuses?limit=%d", ap.host, url.PathEscape(acc.ID), pageSize) if maxID != "" { endpoint += "&max_id=" + url.QueryEscape(maxID) } var page []*internalPost if err := ap.get(ctx, endpoint, &page); err != nil { return nil, err } if len(page) == 0 { break } newMaxID := page[len(page)-1].ID if newMaxID == "" { break } if maxID != "" && newMaxID == maxID { break } internal = append(internal, page...) maxID = newMaxID } if len(internal) > limit { internal = internal[:limit] } return internal, nil } func (ap *APClient) GetPostsInTimeframe(ctx context.Context, actorID string, start, end time.Time, limit int) ([]*Post, error) { if limit <= 0 { return nil, nil } if limit > maxLimit { limit = maxLimit } acc, err := ap.resolveAccount(ctx, actorID) if err != nil { return nil, err } cached, cacheErr := ap.cachePostsInTimeframe(ctx, acc, start, end, limit) if cacheErr != nil { return nil, cacheErr } if len(cached) >= limit { return cachePostsToPosts(cached), nil } if !isRemoteAccount(acc) || acc.URL == "" { return cachePostsToPosts(cached), nil } outboxPosts, apErr := ap.outbox.getInTimeframe(ctx, acc.URL, acc.Acct, start, end, limit-len(cached)) if apErr != nil { return nil, fmt.Errorf("outbox fallback failed: %w", apErr) } merged := mergePostsBySource(cached, outboxPosts) if len(outboxPosts) > 0 { urls := make([]string, 0, len(outboxPosts)) for _, p := range outboxPosts { if p != nil && p.ID != "" { urls = append(urls, p.ID) } } go ap.warmGTSAsync(context.WithoutCancel(ctx), urls) } return merged, nil } func (ap *APClient) cachePostsInTimeframe(ctx context.Context, acc *Account, start, end time.Time, limit int) ([]*internalPost, error) { var ( internal []*internalPost maxID string ) for pages := 0; pages < maxTimeframePages; pages++ { if ctx.Err() != nil { return nil, ctx.Err() } if len(internal) >= limit { break } endpoint := fmt.Sprintf("%s/api/v1/accounts/%s/statuses?limit=%d", ap.host, url.PathEscape(acc.ID), maxMastodonPageSize) if maxID != "" { endpoint += "&max_id=" + url.QueryEscape(maxID) } var page []*internalPost if err := ap.get(ctx, endpoint, &page); err != nil { return nil, err } if len(page) == 0 { break } newMaxID := page[len(page)-1].ID if newMaxID == "" { break } if maxID != "" && newMaxID == maxID { break } for _, i := range page { if i.CreatedAt.After(end) { continue } if i.CreatedAt.Before(start) { return internal, nil } internal = append(internal, i) if len(internal) >= limit { return internal, nil } } maxID = newMaxID } return internal, nil } func isRemoteAccount(acc *Account) bool { return acc != nil && strings.Contains(acc.Acct, "@") } func cachePostsToPosts(cached []*internalPost) []*Post { if len(cached) == 0 { return nil } posts := make([]*Post, 0, len(cached)) for _, c := range cached { if c == nil { continue } posts = append(posts, c.toPost()) } return posts } func (ap *APClient) GetActor(ctx context.Context, actorHandle string) (*actor.Profile, error) { acc, err := ap.resolveAccount(ctx, actorHandle) if err != nil { return nil, err } return ap.toProfile(acc), nil } func (ap *APClient) SearchActors(ctx context.Context, query string, limit int) ([]*actor.Profile, error) { endpoint := fmt.Sprintf("%s/api/v1/accounts/search?q=%s&resolve=true&limit=%d", ap.host, url.QueryEscape(query), limit) var accounts []*Account if err := ap.get(ctx, endpoint, &accounts); err != nil { return nil, fmt.Errorf("failed to search actors: %w", err) } results := make([]*actor.Profile, 0, len(accounts)) for _, acc := range accounts { if acc == nil { continue } results = append(results, ap.toProfile(acc)) } return results, nil } func (ap *APClient) toProfile(acc *Account) *actor.Profile { return &actor.Profile{ Platform: "ActivityPub", Handle: acc.Acct, DisplayName: acc.DisplayName, AvatarURL: acc.Avatar, Bio: acc.Note, FollowersCount: acc.FollowersCount, FollowingCount: acc.FollowingCount, PostCount: acc.StatusesCount, URL: acc.URL, } } func extractHandleFromURL(postURL string) string { u, err := url.Parse(postURL) if err != nil { return "" } parts := strings.Split(strings.TrimPrefix(u.Path, "/"), "/") for _, p := range parts { if strings.HasPrefix(p, "@") { return strings.TrimPrefix(p, "@") } } return "" } func (ap *APClient) warmGTSAsync(ctx context.Context, urls []string) { if len(urls) == 0 { return } if ap.limiter == nil { ap.limiter = newHostLimiter() } ctx, cancel := context.WithTimeout(ctx, 5*time.Minute) defer cancel() seen := make(map[string]struct{}, len(urls)) uniq := make([]string, 0, len(urls)) for _, u := range urls { if _, ok := seen[u]; ok { continue } seen[u] = struct{}{} uniq = append(uniq, u) } if len(uniq) > 50 { uniq = uniq[:50] } byHost := make(map[string][]string) for _, u := range uniq { parsed, err := url.Parse(u) if err != nil || parsed.Host == "" { continue } byHost[parsed.Host] = append(byHost[parsed.Host], u) } if len(byHost) == 0 { return } var wg sync.WaitGroup for host, us := range byHost { wg.Add(1) go func(host string, us []string) { defer wg.Done() for _, u := range us { if err := ap.limiter.Wait(ctx, host); err != nil { return } if ctx.Err() != nil { return } var res searchResult _ = ap.get(ctx, fmt.Sprintf("%s/api/v2/search?q=%s&resolve=true&type=statuses&limit=1", ap.host, url.QueryEscape(u)), &res) } }(host, us) } wg.Wait() }