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

package pds

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"

	"atcr.io/pkg/atproto"
	"atcr.io/pkg/s3"
	"github.com/bluesky-social/indigo/api/bsky"
	lexutil "github.com/bluesky-social/indigo/lex/util"
	"github.com/bluesky-social/indigo/repo"
	"github.com/distribution/distribution/v3/registry/storage/driver"
	"github.com/go-chi/chi/v5"
	"github.com/gorilla/websocket"
	"github.com/ipfs/go-cid"
	"github.com/ipld/go-car"
	carutil "github.com/ipld/go-car/util"

	"crypto/sha256"
	"io"
	"log/slog"
	"net/http"
	"strconv"
	"strings"
	"time"

	"github.com/multiformats/go-multihash"

	awss3 "github.com/aws/aws-sdk-go/service/s3"
)

// XRPC handler for ATProto endpoints

// Context keys for storing user info in request context
type contextKey string

const (
	contextKeyUser contextKey = "user"
)

// XRPCHandler handles XRPC requests for the embedded PDS
type XRPCHandler struct {
	pds           *HoldPDS
	s3Service     s3.S3Service
	storageDriver driver.StorageDriver
	broadcaster   *EventBroadcaster
	httpClient    HTTPClient // For testing - allows injecting mock HTTP client
}

// PartInfo represents a completed part in a multipart upload
type PartInfo struct {
	PartNumber int    `json:"partNumber"`
	ETag       string `json:"etag"`
}

// PartUploadInfo contains structured information for uploading a part
// Used for both S3 presigned URLs and buffered mode with headers
type PartUploadInfo struct {
	URL     string            `json:"url"`               // URL to PUT the part to
	Method  string            `json:"method,omitempty"`  // HTTP method (usually "PUT")
	Headers map[string]string `json:"headers,omitempty"` // Additional headers required for the request
}

// NewXRPCHandler creates a new XRPC handler
func NewXRPCHandler(pds *HoldPDS, s3Service s3.S3Service, storageDriver driver.StorageDriver, broadcaster *EventBroadcaster, httpClient HTTPClient) *XRPCHandler {
	return &XRPCHandler{
		pds:           pds,
		s3Service:     s3Service,
		storageDriver: storageDriver,
		broadcaster:   broadcaster,
		httpClient:    httpClient,
	}
}

// CORSMiddleware returns a simple CORS middleware configured for ATProto
// This should be applied in the main router before registering any routes
func (h *XRPCHandler) CORSMiddleware() func(http.Handler) http.Handler {
	return func(next http.Handler) http.Handler {
		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			// Set CORS headers for all requests
			origin := r.Header.Get("Origin")
			if origin == "" {
				origin = "*"
			}
			w.Header().Set("Access-Control-Allow-Origin", origin)
			w.Header().Set("Access-Control-Allow-Methods", "GET, HEAD, POST, PUT, DELETE, OPTIONS")
			w.Header().Set("Access-Control-Allow-Headers", "*")
			w.Header().Set("Access-Control-Expose-Headers", "*")
			w.Header().Set("Access-Control-Max-Age", "300")

			// Handle OPTIONS preflight
			if r.Method == "OPTIONS" {
				w.WriteHeader(http.StatusOK)
				return
			}

			next.ServeHTTP(w, r)
		})
	}
}

// requireOwnerOrCrewAdmin middleware - validates owner or crew admin access
// Stores validated user in request context
func (h *XRPCHandler) requireOwnerOrCrewAdmin(next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		user, err := ValidateOwnerOrCrewAdmin(r, h.pds, h.httpClient)
		if err != nil {
			http.Error(w, fmt.Sprintf("unauthorized: %v", err), http.StatusForbidden)
			return
		}
		// Store user in context for handlers to access
		ctx := context.WithValue(r.Context(), contextKeyUser, user)
		next.ServeHTTP(w, r.WithContext(ctx))
	})
}

// requireAuth middleware - validates service token authentication
// Stores validated user in request context
func (h *XRPCHandler) requireAuth(next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		// Service token authentication
		user, err := ValidateServiceToken(r, h.pds.did, h.httpClient)
		if err != nil {
			http.Error(w, fmt.Sprintf("unauthorized: %v", err), http.StatusUnauthorized)
			return
		}
		// Store user in context for handlers to access
		ctx := context.WithValue(r.Context(), contextKeyUser, user)
		next.ServeHTTP(w, r.WithContext(ctx))
	})
}

// getUserFromContext extracts the authenticated user from request context
// Returns nil if no user is in context (handler should be protected by auth middleware)
func getUserFromContext(r *http.Request) *ValidatedUser {
	user, ok := r.Context().Value(contextKeyUser).(*ValidatedUser)
	if !ok {
		return nil
	}
	return user
}

// RegisterHandlers registers all XRPC endpoints using chi router
// Note: CORS middleware must be applied in the main router before calling this
func (h *XRPCHandler) RegisterHandlers(r chi.Router) {
	// Public read-only endpoints (no auth)
	r.Group(func(r chi.Router) {
		// Health and server info
		r.Get("/xrpc/_health", h.HandleHealth)
		r.Get(atproto.ServerDescribeServer, h.HandleDescribeServer)

		// Repository metadata
		r.Get(atproto.RepoDescribeRepo, h.HandleDescribeRepo)
		r.Get(atproto.RepoGetRecord, h.HandleGetRecord)
		r.Get(atproto.RepoListRecords, h.HandleListRecords)

		// Sync endpoints
		r.Get(atproto.SyncListRepos, h.HandleListRepos)
		r.Get(atproto.SyncGetRecord, h.HandleSyncGetRecord)
		r.Get(atproto.SyncGetRepo, h.HandleGetRepo)
		r.Get(atproto.SyncGetRepoStatus, h.HandleGetRepoStatus)
		r.Get(atproto.SyncSubscribeRepos, h.HandleSubscribeRepos)

		// DID document and handle resolution
		r.Get("/.well-known/did.json", h.HandleDIDDocument)
		r.Get("/.well-known/atproto-did", h.HandleAtprotoDID)

		// Identity and profile endpoints
		r.Get(atproto.IdentityResolveHandle, h.HandleResolveHandle)
		r.Get(atproto.ActorGetProfile, h.HandleGetProfile)
		r.Get(atproto.ActorGetProfiles, h.HandleGetProfiles)
	})

	// Blob read endpoints (conditional auth based on captain.public)
	// Auth is handled inside HandleGetBlob
	r.Group(func(r chi.Router) {
		r.Get(atproto.SyncGetBlob, h.HandleGetBlob)
		r.Head(atproto.SyncGetBlob, h.HandleGetBlob)
	})

	// Write endpoints (owner/crew admin auth)
	r.Group(func(r chi.Router) {
		r.Use(h.requireOwnerOrCrewAdmin)
		r.Post(atproto.RepoDeleteRecord, h.HandleDeleteRecord)
		r.Post(atproto.RepoUploadBlob, h.HandleUploadBlob)
	})

	// Auth-only endpoints (DPoP auth)
	r.Group(func(r chi.Router) {
		r.Use(h.requireAuth)
		r.Post(atproto.HoldRequestCrew, h.HandleRequestCrew)
	})
}

