package source

import (
	"context"
	"database/sql"
	"testing"
	"time"

	"github.com/DATA-DOG/go-sqlmock"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	"gorm.io/driver/mysql"
	"gorm.io/gorm"
	"gorm.io/gorm/logger"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/client-go/util/workqueue"
	"sigs.k8s.io/controller-runtime/pkg/reconcile"
)

func setupMockDBForWorkqueue(t *testing.T) (*sql.DB, sqlmock.Sqlmock, *gorm.DB) {
	sqlDB, mock, err := sqlmock.New()
	require.NoError(t, err)

	gormDB, err := gorm.Open(mysql.New(mysql.Config{
		Conn:                      sqlDB,
		SkipInitializeWithVersion: true,
	}), &gorm.Config{
		Logger: logger.Default.LogMode(logger.Silent),
	})
	require.NoError(t, err)

	return sqlDB, mock, gormDB
}

func TestNewMysqlRateLimitingQueueWithConfig(t *testing.T) {
	t.Run("creates queue with provided config", func(t *testing.T) {
		rateLimiter := workqueue.NewTypedItemExponentialFailureRateLimiter[any](1*time.Second, 10*time.Second)
		config := workqueue.TypedRateLimitingQueueConfig[any]{
			Name: "test-queue",
		}

		queue := NewMysqlRateLimitingQueueWithConfig(rateLimiter, config, 5*time.Minute)

		assert.NotNil(t, queue)
	})

	t.Run("creates queue with custom delaying queue", func(t *testing.T) {
		rateLimiter := workqueue.NewTypedItemExponentialFailureRateLimiter[any](1*time.Second, 10*time.Second)
		delayingQueue := workqueue.NewTypedDelayingQueue[any]()
		config := workqueue.TypedRateLimitingQueueConfig[any]{
			Name:          "test-queue",
			DelayingQueue: delayingQueue,
		}

		queue := NewMysqlRateLimitingQueueWithConfig(rateLimiter, config, 5*time.Minute)

		assert.NotNil(t, queue)
		delayingQueue.ShutDown()
	})
}

func TestMysqlRateLimitingType_Add(t *testing.T) {
	t.Run("adds item to queue", func(t *testing.T) {
		rateLimiter := workqueue.NewTypedItemExponentialFailureRateLimiter[any](1*time.Millisecond, 10*time.Millisecond)
		config := workqueue.TypedRateLimitingQueueConfig[any]{
			Name: "test-queue",
		}
		queue := NewMysqlRateLimitingQueueWithConfig(rateLimiter, config, 5*time.Minute)
		defer queue.ShutDown()

		item := MySQLItem{
			Ctx:       context.Background(),
			EventID:   1,
			Namespace: "default",
			Name:      "cluster-1",
		}

		queue.Add(item)

		assert.Equal(t, 1, queue.Len())
	})

	t.Run("handles duplicate items with fixed queue", func(t *testing.T) {
		rateLimiter := workqueue.NewTypedItemExponentialFailureRateLimiter[any](1*time.Millisecond, 10*time.Millisecond)
		config := workqueue.TypedRateLimitingQueueConfig[any]{
			Name: "test-queue",
		}
		queue := NewMysqlRateLimitingQueueWithConfig(rateLimiter, config, 5*time.Minute)
		defer queue.ShutDown()

		item1 := MySQLItem{
			Ctx:       context.Background(),
			EventID:   1,
			Namespace: "default",
			Name:      "cluster-1",
		}
		item2 := MySQLItem{
			Ctx:       context.Background(),
			EventID:   2,
			Namespace: "default",
			Name:      "cluster-1",
		}

		queue.Add(item1)
		queue.Add(item2)

		// Should still be 1 because same namespace/name
		assert.Equal(t, 1, queue.Len())
	})

	t.Run("adds multiple different items", func(t *testing.T) {
		rateLimiter := workqueue.NewTypedItemExponentialFailureRateLimiter[any](1*time.Millisecond, 10*time.Millisecond)
		config := workqueue.TypedRateLimitingQueueConfig[any]{
			Name: "test-queue",
		}
		queue := NewMysqlRateLimitingQueueWithConfig(rateLimiter, config, 5*time.Minute)
		defer queue.ShutDown()

		item1 := MySQLItem{Namespace: "default", Name: "cluster-1"}
		item2 := MySQLItem{Namespace: "default", Name: "cluster-2"}

		queue.Add(item1)
		queue.Add(item2)

		assert.Equal(t, 2, queue.Len())
	})
}

