package managerserver

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

	pkginfra "github.com/tidbcloud/aws-shared-provider/pkg/infra"
	"github.com/tidbcloud/aws-shared-provider/pkg/metrics"
	"github.com/tidbcloud/aws-shared-provider/pkg/types"
	"github.com/tidbcloud/aws-shared-provider/pkg/utils"
	ep "github.com/tidbcloud/aws-shared-provider/pkg/utils/endpoint"
	"github.com/tidbcloud/aws-shared-provider/pkg/utils/tidb"
	infra "github.com/tidbcloud/tidb-management-service/api/serverless_infra"
	"github.com/tidbcloud/tidb-management-service/api/serverless_infra/meta"
)

func (s *server) assign(w http.ResponseWriter, r *http.Request) {
	logger := log.WithName("tidb/assign")
	logger.V(2).Info("req param", "req", r.URL)

	var err error
	var token, ns string
	start := time.Now()

	defer func() {
		if err != nil {
			w.WriteHeader(http.StatusInternalServerError)
			_, _ = io.WriteString(w, err.Error())
			metrics.RequestHistogram.WithLabelValues("assign", "500", ns).Observe(time.Since(start).Seconds())
			return
		}
		metrics.RequestHistogram.WithLabelValues("assign", "200", ns).Observe(time.Since(start).Seconds())
	}()

	token, ns = r.URL.Query().Get("token"), r.URL.Query().Get("ns")
	if token == "" {
		logger.Error(noTokenErr, "get token")
		err = noTokenErr
		return
	}

	regionalInfra := pkginfra.NewRegionalInfraWithClient(s.infraClusterCli, s.cfg.KubeClusterName)
	meta, err := pkginfra.GetClusterOrBranchMeta(r.Context(), regionalInfra, s.infraApi, utils.TrimToken(token), s.cfg.KubeClusterName)
	if err != nil {
		logger.Error(err, "fail to get token", "token", token)
		return
	}
	isExport := isExportToken(token)
	if !isExport && meta.TiDBPool != ns {
		logger.Error(metaNotSyncErr, "ns not match in meta.TiDBPool", "token", token, "ns", ns, "meta.TiDBPool", meta.TiDBPool)
		err = metaNotSyncErr
		return
	}

	var maxIdleSeconds uint = 0
	m, err := strconv.Atoi(r.URL.Query().Get("max_idle_seconds"))
	if err == nil && m > 0 {
		maxIdleSeconds = uint(m)
	}
	// RCU>0 means essential plan, rewrite maxIdleSeconds with the highest priority.
	if meta.IsEssential() {
		maxIdleSeconds = uint(types.TidbMaxIdleSeconds_EssentialPlan)
	}

	enableAuditLog := meta.EnableAuditLog
	auditLogRedacted := meta.AuditLogRedacted
	projectID := meta.ProjectID
	clusterID := meta.ClusterID
	tenantID := meta.TenantID
	keyspaceID := meta.KeyspaceID

	logger = logger.WithValues(
		"token", token,
		"ns", ns,
		"enableAuditLog", enableAuditLog,
		"auditLogRedacted", auditLogRedacted,
		"maxIdleSeconds", maxIdleSeconds,
		"keyspaceID", keyspaceID,
		"projectID", projectID,
		"clusterID", clusterID,
		"tenantID", tenantID,
	)

	ips, err := s.assignTidb(
		r.Context(),
		token,
		ns,
		&tidb.ActivateParameter{
			EnableAuditLog:   enableAuditLog,
			AuditLogRedacted: auditLogRedacted,
			MaxIdleSeconds:   maxIdleSeconds,
			KeyspaceID:       keyspaceID,
			ProjectID:        projectID,
			ClusterID:        clusterID,
			TenantID:         tenantID,
		},
	)
	if err != nil {
		logger.Error(err, "assignTidb")
		if errors.Is(err, utils.ErrNotFoundToken) && !isExport {
			s.tryResumeEtcdToken(r.Context(), meta)
		}
		return
	}
	_, _ = io.WriteString(w, strings.Join(ips, ","))
}

func (s *server) assignTidb(
	ctx context.Context,
	token,
	ns string,
	param *tidb.ActivateParameter,
) ([]string, error) {
	tManager := s.tManagers[ns]
	if tManager == nil {
		return nil, noTidbPoolErr
	}
	endpoints, err := tManager.AssignTidb(ctx, token, param)
	if err != nil {
		return nil, err
	}

	ips := make([]string, 0, len(endpoints))
	for _, endpoint := range endpoints {
		_, ip := ep.ParsePodFromEndpoint(endpoint)
		if ip == "" {
			log.Info("[assignTidb] pod ip is empty, ignore")
			continue
		}
		ips = append(ips, ip)
	}
	return ips, nil
}

func (s *server) tryResumeEtcdToken(ctx context.Context, m *pkginfra.ClusterOrBranchMeta) {
	logger := log.WithName("tryResumeEtcdToken").WithValues(
		"cluster_type", m.ClusterType, "cluster_id", m.ClusterID, "tidbPool", m.TiDBPool, "token", m.Token)

	if m.Annotations[types.AnnoKeyResumeEtcdToken] != "" {
		logger.V(1).Info("etcd token has been resumed recently, skip",
			"last_resume_time", m.Annotations[types.AnnoKeyResumeEtcdToken])
		return
	}

	switch m.ClusterType {
	case pkginfra.TypeCluster:
		err := s.infraClusterCli.Patch(ctx, map[string]any{
			fmt.Sprintf("annotations.'%s'", types.AnnoKeyResumeEtcdToken): time.Now().Format(time.RFC3339),
		}, &infra.Cluster{ObjectMeta: meta.ObjectMeta{Namespace: m.InfraNamespace, Name: m.ClusterID}})
		if err != nil {
			logger.Error(err, "failed to patch cluster")
			return
		}
	case pkginfra.TypeBranch:
		data := fmt.Sprintf(`{"metadata":{"annotations":{"%s":"%s"}}}`,
			types.AnnoKeyResumeEtcdToken, time.Now().Format(time.RFC3339))
		_, err := s.infraApi.PatchBranch(ctx, m.InfraNamespace, m.ClusterID, []byte(data))
		if err != nil {
			logger.Error(err, "failed to patch branch")
			return
		}
	default:
		logger.Error(nil, "unknown cluster type")
		return
	}

	logger.Info("succeed to patch resume etcd token annotations")
	metrics.ResumeEtcdTokenCounter.WithLabelValues(m.TiDBPool).Inc()
}

func isExportToken(token string) bool {
	// export token is in the format of "keyspaceToken:exportID", we can check if it contains ":"
	return strings.Contains(token, ":")
}