package taskqueue

import (
	"container/list"
	"context"
	"fmt"
	"sync"
	"testing"
	"time"

	"github.com/stretchr/testify/require"
	"go.uber.org/zap"
)

type memoryPersistClient struct {
	mu   sync.Mutex
	data map[string]string
}

func (m *memoryPersistClient) Init() error {
	return nil
}

func (m *memoryPersistClient) Get(key string) (string, error) {
	m.mu.Lock()
	defer m.mu.Unlock()

	value, ok := m.data[key]
	if !ok {
		return "", fmt.Errorf("key %q not found", key)
	}
	return value, nil
}

func (m *memoryPersistClient) Set(key string, value string) error {
	m.mu.Lock()
	defer m.mu.Unlock()

	m.data[key] = value
	return nil
}

func TestAckTask(t *testing.T) {
	tqe := createRunningTestQueue(t, &Config{Interval: "15s"})
	consumerQueue := tqe.Subscribe(context.Background())

	var prevTs time.Time
	for i := 0; i < 10; i++ {
		task := receiveTask(t, consumerQueue)
		task.AckTimestamp = task.EndTimestamp
		require.NoError(t, tqe.Ack(task))

		if i > 0 {
			require.Equal(t, prevTs.Add(tqe.interval), task.Timestamp)
		}

		prevTs = task.Timestamp
	}
}

func TestNackAndRequeueTask(t *testing.T) {
	tqe := createRunningTestQueue(t, &Config{Interval: "15s"})
	consumerQueue := tqe.Subscribe(context.Background())

	var prevTs time.Time
	for i := 0; i < 10; i++ {
		task := receiveTask(t, consumerQueue)
		require.NoError(t, tqe.Nack(task, true))

		if i > 0 {
			require.Equal(t, prevTs, task.Timestamp)
		}

		prevTs = task.Timestamp
	}
}

func TestDuplicateAck(t *testing.T) {
	tqe := createRunningTestQueue(t, &Config{Interval: "15s"})
	consumerQueue := tqe.Subscribe(context.Background())

	task := receiveTask(t, consumerQueue)
	tqe.logger.Info("task received", zap.Time("task_timestamp", task.Timestamp), zap.Time("task_createAt", task.createAt))
	task.AckTimestamp = task.EndTimestamp
	require.NoError(t, tqe.Ack(task))

	task1 := receiveTask(t, consumerQueue)
	tqe.logger.Info("task received", zap.Time("task_timestamp", task1.Timestamp), zap.Time("task_createAt", task1.createAt))

	task.AckTimestamp = task.EndTimestamp
	task1.AckTimestamp = task1.EndTimestamp
	require.NoError(t, tqe.Ack(task))
	require.NoError(t, tqe.Ack(task1))

	task2 := receiveTask(t, consumerQueue)
	require.Equal(t, task1.Timestamp.Add(tqe.interval), task2.Timestamp)
}

func TestNackWithRequeue_OverridesRetryStrategyAndKeepsWindow(t *testing.T) {
	tqe := createRunningTestQueue(t, &Config{Interval: "15s"})
	consumerQueue := tqe.Subscribe(context.Background())

	task := receiveTask(t, consumerQueue)
	task.SetRetryStrategy(RETRYSTRATEGY_REGENERATE)

	require.NoError(t, NackWithRequeue(tqe, task))

	next := receiveTask(t, consumerQueue)
	require.Equal(t, task.Timestamp, next.Timestamp)
	require.Equal(t, task.EndTimestamp, next.EndTimestamp)
	require.Equal(t, RETRYSTRATEGY_REQUEUE, next.RetryStrategy())
	require.Equal(t, task.GetRetriedCount()+1, next.GetRetriedCount())
}

func TestNackWithRequeue_OldWindowStillRetriesBeforeBypassDuration(t *testing.T) {
	tqe := createRunningTestQueue(t, &Config{Interval: "15s"})
	consumerQueue := tqe.Subscribe(context.Background())

	task := receiveTask(t, consumerQueue)
	task.Timestamp = time.Now().Add(-6 * time.Minute).Truncate(tqe.interval)
	task.EndTimestamp = task.Timestamp

	require.NoError(t, NackWithRequeue(tqe, task))

	next := receiveTask(t, consumerQueue)
	require.Equal(t, task.Timestamp, next.Timestamp)
	require.Equal(t, task.EndTimestamp, next.EndTimestamp)
	require.Equal(t, RETRYSTRATEGY_REQUEUE, next.RetryStrategy())
	require.False(t, next.FirstFailedAt().IsZero())
	require.Equal(t, task.GetRetriedCount()+1, next.GetRetriedCount())
}

func TestNackWithRequeue_BypassAfterRetryDuration(t *testing.T) {
	tqe := createRunningTestQueue(t, &Config{Interval: "15s"})
	consumerQueue := tqe.Subscribe(context.Background())

	task := receiveTask(t, consumerQueue)
	task.firstFailedAt = time.Now().Add(-6 * time.Minute)

	require.NoError(t, NackWithRequeue(tqe, task))

	next := receiveTask(t, consumerQueue)
	require.True(t, next.Timestamp.After(task.Timestamp), "should move to next window after bypass")
}

func receiveTask(t *testing.T, consumerQueue <-chan Task) Task {
	t.Helper()

	select {
	case task := <-consumerQueue:
		return task
	case <-time.After(2 * time.Second):
		t.Fatal("timed out waiting for task")
		return Task{}
	}
}

func createRunningTestQueue(t *testing.T, cfg *Config) *taskQueueExtension {
	t.Helper()

	tqe, err := createtestQueue(cfg)
	require.NoError(t, err)

	tqe.wg.Add(1)
	go tqe.runDispatchConsumer(context.Background())

	t.Cleanup(func() {
		close(tqe.stopChan)
		close(tqe.consumer.stopChan)
	})

	return tqe
}

func createtestQueue(cfg *Config) (*taskQueueExtension, error) {
	interval, err := time.ParseDuration(cfg.Interval)
	if err != nil {
		return nil, fmt.Errorf("parse interval failed: %w", err)
	}
	logger, _ := zap.NewDevelopment()

	rdr := &taskQueueExtension{
		logger: logger,

		config:   cfg,
		stopChan: make(chan struct{}),

		consumer: &consumerGroup{
			queue:    make(chan Task),
			ackChan:  make(chan Task),
			nackChan: make(chan nackRequest),
			stopChan: make(chan struct{}),
		},

		queue: list.New(),

		interval: interval,
		timeout:  time.Second,
		ackedTs:  time.Now().Truncate(interval),
		persistCli: &memoryPersistClient{
			data: make(map[string]string),
		},
	}

	rdr.queueCond = sync.NewCond(&rdr.queueMutex)

	if err := rdr.GenerateInitialTask(); err != nil {
		return nil, fmt.Errorf("generate initial task failed:%w", err)
	}

	return rdr, nil
}