Allow running scheduled workflows

Closes #44865

Signed-off-by: Pedro Igor <pigor.craveiro@gmail.com>

core/src/main/java/org/keycloak/representations/workflows/WorkflowConstants.java

@@ -9,6 +9,7 @@ public final class WorkflowConstants {
 
     // Entry configuration keys for Workflow
     public static final String CONFIG_ON_EVENT = "on";
+    public static final String CONFIG_SCHEDULE = "schedule";
     public static final String CONFIG_CONCURRENCY = "concurrency";
     public static final String CONFIG_RESTART_IN_PROGRESS = "restart-in-progress";
     public static final String CONFIG_CANCEL_IN_PROGRESS = "cancel-in-progress";
@@ -26,4 +27,9 @@ public final class WorkflowConstants {
     public static final String CONFIG_AFTER = "after";
     public static final String CONFIG_PRIORITY = "priority";
     public static final String CONFIG_SCHEDULED_AT = "scheduled-at";
+
+    // Entry configuration keys for WorkflowSchedule
+    public static final String CONFIG_SCHEDULE_AFTER = "schedule." + CONFIG_AFTER;
+    public static final String CONFIG_BATCH_SIZE = "batch-size";
+    public static final String CONFIG_SCHEDULE_BATCH_SIZE = "schedule." + CONFIG_BATCH_SIZE;
 }

core/src/main/java/org/keycloak/representations/workflows/WorkflowRepresentation.java

@@ -20,12 +20,15 @@ import static org.keycloak.representations.workflows.WorkflowConstants.CONFIG_IF
 import static org.keycloak.representations.workflows.WorkflowConstants.CONFIG_NAME;
 import static org.keycloak.representations.workflows.WorkflowConstants.CONFIG_ON_EVENT;
 import static org.keycloak.representations.workflows.WorkflowConstants.CONFIG_RESTART_IN_PROGRESS;
+import static org.keycloak.representations.workflows.WorkflowConstants.CONFIG_SCHEDULE;
+import static org.keycloak.representations.workflows.WorkflowConstants.CONFIG_SCHEDULE_AFTER;
+import static org.keycloak.representations.workflows.WorkflowConstants.CONFIG_SCHEDULE_BATCH_SIZE;
 import static org.keycloak.representations.workflows.WorkflowConstants.CONFIG_STATE;
 import static org.keycloak.representations.workflows.WorkflowConstants.CONFIG_STEPS;
 import static org.keycloak.representations.workflows.WorkflowConstants.CONFIG_USES;
 import static org.keycloak.representations.workflows.WorkflowConstants.CONFIG_WITH;
 
-@JsonPropertyOrder({"id", CONFIG_NAME, CONFIG_USES, CONFIG_ENABLED, CONFIG_ON_EVENT, CONFIG_CONCURRENCY, CONFIG_IF, CONFIG_STEPS, CONFIG_STATE})
+@JsonPropertyOrder({"id", CONFIG_NAME, CONFIG_USES, CONFIG_ENABLED, CONFIG_ON_EVENT, CONFIG_SCHEDULE, CONFIG_CONCURRENCY, CONFIG_IF, CONFIG_STEPS, CONFIG_STATE})
 @JsonIgnoreProperties(CONFIG_WITH)
 @JsonInclude(JsonInclude.Include.NON_NULL)
 public final class WorkflowRepresentation extends AbstractWorkflowComponentRepresentation {
@@ -41,6 +44,9 @@ public final class WorkflowRepresentation extends AbstractWorkflowComponentRepre
     @JsonProperty(CONFIG_CONCURRENCY)
     private WorkflowConcurrencyRepresentation concurrency;
 
+    @JsonProperty(CONFIG_SCHEDULE)
+    private WorkflowScheduleRepresentation schedule;
+
     public WorkflowRepresentation() {
         super(null, null);
     }
@@ -59,6 +65,29 @@ public final class WorkflowRepresentation extends AbstractWorkflowComponentRepre
         setConfigValue(CONFIG_ON_EVENT, eventConditions);
     }
 
+    public WorkflowScheduleRepresentation getSchedule() {
+        if (schedule == null) {
+            String after = getConfigValue(CONFIG_SCHEDULE_AFTER, String.class);
+            Integer batchSize = getConfigValue(CONFIG_SCHEDULE_BATCH_SIZE, Integer.class);
+
+            if (after != null || batchSize != null) {
+                this.schedule = new WorkflowScheduleRepresentation();
+                this.schedule.setAfter(after);
+                this.schedule.setBatchSize(batchSize);
+            }
+        }
+
+        return this.schedule;
+    }
+
+    public void setSchedule(WorkflowScheduleRepresentation schedule) {
+        this.schedule = schedule;
+        if (schedule != null) {
+            setConfigValue(CONFIG_SCHEDULE_AFTER, schedule.getAfter());
+            setConfigValue(CONFIG_SCHEDULE_BATCH_SIZE, schedule.getBatchSize());
+        }
+    }
+
     public String getName() {
         return getConfigValue(CONFIG_NAME, String.class);
     }
@@ -217,6 +246,11 @@ public final class WorkflowRepresentation extends AbstractWorkflowComponentRepre
             return this;
         }
 
+        public Builder schedule(WorkflowScheduleRepresentation schedule) {
+            representation.setSchedule(schedule);
+            return this;
+        }
+
         public WorkflowRepresentation build() {
             return representation;
         }

core/src/main/java/org/keycloak/representations/workflows/WorkflowScheduleRepresentation.java

