rebuild repomgr into a custom repo operator. up to 2x faster

docs/REPOMGR_MIGRATION.md

@@ -0,0 +1,148 @@
+# Incremental Migration: Vendored repomgr → Direct `indigo/repo`
+
+## Context
+
+The hold PDS uses a vendored copy of indigo's `repomgr` (~1450 lines in `pkg/hold/pds/repomgr.go`). Upstream, repomgr is [soft-deprecated](https://github.com/bluesky-social/indigo/pull/1102#issuecomment-2985956040) — bnewbold recommends using `indigo/repo` directly (as [cocoon](https://github.com/haileyok/cocoon) does). The vendored copy already has custom patches (PutRecord, UpsertRecord, prevData) and will continue to drift. This migration defines a clean interface, then swaps the implementation behind it.
+
+## Phase 0: Save plan to docs, remove dead code, define interface ✅
+
+**Goal:** Persist this migration plan as a reference doc. Shrink repomgr.go from ~1450 lines to ~750 by removing dead code. Fix import.go encapsulation. Define `RepoOperator` interface so all code accesses repomgr through it. No behavior change.
+
+**Completed 2026-02-28.** repomgr.go: 1453 → 871 lines (-40%). import.go: 189 → 88 lines (-53%). New repo_operator.go: 82 lines.
+
+### Step 1: Save plan to docs ✅
+- Write this plan to `docs/REPOMGR_MIGRATION.md`
+
+### Step 2: Dead code deleted from `repomgr.go` ✅
+- `HandleExternalUserEvent()` + `handleExternalUserEventNoArchive()` + `handleExternalUserEventArchive()`
+- `ImportNewRepo()` + `processNewRepo()` + `walkTree()` + `processOp()` + `stringOrNil()`
+- `CheckRepoSig()`
+- `TakeDownRepo()`, `ResetRepo()`, `VerifyRepo()`
+- `GetProfile()`
+- `CarStore()`
+- `NextTID()` / `nextTID()` (removed entirely — `BatchWrite` uses `rm.clk.Next()` directly)
+- `noArchive` field removed from struct
+- 12 unused imports cleaned up
+
+### Step 3: Fixed `import.go` encapsulation ✅
+Added `BulkUpsert()` method to `RepoManager`. Rewrote `ImportFromCAR` to call `p.repomgr.BulkUpsert()` instead of reaching into `p.repomgr.lockUser`, `p.repomgr.cs`, `p.repomgr.kmgr`, `p.repomgr.events`. Removed the 88-line `bulkImportRecords()` private method.
+
+### Step 4: Defined `RepoOperator` interface ✅
+
+New file: `pkg/hold/pds/repo_operator.go` — interface with 16 methods, compile-time check, shared types (`RepoEvent`, `RepoOp`, `EventKind`, `BulkRecord`).
+
+### Step 5: Updated callers to use interface ✅
+- `pkg/hold/pds/server.go` — `repomgr *RepoManager` → `repomgr RepoOperator`, `RepomgrRef()` returns `RepoOperator`
+- Downstream callers (`hold/server.go`, `admin/handlers_relays.go`, tests) unchanged — they go through `RepomgrRef()` which returns the interface
+- `var _ RepoOperator = (*RepoManager)(nil)` compile-time check in `repo_operator.go`
+
+### Verification: ✅
+- `make lint` — 0 issues (also fixed pre-existing unchecked error in `events.go`)
+- `make test` — all tests pass
+
+---
+
+## Phase 1: Test hardening against the interface ✅
+
+**Goal:** Write tests against `RepoOperator` that verify current behavior while `RepoManager` is the only implementation. These become the regression safety net when swapping to the new implementation in Phase 3.
+
+**Completed 2026-02-28.** New `repo_operator_test.go`: 38 subtests covering all CRUD, read, event emission, error paths, and edge cases. `runRepoOperatorTests(t, setup)` pattern ready for Phase 2's `DirectRepoOperator`.
+
+### Gaps covered ✅
+- `CreateRecord` — round-trip, TID 13-char rkey format, no-panic without event handler
+- `UpdateRecord` — CID changes, new data returned, non-existent record error, hydrated events
+- `PutRecord` — explicit rkey, duplicate rkey error, hydrated events
+- `UpsertRecord` — create path (created=true), update path (created=false, CID changes)
+- `DeleteRecord` — delete + verify gone, non-existent rkey error
+- `BatchWrite` — create+delete batch, update write type, auto-rkey (nil Rkey), delete-not-found error, empty write elem error, multi-op event emission with hydration, update hydration
+- `BulkUpsert` — create + re-upsert with changed data, multi-op event emission
+- `GetRecord` — CID match, CID mismatch error, not-found error
+- `GetRecordProof` — head CID + proof blocks, not-found error, no-repo error
+- `GetRepoRoot` — defined CID after init, changes after write
+- `GetRepoRev` — non-empty, changes after write
+- `ReadRepo` — non-empty CAR output, incremental export with `since`
+- `InitNewActor` — empty DID error, zero user error, event emission with hydration
+- Event emission — create/update/delete events verified: prevData, ops, oldRoot, newRoot, rev, since, repoSlice, hydration
+
+### Coverage ✅
+All RepoOperator methods 81–100%. Remaining uncovered lines are internal infrastructure error guards (`GetUserRepoRev`, `NewDeltaSession`, `OpenRepo`, `Commit`, `CloseWithRoot`) — not reachable without mocking the carstore.
+
+### Files modified ✅
+- `pkg/hold/pds/repo_operator_test.go` — new file, 38 subtests
+
+### Verification ✅
+- `make lint` — 0 issues
+- `make test` — all tests pass
+
+---
+
+## Phase 2: Build new implementation ✅
+
+**Goal:** Create `DirectRepoOperator` using `indigo/repo` directly (cocoon pattern).
+
+**Completed 2026-02-28.** New `pkg/hold/pds/repo.go`: 548 lines (vs 852 in repomgr.go, ~36% reduction). All 37 subtests pass identically for both implementations. Race detector, shuffled order, and parallel execution all clean.
+
+### New file: `pkg/hold/pds/repo.go` ✅
+
+Key differences from vendored repomgr:
+- **Single `sync.Mutex`** instead of per-user lock map (`lklk` + `userLocks` + `userLock` struct + reference counting)
+- **No OpenTelemetry tracing** (`otel.Tracer` calls removed)
+- **No `gorm` dependency** (`RepoHead` struct removed, `gorm.io/gorm` dropped from go.mod)
+- **`openWriteSession` / `commitWrite` helpers** extract the repeated 6-step write pattern
+
+Core mutation pattern (same as current, just cleaner):
+1. Lock → get rev → open delta session → open repo
+2. Capture `r.DataCid()` for prevData
+3. Perform operation(s)
+4. `r.Commit()` → `ds.CloseWithRoot()` → emit event → unlock
+
+### Shared types moved to `repo_operator.go` ✅
+- `KeyManager` interface and `ActorInfo` struct moved from `repomgr.go`
+- Both implementations import from the same location
+
+### Test wiring ✅
+- `setupTestDirectRepoOperator` — creates carstore + key manager directly (no `NewHoldPDS`)
+- `runRepoOperatorTests` refactored to accept optional `freshSetup` for `InitNewActor_EventEmission`
+- `TestDirectRepoOperator` runs all 37 subtests identically
+
+### Verification ✅
+- `go build ./cmd/hold` — compiles
+- `TestRepoManager` — 37/37 subtests pass
+- `TestDirectRepoOperator` — 37/37 subtests pass
+- `-race -shuffle=on -count=5 -parallel=8` — all clean
+- `make lint` — 0 issues
+- `make test` — all tests pass
+
+---
+
+## Phase 3: Config flag + switchover
+
+**Goal:** Feature flag to select implementation, default old.
+
+### Changes:
+- `pkg/hold/config.go` — add `UseDirectRepo bool` to DatabaseConfig
+- `pkg/hold/pds/server.go` — select implementation based on config in `NewHoldPDS`/`NewHoldPDSWithDB`
+- Regenerate example configs
+
+### Verification:
+- Deploy with `use_direct_repo: false` (default)
+- Test with `use_direct_repo: true` in staging
+- `make lint && make test`
+
+---
+
+## Phase 4: Remove vendored repomgr
+
+**Goal:** After production validation, delete the old code.
+
+- Delete `repomgr.go`
+- Remove config flag, make `DirectRepoOperator` the only implementation
+- Rename `repo_direct.go` → `repo_operator_impl.go`
+- Regenerate example configs
+- `make lint && make test`
+
+---
+
+## Decision log
+
+- **`indigo/repo` over `atproto/repo`**: `atproto/repo` has the MST primitives (`Insert`, `Remove`, `ApplyOp`) but no high-level PDS API (`OpenRepo`, `CreateRecord`, `Commit(signFn)`). Its own doc.go says "does not yet work for implementing a repository host (PDS)." `indigo/repo` is what the reference PDS and cocoon use. The `RepoOperator` interface means we can swap later if `atproto/repo` adds PDS support.

go.mod

