package worker

import (
	"context"
	"fmt"
	"os"
	"runtime"
	"strings"
	"sync"
	"time"

	"github.com/go-co-op/gocron"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.opentelemetry.io/otel"
	"go.uber.org/zap"

	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/service/worker/br"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/service/worker/resourcepool"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/pkg/config"
	"github.com/tidbcloud/tidb-management-service/utilities/lease"
)

type Runner struct {
	scheduler *gocron.Scheduler
}

const (
	reportAutoBackupMeteringInterval = time.Hour
	handleIdleResourcePoolsInterval  = 5 * time.Minute
	automaticResumeInterval          = 5 * time.Minute
	expirePauseAdmissionsInterval    = time.Minute
	backfillEssentialV2Interval      = 5 * time.Minute
	lockKeyPrefix                    = "nextgen-global-service:worker"
	lockAttemptDivider               = 10
	minLockAttemptInterval           = time.Second
)

func NewRunner(brWorker br.Worker, rpWorker resourcepool.Worker, cfg *config.Config, locker lease.Locker) *Runner {
	scheduler := gocron.NewScheduler(time.UTC)
	scheduler.SingletonModeAll()

	backupConfig := &cfg.BizConfig.BackupConfig
	// should not make the interval too large, if the interval is too large, the backup can not be synced in time
	if _, err := scheduler.Every(calcLockAttemptInterval(backupConfig.SyncConfig.WorkerInterval)).Do(
		FunWrapper(
			"SyncBackups",
			WithLeaseLock("SyncBackups", backupConfig.SyncConfig.WorkerInterval, locker, brWorker.SyncBackups),
		),
	); err != nil {
		panic(err)
	}

	if _, err := scheduler.Every(calcLockAttemptInterval(backupConfig.ExpireConfig.WorkerInterval)).Do(
		FunWrapper(
			"ExpireBackups",
			WithLeaseLock("ExpireBackups", backupConfig.ExpireConfig.WorkerInterval, locker, brWorker.ExpireBackups),
		),
	); err != nil {
		panic(err)
	}

	if _, err := scheduler.Every(calcLockAttemptInterval(backupConfig.SyncMissingConfig.WorkerInterval)).
		Do(FunWrapper(
			"SyncMissingBackups",
			WithLeaseLock("SyncMissingBackups", backupConfig.SyncMissingConfig.WorkerInterval, locker, brWorker.SyncMissingBackups),
		)); err != nil {
		panic(err)
	}

	if _, err := scheduler.Every(calcLockAttemptInterval(backupConfig.UpdateBackupGcSettingConfig.WorkerInterval)).
		Do(FunWrapper(
			"UpdateBackupGcSetting",
			WithLeaseLock(
				"UpdateBackupGcSetting",
				backupConfig.UpdateBackupGcSettingConfig.WorkerInterval,
				locker,
				brWorker.UpdateBackupGcSetting,
			),
		)); err != nil {
		panic(err)
	}

	if _, err := scheduler.Every(calcLockAttemptInterval(reportAutoBackupMeteringInterval)).Do(FunWrapper(
		"ReportAutoBackupMetering",
		WithLeaseLock(
			"ReportAutoBackupMetering",
			reportAutoBackupMeteringInterval,
			locker,
			brWorker.ReportAutoBackupMetering,
		),
	)); err != nil {
		panic(err)
	}

	// This is a temporary, idempotent backfill. Remove it after all existing
	// Essential V2 clusters have converged to flexible retention.
	if _, err := scheduler.Every(calcLockAttemptInterval(backfillEssentialV2Interval)).Do(FunWrapper(
		"BackfillEssentialV2BackupSettings",
		WithLeaseLock(
			"BackfillEssentialV2BackupSettings",
			backfillEssentialV2Interval,
			locker,
			brWorker.BackfillEssentialV2BackupSettings,
		),
	)); err != nil {
		panic(err)
	}

	if _, err := scheduler.Every(calcLockAttemptInterval(handleIdleResourcePoolsInterval)).Do(FunWrapper(
		"HandleIdleResourcePools",
		WithLeaseLock(
			"HandleIdleResourcePools",
			handleIdleResourcePoolsInterval,
			locker,
			rpWorker.HandleIdleResourcePools,
		),
	)); err != nil {
		panic(err)
	}

	if _, err := scheduler.Every(calcLockAttemptInterval(automaticResumeInterval)).Do(FunWrapper(
		"HandleAutomaticResumeResourcePools",
		WithLeaseLock(
			"HandleAutomaticResumeResourcePools",
			automaticResumeInterval,
			locker,
			rpWorker.HandleAutomaticResumeResourcePools,
		),
	)); err != nil {
		panic(err)
	}

	if _, err := scheduler.Every(calcLockAttemptInterval(expirePauseAdmissionsInterval)).Do(FunWrapper(
		"HandleExpiredPauseAdmissions",
		WithLeaseLock(
			"HandleExpiredPauseAdmissions",
			expirePauseAdmissionsInterval,
			locker,
			rpWorker.HandleExpiredPauseAdmissions,
		),
	)); err != nil {
		panic(err)
	}

	return &Runner{
		scheduler: scheduler,
	}
}

func (w *Runner) StartWorker() {
	tilog.Info(context.Background(), "start worker runner...")
	w.scheduler.StartAsync()
	tilog.Info(context.Background(), "worker runner started")
}

func (w *Runner) StopWorker() {
	tilog.Info(context.Background(), "stop worker runner...")
	w.scheduler.Stop()
	tilog.Info(context.Background(), "worker runner stopped")
}

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

type Fun func() error

type Handler func(ctx context.Context) error

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

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

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

func WithLeaseLock(workerName string, leaseDuration time.Duration, locker lease.Locker, f Handler) Handler {
	lockKey := fmt.Sprintf("%s:%s", lockKeyPrefix, workerName)
	return func(ctx context.Context) error {
		ownerID := genLeaseAttemptOwnerID(workerLeaseOwnerID())
		acquired, err := locker.Acquire(ctx, lockKey, ownerID, leaseDuration)
		if err != nil {
			tilog.Error(ctx, "failed to acquire lease for worker run",
				zap.String("worker", workerName),
				zap.String("lockKey", lockKey),
				zap.String("owner", ownerID),
				zap.Duration("lease_duration", leaseDuration),
				zap.Error(err),
			)
			return err
		}
		if !acquired {
			tilog.Info(ctx, "skip worker run because lease is held by another replica",
				zap.String("worker", workerName),
				zap.String("lockKey", lockKey),
				zap.String("owner", ownerID),
			)
			return nil
		}

		return f(ctx)
	}
}

func calcLockAttemptInterval(leaseDuration time.Duration) time.Duration {
	interval := leaseDuration / lockAttemptDivider
	if interval < minLockAttemptInterval {
		return minLockAttemptInterval
	}
	return interval
}

var (
	workerOwnerIDOnce sync.Once
	workerOwnerID     string
)

func workerLeaseOwnerID() string {
	workerOwnerIDOnce.Do(func() {
		podName := os.Getenv(config.PodNameEnvKey)
		if podName != "" {
			workerOwnerID = podName
			return
		}

		workerOwnerID = fmt.Sprintf("pid-%d", os.Getpid())
	})

	return workerOwnerID
}

func genLeaseAttemptOwnerID(baseOwnerID string) string {
	return fmt.Sprintf("%s:%d", baseOwnerID, time.Now().UnixNano())
}