package app

import (
	"context"
	"fmt"
	"io"
	"os"

	"github.com/tidbcloud/cluster-service-ng/pkg/autoscale/framework"
	"github.com/tidbcloud/cluster-service-ng/pkg/autoscale/modules/input_k8s_v1"
	input_o11y_metrics_v1 "github.com/tidbcloud/cluster-service-ng/pkg/autoscale/modules/input_o11y_v1"
	"github.com/tidbcloud/cluster-service-ng/pkg/autoscale/modules/input_tikv_capacity_v1"
	"github.com/tidbcloud/cluster-service-ng/pkg/autoscale/modules/output_blackhole"
	"github.com/tidbcloud/cluster-service-ng/pkg/autoscale/modules/output_capacity_reservation_v1"
	"github.com/tidbcloud/cluster-service-ng/pkg/autoscale/modules/output_direct_hpa_v1"
	scaler_k8s_v1 "github.com/tidbcloud/cluster-service-ng/pkg/autoscale/modules/scaler_k8s_res_based_v1"
	scaler_o11y_res_based_v1 "github.com/tidbcloud/cluster-service-ng/pkg/autoscale/modules/scaler_o11y_res_based_v1"
	"github.com/tidbcloud/cluster-service-ng/pkg/autoscale/modules/scaler_tikv_capacity_v1"
	"github.com/tidbcloud/cluster-service-ng/pkg/dataplane"
)

// Dependencies holds external dependencies that can be injected into the
// autoscale app. All fields are optional — nil means the capability is
// unavailable (e.g. EBS cooldown check will be skipped).
type Dependencies struct {
	DataplaneManager dataplane.Manager
}

type App struct {
	executor *framework.ClusterScalerExecutor
	deps     Dependencies
}

func New(logger io.Writer) (*App, error) {
	if logger == nil {
		logger = os.Stderr
	}

	fwLogger := framework.NewGeneralLogger("autoscale", logger)
	promReceiver := framework.NewPrometheusEventReceiver()
	composite := framework.NewMultiEventReceiver(fwLogger, promReceiver)

	executor, err := framework.NewClusterScalerExecutor(composite)
	if err != nil {
		return nil, fmt.Errorf("failed to create cluster scaler executor: %w", err)
	}

	app := &App{executor: executor}

	if err := app.registerComponents(); err != nil {
		return nil, fmt.Errorf("failed to register components: %w", err)
	}

	if err := app.registerScalingRoutes(); err != nil {
		return nil, fmt.Errorf("failed to register scaling routes: %w", err)
	}

	if err := app.registerScalingExecutor(); err != nil {
		return nil, fmt.Errorf("failed to register executor: %w", err)
	}

	return app, nil
}

func NewWithDependencies(logger io.Writer, deps Dependencies) (*App, error) {
	if logger == nil {
		logger = os.Stderr
	}

	fwLogger := framework.NewGeneralLogger("autoscale", logger)
	promReceiver := framework.NewPrometheusEventReceiver()
	composite := framework.NewMultiEventReceiver(fwLogger, promReceiver)

	executor, err := framework.NewClusterScalerExecutor(composite)
	if err != nil {
		return nil, fmt.Errorf("failed to create cluster scaler executor: %w", err)
	}

	app := &App{executor: executor, deps: deps}

	if err := app.registerComponents(); err != nil {
		return nil, fmt.Errorf("failed to register components: %w", err)
	}

	if err := app.registerScalingRoutes(); err != nil {
		return nil, fmt.Errorf("failed to register scaling routes: %w", err)
	}

	if err := app.registerScalingExecutor(); err != nil {
		return nil, fmt.Errorf("failed to register executor: %w", err)
	}

	return app, nil
}

