package br

import (
	"context"
	"fmt"
	"strings"
	"time"

	"github.com/tidbcloud/cluster-service-ng/domain/workflow"
	"github.com/tidbcloud/cluster-service-ng/pkg/dataplane"
	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/actions"
	"github.com/tidbcloud/cluster-service-ng/pkg/storage/repo"
	"github.com/tidbcloud/cluster-service-ng/pkg/utils"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	batchv1 "k8s.io/api/batch/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

const (
	DrainArchiveBeforeResumeActionName = "drain-pause-archive-jobs-before-resume"
	DrainArchiveBeforeDeleteActionName = "drain-pause-archive-jobs-before-delete"
)

type drainResumeArchiveJobsAction struct {
	name             string
	guard            actions.ActionGuard
	sharedPoolRepo   repo.SharedPoolRepo
	dataplaneManager dataplane.Manager
}

func NewDrainArchiveBeforeResumeAction(sharedPoolRepo repo.SharedPoolRepo, manager dataplane.Manager, guard actions.ActionGuard) actions.Action {
	asyncAction := &drainResumeArchiveJobsAction{name: DrainArchiveBeforeResumeActionName, guard: guard, sharedPoolRepo: sharedPoolRepo, dataplaneManager: manager}
	return actions.CreateGuardedActionFromAsync(asyncAction, 10*time.Second, 10*time.Minute, guard)
}

func NewDrainArchiveBeforeDeleteAction(sharedPoolRepo repo.SharedPoolRepo, manager dataplane.Manager) actions.Action {
	return actions.CreateActionFromAsync(&drainResumeArchiveJobsAction{name: DrainArchiveBeforeDeleteActionName, sharedPoolRepo: sharedPoolRepo, dataplaneManager: manager}, 10*time.Second, 10*time.Minute)
}

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

func (a *drainResumeArchiveJobsAction) Start(args actions.ActionArgs) error {
	return a.keepResumeRetryInsideInvocation(a.drain(args))
}

func (a *drainResumeArchiveJobsAction) drain(args actions.ActionArgs) error {
	ctx := utils.GenerateContext(args)
	if a.guard != nil {
		var cancel context.CancelFunc
		ctx, cancel = context.WithTimeout(ctx, actions.GuardedWriteTimeout)
		defer cancel()
	}
	jobs, remoteClient, err := a.list(ctx, args)
	if err != nil {
		return err
	}
	foreground := metav1.DeletePropagationForeground
	for i := range jobs.Items {
		job := &jobs.Items[i]
		if !strings.HasPrefix(job.Name, t3ArchiveJobPrefix()) {
			continue
		}
		if err := remoteClient.Delete(ctx, job, &client.DeleteOptions{PropagationPolicy: &foreground}); err != nil && !apierrors.IsNotFound(err) {
			return a.classifyResumeKubernetesError(fmt.Errorf("delete pause archive job %s: %w", job.Name, err))
		}
	}
	return nil
}

func (a *drainResumeArchiveJobsAction) CheckFinish(args actions.ActionArgs) (bool, error) {
	jobs, _, err := a.list(utils.GenerateContext(args), args)
	if err != nil {
		return false, a.keepResumeRetryInsideInvocation(err)
	}
	for i := range jobs.Items {
		if strings.HasPrefix(jobs.Items[i].Name, t3ArchiveJobPrefix()) {
			if a.guard != nil {
				err := a.guard(args, func() error { return a.drain(args) })
				return false, a.keepResumeRetryInsideInvocation(err)
			}
			return false, nil
		}
	}
	return true, nil
}

func (a *drainResumeArchiveJobsAction) list(ctx context.Context, args actions.ActionArgs) (*batchv1.JobList, client.Client, error) {
	sharedPoolID, err := args.GetString(workflow.SharedPoolIdKey)
	if err != nil {
		return nil, nil, err
	}
	sharedPool, err := a.sharedPoolRepo.Load(ctx, sharedPoolID, true)
	if err != nil {
		return nil, nil, fmt.Errorf("load shared pool for archive drain: %w", err)
	}
	remoteClient, err := a.dataplaneManager.GetClient(sharedPool.GetDataplaneName())
	if err != nil {
		if a.guard == nil && status.Code(err) == codes.NotFound {
			return &batchv1.JobList{}, nil, nil
		}
		return nil, nil, fmt.Errorf("get dataplane client for archive drain: %w", err)
	}
	jobs := &batchv1.JobList{}
	if err := remoteClient.List(ctx, jobs, client.InNamespace(sharedPool.GetDataPlaneNamespace())); err != nil {
		return nil, nil, a.classifyResumeKubernetesError(fmt.Errorf("list pause archive jobs: %w", err))
	}
	return jobs, remoteClient, nil
}

// Resume has an operation deadline and owner fencing, so transient failures
// stay inside the current async invocation. Delete keeps the executor's
// existing retry-until-success behavior by returning the original error.
func (a *drainResumeArchiveJobsAction) keepResumeRetryInsideInvocation(err error) error {
	if a.guard == nil || err == nil || actions.IsTerminalError(err) {
		return err
	}
	return nil
}

func (a *drainResumeArchiveJobsAction) classifyResumeKubernetesError(err error) error {
	if a.guard == nil {
		return err
	}
	return classifyKubernetesError(err)
}