Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
25 kB · 780 lines
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781package atproto
import ( "bytes" "context" "errors" "fmt" "os" "path/filepath" "sort" "strings" "sync" "time"
"github.com/bluesky-social/indigo/atproto/atcrypto" "github.com/bluesky-social/indigo/carstore" "github.com/bluesky-social/indigo/models" atrepo "github.com/bluesky-social/indigo/repo" "github.com/bluesky-social/indigo/util" "github.com/ipfs/go-cid" cbor "github.com/ipfs/go-ipld-cbor" "github.com/ipld/go-car" glex "github.com/streamplace/glex/runtime" cbg "github.com/whyrusleeping/cbor-gen" "gorm.io/driver/sqlite" "gorm.io/gorm" "stream.place/streamplace/pkg/comatproto"
"stream.place/streamplace/pkg/config" "stream.place/streamplace/pkg/crypto/spkey" "stream.place/streamplace/pkg/log" "stream.place/streamplace/pkg/statedb"
gocrypto "crypto")
var ServerRepo *atrepo.Repovar ServerPubMultibase stringvar ServerCarStore carstore.CarStorevar ServerRepoUser models.Uid = models.Uid(1)
var serverRepoLock sync.Mutexvar serverCommitDB *gorm.DBvar serverRepoSigner func(ctx context.Context, did string, sb []byte) ([]byte, error)
// serverRepoPriv is the node's secp256k1 server-repo private key — the key// behind its did:web identity — captured by MakeServerRepo. Nil until then.var serverRepoPriv *atcrypto.PrivateKeyK256
// ServerCryptoSigner returns a crypto.Signer for the node's own secp256k1// identity (the server-repo key behind its did:web). Used to S2PA-sign// node-produced artifacts — e.g. a transcoded audio track minted at validate// time, signed as a c2pa.transcoded derivative of the streamer's segment.// Errors if the server repo hasn't been initialized yet.func ServerCryptoSigner() (gocrypto.Signer, error) { if serverRepoPriv == nil { return nil, fmt.Errorf("server repo key not initialized") } return spkey.KeyToSigner(serverRepoPriv)}
// serverCommitSubscribers is notified when new commit events are created.var serverCommitSubscribers []chan *ServerCommitEventvar serverCommitSubLock sync.Mutex
// SubscribeServerCommits returns a channel that receives new commit events.// Call UnsubscribeServerCommits to clean up.func SubscribeServerCommits() chan *ServerCommitEvent { serverCommitSubLock.Lock() defer serverCommitSubLock.Unlock() ch := make(chan *ServerCommitEvent, 64) serverCommitSubscribers = append(serverCommitSubscribers, ch) return ch}
// UnsubscribeServerCommits removes a subscriber channel.func UnsubscribeServerCommits(ch chan *ServerCommitEvent) { serverCommitSubLock.Lock() defer serverCommitSubLock.Unlock() for i, sub := range serverCommitSubscribers { if sub == ch { serverCommitSubscribers = append(serverCommitSubscribers[:i], serverCommitSubscribers[i+1:]...) close(ch) return } }}
func notifyServerCommitSubscribers(event *ServerCommitEvent) { serverCommitSubLock.Lock() defer serverCommitSubLock.Unlock() for _, ch := range serverCommitSubscribers { select { case ch <- event: default: // Subscriber is slow, drop the event } }}
// ServerCommitEvent stores commit events in local SQLite for relay replay (~72h TTL).type ServerCommitEvent struct { Seq int64 `gorm:"primaryKey;autoIncrement"` RepoDID string `gorm:"index:idx_server_repo_seq;column:repo_did"` Timestamp time.Time `gorm:"column:timestamp"` Data []byte `gorm:"column:data"` CID string `gorm:"column:cid"`}
type serverRepoCloser struct { carStore *carstore.SQLiteStore}
func (c *serverRepoCloser) Close() error { return c.carStore.Close()}
func MakeServerRepo(ctx context.Context, cli *config.CLI, state *statedb.StatefulDB) (Closer, error) { ctx = log.WithLogValues(ctx, "func", "MakeServerRepo")
// Ensure data directory exists csPath := cli.DataFilePath([]string{"server-repo", "carstore.db"}) if err := os.MkdirAll(filepath.Dir(csPath), 0755); err != nil { return nil, fmt.Errorf("failed to create server-repo directory: %w", err) }
// Open file-backed carstore sqliteStore := &carstore.SQLiteStore{} err := sqliteStore.Open(csPath) if err != nil { return nil, fmt.Errorf("failed to open server carstore at %s: %w", csPath, err) } ServerCarStore = sqliteStore
// Generate or load key (namespaced by server host) configKey := fmt.Sprintf("server-repo-key:%s", cli.ServerHost) var priv *atcrypto.PrivateKeyK256 keyBs, err := state.GetConfig(configKey) if err != nil { return nil, fmt.Errorf("failed to get server repo key config: %w", err) } if keyBs != nil { priv, err = atcrypto.ParsePrivateBytesK256(keyBs.Value) if err != nil { return nil, fmt.Errorf("failed to parse server repo key: %w", err) } } else { priv, err = atcrypto.GeneratePrivateKeyK256() if err != nil { return nil, fmt.Errorf("failed to generate server repo key: %w", err) } err = state.PutConfig(configKey, priv.Bytes()) if err != nil { return nil, fmt.Errorf("failed to save server repo key: %w", err) } }
serverRepoPriv = priv
pub, err := priv.PublicKey() if err != nil { return nil, fmt.Errorf("failed to get server repo public key: %w", err) } ServerPubMultibase = pub.Multibase()
serverRepoSigner = func(ctx context.Context, did string, sb []byte) ([]byte, error) { return priv.HashAndSign(sb) }
// Open or create the repo from carstore head, err := ServerCarStore.GetUserRepoHead(ctx, ServerRepoUser) if err != nil { return nil, fmt.Errorf("failed to get server repo head: %w", err) }
if head == cid.Undef { log.Log(ctx, "no existing server repo, creating new one") ses, err := ServerCarStore.NewDeltaSession(ctx, ServerRepoUser, nil) if err != nil { return nil, fmt.Errorf("failed to create delta session for new server repo: %w", err) } ServerRepo = atrepo.NewRepo(ctx, cli.ServerDID(), ses)
// Do an initial commit so the repo has a valid head root, rev, err := ServerRepo.Commit(ctx, serverRepoSigner) if err != nil { return nil, fmt.Errorf("failed to initial commit server repo: %w", err) } _, err = ses.CloseWithRoot(ctx, root, rev) if err != nil { return nil, fmt.Errorf("failed to close initial server repo session: %w", err) } log.Log(ctx, "created new server repo", "did", cli.ServerDID(), "root", root.String(), "rev", rev) } else { log.Log(ctx, "opening existing server repo", "head", head.String()) ses, err := ServerCarStore.NewDeltaSession(ctx, ServerRepoUser, nil) if err != nil { return nil, fmt.Errorf("failed to create delta session for existing server repo: %w", err) } ServerRepo, err = atrepo.OpenRepo(ctx, ses, head) if err != nil { return nil, fmt.Errorf("failed to open existing server repo: %w", err) } }
// Open local SQLite for commit events commitDBPath := cli.DataFilePath([]string{"server-repo", "commits.db"}) db, err := gorm.Open(sqlite.Open(commitDBPath), &gorm.Config{}) if err != nil { return nil, fmt.Errorf("failed to open server commit db at %s: %w", commitDBPath, err) } sqlDB, err := db.DB() if err != nil { return nil, fmt.Errorf("failed to get underlying sql db: %w", err) } sqlDB.SetMaxOpenConns(1) err = db.AutoMigrate(&ServerCommitEvent{}) if err != nil { return nil, fmt.Errorf("failed to migrate server commit events: %w", err) } serverCommitDB = db
// Prune old commit events (>72h) cutoff := time.Now().Add(-72 * time.Hour) if err := serverCommitDB.Where("timestamp < ?", cutoff).Delete(&ServerCommitEvent{}).Error; err != nil { log.Warn(ctx, "failed to prune old server commit events", "error", err) }
return &serverRepoCloser{carStore: sqliteStore}, nil}
func OpenServerRepo(ctx context.Context) (*atrepo.Repo, *carstore.DeltaSession, error) { ses, err := ServerCarStore.NewDeltaSession(ctx, ServerRepoUser, nil) if err != nil { return nil, nil, fmt.Errorf("OpenServerRepo: failed to create delta session: %w", err) }
base := ses.BaseCid() if base == cid.Undef { return nil, nil, fmt.Errorf("OpenServerRepo: delta session has no base cid") }
r, err := atrepo.OpenRepo(ctx, ses, base) if err != nil { return nil, nil, fmt.Errorf("OpenServerRepo: failed to open repo: %w", err) } return r, ses, nil}
// CreateServerCommitEvent stores a commit event in the local commit DB for relay replay.func CreateServerCommitEvent(commit *comatproto.SyncSubscribeRepos_Commit, signedData string) error { // Get previous seq var prev ServerCommitEvent err := serverCommitDB.Order("seq DESC").Limit(1).First(&prev).Error var seq int64 = 1 if err == nil { seq = prev.Seq + 1 }
commit.Seq = seq
buf := bytes.Buffer{} err = commit.MarshalCBOR(&buf) if err != nil { return fmt.Errorf("failed to marshal commit event: %w", err) } timestamp, err := time.Parse(util.ISO8601, commit.Time) if err != nil { return fmt.Errorf("failed to parse commit time: %w", err) } event := &ServerCommitEvent{ RepoDID: commit.Repo, Timestamp: timestamp.UTC(), Data: buf.Bytes(), CID: commit.Commit.String(), } if err := serverCommitDB.Create(event).Error; err != nil { return err } notifyServerCommitSubscribers(event) return nil}
// GetServerCommitEventsSinceSeq returns commit events after the given seq for relay replay.func GetServerCommitEventsSinceSeq(repoDID string, seq int64) ([]*ServerCommitEvent, error) { var events []*ServerCommitEvent err := serverCommitDB.Where("repo_did = ? AND seq > ?", repoDID, seq). Order("seq ASC").Find(&events).Error if err != nil { return nil, err } return events, nil}
// ToCommitEvent converts a ServerCommitEvent to a SyncSubscribeRepos_Commit.func (ev *ServerCommitEvent) ToCommitEvent() (*comatproto.SyncSubscribeRepos_Commit, error) { commit := &comatproto.SyncSubscribeRepos_Commit{} err := commit.UnmarshalCBOR(bytes.NewReader(ev.Data)) if err != nil { return nil, err } return commit, nil}
// CommitServerRepoRecord puts or updates a record in the server repo and stores the commit event.func CommitServerRepoRecord(ctx context.Context, cli *config.CLI, collection string, rkey string, value cbg.CBORMarshaler) error { serverRepoLock.Lock() defer serverRepoLock.Unlock()
ses, err := ServerCarStore.NewDeltaSession(ctx, ServerRepoUser, nil) if err != nil { return fmt.Errorf("CommitServerRepoRecord: failed to create delta session: %w", err) }
head, err := ServerCarStore.GetUserRepoHead(ctx, ServerRepoUser) if err != nil { return fmt.Errorf("CommitServerRepoRecord: failed to get repo head: %w", err) }
r, err := atrepo.OpenRepo(ctx, ses, head) if err != nil { return fmt.Errorf("CommitServerRepoRecord: failed to open repo: %w", err) }
rpath := fmt.Sprintf("%s/%s", collection, rkey) var recordCid cid.Cid var action string _, _, err = r.GetRecordBytes(ctx, rpath) if err != nil { // Record doesn't exist, create it recordCid, err = r.PutRecord(ctx, rpath, value) if err != nil { return fmt.Errorf("CommitServerRepoRecord: failed to put record: %w", err) } action = ActionCreate } else { // Record exists, update it recordCid, err = r.UpdateRecord(ctx, rpath, value) if err != nil { return fmt.Errorf("CommitServerRepoRecord: failed to update record: %w", err) } action = ActionUpdate }
root, rev, err := r.Commit(ctx, serverRepoSigner) if err != nil { return fmt.Errorf("CommitServerRepoRecord: failed to commit: %w", err) }
blocks, err := ses.CloseWithRoot(ctx, root, rev) if err != nil { return fmt.Errorf("CommitServerRepoRecord: failed to close session: %w", err) }
ServerRepo = r
cidLink := glex.Link(recordCid) signed := r.SignedCommit() commit := &comatproto.SyncSubscribeRepos_Commit{ Repo: cli.ServerDID(), Blocks: blocks, Rev: rev, Commit: glex.Link(root), Time: time.Now().Format(util.ISO8601), Ops: []comatproto.SyncSubscribeRepos_RepoOp{ { Action: action, Path: rpath, Cid: cidLink, }, }, TooBig: false, } err = CreateServerCommitEvent(commit, signed.Data.String()) if err != nil { return fmt.Errorf("CommitServerRepoRecord: failed to create commit event: %w", err) }
return nil}
// Server repo query functions
func ServerRepoMerkleProof(ctx context.Context, collection string, rkey string) ([]byte, error) { serverRepoLock.Lock() defer serverRepoLock.Unlock()
_, robs, err := OpenServerRepo(ctx) if err != nil { return nil, fmt.Errorf("ServerRepoMerkleProof: failed to open repo: %w", err) }
bs := util.NewLoggingBstore(robs)
root, err := ServerCarStore.GetUserRepoHead(ctx, ServerRepoUser) if err != nil { return nil, fmt.Errorf("ServerRepoMerkleProof: failed to get user repo head: %w", err) }
r, err := atrepo.OpenRepo(ctx, bs, root) if err != nil { return nil, fmt.Errorf("ServerRepoMerkleProof: failed to open repo: %w", err) }
_, _, err = r.GetRecordBytes(ctx, collection+"/"+rkey) if err != nil { return nil, fmt.Errorf("ServerRepoMerkleProof: failed to get record bytes: %w", err) }
blocks := bs.GetLoggedBlocks()
buf := new(bytes.Buffer) hb, err := cbor.DumpObject(&car.CarHeader{ Roots: []cid.Cid{root}, Version: 1, }) if err != nil { return nil, fmt.Errorf("ServerRepoMerkleProof: failed to dump car header: %w", err) } if _, err := carstore.LdWrite(buf, hb); err != nil { return nil, err }
for _, blk := range blocks { if _, err := carstore.LdWrite(buf, blk.Cid().Bytes(), blk.RawData()); err != nil { return nil, err } }
return buf.Bytes(), nil}
// ServerRepoListCollections walks the server repo's MST and returns// the distinct collection NSIDs currently holding at least one record.// Used by com.atproto.repo.describeRepo to advertise what's actually// in the repo rather than a hardcoded list. Returned collections are// sorted lexicographically for stable output.func ServerRepoListCollections(ctx context.Context) ([]string, error) { serverRepoLock.Lock() defer serverRepoLock.Unlock()
r, _, err := OpenServerRepo(ctx) if err != nil { return nil, fmt.Errorf("ServerRepoListCollections: failed to open repo: %w", err) } seen := map[string]struct{}{} err = r.ForEach(ctx, "", func(rpath string, _ cid.Cid) error { // rpath is "<collection>/<rkey>"; pull the prefix. slash := strings.IndexByte(rpath, '/') if slash <= 0 { return nil } seen[rpath[:slash]] = struct{}{} return nil }) if err != nil { return nil, fmt.Errorf("ServerRepoListCollections: error iterating records: %w", err) } out := make([]string, 0, len(seen)) for c := range seen { out = append(out, c) } sort.Strings(out) return out, nil}
// defaultListRecordsLimit / maxListRecordsLimit mirror the standard// com.atproto.repo.listRecords lexicon defaults so callers get// predictable pagination regardless of whether they hit a Streamplace// node or a stock PDS.const ( defaultListRecordsLimit = 50 maxListRecordsLimit = 100)
// ServerRepoListRecords returns records in a single collection of the// server's atproto repo, honoring the standard listRecords contract://// - filters strictly to the requested `collection`// - paginates via `cursor` (opaque to the client; we use the last// rkey of the prior page) and `limit` (clamped to [1, 100],// defaulting to 50)// - **natural order is reverse-lexical (descending rkey)**, so for// TID-shaped rkeys callers get newest-first by default. The// `reverse` param flips back to ascending for callers that want// chronological order.//// `repo` is used only to build the `at://<repo>/<collection>/<rkey>`// URIs in the response; the underlying repo is always the server's own.//// The implementation walks the MST once to collect (rkey, cid) pairs// (cheap — these entries are already in memory after a normal repo// load), sorts them in the requested direction, then fetches the// record bodies only for the page we're returning. Per-call cost// scales with the collection's record count, not with the page size,// which is fine for server-repo-sized collections (origins, view// counts) but worth revisiting if a collection ever explodes.func ServerRepoListRecords(ctx context.Context, collection string, cursor string, limit int, repo string, reverse *bool) (*comatproto.RepoListRecords_Output, error) { serverRepoLock.Lock() defer serverRepoLock.Unlock()
r, ses, err := OpenServerRepo(ctx) if err != nil { return nil, fmt.Errorf("ServerRepoListRecords: failed to open repo: %w", err) }
if limit <= 0 { limit = defaultListRecordsLimit } if limit > maxListRecordsLimit { limit = maxListRecordsLimit }
prefix := collection + "/"
// 1. Walk every rkey in the collection. No body fetches here; // those happen below only for the records we end up returning. type entry struct { rkey string c cid.Cid } var entries []entry err = r.ForEach(ctx, prefix, func(rpath string, c cid.Cid) error { // ForEach walks lex-ascending from the prefix. As soon as we // see a key that doesn't carry the prefix we're past the end // of the requested collection — stop the walk. if !strings.HasPrefix(rpath, prefix) { return atrepo.ErrDoneIterating } entries = append(entries, entry{ rkey: strings.TrimPrefix(rpath, prefix), c: c, }) return nil }) // Repo.ForEach compares the underlying mst walker's error against // atrepo.ErrDoneIterating with `==`, but the walker wraps callback // errors with `%w` before bubbling them up — so the sentinel never // matches the equality check. Use errors.Is on our side so the // early-exit path is observable through the wrapped error. if err != nil && !errors.Is(err, atrepo.ErrDoneIterating) { return nil, fmt.Errorf("ServerRepoListRecords: error iterating records: %w", err) }
// 2. Order: newest-first by default (descending rkey, which is // descending TID-time for TID-shaped rkeys). `reverse=true` // flips back to oldest-first. ascending := reverse != nil && *reverse sort.Slice(entries, func(i, j int) bool { if ascending { return entries[i].rkey < entries[j].rkey } return entries[i].rkey > entries[j].rkey })
// 3. Apply cursor. Cursor is the last rkey of the previous page; // skip past it in the active direction. start := 0 if cursor != "" { for i, e := range entries { past := false if ascending { past = e.rkey > cursor } else { past = e.rkey < cursor } if past { start = i break } start = i + 1 } } end := start + limit if end > len(entries) { end = len(entries) } page := entries[start:end]
// 4. Fetch + decode bodies only for the page. out := &comatproto.RepoListRecords_Output{ Records: make([]comatproto.RepoListRecords_Record, 0, len(page)), } for _, e := range page { raw, err := getBlock(ctx, ses, e.c) if err != nil { return nil, fmt.Errorf("ServerRepoListRecords: %w", err) } val, err := glex.CborDecodeValue(raw) if err != nil { return nil, fmt.Errorf("ServerRepoListRecords: failed to decode record for rkey %q: %w", e.rkey, err) } out.Records = append(out.Records, comatproto.RepoListRecords_Record{ Uri: fmt.Sprintf("at://%s/%s%s", repo, prefix, e.rkey), Cid: e.c.String(), Value: &glex.LexiconTypeDecoder{Val: val}, }) }
// 5. Surface a cursor whenever there are more entries past this // page. If end == len(entries) we've returned everything left. if end < len(entries) && len(page) > 0 { cur := page[len(page)-1].rkey out.Cursor = &cur }
return out, nil}
func ServerRepoGetRecord(ctx context.Context, repo string, collection string, rkey string) (*comatproto.RepoGetRecord_Output, error) { serverRepoLock.Lock() defer serverRepoLock.Unlock()
r, ses, err := OpenServerRepo(ctx) if err != nil { return nil, fmt.Errorf("ServerRepoGetRecord: failed to open repo: %w", err) } outCID, _, err := r.GetRecordBytes(ctx, fmt.Sprintf("%s/%s", collection, rkey)) if err != nil { return nil, err } raw, err := getBlock(ctx, ses, outCID) if err != nil { return nil, fmt.Errorf("ServerRepoGetRecord: %w", err) } rec, err := glex.CborDecodeValue(raw) if err != nil { return nil, fmt.Errorf("ServerRepoGetRecord: failed to decode record: %w", err) } str := outCID.String() return &comatproto.RepoGetRecord_Output{ Uri: fmt.Sprintf("at://%s/%s/%s", repo, collection, rkey), Cid: &str, Value: &glex.LexiconTypeDecoder{Val: rec}, }, nil}
func ServerRepoGetRepo(ctx context.Context, since string) ([]byte, error) { serverRepoLock.Lock() defer serverRepoLock.Unlock()
_, robs, err := OpenServerRepo(ctx) if err != nil { return nil, fmt.Errorf("ServerRepoMerkleProof: failed to open repo: %w", err) }
bs := util.NewLoggingBstore(robs)
root, err := ServerCarStore.GetUserRepoHead(ctx, ServerRepoUser) if err != nil { return nil, fmt.Errorf("ServerRepoMerkleProof: failed to get user repo head: %w", err) }
r, err := atrepo.OpenRepo(ctx, bs, root) if err != nil { return nil, fmt.Errorf("ServerRepoMerkleProof: failed to open repo: %w", err) }
err = r.ForEach(ctx, "", func(rkey string, c cid.Cid) error { _, _, err = r.GetRecordBytes(ctx, rkey) if err != nil { return fmt.Errorf("ServerRepoMerkleProof: failed to get record bytes: %w", err) } return nil }) if err != nil { return nil, fmt.Errorf("error doing foreach: %w", err) }
blocks := bs.GetLoggedBlocks()
buf := new(bytes.Buffer) hb, err := cbor.DumpObject(&car.CarHeader{ Roots: []cid.Cid{root}, Version: 1, }) if err != nil { return nil, fmt.Errorf("ServerRepoMerkleProof: failed to dump car header: %w", err) } if _, err := carstore.LdWrite(buf, hb); err != nil { return nil, err }
for _, blk := range blocks { if _, err := carstore.LdWrite(buf, blk.Cid().Bytes(), blk.RawData()); err != nil { return nil, err } }
return buf.Bytes(), nil}
// ServerRepoLatestCommit returns the server repo's current commit CID and its// revision, backing com.atproto.sync.getLatestCommit.//// Both values come out of a single delta session so they can never be torn:// the session's base CID *is* the commit the repo was opened at, and the rev is// read off that same commit. Reading the head and the rev through two separate// calls would let a concurrent CommitServerRepoRecord slip between them and// hand a peer a (cid, rev) pair that never existed -- which a verifying client// like pkg/reposync rejects outright.func ServerRepoLatestCommit(ctx context.Context) (cid.Cid, string, error) { serverRepoLock.Lock() defer serverRepoLock.Unlock()
r, ses, err := OpenServerRepo(ctx) if err != nil { return cid.Undef, "", fmt.Errorf("ServerRepoLatestCommit: %w", err) } return ses.BaseCid(), r.SignedCommit().Rev, nil}
// ServerRepoGetBlocks returns the requested blocks of the server repo as a// rootless CARv1, backing com.atproto.sync.getBlocks.func ServerRepoGetBlocks(ctx context.Context, cids []cid.Cid) ([]byte, error) { serverRepoLock.Lock() defer serverRepoLock.Unlock()
_, ses, err := OpenServerRepo(ctx) if err != nil { return nil, fmt.Errorf("ServerRepoGetBlocks: %w", err) } return getBlocksCAR(ctx, ses, cids)}
// ErrBlockNotFound reports that a block asked for by CID is not in the repo's// store. It maps to the BlockNotFound error of com.atproto.sync.getBlocks.//// Omitting the block from the response instead would be worse than useless: a// walker cannot distinguish "you don't have it" from "I mis-parsed the CAR",// and pkg/reposync treats any missing block as a hard error anyway.var ErrBlockNotFound = errors.New("BlockNotFound")
// getBlocksCAR writes cids out of ses as a CARv1 with an EMPTY roots list.//// Rootless is the shape com.atproto.sync.getBlocks is specified to return (and// what the reference PDS returns): the response is a bag of blocks, not a DAG// with an entry point. Note that go-car's NewCarReader refuses to parse such a// CAR -- consumers need raw car.ReadHeader, which is what pkg/reposync does.//// Duplicate CIDs are written once. A CID that isn't in the store fails the// whole request with ErrBlockNotFound.func getBlocksCAR(ctx context.Context, ses *carstore.DeltaSession, cids []cid.Cid) ([]byte, error) { buf := new(bytes.Buffer) hb, err := cbor.DumpObject(&car.CarHeader{ Roots: []cid.Cid{}, Version: 1, }) if err != nil { return nil, fmt.Errorf("getBlocksCAR: failed to dump car header: %w", err) } if _, err := carstore.LdWrite(buf, hb); err != nil { return nil, err }
seen := make(map[cid.Cid]struct{}, len(cids)) for _, c := range cids { if _, ok := seen[c]; ok { continue } seen[c] = struct{}{} raw, err := getBlock(ctx, ses, c) if err != nil { return nil, fmt.Errorf("%w: %s", ErrBlockNotFound, c.String()) } if _, err := carstore.LdWrite(buf, c.Bytes(), raw); err != nil { return nil, err } }
return buf.Bytes(), nil}
func getBlock(ctx context.Context, ses *carstore.DeltaSession, c cid.Cid) ([]byte, error) { b, err := ses.Get(ctx, c) if err != nil { return nil, fmt.Errorf("failed to get block %s: %w", c.String(), err) } return b.RawData(), nil}