//go:build integration

package resourcepool

import (
	"context"
	"fmt"
	"runtime"
	"strconv"
	"sync"
	"testing"
	"time"

	"github.com/stretchr/testify/require"
	"gorm.io/gorm"

	globalcommonv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/common/v1"
	globalngv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/nextgen/v1"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/entity"
	resourcepoolrepo "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/resourcepool"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/infra/repoimpl/schema"
	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
)

func TestResourcePoolRepo_LockOrganizationNamespaceSerializesTransactions(t *testing.T) {
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	repo := NewResourcePoolRepo(integrationDBProvider)
	const organizationID = uint64(9_100_001)

	firstLocked := make(chan struct{})
	releaseFirst := make(chan struct{})
	secondStarted := make(chan struct{})
	secondLocked := make(chan struct{})
	errorsCh := make(chan error, 2)

	var wg sync.WaitGroup
	wg.Add(2)
	go func() {
		defer wg.Done()
		err := integrationDBProvider.Transaction(ctx, func(txCtx context.Context) error {
			if lockErr := repo.LockOrganizationNamespace(txCtx, organizationID); lockErr != nil {
				return lockErr
			}
			close(firstLocked)
			select {
			case <-releaseFirst:
				return nil
			case <-txCtx.Done():
				return txCtx.Err()
			}
		})
		errorsCh <- err
	}()

	select {
	case <-firstLocked:
	case <-ctx.Done():
		t.Fatal("first transaction did not acquire the namespace lock")
	}

	go func() {
		defer wg.Done()
		err := integrationDBProvider.Transaction(ctx, func(txCtx context.Context) error {
			close(secondStarted)
			if lockErr := repo.LockOrganizationNamespace(txCtx, organizationID); lockErr != nil {
				return lockErr
			}
			close(secondLocked)
			return nil
		})
		errorsCh <- err
	}()

	select {
	case <-secondStarted:
	case <-ctx.Done():
		t.Fatal("second transaction did not start")
	}

	close(releaseFirst)
	select {
	case <-secondLocked:
	case <-ctx.Done():
		t.Fatal("second transaction did not acquire the namespace after the first transaction committed")
	}

	wg.Wait()
	close(errorsCh)
	for err := range errorsCh {
		require.NoError(t, err)
	}

	var namespaceCount int64
	require.NoError(t, integrationDBProvider.ReadDB(ctx).
		Model(&schema.ResourcePoolNamespace{}).
		Where("organization_id = ?", organizationID).
		Count(&namespaceCount).Error)
	require.Equal(t, int64(1), namespaceCount)
}

func TestResourcePoolRepo_LoadForUpdateSerializesTransactions(t *testing.T) {
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	repo := NewResourcePoolRepo(integrationDBProvider)
	pool := resourcePoolModel(9_200_000, 9_100_000, time.Now().UTC())
	require.NoError(t, integrationDBProvider.WriteDB(ctx).Create(pool).Error)
	t.Cleanup(func() {
		require.NoError(t, integrationDBProvider.WriteDB(context.Background()).Unscoped().Delete(pool).Error)
	})

	firstLocked := make(chan struct{})
	releaseFirst := make(chan struct{})
	secondStarted := make(chan struct{})
	secondLocked := make(chan struct{})
	errorsCh := make(chan error, 2)

	var wg sync.WaitGroup
	wg.Add(2)
	go func() {
		defer wg.Done()
		errorsCh <- integrationDBProvider.Transaction(ctx, func(txCtx context.Context) error {
			if _, lockErr := repo.LoadForUpdate(txCtx, pool.ID, false); lockErr != nil {
				return lockErr
			}
			close(firstLocked)
			select {
			case <-releaseFirst:
				return nil
			case <-txCtx.Done():
				return txCtx.Err()
			}
		})
	}()

	select {
	case <-firstLocked:
	case <-ctx.Done():
		t.Fatal("first transaction did not lock the resource pool")
	}

	go func() {
		defer wg.Done()
		errorsCh <- integrationDBProvider.Transaction(ctx, func(txCtx context.Context) error {
			close(secondStarted)
			if _, lockErr := repo.LoadForUpdate(txCtx, pool.ID, false); lockErr != nil {
				return lockErr
			}
			close(secondLocked)
			return nil
		})
	}()

	select {
	case <-secondStarted:
	case <-ctx.Done():
		t.Fatal("second transaction did not start")
	}

	close(releaseFirst)
	select {
	case <-secondLocked:
	case <-ctx.Done():
		t.Fatal("second transaction did not acquire the row after the first transaction committed")
	}

	wg.Wait()
	close(errorsCh)
	for err := range errorsCh {
		require.NoError(t, err)
	}
}

