package resourcepool

import (
	"bytes"
	"context"
	"database/sql"
	"fmt"
	"log"
	"strings"
	"testing"
	"time"

	sqlmock "github.com/DATA-DOG/go-sqlmock"
	"github.com/stretchr/testify/require"
	"gorm.io/driver/mysql"
	"gorm.io/gorm"
	"gorm.io/gorm/logger"

	globalcommonv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/common/v1"
	globalngv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/nextgen/v1"
	resourcepoolrepo "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/resourcepool"
)

type resourcePoolBackfillTestProvider struct {
	db *gorm.DB
}

func (p resourcePoolBackfillTestProvider) ReadDB(context.Context) *gorm.DB {
	return p.db
}

func (p resourcePoolBackfillTestProvider) WriteDB(context.Context) *gorm.DB {
	return p.db
}

func TestBackfillMetadataUsesIndependentConditionalUpdates(t *testing.T) {
	var output bytes.Buffer
	db, err := gorm.Open(mysql.New(mysql.Config{
		Conn:                      &sql.DB{},
		SkipInitializeWithVersion: true,
	}), &gorm.Config{
		DisableAutomaticPing:   true,
		DryRun:                 true,
		SkipDefaultTransaction: true,
		Logger: logger.New(
			log.New(&output, "", 0),
			logger.Config{LogLevel: logger.Info},
		),
	})
	require.NoError(t, err)

	repo := NewResourcePoolRepo(resourcePoolBackfillTestProvider{db: db})
	name := "backfilled"
	lifecycle := globalngv1.ResourcePool_ACTIVE
	vcpuMode := globalngv1.ResourcePool_LIMITED
	_, backfillErr := repo.BackfillMetadata(t.Context(), &resourcepoolrepo.MetadataBackfillPatch{
		ResourcePoolID:    101,
		NormalizedName:    &name,
		CIDRStartIPv4:     []byte{10, 0, 0, 0},
		CIDREndIPv4:       []byte{10, 0, 0, 255},
		LifecycleState:    &lifecycle,
		VcpuLimitMode:     &vcpuMode,
		CloudResourceTags: map[string]string{},
		BackfillTags:      true,
	})
	require.NoError(t, backfillErr)

	cidrSQL := loggedSQLContaining(t, output.String(), "cidr_start_ipv4")
	lifecycleSQL := loggedSQLContaining(t, output.String(), "SET `lifecycle_state`")
	vcpuSQL := loggedSQLContaining(t, output.String(), "SET `vcpu_limit_mode`")
	tagsSQL := loggedSQLContaining(t, output.String(), "cloud_resource_tags_json")

	for _, statement := range []string{cidrSQL, lifecycleSQL, vcpuSQL, tagsSQL} {
		require.NotContains(t, statement, "normalized_name = ''")
	}
	require.Contains(t, cidrSQL, "cidr_start_ipv4 IS NULL OR cidr_end_ipv4 IS NULL")
	require.NotContains(t, lifecycleSQL, "cidr_start_ipv4 IS NULL")
	require.NotContains(t, vcpuSQL, "lifecycle_state = ''")
	require.NotContains(t, tagsSQL, "vcpu_limit_mode IN")
}

func TestCompletePauseResumeLifecycleUpdatesResourcePoolDirectly(t *testing.T) {
	var output bytes.Buffer
	db, err := gorm.Open(mysql.New(mysql.Config{
		Conn:                      &sql.DB{},
		SkipInitializeWithVersion: true,
	}), &gorm.Config{
		DisableAutomaticPing:   true,
		DryRun:                 true,
		SkipDefaultTransaction: true,
		Logger: logger.New(
			log.New(&output, "", 0),
			logger.Config{LogLevel: logger.Info},
		),
	})
	require.NoError(t, err)

	repo := NewResourcePoolRepo(resourcePoolBackfillTestProvider{db: db})
	updated, updateErr := repo.CompletePauseResumeLifecycle(
		t.Context(),
		101,
		globalngv1.ResourcePool_PAUSING,
		globalngv1.ResourcePool_PAUSED,
	)
	require.NoError(t, updateErr)
	require.False(t, updated)

	updateSQL := loggedSQLContaining(t, output.String(), "UPDATE `resource_pools`")
	require.Contains(t, updateSQL, "lifecycle_state")
	require.Contains(t, updateSQL, "paused_at")
	require.Contains(t, updateSQL, "automatic_resume_at")
	require.Contains(t, updateSQL, "lifecycle_state = 'PAUSING'")
	require.Contains(t, output.String(), "SELECT")
	require.Contains(t, output.String(), "deleted_at")
}

