Added e2e tests

build.sbt

@@ -54,6 +54,7 @@ lazy val modules = List(
   doobieJsoniterCodecs,
   doobieUpickleCodecs,
   driverTests,
+  e2eTests,
   munitTestkit,
   docs,
   unidocs,
@@ -273,6 +274,26 @@ lazy val driverTests = module("backend-tests") {
     )
 }
 
+lazy val e2eTests = module("e2e") {
+  crossProject(JVMPlatform)
+    .crossType(CrossType.Pure)
+    .enablePlugins(NoPublishPlugin)
+    .dependsOn(
+      driverTests,
+      skunkBackend,
+      skunkCirceCodecs,
+      doobieBackend,
+      doobieCirceCodecs
+    )
+    .settings(
+      description := "E2E tests for postgres backends",
+      libraryDependencies ++= Seq(
+        "io.circe" %%% "circe-generic" % Versions.circe
+      )
+    )
+
+}
+
 lazy val munitTestkit = module("munit") {
   crossProject(JVMPlatform, JSPlatform, NativePlatform)
     .crossType(CrossType.Pure)
@@ -292,7 +313,9 @@ lazy val examples =
       skunkUpickleCodecs
     )
     .settings(
-      libraryDependencies += "io.circe" %%% "circe-generic" % Versions.circe
+      libraryDependencies ++= Seq(
+        "io.circe" %%% "circe-generic" % Versions.circe
+      )
     )
     .enablePlugins(NoPublishPlugin)
 

modules/e2e/src/main/scala/accounts/Domain.scala

