package controller

import (
	"context"
	"dataservices/commons/config"
	"dataservices/commons/constants"
	"dataservices/commons/contextvar"
	"dataservices/commons/logger"
	"dataservices/commons/measure_report"
	"dataservices/commons/metrics"
	"dataservices/commons/utils"
	"dataservices/dbhubsdk"
	dto "dataservices/internal/dto/dataapi"
	service "dataservices/internal/service/dataapi_executor"
	"fmt"
	"strconv"
	"time"
	"unicode/utf8"
	"unsafe"

	"github.com/google/uuid"
	"github.com/tidbcloud/dataservices/commons/bizerr"
	"github.com/tidbcloud/tiinf/sdk/measure"

	"github.com/tidbcloud/dataservices/idl/pbgen/proto/dataapi"
	"github.com/tidbcloud/tiinf/errs"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
)

type dataAPIExecutorService struct {
	dataapi.UnimplementedDataAPIExecutorServiceServer
}

func NewDataAPIExecutorService() dataapi.DataAPIExecutorServiceServer {
	return &dataAPIExecutorService{}
}

func (c *dataAPIExecutorService) GetDatabaseMeta(ctx context.Context, in *dataapi.GetDatabaseMetaReq) (*dataapi.GetDatabaseMetaResp, error) {
	logRequest(ctx, in, "GetDatabaseMeta begin")
	res, err := service.GetDatabaseMeta(ctx, in)
	if err != nil {
		logger.LogWithError(ctx, err, "DS-Region-Dataapi-grpc-GetDatabaseMeta execution failed.", zap.String("region", config.Global().AwsConf.Region), zap.String("user", in.Credential.GetAccountId()))
		return nil, err
	}
	resp := dto.ConvertGetDatabaseMetaResp(res, in.Database)
	return resp, nil
}

func (c *dataAPIExecutorService) ExecSql(ctx context.Context, in *dataapi.BasicEndpointReqV2) (*dataapi.ExecSqlRespV2, error) {
	endpointCounterAdd(1)
	l := tilog.WithContextL(ctx).With(
		zap.String("region", config.Global().AwsConf.Region),
		zap.Uint64("projectId", in.ProjectId),
		zap.Uint64("clusterId", in.ClusterId),
		zap.String("source", in.Source),
		zap.String("sessionId", in.SessionId),
		zap.String("cretType", in.Credential.GetType().String()),
		logger.Redact("cretAccount", in.Credential.GetAccountId()),
	)
	l = l.With(zap.String("cretKeyMd5WithSalt", utils.Md5WithDefaultSalt(in.Credential.GetKey())))
	ctx = contextvar.WithAccountID(contextvar.WithLogger(ctx, l), in.Credential.GetAccountId())

	logRequest(ctx, in, "ExecSql begin")
	if in.TimeoutMs < 1 {
		in.TimeoutMs = constants.ExecSQLDefaultTimeoutMs
	}
	if in.LimitRows < 1 {
		in.LimitRows = constants.ExecSQLDefaultRowlimit
	}
	ctx, cancel := context.WithTimeout(ctx, time.Millisecond*time.Duration(in.TimeoutMs))
	defer cancel()
	measureRequest(ctx, in)
	stmtResults, err := service.ExecSql(ctx, in)
	bytesCount := measureNetwork(ctx, stmtResults, in)
	if err != nil {
		logger.LogWithError(ctx, err, "DS-Region-Dataapi-grpc-ExecSql execute failed.", zap.String("sql", in.Sql))
		if !utf8.ValidString(err.Error()) {
			errStr := utils.SanitizeString(err.Error())
			return nil, errs.NewError(bizerr.ErrInvalidUserSql, errStr, nil)
		}
		return nil, err
	}
	if len(stmtResults) > 0 && stmtResults[len(stmtResults)-1].ErrCode != 0 {
		lastStmtRes := stmtResults[len(stmtResults)-1]
		logLevel := zap.ErrorLevel
		if dbhubsdk.IsCausedByUser(uint16(stmtResults[len(stmtResults)-1].ErrCode), stmtResults[len(stmtResults)-1].ErrMessage) {
			logLevel = zap.WarnLevel
		}
		logger.Log(ctx, logLevel, fmt.Sprintf("DS-Region-Dataapi-grpc-ExecSql failed with last sql, err msg: %s", lastStmtRes.ErrMessage),
			zap.String("sql", in.Sql),
			zap.Any("args", in.Args),
			zap.Any("arg_values", in.ArgValues),
			zap.Int64("err_code", lastStmtRes.ErrCode),
			zap.String("err_message", lastStmtRes.ErrMessage),
		)
	} else {
		logger.Info(ctx, "DS-Region-Dataapi-grpc-ExecSql succeed",
			zap.String("sql", in.Sql),
			zap.Any("args", in.Args),
			zap.Any("arg_values", in.ArgValues),
		)
	}
	logger.Info(ctx, "ExecSql(dataapi_executor/executor.go)", zap.Any("bytesCount", bytesCount))
	return &dataapi.ExecSqlRespV2{StmtResults: stmtResults, BytesCount: bytesCount}, nil
}

