package action

import (
	"context"
	"errors"
	"fmt"
	"strings"
	"time"
	"unicode/utf8"

	apiv1 "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	"github.com/tidbcloud/cluster-service-ng/domain/entity"
	"github.com/tidbcloud/cluster-service-ng/domain/workflow"
	braction "github.com/tidbcloud/cluster-service-ng/domain/workflow/action/br"
	"github.com/tidbcloud/cluster-service-ng/pkg/dataplane"
	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/actions"
	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/executor"
	"github.com/tidbcloud/cluster-service-ng/pkg/storage/repo"
	"github.com/tidbcloud/cluster-service-ng/pkg/tioperator"
	"github.com/tidbcloud/cluster-service-ng/pkg/utils"
	"github.com/tidbcloud/pkgv2/db"
)

const (
	PausePrecheckActionName       = "pause-sharedpool-precheck"
	ApplySuspendComputeActionName = "apply-suspend-compute"
	WaitSuspendedActionName       = "wait-sharedpool-suspended"
	CompletePauseActionName       = "complete-pause-sharedpool"
	ApplyResumeComputeActionName  = "apply-resume-compute"
	WaitAvailableActionName       = "wait-sharedpool-available"
	CompleteResumeActionName      = "complete-resume-sharedpool"
)

type pauseResumeStageAction struct {
	operationRepo repo.PauseResumeOperationRepo
	stage         apiv1.PauseResumeOperation_Stage
	action        actions.Action
}

var errPauseResumeOwnershipLost = errors.New("pause/resume operation ownership lost")

func DecoratePauseResumeStage(operationRepo repo.PauseResumeOperationRepo, stage apiv1.PauseResumeOperation_Stage, action actions.Action) actions.Action {
	return &pauseResumeStageAction{operationRepo: operationRepo, stage: stage, action: action}
}

func (a *pauseResumeStageAction) Name() string { return a.action.Name() }

func (a *pauseResumeStageAction) CloneAction() actions.Action {
	// Keep the stage-tracking decorator while cloning a wrapped async action. The
	// latter carries per-execution abort state and must not be shared by DAG runs.
	inner := a.action
	if cloneable, ok := inner.(interface{ CloneAction() actions.Action }); ok {
		inner = cloneable.CloneAction()
	}
	return &pauseResumeStageAction{operationRepo: a.operationRepo, stage: a.stage, action: inner}
}

func (a *pauseResumeStageAction) TryAbort() bool { return actions.TryAbortAction(a.action) }

func (a *pauseResumeStageAction) Invoke(args actions.ActionArgs) error {
	ctx := utils.GenerateContext(args)
	operation, err := loadActionOperation(ctx, a.operationRepo, args)
	if err != nil {
		return err
	}
	updated, err := a.operationRepo.UpdateStageIfRunning(ctx, operation.GetSharedPoolId(), operation.GetOperationId(), a.stage)
	if err != nil {
		return fmt.Errorf("save pause/resume operation stage: %w", err)
	}
	if !updated {
		return actions.NewTerminalError(errPauseResumeOwnershipLost)
	}
	if err := a.action.Invoke(args); err != nil {
		if actions.IsAsyncTimeoutError(err) {
			err = actions.NewTerminalError(err)
		}
		if !actions.IsTerminalError(err) {
			return err
		}
		updated, saveErr := a.operationRepo.MarkFailedIfRunning(ctx, operation.GetSharedPoolId(), operation.GetOperationId(), pauseResumeResultSummary(err), time.Now())
		if saveErr != nil {
			return fmt.Errorf("save terminal pause/resume operation failure: %w", saveErr)
		}
		if !updated {
			return actions.NewTerminalError(errPauseResumeOwnershipLost)
		}
		return err
	}
	return nil
}

func pauseResumeResultSummary(err error) string {
	const maxBytes = 1024
	result := strings.Join(strings.Fields(err.Error()), " ")
	if len(result) > maxBytes {
		result = result[:maxBytes]
		for !utf8.ValidString(result) {
			result = result[:len(result)-1]
		}
	}
	return result
}

type PauseResumeOwnerGuard struct {
	operationRepo  repo.PauseResumeOperationRepo
	sharedPoolRepo repo.SharedPoolRepo
	lockRepo       repo.LockRepo
}

