package service

import (
	"bytes"
	"context"
	"encoding/base64"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"strconv"
	"strings"
	"time"

	"github.com/aws/aws-sdk-go-v2/aws"
	awsconfig "github.com/aws/aws-sdk-go-v2/config"
	"github.com/aws/aws-sdk-go-v2/service/secretsmanager"
	pingcaperr "github.com/pingcap/errors"
	"github.com/pingcap/log"
	"github.com/tidbcloud/pkgv2/auth/auth0"
	"go.uber.org/zap"
	"gorm.io/gorm"
	"gorm.io/gorm/clause"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/client-go/dynamic"
	"k8s.io/client-go/tools/clientcmd"

	apierror "github.com/pingcap-inc/clinic/pkg/api/error"
	"github.com/pingcap-inc/clinic/pkg/db"
	"github.com/pingcap-inc/clinic/pkg/db/model"
	"github.com/pingcap-inc/clinic/pkg/ngm/config"
	"github.com/pingcap-inc/clinic/pkg/ngm/performance_insight/compute_engine"
	"github.com/pingcap-inc/clinic/pkg/pagination"
)

const (
	DefaultOffsetHours = 3
	// Dbaas tenant and dbaas cluster id.
	KeyIDFormatter = "insight/tenant-%d/cluster-%d/mysql-account"
)

var (
	ErrCheckCluster = apierror.ErrInternal.New("internal error: failed to check performance insight cluster")
	ErrGetTask      = apierror.ErrInternal.New("internal error: failed to get performance insight task")
	ErrTaskNotFound = apierror.ErrNotFound.New("internal error: performance insight task not found")
)

var errRunningTaskExisted = errors.New("there is already a running task")

type TaskService interface {
	// For each cluster, only one running task can be existed at the same time.
	// If there is a task not in "succeeded" or "failed" status, return error.ErrPreConditionFailed.
	CreateTask(ctx context.Context, dbaasOrgID, userID, dbaasClusterID uint64) (taskID uint, err error)
	ListTasks(ctx context.Context, dbaasOrgID, dbaasClusterID uint64, statusQ, typesQ []string, updateTimeOrder string, limit, page uint) (tasks []db.PerformanceInsightTask, total int64, err error)
	GetTask(ctx context.Context, dbaasOrgID, dbaasClusterID uint64, taskID uint) (task *db.PerformanceInsightTask, err error)
	GetLatestTask(ctx context.Context, dbaasOrgID, dbaasClusterID uint64) (task *db.PerformanceInsightTask, err error)
	GetLatestFinishTask(ctx context.Context, dbaasOrgID, dbaasClusterID uint64) (task *db.PerformanceInsightTask, err error)
	GetLatestTaskStatus(ctx context.Context, dbaasOrgID, dbaasClusterID uint64) (status string, err error)
	GetTaskResultCount(ctx context.Context, taskID uint) (count int64, err error)
	CancelTask(ctx context.Context, dbaasOrgID, dbaasClusterID uint64, taskID uint) (err error)
	Activate(ctx context.Context, dbaasOrgID, userID, dbaasProjectID, dbaasClusterID uint64, account, password string) error
	Deactivate(ctx context.Context, dbaasOrgID, userID, dbaasClusterID uint64) error
	ClusterActivated(ctx context.Context, dbaasOrgID, userID, dbaasClusterID uint64) (bool, error)
	GetClinicPublicAddresses(ctx context.Context) (addresses []string, err error)
}

func NewTaskService(cfg *config.NgmApiServerConfig, service SecretService, ceClient *compute_engine.ComputerEngineClient) TaskService {
	return &taskService{
		serverConfig:  cfg,
		secretService: service,
		ceClient:      ceClient,
	}
}

type taskService struct {
	secretService SecretService
	serverConfig  *config.NgmApiServerConfig
	ceClient      *compute_engine.ComputerEngineClient
}

