package dataapi_gateway

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io/ioutil"
	"net"
	"net/http"
	"sort"
	"strconv"
	"testing"
	"time"

	// _ "github.com/go-sql-driver/mysql"
	_ "github.com/tidbcloud/go-mysql-driver"

	"database/sql"
	"dataservices/api"
	"dataservices/commons/apikeygenerator"
	"dataservices/commons/config"
	"dataservices/commons/database"
	"dataservices/commons/httpclient"
	"dataservices/commons/metrics"
	"dataservices/commons/sentry"
	"dataservices/dbhubsdk"
	controller "dataservices/internal/controller/dataapi_executor"
	controller_gateway "dataservices/internal/controller/dataapi_gateway"
	dto "dataservices/internal/dto/dataapi"
	dto2 "dataservices/internal/dto/dataapi_gateway"
	"dataservices/internal/entity"

	grpc_middleware "github.com/grpc-ecosystem/go-grpc-middleware"
	"github.com/pingcap/tidb/pkg/parser"
	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/dataservices/idl/pbgen/proto/dataapi"
	"github.com/tidbcloud/tiinf/errs"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"github.com/tidbcloud/tiinf/sdk/rbac/helperimpl"
	"github.com/tidbcloud/tiinf/transport/http/tigin"
	tigrpc "github.com/tidbcloud/tiinf/transport/tigrpc/server"
	"go.uber.org/zap"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"
	"gorm.io/driver/mysql"
	"gorm.io/gorm"
	"gorm.io/gorm/logger"
)

func TestHelloworld(t *testing.T) {
	argsList := []*entity.EndpointArgs{}
	err := json.Unmarshal([]byte(""), &argsList)
	require.NoError(t, err)
}

/*
//prepare data in test TiDB cluster.

use test;
create table t1 (id int key, name varchar(100));
create table t2 (id int key, name varchar(100));
insert into t1 values (1,'a'),(2,'b');
insert into t2 values (1, 'a\'b'),(2, 'a\"b');
create database sample_data;
use sample_data;
create table github_events (id int);
insert into github_events values (1);
*/

func TestGatewayHttpService1(t *testing.T) {
	setupCentral("/Users/cs/.infra-api/dataapi1.yaml")
	centralHttpAddr := fmt.Sprintf("http://0.0.0.0:%v", config.Global().ServerConf.HttpPort)
	setupExecutor("/Users/cs/.infra-api/dataapi_executor.yaml")
	setupGateway("/Users/cs/.infra-api/dataapi_gateway.yaml")
	//err := config.InitConfig("/Users/cs/.infra-api/dataapi_gateway.yaml")
	//require.NoError(t, err)
	// wait grpc server started.
	time.Sleep(500 * time.Millisecond)

	gatewayHttpAddr := fmt.Sprintf("http://0.0.0.0:%v", config.Global().ServerConf.HttpPort)

	//ctx := context.Background()
	projectId := uint64(43563)
	clusterId := uint64(270048)
	clusterIdStr := "270048"
	appId := uint64(5)
	databaseName := "test"
	testEmail := "chenshuang@pingcap.com"
	orgId := 30061
	publicKey := "DsioGVSB"

	createEndpointParam := &entity.CreateEndpointParam{
		Name:         "cs-endpoint-test00",
		Endpoint:     "/cstest00",
		Method:       "get",
		ClusterIdStr: clusterIdStr,
		Database:     databaseName,
		SqlTemplate:  "select * from t1 where id = ${id};",
		Args: []entity.EndpointArgs{
			{
				Name:     "id",
				Type:     "NUMBER",
				Required: 1,
			},
		},
		Settings: entity.EndpointSettings{
			TimeOut:   10000,
			RowsLimit: 1000,
		},
	}
	endpointID := testCreateEndpoint(t, centralHttpAddr, testEmail, publicKey, orgId, int(projectId), int(appId), createEndpointParam)
	defer func() {
		testDeleteEndpoint(t, centralHttpAddr, testEmail, orgId, int(projectId), int(appId), int(endpointID))
	}()

	req := dto2.TestEndpointReq{
		ProjectId:           projectId,
		AppId:               appId,
		EndpointId:          endpointID,
		EndpointType:        "",
		Args:                nil,
		ReturnAllStmtResult: false,
		Credential: dto2.Credential{
			Type: "AUTH0",
			Key:  "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCIsImtpZCI6Im1oSjk3OVJEQlFhOWU1U1I4ZUxwRCJ9.eyJodHRwczovL3RpZGJjbG91ZC5jb20vbG9naW5fcmVjZWl2ZWRfYXQiOjE2Nzc0NjM5NjUyMTksImVtYWlsIjoiaGFueGlhb21lbmcxMTFAZ21haWwuY29tIiwiZW1haWxfdmVyaWZpZWQiOnRydWUsImlzcyI6Imh0dHBzOi8vdGlkYi1kZXYudXMuYXV0aDAuY29tLyIsImF1ZCI6Iktpd1dhakxKM3B0bzhTc1o3RjM1TmFxTmh3bm9SaDFjIiwiaWF0IjoxNjc3NDYzOTY1LCJleHAiOjE2Nzc0OTk5NjUsInN1YiI6Imdvb2dsZS1vYXV0aDJ8MTA3MDYyNjE0NDEzNTMyMjY4MTE4IiwiYXRfaGFzaCI6ImxpdFpQT0Q0U1RzSDNfam1ZTDFPVnciLCJzaWQiOiJ4ZXNHT1ptOE9RTkVidkxYRXYtNm5VOTNvQmFGeW1QNyIsIm5vbmNlIjoiRDBGT1lzTHJ0aFNQYTUtQmRUbHVBRGpaYmRRR2NWMGcifQ.yTIVyiqOkbr7rvKUVeJ9-dckG3sWBgrblAzHb5Dz9ajGZYkqtciWaA-OPLJuYYS9PAuz1gyhrRCEuScHh0NO_T6XqRgoiRiGPf86rcj4moCCu2v4_pUXdp-fyaVL57wgZ6gGfmHdjvK7LVEZSUXMqsZivoKbTdRp2kOOLxiV5S1wxNZHo1wQCwIsiRLNGPpRIWtv1TWxCbz0Yk2FdjYQ1CUd8x1VIeZ8IQ6cmRX5yizHtwpBm-zkH-vIH1BVFBDP7D_xtEq7BtWbCY1seFs6xnOjbEW17-U_MGkvFWAZclzMaU72_oULGgV6R31Ur1tFWf3mxSIdBTvBWnQxMCUrmQ",
		},
	}
	checkResp := func(resp string, rpcCode int64, rpcMsg string, rowCount, rowAffect, limitRows int64, query string) *dto2.ExecSqlResp {
		res := &dto2.ExecSqlResp{}
		err := json.Unmarshal([]byte(resp), res)
		require.NoError(t, err)
		require.Equal(t, rpcMsg, res.Data.Result.Message)
		require.Equal(t, rpcCode, res.Data.Result.Code)
		require.Equal(t, rowCount, res.Data.Result.RowCount)
		require.Equal(t, rowAffect, res.Data.Result.RowAffect)
		require.Equal(t, int(limitRows), int(res.Data.Result.Limit))
		//require.Contains(t, res.Data.Result.Query, query)
		return res
	}
	resp, code := testEndpointHttpApi(t, gatewayHttpAddr, req)
	require.Equal(t, 400, code)
	checkResp(resp, errs.ErrCodeBadRequest, "Parse endpoint sql failed, Param 'id' is required", 0, 0, 0, "")

	req.Args = []dto2.EndpointArg{
		{
			Name:  "id",
			Value: "1",
			Type:  "bigint",
		},
	}
	defaultLimitRows := int64(1000)
	resp, code = testEndpointHttpApi(t, gatewayHttpAddr, req)
	fmt.Printf("\n---- resp: %v --1--\n\n", resp)
	require.Equal(t, 200, code)
	execSqlResp := checkResp(resp, 200, "ok", 1, 0, defaultLimitRows, "Query OK!")
	require.Equal(t, 2, len(execSqlResp.Data.Columns))
	require.Equal(t, [][]string{{"1", "a"}}, execSqlResp.Data.Rows)

	// Test endpoint not found
	req.EndpointId = 12345678901
	resp, code = testEndpointHttpApi(t, gatewayHttpAddr, req)
	fmt.Printf("\n---- resp: %v --1--\n\n", resp)
	require.Equal(t, 404, code)
	execSqlResp = checkResp(resp, errs.ErrCodeNotFound, "Failed to get endpoint", 0, 0, 0, "")

	// Test exec endpoint sql failed.
	testDeleteEndpoint(t, centralHttpAddr, testEmail, orgId, int(projectId), int(appId), int(endpointID))
	createEndpointParam.SqlTemplate = "select * from t_not_exist;"
	createEndpointParam.Settings.RowsLimit = 1
	endpointID = testCreateEndpoint(t, centralHttpAddr, testEmail, publicKey, orgId, int(projectId), int(appId), createEndpointParam)
	req.EndpointId = endpointID
	resp, code = testEndpointHttpApi(t, gatewayHttpAddr, req)
	fmt.Printf("\n---- resp: %v --1--\n\n", resp)
	require.Equal(t, 500, code)
	execSqlResp = checkResp(resp, 1146, "Table 'test.t_not_exist' doesn't exist", 0, 0, 1, "")

	// Test endpoint sql wrong syntax.
	testDeleteEndpoint(t, centralHttpAddr, testEmail, orgId, int(projectId), int(appId), int(endpointID))
	createEndpointParam.SqlTemplate = "invalid sql;"
	endpointID = testCreateEndpoint(t, centralHttpAddr, testEmail, publicKey, orgId, int(projectId), int(appId), createEndpointParam)
	req.EndpointId = endpointID
	resp, code = testEndpointHttpApi(t, gatewayHttpAddr, req)
	fmt.Printf("\n---- resp: %v --1--\n\n", resp)
	require.Equal(t, 400, code)
	execSqlResp = checkResp(resp, errs.ErrCodeBadRequest, "Parsing sql failed, Please check if sql syntax is correct, line 1 column 7 near \"invalid sql;\" ", 0, 0, 0, "")

	// Test wrong credential type
	req.Credential.Type = "key"
	resp, code = testEndpointHttpApi(t, gatewayHttpAddr, req)
	fmt.Printf("\n---- resp: %v --1--\n\n", resp)
	require.Equal(t, 400, code)
	execSqlResp = checkResp(resp, 400, "Unknown credential type: 'key', expected values: 'AUTH0', 'APIKEY'", 0, 0, 0, "")

	execSQLTemplateReq := dto2.ExecSqlTemplateReq{
		ProjectId:           projectId,
		ClusterId:           clusterId,
		Database:            databaseName,
		SqlTemplate:         "select * from t1 where id = ${id};",
		SessionId:           "",
		LimitRows:           1,
		Args:                nil,
		TimeoutMs:           20000,
		ReturnAllStmtResult: false,
	}

	// Test ExecSqlTemplate failed cause by no args.
	resp, code = execSqlTemplateHttpApi(t, gatewayHttpAddr, execSQLTemplateReq)
	fmt.Printf("\n---- resp: %v --1--\n\n", resp)
	require.Equal(t, 400, code)
	checkResp(resp, errs.ErrCodeBadRequest, "Parsing sql failed, Please check if sql syntax is correct, line 1 column 30 near \"{id};\" ", 0, 0, 0, "")

	execSQLTemplateReq.Args = []dto2.EndpointArg{
		{
			Name:  "id",
			Value: "2",
			Type:  "bigint",
		},
	}
	resp, code = execSqlTemplateHttpApi(t, gatewayHttpAddr, execSQLTemplateReq)
	fmt.Printf("\n---- resp: %v --1--\n\n", resp)
	require.Equal(t, 200, code)
	execSqlResp = checkResp(resp, 200, "ok", 1, 0, 1, "Query OK!")
	require.Equal(t, 2, len(execSqlResp.Data.Columns))
	require.Equal(t, [][]string{{"2", "b"}}, execSqlResp.Data.Rows)

	execSQLTemplateReq.SessionId = "unknown_session_id"
	resp, code = execSqlTemplateHttpApi(t, gatewayHttpAddr, execSQLTemplateReq)
	fmt.Printf("\n---- resp: %v --1--\n\n", resp)
	require.Equal(t, 400, code)
	checkResp(resp, errs.ErrCodeBadRequest, "Sorry, your session has expired, please refresh the page and try again ^_^!", 0, 0, 0, "")

	// Test ExecSqlTemplate failed cause by db error.
	execSQLTemplateReq.SqlTemplate = "select * from t_not_exist;"
	execSQLTemplateReq.SessionId = ""
	resp, code = execSqlTemplateHttpApi(t, gatewayHttpAddr, execSQLTemplateReq)
	fmt.Printf("\n---- resp: %v --1--\n\n", resp)
	require.Equal(t, 500, code)
	execSqlResp = checkResp(resp, 1146, "Table 'test.t_not_exist' doesn't exist", 0, 0, 1, "")

	execSQLReq := dto2.ExecSqlReq{
		ProjectId:           projectId,
		ClusterId:           clusterId,
		Database:            databaseName,
		Sql:                 "select * from t1 where id = ?;",
		SessionId:           "",
		LimitRows:           1,
		Args:                nil,
		TimeoutMs:           20000,
		ReturnAllStmtResult: false,
	}

	// Test ExecSql failed cause by no args.
	resp, code = execSqlHttpApi(t, gatewayHttpAddr, execSQLReq)
	fmt.Printf("\n---- resp: %v --1--\n\n", resp)
	require.Equal(t, 400, code)
	checkResp(resp, errs.ErrCodeBadRequest, "Parsing sql failed, sql parameters not match, need 1 params, but only got 0", 0, 0, 0, "")

	execSQLReq.Args = []dto2.SqlArg{
		{
			Name:  "id",
			Value: "2",
			Tp:    "bigint",
		},
	}
	resp, code = execSqlHttpApi(t, gatewayHttpAddr, execSQLReq)
	fmt.Printf("\n---- resp: %v --1--\n\n", resp)
	require.Equal(t, 200, code)
	execSqlResp = checkResp(resp, 200, "ok", 1, 0, 1, "Query OK!")
	require.Equal(t, 2, len(execSqlResp.Data.Columns))
	require.Equal(t, [][]string{{"2", "b"}}, execSqlResp.Data.Rows)

	execSQLReq.SessionId = "unknown_session_id"
	resp, code = execSqlHttpApi(t, gatewayHttpAddr, execSQLReq)
	fmt.Printf("\n---- resp: %v --1--\n\n", resp)
	require.Equal(t, 400, code)
	checkResp(resp, errs.ErrCodeBadRequest, "Sorry, your session has expired, please refresh the page and try again ^_^!", 0, 0, 0, "")

	// Test ExecSqlTemplate failed cause by db error.
	execSQLReq.Sql = "select * from t_not_exist;"
	execSQLReq.SessionId = ""
	resp, code = execSqlHttpApi(t, gatewayHttpAddr, execSQLReq)
	fmt.Printf("\n---- resp: %v --1--\n\n", resp)
	require.Equal(t, 500, code)
	execSqlResp = checkResp(resp, 1146, "Table 'test.t_not_exist' doesn't exist", 0, 0, 1, "")

	getDatabaseMetaReq := dto2.GetDatabaseMetaReq{
		ProjectId: projectId,
		ClusterId: clusterId,
		Database:  databaseName,
	}
	resp, code = getDatabaseMetaHttpApi(t, gatewayHttpAddr, getDatabaseMetaReq)
	fmt.Printf("\n---- resp: %v --1--\n\n", resp)
	require.Equal(t, 200, code)
	var getDatabaseMetaResp dto2.GetDatabaseMetaResp
	err := json.Unmarshal([]byte(resp), &getDatabaseMetaResp)
	require.NoError(t, err)
	require.Equal(t, int64(0), getDatabaseMetaResp.Code)
	require.Equal(t, "", getDatabaseMetaResp.Message)
	require.Equal(t, 2, len(getDatabaseMetaResp.Metas))
	sort.Slice(getDatabaseMetaResp.Metas, func(i, j int) bool {
		return getDatabaseMetaResp.Metas[i].Table < getDatabaseMetaResp.Metas[j].Table
	})
	require.Equal(t, &dto2.TableMeta{
		Table:    "t1",
		Columns:  []string{"id", "name"},
		Database: "test",
	}, getDatabaseMetaResp.Metas[0])
	require.Equal(t, &dto2.TableMeta{
		Table:    "t2",
		Columns:  []string{"id", "name"},
		Database: "test",
	}, getDatabaseMetaResp.Metas[1])

	getDatabaseMetaReq.Database = "db_not_exist"
	resp, code = getDatabaseMetaHttpApi(t, gatewayHttpAddr, getDatabaseMetaReq)
	fmt.Printf("\n---- resp: %v --1--\n\n", resp)
	require.Equal(t, 200, code)
	err = json.Unmarshal([]byte(resp), &getDatabaseMetaResp)
	require.NoError(t, err)
	require.Equal(t, int64(0), getDatabaseMetaResp.Code)
	require.Equal(t, "", getDatabaseMetaResp.Message)
	require.Equal(t, 0, len(getDatabaseMetaResp.Metas))

	createSessionReq := dto2.CreateSessionReq{
		ProjectId:  projectId,
		ClusterId:  clusterId,
		Database:   databaseName,
		ExpireSec:  360,
		Credential: dto2.Credential{},
	}
	resp, code = postHttpApi(t, gatewayHttpAddr+"/api/v1beta/session/create", createSessionReq)
	require.Equal(t, 200, code)
	var createSessionResp dto2.CreateSessionResp
	err = json.Unmarshal([]byte(resp), &createSessionResp)
	require.NoError(t, err)
	require.Equal(t, int64(0), createSessionResp.Code)
	require.Equal(t, "", createSessionResp.Message)
	require.NotEqual(t, "", createSessionResp.SessionId)

	execSQLReq.Sql = "begin; insert into t1 values (10,'abc')"
	execSQLReq.SessionId = createSessionResp.SessionId
	resp, code = execSqlHttpApi(t, gatewayHttpAddr, execSQLReq)
	require.Equal(t, 200, code)
	execSqlResp = checkResp(resp, 200, "ok", 0, 1, 1, "Query OK")

	execSQLReq.Sql = "select * from t1 where id = 10"
	execSQLReq.LimitRows = 30
	execSQLReq.SessionId = createSessionResp.SessionId
	resp, code = execSqlHttpApi(t, gatewayHttpAddr, execSQLReq)
	require.Equal(t, 200, code)
	execSqlResp = checkResp(resp, 200, "ok", 1, 0, 30, "Query OK")
	require.Equal(t, 2, len(execSqlResp.Data.Columns))
	require.Equal(t, [][]string{{"10", "abc"}}, execSqlResp.Data.Rows)

	execSQLReq.SessionId = ""
	resp, code = execSqlHttpApi(t, gatewayHttpAddr, execSQLReq)
	require.Equal(t, 200, code)
	execSqlResp = checkResp(resp, 200, "ok", 0, 0, 30, "Query OK")
	require.Equal(t, [][]string{}, execSqlResp.Data.Rows)

	closeSessionReq := dto2.CloseSessionReq{
		SessionId: createSessionResp.SessionId,
	}
	resp, code = postHttpApi(t, gatewayHttpAddr+"/api/v1beta/session/close", closeSessionReq)
	fmt.Printf("\n---- resp: %v --1--\n\n", resp)
	require.Equal(t, 200, code)
	require.Equal(t, `{"code":0,"message":""}`, resp)

	// Test ParseOneStmt api
	parseOneStmtReq := dto2.ParseOneStmtReq{Sql: "select * from t1;"}
	resp, code = postHttpApi(t, gatewayHttpAddr+"/api/v1beta/parse/stmt", parseOneStmtReq)
	fmt.Printf("\n---- resp: %v --1--\n\n", resp)
	require.Equal(t, 200, code)
	var parseOneStmtResp dto2.ParseOneStmtResp
	err = json.Unmarshal([]byte(resp), &parseOneStmtResp)
	require.NoError(t, err)
	require.Equal(t, int64(0), parseOneStmtResp.Code)
	require.Equal(t, "", parseOneStmtResp.Message)
	require.Equal(t, dataapi.StmtType_Select, parseOneStmtResp.StmtType)

	parseOneStmtReq = dto2.ParseOneStmtReq{Sql: "insert into t1 values (1,'a')"}
	resp, code = postHttpApi(t, gatewayHttpAddr+"/api/v1beta/parse/stmt", parseOneStmtReq)
	require.Equal(t, 200, code)
	err = json.Unmarshal([]byte(resp), &parseOneStmtResp)
	require.NoError(t, err)
	require.Equal(t, int64(0), parseOneStmtResp.Code)
	require.Equal(t, "", parseOneStmtResp.Message)
	require.Equal(t, dataapi.StmtType_Insert, parseOneStmtResp.StmtType)
}

