package action

import (
	"context"
	"errors"
	"fmt"
	"strconv"
	"time"

	clusterngv1beta "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	eventdata "github.com/tidbcloud/eventbus-datastruct/event"
	"github.com/tidbcloud/fastflow/pkg/entity/run"
	"github.com/tidbcloud/pkgv2/db"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"google.golang.org/protobuf/encoding/protojson"

	globalngv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/nextgen/v1"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/adapter/eventbus"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/adapter/regionalserver"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/entity"
	brrepo "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/br"
	clusterrepo "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/cluster"
	regionrepo "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/region"
	resourcepoolrepo "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/resourcepool"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/resourcepoolpolicy"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/service/displayevent"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/service/workflow/common"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/infra/repoimpl/schema"
	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
	"github.com/tidbcloud/tidb-management-service/utilities/workflow"
)

const (
	CreateManualBackup = "create_manual_backup"
)

type CreateManualBackupImpl struct {
	backupRepo         brrepo.BackupRepo
	regionRepo         regionrepo.Repo
	clusterRepo        clusterrepo.Repo
	resourcePoolRepo   resourcepoolrepo.Repo
	pauseOperationRepo resourcepoolrepo.PauseOperationRepo
	regionalserverCli  regionalserver.Client
	transaction        db.Transactioner
	eventCommitter     eventbus.WALCommitter
}

func NewCreateManualBackup(
	backupRepo brrepo.BackupRepo,
	regionRepo regionrepo.Repo,
	clusterRepo clusterrepo.Repo,
	resourcePoolRepo resourcepoolrepo.Repo,
	pauseOperationRepo resourcepoolrepo.PauseOperationRepo,
	regionalserverCli regionalserver.Client,
	transaction db.Transactioner,
	eventCommitter eventbus.WALCommitter,
) *CreateManualBackupImpl {
	return &CreateManualBackupImpl{
		backupRepo:         backupRepo,
		regionRepo:         regionRepo,
		clusterRepo:        clusterRepo,
		resourcePoolRepo:   resourcePoolRepo,
		pauseOperationRepo: pauseOperationRepo,
		regionalserverCli:  regionalserverCli,
		transaction:        transaction,
		eventCommitter:     eventCommitter,
	}
}

func (a *CreateManualBackupImpl) Name() string {
	return CreateManualBackup
}

func (a *CreateManualBackupImpl) Run(ctx run.ExecuteContext, params any) error {
	cctx, span := workflow.GenerateContext(ctx)
	defer span.End()
	tilog.Info(cctx, "create manual backup action start...")

	request, region, backupId, sharedPoolID, err := getManualBackupCtxInfo(cctx, ctx, a.regionRepo, a.clusterRepo)
	if err != nil {
		return err
	}
	resourcePoolID, err := strconv.ParseUint(sharedPoolID, 10, 64)
	if err != nil || resourcePoolID == 0 {
		return fmt.Errorf("failed to parse shared pool id %q", sharedPoolID)
	}
	backup, berr := a.backupRepo.Get(cctx, backupId)
	if berr != nil {
		tilog.Error(cctx, "failed to find backup", zap.Error(berr), zap.Uint64("backup_id", backupId))
		return fmt.Errorf("failed to find backup: %w", berr)
	}
	preparationBackup := backup.PauseOperationId != nil
	if backup.State != schema.BackupStatePending &&
		(!preparationBackup || backup.State != schema.BackupStateSubmitting) {
		tilog.Info(cctx, "backup is no longer pending", zap.Uint64("backup_id", backupId), zap.String("state", string(backup.State)))
		return nil
	}
	if preparationBackup {
		if err := a.submitPausePreparationBackup(
			cctx,
			resourcePoolID,
			backupId,
			region,
			request,
			*backup.PauseOperationId,
		); err != nil {
			tilog.Error(cctx, "failed to submit pause preparation backup", zap.Error(err), zap.Uint64("backup_id", backupId))
			return err
		}
		return nil
	}
	operation := resourcepoolpolicy.OperationManualBackup
	stop, err := guardResourcePoolMutation(
		cctx,
		a.resourcePoolRepo,
		resourcePoolID,
		operation,
	)
	if err != nil {
		return err
	}
	if stop {
		return nil
	}
	if berr = a.regionalserverCli.CreateManualBackup(cctx, region, request); berr != nil {
		tilog.Error(cctx, "send create manual backup request failed", zap.Error(berr), zap.Any("request", request))
		return fmt.Errorf("send create manual backup request failed: %w", berr)
	}
	tilog.Info(cctx, "create manual backup request sent", zap.Any("request", request))

	err = a.transaction.Transaction(cctx, func(ctx context.Context) error {
		backup, berr = a.backupRepo.Get(ctx, backupId)
		if berr != nil {
			return fmt.Errorf("failed to find backup after submission: %w", berr)
		}
		if backup.State != schema.BackupStatePending {
			return nil
		}
		backup.State = schema.BackupStateRunning
		if berr = a.backupRepo.Save(ctx, backup); berr != nil {
			return fmt.Errorf("failed to save backup state to running: %w", berr)
		}
		displayEv := displayevent.New(
			ctx,
			displayevent.BackupCluster,
			eventdata.EventStatusRunning,
			backup.TenantId,
			backup.ClusterId,
			strconv.FormatUint(backup.ID, 10),
			"Manual backup is running.",
			displayevent.BackupDetail{
				BackupID:    strconv.FormatUint(backup.ID, 10),
				DisplayName: backup.DisplayName,
				BackupType:  string(backup.Type),
				TriggerType: string(backup.TriggerType),
				State:       string(backup.State),
			},
		)
		displayevent.Write(ctx, a.eventCommitter, displayEv)
		return nil
	})
	if err != nil {
		tilog.Error(cctx, "failed to persist submitted manual backup", zap.Error(err), zap.Uint64("backup_id", backupId))
		return fmt.Errorf("failed to persist submitted manual backup: %w", err)
	}

	return nil
}

