services/notifications/queue/queue.go

// Package queue provides queue management for the notification service.
package queue

import (
	"context"
	"encoding/json"
	"fmt"
	"strconv"
	"time"

	"github.com/redis/go-redis/v9"
	"go.uber.org/zap"

	notifications "github.com/AmaniQuery/amaniquery/services/notifications"
)

// QueueManager handles notification queue operations
type QueueManager struct {
	redis  *redis.Client
	logger *zap.Logger
}

// NewQueueManager creates a new queue manager
func NewQueueManager(redisClient *redis.Client, logger *zap.Logger) *QueueManager {
	return &QueueManager{
		redis:  redisClient,
		logger: logger,
	}
}

// Enqueue adds a notification to the appropriate queue based on priority
func (m *QueueManager) Enqueue(ctx context.Context, msg notifications.QueueMessage, channel notifications.NotificationChannel) error {
	queueName := m.getQueueName(msg.Request.Priority)

	msgJson, err := json.Marshal(msg)
	if err != nil {
		return fmt.Errorf("failed to marshal queue message: %w", err)
	}

	return m.redis.XAdd(ctx, &redis.XAddArgs{
		Stream: queueName,
		Values: map[string]interface{}{
			"data":    string(msgJson),
			"channel": string(channel),
		},
	}).Err()
}

// EnqueueScheduled adds a notification to the scheduled queue for later delivery
func (m *QueueManager) EnqueueScheduled(ctx context.Context, msg notifications.QueueMessage, channel notifications.NotificationChannel, deliverAt time.Time) error {
	payload := struct {
		Message notifications.QueueMessage `json:"message"`
		Channel string                     `json:"channel"`
	}{
		Message: msg,
		Channel: string(channel),
	}

	msgJson, err := json.Marshal(payload)
	if err != nil {
		return fmt.Errorf("failed to marshal scheduled message: %w", err)
	}

	score := float64(deliverAt.Unix())
	return m.redis.ZAdd(ctx, notifications.QueueScheduled, redis.Z{
		Score:  score,
		Member: string(msgJson),
	}).Err()
}

// ProcessScheduledQueue moves scheduled notifications to active queues when due
func (m *QueueManager) ProcessScheduledQueue(ctx context.Context) {
	ticker := time.NewTicker(1 * time.Minute)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			m.processScheduledBatch(ctx)
		}
	}
}

// processScheduledBatch processes a batch of scheduled notifications
func (m *QueueManager) processScheduledBatch(ctx context.Context) {
	now := time.Now().Unix()

	// Get all scheduled notifications that are ready
	msgs, err := m.redis.ZRangeByScore(ctx, notifications.QueueScheduled, &redis.ZRangeBy{
		Min:   "-inf",
		Max:   strconv.FormatInt(now, 10),
		Count: 100,
	}).Result()

	if err != nil {
		m.logger.Error("Failed to get scheduled notifications", zap.Error(err))
		return
	}

	if len(msgs) == 0 {
		return
	}

	m.logger.Info("Processing scheduled notifications", zap.Int("count", len(msgs)))

	for _, msgStr := range msgs {
		var payload struct {
			Message notifications.QueueMessage `json:"message"`
			Channel string                     `json:"channel"`
		}

		if err := json.Unmarshal([]byte(msgStr), &payload); err != nil {
			m.logger.Error("Failed to unmarshal scheduled message", zap.Error(err))
			m.redis.ZRem(ctx, notifications.QueueScheduled, msgStr)
			continue
		}

		// Move to appropriate priority queue
		queueName := m.getQueueName(payload.Message.Request.Priority)

		msgJson, _ := json.Marshal(payload.Message)
		err = m.redis.XAdd(ctx, &redis.XAddArgs{
			Stream: queueName,
			Values: map[string]interface{}{
				"data":    string(msgJson),
				"channel": payload.Channel,
			},
		}).Err()

		if err != nil {
			m.logger.Error("Failed to enqueue scheduled notification", zap.Error(err))
			continue
		}

		// Remove from scheduled queue
		m.redis.ZRem(ctx, notifications.QueueScheduled, msgStr)

		m.logger.Debug("Scheduled notification moved to active queue",
			zap.String("requestId", payload.Message.Request.ID),
			zap.String("queue", queueName))
	}
}

// GetQueueStats returns statistics for all queues
func (m *QueueManager) GetQueueStats(ctx context.Context) (map[string]QueueStats, error) {
	queues := []string{
		notifications.QueueHighPriority,
		notifications.QueueNormalPriority,
		notifications.QueueLowPriority,
	}

	stats := make(map[string]QueueStats)

	for _, queue := range queues {
		info, err := m.redis.XInfoStream(ctx, queue).Result()
		if err != nil {
			if err.Error() == "ERR no such key" {
				stats[queue] = QueueStats{Length: 0}
				continue
			}
			return nil, fmt.Errorf("failed to get info for queue %s: %w", queue, err)
		}

		stats[queue] = QueueStats{
			Length:         info.Length,
			FirstEntry:     extractTimestamp(info.FirstEntry),
			LastEntry:      extractTimestamp(info.LastEntry),
			ConsumerGroups: info.Groups,
		}
	}

	// Get scheduled queue stats
	scheduledCount, err := m.redis.ZCard(ctx, notifications.QueueScheduled).Result()
	if err == nil {
		stats[notifications.QueueScheduled] = QueueStats{
			Length: scheduledCount,
		}
	}

	return stats, nil
}

