package service

import (
	"context"
	"encoding/json"
	"fmt"
	"strings"
	"time"

	"dataservices/commons/config"
	"dataservices/commons/constants"
	comerrs "dataservices/commons/errs"
	"dataservices/commons/grpc"
	"dataservices/commons/http"
	"dataservices/commons/metrics"
	"dataservices/commons/utils"
	dto "dataservices/internal/dto/dataapi"
	dto2 "dataservices/internal/dto/dataapi_gateway"
	"dataservices/internal/entity"
	"dataservices/internal/repository"

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

const (
	SESSION_EXPIRE_SEC = 2 * 14400
	TIME_OUT           = 30000
	LIMIT_ROWS         = 200
)

var (
	EXCLUDEDBANDSYS = []string{"__TiDB_BR_Temporary_mysql", "tidbcloud_dm_meta", "tidb_cdc", "lightning_task_info", "INFORMATION_SCHEMA", "mysql", "PERFORMANCE_SCHEMA"}
)

func Statements(ctx context.Context, createdBy string, in entity.StatementParamV2) (res []dto.EndpointDataRes, err error) {

	defer recordCounterMetric(metrics.EndpointSqlRunCounter, err)

	startTime := utils.Now()
	defer func() {
		var (
			args    []byte
			endTime utils.JsonTime
			status  string
		)
		if len(in.Args) > 0 {
			args, _ = json.Marshal(in.Args)
		}
		endTime = utils.Now()
		var errormessage string
		if err != nil {
			status = "failed"
			errormessage = err.Error()
		} else {
			status = "succeed"
		}
		message := entity.DataAPIConsoleMessage{
			OrgId:        in.OrgId,
			ProjectId:    in.ProjectId,
			DataappId:    in.DataappId,
			ClusterId:    in.ClusterId,
			Sql:          in.Sql,
			Args:         string(args),
			CreatedBy:    createdBy,
			StartTime:    startTime,
			EndTime:      endTime,
			Duration:     endTime.UnixMilli() - startTime.UnixMilli(),
			Status:       status,
			ErrorMessage: errormessage,
		}
		repository.CreateDataAPIConsoleMessage(message)
	}()

	return ExecSqlTemplateByHttpV2(ctx, in)
}

func ShowDBs(ctx context.Context, in entity.ShowDBParam, credential dto2.Credential) ([]string, error) {
	excludeDB := ""
	for _, db := range EXCLUDEDBANDSYS {
		excludeDB = excludeDB + "'" + db + "',"
	}
	sql := "select schema_name from information_schema.schemata where schema_name not in (" + excludeDB[:len(excludeDB)-1] + ") group by schema_name order by lower(schema_name);"
	if in.ShowSysDB == "true" {
		sql = "select schema_name from information_schema.schemata group by schema_name order by lower(schema_name);"
	}

	res, err := ExeSQLByHttp(ctx, &dto2.ExecSqlReq{
		ProjectId:  in.ProjectId,
		ClusterId:  in.ClusterId,
		Database:   "",
		Sql:        sql,
		LimitRows:  LIMIT_ROWS,
		TimeoutMs:  TIME_OUT,
		Credential: credential,
	})
	if err != nil {
		return nil, err
	}

	dbList := make([]string, 0)
	for _, v := range res.Data.Rows {
		dbList = append(dbList, v["schema_name"])
	}
	return dbList, nil

}

func ShowTables(ctx context.Context, in entity.ShowTableParam) ([]string, error) {
	sql := "show tables;"
	res, err := ExeSQLByHttp(ctx, &dto2.ExecSqlReq{
		ProjectId: in.ProjectId,
		ClusterId: in.ClusterId,
		Database:  in.DBName,
		Sql:       sql,
		LimitRows: LIMIT_ROWS,
		TimeoutMs: TIME_OUT,
		Credential: dto2.Credential{
			Type: dataapi.CrtType_AUTH0.String(),
			Key:  in.Auth0Token,
		},
	})
	if err != nil {
		return nil, err
	}

	dbList := make([]string, 0)
	for _, v := range res.Data.Rows {
		dbList = append(dbList, v["Tables_in_"+in.DBName])
	}
	return dbList, nil
}

func ShowVectorFields(ctx context.Context, in entity.ShowVectorFieldParam) ([]string, error) {
	sql := "desc " + in.TableName + ";"
	res, err := ExeSQLByHttp(ctx, &dto2.ExecSqlReq{
		ProjectId: in.ProjectId,
		ClusterId: in.ClusterId,
		Database:  in.DBName,
		Sql:       sql,
		LimitRows: LIMIT_ROWS,
		TimeoutMs: TIME_OUT,
		Credential: dto2.Credential{
			Type: dataapi.CrtType_AUTH0.String(),
			Key:  in.Auth0Token,
		},
	})
	if err != nil {
		return nil, err
	}

	vectorFieldList := make([]string, 0)
	for _, v := range res.Data.Rows {
		if strings.Contains(v["Type"], "vector") {
			vectorFieldList = append(vectorFieldList, v["Field"])
		}
	}
	return vectorFieldList, nil
}

func ExecSqlTemplateByHttpV2(ctx context.Context, in entity.StatementParamV2) (res []dto.EndpointDataRes, err error) {
	args, argValues, err := Convert2TestArgs(in)
	if err != nil {
		return nil, err
	}

	if err := checkSqlSyntax(in.Sql); err != nil {
		return nil, err
	}
	matchs := argsParamsRegexp(in.Sql)
	for _, match := range matchs {
		if match[1] == "" {
			return nil, errs.NewBadRequest("Invalid endpoint sql params definition", nil)
		}
		if match[1] == constants.EndpointPageParam || match[1] == constants.EndpointPageSizeParam {
			return nil, comerrs.ErrPaginationConflict
		}
	}

	cluster, err := grpc.GetCluster(ctx, in.OrgId, in.ProjectId, in.ClusterId)
	if err != nil {
		tilog.Error(ctx, "GetClusterWithoutPermCheck failed", zap.Error(err), zap.Uint64("cluster_id", in.ClusterId))
		return nil, err
	}

	sqlRes, err := http.ExecSqlTemplateV2WithProvider(ctx, cluster.RegionName, http.ProviderForRegionalHostPrefix(cluster.CloudProviderName, cluster.GetDeployType()), &dto2.ExecSqlTemplateReqV2{
		ProjectId:   in.ProjectId,
		ClusterId:   in.ClusterId,
		Database:    in.Database,
		SqlTemplate: in.Sql,
		Settings:    in.Settings,
		Args:        args,
		ArgValues:   argValues,
		SessionId:   in.SessionID,
		Credential: dto2.Credential{
			Type:      dataapi.CrtType_AUTH0.String(),
			Key:       in.Auth0Token,
			AccountID: in.CreatedBy,
		},
		Source: entity.DataAPISourceDataAPIConsole,
	})
	tilog.Info(ctx, "ExecSqlTemplateV2 result", zap.Any("sqlRes", sqlRes))
	if err != nil {
		return nil, err
	}
	res = convertExecSqlRespToEndpointDataRess(sqlRes)
	return res, nil
}

func Convert2TestArgs(in entity.StatementParamV2) ([]entity.EndpointArgs, []map[string]any, error) {
	if len(in.Args) == 0 {
		return nil, nil, nil
	}
	argsLen := len(in.Args[0])
	args := make([]entity.EndpointArgs, 0, argsLen)
	argValues := make([]map[string]any, 0, len(in.Args))
	for idx, v := range in.Args {
		_argValues := make(map[string]any, argsLen)
		for _, arg := range v {
			if arg.Name == "" || !validDataType[arg.Type] {
				return nil, nil, errs.NewBadRequest("Invalid sql params name/type", nil)
			}
			_argValues[arg.Name] = arg.Value
			if idx == 0 {
				args = append(args, arg.EndpointArgs)
			}
		}
		argValues = append(argValues, _argValues)
	}
	return args, argValues, nil
}

func CreateSession(ctx context.Context, in dto.CreateSessionIn) (res *dto.CreateSessionOut, err error) {
	defer recordCounterMetric(metrics.EndpointSessionCounter, err)
	cluster, err := grpc.GetClusterWithoutPermCheck(ctx, in.ClusterId)
	if err != nil {
		return nil, err
	}

	sessionRes, err := http.CreateSessionWithProvider(ctx, cluster.RegionName, http.ProviderForRegionalHostPrefix(cluster.CloudProviderName, cluster.GetDeployType()), &dto2.CreateSessionReq{
		ProjectId: in.ProjectId,
		ClusterId: in.ClusterId,
		Database:  in.Database,
		ExpireSec: SESSION_EXPIRE_SEC,
		Credential: dto2.Credential{
			Type:      dto2.TypeCredentialAuth0,
			Key:       in.Auth0Token,
			AccountID: in.CreatedBy,
		},
	})
	if err != nil {
		return nil, err
	}
	res = &dto.CreateSessionOut{
		SessionID: sessionRes.SessionId,
	}
	return res, nil
}

func ExeSQLByHttp(ctx context.Context, in *dto2.ExecSqlReq) (*dto2.ExecSqlResp, error) {
	regionName := "us-west-2"
	provider := ""
	if config.Global().ServerConf.Env == "prod" {
		cluster, err := grpc.GetClusterWithoutPermCheck(ctx, in.ClusterId)
		if err != nil {
			return nil, err
		} else if cluster.ProjectId != in.ProjectId {
			return nil, errs.NewBadRequest("Cluster does not belong to the project", nil)
		}
		regionName = cluster.RegionName
		provider = http.ProviderForRegionalHostPrefix(cluster.CloudProviderName, cluster.GetDeployType())
	}

	exeSqlRes, err := http.ExecSqlWithProvider(ctx, regionName, provider, in)
	if err != nil {
		return nil, err
	}
	return exeSqlRes, nil
}

func convertExecSqlRespToEndpointDataRess(multRes *dto2.ExecSqlResp) []dto.EndpointDataRes {
	result := make([]dto.EndpointDataRes, 0, 1)
	sqlRes := multRes.Data
	if sqlRes.Result.Code != 0 && sqlRes.Result.Code != 200 {
		result = append(result, dto.EndpointDataRes{
			ErrCode:    sqlRes.Result.Code,
			ErrMessage: sqlRes.Result.Message,
		})
		return result
	}
	columns := make([]dto.DBColumn, 0, len(sqlRes.Columns))
	rows := make([][]string, 0, len(sqlRes.Rows))
	for _, column := range sqlRes.Columns {
		columns = append(columns, dto.DBColumn{
			Col:      column.Col,
			DataType: column.DataType,
			Nullable: column.Nullable,
		})
	}
	if len(sqlRes.Columns) > 0 {
		for _, rowMap := range sqlRes.Rows {
			row := make([]string, len(sqlRes.Columns))
			for i, col := range sqlRes.Columns {
				row[i] = fmt.Sprintf("%v", rowMap[col.Col])
			}
			rows = append(rows, row)
		}
	} else {
		if len(sqlRes.Rows) > 0 {
			queryError := ""
			for _, rowMap := range sqlRes.Rows {
				if success, ok := rowMap["success"]; ok {
					if success == "false" {
						if index, ok := rowMap["index"]; ok {
							if message, ok := rowMap["message"]; ok {
								queryError = queryError + "Test Value Group [" + index + "]: " + message + ";\n"
							}
						}
					}
				}
			}
			if queryError != "" {
				result = append(result, dto.EndpointDataRes{
					ErrCode:    errs.ErrCodeInternalServerError,
					ErrMessage: queryError,
				})
				return result
			}
		}
	}
	startTime := time.UnixMilli(sqlRes.Result.StartMs)
	endTime := time.UnixMilli(sqlRes.Result.EndMs)
	res := dto.EndpointDataRes{
		Columns:      columns,
		Rows:         rows,
		StartMs:      sqlRes.Result.StartMs,
		EndMs:        sqlRes.Result.EndMs,
		Latency:      endTime.Sub(startTime).String(),
		RowCount:     sqlRes.Result.RowCount,
		RowAffect:    sqlRes.Result.RowAffect,
		Limit:        sqlRes.Result.Limit,
		Query:        sqlRes.Result.Message,
		WarnCount:    sqlRes.Result.WarnCount,
		WarnMessages: sqlRes.Result.WarnMessages,
	}
	result = append(result, res)
	return result
}

func convertExecSqlRespToEndpointResult(multRes *dto2.ExecSqlResp) *dto.EndpointResult {
	if multRes == nil {
		return nil
	}

	result := dto.NewEndpointResult()
	result.Type = constants.EndpointResTypeSQL

	sqlRes := multRes.Data.Result
	if sqlRes.Code != 0 && sqlRes.Code != 200 {
		result.Data.Result.Code = int(sqlRes.Code)
		result.Data.Result.Message = sqlRes.Message
		return &result
	}

	startTime := time.UnixMilli(sqlRes.StartMs)
	endTime := time.UnixMilli(sqlRes.EndMs)
	result.Data.Columns = convertToDBColumns(multRes.Data.Columns)
	result.Data.Rows = convertToDBRows(multRes.Data.Rows)
	result.Data.Result.Code = 200
	result.Data.Result.StartMs = sqlRes.StartMs
	result.Data.Result.EndMs = sqlRes.EndMs
	result.Data.Result.Latency = endTime.Sub(startTime).String()
	result.Data.Result.RowCount = sqlRes.RowCount
	result.Data.Result.RowAffect = sqlRes.RowAffect
	result.Data.Result.Limit = sqlRes.Limit
	result.Data.Result.Message = sqlRes.Message
	return &result
}

func convertToDBColumns(cols []entity.Column) []dto.DBColumn {
	colRes := make([]dto.DBColumn, 0, len(cols))
	for _, col := range cols {
		colRes = append(colRes, dto.DBColumn{
			Col:      col.Col,
			DataType: col.DataType,
			Nullable: col.Nullable,
		})
	}
	return colRes
}

func convertToDBRows(rows []entity.Row) []dto.DBRow {
	rowRes := make([]dto.DBRow, 0, len(rows))
	for _, row := range rows {
		dbRow := dto.DBRow{}
		for k, v := range row {
			dbRow[k] = v
		}
		rowRes = append(rowRes, dbRow)
	}
	return rowRes
}