package console

import (
	"context"
	"strings"
	"testing"
	"time"

	"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/account/accountdb"
	"github.com/tidbcloud/dataflow-service-ng/pkg/account/accounttypes"
	"github.com/tidbcloud/dataflow-service-ng/pkg/grpc/grpcli"
	"github.com/tidbcloud/pkgv2/auth/auth0"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/logid"
	mgmt "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"google.golang.org/protobuf/encoding/protojson"
	"google.golang.org/protobuf/proto"
)

const (
	testDataPipelineUserID    = uint64(73)
	testDataPipelineClusterID = uint64(101)
	testDataPipelineID        = uint64(202)

	testSourcePassword           = "test-source-password"
	testS3SecretAccessKey        = "test-s3-secret-access-key"
	testSourcePasswordCiphertext = "enc:test-source-password-ciphertext"
	testS3SecretAccessCiphertext = "enc:test-s3-secret-access-key-ciphertext"
)

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

	model := &fakeDataPipelineModel{
		createResp: &domain.CreateDataPipelineResp{PipelineID: testDataPipelineID},
		getResp:    &domain.GetDataPipelineResp{DataPipeline: newCompleteDataPipelineDetails()},
		listResp: &domain.ListDataPipelinesResp{
			DataPipelines: []*domain.DataPipelineDetails{newCompleteDataPipelineDetails()},
			Total:         1,
		},
	}
	mgmtClient := &fakeDataPipelineMgmtClient{}
	controller := NewDataPipelineController(model, mgmtClient)
	ctx := newDataPipelineTestContext()

	createReq := newCreateDataPipelineRequest()
	createResp, err := controller.CreateDataPipeline(ctx, createReq)
	require.NoError(t, err)
	require.Equal(t, testDataPipelineClusterID, createResp.GetClusterId())
	require.Equal(t, testDataPipelineID, createResp.GetPipelineId())
	require.Equal(t, &domain.CreateDataPipelineReq{
		ClusterID: testDataPipelineClusterID,
		UserID:    testDataPipelineUserID,
		Name:      "test-pipeline",
		Source: domain.DataPipelineSourceInput{
			Username:      "test-source-user",
			Password:      testSourcePassword,
			TableFilter:   []string{"test_db_1.*", "test_db_2.test_table"},
			CaseSensitive: true,
		},
		Destination: domain.DataPipelineDestinationInput{
			Type:      domain.DataPipelineDestinationTypeTiDBLake,
			Warehouse: "test-warehouse",
			TargetNaming: domain.DataPipelineTargetNaming{
				DatabasePrefix: "mirror_${database}",
				TablePrefix:    "copy_${table}",
			},
			FlushIntervalSeconds: 90,
		},
		Staging: domain.DataPipelineStagingInput{
			Type: domain.DataPipelineStagingTypeCustomerManagedS3,
			S3: &domain.DataPipelineS3StagingInput{
				URI:      "s3://test-pipeline-bucket/customer/base",
				Region:   "us-west-2",
				AuthType: domain.DataPipelineS3AuthTypeAccessKey,

				AccessKeyID:     "test-access-key-id",
				SecretAccessKey: testS3SecretAccessKey,
				Endpoint:        "https://test-s3.example.invalid",
			},
		},
		SyncMode:                domain.DataPipelineSyncModeFullAndIncremental,
		ChangefeedSpecification: "test-changefeed-spec",
	}, model.createReq)

	getResp, err := controller.GetDataPipeline(ctx, &dataflow.DataPipelineID{
		ClusterId:  testDataPipelineClusterID,
		PipelineId: testDataPipelineID,
	})
	require.NoError(t, err)
	require.NotNil(t, getResp)
	require.Equal(t, &domain.GetDataPipelineReq{
		ClusterID:  testDataPipelineClusterID,
		PipelineID: testDataPipelineID,
		UserID:     testDataPipelineUserID,
	}, model.getReq)

	listResp, err := controller.ListDataPipelines(ctx, &dataflow.ListDataPipelinesRequest{
		ClusterId: testDataPipelineClusterID,
		Page:      2,
		PerPage:   25,
		Status:    dataflow.DataPipelineStatus_DATA_PIPELINE_STATUS_COMPLETED,
	})
	require.NoError(t, err)
	require.Len(t, listResp.GetItems(), 1)
	require.Equal(t, uint32(1), listResp.GetTotal())
	require.NotNil(t, model.listReq.Status)
	require.Equal(t, domain.DataPipelineStatusCompleted, *model.listReq.Status)
	require.Equal(t, uint32(2), model.listReq.Page)
	require.Equal(t, uint32(25), model.listReq.PerPage)
	require.Equal(t, testDataPipelineUserID, model.listReq.UserID)

	pipelineIDReq := &dataflow.DataPipelineID{
		ClusterId:  testDataPipelineClusterID,
		PipelineId: testDataPipelineID,
	}
	_, err = controller.PauseDataPipeline(ctx, pipelineIDReq)
	require.NoError(t, err)
	require.Equal(t, &domain.PauseDataPipelineReq{
		ClusterID: testDataPipelineClusterID, PipelineID: testDataPipelineID, UserID: testDataPipelineUserID,
	}, model.pauseReq)

	editReq := &dataflow.EditDataPipelineRequest{
		ClusterId:  testDataPipelineClusterID,
		PipelineId: testDataPipelineID,
		Name:       "test-pipeline-edited",
		Source: &dataflow.DataPipelineTableSelection{
			TableFilter:   []string{"test_db_3.*"},
			CaseSensitive: false,
		},
		Destination: &dataflow.DataPipelineEditableDestination{
			Warehouse: "test-warehouse-edited",
			TargetNaming: &dataflow.DataPipelineTargetNaming{
				DatabasePrefix: "edited_${database}",
				TablePrefix:    "edited_${table}",
			},
			FlushIntervalSeconds: 120,
		},
	}
	_, err = controller.EditDataPipeline(ctx, editReq)
	require.NoError(t, err)
	require.Equal(t, &domain.EditDataPipelineReq{
		ClusterID:  testDataPipelineClusterID,
		PipelineID: testDataPipelineID,
		UserID:     testDataPipelineUserID,
		Name:       "test-pipeline-edited",
		Source: domain.DataPipelineEditableSourceInput{
			TableFilter: []string{"test_db_3.*"},
		},
		Destination: domain.DataPipelineEditableDestinationInput{
			Warehouse: "test-warehouse-edited",
			TargetNaming: domain.DataPipelineTargetNaming{
				DatabasePrefix: "edited_${database}",
				TablePrefix:    "edited_${table}",
			},
			FlushIntervalSeconds: 120,
		},
	}, model.editReq)

	_, err = controller.ScaleDataPipeline(ctx, &dataflow.ScaleDataPipelineRequest{
		ClusterId:               testDataPipelineClusterID,
		PipelineId:              testDataPipelineID,
		ChangefeedSpecification: "test-changefeed-spec-scaled",
	})
	require.NoError(t, err)
	require.Equal(t, &domain.ScaleDataPipelineReq{
		ClusterID:               testDataPipelineClusterID,
		PipelineID:              testDataPipelineID,
		UserID:                  testDataPipelineUserID,
		ChangefeedSpecification: "test-changefeed-spec-scaled",
	}, model.scaleReq)

	_, err = controller.ResumeDataPipeline(ctx, pipelineIDReq)
	require.NoError(t, err)
	require.Equal(t, &domain.ResumeDataPipelineReq{
		ClusterID: testDataPipelineClusterID, PipelineID: testDataPipelineID, UserID: testDataPipelineUserID,
	}, model.resumeReq)

	_, err = controller.DeleteDataPipeline(ctx, pipelineIDReq)
	require.NoError(t, err)
	require.Equal(t, &domain.DeleteDataPipelineReq{
		ClusterID: testDataPipelineClusterID, PipelineID: testDataPipelineID, UserID: testDataPipelineUserID,
	}, model.deleteReq)

	_, err = controller.ValidateDataPipelineSourceConnection(
		ctx,
		&dataflow.ValidateDataPipelineSourceConnectionRequest{
			ClusterId: testDataPipelineClusterID,
			Username:  "test-validation-user",
			Password:  "test-validation-password",
		},
	)
	require.NoError(t, err)
	require.Equal(t, &domain.ValidateDataPipelineSourceConnectionReq{
		ClusterID: testDataPipelineClusterID,
		UserID:    testDataPipelineUserID,
		Username:  "test-validation-user",
		Password:  "test-validation-password",
	}, model.validateSourceReq)

	_, err = controller.ValidateDataPipelineStaging(ctx, &dataflow.ValidateDataPipelineStagingRequest{
		ClusterId: testDataPipelineClusterID,
		Staging:   createReq.GetStaging(),
	})
	require.NoError(t, err)
	require.Equal(t, &domain.ValidateDataPipelineStagingReq{
		ClusterID: testDataPipelineClusterID,
		UserID:    testDataPipelineUserID,
		Staging: domain.DataPipelineStagingInput{
			Type: domain.DataPipelineStagingTypeCustomerManagedS3,
			S3: &domain.DataPipelineS3StagingInput{
				URI:      "s3://test-pipeline-bucket/customer/base",
				Region:   "us-west-2",
				AuthType: domain.DataPipelineS3AuthTypeAccessKey,

				AccessKeyID:     "test-access-key-id",
				SecretAccessKey: testS3SecretAccessKey,
				Endpoint:        "https://test-s3.example.invalid",
			},
		},
	}, model.validateStagingReq)

	// Every write and validation endpoint resolves the cluster first; Get and
	// List do not, because the model scopes them itself.
	require.Equal(t, []uint64{
		testDataPipelineClusterID,
		testDataPipelineClusterID,
		testDataPipelineClusterID,
		testDataPipelineClusterID,
		testDataPipelineClusterID,
		testDataPipelineClusterID,
		testDataPipelineClusterID,
		testDataPipelineClusterID,
	}, mgmtClient.clusterIDs)
	require.Len(t, model.traceIDs, 10)
	for _, traceID := range model.traceIDs {
		require.NotEmpty(t, traceID)
	}
}

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

	details := newCompleteDataPipelineDetails()
	model := &fakeDataPipelineModel{
		getResp: &domain.GetDataPipelineResp{DataPipeline: details},
		listResp: &domain.ListDataPipelinesResp{
			DataPipelines: []*domain.DataPipelineDetails{details},
			Total:         1,
		},
	}
	controller := NewDataPipelineController(model, &fakeDataPipelineMgmtClient{})
	ctx := newDataPipelineTestContext()

	getResp, err := controller.GetDataPipeline(ctx, &dataflow.DataPipelineID{
		ClusterId:  testDataPipelineClusterID,
		PipelineId: testDataPipelineID,
	})
	require.NoError(t, err)
	assertCompleteDataPipelineResponse(t, getResp)

	listResp, err := controller.ListDataPipelines(ctx, &dataflow.ListDataPipelinesRequest{
		ClusterId: testDataPipelineClusterID,
		Page:      1,
		PerPage:   10,
	})
	require.NoError(t, err)
	require.Len(t, listResp.GetItems(), 1)
	require.True(t, proto.Equal(getResp, listResp.GetItems()[0]))

	for _, response := range []proto.Message{getResp, listResp} {
		encoded, marshalErr := protojson.Marshal(response)
		require.NoError(t, marshalErr)
		payload := string(encoded)
		for _, secret := range []string{
			testSourcePassword,
			testS3SecretAccessKey,
			testSourcePasswordCiphertext,
			testS3SecretAccessCiphertext,
		} {
			require.NotContains(t, payload, secret)
		}
		require.NotContains(t, strings.ToLower(payload), "password")
		require.NotContains(t, strings.ToLower(payload), "secretaccesskey")
	}
}

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

	details := newCompleteDataPipelineDetails()
	details.Pipeline.CreatedAt = time.Time{}
	details.Pipeline.UpdatedAt = nil
	details.Observation.ChangefeedCheckpointTime = nil
	details.Observation.LakeResolvedTime = nil
	details.Observation.ObservedAt = nil
	model := &fakeDataPipelineModel{
		getResp: &domain.GetDataPipelineResp{DataPipeline: details},
	}
	controller := NewDataPipelineController(model, &fakeDataPipelineMgmtClient{})

	resp, err := controller.GetDataPipeline(newDataPipelineTestContext(), &dataflow.DataPipelineID{
		ClusterId:  testDataPipelineClusterID,
		PipelineId: testDataPipelineID,
	})
	require.NoError(t, err)
	require.Nil(t, resp.GetCreatedAt())
	require.Nil(t, resp.GetUpdatedAt())
	require.Nil(t, resp.GetObservation().GetChangefeedCheckpointTime())
	require.Nil(t, resp.GetObservation().GetLakeResolvedTime())
	require.Nil(t, resp.GetObservation().GetObservedAt())
}

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

	model := &fakeDataPipelineModel{createResp: &domain.CreateDataPipelineResp{PipelineID: testDataPipelineID}}
	lookupErr := bizerr.New(bizerr.ErrCodeRecordNotFound, "test cluster not found")
	controller := NewDataPipelineController(model, &fakeDataPipelineMgmtClient{err: lookupErr})

	resp, err := controller.CreateDataPipeline(newDataPipelineTestContext(), newCreateDataPipelineRequest())
	require.Nil(t, resp)
	require.Error(t, err)
	require.Nil(t, model.createReq)
}

