package container

import (
	"context"
	"os"

	console_api "github.com/tidbcloud/dataflow-service-ng/app/controllers/console"
	import_into_controller "github.com/tidbcloud/dataflow-service-ng/app/controllers/import_into"
	"github.com/tidbcloud/dataflow-service-ng/app/controllers/inner"
	open_changefeed "github.com/tidbcloud/dataflow-service-ng/app/controllers/open/changefeed"
	premium_dm_controller "github.com/tidbcloud/dataflow-service-ng/app/controllers/premium_dm"
	"github.com/tidbcloud/dataflow-service-ng/app/event_bus/publisher/center"
	"github.com/tidbcloud/dataflow-service-ng/app/internal/cluster_job"
	"github.com/tidbcloud/dataflow-service-ng/app/internal/feature_gate"
	feature_gate_impl "github.com/tidbcloud/dataflow-service-ng/app/internal/feature_gate/impl"
	"github.com/tidbcloud/dataflow-service-ng/app/internal/infra_metrics"
	infra_metrics_impl "github.com/tidbcloud/dataflow-service-ng/app/internal/infra_metrics/impl"
	billing_model "github.com/tidbcloud/dataflow-service-ng/app/models/billing"
	billing_model_impl "github.com/tidbcloud/dataflow-service-ng/app/models/billing/impl"
	"github.com/tidbcloud/dataflow-service-ng/app/models/biz_changefeed"
	biz_changefeed_impl "github.com/tidbcloud/dataflow-service-ng/app/models/biz_changefeed/impl"
	"github.com/tidbcloud/dataflow-service-ng/app/models/config_manager"
	config_manager_impl "github.com/tidbcloud/dataflow-service-ng/app/models/config_manager/impl"
	data_pipeline_model "github.com/tidbcloud/dataflow-service-ng/app/models/data_pipeline"
	import_into_model "github.com/tidbcloud/dataflow-service-ng/app/models/import_into"
	premium_dm_model "github.com/tidbcloud/dataflow-service-ng/app/models/premium_dm"
	"github.com/tidbcloud/dataflow-service-ng/app/models/sink_endpoint"
	sink_endpoint_impl "github.com/tidbcloud/dataflow-service-ng/app/models/sink_endpoint/impl"
	"github.com/tidbcloud/dataflow-service-ng/app/models/ticdc"
	ticdc_impl "github.com/tidbcloud/dataflow-service-ng/app/models/ticdc/impl"
	"github.com/tidbcloud/dataflow-service-ng/app/models/ticdc_spec"
	"github.com/tidbcloud/dataflow-service-ng/app/repo"
	billing_repo "github.com/tidbcloud/dataflow-service-ng/app/repo/billing"
	biz_changefeed_repo "github.com/tidbcloud/dataflow-service-ng/app/repo/biz_changefeed"
	data_pipeline_repo "github.com/tidbcloud/dataflow-service-ng/app/repo/data_pipeline"
	"github.com/tidbcloud/dataflow-service-ng/app/repo/entity"
	sink_endpoint_repo "github.com/tidbcloud/dataflow-service-ng/app/repo/sink_endpoint"
	ticdc_repo "github.com/tidbcloud/dataflow-service-ng/app/repo/ticdc"
	"github.com/tidbcloud/dataflow-service-ng/app/workflow"
	bizworkflow "github.com/tidbcloud/dataflow-service-ng/app/workflow/action/biz_changefeed"
	"github.com/tidbcloud/dataflow-service-ng/app/workflow/action/biz_changefeed/scheduler"
	workflow_impl "github.com/tidbcloud/dataflow-service-ng/app/workflow/impl"
	importintopb "github.com/tidbcloud/dataflow-service-ng/idl/pbgen/proto/import_into/v1beta2"
	service_config "github.com/tidbcloud/dataflow-service-ng/internal/config"
	"github.com/tidbcloud/dataflow-service-ng/pkg/agent"
	"github.com/tidbcloud/dataflow-service-ng/pkg/billing"
	encryptor_wrapper "github.com/tidbcloud/dataflow-service-ng/pkg/encryptor-wrapper"
	"github.com/tidbcloud/dataflow-service-ng/pkg/grpc/grpcli"
	"github.com/tidbcloud/dataflow-service-ng/pkg/id_generator"
	"github.com/tidbcloud/dataflow-service-ng/pkg/lake"
	"github.com/tidbcloud/dataflow-service-ng/pkg/log"
	regionalapi "github.com/tidbcloud/dataflow-service-ng/pkg/regional-api"
	"github.com/tidbcloud/pkgv2/db"
	"go.uber.org/zap"
	"gorm.io/gorm"
)

