package changefeed

import (
	"strconv"
	"testing"

	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	changefeedopenbp "github.com/tidbcloud/dataflow-service/idl/pbgen/proto/openapi/v1beta1"
	mgmt "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
)

func TestConvertConnectorToChangefeed(t *testing.T) {
	connector := &dataflow.Connector{
		ConnectorId:   123,
		Name:          "test-cf",
		Status:        dataflow.Connector_RUNNING,
		CheckpointTso: 42000,
		Request: &dataflow.ConnectorCreateReq{
			ClusterId: 456,
		},
	}

	cf := convertConnectorToChangefeed(connector)
	require.NotNil(t, cf)
	require.Equal(t, strconv.FormatUint(connector.ConnectorId, 10), cf.Id)
	require.Equal(t, strconv.FormatUint(connector.GetRequest().GetClusterId(), 10), cf.ClusterId)
	require.Equal(t, connector.Name, cf.Name)
	require.Equal(t, connector.CheckpointTso, cf.CheckpointTso)
	require.Equal(t, changefeedopenbp.Changefeed_RUNNING, cf.State)

	// Test a different status
	connector.Status = dataflow.Connector_PAUSED
	cf = convertConnectorToChangefeed(connector)
	require.Equal(t, changefeedopenbp.Changefeed_PAUSED, cf.State)
}

func TestToDataflowS3DateSeparator(t *testing.T) {
	tests := map[string]struct {
		input    changefeedopenbp.S3DataFormat_DateSeparator
		expected string
	}{
		"unspecified defaults to none": {
			input:    changefeedopenbp.S3DataFormat_DATE_SEPARATOR_UNSPECIFIED,
			expected: "none",
		},
		"none": {
			input:    changefeedopenbp.S3DataFormat_NONE,
			expected: "none",
		},
		"year": {
			input:    changefeedopenbp.S3DataFormat_YEAR,
			expected: "year",
		},
		"month": {
			input:    changefeedopenbp.S3DataFormat_MONTH,
			expected: "month",
		},
		"day": {
			input:    changefeedopenbp.S3DataFormat_DAY,
			expected: "day",
		},
		"unknown defaults to none": {
			input:    changefeedopenbp.S3DataFormat_DateSeparator(99),
			expected: "none",
		},
	}

	for name, tc := range tests {
		t.Run(name, func(t *testing.T) {
			require.Equal(t, tc.expected, toDataflowS3DateSeparator(tc.input))
		})
	}
}

