feat: Transform event and metadata of replicated event (#32650)

* transform incoming replicated event and metadata before it is stored
* test and hardening of metadata

Author: patriknw

akka-persistence-typed-tests/src/test/scala/akka/persistence/typed/ReplicatedEventPublishingSpec.scala

@@ -5,6 +5,7 @@
 package akka.persistence.typed
 
 import org.scalatest.wordspec.AnyWordSpecLike
+
 import akka.Done
 import akka.actor.testkit.typed.TestException
 import akka.actor.testkit.typed.scaladsl.LogCapturing
@@ -18,11 +19,12 @@ import akka.persistence.typed.internal.{ ReplicatedPublishedEventMetaData, Versi
 import akka.persistence.typed.scaladsl.Effect
 import akka.persistence.typed.scaladsl.EventSourcedBehavior
 import akka.persistence.typed.scaladsl.ReplicatedEventSourcing
-
 import scala.concurrent.ExecutionContext
 import scala.concurrent.Future
 import scala.concurrent.duration.DurationInt
 
+import akka.persistence.typed.scaladsl.EventWithMetadata
+
 object ReplicatedEventPublishingSpec {
 
   val EntityType = "EventPublishingSpec"
@@ -433,6 +435,37 @@ class ReplicatedEventPublishingSpec
       probe.expectTerminated(actor)
     }
 
+    "transform replicated events between two entities" in {
+      val id = nextEntityId()
+      val probe = createTestProbe[Any]()
+      case class Intercepted(origin: ReplicaId, seqNr: Long, event: String)
+      val addTransformation
+          : EventSourcedBehavior[MyReplicatedBehavior.Command, String, Set[String]] => EventSourcedBehavior[
+            MyReplicatedBehavior.Command,
+            String,
+            Set[String]] =
+        _.withReplicatedEventTransformation { (_, eventWithMeta) =>
+          EventWithMetadata(eventWithMeta.event.toUpperCase, Nil)
+        }
+      val actor = spawn(MyReplicatedBehavior(id, DCA, Set(DCA, DCB), modifyBehavior = addTransformation))
+      actor ! MyReplicatedBehavior.Add("one", probe.ref)
+      probe.expectMessage(Done)
+
+      // simulate a published event from another replica
+      actor.asInstanceOf[ActorRef[Any]] ! internal.PublishedEventImpl(
+        ReplicationId(EntityType, id, DCB).persistenceId,
+        1L,
+        "two",
+        System.currentTimeMillis(),
+        Some(new ReplicatedPublishedEventMetaData(DCB, VersionVector.empty, None)),
+        None)
+      actor ! MyReplicatedBehavior.Add("three", probe.ref)
+      probe.expectMessage(Done)
+
+      actor ! MyReplicatedBehavior.Get(probe.ref)
+      probe.expectMessage(Set("one", "TWO", "three"))
+    }
+
   }
 
 }

akka-persistence-typed-tests/src/test/scala/akka/persistence/typed/ReplicatedEventSourcingSpec.scala

@@ -7,8 +7,10 @@ package akka.persistence.typed
 import java.util.concurrent.CountDownLatch
 import java.util.concurrent.TimeUnit
 import java.util.concurrent.atomic.AtomicInteger
+
 import org.scalatest.concurrent.Eventually
 import org.scalatest.wordspec.AnyWordSpecLike
+
 import akka.Done
 import akka.actor.testkit.typed.TestException
 import akka.actor.testkit.typed.scaladsl.LogCapturing
@@ -20,35 +22,53 @@ import akka.persistence.testkit.query.scaladsl.PersistenceTestKitReadJournal
 import akka.persistence.testkit.scaladsl.PersistenceTestKit
 import akka.persistence.typed.scaladsl.{ Effect, EventSourcedBehavior, ReplicatedEventSourcing, ReplicationContext }
 import akka.serialization.jackson.CborSerializable