@@ -0,0 +1,91 @@
+/*
+ * Copyright 2021 Hossein Naderi
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package dev.hnaderi.example.accounts
+
+import edomata.core.*
+import edomata.syntax.all.*
+import cats.implicits.*
+import cats.data.ValidatedNec
+
+enum Event {
+  case Opened
+  case Deposited(amount: BigDecimal)
+  case Withdrawn(amount: BigDecimal)
+  case Closed
+}
+
+enum Rejection {
+  case ExistingAccount
+  case NoSuchAccount
+  case InsufficientBalance
+  case NotSettled
+  case AlreadyClosed
+  case BadRequest
+}
+
+enum Account {
+  case New
+  case Open(balance: BigDecimal)
+  case Close
+
+  def open: Decision[Rejection, Event, Open] = this
+    .decide {
+      case New => Decision.accept(Event.Opened)
+      case _   => Decision.reject(Rejection.ExistingAccount)
+    }
+    .validate(_.mustBeOpen)
+
+  def close: Decision[Rejection, Event, Account] =
+    this.perform(mustBeOpen.toDecision.flatMap { account =>
+      if account.balance == 0 then Event.Closed.accept
+      else Decision.reject(Rejection.NotSettled)
+    })
+
+  def withdraw(amount: BigDecimal): Decision[Rejection, Event, Open] = this
+    .perform(mustBeOpen.toDecision.flatMap { account =>
+      if account.balance >= amount && amount > 0
+      then Decision.accept(Event.Withdrawn(amount))
+      else Decision.reject(Rejection.InsufficientBalance)
+      // We can model rejections to have values, which helps a lot for showing error messages, but it's out of scope for this document
+    })
+    .validate(_.mustBeOpen)
+
+  def deposit(amount: BigDecimal): Decision[Rejection, Event, Open] = this
+    .perform(mustBeOpen.toDecision.flatMap { account =>
+      if amount > 0 then Decision.accept(Event.Deposited(amount))
+      else Decision.reject(Rejection.BadRequest)
+    })
+    .validate(_.mustBeOpen)
+
+  private def mustBeOpen: ValidatedNec[Rejection, Open] = this match {
+    case o @ Open(_) => o.validNec
+    case New         => Rejection.NoSuchAccount.invalidNec
+    case Close       => Rejection.AlreadyClosed.invalidNec
+  }
+}
+
+object Account extends DomainModel[Account, Event, Rejection] {
+  def initial = New
+  def transition = {
+    case Event.Opened => _ => Open(0).validNec
+    case Event.Withdrawn(b) =>
+      _.mustBeOpen.map(s => s.copy(balance = s.balance - b))
+    case Event.Deposited(b) =>
+      _.mustBeOpen.map(s => s.copy(balance = s.balance + b))
+    case Event.Closed => _ => Close.validNec
+  }
+}

modules/e2e/src/main/scala/accounts/Service.scala

@@ -0,0 +1,62 @@
+/*
+ * Copyright 2021 Hossein Naderi
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package dev.hnaderi.example.accounts
+
+enum Command {
+  case Open
+  case Deposit(amount: BigDecimal)
+  case Withdraw(amount: BigDecimal)
+  case Close
+}
+
+enum Notification {
+  case AccountOpened(accountId: String)
+  case BalanceUpdated(accountId: String, balance: BigDecimal)
+  case AccountClosed(accountId: String)
+}
+
+object AccountService extends Account.Service[Command, Notification] {
+  import cats.Monad
+
+  def apply[F[_]: Monad]: App[F, Unit] = App.router {
+
+    case Command.Open =>
+      for {
+        ns <- App.state.decide(_.open)
+        acc <- App.aggregateId
+        _ <- App.publish(Notification.AccountOpened(acc))
+      } yield ()
+
+    case Command.Deposit(amount) =>
+      for {
+        deposited <- App.state.decide(_.deposit(amount))
+        accId <- App.aggregateId
+        _ <- App.publish(Notification.BalanceUpdated(accId, deposited.balance))
+      } yield ()
+
+    case Command.Withdraw(amount) =>
+      for {
+        withdrawn <- App.state.decide(_.withdraw(amount))
+        accId <- App.aggregateId
+        _ <- App.publish(Notification.BalanceUpdated(accId, withdrawn.balance))
+      } yield ()
+
+    case Command.Close =>
+      App.state.decide(_.close).void
+  }
+
+}

modules/e2e/src/main/scala/e2e.scala

@@ -0,0 +1,147 @@
+/*
+ * Copyright 2021 Hossein Naderi
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package tests
+
+import cats.effect.IO
+import cats.effect.Resource
+import dev.hnaderi.example.accounts.*
+import edomata.backend.Backend
+import edomata.backend.eventsourcing
+import edomata.backend.eventsourcing.AggregateState
+import edomata.backend.eventsourcing.StorageDriver
+import edomata.core.CommandMessage
+import munit.CatsEffectSuite
+import munit.Location
+
+import java.time.Instant
+
+abstract class e2e[Codec[_]](driver: Resource[IO, StorageDriver[IO, Codec]])(
+    using
+    Codec[Account],
+    Codec[Event],
+    Codec[Notification]
+) extends CatsEffectSuite {
+
+  private final case class SUT(
+      app: eventsourcing.Backend[IO, Account, Event, Rejection, Notification]
+  ) {
+    val service = app.compile(AccountService[IO])
+
+    def open(address: String) = cmd(address, Command.Open).flatMap(service)
+    def deposit(address: String, amount: BigDecimal) =
+      cmd(address, Command.Deposit(amount)).flatMap(service)
+
+    def state(
+        address: String
+    ) =
+      app.repository
+        .get(address)
+
+    def assertState(
+        address: String,
+        balance: BigDecimal,
+        version: Long
+    )(using Location) =
+      state(address)
+        .assertEquals(
+          AggregateState.Valid(Account.Open(balance), version)
+        )
+  }
+
+  private val AppWithCache = Backend
+    .builder(AccountService)
+    .from(driver)
+    .persistedSnapshot(maxInMem = 100)
+    .withRetryConfig(0)
+    .build
+    .map(SUT(_))
+
+  private val AppNoCache = Backend
+    .builder(AccountService)
+    .from(driver)
+    .disableCache
+    .persistedSnapshot(maxInMem = 100)
+    .withRetryConfig(0)
+    .build
+    .map(SUT(_))
+
+  def randomString = IO.randomUUID.map(_.toString())
+
+  def cmd(address: String, cmd: Command) = for {
+    newID <- randomString
+    now <- IO.realTime.map(fd => Instant.ofEpochMilli(fd.toMillis))
+  } yield CommandMessage(
+    newID.toString(),
+    now,
+    address,
+    cmd
+  )
+
+  test("Sanity") {
+    AppNoCache.use(app =>
+      for {
+        address <- randomString
+        _ <- app.open(address)
+        _ <- app.assertState(address, 0, 1)
+      } yield ()
+    )
+  }
+
+  test("Distributed workload should work without cache") {
+    Resource
+      .both(AppNoCache, AppNoCache)
+      .use((appA, appB) =>
+        for {
+          address <- tests.randomString
+          _ <- appA.open(address)
+          _ <- appA.deposit(address, 100)
+          _ <- appB.deposit(address, 50)
+
+          _ <- appA.assertState(address, 150, 3)
+          _ <- appB.assertState(address, 150, 3)
+        } yield ()
+      )
+  }
+
+  test("Distributed workload doesn't work with default cache") {
+    Resource
+      .both(AppWithCache, AppWithCache)
+      .use((appA, appB) =>
+        for {
+          address <- tests.randomString
+          _ <- appA.open(address)
+          _ <- appA.deposit(address, 100)
+
+          // B doesn't contain entity yet, so no problem
+          _ <- appB.deposit(address, 50)
+
+          _ <- appA.assertState(address, 150, 3)
+          // Now, both the applications have the same entity cached in memory
+          // If one of the apps changes that entity, it will make the other application's cache incorrect
+          // So the other application won't be able to issue a command on that entity anymore due to version conflicts
+          _ <- appB.assertState(address, 150, 3)
+
+          _ <- appA.deposit(address, 100).attempt
+          _ <- appB.deposit(address, 50)
+
+          _ <- appA.assertState(address, 200, 4)
+          _ <- appB.assertState(address, 200, 4)
+        } yield ()
+      )
+  }
+
+}

modules/e2e/src/test/scala/DoobieE2ETestSuites.scala

@@ -0,0 +1,50 @@
+/*
+ * Copyright 2021 Hossein Naderi
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package tests
+package doobie
+
+import _root_.doobie.util.transactor.Transactor
+import cats.effect.IO
+import cats.effect.kernel.Resource
+import dev.hnaderi.example.accounts.*
+import edomata.backend.*
+import edomata.doobie.*
+import edomata.doobie.BackendCodec
+import edomata.doobie.CirceCodec
+import edomata.doobie.DoobieDriver
+import io.circe.generic.auto.*
+
+private given BackendCodec[Event] = CirceCodec.jsonb
+private given BackendCodec[Notification] = CirceCodec.jsonb
+private given BackendCodec[Account] = CirceCodec.jsonb
+
+private def driver =
+  Resource.eval(
+    DoobieDriver(
+      "doobie_e2e",
+      Transactor
+        .fromDriverManager[IO](
+          driver = "org.postgresql.Driver",
+          url = "jdbc:postgresql:postgres",
+          user = "postgres",
+          password = "postgres",
+          logHandler = None
+        )
+    )
+  )
+
+class DoobieE2ETestSuites extends e2e(driver)