var globalContainer = &GlobalContainer{}

type GlobalContainer struct {
	DB            *gorm.DB
	TransProvider db.TransProvider

	ChangefeedSpecificationRepo       biz_changefeed_repo.ChangefeedSpecificationRepo
	CDCInstanceTypeRepo               biz_changefeed_repo.CDCInstanceTypeRepo
	CDCMachineTypeRepo                biz_changefeed_repo.CDCMachineTypeRepo
	BillingRecordRepo                 repo.BillingRecordRepo
	TiCDCClusterRepo                  repo.TiCDCClusterRepo
	TiCDCBizChangefeedResourceRepo    repo.TiCDCBizChangefeedResourceRepo
	TiCDCChangefeedRepo               repo.TiCDCChangefeedRepo
	KernelChangefeedTrafficRecordRepo repo.KernelChangefeedTrafficRecordRepo
	BizChangefeedRepo                 repo.BizChangefeedRepo
	BizSinkEndpointRepo               repo.BizSinkEndpointRepo
	BizSinkEndpointUsageRepo          repo.BizSinkEndpointUsageRepo
	TiCDCCustomStorageRepo            repo.TiCDCCustomStorageRepo
	DataPipelineRepo                  data_pipeline_repo.DataPipelineRepo

	// grpc clients
	AccountSvcGrpcClient   grpcli.AccountSvcGrpcClient
	AccountAclGrpcClient   grpcli.AccountAclGrpcClient
	ConfigCenterGrpcClient grpcli.ConfigCenterGrpcClient
	RegionGrpcClient       grpcli.RegionGrpcClient
	MgmtSvcGrpcClient      grpcli.MgmtSvcGrpcClient

	// regional api client
	RegionalClient            regionalapi.RegionalClient
	TiCDCClusterAPIClient     regionalapi.TiCDCClusterClient
	ChangefeedAPIClient       regionalapi.ChangefeedClient
	TiCDCPrivateLinkAPIClient regionalapi.TiCDCPrivateLinkClient
	TiCDCPreCheckAPIClient    regionalapi.TiCDCPreCheckClient
	NGRegionalClientProvider  agent.NGRegionalClientProvider

	// cdc models
	CDCInstanceTypeModel                 ticdc_spec.CDCInstanceTypeModel
	ClusterJobModel                      cluster_job.Model
	TiCDCMachineManager                  ticdc.MachineManager
	RegionalTiCDCClusterModel            ticdc.ClusterModel
	RegionalTiCDCChangefeedModel         ticdc.ChangefeedModel
	ChangefeedTrafficMeter               ticdc.ChangefeedTrafficMeter
	RegionalTiCDCClusterWorker           ticdc.ClusterWorker
	RegionalTiCDCChangefeedWorker        ticdc.LongTermSynchronization
	SinkEndpointModel                    sink_endpoint.SinkEndpointModel
	EndpointServiceAvailabilityZoneModel sink_endpoint.EndpointServiceAvailabilityZoneModel
	SinkEndpointWorker                   sink_endpoint.SinkEndpointWorker

	// billing models
	SyncBillingRecord billing_model.SyncBillingRecord

	// import into models
	ImportIntoModel import_into_model.ImportIntoModel

	// config manager models
	TiCDCClusterConfigManager config_manager.TiCDCClusterConfigManager
	ChangefeedConfigManager   config_manager.ChangefeedConfigManager

	// biz_changefeed model
	BizChangefeedDownstreamFactory *biz_changefeed_impl.DownstreamAdaptorFactory
	BizChangefeedScheduler         bizworkflow.BizChangefeedScheduler
	BizChangefeedHooker            biz_changefeed.BizChangefeedHooker
	BizChangefeedModel             biz_changefeed.BizChangefeedModel
	BizChangefeedMeteringWorker    biz_changefeed.BizChangefeedMeteringWorker
	BizChangefeedMetricsWorker     biz_changefeed.BizChangefeedMetricsWorker

	// data pipeline runtime
	//
	// DataPipelineConfig is the resolved configuration: the file settings after
	// the secret-backed and environment-supplied values are folded in. Anything
	// deciding whether Data Pipeline is configured must read this rather than
	// the raw config, or the two can disagree about it.
	DataPipelineConfig               service_config.DataPipelineConfig
	DataPipelineExportClient         grpcli.ExportGrpcClient
	DataPipelineLakeClient           lake.Client
	DataPipelineValidator            data_pipeline_model.ExternalResourceValidator
	DataPipelineLifecycleCoordinator data_pipeline_model.LifecycleCoordinator
	DataPipelineModel                data_pipeline_model.Model
	DataPipelineObservationModel     data_pipeline_model.ObservationModel

	// console-api
	BizChangefeedSpecController console_api.ChangefeedSpecController
	BizChangefeedController     console_api.ChangefeedController
	SinkEndpointController      console_api.SinkEndpointController
	ImportIntoController        *import_into_controller.ImportIntoService
	ImportIntoAPI               importintopb.ImportIntoServiceServer
	DataPipelineController      console_api.DataPipelineController
	OpenChangefeedAPI           *open_changefeed.Service

	// premium dm
	PremiumDMDAO            *premium_dm_model.PremiumDMTaskDAO
	PremiumDMWorker         *premium_dm_model.PremiumDMTaskWorker
	PremiumDMPrecheckWorker *premium_dm_model.PremiumDMPrecheckWorker
	PremiumDMAPI            *premium_dm_controller.PremiumDMService

	WorkflowModel workflow.Workflow
	FeatureGate   feature_gate.FeatureGate
	Encryptor     encryptor_wrapper.Encryptor
	InfraMetrics  infra_metrics.InfraMetrics

	EventCenterPublisher       center.EventCenterPublisher
	DataflowEventHookerManager entity.DataflowEventHookerManager

	// grpc inner service for ops portal
	OpsInnerController inner.OpsInnerController
}