func (a *App) registerComponents() error {
	clusterScalerManager := a.executor.GetClusterScalerManager()

	outputManager := clusterScalerManager.GetOutputManager()
	err := outputManager.Register(
		output_blackhole.OutputBlackholeMeta,
		output_blackhole.NewOutputBlackhole,
	)
	if err != nil {
		return fmt.Errorf("failed to register blackhole output: %w", err)
	}

	err = outputManager.Register(
		output_direct_hpa_v1.OutputDirectHPAV1Meta,
		output_direct_hpa_v1.NewOutputDirectHPAV1,
	)
	if err != nil {
		return fmt.Errorf("failed to register HPA output: %w", err)
	}

	err = outputManager.Register(
		output_capacity_reservation_v1.OutputCapacityReservationV1Meta,
		output_capacity_reservation_v1.NewOutputCapacityReservationV1,
	)
	if err != nil {
		return fmt.Errorf("failed to register capacity reservation output: %w", err)
	}

	// Register capacity reservation output with V2 meta — same creator, advertises OutputInterfaceV2.
	err = outputManager.Register(
		output_capacity_reservation_v1.OutputCapacityReservationV2Meta,
		output_capacity_reservation_v1.NewOutputCapacityReservationV1,
	)
	if err != nil {
		return fmt.Errorf("failed to register capacity reservation output (V2): %w", err)
	}

	inputManager := clusterScalerManager.GetInputProviderManager()
	err = inputManager.Register(
		input_k8s_v1.InputK8sV1Meta,
		input_k8s_v1.NewInputK8sV1,
	)
	if err != nil {
		return fmt.Errorf("failed to register K8s input provider: %w", err)
	}

	err = inputManager.Register(
		input_o11y_metrics_v1.InputO11yMetricsV1Meta,
		input_o11y_metrics_v1.NewInputO11yMetricsV1,
	)
	if err != nil {
		return fmt.Errorf("failed to register O11y input provider: %w", err)
	}

	err = inputManager.Register(
		input_tikv_capacity_v1.InputTiKVCapacityV1Meta,
		input_tikv_capacity_v1.NewInputTiKVCapacityV1,
	)
	if err != nil {
		return fmt.Errorf("failed to register TiKV capacity input provider: %w", err)
	}

	// Register TiKV input with V2 meta — same creator, advertises InputInterfaceV2.
	err = inputManager.Register(
		input_tikv_capacity_v1.InputTiKVCapacityV2Meta,
		input_tikv_capacity_v1.NewInputTiKVCapacityV1,
	)
	if err != nil {
		return fmt.Errorf("failed to register TiKV capacity input provider (V2): %w", err)
	}

	phasedComponentScalerManager := clusterScalerManager.GetPhasedComponentScalerManager()
	err = phasedComponentScalerManager.Register(
		scaler_o11y_res_based_v1.ScalerO11yResBasedV1IdStr,
		func(
			eventReceiver framework.EventReceiver,
			clusterId framework.ClusterId,
			componentName framework.ComponentName,
			config framework.ConfigData,
			pInputInterface any,
			pOutputInterface any) (any, error) {

			input, ok := pInputInterface.(framework.InputInterfaceV1)
			if !ok {
				return nil, fmt.Errorf("input interface unmatched")
			}
			output, ok := pOutputInterface.(framework.OutputInterfaceV1)
			if !ok {
				return nil, fmt.Errorf("output interface unmatched")
			}
			cfg, ok := config.(*scaler_o11y_res_based_v1.ScalerO11yResBasedV1Config)
			if !ok || cfg == nil {
				cfg = scaler_o11y_res_based_v1.ScalerO11yResBasedV1DefaultConfig
			}
			return scaler_o11y_res_based_v1.NewScalerO11yResBasedV1(eventReceiver, clusterId, componentName, cfg, input, output)
		})
	if err != nil {
		return err
	}

	err = phasedComponentScalerManager.Register(
		scaler_k8s_v1.ScalerK8sV1IdStr,
		func(
			eventReceiver framework.EventReceiver,
			clusterId framework.ClusterId,
			componentName framework.ComponentName,
			config framework.ConfigData,
			pInputInterface any,
			pOutputInterface any) (any, error) {

			input, ok := pInputInterface.(framework.InputInterfaceV1)
			if !ok {
				return nil, fmt.Errorf("input interface unmatched")
			}
			output, ok := pOutputInterface.(framework.OutputInterfaceV1)
			if !ok {
				return nil, fmt.Errorf("output interface unmatched")
			}
			cfg, ok := config.(*scaler_k8s_v1.ScalerK8sV1Config)
			if !ok || cfg == nil {
				cfg = scaler_k8s_v1.ScalerK8sV1DefaultConfig
			}
			return scaler_k8s_v1.NewScalerK8sV1(eventReceiver, clusterId, componentName, cfg, input, output)
		})
	if err != nil {
		return err
	}

	err = phasedComponentScalerManager.Register(
		scaler_tikv_capacity_v1.ScalerTiKVCapacityV1IdStr,
		func(
			eventReceiver framework.EventReceiver,
			clusterId framework.ClusterId,
			componentName framework.ComponentName,
			config framework.ConfigData,
			pInputInterface any,
			pOutputInterface any) (any, error) {

			input, ok := pInputInterface.(framework.InputInterfaceV2)
			if !ok {
				return nil, fmt.Errorf("input interface unmatched: expected InputInterfaceV2")
			}
			output, ok := pOutputInterface.(framework.OutputInterfaceV2)
			if !ok {
				return nil, fmt.Errorf("output interface unmatched: expected OutputInterfaceV2")
			}
			return scaler_tikv_capacity_v1.NewScalerTiKVCapacityV1(eventReceiver, clusterId, componentName, config, input, output)
		})
	if err != nil {
		return err
	}

	return nil
}