func TestMysqlRateLimitingType_AddAfter(t *testing.T) {
	t.Run("adds item with delay", func(t *testing.T) {
		rateLimiter := workqueue.NewTypedItemExponentialFailureRateLimiter[any](1*time.Millisecond, 10*time.Millisecond)
		config := workqueue.TypedRateLimitingQueueConfig[any]{
			Name: "test-queue",
		}
		queue := NewMysqlRateLimitingQueueWithConfig(rateLimiter, config, 5*time.Minute)
		defer queue.ShutDown()

		req := reconcile.Request{
			NamespacedName: types.NamespacedName{
				Namespace: "default",
				Name:      "cluster-1",
			},
		}

		queue.AddAfter(req, 10*time.Millisecond)

		// Item should not be immediately available
		assert.Equal(t, 0, queue.Len())
	})

	t.Run("marks item as not forgot when adding after", func(t *testing.T) {
		rateLimiter := workqueue.NewTypedItemExponentialFailureRateLimiter[any](1*time.Millisecond, 10*time.Millisecond)
		config := workqueue.TypedRateLimitingQueueConfig[any]{
			Name: "test-queue",
		}
		q := NewMysqlRateLimitingQueueWithConfig(rateLimiter, config, 5*time.Minute).(*mysqlRateLimitingType)
		defer q.ShutDown()

		item := MySQLItem{Namespace: "default", Name: "cluster-1"}
		req := reconcile.Request{
			NamespacedName: types.NamespacedName{
				Namespace: "default",
				Name:      "cluster-1",
			},
		}

		// Add item first
		q.Add(item)

		// Mark as forgot
		q.Forget(req)

		// AddAfter should mark as not forgot
		q.AddAfter(req, 10*time.Millisecond)

		// Verify the forgot flag is reset
		q.mu.Lock()
		fixedQueue := q.processing[req]
		q.mu.Unlock()

		assert.NotNil(t, fixedQueue)
		front := fixedQueue.Peek()
		if front != nil {
			assert.False(t, front.forgot)
		}
	})
}

func TestMysqlRateLimitingType_AddRateLimited(t *testing.T) {
	t.Run("adds item with rate limiting", func(t *testing.T) {
		rateLimiter := workqueue.NewTypedItemExponentialFailureRateLimiter[any](1*time.Millisecond, 10*time.Millisecond)
		config := workqueue.TypedRateLimitingQueueConfig[any]{
			Name: "test-queue",
		}
		queue := NewMysqlRateLimitingQueueWithConfig(rateLimiter, config, 5*time.Minute)
		defer queue.ShutDown()

		req := reconcile.Request{
			NamespacedName: types.NamespacedName{
				Namespace: "default",
				Name:      "cluster-1",
			},
		}

		queue.AddRateLimited(req)

		// Should be delayed based on rate limiter
		assert.Equal(t, 0, queue.Len())
	})
}