func TestResourcePoolRepo_AttachCommitsBeforeDeleteRechecksUsage(t *testing.T) {
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	repo := NewResourcePoolRepo(integrationDBProvider)
	pool := resourcePoolModel(9_200_100, 9_100_100, time.Now().UTC())
	cluster := &schema.Cluster{
		ID:           9_400_100,
		TenantId:     pool.OwnerId,
		ProviderId:   pool.ProviderId,
		RegionId:     pool.RegionId,
		State:        "CREATING",
		ServicePlan:  globalcommonv1.ServicePlan_BYOC.String(),
		SharedPoolId: strconv.FormatUint(pool.ID, 10),
		Attributes:   schema.ClusterAttributes{},
	}
	require.NoError(t, integrationDBProvider.WriteDB(ctx).Create(pool).Error)
	t.Cleanup(func() {
		require.NoError(t, integrationDBProvider.WriteDB(context.Background()).
			Unscoped().
			Delete(cluster).Error)
		require.NoError(t, integrationDBProvider.WriteDB(context.Background()).
			Unscoped().
			Delete(pool).Error)
	})

	attachLocked := make(chan struct{})
	releaseAttach := make(chan struct{})
	deleteConnectionID := make(chan uint64, 1)
	attachResult := make(chan error, 1)
	deleteResult := make(chan error, 1)

	go func() {
		attachResult <- integrationDBProvider.Transaction(ctx, func(txCtx context.Context) error {
			if _, err := repo.LoadForUpdate(txCtx, pool.ID, false); err != nil {
				return err
			}
			close(attachLocked)
			select {
			case <-releaseAttach:
			case <-txCtx.Done():
				return txCtx.Err()
			}
			return integrationDBProvider.WriteDB(txCtx).Create(cluster).Error
		})
	}()

	select {
	case <-attachLocked:
	case <-ctx.Done():
		t.Fatal("attach transaction did not lock the resource pool")
	}

	go func() {
		deleteResult <- integrationDBProvider.Transaction(ctx, func(txCtx context.Context) error {
			connectionID, err := currentMySQLConnectionID(txCtx)
			if err != nil {
				return err
			}
			deleteConnectionID <- connectionID
			if _, err := repo.LoadForUpdate(txCtx, pool.ID, false); err != nil {
				return err
			}
			var attachedClusters int64
			if err := integrationDBProvider.ReadDB(txCtx).
				Model(&schema.Cluster{}).
				Where("shared_pool_id = ?", strconv.FormatUint(pool.ID, 10)).
				Count(&attachedClusters).Error; err != nil {
				return err
			}
			if attachedClusters > 0 {
				return fmt.Errorf("resource pool is used by some clusters")
			}
			return repo.Delete(txCtx, pool.ID)
		})
	}()

	var connectionID uint64
	select {
	case connectionID = <-deleteConnectionID:
	case <-ctx.Done():
		t.Fatal("delete transaction did not attempt the resource-pool lock")
	}
	require.NoError(t, waitForMySQLLockWait(ctx, connectionID))
	close(releaseAttach)

	require.NoError(t, <-attachResult)
	require.ErrorContains(t, <-deleteResult, "used by some clusters")

	_, err := repo.Load(ctx, pool.ID, false)
	require.NoError(t, err)
	var attachedClusters int64
	require.NoError(t, integrationDBProvider.ReadDB(ctx).
		Model(&schema.Cluster{}).
		Where("shared_pool_id = ?", strconv.FormatUint(pool.ID, 10)).
		Count(&attachedClusters).Error)
	require.Equal(t, int64(1), attachedClusters)
}

func TestResourcePoolRepo_NamespaceSerializesCreateCreateConflict(t *testing.T) {
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	repo := NewResourcePoolRepo(integrationDBProvider)
	const organizationID = uint64(9_100_010)
	first := resourcePoolModel(9_200_010, organizationID, time.Now().UTC())
	second := resourcePoolModel(9_200_011, organizationID, first.CreatedAt.Add(time.Millisecond))
	first.Name, first.NormalizedName = "same-name", "same-name"
	second.Name, second.NormalizedName = "same-name", "same-name"
	t.Cleanup(func() {
		require.NoError(t, integrationDBProvider.WriteDB(context.Background()).Unscoped().
			Where("owner_id = ?", organizationID).
			Delete(&schema.ResourcePool{}).Error)
	})

	firstValidated := make(chan struct{})
	releaseFirst := make(chan struct{})
	secondConnectionID := make(chan uint64, 1)
	firstResult := make(chan error, 1)
	secondResult := make(chan error, 1)
	go func() {
		firstResult <- createResourcePoolUnderNamespace(
			ctx,
			repo,
			first,
			nil,
			firstValidated,
			releaseFirst,
		)
	}()

	select {
	case <-firstValidated:
	case <-ctx.Done():
		t.Fatal("first create did not validate while holding the namespace")
	}
	go func() {
		secondResult <- createResourcePoolUnderNamespace(
			ctx,
			repo,
			second,
			secondConnectionID,
			nil,
			nil,
		)
	}()
	var connectionID uint64
	select {
	case connectionID = <-secondConnectionID:
	case <-ctx.Done():
		t.Fatal("second create did not attempt the namespace transaction")
	}
	require.NoError(t, waitForMySQLLockWait(ctx, connectionID))
	close(releaseFirst)

	require.NoError(t, <-firstResult)
	require.ErrorContains(t, <-secondResult, "resource pool name conflict")
	var liveCount int64
	require.NoError(t, integrationDBProvider.ReadDB(ctx).Model(&schema.ResourcePool{}).
		Where("owner_id = ? AND normalized_name = ?", organizationID, "same-name").
		Count(&liveCount).Error)
	require.Equal(t, int64(1), liveCount)
}

