package source

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

	"github.com/DATA-DOG/go-sqlmock"
	"github.com/go-logr/logr"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/observability/pkg/mysql/client"
	"github.com/tidbcloud/pkgv2/gormutil"
	"gorm.io/driver/mysql"
	"gorm.io/gorm"
	"gorm.io/gorm/logger"
)

func setupMockDB(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
}

type mockAdder struct {
	items []any
}

func (m *mockAdder) Add(item any) {
	m.items = append(m.items, item)
}

func TestConsumer_SetLogger(t *testing.T) {
	c := &Consumer{}
	log := logr.Discard()
	c.SetLogger(log)
	assert.NotNil(t, c.log)
}

func TestConsumer_Init(t *testing.T) {
	t.Run("fails with invalid config", func(t *testing.T) {
		c := &Consumer{
			ConnCfg: &gormutil.Config{
				Host:     "invalid-host",
				Port:     9999,
				User:     "test",
				Password: "test",
				Database: "test",
			},
		}
		c.SetLogger(logr.Discard())

		err := c.Init(context.Background())
		assert.Error(t, err)
	})
}

func TestConsumer_Start(t *testing.T) {
	t.Run("stops on context cancellation", func(t *testing.T) {
		sqlDB, _, gormDB := setupMockDB(t)
		defer sqlDB.Close()

		c := &Consumer{
			DB:              gormDB,
			WorkQueue:       &mockAdder{},
			PollingInterval: 10 * time.Millisecond,
			CleanupInterval: 20 * time.Millisecond,
			TTL:             1 * time.Minute,
		}
		c.SetLogger(logr.Discard())

		ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
		defer cancel()

		err := c.Start(ctx)
		assert.NoError(t, err)
	})

	t.Run("uses default intervals when not set", func(t *testing.T) {
		sqlDB, _, gormDB := setupMockDB(t)
		defer sqlDB.Close()

		c := &Consumer{
			DB:        gormDB,
			WorkQueue: &mockAdder{},
			TTL:       1 * time.Minute,
		}
		c.SetLogger(logr.Discard())

		ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
		defer cancel()

		err := c.Start(ctx)
		assert.NoError(t, err)
	})
}

func TestConsumer_fetchNewEvents(t *testing.T) {
	t.Run("fetches and merges events successfully", func(t *testing.T) {
		sqlDB, mock, gormDB := setupMockDB(t)
		defer sqlDB.Close()

		adder := &mockAdder{}
		c := &Consumer{
			DB:          gormDB,
			WorkQueue:   adder,
			TTL:         1 * time.Minute,
			lastEventID: 0,
		}
		c.SetLogger(logr.Discard())

		mock.ExpectBegin()
		rows := sqlmock.NewRows([]string{"event_id", "namespace", "name", "status"}).
			AddRow(1, "default", "cluster-1", client.EventStatusPending).
			AddRow(2, "default", "cluster-1", client.EventStatusPending).
			AddRow(3, "default", "cluster-2", client.EventStatusPending)
		mock.ExpectQuery("SELECT .* FROM `cluster_events`").
			WillReturnRows(rows)
		mock.ExpectCommit()

		err := c.fetchNewEvents(context.Background())

		assert.NoError(t, err)
		assert.Equal(t, uint64(3), c.lastEventID)
		assert.Len(t, adder.items, 2) // Merged to 2 unique clusters
		assert.NoError(t, mock.ExpectationsWereMet())
	})

	t.Run("handles no new events", func(t *testing.T) {
		sqlDB, mock, gormDB := setupMockDB(t)
		defer sqlDB.Close()

		adder := &mockAdder{}
		c := &Consumer{
			DB:          gormDB,
			WorkQueue:   adder,
			TTL:         1 * time.Minute,
			lastEventID: 10,
		}
		c.SetLogger(logr.Discard())

		mock.ExpectBegin()
		rows := sqlmock.NewRows([]string{"event_id", "namespace", "name", "status"})
		mock.ExpectQuery("SELECT .* FROM `cluster_events`").
			WillReturnRows(rows)
		mock.ExpectCommit()

		err := c.fetchNewEvents(context.Background())

		assert.NoError(t, err)
		assert.Equal(t, uint64(10), c.lastEventID)
		assert.Len(t, adder.items, 0)
		assert.NoError(t, mock.ExpectationsWereMet())
	})

	t.Run("handles query error", func(t *testing.T) {
		sqlDB, mock, gormDB := setupMockDB(t)
		defer sqlDB.Close()

		c := &Consumer{
			DB:          gormDB,
			WorkQueue:   &mockAdder{},
			TTL:         1 * time.Minute,
			lastEventID: 0,
		}
		c.SetLogger(logr.Discard())

		mock.ExpectBegin()
		mock.ExpectQuery("SELECT .* FROM `cluster_events`").
			WillReturnError(sql.ErrConnDone)
		mock.ExpectRollback()

		err := c.fetchNewEvents(context.Background())

		assert.Error(t, err)
		assert.NoError(t, mock.ExpectationsWereMet())
	})

	t.Run("handles transaction begin error", func(t *testing.T) {
		sqlDB, mock, gormDB := setupMockDB(t)
		defer sqlDB.Close()

		c := &Consumer{
			DB:        gormDB,
			WorkQueue: &mockAdder{},
			TTL:       1 * time.Minute,
		}
		c.SetLogger(logr.Discard())

		mock.ExpectBegin().WillReturnError(sql.ErrConnDone)

		err := c.fetchNewEvents(context.Background())

		assert.Error(t, err)
		assert.Contains(t, err.Error(), "failed to begin transaction")
		assert.NoError(t, mock.ExpectationsWereMet())
	})
}

