package regional_dataapi_gateway

import (
	"bytes"
	"context"
	"dataservices/api"
	"dataservices/commons/config"
	http2 "dataservices/commons/http"
	"dataservices/commons/metrics"
	"dataservices/commons/sentry"
	"dataservices/dbhubsdk"
	"dataservices/ddb"
	controller "dataservices/internal/controller/dataapi_executor"
	controller_gateway "dataservices/internal/controller/dataapi_gateway"
	dto2 "dataservices/internal/dto/dataapi_gateway"
	"dataservices/internal/entity"
	"dataservices/internal/repository_ddb"
	"encoding/json"
	"fmt"
	"io"
	"io/ioutil"
	"net"
	"net/http"
	"sort"
	"testing"
	"time"

	grpc_middleware "github.com/grpc-ecosystem/go-grpc-middleware"
	"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/transport/http/tigin"
	tigrpc "github.com/tidbcloud/tiinf/transport/tigrpc/server"
	"go.uber.org/zap"
	"google.golang.org/grpc"
)

/*
//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 TestUpdateData2DDB(t *testing.T) {
	ctx := context.Background()
	configPath := "/Users/lvhongmeng/dataapi_executor.yaml"
	//setupExecutor(configPath)
	//setupGateway(configPath)
	//createTableInDdb()
	// wait grpc server started.
	//time.Sleep(500 * time.Millisecond)
	err := config.InitConfig(configPath)
	if err != nil {
		panic(fmt.Sprintf("Initialize config failed: %+v", err))
	}

	config := config.Global()
	err = ddb.Init(ctx, config)
	if err != nil {
		panic(fmt.Sprintf("Initialize ddb failed: %+v", err))
	}

	//dbhubsdk.InitRegionalSdk(ddb.Client())
	endpointArgs, err := json.Marshal([]entity.EndpointArgs{
		{
			Name:     "phone",
			Type:     "string",
			Required: 1,
		},
		{
			Name:     "name",
			Type:     "string",
			Required: 1,
		},
	})
	require.NoError(t, err)

	genEndpointSetting := func(timeout, rowLimit int) string {
		endpointSetting, err := json.Marshal(entity.EndpointSettings{
			TimeOut:   30000,
			RowsLimit: 50,
		})
		require.NoError(t, err)
		return string(endpointSetting)
	}

	endpoint := entity.Endpoint{
		ID:          1868378,
		DataappId:   1470223,
		Name:        "/student3s",
		Type:        "chat2sql",
		Endpoint:    "/student3s",
		Method:      "DELETE",
		ClusterId:   3299042,
		Database:    "sample_data",
		SqlTemplate: "USE sample_data;\nDELETE FROM `student3` WHERE phone = ${phone} and name = ${name} ;",
		Args:        string(endpointArgs),
		Settings:    genEndpointSetting(100000, 50),
		CreatedBy:   "hongmeng.lv@pingcap.com",
		Batch:       false,
		Tag:         "Default",
		RWMode:      "READWRITE",
	}
	client := ddb.Client()
	err = repository_ddb.PutEndpoint(ctx, client, &endpoint)
	require.NoError(t, err)
}

func TestGatewayHttpService1(t *testing.T) {
	setupExecutor("/Users/cs/.infra-api/regional_dataapi_executor.yaml")
	setupGateway("/Users/cs/.infra-api/regional_dataapi_gateway.yaml")
	//createTableInDdb()
	// wait grpc server started.
	time.Sleep(500 * time.Millisecond)

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

	ctx := context.Background()
	projectId := uint64(43563)
	clusterId := uint64(3148938)
	//clusterIdStr := "270048"
	appId := uint64(5)
	databaseName := "test"
	testEmail := "chenshuang@pingcap.com"
	orgId := 30061
	credential := dto2.Credential{
		Type: "APIKEY",
		Key:  "APZPEHNb",
	}
	//publicKey := "DsioGVSB"

	var err error

	//userClusterInfo := entity.UserClusterInfo{
	//	ProjectId:       projectId,
	//	ClusterId:       clusterId,
	//	IsDedicate:      false,
	//	EndpointService: "",
	//	EndpointId:      "",
	//	Host:            "gateway01.us-east-1.dev.shared.aws.tidbcloud.com",
	//	Port:            4000,
	//}
	//err = repository_ddb.PutUserClusterInfo(ctx, ddb.Client(), &userClusterInfo)
	//require.NoError(t, err)

	app := entity.DataApp{
		ID:            appId,
		OrgId:         uint64(orgId),
		ProjectId:     projectId,
		ExternalAppID: "app_id_cs",
		AppName:       "test_cs",
		AppType:       "tp-01",
		CreatedBy:     "chenshuang@pingcap.com",
		IsDeleted:     0,
	}
	err = repository_ddb.PutDataApp(ctx, ddb.Client(), &app)
	require.NoError(t, err)

	endpointArgs, err := json.Marshal([]entity.EndpointArgs{
		{
			Name:     "id",
			Type:     "NUMBER",
			Required: 1,
		},
	})
	require.NoError(t, err)

	genEndpointSetting := func(timeout, rowLimit int) string {
		endpointSetting, err := json.Marshal(entity.EndpointSettings{
			TimeOut:   timeout,
			RowsLimit: rowLimit,
		})
		require.NoError(t, err)
		return string(endpointSetting)
	}

	defaultLimitRows := int64(1000)
	endpointID := uint64(101)
	endpoint := entity.Endpoint{
		ID:          endpointID,
		DataappId:   app.ID,
		Name:        "endpoint-1",
		Type:        "draft",
		Endpoint:    "/testcs00",
		Method:      "get",
		ClusterId:   clusterId,
		Database:    databaseName,
		SqlTemplate: "select * from t1 where id = ${id};",
		Args:        string(endpointArgs),
		Settings:    genEndpointSetting(100000, int(defaultLimitRows)),
		CreatedBy:   testEmail,
	}
	err = repository_ddb.PutEndpoint(ctx, ddb.Client(), &endpoint)
	require.NoError(t, err)

	req := dto2.TestEndpointReq{
		ProjectId:           projectId,
		AppId:               appId,
		EndpointId:          endpointID,
		EndpointType:        "",
		Args:                nil,
		ReturnAllStmtResult: false,
		Credential:          credential,
	}
	//APZPEHNb:152bec2b-2fa2-4355-aa84-173dd7a15e05
	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.Contains(t, res.Data.Result.Message, rpcMsg)
		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",
		},
	}
	resp, code = testEndpointHttpApi(t, gatewayHttpAddr, req)
	require.Equal(t, 200, code)
	execSqlResp := checkResp(resp, 200, "Query OK!", 1, 0, defaultLimitRows, "Query OK!")
	require.Equal(t, 2, len(execSqlResp.Data.Columns))
	require.Equal(t, []entity.Row{{"id": "1", "name": "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, "endpoint not found", 0, 0, 0, "")
	execSqlResp = checkResp(resp, errs.ErrCodeNotFound, "item not exists", 0, 0, 0, "")

	// Test exec endpoint sql failed.
	err = repository_ddb.DeleteEndpoint(ctx, ddb.Client(), endpointID)
	require.NoError(t, err)
	endpoint.SqlTemplate = "select * from t_not_exist;"
	endpoint.Settings = genEndpointSetting(100000, 1)
	err = repository_ddb.PutEndpoint(ctx, ddb.Client(), &endpoint)
	require.NoError(t, err)
	req.EndpointId = endpointID
	resp, code = testEndpointHttpApi(t, gatewayHttpAddr, req)
	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.
	err = repository_ddb.DeleteEndpoint(ctx, ddb.Client(), endpointID)
	require.NoError(t, err)
	endpoint.SqlTemplate = "invalid sql;"
	err = repository_ddb.PutEndpoint(ctx, ddb.Client(), &endpoint)
	require.NoError(t, err)
	req.EndpointId = endpointID
	resp, code = testEndpointHttpApi(t, gatewayHttpAddr, req)
	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,
		Credential:          credential,
	}

	// 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, "Query OK!", 1, 0, 1, "Query OK!")
	require.Equal(t, 2, len(execSqlResp.Data.Columns))
	require.Equal(t, []entity.Row{{"id": "2", "name": "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,
		Credential:          credential,
	}

	// 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, "Query OK!", 1, 0, 1, "Query OK!")
	require.Equal(t, 2, len(execSqlResp.Data.Columns))
	require.Equal(t, []entity.Row{{"id": "2", "name": "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,
		Credential: credential,
	}
	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: credential,
	}
	resp, code = postHttpApi(t, gatewayHttpAddr+"/inner/dataservice/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, "Query 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, "Query OK", 1, 0, 30, "Query OK")
	require.Equal(t, 2, len(execSqlResp.Data.Columns))
	require.Equal(t, []entity.Row{{"id": "10", "name": "abc"}}, execSqlResp.Data.Rows)

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

	closeSessionReq := dto2.CloseSessionReq{
		SessionId: createSessionResp.SessionId,
	}
	resp, code = postHttpApi(t, gatewayHttpAddr+"/inner/dataservice/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+"/inner/dataservice/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+"/inner/dataservice/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 TestGatewayHttpService2(t *testing.T) {
	metrics.RegisterCentralMetrics()
	setupExecutor("/Users/cs/.infra-api/regional_dataapi_executor.yaml")
	setupGateway("/Users/cs/.infra-api/regional_dataapi_gateway.yaml")
	//createTableInDdb()
	// wait grpc server started.
	time.Sleep(500 * time.Millisecond)

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

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

	var err error

	userClusterInfo := entity.UserClusterInfo{
		ProjectId:       projectId,
		ClusterId:       clusterId,
		IsDedicate:      false,
		EndpointService: "",
		EndpointId:      "",
		Host:            "gateway01.us-east-1.dev.shared.aws.tidbcloud.com",
		Port:            4000,
	}
	err = repository_ddb.PutUserClusterInfo(ctx, ddb.Client(), &userClusterInfo)
	require.NoError(t, err)

	app := entity.DataApp{
		ID:            appId,
		OrgId:         uint64(orgId),
		ProjectId:     projectId,
		ExternalAppID: "app_id_cs",
		AppName:       "test_cs",
		AppType:       "tp-01",
		CreatedBy:     "chenshuang@pingcap.com",
		IsDeleted:     0,
	}
	err = repository_ddb.PutDataApp(ctx, ddb.Client(), &app)
	require.NoError(t, err)

	endpointArgs, err := json.Marshal([]entity.EndpointArgs{
		{
			Name:     "id",
			Type:     "NUMBER",
			Required: 1,
		},
	})
	require.NoError(t, err)

	genEndpointSetting := func(timeout, rowLimit int) string {
		endpointSetting, err := json.Marshal(entity.EndpointSettings{
			TimeOut:   timeout,
			RowsLimit: rowLimit,
		})
		require.NoError(t, err)
		return string(endpointSetting)
	}

	defaultLimitRows := int64(1000)
	endpointID := uint64(101)
	endpoint := entity.Endpoint{
		ID:          endpointID,
		DataappId:   app.ID,
		Name:        "endpoint-1",
		Type:        "draft",
		Endpoint:    "/testcs00",
		Method:      "get",
		ClusterId:   clusterId,
		Database:    databaseName,
		SqlTemplate: "select * from t1 where id = ${id};",
		Args:        string(endpointArgs),
		Settings:    genEndpointSetting(100000, int(defaultLimitRows)),
		CreatedBy:   testEmail,
	}
	err = repository_ddb.PutEndpoint(ctx, ddb.Client(), &endpoint)
	require.NoError(t, err)

	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(res *dto2.ExecSqlResp, rpcCode int64, rpcMsg string, rowCount, rowAffect, limitRows int64, query string) *dto2.ExecSqlResp {
		//require.Equal(t, rpcMsg, res.Data.Result.Message)
		require.Contains(t, res.Data.Result.Message, rpcMsg)
		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
	}

	DataserviceGatewayAddr = gatewayHttpAddr
	region := "" // no need in test.

	resp, err := http2.TestEndpoint(ctx, region, &req)
	fmt.Printf("%#v   %#v  ---\n\n", resp, err)
	require.Error(t, err)
	require.Equal(t, "Parse endpoint sql failed, Param 'id' is required", err.Error())
	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",
		},
	}
	resp, err = http2.TestEndpoint(ctx, region, &req)
	fmt.Printf("%#v   %#v  ---\n\n", resp, err)
	require.NoError(t, err)
	execSqlResp := checkResp(resp, 200, "Query OK!", 1, 0, defaultLimitRows, "Query OK!")
	require.Equal(t, 2, len(execSqlResp.Data.Columns))
	require.Equal(t, []entity.Row{{"id": "1", "name": "a"}}, execSqlResp.Data.Rows)

	// Test endpoint not found
	req.EndpointId = 12345678901
	resp, err = http2.TestEndpoint(ctx, region, &req)
	fmt.Printf("%#v   %#v  ---\n\n", resp, err)
	require.Error(t, err)
	require.Equal(t, "endpoint not found", err.Error())
	execSqlResp = checkResp(resp, errs.ErrCodeNotFound, "endpoint not found", 0, 0, 0, "")

	// Test exec endpoint sql failed.
	err = repository_ddb.DeleteEndpoint(ctx, ddb.Client(), endpointID)
	require.NoError(t, err)
	endpoint.SqlTemplate = "select * from t_not_exist;"
	endpoint.Settings = genEndpointSetting(100000, 1)
	err = repository_ddb.PutEndpoint(ctx, ddb.Client(), &endpoint)
	require.NoError(t, err)
	req.EndpointId = endpointID
	resp, err = http2.TestEndpoint(ctx, region, &req)
	require.Error(t, err)
	require.Equal(t, "TestEndpoint failed[1146]", err.Error())
	execSqlResp = checkResp(resp, 1146, "Table 'test.t_not_exist' doesn't exist", 0, 0, 1, "")

	// Test endpoint sql wrong syntax.
	err = repository_ddb.DeleteEndpoint(ctx, ddb.Client(), endpointID)
	require.NoError(t, err)
	endpoint.SqlTemplate = "invalid sql;"
	err = repository_ddb.PutEndpoint(ctx, ddb.Client(), &endpoint)
	require.NoError(t, err)
	req.EndpointId = endpointID
	resp, err = http2.TestEndpoint(ctx, region, &req)
	require.Error(t, err)
	require.Equal(t, "Parsing sql failed, Please check if sql syntax is correct, line 1 column 7 near \"invalid sql;\" ", err.Error())
	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, err = http2.TestEndpoint(ctx, region, &req)
	fmt.Printf("%#v   %#v  ---\n\n", resp, err)
	require.Error(t, err)
	require.Equal(t, "Unknown credential type: 'key', expected values: 'AUTH0', 'APIKEY'", err.Error())
	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, err = http2.ExecSqlTemplate(ctx, region, &execSQLTemplateReq)
	fmt.Printf("%#v   %#v  ---\n\n", resp, err)
	require.Error(t, err)
	require.Equal(t, "Parsing sql failed, Please check if sql syntax is correct, line 1 column 30 near \"{id};\" ", err.Error())
	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, err = http2.ExecSqlTemplate(ctx, region, &execSQLTemplateReq)
	fmt.Printf("%#v   %#v  ---\n\n", resp, err)
	require.NoError(t, err)
	execSqlResp = checkResp(resp, 200, "Query OK!", 1, 0, 1, "Query OK!")
	require.Equal(t, 2, len(execSqlResp.Data.Columns))
	require.Equal(t, []entity.Row{{"id": "2", "name": "b"}}, execSqlResp.Data.Rows)

	execSQLTemplateReq.SessionId = "unknown_session_id"
	resp, err = http2.ExecSqlTemplate(ctx, region, &execSQLTemplateReq)
	fmt.Printf("%#v   %#v  ---\n\n", resp, err)
	require.Error(t, err)
	require.Equal(t, "Sorry, your session has expired, please refresh the page and try again ^_^!", err.Error())
	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, err = http2.ExecSqlTemplate(ctx, region, &execSQLTemplateReq)
	fmt.Printf("%#v   %#v  ---\n\n", resp, err)
	require.Error(t, err)
	require.Equal(t, "ExecSqlTemplate failed[1146]", err.Error())
	execSqlResp = checkResp(resp, 1146, "Table 'test.t_not_exist' doesn't exist", 0, 0, 1, "")

	getDatabaseMetaReq := dto2.GetDatabaseMetaReq{
		ProjectId: projectId,
		ClusterId: clusterId,
		Database:  databaseName,
	}
	getDatabaseMetaResp, err := http2.GetDatabaseMeta(ctx, region, &getDatabaseMetaReq)
	fmt.Printf("%#v   %#v  ---\n\n", getDatabaseMetaResp, err)
	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"
	getDatabaseMetaResp, err = http2.GetDatabaseMeta(ctx, region, &getDatabaseMetaReq)
	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{},
	}
	createSessionResp, err := http2.CreateSession(ctx, region, &createSessionReq)
	require.NoError(t, err)
	require.Equal(t, int64(0), createSessionResp.Code)
	require.Equal(t, "", createSessionResp.Message)
	require.NotEqual(t, "", createSessionResp.SessionId)

	execSQLReq := dto2.ExecSqlReq{
		ProjectId:           projectId,
		ClusterId:           clusterId,
		Database:            databaseName,
		Sql:                 "select * from t1 where id = ?;",
		SessionId:           "",
		LimitRows:           1,
		Args:                nil,
		TimeoutMs:           20000,
		ReturnAllStmtResult: false,
	}
	execSQLReq.Sql = "begin; insert into t1 values (10,'abc')"
	execSQLReq.SessionId = createSessionResp.SessionId
	resp, err = http2.ExecSql(ctx, region, &execSQLReq)
	require.NoError(t, err)
	execSqlResp = checkResp(resp, 200, "Query OK", 0, 1, 1, "Query OK")

	execSQLReq.Sql = "select * from t1 where id = 10"
	execSQLReq.LimitRows = 30
	execSQLReq.SessionId = createSessionResp.SessionId
	resp, err = http2.ExecSql(ctx, region, &execSQLReq)
	require.NoError(t, err)
	execSqlResp = checkResp(resp, 200, "Query OK", 1, 0, 30, "Query OK")
	require.Equal(t, 2, len(execSqlResp.Data.Columns))
	require.Equal(t, []entity.Row{{"id": "10", "name": "abc"}}, execSqlResp.Data.Rows)

	execSQLReq.SessionId = ""
	resp, err = http2.ExecSql(ctx, region, &execSQLReq)
	require.NoError(t, err)
	execSqlResp = checkResp(resp, 200, "Query OK", 0, 0, 30, "Query OK")
	require.Equal(t, []entity.Row{}, execSqlResp.Data.Rows)

	//closeSessionReq := dto2.CloseSessionReq{
	//	SessionId: createSessionResp.SessionId,
	//}
	//closeSessionResp, err := http2.CloseSession(ctx, region, &closeSessionReq)
	//require.NoError(t, err)
	//require.Equal(t, int64(0), closeSessionResp.Code)
	//require.Equal(t, "", closeSessionResp.Message)
}

var DataserviceGatewayAddr = ""

func PostRequestToLocalGateway(ctx context.Context, region, url string, data interface{}, result interface{}, method string) (int, error) {
	tilog.Info(ctx, fmt.Sprintf("http/dataapi-gateway:%s begin!payload:%+v", method, data))

	url = DataserviceGatewayAddr + url
	header := map[string]string{
		"Content-Type": "application/json",
	}
	return Post(url, data, &header, result)
}

func Post(url string, data interface{}, header *map[string]string, result interface{}) (int, error) {
	req, err := http.NewRequest(http.MethodPost, url, encodeJSON(data))
	if err != nil {
		return 0, err
	}
	if header != nil {
		for k, v := range *header {
			req.Header.Set(k, v)
		}
	}
	httpClient := &http.Client{}
	resp, err := httpClient.Do(req)
	if nil != err {
		return 0, err
	}
	respdata, err := ioutil.ReadAll(resp.Body)
	if err != nil && err != io.EOF {
		return resp.StatusCode, err
	}
	err = json.Unmarshal(respdata, result)
	return resp.StatusCode, err
}

func encodeJSON(data interface{}) io.Reader {
	b, _ := json.Marshal(data)
	return bytes.NewReader(b)
}

func testEndpointHttpApi(t *testing.T, host string, req dto2.TestEndpointReq) (string, int) {
	header := map[string]string{
		"Content-Type": "application/json",
	}
	url := host + "/inner/dataservice/endpoint/test"
	resp, code, err := PostWithHeader(url, req, &header)
	require.NoError(t, err)
	fmt.Printf("%v %v --------\n\n", resp, code)
	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 + "/inner/dataservice/exec/sqltemplate"
	resp, code, err := 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 + "/inner/dataservice/exec/sql"
	resp, code, err := 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 + "/inner/dataservice/database/meta"
	resp, code, err := 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 := PostWithHeader(url, req, &header)
	require.NoError(t, err)
	fmt.Printf("\n---- resp: %v code: %v  --1--\n\n", resp, code)
	return resp, code
}

var (
	httpClient = &http.Client{Timeout: time.Second * 120}
)

func PostWithHeader(url string, data interface{}, header *map[string]string) (string, int, error) {
	req, err := http.NewRequest(http.MethodPost, url, encodeJSON(data))
	if err != nil {
		return "", 0, err
	}
	if header != nil {
		for k, v := range *header {
			req.Header.Set(k, v)
		}
	}
	resp, err := httpClient.Do(req)
	if nil != err {
		return "", 0, err
	}
	respdata, err := ioutil.ReadAll(resp.Body)
	return string(respdata), resp.StatusCode, err
}

func createTableInDdb() {
	ctx := context.Background()
	cli := ddb.Client()
	err := repository_ddb.CreateDataAppRegionTable(ctx, cli)
	fmt.Printf("CreateDataAppRegionTable: %v\n", err)
	err = repository_ddb.CreateEndpointRegionTable(ctx, cli)
	fmt.Printf("CreateEndpointRegionTable: %v\n", err)
	err = repository_ddb.CreateEndpointPublishRegionTable(ctx, cli)
	fmt.Printf("CreateEndpointPublishRegionTable: %v\n", err)
	err = repository_ddb.CreateUserClusterInfoTableIfNotExist(ctx, cli)
	fmt.Printf("CreateUserClusterInfoTableIfNotExist: %v\n", err)
}

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

	config := config.Global()

	metrics.RegisterDataApiGatewayMetrics()

	controller_gateway.InitRBACConfig(config.AwsConf, config.RBACConf)

	err = ddb.Init(context.Background(), config)
	if err != nil {
		panic(fmt.Sprintf("Initialize ddb failed: %+v", err))
	}

	go setupGatewayHTTP()
	go setupGatewayGRPC()
}

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 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()

	metrics.RegisterDataApiExecutorMetrics()

	err = ddb.Init(context.Background(), config)
	if err != nil {
		panic(fmt.Sprintf("Initialize ddb failed: %+v", err))
	}

	dbhubsdk.InitRegionalSdk(ddb.Client())

	go setupExecutorGRPC()
}

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 {
	server := tigrpc.NewServer(
		tigrpc.WithDefault(),
		tigrpc.WithUnaryInterceptors(
			grpc_middleware.ChainUnaryServer(
				metrics.UnaryServerInterceptor(),
				errs.UnaryServerInterceptor(),
			),
		),
	)
	return server
}