johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package remote

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

	"github.com/bluesky-social/indigo/atproto/syntax"
	"github.com/ipfs/go-cid"

	commonv1 "github.com/johnandersen777/socialweb-computer-kcp/api/common/v1alpha1"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/atproto/pdsclient"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/atproto/xrpc"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/contract/atprotorepo"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/contract/pds"
)

const DefaultTimeout = 30 * time.Second

type Config struct {
	DID string

	Transport pds.Transport

	Auth xrpc.Auth

	Timeout time.Duration
}

type Repo struct {
	pds pdsclient.PDS

	did syntax.DID
}

var (
	_ atprotorepo.Repo         = (*Repo)(nil)
	_ atprotorepo.RecordReader = (*Repo)(nil)
	_ atprotorepo.RecordWriter = (*Repo)(nil)
	_ atprotorepo.Subscriber   = (*Repo)(nil)
	_ atprotorepo.SwapDeleter  = (*Repo)(nil)
	_ atprotorepo.SwapPutter   = (*Repo)(nil)
)

func New(cfg Config) (*Repo, error) {
	did, err := syntax.ParseDID(strings.TrimSpace(cfg.DID))
	if err != nil {
		return nil, fmt.Errorf("remote: the repo did %q: %w", cfg.DID, err)
	}
	if cfg.Transport == nil {
		return nil, errors.New("remote: the repo has no PDS transport")
	}
	timeout := cfg.Timeout
	if timeout <= 0 {
		timeout = DefaultTimeout
	}
	client, err := pdsclient.NewRemote(pdsclient.RemoteConfig{BaseURL: cfg.Transport.Address(), HTTPClient: cfg.Transport.Client(),
		TLS: cfg.Transport.TLS(), Auth: cfg.Auth, Timeout: timeout})
	if err != nil {
		return nil, err
	}
	return Wrap(client, did), nil
}

func Wrap(client pdsclient.PDS, did syntax.DID) *Repo {
	return &Repo{pds: client, did: did}
}

func (r *Repo) DID() string { return r.did.String() }

func notFound(err error) error {
	var failure *pdsclient.XRPCError
	if errors.As(err, &failure) && failure.Code == "RecordNotFound" {
		return fmt.Errorf("%w: %w: %v", atprotorepo.ErrNotFound, atprotorepo.ErrRecordNotFound, err)
	}
	if errors.As(err, &failure) && (failure.StatusCode == http.StatusNotFound || failure.Code == "RepoNotFound") {
		return fmt.Errorf("%w: %v", atprotorepo.ErrNotFound, err)
	}
	return err
}

func (r *Repo) path(repo, collection, rkey string) (pdsclient.RepoPath, error) {
	if strings.TrimSpace(repo) == "" {
		repo = r.did.String()
	}
	return pdsclient.ParseRepoPath(repo, collection, rkey)
}

func toRecord(rec pdsclient.Record) atprotorepo.Record {
	return atprotorepo.Record{URI: rec.URI.String(), CID: rec.CID.String(), Collection: rec.URI.Collection().String(),
		RKey: rec.URI.RecordKey().String(), Value: rec.Value}
}

func (r *Repo) GetRecord(ctx context.Context, repo, collection, rkey string) (atprotorepo.Record, error) {
	path, err := r.path(repo, collection, rkey)
	if err != nil {
		return atprotorepo.Record{}, err
	}
	rec, err := r.pds.GetRecord(ctx, path, nil)
	if err != nil {
		return atprotorepo.Record{}, notFound(err)
	}
	return toRecord(rec), nil
}

func (r *Repo) ListRecords(ctx context.Context, repo, collection string, opts atprotorepo.ListOptions) (atprotorepo.ListPage, error) {
	path, err := r.path(repo, collection, "")
	if err != nil {
		return atprotorepo.ListPage{}, err
	}
	limit := opts.Limit
	if limit <= 0 || limit > 100 {
		limit = 100
	}
	out, err := r.pds.ListRecords(ctx, pdsclient.ListRecordsRequest{Repo: path.Repo, Collection: path.Collection, Cursor: opts.Cursor,
		Limit: limit, Reverse: opts.Reverse})
	if err != nil {
		return atprotorepo.ListPage{}, notFound(err)
	}
	page := atprotorepo.ListPage{Cursor: out.Cursor, Records: make([]atprotorepo.Record, 0, len(out.Records))}
	for _, rec := range out.Records {
		page.Records = append(page.Records, toRecord(rec))
	}
	return page, nil
}

func (r *Repo) Resolve(ctx context.Context, ref commonv1.StrongRef) (atprotorepo.Record, error) {
	uri, err := syntax.ParseATURI(ref.URI)
	if err != nil {
		return atprotorepo.Record{}, fmt.Errorf("remote: the record uri %q: %w", ref.URI, err)
	}
	rec, err := r.GetRecord(ctx, uri.Authority().String(), uri.Collection().String(), uri.RecordKey().String())
	if err != nil {
		return atprotorepo.Record{}, err
	}
	if ref.CID != "" && rec.CID != ref.CID {
		return atprotorepo.Record{}, fmt.Errorf("%w: %s holds %s, not %s", atprotorepo.ErrNotFound, ref.URI, rec.CID, ref.CID)
	}
	return rec, nil
}

