package queue_test

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"net"
	"os/exec"
	"strings"
	"testing"
	"time"

	"github.com/redis/go-redis/v9"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"

	"security-center-backend/internal/modules/caasm/queue"
)

func TestRedisRetryBackoffDoesNotHotLoop(t *testing.T) {
	ctx := context.Background()
	client := startRedisForQueueTest(t)
	jq := newTestJobQueue(t, client)

	job := &queue.SyncJobPayload{
		JobID:          "retry-job",
		CloudAccountID: 42,
		JobType:        queue.JobTypeCloudAccountSync,
		MaxRetries:     3,
	}
	require.NoError(t, jq.Enqueue(ctx, job))

	claimed, streamID, err := jq.Dequeue(ctx, "worker-1")
	require.NoError(t, err)
	require.NotNil(t, claimed)

	require.NoError(t, jq.Fail(ctx, claimed, streamID, errors.New("collector failed")))

	pending, err := client.XPending(ctx, testStreamName(t), queue.ConsumerGroup).Result()
	require.NoError(t, err)
	assert.EqualValues(t, 0, pending.Count)
	assert.EqualValues(t, 1, client.ZCard(ctx, jq.DelayedRetrySetForTest()).Val())

	streams, err := client.XReadGroup(ctx, &redis.XReadGroupArgs{
		Group:    queue.ConsumerGroup,
		Consumer: "worker-2",
		Streams:  []string{testStreamName(t), ">"},
		Count:    1,
		Block:    20 * time.Millisecond,
	}).Result()
	assert.ErrorIs(t, err, redis.Nil)
	assert.Empty(t, streams)

	members, err := client.ZRange(ctx, jq.DelayedRetrySetForTest(), 0, -1).Result()
	require.NoError(t, err)
	require.Len(t, members, 1)

	var delayed queue.SyncJobPayload
	require.NoError(t, json.Unmarshal([]byte(members[0]), &delayed))
	assert.Equal(t, 1, delayed.RetryCount)
	assert.Contains(t, delayed.LastError, "collector failed")

	require.NoError(t, client.ZAdd(ctx, jq.DelayedRetrySetForTest(), redis.Z{
		Score:  float64(time.Now().Add(-time.Second).UnixMilli()),
		Member: members[0],
	}).Err())
	require.NoError(t, jq.PromoteDelayedRetriesForTest(ctx))
	assert.EqualValues(t, 0, client.ZCard(ctx, jq.DelayedRetrySetForTest()).Val())

	promoted, _, err := jq.Dequeue(ctx, "worker-2")
	require.NoError(t, err)
	require.NotNil(t, promoted)
	assert.Equal(t, "retry-job", promoted.JobID)
	assert.Equal(t, 1, promoted.RetryCount)
}

func TestCleanupStuckJobsSchedulesRetryForRetryablePendingJob(t *testing.T) {
	ctx := context.Background()
	client := startRedisForQueueTest(t)
	jq := newTestJobQueue(t, client)
	jq.SetJobTimeoutForTest(10 * time.Millisecond)

	job := &queue.SyncJobPayload{
		JobID:          "stuck-retry-job",
		CloudAccountID: 43,
		JobType:        queue.JobTypeCloudAccountSync,
		MaxRetries:     2,
	}
	require.NoError(t, jq.Enqueue(ctx, job))

	claimed, _, err := jq.Dequeue(ctx, "stuck-worker")
	require.NoError(t, err)
	require.NotNil(t, claimed)

	time.Sleep(25 * time.Millisecond)
	require.NoError(t, jq.CleanupStuckJobs(ctx))

	pending, err := client.XPending(ctx, testStreamName(t), queue.ConsumerGroup).Result()
	require.NoError(t, err)
	assert.EqualValues(t, 0, pending.Count)
	assert.EqualValues(t, 1, client.ZCard(ctx, jq.DelayedRetrySetForTest()).Val())
	assert.EqualValues(t, 0, client.XLen(ctx, testStreamName(t)+":dlq").Val())

	members, err := client.ZRange(ctx, jq.DelayedRetrySetForTest(), 0, -1).Result()
	require.NoError(t, err)
	require.Len(t, members, 1)

	var delayed queue.SyncJobPayload
	require.NoError(t, json.Unmarshal([]byte(members[0]), &delayed))
	assert.Equal(t, "stuck-retry-job", delayed.JobID)
	assert.Equal(t, 1, delayed.RetryCount)
	assert.Contains(t, delayed.LastError, "job stuck pending")
}

