johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package repo

import (
	"context"
	"errors"
	"time"

	comatproto "github.com/bluesky-social/indigo/api/atproto"
	"github.com/bluesky-social/indigo/atproto/atcrypto"
	indigorepo "github.com/bluesky-social/indigo/atproto/repo"
	"github.com/bluesky-social/indigo/atproto/syntax"
	lexutil "github.com/bluesky-social/indigo/lex/util"
	"github.com/ipfs/go-cid"
)

const (
	MaxWrites      = 200
	DefaultList    = 50
	MaxList        = 100
	DefaultCommits = 100
	MaxCommits     = 1000
)

var (
	ErrRepoNotFound    = errors.New("repo not found")
	ErrRepoExists      = errors.New("repo already exists")
	ErrCommitNotFound  = errors.New("commit not found")
	ErrRecordNotFound  = errors.New("record not found")
	ErrRecordExists    = errors.New("record already exists")
	ErrInvalidSwap     = errors.New("invalid swap")
	ErrInvalidRequest  = errors.New("invalid request")
	ErrReadOnly        = errors.New("repo has no signing key")
	ErrCorruptRepo     = errors.New("repo is corrupt")
	ErrUnsupportedRepo = errors.New("unsupported repo")
)

type Commit = indigorepo.Commit

type WriteAction string

const (
	ActionCreate WriteAction = "create"
	ActionUpdate WriteAction = "update"
	ActionDelete WriteAction = "delete"

	actionPut WriteAction = "put"
)

type WriteOp struct {
	Action     WriteAction
	Collection syntax.NSID
	Rkey       syntax.RecordKey
	Record     map[string]any
	Prev       *cid.Cid
}

type CommitOp struct {
	Action WriteAction
	Path   string
	CID    *cid.Cid
	Prev   *cid.Cid
}

type CommitEvent struct {
	Repo     syntax.DID
	Seq      int64
	Commit   cid.Cid
	Rev      syntax.TID
	Since    *syntax.TID
	Data     cid.Cid
	PrevData *cid.Cid
	Blocks   []byte
	Ops      []CommitOp
	Blobs    []cid.Cid
	Time     time.Time
}

func (ev *CommitEvent) CommitMessage() *comatproto.SyncSubscribeRepos_Commit {
	ops := make([]*comatproto.SyncSubscribeRepos_RepoOp, 0, len(ev.Ops))
	for _, op := range ev.Ops {
		rop := &comatproto.SyncSubscribeRepos_RepoOp{
			Action: string(op.Action),
			Path:   op.Path,
		}
		if op.CID != nil {
			c := lexutil.LexLink(*op.CID)
			rop.Cid = &c
		}
		if op.Prev != nil {
			p := lexutil.LexLink(*op.Prev)
			rop.Prev = &p
		}
		ops = append(ops, rop)
	}
	var since *string
	if ev.Since != nil {
		s := ev.Since.String()
		since = &s
	}
	var prevData *lexutil.LexLink
	if ev.PrevData != nil {
		p := lexutil.LexLink(*ev.PrevData)
		prevData = &p
	}
	blobs := make([]lexutil.LexLink, 0, len(ev.Blobs))
	for _, b := range ev.Blobs {
		blobs = append(blobs, lexutil.LexLink(b))
	}
	return &comatproto.SyncSubscribeRepos_Commit{
		Blobs:    blobs,
		Blocks:   lexutil.LexBytes(ev.Blocks),
		Commit:   lexutil.LexLink(ev.Commit),
		Ops:      ops,
		PrevData: prevData,
		Repo:     ev.Repo.String(),
		Rev:      ev.Rev.String(),
		Seq:      ev.Seq,
		Since:    since,
		Time:     ev.Time.UTC().Format("2006-01-02T15:04:05.000Z"),
	}
}

type Record struct {
	Collection syntax.NSID
	Rkey       syntax.RecordKey
	CID        cid.Cid
	Bytes      []byte
	Value      map[string]any
}