func (t *taskService) CreateTask(ctx context.Context, dbaasOrgID, userID, dbaasClusterID uint64) (taskID uint, err error) {
	// Check if the cluster exists.
	piClusterID, exists, err := t.checkPerformanceInsightClusterExists(ctx, dbaasOrgID, dbaasClusterID)
	if err != nil {
		log.Error("failed to check performance insight cluster",
			zap.Error(err),
			zap.Uint64("userID", userID),
			zap.Uint64("dbaasOrgID", dbaasOrgID),
			zap.Uint64("dbaasClusterID", dbaasClusterID))
		return 0, ErrCheckCluster
	}
	if !exists {
		return 0, apierror.ErrBadRequest.New("performance insight cluster does not exists, please activate it first")
	}

	// Ensure there is no running task then create a new one.
	var piTask db.PerformanceInsightTask
	err = db.TransactionWithContext(ctx, func(tx *gorm.DB) error {
		// Require the row lock, on the performance cluster record.
		q := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&db.PerformanceInsightCluster{}, piClusterID)
		if err := q.Error; err != nil {
			return err
		}

		// Get running task.
		queryErr := db.
			NewPerformanceInsightTaskQuerySet(tx).
			PerformanceInsightClusterIDEq(piClusterID).
			StatusIn(TaskStatusCreated, TaskStatusRunning).
			One(&piTask)

		if queryErr != nil && !errors.Is(queryErr, gorm.ErrRecordNotFound) {
			// Query error.
			return queryErr
		} else if queryErr == nil && piTask.ID != 0 {
			// Running task exists.
			return errRunningTaskExisted
		}

		// Pass the check, create task.
		piTask = db.PerformanceInsightTask{
			PerformanceInsightClusterID: piClusterID,
			Type:                        TaskTypeIndexInsight, // TODO: hardcode here, need add param
			Status:                      TaskStatusCreated,
			CreatorID:                   uint(userID),
			ExecutionCount:              0,
			OffsetHours:                 DefaultOffsetHours,
		}
		if err := tx.Create(&piTask).Error; err != nil {
			return err
		}

		return nil
	})

	if errors.Is(err, errRunningTaskExisted) {
		return 0, apierror.ErrPreConditionFailed.New("there is already a running task")
	} else if err != nil {
		log.Error("failed to create the performance insight task",
			zap.Error(err),
			zap.Uint64("userID", userID),
			zap.Uint64("dbaasOrgID", dbaasOrgID),
			zap.Uint64("dbaasClusterID", dbaasClusterID))
		return 0, apierror.ErrInternal.New("internal error: failed to create the performance insight task")
	}

	return piTask.ID, nil
}

func (t *taskService) ListTasks(ctx context.Context, dbaasOrgID, dbaasClusterID uint64, statusQ, typesQ []string, updateTimeOrder string, limit, page uint) (tasks []db.PerformanceInsightTask, total int64, err error) {
	// Check if the cluster exists.
	piClusterID, exists, err := t.checkPerformanceInsightClusterExists(ctx, dbaasOrgID, dbaasClusterID)
	if err != nil {
		log.Error("failed to check performance insight cluster",
			zap.Error(err),
			zap.Uint64("dbaasOrgID", dbaasOrgID),
			zap.Uint64("dbaasClusterID", dbaasClusterID))
		return nil, 0, ErrCheckCluster
	}
	if !exists {
		return nil, 0, nil
	}
	err = db.TransactionWithContext(ctx, func(tx *gorm.DB) error {
		// Build sql query.
		q := tx.Model(&db.PerformanceInsightTask{}).
			Preload("PerformanceInsightCluster").
			Preload("PerformanceInsightCluster.Cluster").
			Preload("PerformanceInsightCluster.Cluster.Organization").
			Where("performance_insight_cluster_id = ?", piClusterID)
		if len(statusQ) > 0 {
			q = q.Where("performance_insight_tasks.status IN (?)", statusQ)
		}
		if len(typesQ) > 0 {
			q = q.Where("performance_insight_tasks.type IN (?)", typesQ)
		}

		// Count # of tasks that match the query.
		if err = q.Count(&total).Error; err != nil {
			return pingcaperr.Annotate(err, "count tasks error")
		}

		// Pagination.
		offset := pagination.GetStartIndex(int(page), int(limit))
		q = q.Limit(int(limit)).Offset(offset).Order("performance_insight_tasks.updated_at " + updateTimeOrder)

		// Do query.
		return q.Find(&tasks).Error
	})
	if err != nil {
		log.Error("failed to list performance insight tasks",
			zap.Error(err),
			zap.Uint64("dbaasOrgID", dbaasOrgID),
			zap.Uint64("dbaasClusterID", dbaasClusterID))
		return nil, 0, apierror.ErrInternal.New("internal error: failed to list performance insight tasks")
	}

	return tasks, total, nil
}