const (
	ScalingRouteK8s2BlackholeByResourceUsageBasedType  framework.ClusterScalerType = "ScalingRouteK8s2BlackholeByResourceUsageBased"
	ScalingRouteK8s2HPAByResourceUsageBasedType        framework.ClusterScalerType = "ScalingRouteK8s2HPAByResourceUsageBased"
	ScalingRouteK8s2BlackholeByCPUUsageBasedType       framework.ClusterScalerType = "ScalingRouteK8s2BlackholeByCPUUsageBased"
	ScalingRouteK8s2HPAByCPUUsageBasedType             framework.ClusterScalerType = "ScalingRouteK8s2HPAByCPUUsageBased"
	ScalingRouteO11y2BlackholeByResourceUsageBasedType framework.ClusterScalerType = "ScalingRouteO11y2BlackholeByResourceUsageBased"
	ScalingRouteO11y2HPAByResourceUsageBasedType       framework.ClusterScalerType = "ScalingRouteO11y2HPAByResourceUsageBased"
	ScalingRouteTiKVCapacity2ReservationType           framework.ClusterScalerType = "ScalingRouteTiKVCapacity2Reservation"
)

func (a *App) registerScalingRoutes() error {
	clusterScalerManager := a.executor.GetClusterScalerManager()

	scalingRouteK8s2BlackholeByResourceUsageBased := &framework.ClusterScalerMeta{
		ScalerMeta: framework.UnattachedScalerMeta{
			Type:             ScalingRouteK8s2BlackholeByResourceUsageBasedType,
			InterfaceVersion: framework.ClusterScalerInterfaceV1IdStr,
		},
		InputMeta:  input_k8s_v1.InputK8sV1Meta,
		OutputMeta: output_blackhole.OutputBlackholeMeta,
	}
	err := clusterScalerManager.Register(scalingRouteK8s2BlackholeByResourceUsageBased, a.createPhasedComponentScalerCreator(scaler_o11y_res_based_v1.ScalerO11yResBasedV1IdStr))
	if err != nil {
		return err
	}

	scalingRouteK8s2HPAByResourceUsageBased := &framework.ClusterScalerMeta{
		ScalerMeta: framework.UnattachedScalerMeta{
			Type:             ScalingRouteK8s2HPAByResourceUsageBasedType,
			InterfaceVersion: framework.ClusterScalerInterfaceV1IdStr,
		},
		InputMeta:  input_k8s_v1.InputK8sV1Meta,
		OutputMeta: output_direct_hpa_v1.OutputDirectHPAV1Meta,
	}
	err = clusterScalerManager.Register(scalingRouteK8s2HPAByResourceUsageBased, a.createPhasedComponentScalerCreator(scaler_o11y_res_based_v1.ScalerO11yResBasedV1IdStr))
	if err != nil {
		return err
	}

	scalingRouteK8s2BlackholeByCPUUsageBased := &framework.ClusterScalerMeta{
		ScalerMeta: framework.UnattachedScalerMeta{
			Type:             ScalingRouteK8s2BlackholeByCPUUsageBasedType,
			InterfaceVersion: framework.ClusterScalerInterfaceV1IdStr,
		},
		InputMeta:  input_k8s_v1.InputK8sV1Meta,
		OutputMeta: output_blackhole.OutputBlackholeMeta,
	}
	err = clusterScalerManager.Register(scalingRouteK8s2BlackholeByCPUUsageBased, a.createPhasedComponentScalerCreator(scaler_k8s_v1.ScalerK8sV1IdStr))
	if err != nil {
		return err
	}

	scalingRouteK8s2HPAByCPUUsageBased := &framework.ClusterScalerMeta{
		ScalerMeta: framework.UnattachedScalerMeta{
			Type:             ScalingRouteK8s2HPAByCPUUsageBasedType,
			InterfaceVersion: framework.ClusterScalerInterfaceV1IdStr,
		},
		InputMeta:  input_k8s_v1.InputK8sV1Meta,
		OutputMeta: output_direct_hpa_v1.OutputDirectHPAV1Meta,
	}
	err = clusterScalerManager.Register(scalingRouteK8s2HPAByCPUUsageBased, a.createPhasedComponentScalerCreator(scaler_k8s_v1.ScalerK8sV1IdStr))
	if err != nil {
		return err
	}

	scalingRouteO11y2BlackholeByResourceUsageBased := &framework.ClusterScalerMeta{
		ScalerMeta: framework.UnattachedScalerMeta{
			Type:             ScalingRouteO11y2BlackholeByResourceUsageBasedType,
			InterfaceVersion: framework.ClusterScalerInterfaceV1IdStr,
		},
		InputMeta:  input_o11y_metrics_v1.InputO11yMetricsV1Meta,
		OutputMeta: output_blackhole.OutputBlackholeMeta,
	}
	err = clusterScalerManager.Register(scalingRouteO11y2BlackholeByResourceUsageBased, a.createPhasedComponentScalerCreator(scaler_o11y_res_based_v1.ScalerO11yResBasedV1IdStr))
	if err != nil {
		return err
	}

	scalingRouteO11y2HPAByResourceUsageBased := &framework.ClusterScalerMeta{
		ScalerMeta: framework.UnattachedScalerMeta{
			Type:             ScalingRouteO11y2HPAByResourceUsageBasedType,
			InterfaceVersion: framework.ClusterScalerInterfaceV1IdStr,
		},
		InputMeta:  input_o11y_metrics_v1.InputO11yMetricsV1Meta,
		OutputMeta: output_direct_hpa_v1.OutputDirectHPAV1Meta,
	}
	err = clusterScalerManager.Register(scalingRouteO11y2HPAByResourceUsageBased, a.createPhasedComponentScalerCreator(scaler_o11y_res_based_v1.ScalerO11yResBasedV1IdStr))
	if err != nil {
		return err
	}

	scalingRouteTiKVCapacity2Reservation := &framework.ClusterScalerMeta{
		ScalerMeta: framework.UnattachedScalerMeta{
			Type:             ScalingRouteTiKVCapacity2ReservationType,
			InterfaceVersion: framework.ClusterScalerInterfaceV2IdStr,
		},
		InputMeta:  input_tikv_capacity_v1.InputTiKVCapacityV2Meta,
		OutputMeta: output_capacity_reservation_v1.OutputCapacityReservationV2Meta,
	}
	err = clusterScalerManager.Register(scalingRouteTiKVCapacity2Reservation, a.createPhasedComponentScalerV2Creator(scaler_tikv_capacity_v1.ScalerTiKVCapacityV1IdStr))
	if err != nil {
		return err
	}

	return nil
}