func TestGatewayService0(t *testing.T) {
	setupExecutor("/Users/cs/.infra-api/dataapi_executor.yaml")
	setupGateway("/Users/cs/.infra-api/dataapi_gateway.yaml")
	// wait grpc server started.
	time.Sleep(500 * time.Millisecond)
	conn, err := grpc.Dial("0.0.0.0:"+config.Global().ServerConf.GrpcPort, grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		t.Fatal(err)
	}

	gatewayClient := dataapi.NewDataAPIGatewayServiceClient(conn)

	ctx := context.Background()
	projectId := uint64(43563)
	clusterId := uint64(270048)
	databaseName := "test"

	execSQLTemplateReq := &dataapi.ExecSqlTemplateReq{
		ProjectId:   projectId,
		ClusterId:   clusterId,
		Database:    databaseName,
		SqlTemplate: "select * from t1 limit ${line}",
		Args: []*dataapi.EndpointArg{
			{
				Name:  "line",
				Value: "1",
				Type:  "bigint",
			},
		},
		SessionId: "",
		LimitRows: 10,
		TimeoutMs: 1000000,
	}
	resp, err := gatewayClient.ExecSqlTemplate(ctx, execSQLTemplateReq)
	require.NoError(t, err)
	require.NotNil(t, resp.BaseResp)
	require.Equal(t, "", resp.BaseResp.ErrMsg)
	require.Equal(t, 1, len(resp.StmtResults))
	require.Equal(t, "", resp.StmtResults[0].ErrMessage)
	require.Equal(t, int64(0), resp.StmtResults[0].ErrCode)
	require.Equal(t, int64(1), resp.StmtResults[0].RowNum)
	require.Equal(t, []string{"1", "a"}, resp.StmtResults[0].Rows[0].Items)
	require.Equal(t, dataapi.StmtType_Select, resp.StmtResults[0].StmtType)
}

