package main

import (
	"context"
	"encoding/base64"
	"encoding/json"
	"flag"
	"fmt"
	"path/filepath"
	"strings"

	kms "github.com/alibabacloud-go/kms-20160120/v3/client"
	"github.com/aliyun/credentials-go/credentials"
	"github.com/aws/aws-sdk-go-v2/aws"
	awsconfig "github.com/aws/aws-sdk-go-v2/config"
	"github.com/aws/aws-sdk-go-v2/service/secretsmanager"
	"github.com/go-logr/zapr"
	"github.com/tidbcloud/observability/pkg/regionalservice"
	gcpsecretmanager "google.golang.org/api/secretmanager/v1"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/tools/clientcmd"

	"github.com/pingcap/log"
	"github.com/tidbcloud/observability/pkg/resourcemanager"
	"k8s.io/client-go/util/homedir"

	aliopenapi "github.com/alibabacloud-go/darabonba-openapi/v2/client"
	teautil "github.com/alibabacloud-go/tea-utils/v2/service"
	"github.com/tidbcloud/observability/pkg/common/configs"
	"github.com/tidbcloud/observability/pkg/common/constants"
	"github.com/tidbcloud/observability/pkg/common/errors"
	"github.com/tidbcloud/observability/pkg/common/metrics"
	"github.com/tidbcloud/observability/pkg/common/structs"
	"github.com/tidbcloud/observability/pkg/common/utils"
	"github.com/tidbcloud/observability/pkg/diagcontroller"
	diagconfig "github.com/tidbcloud/observability/pkg/diagcontroller/config"
	myclient "github.com/tidbcloud/observability/pkg/mysql/client"
	"github.com/tidbcloud/observability/pkg/regionalapi"
	"github.com/tidbcloud/observability/pkg/resourceworker/providers"
	"github.com/tidbcloud/observability/pkg/resourceworker/worker"
	"go.uber.org/zap"
	controller_metrics "sigs.k8s.io/controller-runtime/pkg/metrics"
)

// @title Observability RegionalServer API
// @version 1.0
// @description Observability RegionalServer API

// @contact.name zhangpeijin
// @contact.email zhangpeijin@pingcap.com

// @license.name Apache 2.0
// @license.url http://www.apache.org/licenses/LICENSE-2.0.html

