package data_pipeline

import (
	"context"
	"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"
	workflow_action "github.com/tidbcloud/dataflow-service-ng/app/workflow/action"
	app_log "github.com/tidbcloud/dataflow-service-ng/pkg/log"
	"github.com/tidbcloud/pkgv2/bizerr"
	"go.uber.org/zap"
	"go.uber.org/zap/zaptest/observer"
)

type coordinatorCallRecord struct {
	method     string
	pipelineID uint64
}

type coordinatorSpy struct {
	calls []coordinatorCallRecord
	err   bizerr.BizError
}

func (s *coordinatorSpy) record(method string, pipelineID uint64) bizerr.BizError {
	s.calls = append(s.calls, coordinatorCallRecord{method: method, pipelineID: pipelineID})
	return s.err
}

func (s *coordinatorSpy) CreateExport(_ context.Context, id uint64) bizerr.BizError {
	return s.record("CreateExport", id)
}

func (s *coordinatorSpy) CreateChangefeed(_ context.Context, id uint64) bizerr.BizError {
	return s.record("CreateChangefeed", id)
}

func (s *coordinatorSpy) WaitSnapshotReady(_ context.Context, id uint64) bizerr.BizError {
	return s.record("WaitSnapshotReady", id)
}

func (s *coordinatorSpy) CreateDestination(_ context.Context, id uint64) bizerr.BizError {
	return s.record("CreateDestination", id)
}

func (s *coordinatorSpy) StartDestination(_ context.Context, id uint64) bizerr.BizError {
	return s.record("StartDestination", id)
}

func (s *coordinatorSpy) FinishCreate(_ context.Context, id uint64) bizerr.BizError {
	return s.record("FinishCreate", id)
}

func (s *coordinatorSpy) PauseChangefeed(_ context.Context, id uint64) bizerr.BizError {
	return s.record("PauseChangefeed", id)
}

func (s *coordinatorSpy) StopDestination(_ context.Context, id uint64) bizerr.BizError {
	return s.record("StopDestination", id)
}

func (s *coordinatorSpy) FinishPause(_ context.Context, id uint64) bizerr.BizError {
	return s.record("FinishPause", id)
}

func (s *coordinatorSpy) ApplyChangefeed(_ context.Context, id uint64) bizerr.BizError {
	return s.record("ApplyChangefeed", id)
}

func (s *coordinatorSpy) ApplyDestination(_ context.Context, id uint64) bizerr.BizError {
	return s.record("ApplyDestination", id)
}

func (s *coordinatorSpy) FinishEdit(_ context.Context, id uint64) bizerr.BizError {
	return s.record("FinishEdit", id)
}

func (s *coordinatorSpy) ResumeChangefeed(_ context.Context, id uint64) bizerr.BizError {
	return s.record("ResumeChangefeed", id)
}

func (s *coordinatorSpy) FinishResume(_ context.Context, id uint64) bizerr.BizError {
	return s.record("FinishResume", id)
}

func (s *coordinatorSpy) DeleteDestination(_ context.Context, id uint64) bizerr.BizError {
	return s.record("DeleteDestination", id)
}

func (s *coordinatorSpy) DeleteChangefeed(_ context.Context, id uint64) bizerr.BizError {
	return s.record("DeleteChangefeed", id)
}

func (s *coordinatorSpy) DeleteExport(_ context.Context, id uint64) bizerr.BizError {
	return s.record("DeleteExport", id)
}

func (s *coordinatorSpy) FinishDelete(_ context.Context, id uint64) bizerr.BizError {
	return s.record("FinishDelete", id)
}

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

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

