package service

import (
	"context"
	"dataapps/commons/grpc"
	"dataapps/commons/logger"
	dto "dataapps/internal/dto/sqleditor"
	entity "dataapps/internal/entity/sqleditor"
	"dataapps/internal/service/util"
	"fmt"
	"strconv"
	"strings"

	"github.com/tidbcloud/dataservices/idl/pbgen/proto/dataapi"
	"github.com/tidbcloud/tiinf/errs"
)

const (
	defaultSchemaSearchLimit = 100
	maxSchemaSearchLimit     = 1000
)

func ListDatabases(ctx context.Context, in dto.ListDatabasesInput) ([]*dto.DatabaseItem, error) {
	excludeDBs, err := getExcludeDBsBySettings(ctx, in.CreatedBy, in.ProjectID, in.ClusterID)
	if err != nil {
		return nil, err
	}
	credential := &dataapi.Credential{
		Type:      dataapi.CrtType_AUTH0,
		Key:       in.Auth0Token,
		AccountId: in.CreatedBy,
	}

	notIn := quoteStringList(excludeDBs)
	sql := "select s.schema_name, if(t.cnt is null, 0, 1) as has_table from information_schema.schemata s left join (select table_schema, 1 as cnt from information_schema.tables group by table_schema) t on s.schema_name = t.table_schema where s.schema_name not in (" + notIn + ") order by lower(s.schema_name);"
	resp, bizErr := grpc.ExecSqlDirectly(ctx, &dataapi.BasicEndpointReq{
		ProjectId:  in.ProjectID,
		ClusterId:  in.ClusterID,
		Sql:        sql,
		LimitRows:  uint64(SCHEMALIMIT),
		Credential: credential,
	})
	if bizErr != nil {
		logger.LogWithError(ctx, bizErr, fmt.Sprintf("ERROR: ListDatabases failed! sql:%s, user:%s", sql, in.CreatedBy))
		return nil, bizErr
	}

	dbs := make([]*dto.DatabaseItem, 0)
	for _, r := range resp.Rows {
		if len(r.Items) == 0 || len(r.Items[0]) == 0 {
			continue
		}
		hasTable := false
		if len(r.Items) > 1 && r.Items[1] == "1" {
			hasTable = true
		}
		dbs = append(dbs, &dto.DatabaseItem{Name: r.Items[0], HasTable: hasTable})
	}
	return dbs, nil
}

func ListTables(ctx context.Context, in dto.ListTablesInput) ([]string, error) {
	excludeDBs, err := getExcludeDBsBySettings(ctx, in.CreatedBy, in.ProjectID, in.ClusterID)
	if err != nil {
		return nil, err
	}
	if util.ArrayIn(in.Database, excludeDBs) {
		return nil, errs.NewBadRequest("db_name is not allowed by current settings", nil)
	}
	credential := &dataapi.Credential{
		Type:      dataapi.CrtType_AUTH0,
		Key:       in.Auth0Token,
		AccountId: in.CreatedBy,
	}

	db := escapeSQLString(in.Database)
	sql := "select table_name from information_schema.tables where table_schema='" + db + "' order by lower(table_name);"
	resp, bizErr := grpc.ExecSqlDirectly(ctx, &dataapi.BasicEndpointReq{
		ProjectId:  in.ProjectID,
		ClusterId:  in.ClusterID,
		Database:   in.Database,
		Sql:        sql,
		LimitRows:  uint64(SCHEMALIMIT),
		Credential: credential,
	})
	if bizErr != nil {
		logger.LogWithError(ctx, bizErr, fmt.Sprintf("ERROR: ListTables failed! sql:%s, user:%s", sql, in.CreatedBy))
		return nil, bizErr
	}

	tables := make([]string, 0)
	for _, r := range resp.Rows {
		if len(r.Items) == 0 || len(r.Items[0]) == 0 {
			continue
		}
		tables = append(tables, r.Items[0])
	}
	return tables, nil
}

func ListColumns(ctx context.Context, in dto.ListColumnsInput) ([]*entity.Columns, error) {
	excludeDBs, err := getExcludeDBsBySettings(ctx, in.CreatedBy, in.ProjectID, in.ClusterID)
	if err != nil {
		return nil, err
	}
	if util.ArrayIn(in.Database, excludeDBs) {
		return nil, errs.NewBadRequest("db_name is not allowed by current settings", nil)
	}
	credential := &dataapi.Credential{
		Type:      dataapi.CrtType_AUTH0,
		Key:       in.Auth0Token,
		AccountId: in.CreatedBy,
	}

	db := escapeSQLString(in.Database)
	table := escapeSQLString(in.Table)
	// column_name, data_type, is_nullable, ordinal_position
	sql := "select column_name,data_type,is_nullable from information_schema.columns where table_schema='" + db + "' and table_name='" + table + "' order by ordinal_position;"
	resp, bizErr := grpc.ExecSqlDirectly(ctx, &dataapi.BasicEndpointReq{
		ProjectId:  in.ProjectID,
		ClusterId:  in.ClusterID,
		Database:   in.Database,
		Sql:        sql,
		LimitRows:  uint64(SCHEMALIMIT),
		Credential: credential,
	})
	if bizErr != nil {
		logger.LogWithError(ctx, bizErr, fmt.Sprintf("ERROR: ListColumns failed! sql:%s, user:%s", sql, in.CreatedBy))
		return nil, bizErr
	}

	cols := make([]*entity.Columns, 0)
	for _, r := range resp.Rows {
		if len(r.Items) < 3 {
			continue
		}
		col := &entity.Columns{
			Col:      r.Items[0],
			DataType: r.Items[1],
			Nullable: strings.ToLower(r.Items[2]) == "yes",
		}
		cols = append(cols, col)
	}
	return cols, nil
}

