package server

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

	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"
	"github.com/tidbcloud/cluster-service-ng/domain/workflow/dag"
	"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/utils"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/encoding/protojson"
	"gorm.io/gorm"
)

func (s *serverImpl) DeleteSharedPool(ctx context.Context, sharedPoolId string, pruneDataplane bool) (*entity.SharedPool, error) {
	if sharedPoolId == "" {
		return nil, status.Errorf(codes.InvalidArgument, "sharedPool id is required")
	}

	unlock, err := s.lockWithTimeout(ctx, sharedPoolId, 60)
	if err != nil {
		tilog.Error(ctx, "failed to lock sharedPool", zap.String("sharedPoolId", sharedPoolId), zap.Error(err))
		return nil, err
	}
	defer func() {
		if err := unlock(); err != nil {
			tilog.Error(ctx, "failed to unlock sharedPool", zap.String("sharedPoolId", sharedPoolId), zap.Error(err))
		}
	}()

	sharedPool, err := s.sharedPoolRepo.Load(ctx, sharedPoolId, false)
	if err != nil {
		if status.Code(err) != codes.NotFound {
			return nil, status.Errorf(codes.Internal, "failed to load sharedPool: %s", err)
		}
		return nil, status.Errorf(codes.NotFound, "sharedPool %s not found", sharedPoolId)
	}
	if sharedPool.Phase == apiv1.SharedPool_DELETING {
		return sharedPool, nil
	}
	if err := guardSharedPoolMutation(sharedPool, sharedPoolMutationDeletePool); err != nil {
		return nil, err
	}

	// if sharedpool has any available tidb cluster, we should not delete it
	tidbClusters, err := s.tidbClusterRepo.List(ctx, &repo.TiDBClusterListFilter{
		SharedPoolIds: []string{sharedPoolId},
		Deleted:       false,
	})
	if err != nil {
		return nil, status.Errorf(codes.Internal, "failed to list tidb clusters: %s", err)
	}
	if len(tidbClusters) > 0 {
		tilog.Warn(ctx, "sharedPoolId has available tidb clusters, please delete them first", zap.String("sharedPoolId", sharedPoolId))
		return nil, status.Error(codes.FailedPrecondition, "sharedPoolId has available tidb clusters, please delete them first")
	}

	// delete all dm clusters belonging to this sharedpool
	dmClusters, err := s.dmClusterRepo.List(ctx, &repo.DMClusterListFilter{
		SharedPoolIds: []string{sharedPoolId},
		Deleted:       false,
	})
	if err != nil {
		return nil, status.Errorf(codes.Internal, "failed to list dm clusters: %s", err)
	}
	transitional := sharedPool.PauseResumeState == entity.PauseResumeStatePausing || sharedPool.PauseResumeState == entity.PauseResumeStateResuming
	if len(dmClusters) > 0 && transitional {
		// TODO: DM owns independent workflows and external writes that do not yet
		// participate in Pause/Resume ownership fencing. Keep Delete fail-closed
		// until the DM team defines cancellation and takeover semantics.
		return nil, status.Error(codes.FailedPrecondition, "sharedPool has DM clusters; delete them before deleting the shared pool")
	}
	if len(dmClusters) > 0 {
		tilog.Info(ctx, "sharedPool has dm clusters, deleting them first", zap.String("sharedPoolId", sharedPoolId), zap.Int("dmClusterCount", len(dmClusters)))
		for _, dmCluster := range dmClusters {
			if err := s.DeleteDMCluster(ctx, sharedPoolId, dmCluster.GetId()); err != nil {
				tilog.Error(ctx, "failed to delete dm cluster", zap.String("sharedPoolId", sharedPoolId), zap.String("dmClusterId", dmCluster.GetId()), zap.Error(err))
				return nil, status.Errorf(codes.Internal, "failed to delete dm cluster %s: %s", dmCluster.GetId(), err)
			}
		}
	}

	// query deleting tidb groups - must wait before proceeding to avoid a race where
	// ClearDataPlaneAction destroys the dataplane while a delete-tidbgroup DAG is still
	// running ApplyTiDBGroupAction, causing that DAG to fail and leaving its TiDBGroupAgent
	// orphaned forever.
	deleteGroupIns, err := s.getRunningDagIns(ctx,
		[]string{dag.PlanDeleteTiDBGroup},
		[]string{sharedPool.GetId()},
		nil,
		nil,
		nil)
	if err != nil {
		return nil, status.Errorf(codes.Internal, "failed to get running delete tidb group dag instance: %s", err)
	}
	if len(deleteGroupIns) > 0 {
		tilog.Info(ctx, "sharedPool has deleting tidb groups, please wait for them to be deleted first", zap.String("sharedPoolId", sharedPoolId), zap.Int("deletingTidbGroupCount", len(deleteGroupIns)))
		return nil, status.Error(codes.FailedPrecondition, "sharedPool has deleting tidb groups, please wait for them to be deleted first")
	}

	// query deleting tidb clusters
	deleteClusterIns, err := s.getRunningDagIns(ctx,
		[]string{dag.PlanDeleteTiDBCluster},
		[]string{sharedPool.GetId()},
		nil,
		nil,
		nil)
	if err != nil {
		return nil, status.Errorf(codes.Internal, "failed to get running delete tidb cluster dag instance: %s", err)
	}
	if len(deleteClusterIns) > 0 {
		tilog.Info(ctx, "sharedPool has deleting tidb clusters, please wait for them to be deleted first", zap.String("sharedPoolId", sharedPoolId), zap.Int("deletingTidbClusterCount", len(deleteClusterIns)))
		return nil, status.Error(codes.FailedPrecondition, "sharedPool has deleting tidb clusters, please wait for them to be deleted first")
	}

	// Remote-worker switching changes the same TiKV/COPROCESSOR topology that
	// teardown destroys. Abort is asynchronous and no completion signal is
	// available, so fail closed. Note a failing enable/disable DAG never
	// reaches a terminal state on its own - its errored action is
	// rescheduled indefinitely - so "retry later" only works once an
	// operator has manually aborted the DAG via the executor API.
	remoteWorkerIns, err := s.getRunningDagIns(ctx, tikvRemoteWorkersDAGPlans(), []string{sharedPool.GetId()}, nil, nil, nil)
	if err != nil {
		return nil, status.Errorf(codes.Internal, "failed to get running tikv remote worker dag instance: %s", err)
	}
	if len(remoteWorkerIns) > 0 {
		return nil, status.Errorf(codes.FailedPrecondition, "tikv remote worker routing switch is running for shared pool %s", sharedPoolId)
	}

	if transitional {
		// TODO: independent Backup/Restore operations have no common owner fence
		// with Pool Delete yet. Detect the CSNG-visible DAGs, but never Abort or
		// claim takeover until the owning teams add a durable cancellation model.
		running, queryErr := s.getRunningDagIns(ctx, dag.PauseResumeBackupRestorePlans(), []string{sharedPoolId}, nil, nil, nil)
		if queryErr != nil {
			return nil, status.Errorf(codes.Internal, "failed to query backup or restore conflicts: %s", queryErr)
		}
		if len(running) > 0 {
			return nil, status.Errorf(codes.FailedPrecondition, "backup or restore operation is running for shared pool %s", sharedPoolId)
		}
		running, queryErr = s.queryRunningDagIns(ctx, dag.PauseResumeBackupRestorePlans(), executor.NewMatchFilter().Add(workflow.SourceSharedPoolIdKey, []string{sharedPoolId}))
		if queryErr != nil {
			return nil, status.Errorf(codes.Internal, "failed to query backup or restore source-pool conflicts: %s", queryErr)
		}
		if len(running) > 0 {
			return nil, status.Errorf(codes.FailedPrecondition, "backup or restore operation is running for shared pool %s", sharedPoolId)
		}
	}

	needCancelPlans := []string{
		dag.PlanCreateSharedPool,
		dag.PlanApplyComponent,
		dag.PlanUpdateComponent,
		dag.PlanCreateTiDBCluster,
		dag.PlanUpdateTiDBCluster,
		dag.PlanCreateTiDBGroup,
		dag.PlanUpdateTiDBGroup,
	}
	dagIns, err := s.getRunningDagIns(ctx, needCancelPlans, []string{sharedPool.GetId()}, nil, nil, nil)
	if err != nil {
		return nil, status.Errorf(codes.Internal, "failed to get running dag instances: %s", err)
	}

	var takeoverOperation *entity.PauseResumeOperation
	if transitional {
		takeoverOperation, err = s.pauseResumeOperationRepo.LoadLatestBySharedPoolID(ctx, sharedPoolId)
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return nil, status.Error(codes.FailedPrecondition, "transitional shared pool has no pause/resume journal")
		}
		if err != nil {
			return nil, status.Errorf(codes.Internal, "failed to load pause/resume journal: %s", err)
		}
		expectedType := apiv1.PauseResumeOperation_PAUSE
		if sharedPool.PauseResumeState == entity.PauseResumeStateResuming {
			expectedType = apiv1.PauseResumeOperation_RESUME
		}
		if takeoverOperation.GetType() != expectedType {
			return nil, status.Error(codes.FailedPrecondition, "pause/resume journal does not match transitional shared pool")
		}
		switch takeoverOperation.GetStatus() {
		case apiv1.PauseResumeOperation_RUNNING:
			if err := s.transactioner.Transaction(ctx, func(txCtx context.Context) error {
				cancelled, err := s.pauseResumeOperationRepo.CancelIfRunning(txCtx, sharedPoolId, takeoverOperation.GetOperationId(), "cancelled by shared pool deletion", time.Now())
				if err != nil {
					return err
				}
				if !cancelled {
					return status.Error(codes.Aborted, "pause/resume operation changed during delete takeover")
				}
				return nil
			}); err != nil {
				return nil, err
			}
		case apiv1.PauseResumeOperation_FAILED, apiv1.PauseResumeOperation_CANCELLED:
			// A previous failure or Delete attempt has already revoked the owner.
		case apiv1.PauseResumeOperation_SUCCEEDED, apiv1.PauseResumeOperation_STATUS_UNSPECIFIED:
			return nil, status.Error(codes.FailedPrecondition, "pause/resume journal does not match transitional shared pool")
		default:
			return nil, status.Error(codes.FailedPrecondition, "pause/resume journal has an unsupported status")
		}
	}

	if takeoverOperation != nil {
		execID := workflow.PauseResumeExecID(sharedPoolId, takeoverOperation.GetOperationId(), takeoverOperation.GetType().String())
		if abortErr := s.dagExecutor.Abort(ctx, execID); abortErr != nil {
			tilog.Warn(ctx, "failed to abort cancelled pause/resume DAG", zap.String("execID", execID), zap.Error(abortErr))
		}
	}
	for id, task := range dagIns {
		// abort create cluster dag
		tilog.Info(ctx, "aborting sharedpool dag", zap.String("exec id", id), zap.String("root name", task.ExecTask.RootName))
		if err := s.dagExecutor.Abort(ctx, task.ExecTask.ExecId); err != nil {
			tilog.Error(ctx, "failed to abort sharedpool dag", zap.String("task id", task.ExecTask.ExecId), zap.String("root name", task.ExecTask.RootName), zap.Error(err))
			return nil, status.Errorf(codes.Internal, "failed to abort sharedpool dag %s: %s", task.ExecTask.ExecId, err)
		}
	}

	// set sharedpool phase to deleting
	sharedPool.Phase = apiv1.SharedPool_DELETING
	deleteIns, err := s.getRunningDagIns(ctx, []string{
		dag.PlanDeleteSharedPool,
	}, []string{sharedPool.GetId()}, nil, nil, nil)
	if err != nil {
		return nil, status.Errorf(codes.Internal, "failed to get running delete dag instance: %s", err)
	}
	var dagExecId string
	dagSubmitted := false
	err = s.transactioner.Transaction(ctx, func(txCtx context.Context) error {
		if err := s.deleteSharedPoolMeta(txCtx, sharedPoolId); err != nil {
			return err
		}
		if len(deleteIns) > 0 {
			tilog.Info(ctx, "delete sharedpool dag is running", zap.Any("deleteIns", deleteIns))
			return nil
		}
		sharedPoolStr, err := protojson.Marshal(sharedPool)
		if err != nil {
			return status.Errorf(codes.Internal, "failed to marshal sharedPool: %s", err)
		}

		// start delete cluster dag
		deleteTriggeredAt := time.Now().Unix()
		args := actions.ActionArgs{
			workflow.PruneDataplaneKey:           actions.ActionArgString(strconv.FormatBool(pruneDataplane)),
			workflow.SharedPoolIdKey:             actions.ActionArgString(sharedPool.GetId()),
			workflow.SharedPoolKey:               actions.ActionArgString(string(sharedPoolStr)),
			workflow.MeteringTypeKey:             actions.ActionArgString(workflow.MeteringTypeValueSharedPoolMeta),
			workflow.MeteringDeleteSharedPoolKey: actions.ActionArgString(workflow.MeteringDeleteSharedPoolValue),
			workflow.MeteringModifyTSKey:         actions.ActionArgString(strconv.FormatInt(deleteTriggeredAt, 10)),
			workflow.TenantIdKey:                 actions.ActionArgString(sharedPool.TenantId),
			utils.TraceIdKey:                     utils.GetTraceId(txCtx),
		}
		dagExecId = utils.GeneratedDagExecId(txCtx, fmt.Sprintf("delete-sharedpool-%s", sharedPool.GetId()))
		if err := s.dagExecutor.Submit(txCtx, dag.PlanDeleteSharedPool, args, dagExecId); err != nil {
			return status.Errorf(codes.Internal, "failed to submit delete cluster dag: %s", err)
		}
		dagSubmitted = true
		return nil
	})
	if err != nil {
		if dagSubmitted {
			if abortErr := s.dagExecutor.Abort(ctx, dagExecId); abortErr != nil {
				tilog.Error(ctx, "failed to abort dag after transaction failure",
					zap.String("dagExecId", dagExecId),
					zap.Error(abortErr))
			}
		}
		return nil, err
	}
	return sharedPool, nil
}

func (s *serverImpl) deleteSharedPoolMeta(ctx context.Context, sharedPoolId string) error {
	if err := s.sharedPoolRepo.Delete(ctx, sharedPoolId); err != nil {
		tilog.Error(ctx, "failed to delete shared pool meta", zap.Error(err))
		return fmt.Errorf("failed to delete shared pool meta: %w", err)
	}

	if err := s.ruConfigRepo.DeleteRuConfig(ctx, sharedPoolId, apiv1.RuConfig_SHARED_POOL); err != nil {
		tilog.Error(ctx, "failed to delete sharedPool ruConfig meta", zap.Error(err))
		return fmt.Errorf("failed to delete sharedPool ruConfig meta: %w", err)
	}

	return nil
}