// HandleHealth returns health check information
func (h *XRPCHandler) HandleHealth(w http.ResponseWriter, r *http.Request) {
	response := map[string]any{
		"version": "0.4.999",
	}

	w.Header().Set("Content-Type", "application/json")
	json.NewEncoder(w).Encode(response)
}

// HandleDescribeServer returns server metadata
func (h *XRPCHandler) HandleDescribeServer(w http.ResponseWriter, r *http.Request) {
	// Extract hostname from public URL for availableUserDomains
	// For hold01.atcr.io, return [".hold01.atcr.io"] to match stream.place pattern
	hostname := h.pds.PublicURL
	hostname = strings.TrimPrefix(hostname, "http://")
	hostname = strings.TrimPrefix(hostname, "https://")
	hostname = strings.Split(hostname, "/")[0] // Remove path
	hostname = strings.Split(hostname, ":")[0] // Remove port

	response := map[string]any{
		"did":                  h.pds.DID(),
		"availableUserDomains": []string{"." + hostname},
		"inviteCodeRequired":   true, // Single-user PDS, no account creation
	}

	w.Header().Set("Content-Type", "application/json")
	json.NewEncoder(w).Encode(response)
}

// HandleResolveHandle resolves a handle to a DID
func (h *XRPCHandler) HandleResolveHandle(w http.ResponseWriter, r *http.Request) {
	// Get handle parameter
	handle := r.URL.Query().Get("handle")
	if handle == "" {
		http.Error(w, "handle parameter required", http.StatusBadRequest)
		return
	}

	// For this hold PDS, the handle is the domain part of the DID
	// e.g., "hold01.atcr.io did:web:hold01.atcr.io"
	expectedHandle := strings.TrimPrefix(h.pds.DID(), "did:web:")

	// Check if the handle matches
	if handle != expectedHandle {
		http.Error(w, "handle not found", http.StatusNotFound)
		return
	}

	// Return the DID
	response := map[string]string{
		"did": h.pds.DID(),
	}

	w.Header().Set("Content-Type", "application/json")
	json.NewEncoder(w).Encode(response)
}

// HandleGetProfile returns aggregated profile information
func (h *XRPCHandler) HandleGetProfile(w http.ResponseWriter, r *http.Request) {
	// Get actor parameter (can be DID or handle)
	actor := r.URL.Query().Get("actor")
	if actor == "" {
		http.Error(w, "actor parameter required", http.StatusBadRequest)
		return
	}

	// Normalize actor to DID
	actorDID := actor
	if !atproto.IsDID(actor) {
		// It's a handle, resolve to DID
		expectedHandle := strings.TrimPrefix(h.pds.DID(), "did:web:")
		if actor == expectedHandle {
			actorDID = h.pds.DID()
		} else {
			http.Error(w, "actor not found", http.StatusNotFound)
			return
		}
	}

	// Verify it's this hold's DID
	if actorDID != h.pds.DID() {
		http.Error(w, "actor not found", http.StatusNotFound)
		return
	}

	// Build profile response using shared function
	response := h.buildProfileResponse(r.Context())

	w.Header().Set("Content-Type", "application/json")
	json.NewEncoder(w).Encode(response)
}

// HandleGetProfiles returns aggregated profile information for multiple actors
func (h *XRPCHandler) HandleGetProfiles(w http.ResponseWriter, r *http.Request) {
	// Get actors parameters (can be multiple)
	actors := r.URL.Query()["actors"]
	if len(actors) == 0 {
		http.Error(w, "actors parameter required", http.StatusBadRequest)
		return
	}

	// Initialize profiles array (always return array, even if empty)
	profiles := []map[string]any{}

	// Expected handle for this hold
	expectedHandle := strings.TrimPrefix(h.pds.DID(), "did:web:")

	// Check each actor to see if it matches this hold's DID
	for _, actor := range actors {
		// Normalize actor to DID
		actorDID := actor
		if !atproto.IsDID(actor) {
			// It's a handle, check if it matches
			if actor == expectedHandle {
				actorDID = h.pds.DID()
			} else {
				// Not this hold's handle, skip
				continue
			}
		}

		// Check if it's this hold's DID
		if actorDID != h.pds.DID() {
			// Not this hold, skip
			continue
		}

		// Build profile for this hold
		profile := h.buildProfileResponse(r.Context())
		if profile != nil {
			profiles = append(profiles, profile)
		}
	}

	// Return profiles array
	response := map[string]any{
		"profiles": profiles,
	}

	w.Header().Set("Content-Type", "application/json")
	json.NewEncoder(w).Encode(response)
}