func TestGatewayService1(t *testing.T) {
	setupCentral("/Users/cs/.infra-api/dataapi1.yaml")
	centralHttpAddr := fmt.Sprintf("http://0.0.0.0:%v", config.Global().ServerConf.HttpPort)
	setupExecutor("/Users/cs/.infra-api/dataapi_executor.yaml")
	setupGateway("/Users/cs/.infra-api/dataapi_gateway.yaml")
	//err := config.InitConfig("/Users/cs/.infra-api/dataapi_gateway.yaml")
	//require.NoError(t, err)
	// wait grpc server started.
	time.Sleep(500 * time.Millisecond)
	conn, err := grpc.Dial("0.0.0.0:"+config.Global().ServerConf.GrpcPort, grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		t.Fatal(err)
	}

	gatewayClient := dataapi.NewDataAPIGatewayServiceClient(conn)

	ctx := context.Background()
	projectId := uint64(43563)
	clusterId := uint64(270048)
	clusterIdStr := "270048"
	appId := uint64(5)
	databaseName := "test"
	testEmail := "chenshuang@pingcap.com"
	orgId := 30061
	publicKey := "DsioGVSB"

	createEndpointParam := &entity.CreateEndpointParam{
		Name:         "cs-endpoint-test00",
		Endpoint:     "/cstest00",
		Method:       "get",
		ClusterIdStr: clusterIdStr,
		Database:     databaseName,
		SqlTemplate:  "select * from t1 where id = ${id};",
		Args: []entity.EndpointArgs{
			{
				Name:     "id",
				Type:     "NUMBER",
				Required: 1,
			},
		},
		Settings: entity.EndpointSettings{
			TimeOut:   10000,
			RowsLimit: 1000,
		},
	}
	endpointID := testCreateEndpoint(t, centralHttpAddr, testEmail, publicKey, orgId, int(projectId), int(appId), createEndpointParam)
	fmt.Printf("create endpoint %v ------\n\n", endpointID)
	defer func() {
		testDeleteEndpoint(t, centralHttpAddr, testEmail, orgId, int(projectId), int(appId), int(endpointID))
	}()

	testEndpointReq := &dataapi.TestEndpointReq{
		ProjectId:  projectId,
		AppId:      appId,
		EndpointId: endpointID,
		Args:       nil,
		Credential: &dataapi.Credential{
			Type: dataapi.CrtType_AUTH0,
			Key:  "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCIsImtpZCI6Im1oSjk3OVJEQlFhOWU1U1I4ZUxwRCJ9.eyJodHRwczovL3RpZGJjbG91ZC5jb20vbG9naW5fcmVjZWl2ZWRfYXQiOjE2Nzc0NjM5NjUyMTksImVtYWlsIjoiaGFueGlhb21lbmcxMTFAZ21haWwuY29tIiwiZW1haWxfdmVyaWZpZWQiOnRydWUsImlzcyI6Imh0dHBzOi8vdGlkYi1kZXYudXMuYXV0aDAuY29tLyIsImF1ZCI6Iktpd1dhakxKM3B0bzhTc1o3RjM1TmFxTmh3bm9SaDFjIiwiaWF0IjoxNjc3NDYzOTY1LCJleHAiOjE2Nzc0OTk5NjUsInN1YiI6Imdvb2dsZS1vYXV0aDJ8MTA3MDYyNjE0NDEzNTMyMjY4MTE4IiwiYXRfaGFzaCI6ImxpdFpQT0Q0U1RzSDNfam1ZTDFPVnciLCJzaWQiOiJ4ZXNHT1ptOE9RTkVidkxYRXYtNm5VOTNvQmFGeW1QNyIsIm5vbmNlIjoiRDBGT1lzTHJ0aFNQYTUtQmRUbHVBRGpaYmRRR2NWMGcifQ.yTIVyiqOkbr7rvKUVeJ9-dckG3sWBgrblAzHb5Dz9ajGZYkqtciWaA-OPLJuYYS9PAuz1gyhrRCEuScHh0NO_T6XqRgoiRiGPf86rcj4moCCu2v4_pUXdp-fyaVL57wgZ6gGfmHdjvK7LVEZSUXMqsZivoKbTdRp2kOOLxiV5S1wxNZHo1wQCwIsiRLNGPpRIWtv1TWxCbz0Yk2FdjYQ1CUd8x1VIeZ8IQ6cmRX5yizHtwpBm-zkH-vIH1BVFBDP7D_xtEq7BtWbCY1seFs6xnOjbEW17-U_MGkvFWAZclzMaU72_oULGgV6R31Ur1tFWf3mxSIdBTvBWnQxMCUrmQ",
		},
	}

	resp, err := gatewayClient.TestEndpoint(ctx, testEndpointReq)
	require.NoError(t, err)
	require.NotNil(t, resp.BaseResp)
	require.Equal(t, int64(errs.ErrCodeBadRequest), resp.BaseResp.ErrCode)
	require.Equal(t, "Parse endpoint sql failed, Param 'id' is required", resp.BaseResp.ErrMsg)

	testEndpointReq.Args = []*dataapi.EndpointArg{
		{
			Name:  "id",
			Value: "1",
			Type:  "bigint",
		},
	}
	resp, err = gatewayClient.TestEndpoint(ctx, testEndpointReq)
	require.NoError(t, err)
	require.Equal(t, 1, len(resp.StmtResults))
	require.Equal(t, int64(1), resp.StmtResults[0].RowNum)
	require.Equal(t, []string{"1", "a"}, resp.StmtResults[0].Rows[0].Items)
	require.Equal(t, dataapi.StmtType_Select, resp.StmtResults[0].StmtType)

	// Test endpoint not found
	testEndpointReq.EndpointId = 12345678901
	resp, err = gatewayClient.TestEndpoint(ctx, testEndpointReq)
	require.NoError(t, err)
	require.NotNil(t, resp.BaseResp)
	require.Equal(t, "Failed to get endpoint", resp.BaseResp.ErrMsg)
	require.Equal(t, errs.ErrCodeNotFound, resp.BaseResp.ErrCode)

	execSQLTemplateReq := &dataapi.ExecSqlTemplateReq{
		ProjectId:   projectId,
		ClusterId:   clusterId,
		Database:    databaseName,
		SqlTemplate: "select * from t1 where id = ${id};",
		SessionId:   "",
		LimitRows:   10,
	}

	resp, err = gatewayClient.ExecSqlTemplate(ctx, execSQLTemplateReq)
	require.NoError(t, err)
	require.NotNil(t, resp.BaseResp)
	require.Equal(t, errs.ErrCodeBadRequest, resp.BaseResp.ErrCode)
	require.Equal(t, "Parsing sql failed, Please check if sql syntax is correct, line 1 column 30 near \"{id};\" ", resp.BaseResp.ErrMsg)
	require.Equal(t, 0, len(resp.StmtResults))

	execSQLTemplateReq.Args = []*dataapi.EndpointArg{
		{
			Name:  "id",
			Value: "2",
			Type:  "bigint",
		},
	}
	resp, err = gatewayClient.ExecSqlTemplate(ctx, execSQLTemplateReq)
	require.NoError(t, err)
	require.Equal(t, 1, len(resp.StmtResults))
	require.Equal(t, int64(1), resp.StmtResults[0].RowNum)
	require.Equal(t, []string{"2", "b"}, resp.StmtResults[0].Rows[0].Items)
	require.Equal(t, dataapi.StmtType_Select, resp.StmtResults[0].StmtType)

	execSQLTemplateReq.SessionId = "unknown_session_id"
	resp, err = gatewayClient.ExecSqlTemplate(ctx, execSQLTemplateReq)
	require.NoError(t, err)
	require.Equal(t, int64(49900001), resp.BaseResp.ErrCode)
	require.Equal(t, "Sorry, your session has expired, please refresh the page and try again ^_^!", resp.BaseResp.ErrMsg)

	seResp, err := gatewayClient.CreateSession(ctx, &dataapi.CreateSessionReq{
		ProjectId:  projectId,
		ClusterId:  clusterId,
		Database:   databaseName,
		ExpireSec:  0,
		Credential: nil,
	})
	require.NoError(t, err)
	require.NotEqual(t, "", seResp.SessionId)

	execSQLTemplateReq.SessionId = seResp.SessionId
	resp, err = gatewayClient.ExecSqlTemplate(ctx, execSQLTemplateReq)
	require.NoError(t, err)
	require.Equal(t, 1, len(resp.StmtResults))
	require.Equal(t, int64(1), resp.StmtResults[0].RowNum)
	require.Equal(t, []string{"2", "b"}, resp.StmtResults[0].Rows[0].Items)
	require.Equal(t, dataapi.StmtType_Select, resp.StmtResults[0].StmtType)

	resp2, err := gatewayClient.CloseSession(ctx, &dataapi.CloseSessionReq{SessionId: "unknown_session_id"})
	require.NoError(t, err)
	require.Equal(t, errs.ErrCodeBadRequest, resp2.BaseResp.ErrCode)
	require.Equal(t, "Close session failed, no sessionId found in cache: unknown_session_id", resp2.BaseResp.ErrMsg)

	endpointReq := &dataapi.BasicEndpointReqV2{
		ProjectId:  projectId,
		ClusterId:  clusterId,
		Database:   databaseName,
		Sql:        "select * from t1 where id = 1;",
		Args:       nil,
		SessionId:  "",
		LimitRows:  10,
		Credential: nil,
		TimeoutMs:  1000000,
	}
	res, err := gatewayClient.ExecSqlDirectly(ctx, endpointReq)
	require.NoError(t, err)
	require.Equal(t, 1, len(res.StmtResults))
	require.Equal(t, int64(1), res.StmtResults[0].RowNum)
	require.Equal(t, []string{"1", "a"}, res.StmtResults[0].Rows[0].Items)
	require.Equal(t, dataapi.StmtType_Select, res.StmtResults[0].StmtType)

	res, err = gatewayClient.ExecSqlWithSession(ctx, endpointReq)
	require.NoError(t, err)
	require.Equal(t, errs.ErrCodeBadRequest, res.BaseResp.ErrCode)
	require.Equal(t, "Session id is required", res.BaseResp.ErrMsg)

	endpointReq.SessionId = "1"
	res, err = gatewayClient.ExecSqlWithSession(ctx, endpointReq)
	require.NoError(t, err)
	require.Equal(t, int64(49900001), res.BaseResp.ErrCode)
	require.Equal(t, "Sorry, your session has expired, please refresh the page and try again ^_^!", res.BaseResp.ErrMsg)

	endpointReq.SessionId = seResp.SessionId
	res, err = gatewayClient.ExecSqlWithSession(ctx, endpointReq)
	require.NoError(t, err)
	require.Equal(t, 1, len(res.StmtResults))
	require.Equal(t, int64(1), res.StmtResults[0].RowNum)
	require.Equal(t, []string{"1", "a"}, res.StmtResults[0].Rows[0].Items)
	require.Equal(t, dataapi.StmtType_Select, res.StmtResults[0].StmtType)

	endpointReq.Sql = "bad sql;"
	res, err = gatewayClient.ExecSqlDirectly(ctx, endpointReq)
	require.NoError(t, err)
	require.NotNil(t, res.BaseResp)
	require.Equal(t, errs.ErrCodeBadRequest, res.BaseResp.ErrCode)
	require.Equal(t, "Parsing sql failed, Please check if sql syntax is correct, line 1 column 3 near \"bad sql;\" ", res.BaseResp.ErrMsg)
	require.Equal(t, 0, len(res.StmtResults))

	endpointReq.Sql = "use test;bad sql;"
	endpointReq.ReturnAllStmtResult = true
	res, err = gatewayClient.ExecSqlDirectly(ctx, endpointReq)
	require.NoError(t, err)
	require.NotNil(t, res.BaseResp)
	require.Equal(t, errs.ErrCodeBadRequest, res.BaseResp.ErrCode)
	require.Equal(t, "Parsing sql failed, Please check if sql syntax is correct, line 1 column 12 near \"bad sql;\" ", res.BaseResp.ErrMsg)
	require.Equal(t, 0, len(res.StmtResults))

	endpointReq.SessionId = seResp.SessionId
	res, err = gatewayClient.ExecSqlWithSession(ctx, endpointReq)
	require.NoError(t, err)
	require.NotNil(t, res.BaseResp)
	require.Equal(t, errs.ErrCodeBadRequest, res.BaseResp.ErrCode)
	require.Equal(t, "Parsing sql failed, Please check if sql syntax is correct, line 1 column 12 near \"bad sql;\" ", res.BaseResp.ErrMsg)
	require.Equal(t, 0, len(res.StmtResults))
	require.Equal(t, 0, len(res.StmtResults))

	seResp, err = gatewayClient.CreateSession(ctx, &dataapi.CreateSessionReq{
		ProjectId:  projectId,
		ClusterId:  clusterId,
		ExpireSec:  0,
		Credential: nil,
	})
	require.NoError(t, err)
	require.NotEqual(t, "", seResp.SessionId)

	// change db.
	endpointReq.ReturnAllStmtResult = false
	endpointReq.SessionId = seResp.SessionId
	endpointReq.Database = "test"
	endpointReq.Sql = "select * from t1 where id = 1;"
	res, err = gatewayClient.ExecSqlWithSession(ctx, endpointReq)
	require.NoError(t, err)
	require.Equal(t, int64(0), res.BaseResp.ErrCode)
	require.Equal(t, "", res.BaseResp.ErrMsg)
	require.Equal(t, 1, len(res.StmtResults))
	require.Equal(t, "", res.StmtResults[0].ErrMessage)
	require.Equal(t, int64(0), res.StmtResults[0].ErrCode)
	require.Equal(t, []string{"1", "a"}, res.StmtResults[0].Rows[0].Items)
	require.Equal(t, dataapi.StmtType_Select, res.StmtResults[0].StmtType)

	// change db.
	endpointReq.Database = "sample_data"
	endpointReq.Sql = "select 1 from github_events limit 1;"
	res, err = gatewayClient.ExecSqlWithSession(ctx, endpointReq)
	require.NoError(t, err)
	require.Equal(t, 1, len(res.StmtResults))
	require.Equal(t, "", res.StmtResults[0].ErrMessage)
	require.Equal(t, int64(0), res.StmtResults[0].ErrCode)
	require.Equal(t, int64(1), res.StmtResults[0].RowNum)
	require.Equal(t, []string{"1"}, res.StmtResults[0].Rows[0].Items)
	require.Equal(t, dataapi.StmtType_Select, res.StmtResults[0].StmtType)
}

