package export

import (
	"context"
	"strconv"

	eventdata "github.com/tidbcloud/eventbus-datastruct/event"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"

	globalngv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/nextgen/v1"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/service/displayevent"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/service/workflow/common"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/infra/repoimpl/schema"
	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
	"github.com/tidbcloud/tidb-management-service/utilities/workflow"
)

func (s *serviceImpl) CancelExport(ctx context.Context, req *globalngv1.CancelExportRequest) (*globalngv1.Export, bizerr.BizError) {
	if req == nil || req.GetClusterId() == "" || req.GetExportId() == "" {
		return nil, bizerr.New(bizerr.ErrCodeParams, "cluster_id and export_id are required")
	}

	tilog.Info(ctx, "cancel export request", zap.String("cluster_id", req.GetClusterId()), zap.String("export_id", req.GetExportId()))

	ent, err := s.exportRepo.GetByClusterAndExportId(ctx, req.GetClusterId(), req.GetExportId())
	if err != nil {
		return nil, err
	}
	if ent == nil {
		return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "export not found")
	}
	if ent.State != schema.ExportStateRunning {
		return nil, bizerr.New(bizerr.ErrCodeParams, "export can not be canceled because it is not in running state")
	}
	clusterID, parseErr := strconv.ParseUint(req.GetClusterId(), 10, 64)
	if parseErr != nil || clusterID == 0 {
		return nil, bizerr.New(bizerr.ErrCodeParams, "invalid cluster_id")
	}
	if lifecycleErr := s.checkResourcePoolLifecycle(ctx, clusterID, false); lifecycleErr != nil {
		return nil, lifecycleErr
	}

	errx := s.transaction.Transaction(ctx, func(ctx context.Context) error {
		if lifecycleErr := s.checkResourcePoolLifecycle(ctx, clusterID, true); lifecycleErr != nil {
			return lifecycleErr
		}
		ent.State = schema.ExportStateCanceled
		if err := s.exportRepo.Cancel(ctx, ent.ClusterId, ent.ExportId); err != nil {
			tilog.Error(ctx, "failed to cancel export", zap.Error(err))
			return err
		}
		if err := s.workflow.DeleteDagInstances(ctx, ent.ExportId, nil); err != nil {
			return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeUnknown, "failed to delete create_export dag instances")
		}

		workFlowErr := s.workflow.RunDagWithBizId(
			ctx,
			common.CancelExport,
			strconv.FormatUint(ent.OrganizationId, 10),
			"0",
			req.GetExportId(),
			map[string]string{
				workflow.KeyTraceId: workflow.GetTraceId(ctx),
				common.KeyExportId:  req.GetExportId(),
				common.KeyClusterId: req.GetClusterId(),
			},
		)
		if workFlowErr != nil {
			return bizerr.WrapErrorWithCode(workFlowErr, bizerr.ErrCodeUnknown, "create cancel export workflow failed")
		}
		displayEv := newExportDisplayEvent(ctx, ent, displayevent.CancelExport, eventdata.EventStatusSucceeded, "Export canceled successfully.")
		displayevent.Write(ctx, s.eventCommitter, displayEv)
		return nil
	})
	if errx != nil {
		tilog.Error(ctx, "cancel export failed", zap.Error(errx), zap.String("export_id", req.GetExportId()))
		return nil, bizerr.WrapError(errx, "cancel export failed")
	}
	return schemaToProto(ent.Export), nil
}