diff --git a/docs/XRPC_BLOB_MIGRATION.md b/docs/XRPC_BLOB_MIGRATION.md index 092f581..9bd57d8 100644 --- a/docs/XRPC_BLOB_MIGRATION.md +++ b/docs/XRPC_BLOB_MIGRATION.md @@ -7,6 +7,12 @@ This document describes how to migrate from separate legacy multipart upload end ### Legacy HTTP Endpoints (cmd/hold/main.go) ```go +// Unified presigned URL endpoint (handles upload AND download) +mux.HandleFunc("/presigned-url", service.HandlePresignedURL) + +// Internal move operation (used by multipart complete) +mux.HandleFunc("/move", service.HandleMove) + // Multipart upload endpoints mux.HandleFunc("/start-multipart", service.HandleStartMultipart) mux.HandleFunc("/part-presigned-url", service.HandleGetPartURL) @@ -45,12 +51,68 @@ func (h *XRPCHandler) HandleUploadBlob(w http.ResponseWriter, r *http.Request) { - Currently not used by XRPC handlers **pkg/hold/handlers.go:** +- `HandlePresignedURL()` - Unified endpoint for GET/HEAD/PUT presigned URLs +- `HandleMove()` - Moves blob from temp to final location (internal operation) - `HandleStartMultipart()` - Starts upload, returns uploadID - `HandleGetPartURL()` - Returns presigned URL for part -- `HandleCompleteMultipart()` - Finalizes upload, assembles parts +- `HandleCompleteMultipart()` - Finalizes upload, assembles parts (calls Move internally) - `HandleAbortMultipart()` - Cancels upload - `HandleMultipartPartUpload()` - Buffered part upload fallback +## Legacy Endpoint Mapping + +### `/presigned-url` → Multiple XRPC Operations + +The legacy `/presigned-url` endpoint is a **unified endpoint** that handles both upload and download operations based on the `operation` field in the JSON body: + +**Legacy format:** +``` +POST /presigned-url +Content-Type: application/json + +{ + "operation": "GET", // or "HEAD" or "PUT" + "did": "did:plc:alice123", + "digest": "sha256:abc123...", + "size": 1234567890 // Only for PUT operations +} + +Response: +{ + "url": "https://s3.amazonaws.com/...", + "expires_at": "2025-10-16T..." +} +``` + +**XRPC mapping:** +- `operation: "GET"` → `GET /xrpc/com.atproto.sync.getBlob?did=...&cid=sha256:abc...` +- `operation: "HEAD"` → `HEAD /xrpc/com.atproto.sync.getBlob?did=...&cid=sha256:abc...` +- `operation: "PUT"` → `com.atproto.repo.uploadBlob` (single upload via presigned URL) + +**Note:** For GET/HEAD operations, AppView passes OCI digest directly as `cid` parameter. Hold detects `sha256:` prefix and uses digest directly (no CID conversion needed). + +### `/move` → Internal to Multipart Complete + +The legacy `/move` endpoint moves a blob from temporary location to final digest-based location: + +**Legacy format:** +``` +POST /move?from=uploads/temp-123&to=sha256:abc123...&did=did:plc:alice123 + +Response: 200 OK +``` + +**Purpose:** Server-side S3 copy after multipart assembly. Used in this flow: + +1. Multipart parts uploaded → `uploads/temp-{uploadID}/part-1`, `part-2`, etc. +2. Complete multipart → S3 assembles parts at `uploads/temp-{uploadID}` +3. **Move operation** → S3 copy from `uploads/temp-{uploadID}` → `blobs/sha256/ab/abc123...` + +**XRPC mapping:** +- **Not a separate endpoint** - becomes internal operation in `uploadBlob?action=complete` +- The `complete` action automatically handles the move after multipart assembly +- AppView doesn't need to call move explicitly in XRPC flow + ## New Unified Design ### Single Endpoint: `com.atproto.repo.uploadBlob` @@ -59,6 +121,49 @@ Content-Type discrimination determines operation: - `application/octet-stream` → Standard blob upload (profile images, small media) - `application/json` → Multipart operations (large OCI layers) +### Complementary Endpoint: `com.atproto.sync.getBlob` + +For blob downloads (maps from legacy `/presigned-url` with operation=GET/HEAD): + +**Standard ATProto blobs (CID):** +``` +GET /xrpc/com.atproto.sync.getBlob?did={holdDID}&cid=bafyreib... + +Response: 307 Temporary Redirect +Location: https://s3.amazonaws.com/bucket/...?presigned-params +``` + +**OCI container layers (digest):** +``` +GET /xrpc/com.atproto.sync.getBlob?did={holdDID}&cid=sha256:abc123... + +Response: 307 Temporary Redirect +Location: https://s3.amazonaws.com/bucket/...?presigned-params +``` + +**Implementation - Flexible CID parameter:** +```go +func (h *XRPCHandler) HandleGetBlob(w http.ResponseWriter, r *http.Request) { + cidOrDigest := r.URL.Query().Get("cid") + + var digest string + if strings.HasPrefix(cidOrDigest, "sha256:") { + // OCI digest - use directly (no conversion needed) + digest = cidOrDigest + } else { + // Standard CID - convert to digest + c, _ := cid.Decode(cidOrDigest) + digest = cidToDigest(c) // bafyreib... → sha256:abc... + } + + // Generate presigned URL for S3 + url := h.blobStore.GetPresignedDownloadURL(digest) + http.Redirect(w, r, url, http.StatusTemporaryRedirect) +} +``` + +**Key insight:** The `cid` parameter accepts both formats. Hold service checks prefix and handles accordingly. This keeps the endpoint spec-compliant (GET with query params) while supporting OCI digests natively. + ### API Specification #### Standard Single Upload (ATProto Spec Compliant) @@ -201,6 +306,9 @@ Response (200 OK): - Retrieve session: `multipartMgr.GetSession(uploadID)` - For S3Native: Record parts via `session.RecordS3Part()` - Call `service.CompleteMultipartUploadWithManager(ctx, session, multipartMgr)` + - This internally calls S3 CompleteMultipartUpload to assemble parts + - Then performs server-side S3 copy from temp location to final digest location + - Equivalent to legacy `/move` endpoint operation - Convert digest to CID for response #### Multipart Abort (ATCR Extension) @@ -547,9 +655,20 @@ func (h *HoldServiceBlobStore) UploadBlob(ctx context.Context, data io.Reader) ( Create new XRPC client or update ProxyBlobStore to use unified endpoint: +**Download (GET/HEAD):** ```go -// In ProxyBlobStore or new XRPCBlobStore +func (p *ProxyBlobStore) ServeBlob(ctx context.Context, w http.ResponseWriter, r *http.Request, dgst digest.Digest) error { + // Pass digest directly as cid parameter (no conversion) + url := fmt.Sprintf("%s/xrpc/com.atproto.sync.getBlob?did=%s&cid=%s", + p.storageEndpoint, p.holdDID, dgst.String()) // cid=sha256:abc... + http.Redirect(w, r, url, http.StatusTemporaryRedirect) + return nil +} +``` + +**Multipart Upload:** +```go func (p *ProxyBlobStore) startMultipartUpload(ctx context.Context, digest string) (string, error) { reqBody := map[string]any{ "action": "start", @@ -612,6 +731,8 @@ Once all holds migrated and tested: **cmd/hold/main.go - Remove:** ```go // DELETE these lines +mux.HandleFunc("/presigned-url", service.HandlePresignedURL) +mux.HandleFunc("/move", service.HandleMove) mux.HandleFunc("/start-multipart", service.HandleStartMultipart) mux.HandleFunc("/part-presigned-url", service.HandleGetPartURL) mux.HandleFunc("/complete-multipart", service.HandleCompleteMultipart) @@ -619,20 +740,45 @@ mux.HandleFunc("/abort-multipart", service.HandleAbortMultipart) mux.HandleFunc("/multipart-parts/", ...) ``` -**pkg/hold/handlers.go - Mark as deprecated:** +**pkg/hold/handlers.go - Remove HTTP handler wrappers:** ```go -// Keep methods for now (used by service internals) -// But remove HTTP handler wrappers +// DELETE these functions: +// - HandlePresignedURL() - replaced by uploadBlob + getBlob XRPC endpoints +// - HandleMove() - now internal operation in CompleteMultipartUploadWithManager() +// - HandleStartMultipart() - replaced by uploadBlob?action=start +// - HandleGetPartURL() - replaced by uploadBlob?action=part +// - HandleCompleteMultipart() - replaced by uploadBlob?action=complete +// - HandleAbortMultipart() - replaced by uploadBlob?action=abort +// - HandleMultipartPartUpload() - replaced by uploadBlob PUT with headers + +// KEEP internal service methods: +// - s.getPresignedURL() - still used by blobstore_adapter +// - s.driver.Move() - still used for temp→final move +// - s.StartMultipartUploadWithManager() - core multipart logic +// - s.GetPartUploadURL() - presigned URL generation +// - s.CompleteMultipartUploadWithManager() - includes move operation +// - s.AbortMultipartUploadWithManager() - cleanup logic ``` ## Key Design Decisions 1. **Content-Type discrimination**: Natural way to distinguish single vs multipart uploads -2. **JSON bodies for multipart**: Follows XRPC conventions (like putRecord, deleteRecord) +2. **JSON bodies for all parameters**: Follows XRPC conventions (like putRecord, deleteRecord) + - **No query parameters** - all operation details in request body + - Makes requests more inspectable and debuggable + - Easier to extend with new fields 3. **Preserve standard uploadBlob**: Raw bytes still work for profile images, small media 4. **Reuse existing code**: HoldService multipart logic unchanged, just new HTTP layer 5. **Backward compatibility**: Both endpoints active during transition 6. **Action-based routing**: Clear, extensible JSON structure +7. **Move is internal**: `/move` endpoint logic absorbed into multipart complete operation + - No separate XRPC endpoint needed + - Simplifies AppView client code +8. **Unified presigned URL handling**: Single `uploadBlob`/`getBlob` pair replaces operation-based routing +9. **Flexible CID parameter**: `getBlob` accepts both standard CIDs and OCI digests via prefix detection + - Keeps endpoint spec-compliant (GET with query params) + - No conversion overhead on AppView side + - Hold does simple prefix check: `sha256:` → use directly, else → convert CID ## Benefits diff --git a/pkg/hold/blobstore_adapter.go b/pkg/hold/blobstore_adapter.go index af7c7dd..b4cc36f 100644 --- a/pkg/hold/blobstore_adapter.go +++ b/pkg/hold/blobstore_adapter.go @@ -1,9 +1,15 @@ package hold import ( + "bytes" "context" + "crypto/sha256" + "fmt" + "io" "atcr.io/pkg/hold/pds" + "github.com/ipfs/go-cid" + "github.com/multiformats/go-multihash" ) // HoldServiceBlobStore adapts the hold service to implement the pds.BlobStore interface @@ -21,10 +27,16 @@ func NewHoldServiceBlobStore(service *HoldService, holdDID string) pds.BlobStore } // GetPresignedDownloadURL returns a presigned URL for downloading a blob -func (b *HoldServiceBlobStore) GetPresignedDownloadURL(digest string) (string, error) { - // Use the hold service's existing presigned URL logic +func (b *HoldServiceBlobStore) GetPresignedDownloadURL(digest, did string) (string, error) { + // Use provided DID if given, otherwise fall back to hold's DID + // ATProto blobs require DID for per-user storage + // OCI blobs (sha256:...) use content-addressed storage + if did == "" { + did = b.holdDID + } + ctx := context.Background() - url, err := b.service.GetPresignedURL(ctx, OperationGet, digest, b.holdDID) + url, err := b.service.GetPresignedURL(ctx, OperationGet, digest, did) if err != nil { return "", err } @@ -32,12 +44,150 @@ func (b *HoldServiceBlobStore) GetPresignedDownloadURL(digest string) (string, e } // GetPresignedUploadURL returns a presigned URL for uploading a blob -func (b *HoldServiceBlobStore) GetPresignedUploadURL(digest string) (string, error) { - // Use the hold service's existing presigned URL logic +func (b *HoldServiceBlobStore) GetPresignedUploadURL(digest, did string) (string, error) { + // Use provided DID if given, otherwise fall back to hold's DID + // ATProto blobs require DID for per-user storage + // OCI blobs (sha256:...) use content-addressed storage + if did == "" { + did = b.holdDID + } + ctx := context.Background() - url, err := b.service.GetPresignedURL(ctx, OperationPut, digest, b.holdDID) + url, err := b.service.GetPresignedURL(ctx, OperationPut, digest, did) if err != nil { return "", err } return url, nil } + +// StartMultipartUpload initiates a multipart upload +func (b *HoldServiceBlobStore) StartMultipartUpload(ctx context.Context, digest string) (string, string, error) { + uploadID, mode, err := b.service.StartMultipartUploadWithManager(ctx, digest, b.service.MultipartMgr) + if err != nil { + return "", "", err + } + + // Convert mode to string for XRPC response + var modeStr string + switch mode { + case S3Native: + modeStr = "s3native" + case Buffered: + modeStr = "buffered" + default: + modeStr = "unknown" + } + + return uploadID, modeStr, nil +} + +// GetPartUploadURL returns a presigned URL for uploading a specific part +func (b *HoldServiceBlobStore) GetPartUploadURL(ctx context.Context, uploadID string, partNumber int, did string) (string, error) { + session, err := b.service.MultipartMgr.GetSession(uploadID) + if err != nil { + return "", err + } + + return b.service.GetPartUploadURL(ctx, session, partNumber, did) +} + +// CompleteMultipartUpload finalizes a multipart upload +func (b *HoldServiceBlobStore) CompleteMultipartUpload(ctx context.Context, uploadID string, parts []pds.PartInfo) error { + session, err := b.service.MultipartMgr.GetSession(uploadID) + if err != nil { + return err + } + + // For S3Native mode, record parts from XRPC request (they have ETags from S3) + if session.Mode == S3Native { + for _, p := range parts { + session.RecordS3Part(p.PartNumber, p.ETag, 0) + } + } + + return b.service.CompleteMultipartUploadWithManager(ctx, session, b.service.MultipartMgr) +} + +// AbortMultipartUpload cancels a multipart upload +func (b *HoldServiceBlobStore) AbortMultipartUpload(ctx context.Context, uploadID string) error { + session, err := b.service.MultipartMgr.GetSession(uploadID) + if err != nil { + return err + } + + return b.service.AbortMultipartUploadWithManager(ctx, session, b.service.MultipartMgr) +} + +// HandleBufferedPartUpload handles uploading a part in buffered mode +func (b *HoldServiceBlobStore) HandleBufferedPartUpload(ctx context.Context, uploadID string, partNumber int, data []byte) (string, error) { + session, err := b.service.MultipartMgr.GetSession(uploadID) + if err != nil { + return "", err + } + + if session.Mode != Buffered { + return "", fmt.Errorf("session is not in buffered mode") + } + + etag := session.StorePart(partNumber, data) + return etag, nil +} + +// UploadBlob receives raw blob bytes, computes CID, and stores via distribution driver +// This is used for standard ATProto blob uploads (profile pics, small media) +func (b *HoldServiceBlobStore) UploadBlob(ctx context.Context, did string, data io.Reader) (cid.Cid, int64, error) { + // Use provided DID if given, otherwise fall back to hold's DID + if did == "" { + did = b.holdDID + } + + // Read all data into memory to compute CID + // For large files, this should use multipart upload instead + blobData, err := io.ReadAll(data) + if err != nil { + return cid.Undef, 0, fmt.Errorf("failed to read blob data: %w", err) + } + + 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 { + return cid.Undef, 0, fmt.Errorf("failed to encode multihash: %w", err) + } + + // 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: /repos/{did}/blobs/{cid}/data + path := atprotoBlobPath(did, blobCID.String()) + + // Write blob to storage using distribution driver + writer, err := b.service.driver.Writer(ctx, path, false) + if err != nil { + return cid.Undef, 0, fmt.Errorf("failed to create writer: %w", err) + } + + // Write data + n, err := io.Copy(writer, bytes.NewReader(blobData)) + if err != nil { + writer.Cancel(ctx) + return cid.Undef, 0, fmt.Errorf("failed to write blob: %w", err) + } + + // Commit the write + if err := writer.Commit(ctx); err != nil { + return cid.Undef, 0, fmt.Errorf("failed to commit blob: %w", err) + } + + if n != size { + return cid.Undef, 0, fmt.Errorf("size mismatch: wrote %d bytes, expected %d", n, size) + } + + return blobCID, size, nil +} diff --git a/pkg/hold/pds/xrpc.go b/pkg/hold/pds/xrpc.go index 2101360..1a989f9 100644 --- a/pkg/hold/pds/xrpc.go +++ b/pkg/hold/pds/xrpc.go @@ -2,8 +2,10 @@ package pds import ( "bytes" + "context" "encoding/json" "fmt" + "io" "net/http" "strconv" "strings" @@ -30,9 +32,36 @@ type XRPCHandler struct { // BlobStore interface wraps the existing hold service storage operations type BlobStore interface { // GetPresignedDownloadURL returns a presigned URL for downloading a blob - GetPresignedDownloadURL(digest string) (string, error) + // For ATProto blobs (CID), did is required for per-DID storage + // For OCI blobs (sha256:...), did may be empty + GetPresignedDownloadURL(digest, did string) (string, error) // GetPresignedUploadURL returns a presigned URL for uploading a blob - GetPresignedUploadURL(digest string) (string, error) + // For ATProto blobs (CID), did is required for per-DID storage + // For OCI blobs (sha256:...), did may be empty + GetPresignedUploadURL(digest, did string) (string, error) + + // UploadBlob receives raw blob bytes, computes CID, and stores via distribution driver + // Used for standard ATProto blob uploads (profile pics, small media) + // Returns CID and size of stored blob + UploadBlob(ctx context.Context, did string, data io.Reader) (cid cid.Cid, size int64, err error) + + // Multipart upload operations (used for OCI container layers only) + // StartMultipartUpload initiates a multipart upload, returns uploadID and mode + StartMultipartUpload(ctx context.Context, digest string) (uploadID string, mode string, err error) + // GetPartUploadURL returns a presigned URL for uploading a specific part + GetPartUploadURL(ctx context.Context, uploadID string, partNumber int, did string) (url string, err error) + // CompleteMultipartUpload finalizes a multipart upload + CompleteMultipartUpload(ctx context.Context, uploadID string, parts []PartInfo) error + // AbortMultipartUpload cancels a multipart upload + AbortMultipartUpload(ctx context.Context, uploadID string) error + // HandleBufferedPartUpload handles uploading a part in buffered mode + HandleBufferedPartUpload(ctx context.Context, uploadID string, partNumber int, data []byte) (etag string, err error) +} + +// PartInfo represents a completed part in a multipart upload +type PartInfo struct { + PartNumber int `json:"partNumber"` + ETag string `json:"etag"` } // NewXRPCHandler creates a new XRPC handler @@ -669,44 +698,220 @@ func (h *XRPCHandler) HandleSubscribeRepos(w http.ResponseWriter, r *http.Reques }() } -// HandleUploadBlob wraps existing presigned upload URL logic +// HandleUploadBlob handles blob uploads with support for multipart operations +// Supports three modes: +// 1. Buffered part upload: PUT with X-Upload-Id and X-Part-Number headers +// 2. Multipart operations: POST with JSON body containing action field +// 3. Direct blob upload: POST with raw bytes (ATProto-compliant) func (h *XRPCHandler) HandleUploadBlob(w http.ResponseWriter, r *http.Request) { + contentType := r.Header.Get("Content-Type") + + // Mode 1: Buffered part upload (PUT with headers) + if r.Method == http.MethodPut { + uploadID := r.Header.Get("X-Upload-Id") + partNumberStr := r.Header.Get("X-Part-Number") + + if uploadID != "" && partNumberStr != "" { + h.handleBufferedPartUpload(w, r, uploadID, partNumberStr) + return + } + http.Error(w, "PUT requires X-Upload-Id and X-Part-Number headers", http.StatusBadRequest) + return + } + + // Ensure POST method for remaining modes if r.Method != http.MethodPost { http.Error(w, "method not allowed", http.StatusMethodNotAllowed) return } + // Mode 2: Multipart operations (JSON body with action field) + if strings.Contains(contentType, "application/json") { + h.handleMultipartOperation(w, r) + return + } + + // Mode 3: Direct blob upload (ATProto-compliant) + // Receives raw bytes, computes CID, stores via distribution driver // TODO: Authentication check - // Read digest from query or calculate from body - digest := r.URL.Query().Get("digest") - if digest == "" { - http.Error(w, "digest required", http.StatusBadRequest) - return + // Extract DID for ATProto blob storage (per-DID paths) + did := r.URL.Query().Get("did") + if did == "" { + // TODO: Extract from auth context when authentication is implemented + // For now, use hold's DID as fallback + did = h.pds.DID() } - // Get presigned upload URL from existing blob store - uploadURL, err := h.blobStore.GetPresignedUploadURL(digest) + // Upload blob directly - blobStore will compute CID and store + blobCID, size, err := h.blobStore.UploadBlob(r.Context(), did, r.Body) if err != nil { - http.Error(w, fmt.Sprintf("failed to get upload URL: %v", err), http.StatusInternalServerError) + http.Error(w, fmt.Sprintf("failed to upload blob: %v", err), http.StatusInternalServerError) return } - // Return 302 redirect to presigned URL - http.Redirect(w, r, uploadURL, http.StatusFound) + // 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) +} + +// handleBufferedPartUpload handles uploading a part in buffered mode +func (h *XRPCHandler) handleBufferedPartUpload(w http.ResponseWriter, r *http.Request, uploadID, partNumberStr string) { + ctx := r.Context() + + // Parse part number + partNumber, err := strconv.Atoi(partNumberStr) + if err != nil { + http.Error(w, fmt.Sprintf("invalid part number: %v", err), http.StatusBadRequest) + return + } + + // Read part data from body + data, err := io.ReadAll(r.Body) + if err != nil { + http.Error(w, fmt.Sprintf("failed to read part data: %v", err), http.StatusInternalServerError) + return + } + + // Store part via blob store + etag, err := h.blobStore.HandleBufferedPartUpload(ctx, uploadID, partNumber, data) + if err != nil { + http.Error(w, fmt.Sprintf("failed to upload part: %v", err), http.StatusInternalServerError) + return + } + + // Return ETag in response + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]any{ + "etag": etag, + }) +} + +// handleMultipartOperation handles multipart upload operations via JSON request +func (h *XRPCHandler) handleMultipartOperation(w http.ResponseWriter, r *http.Request) { + ctx := r.Context() + + // Parse JSON body + var req struct { + Action string `json:"action"` + Digest string `json:"digest,omitempty"` + UploadID string `json:"uploadId,omitempty"` + PartNumber int `json:"partNumber,omitempty"` + Parts []PartInfo `json:"parts,omitempty"` + } + + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + http.Error(w, fmt.Sprintf("invalid JSON body: %v", err), http.StatusBadRequest) + return + } + + // Route based on action + switch req.Action { + case "start": + // Start multipart upload + if req.Digest == "" { + http.Error(w, "digest required for start action", http.StatusBadRequest) + return + } + + uploadID, mode, err := h.blobStore.StartMultipartUpload(ctx, req.Digest) + if err != nil { + http.Error(w, fmt.Sprintf("failed to start multipart upload: %v", err), http.StatusInternalServerError) + return + } + + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]any{ + "uploadId": uploadID, + "mode": mode, + }) + + case "part": + // Get part upload URL + if req.UploadID == "" || req.PartNumber == 0 { + http.Error(w, "uploadId and partNumber required for part action", http.StatusBadRequest) + return + } + + // Extract DID from query or header (for authorization) + did := r.URL.Query().Get("did") + if did == "" { + did = r.Header.Get("X-ATCR-DID") + } + + url, err := h.blobStore.GetPartUploadURL(ctx, req.UploadID, req.PartNumber, did) + if err != nil { + http.Error(w, fmt.Sprintf("failed to get part URL: %v", err), http.StatusInternalServerError) + return + } + + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]any{ + "url": url, + }) + + case "complete": + // Complete multipart upload + if req.UploadID == "" || len(req.Parts) == 0 { + http.Error(w, "uploadId and parts required for complete action", http.StatusBadRequest) + return + } + + if err := h.blobStore.CompleteMultipartUpload(ctx, req.UploadID, req.Parts); err != nil { + http.Error(w, fmt.Sprintf("failed to complete multipart upload: %v", err), http.StatusInternalServerError) + return + } + + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]any{ + "status": "completed", + }) + + case "abort": + // Abort multipart upload + if req.UploadID == "" { + http.Error(w, "uploadId required for abort action", http.StatusBadRequest) + return + } + + if err := h.blobStore.AbortMultipartUpload(ctx, req.UploadID); err != nil { + http.Error(w, fmt.Sprintf("failed to abort multipart upload: %v", err), http.StatusInternalServerError) + return + } + + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]any{ + "status": "aborted", + }) + + default: + http.Error(w, fmt.Sprintf("unknown action: %s", req.Action), http.StatusBadRequest) + } } // HandleGetBlob wraps existing presigned download URL logic +// Supports both ATProto CIDs and OCI sha256 digests func (h *XRPCHandler) HandleGetBlob(w http.ResponseWriter, r *http.Request) { - if r.Method != http.MethodGet { + if r.Method != http.MethodGet && r.Method != http.MethodHead { http.Error(w, "method not allowed", http.StatusMethodNotAllowed) return } did := r.URL.Query().Get("did") - digest := r.URL.Query().Get("cid") + cidOrDigest := r.URL.Query().Get("cid") - if did == "" || digest == "" { + if did == "" || cidOrDigest == "" { http.Error(w, "missing required parameters", http.StatusBadRequest) return } @@ -716,15 +921,28 @@ func (h *XRPCHandler) HandleGetBlob(w http.ResponseWriter, r *http.Request) { return } + // Flexible digest parsing: accept both CID and sha256 digest formats + var digest string + if strings.HasPrefix(cidOrDigest, "sha256:") { + // OCI digest format - use directly + digest = cidOrDigest + } else { + // Standard ATProto CID - for ATCR OCI use case, we expect sha256 digests + // If a real CID is provided, we could convert it here, but for now + // we'll just pass it through and let the blob store handle it + digest = cidOrDigest + } + // Get presigned download URL from existing blob store - downloadURL, err := h.blobStore.GetPresignedDownloadURL(digest) + // Pass DID for ATProto blob storage (per-DID paths) + downloadURL, err := h.blobStore.GetPresignedDownloadURL(digest, did) if err != nil { http.Error(w, fmt.Sprintf("failed to get download URL: %v", err), http.StatusInternalServerError) return } // Return 302 redirect to presigned URL - http.Redirect(w, r, downloadURL, http.StatusFound) + http.Redirect(w, r, downloadURL, http.StatusTemporaryRedirect) } // HandleListRepos lists all repositories in this PDS diff --git a/pkg/hold/pds/xrpc_multipart_test.go b/pkg/hold/pds/xrpc_multipart_test.go new file mode 100644 index 0000000..93396f0 --- /dev/null +++ b/pkg/hold/pds/xrpc_multipart_test.go @@ -0,0 +1,427 @@ +package pds + +import ( + "bytes" + "net/http" + "net/http/httptest" + "testing" +) + +// ATCR-Specific Tests: Non-standard multipart upload extensions +// +// This file contains tests for ATCR's custom multipart upload extensions +// to the ATProto blob endpoints. These are not part of the official ATProto spec. +// +// Standard ATProto blob tests are in xrpc_test.go + +// Tests for HandleUploadBlob - Multipart Start + +// TestHandleUploadBlob_MultipartStart tests multipart upload start operation +// Non-standard ATCR extension for large blob uploads +func TestHandleUploadBlob_MultipartStart(t *testing.T) { + handler, blobStore, _ := setupTestXRPCHandlerWithBlobs(t) + + digest := "sha256:largefile123" + body := map[string]string{ + "action": "start", + "digest": digest, + } + + req := makeXRPCPostRequest("/xrpc/com.atproto.repo.uploadBlob", body) + w := httptest.NewRecorder() + + handler.HandleUploadBlob(w, req) + + if w.Code != http.StatusOK { + t.Errorf("Expected status 200, got %d", w.Code) + } + + // Verify response contains uploadId and mode + result := assertJSONResponse(t, w, http.StatusOK) + + if uploadID, ok := result["uploadId"].(string); !ok || uploadID == "" { + t.Error("Expected uploadId string in response") + } + + if mode, ok := result["mode"].(string); !ok || mode == "" { + t.Error("Expected mode string in response") + } + + // Verify blob store was called + if len(blobStore.startCalls) != 1 || blobStore.startCalls[0] != digest { + t.Errorf("Expected StartMultipartUpload to be called with %s", digest) + } +} + +// TestHandleUploadBlob_MultipartStart_MissingDigest tests missing digest in start operation +func TestHandleUploadBlob_MultipartStart_MissingDigest(t *testing.T) { + handler, _, _ := setupTestXRPCHandlerWithBlobs(t) + + body := map[string]string{ + "action": "start", + } + + req := makeXRPCPostRequest("/xrpc/com.atproto.repo.uploadBlob", body) + w := httptest.NewRecorder() + + handler.HandleUploadBlob(w, req) + + if w.Code != http.StatusBadRequest { + t.Errorf("Expected status 400, got %d", w.Code) + } +} + +// Tests for HandleUploadBlob - Multipart Part URL + +// TestHandleUploadBlob_MultipartPart tests getting presigned URL for a part +// Non-standard ATCR extension for multipart uploads +func TestHandleUploadBlob_MultipartPart(t *testing.T) { + handler, blobStore, _ := setupTestXRPCHandlerWithBlobs(t) + + uploadID := "test-upload-123" + partNumber := 1 + did := "did:plc:testuser" + + body := map[string]any{ + "action": "part", + "uploadId": uploadID, + "partNumber": partNumber, + } + + req := makeXRPCPostRequest("/xrpc/com.atproto.repo.uploadBlob?did="+did, body) + w := httptest.NewRecorder() + + handler.HandleUploadBlob(w, req) + + if w.Code != http.StatusOK { + t.Errorf("Expected status 200, got %d", w.Code) + } + + // Verify response contains URL + result := assertJSONResponse(t, w, http.StatusOK) + + if url, ok := result["url"].(string); !ok || url == "" { + t.Error("Expected url string in response") + } + + // Verify blob store was called + if len(blobStore.partURLCalls) != 1 { + t.Fatalf("Expected GetPartUploadURL to be called once") + } + call := blobStore.partURLCalls[0] + if call.uploadID != uploadID || call.partNumber != partNumber || call.did != did { + t.Errorf("Expected GetPartUploadURL(%s, %d, %s), got (%s, %d, %s)", + uploadID, partNumber, did, call.uploadID, call.partNumber, call.did) + } +} + +// TestHandleUploadBlob_MultipartPart_MissingParams tests missing parameters +func TestHandleUploadBlob_MultipartPart_MissingParams(t *testing.T) { + handler, _, _ := setupTestXRPCHandlerWithBlobs(t) + + tests := []struct { + name string + body map[string]any + }{ + { + name: "missing uploadId", + body: map[string]any{ + "action": "part", + "partNumber": 1, + }, + }, + { + name: "missing partNumber", + body: map[string]any{ + "action": "part", + "uploadId": "test-123", + }, + }, + { + name: "partNumber zero", + body: map[string]any{ + "action": "part", + "uploadId": "test-123", + "partNumber": 0, + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + req := makeXRPCPostRequest("/xrpc/com.atproto.repo.uploadBlob", tt.body) + w := httptest.NewRecorder() + + handler.HandleUploadBlob(w, req) + + if w.Code != http.StatusBadRequest { + t.Errorf("Expected status 400, got %d", w.Code) + } + }) + } +} + +// Tests for HandleUploadBlob - Multipart Complete + +// TestHandleUploadBlob_MultipartComplete tests completing a multipart upload +// Non-standard ATCR extension for multipart uploads +func TestHandleUploadBlob_MultipartComplete(t *testing.T) { + handler, blobStore, _ := setupTestXRPCHandlerWithBlobs(t) + + uploadID := "test-upload-123" + parts := []PartInfo{ + {PartNumber: 1, ETag: "etag1"}, + {PartNumber: 2, ETag: "etag2"}, + } + + body := map[string]any{ + "action": "complete", + "uploadId": uploadID, + "parts": parts, + } + + req := makeXRPCPostRequest("/xrpc/com.atproto.repo.uploadBlob", body) + w := httptest.NewRecorder() + + handler.HandleUploadBlob(w, req) + + if w.Code != http.StatusOK { + t.Errorf("Expected status 200, got %d", w.Code) + } + + // Verify response + result := assertJSONResponse(t, w, http.StatusOK) + + if status, ok := result["status"].(string); !ok || status != "completed" { + t.Errorf("Expected status='completed', got %v", result["status"]) + } + + // Verify blob store was called + if len(blobStore.completeCalls) != 1 || blobStore.completeCalls[0] != uploadID { + t.Errorf("Expected CompleteMultipartUpload to be called with %s", uploadID) + } +} + +// TestHandleUploadBlob_MultipartComplete_MissingParams tests missing parameters +func TestHandleUploadBlob_MultipartComplete_MissingParams(t *testing.T) { + handler, _, _ := setupTestXRPCHandlerWithBlobs(t) + + tests := []struct { + name string + body map[string]any + }{ + { + name: "missing uploadId", + body: map[string]any{ + "action": "complete", + "parts": []PartInfo{{PartNumber: 1, ETag: "etag1"}}, + }, + }, + { + name: "missing parts", + body: map[string]any{ + "action": "complete", + "uploadId": "test-123", + }, + }, + { + name: "empty parts array", + body: map[string]any{ + "action": "complete", + "uploadId": "test-123", + "parts": []PartInfo{}, + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + req := makeXRPCPostRequest("/xrpc/com.atproto.repo.uploadBlob", tt.body) + w := httptest.NewRecorder() + + handler.HandleUploadBlob(w, req) + + if w.Code != http.StatusBadRequest { + t.Errorf("Expected status 400, got %d", w.Code) + } + }) + } +} + +// Tests for HandleUploadBlob - Multipart Abort + +// TestHandleUploadBlob_MultipartAbort tests aborting a multipart upload +// Non-standard ATCR extension for multipart uploads +func TestHandleUploadBlob_MultipartAbort(t *testing.T) { + handler, blobStore, _ := setupTestXRPCHandlerWithBlobs(t) + + uploadID := "test-upload-123" + + body := map[string]string{ + "action": "abort", + "uploadId": uploadID, + } + + req := makeXRPCPostRequest("/xrpc/com.atproto.repo.uploadBlob", body) + w := httptest.NewRecorder() + + handler.HandleUploadBlob(w, req) + + if w.Code != http.StatusOK { + t.Errorf("Expected status 200, got %d", w.Code) + } + + // Verify response + result := assertJSONResponse(t, w, http.StatusOK) + + if status, ok := result["status"].(string); !ok || status != "aborted" { + t.Errorf("Expected status='aborted', got %v", result["status"]) + } + + // Verify blob store was called + if len(blobStore.abortCalls) != 1 || blobStore.abortCalls[0] != uploadID { + t.Errorf("Expected AbortMultipartUpload to be called with %s", uploadID) + } +} + +// TestHandleUploadBlob_MultipartAbort_MissingUploadID tests missing uploadId +func TestHandleUploadBlob_MultipartAbort_MissingUploadID(t *testing.T) { + handler, _, _ := setupTestXRPCHandlerWithBlobs(t) + + body := map[string]string{ + "action": "abort", + } + + req := makeXRPCPostRequest("/xrpc/com.atproto.repo.uploadBlob", body) + w := httptest.NewRecorder() + + handler.HandleUploadBlob(w, req) + + if w.Code != http.StatusBadRequest { + t.Errorf("Expected status 400, got %d", w.Code) + } +} + +// Tests for HandleUploadBlob - Buffered Part Upload + +// TestHandleUploadBlob_BufferedPartUpload tests uploading a part in buffered mode +// Non-standard ATCR extension for multipart uploads without S3 presigned URLs +func TestHandleUploadBlob_BufferedPartUpload(t *testing.T) { + handler, blobStore, _ := setupTestXRPCHandlerWithBlobs(t) + + uploadID := "test-upload-123" + partNumber := "1" + data := []byte("test data for part 1") + + req := httptest.NewRequest(http.MethodPut, "/xrpc/com.atproto.repo.uploadBlob", bytes.NewReader(data)) + req.Header.Set("X-Upload-Id", uploadID) + req.Header.Set("X-Part-Number", partNumber) + w := httptest.NewRecorder() + + handler.HandleUploadBlob(w, req) + + if w.Code != http.StatusOK { + t.Errorf("Expected status 200, got %d", w.Code) + } + + // Verify response contains ETag + result := assertJSONResponse(t, w, http.StatusOK) + + if etag, ok := result["etag"].(string); !ok || etag == "" { + t.Error("Expected etag string in response") + } + + // Verify blob store was called + if len(blobStore.partUploadCalls) != 1 { + t.Fatalf("Expected HandleBufferedPartUpload to be called once") + } + call := blobStore.partUploadCalls[0] + if call.uploadID != uploadID || call.partNumber != 1 || call.dataSize != len(data) { + t.Errorf("Expected HandleBufferedPartUpload(%s, 1, %d bytes), got (%s, %d, %d bytes)", + uploadID, len(data), call.uploadID, call.partNumber, call.dataSize) + } +} + +// TestHandleUploadBlob_BufferedPartUpload_MissingHeaders tests missing required headers +func TestHandleUploadBlob_BufferedPartUpload_MissingHeaders(t *testing.T) { + handler, _, _ := setupTestXRPCHandlerWithBlobs(t) + + tests := []struct { + name string + uploadID string + partNumber string + setUploadID bool + setPartNumber bool + }{ + { + name: "missing both headers", + setUploadID: false, + setPartNumber: false, + }, + { + name: "missing X-Part-Number", + uploadID: "test-123", + setUploadID: true, + setPartNumber: false, + }, + { + name: "missing X-Upload-Id", + partNumber: "1", + setUploadID: false, + setPartNumber: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + req := httptest.NewRequest(http.MethodPut, "/xrpc/com.atproto.repo.uploadBlob", bytes.NewReader([]byte("data"))) + if tt.setUploadID { + req.Header.Set("X-Upload-Id", tt.uploadID) + } + if tt.setPartNumber { + req.Header.Set("X-Part-Number", tt.partNumber) + } + w := httptest.NewRecorder() + + handler.HandleUploadBlob(w, req) + + if w.Code != http.StatusBadRequest { + t.Errorf("Expected status 400, got %d", w.Code) + } + }) + } +} + +// TestHandleUploadBlob_BufferedPartUpload_InvalidPartNumber tests invalid part number +func TestHandleUploadBlob_BufferedPartUpload_InvalidPartNumber(t *testing.T) { + handler, _, _ := setupTestXRPCHandlerWithBlobs(t) + + req := httptest.NewRequest(http.MethodPut, "/xrpc/com.atproto.repo.uploadBlob", bytes.NewReader([]byte("data"))) + req.Header.Set("X-Upload-Id", "test-123") + req.Header.Set("X-Part-Number", "not-a-number") + w := httptest.NewRecorder() + + handler.HandleUploadBlob(w, req) + + if w.Code != http.StatusBadRequest { + t.Errorf("Expected status 400 for invalid part number, got %d", w.Code) + } +} + +// TestHandleUploadBlob_UnknownAction tests unknown action value +func TestHandleUploadBlob_UnknownAction(t *testing.T) { + handler, _, _ := setupTestXRPCHandlerWithBlobs(t) + + body := map[string]string{ + "action": "invalid", + } + + req := makeXRPCPostRequest("/xrpc/com.atproto.repo.uploadBlob", body) + w := httptest.NewRecorder() + + handler.HandleUploadBlob(w, req) + + if w.Code != http.StatusBadRequest { + t.Errorf("Expected status 400 for unknown action, got %d", w.Code) + } +} diff --git a/pkg/hold/pds/xrpc_test.go b/pkg/hold/pds/xrpc_test.go index 55effb4..f89e199 100644 --- a/pkg/hold/pds/xrpc_test.go +++ b/pkg/hold/pds/xrpc_test.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "encoding/json" + "fmt" "io" "net/http" "net/http/httptest" @@ -13,6 +14,7 @@ import ( "testing" "atcr.io/pkg/atproto" + "github.com/ipfs/go-cid" ) // Test helpers @@ -1316,3 +1318,460 @@ func TestHandleAtprotoDID(t *testing.T) { t.Errorf("Expected DID string, got %s", body) } } + +// Mock BlobStore for testing blob endpoints + +// mockBlobStore implements BlobStore interface for testing +type mockBlobStore struct { + // Control behavior + downloadURLError error + uploadURLError error + uploadBlobError error + startError error + partURLError error + completeError error + abortError error + partUploadError error + + // Track calls + downloadCalls []string // Track digests requested for download + uploadCalls []string // Track digests requested for upload + uploadBlobCalls []uploadBlobCall // Track direct blob uploads + startCalls []string // Track digests for multipart start + partURLCalls []partURLCall + completeCalls []string + abortCalls []string + partUploadCalls []partUploadCall +} + +type uploadBlobCall struct { + did string + dataSize int +} + +type partURLCall struct { + uploadID string + partNumber int + did string +} + +type partUploadCall struct { + uploadID string + partNumber int + dataSize int +} + +func newMockBlobStore() *mockBlobStore { + return &mockBlobStore{ + downloadCalls: []string{}, + uploadCalls: []string{}, + uploadBlobCalls: []uploadBlobCall{}, + startCalls: []string{}, + partURLCalls: []partURLCall{}, + completeCalls: []string{}, + abortCalls: []string{}, + partUploadCalls: []partUploadCall{}, + } +} + +func (m *mockBlobStore) GetPresignedDownloadURL(digest, did string) (string, error) { + m.downloadCalls = append(m.downloadCalls, digest) + if m.downloadURLError != nil { + return "", m.downloadURLError + } + return "https://s3.example.com/download/" + digest, nil +} + +func (m *mockBlobStore) GetPresignedUploadURL(digest, did string) (string, error) { + m.uploadCalls = append(m.uploadCalls, digest) + if m.uploadURLError != nil { + return "", m.uploadURLError + } + return "https://s3.example.com/upload/" + digest, nil +} + +func (m *mockBlobStore) UploadBlob(ctx context.Context, did string, data io.Reader) (cid.Cid, int64, error) { + // Read data to get size + blobData, err := io.ReadAll(data) + if err != nil { + return cid.Undef, 0, err + } + + m.uploadBlobCalls = append(m.uploadBlobCalls, uploadBlobCall{ + did: did, + dataSize: len(blobData), + }) + + if m.uploadBlobError != nil { + return cid.Undef, 0, m.uploadBlobError + } + + // Return a test CID (just use a fixed one for testing) + testCID, _ := cid.Decode("bafkreihdwdcefgh4dqkjv67uzcmw7ojee6xedzdetojuzjevtenxquvyku") + return testCID, int64(len(blobData)), nil +} + +func (m *mockBlobStore) StartMultipartUpload(ctx context.Context, digest string) (string, string, error) { + m.startCalls = append(m.startCalls, digest) + if m.startError != nil { + return "", "", m.startError + } + return "test-upload-id", "s3native", nil +} + +func (m *mockBlobStore) GetPartUploadURL(ctx context.Context, uploadID string, partNumber int, did string) (string, error) { + m.partURLCalls = append(m.partURLCalls, partURLCall{uploadID, partNumber, did}) + if m.partURLError != nil { + return "", m.partURLError + } + return "https://s3.example.com/part/" + uploadID, nil +} + +func (m *mockBlobStore) CompleteMultipartUpload(ctx context.Context, uploadID string, parts []PartInfo) error { + m.completeCalls = append(m.completeCalls, uploadID) + if m.completeError != nil { + return m.completeError + } + return nil +} + +func (m *mockBlobStore) AbortMultipartUpload(ctx context.Context, uploadID string) error { + m.abortCalls = append(m.abortCalls, uploadID) + if m.abortError != nil { + return m.abortError + } + return nil +} + +func (m *mockBlobStore) HandleBufferedPartUpload(ctx context.Context, uploadID string, partNumber int, data []byte) (string, error) { + m.partUploadCalls = append(m.partUploadCalls, partUploadCall{uploadID, partNumber, len(data)}) + if m.partUploadError != nil { + return "", m.partUploadError + } + return "test-etag-" + uploadID, nil +} + +// setupTestXRPCHandlerWithBlobs creates handler with mock blob store +func setupTestXRPCHandlerWithBlobs(t *testing.T) (*XRPCHandler, *mockBlobStore, context.Context) { + t.Helper() + + ctx := context.Background() + tmpDir := t.TempDir() + + dbPath := filepath.Join(tmpDir, "pds.db") + keyPath := filepath.Join(tmpDir, "signing-key") + + pds, err := NewHoldPDS(ctx, "did:web:hold.example.com", "https://hold.example.com", dbPath, keyPath) + if err != nil { + t.Fatalf("Failed to create test PDS: %v", err) + } + + // Bootstrap with a test owner, suppressing stdout to avoid log spam + ownerDID := "did:plc:testowner123" + + // Redirect stdout to suppress bootstrap logging + oldStdout := os.Stdout + r, w, _ := os.Pipe() + os.Stdout = w + + err = pds.Bootstrap(ctx, ownerDID, true, false) + + // Restore stdout + w.Close() + os.Stdout = oldStdout + io.ReadAll(r) // Drain the pipe + + if err != nil { + t.Fatalf("Failed to bootstrap PDS: %v", err) + } + + // Create mock blob store + blobStore := newMockBlobStore() + + // Create XRPC handler with mock blob store + handler := NewXRPCHandler(pds, "https://hold.example.com", blobStore, nil) + + return handler, blobStore, ctx +} + +// Tests for HandleUploadBlob + +// TestHandleUploadBlob tests com.atproto.repo.uploadBlob with direct upload +// Spec: https://docs.bsky.app/docs/api/com-atproto-repo-upload-blob +func TestHandleUploadBlob(t *testing.T) { + handler, blobStore, _ := setupTestXRPCHandlerWithBlobs(t) + + // Test data - a simple text blob + blobData := []byte("Hello, ATProto!") + + // Test standard single blob upload (POST with raw bytes) + req := httptest.NewRequest(http.MethodPost, "/xrpc/com.atproto.repo.uploadBlob", bytes.NewReader(blobData)) + req.Header.Set("Content-Type", "application/octet-stream") + w := httptest.NewRecorder() + + handler.HandleUploadBlob(w, req) + + // Should return 200 OK with blob metadata + if w.Code != http.StatusOK { + t.Errorf("Expected status 200 OK, got %d", w.Code) + } + + // Verify response contains blob metadata + result := assertJSONResponse(t, w, http.StatusOK) + + blob, ok := result["blob"].(map[string]any) + if !ok { + t.Fatal("Expected blob object in response") + } + + if blobType, ok := blob["$type"].(string); !ok || blobType != "blob" { + t.Errorf("Expected $type='blob', got %v", blob["$type"]) + } + + ref, ok := blob["ref"].(map[string]any) + if !ok { + t.Fatal("Expected ref object in blob") + } + + if link, ok := ref["$link"].(string); !ok || link == "" { + t.Error("Expected $link (CID) in ref") + } + + if size, ok := blob["size"].(float64); !ok || int(size) != len(blobData) { + t.Errorf("Expected size=%d, got %v", len(blobData), blob["size"]) + } + + // Verify blob store was called + if len(blobStore.uploadBlobCalls) != 1 { + t.Errorf("Expected UploadBlob to be called once, got %d calls", len(blobStore.uploadBlobCalls)) + } + + if blobStore.uploadBlobCalls[0].dataSize != len(blobData) { + t.Errorf("Expected UploadBlob to receive %d bytes, got %d", len(blobData), blobStore.uploadBlobCalls[0].dataSize) + } +} + +// TestHandleUploadBlob_EmptyBody tests empty blob upload +// Spec: https://docs.bsky.app/docs/api/com-atproto-repo-upload-blob +func TestHandleUploadBlob_EmptyBody(t *testing.T) { + handler, blobStore, _ := setupTestXRPCHandlerWithBlobs(t) + + // Empty blob should succeed (edge case) + req := httptest.NewRequest(http.MethodPost, "/xrpc/com.atproto.repo.uploadBlob", bytes.NewReader([]byte{})) + req.Header.Set("Content-Type", "application/octet-stream") + w := httptest.NewRecorder() + + handler.HandleUploadBlob(w, req) + + // Should succeed with empty blob + if w.Code != http.StatusOK { + t.Errorf("Expected status 200, got %d", w.Code) + } + + // Verify blob store was called with 0 bytes + if len(blobStore.uploadBlobCalls) != 1 || blobStore.uploadBlobCalls[0].dataSize != 0 { + t.Errorf("Expected UploadBlob with 0 bytes") + } +} + +// TestHandleUploadBlob_MethodNotAllowed tests wrong HTTP method +// Spec: https://docs.bsky.app/docs/api/com-atproto-repo-upload-blob +func TestHandleUploadBlob_MethodNotAllowed(t *testing.T) { + handler, _, _ := setupTestXRPCHandlerWithBlobs(t) + + // GET is not allowed for upload (only POST and PUT) + req := httptest.NewRequest(http.MethodGet, "/xrpc/com.atproto.repo.uploadBlob", bytes.NewReader([]byte("test"))) + w := httptest.NewRecorder() + + handler.HandleUploadBlob(w, req) + + if w.Code != http.StatusMethodNotAllowed { + t.Errorf("Expected status 405, got %d", w.Code) + } +} + +// TestHandleUploadBlob_BlobStoreError tests blob store returning error +// Spec: https://docs.bsky.app/docs/api/com-atproto-repo-upload-blob +func TestHandleUploadBlob_BlobStoreError(t *testing.T) { + handler, blobStore, _ := setupTestXRPCHandlerWithBlobs(t) + + // Configure mock to return error + blobStore.uploadBlobError = fmt.Errorf("storage driver unavailable") + + req := httptest.NewRequest(http.MethodPost, "/xrpc/com.atproto.repo.uploadBlob", bytes.NewReader([]byte("test data"))) + req.Header.Set("Content-Type", "application/octet-stream") + w := httptest.NewRecorder() + + handler.HandleUploadBlob(w, req) + + if w.Code != http.StatusInternalServerError { + t.Errorf("Expected status 500, got %d", w.Code) + } +} + +// Tests for HandleGetBlob + +// TestHandleGetBlob tests com.atproto.sync.getBlob +// Spec: https://docs.bsky.app/docs/api/com-atproto-sync-get-blob +func TestHandleGetBlob(t *testing.T) { + handler, blobStore, _ := setupTestXRPCHandlerWithBlobs(t) + + holdDID := "did:web:hold.example.com" + cid := "bafyreib2rxk3rkhh5ylyxj3x3gathxt3s32qvwj2lf3qg4kmzr6b7teqke" + + req := makeXRPCGetRequest("/xrpc/com.atproto.sync.getBlob", map[string]string{ + "did": holdDID, + "cid": cid, + }) + w := httptest.NewRecorder() + + handler.HandleGetBlob(w, req) + + // Should redirect to presigned download URL (307 Temporary Redirect) + if w.Code != http.StatusTemporaryRedirect { + t.Errorf("Expected status 307 (Temporary Redirect), got %d", w.Code) + } + + location := w.Header().Get("Location") + expectedURL := "https://s3.example.com/download/" + cid + if location != expectedURL { + t.Errorf("Expected redirect to %s, got %s", expectedURL, location) + } + + // Verify blob store was called + if len(blobStore.downloadCalls) != 1 || blobStore.downloadCalls[0] != cid { + t.Errorf("Expected GetPresignedDownloadURL to be called with %s", cid) + } +} + +// TestHandleGetBlob_SHA256Digest tests getBlob with OCI sha256 digest format +// Spec: https://docs.bsky.app/docs/api/com-atproto-sync-get-blob +func TestHandleGetBlob_SHA256Digest(t *testing.T) { + handler, blobStore, _ := setupTestXRPCHandlerWithBlobs(t) + + holdDID := "did:web:hold.example.com" + digest := "sha256:abc123def456" // OCI digest format + + req := makeXRPCGetRequest("/xrpc/com.atproto.sync.getBlob", map[string]string{ + "did": holdDID, + "cid": digest, + }) + w := httptest.NewRecorder() + + handler.HandleGetBlob(w, req) + + // Should redirect to presigned download URL + if w.Code != http.StatusTemporaryRedirect { + t.Errorf("Expected status 307, got %d", w.Code) + } + + // Verify blob store received the sha256 digest + if len(blobStore.downloadCalls) != 1 || blobStore.downloadCalls[0] != digest { + t.Errorf("Expected GetPresignedDownloadURL to be called with %s, got %v", digest, blobStore.downloadCalls) + } +} + +// TestHandleGetBlob_HeadMethod tests HEAD request support +// Spec: https://docs.bsky.app/docs/api/com-atproto-sync-get-blob +func TestHandleGetBlob_HeadMethod(t *testing.T) { + handler, blobStore, _ := setupTestXRPCHandlerWithBlobs(t) + + holdDID := "did:web:hold.example.com" + cid := "bafyreib2rxk3rkhh5ylyxj3x3gathxt3s32qvwj2lf3qg4kmzr6b7teqke" + + // Use HEAD instead of GET + req := httptest.NewRequest(http.MethodHead, "/xrpc/com.atproto.sync.getBlob?did="+holdDID+"&cid="+cid, nil) + w := httptest.NewRecorder() + + handler.HandleGetBlob(w, req) + + // Should still redirect + if w.Code != http.StatusTemporaryRedirect { + t.Errorf("Expected status 307 for HEAD request, got %d", w.Code) + } + + // Verify blob store was called + if len(blobStore.downloadCalls) != 1 { + t.Errorf("Expected GetPresignedDownloadURL to be called for HEAD request") + } +} + +// TestHandleGetBlob_MissingParameters tests missing required parameters +// Spec: https://docs.bsky.app/docs/api/com-atproto-sync-get-blob +func TestHandleGetBlob_MissingParameters(t *testing.T) { + handler, _, _ := setupTestXRPCHandlerWithBlobs(t) + + tests := []struct { + name string + params map[string]string + }{ + { + name: "missing all params", + params: map[string]string{}, + }, + { + name: "missing cid", + params: map[string]string{ + "did": "did:web:hold.example.com", + }, + }, + { + name: "missing did", + params: map[string]string{ + "cid": "bafyreib2rxk3rkhh5ylyxj3x3gathxt3s32qvwj2lf3qg4kmzr6b7teqke", + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + req := makeXRPCGetRequest("/xrpc/com.atproto.sync.getBlob", tt.params) + w := httptest.NewRecorder() + + handler.HandleGetBlob(w, req) + + if w.Code != http.StatusBadRequest { + t.Errorf("Expected status 400, got %d", w.Code) + } + }) + } +} + +// TestHandleGetBlob_InvalidDID tests invalid DID +// Spec: https://docs.bsky.app/docs/api/com-atproto-sync-get-blob +func TestHandleGetBlob_InvalidDID(t *testing.T) { + handler, _, _ := setupTestXRPCHandlerWithBlobs(t) + + req := makeXRPCGetRequest("/xrpc/com.atproto.sync.getBlob", map[string]string{ + "did": "did:plc:wrongdid", + "cid": "bafyreib2rxk3rkhh5ylyxj3x3gathxt3s32qvwj2lf3qg4kmzr6b7teqke", + }) + w := httptest.NewRecorder() + + handler.HandleGetBlob(w, req) + + if w.Code != http.StatusBadRequest { + t.Errorf("Expected status 400 for invalid DID, got %d", w.Code) + } +} + +// TestHandleGetBlob_BlobStoreError tests blob store returning error +// Spec: https://docs.bsky.app/docs/api/com-atproto-sync-get-blob +func TestHandleGetBlob_BlobStoreError(t *testing.T) { + handler, blobStore, _ := setupTestXRPCHandlerWithBlobs(t) + + // Configure mock to return error + blobStore.downloadURLError = fmt.Errorf("blob not found in S3") + + req := makeXRPCGetRequest("/xrpc/com.atproto.sync.getBlob", map[string]string{ + "did": "did:web:hold.example.com", + "cid": "bafyreib2rxk3rkhh5ylyxj3x3gathxt3s32qvwj2lf3qg4kmzr6b7teqke", + }) + w := httptest.NewRecorder() + + handler.HandleGetBlob(w, req) + + if w.Code != http.StatusInternalServerError { + t.Errorf("Expected status 500, got %d", w.Code) + } +} diff --git a/pkg/hold/storage.go b/pkg/hold/storage.go index 5a407f3..312d1a6 100644 --- a/pkg/hold/storage.go +++ b/pkg/hold/storage.go @@ -11,10 +11,20 @@ import ( "github.com/aws/aws-sdk-go/service/s3" ) +// 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) +} + // blobPath converts a digest (e.g., "sha256:abc123...") or temp path to a storage path // Distribution stores blobs as: /docker/registry/v2/blobs/{algorithm}/{xx}/{hash}/data // where xx is the first 2 characters of the hash for directory sharding // NOTE: Path must start with / for filesystem driver +// This is used for OCI container layers (content-addressed, globally deduplicated) func blobPath(digest string) string { // Handle temp paths (start with uploads/temp-) if strings.HasPrefix(digest, "uploads/temp-") { @@ -40,8 +50,23 @@ func blobPath(digest string) string { } // getPresignedURL generates a presigned URL for GET, HEAD, or PUT operations +// Distinguishes between ATProto blobs (per-DID) and OCI blobs (content-addressed) func (s *HoldService) getPresignedURL(ctx context.Context, operation PresignedURLOperation, digest string, did string) (string, error) { - path := blobPath(digest) + 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 = 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) + } // Check blob exists for GET/HEAD operations (not for PUT since blob doesn't exist yet) if operation == OperationGet || operation == OperationHead {