From 3f5301afb6437e9b5ce18b3385fd2314f8d74908 Mon Sep 17 00:00:00 2001 From: FahdShalhoub Date: Thu, 27 Aug 2026 20:04:38 +0300 Subject: [PATCH] FEAT: Update Status Of Video Row On Media Convert Update --- .claude/skills/atdd/SKILL.md | 121 ++++++++++++ cms/docs/docs.go | 103 ++++++++++ cms/docs/swagger.json | 103 ++++++++++ cms/docs/swagger.yaml | 75 ++++++++ cms/go.mod | 10 +- cms/go.sum | 20 +- cms/internal/api/videos.go | 19 ++ cms/internal/consumers/mediaconvert.go | 135 +++++++++++++ ..._index_videos_mediaconvert_job_id.down.sql | 1 + ...03_index_videos_mediaconvert_job_id.up.sql | 4 + cms/internal/db/repositories/errors.go | 9 + cms/internal/db/repositories/videos.go | 61 ++++++ cms/internal/handlers/videos.go | 61 ++++++ cms/internal/services/sqs.go | 80 ++++++++ cms/main.go | 24 +++ docker-compose.yml | 11 ++ infrastructure/Pulumi.local.yaml | 8 +- infrastructure/main.go | 164 +++++++++++++++- tests/events_test.go | 119 ++++++++++++ tests/features/transcoding_status.feature | 44 +++++ tests/features/video_details.feature | 52 +++++ tests/go.mod | 15 ++ tests/go.sum | 30 +++ tests/steps_test.go | 180 ++++++++++++++++++ tests/suite_test.go | 9 + 25 files changed, 1444 insertions(+), 14 deletions(-) create mode 100644 .claude/skills/atdd/SKILL.md create mode 100644 cms/internal/consumers/mediaconvert.go create mode 100644 cms/internal/db/migrations/0003_index_videos_mediaconvert_job_id.down.sql create mode 100644 cms/internal/db/migrations/0003_index_videos_mediaconvert_job_id.up.sql create mode 100644 cms/internal/db/repositories/errors.go create mode 100644 cms/internal/services/sqs.go create mode 100644 tests/events_test.go create mode 100644 tests/features/transcoding_status.feature create mode 100644 tests/features/video_details.feature diff --git a/.claude/skills/atdd/SKILL.md b/.claude/skills/atdd/SKILL.md new file mode 100644 index 0000000..d2ae046 --- /dev/null +++ b/.claude/skills/atdd/SKILL.md @@ -0,0 +1,121 @@ +--- +description: The ATDD loop this repo develops features with — write the Gherkin scenario first, watch it fail, write just enough code in cms/ to pass it, re-run, move to the next scenario. Use when adding or changing an endpoint, a validation rule, an error response, or any user-visible behaviour in cms/. +--- + +## The loop + +One scenario at a time. Never write two scenarios' worth of code in one pass. + +1. **Write the scenario** in `tests/features/*.feature`, in the language of the + feature rather than of HTTP where you can. Name it for the behaviour + (`Refusing to register a video whose file was never uploaded`), not the + mechanism. +2. **Run it and watch it fail.** `cd tests && go test -count=1 -v ./...` + Confirm it fails for the reason you expect. A scenario that passes before + you have written any code is testing nothing — go back to step 1. +3. **Write just enough code in `cms/`** to turn that one scenario green. + Not the next scenario's code. Not a generalisation you have no test for. +4. **Re-run the suite** — the whole file, not just the new scenario. Green + means done; a scenario that used to pass and now does not means the new + code changed published behaviour, which is a decision, not an accident. +5. **Next scenario.** Repeat until the feature is described. + +Only then tidy: extract helpers, rename, split files. The suite is what makes +that safe, so do it with the tests green and re-run after. + +## Where things go + +Tests are **black box**. `tests/` is its own Go module (`thamanyah/tests`) that +imports nothing from `cms/` and speaks only HTTP. Do not add `_test.go` files +inside `cms/` — if a behaviour cannot be reached through the API, that is a +finding about the design, not a reason to reach past it. + +| file | what goes in it | +|---|---| +| `tests/features/*.feature` | the scenarios | +| `tests/suite_test.go` | step registration, the per-scenario `world` | +| `tests/steps_test.go` | what each step does | +| `tests/client_test.go` | HTTP plumbing and wire types | + +Production code keeps to the seams in `CLAUDE.md`: wire structs in +`internal/api`, domain types in `internal/models`, HTTP in `internal/handlers`, +SQL in `internal/db/repositories`, AWS in `internal/services`. "Just enough +code" still lands on the right seam — a handler that runs its own SQL is not +minimal, it is misplaced. + +## Running it + +```bash +docker compose up -d # localstack + infra + cms (needs LOCALSTACK_AUTH_TOKEN) +docker exec cms ./cms migrate # see Migrations below +cd tests && go test -count=1 -v ./... +``` + +`-count=1` defeats Go's test caching, which otherwise hides a re-run against a +changed service. `-v` prints the Gherkin: gobdd nests a subtest per feature, +scenario and step, so the step text appears in the output. + +`CMS_BASE_URL` (default `http://localhost:8081`) points the suite elsewhere. If +nothing is serving, the suite skips rather than fails. + +**Migrations.** Compose starts `cms` in server mode only — the migrate +container exists in the ECS task definition, not in `docker-compose.yml`. A new +migration is **not** applied by `docker compose watch` rebuilding the image. +Run `docker exec cms ./cms migrate` after adding one, or every scenario fails +on `relation "…" does not exist`. + +**The spec.** Changing a handler, its annotations, or an `internal/api` struct +means regenerating the OpenAPI spec, or the build drifts from the published +contract: + +```bash +$(go env GOPATH)/bin/swag init --generalInfo main.go --dir ./ --parseInternal --output ./docs +``` + +## Reading a red run + +gobdd does **not** stop a scenario at the first failing step. A step that fails +leaves the later steps running against stale state, so one real failure +produces a cascade of noisy ones. **Read the first failure in a scenario and +ignore the rest** until it is fixed. + +`cannot find step definition for step: …` is a legitimate red: the scenario is +written and the step is not. Add the function to `steps_test.go`, register it +in `TestVideoUpload`, and run again. + +## Writing steps + +- **Anchor every pattern** with `^…$`. gobdd matches unanchored and the + first registered pattern that matches wins, so an unanchored pattern can + silently shadow another step. +- State moves between steps through the `*world` in the gobdd context, never + package variables. It is held by **pointer** because gobdd clones the context + between the `Background` and the scenario steps. +- A `Scenario Outline` runs every example row in **one** world. Steps must + overwrite state, never accumulate it. +- Reuse a step before writing a new one. Two steps that differ only in wording + are two ways for the feature file to drift. + +## Asserting errors + +Errors are RFC 9457 Problem Details on `application/problem+json`. `title` is +held identical across every occurrence of a given problem so clients can branch +on it — so **assert the exact title**, not just the status. A new error +condition means a new stable title, and a new `@Failure` annotation on the +handler. + +When a status alone would be ambiguous, distinguish the cases in code rather +than in the assertion: a client's mistake and a server fault reaching the same +`writeProblem` call is the bug, and a sentinel error in +`internal/handlers/errors.go` plus `errors.Is` at the call site is how this repo +separates them. + +## `@known-gap` + +Tag a scenario `@known-gap` when it pins what the service does **today** and +that differs from what it is documented to do. It still passes — it is a record, +not a failure — and it carries a comment saying what the contract promises and +what changing it would take. Fixing the code means changing the scenario in the +same commit and dropping the tag. + +Do not use the tag to park a scenario you could not make pass. diff --git a/cms/docs/docs.go b/cms/docs/docs.go index 4ed94c0..c57b810 100644 --- a/cms/docs/docs.go +++ b/cms/docs/docs.go @@ -157,6 +157,47 @@ const docTemplate = `{ } } }, + "/api/videos/{id}": { + "get": { + "description": "Returns the catalogue record for one video, including the status its transcoding job has reached. This is how a client follows an upload after POST /api/videos: the record starts \"processing\" and moves to \"ready\" or \"failed\" once MediaConvert reports the job finished.", + "produces": [ + "application/json" + ], + "tags": [ + "videos" + ], + "summary": "Read a video", + "parameters": [ + { + "type": "string", + "description": "The video's id, as returned by POST /api/videos", + "name": "id", + "in": "path", + "required": true + } + ], + "responses": { + "200": { + "description": "OK", + "schema": { + "$ref": "#/definitions/api.VideoResponse" + } + }, + "404": { + "description": "No video has that id", + "schema": { + "$ref": "#/definitions/api.ProblemDetails" + } + }, + "500": { + "description": "The video could not be read from the database", + "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 or S3 connections.", @@ -359,6 +400,68 @@ const docTemplate = `{ "example": "about:blank" } } + }, + "api.VideoResponse": { + "type": "object", + "properties": { + "categoryIds": { + "type": "array", + "items": { + "type": "integer" + }, + "example": [ + 1, + 2 + ] + }, + "createdAt": { + "type": "string" + }, + "description": { + "type": "string", + "example": "A behind-the-scenes look at the evening bulletin." + }, + "fileName": { + "type": "string", + "example": "interview-cut.mov" + }, + "id": { + "type": "string", + "example": "0199f3a1-7c2e-7b21-9f0d-1a2b3c4d5e6f" + }, + "mediaConvertJobId": { + "type": "string", + "example": "1755300000000-abcdef" + }, + "sizeBytes": { + "type": "integer", + "example": 60 + }, + "status": { + "type": "string", + "enum": [ + "processing", + "ready", + "failed" + ], + "example": "processing" + }, + "storageKey": { + "type": "string", + "example": "videos/a1b2c3d4e5f6.mov" + }, + "tags": { + "type": "string", + "example": "media, press, riyadh" + }, + "title": { + "type": "string", + "example": "Inside the Newsroom" + }, + "updatedAt": { + "type": "string" + } + } } } }` diff --git a/cms/docs/swagger.json b/cms/docs/swagger.json index c00317e..a8d7b73 100644 --- a/cms/docs/swagger.json +++ b/cms/docs/swagger.json @@ -150,6 +150,47 @@ } } }, + "/api/videos/{id}": { + "get": { + "description": "Returns the catalogue record for one video, including the status its transcoding job has reached. This is how a client follows an upload after POST /api/videos: the record starts \"processing\" and moves to \"ready\" or \"failed\" once MediaConvert reports the job finished.", + "produces": [ + "application/json" + ], + "tags": [ + "videos" + ], + "summary": "Read a video", + "parameters": [ + { + "type": "string", + "description": "The video's id, as returned by POST /api/videos", + "name": "id", + "in": "path", + "required": true + } + ], + "responses": { + "200": { + "description": "OK", + "schema": { + "$ref": "#/definitions/api.VideoResponse" + } + }, + "404": { + "description": "No video has that id", + "schema": { + "$ref": "#/definitions/api.ProblemDetails" + } + }, + "500": { + "description": "The video could not be read from the database", + "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 or S3 connections.", @@ -352,6 +393,68 @@ "example": "about:blank" } } + }, + "api.VideoResponse": { + "type": "object", + "properties": { + "categoryIds": { + "type": "array", + "items": { + "type": "integer" + }, + "example": [ + 1, + 2 + ] + }, + "createdAt": { + "type": "string" + }, + "description": { + "type": "string", + "example": "A behind-the-scenes look at the evening bulletin." + }, + "fileName": { + "type": "string", + "example": "interview-cut.mov" + }, + "id": { + "type": "string", + "example": "0199f3a1-7c2e-7b21-9f0d-1a2b3c4d5e6f" + }, + "mediaConvertJobId": { + "type": "string", + "example": "1755300000000-abcdef" + }, + "sizeBytes": { + "type": "integer", + "example": 60 + }, + "status": { + "type": "string", + "enum": [ + "processing", + "ready", + "failed" + ], + "example": "processing" + }, + "storageKey": { + "type": "string", + "example": "videos/a1b2c3d4e5f6.mov" + }, + "tags": { + "type": "string", + "example": "media, press, riyadh" + }, + "title": { + "type": "string", + "example": "Inside the Newsroom" + }, + "updatedAt": { + "type": "string" + } + } } } } \ No newline at end of file diff --git a/cms/docs/swagger.yaml b/cms/docs/swagger.yaml index 36ebfd4..c8dc9e0 100644 --- a/cms/docs/swagger.yaml +++ b/cms/docs/swagger.yaml @@ -129,6 +129,51 @@ definitions: example: about:blank type: string type: object + api.VideoResponse: + properties: + categoryIds: + example: + - 1 + - 2 + items: + type: integer + type: array + createdAt: + type: string + description: + example: A behind-the-scenes look at the evening bulletin. + type: string + fileName: + example: interview-cut.mov + type: string + id: + example: 0199f3a1-7c2e-7b21-9f0d-1a2b3c4d5e6f + type: string + mediaConvertJobId: + example: 1755300000000-abcdef + type: string + sizeBytes: + example: 60 + type: integer + status: + enum: + - processing + - ready + - failed + example: processing + type: string + storageKey: + example: videos/a1b2c3d4e5f6.mov + type: string + tags: + example: media, press, riyadh + type: string + title: + example: Inside the Newsroom + type: string + updatedAt: + type: string + type: object info: contact: {} description: 'JSON API for ingesting videos into the Thamanyah catalogue. Uploads @@ -208,6 +253,36 @@ paths: summary: Register an uploaded video tags: - videos + /api/videos/{id}: + get: + description: 'Returns the catalogue record for one video, including the status + its transcoding job has reached. This is how a client follows an upload after + POST /api/videos: the record starts "processing" and moves to "ready" or "failed" + once MediaConvert reports the job finished.' + parameters: + - description: The video's id, as returned by POST /api/videos + in: path + name: id + required: true + type: string + produces: + - application/json + responses: + "200": + description: OK + schema: + $ref: '#/definitions/api.VideoResponse' + "404": + description: No video has that id + schema: + $ref: '#/definitions/api.ProblemDetails' + "500": + description: The video could not be read from the database + schema: + $ref: '#/definitions/api.ProblemDetails' + summary: Read a video + tags: + - videos /api/videos/presign: post: consumes: diff --git a/cms/go.mod b/cms/go.mod index b311338..52398ed 100644 --- a/cms/go.mod +++ b/cms/go.mod @@ -3,11 +3,13 @@ module thamanyah/cms/v2 go 1.25.12 require ( - github.com/aws/aws-sdk-go-v2 v1.43.5 + github.com/aws/aws-sdk-go-v2 v1.44.0 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 + github.com/aws/aws-sdk-go-v2/service/sqs v1.47.0 github.com/golang-migrate/migrate/v4 v4.19.1 + github.com/google/uuid v1.6.0 github.com/lib/pq v1.12.3 github.com/swaggo/http-swagger/v2 v2.0.2 github.com/swaggo/swag v1.16.6 @@ -18,8 +20,8 @@ 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.36 // indirect - github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.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/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 @@ -29,7 +31,7 @@ require ( 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 - github.com/aws/smithy-go v1.27.7 // 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/cms/go.sum b/cms/go.sum index cccad89..c88c73f 100644 --- a/cms/go.sum +++ b/cms/go.sum @@ -4,8 +4,8 @@ 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.43.5 h1:yKT5GYnFWhuDo+DqKvE5ZPwVn3RjC4MAeBtZGlh6AVM= -github.com/aws/aws-sdk-go-v2 v1.43.5/go.mod h1:wZjAJppCntyOGgVSmgVTfDyRJK5PHOasO6Wsy8U7Axk= +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/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= @@ -14,10 +14,10 @@ github.com/aws/aws-sdk-go-v2/credentials v1.19.35 h1:Cxua2RVdRwL0sfjHM/SnQoOnQ7x github.com/aws/aws-sdk-go-v2/credentials v1.19.35/go.mod h1:9XQ+RSIGPkycr+oCJYnB1uTv5kMVVR+rd2vYK0Hxj2w= github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.36 h1:gucL1KH/PAYbpTpBg09CiVpBdTu4qkCl8C7xOTBixUg= 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.36 h1:5CrzwxDqf4w3x1Vs3/NiZ0nsC34Hbm3pIDMWbsLebOE= -github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.36/go.mod h1:A3gHdKZIvG/QXERzZwcxNS3RNDFcRCuhhTFBYp+V/nw= -github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.36 h1:A4N2f4YPcST0v+dWtX+xrpPPCL9VTBhoIFFUWYqbacE= -github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.36/go.mod h1:B/Qr859uxWUEfZeGotK5KAEoof4Q9YWgNtPSwV6jcyk= +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/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,14 +34,16 @@ 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/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= github.com/aws/aws-sdk-go-v2/service/sso v1.33.5/go.mod h1:OcT2AhgTuxGAwZk5hgxaNLGpS33W8s8dUQadGVDVY9I= github.com/aws/aws-sdk-go-v2/service/ssooidc v1.38.5 h1:8xo1q9ttkYqMJ6vOXX67FPSpVEI7BWKVTKh77g82w+8= github.com/aws/aws-sdk-go-v2/service/ssooidc v1.38.5/go.mod h1:hbBeEUrZg6VddXYZpbKPyF0tl4XEnM+Dbx92RW3vmZI= github.com/aws/aws-sdk-go-v2/service/sts v1.45.5 h1:eQ5BtXDrPg2wK0AjtVPzeBhUpYPeqHE/ptiH7xJRGek= github.com/aws/aws-sdk-go-v2/service/sts v1.45.5/go.mod h1:f9ImhnOISY7BuTZLM8qHepCYnglHBVLk5wVzatmP++w= -github.com/aws/smithy-go v1.27.7 h1:Zgj5z4LfcDYoQIVk+n/yGdTkP/2y6ZT5vYxe0fp7bqE= -github.com/aws/smithy-go v1.27.7/go.mod h1:YE2RhdIuDbA5E5bTdciG9KrW3+TiEONeUWCqxX9i1Fc= +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= @@ -83,6 +85,8 @@ github.com/golang-migrate/migrate/v4 v4.19.1 h1:OCyb44lFuQfYXYLx1SCxPZQGU7mcaZ7g github.com/golang-migrate/migrate/v4 v4.19.1/go.mod h1:CTcgfjxhaUtsLipnLoQRWCrjYXycRz/g5+RWDuYgPrE= github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI= github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/josharian/intern v1.0.0 h1:vlS4z54oSdjm0bgjRigI+G1HpF+tI+9rE5LLzOg8HmY= github.com/josharian/intern v1.0.0/go.mod h1:5DoeVV0s6jJacbCEi61lwdGj/aVlrQvzHFFd8Hwg//Y= github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo= diff --git a/cms/internal/api/videos.go b/cms/internal/api/videos.go index 0d055e1..4b26fc1 100644 --- a/cms/internal/api/videos.go +++ b/cms/internal/api/videos.go @@ -55,3 +55,22 @@ type CompleteResponse struct { CreatedAt time.Time `json:"createdAt"` UpdatedAt time.Time `json:"updatedAt"` } + +// VideoResponse is a persisted video record as returned by +// GET /api/videos/{id}. It carries the same members as CompleteResponse — +// deliberately a separate type, so the shape POST publishes and the shape GET +// publishes can diverge without one silently dragging the other with it. +type VideoResponse struct { + ID string `json:"id" example:"0199f3a1-7c2e-7b21-9f0d-1a2b3c4d5e6f"` + Title string `json:"title" example:"Inside the Newsroom"` + Description string `json:"description" example:"A behind-the-scenes look at the evening bulletin."` + CategoryIDs []int16 `json:"categoryIds" example:"1,2"` + Tags string `json:"tags" example:"media, press, riyadh"` + FileName string `json:"fileName" example:"interview-cut.mov"` + StorageKey string `json:"storageKey" example:"videos/a1b2c3d4e5f6.mov"` + MediaConvertJobID string `json:"mediaConvertJobId" example:"1755300000000-abcdef"` + Status string `json:"status" enums:"processing,ready,failed" example:"processing"` + SizeBytes int64 `json:"sizeBytes" example:"60"` + CreatedAt time.Time `json:"createdAt"` + UpdatedAt time.Time `json:"updatedAt"` +} diff --git a/cms/internal/consumers/mediaconvert.go b/cms/internal/consumers/mediaconvert.go new file mode 100644 index 0000000..4ce4f7b --- /dev/null +++ b/cms/internal/consumers/mediaconvert.go @@ -0,0 +1,135 @@ +// Package consumers holds the queue-driven half of the service: work that +// arrives on a queue rather than as an HTTP request. It is to SQS what +// internal/handlers is to HTTP — it decodes a message, calls repositories, and +// decides what the message's fate is — so the same seams apply, and no SQL +// lives here. +package consumers + +import ( + "context" + "encoding/json" + "errors" + "log" + "thamanyah/cms/v2/internal/db/repositories" + "thamanyah/cms/v2/internal/models" + "thamanyah/cms/v2/internal/services" + "time" +) + +// receiveBackoff is how long the loop waits after a failed receive, so a queue +// that is unreachable produces a slow trickle of log lines rather than a spin. +const receiveBackoff = 5 * time.Second + +// jobStateChange is the EventBridge event MediaConvert emits on every job +// state transition. The subscription delivers it raw, so this is the whole +// message body — there is no SNS envelope to unwrap. Only the members this +// service acts on are modelled; AWS sends a good deal more. +type jobStateChange struct { + Source string `json:"source"` + DetailType string `json:"detail-type"` + Detail struct { + JobID string `json:"jobId"` + Status string `json:"status"` + } `json:"detail"` +} + +// statusForJobState maps MediaConvert's job states onto the catalogue's. The +// second return is false for states that are not an outcome — the job is still +// running, and the record should stay where it is rather than being rewritten +// with what it already says. +func statusForJobState(state string) (models.VideoStatus, bool) { + switch state { + case "COMPLETE": + return models.VideoStatusReady, true + case "ERROR", "CANCELED": + return models.VideoStatusFailed, true + default: + return "", false + } +} + +// RunMediaConvertEvents consumes job state changes until ctx is cancelled. It +// is meant to be run in its own goroutine for the lifetime of the process. +func RunMediaConvertEvents(ctx context.Context) { + log.Println("mediaconvert job events consumer started") + + for { + if ctx.Err() != nil { + log.Println("mediaconvert job events consumer stopped") + return + } + + messages, err := services.SQSClient.ReceiveMessages(ctx) + if err != nil { + if ctx.Err() != nil { + log.Println("mediaconvert job events consumer stopped") + return + } + log.Printf("Something Went Wrong Receiving Job Events: %s", err) + time.Sleep(receiveBackoff) + continue + } + + for _, message := range messages { + if !handleJobEvent(ctx, message.Body) { + // Left on the queue on purpose: it becomes visible again when + // the visibility timeout expires, and reaches the dead-letter + // queue if it keeps failing. + continue + } + + if err := services.SQSClient.DeleteMessage(ctx, message.ReceiptHandle); err != nil { + log.Printf("Something Went Wrong Acknowledging A Job Event: %s", err) + } + } + } +} + +// handleJobEvent processes one message and reports whether it is finished +// with — that is, whether it should be deleted from the queue. It returns +// false only when a retry could plausibly succeed. Anything a retry cannot fix +// (a body that will never parse, an event for a job nobody has) is finished +// with, however little it accomplished: leaving it on the queue would only +// stall the events behind it. +func handleJobEvent(ctx context.Context, body string) bool { + var event jobStateChange + if err := json.Unmarshal([]byte(body), &event); err != nil { + log.Printf("Discarding A Job Event That Is Not JSON: %s", err) + return true + } + + if event.Source != "aws.mediaconvert" || event.DetailType != "MediaConvert Job State Change" { + log.Printf("Discarding An Event That Is Not A MediaConvert Job State Change: source=%q detail-type=%q", + event.Source, event.DetailType) + return true + } + + if event.Detail.JobID == "" { + log.Printf("Discarding A Job Event That Names No Job") + return true + } + + status, isOutcome := statusForJobState(event.Detail.Status) + if !isOutcome { + // SUBMITTED, PROGRESSING and STATUS_UPDATE say the job is still + // running, which is what "processing" already records. + return true + } + + err := repositories.VideoRepo.UpdateVideoStatusByJobID(ctx, event.Detail.JobID, status) + if errors.Is(err, repositories.ErrVideoNotFound) { + // Not a fault: the topic carries every job in the account, including + // ones this service never submitted. + log.Printf("Ignoring A Job Event For A Job No Video Has: job=%q status=%q", + event.Detail.JobID, event.Detail.Status) + return true + } + if err != nil { + log.Printf("Something Went Wrong Recording A Job Outcome: job=%q status=%q: %s", + event.Detail.JobID, event.Detail.Status, err) + return false + } + + log.Printf("Recorded A Job Outcome: job=%q status=%q", event.Detail.JobID, status) + return true +} diff --git a/cms/internal/db/migrations/0003_index_videos_mediaconvert_job_id.down.sql b/cms/internal/db/migrations/0003_index_videos_mediaconvert_job_id.down.sql new file mode 100644 index 0000000..9cebd9c --- /dev/null +++ b/cms/internal/db/migrations/0003_index_videos_mediaconvert_job_id.down.sql @@ -0,0 +1 @@ +DROP INDEX videos_mediaconvert_job_id_idx; diff --git a/cms/internal/db/migrations/0003_index_videos_mediaconvert_job_id.up.sql b/cms/internal/db/migrations/0003_index_videos_mediaconvert_job_id.up.sql new file mode 100644 index 0000000..0a903e0 --- /dev/null +++ b/cms/internal/db/migrations/0003_index_videos_mediaconvert_job_id.up.sql @@ -0,0 +1,4 @@ +-- Job state changes arrive keyed by MediaConvert's job id, so every event the +-- consumer handles is a lookup on this column. Nothing indexed it before: the +-- column was only ever written, never searched. +CREATE INDEX videos_mediaconvert_job_id_idx ON videos (mediaconvert_job_id); diff --git a/cms/internal/db/repositories/errors.go b/cms/internal/db/repositories/errors.go new file mode 100644 index 0000000..9332d23 --- /dev/null +++ b/cms/internal/db/repositories/errors.go @@ -0,0 +1,9 @@ +package repositories + +import "errors" + +// Sentinel errors the repositories return so handlers can tell "there is no +// such row" from "the query failed", without importing database/sql to do it. + +// ErrVideoNotFound reports that no video has the requested id. +var ErrVideoNotFound = errors.New("video not found") diff --git a/cms/internal/db/repositories/videos.go b/cms/internal/db/repositories/videos.go index 3ed9e4d..111f921 100644 --- a/cms/internal/db/repositories/videos.go +++ b/cms/internal/db/repositories/videos.go @@ -3,6 +3,7 @@ package repositories import ( "context" "database/sql" + "errors" "fmt" "strings" "thamanyah/cms/v2/internal/models" @@ -14,6 +15,66 @@ type VideoRepository struct { var VideoRepo VideoRepository +// UpdateVideoStatusByJobID moves the video a transcoding job belongs to into a +// new status. It returns ErrVideoNotFound when no video carries that job id, +// which the consumer treats as a message to drop rather than a fault. +func (svc VideoRepository) UpdateVideoStatusByJobID(ctx context.Context, jobID string, status models.VideoStatus) error { + result, err := svc.SQLDB.ExecContext(ctx, ` + UPDATE videos + SET status = $1, updated_at = now() + WHERE mediaconvert_job_id = $2 + `, string(status), jobID) + if err != nil { + return err + } + + affected, err := result.RowsAffected() + if err != nil { + return err + } + if affected == 0 { + return ErrVideoNotFound + } + + return nil +} + +// GetVideoByID reads one video and the ids of the categories it is filed +// under. It returns ErrVideoNotFound when no video has that id. +func (svc VideoRepository) GetVideoByID(ctx context.Context, id string) (models.Video, error) { + var v models.Video + + err := svc.SQLDB.QueryRowContext(ctx, ` + SELECT id, title, description, tags, file_name, storage_key, mediaconvert_job_id, status, size_bytes, created_at, updated_at + FROM videos + WHERE id = $1 + `, id).Scan(&v.ID, &v.Title, &v.Description, &v.Tags, &v.FileName, &v.StorageKey, + &v.MediaConvertJobID, &v.Status, &v.SizeBytes, &v.CreatedAt, &v.UpdatedAt) + if errors.Is(err, sql.ErrNoRows) { + return models.Video{}, ErrVideoNotFound + } + if err != nil { + return models.Video{}, err + } + + rows, err := svc.SQLDB.QueryContext(ctx, + `SELECT category_id FROM video_categories WHERE video_id = $1 ORDER BY category_id`, v.ID) + if err != nil { + return models.Video{}, err + } + defer rows.Close() + + for rows.Next() { + var categoryID int16 + if err := rows.Scan(&categoryID); err != nil { + return models.Video{}, err + } + v.CategoryIDs = append(v.CategoryIDs, categoryID) + } + + return v, rows.Err() +} + // VideoExistsWithStorageKey reports whether a video has already been // registered under a storage key. It is served by videos_storage_key_key, the // index Postgres builds for the column's UNIQUE constraint. diff --git a/cms/internal/handlers/videos.go b/cms/internal/handlers/videos.go index de9b433..7649c5f 100644 --- a/cms/internal/handlers/videos.go +++ b/cms/internal/handlers/videos.go @@ -17,6 +17,8 @@ import ( "thamanyah/cms/v2/internal/models" "thamanyah/cms/v2/internal/services" "time" + + "github.com/google/uuid" ) const ( @@ -253,3 +255,62 @@ func randomFilename(original string) (string, error) { } return hex.EncodeToString(buf) + ext, nil } + +// GetVideo returns one video by its id. +// +// @Summary Read a video +// @Description Returns the catalogue record for one video, including the status its transcoding job has reached. This is how a client follows an upload after POST /api/videos: the record starts "processing" and moves to "ready" or "failed" once MediaConvert reports the job finished. +// @Tags videos +// @Produce json +// @Param id path string true "The video's id, as returned by POST /api/videos" +// @Success 200 {object} api.VideoResponse +// @Failure 404 {object} api.ProblemDetails "No video has that id" +// @Failure 500 {object} api.ProblemDetails "The video could not be read from the database" +// @Router /api/videos/{id} [get] +func GetVideo(w http.ResponseWriter, r *http.Request) { + id := r.PathValue("id") + + // Parsed before the query, because Postgres rejects anything that is not a + // uuid the moment it is compared against the column rather than simply + // matching no row — which would surface as a 500. An id that cannot name a + // video is a video that does not exist. + // + // The parsed value, not the raw one, is what goes to the database: uuid + // accepts spellings Postgres will not take verbatim — notably the + // "urn:uuid:" form — and String() puts them all back into the canonical + // form the column holds. + videoID, err := uuid.Parse(id) + if err != nil { + writeProblem(w, http.StatusNotFound, "Video Not Found", + fmt.Sprintf("No video has the id %q. Use the 'id' returned by POST /api/videos.", id)) + return + } + + video, err := repositories.VideoRepo.GetVideoByID(r.Context(), videoID.String()) + if errors.Is(err, repositories.ErrVideoNotFound) { + writeProblem(w, http.StatusNotFound, "Video Not Found", + fmt.Sprintf("No video has the id %q. Use the 'id' returned by POST /api/videos.", id)) + return + } + if err != nil { + log.Printf("Something Went Wrong Loading The Video: %s", err) + writeProblem(w, http.StatusInternalServerError, "Something Went Wrong Loading The Video", + "The video could not be read from the database. This is a server-side fault and the request was not processed; retrying in a few moments may succeed.") + return + } + + writeJSON(w, http.StatusOK, api.VideoResponse{ + ID: video.ID, + Title: video.Title, + Description: video.Description, + CategoryIDs: video.CategoryIDs, + Tags: video.Tags, + FileName: video.FileName, + StorageKey: video.StorageKey, + MediaConvertJobID: video.MediaConvertJobID, + Status: video.Status.String(), + SizeBytes: video.SizeBytes, + CreatedAt: video.CreatedAt, + UpdatedAt: video.UpdatedAt, + }) +} diff --git a/cms/internal/services/sqs.go b/cms/internal/services/sqs.go new file mode 100644 index 0000000..71a543a --- /dev/null +++ b/cms/internal/services/sqs.go @@ -0,0 +1,80 @@ +package services + +import ( + "context" + "fmt" + + "github.com/aws/aws-sdk-go-v2/service/sqs" +) + +var SQSClient SQS + +// QueueMessage is one message taken off the queue. ReceiptHandle is what +// identifies it for deletion — it belongs to this delivery, not to the +// message, so it cannot be held across receives. +type QueueMessage struct { + Body string + ReceiptHandle string +} + +type SQS interface { + ReceiveMessages(ctx context.Context) ([]QueueMessage, error) + DeleteMessage(ctx context.Context, receiptHandle string) error +} + +type SQSConcrete struct { + SQSClient *sqs.Client + QueueURL string +} + +// receiveWaitSeconds turns every receive into a long poll: the call parks on +// the server until a message arrives or this elapses, so an event is picked up +// within milliseconds of being published while an idle consumer costs one +// request every twenty seconds rather than spinning. 20 is the AWS maximum. +const receiveWaitSeconds = 20 + +func (svc SQSConcrete) ReceiveMessages(ctx context.Context) ([]QueueMessage, error) { + output, err := svc.SQSClient.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{ + QueueUrl: &svc.QueueURL, + MaxNumberOfMessages: 10, + WaitTimeSeconds: receiveWaitSeconds, + }) + if err != nil { + return nil, err + } + + messages := make([]QueueMessage, 0, len(output.Messages)) + for _, message := range output.Messages { + if message.Body == nil || message.ReceiptHandle == nil { + continue + } + messages = append(messages, QueueMessage{ + Body: *message.Body, + ReceiptHandle: *message.ReceiptHandle, + }) + } + + return messages, nil +} + +// DeleteMessage acknowledges a message. Until this is called the message is +// merely invisible, and it returns to the queue when its visibility timeout +// expires — which is how a handler that failed gets another attempt. +func (svc SQSConcrete) DeleteMessage(ctx context.Context, receiptHandle string) error { + _, err := svc.SQSClient.DeleteMessage(ctx, &sqs.DeleteMessageInput{ + QueueUrl: &svc.QueueURL, + ReceiptHandle: &receiptHandle, + }) + return err +} + +// AssertSuccessfulConnection verifies the queue is reachable and readable with +// the configured credentials. It panics on failure, since a service whose +// consumer cannot start would silently leave every video stuck "processing". +func (svc SQSConcrete) AssertSuccessfulConnection(ctx context.Context) { + if _, err := svc.SQSClient.GetQueueAttributes(ctx, &sqs.GetQueueAttributesInput{ + QueueUrl: &svc.QueueURL, + }); err != nil { + panic(fmt.Errorf("sqs: cannot connect to queue %q: %w", svc.QueueURL, err)) + } +} diff --git a/cms/main.go b/cms/main.go index 0a5809c..f64f675 100644 --- a/cms/main.go +++ b/cms/main.go @@ -6,6 +6,7 @@ import ( "log" "net/http" "os" + "thamanyah/cms/v2/internal/consumers" "thamanyah/cms/v2/internal/db" "thamanyah/cms/v2/internal/db/repositories" "thamanyah/cms/v2/internal/handlers" @@ -14,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/sqs" _ "github.com/lib/pq" httpSwagger "github.com/swaggo/http-swagger/v2" @@ -66,6 +68,11 @@ func runServer() { panic(fmt.Errorf("missing required env var: MEDIACONVERT_OUTPUT_BUCKET")) } + mediaConvertEventsQueueURL := os.Getenv("MEDIACONVERT_EVENTS_QUEUE_URL") + if mediaConvertEventsQueueURL == "" { + panic(fmt.Errorf("missing required env var: MEDIACONVERT_EVENTS_QUEUE_URL")) + } + // AWS_ENDPOINT_URL is only ever set when pointing at something other than // real S3 — LocalStack, in docker-compose. Virtual-host addressing would // resolve . there, which neither Docker's DNS nor @@ -93,6 +100,14 @@ func runServer() { OutputBucket: mediaConvertOutputBucket, } + concreteSQSClient := &services.SQSConcrete{ + SQSClient: sqs.NewFromConfig(awsConfig), + QueueURL: mediaConvertEventsQueueURL, + } + concreteSQSClient.AssertSuccessfulConnection(context.Background()) + + services.SQSClient = concreteSQSClient + concreteDBClient := db.CreateDBConnection(requireDBConnectionString()) defer db.CloseConnection(concreteDBClient) db.AssertSuccessfulConnection(context.Background(), concreteDBClient) @@ -105,12 +120,21 @@ func runServer() { SQLDB: concreteDBClient, } + // MediaConvert reports job state changes to a topic that fans out to this + // service's queue; the consumer runs alongside the HTTP server for the + // lifetime of the process, so a video's status catches up with its + // transcoding job without anything calling back into this service. + consumerCtx, stopConsumer := context.WithCancel(context.Background()) + defer stopConsumer() + go consumers.RunMediaConvertEvents(consumerCtx) + mux := http.NewServeMux() mux.HandleFunc("GET /health", handlers.Health) mux.HandleFunc("GET /api/categories", handlers.ListCategories) mux.HandleFunc("POST /api/videos/presign", handlers.PresignVideoUpload) mux.HandleFunc("POST /api/videos", handlers.CompleteVideoUpload) + 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 1b2a869..64045b0 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -12,6 +12,13 @@ services: # LocalStack configuration: https://docs.localstack.cloud/references/configuration/ - DEBUG=${DEBUG:-0} - PERSISTENCE=${PERSISTENCE:-0} + # Queue URLs are handed out by LocalStack and then connected to by the + # cms container. The default "standard" strategy builds them on + # sqs..localhost.localstack.cloud, which public DNS points at + # 127.0.0.1 and which this network has no alias for — so cms would fail + # to resolve its own queue. "off" keeps them on the gateway host, the + # one name that resolves from the host and from inside the network. + - SQS_ENDPOINT_STRATEGY=off volumes: - "${LOCALSTACK_VOLUME_DIR:-./volume}:/var/lib/localstack" - "/var/run/docker.sock:/var/run/docker.sock" @@ -82,6 +89,10 @@ services: - MEDIACONVERT_INPUT_BUCKET=${MEDIACONVERT_INPUT_BUCKET:-raw-uploads-bucket} - MEDIACONVERT_OUTPUT_BUCKET=${MEDIACONVERT_OUTPUT_BUCKET:-encoded-bucket} - MEDIACONVERT_ROLE_ARN=${MEDIACONVERT_ROLE_ARN:-arn:aws:iam::000000000000:role/mediaconvert-service-role} + # The queue the job-events topic fans out to, consumed for the lifetime + # 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} # Postgres: the RDS instance LocalStack provisions, which runs inside the # localstack container and speaks plain TCP — hence sslmode=disable. - DB_HOST=${DB_HOST:-localhost.localstack.cloud} diff --git a/infrastructure/Pulumi.local.yaml b/infrastructure/Pulumi.local.yaml index 6a98079..c0b71b5 100644 --- a/infrastructure/Pulumi.local.yaml +++ b/infrastructure/Pulumi.local.yaml @@ -12,7 +12,11 @@ config: aws:accessKey: test aws:secretKey: test aws:skipCredentialsValidation: "true" - aws:skipRequestingAccountId: "true" + # SNS builds topic ARNs from the caller's account id, so unlike the other + # services it cannot work without one — "true" here makes the provider hand + # SNS an empty account and every topic call fails. sts is pointed at + # LocalStack just below, so asking is cheap and answers 000000000000. + aws:skipRequestingAccountId: "false" aws:skipMetadataApiCheck: "true" # Virtual-host addressing would resolve .localhost.localstack.cloud, # which points at 127.0.0.1 — the wrong container from inside compose. @@ -21,6 +25,8 @@ config: - iam: http://localhost.localstack.cloud:4566 rds: http://localhost.localstack.cloud:4566 s3: http://localhost.localstack.cloud:4566 + sns: http://localhost.localstack.cloud:4566 + sqs: http://localhost.localstack.cloud:4566 sts: http://localhost.localstack.cloud:4566 thamanyah:localstack: "true" encryptionsalt: v1:QKoqcSNBvDc=:v1:bTGZko5jHuGNRVoq:2oQ4pFb8MYwspEbxipYeO+rVqVnLBg== diff --git a/infrastructure/main.go b/infrastructure/main.go index 65dcb66..e320ad5 100644 --- a/infrastructure/main.go +++ b/infrastructure/main.go @@ -18,6 +18,8 @@ import ( "github.com/pulumi/pulumi-aws/sdk/v7/go/aws/rds" "github.com/pulumi/pulumi-aws/sdk/v7/go/aws/s3" "github.com/pulumi/pulumi-aws/sdk/v7/go/aws/secretsmanager" + "github.com/pulumi/pulumi-aws/sdk/v7/go/aws/sns" + "github.com/pulumi/pulumi-aws/sdk/v7/go/aws/sqs" "github.com/pulumi/pulumi-postgresql/sdk/v3/go/postgresql" "github.com/pulumi/pulumi-random/sdk/v4/go/random" "github.com/pulumi/pulumi/sdk/v3/go/pulumi" @@ -808,6 +810,154 @@ func main() { return err } + // MediaConvert job events: MediaConvert -> EventBridge -> SNS -> SQS -> + // cms. cms *consumes a queue* rather than being called back on an HTTPS + // endpoint, so nothing new is exposed on the ALB, authenticity comes + // from IAM instead of SNS message signatures, and a failed handler + // leaves the message on the queue to be retried instead of dropping the + // update. The topic sits between the rule and the queue so a second + // consumer — discovery, when it has code — can subscribe its own queue + // without touching this wiring or MediaConvert. + jobEventsTopicArgs := &sns.TopicArgs{} + jobEventsQueueArgs := &sqs.QueueArgs{} + jobEventsDeadLetterArgs := &sqs.QueueArgs{} + if localstack { + // docker-compose.yml and the tests/ suite refer to these literally. + jobEventsTopicArgs.Name = pulumi.String("mediaconvert-job-events") + jobEventsQueueArgs.Name = pulumi.String("cms-mediaconvert-events") + jobEventsDeadLetterArgs.Name = pulumi.String("cms-mediaconvert-events-dlq") + } + + jobEventsTopic, err := sns.NewTopic(ctx, "mediaconvert-job-events", jobEventsTopicArgs) + if err != nil { + return err + } + + // Where a message lands after maxReceiveCount failed handlings, so one + // event cms cannot process never blocks the ones behind it. + jobEventsDeadLetter, err := sqs.NewQueue(ctx, "cms-mediaconvert-events-dlq", jobEventsDeadLetterArgs) + if err != nil { + return err + } + + jobEventsQueueArgs.VisibilityTimeoutSeconds = pulumi.Int(60) + jobEventsQueueArgs.RedrivePolicy = jobEventsDeadLetter.Arn.ApplyT(func(arn string) (string, error) { + b, err := json.Marshal(map[string]any{ + "deadLetterTargetArn": arn, + "maxReceiveCount": 5, + }) + return string(b), err + }).(pulumi.StringOutput) + + jobEventsQueue, err := sqs.NewQueue(ctx, "cms-mediaconvert-events", jobEventsQueueArgs) + 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( + 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": "AllowJobEventsTopicToSend", + "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, "cms-mediaconvert-events-queue-policy", &sqs.QueuePolicyArgs{ + QueueUrl: jobEventsQueue.ID(), + Policy: jobEventsQueuePolicy, + }) + if err != nil { + return err + } + + _, err = sns.NewTopicSubscription(ctx, "cms-mediaconvert-events-subscription", &sns.TopicSubscriptionArgs{ + Topic: jobEventsTopic.Arn, + Protocol: pulumi.String("sqs"), + Endpoint: jobEventsQueue.Arn, + // The queue receives the EventBridge event itself rather than an SNS + // envelope carrying it as a JSON string, so the consumer parses one + // document instead of unwrapping two. + RawMessageDelivery: pulumi.Bool(true), + }) + if err != nil { + return err + } + + jobEventsTopicPolicy := jobEventsTopic.Arn.ApplyT(func(topicArn string) (string, error) { + doc := map[string]any{ + "Version": "2012-10-17", + "Statement": []map[string]any{ + { + "Sid": "AllowEventBridgeToPublish", + "Effect": "Allow", + "Principal": map[string]string{"Service": "events.amazonaws.com"}, + "Action": "sns:Publish", + "Resource": topicArn, + }, + }, + } + b, err := json.Marshal(doc) + return string(b), err + }).(pulumi.StringOutput) + + _, err = sns.NewTopicPolicy(ctx, "mediaconvert-job-events-topic-policy", &sns.TopicPolicyArgs{ + Arn: jobEventsTopic.Arn, + Policy: jobEventsTopicPolicy, + }) + if err != nil { + return err + } + + // MediaConvert does not publish to SNS itself: it reports state changes + // to EventBridge, and this rule is what forwards them to the topic. + // Skipped under LocalStack, whose MediaConvert emulation does not emit + // these — the suite in tests/ publishes to the topic directly instead, + // which is the same seam from cms's side. + if !localstack { + jobStateChangePattern, err := json.Marshal(map[string]any{ + "source": []string{"aws.mediaconvert"}, + "detail-type": []string{"MediaConvert Job State Change"}, + }) + if err != nil { + return err + } + + jobEventsRule, err := cloudwatch.NewEventRule(ctx, "mediaconvert-job-state-change", &cloudwatch.EventRuleArgs{ + Description: pulumi.String("MediaConvert job state changes, forwarded to the cms job events topic"), + EventPattern: pulumi.String(string(jobStateChangePattern)), + }) + if err != nil { + return err + } + + _, err = cloudwatch.NewEventTarget(ctx, "mediaconvert-job-state-change-to-topic", &cloudwatch.EventTargetArgs{ + Rule: jobEventsRule.Name, + Arn: jobEventsTopic.Arn, + }) + if err != nil { + return err + } + } + // cms's own ECS task role (distinct from the shared execRole, which // is only for the ECS agent's pull/logs/secrets access): lets the // running container call S3 and MediaConvert directly. discovery @@ -819,10 +969,11 @@ func main() { return err } - cmsTaskPolicy := pulumi.All(rawUploadsBucket.Arn, mediaConvertRole.Arn).ApplyT( + cmsTaskPolicy := pulumi.All(rawUploadsBucket.Arn, mediaConvertRole.Arn, jobEventsQueue.Arn).ApplyT( func(args []any) (string, error) { rawUploadsArn := args[0].(string) mediaConvertRoleArn := args[1].(string) + jobEventsQueueArn := args[2].(string) doc := map[string]any{ "Version": "2012-10-17", @@ -848,6 +999,16 @@ func main() { "Action": "mediaconvert:CreateJob", "Resource": "*", }, + { + "Sid": "ConsumeJobEvents", + "Effect": "Allow", + "Action": []string{ + "sqs:ReceiveMessage", + "sqs:DeleteMessage", + "sqs:GetQueueAttributes", + }, + "Resource": jobEventsQueueArn, + }, { "Sid": "PassMediaConvertRole", "Effect": "Allow", @@ -884,6 +1045,7 @@ func main() { {Name: "MEDIACONVERT_INPUT_BUCKET", Value: rawUploadsBucket.ID().ToStringOutput()}, {Name: "MEDIACONVERT_OUTPUT_BUCKET", Value: bucket.ID().ToStringOutput()}, {Name: "MEDIACONVERT_ROLE_ARN", Value: mediaConvertRole.Arn}, + {Name: "MEDIACONVERT_EVENTS_QUEUE_URL", Value: jobEventsQueue.Url}, } cmsRepo, cmsService, cmsAlb, err = deployFargateService(ctx, "cms", 8081, diff --git a/tests/events_test.go b/tests/events_test.go new file mode 100644 index 0000000..65355a1 --- /dev/null +++ b/tests/events_test.go @@ -0,0 +1,119 @@ +package tests + +import ( + "context" + "encoding/json" + "fmt" + "os" + "strings" + "time" + + "github.com/aws/aws-sdk-go-v2/aws" + 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" +) + +// These scenarios stand in for EventBridge, not for a client: they publish the +// same "MediaConvert Job State Change" event AWS would put on the topic when a +// job changes state. That is the seam cms owns — the topic, the queue, the +// consumer and the record it updates. The EventBridge rule that feeds the +// topic in AWS is Pulumi configuration, and is not exercised here. +// +// This is the only place the suite reaches for the AWS SDK. Everything a real +// client does still goes over plain HTTP. + +const ( + defaultEndpointURL = "http://localhost.localstack.cloud:4566" + defaultTopicARN = "arn:aws:sns:us-east-1:000000000000:mediaconvert-job-events" + defaultRegion = "us-east-1" +) + +func envOr(name, fallback string) string { + if v := strings.TrimSpace(os.Getenv(name)); v != "" { + return v + } + return fallback +} + +// jobStateChange is the EventBridge event MediaConvert emits. Only the members +// cms reads are modelled; AWS sends a good deal more. +type jobStateChange struct { + Version string `json:"version"` + ID string `json:"id"` + DetailType string `json:"detail-type"` + Source string `json:"source"` + Account string `json:"account"` + Time time.Time `json:"time"` + Region string `json:"region"` + Resources []string `json:"resources"` + Detail jobStateDetail `json:"detail"` +} + +type jobStateDetail struct { + Timestamp int64 `json:"timestamp"` + AccountID string `json:"accountId"` + Queue string `json:"queue"` + JobID string `json:"jobId"` + Status string `json:"status"` +} + +// eventPublisher puts job state changes on the topic cms's queue subscribes to. +type eventPublisher struct { + sns *sns.Client + 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), + // 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"), "")), + ) + if err != nil { + return nil, err + } + + return &eventPublisher{ + sns: sns.NewFromConfig(cfg, func(o *sns.Options) { o.BaseEndpoint = aws.String(endpoint) }), + topicARN: envOr("MEDIACONVERT_EVENTS_TOPIC_ARN", defaultTopicARN), + }, nil +} + +func (p *eventPublisher) publishJobState(ctx context.Context, jobID, state string) error { + event := jobStateChange{ + Version: "0", + ID: fmt.Sprintf("test-%d", time.Now().UnixNano()), + DetailType: "MediaConvert Job State Change", + Source: "aws.mediaconvert", + Account: "000000000000", + Time: time.Now().UTC(), + Region: envOr("AWS_REGION", defaultRegion), + Resources: []string{fmt.Sprintf("arn:aws:mediaconvert:%s:000000000000:jobs/%s", envOr("AWS_REGION", defaultRegion), jobID)}, + Detail: jobStateDetail{ + Timestamp: time.Now().UnixMilli(), + AccountID: "000000000000", + JobID: jobID, + Status: state, + }, + } + + body, err := json.Marshal(event) + if err != nil { + return err + } + + _, err = p.sns.Publish(ctx, &sns.PublishInput{ + TopicArn: aws.String(p.topicARN), + Message: aws.String(string(body)), + }) + if err != nil { + return fmt.Errorf("publishing to %s: %w", p.topicARN, err) + } + return nil +} diff --git a/tests/features/transcoding_status.feature b/tests/features/transcoding_status.feature new file mode 100644 index 0000000..d795a46 --- /dev/null +++ b/tests/features/transcoding_status.feature @@ -0,0 +1,44 @@ +Feature: Transcoding status + A registered video starts life as "processing" and is handed to MediaConvert. + MediaConvert reports every job state change to an SNS topic, which fans the + event out to a queue cms consumes, so the catalogue record catches up with + the transcoding job on its own — nobody polls AWS and nothing is exposed for + AWS to call back into. + + Background: + Given the CMS API is available + + Scenario: A finished job marks the video ready + 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" + + Scenario Outline: A job that did not finish cleanly marks the video failed + Given I have registered a video that is being transcoded + When MediaConvert reports that the job reached "" + Then the video eventually has status "failed" + + Examples: + | state | + | ERROR | + | CANCELED | + + Scenario Outline: A state that is not an outcome leaves the video processing + Given I have registered a video that is being transcoded + When MediaConvert reports that the job reached "" + Then the video keeps status "processing" + + Examples: + | state | + | SUBMITTED | + | PROGRESSING | + | STATUS_UPDATE | + + # A message naming a job nobody has must be consumed and dropped, not left to + # redeliver forever. The second event is the assertion that it was: it can + # only be handled if the first one did not wedge the consumer. + Scenario: An event for a job no video has does not stop the consumer + Given I have registered a video that is being transcoded + When MediaConvert reports that the job "1700000000000-nosuchjob" reached "COMPLETE" + And MediaConvert reports that the job reached "COMPLETE" + Then the video eventually has status "ready" diff --git a/tests/features/video_details.feature b/tests/features/video_details.feature new file mode 100644 index 0000000..172fca1 --- /dev/null +++ b/tests/features/video_details.feature @@ -0,0 +1,52 @@ +Feature: Video details + A registered video can be read back by its id. This is how a client learns + what became of an upload once it was handed to the transcoding pipeline — + without it, the status a video carries is written but never visible. + + Background: + Given the CMS API is available + + Scenario: Reading back a video that was registered + Given I have requested an upload slot for "detail.mp4" of type "video/mp4" + And I have uploaded the file to the upload URL + And I have registered the uploaded video titled "Inside the Newsroom" under categories "documentary, news" + When I ask the CMS for that video + Then the request succeeds with status 200 + And the video has an id + And the video is registered with status "processing" + And the video is stored under the key from the upload slot + And the video is filed under categories "documentary, news" + And the video keeps the metadata I sent + + # A uuid has several accepted spellings, and Postgres takes only some of them + # verbatim — the urn form it rejects outright. The handler parses the id and + # queries with the canonical form, so every spelling names the same video + # instead of some of them faulting. + Scenario Outline: Reading a video by another accepted spelling of its id + Given I have registered a video + When I ask the CMS for that video with its id written
+ Then the request succeeds with status 200 + And the video has an id + + Examples: + | form | + | braced | + | unhyphenated | + | as a urn | + + Scenario: Asking for a video that does not exist + When I ask the CMS for the video with id "0199f3a1-7c2e-7b21-9f0d-1a2b3c4d5e6f" + Then the request is rejected with status 404 + And the problem title is "Video Not Found" + + Scenario Outline: Asking for a video with an id that is not a video id + When I ask the CMS for the video with id "" + Then the request is rejected with status 404 + And the problem title is "Video Not Found" + + Examples: + | id | + | not-a-uuid | + | 12345 | + | 0199f3a1-7c2e-7b21-9f0d-1a2b3c4d5e6 | + | '; DROP TABLE videos; -- | diff --git a/tests/go.mod b/tests/go.mod index 59b49d8..24cd7da 100644 --- a/tests/go.mod +++ b/tests/go.mod @@ -5,6 +5,21 @@ go 1.25.12 require 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/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 + github.com/aws/smithy-go v1.28.1 // indirect github.com/cucumber/gherkin/go/v33 v33.0.0 // indirect github.com/cucumber/messages/go/v28 v28.0.0 // indirect github.com/google/uuid v1.6.0 // indirect diff --git a/tests/go.sum b/tests/go.sum index 8d2aef6..3c89b30 100644 --- a/tests/go.sum +++ b/tests/go.sum @@ -1,3 +1,33 @@ +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/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/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= +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.13.40 h1:gr3Fw1cxZXNCdeo/lQ7isHEHzvHVM7z75qb2zW9aMjw= +github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.40/go.mod h1:8z/9CmfnQhiuXD7Ykbcg4a/whSWsniE0ODSx9uwVzfk= +github.com/aws/aws-sdk-go-v2/service/signin v1.6.0 h1:agcr0j8YeFEzdXNo17Rg9MbbjLRjrimabwNtji4e+lU= +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/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= +github.com/aws/aws-sdk-go-v2/service/ssooidc v1.39.0/go.mod h1:HjjZVhaBz0JBR/kbWKThmNDhFKS7y6EURuk493tJk9Y= +github.com/aws/aws-sdk-go-v2/service/sts v1.46.0 h1:IZ63JdogSNNjex/jsODNv7jGDcO/xJYd9FsgyfCsp1g= +github.com/aws/aws-sdk-go-v2/service/sts v1.46.0/go.mod h1:I+rwAf3spG5dITBaAo3xXRowk8kiOhtU1kYxfvCTC44= +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/cucumber/gherkin/go/v33 v33.0.0 h1:PqQ81cHjD732/GZ7c/6k/sjlYG4LzLp51nggL6172Sg= github.com/cucumber/gherkin/go/v33 v33.0.0/go.mod h1:mnP4fdkoc+LmjSLi9Kq3M5D84GphRWLdgeKAlvUYp2c= github.com/cucumber/messages/go/v28 v28.0.0 h1:BOJmy8LKSbdKxM6Ba1v9ZmpZk7j5cyH+LTAaGxMeflc= diff --git a/tests/steps_test.go b/tests/steps_test.go index 0133097..2a07ab1 100644 --- a/tests/steps_test.go +++ b/tests/steps_test.go @@ -2,8 +2,11 @@ package tests import ( "bytes" + "context" + "net/url" "slices" "strings" + "time" "github.com/go-bdd/gobdd" ) @@ -352,3 +355,180 @@ func theVideoKeepsTheMetadataISent(t gobdd.StepTest, ctx gobdd.Context) { t.Errorf("expected the file name %q, got %q", w.sent.FileName, w.video.FileName) } } + +func askForThatVideo(t gobdd.StepTest, ctx gobdd.Context) { + w := worldOf(t, ctx) + + if w.video.ID == "" { + t.Fatalf("no video has been registered, so there is no id to ask for") + return + } + + result, err := w.client.get("/api/videos/" + w.video.ID) + if err != nil { + t.Fatalf("could not ask for the video: %s", err) + return + } + + w.last = result + w.video = videoBody{} + + if result.status == 200 { + if err := result.json(&w.video); err != nil { + t.Fatalf("could not decode the video: %s", err) + } + } +} + +func askForVideoWithID(t gobdd.StepTest, ctx gobdd.Context, id string) { + w := worldOf(t, ctx) + + result, err := w.client.get("/api/videos/" + url.PathEscape(id)) + if err != nil { + t.Fatalf("could not ask for the video: %s", err) + return + } + + w.last = result + w.video = videoBody{} +} + +// --- transcoding status ----------------------------------------------------- + +// statusSettleTimeout bounds how long a scenario waits for the consumer to +// pick an event off the queue and write the record. The consumer long-polls, +// so in practice this resolves in well under a second; the budget is for a +// loaded machine, not for a slow path. +const ( + statusSettleTimeout = 20 * time.Second + statusPollInterval = 200 * time.Millisecond + statusHoldWindow = 3 * time.Second +) + +func registerVideoBeingTranscoded(t gobdd.StepTest, ctx gobdd.Context) { + w := worldOf(t, ctx) + + requestUploadSlot(t, ctx, "transcoding.mp4", "video/mp4") + uploadTheFile(t, ctx) + registerUploadedVideo(t, ctx, "A Video Being Transcoded", "other") + + 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) + + if strings.TrimSpace(w.video.MediaConvertJobID) == "" { + t.Fatalf("no transcoding job has been started, so there is no job to report on") + return + } + publishJobState(t, w.video.MediaConvertJobID, state) +} + +func mediaConvertReportsStateForJob(t gobdd.StepTest, ctx gobdd.Context, jobID, state string) { + worldOf(t, ctx) // keeps the step honest about needing a scenario world + publishJobState(t, jobID, state) +} + +func publishJobState(t gobdd.StepTest, jobID, state string) { + background := context.Background() + + publisher, err := newEventPublisher(background) + if err != nil { + t.Fatalf("could not reach the job events topic: %s", err) + return + } + if err := publisher.publishJobState(background, jobID, state); err != nil { + t.Fatalf("could not publish the job state change: %s", err) + } +} + +func theVideoEventuallyHasStatus(t gobdd.StepTest, ctx gobdd.Context, want string) { + w := worldOf(t, ctx) + + deadline := time.Now().Add(statusSettleTimeout) + last := "" + for time.Now().Before(deadline) { + last = currentStatus(t, w) + if last == want { + return + } + time.Sleep(statusPollInterval) + } + + t.Errorf("the video never reached status %q within %s; it is still %q", want, statusSettleTimeout, last) +} + +func theVideoKeepsStatus(t gobdd.StepTest, ctx gobdd.Context, want string) { + w := worldOf(t, ctx) + + // Held rather than sampled once: the consumer is asynchronous, so a status + // that is still "processing" the instant after publishing proves nothing. + deadline := time.Now().Add(statusHoldWindow) + for time.Now().Before(deadline) { + if got := currentStatus(t, w); got != want { + t.Errorf("the video moved to status %q, but this event should have left it %q", got, want) + return + } + time.Sleep(statusPollInterval) + } +} + +// currentStatus re-reads the video over the API. It deliberately does not +// touch w.last: these steps assert on the record, not on an exchange. +func currentStatus(t gobdd.StepTest, w *world) string { + result, err := w.client.get("/api/videos/" + w.video.ID) + if err != nil { + t.Fatalf("could not re-read the video: %s", err) + return "" + } + if result.status != 200 { + t.Fatalf("could not re-read the video: %s", result.summary()) + return "" + } + + var body videoBody + if err := result.json(&body); err != nil { + t.Fatalf("could not decode the video: %s", err) + return "" + } + return body.Status +} + +// askForThatVideoWrittenAs re-reads the video by a different, still-valid +// spelling of the same uuid. +func askForThatVideoWrittenAs(t gobdd.StepTest, ctx gobdd.Context, form string) { + w := worldOf(t, ctx) + + if w.video.ID == "" { + t.Fatalf("no video has been registered, so there is no id to rewrite") + return + } + + var id string + switch form { + case "braced": + id = "{" + w.video.ID + "}" + case "unhyphenated": + id = strings.ReplaceAll(w.video.ID, "-", "") + case "as a urn": + id = "urn:uuid:" + w.video.ID + default: + t.Fatalf("no such spelling of a uuid: %q", form) + return + } + + askForVideoWithID(t, ctx, id) + if w.last.status == 200 { + if err := w.last.json(&w.video); err != nil { + t.Fatalf("could not decode the video: %s", err) + } + } +} diff --git a/tests/suite_test.go b/tests/suite_test.go index f38928b..046379a 100644 --- a/tests/suite_test.go +++ b/tests/suite_test.go @@ -139,6 +139,15 @@ func TestVideoUpload(t *testing.T) { suite.AddStep(`^I request an upload slot for "(.*)" of type "(.*)"$`, requestUploadSlot) suite.AddStep(`^I upload the file to the upload URL$`, uploadTheFile) suite.AddStep(`^I send malformed JSON to the upload slot endpoint$`, sendMalformedJSON) + 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$`, registerVideoBeingTranscoded) + suite.AddStep(`^I ask the CMS for that video with its id written (.*)$`, askForThatVideoWrittenAs) + suite.AddStep(`^MediaConvert reports that the job reached "(.*)"$`, mediaConvertReportsState) + suite.AddStep(`^MediaConvert reports that the job "(.*)" reached "(.*)"$`, mediaConvertReportsStateForJob) + suite.AddStep(`^the video eventually has status "(.*)"$`, theVideoEventuallyHasStatus) + suite.AddStep(`^the video keeps status "(.*)"$`, theVideoKeepsStatus) 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)