package worker

import (
	"context"
	"net/http"
	"time"

	"github.com/pingcap/errors"
	"github.com/pingcap/log"
	"github.com/robfig/cron"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	appcontainer "github.com/tidbcloud/dataflow-service/app/container"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/dataflow-service/internal/bizutils"
	"github.com/tidbcloud/dataflow-service/internal/config"
	"github.com/tidbcloud/dataflow-service/internal/db"
	"github.com/tidbcloud/dataflow-service/internal/grpc/grpcli"
	"github.com/tidbcloud/dataflow-service/internal/metrics"
	"github.com/tidbcloud/dataflow-service/internal/model/backup"
	"github.com/tidbcloud/dataflow-service/internal/model/connprofile"
	"github.com/tidbcloud/dataflow-service/internal/model/dm"
	pimport "github.com/tidbcloud/dataflow-service/internal/model/import"
	"github.com/tidbcloud/dataflow-service/internal/model/statistic"
	"github.com/tidbcloud/pkgv2/worker"
	"go.uber.org/multierr"
	"go.uber.org/zap"
)

const (
	O11YClientID     = "Sd4aekPCa5nMUSQBIvf8eGSL895WtV4o"
	O11YClientSecret = "-SLP5UvxJdwGbVaKupxgv53T6otXguApPoG0fBsWCIqDG8_TuvlYnlkhPQHcWaKT"
)

type GlobalWorker struct {
	logger  *zap.Logger
	workers []*worker.PeriodWorker
	manager *Manager
	cronJob *cron.Cron
}

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

func NewGlobalBgWorker(manager *Manager) (*GlobalWorker, error) {
	gWorker := &GlobalWorker{
		logger:  log.L().With(zap.String("worker", "global")),
		manager: manager,
	}
	cronJob := cron.New()

	// schedule aggregate prometheus works job every 1h
	if err := cronJob.AddJob("@every 1h", &AggregatePrometheusWorker{
		cli:         &http.Client{Timeout: 100 * time.Second},
		backupModel: manager.backupModel,
	}); err != nil {
		return nil, errors.Annotate(err, "add aggregate prometheus worker")
	}

	gWorker.cronJob = cronJob
	importWorker := ImportWorker{model: manager.importModel}
	dmJobWorker := DMJobWorker{model: manager.dmModel}
	plEndpointWorker := PrivateLinkEndpointWorker{model: manager.connProfileModel, unusedEndpointMatchCountMap: make(map[string]int)}
	dmMeteringWorker := DMMeteringWorker{model: manager.dmModel}
	statisticWorker := StatisticWorker{model: manager.statisticModel}
	gWorker.workers = []*worker.PeriodWorker{
		worker.NewPeriodWorker(gWorker.logger.With(zap.String("worker", "import")), time.Second*2, withMetrics(0, "import", importWorker.Run)),
		worker.NewPeriodWorker(gWorker.logger.With(zap.String("worker", "import-bucket-cleaner")), 10*time.Minute, withMetrics(0, "import-bucket-cleaner", importWorker.CleanBucket)),
		worker.NewPeriodWorker(gWorker.logger.With(zap.String("worker", "dm-job-v2")), time.Second*10, withMetrics(0, "dm-job-v2", dmJobWorker.RunPer10SecondsV2)),
		worker.NewPeriodWorker(gWorker.logger.With(zap.String("worker", "dm-job-cleaner")), time.Minute, withMetrics(0, "dm-job-cleaner", dmJobWorker.RunPerMinute)),
		worker.NewPeriodWorker(gWorker.logger.With(zap.String("worker", "pl-endpoint")), time.Second*5, withMetrics(0, "pl-endpoint", plEndpointWorker.Run)),
		worker.NewPeriodWorker(gWorker.logger.With(zap.String("worker", "pl-endpoint-cleaner")), time.Minute, withMetrics(0, "pl-endpoint-cleaner", plEndpointWorker.RunPerMinute)),
		worker.NewPeriodWorker(gWorker.logger.With(zap.String("worker", "dm-new-metering")), time.Minute*10, withMetrics(0, "dm-new-metering", dmMeteringWorker.RunNewMeter)),
		worker.NewPeriodWorkerWithDelay(gWorker.logger.With(zap.String("worker", "dm-metering-sync")), time.Minute, time.Minute*10, withMetrics(0, "dm-metering-sync", dmMeteringWorker.SyncDMMetering)),
		worker.NewPeriodWorker(gWorker.logger.With(zap.String("worker", "statistic")), time.Minute*30, withMetrics(0, "statistic", statisticWorker.Run)),
	}
	return gWorker, nil
}

