package child

import (
	"encoding/json"
	"errors"
	"fmt"
	"strconv"
	"strings"
	"testing"

	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/dataflow-service-ng/app/models/domain"
	dataflow "github.com/tidbcloud/dataflow-service-ng/idl/pbgen/proto/consoleapi"
	"github.com/tidbcloud/dataflow-service-ng/pkg/lake"
	globalngv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/nextgen/v1"
	"google.golang.org/protobuf/proto"
)

const (
	testTiDBPassword      = "test-tidb-password-plain"
	testS3SecretAccessKey = "test-s3-secret-access-key-plain"
	testSnapshotTSO       = uint64(1700000000000) << 18
)

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

	tests := map[string]struct {
		pipelineID uint64
		want       ResourceNames
		wantErr    error
	}{
		"stable child and Lake names": {
			pipelineID: 9001,
			want: ResourceNames{
				Export:         "tidb-dp-export-9001",
				Changefeed:     "tidb-dp-changefeed-9001",
				LakeDataSource: "tidb-dp-ds-9001-13b7994fae93",
				LakeTask:       "tidb-dp-task-9001-13b7994fae93",
			},
		},
		"zero pipeline id is rejected": {
			wantErr: ErrInvalidPipeline,
		},
	}
	for name, test := range tests {
		t.Run(name, func(t *testing.T) {
			t.Parallel()
			got, err := DeterministicResourceNames(test.pipelineID)
			if test.wantErr != nil {
				require.ErrorIs(t, err, test.wantErr)
				return
			}
			require.NoError(t, err)
			require.Equal(t, test.want, got)
		})
	}
}

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

	tests := map[string]struct {
		mutate func(*domain.DataPipeline)
		want   func(*domain.DataPipeline) *globalngv1.CreateExportRequest
	}{
		"full and incremental pipeline": {
			want: expectedExportCreateRequest,
		},
		"full-only pipeline uses the same immutable snapshot contract": {
			mutate: func(pipeline *domain.DataPipeline) {
				pipeline.Params.SyncMode = domain.DataPipelineSyncModeFull
				pipeline.ExecutionPlan.Incremental = nil
				pipeline.IntervalPlan = domain.DeliveryIntervalPlan{}
			},
			want: expectedExportCreateRequest,
		},
	}
	for name, test := range tests {
		t.Run(name, func(t *testing.T) {
			t.Parallel()
			pipeline := testPipeline()
			if test.mutate != nil {
				test.mutate(pipeline)
			}
			built, err := BuildExportCreateRequest(pipeline, testCredentials())
			require.NoError(t, err)
			require.Equal(t, testSnapshotTSO, built.SnapshotTSO())
			require.True(t, proto.Equal(test.want(pipeline), built.Request()))
			require.Zero(t, built.Request().GetExport().GetSnapshotTso(),
				"output-only Export.snapshot_tso must remain unset")

			first := built.Request()
			first.Export.Password = "mutated-by-caller"
			require.Equal(t, testTiDBPassword, built.Request().GetExport().GetPassword(),
				"Request must return a clone")
		})
	}
}

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

	tests := map[string]struct {
		build func(*domain.DataPipeline) (any, error)
		want  func(*domain.DataPipeline) any
	}{
		"create": {
			build: func(pipeline *domain.DataPipeline) (any, error) {
				return BuildChangefeedCreateRequest(pipeline, NewSecret(testS3SecretAccessKey))
			},
			want: func(pipeline *domain.DataPipeline) any {
				return &domain.CreateBizChangefeedReq{
					Name:                "tidb-dp-changefeed-9001",
					ClusterID:           8001,
					DownstreamType:      dataflow.ChangefeedType_S3,
					DownstreamParameter: expectedChangefeedS3Info(pipeline),
					SpecificationCode:   "test-4-rcu",
					UserID:              7001,
				}
			},
		},
		"stopped edit": {
			build: func(pipeline *domain.DataPipeline) (any, error) {
				return BuildChangefeedEditRequest(
					pipeline, 7002, NewSecret(testS3SecretAccessKey))
			},
			want: func(pipeline *domain.DataPipeline) any {
				return &domain.ApplyStoppedBizChangefeedConfigReq{
					BizChangefeedID:     6001,
					Name:                "tidb-dp-changefeed-9001",
					DownstreamParameter: expectedChangefeedS3Info(pipeline),
					UserID:              7002,
				}
			},
		},
	}
	for name, test := range tests {
		t.Run(name, func(t *testing.T) {
			t.Parallel()
			pipeline := testPipeline()
			got, err := test.build(pipeline)
			require.NoError(t, err)
			switch built := got.(type) {
			case ChangefeedCreateRequest:
				assertCreateChangefeedRequestEqual(
					t, test.want(pipeline).(*domain.CreateBizChangefeedReq), built.Request())
				first := built.Request()
				first.DownstreamParameter.(*dataflow.S3Info).Endpoint.SecretKey = "mutated"
				require.Equal(t, testS3SecretAccessKey,
					built.Request().DownstreamParameter.(*dataflow.S3Info).GetEndpoint().GetSecretKey())
			case ChangefeedEditRequest:
				assertEditChangefeedRequestEqual(
					t, test.want(pipeline).(*domain.ApplyStoppedBizChangefeedConfigReq), built.Request())
				first := built.Request()
				first.DownstreamParameter.(*dataflow.S3Info).Endpoint.SecretKey = "mutated"
				require.Equal(t, testS3SecretAccessKey,
					built.Request().DownstreamParameter.(*dataflow.S3Info).GetEndpoint().GetSecretKey())
			default:
				t.Fatalf("unexpected built request type %T", got)
			}
		})
	}
}

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

	tests := map[string]struct {
		mutate func(*domain.DataPipeline)
		want   lake.DataSourceSpec
	}{
		"the filter rules reach Lake unchanged": {
			mutate: func(pipeline *domain.DataPipeline) {
				pipeline.Params.Staging.S3.Endpoint = "https://s3.test.invalid"
			},
			want: lake.DataSourceSpec{
				Name: "tidb-dp-ds-9001-13b7994fae93",
				S3: lake.S3Credentials{
					Bucket:    "test-pipeline-bucket",
					Region:    "us-test-1",
					Endpoint:  "https://s3.test.invalid",
					AccessKey: lake.NewSecret("test-access-key-id"),
					SecretKey: lake.NewSecret(testS3SecretAccessKey),
				},
			},
		},
		// The queue lives on the datasource next to the bucket, so an Edit that
		// attaches one reaches Lake through the datasource update rather than
		// the task update.
		"a notification queue reaches Lake with the bucket": {
			mutate: func(pipeline *domain.DataPipeline) {
				pipeline.Params.Staging.S3.SQSQueueURL =
					"https://sqs.us-test-1.amazonaws.com/000000000000/test-queue.fifo"
				pipeline.Params.Staging.S3.SQSRegion = "us-test-2"
			},
			want: lake.DataSourceSpec{
				Name: "tidb-dp-ds-9001-13b7994fae93",
				S3: lake.S3Credentials{
					Bucket:      "test-pipeline-bucket",
					Region:      "us-test-1",
					AccessKey:   lake.NewSecret("test-access-key-id"),
					SecretKey:   lake.NewSecret(testS3SecretAccessKey),
					SQSQueueURL: "https://sqs.us-test-1.amazonaws.com/000000000000/test-queue.fifo",
					SQSRegion:   "us-test-2",
				},
			},
		},
	}
	for name, test := range tests {
		t.Run(name, func(t *testing.T) {
			t.Parallel()
			pipeline := testPipeline()
			if test.mutate != nil {
				test.mutate(pipeline)
			}
			got, err := BuildLakeDataSourceSpec(
				pipeline, NewSecret(testS3SecretAccessKey))
			require.NoError(t, err)
			require.Equal(t, test.want, got)
		})
	}
}

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

	tests := map[string]struct {
		mutate func(*domain.DataPipeline)
		want   lake.TaskSpec
	}{
		"full and incremental": {
			want: lake.TaskSpec{
				Name:                "tidb-dp-task-9001-13b7994fae93",
				DataSourceName:      "tidb-dp-ds-9001-13b7994fae93",
				Warehouse:           "test-warehouse-id",
				ContinuousIngestion: true,
				TiDB: lake.TiDBTaskSpec{
					SyncMode:                lake.SyncModeSnapshotAndCDC,
					TableRules:              expectedLakeTableRules(),
					PollIntervalSeconds:     int32Pointer(6),
					AllowDelete:             true,
					DumplingS3Prefix:        "customer/base/tidb-data-pipelines/staging/8001/9001/snapshot/",
					DumplingCSVSeparator:    ",",
					DumplingCSVNull:         `\N`,
					DumplingCSVSkipHeader:   int32Pointer(1),
					DumplingAutoCreateTable: true,
					DumplingOnError:         "abort",
					DumplingPurge:           false,
					ChangefeedS3Prefix:      "customer/base/tidb-data-pipelines/staging/8001/9001/incremental/",
					TargetNaming: lake.TargetNaming{
						DatabasePrefix: "lake_",
						DatabaseSuffix: "_archive",
						TablePrefix:    "raw_",
						TableSuffix:    "_copy",
					},
				},
			},
		},
		"full only omits CDC and polling": {
			mutate: func(pipeline *domain.DataPipeline) {
				pipeline.Params.SyncMode = domain.DataPipelineSyncModeFull
				pipeline.ExecutionPlan.Incremental = nil
				pipeline.IntervalPlan = domain.DeliveryIntervalPlan{}
			},
			want: lake.TaskSpec{
				Name:           "tidb-dp-task-9001-13b7994fae93",
				DataSourceName: "tidb-dp-ds-9001-13b7994fae93",
				Warehouse:      "test-warehouse-id",
				TiDB: lake.TiDBTaskSpec{
					SyncMode:                lake.SyncModeSnapshot,
					TableRules:              expectedLakeTableRules(),
					DumplingS3Prefix:        "customer/base/tidb-data-pipelines/staging/8001/9001/snapshot/",
					DumplingCSVSeparator:    ",",
					DumplingCSVNull:         `\N`,
					DumplingCSVSkipHeader:   int32Pointer(1),
					DumplingAutoCreateTable: true,
					DumplingOnError:         "abort",
					DumplingPurge:           false,
					TargetNaming: lake.TargetNaming{
						DatabasePrefix: "lake_",
						DatabaseSuffix: "_archive",
						TablePrefix:    "raw_",
						TableSuffix:    "_copy",
					},
				},
			},
		},
		// A notified pipeline is planned with no consumer interval, and the
		// zero is sent rather than omitted: an absent field cannot clear an
		// interval Lake already holds, which is what an Edit attaching a queue
		// to a polling Pipeline has to do.
		"notified incremental sends a zero poll interval": {
			mutate: func(pipeline *domain.DataPipeline) {
				pipeline.Params.Staging.S3.SQSQueueURL =
					"https://sqs.us-test-1.amazonaws.com/000000000000/test-queue.fifo"
				pipeline.IntervalPlan.ProducerFlushIntervalSeconds = 10
				pipeline.IntervalPlan.ConsumerPollIntervalSeconds = 0
			},
			want: lake.TaskSpec{
				Name:                "tidb-dp-task-9001-13b7994fae93",
				DataSourceName:      "tidb-dp-ds-9001-13b7994fae93",
				Warehouse:           "test-warehouse-id",
				ContinuousIngestion: true,
				TiDB: lake.TiDBTaskSpec{
					SyncMode:                lake.SyncModeSnapshotAndCDC,
					TableRules:              expectedLakeTableRules(),
					PollIntervalSeconds:     int32Pointer(0),
					AllowDelete:             true,
					DumplingS3Prefix:        "customer/base/tidb-data-pipelines/staging/8001/9001/snapshot/",
					DumplingCSVSeparator:    ",",
					DumplingCSVNull:         `\N`,
					DumplingCSVSkipHeader:   int32Pointer(1),
					DumplingAutoCreateTable: true,
					DumplingOnError:         "abort",
					DumplingPurge:           false,
					ChangefeedS3Prefix:      "customer/base/tidb-data-pipelines/staging/8001/9001/incremental/",
					TargetNaming: lake.TargetNaming{
						DatabasePrefix: "lake_",
						DatabaseSuffix: "_archive",
						TablePrefix:    "raw_",
						TableSuffix:    "_copy",
					},
				},
			},
		},
	}
	for name, test := range tests {
		t.Run(name, func(t *testing.T) {
			t.Parallel()
			pipeline := testPipeline()
			if test.mutate != nil {
				test.mutate(pipeline)
			}
			got, err := BuildLakeTaskSpec(pipeline)
			require.NoError(t, err)
			require.Equal(t, test.want, got)
		})
	}
}

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

	pipeline := testPipeline()
	exportRequest, err := BuildExportCreateRequest(pipeline, testCredentials())
	require.NoError(t, err)
	createRequest, err := BuildChangefeedCreateRequest(
		pipeline, NewSecret(testS3SecretAccessKey))
	require.NoError(t, err)
	editRequest, err := BuildChangefeedEditRequest(
		pipeline, 7002, NewSecret(testS3SecretAccessKey))
	require.NoError(t, err)
	dataSource, err := BuildLakeDataSourceSpec(
		pipeline, NewSecret(testS3SecretAccessKey))
	require.NoError(t, err)

	tests := map[string]any{
		"decrypted credentials": testCredentials(),
		"Export request":        exportRequest,
		"Changefeed create":     createRequest,
		"Changefeed edit":       editRequest,
		"Lake datasource":       dataSource,
	}
	for name, value := range tests {
		t.Run(name, func(t *testing.T) {
			t.Parallel()
			formatted := fmt.Sprintf("%v %+v %#v %s %q", value, value, value, value, value)
			encoded, marshalErr := json.Marshal(value)
			require.NoError(t, marshalErr)
			assertSecretsAbsent(t, formatted)
			assertSecretsAbsent(t, string(encoded))
		})
	}
}

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

	tests := map[string]func() error{
		"Export rejects unsupported endpoint": func() error {
			pipeline := testPipeline()
			pipeline.Params.Staging.S3.Endpoint = testS3SecretAccessKey
			_, err := BuildExportCreateRequest(pipeline, testCredentials())
			return err
		},
		"Export rejects malformed secret-like staging path": func() error {
			pipeline := testPipeline()
			pipeline.StagingRoot = testTiDBPassword + "/../"
			_, err := BuildExportCreateRequest(pipeline, testCredentials())
			return err
		},
		"Lake rejects a task with no table rule": func() error {
			pipeline := testPipeline()
			pipeline.Params.Source.TableFilter = nil
			_, err := BuildLakeTaskSpec(pipeline)
			return err
		},
	}
	for name, buildError := range tests {
		t.Run(name, func(t *testing.T) {
			t.Parallel()
			err := buildError()
			require.Error(t, err)
			assertSecretsAbsent(t, err.Error())
			assertSecretsAbsent(t, fmt.Sprintf("%v %+v %#v", err, err, err))
			wrapped := fmt.Errorf("child request failed: %w", err)
			assertSecretsAbsent(t, wrapped.Error())
		})
	}
}

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

	tests := map[string]struct {
		build   func(*domain.DataPipeline) error
		mutate  func(*domain.DataPipeline)
		wantErr error
	}{
		"Export requires decrypted password": {
			build: func(pipeline *domain.DataPipeline) error {
				_, err := BuildExportCreateRequest(pipeline, DecryptedCredentials{
					S3SecretAccessKey: NewSecret(testS3SecretAccessKey),
				})
				return err
			},
			wantErr: ErrMissingDecryptedCredential,
		},
		"Changefeed requires immutable TiCDC plan": {
			mutate: func(pipeline *domain.DataPipeline) {
				pipeline.ExecutionPlan.Incremental.Format = "test-unknown-format"
			},
			build: func(pipeline *domain.DataPipeline) error {
				_, err := BuildChangefeedCreateRequest(
					pipeline, NewSecret(testS3SecretAccessKey))
				return err
			},
			wantErr: ErrUnsupportedExecutionPlan,
		},
		"stopped edit requires persisted child": {
			mutate: func(pipeline *domain.DataPipeline) {
				pipeline.Children.ChangefeedID = 0
			},
			build: func(pipeline *domain.DataPipeline) error {
				_, err := BuildChangefeedEditRequest(
					pipeline, 7002, NewSecret(testS3SecretAccessKey))
				return err
			},
			wantErr: ErrMissingChangefeedChild,
		},
		"Lake task requires a datasource name": {
			mutate: func(pipeline *domain.DataPipeline) {
				pipeline.Children.LakeDataSourceName = ""
			},
			build: func(pipeline *domain.DataPipeline) error {
				_, err := BuildLakeTaskSpec(pipeline)
				return err
			},
			wantErr: ErrInvalidLakeDataSource,
		},
	}
	for name, test := range tests {
		t.Run(name, func(t *testing.T) {
			t.Parallel()
			pipeline := testPipeline()
			if test.mutate != nil {
				test.mutate(pipeline)
			}
			err := test.build(pipeline)
			require.ErrorIs(t, err, test.wantErr)
		})
	}
}

