454 lines
16 KiB
Go
454 lines
16 KiB
Go
// app/persistence/redis.go
|
||
// Redis persistence layer для shared-sqs — write-through cache.
|
||
// Стратегия: память — источник правды для чтения (быстро),
|
||
// Redis — источник правды для восстановления после рестарта.
|
||
// Все записи в Redis асинхронны (горутина) — не блокируют SQS-операции.
|
||
// Сериализация (json.Marshal) происходит синхронно пока вызывающий держит мьютекс — консистентный снапшот.
|
||
//
|
||
// Схема хранения v2 (2026-04-11):
|
||
// ssq:queues (HASH) — queueKey → JSON(метаданные очереди без Messages)
|
||
// ssq:msg:{queueKey} (HASH) — uuid → JSON(SqsMessage)
|
||
// Это позволяет O(1) на каждое сообщение вместо O(N×msg_size) при SaveQueue.
|
||
// Миграция со старого формата (Messages внутри ssq:queues) — автоматическая при LoadAllQueues.
|
||
// Created: 2026-04-10 | Modified: 2026-04-11
|
||
|
||
package persistence
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"fmt"
|
||
"sync"
|
||
"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
|
||
|
||
var (
|
||
seqMu sync.Mutex
|
||
queueWriteSeq = map[string]uint64{}
|
||
tenantWriteSeq = map[string]uint64{}
|
||
)
|
||
|
||
const (
|
||
// redisHashTenants — HASH: tenantID → JSON тенанта
|
||
redisHashTenants = "ssq:tenants"
|
||
// redisHashQueues — HASH: queueKey → JSON метаданных очереди (без Messages с v2)
|
||
redisHashQueues = "ssq:queues"
|
||
// redisMsgHashPrefix — префикс для per-message HASH: ssq:msg:{queueKey} → uuid → JSON(SqsMessage)
|
||
redisMsgHashPrefix = "ssq:msg:"
|
||
)
|
||
|
||
// 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()
|
||
}()
|
||
}
|
||
|
||
// marshalQueueMeta — сериализует очередь БЕЗ Messages (и без Messages в DLQ).
|
||
// DeadLetterQueue сохраняется как ссылка (Name/URL/Arn), но без сообщений.
|
||
// ВАЖНО: вызывать под SyncQueues.Lock() — временно nil'ит Messages.
|
||
func marshalQueueMeta(queue *models.Queue) ([]byte, error) {
|
||
savedMsgs := queue.Messages
|
||
queue.Messages = nil
|
||
var savedDLQMsgs []models.SqsMessage
|
||
if queue.DeadLetterQueue != nil {
|
||
savedDLQMsgs = queue.DeadLetterQueue.Messages
|
||
queue.DeadLetterQueue.Messages = nil
|
||
}
|
||
data, err := json.Marshal(queue)
|
||
queue.Messages = savedMsgs
|
||
if queue.DeadLetterQueue != nil {
|
||
queue.DeadLetterQueue.Messages = savedDLQMsgs
|
||
}
|
||
return data, err
|
||
}
|
||
|
||
// SaveQueue — сохраняет МЕТАДАННЫЕ очереди (без сообщений) в Redis асинхронно.
|
||
// ВАЖНО: вызывать пока вызывающий держит SyncQueues.Lock() — тогда marshalQueueMeta
|
||
// создаёт консистентный снапшот. Горутина только делает сетевой вызов.
|
||
// Для сохранения сообщений используй SaveMessage/SaveMessages.
|
||
func SaveQueue(key string, queue *models.Queue) {
|
||
if Client == nil {
|
||
return
|
||
}
|
||
// Сериализуем метаданные синхронно под мьютексом вызывающего → консистентный снапшот
|
||
data, err := marshalQueueMeta(queue)
|
||
if err != nil {
|
||
log.Errorf("persistence: marshal queue meta %q: %v", key, err)
|
||
return
|
||
}
|
||
writeSeq := nextQueueWriteSeq(key)
|
||
asyncWrite(func() {
|
||
if !isLatestQueueWriteSeq(key, writeSeq) {
|
||
return
|
||
}
|
||
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 meta %q: %v", key, err)
|
||
}
|
||
})
|
||
}
|
||
|
||
// SaveMessage — сохраняет ОДНО сообщение в Redis асинхронно.
|
||
// Ключ: ssq:msg:{queueKey}, поле: msg.Uuid, значение: JSON(SqsMessage).
|
||
// Вызывать под Lock — json.Marshal делает консистентный снапшот сообщения.
|
||
func SaveMessage(queueKey string, msg *models.SqsMessage) {
|
||
if Client == nil {
|
||
return
|
||
}
|
||
data, err := json.Marshal(msg)
|
||
if err != nil {
|
||
log.Errorf("persistence: marshal message %s/%s: %v", queueKey, msg.Uuid, err)
|
||
return
|
||
}
|
||
hashKey := redisMsgHashPrefix + queueKey
|
||
asyncWrite(func() {
|
||
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
|
||
defer cancel()
|
||
if err := Client.HSet(ctx, hashKey, msg.Uuid, string(data)).Err(); err != nil {
|
||
log.Errorf("persistence: HSet message %s/%s: %v", queueKey, msg.Uuid, err)
|
||
}
|
||
})
|
||
}
|
||
|
||
// SaveMessages — сохраняет несколько сообщений в Redis одним pipeline.
|
||
// Вызывать под Lock.
|
||
func SaveMessages(queueKey string, msgs []models.SqsMessage) {
|
||
if Client == nil || len(msgs) == 0 {
|
||
return
|
||
}
|
||
// Сериализуем все сообщения синхронно под Lock
|
||
fields := make(map[string]string, len(msgs))
|
||
for i := range msgs {
|
||
data, err := json.Marshal(&msgs[i])
|
||
if err != nil {
|
||
log.Errorf("persistence: marshal message %s/%s: %v", queueKey, msgs[i].Uuid, err)
|
||
continue
|
||
}
|
||
fields[msgs[i].Uuid] = string(data)
|
||
}
|
||
if len(fields) == 0 {
|
||
return
|
||
}
|
||
hashKey := redisMsgHashPrefix + queueKey
|
||
asyncWrite(func() {
|
||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||
defer cancel()
|
||
// Конвертируем map[string]string → []interface{} для HSet
|
||
args := make([]interface{}, 0, len(fields)*2)
|
||
for k, v := range fields {
|
||
args = append(args, k, v)
|
||
}
|
||
if err := Client.HSet(ctx, hashKey, args...).Err(); err != nil {
|
||
log.Errorf("persistence: HSet messages %s (%d msgs): %v", queueKey, len(fields), err)
|
||
}
|
||
})
|
||
}
|
||
|
||
// DeleteMessagePersist — удаляет одно сообщение из Redis асинхронно.
|
||
// Вызывать при DeleteMessage (после удаления из in-memory).
|
||
func DeleteMessagePersist(queueKey string, msgUuid string) {
|
||
if Client == nil || msgUuid == "" {
|
||
return
|
||
}
|
||
hashKey := redisMsgHashPrefix + queueKey
|
||
asyncWrite(func() {
|
||
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
|
||
defer cancel()
|
||
if err := Client.HDel(ctx, hashKey, msgUuid).Err(); err != nil {
|
||
log.Errorf("persistence: HDel message %s/%s: %v", queueKey, msgUuid, err)
|
||
}
|
||
})
|
||
}
|
||
|
||
// DeleteMessagesPersist — удаляет несколько сообщений из Redis.
|
||
// Вызывать при DeleteMessageBatch.
|
||
func DeleteMessagesPersist(queueKey string, uuids []string) {
|
||
if Client == nil || len(uuids) == 0 {
|
||
return
|
||
}
|
||
hashKey := redisMsgHashPrefix + queueKey
|
||
// Копируем uuids — вызывающий может переиспользовать slice
|
||
ids := make([]string, len(uuids))
|
||
copy(ids, uuids)
|
||
asyncWrite(func() {
|
||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||
defer cancel()
|
||
if err := Client.HDel(ctx, hashKey, ids...).Err(); err != nil {
|
||
log.Errorf("persistence: HDel messages %s (%d): %v", queueKey, len(ids), err)
|
||
}
|
||
})
|
||
}
|
||
|
||
// PurgeMessagesPersist — удаляет ВСЕ сообщения очереди из Redis (для PurgeQueue).
|
||
// Удаляет весь HASH ssq:msg:{queueKey}.
|
||
func PurgeMessagesPersist(queueKey string) {
|
||
if Client == nil {
|
||
return
|
||
}
|
||
hashKey := redisMsgHashPrefix + queueKey
|
||
asyncWrite(func() {
|
||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||
defer cancel()
|
||
if err := Client.Del(ctx, hashKey).Err(); err != nil {
|
||
log.Errorf("persistence: DEL messages hash %s: %v", queueKey, err)
|
||
}
|
||
})
|
||
}
|
||
|
||
// DeleteQueue — удаляет метаданные очереди И все её сообщения из Redis асинхронно.
|
||
func DeleteQueue(key string) {
|
||
if Client == nil {
|
||
return
|
||
}
|
||
writeSeq := nextQueueWriteSeq(key)
|
||
msgHashKey := redisMsgHashPrefix + key
|
||
asyncWrite(func() {
|
||
if !isLatestQueueWriteSeq(key, writeSeq) {
|
||
return
|
||
}
|
||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||
defer cancel()
|
||
// Удаляем метаданные из общего HASH
|
||
if err := Client.HDel(ctx, redisHashQueues, key).Err(); err != nil {
|
||
log.Errorf("persistence: HDel queue meta %q: %v", key, err)
|
||
}
|
||
// Удаляем весь HASH с сообщениями
|
||
if err := Client.Del(ctx, msgHashKey).Err(); err != nil {
|
||
log.Errorf("persistence: DEL messages hash %q: %v", key, err)
|
||
}
|
||
})
|
||
}
|
||
|
||
// LoadAllQueues — загружает все очереди из Redis в память при старте сервиса.
|
||
// Схема v2: метаданные из ssq:queues, сообщения из ssq:msg:{key}.
|
||
// Миграция v1→v2: если в ssq:queues лежит JSON со встроенными Messages,
|
||
// они извлекаются в отдельный HASH и метаданные пересохраняются без Messages.
|
||
func LoadAllQueues() (map[string]*models.Queue, error) {
|
||
if Client == nil {
|
||
return nil, nil
|
||
}
|
||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||
defer cancel()
|
||
|
||
// 1. Загружаем метаданные очередей
|
||
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))
|
||
migrateKeys := make([]string, 0) // ключи для миграции v1→v2
|
||
|
||
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)
|
||
}
|
||
|
||
// Миграция v1→v2: если в JSON есть Messages — это старый формат
|
||
if len(q.Messages) > 0 {
|
||
migrateKeys = append(migrateKeys, k)
|
||
log.Infof("persistence: миграция v1→v2 для %q (%d сообщений)", k, len(q.Messages))
|
||
}
|
||
|
||
queues[k] = &q
|
||
}
|
||
|
||
// 2. Миграция: выносим Messages из старого формата в отдельные HASH'ы
|
||
for _, k := range migrateKeys {
|
||
q := queues[k]
|
||
hashKey := redisMsgHashPrefix + k
|
||
// Сохраняем каждое сообщение отдельно
|
||
if len(q.Messages) > 0 {
|
||
args := make([]interface{}, 0, len(q.Messages)*2)
|
||
for i := range q.Messages {
|
||
data, err := json.Marshal(&q.Messages[i])
|
||
if err != nil {
|
||
log.Errorf("persistence: migrate marshal msg %s/%s: %v", k, q.Messages[i].Uuid, err)
|
||
continue
|
||
}
|
||
args = append(args, q.Messages[i].Uuid, string(data))
|
||
}
|
||
if len(args) > 0 {
|
||
if err := Client.HSet(ctx, hashKey, args...).Err(); err != nil {
|
||
log.Errorf("persistence: migrate HSet messages %q: %v", k, err)
|
||
}
|
||
}
|
||
}
|
||
// Пересохраняем метаданные без Messages
|
||
metaData, err := marshalQueueMeta(q)
|
||
if err == nil {
|
||
if err := Client.HSet(ctx, redisHashQueues, k, string(metaData)).Err(); err != nil {
|
||
log.Errorf("persistence: migrate HSet meta %q: %v", k, err)
|
||
}
|
||
}
|
||
log.Infof("persistence: миграция %q завершена — %d сообщений вынесены в %s", k, len(q.Messages), hashKey)
|
||
}
|
||
|
||
// 3. Загружаем сообщения из отдельных HASH'ов для каждой очереди
|
||
for k, q := range queues {
|
||
hashKey := redisMsgHashPrefix + k
|
||
msgRaw, err := Client.HGetAll(ctx, hashKey).Result()
|
||
if err != nil {
|
||
log.Errorf("persistence: HGetAll messages %q: %v", k, err)
|
||
continue
|
||
}
|
||
if len(msgRaw) > 0 {
|
||
msgs := make([]models.SqsMessage, 0, len(msgRaw))
|
||
for uuid, v := range msgRaw {
|
||
var m models.SqsMessage
|
||
if err := json.Unmarshal([]byte(v), &m); err != nil {
|
||
log.Errorf("persistence: unmarshal message %s/%s: %v", k, uuid, err)
|
||
continue
|
||
}
|
||
msgs = append(msgs, m)
|
||
}
|
||
q.Messages = msgs
|
||
} else if len(q.Messages) == 0 {
|
||
// Пустая очередь — Messages уже nil, оставляем
|
||
q.Messages = nil
|
||
}
|
||
}
|
||
|
||
log.Infof("persistence: загружено %d очередей из Redis (миграций: %d)", len(queues), len(migrateKeys))
|
||
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)
|
||
writeSeq := nextTenantWriteSeq(id)
|
||
asyncWrite(func() {
|
||
if !isLatestTenantWriteSeq(id, writeSeq) {
|
||
return
|
||
}
|
||
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
|
||
}
|
||
writeSeq := nextTenantWriteSeq(id)
|
||
asyncWrite(func() {
|
||
if !isLatestTenantWriteSeq(id, writeSeq) {
|
||
return
|
||
}
|
||
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)
|
||
}
|
||
})
|
||
}
|
||
|
||
func nextQueueWriteSeq(key string) uint64 {
|
||
seqMu.Lock()
|
||
defer seqMu.Unlock()
|
||
queueWriteSeq[key]++
|
||
return queueWriteSeq[key]
|
||
}
|
||
|
||
func isLatestQueueWriteSeq(key string, seq uint64) bool {
|
||
seqMu.Lock()
|
||
defer seqMu.Unlock()
|
||
return queueWriteSeq[key] == seq
|
||
}
|
||
|
||
func nextTenantWriteSeq(id string) uint64 {
|
||
seqMu.Lock()
|
||
defer seqMu.Unlock()
|
||
tenantWriteSeq[id]++
|
||
return tenantWriteSeq[id]
|
||
}
|
||
|
||
func isLatestTenantWriteSeq(id string, seq uint64) bool {
|
||
seqMu.Lock()
|
||
defer seqMu.Unlock()
|
||
return tenantWriteSeq[id] == seq
|
||
}
|
||
|
||
// 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
|
||
}
|