package workflow

import (
	"context"
	"encoding/json"
	"time"

	"github.com/prometheus/client_golang/prometheus"
	"github.com/samber/lo"
	"github.com/tidbcloud/fastflow/pkg/entity"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.opentelemetry.io/otel"
	"go.uber.org/zap"
)

const (
	NS = "workflow"
)

var (
	WorkflowInfo = prometheus.NewDesc(
		prometheus.BuildFQName(NS, "", "info"),
		"Information about lifecycle of a cluster.",
		[]string{
			"dag_ins_id",
			"biz_id",
			"tenant_id",
			"project_id",
			"workflow_dag", // dag_id of dag_instances table
			"component_type",
			"operation_type",
		}, nil,
	)
)

var (
	Tracer = otel.Tracer("metrics")
)

type collector struct {
	wf Workflow
}

func newWorkflowCollector(wf Workflow) *collector {
	return &collector{
		wf: wf,
	}
}

func (c *collector) Describe(ch chan<- *prometheus.Desc) {
	ch <- WorkflowInfo
}

func (c *collector) Collect(ch chan<- prometheus.Metric) {
	ctx, span := Tracer.Start(context.Background(), "ScaleTypeCollector.Collect")
	defer span.End()

	dagInstances, err := c.wf.ListAllRunningDagInstances()
	if err != nil {
		tilog.Error(ctx, "list running dag instances failed", zap.Error(err))
		return
	}

	lo.ForEach(dagInstances, func(dagIns *entity.DagInstance, index int) {
		// Associated cluster not found, could be project scope dag instances.
		// Record metric with dag instance tags.

		if s, ok := dagIns.Vars[KeyScheduledTime]; ok {
			scheduledTime, err := time.Parse(time.RFC3339, s.Value)
			if err != nil {
				tilog.Error(ctx, "parse scheduled time failed", zap.Error(err))
			} else if time.Now().Before(scheduledTime) { // if waiting for scheduled time, skip record metric
				return
			}
			// if after scheduled time, set dag instance time to scheduled time
			// set dag instance time to scheduled time
			dagIns.CreatedAt = scheduledTime.Unix()
		}

		var tenantId, projectId, bizId string
		prjIdTag, ok := lo.Find(dagIns.Tags, func(tag entity.DagInstanceTag) bool {
			return tag.Key == ProjectKey
		})
		if ok {
			projectId = prjIdTag.Value
		}
		orgIdTag, ok := lo.Find(dagIns.Tags, func(tag entity.DagInstanceTag) bool {
			return tag.Key == OrgIdKey
		})
		if ok {
			tenantId = orgIdTag.Value
		}

		bizIdTag, ok := lo.Find(dagIns.Tags, func(tag entity.DagInstanceTag) bool {
			return tag.Key == BizIdKey
		})
		if ok {
			bizId = bizIdTag.Value
		}

		// Fallback use empty component_type & operation_type
		sendMetric := func(componentType, operationType string) {
			ch <- prometheus.MustNewConstMetric(WorkflowInfo, prometheus.GaugeValue, float64(time.Now().Unix()-dagIns.CreatedAt),
				dagIns.ID,
				bizId,
				tenantId,
				projectId,
				dagIns.DagID,
				componentType,
				operationType,
			)
		}
		if v, ok := dagIns.Vars[KeyComponentOperations]; ok {
			var componentOperations map[string][]string
			if err := json.Unmarshal([]byte(v.Value), &componentOperations); err != nil {
				tilog.Error(ctx, "parse component operations failed", zap.Error(err))
				sendMetric("", "")
			} else {
				if len(componentOperations) > 0 {
					for component, operations := range componentOperations {
						for _, operation := range operations {
							sendMetric(component, operation)
						}
					}
				} else {
					sendMetric("", "")
				}
			}
		} else {
			sendMetric("", "")
		}
	})
}