func TestGatewayService2(t *testing.T) {
	setupExecutor("/Users/cs/.infra-api/dataapi_executor.yaml")
	setupGateway("/Users/cs/.infra-api/dataapi_gateway.yaml")
	err := config.InitConfig("/Users/cs/.infra-api/dataapi_gateway.yaml")
	require.NoError(t, err)
	conn, err := grpc.Dial("0.0.0.0:"+config.Global().ServerConf.GrpcPort, grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		t.Fatal(err)
	}

	gatewayClient := dataapi.NewDataAPIGatewayServiceClient(conn)

	ctx := context.Background()
	projectId := uint64(43563)
	clusterId := uint64(270048)
	//endpointId := uint64(90002)
	//appId := uint64(5)
	databaseName := "test"

	checkStmtResult := func(stmt *dataapi.ExecStmtResult, code int64, msg string, affect, rowNum int64, dbs []string, tp dataapi.StmtType) {
		require.Equal(t, code, stmt.ErrCode)
		require.Equal(t, msg, stmt.ErrMessage)
		require.Equal(t, affect, stmt.Affect)
		require.Equal(t, rowNum, stmt.RowNum)
		require.Equal(t, int(rowNum), len(stmt.Rows))
		require.Equal(t, dbs, stmt.Databases)
		require.Equal(t, tp, stmt.StmtType)
	}

	endpointReq := &dataapi.BasicEndpointReqV2{
		ProjectId:  projectId,
		ClusterId:  clusterId,
		Database:   databaseName,
		Args:       nil,
		SessionId:  "",
		LimitRows:  10,
		Credential: nil,
	}
	endpointReq.Sql = "delete from t1;" +
		"insert into t1 values (1,'a');" +
		"insert into t1 values (2, 'b');" +
		"select * from t1 order by id;"
	res, err := gatewayClient.ExecSqlDirectly(ctx, endpointReq)
	require.NoError(t, err)
	require.Equal(t, 1, len(res.StmtResults))
	checkStmtResult(res.StmtResults[0], 0, "", 0, 2, []string{""}, dataapi.StmtType_Select)
	require.Equal(t, []string{"1", "a"}, res.StmtResults[0].Rows[0].Items)
	require.Equal(t, []string{"2", "b"}, res.StmtResults[0].Rows[1].Items)

	// Test run multi-statement
	endpointReq.Sql = "begin;" +
		"delete from t1;" +
		"insert into t1 values (1,'aa');" +
		"insert into t1 values (2, 'bb');" +
		"select * from t1 order by id;" // doesn't commit
	res, err = gatewayClient.ExecSqlDirectly(ctx, endpointReq)
	require.NoError(t, err)
	require.Equal(t, 1, len(res.StmtResults))
	checkStmtResult(res.StmtResults[0], 0, "", 0, 2, []string{""}, dataapi.StmtType_Select)
	require.Equal(t, []string{"1", "aa"}, res.StmtResults[0].Rows[0].Items)
	require.Equal(t, []string{"2", "bb"}, res.StmtResults[0].Rows[1].Items)

	endpointReq.ReturnAllStmtResult = true
	endpointReq.Sql = "begin;" +
		"delete from t1;" +
		"insert into t1 values  (1,'aa');" +
		"replace into t1 values (1, 'bb');" +
		"select * from t1 order by id;" +
		"rollback;" +
		"select * from t1 order by id;"
	res, err = gatewayClient.ExecSqlDirectly(ctx, endpointReq)
	require.NoError(t, err)
	require.Equal(t, 7, len(res.StmtResults))
	checkStmtResult(res.StmtResults[0], 0, "", 0, 0, nil, dataapi.StmtType_Other)
	checkStmtResult(res.StmtResults[1], 0, "", 2, 0, nil, dataapi.StmtType_Delete)
	checkStmtResult(res.StmtResults[2], 0, "", 1, 0, nil, dataapi.StmtType_Insert)
	checkStmtResult(res.StmtResults[3], 0, "", 2, 0, nil, dataapi.StmtType_Insert)
	checkStmtResult(res.StmtResults[4], 0, "", 0, 1, []string{""}, dataapi.StmtType_Select)
	require.Equal(t, []string{"1", "bb"}, res.StmtResults[4].Rows[0].Items)
	checkStmtResult(res.StmtResults[5], 0, "", 0, 0, nil, dataapi.StmtType_Other)
	checkStmtResult(res.StmtResults[6], 0, "", 0, 2, []string{""}, dataapi.StmtType_Select)
	require.Equal(t, []string{"1", "a"}, res.StmtResults[6].Rows[0].Items)
	require.Equal(t, []string{"2", "b"}, res.StmtResults[6].Rows[1].Items)

	endpointReq.Sql = "select * from t1 order by id;"
	res, err = gatewayClient.ExecSqlDirectly(ctx, endpointReq)
	require.NoError(t, err)
	require.Equal(t, 1, len(res.StmtResults))
	checkStmtResult(res.StmtResults[0], 0, "", 0, 2, []string{""}, dataapi.StmtType_Select)
	require.Equal(t, []string{"1", "a"}, res.StmtResults[0].Rows[0].Items)
	require.Equal(t, []string{"2", "b"}, res.StmtResults[0].Rows[1].Items)

	// Test exec error
	endpointReq.ReturnAllStmtResult = true
	endpointReq.Sql = "begin;" +
		"delete from t1;" +
		"insert into t1 values  (1,'aa');" +
		"replace into t1 values (1, 'bb');" +
		"select * from t_not_exist order by id;" + // exec error
		"rollback;" +
		"select * from t1 order by id;"
	res, err = gatewayClient.ExecSqlDirectly(ctx, endpointReq)
	require.NoError(t, err)
	require.Equal(t, int64(0), res.BaseResp.ErrCode)
	require.Equal(t, "", res.BaseResp.ErrMsg)
	require.Equal(t, 5, len(res.StmtResults))
	checkStmtResult(res.StmtResults[0], 0, "", 0, 0, nil, dataapi.StmtType_Other)
	checkStmtResult(res.StmtResults[1], 0, "", 2, 0, nil, dataapi.StmtType_Delete)
	checkStmtResult(res.StmtResults[2], 0, "", 1, 0, nil, dataapi.StmtType_Insert)
	checkStmtResult(res.StmtResults[3], 0, "", 2, 0, nil, dataapi.StmtType_Insert)
	checkStmtResult(res.StmtResults[4], 1146, "Table 'test.t_not_exist' doesn't exist", 0, 0, []string{""}, dataapi.StmtType_Select)

	endpointReq.ReturnAllStmtResult = false
	res, err = gatewayClient.ExecSqlDirectly(ctx, endpointReq)
	require.NoError(t, err)
	require.Equal(t, "", res.BaseResp.ErrMsg)
	require.Equal(t, int64(0), res.BaseResp.ErrCode)
	require.Equal(t, 1, len(res.StmtResults))
	checkStmtResult(res.StmtResults[0], 1146, "Table 'test.t_not_exist' doesn't exist", 0, 0, []string{""}, dataapi.StmtType_Select)
}

func TestGatewayService3(t *testing.T) {
	setupCentral("/Users/cs/.infra-api/dataapi1.yaml")
	centralHttpAddr := fmt.Sprintf("http://0.0.0.0:%v", config.Global().ServerConf.HttpPort)
	setupExecutor("/Users/cs/.infra-api/dataapi_executor.yaml")
	setupGateway("/Users/cs/.infra-api/dataapi_gateway.yaml")
	gatewayHttpAddr := fmt.Sprintf("http://0.0.0.0:%v", config.Global().ServerConf.HttpPort)
	fmt.Printf("%v    %v --------\n\n", centralHttpAddr, gatewayHttpAddr)

	time.Sleep(time.Second)

	projectId := uint64(43563)
	clusterId := "270048"
	databaseName := "test"
	testEmail := "chenshuang@pingcap.com"
	orgId := 30061
	appId := 5
	publicKey := "DsioGVSB"

	createEndpointParam := &entity.CreateEndpointParam{
		Name:         "cs-endpoint-test00",
		Endpoint:     "/cstest00",
		Method:       "get",
		ClusterIdStr: clusterId,
		Database:     databaseName,
		SqlTemplate:  "select * from t1;",
		Settings: entity.EndpointSettings{
			TimeOut:   3600,
			RowsLimit: 1000,
		},
	}
	endpointID := testCreateEndpoint(t, centralHttpAddr, testEmail, publicKey, orgId, int(projectId), appId, createEndpointParam)
	fmt.Printf("create endpoint %v ------\n\n", endpointID)
	defer func() {
		testDeletePublishEndpoint(t, centralHttpAddr, testEmail, publicKey, orgId, int(projectId), appId, int(endpointID))
		testDeleteEndpoint(t, centralHttpAddr, testEmail, orgId, int(projectId), appId, int(endpointID))
	}()

	externalAppID := "app-uzytMOJP"
	resp, _ := testAdvanceEndpointByPost(t, gatewayHttpAddr, testEmail, "draft", "DsioGVSB", true, externalAppID, "/cstest00", nil, "get")
	checkAdvanceResp := func(resp string, rpcCode int, rpcMsg string, sqlRespCount int, sql string) *dto2.AdvancedRes {
		advanceRes := &dto2.AdvancedRes{}
		err := json.Unmarshal([]byte(resp), advanceRes)
		require.NoError(t, err)
		require.Equal(t, rpcMsg, advanceRes.Data.Result.Message)
		require.Equal(t, rpcCode, advanceRes.Data.Result.Code)
		//require.Equal(t, sqlRespCount, len(advanceRes.Data))
		//require.Equal(t, sql, advanceRes.Data[0].SQL)
		return advanceRes
	}
	advanceRes := checkAdvanceResp(resp, 200, "ok", 1, "select * from t1;")
	//require.Equal(t, int64(0), advanceRes.Data.Result.Code)
	//require.Equal(t, "", advanceRes.Data.Result.Message)
	require.Equal(t, [][]string{{"1", "a"}, {"2", "b"}}, advanceRes.Data.Rows)
	require.Equal(t, int64(2), advanceRes.Data.Result.RowCount)
	// Test publish endpoint
	testPublishEndpoint(t, centralHttpAddr, testEmail, publicKey, orgId, int(projectId), appId, int(endpointID))
	resp, _ = testAdvanceEndpointByPost(t, gatewayHttpAddr, testEmail, "", "DsioGVSB", true, externalAppID, "/cstest00", nil, "get")
	advanceRes = checkAdvanceResp(resp, 200, "ok", 1, "select * from t1;")
	//require.Equal(t, int64(0), advanceRes.Data.Result.Code)
	//require.Equal(t, "", advanceRes.Data.Result.Message)
	require.Equal(t, [][]string{{"1", "a"}, {"2", "b"}}, advanceRes.Data.Rows)
	require.Equal(t, int64(2), advanceRes.Data.Result.RowCount)

	testDeletePublishEndpoint(t, centralHttpAddr, testEmail, publicKey, orgId, int(projectId), appId, int(endpointID))
	testDeleteEndpoint(t, centralHttpAddr, testEmail, orgId, int(projectId), appId, int(endpointID))

	// Test advance a deleted endpoint  ----???
	resp, _ = testAdvanceEndpointByPost(t, gatewayHttpAddr, testEmail, "draft", "DsioGVSB", true, externalAppID, "/cstest00", nil, "get")
	advanceRes = &dto2.AdvancedRes{}
	err := json.Unmarshal([]byte(resp), advanceRes)
	require.NoError(t, err)
	require.Equal(t, "draft endpoint not found. app_id: 5, endpoint_path: /cstest00, method: get", advanceRes.Data.Result.Message)
	require.Equal(t, 404, advanceRes.Data.Result.Code)

	// Test sql with params
	createEndpointParam.SqlTemplate = "select * from t1 where id = ${id};"
	createEndpointParam.Args = []entity.EndpointArgs{
		{
			Name:     "id",
			Type:     "NUMBER",
			Required: 1,
		},
	}
	endpointID = testCreateEndpoint(t, centralHttpAddr, testEmail, publicKey, orgId, int(projectId), appId, createEndpointParam)
	resp, _ = testAdvanceEndpointByPost(t, gatewayHttpAddr, testEmail, "draft", "DsioGVSB", true, externalAppID, "/cstest00", &map[string]string{"id": "1"}, "get")
	//advanceRes = checkAdvanceResp(resp, 200, "ok", 1, "select * from t1 where id = 1;")
	advanceRes = checkAdvanceResp(resp, 200, "ok", 1, "select * from t1 where id = 1;")
	//require.Equal(t, int64(0), advanceRes.Data.Result.Code)
	//require.Equal(t, "", advanceRes.Data.Result.Message)
	require.Equal(t, int64(1), advanceRes.Data.Result.RowCount)
	require.Equal(t, [][]string{{"1", "a"}}, advanceRes.Data.Rows)

	testDeleteEndpoint(t, centralHttpAddr, testEmail, orgId, int(projectId), appId, int(endpointID))
	createEndpointParam.Method = "post"
	endpointID = testCreateEndpoint(t, centralHttpAddr, testEmail, publicKey, orgId, int(projectId), appId, createEndpointParam)
	resp, _ = testAdvanceEndpointByPost(t, gatewayHttpAddr, testEmail, "draft", "DsioGVSB", true, externalAppID, "/cstest00", &map[string]string{"id": "1"}, "post")
	//advanceRes = checkAdvanceResp(resp, 200, "ok", 1, "select * from t1 where id = 1;")
	advanceRes = checkAdvanceResp(resp, 200, "ok", 1, "select * from t1 where id = 1;")
	//require.Equal(t, int64(0), advanceRes.Data.Result.Code)
	//require.Equal(t, "", advanceRes.Data.Result.Message)
	require.Equal(t, int64(1), advanceRes.Data.Result.RowCount)
	require.Equal(t, [][]string{{"1", "a"}}, advanceRes.Data.Rows)

	testDeleteEndpoint(t, centralHttpAddr, testEmail, orgId, int(projectId), appId, int(endpointID))
	createEndpointParam.Method = "get"
	createEndpointParam.SqlTemplate = "select * from t1;"
	createEndpointParam.Args = nil
	endpointID = testCreateEndpoint(t, centralHttpAddr, testEmail, publicKey, orgId, int(projectId), appId, createEndpointParam)
	resp, _ = testAdvanceEndpointByPost(t, gatewayHttpAddr, testEmail, "draft", "DsioGVSB", true, externalAppID, "/cstest00", &map[string]string{"id": "3", "name": "c"}, "get")
	advanceRes = checkAdvanceResp(resp, 200, "ok", 1, "select * from t1;")
	//require.Equal(t, int64(0), advanceRes.Data.Result.Code)
	//require.Equal(t, "", advanceRes.Data.Result.Message)
	require.Equal(t, [][]string{{"1", "a"}, {"2", "b"}}, advanceRes.Data.Rows)
	require.Equal(t, int64(2), advanceRes.Data.Result.RowCount)

	createEndpointParam.SqlTemplate = "select * from t2 where name = ${name}"
	createEndpointParam.Args = []entity.EndpointArgs{
		{
			Name:     "name",
			Type:     "string",
			Required: 1,
		},
	}
	createEndpointParam.Method = "get"
	testDeleteEndpoint(t, centralHttpAddr, testEmail, orgId, int(projectId), appId, int(endpointID))
	endpointID = testCreateEndpoint(t, centralHttpAddr, testEmail, publicKey, orgId, int(projectId), appId, createEndpointParam)
	resp, _ = testAdvanceEndpoint(t, gatewayHttpAddr, testEmail, "draft", "DsioGVSB", true, false, externalAppID, "/cstest00", &map[string]string{"name": "a'b"}, "get")
	advanceRes = checkAdvanceResp(resp, 200, "ok", 1, "select * from t2 where name = \"a'b\"")
	//require.Equal(t, int64(0), advanceRes.Data.Result.Code)
	//require.Equal(t, "", advanceRes.Data.Result.Message)
	require.Equal(t, int64(1), advanceRes.Data.Result.RowCount)
	require.Equal(t, [][]string{{"1", "a'b"}}, advanceRes.Data.Rows)

	resp, _ = testAdvanceEndpointByPost(t, gatewayHttpAddr, testEmail, "draft", "DsioGVSB", true, externalAppID, "/cstest00", &map[string]string{"name": "a\"b"}, "get")
	advanceRes = checkAdvanceResp(resp, 200, "ok", 1, `select * from t2 where name = "a\"b"`)
	//require.Equal(t, int64(0), advanceRes.Data.Result.Code)
	//require.Equal(t, "", advanceRes.Data.Result.Message)
	require.Equal(t, int64(1), advanceRes.Data.Result.RowCount)
	require.Equal(t, [][]string{{"2", "a\"b"}}, advanceRes.Data.Rows)

	createEndpointParam.Method = "post"
	testDeleteEndpoint(t, centralHttpAddr, testEmail, orgId, int(projectId), appId, int(endpointID))
	endpointID = testCreateEndpoint(t, centralHttpAddr, testEmail, publicKey, orgId, int(projectId), appId, createEndpointParam)
	resp, _ = testAdvanceEndpoint(t, gatewayHttpAddr, testEmail, "draft", "DsioGVSB", true, false, externalAppID, "/cstest00", &map[string]string{"name": "a'b"}, "post")
	advanceRes = checkAdvanceResp(resp, 200, "ok", 1, "select * from t2 where name = \"a'b\"")
	//require.Equal(t, int64(0), advanceRes.Data.Result.Code)
	//require.Equal(t, "", advanceRes.Data.Result.Message)
	require.Equal(t, int64(1), advanceRes.Data.Result.RowCount)
	require.Equal(t, [][]string{{"1", "a'b"}}, advanceRes.Data.Rows)

	resp, _ = testAdvanceEndpointByPost(t, gatewayHttpAddr, testEmail, "draft", "DsioGVSB", true, externalAppID, "/cstest00", &map[string]string{"name": "a\"b"}, "post")
	advanceRes = checkAdvanceResp(resp, 200, "ok", 1, `select * from t2 where name = "a\"b"`)
	//require.Equal(t, int64(0), advanceRes.Data.Result.Code)
	//require.Equal(t, "", advanceRes.Data.Result.Message)
	require.Equal(t, int64(1), advanceRes.Data.Result.RowCount)
	require.Equal(t, [][]string{{"2", "a\"b"}}, advanceRes.Data.Rows)
}

