package dbhubsdk

import (
	"context"
	"database/sql"
	"dataservices/commons/config"
	"dataservices/commons/constants"
	"dataservices/commons/contextvar"
	"dataservices/commons/utils"
	"fmt"
	"slices"
	"strconv"
	"strings"
	"time"

	"dataservices/commons/metrics"
	"dataservices/internal/entity"

	"github.com/aws/aws-sdk-go-v2/service/dynamodb"
	"github.com/pingcap/log"
	"github.com/tidbcloud/dataservices/idl/pbgen/proto/dataapi"
	"github.com/tidbcloud/tiinf/errs"
	errs2 "github.com/tidbcloud/tiinf/errs"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"gorm.io/gorm"
)

type Column struct {
	Col        string `json:"col"`
	DataType   string `json:"data_type"`
	Nullable   bool   `json:"nullable"`
	IsTimeType bool   `json:"is_time_type"`
	TimeType   string `json:"time_type"`
}

type QueryRes struct {
	Code         uint16
	Message      string
	Columns      []*Column  `json:"colomns"`
	Rows         [][]string `json:"rows"`
	RowNum       int64
	StartMs      int64
	EndMs        int64
	Affect       int64
	Databases    []string
	UseDatabase  string
	DropDatabase string
	StmtType     StmtType
	LastInsertID int64
	WarnCount    int      `json:"warn_count,omitempty"`
	WarnMessages []string `json:"warn_messages,omitempty"`
}

func (s *DBHubSdkImpl) QueryByPrepare(ctx context.Context, db *sql.Conn, sql string, args []any, limits uint64) (res *QueryRes, err error) {
	start := time.Now()
	defer func() {
		if r := recover(); r != nil {
			tilog.Error(ctx, "execute sql panic", zap.String("sql", sql), zap.Any("args", args), zap.Any("panic", r))
			err = errs2.NewInternalPanic(fmt.Sprintf("exec sql panic, %v", r), nil)
		}
		ok := strconv.FormatBool(err == nil)
		metrics.SqlExecCounter.WithLabelValues(ok).Inc()
		metrics.SqlExecDuration.WithLabelValues(ok).Observe(time.Since(start).Seconds())
	}()
	return queryDBByPrepare(ctx, db, sql, args, limits)
}

func (s *DBHubSdkImpl) CloseConnection(ctx context.Context, gormDB *gorm.DB) error {
	sqlDB, sqlErr := gormDB.DB()
	if sqlErr != nil {
		return sqlErr
	}
	sqlDB.Close()
	return nil
}