func testPipeline() *domain.DataPipeline {
	return &domain.DataPipeline{
		ID:                      9001,
		TenantID:                1001,
		ClusterID:               8001,
		CreatorID:               7001,
		Name:                    "user-editable-name",
		ChangefeedSpecification: "test-4-rcu",
		Params: domain.DataPipelineParams{
			Version:  1,
			SyncMode: domain.DataPipelineSyncModeFullAndIncremental,
			Source: domain.DataPipelineSourceConfig{
				Username:           "test-source-user",
				PasswordCiphertext: "test-tidb-password-ciphertext",
				TableFilter:        []string{"test_app.*", "test_audit.events"},
				CaseSensitive:      true,
			},
			Staging: domain.DataPipelineStagingConfig{
				Type: domain.DataPipelineStagingTypeCustomerManagedS3,
				S3: domain.DataPipelineS3Config{
					Bucket:                    "test-pipeline-bucket",
					BasePrefix:                "customer/base",
					Region:                    "us-test-1",
					AccessKeyID:               "test-access-key-id",
					SecretAccessKeyCiphertext: "test-s3-secret-ciphertext",
				},
			},
			Destination: domain.DataPipelineDestinationConfig{
				Type:      domain.DataPipelineDestinationTypeTiDBLake,
				LakeOrgID: "7",
				Warehouse: "test-warehouse-id",
				TargetNaming: domain.DataPipelineTargetNaming{
					DatabasePrefix: "lake_",
					DatabaseSuffix: "_archive",
					TablePrefix:    "raw_",
					TableSuffix:    "_copy",
				},
			},
		},
		ExecutionPlan: domain.DataPipelineExecutionPlan{
			Version:         1,
			DestinationType: domain.DataPipelineDestinationTypeTiDBLake,
			Snapshot: domain.DataPipelineSnapshotPlan{
				Engine: domain.DataPipelineSnapshotEngineDumpling,
				Format: domain.DataPipelineSnapshotFormatCSV,
			},
			Incremental: &domain.DataPipelineIncrementalPlan{
				Engine: domain.DataPipelineIncrementalEngineTiCDC,
				Format: domain.DataPipelineIncrementalFormatCanalJSON,
			},
		},
		IntervalPlan: domain.DeliveryIntervalPlan{
			Version:                      1,
			EndToEndSeconds:              10,
			ProducerFlushIntervalSeconds: 4,
			ConsumerPollIntervalSeconds:  6,
		},
		SnapshotTSO: testSnapshotTSO,
		StagingRoot: "customer/base/tidb-data-pipelines/staging/8001/9001/",
		Children: domain.DataPipelineChildResources{
			ExportName:         "tidb-dp-export-9001",
			ExportID:           "6002",
			ChangefeedName:     "tidb-dp-changefeed-9001",
			ChangefeedID:       6001,
			LakeDataSourceName: lakeNames(9001).DataSource,
			LakeTaskName:       lakeNames(9001).Task,
		},
	}
}