func TestLifecycleActionsDelegateExactlyOnce(t *testing.T) {
	type actionFactory func(data_pipeline_model.LifecycleCoordinator) *workflow_action.Wrapper
	tests := []struct {
		name       string
		factory    actionFactory
		method     string
		runInAfter bool
	}{
		{name: "create export", factory: NewCreateExportAction, method: "CreateExport"},
		{name: "create changefeed", factory: NewCreateChangefeedAction, method: "CreateChangefeed"},
		{name: "wait snapshot", factory: NewWaitSnapshotReadyAction, method: "WaitSnapshotReady", runInAfter: true},
		{name: "create destination", factory: NewCreateDestinationAction, method: "CreateDestination"},
		{name: "start destination", factory: NewStartDestinationAction, method: "StartDestination", runInAfter: true},
		{name: "finish create", factory: NewFinishCreateAction, method: "FinishCreate"},
		{name: "pause changefeed", factory: NewPauseChangefeedAction, method: "PauseChangefeed", runInAfter: true},
		{name: "stop destination", factory: NewStopDestinationAction, method: "StopDestination", runInAfter: true},
		{name: "finish pause", factory: NewFinishPauseAction, method: "FinishPause"},
		{name: "apply changefeed", factory: NewApplyChangefeedAction, method: "ApplyChangefeed"},
		{name: "apply destination", factory: NewApplyDestinationAction, method: "ApplyDestination", runInAfter: true},
		{name: "resume destination", factory: NewResumeDestinationAction, method: "StartDestination", runInAfter: true},
		{name: "resume changefeed", factory: NewResumeChangefeedAction, method: "ResumeChangefeed", runInAfter: true},
		{name: "finish resume", factory: NewFinishResumeAction, method: "FinishResume"},
		{name: "delete destination", factory: NewDeleteDestinationAction, method: "DeleteDestination"},
		{name: "delete changefeed", factory: NewDeleteChangefeedAction, method: "DeleteChangefeed"},
		{name: "delete export", factory: NewDeleteExportAction, method: "DeleteExport"},
		{name: "finish delete", factory: NewFinishDeleteAction, method: "FinishDelete"},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			coordinator := &coordinatorSpy{}
			lifecycleAction := test.factory(coordinator)
			params, ok := lifecycleAction.ParameterNew().(*Params)
			require.True(t, ok)
			params.PipelineID = 42

			err := lifecycleAction.Run(newExecuteContext(), params)
			require.NoError(t, err)
			if test.runInAfter {
				assert.Empty(t, coordinator.calls, "Run must not poll long-running stages")
				err = lifecycleAction.RunAfter(newExecuteContext(), params)
				require.NoError(t, err)
			}

			require.Equal(t, []coordinatorCallRecord{{method: test.method, pipelineID: 42}}, coordinator.calls)
		})
	}
}

func TestLifecycleActionInvalidParamsDoNotCallCoordinator(t *testing.T) {
	invalidParams := []interface{}{
		nil,
		Params{PipelineID: 42},
		(*Params)(nil),
		&Params{},
	}

	for _, params := range invalidParams {
		coordinator := &coordinatorSpy{}
		runAction := &runLifecycleAction{call: coordinator.CreateExport}
		err := runAction.Run(context.Background(), zap.NewNop(), params)
		require.Error(t, err)
		assert.Empty(t, coordinator.calls)

		runAfterAction := &runAfterLifecycleAction{call: coordinator.WaitSnapshotReady}
		err = runAfterAction.RunAfter(context.Background(), zap.NewNop(), params)
		require.Error(t, err)
		assert.Empty(t, coordinator.calls)
	}
}

// Every lifecycle action goes through invokeCoordinator, so binding the id to
// the logger here is what makes "show me everything that happened to this
// Pipeline" a single query. Without it the coordinator's own statements carry
// only an operation name and a trace id, and finding the Pipeline they belong
// to means first finding some other line that happens to name it.
func TestInvokeCoordinatorBindsThePipelineIDToTheLogger(t *testing.T) {
	t.Parallel()

	core, logs := observer.New(zap.InfoLevel)
	ctx := app_log.WithLogger(context.Background(), zap.New(core))

	var seen uint64
	err := invokeCoordinator(ctx, &Params{PipelineID: 987654321}, func(inner context.Context, id uint64) bizerr.BizError {
		seen = id
		// Anything the coordinator logs downstream, without naming the id.
		app_log.GetLogger(inner).Info("something happened downstream")
		return nil
	})

	require.NoError(t, err)
	require.Equal(t, uint64(987654321), seen)
	entries := logs.All()
	require.Len(t, entries, 1)
	require.Equal(t, uint64(987654321), entries[0].ContextMap()["pipelineID"],
		"a downstream statement that never mentions the id still carries it")
}