@@ -25,7 +25,6 @@ require (
 	github.com/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674
 	github.com/ipfs/go-block-format v0.2.3
 	github.com/ipfs/go-cid v0.6.0
-	github.com/ipfs/go-datastore v0.9.1
 	github.com/ipfs/go-ipfs-blockstore v1.3.1
 	github.com/ipfs/go-ipld-cbor v0.2.1
 	github.com/ipfs/go-ipld-format v0.6.3
@@ -50,7 +49,6 @@ require (
 	golang.org/x/image v0.36.0
 	golang.org/x/sys v0.41.0
 	golang.org/x/xerrors v0.0.0-20240903120638-7835f813f4da
-	gorm.io/gorm v1.31.1
 )
 
 require (
@@ -120,6 +118,7 @@ require (
 	github.com/ipfs/bbloom v0.0.4 // indirect
 	github.com/ipfs/boxo v0.36.0 // indirect
 	github.com/ipfs/go-cidutil v0.1.1 // indirect
+	github.com/ipfs/go-datastore v0.9.1 // indirect
 	github.com/ipfs/go-dsqueue v0.2.0 // indirect
 	github.com/ipfs/go-ipfs-ds-help v1.1.1 // indirect
 	github.com/ipfs/go-ipfs-util v0.0.3 // indirect
@@ -213,5 +212,6 @@ require (
 	google.golang.org/protobuf v1.36.11 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
+	gorm.io/gorm v1.31.1 // indirect
 	lukechampine.com/blake3 v1.4.1 // indirect
 )

pkg/hold/pds/events.go

@@ -56,7 +56,7 @@ type RepoCommitEvent struct {
 	Rev      string                               `json:"rev" cborgen:"rev"`
 	Since    *string                              `json:"since,omitempty" cborgen:"since,omitempty"`
 	PrevData string                               `json:"prevData,omitempty" cborgen:"prevData,omitempty"` // MST root CID string of previous commit
-	Blocks   []byte                               `json:"blocks" cborgen:"blocks"`                        // CAR slice bytes
+	Blocks   []byte                               `json:"blocks" cborgen:"blocks"`                         // CAR slice bytes
 	Ops      []*atproto.SyncSubscribeRepos_RepoOp `json:"ops" cborgen:"ops"`
 	Time     string                               `json:"time" cborgen:"time"`
 	Type     string                               `json:"$type" cborgen:"$type"` // Always "#commit"
@@ -170,8 +170,9 @@ func (b *EventBroadcaster) initSchema() error {
 		}
 	}
 
-	// Migration: add prev_data column if missing (existing databases)
-	b.db.Exec("ALTER TABLE firehose_events ADD COLUMN prev_data TEXT")
+	// Migration: add prev_data column if missing (existing databases).
+	// Intentionally ignore error — fails with "duplicate column" if already present.
+	_, _ = b.db.Exec("ALTER TABLE firehose_events ADD COLUMN prev_data TEXT")
 
 	// Load last sequence number from database
 	var lastSeq sql.NullInt64

pkg/hold/pds/import.go

@@ -8,7 +8,6 @@ import (
 
 	"github.com/bluesky-social/indigo/repo"
 	"github.com/ipfs/go-cid"
-	"go.opentelemetry.io/otel"
 )
 
 // rawCBOR wraps raw bytes to satisfy cbg.CBORMarshaler.
@@ -20,13 +19,6 @@ func (r rawCBOR) MarshalCBOR(w io.Writer) error {
 	return err
 }
 
-// bulkRecord holds a single record to import.
-type bulkRecord struct {
-	Collection string
-	Rkey       string
-	Data       rawCBOR
-}
-
 // ImportResult summarizes a CAR import operation.
 type ImportResult struct {
 	Total         int
@@ -52,7 +44,7 @@ func (p *HoldPDS) ImportFromCAR(ctx context.Context, r io.Reader) (*ImportResult
 	}
 
 	// Collect all records
-	var records []bulkRecord
+	var records []BulkRecord
 	err = sourceRepo.ForEach(ctx, "", func(k string, v cid.Cid) error {
 		_, recBytes, err := sourceRepo.GetRecordBytes(ctx, k)
 		if err != nil {
@@ -64,7 +56,7 @@ func (p *HoldPDS) ImportFromCAR(ctx context.Context, r io.Reader) (*ImportResult
 			return fmt.Errorf("unexpected record path format: %s", k)
 		}
 
-		records = append(records, bulkRecord{
+		records = append(records, BulkRecord{
 			Collection: parts[0],
 			Rkey:       parts[1],
 			Data:       rawCBOR(*recBytes),
@@ -79,8 +71,8 @@ func (p *HoldPDS) ImportFromCAR(ctx context.Context, r io.Reader) (*ImportResult
 		return &ImportResult{PerCollection: map[string]int{}}, nil
 	}
 
-	// Bulk upsert all records in a single commit
-	if err := p.bulkImportRecords(ctx, records); err != nil {
+	// Bulk upsert all records in a single commit via the RepoOperator interface
+	if err := p.repomgr.BulkUpsert(ctx, p.uid, records); err != nil {
 		return nil, fmt.Errorf("failed to import records: %w", err)
 	}
 
@@ -94,95 +86,3 @@ func (p *HoldPDS) ImportFromCAR(ctx context.Context, r io.Reader) (*ImportResult
 	}
 	return result, nil
 }
-
-// bulkImportRecords writes all records in a single delta session + commit.
-// Each record is upserted: created if new, updated if exists.
-func (p *HoldPDS) bulkImportRecords(ctx context.Context, records []bulkRecord) error {
-	ctx, span := otel.Tracer("repoman").Start(ctx, "BulkImportRecords")
-	defer span.End()
-
-	unlock := p.repomgr.lockUser(ctx, p.uid)
-	defer unlock()
-
-	rev, err := p.repomgr.cs.GetUserRepoRev(ctx, p.uid)
-	if err != nil {
-		return err
-	}
-
-	ds, err := p.repomgr.cs.NewDeltaSession(ctx, p.uid, &rev)
-	if err != nil {
-		return err
-	}
-
-	head := ds.BaseCid()
-	r, err := repo.OpenRepo(ctx, ds, head)
-	if err != nil {
-		return err
-	}
-
-	// Capture previous MST root before commit overwrites it
-	var prevData *cid.Cid
-	if head.Defined() {
-		pd := r.DataCid()
-		prevData = &pd
-	}
-
-	ops := make([]RepoOp, 0, len(records))
-	for _, rec := range records {
-		rpath := rec.Collection + "/" + rec.Rkey
-
-		// Check if record exists to determine create vs update
-		_, _, getErr := r.GetRecordBytes(ctx, rpath)
-		recordExists := getErr == nil
-
-		var cc cid.Cid
-		var evtKind EventKind
-		if recordExists {
-			cc, err = r.UpdateRecord(ctx, rpath, rec.Data)
-			evtKind = EvtKindUpdateRecord
-		} else {
-			cc, err = r.PutRecord(ctx, rpath, rec.Data)
-			evtKind = EvtKindCreateRecord
-		}
-		if err != nil {
-			return fmt.Errorf("failed to write %s: %w", rpath, err)
-		}
-
-		ops = append(ops, RepoOp{
-			Kind:       evtKind,
-			Collection: rec.Collection,
-			Rkey:       rec.Rkey,
-			RecCid:     &cc,
-		})
-	}
-
-	nroot, nrev, err := r.Commit(ctx, p.repomgr.kmgr.SignForUser)
-	if err != nil {
-		return err
-	}
-
-	rslice, err := ds.CloseWithRoot(ctx, nroot, nrev)
-	if err != nil {
-		return fmt.Errorf("close with root: %w", err)
-	}
-
-	var oldroot *cid.Cid
-	if head.Defined() {
-		oldroot = &head
-	}
-
-	if p.repomgr.events != nil {
-		p.repomgr.events(ctx, &RepoEvent{
-			User:      p.uid,
-			OldRoot:   oldroot,
-			NewRoot:   nroot,
-			PrevData:  prevData,
-			Rev:       nrev,
-			Since:     &rev,
-			Ops:       ops,
-			RepoSlice: rslice,
-		})
-	}
-
-	return nil
-}

pkg/hold/pds/repo.go

@@ -0,0 +1,548 @@
+package pds
+
+// repo.go — DirectRepoOperator manages ATProto repository operations using
+// indigo/repo directly, replacing RepoManager with a simpler implementation.
+//
+// Key simplifications vs RepoManager:
+//   - Single sync.Mutex instead of per-user lock map (hold is always uid=1)
+//   - No OpenTelemetry tracing
+//   - No gorm dependency
+//
+// Implements the RepoOperator interface (see repo_operator.go).
+// See docs/REPOMGR_MIGRATION.md for the migration plan.
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"log/slog"
+	"sync"
+
+	holddb "atcr.io/pkg/hold/db"
+	atproto "github.com/bluesky-social/indigo/api/atproto"
+	bsky "github.com/bluesky-social/indigo/api/bsky"
+	"github.com/bluesky-social/indigo/atproto/syntax"
+	"github.com/bluesky-social/indigo/models"
+	"github.com/bluesky-social/indigo/repo"
+	"github.com/bluesky-social/indigo/util"
+
+	blocks "github.com/ipfs/go-block-format"
+	"github.com/ipfs/go-cid"
+	cbg "github.com/whyrusleeping/cbor-gen"
+)
+
+// Compile-time check that DirectRepoOperator implements RepoOperator.
+var _ RepoOperator = (*DirectRepoOperator)(nil)
+
+// DirectRepoOperator implements RepoOperator using indigo/repo directly.
+type DirectRepoOperator struct {
+	cs             holddb.CarStore
+	kmgr           KeyManager
+	mu             sync.Mutex // single mutex (hold is always uid=1)
+	events         func(context.Context, *RepoEvent)
+	hydrateRecords bool
+	log            *slog.Logger
+	clk            *syntax.TIDClock // for BatchWrite auto-rkey generation
+}
+
+// NewDirectRepoOperator creates a new DirectRepoOperator.
+func NewDirectRepoOperator(cs holddb.CarStore, kmgr KeyManager) *DirectRepoOperator {
+	return &DirectRepoOperator{
+		cs:   cs,
+		kmgr: kmgr,
+		log:  slog.Default().With("system", "repo"),
+		clk:  syntax.NewTIDClock(0),
+	}
+}
+
+// writeSession holds state for an in-progress write transaction.
+type writeSession struct {
+	ds       *holddb.DeltaSession
+	r        *repo.Repo
+	head     cid.Cid
+	prevData *cid.Cid
+	rev      string
+}
+
+// openWriteSession opens a write session, acquiring the lock.
+// On success, the caller holds d.mu and must defer d.mu.Unlock().
+// On error, the lock is released before returning.
+func (d *DirectRepoOperator) openWriteSession(ctx context.Context, user models.Uid) (*writeSession, error) {
+	d.mu.Lock()
+
+	rev, err := d.cs.GetUserRepoRev(ctx, user)
+	if err != nil {
+		d.mu.Unlock()
+		return nil, err
+	}
+
+	ds, err := d.cs.NewDeltaSession(ctx, user, &rev)
+	if err != nil {
+		d.mu.Unlock()
+		return nil, err
+	}
+
+	head := ds.BaseCid()
+
+	r, err := repo.OpenRepo(ctx, ds, head)
+	if err != nil {
+		d.mu.Unlock()
+		return nil, err
+	}
+
+	var prevData *cid.Cid
+	if head.Defined() {
+		pd := r.DataCid()
+		prevData = &pd
+	}
+
+	return &writeSession{
+		ds:       ds,
+		r:        r,
+		head:     head,
+		prevData: prevData,
+		rev:      rev,
+	}, nil
+}
+
+// commitWrite commits a write session and emits an event if configured.
+func (d *DirectRepoOperator) commitWrite(ctx context.Context, ws *writeSession, user models.Uid, ops []RepoOp) (cid.Cid, string, error) {
+	nroot, nrev, err := ws.r.Commit(ctx, d.kmgr.SignForUser)
+	if err != nil {
+		return cid.Undef, "", err
+	}
+
+	rslice, err := ws.ds.CloseWithRoot(ctx, nroot, nrev)
+	if err != nil {
+		return cid.Undef, "", fmt.Errorf("close with root: %w", err)
+	}
+
+	if d.events != nil {
+		var oldroot *cid.Cid
+		if ws.head.Defined() {
+			oldroot = &ws.head
+		}
+
+		d.events(ctx, &RepoEvent{
+			User:      user,
+			OldRoot:   oldroot,
+			NewRoot:   nroot,
+			PrevData:  ws.prevData,
+			Rev:       nrev,
+			Since:     &ws.rev,
+			Ops:       ops,
+			RepoSlice: rslice,
+		})
+	}
+
+	return nroot, nrev, nil
+}
+
+func (d *DirectRepoOperator) SetEventHandler(cb func(context.Context, *RepoEvent), hydrateRecords bool) {
+	d.events = cb
+	d.hydrateRecords = hydrateRecords
+}
+
+func (d *DirectRepoOperator) CreateRecord(ctx context.Context, user models.Uid, collection string, rec cbg.CBORMarshaler) (string, cid.Cid, error) {
+	ws, err := d.openWriteSession(ctx, user)
+	if err != nil {
+		return "", cid.Undef, err
+	}
+	defer d.mu.Unlock()
+
+	cc, tid, err := ws.r.CreateRecord(ctx, collection, rec)
+	if err != nil {
+		return "", cid.Undef, err
+	}
+
+	ops := []RepoOp{{
+		Kind:       EvtKindCreateRecord,
+		Collection: collection,
+		Rkey:       tid,
+		Record:     rec, // CreateRecord always includes Record (no hydration check)
+		RecCid:     &cc,
+	}}
+
+	if _, _, err := d.commitWrite(ctx, ws, user, ops); err != nil {
+		return "", cid.Undef, err
+	}
+
+	return collection + "/" + tid, cc, nil
+}
+
+func (d *DirectRepoOperator) UpdateRecord(ctx context.Context, user models.Uid, collection, rkey string, rec cbg.CBORMarshaler) (cid.Cid, error) {
+	ws, err := d.openWriteSession(ctx, user)
+	if err != nil {
+		return cid.Undef, err
+	}
+	defer d.mu.Unlock()
+
+	rpath := collection + "/" + rkey
+	cc, err := ws.r.UpdateRecord(ctx, rpath, rec)
+	if err != nil {
+		return cid.Undef, err
+	}
+
+	op := RepoOp{
+		Kind:       EvtKindUpdateRecord,
+		Collection: collection,
+		Rkey:       rkey,
+		RecCid:     &cc,
+	}
+	if d.hydrateRecords {
+		op.Record = rec
+	}
+
+	if _, _, err := d.commitWrite(ctx, ws, user, []RepoOp{op}); err != nil {
+		return cid.Undef, err
+	}
+
+	return cc, nil
+}
+
+func (d *DirectRepoOperator) PutRecord(ctx context.Context, user models.Uid, collection, rkey string, rec cbg.CBORMarshaler) (string, cid.Cid, error) {
+	ws, err := d.openWriteSession(ctx, user)
+	if err != nil {
+		return "", cid.Undef, err
+	}
+	defer d.mu.Unlock()
+
+	rpath := collection + "/" + rkey
+	cc, err := ws.r.PutRecord(ctx, rpath, rec)
+	if err != nil {
+		return "", cid.Undef, err
+	}
+
+	op := RepoOp{
+		Kind:       EvtKindCreateRecord,
+		Collection: collection,
+		Rkey:       rkey,
+		RecCid:     &cc,
+	}
+	if d.hydrateRecords {
+		op.Record = rec
+	}
+
+	if _, _, err := d.commitWrite(ctx, ws, user, []RepoOp{op}); err != nil {
+		return "", cid.Undef, err
+	}
+
+	return rpath, cc, nil
+}
+
+func (d *DirectRepoOperator) UpsertRecord(ctx context.Context, user models.Uid, collection, rkey string, rec cbg.CBORMarshaler) (string, cid.Cid, bool, error) {
+	ws, err := d.openWriteSession(ctx, user)
+	if err != nil {
+		return "", cid.Undef, false, err
+	}
+	defer d.mu.Unlock()
+
+	rpath := collection + "/" + rkey
+
+	// Check if record exists
+	_, _, err = ws.r.GetRecordBytes(ctx, rpath)
+	recordExists := err == nil
+
+	var cc cid.Cid
+	var evtKind EventKind
+	if recordExists {
+		cc, err = ws.r.UpdateRecord(ctx, rpath, rec)
+		evtKind = EvtKindUpdateRecord
+	} else {
+		cc, err = ws.r.PutRecord(ctx, rpath, rec)
+		evtKind = EvtKindCreateRecord
+	}
+	if err != nil {
+		return "", cid.Undef, false, err
+	}
+
+	op := RepoOp{
+		Kind:       evtKind,
+		Collection: collection,
+		Rkey:       rkey,
+		RecCid:     &cc,
+	}
+	if d.hydrateRecords {
+		op.Record = rec
+	}
+
+	if _, _, err := d.commitWrite(ctx, ws, user, []RepoOp{op}); err != nil {
+		return "", cid.Undef, false, err
+	}
+
+	return rpath, cc, !recordExists, nil
+}
+
+func (d *DirectRepoOperator) DeleteRecord(ctx context.Context, user models.Uid, collection, rkey string) error {
+	ws, err := d.openWriteSession(ctx, user)
+	if err != nil {
+		return err
+	}
+	defer d.mu.Unlock()
+
+	rpath := collection + "/" + rkey
+	if err := ws.r.DeleteRecord(ctx, rpath); err != nil {
+		return err
+	}
+
+	ops := []RepoOp{{
+		Kind:       EvtKindDeleteRecord,
+		Collection: collection,
+		Rkey:       rkey,
+	}}
+
+	_, _, err = d.commitWrite(ctx, ws, user, ops)
+	return err
+}
+
+func (d *DirectRepoOperator) BatchWrite(ctx context.Context, user models.Uid, writes []*atproto.RepoApplyWrites_Input_Writes_Elem) error {
+	ws, err := d.openWriteSession(ctx, user)
+	if err != nil {
+		return err
+	}
+	defer d.mu.Unlock()
+
+	ops := make([]RepoOp, 0, len(writes))
+	for _, w := range writes {
+		switch {
+		case w.RepoApplyWrites_Create != nil:
+			c := w.RepoApplyWrites_Create
+			var rkey string
+			if c.Rkey != nil {
+				rkey = *c.Rkey
+			} else {
+				rkey = d.clk.Next().String()
+			}
+
+			nsid := c.Collection + "/" + rkey
+			cc, err := ws.r.PutRecord(ctx, nsid, c.Value.Val)
+			if err != nil {
+				return err
+			}
+
+			op := RepoOp{
+				Kind:       EvtKindCreateRecord,
+				Collection: c.Collection,
+				Rkey:       rkey,
+				RecCid:     &cc,
+			}
+			if d.hydrateRecords {
+				op.Record = c.Value.Val
+			}
+			ops = append(ops, op)
+
+		case w.RepoApplyWrites_Update != nil:
+			u := w.RepoApplyWrites_Update
+
+			// Known quirk: uses PutRecord (mst.Add) not UpdateRecord
+			cc, err := ws.r.PutRecord(ctx, u.Collection+"/"+u.Rkey, u.Value.Val)
+			if err != nil {
+				return err
+			}
+
+			op := RepoOp{
+				Kind:       EvtKindUpdateRecord,
+				Collection: u.Collection,
+				Rkey:       u.Rkey,
+				RecCid:     &cc,
+			}
+			if d.hydrateRecords {
+				op.Record = u.Value.Val
+			}
+			ops = append(ops, op)
+
+		case w.RepoApplyWrites_Delete != nil:
+			del := w.RepoApplyWrites_Delete
+
+			if err := ws.r.DeleteRecord(ctx, del.Collection+"/"+del.Rkey); err != nil {
+				return err
+			}
+
+			ops = append(ops, RepoOp{
+				Kind:       EvtKindDeleteRecord,
+				Collection: del.Collection,
+				Rkey:       del.Rkey,
+			})
+
+		default:
+			return fmt.Errorf("no operation set in write enum")
+		}
+	}
+
+	_, _, err = d.commitWrite(ctx, ws, user, ops)
+	return err
+}
+
+func (d *DirectRepoOperator) BulkUpsert(ctx context.Context, user models.Uid, records []BulkRecord) error {
+	ws, err := d.openWriteSession(ctx, user)
+	if err != nil {
+		return err
+	}
+	defer d.mu.Unlock()
+
+	ops := make([]RepoOp, 0, len(records))
+	for _, rec := range records {
+		rpath := rec.Collection + "/" + rec.Rkey
+
+		// Check if record exists to determine create vs update
+		_, _, getErr := ws.r.GetRecordBytes(ctx, rpath)
+		recordExists := getErr == nil
+
+		var cc cid.Cid
+		var evtKind EventKind
+		if recordExists {
+			cc, err = ws.r.UpdateRecord(ctx, rpath, rec.Data)
+			evtKind = EvtKindUpdateRecord
+		} else {
+			cc, err = ws.r.PutRecord(ctx, rpath, rec.Data)
+			evtKind = EvtKindCreateRecord
+		}
+		if err != nil {
+			return fmt.Errorf("failed to write %s: %w", rpath, err)
+		}
+
+		// No hydration for BulkUpsert (records never included in events)
+		ops = append(ops, RepoOp{
+			Kind:       evtKind,
+			Collection: rec.Collection,
+			Rkey:       rec.Rkey,
+			RecCid:     &cc,
+		})
+	}
+
+	_, _, err = d.commitWrite(ctx, ws, user, ops)
+	return err
+}
+
+func (d *DirectRepoOperator) InitNewActor(ctx context.Context, user models.Uid, handle, did, displayname string, declcid, actortype string) error {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+
+	if did == "" {
+		return fmt.Errorf("must specify DID for new actor")
+	}
+
+	if user == 0 {
+		return fmt.Errorf("must specify user for new actor")
+	}
+
+	ds, err := d.cs.NewDeltaSession(ctx, user, nil)
+	if err != nil {
+		return fmt.Errorf("creating new delta session: %w", err)
+	}
+
+	r := repo.NewRepo(ctx, did, ds)
+
+	profile := &bsky.ActorProfile{
+		DisplayName: &displayname,
+	}
+
+	_, err = r.PutRecord(ctx, "app.bsky.actor.profile/self", profile)
+	if err != nil {
+		return fmt.Errorf("setting initial actor profile: %w", err)
+	}
+
+	root, nrev, err := r.Commit(ctx, d.kmgr.SignForUser)
+	if err != nil {
+		return fmt.Errorf("committing repo for actor init: %w", err)
+	}
+
+	rslice, err := ds.CloseWithRoot(ctx, root, nrev)
+	if err != nil {
+		return fmt.Errorf("close with root: %w", err)
+	}
+
+	if d.events != nil {
+		op := RepoOp{
+			Kind:       EvtKindCreateRecord,
+			Collection: "app.bsky.actor.profile",
+			Rkey:       "self",
+		}
+
+		if d.hydrateRecords {
+			op.Record = profile
+		}
+
+		d.events(ctx, &RepoEvent{
+			User:      user,
+			NewRoot:   root,
+			Rev:       nrev,
+			Ops:       []RepoOp{op},
+			RepoSlice: rslice,
+		})
+	}
+
+	return nil
+}
+
+func (d *DirectRepoOperator) GetRecord(ctx context.Context, user models.Uid, collection string, rkey string, maybeCid cid.Cid) (cid.Cid, cbg.CBORMarshaler, error) {
+	bs, err := d.cs.ReadOnlySession(user)
+	if err != nil {
+		return cid.Undef, nil, err
+	}
+
+	head, err := d.cs.GetUserRepoHead(ctx, user)
+	if err != nil {
+		return cid.Undef, nil, err
+	}
+
+	r, err := repo.OpenRepo(ctx, bs, head)
+	if err != nil {
+		return cid.Undef, nil, err
+	}
+
+	ocid, val, err := r.GetRecord(ctx, collection+"/"+rkey)
+	if err != nil {
+		return cid.Undef, nil, err
+	}
+
+	if maybeCid.Defined() && ocid != maybeCid {
+		return cid.Undef, nil, fmt.Errorf("record at specified key had different CID than expected")
+	}
+
+	return ocid, val, nil
+}
+
+func (d *DirectRepoOperator) GetRecordProof(ctx context.Context, user models.Uid, collection string, rkey string) (cid.Cid, []blocks.Block, error) {
+	robs, err := d.cs.ReadOnlySession(user)
+	if err != nil {
+		return cid.Undef, nil, err
+	}
+
+	bs := util.NewLoggingBstore(robs)
+
+	head, err := d.cs.GetUserRepoHead(ctx, user)
+	if err != nil {
+		return cid.Undef, nil, err
+	}
+
+	r, err := repo.OpenRepo(ctx, bs, head)
+	if err != nil {
+		return cid.Undef, nil, err
+	}
+
+	_, _, err = r.GetRecordBytes(ctx, collection+"/"+rkey)
+	if err != nil {
+		return cid.Undef, nil, err
+	}
+
+	return head, bs.GetLoggedBlocks(), nil
+}
+
+func (d *DirectRepoOperator) GetRepoRoot(ctx context.Context, user models.Uid) (cid.Cid, error) {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+
+	return d.cs.GetUserRepoHead(ctx, user)
+}
+
+func (d *DirectRepoOperator) GetRepoRev(ctx context.Context, user models.Uid) (string, error) {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+
+	return d.cs.GetUserRepoRev(ctx, user)
+}
+
+func (d *DirectRepoOperator) ReadRepo(ctx context.Context, user models.Uid, since string, w io.Writer) error {
+	return d.cs.ReadUserCar(ctx, user, since, true, w)
+}

pkg/hold/pds/repo_operator.go

@@ -0,0 +1,96 @@
+// Package pds implements a minimal ATProto PDS for the hold service.
+package pds
+
+import (
+	"context"
+	"io"
+
+	atproto "github.com/bluesky-social/indigo/api/atproto"
+	"github.com/bluesky-social/indigo/models"
+
+	blocks "github.com/ipfs/go-block-format"
+	"github.com/ipfs/go-cid"
+	cbg "github.com/whyrusleeping/cbor-gen"
+)
+
+// RepoOperator defines the interface for ATProto repository operations.
+// RepoManager implements this interface. Future implementations (e.g., using
+// indigo/repo directly) can be swapped in behind this interface.
+// See docs/REPOMGR_MIGRATION.md for the migration plan.
+type RepoOperator interface {
+	// Record CRUD
+	CreateRecord(ctx context.Context, user models.Uid, collection string, rec cbg.CBORMarshaler) (string, cid.Cid, error)
+	UpdateRecord(ctx context.Context, user models.Uid, collection, rkey string, rec cbg.CBORMarshaler) (cid.Cid, error)
+	PutRecord(ctx context.Context, user models.Uid, collection, rkey string, rec cbg.CBORMarshaler) (string, cid.Cid, error)
+	UpsertRecord(ctx context.Context, user models.Uid, collection, rkey string, rec cbg.CBORMarshaler) (string, cid.Cid, bool, error)
+	DeleteRecord(ctx context.Context, user models.Uid, collection, rkey string) error
+	BatchWrite(ctx context.Context, user models.Uid, writes []*atproto.RepoApplyWrites_Input_Writes_Elem) error
+	BulkUpsert(ctx context.Context, user models.Uid, records []BulkRecord) error
+
+	// Read
+	GetRecord(ctx context.Context, user models.Uid, collection string, rkey string, maybeCid cid.Cid) (cid.Cid, cbg.CBORMarshaler, error)
+	GetRecordProof(ctx context.Context, user models.Uid, collection string, rkey string) (cid.Cid, []blocks.Block, error)
+	GetRepoRoot(ctx context.Context, user models.Uid) (cid.Cid, error)
+	GetRepoRev(ctx context.Context, user models.Uid) (string, error)
+	ReadRepo(ctx context.Context, user models.Uid, since string, w io.Writer) error
+
+	// Lifecycle
+	InitNewActor(ctx context.Context, user models.Uid, handle, did, displayname string, declcid, actortype string) error
+	SetEventHandler(cb func(context.Context, *RepoEvent), hydrateRecords bool)
+}
+
+// KeyManager handles cryptographic signing for repository commits.
+type KeyManager interface {
+	VerifyUserSignature(context.Context, string, []byte, []byte) error
+	SignForUser(context.Context, string, []byte) ([]byte, error)
+}
+
+// ActorInfo holds identity information for a repository actor.
+type ActorInfo struct {
+	Did         string
+	Handle      string
+	DisplayName string
+	Type        string
+}
+
+// Compile-time check that RepoManager implements RepoOperator.
+var _ RepoOperator = (*RepoManager)(nil)
+
+// RepoEvent represents a mutation event emitted by a RepoOperator.
+type RepoEvent struct {
+	User      models.Uid
+	OldRoot   *cid.Cid
+	NewRoot   cid.Cid
+	PrevData  *cid.Cid // MST root CID of the previous commit (for firehose prevData field)
+	Since     *string
+	Rev       string
+	RepoSlice []byte
+	PDS       uint
+	Ops       []RepoOp
+}
+
+// RepoOp represents a single operation within a RepoEvent.
+type RepoOp struct {
+	Kind       EventKind
+	Collection string
+	Rkey       string
+	RecCid     *cid.Cid
+	Record     any
+	ActorInfo  *ActorInfo
+}
+
+// EventKind identifies the type of repository mutation.
+type EventKind string
+
+const (
+	EvtKindCreateRecord = EventKind("create")
+	EvtKindUpdateRecord = EventKind("update")
+	EvtKindDeleteRecord = EventKind("delete")
+)
+
+// BulkRecord holds a single record for bulk import/upsert operations.
+type BulkRecord struct {
+	Collection string
+	Rkey       string
+	Data       cbg.CBORMarshaler
+}

pkg/hold/pds/repo_operator_benchmark_test.go

@@ -0,0 +1,418 @@
+package pds
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"io"
+	"log/slog"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"atcr.io/pkg/atproto"
+	"atcr.io/pkg/auth/oauth"
+	holddb "atcr.io/pkg/hold/db"
+	"github.com/bluesky-social/indigo/models"
+	"github.com/ipfs/go-cid"
+
+	indigoatproto "github.com/bluesky-social/indigo/api/atproto"
+	lexutil "github.com/bluesky-social/indigo/lex/util"
+)
+
+// benchSetup creates a RepoOperator and returns it with the user ID.
+type benchSetup func(b *testing.B) (RepoOperator, models.Uid)
+
+func suppressLogs(b *testing.B) {
+	b.Helper()
+	prev := slog.Default()
+	slog.SetDefault(slog.New(slog.NewTextHandler(io.Discard, nil)))
+	b.Cleanup(func() { slog.SetDefault(prev) })
+}
+
+func setupBenchRepoManager(b *testing.B) (RepoOperator, models.Uid) {
+	b.Helper()
+	suppressLogs(b)
+	ctx := context.Background()
+	tmpDir := b.TempDir()
+	keyPath := filepath.Join(tmpDir, "signing-key")
+
+	if err := os.WriteFile(keyPath, sharedTestKey, 0600); err != nil {
+		b.Fatalf("write signing key: %v", err)
+	}
+
+	pds, err := NewHoldPDS(ctx, "did:web:hold.bench", "https://hold.bench", "https://atcr.io", ":memory:", keyPath, false)
+	if err != nil {
+		b.Fatalf("NewHoldPDS: %v", err)
+	}
+
+	if err := pds.repomgr.InitNewActor(ctx, pds.uid, "", pds.did, "", "", ""); err != nil {
+		b.Fatalf("InitNewActor: %v", err)
+	}
+
+	b.Cleanup(func() { pds.Close() })
+	return pds.repomgr, pds.uid
+}
+
+func setupBenchDirectRepoOperator(b *testing.B) (RepoOperator, models.Uid) {
+	b.Helper()
+	suppressLogs(b)
+	ctx := context.Background()
+	keyPath := filepath.Join(b.TempDir(), "signing-key")
+
+	if err := os.WriteFile(keyPath, sharedTestKey, 0600); err != nil {
+		b.Fatalf("write signing key: %v", err)
+	}
+	signingKey, err := oauth.GenerateOrLoadPDSKey(keyPath)
+	if err != nil {
+		b.Fatalf("GenerateOrLoadPDSKey: %v", err)
+	}
+
+	sqlStore := new(holddb.SQLiteStore)
+	if err := sqlStore.Open(":memory:"); err != nil {
+		b.Fatalf("SQLiteStore.Open: %v", err)
+	}
+	b.Cleanup(func() { sqlStore.Close() })
+
+	kmgr := NewHoldKeyManager(signingKey)
+	op := NewDirectRepoOperator(sqlStore, kmgr)
+	uid := models.Uid(1)
+
+	if err := op.InitNewActor(ctx, uid, "", "did:web:hold.bench", "", "", ""); err != nil {
+		b.Fatalf("InitNewActor: %v", err)
+	}
+
+	return op, uid
+}
+
+// seedRecords writes n records and returns their rkeys.
+func seedRecords(b *testing.B, op RepoOperator, uid models.Uid, n int) []string {
+	b.Helper()
+	ctx := context.Background()
+	rkeys := make([]string, n)
+	for i := 0; i < n; i++ {
+		rkey := fmt.Sprintf("seed%d", i)
+		_, _, err := op.PutRecord(ctx, uid, atproto.CrewCollection, rkey, newCrewRecord(fmt.Sprintf("did:plc:seed%d", i)))
+		if err != nil {
+			b.Fatalf("seed PutRecord %d: %v", i, err)
+		}
+		rkeys[i] = rkey
+	}
+	return rkeys
+}
+
+func runRepoOperatorBenchmarks(b *testing.B, name string, setup benchSetup) {
+	b.Run(name, func(b *testing.B) {
+		b.Run("CreateRecord", func(b *testing.B) {
+			op, uid := setup(b)
+			ctx := context.Background()
+
+			b.ResetTimer()
+			for i := 0; i < b.N; i++ {
+				_, _, err := op.CreateRecord(ctx, uid, atproto.CrewCollection, newCrewRecord(fmt.Sprintf("did:plc:bench%d", i)))
+				if err != nil {
+					b.Fatalf("CreateRecord: %v", err)
+				}
+			}
+		})
+
+		b.Run("PutRecord", func(b *testing.B) {
+			op, uid := setup(b)
+			ctx := context.Background()
+
+			b.ResetTimer()
+			for i := 0; i < b.N; i++ {
+				rkey := fmt.Sprintf("put%d", i)
+				_, _, err := op.PutRecord(ctx, uid, atproto.CrewCollection, rkey, newCrewRecord(fmt.Sprintf("did:plc:put%d", i)))
+				if err != nil {
+					b.Fatalf("PutRecord: %v", err)
+				}
+			}
+		})
+
+		b.Run("UpdateRecord", func(b *testing.B) {
+			op, uid := setup(b)
+			ctx := context.Background()
+
+			// Seed one record to update repeatedly
+			_, _, err := op.PutRecord(ctx, uid, atproto.CrewCollection, "updbench", newCrewRecord("did:plc:updbench"))
+			if err != nil {
+				b.Fatalf("seed PutRecord: %v", err)
+			}
+
+			b.ResetTimer()
+			for i := 0; i < b.N; i++ {
+				rec := newCrewRecord("did:plc:updbench")
+				rec.Role = fmt.Sprintf("role%d", i)
+				_, err := op.UpdateRecord(ctx, uid, atproto.CrewCollection, "updbench", rec)
+				if err != nil {
+					b.Fatalf("UpdateRecord: %v", err)
+				}
+			}
+		})
+
+		b.Run("DeleteRecord", func(b *testing.B) {
+			op, uid := setup(b)
+			ctx := context.Background()
+
+			// Pre-create all records to delete
+			rkeys := make([]string, b.N)
+			for i := 0; i < b.N; i++ {
+				rkey := fmt.Sprintf("del%d", i)
+				_, _, err := op.PutRecord(ctx, uid, atproto.CrewCollection, rkey, newCrewRecord(fmt.Sprintf("did:plc:del%d", i)))
+				if err != nil {
+					b.Fatalf("seed PutRecord: %v", err)
+				}
+				rkeys[i] = rkey
+			}
+
+			b.ResetTimer()
+			for i := 0; i < b.N; i++ {
+				if err := op.DeleteRecord(ctx, uid, atproto.CrewCollection, rkeys[i]); err != nil {
+					b.Fatalf("DeleteRecord: %v", err)
+				}
+			}
+		})
+
+		b.Run("GetRecord", func(b *testing.B) {
+			op, uid := setup(b)
+			ctx := context.Background()
+
+			_, _, err := op.PutRecord(ctx, uid, atproto.CrewCollection, "getbench", newCrewRecord("did:plc:getbench"))
+			if err != nil {
+				b.Fatalf("seed PutRecord: %v", err)
+			}
+
+			b.ResetTimer()
+			for i := 0; i < b.N; i++ {
+				_, _, err := op.GetRecord(ctx, uid, atproto.CrewCollection, "getbench", cid.Undef)
+				if err != nil {
+					b.Fatalf("GetRecord: %v", err)
+				}
+			}
+		})
+
+		b.Run("GetRepoRev", func(b *testing.B) {
+			op, uid := setup(b)
+			ctx := context.Background()
+
+			b.ResetTimer()
+			for i := 0; i < b.N; i++ {
+				_, err := op.GetRepoRev(ctx, uid)
+				if err != nil {
+					b.Fatalf("GetRepoRev: %v", err)
+				}
+			}
+		})
+
+		b.Run("GetRepoRoot", func(b *testing.B) {
+			op, uid := setup(b)
+			ctx := context.Background()
+
+			b.ResetTimer()
+			for i := 0; i < b.N; i++ {
+				_, err := op.GetRepoRoot(ctx, uid)
+				if err != nil {
+					b.Fatalf("GetRepoRoot: %v", err)
+				}
+			}
+		})
+
+		// BatchWrite at different sizes — shows commit overhead vs per-record cost
+		for _, size := range []int{1, 10, 100} {
+			b.Run(fmt.Sprintf("BatchWrite_%d", size), func(b *testing.B) {
+				op, uid := setup(b)
+				ctx := context.Background()
+
+				b.ResetTimer()
+				for i := 0; i < b.N; i++ {
+					writes := make([]*indigoatproto.RepoApplyWrites_Input_Writes_Elem, size)
+					for j := 0; j < size; j++ {
+						rkey := fmt.Sprintf("batch%d_%d", i, j)
+						writes[j] = &indigoatproto.RepoApplyWrites_Input_Writes_Elem{
+							RepoApplyWrites_Create: &indigoatproto.RepoApplyWrites_Create{
+								Collection: atproto.CrewCollection,
+								Rkey:       &rkey,
+								Value:      &lexutil.LexiconTypeDecoder{Val: newCrewRecord(fmt.Sprintf("did:plc:batch%d_%d", i, j))},
+							},
+						}
+					}
+					if err := op.BatchWrite(ctx, uid, writes); err != nil {
+						b.Fatalf("BatchWrite: %v", err)
+					}
+				}
+			})
+		}
+
+		// ReadRepo at different repo sizes
+		for _, size := range []int{10, 100, 500} {
+			b.Run(fmt.Sprintf("ReadRepo_%drecords", size), func(b *testing.B) {
+				op, uid := setup(b)
+				ctx := context.Background()
+				seedRecords(b, op, uid, size)
+
+				var buf bytes.Buffer
+				b.ResetTimer()
+				for i := 0; i < b.N; i++ {
+					buf.Reset()
+					if err := op.ReadRepo(ctx, uid, "", &buf); err != nil {
+						b.Fatalf("ReadRepo: %v", err)
+					}
+				}
+				b.SetBytes(int64(buf.Len()))
+			})
+		}
+
+		// Event overhead: with vs without handler
+		b.Run("PutRecord_NoEvents", func(b *testing.B) {
+			op, uid := setup(b)
+			ctx := context.Background()
+			// No event handler set
+
+			b.ResetTimer()
+			for i := 0; i < b.N; i++ {
+				rkey := fmt.Sprintf("noevt%d", i)
+				_, _, err := op.PutRecord(ctx, uid, atproto.CrewCollection, rkey, newCrewRecord(fmt.Sprintf("did:plc:noevt%d", i)))
+				if err != nil {
+					b.Fatalf("PutRecord: %v", err)
+				}
+			}
+		})
+
+		b.Run("PutRecord_WithEvents", func(b *testing.B) {
+			op, uid := setup(b)
+			ctx := context.Background()
+			op.SetEventHandler(func(_ context.Context, _ *RepoEvent) {}, false)
+
+			b.ResetTimer()
+			for i := 0; i < b.N; i++ {
+				rkey := fmt.Sprintf("evt%d", i)
+				_, _, err := op.PutRecord(ctx, uid, atproto.CrewCollection, rkey, newCrewRecord(fmt.Sprintf("did:plc:evt%d", i)))
+				if err != nil {
+					b.Fatalf("PutRecord: %v", err)
+				}
+			}
+		})
+
+		b.Run("PutRecord_WithHydration", func(b *testing.B) {
+			op, uid := setup(b)
+			ctx := context.Background()
+			op.SetEventHandler(func(_ context.Context, _ *RepoEvent) {}, true)
+
+			b.ResetTimer()
+			for i := 0; i < b.N; i++ {
+				rkey := fmt.Sprintf("hyd%d", i)
+				_, _, err := op.PutRecord(ctx, uid, atproto.CrewCollection, rkey, newCrewRecord(fmt.Sprintf("did:plc:hyd%d", i)))
+				if err != nil {
+					b.Fatalf("PutRecord: %v", err)
+				}
+			}
+		})
+
+		// Read from a repo with many records (MST depth)
+		b.Run("GetRecord_LargeRepo", func(b *testing.B) {
+			op, uid := setup(b)
+			ctx := context.Background()
+			rkeys := seedRecords(b, op, uid, 500)
+			target := rkeys[len(rkeys)/2] // pick a middle record
+
+			b.ResetTimer()
+			for i := 0; i < b.N; i++ {
+				_, _, err := op.GetRecord(ctx, uid, atproto.CrewCollection, target, cid.Undef)
+				if err != nil {
+					b.Fatalf("GetRecord: %v", err)
+				}
+			}
+		})
+
+		// ReadRepo incremental (since) vs full
+		b.Run("ReadRepo_Incremental", func(b *testing.B) {
+			op, uid := setup(b)
+			ctx := context.Background()
+			seedRecords(b, op, uid, 100)
+
+			// Get rev before the last write
+			rev, err := op.GetRepoRev(ctx, uid)
+			if err != nil {
+				b.Fatalf("GetRepoRev: %v", err)
+			}
+
+			// Write one more record
+			_, _, err = op.PutRecord(ctx, uid, atproto.CrewCollection, "incremental", newCrewRecord("did:plc:incremental"))
+			if err != nil {
+				b.Fatalf("PutRecord: %v", err)
+			}
+
+			var buf bytes.Buffer
+			b.ResetTimer()
+			for i := 0; i < b.N; i++ {
+				buf.Reset()
+				if err := op.ReadRepo(ctx, uid, rev, &buf); err != nil {
+					b.Fatalf("ReadRepo: %v", err)
+				}
+			}
+			b.SetBytes(int64(buf.Len()))
+		})
+	})
+}
+
+func BenchmarkRepoManager(b *testing.B) {
+	runRepoOperatorBenchmarks(b, "RepoManager", setupBenchRepoManager)
+}
+
+func BenchmarkDirectRepoOperator(b *testing.B) {
+	runRepoOperatorBenchmarks(b, "DirectRepoOperator", setupBenchDirectRepoOperator)
+}
+
+// BenchmarkBatchVsSingle compares the cost of N individual PutRecord calls
+// vs a single BatchWrite with N records.
+func BenchmarkBatchVsSingle(b *testing.B) {
+	for _, impl := range []struct {
+		name  string
+		setup benchSetup
+	}{
+		{"RepoManager", setupBenchRepoManager},
+		{"DirectRepoOperator", setupBenchDirectRepoOperator},
+	} {
+		for _, size := range []int{1, 10, 50} {
+			b.Run(fmt.Sprintf("%s/Single_%d", impl.name, size), func(b *testing.B) {
+				op, uid := impl.setup(b)
+				ctx := context.Background()
+
+				b.ResetTimer()
+				for i := 0; i < b.N; i++ {
+					for j := 0; j < size; j++ {
+						rkey := fmt.Sprintf("single%d_%d", i, j)
+						_, _, err := op.PutRecord(ctx, uid, atproto.CrewCollection, rkey, newCrewRecord(fmt.Sprintf("did:plc:s%d_%d", i, j)))
+						if err != nil {
+							b.Fatalf("PutRecord: %v", err)
+						}
+					}
+				}
+			})
+
+			b.Run(fmt.Sprintf("%s/Batch_%d", impl.name, size), func(b *testing.B) {
+				op, uid := impl.setup(b)
+				ctx := context.Background()
+
+				b.ResetTimer()
+				for i := 0; i < b.N; i++ {
+					writes := make([]*indigoatproto.RepoApplyWrites_Input_Writes_Elem, size)
+					for j := 0; j < size; j++ {
+						rkey := fmt.Sprintf("batch%d_%d", i, j)
+						writes[j] = &indigoatproto.RepoApplyWrites_Input_Writes_Elem{
+							RepoApplyWrites_Create: &indigoatproto.RepoApplyWrites_Create{
+								Collection: atproto.CrewCollection,
+								Rkey:       &rkey,
+								Value:      &lexutil.LexiconTypeDecoder{Val: newCrewRecord(fmt.Sprintf("did:plc:b%d_%d", i, j))},
+							},
+						}
+					}
+					if err := op.BatchWrite(ctx, uid, writes); err != nil {
+						b.Fatalf("BatchWrite: %v", err)
+					}
+				}
+			})
+		}
+	}
+}

pkg/hold/pds/repomgr.go

@@ -1,51 +1,34 @@
-// Package pds contains a vendored copy of RepoManager from github.com/bluesky-social/indigo
-//
-// Source: github.com/bluesky-social/indigo/repomgr (v0.0.0-20251014222321)
-// Reference: github.com/streamplace/indigo (67ae5a5) for PutRecord implementation
-// Reason: The indigo library is unmaintained and contains a critical bug in UpdateRecord
-//
-// Modifications from original:
-//   - Changed package from 'repomgr' to 'pds' for integration with hold service
-//   - Fixed UpdateRecord bug (line 263): Changed r.PutRecord to r.UpdateRecord
-//     (UpdateRecord was incorrectly calling PutRecord, causing incorrect MST operations)
-//   - Removed 5 Prometheus metrics calls (openAndSigCheckDuration, calcDiffDuration,
-//     writeCarSliceDuration, repoOpsImported) as metrics are not used in this project
-//   - Added PutRecord method (lines 309-381) for creating records with explicit rkeys
-//     (like CreateRecord but with specified rkey instead of auto-generated TID)
-//     Based on streamplace/indigo implementation
-//   - Added prevData to support Sync 1.1
 package pds
 
+// repomgr.go — RepoManager manages ATProto repository operations for the hold PDS.
+//
+// Originally vendored from github.com/bluesky-social/indigo/repomgr with fixes:
+//   - Fixed UpdateRecord bug (was calling PutRecord internally)
+//   - Added PutRecord/UpsertRecord for explicit rkey operations
+//   - Added prevData support for Sync 1.1
+//
+// Implements the RepoOperator interface (see repo_operator.go).
+// See docs/REPOMGR_MIGRATION.md for planned migration to indigo/repo directly.
+
 import (
-	"bytes"
 	"context"
-	"errors"
 	"fmt"
 	"io"
 	"log/slog"
-	"strings"
 	"sync"
 
 	holddb "atcr.io/pkg/hold/db"
 	atproto "github.com/bluesky-social/indigo/api/atproto"
 	bsky "github.com/bluesky-social/indigo/api/bsky"
 	"github.com/bluesky-social/indigo/atproto/syntax"
-	lexutil "github.com/bluesky-social/indigo/lex/util"
 	"github.com/bluesky-social/indigo/models"
-	"github.com/bluesky-social/indigo/mst"
 	"github.com/bluesky-social/indigo/repo"
 	"github.com/bluesky-social/indigo/util"
 
 	blocks "github.com/ipfs/go-block-format"
 	"github.com/ipfs/go-cid"
-	"github.com/ipfs/go-datastore"
-	blockstore "github.com/ipfs/go-ipfs-blockstore"
-	ipld "github.com/ipfs/go-ipld-format"
-	"github.com/ipld/go-car"
 	cbg "github.com/whyrusleeping/cbor-gen"
 	"go.opentelemetry.io/otel"
-	"go.opentelemetry.io/otel/attribute"
-	"gorm.io/gorm"
 )
 
 func NewRepoManager(cs holddb.CarStore, kmgr KeyManager) *RepoManager {
@@ -56,16 +39,10 @@ func NewRepoManager(cs holddb.CarStore, kmgr KeyManager) *RepoManager {
 		userLocks: make(map[models.Uid]*userLock),
 		kmgr:      kmgr,
 		log:       slog.Default().With("system", "repomgr"),
-		noArchive: false, // NonArchivalCarstore not used in hold service
 		clk:       clk,
 	}
 }
 
-type KeyManager interface {
-	VerifyUserSignature(context.Context, string, []byte, []byte) error
-	SignForUser(context.Context, string, []byte) ([]byte, error)
-}
-
 func (rm *RepoManager) SetEventHandler(cb func(context.Context, *RepoEvent), hydrateRecords bool) {
 	rm.events = cb
 	rm.hydrateRecords = hydrateRecords
@@ -81,59 +58,10 @@ type RepoManager struct {
 	events         func(context.Context, *RepoEvent)
 	hydrateRecords bool
 
-	log       *slog.Logger
-	noArchive bool
-
+	log *slog.Logger
 	clk *syntax.TIDClock
 }
 
-// NextTID generates a new TID for use as a record key.
-func (rm *RepoManager) NextTID() string {
-	return rm.clk.Next().String()
-}
-
-type ActorInfo struct {
-	Did         string
-	Handle      string
-	DisplayName string
-	Type        string
-}
-
-type RepoEvent struct {
-	User      models.Uid
-	OldRoot   *cid.Cid
-	NewRoot   cid.Cid
-	PrevData  *cid.Cid // MST root CID of the previous commit (for firehose prevData field)
-	Since     *string
-	Rev       string
-	RepoSlice []byte
-	PDS       uint
-	Ops       []RepoOp
-}
-
-type RepoOp struct {
-	Kind       EventKind
-	Collection string
-	Rkey       string
-	RecCid     *cid.Cid
-	Record     any
-	ActorInfo  *ActorInfo
-}
-
-type EventKind string
-
-const (
-	EvtKindCreateRecord = EventKind("create")
-	EvtKindUpdateRecord = EventKind("update")
-	EvtKindDeleteRecord = EventKind("delete")
-)
-
-type RepoHead struct {
-	gorm.Model
-	Usr  models.Uid `gorm:"uniqueIndex"`
-	Root string
-}
-
 type userLock struct {
 	lk    sync.Mutex
 	count int
@@ -170,10 +98,6 @@ func (rm *RepoManager) lockUser(ctx context.Context, user models.Uid) func() {
 	}
 }
 
-func (rm *RepoManager) CarStore() holddb.CarStore {
-	return rm.cs
-}
-
 func (rm *RepoManager) CreateRecord(ctx context.Context, user models.Uid, collection string, rec cbg.CBORMarshaler) (string, cid.Cid, error) {
 	ctx, span := otel.Tracer("repoman").Start(ctx, "CreateRecord")
 	defer span.End()
@@ -707,302 +631,6 @@ func (rm *RepoManager) GetRecordProof(ctx context.Context, user models.Uid, coll
 	return head, bs.GetLoggedBlocks(), nil
 }
 
-func (rm *RepoManager) GetProfile(ctx context.Context, uid models.Uid) (*bsky.ActorProfile, error) {
-	bs, err := rm.cs.ReadOnlySession(uid)
-	if err != nil {
-		return nil, err
-	}
-
-	head, err := rm.cs.GetUserRepoHead(ctx, uid)
-	if err != nil {
-		return nil, err
-	}
-
-	r, err := repo.OpenRepo(ctx, bs, head)
-	if err != nil {
-		return nil, err
-	}
-
-	_, val, err := r.GetRecord(ctx, "app.bsky.actor.profile/self")
-	if err != nil {
-		return nil, err
-	}
-
-	ap, ok := val.(*bsky.ActorProfile)
-	if !ok {
-		return nil, fmt.Errorf("found wrong type in actor profile location in tree")
-	}
-
-	return ap, nil
-}
-
-func (rm *RepoManager) CheckRepoSig(ctx context.Context, r *repo.Repo, expdid string) error {
-	ctx, span := otel.Tracer("repoman").Start(ctx, "CheckRepoSig")
-	defer span.End()
-
-	repoDid := r.RepoDid()
-	if expdid != repoDid {
-		return fmt.Errorf("DID in repo did not match (%q != %q)", expdid, repoDid)
-	}
-
-	scom := r.SignedCommit()
-
-	usc := scom.Unsigned()
-	sb, err := usc.BytesForSigning()
-	if err != nil {
-		return fmt.Errorf("commit serialization failed: %w", err)
-	}
-	if err := rm.kmgr.VerifyUserSignature(ctx, repoDid, scom.Sig, sb); err != nil {
-		return fmt.Errorf("signature check failed (sig: %x) (sb: %x) : %w", scom.Sig, sb, err)
-	}
-
-	return nil
-}
-
-func (rm *RepoManager) HandleExternalUserEvent(ctx context.Context, pdsid uint, uid models.Uid, did string, since *string, nrev string, carslice []byte, ops []*atproto.SyncSubscribeRepos_RepoOp) error {
-	if rm.noArchive {
-		return rm.handleExternalUserEventNoArchive(ctx, pdsid, uid, did, since, nrev, carslice, ops)
-	} else {
-		return rm.handleExternalUserEventArchive(ctx, pdsid, uid, did, since, nrev, carslice, ops)
-	}
-}
-
-func (rm *RepoManager) handleExternalUserEventNoArchive(ctx context.Context, pdsid uint, uid models.Uid, did string, since *string, nrev string, carslice []byte, ops []*atproto.SyncSubscribeRepos_RepoOp) error {
-	ctx, span := otel.Tracer("repoman").Start(ctx, "HandleExternalUserEvent")
-	defer span.End()
-
-	span.SetAttributes(attribute.Int64("uid", int64(uid)))
-
-	rm.log.Debug("HandleExternalUserEvent", "pds", pdsid, "uid", uid, "since", since, "nrev", nrev)
-
-	unlock := rm.lockUser(ctx, uid)
-	defer unlock()
-
-	root, ds, err := rm.cs.ImportSlice(ctx, uid, since, carslice)
-	if err != nil {
-		return fmt.Errorf("importing external carslice: %w", err)
-	}
-
-	r, err := repo.OpenRepo(ctx, ds, root)
-	if err != nil {
-		return fmt.Errorf("opening external user repo (%d, root=%s): %w", uid, root, err)
-	}
-
-	if err := rm.CheckRepoSig(ctx, r, did); err != nil {
-		return fmt.Errorf("check repo sig: %w", err)
-	}
-
-	// Capture previous MST root from old repo state if it exists
-	var prevData *cid.Cid
-	if ds.BaseCid().Defined() {
-		oldrepo, err := repo.OpenRepo(ctx, ds, ds.BaseCid())
-		if err == nil {
-			pd := oldrepo.DataCid()
-			prevData = &pd
-		}
-	}
-
-	evtops := make([]RepoOp, 0, len(ops))
-	for _, op := range ops {
-		parts := strings.SplitN(op.Path, "/", 2)
-		if len(parts) != 2 {
-			return fmt.Errorf("invalid rpath in mst diff, must have collection and rkey")
-		}
-
-		switch EventKind(op.Action) {
-		case EvtKindCreateRecord:
-			rop := RepoOp{
-				Kind:       EvtKindCreateRecord,
-				Collection: parts[0],
-				Rkey:       parts[1],
-				RecCid:     (*cid.Cid)(op.Cid),
-			}
-
-			if rm.hydrateRecords {
-				_, rec, err := r.GetRecord(ctx, op.Path)
-				if err != nil {
-					return fmt.Errorf("reading changed record from car slice: %w", err)
-				}
-				rop.Record = rec
-			}
-
-			evtops = append(evtops, rop)
-		case EvtKindUpdateRecord:
-			rop := RepoOp{
-				Kind:       EvtKindUpdateRecord,
-				Collection: parts[0],
-				Rkey:       parts[1],
-				RecCid:     (*cid.Cid)(op.Cid),
-			}
-
-			if rm.hydrateRecords {
-				_, rec, err := r.GetRecord(ctx, op.Path)
-				if err != nil {
-					return fmt.Errorf("reading changed record from car slice: %w", err)
-				}
-
-				rop.Record = rec
-			}
-
-			evtops = append(evtops, rop)
-		case EvtKindDeleteRecord:
-			evtops = append(evtops, RepoOp{
-				Kind:       EvtKindDeleteRecord,
-				Collection: parts[0],
-				Rkey:       parts[1],
-			})
-		default:
-			return fmt.Errorf("unrecognized external user event kind: %q", op.Action)
-		}
-	}
-
-	if rm.events != nil {
-		rm.events(ctx, &RepoEvent{
-			User: uid,
-			//OldRoot:   prev,
-			NewRoot:   root,
-			PrevData:  prevData,
-			Rev:       nrev,
-			Since:     since,
-			Ops:       evtops,
-			RepoSlice: carslice,
-			PDS:       pdsid,
-		})
-	}
-
-	return nil
-}
-
-func (rm *RepoManager) handleExternalUserEventArchive(ctx context.Context, pdsid uint, uid models.Uid, did string, since *string, nrev string, carslice []byte, ops []*atproto.SyncSubscribeRepos_RepoOp) error {
-	ctx, span := otel.Tracer("repoman").Start(ctx, "HandleExternalUserEvent")
-	defer span.End()
-
-	span.SetAttributes(attribute.Int64("uid", int64(uid)))
-
-	rm.log.Debug("HandleExternalUserEvent", "pds", pdsid, "uid", uid, "since", since, "nrev", nrev)
-
-	unlock := rm.lockUser(ctx, uid)
-	defer unlock()
-
-	root, ds, err := rm.cs.ImportSlice(ctx, uid, since, carslice)
-	if err != nil {
-		return fmt.Errorf("importing external carslice: %w", err)
-	}
-
-	r, err := repo.OpenRepo(ctx, ds, root)
-	if err != nil {
-		return fmt.Errorf("opening external user repo (%d, root=%s): %w", uid, root, err)
-	}
-
-	if err := rm.CheckRepoSig(ctx, r, did); err != nil {
-		return err
-	}
-
-	var skipcids map[cid.Cid]bool
-	var prevData *cid.Cid
-	if ds.BaseCid().Defined() {
-		oldrepo, err := repo.OpenRepo(ctx, ds, ds.BaseCid())
-		if err != nil {
-			return fmt.Errorf("failed to check data root in old repo: %w", err)
-		}
-
-		// Capture previous MST root for prevData
-		pd := oldrepo.DataCid()
-		prevData = &pd
-
-		// if the old commit has a 'prev', CalcDiff will error out while trying
-		// to walk it. This is an old repo thing that is being deprecated.
-		// This check is a temporary workaround until all repos get migrated
-		// and this becomes no longer an issue
-		prev, _ := oldrepo.PrevCommit(ctx)
-		if prev != nil {
-			skipcids = map[cid.Cid]bool{
-				*prev: true,
-			}
-		}
-	}
-
-	if err := ds.CalcDiff(ctx, skipcids); err != nil {
-		return fmt.Errorf("failed while calculating mst diff (since=%v): %w", since, err)
-	}
-
-	evtops := make([]RepoOp, 0, len(ops))
-
-	for _, op := range ops {
-		parts := strings.SplitN(op.Path, "/", 2)
-		if len(parts) != 2 {
-			return fmt.Errorf("invalid rpath in mst diff, must have collection and rkey")
-		}
-
-		switch EventKind(op.Action) {
-		case EvtKindCreateRecord:
-			rop := RepoOp{
-				Kind:       EvtKindCreateRecord,
-				Collection: parts[0],
-				Rkey:       parts[1],
-				RecCid:     (*cid.Cid)(op.Cid),
-			}
-
-			if rm.hydrateRecords {
-				_, rec, err := r.GetRecord(ctx, op.Path)
-				if err != nil {
-					return fmt.Errorf("reading changed record from car slice: %w", err)
-				}
-				rop.Record = rec
-			}
-
-			evtops = append(evtops, rop)
-		case EvtKindUpdateRecord:
-			rop := RepoOp{
-				Kind:       EvtKindUpdateRecord,
-				Collection: parts[0],
-				Rkey:       parts[1],
-				RecCid:     (*cid.Cid)(op.Cid),
-			}
-
-			if rm.hydrateRecords {
-				_, rec, err := r.GetRecord(ctx, op.Path)
-				if err != nil {
-					return fmt.Errorf("reading changed record from car slice: %w", err)
-				}
-
-				rop.Record = rec
-			}
-
-			evtops = append(evtops, rop)
-		case EvtKindDeleteRecord:
-			evtops = append(evtops, RepoOp{
-				Kind:       EvtKindDeleteRecord,
-				Collection: parts[0],
-				Rkey:       parts[1],
-			})
-		default:
-			return fmt.Errorf("unrecognized external user event kind: %q", op.Action)
-		}
-	}
-
-	rslice, err := ds.CloseWithRoot(ctx, root, nrev)
-	if err != nil {
-		return fmt.Errorf("close with root: %w", err)
-	}
-
-	if rm.events != nil {
-		rm.events(ctx, &RepoEvent{
-			User: uid,
-			//OldRoot:   prev,
-			NewRoot:   root,
-			PrevData:  prevData,
-			Rev:       nrev,
-			Since:     since,
-			Ops:       evtops,
-			RepoSlice: rslice,
-			PDS:       pdsid,
-		})
-	}
-
-	return nil
-}
-
 func (rm *RepoManager) BatchWrite(ctx context.Context, user models.Uid, writes []*atproto.RepoApplyWrites_Input_Writes_Elem) error {
 	ctx, span := otel.Tracer("repoman").Start(ctx, "BatchWrite")
 	defer span.End()
@@ -1131,321 +759,93 @@ func (rm *RepoManager) BatchWrite(ctx context.Context, user models.Uid, writes [
 	return nil
 }
 
-func (rm *RepoManager) ImportNewRepo(ctx context.Context, user models.Uid, repoDid string, r io.Reader, rev *string) error {
-	ctx, span := otel.Tracer("repoman").Start(ctx, "ImportNewRepo")
+// BulkUpsert writes multiple records in a single delta session and commit.
+// Each record is upserted: created if new, updated if it already exists.
+func (rm *RepoManager) BulkUpsert(ctx context.Context, user models.Uid, records []BulkRecord) error {
+	ctx, span := otel.Tracer("repoman").Start(ctx, "BulkUpsert")
 	defer span.End()
 
 	unlock := rm.lockUser(ctx, user)
 	defer unlock()
 
-	currev, err := rm.cs.GetUserRepoRev(ctx, user)
+	rev, err := rm.cs.GetUserRepoRev(ctx, user)
 	if err != nil {
 		return err
 	}
 
-	curhead, err := rm.cs.GetUserRepoHead(ctx, user)
+	ds, err := rm.cs.NewDeltaSession(ctx, user, &rev)
 	if err != nil {
 		return err
 	}
 
-	if rev != nil && *rev == "" {
-		rev = nil
-	}
-	if rev == nil {
-		// if 'rev' is nil, this implies a fresh sync.
-		// in this case, ignore any existing blocks we have and treat this like a clean import.
-		curhead = cid.Undef
+	head := ds.BaseCid()
+	r, err := repo.OpenRepo(ctx, ds, head)
+	if err != nil {
+		return err
 	}
 
-	if rev != nil && *rev != currev {
-		// TODO: we could probably just deal with this
-		return fmt.Errorf("ImportNewRepo called with incorrect base")
-	}
-
-	// Capture previous MST root before import overwrites it
+	// Capture previous MST root before commit overwrites it
 	var prevData *cid.Cid
-	if curhead.Defined() {
-		robs, err := rm.cs.ReadOnlySession(user)
-		if err == nil {
-			oldrepo, err := repo.OpenRepo(ctx, robs, curhead)
-			if err == nil {
-				pd := oldrepo.DataCid()
-				prevData = &pd
-			}
-		}
+	if head.Defined() {
+		pd := r.DataCid()
+		prevData = &pd
 	}
 
-	err = rm.processNewRepo(ctx, user, r, rev, func(ctx context.Context, root cid.Cid, finish func(context.Context, string) ([]byte, error), bs blockstore.Blockstore) error {
-		r, err := repo.OpenRepo(ctx, bs, root)
+	ops := make([]RepoOp, 0, len(records))
+	for _, rec := range records {
+		rpath := rec.Collection + "/" + rec.Rkey
+
+		// Check if record exists to determine create vs update
+		_, _, getErr := r.GetRecordBytes(ctx, rpath)
+		recordExists := getErr == nil
+
+		var cc cid.Cid
+		var evtKind EventKind
+		if recordExists {
+			cc, err = r.UpdateRecord(ctx, rpath, rec.Data)
+			evtKind = EvtKindUpdateRecord
+		} else {
+			cc, err = r.PutRecord(ctx, rpath, rec.Data)
+			evtKind = EvtKindCreateRecord
+		}
 		if err != nil {
-			return fmt.Errorf("opening new repo: %w", err)
+			return fmt.Errorf("failed to write %s: %w", rpath, err)
 		}
 
-		scom := r.SignedCommit()
+		ops = append(ops, RepoOp{
+			Kind:       evtKind,
+			Collection: rec.Collection,
+			Rkey:       rec.Rkey,
+			RecCid:     &cc,
+		})
+	}
 
-		usc := scom.Unsigned()
-		sb, err := usc.BytesForSigning()
-		if err != nil {
-			return fmt.Errorf("commit serialization failed: %w", err)
-		}
-		if err := rm.kmgr.VerifyUserSignature(ctx, repoDid, scom.Sig, sb); err != nil {
-			return fmt.Errorf("new user signature check failed: %w", err)
-		}
-
-		diffops, err := r.DiffSince(ctx, curhead)
-		if err != nil {
-			return fmt.Errorf("diff trees (curhead: %s): %w", curhead, err)
-		}
-
-		ops := make([]RepoOp, 0, len(diffops))
-		for _, op := range diffops {
-			out, err := rm.processOp(ctx, bs, op, rm.hydrateRecords)
-			if err != nil {
-				rm.log.Error("failed to process repo op", "err", err, "path", op.Rpath, "repo", repoDid)
-			}
-
-			if out != nil {
-				ops = append(ops, *out)
-			}
-		}
-
-		slice, err := finish(ctx, scom.Rev)
-		if err != nil {
-			return err
-		}
-
-		if rm.events != nil {
-			rm.events(ctx, &RepoEvent{
-				User: user,
-				//OldRoot:   oldroot,
-				NewRoot:   root,
-				PrevData:  prevData,
-				Rev:       scom.Rev,
-				Since:     &currev,
-				RepoSlice: slice,
-				Ops:       ops,
-			})
-		}
-
-		return nil
-	})
+	nroot, nrev, err := r.Commit(ctx, rm.kmgr.SignForUser)
 	if err != nil {
-		return fmt.Errorf("process new repo (current rev: %s): %w", currev, err)
-	}
-
-	return nil
-}
-
-func (rm *RepoManager) processOp(ctx context.Context, bs blockstore.Blockstore, op *mst.DiffOp, hydrateRecords bool) (*RepoOp, error) {
-	parts := strings.SplitN(op.Rpath, "/", 2)
-	if len(parts) != 2 {
-		return nil, fmt.Errorf("repo mst had invalid rpath: %q", op.Rpath)
-	}
-
-	switch op.Op {
-	case "add", "mut":
-
-		kind := EvtKindCreateRecord
-		if op.Op == "mut" {
-			kind = EvtKindUpdateRecord
-		}
-
-		outop := &RepoOp{
-			Kind:       kind,
-			Collection: parts[0],
-			Rkey:       parts[1],
-			RecCid:     &op.NewCid,
-		}
-
-		if hydrateRecords {
-			blk, err := bs.Get(ctx, op.NewCid)
-			if err != nil {
-				return nil, err
-			}
-
-			rec, err := lexutil.CborDecodeValue(blk.RawData())
-			if err != nil {
-				if !errors.Is(err, lexutil.ErrUnrecognizedType) {
-					return nil, err
-				}
-
-				rm.log.Warn("failed processing repo diff", "err", err)
-			} else {
-				outop.Record = rec
-			}
-		}
-
-		return outop, nil
-	case "del":
-		return &RepoOp{
-			Kind:       EvtKindDeleteRecord,
-			Collection: parts[0],
-			Rkey:       parts[1],
-			RecCid:     nil,
-		}, nil
-
-	default:
-		return nil, fmt.Errorf("diff returned invalid op type: %q", op.Op)
-	}
-}
-
-func (rm *RepoManager) processNewRepo(ctx context.Context, user models.Uid, r io.Reader, rev *string, cb func(ctx context.Context, root cid.Cid, finish func(context.Context, string) ([]byte, error), bs blockstore.Blockstore) error) error {
-	ctx, span := otel.Tracer("repoman").Start(ctx, "processNewRepo")
-	defer span.End()
-
-	carr, err := car.NewCarReader(r)
-	if err != nil {
 		return err
 	}
 
-	if len(carr.Header.Roots) != 1 {
-		return fmt.Errorf("invalid car file, header must have a single root (has %d)", len(carr.Header.Roots))
-	}
-
-	membs := blockstore.NewBlockstore(datastore.NewMapDatastore())
-
-	for {
-		blk, err := carr.Next()
-		if err != nil {
-			if err == io.EOF {
-				break
-			}
-			return err
-		}
-
-		if err := membs.Put(ctx, blk); err != nil {
-			return err
-		}
-	}
-
-	seen := make(map[cid.Cid]bool)
-
-	root := carr.Header.Roots[0]
-	// TODO: if there are blocks that get convergently recreated throughout
-	// the repos lifecycle, this will end up erroneously not including
-	// them. We should compute the set of blocks needed to read any repo
-	// ops that happened in the commit and use that for our 'output' blocks
-	cids, err := rm.walkTree(ctx, seen, root, membs, true)
+	rslice, err := ds.CloseWithRoot(ctx, nroot, nrev)
 	if err != nil {
-		return fmt.Errorf("walkTree: %w", err)
-	}
-
-	ds, err := rm.cs.NewDeltaSession(ctx, user, rev)
-	if err != nil {
-		return fmt.Errorf("opening delta session: %w", err)
-	}
-
-	for _, c := range cids {
-		blk, err := membs.Get(ctx, c)
-		if err != nil {
-			return fmt.Errorf("copying walked cids to carstore: %w", err)
-		}
-
-		if err := ds.Put(ctx, blk); err != nil {
-			return err
-		}
-	}
-
-	finish := func(ctx context.Context, nrev string) ([]byte, error) {
-		return ds.CloseWithRoot(ctx, root, nrev)
+		return fmt.Errorf("close with root: %w", err)
 	}
 
-	if err := cb(ctx, root, finish, ds); err != nil {
-		return fmt.Errorf("cb errored root: %s, rev: %s: %w", root, stringOrNil(rev), err)
+	var oldroot *cid.Cid
+	if head.Defined() {
+		oldroot = &head
 	}
 
-	return nil
-}
-
-func stringOrNil(s *string) string {
-	if s == nil {
-		return "nil"
-	}
-	return *s
-}
-
-// walkTree returns all cids linked recursively by the root, skipping any cids
-// in the 'skip' map, and not erroring on 'not found' if prevMissing is set
-func (rm *RepoManager) walkTree(ctx context.Context, skip map[cid.Cid]bool, root cid.Cid, bs blockstore.Blockstore, prevMissing bool) ([]cid.Cid, error) {
-	// TODO: what if someone puts non-cbor links in their repo?
-	if root.Prefix().Codec != cid.DagCBOR {
-		return nil, fmt.Errorf("can only handle dag-cbor objects in repos (%s is %d)", root, root.Prefix().Codec)
-	}
-
-	blk, err := bs.Get(ctx, root)
-	if err != nil {
-		return nil, err
-	}
-
-	var links []cid.Cid
-	if err := cbg.ScanForLinks(bytes.NewReader(blk.RawData()), func(c cid.Cid) {
-		if c.Prefix().Codec == cid.Raw {
-			rm.log.Debug("skipping 'raw' CID in record", "recordCid", root, "rawCid", c)
-			return
-		}
-		if skip[c] {
-			return
-		}
-
-		links = append(links, c)
-		skip[c] = true
-	}); err != nil {
-		return nil, err
-	}
-
-	out := []cid.Cid{root}
-	skip[root] = true
-
-	// TODO: should do this non-recursive since i expect these may get deep
-	for _, c := range links {
-		sub, err := rm.walkTree(ctx, skip, c, bs, prevMissing)
-		if err != nil {
-			if prevMissing && !ipld.IsNotFound(err) {
-				return nil, err
-			}
-		}
-
-		out = append(out, sub...)
-	}
-
-	return out, nil
-}
-
-func (rm *RepoManager) TakeDownRepo(ctx context.Context, uid models.Uid) error {
-	unlock := rm.lockUser(ctx, uid)
-	defer unlock()
-
-	return rm.cs.WipeUserData(ctx, uid)
-}
-
-// ResetRepo is technically identical to TakeDownRepo, for now
-func (rm *RepoManager) ResetRepo(ctx context.Context, uid models.Uid) error {
-	unlock := rm.lockUser(ctx, uid)
-	defer unlock()
-
-	return rm.cs.WipeUserData(ctx, uid)
-}
-
-func (rm *RepoManager) VerifyRepo(ctx context.Context, uid models.Uid) error {
-	ses, err := rm.cs.ReadOnlySession(uid)
-	if err != nil {
-		return err
-	}
-
-	r, err := repo.OpenRepo(ctx, ses, ses.BaseCid())
-	if err != nil {
-		return err
-	}
-
-	if err := r.ForEach(ctx, "", func(k string, v cid.Cid) error {
-		_, err := ses.Get(ctx, v)
-		if err != nil {
-			return fmt.Errorf("failed to get record %s (%s): %w", k, v, err)
-		}
-
-		return nil
-	}); err != nil {
-		return err
+	if rm.events != nil {
+		rm.events(ctx, &RepoEvent{
+			User:      user,
+			OldRoot:   oldroot,
+			NewRoot:   nroot,
+			PrevData:  prevData,
+			Rev:       nrev,
+			Since:     &rev,
+			Ops:       ops,
+			RepoSlice: rslice,
+		})
 	}
 
 	return nil

pkg/hold/pds/server.go

@@ -41,7 +41,7 @@ type HoldPDS struct {
 	appviewURL         string
 	appviewMeta        *atproto.AppviewMetadata
 	carstore           holddb.CarStore
-	repomgr            *RepoManager
+	repomgr            RepoOperator
 	dbPath             string
 	uid                models.Uid
 	signingKey         *atcrypto.PrivateKeyK256
@@ -106,8 +106,8 @@ func NewHoldPDS(ctx context.Context, did, publicURL, appviewURL, dbPath, keyPath
 	// Create KeyManager wrapper for our signing key
 	kmgr := NewHoldKeyManager(signingKey)
 
-	// Create RepoManager - it will handle all session/repo lifecycle
-	rm := NewRepoManager(cs, kmgr)
+	// Create repo operator - handles all session/repo lifecycle
+	rm := NewDirectRepoOperator(cs, kmgr)
 
 	// Check if repo already exists, if not create initial commit
 	head, err := cs.GetUserRepoHead(ctx, uid)
@@ -162,7 +162,7 @@ func NewHoldPDSWithDB(ctx context.Context, did, publicURL, appviewURL, dbPath, k
 	cs := sqlStore
 	uid := models.Uid(1)
 	kmgr := NewHoldKeyManager(signingKey)
-	rm := NewRepoManager(cs, kmgr)
+	rm := NewDirectRepoOperator(cs, kmgr)
 
 	head, err := cs.GetUserRepoHead(ctx, uid)
 	hasValidRepo := (err == nil && head.Defined())
@@ -200,8 +200,8 @@ func (p *HoldPDS) SigningKey() *atcrypto.PrivateKeyK256 {
 	return p.signingKey
 }
 
-// RepomgrRef returns a reference to the RepoManager for event handler setup
-func (p *HoldPDS) RepomgrRef() *RepoManager {
+// RepomgrRef returns a reference to the RepoOperator for event handler setup
+func (p *HoldPDS) RepomgrRef() RepoOperator {
 	return p.repomgr
 }