package autoanalyze

import (
	"context"
	"fmt"
	"time"

	"github.com/tidbcloud/aws-shared-provider/pkg/tidbworker/job"
	"github.com/tidbcloud/aws-shared-provider/pkg/tidbworker/schema/autoanalyze"
	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-auto-analyze"
)

// mkJobName generates runner name in the format:
// tidb-worker-auto-analyze-{keyspaceID}-{registeredAt}
func mkJobName(keyspaceID uint32, registeredAt time.Time) string {
	return fmt.Sprintf("%s-%d-%d", jobNamePrefix, keyspaceID, registeredAt.Unix())
}

func (h *handler) activateJob(
	ctx context.Context,
	keyspaceName string,
	keyspaceID uint32,
	tidbPool string,
	registeredAt time.Time) error {
	db := h.db.WithContext(ctx)

	// Build auto-analyze worker job spec.
	jobName := mkJobName(keyspaceID, registeredAt)
	jobSpec, err := h.jobBuilder.Build(ctx, job.Options{
		TiDBPool:         tidbPool,
		WorkerImage:      h.adminConfig.GetWorkerImage(),
		KeyspaceName:     keyspaceName,
		JobType:          job.TypeAutoAnalyze,
		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 auto-analyze job.
	h.logger.Info("create auto-analyze 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
	}

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

	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 do not sync to the database, we should delete it.
	return h.clusterCli.Delete(ctx, jobSpec, client.PropagationPolicy(metav1.DeletePropagationBackground))
}