package slprober

import (
	"fmt"
	"math/rand"
	"sync"
	"time"

	"github.com/pingcap/dbaas/pkg/constants"
	"github.com/pingcap/log"
	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	tidbClusterInformers "github.com/pingcap/tidb-operator/pkg/client/informers/externalversions"
	tidbClusterV1alpha1listers "github.com/pingcap/tidb-operator/pkg/client/listers/pingcap/v1alpha1"
	"github.com/uber-go/atomic"
	"go.uber.org/zap"
	"k8s.io/apimachinery/pkg/labels"
	kubeinformers "k8s.io/client-go/informers"
	"k8s.io/client-go/kubernetes"
	corev1list "k8s.io/client-go/listers/core/v1"

	"github.com/tidbcloud/billing-reporter/slprober/handshake"
	"github.com/tidbcloud/billing-reporter/utils"
)

type ServiceStatus string

const (
	ServiceStatusDetectFailed               ServiceStatus = "detect_failed"
	ServiceStatusSkippedByNotReady          ServiceStatus = "skipped_by_not_ready"
	ServiceStatusDetectedAsDead             ServiceStatus = "dead"
	ServiceStatusDetectedAsAliveButOutdated ServiceStatus = "alive_outdated"
	ServiceStatusDetectedAsAlive            ServiceStatus = "alive"
)

const (
	ProbeTimeout = time.Second * 5
)

type ServiceLevelProber struct {
	kubeCli   kubernetes.Interface
	svcLister corev1list.ServiceLister
	tcLister  tidbClusterV1alpha1listers.TidbClusterLister
}

func NewServiceLevelProber(
	kubeCli kubernetes.Interface,
	kubeFactory kubeinformers.SharedInformerFactory,
	informerFactory tidbClusterInformers.SharedInformerFactory) *ServiceLevelProber {
	return &ServiceLevelProber{
		kubeCli:   kubeCli,
		svcLister: kubeFactory.Core().V1().Services().Lister(),
		tcLister:  informerFactory.Pingcap().V1alpha1().TidbClusters().Lister(),
	}
}

type probeResult struct {
	cluster *v1alpha1.TidbCluster
	status  ServiceStatus
	err     error
}

func (m *ServiceLevelProber) ProbeStatus() {
	log.Debug("ServiceLevelProber::BeginProbeClusters")

	timeBegin := time.Now()
	clusters, err := m.tcLister.List(labels.Everything())
	timeListDuration := time.Since(timeBegin)
	if err != nil {
		log.Error("ServiceLevelProber::ListClustersFailed",
			zap.Duration("list_elapsed", timeListDuration),
			zap.Error(err))
		ProbeDuration.WithLabelValues("list_clusters_failed").Observe(timeListDuration.Seconds())
		return
	}
	ProbeDuration.WithLabelValues("list_clusters").Observe(timeListDuration.Seconds())

	timeProbeBegin := time.Now()

	// Let's ping all clusters concurrently
	probeResultsCh := make(chan probeResult, len(clusters))
	availableResults := atomic.NewUint32(0)
	isCanceled := atomic.NewBool(false)
	var wg sync.WaitGroup
	for _, cluster := range clusters {
		c := cluster
		wg.Add(1)
		go m.probeCluster(c, &wg, probeResultsCh, availableResults, isCanceled)
	}
	waitTimeout(&wg, ProbeTimeout)
	resultsN := int(availableResults.Load()) // Not all results might be ready
	isCanceled.Store(true)                   // Cancel all flying requests

	// Generate statistics summary for output
	statsProbeSuccess := 0
	statsProbeFailed := 0
	statsProbeTimeout := len(clusters) - resultsN
	statsAliveClusters := 0
	statsDeadClusters := 0
	statsSkipClusters := 0
	for i := 0; i < resultsN; i++ {
		r := <-probeResultsCh
		switch r.status {
		case ServiceStatusDetectedAsAlive:
			statsProbeSuccess++
			statsAliveClusters++
		case ServiceStatusDetectedAsDead:
			statsProbeSuccess++
			statsDeadClusters++
		case ServiceStatusDetectFailed:
			statsProbeFailed++
		case ServiceStatusSkippedByNotReady:
			statsProbeSuccess++
			statsSkipClusters++
		case ServiceStatusDetectedAsAliveButOutdated:
			// Already counted in probe timeout?
		}
	}
	timeProbeDuration := time.Since(timeProbeBegin)
	timeDuration := time.Since(timeBegin)
	log.Info("ServiceLevelProber::ProbeClustersFinished",
		zap.Int("all_clusters", len(clusters)),
		zap.Int("probe_success_clusters", statsProbeSuccess),
		zap.Int("probe_failed_clusters", statsProbeFailed),
		zap.Int("probe_timeout_clusters", statsProbeTimeout),
		zap.Int("alive_clusters", statsAliveClusters),
		zap.Int("dead_clusters", statsDeadClusters),
		zap.Int("skip_clusters", statsSkipClusters),
		zap.Duration("list_elapsed", timeListDuration),
		zap.Duration("probe_elapsed", timeProbeDuration),
		zap.Duration("total_elapsed", timeDuration),
	)
	ProbeDuration.WithLabelValues("probe_clusters").Observe(timeProbeDuration.Seconds())
	ProbeDuration.WithLabelValues("op").Observe(timeDuration.Seconds())
}

