package impl

import (
	"context"
	"fmt"
	"strings"
	"time"

	"github.com/hashicorp/go-multierror"
	ffentity "github.com/shiningrush/fastflow/pkg/entity"
	"github.com/shiningrush/fastflow/pkg/mod"
	"github.com/tidbcloud/dataflow-service/app/internal/aws_facade"
	"github.com/tidbcloud/dataflow-service/app/internal/gcp_facade"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	"github.com/tidbcloud/dataflow-service/app/repo"
	backup_repo "github.com/tidbcloud/dataflow-service/app/repo/backup"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/dataflow-service/app/workflow"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	"github.com/tidbcloud/dataflow-service/internal/constants"
	"github.com/tidbcloud/dataflow-service/pkg/account/accountdb"
	bizerr_wrapper "github.com/tidbcloud/dataflow-service/pkg/bizerr-wrapper"
	"github.com/tidbcloud/dataflow-service/pkg/log"
	"github.com/tidbcloud/pkgv2/bizerr"
	"go.uber.org/zap"
)

type BackupExportModelImpl struct {
	backupRepo            backup_repo.BackupRepo
	bucketRepo            repo.BucketRepo
	backupExportRepo      backup_repo.BackupExportRepo
	storageTransferClient gcp_facade.GCPStorageTransferClient
	batchOpsClient        aws_facade.BatchOpsClient
	workflowClient        workflow.Workflow
}

func NewBackupExportModel(
	backupRepo backup_repo.BackupRepo,
	bucketRepo repo.BucketRepo,
	backupExportRepo backup_repo.BackupExportRepo,
	storageTransferClient gcp_facade.GCPStorageTransferClient,
	batchOpsClient aws_facade.BatchOpsClient,
	workflowClient workflow.Workflow,
) *BackupExportModelImpl {
	return &BackupExportModelImpl{
		backupRepo:            backupRepo,
		bucketRepo:            bucketRepo,
		backupExportRepo:      backupExportRepo,
		storageTransferClient: storageTransferClient,
		batchOpsClient:        batchOpsClient,
		workflowClient:        workflowClient,
	}
}

func (m *BackupExportModelImpl) GetBackupExportProfile(
	ctx context.Context, req *domain.GetBackupExportProfileReq,
) (*domain.GetBackupExportProfileResp, bizerr.BizError) {
	logger := log.GetLogger(ctx)
	backup, err := m.backupRepo.Load(ctx, req.BackupID)
	if err != nil {
		return nil, err
	}

	bucket, err := m.bucketRepo.Load(ctx, backup.BucketID.Uint64())
	if err != nil {
		return nil, err
	}

	resp := &domain.GetBackupExportProfileResp{}

	// Only get service account for GCP backups
	if bucket.Provider == constants.CloudProviderGCP {
		serviceAccount, err := m.storageTransferClient.GetServiceAccount(ctx, *bucket.CloudAccountID)
		if err != nil {
			logger.Error("failed to get the GCP service account of Storage Transfer Service", zap.Error(err))
			return nil, bizerr_wrapper.NewInternalErr(
				"failed to get the GCP service account of Storage Transfer Service for %s", *bucket.CloudAccountID,
			)
		}
		resp.GCPServiceAccount = serviceAccount
	}
	if bucket.Provider == constants.CloudProviderAWS {
		if bucket.CloudAccountID == nil || *bucket.CloudAccountID == "" {
			logger.Error("got empty cloud account from backup bucket", zap.Uint64("BackupID", req.BackupID))
			return nil, bizerr_wrapper.NewInternalErr(
				"got empty cloud account from backup bucket, backupId: %d", req.BackupID,
			)
		}

		accountId := *bucket.CloudAccountID
		projectId := backup.ProjectID.Uint64()
		clusterId := backup.ClusterID.Uint64()

		// cluster-specific role used for batchOps-export, such as:
		// arn:aws:iam::385595570414:role/batchops-role-3200560-10371783398525555697
		resp.S3RoleArn = m.batchOpsClient.GenerateServiceRole(ctx, accountId, projectId, clusterId)
	}

	return resp, nil
}