// buildProfileResponse builds a profile response map (shared by GetProfile and GetProfiles)
func (h *XRPCHandler) buildProfileResponse(ctx context.Context) map[string]any {
	// Get profile record from repo
	_, profileVal, _ := h.pds.repomgr.GetRecord(
		ctx,
		h.pds.uid,
		"app.bsky.actor.profile",
		"self",
		cid.Undef,
	)

	// Base response with minimal info
	response := map[string]any{
		"did":            h.pds.DID(),
		"handle":         strings.TrimPrefix(h.pds.DID(), "did:web:"),
		"postsCount":     0,
		"followersCount": 0,
		"followsCount":   0,
	}

	// Count posts
	session, err := h.pds.carstore.ReadOnlySession(h.pds.uid)
	if err == nil {
		head, err := h.pds.carstore.GetUserRepoHead(ctx, h.pds.uid)
		if err == nil && head.Defined() {
			repoHandle, err := repo.OpenRepo(ctx, session, head)
			if err == nil {
				postCount := 0
				_ = repoHandle.ForEach(ctx, "app.bsky.feed.post", func(k string, v cid.Cid) error {
					postCount++
					return nil
				})
				response["postsCount"] = postCount
			}
		}
	}

	// Add profile fields if profile record exists
	if err == nil {
		profileRecord, ok := profileVal.(*bsky.ActorProfile)
		if ok {
			if profileRecord.DisplayName != nil && *profileRecord.DisplayName != "" {
				response["displayName"] = *profileRecord.DisplayName
			}
			if profileRecord.Description != nil && *profileRecord.Description != "" {
				response["description"] = *profileRecord.Description
			}
			if profileRecord.Avatar != nil {
				avatarURL := fmt.Sprintf("%s/xrpc/com.atproto.sync.getBlob?did=%s&cid=%s",
					h.pds.PublicURL, h.pds.DID(), profileRecord.Avatar.Ref.String())
				response["avatar"] = avatarURL
			}
		}
	}

	return response
}

// HandleDescribeRepo returns repository information
func (h *XRPCHandler) HandleDescribeRepo(w http.ResponseWriter, r *http.Request) {
	// Get repo parameter
	repoDID := r.URL.Query().Get("repo")
	if repoDID == "" || repoDID != h.pds.DID() {
		http.Error(w, "invalid repo", http.StatusBadRequest)
		return
	}

	// Generate DID document
	didDoc, err := h.pds.GenerateDIDDocument(h.pds.PublicURL)
	if err != nil {
		http.Error(w, fmt.Sprintf("failed to generate DID document: %v", err), http.StatusInternalServerError)
		return
	}

	// Get actual collections from repo
	collections, err := h.pds.ListCollections(r.Context())
	if err != nil {
		http.Error(w, fmt.Sprintf("failed to list collections: %v", err), http.StatusInternalServerError)
		return
	}

	// Note: For did:web, the handle IS the DID (not just hostname)
	response := map[string]any{
		"did":             h.pds.DID(),
		"handle":          h.pds.DID(),
		"didDoc":          didDoc,
		"collections":     collections,
		"handleIsCorrect": true,
	}

	w.Header().Set("Content-Type", "application/json")
	json.NewEncoder(w).Encode(response)
}

// HandleGetRecord retrieves a record from the repository
func (h *XRPCHandler) HandleGetRecord(w http.ResponseWriter, r *http.Request) {
	repoDID := r.URL.Query().Get("repo")
	collection := r.URL.Query().Get("collection")
	rkey := r.URL.Query().Get("rkey")

	if repoDID == "" || collection == "" || rkey == "" {
		http.Error(w, "missing required parameters", http.StatusBadRequest)
		return
	}

	if repoDID != h.pds.DID() {
		http.Error(w, "invalid repo", http.StatusBadRequest)
		return
	}

	// Use generic repomgr.GetRecord - works for any collection
	// lexutil type registry automatically unmarshals to correct type
	recordCID, recordValue, err := h.pds.repomgr.GetRecord(
		r.Context(),
		h.pds.uid,
		collection,
		rkey,
		cid.Undef,
	)
	if err != nil {
		if strings.Contains(err.Error(), "not found") {
			http.Error(w, "record not found", http.StatusNotFound)
		} else {
			http.Error(w, fmt.Sprintf("failed to get record: %v", err), http.StatusInternalServerError)
		}
		return
	}

	response := map[string]any{
		"uri":   fmt.Sprintf("at://%s/%s/%s", h.pds.DID(), collection, rkey),
		"cid":   recordCID.String(),
		"value": recordValue,
	}

	w.Header().Set("Content-Type", "application/json")
	json.NewEncoder(w).Encode(response)
}