func (a *App) createPhasedComponentScalerCreator(componentScalerType framework.PhasedComponentScalerType) framework.ClusterScalerCreator {
	return func(eventReceiver framework.EventReceiver,
		clusterId framework.ClusterId,
		config framework.ConfigData,
		pInputInterface,
		pOutputInterface any) (any, error) {

		input, ok := pInputInterface.(framework.InputInterfaceV1)
		if !ok {
			return nil, fmt.Errorf("input interface unmatched")
		}
		output, ok := pOutputInterface.(framework.OutputInterfaceV1)
		if !ok {
			return nil, fmt.Errorf("output interface unmatched")
		}
		if config == nil {
			return nil, fmt.Errorf("cluster scaler config is empty")
		}
		clusterScalerConfig, ok := config.(*framework.ClusterScalerPhasedComponentScalerV1Config)
		if !ok {
			return nil, fmt.Errorf("cluster scaler config type unmatched")
		}
		componentScaler, err := a.executor.GetClusterScalerManager().GetPhasedComponentScalerManager().NewPhasedComponentScaler(
			componentScalerType,
			clusterId,
			clusterScalerConfig.ComponentName,
			clusterScalerConfig.ComponentScalerConfig,
			pInputInterface,
			pOutputInterface)
		if err != nil {
			return nil, err
		}
		scaler := framework.NewClusterScalerPhasedComponentScalerV1(
			eventReceiver,
			clusterId,
			input,
			output,
			componentScaler.(framework.PhasedComponentScalerV1),
			clusterScalerConfig.ComponentName)
		return scaler, nil
	}
}

