package lease

import (
	"context"
	"fmt"
	"os"
	"strconv"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	gomysql "github.com/go-sql-driver/mysql"
	"github.com/stretchr/testify/require"
	"github.com/testcontainers/testcontainers-go"
	mysqlcontainer "github.com/testcontainers/testcontainers-go/modules/mysql"
	pkgv2db "github.com/tidbcloud/pkgv2/db"
	gormmysql "gorm.io/driver/mysql"
	"gorm.io/gorm"
)

const (
	leaseTestDBName   = "lease_test"
	leaseTestUser     = "root"
	leaseTestPassword = "password"
)

var leaseTestDB *gorm.DB
var leaseTestProvider pkgv2db.Provider

func TestMain(m *testing.M) {
	ctx := context.Background()
	mysqlContainer, host, port, err := startLeaseMysqlContainer(ctx)
	if err != nil {
		panic(err)
	}
	defer func() {
		if terminateErr := mysqlContainer.Terminate(ctx); terminateErr != nil {
			panic(terminateErr)
		}
	}()

	cfg := gomysql.NewConfig()
	cfg.User = leaseTestUser
	cfg.Passwd = leaseTestPassword
	cfg.Net = "tcp"
	cfg.Addr = fmt.Sprintf("%s:%d", host, port)
	cfg.DBName = leaseTestDBName
	cfg.ParseTime = true

	gdb, err := gorm.Open(gormmysql.Open(cfg.FormatDSN()), &gorm.Config{})
	if err != nil {
		panic(err)
	}
	leaseTestDB = gdb
	leaseTestProvider = pkgv2db.NewProvider(gdb)
	if err := leaseTestDB.AutoMigrate(&Lease{}); err != nil {
		panic(err)
	}

	m.Run()
}

func TestAcquireRenewRelease(t *testing.T) {
	ctx := context.Background()
	locker := newTestLocker(t)

	key := fmt.Sprintf("acquire-renew-release-%d", time.Now().UnixNano())
	ownerA := "owner-a"
	ownerB := "owner-b"

	acquired, err := locker.Acquire(ctx, key, ownerA, 2*time.Second)
	require.NoError(t, err)
	require.True(t, acquired)

	acquired, err = locker.Acquire(ctx, key, ownerB, 2*time.Second)
	require.NoError(t, err)
	require.False(t, acquired)

	renewed, err := locker.Renew(ctx, key, ownerA, 2*time.Second)
	require.NoError(t, err)
	require.True(t, renewed)

	renewed, err = locker.Renew(ctx, key, ownerB, 2*time.Second)
	require.NoError(t, err)
	require.False(t, renewed)

	released, err := locker.Release(ctx, key, ownerB)
	require.NoError(t, err)
	require.False(t, released)

	released, err = locker.Release(ctx, key, ownerA)
	require.NoError(t, err)
	require.True(t, released)

	acquired, err = locker.Acquire(ctx, key, ownerB, 2*time.Second)
	require.NoError(t, err)
	require.True(t, acquired)
}

func TestLeaseExpiryAllowsTakeover(t *testing.T) {
	ctx := context.Background()
	locker := newTestLocker(t)

	key := fmt.Sprintf("lease-expiry-%d", time.Now().UnixNano())

	acquired, err := locker.Acquire(ctx, key, "owner-a", 200*time.Millisecond)
	require.NoError(t, err)
	require.True(t, acquired)

	time.Sleep(350 * time.Millisecond)

	acquired, err = locker.Acquire(ctx, key, "owner-b", 2*time.Second)
	require.NoError(t, err)
	require.True(t, acquired)
}

func TestConcurrentAcquireOnlyOneWinner(t *testing.T) {
	ctx := context.Background()
	locker := newTestLocker(t)

	key := fmt.Sprintf("concurrent-acquire-%d", time.Now().UnixNano())

	const contenders = 20
	var winners int64
	errCh := make(chan error, contenders)
	var wg sync.WaitGroup

	for i := range contenders {
		wg.Add(1)
		go func(i int) {
			defer wg.Done()
			owner := fmt.Sprintf("owner-%02d", i)
			acquired, err := locker.Acquire(ctx, key, owner, 5*time.Second)
			if err != nil {
				errCh <- err
				return
			}
			if acquired {
				atomic.AddInt64(&winners, 1)
			}
		}(i)
	}

	wg.Wait()
	close(errCh)
	for err := range errCh {
		require.NoError(t, err)
	}
	require.Equal(t, int64(1), winners)
}

func TestInputValidation(t *testing.T) {
	ctx := context.Background()
	locker := newTestLocker(t)

	_, err := locker.Acquire(ctx, "", "owner", time.Second)
	require.ErrorIs(t, err, ErrInvalidKey)

	_, err = locker.Acquire(ctx, "key", "", time.Second)
	require.ErrorIs(t, err, ErrInvalidOwner)

	_, err = locker.Acquire(ctx, "key", "owner", 0)
	require.ErrorIs(t, err, ErrInvalidTTL)
}

func newTestLocker(t *testing.T) Locker {
	t.Helper()

	locker, err := NewLocker(leaseTestProvider)
	require.NoError(t, err)
	return locker
}

func startLeaseMysqlContainer(ctx context.Context) (*mysqlcontainer.MySQLContainer, string, uint16, error) {
	if err := os.Setenv("TESTCONTAINERS_RYUK_DISABLED", "true"); err != nil {
		return nil, "", 0, err
	}
	if err := os.Setenv("DOCKER_API_VERSION", "1.44"); err != nil {
		return nil, "", 0, err
	}

	container, err := mysqlcontainer.RunContainer(
		ctx,
		testcontainers.WithImage("gcr.io/pingcap-public/third-party/mysql:8.0"),
		mysqlcontainer.WithDatabase(leaseTestDBName),
		mysqlcontainer.WithUsername(leaseTestUser),
		mysqlcontainer.WithPassword(leaseTestPassword),
		testcontainers.CustomizeRequestOption(func(req *testcontainers.GenericContainerRequest) {
			req.Tmpfs = map[string]string{"/var/lib/mysql": "rw"}
		}),
	)
	if err != nil {
		return nil, "", 0, err
	}

	host, err := container.Host(ctx)
	if err != nil {
		return nil, "", 0, terminateLeaseMysqlContainer(ctx, container, err)
	}

	port, err := container.MappedPort(ctx, "3306/tcp")
	if err != nil {
		return nil, "", 0, terminateLeaseMysqlContainer(ctx, container, err)
	}

	parsed, err := strconv.ParseUint(port.Port(), 10, 16)
	if err != nil {
		return nil, "", 0, terminateLeaseMysqlContainer(ctx, container, err)
	}

	return container, host, uint16(parsed), nil
}

func terminateLeaseMysqlContainer(
	ctx context.Context,
	container *mysqlcontainer.MySQLContainer,
	cause error,
) error {
	if terminateErr := container.Terminate(ctx); terminateErr != nil {
		return fmt.Errorf("%w; terminate mysql container failed: %w", cause, terminateErr)
	}
	return cause
}