func (t *taskService) GetTask(ctx context.Context, dbaasOrgID, dbaasClusterID uint64, taskID uint) (task *db.PerformanceInsightTask, err error) {
	task = &db.PerformanceInsightTask{}
	err = db.TransactionWithContext(ctx, func(tx *gorm.DB) error {
		q := tx.Model(&db.PerformanceInsightTask{}).
			Preload("PerformanceInsightCluster").
			Preload("PerformanceInsightCluster.Cluster").
			Preload("PerformanceInsightCluster.Cluster.Organization").
			Take(task, taskID)
		return q.Error
	})

	if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
		log.Error("failed to get performance insight task",
			zap.Error(err),
			zap.Uint64("dbaasOrgID", dbaasOrgID),
			zap.Uint64("dbaasClusterID", dbaasClusterID),
			zap.Uint("taskID", taskID))
		return nil, ErrGetTask
	} else if errors.Is(err, gorm.ErrRecordNotFound) {
		return nil, ErrTaskNotFound
	}
	return task, nil
}

func (t *taskService) GetLatestTask(ctx context.Context, dbaasOrgID, dbaasClusterID uint64) (task *db.PerformanceInsightTask, err error) {
	// Check if the cluster exists.
	piClusterID, exists, err := t.checkPerformanceInsightClusterExists(ctx, dbaasOrgID, dbaasClusterID)
	if err != nil {
		log.Error("failed to check performance insight cluster",
			zap.Error(err),
			zap.Uint64("dbaasOrgID", dbaasOrgID),
			zap.Uint64("dbaasClusterID", dbaasClusterID))
		return nil, ErrCheckCluster
	}
	if !exists {
		return nil, nil
	}
	task = &db.PerformanceInsightTask{}
	err = db.TransactionWithContext(ctx, func(tx *gorm.DB) error {
		q := tx.Model(&db.PerformanceInsightTask{}).
			Preload("PerformanceInsightCluster").
			Preload("PerformanceInsightCluster.Cluster").
			Preload("PerformanceInsightCluster.Cluster.Organization").
			Where("performance_insight_cluster_id = ?", piClusterID).
			Last(task)
		return q.Error
	})

	if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
		log.Error("failed to get performance insight task",
			zap.Error(err),
			zap.Uint64("dbaasOrgID", dbaasOrgID),
			zap.Uint64("dbaasClusterID", dbaasClusterID))
		return nil, ErrGetTask
	} else if errors.Is(err, gorm.ErrRecordNotFound) {
		return nil, ErrTaskNotFound
	}
	return task, nil
}

func (t *taskService) GetLatestFinishTask(ctx context.Context, dbaasOrgID, dbaasClusterID uint64) (task *db.PerformanceInsightTask, err error) {
	// Check if the cluster exists.
	piClusterID, exists, err := t.checkPerformanceInsightClusterExists(ctx, dbaasOrgID, dbaasClusterID)
	if err != nil {
		log.Error("failed to check performance insight cluster",
			zap.Error(err),
			zap.Uint64("dbaasOrgID", dbaasOrgID),
			zap.Uint64("dbaasClusterID", dbaasClusterID))
		return nil, ErrCheckCluster
	}
	if !exists {
		return nil, nil
	}
	task = &db.PerformanceInsightTask{}
	err = db.TransactionWithContext(ctx, func(tx *gorm.DB) error {
		q := tx.Model(&db.PerformanceInsightTask{}).
			Preload("PerformanceInsightCluster").
			Preload("PerformanceInsightCluster.Cluster").
			Preload("PerformanceInsightCluster.Cluster.Organization").
			Where("performance_insight_cluster_id = ? and status != ? and status != ?",
				piClusterID, TaskStatusCreated, TaskStatusRunning).
			Last(task)
		return q.Error
	})

	if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
		log.Error("failed to get performance insight task",
			zap.Error(err),
			zap.Uint64("dbaasOrgID", dbaasOrgID),
			zap.Uint64("dbaasClusterID", dbaasClusterID))
		return nil, ErrGetTask
	} else if errors.Is(err, gorm.ErrRecordNotFound) {
		return nil, ErrTaskNotFound
	}
	return task, nil
}

