package gateway

import (
	"context"
	"crypto/tls"
	"fmt"
	"os"
	"strings"
	"sync/atomic"
	"time"

	alicred "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/smithy-go/logging"
	grpcprom "github.com/grpc-ecosystem/go-grpc-prometheus"
	"github.com/pingcap/tidb/pkg/util"
	tiproxycfg "github.com/pingcap/tiproxy/lib/config"
	"github.com/pingcap/tiproxy/pkg/manager/cert"
	"github.com/pingcap/tiproxy/pkg/proxy"
	"github.com/prometheus/client_golang/prometheus/push"
	pbm "github.com/tidbcloud/aws-shared-provider/pbgen/manager"
	"github.com/tidbcloud/aws-shared-provider/pkg/gateway/client"
	"github.com/tidbcloud/aws-shared-provider/pkg/gateway/config"
	"github.com/tidbcloud/aws-shared-provider/pkg/gateway/handshake"
	"github.com/tidbcloud/aws-shared-provider/pkg/gateway/metrics"
	"github.com/tidbcloud/aws-shared-provider/pkg/gateway/probe"
	"github.com/tidbcloud/aws-shared-provider/pkg/gateway/router"
	"github.com/tidbcloud/aws-shared-provider/pkg/infra"
	"github.com/tidbcloud/aws-shared-provider/pkg/types"
	"github.com/tidbcloud/aws-shared-provider/pkg/utils"
	"github.com/tidbcloud/aws-shared-provider/pkg/utils/tidb"
	consumerredis "github.com/tidbcloud/tidb-management-service/api/serverless_infra/consumer/redis"
	"go.uber.org/zap"
	"google.golang.org/grpc"
	"google.golang.org/grpc/backoff"
	"google.golang.org/grpc/credentials"
	"google.golang.org/grpc/keepalive"
	ctrl "sigs.k8s.io/controller-runtime"
)

var (
	DialGrpcManagerTimeout    = 10 * time.Second
	PushTrafficMetricsTimeout = 10 * time.Second
	ConnectTimeout            = 300
)

// Gateway is a proxy server that connects to backend TiDB servers.
type Gateway struct {
	conf                      *config.Config
	ctx                       context.Context
	cancel                    context.CancelFunc
	sqlProxy                  *proxy.SQLServer
	serving                   uint32
	infraClusterRedisConsumer *consumerredis.Consumer
	infraApi                  infra.InfraAPI
	localKube                 infra.LocalKube
	routerBundle              *router.RouterBundle
	checkBackendServer        *probe.CheckBackendServer
	probeTiDB                 *probe.TiDBProber
	grpcConn                  *grpc.ClientConn
	configWatcher             *config.DynamicConfigWatcher
}

func newGrpcConn(ctx context.Context) (*grpc.ClientConn, error) {
	//nolint:mnd // ignore
	return grpc.DialContext(ctx, "dns:///manager-server:8000",
		grpc.WithStreamInterceptor(grpcprom.StreamClientInterceptor),
		grpc.WithUnaryInterceptor(grpcprom.UnaryClientInterceptor),
		grpc.WithTransportCredentials(credentials.NewTLS(&tls.Config{InsecureSkipVerify: true})), //nolint:gosec // ignore
		grpc.WithReturnConnectionError(),
		grpc.WithConnectParams(grpc.ConnectParams{
			Backoff: backoff.Config{
				BaseDelay:  100 * time.Millisecond, // Default was 1s.
				Multiplier: 1.6,                    // Default
				Jitter:     0.2,                    // Default
				MaxDelay:   3 * time.Second,        // Default was 120s.
			},
			MinConnectTimeout: 2 * time.Second,
		}),
		grpc.WithKeepaliveParams(keepalive.ClientParameters{
			Time:    10 * time.Second,
			Timeout: 2 * time.Second,
		}),
		grpc.WithDefaultServiceConfig(`{"loadBalancingConfig": [ {"round_robin": {}} ]}`),
	)
}

