Add Provisioning dispatch executor

components/org.wso2.carbon.identity.recovery/src/main/java/org/wso2/carbon/identity/recovery/executor/ProvisioningDispatchExecutor.java

@@ -0,0 +1,134 @@
+/*
+ * Copyright (c) 2026, WSO2 LLC. (https://www.wso2.com).
+ *
+ * WSO2 LLC. licenses this file to you under the Apache License,
+ * Version 2.0 (the "License"); you may not use this file except
+ * in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.wso2.carbon.identity.recovery.executor;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.wso2.carbon.identity.flow.execution.engine.exception.FlowEngineException;
+import org.wso2.carbon.identity.flow.execution.engine.graph.Executor;
+import org.wso2.carbon.identity.flow.execution.engine.model.ExecutorResponse;
+import org.wso2.carbon.identity.flow.execution.engine.model.FlowExecutionContext;
+import org.wso2.carbon.identity.recovery.internal.IdentityRecoveryServiceDataHolder;
+
+import java.util.Collections;
+import java.util.List;
+
+import static org.wso2.carbon.identity.flow.execution.engine.Constants.ExecutorStatus.STATUS_COMPLETE;
+import static org.wso2.carbon.identity.flow.execution.engine.Constants.ExecutorStatus.STATUS_ERROR;
+
+/**
+ * Flow executor that provisions a user and then the organization the flow collected, in that order.
+ * <p>
+ * The flow builder decides whether this executor is needed: it names this executor on the END step when
+ * the flow collects organization details, and {@code UserProvisioningExecutor} when it does not. So by
+ * the time this runs the decision is already made, and it dispatches to both unconditionally rather than
+ * classifying the flow itself.
+ * <p>
+ * Ordering matters. {@code OrganizationProvisioningExecutor} requires a provisioned user, because the
+ * creating user becomes the organization owner. Running user provisioning first leaves the user ID on
+ * the flow user, which the organization executor then reads from the same context.
+ * <p>
+ * The executors are resolved by name from executors contributed across all bundles, so this component
+ * does not depend on the one that owns the organization executor.
+ */
+public class ProvisioningDispatchExecutor implements Executor {
+
+    private static final Log LOG = LogFactory.getLog(ProvisioningDispatchExecutor.class);
+
+    private static final String EXECUTOR_NAME = "ProvisioningDispatchExecutor";
+    private static final String USER_PROVISIONING_EXECUTOR = "UserProvisioningExecutor";
+    private static final String ORGANIZATION_PROVISIONING_EXECUTOR = "OrganizationProvisioningExecutor";
+
+    @Override
+    public String getName() {
+
+        return EXECUTOR_NAME;
+    }
+
+    @Override
+    public ExecutorResponse execute(FlowExecutionContext context) throws FlowEngineException {
+
+        ExecutorResponse userResponse = dispatch(USER_PROVISIONING_EXECUTOR, context);
+        if (userResponse == null) {
+            return unavailableExecutorResponse(USER_PROVISIONING_EXECUTOR);
+        }
+        // Anything other than completion is the user provisioning step's own outcome to report: it may
+        // need more input, or it may have failed. The organization must not be created either way.
+        if (!STATUS_COMPLETE.equals(userResponse.getResult())) {
+            return userResponse;
+        }
+
+        ExecutorResponse organizationResponse = dispatch(ORGANIZATION_PROVISIONING_EXECUTOR, context);
+        if (organizationResponse == null) {
+            return unavailableExecutorResponse(ORGANIZATION_PROVISIONING_EXECUTOR);
+        }
+        return organizationResponse;
+    }
+
+    /**
+     * Resolves an executor by name and runs it against the same flow context.
+     *
+     * @param executorName Name the executor is registered under.
+     * @param context      Flow execution context, shared by both executors.
+     * @return The executor's response, or {@code null} if no executor is registered under that name.
+     * @throws FlowEngineException If the executor fails.
+     */
+    private ExecutorResponse dispatch(String executorName, FlowExecutionContext context)
+            throws FlowEngineException {
+
+        Executor executor = IdentityRecoveryServiceDataHolder.getInstance().getFlowExecutor(executorName);
+        if (executor == null) {
+            return null;
+        }
+        if (LOG.isDebugEnabled()) {
+            LOG.debug("Dispatching to executor: " + executorName + " for flow: "
+                    + context.getContextIdentifier());
+        }
+        return executor.execute(context);
+    }
+
+    /**
+     * Builds the response for a flow that names this executor while one of the executors it dispatches
+     * to is not deployed. The end user cannot resolve this by retrying.
+     *
+     * @param executorName Name of the executor that could not be resolved.
+     * @return An error response.
+     */
+    private ExecutorResponse unavailableExecutorResponse(String executorName) {
+
+        LOG.error("Executor not found: " + executorName + ". The provisioning dispatch executor requires "
+                + "both the user and organization provisioning executors to be deployed.");
+        ExecutorResponse response = new ExecutorResponse();
+        response.setResult(STATUS_ERROR);
+        response.setErrorMessage("Provisioning is not available.");
+        return response;
+    }
+
+    @Override
+    public List<String> getInitiationData() {
+
+        return Collections.emptyList();
+    }
+
+    @Override
+    public ExecutorResponse rollback(FlowExecutionContext context) throws FlowEngineException {
+
+        return null;
+    }
+}

