// app/persistence/redis.go // Redis persistence layer для shared-sqs — write-through cache. // Стратегия: память — источник правды для чтения (быстро), // Redis — источник правды для восстановления после рестарта. // Все записи в Redis асинхронны (горутина) — не блокируют SQS-операции. // Сериализация (json.Marshal) происходит синхронно пока вызывающий держит мьютекс — консистентный снапшот. // Created: 2026-04-10 package persistence import ( "context" "encoding/json" "fmt" "time" "github.com/redis/go-redis/v9" log "github.com/sirupsen/logrus" "shared-sqs/app/models" ) // Client — глобальный Redis клиент. // nil означает режим "только память" — все функции тихо no-op. var Client *redis.Client const ( // redisHashTenants — HASH: tenantID → JSON тенанта redisHashTenants = "ssq:tenants" // redisHashQueues — HASH: queueKey → JSON очереди (включая сообщения) redisHashQueues = "ssq:queues" ) // Connect — подключается к Redis и проверяет ping. // Если addr пустой — не подключается, остаёмся в memory-only режиме. func Connect(addr, username, password string) error { if addr == "" { log.Info("persistence: REDIS_ADDR не задан, работаем в memory-only режиме") return nil } rdb := redis.NewClient(&redis.Options{ Addr: addr, Username: username, Password: password, DB: 0, }) ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() if err := rdb.Ping(ctx).Err(); err != nil { return fmt.Errorf("redis ping %s: %w", addr, err) } Client = rdb log.Infof("persistence: подключились к Redis %s", addr) return nil } // asyncWrite — запускает fn в горутине, перехватывает panic и логирует. // Используется для записей в Redis чтобы не задерживать SQS-ответы. func asyncWrite(fn func()) { go func() { defer func() { if r := recover(); r != nil { log.Errorf("persistence: panic в asyncWrite: %v", r) } }() fn() }() } // SaveQueue — сохраняет очередь (с сообщениями) в Redis асинхронно. // ВАЖНО: вызывать пока вызывающий держит SyncQueues.Lock() — тогда json.Marshal // создаёт консистентный снапшот. Горутина только делает сетевой вызов. func SaveQueue(key string, queue *models.Queue) { if Client == nil { return } // Сериализуем синхронно под мьютексом вызывающего → консистентный снапшот data, err := json.Marshal(queue) if err != nil { log.Errorf("persistence: marshal queue %q: %v", key, err) return } asyncWrite(func() { ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) defer cancel() if err := Client.HSet(ctx, redisHashQueues, key, string(data)).Err(); err != nil { log.Errorf("persistence: HSet queue %q: %v", key, err) } }) } // DeleteQueue — удаляет очередь из Redis асинхронно. func DeleteQueue(key string) { if Client == nil { return } asyncWrite(func() { ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) defer cancel() if err := Client.HDel(ctx, redisHashQueues, key).Err(); err != nil { log.Errorf("persistence: HDel queue %q: %v", key, err) } }) } // LoadAllQueues — загружает все очереди из Redis в память при старте сервиса. // Инициализирует nil-maps чтобы избежать panic при deduplication/FIFO операциях. func LoadAllQueues() (map[string]*models.Queue, error) { if Client == nil { return nil, nil } ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() raw, err := Client.HGetAll(ctx, redisHashQueues).Result() if err != nil { return nil, fmt.Errorf("redis HGetAll queues: %w", err) } queues := make(map[string]*models.Queue, len(raw)) for k, v := range raw { var q models.Queue if err := json.Unmarshal([]byte(v), &q); err != nil { log.Errorf("persistence: unmarshal queue %q: %v", k, err) continue } // Инициализируем nil-maps — их json.Unmarshal не создаёт если поле было nil if q.Duplicates == nil { q.Duplicates = make(map[string]time.Time) } if q.FIFOMessages == nil { q.FIFOMessages = make(map[string]int) } if q.FIFOSequenceNumbers == nil { q.FIFOSequenceNumbers = make(map[string]int) } queues[k] = &q } log.Infof("persistence: загружено %d очередей из Redis", len(queues)) return queues, nil } // SaveTenantRaw — сохраняет тенанта (сырой JSON) в Redis асинхронно. // Принимает []byte чтобы избежать циклического импорта с пакетом tenant. // Сериализацию делает вызывающий (tenant_store.go). func SaveTenantRaw(id string, jsonData []byte) { if Client == nil { return } // Копируем bytes — вызывающий может переиспользовать буфер dataCopy := make([]byte, len(jsonData)) copy(dataCopy, jsonData) asyncWrite(func() { ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) defer cancel() if err := Client.HSet(ctx, redisHashTenants, id, string(dataCopy)).Err(); err != nil { log.Errorf("persistence: HSet tenant %q: %v", id, err) } }) } // DeleteTenant — удаляет тенанта из Redis асинхронно. func DeleteTenant(id string) { if Client == nil { return } asyncWrite(func() { ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) defer cancel() if err := Client.HDel(ctx, redisHashTenants, id).Err(); err != nil { log.Errorf("persistence: HDel tenant %q: %v", id, err) } }) } // LoadAllTenantsRaw — загружает всех тенантов из Redis при старте. // Возвращает map[tenantID]rawJSON — десериализацию делает tenant_store.go. func LoadAllTenantsRaw() (map[string][]byte, error) { if Client == nil { return nil, nil } ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() raw, err := Client.HGetAll(ctx, redisHashTenants).Result() if err != nil { return nil, fmt.Errorf("redis HGetAll tenants: %w", err) } result := make(map[string][]byte, len(raw)) for id, v := range raw { result[id] = []byte(v) } log.Infof("persistence: загружено %d тенантов из Redis", len(result)) return result, nil }