package export

import (
	"context"
	"fmt"
	"strings"

	"github.com/pingcap/tidb/pkg/parser"
	"github.com/pingcap/tidb/pkg/parser/ast"
	_ "github.com/pingcap/tidb/pkg/parser/test_driver"
	tablefilter "github.com/pingcap/tidb/pkg/util/table-filter"

	globalcommonv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/common/v1"
	globalngv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/nextgen/v1"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/service/cloudstorage"
	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
)

const tsoPhysicalShiftBits = 18

func (s *serviceImpl) validateExportId(ctx context.Context, clusterId, exportId string) bizerr.BizError {
	if strings.TrimSpace(exportId) != exportId {
		return bizerr.New(bizerr.ErrCodeParams, "export_id must not contain leading/trailing whitespace")
	}
	if exportId == "" {
		return bizerr.New(bizerr.ErrCodeParams, "export_id must not be empty")
	}
	if len(exportId) > 64 {
		return bizerr.New(bizerr.ErrCodeParams, "export_id must be at most 64 characters")
	}
	for i := 0; i < len(exportId); i++ {
		c := exportId[i]
		if (c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z') || (c >= '0' && c <= '9') || c == '-' || c == '_' {
			continue
		}
		return bizerr.New(bizerr.ErrCodeParams, "export_id contains invalid character %q", c)
	}

	// Check if export_id already exists
	_, err := s.exportRepo.GetByClusterAndExportId(ctx, clusterId, exportId)
	if err == nil {
		// Export already exists
		return bizerr.New(bizerr.ErrCodeRecordExist, "export_id %q already exists in cluster %s", exportId, clusterId)
	}
	// If error is not "not found", return the error
	if err.ErrCode() != bizerr.ErrCodeRecordNotFound {
		return bizerr.WrapError(err, "failed to check export_id existence")
	}
	// export_id is available
	return nil
}

func (s *serviceImpl) validateCreateExport(
	ctx context.Context,
	req *globalngv1.CreateExportRequest,
	cloudProvider globalcommonv1.CloudProvider_Name,
) bizerr.BizError {
	// validate export_id if provided
	if req.GetExport().GetExportId() != "" {
		if err := s.validateExportId(ctx, req.GetClusterId(), req.GetExport().GetExportId()); err != nil {
			return err
		}
	}
	if req.GetExport().GetSnapshotTso() != 0 {
		if err := validateSnapshotTSO(req.GetExport().GetSnapshotTso()); err != nil {
			return err
		}
	}
	if err := validateExportParams(ctx, req.Export, cloudProvider); err != nil {
		return err
	}
	if err := s.precheckExport(ctx, req); err != nil {
		return err
	}
	return nil
}

func validateSnapshotTSO(snapshotTSO uint64) bizerr.BizError {
	if snapshotTSO == 0 {
		return bizerr.New(bizerr.ErrCodeParams, "snapshot_tso must be greater than 0 when specified")
	}
	if snapshotTSO>>tsoPhysicalShiftBits == 0 {
		return bizerr.New(bizerr.ErrCodeParams, "snapshot_tso has an invalid physical timestamp")
	}
	return nil
}

func validateExportParams(ctx context.Context, exp *globalngv1.Export, cloudProvider globalcommonv1.CloudProvider_Name) bizerr.BizError {
	if exp.GetUsername() == "" {
		return bizerr.New(bizerr.ErrCodeParams, "export.username is required")
	}
	if exp.GetPassword() == "" {
		return bizerr.New(bizerr.ErrCodeParams, "export.password is required")
	}
	// validate cloud storage
	if err := cloudstorage.Validate(exp.GetCloudStorage(), cloudProvider); err != nil {
		return err
	}
	if exp.GetCloudStorage().GetType() == globalngv1.CloudStorageType_OSS {
		if exp.GetCloudStorage().GetOss().GetAuthType() == globalngv1.OSSCloudStorage_ROLE_ARN {
			return bizerr.New(bizerr.ErrCodeParams, "oss role_arn is not supported yet")
		}
	}
	if exp.GetCloudStorage().GetType() == globalngv1.CloudStorageType_GCS {
		return bizerr.New(bizerr.ErrCodeParams, "export to GCS is not supported yet")
	}
	if cloudProvider == globalcommonv1.CloudProvider_AZURE {
		return bizerr.New(bizerr.ErrCodeParams, "export is not supported for Azure cloud provider yet")
	}
	if err := validateExportOptions(ctx, exp.GetExportOptions()); err != nil {
		return err
	}
	return nil
}

