From 7004aa6850320d32323939afb67a31d8fb97804f Mon Sep 17 00:00:00 2001
From: Yufei Cai <yufei.cai@bosch.io>
Date: Thu, 14 Jan 2021 19:32:42 +0100
Subject: [PATCH] [#926] Add generic TopLevelActionCommand for policies.

Signed-off-by: Yufei Cai <yufei.cai@bosch.io>
---
 .../enforcement/PolicyCommandEnforcement.java |  31 +-
 .../PolicyCommandEnforcementTest.java         |  39 ++-
 .../routes/policies/PoliciesRoute.java        |  18 +-
 .../routes/policies/PoliciesRouteTest.java    |  25 +-
 .../AbstractPolicyActionCommandStrategy.java  |  12 +-
 .../AbstractPolicyCommandStrategy.java        |   7 +-
 .../AbstractPolicyQueryCommandStrategy.java   |   3 +-
 ...ctivatePolicyTokenIntegrationStrategy.java |   8 +-
 .../ActivateTokenIntegrationStrategy.java     |  13 +-
 .../commands/CreatePolicyStrategy.java        |   2 +-
 ...ctivatePolicyTokenIntegrationStrategy.java |   6 +-
 .../DeactivateTokenIntegrationStrategy.java   |   9 +-
 .../commands/DeletePolicyEntryStrategy.java   |   2 +-
 .../commands/DeletePolicyStrategy.java        |   2 +-
 .../commands/DeleteResourceStrategy.java      |   2 +-
 .../commands/DeleteSubjectStrategy.java       |   2 +-
 .../commands/ModifyPolicyEntriesStrategy.java |   7 +-
 .../commands/ModifyPolicyEntryStrategy.java   |   2 +-
 .../commands/ModifyPolicyStrategy.java        |   2 +-
 .../commands/ModifyResourceStrategy.java      |   2 +-
 .../commands/ModifyResourcesStrategy.java     |   2 +-
 .../commands/ModifySubjectStrategy.java       |   2 +-
 .../commands/ModifySubjectsStrategy.java      |   2 +-
 .../commands/PolicyCommandStrategies.java     |   9 +-
 .../commands/PolicyConflictStrategy.java      |   2 +-
 .../TopLevelActionCommandStrategy.java        | 180 +++++++++++
 .../AbstractPolicyCommandStrategyTest.java    |  45 ++-
 ...atePolicyTokenIntegrationStrategyTest.java |   6 +-
 .../ActivateTokenIntegrationStrategyTest.java |   2 +-
 .../TopLevelActionCommandStrategyTest.java    | 304 ++++++++++++++++++
 .../commands/AbstractCommandStrategies.java   |  24 +-
 .../commands/CommandStrategy.java             |  23 ++
 .../ActivatePolicyTokenIntegration.java       |  11 +
 .../actions/ActivateTokenIntegration.java     |  23 ++
 .../DeactivatePolicyTokenIntegration.java     |  11 +
 .../actions/DeactivateTokenIntegration.java   |  11 +
 .../policies/actions/PolicyActionCommand.java |  18 ++
 .../actions/TopLevelActionCommand.java        | 233 ++++++++++++++
 .../TopLevelActionCommandResponse.java        | 147 +++++++++
 .../PolicyActionFailedException.java          |   9 +
 ...licyCommandToActionsExceptionRegistry.java |  14 +-
 .../TopLevelActionCommandResponseTest.java    |  81 +++++
 .../actions/TopLevelActionCommandTest.java    |  98 ++++++
 .../policies/AbstractPolicyActionEvent.java   |  96 ++++++
 .../events/policies/PolicyActionEvent.java    |  32 ++
 .../events/policies/SubjectCreated.java       |  33 +-
 .../events/policies/SubjectDeleted.java       |  30 +-
 .../events/policies/SubjectModified.java      |  34 +-
 .../policies/SubjectsDeletedPartially.java    |  11 +-
 .../policies/SubjectsModifiedPartially.java   |   8 +-
 50 files changed, 1525 insertions(+), 170 deletions(-)
 create mode 100644 services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/TopLevelActionCommandStrategy.java
 create mode 100644 services/policies/persistence/src/test/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/TopLevelActionCommandStrategyTest.java
 create mode 100755 signals/commands/policies/src/main/java/org/eclipse/ditto/signals/commands/policies/actions/TopLevelActionCommand.java
 create mode 100755 signals/commands/policies/src/main/java/org/eclipse/ditto/signals/commands/policies/actions/TopLevelActionCommandResponse.java
 create mode 100755 signals/commands/policies/src/test/java/org/eclipse/ditto/signals/commands/policies/actions/TopLevelActionCommandResponseTest.java
 create mode 100755 signals/commands/policies/src/test/java/org/eclipse/ditto/signals/commands/policies/actions/TopLevelActionCommandTest.java
 create mode 100755 signals/events/policies/src/main/java/org/eclipse/ditto/signals/events/policies/AbstractPolicyActionEvent.java
 create mode 100755 signals/events/policies/src/main/java/org/eclipse/ditto/signals/events/policies/PolicyActionEvent.java

diff --git a/services/concierge/enforcement/src/main/java/org/eclipse/ditto/services/concierge/enforcement/PolicyCommandEnforcement.java b/services/concierge/enforcement/src/main/java/org/eclipse/ditto/services/concierge/enforcement/PolicyCommandEnforcement.java
index 88b444732b..56bdb08d93 100644
--- a/services/concierge/enforcement/src/main/java/org/eclipse/ditto/services/concierge/enforcement/PolicyCommandEnforcement.java
+++ b/services/concierge/enforcement/src/main/java/org/eclipse/ditto/services/concierge/enforcement/PolicyCommandEnforcement.java
@@ -50,9 +50,8 @@
 import org.eclipse.ditto.services.utils.cluster.DistPubSubAccess;
 import org.eclipse.ditto.signals.commands.base.CommandToExceptionRegistry;
 import org.eclipse.ditto.signals.commands.policies.PolicyCommand;
-import org.eclipse.ditto.signals.commands.policies.actions.ActivatePolicyTokenIntegration;
-import org.eclipse.ditto.signals.commands.policies.actions.DeactivatePolicyTokenIntegration;
 import org.eclipse.ditto.signals.commands.policies.actions.PolicyActionCommand;
+import org.eclipse.ditto.signals.commands.policies.actions.TopLevelActionCommand;
 import org.eclipse.ditto.signals.commands.policies.exceptions.PolicyCommandToAccessExceptionRegistry;
 import org.eclipse.ditto.signals.commands.policies.exceptions.PolicyCommandToActionsExceptionRegistry;
 import org.eclipse.ditto.signals.commands.policies.exceptions.PolicyCommandToModifyExceptionRegistry;
@@ -131,11 +130,13 @@ public static <T extends PolicyCommand<?>> Optional<T> authorizePolicyCommand(fi
         return authorizedCommand;
     }
 
+    @SuppressWarnings("unchecked")
     private static <T extends PolicyCommand<?>> Optional<T> authorizeActionCommand(final PolicyEnforcer enforcer,
             final T command, final ResourceKey resourceKey, final AuthorizationContext authorizationContext) {
 
-        if (isTopLevelActionCommand(command)) {
-            return authorizeTopLevelAction(enforcer, command, authorizationContext);
+        if (command instanceof TopLevelActionCommand) {
+            final TopLevelActionCommand topLevelActionCommand = (TopLevelActionCommand) command;
+            return (Optional<T>) authorizeTopLevelAction(enforcer, topLevelActionCommand, authorizationContext);
         } else {
             return authorizeEntryLevelAction(enforcer.getEnforcer(), command, resourceKey, authorizationContext);
         }
@@ -152,9 +153,8 @@ private static <T extends PolicyCommand<?>> Optional<T> authorizeEntryLevelActio
                 : Optional.empty();
     }
 
-    @SuppressWarnings("unchecked")
-    private static <T extends PolicyCommand<?>> Optional<T> authorizeTopLevelAction(final PolicyEnforcer policyEnforcer,
-            final T command, final AuthorizationContext authorizationContext) {
+    private static Optional<TopLevelActionCommand> authorizeTopLevelAction(final PolicyEnforcer policyEnforcer,
+            final TopLevelActionCommand command, final AuthorizationContext authorizationContext) {
         final Enforcer enforcer = policyEnforcer.getEnforcer();
         final List<Label> authorizedLabels = policyEnforcer.getPolicy()
                 .map(policy -> policy.getEntriesSet().stream()
@@ -165,21 +165,10 @@ private static <T extends PolicyCommand<?>> Optional<T> authorizeTopLevelAction(
                 .orElse(List.of());
         if (authorizedLabels.isEmpty()) {
             return Optional.empty();
-        } else if (command instanceof ActivatePolicyTokenIntegration) {
-            final ActivatePolicyTokenIntegration c = (ActivatePolicyTokenIntegration) command;
-            final T adjustedCommand =
-                    (T) ActivatePolicyTokenIntegration.of(c.getEntityId(), c.getSubjectId(), c.getExpiry(),
-                            authorizedLabels,
-                            c.getDittoHeaders());
-            return Optional.of(adjustedCommand);
-        } else if (command instanceof DeactivatePolicyTokenIntegration) {
-            final DeactivatePolicyTokenIntegration c = (DeactivatePolicyTokenIntegration) command;
-            final T adjustedCommand =
-                    (T) DeactivatePolicyTokenIntegration.of(c.getEntityId(), c.getSubjectId(), authorizedLabels,
-                            c.getDittoHeaders());
-            return Optional.of(adjustedCommand);
         } else {
-            return Optional.empty();
+            final TopLevelActionCommand adjustedCommand =
+                    TopLevelActionCommand.of(command.getPolicyActionCommand(), authorizedLabels);
+            return Optional.of(adjustedCommand);
         }
     }
 
diff --git a/services/concierge/enforcement/src/test/java/org/eclipse/ditto/services/concierge/enforcement/PolicyCommandEnforcementTest.java b/services/concierge/enforcement/src/test/java/org/eclipse/ditto/services/concierge/enforcement/PolicyCommandEnforcementTest.java
index ef2daeafbf..85a4b64615 100644
--- a/services/concierge/enforcement/src/test/java/org/eclipse/ditto/services/concierge/enforcement/PolicyCommandEnforcementTest.java
+++ b/services/concierge/enforcement/src/test/java/org/eclipse/ditto/services/concierge/enforcement/PolicyCommandEnforcementTest.java
@@ -62,6 +62,7 @@
 import org.eclipse.ditto.signals.commands.policies.actions.ActivateTokenIntegration;
 import org.eclipse.ditto.signals.commands.policies.actions.DeactivatePolicyTokenIntegration;
 import org.eclipse.ditto.signals.commands.policies.actions.DeactivateTokenIntegration;
+import org.eclipse.ditto.signals.commands.policies.actions.TopLevelActionCommand;
 import org.eclipse.ditto.signals.commands.policies.exceptions.PolicyActionFailedException;
 import org.eclipse.ditto.signals.commands.policies.exceptions.PolicyNotAccessibleException;
 import org.eclipse.ditto.signals.commands.policies.exceptions.PolicyNotModifiableException;
@@ -625,19 +626,22 @@ public void activatePolicyTokenIntegration() {
         new TestKit(system) {{
             final SubjectId subjectId = SubjectId.newInstance("issuer:{{policy-entry:label}}:subject");
             final Instant expiry = Instant.now();
-            final ActivatePolicyTokenIntegration activatePolicyTokenIntegration =
-                    ActivatePolicyTokenIntegration.of(POLICY_ID, subjectId, expiry, List.of(), DITTO_HEADERS);
+            final TopLevelActionCommand command = TopLevelActionCommand.of(
+                    ActivateTokenIntegration.of(POLICY_ID, Label.of("-"), subjectId, expiry, DITTO_HEADERS),
+                    List.of()
+            );
 
-            enforcer.tell(activatePolicyTokenIntegration, getRef());
+            enforcer.tell(command, getRef());
 
             policiesShardRegionProbe.expectMsgClass(SudoRetrievePolicy.class);
             policiesShardRegionProbe.reply(createPolicyResponseForActions());
 
-            final ActivatePolicyTokenIntegration
-                    forwarded = policiesShardRegionProbe.expectMsgClass(ActivatePolicyTokenIntegration.class);
-            assertThat(forwarded).isEqualTo(
-                    ActivatePolicyTokenIntegration.of(POLICY_ID, subjectId, expiry, List.of(Label.of("allowed")),
-                            DITTO_HEADERS));
+            final TopLevelActionCommand
+                    forwarded = policiesShardRegionProbe.expectMsgClass(TopLevelActionCommand.class);
+            assertThat(forwarded).isEqualTo(TopLevelActionCommand.of(
+                    ActivateTokenIntegration.of(POLICY_ID, Label.of("-"), subjectId, expiry, DITTO_HEADERS),
+                    List.of(Label.of("allowed"))
+            ));
         }};
     }
 
@@ -645,19 +649,22 @@ public void activatePolicyTokenIntegration() {
     public void deactivatePolicyTokenIntegration() {
         new TestKit(system) {{
             final SubjectId subjectId = SubjectId.newInstance("issuer:{{policy-entry:label}}:subject");
-            final DeactivatePolicyTokenIntegration deactivatePolicyTokenIntegration =
-                    DeactivatePolicyTokenIntegration.of(POLICY_ID, subjectId, List.of(), DITTO_HEADERS);
+            final TopLevelActionCommand command = TopLevelActionCommand.of(
+                    DeactivateTokenIntegration.of(POLICY_ID, Label.of("-"), subjectId, DITTO_HEADERS),
+                    List.of()
+            );
 
-            enforcer.tell(deactivatePolicyTokenIntegration, getRef());
+            enforcer.tell(command, getRef());
 
             policiesShardRegionProbe.expectMsgClass(SudoRetrievePolicy.class);
             policiesShardRegionProbe.reply(createPolicyResponseForActions());
 
-            final DeactivatePolicyTokenIntegration
-                    forwarded = policiesShardRegionProbe.expectMsgClass(DeactivatePolicyTokenIntegration.class);
-            assertThat(forwarded).isEqualTo(
-                    DeactivatePolicyTokenIntegration.of(POLICY_ID, subjectId, List.of(Label.of("allowed")),
-                            DITTO_HEADERS));
+            final TopLevelActionCommand
+                    forwarded = policiesShardRegionProbe.expectMsgClass(TopLevelActionCommand.class);
+            assertThat(forwarded).isEqualTo(TopLevelActionCommand.of(
+                    DeactivateTokenIntegration.of(POLICY_ID, Label.of("-"), subjectId, DITTO_HEADERS),
+                    List.of(Label.of("allowed"))
+            ));
         }};
     }
 
diff --git a/services/gateway/endpoints/src/main/java/org/eclipse/ditto/services/gateway/endpoints/routes/policies/PoliciesRoute.java b/services/gateway/endpoints/src/main/java/org/eclipse/ditto/services/gateway/endpoints/routes/policies/PoliciesRoute.java
index 77025328b5..888341ae33 100755
--- a/services/gateway/endpoints/src/main/java/org/eclipse/ditto/services/gateway/endpoints/routes/policies/PoliciesRoute.java
+++ b/services/gateway/endpoints/src/main/java/org/eclipse/ditto/services/gateway/endpoints/routes/policies/PoliciesRoute.java
@@ -24,6 +24,7 @@
 import org.eclipse.ditto.json.JsonValue;
 import org.eclipse.ditto.model.base.headers.DittoHeaders;
 import org.eclipse.ditto.model.jwt.JsonWebToken;
+import org.eclipse.ditto.model.policies.Label;
 import org.eclipse.ditto.model.policies.PoliciesModelFactory;
 import org.eclipse.ditto.model.policies.Policy;
 import org.eclipse.ditto.model.policies.PolicyId;
@@ -34,10 +35,9 @@
 import org.eclipse.ditto.services.gateway.security.authentication.jwt.JwtAuthenticationResult;
 import org.eclipse.ditto.services.gateway.util.config.endpoints.CommandConfig;
 import org.eclipse.ditto.services.gateway.util.config.endpoints.HttpConfig;
-import org.eclipse.ditto.signals.commands.policies.actions.ActivatePolicyTokenIntegration;
 import org.eclipse.ditto.signals.commands.policies.actions.ActivateTokenIntegration;
-import org.eclipse.ditto.signals.commands.policies.actions.DeactivatePolicyTokenIntegration;
 import org.eclipse.ditto.signals.commands.policies.actions.DeactivateTokenIntegration;
+import org.eclipse.ditto.signals.commands.policies.actions.TopLevelActionCommand;
 import org.eclipse.ditto.signals.commands.policies.exceptions.PolicyActionFailedException;
 import org.eclipse.ditto.signals.commands.policies.exceptions.PolicyIdNotExplicitlySettableException;
 import org.eclipse.ditto.signals.commands.policies.modify.DeletePolicy;
@@ -62,6 +62,8 @@ public final class PoliciesRoute extends AbstractRoute {
     private static final String PATH_POLICIES = "policies";
     private static final String PATH_ENTRIES = "entries";
 
+    private static final Label DUMMY_LABEL = Label.of("-");
+
     private final PolicyEntriesRoute policyEntriesRoute;
     private final TokenIntegrationSubjectIdFactory tokenIntegrationSubjectIdFactory;
 
@@ -187,19 +189,23 @@ private Route policyActions(final RequestContext ctx, final DittoHeaders dittoHe
         ));
     }
 
-    private ActivatePolicyTokenIntegration activatePolicyTokenIntegration(final DittoHeaders dittoHeaders,
+    private TopLevelActionCommand activatePolicyTokenIntegration(final DittoHeaders dittoHeaders,
             final PolicyId policyId, final JsonWebToken jwt) {
 
         final SubjectId subjectId = tokenIntegrationSubjectIdFactory.getSubjectId(dittoHeaders, jwt);
         final Instant expiry = jwt.getExpirationTime();
-        return ActivatePolicyTokenIntegration.of(policyId, subjectId, expiry, List.of(), dittoHeaders);
+        final ActivateTokenIntegration activateTokenIntegration =
+                ActivateTokenIntegration.of(policyId, DUMMY_LABEL, subjectId, expiry, dittoHeaders);
+        return TopLevelActionCommand.of(activateTokenIntegration, List.of());
     }
 
-    private DeactivatePolicyTokenIntegration deactivatePolicyTokenIntegration(final DittoHeaders dittoHeaders,
+    private TopLevelActionCommand deactivatePolicyTokenIntegration(final DittoHeaders dittoHeaders,
             final PolicyId policyId, final JsonWebToken jwt) {
 
         final SubjectId subjectId = tokenIntegrationSubjectIdFactory.getSubjectId(dittoHeaders, jwt);
-        return DeactivatePolicyTokenIntegration.of(policyId, subjectId, List.of(), dittoHeaders);
+        final DeactivateTokenIntegration deactivateTokenIntegration =
+                DeactivateTokenIntegration.of(policyId, DUMMY_LABEL, subjectId, dittoHeaders);
+        return TopLevelActionCommand.of(deactivateTokenIntegration, List.of());
     }
 
     static Route extractJwt(final DittoHeaders dittoHeaders,
diff --git a/services/gateway/endpoints/src/test/java/org/eclipse/ditto/services/gateway/endpoints/routes/policies/PoliciesRouteTest.java b/services/gateway/endpoints/src/test/java/org/eclipse/ditto/services/gateway/endpoints/routes/policies/PoliciesRouteTest.java
index 0a3c97885b..21003cb9ce 100644
--- a/services/gateway/endpoints/src/test/java/org/eclipse/ditto/services/gateway/endpoints/routes/policies/PoliciesRouteTest.java
+++ b/services/gateway/endpoints/src/test/java/org/eclipse/ditto/services/gateway/endpoints/routes/policies/PoliciesRouteTest.java
@@ -28,10 +28,9 @@
 import org.eclipse.ditto.services.gateway.security.authentication.DefaultAuthenticationResult;
 import org.eclipse.ditto.services.gateway.security.authentication.jwt.JwtAuthenticationResult;
 import org.eclipse.ditto.services.utils.protocol.ProtocolAdapterProvider;
-import org.eclipse.ditto.signals.commands.policies.actions.ActivatePolicyTokenIntegration;
 import org.eclipse.ditto.signals.commands.policies.actions.ActivateTokenIntegration;
-import org.eclipse.ditto.signals.commands.policies.actions.DeactivatePolicyTokenIntegration;
 import org.eclipse.ditto.signals.commands.policies.actions.DeactivateTokenIntegration;
+import org.eclipse.ditto.signals.commands.policies.actions.TopLevelActionCommand;
 import org.eclipse.ditto.signals.commands.policies.exceptions.PolicyActionFailedException;
 import org.junit.Before;
 import org.junit.Rule;
@@ -150,11 +149,13 @@ public void nonexistentPolicyAction() {
     public void activateTokenIntegration() {
         getRoute(getTokenAuthResult()).run(HttpRequest.POST("/policies/ns%3An/actions/activateTokenIntegration/"))
                 .assertStatusCode(StatusCodes.OK)
-                .assertEntity(ActivatePolicyTokenIntegration.of(PolicyId.of("ns:n"),
-                        SubjectId.newInstance("dummy-issuer:{{policy-entry:label}}:dummy-subject"),
-                        DummyJwt.EXPIRY,
-                        List.of(),
-                        DittoHeaders.empty()
+                .assertEntity(TopLevelActionCommand.of(
+                        ActivateTokenIntegration.of(PolicyId.of("ns:n"),
+                                Label.of("-"),
+                                SubjectId.newInstance("dummy-issuer:{{policy-entry:label}}:dummy-subject"),
+                                DummyJwt.EXPIRY,
+                                DittoHeaders.empty()),
+                        List.of()
                 ).toJsonString());
     }
 
@@ -162,10 +163,12 @@ public void activateTokenIntegration() {
     public void deactivateTokenIntegration() {
         getRoute(getTokenAuthResult()).run(HttpRequest.POST("/policies/ns%3An/actions/deactivateTokenIntegration"))
                 .assertStatusCode(StatusCodes.OK)
-                .assertEntity(DeactivatePolicyTokenIntegration.of(PolicyId.of("ns:n"),
-                        SubjectId.newInstance("dummy-issuer:{{policy-entry:label}}:dummy-subject"),
-                        List.of(),
-                        DittoHeaders.empty()
+                .assertEntity(TopLevelActionCommand.of(
+                        DeactivateTokenIntegration.of(PolicyId.of("ns:n"),
+                                Label.of("-"),
+                                SubjectId.newInstance("dummy-issuer:{{policy-entry:label}}:dummy-subject"),
+                                DittoHeaders.empty()),
+                        List.of()
                 ).toJsonString());
     }
 
diff --git a/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/AbstractPolicyActionCommandStrategy.java b/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/AbstractPolicyActionCommandStrategy.java
index 2a70b9b3a7..108875451c 100644
--- a/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/AbstractPolicyActionCommandStrategy.java
+++ b/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/AbstractPolicyActionCommandStrategy.java
@@ -22,6 +22,7 @@
 import org.eclipse.ditto.services.utils.akka.AkkaClassLoader;
 import org.eclipse.ditto.signals.commands.policies.actions.PolicyActionCommand;
 import org.eclipse.ditto.signals.commands.policies.exceptions.PolicyActionFailedException;
+import org.eclipse.ditto.signals.events.policies.PolicyActionEvent;
 
 import akka.actor.ActorSystem;
 
@@ -31,7 +32,7 @@
  * @param <C> the type of the handled command
  */
 abstract class AbstractPolicyActionCommandStrategy<C extends PolicyActionCommand<C>>
-        extends AbstractPolicyCommandStrategy<C> {
+        extends AbstractPolicyCommandStrategy<C, PolicyActionEvent<?>> {
 
     protected final SubjectIdFromActionResolver subjectIdFromActionResolver;
 
@@ -61,7 +62,14 @@ boolean containsThingReadPermission(final PolicyEntry policyEntry) {
                 });
     }
 
-    PolicyActionFailedException getExceptionForNoEntryWithThingReadPermission() {
+    /**
+     * Get the exception for when a policy action command is not applicable to the designated policy entry.
+     * For now there is 1 not-applicable exception defined here.
+     * New policy action command strategies may override this method to provide their own exceptions.
+     *
+     * @return the exception for when a policy action command is not applicable.
+     */
+    PolicyActionFailedException getNotApplicableException() {
         return PolicyActionFailedException.newBuilderForActivateTokenIntegration()
                 .status(HttpStatusCode.NOT_FOUND)
                 .description("No policy entry found with READ permission for things.")
diff --git a/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/AbstractPolicyCommandStrategy.java b/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/AbstractPolicyCommandStrategy.java
index d47f284616..f4a0804eac 100644
--- a/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/AbstractPolicyCommandStrategy.java
+++ b/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/AbstractPolicyCommandStrategy.java
@@ -57,8 +57,8 @@
  * @param <C> the type of the handled command - of type {@code Command} as also
  * {@link org.eclipse.ditto.services.models.policies.commands.sudo.SudoCommand} are handled which are no PolicyCommands.
  */
-abstract class AbstractPolicyCommandStrategy<C extends Command<C>>
-        extends AbstractConditionHeaderCheckingCommandStrategy<C, Policy, PolicyId, PolicyEvent<?>> {
+abstract class AbstractPolicyCommandStrategy<C extends Command<C>, E extends PolicyEvent<?>>
+        extends AbstractConditionHeaderCheckingCommandStrategy<C, Policy, PolicyId, E> {
 
     private final PolicyExpiryGranularity policyExpiryGranularity;
 
@@ -235,7 +235,8 @@ protected SubjectExpiry roundPolicySubjectExpiry(final SubjectExpiry expiry) {
      * @param command the command which caused the change of the policy entries.
      * @return an Optional with ErrorResponse if a subject was invalid, an empty Optional if everything was valid.
      */
-    protected static Optional<Result<PolicyEvent<?>>> checkForAlreadyExpiredSubject(final Iterable<PolicyEntry> entries,
+    protected static <T extends PolicyEvent<?>> Optional<Result<T>> checkForAlreadyExpiredSubject(
+            final Iterable<PolicyEntry> entries,
             final DittoHeaders dittoHeaders, final Command<?> command) {
 
         return StreamSupport.stream(entries.spliterator(), false)
diff --git a/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/AbstractPolicyQueryCommandStrategy.java b/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/AbstractPolicyQueryCommandStrategy.java
index fe753f3cfb..7006954803 100644
--- a/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/AbstractPolicyQueryCommandStrategy.java
+++ b/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/AbstractPolicyQueryCommandStrategy.java
@@ -20,6 +20,7 @@
 import org.eclipse.ditto.model.policies.Policy;
 import org.eclipse.ditto.services.policies.common.config.PolicyConfig;
 import org.eclipse.ditto.signals.commands.base.Command;
+import org.eclipse.ditto.signals.events.policies.PolicyEvent;
 
 /**
  * Abstract base class for
@@ -27,7 +28,7 @@
  * {@link org.eclipse.ditto.services.models.policies.commands.sudo.SudoCommand} are handled which are no
  * PolicyQueryCommands.
  */
-abstract class AbstractPolicyQueryCommandStrategy<C extends Command<C>> extends AbstractPolicyCommandStrategy<C> {
+abstract class AbstractPolicyQueryCommandStrategy<C extends Command<C>> extends AbstractPolicyCommandStrategy<C, PolicyEvent<?>> {
 
     AbstractPolicyQueryCommandStrategy(final Class<C> theMatchingClass, final PolicyConfig policyConfig) {
         super(theMatchingClass, policyConfig);
diff --git a/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/ActivatePolicyTokenIntegrationStrategy.java b/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/ActivatePolicyTokenIntegrationStrategy.java
index b28d9d67a3..7f0bbb8ee7 100644
--- a/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/ActivatePolicyTokenIntegrationStrategy.java
+++ b/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/ActivatePolicyTokenIntegrationStrategy.java
@@ -43,7 +43,7 @@
 import org.eclipse.ditto.services.utils.persistentactors.results.ResultFactory;
 import org.eclipse.ditto.signals.commands.policies.actions.ActivatePolicyTokenIntegration;
 import org.eclipse.ditto.signals.commands.policies.actions.ActivatePolicyTokenIntegrationResponse;
-import org.eclipse.ditto.signals.events.policies.PolicyEvent;
+import org.eclipse.ditto.signals.events.policies.PolicyActionEvent;
 import org.eclipse.ditto.signals.events.policies.SubjectsModifiedPartially;
 
 import akka.actor.ActorSystem;
@@ -59,7 +59,7 @@ final class ActivatePolicyTokenIntegrationStrategy
     }
 
     @Override
-    protected Result<PolicyEvent<?>> doApply(final Context<PolicyId> context,
+    protected Result<PolicyActionEvent<?>> doApply(final Context<PolicyId> context,
             @Nullable final Policy policy,
             final long nextRevision,
             final ActivatePolicyTokenIntegration command,
@@ -76,7 +76,7 @@ protected Result<PolicyEvent<?>> doApply(final Context<PolicyId> context,
                 .filter(this::containsThingReadPermission)
                 .collect(Collectors.toList());
         if (entries.isEmpty()) {
-            return ResultFactory.newErrorResult(getExceptionForNoEntryWithThingReadPermission(), command);
+            return ResultFactory.newErrorResult(getNotApplicableException(), command);
         }
         final PolicyBuilder policyBuilder = nonNullPolicy.toBuilder();
         final Map<Label, Subject> activatedSubjects = new HashMap<>();
@@ -99,7 +99,7 @@ protected Result<PolicyEvent<?>> doApply(final Context<PolicyId> context,
         final Policy newPolicy = policyBuilder.build();
         final PoliciesValidator validator = PoliciesValidator.newInstance(newPolicy);
         if (validator.isValid()) {
-            final PolicyEvent<?> event =
+            final SubjectsModifiedPartially event =
                     SubjectsModifiedPartially.of(policyId, activatedSubjects, nextRevision, getEventTimestamp(),
                             dittoHeaders);
             final ActivatePolicyTokenIntegrationResponse rawResponse =
diff --git a/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/ActivateTokenIntegrationStrategy.java b/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/ActivateTokenIntegrationStrategy.java
index c1cf7cdd83..a21fc45acb 100644
--- a/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/ActivateTokenIntegrationStrategy.java
+++ b/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/ActivateTokenIntegrationStrategy.java
@@ -38,7 +38,7 @@
 import org.eclipse.ditto.services.utils.persistentactors.results.ResultFactory;
 import org.eclipse.ditto.signals.commands.policies.actions.ActivateTokenIntegration;
 import org.eclipse.ditto.signals.commands.policies.actions.ActivateTokenIntegrationResponse;
-import org.eclipse.ditto.signals.events.policies.PolicyEvent;
+import org.eclipse.ditto.signals.events.policies.PolicyActionEvent;
 import org.eclipse.ditto.signals.events.policies.SubjectCreated;
 import org.eclipse.ditto.signals.events.policies.SubjectModified;
 
@@ -47,14 +47,15 @@
 /**
  * This strategy handles the {@link ActivateTokenIntegration} command.
  */
-final class ActivateTokenIntegrationStrategy extends AbstractPolicyActionCommandStrategy<ActivateTokenIntegration> {
+final class ActivateTokenIntegrationStrategy
+        extends AbstractPolicyActionCommandStrategy<ActivateTokenIntegration> {
 
     ActivateTokenIntegrationStrategy(final PolicyConfig policyConfig, final ActorSystem system) {
         super(ActivateTokenIntegration.class, policyConfig, system);
     }
 
     @Override
-    protected Result<PolicyEvent<?>> doApply(final Context<PolicyId> context,
+    protected Result<PolicyActionEvent<?>> doApply(final Context<PolicyId> context,
             @Nullable final Policy policy,
             final long nextRevision,
             final ActivateTokenIntegration command,
@@ -87,7 +88,7 @@ protected Result<PolicyEvent<?>> doApply(final Context<PolicyId> context,
             // Validation is necessary because activation may add expiry to the policy admin subject.
             final Policy newPolicy = nonNullPolicy.setSubjectFor(label, adjustedSubject);
 
-            final Optional<Result<PolicyEvent<?>>> alreadyExpiredSubject =
+            final Optional<Result<PolicyActionEvent<?>>> alreadyExpiredSubject =
                     checkForAlreadyExpiredSubject(newPolicy, dittoHeaders, command);
             if (alreadyExpiredSubject.isPresent()) {
                 return alreadyExpiredSubject.get();
@@ -96,7 +97,7 @@ protected Result<PolicyEvent<?>> doApply(final Context<PolicyId> context,
             final PoliciesValidator validator = PoliciesValidator.newInstance(newPolicy);
             if (validator.isValid()) {
 
-                final PolicyEvent<?> event;
+                final PolicyActionEvent<?> event;
                 if (policyEntry.getSubjects().getSubject(adjustedSubject.getId()).isPresent()) {
                     event = SubjectModified.of(policyId, label, adjustedSubject, nextRevision, getEventTimestamp(),
                             dittoHeaders);
@@ -114,7 +115,7 @@ protected Result<PolicyEvent<?>> doApply(final Context<PolicyId> context,
                         command);
             }
         } else {
-            return ResultFactory.newErrorResult(getExceptionForNoEntryWithThingReadPermission(), command);
+            return ResultFactory.newErrorResult(getNotApplicableException(), command);
         }
     }
 
diff --git a/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/CreatePolicyStrategy.java b/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/CreatePolicyStrategy.java
index d1d1f2f183..8bcf5de98d 100644
--- a/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/CreatePolicyStrategy.java
+++ b/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/CreatePolicyStrategy.java
@@ -41,7 +41,7 @@
  * This strategy handles the {@link org.eclipse.ditto.signals.commands.policies.modify.CreatePolicy} command for a
  * new Policy.
  */
-final class CreatePolicyStrategy extends AbstractPolicyCommandStrategy<CreatePolicy> {
+final class CreatePolicyStrategy extends AbstractPolicyCommandStrategy<CreatePolicy, PolicyEvent<?>> {
 
     CreatePolicyStrategy(final PolicyConfig policyConfig) {
         super(CreatePolicy.class, policyConfig);
diff --git a/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/DeactivatePolicyTokenIntegrationStrategy.java b/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/DeactivatePolicyTokenIntegrationStrategy.java
index a1fd234153..b53bc797cf 100644
--- a/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/DeactivatePolicyTokenIntegrationStrategy.java
+++ b/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/DeactivatePolicyTokenIntegrationStrategy.java
@@ -38,7 +38,7 @@
 import org.eclipse.ditto.signals.commands.policies.actions.DeactivatePolicyTokenIntegration;
 import org.eclipse.ditto.signals.commands.policies.actions.DeactivatePolicyTokenIntegrationResponse;
 import org.eclipse.ditto.signals.commands.policies.exceptions.PolicyActionFailedException;
-import org.eclipse.ditto.signals.events.policies.PolicyEvent;
+import org.eclipse.ditto.signals.events.policies.PolicyActionEvent;
 import org.eclipse.ditto.signals.events.policies.SubjectsDeletedPartially;
 
 import akka.actor.ActorSystem;
@@ -54,7 +54,7 @@ final class DeactivatePolicyTokenIntegrationStrategy
     }
 
     @Override
-    protected Result<PolicyEvent<?>> doApply(final Context<PolicyId> context,
+    protected Result<PolicyActionEvent<?>> doApply(final Context<PolicyId> context,
             @Nullable final Policy policy,
             final long nextRevision,
             final DeactivatePolicyTokenIntegration command,
@@ -94,7 +94,7 @@ protected Result<PolicyEvent<?>> doApply(final Context<PolicyId> context,
             }
         }
         // Validation is not necessary because temporary subjects do not affect validity
-        final PolicyEvent<?> event =
+        final SubjectsDeletedPartially event =
                 SubjectsDeletedPartially.of(policyId, deactivatedSubjectsIds, nextRevision, getEventTimestamp(),
                         dittoHeaders);
         final DeactivatePolicyTokenIntegrationResponse rawResponse =
diff --git a/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/DeactivateTokenIntegrationStrategy.java b/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/DeactivateTokenIntegrationStrategy.java
index 50018997a7..c0d70d005c 100644
--- a/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/DeactivateTokenIntegrationStrategy.java
+++ b/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/DeactivateTokenIntegrationStrategy.java
@@ -35,7 +35,7 @@
 import org.eclipse.ditto.signals.commands.policies.actions.DeactivateTokenIntegrationResponse;
 import org.eclipse.ditto.signals.commands.policies.exceptions.PolicyActionFailedException;
 import org.eclipse.ditto.signals.commands.policies.exceptions.PolicyEntryNotAccessibleException;
-import org.eclipse.ditto.signals.events.policies.PolicyEvent;
+import org.eclipse.ditto.signals.events.policies.PolicyActionEvent;
 import org.eclipse.ditto.signals.events.policies.SubjectDeleted;
 
 import akka.actor.ActorSystem;
@@ -43,14 +43,15 @@
 /**
  * This strategy handles the {@link DeactivateTokenIntegration} command.
  */
-final class DeactivateTokenIntegrationStrategy extends AbstractPolicyActionCommandStrategy<DeactivateTokenIntegration> {
+final class DeactivateTokenIntegrationStrategy
+        extends AbstractPolicyActionCommandStrategy<DeactivateTokenIntegration> {
 
     DeactivateTokenIntegrationStrategy(final PolicyConfig policyConfig, final ActorSystem system) {
         super(DeactivateTokenIntegration.class, policyConfig, system);
     }
 
     @Override
-    protected Result<PolicyEvent<?>> doApply(final Context<PolicyId> context,
+    protected Result<PolicyActionEvent<?>> doApply(final Context<PolicyId> context,
             @Nullable final Policy policy,
             final long nextRevision,
             final DeactivateTokenIntegration command,
@@ -83,7 +84,7 @@ protected Result<PolicyEvent<?>> doApply(final Context<PolicyId> context,
                 return ResultFactory.newErrorResult(error, command);
             } else {
                 // Expiring subjects are not considered for validation. The result is always valid.
-                final PolicyEvent<?> event =
+                final SubjectDeleted event =
                         SubjectDeleted.of(policyId, label, subjectId, nextRevision, getEventTimestamp(),
                                 dittoHeaders);
                 final DeactivateTokenIntegrationResponse rawResponse =
diff --git a/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/DeletePolicyEntryStrategy.java b/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/DeletePolicyEntryStrategy.java
index d338f659a9..0bcef883a8 100644
--- a/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/DeletePolicyEntryStrategy.java
+++ b/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/DeletePolicyEntryStrategy.java
@@ -37,7 +37,7 @@
 /**
  * This strategy handles the {@link org.eclipse.ditto.signals.commands.policies.modify.DeletePolicyEntry} command.
  */
-final class DeletePolicyEntryStrategy extends AbstractPolicyCommandStrategy<DeletePolicyEntry> {
+final class DeletePolicyEntryStrategy extends AbstractPolicyCommandStrategy<DeletePolicyEntry, PolicyEvent<?>> {
 
     DeletePolicyEntryStrategy(final PolicyConfig policyConfig) {
         super(DeletePolicyEntry.class, policyConfig);
diff --git a/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/DeletePolicyStrategy.java b/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/DeletePolicyStrategy.java
index f9e4ff2d7b..6c3b97b627 100644
--- a/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/DeletePolicyStrategy.java
+++ b/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/DeletePolicyStrategy.java
@@ -33,7 +33,7 @@
 /**
  * This strategy handles the {@link org.eclipse.ditto.signals.commands.policies.modify.DeletePolicy} command.
  */
-final class DeletePolicyStrategy extends AbstractPolicyCommandStrategy<DeletePolicy> {
+final class DeletePolicyStrategy extends AbstractPolicyCommandStrategy<DeletePolicy, PolicyEvent<?>> {
 
     DeletePolicyStrategy(final PolicyConfig policyConfig) {
         super(DeletePolicy.class, policyConfig);
diff --git a/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/DeleteResourceStrategy.java b/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/DeleteResourceStrategy.java
index 8fb4c7fb62..894e2c5af7 100644
--- a/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/DeleteResourceStrategy.java
+++ b/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/DeleteResourceStrategy.java
@@ -39,7 +39,7 @@
 /**
  * This strategy handles the {@link org.eclipse.ditto.signals.commands.policies.modify.DeleteResource} command.
  */
-final class DeleteResourceStrategy extends AbstractPolicyCommandStrategy<DeleteResource> {
+final class DeleteResourceStrategy extends AbstractPolicyCommandStrategy<DeleteResource, PolicyEvent<?>> {
 
     DeleteResourceStrategy(final PolicyConfig policyConfig) {
         super(DeleteResource.class, policyConfig);
diff --git a/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/DeleteSubjectStrategy.java b/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/DeleteSubjectStrategy.java
index 560725e4e7..6b8dacabc5 100644
--- a/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/DeleteSubjectStrategy.java
+++ b/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/DeleteSubjectStrategy.java
@@ -39,7 +39,7 @@
 /**
  * This strategy handles the {@link org.eclipse.ditto.signals.commands.policies.modify.DeleteSubject} command.
  */
-final class DeleteSubjectStrategy extends AbstractPolicyCommandStrategy<DeleteSubject> {
+final class DeleteSubjectStrategy extends AbstractPolicyCommandStrategy<DeleteSubject, PolicyEvent<?>> {
 
     DeleteSubjectStrategy(final PolicyConfig policyConfig) {
         super(DeleteSubject.class, policyConfig);
diff --git a/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/ModifyPolicyEntriesStrategy.java b/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/ModifyPolicyEntriesStrategy.java
index 333b5ca7cd..ff1adc8ea7 100644
--- a/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/ModifyPolicyEntriesStrategy.java
+++ b/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/ModifyPolicyEntriesStrategy.java
@@ -28,15 +28,10 @@
 import org.eclipse.ditto.model.policies.PolicyEntry;
 import org.eclipse.ditto.model.policies.PolicyId;
 import org.eclipse.ditto.model.policies.PolicyTooLargeException;
-import org.eclipse.ditto.model.policies.Subject;
-import org.eclipse.ditto.model.policies.SubjectExpiry;
-import org.eclipse.ditto.model.policies.SubjectExpiryInvalidException;
-import org.eclipse.ditto.model.policies.Subjects;
 import org.eclipse.ditto.services.models.policies.PoliciesValidator;
 import org.eclipse.ditto.services.policies.common.config.PolicyConfig;
 import org.eclipse.ditto.services.utils.persistentactors.results.Result;
 import org.eclipse.ditto.services.utils.persistentactors.results.ResultFactory;
-import org.eclipse.ditto.signals.commands.base.Command;
 import org.eclipse.ditto.signals.commands.policies.PolicyCommandSizeValidator;
 import org.eclipse.ditto.signals.commands.policies.modify.ModifyPolicyEntries;
 import org.eclipse.ditto.signals.commands.policies.modify.ModifyPolicyEntriesResponse;
@@ -46,7 +41,7 @@
 /**
  * This strategy handles the {@link org.eclipse.ditto.signals.commands.policies.modify.ModifyPolicyEntries} command.
  */
-final class ModifyPolicyEntriesStrategy extends AbstractPolicyCommandStrategy<ModifyPolicyEntries> {
+final class ModifyPolicyEntriesStrategy extends AbstractPolicyCommandStrategy<ModifyPolicyEntries, PolicyEvent<?>> {
 
     ModifyPolicyEntriesStrategy(final PolicyConfig policyConfig) {
         super(ModifyPolicyEntries.class, policyConfig);
diff --git a/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/ModifyPolicyEntryStrategy.java b/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/ModifyPolicyEntryStrategy.java
index 4ca16dba44..070ee3903b 100644
--- a/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/ModifyPolicyEntryStrategy.java
+++ b/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/ModifyPolicyEntryStrategy.java
@@ -44,7 +44,7 @@
  * This strategy handles the {@link org.eclipse.ditto.signals.commands.policies.modify.ModifyPolicyEntry} command.
  */
 @NotThreadSafe
-final class ModifyPolicyEntryStrategy extends AbstractPolicyCommandStrategy<ModifyPolicyEntry> {
+final class ModifyPolicyEntryStrategy extends AbstractPolicyCommandStrategy<ModifyPolicyEntry, PolicyEvent<?>> {
 
     ModifyPolicyEntryStrategy(final PolicyConfig policyConfig) {
         super(ModifyPolicyEntry.class, policyConfig);
diff --git a/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/ModifyPolicyStrategy.java b/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/ModifyPolicyStrategy.java
index 490f81de7e..8c99fbb2fa 100644
--- a/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/ModifyPolicyStrategy.java
+++ b/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/ModifyPolicyStrategy.java
@@ -42,7 +42,7 @@
  * This strategy handles the {@link org.eclipse.ditto.signals.commands.policies.modify.ModifyPolicy} command for an
  * already existing Policy.
  */
-final class ModifyPolicyStrategy extends AbstractPolicyCommandStrategy<ModifyPolicy> {
+final class ModifyPolicyStrategy extends AbstractPolicyCommandStrategy<ModifyPolicy, PolicyEvent<?>> {
 
     ModifyPolicyStrategy(final PolicyConfig policyConfig) {
         super(ModifyPolicy.class, policyConfig);
diff --git a/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/ModifyResourceStrategy.java b/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/ModifyResourceStrategy.java
index f71b0c1c4f..a365741f0f 100644
--- a/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/ModifyResourceStrategy.java
+++ b/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/ModifyResourceStrategy.java
@@ -39,7 +39,7 @@
 /**
  * This strategy handles the {@link org.eclipse.ditto.signals.commands.policies.modify.ModifyResource} command.
  */
-final class ModifyResourceStrategy extends AbstractPolicyCommandStrategy<ModifyResource> {
+final class ModifyResourceStrategy extends AbstractPolicyCommandStrategy<ModifyResource, PolicyEvent<?>> {
 
     ModifyResourceStrategy(final PolicyConfig policyConfig) {
         super(ModifyResource.class, policyConfig);
diff --git a/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/ModifyResourcesStrategy.java b/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/ModifyResourcesStrategy.java
index 551140db21..b308b2d18d 100644
--- a/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/ModifyResourcesStrategy.java
+++ b/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/ModifyResourcesStrategy.java
@@ -46,7 +46,7 @@
 /**
  * This strategy handles the {@link org.eclipse.ditto.signals.commands.policies.modify.ModifyResources} command.
  */
-final class ModifyResourcesStrategy extends AbstractPolicyCommandStrategy<ModifyResources> {
+final class ModifyResourcesStrategy extends AbstractPolicyCommandStrategy<ModifyResources, PolicyEvent<?>> {
 
     ModifyResourcesStrategy(final PolicyConfig policyConfig) {
         super(ModifyResources.class, policyConfig);
diff --git a/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/ModifySubjectStrategy.java b/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/ModifySubjectStrategy.java
index 541eeae69c..1db3b28156 100644
--- a/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/ModifySubjectStrategy.java
+++ b/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/ModifySubjectStrategy.java
@@ -39,7 +39,7 @@
 /**
  * This strategy handles the {@link org.eclipse.ditto.signals.commands.policies.modify.ModifySubject} command.
  */
-final class ModifySubjectStrategy extends AbstractPolicyCommandStrategy<ModifySubject> {
+final class ModifySubjectStrategy extends AbstractPolicyCommandStrategy<ModifySubject, PolicyEvent<?>> {
 
     ModifySubjectStrategy(final PolicyConfig policyConfig) {
         super(ModifySubject.class, policyConfig);
diff --git a/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/ModifySubjectsStrategy.java b/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/ModifySubjectsStrategy.java
index 75a8118ccc..1840223732 100644
--- a/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/ModifySubjectsStrategy.java
+++ b/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/ModifySubjectsStrategy.java
@@ -39,7 +39,7 @@
 /**
  * This strategy handles the {@link org.eclipse.ditto.signals.commands.policies.modify.ModifySubjects} command.
  */
-final class ModifySubjectsStrategy extends AbstractPolicyCommandStrategy<ModifySubjects> {
+final class ModifySubjectsStrategy extends AbstractPolicyCommandStrategy<ModifySubjects, PolicyEvent<?>> {
 
     /**
      * Constructs a new {@code ModifySubjectsStrategy} object.
diff --git a/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/PolicyCommandStrategies.java b/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/PolicyCommandStrategies.java
index a206a9bc03..eca7665e87 100755
--- a/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/PolicyCommandStrategies.java
+++ b/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/PolicyCommandStrategies.java
@@ -44,8 +44,9 @@ private PolicyCommandStrategies(final PolicyConfig policyConfig, final ActorSyst
         addStrategy(new ModifyPolicyStrategy(policyConfig));
         addStrategy(new RetrievePolicyStrategy(policyConfig));
         addStrategy(new DeletePolicyStrategy(policyConfig));
-        addStrategy(new ActivatePolicyTokenIntegrationStrategy(policyConfig, system));
-        addStrategy(new DeactivatePolicyTokenIntegrationStrategy(policyConfig, system));
+        addStrategy(new TopLevelActionCommandStrategy(policyConfig, system));
+        addStrategy(this, new ActivatePolicyTokenIntegrationStrategy(policyConfig, system));
+        addStrategy(this, new DeactivatePolicyTokenIntegrationStrategy(policyConfig, system));
 
         // Policy Entries
         addStrategy(new ModifyPolicyEntriesStrategy(policyConfig));
@@ -55,8 +56,8 @@ private PolicyCommandStrategies(final PolicyConfig policyConfig, final ActorSyst
         addStrategy(new ModifyPolicyEntryStrategy(policyConfig));
         addStrategy(new RetrievePolicyEntryStrategy(policyConfig));
         addStrategy(new DeletePolicyEntryStrategy(policyConfig));
-        addStrategy(new ActivateTokenIntegrationStrategy(policyConfig, system));
-        addStrategy(new DeactivateTokenIntegrationStrategy(policyConfig, system));
+        addStrategy(this, new ActivateTokenIntegrationStrategy(policyConfig, system));
+        addStrategy(this, new DeactivateTokenIntegrationStrategy(policyConfig, system));
 
         // Subjects
         addStrategy(new ModifySubjectsStrategy(policyConfig));
diff --git a/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/PolicyConflictStrategy.java b/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/PolicyConflictStrategy.java
index f45111e13c..dcc355865d 100644
--- a/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/PolicyConflictStrategy.java
+++ b/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/PolicyConflictStrategy.java
@@ -31,7 +31,7 @@
  * This strategy handles the {@link org.eclipse.ditto.signals.commands.policies.modify.CreatePolicy} command for an
  * already existing Policy.
  */
-final class PolicyConflictStrategy extends AbstractPolicyCommandStrategy<CreatePolicy> {
+final class PolicyConflictStrategy extends AbstractPolicyCommandStrategy<CreatePolicy, PolicyEvent<?>> {
 
     PolicyConflictStrategy(final PolicyConfig policyConfig) {
         super(CreatePolicy.class, policyConfig);
diff --git a/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/TopLevelActionCommandStrategy.java b/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/TopLevelActionCommandStrategy.java
new file mode 100644
index 0000000000..f0402c390d
--- /dev/null
+++ b/services/policies/persistence/src/main/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/TopLevelActionCommandStrategy.java
@@ -0,0 +1,180 @@
+/*
+ * Copyright (c) 2021 Contributors to the Eclipse Foundation
+ *
+ * See the NOTICE file(s) distributed with this work for additional
+ * information regarding copyright ownership.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License 2.0 which is available at
+ * http://www.eclipse.org/legal/epl-2.0
+ *
+ * SPDX-License-Identifier: EPL-2.0
+ */
+package org.eclipse.ditto.services.policies.persistence.actors.strategies.commands;
+
+import static org.eclipse.ditto.model.base.common.ConditionChecker.checkNotNull;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+import javax.annotation.Nullable;
+
+import org.eclipse.ditto.model.base.entity.metadata.Metadata;
+import org.eclipse.ditto.model.base.exceptions.DittoRuntimeException;
+import org.eclipse.ditto.model.base.headers.WithDittoHeaders;
+import org.eclipse.ditto.model.base.headers.entitytag.EntityTag;
+import org.eclipse.ditto.model.policies.Policy;
+import org.eclipse.ditto.model.policies.PolicyEntry;
+import org.eclipse.ditto.model.policies.PolicyId;
+import org.eclipse.ditto.services.policies.common.config.PolicyConfig;
+import org.eclipse.ditto.services.utils.persistentactors.results.Result;
+import org.eclipse.ditto.services.utils.persistentactors.results.ResultFactory;
+import org.eclipse.ditto.services.utils.persistentactors.results.ResultVisitor;
+import org.eclipse.ditto.signals.commands.base.Command;
+import org.eclipse.ditto.signals.commands.policies.actions.ActivateTokenIntegration;
+import org.eclipse.ditto.signals.commands.policies.actions.DeactivateTokenIntegration;
+import org.eclipse.ditto.signals.commands.policies.actions.PolicyActionCommand;
+import org.eclipse.ditto.signals.commands.policies.actions.TopLevelActionCommand;
+import org.eclipse.ditto.signals.commands.policies.actions.TopLevelActionCommandResponse;
+import org.eclipse.ditto.signals.commands.policies.exceptions.PolicyActionFailedException;
+import org.eclipse.ditto.signals.events.policies.PolicyActionEvent;
+import org.eclipse.ditto.signals.events.policies.PolicyEvent;
+
+import akka.actor.ActorSystem;
+
+/**
+ * This strategy handles the {@link org.eclipse.ditto.signals.commands.policies.actions.TopLevelActionCommand} command.
+ */
+final class TopLevelActionCommandStrategy
+        extends AbstractPolicyCommandStrategy<TopLevelActionCommand, PolicyEvent<?>> {
+
+    private final Map<String, AbstractPolicyActionCommandStrategy<?>> policyActionCommandStrategyMap;
+
+    TopLevelActionCommandStrategy(final PolicyConfig policyConfig, final ActorSystem system) {
+        super(TopLevelActionCommand.class, policyConfig);
+        policyActionCommandStrategyMap = instantiatePolicyActionCommandStrategies(policyConfig, system);
+    }
+
+    @Override
+    protected Result<PolicyEvent<?>> doApply(final Context<PolicyId> context,
+            @Nullable final Policy policy,
+            final long nextRevision,
+            final TopLevelActionCommand command,
+            @Nullable final Metadata metadata) {
+
+        final Policy nonNullPolicy = checkNotNull(policy, "policy");
+        final PolicyActionCommand<?> actionCommand = command.getPolicyActionCommand();
+        final List<PolicyEntry> entries = command.getAuthorizedLabels()
+                .stream()
+                .map(nonNullPolicy::getEntryFor)
+                .flatMap(Optional::stream)
+                .filter(actionCommand::isApplicable)
+                .collect(Collectors.toList());
+        final AbstractPolicyActionCommandStrategy<?> strategy =
+                policyActionCommandStrategyMap.get(actionCommand.getName());
+
+        if (strategy == null) {
+            final PolicyActionFailedException exception = PolicyActionFailedException.newBuilder()
+                    .action(command.getPolicyActionCommand().getName())
+                    .build();
+            context.getLog()
+                    .withCorrelationId(command)
+                    .error(exception, "Strategy not found for top-level action <{}>", actionCommand.getName());
+            return ResultFactory.newErrorResult(exception, command);
+        } else if (!entries.isEmpty()) {
+            final List<PolicyActionCommand<?>> commands = entries.stream()
+                    .map(PolicyEntry::getLabel)
+                    .map(actionCommand::setLabel)
+                    .collect(Collectors.toList());
+            final ResultCollectionVisitor visitor =
+                    collectResults(strategy, context, policy, nextRevision, commands);
+            if (visitor.error != null) {
+                return ResultFactory.newErrorResult(visitor.error, command);
+            } else {
+                final Optional<PolicyEvent<?>> event = visitor.aggregateEvents();
+                if (event.isPresent()) {
+                    final TopLevelActionCommandResponse response =
+                            TopLevelActionCommandResponse.of(context.getState(), command.getDittoHeaders());
+                    return ResultFactory.newMutationResult(command, event.get(), response);
+                }
+            }
+        }
+        return ResultFactory.newErrorResult(strategy.getNotApplicableException(), command);
+    }
+
+    @Override
+    public Optional<EntityTag> previousEntityTag(final TopLevelActionCommand command,
+            @Nullable final Policy previousEntity) {
+        // activated subjects do not support entity tag
+        return Optional.empty();
+    }
+
+    @Override
+    public Optional<EntityTag> nextEntityTag(final TopLevelActionCommand command,
+            @Nullable final Policy newEntity) {
+        // activated subjects do not support entity tag
+        return Optional.empty();
+    }
+
+    private static Map<String, AbstractPolicyActionCommandStrategy<?>> instantiatePolicyActionCommandStrategies(
+            final PolicyConfig policyConfig, final ActorSystem system) {
+
+        return Map.of(
+                ActivateTokenIntegration.NAME, new ActivateTokenIntegrationStrategy(policyConfig, system),
+                DeactivateTokenIntegration.NAME, new DeactivateTokenIntegrationStrategy(policyConfig, system)
+        );
+    }
+
+    private static ResultCollectionVisitor collectResults(
+            final AbstractPolicyActionCommandStrategy<?> strategy,
+            final Context<PolicyId> context,
+            @Nullable final Policy policy,
+            final long nextRevision,
+            final Collection<PolicyActionCommand<?>> commands) {
+        final ResultCollectionVisitor visitor = new ResultCollectionVisitor();
+        for (final PolicyActionCommand<?> command : commands) {
+            strategy.typeCheckAndApply(context, policy, nextRevision, command)
+                    .ifPresent(result -> result.accept(visitor));
+        }
+        return visitor;
+    }
+
+    private static final class ResultCollectionVisitor implements ResultVisitor<PolicyActionEvent<?>> {
+
+        @Nullable private DittoRuntimeException error;
+        @Nullable private PolicyActionEvent<?> firstEvent;
+        private final List<PolicyActionEvent<?>> otherEvents = new ArrayList<>();
+
+        private Optional<PolicyEvent<?>> aggregateEvents() {
+            if (firstEvent == null) {
+                return Optional.empty();
+            } else {
+                return Optional.of(firstEvent.aggregateWith(otherEvents));
+            }
+        }
+
+        @Override
+        public void onMutation(final Command<?> command, final PolicyActionEvent<?> event,
+                final WithDittoHeaders<?> response, final boolean becomeCreated, final boolean becomeDeleted) {
+            if (firstEvent == null) {
+                firstEvent = event;
+            } else {
+                otherEvents.add(event);
+            }
+        }
+
+        @Override
+        public void onQuery(final Command<?> command, final WithDittoHeaders<?> response) {
+            // do nothing
+        }
+
+        @Override
+        public void onError(final DittoRuntimeException error, final Command<?> errorCausingCommand) {
+            this.error = error;
+        }
+    }
+}
diff --git a/services/policies/persistence/src/test/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/AbstractPolicyCommandStrategyTest.java b/services/policies/persistence/src/test/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/AbstractPolicyCommandStrategyTest.java
index 306c2ef8a7..55774dbd39 100644
--- a/services/policies/persistence/src/test/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/AbstractPolicyCommandStrategyTest.java
+++ b/services/policies/persistence/src/test/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/AbstractPolicyCommandStrategyTest.java
@@ -16,7 +16,6 @@
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyBoolean;
 import static org.mockito.ArgumentMatchers.eq;
-import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
 
 import java.util.function.Consumer;
@@ -38,6 +37,7 @@
 import org.eclipse.ditto.signals.commands.base.Command;
 import org.eclipse.ditto.signals.commands.base.CommandResponse;
 import org.eclipse.ditto.signals.commands.policies.PolicyCommandResponse;
+import org.eclipse.ditto.signals.events.base.Event;
 import org.eclipse.ditto.signals.events.policies.PolicyEvent;
 import org.junit.BeforeClass;
 import org.mockito.ArgumentCaptor;
@@ -65,7 +65,7 @@ protected static CommandStrategy.Context<PolicyId> getDefaultContext() {
     }
 
     protected static <C extends Command<?>, T extends PolicyEvent<?>> T assertModificationResult(
-            final CommandStrategy<C, Policy, PolicyId, Result<PolicyEvent<?>>> underTest,
+            final CommandStrategy<C, Policy, PolicyId, ?> underTest,
             @Nullable final Policy policy,
             final C command,
             final Class<T> expectedEventClass,
@@ -76,7 +76,7 @@ protected static <C extends Command<?>, T extends PolicyEvent<?>> T assertModifi
 
     protected static <C extends Command<?>, T extends PolicyEvent<?>, R extends PolicyCommandResponse<R>>
     void assertModificationResult(
-            final CommandStrategy<C, Policy, PolicyId, Result<PolicyEvent<?>>> underTest,
+            final CommandStrategy<C, Policy, PolicyId, ?> underTest,
             @Nullable final Policy policy,
             final C command,
             final Class<T> expectedEventClass,
@@ -85,12 +85,12 @@ void assertModificationResult(
             final Consumer<R> responseSatisfactions) {
 
         final CommandStrategy.Context<PolicyId> context = getDefaultContext();
-        final Result<PolicyEvent<?>> result = applyStrategy(underTest, context, policy, command);
+        final Result<?> result = applyStrategy(underTest, context, policy, command);
 
         final ArgumentCaptor<T> event = ArgumentCaptor.forClass(expectedEventClass);
         final ArgumentCaptor<R> response = ArgumentCaptor.forClass(expectedResponseClass);
-        final ResultVisitor<PolicyEvent<?>> mock = mock(Dummy.class);
-        result.accept(mock);
+        final Dummy<T> mock = Dummy.mock();
+        result.accept(cast(mock));
 
         verify(mock).onMutation(any(), event.capture(), response.capture(), anyBoolean(), eq(false));
         assertThat(event.getValue()).isInstanceOf(expectedEventClass);
@@ -107,7 +107,7 @@ void assertModificationResult(
     }
 
     protected static <C extends Command<?>, T extends PolicyEvent<?>> T assertModificationResult(
-            final CommandStrategy<C, Policy, PolicyId, Result<PolicyEvent<?>>> underTest,
+            final CommandStrategy<C, Policy, PolicyId, ?> underTest,
             @Nullable final Policy policy,
             final C command,
             final Class<T> expectedEventClass,
@@ -115,40 +115,40 @@ protected static <C extends Command<?>, T extends PolicyEvent<?>> T assertModifi
             final boolean becomeDeleted) {
 
         final CommandStrategy.Context<PolicyId> context = getDefaultContext();
-        final Result<PolicyEvent<?>> result = applyStrategy(underTest, context, policy, command);
+        final Result<?> result = applyStrategy(underTest, context, policy, command);
 
         return assertModificationResult(result, expectedEventClass, expectedCommandResponse, becomeDeleted);
     }
 
     protected static <C extends Command<?>> void assertErrorResult(
-            final CommandStrategy<C, Policy, PolicyId, Result<PolicyEvent<?>>> underTest,
+            final CommandStrategy<C, Policy, PolicyId, ?> underTest,
             @Nullable final Policy policy,
             final C command,
             final DittoRuntimeException expectedException) {
 
-        final ResultVisitor<PolicyEvent<?>> mock = mock(Dummy.class);
-        applyStrategy(underTest, getDefaultContext(), policy, command).accept(mock);
+        final Dummy<?> mock = Dummy.mock();
+        applyStrategy(underTest, getDefaultContext(), policy, command).accept(cast(mock));
         verify(mock).onError(eq(expectedException), eq(command));
     }
 
-    private static <T extends PolicyEvent<?>> T assertModificationResult(final Result<PolicyEvent<?>> result,
+    private static <T extends PolicyEvent<?>> T assertModificationResult(final Result<?> result,
             final Class<T> eventClazz,
             final WithDittoHeaders<?> expectedResponse,
             final boolean becomeDeleted) {
 
         final ArgumentCaptor<T> event = ArgumentCaptor.forClass(eventClazz);
 
-        final ResultVisitor<PolicyEvent<?>> mock = mock(Dummy.class);
+        final Dummy<T> mock = Dummy.mock();
 
-        result.accept(mock);
+        result.accept(cast(mock));
 
         verify(mock).onMutation(any(), event.capture(), eq(expectedResponse), anyBoolean(), eq(becomeDeleted));
         assertThat(event.getValue()).isInstanceOf(eventClazz);
         return event.getValue();
     }
 
-    private static <C extends Command<?>> Result<PolicyEvent<?>> applyStrategy(
-            final CommandStrategy<C, Policy, PolicyId, Result<PolicyEvent<?>>> underTest,
+    private static <C extends Command<?>> Result<?> applyStrategy(
+            final CommandStrategy<C, Policy, PolicyId, ?> underTest,
             final CommandStrategy.Context<PolicyId> context,
             @Nullable final Policy policy,
             final C command) {
@@ -156,7 +156,18 @@ private static <C extends Command<?>> Result<PolicyEvent<?>> applyStrategy(
         return underTest.apply(context, policy, NEXT_REVISION, command);
     }
 
-    interface Dummy extends ResultVisitor<PolicyEvent<?>> {}
+    @SuppressWarnings({"unchecked", "rawtypes"})
+    private static <T extends Event<?>> Dummy<T> cast(final Dummy<?> dummy) {
+        return (Dummy) dummy;
+    }
+
+    interface Dummy<T extends Event<?>> extends ResultVisitor<T> {
+
+        @SuppressWarnings("unchecked")
+        static <T extends Event<?>> Dummy<T> mock() {
+            return Mockito.mock(Dummy.class);
+        }
+    }
 
     protected static DittoHeaders appendETagToDittoHeaders(final Object object, final DittoHeaders dittoHeaders) {
 
diff --git a/services/policies/persistence/src/test/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/ActivatePolicyTokenIntegrationStrategyTest.java b/services/policies/persistence/src/test/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/ActivatePolicyTokenIntegrationStrategyTest.java
index 1f42429746..d2c7205e21 100644
--- a/services/policies/persistence/src/test/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/ActivatePolicyTokenIntegrationStrategyTest.java
+++ b/services/policies/persistence/src/test/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/ActivatePolicyTokenIntegrationStrategyTest.java
@@ -132,7 +132,7 @@ public void rejectEmptyLabels() {
         final ActivatePolicyTokenIntegration command =
                 ActivatePolicyTokenIntegration.of(context.getState(), subjectId, expiry, List.of(), dittoHeaders);
         assertErrorResult(underTest, TestConstants.Policy.POLICY, command,
-                underTest.getExceptionForNoEntryWithThingReadPermission());
+                underTest.getNotApplicableException());
     }
 
     @Test
@@ -146,7 +146,7 @@ public void rejectNonexistentLabel() {
                 ActivatePolicyTokenIntegration.of(context.getState(), subjectId, expiry, List.of(nonexistentLabel),
                         dittoHeaders);
         assertErrorResult(underTest, TestConstants.Policy.POLICY, command,
-                underTest.getExceptionForNoEntryWithThingReadPermission());
+                underTest.getNotApplicableException());
     }
 
     @Test
@@ -163,6 +163,6 @@ public void rejectEntryWithoutThingReadPermission() {
                 .setSubject(TestConstants.Policy.SUPPORT_SUBJECT)
                 .setGrantedPermissions(ResourceKey.newInstance("policy:/"), Permission.READ)
                 .build();
-        assertErrorResult(underTest, policy, command, underTest.getExceptionForNoEntryWithThingReadPermission());
+        assertErrorResult(underTest, policy, command, underTest.getNotApplicableException());
     }
 }
diff --git a/services/policies/persistence/src/test/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/ActivateTokenIntegrationStrategyTest.java b/services/policies/persistence/src/test/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/ActivateTokenIntegrationStrategyTest.java
index 15ee93ca26..f2d9dd0cb2 100644
--- a/services/policies/persistence/src/test/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/ActivateTokenIntegrationStrategyTest.java
+++ b/services/policies/persistence/src/test/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/ActivateTokenIntegrationStrategyTest.java
@@ -126,6 +126,6 @@ public void rejectEntryWithoutThingReadPermission() {
                 .setSubject(TestConstants.Policy.SUPPORT_SUBJECT)
                 .setGrantedPermissions(ResourceKey.newInstance("policy:/"), Permission.READ)
                 .build();
-        assertErrorResult(underTest, policy, command, underTest.getExceptionForNoEntryWithThingReadPermission());
+        assertErrorResult(underTest, policy, command, underTest.getNotApplicableException());
     }
 }
diff --git a/services/policies/persistence/src/test/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/TopLevelActionCommandStrategyTest.java b/services/policies/persistence/src/test/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/TopLevelActionCommandStrategyTest.java
new file mode 100644
index 0000000000..5329aa9b2a
--- /dev/null
+++ b/services/policies/persistence/src/test/java/org/eclipse/ditto/services/policies/persistence/actors/strategies/commands/TopLevelActionCommandStrategyTest.java
@@ -0,0 +1,304 @@
+/*
+ * Copyright (c) 2021 Contributors to the Eclipse Foundation
+ *
+ * See the NOTICE file(s) distributed with this work for additional
+ * information regarding copyright ownership.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License 2.0 which is available at
+ * http://www.eclipse.org/legal/epl-2.0
+ *
+ * SPDX-License-Identifier: EPL-2.0
+ */
+package org.eclipse.ditto.services.policies.persistence.actors.strategies.commands;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.eclipse.ditto.services.policies.persistence.TestConstants.Policy.LABEL;
+import static org.mutabilitydetector.unittesting.MutabilityAssert.assertInstancesOf;
+import static org.mutabilitydetector.unittesting.MutabilityMatchers.areImmutable;
+
+import java.text.MessageFormat;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.List;
+import java.util.Map;
+
+import org.eclipse.ditto.model.base.headers.DittoHeaders;
+import org.eclipse.ditto.model.placeholders.UnresolvedPlaceholderException;
+import org.eclipse.ditto.model.policies.Label;
+import org.eclipse.ditto.model.policies.PoliciesModelFactory;
+import org.eclipse.ditto.model.policies.Policy;
+import org.eclipse.ditto.model.policies.PolicyId;
+import org.eclipse.ditto.model.policies.ResourceKey;
+import org.eclipse.ditto.model.policies.SubjectExpiry;
+import org.eclipse.ditto.model.policies.SubjectId;
+import org.eclipse.ditto.model.policies.SubjectIdInvalidException;
+import org.eclipse.ditto.model.policies.SubjectIssuer;
+import org.eclipse.ditto.services.models.policies.Permission;
+import org.eclipse.ditto.services.policies.common.config.DefaultPolicyConfig;
+import org.eclipse.ditto.services.policies.common.config.PolicyConfig;
+import org.eclipse.ditto.services.policies.persistence.TestConstants;
+import org.eclipse.ditto.services.utils.persistentactors.commands.CommandStrategy;
+import org.eclipse.ditto.signals.commands.policies.actions.ActivatePolicyTokenIntegration;
+import org.eclipse.ditto.signals.commands.policies.actions.ActivateTokenIntegration;
+import org.eclipse.ditto.signals.commands.policies.actions.DeactivateTokenIntegration;
+import org.eclipse.ditto.signals.commands.policies.actions.TopLevelActionCommand;
+import org.eclipse.ditto.signals.commands.policies.actions.TopLevelActionCommandResponse;
+import org.eclipse.ditto.signals.commands.policies.exceptions.PolicyActionFailedException;
+import org.eclipse.ditto.signals.events.policies.SubjectsDeletedPartially;
+import org.eclipse.ditto.signals.events.policies.SubjectsModifiedPartially;
+import org.junit.Before;
+import org.junit.Test;
+
+import com.typesafe.config.ConfigFactory;
+
+import akka.actor.ActorSystem;
+
+/**
+ * Tests {@link TopLevelActionCommandStrategy}.
+ */
+public final class TopLevelActionCommandStrategyTest extends AbstractPolicyCommandStrategyTest {
+
+    private static final Label DUMMY_LABEL = Label.of("-");
+
+    private TopLevelActionCommandStrategy underTest;
+    private ActivateTokenIntegrationStrategy activateTokenIntegrationStrategy;
+
+    @Before
+    public void setUp() {
+        final PolicyConfig policyConfig = DefaultPolicyConfig.of(ConfigFactory.load("policy-test"));
+        final ActorSystem system = ActorSystem.create("test");
+        underTest = new TopLevelActionCommandStrategy(policyConfig, system);
+        activateTokenIntegrationStrategy = new ActivateTokenIntegrationStrategy(policyConfig, system);
+    }
+
+    @Test
+    public void assertImmutability() {
+        assertInstancesOf(ActivatePolicyTokenIntegrationStrategy.class, areImmutable());
+    }
+
+    @Test
+    public void activateTokenIntegration() {
+        final CommandStrategy.Context<PolicyId> context = getDefaultContext();
+        final Instant expiry = Instant.now().plus(Duration.ofDays(1L));
+        final SubjectId subjectId =
+                SubjectId.newInstance(SubjectIssuer.INTEGRATION, "{{policy-entry:label}}:this-is-me");
+        final SubjectId expectedSubjectId =
+                SubjectId.newInstance(SubjectIssuer.INTEGRATION, LABEL + ":this-is-me");
+        final DittoHeaders dittoHeaders = DittoHeaders.empty();
+        final TopLevelActionCommand command = TopLevelActionCommand.of(
+                ActivateTokenIntegration.of(context.getState(), DUMMY_LABEL, subjectId, expiry, dittoHeaders),
+                List.of(LABEL)
+        );
+        assertModificationResult(underTest, TestConstants.Policy.POLICY, command,
+                SubjectsModifiedPartially.class,
+                event -> {
+                    assertThat(event.getModifiedSubjects()).containsOnlyKeys(LABEL);
+                    assertThat(event.getModifiedSubjects().get(LABEL).getId()).isEqualTo(expectedSubjectId);
+                    assertThat(event.getModifiedSubjects().get(LABEL).getExpiry()).isNotEmpty();
+                },
+                TopLevelActionCommandResponse.class,
+                response -> assertThat(response)
+                        .isEqualTo(TopLevelActionCommandResponse.of(context.getState(), dittoHeaders)));
+    }
+
+    @Test
+    public void activateInvalidSubject() {
+        final CommandStrategy.Context<PolicyId> context = getDefaultContext();
+        final Instant expiry = Instant.now().plus(Duration.ofDays(1L));
+        final SubjectId subjectId = SubjectId.newInstance("{{policy-entry:label}}");
+        final DittoHeaders dittoHeaders = DittoHeaders.empty();
+        final TopLevelActionCommand command = TopLevelActionCommand.of(
+                ActivateTokenIntegration.of(context.getState(), DUMMY_LABEL, subjectId, expiry, dittoHeaders),
+                List.of(LABEL)
+        );
+        assertErrorResult(underTest, TestConstants.Policy.POLICY, command,
+                SubjectIdInvalidException.newBuilder(LABEL).build());
+    }
+
+    @Test
+    public void activateUnresolvableSubject() {
+        final CommandStrategy.Context<PolicyId> context = getDefaultContext();
+        final Instant expiry = Instant.now().plus(Duration.ofDays(1L));
+        final SubjectId subjectId = SubjectId.newInstance(SubjectIssuer.INTEGRATION, "{{fn:delete()}}");
+        final DittoHeaders dittoHeaders = DittoHeaders.empty();
+        final TopLevelActionCommand command = TopLevelActionCommand.of(
+                ActivateTokenIntegration.of(context.getState(), DUMMY_LABEL, subjectId, expiry, dittoHeaders),
+                List.of(LABEL)
+        );
+        assertErrorResult(underTest, TestConstants.Policy.POLICY, command,
+                UnresolvedPlaceholderException.newBuilder("integration:{{fn:delete()}}").build());
+    }
+
+    @Test
+    public void activateTokenIntegrationWithUnsupportedPlaceholder() {
+        final CommandStrategy.Context<PolicyId> context = getDefaultContext();
+        final Instant expiry = Instant.now().plus(Duration.ofDays(1L));
+        final SubjectId subjectId = SubjectId.newInstance("{{request:subjectId}}");
+        final DittoHeaders dittoHeaders = DittoHeaders.empty();
+        final TopLevelActionCommand command = TopLevelActionCommand.of(
+                ActivateTokenIntegration.of(context.getState(), DUMMY_LABEL, subjectId, expiry, dittoHeaders),
+                List.of(LABEL)
+        );
+        assertErrorResult(underTest, TestConstants.Policy.POLICY, command,
+                UnresolvedPlaceholderException.newBuilder("{{request:subjectId}}").build());
+    }
+
+    @Test
+    public void rejectEmptyLabels() {
+        final CommandStrategy.Context<PolicyId> context = getDefaultContext();
+        final Instant expiry = Instant.now().plus(Duration.ofDays(1L));
+        final SubjectId subjectId = SubjectId.newInstance("integration:this-is-me");
+        final DittoHeaders dittoHeaders = DittoHeaders.empty();
+        final TopLevelActionCommand command = TopLevelActionCommand.of(
+                ActivateTokenIntegration.of(context.getState(), DUMMY_LABEL, subjectId, expiry, dittoHeaders),
+                List.of()
+        );
+        assertErrorResult(underTest, TestConstants.Policy.POLICY, command,
+                activateTokenIntegrationStrategy.getNotApplicableException());
+    }
+
+    @Test
+    public void rejectNonexistentLabel() {
+        final CommandStrategy.Context<PolicyId> context = getDefaultContext();
+        final Label nonexistentLabel = Label.of("nonexistent-label");
+        final Instant expiry = Instant.now().plus(Duration.ofDays(1L));
+        final SubjectId subjectId = SubjectId.newInstance("integration:this-is-me");
+        final DittoHeaders dittoHeaders = DittoHeaders.empty();
+        final TopLevelActionCommand command = TopLevelActionCommand.of(
+                ActivateTokenIntegration.of(context.getState(), DUMMY_LABEL, subjectId, expiry, dittoHeaders),
+                List.of(nonexistentLabel)
+        );
+        assertErrorResult(underTest, TestConstants.Policy.POLICY, command,
+                activateTokenIntegrationStrategy.getNotApplicableException());
+    }
+
+    @Test
+    public void rejectEntryWithoutThingReadPermission() {
+        final CommandStrategy.Context<PolicyId> context = getDefaultContext();
+        final Label label = Label.of("empty-entry");
+        final Instant expiry = Instant.now().plus(Duration.ofDays(1L));
+        final SubjectId subjectId = SubjectId.newInstance("integration:this-is-me");
+        final DittoHeaders dittoHeaders = DittoHeaders.empty();
+        final TopLevelActionCommand command = TopLevelActionCommand.of(
+                ActivateTokenIntegration.of(context.getState(), DUMMY_LABEL, subjectId, expiry, dittoHeaders),
+                List.of(label)
+        );
+        final Policy policy = TestConstants.Policy.POLICY.toBuilder()
+                .forLabel(label)
+                .setSubject(TestConstants.Policy.SUPPORT_SUBJECT)
+                .setGrantedPermissions(ResourceKey.newInstance("policy:/"), Permission.READ)
+                .build();
+        assertErrorResult(underTest, policy, command, activateTokenIntegrationStrategy.getNotApplicableException());
+    }
+
+    @Test
+    public void deactivateTokenIntegration() {
+        final CommandStrategy.Context<PolicyId> context = getDefaultContext();
+        final SubjectId subjectId =
+                SubjectId.newInstance(SubjectIssuer.INTEGRATION, "{{policy-entry:label}}:this-is-me");
+        final SubjectId expectedSubjectId =
+                SubjectId.newInstance(SubjectIssuer.INTEGRATION, LABEL + ":this-is-me");
+        final DittoHeaders dittoHeaders = DittoHeaders.empty();
+        final TopLevelActionCommand command = TopLevelActionCommand.of(
+                DeactivateTokenIntegration.of(context.getState(), DUMMY_LABEL, subjectId, dittoHeaders),
+                List.of(LABEL)
+        );
+        final Policy policy = TestConstants.Policy.POLICY.toBuilder()
+                .setSubjectFor(LABEL, PoliciesModelFactory.newSubject(expectedSubjectId,
+                        PoliciesModelFactory.newSubjectType(MessageFormat.format("via action <{0}>",
+                                ActivatePolicyTokenIntegration.NAME)),
+                        SubjectExpiry.newInstance(Instant.now().plus(Duration.ofDays(1L)))))
+                .build();
+        assertModificationResult(underTest, policy, command,
+                SubjectsDeletedPartially.class,
+                event -> {
+                    assertThat(event.getDeletedSubjectIds()).containsOnlyKeys(LABEL);
+                    assertThat(event.getDeletedSubjectIds().get(LABEL)).isEqualTo(expectedSubjectId);
+                },
+                TopLevelActionCommandResponse.class,
+                response -> assertThat(response)
+                        .isEqualTo(TopLevelActionCommandResponse.of(context.getState(), dittoHeaders)));
+    }
+
+    @Test
+    public void deactivateNonexistentSubject() {
+        final CommandStrategy.Context<PolicyId> context = getDefaultContext();
+        final SubjectId subjectId =
+                SubjectId.newInstance(SubjectIssuer.INTEGRATION, "{{policy-entry:label}}:this-is-me");
+        final DittoHeaders dittoHeaders = DittoHeaders.empty();
+        final TopLevelActionCommand command = TopLevelActionCommand.of(
+                DeactivateTokenIntegration.of(context.getState(), DUMMY_LABEL, subjectId, dittoHeaders),
+                List.of(LABEL)
+        );
+        assertModificationResult(underTest, TestConstants.Policy.POLICY, command,
+                SubjectsDeletedPartially.class,
+                // nonexistent subjects are present in the event.
+                event -> assertThat(event.getDeletedSubjectIds()).containsExactly(
+                        Map.entry(LABEL, SubjectId.newInstance("integration:" + LABEL + ":this-is-me"))),
+                TopLevelActionCommandResponse.class,
+                response -> assertThat(response)
+                        .isEqualTo(TopLevelActionCommandResponse.of(context.getState(), dittoHeaders)));
+    }
+
+    @Test
+    public void deactivatePermanentSubject() {
+        final CommandStrategy.Context<PolicyId> context = getDefaultContext();
+        final SubjectId subjectId =
+                SubjectId.newInstance(SubjectIssuer.INTEGRATION, "{{policy-entry:label}}:this-is-me");
+        final SubjectId expectedSubjectId =
+                SubjectId.newInstance(SubjectIssuer.INTEGRATION, LABEL + ":this-is-me");
+        final DittoHeaders dittoHeaders = DittoHeaders.empty();
+        final TopLevelActionCommand command = TopLevelActionCommand.of(
+                DeactivateTokenIntegration.of(context.getState(), DUMMY_LABEL, subjectId, dittoHeaders),
+                List.of(LABEL)
+        );
+        final Policy policy = TestConstants.Policy.POLICY.toBuilder()
+                .setSubjectFor(LABEL, PoliciesModelFactory.newSubject(expectedSubjectId,
+                        PoliciesModelFactory.newSubjectType(MessageFormat.format("added via action <{0}>",
+                                ActivatePolicyTokenIntegration.NAME))))
+                .build();
+        assertErrorResult(underTest, policy, command,
+                PolicyActionFailedException.newBuilderForDeactivatingPermanentSubjects().build());
+    }
+
+    @Test
+    public void deactivateInvalidSubject() {
+        final CommandStrategy.Context<PolicyId> context = getDefaultContext();
+        final SubjectId subjectId = SubjectId.newInstance("{{policy-entry:label}}");
+        final DittoHeaders dittoHeaders = DittoHeaders.empty();
+        final TopLevelActionCommand command = TopLevelActionCommand.of(
+                DeactivateTokenIntegration.of(context.getState(), DUMMY_LABEL, subjectId, dittoHeaders),
+                List.of(LABEL)
+        );
+        assertErrorResult(underTest, TestConstants.Policy.POLICY, command,
+                SubjectIdInvalidException.newBuilder(LABEL).build());
+    }
+
+    @Test
+    public void deactivateUnresolvableSubject() {
+        final CommandStrategy.Context<PolicyId> context = getDefaultContext();
+        final SubjectId subjectId = SubjectId.newInstance(SubjectIssuer.INTEGRATION, "{{fn:delete()}}");
+        final DittoHeaders dittoHeaders = DittoHeaders.empty();
+        final TopLevelActionCommand command = TopLevelActionCommand.of(
+                DeactivateTokenIntegration.of(context.getState(), DUMMY_LABEL, subjectId, dittoHeaders),
+                List.of(LABEL)
+        );
+        assertErrorResult(underTest, TestConstants.Policy.POLICY, command,
+                UnresolvedPlaceholderException.newBuilder("integration:{{fn:delete()}}").build());
+    }
+
+    @Test
+    public void deactivateTokenIntegrationWithUnsupportedPlaceholder() {
+        final CommandStrategy.Context<PolicyId> context = getDefaultContext();
+        final SubjectId subjectId = SubjectId.newInstance("{{request:subjectId}}");
+        final DittoHeaders dittoHeaders = DittoHeaders.empty();
+        final TopLevelActionCommand command = TopLevelActionCommand.of(
+                DeactivateTokenIntegration.of(context.getState(), DUMMY_LABEL, subjectId, dittoHeaders),
+                List.of(LABEL)
+        );
+        assertErrorResult(underTest, TestConstants.Policy.POLICY, command,
+                UnresolvedPlaceholderException.newBuilder("{{request:subjectId}}").build());
+    }
+
+}
\ No newline at end of file
diff --git a/services/utils/persistent-actors/src/main/java/org/eclipse/ditto/services/utils/persistentactors/commands/AbstractCommandStrategies.java b/services/utils/persistent-actors/src/main/java/org/eclipse/ditto/services/utils/persistentactors/commands/AbstractCommandStrategies.java
index 1049b02cca..a2abe14d68 100644
--- a/services/utils/persistent-actors/src/main/java/org/eclipse/ditto/services/utils/persistentactors/commands/AbstractCommandStrategies.java
+++ b/services/utils/persistent-actors/src/main/java/org/eclipse/ditto/services/utils/persistentactors/commands/AbstractCommandStrategies.java
@@ -22,6 +22,7 @@
 import org.eclipse.ditto.model.base.entity.metadata.Metadata;
 import org.eclipse.ditto.services.utils.persistentactors.results.Result;
 import org.eclipse.ditto.signals.commands.base.Command;
+import org.eclipse.ditto.signals.events.base.Event;
 
 /**
  * This <em>Singleton</em> delegates a {@code Command} to a dedicated strategy - if one is available - to be handled.
@@ -35,7 +36,7 @@
 public abstract class AbstractCommandStrategies<C extends Command<?>, S, K, R extends Result<?>>
         extends AbstractCommandStrategy<C, S, K, R> {
 
-    protected final Map<Class<? extends C>, CommandStrategy<? extends C, S, K, R>> strategies;
+    protected final Map<Class<? extends C>, CommandStrategy<? extends C, S, K, ? extends R>> strategies;
 
     /**
      * Constructs a new {@code AbstractCommandStrategy} object.
@@ -59,7 +60,7 @@ protected AbstractCommandStrategies(final Class theMatchingClass) {
      *
      * @param strategy the strategy.
      */
-    protected void addStrategy(final CommandStrategy<? extends C, S, K, R> strategy) {
+    protected void addStrategy(final CommandStrategy<? extends C, S, K, ? extends R> strategy) {
         final Class<? extends C> matchingClass = strategy.getMatchingClass();
         strategies.put(matchingClass, strategy);
     }
@@ -108,4 +109,23 @@ private CommandStrategy<C, S, K, R> getAppropriateStrategy(final Class<?> comman
         return (CommandStrategy<C, S, K, R>) strategies.get(commandClass);
     }
 
+    /**
+     * Add command strategy in a covariant way.
+     * TODO: Replace Result by event type; hard code the result type, then delete this method.
+     *
+     * @param strategies the command strategies.
+     * @param strategy the strategy to add.
+     * @param <E> type of events.
+     * @param <C> type of commands.
+     * @param <S> type of entities.
+     * @param <K> type of contexts.
+     */
+    @SuppressWarnings({"unchecked", "rawtypes"})
+    protected static <E extends Event<?>, F extends E, C extends Command<?>, S, K> void addStrategy(
+            final AbstractCommandStrategies<C, S, K, Result<E>> strategies,
+            CommandStrategy<? extends C, S, K, Result<F>> strategy) {
+
+        strategies.strategies.put(strategy.getMatchingClass(), (CommandStrategy) strategy);
+    }
+
 }
diff --git a/services/utils/persistent-actors/src/main/java/org/eclipse/ditto/services/utils/persistentactors/commands/CommandStrategy.java b/services/utils/persistent-actors/src/main/java/org/eclipse/ditto/services/utils/persistentactors/commands/CommandStrategy.java
index a207799187..d2d72db67b 100644
--- a/services/utils/persistent-actors/src/main/java/org/eclipse/ditto/services/utils/persistentactors/commands/CommandStrategy.java
+++ b/services/utils/persistent-actors/src/main/java/org/eclipse/ditto/services/utils/persistentactors/commands/CommandStrategy.java
@@ -12,6 +12,8 @@
  */
 package org.eclipse.ditto.services.utils.persistentactors.commands;
 
+import java.util.Optional;
+
 import javax.annotation.Nullable;
 
 import org.eclipse.ditto.services.utils.akka.logging.DittoDiagnosticLoggingAdapter;
@@ -66,6 +68,27 @@ default boolean isDefined(final Context<K> context, @Nullable final S entity, fi
         return isDefined(command);
     }
 
+    /**
+     * Applies the strategy to the given command of unknown type using the given context.
+     *
+     * @param context the context.
+     * @param entity the current entity of the persistence actor.
+     * @param nextRevision the next revision number of the entity.
+     * @param command the command of unknown type.
+     * @return the result of the strategy if the strategy is defined for the command, or an empty optional otherwise.
+     */
+    default Optional<R> typeCheckAndApply(final Context<K> context, @Nullable final S entity, final long nextRevision,
+            final Object command) {
+
+        if (getMatchingClass().isInstance(command)) {
+            final C theCommand = getMatchingClass().cast(command);
+            if (isDefined(context, entity, theCommand)) {
+                return Optional.of(apply(context, entity, nextRevision, theCommand));
+            }
+        }
+        return Optional.empty();
+    }
+
     /**
      * The Context in which a strategy is executed.
      *
diff --git a/signals/commands/policies/src/main/java/org/eclipse/ditto/signals/commands/policies/actions/ActivatePolicyTokenIntegration.java b/signals/commands/policies/src/main/java/org/eclipse/ditto/signals/commands/policies/actions/ActivatePolicyTokenIntegration.java
index 050bea9636..0ab0431572 100755
--- a/signals/commands/policies/src/main/java/org/eclipse/ditto/signals/commands/policies/actions/ActivatePolicyTokenIntegration.java
+++ b/signals/commands/policies/src/main/java/org/eclipse/ditto/signals/commands/policies/actions/ActivatePolicyTokenIntegration.java
@@ -41,6 +41,7 @@
 import org.eclipse.ditto.model.base.json.JsonSchemaVersion;
 import org.eclipse.ditto.model.policies.Label;
 import org.eclipse.ditto.model.policies.PoliciesModelFactory;
+import org.eclipse.ditto.model.policies.PolicyEntry;
 import org.eclipse.ditto.model.policies.PolicyId;
 import org.eclipse.ditto.model.policies.SubjectId;
 import org.eclipse.ditto.signals.commands.base.AbstractCommand;
@@ -146,6 +147,16 @@ public SubjectId getSubjectId() {
         return subjectId;
     }
 
+    @Override
+    public ActivatePolicyTokenIntegration setLabel(final Label label) {
+        return this;
+    }
+
+    @Override
+    public boolean isApplicable(final PolicyEntry policyEntry) {
+        return false;
+    }
+
     /**
      * Returns the expiry of the subject ID.
      *
diff --git a/signals/commands/policies/src/main/java/org/eclipse/ditto/signals/commands/policies/actions/ActivateTokenIntegration.java b/signals/commands/policies/src/main/java/org/eclipse/ditto/signals/commands/policies/actions/ActivateTokenIntegration.java
index 058737deb2..561f43f9ca 100755
--- a/signals/commands/policies/src/main/java/org/eclipse/ditto/signals/commands/policies/actions/ActivateTokenIntegration.java
+++ b/signals/commands/policies/src/main/java/org/eclipse/ditto/signals/commands/policies/actions/ActivateTokenIntegration.java
@@ -34,9 +34,12 @@
 import org.eclipse.ditto.model.base.json.FieldType;
 import org.eclipse.ditto.model.base.json.JsonParsableCommand;
 import org.eclipse.ditto.model.base.json.JsonSchemaVersion;
+import org.eclipse.ditto.model.policies.EffectedPermissions;
 import org.eclipse.ditto.model.policies.Label;
 import org.eclipse.ditto.model.policies.PoliciesModelFactory;
+import org.eclipse.ditto.model.policies.PoliciesResourceType;
 import org.eclipse.ditto.model.policies.Policy;
+import org.eclipse.ditto.model.policies.PolicyEntry;
 import org.eclipse.ditto.model.policies.PolicyId;
 import org.eclipse.ditto.model.policies.SubjectId;
 import org.eclipse.ditto.signals.commands.base.AbstractCommand;
@@ -139,6 +142,26 @@ public SubjectId getSubjectId() {
         return subjectId;
     }
 
+    @Override
+    public ActivateTokenIntegration setLabel(final Label label) {
+        return new ActivateTokenIntegration(policyId, label, subjectId, expiry, getDittoHeaders());
+    }
+
+    @Override
+    public boolean isApplicable(final PolicyEntry policyEntry) {
+        // This action is applicable to policy entries containing a READ permission grated to a thing resource.
+        final String readPermission = "READ";
+        return policyEntry.getResources()
+                .stream()
+                .anyMatch(resource -> {
+                    final String resourceType = resource.getResourceKey().getResourceType();
+                    final EffectedPermissions permissions = resource.getEffectedPermissions();
+                    return PoliciesResourceType.THING.equals(resourceType) &&
+                            permissions.getGrantedPermissions().contains(readPermission) &&
+                            !permissions.getRevokedPermissions().contains(readPermission);
+                });
+    }
+
     /**
      * Returns the expiry of the subject ID.
      *
diff --git a/signals/commands/policies/src/main/java/org/eclipse/ditto/signals/commands/policies/actions/DeactivatePolicyTokenIntegration.java b/signals/commands/policies/src/main/java/org/eclipse/ditto/signals/commands/policies/actions/DeactivatePolicyTokenIntegration.java
index 17c56f3940..4e0b2d873b 100755
--- a/signals/commands/policies/src/main/java/org/eclipse/ditto/signals/commands/policies/actions/DeactivatePolicyTokenIntegration.java
+++ b/signals/commands/policies/src/main/java/org/eclipse/ditto/signals/commands/policies/actions/DeactivatePolicyTokenIntegration.java
@@ -39,6 +39,7 @@
 import org.eclipse.ditto.model.base.json.JsonSchemaVersion;
 import org.eclipse.ditto.model.policies.Label;
 import org.eclipse.ditto.model.policies.PoliciesModelFactory;
+import org.eclipse.ditto.model.policies.PolicyEntry;
 import org.eclipse.ditto.model.policies.PolicyId;
 import org.eclipse.ditto.model.policies.SubjectId;
 import org.eclipse.ditto.signals.commands.base.AbstractCommand;
@@ -133,6 +134,16 @@ public SubjectId getSubjectId() {
         return subjectId;
     }
 
+    @Override
+    public DeactivatePolicyTokenIntegration setLabel(final Label label) {
+        return this;
+    }
+
+    @Override
+    public boolean isApplicable(final PolicyEntry policyEntry) {
+        return false;
+    }
+
     /**
      * Returns the labels of policy entries from which the subject is to be removed.
      *
diff --git a/signals/commands/policies/src/main/java/org/eclipse/ditto/signals/commands/policies/actions/DeactivateTokenIntegration.java b/signals/commands/policies/src/main/java/org/eclipse/ditto/signals/commands/policies/actions/DeactivateTokenIntegration.java
index 4c522816a5..af4eaeae7b 100755
--- a/signals/commands/policies/src/main/java/org/eclipse/ditto/signals/commands/policies/actions/DeactivateTokenIntegration.java
+++ b/signals/commands/policies/src/main/java/org/eclipse/ditto/signals/commands/policies/actions/DeactivateTokenIntegration.java
@@ -34,6 +34,7 @@
 import org.eclipse.ditto.model.policies.Label;
 import org.eclipse.ditto.model.policies.PoliciesModelFactory;
 import org.eclipse.ditto.model.policies.Policy;
+import org.eclipse.ditto.model.policies.PolicyEntry;
 import org.eclipse.ditto.model.policies.PolicyId;
 import org.eclipse.ditto.model.policies.SubjectId;
 import org.eclipse.ditto.signals.commands.base.AbstractCommand;
@@ -129,6 +130,16 @@ public SubjectId getSubjectId() {
         return subjectId;
     }
 
+    @Override
+    public DeactivateTokenIntegration setLabel(final Label label) {
+        return new DeactivateTokenIntegration(policyId, label, subjectId, getDittoHeaders());
+    }
+
+    @Override
+    public boolean isApplicable(final PolicyEntry policyEntry) {
+        return true;
+    }
+
     @Override
     public PolicyId getEntityId() {
         return policyId;
diff --git a/signals/commands/policies/src/main/java/org/eclipse/ditto/signals/commands/policies/actions/PolicyActionCommand.java b/signals/commands/policies/src/main/java/org/eclipse/ditto/signals/commands/policies/actions/PolicyActionCommand.java
index a070591f80..d8233ab4dc 100755
--- a/signals/commands/policies/src/main/java/org/eclipse/ditto/signals/commands/policies/actions/PolicyActionCommand.java
+++ b/signals/commands/policies/src/main/java/org/eclipse/ditto/signals/commands/policies/actions/PolicyActionCommand.java
@@ -13,6 +13,8 @@
 package org.eclipse.ditto.signals.commands.policies.actions;
 
 import org.eclipse.ditto.json.JsonPointer;
+import org.eclipse.ditto.model.policies.Label;
+import org.eclipse.ditto.model.policies.PolicyEntry;
 import org.eclipse.ditto.model.policies.SubjectId;
 import org.eclipse.ditto.signals.base.WithOptionalEntity;
 import org.eclipse.ditto.signals.commands.policies.PolicyCommand;
@@ -37,6 +39,22 @@ public interface PolicyActionCommand<T extends PolicyActionCommand<T>> extends P
      */
     SubjectId getSubjectId();
 
+    /**
+     * Set the label of the policy entry where this action is executed, if applicable.
+     *
+     * @param label the policy entry label to execute the action.
+     * @return a new command to execute the action at the new policy entry if applicable, or this object if not.
+     */
+    T setLabel(Label label);
+
+    /**
+     * Check if this command is applicable to a policy entry.
+     *
+     * @param policyEntry the policy entry.
+     * @return whether this command is applicable.
+     */
+    boolean isApplicable(PolicyEntry policyEntry);
+
     @Override
     default Category getCategory() {
         return Category.ACTION;
diff --git a/signals/commands/policies/src/main/java/org/eclipse/ditto/signals/commands/policies/actions/TopLevelActionCommand.java b/signals/commands/policies/src/main/java/org/eclipse/ditto/signals/commands/policies/actions/TopLevelActionCommand.java
new file mode 100755
index 0000000000..2151c14ce5
--- /dev/null
+++ b/signals/commands/policies/src/main/java/org/eclipse/ditto/signals/commands/policies/actions/TopLevelActionCommand.java
@@ -0,0 +1,233 @@
+/*
+ * Copyright (c) 2021 Contributors to the Eclipse Foundation
+ *
+ * See the NOTICE file(s) distributed with this work for additional
+ * information regarding copyright ownership.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License 2.0 which is available at
+ * http://www.eclipse.org/legal/epl-2.0
+ *
+ * SPDX-License-Identifier: EPL-2.0
+ */
+package org.eclipse.ditto.signals.commands.policies.actions;
+
+import static org.eclipse.ditto.model.base.common.ConditionChecker.checkNotNull;
+
+import java.io.NotSerializableException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+
+import javax.annotation.Nullable;
+import javax.annotation.concurrent.Immutable;
+
+import org.eclipse.ditto.json.JsonArray;
+import org.eclipse.ditto.json.JsonCollectors;
+import org.eclipse.ditto.json.JsonFactory;
+import org.eclipse.ditto.json.JsonField;
+import org.eclipse.ditto.json.JsonFieldDefinition;
+import org.eclipse.ditto.json.JsonKey;
+import org.eclipse.ditto.json.JsonObject;
+import org.eclipse.ditto.json.JsonObjectBuilder;
+import org.eclipse.ditto.json.JsonParseException;
+import org.eclipse.ditto.json.JsonPointer;
+import org.eclipse.ditto.json.JsonValue;
+import org.eclipse.ditto.model.base.headers.DittoHeaders;
+import org.eclipse.ditto.model.base.json.FieldType;
+import org.eclipse.ditto.model.base.json.JsonParsableCommand;
+import org.eclipse.ditto.model.base.json.JsonSchemaVersion;
+import org.eclipse.ditto.model.policies.Label;
+import org.eclipse.ditto.model.policies.PolicyEntry;
+import org.eclipse.ditto.model.policies.PolicyId;
+import org.eclipse.ditto.model.policies.SubjectId;
+import org.eclipse.ditto.signals.base.JsonParsable;
+import org.eclipse.ditto.signals.commands.base.AbstractCommand;
+import org.eclipse.ditto.signals.commands.base.CommandJsonDeserializer;
+import org.eclipse.ditto.signals.commands.policies.PolicyCommand;
+
+/**
+ * This command executes a policy action on every authorized entries.
+ *
+ * @since 2.0.0
+ */
+@Immutable
+@JsonParsableCommand(typePrefix = PolicyCommand.TYPE_PREFIX, name = TopLevelActionCommand.NAME)
+public final class TopLevelActionCommand extends AbstractCommand<TopLevelActionCommand>
+        implements PolicyActionCommand<TopLevelActionCommand> {
+
+    /**
+     * NAME of this command.
+     */
+    public static final String NAME = "topLevelAction";
+
+    /**
+     * Type of this command.
+     */
+    public static final String TYPE = TYPE_PREFIX + NAME;
+
+    static final JsonFieldDefinition<JsonObject> JSON_ACTION =
+            JsonFactory.newJsonObjectFieldDefinition("action", FieldType.REGULAR, JsonSchemaVersion.V_2);
+
+    static final JsonFieldDefinition<JsonArray> JSON_LABELS =
+            JsonFactory.newJsonArrayFieldDefinition("labels", FieldType.REGULAR, JsonSchemaVersion.V_2);
+
+    private final PolicyActionCommand<?> policyActionCommand;
+    private final List<Label> authorizedLabels;
+
+    private TopLevelActionCommand(final PolicyActionCommand<?> policyActionCommand,
+            final List<Label> authorizedLabels) {
+        super(TYPE, policyActionCommand.getDittoHeaders());
+        // Null check and copying in the factory method in order to share known unmodifiable fields between instances.
+        this.authorizedLabels = authorizedLabels;
+        this.policyActionCommand = policyActionCommand;
+    }
+
+    /**
+     * Creates a command for executing a policy action on all authorized entries.
+     *
+     * @param policyActionCommand the action to execute.
+     * @param authorizedLabels labels of policy entries authorized to execute the action.
+     * @return the command.
+     * @throws NullPointerException if any argument is {@code null}.
+     */
+    public static TopLevelActionCommand of(final PolicyActionCommand<?> policyActionCommand,
+            final List<Label> authorizedLabels) {
+
+        return new TopLevelActionCommand(
+                policyActionCommand,
+                Collections.unmodifiableList(new ArrayList<>(checkNotNull(authorizedLabels, "labels")))
+        );
+    }
+
+    /**
+     * Creates a command for executing a policy action on all authorized entries.
+     *
+     * @param jsonObject the JSON object of which the command is to be created.
+     * @param dittoHeaders headers of the command.
+     * @param parseInnerJson function to parse the inner JSON of the command.
+     * @return the command.
+     * @throws NullPointerException if {@code jsonObject} is {@code null}.
+     * @throws org.eclipse.ditto.json.JsonParseException if the passed in {@code jsonObject} was not in the expected
+     * format.
+     */
+    public static TopLevelActionCommand fromJson(final JsonObject jsonObject,
+            final DittoHeaders dittoHeaders,
+            final JsonParsable.ParseInnerJson parseInnerJson) {
+
+        return new CommandJsonDeserializer<TopLevelActionCommand>(TYPE, jsonObject).deserialize(() -> {
+            try {
+                final JsonObject commandJson = jsonObject.getValueOrThrow(JSON_ACTION);
+                final PolicyActionCommand<?> policyActionCommand =
+                        (PolicyActionCommand<?>) parseInnerJson.parseInnerJson(commandJson);
+                final List<Label> labels = Collections.unmodifiableList(
+                        jsonObject.getValueOrThrow(JSON_LABELS)
+                                .stream()
+                                .map(JsonValue::asString)
+                                .map(Label::of)
+                                .collect(Collectors.toList())
+                );
+                return new TopLevelActionCommand(policyActionCommand.setDittoHeaders(dittoHeaders), labels);
+            } catch (final NotSerializableException e) {
+                throw new JsonParseException(e.getMessage());
+            }
+        });
+    }
+
+    @Override
+    public SubjectId getSubjectId() {
+        return policyActionCommand.getSubjectId();
+    }
+
+    @Override
+    public TopLevelActionCommand setLabel(final Label label) {
+        return this;
+    }
+
+    @Override
+    public boolean isApplicable(final PolicyEntry policyEntry) {
+        return false;
+    }
+
+    /**
+     * Returns the labels of policy entries authorized to execute the action.
+     *
+     * @return the policy entry labels.
+     */
+    public List<Label> getAuthorizedLabels() {
+        return authorizedLabels;
+    }
+
+    /**
+     * Returns the policy action command to execute on each authorized entry.
+     *
+     * @return the policy action command.
+     */
+    public PolicyActionCommand<?> getPolicyActionCommand() {
+        return policyActionCommand;
+    }
+
+    @Override
+    public PolicyId getEntityId() {
+        return policyActionCommand.getEntityId();
+    }
+
+    @Override
+    public JsonPointer getResourcePath() {
+        // actions/<ACTION-NAME>
+        return RESOURCE_PATH_ACTIONS.addLeaf(JsonKey.of(policyActionCommand.getName()));
+    }
+
+    @Override
+    protected void appendPayload(final JsonObjectBuilder jsonObjectBuilder, final JsonSchemaVersion schemaVersion,
+            final Predicate<JsonField> thePredicate) {
+
+        final Predicate<JsonField> predicate = schemaVersion.and(thePredicate);
+        final JsonObject policyActionCommandJson = policyActionCommand.toJson(schemaVersion, thePredicate);
+        final JsonArray authorizedLabelsJson =
+                authorizedLabels.stream().map(JsonValue::of).collect(JsonCollectors.valuesToArray());
+        jsonObjectBuilder.set(JSON_ACTION, policyActionCommandJson, predicate);
+        jsonObjectBuilder.set(JSON_LABELS, authorizedLabelsJson, predicate);
+    }
+
+    @Override
+    public TopLevelActionCommand setDittoHeaders(final DittoHeaders dittoHeaders) {
+        return new TopLevelActionCommand(policyActionCommand.setDittoHeaders(dittoHeaders), authorizedLabels);
+    }
+
+    @Override
+    protected boolean canEqual(@Nullable final Object other) {
+        return other instanceof TopLevelActionCommand;
+    }
+
+    @Override
+    public boolean equals(@Nullable final Object obj) {
+        if (this == obj) {
+            return true;
+        }
+        if (null == obj || getClass() != obj.getClass()) {
+            return false;
+        }
+        final TopLevelActionCommand that = (TopLevelActionCommand) obj;
+        return Objects.equals(policyActionCommand, that.policyActionCommand) &&
+                Objects.equals(authorizedLabels, that.authorizedLabels) &&
+                super.equals(obj);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(super.hashCode(), policyActionCommand, authorizedLabels);
+    }
+
+    @Override
+    public String toString() {
+        return getClass().getSimpleName() + " [" + super.toString() +
+                ", policyActionCommand=" + policyActionCommand +
+                ", labels=" + authorizedLabels +
+                "]";
+    }
+
+}
diff --git a/signals/commands/policies/src/main/java/org/eclipse/ditto/signals/commands/policies/actions/TopLevelActionCommandResponse.java b/signals/commands/policies/src/main/java/org/eclipse/ditto/signals/commands/policies/actions/TopLevelActionCommandResponse.java
new file mode 100755
index 0000000000..22acedd2c9
--- /dev/null
+++ b/signals/commands/policies/src/main/java/org/eclipse/ditto/signals/commands/policies/actions/TopLevelActionCommandResponse.java
@@ -0,0 +1,147 @@
+/*
+ * Copyright (c) 2021 Contributors to the Eclipse Foundation
+ *
+ * See the NOTICE file(s) distributed with this work for additional
+ * information regarding copyright ownership.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License 2.0 which is available at
+ * http://www.eclipse.org/legal/epl-2.0
+ *
+ * SPDX-License-Identifier: EPL-2.0
+ */
+package org.eclipse.ditto.signals.commands.policies.actions;
+
+import static org.eclipse.ditto.model.base.common.ConditionChecker.checkNotNull;
+
+import java.util.Objects;
+import java.util.function.Predicate;
+
+import javax.annotation.Nullable;
+import javax.annotation.concurrent.Immutable;
+
+import org.eclipse.ditto.json.JsonField;
+import org.eclipse.ditto.json.JsonObject;
+import org.eclipse.ditto.json.JsonObjectBuilder;
+import org.eclipse.ditto.json.JsonPointer;
+import org.eclipse.ditto.model.base.common.HttpStatusCode;
+import org.eclipse.ditto.model.base.headers.DittoHeaders;
+import org.eclipse.ditto.model.base.json.JsonParsableCommandResponse;
+import org.eclipse.ditto.model.base.json.JsonSchemaVersion;
+import org.eclipse.ditto.model.policies.PolicyId;
+import org.eclipse.ditto.signals.commands.base.AbstractCommandResponse;
+import org.eclipse.ditto.signals.commands.policies.PolicyCommandResponse;
+
+/**
+ * Response to an {@link TopLevelActionCommand} command.
+ *
+ * @since 2.0.0
+ */
+@Immutable
+@JsonParsableCommandResponse(type = TopLevelActionCommandResponse.TYPE)
+public final class TopLevelActionCommandResponse
+        extends AbstractCommandResponse<TopLevelActionCommandResponse>
+        implements PolicyActionCommandResponse<TopLevelActionCommandResponse> {
+
+    /**
+     * Type of this response.
+     */
+    public static final String TYPE = TYPE_PREFIX + TopLevelActionCommand.NAME;
+
+    /**
+     * Status code of this response.
+     */
+    public static final HttpStatusCode STATUS = HttpStatusCode.NO_CONTENT;
+
+    private final PolicyId policyId;
+
+    private TopLevelActionCommandResponse(final PolicyId policyId, final DittoHeaders dittoHeaders) {
+        super(TYPE, STATUS, dittoHeaders);
+        this.policyId = checkNotNull(policyId, "policyId");
+    }
+
+    /**
+     * Creates a response to an {@code TopLevelActionCommand} command.
+     *
+     * @param policyId the policy ID.
+     * @param dittoHeaders the headers of the preceding command.
+     * @return the response.
+     * @throws NullPointerException if any argument is {@code null}.
+     */
+    public static TopLevelActionCommandResponse of(final PolicyId policyId, final DittoHeaders dittoHeaders) {
+        return new TopLevelActionCommandResponse(policyId, dittoHeaders);
+    }
+
+    /**
+     * Creates a response to a {@code TopLevelActionCommand} command from a JSON object.
+     *
+     * @param jsonObject the JSON object of which the response is to be created.
+     * @param dittoHeaders the headers of the preceding command.
+     * @return the response.
+     * @throws NullPointerException if {@code jsonObject} is {@code null}.
+     * @throws org.eclipse.ditto.json.JsonParseException if the passed in {@code jsonObject} was not in the expected
+     * format.
+     */
+    @SuppressWarnings("unused") // called by reflection
+    public static TopLevelActionCommandResponse fromJson(final JsonObject jsonObject,
+            final DittoHeaders dittoHeaders) {
+        final PolicyId policyId =
+                PolicyId.of(jsonObject.getValueOrThrow(PolicyCommandResponse.JsonFields.JSON_POLICY_ID));
+        return new TopLevelActionCommandResponse(policyId, dittoHeaders);
+    }
+
+    @Override
+    public PolicyId getEntityId() {
+        return policyId;
+    }
+
+    @Override
+    public JsonPointer getResourcePath() {
+        return JsonPointer.empty();
+    }
+
+    @Override
+    protected void appendPayload(final JsonObjectBuilder jsonObjectBuilder, final JsonSchemaVersion schemaVersion,
+            final Predicate<JsonField> thePredicate) {
+
+        final Predicate<JsonField> predicate = schemaVersion.and(thePredicate);
+        jsonObjectBuilder.set(PolicyCommandResponse.JsonFields.JSON_POLICY_ID, policyId.toString(), predicate);
+    }
+
+    @Override
+    public TopLevelActionCommandResponse setDittoHeaders(final DittoHeaders dittoHeaders) {
+        return new TopLevelActionCommandResponse(policyId, dittoHeaders);
+    }
+
+    @Override
+    protected boolean canEqual(@Nullable final Object other) {
+        return other instanceof TopLevelActionCommandResponse;
+    }
+
+    @Override
+    public boolean equals(@Nullable final Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        final TopLevelActionCommandResponse that = (TopLevelActionCommandResponse) o;
+        return Objects.equals(policyId, that.policyId) &&
+                super.equals(o);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(super.hashCode(), policyId);
+    }
+
+    @Override
+    public String toString() {
+        return getClass().getSimpleName() +
+                " [" + super.toString() +
+                ", policyId=" + policyId +
+                "]";
+    }
+
+}
diff --git a/signals/commands/policies/src/main/java/org/eclipse/ditto/signals/commands/policies/exceptions/PolicyActionFailedException.java b/signals/commands/policies/src/main/java/org/eclipse/ditto/signals/commands/policies/exceptions/PolicyActionFailedException.java
index 396e527958..e3b6682610 100755
--- a/signals/commands/policies/src/main/java/org/eclipse/ditto/signals/commands/policies/exceptions/PolicyActionFailedException.java
+++ b/signals/commands/policies/src/main/java/org/eclipse/ditto/signals/commands/policies/exceptions/PolicyActionFailedException.java
@@ -60,6 +60,15 @@ private PolicyActionFailedException(final DittoHeaders dittoHeaders,
         super(ERROR_CODE, status, dittoHeaders, message, description, cause, href);
     }
 
+    /**
+     * A mutable builder for a {@code PolicyActionFailedException}.
+     *
+     * @return the builder.
+     */
+    public static Builder newBuilder() {
+        return new Builder();
+    }
+
     /**
      * A mutable builder for a {@code PolicyActionFailedException} for the action {@code activateTokenIntegration}.
      *
diff --git a/signals/commands/policies/src/main/java/org/eclipse/ditto/signals/commands/policies/exceptions/PolicyCommandToActionsExceptionRegistry.java b/signals/commands/policies/src/main/java/org/eclipse/ditto/signals/commands/policies/exceptions/PolicyCommandToActionsExceptionRegistry.java
index 62fddaa538..b547b4f84d 100755
--- a/signals/commands/policies/src/main/java/org/eclipse/ditto/signals/commands/policies/exceptions/PolicyCommandToActionsExceptionRegistry.java
+++ b/signals/commands/policies/src/main/java/org/eclipse/ditto/signals/commands/policies/exceptions/PolicyCommandToActionsExceptionRegistry.java
@@ -20,10 +20,9 @@
 import org.eclipse.ditto.model.base.exceptions.DittoRuntimeException;
 import org.eclipse.ditto.signals.commands.base.AbstractCommandToExceptionRegistry;
 import org.eclipse.ditto.signals.commands.policies.PolicyCommand;
-import org.eclipse.ditto.signals.commands.policies.actions.ActivatePolicyTokenIntegration;
 import org.eclipse.ditto.signals.commands.policies.actions.ActivateTokenIntegration;
-import org.eclipse.ditto.signals.commands.policies.actions.DeactivatePolicyTokenIntegration;
 import org.eclipse.ditto.signals.commands.policies.actions.DeactivateTokenIntegration;
+import org.eclipse.ditto.signals.commands.policies.actions.TopLevelActionCommand;
 
 /**
  * Registry to map policy commands to their according policy actions exception.
@@ -51,8 +50,9 @@ private static PolicyCommandToActionsExceptionRegistry createInstance() {
         final Map<String, Function<PolicyCommand<?>, DittoRuntimeException>> mappingStrategies = new HashMap<>();
 
         final String insufficientPermissions = "The requester has insufficient permissions. 'EXECUTE' is required.";
-        mappingStrategies.put(ActivatePolicyTokenIntegration.TYPE,
-                command -> PolicyActionFailedException.newBuilderForActivateTokenIntegration()
+        mappingStrategies.put(TopLevelActionCommand.TYPE,
+                command -> PolicyActionFailedException.newBuilder()
+                        .action(((TopLevelActionCommand) command).getPolicyActionCommand().getName())
                         .status(HttpStatusCode.FORBIDDEN)
                         .description(insufficientPermissions)
                         .dittoHeaders(command.getDittoHeaders())
@@ -63,12 +63,6 @@ private static PolicyCommandToActionsExceptionRegistry createInstance() {
                         .description(insufficientPermissions)
                         .dittoHeaders(command.getDittoHeaders())
                         .build());
-        mappingStrategies.put(DeactivatePolicyTokenIntegration.TYPE,
-                command -> PolicyActionFailedException.newBuilderForDeactivateTokenIntegration()
-                        .status(HttpStatusCode.FORBIDDEN)
-                        .description(insufficientPermissions)
-                        .dittoHeaders(command.getDittoHeaders())
-                        .build());
         mappingStrategies.put(DeactivateTokenIntegration.TYPE,
                 command -> PolicyActionFailedException.newBuilderForDeactivateTokenIntegration()
                         .status(HttpStatusCode.FORBIDDEN)
diff --git a/signals/commands/policies/src/test/java/org/eclipse/ditto/signals/commands/policies/actions/TopLevelActionCommandResponseTest.java b/signals/commands/policies/src/test/java/org/eclipse/ditto/signals/commands/policies/actions/TopLevelActionCommandResponseTest.java
new file mode 100755
index 0000000000..f5ff770d62
--- /dev/null
+++ b/signals/commands/policies/src/test/java/org/eclipse/ditto/signals/commands/policies/actions/TopLevelActionCommandResponseTest.java
@@ -0,0 +1,81 @@
+/*
+ * Copyright (c) 2021 Contributors to the Eclipse Foundation
+ *
+ * See the NOTICE file(s) distributed with this work for additional
+ * information regarding copyright ownership.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License 2.0 which is available at
+ * http://www.eclipse.org/legal/epl-2.0
+ *
+ * SPDX-License-Identifier: EPL-2.0
+ */
+package org.eclipse.ditto.signals.commands.policies.actions;
+
+import static org.eclipse.ditto.json.assertions.DittoJsonAssertions.assertThat;
+import static org.mutabilitydetector.unittesting.AllowedReason.provided;
+import static org.mutabilitydetector.unittesting.MutabilityAssert.assertInstancesOf;
+import static org.mutabilitydetector.unittesting.MutabilityMatchers.areImmutable;
+
+import org.eclipse.ditto.json.JsonFactory;
+import org.eclipse.ditto.json.JsonObject;
+import org.eclipse.ditto.model.base.json.FieldType;
+import org.eclipse.ditto.model.policies.PolicyId;
+import org.eclipse.ditto.signals.commands.policies.PolicyCommandResponse;
+import org.eclipse.ditto.signals.commands.policies.TestConstants;
+import org.junit.Test;
+
+import nl.jqno.equalsverifier.EqualsVerifier;
+
+/**
+ * Unit test for {@link TopLevelActionCommandResponse}.
+ */
+public final class TopLevelActionCommandResponseTest {
+
+    private static final JsonObject KNOWN_JSON = JsonFactory.newObjectBuilder()
+            .set(PolicyCommandResponse.JsonFields.TYPE, TopLevelActionCommandResponse.TYPE)
+            .set(PolicyCommandResponse.JsonFields.STATUS, TopLevelActionCommandResponse.STATUS.toInt())
+            .set(PolicyCommandResponse.JsonFields.JSON_POLICY_ID, TestConstants.Policy.POLICY_ID.toString())
+            .build();
+
+    @Test
+    public void assertImmutability() {
+        assertInstancesOf(TopLevelActionCommandResponse.class,
+                areImmutable(),
+                provided(PolicyId.class).isAlsoImmutable());
+    }
+
+    @Test
+    public void testHashCodeAndEquals() {
+        EqualsVerifier.forClass(TopLevelActionCommandResponse.class)
+                .withRedefinedSuperclass()
+                .verify();
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void tryToCreateInstanceWithNullPolicyId() {
+        TopLevelActionCommandResponse.of(null, TestConstants.EMPTY_DITTO_HEADERS);
+    }
+
+    @Test
+    public void toJsonReturnsExpected() {
+        final TopLevelActionCommandResponse underTest =
+                TopLevelActionCommandResponse.of(TestConstants.Policy.POLICY_ID,
+                        TestConstants.EMPTY_DITTO_HEADERS);
+        final JsonObject actualJson = underTest.toJson(FieldType.regularOrSpecial());
+
+        assertThat(actualJson).isEqualTo(KNOWN_JSON);
+    }
+
+    @Test
+    public void createInstanceFromValidJson() {
+        final TopLevelActionCommandResponse underTest =
+                TopLevelActionCommandResponse.fromJson(KNOWN_JSON, TestConstants.EMPTY_DITTO_HEADERS);
+
+        final TopLevelActionCommandResponse expectedCommand =
+                TopLevelActionCommandResponse.of(TestConstants.Policy.POLICY_ID,
+                        TestConstants.EMPTY_DITTO_HEADERS);
+        assertThat(underTest).isEqualTo(expectedCommand);
+    }
+
+}
diff --git a/signals/commands/policies/src/test/java/org/eclipse/ditto/signals/commands/policies/actions/TopLevelActionCommandTest.java b/signals/commands/policies/src/test/java/org/eclipse/ditto/signals/commands/policies/actions/TopLevelActionCommandTest.java
new file mode 100755
index 0000000000..0a63c162cc
--- /dev/null
+++ b/signals/commands/policies/src/test/java/org/eclipse/ditto/signals/commands/policies/actions/TopLevelActionCommandTest.java
@@ -0,0 +1,98 @@
+/*
+ * Copyright (c) 2021 Contributors to the Eclipse Foundation
+ *
+ * See the NOTICE file(s) distributed with this work for additional
+ * information regarding copyright ownership.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License 2.0 which is available at
+ * http://www.eclipse.org/legal/epl-2.0
+ *
+ * SPDX-License-Identifier: EPL-2.0
+ */
+package org.eclipse.ditto.signals.commands.policies.actions;
+
+import static org.eclipse.ditto.json.assertions.DittoJsonAssertions.assertThat;
+import static org.mutabilitydetector.unittesting.AllowedReason.assumingFields;
+import static org.mutabilitydetector.unittesting.AllowedReason.provided;
+import static org.mutabilitydetector.unittesting.MutabilityAssert.assertInstancesOf;
+import static org.mutabilitydetector.unittesting.MutabilityMatchers.areImmutable;
+
+import java.time.Instant;
+import java.util.Collections;
+import java.util.List;
+
+import org.eclipse.ditto.json.JsonArray;
+import org.eclipse.ditto.json.JsonFactory;
+import org.eclipse.ditto.json.JsonObject;
+import org.eclipse.ditto.model.base.headers.DittoHeaders;
+import org.eclipse.ditto.model.base.json.FieldType;
+import org.eclipse.ditto.model.policies.Label;
+import org.eclipse.ditto.signals.commands.policies.PolicyCommand;
+import org.eclipse.ditto.signals.commands.policies.TestConstants;
+import org.junit.Test;
+
+import nl.jqno.equalsverifier.EqualsVerifier;
+
+/**
+ * Unit test for {@link TopLevelActionCommand}.
+ */
+public final class TopLevelActionCommandTest {
+
+    private static final List<Label> LABELS = Collections.singletonList(TestConstants.Policy.LABEL);
+
+    private static final ActivateTokenIntegration POLICY_ACTION_COMMAND =
+            ActivateTokenIntegration.of(TestConstants.Policy.POLICY_ID, TestConstants.Policy.LABEL,
+                    TestConstants.Policy.SUBJECT_ID, Instant.EPOCH, TestConstants.EMPTY_DITTO_HEADERS);
+
+    private static final JsonObject KNOWN_JSON = JsonFactory.newObjectBuilder()
+            .set(PolicyCommand.JsonFields.TYPE, TopLevelActionCommand.TYPE)
+            .set(TopLevelActionCommand.JSON_ACTION, POLICY_ACTION_COMMAND.toJson())
+            .set(TopLevelActionCommand.JSON_LABELS, JsonArray.of(TestConstants.Policy.LABEL))
+            .build();
+
+    @Test
+    public void assertImmutability() {
+        assertInstancesOf(TopLevelActionCommand.class,
+                areImmutable(),
+                provided(PolicyActionCommand.class).areAlsoImmutable(),
+                assumingFields("authorizedLabels").areSafelyCopiedUnmodifiableCollectionsWithImmutableElements());
+    }
+
+    @Test
+    public void testHashCodeAndEquals() {
+        EqualsVerifier.forClass(TopLevelActionCommand.class)
+                .withRedefinedSuperclass()
+                .verify();
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void tryToCreateInstanceWithNullCommand() {
+        TopLevelActionCommand.of(null, LABELS);
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void tryToCreateInstanceWithNullLabels() {
+        TopLevelActionCommand.of(POLICY_ACTION_COMMAND, null);
+    }
+
+    @Test
+    public void toJsonReturnsExpected() {
+        final TopLevelActionCommand underTest =
+                TopLevelActionCommand.of(POLICY_ACTION_COMMAND, LABELS);
+        final JsonObject actualJson = underTest.toJson(FieldType.regularOrSpecial());
+
+        assertThat(actualJson).isEqualTo(KNOWN_JSON);
+    }
+
+    @Test
+    public void createInstanceFromValidJson() {
+        final TopLevelActionCommand underTest =
+                TopLevelActionCommand.fromJson(KNOWN_JSON, TestConstants.EMPTY_DITTO_HEADERS,
+                        jsonObject -> ActivateTokenIntegration.fromJson(jsonObject, DittoHeaders.empty()));
+
+        final TopLevelActionCommand expectedCommand =
+                TopLevelActionCommand.of(POLICY_ACTION_COMMAND, LABELS);
+        assertThat(underTest).isEqualTo(expectedCommand);
+    }
+}
diff --git a/signals/events/policies/src/main/java/org/eclipse/ditto/signals/events/policies/AbstractPolicyActionEvent.java b/signals/events/policies/src/main/java/org/eclipse/ditto/signals/events/policies/AbstractPolicyActionEvent.java
new file mode 100755
index 0000000000..7491270ec7
--- /dev/null
+++ b/signals/events/policies/src/main/java/org/eclipse/ditto/signals/events/policies/AbstractPolicyActionEvent.java
@@ -0,0 +1,96 @@
+/*
+ * Copyright (c) 2017 Contributors to the Eclipse Foundation
+ *
+ * See the NOTICE file(s) distributed with this work for additional
+ * information regarding copyright ownership.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License 2.0 which is available at
+ * http://www.eclipse.org/legal/epl-2.0
+ *
+ * SPDX-License-Identifier: EPL-2.0
+ */
+package org.eclipse.ditto.signals.events.policies;
+
+import java.time.Instant;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+
+import javax.annotation.Nullable;
+
+import org.eclipse.ditto.model.base.headers.DittoHeaders;
+import org.eclipse.ditto.model.policies.Label;
+import org.eclipse.ditto.model.policies.PolicyId;
+import org.eclipse.ditto.model.policies.Subject;
+import org.eclipse.ditto.model.policies.SubjectId;
+
+/**
+ * Abstract base class of a {@link PolicyActionEvent}.
+ *
+ * @param <T> the type of the implementing class.
+ */
+abstract class AbstractPolicyActionEvent<T extends AbstractPolicyActionEvent<T>>
+        extends AbstractPolicyEvent<T> implements PolicyActionEvent<T> {
+
+    /**
+     * Constructs a new {@code AbstractPolicyActionEvent} object.
+     *
+     * @param type the type of this event.
+     * @param policyId the TYPE of the Policy with which this event is associated.
+     * @param revision the revision of the Policy.
+     * @param timestamp the timestamp of the event.
+     * @param dittoHeaders the headers of the command which was the cause of this event.
+     * @throws NullPointerException if any argument but {@code timestamp} is {@code null}.
+     */
+    protected AbstractPolicyActionEvent(final String type, final PolicyId policyId, final long revision,
+            @Nullable final Instant timestamp, final DittoHeaders dittoHeaders) {
+        super(type, policyId, revision, timestamp, dittoHeaders);
+    }
+
+    /**
+     * Aggregate a policy action event for subject creation or modification with similar events.
+     *
+     * @param initialModifiedSubjects subjects created or modified by this event indexed by their policy entry labels.
+     * @param otherEvents other subject creation or modification events.
+     * @return the aggregated event.
+     */
+    protected SubjectsModifiedPartially aggregateWithSubjectCreatedOrModified(
+            final Map<Label, Subject> initialModifiedSubjects,
+            final Collection<PolicyActionEvent<?>> otherEvents) {
+
+        final Map<Label, Subject> modifiedSubjects = new HashMap<>(initialModifiedSubjects);
+        for (final PolicyActionEvent<?> event : otherEvents) {
+            if (event instanceof SubjectCreated) {
+                final SubjectCreated subjectCreated = (SubjectCreated) event;
+                modifiedSubjects.put(subjectCreated.getLabel(), subjectCreated.getSubject());
+            } else if (event instanceof SubjectModified) {
+                final SubjectModified subjectModified = (SubjectModified) event;
+                modifiedSubjects.put(subjectModified.getLabel(), subjectModified.getSubject());
+            }
+        }
+        return SubjectsModifiedPartially.of(getPolicyEntityId(), modifiedSubjects, getRevision(),
+                getTimestamp().orElse(null), getDittoHeaders());
+    }
+
+    /**
+     * Aggregate a policy action event for subject deletion.
+     *
+     * @param initialDeletedSubjectIds Subject IDs deleted by this event indexed by their policy entry labels.
+     * @param otherEvents other subject deletion events.
+     * @return the aggregated event.
+     */
+    protected SubjectsDeletedPartially aggregateWithSubjectDeleted(final Map<Label, SubjectId> initialDeletedSubjectIds,
+            final Collection<PolicyActionEvent<?>> otherEvents) {
+
+        final Map<Label, SubjectId> deletedSubjectIds = new HashMap<>(initialDeletedSubjectIds);
+        for (final PolicyActionEvent<?> event : otherEvents) {
+            if (event instanceof SubjectDeleted) {
+                final SubjectDeleted subjectDeleted = (SubjectDeleted) event;
+                deletedSubjectIds.put(subjectDeleted.getLabel(), subjectDeleted.getSubjectId());
+            }
+        }
+        return SubjectsDeletedPartially.of(getPolicyEntityId(), deletedSubjectIds, getRevision(),
+                getTimestamp().orElse(null), getDittoHeaders());
+    }
+}
diff --git a/signals/events/policies/src/main/java/org/eclipse/ditto/signals/events/policies/PolicyActionEvent.java b/signals/events/policies/src/main/java/org/eclipse/ditto/signals/events/policies/PolicyActionEvent.java
new file mode 100755
index 0000000000..377105c4bb
--- /dev/null
+++ b/signals/events/policies/src/main/java/org/eclipse/ditto/signals/events/policies/PolicyActionEvent.java
@@ -0,0 +1,32 @@
+/*
+ * Copyright (c) 2021 Contributors to the Eclipse Foundation
+ *
+ * See the NOTICE file(s) distributed with this work for additional
+ * information regarding copyright ownership.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License 2.0 which is available at
+ * http://www.eclipse.org/legal/epl-2.0
+ *
+ * SPDX-License-Identifier: EPL-2.0
+ */
+package org.eclipse.ditto.signals.events.policies;
+
+import java.util.Collection;
+
+/**
+ * Interface for all policy-action-related events.
+ *
+ * @param <T> the type of the implementing class.
+ * @since 2.0.0
+ */
+public interface PolicyActionEvent<T extends PolicyActionEvent<T>> extends PolicyEvent<T> {
+
+    /**
+     * Create a policy event that has the same effect as a collection of policy action events together with this event.
+     *
+     * @param otherPolicyActionEvents the collection of policy action events.
+     * @return the aggregated event.
+     */
+    PolicyEvent<?> aggregateWith(Collection<PolicyActionEvent<?>> otherPolicyActionEvents);
+}
diff --git a/signals/events/policies/src/main/java/org/eclipse/ditto/signals/events/policies/SubjectCreated.java b/signals/events/policies/src/main/java/org/eclipse/ditto/signals/events/policies/SubjectCreated.java
index 0e45143ef7..08dfa1c5f9 100755
--- a/signals/events/policies/src/main/java/org/eclipse/ditto/signals/events/policies/SubjectCreated.java
+++ b/signals/events/policies/src/main/java/org/eclipse/ditto/signals/events/policies/SubjectCreated.java
@@ -15,9 +15,13 @@
 import static org.eclipse.ditto.model.base.common.ConditionChecker.checkNotNull;
 
 import java.time.Instant;
+import java.util.Collection;
+import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.function.Predicate;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 import javax.annotation.Nullable;
 import javax.annotation.concurrent.Immutable;
@@ -43,8 +47,8 @@
  * This event is emitted after a {@link Subject} was created.
  */
 @Immutable
-@JsonParsableEvent(name = SubjectCreated.NAME, typePrefix= SubjectCreated.TYPE_PREFIX)
-public final class SubjectCreated extends AbstractPolicyEvent<SubjectCreated> implements PolicyEvent<SubjectCreated> {
+@JsonParsableEvent(name = SubjectCreated.NAME, typePrefix = SubjectCreated.TYPE_PREFIX)
+public final class SubjectCreated extends AbstractPolicyActionEvent<SubjectCreated> {
 
     /**
      * Name of this event
@@ -197,15 +201,16 @@ public static SubjectCreated fromJson(final String jsonString, final DittoHeader
     public static SubjectCreated fromJson(final JsonObject jsonObject, final DittoHeaders dittoHeaders) {
         return new EventJsonDeserializer<SubjectCreated>(TYPE, jsonObject).deserialize(
                 (revision, timestamp, metadata) -> {
-            final String extractedPolicyId = jsonObject.getValueOrThrow(JsonFields.POLICY_ID);
-            final PolicyId policyId = PolicyId.of(extractedPolicyId);
-            final Label label = Label.of(jsonObject.getValueOrThrow(JSON_LABEL));
-            final String subjectId = jsonObject.getValueOrThrow(JSON_SUBJECT_ID);
-            final JsonObject subjectJsonObject = jsonObject.getValueOrThrow(JSON_SUBJECT);
-            final Subject extractedCreatedSubject = PoliciesModelFactory.newSubject(subjectId, subjectJsonObject);
-
-            return of(policyId, label, extractedCreatedSubject, revision, timestamp, dittoHeaders);
-        });
+                    final String extractedPolicyId = jsonObject.getValueOrThrow(JsonFields.POLICY_ID);
+                    final PolicyId policyId = PolicyId.of(extractedPolicyId);
+                    final Label label = Label.of(jsonObject.getValueOrThrow(JSON_LABEL));
+                    final String subjectId = jsonObject.getValueOrThrow(JSON_SUBJECT_ID);
+                    final JsonObject subjectJsonObject = jsonObject.getValueOrThrow(JSON_SUBJECT);
+                    final Subject extractedCreatedSubject =
+                            PoliciesModelFactory.newSubject(subjectId, subjectJsonObject);
+
+                    return of(policyId, label, extractedCreatedSubject, revision, timestamp, dittoHeaders);
+                });
     }
 
     /**
@@ -290,4 +295,10 @@ public String toString() {
         return getClass().getSimpleName() + " [" + super.toString() + ", label=" + label + ", subject=" + subject + "]";
     }
 
+    @Override
+    public SubjectsModifiedPartially aggregateWith(final Collection<PolicyActionEvent<?>> otherPolicyActionEvents) {
+        final Map<Label, Subject> initialCreatedSubjects =
+                Stream.of(0).collect(Collectors.toMap(i -> label, i -> subject));
+        return aggregateWithSubjectCreatedOrModified(initialCreatedSubjects, otherPolicyActionEvents);
+    }
 }
diff --git a/signals/events/policies/src/main/java/org/eclipse/ditto/signals/events/policies/SubjectDeleted.java b/signals/events/policies/src/main/java/org/eclipse/ditto/signals/events/policies/SubjectDeleted.java
index 80314b2660..8740f46b44 100755
--- a/signals/events/policies/src/main/java/org/eclipse/ditto/signals/events/policies/SubjectDeleted.java
+++ b/signals/events/policies/src/main/java/org/eclipse/ditto/signals/events/policies/SubjectDeleted.java
@@ -15,8 +15,12 @@
 import static org.eclipse.ditto.model.base.common.ConditionChecker.checkNotNull;
 
 import java.time.Instant;
+import java.util.Collection;
+import java.util.Map;
 import java.util.Objects;
 import java.util.function.Predicate;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 import javax.annotation.Nullable;
 import javax.annotation.concurrent.Immutable;
@@ -40,8 +44,8 @@
  * This event is emitted after a {@link org.eclipse.ditto.model.policies.Subject} was deleted.
  */
 @Immutable
-@JsonParsableEvent(name = SubjectDeleted.NAME, typePrefix= SubjectDeleted.TYPE_PREFIX)
-public final class SubjectDeleted extends AbstractPolicyEvent<SubjectDeleted> implements PolicyEvent<SubjectDeleted> {
+@JsonParsableEvent(name = SubjectDeleted.NAME, typePrefix = SubjectDeleted.TYPE_PREFIX)
+public final class SubjectDeleted extends AbstractPolicyActionEvent<SubjectDeleted> {
 
     /**
      * Name of this event.
@@ -191,14 +195,14 @@ public static SubjectDeleted fromJson(final String jsonString, final DittoHeader
     public static SubjectDeleted fromJson(final JsonObject jsonObject, final DittoHeaders dittoHeaders) {
         return new EventJsonDeserializer<SubjectDeleted>(TYPE, jsonObject).deserialize(
                 (revision, timestamp, metadata) -> {
-            final String extractedPolicyId = jsonObject.getValueOrThrow(JsonFields.POLICY_ID);
-            final PolicyId policyId = PolicyId.of(extractedPolicyId);
-            final Label label = Label.of(jsonObject.getValueOrThrow(JSON_LABEL));
-            final SubjectId extractedDeletedSubjectId =
-                    SubjectId.newInstance(jsonObject.getValueOrThrow(JSON_SUBJECT_ID));
-
-            return of(policyId, label, extractedDeletedSubjectId, revision, timestamp, dittoHeaders);
-        });
+                    final String extractedPolicyId = jsonObject.getValueOrThrow(JsonFields.POLICY_ID);
+                    final PolicyId policyId = PolicyId.of(extractedPolicyId);
+                    final Label label = Label.of(jsonObject.getValueOrThrow(JSON_LABEL));
+                    final SubjectId extractedDeletedSubjectId =
+                            SubjectId.newInstance(jsonObject.getValueOrThrow(JSON_SUBJECT_ID));
+
+                    return of(policyId, label, extractedDeletedSubjectId, revision, timestamp, dittoHeaders);
+                });
     }
 
     /**
@@ -278,4 +282,10 @@ public String toString() {
                 + "]";
     }
 
+    @Override
+    public SubjectsDeletedPartially aggregateWith(final Collection<PolicyActionEvent<?>> otherPolicyActionEvents) {
+        final Map<Label, SubjectId> initialDeletedSubjectId =
+                Stream.of(0).collect(Collectors.toMap(i -> label, i -> subjectId));
+        return aggregateWithSubjectDeleted(initialDeletedSubjectId, otherPolicyActionEvents);
+    }
 }
diff --git a/signals/events/policies/src/main/java/org/eclipse/ditto/signals/events/policies/SubjectModified.java b/signals/events/policies/src/main/java/org/eclipse/ditto/signals/events/policies/SubjectModified.java
index e5a9659003..30e7b3cbdf 100755
--- a/signals/events/policies/src/main/java/org/eclipse/ditto/signals/events/policies/SubjectModified.java
+++ b/signals/events/policies/src/main/java/org/eclipse/ditto/signals/events/policies/SubjectModified.java
@@ -15,9 +15,13 @@
 import static org.eclipse.ditto.model.base.common.ConditionChecker.checkNotNull;
 
 import java.time.Instant;
+import java.util.Collection;
+import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.function.Predicate;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 import javax.annotation.Nullable;
 import javax.annotation.concurrent.Immutable;
@@ -43,9 +47,8 @@
  * This event is emitted after a {@link Subject} was modified.
  */
 @Immutable
-@JsonParsableEvent(name = SubjectModified.NAME, typePrefix= SubjectModified.TYPE_PREFIX)
-public final class SubjectModified extends AbstractPolicyEvent<SubjectModified>
-        implements PolicyEvent<SubjectModified> {
+@JsonParsableEvent(name = SubjectModified.NAME, typePrefix = SubjectModified.TYPE_PREFIX)
+public final class SubjectModified extends AbstractPolicyActionEvent<SubjectModified> {
 
     /**
      * Name of this event
@@ -198,15 +201,16 @@ public static SubjectModified fromJson(final String jsonString, final DittoHeade
     public static SubjectModified fromJson(final JsonObject jsonObject, final DittoHeaders dittoHeaders) {
         return new EventJsonDeserializer<SubjectModified>(TYPE, jsonObject).deserialize(
                 (revision, timestamp, metadata) -> {
-            final String extractedPolicyId = jsonObject.getValueOrThrow(JsonFields.POLICY_ID);
-            final PolicyId policyId = PolicyId.of(extractedPolicyId);
-            final Label label = Label.of(jsonObject.getValueOrThrow(JSON_LABEL));
-            final String subjectId = jsonObject.getValueOrThrow(JSON_SUBJECT_ID);
-            final JsonObject subjectJsonObject = jsonObject.getValueOrThrow(JSON_SUBJECT);
-            final Subject extractedModifiedSubject = PoliciesModelFactory.newSubject(subjectId, subjectJsonObject);
-
-            return of(policyId, label, extractedModifiedSubject, revision, timestamp, dittoHeaders);
-        });
+                    final String extractedPolicyId = jsonObject.getValueOrThrow(JsonFields.POLICY_ID);
+                    final PolicyId policyId = PolicyId.of(extractedPolicyId);
+                    final Label label = Label.of(jsonObject.getValueOrThrow(JSON_LABEL));
+                    final String subjectId = jsonObject.getValueOrThrow(JSON_SUBJECT_ID);
+                    final JsonObject subjectJsonObject = jsonObject.getValueOrThrow(JSON_SUBJECT);
+                    final Subject extractedModifiedSubject =
+                            PoliciesModelFactory.newSubject(subjectId, subjectJsonObject);
+
+                    return of(policyId, label, extractedModifiedSubject, revision, timestamp, dittoHeaders);
+                });
     }
 
     /**
@@ -292,4 +296,10 @@ public String toString() {
         return getClass().getSimpleName() + " [" + super.toString() + ", label=" + label + ", subject=" + subject + "]";
     }
 
+    @Override
+    public SubjectsModifiedPartially aggregateWith(final Collection<PolicyActionEvent<?>> otherPolicyActionEvents) {
+        final Map<Label, Subject> initialModifiedSubjects =
+                Stream.of(0).collect(Collectors.toMap(i -> label, i -> subject));
+        return aggregateWithSubjectCreatedOrModified(initialModifiedSubjects, otherPolicyActionEvents);
+    }
 }
diff --git a/signals/events/policies/src/main/java/org/eclipse/ditto/signals/events/policies/SubjectsDeletedPartially.java b/signals/events/policies/src/main/java/org/eclipse/ditto/signals/events/policies/SubjectsDeletedPartially.java
index 3ee2ed20ab..22bdf6e274 100755
--- a/signals/events/policies/src/main/java/org/eclipse/ditto/signals/events/policies/SubjectsDeletedPartially.java
+++ b/signals/events/policies/src/main/java/org/eclipse/ditto/signals/events/policies/SubjectsDeletedPartially.java
@@ -15,6 +15,7 @@
 import static org.eclipse.ditto.model.base.common.ConditionChecker.checkNotNull;
 
 import java.time.Instant;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
@@ -51,8 +52,7 @@
  */
 @Immutable
 @JsonParsableEvent(name = SubjectsDeletedPartially.NAME, typePrefix = SubjectsDeletedPartially.TYPE_PREFIX)
-public final class SubjectsDeletedPartially extends AbstractPolicyEvent<SubjectsDeletedPartially>
-        implements PolicyEvent<SubjectsDeletedPartially> {
+public final class SubjectsDeletedPartially extends AbstractPolicyActionEvent<SubjectsDeletedPartially> {
 
     /**
      * Name of this event.
@@ -183,7 +183,8 @@ public SubjectsDeletedPartially setRevision(final long revision) {
 
     @Override
     public SubjectsDeletedPartially setDittoHeaders(final DittoHeaders dittoHeaders) {
-        return new SubjectsDeletedPartially(getEntityId(), deletedSubjectIds, getRevision(), getTimestamp().orElse(null),
+        return new SubjectsDeletedPartially(getEntityId(), deletedSubjectIds, getRevision(),
+                getTimestamp().orElse(null),
                 dittoHeaders);
     }
 
@@ -238,4 +239,8 @@ private static Map<Label, SubjectId> deletedSubjectsFromJson(final JsonObject js
         return Collections.unmodifiableMap(map);
     }
 
+    @Override
+    public SubjectsDeletedPartially aggregateWith(final Collection<PolicyActionEvent<?>> otherPolicyActionEvents) {
+        return aggregateWithSubjectDeleted(deletedSubjectIds, otherPolicyActionEvents);
+    }
 }
diff --git a/signals/events/policies/src/main/java/org/eclipse/ditto/signals/events/policies/SubjectsModifiedPartially.java b/signals/events/policies/src/main/java/org/eclipse/ditto/signals/events/policies/SubjectsModifiedPartially.java
index 062fb273a6..795cd8cab2 100755
--- a/signals/events/policies/src/main/java/org/eclipse/ditto/signals/events/policies/SubjectsModifiedPartially.java
+++ b/signals/events/policies/src/main/java/org/eclipse/ditto/signals/events/policies/SubjectsModifiedPartially.java
@@ -15,6 +15,7 @@
 import static org.eclipse.ditto.model.base.common.ConditionChecker.checkNotNull;
 
 import java.time.Instant;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
@@ -53,8 +54,7 @@
  */
 @Immutable
 @JsonParsableEvent(name = SubjectsModifiedPartially.NAME, typePrefix = SubjectsModifiedPartially.TYPE_PREFIX)
-public final class SubjectsModifiedPartially extends AbstractPolicyEvent<SubjectsModifiedPartially>
-        implements PolicyEvent<SubjectsModifiedPartially> {
+public final class SubjectsModifiedPartially extends AbstractPolicyActionEvent<SubjectsModifiedPartially> {
 
     /**
      * Name of this event
@@ -255,4 +255,8 @@ private static Map<Label, Subject> modifiedSubjectsFromJson(final JsonObject jso
         return Collections.unmodifiableMap(map);
     }
 
+    @Override
+    public SubjectsModifiedPartially aggregateWith(final Collection<PolicyActionEvent<?>> otherPolicyActionEvents) {
+        return aggregateWithSubjectCreatedOrModified(modifiedSubjects, otherPolicyActionEvents);
+    }
 }