package app

import (
	"context"
	"fmt"
	"os"
	"strconv"
	"sync"
	"time"

	"github.com/prometheus/client_golang/prometheus"
	"github.com/samber/lo"
	"github.com/tidbcloud/pkgv2/db"
	"github.com/tidbcloud/pkgv2/worker"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"

	globaldedicatedv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/dedicated/v1"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/entity"
	clusterrepo "github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/repository/cluster"
	clustersvc "github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/service/cluster"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/service/metering"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/pkg/config"
	"github.com/tidbcloud/tidb-management-service/utilities/grpc/metadata"
	"github.com/tidbcloud/tidb-management-service/utilities/lease"
	"github.com/tidbcloud/tidb-management-service/utilities/workflow"
)

type GlobalWorker struct {
	workers []*worker.PeriodWorker
}

func (w *GlobalWorker) Start() {
	for _, item := range w.workers {
		item.Start()
	}
}

const (
	NameLblName     = "name"
	TenantIDLblName = "tenant"
	LoopError       = "error"
	LoopOK          = "ok"
)

var workerDuration = prometheus.NewHistogramVec(
	prometheus.HistogramOpts{
		Namespace: metering.MetricsNamespace,
		Subsystem: "worker",
		Name:      "duration_millisecond",
		Help:      "duration of worker loop.",
	}, []string{TenantIDLblName, NameLblName, "status"},
)

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

func NewGlobalBgWorker(
	clusterMeteringWorker *ClusterMeteringWorker,
	reportMetering2billingWorker *ReportMetering2billingWorker,
	dbCleanerWorker *DBCleanerWorker,
	billingLagReporterWorker *BillingLagReporterWorker,
	normalPauseWorker *NormalPauseWorker,
	clusterConfigBackupWorker *ClusterConfigBackupWorker,
) (*GlobalWorker, error) {
	gWorker := &GlobalWorker{}
	const (
		meterClusterInterval        = time.Minute
		report2billingInterval      = time.Minute
		dbCleanerInterval           = 10 * time.Minute
		billingLagReporterInterval  = 10 * time.Minute
		clusterConfigBackupInterval = 10 * time.Minute
	)

	newPeriodWorker := func(workerName string, interval time.Duration, f worker.Handler) *worker.PeriodWorker {
		logger := tilog.With(zap.String("worker", workerName))
		return worker.NewPeriodWorker(logger, interval, withMetrics(0, workerName, f))
	}

	gWorker.workers = []*worker.PeriodWorker{
		newPeriodWorker("meterCluster", meterClusterInterval, clusterMeteringWorker.Run),
		newPeriodWorker("report2billing", report2billingInterval, reportMetering2billingWorker.Run),
		newPeriodWorker("dbCleaner", dbCleanerInterval, dbCleanerWorker.Run),
		newPeriodWorker("billingLagReporter", billingLagReporterInterval, billingLagReporterWorker.Run),
		newPeriodWorker("normalPause", calcWorkerLockAttemptInterval(normalPauseWorkerLeaseTTL), normalPauseWorker.Run),
		newPeriodWorker("clusterConfigBackup", clusterConfigBackupInterval, clusterConfigBackupWorker.Run),
	}

	return gWorker, nil
}

type ClusterMeteringWorker struct {
	recordMeteringSvc *metering.RecordMeteringSvc
	workflow          workflow.Workflow
}

func NewClusterMeteringWorker(
	recordMeteringSvc *metering.RecordMeteringSvc,
	flow workflow.Workflow,
) *ClusterMeteringWorker {
	return &ClusterMeteringWorker{
		recordMeteringSvc: recordMeteringSvc,
		workflow:          flow,
	}
}

func (w *ClusterMeteringWorker) Run(ctx context.Context) error {
	logger := tilog.WithContextS(ctx)
	logger.Info("ClusterMeteringWorker has been taken offline")
	return nil
	/*
		if !isPrimaryPod() {
			return nil
		}

		logger.Info("ClusterMeteringWorker start")

		now := time.Now()
		if err := w.recordMeteringSvc.RunClusterMetering(ctx, now); err != nil {
			logger.Errorf("run cluster metering failed: %s", err)
		}

		logger.Info("ClusterMeteringWorker done")
		return nil
	*/
}

type ReportMetering2billingWorker struct {
	reportMetering2billing metering.ReportMetering2billingSvc
	workflow               workflow.Workflow
}

func NewReportMetering2billingWorker(
	reportMetering2billing metering.ReportMetering2billingSvc,
	flow workflow.Workflow,
) *ReportMetering2billingWorker {
	return &ReportMetering2billingWorker{
		reportMetering2billing: reportMetering2billing,
		workflow:               flow,
	}
}

