SQS-service/app/persistence/redis.go

// 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"
	"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 очереди (включая сообщения)
	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
	}
	// Fix #4.3: Redis size guard — не сохранять если > 50MB (OOM protection)
	if len(data) > 50*1024*1024 {
		log.Warnf("persistence: queue %q too large for Redis (%d bytes), skipping", key, len(data))
		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 %q: %v", key, err)
		}
	})
}

// DeleteQueue — удаляет очередь из Redis асинхронно.
func DeleteQueue(key string) {
	if Client == nil {
		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.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)
	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
}