// HandleListRecords lists records in a collection
// Spec: https://docs.bsky.app/docs/api/com-atproto-repo-list-records
// Supports pagination via limit, cursor, and reverse parameters
func (h *XRPCHandler) HandleListRecords(w http.ResponseWriter, r *http.Request) {
	repoDID := r.URL.Query().Get("repo")
	collection := r.URL.Query().Get("collection")

	if repoDID == "" || collection == "" {
		http.Error(w, "missing required parameters", http.StatusBadRequest)
		return
	}

	if repoDID != h.pds.DID() {
		http.Error(w, "invalid repo", http.StatusBadRequest)
		return
	}

	// Parse pagination parameters (per spec)
	limit := 50 // default
	if limitStr := r.URL.Query().Get("limit"); limitStr != "" {
		parsedLimit, err := strconv.Atoi(limitStr)
		if err != nil || parsedLimit < 1 || parsedLimit > 100 {
			http.Error(w, "invalid limit (must be 1-100)", http.StatusBadRequest)
			return
		}
		limit = parsedLimit
	}

	cursor := r.URL.Query().Get("cursor")
	reverse := r.URL.Query().Get("reverse") == "true"

	// Generic implementation using repo.ForEach
	session, err := h.pds.carstore.ReadOnlySession(h.pds.uid)
	if err != nil {
		http.Error(w, fmt.Sprintf("failed to create session: %v", err), http.StatusInternalServerError)
		return
	}

	head, err := h.pds.carstore.GetUserRepoHead(r.Context(), h.pds.uid)
	if err != nil {
		http.Error(w, fmt.Sprintf("failed to get repo head: %v", err), http.StatusInternalServerError)
		return
	}

	if !head.Defined() {
		// Empty repo, return empty list
		response := map[string]any{"records": []any{}}
		w.Header().Set("Content-Type", "application/json")
		json.NewEncoder(w).Encode(response)
		return
	}

	repoHandle, err := repo.OpenRepo(r.Context(), session, head)
	if err != nil {
		http.Error(w, fmt.Sprintf("failed to open repo: %v", err), http.StatusInternalServerError)
		return
	}

	// Initialize as empty slice (not nil) to ensure JSON encodes as [] not null
	records := []map[string]any{}
	var nextCursor string
	skipUntilCursor := cursor != ""

	// Iterate over all records in the collection
	err = repoHandle.ForEach(r.Context(), collection, func(k string, v cid.Cid) error {
		// k is like "io.atcr.hold.captain/self" or "io.atcr.hold.crew/3m3by7msdln22"
		parts := strings.Split(k, "/")
		if len(parts) < 2 {
			return nil // Skip invalid keys
		}

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

		// Filter: only include records that match the requested collection
		// MST keys are sorted lexicographically, so once we hit a different
		// collection prefix, all remaining keys will also be outside our range
		if actualCollection != collection {
			return repo.ErrDoneIterating // Stop walking the tree
		}

		// Handle cursor-based pagination
		if skipUntilCursor {
			if rkey == cursor {
				skipUntilCursor = false // Found cursor, start including records after this
			}
			return nil // Skip this record
		}

		// Check if we've hit the limit
		if len(records) >= limit {
			// Set next cursor to current rkey
			nextCursor = rkey
			return repo.ErrDoneIterating // Stop iteration
		}

		// Get the record bytes
		recordCID, recBytes, err := repoHandle.GetRecordBytes(r.Context(), k)
		if err != nil {
			return fmt.Errorf("failed to get record: %v", err)
		}

		// Decode using lexutil (type registry handles unmarshaling)
		recordValue, err := lexutil.CborDecodeValue(*recBytes)
		if err != nil {
			return fmt.Errorf("failed to decode record: %v", err)
		}

		records = append(records, map[string]any{
			"uri":   fmt.Sprintf("at://%s/%s/%s", h.pds.DID(), actualCollection, rkey),
			"cid":   recordCID.String(),
			"value": recordValue,
		})
		return nil
	})

	if err != nil {
		// ErrDoneIterating is expected when we stop walking early (reached collection boundary or hit limit)
		// Check using strings.Contains because the error may be wrapped
		if err == repo.ErrDoneIterating || strings.Contains(err.Error(), "done iterating") {
			// Successfully stopped at collection boundary or hit pagination limit, continue with collected records
		} else if strings.Contains(err.Error(), "not found") {
			// If the collection doesn't exist yet, return empty list
			records = []map[string]any{}
		} else {
			http.Error(w, fmt.Sprintf("failed to list records: %v", err), http.StatusInternalServerError)
			return
		}
	}

	// Handle reverse order if requested
	if reverse && len(records) > 0 {
		// Reverse the slice
		for i, j := 0, len(records)-1; i < j; i, j = i+1, j-1 {
			records[i], records[j] = records[j], records[i]
		}
	}

	response := map[string]any{
		"records": records,
	}

	// Include cursor in response if there are more records
	if nextCursor != "" {
		response["cursor"] = nextCursor
	}

	w.Header().Set("Content-Type", "application/json")
	json.NewEncoder(w).Encode(response)
}

// HandleDeleteRecord deletes a record from the repository
// Spec: https://docs.bsky.app/docs/api/com-atproto-repo-delete-record
// Accepts JSON input with repo, collection, rkey, and optional swap parameters
func (h *XRPCHandler) HandleDeleteRecord(w http.ResponseWriter, r *http.Request) {
	var err error

	// Parse JSON body (per spec - input is in body, not query params)
	var input struct {
		Repo       string  `json:"repo"`
		Collection string  `json:"collection"`
		Rkey       string  `json:"rkey"`
		SwapRecord *string `json:"swapRecord,omitempty"` // Optional CID for compare-and-swap
		SwapCommit *string `json:"swapCommit,omitempty"` // Optional CID for compare-and-swap
	}

	if err = json.NewDecoder(r.Body).Decode(&input); err != nil {
		http.Error(w, fmt.Sprintf("invalid JSON body: %v", err), http.StatusBadRequest)
		return
	}

	if input.Repo == "" || input.Collection == "" || input.Rkey == "" {
		http.Error(w, "missing required parameters", http.StatusBadRequest)
		return
	}

	if input.Repo != h.pds.DID() {
		http.Error(w, "invalid repo", http.StatusBadRequest)
		return
	}

	// TODO: Implement swap record/commit validation
	// For now, if swap parameters are provided, we should validate them
	// against the current record/commit CID before deleting
	if input.SwapRecord != nil || input.SwapCommit != nil {
		// Parse swap CIDs
		var swapRecordCID, swapCommitCID cid.Cid
		if input.SwapRecord != nil {
			swapRecordCID, err = cid.Decode(*input.SwapRecord)
			if err != nil {
				http.Error(w, "invalid swapRecord CID", http.StatusBadRequest)
				return
			}
		}
		if input.SwapCommit != nil {
			swapCommitCID, err = cid.Decode(*input.SwapCommit)
			if err != nil {
				http.Error(w, "invalid swapCommit CID", http.StatusBadRequest)
				return
			}
		}

		// Validate swap conditions
		if input.SwapRecord != nil {
			// Get current record CID
			currentCID, _, err := h.pds.repomgr.GetRecord(r.Context(), h.pds.uid, input.Collection, input.Rkey, cid.Undef)
			if err != nil {
				if strings.Contains(err.Error(), "not found") {
					http.Error(w, "record not found", http.StatusNotFound)
				} else {
					http.Error(w, fmt.Sprintf("failed to get current record: %v", err), http.StatusInternalServerError)
				}
				return
			}

			if !currentCID.Equals(swapRecordCID) {
				// Swap failed - record CID doesn't match
				w.WriteHeader(http.StatusBadRequest)
				json.NewEncoder(w).Encode(map[string]any{
					"error":   "InvalidSwap",
					"message": "record CID does not match swapRecord",
				})
				return
			}
		}

		// SwapCommit validation would require checking the repo head CID
		// For now, we'll skip this as it's complex and not critical for MVP
		_ = swapCommitCID
	}

	// Delete the record using repomgr
	err = h.pds.repomgr.DeleteRecord(r.Context(), h.pds.uid, input.Collection, input.Rkey)
	if err != nil {
		if strings.Contains(err.Error(), "not found") {
			http.Error(w, "record not found", http.StatusNotFound)
		} else {
			http.Error(w, fmt.Sprintf("failed to delete record: %v", err), http.StatusInternalServerError)
		}
		return
	}

	// Get commit info for response (per spec)
	// The spec requires returning commit metadata
	head, err := h.pds.carstore.GetUserRepoHead(r.Context(), h.pds.uid)
	if err != nil {
		http.Error(w, fmt.Sprintf("failed to get repo head: %v", err), http.StatusInternalServerError)
		return
	}

	rev, err := h.pds.repomgr.GetRepoRev(r.Context(), h.pds.uid)
	if err != nil {
		http.Error(w, fmt.Sprintf("failed to get repo rev: %v", err), http.StatusInternalServerError)
		return
	}

	// Return commit response (per spec)
	response := map[string]any{
		"commit": map[string]any{
			"cid": head.String(),
			"rev": rev,
		},
	}

	w.Header().Set("Content-Type", "application/json")
	json.NewEncoder(w).Encode(response)
}

