package export

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

	"github.com/pingcap/errors"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/push"
	"github.com/tidbcloud/aws-shared-provider/pkg/types"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	batchv1 "k8s.io/api/batch/v1"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

// Export metrics

const (
	PushExportSizeMetricsTimeout = 100 * time.Second
	PrometheusPushGatewayURL     = "serverless-metering.serverless-metering:9091"
)

func (r *reconcileContext) reconcileExport() error {
	r.Info("reconcile export")

	if r.export.Annotations[types.AnnoExportJobStatus] == "" {
		r.Info("start watch job status")
		if err := r.watchJobStatus(); err != nil {
			return errors.Annotate(err, "failed to ensure export job status")
		}
		return nil
	}

	switch r.export.Annotations[types.AnnoExportJobStatus] {
	case string(cloudv1.ExportPhaseCompleted):
		_, err := r.updateProgress(cloudv1.ExportPhaseCompleted, "export success")
		if err != nil {
			r.Error(err, "update status failed")
			return err
		}
		return nil
	case string(cloudv1.ExportPhaseFailed):
		message := r.export.Annotations[types.AnnoExportJobFailedMessage]
		if message == "" {
			message = unknownErrorMessage
		}
		_, err := r.updateProgress(cloudv1.ExportPhaseFailed, message)
		if err != nil {
			r.Error(err, "update status failed")
			return err
		}
		return nil
	default:
		return errors.New("invalid export job status")
	}
}

func (r *reconcileContext) watchJobStatus() error {
	exportJob := &batchv1.Job{}
	if err := r.Client.Get(r.ctx, client.ObjectKey{
		Name:      r.BuildDumplingJobName(),
		Namespace: r.ExportJobNS,
	}, exportJob); err != nil {
		r.Error(err, "get export job failed")
		return err
	}
	for _, cond := range exportJob.Status.Conditions {
		if cond.Type == batchv1.JobComplete && cond.Status == corev1.ConditionTrue {
			r.export.Annotations[types.AnnoExportJobStatus] = string(cloudv1.ExportPhaseCompleted)
			return r.InfraClient.Update(r.ctx, r.export)
		} else if cond.Type == batchv1.JobFailed && cond.Status == corev1.ConditionTrue {
			r.export.Annotations[types.AnnoExportJobStatus] = string(cloudv1.ExportPhaseFailed)
			message, err := r.getPodFailedMessage()
			if err != nil {
				r.Error(err, "failed to get failed message")
				return err
			}
			r.export.Annotations[types.AnnoExportJobFailedMessage] = message
			return r.InfraClient.Update(r.ctx, r.export)
		}
	}
	return nil
}

func (r *reconcileContext) getPodFailedMessage() (string, error) {
	exportPod, err := r.getJobPod()
	if err != nil {
		r.Error(err, "get export pod failed")
		return unknownErrorMessage, err
	}
	containerStatuses := exportPod.Status.ContainerStatuses
	var message string
	for _, containerStatus := range containerStatuses {
		if containerStatus.Name == DumplingContainerName {
			if containerStatus.State.Terminated != nil {
				message = containerStatus.State.Terminated.Message
			}
			break
		}
	}
	r.Info("get export pod message", "podName", exportPod.Name, "message", message)
	return r.generateExportFailedMessage(message), nil
}

func (r *reconcileContext) generateExportFailedMessage(message string) string {
	if message == "" {
		return unknownErrorMessage
	}
	// get dump failed message
	messages := strings.Split(message, DumpFailMessageTemplate)
	//nolint:mnd // ignore magic number
	if len(messages) < 2 {
		r.Info("failed message is not contain dump failed message", "message", message)
		return unknownErrorMessage
	}
	dumpFailMessage := strings.Trim(messages[len(messages)-1], "\n ")

	r.Info("get export dump fail message", "dump_fail_message", dumpFailMessage)

	// SQL dump failed
	if strings.HasPrefix(dumpFailMessage, SQLFailMessageTemplate) {
		if r.export.Spec.Options.SQL != "" {
			return fmt.Sprintf("%s %s.", ExecuteSQLErrorMessage, dumpFailMessage)
		}
		if r.export.Spec.Options.Where != "" {
			return fmt.Sprintf("%s %s.", ExecuteSQLErrorMessage, dumpFailMessage)
		}
		r.Info("execute sql failed without sql or where filter")
		return unknownErrorMessage
	}
	// Partition failed
	if strings.HasPrefix(dumpFailMessage, PartitionFailMessageTemplate) {
		if len(r.export.Spec.Options.Partition) > 0 {
			return fmt.Sprintf("%s %s.", PartitionErrorMessage, dumpFailMessage)
		}
		r.Info("partition failed without partition flag")
		return unknownErrorMessage
	}
	// AccessDeny
	if strings.HasPrefix(dumpFailMessage, AccessDeniedMessageTemplate) {
		switch r.export.Spec.Target.ExportType {
		case cloudv1.ExportTypeLOCAL:
			return unknownErrorMessage
		case cloudv1.ExportTypeS3:
			if r.export.Spec.Target.S3.RoleArn != "" {
				return S3RoleArnAccessDeniedMessage
			}
			if r.export.Spec.Target.S3.AccessKey != "" {
				if r.export.Spec.Target.S3.Provider == "alibaba" {
					return OSSAccessKeyAccessDeniedMessage
				} else {
					return S3AccessKeyAccessDeniedMessage
				}
			}
		case cloudv1.ExportTypeGCS:
			return GCSAccessKeyAccessDeniedMessage
		case cloudv1.ExportTypeAzure:
			return AzureAccessKeyAccessDeniedMessage
		}
	}
	return unknownErrorMessage
}

