package metrics

import (
	"errors"
	"net/http"
	"net/http/pprof"
	"time"

	"github.com/pingcap/log"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/promhttp"
	"github.com/tidbcloud/observability/cmd/diagnosis-compute/pkg/types"
	"go.uber.org/zap"
)

const (
	labelEnv       = "env"
	labelTenantID  = "tenant_id"
	labelCluster   = "cluster"
	labelComponent = "component"
	labelClusterID = "cluster_id"
	labelStatus    = "status"
)

func logLabels(info types.ClusterInfo, component string) prometheus.Labels {
	return prometheus.Labels{
		labelEnv:       info.Env,
		labelTenantID:  info.TenantID,
		labelCluster:   info.ClusterID,
		labelComponent: component,
	}
}

var (
	// Prometheus metrics
	FilesProcessed = prometheus.NewCounterVec(
		prometheus.CounterOpts{
			Name: "diagnosis_compute_files_processed_total",
			Help: "Total number of files processed by cluster and status",
		},
		[]string{"cluster_id", "status"},
	)

	UniqueLogPatterns = prometheus.NewGaugeVec(
		prometheus.GaugeOpts{
			Name: "diagnosis_compute_unique_log_patterns",
			Help: "Number of unique log patterns by cluster and component",
		},
		[]string{"cluster_id", "component"},
	)

	FileProcessingDuration = prometheus.NewHistogramVec(
		prometheus.HistogramOpts{
			Name:    "diagnosis_compute_file_processing_duration_seconds",
			Help:    "Time taken to process log files in seconds",
			Buckets: []float64{0.1, 0.5, 1, 5, 10, 30, 60, 120, 300},
		},
		[]string{"cluster_id", "component"},
	)

	CheckpointTs = prometheus.NewGaugeVec(
		prometheus.GaugeOpts{
			Name: "diagnosis_compute_checkpoint_ts",
			Help: "The checkpoint ts for clusters",
		},
		[]string{"cluster_id"},
	)

	UploadMergedFilesToS3Bytes = prometheus.NewCounterVec(
		prometheus.CounterOpts{
			Name: "diagnosis_compute_upload_merged_files_to_s3_bytes_total",
			Help: "Total number of bytes uploaded for merged files to S3 by cluster",
		},
		[]string{"cluster_id", "components"},
	)

	UploadMergedFilesToS3Duration = prometheus.NewHistogramVec(
		prometheus.HistogramOpts{
			Name:    "diagnosis_compute_upload_merged_files_to_s3_duration_seconds",
			Help:    "Time taken to upload merged files to S3 in seconds",
			Buckets: []float64{0.1, 0.5, 1, 5, 10, 30, 60, 120, 300},
		},
		[]string{"cluster_id", "components"},
	)

	RawLogFiles = prometheus.NewCounterVec(
		prometheus.CounterOpts{
			Name: "diagnosis_compute_raw_log_files_total",
			Help: "Total number of raw S3 log objects processed",
		},
		[]string{labelEnv, labelTenantID, labelCluster, labelComponent},
	)

	RawLogBytes = prometheus.NewCounterVec(
		prometheus.CounterOpts{
			Name: "diagnosis_compute_raw_log_bytes_total",
			Help: "Total compressed bytes of raw S3 log objects processed",
		},
		[]string{labelEnv, labelTenantID, labelCluster, labelComponent},
	)

	ParquetBytes = prometheus.NewCounterVec(
		prometheus.CounterOpts{
			Name: "diagnosis_compute_parquet_bytes_total",
			Help: "Total bytes uploaded for parquet files",
		},
		[]string{labelEnv, labelTenantID, labelCluster, labelComponent},
	)

	ParquetFiles = prometheus.NewCounterVec(
		prometheus.CounterOpts{
			Name: "diagnosis_compute_parquet_files_total",
			Help: "Total number of parquet files uploaded",
		},
		[]string{labelEnv, labelTenantID, labelCluster, labelComponent},
	)

	UploadParquetBytes = prometheus.NewCounterVec(
		prometheus.CounterOpts{
			Name: "diagnosis_compute_upload_parquet_bytes_total",
			Help: "Total bytes uploaded for parquet files by storage cluster id (legacy)",
		},
		[]string{labelClusterID, labelComponent},
	)

	UploadParquetDuration = prometheus.NewHistogramVec(
		prometheus.HistogramOpts{
			Name:    "diagnosis_compute_upload_parquet_duration_seconds",
			Help:    "Time taken to upload parquet files in seconds",
			Buckets: []float64{0.1, 0.5, 1, 5, 10, 30, 60, 120, 300},
		},
		[]string{labelEnv, labelTenantID, labelCluster, labelComponent},
	)

	CurrentAnalyzeTasks = prometheus.NewGaugeVec(
		prometheus.GaugeOpts{
			Name: "diagnosis_compute_current_analyze_tasks",
			Help: "Current analyze tasks",
		},
		[]string{"cluster_id"},
	)

	CurrentMergeTasks = prometheus.NewGaugeVec(
		prometheus.GaugeOpts{
			Name: "diagnosis_compute_current_merge_tasks",
			Help: "Current merge tasks",
		},
		[]string{"cluster_id"},
	)

	ConprofCleanupFiles = prometheus.NewCounterVec(
		prometheus.CounterOpts{
			Name: "diagnosis_compute_conprof_cleanup_files_total",
			Help: "Total conprof profile files deleted by retention cleanup",
		},
		[]string{labelEnv, labelTenantID, labelCluster, "phase", labelStatus},
	)

	ConprofCleanupBytes = prometheus.NewCounterVec(
		prometheus.CounterOpts{
			Name: "diagnosis_compute_conprof_cleanup_bytes_total",
			Help: "Total bytes of conprof profile files deleted by retention cleanup",
		},
		[]string{labelEnv, labelTenantID, labelCluster, "phase"},
	)
)

