package managerserver

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

	"github.com/go-logr/logr"
	"github.com/tidbcloud/aws-shared-provider/pkg/types"
	"github.com/tidbcloud/aws-shared-provider/pkg/utils"
	utilslabels "github.com/tidbcloud/aws-shared-provider/pkg/utils/labels"
	"github.com/tidbcloud/aws-shared-provider/pkg/utils/pd"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/labels"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

// Mirrored from pkg/tidbworker (owned by another team) so we don't import
// from that package.
const (
	defaultTiDBWorkerNamespace    = "tidb-worker"
	tidbWorkerComponentLabelValue = "worker"
)

const (
	migrateKeyspaceNameParam       = "keyspace-name"
	migrateMetaServiceGroupIDParam = "meta-service-group-id"
	migrateWaitDurationParam       = "wait-duration"
)

type migrateResult struct {
	KeyspaceName       string `json:"keyspace_name"`
	MetaServiceGroupID string `json:"meta_service_group_id"`
	Status             string `json:"status"`
}

// migrateMetaServiceGroup transfers a keyspace onto a different meta-service
// group. Assumes a single tidb-manager replica is running for the duration.
func (s *server) migrateMetaServiceGroup(w http.ResponseWriter, r *http.Request) {
	logger := log.WithName("keyspace/migrate_meta_service_group")
	ctx := r.Context()

	keyspaceName := strings.TrimSpace(r.URL.Query().Get(migrateKeyspaceNameParam))
	targetGroupID := strings.TrimSpace(r.URL.Query().Get(migrateMetaServiceGroupIDParam))
	logger = logger.WithValues("keyspace-name", keyspaceName, "meta-service-group-id", targetGroupID)

	if keyspaceName == "" {
		writeMigrateError(w, logger, http.StatusBadRequest,
			fmt.Errorf("query parameter %q is required", migrateKeyspaceNameParam))
		return
	}
	if targetGroupID == "" {
		writeMigrateError(w, logger, http.StatusBadRequest,
			fmt.Errorf("query parameter %q is required", migrateMetaServiceGroupIDParam))
		return
	}
	if s.pdClient == nil {
		writeMigrateError(w, logger, http.StatusInternalServerError,
			errors.New("pd client is not configured"))
		return
	}
	if s.migrationLocker == nil {
		writeMigrateError(w, logger, http.StatusInternalServerError,
			errors.New("migration locker is not configured"))
		return
	}

	logger.Info("migration request received")
	start := time.Now()

	waitDuration, err := parseWaitDuration(r)
	if err != nil {
		writeMigrateError(w, logger, http.StatusBadRequest, err)
		return
	}

	keyspaceMeta, err := s.pdClient.GetKeyspaceByName(ctx, keyspaceName)
	if err != nil {
		writeMigrateError(w, logger, http.StatusBadRequest,
			fmt.Errorf("get keyspace %q from pd: %w", keyspaceName, err))
		return
	}
	logger.Info("fetched keyspace from pd",
		"current-meta-service-group-id", keyspaceMeta.Config.MetaServiceGroupID)
	if keyspaceMeta.Config.MetaServiceGroupID == targetGroupID {
		logger.Info("keyspace already on target meta-service group, nothing to do")
		writeMigrateOK(w, logger, &migrateResult{
			KeyspaceName:       keyspaceName,
			MetaServiceGroupID: targetGroupID,
			Status:             "noop",
		})
		return
	}
	if err := s.assertMetaServiceGroupExists(ctx, targetGroupID); err != nil {
		writeMigrateError(w, logger, http.StatusBadRequest, err)
		return
	}
	logger.Info("target meta-service group exists in pd")

	logger.Info("running pre-check for active components")
	if err := s.assertNoKeyspaceComponentsRunning(ctx, keyspaceName); err != nil {
		writeMigrateError(w, logger, http.StatusConflict, err)
		return
	}
	logger.Info("pre-check passed, no components running")

	release, ok := s.migrationLocker.TryAcquire(keyspaceName)
	if !ok {
		writeMigrateError(w, logger, http.StatusConflict,
			fmt.Errorf("another migration is already in progress for keyspace %q", keyspaceName))
		return
	}
	logger.Info("acquired migration lock")
	defer func() {
		release()
		logger.Info("released migration lock")
	}()

	if s.tidbWorkerApi != nil {
		logger.Info("pausing tidb-worker tasks")
		if err := s.tidbWorkerApi.Pause(ctx, keyspaceName); err != nil {
			writeMigrateError(w, logger, http.StatusInternalServerError,
				fmt.Errorf("pause tidb-worker for keyspace %q: %w", keyspaceName, err))
			return
		}
		logger.Info("paused tidb-worker tasks")
	}

	if waitDuration > 0 {
		logger.Info("waiting before re-check", "duration", waitDuration.String())
		if err := sleepCtx(ctx, waitDuration); err != nil {
			s.tryResumeTiDBWorker(ctx, logger, keyspaceName)
			writeMigrateError(w, logger, http.StatusInternalServerError, err)
			return
		}
	}

	logger.Info("re-checking for active components after pause")
	if err := s.assertNoKeyspaceComponentsRunning(ctx, keyspaceName); err != nil {
		s.tryResumeTiDBWorker(ctx, logger, keyspaceName)
		writeMigrateError(w, logger, http.StatusConflict, err)
		return
	}
	logger.Info("re-check passed")

	logger.Info("updating keyspace meta-service-group in pd")
	if err := s.pdClient.UpdateKeyspaceConfig(ctx, keyspaceName, &pd.KeyspaceConfig{
		MetaServiceGroupID: targetGroupID,
	}); err != nil {
		s.tryResumeTiDBWorker(ctx, logger, keyspaceName)
		writeMigrateError(w, logger, http.StatusInternalServerError,
			fmt.Errorf("update keyspace config in pd: %w", err))
		return
	}
	logger.Info("pd config updated")

	if s.tidbWorkerApi != nil {
		logger.Info("resuming tidb-worker tasks")
		// PD already updated; surface a resume error but do not roll back.
		if err := s.tidbWorkerApi.Resume(ctx, keyspaceName); err != nil {
			writeMigrateError(w, logger, http.StatusInternalServerError,
				fmt.Errorf("pd update succeeded but resume tidb-worker failed for keyspace %q: %w", keyspaceName, err))
			return
		}
		logger.Info("resumed tidb-worker tasks")
	}

	logger.Info("migration completed", "elapsed", time.Since(start).String())
	writeMigrateOK(w, logger, &migrateResult{
		KeyspaceName:       keyspaceName,
		MetaServiceGroupID: targetGroupID,
		Status:             "migrated",
	})
}