// HandleSyncGetRecord returns a single record as a CAR file for sync
func (h *XRPCHandler) HandleSyncGetRecord(w http.ResponseWriter, r *http.Request) {
	did := r.URL.Query().Get("did")
	collection := r.URL.Query().Get("collection")
	rkey := r.URL.Query().Get("rkey")

	if did == "" || collection == "" || rkey == "" {
		http.Error(w, "missing required parameters", http.StatusBadRequest)
		return
	}

	if did != h.pds.DID() {
		http.Error(w, "invalid did", http.StatusBadRequest)
		return
	}

	// Use repomgr to get record proof (repo head + all blocks in MST path to record)
	repoHead, blocks, err := h.pds.repomgr.GetRecordProof(r.Context(), h.pds.uid, collection, rkey)
	if err != nil {
		http.Error(w, fmt.Sprintf("failed to get record: %v", err), http.StatusNotFound)
		return
	}

	// Write CAR file with all accessed blocks
	w.Header().Set("Content-Type", "application/vnd.ipld.car")

	// Create a buffer to write the CAR data
	var buf bytes.Buffer

	// Create CAR header with the repo head as root (not the record CID)
	// The CAR file represents a slice of the repo from head to record
	header := &car.CarHeader{
		Roots:   []cid.Cid{repoHead},
		Version: 1,
	}

	// Write the CAR header
	if err := car.WriteHeader(header, &buf); err != nil {
		http.Error(w, fmt.Sprintf("failed to write CAR header: %v", err), http.StatusInternalServerError)
		return
	}

	// Write all logged blocks to the CAR file
	for _, blk := range blocks {
		if err := carutil.LdWrite(&buf, blk.Cid().Bytes(), blk.RawData()); err != nil {
			http.Error(w, fmt.Sprintf("failed to write block to CAR: %v", err), http.StatusInternalServerError)
			return
		}
	}

	// Write the CAR data to the response
	w.Write(buf.Bytes())
}

// HandleGetRepo returns the full repository as a CAR file
// This is the critical endpoint for relay crawling and Bluesky discovery
func (h *XRPCHandler) HandleGetRepo(w http.ResponseWriter, r *http.Request) {
	// Get required 'did' parameter
	did := r.URL.Query().Get("did")
	if did == "" {
		http.Error(w, "missing required parameter: did", http.StatusBadRequest)
		return
	}

	// Validate DID matches this PDS
	if did != h.pds.DID() {
		http.Error(w, "repo not found", http.StatusNotFound)
		return
	}

	// Get optional 'since' parameter for diff export
	since := r.URL.Query().Get("since")

	// Set CAR content type
	w.Header().Set("Content-Type", "application/vnd.ipld.car")

	// Stream the repository CAR file directly to the response
	// ReadRepo handles full export or diff based on 'since' parameter
	err := h.pds.repomgr.ReadRepo(r.Context(), h.pds.uid, since, w)
	if err != nil {
		// Error already written to response by ReadRepo streaming
		// Log it but don't try to write another HTTP error
		slog.Error("Error streaming repo CAR", "error", err)
		return
	}
}

// WebSocket upgrader
var upgrader = websocket.Upgrader{
	CheckOrigin: func(r *http.Request) bool {
		// Allow all origins for MVP (ATProto firehose is public)
		return true
	},
}

// HandleSubscribeRepos handles WebSocket connections for the firehose
// This is the real-time event stream for repo changes
func (h *XRPCHandler) HandleSubscribeRepos(w http.ResponseWriter, r *http.Request) {
	// Check if broadcaster is configured
	if h.broadcaster == nil {
		http.Error(w, "firehose not enabled", http.StatusNotImplemented)
		return
	}

	// Get optional cursor parameter for backfill
	// Default to -1 (no backfill, only stream new events)
	// cursor=0 means "replay all events from the beginning"
	var cursor int64 = -1
	if cursorStr := r.URL.Query().Get("cursor"); cursorStr != "" {
		var err error
		cursor, err = strconv.ParseInt(cursorStr, 10, 64)
		if err != nil {
			http.Error(w, "invalid cursor parameter", http.StatusBadRequest)
			return
		}
	}

	// Upgrade to WebSocket
	conn, err := upgrader.Upgrade(w, r, nil)
	if err != nil {
		slog.Error("WebSocket upgrade failed", "error", err)
		return
	}

	// Subscribe to events
	// The broadcaster's handleSubscriber goroutine will manage this connection
	// and handle cleanup when the client disconnects
	h.broadcaster.Subscribe(conn, cursor)
}

