package backup

import (
	"context"
	"fmt"
	"time"

	"github.com/elliotchance/pie/v2"
	"github.com/prometheus/client_golang/prometheus"
	"go.uber.org/zap"
	"gorm.io/gorm"

	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/dataflow-service/internal/db"
	"github.com/tidbcloud/dataflow-service/internal/grpc/grpcli"
	"github.com/tidbcloud/dataflow-service/internal/metrics"
	"github.com/tidbcloud/dataflow-service/internal/model/util"
	"github.com/tidbcloud/dataflow-service/pkg/log"
)

var logBackupLagDurationSeconds = prometheus.NewDesc(
	prometheus.BuildFQName(metrics.DataflowNamespace, "log_backup", "lag_duration_seconds"),
	"The lag duration of log backup job",
	append(metrics.CommonIDLabels, "backup_id", "backup_name", "create_at"),
	nil)

type LogBackupLagMetricsCollector struct{}

func NewLogBackupLagMetricsCollector() prometheus.Collector {
	return &LogBackupLagMetricsCollector{}
}

func (c *LogBackupLagMetricsCollector) Describe(ch chan<- *prometheus.Desc) {
	ch <- logBackupLagDurationSeconds
}

func (c *LogBackupLagMetricsCollector) Collect(ch chan<- prometheus.Metric) {
	logger := log.GetLogger(context.Background())
	start := time.Now()
	defer func() {
		logger.Info("log backup lag metrics collect time", zap.Int64("time", time.Since(start).Milliseconds()))
	}()

	backups, err := c.queryLogBackups()
	if err != nil {
		logger.Error("Retrieve log backup failed", zap.Error(err))
		return
	}
	if len(backups) == 0 {
		return
	}

	ctx := context.Background()
	for _, br := range backups {

		cluster, err := grpcli.GetClusterByID(ctx, uint64(br.ClusterID))
		if err != nil {
			logger.Error("failed to get cluster by id",
				zap.Uint64("backupID", uint64(br.ID)),
				zap.Uint64("clusterID", uint64(br.ClusterID)),
				zap.Error(err))
			continue
		}
		if cluster.Status != core.ClusterStatus_CLUSTER_STATUS_SCALING && cluster.Status != core.ClusterStatus_CLUSTER_STATUS_NORMAL {
			logger.Info("cluster is not in normal or scaling status",
				zap.Uint64("backupID", uint64(br.ID)),
				zap.Uint64("clusterID", uint64(br.ClusterID)),
				zap.Int32("clusterStatus", int32(cluster.Status)))
			continue
		}

		timeTaken := c.calculateTimeTaken(&br)
		if timeTaken == nil {
			continue
		}

		ch <- prometheus.MustNewConstMetric(
			logBackupLagDurationSeconds,
			prometheus.GaugeValue,
			*timeTaken,
			fmt.Sprintf("%d", br.TenantID),
			fmt.Sprintf("%d", br.ClusterID),
			fmt.Sprintf("%d", br.ID),
			br.Name,
			br.CreatedAt.String(),
		)
	}
}

func (c *LogBackupLagMetricsCollector) calculateTimeTaken(br *entity.LogBackup) *float64 {
	logger := log.GetLogger(context.Background())
	finishedAt := time.Now().Unix()
	checkpointTS, err := util.ParseTSFromTSString(br.CheckpointTS)
	if err != nil {
		logger.Error("failed to parse log-backup checkpointTS",
			zap.Uint64("backupID", uint64(br.ID)),
			zap.Uint64("clusterID", uint64(br.ClusterID)),
			zap.String("checkpointTS", br.CheckpointTS),
			zap.Error(err))
		return nil
	}
	timeTaken := float64(finishedAt - checkpointTS)
	return &timeTaken
}

func (c *LogBackupLagMetricsCollector) queryLogBackups() ([]entity.LogBackup, error) {
	var backups []entity.LogBackup
	err := db.Transaction(func(tx *gorm.DB) error {
		return entity.NewLogBackupQuerySet(tx).
			DeletedAtIsNull().
			StatusIn(entity.LogBackupStatusRunning).
			PreloadBackupSchedule().
			All(&backups)
	})
	if err != nil {
		return nil, err
	}

	if len(backups) == 0 {
		return backups, nil
	}

	clearBackups := pie.Of(backups).Filter(func(backup entity.LogBackup) bool {
		return backup.BackupSchedule != nil
	}).Result
	return clearBackups, nil
}