package queue

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

	"github.com/google/uuid"
	"github.com/redis/go-redis/v9"

	"security-center-backend/internal/config"
	"security-center-backend/internal/pkg/logger"
)

const (
	// Stream names
	StreamNameSyncJobs = "caasm:sync-jobs"
	StreamNameDLQ      = "caasm:sync-jobs:dlq"

	// Consumer group
	ConsumerGroup = "caasm-workers"

	// Default configuration
	DefaultMaxRetries    = 3
	DefaultJobTimeout    = 30 * time.Minute
	DefaultBlockDuration = 5 * time.Second

	delayedRetrySuffix = ":delayed-retries"
	cleanupConsumer    = "caasm-cleanup"
)

// JobQueueInterface defines the methods for the job queue
type JobQueueInterface interface {
	Enqueue(ctx context.Context, job *SyncJobPayload) error
	Dequeue(ctx context.Context, consumerName string) (*SyncJobPayload, string, error)
	Acknowledge(ctx context.Context, streamID string) error
	Fail(ctx context.Context, job *SyncJobPayload, streamID string, err error) error
	GetQueueDepth(ctx context.Context) (int64, error)
	GetDLQDepth(ctx context.Context) (int64, error)
	CleanupStuckJobs(ctx context.Context) error
	// ExistsActiveAccount checks queue + pending groups for any job for the account.
	ExistsActiveAccount(ctx context.Context, accountID uint64) bool
}

type AdmissionDecision struct {
	Allowed  bool
	Delay    time.Duration
	Reason   string
	Metadata map[string]interface{}
}

type AdmissionController func(ctx context.Context, job *SyncJobPayload, activeLocalJobs int) AdmissionDecision
type JobEventReporter func(ctx context.Context, syncJobID uint64, eventType string, message string, metadata map[string]interface{})

type StuckJobDecision struct {
	SkipRetry bool
	Reason    string
	Metadata  map[string]interface{}
}

type StuckJobGuard func(ctx context.Context, job *SyncJobPayload, idle time.Duration) StuckJobDecision

// JobQueue manages the Redis Streams-based job queue
type JobQueue struct {
	redis      *redis.Client
	logger     *logger.Logger
	streamName string
	dlqStream  string
	delayedSet string
	consGroup  string
	maxRetries int
	jobTimeout time.Duration
	stuckGuard StuckJobGuard
}

// NewJobQueue creates a new job queue instance
func NewJobQueue(redisClient *redis.Client, streamName, consumerGroup string) (*JobQueue, error) {
	if redisClient == nil {
		return nil, fmt.Errorf("redis client is required")
	}

	jq := &JobQueue{
		redis:      redisClient,
		logger:     logger.New(config.LogConfig{Level: "info", Format: "json"}), // Create internal logger
		streamName: streamName,
		dlqStream:  streamName + ":dlq",
		delayedSet: streamName + delayedRetrySuffix,
		consGroup:  consumerGroup,
		maxRetries: DefaultMaxRetries,
		jobTimeout: DefaultJobTimeout,
	}

	// Ensure consumer group exists
	if err := jq.ensureConsumerGroup(context.Background()); err != nil {
		return nil, fmt.Errorf("failed to create consumer group: %w", err)
	}

	return jq, nil
}

func (jq *JobQueue) SetStuckJobGuard(guard StuckJobGuard) {
	jq.stuckGuard = guard
}

// ensureConsumerGroup creates the consumer group if it doesn't exist
func (jq *JobQueue) ensureConsumerGroup(ctx context.Context) error {
	// Try to create the group; ignore error if it already exists
	err := jq.redis.XGroupCreateMkStream(ctx, jq.streamName, jq.consGroup, "0").Err()
	if err != nil && err.Error() != "BUSYGROUP Consumer Group name already exists" {
		return err
	}
	return nil
}

