313 lines
8.6 KiB
Go
313 lines
8.6 KiB
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. "switchboard/")
|
|
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. "switchboard/")
|
|
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
|
|
}
|
|
|
|
// 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 attachments prefix
|
|
attPrefix := s.fullKey("attachments/")
|
|
objectsCh := s.client.ListObjects(ctx, s.bucket, minio.ListObjectsOptions{
|
|
Prefix: attPrefix,
|
|
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
|
|
}
|