Add possibility to unpersist data one generation younger

stamina-core/src/main/scala/stamina/migrations.scala

@@ -26,7 +26,7 @@ package object migrations {
    * creating Migrator[T, V2], etc. Its migration will be the identity
    * function so calling its migrate function will not have any effect.
    */
-  def from[T, V <: V1: VersionInfo]: Migrator[T, V] = new Migrator[T, V](Map(Version.numberFor[V] → identityMigration[T]))
+  def from[T, V <: V1: VersionInfo]: Migrator[T, V] = new Migrator[T, V](Map(Version.numberFor[V] → identityMigration[T]), None)
 }
 
 package migrations {
@@ -36,40 +36,71 @@ package migrations {
 
   /**
    * A `Migrator[R, V]` can migrate raw values of type R from older
-   * versions to version `V` by applying a specific `Migration[R]` to it.
+   * versions to version `V` or from version one generation younger
+   * than `V` back to `V` by applying a specific `Migration[R]` to it.
    *
    * You can create instances of `Migrator[R, V]` by using
    * a small type-safe DSL consisting of two parts: the
    * `from[R, V1]` function will create a
    * `Migrator[R, V1]` and then you can use the
    * `to[V](migration: Migration[R])` function to build
    * instances that can migrate multiple versions.
+   * `backFrom[V]` can be used to define migration from version
+   * one generation younger than current `Migrator` version.
+   * This functionality can be useful when you want to do
+   * e.g. rolling update in clustered application - first
+   * deploy to all nodes app that still persists events with current
+   * version but is able to read also newer events, then deploy
+   * app that actually saves data in new format.
+   * Note that `backFrom[V]` is effective only when
+   * it is called as the last one in call chain when defining
+   * `Migrator[R, V]`
    *
    * @example Using the json implementation:
    * {{{
    * val p = persister[CartCreated, V3]("cart-created",
    *   from[JsValue, V1]
    *     .to[V2](_.update('cart / 'items / * / 'price ! set[Int](1000)))
    *     .to[V3](_.update('timestamp ! set[Long](System.currentTimeMillis - 3600000L)))
+   *     .backFrom[V4](_.update('cart / 'items / * / 'name ! set[String]("unknown")))
    * )
    * }}}
    *
    *  @tparam R The type of raw data being migrated. In the JSON implementation this would be `JsValue`.
-   *  @tparam V The "current" version of this Migrator, i.e. it can migrate values from V1 to this version or any version in between.
+   *  @tparam V The "current" version of this Migrator, i.e. it can migrate values from V1 to this version or any version in between
+    *            and optionally from next version back to this one.
    */
-  class Migrator[R, V <: Version: VersionInfo] private[stamina] (migrations: Map[Int, Migration[R]] = Map.empty) {
-    def canMigrate(fromVersion: Int): Boolean = migrations.contains(fromVersion)
+  class Migrator[R, V <: Version : VersionInfo] private[stamina](migrations: Map[Int, Migration[R]] = Map.empty, backwardMigration: Option[Migration[R]] = None) {
+    def canMigrate(fromVersion: Int): Boolean = migrations.contains(fromVersion) || (backwardMigration.isDefined && fromVersion == Version.numberFor[V] + 1)
 
     def migrate(value: R, fromVersion: Int): R = {
-      migrations.get(fromVersion).map(_.apply(value)).getOrElse(
-        throw UndefinedMigrationException(fromVersion, Version.numberFor[V])
-      )
+      val thisVersion = Version.numberFor[V]
+      if (fromVersion <= thisVersion) {
+        migrations.get(fromVersion).map(_.apply(value)).getOrElse(
+          throw UndefinedMigrationException(fromVersion, thisVersion)
+        )
+      } else if (fromVersion == thisVersion + 1) {
+        backwardMigration.map(_.apply(value)).getOrElse(
+          throw UndefinedMigrationException(fromVersion, thisVersion)
+        )
+      } else {
+        throw UndefinedMigrationException(fromVersion, thisVersion)
+      }
     }
 
-    def to[NextV <: Version: VersionInfo](migration: Migration[R])(implicit isNextAfter: IsNextVersionAfter[NextV, V]) = {
+    def to[NextV <: Version : VersionInfo](migration: Migration[R])(implicit isNextAfter: IsNextVersionAfter[NextV, V]): Migrator[R, NextV] = {
       new Migrator[R, NextV](
-        migrations.mapValues(_ && migration) + (Version.numberFor[NextV] → identityMigration[R])
+        migrations.mapValues(_ && migration) + (Version.numberFor[NextV] → identityMigration[R]),
+        None
+      )
+    }
+
+    def backFrom[NextV <: Version : VersionInfo](migration: Migration[R])(implicit isNextAfter: IsNextVersionAfter[NextV, V]): Migrator[R, V] = {
+      new Migrator[R, V](
+        migrations,
+        Some(migration)
       )
     }
   }
+
 }

stamina-core/src/test/scala/stamina/migrations/MigratorSpec.scala

@@ -0,0 +1,118 @@
+package stamina.migrations
+
+import stamina._
+
+class MigratorSpec extends StaminaSpec {
+
+  val mV3WithBackwardMigration: Migrator[String, V3] =
+    from[String, V1]
+      .to[V2](_ + "V2")
+      .to[V3](_ + "V3")
+      .backFrom[V4](_.replace("V4", ""))
+
+  val mV3WithIgnoredBackwardMigration: Migrator[String, V3] =
+    from[String, V1]
+      .to[V2](_ + "V2")
+      .backFrom[V3](_ + "this should not be added")
+      .to[V3](_ + "V3")
+
+  val mV1WithBackwardMigration: Migrator[String, V1] =
+    from[String, V1]
+      .backFrom[V2](_.replace("V2", ""))
+
+  "Migrator with backward migration" should {
+    "be able to migrate" when {
+      "migration is from V1" in {
+        mV3WithBackwardMigration.canMigrate(1) shouldBe true
+      }
+
+      "migration is from V2" in {
+        mV3WithBackwardMigration.canMigrate(2) shouldBe true
+      }
+
+      "migration is from V3 (identity)" in {
+        mV3WithBackwardMigration.canMigrate(3) shouldBe true
+      }
+
+      "migration is from V4 (backward migration)" in {
+        mV3WithBackwardMigration.canMigrate(4) shouldBe true
+      }
+    }
+
+    "not be able to migrate" when {
+      "migration is from V5" in {
+        mV3WithBackwardMigration.canMigrate(5) shouldBe false
+      }
+    }
+
+    "migrate forward" when {
+      "migration is from V1" in {
+        mV3WithBackwardMigration.migrate("V1", 1) shouldBe "V1V2V3"
+      }
+
+      "migration is from V2" in {
+        mV3WithBackwardMigration.migrate("V1V2", 2) shouldBe "V1V2V3"
+      }
+
+      "migration is from V3" in {
+        mV3WithBackwardMigration.migrate("V1V2V3", 3) shouldBe "V1V2V3"
+      }
+    }
+
+    "migrate backward" when {
+      "migration is from V4" in {
+        mV3WithBackwardMigration.migrate("V1V2V3V4", 4) shouldBe "V1V2V3"
+      }
+    }
+  }
+
+  "Migrator with ignored backward migration" should {
+    "be able to migrate" when {
+      "migration is from V1" in {
+        mV3WithIgnoredBackwardMigration.canMigrate(1) shouldBe true
+      }
+
+      "migration is from V2" in {
+        mV3WithIgnoredBackwardMigration.canMigrate(2) shouldBe true
+      }
+
+      "migration is from V3 (identity)" in {
+        mV3WithIgnoredBackwardMigration.canMigrate(3) shouldBe true
+      }
+    }
+
+    "not be able to migrate" when {
+      "migration is from V4" in {
+        mV3WithIgnoredBackwardMigration.canMigrate(4) shouldBe false
+      }
+    }
+
+    "migrate" when {
+      "migration is from V1" in {
+        mV3WithIgnoredBackwardMigration.migrate("V1", 1) shouldBe "V1V2V3"
+      }
+
+      "migration is from V2" in {
+        mV3WithIgnoredBackwardMigration.migrate("V1V2", 2) shouldBe "V1V2V3"
+      }
+
+      "migration is from V3" in {
+        mV3WithIgnoredBackwardMigration.migrate("V1V2V3", 3) shouldBe "V1V2V3"
+      }
+    }
+  }
+
+  "Migrator V1 with backward migration" should {
+    "be able to migrate" when {
+      "migration is from V2" in {
+        mV1WithBackwardMigration.canMigrate(2) shouldBe true
+      }
+    }
+
+    "migrate" when {
+      "migration is from V2" in {
+        mV1WithBackwardMigration.migrate("V1V2", 2) shouldBe "V1"
+      }
+    }
+  }
+}

stamina-json/src/main/scala/stamina/json/json.scala

@@ -45,13 +45,22 @@ package object json {
    * and unpersist version 1. Use this function to produce the initial persister
    * for a new domain class/event/entity.
    */
-  def persister[T: RootJsonFormat: ClassTag](key: String): JsonPersister[T, V1] = new V1JsonPersister[T](key)
+  def persister[T: RootJsonFormat: ClassTag](key: String): JsonPersister[T, V1] = new V1JsonPersister[T](key, from[V1])
+
+  /**
+   * Creates a JsonPersister[T, V1], i.e. a JsonPersister that will only persist
+   * version 1 and can unpersist from both versions 1 and 2.
+   * Provided Migrator should be able to migrate values from version 2 back to
+   * version 1, it can be achieved by defining it like
+   * `from[V1].backFrom[V2](identity)`.
+   */
+  def persister[T: RootJsonFormat: ClassTag](key: String, migrator: JsonMigrator[V1]): JsonPersister[T, V1] = new V1JsonPersister[T](key, migrator)
 
   /**
    * Creates a JsonPersister[T, V] where V is a version greater than V1.
    * It will always persist instances of T to version V but it will use the specified
-   * JsonMigrator[V] to migrate any values older than version V to version V before
-   * unpersisting them.
+   * JsonMigrator[V] to migrate any values older or or one generation younger
+   * than version V to version V before unpersisting them.
    */
   def persister[T: RootJsonFormat: ClassTag, V <: Version: VersionInfo: MigratableVersion](key: String, migrator: JsonMigrator[V]): JsonPersister[T, V] = new VnJsonPersister[T, V](key, migrator)
 
@@ -69,7 +78,9 @@ package json {
       s"""JsonPersister[${implicitly[ClassTag[T]].runtimeClass.getSimpleName}, V${currentVersion}](key = "${key}") cannot unpersist data with key "${p.key}" and version ${p.version}."""
   }
 
-  private[json] class V1JsonPersister[T: RootJsonFormat: ClassTag](key: String) extends JsonPersister[T, V1](key) {
+  private[json] class V1JsonPersister[T: RootJsonFormat: ClassTag](key: String, migrator: JsonMigrator[V1]) extends JsonPersister[T, V1](key) {
+    override def canUnpersist(p: Persisted): Boolean = p.key == key && migrator.canMigrate(p.version)
+
     def persist(t: T): Persisted = Persisted(key, currentVersion, toJsonBytes(t))
     def unpersist(p: Persisted): T = {
       if (canUnpersist(p)) fromJsonBytes[T](p.bytes)

stamina-json/src/test/scala/stamina/json/JsonPersisterSpec.scala

@@ -2,29 +2,55 @@ package stamina
 package json
 
 class JsonPersisterSpec extends StaminaJsonSpec {
+
   import JsonTestDomain._
-  import spray.json.lenses.JsonLenses._
   import fommil.sjs.FamilyFormats._
+  import spray.json.lenses.JsonLenses._
 
   val v1CartCreatedPersister = persister[CartCreatedV1]("cart-created")
 
+  val v1CartCreatedPersisterWithBackwardMigration = persister[CartCreatedV1](
+    "cart-created",
+    from[V1].backFrom[V2](identity)
+  )
+
   val v2CartCreatedPersister = persister[CartCreatedV2, V2](
     "cart-created",
     from[V1].to[V2](_.update('cart / 'items / * / 'price ! set[Int](1000)))
   )
 
-  val v3CartCreatedPersister = persister[CartCreatedV3, V3](
-    "cart-created",
+  val migratorV3 =
     from[V1]
       .to[V2](_.update('cart / 'items / * / 'price ! set[Int](1000)))
       .to[V3](_.update('timestamp ! set[Long](System.currentTimeMillis - 3600000L)))
+
+  val v3CartCreatedPersister = persister[CartCreatedV3, V3](
+    "cart-created",
+    migratorV3
+  )
+
+  val v3CartCreatedPersisterWithBackwardMigration = persister[CartCreatedV3, V3](
+    "cart-created",
+    migratorV3
+      .backFrom[V4](_.update(('cart / 'items / * / 'name.?) ! setOrUpdateField[String]("unknown")(identity)))
+  )
+
+  val v4SimpleCartCreatedPersister = persister[CartCreatedV4, V4](
+    "cart-created",
+    from[V1].to[V2](identity).to[V3](identity).to[V4](identity)
   )
 
   "V1 persisters produced by SprayJsonPersister" should {
     "correctly persist and unpersist domain events " in {
       import v1CartCreatedPersister._
       unpersist(persist(v1CartCreated)) should equal(v1CartCreated)
     }
+
+    "fail to unpersist V2 domain events" in {
+      val v2Persisted = v2CartCreatedPersister.persist(v2CartCreated)
+      val e = intercept[IllegalArgumentException](v1CartCreatedPersister.unpersist(v2Persisted))
+      e.getMessage.contains("cannot unpersist data") shouldBe true
+    }
   }
 
   "V2 persisters with migrators produced by SprayJsonPersister" should {
@@ -57,5 +83,57 @@ class JsonPersisterSpec extends StaminaJsonSpec {
       v1Unpersisted.cart.items.map(_.price).toSet should equal(Set(1000))
       v2Unpersisted.timestamp should (be > 0L and be < System.currentTimeMillis)
     }
+
+    "fail to migrate and unpersist V4 domain events" in {
+      val v4Persisted = v4SimpleCartCreatedPersister.persist(v4CartCreated)
+      val e = intercept[IllegalArgumentException](v3CartCreatedPersister.unpersist(v4Persisted))
+      e.getMessage.contains("cannot unpersist data") shouldBe true
+    }
+  }
+
+  "V1 persisters with migrators with backward migrations produced by SprayJsonPersister" should {
+    "correctly persist and unpersist domain events" in {
+      import v1CartCreatedPersisterWithBackwardMigration._
+      unpersist(persist(v1CartCreated)) should equal(v1CartCreated)
+    }
+
+    "correctly migrte and unpersist V2 domain events" in {
+      val v2Persister = v2CartCreatedPersister.persist(v2CartCreated)
+      val v1UnpersistedFromV2 = v1CartCreatedPersisterWithBackwardMigration.unpersist(v2Persister)
+      v1UnpersistedFromV2 should equal(v1CartCreated)
+    }
+  }
+
+  "V4 dummy persister" should {
+    "correctly persist and unpersist domain events " in {
+      import v4SimpleCartCreatedPersister._
+      unpersist(persist(v4CartCreated)) should equal(v4CartCreated)
+    }
+  }
+
+  "V3 persisters with backward migrations and migrators produced by SprayJsonPersister" should {
+    "correctly persist and unpersist domain events " in {
+      import v3CartCreatedPersister._
+      unpersist(persist(v3CartCreated)) should equal(v3CartCreated)
+    }
+
+    "correctly migrate and unpersist V1 domain events" in {
+      val v1Persisted = v1CartCreatedPersister.persist(v1CartCreated)
+      val v2Persisted = v2CartCreatedPersister.persist(v2CartCreated)
+
+      val v1Unpersisted = v3CartCreatedPersisterWithBackwardMigration.unpersist(v1Persisted)
+      val v2Unpersisted = v3CartCreatedPersisterWithBackwardMigration.unpersist(v2Persisted)
+
+      v1Unpersisted.cart.items.map(_.price).toSet should equal(Set(1000))
+      v2Unpersisted.timestamp should (be > 0L and be < System.currentTimeMillis)
+    }
+
+    "correctly migrate and unpersist V4 domain events setting default value for removed field" in {
+      val v4Persisted = v4SimpleCartCreatedPersister.persist(v4CartCreated)
+
+      val v3UnpersistedFromV4 = v3CartCreatedPersisterWithBackwardMigration.unpersist(v4Persisted)
+
+      v3UnpersistedFromV4.cart.items.map(_.name) should equal(List("Wonka Bar", "unknown"))
+    }
   }
 }

stamina-json/src/test/scala/stamina/json/JsonTestDomain.scala

@@ -42,4 +42,17 @@ object JsonTestDomain {
   val v3Item2 = ItemV3(2, "Everlasting Gobstopper", 489)
   val v3Cart = CartV3(1, List(v3Item1, v3Item2))
   val v3CartCreated = CartCreatedV3(v3Cart, System.currentTimeMillis)
+
+  // ==========================================================================
+  // V4
+  // ==========================================================================
+
+  case class ItemV4(id: ItemId, name: Option[String], price: Int)
+  case class CartV4(id: CartId, items: List[ItemV4])
+  case class CartCreatedV4(cart: CartV4, timestamp: Long)
+
+  val v4Item1 = ItemV4(1, Some("Wonka Bar"), 500)
+  val v4Item2 = ItemV4(2, None, 489)
+  val v4Cart = CartV4(1, List(v4Item1, v4Item2))
+  val v4CartCreated = CartCreatedV4(v4Cart, System.currentTimeMillis)
 }