func TestCompletePauseResumeLifecycleTreatsBenignRacesAsSuccess(t *testing.T) {
	t.Parallel()

	tests := []struct {
		name           string
		lifecycleState string
		deletedAt      *time.Time
		wantUpdated    bool
	}{
		{
			name:           "already reached target state",
			lifecycleState: globalngv1.ResourcePool_PAUSED.String(),
			wantUpdated:    true,
		},
		{
			name:           "soft deleted",
			lifecycleState: globalngv1.ResourcePool_PAUSING.String(),
			deletedAt:      func() *time.Time { value := time.Now().UTC(); return &value }(),
			wantUpdated:    true,
		},
		{
			name:           "ownership lost",
			lifecycleState: globalngv1.ResourcePool_RESUMING.String(),
			wantUpdated:    false,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			db, mock := newResourcePoolLifecycleMockDB(t)
			repo := NewResourcePoolRepo(resourcePoolBackfillTestProvider{db: db})

			mock.ExpectExec("UPDATE `resource_pools`").
				WillReturnResult(sqlmock.NewResult(0, 0))
			rows := sqlmock.NewRows([]string{"lifecycle_state", "deleted_at"}).
				AddRow(tt.lifecycleState, tt.deletedAt)
			mock.ExpectQuery("SELECT .*lifecycle_state.*deleted_at.*FROM `resource_pools`").
				WillReturnRows(rows)

			updated, updateErr := repo.CompletePauseResumeLifecycle(
				t.Context(),
				101,
				globalngv1.ResourcePool_PAUSING,
				globalngv1.ResourcePool_PAUSED,
			)
			require.NoError(t, updateErr)
			require.Equal(t, tt.wantUpdated, updated)
			require.NoError(t, mock.ExpectationsWereMet())
		})
	}
}

func TestLoadLifecycleForUpdateHonorsDeletedScopeAndLock(t *testing.T) {
	for _, tt := range []struct {
		name        string
		showDeleted bool
		deletedAt   any
		wantDeleted bool
	}{
		{
			name:        "live pool",
			deletedAt:   nil,
			wantDeleted: false,
		},
		{
			name:        "soft deleted pool",
			showDeleted: true,
			deletedAt:   time.Now().UTC(),
			wantDeleted: true,
		},
	} {
		t.Run(tt.name, func(t *testing.T) {
			matcher := sqlmock.QueryMatcherFunc(func(_, actualSQL string) error {
				normalizedSQL := strings.ReplaceAll(strings.ToUpper(strings.Join(strings.Fields(actualSQL), " ")), "`", "")
				if !strings.Contains(normalizedSQL, "FOR UPDATE") {
					return fmt.Errorf("expected lifecycle query to lock the row: %s", actualSQL)
				}
				hasLivePoolScope := strings.Contains(normalizedSQL, "DELETED_AT IS NULL")
				if hasLivePoolScope == tt.showDeleted {
					return fmt.Errorf("unexpected deleted_at scope in lifecycle query: %s", actualSQL)
				}
				return nil
			})
			sqlDB, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(matcher))
			require.NoError(t, err)
			t.Cleanup(func() { _ = sqlDB.Close() })

			db, err := gorm.Open(mysql.New(mysql.Config{
				Conn:                      sqlDB,
				SkipInitializeWithVersion: true,
			}), &gorm.Config{
				SkipDefaultTransaction: true,
				Logger:                 logger.Default.LogMode(logger.Silent),
			})
			require.NoError(t, err)
			repo := NewResourcePoolRepo(resourcePoolBackfillTestProvider{db: db})
			mock.ExpectQuery("lifecycle_state").
				WillReturnRows(sqlmock.NewRows([]string{"lifecycle_state", "deleted_at"}).AddRow(
					globalngv1.ResourcePool_RESUMING.String(),
					tt.deletedAt,
				))

			got, loadErr := repo.LoadLifecycleForUpdate(t.Context(), 101, tt.showDeleted)
			require.NoError(t, loadErr)
			require.Equal(t, globalngv1.ResourcePool_RESUMING, got.State)
			require.Equal(t, tt.wantDeleted, got.IsDeleted)
			require.NoError(t, mock.ExpectationsWereMet())
		})
	}
}