func TestResourcePoolRepo_NamespaceSerializesCreateRenameConflict(t *testing.T) {
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	repo := NewResourcePoolRepo(integrationDBProvider)
	const organizationID = uint64(9_100_011)
	existing := resourcePoolModel(9_200_012, organizationID, time.Now().UTC())
	existing.Name, existing.NormalizedName = "old-name", "old-name"
	creating := resourcePoolModel(9_200_013, organizationID, existing.CreatedAt.Add(time.Millisecond))
	creating.Name, creating.NormalizedName = "new-name", "new-name"
	require.NoError(t, integrationDBProvider.WriteDB(ctx).Create(existing).Error)
	t.Cleanup(func() {
		require.NoError(t, integrationDBProvider.WriteDB(context.Background()).Unscoped().
			Where("owner_id = ?", organizationID).
			Delete(&schema.ResourcePool{}).Error)
	})

	createValidated := make(chan struct{})
	releaseCreate := make(chan struct{})
	renameConnectionID := make(chan uint64, 1)
	createResult := make(chan error, 1)
	renameResult := make(chan error, 1)
	go func() {
		createResult <- createResourcePoolUnderNamespace(
			ctx,
			repo,
			creating,
			nil,
			createValidated,
			releaseCreate,
		)
	}()
	select {
	case <-createValidated:
	case <-ctx.Done():
		t.Fatal("create did not validate while holding the namespace")
	}

	go func() {
		renameResult <- renameResourcePoolUnderNamespace(
			ctx,
			repo,
			existing.ID,
			existing.OwnerId,
			"new-name",
			renameConnectionID,
		)
	}()
	var connectionID uint64
	select {
	case connectionID = <-renameConnectionID:
	case <-ctx.Done():
		t.Fatal("rename did not attempt the namespace transaction")
	}
	require.NoError(t, waitForMySQLLockWait(ctx, connectionID))
	close(releaseCreate)

	require.NoError(t, <-createResult)
	require.ErrorContains(t, <-renameResult, "resource pool name conflict")
	loaded, err := repo.Load(ctx, existing.ID, false)
	require.NoError(t, err)
	require.Equal(t, "old-name", loaded.GetName())
	var targetNameCount int64
	require.NoError(t, integrationDBProvider.ReadDB(ctx).Model(&schema.ResourcePool{}).
		Where("owner_id = ? AND normalized_name = ?", organizationID, "new-name").
		Count(&targetNameCount).Error)
	require.Equal(t, int64(1), targetNameCount)
}

func createResourcePoolUnderNamespace(
	ctx context.Context,
	repo resourcepoolrepo.Repo,
	pool *schema.ResourcePool,
	connectionID chan<- uint64,
	validated chan<- struct{},
	release <-chan struct{},
) error {
	return integrationDBProvider.Transaction(ctx, func(txCtx context.Context) error {
		if connectionID != nil {
			id, err := currentMySQLConnectionID(txCtx)
			if err != nil {
				return err
			}
			connectionID <- id
		}
		if err := repo.LockOrganizationNamespace(txCtx, pool.OwnerId); err != nil {
			return err
		}
		conflict, err := repo.HasLiveNormalizedName(txCtx, pool.OwnerId, pool.NormalizedName, 0)
		if err != nil {
			return err
		}
		if conflict {
			return fmt.Errorf("resource pool name conflict")
		}
		if validated != nil {
			close(validated)
		}
		if release != nil {
			select {
			case <-release:
			case <-txCtx.Done():
				return txCtx.Err()
			}
		}
		return integrationDBProvider.WriteDB(txCtx).Create(pool).Error
	})
}

func renameResourcePoolUnderNamespace(
	ctx context.Context,
	repo resourcepoolrepo.Repo,
	resourcePoolID uint64,
	organizationID uint64,
	name string,
	connectionID chan<- uint64,
) error {
	return integrationDBProvider.Transaction(ctx, func(txCtx context.Context) error {
		if connectionID != nil {
			id, err := currentMySQLConnectionID(txCtx)
			if err != nil {
				return err
			}
			connectionID <- id
		}
		if err := repo.LockOrganizationNamespace(txCtx, organizationID); err != nil {
			return err
		}
		locked, err := repo.LoadForUpdate(txCtx, resourcePoolID, false)
		if err != nil {
			return err
		}
		normalizedName := entity.NormalizeResourcePoolName(name)
		conflict, err := repo.HasLiveNormalizedName(txCtx, locked.GetOwnerId(), normalizedName, resourcePoolID)
		if err != nil {
			return err
		}
		if conflict {
			return fmt.Errorf("resource pool name conflict")
		}
		locked.Name = name
		return repo.Save(txCtx, locked)
	})
}