func (a *CreateManualBackupImpl) submitPausePreparationBackup(
	ctx context.Context,
	resourcePoolID uint64,
	backupID uint64,
	region *entity.Region,
	request *clusterngv1beta.CreateManualBackupRequest,
	pauseOperationID string,
) error {
	var shouldSubmit, stale, expired bool
	err := a.transaction.Transaction(ctx, func(ctx context.Context) error {
		resourcePool, berr := a.resourcePoolRepo.LoadForUpdate(ctx, resourcePoolID, false)
		if berr != nil {
			return berr
		}
		backup, berr := a.backupRepo.Get(ctx, backupID)
		if berr != nil {
			return berr
		}
		if backup.State != schema.BackupStatePending && backup.State != schema.BackupStateSubmitting {
			return nil
		}

		owned := resourcePool.GetLifecycleState() == globalngv1.ResourcePool_PAUSE_PRECHECKING
		var operation *entity.ResourcePoolPauseOperation
		if owned {
			operation, operationErr := a.pauseOperationRepo.GetRunning(ctx, resourcePoolID)
			switch {
			case operationErr == nil:
				owned = operation != nil && operation.OperationID == pauseOperationID
			case operationErr.ErrCode() == bizerr.ErrCodeRecordNotFound:
				owned = false
			default:
				return operationErr
			}
		}
		if !owned {
			backup.State = schema.BackupStateCancelled
			stale = true
			return a.backupRepo.Save(ctx, backup)
		}
		if isExpired, deadlineErr := pausePreparationExpired(operation, time.Now().UTC()); isExpired {
			now := time.Now().UTC()
			backup.State = schema.BackupStateCancelled
			if berr := a.backupRepo.Save(ctx, backup); berr != nil {
				return berr
			}
			resourcePool.LifecycleState = globalngv1.ResourcePool_ACTIVE
			resourcePool.StateMessageCode = ""
			if berr := a.resourcePoolRepo.Save(ctx, resourcePool); berr != nil {
				return berr
			}
			operation.Status = entity.ResourcePoolPauseOperationFailed
			operation.ResultSummary = deadlineErr.Error()
			operation.CompletedAt = &now
			if berr := a.pauseOperationRepo.Save(ctx, operation); berr != nil {
				return berr
			}
			expired = true
			return nil
		}
		if backup.State == schema.BackupStatePending {
			backup.State = schema.BackupStateSubmitting
			if berr := a.backupRepo.Save(ctx, backup); berr != nil {
				return fmt.Errorf("failed to claim backup submission: %w", berr)
			}
		}
		shouldSubmit = true
		return nil
	})
	if err != nil {
		return err
	}
	if stale {
		return workflow.NewTerminalError(fmt.Errorf("pause preparation backup belongs to an inactive operation"))
	}
	if expired {
		return workflow.NewTerminalError(fmt.Errorf("pause preparation deadline exceeded"))
	}
	if !shouldSubmit {
		return nil
	}

	if berr := a.regionalserverCli.CreateManualBackup(ctx, region, request); berr != nil {
		terminal := isTerminalPausePreparationError(berr)
		persistErr := a.persistPauseBackupSubmissionResult(ctx, backupID, false, terminal)
		resultErr := fmt.Errorf("send create manual backup request failed: %w", berr)
		if persistErr != nil {
			resultErr = errors.Join(resultErr, persistErr)
		}
		if terminal {
			return workflow.NewTerminalError(resultErr)
		}
		return resultErr
	}
	return a.persistPauseBackupSubmissionResult(ctx, backupID, true, false)
}

