package backup

import (
	"context"
	"fmt"
	"strings"
	"time"

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

	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/dataflow-service/internal/db"
	"github.com/tidbcloud/dataflow-service/internal/metrics"
	"github.com/tidbcloud/dataflow-service/pkg/log"
)

var (
	compactBackupExecutionStatusDurationSeconds = prometheus.NewDesc(
		prometheus.BuildFQName(metrics.DataflowNamespace, "compact_backup", "execution_status_duration_seconds"),
		"The time taken and status of compact backup job execution",
		append(metrics.CommonIDLabels, "log_backup_id", "compact_backup_id", "status", "create_at", "downstream_id"),
		nil)

	compactBackupTotal = prometheus.NewDesc(
		prometheus.BuildFQName(metrics.DataflowNamespace, "compact_backup", "total"),
		"Total number of compact backups by status",
		append(metrics.CommonIDLabels, "status"),
		nil)

	compactBackupFailedTotal = prometheus.NewDesc(
		prometheus.BuildFQName(metrics.DataflowNamespace, "compact_backup", "failed_total"),
		"Total number of failed compact backups",
		metrics.CommonIDLabels,
		nil)

	compactBackupInProgress = prometheus.NewDesc(
		prometheus.BuildFQName(metrics.DataflowNamespace, "compact_backup", "in_progress"),
		"Number of compact backups currently in progress",
		metrics.CommonIDLabels,
		nil)
)

type CompactBackupMetricsCollector struct{}

func NewCompactBackupMetricsCollector() prometheus.Collector {
	return &CompactBackupMetricsCollector{}
}

func (c *CompactBackupMetricsCollector) Describe(ch chan<- *prometheus.Desc) {
	ch <- compactBackupExecutionStatusDurationSeconds
	ch <- compactBackupTotal
	ch <- compactBackupFailedTotal
	ch <- compactBackupInProgress
}

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

	compactBackups, err := c.queryCompactBackups()
	if err != nil {
		logger.Error("Retrieve compact backups failed", zap.Error(err))
		return
	}
	if len(compactBackups) == 0 {
		return
	}

	// Aggregate metrics by status
	statusCount := make(map[string]map[string]int) // tenantID_clusterID -> status -> count
	failedCount := make(map[string]int)            // tenantID_clusterID -> count
	inProgressCount := make(map[string]int)        // tenantID_clusterID -> count

	for _, cb := range compactBackups {
		timeTaken := c.calculateTimeTaken(&cb)

		// Handle status with fallback to unknown
		status := metrics.Unknown
		if cb.Status != "" {
			status = string(cb.Status)
		}

		// Get downstream ID or use "none" if not set
		downstreamID := "none"
		if cb.DownstreamCompactID != nil {
			downstreamID = *cb.DownstreamCompactID
		}

		// Get tenant ID from LogBackup if available
		tenantID := "0"
		if cb.LogBackup != nil {
			tenantID = fmt.Sprintf("%d", cb.LogBackup.TenantID)
		}
		clusterID := fmt.Sprintf("%d", cb.ClusterID)
		key := tenantID + "_" + clusterID

		// Update aggregated metrics
		if statusCount[key] == nil {
			statusCount[key] = make(map[string]int)
		}
		statusCount[key][status]++

		if cb.Status == entity.CompactBackupStatusFailed {
			failedCount[key]++
		}

		if cb.Status == entity.CompactBackupStatusRunning || cb.Status == entity.CompactBackupStatusPending {
			inProgressCount[key]++
		}

		ch <- prometheus.MustNewConstMetric(
			compactBackupExecutionStatusDurationSeconds,
			prometheus.GaugeValue,
			timeTaken,
			tenantID,
			clusterID,
			fmt.Sprintf("%d", cb.LogBackupID),
			fmt.Sprintf("%d", cb.ID),
			status,
			cb.CreatedAt.String(),
			downstreamID,
		)
	}

	// Emit aggregated metrics
	for key, statuses := range statusCount {
		parts := strings.Split(key, "_")
		tenantID := parts[0]
		clusterID := parts[1]

		for status, count := range statuses {
			ch <- prometheus.MustNewConstMetric(
				compactBackupTotal,
				prometheus.GaugeValue,
				float64(count),
				tenantID,
				clusterID,
				status,
			)
		}

		// Emit failed count
		if failed, ok := failedCount[key]; ok {
			ch <- prometheus.MustNewConstMetric(
				compactBackupFailedTotal,
				prometheus.GaugeValue,
				float64(failed),
				tenantID,
				clusterID,
			)
		}

		// Emit in-progress count
		if inProgress, ok := inProgressCount[key]; ok {
			ch <- prometheus.MustNewConstMetric(
				compactBackupInProgress,
				prometheus.GaugeValue,
				float64(inProgress),
				tenantID,
				clusterID,
			)
		}
	}
}

func (c *CompactBackupMetricsCollector) queryCompactBackups() ([]entity.CompactBackup, error) {
	var compactBackups []entity.CompactBackup
	err := db.Transaction(func(tx *gorm.DB) error {
		return entity.NewCompactBackupQuerySet(tx).
			PreloadLogBackup().
			DeletedAtIsNull().
			CreatedAtGte(time.Now().Add(-24 * time.Hour)).
			All(&compactBackups)
	})
	if err != nil {
		return nil, err
	}
	return compactBackups, nil
}

func (c *CompactBackupMetricsCollector) calculateTimeTaken(cb *entity.CompactBackup) float64 {
	createAt := cb.CreatedAt

	// For completed tasks, use the finish time
	if cb.Status == entity.CompactBackupStatusSuccess || cb.Status == entity.CompactBackupStatusFailed {
		if cb.FinishedAt != nil {
			return cb.FinishedAt.Sub(createAt).Seconds()
		}
	}

	// For running or pending tasks, calculate from creation to now
	return time.Since(createAt).Seconds()
}