func newResourcePoolLifecycleMockDB(t *testing.T) (*gorm.DB, sqlmock.Sqlmock) {
	t.Helper()

	sqlDB, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp))
	require.NoError(t, err)
	t.Cleanup(func() {
		_ = sqlDB.Close()
	})

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

func TestLoadLifecycleForUpdateDoesNotLoadAssociations(t *testing.T) {
	db, mock := newResourcePoolLifecycleMockDB(t)
	repo := NewResourcePoolRepo(resourcePoolBackfillTestProvider{db: db})
	deletedAt := time.Now().UTC()

	rows := sqlmock.NewRows([]string{"lifecycle_state", "deleted_at"}).
		AddRow(globalngv1.ResourcePool_PAUSED.String(), deletedAt)
	mock.ExpectQuery("SELECT .*lifecycle_state.*deleted_at.*FROM `resource_pools`.*FOR UPDATE").
		WithArgs(uint64(42), 1).
		WillReturnRows(rows)

	lifecycle, loadErr := repo.LoadLifecycleForUpdate(t.Context(), 42, true)
	require.NoError(t, loadErr)
	require.Equal(t, globalngv1.ResourcePool_PAUSED, lifecycle.State)
	require.True(t, lifecycle.IsDeleted)
	require.NoError(t, mock.ExpectationsWereMet())
}

func TestLoadForUpdateUsesIndependentQueriesWhenIncludingDeleted(t *testing.T) {
	matcher := sqlmock.QueryMatcherFunc(func(expectedSQL, actualSQL string) error {
		normalizedSQL := strings.ReplaceAll(strings.ToUpper(strings.Join(strings.Fields(actualSQL), " ")), "`", "")
		switch expectedSQL {
		case "resource pool":
			if !strings.Contains(normalizedSQL, "FROM RESOURCE_POOLS") ||
				!strings.Contains(normalizedSQL, "FOR UPDATE") ||
				strings.Contains(normalizedSQL, "DELETED_AT IS NULL") {
				return fmt.Errorf("expected a resource pool locking query: %s", actualSQL)
			}
		case "provider":
			if !strings.Contains(normalizedSQL, "FROM PROVIDERS") ||
				strings.Contains(normalizedSQL, "RESOURCE_POOLS.ID") ||
				strings.Contains(normalizedSQL, "FOR UPDATE") {
				return fmt.Errorf("expected an independent provider query: %s", actualSQL)
			}
		case "region":
			if !strings.Contains(normalizedSQL, "FROM REGIONS") ||
				strings.Contains(normalizedSQL, "RESOURCE_POOLS.ID") ||
				strings.Contains(normalizedSQL, "FOR UPDATE") {
				return fmt.Errorf("expected an independent region query: %s", actualSQL)
			}
		default:
			return fmt.Errorf("unexpected query expectation: %s", expectedSQL)
		}
		return nil
	})
	sqlDB, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(matcher))
	require.NoError(t, err)
	t.Cleanup(func() { _ = sqlDB.Close() })

	db, err := gorm.Open(mysql.New(mysql.Config{
		Conn:                      sqlDB,
		SkipInitializeWithVersion: true,
	}), &gorm.Config{
		SkipDefaultTransaction: true,
		Logger:                 logger.Default.LogMode(logger.Silent),
	})
	require.NoError(t, err)
	repo := NewResourcePoolRepo(resourcePoolBackfillTestProvider{db: db})
	deletedAt := time.Now().UTC()

	mock.ExpectQuery("resource pool").
		WithArgs(uint64(42), 1).
		WillReturnRows(sqlmock.NewRows([]string{
			"id", "provider_id", "region_id", "lifecycle_state", "vcpu_limit_mode", "service_plan", "deleted_at",
		}).AddRow(
			uint64(42), uint64(7), uint64(8), globalngv1.ResourcePool_ACTIVE.String(),
			globalngv1.ResourcePool_INHERITED.String(), globalcommonv1.ServicePlan_Premium.String(), deletedAt,
		))
	mock.ExpectQuery("provider").
		WithArgs(uint64(7), 1).
		WillReturnRows(sqlmock.NewRows([]string{"id", "name"}).AddRow(uint64(7), "aws"))
	mock.ExpectQuery("region").
		WithArgs(uint64(8), 1).
		WillReturnRows(sqlmock.NewRows([]string{"id", "region_name"}).AddRow(uint64(8), "us-east-1"))

	resourcePool, loadErr := repo.LoadForUpdate(t.Context(), 42, true)
	require.NoError(t, loadErr)
	require.Equal(t, uint64(42), resourcePool.GetId())
	require.NotNil(t, resourcePool.GetDeleteTime())
	require.NoError(t, mock.ExpectationsWereMet())
}