type CommitRecord struct {
	DID    syntax.DID
	CID    cid.Cid
	Rev    syntax.TID
	Data   cid.Cid
	Prev   *cid.Cid
	Signed *Commit
	Seq    int64
	Time   time.Time
}

type EventSink interface {
	OnCommit(ctx context.Context, ev *CommitEvent) error
}

type EventSinkFunc func(ctx context.Context, ev *CommitEvent) error

func (f EventSinkFunc) OnCommit(ctx context.Context, ev *CommitEvent) error {
	return f(ctx, ev)
}

type RecordStore interface {
	GetRecord(ctx context.Context, did syntax.DID, collection syntax.NSID, rkey syntax.RecordKey) (*Record, error)
	ListRecords(ctx context.Context, did syntax.DID, collection syntax.NSID, cursor syntax.RecordKey, limit int) ([]Record, *syntax.RecordKey, error)
}

type CommitStore interface {
	Head(ctx context.Context, did syntax.DID) (*CommitRecord, error)
	GetCommit(ctx context.Context, did syntax.DID, rev syntax.TID) (*CommitRecord, error)
	ListCommits(ctx context.Context, did syntax.DID, since *syntax.TID, limit int) ([]CommitRecord, error)
	CAR(ctx context.Context, did syntax.DID, since *syntax.TID) ([]byte, error)
	CommitCAR(ctx context.Context, did syntax.DID, commit cid.Cid) ([]byte, error)
}

type RepoStore interface {
	RecordStore
	CommitStore

	CreateRepo(ctx context.Context, did syntax.DID, key atcrypto.PrivateKey) (Repo, error)
	OpenRepo(ctx context.Context, did syntax.DID, key atcrypto.PrivateKey) (Repo, error)
	DeleteRepo(ctx context.Context, did syntax.DID) error
	RepoExists(ctx context.Context, did syntax.DID) (bool, error)
	ListRepos(ctx context.Context) ([]syntax.DID, error)
	ListEvents(ctx context.Context, since int64, limit int) ([]*CommitEvent, error)
	SetEventSink(sink EventSink)
	Close() error
}

type Repo interface {
	DID() syntax.DID
	Head(ctx context.Context) (*CommitRecord, error)
	GetRecord(ctx context.Context, collection syntax.NSID, rkey syntax.RecordKey) (*Record, error)
	ListRecords(ctx context.Context, collection syntax.NSID, cursor syntax.RecordKey, limit int) ([]Record, *syntax.RecordKey, error)
	Walk(ctx context.Context, prefix string, fn func(path string, c cid.Cid) error) error
	ApplyWrites(ctx context.Context, writes []WriteOp, swapCommit *cid.Cid) (*CommitEvent, error)
	CreateRecord(ctx context.Context, collection syntax.NSID, rkey syntax.RecordKey, record map[string]any) (*Record, *CommitEvent, error)
	PutRecord(ctx context.Context, collection syntax.NSID, rkey syntax.RecordKey, record map[string]any, swapRecord *cid.Cid) (*Record, *CommitEvent, error)
	DeleteRecord(ctx context.Context, collection syntax.NSID, rkey syntax.RecordKey, swapRecord *cid.Cid) (*CommitEvent, error)
	GetCommit(ctx context.Context, rev syntax.TID) (*CommitRecord, error)
	Commits(ctx context.Context, since *syntax.TID, limit int) ([]CommitRecord, error)
	CAR(ctx context.Context, since *syntax.TID) ([]byte, error)
	CommitCAR(ctx context.Context, commit cid.Cid) ([]byte, error)
	Proof(ctx context.Context, collection syntax.NSID, rkey syntax.RecordKey) ([]byte, error)
	Diff(ctx context.Context, since syntax.TID) ([]CommitOp, error)
	Verify(ctx context.Context) error
}