package grpc

import (
	"bytes"
	"context"
	"database/sql"
	"fmt"
	"strings"
	"time"

	"github.com/pingcap/tidb/pkg/util/dbutil"
	"github.com/tidbcloud/dataflow-service/dataflow-agent-ng/idl/pbgen/proto/ng_agent"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/common"
)

const (
	dbTimeout = 5 * time.Second

	DefaultSQLFilePatternWithTargetTable            = `(?i)^%s.%s(?:(?:\.\d+)|(?:\.?(?:_\d+){3}))?\.sql(?:\.(\w+))?$`
	DefaultCSVFilePatternWithTargetTable            = `(?i)^%s.%s(?:(?:\.\d+)|(?:\.?(?:_\d+){3}))?\.csv(?:\.(\w+))?$`
	DefaultRawParquetFilePatternWithTargetTable     = `(?i)^%s.%s(?:(?:\.\d+)|(?:\.?(?:_\d+){3}))?(?:\.(?:snappy|gz|zst|zstd|gzip))?\.parquet$`
	DefaultAuroraSnapshotFilePatternWithTargetTable = `(?i)^%s/%s.%s/(?:[^/]*/)*(?:[a-z0-9\-_.]+\.parquet)$`
	DefaultDBSchemaFilePattern                      = `(?i)^(?:[^/]*/)*([^/.]+)-schema-create\.sql(?:\.(\w+))?$`
	DefaultTableSchemaFilePattern                   = `(?i)^(?:[^/]*/)*([^/.]+)\.(.*?)-schema\.sql(?:\.(\w+))?$`
	DefaultViewSchemaFilePattern                    = `(?i)^(?:[^/]*/)*([^/.]+)\.(.*?)-schema-view\.sql(?:\.(\w+))?$`
	DefaultFilePattern                              = `(?i)^(?:[^/]*/)*([^/.]+)\.([^/.]+?)(?:(?:\.\d+)|(?:\.?(?:_\d+){3}))?\.%s(?:\.(\w+))?$`
	DefaultAuroraSnapshotFilePattern                = `(?i)^(?:[^/]*/)*([a-z0-9_]+)\.([a-z0-9_]+)/(?:[^/]*/)*(?:[a-z0-9\-_.]+\.parquet)$`
	DefaultRawParquetFilePattern                    = `(?i)^(?:[^/]*/)*([^/.]+)\.([^/.]+?)(?:(?:\.\d+)|(?:\.?(?:_\d+){3}))?(?:\.(?:snappy|gz|zst|zstd|gzip))?\.parquet$`
)

func GetDBVersion(ctx context.Context, db *sql.DB) (string, error) {
	ctx2, cancel := context.WithTimeout(ctx, dbTimeout)
	defer cancel()
	return dbutil.GetDBVersion(ctx2, db)
}

func GetSchemas(ctx context.Context, db *sql.DB) ([]string, error) {
	dbCtx, cancel := context.WithTimeout(ctx, dbTimeout)
	defer cancel()
	return dbutil.GetSchemas(dbCtx, db)
}

func GetTables(ctx context.Context, db *sql.DB, schemaName string) ([]string, error) {
	dbCtx, cancel := context.WithTimeout(ctx, dbTimeout)
	defer cancel()
	return dbutil.GetTables(dbCtx, db, schemaName)
}

func GetAllTables(ctx context.Context, db *sql.DB, shouldIngore func(string) bool) ([]*agent.Schema, error) {
	schemas, err := GetSchemas(ctx, db)
	if err != nil {
		return nil, err
	}
	agentSchemas := make([]*agent.Schema, 0, len(schemas))
	for _, schema := range schemas {
		if shouldIngore(strings.ToLower(schema)) {
			continue
		}
		tables, err := GetTables(ctx, db, schema)
		if err != nil {
			return nil, err
		}
		currentSchemaInfo := &agent.Schema{
			SchemaName: schema,
			Tables:     make([]*agent.Table, len(tables)),
		}
		for j, table := range tables {
			currentSchemaInfo.Tables[j] = &agent.Table{
				SchemaName: schema,
				TableName:  table,
			}
		}
		agentSchemas = append(agentSchemas, currentSchemaInfo)
	}
	return agentSchemas, nil
}

func generateLoadDataSql(
	sourceUrl, targetDb, targetTable string,
	terminator, separaor, delimiter string,
	backslashEscape, ignoreOneLine bool,
) string {
	var buf bytes.Buffer
	buf.WriteString(fmt.Sprintf("LOAD DATA INFILE '%s' INTO TABLE `%s`.`%s`", sourceUrl, targetDb, targetTable))

	if len(separaor) != 0 {
		buf.WriteString(fmt.Sprintf(" FIELDS TERMINATED BY '%s'", separaor))
	}

	if len(delimiter) != 0 {
		buf.WriteString(fmt.Sprintf(" ENCLOSED BY '%s'", delimiter))
	}

	if backslashEscape {
		buf.WriteString(" ESCAPED BY '\\\\'")
	}

	if len(terminator) != 0 {
		buf.WriteString(fmt.Sprintf(" LINES TERMINATED BY '%s'", terminator))
	}

	if ignoreOneLine {
		buf.WriteString(" IGNORE 1 LINES")
	}

	buf.WriteString(";")
	return buf.String()
}

func generateDropTableSql(tables []*common.Table) string {
	tbls := make([]string, 0, len(tables))
	for _, tbl := range tables {
		tbls = append(tbls, fmt.Sprintf("`%s`.`%s`", tbl.Schema, tbl.Table))
	}
	return fmt.Sprintf("DROP TABLE %s;", strings.Join(tbls, ","))
}