@@ -0,0 +1,57 @@
+package org.keycloak.representations.workflows;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonPropertyOrder;
+
+import static org.keycloak.representations.workflows.WorkflowConstants.CONFIG_AFTER;
+import static org.keycloak.representations.workflows.WorkflowConstants.CONFIG_BATCH_SIZE;
+
+@JsonPropertyOrder({CONFIG_AFTER, CONFIG_BATCH_SIZE})
+@JsonInclude(JsonInclude.Include.NON_NULL)
+public class WorkflowScheduleRepresentation {
+
+    private String after;
+
+    @JsonProperty(CONFIG_BATCH_SIZE)
+    private Integer batchSize;
+
+    public static Builder create() {
+        return new Builder();
+    }
+
+    public String getAfter() {
+        return after;
+    }
+
+    public void setAfter(String after) {
+        this.after = after;
+    }
+
+    public Integer getBatchSize() {
+        return this.batchSize;
+    }
+
+    public void setBatchSize(Integer batchSize) {
+        this.batchSize = batchSize;
+    }
+
+    public static class Builder {
+
+        private final WorkflowScheduleRepresentation schedule = new WorkflowScheduleRepresentation();
+
+        public Builder after(String after) {
+            schedule.setAfter(after);
+            return this;
+        }
+
+        public Builder batchSize(int batchSize) {
+            schedule.setBatchSize(batchSize);
+            return this;
+        }
+
+        public WorkflowScheduleRepresentation build() {
+            return schedule;
+        }
+    }
+}

docs/documentation/server_admin/topics/assembly-managing-workflows.adoc

@@ -8,6 +8,7 @@ include::workflows/understanding-workflow-definition.adoc[leveloffset=+2]
 include::workflows/understanding-workflow-expression.adoc[leveloffset=+2]
 include::workflows/managing-workflows.adoc[leveloffset=+2]
 include::workflows/listening-workflow-events.adoc[leveloffset=+2]
+include::workflows/scheduling-workflows.adoc[leveloffset=+2]
 include::workflows/defining-conditions.adoc[leveloffset=+2]
 include::workflows/defining-steps.adoc[leveloffset=+2]
 include::workflows/understanding-workflows-engine.adoc[leveloffset=+2]

docs/documentation/server_admin/topics/organizations/managing-identity-providers.adoc

