package main

import (
	"context"
	"fmt"
	"net"
	"net/http"
	"net/http/pprof"
	"os"
	"runtime/debug"
	"strconv"

	"github.com/getsentry/sentry-go"
	"github.com/go-logr/zapr"
	grpcrecovery "github.com/grpc-ecosystem/go-grpc-middleware/recovery"
	grpcvalidator "github.com/grpc-ecosystem/go-grpc-middleware/validator"
	grpcprometheus "github.com/grpc-ecosystem/go-grpc-prometheus"
	"github.com/pingcap/errors"
	"github.com/pingcap/log"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/promhttp"
	"github.com/spf13/cobra"
	"github.com/tidbcloud/core-svc/core_ctx/application/cmd/manually"
	"github.com/tidbcloud/core-svc/core_ctx/application/cmd/measure"
	"github.com/tidbcloud/core-svc/core_ctx/infra/config"
	"github.com/tidbcloud/core-svc/core_ctx/infra/controller"
	"github.com/tidbcloud/core-svc/core_ctx/infra/facade_imp"
	"github.com/tidbcloud/core-svc/core_ctx/infra/metrics"
	"github.com/tidbcloud/core-svc/core_ctx/util"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	pbutil "github.com/tidbcloud/core-svc/idl/pbgen/util"
	"github.com/tidbcloud/core-svc/knife"
	"github.com/tidbcloud/core-svc/migrate"
	"github.com/tidbcloud/pkgv2/bizerr"
	grpcbizerr "github.com/tidbcloud/pkgv2/grpc/middleware/bizerr"
	grpcsentry "github.com/tidbcloud/pkgv2/grpc/middleware/sentry"
	grpctrace "github.com/tidbcloud/pkgv2/grpc/middleware/trace"
	"github.com/tidbcloud/pkgv2/infra/aws"
	pkglog "github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/pkgv2/logid"
	"github.com/tidbcloud/tiinf/tiinf"
	tigrpc "github.com/tidbcloud/tiinf/transport/tigrpc/server"
	"go.opentelemetry.io/otel"
	"go.uber.org/zap"
	"google.golang.org/grpc"
	"google.golang.org/grpc/metadata"
	"google.golang.org/grpc/peer"
	"gorm.io/gorm"
	controllerLog "sigs.k8s.io/controller-runtime/pkg/log"
	controllerMetrics "sigs.k8s.io/controller-runtime/pkg/metrics"
)

const (
	defaultConfigPath = "/etc/core-svc/config.yaml"

	allRole        = "all"
	coreSvcRole    = "CORE_SVC_ROLE"
	apiServiceRole = "API_SERVICE"
	workerRole     = "WORKER"
)

var (
	configPath string
	serveCmd   = &cobra.Command{
		Use:   "serve",
		Short: "serve a gRPC server or a worker",
		Run: func(cmd *cobra.Command, args []string) {
			cobra.CheckErr(serve())
		},
	}

	printErrCode = &cobra.Command{
		Use:   "print-errcode",
		Short: "print all defined error code",
		Run: func(cmd *cobra.Command, args []string) {
			bizerr.PrintAllCustomErrCode()
		},
	}
)

var (
	Tracer                 = otel.Tracer("core")
	MetricsGrpcCallerTrack = prometheus.NewCounterVec(prometheus.CounterOpts{
		Namespace: "dbaas",
		Subsystem: "core_svc",
		Name:      "grpc_caller_track",
		Help:      "grpc caller track",
	}, []string{"ip", "hostname", "method"})
	inf       tiinf.TiInf
	tiinfConf *tiinf.Config
)

func init() {
	prometheus.MustRegister(MetricsGrpcCallerTrack)
}

func newMigrateCidrCmd() *cobra.Command {
	cmd := &cobra.Command{
		Use:   "migrate-cidr",
		Short: "migrate cidr",
		Run: func(cmd *cobra.Command, args []string) {
			cobra.CheckErr(migrateCidr())
		},
	}

	return cmd
}

func newRerunProjectMeteringCmd() *cobra.Command {
	var dryrun bool
	cmd := &cobra.Command{
		Use:   "rerun-project-metering [projectMeteringId1, ...]",
		Short: "rerun project metering for specified projectMeteringIds",
		Run: func(cmd *cobra.Command, args []string) {
			cobra.CheckErr(rerunProjectMetering(dryrun, args))
		},
	}
	cmd.Flags().BoolVar(&dryrun, "dryrun", false, "whether amend projectMetering and billingReportRecord or not")

	return cmd
}

