package bgtask

import (
	"cmp"
	"context"
	"errors"
	"fmt"
	"strconv"
	"strings"

	"github.com/tidbcloud/aws-shared-provider/pkg/tidbworker/job"
	"github.com/tidbcloud/aws-shared-provider/pkg/tidbworker/schema/bgtask"
	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-"
	execIDParts          = 2
	workerNameIDPartsMin = 2

	bgtaskComponentNode = "general"
)

// getWorkerIndexFromExecID extract worker index from execID.
func getWorkerIndexFromExecID(execID string) (int, error) {
	parts := strings.Split(execID, ":")
	if len(parts) != execIDParts {
		return 0, fmt.Errorf("invalid execID %s", execID)
	}
	index, err := strconv.Atoi(parts[1])
	if err != nil {
		return 0, err
	}
	return index, nil
}

// getWorkerIndexFromJobName extract worker index from jobName.
func getWorkerIndexFromJobName(jobName string) (int, error) {
	parts := strings.Split(jobName, "-")
	if len(parts) < workerNameIDPartsMin {
		return 0, fmt.Errorf("invalid job name %s", jobName)
	}
	index, err := strconv.Atoi(parts[len(parts)-1])
	if err != nil {
		return 0, err
	}
	return index, nil
}

// mkJobName generates runner name in the format:
// tidb-worker-{type}-{keyspaceID}-{globalTaskID}-{workerIndex}
func mkJobName(taskType job.Type, keyspaceID uint32, globalTaskID int64, workerIndex int) string {
	return fmt.Sprintf("%s-%d-%d-%d", jobNamePrefix+taskType, keyspaceID, globalTaskID, workerIndex)
}

func (h *handler) getComponentNode() string {
	return cmp.Or(h.componentNode, bgtaskComponentNode)
}

func (h *handler) activateJob(
	ctx context.Context,
	keyspaceName string,
	keyspaceID uint32,
	tidbPool string,
	taskType job.Type,
	globalTaskID int64,
	execID string) error {
	db := h.db.WithContext(ctx)

	// Get worker index from execID.
	workerIndex, err := getWorkerIndexFromExecID(execID)
	if err != nil {
		h.logger.Error(err, "failed to get worker index", "execID", execID)
		return err
	}

	// Build bg worker job spec.
	jobName := mkJobName(taskType, keyspaceID, globalTaskID, workerIndex)
	jobSpec, err := h.jobBuilder.Build(ctx, job.Options{
		TiDBPool:         tidbPool,
		WorkerImage:      h.adminConfig.GetWorkerImage(),
		KeyspaceName:     keyspaceName,
		JobType:          taskType,
		JobName:          jobName,
		ExecID:           execID,
		WorkerCertSecret: h.workerCertSecret,
		ComponentNode:    h.getComponentNode(),
	})
	if err != nil {
		h.logger.Error(err, "failed to build job spec", "keyspace-name", keyspaceName, "job", jobName)
		return err
	}

	// Create bg job.
	h.logger.Info("create bg job", "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
	}

	// Sync job info to task table.
	result := db.Model(&bgtask.Task{}).
		Where("keyspace_name = ?", keyspaceName).
		Where("exec_id = ?", execID).
		Where("state = ?", bgtask.StatePending).
		Updates(map[string]interface{}{
			"state":  bgtask.StateRunning,
			"runner": jobName,
		})

	if result.RowsAffected > 0 {
		h.logger.Info("sync job info to database", "job", jobName)
		return nil
	}

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

// mkJobNameV2 generates runner name in the format:
// tidb-worker-shared-{keyspaceID}-{workerIndex}
func mkJobNameV2(keyspaceID uint32, workerIndex int) string {
	return fmt.Sprintf("%s-%d-%d", "tidb-worker-shared", keyspaceID, workerIndex)
}

func (h *handler) activateJobV2(
	ctx context.Context,
	keyspaceName string,
	keyspaceID uint32,
	tidbPool string,
	execID string) error {
	// Get worker index from execID.
	workerIndex, err := getWorkerIndexFromExecID(execID)
	if err != nil {
		h.logger.Error(err, "failed to get worker index", "execID", execID)
		return err
	}

	// Build bg worker job spec.
	jobName := mkJobNameV2(keyspaceID, workerIndex)
	jobSpec, err := h.jobBuilder.Build(ctx, job.Options{
		TiDBPool:         tidbPool,
		WorkerImage:      h.adminConfig.GetWorkerImage(),
		KeyspaceName:     keyspaceName,
		JobType:          job.TypeShared,
		JobName:          jobName,
		ExecID:           execID,
		WorkerCertSecret: h.workerCertSecret,
		ComponentNode:    h.getComponentNode(),
	})
	if err != nil {
		h.logger.Error(err, "failed to build job spec", "keyspace-name", keyspaceName, "job", jobName)
		return err
	}

	// Create bg job.
	h.logger.Info("create bg job", "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)
		} else {
			return err
		}
	}
	return nil
}

func (h *handler) markTaskRunning(ctx context.Context, keyspaceName string, execID string) error {
	db := h.db.WithContext(ctx)
	// Sync job info to task table.
	result := db.Model(&bgtask.Task{}).
		Where("keyspace_name = ?", keyspaceName).
		Where("exec_id = ?", execID).
		Where("state = ?", bgtask.StatePending).
		Updates(map[string]interface{}{
			"state": bgtask.StateRunning,
		})

	if result.RowsAffected > 0 {
		h.logger.Info("sync job info to database", "keyspaceName", keyspaceName)
		return nil
	}
	return errors.New("no affect rows")
}