@@ -70,7 +70,7 @@ image:images/organizations-edit-identity-provider.png[alt="Editing linked identi
 
 == Unlinking an identity provider from an organization
 
-When an identity provider is unlinked from an organization, it remains available as a realm-level provider that is no longer ssociated with an organization. To delete the unlinked provider, use the *Identity Providers* section in the menu.
+When an identity provider is unlinked from an organization, it remains available as a realm-level provider that is no longer associated with an organization. To delete the unlinked provider, use the *Identity Providers* section in the menu.
 
 .Procedure
 

docs/documentation/server_admin/topics/workflows/defining-conditions.adoc

@@ -3,7 +3,8 @@
 [[_workflow_conditions_]]
 = Defining conditions
 
-The optional `if` setting allows you to define the conditions the target resource must meet in order for the workflow to be triggered.
+The optional `if` setting allows you to define the conditions, as expressions, that the target resource must meet in
+order for the workflow to be triggered. See <<_workflow_expression_language_>> for more details.
 
 Conditions provide fine-grained control over whether a workflow execution should be created.
 They allow you to inspect the context of the event and the state of the resource.

docs/documentation/server_admin/topics/workflows/scheduling-workflows.adoc

@@ -0,0 +1,34 @@
+[id="scheduling-workflows_{context}"]
+
+[[_scheduling_workflows_]]
+= Scheduling workflows
+
+Workflows can be scheduled to run periodically according to a defined interval. This is done using the `schedule` setting
+in the workflow definition.
+
+A key difference between event-based triggering and scheduled workflows is that scheduled workflows follows a passive execution model
+where the workflow engine periodically checks for realm resources that meet the defined condition.
+
+```yaml
+name: Track inactive users
+schedule:
+  after: 5s
+  batch-size: 100
+```
+
+This method of scheduling is useful for automating tasks that need to be performed regularly, such as cleaning up inactive
+users or enforcing specific policies on realm resources. It is an alternative to event-based triggering, but it can also
+be used in combination with it. When used together, the workflow will be triggered either by the defined event or by the schedule.
+
+When a workflow is scheduled, it will be triggered automatically at the defined interval. At each run, the workflow engine
+will query for the realm resources that matches the workflow's condition and will create a workflow execution for each of them,
+up to the defined batch size. If no condition is defined, the workflow will be executed for all realm resources of the type associated
+with the workflow. In the example above, the workflow is scheduled to run every 5 seconds and will process up to 100 realm resources at each run.
+
+The `schedule` setting supports the following parameters:
+
+*   `after`: Defines the interval between each run of the workflow.
+
+*   `batch-size`: Defines the maximum number of realm resources to process at each run.
+
+

docs/documentation/server_admin/topics/workflows/understanding-common-use-cases.adoc

@@ -42,7 +42,10 @@ When a user has been inactive for a certain period, a workflow can send reminder
 
 ```yaml
 name: Track inactive users
-on: user_authenticated
+on: user-authenticated
+schedule:
+  after: 5s
+  batch-size: 2
 concurrency:
   restart-in-progress: true
 steps:

docs/documentation/server_admin/topics/workflows/understanding-workflow-definition.adoc

@@ -53,6 +53,12 @@ This setting is mandatory.
 `on`::
 Define a condition that determines the event that will trigger the workflow.
 The condition is written using an expression language that supports a variety of checks on the event.
+See <<_workflow_events_>> for more details.
+This setting is optional.
+
+`schedule`::
+Define a schedule that will trigger the workflow at defined intervals.
+See <<_scheduling_workflows_>> for more details.
 This setting is optional.
 
 `if`::
@@ -61,10 +67,12 @@ The condition is written using an expression language that supports a variety of
 associated with the event.
 A workflow execution is only created if the expression evaluates to `true`.
 If this setting is omitted, the event defined in the `on` setting will always create the workflow execution.
+See <<_workflow_conditions_>> for more details.
 This setting is optional.
 
 `steps`::
 Define the step chain consisting of one or more steps to be sequentially executed during the lifetime of a workflow.
+See <<_workflow_steps_>> for more details.
 This setting is mandatory.
 
 `concurrency`::

integration/admin-client/src/main/java/org/keycloak/admin/client/resource/WorkflowResource.java

@@ -28,14 +28,6 @@ public interface WorkflowResource {
     @Produces(APPLICATION_JSON)
     WorkflowRepresentation toRepresentation();
 
-    @Path("activate-all")
-    @POST
-    void activateAll();
-
-    @Path("activate-all")
-    @POST
-    void activateAll(@QueryParam("notBefore") String notBefore);
-
     @Path("activate/{type}/{resourceId}")
     @POST
     void activate(@PathParam("type") String type, @PathParam("resourceId") String resourceId);

model/jpa/src/main/java/org/keycloak/models/workflow/DefaultWorkflowProvider.java

@@ -1,5 +1,6 @@
 package org.keycloak.models.workflow;
 
+import java.time.Duration;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -24,6 +25,8 @@ import org.keycloak.models.workflow.WorkflowStateProvider.ScheduledStep;
 import org.keycloak.representations.workflows.WorkflowConstants;
 import org.keycloak.representations.workflows.WorkflowRepresentation;
 import org.keycloak.representations.workflows.WorkflowStepRepresentation;
+import org.keycloak.services.scheduled.ClusterAwareScheduledTaskRunner;
+import org.keycloak.timer.TimerProvider;
 
 import org.jboss.logging.Logger;
 
@@ -95,6 +98,9 @@ public class DefaultWorkflowProvider implements WorkflowProvider {
             // finally, update the workflow's config along with the steps' configs
             workflow.updateConfig(representation.getConfig(), newSteps);
         }
+
+        cancelScheduledWorkflow(workflow);
+        scheduleWorkflow(workflow);
     }
 
     @Override
@@ -104,6 +110,7 @@ public class DefaultWorkflowProvider implements WorkflowProvider {
         realm.getComponentsStream(workflow.getId(), WorkflowStepProvider.class.getName()).forEach(realm::removeComponent);
         realm.removeComponent(component);
         stateProvider.removeByWorkflow(workflow.getId());
+        cancelScheduledWorkflow(workflow);
     }
 
     @Override
@@ -341,6 +348,29 @@ public class DefaultWorkflowProvider implements WorkflowProvider {
             model.setConfig(config);
         }
 
-        return new Workflow(session, realm.addComponentModel(model));
+        workflow = new Workflow(session, realm.addComponentModel(model));
+
+        scheduleWorkflow(workflow);
+
+        return workflow;
+    }
+
+    private void scheduleWorkflow(Workflow workflow) {
+        String scheduled = workflow.getConfig().getFirst(WorkflowConstants.CONFIG_SCHEDULE_AFTER);
+
+        if (scheduled != null) {
+            Duration duration = DurationConverter.parseDuration(scheduled);
+            TimerProvider timer = session.getProvider(TimerProvider.class);
+            timer.schedule(new ClusterAwareScheduledTaskRunner(sessionFactory, new ScheduledWorkflowRunner(workflow.getId(), realm.getId()), duration.toMillis()), duration.toMillis());
+        }
+    }
+
+    void cancelScheduledWorkflow(Workflow workflow) {
+        session.getProvider(TimerProvider.class).cancelTask(new ScheduledWorkflowRunner(workflow.getId(), realm.getId()).getTaskName());
+    }
+
+    void rescheduleWorkflow(Workflow workflow) {
+        cancelScheduledWorkflow(workflow);
+        scheduleWorkflow(workflow);
     }
 }

model/jpa/src/main/java/org/keycloak/models/workflow/DefaultWorkflowProviderFactory.java

@@ -8,10 +8,15 @@ import org.keycloak.component.ComponentModel;
 import org.keycloak.executors.ExecutorsProvider;
 import org.keycloak.models.KeycloakSession;
 import org.keycloak.models.KeycloakSessionFactory;
+import org.keycloak.models.RealmModel.RealmRemovedEvent;
+import org.keycloak.models.utils.KeycloakModelUtils;
+import org.keycloak.models.utils.PostMigrationEvent;
 import org.keycloak.provider.ProviderConfigProperty;
 import org.keycloak.provider.ProviderConfigurationBuilder;
+import org.keycloak.provider.ProviderEvent;
+import org.keycloak.provider.ProviderEventListener;
 
-public class DefaultWorkflowProviderFactory implements WorkflowProviderFactory<DefaultWorkflowProvider> {
+public class DefaultWorkflowProviderFactory implements WorkflowProviderFactory<DefaultWorkflowProvider>, ProviderEventListener {
 
     static final String ID = "default";
     private static final long DEFAULT_EXECUTOR_TASK_TIMEOUT = 1000L;
@@ -44,8 +49,37 @@ public class DefaultWorkflowProviderFactory implements WorkflowProviderFactory<D
     @Override
     public void postInit(KeycloakSessionFactory factory) {
         this.executor = new WorkflowExecutor(getTaskExecutor(factory), blocking, taskTimeout);
+        factory.register(this);
     }
 
+    @Override
+    public void onEvent(ProviderEvent event) {
+        if (event instanceof PostMigrationEvent ev) {
+            KeycloakModelUtils.runJobInTransaction(ev.getFactory(), session ->
+                    session.realms().getRealmsStream().forEach(realm -> {
+                        session.getContext().setRealm(realm);
+                        DefaultWorkflowProvider provider = create(session);
+
+                        try {
+                            provider.getWorkflows().forEach(provider::rescheduleWorkflow);
+                        } finally {
+                            session.getContext().setRealm(null);
+                            provider.close();
+                        }
+                    }));
+        } else if (event instanceof RealmRemovedEvent ev) {
+            KeycloakSession session = ev.getKeycloakSession();
+            DefaultWorkflowProvider provider = create(session);
+
+            try {
+                provider.getWorkflows().forEach(provider::cancelScheduledWorkflow);
+            } finally {
+                provider.close();
+            }
+        }
+    }
+
+
     @Override
     public void close() {
 

model/jpa/src/main/java/org/keycloak/models/workflow/ScheduledWorkflowRunner.java

@@ -0,0 +1,54 @@
+package org.keycloak.models.workflow;
+
+import org.keycloak.models.KeycloakSession;
+import org.keycloak.models.RealmModel;
+import org.keycloak.timer.ScheduledTask;
+
+import org.jboss.logging.Logger;
+
+public class ScheduledWorkflowRunner implements ScheduledTask {
+
+    private static final Logger log = Logger.getLogger(DefaultWorkflowProvider.class);
+
+    private final String workflowId;
+    private final String realmId;
+
+    public ScheduledWorkflowRunner(String workflowId, String realmId) {
+        this.workflowId = workflowId;
+        this.realmId = realmId;
+    }
+
+    @Override
+    public void run(KeycloakSession session) {
+        RealmModel realm = session.realms().getRealm(realmId);
+
+        if (realm == null) {
+            log.warnf("Realm %s for scheduled workflow %s not found, cancelling task", realmId, workflowId);
+            throw new IllegalStateException("Realm for scheduled workflow not found: " + realmId);
+        }
+
+        session.getContext().setRealm(realm);
+        WorkflowProvider provider = session.getProvider(WorkflowProvider.class);
+        Workflow workflow = provider.getWorkflow(workflowId);
+
+        if (workflow == null) {
+            log.warnf("Scheduled workflow %s in realm %s not found, cancelling task", workflowId, realmId);
+            throw new IllegalStateException("Scheduled workflow not found: " + workflowId);
+        }
+
+        log.debugf("Executing scheduled workflow '%s' in realm %s", workflow.getName(), realm.getName());
+
+        try {
+            provider.activateForAllEligibleResources(workflow);
+        } catch (Exception e) {
+            log.errorf(e, "Error while executing scheduled workflow %s in realm %s", workflow.getName(), realm.getName());
+        }
+
+        log.debugf("Finished executing scheduled workflow '%s' in realm %s", workflow.getName(), realm.getName());
+    }
+
+    @Override
+    public String getTaskName() {
+        return "workflow-" + workflowId;
+    }
+}

model/jpa/src/main/java/org/keycloak/models/workflow/UserResourceTypeWorkflowProvider.java

@@ -36,6 +36,7 @@ import org.keycloak.models.jpa.entities.UserEntity;
 import org.keycloak.models.workflow.conditions.expression.BooleanConditionParser;
 import org.keycloak.models.workflow.conditions.expression.EvaluatorUtils;
 import org.keycloak.models.workflow.conditions.expression.PredicateEvaluator;
+import org.keycloak.representations.workflows.WorkflowConstants;
 import org.keycloak.utils.StringUtil;
 
 import static org.keycloak.representations.workflows.WorkflowConstants.CONFIG_CONDITIONS;
@@ -77,7 +78,9 @@ public class UserResourceTypeWorkflowProvider implements ResourceTypeSelector {
 
         query.select(userRoot.get("id")).where(predicates);
 
-        return em.createQuery(query).getResultList();
+        int batchSize = Integer.parseInt(workflow.getConfig().getFirstOrDefault(WorkflowConstants.CONFIG_SCHEDULE_BATCH_SIZE, "100"));
+
+        return em.createQuery(query).setMaxResults(batchSize).getResultList();
     }
 
     @Override

services/src/main/java/org/keycloak/workflow/admin/resource/WorkflowResource.java

@@ -76,6 +76,7 @@ public class WorkflowResource {
     })
     public void update(WorkflowRepresentation rep) {
         try {
+            rep.setId(workflow.getId());
             provider.updateWorkflow(workflow, rep);
         } catch (ModelException me) {
             throw ErrorResponse.error(me.getMessage(), Response.Status.BAD_REQUEST);
@@ -152,23 +153,6 @@ public class WorkflowResource {
         provider.activate(workflow, type, resourceId);
     }
 
-    @POST
-    @Path("activate-all")
-    @Tag(name = KeycloakOpenAPI.Admin.Tags.WORKFLOWS)
-    @Operation(summary = "Activate workflow for all eligible resources", description = "Activate the workflow for all eligible resources; an optional notBefore may schedule the first step for all activations.")
-    @APIResponses(value = {
-            @APIResponse(responseCode = "204", description = "No Content"),
-            @APIResponse(responseCode = "400", description = "Bad Request")
-    })
-    public void activateAll(@QueryParam("notBefore") String notBefore) {
-
-        if (notBefore != null) {
-            workflow.setNotBefore(notBefore);
-        }
-
-        provider.activateForAllEligibleResources(workflow);
-    }
-
     /**
      * Deactivate the workflow for the resource.
      *

tests/base/src/test/java/org/keycloak/tests/workflow/WorkflowManagementTest.java

@@ -29,6 +29,7 @@ import jakarta.ws.rs.core.Response.Status;
 
 import org.keycloak.admin.client.Keycloak;
 import org.keycloak.admin.client.resource.BearerAuthFilter;
+import org.keycloak.admin.client.resource.WorkflowResource;
 import org.keycloak.admin.client.resource.WorkflowsResource;
 import org.keycloak.models.workflow.DeleteUserStepProviderFactory;
 import org.keycloak.models.workflow.DisableUserStepProviderFactory;
@@ -477,6 +478,32 @@ public class WorkflowManagementTest extends AbstractWorkflowTest {
 
     }
 
+    @Test
+    public void testUpdateWorkflowNoIdInRepresentation() {
+        WorkflowRepresentation expectedWorkflows = WorkflowRepresentation.withName("test-workflow")
+                .withSteps(
+                        WorkflowStepRepresentation.create().of(DisableUserStepProviderFactory.ID)
+                                .after(Duration.ofDays(5))
+                                .build()
+                ).build();
+
+        WorkflowsResource workflows = managedRealm.admin().workflows();
+
+        String workflowId;
+        try (Response response = workflows.create(expectedWorkflows)) {
+            assertThat(response.getStatus(), is(Response.Status.CREATED.getStatusCode()));
+            workflowId = ApiUtil.getCreatedId(response);
+        }
+
+        WorkflowResource workflow = managedRealm.admin().workflows().workflow(workflowId);
+        WorkflowRepresentation rep = workflow.toRepresentation();
+        rep.setId(null);
+        rep.setConditions(RoleWorkflowConditionFactory.ID + "(realm-management/realm-admin)");
+        try (Response response = workflow.update(rep)) {
+            assertThat(response.getStatus(), is(Status.NO_CONTENT.getStatusCode()));
+        }
+    }
+
     @Test
     public void testSearch() {
         // create a few workflows with different names

tests/base/src/test/java/org/keycloak/tests/workflow/WorkflowScheduleTest.java

@@ -0,0 +1,67 @@
+package org.keycloak.tests.workflow;
+
+import java.time.Duration;
+import java.util.List;
+import java.util.Map;
+
+import org.keycloak.admin.client.resource.UserResource;
+import org.keycloak.models.workflow.SetUserAttributeStepProviderFactory;
+import org.keycloak.representations.workflows.WorkflowRepresentation;
+import org.keycloak.representations.workflows.WorkflowScheduleRepresentation;
+import org.keycloak.representations.workflows.WorkflowStepRepresentation;
+import org.keycloak.testframework.annotations.InjectUser;
+import org.keycloak.testframework.annotations.KeycloakIntegrationTest;
+import org.keycloak.testframework.injection.LifeCycle;
+import org.keycloak.testframework.realm.ManagedUser;
+import org.keycloak.testframework.realm.UserConfig;
+import org.keycloak.testframework.realm.UserConfigBuilder;
+import org.keycloak.tests.workflow.config.WorkflowsBlockingServerConfig;
+
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.Test;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.containsInAnyOrder;
+import static org.hamcrest.Matchers.notNullValue;
+
+@KeycloakIntegrationTest(config = WorkflowsBlockingServerConfig.class)
+public class WorkflowScheduleTest extends AbstractWorkflowTest {
+
+    @InjectUser(ref = "alice", config = DefaultUserConfig.class, lifecycle = LifeCycle.METHOD, realmRef = DEFAULT_REALM_NAME)
+    private ManagedUser userAlice;
+
+    @Test
+    public void testSchedule() {
+        WorkflowRepresentation expectedWorkflow = WorkflowRepresentation.withName("myworkflow")
+                .schedule(WorkflowScheduleRepresentation.create().after("1s").batchSize(10).build())
+                .withSteps(
+                        WorkflowStepRepresentation.create().of(SetUserAttributeStepProviderFactory.ID)
+                                .withConfig("test", "test")
+                                .build()
+                ).build();
+
+        managedRealm.admin().workflows().create(expectedWorkflow).close();
+
+        Awaitility.await()
+                .timeout(Duration.ofSeconds(15))
+                .pollInterval(Duration.ofSeconds(1))
+                .untilAsserted(() -> {
+                    UserResource user = managedRealm.admin().users().get(userAlice.getId());
+                    Map<String, List<String>> attributes = user.getUnmanagedAttributes();
+                    assertThat(attributes, notNullValue());
+                    assertThat(attributes.get("test"), containsInAnyOrder("test"));
+                });
+    }
+
+    private static class DefaultUserConfig implements UserConfig {
+
+        @Override
+        public UserConfigBuilder configure(UserConfigBuilder user) {
+            user.username("alice");
+            user.password("alice");
+            user.name("alice", "alice");
+            user.email("master-admin@email.org");
+            return user;
+        }
+    }
+}

tests/base/src/test/java/org/keycloak/tests/workflow/condition/GroupWorkflowConditionTest.java

@@ -19,6 +19,7 @@ import org.keycloak.models.workflow.conditions.GroupMembershipWorkflowConditionF
 import org.keycloak.representations.idm.UserRepresentation;
 import org.keycloak.representations.userprofile.config.UPConfig;
 import org.keycloak.representations.workflows.WorkflowRepresentation;
+import org.keycloak.representations.workflows.WorkflowScheduleRepresentation;
 import org.keycloak.representations.workflows.WorkflowStepRepresentation;
 import org.keycloak.testframework.annotations.KeycloakIntegrationTest;
 import org.keycloak.testframework.realm.GroupConfigBuilder;
@@ -28,6 +29,7 @@ import org.keycloak.testframework.util.ApiUtil;
 import org.keycloak.tests.workflow.AbstractWorkflowTest;
 import org.keycloak.tests.workflow.config.WorkflowsBlockingServerConfig;
 
+import org.awaitility.Awaitility;
 import org.junit.jupiter.api.Test;
 
 import static org.hamcrest.MatcherAssert.assertThat;
@@ -101,6 +103,7 @@ public class GroupWorkflowConditionTest extends AbstractWorkflowTest {
 
         managedRealm.admin().workflows().create(WorkflowRepresentation.withName("group-membership-workflow")
                 .onCondition(GROUP_CONDITION)
+                .schedule(WorkflowScheduleRepresentation.create().after("1s").build())
                 .withSteps(
                         WorkflowStepRepresentation.create().of(NotifyUserStepProviderFactory.ID)
                                 .after(Duration.ofDays(5))
@@ -112,30 +115,32 @@ public class GroupWorkflowConditionTest extends AbstractWorkflowTest {
 
         List<WorkflowRepresentation> workflows = managedRealm.admin().workflows().list();
         assertThat(workflows, hasSize(1));
-        // activate the workflow for all eligible users
-        managedRealm.admin().workflows().workflow(workflows.get(0).getId()).activateAll();
 
-        runOnServer.run((RunOnServer) session -> {
-            // check the same users are now scheduled to run the second step.
-            WorkflowProvider provider = session.getProvider(WorkflowProvider.class);
-            List<Workflow> registeredWorkflows = provider.getWorkflows().toList();
-            assertThat(registeredWorkflows, hasSize(1));
-            Workflow workflow = registeredWorkflows.get(0);
+        Awaitility.await()
+                .timeout(Duration.ofSeconds(15))
+                .pollInterval(Duration.ofSeconds(1))
+                .untilAsserted(() -> {
+                    runOnServer.run((RunOnServer) session -> {
+                        // check the same users are now scheduled to run the second step.
+                        WorkflowProvider provider = session.getProvider(WorkflowProvider.class);
+                        List<Workflow> registeredWorkflows = provider.getWorkflows().toList();
+                        assertThat(registeredWorkflows, hasSize(1));
+                        Workflow workflow = registeredWorkflows.get(0);
+                        // check workflow was correctly assigned to the users
+                        WorkflowStateProvider stateProvider = session.getProvider(WorkflowStateProvider.class);
+                        RealmModel realm = session.getContext().getRealm();
+                        List<String> scheduledUsers = stateProvider.getScheduledStepsByWorkflow(workflow)
+                                .map(step -> session.users().getUserById(realm, step.resourceId()).getUsername()).toList();
+                        assertThat(scheduledUsers, hasSize(10));
 
-            // check workflow was correctly assigned to the users
-            RealmModel realm = session.getContext().getRealm();
-            WorkflowStateProvider stateProvider = session.getProvider(WorkflowStateProvider.class);
-            List<String> scheduledUsers = stateProvider.getScheduledStepsByWorkflow(workflow)
-                    .map(step -> session.users().getUserById(realm, step.resourceId()).getUsername()).toList();
-            assertThat(scheduledUsers, hasSize(10));
-
-            List<String> expectedUsers = session.users().searchForUserStream(realm, Map.of())
-                    .map(UserModel::getUsername)
-                    .filter(username -> username.startsWith("group-member-"))
-                    .filter(username -> Integer.parseInt(username.substring("group-member-".length())) % 2 == 0)
-                    .toList();
-            assertThat(scheduledUsers, containsInAnyOrder(expectedUsers.toArray()));
-        });
+                        List<String> expectedUsers = session.users().searchForUserStream(realm, Map.of())
+                                .map(UserModel::getUsername)
+                                .filter(username -> username.startsWith("group-member-"))
+                                .filter(username -> Integer.parseInt(username.substring("group-member-".length())) % 2 == 0)
+                                .toList();
+                        assertThat(scheduledUsers, containsInAnyOrder(expectedUsers.toArray()));
+                    });
+                });
     }
 
 }

tests/base/src/test/java/org/keycloak/tests/workflow/condition/IdpLinkConditionWorkflowTest.java

@@ -40,6 +40,7 @@ import org.keycloak.representations.idm.IdentityProviderRepresentation;
 import org.keycloak.representations.idm.UserRepresentation;
 import org.keycloak.representations.userprofile.config.UPConfig;
 import org.keycloak.representations.workflows.WorkflowRepresentation;
+import org.keycloak.representations.workflows.WorkflowScheduleRepresentation;
 import org.keycloak.representations.workflows.WorkflowStepRepresentation;
 import org.keycloak.testframework.annotations.KeycloakIntegrationTest;
 import org.keycloak.testframework.realm.UserConfigBuilder;
@@ -49,6 +50,7 @@ import org.keycloak.tests.workflow.AbstractWorkflowTest;
 import org.keycloak.tests.workflow.config.WorkflowsBlockingServerConfig;
 import org.keycloak.testsuite.util.IdentityProviderBuilder;
 
+import org.awaitility.Awaitility;
 import org.junit.jupiter.api.Test;
 
 import static org.keycloak.models.workflow.ResourceOperationType.USER_CREATED;
@@ -133,6 +135,7 @@ public class IdpLinkConditionWorkflowTest extends AbstractWorkflowTest {
         managedRealm.admin().workflows().create(WorkflowRepresentation.withName("myworkflow")
                 .onEvent(ResourceOperationType.USER_FEDERATED_IDENTITY_ADDED.name())
                 .onCondition(IdentityProviderWorkflowConditionFactory.ID + "(" + IDP_OIDC_ALIAS + ")")
+                .schedule(WorkflowScheduleRepresentation.create().after("1s").build())
                 .withSteps(
                         WorkflowStepRepresentation.create().of(NotifyUserStepProviderFactory.ID)
                                 .after(Duration.ofDays(5))
@@ -200,43 +203,46 @@ public class IdpLinkConditionWorkflowTest extends AbstractWorkflowTest {
 
         List<WorkflowRepresentation> workflows = managedRealm.admin().workflows().list();
         assertThat(workflows, hasSize(1));
-        // activate the workflow for all eligible users - i.e. only users from the same idp who are not yet assigned to the workflow.
-        managedRealm.admin().workflows().workflow(workflows.get(0).getId()).activateAll();
 
-        runOnServer.run((RunOnServer) session -> {
-            RealmModel realm = session.getContext().getRealm();
-            WorkflowProvider provider = session.getProvider(WorkflowProvider.class);
-            List<Workflow> registeredWorkflows = provider.getWorkflows().toList();
-            assertEquals(1, registeredWorkflows.size());
-            Workflow workflow = registeredWorkflows.get(0);
+        Awaitility.await()
+                .timeout(Duration.ofSeconds(15))
+                .pollInterval(Duration.ofSeconds(1))
+                .untilAsserted(() -> {
+                    runOnServer.run((RunOnServer) session -> {
+                        RealmModel realm = session.getContext().getRealm();
+                        // check the same users are now scheduled to run the second step.
+                        WorkflowProvider provider = session.getProvider(WorkflowProvider.class);
+                        List<Workflow> registeredWorkflows = provider.getWorkflows().toList();
+                        assertEquals(1, registeredWorkflows.size());
+                        Workflow workflow = registeredWorkflows.get(0);
+                        // check workflow was correctly assigned to the old users, not affecting users already associated with the workflow.
+                        WorkflowStateProvider stateProvider = session.getProvider(WorkflowStateProvider.class);
+                        List<ScheduledStep> scheduledSteps = stateProvider.getScheduledStepsByWorkflow(workflow).toList();
+                        assertEquals(13, scheduledSteps.size());
 
-            // check workflow was correctly assigned to the old users, not affecting users already associated with the workflow.
-            WorkflowStateProvider stateProvider = session.getProvider(WorkflowStateProvider.class);
-            List<ScheduledStep> scheduledSteps = stateProvider.getScheduledStepsByWorkflow(workflow).toList();
-            assertEquals(13, scheduledSteps.size()); // total users now associated with the workflow
+                        List<WorkflowStep> steps = workflow.getSteps().toList();
+                        assertEquals(2, steps.size());
+                        WorkflowStep notifyStep = steps.get(0);
+                        List<ScheduledStep> scheduledToNotify = scheduledSteps.stream()
+                                .filter(step -> notifyStep.getId().equals(step.stepId())).toList();
+                        assertEquals(10, scheduledToNotify.size());
+                        scheduledToNotify.forEach(scheduledStep -> {
+                            UserModel user = session.users().getUserById(realm, scheduledStep.resourceId());
+                            assertNotNull(user);
+                            assertTrue(user.getUsername().startsWith("idp-user-"));
+                        });
 
-            List<WorkflowStep> steps = workflow.getSteps().toList();
-            assertEquals(2, steps.size());
-            WorkflowStep notifyStep = steps.get(0);
-            List<ScheduledStep> scheduledToNotify = scheduledSteps.stream()
-                    .filter(step -> notifyStep.getId().equals(step.stepId())).toList();
-            assertEquals(10, scheduledToNotify.size()); // only the 10 old users should be assigned to the first step
-            scheduledToNotify.forEach(scheduledStep -> {
-                UserModel user = session.users().getUserById(realm, scheduledStep.resourceId());
-                assertNotNull(user);
-                assertTrue(user.getUsername().startsWith("idp-user-"));
-            });
-
-            WorkflowStep disableStep = workflow.getSteps().toList().get(1);
-            List<ScheduledStep> scheduledToDisable = scheduledSteps.stream()
-                    .filter(step -> disableStep.getId().equals(step.stepId())).toList();
-            assertEquals(3, scheduledToDisable.size()); // the 3 "new" users should still be at the disable step
-            scheduledToDisable.forEach(scheduledStep -> {
-                UserModel user = session.users().getUserById(realm, scheduledStep.resourceId());
-                assertNotNull(user);
-                assertTrue(user.getUsername().startsWith("new-idp-user-"));
-            });
-        });
+                        WorkflowStep disableStep = workflow.getSteps().toList().get(1);
+                        List<ScheduledStep> scheduledToDisable = scheduledSteps.stream()
+                                .filter(step -> disableStep.getId().equals(step.stepId())).toList();
+                        assertEquals(3, scheduledToDisable.size());
+                        scheduledToDisable.forEach(scheduledStep -> {
+                            UserModel user = session.users().getUserById(realm, scheduledStep.resourceId());
+                            assertNotNull(user);
+                            assertTrue(user.getUsername().startsWith("new-idp-user-"));
+                        });
+                    });
+                });
     }
 
     private void setupIdentityProvider() {

tests/base/src/test/java/org/keycloak/tests/workflow/condition/RoleWorkflowConditionTest.java

@@ -25,6 +25,7 @@ import org.keycloak.representations.idm.RoleRepresentation;
 import org.keycloak.representations.userprofile.config.UPConfig;
 import org.keycloak.representations.userprofile.config.UPConfig.UnmanagedAttributePolicy;
 import org.keycloak.representations.workflows.WorkflowRepresentation;
+import org.keycloak.representations.workflows.WorkflowScheduleRepresentation;
 import org.keycloak.representations.workflows.WorkflowStepRepresentation;
 import org.keycloak.testframework.annotations.KeycloakIntegrationTest;
 import org.keycloak.testframework.realm.RoleConfigBuilder;
@@ -34,6 +35,7 @@ import org.keycloak.testframework.util.ApiUtil;
 import org.keycloak.tests.workflow.AbstractWorkflowTest;
 import org.keycloak.tests.workflow.config.WorkflowsBlockingServerConfig;
 
+import org.awaitility.Awaitility;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
@@ -86,7 +88,7 @@ public class RoleWorkflowConditionTest extends AbstractWorkflowTest {
         }
 
         managedRealm.admin().workflows().create(WorkflowRepresentation.withName("test-role-workflow")
-                .onEvent(ResourceOperationType.USER_ROLE_GRANTED.name())
+                .schedule(WorkflowScheduleRepresentation.create().after("1s").build())
                 .onCondition(RoleWorkflowConditionFactory.ID + "(testRole)")
                 .withSteps(
                         WorkflowStepRepresentation.create().of(NotifyUserStepProviderFactory.ID)
@@ -99,20 +101,23 @@ public class RoleWorkflowConditionTest extends AbstractWorkflowTest {
 
         List<WorkflowRepresentation> workflows = managedRealm.admin().workflows().list();
         assertThat(workflows, hasSize(1));
-        // activate the workflow for all eligible users
-        managedRealm.admin().workflows().workflow(workflows.get(0).getId()).activateAll();
 
-        runOnServer.run((RunOnServer) session -> {
-            // check the same users are now scheduled to run the second step.
-            WorkflowProvider provider = session.getProvider(WorkflowProvider.class);
-            List<Workflow> registeredWorkflows = provider.getWorkflows().toList();
-            assertThat(registeredWorkflows, hasSize(1));
-            Workflow workflow = registeredWorkflows.get(0);
-            // check workflow was correctly assigned to the users
-            WorkflowStateProvider stateProvider = session.getProvider(WorkflowStateProvider.class);
-            List<ScheduledStep> scheduledSteps = stateProvider.getScheduledStepsByWorkflow(workflow).toList();
-            assertThat(scheduledSteps, hasSize(10));
-        });
+        Awaitility.await()
+                .timeout(Duration.ofSeconds(15))
+                .pollInterval(Duration.ofSeconds(1))
+                .untilAsserted(() -> {
+                    runOnServer.run((RunOnServer) session -> {
+                        // check the same users are now scheduled to run the second step.
+                        WorkflowProvider provider = session.getProvider(WorkflowProvider.class);
+                        List<Workflow> registeredWorkflows = provider.getWorkflows().toList();
+                        assertThat(registeredWorkflows, hasSize(1));
+                        Workflow workflow = registeredWorkflows.get(0);
+                        // check workflow was correctly assigned to the users
+                        WorkflowStateProvider stateProvider = session.getProvider(WorkflowStateProvider.class);
+                        List<ScheduledStep> scheduledSteps = stateProvider.getScheduledStepsByWorkflow(workflow).toList();
+                        assertThat(scheduledSteps, hasSize(10));
+                    });
+                });
     }
 
     private void assertUserRoles(String username, boolean shouldExist, String... roles) {

tests/base/src/test/java/org/keycloak/tests/workflow/condition/UserAttributeWorkflowConditionTest.java

@@ -22,6 +22,7 @@ import org.keycloak.models.workflow.conditions.UserAttributeWorkflowConditionFac
 import org.keycloak.representations.userprofile.config.UPConfig;
 import org.keycloak.representations.userprofile.config.UPConfig.UnmanagedAttributePolicy;
 import org.keycloak.representations.workflows.WorkflowRepresentation;
+import org.keycloak.representations.workflows.WorkflowScheduleRepresentation;
 import org.keycloak.representations.workflows.WorkflowStepRepresentation;
 import org.keycloak.testframework.annotations.KeycloakIntegrationTest;
 import org.keycloak.testframework.realm.UserConfigBuilder;
@@ -29,6 +30,7 @@ import org.keycloak.testframework.remote.providers.runonserver.RunOnServer;
 import org.keycloak.tests.workflow.AbstractWorkflowTest;
 import org.keycloak.tests.workflow.config.WorkflowsBlockingServerConfig;
 
+import org.awaitility.Awaitility;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
@@ -94,20 +96,23 @@ public class UserAttributeWorkflowConditionTest extends AbstractWorkflowTest {
 
         List<WorkflowRepresentation> workflows = managedRealm.admin().workflows().list();
         assertThat(workflows, hasSize(1));
-        // activate the workflow for all eligible users
-        managedRealm.admin().workflows().workflow(workflows.get(0).getId()).activateAll();
 
-        runOnServer.run((RunOnServer) session -> {
-            // check the same users are now scheduled to run the second step.
-            WorkflowProvider provider = session.getProvider(WorkflowProvider.class);
-            List<Workflow> registeredWorkflows = provider.getWorkflows().toList();
-            assertThat(registeredWorkflows, hasSize(1));
-            Workflow workflow = registeredWorkflows.get(0);
-            // check workflow was correctly assigned to the users
-            WorkflowStateProvider stateProvider = session.getProvider(WorkflowStateProvider.class);
-            List<ScheduledStep> scheduledSteps = stateProvider.getScheduledStepsByWorkflow(workflow).toList();
-            assertThat(scheduledSteps, hasSize(10));
-        });
+        Awaitility.await()
+                .timeout(Duration.ofSeconds(15))
+                .pollInterval(Duration.ofSeconds(1))
+                .untilAsserted(() -> {
+                    runOnServer.run((RunOnServer) session -> {
+                        // check the same users are now scheduled to run the second step.
+                        WorkflowProvider provider = session.getProvider(WorkflowProvider.class);
+                        List<Workflow> registeredWorkflows = provider.getWorkflows().toList();
+                        assertThat(registeredWorkflows, hasSize(1));
+                        Workflow workflow = registeredWorkflows.get(0);
+                        // check workflow was correctly assigned to the users
+                        WorkflowStateProvider stateProvider = session.getProvider(WorkflowStateProvider.class);
+                        List<ScheduledStep> scheduledSteps = stateProvider.getScheduledStepsByWorkflow(workflow).toList();
+                        assertThat(scheduledSteps, hasSize(10));
+                    });
+                });
     }
 
     private void assertUserAttribute(String username, boolean shouldExist, String... values) {
@@ -157,6 +162,7 @@ public class UserAttributeWorkflowConditionTest extends AbstractWorkflowTest {
 
         WorkflowRepresentation expectedWorkflow = WorkflowRepresentation.withName("myworkflow")
                 .onEvent(ResourceOperationType.USER_CREATED.name())
+                .schedule(WorkflowScheduleRepresentation.create().after("1s").build())
                 .onCondition(attributeCondition)
                 .withSteps(
                         WorkflowStepRepresentation.create()