FEAT: Update Status Of Video Row On Media Convert Update

cms/internal/api/videos.go

@@ -55,3 +55,22 @@ type CompleteResponse struct {
 	CreatedAt         time.Time `json:"createdAt"`
 	UpdatedAt         time.Time `json:"updatedAt"`
 }
+
+// VideoResponse is a persisted video record as returned by
+// GET /api/videos/{id}. It carries the same members as CompleteResponse —
+// deliberately a separate type, so the shape POST publishes and the shape GET
+// publishes can diverge without one silently dragging the other with it.
+type VideoResponse struct {
+	ID                string    `json:"id" example:"0199f3a1-7c2e-7b21-9f0d-1a2b3c4d5e6f"`
+	Title             string    `json:"title" example:"Inside the Newsroom"`
+	Description       string    `json:"description" example:"A behind-the-scenes look at the evening bulletin."`
+	CategoryIDs       []int16   `json:"categoryIds" example:"1,2"`
+	Tags              string    `json:"tags" example:"media, press, riyadh"`
+	FileName          string    `json:"fileName" example:"interview-cut.mov"`
+	StorageKey        string    `json:"storageKey" example:"videos/a1b2c3d4e5f6.mov"`
+	MediaConvertJobID string    `json:"mediaConvertJobId" example:"1755300000000-abcdef"`
+	Status            string    `json:"status" enums:"processing,ready,failed" example:"processing"`
+	SizeBytes         int64     `json:"sizeBytes" example:"60"`
+	CreatedAt         time.Time `json:"createdAt"`
+	UpdatedAt         time.Time `json:"updatedAt"`
+}

cms/internal/consumers/mediaconvert.go