func Global() *GlobalContainer {
	return globalContainer
}

// SetUpContainers this func only called once in main.go
func SetUpContainers(ctx context.Context, db *gorm.DB) error {
	logger := log.GetLogger(ctx)

	container := &GlobalContainer{}

	// init db
	container.DB = db
	container.TransProvider = newTransProvider(container.DB)

	// init event center
	eventCenterPublisher, err := center.NewEventCenterPublisher()
	if err != nil {
		return err
	}
	container.EventCenterPublisher = eventCenterPublisher

	// init workflow
	err = initWorkflow(container)
	if err != nil {
		log.GetLogger(ctx).Error("init workflow failed", zap.Error(err))
		return err
	}

	// init repo
	initRepos(container)

	// init grpc
	initGrpcClient(ctx, container)

	// init regional api
	initRegionalAPIClient(container)

	// init KMS
	encryptor, err := encryptor_wrapper.NewEncryptor(service_config.Global().InfraConfig.CMKArn)
	if err != nil {
		return err
	}
	container.Encryptor = encryptor

	// init internal helper
	initInternalHelper(container)

	// init model
	initModels(container)
	if err := initDataPipeline(container); err != nil {
		return err
	}

	// init premium dm
	initPremiumDM(container)

	// init console controller
	initController(container)

	globalContainer = container
	logger.Info("container setup success")
	return nil
}

func initInternalHelper(container *GlobalContainer) {
	container.FeatureGate = feature_gate_impl.NewFeatureGate()
	container.InfraMetrics = infra_metrics_impl.NewInfraMetrics()
}