func (t *taskService) GetLatestTaskStatus(ctx context.Context, dbaasOrgID, dbaasClusterID uint64) (status string, err error) {
	// Check if the cluster exists.
	piClusterID, exists, err := t.checkPerformanceInsightClusterExists(ctx, dbaasOrgID, dbaasClusterID)
	if err != nil {
		log.Error("failed to check performance insight cluster",
			zap.Error(err),
			zap.Uint64("dbaasOrgID", dbaasOrgID),
			zap.Uint64("dbaasClusterID", dbaasClusterID))
		return "", ErrCheckCluster
	}
	if !exists {
		return "", nil
	}
	task := &db.PerformanceInsightTask{}
	err = db.TransactionWithContext(ctx, func(tx *gorm.DB) error {
		q := tx.Model(&db.PerformanceInsightTask{}).
			Preload("PerformanceInsightCluster").
			Preload("PerformanceInsightCluster.Cluster").
			Preload("PerformanceInsightCluster.Cluster.Organization").
			Where("performance_insight_cluster_id = ?", piClusterID).
			Last(task)
		return q.Error
	})

	if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
		log.Error("failed to get performance insight task",
			zap.Error(err),
			zap.Uint64("dbaasOrgID", dbaasOrgID),
			zap.Uint64("dbaasClusterID", dbaasClusterID))
		return "", ErrGetTask
	} else if errors.Is(err, gorm.ErrRecordNotFound) {
		return "", ErrTaskNotFound
	}
	return task.Status, nil
}

func (t *taskService) GetTaskResultCount(ctx context.Context, taskID uint) (count int64, err error) {
	err = db.TransactionWithContext(ctx, func(tx *gorm.DB) error {
		return tx.
			Model(&db.IndexAdvisorResult{}).
			Where("performance_insight_task_id = ?", taskID).
			Count(&count).
			Error
	})
	if err != nil {
		log.Error("failed to query index advisor result count for latest task",
			zap.Error(err),
			zap.Uint("taskID", taskID))
		return 0, apierror.ErrInternal.New("internal error: failed to query index advisor result count for latest task")
	}
	return
}

func (t *taskService) CancelTask(ctx context.Context, dbaasOrgID, dbaasClusterID uint64, taskID uint) (err error) {
	err = db.TransactionWithContext(ctx, func(tx *gorm.DB) error {
		q := tx.Debug().Model(&db.PerformanceInsightTask{}).
			Where("id = ?", taskID).
			Updates(map[string]any{
				"status": TaskStatusCanceled,
			})
		return q.Error
	})
	if err != nil {
		log.Error("failed to cancel performance insight task",
			zap.Error(err),
			zap.Uint64("dbaasOrgID", dbaasOrgID),
			zap.Uint64("dbaasClusterID", dbaasClusterID),
			zap.Uint("taskID", taskID))
		return apierror.ErrInternal.New("internal error: failed to cancel performance insight task")
	}
	return nil
}