// TestDataPipelineStatusConversionRoundTripsEveryValue drives both directions
// from the generated enum, so a status added to the proto without a conversion
// fails here instead of silently becoming UNSPECIFIED on read or, worse, a nil
// list filter that returns every pipeline.
func TestDataPipelineStatusConversionRoundTripsEveryValue(t *testing.T) {
	t.Parallel()

	for value, name := range dataflow.DataPipelineStatus_name {
		status := dataflow.DataPipelineStatus(value)
		if status == dataflow.DataPipelineStatus_DATA_PIPELINE_STATUS_UNSPECIFIED {
			require.Nil(t, dataPipelineStatusDataflow2Domain(status),
				"UNSPECIFIED is the no-filter sentinel")
			continue
		}
		inbound := dataPipelineStatusDataflow2Domain(status)
		require.NotNil(t, inbound, "%s has no inbound conversion", name)
		require.Equal(t, status, dataPipelineStatusDomain2Dataflow(*inbound),
			"%s does not round trip", name)
	}
}

// TestDataPipelinePhaseCoversEveryStatus pins that no reachable status leaves
// the Console without a phase to render.
func TestDataPipelinePhaseCoversEveryStatus(t *testing.T) {
	t.Parallel()

	statuses := []domain.DataPipelineStatus{
		domain.DataPipelineStatusCreating, domain.DataPipelineStatusRunning,
		domain.DataPipelineStatusPausing, domain.DataPipelineStatusPaused,
		domain.DataPipelineStatusEditing, domain.DataPipelineStatusResuming,
		domain.DataPipelineStatusDeleting, domain.DataPipelineStatusFailed,
		domain.DataPipelineStatusCompleted,
	}
	for _, status := range statuses {
		pipeline := &domain.DataPipeline{Status: status}
		require.NotEqual(t,
			dataflow.DataPipelinePhase_DATA_PIPELINE_PHASE_UNSPECIFIED,
			dataPipelinePhaseDomain2Dataflow(pipeline),
			"status %s has no phase", status)
	}

	// Running splits on the destination's own progress, which is the one
	// distinction the Pipeline cannot make from its own state.
	loading := &domain.DataPipeline{Status: domain.DataPipelineStatusRunning}
	require.Equal(t,
		dataflow.DataPipelinePhase_DATA_PIPELINE_PHASE_INITIAL_LOADING,
		dataPipelinePhaseDomain2Dataflow(loading))
	loading.Observation.LakeFullSyncCompleted = true
	require.Equal(t,
		dataflow.DataPipelinePhase_DATA_PIPELINE_PHASE_STREAMING,
		dataPipelinePhaseDomain2Dataflow(loading))
}

