naco-api/internal/activitypub/outbox.go

package activitypub

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"sort"
	"strings"
	"time"
)

const outboxMaxPages = 100

var webfingerScheme = "https"

type outboxClient struct {
	client *http.Client
}

type apStringOrArray []string

func (s *apStringOrArray) UnmarshalJSON(b []byte) error {
	b = bytes.TrimSpace(b)
	if len(b) == 0 || string(b) == "null" {
		*s = nil
		return nil
	}
	if b[0] == '"' {
		var v string
		if err := json.Unmarshal(b, &v); err != nil {
			return err
		}
		*s = []string{v}
		return nil
	}
	var arr []string
	if err := json.Unmarshal(b, &arr); err != nil {
		return err
	}
	*s = arr
	return nil
}

type apIRI string

func (i *apIRI) UnmarshalJSON(b []byte) error {
	b = bytes.TrimSpace(b)
	if len(b) == 0 || string(b) == "null" {
		*i = ""
		return nil
	}
	if b[0] == '"' {
		var s string
		if err := json.Unmarshal(b, &s); err != nil {
			return err
		}
		*i = apIRI(s)
		return nil
	}
	var o struct {
		ID string `json:"id"`
	}
	if err := json.Unmarshal(b, &o); err != nil {
		return err
	}
	*i = apIRI(o.ID)
	return nil
}

type apRef struct {
	IRI    string
	Inline json.RawMessage
}

func (r *apRef) UnmarshalJSON(b []byte) error {
	b = bytes.TrimSpace(b)
	if len(b) == 0 || string(b) == "null" {
		r.IRI = ""
		r.Inline = nil
		return nil
	}
	if b[0] == '"' {
		var s string
		if err := json.Unmarshal(b, &s); err != nil {
			return err
		}
		r.IRI = s
		return nil
	}
	r.Inline = append([]byte(nil), b...)
	return nil
}

type apActor struct {
	ID                string `json:"id"`
	Type              string `json:"type"`
	PreferredUsername string `json:"preferredUsername"`
	URL               string `json:"url"`
	Outbox            apIRI  `json:"outbox"`
	Inbox             apIRI  `json:"inbox"`
}

type apActivity struct {
	ID        string          `json:"id"`
	Type      string          `json:"type"`
	Published string          `json:"published"`
	Content   string          `json:"content"`
	URL       apStringOrArray `json:"url"`
	Object    json.RawMessage `json:"object"`
}

type apCollection struct {
	Type  string `json:"type"`
	First apRef  `json:"first"`
}

type apPage struct {
	Type         string          `json:"type"`
	OrderedItems json.RawMessage `json:"orderedItems"`
	Next         apRef           `json:"next"`
}

type apNote struct {
	ID        string          `json:"id"`
	Type      string          `json:"type"`
	Published string          `json:"published"`
	Content   string          `json:"content"`
	URL       apStringOrArray `json:"url"`
}

type webfingerResponse struct {
	Subject string `json:"subject"`
	Links   []struct {
		Rel  string `json:"rel"`
		Type string `json:"type"`
		Href string `json:"href"`
	} `json:"links"`
}

func newOutboxClient() *outboxClient {
	return &outboxClient{client: &http.Client{Timeout: 30 * time.Second}}
}

