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/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 logBackupExecutionStatusDurationSeconds = prometheus.NewDesc(
	prometheus.BuildFQName(metrics.DataflowNamespace, "log_backup", "execution_status_duration_seconds"),
	"The time taken and status of log backup job execution",
	append(metrics.CommonIDLabels, "backup_id", "backup_name", "status", "create_at"),
	nil)

type LogBackupMetricsCollector struct{}

func NewLogBackupMetricsCollector() prometheus.Collector {
	return &LogBackupMetricsCollector{}
}

func (c *LogBackupMetricsCollector) Describe(ch chan<- *prometheus.Desc) {
	ch <- logBackupExecutionStatusDurationSeconds
}

func (c *LogBackupMetricsCollector) Collect(ch chan<- prometheus.Metric) {
	logger := log.GetLogger(context.Background())
	start := time.Now()
	defer func() {
		logger.Info("log backup 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
	}

	for _, br := range backups {

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

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

func (c *LogBackupMetricsCollector) calculateTimeTaken(br *entity.LogBackup) *float64 {
	finishedAt := time.Now()
	switch br.Status {
	case entity.LogBackupStatusScheduled:
		if br.ScheduleAt == nil {
			return nil
		}
		timeTaken := finishedAt.Sub(*br.ScheduleAt).Seconds()
		return &timeTaken
	case entity.LogBackupStatusPending:
		if br.PendingAt == nil {
			return nil
		}
		timeTaken := finishedAt.Sub(*br.PendingAt).Seconds()
		return &timeTaken
	case entity.LogBackupStatusStopping:
		if br.StoppingAt == nil {
			return nil
		}
		timeTaken := finishedAt.Sub(*br.StoppingAt).Seconds()
		return &timeTaken
	case entity.LogBackupStatusFailed:
		// do nothing, just use create time as start time, now as finish time
		startAt := br.CreatedAt
		timeTaken := finishedAt.Sub(startAt).Seconds()
		return &timeTaken
	default:
		// other status can be considered as running and no need to monitor normally running log backup
		return nil
	}
}

func (c *LogBackupMetricsCollector) queryLogBackups() ([]entity.LogBackup, error) {
	var backups []entity.LogBackup
	err := db.Transaction(func(tx *gorm.DB) error {
		return entity.NewLogBackupQuerySet(tx).
			DeletedAtIsNull().
			StatusNotIn(entity.LogBackupStatusStopped).
			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
}