From 8e22bc4b9b440761425e7562d5a8b338814a00d7 Mon Sep 17 00:00:00 2001 From: Repinoid Date: Sat, 10 Oct 2026 16:09:52 +0300 Subject: [PATCH] =?UTF-8?q?feat(step3):=20=D1=82=D0=B5=D0=BD=D0=B0=D0=BD?= =?UTF-8?q?=D1=82=D1=8B=20read-through=20Redis=20(=D0=B8=D0=BD=D0=B4=D0=B5?= =?UTF-8?q?=D0=BA=D1=81=20accessKey=E2=86=92tenant,=20fallback=20=D0=B2=20?= =?UTF-8?q?auth,=20=D0=B1=D1=8D=D0=BA=D1=84=D0=B8=D0=BB=D0=BB=20=D0=BD?= =?UTF-8?q?=D0=B0=20=D1=81=D1=82=D0=B0=D1=80=D1=82=D0=B5)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/auth/auth_middleware.go | 13 +++++++++++++ app/cmd/goaws.go | 10 ++++++++++ app/persistence/redis.go | 27 +++++++++++++++++++++++++++ 3 files changed, 50 insertions(+) diff --git a/app/auth/auth_middleware.go b/app/auth/auth_middleware.go index 90a0029..e2c6a53 100644 --- a/app/auth/auth_middleware.go +++ b/app/auth/auth_middleware.go @@ -10,6 +10,7 @@ import ( "crypto/sha256" "crypto/subtle" "encoding/hex" + "encoding/json" "encoding/xml" "errors" "fmt" @@ -21,6 +22,7 @@ import ( "strings" "time" + "shared-sqs/app/persistence" "shared-sqs/app/tenant" ) @@ -59,6 +61,17 @@ func AuthMiddleware(store *tenant.TenantStore) func(http.Handler) http.Handler { } t, ok := store.GetByAccessKey(accessKeyID) + if !ok { + // Read-through из Redis: тенант мог быть создан control-plane / другим + // инстансом и ещё не попасть в локальную память. + if raw, found := persistence.LoadTenantByAccessKey(accessKeyID); found { + var loaded tenant.Tenant + if err := json.Unmarshal(raw, &loaded); err == nil { + store.LoadTenant(&loaded) + t, ok = store.GetByAccessKey(accessKeyID) + } + } + } if !ok || !t.Active { writeSQSAuthError(w, "InvalidClientTokenId", "The security token included in the request is invalid") return diff --git a/app/cmd/goaws.go b/app/cmd/goaws.go index a9b47c2..335f2ab 100644 --- a/app/cmd/goaws.go +++ b/app/cmd/goaws.go @@ -124,6 +124,16 @@ func main() { } } + // Бэкфилл индекса accessKey→tenant для read-through (идемпотентно; нужен для + // тенантов, созданных до появления индекса, и для multi-instance сценария). + if persistence.Client != nil { + for _, t := range tenantStore.List() { + if data, err := json.Marshal(t); err == nil { + persistence.SaveTenantRaw(t.ID, data) + } + } + } + // Автосид демо-данных при SHARED_SQS_SEED_DEMO=true if os.Getenv("SHARED_SQS_SEED_DEMO") == "true" { seedDemoData(tenantStore) diff --git a/app/persistence/redis.go b/app/persistence/redis.go index 9710dd2..ab87379 100644 --- a/app/persistence/redis.go +++ b/app/persistence/redis.go @@ -48,6 +48,8 @@ var ( const ( // redisHashTenants — HASH: tenantID → JSON тенанта redisHashTenants = "ssq:tenants" + // redisHashTenantAK — HASH: accessKey → JSON тенанта (индекс read-through в auth) + redisHashTenantAK = "ssq:tenant_ak" // redisHashQueues — HASH: queueKey → JSON метаданных очереди (без Messages с v2) redisHashQueues = "ssq:queues" // redisMsgHashPrefix — префикс для per-message HASH: ssq:msg:{queueKey} → uuid → JSON(SqsMessage) @@ -519,6 +521,11 @@ func SaveTenantRaw(id string, jsonData []byte) { // Копируем bytes — вызывающий может переиспользовать буфер dataCopy := make([]byte, len(jsonData)) copy(dataCopy, jsonData) + // Извлекаем accessKey для индекса read-through (без импорта пакета tenant) + var meta struct { + AccessKey string + } + _ = json.Unmarshal(dataCopy, &meta) writeSeq := nextTenantWriteSeq(id) asyncWrite(func() { if !isLatestTenantWriteSeq(id, writeSeq) { @@ -529,9 +536,29 @@ func SaveTenantRaw(id string, jsonData []byte) { if err := Client.HSet(ctx, redisHashTenants, id, string(dataCopy)).Err(); err != nil { log.Errorf("persistence: HSet tenant %q: %v", id, err) } + if meta.AccessKey != "" { + if err := Client.HSet(ctx, redisHashTenantAK, meta.AccessKey, string(dataCopy)).Err(); err != nil { + log.Errorf("persistence: HSet tenant_ak %q: %v", meta.AccessKey, err) + } + } }) } +// LoadTenantByAccessKey — read-through тенанта по accessKey (для auth-middleware). +// Возвращает сырой JSON; десериализацию делает вызывающий. +func LoadTenantByAccessKey(accessKey string) ([]byte, bool) { + if Client == nil || accessKey == "" { + return nil, false + } + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + v, err := Client.HGet(ctx, redisHashTenantAK, accessKey).Result() + if err != nil { + return nil, false + } + return []byte(v), true +} + // DeleteTenant — удаляет тенанта из Redis асинхронно. func DeleteTenant(id string) { if Client == nil {