package agent

import (
	"context"
	"encoding/json"
	"fmt"
	"sync"
	"testing"

	"github.com/pingcap/tiflow/dm/openapi"
	agentpb "github.com/tidbcloud/dataflow-service/idl/pbgen/proto/agent"
	"github.com/tidbcloud/dataflow-service/internal/config"
	"github.com/tidbcloud/dataflow-service/internal/grpc/grpcli"
	"github.com/tidbcloud/dataflow-service/internal/utils"
	infraapi "github.com/tidbcloud/dataflow-service/pkg/infra-api"
)

type testConfig struct {
	ctx              context.Context
	dmAddr           string
	clusterID        uint64
	dmOpenapiClient  *DMOpenapiClient
	Source           openapi.Source
	BinlogName       string
	BinlogPos        int
	DataDir          string
	ImportMode       openapi.TaskFullMigrateConfImportMode
	PdAddr           string
	TableMigrateRule []openapi.TaskTableMigrateRule
	TargetDB         *openapi.TaskTargetDataBase
	LoadTargetDB     *openapi.TaskTargetDataBase
	TaskName         string
	TaskModel        openapi.TaskTaskMode
	PDSecurity       *openapi.Security
}

var testCfg *testConfig
var testOnce sync.Once

func initConfig(t *testing.T) *testConfig {
	testOnce.Do(func() {
		if err := config.InitConfig("/Users/jiafangdong/backup/dataflow-service.yaml"); err != nil {
			t.Fatalf("config init failed:%v", err)
			return
		}

		ctx := context.TODO()
		// dmAddr := "http://fangdong-test-001-dm-master.dm-test.svc.cluster.local:8261"
		dmAddr := "http://localhost:8261"
		// dev
		// projectID := uint64(1369847559694037567)
		// region := "eastus"
		// staging
		projectID := uint64(1369847559694040687)
		clusterID := uint64(10731894235153974141)
		region := "eastus2"
		provider := "azure"
		sourceName := "source-1223"
		sourceHost := "123.57.73.55"
		sourcePort := 3306
		sourceUser := "dm"
		sourcePassword := "dataflow"
		sourceSchema := "dmtest"
		sourceTable := "user_001"
		targetUser := "root"
		targetPassword := "zkwC4OAVGJcPRyUi"
		cloudAdminPassword := "Ckz9iHYzcAu8CdEW"

		cfg := &testConfig{
			ctx:       context.TODO(),
			dmAddr:    dmAddr,
			clusterID: clusterID,
			Source: openapi.Source{
				Enable:     true,
				Host:       sourceHost,
				Port:       sourcePort,
				User:       sourceUser,
				Password:   &sourcePassword,
				SourceName: sourceName,
			},
			TaskName:  "task-dump",
			TaskModel: openapi.TaskTaskModeDump,
			// TaskName:  "task-load",
			// TaskModel: openapi.TaskTaskModeLoad,
			// TaskName:   "task-sync",
			// TaskModel:  openapi.TaskTaskModeIncremental,
			ImportMode: openapi.TaskFullMigrateConfImportModePhysical,
			PdAddr:     fmt.Sprintf("db-pd.tidb%d.svc:2379", clusterID),
		}

		cluster, cerr := grpcli.GetClusterWithDataplaneByID(ctx, clusterID)
		if cerr != nil {
			t.Fatalf("failed to get cluster info:%v", cerr)
			return
		}
		fmt.Println(cluster)

		agentClient, err := NewRegionalAgentClient(ctx, clusterID, provider, region)
		if err != nil {
			t.Fatalf("failed to get regional agent client:%v", err)
			return
		}
		cfg.dmOpenapiClient = agentClient.GetDMOpenapiClient(clusterID, dmAddr)

		tcNameGetter := infraapi.ClusterNameGetter{
			InfraProvider:        provider,
			InfraAPIObjNamespace: infraapi.GetObjectNamespace(projectID),
			ClusterName:          infraapi.ClusterName(clusterID),
		}

		tidbServerCertSecret, err := agentClient.GetTiDBServerSecret(ctx, &agentpb.GetTiDBServerSecretReq{
			ClusterNamePrefix: tcNameGetter.ClusterServicePrefix(),
			ClusterNamespace:  tcNameGetter.ClusterNamespace(),
			ProviderType:      provider,
		})
		if err != nil {
			t.Fatalf("failed to get tidb cluster server secret:%v", err)
			return
		}

		clientCertSecret, err := agentClient.GetClusterClientSecret(ctx, &agentpb.GetClusterClientSecretReq{
			ClusterNamePrefix: tcNameGetter.ClusterServicePrefix(),
			ClusterNamespace:  tcNameGetter.ClusterNamespace(),
			ProviderType:      provider,
			RequestId:         utils.UniqueID(),
		})
		if err != nil {
			t.Fatalf("failed to get tidb cluster server secret:%v", err)
			return
		}
		fmt.Println("tidbServerCertSecret.ca", tidbServerCertSecret.SslCa)
		fmt.Println("tidbServerCertSecret.cert", tidbServerCertSecret.SslCert)
		fmt.Println("tidbServerCertSecret.key", tidbServerCertSecret.SslKey)
		fmt.Println("clientCertSecret.ca", clientCertSecret.SslCa)
		fmt.Println("clientCertSecret.cert", clientCertSecret.SslCert)
		fmt.Println("clientCertSecret.key", clientCertSecret.SslKey)

		cfg.PDSecurity = &openapi.Security{
			SslCaContent:   clientCertSecret.SslCa,
			SslCertContent: clientCertSecret.SslCert,
			SslKeyContent:  clientCertSecret.SslKey,
			// SslCa:          "/root/tls/client.ca",
			// SslCert:        "/root/tls/client.cert",
			// SslKey:         "/root/tls/client.key",
		}

		tableMigrateRule := openapi.TaskTableMigrateRule{}
		tableMigrateRule.Source.Schema = sourceSchema
		tableMigrateRule.Source.Table = sourceTable
		tableMigrateRule.Source.SourceName = sourceName
		tableMigrateRule.Target = &openapi.TaskTableMigrateRuleTarget{
			Schema: utils.ToPtr(sourceSchema),
			Table:  utils.ToPtr(sourceTable),
		}
		cfg.TableMigrateRule = append(cfg.TableMigrateRule, tableMigrateRule)

		cfg.TargetDB = &openapi.TaskTargetDataBase{
			Host: tcNameGetter.TiDBNamespacedServiceName(),
			// user can self-define port of tidb, but port of db-tidb(k8s service) is always 4000
			Port:     4000,
			User:     targetUser,
			Password: targetPassword,
			Security: &openapi.Security{
				SslCaContent:   tidbServerCertSecret.SslCa,
				SslCertContent: tidbServerCertSecret.SslCert,
				SslKeyContent:  tidbServerCertSecret.SslKey,
				// SslCa:          "/root/tls/server.ca",
				// SslCert:        "/root/tls/server.cert",
				// SslKey:         "/root/tls/server.key",
			},
		}

		// confidentialInfo, err := grpcli.GetClusterConfidentialInfoByID(ctx, clusterID)
		// if err != nil {
		// 	t.Fatalf("get cluster's confidential information by ID error:%v", err)
		// 	return
		// }
		// cloudAdminPassword := confidentialInfo.GetEncryptedCloudAdminPasswd()
		// encryptor, err := secretprovider.NewSecretEncryptor(config.Global().InfraConfig.CMKArn)
		// if err != nil {
		// 	t.Fatalf("NewSecretEncryptor failed:%v", err)
		// 	return
		// }
		// secret, err := encrypt.DecryptBase64(encryptor, cloudAdminPassword)
		// if err != nil {
		// 	t.Fatalf("encrypt.DecryptBase64 failed:%v", err)
		// 	return
		// }
		// cloudAdminPassword = string(secret)
		cfg.LoadTargetDB = &openapi.TaskTargetDataBase{
			Host: tcNameGetter.TiDBNamespacedServiceName(),
			// user can self-define port of tidb, but port of db-tidb(k8s service) is always 4000
			Port:     4000,
			User:     "cloud_admin",
			Password: cloudAdminPassword,
			Security: &openapi.Security{
				SslCaContent:   tidbServerCertSecret.SslCa,
				SslCertContent: tidbServerCertSecret.SslCert,
				SslKeyContent:  tidbServerCertSecret.SslKey,
				// SslCa:          "/root/tls/server.ca",
				// SslCert:        "/root/tls/server.cert",
				// SslKey:         "/root/tls/server.key",
			},
		}

		cfg.DataDir = "azure://dm-1369847559694040687-eastus2-1df90609/dmjob-002?account-name=dm1369847559694040687&sas-token=sv%3D2022-11-02%26ss%3Db%26srt%3Dco%26sp%3Drwdlacx%26se%3D2025-04-15T17%3A23%3A22Z%26st%3D2024-12-23T09%3A23%3A22Z%26spr%3Dhttps%26sig%3DzRX7v8c0dhGXEtkJBetH4tWBPbECxhOSmQZhBMK48%252FQ%253D"
		// cfg.DataDir = "azure://dm-1369847559694040687-eastus2-7dccf2c7/dmjob-001?account-name=dm1369847559694040687&account-key=5EH/RBNtbfBXjUb7EfD4GGzPzEf6seThUZeEG3HgiJEB8qGaV+Y4+lfEUQiTpVjWi585H2p1jqF8+AStyFa6VA=="
		// cfg.DataDir = "/tmp/dmjob1220"
		testCfg = cfg
	})
	if testCfg == nil {
		t.Fatal("failed to get init testConfig")
	}
	return testCfg
}

