package grpc

// @Time    : 2022/11/14 4:47 PM
// @Author  : lvhongmeng

import (
	"context"
	"dataapps/commons/config"
	"dataapps/commons/logger"
	"fmt"
	"sync"

	"github.com/tidbcloud/dataservices/idl/pbgen/proto/dataapi"
	"github.com/tidbcloud/tiinf/errs"
	tierr "github.com/tidbcloud/tiinf/errs"
	tigrpc "github.com/tidbcloud/tiinf/transport/tigrpc/client"
	"go.uber.org/zap"
	"google.golang.org/grpc"
)

const (
	ERRACCESSDENY       = 1045
	DefaultSqlTimeoutMS = 60000
	grpcClienMaxMsgSize = 8 * 1024 * 1024
)

var (
	dataAPISvcOnce sync.Once
	dataAPISvcCli  dataapi.DataAPIServiceClient
)

func getDataAPISvcClient() dataapi.DataAPIServiceClient {
	dataAPISvcOnce.Do(func() {
		conn, err := tigrpc.NewClientConn(
			config.Global().GrpcConfig.DataAPISvcAddr,
			tigrpc.WithDefault(),
			tigrpc.WithGrpcOption(grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(grpcClienMaxMsgSize))),
		)
		if err != nil {
			panic(fmt.Sprintf("dial dataapi-svc: %s", err))
		}

		dataAPISvcCli = dataapi.NewDataAPIServiceClient(conn)
	})

	return dataAPISvcCli
}

func CreateSession(ctx context.Context, req *dataapi.CreateSessionReq) (resp *dataapi.CreateSessionResp, e error) {
	client := getDataAPISvcClient()

	logger.Info(ctx, "rpc/dataapi:CreateSession begin", logger.Redact("request", req))
	defer func() {
		if e != nil {
			logger.LogWithError(ctx, e, "rpc/dataapi:CreateSession failed")
		}
	}()
	resp, e = client.CreateSession(ctx, req)
	if e != nil || resp.BaseResp.ErrCode != 0 {
		logger.LogWithError(ctx, e, "rpc/dataapi:CreateSession first failed, we will try again latter.", zap.String("user", req.Credential.AccountId))
		return retryCreateSession(ctx, client, req)
	}
	if resp.BaseResp.ErrCode != 0 {
		return nil, errs.GetTiErrFromBaseResp(resp.BaseResp)
	}
	return resp, nil
}

func ExecSqlWithSession(ctx context.Context, req *dataapi.BasicEndpointReq) (resp *dataapi.ExecSqlResp, e error) {
	client := getDataAPISvcClient()

	if req.TimeoutMs == 0 {
		req.TimeoutMs = DefaultSqlTimeoutMS
	}
	logger.Info(ctx, "rpc/dataapi:ExecSqlWithSession begin", logger.Redact("request", req))
	defer func() {
		logger.LogWithError(ctx, e, "rpc/dataapi:ExecSqlWithSession failed")
	}()
	resp, e = client.ExecSqlWithSession(ctx, req)
	if e != nil {
		logger.LogWithError(ctx, e, "rpc/dataapi:ExecSqlWithSession first failed, we will try again latter.")
		// rpc internal error, retry one more time
		return retryExecutionWithSession(ctx, client, req)
	}
	if resp.BaseResp.ErrCode != 0 {
		return nil, errs.GetTiErrFromBaseResp(resp.BaseResp)
	}
	return resp, nil
}

func CloseSession(ctx context.Context, req *dataapi.CloseSessionReq) (resp *dataapi.CloseSessionResp, e error) {
	client := getDataAPISvcClient()

	logger.Info(ctx, "rpc/dataapi:CloseSession begin", logger.Redact("request", req))
	defer func() {
		if e != nil {
			logger.LogWithError(ctx, e, "rpc/dataapi:CloseSession failed", zap.String("session_id", req.SessionId), zap.String("user", req.Credential.AccountId))
		}
	}()
	resp, e = client.CloseSession(ctx, req)
	if e != nil {
		return nil, errs.NewInternalGRPC("failed to close session", e)
	}
	if resp.BaseResp.ErrCode != 0 {
		return nil, errs.GetTiErrFromBaseResp(resp.BaseResp)
	}
	return resp, nil
}

func ExecSqlDirectly(ctx context.Context, req *dataapi.BasicEndpointReq) (resp *dataapi.ExecSqlResp, e error) {
	client := getDataAPISvcClient()

	if req.TimeoutMs == 0 {
		req.TimeoutMs = DefaultSqlTimeoutMS
	}
	logger.Info(ctx, "rpc/dataapi:ExecSqlDirectly begin", logger.Redact("request", req))
	defer func() {
		if e != nil {
			logger.LogWithError(ctx, e, "rpc/dataapi:ExecSqlDirectly failed", zap.String("user", req.Credential.AccountId), zap.Error(e))
		}
	}()
	resp, e = client.ExecSqlDirectly(ctx, req)
	if e != nil {
		return resp, errs.NewInternalGRPC("failed to execute sql directly", e)
	}
	if resp.BaseResp.ErrCode != 0 {
		return nil, errs.GetTiErrFromBaseResp(resp.BaseResp)
	}
	return resp, nil
}

