FEAT: Discovery Video Service Subscription

discovery/internal/api/videos.go

@@ -0,0 +1,13 @@
+package api
+
+// Video is the catalogue's copy of a video, as GET /api/videos/{id} serves it.
+// It is deliberately not cms's video record of the same name: the catalogue
+// publishes what a reader needs to show and play a video, and none of the
+// ingestion detail — no storage key, no transcoding job id, no status, since
+// a video that is not ready never reaches the catalogue at all.
+type Video struct {
+	ID          string   `json:"id" example:"0f8fad5b-d9cb-469f-a165-70867728950e"`
+	Title       string   `json:"title" example:"The Evening Bulletin"`
+	PlaybackURL string   `json:"playbackUrl" example:"https://d111111abcdef8.cloudfront.net/videos/abc123/index.m3u8"`
+	Categories  []string `json:"categories" example:"documentary,news"`
+}

discovery/internal/consumers/catalogue.go

@@ -0,0 +1,110 @@
+// 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"
+	"log"
+	"strings"
+	"thamanyah/discovery/internal/db/repositories"
+	"thamanyah/discovery/internal/models"
+	"thamanyah/discovery/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
+
+// announcement is what cms publishes to the catalogue topic when a video
+// becomes ready. It is cms's published contract, mirrored here rather than
+// imported: the two services share no code, and this struct is exactly the
+// coupling between them — so a field renamed on one side has to be renamed
+// here too, deliberately.
+//
+// The subscription delivers raw, so this is the whole message body; there is
+// no SNS envelope to unwrap.
+type announcement struct {
+	VideoID     string   `json:"videoId"`
+	Title       string   `json:"title"`
+	PlaybackURL string   `json:"playbackUrl"`
+	Categories  []string `json:"categories"`
+}
+
+// CatalogueEvents consumes the announcements cms publishes.
+type CatalogueEvents struct{}
+
+// Run consumes announcements until ctx is cancelled. It is meant to be run in
+// its own goroutine for the lifetime of the process.
+func (c CatalogueEvents) Run(ctx context.Context) {
+	log.Println("catalogue announcements consumer started")
+
+	for {
+		if ctx.Err() != nil {
+			log.Println("catalogue announcements consumer stopped")
+			return
+		}
+
+		messages, err := services.SQSClient.ReceiveMessages(ctx)
+		if err != nil {
+			if ctx.Err() != nil {
+				log.Println("catalogue announcements consumer stopped")
+				return
+			}
+			log.Printf("Something Went Wrong Receiving Announcements: %s", err)
+			time.Sleep(receiveBackoff)
+			continue
+		}
+
+		for _, message := range messages {
+			if !c.handleAnnouncement(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 An Announcement: %s", err)
+			}
+		}
+	}
+}
+
+// handleAnnouncement 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 announcement naming no video) is finished
+// with, however little it accomplished: leaving it on the queue would only
+// stall the announcements behind it.
+func (c CatalogueEvents) handleAnnouncement(ctx context.Context, body string) bool {
+	var announced announcement
+	if err := json.Unmarshal([]byte(body), &announced); err != nil {
+		log.Printf("Discarding An Announcement That Is Not JSON: %s", err)
+		return true
+	}
+
+	if strings.TrimSpace(announced.VideoID) == "" {
+		log.Printf("Discarding An Announcement That Names No Video")
+		return true
+	}
+
+	if err := repositories.VideoRepo.SaveVideo(ctx, models.Video{
+		ID:          announced.VideoID,
+		Title:       announced.Title,
+		PlaybackURL: announced.PlaybackURL,
+		Categories:  announced.Categories,
+	}); err != nil {
+		// Worth another attempt: the save is an upsert, so a redelivery
+		// re-runs it harmlessly.
+		log.Printf("Something Went Wrong Saving An Announced Video: video=%q: %s", announced.VideoID, err)
+		return false
+	}
+
+	log.Printf("Catalogued An Announced Video: video=%q title=%q", announced.VideoID, announced.Title)
+	return true
+}

discovery/internal/db/migrations/0002_create_videos_table.down.sql

@@ -0,0 +1 @@
+DROP TABLE videos;

discovery/internal/db/migrations/0002_create_videos_table.up.sql

@@ -0,0 +1,19 @@
+-- The catalogue's own copy of a video, built from the announcements cms
+-- publishes on the catalogue topic. It is a read model, not a second source of
+-- truth: every column here arrives in an announcement, and nothing in this
+-- service ever writes a video of its own.
+--
+-- id is the id cms issued, carried in the announcement and reused verbatim, so
+-- the two services can talk about the same video without a mapping table. It
+-- is therefore NOT generated here: no DEFAULT, unlike cms's videos.id.
+CREATE TABLE videos (
+    id UUID PRIMARY KEY,
+    title TEXT NOT NULL,
+    playback_url TEXT NOT NULL DEFAULT '',
+    -- Denormalized on purpose. cms owns the category vocabulary and sends
+    -- names, so there is nothing here to join to and no id to keep in step;
+    -- an array is what the read side actually serves.
+    categories TEXT[] NOT NULL DEFAULT '{}',
+    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);

discovery/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 the catalogue holds no video with that id.
+var ErrVideoNotFound = errors.New("video not found")

discovery/internal/db/repositories/videos.go