func TestMysqlRateLimitingType_NumRequeues(t *testing.T) {
	t.Run("returns number of requeues", func(t *testing.T) {
		rateLimiter := workqueue.NewTypedItemExponentialFailureRateLimiter[any](1*time.Millisecond, 10*time.Millisecond)
		config := workqueue.TypedRateLimitingQueueConfig[any]{
			Name: "test-queue",
		}
		queue := NewMysqlRateLimitingQueueWithConfig(rateLimiter, config, 5*time.Minute)
		defer queue.ShutDown()

		req := reconcile.Request{
			NamespacedName: types.NamespacedName{
				Namespace: "default",
				Name:      "cluster-1",
			},
		}

		count := queue.NumRequeues(req)
		assert.Equal(t, 0, count)

		queue.AddRateLimited(req)
		count = queue.NumRequeues(req)
		assert.Equal(t, 1, count)
	})
}

func TestMysqlRateLimitingType_Forget(t *testing.T) {
	t.Run("forgets item and marks flag", func(t *testing.T) {
		rateLimiter := workqueue.NewTypedItemExponentialFailureRateLimiter[any](1*time.Millisecond, 10*time.Millisecond)
		config := workqueue.TypedRateLimitingQueueConfig[any]{
			Name: "test-queue",
		}
		q := NewMysqlRateLimitingQueueWithConfig(rateLimiter, config, 5*time.Minute).(*mysqlRateLimitingType)
		defer q.ShutDown()

		item := MySQLItem{Namespace: "default", Name: "cluster-1"}
		req := reconcile.Request{
			NamespacedName: types.NamespacedName{
				Namespace: "default",
				Name:      "cluster-1",
			},
		}

		q.Add(item)
		q.Forget(req)

		// Verify forgot flag is set
		q.mu.Lock()
		fixedQueue := q.processing[req]
		q.mu.Unlock()

		assert.NotNil(t, fixedQueue)
		front := fixedQueue.Peek()
		assert.NotNil(t, front)
		assert.True(t, front.forgot)
	})

	t.Run("handles forget on non-existent item", func(t *testing.T) {
		rateLimiter := workqueue.NewTypedItemExponentialFailureRateLimiter[any](1*time.Millisecond, 10*time.Millisecond)
		config := workqueue.TypedRateLimitingQueueConfig[any]{
			Name: "test-queue",
		}
		queue := NewMysqlRateLimitingQueueWithConfig(rateLimiter, config, 5*time.Minute)
		defer queue.ShutDown()

		req := reconcile.Request{
			NamespacedName: types.NamespacedName{
				Namespace: "default",
				Name:      "nonexistent",
			},
		}

		// Should not panic
		queue.Forget(req)
	})
}