func TestConvertConnectorToChangefeed_RedactsInputOnlyFields(t *testing.T) {
	t.Run("mysql password", func(t *testing.T) {
		connector := &dataflow.Connector{
			ConnectorType: dataflow.ConnectorType_MYSQL,
			Request: &dataflow.ConnectorCreateReq{
				ClusterId:     1,
				ConnectorType: dataflow.ConnectorType_MYSQL,
				MysqlInfo: &dataflow.MySQLInfo{
					Connection: &dataflow.Connection{
						Endpoint: "mysql.example.com:3306",
						Username: "user",
						Password: "secret-password",
					},
				},
			},
		}
		cf := convertConnectorToChangefeed(connector)
		require.NotNil(t, cf.GetMysql())
		require.Equal(t, "", cf.GetMysql().GetConnection().GetPassword())
	})

	t.Run("kafka auth password and avro schema registry password", func(t *testing.T) {
		connector := &dataflow.Connector{
			ConnectorType: dataflow.ConnectorType_KAFKA,
			Request: &dataflow.ConnectorCreateReq{
				ClusterId:     1,
				ConnectorType: dataflow.ConnectorType_KAFKA,
				KafkaInfo: &dataflow.KafkaInfo{
					Authentication: &dataflow.Authentication{
						AuthType: dataflow.KafkaAuthType_SASL_PLAIN,
						UserName: "kafka-user",
						Password: "kafka-secret",
					},
					DataFormat: &dataflow.DataFormat{
						Protocol: dataflow.Protocol_AVRO,
						SchemaRegistry: &dataflow.SchemaRegistry{
							SchemaRegistryEndpoints: "http://schema-registry:8081",
							EnableHttpAuth:          true,
							UserName:                "schema-user",
							Password:                "schema-secret",
						},
					},
				},
			},
		}
		cf := convertConnectorToChangefeed(connector)
		require.NotNil(t, cf.GetKafka())
		require.NotNil(t, cf.GetKafka().GetAuthentication())
		require.Nil(t, cf.GetKafka().GetAuthentication().Password)
		require.Equal(t, "", cf.GetKafka().GetDataFormat().GetAvroConfig().GetSchemaRegistry().GetBasicAuthPassword())
	})

	t.Run("s3 access key and secret key", func(t *testing.T) {
		connector := &dataflow.Connector{
			ConnectorType: dataflow.ConnectorType_S3,
			Request: &dataflow.ConnectorCreateReq{
				ClusterId:     1,
				ConnectorType: dataflow.ConnectorType_S3,
				S3Info: &dataflow.S3Info{
					Endpoint: &dataflow.S3Endpoint{
						S3Uri:     "s3://my-bucket/cdc-data",
						AuthType:  dataflow.S3AuthType_S3_AUTH_TYPE_AK_SK,
						AccessKey: "access-key",
						SecretKey: "secret-key",
					},
				},
			},
		}
		cf := convertConnectorToChangefeed(connector)
		require.NotNil(t, cf.GetS3())
		require.Equal(t, "access-key", cf.GetS3().GetEndpoint().GetAccessKey())
		require.Equal(t, "", cf.GetS3().GetEndpoint().GetSecretKey())
	})

	t.Run("azure blob sas token", func(t *testing.T) {
		connector := &dataflow.Connector{
			ConnectorType: dataflow.ConnectorType_AZURE_BLOB_STORAGE,
			Request: &dataflow.ConnectorCreateReq{
				ClusterId:     1,
				ConnectorType: dataflow.ConnectorType_AZURE_BLOB_STORAGE,
				AzureBlobStorageInfo: &dataflow.AzureBlobStorageInfo{
					Endpoint: &dataflow.AzureBlobStorageEndpoint{
						BlobUri:  "https://myaccount.blob.core.windows.net/mycontainer/cdc-data",
						SasToken: "sas-token",
					},
				},
			},
		}
		cf := convertConnectorToChangefeed(connector)
		require.NotNil(t, cf.GetAzureBlob())
		require.Equal(t, "", cf.GetAzureBlob().GetEndpoint().GetSasToken())
	})
}

func TestConvertConnectorToOpenAPIFilters(t *testing.T) {
	t.Run("correctly converts filters and fixes bug", func(t *testing.T) {
		rule := &dataflow.EventFilterRule{
			Matcher:               []string{"test.tbl"},
			IgnoreEvent:           []string{"update"},
			IgnoreDeleteValueExpr: "col = 1", // This is the field that was previously mapped incorrectly
		}
		connector := &dataflow.Connector{
			ConnectorType: dataflow.ConnectorType_KAFKA,
			Request: &dataflow.ConnectorCreateReq{
				KafkaInfo: &dataflow.KafkaInfo{
					TableConfig: &dataflow.TableConfig{EventFilterRule: []*dataflow.EventFilterRule{rule}},
				},
			},
		}

		filters, err := convertConnectorToOpenAPIFilters(connector)
		require.NoError(t, err)
		require.Len(t, filters.EventFilters, 1)
		apiFilter := filters.EventFilters[0]
		require.Equal(t, rule.Matcher, apiFilter.TableMatchers)
		require.Equal(t, rule.IgnoreEvent, apiFilter.IgnoredEvents)
		require.Equal(t, rule.IgnoreDeleteValueExpr, apiFilter.IgnoredDeleteValueExpression)
	})

	t.Run("unsupported type returns error", func(t *testing.T) {
		connector := &dataflow.Connector{ConnectorType: dataflow.ConnectorType_TOTAL}
		_, err := convertConnectorToOpenAPIFilters(connector)
		require.Error(t, err)
	})

	t.Run("nil table config returns empty filters", func(t *testing.T) {
		connector := &dataflow.Connector{
			ConnectorType: dataflow.ConnectorType_S3,
			Request:       &dataflow.ConnectorCreateReq{S3Info: &dataflow.S3Info{}}, // Nil TableConfig
		}
		filters, err := convertConnectorToOpenAPIFilters(connector)
		require.NoError(t, err)
		require.NotNil(t, filters)
		require.Len(t, filters.EventFilters, 0)
	})
}