func currentMySQLConnectionID(ctx context.Context) (uint64, error) {
	var connectionID uint64
	if err := integrationDBProvider.WriteDB(ctx).
		Raw("SELECT CONNECTION_ID()").
		Scan(&connectionID).Error; err != nil {
		return 0, err
	}
	return connectionID, nil
}

func waitForMySQLLockWait(ctx context.Context, connectionID uint64) error {
	for {
		var waitingTransactions int64
		if err := integrationDBProvider.ReadDB(ctx).
			Raw(
				"SELECT COUNT(*) FROM information_schema.innodb_trx "+
					"WHERE trx_mysql_thread_id = ? AND trx_state = 'LOCK WAIT'",
				connectionID,
			).
			Scan(&waitingTransactions).Error; err != nil {
			return err
		}
		if waitingTransactions == 1 {
			return nil
		}
		select {
		case <-ctx.Done():
			return ctx.Err()
		default:
			runtime.Gosched()
		}
	}
}

func TestResourcePoolRepo_LiveNamespaceConflictsAndDeletionReuse(t *testing.T) {
	ctx := context.Background()
	repo := NewResourcePoolRepo(integrationDBProvider)
	const (
		organizationID = uint64(9_100_002)
		resourcePoolID = uint64(9_200_001)
	)

	createdAt := time.Date(2026, time.July, 21, 10, 0, 0, 0, time.UTC)
	pool := resourcePoolModel(resourcePoolID, organizationID, createdAt)
	pool.NormalizedName = "production-west"
	pool.CidrStartIPv4 = []byte{10, 40, 0, 0}
	pool.CidrEndIPv4 = []byte{10, 40, 255, 255}
	require.NoError(t, integrationDBProvider.WriteDB(ctx).Create(pool).Error)
	legacyPool := resourcePoolModel(resourcePoolID+1, organizationID, createdAt.Add(time.Second))
	legacyPool.Name = "  Legacy-Global  "
	legacyPool.NormalizedName = ""
	legacyPool.Cidr = "192.168.7.42/24"
	legacyPool.CidrStartIPv4 = nil
	legacyPool.CidrEndIPv4 = nil
	legacyPool.ServicePlan = globalcommonv1.ServicePlan_Premium_Reserved.String()
	legacyPool.RegionId = 4
	require.NoError(t, integrationDBProvider.WriteDB(ctx).Create(legacyPool).Error)
	crossPlanPool := resourcePoolModel(resourcePoolID+5, organizationID, createdAt.Add(5*time.Second))
	crossPlanPool.Name = "production-west"
	crossPlanPool.NormalizedName = "production-west"
	crossPlanPool.ServicePlan = globalcommonv1.ServicePlan_Premium_Reserved.String()
	require.NoError(t, integrationDBProvider.WriteDB(ctx).Create(crossPlanPool).Error)
	legacyListPool := resourcePoolModel(resourcePoolID+2, organizationID, createdAt.Add(2*time.Second))
	legacyListPool.Name = "Legacy-List"
	legacyListPool.NormalizedName = ""
	legacyListPool.Cidr = "172.31.8.0/24"
	legacyListPool.CidrStartIPv4 = nil
	legacyListPool.CidrEndIPv4 = nil
	legacyListPool.ServicePlan = globalcommonv1.ServicePlan_Premium_Reserved.String()
	require.NoError(t, integrationDBProvider.WriteDB(ctx).Create(legacyListPool).Error)
	legacyBYOCPool := resourcePoolModel(resourcePoolID+3, organizationID, createdAt.Add(3*time.Second))
	legacyBYOCPool.Name = "  Legacy-BYOC  "
	legacyBYOCPool.NormalizedName = ""
	legacyBYOCPool.Cidr = "192.168.9.42/24"
	legacyBYOCPool.CidrStartIPv4 = nil
	legacyBYOCPool.CidrEndIPv4 = nil
	legacyBYOCPool.RegionId = 4
	require.NoError(t, integrationDBProvider.WriteDB(ctx).Create(legacyBYOCPool).Error)
	samePhysicalRegionPool := resourcePoolModel(resourcePoolID+4, organizationID, createdAt.Add(4*time.Second))
	samePhysicalRegionPool.Name = "same-physical-region"
	samePhysicalRegionPool.Cidr = "192.168.11.42/24"
	samePhysicalRegionPool.CidrStartIPv4 = nil
	samePhysicalRegionPool.CidrEndIPv4 = nil
	samePhysicalRegionPool.RegionId = 2
	require.NoError(t, integrationDBProvider.WriteDB(ctx).Create(samePhysicalRegionPool).Error)
	t.Cleanup(func() {
		require.NoError(t, integrationDBProvider.WriteDB(context.Background()).Unscoped().
			Where("owner_id = ?", organizationID).
			Delete(&schema.ResourcePool{}).Error)
	})

	nameConflict, err := repo.HasLiveNormalizedName(ctx, organizationID, "production-west", 0)
	require.NoError(t, err)
	require.True(t, nameConflict)
	nameConflict, err = repo.HasLiveNormalizedName(ctx, organizationID, "production-west", resourcePoolID)
	require.NoError(t, err)
	require.True(t, nameConflict, "a matching name in another service plan must remain a conflict")
	nameConflict, err = repo.HasLiveNormalizedName(ctx, organizationID+1, "production-west", 0)
	require.NoError(t, err)
	require.False(t, nameConflict)
	nameConflict, err = repo.HasLiveNormalizedName(ctx, organizationID, "legacy-global", 0)
	require.NoError(t, err)
	require.True(t, nameConflict, "non-BYOC organization resource pool names must conflict")
	nameConflict, err = repo.HasLiveNormalizedName(ctx, organizationID, "production-west", 0)
	require.NoError(t, err)
	require.True(t, nameConflict, "resource pool names must conflict across service plans")
	legacyCIDROverlap, err := repo.HasLiveCIDROverlap(
		ctx,
		organizationID,
		globalcommonv1.CloudProvider_AWS,
		globalcommonv1.Region_AWS_US_WEST_2,
		[]byte{192, 168, 7, 200},
		[]byte{192, 168, 8, 10},
		0,
	)
	require.NoError(t, err)
	require.False(t, legacyCIDROverlap, "non-BYOC resource pool CIDRs must not conflict")
	nameConflict, err = repo.HasLiveNormalizedName(ctx, organizationID, "legacy-byoc", 0)
	require.NoError(t, err)
	require.True(t, nameConflict, "legacy BYOC metadata must still conflict across regions")
	legacyCIDROverlap, err = repo.HasLiveCIDROverlap(
		ctx,
		organizationID,
		globalcommonv1.CloudProvider_AWS,
		globalcommonv1.Region_AWS_US_WEST_2,
		[]byte{192, 168, 9, 200},
		[]byte{192, 168, 10, 10},
		0,
	)
	require.NoError(t, err)
	require.False(t, legacyCIDROverlap, "BYOC CIDRs in different public regions must not conflict")
	samePhysicalRegionOverlap, err := repo.HasLiveCIDROverlap(
		ctx,
		organizationID,
		globalcommonv1.CloudProvider_AWS,
		globalcommonv1.Region_AWS_US_WEST_2,
		[]byte{192, 168, 11, 200},
		[]byte{192, 168, 12, 10},
		0,
	)
	require.NoError(t, err)
	require.True(t, samePhysicalRegionOverlap, "logical regions in the same public region must still conflict")
	legacyList, legacyListTotal, err := repo.Query(ctx, &resourcepoolrepo.Filter{
		OwnerType: int(globalngv1.ResourcePool_ORGANIZATION),
		OwnerIds:  []uint64{organizationID},
		Name:      "legacy-list",
		PageSize:  10,
	})
	require.NoError(t, err)
	require.Equal(t, int64(1), legacyListTotal)
	require.Equal(t, []uint64{legacyListPool.ID}, resourcePoolIDs(legacyList))

	tests := []struct {
		name  string
		start []byte
		end   []byte
		want  bool
	}{
		{name: "same range", start: []byte{10, 40, 0, 0}, end: []byte{10, 40, 255, 255}, want: true},
		{name: "contained range", start: []byte{10, 40, 1, 0}, end: []byte{10, 40, 1, 255}, want: true},
		{name: "touches last address", start: []byte{10, 40, 255, 255}, end: []byte{10, 41, 0, 0}, want: true},
		{name: "adjacent higher range", start: []byte{10, 41, 0, 0}, end: []byte{10, 41, 255, 255}, want: false},
		{name: "lexically lower range", start: []byte{10, 39, 0, 0}, end: []byte{10, 39, 255, 255}, want: false},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			overlaps, overlapErr := repo.HasLiveCIDROverlap(
				ctx,
				organizationID,
				globalcommonv1.CloudProvider_AWS,
				globalcommonv1.Region_AWS_US_WEST_2,
				tt.start,
				tt.end,
				0,
			)
			require.NoError(t, overlapErr)
			require.Equal(t, tt.want, overlaps)
		})
	}

	overlaps, err := repo.HasLiveCIDROverlap(
		ctx,
		organizationID,
		globalcommonv1.CloudProvider_AWS,
		globalcommonv1.Region_AWS_US_WEST_2,
		[]byte{10, 40, 0, 0},
		[]byte{10, 40, 255, 255},
		resourcePoolID,
	)
	require.NoError(t, err)
	require.False(t, overlaps)

	require.NoError(t, repo.Delete(ctx, resourcePoolID))
	nameConflict, err = repo.HasLiveNormalizedName(ctx, organizationID, "production-west", 0)
	require.NoError(t, err)
	require.False(t, nameConflict)
	overlaps, err = repo.HasLiveCIDROverlap(
		ctx,
		organizationID,
		globalcommonv1.CloudProvider_AWS,
		globalcommonv1.Region_AWS_US_WEST_2,
		[]byte{10, 40, 0, 0},
		[]byte{10, 40, 255, 255},
		0,
	)
	require.NoError(t, err)
	require.False(t, overlaps)
}

