Stream post videos to Spaces (#23)

Pass the spooled video file through instead of buffering it in RAM, and sign Spaces uploads as UNSIGNED-PAYLOAD so the client does not hash the body first.

Reviewed-on: #23
Co-authored-by: codegirl-007 <s.raide@gmail.com>

go.mod

@@ -7,6 +7,7 @@ require (
 	github.com/aws/aws-sdk-go-v2 v1.43.7
 	github.com/aws/aws-sdk-go-v2/credentials v1.19.37
 	github.com/aws/aws-sdk-go-v2/service/s3 v1.107.3
+	github.com/aws/smithy-go v1.27.8
 	github.com/bwmarrin/discordgo v0.29.0
 	github.com/go-chi/chi/v5 v5.3.1
 	github.com/google/uuid v1.6.0
@@ -27,7 +28,6 @@ require (
 	github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.31 // indirect
 	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.38 // indirect
 	github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.39 // indirect
-	github.com/aws/smithy-go v1.27.8 // indirect
 	github.com/gorilla/websocket v1.4.2 // indirect
 	github.com/jackc/pgpassfile v1.0.0 // indirect
 	github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect

internal/blob/spaces.go

@@ -8,9 +8,11 @@ import (
 	"strings"
 
 	"github.com/aws/aws-sdk-go-v2/aws"
+	v4 "github.com/aws/aws-sdk-go-v2/aws/signer/v4"
 	"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/middleware"
 )
 
 // Uploader stores public avatar objects.
@@ -78,13 +80,26 @@ func NewSpaces(cfg SpacesConfig) Uploader {
 		return Disabled{}
 	}
 	client := s3.New(s3.Options{
-		Region:       cfg.Region,
+		Region:                     cfg.Region,
-		Credentials:  credentials.NewStaticCredentialsProvider(cfg.Key, cfg.Secret, ""),
+		Credentials:                credentials.NewStaticCredentialsProvider(cfg.Key, cfg.Secret, ""),
-		BaseEndpoint: aws.String(cfg.Endpoint),
+		BaseEndpoint:               aws.String(cfg.Endpoint),
+		RequestChecksumCalculation: aws.RequestChecksumCalculationWhenRequired,
+		APIOptions: []func(*middleware.Stack) error{
+			spacesUnsignedPayload,
+		},
 	})
 	return &spaces{client: client, cfg: cfg}
 }
 
+// spacesUnsignedPayload signs Spaces PUTs as UNSIGNED-PAYLOAD so the client
+// can stream the body without hashing it first.
+func spacesUnsignedPayload(stack *middleware.Stack) error {
+	if err := v4.SwapComputePayloadSHA256ForUnsignedPayloadMiddleware(stack); err != nil {
+		return v4.AddUnsignedPayloadMiddleware(stack)
+	}
+	return nil
+}
+
 func (s *spaces) Enabled() bool { return true }
 
 func (s *spaces) Upload(ctx context.Context, obj FileUpload) (string, error) {

internal/web/post_images.go

@@ -190,13 +190,14 @@ func (s *Server) uploadPostMedia(
 		if err != nil {
 			return store.PostImage{}, "", err
 		}
+		defer prepared.body.Close()
 		mediaID := uuid.NewString()
 		objectKey := path.Join("post-videos", postID, mediaID+prepared.extension)
 		publicURL, err := s.cfg.Blob.Upload(ctx, blob.FileUpload{
 			Key:         objectKey,
-			Body:        bytes.NewReader(prepared.body),
+			Body:        prepared.body,
 			ContentType: prepared.contentType,
-			Size:        int64(len(prepared.body)),
+			Size:        prepared.size,
 		})
 		if err != nil {
 			return store.PostImage{}, "", &postImageRequestError{
@@ -405,7 +406,8 @@ func isWebM(raw []byte) bool {
 }
 
 type preparedPostVideo struct {
-	body        []byte
+	body        io.ReadCloser
+	size        int64
 	extension   string
 	contentType string
 }
@@ -414,6 +416,9 @@ func preparePostVideo(header *multipart.FileHeader) (preparedPostVideo, error) {
 	if header == nil {
 		return preparedPostVideo{}, invalidPostImage("Select a valid video.", nil)
 	}
+	if header.Size == 0 {
+		return preparedPostVideo{}, invalidPostImage("Videos cannot be empty.", nil)
+	}
 	if header.Size > postVideoMaxFileBytes {
 		return preparedPostVideo{}, &postImageRequestError{
 			status:  http.StatusRequestEntityTooLarge,
@@ -424,27 +429,28 @@ func preparePostVideo(header *multipart.FileHeader) (preparedPostVideo, error) {
 	if err != nil {
 		return preparedPostVideo{}, invalidPostImage("Could not read video.", err)
 	}
-	defer file.Close()
+	peek := make([]byte, 512)
-	raw, err := io.ReadAll(io.LimitReader(file, postVideoMaxFileBytes+1))
+	n, err := io.ReadFull(file, peek)
-	if err != nil {
+	if err != nil && !errors.Is(err, io.ErrUnexpectedEOF) && !errors.Is(err, io.EOF) {
+		file.Close()
 		return preparedPostVideo{}, invalidPostImage("Could not read video.", err)
 	}
-	if len(raw) == 0 {
+	if n == 0 {
+		file.Close()
 		return preparedPostVideo{}, invalidPostImage("Videos cannot be empty.", nil)
 	}
-	if int64(len(raw)) > postVideoMaxFileBytes {
+	switch mediaKindFromBytes(peek[:n]) {
-		return preparedPostVideo{}, &postImageRequestError{
-			status:  http.StatusRequestEntityTooLarge,
-			message: "Each video must be 25 MB or smaller.",
-		}
-	}
-	switch mediaKindFromBytes(raw) {
 	case store.MediaKindVideo:
 	default:
+		file.Close()
 		return preparedPostVideo{}, invalidPostImage("Videos must be MP4 or WebM.", nil)
 	}
-	result := preparedPostVideo{body: raw}
+	if _, err := file.Seek(0, io.SeekStart); err != nil {
-	if isWebM(raw) {
+		file.Close()
+		return preparedPostVideo{}, invalidPostImage("Could not read video.", err)
+	}
+	result := preparedPostVideo{body: file, size: header.Size}
+	if isWebM(peek[:n]) {
 		result.extension = ".webm"
 		result.contentType = "video/webm"
 		return result, nil

internal/web/post_images_test.go

@@ -60,20 +60,34 @@ func TestPreparePostImage(t *testing.T) {
 func TestPreparePostVideo(t *testing.T) {
 	t.Parallel()
 
-	prepared, err := preparePostVideoHeader(t, "clip.mp4", tinyMP4())
+	mp4 := tinyMP4()
+	prepared, err := preparePostVideoHeader(t, "clip.mp4", mp4)
 	if err != nil {
 		t.Fatal(err)
 	}
-	if prepared.extension != ".mp4" || prepared.contentType != "video/mp4" || len(prepared.body) == 0 {
+	defer prepared.body.Close()
-		t.Fatalf("prepared MP4 = %+v", prepared)
+	got, err := io.ReadAll(prepared.body)
-	}
-	prepared, err = preparePostVideoHeader(t, "clip.webm", tinyWebM())
 	if err != nil {
 		t.Fatal(err)
 	}
-	if prepared.extension != ".webm" || prepared.contentType != "video/webm" {
+	if prepared.extension != ".mp4" || prepared.contentType != "video/mp4" ||
+		prepared.size != int64(len(mp4)) || !bytes.Equal(got, mp4) {
+		t.Fatalf("prepared MP4 = %+v len(body)=%d", prepared, len(got))
+	}
+
+	webm := tinyWebM()
+	prepared, err = preparePostVideoHeader(t, "clip.webm", webm)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer prepared.body.Close()
+	if prepared.extension != ".webm" || prepared.contentType != "video/webm" ||
+		prepared.size != int64(len(webm)) {
 		t.Fatalf("prepared WebM = %+v", prepared)
 	}
+	if _, err := preparePostVideoHeader(t, "empty.mp4", nil); err == nil {
+		t.Fatal("empty video unexpectedly succeeded")
+	}
 	if _, err := preparePostVideoHeader(t, "notes.txt", []byte("not a video")); err == nil {
 		t.Fatal("text video upload unexpectedly succeeded")
 	}
@@ -269,6 +283,17 @@ func TestPostVideoMultipartLifecycle(t *testing.T) {
 		!strings.HasPrefix(root.Images[1].ObjectKey, "post-videos/") {
 		t.Fatalf("root media = %+v", root.Images)
 	}
+	clip := tinyMP4()
+	var streamed recordedImageUpload
+	for _, upload := range blobs.recordedUploads() {
+		if strings.HasPrefix(upload.key, "post-videos/") {
+			streamed = upload
+			break
+		}
+	}
+	if streamed.size != int64(len(clip)) || !bytes.Equal(streamed.body, clip) {
+		t.Fatalf("streamed video upload = %+v", streamed)
+	}
 
 	rec = multipartPost(t, handler, "/posts/"+root.ID+"/edit", map[string][]string{
 		"_csrf":                      {csrf},
@@ -513,6 +538,7 @@ func solidJPEG(t *testing.T, width, height int) []byte {
 type recordedImageUpload struct {
 	key         string
 	contentType string
+	size        int64
 	body        []byte
 }
 
@@ -540,6 +566,7 @@ func (b *recordingImageBlob) Upload(_ context.Context, object blob.FileUpload) (
 	b.uploads = append(b.uploads, recordedImageUpload{
 		key:         object.Key,
 		contentType: object.ContentType,
+		size:        object.Size,
 		body:        body,
 	})
 	return "https://cdn.example/" + object.Key, nil
@@ -553,9 +580,13 @@ func (b *recordingImageBlob) Delete(_ context.Context, key string) error {
 }
 
 func (b *recordingImageBlob) uploadCount() int {
+	return len(b.recordedUploads())
+}
+
+func (b *recordingImageBlob) recordedUploads() []recordedImageUpload {
 	b.mu.Lock()
 	defer b.mu.Unlock()
-	return len(b.uploads)
+	return append([]recordedImageUpload(nil), b.uploads...)
 }
 
 func (b *recordingImageBlob) deletedKeys() []string {