func newRepairServerlessDelegateUserCmd() *cobra.Command {
	params := &manually.ServerlessDelegateUserRepairerParams{}
	cmd := &cobra.Command{
		Use:   "repair-serverless-delegate-user [clusterId1, ...]",
		Short: "recreate delegate user by 'cloud_admin' for serverless clusters",
		Run: func(cmd *cobra.Command, args []string) {
			ctx := context.Background()
			ctx = logid.WithLogid(ctx, logid.GenLogid())

			params.ClusterIDs = args
			cobra.CheckErr(util.DiCall(func(r *manually.ServerlessDelegateUserRepairer) error {
				return r.Repair(ctx, params)
			}))
		},
	}
	cmd.Flags().BoolVar(&params.Force, "force", false, "force to recreate delegate users")

	return cmd
}

func newScaleClusterCmd() *cobra.Command {
	params := &manually.ScaleClusterParams{}
	cmd := &cobra.Command{
		Use:   "scale-cluster clusterId",
		Short: "scale iops of components for cluster",
		RunE: func(cmd *cobra.Command, args []string) error {
			ctx := context.Background()
			ctx = logid.WithLogid(ctx, logid.GenLogid())
			if len(args) != 1 {
				return fmt.Errorf("there should be one and only one cluster id")
			}
			clusterId, err := strconv.ParseUint(args[0], 10, 64)
			if err != nil {
				return fmt.Errorf("invalid cluster id: %s", err)
			}
			params.ClusterID = clusterId

			return util.DiCall(func(r *manually.ScaleClusterSvc) error {
				return r.Scale(ctx, params)
			})
		},
	}
	cmd.Flags().StringVar(&params.Tidb.Tier, "tidb-tier", "", "tier of tidb, such as '8C32G'. '' means no change")
	cmd.Flags().Int32Var(&params.Tidb.Replica, "tidb-replica", 0, "replica of tidb. 0 means no change")
	cmd.Flags().Uint32Var(&params.Tikv.IOPS, "tikv-iops", 0, "iops of tikv. 0 means no change")
	cmd.Flags().Int64Var(&params.Tikv.Size, "tikv-size", 0, "storage size of tikv, unit: GiB. 0 means no change")
	cmd.Flags().StringVar(&params.Tikv.Tier, "tikv-tier", "", "tier of tikv, such as '8C32G'. '' means no change")
	cmd.Flags().Int32Var(&params.Tikv.Replica, "tikv-replica", 0, "replica of tikv. 0 means no change")
	cmd.Flags().Uint32Var(&params.Tiflash.IOPS, "tiflash-iops", 0, "iops of tiflash. 0 means no change")
	cmd.Flags().Int64Var(&params.Tiflash.Size, "tiflash-size", 0, "storage size of tiflash, unit: GiB. 0 means no change")
	cmd.Flags().StringVar(&params.Tiflash.Tier, "tiflash-tier", "", "tier of tiflash, such as '8C32G'. '' means no change")
	cmd.Flags().Int32Var(&params.Tiflash.Replica, "tiflash-replica", 0, "replica of tiflash. 0 means no change")

	return cmd
}

func newRootCmd() *cobra.Command {
	cmd := &cobra.Command{
		Use:               "core-svc",
		CompletionOptions: cobra.CompletionOptions{DisableDefaultCmd: true},
	}
	cmd.AddCommand(
		serveCmd,
		newMigrateCidrCmd(),
		newRerunProjectMeteringCmd(),
		newRepairServerlessDelegateUserCmd(),
		newScaleClusterCmd(),
		printErrCode)
	cmd.PersistentFlags().StringVar(&configPath, "config-path", defaultConfigPath, "dbaas configuration file path")
	return cmd
}

func main() {
	cobra.OnInitialize(func() {
		log.Info("OnInitialize", zap.String("configPath", configPath))
		err := config.InitConfig(configPath)
		if err != nil {
			log.Fatal("Initialize configuration failed", zap.Error(err))
		}

		if err := setup(); err != nil {
			log.Fatal("setup failed", zap.Error(err))
		}
		log.Info("setup finished")
	})

	rootCmd := newRootCmd()
	if err := rootCmd.Execute(); err != nil {
		log.Fatal("serve return err", zap.Error(err))
	}
}

