package encryptionkey

import (
	"context"
	"strconv"
	"sync/atomic"
	"testing"

	"github.com/tidbcloud/fastflow/pkg/entity/run"
	"github.com/tidbcloud/fastflow/pkg/utils"
	pkgv2db "github.com/tidbcloud/pkgv2/db"
	"github.com/tidbcloud/pkgv2/gormutil"
	"go.uber.org/mock/gomock"

	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/adapter/metrics/mock_metrics"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/adapter/regionalserver/mock_regionalserver"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/entity"
	clusterrepo "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/cluster"
	bindingrepo "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/clusterencryptionbinding"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/encryptionkey"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/region"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/resourcepool/mock_resourcepool"
	regionsvc "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/service/region"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/service/workflow/common"
	clusterrepoimpl "github.com/tidbcloud/tidb-management-service/nextgen-global-service/infra/repoimpl/cluster"
	bindingrepoimpl "github.com/tidbcloud/tidb-management-service/nextgen-global-service/infra/repoimpl/clusterencryptionbinding"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/infra/repoimpl/db"
	encryptionkeyrepoimpl "github.com/tidbcloud/tidb-management-service/nextgen-global-service/infra/repoimpl/encryptionkey"
	migrate "github.com/tidbcloud/tidb-management-service/nextgen-global-service/infra/repoimpl/migration"
	regionrepoimpl "github.com/tidbcloud/tidb-management-service/nextgen-global-service/infra/repoimpl/region"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/infra/repoimpl/schema"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/pkg/config"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/pkg/testdata"
	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
	"github.com/tidbcloud/tidb-management-service/utilities/idgenerator"
	mockworkflow "github.com/tidbcloud/tidb-management-service/utilities/workflow/mock_workflow"
)

var (
	dbProvider      atomic.Pointer[pkgv2db.TransProvider]
	dbTransactioner atomic.Pointer[pkgv2db.Transactioner]
)

const testOrgID = testdata.TestOrgID

// testService holds the service under test and its dependencies.
type testService struct {
	svc  Service
	ctrl *gomock.Controller

	// real repos (backed by testcontainer MySQL)
	regionRepo           region.Repo
	providerIdentityRepo region.ProviderIdentityRepo
	regionService        regionsvc.Service
	encryptionKeyRepo    encryptionkey.Repo
	clusterRepo          clusterrepo.Repo
	bindingRepo          bindingrepo.Repo

	// mocks
	mockRegionalCli      *mock_regionalserver.MockClient
	mockWorkflow         *mockworkflow.MockWorkflow
	resourcePoolResolver *fakeDefaultResourcePoolResolver
}

func newTestService(t *testing.T) *testService {
	t.Helper()
	ctrl := gomock.NewController(t)

	pvd := *dbProvider.Load()
	trans := *dbTransactioner.Load()

	// Real repos backed by the testcontainer DB
	regionRepo := regionrepoimpl.NewRegionRepo(pvd)
	providerIdentityRepo := regionrepoimpl.NewRegionProviderIdentityRepo(pvd)
	clusterRepo := clusterrepoimpl.NewClusterRepo(pvd)
	ekRepo := encryptionkeyrepoimpl.NewEncryptionKeyRepo(pvd)
	bindingRepo := bindingrepoimpl.NewClusterEncryptionBindingRepo(pvd)
	// Mocks for external adapters
	mockCli := mock_regionalserver.NewMockClient(ctrl)
	mockWf := mockworkflow.NewMockWorkflow(ctrl)
	mockRPRepo := mock_resourcepool.NewMockRepo(ctrl)
	mockMetrics := mock_metrics.NewMockClient(ctrl)
	resourcePoolResolver := &fakeDefaultResourcePoolResolver{}
	sfid := idgenerator.NewSonyflakeID(1, "")

	// Build region service with real repos + mocked external deps
	regionService := regionsvc.NewService(
		regionRepo,
		clusterRepo,
		providerIdentityRepo,
		mockRPRepo,
		nil,
		trans,
		mockCli,
		sfid,
		mockMetrics,
		nil,
	)

	svc := NewService(
		regionRepo,
		providerIdentityRepo,
		regionService,
		ekRepo,
		clusterRepo,
		bindingRepo,
		trans,
		mockCli,
		mockWf,
		resourcePoolResolver,
		sfid,
	)

	return &testService{
		svc:                  svc,
		ctrl:                 ctrl,
		regionRepo:           regionRepo,
		providerIdentityRepo: providerIdentityRepo,
		regionService:        regionService,
		encryptionKeyRepo:    ekRepo,
		clusterRepo:          clusterRepo,
		bindingRepo:          bindingRepo,
		mockRegionalCli:      mockCli,
		mockWorkflow:         mockWf,
		resourcePoolResolver: resourcePoolResolver,
	}
}