// @host localhost:8081
// @BasePath /api/v1/
// @query.collection.format multi
func main() {
	ctx := utils.SetupSignalHandler()

	flag.IntVar(&port, "port", constants.DefaultObservabilityRegionalServerPort, "listening port")
	flag.IntVar(&metricsPort, "metrics-port", constants.DefaultObservabilityRegionalServerMetricsPort, "listening metrics port")
	flag.Float64Var(&defaultOversell, "default-oversell", 1, "enable default oversell if not set to zero")
	flag.StringVar(&vendor, "vendor", string(constants.AWS), "vendor")
	flag.BoolVar(&clientMode, "client-mode", true, "client mode (true: in-cluster, false: out-of-cluster)")
	flag.StringVar(&selfMonitorEndpoint, "selfMonitorEndpoint", "http://prometheus-operated.self-monitoring:9090", "self monitoring prometheus endpoint")
	// specify the kube config absolute path, required if the clientMode value is false.
	flag.StringVar(&kubeConfigPath, "kube-config-path", filepath.Join(homedir.HomeDir(), ".kube", "config"), "(optional) absolute path to the kube config file")
	flag.StringVar(&logLevel, "log-level", "info", "log level. default: info")
	flag.StringVar(&logDir, "log-dir", "./logs", "log dir. default: ./logs")
	flag.StringVar(&azConfigPath, "azconfig", "./config/azconfig.yaml", "azconfig file path. default: ./config/azconfig.yaml")
	flag.StringVar(&bizTypeConfigPath, "bizTypeConfigPath", "./config/biztypeconfig.yaml", "bizType config file path. default: ./config/biztypeconfig.yaml")
	flag.StringVar(&monitorSpecPath, "monitorSpec", "./config/monitor_spec.yaml", "available specs config path. default: ./config/monitor_spec.yaml")
	flag.StringVar(&accountInfoPath, "accountInfo", "./config/accountinfo.yaml", "account info config path. default: ./config/accountinfo.yaml")
	flag.StringVar(&alertManagerInfoPath, "alertManagerInfo", "./config/alertmanagerinfo.yaml", "alertmanager info config path. default: ./config/alertmanagerinfo.yaml")
	flag.StringVar(&vmConfigPath, "vmConfigPath", "./config/vmconfig.yaml", "vm info config path. default: ./config/vmconfig.yaml")
	flag.StringVar(&syncAlertRuleConfigPath, "syncAlertRuleConfig", "./config/syncalertruleconfig.yaml", "syncalertruleconfig info config path. default: ./config/syncalertruleconfig.yaml")
	flag.StringVar(&snsConfigPath, "snsConfigPath", "./config/snsconfig.yaml", "sns info config path. default: ./config/snsconfig.yaml")
	flag.StringVar(&overlayAlertRuleCacheConfigPath, "alertrulecacheconfig", "./config/alertrulecacheconfig.yaml", "alertrulecacheconfig info config path. default: ./config/alertrulecacheconfig.yaml")
	flag.StringVar(&diagControllerConfigPath, "diagControllerConfig", "./config/diagcontrollerconfig.yaml", "diagcontrollerconfig info config path. default: ./config/diagcontrollerconfig.yaml")
	flag.StringVar(&controllerConfigPath, "controllerConfig", "./config/controllerconfig.yaml", "controllerconfig info config path. default: ./config/controllerconfig.yaml")
	flag.StringVar(&byocVPCPeeringConfigPath, "byocVPCPeeringConfig", "./config/byocvpcpeeringconfig.yaml", "BYOC VPC peering route config path. default: ./config/byocvpcpeeringconfig.yaml")
	flag.Parse()
	// init log
	utils.InitLog(logLevel)
	log.Info("regional server running in vendor", zap.String("Vendor", vendor))
	// If clientMode is false (out-of-cluster) and the kubeConfigPath file does not exist, then exit
	if !clientMode && !utils.IsFile(kubeConfigPath) {
		log.Panic("please check if parameter `client-mode` or `kube-config-path` is correct.")
	}

	// build kube config
	var err error
	var kubeConfig *rest.Config
	if clientMode {
		// creates the in-cluster config
		kubeConfig, err = rest.InClusterConfig()
	} else {
		// creates the out-of-cluster config
		kubeConfig, err = clientcmd.BuildConfigFromFlags("", kubeConfigPath)
	}
	if err != nil {
		log.Panic("init kubeconfig failed", zap.Error(err))
	}
	//Load AZconfig from file
	azconfig := configs.LoadAZConfig(azConfigPath)
	bizTypeConfigs := configs.LoadBizTypeConfigs(bizTypeConfigPath)
	monitorSpec := configs.LoadMonitorSpec(monitorSpecPath)
	accountInfo := configs.LoadAccountInfo(accountInfoPath)
	vmConfig := configs.LoadVMConfig(vmConfigPath)
	alertmanagerInfo := configs.LoadAlertManagerConfig(alertManagerInfoPath)
	syncAlertRuleConfig := configs.LoadSyncAlertRuleConfig(syncAlertRuleConfigPath)
	alertRuleCacheConfig := configs.LoadOverlayAlertRuleCacheConfig(overlayAlertRuleCacheConfigPath)
	snsConfig := configs.LoadSNSConfig(snsConfigPath)
	controllerConfig := configs.LoadControllerConfig(controllerConfigPath)
	byocVPCPeeringConfig := configs.LoadBYOCVPCPeeringConfig(byocVPCPeeringConfigPath)
	diagConfig := diagconfig.LoadFromPath(diagControllerConfigPath)
	region := accountInfo.Region
	logicalRegion := region
	if accountInfo.LogicalRegionName != "" {
		logicalRegion = accountInfo.LogicalRegionName
	}
	env := accountInfo.Env
	serviceInfo := structs.ServiceInfo{
		ServerName:           constants.ObservabilityRegionalServer,
		Port:                 port,
		MetricsPort:          metricsPort,
		DefaultOversell:      defaultOversell,
		Vendor:               vendor,
		Region:               region,
		LogicalRegionName:    logicalRegion,
		DomainPostfix:        accountInfo.DomainPostfix,
		Env:                  env,
		LogDir:               logDir,
		LogLevel:             logLevel,
		EnableHttps:          false,
		AZConfig:             azconfig,
		BizTypeConfigs:       bizTypeConfigs,
		MonitorSpec:          monitorSpec,
		AccountInfo:          accountInfo,
		AlertManagerInfo:     alertmanagerInfo,
		VMClusterInfo:        vmConfig,
		SNSInfo:              snsConfig,
		AlertRuleCacheConfig: alertRuleCacheConfig,
		SelfMonitorEndpoint:  selfMonitorEndpoint,
		ControllerConfig:     controllerConfig,
		BYOCVPCPeering:       byocVPCPeeringConfig,
	}
	configs.Init(serviceInfo)

	err = providers.InitClient(serviceInfo)
	if err != nil {
		log.Panic("init provider client failed.", zap.Error(err))
	}

	metrics.SetupMetrics(controller_metrics.Registry)

	var rdsClient myclient.Client
	if controllerConfig.ServerlessConfig.EnableRDSClient {
		rdsSecret, rdsConfig := controllerConfig.ServerlessConfig.RDSSecretArn, controllerConfig.ServerlessConfig.RdsConfig
		if rdsConfig == nil && rdsSecret == "" {
			log.Panic("rds config is nil and rds secret arn is empty.")
		}
		// we prefer using the one defined in rdsSecret
		if rdsSecret != "" {
			switch vendor {
			case string(constants.AWS):
				if rdsConfig, err = getRdsConfigWithAWS(rdsSecret, region); err != nil {
					log.Panic("init rds config with aws failed.", zap.Error(err))
				}
			case string(constants.AliCloud):
				if rdsConfig, err = getRdsConfigWithAliCloud(rdsSecret, region); err != nil {
					log.Panic("init rds config with alibaba cloud failed.", zap.Error(err))
				}
			case string(constants.GCP):
				if rdsConfig, err = getRdsConfigWithGCP(rdsSecret); err != nil {
					log.Panic("init rds config with gcp failed.", zap.Error(err))
				}
			default:
				log.Panic("unsupported vendor")
			}
		}
		serviceInfo.ControllerConfig.ServerlessConfig.RdsConfig = rdsConfig
		rdsClient, err = myclient.NewMySQLClient(
			rdsConfig.ToGormutilConfig(),
			zapr.NewLogger(log.L()).WithName("rdsClusterClient"),
			"rdsClusterClient",
			0,
		)
		if err != nil {
			log.Panic("init mysql client failed.", zap.Error(err))
		}
	}

	worker.InitManager(ctx, rdsClient, serviceInfo)
	//Start Manager must before GetManager
	go worker.StartWorker(ctx)

	kubeConfig.QPS = 50
	kubeConfig.Burst = 100
	resourcemanager.InitReaderWriter(kubeConfig, worker.GetManager(), rdsClient, true /*regional*/)

	if syncAlertRuleConfig.Enabled {
		regionalservice.StartSync(ctx, syncAlertRuleConfig, serviceInfo)
	}

	if alertRuleCacheConfig.Enabled {
		regionalservice.InitAlertRuleCache(alertRuleCacheConfig)
	}

	if diagConfig.Enable {
		ctr, err := diagcontroller.NewController(diagConfig)
		if err != nil {
			log.Error("failed to create diag controller", zap.Error(err))
		} else {
			ctr.Start(ctx)
		}
	}

	regionalapi.StartService(ctx, serviceInfo)
}