// lakeNames mirrors what the model settles onto the row at insert time.
func lakeNames(pipelineID uint64) lake.ResourceNames {
	names, err := lake.DeterministicResourceNames(strconv.FormatUint(pipelineID, 10))
	if err != nil {
		panic(err)
	}
	return names
}

func testCredentials() DecryptedCredentials {
	return DecryptedCredentials{
		TiDBPassword:      NewSecret(testTiDBPassword),
		S3SecretAccessKey: NewSecret(testS3SecretAccessKey),
	}
}

// The snapshot has to be RFC 4180 because that is the only dialect the
// destination reads: its CSV reader runs with no escape character, so a quote
// Dumpling escaped as \" ends the field and every comma after it splits the
// row - the load then fails on a column count that does not match the table.
//
// Sending the field is what matters, not just its value: the option is optional
// on the wire, and omitting it leaves Dumpling on its default, which escapes.
func TestBuildExportCreateRequestTurnsOffBackslashEscaping(t *testing.T) {
	t.Parallel()

	request, err := BuildExportCreateRequest(testPipeline(), testCredentials())
	require.NoError(t, err)
	options := request.Request().GetExport().GetExportOptions()

	require.NotNil(t, options.EscapeBackslash,
		"the option must be sent, or Dumpling keeps its escaping default")
	require.False(t, options.GetEscapeBackslash())

	// Doubling only happens when a delimiter is set; without one Dumpling
	// writes the value through untouched and a quote would break the row.
	require.Equal(t, `"`, options.GetCsvFormat().GetDelimiter())
}