// New creates a new Gateway instance.
//
//nolint:funlen // by design
func New(conf *config.Config, zapLogger *zap.Logger) (*Gateway, error) {
	certConfig := tiproxycfg.Config{
		Security: tiproxycfg.Security{
			ServerSQLTLS:      tiproxycfg.TLSConfig{Cert: conf.TLS.Cert, Key: conf.TLS.Key, MinTLSVersion: conf.TLS.MinVersion},
			SQLTLS:            tiproxycfg.TLSConfig{Cert: conf.SQLTLS.Cert, Key: conf.SQLTLS.Key, CA: conf.SQLTLS.CA},
			RequireBackendTLS: conf.RequireBackendTLS,
		},
	}
	certManager := cert.NewCertManager()
	err := certManager.Init(&certConfig, zapLogger, nil)
	if err != nil {
		return nil, fmt.Errorf("init cert manager err: %w", err)
	}

	httpClient := client.NewManagerServerClient()
	clusterTLSConfig, err := util.NewTLSConfig(
		util.WithCAPath(conf.ClusterTLS.CA),
		util.WithCertAndKeyPath(
			conf.ClusterTLS.Cert,
			conf.ClusterTLS.Key,
		),
	)
	if err != nil {
		return nil, fmt.Errorf("load cluster tls config err: %w", err)
	}

	var pAwsCfg *aws.Config
	var alicloudCred alicred.Credential
	switch conf.CloudProviderType {
	case types.ProviderTypeAliCloudFreeTier:
		if conf.OtsInstanceName != "" {
			if alicloudCred, err = alicred.NewCredential(nil); err != nil {
				return nil, fmt.Errorf("load alicloud credential err: %w", err)
			}
			break
		}
		ctrl.Log.Info("--ots-instance-name is empty in alicloud, use aws for its regional infra!")
		fallthrough
	case types.ProviderTypeAWSFreeTier:
		awsSDKLogger, awsSDKLogMode := logging.Logger(logging.Nop{}), aws.ClientLogMode(0)
		if conf.AWSSDKLogMode != "" {
			awsSDKLogger = logging.NewStandardLogger(os.Stderr)
			for _, mode := range strings.Split(conf.AWSSDKLogMode, ",") {
				switch strings.TrimSpace(mode) {
				case "signing":
					awsSDKLogMode |= aws.LogSigning
				case "retries":
					awsSDKLogMode |= aws.LogRetries
				case "request":
					awsSDKLogMode |= aws.LogRequest
				case "requestWithBody":
					awsSDKLogMode |= aws.LogRequestWithBody
				case "response":
					awsSDKLogMode |= aws.LogResponse
				case "responseWithBody":
					awsSDKLogMode |= aws.LogResponseWithBody
				default:
					return nil, fmt.Errorf("unknown aws sdk log mode: %s", mode)
				}
			}
		}
		awsCfg, err := awsconfig.LoadDefaultConfig(context.Background(),
			awsconfig.WithLogger(awsSDKLogger),
			awsconfig.WithClientLogMode(awsSDKLogMode),
		)
		if err != nil {
			return nil, fmt.Errorf("load aws configuration err: %w", err)
		}

		pAwsCfg = &awsCfg
	default:
		return nil, fmt.Errorf("unsupported cloud provider type: %s", conf.CloudProviderType)
	}

	ctx, cancel := context.WithCancel(context.Background())
	tidbController := tidb.NewController(clusterTLSConfig, "")

	regionalInfraConfig := &infra.RegionalInfraConfig{
		InfraRedisClientTLSConfig: infra.TLSConfig(conf.InfraRedisClientTLSConfig),
		InfraClusterTable:         conf.InfraClusterTable,
		OtsInstanceName:           conf.OtsInstanceName,
		Region:                    conf.Region,
		InfraRedisAddr:            conf.InfraRedisAddr,
		KubeClusterName:           conf.KubeClusterName,
		AwsCfg:                    pAwsCfg,
		AlicloudCred:              alicloudCred,

		CloudProviderType:      conf.CloudProviderType,
		EnableInfraMysqlClient: conf.EnableInfraMysqlClient,
		InfraRdsSecretArn:      conf.InfraRdsSecretArn,
		CacheServerAddr:        conf.CacheServerAddr,
	}
	regionalInfra, consumer, err := infra.NewRegionalInfra(ctx, regionalInfraConfig)
	if err != nil {
		cancel()
		return nil, fmt.Errorf("init regional infra err: %w", err)
	}

	infraConfig := &infra.InfraConfig{
		Rest:              conf.LocalConfig,
		CloudProviderType: conf.CloudProviderType,
		KubeClusterName:   conf.KubeClusterName,
	}
	infraApi, err := infra.NewInfraAPI(ctx, infraConfig)
	if err != nil {
		cancel()
		return nil, fmt.Errorf("init infra api err: %w", err)
	}

	localKube, err := infra.NewLocalKube(ctx, conf.LocalConfig)
	if err != nil {
		cancel()
		return nil, fmt.Errorf("init local kube err: %w", err)
	}

	grpcCtx, grpcCancel := context.WithTimeout(ctx, DialGrpcManagerTimeout)
	defer grpcCancel()
	grpcConn, err := newGrpcConn(grpcCtx)
	if err != nil {
		cancel()
		return nil, fmt.Errorf("init grpc conn err: %w", err)
	}
	tbClient := pbm.NewTidbBackendClient(grpcConn)

	rb := router.NewRouterBundle(regionalInfra, infraApi, localKube, httpClient, tbClient)

	checkBackendServer := probe.NewCheckBackendServer(ctx, httpClient)
	probeTiDB := probe.NewTiDBProber(ctx, rb, regionalInfra, infraApi, tidbController)

	handshakeHandler := handshake.NewHandshakeHandler(
		ctx,
		conf,
		regionalInfra,
		infraApi,
		rb,
		zapLogger,
		httpClient,
		probeTiDB,
		checkBackendServer,
	)

	tiproxyLogger := zapLogger.With(zap.String("component", "tiproxy"))
	sqlConfig := tiproxycfg.ProxyServer{Addr: conf.Address}
	sqlConfig.GracefulWaitBeforeShutdown = 60
	sqlConfig.ProxyProtocol = "v2"
	sqlConfig.FrontendKeepalive, sqlConfig.BackendHealthyKeepalive,
		sqlConfig.BackendUnhealthyKeepalive = tiproxycfg.DefaultKeepAlive()
	sqlConfig.FrontendReadTimeout = conf.FrontendReadTimeout
	sqlConfig.ConnectTimeout = ConnectTimeout
	sqlProxy, err := proxy.NewSQLServer(tiproxyLogger, &tiproxycfg.Config{Proxy: sqlConfig}, certManager, handshakeHandler)
	if err != nil {
		if grpcConn != nil {
			grpcConn.Close() //nolint:errcheck // no need check error
		}
		cancel()
		return nil, fmt.Errorf("init sql proxy err: %w", err)
	}

	return &Gateway{
		conf:                      conf,
		ctx:                       ctx,
		cancel:                    cancel,
		routerBundle:              rb,
		sqlProxy:                  sqlProxy,
		infraClusterRedisConsumer: consumer,
		grpcConn:                  grpcConn,
		infraApi:                  infraApi,
		localKube:                 localKube,
		checkBackendServer:        checkBackendServer,
		probeTiDB:                 probeTiDB,
		configWatcher: config.NewDynamicConfigWatcher(
			&config.DynamicConfigWatcherConfig{},
		),
	}, nil
}