@@ -0,0 +1,59 @@
+package repositories
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"thamanyah/discovery/internal/models"
+
+	"github.com/lib/pq"
+)
+
+type VideoRepository struct {
+	SQLDB *sql.DB
+}
+
+var VideoRepo VideoRepository
+
+// SaveVideo records an announced video, replacing what the catalogue already
+// held for that id.
+//
+// The upsert is what makes the consumer safe to retry: the catalogue topic
+// delivers at-least-once, so the same announcement can arrive twice, and a
+// plain INSERT would fail the second time on the primary key. Re-announcing a
+// video is also how it is corrected — a later announcement wins, which is why
+// every column is overwritten rather than merged.
+func (svc VideoRepository) SaveVideo(ctx context.Context, v models.Video) error {
+	_, err := svc.SQLDB.ExecContext(ctx, `
+		INSERT INTO videos (id, title, playback_url, categories)
+		VALUES ($1, $2, $3, $4)
+		ON CONFLICT (id) DO UPDATE
+		SET title = EXCLUDED.title,
+		    playback_url = EXCLUDED.playback_url,
+		    categories = EXCLUDED.categories,
+		    updated_at = now()
+	`, v.ID, v.Title, v.PlaybackURL, pq.Array(v.Categories))
+	return err
+}
+
+// GetVideoByID reads the catalogue's copy of one video. It returns
+// ErrVideoNotFound when the catalogue has not heard about it — which, for a
+// video cms has only just made ready, is a matter of timing rather than a
+// mistake.
+func (svc VideoRepository) GetVideoByID(ctx context.Context, id string) (models.Video, error) {
+	var v models.Video
+
+	err := svc.SQLDB.QueryRowContext(ctx, `
+		SELECT id, title, playback_url, categories, created_at, updated_at
+		FROM videos
+		WHERE id = $1
+	`, id).Scan(&v.ID, &v.Title, &v.PlaybackURL, pq.Array(&v.Categories), &v.CreatedAt, &v.UpdatedAt)
+	if errors.Is(err, sql.ErrNoRows) {
+		return models.Video{}, ErrVideoNotFound
+	}
+	if err != nil {
+		return models.Video{}, err
+	}
+
+	return v, nil
+}

discovery/internal/handlers/videos.go

@@ -0,0 +1,49 @@
+package handlers
+
+import (
+	"errors"
+	"log"
+	"net/http"
+	"thamanyah/discovery/internal/api"
+	"thamanyah/discovery/internal/db/repositories"
+)
+
+// GetVideo serves the catalogue's copy of one video.
+//
+//	@Summary		Get a catalogued video
+//	@Description	Returns the catalogue's copy of a video: what a reader needs to show and play it. A video appears here only after the CMS has announced it as ready, so a video that is still transcoding — or one the CMS never made ready — is a 404.
+//	@Tags			videos
+//	@Produce		json
+//	@Param			id	path		string	true	"Video id, as issued by the CMS"
+//	@Success		200	{object}	api.Video
+//	@Failure		404	{object}	api.ProblemDetails
+//	@Failure		500	{object}	api.ProblemDetails
+//	@Router			/api/videos/{id} [get]
+func GetVideo(w http.ResponseWriter, r *http.Request) {
+	id := r.PathValue("id")
+
+	video, err := repositories.VideoRepo.GetVideoByID(r.Context(), id)
+	if errors.Is(err, repositories.ErrVideoNotFound) {
+		writeProblem(w, http.StatusNotFound, "Video Not In The Catalogue",
+			"No video with that id is in the catalogue. A video appears here once the CMS announces it as ready, so one that is still being transcoded is not here yet.")
+		return
+	}
+	if err != nil {
+		log.Printf("Something Went Wrong Loading A Catalogued Video: id=%q: %s", id, err)
+		writeProblem(w, http.StatusInternalServerError, "Something Went Wrong Loading The Catalogue",
+			"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
+	}
+
+	// Translated field by field rather than returned as-is: models.Video is the
+	// row, api.Video is the published schema, and neither should drag the other
+	// along when it changes.
+	writeJSON(w, http.StatusOK, api.Video{
+		ID:          video.ID,
+		Title:       video.Title,
+		PlaybackURL: video.PlaybackURL,
+		// Never null on the wire: a video with no categories has an empty
+		// list, which is a different thing from "unknown".
+		Categories: append([]string{}, video.Categories...),
+	})
+}

discovery/internal/models/video.go

@@ -0,0 +1,18 @@
+package models
+
+import "time"
+
+// Video is the catalogue's copy of a video cms has announced as ready. Every
+// member is filled from the announcement — discovery originates none of it.
+type Video struct {
+	// ID is the id cms issued, reused verbatim so both services name the same
+	// video the same way.
+	ID          string
+	Title       string
+	PlaybackURL string
+	// Categories are names, not ids: cms owns the vocabulary and announces it
+	// by name, so nothing here has to track its numbering.
+	Categories []string
+	CreatedAt  time.Time
+	UpdatedAt  time.Time
+}

discovery/internal/services/sqs.go

@@ -0,0 +1,83 @@
+// Package services holds this service's AWS clients. Discovery reaches exactly
+// one AWS API — the SQS queue subscribed to the catalogue topic — so unlike
+// cms there is no S3 or MediaConvert here, and nothing database-related.
+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 serve a catalogue that silently stops growing.
+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))
+	}
+}