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

package pds

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"strings"
	"time"

	"atcr.io/pkg/atproto"
	"github.com/bluesky-social/indigo/mst"
	"github.com/bluesky-social/indigo/repo"
	"github.com/ipfs/go-cid"
)

// ErrCrewMemberNotFound reports that this hold's repo has no crew record for the
// member. It is deliberately distinct from a storage failure: callers that treat
// "not crew" as a benign no-op must not swallow a CAR-store error the same way,
// or a failed lookup becomes indistinguishable from a genuine non-member.
var ErrCrewMemberNotFound = errors.New("crew member not found")

// AddCrewMember adds a new crew member to the hold and commits to carstore
// Uses deterministic rkey based on member DID hash for O(1) lookups and automatic deduplication
// If the member already exists, updates their record (upsert behavior)
func (p *HoldPDS) AddCrewMember(ctx context.Context, memberDID, role string, permissions []string, tier string) (cid.Cid, error) {
	crewRecord := &atproto.CrewRecord{
		Type:        atproto.CrewCollection,
		Member:      memberDID,
		Role:        role,
		Permissions: permissions,
		Tier:        tier,
		AddedAt:     time.Now().Format(time.RFC3339),
	}

	// Use deterministic rkey based on member DID hash
	// UpsertRecord handles create-or-update automatically
	rkey := atproto.CrewRecordKey(memberDID)
	_, recordCID, _, err := p.repomgr.UpsertRecord(ctx, p.uid, atproto.CrewCollection, rkey, crewRecord)
	if err != nil {
		return cid.Undef, fmt.Errorf("failed to upsert crew record: %w", err)
	}

	return recordCID, nil
}

// GetCrewMember retrieves a crew member by their record key
func (p *HoldPDS) GetCrewMember(ctx context.Context, rkey string) (cid.Cid, *atproto.CrewRecord, error) {
	// Use repomgr.GetRecord - our types are registered in init()
	recordCID, val, err := p.repomgr.GetRecord(ctx, p.uid, atproto.CrewCollection, rkey, cid.Undef)
	if err != nil {
		if errors.Is(err, mst.ErrNotFound) {
			return cid.Undef, nil, fmt.Errorf("%w: %s", ErrCrewMemberNotFound, rkey)
		}
		return cid.Undef, nil, fmt.Errorf("failed to get crew record: %w", err)
	}

	// Type assert to our concrete type
	crewRecord, ok := val.(*atproto.CrewRecord)
	if !ok {
		return cid.Undef, nil, fmt.Errorf("unexpected type for crew record: %T", val)
	}

	return recordCID, crewRecord, nil
}

// GetCrewMemberByDID retrieves a crew member by their DID using O(1) lookup
// Uses deterministic rkey based on member DID hash
func (p *HoldPDS) GetCrewMemberByDID(ctx context.Context, memberDID string) (cid.Cid, *atproto.CrewRecord, error) {
	rkey := atproto.CrewRecordKey(memberDID)
	return p.GetCrewMember(ctx, rkey)
}

// CrewMemberWithKey pairs a crew record with its rkey and CID
type CrewMemberWithKey struct {
	Rkey   string
	Cid    cid.Cid
	Record *atproto.CrewRecord
}

