diff --git a/CLAUDE.md b/CLAUDE.md index f81e345..b4f245f 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -13,16 +13,19 @@ This file provides guidance to Claude Code (claude.ai/code) when working with co └── .gitea/workflows/ Gitea Actions CI/CD pipelines ``` -`discovery` is a skeleton, not a feature: it mirrors `cms`'s layout, boot -sequence and Postgres wiring, and serves `GET /health` plus the Swagger UI, but -it has no domain yet — `internal/models` and `internal/db/repositories` are -package doc comments, and migration `0001_init` creates no tables (it exists -only because `internal/db/migrate.go` embeds `migrations/*.sql`, which will not -compile against an empty directory). It has no `internal/services` and no -`internal/consumers`: those are AWS-only in `cms`, and `discovery` has no task -role, so it reaches nothing but its own database. It is deployable — its ECR -repo, ECS service and ALB have always existed in infra, and it now has an image -to put in them. +`discovery` is the read side. It owns no ingestion: it learns what exists by +draining `discovery-catalogue-events`, the queue subscribed to the catalogue +topic `cms` announces ready videos on, and keeps its own copy in its own +database. Neither service reads the other's tables and nothing polls `cms`. + +It serves `GET /health`, `GET /api/videos/{id}` and the Swagger UI. Migration +`0001_init` still creates nothing (it predates the domain and exists only +because `internal/db/migrate.go` embeds `migrations/*.sql`, which will not +compile against an empty directory); `0002_create_videos_table` is where the +schema actually starts. It now has an `internal/services` (SQS only) and an +`internal/consumers`, and a task role scoped to that one queue — receive, +delete, get-attributes, and nothing else. It cannot publish back onto the +topic: it is a subscriber, not a participant. ## Commands @@ -45,7 +48,7 @@ never had a v1. ```bash cd discovery -go run . # serves on :8080 (requires DB_* env vars only — no S3/MediaConvert) +go run . # serves on :8080 (requires DB_* plus AWS_REGION/CATALOGUE_EVENTS_QUEUE_URL — no S3/MediaConvert) go run . migrate # applies pending DB migrations, then exits (no HTTP server) go build ./... go vet ./... @@ -54,9 +57,20 @@ go vet ./... Its OpenAPI spec is generated into `discovery/docs` by the same `swag init` invocation as `cms`, run from `discovery/`. In `docker-compose.yml` it is a service of its own on `127.0.0.1:8080`, wired to the `discovery` database and -role LocalStack provisions — and, like `cms`, started in server mode only, so a -freshly created local database needs `docker exec discovery ./discovery migrate` -once. +role LocalStack provisions, plus the AWS_* vars its consumer needs — and, like +`cms`, started in server mode only, so a freshly created local database needs +`docker exec discovery ./discovery migrate` once. Until that runs, the consumer +logs `relation "videos" does not exist` per announcement and leaves them on the +queue; they are picked up once the table exists. + +`discovery`'s data model is one table, `videos`: the catalogue's copy of an +announced video, keyed by **the id `cms` issued** (no `DEFAULT` — the id +arrives in the announcement and is reused verbatim, so both services name the +same video the same way). `categories` is a denormalized `TEXT[]` of names: +`cms` owns the vocabulary and announces names, so there is nothing to join to. +`repositories.SaveVideo` is an **upsert** — the topic delivers at-least-once, +so the same announcement can arrive twice, and a plain `INSERT` would fail on +the second, wedging every message queued behind it. A later announcement wins. `cms`, `discovery` and `infrastructure` hold no test files of their own. The only tests in the repo are the black-box BDD scenarios in `tests/` — see below. @@ -127,7 +141,7 @@ on the same seams: | `internal/models` | the **domain types** the rest of the code passes around: `Video`, `Category`, `VideoStatus`. No JSON tags, no SQL. | | `internal/handlers` | HTTP: decode `api.*`, validate, call repositories/services, encode `api.*`. | | `internal/db` | the `*sql.DB` connection lifecycle and migrations; `internal/db/repositories` holds all SQL. | -| `internal/services` | **AWS only** — S3 and MediaConvert. Nothing database-related lives here. | +| `internal/services` | **AWS only** — S3, MediaConvert and SQS. Nothing database-related lives here. | `api` and `models` are deliberately separate types even where the fields look alike: `api.CompleteResponse` is the published schema, `models.Video` is the @@ -255,9 +269,27 @@ it ever matters. `repositories.VideoRepo.CreateVideo` persists the `videos` row (status `"processing"`) and its `video_categories` links. 4. Finished output lands in `encoded-bucket`, served via CloudFront. +5. MediaConvert reports the job's state changes to an SNS topic that fans out + to `cms-mediaconvert-events`, which `internal/consumers` reads for the + lifetime of the process: it records the outcome (`ready`/`failed`) and the + playback URL, rewriting the `s3://` playlist path onto `PLAYBACK_BASE_URL`. + A job that came out **ready** is then announced on the `catalogue-events` + SNS topic via `services.CatalogueClient` — `{videoId, title, playbackUrl, + categories}`, categories by **name** so a subscriber needs nothing from + cms to interpret them. Failed jobs are recorded but never announced. A + topic, not a queue, so the announcement fans out: `discovery-catalogue-events` + subscribes today (raw delivery, so there is no SNS envelope to unwrap), and + a second reader can subscribe its own queue without cms changing. Note a + topic only delivers to subscriptions that exist when it publishes, so a new + subscriber's queue has to be in place before the announcement, not after. + Delivery is at-least-once: a failed publish leaves the *job* event on the + consumer's queue, and the outcome update is idempotent, so a redelivery + re-announces and a subscriber can see a video twice. Config wiring: cms reads `S3_BUCKET`, `MEDIACONVERT_INPUT_BUCKET`, `MEDIACONVERT_OUTPUT_BUCKET`, `MEDIACONVERT_ROLE_ARN`, `AWS_REGION`, +`MEDIACONVERT_EVENTS_QUEUE_URL`, `CATALOGUE_EVENTS_TOPIC_ARN`, +`PLAYBACK_BASE_URL`, `DB_HOST`, `DB_PORT`, `DB_NAME`, `DB_USER`, `DB_PASSWORD` from env vars injected by the ECS task definition (`extraEnv` and the unconditional DB_* vars in `deployFargateService`, `infrastructure/main.go`) — `main.go` panics @@ -295,16 +327,23 @@ on boot if any required var is empty. scoped to `PUT` only, from the `cms` ALB's own origin — correct back when cms served the upload page itself, but stale now that it serves no UI (see Known gaps). -- **IAM roles** — four distinct roles/users, each scoped narrowly, don't +- **IAM roles** — five distinct roles/users, each scoped narrowly, don't conflate them: - `ecs-task-execution-role` — shared by both services' ECS *agent* (image pull, log write, Secrets Manager read for DB password). Not usable by application code inside the container. - `cms-task-role` — the `cms` container's own AWS identity: S3 `ListBucket`/`PutObject`/`GetObject` on `raw-uploads-bucket` only, - `mediaconvert:CreateJob`, and `iam:PassRole` scoped to the MediaConvert - service role (`iam:PassedToService` condition). `discovery` has no task - role — it doesn't touch S3 or MediaConvert. + `mediaconvert:CreateJob`, `iam:PassRole` scoped to the MediaConvert + service role (`iam:PassedToService` condition), receive/delete on + `cms-mediaconvert-events`, and `sns:Publish` on the `catalogue-events` + topic — the only thing it writes to. It has no access to + `discovery-catalogue-events`, the queue subscribed to that topic: cms + publishes, it does not reach into a subscriber. + - `discovery-task-role` — the `discovery` container's own AWS identity, and + its only one: receive/delete/get-attributes on + `discovery-catalogue-events`. No S3, no MediaConvert, and no `sns:Publish` + — it consumes the catalogue, it does not add to it. - `mediaconvert-service-role` — trusted by `mediaconvert.amazonaws.com`, not by ECS; the role MediaConvert itself assumes (passed as `CreateJobInput.Role`) to read `raw-uploads-bucket` and write diff --git a/cms/go.mod b/cms/go.mod index 52398ed..99ca37a 100644 --- a/cms/go.mod +++ b/cms/go.mod @@ -3,7 +3,7 @@ module thamanyah/cms/v2 go 1.25.12 require ( - github.com/aws/aws-sdk-go-v2 v1.44.0 + github.com/aws/aws-sdk-go-v2 v1.45.1 github.com/aws/aws-sdk-go-v2/config v1.32.36 github.com/aws/aws-sdk-go-v2/service/mediaconvert v1.97.2 github.com/aws/aws-sdk-go-v2/service/s3 v1.107.1 @@ -20,14 +20,15 @@ require ( github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.17 // indirect github.com/aws/aws-sdk-go-v2/credentials v1.19.35 // indirect github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.36 // indirect - github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.40 // indirect - github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.40 // indirect + github.com/aws/aws-sdk-go-v2/internal/configsources v1.5.1 // indirect + github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.8.1 // indirect github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.37 // indirect github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.16 // indirect github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.29 // indirect github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.36 // indirect github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.37 // indirect github.com/aws/aws-sdk-go-v2/service/signin v1.5.5 // indirect + github.com/aws/aws-sdk-go-v2/service/sns v1.44.1 // indirect github.com/aws/aws-sdk-go-v2/service/sso v1.33.5 // indirect github.com/aws/aws-sdk-go-v2/service/ssooidc v1.38.5 // indirect github.com/aws/aws-sdk-go-v2/service/sts v1.45.5 // indirect diff --git a/cms/go.sum b/cms/go.sum index c88c73f..7eeca87 100644 --- a/cms/go.sum +++ b/cms/go.sum @@ -6,6 +6,8 @@ github.com/Microsoft/go-winio v0.6.2 h1:F2VQgta7ecxGYO8k3ZZz3RS8fVIXVxONVUPlNERo github.com/Microsoft/go-winio v0.6.2/go.mod h1:yd8OoFMLzJbo9gZq8j5qaps8bJ9aShtEA8Ipt1oGCvU= github.com/aws/aws-sdk-go-v2 v1.44.0 h1:4IbaHhtzy+4h37z4JQyO9a2QsiCml3CNYHtq5hIHigo= github.com/aws/aws-sdk-go-v2 v1.44.0/go.mod h1:bttEH6JqnUL8LepvDVfdrds/fZ5bCIxzpe3abyUrhDU= +github.com/aws/aws-sdk-go-v2 v1.45.1 h1:iIoG3NaLhV6UZpPXyPXlDj2I9oS8tV/nMcMnITCC6Ks= +github.com/aws/aws-sdk-go-v2 v1.45.1/go.mod h1:bttEH6JqnUL8LepvDVfdrds/fZ5bCIxzpe3abyUrhDU= github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.17 h1:mn+Vxb9zgz/FE/yDTcFim3DZ1qpcrxR+qBQkBrl6bzA= github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.17/go.mod h1:eDfmEFxu+BSVsUGLbzJhWjpOurv1mqczClS97yI8wdk= github.com/aws/aws-sdk-go-v2/config v1.32.36 h1:mX6ietU7UlB4w/2IUaexJdsyUDvhTd+jYPjVePiyi6s= @@ -16,8 +18,12 @@ github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.36 h1:gucL1KH/PAYbpTpBg09CiV github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.36/go.mod h1:usTB+PHhNMhrx2dxUeHcM7OrT5pySvmjYI++IsefPN0= github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.40 h1:UIXlbijuB2XK1Kr57fo8iIxCuaSHJzwZ1uo+2tbEYIk= github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.40/go.mod h1:wcEsL6jscjZjVUinb0Q5qD/GXOG1yT3GNfmT9HuDwzU= +github.com/aws/aws-sdk-go-v2/internal/configsources v1.5.1 h1:pc138gM1CW+XPc60rEwUlwwuwWFQK16CI1T7v1F9Oec= +github.com/aws/aws-sdk-go-v2/internal/configsources v1.5.1/go.mod h1:1+koxpPIbfBdfzP6vojm5/zTpTQ/micYwlxIiNB3TxI= github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.40 h1:xLQVRDs2NddDmK9BEyh5KSlJ1Gpy5/GIJXrV6WcVGAE= github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.40/go.mod h1:XRXnpFVFGLaEVK+olDdFIM1vNa04ETW452oFGEPUxAo= +github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.8.1 h1:K0JsbZQj+1h208Ro1zHeA4l7bMp0NvRffHQ91q8Ol1s= +github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.8.1/go.mod h1:W3/vL6EtCIatICGy9ab29QhMuae+cOKPWcMxv02CO+Q= github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.37 h1:oyd3ke4V9AhKcRR7rRgxk1VyI+DjK2CBQtbxh3OkdaA= github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.37/go.mod h1:aA9D7SqfG9IC1b7FLD7Iyc8Q4JN0a8gHhNjN4zPlIaI= github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.16 h1:iE4NGbvqUZnHDqddQAauZzCILYtFjOHwRM5MOOKLB5A= @@ -34,6 +40,8 @@ github.com/aws/aws-sdk-go-v2/service/s3 v1.107.1 h1:VUTtUJMuRNMkb/7NIKmd8NQaeQLP github.com/aws/aws-sdk-go-v2/service/s3 v1.107.1/go.mod h1:WvUaO0lP5GNMs1R6cs6qvB3mqo16GLta8yfOuf55Rpc= github.com/aws/aws-sdk-go-v2/service/signin v1.5.5 h1:0VTFBfOgPJrUSpGMgzoi8qLcXF5dbmiBuxpo14eBWUw= github.com/aws/aws-sdk-go-v2/service/signin v1.5.5/go.mod h1:sNZYlBxoohYMBYl47BO/bFtAM6I8HSsPa1qwwPPRGoQ= +github.com/aws/aws-sdk-go-v2/service/sns v1.44.1 h1:dMIcbUQ8fPJPbX9jZV19JtL2lCAgEh1LUNlpe6sdgqE= +github.com/aws/aws-sdk-go-v2/service/sns v1.44.1/go.mod h1:ucBUMGW8avqGmbyQoXyoC6Cgt+WsNBrhL9DA4Bb+jN4= github.com/aws/aws-sdk-go-v2/service/sqs v1.47.0 h1:vNsYthHgT4sUo0KVqpkZlz+8ZDqy/MdlqdvZdP6IoAc= github.com/aws/aws-sdk-go-v2/service/sqs v1.47.0/go.mod h1:FSB4mnod1TCBhs3vp2tWVVGHbqxluzA0Fo6LBOXZByw= github.com/aws/aws-sdk-go-v2/service/sso v1.33.5 h1:jDQARFp1mJ2PEnllQf01nfFXGfWMJ59e0/HCHUTTZCk= diff --git a/cms/internal/consumers/mediaconvert.go b/cms/internal/consumers/mediaconvert.go index 3e8d769..4664e58 100644 --- a/cms/internal/consumers/mediaconvert.go +++ b/cms/internal/consumers/mediaconvert.go @@ -187,7 +187,7 @@ func (c MediaConvertEvents) handleJobEvent(ctx context.Context, body string) boo } } - err := repositories.VideoRepo.UpdateVideoOutcomeByJobID(ctx, event.Detail.JobID, status, playbackURL) + video, err := repositories.VideoRepo.UpdateVideoOutcomeByJobID(ctx, event.Detail.JobID, status, playbackURL) if errors.Is(err, repositories.ErrVideoNotFound) { // Not a fault: the topic carries every job in the account, including // ones this service never submitted. @@ -202,5 +202,29 @@ func (c MediaConvertEvents) handleJobEvent(ctx context.Context, body string) boo } log.Printf("Recorded A Job Outcome: job=%q status=%q playback=%q", event.Detail.JobID, status, playbackURL) + + // Only a video that came out ready is worth showing, so only that one is + // announced. A failed job has already had its status recorded above, and + // the catalogue has no use for it. + if status != models.VideoStatusReady { + return true + } + + if err := services.CatalogueClient.AnnounceVideo(ctx, services.CatalogueVideo{ + VideoID: video.ID, + Title: video.Title, + PlaybackURL: video.PlaybackURL, + Categories: video.CategoryNames, + }); err != nil { + // Worth another attempt: the outcome is already recorded, and the + // update is idempotent, so a redelivery re-runs it and tries the + // announcement again. The cost is that a consumer can see the same + // video twice, which is what at-least-once delivery means anyway. + log.Printf("Something Went Wrong Announcing A Ready Video: video=%q job=%q: %s", + video.ID, event.Detail.JobID, err) + return false + } + + log.Printf("Announced A Ready Video: video=%q playback=%q", video.ID, video.PlaybackURL) return true } diff --git a/cms/internal/db/repositories/videos.go b/cms/internal/db/repositories/videos.go index b347d04..a2b564c 100644 --- a/cms/internal/db/repositories/videos.go +++ b/cms/internal/db/repositories/videos.go @@ -21,25 +21,52 @@ var VideoRepo VideoRepository // is cleared along with the 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) UpdateVideoOutcomeByJobID(ctx context.Context, jobID string, status models.VideoStatus, playbackURL string) error { - result, err := svc.SQLDB.ExecContext(ctx, ` +// +// It returns the video it updated, carrying the members an announcement is +// built from — the row's id and title, and the names of the categories it is +// filed under. The consumer would otherwise have to read back what it just +// wrote, and reading it here keeps that SQL on this side of the seam. +func (svc VideoRepository) UpdateVideoOutcomeByJobID(ctx context.Context, jobID string, status models.VideoStatus, playbackURL string) (models.Video, error) { + updated := models.Video{ + Status: status, + PlaybackURL: playbackURL, + MediaConvertJobID: jobID, + } + + err := svc.SQLDB.QueryRowContext(ctx, ` UPDATE videos SET status = $1, playback_url = $2, updated_at = now() WHERE mediaconvert_job_id = $3 - `, string(status), playbackURL, jobID) + RETURNING id, title + `, string(status), playbackURL, jobID).Scan(&updated.ID, &updated.Title) + if errors.Is(err, sql.ErrNoRows) { + return models.Video{}, ErrVideoNotFound + } if err != nil { - return err + return models.Video{}, err } - affected, err := result.RowsAffected() + rows, err := svc.SQLDB.QueryContext(ctx, ` + SELECT c.name + FROM video_categories vc + JOIN categories c ON c.id = vc.category_id + WHERE vc.video_id = $1 + ORDER BY c.name + `, updated.ID) if err != nil { - return err + return models.Video{}, err } - if affected == 0 { - return ErrVideoNotFound + defer rows.Close() + + for rows.Next() { + var name string + if err := rows.Scan(&name); err != nil { + return models.Video{}, err + } + updated.CategoryNames = append(updated.CategoryNames, name) } - return nil + return updated, rows.Err() } // GetVideoByID reads one video and the ids of the categories it is filed diff --git a/cms/internal/models/video.go b/cms/internal/models/video.go index 7fcd11c..4e79802 100644 --- a/cms/internal/models/video.go +++ b/cms/internal/models/video.go @@ -3,10 +3,14 @@ package models import "time" type Video struct { - ID string - Title string - Description string - CategoryIDs []int16 + ID string + Title string + Description string + CategoryIDs []int16 + // CategoryNames is the same filing read by name rather than id. It is + // filled only where a caller needs categories to be self-describing — + // the catalogue announcement — and is empty elsewhere. + CategoryNames []string Tags string FileName string StorageKey string diff --git a/cms/internal/services/catalogue.go b/cms/internal/services/catalogue.go new file mode 100644 index 0000000..25a5e70 --- /dev/null +++ b/cms/internal/services/catalogue.go @@ -0,0 +1,75 @@ +package services + +import ( + "context" + "encoding/json" + "fmt" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/service/sns" +) + +var CatalogueClient Catalogue + +// CatalogueVideo is what cms announces when a video becomes ready. It is a +// published contract — every subscriber of the catalogue topic decodes it — +// so the JSON names here are as much an API as the ones in internal/api, and +// changing one is a change every subscriber sees. +// +// Categories are named rather than numbered on purpose: the ids are cms's +// own, and a consumer holding a name needs nothing from cms to make sense +// of it. +type CatalogueVideo struct { + VideoID string `json:"videoId"` + Title string `json:"title"` + PlaybackURL string `json:"playbackUrl"` + Categories []string `json:"categories"` +} + +type Catalogue interface { + AnnounceVideo(ctx context.Context, video CatalogueVideo) error +} + +type CatalogueConcrete struct { + SNSClient *sns.Client + TopicARN string +} + +// AnnounceVideo publishes one video to the catalogue topic, which fans it out +// to whatever has subscribed. Delivery is at-least-once: the caller +// republishes when a later step fails, so a subscriber has to tolerate seeing +// the same video twice. +// +// A topic delivers only to the subscriptions that exist at the moment it +// publishes, so a subscriber that wants the backlog needs its queue in place +// before the announcement, not after. +func (svc CatalogueConcrete) AnnounceVideo(ctx context.Context, video CatalogueVideo) error { + // A nil slice marshals to null, which would make "has no categories" and + // "categories unknown" indistinguishable to a reader. An empty list says + // the first, which is what an unfiled video means. + if video.Categories == nil { + video.Categories = []string{} + } + + body, err := json.Marshal(video) + if err != nil { + return err + } + + _, err = svc.SNSClient.Publish(ctx, &sns.PublishInput{ + TopicArn: aws.String(svc.TopicARN), + Message: aws.String(string(body)), + }) + return err +} + +// AssertSuccessfulConnection verifies the topic exists and is readable with +// the configured credentials. It panics on failure: a service that cannot +// announce would transcode videos the catalogue never hears about. +func (svc CatalogueConcrete) AssertSuccessfulConnection(ctx context.Context) { + if _, err := svc.SNSClient.GetTopicAttributes(ctx, &sns.GetTopicAttributesInput{ + TopicArn: aws.String(svc.TopicARN), + }); err != nil { + panic(fmt.Errorf("sns: cannot connect to catalogue topic %q: %w", svc.TopicARN, err)) + } +} diff --git a/cms/main.go b/cms/main.go index 74a7cca..6fa5fef 100644 --- a/cms/main.go +++ b/cms/main.go @@ -15,6 +15,7 @@ import ( "github.com/aws/aws-sdk-go-v2/config" "github.com/aws/aws-sdk-go-v2/service/mediaconvert" "github.com/aws/aws-sdk-go-v2/service/s3" + "github.com/aws/aws-sdk-go-v2/service/sns" "github.com/aws/aws-sdk-go-v2/service/sqs" _ "github.com/lib/pq" httpSwagger "github.com/swaggo/http-swagger/v2" @@ -73,6 +74,13 @@ func runServer() { panic(fmt.Errorf("missing required env var: MEDIACONVERT_EVENTS_QUEUE_URL")) } + // The topic ready videos are announced on, which fans out to whatever has + // subscribed — the read side's queue, today. + catalogueEventsTopicARN := os.Getenv("CATALOGUE_EVENTS_TOPIC_ARN") + if catalogueEventsTopicARN == "" { + panic(fmt.Errorf("missing required env var: CATALOGUE_EVENTS_TOPIC_ARN")) + } + // The public host the encoded output is served from — the CloudFront // distribution in front of the output bucket. MediaConvert reports where // it wrote as an s3:// path into a bucket that blocks public access, so @@ -117,6 +125,14 @@ func runServer() { services.SQSClient = concreteSQSClient + concreteCatalogueClient := &services.CatalogueConcrete{ + SNSClient: sns.NewFromConfig(awsConfig), + TopicARN: catalogueEventsTopicARN, + } + concreteCatalogueClient.AssertSuccessfulConnection(context.Background()) + + services.CatalogueClient = concreteCatalogueClient + concreteDBClient := db.CreateDBConnection(requireDBConnectionString()) defer db.CloseConnection(concreteDBClient) db.AssertSuccessfulConnection(context.Background(), concreteDBClient) diff --git a/discovery/docs/docs.go b/discovery/docs/docs.go index be9ed89..2e65b8e 100644 --- a/discovery/docs/docs.go +++ b/discovery/docs/docs.go @@ -15,6 +15,47 @@ const docTemplate = `{ "host": "{{.Host}}", "basePath": "{{.BasePath}}", "paths": { + "/api/videos/{id}": { + "get": { + "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.", + "produces": [ + "application/json" + ], + "tags": [ + "videos" + ], + "summary": "Get a catalogued video", + "parameters": [ + { + "type": "string", + "description": "Video id, as issued by the CMS", + "name": "id", + "in": "path", + "required": true + } + ], + "responses": { + "200": { + "description": "OK", + "schema": { + "$ref": "#/definitions/api.Video" + } + }, + "404": { + "description": "Not Found", + "schema": { + "$ref": "#/definitions/api.ProblemDetails" + } + }, + "500": { + "description": "Internal Server Error", + "schema": { + "$ref": "#/definitions/api.ProblemDetails" + } + } + } + } + }, "/health": { "get": { "description": "Reports that the service is up and serving. Used as the ALB target group health check; it does not verify the database connection.", @@ -45,6 +86,54 @@ const docTemplate = `{ "example": "ok" } } + }, + "api.ProblemDetails": { + "type": "object", + "properties": { + "detail": { + "type": "string", + "example": "The catalogue 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." + }, + "status": { + "type": "integer", + "example": 500 + }, + "title": { + "type": "string", + "example": "Something Went Wrong Loading The Catalogue" + }, + "type": { + "type": "string", + "example": "about:blank" + } + } + }, + "api.Video": { + "type": "object", + "properties": { + "categories": { + "type": "array", + "items": { + "type": "string" + }, + "example": [ + "documentary", + "news" + ] + }, + "id": { + "type": "string", + "example": "0f8fad5b-d9cb-469f-a165-70867728950e" + }, + "playbackUrl": { + "type": "string", + "example": "https://d111111abcdef8.cloudfront.net/videos/abc123/index.m3u8" + }, + "title": { + "type": "string", + "example": "The Evening Bulletin" + } + } } } }` diff --git a/discovery/docs/swagger.json b/discovery/docs/swagger.json index 1f75fba..a1f2c8f 100644 --- a/discovery/docs/swagger.json +++ b/discovery/docs/swagger.json @@ -8,6 +8,47 @@ }, "basePath": "/", "paths": { + "/api/videos/{id}": { + "get": { + "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.", + "produces": [ + "application/json" + ], + "tags": [ + "videos" + ], + "summary": "Get a catalogued video", + "parameters": [ + { + "type": "string", + "description": "Video id, as issued by the CMS", + "name": "id", + "in": "path", + "required": true + } + ], + "responses": { + "200": { + "description": "OK", + "schema": { + "$ref": "#/definitions/api.Video" + } + }, + "404": { + "description": "Not Found", + "schema": { + "$ref": "#/definitions/api.ProblemDetails" + } + }, + "500": { + "description": "Internal Server Error", + "schema": { + "$ref": "#/definitions/api.ProblemDetails" + } + } + } + } + }, "/health": { "get": { "description": "Reports that the service is up and serving. Used as the ALB target group health check; it does not verify the database connection.", @@ -38,6 +79,54 @@ "example": "ok" } } + }, + "api.ProblemDetails": { + "type": "object", + "properties": { + "detail": { + "type": "string", + "example": "The catalogue 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." + }, + "status": { + "type": "integer", + "example": 500 + }, + "title": { + "type": "string", + "example": "Something Went Wrong Loading The Catalogue" + }, + "type": { + "type": "string", + "example": "about:blank" + } + } + }, + "api.Video": { + "type": "object", + "properties": { + "categories": { + "type": "array", + "items": { + "type": "string" + }, + "example": [ + "documentary", + "news" + ] + }, + "id": { + "type": "string", + "example": "0f8fad5b-d9cb-469f-a165-70867728950e" + }, + "playbackUrl": { + "type": "string", + "example": "https://d111111abcdef8.cloudfront.net/videos/abc123/index.m3u8" + }, + "title": { + "type": "string", + "example": "The Evening Bulletin" + } + } } } } \ No newline at end of file diff --git a/discovery/docs/swagger.yaml b/discovery/docs/swagger.yaml index 7a5b1e2..c3ab21f 100644 --- a/discovery/docs/swagger.yaml +++ b/discovery/docs/swagger.yaml @@ -6,6 +6,41 @@ definitions: example: ok type: string type: object + api.ProblemDetails: + properties: + detail: + example: The catalogue 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. + type: string + status: + example: 500 + type: integer + title: + example: Something Went Wrong Loading The Catalogue + type: string + type: + example: about:blank + type: string + type: object + api.Video: + properties: + categories: + example: + - documentary + - news + items: + type: string + type: array + id: + example: 0f8fad5b-d9cb-469f-a165-70867728950e + type: string + playbackUrl: + example: https://d111111abcdef8.cloudfront.net/videos/abc123/index.m3u8 + type: string + title: + example: The Evening Bulletin + type: string + type: object info: contact: {} description: JSON API for browsing the Thamanyah catalogue. Read-side counterpart @@ -13,6 +48,36 @@ info: title: Thamanyah Discovery API version: "1.0" paths: + /api/videos/{id}: + get: + 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.' + parameters: + - description: Video id, as issued by the CMS + in: path + name: id + required: true + type: string + produces: + - application/json + responses: + "200": + description: OK + schema: + $ref: '#/definitions/api.Video' + "404": + description: Not Found + schema: + $ref: '#/definitions/api.ProblemDetails' + "500": + description: Internal Server Error + schema: + $ref: '#/definitions/api.ProblemDetails' + summary: Get a catalogued video + tags: + - videos /health: get: description: Reports that the service is up and serving. Used as the ALB target diff --git a/discovery/go.mod b/discovery/go.mod index 7aca8c7..032b50d 100644 --- a/discovery/go.mod +++ b/discovery/go.mod @@ -11,6 +11,21 @@ require ( require ( github.com/KyleBanks/depth v1.2.1 // indirect + github.com/aws/aws-sdk-go-v2 v1.45.1 // indirect + github.com/aws/aws-sdk-go-v2/config v1.33.1 // indirect + github.com/aws/aws-sdk-go-v2/credentials v1.20.1 // indirect + github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.19.1 // indirect + github.com/aws/aws-sdk-go-v2/internal/configsources v1.5.1 // indirect + github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.8.1 // indirect + github.com/aws/aws-sdk-go-v2/internal/v4a v1.5.1 // indirect + github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.19 // indirect + github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.14.1 // indirect + github.com/aws/aws-sdk-go-v2/service/signin v1.7.1 // indirect + github.com/aws/aws-sdk-go-v2/service/sqs v1.48.1 // indirect + github.com/aws/aws-sdk-go-v2/service/sso v1.35.1 // indirect + github.com/aws/aws-sdk-go-v2/service/ssooidc v1.40.1 // indirect + github.com/aws/aws-sdk-go-v2/service/sts v1.47.1 // indirect + github.com/aws/smithy-go v1.28.1 // indirect github.com/go-openapi/jsonpointer v0.19.5 // indirect github.com/go-openapi/jsonreference v0.20.0 // indirect github.com/go-openapi/spec v0.20.6 // indirect diff --git a/discovery/go.sum b/discovery/go.sum index d833e7f..10fbae7 100644 --- a/discovery/go.sum +++ b/discovery/go.sum @@ -4,6 +4,36 @@ github.com/KyleBanks/depth v1.2.1 h1:5h8fQADFrWtarTdtDudMmGsC7GPbOAu6RVB3ffsVFHc github.com/KyleBanks/depth v1.2.1/go.mod h1:jzSb9d0L43HxTQfT+oSA1EEp2q+ne2uh6XgeJcm8brE= github.com/Microsoft/go-winio v0.6.2 h1:F2VQgta7ecxGYO8k3ZZz3RS8fVIXVxONVUPlNERoyfY= github.com/Microsoft/go-winio v0.6.2/go.mod h1:yd8OoFMLzJbo9gZq8j5qaps8bJ9aShtEA8Ipt1oGCvU= +github.com/aws/aws-sdk-go-v2 v1.45.1 h1:iIoG3NaLhV6UZpPXyPXlDj2I9oS8tV/nMcMnITCC6Ks= +github.com/aws/aws-sdk-go-v2 v1.45.1/go.mod h1:bttEH6JqnUL8LepvDVfdrds/fZ5bCIxzpe3abyUrhDU= +github.com/aws/aws-sdk-go-v2/config v1.33.1 h1:bq9jze1hQ5YTCLoVxNnbp0T7rglrlOE7N9YsHqjGkEw= +github.com/aws/aws-sdk-go-v2/config v1.33.1/go.mod h1:2A3HQwG4zaL5Tm80rc6RZj8LmWWv4WYT5v8raSz/L7A= +github.com/aws/aws-sdk-go-v2/credentials v1.20.1 h1:Z8GRNEx0u9sDkZOq4PUnN8mjGwbUQGRzMSXpvt3d8xQ= +github.com/aws/aws-sdk-go-v2/credentials v1.20.1/go.mod h1:uBIK00kFo95dnemqfFMTWx0X8YRqsh6ecIoCjjOkZqM= +github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.19.1 h1:YIEBqcqRnpi4Pfv0YHImtgi6czGCwKHANC7SwmUAVD0= +github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.19.1/go.mod h1:imEf0oufgAo8KAkCHhrOdqGEC0YWx1PPBQH82shSxGw= +github.com/aws/aws-sdk-go-v2/internal/configsources v1.5.1 h1:pc138gM1CW+XPc60rEwUlwwuwWFQK16CI1T7v1F9Oec= +github.com/aws/aws-sdk-go-v2/internal/configsources v1.5.1/go.mod h1:1+koxpPIbfBdfzP6vojm5/zTpTQ/micYwlxIiNB3TxI= +github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.8.1 h1:K0JsbZQj+1h208Ro1zHeA4l7bMp0NvRffHQ91q8Ol1s= +github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.8.1/go.mod h1:W3/vL6EtCIatICGy9ab29QhMuae+cOKPWcMxv02CO+Q= +github.com/aws/aws-sdk-go-v2/internal/v4a v1.5.1 h1:yhw5KD1phVyP9vijxOUzDfEtJx+bt+L63k+VfuiYFAA= +github.com/aws/aws-sdk-go-v2/internal/v4a v1.5.1/go.mod h1:ZW2e0d7DYlRxlS9hEiMXE47gTdX5KRN4byUiNbUpG+Q= +github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.19 h1:bAdDl/HkGCcGPoe25ToSHEw23VIxt6CT5fLcg111BKg= +github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.19/go.mod h1:KaUzbLxv4CeSxh6ZCl9B4m7CuFenS8kUEaDs+f/DQr4= +github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.14.1 h1:RmmWQPREQdk9U+PfqeHW3MqZaBaNK7TpV9W3RY+b+7g= +github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.14.1/go.mod h1:0A3W4F+68ZnNk5XcNL/e9HFMwnP8RlEicFfy6eOEDyw= +github.com/aws/aws-sdk-go-v2/service/signin v1.7.1 h1:mdMtSVKdQ3+mzBh+l0ogrFYZVQUCg6pJZOirA2ARsYE= +github.com/aws/aws-sdk-go-v2/service/signin v1.7.1/go.mod h1:9IqUlsJDbUPcg6cgx3WEzXdjrbWzLDQrak0aaSqlTcI= +github.com/aws/aws-sdk-go-v2/service/sqs v1.48.1 h1:jXP3BdVenFa8RfLVH+D2gswrWZHJcgtygKCf22APFqo= +github.com/aws/aws-sdk-go-v2/service/sqs v1.48.1/go.mod h1:d4DToDhLnEofHKvFu4yCF0Be65pZW267COfKOztsZOQ= +github.com/aws/aws-sdk-go-v2/service/sso v1.35.1 h1:B6WFn91tobD6gG4724ONHaqrpKsoETGnv98LHe/yIGM= +github.com/aws/aws-sdk-go-v2/service/sso v1.35.1/go.mod h1:tWuiVBUtPBr8/rgRiYS8Uf85sHcAN+G7XS3D3CEoUh8= +github.com/aws/aws-sdk-go-v2/service/ssooidc v1.40.1 h1:6yeYCWFvgbI2TI3K6jr9LtBNhXgJ7g4xqD+DEiaDDmM= +github.com/aws/aws-sdk-go-v2/service/ssooidc v1.40.1/go.mod h1:naFe83jSMuYkH+QjQPX8n1MLhBkeCFM5Lsnh5m5wz3c= +github.com/aws/aws-sdk-go-v2/service/sts v1.47.1 h1:Sv2xPnRHlThSUtVujYuUBPI/Il8si6UPHXL8DMiB/F0= +github.com/aws/aws-sdk-go-v2/service/sts v1.47.1/go.mod h1:mKo/CzaCz8qytGW70NG4vIIGAx1HXTlb5lHNkC5k3lk= +github.com/aws/smithy-go v1.28.1 h1:R/nXH00c8qcfCzQVELtRw+eLQWtzv+VAIEFJ1/xxXlQ= +github.com/aws/smithy-go v1.28.1/go.mod h1:YE2RhdIuDbA5E5bTdciG9KrW3+TiEONeUWCqxX9i1Fc= github.com/containerd/errdefs v1.0.0 h1:tg5yIfIlQIrxYtu9ajqY42W3lpS19XqdxRQeEwYG8PI= github.com/containerd/errdefs v1.0.0/go.mod h1:+YBYIdtsnF4Iw6nWZhJcqGSg/dwvV7tyJ/kCkyJ2k+M= github.com/containerd/errdefs/pkg v0.3.0 h1:9IKJ06FvyNlexW690DXuQNx2KA2cUJXx151Xdx3ZPPE= diff --git a/discovery/internal/api/videos.go b/discovery/internal/api/videos.go new file mode 100644 index 0000000..6988845 --- /dev/null +++ b/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"` +} diff --git a/discovery/internal/consumers/catalogue.go b/discovery/internal/consumers/catalogue.go new file mode 100644 index 0000000..5ad5bf1 --- /dev/null +++ b/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 +} diff --git a/discovery/internal/db/migrations/0002_create_videos_table.down.sql b/discovery/internal/db/migrations/0002_create_videos_table.down.sql new file mode 100644 index 0000000..85bd277 --- /dev/null +++ b/discovery/internal/db/migrations/0002_create_videos_table.down.sql @@ -0,0 +1 @@ +DROP TABLE videos; diff --git a/discovery/internal/db/migrations/0002_create_videos_table.up.sql b/discovery/internal/db/migrations/0002_create_videos_table.up.sql new file mode 100644 index 0000000..c74c35f --- /dev/null +++ b/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() +); diff --git a/discovery/internal/db/repositories/errors.go b/discovery/internal/db/repositories/errors.go new file mode 100644 index 0000000..0ee82ec --- /dev/null +++ b/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") diff --git a/discovery/internal/db/repositories/videos.go b/discovery/internal/db/repositories/videos.go new file mode 100644 index 0000000..71eaa5c --- /dev/null +++ b/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 +} diff --git a/discovery/internal/handlers/videos.go b/discovery/internal/handlers/videos.go new file mode 100644 index 0000000..e3390c9 --- /dev/null +++ b/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...), + }) +} diff --git a/discovery/internal/models/video.go b/discovery/internal/models/video.go new file mode 100644 index 0000000..552bf38 --- /dev/null +++ b/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 +} diff --git a/discovery/internal/services/sqs.go b/discovery/internal/services/sqs.go new file mode 100644 index 0000000..c4526fe --- /dev/null +++ b/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)) + } +} diff --git a/discovery/main.go b/discovery/main.go index 97e2f09..8c70285 100644 --- a/discovery/main.go +++ b/discovery/main.go @@ -6,9 +6,14 @@ import ( "log" "net/http" "os" + "thamanyah/discovery/internal/consumers" "thamanyah/discovery/internal/db" + "thamanyah/discovery/internal/db/repositories" "thamanyah/discovery/internal/handlers" + "thamanyah/discovery/internal/services" + awsconfig "github.com/aws/aws-sdk-go-v2/config" + "github.com/aws/aws-sdk-go-v2/service/sqs" _ "github.com/lib/pq" httpSwagger "github.com/swaggo/http-swagger/v2" @@ -38,13 +43,44 @@ func runMigrate() { } func runServer() { + // The queue subscribed to cms's catalogue topic. This is the only AWS + // service discovery talks to. + catalogueQueueURL := os.Getenv("CATALOGUE_EVENTS_QUEUE_URL") + if catalogueQueueURL == "" { + panic(fmt.Errorf("missing required env var: CATALOGUE_EVENTS_QUEUE_URL")) + } + + awsConfig, err := awsconfig.LoadDefaultConfig(context.Background()) + if err != nil { + panic(err) + } + + concreteSQSClient := &services.SQSConcrete{ + SQSClient: sqs.NewFromConfig(awsConfig), + QueueURL: catalogueQueueURL, + } + concreteSQSClient.AssertSuccessfulConnection(context.Background()) + + services.SQSClient = concreteSQSClient + concreteDBClient := db.CreateDBConnection(requireDBConnectionString()) defer db.CloseConnection(concreteDBClient) db.AssertSuccessfulConnection(context.Background(), concreteDBClient) + repositories.VideoRepo = repositories.VideoRepository{ + SQLDB: concreteDBClient, + } + + // Consumes announcements for the lifetime of the process. Cancelled when + // runServer returns, which today only happens if the server itself fails. + consumerCtx, stopConsumer := context.WithCancel(context.Background()) + defer stopConsumer() + go consumers.CatalogueEvents{}.Run(consumerCtx) + mux := http.NewServeMux() mux.HandleFunc("GET /health", handlers.Health) + mux.HandleFunc("GET /api/videos/{id}", handlers.GetVideo) // Swagger UI and the generated spec. The UI assets are embedded in the // binary by swaggo/files, so this needs no static directory on disk. diff --git a/docker-compose.yml b/docker-compose.yml index bb0d6c0..42dbba9 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -55,7 +55,7 @@ services: environment: - PULUMI_CONFIG_PASSPHRASE=${PULUMI_CONFIG_PASSPHRASE:-local} - PULUMI_SKIP_UPDATE_CHECK=true - - DB_PASSWORD=${DB_PASSWORD:?} + - DB_PASSWORD=example_password volumes: - "./infrastructure:/infra" - "pulumi-state:/state" # stack state, kept out of the repo @@ -100,6 +100,9 @@ services: # of the process. Pinned by the localstack branch in infrastructure/main.go; # LocalStack always uses account 000000000000. - MEDIACONVERT_EVENTS_QUEUE_URL=${MEDIACONVERT_EVENTS_QUEUE_URL:-http://localhost.localstack.cloud:4566/000000000000/cms-mediaconvert-events} + # The topic ready videos are announced on, which fans out to the read + # side's queue. Pinned by the localstack branch in infrastructure/main.go. + - CATALOGUE_EVENTS_TOPIC_ARN=${CATALOGUE_EVENTS_TOPIC_ARN:-arn:aws:sns:us-east-1:000000000000:catalogue-events} # Where finished HLS output is served from. In AWS this is the CloudFront # distribution; there is none under LocalStack, so the encoded bucket is # addressed directly — path-style, for the same reason S3 is elsewhere. @@ -110,7 +113,7 @@ services: - DB_PORT=${DB_PORT:-4510} - DB_NAME=${DB_NAME:-cms} - DB_USER=${DB_USER:-cms} - - DB_PASSWORD=${DB_PASSWORD:?} + - DB_PASSWORD=example_password - DB_SSLMODE=${DB_SSLMODE:-disable} depends_on: infra: @@ -131,15 +134,22 @@ services: ports: - "127.0.0.1:8080:8080" # JSON API + Swagger UI at /swagger/ environment: - # Postgres only — discovery touches no AWS service, which is why it has - # no task role in infrastructure/main.go and no AWS_* wiring here. Its - # own role and database, provisioned alongside the cms ones by - # newServiceDatabase, so the two services share no credentials. + # AWS: discovery reaches exactly one AWS API — the queue subscribed to + # cms's catalogue topic, which it drains for the lifetime of the process. + # That is why it now has a task role in infrastructure/main.go, scoped to + # this queue and nothing else. + - AWS_REGION=${AWS_REGION:-us-east-1} + - AWS_ACCESS_KEY_ID=${AWS_ACCESS_KEY_ID:-test} + - AWS_SECRET_ACCESS_KEY=${AWS_SECRET_ACCESS_KEY:-test} + - AWS_ENDPOINT_URL=${AWS_ENDPOINT_URL:-http://localhost.localstack.cloud:4566} + - CATALOGUE_EVENTS_QUEUE_URL=${CATALOGUE_EVENTS_QUEUE_URL:-http://localhost.localstack.cloud:4566/000000000000/discovery-catalogue-events} + # Postgres: its own role and database, provisioned alongside the cms ones + # by newServiceDatabase, so the two services share no credentials. - DB_HOST=${DISCOVERY_DB_HOST:-localhost.localstack.cloud} - DB_PORT=${DISCOVERY_DB_PORT:-4510} - DB_NAME=${DISCOVERY_DB_NAME:-discovery} - DB_USER=${DISCOVERY_DB_USER:-discovery} - - DB_PASSWORD=${DB_PASSWORD:?} + - DB_PASSWORD=example_password - DB_SSLMODE=${DISCOVERY_DB_SSLMODE:-disable} depends_on: infra: diff --git a/infrastructure/main.go b/infrastructure/main.go index 0649aab..c2440f5 100644 --- a/infrastructure/main.go +++ b/infrastructure/main.go @@ -854,6 +854,158 @@ func main() { return err } + // The catalogue topic: where cms announces a video once its transcode + // has finished. A topic rather than a queue so the announcement fans + // out — the read side subscribes the queue below, and a second reader + // (search indexing, notifications) can subscribe its own without cms + // knowing it exists or this wiring changing. + catalogueTopicArgs := &sns.TopicArgs{} + catalogueQueueArgs := &sqs.QueueArgs{} + catalogueDeadLetterArgs := &sqs.QueueArgs{} + if localstack { + // docker-compose.yml and the tests/ suite refer to these literally. + catalogueTopicArgs.Name = pulumi.String("catalogue-events") + catalogueQueueArgs.Name = pulumi.String("discovery-catalogue-events") + catalogueDeadLetterArgs.Name = pulumi.String("discovery-catalogue-events-dlq") + } + + catalogueTopic, err := sns.NewTopic(ctx, "catalogue-events", catalogueTopicArgs) + if err != nil { + return err + } + + catalogueDeadLetter, err := sqs.NewQueue(ctx, "discovery-catalogue-events-dlq", catalogueDeadLetterArgs) + if err != nil { + return err + } + + catalogueQueueArgs.VisibilityTimeoutSeconds = pulumi.Int(60) + catalogueQueueArgs.RedrivePolicy = catalogueDeadLetter.Arn.ApplyT(func(arn string) (string, error) { + b, err := json.Marshal(map[string]any{ + "deadLetterTargetArn": arn, + "maxReceiveCount": 5, + }) + return string(b), err + }).(pulumi.StringOutput) + + // The read side's own queue on the catalogue topic. It exists before + // discovery has code to drain it so the announcements pile up rather + // than being dropped on the floor — a topic delivers only to the + // subscriptions that exist when it publishes. + catalogueQueue, err := sqs.NewQueue(ctx, "discovery-catalogue-events", catalogueQueueArgs) + if err != nil { + return err + } + + // SNS is not an IAM principal the queue trusts by default; without this + // the subscription is created and silently delivers nothing. + catalogueQueuePolicy := pulumi.All(catalogueQueue.Arn, catalogueTopic.Arn).ApplyT( + func(args []any) (string, error) { + queueArn := args[0].(string) + topicArn := args[1].(string) + + doc := map[string]any{ + "Version": "2012-10-17", + "Statement": []map[string]any{ + { + "Sid": "AllowCatalogueTopicToSend", + "Effect": "Allow", + "Principal": map[string]string{"Service": "sns.amazonaws.com"}, + "Action": "sqs:SendMessage", + "Resource": queueArn, + "Condition": map[string]any{ + "ArnEquals": map[string]string{"aws:SourceArn": topicArn}, + }, + }, + }, + } + b, err := json.Marshal(doc) + return string(b), err + }, + ).(pulumi.StringOutput) + + _, err = sqs.NewQueuePolicy(ctx, "discovery-catalogue-events-queue-policy", &sqs.QueuePolicyArgs{ + QueueUrl: catalogueQueue.ID(), + Policy: catalogueQueuePolicy, + }) + if err != nil { + return err + } + + _, err = sns.NewTopicSubscription(ctx, "discovery-catalogue-events-subscription", &sns.TopicSubscriptionArgs{ + Topic: catalogueTopic.Arn, + Protocol: pulumi.String("sqs"), + Endpoint: catalogueQueue.Arn, + // The queue receives the announcement itself rather than an SNS + // envelope carrying it as a JSON string, so a consumer parses one + // document instead of unwrapping two. + RawMessageDelivery: pulumi.Bool(true), + }) + if err != nil { + return err + } + + // A second subscriber, for the tests/ suite only. The BDD scenarios + // assert on the announcement itself — the field names cms publishes, + // which no assertion through discovery's API can pin — and a consumer + // destroys what it reads. Sharing discovery's queue would mean the two + // racing for every message and each seeing about half, so the suite + // gets its own subscription. This is what the topic is for. + // + // LocalStack only: in AWS this queue would fill up unread, since + // nothing runs the suite there. + if localstack { + testsCatalogueQueue, err := sqs.NewQueue(ctx, "tests-catalogue-events", &sqs.QueueArgs{ + Name: pulumi.String("tests-catalogue-events"), + }) + if err != nil { + return err + } + + testsCatalogueQueuePolicy := pulumi.All(testsCatalogueQueue.Arn, catalogueTopic.Arn).ApplyT( + func(args []any) (string, error) { + queueArn := args[0].(string) + topicArn := args[1].(string) + + doc := map[string]any{ + "Version": "2012-10-17", + "Statement": []map[string]any{ + { + "Sid": "AllowCatalogueTopicToSend", + "Effect": "Allow", + "Principal": map[string]string{"Service": "sns.amazonaws.com"}, + "Action": "sqs:SendMessage", + "Resource": queueArn, + "Condition": map[string]any{ + "ArnEquals": map[string]string{"aws:SourceArn": topicArn}, + }, + }, + }, + } + b, err := json.Marshal(doc) + return string(b), err + }, + ).(pulumi.StringOutput) + + _, err = sqs.NewQueuePolicy(ctx, "tests-catalogue-events-queue-policy", &sqs.QueuePolicyArgs{ + QueueUrl: testsCatalogueQueue.ID(), + Policy: testsCatalogueQueuePolicy, + }) + if err != nil { + return err + } + + _, err = sns.NewTopicSubscription(ctx, "tests-catalogue-events-subscription", &sns.TopicSubscriptionArgs{ + Topic: catalogueTopic.Arn, + Protocol: pulumi.String("sqs"), + Endpoint: testsCatalogueQueue.Arn, + RawMessageDelivery: pulumi.Bool(true), + }) + if err != nil { + return err + } + } + // SNS is not an IAM principal the queue trusts by default; without this // the subscription is created and silently delivers nothing. jobEventsQueuePolicy := pulumi.All(jobEventsQueue.Arn, jobEventsTopic.Arn).ApplyT( @@ -969,11 +1121,12 @@ func main() { return err } - cmsTaskPolicy := pulumi.All(rawUploadsBucket.Arn, mediaConvertRole.Arn, jobEventsQueue.Arn).ApplyT( + cmsTaskPolicy := pulumi.All(rawUploadsBucket.Arn, mediaConvertRole.Arn, jobEventsQueue.Arn, catalogueTopic.Arn).ApplyT( func(args []any) (string, error) { rawUploadsArn := args[0].(string) mediaConvertRoleArn := args[1].(string) jobEventsQueueArn := args[2].(string) + catalogueTopicArn := args[3].(string) doc := map[string]any{ "Version": "2012-10-17", @@ -1009,6 +1162,15 @@ func main() { }, "Resource": jobEventsQueueArn, }, + { + "Sid": "AnnounceReadyVideos", + "Effect": "Allow", + "Action": []string{ + "sns:Publish", + "sns:GetTopicAttributes", + }, + "Resource": catalogueTopicArn, + }, { "Sid": "PassMediaConvertRole", "Effect": "Allow", @@ -1035,6 +1197,45 @@ func main() { return err } + // discovery's own ECS task role. It used to have none — it touched no + // AWS API at all — and it still gets nothing but the one queue it + // drains: no S3, no MediaConvert, and no ability to publish back onto + // the catalogue topic. It is a subscriber, not a participant. + discoveryTaskRole, err := iam.NewRole(ctx, "discovery-task-role", &iam.RoleArgs{ + AssumeRolePolicy: pulumi.String(execRoleAssumePolicy), + }) + if err != nil { + return err + } + + discoveryTaskPolicy := catalogueQueue.Arn.ApplyT(func(catalogueQueueArn string) (string, error) { + doc := map[string]any{ + "Version": "2012-10-17", + "Statement": []map[string]any{ + { + "Sid": "ConsumeCatalogueAnnouncements", + "Effect": "Allow", + "Action": []string{ + "sqs:ReceiveMessage", + "sqs:DeleteMessage", + "sqs:GetQueueAttributes", + }, + "Resource": catalogueQueueArn, + }, + }, + } + b, err := json.Marshal(doc) + return string(b), err + }).(pulumi.StringOutput) + + _, err = iam.NewRolePolicy(ctx, "discovery-task-role-policy", &iam.RolePolicyArgs{ + Role: discoveryTaskRole.ID(), + Policy: discoveryTaskPolicy, + }) + if err != nil { + return err + } + // The two ECS services and the CI user that deploys them. Under // LocalStack the equivalent of all this is the cms service in // docker-compose.yml, which reads the same env vars from .env. @@ -1046,6 +1247,7 @@ func main() { {Name: "MEDIACONVERT_OUTPUT_BUCKET", Value: bucket.ID().ToStringOutput()}, {Name: "MEDIACONVERT_ROLE_ARN", Value: mediaConvertRole.Arn}, {Name: "MEDIACONVERT_EVENTS_QUEUE_URL", Value: jobEventsQueue.Url}, + {Name: "CATALOGUE_EVENTS_TOPIC_ARN", Value: catalogueTopic.Arn}, // The CDN in front of the encoded bucket, which is the only way // that bucket is readable — it blocks public access and grants // only CloudFront's OAC principal. cms rewrites the s3:// paths @@ -1093,8 +1295,13 @@ func main() { // discovery embeds its own golang-migrate migrations and applies // them through `./discovery migrate`, so ECS runs that container to // completion before the service accepts traffic. + discoveryExtraEnv := []envVar{ + {Name: "AWS_REGION", Value: pulumi.String(awsRegion).ToStringOutput()}, + {Name: "CATALOGUE_EVENTS_QUEUE_URL", Value: catalogueQueue.Url}, + } + discoveryRepo, discoveryService, discoveryAlb, err = deployFargateService(ctx, "discovery", 8080, - cluster, execRole, nil, nil, vpcID, subnetIDs, albSecurityGroup, serviceSecurityGroup, + cluster, execRole, discoveryTaskRole, discoveryExtraEnv, vpcID, subnetIDs, albSecurityGroup, serviceSecurityGroup, db.Address, db.Port, discoveryPassword, true) if err != nil { return err diff --git a/tests/README.md b/tests/README.md index 374f59e..b24ea8c 100644 --- a/tests/README.md +++ b/tests/README.md @@ -12,7 +12,7 @@ The upload to storage is a plain `PUT` to the presigned URL the API hands out, with no AWS SDK involved, because that is exactly what a real client does. ``` -features/video_upload.feature the scenarios, in Gherkin +features/*.feature the scenarios, in Gherkin suite_test.go the gobdd suite: step registration + per-scenario world steps_test.go what each step does client_test.go HTTP plumbing and the wire types @@ -68,6 +68,26 @@ Both steps of the upload flow, end to end: the category lookup itself failing produces - answering 409 for a key that was already registered, rather than queueing a second transcoding job for the same file +- announcing a video on the catalogue topic once its job comes back COMPLETE, + carrying its id, title, playback URL and the **names** of its categories — + and announcing nothing for a job that ended in an error + +- the read side building its catalogue from those announcements: the video + turns up in discovery under the id cms issued, a video still transcoding is + not there at all, and a video announced a second time is updated rather than + colliding with its own row + +The publication scenarios read `tests-catalogue-events`, the suite's **own** +queue on the catalogue topic — not discovery's. A consumer destroys what it +reads, so sharing discovery's queue would have the suite and the running +service race for every announcement and each see about half. Delivery is raw, +so a message body is the announcement with no SNS envelope. + +The ingestion scenarios need `discovery` up as well as `cms`; they read it over +HTTP at `DISCOVERY_BASE_URL` (default `http://localhost:8080`). +Like the job-state publisher they stand in for AWS, so they are the second +place the suite reaches for the AWS SDK; everything a real client does still +goes over plain HTTP. ## Adding a scenario diff --git a/tests/client_test.go b/tests/client_test.go index 3d56f09..fd4dfcf 100644 --- a/tests/client_test.go +++ b/tests/client_test.go @@ -23,6 +23,19 @@ func baseURL() string { return defaultBaseURL } +// defaultDiscoveryBaseURL is where docker-compose publishes discovery on the +// host. +const defaultDiscoveryBaseURL = "http://localhost:8080" + +// discoveryBaseURL is the Discovery service the scenarios read the catalogue +// from. Override it the same way as CMS_BASE_URL. +func discoveryBaseURL() string { + if v := strings.TrimSpace(os.Getenv("DISCOVERY_BASE_URL")); v != "" { + return strings.TrimRight(v, "/") + } + return defaultDiscoveryBaseURL +} + // response is one HTTP exchange, kept whole so a failing assertion can print // the body it actually got rather than just a status code. type response struct { @@ -107,6 +120,25 @@ func newClient() *client { } } +// newDiscoveryClient is newClient pointed at the read side. Same plumbing — +// only the host differs, since both services speak the same JSON dialect. +func newDiscoveryClient() *client { + return &client{ + base: discoveryBaseURL(), + http: &http.Client{Timeout: 30 * time.Second}, + } +} + +// catalogueVideoBody is GET /api/videos/{id} on discovery: the catalogue's own +// copy of a video, which is a different published shape from the CMS record of +// the same name. +type catalogueVideoBody struct { + ID string `json:"id"` + Title string `json:"title"` + PlaybackURL string `json:"playbackUrl"` + Categories []string `json:"categories"` +} + // postJSON marshals payload and posts it to a path on the CMS. func (c *client) postJSON(path string, payload any) (response, error) { encoded, err := json.Marshal(payload) diff --git a/tests/events_test.go b/tests/events_test.go index 6110501..680e623 100644 --- a/tests/events_test.go +++ b/tests/events_test.go @@ -13,6 +13,7 @@ import ( 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/sns" + "github.com/aws/aws-sdk-go-v2/service/sqs" ) // These scenarios stand in for EventBridge, not for a client: they publish the @@ -30,6 +31,15 @@ const ( defaultRegion = "us-east-1" defaultOutputBucket = "encoded-bucket" defaultPlaybackBaseURL = "http://localhost.localstack.cloud:4566/encoded-bucket" + // The suite's own queue on the catalogue topic cms announces ready videos + // to. The suite reads a queue rather than the topic because that is what a + // subscriber sees — SNS has nothing to poll. + // + // Deliberately not discovery's queue: a consumer destroys what it reads, so + // sharing one would have the suite and the running service race for every + // announcement and each see about half. Pinned by the localstack branch in + // infrastructure/main.go; LocalStack always uses account 000000000000. + defaultCatalogueQueueURL = "http://localhost.localstack.cloud:4566/000000000000/tests-catalogue-events" ) // playbackBaseURL is the public host the encoded output is served from — the @@ -97,21 +107,27 @@ type eventPublisher struct { topicARN string } -func newEventPublisher(ctx context.Context) (*eventPublisher, error) { - endpoint := envOr("AWS_ENDPOINT_URL", defaultEndpointURL) - region := envOr("AWS_REGION", defaultRegion) - - cfg, err := awsconfig.LoadDefaultConfig(ctx, - awsconfig.WithRegion(region), +// awsConfig is the SDK configuration both ends of the event seam use: the +// publisher that stands in for EventBridge, and the reader that stands in for +// whoever consumes the catalogue queue. +func awsConfig(ctx context.Context) (aws.Config, error) { + return awsconfig.LoadDefaultConfig(ctx, + awsconfig.WithRegion(envOr("AWS_REGION", defaultRegion)), // LocalStack accepts any credentials; these keep the SDK from hunting // for a profile that a developer's machine may not have. awsconfig.WithCredentialsProvider(credentials.NewStaticCredentialsProvider( envOr("AWS_ACCESS_KEY_ID", "test"), envOr("AWS_SECRET_ACCESS_KEY", "test"), "")), ) +} + +func newEventPublisher(ctx context.Context) (*eventPublisher, error) { + cfg, err := awsConfig(ctx) if err != nil { return nil, err } + endpoint := envOr("AWS_ENDPOINT_URL", defaultEndpointURL) + return &eventPublisher{ sns: sns.NewFromConfig(cfg, func(o *sns.Options) { o.BaseEndpoint = aws.String(endpoint) }), topicARN: envOr("MEDIACONVERT_EVENTS_TOPIC_ARN", defaultTopicARN), @@ -160,3 +176,84 @@ func (p *eventPublisher) publishJobState(ctx context.Context, jobID, state, play } return nil } + +// --- the catalogue queue ---------------------------------------------------- + +// catalogueAnnouncement is what cms puts on the catalogue queue when a video +// becomes ready. This is the published contract, so the suite spells it out +// rather than importing it: a change here is a change other services see. +type catalogueAnnouncement struct { + VideoID string `json:"videoId"` + Title string `json:"title"` + PlaybackURL string `json:"playbackUrl"` + Categories []string `json:"categories"` +} + +// catalogueReader stands in for the read side, draining its own queue on the +// catalogue topic. It consumes what it reads, exactly as a real subscriber +// would. The subscription delivers raw, so a message body is the announcement +// itself with no SNS envelope to unwrap. +type catalogueReader struct { + sqs *sqs.Client + queueURL string +} + +func newCatalogueReader(ctx context.Context) (*catalogueReader, error) { + cfg, err := awsConfig(ctx) + if err != nil { + return nil, err + } + + endpoint := envOr("AWS_ENDPOINT_URL", defaultEndpointURL) + + return &catalogueReader{ + sqs: sqs.NewFromConfig(cfg, func(o *sqs.Options) { o.BaseEndpoint = aws.String(endpoint) }), + queueURL: envOr("CATALOGUE_QUEUE_URL", defaultCatalogueQueueURL), + }, nil +} + +// await waits for the announcement naming videoID and returns it. Messages for +// other videos are consumed and discarded on the way: scenarios run one at a +// time, so anything else on the queue is a leftover, and leaving it would only +// make the next scenario wade through it. It returns nil if nothing names that +// video before the deadline. +func (r *catalogueReader) await(ctx context.Context, videoID string, within time.Duration) (*catalogueAnnouncement, error) { + deadline := time.Now().Add(within) + + for time.Now().Before(deadline) { + output, err := r.sqs.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{ + QueueUrl: aws.String(r.queueURL), + MaxNumberOfMessages: 10, + WaitTimeSeconds: 2, + }) + if err != nil { + return nil, fmt.Errorf("receiving from %s: %w", r.queueURL, err) + } + + var found *catalogueAnnouncement + for _, message := range output.Messages { + if message.Body == nil || message.ReceiptHandle == nil { + continue + } + + var announcement catalogueAnnouncement + if err := json.Unmarshal([]byte(*message.Body), &announcement); err == nil && + announcement.VideoID == videoID { + found = &announcement + } + + if _, err := r.sqs.DeleteMessage(ctx, &sqs.DeleteMessageInput{ + QueueUrl: aws.String(r.queueURL), + ReceiptHandle: message.ReceiptHandle, + }); err != nil { + return nil, fmt.Errorf("acknowledging a message on %s: %w", r.queueURL, err) + } + } + + if found != nil { + return found, nil + } + } + + return nil, nil +} diff --git a/tests/features/catalogue_ingestion.feature b/tests/features/catalogue_ingestion.feature new file mode 100644 index 0000000..34e2726 --- /dev/null +++ b/tests/features/catalogue_ingestion.feature @@ -0,0 +1,40 @@ +Feature: Building the catalogue from announcements + Discovery is the read side. It owns no ingestion of its own: it learns what + exists by subscribing to the catalogue topic cms announces ready videos on, + and keeps its own copy in its own database. Nothing polls cms and neither + service reads the other's tables. + + A video therefore reaches the catalogue only once its transcode has finished + — which is the point, since a catalogue entry nobody can play is no use. + + Background: + Given the CMS API is available + And the Discovery API is available + + Scenario: A video the CMS announces appears in the catalogue + Given I have registered a video that is being transcoded + When MediaConvert reports that the job reached "COMPLETE" + Then the catalogue eventually holds that video + And the catalogue's copy carries its title, playback URL and categories + + # The catalogue is built from announcements, and cms announces only what came + # out ready. A video part-way through its transcode is therefore genuinely + # absent rather than merely late, and saying so is the honest answer. + Scenario: A video that is still being transcoded is not in the catalogue + Given I have registered a video that is being transcoded + Then the catalogue does not hold that video + And the request is rejected with status 404 + And the problem title is "Video Not In The Catalogue" + + # The catalogue topic delivers at-least-once, and a job's outcome can be + # reported more than once, so the same video can be announced again. The + # later announcement is the truth: it lands on top of what is already there + # rather than colliding with it. Announcing it again with no playlist is what + # makes that visible — a second announcement carrying identical content is + # indistinguishable from never having arrived. + Scenario: A video announced again is updated rather than colliding + Given I have registered a video that is being transcoded + And MediaConvert reports that the job reached "COMPLETE" + And the catalogue eventually holds that video + When MediaConvert reports that the job reached "COMPLETE" without a playlist + Then the catalogue eventually holds that video with no playback URL diff --git a/tests/features/catalogue_publication.feature b/tests/features/catalogue_publication.feature new file mode 100644 index 0000000..4afa838 --- /dev/null +++ b/tests/features/catalogue_publication.feature @@ -0,0 +1,43 @@ +Feature: Announcing ready videos to the catalogue + A video is only worth showing once its transcode has finished. When + MediaConvert reports a job complete, cms announces that video on a queue of + its own — the seam the read side reads to build the public catalogue, so + nothing downstream has to poll cms or reach into its database. + + The announcement is the published contract: the video's id, its title, the + URL a player can open, and the categories it is filed under, named rather + than numbered so a reader needs nothing from cms to make sense of it. + + Background: + Given the CMS API is available + + Scenario: A finished job announces the video to the catalogue + Given I have registered a video that is being transcoded + When MediaConvert reports that the job reached "COMPLETE" + Then the video eventually has status "ready" + And the catalogue is told about the video + And the announcement carries the video's title, playback URL and categories + + # The catalogue is a list of things worth showing. A job that ended in an + # error produced nothing to play, so it is recorded against the video and + # goes no further. + Scenario Outline: A job that did not finish is not announced + Given I have registered a video that is being transcoded + When MediaConvert reports that the job reached "" + Then the video eventually has status "failed" + And the catalogue is told nothing about the video + + Examples: + | state | + | ERROR | + | CANCELED | + + # A video can be filed under several categories, so the announcement carries + # all of them. Names, not the ids the register call took: what cms numbers + # them is its own business. + Scenario: The announcement lists every category the video is filed under + Given I have registered a video being transcoded under categories "documentary, news, podcast" + When MediaConvert reports that the job reached "COMPLETE" + Then the video eventually has status "ready" + And the catalogue is told about the video + And the announcement files it under "documentary, news, podcast" diff --git a/tests/go.mod b/tests/go.mod index 24cd7da..bd8e2e7 100644 --- a/tests/go.mod +++ b/tests/go.mod @@ -2,20 +2,23 @@ module thamanyah/tests go 1.25.12 -require github.com/go-bdd/gobdd v1.1.4 +require ( + github.com/aws/aws-sdk-go-v2 v1.45.1 + github.com/aws/aws-sdk-go-v2/config v1.32.40 + github.com/aws/aws-sdk-go-v2/credentials v1.19.39 + github.com/aws/aws-sdk-go-v2/service/sns v1.43.0 + github.com/aws/aws-sdk-go-v2/service/sqs v1.48.1 + github.com/go-bdd/gobdd v1.1.4 +) require ( - github.com/aws/aws-sdk-go-v2 v1.44.0 // indirect - github.com/aws/aws-sdk-go-v2/config v1.32.40 // indirect - github.com/aws/aws-sdk-go-v2/credentials v1.19.39 // indirect github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.40 // indirect - github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.40 // indirect - github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.40 // indirect + github.com/aws/aws-sdk-go-v2/internal/configsources v1.5.1 // indirect + github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.8.1 // indirect github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.41 // indirect github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.19 // indirect github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.40 // indirect github.com/aws/aws-sdk-go-v2/service/signin v1.6.0 // indirect - github.com/aws/aws-sdk-go-v2/service/sns v1.43.0 // indirect github.com/aws/aws-sdk-go-v2/service/sso v1.34.0 // indirect github.com/aws/aws-sdk-go-v2/service/ssooidc v1.39.0 // indirect github.com/aws/aws-sdk-go-v2/service/sts v1.46.0 // indirect diff --git a/tests/go.sum b/tests/go.sum index 3c89b30..8fb54e3 100644 --- a/tests/go.sum +++ b/tests/go.sum @@ -1,15 +1,15 @@ -github.com/aws/aws-sdk-go-v2 v1.44.0 h1:4IbaHhtzy+4h37z4JQyO9a2QsiCml3CNYHtq5hIHigo= -github.com/aws/aws-sdk-go-v2 v1.44.0/go.mod h1:bttEH6JqnUL8LepvDVfdrds/fZ5bCIxzpe3abyUrhDU= +github.com/aws/aws-sdk-go-v2 v1.45.1 h1:iIoG3NaLhV6UZpPXyPXlDj2I9oS8tV/nMcMnITCC6Ks= +github.com/aws/aws-sdk-go-v2 v1.45.1/go.mod h1:bttEH6JqnUL8LepvDVfdrds/fZ5bCIxzpe3abyUrhDU= github.com/aws/aws-sdk-go-v2/config v1.32.40 h1:lAVC9gMmKusmqDRe32dPtgKl/BWvJmMJoWELKHCAObw= github.com/aws/aws-sdk-go-v2/config v1.32.40/go.mod h1:8xOJLbe/hOj1g4PVsfJYV7O2byq+UGET1onDdUgbwqc= github.com/aws/aws-sdk-go-v2/credentials v1.19.39 h1:XOg8LC3Kgnsa3WiPQjc7Bi8k5IBN92cPYfIV9XMFss0= github.com/aws/aws-sdk-go-v2/credentials v1.19.39/go.mod h1:GonTDBQ+mTpCVNwaHjj0PagspfrYYMEqOx7FehoEP/I= github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.40 h1:r5aGipEVgI9aT/tAGjdrPbDQvIAKdTrS3rUPQtG4Rmo= github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.40/go.mod h1:vOD3CnPxAdkL6MWZeROkZsTlskklMFfgVFkHzx/oZpY= -github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.40 h1:UIXlbijuB2XK1Kr57fo8iIxCuaSHJzwZ1uo+2tbEYIk= -github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.40/go.mod h1:wcEsL6jscjZjVUinb0Q5qD/GXOG1yT3GNfmT9HuDwzU= -github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.40 h1:xLQVRDs2NddDmK9BEyh5KSlJ1Gpy5/GIJXrV6WcVGAE= -github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.40/go.mod h1:XRXnpFVFGLaEVK+olDdFIM1vNa04ETW452oFGEPUxAo= +github.com/aws/aws-sdk-go-v2/internal/configsources v1.5.1 h1:pc138gM1CW+XPc60rEwUlwwuwWFQK16CI1T7v1F9Oec= +github.com/aws/aws-sdk-go-v2/internal/configsources v1.5.1/go.mod h1:1+koxpPIbfBdfzP6vojm5/zTpTQ/micYwlxIiNB3TxI= +github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.8.1 h1:K0JsbZQj+1h208Ro1zHeA4l7bMp0NvRffHQ91q8Ol1s= +github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.8.1/go.mod h1:W3/vL6EtCIatICGy9ab29QhMuae+cOKPWcMxv02CO+Q= github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.41 h1:nv/ILuCY0yXACzMQwvtt/HbqDDjemZiI0AeDbxGQlnU= github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.41/go.mod h1:dzvOSpxaPqQ3j0xS6Lc1vyVuWW0RBj7s/QqYpzu3Q/0= github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.19 h1:bAdDl/HkGCcGPoe25ToSHEw23VIxt6CT5fLcg111BKg= @@ -20,6 +20,8 @@ github.com/aws/aws-sdk-go-v2/service/signin v1.6.0 h1:agcr0j8YeFEzdXNo17Rg9MbbjL github.com/aws/aws-sdk-go-v2/service/signin v1.6.0/go.mod h1:qU5PxgQ4JiUOOMotzfO3+5oUda5W+8JDVKyLQqlrJik= github.com/aws/aws-sdk-go-v2/service/sns v1.43.0 h1:VPYjwn0BoX34hb44OT8T+Ikgn4NzsN7fHetaHaevsDc= github.com/aws/aws-sdk-go-v2/service/sns v1.43.0/go.mod h1:I1vnLPvvi9KBqxddu8nJ4vktoPJvaIG05UmjBD9sqm8= +github.com/aws/aws-sdk-go-v2/service/sqs v1.48.1 h1:jXP3BdVenFa8RfLVH+D2gswrWZHJcgtygKCf22APFqo= +github.com/aws/aws-sdk-go-v2/service/sqs v1.48.1/go.mod h1:d4DToDhLnEofHKvFu4yCF0Be65pZW267COfKOztsZOQ= github.com/aws/aws-sdk-go-v2/service/sso v1.34.0 h1:FxaN8/sn61DTXNI6Gt678tFJUY8iUsCchm6Y/F/RjaA= github.com/aws/aws-sdk-go-v2/service/sso v1.34.0/go.mod h1:vu4OY6s8LJtT8BtYG2LD6BGSZMptkYn3o5hvCPB22jc= github.com/aws/aws-sdk-go-v2/service/ssooidc v1.39.0 h1:crWKPeGYTBTuBxQ3p73kjfJvt4brUIsr+Fuypko8FxY= diff --git a/tests/steps_test.go b/tests/steps_test.go index 5d8087c..7e06539 100644 --- a/tests/steps_test.go +++ b/tests/steps_test.go @@ -422,6 +422,25 @@ func registerVideoBeingTranscoded(t gobdd.StepTest, ctx gobdd.Context) { } } +// registerVideoBeingTranscodedUnder is registerVideoBeingTranscoded for a +// scenario that cares which categories the video is filed under. +func registerVideoBeingTranscodedUnder(t gobdd.StepTest, ctx gobdd.Context, categories string) { + w := worldOf(t, ctx) + + requestUploadSlot(t, ctx, "transcoding.mp4", "video/mp4") + uploadTheFile(t, ctx) + registerUploadedVideo(t, ctx, "A Video In Several Categories", categories) + + if w.last.status != 201 { + t.Fatalf("could not register a video to transcode: %s", w.last.summary()) + return + } + if strings.TrimSpace(w.video.MediaConvertJobID) == "" { + t.Fatalf("the registered video carries no transcoding job id, so no event could name it") + return + } +} + func mediaConvertReportsState(t gobdd.StepTest, ctx gobdd.Context, state string) { w := worldOf(t, ctx) @@ -581,3 +600,305 @@ func askForThatVideoWrittenAs(t gobdd.StepTest, ctx gobdd.Context, form string) } } } + +// --- the catalogue queue ---------------------------------------------------- + +// announcementTimeout bounds how long a scenario waits for the announcement. +// It is the status timeout's sibling: the publish happens in the same consumer +// pass as the status write, so a video that is ready and still unannounced +// after this is not slow, it is unannounced. +const announcementTimeout = 20 * time.Second + +func theCatalogueIsToldAboutTheVideo(t gobdd.StepTest, ctx gobdd.Context) { + w := worldOf(t, ctx) + + if strings.TrimSpace(w.video.ID) == "" { + t.Fatalf("no video has been registered, so nothing could be announced") + return + } + + background := context.Background() + + reader, err := newCatalogueReader(background) + if err != nil { + t.Fatalf("could not reach the catalogue queue: %s", err) + return + } + + announcement, err := reader.await(background, w.video.ID, announcementTimeout) + if err != nil { + t.Fatalf("could not read the catalogue queue: %s", err) + return + } + if announcement == nil { + t.Errorf("the catalogue was never told about video %q within %s", w.video.ID, announcementTimeout) + return + } + + w.announcement = announcement +} + +// announcementHoldWindow is how long "told nothing" watches for before it is +// satisfied. The announcement rides in the same consumer pass as the status +// write, and the status is already settled by the time this step runs, so a +// message that is not here by now is not coming. +const announcementHoldWindow = 5 * time.Second + +func theCatalogueIsToldNothingAboutTheVideo(t gobdd.StepTest, ctx gobdd.Context) { + w := worldOf(t, ctx) + + if strings.TrimSpace(w.video.ID) == "" { + t.Fatalf("no video has been registered, so there is nothing to look for") + return + } + + background := context.Background() + + reader, err := newCatalogueReader(background) + if err != nil { + t.Fatalf("could not reach the catalogue queue: %s", err) + return + } + + announcement, err := reader.await(background, w.video.ID, announcementHoldWindow) + if err != nil { + t.Fatalf("could not read the catalogue queue: %s", err) + return + } + if announcement != nil { + t.Errorf("the catalogue was told about video %q, which has nothing to play", w.video.ID) + } +} + +func theAnnouncementCarriesTheVideosDetails(t gobdd.StepTest, ctx gobdd.Context) { + w := worldOf(t, ctx) + + if w.announcement == nil { + t.Fatalf("no announcement has been received, so there is nothing to inspect") + return + } + + if w.announcement.Title != w.sent.Title { + t.Errorf("the announcement calls the video %q, but it was registered as %q", + w.announcement.Title, w.sent.Title) + } + + if want := playbackURLFor(w.slot.Key); w.announcement.PlaybackURL != want { + t.Errorf("the announcement points a player at %q, but the output landed at %q", + w.announcement.PlaybackURL, want) + } + + // Named, not numbered: a reader of the queue has no access to the category + // ids cms issues, so the names are what make the message self-contained. + want := make([]string, 0, len(w.sent.CategoryIDs)) + for _, id := range w.sent.CategoryIDs { + want = append(want, w.categoryName(t, id)) + } + + got := slices.Clone(w.announcement.Categories) + slices.Sort(got) + slices.Sort(want) + + if !slices.Equal(got, want) { + t.Errorf("the announcement files the video under %v, but it was registered under %v", + w.announcement.Categories, want) + } +} + +func theAnnouncementFilesItUnder(t gobdd.StepTest, ctx gobdd.Context, names string) { + w := worldOf(t, ctx) + + if w.announcement == nil { + t.Fatalf("no announcement has been received, so there is nothing to inspect") + return + } + + want := []string{} + for _, name := range strings.Split(names, ",") { + if name = strings.TrimSpace(name); name != "" { + want = append(want, name) + } + } + + got := slices.Clone(w.announcement.Categories) + slices.Sort(got) + slices.Sort(want) + + if !slices.Equal(got, want) { + t.Errorf("the announcement files the video under %v, but it was registered under %v", + w.announcement.Categories, want) + } +} + +// --- the catalogue, as discovery serves it ---------------------------------- + +// catalogueSettleTimeout bounds how long a scenario waits for the read side to +// catch up. It is longer than the CMS's own status timeout because two hops +// have to happen — cms consuming the job event and announcing, then discovery +// consuming that announcement — each with its own long poll. +const catalogueSettleTimeout = 30 * time.Second + +func theDiscoveryAPIIsAvailable(t gobdd.StepTest, ctx gobdd.Context) { + w := worldOf(t, ctx) + + w.discovery = newDiscoveryClient() + + result, err := w.discovery.get("/health") + if err != nil { + t.Fatalf("no Discovery at %s (%s) — start one with `docker compose up` from the repo root, "+ + "or set DISCOVERY_BASE_URL to point at a running instance", w.discovery.base, err) + return + } + if result.status != 200 { + t.Fatalf("Discovery at %s is not healthy: %s", w.discovery.base, result.summary()) + } +} + +func theCatalogueEventuallyHoldsThatVideo(t gobdd.StepTest, ctx gobdd.Context) { + w := worldOf(t, ctx) + + if strings.TrimSpace(w.video.ID) == "" { + t.Fatalf("no video has been registered, so the catalogue could hold nothing") + return + } + + deadline := time.Now().Add(catalogueSettleTimeout) + last := response{} + for time.Now().Before(deadline) { + result, err := w.discovery.get("/api/videos/" + w.video.ID) + if err != nil { + t.Fatalf("could not ask the catalogue for the video: %s", err) + return + } + last = result + + if result.status == 200 { + var body catalogueVideoBody + if err := result.json(&body); err != nil { + t.Fatalf("could not decode the catalogue's copy: %s", err) + return + } + w.catalogued = &body + return + } + + time.Sleep(statusPollInterval) + } + + t.Errorf("the catalogue never held video %q within %s; asking for it still gives %s", + w.video.ID, catalogueSettleTimeout, last.summary()) +} + +func theCataloguesCopyCarriesTheDetails(t gobdd.StepTest, ctx gobdd.Context) { + w := worldOf(t, ctx) + + if w.catalogued == nil { + t.Fatalf("the catalogue holds no copy of the video, so there is nothing to inspect") + return + } + + if w.catalogued.ID != w.video.ID { + t.Errorf("the catalogue filed the video as %q, but the CMS issued id %q", + w.catalogued.ID, w.video.ID) + } + + if w.catalogued.Title != w.sent.Title { + t.Errorf("the catalogue calls the video %q, but it was registered as %q", + w.catalogued.Title, w.sent.Title) + } + + if want := playbackURLFor(w.slot.Key); w.catalogued.PlaybackURL != want { + t.Errorf("the catalogue points a player at %q, but the output landed at %q", + w.catalogued.PlaybackURL, want) + } + + want := make([]string, 0, len(w.sent.CategoryIDs)) + for _, id := range w.sent.CategoryIDs { + want = append(want, w.categoryName(t, id)) + } + + got := slices.Clone(w.catalogued.Categories) + slices.Sort(got) + slices.Sort(want) + + if !slices.Equal(got, want) { + t.Errorf("the catalogue files the video under %v, but it was registered under %v", + w.catalogued.Categories, want) + } +} + +// catalogueAbsenceWindow is how long "does not hold" watches before it is +// satisfied. Held rather than sampled once: the read side is asynchronous, so +// a video that is absent the instant after registration proves nothing — it +// might simply not have been announced yet. +const catalogueAbsenceWindow = 5 * time.Second + +func theCatalogueDoesNotHoldThatVideo(t gobdd.StepTest, ctx gobdd.Context) { + w := worldOf(t, ctx) + + if strings.TrimSpace(w.video.ID) == "" { + t.Fatalf("no video has been registered, so there is nothing to look for") + return + } + + deadline := time.Now().Add(catalogueAbsenceWindow) + for { + result, err := w.discovery.get("/api/videos/" + w.video.ID) + if err != nil { + t.Fatalf("could not ask the catalogue for the video: %s", err) + return + } + + // Kept as the scenario's last exchange so the assertions that follow + // can read the status and the problem body the usual way. + w.last = result + + if result.status != 404 { + t.Errorf("the catalogue holds video %q, which the CMS never announced: %s", + w.video.ID, result.summary()) + return + } + + if time.Now().After(deadline) { + return + } + time.Sleep(statusPollInterval) + } +} + +func theCatalogueEventuallyHoldsThatVideoWithNoPlaybackURL(t gobdd.StepTest, ctx gobdd.Context) { + w := worldOf(t, ctx) + + if strings.TrimSpace(w.video.ID) == "" { + t.Fatalf("no video has been registered, so the catalogue could hold nothing") + return + } + + deadline := time.Now().Add(catalogueSettleTimeout) + last := "" + for time.Now().Before(deadline) { + result, err := w.discovery.get("/api/videos/" + w.video.ID) + if err != nil { + t.Fatalf("could not ask the catalogue for the video: %s", err) + return + } + + if result.status == 200 { + var body catalogueVideoBody + if err := result.json(&body); err != nil { + t.Fatalf("could not decode the catalogue's copy: %s", err) + return + } + if body.PlaybackURL == "" { + w.catalogued = &body + return + } + last = body.PlaybackURL + } + + time.Sleep(statusPollInterval) + } + + t.Errorf("the catalogue still points a player at %q for video %q after %s; the later "+ + "announcement carried no playlist and should have replaced it", last, w.video.ID, catalogueSettleTimeout) +} diff --git a/tests/suite_test.go b/tests/suite_test.go index 33e7c62..372865d 100644 --- a/tests/suite_test.go +++ b/tests/suite_test.go @@ -16,6 +16,10 @@ import ( type world struct { client *client + // discovery is the read side, set by the Background step that checks it is + // up. Nil in scenarios that never mention the catalogue. + discovery *client + // last is the most recent exchange with the CMS. Every "Then the request // …" step reads it. last response @@ -36,6 +40,14 @@ type world struct { // category instead of hard-coding the id the seed migration happened to // give it. categoriesByName map[string]int16 + + // announcement is the message the catalogue topic carried for this + // scenario's video, once a step has waited for it. + announcement *catalogueAnnouncement + + // catalogued is the read side's own copy of the video, once a step has + // waited for it to arrive. + catalogued *catalogueVideoBody } type worldKey struct{} @@ -91,6 +103,19 @@ func (w *world) categoryID(t gobdd.StepTest, name string) int16 { return id } +// categoryName resolves a category id back to its name, for asserting on a +// message that names categories rather than numbering them. +func (w *world) categoryName(t gobdd.StepTest, id int16) string { + w.categoryID(t, "other") // ensures the list is loaded + for name, known := range w.categoriesByName { + if known == id { + return name + } + } + t.Fatalf("the CMS knows no category with id %d; it offers %v", id, w.categoriesByName) + return "" +} + // categoryIDs turns a comma-separated list of category names into the ids the // register endpoint expects. An empty list stays empty — that is a scenario in // its own right. @@ -130,6 +155,7 @@ func TestVideoUpload(t *testing.T) { // Given — the stack is up, and steps that arrange state a later When acts on. suite.AddStep(`^the CMS API is available$`, theAPIIsAvailable) + suite.AddStep(`^the Discovery API is available$`, theDiscoveryAPIIsAvailable) suite.AddStep(`^I have requested an upload slot for "(.*)" of type "(.*)"$`, requestUploadSlot) suite.AddStep(`^I have uploaded the file to the upload URL$`, uploadTheFile) suite.AddStep(`^I have registered the uploaded video titled "(.*)" under categories "(.*)"$`, registerUploadedVideo) @@ -142,6 +168,7 @@ func TestVideoUpload(t *testing.T) { suite.AddStep(`^I ask the CMS for that video$`, askForThatVideo) suite.AddStep(`^I ask the CMS for the video with id "(.*)"$`, askForVideoWithID) suite.AddStep(`^I have registered a video that is being transcoded$`, registerVideoBeingTranscoded) + suite.AddStep(`^I have registered a video being transcoded under categories "(.*)"$`, registerVideoBeingTranscodedUnder) suite.AddStep(`^I have registered a video$`, registerVideoBeingTranscoded) suite.AddStep(`^I ask the CMS for that video with its id written (.*)$`, askForThatVideoWrittenAs) suite.AddStep(`^MediaConvert reports that the job reached "(.*)"$`, mediaConvertReportsState) @@ -151,6 +178,14 @@ func TestVideoUpload(t *testing.T) { suite.AddStep(`^the video keeps status "(.*)"$`, theVideoKeepsStatus) suite.AddStep(`^the video has a playback URL for its HLS playlist$`, theVideoHasAPlaybackURL) suite.AddStep(`^the video has no playback URL$`, theVideoHasNoPlaybackURL) + suite.AddStep(`^the catalogue is told about the video$`, theCatalogueIsToldAboutTheVideo) + suite.AddStep(`^the catalogue is told nothing about the video$`, theCatalogueIsToldNothingAboutTheVideo) + suite.AddStep(`^the announcement carries the video's title, playback URL and categories$`, theAnnouncementCarriesTheVideosDetails) + suite.AddStep(`^the announcement files it under "(.*)"$`, theAnnouncementFilesItUnder) + suite.AddStep(`^the catalogue eventually holds that video$`, theCatalogueEventuallyHoldsThatVideo) + suite.AddStep(`^the catalogue does not hold that video$`, theCatalogueDoesNotHoldThatVideo) + suite.AddStep(`^the catalogue eventually holds that video with no playback URL$`, theCatalogueEventuallyHoldsThatVideoWithNoPlaybackURL) + suite.AddStep(`^the catalogue's copy carries its title, playback URL and categories$`, theCataloguesCopyCarriesTheDetails) suite.AddStep(`^I register the uploaded video titled "(.*)" under categories "(.*)"$`, registerUploadedVideo) suite.AddStep(`^I register the uploaded video titled "(.*)" under category id (\d+)$`, registerUnderCategoryID) suite.AddStep(`^I register a video titled "(.*)" with the key "(.*)" under categories "(.*)"$`, registerWithKey)