func expectedExportCreateRequest(pipeline *domain.DataPipeline) *globalngv1.CreateExportRequest {
	return &globalngv1.CreateExportRequest{
		ClusterId: "8001",
		Export: &globalngv1.Export{
			ExportId:    "6002",
			DisplayName: "tidb-dp-export-9001",
			TidbId:      "8001",
			Username:    "test-source-user",
			Password:    testTiDBPassword,
			ExportOptions: &globalngv1.ExportOptions{
				FileType:    globalngv1.ExportOptions_CSV,
				Compression: globalngv1.ExportOptions_NONE,
				Filter: &globalngv1.ExportOptions_Filter{
					Type: &globalngv1.ExportOptions_Filter_Table_{
						Table: &globalngv1.ExportOptions_Filter_Table{
							Patterns: append([]string(nil), pipeline.Params.Source.TableFilter...),
						},
					},
				},
				CsvFormat: &globalngv1.ExportOptions_CSVFormat{
					Separator:  ",",
					Delimiter:  stringPointer(`"`),
					NullValue:  stringPointer(`\N`),
					SkipHeader: false,
					// Binary columns must be hex: the raw bytes a BLOB can hold
					// include the separator, the delimiter and newlines, none of
					// which a CSV reader can tell apart from structure.
					Dialect: globalngv1.ExportOptions_ExportCSVDialect_SNOWFLAKE.Enum(),
				},
				// Off, so a quote inside a field is doubled rather than
				// backslash-escaped. The destination reads RFC 4180 with no
				// escape character and would end the field at the quote.
				EscapeBackslash: proto.Bool(false),
			},
			CloudStorage: &globalngv1.CloudStorage{
				Type: globalngv1.CloudStorageType_S3,
				Storage: &globalngv1.CloudStorage_S3{
					S3: &globalngv1.S3CloudStorage{
						Uri:      "s3://test-pipeline-bucket/customer/base/tidb-data-pipelines/staging/8001/9001/snapshot/",
						AuthType: globalngv1.S3CloudStorage_ACCESS_KEY,
						Auth: &globalngv1.S3CloudStorage_AccessKey_{
							AccessKey: &globalngv1.S3CloudStorage_AccessKey{
								Id:     "test-access-key-id",
								Secret: testS3SecretAccessKey,
							},
						},
					},
				},
			},
		},
	}
}