func InitMetrics() {
	prometheus.MustRegister(FilesProcessed)
	prometheus.MustRegister(FileProcessingDuration)
	prometheus.MustRegister(CheckpointTs)
	prometheus.MustRegister(UniqueLogPatterns)
	prometheus.MustRegister(UploadMergedFilesToS3Bytes)
	prometheus.MustRegister(UploadMergedFilesToS3Duration)
	prometheus.MustRegister(RawLogFiles)
	prometheus.MustRegister(RawLogBytes)
	prometheus.MustRegister(ParquetBytes)
	prometheus.MustRegister(ParquetFiles)
	prometheus.MustRegister(UploadParquetBytes)
	prometheus.MustRegister(UploadParquetDuration)
	prometheus.MustRegister(CurrentAnalyzeTasks)
	prometheus.MustRegister(CurrentMergeTasks)
	prometheus.MustRegister(ConprofCleanupFiles)
	prometheus.MustRegister(ConprofCleanupBytes)

	mux := http.NewServeMux()

	// prometheus metrics
	mux.Handle("/metrics", promhttp.Handler())

	// pprof debug endpoints
	mux.HandleFunc("/debug/pprof/", pprof.Index)
	mux.HandleFunc("/debug/pprof/cmdline", pprof.Cmdline)
	mux.HandleFunc("/debug/pprof/profile", pprof.Profile)
	mux.HandleFunc("/debug/pprof/symbol", pprof.Symbol)
	mux.HandleFunc("/debug/pprof/trace", pprof.Trace)

	srv := &http.Server{
		Addr:         ":8000",
		Handler:      mux,
		ReadTimeout:  5 * time.Second,
		WriteTimeout: 10 * time.Second,
		IdleTimeout:  15 * time.Second,
	}
	go func() {
		if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
			log.Error("Failed to start Prometheus metrics server", zap.Error(err))
		}
	}()
}

func RecordFileProcessed(clusterID string, status string) {
	FilesProcessed.WithLabelValues(clusterID, status).Inc()
}