// ListCrewMembers returns all crew members with their rkeys
func (p *HoldPDS) ListCrewMembers(ctx context.Context) ([]*CrewMemberWithKey, error) {
	var crew []*CrewMemberWithKey

	// Create read-only session for ForEach access
	// repomgr doesn't expose ForEach, so we need direct repo access
	session, err := p.carstore.ReadOnlySession(p.uid)
	if err != nil {
		return nil, fmt.Errorf("failed to create read-only session: %w", err)
	}

	// Get repo head
	head, err := p.carstore.GetUserRepoHead(ctx, p.uid)
	if err != nil {
		return nil, fmt.Errorf("failed to get repo head: %w", err)
	}

	if !head.Defined() {
		return nil, fmt.Errorf("repo not initialized")
	}

	// Open repo
	r, err := repo.OpenRepo(ctx, session, head)
	if err != nil {
		return nil, fmt.Errorf("failed to open repo: %w", err)
	}

	// Iterate over all crew records
	err = r.ForEach(ctx, atproto.CrewCollection, func(k string, v cid.Cid) error {
		// Extract collection and rkey from full path (k is like "io.atcr.hold.crew/3m37dr2ddit22")
		parts := strings.Split(k, "/")
		if len(parts) < 2 {
			return nil // Skip invalid keys
		}

		// Extract actual collection and rkey
		actualCollection := strings.Join(parts[:len(parts)-1], "/")
		rkey := parts[len(parts)-1]

		// MST keys are sorted, so once we hit a different collection, stop walking
		if actualCollection != atproto.CrewCollection {
			return repo.ErrDoneIterating
		}

		// Get the record directly from the repo we already have open
		// (calling GetCrewMember would open a new session unnecessarily)
		recordCID, recBytes, err := r.GetRecordBytes(ctx, k)
		if err != nil {
			return fmt.Errorf("failed to get crew record: %w", err)
		}

		// Unmarshal the CBOR bytes into our concrete type
		var crewRecord atproto.CrewRecord
		if err := crewRecord.UnmarshalCBOR(bytes.NewReader(*recBytes)); err != nil {
			return fmt.Errorf("failed to decode crew record: %w", err)
		}

		crew = append(crew, &CrewMemberWithKey{
			Rkey:   rkey,
			Cid:    recordCID,
			Record: &crewRecord,
		})
		return nil
	})

	if err != nil {
		// ErrDoneIterating is expected when we stop walking early
		// Use errors.Is to handle wrapped errors (indigo wraps with %w in MST walk)
		if errors.Is(err, repo.ErrDoneIterating) {
			// Successfully stopped at collection boundary
		} else if strings.Contains(err.Error(), "not found") {
			// If the collection doesn't exist yet (empty repo or no records created),
			// return empty list instead of error
			return []*CrewMemberWithKey{}, nil
		} else {
			return nil, fmt.Errorf("failed to list crew members: %w", err)
		}
	}

	return crew, nil
}

// RemoveCrewMember removes a crew member by rkey
func (p *HoldPDS) RemoveCrewMember(ctx context.Context, rkey string) error {
	// Use repomgr.DeleteRecord - it will automatically commit!
	// This fixes the bug where deletions weren't being committed
	err := p.repomgr.DeleteRecord(ctx, p.uid, atproto.CrewCollection, rkey)
	if err != nil {
		return fmt.Errorf("failed to delete crew record: %w", err)
	}

	return nil
}

// RemoveCrewMemberByDID removes a crew member by their DID using O(1) lookup
func (p *HoldPDS) RemoveCrewMemberByDID(ctx context.Context, memberDID string) error {
	rkey := atproto.CrewRecordKey(memberDID)
	return p.RemoveCrewMember(ctx, rkey)
}

// UpdateCrewMemberTier updates a crew member's tier
// Uses O(1) lookup via hash-based rkey and PutRecord for atomic upsert
func (p *HoldPDS) UpdateCrewMemberTier(ctx context.Context, memberDID, tier string) error {
	// O(1) lookup using hash-based rkey
	_, existing, err := p.GetCrewMemberByDID(ctx, memberDID)
	if err != nil {
		return fmt.Errorf("crew member not found: %w", err)
	}

	// If tier is already the same, no update needed
	if existing.Tier == tier {
		return nil
	}

	// Create updated record (UpdateRecord handles in-place update with same rkey)
	newRecord := &atproto.CrewRecord{
		Type:        atproto.CrewCollection,
		Member:      existing.Member,
		Role:        existing.Role,
		Permissions: existing.Permissions,
		Tier:        tier,
		Plankowner:  existing.Plankowner, // Preserve early adopter flag
		AddedAt:     existing.AddedAt,    // Preserve original add time
	}

	rkey := atproto.CrewRecordKey(memberDID)
	_, err = p.repomgr.UpdateRecord(ctx, p.uid, atproto.CrewCollection, rkey, newRecord)
	if err != nil {
		return fmt.Errorf("failed to update crew record: %w", err)
	}

	return nil
}