func expectedChangefeedS3Info(pipeline *domain.DataPipeline) *dataflow.S3Info {
	return &dataflow.S3Info{
		Endpoint: &dataflow.S3Endpoint{
			S3Uri:     "s3://test-pipeline-bucket/customer/base/tidb-data-pipelines/staging/8001/9001/incremental/",
			AccessKey: "test-access-key-id",
			SecretKey: testS3SecretAccessKey,
			AuthType:  dataflow.S3AuthType_S3_AUTH_TYPE_AK_SK,
			Provider:  dataflow.S3Provider_S3_PROVIDER_AWS,
		},
		DataFormat: &dataflow.S3Dataformat{
			Protocol:            dataflow.S3Protocol_S3_PROTOCOL_CANAL_JSON,
			Terminator:          "\n",
			DateSeparator:       "none",
			EnableTidbExtension: true,
		},
		TableConfig: &dataflow.TableConfig{
			FilterRule:    append([]string(nil), pipeline.Params.Source.TableFilter...),
			Mode:          dataflow.TableMode_IGNORE_NOT_SUPPORT_TABLE,
			CaseSensitive: true,
		},
		StartPosition: &dataflow.StartPosition{
			Mode: dataflow.StartMode_FROM_TSO,
			Tso:  testSnapshotTSO,
		},
		StorageFlushConfig: &dataflow.StorageFlushConfig{
			FlushInterval: "4s",
			FileSize:      changefeedFileSizeBytes,
		},
	}
}

