package data_pipeline

import (
	"context"
	"fmt"
	"testing"

	"github.com/shiningrush/fastflow/pkg/entity"
	"github.com/shiningrush/fastflow/pkg/entity/run"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	data_pipeline_model "github.com/tidbcloud/dataflow-service-ng/app/models/data_pipeline"
	"github.com/tidbcloud/dataflow-service-ng/app/workflow"
	data_pipeline_action "github.com/tidbcloud/dataflow-service-ng/app/workflow/action/data_pipeline"
	"github.com/tidbcloud/pkgv2/bizerr"
)

type noopCoordinator struct{}

func (*noopCoordinator) CreateExport(context.Context, uint64) bizerr.BizError { return nil }
func (*noopCoordinator) CreateChangefeed(context.Context, uint64) bizerr.BizError {
	return nil
}
func (*noopCoordinator) WaitSnapshotReady(context.Context, uint64) bizerr.BizError {
	return nil
}
func (*noopCoordinator) CreateDestination(context.Context, uint64) bizerr.BizError {
	return nil
}
func (*noopCoordinator) StartDestination(context.Context, uint64) bizerr.BizError  { return nil }
func (*noopCoordinator) FinishCreate(context.Context, uint64) bizerr.BizError      { return nil }
func (*noopCoordinator) PauseChangefeed(context.Context, uint64) bizerr.BizError   { return nil }
func (*noopCoordinator) StopDestination(context.Context, uint64) bizerr.BizError   { return nil }
func (*noopCoordinator) FinishPause(context.Context, uint64) bizerr.BizError       { return nil }
func (*noopCoordinator) ApplyChangefeed(context.Context, uint64) bizerr.BizError   { return nil }
func (*noopCoordinator) ApplyDestination(context.Context, uint64) bizerr.BizError  { return nil }
func (*noopCoordinator) FinishEdit(context.Context, uint64) bizerr.BizError        { return nil }
func (*noopCoordinator) ResumeChangefeed(context.Context, uint64) bizerr.BizError  { return nil }
func (*noopCoordinator) FinishResume(context.Context, uint64) bizerr.BizError      { return nil }
func (*noopCoordinator) DeleteDestination(context.Context, uint64) bizerr.BizError { return nil }
func (*noopCoordinator) DeleteChangefeed(context.Context, uint64) bizerr.BizError  { return nil }
func (*noopCoordinator) DeleteExport(context.Context, uint64) bizerr.BizError      { return nil }
func (*noopCoordinator) FinishDelete(context.Context, uint64) bizerr.BizError      { return nil }

var _ data_pipeline_model.LifecycleCoordinator = (*noopCoordinator)(nil)

type pollingCoordinator struct {
	noopCoordinator
	calls []string
}

func (c *pollingCoordinator) record(method string) bizerr.BizError {
	c.calls = append(c.calls, method)
	return nil
}

func (c *pollingCoordinator) WaitSnapshotReady(context.Context, uint64) bizerr.BizError {
	return c.record("WaitSnapshotReady")
}

func (c *pollingCoordinator) StartDestination(context.Context, uint64) bizerr.BizError {
	return c.record("StartDestination")
}

func (c *pollingCoordinator) PauseChangefeed(context.Context, uint64) bizerr.BizError {
	return c.record("PauseChangefeed")
}

func (c *pollingCoordinator) StopDestination(context.Context, uint64) bizerr.BizError {
	return c.record("StopDestination")
}

func (c *pollingCoordinator) ResumeChangefeed(context.Context, uint64) bizerr.BizError {
	return c.record("ResumeChangefeed")
}

var _ data_pipeline_model.LifecycleCoordinator = (*pollingCoordinator)(nil)

func newExecuteContext() run.ExecuteContext {
	ctx := entity.CtxWithRunningTaskIns(context.Background(), &entity.TaskInstance{})
	return run.NewDefExecuteContext(ctx, nil, nil, nil, nil)
}