func TestGatewayService4(t *testing.T) {
	setupCentral("/Users/cs/.infra-api/dataapi1.yaml")
	centralHttpAddr := fmt.Sprintf("http://0.0.0.0:%v", config.Global().ServerConf.HttpPort)
	setupExecutor("/Users/cs/.infra-api/dataapi_executor.yaml")
	setupGateway("/Users/cs/.infra-api/dataapi_gateway.yaml")
	gatewayHttpAddr := fmt.Sprintf("http://0.0.0.0:%v", config.Global().ServerConf.HttpPort)

	time.Sleep(time.Second)

	projectId := uint64(43563)
	clusterId := "270048"
	databaseName := "test"
	testEmail := "chenshuang@pingcap.com"
	orgId := 30061
	appId := 5
	externalAppID := "app-uzytMOJP"
	publicKey := "DsioGVSB"

	resp, code := testAdvanceEndpointByPost(t, gatewayHttpAddr, testEmail, "draft", "DsioGVSB", true, externalAppID, "/endpoint_not_exist", nil, "get")
	require.Equal(t, 404, code)
	require.Equal(t, `{"type":"sql_endpoint","data":{"columns":null,"rows":null,"result":{"code":404,"message":"draft endpoint not found. app: app-uzytMOJP, endpoint_path: /endpoint_not_exist, method: get","start_ms":0,"end_ms":0,"latency":"","row_count":0,"row_affect":0,"limit":0,"query":""}}}`, resp)

	//testDeleteEndpoint(t, centralHttpAddr, testEmail, orgId, int(projectId), appId, 300064)
	createEndpointParam := &entity.CreateEndpointParam{
		Name:         "cs-endpoint-test00",
		Endpoint:     "/cstest00",
		Method:       "get",
		ClusterIdStr: clusterId,
		Database:     databaseName,
		SqlTemplate:  "select * from t1 where id=${id};",
		Args: []entity.EndpointArgs{
			{
				Name:     "id",
				Type:     "number",
				Required: 1,
			},
		},
		Settings: entity.EndpointSettings{
			TimeOut:   3600,
			RowsLimit: 1000,
		},
	}
	endpointID := testCreateEndpoint(t, centralHttpAddr, testEmail, publicKey, orgId, int(projectId), appId, createEndpointParam)
	defer func() {
		testDeleteEndpoint(t, centralHttpAddr, testEmail, orgId, int(projectId), appId, int(endpointID))
	}()

	// auth failed
	//resp, code = testAdvanceEndpointByPost(t, gatewayHttpAddr, testEmail, "draft", "xx", false, externalAppID, "/cstest00", nil, "get")
	//require.Equal(t, 500, code)
	//require.Equal(t, `{"code":500,"message":"internal error! CheckApiKeyPermission failed, error:defaultPermissionHelper: rpc error: code = DeadlineExceeded desc = context deadline exceeded"}`, resp)
	helperimpl.DefaultApiKeyPermissionHelper.SetEndpoint("0.0.0.0:10000")
	//resp, code = testAdvanceEndpointByPost(t, gatewayHttpAddr, testEmail, "draft", "xx", false, externalAppID, "/cstest00", nil, "get")
	//require.Equal(t, 401, code)
	//require.Equal(t, `{"code":401,"message":"auth failed! msg:identity has no permission"}`, resp)

	resp, code = testAdvanceEndpointByPost(t, gatewayHttpAddr, testEmail, "draft", "DsioGVSB", true, externalAppID, "/cstest00", &map[string]string{"id": "abc"}, "get")
	require.Equal(t, 400, code)
	require.Equal(t, `{"type":"sql_endpoint","data":{"columns":null,"rows":null,"result":{"code":400,"message":"Parse endpoint sql failed, Parsing argument failed: id","start_ms":0,"end_ms":0,"latency":"","row_count":0,"row_affect":0,"limit":0,"query":""}}}`, resp)

	// test execute sql failed.
	createEndpointParam.SqlTemplate = "select * from t_not_exist;"
	createEndpointParam.Args = nil
	testDeleteEndpoint(t, centralHttpAddr, testEmail, orgId, int(projectId), appId, int(endpointID))
	endpointID = testCreateEndpoint(t, centralHttpAddr, testEmail, publicKey, orgId, int(projectId), appId, createEndpointParam)
	resp, code = testAdvanceEndpointByPost(t, gatewayHttpAddr, testEmail, "draft", "DsioGVSB", true, externalAppID, "/cstest00", nil, "get")
	require.Equal(t, 200, code)
	require.Contains(t, resp, `test.t_not_exist' doesn't exist`)

	// test execute invalid sql failed.
	createEndpointParam.SqlTemplate = "a invalid sql;"
	createEndpointParam.Args = nil
	testDeleteEndpoint(t, centralHttpAddr, testEmail, orgId, int(projectId), appId, int(endpointID))
	endpointID = testCreateEndpoint(t, centralHttpAddr, testEmail, publicKey, orgId, int(projectId), appId, createEndpointParam)
	resp, code = testAdvanceEndpointByPost(t, gatewayHttpAddr, testEmail, "draft", "DsioGVSB", true, externalAppID, "/cstest00", nil, "get")
	require.Equal(t, `{"type":"sql_endpoint","data":{"columns":null,"rows":null,"result":{"code":400,"message":"Parsing sql failed, Please check if sql syntax is correct, line 1 column 1 near \"a invalid sql;\" ","start_ms":0,"end_ms":0,"latency":"","row_count":0,"row_affect":0,"limit":0,"query":""}}}`, resp)
	require.Equal(t, 400, code)
}

func checkDataRes(t *testing.T, res *entity.DataRes, code int64, msg string, rows [][]string, rowCount, rowAffect int64, sql, query string) {
	require.Equal(t, msg, res.Result.Message)
	require.Equal(t, code, res.Result.Code)
	//require.Equal(t, sql, res.SQL)
	require.Equal(t, rows, res.Rows)
	require.Equal(t, rowCount, res.Result.RowCount)
	require.Equal(t, rowAffect, res.Result.RowAffect)
	//require.Regexp(t, query, res.Result.Query)
}