func assertCompleteDataPipelineResponse(t *testing.T, resp *dataflow.DataPipeline) {
	t.Helper()

	require.Equal(t, testDataPipelineID, resp.GetPipelineId())
	require.Equal(t, testDataPipelineClusterID, resp.GetClusterId())
	require.Equal(t, "test-pipeline", resp.GetName())
	require.Equal(t, "test-source-user", resp.GetSource().GetUsername())
	require.Equal(t, []string{"test_db_1.*", "test_db_2.test_table"}, resp.GetSource().GetTableFilter())
	require.True(t, resp.GetSource().GetCaseSensitive())
	require.Equal(t, dataflow.DataPipelineDestinationType_DATA_PIPELINE_DESTINATION_TYPE_TIDB_LAKE, resp.GetDestination().GetType())
	require.Equal(t, "test-warehouse", resp.GetDestination().GetWarehouse())
	require.Equal(t, "mirror_${database}", resp.GetDestination().GetTargetNaming().GetDatabasePrefix())
	require.Equal(t, "copy_${table}", resp.GetDestination().GetTargetNaming().GetTablePrefix())
	require.Equal(t, uint32(90), resp.GetDestination().GetFlushIntervalSeconds())
	require.Equal(t, dataflow.DataPipelineStagingType_DATA_PIPELINE_STAGING_TYPE_CUSTOMER_MANAGED_S3, resp.GetStaging().GetType())
	require.Equal(t, "s3://test-pipeline-bucket/customer/base", resp.GetStaging().GetS3().GetUri())
	require.Equal(t, "us-west-2", resp.GetStaging().GetS3().GetRegion())
	require.Equal(t, "test-access-key-id", resp.GetStaging().GetS3().GetAccessKeyId())
	require.Equal(t, "https://test-s3.example.invalid", resp.GetStaging().GetS3().GetEndpoint())
	require.Equal(t, "s3://test-pipeline-bucket/customer/base/tidb-data-pipelines/test/101/202/snapshot/", resp.GetStaging().GetS3().GetSnapshotUri())
	require.Equal(t, "s3://test-pipeline-bucket/customer/base/tidb-data-pipelines/test/101/202/incremental/", resp.GetStaging().GetS3().GetIncrementalUri())
	require.Equal(t, dataflow.DataPipelineSyncMode_DATA_PIPELINE_SYNC_MODE_FULL_AND_INCREMENTAL, resp.GetSyncMode())
	require.Equal(t, "test-changefeed-spec", resp.GetChangefeedSpecification())
	require.Equal(t, dataflow.DataPipelineStatus_DATA_PIPELINE_STATUS_CREATING, resp.GetStatus())
	// Nothing has been observed yet, so Creating reports the earliest phase.
	require.Equal(t, dataflow.DataPipelinePhase_DATA_PIPELINE_PHASE_PREPARING, resp.GetPhase())
	require.Equal(t, dataflow.DataPipelineSnapshotEngine_DATA_PIPELINE_SNAPSHOT_ENGINE_DUMPLING, resp.GetExecutionPlan().GetSnapshotEngine())
	require.Equal(t, dataflow.DataPipelineSnapshotFormat_DATA_PIPELINE_SNAPSHOT_FORMAT_CSV, resp.GetExecutionPlan().GetSnapshotFormat())
	require.Equal(t, dataflow.DataPipelineIncrementalEngine_DATA_PIPELINE_INCREMENTAL_ENGINE_TICDC, resp.GetExecutionPlan().GetIncrementalEngine())
	require.Equal(t, dataflow.DataPipelineIncrementalFormat_DATA_PIPELINE_INCREMENTAL_FORMAT_CANAL_JSON, resp.GetExecutionPlan().GetIncrementalFormat())
	require.Equal(t, uint64(123456789<<18), resp.GetExecutionPlan().GetSnapshotTso())
	require.Equal(t, uint32(90), resp.GetDeliveryIntervalPlan().GetEndToEndSeconds())
	require.Equal(t, uint32(30), resp.GetDeliveryIntervalPlan().GetProducerFlushIntervalSeconds())
	require.Equal(t, uint32(60), resp.GetDeliveryIntervalPlan().GetConsumerPollIntervalSeconds())
	require.Equal(t, "test-export-id", resp.GetChildResources().GetExport().GetExportId())
	require.Equal(t, "SUCCEEDED", resp.GetChildResources().GetExport().GetStatus())
	require.Equal(t, "test export warning", resp.GetChildResources().GetExport().GetErrorMessage())
	require.Equal(t, uint64(303), resp.GetChildResources().GetChangefeed().GetChangefeedId())
	require.Equal(t, "NORMAL", resp.GetChildResources().GetChangefeed().GetStatus())
	require.Equal(t, "test changefeed warning", resp.GetChildResources().GetChangefeed().GetErrorMessage())
	require.Equal(t, "test-datasource-name", resp.GetChildResources().GetLake().GetDatasourceName())
	require.Equal(t, "Ready", resp.GetChildResources().GetLake().GetDatasourceStatus())
	require.Equal(t, "test-integration-name", resp.GetChildResources().GetLake().GetIntegrationName())
	require.Equal(t, "TaskRunning", resp.GetChildResources().GetLake().GetIntegrationStatus())
	require.Equal(t, "test lake warning", resp.GetChildResources().GetLake().GetErrorMessage())
	require.Equal(t, uint64(123456999<<18), resp.GetObservation().GetChangefeedCheckpointTso())
	require.Equal(t, uint64(123456900<<18), resp.GetObservation().GetLakeResolvedTso())
	require.Equal(t, uint64(99), resp.GetObservation().GetLatencySeconds())
	require.True(t, resp.GetObservation().GetStale())
	require.Equal(t, "test observation error", resp.GetObservation().GetObservationError())
	require.Equal(t, "TEST_PIPELINE_ERROR", resp.GetError().GetCode())
	require.Equal(t, "test lifecycle error", resp.GetError().GetMessage())

	expected := dataPipelineTestTimes()
	require.True(t, resp.GetCreatedAt().AsTime().Equal(expected.createdAt))
	require.True(t, resp.GetUpdatedAt().AsTime().Equal(expected.updatedAt))
	require.True(t, resp.GetObservation().GetChangefeedCheckpointTime().AsTime().Equal(expected.checkpointAt))
	require.True(t, resp.GetObservation().GetLakeResolvedTime().AsTime().Equal(expected.lakeResolvedAt))
	require.True(t, resp.GetObservation().GetObservedAt().AsTime().Equal(expected.observedAt))
}

