// Package worker_sampledataapp add by blacknc <blacknc@163.com> at 2023/09/25 17:08:31
package worker_sampledataapp

import (
	"dataservices/commons/grpc"
	"dataservices/commons/http"
	"dataservices/dbhubsdk"
	dtoGateway "dataservices/internal/dto/dataapi_gateway"
	"dataservices/internal/entity"
	"fmt"

	"github.com/tidbcloud/account/idl/pbgen/proto/acl"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
)

type ImportDataState struct{}

func (s *ImportDataState) String() string {
	return entity.SampleDataTaskStateImportData
}

func (s *ImportDataState) Exec(c *SampleDataappContext) {
	var err error
	defer func() {
		if err != nil {
			c.Trans(&FailState{s, err})
		} else {
			c.Trans(&CreateEndpointState{})
		}
	}()

	ctx := c.ctx
	orgID := c.dataapp.OrgId
	projectID := c.dataapp.ProjectId
	dataappID := c.dataapp.ID
	createdBy := c.dataapp.CreatedBy
	clusterID := c.clusterID

	clusterInfo, err := grpc.GetCluster(ctx, orgID, projectID, clusterID)
	if err != nil {
		return
	}

	credential, err := getCredential(c, clusterID, createdBy)
	if err != nil {
		return
	}

	sessionRes, err := http.CreateSession(ctx, clusterInfo.RegionName, &dtoGateway.CreateSessionReq{
		ProjectId:  projectID,
		ClusterId:  clusterID,
		ExpireSec:  300,
		Credential: credential,
	})
	if err != nil {
		return
	}
	defer func() {
		if err := http.CloseSession(ctx, clusterInfo.RegionName, &dtoGateway.CloseSessionReq{
			ProjectId:  projectID,
			ClusterId:  clusterID,
			Credential: credential,
			SessionId:  sessionRes.SessionId,
		}); err != nil {
			tilog.Error(ctx, "close session failed", zap.Uint64("dataapp_id", dataappID), zap.String("session_id", sessionRes.SessionId), zap.Error(err))
		}
	}()
	sessionID := sessionRes.SessionId

	database, table := "fortune500", "fortune500_2018_2022"
	exists, err := databaseExists(c, credential, clusterInfo, sessionID, database)
	if err != nil {
		return
	} else if !exists {
		sql := fmt.Sprintf("CREATE DATABASE `%s`;", database)
		if _, err = execSql(c, credential, clusterInfo, sessionID, sql); err != nil {
			return
		}
	}

	exists, err = tableExists(c, credential, clusterInfo, sessionID, database, table)
	if err != nil {
		return
	} else if !exists {
		sql := fmt.Sprintf("USE `%s`;", database)
		if _, err = execSql(c, credential, clusterInfo, sessionID, sql); err != nil {
			return
		}
		if _, err = execSql(c, credential, clusterInfo, sessionID, Fortune500DDL); err != nil {
			return
		}
		if _, err = execSql(c, credential, clusterInfo, sessionID, Fortune500Data); err != nil {
			return
		}
	}
}

func getCredential(c *SampleDataappContext, clusterID uint64, email string) (credential dtoGateway.Credential, err error) {
	res, err := grpc.GetJwtByUser(c.ctx, &acl.GetJwtByUserReq{
		ClusterId: clusterID,
		Email:     email,
	})
	if err != nil {
		return
	}
	credential.Type = dtoGateway.TypeCredentialAccount
	credential.Key = dbhubsdk.EncryptDBAccount(res.DbUser, res.Jwt)
	credential.AccountID = email
	return
}

func databaseExists(c *SampleDataappContext, credential dtoGateway.Credential, clusterInfo *core.Cluster, sessionID, database string) (bool, error) {
	sql := fmt.Sprintf("SELECT schema_name FROM INFORMATION_SCHEMA.schemata WHERE schema_name='%s'", database)
	metaRes, err := execSql(c, credential, clusterInfo, sessionID, sql)
	if err != nil {
		return false, err
	}
	return len(metaRes.Data.Rows) > 0, nil
}

func tableExists(c *SampleDataappContext, credential dtoGateway.Credential, clusterInfo *core.Cluster, sessionID, database, table string) (bool, error) {
	sql := fmt.Sprintf("SELECT table_schema,table_name FROM INFORMATION_SCHEMA.tables WHERE table_schema='%s' AND table_name='%s'", database, table)
	metaRes, err := execSql(c, credential, clusterInfo, sessionID, sql)
	if err != nil {
		return false, err
	}
	return len(metaRes.Data.Rows) > 0, nil
}

func execSql(c *SampleDataappContext, credential dtoGateway.Credential, clusterInfo *core.Cluster, sessionID, sql string) (*dtoGateway.ExecSqlResp, error) {
	return http.ExecSql(c.ctx, clusterInfo.RegionName, &dtoGateway.ExecSqlReq{
		ProjectId:  c.dataapp.ProjectId,
		ClusterId:  clusterInfo.Id,
		Sql:        sql,
		SessionId:  sessionID,
		Credential: credential,
	})
}