func setup() error {
	inf = tiinf.NewInf()
	// initialize
	var err error
	tiinfConf, err = tiinf.ParseConfigFromFile(configPath)
	if err != nil {
		return fmt.Errorf("parse tiinf config file: %s", err)
	}
	if err := inf.Init(
		tiinf.WithConfigFile(configPath),
		tiinf.WithGrpcOptions(tigrpc.WithUnaryInterceptors(setupGrpcMiddlewares()...)),
		tiinf.WithGrpcOptions(tigrpc.WithRecovery(grpcrecovery.WithRecoveryHandlerContext(panicHandler))),
	); err != nil {
		return fmt.Errorf("init TiInf: %w", err)
	}

	http.Handle("/metrics_controller",
		promhttp.InstrumentMetricHandler(
			controllerMetrics.Registry, promhttp.HandlerFor(controllerMetrics.Registry, promhttp.HandlerOpts{}),
		))

	controllerLog.SetLogger(zapr.NewLogger(log.L()))

	setupDI()
	warmup()
	metrics.RegisterMetrics()
	eventbus := controller.NewEventbus(config.Global())
	k8sControllerMgr := facade_imp.NewK8sControllerManager(config.Global())

	if enableWorker() {
		// worker do the DB migration
		if err := util.DiCall(func(db *gorm.DB) error {
			return migrate.RunCoreMigrate(db)
		}); err != nil {
			panic(fmt.Sprintf("do DB migration failed: %s", err))
		}
		log.Info("migrate finished")

		setupCronWorker()

		setupDBSubscriber()
		// only the worker is expected to subscribe events, but in theory,
		// both the worker and server have the ability to do so.
		if err := controller.RegisterEventbusConsumers(eventbus); err != nil {
			return fmt.Errorf("register eventbus consumers: %w", err)
		}

		if err := util.DiCall(controller.NewInfraApiWatcher); err != nil {
			return fmt.Errorf("new infraApi wather: %s", err)
		}
	}
	setupMetricsAndPprof()

	if enableApiService() {
		core.RegisterCoreServiceServer(inf.GrpcServer(), &controller.CoreController{})
	}

	facade_imp.RunK8sControllerManager(k8sControllerMgr)
	log.Info("k8s controller manager is running")
	// run eventbus after all producers and consumers are initialized.
	// both worker and server are expected to run eventbus.
	if err := eventbus.Run(); err != nil {
		return fmt.Errorf("run eventbus: %w", err)
	}
	log.Info("eventbus is running")

	return nil
}

func serve() error {
	// flush all buffering log
	flushFn := func() {
		if err := log.L().Sync(); err != nil {
			fmt.Println(err.Error())
		}
	}
	defer flushFn()

	log.S().Infof("listen on: %d", tiinfConf.TiInfConfig.GrpcServerConfig.Port)
	// start all components
	return inf.Run()
}

func migrateCidr() error {
	return util.DiCall(func(svc *manually.MigrateCidrSvc) error {
		return svc.Migrate(context.Background())
	})
}

func rerunProjectMetering(dryrun bool, args []string) error {
	return util.DiCall(func(svc measure.ProjectMeteringRecorder) error {
		meteringIds := make([]uint64, 0, len(args))
		for _, each := range args {
			id, err := strconv.ParseUint(each, 10, 64)
			if err != nil {
				return err
			}
			meteringIds = append(meteringIds, id)
		}
		log.S().Infof("rerunProjectMetering, dryrun=%t,ids=%v", dryrun, meteringIds)

		return svc.Rerun(context.Background(), dryrun, meteringIds)
	})
}

func panicHandler(ctx context.Context, p interface{}) error {
	logger := pkglog.WithContextL(ctx)
	err, ok := p.(error)
	if ok && errors.HasStack(err) {
		logger.Sugar().Warnf("panic, err: %s, stack: %s", err, errors.GetStackTracer(err).StackTrace())
	} else {
		logger.Sugar().Warnf("panic, err: %s, stack: %s", p, debug.Stack())
	}

	// TODO(chenfei): opt, Should replaced by grpc_sentry.UnaryServerInterceptor
	// report panic to sentry
	sentry.CurrentHub().Recover(err)
	return bizerr.New(bizerr.ErrCodePanic, "panic with: %s", p)
}

func setupGrpcMiddlewares() []grpc.UnaryServerInterceptor {
	return []grpc.UnaryServerInterceptor{
		// 3. set err to nil if err is BizError, error message and code has been set in BaseResp
		grpcbizerr.UnaryServerInterceptor(func(resp interface{}, err bizerr.BizError) (interface{}, error) {
			return resp, nil
		}),
		// 2. provides Prometheus monitoring for Unary RPCs
		grpcprometheus.UnaryServerInterceptor,
		// 1. pbutil.SetCodeMsg: set BaseResp.ErrCode and BaseResp.ErrMsg if err is BizError and response contains BaseResp
		//    return err if err is internal or exception error, otherwise return nil, so that the grpcprometheus.UnaryServerInterceptor can record the correct status code.
		grpcbizerr.UnaryServerInterceptor(func(resp interface{}, err bizerr.BizError) (interface{}, error) {
			pbutil.SetCodeMsg(resp, int64(err.ErrCode()), err.ErrMsg())
			if bizerr.IsInternalErr(err.ErrCode()) || bizerr.IsException(err.ErrCode()) {
				return resp, err
			}
			return resp, nil
		}),
		//otelgrpc.UnaryServerInterceptor(),
		LogidUnaryServerInterceptor(),
		grpcsentry.UnaryServerInterceptor(),
		grpcvalidator.UnaryServerInterceptor(),
		grpctrace.UnaryServerInterceptor(),
		//grpcrecovery.UnaryServerInterceptor(opts...),
		CallerTrackUnaryServerInterceptor(),
		knife.DegradationInterceptor(func() knife.DegradationConfig {
			return config.Global().Degradation
		}),
	}
}

