package pimport

import (
	"fmt"
	"github.com/tidbcloud/dataflow-service/internal/openapi/service"
	"net/url"
	"strconv"
	"strings"

	"github.com/pingcap/errcode"
	"github.com/pingcap/errors"
	"github.com/tidbcloud/dataflow-service/internal/db/entity"
	"github.com/tidbcloud/dataflow-service/internal/utils"
	"github.com/tidbcloud/idl/pbgen/proto/openapi/dp"
	"go.uber.org/zap"
)

type ImportSourceScheme string

const (
	S3Scheme         ImportSourceScheme = "s3"
	GCSScheme        ImportSourceScheme = "gs"
	UploadStubScheme ImportSourceScheme = "file"
)

var gMapSourceTypeToScheme = map[dp.ImportSource_ImportSourceType]ImportSourceScheme{
	dp.ImportSource_S3:         S3Scheme,
	dp.ImportSource_GCS:        GCSScheme,
	dp.ImportSource_LOCAL_FILE: UploadStubScheme,
}

func checkListImportTasks(userContext *service.UserContext, req *dp.ListImportTasksReq, logger *zap.Logger) error {
	if err := userContext.CheckProjectID(entity.ID(req.GetProjectId())); err != nil {
		logger.Error("check project ID error", zap.Error(err))
		return errcode.NewInvalidInputErr(errors.Errorf(
			"The project ID %v is invalid for the user to list the import tasks. Maybe the project doesn't belong to the user. Please specify the right project ID", req.GetProjectId()),
		)
	}
	return nil
}

func checkGetImportTask(userContext *service.UserContext, req *dp.GetImportTaskReq, logger *zap.Logger) error {
	if err := userContext.CheckProjectID(entity.ID(req.GetProjectId())); err != nil {
		logger.Error("check project ID error", zap.Error(err))
		return errcode.NewInvalidInputErr(errors.Errorf(
			"The project ID %v is invalid for the user to get the import task. Maybe the project doesn't belong to the user. Please specify the right project ID", req.GetProjectId()),
		)
	}
	return nil
}

func checkGetImportTaskRoleInfo(userContext *service.UserContext, req *dp.GetImportTaskRoleInfoReq, logger *zap.Logger) error {
	if err := userContext.CheckProjectID(entity.ID(req.GetProjectId())); err != nil {
		logger.Error("check project ID error", zap.Error(err))
		return errcode.NewInvalidInputErr(errors.Errorf(
			"The project ID %v is invalid for the user to get the import role information. Maybe the project doesn't belong to the user. Please specify the right project ID", req.GetProjectId()),
		)
	}
	return nil
}

func checkCreateImportTask(userContext *service.UserContext, req *dp.CreateImportTaskReq, logger *zap.Logger) (*parsedImportSpecInfo, error) {
	if err := userContext.CheckProjectID(entity.ID(req.GetProjectId())); err != nil {
		logger.Error("check project ID error", zap.Error(err))
		return nil, errcode.NewInvalidInputErr(errors.Errorf(
			"The project ID %v is invalid for the user to create the import task. Maybe the project doesn't belong to the user. Please specify the right project ID", req.GetProjectId()),
		)
	}
	parsedSpecInfo, err := checkImportSpecForRequests(req.GetSpec(), logger)
	if err != nil {
		logger.Error("check import spec for requests failed", zap.Error(err))
		return nil, utils.WrapAsErrorCodeErr(err, "check import spec field failed")
	}
	// TODO: extra arg check: check param limitations
	return parsedSpecInfo, nil
}

func checkUpdateImportTask(userContext *service.UserContext, req *dp.UpdateImportTaskReq, logger *zap.Logger) error {
	if err := userContext.CheckProjectID(entity.ID(req.GetProjectId())); err != nil {
		logger.Error("check project ID error", zap.Error(err))
		return errcode.NewInvalidInputErr(errors.Errorf(
			"The project ID %v is invalid for the user to update the import task. Maybe the project doesn't belong to the user. Please specify the right project ID", req.GetProjectId()),
		)
	}
	phase := req.GetAction()
	if phase != dp.UpdateImportTaskReq_CANCEL {
		return errcode.NewInvalidInputErr(errors.Errorf("invalid action: %v . Currently, only CANCEL is supported", phase))
	}
	return nil
}

func checkUploadLocalFile(userContext *service.UserContext, req *dp.UploadLocalFileReq, logger *zap.Logger) error {
	if err := userContext.CheckProjectID(entity.ID(req.GetProjectId())); err != nil {
		logger.Error("check project ID error", zap.Error(err))
		return errcode.NewInvalidInputErr(errors.Errorf(
			"The project ID %v is invalid for the user to upload the local file data. Maybe the project doesn't belong to the user. Please specify the right project ID", req.GetProjectId()),
		)
	}
	return nil
}