// multi-sqls
/*
func TestGatewayService5(t *testing.T) {
	setupCentral("/Users/cs/.infra-api/dataapi1.yaml")
	centralHttpAddr := fmt.Sprintf("http://0.0.0.0:%v", config.Global().ServerConf.HttpPort)
	setupExecutor("/Users/cs/.infra-api/dataapi_executor.yaml")
	setupGateway("/Users/cs/.infra-api/dataapi_gateway.yaml")
	gatewayHttpAddr := fmt.Sprintf("http://0.0.0.0:%v", config.Global().ServerConf.HttpPort)
	time.Sleep(time.Millisecond * 200)

	projectId := uint64(43563)
	clusterId := "270048"
	databaseName := "test"
	testEmail := "chenshuang@pingcap.com"
	orgId := 30061
	appId := 5
	publicKey := "DsioGVSB"

	createEndpointParam := &entity.CreateEndpointParam{
		Name:         "cs-endpoint-test00",
		Endpoint:     "/cstest00",
		Method:       "get",
		ClusterIdStr: clusterId,
		Database:     databaseName,
		SqlTemplate:  "select * from t1 where id = ${id1};select * from t1 where id in (${id2});",
		Args: []entity.EndpointArgs{
			{
				Name:     "id1",
				Type:     "NUMBER",
				Required: 1,
			},
			{
				Name:     "id2",
				Type:     "NUMBER",
				Required: 1,
			},
		},
		Settings: entity.EndpointSettings{
			TimeOut:   3600,
			RowsLimit: 1000,
		},
	}
	endpointID := testCreateEndpoint(t, centralHttpAddr, testEmail, publicKey, orgId, int(projectId), appId, createEndpointParam)
	defer func() {
		testDeletePublishEndpoint(t, centralHttpAddr, testEmail, publicKey, orgId, int(projectId), appId, int(endpointID))
		testDeleteEndpoint(t, centralHttpAddr, testEmail, orgId, int(projectId), appId, int(endpointID))
	}()

	externalAppID := "app-uzytMOJP"
	resp, _ := testAdvanceEndpointByPost1(t, gatewayHttpAddr, testEmail, "draft", "DsioGVSB", true, false, externalAppID, "/cstest00", &map[string]string{"id1": "1", "id2": "2"}, "get")
	checkAdvanceResp := func(resp string, rpcCode int, rpcMsg string, sqlRespCount int) *dto2.AdvancedRes {
		advanceRes := &dto2.AdvancedRes{}
		err := json.Unmarshal([]byte(resp), advanceRes)
		require.NoError(t, err)
		require.Equal(t, rpcMsg, advanceRes.Data.Result.Message)
		require.Equal(t, rpcCode, advanceRes.Data.Result.Code)
		//require.Equal(t, sqlRespCount, len(advanceRes.Data))
		return advanceRes
	}
	advanceRes := checkAdvanceResp(resp, 200, "ok", 1)
	checkDataRes(t, advanceRes.Data, 0, "", [][]string{{"2", "b"}}, 1, 0, "select * from t1 where id in (2);", "Query OK!")

	resp, _ = testAdvanceEndpointByPost1(t, gatewayHttpAddr, testEmail, "draft", "DsioGVSB", true, true, externalAppID, "/cstest00", &map[string]string{"id1": "1", "id2": "2"}, "get")
	advanceRes = checkAdvanceResp(resp, 200, "ok", 2)
	checkDataRes(t, advanceRes.Data[0], 0, "", [][]string{{"1", "a"}}, 1, 0, "select * from t1 where id = 1;", "Query OK!")
	checkDataRes(t, advanceRes.Data[1], 0, "", [][]string{{"2", "b"}}, 1, 0, "select * from t1 where id in (2);", "Query OK!")

	// Test publish endpoint
	testPublishEndpoint(t, centralHttpAddr, testEmail, publicKey, orgId, int(projectId), appId, int(endpointID))
	resp, _ = testAdvanceEndpointByPost1(t, gatewayHttpAddr, testEmail, "", "DsioGVSB", true, true, externalAppID, "/cstest00", &map[string]string{"id1": "1", "id2": "2"}, "get")
	advanceRes = checkAdvanceResp(resp, 200, "ok", 2)
	checkDataRes(t, advanceRes.Data[0], 0, "", [][]string{{"1", "a"}}, 1, 0, "select * from t1 where id = 1;", "Query OK!")
	checkDataRes(t, advanceRes.Data[1], 0, "", [][]string{{"2", "b"}}, 1, 0, "select * from t1 where id in (2);", "Query OK!")

	resp, _ = testAdvanceEndpointByPost1(t, gatewayHttpAddr, testEmail, "", "DsioGVSB", true, true, externalAppID, "/cstest00", &map[string]string{"id1": "2", "id2": "4"}, "get")
	advanceRes = checkAdvanceResp(resp, 200, "ok", 2)
	checkDataRes(t, advanceRes.Data[0], 0, "", [][]string{{"2", "b"}}, 1, 0, "select * from t1 where id = 2;", "Query OK!")
	checkDataRes(t, advanceRes.Data[1], 0, "", [][]string{}, 0, 0, "select * from t1 where id in (4);", "Query OK!")

	testDeletePublishEndpoint(t, centralHttpAddr, testEmail, publicKey, orgId, int(projectId), appId, int(endpointID))
	testDeleteEndpoint(t, centralHttpAddr, testEmail, orgId, int(projectId), appId, int(endpointID))

	// ------------------------------------
	createEndpointParam.SqlTemplate = `use test;
  select * from t1 where id = ${id} or name = ${name};
    select * from t1 where id > ${id};
    select * from t1 where name != ${name};
		`
	createEndpointParam.Args = []entity.EndpointArgs{
		{
			Name:     "name",
			Type:     "string",
			Required: 1,
		},
		{
			Name:     "id",
			Type:     "NUMBER",
			Required: 1,
		},
	}
	endpointID = testCreateEndpoint(t, centralHttpAddr, testEmail, publicKey, orgId, int(projectId), appId, createEndpointParam)
	resp, _ = testAdvanceEndpointByPost1(t, gatewayHttpAddr, testEmail, "draft", "DsioGVSB", true, false, externalAppID, "/cstest00", &map[string]string{"id": "1", "name": "b"}, "get")
	advanceRes = checkAdvanceResp(resp, 200, "ok", 1)
	checkDataRes(t, advanceRes.Data[0], 0, "", [][]string{{"1", "a"}}, 1, 0, "    select * from t1 where name != \"b\";", "Query OK!")

	resp, _ = testAdvanceEndpointByPost1(t, gatewayHttpAddr, testEmail, "draft", "DsioGVSB", true, true, externalAppID, "/cstest00", &map[string]string{"id": "1", "name": "b"}, "get")
	advanceRes = checkAdvanceResp(resp, 200, "ok", 4)
	checkDataRes(t, advanceRes.Data[0], 0, "", [][]string{}, 0, 0, "use test;", "Database changed!")
	checkDataRes(t, advanceRes.Data[1], 0, "", [][]string{{"1", "a"}, {"2", "b"}}, 2, 0, "  select * from t1 where id = 1 or name = \"b\";", "Query OK!")
	checkDataRes(t, advanceRes.Data[2], 0, "", [][]string{{"2", "b"}}, 1, 0, "    select * from t1 where id > 1;", "Query OK!")
	checkDataRes(t, advanceRes.Data[3], 0, "", [][]string{{"1", "a"}}, 1, 0, "    select * from t1 where name != \"b\";", "Query OK!")

	testDeletePublishEndpoint(t, centralHttpAddr, testEmail, publicKey, orgId, int(projectId), appId, int(endpointID))
	testDeleteEndpoint(t, centralHttpAddr, testEmail, orgId, int(projectId), appId, int(endpointID))

	// ------------------------------------
	createEndpointParam.SqlTemplate = `drop database if exists test2;
create database test2;
use test2;
create table t1 (id int key, name varchar(10));
insert into t1 values (${id0}, ${name0}),(${id1}, ${name1});
select * from test2.t1 where id in ( ${id0}, ${id1} ) or name = ${name1};
update t1 set id=id+10 where name in ( ${name0}, ${name1});
select * from test2.t1 where name in ( ${name0}, ${name1} );
delete from t1 where name = ${name1};
select * from test2.t1 where name in ( ${name0}, ${name1} );
`
	createEndpointParam.Args = []entity.EndpointArgs{
		{
			Name:     "id0",
			Type:     "NUMBER",
			Required: 1,
		},
		{
			Name:     "id1",
			Type:     "NUMBER",
			Required: 1,
		},
		{
			Name:     "name0",
			Type:     "string",
			Required: 1,
		},
		{
			Name:     "name1",
			Type:     "string",
			Required: 1,
		},
	}
	endpointID = testCreateEndpoint(t, centralHttpAddr, testEmail, publicKey, orgId, int(projectId), appId, createEndpointParam)
	resp, _ = testAdvanceEndpointByPost1(t, gatewayHttpAddr, testEmail, "draft", "DsioGVSB", true, false, externalAppID, "/cstest00", &map[string]string{"id0": "0", "id1": "1", "name0": "aa0", "name1": "bb1"}, "get")
	advanceRes = checkAdvanceResp(resp, 200, "ok", 1)
	checkDataRes(t, advanceRes.Data[0], 0, "", [][]string{{"10", "aa0"}}, 1, 0, "select * from test2.t1 where name in ( \"aa0\", \"bb1\" );", "Query OK!")

	resp, _ = testAdvanceEndpointByPost1(t, gatewayHttpAddr, testEmail, "draft", "DsioGVSB", true, true, externalAppID, "/cstest00", &map[string]string{"id0": "0", "id1": "1", "name0": "aa0", "name1": "bb1"}, "get")
	advanceRes = checkAdvanceResp(resp, 200, "ok", 10)
	checkDataRes(t, advanceRes.Data[0], 0, "", [][]string{}, 0, 0, "drop database if exists test2;", "Query OK, 0 rows affected.*")
	checkDataRes(t, advanceRes.Data[1], 0, "", [][]string{}, 0, 0, "create database test2;", "Query OK, 0 rows affected.*")
	checkDataRes(t, advanceRes.Data[2], 0, "", [][]string{}, 0, 0, "use test2;", "Database changed!")
	checkDataRes(t, advanceRes.Data[3], 0, "", [][]string{}, 0, 0, "create table t1 (id int key, name varchar(10));", "Query OK, 0 rows affected.*")
	checkDataRes(t, advanceRes.Data[4], 0, "", [][]string{}, 0, 2, "insert into t1 values (0, \"aa0\"),(1, \"bb1\");", "Query OK, 2 rows affected.*")
	checkDataRes(t, advanceRes.Data[5], 0, "", [][]string{{"0", "aa0"}, {"1", "bb1"}}, 2, 0, "select * from test2.t1 where id in ( 0, 1 ) or name = \"bb1\";", "Query OK!")
	checkDataRes(t, advanceRes.Data[6], 0, "", [][]string{}, 0, 2, "update t1 set id=id+10 where name in ( \"aa0\", \"bb1\");", "Query OK, 2 rows affected.*")
	checkDataRes(t, advanceRes.Data[7], 0, "", [][]string{{"10", "aa0"}, {"11", "bb1"}}, 2, 0, "select * from test2.t1 where name in ( \"aa0\", \"bb1\" );", "Query OK!")
	checkDataRes(t, advanceRes.Data[8], 0, "", [][]string{}, 0, 1, "delete from t1 where name = \"bb1\";", "Query OK, 1 rows affected.*")
	checkDataRes(t, advanceRes.Data[9], 0, "", [][]string{{"10", "aa0"}}, 1, 0, "select * from test2.t1 where name in ( \"aa0\", \"bb1\" );", "Query OK!")
}
*/

// test for timeout
func TestGatewayService6(t *testing.T) {
	setupCentral("/Users/cs/.infra-api/dataapi1.yaml")
	centralHttpAddr := fmt.Sprintf("http://0.0.0.0:%v", config.Global().ServerConf.HttpPort)
	setupExecutor("/Users/cs/.infra-api/dataapi_executor.yaml")
	setupGateway("/Users/cs/.infra-api/dataapi_gateway.yaml")
	gatewayHttpAddr := fmt.Sprintf("http://0.0.0.0:%v", config.Global().ServerConf.HttpPort)

	time.Sleep(time.Second)

	projectId := uint64(43563)
	clusterId := "270048"
	databaseName := "test"
	testEmail := "chenshuang@pingcap.com"
	orgId := 30061
	appId := 5
	publicKey := "DsioGVSB"

	createEndpointParam := &entity.CreateEndpointParam{
		Name:         "cs-endpoint-test00",
		Endpoint:     "/cstest00",
		Method:       "get",
		ClusterIdStr: clusterId,
		Database:     databaseName,
		SqlTemplate:  "select sleep( ${time} );",
		Args: []entity.EndpointArgs{
			{
				Name:     "time",
				Type:     "number",
				Required: 1,
			},
		},
		Settings: entity.EndpointSettings{
			TimeOut:   1000,
			RowsLimit: 10,
		},
	}
	endpointID := testCreateEndpoint(t, centralHttpAddr, testEmail, publicKey, orgId, int(projectId), appId, createEndpointParam)
	defer func() {
		testDeleteEndpoint(t, centralHttpAddr, testEmail, orgId, int(projectId), appId, int(endpointID))
	}()

	externalAppID := "app-uzytMOJP"
	start := time.Now()
	resp, _ := testAdvanceEndpointByPost1(t, gatewayHttpAddr, testEmail, "draft", "DsioGVSB", true, false, externalAppID, "/cstest00", &map[string]string{"time": "10"}, "get")
	checkAdvanceResp := func(resp string, rpcCode int, rpcMsg string, sqlRespCount int) *dto2.AdvancedRes {
		advanceRes := &dto2.AdvancedRes{}
		err := json.Unmarshal([]byte(resp), advanceRes)
		require.NoError(t, err)
		require.Equal(t, rpcMsg, advanceRes.Data.Result.Message)
		require.Equal(t, rpcCode, advanceRes.Data.Result.Code)
		//require.Equal(t, sqlRespCount, len(advanceRes.Data))
		return advanceRes
	}
	checkAdvanceResp(resp, 400, "Execute endpoint sql timeout, timeout is 1s", 0)
	require.Less(t, time.Since(start), time.Second*10)
}

func TestGatewayService7(t *testing.T) {
	setupCentral("/Users/cs/.infra-api/dataapi1.yaml")
	centralHttpAddr := fmt.Sprintf("http://0.0.0.0:%v", config.Global().ServerConf.HttpPort)
	setupExecutor("/Users/cs/.infra-api/dataapi_executor.yaml")
	setupGateway("/Users/cs/.infra-api/dataapi_gateway.yaml")
	gatewayHttpAddr := fmt.Sprintf("http://0.0.0.0:%v", config.Global().ServerConf.HttpPort)

	time.Sleep(time.Second)

	projectId := uint64(43563)
	clusterId := "270048"
	databaseName := "test"
	testEmail := "chenshuang@pingcap.com"
	orgId := 30061
	appId := 5
	publicKey := "DsioGVSB"
	endpointPath := "/cs/test001"

	createEndpointParam := &entity.CreateEndpointParam{
		Name:         "cs-endpoint-test00",
		Endpoint:     endpointPath,
		Method:       "get",
		ClusterIdStr: clusterId,
		Database:     databaseName,
		SqlTemplate:  "select sleep(0.01);",
		Settings: entity.EndpointSettings{
			TimeOut:   10000,
			RowsLimit: 10,
		},
	}
	endpointID := testCreateEndpoint(t, centralHttpAddr, testEmail, publicKey, orgId, int(projectId), appId, createEndpointParam)
	defer func() {
		testDeleteEndpoint(t, centralHttpAddr, testEmail, orgId, int(projectId), appId, int(endpointID))
	}()

	externalAppID := "app-uzytMOJP"
	resp, _ := testAdvanceEndpointByPost1(t, gatewayHttpAddr, testEmail, "draft", "DsioGVSB", true, false, externalAppID, endpointPath, nil, "get")
	checkAdvanceResp := func(resp string, rpcCode int, rpcMsg string, sqlRespCount int) *dto2.AdvancedRes {
		advanceRes := &dto2.AdvancedRes{}
		err := json.Unmarshal([]byte(resp), advanceRes)
		require.NoError(t, err)
		require.Equal(t, rpcMsg, advanceRes.Data.Result.Message)
		require.Equal(t, rpcCode, advanceRes.Data.Result.Code)
		//require.Equal(t, sqlRespCount, len(advanceRes.Data))
		return advanceRes
	}
	checkAdvanceResp(resp, 200, "ok", 1)
}

