naco-api/internal/activitypub/client.go

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