type fakeDefaultResourcePoolResolver struct {
	calls atomic.Int32
}

func (r *fakeDefaultResourcePoolResolver) GetOrCreateDefaultResourcePool(
	_ context.Context,
	_ entity.ResourcePool,
	_ uint32,
) (*entity.ResourcePool, bizerr.BizError) {
	r.calls.Add(1)
	return &entity.ResourcePool{
		ResourcePool: &globalngv1.ResourcePool{
			Id: 1,
		},
	}, nil
}

// cleanTestData removes test-created rows (clusters, encryption keys, operations)
// while leaving seed data (providers, regions, provider identities) intact.
func cleanTestData(ctx context.Context, t *testing.T) {
	t.Helper()
	wdb := (*dbProvider.Load()).WriteDB(ctx)
	for _, table := range []string{"cluster_encryption_bindings", "clusters", "encryption_keys"} {
		if err := wdb.Exec("DELETE FROM " + table).Error; err != nil {
			t.Fatalf("cleanTestData(%s): %v", table, err)
		}
	}
}

// createTestCluster inserts a cluster record and returns it.
func createTestCluster(
	ctx context.Context,
	t *testing.T,
	repo clusterrepo.Repo,
	sfid idgenerator.IDGenerator,
	opts ...func(*entity.Cluster),
) *entity.Cluster {
	t.Helper()
	cluster := &entity.Cluster{
		Cluster: &globalngv1.Cluster{
			ClusterId:       sfid.MustNew(),
			OrganizationId:  testOrgID,
			State:           globalngv1.Cluster_AVAILABLE,
			SharedPoolId:    "1",
			ServicePlan:     globalcommonv1.ServicePlan_Premium,
			CloudProviderId: globalcommonv1.CloudProvider_AWS,
			RegionId:        globalcommonv1.Region_AWS_US_WEST_2,
			DisplayName:     "test-cluster",
		},
		EncryptionKeyId: 0,
		RegionId:        1,
		CloudProviderId: 1,
	}
	for _, o := range opts {
		o(cluster)
	}
	if err := repo.Create(ctx, cluster); err != nil {
		t.Fatalf("createTestCluster: %v", err)
	}
	return cluster
}

// createTestEncryptionKey inserts an encryption key record and returns it.
func createTestEncryptionKey(
	ctx context.Context,
	t *testing.T,
	repo encryptionkey.Repo,
	sfid idgenerator.IDGenerator,
	opts ...func(*entity.EncryptionKey),
) *entity.EncryptionKey {
	t.Helper()
	key := &entity.EncryptionKey{
		Id:                 sfid.MustNew(),
		TenantId:           testOrgID,
		ProviderName:       globalcommonv1.CloudProvider_AWS,
		RegionName:         globalcommonv1.Region_AWS_US_WEST_2,
		ServicePlan:        globalcommonv1.ServicePlan_Premium,
		ProviderIdentityId: 1,
		CustomerManaged:    true,
		MasterKey:          "arn:aws:kms:us-west-2:123456789012:key/test-key-id",
		InitialDataplaneId: "dataplane-1",
		ProviderId:         1,
		RegionId:           1,
	}
	for _, o := range opts {
		o(key)
	}
	if err := repo.Create(ctx, key); err != nil {
		t.Fatalf("createTestEncryptionKey: %v", err)
	}
	return key
}

func TestMain(m *testing.M) {
	var (
		ctx = context.Background()
		c   = &config.Config{
			MysqlConfig: gormutil.Config{
				User:     "root",
				Database: "local_test",
				Password: "password",
				LogLevel: "warn",
			},
		}
	)

	mysqlContainer, host, port, err := testdata.StartMysqlContainer(
		ctx,
		c.MysqlConfig.Database,
		c.MysqlConfig.User,
		c.MysqlConfig.Password,
	)
	if err != nil {
		panic(err)
	}
	defer func() {
		if err := mysqlContainer.Terminate(ctx); err != nil {
			panic(err)
		}
	}()

	c.MysqlConfig.Host = host
	c.MysqlConfig.Port = port

	// Create schemas.
	_, pvd, trans := db.NewDBProvider(db.NewDB(c))
	if err := migrate.RunDBMigrate(pvd.WriteDB(ctx)); err != nil {
		panic(err)
	}

	dbProvider.Store(&pvd)
	dbTransactioner.Store(&trans)

	if err := seedBaseData(ctx, pvd); err != nil {
		panic(err)
	}

	m.Run()
}