-
 import scala.concurrent.ExecutionContext
 import scala.concurrent.Future
 import scala.concurrent.duration.DurationInt
 
+import akka.actor.typed.scaladsl.ActorContext
+import akka.actor.typed.scaladsl.Behaviors
+import akka.persistence.typed.internal.ReplicatedEventMetadata
+import akka.persistence.typed.scaladsl.EventWithMetadata
+
 object ReplicatedEventSourcingSpec {
 
   val AllReplicas = Set(ReplicaId("R1"), ReplicaId("R2"), ReplicaId("R3"))
 
   sealed trait Command
-  case class GetState(replyTo: ActorRef[State]) extends Command
-  case class StoreMe(description: String, replyTo: ActorRef[Done], latch: CountDownLatch = new CountDownLatch(1))
+  final case class GetState(replyTo: ActorRef[State]) extends Command
+  final case class StoreMe(description: String, replyTo: ActorRef[Done], latch: CountDownLatch = new CountDownLatch(1))
       extends Command
-  case class StoreUs(descriptions: List[String], replyTo: ActorRef[Done], latch: CountDownLatch = new CountDownLatch(1))
+  final case class StoreUs(
+      descriptions: List[String],
+      replyTo: ActorRef[Done],
+      latch: CountDownLatch = new CountDownLatch(1))
       extends Command
-  case class GetReplica(replyTo: ActorRef[(ReplicaId, Set[ReplicaId])]) extends Command
-  case object Stop extends Command
+  final case class StoreMeWithMeta(description: String, replyTo: ActorRef[Done], meta: Meta) extends Command
+  final case class GetReplica(replyTo: ActorRef[(ReplicaId, Set[ReplicaId])]) extends Command
+  final case object Stop extends Command
 
-  case class State(all: List[String]) extends CborSerializable
+  final case class State(all: Vector[String]) extends CborSerializable
+
+  final case class Meta(value: String) extends CborSerializable
+
+  final case class EventAndContext(
+      event: Any,
+      origin: ReplicaId,
+      recoveryRunning: Boolean,
+      concurrent: Boolean,
+      meta: Option[Meta] = None)
 
   def testBehavior(entityId: String, replicaId: String, probe: ActorRef[EventAndContext]): Behavior[Command] =
     testBehavior(entityId, replicaId, Some(probe))
 
   def eventSourcedBehavior(
+      context: ActorContext[Command],
       replicationContext: ReplicationContext,
       probe: Option[ActorRef[EventAndContext]]): EventSourcedBehavior[Command, String, State] = {
     EventSourcedBehavior[Command, String, State](
       replicationContext.persistenceId,
-      State(Nil),
+      State(Vector.empty),
       (state, command) =>
         command match {
           case GetState(replyTo) =>
@@ -65,17 +85,21 @@ object ReplicatedEventSourcingSpec {
             latch.countDown()
             latch.await(10, TimeUnit.SECONDS)
             Effect.persist(evts).thenRun(_ => replyTo ! Done)
+          case StoreMeWithMeta(evt, ack, meta) =>
+            Effect.persistWithMetadata(EventWithMetadata(evt, meta)).thenRun(_ => ack ! Done)
           case Stop =>
             Effect.stop()
         },
       (state, event) => {
+        val meta = EventSourcedBehavior.currentMetadata[Meta](context)
         probe.foreach(
           _ ! EventAndContext(
             event,
             replicationContext.origin,
             replicationContext.recoveryRunning,
-            replicationContext.concurrent))
-        state.copy(all = event :: state.all)
+            replicationContext.concurrent,
+            meta))
+        state.copy(all = state.all :+ event)
       })
   }
 
