package data_pipeline

import (
	"context"
	"errors"

	data_pipeline_model "github.com/tidbcloud/dataflow-service-ng/app/models/data_pipeline"
	"github.com/tidbcloud/dataflow-service-ng/app/workflow"
	"github.com/tidbcloud/dataflow-service-ng/app/workflow/action"
	"github.com/tidbcloud/dataflow-service-ng/pkg/log"
	"github.com/tidbcloud/pkgv2/bizerr"
	"go.uber.org/zap"
)

type Params struct {
	PipelineID uint64 `json:"data_pipeline_id"`
}

type coordinatorCall func(context.Context, uint64) bizerr.BizError

type runLifecycleAction struct {
	call coordinatorCall
}

func (a *runLifecycleAction) Params() interface{} {
	return &Params{}
}

func (a *runLifecycleAction) Run(
	ctx context.Context,
	_ *zap.Logger,
	params interface{},
) error {
	return invokeCoordinator(ctx, params, a.call)
}

type runAfterLifecycleAction struct {
	call coordinatorCall
}

func (a *runAfterLifecycleAction) Params() interface{} {
	return &Params{}
}

func (*runAfterLifecycleAction) Run(context.Context, *zap.Logger, interface{}) error {
	return nil
}

func (a *runAfterLifecycleAction) RunAfter(
	ctx context.Context,
	_ *zap.Logger,
	params interface{},
) error {
	return invokeCoordinator(ctx, params, a.call)
}

func invokeCoordinator(ctx context.Context, params interface{}, call coordinatorCall) error {
	p, ok := params.(*Params)
	if !ok || p == nil {
		return errors.New("invalid data pipeline action params")
	}
	if p.PipelineID == 0 {
		return errors.New("data_pipeline_id must be greater than zero")
	}
	// Bound to the logger once, here, because this is the single entry point
	// every lifecycle action goes through. Everything the coordinator logs
	// downstream then carries the Pipeline it is about, without each statement
	// having to be handed the id - and "show me everything that happened to
	// this Pipeline" becomes one query instead of a hunt for a line that
	// happens to name it and a second search on its trace.
	ctx = log.WithLogger(ctx, log.GetLogger(ctx).With(
		zap.Uint64("pipelineID", p.PipelineID),
	))
	return call(ctx, p.PipelineID)
}

func newBoundedAction(name string, call coordinatorCall) *action.Wrapper {
	return action.MustNewActionWrapper(
		name,
		&runLifecycleAction{call: call},
		action.SetRunRetryFun(action.InvokeRunActionWithExponentialRetry),
	)
}

func newRunAfterAction(name string, call coordinatorCall) *action.Wrapper {
	return action.MustNewActionWrapper(
		name,
		&runAfterLifecycleAction{call: call},
		action.SetRunAfterRetryFun(action.InvokeRunActionWithInfinitelyExponentialRetry),
	)
}

func newDeleteAction(name string, call coordinatorCall) *action.Wrapper {
	return action.MustNewActionWrapper(
		name,
		&runLifecycleAction{call: call},
		action.SetRunRetryFun(action.InvokeRunActionWithInfinitelyRetry),
	)
}

func NewCreateExportAction(coordinator data_pipeline_model.LifecycleCoordinator) *action.Wrapper {
	return newBoundedAction(workflow.ActionDataPipelineCreateExport, coordinator.CreateExport)
}

func NewCreateChangefeedAction(coordinator data_pipeline_model.LifecycleCoordinator) *action.Wrapper {
	return newBoundedAction(workflow.ActionDataPipelineCreateChangefeed, coordinator.CreateChangefeed)
}

func NewWaitSnapshotReadyAction(coordinator data_pipeline_model.LifecycleCoordinator) *action.Wrapper {
	return newRunAfterAction(workflow.ActionDataPipelineWaitSnapshot, coordinator.WaitSnapshotReady)
}