// Enqueue adds a new job to the queue
func (jq *JobQueue) Enqueue(ctx context.Context, job *SyncJobPayload) error {
	// Generate job ID if not provided
	if job.JobID == "" {
		job.JobID = uuid.New().String()
	}

	// Set defaults
	if job.EnqueuedAt.IsZero() {
		job.EnqueuedAt = time.Now()
	}
	if job.MaxRetries == 0 {
		job.MaxRetries = jq.maxRetries
	}

	// Serialize job payload
	jobData, err := json.Marshal(job)
	if err != nil {
		return fmt.Errorf("failed to marshal job: %w", err)
	}

	// Add to stream
	args := &redis.XAddArgs{
		Stream: jq.streamName,
		Values: map[string]interface{}{
			"job_id":   job.JobID,
			"payload":  string(jobData),
			"priority": int(job.Priority),
		},
	}

	id, err := jq.redis.XAdd(ctx, args).Result()
	if err != nil {
		return fmt.Errorf("failed to enqueue job: %w", err)
	}

	jq.logger.Info("job enqueued",
		"job_id", job.JobID,
		"stream_id", id,
		"account_id", job.CloudAccountID,
		"job_type", job.JobType,
	)

	return nil
}

// Dequeue retrieves the next job from the queue for processing
func (jq *JobQueue) Dequeue(ctx context.Context, consumerName string) (*SyncJobPayload, string, error) {
	if err := jq.promoteDelayedRetries(ctx); err != nil {
		return nil, "", fmt.Errorf("failed to promote delayed retries: %w", err)
	}

	// XREADGROUP with blocking
	streams, err := jq.redis.XReadGroup(ctx, &redis.XReadGroupArgs{
		Group:    jq.consGroup,
		Consumer: consumerName,
		Streams:  []string{jq.streamName, ">"},
		Count:    1,
		Block:    DefaultBlockDuration,
	}).Result()

	if err != nil {
		if err == redis.Nil {
			// No messages available (timeout)
			return nil, "", nil
		}
		return nil, "", fmt.Errorf("failed to read from stream: %w", err)
	}

	if len(streams) == 0 || len(streams[0].Messages) == 0 {
		return nil, "", nil
	}

	msg := streams[0].Messages[0]

	// Parse job payload
	payloadStr, ok := msg.Values["payload"].(string)
	if !ok {
		return nil, "", fmt.Errorf("invalid payload format")
	}

	var job SyncJobPayload
	if err := json.Unmarshal([]byte(payloadStr), &job); err != nil {
		return nil, "", fmt.Errorf("failed to unmarshal job: %w", err)
	}

	jq.logger.Debug("job dequeued",
		"job_id", job.JobID,
		"stream_id", msg.ID,
		"consumer", consumerName,
	)

	return &job, msg.ID, nil
}

// Acknowledge marks a job as successfully completed
func (jq *JobQueue) Acknowledge(ctx context.Context, streamID string) error {
	err := jq.redis.XAck(ctx, jq.streamName, jq.consGroup, streamID).Err()
	if err != nil {
		return fmt.Errorf("failed to acknowledge job: %w", err)
	}
	return nil
}

// Fail handles job failure - either retry or move to DLQ
func (jq *JobQueue) Fail(ctx context.Context, job *SyncJobPayload, streamID string, err error) error {
	job.IncrementRetry(err)

	if job.ShouldRetry() {
		// Requeue with backoff
		return jq.requeueWithBackoff(ctx, job, streamID)
	}

	// Move to DLQ
	return jq.moveToDLQ(ctx, job, streamID)
}

// requeueWithBackoff requeues a job after exponential backoff delay
func (jq *JobQueue) requeueWithBackoff(ctx context.Context, job *SyncJobPayload, streamID string) error {
	delay := job.NextRetryDelay()

	jq.logger.Warn("job failed, will retry",
		"job_id", job.JobID,
		"retry_count", job.RetryCount,
		"delay", delay,
		"error", job.LastError,
	)

	jobData, err := json.Marshal(job)
	if err != nil {
		return fmt.Errorf("failed to marshal delayed retry job: %w", err)
	}

	score := float64(time.Now().Add(delay).UnixMilli())
	if err := jq.redis.ZAdd(ctx, jq.delayedSet, redis.Z{
		Score:  score,
		Member: string(jobData),
	}).Err(); err != nil {
		return fmt.Errorf("failed to schedule delayed retry: %w", err)
	}

	// Acknowledge the original message
	return jq.Acknowledge(ctx, streamID)
}

