package bootstrap

import (
	"context"

	"github.com/prometheus/client_golang/prometheus"
	"github.com/shiningrush/fastflow"
	"github.com/shiningrush/fastflow/pkg/exporter"
	"github.com/shiningrush/fastflow/pkg/mod"
	"github.com/shiningrush/goevent"
	appmetrcisregister "github.com/tidbcloud/dataflow-service/app/metrics/register"
	"github.com/tidbcloud/dataflow-service/app/workflow/dag"
	"github.com/tidbcloud/dataflow-service/app/workflow/hook"
	"github.com/tidbcloud/dataflow-service/internal/cmd/bootstrap/server"
	"github.com/tidbcloud/dataflow-service/pkg/log"
	"go.uber.org/zap"
	"golang.org/x/sync/errgroup"
)

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

func NewDataflowFastflow() (*DataflowFastflow, error) {
	s := &DataflowFastflow{}
	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

	if err := dag.RegisterDags(); err != nil {
		logger.Fatal("init workflow dag", zap.Error(err))
		return nil, err
	}

	// Register cross-DAG lifecycle hooks. Must run before fastflow.Init so any
	// DagInstance that fails immediately on the first parse tick already sees
	// the hook installed.
	hook.RegisterBackupDagFailureHook()
	hook.RegisterBackupExportDagFailureHook()
	hook.RegisterPitBackupDagFailureHook()

	// init worker
	if err := fastflow.Init(&fastflow.InitialOption{
		Keeper:            mod.GetKeeper(),
		Store:             mod.GetStore(),
		ParserWorkersCnt:  300,
		ExecutorWorkerCnt: 3000,
	}); err != nil {
		logger.Fatal("init workflow failed", zap.Error(err))
		return nil, err
	}

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

func registerMetrics() error {
	execCollector := &exporter.ExecutorCollector{}
	if err := goevent.Subscribe(execCollector); err != nil {
		return err
	}
	leaderCollector := &exporter.LeaderCollector{}
	if err := goevent.Subscribe(leaderCollector); err != nil {
		return err
	}
	prometheus.MustRegister(execCollector)
	prometheus.MustRegister(leaderCollector)
	appmetrcisregister.RegisterMetrics()
	return nil
}

func (s *DataflowFastflow) RunDataflowFastflow() error {
	ctx := context.Background()
	eg := errgroup.Group{}
	eg.Go(func() error {
		return s.metricsSerer.ServerMetrics(ctx)
	})
	return eg.Wait()
}