feat(step3): тенанты read-through Redis (индекс accessKey→tenant, fallback в auth, бэкфилл на старте)
This commit is contained in:
@@ -10,6 +10,7 @@ import (
|
|||||||
"crypto/sha256"
|
"crypto/sha256"
|
||||||
"crypto/subtle"
|
"crypto/subtle"
|
||||||
"encoding/hex"
|
"encoding/hex"
|
||||||
|
"encoding/json"
|
||||||
"encoding/xml"
|
"encoding/xml"
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
@@ -21,6 +22,7 @@ import (
|
|||||||
"strings"
|
"strings"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
|
"shared-sqs/app/persistence"
|
||||||
"shared-sqs/app/tenant"
|
"shared-sqs/app/tenant"
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -59,6 +61,17 @@ func AuthMiddleware(store *tenant.TenantStore) func(http.Handler) http.Handler {
|
|||||||
}
|
}
|
||||||
|
|
||||||
t, ok := store.GetByAccessKey(accessKeyID)
|
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 {
|
if !ok || !t.Active {
|
||||||
writeSQSAuthError(w, "InvalidClientTokenId", "The security token included in the request is invalid")
|
writeSQSAuthError(w, "InvalidClientTokenId", "The security token included in the request is invalid")
|
||||||
return
|
return
|
||||||
|
|||||||
@@ -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
|
// Автосид демо-данных при SHARED_SQS_SEED_DEMO=true
|
||||||
if os.Getenv("SHARED_SQS_SEED_DEMO") == "true" {
|
if os.Getenv("SHARED_SQS_SEED_DEMO") == "true" {
|
||||||
seedDemoData(tenantStore)
|
seedDemoData(tenantStore)
|
||||||
|
|||||||
@@ -48,6 +48,8 @@ var (
|
|||||||
const (
|
const (
|
||||||
// redisHashTenants — HASH: tenantID → JSON тенанта
|
// redisHashTenants — HASH: tenantID → JSON тенанта
|
||||||
redisHashTenants = "ssq:tenants"
|
redisHashTenants = "ssq:tenants"
|
||||||
|
// redisHashTenantAK — HASH: accessKey → JSON тенанта (индекс read-through в auth)
|
||||||
|
redisHashTenantAK = "ssq:tenant_ak"
|
||||||
// redisHashQueues — HASH: queueKey → JSON метаданных очереди (без Messages с v2)
|
// redisHashQueues — HASH: queueKey → JSON метаданных очереди (без Messages с v2)
|
||||||
redisHashQueues = "ssq:queues"
|
redisHashQueues = "ssq:queues"
|
||||||
// redisMsgHashPrefix — префикс для per-message HASH: ssq:msg:{queueKey} → uuid → JSON(SqsMessage)
|
// redisMsgHashPrefix — префикс для per-message HASH: ssq:msg:{queueKey} → uuid → JSON(SqsMessage)
|
||||||
@@ -519,6 +521,11 @@ func SaveTenantRaw(id string, jsonData []byte) {
|
|||||||
// Копируем bytes — вызывающий может переиспользовать буфер
|
// Копируем bytes — вызывающий может переиспользовать буфер
|
||||||
dataCopy := make([]byte, len(jsonData))
|
dataCopy := make([]byte, len(jsonData))
|
||||||
copy(dataCopy, jsonData)
|
copy(dataCopy, jsonData)
|
||||||
|
// Извлекаем accessKey для индекса read-through (без импорта пакета tenant)
|
||||||
|
var meta struct {
|
||||||
|
AccessKey string
|
||||||
|
}
|
||||||
|
_ = json.Unmarshal(dataCopy, &meta)
|
||||||
writeSeq := nextTenantWriteSeq(id)
|
writeSeq := nextTenantWriteSeq(id)
|
||||||
asyncWrite(func() {
|
asyncWrite(func() {
|
||||||
if !isLatestTenantWriteSeq(id, writeSeq) {
|
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 {
|
if err := Client.HSet(ctx, redisHashTenants, id, string(dataCopy)).Err(); err != nil {
|
||||||
log.Errorf("persistence: HSet tenant %q: %v", id, err)
|
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 асинхронно.
|
// DeleteTenant — удаляет тенанта из Redis асинхронно.
|
||||||
func DeleteTenant(id string) {
|
func DeleteTenant(id string) {
|
||||||
if Client == nil {
|
if Client == nil {
|
||||||
|
|||||||
Reference in New Issue
Block a user