func TestQueryKeepsCountSeparateFromSeekPageFetch(t *testing.T) {
	var output bytes.Buffer
	db, err := gorm.Open(mysql.New(mysql.Config{
		Conn:                      &sql.DB{},
		SkipInitializeWithVersion: true,
	}), &gorm.Config{
		DisableAutomaticPing:   true,
		DryRun:                 true,
		SkipDefaultTransaction: true,
		Logger: logger.New(
			log.New(&output, "", 0),
			logger.Config{LogLevel: logger.Info},
		),
	})
	require.NoError(t, err)

	repo := NewResourcePoolRepo(resourcePoolBackfillTestProvider{db: db})
	_, _, queryErr := repo.Query(t.Context(), &resourcepoolrepo.Filter{PageSize: 2})
	require.NoError(t, queryErr)

	sqlLog := output.String()
	require.Contains(t, sqlLog, "SELECT count(*)")
	pageQuery := loggedSQLContaining(t, sqlLog, "ORDER BY `resource_pools`.`created_at` DESC")
	require.NotContains(t, pageQuery, "SELECT count(*)")
	require.Contains(t, pageQuery, "LIMIT 2")
}

func TestListDueAutomaticResumeFiltersPausedLivePoolsByDeadline(t *testing.T) {
	var output bytes.Buffer
	db, err := gorm.Open(mysql.New(mysql.Config{
		Conn:                      &sql.DB{},
		SkipInitializeWithVersion: true,
	}), &gorm.Config{
		DisableAutomaticPing:   true,
		DryRun:                 true,
		SkipDefaultTransaction: true,
		Logger: logger.New(
			log.New(&output, "", 0),
			logger.Config{LogLevel: logger.Info},
		),
	})
	require.NoError(t, err)

	repo := NewResourcePoolRepo(resourcePoolBackfillTestProvider{db: db})
	_, listErr := repo.ListDueAutomaticResume(t.Context())
	require.NoError(t, listErr)

	query := loggedSQLContaining(t, output.String(), "automatic_resume_at")
	require.Contains(t, query, "lifecycle_state = 'PAUSED'")
	require.Contains(t, query, "automatic_resume_at <= CURRENT_TIMESTAMP(3)")
	require.Contains(t, query, "deleted_at` IS NULL")
	require.Contains(t, query, "ORDER BY id")
}

func TestHasLiveCIDROverlapScopesQueryToPublicRegion(t *testing.T) {
	var output bytes.Buffer
	db, err := gorm.Open(mysql.New(mysql.Config{
		Conn:                      &sql.DB{},
		SkipInitializeWithVersion: true,
	}), &gorm.Config{
		DisableAutomaticPing:   true,
		DryRun:                 true,
		SkipDefaultTransaction: true,
		Logger: logger.New(
			log.New(&output, "", 0),
			logger.Config{LogLevel: logger.Info},
		),
	})
	require.NoError(t, err)

	repo := NewResourcePoolRepo(resourcePoolBackfillTestProvider{db: db})
	_, overlapErr := repo.HasLiveCIDROverlap(
		t.Context(),
		42,
		globalcommonv1.CloudProvider_AWS,
		globalcommonv1.Region_AWS_US_WEST_2,
		[]byte{10, 0, 0, 0},
		[]byte{10, 0, 255, 255},
		0,
	)
	require.NoError(t, overlapErr)

	query := loggedSQLContaining(t, output.String(), "cidr_start_ipv4")
	require.Contains(t, query, "resource_pools.owner_id = 42")
	require.Contains(t, query, "FROM `providers` WHERE name = 'aws'")
	require.Contains(t, query, "FROM `regions` WHERE region_name = 'us-west-2'")
}

func loggedSQLContaining(t *testing.T, output, needle string) string {
	t.Helper()

	for line := range strings.SplitSeq(output, "\n") {
		if strings.Contains(line, needle) {
			return line
		}
	}
	t.Fatalf("SQL log does not contain %q:\n%s", needle, output)
	return ""
}