func assertCreateChangefeedRequestEqual(
	t *testing.T,
	want, got *domain.CreateBizChangefeedReq,
) {
	t.Helper()
	require.NotNil(t, got)
	require.Equal(t, want.Name, got.Name)
	require.Equal(t, want.ClusterID, got.ClusterID)
	require.Equal(t, want.DownstreamType, got.DownstreamType)
	require.Equal(t, want.NetworkInfo, got.NetworkInfo)
	require.Equal(t, want.SpecificationCode, got.SpecificationCode)
	require.Equal(t, want.UserID, got.UserID)
	require.Equal(t, want.PrivateLinkEndpointID, got.PrivateLinkEndpointID)
	require.True(t, proto.Equal(
		want.DownstreamParameter.(*dataflow.S3Info),
		got.DownstreamParameter.(*dataflow.S3Info),
	))
}

func assertEditChangefeedRequestEqual(
	t *testing.T,
	want, got *domain.ApplyStoppedBizChangefeedConfigReq,
) {
	t.Helper()
	require.NotNil(t, got)
	require.Equal(t, want.BizChangefeedID, got.BizChangefeedID)
	require.Equal(t, want.Name, got.Name)
	require.Equal(t, want.UserID, got.UserID)
	require.True(t, proto.Equal(
		want.DownstreamParameter.(*dataflow.S3Info),
		got.DownstreamParameter.(*dataflow.S3Info),
	))
}