func TestMysqlRateLimitingType_Done(t *testing.T) {
	t.Run("marks item as done and removes from processing", func(t *testing.T) {
		sqlDB, _, gormDB := setupMockDBForWorkqueue(t)
		defer sqlDB.Close()

		rateLimiter := workqueue.NewTypedItemExponentialFailureRateLimiter[any](1*time.Millisecond, 10*time.Millisecond)
		config := workqueue.TypedRateLimitingQueueConfig[any]{
			Name: "test-queue",
		}
		q := NewMysqlRateLimitingQueueWithConfig(rateLimiter, config, 5*time.Minute).(*mysqlRateLimitingType)
		defer q.ShutDown()

		ctx := context.Background()
		item := MySQLItem{
			Ctx:       ctx,
			DB:        gormDB,
			EventID:   1,
			Namespace: "default",
			Name:      "cluster-1",
			TTL:       5 * time.Minute,
		}
		req := reconcile.Request{
			NamespacedName: types.NamespacedName{
				Namespace: "default",
				Name:      "cluster-1",
			},
		}

		q.Add(item)

		// Get the item
		gotItem, shutdown := q.Get()
		assert.False(t, shutdown)
		assert.NotNil(t, gotItem)

		// Mark as done - no deleteMessage call expected when queue becomes empty and not forgotten
		q.Done(gotItem)

		// Processing queue should be empty
		q.mu.Lock()
		fixedQueue := q.processing[req]
		q.mu.Unlock()

		assert.Nil(t, fixedQueue)
	})

	t.Run("keeps processing queue when more items exist", func(t *testing.T) {
		sqlDB, mock, gormDB := setupMockDBForWorkqueue(t)
		defer sqlDB.Close()

		rateLimiter := workqueue.NewTypedItemExponentialFailureRateLimiter[any](1*time.Millisecond, 10*time.Millisecond)
		config := workqueue.TypedRateLimitingQueueConfig[any]{
			Name: "test-queue",
		}
		q := NewMysqlRateLimitingQueueWithConfig(rateLimiter, config, 5*time.Minute).(*mysqlRateLimitingType)
		defer q.ShutDown()

		ctx := context.Background()
		item1 := MySQLItem{Ctx: ctx, DB: gormDB, EventID: 1, Namespace: "default", Name: "cluster-1", TTL: 5 * time.Minute}
		item2 := MySQLItem{Ctx: ctx, DB: gormDB, EventID: 2, Namespace: "default", Name: "cluster-1", TTL: 5 * time.Minute}
		req := reconcile.Request{
			NamespacedName: types.NamespacedName{
				Namespace: "default",
				Name:      "cluster-1",
			},
		}

		q.Add(item1)
		q.Add(item2)

		gotItem, _ := q.Get()

		// Mock the deleteMessage DB call
		mock.ExpectBegin()
		mock.ExpectExec("UPDATE `cluster_events`").
			WillReturnResult(sqlmock.NewResult(0, 1))
		mock.ExpectCommit()

		q.Done(gotItem)

		// Processing queue should still exist
		q.mu.Lock()
		fixedQueue := q.processing[req]
		q.mu.Unlock()

		assert.NotNil(t, fixedQueue)
		assert.Equal(t, 1, fixedQueue.Len())
		assert.NoError(t, mock.ExpectationsWereMet())
	})

	t.Run("handles done on forgotten item", func(t *testing.T) {
		sqlDB, mock, gormDB := setupMockDBForWorkqueue(t)
		defer sqlDB.Close()

		rateLimiter := workqueue.NewTypedItemExponentialFailureRateLimiter[any](1*time.Millisecond, 10*time.Millisecond)
		config := workqueue.TypedRateLimitingQueueConfig[any]{
			Name: "test-queue",
		}
		q := NewMysqlRateLimitingQueueWithConfig(rateLimiter, config, 5*time.Minute).(*mysqlRateLimitingType)
		defer q.ShutDown()

		ctx := context.Background()
		item := MySQLItem{Ctx: ctx, DB: gormDB, EventID: 1, Namespace: "default", Name: "cluster-1", TTL: 5 * time.Minute}
		req := reconcile.Request{
			NamespacedName: types.NamespacedName{
				Namespace: "default",
				Name:      "cluster-1",
			},
		}

		q.Add(item)
		gotItem, _ := q.Get()

		q.Forget(req)

		// Mock the deleteMessage DB call
		mock.ExpectBegin()
		mock.ExpectExec("UPDATE `cluster_events`").
			WillReturnResult(sqlmock.NewResult(0, 1))
		mock.ExpectCommit()

		q.Done(gotItem)

		// Processing queue should be removed
		q.mu.Lock()
		fixedQueue := q.processing[req]
		q.mu.Unlock()

		assert.Nil(t, fixedQueue)
		assert.NoError(t, mock.ExpectationsWereMet())
	})
}

func TestMysqlRateLimitingType_deleteMessage(t *testing.T) {
	t.Run("handles nil item", func(t *testing.T) {
		rateLimiter := workqueue.NewTypedItemExponentialFailureRateLimiter[any](1*time.Millisecond, 10*time.Millisecond)
		config := workqueue.TypedRateLimitingQueueConfig[any]{
			Name: "test-queue",
		}
		q := NewMysqlRateLimitingQueueWithConfig(rateLimiter, config, 5*time.Minute).(*mysqlRateLimitingType)
		defer q.ShutDown()

		// Should not panic
		q.deleteMessage(nil)
	})
}