func TestGatewayService8(t *testing.T) {
	setupCentral("/Users/cs/.infra-api/dataapi1.yaml")
	centralHttpAddr := fmt.Sprintf("http://0.0.0.0:%v", config.Global().ServerConf.HttpPort)
	setupExecutor("/Users/cs/.infra-api/dataapi_executor.yaml")
	setupGateway("/Users/cs/.infra-api/dataapi_gateway.yaml")
	// wait grpc server started.
	time.Sleep(500 * time.Millisecond)
	conn, err := grpc.Dial("0.0.0.0:"+config.Global().ServerConf.GrpcPort, grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		t.Fatal(err)
	}

	gatewayClient := dataapi.NewDataAPIGatewayServiceClient(conn)

	ctx := context.Background()
	projectId := uint64(43563)
	clusterIdStr := "270048"
	appId := uint64(5)
	databaseName := "test"
	testEmail := "chenshuang@pingcap.com"
	orgId := 30061
	publicKey := "DsioGVSB"

	createEndpointParam := &entity.CreateEndpointParam{
		Name:         "cs-endpoint-test00",
		Endpoint:     "/cstest00",
		Method:       "get",
		ClusterIdStr: clusterIdStr,
		Database:     databaseName,
		SqlTemplate:  "select * from t1 limit ${line};",
		Args: []entity.EndpointArgs{
			{
				Name:     "line",
				Type:     "NUMBER",
				Required: 1,
			},
		},
		Settings: entity.EndpointSettings{
			TimeOut:   10000,
			RowsLimit: 1,
		},
	}
	endpointID := testCreateEndpoint(t, centralHttpAddr, testEmail, publicKey, orgId, int(projectId), int(appId), createEndpointParam)
	defer func() {
		testDeleteEndpoint(t, centralHttpAddr, testEmail, orgId, int(projectId), int(appId), int(endpointID))
	}()

	testEndpointReq := &dataapi.TestEndpointReq{
		ProjectId:  projectId,
		AppId:      appId,
		EndpointId: endpointID,
		Args: []*dataapi.EndpointArg{
			{
				Name:  "line",
				Value: "10",
				Type:  "bigint",
			},
		},
		Credential: &dataapi.Credential{
			Type: dataapi.CrtType_AUTH0,
			Key:  "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCIsImtpZCI6Im1oSjk3OVJEQlFhOWU1U1I4ZUxwRCJ9.eyJodHRwczovL3RpZGJjbG91ZC5jb20vbG9naW5fcmVjZWl2ZWRfYXQiOjE2Nzc0NjM5NjUyMTksImVtYWlsIjoiaGFueGlhb21lbmcxMTFAZ21haWwuY29tIiwiZW1haWxfdmVyaWZpZWQiOnRydWUsImlzcyI6Imh0dHBzOi8vdGlkYi1kZXYudXMuYXV0aDAuY29tLyIsImF1ZCI6Iktpd1dhakxKM3B0bzhTc1o3RjM1TmFxTmh3bm9SaDFjIiwiaWF0IjoxNjc3NDYzOTY1LCJleHAiOjE2Nzc0OTk5NjUsInN1YiI6Imdvb2dsZS1vYXV0aDJ8MTA3MDYyNjE0NDEzNTMyMjY4MTE4IiwiYXRfaGFzaCI6ImxpdFpQT0Q0U1RzSDNfam1ZTDFPVnciLCJzaWQiOiJ4ZXNHT1ptOE9RTkVidkxYRXYtNm5VOTNvQmFGeW1QNyIsIm5vbmNlIjoiRDBGT1lzTHJ0aFNQYTUtQmRUbHVBRGpaYmRRR2NWMGcifQ.yTIVyiqOkbr7rvKUVeJ9-dckG3sWBgrblAzHb5Dz9ajGZYkqtciWaA-OPLJuYYS9PAuz1gyhrRCEuScHh0NO_T6XqRgoiRiGPf86rcj4moCCu2v4_pUXdp-fyaVL57wgZ6gGfmHdjvK7LVEZSUXMqsZivoKbTdRp2kOOLxiV5S1wxNZHo1wQCwIsiRLNGPpRIWtv1TWxCbz0Yk2FdjYQ1CUd8x1VIeZ8IQ6cmRX5yizHtwpBm-zkH-vIH1BVFBDP7D_xtEq7BtWbCY1seFs6xnOjbEW17-U_MGkvFWAZclzMaU72_oULGgV6R31Ur1tFWf3mxSIdBTvBWnQxMCUrmQ",
		},
	}
	resp, err := gatewayClient.TestEndpoint(ctx, testEndpointReq)
	require.NoError(t, err)
	require.Equal(t, 1, len(resp.StmtResults))
	require.Equal(t, int64(1), resp.StmtResults[0].RowNum)
	require.Equal(t, []string{"1", "a"}, resp.StmtResults[0].Rows[0].Items)
	require.Equal(t, dataapi.StmtType_Select, resp.StmtResults[0].StmtType)
}

func TestGatewayHTTPAPIError(t *testing.T) {
	setupCentral("/Users/cs/.infra-api/dataapi1.yaml")
	centralHttpAddr := fmt.Sprintf("http://0.0.0.0:%v", config.Global().ServerConf.HttpPort)
	setupExecutor("/Users/cs/.infra-api/dataapi_executor.yaml")
	setupGateway("/Users/cs/.infra-api/dataapi_gateway.yaml")
	gatewayHttpAddr := fmt.Sprintf("http://0.0.0.0:%v", config.Global().ServerConf.HttpPort)
	fmt.Printf("%v    %v --------\n\n", centralHttpAddr, gatewayHttpAddr)

	time.Sleep(time.Second)

	projectId := uint64(43563)
	clusterId := "2700480000"
	databaseName := "test"
	testEmail := "chenshuang@pingcap.com"
	orgId := 30061
	appId := 5
	externalAppID := "app-uzytMOJP"
	publicKey := "DsioGVSB"

	createEndpointParam := &entity.CreateEndpointParam{
		Name:         "cs-endpoint-test00",
		Endpoint:     "/cstest00",
		Method:       "get",
		ClusterIdStr: clusterId,
		Database:     databaseName,
		SqlTemplate:  "select * from t1 where id=${id};",
		Args: []entity.EndpointArgs{
			{
				Name:     "id",
				Type:     "number",
				Required: 1,
			},
		},
		Settings: entity.EndpointSettings{
			TimeOut:   3600,
			RowsLimit: 1000,
		},
	}
	endpointID := testCreateEndpoint(t, centralHttpAddr, testEmail, publicKey, orgId, int(projectId), appId, createEndpointParam)
	fmt.Printf("create endpoint %v ------\n\n", endpointID)
	defer func() {
		testDeleteEndpoint(t, centralHttpAddr, testEmail, orgId, int(projectId), appId, int(endpointID))
	}()

	// mock fail by connect user cluster failed by mock clusterID
	resp, code := testAdvanceEndpointByPost(t, gatewayHttpAddr, testEmail, "draft", publicKey, true, externalAppID, "/cstest00", &map[string]string{"id": "1"}, "get")
	require.Equal(t, `{"code":500,"message":"BuildConnection failed, Cluster.cloud.pingcap.com \"2700480000\" not found"}`, resp)
	require.Equal(t, 500, code)
}

func testAdvanceEndpointByPost(t *testing.T, host, email, endpointType, publicKey string, isTest bool, externalAppId, endpoint string, params *map[string]string, method string) (string, int) {
	return testAdvanceEndpoint(t, host, email, endpointType, publicKey, isTest, false, externalAppId, endpoint, params, method)
}

func testAdvanceEndpointByPost1(t *testing.T, host, email, endpointType, publicKey string, isTest, returnAll bool, externalAppId, endpoint string, params *map[string]string, method string) (string, int) {
	return testAdvanceEndpoint(t, host, email, endpointType, publicKey, isTest, returnAll, externalAppId, endpoint, params, method)
}

func testAdvanceEndpoint(t *testing.T, host, email, endpointType, publicKey string, isTest, returnAll bool, externalAppId, endpoint string, params *map[string]string, method string) (string, int) {
	header := map[string]string{
		"test-email":     email,
		"endpoint-type":  endpointType,
		"x-auth-type":    "openapi",
		"x-auth-content": fmt.Sprintf(`{"public_key":"%v"}`, publicKey),
	}
	if isTest {
		defer func() {
			controller_gateway.InTest = false
		}()
		controller_gateway.InTest = true
	}
	if returnAll {
		header["return_all_stmt_result"] = "true"
	}
	url := host + "/api/v1beta/app/" + externalAppId + "/endpoint" + endpoint
	var resp string
	var code int
	var err error
	switch method {
	case "post":
		resp, code, err = httpclient.PostWithHeader(url, params, &header)
	case "get":
		resp, code, err = httpclient.GetWithHeader(url, params, &header)
	case "put":
		resp, code, err = httpclient.PutWithHeader(url, params, &header)
	case "delete":
		resp, code, err = DeleteWithHeader(url, params, &header)
	default:
		t.Fatalf("unknown method %v", method)
	}
	require.NoError(t, err)
	return resp, code
}

func testEndpointHttpApi(t *testing.T, host string, req dto2.TestEndpointReq) (string, int) {
	header := map[string]string{
		"Content-Type": "application/json",
	}
	url := host + "/api/v1beta/endpoint/test"
	resp, code, err := httpclient.PostWithHeader(url, req, &header)
	require.NoError(t, err)
	return resp, code
}
func execSqlTemplateHttpApi(t *testing.T, host string, req dto2.ExecSqlTemplateReq) (string, int) {
	header := map[string]string{
		"Content-Type": "application/json",
	}
	url := host + "/api/v1beta/exec/sqltemplate"
	resp, code, err := httpclient.PostWithHeader(url, req, &header)
	require.NoError(t, err)
	return resp, code
}

func execSqlHttpApi(t *testing.T, host string, req dto2.ExecSqlReq) (string, int) {
	header := map[string]string{
		"Content-Type": "application/json",
	}
	url := host + "/api/v1beta/exec/sql"
	resp, code, err := httpclient.PostWithHeader(url, req, &header)
	require.NoError(t, err)
	return resp, code
}

func getDatabaseMetaHttpApi(t *testing.T, host string, req dto2.GetDatabaseMetaReq) (string, int) {
	header := map[string]string{
		"Content-Type": "application/json",
	}
	url := host + "/api/v1beta/database/meta"
	resp, code, err := httpclient.PostWithHeader(url, req, &header)
	require.NoError(t, err)
	return resp, code
}

func postHttpApi(t *testing.T, url string, req interface{}) (string, int) {
	header := map[string]string{
		"Content-Type": "application/json",
	}
	resp, code, err := httpclient.PostWithHeader(url, req, &header)
	require.NoError(t, err)
	fmt.Printf("\n---- resp: %v code: %v  --1--\n\n", resp, code)
	return resp, code
}

func DeleteWithHeader(url string, data interface{}, header *map[string]string) (string, int, error) {
	b, _ := json.Marshal(data)
	req, err := http.NewRequest(http.MethodDelete, url, bytes.NewReader(b))
	if err != nil {
		return "", 0, err
	}
	if header != nil {
		for k, v := range *header {
			req.Header.Set(k, v)
		}
	}
	httpClient := &http.Client{Timeout: time.Second * 360}
	resp, err := httpClient.Do(req)
	if nil != err {
		return "", 0, err
	}
	respdata, err := ioutil.ReadAll(resp.Body)
	return string(respdata), resp.StatusCode, err
}

func testCreateEndpoint(t *testing.T, host, email, publicKey string, orgId int, projectId int, appId int, createEndpointParam *entity.CreateEndpointParam) uint64 {
	header := map[string]string{
		"test-email":     email,
		"x-auth-type":    "openapi",
		"x-auth-content": fmt.Sprintf(`{"public_key":"%v"}`, publicKey),
	}
	url := host + "/api/v1/dataservices/central/orgs/" + strconv.Itoa(orgId) + "/projects/" + strconv.Itoa(int(projectId)) + "/apps/" + strconv.Itoa(appId) + "/endpoints"
	resp, code, err := httpclient.PostWithHeader(url, createEndpointParam, &header)
	require.NoError(t, err)
	require.Equal(t, 200, code)
	var res dto.CreateEndpointRes
	err = json.Unmarshal([]byte(resp), &res)
	require.Equal(t, "ok", res.Message)
	require.Equal(t, 200, res.Code)
	fmt.Printf("create endpoint %v ------\n\n", res.Endpoint.ID)
	return res.Endpoint.ID
}

