package cluster_proxy

import (
	"context"
	"fmt"
	"time"

	"github.com/patrickmn/go-cache"
	"github.com/tidbcloud/account/idl/pbgen/proto/acl"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	"github.com/tidbcloud/dataflow-service/app/config"
	"github.com/tidbcloud/dataflow-service/internal/model/cluster/clusterproxy/tidbcloud"
	bizerr_wrapper "github.com/tidbcloud/dataflow-service/pkg/bizerr-wrapper"
	encryptor_wrapper "github.com/tidbcloud/dataflow-service/pkg/encryptor-wrapper"
	"github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/clusterproxy"
)

type Model interface {
	GetClusterProxy(ctx context.Context,
		projectId uint64,
		clusterId uint64,
	) (clusterproxy.ClusterProxy, error)
	DecryptCloudAdminPassword(adminPasswd string) (string, error)
	ExecuteMultiSQL(ctx context.Context, req *ExecuteMultiSQLReq) error
	QuerySQL(ctx context.Context, req *QuerySQLReq) ([]byte, error)
}

type ExecuteMultiSQLReq struct {
	ProjectID uint64
	ClusterID uint64
	// if UserID is null, use cloud_admin to execute sql
	UserID uint64
	SQL    []string
}

type QuerySQLReq struct {
	ProjectID uint64
	ClusterID uint64
	// if UserID is null, use cloud_admin to execute sql
	UserID uint64
	SQL    string
}

// The clusterProxyModelImpl struct provides the implementation of the Model interface.
type clusterProxyModelImpl struct {
	cache           *cache.Cache
	clusterPwdCache *cache.Cache

	coreCli    grpcli.CoreGrpcClient
	aclGrpcCli grpcli.AccountAclGrpcClient
	enc        encryptor_wrapper.Encryptor
}

//nolint:gomnd
func NewClusterProxyModelImpl(
	enc encryptor_wrapper.Encryptor,
	coreCli grpcli.CoreGrpcClient,
	aclGrpcCli grpcli.AccountAclGrpcClient,
) Model {
	return &clusterProxyModelImpl{
		enc:             enc,
		coreCli:         coreCli,
		aclGrpcCli:      aclGrpcCli,
		cache:           cache.New(1*time.Minute, 30*time.Second),
		clusterPwdCache: cache.New(1*time.Minute, 30*time.Second),
	}
}

func (p *clusterProxyModelImpl) GetClusterProxy(
	ctx context.Context, projectId uint64, clusterId uint64,
) (clusterproxy.ClusterProxy, error) {
	cacheKey := fmt.Sprintf("%d-%d", projectId, clusterId)
	if proxy, ok := p.cache.Get(cacheKey); ok {
		return proxy.(clusterproxy.ClusterProxy), nil
	}

	cluster, bizErr := p.coreCli.GetAnyClusterByID(ctx, clusterId)
	if bizErr != nil {
		return nil, fmt.Errorf("failed get cluster %d, %s", clusterId, bizErr.Error())
	}

	if cluster.Status == core.ClusterStatus_CLUSTER_STATUS_DELETECOMPLETE ||
		cluster.Status == core.ClusterStatus_CLUSTER_STATUS_CLEARED {
		return nil, fmt.Errorf("cluster id is %d, the status is not normal, is deleted", clusterId)
	}

	if cluster.Status == core.ClusterStatus_CLUSTER_STATUS_PAUSED {
		return nil, fmt.Errorf("cluster id is %d, the status is not normal, is paused", clusterId)
	}

	proxy, err := tidbcloud.BuildClusterProxyV2(
		ctx,
		clusterId,
		cluster.Version,
		cluster.InternalEndpoint,
		config.ClusterProxyCDCHeavyOperationTimeout,
	)
	if err != nil {
		return nil, fmt.Errorf("failed to get proxy, %s", err.Error())
	}
	p.cache.SetDefault(cacheKey, proxy)
	return proxy, nil
}

func (p *clusterProxyModelImpl) DecryptCloudAdminPassword(adminPasswd string) (string, error) {
	// Clusters that have not yet been upgraded to use cloud_admin
	if len(adminPasswd) == 0 {
		return "", nil
	}
	pwd, err := p.enc.DecryptBase64(adminPasswd)
	if err != nil {
		return "", bizerr.WrapError(err, "decrypt encrypted cloud admin password")
	}
	return pwd, nil
}

func (p *clusterProxyModelImpl) ExecuteMultiSQL(
	ctx context.Context, req *ExecuteMultiSQLReq,
) error {
	if req.UserID == 0 {
		return p.executeMultiSQLByCloudAdmin(ctx, req.ProjectID, req.ClusterID, req.SQL)
	}
	return p.executeMultiSQLByUser(ctx, req.ProjectID, req.ClusterID, req.UserID, req.SQL)
}

