ds-galo/backend/internal/storage/storage.go

package storage

import (
	"context"
	"errors"
	"fmt"
	"io"
	"net/http"
	"strconv"
	"strings"
	"time"

	"github.com/aws/aws-sdk-go-v2/aws"
	awsconfig "github.com/aws/aws-sdk-go-v2/config"
	"github.com/aws/aws-sdk-go-v2/credentials"
	"github.com/aws/aws-sdk-go-v2/service/s3"
	"github.com/aws/aws-sdk-go-v2/service/s3/types"
	"github.com/aws/smithy-go"

	"galo/internal/config"
)

// ErrSignedURLInvalid is returned when an app-served object URL has a missing,
// malformed, expired or forged signature.
var ErrSignedURLInvalid = errors.New("invalid or expired object signature")

// ErrObjectNotFound is returned when the requested blob does not exist.
var ErrObjectNotFound = errors.New("object not found")

// ErrRangeNotSatisfiable is returned when a requested byte range is invalid.
var ErrRangeNotSatisfiable = errors.New("requested range not satisfiable")

// ObjectInfo describes one stored blob for inventory and reconciliation.
type ObjectInfo struct {
	Key     string
	Size    int64
	ModTime time.Time
}

// Backend is the object storage abstraction. Keys are opaque and independent of
// the backend: every implementation stores and serves the same logical blob.
// Blobs are always delivered through the app.
type Backend interface {
	Put(ctx context.Context, key string, body io.Reader, size int64, contentType string) error
	Delete(ctx context.Context, keys ...string) error
	Get(ctx context.Context, key string) ([]byte, error)
	// URL returns a time-limited signed app URL the client can use to fetch the blob.
	URL(ctx context.Context, key string) (string, error)
	// ServeObject validates the signed URL and writes the blob, including
	// Content-Type and range handling, to the response.
	ServeObject(w http.ResponseWriter, r *http.Request, key string) error
	// List returns the blobs under prefix for inventory and reconciliation.
	List(ctx context.Context, prefix string) ([]ObjectInfo, error)
	Ping(ctx context.Context) error
	EnsureBucket(ctx context.Context) error
	// Bucket returns a human-readable identifier for the storage location.
	Bucket() string
}

// S3 is a Backend backed by an S3-compatible object store (Garage). Blobs are
// delivered through the app: the client receives a signed app URL and the bytes
// are streamed from S3 over the internal network.
type S3 struct {
	s3     *s3.Client
	bucket string
	signer *Signer
}

// NewS3 builds an S3-compatible storage backend. signer issues the app URLs used
// to serve blobs.
func NewS3(ctx context.Context, cfg config.S3, signer *Signer) (*S3, error) {
	if signer == nil {
		return nil, errors.New("s3: signer is required")
	}
	awsCfg, err := awsconfig.LoadDefaultConfig(ctx,
		awsconfig.WithRegion(cfg.Region),
		awsconfig.WithCredentialsProvider(credentials.NewStaticCredentialsProvider(
			cfg.AccessKey, cfg.SecretKey, "",
		)),
	)
	if err != nil {
		return nil, fmt.Errorf("load aws config: %w", err)
	}

	client := s3.NewFromConfig(awsCfg, func(o *s3.Options) {
		o.BaseEndpoint = aws.String(cfg.Endpoint)
		o.UsePathStyle = cfg.UsePathStyle
	})

	return &S3{
		s3:     client,
		bucket: cfg.Bucket,
		signer: signer,
	}, nil
}

func (c *S3) Bucket() string { return c.bucket }

// Open builds the storage backend selected by configuration: S3-compatible by
// default, or the local file system when STORAGE_BACKEND=fs. Every backend
// delivers blobs through the app using a shared Signer. jwtSecret is the
// application secret used to sign object URLs when URL_SECRET is not set.
func Open(ctx context.Context, cfg *config.Config, jwtSecret string) (Backend, error) {
	secret := cfg.Storage.URLSecret
	if secret == "" {
		secret = jwtSecret
	}
	if secret == "" {
		return nil, errors.New("object URL secret is not available")
	}
	signer, err := NewSigner(cfg.Server.BaseURL, secret, cfg.Storage.URLTTL, cfg.Storage.URLSignWindow)
	if err != nil {
		return nil, err
	}
	switch cfg.Storage.Backend {
	case "", "s3":
		return NewS3(ctx, cfg.S3, signer)
	case "fs":
		return NewFS(cfg.Storage.FSRoot, signer)
	default:
		return nil, fmt.Errorf("unknown storage backend %q", cfg.Storage.Backend)
	}
}

func (c *S3) EnsureBucket(ctx context.Context) error {
	_, err := c.s3.HeadBucket(ctx, &s3.HeadBucketInput{Bucket: aws.String(c.bucket)})
	if err == nil {
		return nil
	}
	var nf *types.NotFound
	if !errors.As(err, &nf) && !isNoSuchBucket(err) {
		return fmt.Errorf("head bucket: %w", err)
	}
	if _, err := c.s3.CreateBucket(ctx, &s3.CreateBucketInput{Bucket: aws.String(c.bucket)}); err != nil {
		return fmt.Errorf("create bucket: %w", err)
	}
	return nil
}

func (c *S3) Ping(ctx context.Context) error {
	_, err := c.s3.HeadBucket(ctx, &s3.HeadBucketInput{Bucket: aws.String(c.bucket)})
	return err
}

func (c *S3) Put(ctx context.Context, key string, body io.Reader, size int64, contentType string) error {
	in := &s3.PutObjectInput{
		Bucket:      aws.String(c.bucket),
		Key:         aws.String(key),
		Body:        body,
		ContentType: aws.String(contentType),
	}
	if size >= 0 {
		in.ContentLength = aws.Int64(size)
	}
	if _, err := c.s3.PutObject(ctx, in); err != nil {
		return fmt.Errorf("put object %s: %w", key, err)
	}
	return nil
}