func testPublishEndpoint(t *testing.T, host, email, publicKey string, orgId int, projectId int, appId, endpointID int) {
	header := map[string]string{
		"test-email":     email,
		"x-auth-type":    "openapi",
		"x-auth-content": fmt.Sprintf(`{"public_key":"%v"}`, publicKey),
	}
	url := host + "/api/v1/dataservices/central/orgs/" + strconv.Itoa(orgId) + "/projects/" + strconv.Itoa(int(projectId)) + "/apps/" + strconv.Itoa(appId) + "/endpoints/" + strconv.Itoa(endpointID) + "/online"
	resp, code, err := httpclient.PostWithHeader(url, nil, &header)
	require.NoError(t, err)
	require.Equal(t, 200, code)
	fmt.Printf("publish endpoint resp: %v     --\n\n", resp)
}

func testDeletePublishEndpoint(t *testing.T, host, email, publicKey string, orgId int, projectId int, appId, endpointID int) {
	header := map[string]string{
		"test-email":     email,
		"x-auth-type":    "openapi",
		"x-auth-content": fmt.Sprintf(`{"public_key":"%v"}`, publicKey),
	}
	url := host + "/api/v1/dataservices/central/orgs/" + strconv.Itoa(orgId) + "/projects/" + strconv.Itoa(int(projectId)) + "/apps/" + strconv.Itoa(appId) + "/endpoints/" + strconv.Itoa(endpointID) + "/offline"
	resp, code, err := httpclient.PutWithHeader(url, nil, &header)
	require.NoError(t, err)
	require.Equal(t, 200, code)
	fmt.Printf("offline endpoint resp: %v     --\n\n", resp)
}

func testDeleteEndpoint(t *testing.T, host, email string, orgId int, projectId int, appId int, endpointId int) {
	header := map[string]string{"test-email": email}
	url := host + "/api/v1/dataservices/central/orgs/" + strconv.Itoa(orgId) + "/projects/" + strconv.Itoa(projectId) + "/apps/" + strconv.Itoa(appId) + "/endpoints/" + strconv.Itoa(endpointId)

	_, code, err := httpclient.DeleteWithHeader(url, &header)
	require.NoError(t, err)
	require.Equal(t, 200, code)
}

func TestExecutorGrpc(t *testing.T) {
	//setupExecutor("/Users/cs/.infra-api/dataapi_executor.yaml")

	//conn, err := grpc.Dial("0.0.0.0:"+config.Global().ServerConf.GrpcPort, grpc.WithTransportCredentials(insecure.NewCredentials()))
	conn, err := grpc.Dial("0.0.0.0:10000", grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		t.Fatal(err)
	}
	executorClient := dataapi.NewDataAPIExecutorServiceClient(conn)

	ctx := context.Background()
	projectId := uint64(43563)
	clusterId := uint64(270048)
	databaseName := "test"
	endpointReq := &dataapi.BasicEndpointReqV2{
		ProjectId:  projectId,
		ClusterId:  clusterId,
		Database:   databaseName,
		Sql:        "select * from t1 where id = 1;",
		Args:       nil,
		SessionId:  "",
		LimitRows:  10,
		Credential: nil,
	}
	res, err := executorClient.ExecSqlDirectly(ctx, endpointReq)
	require.NoError(t, err)
	require.Equal(t, 1, len(res.StmtResults))
	require.Equal(t, int64(1), res.StmtResults[0].RowNum)
	require.Equal(t, []string{"1", "a"}, res.StmtResults[0].Rows[0].Items)

	res, err = executorClient.ExecSqlWithSession(ctx, endpointReq)
	require.NoError(t, err)
	require.Equal(t, int64(49900003), res.BaseResp.ErrCode)
	require.Equal(t, "Session id is required", res.BaseResp.ErrMsg)

	endpointReq.SessionId = "1"
	res, err = executorClient.ExecSqlWithSession(ctx, endpointReq)
	require.NoError(t, err)
	require.Equal(t, int64(49900001), res.BaseResp.ErrCode)
	require.Equal(t, "Sorry, your session has expired, please refresh the page and try again ^_^!", res.BaseResp.ErrMsg)

	resp, err := executorClient.CreateSession(ctx, &dataapi.CreateSessionReq{
		ProjectId:  projectId,
		ClusterId:  clusterId,
		Database:   databaseName,
		ExpireSec:  0,
		Credential: nil,
	})
	require.NoError(t, err)
	require.NotEqual(t, "", resp.SessionId)

	endpointReq.SessionId = resp.SessionId
	res, err = executorClient.ExecSqlWithSession(ctx, endpointReq)
	require.NoError(t, err)
	require.Equal(t, int64(1), res.StmtResults[0].RowNum)
	require.Equal(t, []string{"1", "a"}, res.StmtResults[0].Rows[0].Items)

	resp2, err := executorClient.CloseSession(ctx, &dataapi.CloseSessionReq{SessionId: resp.SessionId})
	require.NoError(t, err)
	require.Equal(t, int64(0), resp2.BaseResp.ErrCode)
	require.Equal(t, "", resp2.BaseResp.ErrMsg)
}

func TestGormExecMultiStmt(t *testing.T) {
	url := fmt.Sprintf("%s:%s@tcp(%s:%d)/%s?timeout=30s&allowCleartextPasswords=true&tidb_multi_statement_mode=true",
		"root",
		"",
		"0.0.0.0",
		4000,
		"test",
	)
	conn, err := gorm.Open(mysql.Open(url), &gorm.Config{
		Logger: logger.Default.LogMode(logger.Info),
	})
	require.NoError(t, err)
	sql := "dElete from t2;" +
		"insert into t2 values (1);" +
		"insert into t2 values (2);" +
		"insert into t2 values (3);" +
		"select * from t2;" +
		"insert into t2 values (4);" +
		"select * from t2;" +
		"insert into t2 values (5);"

	stmts, _, err := parser.New().ParseSQL(sql)
	require.NoError(t, err)
	for _, stmt := range stmts {
		stmtSQL := stmt.Text()
		fmt.Printf("%v\n", stmtSQL)
	}

	rows, err := conn.Raw(sql).Rows()
	require.NoError(t, err)
	res := &dbhubsdk.QueryRes{}
	dbhubsdk.ScanRows(context.Background(), rows, 0, res)
	fmt.Printf("rows: %v\n", res.Rows)
}

func TestGormExecTimeout(t *testing.T) {
	url := fmt.Sprintf("%s:%s@tcp(%s:%d)/%s?timeout=30s&allowCleartextPasswords=true&tidb_multi_statement_mode=true",
		"root",
		"",
		"0.0.0.0",
		4000,
		"test",
	)
	conn, err := gorm.Open(mysql.Open(url), &gorm.Config{
		Logger: logger.Default.LogMode(logger.Info),
	})
	require.NoError(t, err)

	db, err := conn.DB()
	require.NoError(t, err)

	ctx, cancel := context.WithTimeout(context.Background(), time.Second*3)
	defer cancel()
	start := time.Now()
	stmt, err := db.PrepareContext(ctx, "select * from t1 where a = ?")
	require.NoError(t, err)
	rows, err := stmt.QueryContext(ctx, true)
	//require.Error(t, err)
	//require.Equal(t, "context deadline exceeded", err.Error())
	res := &dbhubsdk.QueryRes{}
	dbhubsdk.ScanRows(ctx, rows, 0, res)
	require.Less(t, time.Since(start), time.Second*10)
}

func TestMySQLDriverExecMultiStmt(t *testing.T) {
	url := fmt.Sprintf("%s:%s@tcp(%s:%d)/%s?timeout=30s&allowCleartextPasswords=true&tidb_multi_statement_mode=true",
		"root",
		"",
		"0.0.0.0",
		4000,
		"test",
	)
	db, err := sql.Open("mysql", url)
	require.NoError(t, err)
	require.NoError(t, err)
	sql := "delete from t2;" +
		"insert into t2 values (1);" +
		"insert into t2 values (2);" +
		"insert into t2 values (3);" +
		"select * from t2;" +
		"insert into t2 values (4);" +
		"select * from t2;" +
		"insert into t2 values (5);"
	rows, err := db.Query(sql)
	require.NoError(t, err)
	res := &dbhubsdk.QueryRes{}
	dbhubsdk.ScanRows(context.Background(), rows, 0, res)
	fmt.Printf("rows: %v\n", res.Rows)
}

func setupExecutor(configPath string) {
	err := config.InitConfig(configPath)
	if err != nil {
		panic(fmt.Sprintf("Initialize config failed: %+v", err))
	}

	err = sentry.SetupLoggerAndSentry(config.Global().SentryConf.DSN, config.Global().SentryConf.Environment, config.Global().SentryConf.LogLevel)
	if err != nil {
		panic(fmt.Sprintf("Initialize logger and sentry failed: %+v", err))
	}
	config := config.Global()
	database.Connect(context.Background(),
		config.MysqlConf.User,
		config.MysqlConf.Password,
		config.MysqlConf.Host,
		config.MysqlConf.Port,
		config.MysqlConf.Database,
		config.MysqlConf.CharSet,
		config.MysqlConf.TimeZone,
	)

	metrics.RegisterDataApiExecutorMetrics()

	go setupExecutorGRPC()
}

func setupGateway(configPath string) {
	err := config.InitConfig(configPath)
	if err != nil {
		panic(fmt.Sprintf("Initialize config failed: %+v", err))
	}

	config := config.Global()
	database.Connect(context.Background(),
		config.MysqlConf.User,
		config.MysqlConf.Password,
		config.MysqlConf.Host,
		config.MysqlConf.Port,
		config.MysqlConf.Database,
		config.MysqlConf.CharSet,
		config.MysqlConf.TimeZone,
	)

	apikeygenerator.InitEncrypter(
		config.DataApiKeyConf.SecretKey,
		config.DataApiKeyConf.SecretIv,
	)

	metrics.RegisterDataApiGatewayMetrics()

	//err = dbhubsdk.InitSdk(
	//	config.DBHubSdkConf.KubeConfigPath,
	//	config.DBHubSdkConf.GardenerProjectName,
	//	config.DBHubSdkConf.InfraAPIConfigPath)
	//if err != nil {
	//	panic(fmt.Sprintf("Initialize dbhubsdk failed: %+v", err))
	//}

	go setupGatewayHTTP()
	go setupGatewayGRPC()
}

func setupCentral(configPath string) {
	err := config.InitConfig(configPath)
	if err != nil {
		panic(fmt.Sprintf("Initialize config failed: %+v", err))
	}

	metrics.RegisterCentralMetrics()

	go startupCentral(context.Background(), config.Global())
}

func startupCentral(ctx context.Context, centralConfig *config.CentralConfig) {
	serverConfig := centralConfig.ServerConf
	ginEng := tigin.Default()
	api.RouteCentralAPI(ginEng)
	addr := ":" + serverConfig.HttpPort
	tilog.Info(ctx, "serving HTTP Service", zap.String("address", addr))
	panic(fmt.Sprintf("HTTP Service died %+v", ginEng.Run(addr)))
}

func setupGatewayHTTP() {
	ginEng := tigin.Default()
	api.RouteDataApiGatewayAPI(ginEng)
	addr := ":" + config.Global().ServerConf.HttpPort
	tilog.Info(context.Background(), "gateway serving HTTP Service", zap.String("address", addr))
	panic(fmt.Sprintf("HTTP Service died %+v", ginEng.Run(addr)))
}

func setupGatewayGRPC() {
	grpcServer := newGRPCServer()
	config := config.Global()
	server, err := controller_gateway.CreateDataAPIGatewayService(config.GrpcConf)
	if err != nil {
		panic(err)
	}
	dataapi.RegisterDataAPIGatewayServiceServer(grpcServer, server)
	addr := ":" + config.ServerConf.GrpcPort
	tilog.Info(context.Background(), "start gateway rpc", zap.String("addr", addr))
	listen, err := net.Listen("tcp", addr)
	if err != nil {
		panic(err)
	}
	err = grpcServer.Serve(listen)
	if err != nil {
		panic(err)
	}
}

func setupExecutorGRPC() {
	grpcServer := newGRPCServer()
	dataapi.RegisterDataAPIExecutorServiceServer(grpcServer, controller.NewDataAPIExecutorService())
	addr := ":" + config.Global().ServerConf.GrpcPort
	tilog.Info(context.Background(), "start executor rpc", zap.String("addr", addr))
	listen, err := net.Listen("tcp", addr)
	if err != nil {
		panic(err)
	}
	err = grpcServer.Serve(listen)
	if err != nil {
		panic(err)
	}
}

func newGRPCServer() *grpc.Server {
	ctx := context.Background()
	server := tigrpc.NewServer(
		tigrpc.WithDefault(tilog.WithContextL(ctx)),
		tigrpc.WithUnaryInterceptors(
			grpc_middleware.ChainUnaryServer(
				metrics.UnaryServerInterceptor(),
				errs.UnaryServerInterceptor(),
			),
		),
	)
	return server
}