// Stop stops the Gateway.
func (g *Gateway) Stop() {
	atomic.StoreUint32(&g.serving, 0)
	ctrl.Log.Info("gateway starts to stop")

	g.cancel()
	err := g.sqlProxy.Close()
	if err != nil {
		ctrl.Log.Error(err, "failed to close sql proxy")
	}
	g.checkBackendServer.Stop()
	g.probeTiDB.Stop()
	if g.grpcConn != nil {
		g.grpcConn.Close() //nolint:errcheck // no need check error
	}
}

func (g *Gateway) WaitForConnectionsClose(gracefulWaitBeforeCloseServer int) {
	ctrl.Log.Info("start to wait for connections to close", "timeout", gracefulWaitBeforeCloseServer)
	timeout := time.After(time.Duration(gracefulWaitBeforeCloseServer) * time.Second)
	ticker := time.NewTicker(30 * time.Second) //nolint:mnd // no need
	defer ticker.Stop()
	for {
		select {
		case <-ticker.C:
			connCount := g.routerBundle.ConnCount()
			if connCount <= 0 {
				ctrl.Log.Info("all connections are closed", "conn_count", connCount)
				return
			}
			ctrl.Log.Info("waiting for connections to close", "conn_count", connCount)
		case <-timeout:
			ctrl.Log.Info("graceful shutdown timeout, stop waiting for connections to close")
			return
		}
	}
}