func getRdsConfigWithAWS(rdsSecretArn, region string) (*structs.RdsConfig, error) {
	cfg, err := awsconfig.LoadDefaultConfig(context.TODO(), awsconfig.WithRegion(region))
	if err != nil {
		return nil, errors.New(errors.K8SClientError, "unable to get aws config")
	}
	sm := secretsmanager.NewFromConfig(cfg)
	result, err := sm.GetSecretValue(context.TODO(), &secretsmanager.GetSecretValueInput{
		SecretId: aws.String(rdsSecretArn),
	})
	if err != nil {
		return nil, fmt.Errorf("load rds secret configuration err: %w", err)
	}
	rdsConfig := &structs.RdsConfig{}
	if err := json.Unmarshal([]byte(*result.SecretString), rdsConfig); err != nil {
		return nil, fmt.Errorf("unmarshal rds secret configuration err: %w", err)
	}
	return rdsConfig, nil
}

// getRdsConfigWithGCP reads the rds credentials out of GCP Secret Manager, so that they never
// have to be written into controllerconfig.yaml - which is rendered into a ConfigMap and is
// readable by anyone with get on the namespace.
//
// rdsSecretArn carries a Secret Manager resource name here rather than an ARN, the same way the
// AliCloud branch carries a KMS secret name:
//
//	projects/<project>/secrets/<name>/versions/latest
//
// The payload is the same JSON shape as the AWS and AliCloud secrets, so it unmarshals straight
// into RdsConfig. Credentials come from ADC, which on GKE resolves to the workload identity of
// the pod; that service account needs roles/secretmanager.secretAccessor on the secret.
func getRdsConfigWithGCP(rdsSecretArn string) (*structs.RdsConfig, error) {
	ctx := context.TODO()
	svc, err := gcpsecretmanager.NewService(ctx)
	if err != nil {
		return nil, fmt.Errorf("unable to get gcp secret manager client: %w", err)
	}
	result, err := svc.Projects.Secrets.Versions.Access(rdsSecretArn).Context(ctx).Do()
	if err != nil {
		return nil, fmt.Errorf("load rds secret configuration err: %w", err)
	}
	if result.Payload == nil {
		return nil, fmt.Errorf("rds secret %s has no payload", rdsSecretArn)
	}
	// The REST API base64-encodes the payload; the gRPC client would not.
	data, err := base64.StdEncoding.DecodeString(result.Payload.Data)
	if err != nil {
		return nil, fmt.Errorf("decode rds secret payload err: %w", err)
	}
	rdsConfig := &structs.RdsConfig{}
	if err := json.Unmarshal(data, rdsConfig); err != nil {
		return nil, fmt.Errorf("unmarshal rds secret configuration err: %w", err)
	}
	return rdsConfig, nil
}

