package admin

import (
	"context"
	"net/http"
	"os"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/go-logr/logr"
	"github.com/gorilla/mux"
	"github.com/pingcap/log"
	"github.com/pingcap/tidb/pkg/util/logutil"
	"github.com/tidbcloud/aws-shared-provider/pkg/tidbworker/schema/autoanalyze"
	"github.com/tidbcloud/aws-shared-provider/pkg/tidbworker/schema/bgtask"
	"github.com/tidbcloud/aws-shared-provider/pkg/tidbworker/schema/gc"
	"github.com/tidbcloud/aws-shared-provider/pkg/tidbworker/schema/gcv2"
	"github.com/tidbcloud/aws-shared-provider/pkg/tidbworker/schema/ttl_task"
	"github.com/tidbcloud/aws-shared-provider/pkg/types"
	"github.com/tidbcloud/aws-shared-provider/pkg/utils/labels"
	"go.uber.org/zap"
	"gopkg.in/yaml.v2"
	"gorm.io/gorm"
	batchv1 "k8s.io/api/batch/v1"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/utils/ptr"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller"
	"sigs.k8s.io/controller-runtime/pkg/handler"
	ctrl "sigs.k8s.io/controller-runtime/pkg/manager"
	"sigs.k8s.io/controller-runtime/pkg/predicate"
	"sigs.k8s.io/controller-runtime/pkg/reconcile"
	"sigs.k8s.io/controller-runtime/pkg/source"
)

const (
	cleanupJobsInterval = 5 * time.Minute
	tidbWorkerNamespace = "tidb-worker"
	tidbWorkerNameLabel = "tidb-worker"
	workerImageKey      = "worker-image"
)

// WorkerConfig is the admin config for a worker.
type WorkerConfig struct {
	Paused  bool `yaml:"paused"`
	Cleanup bool `yaml:"cleanup"`
}

// ConfigReconciler watches updates of admin-cm.
type ConfigReconciler struct {
	manager ctrl.Manager
	logger  logr.Logger
	db      *gorm.DB

	sync.RWMutex
	workerImage     string
	config          map[string]WorkerConfig
	TestRMkeyspaces map[string]string
}

// NewReconciler creates a new ConfigReconciler.
func NewReconciler(mgr ctrl.Manager, db *gorm.DB) (*ConfigReconciler, error) {
	cw := &ConfigReconciler{
		manager:         mgr,
		logger:          mgr.GetLogger().WithName("configwatch"),
		db:              db,
		config:          make(map[string]WorkerConfig),
		TestRMkeyspaces: make(map[string]string),
	}
	c, err := controller.New("configwatch-controller", mgr, controller.Options{
		Reconciler:         cw,
		NeedLeaderElection: ptr.To(false),
	})
	if err != nil {
		return nil, err
	}

	err = c.Watch(source.Kind(mgr.GetCache(), &corev1.ConfigMap{},
		&handler.TypedEnqueueRequestForObject[*corev1.ConfigMap]{},
		predicate.NewTypedPredicateFuncs(func(cm *corev1.ConfigMap) bool {
			return cm.GetNamespace() == tidbWorkerNamespace && cm.GetName() == "admin-cm"
		}),
	))
	if err != nil {
		return nil, err
	}

	err = mgr.Add(cw)
	if err != nil {
		return nil, err
	}
	return cw, nil
}

// NeedLeaderElection implements manager.LeaderElectionRunnable interface.
func (cw *ConfigReconciler) NeedLeaderElection() bool {
	return false
}

// Reconcile implements reconcile.Reconciler interface.
func (cw *ConfigReconciler) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) {
	var configMap corev1.ConfigMap
	if err := cw.manager.GetClient().Get(ctx, req.NamespacedName, &configMap); err != nil {
		if apierrors.IsNotFound(err) {
			cw.logger.Info("admin configmap deleted")
			cw.Lock()
			defer cw.Unlock()
			cw.config = make(map[string]WorkerConfig)
			return reconcile.Result{}, nil
		}
		return reconcile.Result{}, err
	}
	cw.logger.Info("admin configmap updated", "config", configMap.Data)
	config := make(map[string]WorkerConfig)
	var workerImage string
	for key, value := range configMap.Data {
		if key == workerImageKey {
			workerImage = value
			continue
		}
		var workerConfig WorkerConfig
		if err := yaml.Unmarshal([]byte(value), &workerConfig); err != nil {
			cw.logger.Error(err, "unable to parse config", "key", key)
			continue
		}
		config[key] = workerConfig
	}
	cw.Lock()
	defer cw.Unlock()
	cw.config = config
	cw.workerImage = workerImage
	return reconcile.Result{}, nil
}