type ImportWorker struct {
	model *pimport.Model
}

func (w ImportWorker) Run(ctx context.Context) error {
	return w.model.Run(ctx)
}

func (w ImportWorker) CleanBucket(ctx context.Context) error {
	return w.model.CleanBucket(ctx)
}

type DMJobWorker struct {
	model *dm.Model
}

func (w *DMJobWorker) RunPer10SecondsV2(ctx context.Context) error {
	return multierr.Combine(
		w.model.RunDMJobs(ctx),
	)
}

func (w *DMJobWorker) RunPerMinute(ctx context.Context) error {
	return multierr.Combine(
		w.model.RunCleanupJobsForDeletedCluster(ctx),
	)
}

type PrivateLinkEndpointWorker struct {
	model *connprofile.Model

	// how many times a pl-endpoint in available phase is matched as unused, if count >= 10(10 minutes), we clear it by mark it as
	// failed. Since pl-endpoint is created separated from dm job, we don't know for sure when the pl-endpoint
	// is unused completely. Even if count >= 10, user might still use the pl-endpoint to create job.
	// todo: right now, a pl-endpoint is shared in some project & region, maybe let user manage pl-endpoints or we
	// create pl-endpoint for each job in some project & region, so we can deleted the pl-endpoint when job is deleted.
	//
	// map key: project-id/region/endpoint-service-name
	unusedEndpointMatchCountMap map[string]int
}

func (w *PrivateLinkEndpointWorker) Run(ctx context.Context) error {
	return multierr.Combine(
		w.model.RunDeduplicatePLEndpoints(ctx),
		w.model.RunCreatePLEndpoints(ctx),
		w.model.RunDeleteFailedPLEndpoints(ctx),
		// todo: The service provider can also reject a connection after it is available for use. we need to update it's state.
		// in this case, all jobs depends on it will fail.
	)
}

func (w *PrivateLinkEndpointWorker) RunPerMinute(ctx context.Context) error {
	return multierr.Combine(
		w.model.RunClearUnusedPLEndpoints(ctx, w.unusedEndpointMatchCountMap),
	)
}

type DMMeteringWorker struct {
	model *dm.Model
}

func (w *DMMeteringWorker) RunNewMeter(ctx context.Context) error {
	startMetering := config.Global().ServerConfig.EnableDMMetering
	if !startMetering {
		return nil
	}
	return w.model.NewDMMetering(ctx)
}

func (w *DMMeteringWorker) SyncDMMetering(ctx context.Context) error {
	return w.model.SyncDMMetering(ctx)
}

type StatisticWorker struct {
	model *statistic.Model
}

func (w StatisticWorker) Run(ctx context.Context) error {
	return multierr.Combine(
		w.model.CollectCountStatisticDFSBusiness(ctx),
	)
}

// AggregatePrometheusWorker is used for collect all data plane cluters' metrics
// into control plane
// 1. to generate a global view of all Backup/CDC customers.
// 2. to measure the DR scenario adoption.
type AggregatePrometheusWorker struct {
	cli         *http.Client
	backupModel *backup.Model

	bizType string
}

