package workflow

import (
	"context"
	"runtime/debug"
	"time"

	"github.com/prometheus/client_golang/prometheus"
	"github.com/tidbcloud/fastflow/pkg/mod"
	"github.com/tidbcloud/pkgv2/db"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
)

const (
	MetricsNamespace       = "workflow"
	watchDogCandidateLimit = 1000
	watchDogRepairReason   = "force failed by workflow watchdog because workflow and all tasks are stale beyond their configured timeout"
)

var (
	MetricsWatchDogRetryDagInsTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
		Namespace: MetricsNamespace,
		Subsystem: "watchdog",
		Name:      "retried_dag_instance_total",
		Help:      "retry dag instance count",
	}, []string{"workflow_dag", "dag_ins_id", "retry_command_error"})

	MetricsWatchDogCleanDagInsTotal = prometheus.NewCounter(prometheus.CounterOpts{
		Namespace: MetricsNamespace,
		Subsystem: "watchdog",
		Name:      "cleaned_dag_instance_total",
		Help:      "watchdog clean count",
	})
)

func init() {
	prometheus.MustRegister(MetricsWatchDogRetryDagInsTotal)
	prometheus.MustRegister(MetricsWatchDogCleanDagInsTotal)
}

type watchDog struct {
	closeCh  chan struct{}
	provider db.TransProvider
	mode     Mode

	dispatcher  *dispatcherWatchDog
	competitive *competitiveWatchDog
}

func newWatchDog(provider db.TransProvider, mode Mode) *watchDog {
	workerMode := mode
	if workerMode == "" {
		workerMode = ModeDispatcher
	}
	wd := &watchDog{
		provider: provider,
		closeCh:  make(chan struct{}),
		mode:     workerMode,
	}
	wd.dispatcher = newDispatcherWatchDog(provider)
	wd.competitive = newCompetitiveWatchDog(provider)
	return wd
}

func (wd *watchDog) start() {
	ctx := context.TODO()
	interval := wd.retryIntervalByMode()
	ticker := time.NewTicker(interval)
	defer ticker.Stop()

	for {
		select {
		case <-wd.closeCh:
			return
		case <-ticker.C:
			if wd.mode == ModeDispatcher && !mod.GetKeeper().IsLeader() {
				tilog.Debug(ctx, "watchdog is not leader, skip retry dag instances and clean complete dag instances")
				continue
			}

			if err := wd.retryDagInstances(); err != nil {
				tilog.PlainError("retry dag instances failed", zap.Error(err))
			}

			if err := wd.cleanCompleteDagInstances(ctx); err != nil {
				tilog.PlainError("clean dag instances failed", zap.Error(err))
			}
		}
	}
}

func (wd *watchDog) retryIntervalByMode() time.Duration {
	if wd.mode == ModeCompetitive {
		return time.Second
	}
	return 5 * time.Second
}

func (wd *watchDog) retryDagInstances() error {
	ctx := context.TODO()
	defer func() {
		if err := recover(); err != nil {
			tilog.PlainError("panic happen", zap.Any("err", err), zap.String("stack", string(debug.Stack())))
		}
	}()

	if wd.mode == ModeCompetitive {
		return wd.competitive.retry(ctx)
	}
	return wd.dispatcher.retry(ctx)
}

func (wd *watchDog) retryDagInstancesWithContext(ctx context.Context) error {
	if wd.mode == ModeCompetitive {
		return wd.competitive.retry(ctx)
	}
	return wd.dispatcher.retry(ctx)
}

func (wd *watchDog) close() {
	close(wd.closeCh)
}