Issue-111: add event and snapshot adapters (#116)

Converts lagom-pb wrapper protos to COS protos at rest (as part of decoupling us from lagom-pb)

Author: Zen Yui

code/service/src/main/scala/com/namely/chiefofstate/Aggregate.scala

@@ -5,6 +5,11 @@ import com.google.protobuf.empty.Empty
 import com.typesafe.config.Config
 import io.superflat.lagompb.encryption.EncryptionAdapter
 import io.superflat.lagompb.{AggregateRoot, CommandHandler, EventHandler}
+import akka.persistence.typed.PersistenceId
+import akka.persistence.typed.scaladsl.EventSourcedBehavior
+import io.superflat.lagompb.Command
+import io.superflat.lagompb.protobuf.v1.core.{EventWrapper, StateWrapper}
+import com.namely.chiefofstate.persistence.{CosEventAdapter, CosSnapshotAdapter}
 
 /**
  * ChiefOfStateAggregate
@@ -24,4 +29,17 @@ class Aggregate(
 
   override def aggregateName: String = "chiefOfState"
 
+  /**
+   * generate the lagom-pb event-sourced behavior and inject event and
+   * snapshot adapters
+   *
+   * @param persistenceId aggregate persistence ID
+   * @return EventSourcedBehavior
+   */
+  override def create(persistenceId: PersistenceId): EventSourcedBehavior[Command, EventWrapper, StateWrapper] = {
+    super
+      .create(persistenceId)
+      .eventAdapter(CosEventAdapter)
+      .snapshotAdapter(CosSnapshotAdapter)
+  }
 }

code/service/src/main/scala/com/namely/chiefofstate/Util.scala