func (p *clusterProxyModelImpl) QuerySQL(ctx context.Context, req *QuerySQLReq) ([]byte, error) {
	if req.UserID == 0 {
		return p.querySQLByCloudAdmin(ctx, req.ProjectID, req.ClusterID, req.SQL)
	}
	return p.querySQLByUser(ctx, req.ProjectID, req.ClusterID, req.UserID, req.SQL)
}

func (p *clusterProxyModelImpl) querySQLByCloudAdmin(
	ctx context.Context, projectId uint64, clusterId uint64, sql string,
) ([]byte, error) {
	proxy, err := p.GetClusterProxy(ctx, projectId, clusterId)
	if err != nil {
		return nil, err
	}
	adminPwd, err := p.queryClusterCloudAdminPwdWithCache(ctx, projectId, clusterId)
	if err != nil {
		return nil, err
	}
	return proxy.QuerySQLV2(ctx, &clusterproxy.ClusterOperator{
		Username:   "cloud_admin",
		AuthMethod: clusterproxy.AuthMethodPassword,
		Credential: adminPwd,
	}, sql)
}

func (p *clusterProxyModelImpl) querySQLByUser(
	ctx context.Context, projectId uint64, clusterId uint64, userId uint64, sql string,
) ([]byte, error) {
	proxy, err := p.GetClusterProxy(ctx, projectId, clusterId)
	if err != nil {
		return nil, err
	}
	tokenOperator, err := p.queryClusterUserJwtToken(ctx, clusterId, userId)
	if err != nil {
		return nil, err
	}
	return proxy.QuerySQLV2(ctx, tokenOperator, sql)
}

func (p *clusterProxyModelImpl) executeMultiSQLByUser(
	ctx context.Context, projectId uint64, clusterId uint64, userId uint64, sql []string,
) error {
	proxy, err := p.GetClusterProxy(ctx, projectId, clusterId)
	if err != nil {
		return err
	}
	tokenOperator, err := p.queryClusterUserJwtToken(ctx, clusterId, userId)
	if err != nil {
		return err
	}
	return proxy.ExecuteMultiSQLV2(ctx, tokenOperator, sql)
}

func (p *clusterProxyModelImpl) executeMultiSQLByCloudAdmin(
	ctx context.Context, projectId uint64, clusterId uint64, sql []string,
) error {
	proxy, err := p.GetClusterProxy(ctx, projectId, clusterId)
	if err != nil {
		return err
	}
	adminPwd, err := p.queryClusterCloudAdminPwdWithCache(ctx, projectId, clusterId)
	if err != nil {
		return err
	}
	return proxy.ExecuteMultiSQLV2(ctx, &clusterproxy.ClusterOperator{
		Username:   "cloud_admin",
		AuthMethod: clusterproxy.AuthMethodPassword,
		Credential: adminPwd,
	}, sql)
}

func (p *clusterProxyModelImpl) queryClusterCloudAdminPwdWithCache(
	ctx context.Context, projectId uint64, clusterId uint64,
) (string, error) {
	cacheKey := fmt.Sprintf("%d-%d", projectId, clusterId)
	if pwd, ok := p.clusterPwdCache.Get(cacheKey); ok {
		return pwd.(string), nil
	}
	adminPwd, err := p.queryClusterCloudAdminPwd(ctx, clusterId)
	if err != nil {
		return "", err
	}
	p.clusterPwdCache.SetDefault(cacheKey, adminPwd)
	return adminPwd, nil
}

func (p *clusterProxyModelImpl) queryClusterCloudAdminPwd(ctx context.Context, clusterId uint64) (string, error) {
	resp, bizErr := p.coreCli.GetClusterConfidentialInfo(
		ctx,
		&core.GetClusterConfidentialInfoReq{ClusterId: clusterId},
	)
	if bizErr != nil {
		return "", bizerr.WrapError(bizErr, "failed to get cluster admin account %d", clusterId)
	}
	if resp.Info == nil || resp.Info.EncryptedCloudAdminPasswd == "" {
		return "", bizerr_wrapper.NewInternalErr("cluster admin password is empty")
	}
	adminPwd, err := p.DecryptCloudAdminPassword(resp.GetInfo().GetEncryptedCloudAdminPasswd())
	if err != nil {
		return "", bizerr.WrapError(err, "failed to decrypt admin password")
	}
	return adminPwd, nil
}

func (p *clusterProxyModelImpl) queryClusterUserJwtToken(
	ctx context.Context, clusterId uint64, userId uint64,
) (*clusterproxy.ClusterOperator, error) {
	jwtResp, grpcErr := p.aclGrpcCli.GetDelegateJwtByUser(ctx, &acl.GetJwtByUserReq{
		ClusterId: clusterId,
		UserId:    userId,
		Email:     "",
	})
	if grpcErr != nil {
		return nil, grpcErr
	}

	return &clusterproxy.ClusterOperator{
		Username:   jwtResp.DbUser,
		AuthMethod: clusterproxy.AuthMethodToken,
		Credential: jwtResp.Jwt,
	}, nil
}