func getRdsConfigWithAliCloud(rdsSecretArn, region string) (*structs.RdsConfig, error) {
	cred, err := credentials.NewCredential(nil)
	if err != nil {
		return nil, fmt.Errorf("unable to get ali cloud config: %w", err)
	}
	// acs:kms:ap-southeast-1:5092239846523038:secret/*
	parts := strings.Split(rdsSecretArn, ":")
	if len(parts) > 2 {
		region = parts[2]
	}
	config := &aliopenapi.Config{
		Credential: cred,
		RegionId:   &region,
	}
	kmsClient, err := kms.NewClient(config)
	if err != nil {
		return nil, fmt.Errorf("unable to get ali cloud kms client: %w", err)
	}

	result, err := kmsClient.GetSecretValueWithOptions(&kms.GetSecretValueRequest{
		SecretName: aws.String(rdsSecretArn),
	}, &teautil.RuntimeOptions{})
	if err != nil {
		return nil, fmt.Errorf("load rds secret configuration err: %w", err)
	}
	rdsConfig := &structs.RdsConfig{}
	if err := json.Unmarshal([]byte(*result.Body.SecretData), rdsConfig); err != nil {
		return nil, fmt.Errorf("unmarshal rds secret configuration err: %w", err)
	}
	return rdsConfig, nil
}