// Start implements manager.Runnable interface.
func (cw *ConfigReconciler) Start(ctx context.Context) error {
	for {
		select {
		case <-ctx.Done():
			return nil
		case <-time.After(cleanupJobsInterval):
			var cleanupTypes []string
			cw.RLock()
			for key, config := range cw.config {
				if config.Cleanup {
					cleanupTypes = append(cleanupTypes, key)
				}
			}
			cw.RUnlock()
			for _, t := range cleanupTypes {
				cw.logger.Info("cleanup jobs", "type", t)
				err := cw.manager.GetClient().DeleteAllOf(ctx, &batchv1.Job{},
					client.InNamespace(os.Getenv("NAMESPACE")),
					client.MatchingLabels(map[string]string{types.JobTypeLabelKey: t}),
					client.PropagationPolicy(metav1.DeletePropagationBackground))
				if err != nil {
					cw.logger.Error(err, "failed to cleanup jobs", "type", t)
				}
			}
		}
	}
}

// IsPaused returns true if the worker is paused.
func (cw *ConfigReconciler) IsPaused(name string) bool {
	cw.RLock()
	defer cw.RUnlock()

	config, ok := cw.config[name]
	return ok && config.Paused
}

func (cw *ConfigReconciler) GetWorkerImage() string {
	if cw == nil {
		return ""
	}
	cw.RLock()
	defer cw.RUnlock()

	return cw.workerImage
}

func (cw *ConfigReconciler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	cw.logger.Info("received request", "method", r.Method, "url", r.URL.Path)
	router := mux.NewRouter()
	router.HandleFunc("/scaler/api/v1/admin/pause", cw.pause)
	router.HandleFunc("/scaler/api/v1/admin/resume", cw.resume)
	router.HandleFunc("/scaler/api/v1/admin/clean", cw.clean)
	router.HandleFunc("/scaler/api/v1/admin/loglevel", cw.setLogLevel)
	router.HandleFunc("/scaler/api/v1/admin/testkeyspace", cw.setTestKeyspace)
	router.HandleFunc("/scaler/api/v1/admin/get_keyspace_pod_ct", cw.getKeyspacePodCount)
	router.ServeHTTP(w, r)
}

// targetJobs returns the scoping function for gorm to find the target jobs based on keyspace and pool input.
func targetJobs(keyspace, pool string) func(db *gorm.DB) *gorm.DB {
	return func(db *gorm.DB) *gorm.DB {
		if keyspace != "" {
			db = db.Where("keyspace_name = ?", keyspace)
		}
		if pool != "" {
			db = db.Where("tidb_pool = ?", pool)
		}
		return db.Where("1=1") // Avoid global operation
	}
}

func targetLabels(keyspace, pool string) map[string]string {
	target := map[string]string{
		labels.NameLabelKey:      tidbWorkerNameLabel,
		labels.ComponentLabelKey: "worker",
	}
	if keyspace != "" {
		target[types.KeyspaceNameLabelKey] = keyspace
	}
	if pool != "" {
		target[types.TiDBPoolLabelKey] = pool
	}
	return target
}

func isTableNotExistsError(err error) bool {
	return err != nil && strings.Contains(err.Error(), "1146")
}