@@ -0,0 +1,135 @@
+// Package consumers holds the queue-driven half of the service: work that
+// arrives on a queue rather than as an HTTP request. It is to SQS what
+// internal/handlers is to HTTP — it decodes a message, calls repositories, and
+// decides what the message's fate is — so the same seams apply, and no SQL
+// lives here.
+package consumers
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"log"
+	"thamanyah/cms/v2/internal/db/repositories"
+	"thamanyah/cms/v2/internal/models"
+	"thamanyah/cms/v2/internal/services"
+	"time"
+)
+
+// receiveBackoff is how long the loop waits after a failed receive, so a queue
+// that is unreachable produces a slow trickle of log lines rather than a spin.
+const receiveBackoff = 5 * time.Second
+
+// jobStateChange is the EventBridge event MediaConvert emits on every job
+// state transition. The subscription delivers it raw, so this is the whole
+// message body — there is no SNS envelope to unwrap. Only the members this
+// service acts on are modelled; AWS sends a good deal more.
+type jobStateChange struct {
+	Source     string `json:"source"`
+	DetailType string `json:"detail-type"`
+	Detail     struct {
+		JobID  string `json:"jobId"`
+		Status string `json:"status"`
+	} `json:"detail"`
+}
+
+// statusForJobState maps MediaConvert's job states onto the catalogue's. The
+// second return is false for states that are not an outcome — the job is still
+// running, and the record should stay where it is rather than being rewritten
+// with what it already says.
+func statusForJobState(state string) (models.VideoStatus, bool) {
+	switch state {
+	case "COMPLETE":
+		return models.VideoStatusReady, true
+	case "ERROR", "CANCELED":
+		return models.VideoStatusFailed, true
+	default:
+		return "", false
+	}
+}
+
+// RunMediaConvertEvents consumes job state changes until ctx is cancelled. It
+// is meant to be run in its own goroutine for the lifetime of the process.
+func RunMediaConvertEvents(ctx context.Context) {
+	log.Println("mediaconvert job events consumer started")
+
+	for {
+		if ctx.Err() != nil {
+			log.Println("mediaconvert job events consumer stopped")
+			return
+		}
+
+		messages, err := services.SQSClient.ReceiveMessages(ctx)
+		if err != nil {
+			if ctx.Err() != nil {
+				log.Println("mediaconvert job events consumer stopped")
+				return
+			}
+			log.Printf("Something Went Wrong Receiving Job Events: %s", err)
+			time.Sleep(receiveBackoff)
+			continue
+		}
+
+		for _, message := range messages {
+			if !handleJobEvent(ctx, message.Body) {
+				// Left on the queue on purpose: it becomes visible again when
+				// the visibility timeout expires, and reaches the dead-letter
+				// queue if it keeps failing.
+				continue
+			}
+
+			if err := services.SQSClient.DeleteMessage(ctx, message.ReceiptHandle); err != nil {
+				log.Printf("Something Went Wrong Acknowledging A Job Event: %s", err)
+			}
+		}
+	}
+}
+
+// handleJobEvent processes one message and reports whether it is finished
+// with — that is, whether it should be deleted from the queue. It returns
+// false only when a retry could plausibly succeed. Anything a retry cannot fix
+// (a body that will never parse, an event for a job nobody has) is finished
+// with, however little it accomplished: leaving it on the queue would only
+// stall the events behind it.
+func handleJobEvent(ctx context.Context, body string) bool {
+	var event jobStateChange
+	if err := json.Unmarshal([]byte(body), &event); err != nil {
+		log.Printf("Discarding A Job Event That Is Not JSON: %s", err)
+		return true
+	}
+
+	if event.Source != "aws.mediaconvert" || event.DetailType != "MediaConvert Job State Change" {
+		log.Printf("Discarding An Event That Is Not A MediaConvert Job State Change: source=%q detail-type=%q",
+			event.Source, event.DetailType)
+		return true
+	}
+
+	if event.Detail.JobID == "" {
+		log.Printf("Discarding A Job Event That Names No Job")
+		return true
+	}
+
+	status, isOutcome := statusForJobState(event.Detail.Status)
+	if !isOutcome {
+		// SUBMITTED, PROGRESSING and STATUS_UPDATE say the job is still
+		// running, which is what "processing" already records.
+		return true
+	}
+
+	err := repositories.VideoRepo.UpdateVideoStatusByJobID(ctx, event.Detail.JobID, status)
+	if errors.Is(err, repositories.ErrVideoNotFound) {
+		// Not a fault: the topic carries every job in the account, including
+		// ones this service never submitted.
+		log.Printf("Ignoring A Job Event For A Job No Video Has: job=%q status=%q",
+			event.Detail.JobID, event.Detail.Status)
+		return true
+	}
+	if err != nil {
+		log.Printf("Something Went Wrong Recording A Job Outcome: job=%q status=%q: %s",
+			event.Detail.JobID, event.Detail.Status, err)
+		return false
+	}
+
+	log.Printf("Recorded A Job Outcome: job=%q status=%q", event.Detail.JobID, status)
+	return true
+}

cms/internal/db/migrations/0003_index_videos_mediaconvert_job_id.down.sql

@@ -0,0 +1 @@
+DROP INDEX videos_mediaconvert_job_id_idx;

cms/internal/db/migrations/0003_index_videos_mediaconvert_job_id.up.sql

@@ -0,0 +1,4 @@
+-- Job state changes arrive keyed by MediaConvert's job id, so every event the
+-- consumer handles is a lookup on this column. Nothing indexed it before: the
+-- column was only ever written, never searched.
+CREATE INDEX videos_mediaconvert_job_id_idx ON videos (mediaconvert_job_id);

cms/internal/db/repositories/errors.go

@@ -0,0 +1,9 @@
+package repositories
+
+import "errors"
+
+// Sentinel errors the repositories return so handlers can tell "there is no
+// such row" from "the query failed", without importing database/sql to do it.
+
+// ErrVideoNotFound reports that no video has the requested id.
+var ErrVideoNotFound = errors.New("video not found")

cms/internal/db/repositories/videos.go

