package import_collector

import (
	"context"
	"fmt"
	"strconv"
	"time"

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/tidbcloud/serverless-service/internal/config"
	"github.com/tidbcloud/serverless-service/internal/domain/constant"
	"github.com/tidbcloud/serverless-service/internal/domain/entity"
	"github.com/tidbcloud/serverless-service/internal/domain/repo"
	errhandler "github.com/tidbcloud/serverless-service/internal/domain/vo"
	"github.com/tidbcloud/serverless-service/internal/metric"
	"github.com/tidbcloud/serverless-service/internal/service/import_svc"
	"github.com/tidbcloud/serverless-service/internal/utils"
	"go.uber.org/zap"
)

const recentFailDuration = time.Second * 120

var _ prometheus.Collector = &ImportCollector{}

type ImportCollector struct {
	importHelper *import_svc.ImportHelperService
}

func NewImportCollector(importHelper *import_svc.ImportHelperService) *ImportCollector {
	return &ImportCollector{
		importHelper: importHelper,
	}
}

func (c *ImportCollector) Describe(ch chan<- *prometheus.Desc) {
	ch <- metric.SyncingImportSinceCreatedSeconds
	ch <- metric.RecentFailImports
	ch <- metric.ImportSizeWithIndexBytes
}

func (c *ImportCollector) Collect(ch chan<- prometheus.Metric) {
	ctx := context.WithValue(context.Background(), config.CtxWorkerNameKey, "ImportCollector")
	logger := utils.GetWorkerLoggerFromContext(ctx)
	start := time.Now()
	defer func() {
		logger.Info("imports metrics collect time", zap.Int64("time", time.Since(start).Milliseconds()))
	}()

	imports, err := c.importHelper.GetImportsToSync(ctx, nil)
	if err != nil {
		logger.Error("GetImportsToSync failed", zap.Error(err))
	}

	for _, imp := range imports {
		dur := time.Since(imp.CreatedAt).Seconds()
		ch <- prometheus.MustNewConstMetric(
			metric.SyncingImportSinceCreatedSeconds,
			prometheus.GaugeValue,
			dur,
			imp.TenantID,
			imp.ClusterID,
			imp.ID,
			entity.ImportProviderType(imp.Cluster),
			string(imp.Status),
		)
		if imp.TotalSizeWithIndex != nil {
			ch <- prometheus.MustNewConstMetric(
				metric.ImportSizeWithIndexBytes,
				prometheus.GaugeValue,
				float64(*imp.TotalSizeWithIndex),
				imp.TenantID,
				imp.ClusterID,
				imp.ID,
				entity.ImportProviderType(imp.Cluster),
				string(imp.Status))
		}
	}

	imports, err = c.importHelper.GetFailImportsInRecent(ctx, nil /* tenantID */, recentFailDuration)
	if err != nil {
		logger.Error("GetFailImportsInRecent failed", zap.Error(err))
	}

	for _, imp := range imports {
		dur := time.Since(imp.CreatedAt).Seconds()
		ch <- prometheus.MustNewConstMetric(
			metric.RecentFailImports,
			prometheus.GaugeValue,
			dur,
			imp.TenantID,
			imp.ClusterID,
			imp.ID,
			entity.ImportProviderType(imp.Cluster),
			errhandler.PrettyMessage(imp.Message, imp.DataFormat),
			imp.Org.Plan,
			imp.Org.Kind,
		)
	}
}

var ImportStatusCount = prometheus.NewDesc(
	prometheus.BuildFQName(metric.MetricNamespace, "import", "status_count"),
	"The execution result count of import",
	append(metric.CommonLabels, metric.ProviderTypeLabel, "status", "tenant_plan"),
	nil)

const ImportTimeOut = 8 * time.Hour

var _ prometheus.Collector = &StatusCollector{}

type StatusCollector struct {
	importRepo   repo.ImportRepo
	importHelper *import_svc.ImportHelperService
}

type importObject struct {
	clusterName  string
	tenantID     entity.ID
	tenantName   string
	succNum      float64
	failNum      float64
	tenantPlan   string
	providerType string
}

func NewStatusCollector(
	importRepo repo.ImportRepo,
	importHelper *import_svc.ImportHelperService,
) *StatusCollector {
	return &StatusCollector{
		importRepo:   importRepo,
		importHelper: importHelper,
	}
}

func (c *StatusCollector) Describe(ch chan<- *prometheus.Desc) {
	ch <- ImportStatusCount
}

func (c *StatusCollector) Collect(ch chan<- prometheus.Metric) {
	ctx := context.WithValue(context.Background(), config.CtxWorkerNameKey, "ImportStatusCollector")
	logger := utils.GetWorkerLoggerFromContext(ctx)
	start := time.Now()
	defer func() {
		logger.Info("imports metrics collect time", zap.Int64("time", time.Since(start).Milliseconds()))
	}()
	var usedTime time.Duration
	todayTime := time.Now().Truncate(24 * time.Hour)
	imports, bizErr := c.importRepo.ListImportsByFilter(ctx, &entity.ImportFilter{
		CreatedAtAfter: &todayTime,
		CleanUp:        aws.Bool(false),
	})
	if bizErr != nil {
		logger.Error("Retrieve import job failed", zap.Error(bizErr))
		return
	}
	importTasks := make([]*entity.ImportTask, len(imports))
	for i := range imports {
		importTasks[i] = &entity.ImportTask{
			Import: imports[i],
		}
	}
	var err error
	if importTasks, err = c.importHelper.BatchLoadImportTenantInfo(ctx, importTasks); err != nil {
		logger.Error("Batch load import tenant info error", zap.Error(err))
	}

	importMap := make(map[uint64]*importObject)
	for _, im := range importTasks {
		if im.Org.Kind != string(constant.TenantKindExternal) {
			continue
		}
		clusterID, err := strconv.ParseUint(im.ClusterID, 10, 64)
		if err != nil {
			logger.Error("Parse cluster id failed", zap.Error(err))
			continue
		}
		if _, exist := importMap[clusterID]; !exist {
			importMap[clusterID] = &importObject{
				clusterName:  im.Cluster.ClusterName,
				tenantID:     entity.ID(im.Cluster.OrganizationId),
				tenantName:   im.Org.Name,
				succNum:      0,
				failNum:      0,
				tenantPlan:   im.Org.Plan,
				providerType: entity.ImportProviderType(im.Cluster),
			}
		}
		if im.CompletedAt != nil {
			usedTime = im.CompletedAt.Sub(im.CreatedAt)
		} else {
			usedTime = time.Since(im.CreatedAt)
		}
		if usedTime > ImportTimeOut {
			importMap[clusterID].failNum++
			continue
		}
		switch im.Status {
		case constant.ImportStatusCompleted:
			importMap[clusterID].succNum++
		case constant.ImportStatusFailed:
			importMap[clusterID].failNum++
		}
	}

	for id, ob := range importMap {
		ch <- prometheus.MustNewConstMetric(
			ImportStatusCount,
			prometheus.CounterValue,
			ob.succNum,
			fmt.Sprintf("%d", ob.tenantID),
			ob.tenantName,
			fmt.Sprintf("%d", id),
			ob.clusterName,
			ob.providerType,
			"ok",
			ob.tenantPlan,
		)

		ch <- prometheus.MustNewConstMetric(
			ImportStatusCount,
			prometheus.CounterValue,
			ob.failNum,
			fmt.Sprintf("%d", ob.tenantID),
			ob.tenantName,
			fmt.Sprintf("%d", id),
			ob.clusterName,
			ob.providerType,
			"fail",
			ob.tenantPlan,
		)
	}
}