mirrors/at-container-registry
1
0
Fork 0
mirror of https://tangled.org/evan.jarrett.net/at-container-registry synced 2026-09-03 08:46:57 +00:00
Code Issues Packages Projects Releases Wiki Activity

combine s3 into multipart

Browse Source
This commit is contained in:
Evan Jarrett
2025-10-19 09:14:47 -05:00
parent 90ef4e90e5
commit fa931aca3b
10 changed files with 591 additions and 702 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/hold/main.go
+6 -9
View File
@@ -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
pkg/appview/storage/proxy_blob_store.go
-1
View File
@@ -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
pkg/hold/blobstore_adapter.go
-198
View File
@@ -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
}
pkg/hold/multipart.go
+253 -63
View File
@@ -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-<uuid>")
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)
}
pkg/hold/pds/xrpc.go
+15 -119
View File
@@ -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)
pkg/hold/pds/xrpc_multipart_test.go
+12 -12
View File
@@ -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)
pkg/hold/pds/xrpc_test.go
+143 -39
View File
@@ -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",
pkg/hold/s3.go
-221
View File
@@ -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
}
pkg/hold/service.go
+155 -34
View File
@@ -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: &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 = &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
}
pkg/hold/storage.go
+7 -6
View File
@@ -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",