johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package market

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

	"github.com/bluesky-social/indigo/atproto/syntax"

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

type Foreign interface {
	RecordGetter
	RecordLister
	RecordResolver
}

type RepoStoreConfig struct {
	DID string

	Open func(ctx context.Context) (atprotorepo.Repo, error)

	Foreign Foreign
}

type RepoStore struct {
	did string

	open func(ctx context.Context) (atprotorepo.Repo, error)

	foreign Foreign
}

var (
	_ RecordWriter   = (*RepoStore)(nil)
	_ RecordGetter   = (*RepoStore)(nil)
	_ RecordLister   = (*RepoStore)(nil)
	_ RecordResolver = (*RepoStore)(nil)
)

func NewRepoStore(cfg RepoStoreConfig) *RepoStore {
	return &RepoStore{did: strings.TrimSpace(cfg.DID), open: cfg.Open, foreign: cfg.Foreign}
}

func (s *RepoStore) DID() string { return s.did }

func (s *RepoStore) repo(ctx context.Context, did string) (atprotorepo.Repo, error) {
	if s.open == nil {
		return nil, errors.New("market: the repo store has no repo handle")
	}
	if did != "" && did != s.did {
		return nil, fmt.Errorf("%w: the repo store writes %s, not %s", ErrInvalidRef, s.did, did)
	}
	return s.open(ctx)
}

func (s *RepoStore) local(repo string) bool {
	return s.foreign == nil || repo == "" || repo == s.did
}

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

func encodeValue(record map[string]any) (json.RawMessage, error) {
	raw, err := json.Marshal(record)
	if err != nil {
		return nil, fmt.Errorf("market: encoding a record: %w", err)
	}
	return raw, nil
}

func decodeValue(raw json.RawMessage) (map[string]any, error) {
	trimmed := bytes.TrimSpace(raw)
	if len(trimmed) == 0 || bytes.Equal(trimmed, []byte("null")) {
		return nil, fmt.Errorf("%w: the record has no value", ErrNotFound)
	}
	var value map[string]any
	if err := json.Unmarshal(trimmed, &value); err != nil {
		return nil, fmt.Errorf("market: decoding a record value: %w", err)
	}
	if value == nil {
		return nil, fmt.Errorf("%w: the record has no value", ErrNotFound)
	}
	return value, nil
}

func (s *RepoStore) write(ctx context.Context, repoDID, collection, rkey string, record map[string]any, create bool) (StrongRef, error) {
	repo, err := s.repo(ctx, repoDID)
	if err != nil {
		return StrongRef{}, err
	}
	raw, err := encodeValue(record)
	if err != nil {
		return StrongRef{}, err
	}
	var ref commonv1.StrongRef
	if create {
		ref, err = repo.CreateRecord(ctx, collection, rkey, raw)
	} else {
		ref, err = repo.PutRecord(ctx, collection, rkey, raw)
	}
	if err != nil {
		return StrongRef{}, fmt.Errorf("market: writing %s/%s/%s: %w", s.did, collection, rkey, err)
	}
	return NewStrongRef(ref.URI, ref.CID), nil
}

func (s *RepoStore) CreateRecord(ctx context.Context, repo, collection string, record map[string]any) (StrongRef, error) {
	return s.write(ctx, repo, collection, "", record, true)
}

func (s *RepoStore) CreateRecordAt(ctx context.Context, repo, collection, rkey string, record map[string]any) (StrongRef, error) {
	return s.write(ctx, repo, collection, rkey, record, true)
}

func (s *RepoStore) PutRecord(ctx context.Context, repo, collection, rkey string, record map[string]any) (StrongRef, error) {
	if rkey == "" {
		rkey = syntax.NewTIDNow(0).String()
	}
	return s.write(ctx, repo, collection, rkey, record, false)
}

func (s *RepoStore) PutRecordSwap(ctx context.Context, repoDID, collection, rkey string, record map[string]any, swapCID string) (StrongRef, error) {
	repo, err := s.repo(ctx, repoDID)
	if err != nil {
		return StrongRef{}, err
	}
	swapper, ok := repo.(atprotorepo.SwapPutter)
	if !ok {
		return StrongRef{}, errors.New("market: the repo cannot put a record by its cid")
	}
	raw, err := encodeValue(record)
	if err != nil {
		return StrongRef{}, err
	}
	ref, err := swapper.PutRecordSwap(ctx, collection, rkey, raw, swapCID)
	if errors.Is(err, atprotorepo.ErrSwapMismatch) {
		return StrongRef{}, fmt.Errorf("%w: %s/%s/%s: %v", ErrSwapMismatch, s.did, collection, rkey, err)
	}
	if err != nil {
		return StrongRef{}, fmt.Errorf("market: writing %s/%s/%s: %w", s.did, collection, rkey, err)
	}
	return NewStrongRef(ref.URI, ref.CID), nil
}

func (s *RepoStore) DeleteRecord(ctx context.Context, repoDID, collection, rkey string) error {
	repo, err := s.repo(ctx, repoDID)
	if err != nil {
		return err
	}
	return mapError(repo.DeleteRecord(ctx, collection, rkey))
}

func (s *RepoStore) GetRecord(ctx context.Context, repoDID, collection, rkey string) (ResolvedRecord, error) {
	if !s.local(repoDID) {
		return s.foreign.GetRecord(ctx, repoDID, collection, rkey)
	}
	repo, err := s.repo(ctx, "")
	if err != nil {
		return ResolvedRecord{}, err
	}
	rec, err := repo.GetRecord(ctx, repoDID, collection, rkey)
	if err != nil {
		return ResolvedRecord{}, mapError(err)
	}
	value, err := decodeValue(rec.Value)
	if err != nil {
		return ResolvedRecord{}, err
	}
	return ResolvedRecord{URI: rec.URI, CID: rec.CID, Value: value}, nil
}

func (s *RepoStore) ListRecords(ctx context.Context, repoDID, collection string, limit int, cursor string) (ListPage, error) {
	if !s.local(repoDID) {
		return s.foreign.ListRecords(ctx, repoDID, collection, limit, cursor)
	}
	repo, err := s.repo(ctx, "")
	if err != nil {
		return ListPage{}, err
	}
	if limit <= 0 || limit > 100 {
		limit = 100
	}
	page, err := repo.ListRecords(ctx, repoDID, collection, atprotorepo.ListOptions{Limit: limit, Cursor: cursor})
	if err != nil {
		return ListPage{}, mapError(err)
	}
	out := ListPage{Cursor: page.Cursor, Records: make([]ListedRecord, 0, len(page.Records))}
	for _, rec := range page.Records {
		value, err := decodeValue(rec.Value)
		if err != nil {
			return ListPage{}, err
		}
		out.Records = append(out.Records, ListedRecord{URI: rec.URI, CID: rec.CID, Value: value})
	}
	return out, nil
}

func (s *RepoStore) 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, collection, rkey := aturi.Authority().String(), aturi.Collection().String(), aturi.RecordKey().String()
	if repo == "" || collection == "" || rkey == "" {
		return ResolvedRecord{}, fmt.Errorf("%w: %q is not a record AT URI", ErrInvalidRef, ref.URI)
	}
	if !s.local(repo) {
		return s.foreign.Resolve(ctx, ref)
	}
	rec, err := s.GetRecord(ctx, repo, collection, rkey)
	if err != nil {
		return ResolvedRecord{}, err
	}
	if rec.CID == "" {
		rec.CID = ref.CID
	}
	return rec, nil
}