@@ -3,6 +3,7 @@ package repositories
 import (
 	"context"
 	"database/sql"
+	"errors"
 	"fmt"
 	"strings"
 	"thamanyah/cms/v2/internal/models"
@@ -14,6 +15,66 @@ type VideoRepository struct {
 
 var VideoRepo VideoRepository
 
+// UpdateVideoStatusByJobID moves the video a transcoding job belongs to into a
+// new status. It returns ErrVideoNotFound when no video carries that job id,
+// which the consumer treats as a message to drop rather than a fault.
+func (svc VideoRepository) UpdateVideoStatusByJobID(ctx context.Context, jobID string, status models.VideoStatus) error {
+	result, err := svc.SQLDB.ExecContext(ctx, `
+		UPDATE videos
+		SET status = $1, updated_at = now()
+		WHERE mediaconvert_job_id = $2
+	`, string(status), jobID)
+	if err != nil {
+		return err
+	}
+
+	affected, err := result.RowsAffected()
+	if err != nil {
+		return err
+	}
+	if affected == 0 {
+		return ErrVideoNotFound
+	}
+
+	return nil
+}
+
+// GetVideoByID reads one video and the ids of the categories it is filed
+// under. It returns ErrVideoNotFound when no video has that id.
+func (svc VideoRepository) GetVideoByID(ctx context.Context, id string) (models.Video, error) {
+	var v models.Video
+
+	err := svc.SQLDB.QueryRowContext(ctx, `
+		SELECT id, title, description, tags, file_name, storage_key, mediaconvert_job_id, status, size_bytes, created_at, updated_at
+		FROM videos
+		WHERE id = $1
+	`, id).Scan(&v.ID, &v.Title, &v.Description, &v.Tags, &v.FileName, &v.StorageKey,
+		&v.MediaConvertJobID, &v.Status, &v.SizeBytes, &v.CreatedAt, &v.UpdatedAt)
+	if errors.Is(err, sql.ErrNoRows) {
+		return models.Video{}, ErrVideoNotFound
+	}
+	if err != nil {
+		return models.Video{}, err
+	}
+
+	rows, err := svc.SQLDB.QueryContext(ctx,
+		`SELECT category_id FROM video_categories WHERE video_id = $1 ORDER BY category_id`, v.ID)
+	if err != nil {
+		return models.Video{}, err
+	}
+	defer rows.Close()
+
+	for rows.Next() {
+		var categoryID int16
+		if err := rows.Scan(&categoryID); err != nil {
+			return models.Video{}, err
+		}
+		v.CategoryIDs = append(v.CategoryIDs, categoryID)
+	}
+
+	return v, rows.Err()
+}
+
 // VideoExistsWithStorageKey reports whether a video has already been
 // registered under a storage key. It is served by videos_storage_key_key, the
 // index Postgres builds for the column's UNIQUE constraint.

cms/internal/handlers/videos.go

@@ -17,6 +17,8 @@ import (
 	"thamanyah/cms/v2/internal/models"
 	"thamanyah/cms/v2/internal/services"
 	"time"
+
+	"github.com/google/uuid"
 )
 
 const (
@@ -253,3 +255,62 @@ func randomFilename(original string) (string, error) {
 	}
 	return hex.EncodeToString(buf) + ext, nil
 }
+
+// GetVideo returns one video by its id.
+//
+//	@Summary		Read a video
+//	@Description	Returns the catalogue record for one video, including the status its transcoding job has reached. This is how a client follows an upload after POST /api/videos: the record starts "processing" and moves to "ready" or "failed" once MediaConvert reports the job finished.
+//	@Tags			videos
+//	@Produce		json
+//	@Param			id	path		string	true	"The video's id, as returned by POST /api/videos"
+//	@Success		200	{object}	api.VideoResponse
+//	@Failure		404	{object}	api.ProblemDetails	"No video has that id"
+//	@Failure		500	{object}	api.ProblemDetails	"The video could not be read from the database"
+//	@Router			/api/videos/{id} [get]
+func GetVideo(w http.ResponseWriter, r *http.Request) {
+	id := r.PathValue("id")
+
+	// Parsed before the query, because Postgres rejects anything that is not a
+	// uuid the moment it is compared against the column rather than simply
+	// matching no row — which would surface as a 500. An id that cannot name a
+	// video is a video that does not exist.
+	//
+	// The parsed value, not the raw one, is what goes to the database: uuid
+	// accepts spellings Postgres will not take verbatim — notably the
+	// "urn:uuid:" form — and String() puts them all back into the canonical
+	// form the column holds.
+	videoID, err := uuid.Parse(id)
+	if err != nil {
+		writeProblem(w, http.StatusNotFound, "Video Not Found",
+			fmt.Sprintf("No video has the id %q. Use the 'id' returned by POST /api/videos.", id))
+		return
+	}
+
+	video, err := repositories.VideoRepo.GetVideoByID(r.Context(), videoID.String())
+	if errors.Is(err, repositories.ErrVideoNotFound) {
+		writeProblem(w, http.StatusNotFound, "Video Not Found",
+			fmt.Sprintf("No video has the id %q. Use the 'id' returned by POST /api/videos.", id))
+		return
+	}
+	if err != nil {
+		log.Printf("Something Went Wrong Loading The Video: %s", err)
+		writeProblem(w, http.StatusInternalServerError, "Something Went Wrong Loading The Video",
+			"The video could not be read from the database. This is a server-side fault and the request was not processed; retrying in a few moments may succeed.")
+		return
+	}
+
+	writeJSON(w, http.StatusOK, api.VideoResponse{
+		ID:                video.ID,
+		Title:             video.Title,
+		Description:       video.Description,
+		CategoryIDs:       video.CategoryIDs,
+		Tags:              video.Tags,
+		FileName:          video.FileName,
+		StorageKey:        video.StorageKey,
+		MediaConvertJobID: video.MediaConvertJobID,
+		Status:            video.Status.String(),
+		SizeBytes:         video.SizeBytes,
+		CreatedAt:         video.CreatedAt,
+		UpdatedAt:         video.UpdatedAt,
+	})
+}

cms/internal/services/sqs.go

@@ -0,0 +1,80 @@
+package services
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/aws/aws-sdk-go-v2/service/sqs"
+)
+
+var SQSClient SQS
+
+// QueueMessage is one message taken off the queue. ReceiptHandle is what
+// identifies it for deletion — it belongs to this delivery, not to the
+// message, so it cannot be held across receives.
+type QueueMessage struct {
+	Body          string
+	ReceiptHandle string
+}
+
+type SQS interface {
+	ReceiveMessages(ctx context.Context) ([]QueueMessage, error)
+	DeleteMessage(ctx context.Context, receiptHandle string) error
+}
+
+type SQSConcrete struct {
+	SQSClient *sqs.Client
+	QueueURL  string
+}
+
+// receiveWaitSeconds turns every receive into a long poll: the call parks on
+// the server until a message arrives or this elapses, so an event is picked up
+// within milliseconds of being published while an idle consumer costs one
+// request every twenty seconds rather than spinning. 20 is the AWS maximum.
+const receiveWaitSeconds = 20
+
+func (svc SQSConcrete) ReceiveMessages(ctx context.Context) ([]QueueMessage, error) {
+	output, err := svc.SQSClient.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{
+		QueueUrl:            &svc.QueueURL,
+		MaxNumberOfMessages: 10,
+		WaitTimeSeconds:     receiveWaitSeconds,
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	messages := make([]QueueMessage, 0, len(output.Messages))
+	for _, message := range output.Messages {
+		if message.Body == nil || message.ReceiptHandle == nil {
+			continue
+		}
+		messages = append(messages, QueueMessage{
+			Body:          *message.Body,
+			ReceiptHandle: *message.ReceiptHandle,
+		})
+	}
+
+	return messages, nil
+}
+
+// DeleteMessage acknowledges a message. Until this is called the message is
+// merely invisible, and it returns to the queue when its visibility timeout
+// expires — which is how a handler that failed gets another attempt.
+func (svc SQSConcrete) DeleteMessage(ctx context.Context, receiptHandle string) error {
+	_, err := svc.SQSClient.DeleteMessage(ctx, &sqs.DeleteMessageInput{
+		QueueUrl:      &svc.QueueURL,
+		ReceiptHandle: &receiptHandle,
+	})
+	return err
+}
+
+// AssertSuccessfulConnection verifies the queue is reachable and readable with
+// the configured credentials. It panics on failure, since a service whose
+// consumer cannot start would silently leave every video stuck "processing".
+func (svc SQSConcrete) AssertSuccessfulConnection(ctx context.Context) {
+	if _, err := svc.SQSClient.GetQueueAttributes(ctx, &sqs.GetQueueAttributesInput{
+		QueueUrl: &svc.QueueURL,
+	}); err != nil {
+		panic(fmt.Errorf("sqs: cannot connect to queue %q: %w", svc.QueueURL, err))
+	}
+}