func checkPreviewImportData(userContext *service.UserContext, req *dp.PreviewImportDataReq, logger *zap.Logger) (*parsedImportSpecInfo, error) {
	if err := userContext.CheckProjectID(entity.ID(req.GetProjectId())); err != nil {
		logger.Error("check project ID error", zap.Error(err))
		return nil, errcode.NewInvalidInputErr(errors.Errorf(
			"The project ID %v is invalid for the user to do a data preview. Maybe the project doesn't belong to the user. Please specify the right project ID", req.GetProjectId()),
		)
	}
	limitRowsCount := req.GetLimitRowsCount()
	const maxPreviewRowsCount = 20
	if limitRowsCount <= 0 {
		limitRowsCount = 10
	}
	if limitRowsCount > maxPreviewRowsCount {
		limitRowsCount = maxPreviewRowsCount
	}
	req.LimitRowsCount = &limitRowsCount

	parsedSpecInfo, err := checkImportSpecForRequests(req.GetSpec(), logger)
	if err != nil {
		logger.Error("check import spec for requests failed", zap.Error(err))
		return nil, utils.WrapAsErrorCodeErr(err, "check import spec field failed")
	}
	switch req.GetSpec().GetSource().GetType() {
	case dp.ImportSource_S3, dp.ImportSource_GCS:
		return nil, errcode.NewInvalidInputErr(errors.Errorf("The source type %v is not supported for preview", req.GetSpec().GetSource().GetType()))
	case dp.ImportSource_UNSPECIFIED:
		return nil, errcode.NewInvalidInputErr(errors.New("The source type is not specified"))
	}
	return parsedSpecInfo, nil
}

type parsedImportSpecInfo struct {
	UploadStubID   uint64
	TargetTableMap map[string]struct{}
}

func getFullTableName(dbName string, tblName string) string {
	return fmt.Sprintf("%s.%s",
		strings.ToLower(strings.TrimSpace(dbName)),
		strings.ToLower(strings.TrimSpace(tblName)),
	)
}

func checkImportSpecForRequests(spec *dp.ImportSpec, logger *zap.Logger) (*parsedImportSpecInfo, error) {
	result := &parsedImportSpecInfo{}
	formatType := spec.GetSource().GetFormat().GetType()
	switch formatType {
	case dp.ImportSourceFormat_UNSPECIFIED:
		return nil, errcode.NewInvalidInputErr(errors.Errorf("Source format type %v is not supported", formatType))
	}
	sourceURI := spec.GetSource().GetUri()
	parsedSourceURI, err := url.Parse(sourceURI)
	if err != nil {
		logger.Error("cannot parse the source URI", zap.Error(err))
		return nil, errcode.NewInvalidInputErr(errors.New("Import source URL is invalid"))
	}
	sourceType := spec.GetSource().GetType()
	switch sourceType {
	case dp.ImportSource_S3, dp.ImportSource_GCS:
		if parsedSourceURI.Scheme != string(gMapSourceTypeToScheme[sourceType]) {
			return nil, errcode.NewInvalidInputErr(errors.New("Import source URL's scheme doesn't match the source type"))
		}
	case dp.ImportSource_LOCAL_FILE:
		if formatType != dp.ImportSourceFormat_CSV {
			return nil, errcode.NewInvalidInputErr(errors.New("Currently, only CSV format is supported for local file import"))
		}
		if parsedSourceURI.Scheme != string(gMapSourceTypeToScheme[sourceType]) {
			return nil, errcode.NewInvalidInputErr(errors.New("Import source URL's scheme doesn't match the source type"))
		}
		stubID, err := strconv.ParseInt(parsedSourceURI.Host, 10, 0)
		if err != nil {
			return nil, errcode.NewInvalidInputErr(errors.New("Import source URL's path is not a stub ID"))
		}
		if stubID <= 0 {
			return nil, errcode.NewInvalidInputErr(errors.New("Import source stub ID is invalid"))
		}
		result.UploadStubID = uint64(stubID)
		if len(spec.GetTarget().GetTables()) != 1 {
			return nil, errcode.NewInvalidInputErr(errors.New("Only one target table is supported when the source type is `LOCAL_FILE`"))
		}
	case dp.ImportSource_UNSPECIFIED:
		return nil, errcode.NewInvalidInputErr(errors.Errorf("Source type %v is not supported to create", sourceType))
	}
	targetTableMap := make(map[string]struct{})
	for _, tbl := range spec.GetTarget().GetTables() {
		targetTableMap[getFullTableName(tbl.GetDatabaseName(), tbl.GetTableName())] = struct{}{}
	}
	result.TargetTableMap = targetTableMap
	return result, nil
}