package common

import (
	"context"
	"fmt"
	"runtime"
	"time"

	"github.com/tidbcloud/dataflow-service-ng/internal/metrics"
	"github.com/tidbcloud/dataflow-service-ng/pkg/log"
	"go.opentelemetry.io/otel"
	"go.uber.org/zap"
)

var Tracer = otel.Tracer("dataflow-worker")

type WorkerFun func() error

type Handler func(ctx context.Context) error

func WorkerFunWrapper(workerName string, f Handler) WorkerFun {
	handler := WithRecover(workerName, WithMetrics(workerName, f))
	return func() error {
		ctx := context.Background()
		ctx, span := Tracer.Start(ctx, genWorkerSpanName(workerName))
		defer span.End()
		return handler(ctx)
	}
}

func WithMetrics(workerName string, f Handler) Handler {
	return func(ctx context.Context) error {
		tBegin := time.Now()
		status := metrics.LoopOK
		err := f(ctx)
		if err != nil {
			status = metrics.LoopError
		}
		duration := float64(time.Since(tBegin)) / float64(time.Millisecond)
		metrics.WorkerDuration.WithLabelValues(fmt.Sprintf("%d", 0), workerName, status).Observe(duration)
		return err
	}
}

func WithRecover(workerName string, f Handler) Handler {
	return func(ctx context.Context) error {
		defer func() {
			if rErr := recover(); rErr != nil {
				var stacktrace string
				for i := 2; ; i++ {
					_, f, l, got := runtime.Caller(i)
					if !got {
						break
					}
					stacktrace += fmt.Sprintf("%s:%d\n", f, l)
				}
				log.GetLogger(ctx).Error("get panic when running action",
					zap.Any("worker", workerName),
					zap.Any("stack trace", stacktrace),
					zap.Any("error msg", rErr))
			}
		}()
		return f(ctx)
	}
}

func genWorkerSpanName(workerName string) string {
	return fmt.Sprintf("dataflow-worker.%s", workerName)
}