package gc

import (
	"context"
	"fmt"
	"strings"

	"github.com/tidbcloud/aws-shared-provider/pkg/tidbworker/job"
	schedulerutils "github.com/tidbcloud/aws-shared-provider/pkg/tidbworker/resourcemanager/utils"
	"github.com/tidbcloud/aws-shared-provider/pkg/tidbworker/schema/gc"
	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 (
	jobNamePrefix = "tidb-worker-gc"
	rmTaskPrefix  = "rm-tidb-worker-gc"
)

// mkJobName generates runner name in the format:
// tidb-worker-gc-{keyspaceID}-{safePoint}
func mkJobName(keyspaceID uint32, safePoint uint64) string {
	return fmt.Sprintf("%s-%d-%d", jobNamePrefix, keyspaceID, safePoint)
}

// mkJobName generates runner name in the format:
// rm-tidb-worker-gc-{keyspaceID}-{safePoint}
func mkRMTaskID(keyspaceID uint32, safePoint uint64) string {
	return fmt.Sprintf("%s-%d-%d", rmTaskPrefix, keyspaceID, safePoint)
}

func isRMTask(jobName string) bool {
	return strings.HasPrefix(jobName, rmTaskPrefix)
}

func (h *handler) activateJob(
	ctx context.Context,
	keyspaceName string,
	keyspaceID uint32,
	tidbPool string,
	safePoint uint64) error {
	if h.taskRM != nil {
		return h.dispatchRMTask(ctx, keyspaceName, keyspaceID, safePoint)
	}
	// Build gc worker job spec.
	jobName := mkJobName(keyspaceID, safePoint)
	jobSpec, err := h.jobBuilder.Build(ctx, job.Options{
		TiDBPool:         tidbPool,
		WorkerImage:      h.adminConfig.GetWorkerImage(),
		KeyspaceName:     keyspaceName,
		JobType:          job.TypeGC,
		JobName:          jobName,
		SafeToEvict:      true,
		WorkerCertSecret: h.workerCertSecret,
	})
	if err != nil {
		h.logger.Error(err, "failed to build job spec", "keyspace-name", keyspaceName, "job", jobName)
		return err
	}

	// Create gc job.
	h.logger.Info("create gc job", "keyspace-name", keyspaceName, "job", jobName)
	err = h.clusterCli.Create(ctx, jobSpec)
	if err != nil {
		if apierrors.IsAlreadyExists(err) {
			h.logger.Info("job already created", "keyspace-name", keyspaceName, "job", jobName)
			return nil
		}
		return err
	}

	if h.syncJobIntoTable(ctx, keyspaceName, safePoint, jobName) {
		h.logger.Info("sync job info to database", "job", jobName)
		return nil
	}

	h.logger.Info("delete orphan job", "job", jobName)
	// If the created job do not sync to the database, we should delete it.
	return h.clusterCli.Delete(ctx, jobSpec, client.PropagationPolicy(metav1.DeletePropagationBackground))
}

func (h *handler) recycleJob(ctx context.Context, keyspaceName, jobName string) error {
	if h.taskRM != nil && isRMTask(jobName) {
		return h.taskRM.RecycleResource(ctx, keyspaceName, jobName)
	}
	j := &batchv1.Job{
		ObjectMeta: metav1.ObjectMeta{
			Name:      jobName,
			Namespace: h.namespace,
		},
	}
	return h.clusterCli.Delete(ctx, j, client.PropagationPolicy(metav1.DeletePropagationBackground))
}

func (h *handler) syncJobIntoTable(
	ctx context.Context,
	keyspaceName string,
	safePoint uint64,
	jobName string,
) bool {
	db := h.db.WithContext(ctx)
	// Sync job info to task table.
	result := db.Model(&gc.Task{}).
		Where("keyspace_name = ?", keyspaceName).
		Where("deletion_ts < ?", safePoint).
		Where("state = ?", gc.StatePending).
		Updates(map[string]interface{}{
			"state":  gc.StateRunning,
			"runner": jobName,
		})
	return result.RowsAffected > 0
}

func (h *handler) dispatchRMTask(ctx context.Context,
	keyspaceName string,
	keyspaceID uint32,
	safePoint uint64,
) error {
	taskID := mkRMTaskID(keyspaceID, safePoint)
	if err := h.taskRM.AllocateResource(ctx, schedulerutils.MakeGCRequest(keyspaceName, taskID)); err != nil {
		return fmt.Errorf("failed to allocate resource for task %s: %w", taskID, err)
	}
	if h.syncJobIntoTable(ctx, keyspaceName, safePoint, taskID) {
		h.logger.Info("sync job info to database", "job", taskID)
		return nil
	}
	h.logger.Info("recycle orphan job", "job", taskID)
	// If the created job did not sync to the database, we should recycle it from RM.
	return h.taskRM.RecycleResource(ctx, keyspaceName, taskID)
}