@@ -85,19 +109,35 @@ object ReplicatedEventSourcingSpec {
       probe: Option[ActorRef[EventAndContext]] = None,
       allReplicas: Set[ReplicaId] = AllReplicas,
       modifyBehavior: EventSourcedBehavior[Command, String, State] => EventSourcedBehavior[Command, String, State] =
-        identity): Behavior[Command] =
-    ReplicatedEventSourcing.commonJournalConfig(
-      ReplicationId("ReplicatedEventSourcingSpec", entityId, ReplicaId(replicaId)),
-      allReplicas,
-      PersistenceTestKitReadJournal.Identifier)(replicationContext =>
-      modifyBehavior(eventSourcedBehavior(replicationContext, probe)))
+        identity): Behavior[Command] = {
+    testBehaviorWithContext(entityId, replicaId, probe, allReplicas, (b, _) => modifyBehavior(b))
+  }
+
+  def testBehaviorWithContext(
+      entityId: String,
+      replicaId: String,
+      probe: Option[ActorRef[EventAndContext]] = None,
+      allReplicas: Set[ReplicaId] = AllReplicas,
+      modifyBehavior: (
+          EventSourcedBehavior[Command, String, State],
+          ActorContext[Command]) => EventSourcedBehavior[Command, String, State]): Behavior[Command] = {
+    Behaviors.setup[Command] { context =>
+      ReplicatedEventSourcing.commonJournalConfig(
+        ReplicationId("ReplicatedEventSourcingSpec", entityId, ReplicaId(replicaId)),
+        allReplicas,
+        PersistenceTestKitReadJournal.Identifier) { replicationContext =>
+        val behv = eventSourcedBehavior(context, replicationContext, probe)
+        modifyBehavior(behv, context)
+      }
+    }
+  }
 
   def nonReplicatedEventSourcedBehavior(
       persistenceId: PersistenceId,
       probe: Option[ActorRef[String]]): EventSourcedBehavior[Command, String, State] = {
     EventSourcedBehavior[Command, String, State](
       persistenceId,
-      State(Nil),
+      State(Vector.empty),
       (state, command) =>
         command match {
           case GetState(replyTo) =>
@@ -113,12 +153,14 @@ object ReplicatedEventSourcingSpec {
             latch.countDown()
             latch.await(10, TimeUnit.SECONDS)
             Effect.persist(evts).thenRun(_ => replyTo ! Done)
+          case StoreMeWithMeta(evt, ack, meta) =>
+            Effect.persistWithMetadata(EventWithMetadata(evt, meta)).thenRun(_ => ack ! Done)
           case Stop =>
             Effect.stop()
         },
       (state, event) => {
         probe.foreach(_ ! event)
-        state.copy(all = event :: state.all)
+        state.copy(all = state.all :+ event)
       })
   }
 
@@ -127,8 +169,6 @@ object ReplicatedEventSourcingSpec {
 
 }
 
-case class EventAndContext(event: Any, origin: ReplicaId, recoveryRunning: Boolean, concurrent: Boolean)
-
 class ReplicatedEventSourcingSpec
     extends ScalaTestWithActorTestKit(PersistenceTestKitPlugin.config)
     with AnyWordSpecLike
@@ -490,9 +530,9 @@ class ReplicatedEventSourcingSpec
 
       // ensure recovery is complete
       r1 ! GetState(stateProbe.ref)
-      stateProbe.expectMessage(State(Nil))
+      stateProbe.expectMessage(State(Vector.empty))
       r2 ! GetState(stateProbe.ref)
-      stateProbe.expectMessage(State(Nil))
+      stateProbe.expectMessage(State(Vector.empty))
 
       // make reads fail for the replication
       testkit.failNextNReads(s"$entityId|R2", 1)
@@ -517,21 +557,21 @@ class ReplicatedEventSourcingSpec
       val r1 = spawn(testBehavior(entityId, "", allReplicas = allReplicas))
       val stateProbe = createTestProbe[State]()
       r1 ! GetState(stateProbe.ref)
-      stateProbe.expectMessageType[State].all.reverse shouldEqual List("from es1")
+      stateProbe.expectMessageType[State].all shouldEqual List("from es1")
 
       r1 ! StoreMe("from r1", probe.ref)
       probe.expectMessage(Done)
       val r2 = spawn(testBehavior(entityId, "R2", allReplicas = allReplicas))
       eventually {
         r2 ! GetState(stateProbe.ref)
-        stateProbe.expectMessageType[State].all.reverse shouldEqual List("from es1", "from r1")
+        stateProbe.expectMessageType[State].all shouldEqual List("from es1", "from r1")
       }
 
       r2 ! StoreMe("from r2", probe.ref)
       probe.expectMessage(Done)
       eventually {
         r1 ! GetState(stateProbe.ref)
-        stateProbe.expectMessageType[State].all.reverse shouldEqual List("from es1", "from r1", "from r2")
+        stateProbe.expectMessageType[State].all shouldEqual List("from es1", "from r1", "from r2")
       }
     }
 
