package planner

import (
	"errors"
	"testing"

	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/dataflow-service-ng/app/models/domain"
)

// A rejection has to be actionable on its own. The Console sends the enum
// number, so a message naming only "tidb_managed" reads to it as a value it
// never sent - that happened on staging and cost a round trip through the
// trace id to resolve. Each rejection therefore names the request field, the
// value that arrived, and what to send instead.
func TestSelectExecutionPlanRejectionsNameTheFieldAndTheWayOut(t *testing.T) {
	t.Parallel()

	t.Run("reserved staging says it is not built rather than wrong", func(t *testing.T) {
		t.Parallel()
		_, err := SelectExecutionPlan(
			domain.DataPipelineSyncModeFull,
			domain.DataPipelineDestinationTypeTiDBLake,
			domain.DataPipelineStagingTypeTiDBManaged,
		)
		require.ErrorIs(t, err, ErrUnsupportedStaging)
		require.Contains(t, err.Error(), "staging.type")
		require.Contains(t, err.Error(), "does not exist yet")
		require.Contains(t, err.Error(), "staging.s3")
		require.Contains(t, err.Error(), string(domain.DataPipelineStagingTypeCustomerManagedS3))
	})

	t.Run("any other staging still points at the supported value", func(t *testing.T) {
		t.Parallel()
		_, err := SelectExecutionPlan(
			domain.DataPipelineSyncModeFull,
			domain.DataPipelineDestinationTypeTiDBLake,
			domain.DataPipelineStagingType("something-else"),
		)
		require.ErrorIs(t, err, ErrUnsupportedStaging)
		require.Contains(t, err.Error(), "staging.type")
		require.Contains(t, err.Error(), string(domain.DataPipelineStagingTypeCustomerManagedS3))
	})

	t.Run("destination names its field and the supported value", func(t *testing.T) {
		t.Parallel()
		_, err := SelectExecutionPlan(
			domain.DataPipelineSyncModeFull,
			domain.DataPipelineDestinationTypeSnowflake,
			domain.DataPipelineStagingTypeCustomerManagedS3,
		)
		require.ErrorIs(t, err, ErrUnsupportedDestination)
		require.Contains(t, err.Error(), "destination.type")
		require.Contains(t, err.Error(), string(domain.DataPipelineDestinationTypeTiDBLake))
	})

	t.Run("sync mode names both supported values", func(t *testing.T) {
		t.Parallel()
		_, err := SelectExecutionPlan(
			domain.DataPipelineSyncMode("unknown"),
			domain.DataPipelineDestinationTypeTiDBLake,
			domain.DataPipelineStagingTypeCustomerManagedS3,
		)
		require.ErrorIs(t, err, ErrUnsupportedSyncMode)
		require.Contains(t, err.Error(), "sync_mode")
		require.Contains(t, err.Error(), string(domain.DataPipelineSyncModeFull))
		require.Contains(t, err.Error(), string(domain.DataPipelineSyncModeFullAndIncremental))
	})
}

func TestSelectExecutionPlan(t *testing.T) {
	tests := map[string]struct {
		syncMode    domain.DataPipelineSyncMode
		destination domain.DataPipelineDestinationType
		staging     domain.DataPipelineStagingType
		wantErr     error
		incremental bool
	}{
		"full uses Dumpling CSV": {
			syncMode:    domain.DataPipelineSyncModeFull,
			destination: domain.DataPipelineDestinationTypeTiDBLake,
			staging:     domain.DataPipelineStagingTypeCustomerManagedS3,
		},
		"full and incremental adds TiCDC Canal JSON": {
			syncMode:    domain.DataPipelineSyncModeFullAndIncremental,
			destination: domain.DataPipelineDestinationTypeTiDBLake,
			staging:     domain.DataPipelineStagingTypeCustomerManagedS3,
			incremental: true,
		},
		"Snowflake fails closed": {
			syncMode:    domain.DataPipelineSyncModeFull,
			destination: domain.DataPipelineDestinationTypeSnowflake,
			staging:     domain.DataPipelineStagingTypeCustomerManagedS3,
			wantErr:     ErrUnsupportedDestination,
		},
		"managed staging fails before execution": {
			syncMode:    domain.DataPipelineSyncModeFull,
			destination: domain.DataPipelineDestinationTypeTiDBLake,
			staging:     domain.DataPipelineStagingTypeTiDBManaged,
			wantErr:     ErrUnsupportedStaging,
		},
		"unknown sync mode fails closed": {
			syncMode:    "unknown",
			destination: domain.DataPipelineDestinationTypeTiDBLake,
			staging:     domain.DataPipelineStagingTypeCustomerManagedS3,
			wantErr:     ErrUnsupportedSyncMode,
		},
	}

	for name, test := range tests {
		t.Run(name, func(t *testing.T) {
			plan, err := SelectExecutionPlan(test.syncMode, test.destination, test.staging)
			if test.wantErr != nil {
				require.ErrorIs(t, err, test.wantErr)
				require.Empty(t, plan)
				return
			}
			require.NoError(t, err)
			require.Equal(t, executionPlanVersion, plan.Version)
			require.Equal(t, domain.DataPipelineSnapshotEngineDumpling, plan.Snapshot.Engine)
			require.Equal(t, domain.DataPipelineSnapshotFormatCSV, plan.Snapshot.Format)
			if test.incremental {
				require.NotNil(t, plan.Incremental)
				require.Equal(t, domain.DataPipelineIncrementalEngineTiCDC, plan.Incremental.Engine)
				require.Equal(t, domain.DataPipelineIncrementalFormatCanalJSON, plan.Incremental.Format)
			} else {
				require.Nil(t, plan.Incremental)
			}
		})
	}
}

func TestValidateExecutionPlanImmutable(t *testing.T) {
	persisted, err := SelectExecutionPlan(
		domain.DataPipelineSyncModeFullAndIncremental,
		domain.DataPipelineDestinationTypeTiDBLake,
		domain.DataPipelineStagingTypeCustomerManagedS3,
	)
	require.NoError(t, err)

	identical, err := SelectExecutionPlan(
		domain.DataPipelineSyncModeFullAndIncremental,
		domain.DataPipelineDestinationTypeTiDBLake,
		domain.DataPipelineStagingTypeCustomerManagedS3,
	)
	require.NoError(t, err)
	require.NoError(t, ValidateExecutionPlanImmutable(persisted, identical))

	mutations := map[string]func(domain.DataPipelineExecutionPlan) domain.DataPipelineExecutionPlan{
		"snapshot engine cannot change to BR": func(plan domain.DataPipelineExecutionPlan) domain.DataPipelineExecutionPlan {
			plan.Snapshot.Engine = domain.DataPipelineSnapshotEngineBR
			return plan
		},
		"snapshot format is fixed": func(plan domain.DataPipelineExecutionPlan) domain.DataPipelineExecutionPlan {
			plan.Snapshot.Format = "parquet"
			return plan
		},
		"incremental encoding is fixed": func(plan domain.DataPipelineExecutionPlan) domain.DataPipelineExecutionPlan {
			incremental := *plan.Incremental
			incremental.Format = "other"
			plan.Incremental = &incremental
			return plan
		},
		"plan version is fixed": func(plan domain.DataPipelineExecutionPlan) domain.DataPipelineExecutionPlan {
			plan.Version++
			return plan
		},
	}
	for name, mutate := range mutations {
		t.Run(name, func(t *testing.T) {
			require.True(t, errors.Is(
				ValidateExecutionPlanImmutable(persisted, mutate(identical)),
				ErrExecutionPlanImmutable,
			))
		})
	}
}