func TestCleanupStuckJobsMovesExhaustedPendingJobToDLQ(t *testing.T) {
	ctx := context.Background()
	client := startRedisForQueueTest(t)
	jq := newTestJobQueue(t, client)
	jq.SetJobTimeoutForTest(10 * time.Millisecond)

	job := &queue.SyncJobPayload{
		JobID:          "stuck-dlq-job",
		CloudAccountID: 44,
		JobType:        queue.JobTypeCloudAccountSync,
		MaxRetries:     1,
	}
	require.NoError(t, jq.Enqueue(ctx, job))

	claimed, _, err := jq.Dequeue(ctx, "stuck-worker")
	require.NoError(t, err)
	require.NotNil(t, claimed)

	time.Sleep(25 * time.Millisecond)
	require.NoError(t, jq.CleanupStuckJobs(ctx))

	pending, err := client.XPending(ctx, testStreamName(t), queue.ConsumerGroup).Result()
	require.NoError(t, err)
	assert.EqualValues(t, 0, pending.Count)
	assert.EqualValues(t, 0, client.ZCard(ctx, jq.DelayedRetrySetForTest()).Val())
	assert.EqualValues(t, 1, client.XLen(ctx, testStreamName(t)+":dlq").Val())

	entries, err := client.XRange(ctx, testStreamName(t)+":dlq", "-", "+").Result()
	require.NoError(t, err)
	require.Len(t, entries, 1)
	payload, ok := entries[0].Values["payload"].(string)
	require.True(t, ok)

	var dlqJob queue.SyncJobPayload
	require.NoError(t, json.Unmarshal([]byte(payload), &dlqJob))
	assert.Equal(t, "stuck-dlq-job", dlqJob.JobID)
	assert.Equal(t, 1, dlqJob.RetryCount)
	assert.Contains(t, dlqJob.LastError, "job stuck pending")
}

func TestCleanupStuckJobsHonorsDBGuardBeforeReclaim(t *testing.T) {
	ctx := context.Background()
	client := startRedisForQueueTest(t)
	jq := newTestJobQueue(t, client)
	jq.SetJobTimeoutForTest(10 * time.Millisecond)

	syncJobID := uint64(501)
	guardCalled := false
	jq.SetStuckJobGuard(func(ctx context.Context, job *queue.SyncJobPayload, idle time.Duration) queue.StuckJobDecision {
		guardCalled = true
		assert.Equal(t, "guarded-stuck-job", job.JobID)
		assert.Equal(t, &syncJobID, job.SyncJobID)
		return queue.StuckJobDecision{
			SkipRetry: true,
			Reason:    "db_sync_job_still_active",
		}
	})

	job := &queue.SyncJobPayload{
		JobID:          "guarded-stuck-job",
		SyncJobID:      &syncJobID,
		CloudAccountID: 45,
		JobType:        queue.JobTypeCloudAccountSync,
		MaxRetries:     2,
	}
	require.NoError(t, jq.Enqueue(ctx, job))

	claimed, _, err := jq.Dequeue(ctx, "guarded-worker")
	require.NoError(t, err)
	require.NotNil(t, claimed)

	time.Sleep(25 * time.Millisecond)
	require.NoError(t, jq.CleanupStuckJobs(ctx))

	assert.True(t, guardCalled)
	pending, err := client.XPending(ctx, testStreamName(t), queue.ConsumerGroup).Result()
	require.NoError(t, err)
	assert.EqualValues(t, 1, pending.Count)
	assert.EqualValues(t, 0, client.ZCard(ctx, jq.DelayedRetrySetForTest()).Val())
	assert.EqualValues(t, 0, client.XLen(ctx, testStreamName(t)+":dlq").Val())
}

func startRedisForQueueTest(t *testing.T) *redis.Client {
	t.Helper()

	redisServer, err := exec.LookPath("redis-server")
	if err != nil {
		t.Skip("redis-server binary is required for Redis queue reliability tests")
	}

	listener, err := net.Listen("tcp", "127.0.0.1:0")
	require.NoError(t, err)
	addr := listener.Addr().String()
	require.NoError(t, listener.Close())
	_, port, err := net.SplitHostPort(addr)
	require.NoError(t, err)

	cmd := exec.Command(redisServer,
		"--bind", "127.0.0.1",
		"--port", port,
		"--save", "",
		"--appendonly", "no",
		"--dir", t.TempDir(),
	)
	require.NoError(t, cmd.Start())
	t.Cleanup(func() {
		_ = cmd.Process.Kill()
		_ = cmd.Wait()
	})

	client := redis.NewClient(&redis.Options{Addr: addr})
	t.Cleanup(func() {
		_ = client.Close()
	})

	deadline := time.Now().Add(2 * time.Second)
	for {
		if err := client.Ping(context.Background()).Err(); err == nil {
			return client
		}
		if time.Now().After(deadline) {
			t.Fatal("redis-server did not become ready")
		}
		time.Sleep(10 * time.Millisecond)
	}
}

func newTestJobQueue(t *testing.T, client *redis.Client) *queue.JobQueue {
	t.Helper()

	jq, err := queue.NewJobQueue(client, testStreamName(t), queue.ConsumerGroup)
	require.NoError(t, err)
	return jq
}

func testStreamName(t *testing.T) string {
	t.Helper()
	return fmt.Sprintf("caasm:test:%s", strings.NewReplacer("/", "-", " ", "-").Replace(t.Name()))
}