feat(transcoder): enhance video metadata handling and notifier integration

Author: subrotokumar

backend/server/media_handler.go

@@ -46,10 +46,10 @@ type (
 		ContentType string `json:"content_type" validate:"required"`
 	}
 	AssetsResponseData struct {
-		UploadUrl string            `json:"upload_url"`
-		Header    map[string]string `json:"header"`
-		Asset     Asset             `json:"asset"`
-		Form      map[string]string `json:"form"`
+		UploadUrl string             `json:"upload_url"`
+		Header    *map[string]string `json:"header,omitempty"`
+		Asset     *Asset             `json:"asset,omitempty"`
+		Form      *map[string]string `json:"form,omitempty"`
 	}
 	AssetsResponse struct {
 		Data    *AssetsResponseData `json:"data,omitempty"`
@@ -117,13 +117,13 @@ func (s *Server) VideoAssetsHandler(c echo.Context) error {
 		s.log.Error(ErrFailedToCreateVideoRecord, "err", err)
 		return c.JSON(http.StatusInternalServerError, AssetsResponse{Error: ErrFailedToCreateVideoRecord})
 	}
-	presignedUrl, err := s.storage.PresignedClient().PresignPostObject(c.Request().Context(), &s3.PutObjectInput{
+	presignedUrl, err := s.storage.PresignedClient().PresignPutObject(c.Request().Context(), &s3.PutObjectInput{
 		Bucket:        aws.String(s.cfg.S3.RawMediaBucket),
 		Key:           aws.String(key),
 		ContentType:   aws.String(body.ContentType),
 		ContentLength: aws.Int64(int64(body.Size)),
 		Metadata:      map[string]string{},
-	}, func(options *s3.PresignPostOptions) {
+	}, func(options *s3.PresignOptions) {
 		options.Expires = time.Duration(POST_PRESIGNED_URL_TTL) * time.Second
 	})
 	if err != nil {
@@ -133,16 +133,16 @@ func (s *Server) VideoAssetsHandler(c echo.Context) error {
 	return c.JSON(http.StatusOK, AssetsResponse{
 		Data: &AssetsResponseData{
 			UploadUrl: presignedUrl.URL,
-			Header:    map[string]string{},
-			Asset: Asset{
+			Header:    &map[string]string{},
+			Asset: &Asset{
 				Id:           videoId,
 				Name:         body.Name,
 				Size:         int(body.Size),
 				ContentType:  body.ContentType,
 				Href:         "",
 				OriginalName: body.Name,
 			},
-			Form: presignedUrl.Values,
+			Form: nil,
 		},
 		Message: MsgPresignedURLGenerated,
 	})

backend/swagger/docs.go

@@ -567,15 +567,19 @@ const docTemplate = `{
                 "PREUPLOAD",
                 "UPLOADED",
                 "PROCESSING",
+                "FAILED",
                 "READY",
-                "FAILED"
+                "PRIVATE",
+                "PUBLIC"
             ],
             "x-enum-varnames": [
                 "VideoStatusPREUPLOAD",
                 "VideoStatusUPLOADED",
                 "VideoStatusPROCESSING",
+                "VideoStatusFAILED",
                 "VideoStatusREADY",
-                "VideoStatusFAILED"
+                "VideoStatusPRIVATE",
+                "VideoStatusPUBLIC"
             ]
         },
         "pgtype.InfinityModifier": {

backend/swagger/swagger.json

@@ -561,15 +561,19 @@
                 "PREUPLOAD",
                 "UPLOADED",
                 "PROCESSING",
+                "FAILED",
                 "READY",
-                "FAILED"
+                "PRIVATE",
+                "PUBLIC"
             ],
             "x-enum-varnames": [
                 "VideoStatusPREUPLOAD",
                 "VideoStatusUPLOADED",
                 "VideoStatusPROCESSING",
+                "VideoStatusFAILED",
                 "VideoStatusREADY",
-                "VideoStatusFAILED"
+                "VideoStatusPRIVATE",
+                "VideoStatusPUBLIC"
             ]
         },
         "pgtype.InfinityModifier": {

backend/swagger/swagger.yaml

@@ -20,15 +20,19 @@ definitions:
     - PREUPLOAD
     - UPLOADED
     - PROCESSING
-    - READY
     - FAILED
+    - READY
+    - PRIVATE
+    - PUBLIC
     type: string
     x-enum-varnames:
     - VideoStatusPREUPLOAD
     - VideoStatusUPLOADED
     - VideoStatusPROCESSING
-    - VideoStatusREADY
     - VideoStatusFAILED
+    - VideoStatusREADY
+    - VideoStatusPRIVATE
+    - VideoStatusPUBLIC
   pgtype.InfinityModifier:
     enum:
     - 1

libs/db/models.go

@@ -6,8 +6,8 @@ sql:
   - schema: "migration"
     queries: "libs/db/queries"
     engine: "postgresql"
-    database:
-      managed: true
+    # database:
+    #   managed: true
     gen:
       go:
         package: "db"

sqlc.yml

@@ -2,6 +2,7 @@ package config
 
 import (
 	"encoding/json"
+	"strings"
 
 	"gitlab.com/subrotokumar/playstack/libs/core"
 	"gitlab.com/subrotokumar/playstack/libs/storage"
@@ -21,6 +22,11 @@ type Config struct {
 		SecretAccessKey string `yaml:"secret_key" envconfig:"AWS_SECRET_ACCESS_KEY"`
 		MediaBucket     string `yaml:"media_bucket" envconfig:"MEDIA_BUCKET" required:"true"`
 	} `yaml:"aws"`
+	NotifierService struct {
+		URL      string `yaml:"api" envconfig:"NOTIFIER_SERVICE_ENDPOINT" default:"http://localhost:8080"`
+		Username string `yaml:"username" envconfig:"BASIC_AUTH_USERNAME"`
+		PASSWORD string `yaml:"password" envconfig:"BASIC_AUTH_PASSWORD"`
+	} `yaml:"notifier_service"`
 	Event   string `yaml:"events" envconfig:"SQS_MESSAGE" required:"true"`
 	S3Event storage.S3Event
 }
@@ -38,6 +44,11 @@ func (cfg *Config) Key() string {
 	return cfg.S3Event.Records[0].S3.Object.Key
 }
 
+func (cfg *Config) UserAndVideoID() (string, string) {
+	keys := strings.Split(cfg.Key(), "/")
+	return keys[1], keys[2]
+}
+
 func (cfg *Config) ObjectSize() int64 {
 	return cfg.S3Event.Records[0].S3.Object.Size
 }

transcoder/config/config.go

@@ -1,10 +1,73 @@
 package service
 
 import (
+	"bytes"
 	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+
+	"gitlab.com/subrotokumar/playstack/libs/db"
+)
+
+type (
+	UpdateMetadataRequest struct {
+		Title       *string        `json:"title"`
+		Status      db.VideoStatus `json:"status" validate:"omitempty,oneof='PREUPLOAD' 'UPLOADED' 'PROCESSING' 'READY' 'FAILED'"`
+		DurationSec *int32         `json:"duration_sec"`
+	}
 )
 
-func (s *Service) UpdateMetadata(ctx context.Context) error {
+func (s *Service) UpdateMetadata(ctx context.Context, request UpdateMetadataRequest) error {
 	s.log.Info("Updating video metadata in database")
+
+	userID, videoID := s.cfg.UserAndVideoID()
+
+	url := s.cfg.NotifierService.URL + "/internal/media/videos/" + videoID
+	payload := make(map[string]any)
+	payload["user_id"] = userID
+	payload["status"] = string(request.Status)
+
+	if request.Title != nil {
+		payload["title"] = *request.Title
+	}
+	if request.DurationSec != nil {
+		payload["duration_sec"] = *request.DurationSec
+	}
+
+	bodyBytes, err := json.Marshal(payload)
+	if err != nil {
+		s.log.Error("failed to marshal update payload", "err", err)
+		return err
+	}
+
+	req, err := http.NewRequestWithContext(ctx, "PATCH", url, bytes.NewReader(bodyBytes))
+	if err != nil {
+		s.log.Error("failed to build request", "err", err)
+		return err
+	}
+
+	username := s.cfg.NotifierService.Username
+	password := s.cfg.NotifierService.PASSWORD
+	req.Header.Set("Content-Type", "application/json")
+	if username != "" || password != "" {
+		req.SetBasicAuth(username, password)
+	}
+
+	res, err := http.DefaultClient.Do(req)
+	if err != nil {
+		s.log.Error("notifier request failed", "err", err)
+		return err
+	}
+	defer res.Body.Close()
+
+	respBody, _ := io.ReadAll(res.Body)
+	if res.StatusCode < 200 || res.StatusCode >= 300 {
+		s.log.Error("notifier returned non-2xx", "status", res.StatusCode, "body", string(respBody))
+		return fmt.Errorf("notifier returned status: %s", res.Status)
+	}
+
+	s.log.Info("Notifier updated metadata", "status", res.StatusCode)
 	return nil
 }

transcoder/service/notifier.go

@@ -11,9 +11,14 @@ import (
 
 	"github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go-v2/service/s3"
+	"gitlab.com/subrotokumar/playstack/libs/db"
 	"gitlab.com/subrotokumar/playstack/transcoder/ffmpeg"
 )
 
+const (
+	MsgVideoMetadataUpdateFailed string = "failed to update video metadata"
+)
+
 func (s *Service) Download(ctx context.Context, destPath string) error {
 	s.log.Info("Downloading file", "path", destPath)
 
@@ -64,10 +69,7 @@ func (s *Service) Transcode(ctx context.Context, inputPath, outputDir string) er
 
 func (s *Service) Upload(ctx context.Context, sourceDir string) error {
 	s.log.Info("Uploading files from", "dir", sourceDir)
-	keys := strings.Split(s.cfg.Key(), "/")
-	userId := keys[0]
-	videoId := keys[1]
-	uploadKey := userId + "/" + videoId + "/" + "output/"
+	uploadKey := strings.ReplaceAll(s.cfg.Key(), "video.mp4", "output/")
 	err := filepath.Walk(sourceDir, func(path string, info os.FileInfo, err error) error {
 		if err != nil {
 			return err
@@ -89,7 +91,7 @@ func (s *Service) Upload(ctx context.Context, sourceDir string) error {
 
 		s.log.Info("Uploading", "key", uploadKey+relPath)
 		_, err = s.storage.Client().PutObject(ctx, &s3.PutObjectInput{
-			Bucket: aws.String(s.cfg.Bucket()),
+			Bucket: aws.String(s.cfg.Aws.MediaBucket),
 			Key:    aws.String(uploadKey + relPath),
 			Body:   file,
 		})
@@ -107,6 +109,10 @@ func (s *Service) Upload(ctx context.Context, sourceDir string) error {
 }
 
 func (s *Service) Process(ctx context.Context) error {
+	if err := s.UpdateMetadata(ctx, UpdateMetadataRequest{Status: db.VideoStatusUPLOADED}); err != nil {
+		s.log.Error(MsgVideoMetadataUpdateFailed, "err", err.Error())
+	}
+
 	workDir := "./tmp/workspace"
 	if err := os.MkdirAll(workDir, 0o755); err != nil {
 		return fmt.Errorf("create work dir: %w", err)
@@ -131,15 +137,22 @@ func (s *Service) Process(ctx context.Context) error {
 		return fmt.Errorf("download video: %w", err)
 	}
 
+	if err := s.UpdateMetadata(ctx, UpdateMetadataRequest{Status: db.VideoStatusPROCESSING}); err != nil {
+		s.log.Error(MsgVideoMetadataUpdateFailed, "err", err.Error())
+	}
+
 	if err := s.Transcode(ctx, inputPath, outputPath); err != nil {
+		s.UpdateMetadata(ctx, UpdateMetadataRequest{Status: db.VideoStatusFAILED})
 		return fmt.Errorf("transcode video: %w", err)
 	}
 
 	if err := s.Upload(ctx, outputPath); err != nil {
+		s.UpdateMetadata(ctx, UpdateMetadataRequest{Status: db.VideoStatusFAILED})
 		return fmt.Errorf("upload files: %w", err)
 	}
-	if err := s.UpdateMetadata(ctx); err != nil {
-		return fmt.Errorf("update video: %w", err)
+	if err := s.UpdateMetadata(ctx, UpdateMetadataRequest{Status: db.VideoStatusREADY}); err != nil {
+		s.log.Error(MsgVideoMetadataUpdateFailed, "err", err.Error())
+		return err
 	}
 	return nil
 }