func TestConsumer_cleanupExpiredEvents(t *testing.T) {
	t.Run("cleans up expired events successfully", func(t *testing.T) {
		sqlDB, mock, gormDB := setupMockDB(t)
		defer sqlDB.Close()

		c := &Consumer{
			DB:  gormDB,
			TTL: 1 * time.Minute,
		}
		c.SetLogger(logr.Discard())

		mock.ExpectBegin()
		mock.ExpectExec("DELETE FROM `cluster_events`").
			WillReturnResult(sqlmock.NewResult(0, 5))
		mock.ExpectCommit()

		err := c.cleanupExpiredEvents(context.Background())

		assert.NoError(t, err)
		assert.NoError(t, mock.ExpectationsWereMet())
	})

	t.Run("handles delete error", func(t *testing.T) {
		sqlDB, mock, gormDB := setupMockDB(t)
		defer sqlDB.Close()

		c := &Consumer{
			DB:  gormDB,
			TTL: 1 * time.Minute,
		}
		c.SetLogger(logr.Discard())

		mock.ExpectBegin()
		mock.ExpectExec("DELETE FROM `cluster_events`").
			WillReturnError(sql.ErrConnDone)
		mock.ExpectRollback()

		err := c.cleanupExpiredEvents(context.Background())

		assert.Error(t, err)
		assert.Contains(t, err.Error(), "failed to cleanup expired events")
		assert.NoError(t, mock.ExpectationsWereMet())
	})

	t.Run("handles transaction begin error", func(t *testing.T) {
		sqlDB, mock, gormDB := setupMockDB(t)
		defer sqlDB.Close()

		c := &Consumer{
			DB:  gormDB,
			TTL: 1 * time.Minute,
		}
		c.SetLogger(logr.Discard())

		mock.ExpectBegin().WillReturnError(sql.ErrConnDone)

		err := c.cleanupExpiredEvents(context.Background())

		assert.Error(t, err)
		assert.Contains(t, err.Error(), "failed to begin transaction")
		assert.NoError(t, mock.ExpectationsWereMet())
	})

	t.Run("handles commit error", func(t *testing.T) {
		sqlDB, mock, gormDB := setupMockDB(t)
		defer sqlDB.Close()

		c := &Consumer{
			DB:  gormDB,
			TTL: 1 * time.Minute,
		}
		c.SetLogger(logr.Discard())

		mock.ExpectBegin()
		mock.ExpectExec("DELETE FROM `cluster_events`").
			WillReturnResult(sqlmock.NewResult(0, 1))
		mock.ExpectCommit().WillReturnError(sql.ErrConnDone)

		err := c.cleanupExpiredEvents(context.Background())

		assert.Error(t, err)
		assert.NoError(t, mock.ExpectationsWereMet())
	})
}