@@ -603,5 +643,70 @@ class ReplicatedEventSourcingSpec
       r2 ! StoreMe("from r2", probe.ref)
       probe.expectTerminated(r1)
     }
+
+    "transform replicated events between two entities" in {
+      val entityId = nextEntityId
+      val probe = createTestProbe[Done]()
+      val eventProbe1 = createTestProbe[EventAndContext]()
+      val eventProbe2 = createTestProbe[EventAndContext]()
+      val addTransformation: (
+          EventSourcedBehavior[Command, String, State],
+          ActorContext[Command]) => EventSourcedBehavior[Command, String, State] = { (behv, context) =>
+        behv.withReplicatedEventTransformation { (_, eventWithMeta) =>
+          val resMeta1 = EventSourcedBehavior.currentMetadata[ReplicatedEventMetadata](context)
+          val resMeta2 = eventWithMeta.metadata[ReplicatedEventMetadata]
+          if (resMeta1 != resMeta2)
+            throw new IllegalStateException(s"Expected RES metadata to be the same, $resMeta1 != $resMeta2")
+
+          val newMeta = eventWithMeta.metadata[Meta].map(m => m.copy(m.value.toUpperCase)).toList
+          EventWithMetadata(eventWithMeta.event.toUpperCase, newMeta)
+        }
+      }
+      val r1 = spawn(
+        testBehaviorWithContext(entityId, "R1", probe = Some(eventProbe1.ref), modifyBehavior = addTransformation))
+      val r2 = spawn(
+        testBehaviorWithContext(entityId, "R2", probe = Some(eventProbe2.ref), modifyBehavior = addTransformation))
+
+      r1 ! StoreMeWithMeta("from r1", probe.ref, Meta("meta from r1"))
+      eventProbe1.expectMessage(
+        EventAndContext(
+          "from r1",
+          ReplicaId("R1"),
+          recoveryRunning = false,
+          concurrent = false,
+          Some(Meta("meta from r1"))))
+      // replicated to r2, and transformed
+      eventProbe2.expectMessage(
+        EventAndContext(
+          "FROM R1",
+          ReplicaId("R1"),
+          recoveryRunning = false,
+          concurrent = false,
+          meta = Some(Meta("META FROM R1"))))
+
+      r2 ! StoreMeWithMeta("from r2", probe.ref, Meta("meta from r2"))
+      eventProbe2.expectMessage(
+        EventAndContext(
+          "from r2",
+          ReplicaId("R2"),
+          recoveryRunning = false,
+          concurrent = false,
+          Some(Meta("meta from r2"))))
+      // replicated to r1, and transformed
+      eventProbe1.expectMessage(
+        EventAndContext(
+          "FROM R2",
+          ReplicaId("R2"),
+          recoveryRunning = false,
+          concurrent = false,
+          meta = Some(Meta("META FROM R2"))))
+
+      val stateProbe = createTestProbe[State]()
+      r1 ! GetState(stateProbe.ref)
+      stateProbe.expectMessage(State(Vector("from r1", "FROM R2")))
+
+      r2 ! GetState(stateProbe.ref)
+      stateProbe.expectMessage(State(Vector("FROM R1", "from r2")))
+    }
   }
 }