func TestDataPipelineDags(t *testing.T) {
	type dagFactory func(data_pipeline_model.LifecycleCoordinator) (*entity.Dag, []run.Action)
	tests := []struct {
		name    string
		factory dagFactory
		dagName string
		actions []string
	}{
		{
			name:    "create",
			factory: GetCreateDataPipelineDag,
			dagName: workflow.DagCreateDataPipeline,
			actions: []string{
				workflow.ActionDataPipelineCreateExport,
				workflow.ActionDataPipelineCreateChangefeed,
				workflow.ActionDataPipelineWaitSnapshot,
				workflow.ActionDataPipelineCreateDestination,
				workflow.ActionDataPipelineStartDestination,
				workflow.ActionDataPipelineFinishCreate,
			},
		},
		{
			name:    "pause",
			factory: GetPauseDataPipelineDag,
			dagName: workflow.DagPauseDataPipeline,
			actions: []string{
				workflow.ActionDataPipelinePauseChangefeed,
				workflow.ActionDataPipelineStopDestination,
				workflow.ActionDataPipelineFinishPause,
			},
		},
		{
			// Edit owns applying the desired revision to the stopped children.
			name:    "edit",
			factory: GetEditDataPipelineDag,
			dagName: workflow.DagEditDataPipeline,
			actions: []string{
				workflow.ActionDataPipelineApplyChangefeed,
				workflow.ActionDataPipelineApplyDestination,
				workflow.ActionDataPipelineFinishEdit,
			},
		},
		{
			// Resume only starts children that already hold the applied
			// configuration, destination before incremental.
			name:    "resume",
			factory: GetResumeDataPipelineDag,
			dagName: workflow.DagResumeDataPipeline,
			actions: []string{
				workflow.ActionDataPipelineResumeDestination,
				workflow.ActionDataPipelineResumeChangefeed,
				workflow.ActionDataPipelineFinishResume,
			},
		},
		{
			// Producers before the destination, and the order is the assertion
			// rather than an incidental detail: with the destination first, a
			// destination that cannot be reached retries forever and the
			// changefeed is never deleted - still running, still writing, still
			// charging, after the user asked for the pipeline to go away.
			name:    "delete",
			factory: GetDeleteDataPipelineDag,
			dagName: workflow.DagDeleteDataPipeline,
			actions: []string{
				workflow.ActionDataPipelineDeleteChangefeed,
				workflow.ActionDataPipelineDeleteExport,
				workflow.ActionDataPipelineDeleteDestination,
				workflow.ActionDataPipelineFinishDelete,
			},
		},
	}

	seenActionNames := make(map[string]string)
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			dag, actions := test.factory(&noopCoordinator{})

			assert.Equal(t, test.dagName, dag.ID)
			assert.Equal(t, test.dagName, dag.Name)
			assert.Equal(t, entity.DagStatusNormal, dag.Status)
			require.Len(t, dag.Vars, 1)
			_, ok := dag.Vars[workflow.KeyDataPipelineID]
			assert.True(t, ok, "data_pipeline_id must be a required DAG variable")
			require.Len(t, dag.Tasks, len(test.actions))
			require.Len(t, actions, len(test.actions))

			for i, expectedName := range test.actions {
				task := dag.Tasks[i]
				assert.Equal(t, expectedName, task.ID)
				assert.Equal(t, expectedName, task.ActionName)
				assert.Equal(t, expectedName, actions[i].Name())
				assert.Equal(t, map[string]interface{}{
					workflow.KeyDataPipelineID: fmt.Sprintf("{{%v}}", workflow.KeyDataPipelineID),
				}, task.Params)
				assert.Positive(t, task.TimeoutSecs)

				if i == 0 {
					assert.Empty(t, task.DependOn)
				} else {
					assert.Equal(t, []string{test.actions[i-1]}, task.DependOn)
				}

				previousDag, duplicate := seenActionNames[expectedName]
				assert.Falsef(t, duplicate, "action name %q is also used by %s DAG", expectedName, previousDag)
				seenActionNames[expectedName] = test.name
			}
		})
	}
}

func TestDataPipelinePollingActionsRunInRunAfter(t *testing.T) {
	type dagFactory func(data_pipeline_model.LifecycleCoordinator) (*entity.Dag, []run.Action)

	expectedCalls := map[string]string{
		workflow.ActionDataPipelineWaitSnapshot:      "WaitSnapshotReady",
		workflow.ActionDataPipelineStartDestination:  "StartDestination",
		workflow.ActionDataPipelinePauseChangefeed:   "PauseChangefeed",
		workflow.ActionDataPipelineStopDestination:   "StopDestination",
		workflow.ActionDataPipelineResumeDestination: "StartDestination",
		workflow.ActionDataPipelineResumeChangefeed:  "ResumeChangefeed",
	}
	factories := []dagFactory{
		GetCreateDataPipelineDag,
		GetPauseDataPipelineDag,
		GetResumeDataPipelineDag,
	}

	seen := make(map[string]struct{}, len(expectedCalls))
	for _, factory := range factories {
		coordinator := &pollingCoordinator{}
		_, actions := factory(coordinator)
		for _, lifecycleAction := range actions {
			expectedCall, ok := expectedCalls[lifecycleAction.Name()]
			if !ok {
				continue
			}

			t.Run(lifecycleAction.Name(), func(t *testing.T) {
				params := &data_pipeline_action.Params{PipelineID: 42}
				before := len(coordinator.calls)

				require.NoError(t, lifecycleAction.Run(newExecuteContext(), params))
				assert.Len(t, coordinator.calls, before, "Run must not poll long-running stages")

				afterAction, ok := lifecycleAction.(run.AfterAction)
				require.True(t, ok)
				require.NoError(t, afterAction.RunAfter(newExecuteContext(), params))
				require.Len(t, coordinator.calls, before+1)
				assert.Equal(t, expectedCall, coordinator.calls[before])
			})
			seen[lifecycleAction.Name()] = struct{}{}
		}
	}

	assert.Len(t, seen, len(expectedCalls), "every polling action must be present in a lifecycle DAG")
}