func newCreateDataPipelineRequest() *dataflow.CreateDataPipelineRequest {
	return &dataflow.CreateDataPipelineRequest{
		ClusterId: testDataPipelineClusterID,
		Name:      "test-pipeline",
		Source: &dataflow.DataPipelineSourceInput{
			Username:      "test-source-user",
			Password:      testSourcePassword,
			TableFilter:   []string{"test_db_1.*", "test_db_2.test_table"},
			CaseSensitive: true,
		},
		Destination: &dataflow.DataPipelineDestination{
			Type:      dataflow.DataPipelineDestinationType_DATA_PIPELINE_DESTINATION_TYPE_TIDB_LAKE,
			Warehouse: "test-warehouse",
			TargetNaming: &dataflow.DataPipelineTargetNaming{
				DatabasePrefix: "mirror_${database}",
				TablePrefix:    "copy_${table}",
			},
			FlushIntervalSeconds: 90,
		},
		Staging: &dataflow.DataPipelineStagingInput{
			Type: dataflow.DataPipelineStagingType_DATA_PIPELINE_STAGING_TYPE_CUSTOMER_MANAGED_S3,
			S3: &dataflow.DataPipelineS3StagingInput{
				Uri:             "s3://test-pipeline-bucket/customer/base",
				Region:          "us-west-2",
				AuthType:        dataflow.DataPipelineS3AuthType_DATA_PIPELINE_S3_AUTH_TYPE_ACCESS_KEY,
				AccessKeyId:     "test-access-key-id",
				SecretAccessKey: testS3SecretAccessKey,
				Endpoint:        "https://test-s3.example.invalid",
			},
		},
		SyncMode:                dataflow.DataPipelineSyncMode_DATA_PIPELINE_SYNC_MODE_FULL_AND_INCREMENTAL,
		ChangefeedSpecification: "test-changefeed-spec",
	}
}