akka-persistence-typed-tests/src/test/scala/akka/persistence/typed/ReplicationSnapshotSpec.scala

@@ -10,6 +10,7 @@ import org.scalatest.wordspec.AnyWordSpecLike
 
 import akka.Done
 import akka.actor.testkit.typed.scaladsl.{ LogCapturing, ScalaTestWithActorTestKit }
+import akka.actor.typed.scaladsl.Behaviors
 import akka.actor.typed.{ ActorRef, Behavior }
 import akka.persistence.testkit.{ PersistenceTestKitPlugin, PersistenceTestKitSnapshotPlugin }
 import akka.persistence.testkit.query.scaladsl.PersistenceTestKitReadJournal
@@ -36,11 +37,15 @@ object ReplicationSnapshotSpec {
       entityId: String,
       replicaId: ReplicaId,
       probe: Option[ActorRef[EventAndContext]]): Behavior[Command] = {
-    ReplicatedEventSourcing.commonJournalConfig(
-      ReplicationId(EntityType, entityId, replicaId),
-      AllReplicas,
-      PersistenceTestKitReadJournal.Identifier)(replicationContext =>
-      eventSourcedBehavior(replicationContext, probe).snapshotWhen((_, _, sequenceNr) => sequenceNr % 2 == 0))
+    Behaviors.setup[Command] { context =>
+      ReplicatedEventSourcing.commonJournalConfig(
+        ReplicationId(EntityType, entityId, replicaId),
+        AllReplicas,
+        PersistenceTestKitReadJournal.Identifier)(
+        replicationContext =>
+          eventSourcedBehavior(context, replicationContext, probe).snapshotWhen((_, _, sequenceNr) =>
+            sequenceNr % 2 == 0))
+    }
 
   }
 }
@@ -79,8 +84,8 @@ class ReplicationSnapshotSpec
         r2EventProbe.expectMessageType[EventAndContext]
         r2EventProbe.expectMessageType[EventAndContext]
 
-        snapshotTestKit.expectNextPersisted(persistenceIdR1, State(List("r1 2", "r1 1")))
-        snapshotTestKit.expectNextPersisted(persistenceIdR2, State(List("r1 2", "r1 1")))
+        snapshotTestKit.expectNextPersisted(persistenceIdR1, State(Vector("r1 1", "r1 2")))
+        snapshotTestKit.expectNextPersisted(persistenceIdR2, State(Vector("r1 1", "r1 2")))
 
         r2.asInstanceOf[ActorRef[Any]] ! internal.PublishedEventImpl(
           ReplicationId(EntityType, entityId, R1).persistenceId,
@@ -108,7 +113,7 @@ class ReplicationSnapshotSpec
 
         val stateProbe = createTestProbe[State]()
         r2 ! GetState(stateProbe.ref)
-        stateProbe.expectMessage(State(List("r1 2", "r1 1")))
+        stateProbe.expectMessage(State(Vector("r1 1", "r1 2")))
       }
     }
   }

akka-persistence-typed-tests/src/test/scala/akka/persistence/typed/scaladsl/EventSourcedBehaviorWatchSpec.scala

@@ -78,6 +78,7 @@ class EventSourcedBehaviorWatchSpec
       internalLoggerFactory = () => logger,
       retentionInProgress = false,
       EmptyEventSourcedBehaviorInstrumentation,
+      None,
       None)
 
   "A typed persistent parent actor watching a child" must {

akka-persistence-typed/src/main/mima-filters/2.10.1.backwards.excludes/32650-res-transform.excludes

@@ -0,0 +1,2 @@
+ProblemFilters.exclude[ReversedMissingMethodProblem]("akka.persistence.typed.scaladsl.EventSourcedBehavior.withReplicatedEventTransformation")
+ProblemFilters.exclude[Problem]("akka.persistence.typed.internal.JournalInteractions#EventToPersist.*")