// 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
}