// HandleUploadBlob handles blob uploads with support for multipart operations
// Direct blob upload: POST with raw bytes (ATProto-compliant)
func (h *XRPCHandler) HandleUploadBlob(w http.ResponseWriter, r *http.Request) {
	// Get authenticated user from context (if coming through middleware)
	// Otherwise validate directly (for tests or direct handler calls)
	user := getUserFromContext(r)
	if user == nil {
		var err error
		user, err = ValidateOwnerOrCrewAdmin(r, h.pds, h.httpClient)
		if err != nil {
			http.Error(w, fmt.Sprintf("authorization failed: %v", err), http.StatusForbidden)
			return
		}
	}

	// Use authenticated user's DID for ATProto blob storage (per-DID paths)
	did := user.DID

	// Read all data into memory to compute CID
	// For large files, this should use multipart upload instead
	blobData, err := io.ReadAll(r.Body)
	if err != nil {
		http.Error(w, fmt.Sprintf("failed to read blob data: %v", err), http.StatusInternalServerError)
		return
	}

	size := int64(len(blobData))

	// Compute SHA-256 hash
	hash := sha256.Sum256(blobData)

	// Create CIDv1 with SHA-256 multihash
	mh, err := multihash.EncodeName(hash[:], "sha2-256")
	if err != nil {
		http.Error(w, fmt.Sprintf("failed to encode multihash: %v", err), http.StatusInternalServerError)
		return
	}

	// Create CIDv1 with raw codec (0x55)
	// ATProto uses CIDv1 with raw codec for blobs
	blobCID := cid.NewCidV1(0x55, mh)

	// Store blob via distribution driver at ATProto path
	path := atprotoBlobPath(did, blobCID.String())

	// Write blob to storage using distribution driver
	writer, err := h.storageDriver.Writer(r.Context(), path, false)
	if err != nil {
		http.Error(w, fmt.Sprintf("failed to create writer: %v", err), http.StatusInternalServerError)
		return
	}

	// Write data
	n, err := io.Copy(writer, bytes.NewReader(blobData))
	if err != nil {
		writer.Cancel(r.Context())
		http.Error(w, fmt.Sprintf("failed to write blob: %v", err), http.StatusInternalServerError)
		return
	}

	// Commit the write
	if err := writer.Commit(r.Context()); err != nil {
		http.Error(w, fmt.Sprintf("failed to commit blob: %v", err), http.StatusInternalServerError)
		return
	}

	if n != size {
		http.Error(w, fmt.Sprintf("size mismatch: wrote %d bytes, expected %d", n, size), http.StatusInternalServerError)
		return
	}

	// Return ATProto-compliant blob response
	response := map[string]any{
		"blob": map[string]any{
			"$type": "blob",
			"ref": map[string]any{
				"$link": blobCID.String(),
			},
			"mimeType": "application/octet-stream",
			"size":     size,
		},
	}

	w.Header().Set("Content-Type", "application/json")
	json.NewEncoder(w).Encode(response)
}

// HandleGetBlob routes blob requests to appropriate handlers based on blob type
// Routes to:
//   - handleGetOCIBlob for OCI image blobs (sha256:...)
//   - handleGetATProtoBlob for ATProto blobs (CID format)
func (h *XRPCHandler) HandleGetBlob(w http.ResponseWriter, r *http.Request) {
	did := r.URL.Query().Get("did")
	cidOrDigest := r.URL.Query().Get("cid")

	slog.Debug("HandleGetBlob request",
		"method", r.Method,
		"did", did,
		"cid", cidOrDigest)

	if did == "" || cidOrDigest == "" {
		http.Error(w, "missing required parameters", http.StatusBadRequest)
		return
	}

	// Route based on blob type
	if strings.HasPrefix(cidOrDigest, "sha256:") {
		// OCI blob (container image layers)
		h.handleGetOCIBlob(w, r, did, cidOrDigest)
		return
	}

	// ATProto blob (profile avatars, etc.)
	h.handleGetATProtoBlob(w, r, did, cidOrDigest)
}

// handleGetOCIBlob handles OCI container image blob requests
// Returns JSON with presigned URL for AppView integration
// Authorization: Protected by hold access control (captain.public or crew with blob:read)
func (h *XRPCHandler) handleGetOCIBlob(w http.ResponseWriter, r *http.Request, did, digest string) {
	slog.Debug("Processing OCI blob", "digest", digest)

	// Validate blob read access (hold access control)
	// If captain.public = true, returns nil (public access allowed)
	// If captain.public = false, validates auth and checks for blob:read permission
	_, err := ValidateBlobReadAccess(r, h.pds, h.httpClient)
	if err != nil {
		slog.Warn("OCI blob authorization failed", "error", err, "digest", digest)
		http.Error(w, fmt.Sprintf("authorization failed: %v", err), http.StatusForbidden)
		return
	}

	// Determine presigned URL operation (GET or HEAD)
	// Check for ?method=HEAD query parameter first (from AppView)
	operation := r.URL.Query().Get("method")
	if operation == "" {
		operation = "GET"
	}

	// Generate presigned URL (use empty DID for content-addressed storage)
	presignedURL, err := h.GetPresignedURL(r.Context(), operation, digest, "")
	if err != nil {
		slog.Error("Failed to get presigned URL for OCI blob",
			"error", err,
			"operation", operation,
			"digest", digest)
		http.Error(w, "failed to get presigned URL", http.StatusInternalServerError)
		return
	}

	slog.Debug("Returning presigned URL for OCI blob",
		"operation", operation,
		"digest", digest,
		"url", presignedURL)

	// Return JSON response with presigned URL (AppView expects this format)
	response := map[string]string{
		"url": presignedURL,
	}
	w.Header().Set("Content-Type", "application/json")
	json.NewEncoder(w).Encode(response)
}

// handleGetATProtoBlob handles standard ATProto blob requests
// Returns 307 redirect to presigned URL (standard ATProto behavior)
// Authorization: Public per ATProto spec (no auth required)
func (h *XRPCHandler) handleGetATProtoBlob(w http.ResponseWriter, r *http.Request, did, cid string) {
	slog.Debug("Processing ATProto blob", "cid", cid)

	// Validate DID (ATProto blobs are stored per-DID for data sovereignty)
	if did != h.pds.DID() {
		slog.Warn("ATProto blob DID mismatch",
			"got", did,
			"expected", h.pds.DID())
		http.Error(w, "invalid did", http.StatusBadRequest)
		return
	}

	// Determine presigned URL operation (GET or HEAD)
	operation := r.URL.Query().Get("method")
	if operation == "" {
		operation = "GET"
	}

	// Generate presigned URL (use DID for per-DID storage path)
	presignedURL, err := h.GetPresignedURL(r.Context(), operation, cid, did)
	if err != nil {
		slog.Error("Failed to get presigned URL for ATProto blob",
			"error", err,
			"operation", operation,
			"cid", cid,
			"did", did)
		http.Error(w, "failed to get presigned URL", http.StatusInternalServerError)
		return
	}

	// Return 307 redirect (standard ATProto behavior - client fetches blob directly)
	http.Redirect(w, r, presignedURL, http.StatusTemporaryRedirect)
}

