From fa931aca3b315ad49968963ee6d0efd49765da53 Mon Sep 17 00:00:00 2001 From: Evan Jarrett Date: Sun, 19 Oct 2025 09:14:47 -0500 Subject: [PATCH] combine s3 into multipart --- cmd/hold/main.go | 15 +- pkg/appview/storage/proxy_blob_store.go | 1 - pkg/hold/blobstore_adapter.go | 198 --------------- pkg/hold/multipart.go | 316 +++++++++++++++++++----- pkg/hold/pds/xrpc.go | 134 ++-------- pkg/hold/pds/xrpc_multipart_test.go | 24 +- pkg/hold/pds/xrpc_test.go | 182 +++++++++++--- pkg/hold/s3.go | 221 ----------------- pkg/hold/service.go | 189 +++++++++++--- pkg/hold/storage.go | 13 +- 10 files changed, 591 insertions(+), 702 deletions(-) delete mode 100644 pkg/hold/blobstore_adapter.go delete mode 100644 pkg/hold/s3.go diff --git a/cmd/hold/main.go b/cmd/hold/main.go index b96f2ed..05d2d33 100644 --- a/cmd/hold/main.go +++ b/cmd/hold/main.go @@ -54,17 +54,14 @@ func main() { log.Fatalf("Database path is required for embedded PDS authorization") } - // Create hold service with PDS - service, err := hold.NewHoldService(cfg, holdPDS) - if err != nil { - log.Fatalf("Failed to create hold service: %v", err) - } - // Create blob store adapter and XRPC handler if holdPDS != nil { - holdDID := holdPDS.DID() - blobStore := hold.NewHoldServiceBlobStore(service, holdDID) - xrpcHandler = pds.NewXRPCHandler(holdPDS, cfg.Server.PublicURL, blobStore, broadcaster, nil) + // Create hold service with PDS + service, err := hold.NewHoldService(cfg, holdPDS) + if err != nil { + log.Fatalf("Failed to create hold service: %v", err) + } + xrpcHandler = pds.NewXRPCHandler(holdPDS, cfg.Server.PublicURL, service, broadcaster, nil) } // Setup HTTP routes diff --git a/pkg/appview/storage/proxy_blob_store.go b/pkg/appview/storage/proxy_blob_store.go index 5909c06..a506ef3 100644 --- a/pkg/appview/storage/proxy_blob_store.go +++ b/pkg/appview/storage/proxy_blob_store.go @@ -528,7 +528,6 @@ func (p *ProxyBlobStore) startMultipartUpload(ctx context.Context, digest string var result struct { UploadID string `json:"uploadId"` - Mode string `json:"mode"` } if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { return "", err diff --git a/pkg/hold/blobstore_adapter.go b/pkg/hold/blobstore_adapter.go deleted file mode 100644 index 5bc8860..0000000 --- a/pkg/hold/blobstore_adapter.go +++ /dev/null @@ -1,198 +0,0 @@ -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 -type HoldServiceBlobStore struct { - service *HoldService - holdDID string -} - -// NewHoldServiceBlobStore creates a blob store adapter for the hold service -func NewHoldServiceBlobStore(service *HoldService, holdDID string) pds.BlobStore { - return &HoldServiceBlobStore{ - service: service, - holdDID: holdDID, - } -} - -// GetPresignedURL returns a presigned URL for the specified operation (GET, HEAD, or PUT) -func (b *HoldServiceBlobStore) GetPresignedURL(operation string, 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() - // Cast operation string to PresignedURLOperation type - url, err := b.service.GetPresignedURL(ctx, PresignedURLOperation(operation), 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 structured upload info for uploading a specific part -func (b *HoldServiceBlobStore) GetPartUploadURL(ctx context.Context, uploadID string, partNumber int, did string) (*pds.PartUploadInfo, error) { - session, err := b.service.MultipartMgr.GetSession(uploadID) - if err != nil { - return nil, err - } - - // For S3Native mode: return presigned URL - if session.Mode == S3Native { - url, err := b.service.GetPartUploadURL(ctx, session, partNumber, did) - if err != nil { - return nil, err - } - return &pds.PartUploadInfo{ - URL: url, - Method: "PUT", - }, nil - } - - // Buffered mode: return XRPC endpoint with headers - return &pds.PartUploadInfo{ - URL: fmt.Sprintf("%s/xrpc/com.atproto.repo.uploadBlob", b.service.config.Server.PublicURL), - Method: "PUT", - Headers: map[string]string{ - "X-Upload-Id": uploadID, - "X-Part-Number": fmt.Sprintf("%d", partNumber), - }, - }, nil -} - -// CompleteMultipartUpload finalizes a multipart upload and moves to final digest location -// finalDigest is the real digest (e.g., "sha256:abc123...") for the final storage location -func (b *HoldServiceBlobStore) CompleteMultipartUpload(ctx context.Context, uploadID string, finalDigest 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, finalDigest) -} - -// 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/multipart.go b/pkg/hold/multipart.go index b182169..d9c2776 100644 --- a/pkg/hold/multipart.go +++ b/pkg/hold/multipart.go @@ -4,11 +4,16 @@ import ( "context" "crypto/sha256" "encoding/hex" + "encoding/json" "fmt" "log" + "net/http" + "sort" + "strings" "sync" "time" + "github.com/aws/aws-sdk-go/service/s3" "github.com/google/uuid" ) @@ -23,11 +28,19 @@ const ( ) // CompletedPart represents an uploaded part with its ETag -type CompletedPart struct { +type PartInfo struct { PartNumber int `json:"part_number"` 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 +} + // MultipartSession tracks an in-progress multipart upload type MultipartSession struct { UploadID string // Unique upload ID @@ -159,24 +172,6 @@ func (s *MultipartSession) StorePart(partNumber int, data []byte) string { return etag } -// RecordS3Part records a part uploaded to S3 (for S3Native mode) -func (s *MultipartSession) RecordS3Part(partNumber int, etag string, size int64) { - s.mu.Lock() - defer s.mu.Unlock() - - part := &MultipartPart{ - PartNumber: partNumber, - ETag: etag, - Size: size, - UploadedAt: time.Now(), - } - - s.Parts[partNumber] = part - s.LastActivity = time.Now() - - log.Printf("Recorded S3 part: uploadID=%s, part=%d, size=%d bytes, etag=%s", s.UploadID, partNumber, size, etag) -} - // AssembleBufferedParts assembles all buffered parts into a single blob // Returns the complete data and total size func (s *MultipartSession) AssembleBufferedParts() ([]byte, int64, error) { @@ -215,82 +210,150 @@ func (s *MultipartSession) AssembleBufferedParts() ([]byte, int64, error) { return assembled, totalSize, nil } -// GetCompletedParts returns the list of completed parts for S3 multipart completion -func (s *MultipartSession) GetCompletedParts() []CompletedPart { - s.mu.RLock() - defer s.mu.RUnlock() - - parts := make([]CompletedPart, 0, len(s.Parts)) - for _, part := range s.Parts { - parts = append(parts, CompletedPart{ - PartNumber: part.PartNumber, - ETag: part.ETag, - }) - } - - return parts -} - // StartMultipartUploadWithManager initiates a multipart upload using the manager // Returns uploadID and mode -func (s *HoldService) StartMultipartUploadWithManager(ctx context.Context, digest string, manager *MultipartManager) (string, MultipartMode, error) { +func (s *HoldService) StartMultipartUploadWithManager(ctx context.Context, digest string) (string, MultipartMode, error) { // Check if presigned URLs are disabled for testing if s.config.Server.DisablePresignedURLs { log.Printf("Presigned URLs disabled (DISABLE_PRESIGNED_URLS=true), using buffered mode") - session := manager.CreateSession(digest, Buffered, "") + session := s.MultipartMgr.CreateSession(digest, Buffered, "") log.Printf("Started buffered multipart: uploadID=%s", session.UploadID) return session.UploadID, Buffered, nil } // Try S3 native multipart first if s.s3Client != nil { - s3UploadID, err := s.startMultipartUpload(ctx, digest) + if s.s3Client == nil { + return "", S3Native, fmt.Errorf("S3 not configured") + } + path := blobPath(digest) + s3Key := strings.TrimPrefix(path, "/") + if s.s3PathPrefix != "" { + s3Key = s.s3PathPrefix + "/" + s3Key + } + + result, err := s.s3Client.CreateMultipartUploadWithContext(ctx, &s3.CreateMultipartUploadInput{ + Bucket: &s.bucket, + Key: &s3Key, + }) if err == nil { + s3UploadID := *result.UploadId // S3 native multipart succeeded - session := manager.CreateSession(digest, S3Native, s3UploadID) - log.Printf("Started S3 native multipart: uploadID=%s, s3UploadID=%s", session.UploadID, s3UploadID) + session := s.MultipartMgr.CreateSession(digest, S3Native, s3UploadID) + log.Printf("Started S3 native multipart: digest=%s, uploadID=%s, s3UploadID=%s", digest, session.UploadID, s3UploadID) return session.UploadID, S3Native, nil } log.Printf("S3 native multipart failed, falling back to buffered mode: %v", err) } // Fallback to buffered mode - session := manager.CreateSession(digest, Buffered, "") + session := s.MultipartMgr.CreateSession(digest, Buffered, "") log.Printf("Started buffered multipart: uploadID=%s", session.UploadID) return session.UploadID, Buffered, nil } // GetPartUploadURL generates a presigned URL for uploading a part // Only used for S3Native mode - Buffered mode is handled by blobstore adapter -func (s *HoldService) GetPartUploadURL(ctx context.Context, session *MultipartSession, partNumber int, did string) (string, error) { - if session.Mode != S3Native { - return "", fmt.Errorf("GetPartUploadURL only supports S3Native mode") +func (s *HoldService) GetPartUploadURL(ctx context.Context, uploadID string, partNumber int, did string) (*PartUploadInfo, error) { + session, err := s.MultipartMgr.GetSession(uploadID) + if err != nil { + return nil, err } - // Generate S3 presigned URL for this part - url, err := s.getPartPresignedURL(ctx, session.Digest, session.S3UploadID, partNumber) - if err != nil { - return "", fmt.Errorf("failed to generate S3 part URL: %w", err) + // For S3Native mode: return presigned URL + if session.Mode == S3Native { + if s.s3Client == nil { + return nil, fmt.Errorf("S3 not configured") + } + + path := blobPath(session.Digest) + s3Key := strings.TrimPrefix(path, "/") + if s.s3PathPrefix != "" { + s3Key = s.s3PathPrefix + "/" + s3Key + } + pnum := int64(partNumber) + req, _ := s.s3Client.UploadPartRequest(&s3.UploadPartInput{ + Bucket: &s.bucket, + Key: &s3Key, + UploadId: &uploadID, + PartNumber: &pnum, + }) + + url, err := req.Presign(15 * time.Minute) + if err != nil { + return nil, err + } + + log.Printf("Generated part presigned URL: digest=%s, uploadID=%s, part=%d", session.Digest, uploadID, partNumber) + + return &PartUploadInfo{ + URL: url, + Method: "PUT", + }, nil } - return url, nil + + // Buffered mode: return XRPC endpoint with headers + return &PartUploadInfo{ + URL: fmt.Sprintf("%s/xrpc/com.atproto.repo.uploadBlob", s.config.Server.PublicURL), + Method: "PUT", + Headers: map[string]string{ + "X-Upload-Id": uploadID, + "X-Part-Number": fmt.Sprintf("%d", partNumber), + }, + }, nil } // CompleteMultipartUploadWithManager completes a multipart upload and moves to final location // finalDigest is the real digest (e.g., "sha256:abc123...") for the final storage location // session.Digest is the temp location (e.g., "uploads/temp-") -func (s *HoldService) CompleteMultipartUploadWithManager(ctx context.Context, session *MultipartSession, manager *MultipartManager, finalDigest string) error { - defer manager.DeleteSession(session.UploadID) +func (s *HoldService) CompleteMultipartUploadWithManager(ctx context.Context, uploadID string, finalDigest string, parts []PartInfo) error { + session, err := s.MultipartMgr.GetSession(uploadID) + defer s.MultipartMgr.DeleteSession(uploadID) + if err != nil { + return err + } if session.Mode == S3Native { - // Complete S3 multipart upload at temp location - parts := session.GetCompletedParts() - if err := s.completeMultipartUpload(ctx, session.Digest, session.S3UploadID, parts); err != nil { - return fmt.Errorf("failed to complete S3 multipart: %w", err) + if s.s3Client == nil { + return fmt.Errorf("S3 not configured") } - log.Printf("Completed S3 native multipart at temp location: uploadID=%s, parts=%d", session.UploadID, len(parts)) + + // Sort parts by part number (S3 requires ascending order) + sort.Slice(parts, func(i, j int) bool { + return parts[i].PartNumber < parts[j].PartNumber + }) + + // Convert to S3 CompletedPart format + // IMPORTANT: S3 requires ETags to be quoted in the CompleteMultipartUpload XML + s3Parts := make([]*s3.CompletedPart, len(parts)) + for i, p := range parts { + etag := normalizeETag(p.ETag) + pnum := int64(p.PartNumber) + s3Parts[i] = &s3.CompletedPart{ + PartNumber: &pnum, + ETag: &etag, + } + } + sourcePath := blobPath(session.Digest) + s3Key := strings.TrimPrefix(sourcePath, "/") + if s.s3PathPrefix != "" { + s3Key = s.s3PathPrefix + "/" + s3Key + } + + _, err = s.s3Client.CompleteMultipartUploadWithContext(ctx, &s3.CompleteMultipartUploadInput{ + Bucket: &s.bucket, + Key: &s3Key, + UploadId: &uploadID, + MultipartUpload: &s3.CompletedMultipartUpload{ + Parts: s3Parts, + }, + }) + if err != nil { + return fmt.Errorf("failed to complete multipart upload: digest=%s, uploadID=%s, err=%v", session.Digest, uploadID, err) + } + log.Printf("Completed S3 native multipart at temp location: digest=%s, uploadID=%s, parts=%d", session.Digest, session.UploadID, len(s3Parts)) // Verify the blob exists at temp location before moving - sourcePath := blobPath(session.Digest) destPath := blobPath(finalDigest) log.Printf("[DEBUG] About to move: source=%s, dest=%s", sourcePath, destPath) @@ -339,15 +402,33 @@ func (s *HoldService) CompleteMultipartUploadWithManager(ctx context.Context, se } // AbortMultipartUploadWithManager aborts a multipart upload -func (s *HoldService) AbortMultipartUploadWithManager(ctx context.Context, session *MultipartSession, manager *MultipartManager) error { - defer manager.DeleteSession(session.UploadID) +func (s *HoldService) AbortMultipartUploadWithManager(ctx context.Context, uploadID string) error { + session, err := s.MultipartMgr.GetSession(uploadID) + defer s.MultipartMgr.DeleteSession(uploadID) + if err != nil { + return err + } if session.Mode == S3Native { - // Abort S3 multipart upload - if err := s.abortMultipartUpload(ctx, session.Digest, session.S3UploadID); err != nil { - return fmt.Errorf("failed to abort S3 multipart: %w", err) + if s.s3Client == nil { + return fmt.Errorf("S3 not configured") } - log.Printf("Aborted S3 native multipart: uploadID=%s", session.UploadID) + path := blobPath(session.Digest) + s3Key := strings.TrimPrefix(path, "/") + if s.s3PathPrefix != "" { + s3Key = s.s3PathPrefix + "/" + s3Key + } + + _, err := s.s3Client.AbortMultipartUploadWithContext(ctx, &s3.AbortMultipartUploadInput{ + Bucket: &s.bucket, + Key: &s3Key, + UploadId: &uploadID, + }) + // Abort S3 multipart upload + if err != nil { + return fmt.Errorf("failed to abort multipart upload: digest=%s, uploadID=%s, err=%v", session.Digest, uploadID, err) + } + log.Printf("Aborted S3 native multipart: digest=%s, uploadID=%s", session.Digest, session.UploadID) return nil } @@ -355,3 +436,112 @@ func (s *HoldService) AbortMultipartUploadWithManager(ctx context.Context, sessi log.Printf("Aborted buffered multipart: uploadID=%s", session.UploadID) return nil } + +// handleMultipartOperation handles multipart upload operations via JSON request +func (s *HoldService) HandleMultipartOperation(w http.ResponseWriter, r *http.Request, did string) { + 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, _, err := s.StartMultipartUploadWithManager(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, + }) + + 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 + } + + uploadInfo, err := s.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(uploadInfo) + + 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 req.Digest == "" { + http.Error(w, "digest required for complete action", http.StatusBadRequest) + return + } + + // Pass the real digest so hold can move temp → final location + if err := s.CompleteMultipartUploadWithManager(ctx, req.UploadID, req.Digest, 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 := s.AbortMultipartUploadWithManager(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) + } +} + +// normalizeETag ensures an ETag has quotes (required by S3 CompleteMultipartUpload) +// S3 returns ETags with quotes, but HTTP clients may strip them +func normalizeETag(etag string) string { + // Already has quotes + if strings.HasPrefix(etag, "\"") && strings.HasSuffix(etag, "\"") { + return etag + } + // Add quotes + return fmt.Sprintf("\"%s\"", etag) +} diff --git a/pkg/hold/pds/xrpc.go b/pkg/hold/pds/xrpc.go index 4cced22..e7e2014 100644 --- a/pkg/hold/pds/xrpc.go +++ b/pkg/hold/pds/xrpc.go @@ -25,35 +25,28 @@ import ( type XRPCHandler struct { pds *HoldPDS publicURL string - blobStore BlobStore + holdService XRPCHoldService broadcaster *EventBroadcaster httpClient HTTPClient // For testing - allows injecting mock HTTP client } -// BlobStore interface wraps the existing hold service storage operations -type BlobStore interface { +// interface wraps the existing hold service storage operations +type XRPCHoldService interface { // GetPresignedURL returns a presigned URL for the specified operation // For ATProto blobs (CID), did is required for per-DID storage // For OCI blobs (sha256:...), did may be empty // operation can be "GET", "HEAD", or "PUT" - GetPresignedURL(operation string, digest, did string) (string, error) + GetPresignedURL(ctx context.Context, operation string, digest string, 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 structured upload info (URL + optional headers) for a specific part - GetPartUploadURL(ctx context.Context, uploadID string, partNumber int, did string) (*PartUploadInfo, error) - // CompleteMultipartUpload finalizes a multipart upload and moves to final digest location - // finalDigest is the real digest (e.g., "sha256:abc123...") for the final storage location - CompleteMultipartUpload(ctx context.Context, uploadID string, finalDigest string, parts []PartInfo) error - // AbortMultipartUpload cancels a multipart upload - AbortMultipartUpload(ctx context.Context, uploadID string) error - // HandleBufferedPartUpload handles uploading a part in buffered mode + // handles multipart upload operations via JSON request + HandleMultipartOperation(w http.ResponseWriter, r *http.Request, did string) + + // handles uploading a part in buffered mode HandleBufferedPartUpload(ctx context.Context, uploadID string, partNumber int, data []byte) (etag string, err error) } @@ -72,11 +65,11 @@ type PartUploadInfo struct { } // NewXRPCHandler creates a new XRPC handler -func NewXRPCHandler(pds *HoldPDS, publicURL string, blobStore BlobStore, broadcaster *EventBroadcaster, httpClient HTTPClient) *XRPCHandler { +func NewXRPCHandler(pds *HoldPDS, publicURL string, holdService XRPCHoldService, broadcaster *EventBroadcaster, httpClient HTTPClient) *XRPCHandler { return &XRPCHandler{ pds: pds, publicURL: publicURL, - blobStore: blobStore, + holdService: holdService, broadcaster: broadcaster, httpClient: httpClient, } @@ -751,8 +744,8 @@ func (h *XRPCHandler) HandleUploadBlob(w http.ResponseWriter, r *http.Request) { // Use authenticated user's DID for ATProto blob storage (per-DID paths) did := user.DID - // Upload blob directly - blobStore will compute CID and store - blobCID, size, err := h.blobStore.UploadBlob(r.Context(), did, r.Body) + // Upload blob directly - holdService will compute CID and store + blobCID, size, err := h.holdService.UploadBlob(r.Context(), did, r.Body) if err != nil { http.Error(w, fmt.Sprintf("failed to upload blob: %v", err), http.StatusInternalServerError) return @@ -801,7 +794,7 @@ func (h *XRPCHandler) handleBufferedPartUpload(w http.ResponseWriter, r *http.Re } // Store part via blob store - etag, err := h.blobStore.HandleBufferedPartUpload(ctx, uploadID, partNumber, data) + etag, err := h.holdService.HandleBufferedPartUpload(ctx, uploadID, partNumber, data) if err != nil { http.Error(w, fmt.Sprintf("failed to upload part: %v", err), http.StatusInternalServerError) return @@ -816,22 +809,6 @@ func (h *XRPCHandler) handleBufferedPartUpload(w http.ResponseWriter, r *http.Re // 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 - } - // Validate blob write access for all multipart operations // This checks DPoP + OAuth tokens and verifies user is captain or crew with blob:write permission user, err := ValidateBlobWriteAccess(r, h.pds, h.httpClient) @@ -840,85 +817,7 @@ func (h *XRPCHandler) handleMultipartOperation(w http.ResponseWriter, r *http.Re 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 - } - - uploadInfo, err := h.blobStore.GetPartUploadURL(ctx, req.UploadID, req.PartNumber, user.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(uploadInfo) - - 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 req.Digest == "" { - http.Error(w, "digest required for complete action", http.StatusBadRequest) - return - } - - // Pass the real digest so hold can move temp → final location - if err := h.blobStore.CompleteMultipartUpload(ctx, req.UploadID, req.Digest, 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) - } + h.holdService.HandleMultipartOperation(w, r, user.DID) } // HandleGetBlob wraps existing presigned download URL logic @@ -984,13 +883,10 @@ func (h *XRPCHandler) HandleGetBlob(w http.ResponseWriter, r *http.Request) { operation := r.URL.Query().Get("method") if operation == "" { operation = "GET" - if r.Method == http.MethodHead { - operation = "HEAD" - } } // Generate presigned URL for the operation - presignedURL, err := h.blobStore.GetPresignedURL(operation, digest, did) + presignedURL, err := h.holdService.GetPresignedURL(r.Context(), operation, digest, did) if err != nil { log.Printf("[HandleGetBlob] Failed to get presigned %s URL: digest=%s, did=%s, err=%v", operation, digest, did, err) http.Error(w, "failed to get presigned URL", http.StatusInternalServerError) diff --git a/pkg/hold/pds/xrpc_multipart_test.go b/pkg/hold/pds/xrpc_multipart_test.go index 0ae9dff..ded6dcc 100644 --- a/pkg/hold/pds/xrpc_multipart_test.go +++ b/pkg/hold/pds/xrpc_multipart_test.go @@ -31,7 +31,7 @@ func addTestDPoPAuth(t *testing.T, req *http.Request, did string) { // 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) + handler, holdService, _ := setupTestXRPCHandlerWithBlobs(t) digest := "sha256:largefile123" body := map[string]string{ @@ -62,7 +62,7 @@ func TestHandleUploadBlob_MultipartStart(t *testing.T) { } // Verify blob store was called - if len(blobStore.startCalls) != 1 || blobStore.startCalls[0] != digest { + if len(holdService.startCalls) != 1 || holdService.startCalls[0] != digest { t.Errorf("Expected StartMultipartUpload to be called with %s", digest) } } @@ -91,7 +91,7 @@ func TestHandleUploadBlob_MultipartStart_MissingDigest(t *testing.T) { // 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) + handler, holdService, _ := setupTestXRPCHandlerWithBlobs(t) uploadID := "test-upload-123" partNumber := 1 @@ -121,10 +121,10 @@ func TestHandleUploadBlob_MultipartPart(t *testing.T) { } // Verify blob store was called with authenticated user's DID - if len(blobStore.partURLCalls) != 1 { + if len(holdService.partURLCalls) != 1 { t.Fatalf("Expected GetPartUploadURL to be called once") } - call := blobStore.partURLCalls[0] + call := holdService.partURLCalls[0] if call.uploadID != uploadID || call.partNumber != partNumber || call.did != expectedDID { t.Errorf("Expected GetPartUploadURL(%s, %d, %s), got (%s, %d, %s)", uploadID, partNumber, expectedDID, call.uploadID, call.partNumber, call.did) @@ -183,7 +183,7 @@ func TestHandleUploadBlob_MultipartPart_MissingParams(t *testing.T) { // TestHandleUploadBlob_MultipartComplete tests completing a multipart upload // Non-standard ATCR extension for multipart uploads func TestHandleUploadBlob_MultipartComplete(t *testing.T) { - handler, blobStore, _ := setupTestXRPCHandlerWithBlobs(t) + handler, holdService, _ := setupTestXRPCHandlerWithBlobs(t) uploadID := "test-upload-123" parts := []PartInfo{ @@ -216,7 +216,7 @@ func TestHandleUploadBlob_MultipartComplete(t *testing.T) { } // Verify blob store was called - if len(blobStore.completeCalls) != 1 || blobStore.completeCalls[0] != uploadID { + if len(holdService.completeCalls) != 1 || holdService.completeCalls[0] != uploadID { t.Errorf("Expected CompleteMultipartUpload to be called with %s", uploadID) } } @@ -284,7 +284,7 @@ func TestHandleUploadBlob_MultipartComplete_MissingParams(t *testing.T) { // TestHandleUploadBlob_MultipartAbort tests aborting a multipart upload // Non-standard ATCR extension for multipart uploads func TestHandleUploadBlob_MultipartAbort(t *testing.T) { - handler, blobStore, _ := setupTestXRPCHandlerWithBlobs(t) + handler, holdService, _ := setupTestXRPCHandlerWithBlobs(t) uploadID := "test-upload-123" @@ -311,7 +311,7 @@ func TestHandleUploadBlob_MultipartAbort(t *testing.T) { } // Verify blob store was called - if len(blobStore.abortCalls) != 1 || blobStore.abortCalls[0] != uploadID { + if len(holdService.abortCalls) != 1 || holdService.abortCalls[0] != uploadID { t.Errorf("Expected AbortMultipartUpload to be called with %s", uploadID) } } @@ -340,7 +340,7 @@ func TestHandleUploadBlob_MultipartAbort_MissingUploadID(t *testing.T) { // 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) + handler, holdService, _ := setupTestXRPCHandlerWithBlobs(t) uploadID := "test-upload-123" partNumber := "1" @@ -366,10 +366,10 @@ func TestHandleUploadBlob_BufferedPartUpload(t *testing.T) { } // Verify blob store was called - if len(blobStore.partUploadCalls) != 1 { + if len(holdService.partUploadCalls) != 1 { t.Fatalf("Expected HandleBufferedPartUpload to be called once") } - call := blobStore.partUploadCalls[0] + call := holdService.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) diff --git a/pkg/hold/pds/xrpc_test.go b/pkg/hold/pds/xrpc_test.go index bfa698e..19746d3 100644 --- a/pkg/hold/pds/xrpc_test.go +++ b/pkg/hold/pds/xrpc_test.go @@ -1330,10 +1330,10 @@ func TestHandleAtprotoDID(t *testing.T) { } } -// Mock BlobStore for testing blob endpoints +// Mock HoldService for testing blob endpoints -// mockBlobStore implements BlobStore interface for testing -type mockBlobStore struct { +// mockHoldService implements XRPCHoldService interface for testing +type mockHoldService struct { // Control behavior downloadURLError error uploadURLError error @@ -1372,8 +1372,8 @@ type partUploadCall struct { dataSize int } -func newMockBlobStore() *mockBlobStore { - return &mockBlobStore{ +func newMockHoldService() *mockHoldService { + return &mockHoldService{ downloadCalls: []string{}, uploadCalls: []string{}, uploadBlobCalls: []uploadBlobCall{}, @@ -1385,7 +1385,7 @@ func newMockBlobStore() *mockBlobStore { } } -func (m *mockBlobStore) GetPresignedURL(operation, digest, did string) (string, error) { +func (m *mockHoldService) GetPresignedURL(ctx context.Context, operation, digest, did string) (string, error) { if operation == "GET" || operation == "HEAD" { // Both GET and HEAD are download operations, just different HTTP methods m.downloadCalls = append(m.downloadCalls, digest) @@ -1405,7 +1405,7 @@ func (m *mockBlobStore) GetPresignedURL(operation, digest, did string) (string, return "https://s3.example.com/upload/" + digest, nil } -func (m *mockBlobStore) UploadBlob(ctx context.Context, did string, data io.Reader) (cid.Cid, int64, error) { +func (m *mockHoldService) 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 { @@ -1426,15 +1426,15 @@ func (m *mockBlobStore) UploadBlob(ctx context.Context, did string, data io.Read return testCID, int64(len(blobData)), nil } -func (m *mockBlobStore) StartMultipartUpload(ctx context.Context, digest string) (string, string, error) { +func (m *mockHoldService) StartMultipartUploadWithManager(ctx context.Context, digest string) (string, int, error) { m.startCalls = append(m.startCalls, digest) if m.startError != nil { - return "", "", m.startError + return "", 0, m.startError } - return "test-upload-id", "s3native", nil + return "test-upload-id", 0, nil // Return 0 for S3Native mode } -func (m *mockBlobStore) GetPartUploadURL(ctx context.Context, uploadID string, partNumber int, did string) (*PartUploadInfo, error) { +func (m *mockHoldService) GetPartUploadURL(ctx context.Context, uploadID string, partNumber int, did string) (*PartUploadInfo, error) { m.partURLCalls = append(m.partURLCalls, partURLCall{uploadID, partNumber, did}) if m.partURLError != nil { return nil, m.partURLError @@ -1445,7 +1445,7 @@ func (m *mockBlobStore) GetPartUploadURL(ctx context.Context, uploadID string, p }, nil } -func (m *mockBlobStore) CompleteMultipartUpload(ctx context.Context, uploadID string, finalDigest string, parts []PartInfo) error { +func (m *mockHoldService) CompleteMultipartUploadWithManager(ctx context.Context, uploadID string, finalDigest string, parts []PartInfo) error { m.completeCalls = append(m.completeCalls, uploadID) if m.completeError != nil { return m.completeError @@ -1453,7 +1453,7 @@ func (m *mockBlobStore) CompleteMultipartUpload(ctx context.Context, uploadID st return nil } -func (m *mockBlobStore) AbortMultipartUpload(ctx context.Context, uploadID string) error { +func (m *mockHoldService) AbortMultipartUploadWithManager(ctx context.Context, uploadID string) error { m.abortCalls = append(m.abortCalls, uploadID) if m.abortError != nil { return m.abortError @@ -1461,7 +1461,7 @@ func (m *mockBlobStore) AbortMultipartUpload(ctx context.Context, uploadID strin return nil } -func (m *mockBlobStore) HandleBufferedPartUpload(ctx context.Context, uploadID string, partNumber int, data []byte) (string, error) { +func (m *mockHoldService) 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 @@ -1469,8 +1469,112 @@ func (m *mockBlobStore) HandleBufferedPartUpload(ctx context.Context, uploadID s return "test-etag-" + uploadID, nil } -// setupTestXRPCHandlerWithBlobs creates handler with mock blob store and mock PDS client -func setupTestXRPCHandlerWithBlobs(t *testing.T) (*XRPCHandler, *mockBlobStore, context.Context) { +func (m *mockHoldService) HandleMultipartOperation(w http.ResponseWriter, r *http.Request, did string) { + ctx := r.Context() + + // Parse JSON body (same as real implementation) + 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": + if req.Digest == "" { + http.Error(w, "digest required for start action", http.StatusBadRequest) + return + } + + uploadID, mode, err := m.StartMultipartUploadWithManager(ctx, req.Digest) + if err != nil { + http.Error(w, fmt.Sprintf("failed to start multipart upload: %v", err), http.StatusInternalServerError) + return + } + + // Convert mode to string + var modeStr string + switch mode { + case 0: + modeStr = "s3native" + case 1: + modeStr = "buffered" + default: + modeStr = "unknown" + } + + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]any{ + "uploadId": uploadID, + "mode": modeStr, + }) + + case "part": + if req.UploadID == "" || req.PartNumber == 0 { + http.Error(w, "uploadId and partNumber required for part action", http.StatusBadRequest) + return + } + + uploadInfo, err := m.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(uploadInfo) + + case "complete": + if req.UploadID == "" || len(req.Parts) == 0 { + http.Error(w, "uploadId and parts required for complete action", http.StatusBadRequest) + return + } + if req.Digest == "" { + http.Error(w, "digest required for complete action", http.StatusBadRequest) + return + } + + if err := m.CompleteMultipartUploadWithManager(ctx, req.UploadID, req.Digest, 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": + if req.UploadID == "" { + http.Error(w, "uploadId required for abort action", http.StatusBadRequest) + return + } + + if err := m.AbortMultipartUploadWithManager(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) + } +} + +// setupTestXRPCHandlerWithBlobs creates handler with mock hold service and mock PDS client +func setupTestXRPCHandlerWithBlobs(t *testing.T) (*XRPCHandler, *mockHoldService, context.Context) { t.Helper() ctx := context.Background() @@ -1503,16 +1607,16 @@ func setupTestXRPCHandlerWithBlobs(t *testing.T) (*XRPCHandler, *mockBlobStore, t.Fatalf("Failed to bootstrap PDS: %v", err) } - // Create mock blob store - blobStore := newMockBlobStore() + // Create mock hold service + holdService := newMockHoldService() // Create mock PDS client for DPoP validation mockClient := &mockPDSClient{} - // Create XRPC handler with mock blob store and mock HTTP client - handler := NewXRPCHandler(pds, "https://hold.example.com", blobStore, nil, mockClient) + // Create XRPC handler with mock hold service and mock HTTP client + handler := NewXRPCHandler(pds, "https://hold.example.com", holdService, nil, mockClient) - return handler, blobStore, ctx + return handler, holdService, ctx } // Tests for HandleUploadBlob @@ -1520,7 +1624,7 @@ func setupTestXRPCHandlerWithBlobs(t *testing.T) (*XRPCHandler, *mockBlobStore, // 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) + handler, holdService, _ := setupTestXRPCHandlerWithBlobs(t) // Test data - a simple text blob blobData := []byte("Hello, ATProto!") @@ -1574,19 +1678,19 @@ func TestHandleUploadBlob(t *testing.T) { } // 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 len(holdService.uploadBlobCalls) != 1 { + t.Errorf("Expected UploadBlob to be called once, got %d calls", len(holdService.uploadBlobCalls)) } - if blobStore.uploadBlobCalls[0].dataSize != len(blobData) { - t.Errorf("Expected UploadBlob to receive %d bytes, got %d", len(blobData), blobStore.uploadBlobCalls[0].dataSize) + if holdService.uploadBlobCalls[0].dataSize != len(blobData) { + t.Errorf("Expected UploadBlob to receive %d bytes, got %d", len(blobData), holdService.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) + handler, holdService, _ := setupTestXRPCHandlerWithBlobs(t) // Empty blob should succeed (edge case) req := httptest.NewRequest(http.MethodPost, "/xrpc/com.atproto.repo.uploadBlob", bytes.NewReader([]byte{})) @@ -1612,7 +1716,7 @@ func TestHandleUploadBlob_EmptyBody(t *testing.T) { } // Verify blob store was called with 0 bytes - if len(blobStore.uploadBlobCalls) != 1 || blobStore.uploadBlobCalls[0].dataSize != 0 { + if len(holdService.uploadBlobCalls) != 1 || holdService.uploadBlobCalls[0].dataSize != 0 { t.Errorf("Expected UploadBlob with 0 bytes") } } @@ -1636,10 +1740,10 @@ func TestHandleUploadBlob_MethodNotAllowed(t *testing.T) { // 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) + handler, holdService, _ := setupTestXRPCHandlerWithBlobs(t) // Configure mock to return error - blobStore.uploadBlobError = fmt.Errorf("storage driver unavailable") + holdService.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") @@ -1669,7 +1773,7 @@ func TestHandleUploadBlob_BlobStoreError(t *testing.T) { // 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) + handler, holdService, _ := setupTestXRPCHandlerWithBlobs(t) holdDID := "did:web:hold.example.com" cid := "bafyreib2rxk3rkhh5ylyxj3x3gathxt3s32qvwj2lf3qg4kmzr6b7teqke" @@ -1706,7 +1810,7 @@ func TestHandleGetBlob(t *testing.T) { } // Verify blob store was called - if len(blobStore.downloadCalls) != 1 || blobStore.downloadCalls[0] != cid { + if len(holdService.downloadCalls) != 1 || holdService.downloadCalls[0] != cid { t.Errorf("Expected GetPresignedURL to be called with %s", cid) } } @@ -1714,7 +1818,7 @@ func TestHandleGetBlob(t *testing.T) { // 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) + handler, holdService, _ := setupTestXRPCHandlerWithBlobs(t) holdDID := "did:web:hold.example.com" digest := "sha256:abc123def456" // OCI digest format @@ -1744,8 +1848,8 @@ func TestHandleGetBlob_SHA256Digest(t *testing.T) { } // Verify blob store received the sha256 digest - if len(blobStore.downloadCalls) != 1 || blobStore.downloadCalls[0] != digest { - t.Errorf("Expected GetPresignedURL to be called with %s, got %v", digest, blobStore.downloadCalls) + if len(holdService.downloadCalls) != 1 || holdService.downloadCalls[0] != digest { + t.Errorf("Expected GetPresignedURL to be called with %s, got %v", digest, holdService.downloadCalls) } } @@ -1754,7 +1858,7 @@ func TestHandleGetBlob_SHA256Digest(t *testing.T) { // AppView is responsible for making the actual HEAD request to S3 // Spec: https://docs.bsky.app/docs/api/com-atproto-sync-get-blob func TestHandleGetBlob_HeadMethod(t *testing.T) { - handler, blobStore, _ := setupTestXRPCHandlerWithBlobs(t) + handler, holdService, _ := setupTestXRPCHandlerWithBlobs(t) holdDID := "did:web:hold.example.com" cid := "bafyreib2rxk3rkhh5ylyxj3x3gathxt3s32qvwj2lf3qg4kmzr6b7teqke" @@ -1789,7 +1893,7 @@ func TestHandleGetBlob_HeadMethod(t *testing.T) { } // Verify blob store was called with HEAD operation - if len(blobStore.downloadCalls) != 1 || blobStore.downloadCalls[0] != cid { + if len(holdService.downloadCalls) != 1 || holdService.downloadCalls[0] != cid { t.Errorf("Expected GetPresignedURL to be called with %s", cid) } } @@ -1856,10 +1960,10 @@ func TestHandleGetBlob_InvalidDID(t *testing.T) { // 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) + handler, holdService, _ := setupTestXRPCHandlerWithBlobs(t) // Configure mock to return error - blobStore.downloadURLError = fmt.Errorf("blob not found in S3") + holdService.downloadURLError = fmt.Errorf("blob not found in S3") req := makeXRPCGetRequest("/xrpc/com.atproto.sync.getBlob", map[string]string{ "did": "did:web:hold.example.com", diff --git a/pkg/hold/s3.go b/pkg/hold/s3.go deleted file mode 100644 index 8885988..0000000 --- a/pkg/hold/s3.go +++ /dev/null @@ -1,221 +0,0 @@ -package hold - -import ( - "context" - "fmt" - "log" - "sort" - "strings" - "time" - - "github.com/aws/aws-sdk-go/aws" - "github.com/aws/aws-sdk-go/aws/credentials" - "github.com/aws/aws-sdk-go/aws/session" - "github.com/aws/aws-sdk-go/service/s3" -) - -// initS3Client initializes the S3 client for presigned URL generation -// Returns nil error if S3 client is successfully initialized -// Returns error if storage is not S3 or if initialization fails (service will fall back to proxy mode) -func (s *HoldService) initS3Client() error { - // Check if presigned URLs are explicitly disabled - if s.config.Server.DisablePresignedURLs { - log.Printf("⚠️ S3 presigned URLs DISABLED by config (DISABLE_PRESIGNED_URLS=true)") - log.Printf(" All uploads will use buffered mode (parts buffered in hold service)") - return nil // Not an error - just using buffered mode - } - - // Check if storage driver is S3 - if s.config.Storage.Type() != "s3" { - log.Printf("Storage driver is %s (not S3), presigned URLs disabled", s.config.Storage.Type()) - return nil // Not an error - just using different driver - } - - // Extract S3 configuration from storage parameters - params := s.config.Storage.Parameters() - - // Extract required S3 configuration - region, _ := params["region"].(string) - if region == "" { - region = "us-east-1" // Default region - } - - accessKey, _ := params["accesskey"].(string) - secretKey, _ := params["secretkey"].(string) - bucket, _ := params["bucket"].(string) - - if bucket == "" { - return fmt.Errorf("S3 bucket not configured") - } - - // Build AWS config - awsConfig := &aws.Config{ - Region: aws.String(region), - } - - // Add credentials if provided (allow IAM role auth if not provided) - if accessKey != "" && secretKey != "" { - awsConfig.Credentials = credentials.NewStaticCredentials(accessKey, secretKey, "") - } - - // Add custom endpoint for S3-compatible services (Storj, MinIO, R2, etc.) - if endpoint, ok := params["regionendpoint"].(string); ok && endpoint != "" { - awsConfig.Endpoint = aws.String(endpoint) - awsConfig.S3ForcePathStyle = aws.Bool(true) // Required for MinIO, Storj - } - - // Create AWS session - sess, err := session.NewSession(awsConfig) - if err != nil { - return fmt.Errorf("failed to create AWS session: %w", err) - } - - // Create S3 client - s.s3Client = s3.New(sess) - s.bucket = bucket - - // Extract path prefix if configured (rootdirectory in S3 params) - if rootDir, ok := params["rootdirectory"].(string); ok && rootDir != "" { - s.s3PathPrefix = strings.TrimPrefix(rootDir, "/") - } - - log.Printf("✅ S3 presigned URLs enabled") - - return nil -} - -// startMultipartUpload initiates a multipart upload and returns upload ID -func (s *HoldService) startMultipartUpload(ctx context.Context, digest string) (string, error) { - if s.s3Client == nil { - return "", fmt.Errorf("S3 not configured") - } - - path := blobPath(digest) - s3Key := strings.TrimPrefix(path, "/") - if s.s3PathPrefix != "" { - s3Key = s.s3PathPrefix + "/" + s3Key - } - - result, err := s.s3Client.CreateMultipartUploadWithContext(ctx, &s3.CreateMultipartUploadInput{ - Bucket: aws.String(s.bucket), - Key: aws.String(s3Key), - }) - if err != nil { - return "", err - } - - log.Printf("Started multipart upload: digest=%s, uploadID=%s", digest, *result.UploadId) - return *result.UploadId, nil -} - -// getPartPresignedURL generates presigned URL for a specific part -func (s *HoldService) getPartPresignedURL(ctx context.Context, digest, uploadID string, partNumber int) (string, error) { - if s.s3Client == nil { - return "", fmt.Errorf("S3 not configured") - } - - path := blobPath(digest) - s3Key := strings.TrimPrefix(path, "/") - if s.s3PathPrefix != "" { - s3Key = s.s3PathPrefix + "/" + s3Key - } - - req, _ := s.s3Client.UploadPartRequest(&s3.UploadPartInput{ - Bucket: aws.String(s.bucket), - Key: aws.String(s3Key), - UploadId: aws.String(uploadID), - PartNumber: aws.Int64(int64(partNumber)), - }) - - url, err := req.Presign(15 * time.Minute) - if err != nil { - return "", err - } - - log.Printf("Generated part presigned URL: digest=%s, uploadID=%s, part=%d", digest, uploadID, partNumber) - return url, nil -} - -// normalizeETag ensures an ETag has quotes (required by S3 CompleteMultipartUpload) -// S3 returns ETags with quotes, but HTTP clients may strip them -func normalizeETag(etag string) string { - // Already has quotes - if strings.HasPrefix(etag, "\"") && strings.HasSuffix(etag, "\"") { - return etag - } - // Add quotes - return fmt.Sprintf("\"%s\"", etag) -} - -// completeMultipartUpload finalizes the multipart upload -func (s *HoldService) completeMultipartUpload(ctx context.Context, digest, uploadID string, parts []CompletedPart) error { - if s.s3Client == nil { - return fmt.Errorf("S3 not configured") - } - - path := blobPath(digest) - s3Key := strings.TrimPrefix(path, "/") - if s.s3PathPrefix != "" { - s3Key = s.s3PathPrefix + "/" + s3Key - } - - // Sort parts by part number (S3 requires ascending order) - sort.Slice(parts, func(i, j int) bool { - return parts[i].PartNumber < parts[j].PartNumber - }) - - // Convert to S3 CompletedPart format - // IMPORTANT: S3 requires ETags to be quoted in the CompleteMultipartUpload XML - s3Parts := make([]*s3.CompletedPart, len(parts)) - for i, p := range parts { - etag := normalizeETag(p.ETag) - s3Parts[i] = &s3.CompletedPart{ - PartNumber: aws.Int64(int64(p.PartNumber)), - ETag: aws.String(etag), - } - } - - _, err := s.s3Client.CompleteMultipartUploadWithContext(ctx, &s3.CompleteMultipartUploadInput{ - Bucket: aws.String(s.bucket), - Key: aws.String(s3Key), - UploadId: aws.String(uploadID), - MultipartUpload: &s3.CompletedMultipartUpload{ - Parts: s3Parts, - }, - }) - - if err != nil { - log.Printf("Failed to complete multipart upload: digest=%s, uploadID=%s, err=%v", digest, uploadID, err) - return err - } - - log.Printf("Completed multipart upload: digest=%s, uploadID=%s, parts=%d", digest, uploadID, len(parts)) - return nil -} - -// abortMultipartUpload aborts an in-progress multipart upload -func (s *HoldService) abortMultipartUpload(ctx context.Context, digest, uploadID string) error { - if s.s3Client == nil { - return fmt.Errorf("S3 not configured") - } - - path := blobPath(digest) - s3Key := strings.TrimPrefix(path, "/") - if s.s3PathPrefix != "" { - s3Key = s.s3PathPrefix + "/" + s3Key - } - - _, err := s.s3Client.AbortMultipartUploadWithContext(ctx, &s3.AbortMultipartUploadInput{ - Bucket: aws.String(s.bucket), - Key: aws.String(s3Key), - UploadId: aws.String(uploadID), - }) - - if err != nil { - log.Printf("Failed to abort multipart upload: digest=%s, uploadID=%s, err=%v", digest, uploadID, err) - return err - } - - log.Printf("Aborted multipart upload: digest=%s, uploadID=%s", digest, uploadID) - return nil -} diff --git a/pkg/hold/service.go b/pkg/hold/service.go index 84840bd..f5623be 100644 --- a/pkg/hold/service.go +++ b/pkg/hold/service.go @@ -4,40 +4,33 @@ import ( "context" "fmt" "log" + "strings" - "atcr.io/pkg/auth" + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/aws/credentials" + "github.com/aws/aws-sdk-go/aws/session" "github.com/aws/aws-sdk-go/service/s3" storagedriver "github.com/distribution/distribution/v3/registry/storage/driver" "github.com/distribution/distribution/v3/registry/storage/driver/factory" -) -// HoldPDSInterface is the minimal interface needed from the embedded PDS -// This avoids a circular import between pkg/hold and pkg/hold/pds -type HoldPDSInterface interface { - DID() string -} + "bytes" + "crypto/sha256" + "io" + + "github.com/ipfs/go-cid" + "github.com/multiformats/go-multihash" +) // HoldService provides presigned URLs for blob storage in a hold type HoldService struct { driver storagedriver.StorageDriver config *Config - s3Client *s3.S3 // S3 client for presigned URLs (nil if not S3 storage) - bucket string // S3 bucket name - s3PathPrefix string // S3 path prefix (if any) - MultipartMgr *MultipartManager // Exported for access in route handlers - pds HoldPDSInterface // Embedded PDS for captain/crew records - authorizer auth.HoldAuthorizer // Authorizer for access control + s3Client *s3.S3 // S3 client for presigned URLs (nil if not S3 storage) + bucket string // S3 bucket name + s3PathPrefix string // S3 path prefix (if any) + MultipartMgr *MultipartManager // Exported for access in route handlers } -// PresignedURLOperation defines the type of presigned URL operation -type PresignedURLOperation string - -const ( - OperationGet PresignedURLOperation = "GET" - OperationHead PresignedURLOperation = "HEAD" - OperationPut PresignedURLOperation = "PUT" -) - // NewHoldService creates a new hold service // holdPDS must be a *pds.HoldPDS but we use any to avoid import cycle func NewHoldService(cfg *Config, holdPDS any) (*HoldService, error) { @@ -48,22 +41,10 @@ func NewHoldService(cfg *Config, holdPDS any) (*HoldService, error) { return nil, fmt.Errorf("failed to create storage driver: %w", err) } - // Create local authorizer using the embedded PDS - // This requires casting holdPDS to the concrete type expected by auth - authorizer := auth.NewLocalHoldAuthorizerFromInterface(holdPDS) - - // Cast to our interface for storage - pdsInterface, ok := holdPDS.(HoldPDSInterface) - if !ok { - return nil, fmt.Errorf("holdPDS must implement HoldPDSInterface") - } - service := &HoldService{ driver: driver, config: cfg, MultipartMgr: NewMultipartManager(), - pds: pdsInterface, - authorizer: authorizer, } // Initialize S3 client for presigned URLs (if using S3 storage) @@ -73,3 +54,143 @@ func NewHoldService(cfg *Config, holdPDS any) (*HoldService, error) { return service, 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 (h *HoldService) UploadBlob(ctx context.Context, did string, data io.Reader) (cid.Cid, int64, error) { + + // 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 := h.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 +} + +// HandleBufferedPartUpload handles uploading a part in buffered mode +func (h *HoldService) HandleBufferedPartUpload(ctx context.Context, uploadID string, partNumber int, data []byte) (string, error) { + session, err := h.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 +} + +// initS3Client initializes the S3 client for presigned URL generation +// Returns nil error if S3 client is successfully initialized +// Returns error if storage is not S3 or if initialization fails (service will fall back to proxy mode) +func (s *HoldService) initS3Client() error { + // Check if presigned URLs are explicitly disabled + if s.config.Server.DisablePresignedURLs { + log.Printf("⚠️ S3 presigned URLs DISABLED by config (DISABLE_PRESIGNED_URLS=true)") + log.Printf(" All uploads will use buffered mode (parts buffered in hold service)") + return nil // Not an error - just using buffered mode + } + + // Check if storage driver is S3 + if s.config.Storage.Type() != "s3" { + log.Printf("Storage driver is %s (not S3), presigned URLs disabled", s.config.Storage.Type()) + return nil // Not an error - just using different driver + } + + // Extract S3 configuration from storage parameters + params := s.config.Storage.Parameters() + + // Extract required S3 configuration + region, _ := params["region"].(string) + if region == "" { + region = "us-east-1" // Default region + } + + accessKey, _ := params["accesskey"].(string) + secretKey, _ := params["secretkey"].(string) + bucket, _ := params["bucket"].(string) + + if bucket == "" { + return fmt.Errorf("S3 bucket not configured") + } + + // Build AWS config + awsConfig := &aws.Config{ + Region: ®ion, + } + + // Add credentials if provided (allow IAM role auth if not provided) + if accessKey != "" && secretKey != "" { + awsConfig.Credentials = credentials.NewStaticCredentials(accessKey, secretKey, "") + } + + // Add custom endpoint for S3-compatible services (Storj, MinIO, R2, etc.) + if endpoint, ok := params["regionendpoint"].(string); ok && endpoint != "" { + awsConfig.Endpoint = &endpoint + awsConfig.S3ForcePathStyle = aws.Bool(true) // Required for MinIO, Storj + } + + // Create AWS session + sess, err := session.NewSession(awsConfig) + if err != nil { + return fmt.Errorf("failed to create AWS session: %w", err) + } + + // Create S3 client + s.s3Client = s3.New(sess) + s.bucket = bucket + + // Extract path prefix if configured (rootdirectory in S3 params) + if rootDir, ok := params["rootdirectory"].(string); ok && rootDir != "" { + s.s3PathPrefix = strings.TrimPrefix(rootDir, "/") + } + + log.Printf("✅ S3 presigned URLs enabled") + + return nil +} diff --git a/pkg/hold/storage.go b/pkg/hold/storage.go index 2ee3de8..e06c5a3 100644 --- a/pkg/hold/storage.go +++ b/pkg/hold/storage.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "log" + "net/http" "strings" "time" @@ -53,7 +54,7 @@ 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) { +func (s *HoldService) GetPresignedURL(ctx context.Context, operation string, digest string, did string) (string, error) { var path string // Determine blob type and construct appropriate path @@ -97,20 +98,20 @@ func (s *HoldService) GetPresignedURL(ctx context.Context, operation PresignedUR Presign(time.Duration) (string, error) } switch operation { - case OperationGet: + case http.MethodGet: // Note: Don't use ResponseContentType - not supported by all S3-compatible services req, _ = s.s3Client.GetObjectRequest(&s3.GetObjectInput{ Bucket: aws.String(s.bucket), Key: aws.String(s3Key), }) - case OperationHead: + case http.MethodHead: req, _ = s.s3Client.HeadObjectRequest(&s3.HeadObjectInput{ Bucket: aws.String(s.bucket), Key: aws.String(s3Key), }) - case OperationPut: + case http.MethodPut: req, _ = s.s3Client.PutObjectRequest(&s3.PutObjectInput{ Bucket: aws.String(s.bucket), Key: aws.String(s3Key), @@ -147,9 +148,9 @@ func (s *HoldService) GetPresignedURL(ctx context.Context, operation PresignedUR // 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 (s *HoldService) getProxyURL(digest, did string, operation PresignedURLOperation) string { +func (s *HoldService) getProxyURL(digest, did string, operation string) string { // For read operations, use XRPC getBlob endpoint - if operation == OperationGet || operation == OperationHead { + if operation == http.MethodGet || operation == http.MethodHead { // Generate hold DID from public URL using shared function holdDID := atproto.ResolveHoldDIDFromURL(s.config.Server.PublicURL) return fmt.Sprintf("%s/xrpc/com.atproto.sync.getBlob?did=%s&cid=%s",