iu
d38ede0201
Add video as a media kind in photos: streaming upload with per-kind limits, ffprobe metadata, poster frame reusing the image variant pipeline, and a playback object (remux when web-compatible, else h264/yuv420p/aac transcode). Video jobs run on a separate pool with a long lease. Frontend gains a video viewer, grid badge/duration and video upload.
343 lines
10 KiB
Go
343 lines
10 KiB
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)
|
|
// Open returns a streaming reader over a blob, for processing objects too
|
|
// large to buffer in memory (for example video). The caller closes it.
|
|
Open(ctx context.Context, key string) (io.ReadCloser, 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
|
|
}
|
|
|
|
// Open streams a blob from S3 for processing without buffering it fully.
|
|
func (c *S3) Open(ctx context.Context, key string) (io.ReadCloser, error) {
|
|
out, err := c.s3.GetObject(ctx, &s3.GetObjectInput{
|
|
Bucket: aws.String(c.bucket),
|
|
Key: aws.String(key),
|
|
})
|
|
if err != nil {
|
|
if isNoSuchKey(err) {
|
|
return nil, ErrObjectNotFound
|
|
}
|
|
return nil, fmt.Errorf("open object %s: %w", key, err)
|
|
}
|
|
return out.Body, 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
|
|
}
|