// pause worker tasks.
// It first updates the state of all target tasks to paused in the database,
// then deletes all target jobs in the namespace.
func (cw *ConfigReconciler) pause(w http.ResponseWriter, r *http.Request) {
	keyspace := r.URL.Query().Get("keyspace")
	pool := r.URL.Query().Get("pool")
	scope := targetJobs(keyspace, pool)
	err := cw.db.WithContext(r.Context()).Transaction(func(tx *gorm.DB) error {
		err := tx.Model(&gc.Task{}).Scopes(scope).
			Update("state", gc.StatePaused).Error
		if err != nil && !isTableNotExistsError(err) {
			return err
		}
		err = tx.Model(&gcv2.Task{}).Scopes(scope).
			Update("state", gcv2.StatePaused).Error
		if err != nil && !isTableNotExistsError(err) {
			return err
		}
		err = tx.Model(&bgtask.Task{}).Scopes(scope).
			Update("state", bgtask.StatePaused).Error
		if err != nil && !isTableNotExistsError(err) {
			return err
		}
		err = tx.Model(&autoanalyze.Task{}).Scopes(scope).
			Update("state", autoanalyze.StatePaused).Error
		if err != nil && !isTableNotExistsError(err) {
			return err
		}
		err = tx.Model(&ttl_task.Task{}).Scopes(scope).
			Update("state", ttl_task.StatePaused).Error
		if err != nil && !isTableNotExistsError(err) {
			return err
		}
		// Special treatment to ttl_table_info, which does not contain pool info.
		query := tx.Model(&ttl_task.TTLTableInfo{}).Where("paused = ?", false)
		if keyspace != "" {
			query = query.Where("keyspace_name = ?", keyspace)
		}
		err = query.Update("paused", true).Error
		if err != nil && !isTableNotExistsError(err) {
			return err
		}
		return nil
	})
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}

	err = cw.manager.GetClient().DeleteAllOf(r.Context(), &batchv1.Job{},
		client.InNamespace(os.Getenv("NAMESPACE")),
		client.MatchingLabels(targetLabels(keyspace, pool)),
		client.PropagationPolicy(metav1.DeletePropagationBackground))
	if err != nil {
		cw.logger.Error(err, "failed to pause jobs", "keyspace", keyspace)
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	w.WriteHeader(http.StatusOK)
	cw.logger.Info("paused tasks", "keyspace", keyspace, "pool", pool)
}