func (r *reconcileContext) getJobPod() (*corev1.Pod, error) {
	ns := r.ExportJobNS
	jobName := r.BuildDumplingJobName()
	pods := &corev1.PodList{}
	if err := r.Client.List(r.ctx, pods,
		client.InNamespace(ns),
		client.MatchingLabels{"job-name": jobName}); err != nil {
		return nil, err
	}

	if len(pods.Items) == 0 {
		return nil, apierrors.NewNotFound(schema.GroupResource{Resource: "pods"}, jobName)
	}

	for _, p := range pods.Items {
		if p.Status.Phase == corev1.PodRunning || p.Status.Phase == corev1.PodSucceeded {
			return &p, nil
		}
	}

	return &pods.Items[0], nil
}

func (r *reconcileContext) sendToMetering(finishedSizeStr, compressedSizeStr string) error {
	r.Info("Sending export size", "finishedSize", finishedSizeStr, "compressedSize", compressedSizeStr)
	finishedSize, err := strconv.ParseFloat(finishedSizeStr, 64)
	if err != nil {
		return fmt.Errorf("failed to parse finished size: %w", err)
	}
	compressedSize, err := strconv.ParseFloat(compressedSizeStr, 64)
	if err != nil {
		return fmt.Errorf("failed to parse compressed size: %w", err)
	}
	pushCtx, cancel := context.WithTimeout(r.ctx, PushExportSizeMetricsTimeout)
	defer cancel()

	collector := r.createExportSizeCollector(finishedSize, compressedSize)

	err = push.New(PrometheusPushGatewayURL, "export_size").
		Collector(collector).
		PushContext(pushCtx)
	if err != nil {
		r.Error(err, "failed to push export size metrics")
		return err
	}

	return nil
}

func (r *reconcileContext) createExportSizeCollector(finishedSize, compressedSize float64) prometheus.Collector {
	finishedSizeGauge := prometheus.NewGaugeVec(
		prometheus.GaugeOpts{
			Name: "export_finished_size_bytes",
			Help: "The finished size of the export in bytes",
		},
		[]string{"cluster_id", "export_id", "export_type", "provider", "region", "target_provider", "target_region"},
	)

	compressedSizeGauge := prometheus.NewGaugeVec(
		prometheus.GaugeOpts{
			Name: "export_compressed_size_bytes",
			Help: "The compressed size of the export in bytes",
		},
		[]string{"cluster_id", "export_id", "export_type", "provider", "region", "target_provider", "target_region"},
	)

	exportType := r.export.Annotations[types.AnnoExportType]
	provider := r.export.Labels[types.LabelKeyProviderType]
	provider = strings.TrimSuffix(provider, "-free-tier")

	region := r.export.Labels[types.LabelKeyRegion]
	targetProvider := getTargetProvider(r.export.Spec.Target)
	targetRegion := r.export.Spec.Target.Region

	finishedSizeGauge.WithLabelValues(r.clusterID, r.export.Name, exportType, provider, region, targetProvider, targetRegion).Set(finishedSize)
	compressedSizeGauge.WithLabelValues(r.clusterID, r.export.Name, exportType, provider, region, targetProvider, targetRegion).Set(compressedSize)

	registry := prometheus.NewRegistry()
	registry.MustRegister(finishedSizeGauge)
	registry.MustRegister(compressedSizeGauge)

	return registry
}

func getTargetProvider(target cloudv1.ExportTarget) string {
	switch target.ExportType {
	case cloudv1.ExportTypeS3:
		if target.S3.Provider == "alibaba" {
			return "alicloud"
		} else {
			return "aws"
		}
	case cloudv1.ExportTypeGCS:
		return "google"
	case cloudv1.ExportTypeAzure:
		return "azure"
	case cloudv1.ExportTypeLOCAL:
		return "local"
	default:
		return "local"
	}
}