func TestResourcePoolRepo_QueryUsesLifecycleAndStableCursor(t *testing.T) {
	ctx := context.Background()
	repo := NewResourcePoolRepo(integrationDBProvider)
	const organizationID = uint64(9_100_003)

	tiedCreatedAt := time.Date(2026, time.July, 21, 11, 0, 0, 123_000_000, time.UTC)
	olderCreatedAt := tiedCreatedAt.Add(-time.Millisecond)
	models := []*schema.ResourcePool{
		resourcePoolModel(9_300_101, organizationID, olderCreatedAt),
		resourcePoolModel(9_300_102, organizationID, tiedCreatedAt),
		resourcePoolModel(9_300_103, organizationID, tiedCreatedAt),
		resourcePoolModel(9_300_104, organizationID, tiedCreatedAt),
		resourcePoolModel(9_300_105, organizationID, tiedCreatedAt),
	}
	models[3].Name = "raw-name-does-not-match-filter"
	models[len(models)-1].LifecycleState = globalngv1.ResourcePool_FAILED.String()
	for _, model := range models {
		require.NoError(t, integrationDBProvider.WriteDB(ctx).Create(model).Error)
	}
	t.Cleanup(func() {
		require.NoError(t, integrationDBProvider.WriteDB(context.Background()).Unscoped().
			Where("owner_id = ?", organizationID).
			Delete(&schema.ResourcePool{}).Error)
	})

	filter := &resourcepoolrepo.Filter{
		OwnerType:       int(globalngv1.ResourcePool_ORGANIZATION),
		OwnerIds:        []uint64{organizationID},
		ServicePlans:    []globalcommonv1.ServicePlan{globalcommonv1.ServicePlan_BYOC},
		LifecycleStates: []globalngv1.ResourcePool_LifecycleState{globalngv1.ResourcePool_ACTIVE},
		PageSize:        2,
	}
	firstPage, total, err := repo.Query(ctx, filter)
	require.NoError(t, err)
	require.Equal(t, int64(4), total)
	require.Equal(t, []uint64{9_300_104, 9_300_103}, resourcePoolIDs(firstPage))

	filter.Cursor = &resourcepoolrepo.Cursor{
		CreatedAt: firstPage[len(firstPage)-1].GetCreateTime().AsTime(),
		ID:        firstPage[len(firstPage)-1].GetId(),
	}
	secondPage, total, err := repo.Query(ctx, filter)
	require.NoError(t, err)
	require.Equal(t, int64(4), total, "total must be counted before the cursor predicate")
	require.Equal(t, []uint64{9_300_102, 9_300_101}, resourcePoolIDs(secondPage))

	nameFiltered, total, err := repo.Query(ctx, &resourcepoolrepo.Filter{
		OwnerType:       int(globalngv1.ResourcePool_ORGANIZATION),
		OwnerIds:        []uint64{organizationID},
		LifecycleStates: []globalngv1.ResourcePool_LifecycleState{globalngv1.ResourcePool_ACTIVE},
		Name:            "resource-pool-9300104",
		PageSize:        10,
	})
	require.NoError(t, err)
	require.Equal(t, int64(1), total)
	require.Equal(t, []uint64{9_300_104}, resourcePoolIDs(nameFiltered))
}