func (a *App) createPhasedComponentScalerV2Creator(componentScalerType framework.PhasedComponentScalerType) framework.ClusterScalerCreator {
	return func(eventReceiver framework.EventReceiver,
		clusterId framework.ClusterId,
		config framework.ConfigData,
		pInputInterface,
		pOutputInterface any) (any, error) {

		if config == nil {
			return nil, fmt.Errorf("cluster scaler config is empty")
		}
		clusterScalerConfig, ok := config.(*framework.ClusterScalerPhasedComponentScalerV1Config)
		if !ok {
			return nil, fmt.Errorf("cluster scaler config type unmatched")
		}
		componentScaler, err := a.executor.GetClusterScalerManager().GetPhasedComponentScalerManager().NewPhasedComponentScaler(
			componentScalerType,
			clusterId,
			clusterScalerConfig.ComponentName,
			clusterScalerConfig.ComponentScalerConfig,
			pInputInterface,
			pOutputInterface)
		if err != nil {
			return nil, err
		}
		scaler := framework.NewClusterScalerPhasedComponentScalerV2(
			eventReceiver,
			clusterId,
			componentScaler.(framework.PhasedComponentScalerV2),
			clusterScalerConfig.ComponentName)
		return scaler, nil
	}
}

func (a *App) registerScalingExecutor() error {
	executorManager := a.executor.GetScalingTaskExecutorManager()

	for _, scalerType := range []framework.ClusterScalerType{
		ScalingRouteK8s2BlackholeByResourceUsageBasedType,
		ScalingRouteK8s2HPAByResourceUsageBasedType,
		ScalingRouteK8s2BlackholeByCPUUsageBasedType,
		ScalingRouteK8s2HPAByCPUUsageBasedType,
		ScalingRouteO11y2BlackholeByResourceUsageBasedType,
		ScalingRouteO11y2HPAByResourceUsageBasedType,
	} {
		meta := &framework.ScalingTaskExecutorMeta{
			ScalerType:             scalerType,
			ScalerInterfaceVersion: framework.ClusterScalerInterfaceV1IdStr,
			InputInterfaceVersion:  framework.InputInterfaceV1IdStr,
			OutputInterfaceVersion: framework.OutputInterfaceV1IdStr,
			ExecutorName:           framework.ScalerInterfaceV1ExecutorSyncLoopIdStr,
		}

		err := executorManager.Register(meta, func(
			eventReceiver framework.EventReceiver,
			clusterId framework.ClusterId,
			pScaler any,
			pInput any,
			pOutput any,
			executorConfig framework.ConfigData,
		) (framework.ScalingTaskExecutor, error) {
			scaler, ok := pScaler.(framework.ClusterScalerInterfaceV1)
			if !ok {
				return nil, fmt.Errorf("scaler interface unmatched")
			}
			input, ok := pInput.(framework.InputInterfaceV1)
			if !ok {
				return nil, fmt.Errorf("input interface unmatched")
			}
			output, ok := pOutput.(framework.OutputInterfaceV1)
			if !ok {
				return nil, fmt.Errorf("output interface unmatched")
			}
			cfg, ok := executorConfig.(*framework.ScalerInterfaceV1ExecutorSyncLoopConfig)
			if !ok || cfg == nil {
				return nil, fmt.Errorf("executor config unmatched")
			}
			return framework.NewScalerInterfaceV1ExecutorSyncLoop(
				eventReceiver,
				clusterId,
				scaler,
				input,
				output,
				cfg,
			), nil
		})
		if err != nil {
			return err
		}
	}

	// Register V2 executor for TiKV capacity route.
	{
		meta := &framework.ScalingTaskExecutorMeta{
			ScalerType:             ScalingRouteTiKVCapacity2ReservationType,
			ScalerInterfaceVersion: framework.ClusterScalerInterfaceV2IdStr,
			InputInterfaceVersion:  framework.InputInterfaceV2IdStr,
			OutputInterfaceVersion: framework.OutputInterfaceV2IdStr,
			ExecutorName:           framework.ScalerInterfaceV2ExecutorSyncLoopIdStr,
		}
		err := executorManager.Register(meta, func(
			eventReceiver framework.EventReceiver,
			clusterId framework.ClusterId,
			pScaler any,
			pInput any,
			pOutput any,
			executorConfig framework.ConfigData,
		) (framework.ScalingTaskExecutor, error) {
			scaler, ok := pScaler.(framework.ClusterScalerInterfaceV2)
			if !ok {
				return nil, fmt.Errorf("scaler interface unmatched: expected ClusterScalerInterfaceV2")
			}
			input, ok := pInput.(framework.InputInterfaceV2)
			if !ok {
				return nil, fmt.Errorf("input interface unmatched: expected InputInterfaceV2")
			}
			output, ok := pOutput.(framework.OutputInterfaceV2)
			if !ok {
				return nil, fmt.Errorf("output interface unmatched: expected OutputInterfaceV2")
			}
			cfg, ok := executorConfig.(*framework.ScalerInterfaceV1ExecutorSyncLoopConfig)
			if !ok || cfg == nil {
				return nil, fmt.Errorf("executor config unmatched")
			}
			return framework.NewScalerInterfaceV2ExecutorSyncLoop(
				eventReceiver,
				clusterId,
				scaler,
				input,
				output,
				cfg,
			), nil
		})
		if err != nil {
			return err
		}
	}

	return nil
}