func (s *DBHubSdkImpl) BuildConnection(ctx context.Context, in *entity.ConnectionInput) (out *entity.ConnectionOutput, err error) {
	startTime := time.Now()
	logger := contextvar.Logger(ctx)
	defer func() {
		cost := time.Since(startTime)
		logger.Info("BuildConnection completed", zap.Uint64("cluster", in.ClusterId), zap.Time("startTime", startTime), zap.String("cost", cost.String()), zap.Error(err))
		if !utils.IsTestUser(in.Credential.AccountId) {
			metrics.BuildUserDBConnDuration.WithLabelValues(strconv.FormatBool(err == nil)).Observe(cost.Seconds())
		}
	}()

	user, err := s.GetUserMeta(ctx, in.ProjectId, in.ClusterId)
	if err != nil {
		return nil, errs2.NewBadRequest("Failed to get cluster information, Please check if the cluster is available.", err)
	}

	useTls := !user.IsDedicate
	if user.UseTLS != nil {
		useTls = *user.UseTLS
	}
	if config.Global().IsRegionMode() && !user.IsDedicate &&
		user.Cloud == constants.ClusterCloudProviderAWS &&
		config.Global().ServerlessConf.PrivatelinkEnabled &&
		config.Global().ServerlessConf.AwsVPCPrivateDNS != "" {
		user.Host = config.Global().ServerlessConf.AwsVPCPrivateDNS
		useTls = false
		if slices.Contains(config.Global().ServerlessConf.ProjectIds02, in.ProjectId) {
			logger.Info("use gateway02 for serverless access", zap.Uint64("projectId", in.ProjectId), zap.Uint64("clusterId", in.ClusterId))
			user.Host = strings.Replace(user.Host, "01", "02", 1)
		}
	}

	// Fill user cluster account
	if err := fillUserAccount(ctx, in, user, in.Force); err != nil {
		logger.Error("fillUserAccount failed", zap.Error(err))
		return nil, errs2.NewBadRequest("Failed to connect to cluster. Cause: get db account failed! Please check if the cluster is available.", err)
	}

	sqlDB, err := connectDB(ctx, user, in, useTls)
	if err != nil && !in.Force {
		logger.Error("first connect db failed, retry one more time.", zap.Error(err))
		// retry one more time
		if errRetry := fillUserAccount(ctx, in, user, true); errRetry != nil {
			logger.Error("retry fillUserAccount failed", zap.Error(errRetry))
			return nil, errs2.NewBadRequest("Failed to connect to cluster. Cause: get db account failed! Please check if the cluster is available.", errRetry)
		}
		sqlDB, err = connectDB(ctx, user, in, useTls)
		if err != nil {
			if in.Credential.GetType() == dataapi.CrtType_APIKEY {
				return nil, errs2.NewBadRequest("Failed to connect to cluster, Please check if the cluster is available.", err)
			} else {
				return nil, errs2.NewBadRequest("Failed to connect to cluster, "+err.Error(), nil)
			}
		}
	}
	out = &entity.ConnectionOutput{
		DB:       sqlDB,
		ExpireAt: user.ExpiresAt,
		JwtMd5:   utils.Md5WithDefaultSalt(user.ServerlessPasswd),
	}
	return
}

func (s *DBHubSdkImpl) GetUserMeta(ctx context.Context, projectId, clusterId uint64) (userMeta *TidbUserMeta, err error) {
	startTime := time.Now()
	defer func() {
		tilog.Info(ctx, "GetUserMeta completed", zap.Time("startTime", startTime),
			zap.Uint("cluster", uint(clusterId)), zap.Duration("cost", time.Since(startTime)), zap.Error(err))
	}()
	if s.InfraApi != nil {
		return s.InfraApi.GetUserMeta(ctx, projectId, clusterId)
	}
	if s.UserMetaProvider != nil {
		return s.UserMetaProvider.GetUserMeta(ctx, projectId, clusterId)
	}
	return nil, fmt.Errorf("DBHubSdkImpl user meta provider doesn't initialize")
}

func fillUserAccount(ctx context.Context, in *entity.ConnectionInput, user *TidbUserMeta, force bool) error {
	startTime := time.Now()
	logger := contextvar.Logger(ctx)
	defer func() {
		cost := time.Since(startTime)
		logger.Info("fillUserAccount completed", zap.Uint64("cluster", in.ClusterId), zap.Time("startTime", startTime), zap.String("cost", cost.String()))
	}()

	if in.Credential == nil || in.Credential.Key == "" {
		return errs2.NewBadRequest("deny access to cluster", nil)
	}

	var err error
	start := time.Now()
	defer func() {
		metrics.AccountIntergrationCounter.WithLabelValues(strconv.FormatBool(err == nil)).Inc()
		metrics.GetUserAccountDuration.Observe(time.Since(start).Seconds())
	}()

	if in.Credential.Type == dataapi.CrtType_ACCOUNT {
		dbuser, dbpwd, err := DecryptDBAccount(in.Credential.Key)
		if err != nil {
			return err
		}
		user.ServerlessUser = dbuser
		user.ServerlessPasswd = dbpwd
		return nil
	} else if in.Credential.Type == dataapi.CrtType_MIXED {
		return errs2.NewBadRequest("deny access to cluster", nil)
	}

	jwtRes, err := globalJwtClient.GetJwt(ctx, in.Credential.Type, in.ClusterId, in.Credential.Key, force)
	if jwtRes != nil {
		jwtMd5 := utils.Md5WithDefaultSalt(jwtRes.Jwt)
		if time.Now().Unix() >= jwtRes.ExpiresAt {
			logger.Error("got expired cluster jwt from account", zap.Uint64("cluster", in.ClusterId), zap.String("crtType", in.Credential.Type.String()),
				zap.Bool("force", force), zap.String("username", jwtRes.DbUser), zap.String("jwtMd5", jwtMd5), zap.Int64("expiresAt", jwtRes.ExpiresAt))
			return errs2.NewInternalGeneralBusiness("got expired cluster jwt from account", nil)
		}
		user.ServerlessUser = jwtRes.DbUser
		user.ServerlessPasswd = jwtRes.Jwt
		user.ExpiresAt = jwtRes.ExpiresAt
		logger.Info("successfully get user cluster jwt", zap.Uint64("cluster", in.ClusterId), zap.String("crtType", in.Credential.Type.String()),
			zap.Bool("force", force), zap.String("username", jwtRes.DbUser), zap.String("jwtMd5", jwtMd5), zap.Int64("expiresAt", jwtRes.ExpiresAt),
			zap.Int64("now", time.Now().Unix()))
	}
	return err
}