func (w *ReportMetering2billingWorker) Run(ctx context.Context) error {
	logger := tilog.WithContextS(ctx)
	logger.Info("ReportMetering2billingWorker has been taken offline")
	return nil
	/*
		const (
			batchSize = 1
		)

		logger.Info("ReportMetering2billingWorker start")

		if err := w.reportMetering2billing.Report(ctx, batchSize); err != nil {
			logger.Errorf("report metering to billing failed:%s", err)
		}

		logger.Info("ReportMetering2billingWorker done")
		return nil
	*/
}

type BillingLagReporterWorker struct {
	lagReporter *metering.LagReporter
	workflow    workflow.Workflow
}

func NewBillingLagReporter(
	lagReporter *metering.LagReporter,
	flow workflow.Workflow,
) *BillingLagReporterWorker {
	return &BillingLagReporterWorker{
		lagReporter: lagReporter,
		workflow:    flow,
	}
}

func (w *BillingLagReporterWorker) Run(ctx context.Context) error {
	logger := tilog.WithContextS(ctx)
	logger.Info("BillingLagReporterWorker has been taken offline")
	return nil
	/*
		if !isPrimaryPod() {
			tilog.Info(ctx, "Not primary pod, skip worker")
			return nil
		}

		logger.Infof("BillingLagReporterWorker start")

		w.lagReporter.Report(ctx)

		logger.Infof("BillingLagReporterWorker done")

		return nil
	*/
}

func NewDBCleanerWorker(
	billingRecordsCleaner *metering.RecordsCleaner,
	dbProvider db.TransProvider,
	flow workflow.Workflow,
	cfg *config.Config,
) *DBCleanerWorker {
	return &DBCleanerWorker{
		billingRecordsCleaner: billingRecordsCleaner,
		dbProvider:            dbProvider,
		workflow:              flow,
		cfg:                   &cfg.BillingConfig,
	}
}

type DBCleanerWorker struct {
	billingRecordsCleaner *metering.RecordsCleaner
	dbProvider            db.TransProvider
	workflow              workflow.Workflow
	cfg                   *config.BillingConfig
}

func (w *DBCleanerWorker) Run(ctx context.Context) error {
	logger := tilog.WithContextS(ctx)
	logger.Info("DBCleanerWorker has been taken offline")
	return nil
	/*
		if !isPrimaryPod() {
			tilog.Info(ctx, "Not primary pod, skip worker")
			return nil
		}

		logger.Info("DBCleanerWorker start")
		meteringCfg := w.cfg

		param := &metering.RecordsCleanParam{
			RecordsToBeCleanedLimit:          meteringCfg.RecordsToBeCleanedLimit,
			ExpireDurationForClusterMetering: meteringCfg.ExpireDurationForClusterMetering,
			ExpireDurationForBillingRecord:   meteringCfg.ExpireDurationForBillingRecord,
		}
		if err := w.billingRecordsCleaner.Clean(ctx, param); err != nil {
			logger.Errorf("clean billing record: %s", err)
		}
		logger.Info("DBCleanerWorker done")

		return nil
	*/
}

func NewNormalPauseWorker(
	clusterSvc clustersvc.Service,
	clusterRepo clusterrepo.Repo,
	pausePlanRepo clusterrepo.PausePlanRepo,
	flow workflow.Workflow,
	cfg *config.Config,
	leaseLocker lease.Locker,
) *NormalPauseWorker {
	return &NormalPauseWorker{
		clusterSvc:    clusterSvc,
		clusterRepo:   clusterRepo,
		pausePlanRepo: pausePlanRepo,
		workflow:      flow,
		cfg:           cfg,
		leaseLocker:   leaseLocker,
	}
}

type NormalPauseWorker struct {
	clusterSvc    clustersvc.Service
	clusterRepo   clusterrepo.Repo
	pausePlanRepo clusterrepo.PausePlanRepo
	workflow      workflow.Workflow
	cfg           *config.Config
	leaseLocker   lease.Locker
}

const (
	normalPauseWorkerLockKey  = "dedicated-global-service:worker:normalPause"
	normalPauseWorkerLeaseTTL = 10 * time.Minute
	lockAttemptDivider        = 10
	minLockAttemptInterval    = time.Second
)

