package bootstrap

import (
	"context"
	"time"

	"github.com/prometheus/client_golang/prometheus"
	"github.com/tidbcloud/dataflow-service/app/container"
	appmetrcisregister "github.com/tidbcloud/dataflow-service/app/metrics/register"
	appworker "github.com/tidbcloud/dataflow-service/app/worker"
	"github.com/tidbcloud/dataflow-service/internal/cmd/bootstrap/server"
	internalContainer "github.com/tidbcloud/dataflow-service/internal/container"
	"github.com/tidbcloud/dataflow-service/internal/metrics"
	backupmetrics "github.com/tidbcloud/dataflow-service/internal/metrics/backup"
	importmetrics "github.com/tidbcloud/dataflow-service/internal/metrics/import"
	restoremetrics "github.com/tidbcloud/dataflow-service/internal/metrics/restore"
	workerpkg "github.com/tidbcloud/dataflow-service/internal/worker"
	"github.com/tidbcloud/dataflow-service/migrate"
	_ "github.com/tidbcloud/dataflow-service/migrate" // blank import to make sure all migration files will be compiled
	"github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli"
	"github.com/tidbcloud/dataflow-service/pkg/log"
	"go.uber.org/zap"
	"golang.org/x/sync/errgroup"
)

type DataflowWorker struct {
	sever        *server.Server
	metricsSerer *server.MetricsServer
}

func NewDataflowWorker() (*DataflowWorker, error) {
	s := &DataflowWorker{}
	ctx := log.WithLogID(context.Background())
	logger := log.GetLogger(ctx)
	ser, err := server.NewServer(ctx, server.ServerOption{
		EnableEventBusConsumer: false,
	})
	if err != nil {
		logger.Error("setup server error", zap.Error(err))
		return nil, err
	}
	s.sever = ser

	metricsServer, err := server.NewMetricsServer(ctx)
	if err != nil {
		logger.Error("init metrics server error", zap.Error(err))
		return nil, err
	}
	registerWorkerMetricsHandler()
	s.metricsSerer = metricsServer
	return s, nil
}

func (s *DataflowWorker) RunDataflowWorker() error {
	ctx := context.Background()
	logger := log.GetLogger(ctx)

	if err := migrate.RunDataflowDBMigrate(container.Global().DB); err != nil {
		logger.Error("do DB migration failed", zap.Error(err))
		return err
	}
	logger.Info("migrate finished")

	setupInfraAPIConnection()
	workerM := workerpkg.New(
		&internalContainer.Global().BackupsModel,
		&internalContainer.Global().ImportsModel,
		&internalContainer.Global().DmModel,
		&internalContainer.Global().StatisticModel,
		&internalContainer.Global().ConnProfileModel)
	if err := workerM.StartBackgroundWorkers(); err != nil {
		logger.Error("failed to old arch workers", zap.Error(err))
		return err
	}
	logger.Info("old arch workers started")

	aw := appworker.MustNewWorker(ctx, time.UTC)
	aw.StartWorker()
	logger.Info("new arch workers started")

	eg := errgroup.Group{}
	eg.Go(func() error {
		return s.metricsSerer.ServerMetrics(ctx)
	})
	return eg.Wait()
}

func registerWorkerMetricsHandler() {
	// todo: optimize to one function
	importmetrics.RegisterMetrics()
	backupmetrics.RegisterMetrics()
	restoremetrics.RegisterMetrics()
	metrics.RegisterMetrics()
	appmetrcisregister.RegisterMetrics()
	prometheus.MustRegister(metrics.WorkerDuration)
	prometheus.MustRegister(metrics.BackupSchedulerAbnormal)
	prometheus.MustRegister(metrics.BackupCleanUpAbnormal)
	prometheus.MustRegister(metrics.BackupNodeGroupAndNodePoolExpired)
}

func setupInfraAPIConnection() {
	grpcli.MustNewTiDBClusterInfraClient()
}