@@ -22,14 +22,28 @@ object Util {
   /**
    * Converts the lagom-pb MetaData class to the chief-of-state MetaData
    *
-   * @param metaData
-   * @return
+   * @param metaData lagom-pb MetaData
+   * @return chief-of-state MetaData instance
    */
   def toCosMetaData(metaData: MetaData): CosMetaData = {
     CosMetaData(
       entityId = metaData.entityId,
-      // TODO: remove .toInt
-      revisionNumber = metaData.revisionNumber.toInt,
+      revisionNumber = metaData.revisionNumber,
+      revisionDate = metaData.revisionDate,
+      data = metaData.data
+    )
+  }
+
+  /**
+   * Converts chief-of-state MetaData to lagom-pb MetaData
+   *
+   * @param metaData COS meta data
+   * @return Lagom-pb MetaData instance
+   */
+  def toLagompbMetaData(metaData: CosMetaData): MetaData = {
+    MetaData(
+      entityId = metaData.entityId,
+      revisionNumber = metaData.revisionNumber,
       revisionDate = metaData.revisionDate,
       data = metaData.data
     )

code/service/src/main/scala/com/namely/chiefofstate/persistence/CosEventAdapter.scala

@@ -0,0 +1,50 @@
+package com.namely.chiefofstate.persistence
+
+import akka.persistence.typed.{EventAdapter, EventSeq}
+import io.superflat.lagompb.protobuf.v1.core.{EventWrapper => LagompbEventWrapper}
+import com.namely.protobuf.chiefofstate.v1.persistence.EventWrapper
+import com.namely.chiefofstate.Util
+
+/**
+ * Akka persistence event adaptor that converts lagom-pb event wrappers
+ * to COS wrappers in preparation of dropping the lagom-pb dependency
+ */
+object CosEventAdapter extends EventAdapter[LagompbEventWrapper, EventWrapper] {
+
+  /**
+   * convert lagom-pb EventWrapper to a cos EventWrapper
+   *
+   * @param e lagom-pb EventWrapper
+   * @return cos EventWrapper instance
+   */
+  def toJournal(e: LagompbEventWrapper): EventWrapper = {
+    EventWrapper(
+      event = e.event,
+      resultingState = e.resultingState,
+      meta = e.meta.map(Util.toCosMetaData)
+    )
+  }
+
+  /**
+   * convert cos EventWrapper to a lagom-pb EventWrapper
+   *
+   * @param p cos EventWrapper
+   * @param manifest the manifest used
+   * @return lagom-pb EventWrapper instance
+   */
+  def fromJournal(p: EventWrapper, manifest: String): EventSeq[LagompbEventWrapper] = {
+    EventSeq(
+      Seq(
+        LagompbEventWrapper(
+          event = p.event,
+          resultingState = p.resultingState,
+          meta = p.meta.map(Util.toLagompbMetaData)
+        )
+      )
+    )
+  }
+
+  val MANIFEST: String = "com.namely.chiefofstate.persistence.CosEventAdapter"
+
+  def manifest(event: LagompbEventWrapper): String = MANIFEST
+}

code/service/src/main/scala/com/namely/chiefofstate/persistence/CosSnapshotAdapter.scala

@@ -0,0 +1,47 @@
+package com.namely.chiefofstate.persistence
+
+import io.superflat.lagompb.protobuf.v1.core.{StateWrapper => LagompbStateWrapper}
+import com.namely.protobuf.chiefofstate.v1.persistence.StateWrapper
+import akka.persistence.typed.SnapshotAdapter
+import com.google.protobuf.any.Any
+import com.namely.chiefofstate.Util
+
+/**
+ * akka persistence SnapshotAdapter for converting to/from lagom-pb
+ * state wrappers in anticipation of dropping that dependency
+ */
+object CosSnapshotAdapter extends SnapshotAdapter[LagompbStateWrapper] {
+
+  /**
+   * convert lagom-pb state wrapper to a COS wrapper
+   *
+   * @param state lagom-pb state wrapper
+   * @return COS state wrapper as a scala.Any
+   */
+  def toJournal(state: LagompbStateWrapper): scala.Any = {
+    StateWrapper(
+      state = state.state,
+      meta = state.meta.map(Util.toCosMetaData)
+    )
+  }
+
+  /**
+   * convert COS state wrapper to a lagom-pb state wrapper
+   *
+   * @param from COS state wrapper as a scala.Any
+   * @return lagom-pb StateWrapper
+   */
+  def fromJournal(from: scala.Any): LagompbStateWrapper = {
+    from match {
+      case state: StateWrapper =>
+        LagompbStateWrapper(
+          state = state.state,
+          meta = state.meta.map(Util.toLagompbMetaData)
+        )
+
+      case x =>
+        throw new Exception(s"snapshot adapter cannot unpack state of type ${from.getClass.getName}")
+    }
+  }
+
+}

code/service/src/test/scala/com/namely/chiefofstate/AggregateSpec.scala

@@ -0,0 +1,24 @@
+package com.namely.chiefofstate
+
+import org.scalamock.scalatest.MockFactory
+import scala.util.Try
+import io.superflat.lagompb.encryption.EncryptionAdapter
+import akka.persistence.typed.PersistenceId
+import com.namely.chiefofstate.test.helpers.TestSpec
+import io.superflat.lagompb.{CommandHandler, EventHandler}
+
+class AggregateSpec extends TestSpec with MockFactory {
+  ".create" should {
+    "return EventSourcedBehavior with adapters" in {
+      val cmdHandler: CommandHandler = mock[CommandHandler]
+      val eventHandler: EventHandler = mock[EventHandler]
+      val encryptionAdapter: EncryptionAdapter = mock[EncryptionAdapter]
+      val agg = new Aggregate(null, null, cmdHandler, eventHandler, encryptionAdapter)
+      val persistenceId: PersistenceId = PersistenceId("typeHint", "entityId")
+      // TODO: find a real way to test this
+      // unfortunately, Akka made the implementation case class private,
+      // so there is no way to observe the eventAdatper and snapshotAdapter
+      Try(agg.create(persistenceId)).isSuccess shouldBe (true)
+    }
+  }
+}

code/service/src/test/scala/com/namely/chiefofstate/UtilSpec.scala

@@ -30,4 +30,26 @@ class UtilSpec extends TestSpec {
       actual shouldBe (expected)
     }
   }
+
+  "toLagompbMetaData" should {
+    "return the right lagom-pb MetaData" in {
+      val ts = Timestamp().withSeconds(3L).withNanos(2)
+      val revisionNumber = 2
+      val data = Map("foo" -> Any.pack(Empty.defaultInstance))
+
+      val lagomMetaData = LagompbMetaData()
+        .withRevisionNumber(revisionNumber)
+        .withRevisionDate(ts)
+        .withData(data)
+
+      val cosMetaData = CosMetaData()
+        .withRevisionNumber(revisionNumber)
+        .withRevisionDate(ts)
+        .withData(data)
+
+      val actual = Util.toLagompbMetaData(cosMetaData)
+
+      actual shouldBe (lagomMetaData)
+    }
+  }
 }

code/service/src/test/scala/com/namely/chiefofstate/persistence/CosEventAdapterSpec.scala

@@ -0,0 +1,61 @@
+package com.namely.chiefofstate.persistence
+
+import com.google.protobuf.any.Any
+import com.google.protobuf.wrappers.StringValue
+import com.namely.chiefofstate.test.helpers.TestSpec
+import com.namely.protobuf.chiefofstate.v1.common.{MetaData => CosMetaData}
+import com.namely.protobuf.chiefofstate.v1.persistence.EventWrapper
+import io.superflat.lagompb.protobuf.v1.core.{MetaData => LagompbMetaData}
+import io.superflat.lagompb.protobuf.v1.core.{EventWrapper => LagompbEventWrapper}
+
+class CosEventAdapterSpec extends TestSpec {
+  ".toJournal" should {
+    "return a cos event wrapper" in {
+      val event = Any.pack(StringValue("event"))
+      val state = Any.pack(StringValue("state"))
+      val revision = 2
+
+      val before = LagompbEventWrapper()
+        .withEvent(event)
+        .withResultingState(state)
+        .withMeta(LagompbMetaData().withRevisionNumber(revision))
+
+      val expected = EventWrapper()
+        .withEvent(event)
+        .withResultingState(state)
+        .withMeta(CosMetaData().withRevisionNumber(revision))
+
+      val actual = CosEventAdapter.toJournal(before)
+
+      actual shouldBe (expected)
+    }
+  }
+
+  ".fromJournal" should {
+    "return a lagom-pb event wrapper" in {
+      val event = Any.pack(StringValue("event"))
+      val state = Any.pack(StringValue("state"))
+      val revision = 2
+
+      val expected = LagompbEventWrapper()
+        .withEvent(event)
+        .withResultingState(state)
+        .withMeta(LagompbMetaData().withRevisionNumber(revision))
+
+      val before = EventWrapper()
+        .withEvent(event)
+        .withResultingState(state)
+        .withMeta(CosMetaData().withRevisionNumber(revision))
+
+      val actual = CosEventAdapter.fromJournal(before, "")
+      actual.events.length shouldBe (1)
+      actual.events.head shouldBe (expected)
+    }
+  }
+
+  ".manifest" should {
+    "yield the stable string" in {
+      CosEventAdapter.manifest(null) shouldBe (CosEventAdapter.MANIFEST)
+    }
+  }
+}

