package pimport

import (
	"context"
	"fmt"

	"github.com/pingcap/errors"
	"github.com/tidbcloud/dataflow-service/internal/db/entity"
	"github.com/tidbcloud/dataflow-service/internal/grpc/grpcli"
	"github.com/tidbcloud/pkgv2/log"
	"go.uber.org/zap"

	agentpb "github.com/tidbcloud/dataflow-service/idl/pbgen/proto/agent"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
)

type Validator interface {
	Validate(ctx context.Context) error
	GetCheckItem() dataflow.ImportValidationItem
}

type SourceAccessCheckFailedError struct {
	AWSRoleARN         string
	SourceURL          string
	AWSAccessKeyID     string
	AWSSecretAccessKey string
	ErrorMsg           string
	DataSourceType     entity.DataSource
	AzblobSASToken     string
}

func (e *SourceAccessCheckFailedError) Error() string {
	return fmt.Sprintf("source access check failed: %s\n", e.ErrorMsg)
}

type sourceAccessValidator struct {
	importEntity *ImportTask
	importModel  Model
}

func (v *sourceAccessValidator) Validate(ctx context.Context) error {
	logger := log.WithContextL(ctx)
	// Note: the SK is not encrypted when doing validation here.
	// Only persisted import entity will encrypt the SK
	plainSourceURL, accessConfig, err := v.importEntity.ParseSourceURL()
	if err != nil {
		logger.Error("cannot get access key info from entity's source URL", zap.Error(err))
		accessConfig = new(entity.ImportSourceAccessConfig)
	}
	logger.Info("source access validator", zap.String("sourceURL", plainSourceURL), zap.Any("accessConfig", accessConfig))
	checkResultErrMsg, dataSourceType, err := v.importModel.CheckDataSourceAccess(
		ctx,
		uint64(v.importEntity.ClusterID),
		v.importEntity.AwsRoleArn,
		plainSourceURL,
		accessConfig.AWSAccessKeyID,
		accessConfig.AWSSecretAccessKey,
		accessConfig.AzblobSASToken,
	)
	if err != nil {
		return errors.Annotate(err, "check data source access from agent error")
	}
	if len(checkResultErrMsg) > 0 {
		return &SourceAccessCheckFailedError{
			AWSRoleARN:         v.importEntity.AwsRoleArn,
			SourceURL:          plainSourceURL,
			AWSAccessKeyID:     accessConfig.AWSAccessKeyID,
			AWSSecretAccessKey: accessConfig.AWSSecretAccessKey,
			ErrorMsg:           checkResultErrMsg,
			DataSourceType:     dataSourceType,
			AzblobSASToken:     accessConfig.AzblobSASToken,
		}
	}
	return nil
}

func (v *sourceAccessValidator) GetCheckItem() dataflow.ImportValidationItem {
	return dataflow.ImportValidationItem_SOURCE_ACCESS_CHECK
}

// deprecated
type targetAccessValidator struct{}

func (v *targetAccessValidator) Validate(ctx context.Context) error {
	return nil
}

func (v *targetAccessValidator) GetCheckItem() dataflow.ImportValidationItem {
	return dataflow.ImportValidationItem_TARGET_ACCESS_CHECK
}

type filterAndRuleValidator struct{}

func (v *filterAndRuleValidator) Validate(ctx context.Context) error {
	return nil
}

func (v *filterAndRuleValidator) GetCheckItem() dataflow.ImportValidationItem {
	return dataflow.ImportValidationItem_FILTER_AND_RULE_CHECK
}

type LightningPreCheckFailedError struct {
	CheckResult *agentpb.CheckResult
}

func (e *LightningPreCheckFailedError) Error() string {
	return fmt.Sprintf("Lightning precheck on item %s not passed: %s",
		e.CheckResult.CheckItem.String(),
		e.CheckResult.Msg,
	)
}

type importPrecheckValidator struct {
	importEntity *ImportTask
	importModel  Model
}

func (v *importPrecheckValidator) Validate(ctx context.Context) error {
	var err error
	cluster, err := grpcli.GetClusterByID(ctx, uint64(v.importEntity.ClusterID))
	if err != nil {
		return errors.Annotate(err, "get cluster information by ID error")
	}
	checkResults, err := v.importModel.doLightningPrecheck(
		ctx,
		v.importEntity,
		cluster.InfraProviderType,
	)
	if err != nil {
		return errors.Annotate(err, "do lightning precheck returns error")
	}

	for _, result := range checkResults {
		if !result.HasPassed {
			return &LightningPreCheckFailedError{
				CheckResult: result,
			}
		}
	}
	return nil
}

func (v *importPrecheckValidator) GetCheckItem() dataflow.ImportValidationItem {
	return dataflow.ImportValidationItem_IMPORT_PRECHECK
}

type importEstimateSizeValidator struct {
	importEntity *ImportTask
	importModel  Model
}

func (v *importEstimateSizeValidator) Validate(ctx context.Context) error {
	return nil
}

func (v *importEstimateSizeValidator) GetCheckItem() dataflow.ImportValidationItem {
	return dataflow.ImportValidationItem_ESTIMATE_SIZE
}

type importDataOnlyValidator struct {
	importEntity *ImportTask
	importModel  Model
}

func (v *importDataOnlyValidator) Validate(ctx context.Context) error {
	const maxCheckTables = 15
	if len(v.importEntity.TargetTableInfos.Infos) >= maxCheckTables {
		return nil
	}

	return v.importModel.PrecheckImportFile(ctx, v.importEntity)
}

func (v *importDataOnlyValidator) GetCheckItem() dataflow.ImportValidationItem {
	return dataflow.ImportValidationItem_IMPORT_DATA_ONLY_CHECK
}

type validatorBuilder struct {
	importEntity *ImportTask
	importModel  Model
}

func (c Model) newValidatorBuilder(imp *ImportTask) (*validatorBuilder, error) {
	return &validatorBuilder{
		importEntity: imp,
		importModel:  c,
	}, nil
}

func (b *validatorBuilder) Build(item dataflow.ImportValidationItem) (Validator, error) {
	switch item {
	case dataflow.ImportValidationItem_SOURCE_ACCESS_CHECK:
		return &sourceAccessValidator{
			importEntity: b.importEntity,
			importModel:  b.importModel,
		}, nil
	case dataflow.ImportValidationItem_TARGET_ACCESS_CHECK:
		return &targetAccessValidator{}, nil
	case dataflow.ImportValidationItem_FILTER_AND_RULE_CHECK:
		return &filterAndRuleValidator{}, nil
	case dataflow.ImportValidationItem_IMPORT_PRECHECK:
		return &importPrecheckValidator{
			importEntity: b.importEntity,
			importModel:  b.importModel,
		}, nil
	case dataflow.ImportValidationItem_ESTIMATE_SIZE:
		return &importEstimateSizeValidator{
			importEntity: b.importEntity,
			importModel:  b.importModel,
		}, nil
	case dataflow.ImportValidationItem_IMPORT_DATA_ONLY_CHECK:
		return &importDataOnlyValidator{
			importEntity: b.importEntity,
			importModel:  b.importModel,
		}, nil
	default:
		return nil, errors.Errorf("unsupported check item: %s", item)
	}
}