func (t *taskService) Activate(ctx context.Context, dbaasOrgID, userID, dbaasProjectID, dbaasClusterID uint64, account, password string) error {
	// Ensure performance insight cluster exists.
	piClusterID, err := t.ensurePerformanceInsightClusterExists(ctx, dbaasOrgID, dbaasClusterID)
	if err != nil {
		log.Error("failed to ensure performance insight cluster",
			zap.Error(err),
			zap.Uint64("dbaasOrgID", dbaasOrgID),
			zap.Uint64("dbaasProjectID", dbaasProjectID),
			zap.Uint64("dbaasClusterID", dbaasClusterID),
			zap.Uint64("userID", userID))
		return apierror.ErrInternal.New("internal error: failed to ensure performance insight cluster")
	}

	// Get cluster proxy endpoint from infra api.
	endpoint, err := getClusterProxyEndpoint(ctx, t.serverConfig.Env, dbaasClusterID, dbaasProjectID)
	if err != nil {
		log.Error("failed to get cluster proxy endpoint",
			zap.Error(err),
			zap.Uint64("dbaasOrgID", dbaasOrgID),
			zap.Uint64("dbaasProjectID", dbaasProjectID),
			zap.Uint64("dbaasClusterID", dbaasClusterID),
			zap.Uint64("userID", userID))
		return apierror.ErrInternal.New("internal error: failed to connect user cluster")
	}
	if endpoint == "" {
		return apierror.ErrInternal.New("internal error: empty cluster proxy endpoint")
	}
	if !strings.HasPrefix(endpoint, "http://") && !strings.HasPrefix(endpoint, "https://") {
		endpoint = "https://" + endpoint
	}

	// Get auth0 token.
	auth0Cfg, err := getClusterProxyAuth0FromSecretManager(ctx, t.serverConfig.Env)
	if err != nil {
		log.Error("failed to get cluster proxy auth0 from secret manager",
			zap.Error(err),
			zap.Uint64("dbaasOrgID", dbaasOrgID),
			zap.Uint64("dbaasProjectID", dbaasProjectID),
			zap.Uint64("dbaasClusterID", dbaasClusterID),
			zap.Uint64("userID", userID))
		return apierror.ErrInternal.New("internal error: failed to connect user cluster")
	}
	token, err := auth0.GetClientToken(auth0Cfg.Domain, auth0Cfg.ClientID, auth0Cfg.ClientSecret, auth0Cfg.Audience)
	if err != nil {
		log.Error("failed to get auth0 client token",
			zap.Error(err),
			zap.Uint64("dbaasOrgID", dbaasOrgID),
			zap.Uint64("dbaasProjectID", dbaasProjectID),
			zap.Uint64("dbaasClusterID", dbaasClusterID),
			zap.Uint64("userID", userID))
		return apierror.ErrInternal.New("internal error: failed to connect user cluster")
	}

	// Validate password and privilege.
	passwordOK, privilegeOK, err := validatePasswordAndPrivilege(endpoint, account, password, token)
	if err != nil {
		log.Error("failed to validate password and privilege",
			zap.Error(err),
			zap.Uint64("dbaasOrgID", dbaasOrgID),
			zap.Uint64("dbaasProjectID", dbaasProjectID),
			zap.Uint64("dbaasClusterID", dbaasClusterID),
			zap.Uint64("userID", userID))
		return apierror.ErrInternal.New("internal error: failed to validate password and privilege")
	}
	if !passwordOK {
		return apierror.ErrBadRequest.New("invalid username or password")
	}
	if !privilegeOK {
		return apierror.ErrBadRequest.New("no sufficient privilege")
	}

	// Save password to aws secret manager.
	keyID := fmt.Sprintf(KeyIDFormatter, dbaasOrgID, dbaasClusterID)
	tags := map[string]string{
		"performance-insight-cluster-id": strconv.FormatUint(uint64(piClusterID), 10),
		"creator-id":                     strconv.FormatUint(userID, 10),
		"mysql-account":                  account,
		"used-by":                        "performance-insight",
	}
	if err := t.secretService.CreateOrUpdateSecret(ctx, keyID, password, tags); err != nil {
		log.Error("failed to save mysql account to aws secret manager",
			zap.Error(err),
			zap.Uint64("dbaasOrgID", dbaasOrgID),
			zap.Uint64("dbaasProjectID", dbaasProjectID),
			zap.Uint64("dbaasClusterID", dbaasClusterID),
			zap.Uint64("userID", userID),
			zap.String("account", account))
		return apierror.ErrBadRequest.New("operations may be too frequent")
	}

	// Update account info in the perform insight cluster table.
	err = db.TransactionWithContext(ctx, func(tx *gorm.DB) error {
		q := db.NewPerformanceInsightClusterQuerySet(tx).
			IDEq(piClusterID).
			GetUpdater().
			SetConnectAccountName(account).
			SetConnectPasswordSecretID(keyID).
			SetConnectHost("").
			SetConnectPort(0).
			SetCreatorID(uint(userID))
		return q.Update()
	})
	if err != nil {
		log.Error("failed to update performance insight cluster",
			zap.Error(err),
			zap.Uint64("dbaasOrgID", dbaasOrgID),
			zap.Uint64("dbaasProjectID", dbaasProjectID),
			zap.Uint64("dbaasClusterID", dbaasClusterID),
			zap.Uint64("userID", userID),
			zap.String("account", account))
		return apierror.ErrInternal.New("internal error: failed to update performance insight cluster")
	}

	return nil
}