func (oc *outboxClient) getJSON(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("Accept", `application/activity+json, application/ld+json; profile="https://www.w3.org/ns/activitystreams", application/json`)

	resp, err := oc.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 (oc *outboxClient) webfinger(ctx context.Context, handle string) (string, error) {
	full := strings.TrimPrefix(handle, "@")
	at := strings.LastIndex(full, "@")
	if at <= 0 || at == len(full)-1 {
		return "", fmt.Errorf("invalid handle %q", handle)
	}
	domain := full[at+1:]

	endpoint := fmt.Sprintf("%s://%s/.well-known/webfinger?resource=%s", webfingerScheme, domain, url.QueryEscape("acct:"+full))
	var wf webfingerResponse
	if err := oc.getJSON(ctx, endpoint, &wf); err != nil {
		return "", fmt.Errorf("webfinger lookup for %q failed: %w", full, err)
	}

	var htmlSelf string
	for _, l := range wf.Links {
		if l.Rel != "self" || l.Href == "" {
			continue
		}
		if strings.Contains(l.Type, "activity+json") || strings.Contains(l.Type, "ld+json") {
			return l.Href, nil
		}
		if htmlSelf == "" && strings.Contains(l.Type, "html") {
			htmlSelf = l.Href
		}
	}
	if htmlSelf != "" {
		return htmlSelf, nil
	}
	return "", fmt.Errorf("webfinger for %q returned no self link", full)
}

func (oc *outboxClient) getActor(ctx context.Context, actorURL string) (*apActor, error) {
	var actor apActor
	if err := oc.getJSON(ctx, actorURL, &actor); err != nil {
		return nil, fmt.Errorf("failed to fetch actor %q: %w", actorURL, err)
	}
	if actor.ID == "" && actor.Outbox == "" {
		return nil, fmt.Errorf("response from %q is not an actor document", actorURL)
	}
	return &actor, nil
}

func (oc *outboxClient) resolveActor(ctx context.Context, actorURL, handle string) (*apActor, error) {
	if actorURL == "" {
		var err error
		actorURL, err = oc.webfinger(ctx, handle)
		if err != nil {
			return nil, err
		}
	}
	return oc.getActor(ctx, actorURL)
}

func (oc *outboxClient) getLatest(ctx context.Context, actorURL, author string, limit int) ([]*Post, error) {
	if limit <= 0 {
		return nil, nil
	}
	return oc.walk(ctx, actorURL, author, limit, func(*Post) bool { return true })
}

func (oc *outboxClient) getInTimeframe(ctx context.Context, actorURL, author string, start, end time.Time, limit int) ([]*Post, error) {
	if limit <= 0 {
		return nil, nil
	}
	return oc.walk(ctx, actorURL, author, limit, func(p *Post) bool {
		return !p.CreatedAt.Before(start) && !p.CreatedAt.After(end)
	})
}

func (oc *outboxClient) walk(ctx context.Context, actorURL, author string, limit int, include func(*Post) bool) ([]*Post, error) {
	actor, err := oc.resolveActor(ctx, actorURL, author)
	if err != nil {
		return nil, err
	}
	if actor.Outbox == "" {
		return nil, fmt.Errorf("actor %q exposes no outbox", actorURL)
	}

	var posts []*Post
	seen := make(map[string]struct{})

	cur, err := oc.firstPage(ctx, string(actor.Outbox))
	if err != nil {
		return nil, err
	}

	for pages := 0; pages < outboxMaxPages; pages++ {
		if ctx.Err() != nil {
			return nil, ctx.Err()
		}
		if len(posts) >= limit {
			break
		}

		page, ok, err := oc.getPage(ctx, cur)
		if err != nil {
			return nil, err
		}
		if !ok {
			break
		}

		var items []apActivity
		if len(page.OrderedItems) > 0 {
			if err := json.Unmarshal(page.OrderedItems, &items); err != nil {
				return nil, fmt.Errorf("failed to decode outbox page items: %w", err)
			}
		}

		for _, item := range items {
			p, valid, err := oc.itemToPost(ctx, item, author)
			if err != nil {
				return nil, err
			}
			if !valid || p == nil {
				continue
			}
			if _, dup := seen[p.ID]; dup {
				continue
			}
			seen[p.ID] = struct{}{}
			if include(p) {
				posts = append(posts, p)
				if len(posts) >= limit {
					return posts, nil
				}
			}
		}

		cur = page.Next
		if cur.IRI == "" && len(cur.Inline) == 0 {
			break
		}
	}

	return posts, nil
}

func (oc *outboxClient) firstPage(ctx context.Context, outbox string) (apRef, error) {
	var coll apCollection
	if err := oc.getJSON(ctx, outbox, &coll); err != nil {
		return apRef{}, fmt.Errorf("failed to fetch outbox %q: %w", outbox, err)
	}
	if coll.Type != "" && coll.Type != "OrderedCollection" {
		return apRef{}, fmt.Errorf("outbox %q has unexpected type %q", outbox, coll.Type)
	}
	if coll.First.IRI == "" && len(coll.First.Inline) == 0 {
		return apRef{}, fmt.Errorf("outbox %q exposes no pages", outbox)
	}
	return coll.First, nil
}

func (oc *outboxClient) getPage(ctx context.Context, ref apRef) (apPage, bool, error) {
	var page apPage
	switch {
	case ref.IRI != "":
		if err := oc.getJSON(ctx, ref.IRI, &page); err != nil {
			return apPage{}, false, fmt.Errorf("failed to fetch outbox page %q: %w", ref.IRI, err)
		}
	case len(ref.Inline) > 0:
		if err := json.Unmarshal(ref.Inline, &page); err != nil {
			return apPage{}, false, fmt.Errorf("failed to decode inline outbox page: %w", err)
		}
	default:
		return apPage{}, false, nil
	}
	return page, true, nil
}

func (oc *outboxClient) itemToPost(ctx context.Context, item apActivity, author string) (*Post, bool, error) {
	var note *apNote
	var err error

	switch item.Type {
	case "Create", "Update":
		note, err = oc.noteFromObject(ctx, item.Object)
	case "Note", "Article", "Question", "Page", "":
		note = &apNote{
			ID:        item.ID,
			Type:      item.Type,
			Published: item.Published,
			Content:   item.Content,
			URL:       item.URL,
		}
	default:
		return nil, false, nil
	}
	if err != nil {
		return nil, false, err
	}
	if note == nil || !isPostNote(note.Type) {
		return nil, false, nil
	}

	id := note.ID
	if id == "" && len(note.URL) > 0 {
		id = note.URL[0]
	}
	if id == "" {
		return nil, false, nil
	}

	created, ok := parseActivityTime(note.Published)
	if !ok || created.IsZero() {
		return nil, false, nil
	}

	return &Post{
		ID:        id,
		Author:    author,
		Content:   note.Content,
		CreatedAt: created,
	}, true, nil
}

func (oc *outboxClient) noteFromObject(ctx context.Context, raw json.RawMessage) (*apNote, error) {
	raw = bytes.TrimSpace(raw)
	if len(raw) == 0 || string(raw) == "null" {
		return nil, nil
	}
	if raw[0] == '"' {
		var id string
		if err := json.Unmarshal(raw, &id); err != nil {
			return nil, err
		}
		return oc.fetchNote(ctx, id)
	}
	var n apNote
	if err := json.Unmarshal(raw, &n); err != nil {
		return nil, err
	}
	return &n, nil
}

func (oc *outboxClient) fetchNote(ctx context.Context, iri string) (*apNote, error) {
	var n apNote
	if err := oc.getJSON(ctx, iri, &n); err != nil {
		return nil, fmt.Errorf("failed to dereference object %q: %w", iri, err)
	}
	return &n, nil
}

func (oc *outboxClient) resolveNote(ctx context.Context, url string) (*apNote, error) {
	var act apActivity
	if err := oc.getJSON(ctx, url, &act); err != nil {
		return nil, fmt.Errorf("failed to dereference status %q: %w", url, err)
	}
	switch act.Type {
	case "Create":
		return oc.noteFromObject(ctx, act.Object)
	case "Note", "Article", "Question", "Page", "":
		return &apNote{
			ID:        act.ID,
			Type:      act.Type,
			Published: act.Published,
			Content:   act.Content,
			URL:       act.URL,
		}, nil
	default:
		return nil, fmt.Errorf("document at %q is not a post (type %q)", url, act.Type)
	}
}

func isPostNote(t string) bool {
	switch t {
	case "", "Note", "Article", "Question", "Page":
		return true
	}
	return false
}

func parseActivityTime(s string) (time.Time, bool) {
	if s == "" {
		return time.Time{}, false
	}
	for _, layout := range []string{time.RFC3339, time.RFC3339Nano, "2006-01-02T15:04:05"} {
		if t, err := time.Parse(layout, s); err == nil {
			return t, true
		}
	}
	return time.Time{}, false
}

func noteToInternalPost(n *apNote, author string) *internalPost {
	p := &internalPost{
		ID:      n.ID,
		URI:     n.ID,
		Content: n.Content,
	}
	if len(n.URL) > 0 {
		p.URL = n.URL[0]
	}
	if t, ok := parseActivityTime(n.Published); ok {
		p.CreatedAt = t
	}
	if author != "" {
		p.Account = Account{Acct: author}
	}
	return p
}

func mergePostsBySource(cached []*internalPost, outbox []*Post) []*Post {
	seen := make(map[string]struct{}, len(cached))
	res := make([]*Post, 0, len(cached)+len(outbox))
	for _, c := range cached {
		key := c.URI
		if key == "" {
			key = c.ID
		}
		if key != "" {
			seen[key] = struct{}{}
		}
		if c != nil {
			res = append(res, c.toPost())
		}
	}
	for _, p := range outbox {
		if p == nil || p.ID == "" {
			continue
		}
		if _, dup := seen[p.ID]; dup {
			continue
		}
		seen[p.ID] = struct{}{}
		res = append(res, p)
	}
	sort.SliceStable(res, func(i, j int) bool { return res[i].CreatedAt.After(res[j].CreatedAt) })
	return res
}