func TestNewConnectorEditRequest(t *testing.T) {
	apiFilters := []*changefeedopenbp.EventFilter{
		{TableMatchers: []string{"new.test"}},
	}
	originalRules := []*dataflow.EventFilterRule{
		{Matcher: []string{"old.test"}},
	}

	baseConnector := &dataflow.Connector{
		ConnectorId:   101,
		Name:          "cf-s3",
		ConnectorType: dataflow.ConnectorType_S3,
		Request: &dataflow.ConnectorCreateReq{
			ClusterId: 1, ProjectId: 2, OrgId: 3,
			S3Info: &dataflow.S3Info{
				TableConfig: &dataflow.TableConfig{EventFilterRule: originalRules},
			},
		},
	}

	t.Run("success", func(t *testing.T) {
		req, err := newConnectorEditRequest(baseConnector, apiFilters)
		require.NoError(t, err)
		require.NotNil(t, req)

		// Check top-level fields
		require.Equal(t, baseConnector.ConnectorId, req.ConnectorId)
		require.Equal(t, baseConnector.Name, req.Name)
		require.Equal(t, baseConnector.ConnectorType, req.ConnectorType)

		// Check that the new filters are set correctly
		require.NotNil(t, req.S3Info)
		require.Len(t, req.S3Info.TableConfig.EventFilterRule, 1)
		require.Equal(t, "new.test", req.S3Info.TableConfig.EventFilterRule[0].Matcher[0])

		// IMPORTANT: Check that the original connector object was not modified (no side effects)
		require.Len(t, baseConnector.Request.S3Info.TableConfig.EventFilterRule, 1)
		require.Equal(t, "old.test", baseConnector.Request.S3Info.TableConfig.EventFilterRule[0].Matcher[0])
	})

	t.Run("unsupported type returns error", func(t *testing.T) {
		unsupportedConnector := &dataflow.Connector{ConnectorType: dataflow.ConnectorType_TOTAL}
		_, err := newConnectorEditRequest(unsupportedConnector, apiFilters)
		require.Error(t, err)
	})
}

func TestConvertAPIFiltersToInternal(t *testing.T) {
	apiFilters := []*changefeedopenbp.EventFilter{
		{
			TableMatchers:                []string{"db.tbl"},
			IgnoredEvents:                []string{"update"},
			IgnoredSqlStatements:         []string{"TRUNCATE TABLE"},
			IgnoredDeleteValueExpression: "col_del = 1",
		},
		{
			TableMatchers: []string{"db.tbl2"},
		},
	}

	internalRules := convertAPIFiltersToInternal(apiFilters)
	require.Len(t, internalRules, 2)

	rule1 := internalRules[0]
	require.Equal(t, apiFilters[0].TableMatchers, rule1.Matcher)
	require.Equal(t, apiFilters[0].IgnoredEvents, rule1.IgnoreEvent)
	require.Equal(t, apiFilters[0].IgnoredSqlStatements, rule1.IgnoreSql)
	require.Equal(t, apiFilters[0].IgnoredDeleteValueExpression, rule1.IgnoreDeleteValueExpr)

	rule2 := internalRules[1]
	require.Equal(t, apiFilters[1].TableMatchers, rule2.Matcher)
}

func TestToOpenAPINetworkType_CollapsesPrivateEndpointEnums(t *testing.T) {
	require.Equal(t,
		changefeedopenbp.NetworkInfo_NETWORK_TYPE_PRIVATE_LINK,
		toOpenAPINetworkType(dataflow.NetworkType_NetworkType_Private_Link),
	)
	require.Equal(t,
		changefeedopenbp.NetworkInfo_NETWORK_TYPE_PRIVATE_LINK,
		toOpenAPINetworkType(dataflow.NetworkType_NetworkType_Private_Service_Connect),
	)
}

func TestToDataflowNetworkInfo_PrivateLinkProviderMapping(t *testing.T) {
	tests := map[string]struct {
		cluster *mgmt.Cluster
		want    dataflow.NetworkType
	}{
		"aws private link": {
			cluster: &mgmt.Cluster{CloudProvider: string(domain.CloudProviderTypeAWS)},
			want:    dataflow.NetworkType_NetworkType_Private_Link,
		},
		"azure private link": {
			cluster: &mgmt.Cluster{CloudProvider: string(domain.CloudProviderTypeAzure)},
			want:    dataflow.NetworkType_NetworkType_Private_Link,
		},
		"gcp private service connect": {
			cluster: &mgmt.Cluster{CloudProvider: string(domain.CloudProviderTypeGCP)},
			want:    dataflow.NetworkType_NetworkType_Private_Service_Connect,
		},
	}

	for name, tc := range tests {
		t.Run(name, func(t *testing.T) {
			sinkEndpointID := uint64(12345)
			got, err := toDataflowNetworkInfo(&changefeedopenbp.NetworkInfo{
				NetworkType:    changefeedopenbp.NetworkInfo_NETWORK_TYPE_PRIVATE_LINK,
				SinkEndpointId: &sinkEndpointID,
				Ports:          []uint32{9092},
			}, tc.cluster)
			require.NoError(t, err)
			require.Equal(t, tc.want, got.GetNetworkType())
			require.Equal(t, sinkEndpointID, got.GetSinkEndpointId())
			require.Equal(t, []uint32{9092}, got.GetPorts())
		})
	}
}

