Migration tool

build.sbt

@@ -4,7 +4,7 @@ lazy val `akka-persistence-jdbc` = project
   .in(file("."))
   .enablePlugins(ScalaUnidocPlugin)
   .disablePlugins(MimaPlugin, SitePlugin)
-  .aggregate(core, migration, docs)
+  .aggregate(core, migration, docs, tools)
   .settings(publish / skip := true)
 
 lazy val core = project
@@ -30,6 +30,16 @@ lazy val migration = project
     libraryDependencies ++= Dependencies.Migration,
     publish / skip := true)
 
+lazy val tools = project
+  .in(file("tools"))
+  .enablePlugins(Publish)
+  .disablePlugins(SitePlugin, MimaPlugin)
+  .settings(
+    name := "akka-persistence-jdbc-schema-utils",
+    libraryDependencies ++= Dependencies.Migration ++ Dependencies.Libraries,
+    publish / skip := true)
+  .dependsOn(core)
+
 lazy val docs = project
   .enablePlugins(ProjectAutoPlugin, AkkaParadoxPlugin, ParadoxSitePlugin, PreprocessPlugin, PublishRsyncPlugin)
   .disablePlugins(MimaPlugin)

project/Dependencies.scala

@@ -33,7 +33,7 @@ object Dependencies {
     "org.scalatest" %% "scalatest" % ScalaTestVersion % Test) ++ JdbcDrivers.map(_ % Test)
 
   val Migration: Seq[ModuleID] = Seq(
-    "org.flywaydb" % "flyway-core" % "7.5.1",
+    "org.flywaydb" % "flyway-core" % "7.5.2",
     "com.typesafe" % "config" % "1.4.1",
     "ch.qos.logback" % "logback-classic" % "1.2.3",
     "org.testcontainers" % "postgresql" % "1.15.1" % Test,

tools/src/main/resources/db/migration/postgres/V1__create-table.sql

@@ -0,0 +1,56 @@
+-- creation of the new akka persistence jdbc schemas
+
+-- event_journal table
+CREATE TABLE IF NOT EXISTS ${apj:event_journal}
+(
+    ordering           BIGSERIAL,
+    persistence_id     VARCHAR(255)          NOT NULL,
+    sequence_number    BIGINT                NOT NULL,
+    deleted            BOOLEAN DEFAULT FALSE NOT NULL,
+
+    writer             VARCHAR(255)          NOT NULL,
+    write_timestamp    BIGINT,
+    adapter_manifest   VARCHAR(255),
+
+    event_ser_id       INTEGER               NOT NULL,
+    event_ser_manifest VARCHAR(255)          NOT NULL,
+    event_payload      BYTEA                 NOT NULL,
+
+    meta_ser_id        INTEGER,
+    meta_ser_manifest  VARCHAR(255),
+    meta_payload       BYTEA,
+
+    PRIMARY KEY (persistence_id, sequence_number)
+);
+
+CREATE UNIQUE INDEX ${apj:event_journal}_ordering_idx ON ${apj:event_journal} (ordering);
+
+-- event_tag table
+CREATE TABLE IF NOT EXISTS ${apj:event_tag}
+(
+    event_id BIGINT,
+    tag      VARCHAR(256),
+    PRIMARY KEY (event_id, tag),
+    CONSTRAINT fk_${apj:event_journal}
+        FOREIGN KEY (event_id)
+            REFERENCES ${apj:event_journal} (ordering)
+            ON DELETE CASCADE
+);
+
+-- state_snapshot table
+CREATE TABLE IF NOT EXISTS ${apj:state_snapshot}
+(
+    persistence_id        VARCHAR(255) NOT NULL,
+    sequence_number       BIGINT       NOT NULL,
+    created               BIGINT       NOT NULL,
+
+    snapshot_ser_id       INTEGER      NOT NULL,
+    snapshot_ser_manifest VARCHAR(255) NOT NULL,
+    snapshot_payload      BYTEA        NOT NULL,
+
+    meta_ser_id           INTEGER,
+    meta_ser_manifest     VARCHAR(255),
+    meta_payload          BYTEA,
+
+    PRIMARY KEY (persistence_id, sequence_number)
+);

tools/src/main/scala/akka/persistence/jdbc/tools/SchemasUtil.scala

@@ -0,0 +1,65 @@
+/*
+ * Copyright (C) 2014 - 2019 Dennis Vriend <https://github.com/dnvriend>
+ * Copyright (C) 2019 - 2021 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package akka.persistence.jdbc.tools
+
+import com.typesafe.config.Config
+import org.flywaydb.core.Flyway
+import org.flywaydb.core.api.Location
+import org.flywaydb.core.api.configuration.FluentConfiguration
+import org.flywaydb.core.api.output.MigrateResult
+
+import scala.collection.JavaConverters.{ collectionAsScalaIterableConverter, mapAsJavaMapConverter }
+
+/**
+ * executes the database migration
+ *
+ * @param config the application config
+ */
+final case class SchemasUtil(config: Config) {
+  private val userKey: String = "jdbc-journal.slick.db.user"
+  private val passwordKey: String = "jdbc-journal.slick.db.password"
+  private val urlKey: String = "jdbc-journal.slick.db.url"
+
+  // Flyway placeholders values keys
+  private val eventJournalPlaceholderValueKey: String = "jdbc-journal.tables.event_journal.tableName"
+  private val eventTagPlaceholderValueKey: String = "jdbc-journal.tables.event_tag.tableName"
+  private val stateSnapshotPlaceholderValueKey: String = "jdbc-snapshot-store.tables.snapshot.tableName"
+
+  // Flyway placeholders
+  private val eventJournalPlaceholder = "apj:event_journal"
+  private val eventTagPlaceholder: String = "apj:event_tag"
+  private val stateSnapshotPlaceholder: String = "apj:state_snapshot"
+
+  private val url: String = config.getString(urlKey)
+  private val user: String = config.getString(userKey)
+  private val password: String = config.getString(passwordKey)
+
+  /**
+   * create the persistence schemas.
+   *
+   * @return the list of migration versions run
+   */
+  def createIfNotExists(): Seq[String] = {
+    val flywayConfig: FluentConfiguration = Flyway.configure
+      .dataSource(url, user, password)
+      .table("apj_schema_history")
+      .locations(new Location("classpath:db/migration/postgres"))
+      .ignoreMissingMigrations(true) // in case someone has some missing migrations
+      .placeholders(Map(
+        eventJournalPlaceholder -> config.getString(eventJournalPlaceholderValueKey),
+        eventTagPlaceholder -> config.getString(eventTagPlaceholderValueKey),
+        stateSnapshotPlaceholder -> config.getString(stateSnapshotPlaceholderValueKey)).asJava)
+
+    val flyway: Flyway = flywayConfig.load
+    flyway.baseline()
+
+    // running the migration
+    val result: MigrateResult = flyway.migrate()
+
+    // let us return the sequence of migration versions applied sorted in a descending order
+    result.migrations.asScala.toSeq.map(_.version).sortWith(_ > _)
+  }
+}

tools/src/main/scala/akka/persistence/jdbc/tools/migration/DataMigrator.scala

@@ -0,0 +1,60 @@
+/*
+ * Copyright (C) 2014 - 2019 Dennis Vriend <https://github.com/dnvriend>
+ * Copyright (C) 2019 - 2021 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package akka.persistence.jdbc.tools.migration
+
+import akka.actor.ActorSystem
+import akka.persistence.jdbc.config.{ JournalConfig, ReadJournalConfig, SnapshotConfig }
+import akka.persistence.jdbc.db.SlickExtension
+import akka.persistence.jdbc.journal.dao.DefaultJournalDao
+import akka.persistence.jdbc.query.dao.legacy.ByteArrayReadJournalDao
+import akka.persistence.jdbc.snapshot.dao.legacy.ByteArraySnapshotDao
+import akka.persistence.jdbc.snapshot.dao.DefaultSnapshotDao
+import akka.serialization.SerializationExtension
+import com.typesafe.config.Config
+import slick.basic.DatabaseConfig
+import slick.jdbc.{ JdbcBackend, JdbcProfile }
+import slick.jdbc
+
+/**
+ * this class will be implemented by both the journal and the snapshot migration class
+ *
+ * @param config the application config
+ * @param system the actor system
+ */
+abstract class DataMigrator(config: Config)(implicit system: ActorSystem) {
+
+  import system.dispatcher
+
+  // get an instance of the database and the Jdbc profile
+  protected val profile: JdbcProfile = DatabaseConfig.forConfig[JdbcProfile]("slick").profile
+
+  // get the various configuration
+  protected val journalConfig = new JournalConfig(config.getConfig("jdbc-journal"))
+  protected val readJournalConfig = new ReadJournalConfig(config.getConfig("jdbc-read-journal"))
+
+  protected val journaldb: JdbcBackend.Database =
+    SlickExtension(system).database(config.getConfig("jdbc-read-journal")).database
+  protected val snapshotdb: jdbc.JdbcBackend.Database =
+    SlickExtension(system).database(config.getConfig("jdbc-snapshot-store")).database
+
+  // get an instance of the legacy read journal dao
+  protected val legacyReadJournalDao: ByteArrayReadJournalDao =
+    new ByteArrayReadJournalDao(journaldb, profile, readJournalConfig, SerializationExtension(system))
+
+  // get an instance of the default journal dao
+  protected val defaultJournalDao: DefaultJournalDao =
+    new DefaultJournalDao(journaldb, profile, journalConfig, SerializationExtension(system))
+
+  protected val snapshotConfig = new SnapshotConfig(config.getConfig("jdbc-snapshot-store"))
+
+  // get the instance of the legacy snapshot dao
+  val legacySnapshotDao: ByteArraySnapshotDao =
+    new ByteArraySnapshotDao(snapshotdb, profile, snapshotConfig, SerializationExtension(system))
+
+  // get the instance if the default snapshot dao
+  val defaultSnapshotDao: DefaultSnapshotDao =
+    new DefaultSnapshotDao(snapshotdb, profile, snapshotConfig, SerializationExtension(system))
+}

tools/src/main/scala/akka/persistence/jdbc/tools/migration/LegacyJournalDataMigrator.scala

@@ -0,0 +1,61 @@
+/*
+ * Copyright (C) 2014 - 2019 Dennis Vriend <https://github.com/dnvriend>
+ * Copyright (C) 2019 - 2021 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package akka.persistence.jdbc.tools.migration
+
+import akka.NotUsed
+import akka.actor.ActorSystem
+import akka.persistence.{ AtomicWrite, Persistence, PersistentRepr }
+import akka.persistence.journal.EventAdapter
+import akka.stream.scaladsl.Source
+import com.typesafe.config.Config
+
+import scala.collection.immutable
+import scala.concurrent.Future
+import scala.util.Try
+
+/**
+ * This will help migrate the legacy journal data onto the new journal schema with the
+ * appropriate serialization
+ *
+ * @param config the application config
+ * @param system the actor system
+ */
+final case class LegacyJournalDataMigrator(config: Config)(implicit system: ActorSystem) extends DataMigrator(config) {
+
+  private val eventAdapters = Persistence(system).adaptersFor("", config)
+
+  private def adaptEvents(repr: PersistentRepr): Seq[PersistentRepr] = {
+    val adapter: EventAdapter = eventAdapters.get(repr.payload.getClass)
+    adapter.fromJournal(repr.payload, repr.manifest).events.map(repr.withPayload)
+  }
+
+  /**
+   * reads all the current events in the legacy journal
+   *
+   * @return the source of all the events
+   */
+  private def allEvents(): Source[PersistentRepr, NotUsed] = {
+    legacyReadJournalDao
+      .allPersistenceIdsSource(Long.MaxValue)
+      .flatMapConcat((persistenceId: String) => {
+        legacyReadJournalDao
+          .messagesWithBatch(persistenceId, 0L, Long.MaxValue, readJournalConfig.maxBufferSize, None)
+          .mapAsync(1)(reprAndOrdNr => Future.fromTry(reprAndOrdNr))
+          .mapConcat { case (repr: PersistentRepr, _) =>
+            adaptEvents(repr)
+          }
+      })
+  }
+
+  /**
+   * write all legacy events into the new journal tables applying the proper serialization
+   */
+  def migrate(): Source[Seq[Try[Unit]], NotUsed] = {
+    allEvents().mapAsync(1) { pr: PersistentRepr =>
+      defaultJournalDao.asyncWriteMessages(immutable.Seq(AtomicWrite(pr)))
+    }
+  }
+}

tools/src/main/scala/akka/persistence/jdbc/tools/migration/LegacySnapshotDataMigrator.scala

@@ -0,0 +1,42 @@
+/*
+ * Copyright (C) 2014 - 2019 Dennis Vriend <https://github.com/dnvriend>
+ * Copyright (C) 2019 - 2021 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package akka.persistence.jdbc.tools.migration
+
+import akka.actor.ActorSystem
+import akka.NotUsed
+import akka.persistence.SnapshotMetadata
+import akka.stream.scaladsl.Source
+import com.typesafe.config.Config
+
+import scala.concurrent.Future
+
+/**
+ * This will help migrate the legacy snapshot data onto the new snapshot schema with the
+ * appropriate serialization
+ *
+ * @param config the application config
+ * @param system the actor system
+ */
+case class LegacySnapshotDataMigrator(config: Config)(implicit system: ActorSystem) extends DataMigrator(config) {
+
+  import system.dispatcher
+
+  /**
+   * write the latest state snapshot into the new snapshot table applying the proper serialization
+   */
+  def migrate(): Source[Option[Future[Unit]], NotUsed] = {
+    legacyReadJournalDao.allPersistenceIdsSource(Long.MaxValue).mapAsync(1) { persistenceId: String =>
+      legacySnapshotDao
+        .latestSnapshot(persistenceId)
+        .map((o: Option[(SnapshotMetadata, Any)]) => {
+          o.map((result: (SnapshotMetadata, Any)) => {
+            val (meta, data) = result
+            defaultSnapshotDao.save(meta, data)
+          })
+        })
+    }
+  }
+}