func int32Pointer(value int32) *int32 {
	return &value
}

func assertSecretsAbsent(t *testing.T, value string) {
	t.Helper()
	for _, secret := range []string{
		testTiDBPassword,
		testS3SecretAccessKey,
		"test-access-key-id",
		"test-tidb-password-ciphertext",
		"test-s3-secret-ciphertext",
	} {
		require.NotContains(t, value, secret)
	}
}

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

	require.False(t, errors.Is(ErrInvalidPipeline, ErrUnsupportedExecutionPlan))
}

// expectedLakeTableRules pins that the user's filter reaches Lake unchanged.
//
// The fixture is deliberately the shape the previous translation could not
// carry: a wildcard table pattern and a second database. Both survive now, so a
// regression that reintroduces per-database flattening fails here.
func expectedLakeTableRules() lake.TableRules {
	return lake.TableRules{"test_app.*", "test_audit.events"}
}

// TestBuildersCarryRoleARNToEveryChild pins that the staging auth choice is
// made once and reaches all three children, each in its own shape. Getting one
// of them wrong would leave that child unable to read the bucket the other two
// can, which surfaces as a partial pipeline rather than a clean failure.
func TestBuildersCarryRoleARNToEveryChild(t *testing.T) {
	t.Parallel()
	const roleARN = "arn:aws:iam::000000000000:role/tidb-pipeline"

	roleARNPipeline := func() *domain.DataPipeline {
		pipeline := testPipeline()
		pipeline.Params.Staging.S3.AuthType = domain.DataPipelineS3AuthTypeRoleARN
		pipeline.Params.Staging.S3.RoleARN = roleARN
		pipeline.Params.Staging.S3.AccessKeyID = ""
		pipeline.Params.Staging.S3.SecretAccessKeyCiphertext = ""
		return pipeline
	}

	t.Run("Export takes a oneof selected by auth type", func(t *testing.T) {
		t.Parallel()
		// No S3 secret: a role-ARN Pipeline never stores one.
		built, err := BuildExportCreateRequest(roleARNPipeline(), DecryptedCredentials{
			TiDBPassword: NewSecret(testTiDBPassword),
		})
		require.NoError(t, err)
		s3 := built.Request().GetExport().GetCloudStorage().GetS3()
		require.Equal(t, globalngv1.S3CloudStorage_ROLE_ARN, s3.GetAuthType())
		require.Equal(t, roleARN, s3.GetRoleArn())
		require.Empty(t, s3.GetAccessKey().GetId())
		require.Empty(t, s3.GetAccessKey().GetSecret())
	})

	t.Run("TiCDC takes a flat endpoint with an assume-role auth type", func(t *testing.T) {
		t.Parallel()
		request, err := BuildChangefeedCreateRequest(roleARNPipeline(), Secret{})
		require.NoError(t, err)
		s3Info, ok := request.Request().DownstreamParameter.(*dataflow.S3Info)
		require.True(t, ok)
		endpoint := s3Info.GetEndpoint()
		require.Equal(t, dataflow.S3AuthType_S3_AUTH_TYPE_ASSUME_ROLE, endpoint.GetAuthType())
		require.Equal(t, roleARN, endpoint.GetIamAssumedRoleArn())
		require.Empty(t, endpoint.GetAccessKey())
		require.Empty(t, endpoint.GetSecretKey())
	})

	t.Run("Lake takes a role ARN and leaves the external ID to the org", func(t *testing.T) {
		t.Parallel()
		spec, err := BuildLakeDataSourceSpec(roleARNPipeline(), Secret{})
		require.NoError(t, err)
		require.Equal(t, roleARN, spec.S3.RoleARN)
		require.True(t, spec.S3.AccessKey.Empty())
		require.True(t, spec.S3.SecretKey.Empty())
		require.True(t, spec.S3.ExternalID.Empty(),
			"Lake fills the external ID with the org slug, which is what its trust policy expects")
	})

	t.Run("a role-ARN Pipeline needs no S3 secret", func(t *testing.T) {
		t.Parallel()
		pipeline := roleARNPipeline()
		pipeline.Params.Staging.S3.RoleARN = ""
		_, err := BuildLakeDataSourceSpec(pipeline, Secret{})
		require.ErrorIs(t, err, ErrInvalidPipeline, "but it does need the role")
	})
}