func (t *taskService) Deactivate(ctx context.Context, dbaasOrgID, dbaasProjectID, dbaasClusterID uint64) error {
	// Delete password from aws secret manager.
	keyID := fmt.Sprintf(KeyIDFormatter, dbaasOrgID, dbaasClusterID)
	if err := t.secretService.DeleteSecret(ctx, keyID); err != nil {
		log.Error("failed to delete mysql account from aws secret manager",
			zap.Error(err),
			zap.Uint64("dbaasOrgID", dbaasOrgID),
			zap.Uint64("dbaasProjectID", dbaasProjectID),
			zap.Uint64("dbaasClusterID", dbaasClusterID))
		return apierror.ErrInternal.New("internal error: failed to delete mysql account from aws secret manager")
	}

	// Check if the cluster exists.
	piClusterID, exists, err := t.checkPerformanceInsightClusterExists(ctx, dbaasOrgID, dbaasClusterID)
	if err != nil {
		log.Error("failed to check performance insight cluster",
			zap.Error(err),
			zap.Uint64("dbaasOrgID", dbaasOrgID),
			zap.Uint64("dbaasProjectID", dbaasProjectID),
			zap.Uint64("dbaasClusterID", dbaasClusterID))
		return ErrCheckCluster
	}
	if !exists {
		return nil
	}

	// Delete cluster, its tasks and its results.
	err = db.TransactionWithContext(ctx, func(tx *gorm.DB) error {
		// Delete all index insight results associated with the cluster.
		err := tx.Where("dbaas_org_id = ? AND dbaas_cluster_id = ?", dbaasOrgID, dbaasClusterID).
			Delete(&db.IndexAdvisorResult{}).
			Error
		if err != nil {
			return err
		}

		// Use gorm association helper to delete the cluster record with related task records.
		// Refer to https://gorm.io/docs/associations.html#Delete-with-Select.
		q := tx.Select("PerformanceInsightTasks").Delete(&db.PerformanceInsightCluster{}, piClusterID)
		return q.Error
	})
	if err != nil {
		log.Error("failed to delete performance insight cluster",
			zap.Error(err),
			zap.Uint64("dbaasOrgID", dbaasOrgID),
			zap.Uint64("dbaasProjectID", dbaasProjectID),
			zap.Uint64("dbaasClusterID", dbaasClusterID))
		return apierror.ErrInternal.New("internal error: failed to delete performance insight cluster")
	}

	return nil
}

func (t *taskService) ClusterActivated(ctx context.Context, dbaasOrgID, dbaasProjectID, dbaasClusterID uint64) (bool, error) {
	// Get clinic cluster.
	cluster, err := model.GetClusterWithDbaasIDs(strconv.FormatUint(dbaasClusterID, 10), uint(dbaasOrgID))
	if err != nil {
		log.Error("failed to get cluster with dbaas id",
			zap.Error(err),
			zap.Uint64("dbaasOrgID", dbaasOrgID),
			zap.Uint64("dbaasProjectID", dbaasProjectID),
			zap.Uint64("dbaasClusterID", dbaasClusterID))
		return false, apierror.ErrInternal.New("internal error: failed to get cluster with dbaas id")
	}
	var pic db.PerformanceInsightCluster
	err = db.TransactionWithContext(ctx, func(tx *gorm.DB) error {
		// Get plan insight cluster.
		return tx.
			Model(&db.PerformanceInsightCluster{}).
			Preload("Cluster").
			Where("cluster_id = ?", cluster.ID). // A PerformanceInsightCluster belongs to a clinic Cluster.
			Take(&pic).
			Error
	})
	if err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return false, nil
		}
		log.Error("failed to query performance insight cluster",
			zap.Error(err),
			zap.Uint64("dbaasOrgID", dbaasOrgID),
			zap.Uint64("dbaasProjectID", dbaasProjectID),
			zap.Uint64("dbaasClusterID", dbaasClusterID))
		return false, apierror.ErrInternal.New("internal error: failed to query performance insight cluster")
	}
	if pic.ConnectPasswordSecretID == "" {
		return false, nil
	}
	return true, nil
}

func (t *taskService) GetClinicPublicAddresses(_ context.Context) (addresses []string, err error) {
	return t.serverConfig.OutgoingNatAddresses, nil
}

