package storage import ( "bytes" "context" "fmt" "io" "log" "strings" "github.com/minio/minio-go/v7" "github.com/minio/minio-go/v7/pkg/credentials" ) // ── S3 Store ────────────────────────────── // S3Config holds all configuration for the S3 backend. type S3Config struct { Endpoint string // e.g. "minio.example.com:9000" or "s3.amazonaws.com" Bucket string // bucket name Region string // AWS region (default "us-east-1") AccessKey string // access key ID SecretKey string // secret access key Prefix string // optional key prefix (e.g. "armature/") ForcePathStyle bool // path-style URLs (required for MinIO, most self-hosted) UseSSL bool // use HTTPS (derived from endpoint scheme) } // S3Store implements ObjectStore using any S3-compatible API. // Tested with: MinIO, Ceph RGW, AWS S3. type S3Store struct { client *minio.Client bucket string prefix string // prepended to all keys (e.g. "armature/") endpoint string // original endpoint for display in Stats } // NewS3 creates an S3Store connected to the given endpoint. // Returns an error if the bucket does not exist or is not accessible. func NewS3(cfg S3Config) (*S3Store, error) { if cfg.Bucket == "" { return nil, fmt.Errorf("storage/s3: bucket name is required") } if cfg.AccessKey == "" || cfg.SecretKey == "" { return nil, fmt.Errorf("storage/s3: access key and secret key are required") } if cfg.Region == "" { cfg.Region = "us-east-1" } // Parse endpoint: strip scheme if present, detect SSL endpoint := cfg.Endpoint useSSL := cfg.UseSSL if strings.HasPrefix(endpoint, "https://") { endpoint = strings.TrimPrefix(endpoint, "https://") useSSL = true } else if strings.HasPrefix(endpoint, "http://") { endpoint = strings.TrimPrefix(endpoint, "http://") useSSL = false } // Default to SSL for AWS endpoints if endpoint == "" || strings.Contains(endpoint, "amazonaws.com") { useSSL = true } opts := &minio.Options{ Creds: credentials.NewStaticV4(cfg.AccessKey, cfg.SecretKey, ""), Secure: useSSL, Region: cfg.Region, } // BucketLookup controls path-style vs virtual-hosted-style URLs. // MinIO, Ceph, and most self-hosted S3 require path-style. if cfg.ForcePathStyle { opts.BucketLookup = minio.BucketLookupPath } else { opts.BucketLookup = minio.BucketLookupAuto } client, err := minio.New(endpoint, opts) if err != nil { return nil, fmt.Errorf("storage/s3: client init: %w", err) } // Normalize prefix: ensure trailing slash if non-empty, no leading slash prefix := strings.TrimPrefix(cfg.Prefix, "/") if prefix != "" && !strings.HasSuffix(prefix, "/") { prefix += "/" } s := &S3Store{ client: client, bucket: cfg.Bucket, prefix: prefix, endpoint: endpoint, } // Validate: check bucket exists and is accessible ctx := context.Background() exists, err := client.BucketExists(ctx, cfg.Bucket) if err != nil { return nil, fmt.Errorf("storage/s3: cannot reach endpoint: %w", err) } if !exists { return nil, fmt.Errorf("storage/s3: bucket %q does not exist", cfg.Bucket) } return s, nil } func (s *S3Store) Backend() string { return "s3" } // Put writes data to S3 at the given key. func (s *S3Store) Put(ctx context.Context, key string, r io.Reader, size int64, contentType string) error { if err := validateKey(key); err != nil { return err } if contentType == "" { contentType = "application/octet-stream" } opts := minio.PutObjectOptions{ ContentType: contentType, } // If size is unknown (0), we need to buffer or use -1. // minio-go handles -1 with multipart upload. uploadSize := size if uploadSize == 0 { uploadSize = -1 } _, err := s.client.PutObject(ctx, s.bucket, s.fullKey(key), r, uploadSize, opts) if err != nil { return fmt.Errorf("storage/s3: put %q: %w", key, err) } return nil } // Get returns a reader for the object at key. func (s *S3Store) Get(ctx context.Context, key string) (io.ReadCloser, int64, string, error) { if err := validateKey(key); err != nil { return nil, 0, "", err } obj, err := s.client.GetObject(ctx, s.bucket, s.fullKey(key), minio.GetObjectOptions{}) if err != nil { return nil, 0, "", fmt.Errorf("storage/s3: get %q: %w", key, err) } // Stat the object to get size and content type. // GetObject is lazy — the actual HTTP request happens on Read or Stat. info, err := obj.Stat() if err != nil { obj.Close() if isS3NotFound(err) { return nil, 0, "", ErrNotFound } return nil, 0, "", fmt.Errorf("storage/s3: stat %q: %w", key, err) } ct := info.ContentType if ct == "" { ct = "application/octet-stream" } return obj, info.Size, ct, nil } // Delete removes the object at key. func (s *S3Store) Delete(ctx context.Context, key string) error { if err := validateKey(key); err != nil { return err } err := s.client.RemoveObject(ctx, s.bucket, s.fullKey(key), minio.RemoveObjectOptions{}) if err != nil { // S3 RemoveObject is already idempotent (no error for missing keys) return fmt.Errorf("storage/s3: delete %q: %w", key, err) } return nil } // DeletePrefix removes all objects under the given prefix. func (s *S3Store) DeletePrefix(ctx context.Context, prefix string) error { if err := validateKey(prefix); err != nil { return err } fullPrefix := s.fullKey(prefix) // Ensure trailing slash for directory-like prefix if !strings.HasSuffix(fullPrefix, "/") { fullPrefix += "/" } // List all objects under prefix objectsCh := s.client.ListObjects(ctx, s.bucket, minio.ListObjectsOptions{ Prefix: fullPrefix, Recursive: true, }) // Build removal channel errCh := s.client.RemoveObjects(ctx, s.bucket, toRemoveChannel(objectsCh), minio.RemoveObjectsOptions{}) // Drain errors var firstErr error for e := range errCh { if e.Err != nil && firstErr == nil { firstErr = fmt.Errorf("storage/s3: delete prefix %q: %w", prefix, e.Err) } } return firstErr } // Exists checks if an object exists at key. func (s *S3Store) Exists(ctx context.Context, key string) (bool, error) { if err := validateKey(key); err != nil { return false, err } _, err := s.client.StatObject(ctx, s.bucket, s.fullKey(key), minio.StatObjectOptions{}) if err != nil { if isS3NotFound(err) { return false, nil } return false, fmt.Errorf("storage/s3: exists %q: %w", key, err) } return true, nil } // List returns objects under the given prefix, up to limit entries. func (s *S3Store) List(ctx context.Context, prefix string, limit int) ([]ObjectEntry, error) { if prefix != "" { if err := validateKey(prefix); err != nil { return nil, err } } if limit <= 0 { limit = 100 } fullPrefix := s.fullKey(prefix) objectsCh := s.client.ListObjects(ctx, s.bucket, minio.ListObjectsOptions{ Prefix: fullPrefix, Recursive: true, }) var entries []ObjectEntry for obj := range objectsCh { if obj.Err != nil { return entries, fmt.Errorf("storage/s3: list %q: %w", prefix, obj.Err) } if len(entries) >= limit { // Drain the remaining channel entries (minio requires it). for range objectsCh { } break } // Strip the s3 prefix to recover the storage key. key := strings.TrimPrefix(obj.Key, s.prefix) ct := obj.ContentType if ct == "" { ct = "application/octet-stream" } entries = append(entries, ObjectEntry{Key: key, Size: obj.Size, ContentType: ct}) } return entries, nil } // Healthy checks connectivity by performing a HeadBucket. func (s *S3Store) Healthy(ctx context.Context) error { exists, err := s.client.BucketExists(ctx, s.bucket) if err != nil { return fmt.Errorf("storage/s3: health check failed: %w", err) } if !exists { return fmt.Errorf("storage/s3: bucket %q no longer exists", s.bucket) } // Write + read a probe object to verify full access probeKey := s.fullKey(".health-probe") _, err = s.client.PutObject(ctx, s.bucket, probeKey, bytes.NewReader([]byte("ok")), 2, minio.PutObjectOptions{ContentType: "text/plain"}) if err != nil { return fmt.Errorf("storage/s3: not writable: %w", err) } s.client.RemoveObject(ctx, s.bucket, probeKey, minio.RemoveObjectOptions{}) return nil } // Stats returns object count and total size in the bucket (under prefix). func (s *S3Store) Stats(ctx context.Context) (*StorageStats, error) { stats := &StorageStats{ Backend: "s3", Bucket: s.bucket, Endpoint: s.endpoint, Configured: true, } // Check health first if err := s.Healthy(ctx); err != nil { stats.Healthy = false return stats, nil } stats.Healthy = true // Count objects under the files prefix filePrefix := s.fullKey("files/") objectsCh := s.client.ListObjects(ctx, s.bucket, minio.ListObjectsOptions{ Prefix: filePrefix, Recursive: true, }) for obj := range objectsCh { if obj.Err != nil { log.Printf("storage/s3: stats list error: %v", obj.Err) break } stats.TotalFiles++ stats.TotalBytes += obj.Size } return stats, nil } // ── Helpers ──────────────────────────────── // fullKey prepends the configured prefix to a storage key. func (s *S3Store) fullKey(key string) string { return s.prefix + key } // isS3NotFound checks if an error represents a "not found" condition. func isS3NotFound(err error) bool { if err == nil { return false } resp := minio.ToErrorResponse(err) return resp.Code == "NoSuchKey" || resp.StatusCode == 404 } // toRemoveChannel converts a ListObjects channel to a RemoveObjects input channel. func toRemoveChannel(objects <-chan minio.ObjectInfo) <-chan minio.ObjectInfo { // RemoveObjects accepts the same channel type — pass through directly. // The channel is already producing ObjectInfo values with Key set. return objects }