// checkExportOptions validates ExportOptions per design: file_type vs filter, sql vs table mutually exclusive.
// Returns error when filter is set but its content is empty (oneof not set, sql empty, or table/patterns empty).
func validateExportOptions(_ context.Context, opts *globalngv1.ExportOptions) bizerr.BizError {
	if opts == nil {
		return bizerr.New(bizerr.ErrCodeParams, "export.export_options is required")
	}
	if opts.GetFilter() == nil || opts.GetFilter().GetType() == nil {
		return bizerr.New(bizerr.ErrCodeParams, "export_options.filter is required")
	}
	if opts.GetFileType() == globalngv1.ExportOptions_FILE_TYPE_UNSPECIFIED {
		return bizerr.New(bizerr.ErrCodeParams, "unsupported file_type")
	}
	if opts.GetCompression() == globalngv1.ExportOptions_COMPRESSION_TYPE_UNSPECIFIED {
		return bizerr.New(bizerr.ErrCodeParams, "unsupported compression")
	}
	if err := validateCSVDialect(opts.GetCsvFormat()); err != nil {
		return err
	}
	if err := validateParquetOptions(opts); err != nil {
		return err
	}

	// check filter
	switch f := opts.GetFilter().GetType().(type) {
	case *globalngv1.ExportOptions_Filter_Sql:
		if f.Sql == "" {
			return bizerr.New(bizerr.ErrCodeParams, "export_options.filter.sql is required when filter type is sql")
		}
		_, err := parseSQL(f.Sql)
		if err != nil {
			return bizerr.New(bizerr.ErrCodeParams, "Invalid SQL error: %s", err.Error())
		}
		if opts.GetFileType() == globalngv1.ExportOptions_SQL {
			return bizerr.New(bizerr.ErrCodeParams, "does not support exporting to sql file when filter with SQL statement")
		}
		// forbid SQL filter as it does not need at current, delete if want to support in the future.
		return bizerr.New(bizerr.ErrCodeParams, "sql filter is not supported yet")
	case *globalngv1.ExportOptions_Filter_Table_:
		if f.Table == nil {
			return bizerr.New(bizerr.ErrCodeParams, "export_options.filter.table is required when filter type is table")
		}
		if len(f.Table.GetPatterns()) == 0 {
			return bizerr.New(bizerr.ErrCodeParams, "export_options.filter.table.patterns is required and cannot be empty")
		}
		for _, pattern := range f.Table.GetPatterns() {
			if err := parseTableFilter(pattern); err != nil {
				return bizerr.New(
					bizerr.ErrCodeParams,
					"Invalid filter patterns %s: %s. See https://docs.pingcap.com/tidb/stable/table-filter for more details.",
					pattern,
					err.Error(),
				)
			}
		}
		if f.Table.GetWhere() != "" {
			sql := fmt.Sprintf("SELECT * FROM test.t WHERE %s", f.Table.GetWhere())
			_, err := parseSQL(sql)
			if err != nil {
				return bizerr.New(bizerr.ErrCodeParams, "Invalid WHERE clause: %s", err.Error())
			}
		}
		for _, partition := range f.Table.GetPartitions() {
			if strings.TrimSpace(partition) == "" {
				return bizerr.New(bizerr.ErrCodeParams, "export_options.filter.table.partitions cannot contain an empty name")
			}
		}
	default:
		return bizerr.New(bizerr.ErrCodeParams, "unsupported filter type")
	}
	return nil
}

func validateParquetOptions(opts *globalngv1.ExportOptions) bizerr.BizError {
	if opts.GetFileType() != globalngv1.ExportOptions_PARQUET {
		if opts.GetParquetFormat() != nil {
			return bizerr.New(bizerr.ErrCodeParams, "parquet_format is only supported for parquet exports")
		}
		return nil
	}

	// Dumpling rejects the general --compress option for Parquet and requires
	// --parquet-compress instead.
	if opts.GetCompression() != globalngv1.ExportOptions_NONE {
		return bizerr.New(bizerr.ErrCodeParams, "parquet exports require compression NONE; use parquet_format.compression")
	}

	if opts.GetParquetFormat() == nil {
		return nil
	}
	switch opts.GetParquetFormat().GetCompression() {
	case globalngv1.ExportParquetCompressionType_PARQUET_COMPRESSION_TYPE_UNSPECIFIED,
		globalngv1.ExportParquetCompressionType_GZIP,
		globalngv1.ExportParquetCompressionType_SNAPPY,
		globalngv1.ExportParquetCompressionType_ZSTD,
		globalngv1.ExportParquetCompressionType_NONE:
		return nil
	default:
		return bizerr.New(bizerr.ErrCodeParams, "unsupported parquet compression")
	}
}

func validateCSVDialect(csvFormat *globalngv1.ExportOptions_CSVFormat) bizerr.BizError {
	if csvFormat == nil || csvFormat.Dialect == nil {
		return nil
	}
	switch csvFormat.GetDialect() {
	case globalngv1.ExportOptions_ExportCSVDialect_DEFAULT,
		globalngv1.ExportOptions_ExportCSVDialect_SNOWFLAKE,
		globalngv1.ExportOptions_ExportCSVDialect_REDSHIFT,
		globalngv1.ExportOptions_ExportCSVDialect_BIGQUERY:
		return nil
	default:
		return bizerr.New(bizerr.ErrCodeParams, "unsupported csv dialect")
	}
}

func parseTableFilter(filter string) error {
	arg := strings.TrimSpace(filter)
	_, err := tablefilter.Parse([]string{arg})
	if err != nil {
		return err
	}
	return nil
}

func parseSQL(sql string) (ast.StmtNode, error) {
	return parser.New().ParseOneStmt(sql, "", "")
}

func (s *serviceImpl) precheckExport(ctx context.Context, req *globalngv1.CreateExportRequest) bizerr.BizError {
	return nil
	// check cloud storage
}