package pimport

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

	"github.com/tidbcloud/dataflow-service/internal/constants"

	"github.com/pingcap/failpoint"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	"github.com/tidbcloud/dataflow-service/internal/config"
	"github.com/tidbcloud/dataflow-service/internal/db/entity"
	"github.com/tidbcloud/dataflow-service/pkg/event"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/pkgv2/secret/encrypt"
	"go.uber.org/zap"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)

type mockPublisher struct {
	mock.Mock
}

func (p *mockPublisher) Publish(ctx context.Context, topic event.Topic, meta *event.Meta, event interface{}) error {
	args := p.Called(ctx, topic, meta, event)
	return args.Error(0)
}

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

	pb := &mockPublisher{}
	tm := time.Now()
	newImportEntity := &ImportTask{
		Import: entity.Import{
			Updatable: entity.Updatable{
				Base: entity.Base{
					ID:        1,
					CreatedAt: tm,
				},
			},
			ClusterID:  123,
			Name:       "import-test",
			Status:     entity.ImportStatusFailed,
			Creator:    "test",
			StartedAt:  &tm,
			FinishedAt: nil,
			Message:    "error-test",
		},
	}
	importEvent := &event.ImportEvent{
		EventType:  event.ImportEventTypeFailed,
		ID:         1,
		Name:       "import-test",
		ClusterID:  123,
		Status:     "failed",
		Creator:    "test",
		CreatedAt:  &tm,
		StartedAt:  &tm,
		FinishedAt: nil,
		ExtraInfo: &event.ImportEventExtraInfo{
			Message: "error-test",
		},
	}
	pb.On("Publish", mock.Anything, event.ImportTopic, mock.Anything, importEvent).Return(nil)
	require.NoError(t, publishEvent(pb, newImportEntity))
}

func TestConvertCsvConfigToCr(t *testing.T) {
	type testCase struct {
		CaseName               string
		DataflowCSVFormatStr   string
		ExpectedCRCSVConfigStr string
	}
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	for _, tc := range []testCase{
		{
			CaseName:               "basic",
			DataflowCSVFormatStr:   `{"separator":",","delimiter":"\"","header":true,"null":"\\N","backslash_escape":true,"header_schema_match":true}`,
			ExpectedCRCSVConfigStr: `{"separator":",","delimiter":"\"","header":true,"notNull":false,"null":"\\N","backslashEscape":true,"trimLastSeparator":false,"headerSchemaMatch":true}`,
		},
		{
			CaseName:               "missing_header_schema_match",
			DataflowCSVFormatStr:   `{"separator":",","delimiter":"\"","header":true,"null":"\\N","backslash_escape":true}`,
			ExpectedCRCSVConfigStr: `{"separator":",","delimiter":"\"","header":true,"notNull":false,"null":"\\N","backslashEscape":true,"trimLastSeparator":false}`,
		},
		{
			CaseName:               "missing_header_schema_match_02",
			DataflowCSVFormatStr:   `{"separator":",","delimiter":"\"","null":"\\N","backslash_escape":true}`,
			ExpectedCRCSVConfigStr: `{"separator":",","delimiter":"\"","header":false,"notNull":false,"null":"\\N","backslashEscape":true,"trimLastSeparator":false}`,
		},
		{
			CaseName:               "tab_separator",
			DataflowCSVFormatStr:   `{"separator":"\t","delimiter":"\"","header":true,"null":"\\N","backslash_escape":true}`,
			ExpectedCRCSVConfigStr: `{"separator":"\t","delimiter":"\"","header":true,"notNull":false,"null":"\\N","backslashEscape":true,"trimLastSeparator":false}`,
		},
		{
			CaseName:               "skip_header_schema_match",
			DataflowCSVFormatStr:   `{"separator":",","delimiter":"\"","header":true,"null":"\\N","backslash_escape":true,"header_schema_match":false}`,
			ExpectedCRCSVConfigStr: `{"separator":",","delimiter":"\"","header":true,"notNull":false,"null":"\\N","backslashEscape":true,"trimLastSeparator":false,"headerSchemaMatch":false}`,
		},
	} {
		crCSVConfig, err := convertCsvStringToCr(ctx, tc.DataflowCSVFormatStr)
		require.NoErrorf(t, err, "case name: %s", tc.CaseName)
		crCSVConfigJSON, err := json.Marshal(crCSVConfig)
		require.NoErrorf(t, err, "case name: %s", tc.CaseName)
		require.Equalf(t, tc.ExpectedCRCSVConfigStr, string(crCSVConfigJSON), "case name: %s", tc.CaseName)
	}
}

