at-container-registry/pkg/hold/pds/import.go

package pds

import (
	"context"
	"fmt"
	"io"
	"strings"

	"github.com/bluesky-social/indigo/repo"
	"github.com/ipfs/go-cid"
	"go.opentelemetry.io/otel"
)

// rawCBOR wraps raw bytes to satisfy cbg.CBORMarshaler.
// Used to pass through record bytes from a CAR without decoding.
type rawCBOR []byte

func (r rawCBOR) MarshalCBOR(w io.Writer) error {
	_, err := w.Write(r)
	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
	PerCollection map[string]int
}

// ImportFromCAR reads a CAR file and imports all records into the hold's repo.
// Records are upserted (overwrite on conflict) in a single atomic commit.
// The repo is initialized if it doesn't exist yet.
func (p *HoldPDS) ImportFromCAR(ctx context.Context, r io.Reader) (*ImportResult, error) {
	// Ensure repo exists
	head, err := p.carstore.GetUserRepoHead(ctx, p.uid)
	if err != nil || !head.Defined() {
		if err := p.repomgr.InitNewActor(ctx, p.uid, "", p.did, "", "", ""); err != nil {
			return nil, fmt.Errorf("failed to initialize repo: %w", err)
		}
	}

	// Parse the CAR into an in-memory repo
	sourceRepo, err := repo.ReadRepoFromCar(ctx, r)
	if err != nil {
		return nil, fmt.Errorf("failed to read CAR: %w", err)
	}

	// Collect all records
	var records []bulkRecord
	err = sourceRepo.ForEach(ctx, "", func(k string, v cid.Cid) error {
		_, recBytes, err := sourceRepo.GetRecordBytes(ctx, k)
		if err != nil {
			return fmt.Errorf("failed to get record bytes for %s: %w", k, err)
		}

		parts := strings.SplitN(k, "/", 2)
		if len(parts) != 2 {
			return fmt.Errorf("unexpected record path format: %s", k)
		}

		records = append(records, bulkRecord{
			Collection: parts[0],
			Rkey:       parts[1],
			Data:       rawCBOR(*recBytes),
		})
		return nil
	})
	if err != nil {
		return nil, fmt.Errorf("failed to iterate CAR records: %w", err)
	}

	if len(records) == 0 {
		return &ImportResult{PerCollection: map[string]int{}}, nil
	}

	// Bulk upsert all records in a single commit
	if err := p.bulkImportRecords(ctx, records); err != nil {
		return nil, fmt.Errorf("failed to import records: %w", err)
	}

	// Build result
	result := &ImportResult{
		Total:         len(records),
		PerCollection: make(map[string]int),
	}
	for _, rec := range records {
		result.PerCollection[rec.Collection]++
	}
	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
	}

	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,
			Rev:       nrev,
			Since:     &rev,
			Ops:       ops,
			RepoSlice: rslice,
		})
	}

	return nil
}