//nolint:funlen
func (m *BackupExportModelImpl) CreateBackupExport(
	ctx context.Context,
	req *dataflow.CreateBackupExportReq,
	user *accountdb.User,
) (uint64, bizerr.BizError) {
	logger := log.GetLogger(ctx)
	backup, err := m.backupRepo.Load(ctx, req.BackupId)
	if err != nil {
		return 0, err
	}
	bucket, err := m.bucketRepo.Load(ctx, backup.BucketID.Uint64())
	if err != nil {
		return 0, err
	}
	exportName := req.BackupExport.Name
	if exportName == "" {
		exportName = fmt.Sprintf("Export-%d-%s", backup.ID, time.Now().Format("20060102150405"))
	}

	targetEntity := entity.BackupExportTarget{}
	switch bucket.Provider {
	case constants.CloudProviderGCP:
		targetEntity.TargetURI = req.BackupExport.GetGcs().GetTargetUri()
	case constants.CloudProviderAWS:
		targetEntity.DryRun = req.BackupExport.GetS3().GetDryRun()
		targetEntity.Region = req.BackupExport.GetS3().GetRegion()
		targetEntity.TargetURI = req.BackupExport.GetS3().GetTargetUri()
	case constants.CloudProviderAzure:
		targetEntity.Region = req.BackupExport.GetAzBlob().GetTargetUri()
	default:
		return 0, bizerr_wrapper.NewInvalidInputError("unsupported cloud provider: %s", bucket.Provider)
	}

	// Create backup export record
	export := &entity.BackupExport{
		UserID:     entity.ID(user.ID),
		TenantID:   backup.TenantID,
		ProjectID:  backup.ProjectID,
		ClusterID:  backup.ClusterID,
		BackupID:   backup.ID,
		Provider:   bucket.Provider,
		Status:     entity.BackupExportStatusPending,
		Updatable:  entity.Updatable{},
		Versioned:  entity.Versioned{},
		DeletedAt:  nil,
		FinishedAt: nil,
		Region:     bucket.MainRegion(),
		Name:       exportName,
		StatusMsg:  "Created",
		Target:     targetEntity,
		JobID:      "",
	}
	if err := m.backupExportRepo.Save(ctx, export); err != nil {
		return 0, err
	}

	// Run backup export workflow
	var dagID string
	switch bucket.Provider {
	case constants.CloudProviderGCP:
		dagID = workflow.DagCreateGCPBackupExport
	case constants.CloudProviderAWS:
		dagID = workflow.DagCreateAWSBackupExport
	default:
		return 0, bizerr_wrapper.NewInvalidInputError(
			"export backup is not supported for cloud provider: %s", bucket.Provider,
		)
	}
	dag, dagErr := m.workflowClient.RunDagWithTags(
		dagID,
		map[string]string{
			workflow.KeyBackupExportID: fmt.Sprintf("%d", export.ID.Uint64()),
			workflow.KeyBackupID:       fmt.Sprintf("%d", backup.ID.Uint64()),
		},
		map[string]string{
			workflow.BusinessTypeKey:   string(workflow.BusinessTypeBackupExport),
			workflow.BusinessActionKey: string(workflow.BusinessActionBackupExportCreate),
			workflow.BusinessID:        fmt.Sprintf("%d", export.ID.Uint64()),
			workflow.ClusterID:         fmt.Sprintf("%d", backup.ClusterID.Uint64()),
		},
	)
	if dagErr != nil {
		return 0, bizerr_wrapper.NewInternalErr("run backup export workflow error: %s", dagErr)
	}

	// write dag ins id to backup export record
	export.DAGInsID = dag.ID
	if err := m.backupExportRepo.Save(ctx, export); err != nil {
		return 0, err
	}
	logger.Info("backup export workflow created",
		zap.String("workflow_id", dag.ID),
		zap.Uint64("export_id", export.ID.Uint64()),
	)

	return export.ID.Uint64(), nil
}

func (m *BackupExportModelImpl) CancelBackupExport(
	ctx context.Context, req *dataflow.CancelBackupExportReq,
) bizerr.BizError {
	export, err := m.backupExportRepo.Load(ctx, req.ExportId)
	if err != nil {
		return err
	}
	preStatus := export.Status
	if err := export.Cancel(time.Now(), "Canceled by user"); err != nil {
		return bizerr_wrapper.NewInvalidInputError("failed to cancel backup export: %s", err)
	}

	switch preStatus {
	case entity.BackupExportStatusPending:
		if err := m.cancelPendingBackupExport(ctx, export); err != nil {
			return err
		}
	case entity.BackupExportStatusRunning:
		if err := m.cancelRunningBackupExport(ctx, export); err != nil {
			return err
		}
	default:
		return bizerr_wrapper.NewInvalidInputError("can't cancel backup export in current status: %s", export.Status)
	}

	// Update export status
	if err := m.backupExportRepo.Save(ctx, export); err != nil {
		return err
	}

	return nil
}