func CreateDataApp(ctx context.Context, req *dataapi.CreateDataAppReq) (*dataapi.CreateDataAppResp, error) {
	client := getDataAPISvcClient()

	logger.Info(ctx, fmt.Sprintf("rpc:CreateDataApp begin!payload:%+v", req))
	resp, e := client.CreateDataApp(ctx, req)
	if e != nil {
		return resp, tierr.NewInternalGRPC("CreateDataApp", e)
	}
	if resp.BaseResp.ErrCode != 0 {
		return resp, tierr.ParseBaseResp(ctx, resp.BaseResp)
	}
	return resp, nil
}

func CreateAPIKey(ctx context.Context, req *dataapi.CreateAPIKeyReq) (*dataapi.CreateAPIKeyResp, error) {
	client := getDataAPISvcClient()

	logger.Info(ctx, fmt.Sprintf("rpc:CreateAPIKey begin!payload:%+v", req))
	resp, e := client.CreateAPIKey(ctx, req)
	if e != nil {
		return resp, tierr.NewInternalGRPC("CreateAPIKey", e)
	}
	if resp.BaseResp.ErrCode != 0 {
		return resp, tierr.ParseBaseResp(ctx, resp.BaseResp)
	}
	return resp, nil
}

func ShowDataApp(ctx context.Context, req *dataapi.ShowDataAppReq) (*dataapi.ShowDataAppResp, error) {
	client := getDataAPISvcClient()

	logger.Info(ctx, fmt.Sprintf("rpc:ShowDataApp begin!payload:%+v", req))
	resp, e := client.ShowDataApp(ctx, req)
	if e != nil {
		return resp, tierr.NewInternalGRPC("ShowDataApp", e)
	}
	if resp.BaseResp.ErrCode != 0 {
		return resp, tierr.ParseBaseResp(ctx, resp.BaseResp)
	}
	return resp, nil
}

func ListDataApp(ctx context.Context, req *dataapi.ListDataAppReq) (*dataapi.ListDataAppResp, error) {
	client := getDataAPISvcClient()

	logger.Info(ctx, fmt.Sprintf("rpc:ListDataApp begin! payload:%+v", req))
	resp, e := client.ListDataApp(ctx, req)
	if e != nil {
		return resp, tierr.NewInternalGRPC("ListDataApp", e)
	}
	if resp.BaseResp.ErrCode != 0 {
		return resp, tierr.ParseBaseResp(ctx, resp.BaseResp)
	}
	return resp, nil
}

func DeleteDataApp(ctx context.Context, req *dataapi.DeleteDataAppReq) (*dataapi.DeleteDataAppResp, error) {
	client := getDataAPISvcClient()

	logger.Info(ctx, fmt.Sprintf("rpc:DeleteDataApp begin!payload:%+v", req))
	resp, e := client.DeleteDataApp(ctx, req)
	if e != nil {
		return resp, tierr.NewInternalGRPC("DeleteDataApp", e)
	}
	if resp.BaseResp.ErrCode != 0 {
		return resp, tierr.ParseBaseResp(ctx, resp.BaseResp)
	}
	return resp, nil
}

func DeleteAPIKey(ctx context.Context, req *dataapi.DeleteAPIKeyReq) (*dataapi.DeleteAPIKeyResp, error) {
	client := getDataAPISvcClient()

	logger.Info(ctx, fmt.Sprintf("rpc:DeleteAPIKey begin!payload:%+v", req))
	resp, e := client.DeleteAPIKey(ctx, req)
	if e != nil {
		return resp, tierr.NewInternalGRPC("DeleteAPIKey", e)
	}
	if resp.BaseResp.ErrCode != 0 {
		return resp, tierr.ParseBaseResp(ctx, resp.BaseResp)
	}
	return resp, nil
}

func ListAPIKey(ctx context.Context, req *dataapi.ListAPIKeyReq) (*dataapi.ListAPIKeyResp, error) {
	client := getDataAPISvcClient()

	logger.Info(ctx, fmt.Sprintf("rpc:ListAPIKey begin!payload:%+v", req))
	resp, e := client.ListAPIKey(ctx, req)
	if e != nil {
		return resp, tierr.NewInternalGRPC("ListAPIKey", e)
	}
	if resp.BaseResp.ErrCode != 0 {
		return resp, tierr.ParseBaseResp(ctx, resp.BaseResp)
	}
	return resp, nil
}