func (a *CreateManualBackupImpl) persistPauseBackupSubmissionResult(
	ctx context.Context,
	backupID uint64,
	submitted bool,
	terminalFailure bool,
) error {
	return a.transaction.Transaction(ctx, func(ctx context.Context) error {
		backup, berr := a.backupRepo.Get(ctx, backupID)
		if berr != nil {
			return berr
		}
		if backup.State != schema.BackupStateSubmitting {
			return nil
		}
		switch {
		case submitted:
			backup.State = schema.BackupStateRunning
		case terminalFailure:
			backup.State = schema.BackupStateFailed
		default:
			backup.State = schema.BackupStatePending
		}
		if berr := a.backupRepo.Save(ctx, backup); berr != nil {
			return fmt.Errorf("failed to save backup submission result: %w", berr)
		}
		if !submitted {
			return nil
		}
		displayEv := displayevent.New(
			ctx,
			displayevent.BackupCluster,
			eventdata.EventStatusRunning,
			backup.TenantId,
			backup.ClusterId,
			strconv.FormatUint(backup.ID, 10),
			"Manual backup is running.",
			displayevent.BackupDetail{
				BackupID:    strconv.FormatUint(backup.ID, 10),
				DisplayName: backup.DisplayName,
				BackupType:  string(backup.Type),
				TriggerType: string(backup.TriggerType),
				State:       string(backup.State),
			},
		)
		displayevent.Write(ctx, a.eventCommitter, displayEv)
		return nil
	})
}

func getManualBackupCtxInfo(
	cctx context.Context,
	ctx run.ExecuteContext,
	regionRepo regionrepo.Repo,
	clusterRepo clusterrepo.Repo,
) (*clusterngv1beta.CreateManualBackupRequest, *entity.Region, uint64, string, error) {
	requestJson, ok := ctx.GetVar(common.KeyRequest)
	if !ok {
		return nil, nil, 0, "", fmt.Errorf("manual backup request is not found in context")
	}

	var request clusterngv1beta.CreateManualBackupRequest
	if err := protojson.Unmarshal([]byte(requestJson), &request); err != nil {
		tilog.Error(cctx, "unmarshal create manual backup request failed", zap.Error(err), zap.String("request", requestJson))
		return nil, nil, 0, "", fmt.Errorf("unmarshal create manual backup request failed: %w", err)
	}

	backupId, err := strconv.ParseUint(request.GetBackupId(), 10, 64)
	if err != nil {
		return nil, nil, 0, "", fmt.Errorf("failed to parse backup id: %w", err)
	}

	clusterId, err := strconv.ParseUint(request.GetClusterId(), 10, 64)
	if err != nil {
		return nil, nil, 0, "", fmt.Errorf("failed to parse cluster id: %w", err)
	}

	cluster, berr := clusterRepo.Get(cctx, clusterId, true)
	if berr != nil {
		return nil, nil, 0, "", berr.Chain("failed to get cluster")
	}

	region, err := regionRepo.Get(cctx, cluster.RegionId)
	if err != nil {
		tilog.Error(cctx, "failed to find region", zap.Error(err), zap.String("cluster_id", request.GetClusterId()))
		return nil, nil, 0, "", fmt.Errorf("failed to find region: %w", err)
	}

	return &request, region, backupId, cluster.GetSharedPoolId(), nil
}