johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package local

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"maps"
	"strings"

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

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

const backfillPage = 500

var errBehind = errors.New("pds: the subscription fell behind the sequencer")

type Repo struct {
	srv *atpds.Server

	did syntax.DID
}

func openRepo(ctx context.Context, srv *atpds.Server, raw string) (*Repo, error) {
	did, err := syntax.ParseDID(raw)
	if err != nil {
		return nil, fmt.Errorf("pds: the repo did %q: %w", raw, err)
	}
	exists, err := srv.Repos().RepoExists(ctx, did)
	if err != nil {
		return nil, err
	}
	if !exists {
		return nil, fmt.Errorf("%w: the Pds hosts no repo %s", atprotorepo.ErrNotFound, did)
	}
	return &Repo{srv: srv, did: did}, nil
}

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

func notFound(err error) error {
	if errors.Is(err, repo.ErrRecordNotFound) {
		return fmt.Errorf("%w: %w: %v", atprotorepo.ErrNotFound, atprotorepo.ErrRecordNotFound, err)
	}
	if errors.Is(err, repo.ErrRepoNotFound) {
		return fmt.Errorf("%w: %v", atprotorepo.ErrNotFound, err)
	}
	return err
}

func (r *Repo) target(raw string) (syntax.DID, error) {
	raw = strings.TrimSpace(raw)
	if raw == "" {
		return r.did, nil
	}
	return syntax.ParseDID(raw)
}

func toRecord(did syntax.DID, rec *repo.Record) (atprotorepo.Record, error) {
	value, err := json.Marshal(rec.Value)
	if err != nil {
		return atprotorepo.Record{}, fmt.Errorf("pds: encoding the record value: %w", err)
	}
	return atprotorepo.Record{URI: "at://" + did.String() + "/" + rec.Collection.String() + "/" + rec.Rkey.String(), CID: rec.CID.String(),
		Collection: rec.Collection.String(), RKey: rec.Rkey.String(), Value: value}, nil
}

func (r *Repo) GetRecord(ctx context.Context, repoDID, collection, rkey string) (atprotorepo.Record, error) {
	did, err := r.target(repoDID)
	if err != nil {
		return atprotorepo.Record{}, err
	}
	rec, err := r.srv.Repos().GetRecord(ctx, did, syntax.NSID(collection), syntax.RecordKey(rkey))
	if err != nil {
		return atprotorepo.Record{}, notFound(err)
	}
	return toRecord(did, rec)
}