func (m *BackupExportModelImpl) cancelPendingBackupExport(
	ctx context.Context, export *entity.BackupExport,
) bizerr.BizError {
	return m.cancelWorkflowIgnoreFinalState(ctx, export)
}

func (m *BackupExportModelImpl) cancelRunningBackupExport(
	ctx context.Context, export *entity.BackupExport,
) bizerr.BizError {
	logger := log.GetLogger(ctx)
	// Get the backup to check provider
	backup, err := m.backupRepo.Load(ctx, export.BackupID.Uint64())
	if err != nil {
		return err
	}
	bucket, err := m.bucketRepo.Load(ctx, backup.BucketID.Uint64())
	if err != nil {
		return err
	}

	// step 1: cancel the workflow
	if err := m.cancelWorkflowIgnoreFinalState(ctx, export); err != nil {
		return err
	}

	// step 2: cancel the transfer operation
	switch bucket.Provider {
	case constants.CloudProviderGCP:
		if export.JobID == "" {
			logger.Info("no job id found for backup export, ignore cancel transfer operation",
				zap.Uint64("export_id", export.ID.Uint64()),
			)
			break
		}
		if err := m.storageTransferClient.CancelOneTimeTransferJob(ctx, *bucket.CloudAccountID, export.JobID); err != nil {
			return bizerr_wrapper.NewInternalErr("failed to cancel transfer operation: %v", err)
		}
	case constants.CloudProviderAWS:
		if export.JobID == "" {
			logger.Info("no job id found for backup export, ignore cancel batch ops job",
				zap.Uint64("export_id", export.ID.Uint64()),
			)
			break
		}
		if bucket.CloudAccountID == nil || *bucket.CloudAccountID == "" {
			return bizerr_wrapper.NewInternalErr("aws cloud account id is empty")
		}
		if err := m.batchOpsClient.CancelJob(ctx, export.Target.Region, *bucket.CloudAccountID, export.JobID); err != nil {
			if strings.Contains(strings.ToLower(err.Error()), aws_facade.JobStatusForbiddenMsg) {
				// Job status cannot be changed, e.g. already completed
				// Still update the export entity to canceled; otherwise the export entity will be stuck
				if saveErr := m.backupExportRepo.Save(ctx, export); saveErr != nil {
					return bizerr_wrapper.NewInternalErr(
						"failed save entity of finished remote job: %s", multierror.Append(saveErr, err).Error())
				}
				logger.Error("Ignore error about canceling export job "+
					"because current status is unable to be updated", zap.Error(err))
				return nil
			}
			return bizerr_wrapper.NewInternalErr("%v", err)
		}
	default:
		return bizerr_wrapper.NewInvalidInputError("cancellation is not supported for cloud provider: %s", bucket.Provider)
	}
	return nil
}

func (m *BackupExportModelImpl) cancelWorkflowIgnoreFinalState(
	ctx context.Context, export *entity.BackupExport,
) bizerr.BizError {
	logger := log.GetLogger(ctx)
	dagInsList, err := m.workflowClient.ListDagIns(&mod.ListDagInstanceInput{
		IDs: []string{export.DAGInsID},
	})
	if err != nil {
		return bizerr_wrapper.NewInternalErr("failed to list dag instance: %v", err)
	}
	if len(dagInsList) == 0 {
		return nil
	}
	if len(dagInsList) != 1 {
		return bizerr_wrapper.NewInternalErr("unexpected number of dag instances: %d", len(dagInsList))
	}
	dagIns := dagInsList[0]
	if dagIns.Status == ffentity.DagInstanceStatusSuccess ||
		dagIns.Status == ffentity.DagInstanceStatusFailed ||
		dagIns.Status == ffentity.DagInstanceStatusCanceled {
		logger.Info("backup export workflow already finished, ignore cancel",
			zap.Uint64("export_id", export.ID.Uint64()),
			zap.String("status", string(dagIns.Status)),
		)
		return nil
	}

	if err := m.workflowClient.CancelDagIns(export.DAGInsID); err != nil {
		return bizerr_wrapper.NewInternalErr("failed to cancel backup export workflow: %v", err)
	}
	return nil
}