func TestDMOpenapiClient_CreateDataSource(t *testing.T) {

	type args struct {
		req *openapi.CreateSourceRequest
	}

	cfg := initConfig(t)
	source := cfg.Source
	tests := []struct {
		name    string
		args    args
		want    *openapi.Source
		wantErr bool
	}{
		{
			name: "OK",
			args: args{
				req: &openapi.CreateSourceRequest{
					Source: source,
				},
			},
			want: &openapi.Source{
				SourceName: source.SourceName,
			},
			wantErr: false,
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			got, err := cfg.dmOpenapiClient.CreateDataSource(cfg.ctx, tt.args.req)
			if (err != nil) != tt.wantErr {
				t.Errorf("DMOpenapiClient.CreateDataSource() error = %v, wantErr %v", err, tt.wantErr)
				return
			}
			if got.SourceName != tt.want.SourceName {
				t.Errorf("DMOpenapiClient.CreateDataSource() = %v, want %v", got, tt.want)
			}
		})
	}
}

func TestDMOpenapiClient_GetDataSource(t *testing.T) {

	type args struct {
		name string
	}

	cfg := initConfig(t)
	tests := []struct {
		name    string
		args    args
		want    *openapi.Source
		wantErr bool
	}{
		{
			name: "OK",
			args: args{
				name: cfg.Source.SourceName,
			},
			want: &openapi.Source{
				SourceName: cfg.Source.SourceName,
			},
			wantErr: false,
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			got, err := cfg.dmOpenapiClient.GetDataSource(cfg.ctx, tt.args.name)
			if (err != nil) != tt.wantErr {
				t.Errorf("DMOpenapiClient.GetDataSource() error = %v, wantErr %v", err, tt.wantErr)
				return
			}
			if got.SourceName != tt.want.SourceName {
				t.Errorf("DMOpenapiClient.GetDataSource() = %v, want %v", got, tt.want)
			}
		})
	}
}

