package pimport

import (
	"context"
	"fmt"
	"time"

	"github.com/pingcap/log"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/tidbcloud/dataflow-service/internal/db"
	"github.com/tidbcloud/dataflow-service/internal/db/dbutil"
	"github.com/tidbcloud/dataflow-service/internal/db/entity"
	"github.com/tidbcloud/dataflow-service/internal/metrics"
	pimport "github.com/tidbcloud/dataflow-service/internal/model/import"
	"github.com/tidbcloud/dataflow-service/pkg/account/accounttypes"
	"go.uber.org/zap"
)

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

const ImportTimeOut = 8 * time.Hour

type StatusCollector struct{}

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

func NewStatusCollector() prometheus.Collector {
	return &StatusCollector{}
}

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

func (c *StatusCollector) Collect(ch chan<- prometheus.Metric) {
	var usedTime time.Duration

	todayTime := time.Now().Truncate(24 * time.Hour)
	imports, err := dbutil.ListActiveImportsCreatedAfter(db.GetDB(), todayTime)
	if err != nil {
		log.L().Error("Retrieve import job failed", zap.Error(err))
		return
	}
	importTasks := make([]*pimport.ImportTask, len(imports))
	for i, imp := range imports {
		importTasks[i] = &pimport.ImportTask{
			Import: imp,
		}
	}
	if importTasks, err = pimport.BatchLoadImportTenantInfo(context.TODO(), importTasks); err != nil {
		log.L().Error("Batch load import tenant info error", zap.Error(err))
	}

	importMap := make(map[entity.ID]*importObject)
	for _, im := range importTasks {
		if im.Tenant.Kind != accounttypes.TenantKindExternal {
			continue
		}
		if _, exist := importMap[im.ClusterID]; !exist {
			importMap[im.ClusterID] = &importObject{
				clusterName: im.Cluster.Name,
				tenantID:    entity.ID(im.Cluster.GetTenantId()),
				tenantName:  im.Tenant.Name,
				devTier:     im.Cluster.GetIsDevTier(),
				succNum:     0,
				failNum:     0,
				tenantPlan:  string(im.Tenant.Plan),
			}
		}
		if im.FinishedAt != nil {
			usedTime = im.FinishedAt.Sub(im.CreatedAt)
		} else {
			usedTime = time.Since(im.CreatedAt)
		}
		if usedTime > ImportTimeOut {
			importMap[im.ClusterID].failNum++
			continue
		}
		if im.Status == entity.ImportStatusCompleted {
			importMap[im.ClusterID].succNum++
		} else if im.Status == entity.ImportStatusFailed {
			importMap[im.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,
			metrics.Dedicate,
			"ok",
			ob.tenantPlan,
		)

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