// resume worker tasks.
func (cw *ConfigReconciler) resume(w http.ResponseWriter, r *http.Request) {
	keyspace := r.URL.Query().Get("keyspace")
	pool := r.URL.Query().Get("pool")
	scope := targetJobs(keyspace, pool)
	err := cw.db.WithContext(r.Context()).Transaction(func(tx *gorm.DB) error {
		err := tx.Model(&gc.Task{}).Scopes(scope).
			Where("state = ?", gc.StatePaused).
			Update("state", gc.StatePending).Error
		if err != nil && !isTableNotExistsError(err) {
			return err
		}
		err = tx.Model(&gcv2.Task{}).Scopes(scope).
			Where("state = ?", gcv2.StatePaused).
			Update("state", gcv2.StatePending).Error
		if err != nil && !isTableNotExistsError(err) {
			return err
		}
		err = tx.Model(&bgtask.Task{}).Scopes(scope).
			Where("state = ?", bgtask.StatePaused).
			Update("state", bgtask.StatePending).Error
		if err != nil && !isTableNotExistsError(err) {
			return err
		}
		err = tx.Model(&autoanalyze.Task{}).Scopes(scope).
			Where("state = ?", autoanalyze.StatePaused).
			Update("state", autoanalyze.StatePending).Error
		if err != nil && !isTableNotExistsError(err) {
			return err
		}
		err = tx.Model(&ttl_task.Task{}).Scopes(scope).
			Where("state = ?", ttl_task.StatePaused).
			Update("state", ttl_task.StatePending).Error
		if err != nil && !isTableNotExistsError(err) {
			return err
		}
		// Special treatment to ttl_table_info, which does not contain pool info.
		query := tx.Model(&ttl_task.TTLTableInfo{}).Where("paused = ?", true)
		if keyspace != "" {
			query = query.Where("keyspace_name = ?", keyspace)
		}
		err = query.Update("paused", false).Error
		if err != nil && !isTableNotExistsError(err) {
			return err
		}
		return nil
	})

	if err != nil {
		cw.logger.Error(err, "failed to resume tasks", "keyspace", keyspace)
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	w.WriteHeader(http.StatusOK)
	cw.logger.Info("resumed tasks", "keyspace", keyspace, "pool", pool)
}

func (cw *ConfigReconciler) clean(w http.ResponseWriter, r *http.Request) {
	keyspace := r.URL.Query().Get("keyspace")
	pool := r.URL.Query().Get("pool")

	if keyspace == "" {
		http.Error(w, "keyspace must be provided for cleaning", http.StatusBadRequest)
		return
	}
	scope := targetJobs(keyspace, pool)

	err := cw.db.WithContext(r.Context()).Transaction(func(tx *gorm.DB) error {
		err := tx.Model(&gc.Task{}).Scopes(scope).Delete(&gc.Task{}).Error
		if err != nil && !isTableNotExistsError(err) {
			return err
		}
		err = tx.Model(&gcv2.Task{}).Scopes(scope).Delete(&gcv2.Task{}).Error
		if err != nil && !isTableNotExistsError(err) {
			return err
		}
		err = tx.Model(&bgtask.Task{}).Scopes(scope).Delete(&bgtask.Task{}).Error
		if err != nil && !isTableNotExistsError(err) {
			return err
		}
		err = tx.Model(&autoanalyze.Task{}).Scopes(scope).Delete(&autoanalyze.Task{}).Error
		if err != nil && !isTableNotExistsError(err) {
			return err
		}
		err = tx.Model(&ttl_task.Task{}).Scopes(scope).Delete(&ttl_task.Task{}).Error
		if err != nil && !isTableNotExistsError(err) {
			return err
		}
		// Special treatment to ttl_table_info, which does not contain pool info.
		query := tx.Model(&ttl_task.TTLTableInfo{})
		if keyspace != "" {
			query = query.Where("keyspace_name = ?", keyspace)
		}
		err = query.Delete(&ttl_task.TTLTableInfo{}).Error
		if err != nil && !isTableNotExistsError(err) {
			return err
		}
		return nil
	})
	if err != nil {
		cw.logger.Error(err, "failed to clean tasks", "keyspace", keyspace)
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	err = cw.manager.GetClient().DeleteAllOf(r.Context(), &batchv1.Job{},
		client.InNamespace(os.Getenv("NAMESPACE")),
		client.MatchingLabels(targetLabels(keyspace, pool)),
		client.PropagationPolicy(metav1.DeletePropagationBackground))
	if err != nil {
		cw.logger.Error(err, "failed to clean jobs", "keyspace", keyspace)
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	w.WriteHeader(http.StatusOK)
	cw.logger.Info("cleaned tasks", "keyspace", keyspace, "pool", pool)
}

// setLogLevel set log level.
func (cw *ConfigReconciler) setLogLevel(w http.ResponseWriter, r *http.Request) {
	levelStr := r.URL.Query().Get("level")
	cw.logger.Info("set log level ", "level", levelStr)
	err := logutil.SetLevel(levelStr)
	if err != nil {
		cw.logger.Error(err, "failed to set log level", "level", levelStr)
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	w.WriteHeader(http.StatusOK)
	cw.logger.Info("set log level succeeded", "level", levelStr)
}

// setTestKeyspace set test keyspace in RM.
func (cw *ConfigReconciler) setTestKeyspace(w http.ResponseWriter, r *http.Request) {
	keyspace_name := r.URL.Query().Get("keyspace_name")
	log.Info("set test keyspace ", zap.String("keyspace_name", keyspace_name))
	cw.TestRMkeyspaces[keyspace_name] = keyspace_name
	w.WriteHeader(http.StatusOK)
	log.Info("append a test keyspace",
		zap.String("keyspace_name", keyspace_name),
		zap.Any("test-keyspaces", cw.TestRMkeyspaces))
}

func (cw *ConfigReconciler) getKeyspacePodCount(w http.ResponseWriter, r *http.Request) {
	keyspace := r.URL.Query().Get("keyspace")
	if keyspace == "" {
		http.Error(w, "keyspace must be provided", http.StatusBadRequest)
		return
	}

	podList := &corev1.PodList{}
	err := cw.manager.GetClient().List(r.Context(), podList,
		client.InNamespace(tidbWorkerNamespace),
		client.MatchingLabels{
			types.KeyspaceNameLabelKey: keyspace,
			labels.ComponentLabelKey:   "worker",
		})
	if err != nil {
		cw.logger.Error(err, "failed to list pods for keyspace", "keyspace", keyspace)
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}

	w.WriteHeader(http.StatusOK)
	_, err = w.Write([]byte(strconv.Itoa(len(podList.Items))))
	if err != nil {
		cw.logger.Error(err, "failed to get keyspace tidb worker pod count", "keyspace", keyspace)
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	cw.logger.Info("get keyspace pod count", "keyspace", keyspace, "count", len(podList.Items))

}