func (c *S3) Delete(ctx context.Context, keys ...string) error {
	if len(keys) == 0 {
		return nil
	}
	objs := make([]types.ObjectIdentifier, 0, len(keys))
	for _, k := range keys {
		objs = append(objs, types.ObjectIdentifier{Key: aws.String(k)})
	}
	if _, err := c.s3.DeleteObjects(ctx, &s3.DeleteObjectsInput{
		Bucket: aws.String(c.bucket),
		Delete: &types.Delete{Objects: objs, Quiet: aws.Bool(true)},
	}); err != nil {
		return fmt.Errorf("delete objects: %w", err)
	}
	return nil
}

func (c *S3) Get(ctx context.Context, key string) ([]byte, error) {
	out, err := c.s3.GetObject(ctx, &s3.GetObjectInput{
		Bucket: aws.String(c.bucket),
		Key:    aws.String(key),
	})
	if err != nil {
		return nil, fmt.Errorf("get object %s: %w", key, err)
	}
	defer out.Body.Close()
	data, err := io.ReadAll(out.Body)
	if err != nil {
		return nil, fmt.Errorf("read object %s: %w", key, err)
	}
	return data, nil
}

// URL returns a signed app URL. The blob itself is streamed by ServeObject, so
// the S3 endpoint is never exposed to the client.
func (c *S3) URL(_ context.Context, key string) (string, error) {
	return c.signer.URL(key), nil
}

// List returns every object under prefix using paginated ListObjectsV2.
func (c *S3) List(ctx context.Context, prefix string) ([]ObjectInfo, error) {
	var out []ObjectInfo
	var token *string
	for {
		in := &s3.ListObjectsV2Input{Bucket: aws.String(c.bucket)}
		if prefix != "" {
			in.Prefix = aws.String(prefix)
		}
		if token != nil {
			in.ContinuationToken = token
		}
		res, err := c.s3.ListObjectsV2(ctx, in)
		if err != nil {
			return nil, fmt.Errorf("list objects: %w", err)
		}
		for _, o := range res.Contents {
			out = append(out, ObjectInfo{
				Key:     aws.ToString(o.Key),
				Size:    aws.ToInt64(o.Size),
				ModTime: aws.ToTime(o.LastModified),
			})
		}
		if res.IsTruncated == nil || !*res.IsTruncated {
			return out, nil
		}
		token = res.NextContinuationToken
	}
}

// ServeObject validates the signed URL and streams the blob from S3. It passes
// the client's Range header through to S3, so byte ranges are handled natively,
// and answers HEAD via HeadObject. It writes the response itself.
func (c *S3) ServeObject(w http.ResponseWriter, r *http.Request, key string) error {
	if err := c.signer.VerifyRequest(key, r); err != nil {
		return err
	}
	if r.Method == http.MethodHead {
		return c.headObject(w, r, key)
	}
	in := &s3.GetObjectInput{Bucket: aws.String(c.bucket), Key: aws.String(key)}
	if rng := r.Header.Get("Range"); rng != "" {
		in.Range = aws.String(rng)
	}
	out, err := c.s3.GetObject(r.Context(), in)
	if err != nil {
		if isNoSuchKey(err) {
			return ErrObjectNotFound
		}
		if isInvalidRange(err) {
			return ErrRangeNotSatisfiable
		}
		return err
	}
	defer out.Body.Close()
	if ct := aws.ToString(out.ContentType); ct != "" {
		w.Header().Set("Content-Type", ct)
	} else {
		w.Header().Set("Content-Type", contentTypeForKey(key))
	}
	if out.ContentLength != nil {
		w.Header().Set("Content-Length", strconv.FormatInt(*out.ContentLength, 10))
	}
	if out.ContentRange != nil {
		w.Header().Set("Content-Range", *out.ContentRange)
	}
	status := http.StatusOK
	if out.ContentRange != nil {
		status = http.StatusPartialContent
	}
	w.WriteHeader(status)
	if _, err := io.Copy(w, out.Body); err != nil {
		return err
	}
	return nil
}

func (c *S3) headObject(w http.ResponseWriter, r *http.Request, key string) error {
	out, err := c.s3.HeadObject(r.Context(), &s3.HeadObjectInput{
		Bucket: aws.String(c.bucket),
		Key:    aws.String(key),
	})
	if err != nil {
		if isNoSuchKey(err) {
			return ErrObjectNotFound
		}
		return err
	}
	if ct := aws.ToString(out.ContentType); ct != "" {
		w.Header().Set("Content-Type", ct)
	} else {
		w.Header().Set("Content-Type", contentTypeForKey(key))
	}
	if out.ContentLength != nil {
		w.Header().Set("Content-Length", strconv.FormatInt(*out.ContentLength, 10))
	}
	w.WriteHeader(http.StatusOK)
	return nil
}

func isNoSuchKey(err error) bool {
	var apiErr smithy.APIError
	if errors.As(err, &apiErr) {
		code := apiErr.ErrorCode()
		return strings.EqualFold(code, "NoSuchKey") || strings.EqualFold(code, "NotFound")
	}
	return false
}

func isInvalidRange(err error) bool {
	var apiErr smithy.APIError
	if errors.As(err, &apiErr) {
		return strings.EqualFold(apiErr.ErrorCode(), "InvalidRange")
	}
	return false
}

func isNoSuchBucket(err error) bool {
	var apiErr smithy.APIError
	if errors.As(err, &apiErr) {
		code := apiErr.ErrorCode()
		return strings.EqualFold(code, "NoSuchBucket") || strings.EqualFold(code, "NotFound")
	}
	return false
}