func (s *server) assertMetaServiceGroupExists(ctx context.Context, groupID string) error {
	groups, err := s.pdClient.GetMetaServiceGroupList(ctx)
	if err != nil {
		return fmt.Errorf("list meta-service groups from pd: %w", err)
	}
	if groups != nil {
		for _, g := range *groups {
			if g.ID == groupID {
				return nil
			}
		}
	}
	return fmt.Errorf("meta-service group %q not registered in pd", groupID)
}

func (s *server) assertNoKeyspaceComponentsRunning(ctx context.Context, keyspaceName string) error {
	for ns, tManager := range s.tManagers {
		if tManager == nil {
			continue
		}
		// Walk every active token rather than exact-match on keyspaceName,
		// because RG / export tokens look like "keyspace/rg" or
		// "keyspace:exportID" and would otherwise be missed.
		var foundToken string
		var foundEndpoints []string
		tManager.WalkActiveTokenFromAssignCache(func(token string, endpoints []string, _ int) bool {
			if utils.TrimToken(token) == keyspaceName && len(endpoints) > 0 {
				foundToken = token
				foundEndpoints = endpoints
				return false
			}
			return true
		})
		if len(foundEndpoints) > 0 {
			return fmt.Errorf("tidb pod(s) still running for keyspace %q in pool %q (token %q): %v",
				keyspaceName, ns, foundToken, foundEndpoints)
		}
	}
	if err := s.assertNoKeyspacePodsInNamespace(ctx, keyspaceName, importNamespace, "lightning"); err != nil {
		return err
	}
	if err := s.assertNoKeyspacePodsInNamespace(ctx, keyspaceName, exportNamespace, "dumpling"); err != nil {
		return err
	}
	return s.assertNoTiDBWorkerPodsRunning(ctx, keyspaceName)
}