func (r *Repo) CreateRecord(ctx context.Context, collection, rkey string, value json.RawMessage) (commonv1.StrongRef, error) {
	path, err := r.path("", collection, rkey)
	if err != nil {
		return commonv1.StrongRef{}, err
	}
	out, err := r.pds.CreateRecord(ctx, path, value, pdsclient.WriteOptions{})
	if err != nil {
		return commonv1.StrongRef{}, err
	}
	return commonv1.StrongRef{URI: out.URI.String(), CID: out.CID.String()}, nil
}

func (r *Repo) PutRecord(ctx context.Context, collection, rkey string, value json.RawMessage) (commonv1.StrongRef, error) {
	return r.putRecord(ctx, collection, rkey, value, pdsclient.WriteOptions{})
}

func (r *Repo) PutRecordSwap(ctx context.Context, collection, rkey string, value json.RawMessage, swapCID string) (commonv1.StrongRef, error) {
	swap, err := cid.Decode(strings.TrimSpace(swapCID))
	if err != nil {
		return commonv1.StrongRef{}, fmt.Errorf("remote: the swap cid %q: %w", swapCID, err)
	}
	return r.putRecord(ctx, collection, rkey, value, pdsclient.WriteOptions{SwapRecord: &swap})
}

func (r *Repo) putRecord(ctx context.Context, collection, rkey string, value json.RawMessage, opts pdsclient.WriteOptions) (commonv1.StrongRef, error) {
	if strings.TrimSpace(rkey) == "" {
		return commonv1.StrongRef{}, errors.New("remote: putRecord needs an rkey")
	}
	path, err := r.path("", collection, rkey)
	if err != nil {
		return commonv1.StrongRef{}, err
	}
	out, err := r.pds.PutRecord(ctx, path, value, opts)
	if err != nil {
		var failure *pdsclient.XRPCError
		if errors.As(err, &failure) && failure.Code == "InvalidSwap" {
			return commonv1.StrongRef{}, fmt.Errorf("%w: %v", atprotorepo.ErrSwapMismatch, err)
		}
		return commonv1.StrongRef{}, err
	}
	return commonv1.StrongRef{URI: out.URI.String(), CID: out.CID.String()}, nil
}

func (r *Repo) DeleteRecord(ctx context.Context, collection, rkey string) error {
	return r.deleteRecord(ctx, collection, rkey, pdsclient.WriteOptions{})
}

func (r *Repo) DeleteRecordSwap(ctx context.Context, collection, rkey, swapCID string) error {
	swap, err := cid.Decode(strings.TrimSpace(swapCID))
	if err != nil {
		return fmt.Errorf("remote: the swap cid %q: %w", swapCID, err)
	}
	return r.deleteRecord(ctx, collection, rkey, pdsclient.WriteOptions{SwapRecord: &swap})
}

func (r *Repo) deleteRecord(ctx context.Context, collection, rkey string, opts pdsclient.WriteOptions) error {
	path, err := r.path("", collection, rkey)
	if err != nil {
		return err
	}
	if err := r.pds.DeleteRecord(ctx, path, opts); err != nil {
		var failure *pdsclient.XRPCError
		if errors.As(err, &failure) && failure.Code == "InvalidSwap" {
			return fmt.Errorf("%w: %v", atprotorepo.ErrSwapMismatch, err)
		}
		return notFound(err)
	}
	return nil
}

func (r *Repo) Subscribe(ctx context.Context, opts atprotorepo.SubscribeOptions, fn func(atprotorepo.Commit) error) error {
	stream, err := r.pds.SubscribeRepos(ctx, opts.Cursor)
	if err != nil {
		return err
	}
	defer stream.Close()
	for {
		ev, err := stream.Next(ctx)
		if err != nil {
			return err
		}
		if ev.Repo != r.did {
			continue
		}
		commit := Commit(ev, opts.Collections)
		if len(commit.Ops) == 0 && len(ev.Ops) > 0 {
			continue
		}
		if err := fn(commit); err != nil {
			return err
		}
	}
}

func Commit(ev pdsclient.RepoEvent, collections []string) atprotorepo.Commit {
	out := atprotorepo.Commit{Seq: ev.Seq, Repo: ev.Repo.String(), CID: ev.Commit.String(), Rev: ev.Rev.String(), Time: ev.Time}
	for _, op := range ev.Ops {
		collection, rkey, _ := strings.Cut(op.Path, "/")
		if len(collections) > 0 && !contains(collections, collection) {
			continue
		}
		next := atprotorepo.Op{Action: op.Action, Collection: collection, RKey: rkey}
		if op.CID != nil {
			next.CID = op.CID.String()
		}
		out.Ops = append(out.Ops, next)
	}
	return out
}

func contains(list []string, value string) bool {
	for _, entry := range list {
		if strings.EqualFold(strings.TrimSpace(entry), value) {
			return true
		}
	}
	return false
}