package pimport

import (
	"fmt"
	"strconv"
	"time"

	"github.com/docker/go-units"
	"github.com/pingcap/errors"
	"github.com/pingcap/log"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/tidbcloud/dataflow-service/internal/config"
	"github.com/tidbcloud/dataflow-service/internal/db/entity"
	"github.com/tidbcloud/dataflow-service/internal/metrics"
	pimport "github.com/tidbcloud/dataflow-service/internal/model/import"
	errhandler "github.com/tidbcloud/dataflow-service/pkg/error-handler"
	"github.com/tidbcloud/pkgv2/logid"
	"go.uber.org/multierr"
	"go.uber.org/zap"
)

const (
	recentFailDuration = time.Minute * 10

	slowTypeOverallDurationLong        = "overall_duration_long"
	slowTypeOverallDurationTooLong     = "overall_duration_too_long"
	slowTypePrepareDurationLong        = "prepare_duration_gt_%s"
	slowTypePrepareDurationTooLong     = "prepare_duration_gt_%s"
	slowTypeEncodeDurationLong         = "encode_speed_lt_%s"
	slowTypeEncodeDurationTooLong      = "encode_speed_lt_%s"
	slowTypeImportDurationLong         = "import_duration_long"
	slowTypeImportDurationTooLong      = "import_duration_too_long"
	slowTypePostProcessDurationLong    = "postproce_duration_long"
	slowTypePostProcessDurationTooLong = "postproce_duration_too_long"
	// the alarm will only be triggered if the running time of each stage
	// exceeds the minimum duration
	importStageMinAlarmDuration = 10 * time.Minute
)

// RegisterMetrics registers the metrics which are ONLY used in DbaaS server.
func RegisterMetrics() {
	prometheus.MustRegister(metrics.ImportThroughputBytes)
	prometheus.MustRegister(newCollector())
	prometheus.MustRegister(NewStatusCollector())
}

var _ prometheus.Collector = &ImportCollector{}

type ImportCollector struct{}

func newCollector() *ImportCollector {
	return &ImportCollector{}
}

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

func (c *ImportCollector) Collect(ch chan<- prometheus.Metric) {
	logger := log.L().With(zap.String("collector", "import"), zap.String("runID", logid.GenLogid()))
	err := c.collect(ch, logger)
	if err != nil {
		logger.Error("failed to collect metrics", zap.Error(err))
	}
}