code/service/src/test/scala/com/namely/chiefofstate/persistence/CosSnapshotAdapterSpec.scala

@@ -0,0 +1,62 @@
+package com.namely.chiefofstate.persistence
+
+import com.google.protobuf.any.Any
+import com.google.protobuf.wrappers.StringValue
+import com.google.protobuf.timestamp.Timestamp
+import com.namely.chiefofstate.test.helpers.TestSpec
+import com.namely.chiefofstate.Util
+import com.namely.protobuf.chiefofstate.v1.common.{MetaData => CosMetaData}
+import com.namely.protobuf.chiefofstate.v1.persistence.StateWrapper
+import io.superflat.lagompb.protobuf.v1.core.{MetaData => LagompbMetaData}
+import io.superflat.lagompb.protobuf.v1.core.{StateWrapper => LagompbStateWrapper}
+
+class CosSnapshotAdapterSpec extends TestSpec {
+  ".toJournal" should {
+    "return a cos state wrapper" in {
+      val state = Any.pack(StringValue("state"))
+      val revision: Int = 2
+
+      val before = LagompbStateWrapper()
+        .withState(state)
+        .withMeta(LagompbMetaData().withRevisionNumber(revision))
+
+      val expected = StateWrapper()
+        .withState(state)
+        .withMeta(Util.toCosMetaData(before.getMeta))
+
+      val actual: StateWrapper = CosSnapshotAdapter
+        .toJournal(before)
+        .asInstanceOf[StateWrapper]
+
+      actual shouldBe (expected)
+    }
+  }
+
+  ".fromJournal" should {
+    "return a lagom-pb state wrapper" in {
+      val state = Any.pack(StringValue("state"))
+      val revision: Int = 2
+
+      val before = StateWrapper()
+        .withState(state)
+        .withMeta(CosMetaData().withRevisionNumber(revision))
+
+      val expected = LagompbStateWrapper()
+        .withState(state)
+        .withMeta(Util.toLagompbMetaData(before.getMeta))
+
+      val actual: LagompbStateWrapper = CosSnapshotAdapter
+        .fromJournal(before)
+        .asInstanceOf[LagompbStateWrapper]
+
+      actual shouldBe (expected)
+    }
+    "fail on unknown snapshot type" in {
+      val failure = intercept[Exception] {
+        CosSnapshotAdapter.fromJournal(StringValue("bad state"))
+      }
+
+      failure.getMessage().startsWith("snapshot adapter cannot unpack state of type") shouldBe (true)
+    }
+  }
+}

project/Dependencies.scala

@@ -5,7 +5,7 @@ object Dependencies {
   // Package versions
   object Versions {
     val Scala213: String = "2.13.1"
-    val LagomPbVersion: String = "1.0.2"
+    val LagomPbVersion: String = "1.0.2+4-ccd75b98-SNAPSHOT"
     val KanelaAgentVersion: String = "1.0.6"
     val SilencerVersion: String = "1.6.0"
     val KamonAkkaGrpcVersion: String = "0.0.9"

proto/internal/persistence.proto

@@ -0,0 +1,33 @@
+syntax = "proto3";
+
+package chief_of_state.v1;
+
+option java_package = "com.namely.protobuf.chiefofstate.v1";
+option java_multiple_files = true;
+option java_outer_classname = "CosPersistenceProto";
+
+import "chief_of_state/v1/common.proto";
+import "google/protobuf/any.proto";
+
+// These protos are used by akka persistence to write to disk. They will
+// likely be offered as part of the open-source protos repo, but until
+// they are concrete/finalized, we are keeping them here.
+
+// Wrap the aggregate state and the meta data.
+message StateWrapper {
+  // the entity state
+  google.protobuf.Any state = 1;
+  // metadata from the event that made this state
+  chief_of_state.v1.MetaData meta = 3;
+}
+
+// EventWrapper is an event wrapper that holds both the
+// event and the corresponding aggregate root state.
+message EventWrapper {
+  // the event emitted
+  google.protobuf.Any event = 1;
+  // the state obtained from processing the event
+  google.protobuf.Any resulting_state = 2;
+  // meta data
+  chief_of_state.v1.MetaData meta = 3;
+}