func (jq *JobQueue) promoteDelayedRetries(ctx context.Context) error {
	now := float64(time.Now().UnixMilli())
	members, err := jq.redis.ZRangeByScore(ctx, jq.delayedSet, &redis.ZRangeBy{
		Min:   "-inf",
		Max:   fmt.Sprintf("%f", now),
		Count: 100,
	}).Result()
	if err != nil {
		return err
	}

	for _, member := range members {
		removed, err := jq.redis.ZRem(ctx, jq.delayedSet, member).Result()
		if err != nil {
			return fmt.Errorf("failed to claim delayed retry job: %w", err)
		}
		if removed == 0 {
			continue
		}

		var job SyncJobPayload
		if err := json.Unmarshal([]byte(member), &job); err != nil {
			jq.logger.Error("dropping invalid delayed retry payload", "error", err)
			continue
		}

		if err := jq.Enqueue(ctx, &job); err != nil {
			if restoreErr := jq.redis.ZAdd(ctx, jq.delayedSet, redis.Z{
				Score:  now,
				Member: member,
			}).Err(); restoreErr != nil {
				return fmt.Errorf("failed to restore delayed retry after enqueue error: %w", restoreErr)
			}
			return fmt.Errorf("failed to enqueue delayed retry job: %w", err)
		}

		jq.logger.Info("delayed retry promoted",
			"job_id", job.JobID,
			"retry_count", job.RetryCount,
		)
	}

	return nil
}

// moveToDLQ moves a job to the dead letter queue
func (jq *JobQueue) moveToDLQ(ctx context.Context, job *SyncJobPayload, streamID string) error {
	jq.logger.Error("job moved to DLQ after max retries",
		"job_id", job.JobID,
		"retry_count", job.RetryCount,
		"error", job.LastError,
	)

	// Serialize job
	jobData, err := json.Marshal(job)
	if err != nil {
		return fmt.Errorf("failed to marshal DLQ job: %w", err)
	}

	// Add to DLQ stream
	args := &redis.XAddArgs{
		Stream: jq.dlqStream,
		Values: map[string]interface{}{
			"job_id":      job.JobID,
			"payload":     string(jobData),
			"original_id": streamID,
			"failed_at":   time.Now().Unix(),
		},
	}

	if _, err := jq.redis.XAdd(ctx, args).Result(); err != nil {
		return fmt.Errorf("failed to add to DLQ: %w", err)
	}

	// Acknowledge the original message
	return jq.Acknowledge(ctx, streamID)
}

// GetQueueDepth returns the number of pending jobs in the queue
func (jq *JobQueue) GetQueueDepth(ctx context.Context) (int64, error) {
	length, err := jq.redis.XLen(ctx, jq.streamName).Result()
	if err != nil {
		return 0, fmt.Errorf("failed to get queue depth: %w", err)
	}
	return length, nil
}

// GetDLQDepth returns the number of jobs in the dead letter queue
func (jq *JobQueue) GetDLQDepth(ctx context.Context) (int64, error) {
	length, err := jq.redis.XLen(ctx, jq.dlqStream).Result()
	if err != nil {
		return 0, fmt.Errorf("failed to get DLQ depth: %w", err)
	}
	return length, nil
}

// ExistsActiveAccount scans recent entries to detect pending work for the account (best-effort).
func (jq *JobQueue) ExistsActiveAccount(ctx context.Context, accountID uint64) bool {
	if jq.redis == nil {
		return false
	}
	accountStr := fmt.Sprintf("%d", accountID)

	// Check recent ready entries
	entries, err := jq.redis.XRevRangeN(ctx, jq.streamName, "+", "-", 50).Result()
	if err == nil {
		for _, entry := range entries {
			if acctVal, ok := entry.Values["cloud_account_id"]; ok && fmt.Sprint(acctVal) == accountStr {
				return true
			}
			if acctVal, ok := entry.Values["CloudAccountID"]; ok && fmt.Sprint(acctVal) == accountStr {
				return true
			}
		}
	}

	// Check pending claims
	pending, err := jq.redis.XPendingExt(ctx, &redis.XPendingExtArgs{
		Stream: jq.streamName,
		Group:  jq.consGroup,
		Start:  "-",
		End:    "+",
		Count:  50,
	}).Result()
	if err == nil {
		for _, p := range pending {
			msgs, xrErr := jq.redis.XRange(ctx, jq.streamName, p.ID, p.ID).Result()
			if xrErr != nil {
				continue
			}
			for _, m := range msgs {
				if acctVal, ok := m.Values["cloud_account_id"]; ok && fmt.Sprint(acctVal) == accountStr {
					return true
				}
				if acctVal, ok := m.Values["CloudAccountID"]; ok && fmt.Sprint(acctVal) == accountStr {
					return true
				}
			}
		}
	}
	return false
}