type dataPipelineTimes struct {
	createdAt      time.Time
	updatedAt      time.Time
	checkpointAt   time.Time
	lakeResolvedAt time.Time
	observedAt     time.Time
}

func dataPipelineTestTimes() dataPipelineTimes {
	return dataPipelineTimes{
		createdAt:      time.Date(2026, time.August, 2, 1, 2, 3, 400_000_000, time.UTC),
		updatedAt:      time.Date(2026, time.August, 2, 2, 3, 4, 500_000_000, time.UTC),
		checkpointAt:   time.Date(2026, time.August, 2, 3, 4, 5, 600_000_000, time.UTC),
		lakeResolvedAt: time.Date(2026, time.August, 2, 3, 2, 26, 600_000_000, time.UTC),
		observedAt:     time.Date(2026, time.August, 2, 3, 4, 35, 700_000_000, time.UTC),
	}
}

func newCompleteDataPipelineDetails() *domain.DataPipelineDetails {
	times := dataPipelineTestTimes()
	incrementalPlan := &domain.DataPipelineIncrementalPlan{
		Engine: domain.DataPipelineIncrementalEngineTiCDC,
		Format: domain.DataPipelineIncrementalFormatCanalJSON,
	}
	return &domain.DataPipelineDetails{
		Pipeline: &domain.DataPipeline{
			ID:                      testDataPipelineID,
			TenantID:                404,
			ClusterID:               testDataPipelineClusterID,
			CreatorID:               testDataPipelineUserID,
			Name:                    "test-pipeline",
			Status:                  domain.DataPipelineStatusCreating,
			ChangefeedSpecification: "test-changefeed-spec",
			Params: domain.DataPipelineParams{
				Version:  1,
				SyncMode: domain.DataPipelineSyncModeFullAndIncremental,
				Source: domain.DataPipelineSourceConfig{
					Username:           "test-source-user",
					PasswordCiphertext: testSourcePasswordCiphertext,
					TableFilter:        []string{"test_db_1.*", "test_db_2.test_table"},
					CaseSensitive:      true,
				},
				Staging: domain.DataPipelineStagingConfig{
					Type: domain.DataPipelineStagingTypeCustomerManagedS3,
					S3: domain.DataPipelineS3Config{
						Bucket:                    "test-pipeline-bucket",
						BasePrefix:                "customer/base",
						Region:                    "us-west-2",
						Endpoint:                  "https://test-s3.example.invalid",
						AccessKeyID:               "test-access-key-id",
						SecretAccessKeyCiphertext: testS3SecretAccessCiphertext,
					},
				},
				Destination: domain.DataPipelineDestinationConfig{
					Type:      domain.DataPipelineDestinationTypeTiDBLake,
					Warehouse: "test-warehouse",
					TargetNaming: domain.DataPipelineTargetNaming{
						DatabasePrefix: "mirror_${database}",
						TablePrefix:    "copy_${table}",
					},
				},
			},
			ExecutionPlan: domain.DataPipelineExecutionPlan{
				Version:         1,
				DestinationType: domain.DataPipelineDestinationTypeTiDBLake,
				Snapshot: domain.DataPipelineSnapshotPlan{
					Engine: domain.DataPipelineSnapshotEngineDumpling,
					Format: domain.DataPipelineSnapshotFormatCSV,
				},
				Incremental: incrementalPlan,
			},
			IntervalPlan: domain.DeliveryIntervalPlan{
				Version:                      1,
				EndToEndSeconds:              90,
				ProducerFlushIntervalSeconds: 30,
				ConsumerPollIntervalSeconds:  60,
			},
			SnapshotTSO: 123456789 << 18,
			StagingRoot: "customer/base/tidb-data-pipelines/test/101/202/",
			Children: domain.DataPipelineChildResources{
				ExportID:           "test-export-id",
				ChangefeedID:       303,
				LakeDataSourceName: "test-datasource-name",
				LakeTaskName:       "test-integration-name",
			},
			LastError: &domain.DataPipelineError{
				Code:      "TEST_PIPELINE_ERROR",
				Message:   "test lifecycle error",
				Retryable: true,
			},
			CreatedAt: times.createdAt,
			UpdatedAt: &times.updatedAt,
			Version:   2,
		},
		ChildResources: domain.DataPipelineChildResourceDetails{
			Export: domain.DataPipelineResourceDetails{
				Status:       "SUCCEEDED",
				ErrorMessage: "test export warning",
			},
			Changefeed: domain.DataPipelineResourceDetails{
				Status:       "NORMAL",
				ErrorMessage: "test changefeed warning",
			},
			Lake: domain.DataPipelineLakeResourceDetails{
				DataSourceStatus:  "Ready",
				IntegrationStatus: "TaskRunning",
				ErrorMessage:      "test lake warning",
			},
		},
		Observation: domain.DataPipelineObservationDetails{
			ChangefeedCheckpointTSO:  123456999 << 18,
			ChangefeedCheckpointTime: &times.checkpointAt,
			LakeResolvedTSO:          123456900 << 18,
			LakeResolvedTime:         &times.lakeResolvedAt,
			LatencySeconds:           99,
			ObservedAt:               &times.observedAt,
			Stale:                    true,
			Error: &domain.DataPipelineError{
				Code:    "TEST_OBSERVATION_ERROR",
				Message: "test observation error",
			},
		},
	}
}

