// Package opworker is the background worker of the central service.
// The central service records intent to make changes.
// The worker finds those in the database, performs them, and updates the database.
package opworker

import (
	"context"
	"fmt"
	"sync"
	"time"

	"github.com/pingcap/dbaas/central/db/entity"
	infraapi "github.com/pingcap/dbaas/central/infra-api"
	"github.com/pingcap/dbaas/central/metrics"
	"github.com/pingcap/dbaas/central/model/backup"
	billingworker "github.com/pingcap/dbaas/central/model/billing/worker"
	"github.com/pingcap/dbaas/central/model/cdc"
	"github.com/pingcap/dbaas/central/model/cluster"
	"github.com/pingcap/dbaas/central/model/cluster/clusterproxy/tidbcloud"
	pimport "github.com/pingcap/dbaas/central/model/import"
	"github.com/pingcap/dbaas/central/model/replication"
	"github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/pkgv2/worker"
	"go.uber.org/multierr"
	"go.uber.org/zap"
)

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

func (w *BgWorker) Stop() {
	wg := sync.WaitGroup{}
	wg.Add(len(w.workers))
	for _, item := range w.workers {
		go func(wi *worker.PeriodWorker) {
			defer wg.Done()
			wi.Stop()
		}(item)
	}
	wg.Wait()
}

// NewBgWorker creates a new worker.
func NewBgWorker(cbg cluster.BgContext, bbg backup.BgContext, ibg pimport.BgContext, rbg replication.BgContext, dbg cdc.BgContext, billingBg billingworker.BgContext, tenantID entity.ID, api *infraapi.API) (*BgWorker, error) {
	logger := log.With(zap.Uint64("tenantID", tenantID.Uint64()))
	bgWorker := &BgWorker{
		clusters:     cbg,
		backups:      bbg,
		imports:      ibg,
		replications: rbg,
		cdcs:         dbg,
		billing:      billingBg,
		tenantID:     tenantID,
		api:          api,
	}
	bgWorker.workers = []*worker.PeriodWorker{
		worker.NewPeriodWorker(logger.With(zap.String("worker", "cluster")), time.Second*5, withMetrics(tenantID.Uint64(), "cluster", clusterWorker{*bgWorker}.Run)),
		worker.NewPeriodWorker(logger.With(zap.String("worker", "backup")), time.Second*5, withMetrics(tenantID.Uint64(), "backup", backupWorker{*bgWorker}.Run)),
		worker.NewPeriodWorker(logger.With(zap.String("worker", "cleanup")), time.Second*30, withMetrics(tenantID.Uint64(), "cleanup", cleanupWorker{*bgWorker}.Run)),
		worker.NewPeriodWorker(logger.With(zap.String("worker", "import")), time.Second*5, withMetrics(tenantID.Uint64(), "import", importWorker{*bgWorker}.Run)),
		worker.NewPeriodWorker(logger.With(zap.String("worker", "replication")), time.Second*30, withMetrics(tenantID.Uint64(), "replication", replicationWorker{*bgWorker}.Run)),
		worker.NewPeriodWorker(logger.With(zap.String("worker", "cdc")), time.Second*30, withMetrics(tenantID.Uint64(), "cdc", cdcWorker{*bgWorker}.Run)),
		worker.NewPeriodWorker(logger.With(zap.String("worker", "cdc")), time.Minute*10, withMetrics(tenantID.Uint64(), "cdc", cdcWorker{*bgWorker}.RunInTenMinutes)),
		worker.NewPeriodWorker(logger.With(zap.String("worker", "billing")), time.Minute*5, withMetrics(tenantID.Uint64(), "billing", billingWorker{*bgWorker}.Run)),
	}
	return bgWorker, nil
}

// BgWorker is a base worker used by other workers
// Using it in WorkLoop() will build the other workers
type BgWorker struct {
	clusters     cluster.BgContext
	backups      backup.BgContext
	imports      pimport.BgContext
	replications replication.BgContext
	cdcs         cdc.BgContext
	billing      billingworker.BgContext
	tenantID     entity.ID
	api          *infraapi.API

	workers []*worker.PeriodWorker
}

type cleanupWorker struct {
	BgWorker
}

func (w cleanupWorker) Run(ctx context.Context) error {
	return multierr.Combine(
		w.backups.CleanupBackupRestoreJobs(ctx),
		// TODO: support cleanup volume in shoot
		// w.runtime.DeleteLocalVolumesWithDeletedNodes(),
		// TODO: clean up empty shoots
	)
}

type replicationWorker struct {
	BgWorker
}

func (w replicationWorker) Run(ctx context.Context) error {
	return multierr.Combine(
		w.replications.Run(ctx),
		w.replications.Clean(ctx),
		w.replications.GetLagTime(ctx),
		w.replications.ScaleSecondary(ctx),
	)
}

type cdcWorker struct {
	BgWorker
}

func (w cdcWorker) Run(ctx context.Context) error {
	return multierr.Combine(
		w.cdcs.ManageCdcChangefeed(ctx, tidbcloud.BuildClusterProxy),
		w.cdcs.CleanUnusedCdcCluster(ctx),
	)
}

func (w cdcWorker) RunInTenMinutes(ctx context.Context) error {
	return multierr.Combine(
		w.cdcs.SendTicdcMesaures(ctx),
	)
}

type importWorker struct {
	BgWorker
}

func (w importWorker) Run(ctx context.Context) error {
	return w.imports.Run(ctx)
}

type backupWorker struct {
	BgWorker
}

func (w backupWorker) Run(ctx context.Context) error {
	return multierr.Combine(
		// already migrated into core service
		//w.backups.TakeBackups(ctx),
		// Running this after take avoids the possibility of an error message about a backup that has not even begun
		w.backups.RestoreBackups(ctx),
		// Taking or restoring backups is a higher priority than scheduling them
		w.backups.ScheduleBackups(ctx),
	)
}

type clusterWorker struct {
	BgWorker
}

// FIXME: if the cluster is delete before the shoot provisioned, cleanup won't work
func (w clusterWorker) Run(ctx context.Context) error {
	return multierr.Combine(
		w.clusters.Run(ctx),
		w.clusters.SyncReadonly(ctx),
		w.clusters.SyncClearCDCInfo(ctx),
	)
}

type billingWorker struct {
	BgWorker
}

func (w billingWorker) Run(ctx context.Context) error {
	return multierr.Combine(
		w.billing.POCReminder(),
		w.billing.POCCleanup(),
		w.billing.NewTenantEmailNotify(),
		w.billing.SyncEnableOnDemandByPayable(),
		w.billing.DisableOutdatedDevTier(),
		w.billing.NotifyUserDevTierStage(),
		w.billing.SyncUnSyncedSalesforce(),
	)
}

func withMetrics(tenantID uint64, workerName string, f worker.Handler) worker.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", tenantID), workerName, status).Observe(duration)
		return err
	}
}