// QueueStats holds queue statistics
type QueueStats struct {
	Length         int64     `json:"length"`
	FirstEntry     time.Time `json:"firstEntry,omitempty"`
	LastEntry      time.Time `json:"lastEntry,omitempty"`
	ConsumerGroups int64     `json:"consumerGroups"`
}

// CleanupOldMessages removes processed messages older than the specified duration
func (m *QueueManager) CleanupOldMessages(ctx context.Context, maxAge time.Duration) (int64, error) {
	queues := []string{
		notifications.QueueHighPriority,
		notifications.QueueNormalPriority,
		notifications.QueueLowPriority,
	}

	var totalTrimmed int64
	threshold := time.Now().Add(-maxAge)
	thresholdID := fmt.Sprintf("%d-0", threshold.UnixMilli())

	for _, queue := range queues {
		trimmed, err := m.redis.XTrimMinID(ctx, queue, thresholdID).Result()
		if err != nil {
			m.logger.Error("Failed to trim queue", zap.String("queue", queue), zap.Error(err))
			continue
		}
		totalTrimmed += trimmed
	}

	return totalTrimmed, nil
}

// Requeue moves a failed message back to the queue with backoff
func (m *QueueManager) Requeue(ctx context.Context, msg notifications.QueueMessage, channel notifications.NotificationChannel) error {
	// Calculate backoff
	backoff := time.Duration(msg.BackoffMs*(1<<msg.Attempt)) * time.Millisecond
	deliverAt := time.Now().Add(backoff)

	// Update attempt count
	msg.Attempt++

	if msg.Attempt >= msg.MaxRetries {
		// Move to dead letter queue
		return m.moveToDeadLetter(ctx, msg, channel, "max_retries_exceeded")
	}

	m.logger.Info("Requeuing notification with backoff",
		zap.String("requestId", msg.Request.ID),
		zap.Int("attempt", msg.Attempt),
		zap.Duration("backoff", backoff))

	return m.EnqueueScheduled(ctx, msg, channel, deliverAt)
}

// moveToDeadLetter moves a message to the dead letter queue
func (m *QueueManager) moveToDeadLetter(ctx context.Context, msg notifications.QueueMessage, channel notifications.NotificationChannel, reason string) error {
	dlqKey := "notifications:dlq"

	payload := map[string]interface{}{
		"message":        msg,
		"channel":        string(channel),
		"reason":         reason,
		"failed_at":      time.Now().Format(time.RFC3339),
		"total_attempts": msg.Attempt,
	}

	msgJson, err := json.Marshal(payload)
	if err != nil {
		return fmt.Errorf("failed to marshal DLQ message: %w", err)
	}

	return m.redis.LPush(ctx, dlqKey, string(msgJson)).Err()
}

// GetDeadLetterMessages retrieves messages from the dead letter queue
func (m *QueueManager) GetDeadLetterMessages(ctx context.Context, start, stop int64) ([]map[string]interface{}, error) {
	dlqKey := "notifications:dlq"

	msgs, err := m.redis.LRange(ctx, dlqKey, start, stop).Result()
	if err != nil {
		return nil, fmt.Errorf("failed to get DLQ messages: %w", err)
	}

	var result []map[string]interface{}
	for _, msgStr := range msgs {
		var msg map[string]interface{}
		if err := json.Unmarshal([]byte(msgStr), &msg); err != nil {
			continue
		}
		result = append(result, msg)
	}

	return result, nil
}

// RetryDeadLetterMessage moves a message from DLQ back to the active queue
func (m *QueueManager) RetryDeadLetterMessage(ctx context.Context, index int64) error {
	dlqKey := "notifications:dlq"

	// Get the message
	msgs, err := m.redis.LRange(ctx, dlqKey, index, index).Result()
	if err != nil || len(msgs) == 0 {
		return fmt.Errorf("message not found at index %d", index)
	}

	var payload struct {
		Message notifications.QueueMessage `json:"message"`
		Channel string                     `json:"channel"`
	}

	if err := json.Unmarshal([]byte(msgs[0]), &payload); err != nil {
		return fmt.Errorf("failed to unmarshal DLQ message: %w", err)
	}

	// Reset attempt counter
	payload.Message.Attempt = 0
	payload.Message.EnqueuedAt = time.Now()

	// Enqueue to normal priority queue
	if err := m.Enqueue(ctx, payload.Message, notifications.NotificationChannel(payload.Channel)); err != nil {
		return fmt.Errorf("failed to re-enqueue message: %w", err)
	}

	// Remove from DLQ (set to empty and trim)
	m.redis.LSet(ctx, dlqKey, index, "DELETED")
	m.redis.LRem(ctx, dlqKey, 0, "DELETED")

	return nil
}

// getQueueName returns the queue name based on priority
func (m *QueueManager) getQueueName(priority notifications.NotificationPriority) string {
	switch priority {
	case notifications.PriorityHigh, notifications.PriorityUrgent:
		return notifications.QueueHighPriority
	case notifications.PriorityNormal:
		return notifications.QueueNormalPriority
	case notifications.PriorityLow:
		return notifications.QueueLowPriority
	default:
		return notifications.QueueNormalPriority
	}
}

// extractTimestamp extracts timestamp from a Redis stream entry
func extractTimestamp(entry redis.XMessage) time.Time {
	if entry.ID == "" {
		return time.Time{}
	}

	// ID format: timestamp-sequence
	var ts int64
	fmt.Sscanf(entry.ID, "%d-", &ts)
	if ts > 0 {
		return time.UnixMilli(ts)
	}
	return time.Time{}
}