func initController(container *GlobalContainer) {
	container.BizChangefeedController = console_api.NewBiZChangefeedController(
		container.TransProvider,
		container.BizChangefeedModel,
		container.SinkEndpointModel,
		container.MgmtSvcGrpcClient,
		container.FeatureGate)
	container.BizChangefeedSpecController = console_api.NewBizChangefeedSpecController(
		container.TransProvider, container.BizChangefeedModel)
	container.SinkEndpointController = console_api.NewSinkEndpointController(
		container.SinkEndpointModel,
		container.EndpointServiceAvailabilityZoneModel,
		container.MgmtSvcGrpcClient,
	)
	container.ImportIntoController = import_into_controller.NewImportIntoService(
		container.MgmtSvcGrpcClient,
		container.Encryptor,
		container.NGRegionalClientProvider,
		container.EventCenterPublisher,
		container.ClusterJobModel,
	)
	container.ImportIntoAPI = container.ImportIntoController
	container.DataPipelineController = console_api.NewDataPipelineController(
		container.DataPipelineModel,
		container.MgmtSvcGrpcClient,
	)
	container.OpenChangefeedAPI = open_changefeed.NewService(
		container.BizChangefeedController,
		container.BizChangefeedSpecController,
		container.BizChangefeedRepo,
	)

	container.OpsInnerController = inner.NewOpsInnerController(
		container.BizChangefeedModel,
		container.RegionalTiCDCClusterModel,
		container.RegionalTiCDCChangefeedModel,
		container.TiCDCClusterConfigManager,
		container.ChangefeedConfigManager,
		container.MgmtSvcGrpcClient,
		container.BizChangefeedController,
	)
}

func initGrpcClient(ctx context.Context, container *GlobalContainer) {
	container.AccountSvcGrpcClient = grpcli.NewAccountSvcGrpcClient(
		ctx,
		service_config.Global().GrpcConfig.AccountProviderAddr)
	container.AccountAclGrpcClient = grpcli.NewAccountAclGrpcClient(
		ctx,
		service_config.Global().GrpcConfig.AccountAclSvcAddr)
	container.ConfigCenterGrpcClient = grpcli.NewConfigCenterGrpcClient(
		ctx,
		service_config.Global().GrpcConfig.ConfigCenterAddr)
	container.RegionGrpcClient = grpcli.NewRegionGrpcClient(ctx, service_config.Global().GrpcConfig.TiDBMgmtSvcAddr)
	container.MgmtSvcGrpcClient = grpcli.NewMgmtSvcGrpcClient(ctx, service_config.Global().GrpcConfig.TiDBMgmtSvcAddr)
}

func initRegionalAPIClient(container *GlobalContainer) {
	container.RegionalClient = regionalapi.NewRegionalClient(container.RegionGrpcClient)
	container.TiCDCClusterAPIClient = regionalapi.NewTiCDCClusterClient(container.RegionalClient)
	container.ChangefeedAPIClient = regionalapi.NewChangefeedClient(container.RegionalClient)
	container.TiCDCPrivateLinkAPIClient = regionalapi.NewTiCDCPrivateLinkClient(container.RegionalClient)
	container.TiCDCPreCheckAPIClient = regionalapi.NewTiCDCPreCheckClient(container.RegionalClient)

	container.NGRegionalClientProvider = agent.NewNGRegionalClientProvider(
		container.MgmtSvcGrpcClient,
		&agent.ClientConfig{
			MgmtSvcProxyAddr:  service_config.Global().GrpcConfig.MgmtSvcProxyAddr,
			MgmtSvcProxyToken: service_config.Global().GrpcConfig.MgmtSvcProxyToken,
		},
	)
}

func initWorkflow(container *GlobalContainer) error {
	impl, err := workflow_impl.NewWorkflow(checkIsWorker())
	if err != nil {
		return err
	}
	container.WorkflowModel = impl
	return nil
}

func checkIsWorker() bool {
	podName := os.Getenv("POD_NAME")
	return podName != ""
}