func (t *taskService) checkPerformanceInsightClusterExists(ctx context.Context, dbaasOrgID, dbaasClusterID uint64) (id uint, exists bool, err error) {
	// Get clinic cluster.
	cluster, err := model.GetClusterWithDbaasIDs(strconv.FormatUint(dbaasClusterID, 10), uint(dbaasOrgID))
	if err != nil {
		log.Error("failed to get cluster with dbaas id",
			zap.Error(err),
			zap.Uint64("dbaasOrgID", dbaasOrgID),
			zap.Uint64("dbaasClusterID", dbaasClusterID))
		return 0, false, apierror.ErrInternal.New("failed to get cluster with dbaas id")
	}
	var pic db.PerformanceInsightCluster
	err = db.TransactionWithContext(ctx, func(tx *gorm.DB) error {
		// Get plan insight cluster.
		return tx.
			Model(&db.PerformanceInsightCluster{}).
			Preload("Cluster").
			Where("cluster_id = ?", cluster.ID). // A PerformanceInsightCluster belongs to a clinic Cluster.
			Take(&pic).
			Error
	})
	if err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return 0, false, nil
		}
		log.Error("failed to query performance insight cluster",
			zap.Error(err),
			zap.Uint64("dbaasOrgID", dbaasOrgID),
			zap.Uint64("dbaasClusterID", dbaasClusterID))
		return 0, false, apierror.ErrInternal.New("internal error: failed to query performance insight cluster")
	}
	return pic.ID, true, nil
}

func (t *taskService) ensurePerformanceInsightClusterExists(
	ctx context.Context,
	dbaasOrgID,
	dbaasClusterID uint64,
) (planInsightClusterID uint, err error) {
	// Get clinic cluster.
	clinicCluster, err := model.GetClusterWithDbaasIDs(strconv.FormatUint(dbaasClusterID, 10), uint(dbaasOrgID))
	if err != nil {
		return 0, err
	}

	var piCluster db.PerformanceInsightCluster
	err = db.TransactionWithContext(ctx, func(tx *gorm.DB) error {
		// Require the row lock, on the clinic cluster record.
		cq := tx.
			Clauses(clause.Locking{Strength: "UPDATE"}).
			Take(&db.Cluster{}, clinicCluster.ID)
		if err := cq.Error; err != nil {
			return err
		}

		// Get plan insight cluster.
		q := tx.
			Model(&db.PerformanceInsightCluster{}).
			Preload("Cluster").
			Where("cluster_id = ?", clinicCluster.ID). // A PerformanceInsightCluster belongs to a clinic Cluster.
			Take(&piCluster)

		if err := q.Error; err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
			// Query error.
			return err
		} else if err == nil {
			// Found.
			return nil
		}

		// Not found. Create a new one.
		piCluster.ClusterID = clinicCluster.ID
		if err := tx.Create(&piCluster).Error; err != nil {
			return err
		}

		return nil
	})
	if err != nil {
		return 0, err
	}

	return piCluster.ID, nil
}

func getClusterProxyEndpoint(ctx context.Context, env string, dbaasClusterID, dbaasProjectID uint64) (string, error) {
	// Get infra api kubeconfig from secret manager.
	kubeconfig, err := getInfraKubeConfigFromSecretManager(ctx, env)
	if err != nil {
		return "", fmt.Errorf("get infra kubeconfig from secret manager: %w", err)
	}
	// Get cluster proxy endpoint from infra api.
	restCfg, err := clientcmd.RESTConfigFromKubeConfig([]byte(kubeconfig))
	if err != nil {
		return "", fmt.Errorf("k8s rest config from kubeconfig: %w", err)
	}
	dynClient, err := dynamic.NewForConfig(restCfg)
	if err != nil {
		return "", fmt.Errorf("k8s new for config: %w", err)
	}
	namespace := getInfraNamespace(env, dbaasProjectID)
	clusterName := fmt.Sprintf("%d", dbaasClusterID)
	cluster, err := dynClient.Resource(schema.GroupVersionResource{
		Group:    "cloud.pingcap.com",
		Version:  "v1",
		Resource: "clusters",
	}).Namespace(namespace).Get(ctx, clusterName, metav1.GetOptions{})
	if err != nil {
		return "", fmt.Errorf("get cluster resource from infra api: %w", err)
	}
	endpoint, ok, err := unstructured.NestedString(cluster.Object, "status", "proxy", "endpoint")
	if err != nil {
		return "", fmt.Errorf("get field from unstructured object: %w", err)
	}
	if !ok {
		return "", errors.New("cluster proxy endpoint not found")
	}
	return endpoint, nil
}

type auth0Config struct {
	Domain       string `json:"auth0_domain"`
	ClientID     string `json:"auth0_client_id"`
	ClientSecret string `json:"auth0_client_secret"`
	Audience     string `json:"auth0_audience"`
}