func NewCreateDestinationAction(coordinator data_pipeline_model.LifecycleCoordinator) *action.Wrapper {
	return newBoundedAction(workflow.ActionDataPipelineCreateDestination, coordinator.CreateDestination)
}

func NewStartDestinationAction(coordinator data_pipeline_model.LifecycleCoordinator) *action.Wrapper {
	return newRunAfterAction(workflow.ActionDataPipelineStartDestination, coordinator.StartDestination)
}

func NewFinishCreateAction(coordinator data_pipeline_model.LifecycleCoordinator) *action.Wrapper {
	return newBoundedAction(workflow.ActionDataPipelineFinishCreate, coordinator.FinishCreate)
}

func NewPauseChangefeedAction(coordinator data_pipeline_model.LifecycleCoordinator) *action.Wrapper {
	return newRunAfterAction(workflow.ActionDataPipelinePauseChangefeed, coordinator.PauseChangefeed)
}

func NewStopDestinationAction(coordinator data_pipeline_model.LifecycleCoordinator) *action.Wrapper {
	return newRunAfterAction(workflow.ActionDataPipelineStopDestination, coordinator.StopDestination)
}

func NewFinishPauseAction(coordinator data_pipeline_model.LifecycleCoordinator) *action.Wrapper {
	return newBoundedAction(workflow.ActionDataPipelineFinishPause, coordinator.FinishPause)
}

func NewApplyChangefeedAction(coordinator data_pipeline_model.LifecycleCoordinator) *action.Wrapper {
	return newBoundedAction(workflow.ActionDataPipelineApplyChangefeed, coordinator.ApplyChangefeed)
}

func NewApplyDestinationAction(coordinator data_pipeline_model.LifecycleCoordinator) *action.Wrapper {
	// Lake accepts updates synchronously but may expose the new configuration
	// asynchronously. Keep confirming through Fastflow RunAfter instead of
	// exhausting the short bounded action retry window while Lake converges.
	return newRunAfterAction(workflow.ActionDataPipelineApplyDestination, coordinator.ApplyDestination)
}

func NewFinishEditAction(coordinator data_pipeline_model.LifecycleCoordinator) *action.Wrapper {
	return newBoundedAction(workflow.ActionDataPipelineFinishEdit, coordinator.FinishEdit)
}

func NewResumeDestinationAction(coordinator data_pipeline_model.LifecycleCoordinator) *action.Wrapper {
	return newRunAfterAction(workflow.ActionDataPipelineResumeDestination, coordinator.StartDestination)
}

func NewResumeChangefeedAction(coordinator data_pipeline_model.LifecycleCoordinator) *action.Wrapper {
	return newRunAfterAction(workflow.ActionDataPipelineResumeChangefeed, coordinator.ResumeChangefeed)
}

func NewFinishResumeAction(coordinator data_pipeline_model.LifecycleCoordinator) *action.Wrapper {
	return newBoundedAction(workflow.ActionDataPipelineFinishResume, coordinator.FinishResume)
}

func NewDeleteDestinationAction(coordinator data_pipeline_model.LifecycleCoordinator) *action.Wrapper {
	return newDeleteAction(workflow.ActionDataPipelineDeleteDestination, coordinator.DeleteDestination)
}

func NewDeleteChangefeedAction(coordinator data_pipeline_model.LifecycleCoordinator) *action.Wrapper {
	return newDeleteAction(workflow.ActionDataPipelineDeleteChangefeed, coordinator.DeleteChangefeed)
}

func NewDeleteExportAction(coordinator data_pipeline_model.LifecycleCoordinator) *action.Wrapper {
	return newDeleteAction(workflow.ActionDataPipelineDeleteExport, coordinator.DeleteExport)
}

func NewFinishDeleteAction(coordinator data_pipeline_model.LifecycleCoordinator) *action.Wrapper {
	return newDeleteAction(workflow.ActionDataPipelineFinishDelete, coordinator.FinishDelete)
}