package adapterimpl

import (
	"context"
	"sync/atomic"
	"time"

	"github.com/prometheus/client_golang/prometheus"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"

	regionalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/regional/serverless/v1"
	"github.com/tidbcloud/tidb-management-service/serverless-global-service/domain/entity"
	"github.com/tidbcloud/tidb-management-service/serverless-global-service/infra/adapterimpl/cli"
	"github.com/tidbcloud/tidb-management-service/serverless-global-service/pkg/config"
)

var (
	RegionalProxyEnabled = prometheus.NewGauge(prometheus.GaugeOpts{
		Namespace: "regional_proxy",
		Name:      "enabled",
		Help:      "whether enable regional proxy",
	})

	RegionalProxyHits = prometheus.NewCounter(prometheus.CounterOpts{
		Namespace: "regional_proxy",
		Name:      "hits_total",
		Help:      "number of regional proxy hit",
	})

	RegionalProxyLatency = prometheus.NewHistogramVec(prometheus.HistogramOpts{
		Namespace: "regional_proxy",
		Name:      "latency",
		Help:      "Latency of regional proxy",
	}, []string{"method"})

	RegionalProxyFailedCount = prometheus.NewCounter(prometheus.CounterOpts{
		Namespace: "regional_proxy",
		Name:      "failed_total",
		Help:      "number of regional proxy failed",
	})
)

func init() {
	prometheus.MustRegister(RegionalProxyEnabled)
	prometheus.MustRegister(RegionalProxyHits)
	prometheus.MustRegister(RegionalProxyLatency)
}

type RegionalServiceProxy struct {
	available *atomic.Bool
	cli       cli.RegionalProxyClient

	expired chan struct{}
}

func NewRegionalServiceProxy(client cli.RegionalProxyClient) *RegionalServiceProxy {
	a := &atomic.Bool{}
	a.Store(true)
	p := &RegionalServiceProxy{
		available: a,
		cli:       client,
		expired:   make(chan struct{}),
	}

	if p.cli == nil {
		RegionalProxyEnabled.Set(0)
	}

	go p.startExpireLoop(context.TODO())

	return p
}

func (s *RegionalServiceProxy) Enabled() bool {
	enabled := s.available.Load() && config.Global().RegionalServiceConfig.RegionalProxyServiceConfig.Enabled &&
		s.cli != nil
	if enabled {
		RegionalProxyEnabled.Set(1)
	} else {
		RegionalProxyEnabled.Set(0)
	}

	return enabled
}

func (s *RegionalServiceProxy) GetReadyClusterByIds(
	ctx context.Context,
	ids []string,
) ([]*regionalserverlessv1.ServerlessRegionalServiceClusterInfo, bool) {
	if !s.Enabled() {
		return nil, false
	}

	start := time.Now()
	defer func() {
		RegionalProxyLatency.WithLabelValues("GET").Observe(time.Since(start).Seconds())
	}()

	infos, ok := s.cli.GetClusterDataByIds(ctx, ids)
	if ok && len(ids) == len(infos) {
		RegionalProxyHits.Add(1)
		return infos, true
	}

	return nil, false
}

func (s *RegionalServiceProxy) ReadyClusterBatchSet(
	ctx context.Context,
	clusters []*regionalserverlessv1.ServerlessRegionalServiceClusterInfo,
) {
	if !s.Enabled() {
		return
	}

	out := make([]*regionalserverlessv1.ServerlessRegionalServiceClusterInfo, 0)
	for _, cluster := range clusters {
		if !entity.IsReady4Launch(ctx, cluster) {
			tilog.Warn(ctx, "cluster is not ready for launch", zap.String("id", cluster.ClusterId))
			continue
		}

		if cluster.Endpoints == nil {
			tilog.Warn(ctx, "empty endpoints", zap.String("id", cluster.ClusterId))
			continue
		}

		if cluster.Endpoints.Private == nil {
			tilog.Warn(ctx, "empty private endpoint", zap.String("id", cluster.ClusterId))
			continue
		}

		if !cluster.Endpoints.Private.Disabled && cluster.Endpoints.Private.Host == "" {
			tilog.Warn(ctx, "empty private endpoint host", zap.String("id", cluster.ClusterId))
			continue
		}

		out = append(out, cluster)
	}

	start := time.Now()
	defer func() {
		RegionalProxyLatency.WithLabelValues("BatchSet").Observe(time.Since(start).Seconds())
	}()

	ok, err := s.cli.BatchSet(ctx, out)
	if err != nil || !ok {
		RegionalProxyFailedCount.Inc()
		tilog.Error(ctx, "set cluster failed", zap.Any("clusters", clusters), zap.Error(err))
		s.Reset(ctx)
		return
	}
}

func (s *RegionalServiceProxy) Delete(ctx context.Context, id string) {
	if !s.Enabled() {
		return
	}

	start := time.Now()
	defer func() {
		RegionalProxyLatency.WithLabelValues("Delete").Observe(time.Since(start).Seconds())
	}()

	ok, err := s.cli.Delete(ctx, id)
	if err != nil || !ok {
		RegionalProxyFailedCount.Inc()
		tilog.Error(ctx, "remove cluster failed", zap.String("id", id), zap.Error(err))
		s.Reset(ctx)
		return
	}
}

func (s *RegionalServiceProxy) startExpireLoop(ctx context.Context) {
	ticker := time.NewTicker(time.Minute * 10)
	for {
		select {
		case <-s.expired:
			s.clean(ctx)
		case <-ticker.C:
			if !s.available.Load() {
				s.clean(ctx)
			}
		}
	}
}

func (s *RegionalServiceProxy) clean(ctx context.Context) {
	start := time.Now()
	defer func() {
		RegionalProxyLatency.WithLabelValues("DeleteAll").Observe(time.Since(start).Seconds())
	}()

	ok, err := s.cli.DeleteAll(ctx)
	if err != nil {
		s.available.Store(false)
		RegionalProxyFailedCount.Inc()
		tilog.Error(ctx, "remove all data failed", zap.Error(err))
		return
	}

	if !ok {
		RegionalProxyFailedCount.Inc()
		s.available.Store(false)
		tilog.Warn(ctx, "remove data failed")
		return
	}

	tilog.Info(ctx, "clean cache done")
	s.available.Store(true)
}

func (s *RegionalServiceProxy) Reset(ctx context.Context) {
	if s.available.CompareAndSwap(true, false) {
		tilog.Info(ctx, "Reset cache")
		select {
		case s.expired <- struct{}{}:
		default:
		}
	}
}