func TestDMOpenapiClient_DeleteDataSource(t *testing.T) {

	type args struct {
		name string
	}

	cfg := initConfig(t)
	source := cfg.Source

	for i := 0; i < 10; i++ {
		source.SourceName = fmt.Sprintf("source-0%d", i)
		tests := []struct {
			name    string
			args    args
			wantErr bool
		}{
			{
				name: "OK",
				args: args{
					name: source.SourceName,
				},
				wantErr: false,
			},
		}
		for _, tt := range tests {
			t.Run(tt.name, func(t *testing.T) {
				if err := cfg.dmOpenapiClient.DeleteDataSource(cfg.ctx, tt.args.name); (err != nil) != tt.wantErr {
					t.Errorf("DMOpenapiClient.DeleteDataSource() error = %v, wantErr %v", err, tt.wantErr)
				}
			})
		}
	}
}

func TestDMOpenapiClient_CreateTask(t *testing.T) {

	type args struct {
		req *openapi.CreateTaskRequest
	}

	cfg := initConfig(t)
	targetConfig := cfg.TargetDB
	if cfg.TaskName == "task-load" {
		targetConfig = cfg.LoadTargetDB
	}
	taskName := cfg.TaskName
	taskShardMode := openapi.TaskShardMode("pessimistic")
	tests := []struct {
		name    string
		args    args
		want    *openapi.OperateTaskResponse
		wantErr bool
	}{
		{
			name: "OK",
			args: args{
				req: &openapi.CreateTaskRequest{
					Task: openapi.Task{
						Name: taskName,
						// whether to enable support for the online ddl plugin
						EnhanceOnlineSchemaChange: false,
						// ignore precheck items
						IgnoreCheckingItems: &[]string{"all"},
						// downstream database for storing meta information
						// TODO whether need this field
						MetaSchema: utils.ToPtr("tidbcloud_dm_meta"),
						// how to handle conflicted data
						OnDuplicate: openapi.TaskOnDuplicateReplace,
						// the way to coordinate DDL
						ShardMode: &taskShardMode,
						// source-related configuration
						SourceConfig: openapi.TaskSourceConfig{
							FullMigrateConf: &openapi.TaskFullMigrateConf{
								ImportMode: &cfg.ImportMode,
								DataDir:    &cfg.DataDir,
								PdAddr:     &cfg.PdAddr,
								Security:   cfg.PDSecurity,
							},
							SourceConf: []openapi.TaskSourceConf{
								{
									BinlogName: &cfg.BinlogName,
									BinlogPos:  &cfg.BinlogPos,
									SourceName: cfg.Source.SourceName,
								},
							},
						},
						// table migrate rule
						TableMigrateRule: cfg.TableMigrateRule,
						// downstream database configuration
						TargetConfig: *targetConfig,
						// migrate mode
						TaskMode: cfg.TaskModel,
					},
				},
			},
			want: &openapi.OperateTaskResponse{
				Task: openapi.Task{
					Name: cfg.TaskName,
				},
			},
			wantErr: false,
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			got, err := cfg.dmOpenapiClient.CreateTask(cfg.ctx, tt.args.req)
			if (err != nil) != tt.wantErr {
				t.Errorf("DMOpenapiClient.CreateTask() error = %v, wantErr %v", err, tt.wantErr)
				return
			}
			if got.Task.Name != tt.want.Task.Name {
				t.Errorf("DMOpenapiClient.CreateTask() = %v, want %v", got, tt.want)
			}
		})
	}
}