func (w *AggregatePrometheusWorker) Run() {
	ctx := context.Background()
	dbIns := db.GetDBWithContext(ctx)

	token, err := metrics.GetAuthToken(w.cli, O11YClientID, O11YClientSecret)
	if err != nil {
		log.L().Error("[Prometheus Aggregation] unable to get token", zap.Error(err))
		return
	}
	env := config.Global().Env
	w.bizType = metrics.GetBizType(env)

	logBackupClusters, err := w.backupModel.GetLogBackupsClusters(ctx, dbIns)
	if err != nil {
		log.L().Error("[Prometheus Aggregation] unable to get log backup clusters", zap.Error(err))
	}
	logBackupMetrics := config.Global().GrafanaConfig.BackupGaugeMetrics
	if len(logBackupClusters) > 0 && len(logBackupMetrics) > 0 {
		log.L().Info("[Prometheus Aggregation] get log backup clusters", zap.Int("count", len(logBackupClusters)))
		w.collectMetricsFromClusters(logBackupClusters, logBackupMetrics, token)
	}

	cdcClusters, err := w.listClustersWithCDCEnabled(ctx)
	if err != nil {
		log.L().Error("[Prometheus Aggregation] unable to get cdc clusters", zap.Error(err))
	}
	cdcMetrics := config.Global().GrafanaConfig.CDCGaugeMetrics
	if len(cdcClusters) > 0 && len(cdcMetrics) > 0 {
		log.L().Info("[Prometheus Aggregation] get cdc clusters", zap.Int("count", len(cdcClusters)))
		w.collectMetricsFromClusters(cdcClusters, cdcMetrics, token)
	}
}

func (w *AggregatePrometheusWorker) collectMetricsFromClusters(clusters []*core.Cluster, colletedMetrics []string, token string) {
	for _, c := range clusters {
		prometheusURL, err := metrics.GetPrometheusURLForCluster(w.cli, c, w.bizType, token)
		if err != nil {
			log.L().Error("[Prometheus Aggregation] unable to get prometheus url", zap.Uint64("id", c.Id), zap.Error(err))
			continue
		}
		log.L().Info("[Prometheus Aggregation] get prometheus url for cluster", zap.Uint64("id", c.Id), zap.String("prometheus", prometheusURL))

		startTime, endTime := metrics.GetPrometheusTimeRange()
		for _, metric := range colletedMetrics {
			err = metrics.CollectMetrics(w.cli, prometheusURL, metric, startTime, endTime)
			if err != nil {
				log.L().Error("[Prometheus Aggregation] unable to collect metrics", zap.Uint64("id", c.Id), zap.Error(err))
			} else {
				log.L().Info("[Prometheus Aggregation] success collect metrics", zap.Uint64("id", c.Id), zap.Any("metric", metric))
			}
		}
	}
}

func (w *AggregatePrometheusWorker) listClustersWithCDCEnabled(ctx context.Context) ([]*core.Cluster, error) {
	changefeeds, bizErr := appcontainer.Global().BizChangefeedRepo.List(ctx,
		func(qs entity.BizChangefeedQuerySet) entity.BizChangefeedQuerySet {
			return qs.OperationStatusNotIn(
				entity.BizChangefeedOperationStatusDeleting,
				entity.BizChangefeedOperationStatusDeleted,
			)
		},
	)
	if bizErr != nil {
		return nil, errors.Trace(bizErr)
	}
	if len(changefeeds) == 0 {
		return nil, nil
	}

	// Deduplicate cluster IDs
	seen := make(map[uint64]struct{}, len(changefeeds))
	clusterIDs := make([]uint64, 0, len(changefeeds))
	for _, cf := range changefeeds {
		id := cf.ClusterID.Uint64()
		if _, ok := seen[id]; !ok {
			seen[id] = struct{}{}
			clusterIDs = append(clusterIDs, id)
		}
	}

	clusterResp, err := grpcli.ListClustersById(ctx, clusterIDs)
	if err != nil {
		return nil, errors.Annotatef(err, "get clusters %v error", clusterIDs)
	}
	clusters := make([]*core.Cluster, 0, len(clusterResp.Clusters))
	for _, c := range clusterResp.Clusters {
		if !bizutils.ClusterDeletedOrCleared(c) {
			clusters = append(clusters, c)
		}
	}
	return clusters, nil
}