// HandleListRepos lists all repositories in this PDS
func (h *XRPCHandler) HandleListRepos(w http.ResponseWriter, r *http.Request) {
	// Single-user PDS: return just this hold's repo
	did := h.pds.DID()

	// Get repo head and rev from repomgr
	// For a single-user PDS, we use a fixed UID (stored in pds.uid)
	head, err := h.pds.repomgr.GetRepoRoot(r.Context(), h.pds.uid)
	if err != nil {
		// If no repo exists yet, return empty list
		response := map[string]any{
			"repos": []any{},
		}
		w.Header().Set("Content-Type", "application/json")
		json.NewEncoder(w).Encode(response)
		return
	}

	rev, err := h.pds.repomgr.GetRepoRev(r.Context(), h.pds.uid)
	if err != nil || rev == "" {
		// No commits yet, return empty list
		// Don't expose repos with no revision (empty/uninitialized)
		response := map[string]any{
			"repos": []any{},
		}
		w.Header().Set("Content-Type", "application/json")
		json.NewEncoder(w).Encode(response)
		return
	}

	repos := []map[string]any{
		{
			"did":    did,
			"head":   head.String(),
			"rev":    rev,
			"active": true,
		},
	}

	response := map[string]any{
		"repos": repos,
	}

	w.Header().Set("Content-Type", "application/json")
	json.NewEncoder(w).Encode(response)
}

// HandleGetRepoStatus returns the hosting status for a repository
// Spec: https://docs.bsky.app/docs/api/com-atproto-sync-get-repo-status
func (h *XRPCHandler) HandleGetRepoStatus(w http.ResponseWriter, r *http.Request) {
	// Get required 'did' parameter
	did := r.URL.Query().Get("did")
	if did == "" {
		http.Error(w, "missing required parameter: did", http.StatusBadRequest)
		return
	}

	// Validate DID matches this PDS (single-user PDS only hosts one repo)
	if did != h.pds.DID() {
		http.Error(w, "repo not found", http.StatusNotFound)
		return
	}

	// Get current repo revision to verify repo is initialized
	rev, err := h.pds.repomgr.GetRepoRev(r.Context(), h.pds.uid)
	if err != nil || rev == "" {
		// Repo exists (DID matches) but no commits yet
		// Per ATProto spec, return active=true even if empty
		response := map[string]any{
			"did":    did,
			"active": true,
		}
		w.Header().Set("Content-Type", "application/json")
		json.NewEncoder(w).Encode(response)
		return
	}

	// Return status with revision
	response := map[string]any{
		"did":    did,
		"active": true,
		"rev":    rev,
	}

	w.Header().Set("Content-Type", "application/json")
	json.NewEncoder(w).Encode(response)
}

// HandleDIDDocument returns the DID document
func (h *XRPCHandler) HandleDIDDocument(w http.ResponseWriter, r *http.Request) {
	doc, err := h.pds.GenerateDIDDocument(h.pds.PublicURL)
	if err != nil {
		http.Error(w, fmt.Sprintf("failed to generate DID document: %v", err), http.StatusInternalServerError)
		return
	}

	w.Header().Set("Content-Type", "application/json")
	json.NewEncoder(w).Encode(doc)
}

// HandleAtprotoDID returns the DID for handle resolution
func (h *XRPCHandler) HandleAtprotoDID(w http.ResponseWriter, r *http.Request) {
	w.Header().Set("Content-Type", "text/plain")
	fmt.Fprint(w, h.pds.DID())
}

// HandleRequestCrew handles crew membership requests
// This endpoint allows authenticated users to request crew membership
// Authorization is checked against captain record settings
func (h *XRPCHandler) HandleRequestCrew(w http.ResponseWriter, r *http.Request) {
	slog.Debug("Starting crew membership request")

	// Get authenticated user from context (if coming through middleware)
	// Otherwise validate directly (for tests or direct handler calls)
	user := getUserFromContext(r)
	if user == nil {
		var err error
		user, err = ValidateDPoPRequest(r, h.httpClient)
		if err != nil {
			slog.Warn("Crew request authentication failed", "error", err)
			http.Error(w, fmt.Sprintf("authentication failed: %v", err), http.StatusUnauthorized)
			return
		}
	}
	slog.Debug("Authenticated user for crew request", "did", user.DID)

	// Parse request body (optional parameters)
	var req struct {
		Role        string   `json:"role"`        // Requested role (default: "member")
		Permissions []string `json:"permissions"` // Requested permissions
	}

	// Body is optional - if empty, just use defaults
	if r.Body != nil && r.ContentLength > 0 {
		if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
			slog.Warn("Failed to parse crew request body", "error", err)
			http.Error(w, fmt.Sprintf("invalid request body: %v", err), http.StatusBadRequest)
			return
		}
	}

	// Get captain record to check authorization settings
	slog.Debug("Getting captain record for crew request")
	_, captain, err := h.pds.GetCaptainRecord(r.Context())
	if err != nil {
		slog.Error("Failed to get captain record", "error", err)
		http.Error(w, fmt.Sprintf("failed to get captain record: %v", err), http.StatusInternalServerError)
		return
	}
	slog.Debug("Captain record retrieved",
		"owner", captain.Owner,
		"allowAllCrew", captain.AllowAllCrew)

	// Check authorization:
	// 1. If allowAllCrew is true, any authenticated user can join
	// 2. If user is the owner, they can always join (though they should already be crew)
	// 3. Otherwise, deny
	isOwner := user.DID == captain.Owner
	if !captain.AllowAllCrew && !isOwner {
		http.Error(w, "crew registration not allowed (HOLD_ALLOW_ALL_CREW=false)", http.StatusForbidden)
		return
	}

	// Set defaults if not provided
	if req.Role == "" {
		req.Role = "member"
	}
	if len(req.Permissions) == 0 {
		req.Permissions = []string{"blob:read", "blob:write"}
	}

	// Check if user is already a crew member
	// List all crew members and check if this DID is already present
	slog.Debug("Checking existing crew membership")
	crew, err := h.pds.ListCrewMembers(r.Context())
	if err != nil {
		slog.Error("Failed to list crew members", "error", err)
		http.Error(w, fmt.Sprintf("failed to list crew members: %v", err), http.StatusInternalServerError)
		return
	}
	slog.Debug("Found existing crew members", "count", len(crew))

	for _, member := range crew {
		if member.Record.Member == user.DID {
			// Already a crew member, return success with existing record
			slog.Debug("User is already a crew member",
				"did", user.DID,
				"rkey", member.Rkey)
			response := map[string]any{
				"uri":     fmt.Sprintf("at://%s/%s/%s", h.pds.DID(), atproto.CrewCollection, member.Rkey),
				"cid":     member.Cid.String(),
				"status":  "already_member",
				"message": "User is already a crew member",
			}
			w.Header().Set("Content-Type", "application/json")
			w.WriteHeader(http.StatusOK)
			json.NewEncoder(w).Encode(response)
			return
		}
	}

	// Create new crew record
	slog.Debug("Creating new crew record",
		"did", user.DID,
		"role", req.Role,
		"permissions", req.Permissions)
	recordCID, err := h.pds.AddCrewMember(r.Context(), user.DID, req.Role, req.Permissions)
	if err != nil {
		slog.Error("Failed to create crew record",
			"error", err,
			"did", user.DID)
		http.Error(w, fmt.Sprintf("failed to create crew record: %v", err), http.StatusInternalServerError)
		return
	}
	slog.Info("Successfully created crew record",
		"did", user.DID,
		"cid", recordCID.String())

	// Return success response
	// Note: rkey is generated by AddCrewMember (TID), we don't have direct access to it
	// For now, return just the CID. In production, AddCrewMember should return both CID and rkey
	response := map[string]any{
		"cid":     recordCID.String(),
		"status":  "created",
		"message": "Successfully added to crew",
	}

	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(http.StatusCreated)
	json.NewEncoder(w).Encode(response)
}