// CleanupStuckJobs finds and handles jobs that have been processing for too long
func (jq *JobQueue) CleanupStuckJobs(ctx context.Context) error {
	// Use XPENDING to find jobs that have been claimed but not acknowledged
	pending, err := jq.redis.XPendingExt(ctx, &redis.XPendingExtArgs{
		Stream: jq.streamName,
		Group:  jq.consGroup,
		Start:  "-",
		End:    "+",
		Count:  100,
	}).Result()

	if err != nil {
		return fmt.Errorf("failed to get pending jobs: %w", err)
	}

	// p.Idle is a time.Duration, not time.Time
	for _, p := range pending {
		if p.Idle > jq.jobTimeout {
			jq.logger.Warn("found stuck job",
				"message_id", p.ID,
				"consumer", p.Consumer,
				"idle_duration", p.Idle,
			)

			msgs, err := jq.redis.XRange(ctx, jq.streamName, p.ID, p.ID).Result()
			if err != nil {
				return fmt.Errorf("failed to load pending job %s: %w", p.ID, err)
			}
			if len(msgs) == 0 {
				continue
			}
			preClaimJob, parseErr := jq.parseJobMessage(msgs[0])
			if parseErr == nil && jq.shouldSkipStuckJobRetry(ctx, preClaimJob, p.ID, p.Idle) {
				continue
			}

			claimed, err := jq.redis.XClaim(ctx, &redis.XClaimArgs{
				Stream:   jq.streamName,
				Group:    jq.consGroup,
				Consumer: cleanupConsumer,
				MinIdle:  jq.jobTimeout,
				Messages: []string{p.ID},
			}).Result()
			if err != nil {
				return fmt.Errorf("failed to claim stuck job %s: %w", p.ID, err)
			}
			if len(claimed) == 0 {
				continue
			}

			for _, msg := range claimed {
				job, err := jq.parseJobMessage(msg)
				if err != nil {
					if dlqErr := jq.moveInvalidMessageToDLQ(ctx, msg.ID, err); dlqErr != nil {
						return dlqErr
					}
					continue
				}

				if err := jq.Fail(ctx, job, msg.ID, fmt.Errorf("job stuck pending for %s", p.Idle)); err != nil {
					return fmt.Errorf("failed to handle stuck job %s: %w", p.ID, err)
				}
			}
		}
	}

	return nil
}

func (jq *JobQueue) shouldSkipStuckJobRetry(ctx context.Context, job *SyncJobPayload, streamID string, idle time.Duration) bool {
	if jq.stuckGuard == nil || job == nil {
		return false
	}
	decision := jq.stuckGuard(ctx, job, idle)
	if !decision.SkipRetry {
		return false
	}
	if decision.Metadata == nil {
		decision.Metadata = map[string]interface{}{}
	}
	jq.logger.Info("skipping stuck job retry due to guard",
		"job_id", job.JobID,
		"stream_id", streamID,
		"account_id", job.CloudAccountID,
		"sync_job_id", job.SyncJobID,
		"idle_duration", idle,
		"reason", decision.Reason,
		"metadata", decision.Metadata,
	)
	return true
}

func (jq *JobQueue) parseJobMessage(msg redis.XMessage) (*SyncJobPayload, error) {
	payloadStr, ok := msg.Values["payload"].(string)
	if !ok {
		return nil, fmt.Errorf("invalid payload format")
	}

	var job SyncJobPayload
	if err := json.Unmarshal([]byte(payloadStr), &job); err != nil {
		return nil, fmt.Errorf("failed to unmarshal job: %w", err)
	}
	if job.MaxRetries == 0 {
		job.MaxRetries = jq.maxRetries
	}
	return &job, nil
}

func (jq *JobQueue) moveInvalidMessageToDLQ(ctx context.Context, streamID string, cause error) error {
	if _, err := jq.redis.XAdd(ctx, &redis.XAddArgs{
		Stream: jq.dlqStream,
		Values: map[string]interface{}{
			"original_id": streamID,
			"failed_at":   time.Now().Unix(),
			"error":       cause.Error(),
		},
	}).Result(); err != nil {
		return fmt.Errorf("failed to add invalid stuck job to DLQ: %w", err)
	}

	return jq.Acknowledge(ctx, streamID)
}