func TestResourcePoolRepo_ListLiveBYOCOrganizationResourcePoolsInRegionIncludesEveryLifecycle(t *testing.T) {
	ctx := context.Background()
	repo := NewResourcePoolRepo(integrationDBProvider)
	const organizationID = uint64(9_100_004)

	states := []string{
		"",
		globalngv1.ResourcePool_CREATING.String(),
		globalngv1.ResourcePool_ACTIVE.String(),
		globalngv1.ResourcePool_MODIFYING.String(),
		globalngv1.ResourcePool_FAILED.String(),
	}
	wantIDs := make([]uint64, 0, len(states))
	for index, state := range states {
		id := uint64(9_400_100 + index)
		model := resourcePoolModel(id, organizationID, time.Now().UTC())
		model.LifecycleState = state
		model.Cidr = fmt.Sprintf("10.%d.0.0/16", index+1)
		if index == len(states)-1 {
			model.RegionId = 2
		}
		require.NoError(t, integrationDBProvider.WriteDB(ctx).Create(model).Error)
		wantIDs = append(wantIDs, id)
	}
	deleted := resourcePoolModel(9_400_200, organizationID, time.Now().UTC())
	require.NoError(t, integrationDBProvider.WriteDB(ctx).Create(deleted).Error)
	require.NoError(t, integrationDBProvider.WriteDB(ctx).Delete(deleted).Error)
	premium := resourcePoolModel(9_400_201, organizationID, time.Now().UTC())
	premium.ServicePlan = globalcommonv1.ServicePlan_Premium.String()
	require.NoError(t, integrationDBProvider.WriteDB(ctx).Create(premium).Error)
	differentRegion := resourcePoolModel(9_400_202, organizationID, time.Now().UTC())
	differentRegion.RegionId = 4
	require.NoError(t, integrationDBProvider.WriteDB(ctx).Create(differentRegion).Error)
	t.Cleanup(func() {
		require.NoError(t, integrationDBProvider.WriteDB(context.Background()).Unscoped().
			Where("owner_id = ?", organizationID).
			Delete(&schema.ResourcePool{}).Error)
	})

	pools, err := repo.ListLiveBYOCOrganizationResourcePoolsInRegion(
		ctx,
		organizationID,
		globalcommonv1.CloudProvider_AWS,
		globalcommonv1.Region_AWS_US_WEST_2,
	)
	require.NoError(t, err)
	require.Equal(t, wantIDs, resourcePoolIDs(pools))
	for _, pool := range pools {
		require.Equal(t, globalcommonv1.CloudProvider_AWS, pool.GetProviderId())
		require.NotEqual(t, globalcommonv1.Region_NAME_UNSPECIFIED, pool.GetRegionId())
	}
}

