core/server/storage/s3.go

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
}