func NewPauseResumeOwnerGuard(operationRepo repo.PauseResumeOperationRepo, sharedPoolRepo repo.SharedPoolRepo, lockRepo repo.LockRepo) *PauseResumeOwnerGuard {
	return &PauseResumeOwnerGuard{operationRepo: operationRepo, sharedPoolRepo: sharedPoolRepo, lockRepo: lockRepo}
}

func (g *PauseResumeOwnerGuard) Run(args actions.ActionArgs, write func() error) (retErr error) {
	ctx := utils.GenerateContext(args)
	sharedPoolID, err := args.GetString(workflow.SharedPoolIdKey)
	if err != nil {
		return actions.NewTerminalError(err)
	}
	operationID, err := args.GetString(workflow.PauseResumeOperationIdKey)
	if err != nil {
		return actions.NewTerminalError(err)
	}
	operationType, err := args.GetString(workflow.PauseResumeOperationTypeKey)
	if err != nil {
		return actions.NewTerminalError(err)
	}
	unlock, err := g.lockRepo.Lock(ctx, &entity.Lock{Key: sharedPoolID, TimeoutSecs: 60})
	if err != nil {
		return err
	}
	defer func() {
		if unlockErr := unlock(); retErr == nil && unlockErr != nil {
			retErr = unlockErr
		}
	}()
	operation, err := g.operationRepo.LoadByID(ctx, sharedPoolID, operationID)
	if err != nil {
		return fmt.Errorf("load pause/resume owner: %w", err)
	}
	if operation.GetStatus() != apiv1.PauseResumeOperation_RUNNING || operation.GetType().String() != operationType {
		return actions.NewTerminalError(errPauseResumeOwnershipLost)
	}
	sharedPool, err := g.sharedPoolRepo.Load(ctx, sharedPoolID, false)
	if err != nil {
		return fmt.Errorf("load shared pool for pause/resume owner: %w", err)
	}
	expectedState := entity.PauseResumeStatePausing
	if operation.GetType() == apiv1.PauseResumeOperation_RESUME {
		expectedState = entity.PauseResumeStateResuming
	}
	if sharedPool.PauseResumeState != expectedState {
		return actions.NewTerminalError(errPauseResumeOwnershipLost)
	}
	return write()
}

func loadActionOperation(ctx context.Context, operationRepo repo.PauseResumeOperationRepo, args actions.ActionArgs) (*entity.PauseResumeOperation, error) {
	sharedPoolID, err := args.GetString(workflow.SharedPoolIdKey)
	if err != nil {
		return nil, err
	}
	operationID, err := args.GetString(workflow.PauseResumeOperationIdKey)
	if err != nil {
		return nil, err
	}
	operation, err := operationRepo.LoadByID(ctx, sharedPoolID, operationID)
	if err != nil {
		return nil, fmt.Errorf("load committed pause/resume operation: %w", err)
	}
	operationType, err := args.GetString(workflow.PauseResumeOperationTypeKey)
	if err != nil {
		return nil, actions.NewTerminalError(err)
	}
	if operation.GetType().String() != operationType {
		return nil, actions.NewTerminalError(fmt.Errorf("pause/resume operation type does not match DAG"))
	}
	return operation, nil
}

type pausePrecheckAction struct {
	sharedPoolRepo  repo.SharedPoolRepo
	dataplaneRepo   repo.DataplaneRepo
	dataplane       dataplane.Manager
	disableFlushWAL bool
	dagQuery        func(string, executor.MatchFilter) (executor.ExecTaskIdStatus, error)
	conflictPlans   []string
	pausePlans      []string
}

func NewPausePrecheckAction(sharedPoolRepo repo.SharedPoolRepo, dataplaneRepo repo.DataplaneRepo, manager dataplane.Manager, disabled bool,
	dagQuery func(string, executor.MatchFilter) (executor.ExecTaskIdStatus, error), pausePlans, conflictPlans []string,
) actions.Action {
	return &pausePrecheckAction{
		sharedPoolRepo: sharedPoolRepo, dataplaneRepo: dataplaneRepo, dataplane: manager, disableFlushWAL: disabled,
		dagQuery: dagQuery, pausePlans: pausePlans, conflictPlans: conflictPlans,
	}
}

func (a *pausePrecheckAction) Name() string { return PausePrecheckActionName }