func seedBaseData(ctx context.Context, d pkgv2db.Provider) error {
	providers := testdata.GetProviderRecords()
	for i := range providers {
		if err := d.WriteDB(ctx).Create(providers[i]).Error; err != nil {
			return err
		}
	}

	regions := testdata.GetRegionRecords()
	for i := range regions {
		regions[i].Status = "active"
		if err := d.WriteDB(ctx).Create(regions[i]).Error; err != nil {
			return err
		}
	}

	regionTenants := testdata.GetRegionTenantRecords()
	for i := range regionTenants {
		if err := d.WriteDB(ctx).Create(regionTenants[i]).Error; err != nil {
			return err
		}
	}

	regionServicePlans := testdata.GetRegionServicePlanRecords()
	for i := range regionServicePlans {
		if err := d.WriteDB(ctx).Create(regionServicePlans[i]).Error; err != nil {
			return err
		}
	}

	// Seed provider identities for AWS us-west-2 regions (IDs 1, 2, 3).
	// SelectRegion may return any of these depending on priority + tenant rules.
	for _, rid := range []uint64{1, 2, 3} {
		pi := &schema.ProviderIdentity{
			ID:       rid, // use region id as identity id for convenience
			RegionId: rid,
			Identity: "123456789012",
		}
		if err := d.WriteDB(ctx).Create(pi).Error; err != nil {
			return err
		}
	}

	// Seed default_provider_identity (catch-all: RegionId=0, TenantId=0)
	// so PickRegionProviderIdentity works for any AWS Premium region.
	defaultPI := &schema.DefaultProviderIdentity{
		ProviderId:  1,
		ServicePlan: "Premium",
		RegionId:    0,
		TenantId:    0,
		Identity:    "123456789012",
	}
	if err := d.WriteDB(ctx).Create(defaultPI).Error; err != nil {
		return err
	}

	// Seed a resource pool for cluster tests.
	rp := &schema.ResourcePool{
		ID:          1,
		Name:        "test-resource-pool",
		OwnerType:   1,
		OwnerId:     testOrgID,
		ProviderId:  1,
		RegionId:    1,
		DataplaneId: "dataplane-1",
	}
	if err := d.WriteDB(ctx).Create(rp).Error; err != nil {
		return err
	}

	return nil
}

// newTestExecuteContext returns a run.ExecuteContext backed by the given vars map.
func newTestExecuteContext(ctx context.Context, vars map[string]string) run.ExecuteContext {
	getter := func(key string) (string, bool) {
		v, ok := vars[key]
		return v, ok
	}
	shareData := &mapShareData{data: make(map[string]string)}
	noopTrace := func(_ string, _ ...run.TraceOp) {}
	iter := utils.KeyValueIterator(func(fn utils.KeyValueIterateFunc) {
		for k, v := range vars {
			if fn(k, v) {
				return
			}
		}
	})
	return run.NewDefExecuteContext(ctx, shareData, noopTrace, getter, iter)
}

type mapShareData struct {
	data map[string]string
}

func (m *mapShareData) Get(key string) (string, bool) {
	v, ok := m.data[key]
	return v, ok
}

func (m *mapShareData) Set(key, val string) {
	m.data[key] = val
}

// createTestClusterEncryptionBinding inserts a binding.
func createTestClusterEncryptionBinding(
	ctx context.Context,
	t *testing.T,
	repo bindingrepo.Repo,
	clusterId, encryptionKeyId uint64,
	state entity.ClusterEncryptionBindingState, //nolint:unparam // Test method by designed.
) {
	t.Helper()
	b := &entity.ClusterEncryptionBinding{
		ClusterId:       clusterId,
		TenantId:        testOrgID,
		EncryptionKeyId: encryptionKeyId,
		State:           state,
		Message:         "test binding",
	}
	if err := repo.Upsert(ctx, b); err != nil {
		t.Fatalf("createTestClusterEncryptionBinding: %v", err)
	}
}

// workflowVars builds the standard vars map that all encryption-change actions expect.
func workflowVars(clusterId, encryptionKeyId uint64, mode, transitionType string) map[string]string {
	return map[string]string{
		common.KeyClusterId:       strconv.FormatUint(clusterId, 10),
		common.KeyEncryptionKeyId: strconv.FormatUint(encryptionKeyId, 10),
		common.KeyEncryptionMode:  mode,
		common.KeyTransitionType:  transitionType,
	}
}