func SearchSchemas(ctx context.Context, in dto.SearchSchemasInput) ([]*dto.SchemaSearchItem, error) {
	excludeDBs, err := getExcludeDBsBySettings(ctx, in.CreatedBy, in.ProjectID, in.ClusterID)
	if err != nil {
		return nil, err
	}
	credential := &dataapi.Credential{
		Type:      dataapi.CrtType_AUTH0,
		Key:       in.Auth0Token,
		AccountId: in.CreatedBy,
	}

	keyword := strings.TrimSpace(in.Keyword)
	if keyword == "" {
		return nil, errs.NewBadRequest("keyword is required", nil)
	}
	if len(keyword) > 128 {
		return nil, errs.NewBadRequest("keyword is too long", nil)
	}

	limit := in.Limit
	if limit <= 0 {
		limit = defaultSchemaSearchLimit
	}
	if limit > maxSchemaSearchLimit {
		limit = maxSchemaSearchLimit
	}

	like := "'%" + escapeSQLString(keyword) + "%'"
	notIn := quoteStringList(excludeDBs)

	var sql string
	switch in.Type {
	case dto.SchemaSearchTypeDatabase:
		sql = "select s.schema_name, if(t.cnt is null, 0, 1) as has_table from information_schema.schemata s left join (select table_schema, 1 as cnt from information_schema.tables group by table_schema) t on s.schema_name = t.table_schema where s.schema_name not in (" + notIn + ") and s.schema_name like " + like + " order by lower(s.schema_name) limit " + strconv.Itoa(limit) + ";"
	case dto.SchemaSearchTypeTable:
		sql = "select table_schema,table_name from information_schema.tables where table_schema not in (" + notIn + ") and table_name like " + like + " order by lower(table_schema),lower(table_name) limit " + strconv.Itoa(limit) + ";"
	case dto.SchemaSearchTypeColumn:
		sql = "select table_schema,table_name,column_name,data_type,is_nullable from information_schema.columns where table_schema not in (" + notIn + ") and column_name like " + like + " order by lower(table_schema),lower(table_name),lower(column_name) limit " + strconv.Itoa(limit) + ";"
	default:
		return nil, errs.NewBadRequest("invalid type, must be one of: database, table, column", nil)
	}

	resp, bizErr := grpc.ExecSqlDirectly(ctx, &dataapi.BasicEndpointReq{
		ProjectId:  in.ProjectID,
		ClusterId:  in.ClusterID,
		Sql:        sql,
		LimitRows:  uint64(limit),
		Credential: credential,
	})
	if bizErr != nil {
		logger.LogWithError(ctx, bizErr, fmt.Sprintf("ERROR: SearchSchemas failed! sql:%s, user:%s", sql, in.CreatedBy))
		return nil, bizErr
	}

	out := make([]*dto.SchemaSearchItem, 0)
	for _, r := range resp.Rows {
		switch in.Type {
		case dto.SchemaSearchTypeDatabase:
			if len(r.Items) < 1 {
				continue
			}
			hasTable := len(r.Items) > 1 && r.Items[1] == "1"
			out = append(out, &dto.SchemaSearchItem{
				Type:     dto.SchemaSearchTypeDatabase,
				Database: r.Items[0],
				HasTable: hasTable,
			})
		case dto.SchemaSearchTypeTable:
			if len(r.Items) < 2 {
				continue
			}
			out = append(out, &dto.SchemaSearchItem{
				Type:     dto.SchemaSearchTypeTable,
				Database: r.Items[0],
				Table:    r.Items[1],
			})
		case dto.SchemaSearchTypeColumn:
			if len(r.Items) < 5 {
				continue
			}
			out = append(out, &dto.SchemaSearchItem{
				Type:     dto.SchemaSearchTypeColumn,
				Database: r.Items[0],
				Table:    r.Items[1],
				Column:   r.Items[2],
				DataType: r.Items[3],
				Nullable: strings.ToLower(r.Items[4]) == "yes",
			})
		}
	}
	return out, nil
}

func getExcludeDBsBySettings(ctx context.Context, createdBy string, projectID, clusterID uint64) ([]string, error) {
	settings, err := ShowSettings(ctx, entity.SQLEditorSettings{
		CreatedBy: createdBy,
		ClusterID: clusterID,
		ProjectID: projectID,
	})
	if err != nil {
		logger.LogWithError(ctx, err, fmt.Sprintf("ERROR: get user settings failed! user:%s", createdBy))
		return nil, errs.NewInternalGeneralBusiness("get user settings failed", err)
	}
	if settings.ShowSystemDbSchema {
		return EXCLUDEDB, nil
	}
	return EXCLUDEDBANDSYS, nil
}

func escapeSQLString(s string) string {
	return strings.ReplaceAll(s, "'", "''")
}

func quoteStringList(items []string) string {
	if len(items) == 0 {
		return "''"
	}
	var b strings.Builder
	for i, it := range items {
		if i > 0 {
			b.WriteByte(',')
		}
		b.WriteByte('\'')
		b.WriteString(escapeSQLString(it))
		b.WriteByte('\'')
	}
	return b.String()
}