func (a *App) Run(waitForFinish bool) error {
	if a.executor == nil {
		return fmt.Errorf("app not initialized, call New() first")
	}
	return a.executor.Run(waitForFinish)
}

// LoadBaseConfig loads input, output, and scaler configs (but NOT clusterScalers).
// This prepares the app for dynamic clusterScaler management via RunOneClusterScaler.
func (a *App) LoadBaseConfig(cfg *AutoscaleConfig) (scalerRefs map[string]ScalerRefInfo, err error) {
	if a.executor == nil {
		return nil, fmt.Errorf("app not initialized, call New() first")
	}

	inputTypes := make(map[string]string)
	for _, input := range cfg.Inputs {
		ic, err := ConvertToInputProviderConfig(input)
		if err != nil {
			return nil, fmt.Errorf("failed to convert input config %s: %w", input.UniqName, err)
		}
		a.injectDepsIntoInputConfig(&ic)
		a.executor.AddInputConfig(ic)
		inputTypes[input.UniqName] = input.Type
	}

	outputTypes := make(map[string]string)
	for _, output := range cfg.Outputs {
		oc, err := ConvertToOutputConfig(output)
		if err != nil {
			return nil, fmt.Errorf("failed to convert output config %s: %w", output.UniqName, err)
		}
		a.executor.AddOutputConfig(oc)
		outputTypes[output.UniqName] = output.Type
	}

	refs := make(map[string]ScalerRefInfo)
	for _, scaler := range cfg.Scalers {
		inputType, ok := inputTypes[scaler.InputUniqName]
		if !ok {
			return nil, fmt.Errorf("input %s referenced by scaler %s not found", scaler.InputUniqName, scaler.UniqName)
		}
		outputType, ok := outputTypes[scaler.OutputUniqName]
		if !ok {
			return nil, fmt.Errorf("output %s referenced by scaler %s not found", scaler.OutputUniqName, scaler.UniqName)
		}
		sc, err := ConvertToScalerConfig(&scaler, inputType, outputType)
		if err != nil {
			return nil, fmt.Errorf("failed to convert scaler config %s: %w", scaler.UniqName, err)
		}
		a.executor.AddScalerConfig(sc)
		refs[scaler.UniqName] = ScalerRefInfo{
			ScalerType: scaler.Type,
			InputType:  inputType,
			OutputType: outputType,
		}
	}
	return refs, nil
}

// ScalerRefInfo holds resolved type info for a scaler, used when converting
// ClusterScalerConfig to the framework representation.
type ScalerRefInfo struct {
	ScalerType string
	InputType  string
	OutputType string
}

