package impl

import (
	"context"

	"github.com/tidbcloud/dataflow-service-ng/app/models/domain"
	"github.com/tidbcloud/dataflow-service-ng/app/repo/entity"
	"github.com/tidbcloud/dataflow-service-ng/app/utils"
	"github.com/tidbcloud/pkgv2/bizerr"
	"google.golang.org/protobuf/proto"
)

// ApplyStoppedBizChangefeedConfig applies caller-owned downstream
// configuration without changing the business status or starting the kernel
// changefeed.
//
// It deliberately cannot change the changefeed specification. Resizing declares
// a new expected specification and then waits for the shared TiCDC cluster to be
// reconciled, which ScaleBizChangefeed and its workflow already own; the status
// matrix also only allows Scale from a running feed. Everything here completes
// within one request.
func (m *BizChangefeedModelImpl) ApplyStoppedBizChangefeedConfig(
	ctx context.Context, req *domain.ApplyStoppedBizChangefeedConfigReq,
) (err bizerr.BizError) {
	if req == nil || req.DownstreamParameter == nil {
		return bizerr.New(bizerr.ErrCodeParams, "downstream parameters are required")
	}

	bizChangefeedEntity, err := m.bizChangefeedRepo.Load(ctx, req.BizChangefeedID)
	if err != nil {
		return err.Chain("load biz_changefeed")
	}
	kernelChangefeed, err := m.queryKernelChangefeed(ctx, bizChangefeedEntity.ID.Uint64())
	if err != nil {
		return err.Chain("query kernel changefeed")
	}
	if err = m.preCheckApplyStoppedBizChangefeedConfig(
		bizChangefeedEntity, kernelChangefeed); err != nil {
		return err.Chain("precheck apply stopped biz_changefeed config")
	}

	cluster, err := m.mgmtCli.GetClusterByID(ctx, bizChangefeedEntity.ClusterID.Uint64())
	if err != nil {
		return err.Chain("get cluster")
	}
	// The downstream is the changefeed's own, exactly as it is on every other
	// path that edits one that already exists.
	downstreamType := domain.TiCDCChangefeedDownStreamType2ConnectorType(
		bizChangefeedEntity.DownstreamType)
	downstreamAdaptor, err := m.downstreamAdaptorFactory.Get(downstreamType)
	if err != nil {
		return err.Chain("get downstream adaptor")
	}

	// CompleteEditParams mutates its protobuf argument when it preserves fixed
	// endpoint fields or encrypts a new secret. Clone it so a workflow retry
	// cannot double-encrypt a caller-owned request object.
	//
	// Every downstream adaptor takes a pbgen message, so the assertion always
	// holds today. An adaptor whose parameters are not a proto.Message would
	// fall through uncloned and be mutated in place - the exact hazard above -
	// so adding one means extending the clone here, not relaxing this check.
	downstreamParams := req.DownstreamParameter
	if message, ok := downstreamParams.(proto.Message); ok {
		downstreamParams = proto.Clone(message)
	}
	if err = downstreamAdaptor.CompleteEditParams(
		ctx, downstreamParams, bizChangefeedEntity.DownstreamParams); err != nil {
		return err.Chain("complete edit parameters")
	}

	defer func() {
		m.handleEventHook(ctx, err, entity.EditChangefeed, req.UserID, bizChangefeedEntity)
	}()

	editReq := &domain.EditBizChangefeedReq{
		BizChangefeedID:     req.BizChangefeedID,
		Name:                req.Name,
		DownstreamType:      downstreamType,
		DownstreamParameter: downstreamParams,
		UserID:              req.UserID,
	}
	cw := utils.NewCluster(cluster)
	transErr := m.transProvider.Transaction(ctx, func(ctx context.Context) error {
		if updateErr := m.updateBizChangefeedEntityForEdit(ctx, editReq, bizChangefeedEntity); updateErr != nil {
			return updateErr.Chain("update stopped biz_changefeed entity")
		}
		if editErr := m.editKernelChangefeed(
			ctx, cw, downstreamAdaptor, bizChangefeedEntity, kernelChangefeed); editErr != nil {
			return editErr.Chain("edit stopped kernel changefeed")
		}
		return nil
	})
	if transErr != nil {
		return convertTransactionError(transErr)
	}

	return nil
}

func (m *BizChangefeedModelImpl) preCheckApplyStoppedBizChangefeedConfig(
	bizChangefeedEntity *entity.BizChangefeed,
	kernelChangefeed *entity.TiCDCChangefeed,
) bizerr.BizError {
	if bizChangefeedEntity.OperationStatus != entity.BizChangefeedOperationStatusPaused {
		return bizerr.New(
			bizerr.ErrCodeParams,
			"current status %s, stopped config requires paused", bizChangefeedEntity.OperationStatus)
	}
	if kernelChangefeed.KernelMetrics.FeedState != entity.ChangefeedKernelStatusStopped {
		return bizerr.New(
			bizerr.ErrCodeParams, "kernel changefeed is not stopped, can't apply stopped config")
	}
	return nil
}