func TestGenImportObjectSpec(t *testing.T) {
	testCases := []struct {
		CaseName     string
		ImportTask   *ImportTask
		ExpectedSpec *cloudv1.ImportSpec
	}{
		{
			CaseName: "import from s3",
			ImportTask: &ImportTask{
				Import: entity.Import{
					ClusterID:      999,
					AwsRoleArn:     "arn:aws:iam::XXXXXXXXXXXX:role/my-role",
					DataSourceType: entity.ImportDataSourceS3,
				},
			},
			ExpectedSpec: &cloudv1.ImportSpec{
				ClusterName: "999",
				Source: cloudv1.Source{
					UserName: "cloud_admin",
				},
				AccessConfig: cloudv1.AccessConfig{
					RoleArn: "arn:aws:iam::XXXXXXXXXXXX:role/my-role",
				},
			},
		},
		{
			CaseName: "import from gcs",
			ImportTask: &ImportTask{
				Import: entity.Import{
					ClusterID:      999,
					DataSourceType: entity.ImportDataSourceGcs,
				},
			},
			ExpectedSpec: &cloudv1.ImportSpec{
				ClusterName: "999",
				Source: cloudv1.Source{
					UserName: "cloud_admin",
				},
			},
		},
	}
	for _, tc := range testCases {
		spec := genImportObjectSpec(tc.ImportTask)
		t.Logf("spec: %+v", spec)
		require.Equalf(t, tc.ExpectedSpec, spec, "case name: %s", tc.CaseName)
	}
}

