package data_pipeline

import (
	"context"
	"fmt"
	"testing"
	"time"

	"github.com/stretchr/testify/require"
	apprepo "github.com/tidbcloud/dataflow-service-ng/app/repo"
	"github.com/tidbcloud/dataflow-service-ng/app/repo/entity"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/db"
	"gorm.io/gorm"
)

type fixedIDGenerator struct {
	next uint64
}

func (g *fixedIDGenerator) NextID() uint64 {
	g.next++
	return g.next
}

func prepareRepoTest(t *testing.T) (*gorm.DB, DataPipelineRepo) {
	t.Helper()
	testDB, err := apprepo.GetFakeDB()
	require.NoError(t, err)
	require.NoError(t, testDB.Exec("DELETE FROM data_pipelines").Error)
	t.Cleanup(func() {
		require.NoError(t, testDB.Exec("DELETE FROM data_pipelines").Error)
	})
	return testDB, NewDataPipelineRepo(db.NewProvider(testDB), &fixedIDGenerator{next: 9000})
}

func TestDataPipelineRepoJSONRoundTripAndFilters(t *testing.T) {
	_, pipelineRepo := prepareRepoTest(t)
	ctx := context.Background()

	pipelines := []*entity.DataPipeline{
		newTestPipeline(1001, 11, 21, entity.DataPipelineStatusRunning),
		newTestPipeline(1002, 11, 21, entity.DataPipelineStatusPaused),
		newTestPipeline(1003, 12, 22, entity.DataPipelineStatusRunning),
	}
	for _, pipeline := range pipelines {
		require.NoError(t, pipelineRepo.Create(ctx, pipeline))
		require.Equal(t, 1, pipeline.Versioning)
	}

	loaded, err := pipelineRepo.Load(ctx, 1001)
	require.NoError(t, err)
	require.Equal(t, pipelines[0].Params, loaded.Params)
	require.Equal(t, pipelines[0].ExecutionPlan, loaded.ExecutionPlan)
	require.Equal(t, pipelines[0].IntervalPlan, loaded.IntervalPlan)
	require.Equal(t, pipelines[0].Children, loaded.Children)
	require.Equal(t, pipelines[0].LastError, loaded.LastError)
	require.Equal(t, pipelines[0].Observation, loaded.Observation)

	items, err := pipelineRepo.List(
		ctx,
		WithTenantID(11),
		WithClusterID(21),
		WithStatuses(entity.DataPipelineStatusRunning, entity.DataPipelineStatusPaused),
		OrderByIDAsc(),
	)
	require.NoError(t, err)
	require.Equal(t, []entity.ID{1001, 1002}, []entity.ID{items[0].ID, items[1].ID})

	items, err = pipelineRepo.List(ctx, WithIDs(1002), WithIDAfter(1001))
	require.NoError(t, err)
	require.Len(t, items, 1)
	require.Equal(t, entity.ID(1002), items[0].ID)

	count, err := pipelineRepo.Count(ctx, WithTenantID(11), WithPagination(0, 1))
	require.NoError(t, err)
	require.Equal(t, int64(2), count, "Count must ignore pagination")

	got, err := pipelineRepo.Get(ctx, WithClusterID(22))
	require.NoError(t, err)
	require.Equal(t, entity.ID(1003), got.ID)
}

func TestDataPipelineRepoAllocatesID(t *testing.T) {
	_, pipelineRepo := prepareRepoTest(t)
	ctx := context.Background()

	pipeline := newTestPipeline(0, 11, 21, entity.DataPipelineStatusCreating)
	pipeline.StagingRoot = "test/tidb-data-pipelines/staging/21/9001/"
	require.NoError(t, pipelineRepo.Create(ctx, pipeline))
	require.Equal(t, entity.ID(9001), pipeline.ID)
}

func TestDataPipelineRepoOptimisticVersioning(t *testing.T) {
	_, pipelineRepo := prepareRepoTest(t)
	ctx := context.Background()
	pipeline := newTestPipeline(2001, 11, 21, entity.DataPipelineStatusPaused)
	require.NoError(t, pipelineRepo.Create(ctx, pipeline))

	first, err := pipelineRepo.Load(ctx, 2001)
	require.NoError(t, err)
	stale, err := pipelineRepo.Load(ctx, 2001)
	require.NoError(t, err)

	first.Name = "updated-test-pipeline"
	require.NoError(t, pipelineRepo.Save(ctx, first))
	require.Equal(t, 2, first.Versioning)

	stale.Status = entity.DataPipelineStatusRunning
	staleErr := pipelineRepo.Save(ctx, stale)
	require.Error(t, staleErr)
	require.Equal(t, bizerr.ErrCodeOptimisticLockConflict, staleErr.ErrCode())

	latest, err := pipelineRepo.Load(ctx, 2001)
	require.NoError(t, err)
	require.Equal(t, "updated-test-pipeline", latest.Name)
	require.Equal(t, entity.DataPipelineStatusPaused, latest.Status)
}

