Akka-free event-sourced persistence

build.sbt

@@ -28,6 +28,7 @@ lazy val root = (project in file(".")
     testkit,
     persistence,
     eventsourcing,
+    `eventsourcing-persistence`,
     cluster,
     `cluster-sharding`))
 
@@ -64,13 +65,20 @@ lazy val testkit = (project in file("testkit")
     Akka.testkit % Test,
     scalatest % Test)))
 
+lazy val `eventsourcing-persistence` = (project in file("eventsourcing-persistence")
+  settings (name := "akka-effect-eventsourcing-persistence")
+  settings commonSettings
+  settings (
+  libraryDependencies ++= Seq(sstream)))
+
 lazy val persistence = (project in file("persistence")
   settings (name := "akka-effect-persistence")
   settings commonSettings
   dependsOn(
-    actor         % "test->test;compile->compile",
-    testkit       % "test->test;test->compile",
-    `actor-tests` % "test->test")
+    `eventsourcing-persistence` % "test->test;compile->compile",
+    actor                       % "test->test;compile->compile",
+    testkit                     % "test->test;test->compile",
+    `actor-tests`               % "test->test")
   settings (
     libraryDependencies ++= Seq(
       Akka.actor,

eventsourcing-persistence/src/main/scala/com/evolution/akkaeffect/eventsopircing/persistence/Event.scala

@@ -0,0 +1,8 @@
+package com.evolution.akkaeffect.eventsopircing.persistence
+
+trait Event[E] {
+
+  def event: E
+  def seqNr: SeqNr
+
+}

eventsourcing-persistence/src/main/scala/com/evolution/akkaeffect/eventsopircing/persistence/EventSourcedStore.scala

@@ -0,0 +1,41 @@
+package com.evolution.akkaeffect.eventsopircing.persistence
+
+/**
+  * Event sourcing persistence API: provides snapshot followed by stream of events
+  * @tparam F effect
+  * @tparam S snapshot
+  * @tparam E event
+  */
+trait EventSourcedStore[F[_], S, E] {
+
+  import EventSourcedStore._
+
+  /**
+    * Start recovery by retrieving snapshot (eager, happening on outer F)
+    * and preparing for loading events (lazy op, happens on [[Recovery#events()]] stream materialisation)
+    * @param id persistent ID
+    * @param criteria snapshot lookup criteria
+    * @return [[Recovery]] instance, representing __started__ recovery
+    */
+  def recover(id: Id, criteria: Criteria): F[Recovery[F, S, E]]
+
+}
+
+object EventSourcedStore {
+
+  /** ID of persistent actor
+    * @see [[com.evolutiongaming.akkaeffect.persistence.EventSourcedId]]
+    * @see [[akka.persistence.PersistentActor.persistenceId]]
+    */
+  final case class Id(value: String) extends AnyVal
+
+  /**
+    * Snapshot lookup criteria
+    * @see [[akka.persistence.SnapshotSelectionCriteria]]
+    */
+  final case class Criteria(maxSequenceNr: Long = Long.MaxValue,
+                            maxTimestamp: Long = Long.MaxValue,
+                            minSequenceNr: Long = 0L,
+                            minTimestamp: Long = 0L)
+
+}

eventsourcing-persistence/src/main/scala/com/evolution/akkaeffect/eventsopircing/persistence/Recovery.scala

@@ -0,0 +1,18 @@
+package com.evolution.akkaeffect.eventsopircing.persistence
+
+import com.evolutiongaming.sstream.Stream
+
+/**
+  * Representation of __started__ recovery process:
+  * snapshot is already loaded in memory (if any)
+  * while events will be loaded only on materialisation of [[Stream]]
+  * @tparam F effect
+  * @tparam S snapshot
+  * @tparam E event
+  */
+trait Recovery[F[_], S, E] {
+
+  def snapshot: Option[Snapshot[S]]
+  def events: Stream[F, Event[E]]
+
+}

eventsourcing-persistence/src/main/scala/com/evolution/akkaeffect/eventsopircing/persistence/Snapshot.scala

@@ -0,0 +1,16 @@
+package com.evolution.akkaeffect.eventsopircing.persistence
+
+import java.time.Instant
+
+trait Snapshot[S] {
+
+  def snapshot: S
+  def metadata: Snapshot.Metadata
+
+}
+
+object Snapshot {
+
+  final case class Metadata(seqNr: SeqNr, timestamp: Instant)
+
+}

eventsourcing-persistence/src/main/scala/com/evolution/akkaeffect/eventsopircing/persistence/package.scala

@@ -0,0 +1,7 @@
+package com.evolution.akkaeffect.eventsopircing
+
+package object persistence {
+
+  type SeqNr = Long
+
+}

persistence/src/main/scala/com/evolutiongaming/akkaeffect/persistence/EventSourcedStoreOf.scala

@@ -0,0 +1,167 @@
+package com.evolutiongaming.akkaeffect.persistence
+
+import akka.persistence.SnapshotSelectionCriteria
+import cats.syntax.all._
+import cats.effect.{Async, Ref, Resource, Sync}
+import akka.persistence.journal.AsyncRecovery
+import akka.persistence.snapshot.SnapshotStore
+import com.evolution.akkaeffect.eventsopircing.persistence.{
+  Event,
+  EventSourcedStore,
+  Recovery,
+  Snapshot
+}
+import com.evolutiongaming.catshelper.FromFuture
+import com.evolutiongaming.catshelper.ToTry
+import com.evolutiongaming.sstream.Stream
+
+import java.time.Instant
+import scala.concurrent.Future
+import scala.util.Try
+import com.evolutiongaming.sstream.FoldWhile._
+
+object EventSourcedStoreOf {
+
+  /**
+    * [[EventSourcedStore]] implementation based on Akka Persistence API.
+    *
+    * The implementation delegates snapshot and events load to [[SnapshotStore]] and [[AsyncRecovery]].
+    * Snapshot loaded on [[EventSourcedStore#recover]] F while events loaded lazily:
+    * first events will be available for [[Stream#foldWhileM]] while tail still loaded by [[AsyncRecovery]]
+    *
+    * @param snapshotStore Akka Persistence snapshot (plugin)
+    * @param journal Akka Persistence journal (plugin)
+    * @tparam F effect
+    * @tparam S snapshot
+    * @tparam E event
+    * @return resource of [[EventSourcedStore]]
+    */
+  def fromAkka[F[_]: Async: ToTry, S, E](snapshotStore: SnapshotStore,
+                                         journal: AsyncRecovery,
+  ): Resource[F, EventSourcedStore[F, S, E]] = {
+
+    val eventSourcedStore = new EventSourcedStore[F, S, E] {
+
+      override def recover(
+        id: EventSourcedStore.Id,
+        criteria: EventSourcedStore.Criteria
+      ): F[Recovery[F, S, E]] = {
+
+        val snapshotSelectionCriteria = SnapshotSelectionCriteria(
+          criteria.maxSequenceNr,
+          criteria.maxTimestamp,
+          criteria.minSequenceNr,
+          criteria.minTimestamp
+        )
+
+        snapshotStore
+          .loadAsync(id.value, snapshotSelectionCriteria)
+          .liftTo[F]
+          .map { offer =>
+            new Recovery[F, S, E] {
+
+              override val snapshot: Option[Snapshot[S]] =
+                offer.map { offer =>
+                  new Snapshot[S] {
+                    override def snapshot: S = offer.snapshot.asInstanceOf[S]
+
+                    override def metadata: Snapshot.Metadata =
+                      Snapshot.Metadata(
+                        seqNr = offer.metadata.sequenceNr,
+                        timestamp =
+                          Instant.ofEpochMilli(offer.metadata.timestamp)
+                      )
+                  }
+                }
+
+              override val events: Stream[F, Event[E]] = {
+                val fromSequenceNr =
+                  snapshot.map(_.metadata.seqNr).getOrElse(0L)
+
+                val stream = for {
+
+                  buffer <- Ref[F].of(Vector.empty[Event[E]])
+
+                  highestSequenceNr <- journal
+                    .asyncReadHighestSequenceNr(id.value, fromSequenceNr)
+                    .liftTo[F]
+
+                  replayed <- Sync[F].delay {
+
+                    journal.asyncReplayMessages(
+                      id.value,
+                      fromSequenceNr,
+                      highestSequenceNr,
+                      Long.MaxValue
+                    ) { persisted =>
+                      if (persisted.deleted) {} else {
+                        val event = new Event[E] {
+                          override val event: E =
+                            persisted.payload.asInstanceOf[E]
+                          override val seqNr: SeqNr =
+                            persisted.sequenceNr
+                        }
+                        val _ = buffer.update(_ :+ event).toTry
+                      }
+                    }
+
+                  }
+                } yield {
+
+                  new Stream[F, Event[E]] {
+
+                    override def foldWhileM[L, R](
+                      l: L
+                    )(f: (L, Event[E]) => F[Either[L, R]]): F[Either[L, R]] = {
+
+                      l.asLeft[R]
+                        .tailRecM {
+                          case Left(l) =>
+                            for {
+                              events <- buffer.getAndSet(Vector.empty[Event[E]])
+                              result <- events.foldWhileM(l)(f)
+                              result <- result match {
+
+                                case l: Left[L, R] =>
+                                  for {
+                                    replayed <- Sync[F].delay(
+                                      replayed.isCompleted
+                                    )
+                                  } yield
+                                    if (replayed) l.asRight[Either[L, R]]
+                                    else l.asLeft[Either[L, R]]
+
+                                case result =>
+                                  result.asRight[Either[L, R]].pure[F]
+
+                              }
+                            } yield result
+
+                          case result => result.asRight[Either[L, R]].pure[F]
+                        }
+                    }
+
+                  }
+                }
+
+                Stream.lift(stream).flatten
+              }
+            }
+          }
+      }
+
+    }
+
+    eventSourcedStore.pure[Resource[F, *]]
+
+  }
+
+  implicit class FromFutureSyntax[A](val future: Future[A]) extends AnyVal {
+    def liftTo[F[_]: FromFuture]: F[A] = FromFuture[F].apply(future)
+  }
+
+  implicit class ToTrySyntax[F[_], A](val fa: F[A]) extends AnyVal {
+    def toTry(implicit F: ToTry[F]): Try[A] = F(fa)
+  }
+
+}

project/Dependencies.scala

@@ -10,6 +10,7 @@ object Dependencies {
   val `kind-projector`            = "org.typelevel"          % "kind-projector"            % "0.13.2"
   val pureconfig                  = "com.github.pureconfig" %% "pureconfig"                % "0.17.3"
   val smetrics                    = "com.evolutiongaming"   %% "smetrics"                  % "2.0.0"
+  val sstream                     = "com.evolutiongaming"   %% "sstream"                   % "1.0.1"
 
   object Cats {
     private val version = "2.9.0"