package ttl_task

import (
	"context"
	"fmt"
	"time"

	"github.com/pingcap/log"
	"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/ttl_task"
	"go.uber.org/zap"
	"gorm.io/gorm/clause"
	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-ttl"
)

// mkJobName generates runner name in the format:
// {jobNamePrefix}-{keyspaceID}-{lastTTLRunTime}
func mkJobName(keyspaceID uint32, tableID int64, lastTTLRunTime int64) string {
	return fmt.Sprintf("%s-%v-%v-%v", jobNamePrefix, keyspaceID, tableID, lastTTLRunTime)
}

// mkTaskIDByKeyspaceName generates runner name in the format:
// rm-{jobNamePrefix}-{keyspaceName}-{lastTTLRunTime}
func mkTaskIDByKeyspaceName(keyspaceName string, tableID int64, lastTTLRunTime int64) string {
	return fmt.Sprintf("rm-%s-%v-%v-%v", jobNamePrefix, keyspaceName, tableID, lastTTLRunTime)
}

func (h *handler) activateJob(
	ctx context.Context,
	keyspaceName string,
	keyspaceID uint32,
	tidbPool string,
	tableID int64,
	lastTTLTaskRunTime int64,
) error {
	now := time.Now()

	if h.taskRM != nil && h.isTestKeyspace(keyspaceName) {
		log.Info("use resource manager", zap.String("keyspace-name", keyspaceName))
		taskID := mkTaskIDByKeyspaceName(keyspaceName, tableID, lastTTLTaskRunTime)
		req := schedulerutils.MakeTTLRequest(keyspaceName, taskID)
		err := h.taskRM.AllocateResource(ctx, req)
		if err != nil {
			log.Error("failed to build get task resource",
				zap.String("keyspace-name", keyspaceName), zap.Error(err))
			return err
		}
		err = h.updateRunningTTLJob(ctx, keyspaceName, tidbPool, tableID, lastTTLTaskRunTime, taskID, now)
		if err != nil {
			log.Error("failed to update running ttl job", zap.String("keyspace-name", keyspaceName))
			return err
		}
		return nil
	}

	// Build ttl worker job spec.
	jobName := mkJobName(keyspaceID, tableID, lastTTLTaskRunTime)
	jobSpec, err := h.jobBuilder.Build(ctx, job.Options{
		TiDBPool:         tidbPool,
		WorkerImage:      h.adminConfig.GetWorkerImage(),
		KeyspaceName:     keyspaceName,
		JobType:          job.TypeTTL,
		JobName:          jobName,
		SafeToEvict:      true,
		WorkerCertSecret: h.workerCertSecret,
	})
	if err != nil {
		log.Error("failed to build job spec",
			zap.String("keyspace-name", keyspaceName), zap.String("job", jobName), zap.Error(err))
		return err
	}

	// Create ttl job.
	log.Info("creating ttl job", zap.String("keyspace-name", keyspaceName), zap.String("job", jobName))
	err = h.clusterCli.Create(ctx, jobSpec)
	if err != nil {
		if apierrors.IsAlreadyExists(err) {
			log.Info("job already created",
				zap.String("keyspace-name", keyspaceName), zap.String("job", jobName))
			return nil
		}
		return err
	}

	err = h.updateRunningTTLJob(ctx, keyspaceName, tidbPool, tableID, lastTTLTaskRunTime, jobName, now)
	if err != nil {
		return h.clusterCli.Delete(ctx, jobSpec, client.PropagationPolicy(metav1.DeletePropagationBackground))
	}
	return nil
}

func (h *handler) updateRunningTTLJob(ctx context.Context,
	keyspaceName string,
	tidbPool string,
	tableID int64,
	lastTTLTaskRunTime int64,
	jobName string,
	now time.Time,
) error {
	db := h.db.WithContext(ctx)

	ttl_task := &ttl_task.Task{
		KeyspaceName:   keyspaceName,
		TidbPool:       tidbPool,
		TableID:        tableID,
		LastTTLRunTime: lastTTLTaskRunTime,
		Runner:         jobName,
		RunnerCreated:  now.Unix(),
		State:          ttl_task.StateRunning,
		RegisteredAt:   now,
	}
	log.Info("update or insert into ttl_tasks running",
		zap.String("keyspace-name", keyspaceName),
		zap.String("job", jobName), zap.Any("task", ttl_task))
	err := db.Clauses(clause.OnConflict{
		Columns:   []clause.Column{{Name: "keyspace_name"}, {Name: "table_id"}, {Name: "last_ttl_run_time"}},
		DoUpdates: clause.AssignmentColumns([]string{"state", "runner", "runner_created", "registered_at"}),
	}).Create(ttl_task).Error
	if err != nil {
		log.Error("failed to update or insert into ttl_tasks,",
			zap.String("keyspace-name", keyspaceName),
			zap.String("job", jobName),
			zap.Any("task", ttl_task),
			zap.Error(err))
		return err
	}
	return nil
}