func TestDataPipelineRepoSoftDelete(t *testing.T) {
	testDB, pipelineRepo := prepareRepoTest(t)
	ctx := context.Background()
	pipeline := newTestPipeline(3001, 11, 21, entity.DataPipelineStatusDeleting)
	require.NoError(t, pipelineRepo.Create(ctx, pipeline))
	require.NoError(t, pipelineRepo.Delete(ctx, 3001))

	_, loadErr := pipelineRepo.Load(ctx, 3001)
	require.Error(t, loadErr)
	require.Equal(t, bizerr.ErrCodeRecordNotFound, loadErr.ErrCode())

	var deleted entity.DataPipeline
	require.NoError(t, testDB.Unscoped().First(&deleted, 3001).Error)
	require.True(t, deleted.DeletedAt.Valid)
}

func newTestPipeline(
	id, tenantID, clusterID uint64,
	status entity.DataPipelineStatus,
) *entity.DataPipeline {
	now := time.Date(2026, time.August, 2, 10, 0, 0, 0, time.UTC)
	return &entity.DataPipeline{
		ID:                      entity.ID(id),
		TenantID:                entity.ID(tenantID),
		ClusterID:               entity.ID(clusterID),
		CreatorID:               31,
		Name:                    fmt.Sprintf("test-pipeline-%d", id),
		Status:                  status,
		ChangefeedSpecification: "standard-2",
		Params: entity.DataPipelineParams{
			Version:  1,
			SyncMode: entity.DataPipelineSyncModeFullAndIncremental,
			Source: entity.DataPipelineSourceConfig{
				Username:           "test-pipeline-user",
				PasswordCiphertext: "encrypted-source-password",
				TableFilter:        []string{"test_app.*", "test_audit.orders"},
			},
			Staging: entity.DataPipelineStagingConfig{
				Type: entity.DataPipelineStagingTypeCustomerManagedS3,
				S3: entity.DataPipelineS3Config{
					Bucket:                    "test-pipeline-bucket",
					Region:                    "us-west-2",
					AccessKeyID:               "test-access-key",
					SecretAccessKeyCiphertext: "encrypted-s3-secret",
				},
			},
			Destination: entity.DataPipelineDestinationConfig{
				Type:      entity.DataPipelineDestinationTypeTiDBLake,
				Warehouse: "test-warehouse",
			},
		},
		ExecutionPlan: entity.DataPipelineExecutionPlan{
			Version:         1,
			DestinationType: entity.DataPipelineDestinationTypeTiDBLake,
			Snapshot: entity.DataPipelineSnapshotPlan{
				Engine: entity.DataPipelineSnapshotEngineDumpling,
				Format: entity.DataPipelineSnapshotFormatCSV,
			},
			Incremental: &entity.DataPipelineIncrementalPlan{
				Engine: entity.DataPipelineIncrementalEngineTiCDC,
				Format: entity.DataPipelineIncrementalFormatCanalJSON,
			},
		},
		IntervalPlan: entity.DeliveryIntervalPlan{
			Version:                      1,
			EndToEndSeconds:              60,
			ProducerFlushIntervalSeconds: 30,
			ConsumerPollIntervalSeconds:  30,
		},
		SnapshotTSO: uint64(now.UnixMilli()) << 18,
		StagingRoot: fmt.Sprintf("test/tidb-data-pipelines/staging/%d/%d/", clusterID, id),
		Children: entity.DataPipelineChildResources{
			ExportID:     fmt.Sprintf("test-export-%d", id),
			ChangefeedID: id + 200,
		},

		Observation: entity.DataPipelineObservation{
			LastAttemptAt:           &now,
			LastSucceededAt:         &now,
			ChangefeedCheckpointTSO: uint64(now.UnixMilli()) << 18,
			ChangefeedCheckpointAt:  &now,
			LakeResolvedAt:          &now,
		},
	}
}