func TestDMOpenapiClient_QueryTaskStatus(t *testing.T) {

	type args struct {
		name       string
		sourceList openapi.SourceNameList
	}

	cfg := initConfig(t)
	tests := []struct {
		name    string
		args    args
		want    *openapi.GetTaskStatusResponse
		wantErr bool
	}{
		{
			name: "OK",
			args: args{
				name: cfg.TaskName,
				sourceList: openapi.SourceNameList{
					cfg.Source.SourceName,
				},
			},
			want: &openapi.GetTaskStatusResponse{
				Data: []openapi.SubTaskStatus{
					{
						Name: cfg.TaskName,
						Unit: "dump",
					},
				},
			},
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			got, err := cfg.dmOpenapiClient.QueryTaskStatus(cfg.ctx, tt.args.name, tt.args.sourceList)
			if (err != nil) != tt.wantErr {
				t.Errorf("DMOpenapiClient.QueryTaskStatus() error = %v, wantErr %v", err, tt.wantErr)
				return
			}
			if len(got.Data) != len(tt.want.Data) {
				t.Errorf("DMOpenapiClient.QueryTaskStatus() = %+v, want %+v", got, tt.want)
			}
			bytes, err := json.Marshal(got)
			fmt.Printf(string(bytes), err)
		})
	}
}

func TestDMOpenapiClient_StartTask(t *testing.T) {

	type args struct {
		name string
		req  *openapi.StartTaskRequest
	}

	cfg := initConfig(t)
	tests := []struct {
		name    string
		args    args
		wantErr bool
	}{
		{
			name: "OK",
			args: args{
				name: cfg.TaskName,
				req:  &openapi.StartTaskRequest{
					// SourceNameList: &openapi.SourceNameList{
					// 	cfg.Source.SourceName,
					// },
				},
			},
			wantErr: false,
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			if err := cfg.dmOpenapiClient.StartTask(cfg.ctx, tt.args.name, tt.args.req); (err != nil) != tt.wantErr {
				t.Errorf("DMOpenapiClient.StartTask() error = %v, wantErr %v", err, tt.wantErr)
			}
		})
	}
}

func TestDMOpenapiClient_StopTask(t *testing.T) {
	type args struct {
		name string
		req  *openapi.StopTaskRequest
	}

	cfg := initConfig(t)
	tests := []struct {
		name    string
		args    args
		wantErr bool
	}{
		{
			name: "OK",
			args: args{
				name: cfg.TaskName,
				req: &openapi.StopTaskRequest{
					SourceNameList: &openapi.SourceNameList{
						cfg.Source.SourceName,
					},
				},
			},
			wantErr: false,
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			if err := cfg.dmOpenapiClient.StopTask(cfg.ctx, tt.args.name, tt.args.req); (err != nil) != tt.wantErr {
				t.Errorf("DMOpenapiClient.StopTask() error = %v, wantErr %v", err, tt.wantErr)
			}
		})
	}
}