components/org.wso2.carbon.identity.recovery/src/main/java/org/wso2/carbon/identity/recovery/internal/IdentityRecoveryServiceComponent.java

@@ -56,6 +56,7 @@
 import org.wso2.carbon.identity.recovery.connector.UserEmailVerificationConfigImpl;
 import org.wso2.carbon.identity.recovery.executor.ConfirmationCodeValidationExecutor;
 import org.wso2.carbon.identity.recovery.executor.PasswordProvisioningExecutor;
+import org.wso2.carbon.identity.recovery.executor.ProvisioningDispatchExecutor;
 import org.wso2.carbon.identity.recovery.executor.UserProvisioningExecutor;
 import org.wso2.carbon.identity.recovery.handler.AccountConfirmationValidationHandler;
 import org.wso2.carbon.identity.recovery.handler.AdminForcedPasswordResetHandler;
@@ -156,6 +157,8 @@ protected void activate(ComponentContext context) {
                     new PasswordProvisioningExecutor(), null);
             bundleContext.registerService(Executor.class.getName(),
                     new UserProvisioningExecutor(), null);
+            bundleContext.registerService(Executor.class.getName(),
+                    new ProvisioningDispatchExecutor(), null);
         } catch (Exception e) {
             log.error("Error while activating identity governance component.", e);
         }
@@ -186,6 +189,28 @@ protected void setRealmService(RealmService realmService) {
         dataHolder.setRealmService(realmService);
     }
 
+    @Reference(
+            name = "flow.executor",
+            service = org.wso2.carbon.identity.flow.execution.engine.graph.Executor.class,
+            cardinality = ReferenceCardinality.MULTIPLE,
+            policy = ReferencePolicy.DYNAMIC,
+            unbind = "unsetFlowExecutor")
+    protected void setFlowExecutor(Executor executor) {
+
+        if (log.isDebugEnabled()) {
+            log.debug("Adding flow executor: " + executor.getName());
+        }
+        dataHolder.addFlowExecutor(executor);
+    }
+
+    protected void unsetFlowExecutor(Executor executor) {
+
+        if (log.isDebugEnabled()) {
+            log.debug("Removing flow executor: " + executor.getName());
+        }
+        dataHolder.removeFlowExecutor(executor);
+    }
+
     @Reference(
             name = "otpgenerator.service",
             service = org.wso2.carbon.identity.governance.service.otp.OTPGenerator.class,

components/org.wso2.carbon.identity.recovery/src/main/java/org/wso2/carbon/identity/recovery/internal/IdentityRecoveryServiceDataHolder.java

@@ -18,6 +18,7 @@
 
 package org.wso2.carbon.identity.recovery.internal;
 
+import org.apache.commons.lang.StringUtils;
 import org.wso2.carbon.consent.mgt.core.ConsentManager;
 import org.wso2.carbon.identity.application.mgt.ApplicationManagementService;
 import org.wso2.carbon.identity.auth.attribute.handler.AuthAttributeHandlerManager;
@@ -26,6 +27,7 @@
 import org.wso2.carbon.identity.consent.mgt.services.ConsentUtilityService;
 import org.wso2.carbon.identity.core.persistence.registry.RegistryResourceMgtService;
 import org.wso2.carbon.identity.event.services.IdentityEventService;
+import org.wso2.carbon.identity.flow.execution.engine.graph.Executor;
 import org.wso2.carbon.identity.governance.IdentityGovernanceService;
 import org.wso2.carbon.identity.governance.service.IdentityDataStoreService;
 import org.wso2.carbon.identity.governance.service.otp.OTPGenerator;
@@ -41,10 +43,12 @@
 
 import java.util.Map;
 import java.util.TreeMap;
+import java.util.concurrent.ConcurrentHashMap;
 
 public class IdentityRecoveryServiceDataHolder {
 
     private static IdentityRecoveryServiceDataHolder instance = new IdentityRecoveryServiceDataHolder();
+    private final Map<String, Executor> flowExecutors = new ConcurrentHashMap<>();
     private RealmService realmService;
     private IdentityEventService identityEventService;
     private IdentityGovernanceService identityGovernanceService;
@@ -367,4 +371,42 @@ public WorkflowManagementService getWorkflowManagementService() {
 
         return this.workflowService;
     }
+
+    /**
+     * Add a flow executor, keyed by the name it is bound to in a flow definition.
+     * <p>
+     * Executors are collected from every bundle that contributes them, so an executor owned by another
+     * component can be resolved here without depending on that component directly.
+     *
+     * @param executor Flow executor contributed as an OSGi service.
+     */
+    public void addFlowExecutor(Executor executor) {
+
+        if (executor != null && StringUtils.isNotBlank(executor.getName())) {
+            flowExecutors.put(executor.getName(), executor);
+        }
+    }
+
+    /**
+     * Remove a flow executor that is no longer available.
+     *
+     * @param executor Flow executor being withdrawn.
+     */
+    public void removeFlowExecutor(Executor executor) {
+
+        if (executor != null && StringUtils.isNotBlank(executor.getName())) {
+            flowExecutors.remove(executor.getName());
+        }
+    }
+
+    /**
+     * Get a flow executor by the name it is bound to in a flow definition.
+     *
+     * @param executorName Name of the executor.
+     * @return The executor, or {@code null} if no executor is registered under that name.
+     */
+    public Executor getFlowExecutor(String executorName) {
+
+        return flowExecutors.get(executorName);
+    }
 }