func RecordProcessingLag(clusterID string, lagSeconds float64) {
	CheckpointTs.WithLabelValues(clusterID).Set(lagSeconds)
}

func RecordUploadMergedFilesToS3(clusterID string, components string, bytes int64, durationSeconds float64) {
	UploadMergedFilesToS3Bytes.WithLabelValues(clusterID, components).Add(float64(bytes))
	UploadMergedFilesToS3Duration.WithLabelValues(clusterID, components).Observe(durationSeconds)
}

func RecordRawLogProcessed(info types.ClusterInfo, component string, bytes int64) {
	labels := logLabels(info, component)
	RawLogFiles.With(labels).Inc()
	if bytes > 0 {
		RawLogBytes.With(labels).Add(float64(bytes))
	}
}

func RecordParquetUpload(info types.ClusterInfo, component string, bytes int64, durationSeconds float64) {
	labels := logLabels(info, component)
	ParquetBytes.With(labels).Add(float64(bytes))
	ParquetFiles.With(labels).Inc()
	UploadParquetBytes.WithLabelValues(info.StorageID, component).Add(float64(bytes))
	UploadParquetDuration.With(labels).Observe(durationSeconds)
}

func RecordConprofCleanupDeleted(info types.ClusterInfo, phase string, count int, bytes int64) {
	if count <= 0 {
		return
	}
	ConprofCleanupFiles.With(prometheus.Labels{
		labelEnv:      info.Env,
		labelTenantID: info.TenantID,
		labelCluster:  info.ClusterID,
		"phase":       phase,
		labelStatus:   "deleted",
	}).Add(float64(count))
	RecordConprofCleanupBytes(info, phase, bytes)
}

func RecordConprofCleanupError(info types.ClusterInfo, phase string) {
	ConprofCleanupFiles.With(prometheus.Labels{
		labelEnv:      info.Env,
		labelTenantID: info.TenantID,
		labelCluster:  info.ClusterID,
		"phase":       phase,
		labelStatus:   "error",
	}).Inc()
}

func RecordConprofCleanupBytes(info types.ClusterInfo, phase string, bytes int64) {
	if bytes <= 0 {
		return
	}
	ConprofCleanupBytes.With(prometheus.Labels{
		labelEnv:      info.Env,
		labelTenantID: info.TenantID,
		labelCluster:  info.ClusterID,
		"phase":       phase,
	}).Add(float64(bytes))
}

func DeleteMetrics(info types.ClusterInfo) {
	FilesProcessed.DeletePartialMatch(prometheus.Labels{labelClusterID: info.StorageID})
	UniqueLogPatterns.DeletePartialMatch(prometheus.Labels{labelClusterID: info.StorageID})
	FileProcessingDuration.DeletePartialMatch(prometheus.Labels{labelClusterID: info.StorageID})
	CheckpointTs.DeleteLabelValues(info.StorageID)
	UploadMergedFilesToS3Bytes.DeletePartialMatch(prometheus.Labels{labelClusterID: info.StorageID})
	UploadMergedFilesToS3Duration.DeletePartialMatch(prometheus.Labels{labelClusterID: info.StorageID})
	UploadParquetBytes.DeletePartialMatch(prometheus.Labels{labelClusterID: info.StorageID})

	clusterLabels := prometheus.Labels{labelTenantID: info.TenantID, labelCluster: info.ClusterID}
	RawLogFiles.DeletePartialMatch(clusterLabels)
	RawLogBytes.DeletePartialMatch(clusterLabels)
	ParquetBytes.DeletePartialMatch(clusterLabels)
	ParquetFiles.DeletePartialMatch(clusterLabels)
	UploadParquetDuration.DeletePartialMatch(clusterLabels)

	ConprofCleanupFiles.DeletePartialMatch(clusterLabels)
	ConprofCleanupBytes.DeletePartialMatch(clusterLabels)

	CurrentAnalyzeTasks.DeleteLabelValues(info.StorageID)
	CurrentMergeTasks.DeleteLabelValues(info.StorageID)
}