feat: persist messages separately and update ingress docs

2026-04-12 08:44:46 +03:00
parent d5d3519836
commit 4a83575491
11 changed files with 359 additions and 104 deletions
@@ -4,7 +4,13 @@
 // Redis — источник правды для восстановления после рестарта.
 // Все записи в Redis асинхронны (горутина) — не блокируют SQS-операции.
 // Сериализация (json.Marshal) происходит синхронно пока вызывающий держит мьютекс — консистентный снапшот.
-// Created: 2026-04-10
+//
+// Схема хранения 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

@@ -34,8 +40,10 @@ var (
 const (
 	// redisHashTenants — HASH: tenantID → JSON тенанта
 	redisHashTenants = "ssq:tenants"
-	// redisHashQueues — HASH: queueKey → JSON очереди (включая сообщения)
+	// 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.
@@ -74,22 +82,37 @@ func asyncWrite(fn func()) {
 	}()
 }

-// SaveQueue — сохраняет очередь (с сообщениями) в Redis асинхронно.
-// ВАЖНО: вызывать пока вызывающий держит SyncQueues.Lock() — тогда json.Marshal
+// 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 := json.Marshal(queue)
+	// Сериализуем метаданные синхронно под мьютексом вызывающего → консистентный снапшот
+	data, err := marshalQueueMeta(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))
+		log.Errorf("persistence: marshal queue meta %q: %v", key, err)
 		return
 	}
 	writeSeq := nextQueueWriteSeq(key)
@@ -100,42 +123,162 @@ func SaveQueue(key string, queue *models.Queue) {
 		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)
+			log.Errorf("persistence: HSet queue meta %q: %v", key, err)
 		}
 	})
 }

-// DeleteQueue — удаляет очередь из Redis асинхронно.
+// 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(), 3*time.Second)
+		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 %q: %v", key, err)
+			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 в память при старте сервиса.
-// Инициализирует nil-maps чтобы избежать panic при deduplication/FIFO операциях.
+// Схема 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(), 10*time.Second)
+	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 {
@@ -152,9 +295,73 @@ func LoadAllQueues() (map[string]*models.Queue, error) {
 		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
 	}
-	log.Infof("persistence: загружено %d очередей из Redis", len(queues))
+
+	// 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
 }