func newDataPipelineTestContext() context.Context {
	return context.WithValue(context.Background(), auth0.AuthTokenClaimKey, &accountdb.User{
		User: accounttypes.User{
			Updatable: accounttypes.Updatable{Base: accounttypes.Base{ID: testDataPipelineUserID}},
		},
	})
}

type fakeDataPipelineMgmtClient struct {
	grpcli.MgmtSvcGrpcClient
	clusterIDs []uint64
	err        bizerr.BizError
}

func (f *fakeDataPipelineMgmtClient) GetClusterByID(
	_ context.Context,
	clusterID uint64,
) (*mgmt.Cluster, bizerr.BizError) {
	f.clusterIDs = append(f.clusterIDs, clusterID)
	if f.err != nil {
		return nil, f.err
	}
	return &mgmt.Cluster{ClusterId: clusterID}, nil
}

type fakeDataPipelineModel struct {
	traceIDs []string

	createReq  *domain.CreateDataPipelineReq
	createResp *domain.CreateDataPipelineResp
	createErr  bizerr.BizError

	getReq  *domain.GetDataPipelineReq
	getResp *domain.GetDataPipelineResp
	getErr  bizerr.BizError

	listReq  *domain.ListDataPipelinesReq
	listResp *domain.ListDataPipelinesResp
	listErr  bizerr.BizError

	pauseReq *domain.PauseDataPipelineReq
	pauseErr bizerr.BizError

	editReq *domain.EditDataPipelineReq
	editErr bizerr.BizError

	scaleReq *domain.ScaleDataPipelineReq
	scaleErr bizerr.BizError

	validateDestinationReq *domain.ValidateDataPipelineDestinationReq
	validateDestinationErr bizerr.BizError

	stagingTrustReq  *domain.GetDataPipelineStagingTrustReq
	stagingTrustResp *domain.GetDataPipelineStagingTrustResp
	stagingTrustErr  bizerr.BizError

	resumeReq *domain.ResumeDataPipelineReq
	resumeErr bizerr.BizError

	deleteReq *domain.DeleteDataPipelineReq
	deleteErr bizerr.BizError

	validateSourceReq *domain.ValidateDataPipelineSourceConnectionReq
	validateSourceErr bizerr.BizError

	validateStagingReq *domain.ValidateDataPipelineStagingReq
	validateStagingErr bizerr.BizError
}