func (a *pausePrecheckAction) Invoke(args actions.ActionArgs) error {
	ctx := utils.GenerateContext(args)
	sharedPoolID, err := args.GetString(workflow.SharedPoolIdKey)
	if err != nil {
		return actions.NewTerminalError(err)
	}
	sharedPool, err := a.sharedPoolRepo.Load(ctx, sharedPoolID, false)
	if err != nil {
		return err
	}
	if sharedPool.PauseResumeState != entity.PauseResumeStatePausing {
		return actions.NewTerminalError(fmt.Errorf("shared pool pause state changed before DAG precheck"))
	}
	operationID, err := args.GetString(workflow.PauseResumeOperationIdKey)
	if err != nil {
		return actions.NewTerminalError(err)
	}
	operationType, err := args.GetString(workflow.PauseResumeOperationTypeKey)
	if err != nil {
		return actions.NewTerminalError(err)
	}
	selfExecID := workflow.PauseResumeExecID(sharedPoolID, operationID, operationType)
	for _, plan := range append(append([]string{}, a.pausePlans...), a.conflictPlans...) {
		filterKeys := []string{workflow.SharedPoolIdKey}
		if isCrossSharedPoolRestorePlan(plan) {
			filterKeys = append(filterKeys, workflow.SourceSharedPoolIdKey)
		}
		for _, filterKey := range filterKeys {
			running, err := a.dagQuery(plan, executor.NewMatchFilter().Add(filterKey, []string{sharedPoolID}))
			if err != nil {
				return fmt.Errorf("query conflicting DAG %s: %w", plan, err)
			}
			for execID := range running {
				if execID != selfExecID {
					// Admission checks normally prevent a conflicting workflow from being
					// accepted. This is still needed for a request that passed admission
					// immediately before the pool transitioned to PAUSING. Keep the pause
					// operation retryable until that workflow converges, rather than
					// failing it and stranding the shared pool in PAUSING.
					return fmt.Errorf("conflicting DAG %s is running", plan)
				}
			}
		}
	}
	remoteCli, err := a.dataplane.GetClient(sharedPool.GetDataplaneName())
	if err != nil {
		return err
	}
	operatorStatus, err := tioperator.InspectClusterSuspendCompute(ctx, remoteCli, sharedPoolID)
	if err != nil {
		return err
	}
	if operatorStatus.SpecDesired && !sharedPool.SuspendComputeDesired {
		return actions.NewTerminalError(fmt.Errorf("operator suspend spec is true while suspend_compute_desired is false"))
	}
	if operatorStatus.SpecDesired && operatorStatus.Observed {
		return nil
	}
	if _, err := braction.ValidateFlushWALPrerequisites(ctx, sharedPoolID, a.sharedPoolRepo, a.dataplaneRepo, a.dataplane, remoteCli, a.disableFlushWAL); err != nil {
		if braction.IsTerminalFlushWALPrerequisiteError(err) {
			return actions.NewTerminalError(err)
		}
		return err
	}
	return nil
}

func isCrossSharedPoolRestorePlan(plan string) bool {
	switch plan {
	case "create-lightweight-x-restore", "delete-lightweight-x-restore",
		"create-manual-backup-x-restore", "delete-manual-backup-x-restore":
		return true
	default:
		return false
	}
}

type setSuspendComputeAction struct {
	name           string
	desired        bool
	state          entity.PauseResumeState
	sharedPoolRepo repo.SharedPoolRepo
	dataplane      dataplane.Manager
}

func NewSetSuspendComputeAction(name string, desired bool, state entity.PauseResumeState, sharedPoolRepo repo.SharedPoolRepo, manager dataplane.Manager) actions.Action {
	return &setSuspendComputeAction{name: name, desired: desired, state: state, sharedPoolRepo: sharedPoolRepo, dataplane: manager}
}

func (a *setSuspendComputeAction) Name() string { return a.name }

func (a *setSuspendComputeAction) Invoke(args actions.ActionArgs) error {
	ctx := utils.GenerateContext(args)
	ctx, cancel := context.WithTimeout(ctx, actions.GuardedWriteTimeout)
	defer cancel()
	sharedPoolID, err := args.GetString(workflow.SharedPoolIdKey)
	if err != nil {
		return err
	}
	sharedPool, err := a.sharedPoolRepo.Load(ctx, sharedPoolID, false)
	if err != nil {
		return err
	}
	if err := a.sharedPoolRepo.UpdatePauseResumeState(ctx, sharedPoolID, a.state, a.desired); err != nil {
		return err
	}
	remoteCli, err := a.dataplane.GetClient(sharedPool.GetDataplaneName())
	if err != nil {
		return err
	}
	return tioperator.SetClusterSuspendCompute(ctx, remoteCli, sharedPoolID, a.desired)
}