func TestNewInfraAPIImportObject(t *testing.T) {
	// nolint:errcheck
	// failpoint.Enable("github.com/tidbcloud/dataflow-service/pkg/infra-api/mockLightningCfg", "return(true)")
	failpoint.Enable("github.com/tidbcloud/dataflow-service/pkg/import-config/mockLightningCfg", "return(true)")
	// nolint:errcheck
	// defer failpoint.Disable("github.com/tidbcloud/dataflow-service/pkg/infra-api/mockLightningCfg")
	defer failpoint.Disable("github.com/tidbcloud/dataflow-service/pkg/import-config/mockLightningCfg")

	fakeToken := "fake-token"

	require.NoError(t, config.MockConfig())
	ctx := context.Background()
	logger, err := zap.NewDevelopment()
	require.NoError(t, err)
	mockEnc := &MockEncryptor{}
	mockEnc.On("Decrypt", mock.Anything).Return(encrypt.Secret("actual_password"), nil)
	m := NewModel(logger, mockEnc, nil, nil)

	testCases := []struct {
		CaseName       string
		ImportTask     *ImportTask
		ExpectedImport *cloudv1.Import
	}{
		{
			CaseName: "eks provider",
			ImportTask: &ImportTask{
				Import: entity.Import{
					Updatable: entity.Updatable{
						Base: entity.Base{
							ID: 9999,
						},
					},
					Name:           "import",
					ClusterID:      999,
					BucketRegion:   "us-east-1",
					AwsRoleArn:     "arn:aws:iam::XXXXXXXXXXXX:role/my-role",
					DataFormat:     entity.ImportDataFormatCsv,
					DataSourceType: entity.ImportDataSourceS3,
					TargetUsername: "cloud_admin",
					TargetPassword: "YWFhCg==",
					SourceURL:      "s3://some_bucket/some_path/",
					TargetTableInfos: entity.TargetTableInfos{
						Infos: []entity.TargetTableInfo{
							{Schema: "db", Table: "table"},
						},
					},
				},
				Cluster: &core.Cluster{
					Id:                999,
					TenantId:          9,
					ProjectId:         99,
					CloudProviderName: constants.CloudProviderAWS,
					InfraProviderType: "eks",
					RegionName:        "us-west-2",
					IsDevTier:         false,
					DataPlaneInfo: &core.Cluster_DataPlaneInfo{
						K8SNamespace: "tidb999",
						NetworkName:  "eks-us-west-2",
					},
				},
			},
			ExpectedImport: &cloudv1.Import{
				ObjectMeta: metav1.ObjectMeta{
					Namespace: "core-99",
					Name:      "import-9999",
					Labels: map[string]string{
						"tags.tidbcloud.com/tenant":  "9",
						"tags.tidbcloud.com/project": "99",
						"tags.tidbcloud.com/cluster": "999",
						cloudv1.ProviderTypeLabel:    "eks",
						cloudv1.RegionLabel:          "us-west-2",
					},
					Annotations: map[string]string{
						"import.tidbcloud.com/image":             "baseimage:v1",
						"import.tidbcloud.com/network-name":      "eks-us-west-2",
						"import.tidbcloud.com/cluster-namespace": "tidb999",
					},
				},
				Spec: cloudv1.ImportSpec{
					ClusterName: "999",
					Source: cloudv1.Source{
						UserName: "cloud_admin",
						Format:   3,
					},
					AccessConfig: cloudv1.AccessConfig{
						RoleArn: "arn:aws:iam::XXXXXXXXXXXX:role/my-role",
					},
					LightningConfig: fmt.Sprintf(`[mydumper]
  data-source-dir = "s3://some_bucket/some_path/"
  %s

  [[mydumper.files]]
    compression = "$1"
    pattern = "(?i)^db.table(?:(?:\\.\\d+)|(?:\\.?(?:_\\d+){3}))?\\.csv(?:\\.(\\w+))?$"
    schema = "db"
    table = "table"
    type = "csv"

[tidb]
  host = "db-tidb"
  password = "actual_password"
  pd-addr = "db-pd:2379"
  port = 4000
  user = "cloud_admin"

[tikv-importer]
  backend = "local"
`, "filter = [\"`db`.`table`\"]"),
				},
			},
		},
		{
			CaseName: "serverless provider",
			ImportTask: &ImportTask{
				Import: entity.Import{
					Updatable: entity.Updatable{
						Base: entity.Base{
							ID: 9999,
						},
					},
					ClusterID:      999,
					AwsRoleArn:     "arn:aws:iam::XXXXXXXXXXXX:role/my-role",
					DataFormat:     entity.ImportDataFormatCsv,
					DataSourceType: entity.ImportDataSourceS3,
					TargetUsername: "cloud_admin",
					TargetPassword: "YWFhCg==",
					SourceURL:      "s3://some_bucket/some_path/",
					TargetTableInfos: entity.TargetTableInfos{
						Infos: []entity.TargetTableInfo{
							{Schema: "db", Table: "table", CustomFile: "s3://some_bucket/some_path/abc*"},
						},
					},
				},
				Cluster: &core.Cluster{
					Id:                999,
					TenantId:          9,
					ProjectId:         99,
					CloudProviderName: constants.CloudProviderAWS,
					InfraProviderType: "aws-free-tier",
					RegionName:        "us-east-1",
					IsDevTier:         true,
					ServerlessToken:   &fakeToken,
					DataPlaneInfo: &core.Cluster_DataPlaneInfo{
						K8SNamespace: "free-tier-nightly-ms-99999",
						NetworkName:  "aws-free-tier-us-east-1",
					},
				},
			},
			ExpectedImport: &cloudv1.Import{
				ObjectMeta: metav1.ObjectMeta{
					Namespace: "core-99",
					Name:      "import-9999",
					Labels: map[string]string{
						"tags.tidbcloud.com/tenant":  "9",
						"tags.tidbcloud.com/project": "99",
						"tags.tidbcloud.com/cluster": "999",
						cloudv1.ProviderTypeLabel:    "aws-free-tier",
						cloudv1.RegionLabel:          "us-east-1",
					},
					Annotations: map[string]string{
						"import.tidbcloud.com/image":             "baseimage:v1",
						"import.tidbcloud.com/network-name":      "aws-free-tier-us-east-1",
						"import.tidbcloud.com/cluster-namespace": "free-tier-nightly-ms-99999",
					},
				},
				Spec: cloudv1.ImportSpec{
					ClusterName: "999",
					Source: cloudv1.Source{
						UserName: "cloud_admin",
						Format:   3,
					},
					AccessConfig: cloudv1.AccessConfig{
						RoleArn: "arn:aws:iam::XXXXXXXXXXXX:role/my-role",
					},
					LightningConfig: fmt.Sprintf(`[mydumper]
  data-source-dir = "s3://some_bucket/some_path/"
  %s

  [[mydumper.files]]
    pattern = "^abc.*$"
    schema = "db"
    table = "table"
    type = "csv"

[tidb]
  host = "tidb-gateway-inner.tidb-admin.svc.cluster.local"
  password = "actual_password"
  pd-addr = "serverless-cluster-pd.tidb-serverless.svc.cluster.local:2379"
  port = 4000
  user = "fake-token.cloud_admin"

[tikv-importer]
  backend = "tidb"
`, "filter = [\"`db`.`table`\"]"),
				},
			},
		},
		{
			CaseName: "gke provider",
			ImportTask: &ImportTask{
				Import: entity.Import{
					Updatable: entity.Updatable{
						Base: entity.Base{
							ID: 9999,
						},
					},
					Name:           "import",
					ClusterID:      999,
					BucketRegion:   "us-west1",
					DataFormat:     entity.ImportDataFormatCsv,
					DataSourceType: entity.ImportDataSourceGcs,
					TargetUsername: "cloud_admin",
					TargetPassword: "YWFhCg==",
					SourceURL:      "gs://some_bucket/some_path/",
				},
				Cluster: &core.Cluster{
					Id:                999,
					TenantId:          9,
					ProjectId:         99,
					CloudProviderName: constants.CloudProviderGCP,
					InfraProviderType: "gke",
					RegionName:        "us-west2",
					IsDevTier:         false,
					DataPlaneInfo: &core.Cluster_DataPlaneInfo{
						K8SNamespace: "tidb999",
						NetworkName:  "gke-us-west2",
					},
				},
			},
			ExpectedImport: &cloudv1.Import{
				ObjectMeta: metav1.ObjectMeta{
					Namespace: "core-99",
					Name:      "import-9999",
					Labels: map[string]string{
						"tags.tidbcloud.com/tenant":  "9",
						"tags.tidbcloud.com/project": "99",
						"tags.tidbcloud.com/cluster": "999",
						cloudv1.ProviderTypeLabel:    "gke",
						cloudv1.RegionLabel:          "us-west2",
					},
					Annotations: map[string]string{
						"import.tidbcloud.com/image":             "baseimage:v1",
						"import.tidbcloud.com/network-name":      "gke-us-west2",
						"import.tidbcloud.com/cluster-namespace": "tidb999",
					},
				},
				Spec: cloudv1.ImportSpec{
					ClusterName: "999",
					Source: cloudv1.Source{
						UserName: "cloud_admin",
						Format:   3,
					},
					LightningConfig: `[mydumper]
  data-source-dir = "gs://some_bucket/some_path/"

  [[mydumper.files]]
    pattern = "(?i)^(?:[^/]*/)*([^/.]+)-schema-create\\.sql$"
    schema = "$1"
    type = "schema-schema"

  [[mydumper.files]]
    pattern = "(?i)^(?:[^/]*/)*([^/.]+)\\.(.*?)-schema\\.sql$"
    schema = "$1"
    table = "$2"
    type = "table-schema"

  [[mydumper.files]]
    compression = "$3"
    pattern = "(?i)^(?:[^/]*/)*([^/.]+)\\.([^/.]+?)(?:(?:\\.\\d+)|(?:\\.?(?:_\\d+){3}))?\\.csv(?:\\.(\\w+))?$"
    schema = "$1"
    table = "$2"
    type = "csv"

[tidb]
  host = "db-tidb"
  password = "actual_password"
  pd-addr = "db-pd:2379"
  port = 4000
  user = "cloud_admin"

[tikv-importer]
  backend = "local"
`,
				},
			},
		},
	}
	for _, tc := range testCases {
		imp, err := m.newInfraAPIImportObject(ctx, logger, tc.ImportTask)
		require.NoErrorf(t, err, "case name: %s", tc.CaseName)
		require.Equalf(t, tc.ExpectedImport, imp, "case name: %s", tc.CaseName)
	}
}