package bootstrap

import (
	"context"
	"time"

	"github.com/prometheus/client_golang/prometheus"
	"github.com/tidbcloud/dataflow-service-ng/app/container"
	appmetrcisregister "github.com/tidbcloud/dataflow-service-ng/app/metrics/register"
	appworker "github.com/tidbcloud/dataflow-service-ng/app/worker"
	"github.com/tidbcloud/dataflow-service-ng/internal/cmd/bootstrap/server"
	"github.com/tidbcloud/dataflow-service-ng/internal/metrics"
	"github.com/tidbcloud/dataflow-service-ng/migrate"
	_ "github.com/tidbcloud/dataflow-service-ng/migrate" // blank import to make sure all migration files will be compiled
	"github.com/tidbcloud/dataflow-service-ng/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")

	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() {
	metrics.RegisterMetrics()
	appmetrcisregister.RegisterMetrics()
	prometheus.MustRegister(metrics.WorkerDuration)
}