func (r *Repo) ListRecords(ctx context.Context, repoDID, collection string, opts atprotorepo.ListOptions) (atprotorepo.ListPage, error) {
	if opts.Reverse {
		return atprotorepo.ListPage{}, errors.New("pds: the in-process repo lists records in key order only")
	}
	did, err := r.target(repoDID)
	if err != nil {
		return atprotorepo.ListPage{}, err
	}
	records, next, err := r.srv.Repos().ListRecords(ctx, did, syntax.NSID(collection), syntax.RecordKey(opts.Cursor), opts.Limit)
	if err != nil {
		return atprotorepo.ListPage{}, notFound(err)
	}
	page := atprotorepo.ListPage{Records: make([]atprotorepo.Record, 0, len(records))}
	for i := range records {
		rec, err := toRecord(did, &records[i])
		if err != nil {
			return atprotorepo.ListPage{}, err
		}
		page.Records = append(page.Records, rec)
	}
	if next != nil {
		page.Cursor = next.String()
	}
	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("pds: 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 decodeValue(collection string, value json.RawMessage) (map[string]any, error) {
	var record map[string]any
	if err := json.Unmarshal(value, &record); err != nil {
		return nil, fmt.Errorf("pds: the record is not a JSON object: %w", err)
	}
	if record == nil {
		return nil, errors.New("pds: the record is empty")
	}
	if _, ok := record["$type"]; !ok {
		out := make(map[string]any, len(record)+1)
		out["$type"] = collection
		maps.Copy(out, record)
		record = out
	}
	return record, nil
}

func (r *Repo) open(ctx context.Context) (repo.Repo, error) {
	keys := r.srv.KeyStore()
	if keys == nil {
		return nil, repo.ErrReadOnly
	}
	key, err := keys.GetRepoKey(ctx, r.did)
	if err != nil {
		return nil, fmt.Errorf("pds: the signing key of %s: %w", r.did, err)
	}
	return r.srv.Repos().OpenRepo(ctx, r.did, key)
}

func (r *Repo) write(ctx context.Context, action repo.WriteAction, collection, rkey string, value json.RawMessage, prev *cid.Cid) (commonv1.StrongRef, error) {
	nsid, err := syntax.ParseNSID(collection)
	if err != nil {
		return commonv1.StrongRef{}, fmt.Errorf("pds: the collection %q: %w", collection, err)
	}
	var key syntax.RecordKey
	if rkey != "" {
		if key, err = syntax.ParseRecordKey(rkey); err != nil {
			return commonv1.StrongRef{}, fmt.Errorf("pds: the rkey %q: %w", rkey, err)
		}
	}
	record, err := decodeValue(collection, value)
	if err != nil {
		return commonv1.StrongRef{}, err
	}
	handle, err := r.open(ctx)
	if err != nil {
		return commonv1.StrongRef{}, err
	}
	ev, err := handle.ApplyWrites(ctx, []repo.WriteOp{{Action: action, Collection: nsid, Rkey: key, Record: record, Prev: prev}}, nil)
	if errors.Is(err, repo.ErrInvalidSwap) {
		return commonv1.StrongRef{}, fmt.Errorf("%w: %v", atprotorepo.ErrSwapMismatch, err)
	}
	if err != nil {
		return commonv1.StrongRef{}, err
	}
	if ev != nil {
		for _, op := range ev.Ops {
			if op.Action != repo.ActionDelete && op.CID != nil {
				return commonv1.StrongRef{URI: "at://" + r.did.String() + "/" + op.Path, CID: op.CID.String()}, nil
			}
		}
	}
	if key == "" {
		return commonv1.StrongRef{}, errors.New("pds: the write produced no record")
	}
	held, err := handle.GetRecord(ctx, nsid, key)
	if err != nil {
		return commonv1.StrongRef{}, err
	}
	return commonv1.StrongRef{URI: "at://" + r.did.String() + "/" + nsid.String() + "/" + key.String(), CID: held.CID.String()}, nil
}

func (r *Repo) CreateRecord(ctx context.Context, collection, rkey string, value json.RawMessage) (commonv1.StrongRef, error) {
	return r.write(ctx, repo.ActionCreate, collection, rkey, value, nil)
}

func (r *Repo) PutRecord(ctx context.Context, collection, rkey string, value json.RawMessage) (commonv1.StrongRef, error) {
	if strings.TrimSpace(rkey) == "" {
		return commonv1.StrongRef{}, errors.New("pds: putRecord needs an rkey")
	}
	return r.write(ctx, repo.WriteAction("put"), collection, rkey, value, nil)
}

func (r *Repo) PutRecordSwap(ctx context.Context, collection, rkey string, value json.RawMessage, swapCID string) (commonv1.StrongRef, error) {
	if strings.TrimSpace(rkey) == "" {
		return commonv1.StrongRef{}, errors.New("pds: putRecord needs an rkey")
	}
	prev, err := cid.Decode(strings.TrimSpace(swapCID))
	if err != nil {
		return commonv1.StrongRef{}, fmt.Errorf("pds: the swap cid %q: %w", swapCID, err)
	}
	return r.write(ctx, repo.WriteAction("put"), collection, rkey, value, &prev)
}

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

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

func (r *Repo) deleteRecord(ctx context.Context, collection, rkey string, prev *cid.Cid) error {
	nsid, err := syntax.ParseNSID(collection)
	if err != nil {
		return fmt.Errorf("pds: the collection %q: %w", collection, err)
	}
	key, err := syntax.ParseRecordKey(rkey)
	if err != nil {
		return fmt.Errorf("pds: the rkey %q: %w", rkey, err)
	}
	handle, err := r.open(ctx)
	if err != nil {
		return err
	}
	if _, err := handle.ApplyWrites(ctx, []repo.WriteOp{{Action: repo.ActionDelete, Collection: nsid, Rkey: key, Prev: prev}}, nil); err != nil {
		if errors.Is(err, repo.ErrInvalidSwap) {
			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 {
	return follow(ctx, r.srv, opts.Cursor, func(ev *repo.CommitEvent) error {
		if ev.Repo != r.did {
			return nil
		}
		commit := toCommit(ev, opts.Collections)
		if len(commit.Ops) == 0 && len(ev.Ops) > 0 {
			return nil
		}
		return fn(commit)
	})
}

func follow(ctx context.Context, srv *atpds.Server, cursor *int64, fn func(*repo.CommitEvent) error) error {
	live, cancel := srv.Sequencer().Subscribe()
	defer cancel()
	last := int64(-1)
	if cursor != nil {
		last = *cursor
		for {
			events, err := srv.Repos().ListEvents(ctx, last, backfillPage)
			if err != nil {
				return err
			}
			for _, ev := range events {
				if ev.Seq <= last {
					continue
				}
				if err := fn(ev); err != nil {
					return err
				}
				last = ev.Seq
			}
			if len(events) < backfillPage {
				break
			}
		}
	}
	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		case ev, ok := <-live:
			if !ok {
				return errBehind
			}
			if ev.Seq <= last {
				continue
			}
			if err := fn(ev); err != nil {
				return err
			}
			last = ev.Seq
		}
	}
}

func toCommit(ev *repo.CommitEvent, 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 && !containsFold(collections, collection) {
			continue
		}
		next := atprotorepo.Op{Action: string(op.Action), Collection: collection, RKey: rkey}
		if op.CID != nil {
			next.CID = op.CID.String()
		}
		out.Ops = append(out.Ops, next)
	}
	return out
}

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