func TestToDataflowNetworkInfo_PrivateLinkProviderError(t *testing.T) {
	req := &changefeedopenbp.NetworkInfo{
		NetworkType: changefeedopenbp.NetworkInfo_NETWORK_TYPE_PRIVATE_LINK,
	}

	_, err := toDataflowNetworkInfo(req, nil)
	require.Error(t, err)
	require.Contains(t, err.Error(), "cluster metadata is required")

	_, err = toDataflowNetworkInfo(req, &mgmt.Cluster{CloudProvider: "alicloud"})
	require.Error(t, err)
	require.Contains(t, err.Error(), "unsupported cluster provider")
}

func TestToDataflowNetworkInfo_UnsupportedNetworkType(t *testing.T) {
	_, err := toDataflowNetworkInfo(&changefeedopenbp.NetworkInfo{
		NetworkType: changefeedopenbp.NetworkInfo_NetworkType(99),
	}, &mgmt.Cluster{CloudProvider: string(domain.CloudProviderTypeAWS)})
	require.Error(t, err)
	require.Contains(t, err.Error(), "unsupported network_info.network_type")
}

func TestBuildDataflowTableConfig(t *testing.T) {
	tests := map[string]struct {
		in   *changefeedopenbp.TableConfig
		want *dataflow.TableConfig
	}{
		"nil returns nil": {
			in: nil,
		},
		"maps filter rules case sensitivity and event filters": {
			in: &changefeedopenbp.TableConfig{
				FilterRules:   []string{"db.tbl", "db.user_*"},
				CaseSensitive: true,
				EventFilters: []*changefeedopenbp.EventFilter{
					{
						TableMatchers:                []string{"db.tbl"},
						IgnoredEvents:                []string{"update"},
						IgnoredSqlStatements:         []string{"truncate table"},
						IgnoredDeleteValueExpression: "deleted = 1",
					},
				},
			},
			want: &dataflow.TableConfig{
				FilterRule:    []string{"db.tbl", "db.user_*"},
				CaseSensitive: true,
				EventFilterRule: []*dataflow.EventFilterRule{
					{
						Matcher:               []string{"db.tbl"},
						IgnoreEvent:           []string{"update"},
						IgnoreSql:             []string{"truncate table"},
						IgnoreDeleteValueExpr: "deleted = 1",
					},
				},
			},
		},
	}

	for name, tc := range tests {
		t.Run(name, func(t *testing.T) {
			require.Equal(t, tc.want, toDataflowTableConfig(tc.in))
		})
	}
}

func TestBuildDataflowStartPosition(t *testing.T) {
	tests := map[string]struct {
		in   *changefeedopenbp.StartPosition
		want *dataflow.StartPosition
	}{
		"nil returns nil": {
			in: nil,
		},
		"from tso": {
			in: &changefeedopenbp.StartPosition{
				Mode: changefeedopenbp.StartPosition_FROM_TSO,
				Position: &changefeedopenbp.StartPosition_StartTso{
					StartTso: 123,
				},
			},
			want: &dataflow.StartPosition{
				Mode: dataflow.StartMode_FROM_TSO,
				Tso:  123,
			},
		},
		"from utc": {
			in: &changefeedopenbp.StartPosition{
				Mode: changefeedopenbp.StartPosition_FROM_UTC,
				Position: &changefeedopenbp.StartPosition_StartTimestamp{
					StartTimestamp: "2026-01-01T00:00:00Z",
				},
			},
			want: &dataflow.StartPosition{
				Mode: dataflow.StartMode_FROM_UTC,
				Utc:  "2026-01-01T00:00:00Z",
			},
		},
		"default mode": {
			in: &changefeedopenbp.StartPosition{},
			want: &dataflow.StartPosition{
				Mode: dataflow.StartMode_FROM_NOW,
			},
		},
	}

	for name, tc := range tests {
		t.Run(name, func(t *testing.T) {
			require.Equal(t, tc.want, toDataflowStartPosition(tc.in))
		})
	}
}