func (w *NormalPauseWorker) Run(ctx context.Context) error {
	owner := genLeaseAttemptOwnerID(globalWorkerOwnerID())
	acquired, err := w.leaseLocker.Acquire(ctx, normalPauseWorkerLockKey, owner, normalPauseWorkerLeaseTTL)
	if err != nil {
		tilog.Error(ctx, "acquire normal pause worker lease failed", zap.Error(err))
		return err
	}
	if !acquired {
		tilog.Info(ctx, "NormalPauseWorker skip as lease held by another replica", zap.String("owner", owner))
		return nil
	}
	tilog.Info(ctx, "NormalPauseWorker start")
	clusters, err := w.clusterRepo.Query(ctx, &clusterrepo.Filter{
		States:      []entity.State{globaldedicatedv1.Cluster_PAUSED},
		ShowDeleted: false,
	})
	if err != nil {
		tilog.Error(ctx, "query paused clusters failed", zap.Error(err))
		return err
	}

	clusterIds := make(map[uint64]*entity.Cluster, len(clusters))
	for _, cluster := range clusters {
		clusterIds[cluster.GetClusterId()] = cluster
	}

	pausePlans, err := w.pausePlanRepo.LoadMany(ctx, lo.Keys(clusterIds))
	if err != nil {
		tilog.Error(ctx, "load pause plans failed", zap.Error(err))
		return err
	}
	pausePlans = lo.MapValues(pausePlans, func(p *entity.ClusterPausePlan, _ uint64) *entity.ClusterPausePlan {
		entity.EnsureScheduledResumeTime(p, w.cfg.BizConfig.NormalPauseScheduleDay)
		return p
	})

	for clusterId, pausePlan := range pausePlans {
		// only support normal pause yet
		if pausePlan.GetPausePlan() != globaldedicatedv1.Cluster_PausePlan_NORMAL {
			continue
		}

		// check resume time
		// if scheduled resume time > now add notify hour, skip resume
		if pausePlan.ScheduledResumeTime.AsTime().
			After(time.Now().Add(time.Hour * time.Duration(w.cfg.BizConfig.NormalPauseNotifyHour))) {
			tilog.Info(ctx, "skip resume cluster", zap.Uint64("cluster_id", clusterId),
				zap.Time("scheduled_resume_time", pausePlan.ScheduledResumeTime.AsTime()),
				zap.Int("notify_hour", w.cfg.BizConfig.NormalPauseNotifyHour))
			continue
		}

		if _, ok := clusterIds[clusterId]; !ok {
			tilog.Warn(ctx, "cluster not found", zap.Uint64("cluster_id", clusterId))
			continue
		}

		tilog.Info(ctx, "start auto resume cluster", zap.Uint64("cluster_id", clusterId))
		// generate system user context
		cctx := metadata.AppendToIncomingMeta(ctx, &metadata.RequestMeta{
			OrgID:            clusterIds[clusterId].OrganizationId,
			DefaultProjectID: clusterIds[clusterId].ProjectId,
			ValidProjectList: []uint64{clusterIds[clusterId].ProjectId},
			UserID:           "0", // system user
			ClusterID:        clusterId,
		})

		var scheduledResumeTime *time.Time
		if pausePlan.ScheduledResumeTime != nil {
			scheduledResumeTime = lo.ToPtr(pausePlan.ScheduledResumeTime.AsTime())
		}

		_, err := w.clusterSvc.Resume(cctx, &globaldedicatedv1.ResumeClusterRequest{
			ClusterId: clusterId,
		}, scheduledResumeTime)

		if err != nil {
			tilog.Error(ctx, "resume cluster failed", zap.Uint64("cluster_id", clusterId), zap.Error(err))
		}
	}

	tilog.Info(ctx, "NormalPauseWorker done")
	return nil
}

type ClusterConfigBackupWorker struct {
	clusterSvc clustersvc.Service
	workflow   workflow.Workflow
}

func NewClusterConfigBackupWorker(
	clusterSvc clustersvc.Service,
	flow workflow.Workflow,
) *ClusterConfigBackupWorker {
	return &ClusterConfigBackupWorker{
		clusterSvc: clusterSvc,
		workflow:   flow,
	}
}

func (w *ClusterConfigBackupWorker) Run(ctx context.Context) error {
	tilog.Info(ctx, "ClusterConfigBackupWorker start")
	if err := w.clusterSvc.RunDueClusterConfigBackups(ctx, 100); err != nil {
		tilog.Error(ctx, "run due cluster config backups failed", zap.Error(err))
		return err
	}
	tilog.Info(ctx, "ClusterConfigBackupWorker done")
	return nil
}

var (
	globalWorkerOwnerOnce sync.Once
	globalWorkerOwner     string
)

func globalWorkerOwnerID() string {
	globalWorkerOwnerOnce.Do(func() {
		podName := os.Getenv("POD_NAME")
		if podName != "" {
			globalWorkerOwner = podName
			return
		}

		globalWorkerOwner = "pid-" + strconv.Itoa(os.Getpid())
	})

	return globalWorkerOwner
}

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

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