package bootstrap

import (
	"context"

	"github.com/prometheus/client_golang/prometheus"
	"github.com/shiningrush/fastflow"
	"github.com/shiningrush/fastflow/pkg/entity"
	"github.com/shiningrush/fastflow/pkg/entity/run"
	"github.com/shiningrush/fastflow/pkg/exporter"
	"github.com/shiningrush/fastflow/pkg/mod"
	"github.com/shiningrush/goevent"
	"github.com/tidbcloud/dataflow-service-ng/app/container"
	appmetrcisregister "github.com/tidbcloud/dataflow-service-ng/app/metrics/register"
	data_pipeline_model "github.com/tidbcloud/dataflow-service-ng/app/models/data_pipeline"
	"github.com/tidbcloud/dataflow-service-ng/app/workflow"
	"github.com/tidbcloud/dataflow-service-ng/app/workflow/dag"
	data_pipeline_dag "github.com/tidbcloud/dataflow-service-ng/app/workflow/dag/data_pipeline"
	"github.com/tidbcloud/dataflow-service-ng/internal/cmd/bootstrap/server"
	"github.com/tidbcloud/dataflow-service-ng/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
	}
	if err := registerDataPipelineDags(
		container.Global().WorkflowModel,
		container.Global().DataPipelineLifecycleCoordinator,
	); err != nil {
		logger.Fatal("init data pipeline workflow dags", zap.Error(err))
		return nil, err
	}

	// 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 registerDataPipelineDags(
	workflowModel workflow.Workflow,
	coordinator data_pipeline_model.LifecycleCoordinator,
) error {
	if coordinator == nil {
		return nil
	}
	dags := []func(data_pipeline_model.LifecycleCoordinator) (*entity.Dag, []run.Action){
		data_pipeline_dag.GetCreateDataPipelineDag,
		data_pipeline_dag.GetPauseDataPipelineDag,
		data_pipeline_dag.GetEditDataPipelineDag,
		data_pipeline_dag.GetResumeDataPipelineDag,
		data_pipeline_dag.GetDeleteDataPipelineDag,
	}
	for _, build := range dags {
		dagModel, actions := build(coordinator)
		if err := workflowModel.RegisterDag(dagModel, actions); err != nil {
			return err
		}
	}
	return 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()
}