func (f *fakeDataPipelineModel) recordContext(ctx context.Context) {
	f.traceIDs = append(f.traceIDs, logid.GetLogid(ctx))
}

func (f *fakeDataPipelineModel) CreateDataPipeline(
	ctx context.Context,
	req *domain.CreateDataPipelineReq,
) (*domain.CreateDataPipelineResp, bizerr.BizError) {
	f.recordContext(ctx)
	f.createReq = req
	return f.createResp, f.createErr
}

func (f *fakeDataPipelineModel) GetDataPipeline(
	ctx context.Context,
	req *domain.GetDataPipelineReq,
) (*domain.GetDataPipelineResp, bizerr.BizError) {
	f.recordContext(ctx)
	f.getReq = req
	return f.getResp, f.getErr
}

func (f *fakeDataPipelineModel) ListDataPipelines(
	ctx context.Context,
	req *domain.ListDataPipelinesReq,
) (*domain.ListDataPipelinesResp, bizerr.BizError) {
	f.recordContext(ctx)
	f.listReq = req
	return f.listResp, f.listErr
}

func (f *fakeDataPipelineModel) PauseDataPipeline(
	ctx context.Context,
	req *domain.PauseDataPipelineReq,
) bizerr.BizError {
	f.recordContext(ctx)
	f.pauseReq = req
	return f.pauseErr
}

func (f *fakeDataPipelineModel) EditDataPipeline(
	ctx context.Context,
	req *domain.EditDataPipelineReq,
) bizerr.BizError {
	f.recordContext(ctx)
	f.editReq = req
	return f.editErr
}

func (f *fakeDataPipelineModel) ScaleDataPipeline(
	ctx context.Context,
	req *domain.ScaleDataPipelineReq,
) bizerr.BizError {
	f.recordContext(ctx)
	f.scaleReq = req
	return f.scaleErr
}

func (f *fakeDataPipelineModel) GetDataPipelineStagingTrust(
	ctx context.Context,
	req *domain.GetDataPipelineStagingTrustReq,
) (*domain.GetDataPipelineStagingTrustResp, bizerr.BizError) {
	f.recordContext(ctx)
	f.stagingTrustReq = req
	return f.stagingTrustResp, f.stagingTrustErr
}

func (f *fakeDataPipelineModel) ValidateDataPipelineDestination(
	ctx context.Context,
	req *domain.ValidateDataPipelineDestinationReq,
) bizerr.BizError {
	f.recordContext(ctx)
	f.validateDestinationReq = req
	return f.validateDestinationErr
}

func (f *fakeDataPipelineModel) ResumeDataPipeline(
	ctx context.Context,
	req *domain.ResumeDataPipelineReq,
) bizerr.BizError {
	f.recordContext(ctx)
	f.resumeReq = req
	return f.resumeErr
}

func (f *fakeDataPipelineModel) DeleteDataPipeline(
	ctx context.Context,
	req *domain.DeleteDataPipelineReq,
) bizerr.BizError {
	f.recordContext(ctx)
	f.deleteReq = req
	return f.deleteErr
}

func (f *fakeDataPipelineModel) ValidateDataPipelineSourceConnection(
	ctx context.Context,
	req *domain.ValidateDataPipelineSourceConnectionReq,
) bizerr.BizError {
	f.recordContext(ctx)
	f.validateSourceReq = req
	return f.validateSourceErr
}

func (f *fakeDataPipelineModel) ValidateDataPipelineStaging(
	ctx context.Context,
	req *domain.ValidateDataPipelineStagingReq,
) bizerr.BizError {
	f.recordContext(ctx)
	f.validateStagingReq = req
	return f.validateStagingErr
}

