package http

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"strings"

	"github.com/gin-gonic/gin"
	"github.com/pingcap/log"
	"github.com/tidbcloud/dataflow-service/pkg/agent/tidbcloud"
	"go.uber.org/zap"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)

type DMConfigSecret struct {
	ClusterID  string `json:"cluster_id"`
	TenantID   string `json:"tenant_id"`
	DBUser     string `json:"db_user"`
	DBPassword string `json:"db_password"`
	DBHost     string `json:"db_host"`
	DBPort     int    `json:"db_port"`
	DBName     string `json:"db_name"`
	TLS        bool   `json:"tls"`
}

type ResetDMErrorReq struct {
	JobID     int64  `json:"job_id"`
	ClusterID string `json:"cluster_id"`
}

func GetDMConfigSecret(ctx context.Context) (map[string]DMConfigSecret, error) {
	cli, err := tidbcloud.GetKubeClient(ctx)
	if err != nil {
		return nil, err
	}

	secret, err := cli.CoreV1().Secrets("tidb-admin").Get(ctx, "dm-config", metav1.GetOptions{})
	if err != nil {
		log.L().Error("failed to get dm config secret", zap.Error(err))
		return nil, err
	}

	secretData, ok := secret.Data["whitelist.json"]
	if !ok {
		log.L().Error("whitelist.json not found")
		return nil, fmt.Errorf("whitelist.json not found")
	}

	var dmSecretMap map[string]DMConfigSecret
	err = json.Unmarshal(secretData, &dmSecretMap)
	if err != nil {
		log.L().Error("failed to unmarshal dm config secret", zap.Error(err))
		return nil, err
	}

	return dmSecretMap, nil
}

// ResetDMError resets the error message of the DM job.
// DO NOT use this API in other cases, this API is only used for DM job recovery, internal use only
// And will be removed after 2025-03-01! After new DM implmentation released.
func ResetDMError(c *gin.Context) {
	var req ResetDMErrorReq
	if err := c.ShouldBindJSON(&req); err != nil {
		log.L().Error("failed to bind request", zap.Error(err))
		c.JSON(400, gin.H{"error": "invalid request"})
		return
	}

	ctx := c.Request.Context()
	// check if the cluster is in the whitelist
	dmSecretMap, err := GetDMConfigSecret(ctx)
	if err != nil {
		log.L().Error("failed to get dm secret", zap.Error(err))
		c.JSON(500, gin.H{"error": "failed to get dm secret"})
		return
	}
	dmSecret, ok := dmSecretMap[req.ClusterID]
	if !ok {
		log.L().Error("cluster_id is not in the whitelist", zap.String("cluster_id", req.ClusterID))
		c.JSON(400, gin.H{"error": "cluster_id is not in the whitelist"})
		return
	}

	db, err := sql.Open("mysql", fmt.Sprintf("%s:%s@tcp(%s:%d)/%s?parseTime=true&tls=%t", dmSecret.DBUser, dmSecret.DBPassword, dmSecret.DBHost, dmSecret.DBPort, dmSecret.DBName, dmSecret.TLS))
	if err != nil {
		log.L().Error("failed to open db connection", zap.Error(err))
		c.JSON(500, gin.H{"error": "failed to connect to database"})
		return
	}
	defer db.Close()

	jobIDs := []string{
		fmt.Sprintf("%d", req.JobID),
		fmt.Sprintf("%d-sync", req.JobID),
		fmt.Sprintf("%d-dump", req.JobID),
	}
	sql := "update master_meta set error_message='' where id in (?,?,?)"
	result, err := db.Exec(sql, jobIDs[0], jobIDs[1], jobIDs[2])
	if err != nil {
		log.L().Error("failed to reset error message", zap.Error(err), zap.Int64("job-id", req.JobID))
		c.JSON(500, gin.H{"error": "failed to reset error message"})
		return
	}

	rowsAffected, _ := result.RowsAffected()
	c.JSON(200, gin.H{
		"message":       "success",
		"rows_affected": rowsAffected,
	})

}

type GetDMConfigResp struct {
	ClusterID        string   `json:"cluster_id"`
	Namespace        string   `json:"namespace"`
	ExecutorPodNames []string `json:"executor_pod_names"`
}

// GetDMConfig gets the DM config of the cluster.
// This API is only used for DM job recovery, internal use only
// And will be removed after 2025-03-01! After new DM implmentation released.
func GetDMConfig(c *gin.Context) {
	clusterID := c.Query("cluster_id")
	if clusterID == "" {
		c.JSON(400, gin.H{"error": "cluster_id is required"})
		return
	}

	// check if the cluster is in the whitelist
	ctx := c.Request.Context()
	dmSecretMap, err := GetDMConfigSecret(ctx)
	if err != nil {
		log.L().Error("failed to get dm secret", zap.Error(err))
		c.JSON(500, gin.H{"error": "failed to get dm secret"})
		return
	}
	_, ok := dmSecretMap[clusterID]
	if !ok {
		log.L().Error("cluster_id is not in the whitelist", zap.String("cluster_id", clusterID))
		c.JSON(400, gin.H{"error": "cluster_id is not in the whitelist"})
		return
	}

	cli, err := tidbcloud.GetKubeClient(ctx)
	if err != nil {
		log.L().Error("failed to get kube client", zap.Error(err))
		c.JSON(500, gin.H{"error": "failed to get kube client"})
		return
	}

	// filter namespace from k8s, namespace should be tiflow-prod-{project_id}-{provider_type}-{region}-{hash}
	var namespace string
	namespaces, err := cli.CoreV1().Namespaces().List(ctx, metav1.ListOptions{})
	if err != nil {
		log.L().Error("failed to get namespaces", zap.Error(err))
		c.JSON(500, gin.H{"error": "failed to get namespaces"})
		return
	}
	for _, ns := range namespaces.Items {
		if strings.HasPrefix(ns.Name, "tiflow-") {
			namespace = ns.Name
			break
		}
	}

	if namespace == "" {
		log.L().Error("no namespace found")
		c.JSON(500, gin.H{"error": "failed to get namespace"})
		return
	}

	// get pods from namespace
	pods, err := cli.CoreV1().Pods(namespace).List(ctx, metav1.ListOptions{})
	if err != nil {
		log.L().Error("failed to get pods", zap.Error(err))
		c.JSON(500, gin.H{"error": "failed to get pods"})
		return
	}

	podNames := []string{}
	for _, pod := range pods.Items {
		if strings.HasPrefix(pod.Name, "dm-job-") {
			podNames = append(podNames, pod.Name)
		}
	}

	resp := GetDMConfigResp{
		ClusterID:        clusterID,
		Namespace:        namespace,
		ExecutorPodNames: podNames,
	}

	c.JSON(200, resp)
}