// StartServe starts the Gateway.
func (g *Gateway) StartServe() error {
	ctrl.Log.Info("gateway starts to serve")

	if g.infraClusterRedisConsumer != nil {
		go g.infraClusterRedisConsumer.Start(g.ctx) //nolint:errcheck // no need check error
	}
	go g.infraApi.Start(g.ctx)  //nolint:errcheck // no need check error
	go g.localKube.Start(g.ctx) //nolint:errcheck // no need check error
	g.infraApi.WaitForCacheSync(g.ctx)
	g.localKube.WaitForCacheSync(g.ctx)

	g.checkBackendServer.Start()
	g.startPushTrafficMetrics()

	go g.configWatcher.Start(g.ctx)

	go g.routerBundle.RunGC(g.ctx)

	if g.conf.GatewayMode == config.InnerMode {
		go probe.RunProbeLatency(g.ctx)
	}
	if g.conf.GatewayMode == config.PublicMode {
		go RunIPRanges(g.ctx)
	}

	g.startServe()
	return nil
}

// IsServing returns true if the gateway is serving.
func (g *Gateway) IsServing() bool {
	return atomic.LoadUint32(&g.serving) == 1
}

func (g *Gateway) startServe() {
	ctrl.Log.Info("gateway starts to accept connections")
	g.sqlProxy.Run(g.ctx, nil)
	atomic.StoreUint32(&g.serving, 1)
}

func (g *Gateway) startPushTrafficMetrics() {
	if g.conf.PushTrafficMetricsAddr == "" || g.conf.PushTrafficMetricsInterval <= 0 {
		ctrl.Log.Info("push traffic metrics is disabled")
		return
	}
	go func() {
		ctrl.Log.Info("start to push traffic metrics")
		ticker := time.NewTicker(time.Duration(g.conf.PushTrafficMetricsInterval) * time.Second)
		defer ticker.Stop()
		for {
			select {
			case <-g.ctx.Done():
				ctrl.Log.Info("stop to push traffic metrics")
				return
			case <-ticker.C:
				g.pushTrafficMetrics()
			}
		}
	}()
}

func (g *Gateway) pushTrafficMetrics() {
	pushCtx, cancel := context.WithTimeout(g.ctx, PushTrafficMetricsTimeout)
	podName := os.Getenv("HOSTNAME")
	if podName == "" {
		podName = "default"
	}
	defer cancel()
	err := push.New(g.conf.PushTrafficMetricsAddr, "tidb_gateway").
		Grouping("pod", podName).
		Grouping("mode", g.conf.GatewayMode).
		Collector(metrics.ConnectionFlow).
		PushContext(pushCtx)
	if err != nil {
		ctrl.Log.Error(err, "push traffic metrics failed")
	}
}

func RunIPRanges(ctx context.Context) {
	utils.AwsIPSvc.Load()
	ticker := time.NewTicker(24 * time.Hour) //nolint:mnd // no need
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			utils.AwsIPSvc.Load()
		}
	}
}