func CallerTrackUnaryServerInterceptor() grpc.UnaryServerInterceptor {
	return func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
		trackCaller(ctx)
		return handler(ctx, req)
	}
}

func trackCaller(ctx context.Context) {
	ctx, span := Tracer.Start(ctx, "CallerTrackUnaryServerInterceptor")
	defer span.End()

	if config.Global().ServerConfig.CallerTrack.EnableCallerTrack {
		logger := pkglog.WithContextS(ctx)
		var (
			addr      string
			ip        string
			method    string
			hostname  string
			hostnames []string
		)

		if p, ok := peer.FromContext(ctx); ok {
			addr = p.Addr.String()
		}
		if m, ok := grpc.Method(ctx); ok {
			method = m
		}

		if config.Global().ServerConfig.CallerTrack.EnableResolveHostname {
			if addr != "" {
				ip, _, _ = net.SplitHostPort(addr)
				hostnames, _ = net.LookupAddr(ip)
				if len(hostnames) >= 1 {
					hostname = hostnames[0]
				}
			}
		}

		logger.Infof("callerTrack, addr=%s, method=%s, hostname=%s, hostnames=%s", addr, method, hostname, hostnames)
		MetricsGrpcCallerTrack.WithLabelValues(ip, hostname, method).Inc()
	}
}

// LogidUnaryServerInterceptor get logid from grpc-metadata
// TODO(chenfei): remove this later if the new framework is ready
func LogidUnaryServerInterceptor() grpc.UnaryServerInterceptor {
	return func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
		// try to get requestid from ctx for compatible with grpc-gateway
		md, ok := metadata.FromIncomingContext(ctx)
		if ok {
			if vals := md.Get(logid.LogidHTTPHeaderKey); len(vals) > 0 {
				ctx = logid.WithLogid(ctx, vals[0])
			} else {
				ctx = logid.WithLogid(ctx, logid.GenLogid())
			}
		}

		ctx = injectCtxInfoIntoCtx(ctx, req)
		return handler(ctx, req)
	}
}

func injectCtxInfoIntoCtx(ctx context.Context, req interface{}) context.Context {
	if req == nil {
		return ctx
	}

	requestid := pbutil.GetTagsFromReq(req)[logid.GRPCHeaderKey]
	if requestid != "" {
		ctx = logid.WithLogid(ctx, requestid)
	}
	return ctx
}

func setupMetricsAndPprof() {
	// Start metrics server.
	mux := http.NewServeMux()
	setupPprofHandler(mux)
}

func setupPprofHandler(mux *http.ServeMux) {
	mux.HandleFunc("/debug/pprof/", pprof.Index)
	mux.HandleFunc("/debug/pprof/cmdline", pprof.Cmdline)
	mux.HandleFunc("/debug/pprof/profile", pprof.Profile)
	mux.HandleFunc("/debug/pprof/symbol", pprof.Symbol)
	mux.HandleFunc("/debug/pprof/trace", pprof.Trace)
}

func setupDI() {
	controller.InitDependencies()
}

func setupCronWorker() {
	if err := util.DiCall(func(w *controller.GlobalWorker) {
		w.Start()
	}); err != nil {
		panic(fmt.Sprintf("start cron worker failed: %s", err))
	}
}

func setupDBSubscriber() {
	if err := util.DiCall(func(m *controller.SubscriberMgr) {
		m.Start()
	}); err != nil {
		panic(fmt.Sprintf("start subscriber manager failed: %s", err))
	}
}

func getRoleFromEnv() string {
	role := os.Getenv(coreSvcRole)
	if role == "" {
		role = allRole
	}
	return role
}

func enableApiService() bool {
	role := getRoleFromEnv()
	return role == allRole || role == apiServiceRole
}

func enableWorker() bool {
	role := getRoleFromEnv()
	return role == allRole || role == workerRole
}

func warmup() {
	go aws.WarmupKmsSvc()
}