func (m *ServiceLevelProber) probeClusterImpl(cluster *v1alpha1.TidbCluster) (ServiceStatus, error) {
	tenantId := getClusterLabel(cluster, constants.TiDBClusterTenantIDLabel)
	clusterId := getClusterLabel(cluster, constants.TiDBClusterClusterIDLabel)

	// Skip creating status
	isCreating, err := utils.IsClusterInCreatingStatus(m.kubeCli, cluster)
	if err != nil {
		log.Warn("ServiceLevelProber::GetCreatingStatusFailed",
			zap.String("namespace", cluster.Namespace),
			zap.String("tenant-id", tenantId),
			zap.String("cluster-id", clusterId),
			zap.Error(err))
	}
	if isCreating {
		return ServiceStatusSkippedByNotReady, nil
	}

	// FIXME: Maybe we should read MySQL port from spec instead?
	addr := fmt.Sprintf("%s.%s.svc.cluster.local:%d", "db-tidb", cluster.Namespace, 4000)
	hsPacket, err := handshake.TryConnect(addr)
	if err != nil {
		return ServiceStatusDetectedAsDead, err
	}

	ProbeClusterSuccessCounter.
		WithLabelValues(clusterId, hsPacket.ServerVersion).
		Inc()
	return ServiceStatusDetectedAsAlive, nil
}

func (m *ServiceLevelProber) probeCluster(cluster *v1alpha1.TidbCluster,
	wg *sync.WaitGroup,
	resultCh chan probeResult,
	resultN *atomic.Uint32,
	cancelFlag *atomic.Bool) {
	// Pick a time randomly in future 200ms to actually start the probe.
	// This avoid we start too many detections simultaneously when there are many clusters.
	// FIXME: G404: Use of weak random number generator (math/rand instead of crypto/rand) (gosec)
	time.Sleep(time.Duration(rand.Intn(200)) * time.Millisecond) //nolint: gosec

	tenantId := getClusterLabel(cluster, constants.TiDBClusterTenantIDLabel)
	clusterId := getClusterLabel(cluster, constants.TiDBClusterClusterIDLabel)
	log.Debug("ServiceLevelProber::BeginProbeCluster",
		zap.String("namespace", cluster.Namespace),
		zap.String("tenant-id", tenantId),
		zap.String("cluster-id", clusterId))

	timeBegin := time.Now()
	status, err := m.probeClusterImpl(cluster)
	timeDuration := time.Since(timeBegin)

	isCanceled := cancelFlag.Load()
	if status == ServiceStatusDetectedAsAlive && isCanceled {
		// Detected that service is alive, but it took too much time. We treat as failures.
		err = fmt.Errorf("service is detected as alive but took too much time")
		status = ServiceStatusDetectedAsAliveButOutdated
	}
	if err != nil {
		log.Error("ServiceLevelProber::ProbeClusterFailedOrNotAlive",
			zap.Duration("elapsed", timeDuration),
			zap.String("namespace", cluster.Namespace),
			zap.String("tenant-id", tenantId),
			zap.String("cluster-id", clusterId),
			zap.Bool("is-canceled", isCanceled),
			zap.String("status", string(status)),
			zap.Error(err))
	}

	ProbeClusterCounter.
		WithLabelValues(string(status), cluster.Namespace, tenantId, clusterId).
		Inc()
	ProbeClusterDuration.
		WithLabelValues(string(status), clusterId).
		Observe(timeDuration.Seconds())

	resultCh <- probeResult{
		cluster: cluster,
		status:  status,
		err:     err,
	}
	resultN.Add(1)
	wg.Done()
}