diff --git a/CLAUDE.md b/CLAUDE.md index 0e663d3..265ad9a 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -23,10 +23,12 @@ catalogue search — see below) 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, and `0003`–`0005` add what the search needs. It now has an `internal/services` (SQS only) and an +actually starts, and `0003`–`0005` add what the search needs. It now has an `internal/services` (SQS and Redis) 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. +topic: it is a subscriber, not a participant. The Redis client needs nothing +from that role: ElastiCache is reached over the Redis protocol on the private +network, so the security group is what grants access, not IAM. ## Commands @@ -49,7 +51,7 @@ never had a v1. ```bash cd discovery -go run . # serves on :8080 (requires DB_* plus AWS_REGION/CATALOGUE_EVENTS_QUEUE_URL — no S3/MediaConvert) +go run . # serves on :8080 (requires DB_* plus AWS_REGION/CATALOGUE_EVENTS_QUEUE_URL/REDIS_ADDR — no S3/MediaConvert) go run . migrate # applies pending DB migrations, then exits (no HTTP server) go build ./... go vet ./... @@ -58,7 +60,8 @@ 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, plus the AWS_* vars its consumer needs — and, like +role LocalStack provisions, plus the AWS_* vars its consumer needs and +`REDIS_ADDR` pointing at the `redis` container — 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 @@ -124,6 +127,41 @@ browse path (no title term) is a true index seek — that is what `videos_recent_idx` is for, and it takes a 200k-row browse from ~15 ms to ~0.03 ms. +#### Search cache (Redis / ElastiCache) + +Because of that caveat, the search reads through a Redis cache before it +touches Postgres — `handlers.cachedSearch`/`cacheSearch` around the repository +call, storing the `api.SearchResults` a page serialises to. `REDIS_ADDR` points +at the ElastiCache node `infrastructure/main.go` provisions (a `redis` +container under compose). Locally it takes a repeated ranked search from +~6.8 ms to ~0.5 ms. + +- **The key is `discovery:search:v1:` + a canonical JSON encoding of + `{t: title, c: categories, l: limit, p: cursor}`.** The cursor *is* the page + identity — keyset paging has no page number to key on. `limit` is in the key + because the same title/categories/cursor at a different limit is a different + set of rows. JSON rather than pasted-together separators so a category name + containing `,` or `|` cannot be mistaken for a boundary. The title is + lowercased (`tsQueryFor` folds case anyway) but the **categories are not**: + `categories && $2` compares them verbatim, so `News` and `news` really are + different searches. Categories are sorted and deduped — `&&` means "any of + these", so order never changed the answer — and never nil, so `"categories": + []` and an omitted member share an entry. +- **TTL is 60 s and nothing invalidates on write.** The consumer writes rows + continuously and would have to know which cached pages a new title belongs + on — for a ranked search, every page it outranks. Expiry is cheaper and + bounds staleness to roughly the announcement's own delivery lag. +- **Only 200s are cached**, so a malformed cursor still reaches the repository + and is still a 400. +- **Every cache failure is a miss.** A Get/Set error is logged and the search + is answered from Postgres; the service also *starts* with an unreachable + cache, warning rather than panicking as the DB and SQS assertions do — a slow + read side beats no read side. Each call is bounded by + `services.cacheOperationTimeout` (100 ms, retries off), because an unbounded + cache miss on a dead node costs the dial timeout *plus* the query it was + avoiding. A total cache outage adds ~200 ms per search (a failed read and a + failed write), not seconds. + `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. @@ -353,6 +391,18 @@ on boot if any required var is empty. backups — intentionally minimal). Each app (`cms`, `discovery`) gets its own login role and same-named database via the `postgresql` provider (`newServiceDatabase`), so services never share DB credentials. +- **ElastiCache**: one `cache.t4g.micro` Redis node (`search-cache`, engine + 7.1, single-AZ, no replica, no snapshots) that `discovery` answers repeated + catalogue searches from. Its contents are derivable from Postgres by + definition, so there is nothing to back up. Its security group admits 6379 + from `ecs-service-sg` **only** — narrower than the database's, which also + admits the deployer's IP for the `postgresql` provider; there is nothing to + administer here from a laptop. The endpoint is the single node's address + (`CacheNodes[0]`, not `ConfigurationEndpoint` — that is a Memcached thing), + exported as `searchCacheAddress` and injected as `REDIS_ADDR`. Skipped under + LocalStack, where docker-compose runs a plain `redis:7-alpine` container + instead: ElastiCache picks its own endpoint, and compose needs a literal + `REDIS_ADDR` before anything is provisioned. - **ECS Fargate**: one cluster (`app-cluster`), one ALB *per service* (each gets its own DNS name rather than sharing a load balancer on different ports). `deployFargateService(...)` is the shared helper building a @@ -402,7 +452,9 @@ on boot if any required var is empty. - `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. + — it consumes the catalogue, it does not add to it. The search cache is + absent from it on purpose: ElastiCache is reached over the Redis protocol + inside the VPC, so `search-cache-sg` is the grant, not IAM. - `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/discovery/go.mod b/discovery/go.mod index 032b50d..2b45d2b 100644 --- a/discovery/go.mod +++ b/discovery/go.mod @@ -3,8 +3,11 @@ module thamanyah/discovery go 1.25.12 require ( + github.com/aws/aws-sdk-go-v2/config v1.33.1 + github.com/aws/aws-sdk-go-v2/service/sqs v1.48.1 github.com/golang-migrate/migrate/v4 v4.19.1 github.com/lib/pq v1.12.3 + github.com/redis/go-redis/v9 v9.22.0 github.com/swaggo/http-swagger/v2 v2.0.2 github.com/swaggo/swag v1.16.6 ) @@ -12,7 +15,6 @@ 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 @@ -21,11 +23,11 @@ require ( 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/cespare/xxhash/v2 v2.3.0 // 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 @@ -33,8 +35,10 @@ require ( github.com/josharian/intern v1.0.0 // indirect github.com/mailru/easyjson v0.7.6 // indirect github.com/swaggo/files/v2 v2.0.0 // indirect + go.uber.org/atomic v1.11.0 // indirect golang.org/x/mod v0.29.0 // indirect golang.org/x/sync v0.18.0 // indirect + golang.org/x/sys v0.38.0 // indirect golang.org/x/tools v0.38.0 // indirect gopkg.in/yaml.v2 v2.4.0 // indirect ) diff --git a/discovery/go.sum b/discovery/go.sum index 10fbae7..56c2079 100644 --- a/discovery/go.sum +++ b/discovery/go.sum @@ -34,6 +34,12 @@ github.com/aws/aws-sdk-go-v2/service/sts v1.47.1 h1:Sv2xPnRHlThSUtVujYuUBPI/Il8s 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/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs= +github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c= +github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA= +github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0= +github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= +github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= 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= @@ -77,6 +83,8 @@ 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/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/klauspost/cpuid/v2 v2.2.10 h1:tBs3QSyvjDyFTq3uoc/9xFpCuOsJQFNPiAhYdw2skhE= +github.com/klauspost/cpuid/v2 v2.2.10/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0= github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo= github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= @@ -105,6 +113,8 @@ github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINE github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U= github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/redis/go-redis/v9 v9.22.0 h1:laDvpYXTJtZLloinw1fA5Kqd6HAEH2XKxOkG/PDq2F0= +github.com/redis/go-redis/v9 v9.22.0/go.mod h1:y2g0Wj8rQvuK0ELM+oxSudcLtC09JScs98I/X9gRWY4= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= @@ -116,6 +126,8 @@ github.com/swaggo/http-swagger/v2 v2.0.2 h1:FKCdLsl+sFCx60KFsyM0rDarwiUSZ8DqbfSy github.com/swaggo/http-swagger/v2 v2.0.2/go.mod h1:r7/GBkAWIfK6E/OLnE8fXnviHiDeAHmgIyooa4xm3AQ= github.com/swaggo/swag v1.16.6 h1:qBNcx53ZaX+M5dxVyTrgQ0PJ/ACK+NzhwcbieTt+9yI= github.com/swaggo/swag v1.16.6/go.mod h1:ngP2etMK5a0P3QBizic5MEwpRmluJZPHjXcMoj4Xesg= +github.com/zeebo/xxh3 v1.1.0 h1:s7DLGDK45Dyfg7++yxI0khrfwq9661w9EN78eP/UZVs= +github.com/zeebo/xxh3 v1.1.0/go.mod h1:IisAie1LELR4xhVinxWS5+zf1lA4p0MW4T+w+W07F5s= go.opentelemetry.io/auto/sdk v1.1.0 h1:cH53jehLUN6UFLY71z+NDOiNJqDdPRaXzTel0sJySYA= go.opentelemetry.io/auto/sdk v1.1.0/go.mod h1:3wSPjt5PWp2RhlCcmmOial7AvC4DQqZb7a7wCow3W8A= go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.61.0 h1:F7Jx+6hwnZ41NSFTO5q4LYDtJRXBf2PD0rNBkeB/lus= @@ -126,6 +138,8 @@ go.opentelemetry.io/otel/metric v1.37.0 h1:mvwbQS5m0tbmqML4NqK+e3aDiO02vsf/Wgbsd go.opentelemetry.io/otel/metric v1.37.0/go.mod h1:04wGrZurHYKOc+RKeye86GwKiTb9FKm1WHtO+4EVr2E= go.opentelemetry.io/otel/trace v1.37.0 h1:HLdcFNbRQBE2imdSEgm/kwqmQj1Or1l/7bW6mxVK7z4= go.opentelemetry.io/otel/trace v1.37.0/go.mod h1:TlgrlQ+PtQO5XFerSPUYG0JSgGyryXewPGyayAWSBS0= +go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE= +go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0= golang.org/x/mod v0.29.0 h1:HV8lRxZC4l2cr3Zq1LvtOsi/ThTgWnUk/y64QSs8GwA= golang.org/x/mod v0.29.0/go.mod h1:NyhrlYXJ2H4eJiRy/WDBO6HMqZQ6q9nk4JzS3NuCK+w= golang.org/x/sync v0.18.0 h1:kr88TuHDroi+UVf+0hZnirlk8o8T+4MrK6mr60WkH/I= diff --git a/discovery/internal/handlers/videos.go b/discovery/internal/handlers/videos.go index c804283..26fd6d8 100644 --- a/discovery/internal/handlers/videos.go +++ b/discovery/internal/handlers/videos.go @@ -1,12 +1,17 @@ package handlers import ( + "context" "encoding/json" "errors" "log" "net/http" + "slices" + "strings" "thamanyah/discovery/internal/api" "thamanyah/discovery/internal/db/repositories" + "thamanyah/discovery/internal/services" + "time" ) // GetVideo serves the catalogue's copy of one video. @@ -64,8 +69,112 @@ const ( // cursor still reaches the rest — because a client that wants everything // is doing something legitimate, just not in one request. maxSearchPageSize = 100 + + // searchCacheTTL is how long a page of results stays servable from the + // cache before it has to be built again. + // + // Nothing invalidates an entry when a video is announced: the consumer + // writes rows continuously and would have to know which of the cached + // pages a new title belongs on, which for a ranked search is every page it + // outranks. Expiry is the cheaper answer, and it bounds the staleness — a + // video is findable within a minute of being catalogued, which is the same + // order as the announcement's own delivery lag. + searchCacheTTL = time.Minute ) +// searchCacheKey is what one page of search results is stored under: the title +// searched for, the categories it was narrowed to, and the page within that +// result set. +// +// The page is named by the cursor, because with keyset paging that is what a +// page *is* — there is no page number to key on, and the cursor identifies the +// same rows whenever it is presented. The empty cursor is the first page. +// +// The limit is in the key too, even though a reader would not call it part of +// "which page". It has to be: the same title, categories and cursor with a +// different limit is a different set of rows, and sharing an entry between +// them would hand a client asking for 50 a page of 20. +// +// The parts are JSON-encoded rather than pasted together with separators so +// that no category name can be mistaken for the boundary between two of them — +// {"c":["a,b"]} and {"c":["a","b"]} stay distinguishable, which "a,b" and +// "a,b" would not. The key stays readable in redis-cli either way. +func searchCacheKey(title string, categories []string, limit int, cursor string) string { + // The title is lowercased because the search itself is: tsQueryFor folds + // case before it builds the tsquery, so two spellings that differ only in + // case are the same search and should be the same entry. The categories + // are not — they are compared to the stored array verbatim, so "News" and + // "news" really do ask different questions. + // + // Sorting the categories is safe for the same reason the SQL uses &&: + // they mean "any of these", so their order never changed the answer. Two + // readers naming the same categories in a different order now share one + // entry instead of building the same page twice. + // + // Never nil, so that a client sending "categories": [] and one omitting + // the member entirely — the same search, since both mean "every category" + // — land on one entry rather than on `null` and `[]`. + sorted := append([]string{}, categories...) + slices.Sort(sorted) + sorted = slices.Compact(sorted) + + parts := struct { + Title string `json:"t"` + Categories []string `json:"c"` + Limit int `json:"l"` + Cursor string `json:"p"` + }{ + Title: strings.ToLower(strings.TrimSpace(title)), + Categories: sorted, + Limit: limit, + Cursor: cursor, + } + + // Cannot fail: every field is a string, a string slice or an int. + encoded, _ := json.Marshal(parts) + return "discovery:search:v1:" + string(encoded) +} + +// cachedSearch returns the page held under key, or false when there is none to +// serve. +// +// Every failure reads as "no cached page": a cache that is unreachable, slow +// or holding something unreadable must cost a reader nothing more than the +// database query they would have paid for anyway. Only faults worth acting on +// are logged — a miss is not one. +func cachedSearch(ctx context.Context, key string) (api.SearchResults, bool) { + encoded, err := services.CacheClient.Get(ctx, key) + if errors.Is(err, services.ErrCacheMiss) { + return api.SearchResults{}, false + } + if err != nil { + log.Printf("Could Not Read The Search Cache, Falling Back To The Database: key=%q: %s", key, err) + return api.SearchResults{}, false + } + + var results api.SearchResults + if err := json.Unmarshal(encoded, &results); err != nil { + // Only reachable if something else wrote this key, or the shape + // changed without the v1 in the prefix changing with it. + log.Printf("Could Not Read A Cached Search Page, Falling Back To The Database: key=%q: %s", key, err) + return api.SearchResults{}, false + } + + return results, true +} + +func cacheSearch(ctx context.Context, key string, results api.SearchResults) { + encoded, err := json.Marshal(results) + if err != nil { + log.Printf("Could Not Encode A Search Page For The Cache: key=%q: %s", key, err) + return + } + + if err := services.CacheClient.Set(ctx, key, encoded, searchCacheTTL); err != nil { + log.Printf("Could Not Write The Search Cache: key=%q: %s", key, err) + } +} + // SearchVideos serves the catalogue's lexical search. // // It is served on QUERY rather than GET or POST: a search is safe and @@ -94,6 +203,19 @@ func SearchVideos(w http.ResponseWriter, r *http.Request) { limit = maxSearchPageSize } + // Read through the cache before touching Postgres. A ranked search is the + // expensive request this service serves — ts_rank is computed per matching + // row, so deep pages scan rather than seek — and readers ask for the same + // few things over and over, which is exactly the shape a cache pays for. + // + // Only successful pages are ever stored, so a malformed cursor still + // reaches the repository and is still rejected below. + cacheKey := searchCacheKey(request.Title, request.Categories, limit, request.Cursor) + if cached, hit := cachedSearch(r.Context(), cacheKey); hit { + writeJSON(w, http.StatusOK, cached) + return + } + found, next, err := repositories.VideoRepo.SearchVideos(r.Context(), request.Title, request.Categories, limit, request.Cursor) if errors.Is(err, repositories.ErrInvalidCursor) { writeProblem(w, http.StatusBadRequest, "Malformed Search Cursor", @@ -119,5 +241,8 @@ func SearchVideos(w http.ResponseWriter, r *http.Request) { }) } - writeJSON(w, http.StatusOK, api.SearchResults{Videos: videos, NextCursor: next}) + results := api.SearchResults{Videos: videos, NextCursor: next} + cacheSearch(r.Context(), cacheKey, results) + + writeJSON(w, http.StatusOK, results) } diff --git a/discovery/internal/services/cache.go b/discovery/internal/services/cache.go new file mode 100644 index 0000000..16ae515 --- /dev/null +++ b/discovery/internal/services/cache.go @@ -0,0 +1,58 @@ +package services + +import ( + "context" + "errors" + "fmt" + "log" + "time" + + "github.com/redis/go-redis/v9" +) + +var CacheClient Cache + +var ErrCacheMiss = errors.New("cache: miss") + +type Cache interface { + Get(ctx context.Context, key string) ([]byte, error) + Set(ctx context.Context, key string, value []byte, ttl time.Duration) error +} + +type RedisConcrete struct { + Redis *redis.Client +} + +const cacheOperationTimeout = 100 * time.Millisecond + +func (svc RedisConcrete) Get(ctx context.Context, key string) ([]byte, error) { + ctx, cancel := context.WithTimeout(ctx, cacheOperationTimeout) + defer cancel() + + value, err := svc.Redis.Get(ctx, key).Bytes() + if errors.Is(err, redis.Nil) { + return nil, ErrCacheMiss + } + if err != nil { + return nil, err + } + return value, nil +} + +func (svc RedisConcrete) Set(ctx context.Context, key string, value []byte, ttl time.Duration) error { + ctx, cancel := context.WithTimeout(ctx, cacheOperationTimeout) + defer cancel() + + return svc.Redis.Set(ctx, key, value, ttl).Err() +} + +func (svc RedisConcrete) Close() error { + return svc.Redis.Close() +} + +func (svc RedisConcrete) AssertSuccessfulConnection(ctx context.Context) { + if err := svc.Redis.Ping(ctx).Err(); err != nil { + log.Printf("WARNING: %s", fmt.Errorf("redis: cannot reach cache at %q, searches will be served from the database: %w", + svc.Redis.Options().Addr, err)) + } +} diff --git a/discovery/main.go b/discovery/main.go index 74664d5..433a4bb 100644 --- a/discovery/main.go +++ b/discovery/main.go @@ -11,10 +11,12 @@ import ( "thamanyah/discovery/internal/db/repositories" "thamanyah/discovery/internal/handlers" "thamanyah/discovery/internal/services" + "time" awsconfig "github.com/aws/aws-sdk-go-v2/config" "github.com/aws/aws-sdk-go-v2/service/sqs" _ "github.com/lib/pq" + "github.com/redis/go-redis/v9" httpSwagger "github.com/swaggo/http-swagger/v2" _ "thamanyah/discovery/docs" @@ -65,6 +67,38 @@ func runServer() { services.SQSClient = concreteSQSClient + // The ElastiCache Redis node the catalogue search is answered from. Not an + // AWS API call — the address resolves inside the VPC and the security + // group is what grants access — so it needs nothing from the task role and + // nothing from awsConfig. + // + // The address is required because a missing one is a deployment mistake, + // not a choice; whether the node actually answers is a different question, + // and one the service deliberately survives getting "no" to. + redisAddress := os.Getenv("REDIS_ADDR") + if redisAddress == "" { + panic(fmt.Errorf("missing required env var: REDIS_ADDR")) + } + + concreteCacheClient := &services.RedisConcrete{ + Redis: redis.NewClient(&redis.Options{ + Addr: redisAddress, + // Per-call deadlines on top of the budget RedisConcrete already + // applies, so a connection that hangs rather than refusing cannot + // tie up a goroutine past the request that opened it. Retries are + // off for the same reason the budget is small: a second attempt at + // a cache costs more than the query it is saving. + DialTimeout: 100 * time.Millisecond, + ReadTimeout: 100 * time.Millisecond, + WriteTimeout: 100 * time.Millisecond, + MaxRetries: -1, + }), + } + defer func() { _ = concreteCacheClient.Close() }() + concreteCacheClient.AssertSuccessfulConnection(context.Background()) + + services.CacheClient = concreteCacheClient + concreteDBClient := db.CreateDBConnection(requireDBConnectionString()) defer db.CloseConnection(concreteDBClient) db.AssertSuccessfulConnection(context.Background(), concreteDBClient) diff --git a/docker-compose.yml b/docker-compose.yml index 42dbba9..573d935 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -126,6 +126,23 @@ services: - action: rebuild path: ./cms + # Stands in for the ElastiCache node infrastructure/main.go provisions on the + # AWS stacks. Not provisioned through LocalStack like the buckets and the + # database are: ElastiCache hands out an endpoint on a port it chooses, and + # the literal REDIS_ADDR below has to be known before anything is created. + # Nothing here is worth persisting — every entry is derivable from Postgres, + # which is the whole point of a cache — so there is no volume. + redis: + container_name: "${REDIS_DOCKER_NAME:-redis}" + image: redis:7-alpine + ports: + - "127.0.0.1:6379:6379" + healthcheck: + test: ["CMD", "redis-cli", "ping"] + interval: 5s + timeout: 3s + retries: 10 + discovery: container_name: "${DISCOVERY_DOCKER_NAME:-discovery}" build: @@ -143,6 +160,10 @@ services: - 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} + # The catalogue search's cache — the redis service below, standing in for + # ElastiCache. main.go panics without an address but only warns if the + # node does not answer: a cold cache means slower searches, not none. + - REDIS_ADDR=${REDIS_ADDR:-redis:6379} # 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} @@ -154,6 +175,8 @@ services: depends_on: infra: condition: service_completed_successfully + redis: + condition: service_healthy develop: watch: # Same as cms: a compiled binary, so every change means a new image. diff --git a/infrastructure/main.go b/infrastructure/main.go index 3736a4c..f1da8bb 100644 --- a/infrastructure/main.go +++ b/infrastructure/main.go @@ -13,6 +13,7 @@ import ( "github.com/pulumi/pulumi-aws/sdk/v7/go/aws/ec2" "github.com/pulumi/pulumi-aws/sdk/v7/go/aws/ecr" "github.com/pulumi/pulumi-aws/sdk/v7/go/aws/ecs" + "github.com/pulumi/pulumi-aws/sdk/v7/go/aws/elasticache" "github.com/pulumi/pulumi-aws/sdk/v7/go/aws/iam" "github.com/pulumi/pulumi-aws/sdk/v7/go/aws/lb" "github.com/pulumi/pulumi-aws/sdk/v7/go/aws/rds" @@ -384,6 +385,7 @@ func main() { cmsRepo, discoveryRepo *ecr.Repository cmsService, discoveryService *ecs.Service cmsAlb, discoveryAlb *lb.LoadBalancer + searchCacheAddress pulumi.StringOutput ) // Bucket names are left to Pulumi's auto-naming in AWS, but pinned under @@ -771,6 +773,67 @@ func main() { } } + if !localstack { + cacheSubnetGroup, err := elasticache.NewSubnetGroup(ctx, "search-cache-subnet-group", &elasticache.SubnetGroupArgs{ + SubnetIds: pulumi.ToStringArray(subnetIDs), + }) + if err != nil { + return err + } + + cacheSecurityGroup, err := ec2.NewSecurityGroup(ctx, "search-cache-sg", &ec2.SecurityGroupArgs{ + Description: pulumi.String("Allow the ECS tasks to reach the search cache"), + VpcId: pulumi.String(vpcID), + Ingress: ec2.SecurityGroupIngressArray{ + &ec2.SecurityGroupIngressArgs{ + Protocol: pulumi.String("tcp"), + FromPort: pulumi.Int(6379), + ToPort: pulumi.Int(6379), + SecurityGroups: pulumi.StringArray{serviceSecurityGroup.ID()}, + Description: pulumi.String("Redis, from the service tasks only"), + }, + }, + Egress: ec2.SecurityGroupEgressArray{ + &ec2.SecurityGroupEgressArgs{ + Protocol: pulumi.String("-1"), + FromPort: pulumi.Int(0), + ToPort: pulumi.Int(0), + CidrBlocks: pulumi.ToStringArray([]string{"0.0.0.0/0"}), + }, + }, + }) + if err != nil { + return err + } + + searchCache, err := elasticache.NewCluster(ctx, "search-cache", &elasticache.ClusterArgs{ + Engine: pulumi.String("redis"), + EngineVersion: pulumi.String("7.1"), + NodeType: pulumi.String("cache.t4g.micro"), + NumCacheNodes: pulumi.Int(1), + ParameterGroupName: pulumi.String("default.redis7"), + Port: pulumi.Int(6379), + SubnetGroupName: cacheSubnetGroup.Name, + SecurityGroupIds: pulumi.StringArray{cacheSecurityGroup.ID()}, + ApplyImmediately: pulumi.Bool(true), + }) + if err != nil { + return err + } + + // A single-node Redis cluster has no configuration endpoint — that + // is a Memcached thing — so the address is the one cache node's. + searchCacheAddress = pulumi.All(searchCache.CacheNodes, searchCache.Port).ApplyT( + func(args []any) (string, error) { + nodes := args[0].([]elasticache.ClusterCacheNode) + if len(nodes) == 0 || nodes[0].Address == nil { + return "", fmt.Errorf("search cache reported no node address") + } + return fmt.Sprintf("%s:%d", *nodes[0].Address, args[1].(int)), nil + }, + ).(pulumi.StringOutput) + } + // Private bucket for raw video uploads (pre-transcode). Kept separate // from encoded-bucket, which is fronted by CloudFront/OAC for public // delivery of finished renditions — raw source video must not be @@ -1359,6 +1422,10 @@ func main() { discoveryExtraEnv := []envVar{ {Name: "AWS_REGION", Value: pulumi.String(awsRegion).ToStringOutput()}, {Name: "CATALOGUE_EVENTS_QUEUE_URL", Value: catalogueQueue.Url}, + // The search cache. Reached over the Redis protocol on the + // private network, so unlike the queue above it needs no + // matching grant on discovery-task-role. + {Name: "REDIS_ADDR", Value: searchCacheAddress}, } discoveryRepo, discoveryService, discoveryAlb, err = deployFargateService(ctx, "discovery", 8080, @@ -1452,6 +1519,7 @@ func main() { ctx.Export("ecsClusterArn", cluster.Arn) ctx.Export("cmsServiceArn", cmsService.Arn) ctx.Export("discoveryServiceArn", discoveryService.Arn) + ctx.Export("searchCacheAddress", searchCacheAddress) } return nil })