func (s *server) tidbWorkerNamespace() string {
	if s.cfg != nil && s.cfg.TiDBWorkerNS != "" {
		return s.cfg.TiDBWorkerNS
	}
	return defaultTiDBWorkerNamespace
}

func (s *server) assertNoTiDBWorkerPodsRunning(ctx context.Context, keyspaceName string) error {
	namespace := s.tidbWorkerNamespace()
	podList := &corev1.PodList{}
	err := s.manager.GetClient().List(ctx, podList,
		client.InNamespace(namespace),
		client.MatchingLabels{
			types.KeyspaceNameLabelKey:    keyspaceName,
			utilslabels.ComponentLabelKey: tidbWorkerComponentLabelValue,
		},
	)
	if err != nil {
		if apierrors.IsNotFound(err) {
			return nil
		}
		return fmt.Errorf("list tidb-worker pods for keyspace %q in %q: %w",
			keyspaceName, namespace, err)
	}
	if len(podList.Items) > 0 {
		names := make([]string, 0, len(podList.Items))
		for _, p := range podList.Items {
			names = append(names, p.Name)
		}
		return fmt.Errorf("tidb-worker pod(s) still running for keyspace %q in %q: %v",
			keyspaceName, namespace, names)
	}
	return nil
}

func (s *server) assertNoKeyspacePodsInNamespace(
	ctx context.Context,
	keyspaceName, namespace, componentName string,
) error {
	podList := &corev1.PodList{}
	err := s.manager.GetClient().List(ctx, podList,
		client.InNamespace(namespace),
		client.MatchingLabelsSelector{
			Selector: labels.SelectorFromSet(labels.Set{
				types.KeyspaceNameLabel: keyspaceName,
			}),
		},
	)
	if err != nil {
		if apierrors.IsNotFound(err) {
			return nil
		}
		return fmt.Errorf("list %s pods for keyspace %q in %q: %w", componentName, keyspaceName, namespace, err)
	}
	if len(podList.Items) > 0 {
		names := make([]string, 0, len(podList.Items))
		for _, p := range podList.Items {
			names = append(names, p.Name)
		}
		return fmt.Errorf("%s pod(s) still running for keyspace %q in %q: %v",
			componentName, keyspaceName, namespace, names)
	}
	return nil
}

// parseWaitDuration validates the wait-duration query parameter.
// Manual-test hook for injecting state changes while the lock is held.
func parseWaitDuration(r *http.Request) (time.Duration, error) {
	raw := r.URL.Query().Get(migrateWaitDurationParam)
	if raw == "" {
		return 0, nil
	}
	seconds, err := strconv.Atoi(raw)
	if err != nil || seconds < 0 {
		return 0, fmt.Errorf("query parameter %q must be a non-negative integer (seconds), got %q",
			migrateWaitDurationParam, raw)
	}
	return time.Duration(seconds) * time.Second, nil
}

func sleepCtx(ctx context.Context, d time.Duration) error {
	select {
	case <-time.After(d):
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}
}

func (s *server) tryResumeTiDBWorker(ctx context.Context, logger logr.Logger, keyspaceName string) {
	if s.tidbWorkerApi == nil {
		return
	}
	if err := s.tidbWorkerApi.Resume(ctx, keyspaceName); err != nil {
		logger.Error(err, "resume tidb-worker")
	}
}

func writeMigrateOK(w http.ResponseWriter, logger logr.Logger, result *migrateResult) {
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(http.StatusOK)
	if err := json.NewEncoder(w).Encode(result); err != nil {
		logger.Error(err, "encode response")
	}
}

func writeMigrateError(w http.ResponseWriter, logger logr.Logger, statusCode int, err error) {
	logger.Error(err, "migrate_meta_service_group failed", "status", statusCode)
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(statusCode)
	_ = json.NewEncoder(w).Encode(map[string]string{"error": err.Error()})
}