func TestDMOpenapiClient_ListWorks(t *testing.T) {

	cfg := initConfig(t)
	tests := []struct {
		name    string
		want    *openapi.GetClusterWorkerListResponse
		wantErr bool
	}{
		{
			name: "OK",
			want: &openapi.GetClusterWorkerListResponse{
				Data: []openapi.ClusterWorker{
					{
						BoundStage: "bound",
					},
				},
			},
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			got, err := cfg.dmOpenapiClient.ListWorks(cfg.ctx)
			if (err != nil) != tt.wantErr {
				t.Errorf("DMOpenapiClient.ListWorks() error = %v, wantErr %v", err, tt.wantErr)
				return
			}
			fmt.Printf("%+v", got.Data)
			if len(got.Data) != len(tt.want.Data) {
				t.Errorf("DMOpenapiClient.ListWorks() = %v, want %v", got, tt.want)
			}
		})
	}
}

func TestDMOpenapiClient_FULL_Task(t *testing.T) {

	// create data source
	// check data source bound status
	// create dump task
	// start dump task
	// wait dump task completed
	// start load task

	TestDMOpenapiClient_CreateDataSource(t)

	type args struct {
		req *openapi.CreateTaskRequest
	}

	cfg := initConfig(t)
	taskName := cfg.TaskName
	taskShardMode := openapi.TaskShardMode("pessimistic")
	tests := []struct {
		name    string
		args    args
		want    *openapi.OperateTaskResponse
		wantErr bool
	}{
		{
			name: "OK",
			args: args{
				req: &openapi.CreateTaskRequest{
					Task: openapi.Task{
						Name: taskName,
						// whether to enable support for the online ddl plugin
						EnhanceOnlineSchemaChange: false,
						// ignore precheck items
						IgnoreCheckingItems: &[]string{"all"},
						// downstream database for storing meta information
						// TODO whether need this field
						MetaSchema: utils.ToPtr("tidbcloud_dm_meta"),
						// how to handle conflicted data
						OnDuplicate: openapi.TaskOnDuplicateReplace,
						// the way to coordinate DDL
						ShardMode: &taskShardMode,
						// source-related configuration
						SourceConfig: openapi.TaskSourceConfig{
							FullMigrateConf: &openapi.TaskFullMigrateConf{
								DataDir: &cfg.DataDir,
							},
							SourceConf: []openapi.TaskSourceConf{
								{
									BinlogName: &cfg.BinlogName,
									BinlogPos:  &cfg.BinlogPos,
									SourceName: cfg.Source.SourceName,
								},
							},
						},
						// table migrate rule
						TableMigrateRule: cfg.TableMigrateRule,
						// downstream database configuration
						TargetConfig: *cfg.TargetDB,
						// migrate mode
						TaskMode: cfg.TaskModel,
					},
				},
			},
			want: &openapi.OperateTaskResponse{
				Task: openapi.Task{
					Name: cfg.TaskName,
				},
			},
			wantErr: false,
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			got, err := cfg.dmOpenapiClient.CreateTask(cfg.ctx, tt.args.req)
			if (err != nil) != tt.wantErr {
				t.Errorf("DMOpenapiClient.CreateTask() error = %v, wantErr %v", err, tt.wantErr)
				return
			}
			if got.Task.Name != tt.want.Task.Name {
				t.Errorf("DMOpenapiClient.CreateTask() = %v, want %v", got, tt.want)
			}
		})
	}
}

func TestDMOpenapiClient_DeleteTask(t *testing.T) {
	type args struct {
		name string
	}

	cfg := initConfig(t)
	tests := []struct {
		name    string
		args    args
		wantErr bool
	}{
		{
			name: "OK",
			args: args{
				name: "task-dump",
			},
			wantErr: false,
		},
		{
			name: "OK",
			args: args{
				name: "task-load",
			},
			wantErr: false,
		},
		{
			name: "OK",
			args: args{
				name: "task-sync",
			},
			wantErr: false,
		},

		{
			name: "OK",
			args: args{
				name: "dm-342059-task-dump",
			},
			wantErr: false,
		},

		{
			name: "OK",
			args: args{
				name: "dm-342059-task-load",
			},
			wantErr: false,
		},

		{
			name: "OK",
			args: args{
				name: "dm-342059-task-sync",
			},
			wantErr: false,
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			if err := cfg.dmOpenapiClient.DeleteTask(cfg.ctx, tt.args.name); (err != nil) != tt.wantErr {
				t.Errorf("DMOpenapiClient.DeleteTask() error = %v, wantErr %v", err, tt.wantErr)
			}
		})
	}
}