johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package market

import (
	"cmp"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"net/http"
	"net/url"
	"strings"
	"time"

	"github.com/bluesky-social/indigo/atproto/identity"
	"github.com/bluesky-social/indigo/atproto/syntax"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/atproto/xrpc"
)

type HTTPDoer interface {
	Do(req *http.Request) (*http.Response, error)
}

type HTTPReader struct {
	Directory identity.Directory
	Client    HTTPDoer
}

func NewHTTPReader(dir identity.Directory) *HTTPReader {
	return &HTTPReader{Directory: dir, Client: http.DefaultClient}
}

func (h *HTTPReader) doer() HTTPDoer {
	if h.Client != nil {
		return h.Client
	}
	return http.DefaultClient
}

func (h *HTTPReader) PDSEndpoint(ctx context.Context, did string) (string, error) {
	if h.Directory == nil {
		return "", fmt.Errorf("market: no identity directory")
	}
	parsed, err := syntax.ParseDID(did)
	if err != nil {
		return "", fmt.Errorf("market: %w", err)
	}
	ident, err := h.Directory.LookupDID(ctx, parsed)
	if err != nil {
		return "", fmt.Errorf("market: resolve %s: %w", did, err)
	}
	endpoint := ident.PDSEndpoint()
	if endpoint == "" {
		return "", fmt.Errorf("market: no PDS endpoint for %s", did)
	}
	return strings.TrimRight(endpoint, "/"), nil
}

type doerTransport struct {
	doer HTTPDoer
}

func (t doerTransport) RoundTrip(req *http.Request) (*http.Response, error) {
	return t.doer.Do(req)
}

func (h *HTTPReader) query(ctx context.Context, pds, nsid string, params url.Values, out any) error {
	client := &xrpc.Client{Host: pds, HTTP: &http.Client{Transport: doerTransport{doer: h.doer()}}}
	var lastErr error
	for attempt := range 5 {
		reply, err := xrpc.Query[xrpc.Reply](ctx, client, nsid, params)
		if err == nil {
			if err := json.Unmarshal(reply.Body, out); err != nil {
				return fmt.Errorf("%w: %v", ErrInvalidRecord, err)
			}
			return nil
		}
		if failure, ok := errors.AsType[*xrpc.Error](err); ok && failure.Status < 500 && failure.Status != http.StatusTooManyRequests {
			return fmt.Errorf("%w: %s: HTTP %d: %w", ErrNotFound, nsid, failure.Status, failure)
		}
		lastErr = err
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-time.After(time.Duration(attempt+1) * 100 * time.Millisecond):
		}
	}
	return lastErr
}

func (h *HTTPReader) GetRecord(ctx context.Context, repo, collection, rkey string) (ResolvedRecord, error) {
	pds, err := h.PDSEndpoint(ctx, repo)
	if err != nil {
		return ResolvedRecord{}, err
	}
	var body struct {
		URI   string         `json:"uri"`
		CID   string         `json:"cid"`
		Value map[string]any `json:"value"`
	}
	params := url.Values{"repo": {repo}, "collection": {collection}, "rkey": {rkey}}
	if err := h.query(ctx, pds, "com.atproto.repo.getRecord", params, &body); err != nil {
		return ResolvedRecord{}, fmt.Errorf("%s/%s/%s: %w", repo, collection, rkey, err)
	}
	if body.Value == nil {
		return ResolvedRecord{}, fmt.Errorf("%w: no record value", ErrNotFound)
	}
	body.URI = cmp.Or(body.URI, "at://"+repo+"/"+collection+"/"+rkey)
	return ResolvedRecord{URI: body.URI, CID: body.CID, Value: body.Value}, nil
}

func (h *HTTPReader) ListRecords(ctx context.Context, repo, collection string, limit int, cursor string) (ListPage, error) {
	pds, err := h.PDSEndpoint(ctx, repo)
	if err != nil {
		return ListPage{}, err
	}
	if limit <= 0 || limit > 100 {
		limit = 100
	}
	params := url.Values{"repo": {repo}, "collection": {collection}, "limit": {fmt.Sprintf("%d", limit)}}
	if cursor != "" {
		params.Set("cursor", cursor)
	}
	var body struct {
		Records []struct {
			URI   string         `json:"uri"`
			CID   string         `json:"cid"`
			Value map[string]any `json:"value"`
		} `json:"records"`
		Cursor string `json:"cursor"`
	}
	if err := h.query(ctx, pds, "com.atproto.repo.listRecords", params, &body); err != nil {
		return ListPage{}, fmt.Errorf("list %s/%s: %w", repo, collection, err)
	}
	page := ListPage{Cursor: body.Cursor, Records: make([]ListedRecord, 0, len(body.Records))}
	for _, rec := range body.Records {
		page.Records = append(page.Records, ListedRecord{URI: rec.URI, CID: rec.CID, Value: rec.Value})
	}
	return page, nil
}

type recordResolver struct {
	remote RecordGetter
	ownDID string
	own    RecordGetter
}

func NewRecordResolver(remote RecordGetter, ownDID string, own RecordGetter) RecordResolver {
	return &recordResolver{remote: remote, ownDID: ownDID, own: own}
}

func (r *recordResolver) Resolve(ctx context.Context, ref RecordRef) (ResolvedRecord, error) {
	aturi, err := syntax.ParseATURI(ref.URI)
	if err != nil {
		return ResolvedRecord{}, fmt.Errorf("%w: %v", ErrInvalidRef, err)
	}
	repo := string(aturi.Authority())
	collection := string(aturi.Collection())
	rkey := string(aturi.RecordKey())
	if repo == "" || collection == "" || rkey == "" {
		return ResolvedRecord{}, fmt.Errorf("%w: %q is not a record AT URI", ErrInvalidRef, ref.URI)
	}
	if r.own != nil && repo == r.ownDID {
		rec, err := r.own.GetRecord(ctx, repo, collection, rkey)
		if err != nil {
			return ResolvedRecord{}, fmt.Errorf("getRecord failed: %s is not in the own repo: %w", ref.URI, err)
		}
		rec.CID = cmp.Or(rec.CID, ref.CID)
		return rec, nil
	}
	if r.remote == nil {
		return ResolvedRecord{}, fmt.Errorf("%w: no remote record getter", ErrNotFound)
	}
	rec, err := r.remote.GetRecord(ctx, repo, collection, rkey)
	if err != nil {
		return ResolvedRecord{}, err
	}
	rec.CID = cmp.Or(rec.CID, ref.CID)
	return rec, nil
}

func (h *HTTPReader) Resolve(ctx context.Context, ref RecordRef) (ResolvedRecord, error) {
	aturi, err := syntax.ParseATURI(ref.URI)
	if err != nil {
		return ResolvedRecord{}, fmt.Errorf("%w: %v", ErrInvalidRef, err)
	}
	repo := string(aturi.Authority())
	collection := string(aturi.Collection())
	rkey := string(aturi.RecordKey())
	if repo == "" || collection == "" || rkey == "" {
		return ResolvedRecord{}, fmt.Errorf("%w: %q is not a record AT URI", ErrInvalidRef, ref.URI)
	}
	rec, err := h.GetRecord(ctx, repo, collection, rkey)
	if err != nil {
		return ResolvedRecord{}, err
	}
	rec.CID = cmp.Or(rec.CID, ref.CID)
	return rec, nil
}