// GetPresignedURL generates a presigned URL for GET, HEAD, or PUT operations
// Distinguishes between ATProto blobs (per-DID) and OCI blobs (content-addressed)
func (h *XRPCHandler) GetPresignedURL(ctx context.Context, operation string, digest string, did string) (string, error) {
	var path string

	// Determine blob type and construct appropriate path
	if strings.HasPrefix(digest, "sha256:") || strings.HasPrefix(digest, "uploads/") {
		// OCI container layer (sha256 digest or temp upload path)
		// Use content-addressed storage (globally deduplicated)
		path = s3.BlobPath(digest)
	} else {
		// ATProto blob (CID format like bafyreib...)
		// Use per-DID storage for data sovereignty
		if did == "" {
			return "", fmt.Errorf("DID required for ATProto blob storage")
		}
		path = atprotoBlobPath(did, digest)
	}

	// Generate presigned URL if S3 client is available
	if h.s3Service.Client != nil {
		// Build S3 key from blob path
		s3Key := strings.TrimPrefix(path, "/")
		if h.s3Service.PathPrefix != "" {
			s3Key = h.s3Service.PathPrefix + "/" + s3Key
		}

		// Create appropriate S3 request based on operation
		var req interface {
			Presign(time.Duration) (string, error)
		}
		contentType := "application/octet-stream"
		switch operation {
		case http.MethodGet:
			// Note: Don't use ResponseContentType - not supported by all S3-compatible services
			req, _ = h.s3Service.Client.GetObjectRequest(&awss3.GetObjectInput{
				Bucket: &h.s3Service.Bucket,
				Key:    &s3Key,
			})

		case http.MethodHead:
			req, _ = h.s3Service.Client.HeadObjectRequest(&awss3.HeadObjectInput{
				Bucket: &h.s3Service.Bucket,
				Key:    &s3Key,
			})

		case http.MethodPut:
			req, _ = h.s3Service.Client.PutObjectRequest(&awss3.PutObjectInput{
				Bucket:      &h.s3Service.Bucket,
				Key:         &s3Key,
				ContentType: &contentType,
			})

		default:
			return "", fmt.Errorf("unsupported operation: %s", operation)
		}

		// Generate presigned URL with 15 minute expiry
		url, err := req.Presign(15 * time.Minute)
		if err != nil {
			slog.Warn("Presign failed, falling back to XRPC endpoint",
				"error", err,
				"operation", operation,
				"digest", digest)
			slog.Debug("Using XRPC proxy fallback")
			proxyURL := getProxyURL(h.pds.PublicURL, digest, did, operation)
			if proxyURL == "" {
				return "", fmt.Errorf("presign failed and XRPC proxy not supported for PUT operations")
			}
			return proxyURL, nil
		}

		return url, nil
	}

	// Fallback: return XRPC endpoint through this service
	proxyURL := getProxyURL(h.pds.PublicURL, digest, did, operation)
	if proxyURL == "" {
		return "", fmt.Errorf("S3 client not available and XRPC proxy not supported for PUT operations")
	}
	return proxyURL, nil
}

// atprotoBlobPath creates a per-DID storage path for ATProto blobs
// ATProto spec stores blobs as: /repos/{did}/blobs/{cid}/data
// This provides data sovereignty - each user's blobs are isolated
func atprotoBlobPath(did, cid string) string {
	// Clean DID for filesystem safety (replace : with -)
	safeDID := strings.ReplaceAll(did, ":", "-")
	return fmt.Sprintf("/repos/%s/blobs/%s/data", safeDID, cid)
}

// getProxyURL returns XRPC endpoint for blob operations (fallback when presigned URLs unavailable)
// For GET/HEAD operations, returns the XRPC getBlob endpoint
// For PUT operations, this fallback is no longer supported - use multipart upload instead
func getProxyURL(publicURL string, digest, did string, operation string) string {
	// For read operations, use XRPC getBlob endpoint
	if operation == http.MethodGet || operation == http.MethodHead {
		// Generate hold DID from public URL using shared function
		holdDID := atproto.ResolveHoldDIDFromURL(publicURL)
		return fmt.Sprintf("%s%s?did=%s&cid=%s",
			publicURL, atproto.SyncGetBlob, holdDID, digest)
	}

	// For PUT operations, proxy fallback is not supported with XRPC
	// Clients should use multipart upload flow via com.atproto.repo.uploadBlob
	return ""
}