func initRepos(container *GlobalContainer) {
	container.BillingRecordRepo = billing_repo.NewBillingRecordRepo(container.TransProvider)
	container.TiCDCClusterRepo = ticdc_repo.NewTiCDCClusterRepo(container.TransProvider)
	container.TiCDCBizChangefeedResourceRepo = ticdc_repo.NewTiCDCBizChangefeedResourceRepo(container.TransProvider)
	container.TiCDCChangefeedRepo = ticdc_repo.NewTiCDCChangefeedRepo(container.TransProvider)
	container.KernelChangefeedTrafficRecordRepo = ticdc_repo.
		NewKernelChangefeedTrafficRecordRepo(container.TransProvider)
	container.BizChangefeedRepo = biz_changefeed_repo.NewBizChangefeedRepo(
		container.TransProvider,
		id_generator.MustNewSnowflakeIDGenerator(id_generator.NewHostSnowflakeMachineIDGenerator()))
	container.ChangefeedSpecificationRepo = biz_changefeed_repo.NewChangefeedSpecificationRepo(container.TransProvider)
	container.CDCInstanceTypeRepo = biz_changefeed_repo.NewCDCInstanceTypeRepo(container.TransProvider)
	container.CDCMachineTypeRepo = biz_changefeed_repo.NewCDCMachineTypeRepo(container.TransProvider)
	container.BizSinkEndpointRepo = sink_endpoint_repo.NewBizSinkEndpointRepo(container.TransProvider)
	container.BizSinkEndpointUsageRepo = sink_endpoint_repo.NewBizSLinkEndpointUsageRepo(container.TransProvider)
	container.TiCDCCustomStorageRepo = ticdc_repo.NewTiCDCCustomStorageRepo(container.TransProvider)
	container.DataPipelineRepo = data_pipeline_repo.NewDataPipelineRepo(
		container.TransProvider,
		id_generator.MustNewSnowflakeIDGenerator(id_generator.NewHostSnowflakeMachineIDGenerator()),
	)
}

//nolint:funlen
func initModels(container *GlobalContainer) {
	clusterJobModel := cluster_job.NewClusterJobModelImpl(container.MgmtSvcGrpcClient)

	container.TiCDCClusterConfigManager = config_manager_impl.NewTiCDCClusterConfigCenterManager(
		container.ConfigCenterGrpcClient)
	container.ChangefeedConfigManager = config_manager_impl.NewChangefeedConfigCenterManager(
		container.ConfigCenterGrpcClient)

	container.ClusterJobModel = clusterJobModel
	container.CDCInstanceTypeModel = ticdc_spec.NewCDCInstanceTypeModel(
		container.ChangefeedSpecificationRepo,
		container.CDCInstanceTypeRepo,
		container.CDCMachineTypeRepo,
	)
	container.TiCDCMachineManager = ticdc_impl.NewMachineManagerRouter(
		container.CDCInstanceTypeModel,
	)
	container.RegionalTiCDCClusterModel = ticdc_impl.NewTiCDCClusterRegionalModel(
		container.TransProvider,
		container.CDCInstanceTypeModel,
		container.TiCDCClusterAPIClient,
		container.TiCDCClusterRepo,
		container.TiCDCCustomStorageRepo,
		container.TiCDCClusterConfigManager,
		container.RegionGrpcClient,
	)
	container.RegionalTiCDCChangefeedModel = ticdc_impl.NewChangefeedRegionalModel(
		container.TransProvider,
		container.ChangefeedAPIClient,
		container.TiCDCClusterAPIClient,
		container.TiCDCChangefeedRepo,
		container.TiCDCClusterRepo,
		container.BizChangefeedRepo,
		container.ChangefeedConfigManager,
	)
	container.RegionalTiCDCClusterWorker = ticdc_impl.NewTiCDCClusterRegionalWorker(
		container.TransProvider,
		container.CDCInstanceTypeModel,
		container.TiCDCClusterAPIClient,
		container.TiCDCClusterRepo,
		container.TiCDCBizChangefeedResourceRepo,
	)
	container.RegionalTiCDCChangefeedWorker = ticdc_impl.NewChangefeedRegionalWorker(
		container.ChangefeedAPIClient,
		container.TiCDCClusterRepo,
		container.TiCDCChangefeedRepo,
	)
	container.ChangefeedTrafficMeter = ticdc_impl.NewChangefeedTrafficMeter(
		container.TransProvider,
		container.InfraMetrics,
		container.KernelChangefeedTrafficRecordRepo,
	)

	// model need sub model , plz take care of the order
	{
		container.BizChangefeedDownstreamFactory = biz_changefeed_impl.NewDownstreamAdaptorFactory(
			container.TiCDCPreCheckAPIClient,
			container.Encryptor,
			container.RegionalTiCDCClusterModel,
		)
		container.BizChangefeedScheduler = scheduler.NewSharedBizChangefeedScheduler(
			container.TransProvider, container.BizChangefeedRepo, container.TiCDCBizChangefeedResourceRepo,
			container.RegionalTiCDCClusterModel, container.CDCInstanceTypeModel, container.MgmtSvcGrpcClient)
		container.BizChangefeedHooker = biz_changefeed_impl.NewBizChangefeedEventBusHooker(
			container.EventCenterPublisher,
		)
		sinkEndpointModel := sink_endpoint_impl.NewSinkEndpointImpl(
			container.TransProvider,
			container.TiCDCPrivateLinkAPIClient,
			container.BizSinkEndpointRepo,
			container.BizSinkEndpointUsageRepo,
			container.MgmtSvcGrpcClient,
			container.EventCenterPublisher,
		)
		container.SinkEndpointModel = sinkEndpointModel
		container.EndpointServiceAvailabilityZoneModel = sinkEndpointModel
		container.BizChangefeedModel = biz_changefeed_impl.NewBizChangefeedModel(
			container.TransProvider,
			container.BizChangefeedRepo,
			container.RegionalTiCDCChangefeedModel,
			container.TiCDCMachineManager,
			container.RegionalTiCDCClusterModel,
			container.MgmtSvcGrpcClient,
			container.WorkflowModel,
			billing.GetHandler(),
			container.BizChangefeedDownstreamFactory,
			container.BizChangefeedHooker,
			container.ClusterJobModel,
			container.SinkEndpointModel,
			container.TiCDCPreCheckAPIClient,
			container.FeatureGate,
		)
		container.BizChangefeedMeteringWorker = biz_changefeed_impl.NewBizChangefeedMeteringWorker(
			container.TiCDCMachineManager,
			container.BizChangefeedRepo,
			container.BillingRecordRepo,
			container.MgmtSvcGrpcClient)
		container.BizChangefeedMetricsWorker = biz_changefeed_impl.NewBizChangefeedMetricsWorker(
			container.RegionalTiCDCChangefeedModel,
			container.BizChangefeedRepo,
		)
	}

	container.SyncBillingRecord = billing_model_impl.NewSyncBillingRecord(
		container.TransProvider,
		container.BillingRecordRepo,
	)
	container.SinkEndpointWorker = sink_endpoint_impl.NewSinkEndpointWorkerImpl(
		container.TransProvider,
		container.TiCDCPrivateLinkAPIClient,
		container.BizSinkEndpointRepo,
		container.BizSinkEndpointUsageRepo,
		container.MgmtSvcGrpcClient,
		container.ChangefeedTrafficMeter,
		container.KernelChangefeedTrafficRecordRepo,
		container.BillingRecordRepo,
		container.BizChangefeedRepo,
		container.EventCenterPublisher,
	)
	container.ImportIntoModel = *import_into_model.NewImportIntoModel(
		container.MgmtSvcGrpcClient,
		container.Encryptor,
		container.NGRegionalClientProvider,
		container.FeatureGate,
		container.EventCenterPublisher,
		container.ClusterJobModel,
	)
}