func TestResourcePoolRepo_CompareAndSetLifecycleRejectsStaleSnapshot(t *testing.T) {
	ctx := context.Background()
	repo := NewResourcePoolRepo(integrationDBProvider)
	pool := resourcePoolModel(9_300_201, 9_100_004, time.Now().UTC())
	pool.StateMessageCode = "CREATE_PENDING"
	require.NoError(t, integrationDBProvider.WriteDB(ctx).Create(pool).Error)
	t.Cleanup(func() {
		require.NoError(t, integrationDBProvider.WriteDB(context.Background()).Unscoped().Delete(pool).Error)
	})

	loaded, err := repo.Load(ctx, pool.ID, false)
	require.NoError(t, err)
	loadedAt := loaded.GetUpdateTime().AsTime()

	updated, err := repo.CompareAndSetLifecycle(
		ctx,
		pool.ID,
		globalngv1.ResourcePool_ACTIVE,
		"CREATE_PENDING",
		loadedAt,
		globalngv1.ResourcePool_MODIFYING,
		"",
	)
	require.NoError(t, err)
	require.True(t, updated)

	current, err := repo.Load(ctx, pool.ID, false)
	require.NoError(t, err)
	require.Equal(t, globalngv1.ResourcePool_MODIFYING, current.GetLifecycleState())
	require.Empty(t, current.GetStateMessageCode())

	updated, err = repo.CompareAndSetLifecycle(
		ctx,
		pool.ID,
		globalngv1.ResourcePool_ACTIVE,
		"CREATE_PENDING",
		loadedAt,
		globalngv1.ResourcePool_FAILED,
		"",
	)
	require.NoError(t, err)
	require.False(t, updated)

	current, err = repo.Load(ctx, pool.ID, false)
	require.NoError(t, err)
	require.Equal(t, globalngv1.ResourcePool_MODIFYING, current.GetLifecycleState())
	require.Empty(t, current.GetStateMessageCode())
}

func TestResourcePoolRepo_FieldScopedUpdateDistinguishesNoChangeFromMissing(t *testing.T) {
	ctx := context.Background()
	repo := NewResourcePoolRepo(integrationDBProvider)
	pool := resourcePoolModel(9_300_202, 9_100_004, time.Now().UTC())
	require.NoError(t, integrationDBProvider.WriteDB(ctx).Create(pool).Error)
	t.Cleanup(func() {
		require.NoError(t, integrationDBProvider.WriteDB(context.Background()).Unscoped().Delete(pool).Error)
	})

	require.NoError(t, repo.UpdateInstanceLimit(ctx, pool.ID, pool.InstanceLimit),
		"a same-value update must be treated as an idempotent success")
	require.NoError(t, repo.Delete(ctx, pool.ID))

	err := repo.UpdateInstanceLimit(ctx, pool.ID, pool.InstanceLimit)
	require.Error(t, err)
	require.Equal(t, bizerr.ErrCodeRecordNotFound, err.ErrCode())
}

