use offset from event envelope

* because then not depending on db migration of the db_timestamp

Author: patriknw

core/src/main/scala/akka/persistence/r2dbc/internal/StartingFromSnapshotStage.scala

@@ -11,6 +11,7 @@ import scala.util.Success
 import scala.util.Try
 
 import akka.annotation.InternalApi
+import akka.persistence.query.TimestampOffset
 import akka.persistence.query.typed.EventEnvelope
 import akka.persistence.r2dbc.internal.SnapshotDao.SerializedSnapshotRow
 import akka.stream.Attributes
@@ -37,7 +38,7 @@ import akka.util.RecencyList
     cacheCapacity: Int,
     sequenceNumberOfSnapshot: String => Future[Option[Long]],
     loadSnapshot: String => Future[Option[SnapshotDao.SerializedSnapshotRow]],
-    createEnvelope: SerializedSnapshotRow => EventEnvelope[Event])
+    createEnvelope: (SerializedSnapshotRow, TimestampOffset) => EventEnvelope[Event])
     extends GraphStage[FlowShape[EventEnvelope[Event], EventEnvelope[Event]]] {
   import StartingFromSnapshotStage._
 
@@ -95,7 +96,7 @@ import akka.util.RecencyList
         case Success((env, Some(snap))) =>
           inProgress = false
           if (env.sequenceNr == snap.seqNr) {
-            push(out, createEnvelope(snap))
+            push(out, createEnvelope(snap, env.offset.asInstanceOf[TimestampOffset]))
             updateState(snap.persistenceId, snap.seqNr, emitted = true)
           } else if (env.sequenceNr > snap.seqNr) {
             // event is ahead of snapshot, emit event

core/src/main/scala/akka/persistence/r2dbc/query/scaladsl/R2dbcReadJournal.scala

@@ -373,10 +373,8 @@ final class R2dbcReadJournal(system: ExtendedActorSystem, config: Config, cfgPat
             cacheCapacity = settings.querySettings.startFromSnapshotCacheCapacity,
             sequenceNumberOfSnapshot = persistenceId => snapshotDao.sequenceNumberOfSnapshot(persistenceId),
             loadSnapshot = persistenceId => snapshotDao.load(persistenceId, SnapshotSelectionCriteria.Latest),
-            createEnvelope = snapshotRow => {
-              val offset = TimestampOffset(snapshotRow.dbTimestamp, Map(snapshotRow.persistenceId -> snapshotRow.seqNr))
-              createEnvelopeFromSnapshot(snapshotRow, offset, transformSnapshot)
-            })))
+            createEnvelope =
+              (snapshotRow, offset) => createEnvelopeFromSnapshot(snapshotRow, offset, transformSnapshot))))
   }
 
   /**
@@ -425,10 +423,8 @@ final class R2dbcReadJournal(system: ExtendedActorSystem, config: Config, cfgPat
             cacheCapacity = settings.querySettings.startFromSnapshotCacheCapacity,
             sequenceNumberOfSnapshot = persistenceId => snapshotDao.sequenceNumberOfSnapshot(persistenceId),
             loadSnapshot = persistenceId => snapshotDao.load(persistenceId, SnapshotSelectionCriteria.Latest),
-            createEnvelope = snapshotRow => {
-              val offset = TimestampOffset(snapshotRow.dbTimestamp, Map(snapshotRow.persistenceId -> snapshotRow.seqNr))
-              createEnvelopeFromSnapshot(snapshotRow, offset, transformSnapshot)
-            })))
+            createEnvelope =
+              (snapshotRow, offset) => createEnvelopeFromSnapshot(snapshotRow, offset, transformSnapshot))))
   }
 
   private def checkStartFromSnapshotEnabled(methodName: String): Unit =

core/src/test/scala/akka/persistence/r2dbc/query/StartingFromSnapshotStageSpec.scala

@@ -92,7 +92,7 @@ class StartingFromSnapshotStageSpec extends ScalaTestWithActorTestKit with AnyWo
     def sequenceNumberOfSnapshot(persistenceId: String): Future[Option[Long]] =
       Future.successful(findEnvelope(PersistenceId.ofUniqueId(persistenceId), snapshotEnvelopes).map(_.sequenceNr))
 
-    def createEnvelopeFromSnapshotRow(snap: SerializedSnapshotRow): EventEnvelope[Any] =
+    def createEnvelopeFromSnapshotRow(snap: SerializedSnapshotRow, offset: TimestampOffset): EventEnvelope[Any] =
       findEnvelope(PersistenceId.ofUniqueId(snap.persistenceId), snap.seqNr, snapshotEnvelopes)
         .getOrElse(throw new IllegalArgumentException(s"Unknown envelope for [$snap]"))