// The reason a child call failed has to reach the log, but a downstream that
// echoed its request back would carry a credential with it. This is what keeps
// the two compatible, so it is worth pinning that it actually replaces.
func TestRedactSecretsRemovesCredentialsFromADownstreamError(t *testing.T) {
	t.Parallel()

	const password = "test-source-password"
	const key = "test-s3-secret-access-key"
	message := "rpc error: rejected request {password: " + password +
		", secretKey: " + key + "}"

	redacted := RedactSecrets(message, NewSecret(password), NewSecret(key))
	require.NotContains(t, redacted, password)
	require.NotContains(t, redacted, key)
	require.Contains(t, redacted, "rpc error: rejected request")
	require.Equal(t, 2, strings.Count(redacted, "[REDACTED]"))

	t.Run("nothing to redact leaves the message intact", func(t *testing.T) {
		t.Parallel()
		require.Equal(t, message, RedactSecrets(message))
	})

	// A one or two character value is not a credential, and replacing it would
	// shred the message it appears in.
	t.Run("very short values are skipped", func(t *testing.T) {
		t.Parallel()
		require.Equal(t, "a b c", RedactSecrets("a b c", NewSecret("a"), NewSecret("")))
	})

	// An empty Secret is what a role-ARN Pipeline carries: there is no secret
	// to scrub, and it must not match every position in the string.
	t.Run("an empty secret changes nothing", func(t *testing.T) {
		t.Parallel()
		require.Equal(t, message, RedactSecrets(message, Secret{}))
	})
}