type waitSuspendComputeAction struct {
	name           string
	desired        bool
	sharedPoolRepo repo.SharedPoolRepo
	dataplane      dataplane.Manager
}

func NewWaitSuspendComputeAction(name string, desired bool, sharedPoolRepo repo.SharedPoolRepo, manager dataplane.Manager) actions.AsyncAction {
	return &waitSuspendComputeAction{name: name, desired: desired, sharedPoolRepo: sharedPoolRepo, dataplane: manager}
}

func (a *waitSuspendComputeAction) Name() string                   { return a.name }
func (a *waitSuspendComputeAction) Start(actions.ActionArgs) error { return nil }

func (a *waitSuspendComputeAction) CheckFinish(args actions.ActionArgs) (bool, error) {
	ctx := utils.GenerateContext(args)
	sharedPoolID, err := args.GetString(workflow.SharedPoolIdKey)
	if err != nil {
		return false, err
	}
	sharedPool, err := a.sharedPoolRepo.Load(ctx, sharedPoolID, false)
	if err != nil {
		return false, err
	}
	remoteCli, err := a.dataplane.GetClient(sharedPool.GetDataplaneName())
	if err != nil {
		return false, err
	}
	return tioperator.IsClusterSuspendComputeObserved(ctx, remoteCli, sharedPoolID, a.desired)
}

type completePauseResumeAction struct {
	name           string
	state          entity.PauseResumeState
	desired        bool
	sharedPoolRepo repo.SharedPoolRepo
	operationRepo  repo.PauseResumeOperationRepo
	transactioner  db.Transactioner
	lockRepo       repo.LockRepo
}

func NewCompletePauseResumeAction(name string, state entity.PauseResumeState, desired bool, sharedPoolRepo repo.SharedPoolRepo, operationRepo repo.PauseResumeOperationRepo, transactioner db.Transactioner, lockRepo repo.LockRepo) actions.Action {
	return &completePauseResumeAction{name: name, state: state, desired: desired, sharedPoolRepo: sharedPoolRepo, operationRepo: operationRepo, transactioner: transactioner, lockRepo: lockRepo}
}

func (a *completePauseResumeAction) Name() string { return a.name }

func (a *completePauseResumeAction) Invoke(args actions.ActionArgs) error {
	ctx := utils.GenerateContext(args)
	sharedPoolID, err := args.GetString(workflow.SharedPoolIdKey)
	if err != nil {
		return actions.NewTerminalError(err)
	}
	unlock, err := a.lockRepo.Lock(ctx, &entity.Lock{Key: sharedPoolID, TimeoutSecs: 60})
	if err != nil {
		return err
	}
	defer func() { _ = unlock() }()
	return a.transactioner.Transaction(ctx, func(txCtx context.Context) error {
		operation, err := loadActionOperation(txCtx, a.operationRepo, args)
		if err != nil {
			return err
		}
		if operation.GetStatus() != apiv1.PauseResumeOperation_RUNNING {
			return actions.NewTerminalError(fmt.Errorf("pause/resume operation is no longer running"))
		}
		if err := a.sharedPoolRepo.UpdatePauseResumeState(txCtx, operation.GetSharedPoolId(), a.state, a.desired); err != nil {
			return err
		}
		updated, err := a.operationRepo.MarkSucceededIfRunning(txCtx, operation.GetSharedPoolId(), operation.GetOperationId(), "", time.Now())
		if err != nil {
			return err
		}
		if !updated {
			return actions.NewTerminalError(errPauseResumeOwnershipLost)
		}
		return nil
	})
}

const pauseResumeWaitInterval = 10 * time.Second
const pauseResumeWaitTimeout = 60 * time.Minute

func NewPauseResumeWaitAction(asyncAction actions.AsyncAction) actions.Action {
	return actions.CreateActionFromAsync(asyncAction, pauseResumeWaitInterval, pauseResumeWaitTimeout)
}