func TestResourcePoolRepo_BackfillMetadataIsFillOnlyAndIdempotent(t *testing.T) {
	ctx := context.Background()
	repo := NewResourcePoolRepo(integrationDBProvider)
	pool := resourcePoolModel(9_300_203, 9_100_005, time.Now().UTC())
	pool.NormalizedName = ""
	pool.CidrStartIPv4 = nil
	pool.CidrEndIPv4 = nil
	pool.LifecycleState = ""
	pool.VcpusLimit = 16
	pool.VcpuLimitMode = globalngv1.ResourcePool_INHERITED.String()
	pool.CloudResourceTags = nil
	require.NoError(t, integrationDBProvider.WriteDB(ctx).Create(pool).Error)
	t.Cleanup(func() {
		require.NoError(t, integrationDBProvider.WriteDB(context.Background()).Unscoped().Delete(pool).Error)
	})

	lifecycle := globalngv1.ResourcePool_ACTIVE
	vcpuMode := globalngv1.ResourcePool_LIMITED
	normalizedName := "backfilled-pool"
	first, err := repo.BackfillMetadata(ctx, &resourcepoolrepo.MetadataBackfillPatch{
		ResourcePoolID:    pool.ID,
		NormalizedName:    &normalizedName,
		CIDRStartIPv4:     []byte{10, 0, 0, 0},
		CIDREndIPv4:       []byte{10, 0, 0, 255},
		LifecycleState:    &lifecycle,
		VcpuLimitMode:     &vcpuMode,
		CloudResourceTags: map[string]string{"team": "database"},
		BackfillTags:      true,
	})
	require.NoError(t, err)
	require.Equal(t, resourcepoolrepo.MetadataBackfillResult{
		NormalizedName:    true,
		CIDR:              true,
		LifecycleState:    true,
		VcpuLimitMode:     true,
		CloudResourceTags: true,
	}, first)

	var stored schema.ResourcePool
	require.NoError(t, integrationDBProvider.ReadDB(ctx).Where("id = ?", pool.ID).Take(&stored).Error)
	require.Equal(t, normalizedName, stored.NormalizedName)
	require.Equal(t, []byte{10, 0, 0, 0}, stored.CidrStartIPv4)
	require.Equal(t, []byte{10, 0, 0, 255}, stored.CidrEndIPv4)
	require.Equal(t, lifecycle.String(), stored.LifecycleState)
	require.Equal(t, vcpuMode.String(), stored.VcpuLimitMode)
	require.Equal(t, map[string]string{"team": "database"}, stored.CloudResourceTags)

	differentName := "must-not-overwrite"
	differentLifecycle := globalngv1.ResourcePool_FAILED
	differentMode := globalngv1.ResourcePool_UNLIMITED
	second, err := repo.BackfillMetadata(ctx, &resourcepoolrepo.MetadataBackfillPatch{
		ResourcePoolID:    pool.ID,
		NormalizedName:    &differentName,
		CIDRStartIPv4:     []byte{192, 0, 2, 0},
		CIDREndIPv4:       []byte{192, 0, 2, 255},
		LifecycleState:    &differentLifecycle,
		VcpuLimitMode:     &differentMode,
		CloudResourceTags: map[string]string{"team": "other"},
		BackfillTags:      true,
	})
	require.NoError(t, err)
	require.Equal(t, resourcepoolrepo.MetadataBackfillResult{}, second)

	stored = schema.ResourcePool{}
	require.NoError(t, integrationDBProvider.ReadDB(ctx).Where("id = ?", pool.ID).Take(&stored).Error)
	require.Equal(t, normalizedName, stored.NormalizedName)
	require.Equal(t, []byte{10, 0, 0, 0}, stored.CidrStartIPv4)
	require.Equal(t, []byte{10, 0, 0, 255}, stored.CidrEndIPv4)
	require.Equal(t, lifecycle.String(), stored.LifecycleState)
	require.Equal(t, vcpuMode.String(), stored.VcpuLimitMode)
	require.Equal(t, map[string]string{"team": "database"}, stored.CloudResourceTags)
}

func resourcePoolModel(id, organizationID uint64, createdAt time.Time) *schema.ResourcePool {
	return &schema.ResourcePool{
		ID:             id,
		CreatedAt:      createdAt,
		UpdatedAt:      createdAt,
		Name:           fmt.Sprintf("resource-pool-%d", id),
		NormalizedName: fmt.Sprintf("resource-pool-%d", id),
		OwnerType:      int(globalngv1.ResourcePool_ORGANIZATION),
		OwnerId:        organizationID,
		Cidr:           "10.0.0.0/24",
		CidrStartIPv4:  []byte{10, 0, 0, 0},
		CidrEndIPv4:    []byte{10, 0, 0, 255},
		ServicePlan:    globalcommonv1.ServicePlan_BYOC.String(),
		InstanceLimit:  50,
		LifecycleState: globalngv1.ResourcePool_ACTIVE.String(),
		VcpuLimitMode:  globalngv1.ResourcePool_INHERITED.String(),
		ProviderId:     1,
		RegionId:       1,
		DataplaneId:    "",
		Provider:       nil,
		Region:         nil,
		DeletedAt:      gorm.DeletedAt{},
	}
}

func resourcePoolIDs(resourcePools []*entity.ResourcePool) []uint64 {
	ids := make([]uint64, 0, len(resourcePools))
	for _, resourcePool := range resourcePools {
		ids = append(ids, resourcePool.GetId())
	}
	return ids
}