// RunOneClusterScaler dynamically creates and starts a single clusterScaler goroutine.
// Returns a cancel function to stop it. The caller must provide scalerRefs obtained
// from LoadBaseConfig.
func (a *App) RunOneClusterScaler(
	cfg *ClusterScalerConfig,
	refs map[string]ScalerRefInfo,
) (cancel context.CancelFunc, done <-chan struct{}, err error) {
	refInfo, ok := refs[cfg.ScalerUniqName]
	if !ok {
		return nil, nil, fmt.Errorf("scaler %s not found in scalerRefs", cfg.ScalerUniqName)
	}
	csc, err := ConvertToClusterScalerConfig(cfg, refInfo.ScalerType, refInfo.InputType, refInfo.OutputType)
	if err != nil {
		return nil, nil, fmt.Errorf("failed to convert cluster scaler config: %w", err)
	}
	return a.executor.RunOne(csc)
}

func (a *App) LoadConfig(cfg *AutoscaleConfig) error {
	if a.executor == nil {
		return fmt.Errorf("app not initialized, call New() first")
	}

	inputTypes := make(map[string]string)
	for _, input := range cfg.Inputs {
		ic, err := ConvertToInputProviderConfig(input)
		if err != nil {
			return fmt.Errorf("failed to convert input config %s: %w", input.UniqName, err)
		}
		a.injectDepsIntoInputConfig(&ic)
		a.executor.AddInputConfig(ic)
		inputTypes[input.UniqName] = input.Type
	}

	outputTypes := make(map[string]string)
	for _, output := range cfg.Outputs {
		oc, err := ConvertToOutputConfig(output)
		if err != nil {
			return fmt.Errorf("failed to convert output config %s: %w", output.UniqName, err)
		}
		a.executor.AddOutputConfig(oc)
		outputTypes[output.UniqName] = output.Type
	}

	scalerRefs := make(map[string]ScalerRefInfo)
	for _, scaler := range cfg.Scalers {
		inputType, ok := inputTypes[scaler.InputUniqName]
		if !ok {
			return fmt.Errorf("input %s referenced by scaler %s not found", scaler.InputUniqName, scaler.UniqName)
		}
		outputType, ok := outputTypes[scaler.OutputUniqName]
		if !ok {
			return fmt.Errorf("output %s referenced by scaler %s not found", scaler.OutputUniqName, scaler.UniqName)
		}
		sc, err := ConvertToScalerConfig(&scaler, inputType, outputType)
		if err != nil {
			return fmt.Errorf("failed to convert scaler config %s: %w", scaler.UniqName, err)
		}
		a.executor.AddScalerConfig(sc)
		scalerRefs[scaler.UniqName] = ScalerRefInfo{
			ScalerType: scaler.Type,
			InputType:  inputType,
			OutputType: outputType,
		}
	}

	for _, clusterScaler := range cfg.ClusterScalers {
		refInfo, ok := scalerRefs[clusterScaler.ScalerUniqName]
		if !ok {
			return fmt.Errorf("scaler %s not found in scalers config", clusterScaler.ScalerUniqName)
		}
		csc, err := ConvertToClusterScalerConfig(&clusterScaler, refInfo.ScalerType, refInfo.InputType, refInfo.OutputType)
		if err != nil {
			return fmt.Errorf("failed to convert cluster scaler config %s: %w", clusterScaler.ScalerUniqName, err)
		}
		a.executor.AddClusterScalerConfig(csc)
	}

	return nil
}

func (a *App) injectDepsIntoInputConfig(ic *framework.InputProviderConfig) {
	if a.deps.DataplaneManager == nil {
		return
	}
	if tikvCfg, ok := ic.Config.(*input_tikv_capacity_v1.InputTiKVCapacityV1Config); ok {
		if tikvCfg.DataplaneManager == nil {
			tikvCfg.DataplaneManager = a.deps.DataplaneManager
		}
	}
}

func Run(cfg *AutoscaleConfig) error {
	autoscaleApp, err := New(os.Stderr)
	if err != nil {
		return fmt.Errorf("failed to create autoscale app: %w", err)
	}

	if err := autoscaleApp.LoadConfig(cfg); err != nil {
		return fmt.Errorf("failed to load autoscale config: %w", err)
	}

	return autoscaleApp.Run(true)
}

func RunWithDependencies(cfg *AutoscaleConfig, deps Dependencies) error {
	autoscaleApp, err := NewWithDependencies(os.Stderr, deps)
	if err != nil {
		return fmt.Errorf("failed to create autoscale app: %w", err)
	}

	if err := autoscaleApp.LoadConfig(cfg); err != nil {
		return fmt.Errorf("failed to load autoscale config: %w", err)
	}

	return autoscaleApp.Run(true)
}