Method CreateCDCFlow

flow/cmd/handler.go:158–221

(
	ctx context.Context, req *protos.CreateCDCFlowRequest,
)

}

func (h *FlowRequestHandler) CreateCDCFlow(
	ctx context.Context, req *protos.CreateCDCFlowRequest,
) (*protos.CreateCDCFlowResponse, APIError) {
	cfg := req.ConnectionConfigs
	if cfg == nil {
		return nil, NewInvalidArgumentApiError(fmt.Errorf("connection configs cannot be nil"))
	}
	if internalVersion, err := internal.PeerDBForceInternalVersion(ctx, cfg.Env); err != nil {
		return nil, NewInternalApiError(err)
	} else {
		cfg.Version = internalVersion
	}
	if flags, err := h.determineFlags(ctx, cfg.Env, cfg.DestinationName); err != nil {
		return nil, NewInternalApiError(err)
	} else {
		cfg.Flags = flags
	}

	if !req.AttachToExisting {
		if exists, err := h.cdcJobEntryExists(ctx, cfg.FlowJobName); err != nil {
			return nil, NewInternalApiError(fmt.Errorf("unable to check flow job entry: %w", err))
		} else if exists {
			return nil, NewAlreadyExistsApiError(fmt.Errorf("flow already exists: %s", cfg.FlowJobName))
		}
	}

	workflowID := getWorkflowID(cfg.FlowJobName)
	desc, err := h.temporalClient.DescribeWorkflow(ctx, workflowID, "")
	if _, ok := errors.AsType[*serviceerror.NotFound](err); err != nil && !ok {
		return nil, NewInternalApiError(fmt.Errorf("failed to query the workflow execution: %w", err))
	} else if err == nil {
		// If workflow is actively running, handle based on AttachToExisting
		// Workflows in terminal states are fine
		if desc.WorkflowExecutionMetadata.Status == tEnums.WORKFLOW_EXECUTION_STATUS_RUNNING ||
			desc.WorkflowExecutionMetadata.Status == tEnums.WORKFLOW_EXECUTION_STATUS_CONTINUED_AS_NEW {
			if req.AttachToExisting {
				// Idempotent attach to running workflow
				return &protos.CreateCDCFlowResponse{
					WorkflowId: workflowID,
				}, nil
			} else {
				// Can't create duplicate of running workflow
				return nil, NewAlreadyExistsApiError(fmt.Errorf("workflow already exists for flow: %s", cfg.FlowJobName))
			}
		}
	}
	// No running workflow, do the validations and start a new one

	// Use idempotent validation that skips mirror existence check
	connectionConfigsCore := pconv.FlowConnectionConfigsToCore(req.ConnectionConfigs, 0)
	if connectionConfigsCore.SkipValidation == nil || !*connectionConfigsCore.SkipValidation {
		if _, err := h.validateCDCMirrorImpl(ctx, connectionConfigsCore, true); err != nil {
			slog.ErrorContext(ctx, "validate mirror error", slog.Any("error", err))
			return nil, NewInternalApiError(fmt.Errorf("invalid mirror: %w", err))
		}
	}

	if resp, err := h.createCDCFlow(ctx, connectionConfigsCore, workflowID); err != nil {