// Database configuration
type DBHubSdkImpl struct {
	InfraApi         *InfraAPI
	DdbClient        *dynamodb.Client
	UserMetaProvider UserMetaProvider
}

var Serverless = &DBHubSdkImpl{}

func initServerlessDBHubSdk(kubeConfigPath string, projectName string, infraAPIConfigPath string) error {
	log.Info("initServerlessDBHubSdk" + kubeConfigPath)
	baseKubeCli, err := createK8sClientset(kubeConfigPath)
	if err != nil {
		log.Info("createK8sClientset failed. ", zap.Error(err))
		return err
	}
	log.Info("initServerlessDBHubSdk fill InfraApi")
	Serverless.InfraApi, err = NewInfraAPI(projectName, baseKubeCli, infraAPIConfigPath)
	if err != nil {
		log.Info("NewInfraAPI failed. ", zap.Error(err))
		return err
	}
	log.Info("initServerlessDBHubSdk succeed. projectName: " + projectName + " , kubeConfigPath: " + kubeConfigPath)
	return nil
}

func InitSdk(kubeConfigPath string, projectName string, infraAPIConfigPath string) error {
	err := initServerlessDBHubSdk(kubeConfigPath, projectName, infraAPIConfigPath)
	return err
}

func InitRegionalSdk(ddbCli *dynamodb.Client) {
	Serverless.DdbClient = ddbCli
	Serverless.UserMetaProvider = &ddbUserMetaProvider{client: ddbCli}
}

func InitRegionalSdkWithUserMetaProvider(provider UserMetaProvider) {
	Serverless.DdbClient = nil
	Serverless.UserMetaProvider = provider
}

func EncryptDBAccount(user, pwd string) string {
	return user + ":" + pwd
	// return apikeygenerator.Encrypt(user + ":" + pwd)
}

func DecryptDBAccount(encryptStr string) (string, string, error) {
	// decryptStr, err := apikeygenerator.Decrypt(encryptStr)
	// if err != nil {
	// 	return "", "", errs.NewBadRequest("invalid cluster credential key", err)
	// }
	if pos := strings.Index(encryptStr, ":"); pos == -1 {
		return "", "", errs.NewBadRequest("invalid cluster credential key", nil)
	} else {
		return encryptStr[:pos], encryptStr[pos+1:], nil
	}
}

// HandleWithDBConn Inject the database connection into the user's handler and automatically call it,
// The handler does not need to worry about obtaining and releasing the connection.
func HandleWithDBConn(ctx context.Context, db *sql.DB, handler func(context.Context, *sql.Conn) error) error {
	conn, err := db.Conn(ctx)
	if err != nil {
		return err
	}
	defer conn.Close()
	return handler(ctx, conn)
}