// The notification queue crosses the API boundary in three places, and each
// one is a different message: it arrives on Create, it arrives again on Edit
// through the one editable part of the staging area, and it is reported back
// so a reader can tell a notified Pipeline from a polling one without
// inferring it from the delivery interval.
func TestDataPipelineControllerCarriesTheNotificationQueueBothWays(t *testing.T) {
	t.Parallel()

	const queue = "https://sqs.us-west-2.amazonaws.com/000000000000/test-queue.fifo"

	edited := editDataPipelineRequestDataflow2Domain(&dataflow.EditDataPipelineRequest{
		ClusterId:   testDataPipelineClusterID,
		PipelineId:  testDataPipelineID,
		Name:        "test-pipeline",
		Source:      &dataflow.DataPipelineTableSelection{},
		Destination: &dataflow.DataPipelineEditableDestination{},
		Staging: &dataflow.DataPipelineEditableStaging{
			SqsQueueUrl: queue,
			SqsRegion:   "us-west-2",
		},
	}, testDataPipelineUserID)
	require.Equal(t, queue, edited.Staging.SQSQueueURL)
	require.Equal(t, "us-west-2", edited.Staging.SQSRegion)

	// Absent detaches, like every other absent field in a full-replacement
	// payload. It must not read as "leave whatever is there".
	detached := editDataPipelineRequestDataflow2Domain(&dataflow.EditDataPipelineRequest{
		ClusterId:   testDataPipelineClusterID,
		PipelineId:  testDataPipelineID,
		Name:        "test-pipeline",
		Source:      &dataflow.DataPipelineTableSelection{},
		Destination: &dataflow.DataPipelineEditableDestination{},
	}, testDataPipelineUserID)
	require.Empty(t, detached.Staging.SQSQueueURL)
	require.Empty(t, detached.Staging.SQSRegion)

	created := dataPipelineStagingInputDataflow2Domain(&dataflow.DataPipelineStagingInput{
		Type: dataflow.DataPipelineStagingType_DATA_PIPELINE_STAGING_TYPE_CUSTOMER_MANAGED_S3,
		S3: &dataflow.DataPipelineS3StagingInput{
			Uri: "s3://test-bucket/base", Region: "us-west-2",
			SqsQueueUrl: queue, SqsRegion: "us-west-2",
		},
	})
	require.Equal(t, queue, created.S3.SQSQueueURL)
	require.Equal(t, "us-west-2", created.S3.SQSRegion)

	details := newCompleteDataPipelineDetails()
	details.Pipeline.Params.Staging.S3.SQSQueueURL = queue
	details.Pipeline.Params.Staging.S3.SQSRegion = "us-west-2"
	model := &fakeDataPipelineModel{getResp: &domain.GetDataPipelineResp{DataPipeline: details}}
	controller := NewDataPipelineController(model, &fakeDataPipelineMgmtClient{})
	read, err := controller.GetDataPipeline(newDataPipelineTestContext(), &dataflow.DataPipelineID{
		ClusterId: testDataPipelineClusterID, PipelineId: testDataPipelineID,
	})
	require.NoError(t, err)
	require.Equal(t, queue, read.GetStaging().GetS3().GetSqsQueueUrl())
	require.Equal(t, "us-west-2", read.GetStaging().GetS3().GetSqsRegion())
}

// The trust answer carries the actions as well as the identities. Knowing
// which roles to trust is only half of a working policy, and the other half
// fails later and less legibly.
func TestDataPipelineControllerReportsTheRequiredStagingActions(t *testing.T) {
	t.Parallel()

	model := &fakeDataPipelineModel{
		stagingTrustResp: &domain.GetDataPipelineStagingTrustResp{
			Principals: []domain.DataPipelineStagingTrustPrincipal{
				{Name: "test", RoleARN: "arn:aws:iam::000000000000:role/test", ExternalID: "x"},
			},
			BucketActions: []string{"s3:GetObject", "s3:PutObject"},
			QueueActions:  []string{"sqs:ReceiveMessage", "sqs:DeleteMessage"},
		},
	}
	controller := NewDataPipelineController(model, &fakeDataPipelineMgmtClient{})
	resp, err := controller.GetDataPipelineStagingTrust(
		newDataPipelineTestContext(),
		&dataflow.GetDataPipelineStagingTrustRequest{ClusterId: testDataPipelineClusterID},
	)
	require.NoError(t, err)
	require.Equal(t, []string{"s3:GetObject", "s3:PutObject"}, resp.GetBucketActions())
	require.Equal(t, []string{"sqs:ReceiveMessage", "sqs:DeleteMessage"}, resp.GetQueueActions())
}

// The Lake selection crosses the boundary in three shapes: it arrives with the
// destination on Create, it arrives on its own on the two endpoints that reach
// Lake before a Pipeline exists, and it is reported back so a reader can tell
// which Lake a Pipeline feeds.
func TestDataPipelineControllerCarriesTheDestinationLake(t *testing.T) {
	t.Parallel()

	in := dataPipelineDestinationInputDataflow2Domain(&dataflow.DataPipelineDestination{
		Type:      dataflow.DataPipelineDestinationType_DATA_PIPELINE_DESTINATION_TYPE_TIDB_LAKE,
		LakeOrgId: "7",
		Warehouse: "test-warehouse",
	})
	require.Equal(t, "7", in.LakeOrgID)

	details := newCompleteDataPipelineDetails()
	details.Pipeline.Params.Destination.LakeOrgID = "7"
	model := &fakeDataPipelineModel{getResp: &domain.GetDataPipelineResp{DataPipeline: details}}
	controller := NewDataPipelineController(model, &fakeDataPipelineMgmtClient{})
	read, err := controller.GetDataPipeline(newDataPipelineTestContext(), &dataflow.DataPipelineID{
		ClusterId: testDataPipelineClusterID, PipelineId: testDataPipelineID,
	})
	require.NoError(t, err)
	require.Equal(t, "7", read.GetDestination().GetLakeOrgId())

	// The two pre-creation endpoints have no destination to carry it, so they
	// take it directly - and it has to reach the model, not be dropped in
	// conversion.
	_, err = controller.GetDataPipelineStagingTrust(newDataPipelineTestContext(),
		&dataflow.GetDataPipelineStagingTrustRequest{
			ClusterId: testDataPipelineClusterID, LakeOrgId: "7",
		})
	require.NoError(t, err)
	require.Equal(t, "7", model.stagingTrustReq.LakeOrgID)
}