func initPremiumDM(container *GlobalContainer) {
	cfg := service_config.Global()
	agentCfg := agent.ClientConfig{
		MgmtSvcProxyAddr:  cfg.GrpcConfig.MgmtSvcProxyAddr,
		MgmtSvcProxyToken: cfg.GrpcConfig.MgmtSvcProxyToken,
	}
	dao := premium_dm_model.NewPremiumDMTaskDAO(container.TransProvider)
	taskWorker := premium_dm_model.NewPremiumDMTaskWorker(
		dao, container.MgmtSvcGrpcClient, container.Encryptor,
		agentCfg, container.SinkEndpointModel, container.BizSinkEndpointRepo,
		container.FeatureGate,
		id_generator.MustNewSnowflakeIDGenerator(id_generator.NewHostSnowflakeMachineIDGenerator()),
		container.EventCenterPublisher,
	)
	precheckWorker := premium_dm_model.NewPremiumDMPrecheckWorker(dao, taskWorker)

	container.PremiumDMDAO = dao
	container.PremiumDMWorker = taskWorker
	container.PremiumDMPrecheckWorker = precheckWorker
	container.PremiumDMAPI = premium_dm_controller.NewPremiumDMService(
		dao, taskWorker, container.MgmtSvcGrpcClient,
		container.Encryptor, agentCfg, container.TransProvider,
		container.SinkEndpointModel, container.BizSinkEndpointRepo,
		container.EventCenterPublisher,
	)
}

func newTransProvider(handle *gorm.DB) db.TransProvider {
	return db.NewProvider(handle)
}