func (c *dataAPIExecutorService) CreateSession(ctx context.Context, in *dataapi.CreateSessionReq) (*dataapi.CreateSessionResp, error) {
	l := tilog.WithContextL(ctx).With(
		zap.String("region", config.Global().AwsConf.Region),
		zap.Uint64("projectId", in.ProjectId),
		zap.Uint64("clusterId", in.ClusterId),
		zap.String("cretType", in.Credential.GetType().String()),
		logger.Redact("cretAccount", in.Credential.GetAccountId()),
	)
	l = l.With(zap.String("cretKeyMd5WithSalt", utils.Md5WithDefaultSalt(in.Credential.GetKey())))
	ctx = contextvar.WithLogger(ctx, l)

	logRequest(ctx, in, "CreateSession begin")
	sessionId, err := service.CreateSession(ctx, in)
	if err != nil {
		logger.LogWithError(ctx, err, "DS-Region-Dataapi-grpc-CreateSession execution failed.")
		return nil, err
	}
	logger.Info(ctx, "createSession succeed.", zap.String("session_id", sessionId), zap.Uint64("expire_sec", in.ExpireSec))
	resp := &dataapi.CreateSessionResp{
		SessionId: sessionId,
	}
	return resp, nil
}

func (c *dataAPIExecutorService) CloseSession(ctx context.Context, in *dataapi.CloseSessionReq) (*dataapi.CloseSessionResp, error) {
	l := tilog.WithContextL(ctx).With(
		zap.String("region", config.Global().AwsConf.Region),
		zap.Uint64("projectId", in.ProjectId),
		zap.Uint64("clusterId", in.ClusterId),
		zap.String("sessionId", in.SessionId),
		zap.String("cretType", in.Credential.GetType().String()),
		logger.Redact("cretAccount", in.Credential.GetAccountId()),
	)
	l = l.With(zap.String("cretKeyMd5WithSalt", utils.Md5WithDefaultSalt(in.Credential.GetKey())))
	ctx = contextvar.WithLogger(ctx, l)

	logRequest(ctx, in, "CloseSession begin")
	err := service.CloseSession(ctx, in)
	if err != nil {
		logger.LogWithError(ctx, err, "DS-Region-Dataapi-grpc-CloseSession failed.")
		return nil, errs.NewBadRequest("Close session failed, "+err.Error(), nil)
	}
	logger.Info(ctx, "CloseSession succeed.")
	resp := &dataapi.CloseSessionResp{}
	return resp, nil
}

func (c *dataAPIExecutorService) ParseOneStmt(ctx context.Context, in *dataapi.ParseOneStmtReq) (*dataapi.ParseOneStmtResp, error) {
	tilog.Info(ctx, "ParseOneStmt begin", zap.Any("request", in))
	if in.Sql == "" {
		tilog.Error(ctx, "DS-Region-Dataapi-grpc-ParseOneStmt sql is required", zap.String("region", config.Global().AwsConf.Region))
		return nil, errs.NewBadRequest("Sql is required.", nil)
	}
	stmtType, err := service.ParseOneStmt(ctx, in.Sql)
	if err != nil {
		tilog.Error(ctx, "DS-Region-Dataapi-grpc-ParseOneStmt execution failed", zap.Error(err), zap.String("sql", in.Sql), zap.String("region", config.Global().AwsConf.Region))
		return nil, errs.NewBadRequest("Parse sql failed, "+err.Error(), nil)
	}
	resp := new(dataapi.ParseOneStmtResp)
	resp.StmtType = dto.StmtTypeMap[stmtType]
	return resp, nil
}

func logRequest(ctx context.Context, req interface{}, msg string) {
	tilog.Info(ctx, msg, logger.Redact("request", req))
}

func createMeasure(param *dataapi.BasicEndpointReqV2, cost int64, sku string) error {

	return measure_report.Report.PutMeasure(measure.BillingMeasure{
		TenantID:      int64(param.OrgId),
		CostID:        uuid.NewString(),
		Cost:          cost,
		CostTimestamp: time.Now().Unix(),
		Labels: map[string]string{
			constants.MeasureLabelKeyAppId1:     strconv.FormatUint(param.DataappId, 10),
			constants.MeasureLabelKeyAppName2:   param.DataappName,
			constants.MeasureLabelKeyProjectId3: strconv.FormatUint(param.ProjectId, 10),
			constants.MeasureLabelKeyEndpoint4:  param.EndpointName,
			constants.MeasureLabelKeyClusterId5: strconv.FormatUint(param.ClusterId, 10),
		},
		Unit: sku,
	})
}
func measureRequest(ctx context.Context, param *dataapi.BasicEndpointReqV2) {

	if len(param.AppType) < 1 || param.AppType != constants.NeedMeasure {
		tilog.Info(ctx, "measureRequest  error ", zap.String("AppType", param.AppType))
		return
	}

	metrics.MeasureRequestCounter.WithLabelValues(strconv.FormatUint(param.OrgId, 10), strconv.FormatUint(param.DataappId, 10), param.EndpointName, param.AccessKey).Inc()

	err := createMeasure(param, 1, config.Global().SkuCon.SkuRequest)
	if err != nil {
		tilog.Error(ctx, "measureRequest  put err ", zap.Error(err))
	}
}

func measureNetwork(ctx context.Context, stmtResults []*dataapi.ExecStmtResult, param *dataapi.BasicEndpointReqV2) int64 {
	var byteCnt = int64(len(param.Args)) + int64(len(param.ArgValues))
	for _, stm := range stmtResults {
		byteCnt += int64(unsafe.Sizeof(stm))
	}

	if len(param.AppType) > 0 && param.AppType == constants.NeedMeasure {
		metrics.MeasureNetworkCounter.WithLabelValues(strconv.FormatUint(param.OrgId, 10), strconv.FormatUint(param.DataappId, 10), param.EndpointName, param.AccessKey).Add(float64(byteCnt))
		err := createMeasure(param, byteCnt, config.Global().SkuCon.SkuNetWork)
		if err != nil {
			tilog.Error(ctx, "measureNetwork  put err ", zap.Error(err))
		}
	}

	return byteCnt
}