func getClusterProxyAuth0FromSecretManager(ctx context.Context, env string) (*auth0Config, error) {
	cfg, err := awsconfig.LoadDefaultConfig(ctx)
	if err != nil {
		return nil, err
	}
	cfg.Region = "us-west-2" // HARDCODE
	cli := secretsmanager.NewFromConfig(cfg)
	out, err := cli.GetSecretValue(ctx, &secretsmanager.GetSecretValueInput{
		SecretId: aws.String("ngm-cluster-proxy-auth0-config-" + strings.ToLower(env)),
	})
	if err != nil {
		return nil, err
	}
	var auth0Cfg auth0Config
	if err := json.Unmarshal([]byte(aws.ToString(out.SecretString)), &auth0Cfg); err != nil {
		return nil, err
	}
	return &auth0Cfg, nil
}

func getInfraKubeConfigFromSecretManager(ctx context.Context, env string) (string, error) {
	cfg, err := awsconfig.LoadDefaultConfig(ctx)
	if err != nil {
		return "", err
	}
	cfg.Region = "us-west-2" // HARDCODE
	cli := secretsmanager.NewFromConfig(cfg)
	out, err := cli.GetSecretValue(ctx, &secretsmanager.GetSecretValueInput{
		SecretId: aws.String("ngm-infra-kubeconfig-" + strings.ToLower(env)),
	})
	if err != nil {
		return "", err
	}
	return aws.ToString(out.SecretString), nil
}

func getInfraNamespace(env string, dbaasProjectID uint64) string {
	switch env {
	case "dev":
		return fmt.Sprintf("nightly-ms-%d", dbaasProjectID)
	case "staging":
		return fmt.Sprintf("staging-%d", dbaasProjectID)
	case "prod":
		return fmt.Sprintf("prod-%d", dbaasProjectID)
	default:
		return ""
	}
}

func validatePasswordAndPrivilege(clusterProxyEndpoint, username, password, token string) (bool, bool, error) {
	validateSQL := `SELECT table_id FROM mysql.stats_meta LIMIT 1`
	status, resp, err := requestClusterProxy(clusterProxyEndpoint, username, password, validateSQL, token)
	if err != nil {
		return false, false, err
	}
	if status == http.StatusOK {
		return true, true, nil
	}
	respStr := string(resp)
	if strings.Contains(respStr, "Error 1045: Access denied for user") {
		return false, false, nil
	}
	if strings.Contains(respStr, "Error 1044:") ||
		strings.Contains(respStr, "Error 1142:") ||
		strings.Contains(respStr, "Error 1143:") ||
		strings.Contains(respStr, "Error 1227:") ||
		strings.Contains(respStr, "Error 8121:") {
		return true, false, nil
	}
	log.Warn("failed to query cluster proxy",
		zap.Int("status", status),
		zap.String("body", respStr))
	return false, false, fmt.Errorf("failed to query cluster proxy, respond code: %d, response: %s", status, respStr)
}

func requestClusterProxy(endpoint, username, password, sql, token string) (int, []byte, error) {
	type operator struct {
		Username   string `json:"username"`
		AuthMethod string `json:"authMethod"` // token, password, privilegeDB
		Credential string `json:"credential"` // TOKEN, PASSWORD
	}

	type request struct {
		Query    string   `json:"query"`
		Operator operator `json:"operator"`
	}

	client := &http.Client{
		Timeout: 10 * time.Second,
	}
	b, err := json.Marshal(request{
		Query: sql,
		Operator: operator{
			Username:   username,
			AuthMethod: "password",
			Credential: base64.URLEncoding.EncodeToString([]byte(password)),
		},
	})
	if err != nil {
		return 0, nil, err
	}
	req, err := http.NewRequest(http.MethodGet, strings.TrimSuffix(endpoint, "/")+"/v1beta1/query", bytes.NewReader(b))
	if err != nil {
		return 0, nil, err
	}
	if token != "" {
		req.Header.Add("Authorization", fmt.Sprintf("bearer %s", token))
	}
	resp, err := client.Do(req)
	if err != nil {
		return 0, nil, err
	}
	defer resp.Body.Close()
	b, err = io.ReadAll(resp.Body)
	if err != nil {
		return 0, nil, err
	}
	return resp.StatusCode, b, nil
}