func ShowAPIKey(ctx context.Context, req *dataapi.ShowAPIKeyReq) (*dataapi.ShowAPIKeyResp, error) {
	client := getDataAPISvcClient()

	logger.Info(ctx, fmt.Sprintf("rpc:ShowAPIKey begin!payload:%+v", req))
	resp, e := client.ShowAPIKey(ctx, req)
	if e != nil {
		return resp, tierr.NewInternalGRPC("ShowAPIKey", e)
	}
	if resp.BaseResp.ErrCode != 0 {
		return resp, tierr.ParseBaseResp(ctx, resp.BaseResp)
	}
	return resp, nil
}

func EndpointCodeExample(ctx context.Context, req *dataapi.EndpointCodeExampleReq) (*dataapi.EndpointCodeExampleResp, error) {
	client := getDataAPISvcClient()

	logger.Info(ctx, "rpc:EndpointCodeExample begin", logger.Redact("request", req))
	resp, e := client.EndpointCodeExample(ctx, req)
	if e != nil {
		return resp, tierr.NewInternalGRPC("EndpointCodeExample", e)
	}
	if resp.BaseResp.ErrCode != 0 {
		return resp, tierr.ParseBaseResp(ctx, resp.BaseResp)
	}
	return resp, nil
}

func retryExecutionWithSession(ctx context.Context, client dataapi.DataAPIServiceClient, req *dataapi.BasicEndpointReq) (*dataapi.ExecSqlResp, error) {
	logger.Info(ctx, "rpc/dataapi:ExecSqlWithSession retry", logger.Redact("request", req))
	resp, e := client.ExecSqlWithSession(ctx, req)
	if e != nil {
		return nil, errs.NewInternalGRPC("failed to execute sql with session", e)
	}
	if resp.BaseResp.ErrCode != 0 {
		return nil, errs.GetTiErrFromBaseResp(resp.BaseResp)
	}
	return resp, nil
}

func retryCreateSession(ctx context.Context, client dataapi.DataAPIServiceClient, req *dataapi.CreateSessionReq) (*dataapi.CreateSessionResp, error) {
	logger.Info(ctx, "rpc/dataapi:CreateSession retry", logger.Redact("request", req))
	resp, e := client.CreateSession(ctx, req)
	if e != nil {
		return nil, errs.NewInternalGRPC("failed to create session", e)
	}
	if resp.BaseResp.ErrCode != 0 {
		return nil, errs.GetTiErrFromBaseResp(resp.BaseResp)
	}
	return resp, nil
}

func AuthAPIKey(ctx context.Context, req *dataapi.AuthAPIKeyReq) (*dataapi.AuthAPIKeyResp, error) {
	client := getDataAPISvcClient()
	logger.Info(ctx, "rpc/dataapi:AuthAPIKey", zap.Any("req", req))

	resp, e := client.AuthAPIKey(ctx, req)
	logger.LogWithError(ctx, e, "rpc/dataapi:AuthAPIKey complete", zap.Any("resp", resp))
	if e != nil {
		return resp, tierr.NewInternalGRPC("AuthAPIKey", e)
	}
	if resp.BaseResp.ErrCode != 0 {
		return resp, tierr.ParseBaseResp(ctx, resp.BaseResp)
	}
	return resp, nil

}

func ParseOneStmt(ctx context.Context, req *dataapi.ParseOneStmtReq) (*dataapi.ParseOneStmtResp, error) {
	client := getDataAPISvcClient()
	logger.Info(ctx, "rpc/dataapi:ParseOneStmt", zap.Any("req", req))
	resp, err := client.ParseOneStmt(ctx, req)
	logger.LogWithError(ctx, err, "rpc/dataapi:ParseOneStmt complete", zap.Any("resp", resp))
	if err != nil {
		return resp, tierr.NewInternalGRPC("ParseOneStmt", err)
	}
	if resp.BaseResp.ErrCode != 0 {
		return resp, tierr.ParseBaseResp(ctx, resp.BaseResp)
	}
	return resp, nil
}

func GetClusterStatus(ctx context.Context, req *dataapi.GetClusterStatusReq) (*dataapi.GetClusterStatusResp, error) {
	client := getDataAPISvcClient()
	logger.Info(ctx, "rpc/dataapi:GetClusterStatus", zap.Any("req", req))
	resp, err := client.GetClusterStatus(ctx, req)
	logger.LogWithError(ctx, err, "rpc/dataapi:GetClusterStatus complete", zap.Any("resp", resp))
	if err != nil {
		return resp, tierr.NewInternalGRPC("ParseOneStmt", err)
	}
	if resp.BaseResp.ErrCode != 0 {
		return resp, tierr.GetTiErrFromBaseResp(resp.BaseResp)
	}
	return resp, nil
}