func (c *ImportCollector) collect(ch chan<- prometheus.Metric, logger *zap.Logger) (err error) {
	start := time.Now()
	defer func() {
		if err != nil {
			logger.Error("failed to collect metrics", zap.Error(err))
		} else {
			logger.Info("imports metrics collect time", zap.Int64("time", time.Since(start).Milliseconds()))
		}
	}()
	imports, err := pimport.ImportsToSync(nil)
	if err != nil {
		return err
	}

	// // more detail: https://pingcap.feishu.cn/wiki/He6MwEbzyiOvSDkO7dncJu3pnUb#share-ZvILd5oOzoJhmrxTIzScTeMwnaf
	// overallSlowSpeed        = 200 * units.GiB
	// overallTooSlowSpeed   = 100 * units.GiB
	// prepareSlowTime           = 10 * time.Minute
	// prepareSlowSlowTime      = 20 * time.Minute
	// encodeSlowSpeed           = 300 * units.MiB
	// encodeSlowSlowSpeed      = 150 * units.MiB
	// importSlowSpeed           = 300 * units.MiB
	// importSlowSlowSpeed      = 150 * units.MiB
	// postProcessDurationFactor = 1.5

	overallSlowSpeed, err1 := units.RAMInBytes(config.Global().ImportConfig.MetricsConfig.OverallSlowSpeed)
	overallTooSlowSpeed, err2 := units.RAMInBytes(config.Global().ImportConfig.MetricsConfig.OverallTooSlowSpeed)
	prepareSlowTime := config.Global().ImportConfig.MetricsConfig.PrepareSlowTime
	prepareTooSlowTime := config.Global().ImportConfig.MetricsConfig.PrepareTooSlowTime
	encodeSlowSpeed, err3 := units.RAMInBytes(config.Global().ImportConfig.MetricsConfig.EncodeSlowSpeed)
	encodeTooSlowSpeed, err4 := units.RAMInBytes(config.Global().ImportConfig.MetricsConfig.EncodeTooSlowSpeed)
	importSlowSpeed, err5 := units.RAMInBytes(config.Global().ImportConfig.MetricsConfig.ImportSlowSpeed)
	importTooSlowSpeed, err6 := units.RAMInBytes(config.Global().ImportConfig.MetricsConfig.ImportTooSlowSpeed)
	postProcessDurationFactor := config.Global().ImportConfig.MetricsConfig.PostProcessDurationFactor

	if err := multierr.Combine(err1, err2, err3, err4, err5, err6); err != nil {
		return errors.Annotate(err, "failed to parse import metrics config")
	}

	for _, imp := range imports {
		dur := time.Since(imp.CreatedAt).Seconds()
		ch <- prometheus.MustNewConstMetric(
			metrics.SyncingImportSinceCreatedSeconds,
			prometheus.GaugeValue,
			dur,
			strconv.FormatUint(imp.Cluster.GetTenantId(), 10),
			strconv.FormatUint(imp.ClusterID.Uint64(), 10),
			strconv.FormatUint(imp.ID.Uint64(), 10),
			metrics.Dedicate,
			imp.Status.String(),
		)
		if imp.ServerTier != "dedicated" {
			continue
		}

		buildMetric := func(slowType string, value float64) prometheus.Metric {
			logger.Info("slow task metrics collect",
				zap.Uint64("import_id", imp.ID.Uint64()),
				zap.String("slow_type", slowType),
				zap.Float64("value", value))
			var tenantPlan, tenantKind string
			if imp.Tenant != nil {
				tenantKind = string(imp.Tenant.Kind)
				tenantPlan = string(imp.Tenant.Plan)
			}
			return prometheus.MustNewConstMetric(
				metrics.ImportSlowTask,
				prometheus.GaugeValue,
				value,
				strconv.FormatUint(imp.TenantID.Uint64(), 10),
				strconv.FormatUint(imp.ClusterID.Uint64(), 10),
				strconv.FormatUint(imp.ID.Uint64(), 10),
				metrics.Dedicate,
				slowType,
				tenantPlan,
				tenantKind,
			)
		}

		// Overall progress is slow
		if imp.TotalSize != nil && dur > prepareTooSlowTime.Seconds() {
			if dur > float64(*imp.TotalSize)/float64(overallTooSlowSpeed)*3600 {
				ch <- buildMetric(slowTypeOverallDurationTooLong, dur)
			} else if dur > float64(*imp.TotalSize)/float64(overallSlowSpeed)*3600 {
				ch <- buildMetric(slowTypeOverallDurationLong, dur)
			}
		}
		if imp.Status == entity.ImportStatusPreparing {
			// Slow task start
			if dur > prepareTooSlowTime.Seconds() {
				ch <- buildMetric(fmt.Sprintf(slowTypePrepareDurationTooLong, prepareTooSlowTime.String()), dur)
			} else if dur > prepareSlowTime.Seconds() {
				ch <- buildMetric(fmt.Sprintf(slowTypePrepareDurationLong, prepareSlowTime.String()), dur)
			}
		} else if imp.Status == entity.ImportStatusImporting {
			// Slow encoding
			if imp.StartedAt != nil && imp.TaskImportingAt == nil &&
				time.Since(*imp.StartedAt) > importStageMinAlarmDuration && imp.ProcessedDataSize != nil {
				encodeSpeed := float64(*imp.ProcessedDataSize) / time.Since(*imp.StartedAt).Seconds()
				if encodeSpeed < float64(encodeTooSlowSpeed) {
					ch <- buildMetric(fmt.Sprintf(slowTypeEncodeDurationTooLong, units.BytesSize(float64(encodeTooSlowSpeed))), encodeSpeed)
				} else if encodeSpeed < float64(encodeSlowSpeed) {
					ch <- buildMetric(fmt.Sprintf(slowTypeEncodeDurationLong, units.BytesSize(float64(encodeSlowSpeed))), encodeSpeed)
				}
			}

			// Slow importing
			if imp.TaskImportingAt != nil && imp.TaskPostproceAt == nil &&
				time.Since(*imp.TaskImportingAt) > importStageMinAlarmDuration {
				dur := time.Since(*imp.TaskImportingAt).Seconds()
				if dur > float64(*imp.TotalSize)/float64(importTooSlowSpeed) {
					ch <- buildMetric(slowTypeImportDurationTooLong, dur)
				} else if dur > float64(*imp.TotalSize)/float64(importSlowSpeed) {
					ch <- buildMetric(slowTypeImportDurationLong, dur)
				}
			}

			// Slow Post-processing
			if imp.TaskPostproceAt != nil && time.Since(*imp.TaskPostproceAt) > importStageMinAlarmDuration {
				dur := time.Since(*imp.TaskPostproceAt).Seconds()
				if dur > float64(*imp.TotalSize)/float64(importTooSlowSpeed)*postProcessDurationFactor {
					ch <- buildMetric(slowTypePostProcessDurationTooLong, dur)
				} else if dur > float64(*imp.TotalSize)/float64(importSlowSpeed)*postProcessDurationFactor {
					ch <- buildMetric(slowTypePostProcessDurationLong, dur)
				}
			}
		}
	}

	imports, err = pimport.FailImportsInRecent(nil /*tenantID*/, recentFailDuration)
	if err != nil {
		return err
	}

	for _, imp := range imports {
		finishedAt := time.Now()
		if imp.FinishedAt != nil {
			finishedAt = *imp.FinishedAt
		}
		dur := finishedAt.Sub(imp.CreatedAt).Seconds()
		var tenantPlan, tenantKind string
		if imp.Tenant != nil {
			tenantKind = string(imp.Tenant.Kind)
			tenantPlan = string(imp.Tenant.Plan)
		}
		pe := errhandler.NewPrettyError(imp.Message, imp.DataFormat)

		ch <- prometheus.MustNewConstMetric(
			metrics.RecentFailImports,
			prometheus.GaugeValue,
			dur,
			strconv.FormatUint(imp.Cluster.GetTenantId(), 10),
			strconv.FormatUint(imp.ClusterID.Uint64(), 10),
			strconv.FormatUint(imp.ID.Uint64(), 10),
			metrics.Dedicate,
			pe.Error(),
			tenantPlan,
			tenantKind,
			strconv.FormatUint(imp.Cluster.GetProjectId(), 10),
			pe.CommonErr.String(),
		)
		logger.Info("recent failed import task metrics collect", zap.Uint64("import_id", imp.ID.Uint64()))
	}

	return nil
}