package entity

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

	"github.com/stretchr/testify/require"
)

func TestDataPipelineTypedJSONRoundTrip(t *testing.T) {
	now := time.Date(2026, time.August, 2, 8, 30, 0, 0, time.UTC)
	value := struct {
		Params        DataPipelineParams         `json:"params"`
		ExecutionPlan DataPipelineExecutionPlan  `json:"execution_plan"`
		IntervalPlan  DeliveryIntervalPlan       `json:"interval_plan"`
		Children      DataPipelineChildResources `json:"children"`
		Observation   DataPipelineObservation    `json:"observation"`
	}{
		Params: DataPipelineParams{
			Version:  1,
			SyncMode: DataPipelineSyncModeFullAndIncremental,
			Source: DataPipelineSourceConfig{
				Username:           "test-pipeline-user",
				PasswordCiphertext: "encrypted-source-password",
				TableFilter:        []string{"test_app.*", "test_audit.orders"},
				CaseSensitive:      true,
			},
			Staging: DataPipelineStagingConfig{
				Type: DataPipelineStagingTypeCustomerManagedS3,
				S3: DataPipelineS3Config{
					Bucket:                    "test-pipeline-bucket",
					BasePrefix:                "exports",
					Region:                    "us-west-2",
					AccessKeyID:               "test-access-key",
					SecretAccessKeyCiphertext: "encrypted-s3-secret",
				},
			},
			Destination: DataPipelineDestinationConfig{
				Type:      DataPipelineDestinationTypeTiDBLake,
				Warehouse: "test-warehouse",
				TargetNaming: DataPipelineTargetNaming{
					DatabasePrefix: "${database}",
					TablePrefix:    "${table}",
				},
			},
		},
		ExecutionPlan: DataPipelineExecutionPlan{
			Version:         1,
			DestinationType: DataPipelineDestinationTypeTiDBLake,
			Snapshot: DataPipelineSnapshotPlan{
				Engine: DataPipelineSnapshotEngineDumpling,
				Format: DataPipelineSnapshotFormatCSV,
			},
			Incremental: &DataPipelineIncrementalPlan{
				Engine: DataPipelineIncrementalEngineTiCDC,
				Format: DataPipelineIncrementalFormatCanalJSON,
			},
		},
		IntervalPlan: DeliveryIntervalPlan{
			Version:                      1,
			EndToEndSeconds:              60,
			ProducerFlushIntervalSeconds: 30,
			ConsumerPollIntervalSeconds:  30,
		},
		Children: DataPipelineChildResources{
			ExportID:     "export-101",
			ChangefeedID: 102,
		},
		Observation: DataPipelineObservation{
			LastAttemptAt:           &now,
			LastSucceededAt:         &now,
			ExportStatus:            "SUCCEEDED",
			ExportErrorMessage:      "",
			ChangefeedStatus:        "NORMAL",
			LakeDataSourceStatus:    "READY",
			LakeIntegrationStatus:   "RUNNING",
			ChangefeedCheckpointTSO: 123 << 18,
			ChangefeedCheckpointAt:  &now,
			LakeResolvedTSO:         122 << 18,
			LakeResolvedAt:          &now,
			LatencySeconds:          1,
			Stale:                   false,
		},
	}

	raw, err := json.Marshal(value)
	require.NoError(t, err)

	var decoded struct {
		Params        DataPipelineParams         `json:"params"`
		ExecutionPlan DataPipelineExecutionPlan  `json:"execution_plan"`
		IntervalPlan  DeliveryIntervalPlan       `json:"interval_plan"`
		Children      DataPipelineChildResources `json:"children"`
		Observation   DataPipelineObservation    `json:"observation"`
	}
	require.NoError(t, json.Unmarshal(raw, &decoded))
	require.Equal(t, value.Params, decoded.Params)
	require.Equal(t, value.ExecutionPlan, decoded.ExecutionPlan)
	require.Equal(t, value.IntervalPlan, decoded.IntervalPlan)
	require.Equal(t, value.Children, decoded.Children)
	require.Equal(t, value.Observation, decoded.Observation)
}

func TestDataPipelineTypedJSONToleratesAdditiveFields(t *testing.T) {
	raw := []byte(`{
		"version":1,
		"sync_mode":"full",
		"source":{"username":"test-user","table_filter":["db.*"],"future_source_field":true},
		"staging":{"type":"customer_managed_s3","s3":{"bucket":"test-bucket","region":"us-west-2"}},
		"destination":{"type":"tidb_lake","warehouse":"test-warehouse","target_naming":{}},
		"future_top_level_field":{"enabled":true}
	}`)

	var params DataPipelineParams
	require.NoError(t, json.Unmarshal(raw, &params))
	require.Equal(t, DataPipelineSyncModeFull, params.SyncMode)
	require.Equal(t, []string{"db.*"}, params.Source.TableFilter)
	require.Equal(t, "test-warehouse", params.Destination.Warehouse)
}

func TestDataPipelineIsRegisteredForORMMigration(t *testing.T) {
	for _, model := range ORMModels {
		if _, ok := model.(*DataPipeline); ok {
			return
		}
	}
	require.Fail(t, "DataPipeline must be present in ORMModels")
}