How to stream from Cassandra and pass the context around using FlowWithContext

CassandraJornalContextStream.scala

case class MsgMetadata(offset: query.Offset, persistenceId: String, seqNum: Long)

trait EventsJournalOffsetDao {
  def offsetFor(projection: ProjectionId): Future[Option[query.Offset]]
  def saveOffset(projection: ProjectionId, currentOffset: query.Offset): Future[Unit]
}

class EventsStreamFactory(
    config: EventsStreamConfig,
    val projectionId: ProjectionId,
    projectionFlow: FlowWithContext[AppEvent, MsgMetadata, AppEvent, MsgMetadata, NotUsed],
    journalOffsetDao: EventsJournalOffsetDao
) {

def createStream(eventJournal: CassandraReadJournal): Source[Unit, NotUsed] = {
  RestartSource
    .withBackoff(minBackoff = 200.millis, maxBackoff = 3.seconds, randomFactor = 0.2) { () =>
      SourceWithContext
        .fromTuples(
          Source
            .fromFuture(journalOffsetDao.offsetFor(projectionId))
            .map(_.getOrElse(query.Offset.timeBasedUUID(eventJournal.firstOffset)))
            .flatMapConcat { currentOffset =>
              eventJournal
                .eventsByTag("MyTag", currentOffset)
            }
            .collect {
              case EventEnvelope(offset, persistenceId, sequenceNr, event: AppEvent) =>
                (event, MsgMetadata(offset, persistenceId, sequenceNr))
            }
        )
        .via(projectionFlow)
        .asSource
        .mapAsync(1) {
          case (_, context) => journalOffsetDao.saveOffset(projectionId, context.offset)
        }
    }
}