Foreword

Thank you for your interest in this book. I hope you’ll have a good time reading it and learn something from it.

About this book

This book is intended for the intermediate Scala programmer who is interested in functional programming and works mainly on the web service backend side. Ideally she has experience with libraries like Akka HTTP¹ and Slick² which are in heavy use in that area.

However maybe you have wondered if we can’t do better even though aforementioned projects are battle tested and proven.

The answer to this can be found in this book which is intended to be read from cover to cover in the given order. Within the book the following libraries will be used: Cats³, Cats Effect⁴, http4s⁵, Doobie⁶, Refined⁷, fs2⁸, tapir⁹, Monocle¹⁰ and probably others. ;-)

This edition includes a chapter about migrating the project to Scala 3. Which includes all the nasty issues that we tend to run into if we touch code after a longer time.

Code and book source can be found in the following repository: https://github.com/jan0sch/pfhais

Copyleft Notice

This book uses the Creative Commons Attribution ShareAlike 4.0 International (CC BY-SA 4.0) license. The code snippets in this book are licensed under CC0 which means you can use them without restriction. Excerpts from libraries maintain their license.

Thanks

I would like to thank my beloved wife and family who bear with me and make all of this possible.
Also I send a big thank you to all the nice people from the Scala community which I’ve had the pleasure to meet. Special thanks go to Adam Warski (tapir), Frank S. Thomas (refined), Julien Truffaut (Monocle) and Ross A. Baker (http4s) for their help, advise and patience.

Our use case

For better understanding we will implement a small use case in both impure and pure way. The following section will outline the specification.

Service specification

First we need to specify the exact scope and API of our service. We’ll design a service with a minimal API to keep things simple. It shall fulfil the following requirements.

The service shall provide HTTP API endpoints for:

1. the creation of a product data type identified by a unique id
2. adding translations for a product name by language code and unique id
3. returning the existing translations for a product
4. returning a list of all existing products with their translations

Data model

We will keep the model very simple to avoid going overboard with the implementation.

1. A language code shall be defined by the ISO 639-1 (e.g. a two letter code).
2. A translation shall contain a language code and a product name (non-empty string).
3. A product shall contain a unique id (UUID version 4) and a list of translations.

Database

The data will be stored in a relational database (RDBMS). Therefore we need to define the tables and relations within the database.

The products table

The table products must contain only the unique id which is also the primary key.

The names table

The table names must contain a column for the product id, one for the language code and one for the name. Its primary key is the combination of the product id and the language code. All columns must not be null. The relation to the products is realised by a foreign key constraint to the products table via the product id.

HTTP API

The HTTP API shall provide the following endpoints on the given paths:

The data shall be encoded in JSON using the following specification:

1 {
2   "lang": "ISO-639-1 Code",
3   "name": "A non empty string."
4 }

1 {
2   "id": "The-UUID-of-the-product",
3   "names": [
4     // A list of translations.
5   ]
6 }

This should be enough to get us started.

The state of the art

Within the Scala ecosystem the Akka-HTTP library is a popular choice for implementing server side backends for HTTP APIs. Another quite popular option is the Play framework but using a full blown web framework to just provide a thin API is overkill in most cases. As most services need a database the Slick library is another popular choice which completes the picture.

However while all mentioned libraries are battle tested and proven they still have problems.

Problems

In the domain of functional programming we want referential transparency which we will define in the following way:

Building on that we need pure functions which are

1. only dependent on their input
2. have no side effects

This means in turn that our functions will be referential transparent.

But, the mentioned libraries are built upon the Future from Scala which uses eager evaluation and breaks referential transparency. Let’s look at an example.

1 import scala.concurrent.Future
2 import scala.concurrent.ExecutionContext.Implicits.global
3 
4 for {
5   _ <- Future { println("Hi there!") }
6   _ <- Future { println("Hi there!") }
7 } yield ()

The code above will print the text Hi there! two times. But how about the following one?

1 import scala.concurrent.Future
2 import scala.concurrent.ExecutionContext.Implicits.global
3 
4 val printF = Future { println("Hi there!") }
5 
6 for {
7   _ <- printF
8   _ <- printF
9 } yield ()

Instead of printing the text two times it will print it only once even when there is no usage of printF at all (try omitting the for comprehension). This means that Future breaks referential transparency!

Maybe there is another way

If we want referential transparency, we must push the side effects to the boundaries of our system (program) which can be done by using lazy evaluation. Let’s repeat the previous example in a different way.

1 import cats.effect.IO
2 import cats.implicits._
3 
4 val effect = for {
5   _ <- IO(println("Hi there!"))
6   _ <- IO(println("Hi there!"))
7 } yield ()

The above code will produce no output. Only if we evaluate the variable effect which is of type IO[Unit] will the output be generated (try effect.unsafeRunSync in the REPL). Also the second approach works like expected.

1 import cats.effect.IO
2 import cats.implicits._
3 
4 val printF = IO(println("Hi there!"))
5 
6 val effect = for {
7   _ <- printF
8   _ <- printF
9 } yield ()

Suddenly we can much more easily reason about our code! And why is that? Well we don’t have unexpected side effects caused by code running even when it doesn’t need to. This is a sneak peak how pure code looks like. Now we only need to implement pure libraries for our use, or do we?

Luckily for us meanwhile there are several pure options available in the Scala ecosystem. We will stick to the Cats family of libraries namely http4s and Doobie as replacements for Akka-HTTP and Slick. They build upon the Cats Effect library which is an implementation of an IO monad for Scala. Some other options exist but we’ll stick to the one from Cats.

To be able to contrast both ways of implementing a service we will first implement it using Akka-HTTP and Slick and will then migrate to http4s and Doobie.

Impure implementation

We’ll be using the following libraries for the impure version of the service:

1. Akka (including Akka-HTTP and Akka-Streams)
2. Slick (as database layer)
3. Flyway for database migrations (or evolutions)
4. Circe for JSON codecs and akka-http-json as wrapper
5. Refined for using refined types
6. the PostgreSQL JDBC driver

I’ll spare you the sbt setup as you can look that up in the code repository (i.e. the impure folder in the book repo).

Models

First we’ll implement our models which are simple and straightforward. At first we need a class to store our translations or better a single translation.

1 final case class Translation(lang: String, name: String)

Technically it is okay but we have a bad feeling about it. Using Option[String] is of no use because both fields have to be set. But a String can always be null and contain a lot of unexpected stuff (literally anything).

So let us define some refined types which we can use later on. At first we need a language code which obeys the restrictions of ISO-639-1 and we need a stronger definition for a product name. For the former we use a regular expression and for the latter we simply expect a string which is not empty.

1 type LanguageCode = String Refined MatchesRegex[W.`"^[a-z]{2}$"`.T]
2 type ProductName = String Refined NonEmpty

Now we can give our translation model another try.

1 final case class Translation(lang: LanguageCode, name: ProductName)

Much better and while we’re at it we can also write the JSON codecs using the refined module of the Circe library. We put them into the companion object of the model.

1 object Translation {
2   implicit val decode: Decoder[Translation] =
3     Decoder.forProduct2("lang", "name")(Translation.apply)
4 
5   implicit val encode: Encoder[Translation] =
6     Encoder.forProduct2("lang", "name")(t => (t.lang, t.name))
7 }

Now onwards to the product model. Because we already know of refined types we can use them from start here.

1 type ProductId = String Refined Uuid
2 final case class Product(id: ProductId, names: List[Translation])

If we look closely we realise that a List maybe empty. Which is valid for the list but not for our product because we need at least one entry. Luckily for us the Cats library has us covered with the NonEmptyList data type. Including the JSON codecs this leads us to our final implementation.
Last but not least we really should be using the existing UUID data type instead of rolling our own refined string version - even when it is cool. ;-)

 1 type ProductId = java.util.UUID
 2 final case class Product(id: ProductId, names: NonEmptyList[Translation])
 3 
 4 object Product {
 5   implicit val decode: Decoder[Product] =
 6     Decoder.forProduct2("id", "names")(Product.apply)
 7 
 8   implicit val encode: Encoder[Product] =
 9     Encoder.forProduct2("id", "names")(p => (p.id, p.names))
10 }

We kept the type name ProductId by using a type alias. This is convenient but remember that a type alias does not add extra type safety (e.g. type Foo = String will be a String).

Well, maybe because a list may contain duplicate entries but the database will surely not because of unique constraints! So, let’s switch to a NonEmptySet which is also provided by Cats.

1 type ProductId = java.util.UUID
2 final case class Product(id: ProductId, names: NonEmptySet[Translation])

Now we have the models covered and can move on to the database layer.

Database layer

The database layer should provide a programmatic access to the database but also should it manage changes in the database. The latter one is called migrations or evolutions. From the available options we chose Flyway as the tool to manage our database schema.

Migrations

Flyway uses raw SQL scripts which have to be put into a certain location being /db/migration (under the resources folder) in our case. Also the files have to be named like VXX__some_name.sql (XX being a number) starting with V1. Please note that there are two underscores between the version prefix and the rest of the name! Because our database schema is very simply we’re done quickly:

 1 CREATE TABLE "products" (
 2   "id" UUID NOT NULL,
 3   CONSTRAINT "products_pk" PRIMARY KEY ("id")
 4 );
 5 
 6 CREATE TABLE "names" (
 7   "product_id" UUID       NOT NULL,
 8   "lang_code"  VARCHAR(2) NOT NULL,
 9   "name"       TEXT       NOT NULL,
10   CONSTRAINT "names_pk" 
11     PRIMARY KEY ("product_id", "lang_code"),
12   CONSTRAINT "names_product_id_fk" 
13     FOREIGN KEY ("product_id") 
14     REFERENCES "products" ("id") 
15     ON DELETE CASCADE ON UPDATE CASCADE
16 );

In the code you’ll see that we additionally set comments which are omitted from the code snippet above. This might be overkill here but it is a very handy feature to have and I advice you to use it for more complicated database schemas. Because the right comment (read information) in the right place might save a lot of time when trying to understand things.

Next we move on to the programmatic part which at first needs a configuration of our database connection. With Slick you have a multitude of options but we’ll use the “Typesafe Config”¹ approach.

 1 database {
 2   profile = "slick.jdbc.PostgresProfile$"
 3   db {
 4     connectionPool = "HikariCP"
 5     dataSourceClass = "org.postgresql.ds.PGSimpleDataSource"
 6     properties {
 7       serverName = "localhost"
 8       portNumber = "5432"
 9       databaseName = "impure"
10       user = "impure"
11       password = "secret"
12     }
13     numThreads = 10
14   }
15 }

After we have this in place we can run the migrations via the API of Flyway. For this we have to load the configuration (we do it by creating an actor system), extract the needed information and create a JDBC url and use that with username and password to obtain a Flyway instance. On that one we simply call the method migrate() which will do the right thing. Basically it will check if the schema exists and decide to either create it, apply pending migrations or simply do nothing. The method will return the number of applied migrations.

 1 implicit val system: ActorSystem    = ActorSystem()
 2 implicit val mat: ActorMaterializer = ActorMaterializer()
 3 implicit val ec: ExecutionContext   = system.dispatcher
 4 
 5 val url = "jdbc:postgresql://" +
 6 system.settings.config.getString("database.db.properties.serverName") +
 7 ":" + system.settings.config.getString("database.db.properties.portNumber") +
 8 "/" + system.settings.config.getString("database.db.properties.databaseName")
 9 val user = system.settings.config.getString("database.db.properties.user")
10 val pass = system.settings.config.getString("database.db.properties.password")
11 val flyway = Flyway.configure().dataSource(url, user, pass).load()
12 val _ = flyway.migrate()

Let us continue to dive into the Slick table definitions.

Slick tables

Slick offers several options for approaching the database. For our example we will be using the lifted embedding but if needed Slick also provides the ability to perform plain SQL queries.
For the lifted embedding we have to define out tables in a way Slick can understand. While this can be tricky under certain circumstances our simple model is straightforward to implement.

1 final class Products(tag: Tag) extends Table[(UUID)](tag, "products") {
2   def id = column[UUID]("id", O.PrimaryKey)
3 
4   def * = (id)
5 }
6 val productsTable = TableQuery[Products]

As you can see above we’re using simple data types (not the refined ones) to have a more easy Slick implementation. However we can also use refined types for the price of using either the slick-refined library or writing custom column mappers.
Next we’ll implement the table for the translations which will also need some constraints.

 1 final class Names(tag: Tag) extends Table[(UUID, String, String)](tag, "names") {
 2   def productId = column[UUID]("product_id")
 3   def langCode  = column[String]("lang_code")
 4   def name      = column[String]("name")
 5 
 6   def pk = primaryKey("names_pk", (productId, langCode))
 7   def productFk =
 8     foreignKey("names_product_id_fk", productId, productsTable)(
 9       _.id,
10       onDelete = ForeignKeyAction.Cascade,
11       onUpdate = ForeignKeyAction.Cascade
12     )
13 
14   def * = (productId, langCode, name)
15 }
16 val namesTable = TableQuery[Names]

As you can see the definition of constraints is also pretty simple. Now our repository needs some functions for a more convenient access to the data.

 1 def loadProduct(id: ProductId): Future[Seq[(UUID, String, String)]] = {
 2   val program = for {
 3     (p, ns) <- productsTable
 4       .filter(_.id === id)
 5       .join(namesTable)
 6       .on(_.id === _.productId)
 7   } yield (p.id, ns.langCode, ns.name)
 8   dbConfig.db.run(program.result)
 9 }
10 
11 def loadProducts(): DatabasePublisher[(UUID, String, String)] = {
12   val program = for {
13     (p, ns) <- productsTable.join(namesTable)
14                  .on(_.id === _.productId).sortBy(_._1.id)
15   } yield (p.id, ns.langCode, ns.name)
16   dbConfig.db.stream(program.result)
17 }
18 
19 def saveProduct(p: Product): Future[List[Int]] = {
20   val cp      = productsTable += (p.id)
21   val program = DBIO.sequence(
22     cp :: saveTranslations(p).toList
23   ).transactionally
24   dbConfig.db.run(program)
25 }
26 
27 def updateProduct(p: Product): Future[List[Int]] = {
28   val program = namesTable
29     .filter(_.productId === p.id)
30     .delete
31     .andThen(DBIO.sequence(saveTranslations(p).toList))
32     .transactionally
33   dbConfig.db.run(program)
34 }
35 
36 protected def saveTranslations(p: Product): NonEmptyList[DBIO[Int]] = {
37   val save = saveTranslation(p.id)(_)
38   p.names.toNonEmptyList.map(t => save(t))
39 }
40 
41 /**
42   * Create a query to insert or update a given translation in the database.
43   *
44   * @param id The unique ID of the product.
45   * @param t  The translation to be saved.
46   * @return A composable sql query for Slick.
47   */
48 protected def saveTranslation(id: ProductId)(t: Translation): DBIO[Int] =
49   namesTable.insertOrUpdate((id, t.lang, t.name))

The last two functions are helpers to enable us to create a load queries which we can compose. They are used in the saveProduct and updateProduct functions to create a list of queries that are executed as bulk while the call to transactionally ensures that they will run within a transaction. When updating a product we first delete all existing translations to allow the removal of existing translations via an update. To be able to do so we use the andThen helper from Slick.
The loadProduct function simply returns a list of database rows from the needed join. Therefore we need a function which builds a Product type out of that.

 1 def fromDatabase(rows: Seq[(UUID, String, String)]): Option[Product] = {
 2   val po = for {
 3     (id, c, n) <- rows.headOption
 4     t          <- Translation.fromUnsafe(c)(n)
 5     p          <- Product(
 6                     id = id,
 7                     names = NonEmptySet.one[Translation](t)
 8                   ).some
 9   } yield p
10   po.map(
11     p =>
12       rows.drop(1).foldLeft(p) { (a, cols) =>
13         val (id, c, n) = cols
14         Translation.fromUnsafe(c)(n).fold(a)(t =>
15           a.copy(names = a.names.add(t))
16         )
17     }
18   )
19 }

But oh no! The compiler refuses to build it:

1 [error] .../impure/models/Product.scala:45:74:
2   could not find implicit value for parameter A: 
3     cats.kernel.Order[com.wegtam.books.pfhais.impure.models.Translation]
4 [error] p <- Product(id = id, names = NonEmptySet.one[Translation](t)).some
5 [error]                                                            ^

It seems we have to provide an instance of Order for our Translation model to make Cats happy. So we have think of an ordering for our model. A simple approach would be to simply order by the language code. Let’s try this:

1 import cats._
2 import cats.syntax.order._
3 
4 implicit val orderLanguageCode: Order[LanguageCode] = 
5   new Order[LanguageCode] {
6     def compare(x: LanguageCode, y: LanguageCode): Int =
7       x.value.compare(y.value)
8   }

You might have noticed the explicit call to .value to get the underlying string instance of our refined type. This is needed because the other option (using x.compare(y)) will compile but bless you with stack overflow errors. The reason is probably that the latter is compiled into code calling OrderOps#compare which is recursive.

1 import cats._
2 import cats.syntax.order._
3 
4 implicit val order: Order[Translation] =
5   new Order[Translation] {
6     def compare(x: Translation, y: Translation): Int =
7       x.lang.compare(y.lang)
8   }

So far we should have everything in place to make use of our database. Now we need to wire it all together.

Akka-HTTP routes

Defining the routes is pretty simple if you’re used to the Akka-HTTP routing DSL syntax.

 1 val route = path("product" / JavaUUID) { id: ProductId =>
 2   get {
 3     ???
 4   } ~ put {
 5     ???
 6   }
 7 } ~ path("products") {
 8   get {
 9     ???
10   } ~
11   post {
12     ???
13   }
14 }

We will fill in the details later on. But now for starting the actual server to make use of our routes.

1 val host       = system.settings.config.getString("api.host")
2 val port       = system.settings.config.getInt("api.port")
3 val srv        = Http().bindAndHandle(route, host, port)
4 val pressEnter = StdIn.readLine()
5 srv.flatMap(_.unbind()).onComplete(_ => system.terminate())

The code will fire up a server using the defined routes and hostname and port from the configuration to start a server. It will run until you press enter and then terminate. Let us now visit the code for each routing endpoint. We will start with the one for returning a single product.

 1 path("product" / JavaUUID) { id: ProductId =>
 2   get {
 3     complete {
 4       for {
 5         rows <- repo.loadProduct(id)
 6         prod <- Future { Product.fromDatabase(rows) }
 7       } yield prod
 8     }
 9   }
10 }

We load the raw product data from the repository and convert it into a proper product model. But to make the types align we have to wrap the second call in a Future otherwise we would get a compiler error. We don’t need to marshal the response because we are using the akka-http-json library which provides for example an ErrorAccumulatingCirceSupport import that handles this. Unless of course you do not have circe codecs defined for your types.

1 val route = path("product" / JavaUUID) { id: ProductId =>
2   put {
3     entity(as[Product]) { p =>
4       complete {
5         repo.updateProduct(p)
6       }
7     }
8   }
9 }

The route for updating a product is also very simple. We’re extracting the product entity via the entity(as[T]) directive from the request body and simply give it to the appropriate repository function. Now onwards to creating a new product.

1 path("products") {
2   post {
3     entity(as[Product]) { p =>
4       complete {
5         repo.saveProduct(p)
6       }
7     }
8   }
9 }

As you can see the function is basically the same except that we’re calling a different function from the repository. Last but not least let us take a look at the return all products endpoint.

 1 path("products") {
 2   get {
 3     complete {
 4       val products = for {
 5         rows <- repo.loadProducts()
 6         ps <- Future {
 7           rows.toList.groupBy(_._1).map {
 8             case (_, cols) => Product.fromDatabase(cols)
 9           }
10         }
11       } yield ps
12       products.map(_.toList.flatten)
13     }
14   }
15 }

This looks more complicated that the other endpoints. So what exactly are we doing here?
Well first we load the raw product data from the repository. Afterwards we convert it into the proper data model or to be more exact into a list of product entities.

The first thing that comes to mind is that we’re performing operations in memory. This is not different from the last time when we converted the data for a single product. Now however we’re talking about all products which may be a lot of data. Another obvious point is that we get a list of Option[Product] which we explicitly flatten at the end.

Maybe we should consider streaming the results. But we still have to group and combine the rows which belong to a single product into a product entity. Can we achieve that with streaming? Well, let’s look at our data flow.
We receive a list of 3 columns from the database in the following format: product id, language code, name. The tricky part being that multiple rows (list entries) can belong to the same product recognizable by the same value for the first column product id. At first we should simplify our problem by ensuring that the list will be sorted by the product id. This is done by adjusting the function loadProducts in the repository.

1 def loadProducts(): DatabasePublisher[(UUID, String, String)] = {
2   val program = for {
3     (p, ns) <- productsTable.join(namesTable).on(_.id === _.productId)
4                .sortBy(_._1.id)
5   } yield (p.id, ns.langCode, ns.name)
6   dbConfig.db.stream(program.result)
7 }

Now we can rely on the fact that we have seen all entries for one product if the product id in our list changes. Let’s adjust our code in the endpoint to make use of streaming now. Because Akka-HTTP is based on Akka-Streams we can simply use that.

 1 path("products") {
 2   get {
 3     implicit val jsonStreamingSupport: JsonEntityStreamingSupport =
 4       EntityStreamingSupport.json()
 5 
 6     val src = Source.fromPublisher(repo.loadProducts())
 7     val products: Source[Product, NotUsed] = src
 8       .collect(
 9         cs =>
10           Product.fromDatabase(Seq(cs)) match {
11             case Some(p) => p
12         }
13       )
14       .groupBy(Int.MaxValue, _.id)
15       .fold(Option.empty[Product])(
16         (op, x) => op.fold(x.some)(p =>
17           p.copy(names = p.names ::: x.names).some
18         )
19       )
20       .mergeSubstreams
21       .collect(
22         op =>
23           op match {
24             case Some(p) => p
25         }
26       )
27     complete(products)
28   }
29 }

Wow, this may look scary but let’s break it apart piece by piece. At first we need an implicit value which provides streaming support for JSON. Next we create a Source from the database stream. Now we implement the processing logic via the high level streams API. We collect every defined output of our helper function fromDatabase which leads to a stream of Product entities. But we have created way too many (Each product will be created as often as it has translations.). So we group our stream by the product id which creates a new stream for each product id holding only the entities for the specific product. We fold over each of these streams by merging together the list of translations (names). Afterwards we merge the streams back together and run another collect function to simply get a result stream of Product and not of Option[Product]. Last but not least the stream is passed to the complete function which will do the right thing.

Problems with the solution

The solution has two problems:

1. The number of individual streams (and thus products) is limited to Int.MaxValue.
2. The groupBy operator holds the references to these streams in memory opening a possible out of memory issue here.

As the first problem is simply related to the usage of groupBy we may say that we only have one problem: The usage of groupBy. ;-)
For a limited amount of data the proposed solution is perfectly fine so we will leave it as is for now.

Regarding the state of our service we have a working solution, so congratulations and let’s move on to the pure implementation.

Pure implementation

Like in the previous section I will spare you the details of the sbt setup. We will be using the following set of libraries:

1. http4s
2. Doobie (as database layer)
3. Flyway for database migrations (or evolutions)
4. Circe for JSON codecs
5. Refined for using refined types
6. the PostgreSQL JDBC driver
7. pureconfig (for proper configuration loading)

Pure configuration handling

Last time we simply loaded our configuration via the typesafe config library but can’t we do a bit better here? The answer is yes by using the pureconfig¹ library. First we start by implementing the necessary parts of our configuration as data types.

 1 final case class ApiConfig(host: NonEmptyString, port: PortNumber)
 2 
 3 object ApiConfig {
 4   implicit val configReader: ConfigReader[ApiConfig] =
 5     deriveReader[ApiConfig]
 6 }
 7 
 8 final case class DatabaseConfig(driver: NonEmptyString,
 9                                 url: DatabaseUrl,
10                                 user: DatabaseLogin,
11                                 pass: DatabasePassword)
12 
13 object DatabaseConfig {
14   implicit val configReader: ConfigReader[DatabaseConfig] =
15     deriveReader[DatabaseConfig]
16 }

As we can see the code is pretty simple. The implicits in the companion objects are needed for pureconfig to actually map from a configuration to your data types. As you can see we are using a function deriveReader which will derive (like in mathematics) the codec (Yes, it is similar to a JSON codec thus the name.) for us.

Below is an example of deriving a Order instance using the kittens ²library. It uses shapeless under the hood and provides automatic and semi automatic derivation for a lot of type class instances from Cats like Eq, Order, Show, Functor and so on.

1 import cats._
2 import cats.derived
3 
4 implicit val order: Order[Translation] = {
5   import derived.auto.order._
6   derived.semi.order[Translation]
7 }

Models

Because we have already written our models we just re-use them here. The only thing we change is the semi automatic derivation of the JSON codecs. We just need to import the appropriate circe package and call the derive functions.

1 import io.circe._
2 import io.circe.generic.semiauto._
3 
4 implicit val decode: Decoder[Product] = deriveDecoder[Product]
5 implicit val encode: Encoder[Product] = deriveEncoder[Product]
6 implicit val decode: Decoder[Translation] = deriveDecoder[Translation]
7 implicit val encode: Encoder[Translation] = deriveEncoder[Translation]

Database layer

In general the same applies to the database layer as we have already read in the “impure” section.

Migrations

For the sake of simplicity we will stick to Flyway for our database migrations. However we will wrap the migration code in a different way (read Encapsulate it properly within an IO to defer side effects.). While we’re at it we may just as well write our migration code using the interpreter pattern (it became famous under the name “tagless final” in Scala).

1 trait DatabaseMigrator[F[_]] {
2   def migrate(url: DatabaseUrl,
3               user: DatabaseLogin,
4               pass: DatabasePassword): F[Int]
5 }

We define a trait which describes the functionality desired by our interpreter and use a higher kinded type parameter to be able to abstract over the type. But now let’s continue with our Flyway interpreter.

 1 final class FlywayDatabaseMigrator extends DatabaseMigrator[IO] {
 2   override def migrate(url: DatabaseUrl,
 3                        user: DatabaseLogin,
 4                        pass: DatabasePassword): IO[Int] =
 5     IO {
 6       val flyway: Flyway = Flyway.configure()
 7         .dataSource(url, user, pass)
 8         .load()
 9       flyway.migrate()
10     }
11 }

As we can see, the implementation is pretty simple and we just wrap our code into an IO monad to constrain the effect. Having the migration code settled we can move on to the repository.

If we take a closer look at the method definition of Flyway.migrate, we see this:

1 public int migrate() throws FlywayException

While IO will gladly defer side effects for us it won’t stop enclosed code from throwing exceptions. This is not that great. So what can we do about it?
Having an instance of MonadError in scope we could just use the .attempt function provided by it. But is this enough or better does this provide a sensible solution for us? Let’s play a bit on the REPL.

 1 @ import cats._, cats.effect._, cats.implicits._
 2 @ val program = for {
 3            _ <- IO(println("one"))
 4            _ <- IO(println("two"))
 5            x <- IO.pure(42)
 6            } yield x
 7 @ program.attempt.unsafeRunSync match {
 8            case Left(e) =>
 9              println(e.getMessage)
10              -1
11            case Right(r) => r
12            }
13 one
14 two
15 res3: Int = 42
16 @ val program = for {
17            _ <- IO(println("one"))
18            _ <- IO(throw new Error("BOOM!"))
19            x <- IO.pure(42)
20            } yield x
21 @ program.attempt.unsafeRunSync match {
22            case Left(e) =>
23              println(e.getMessage)
24              -1
25            case Right(r) => r
26            }
27 one
28 BOOM!
29 res5: Int = -1

This looks like we just have to use MonadError then. Hurray, we don’t need to change our code in the migrator. As model citizens of the functional programming camp we just defer the responsibility upwards to the calling site.

Doobie

As we already started with using a tagless final approach we might as well continue with it and define a base for our repository.

1 trait Repository[F[_]] {
2   def loadProduct(id: ProductId): F[Seq[(ProductId, LanguageCode, ProductName)]]
3 
4   def loadProducts(): Stream[F, (ProductId, LanguageCode, ProductName)]
5 
6   def saveProduct(p: Product): F[Int]
7 
8   def updateProduct(p: Product): F[Int]
9 }

There is nothing exciting here except that we feel brave now and try to use proper refined types in our database functions. This is possible due to the usage of the doobie-refined module. To be able to map the UUID data type (and others) we also need to include the doobie-postgresql module. For convenience we are still using ProductId instead of UUID in our definition. In addition we wire the return type of loadProducts to be a fs2.Stream because we want to achieve pure functional streaming here. :-)
So let’s see what a repository using doobie looks like.

 1 final class DoobieRepository[F[_]: Sync](tx: Transactor[F])
 2   extends Repository[F] {
 3 
 4   override def loadProduct(id: ProductId) = ???
 5 
 6   override def loadProducts() = ???
 7 
 8   override def saveProduct(p: Product) = ???
 9 
10   override def updateProduct(p: Product) = ???
11 }

We keep our higher kinded type as abstract as we can but we want it to be able to suspend our side effects. Therefore we require an implicit Sync.³
If we look at the detailed function definitions further below, the first big difference is that with doobie you write plain SQL queries. You can do this with Slick too⁴ but with doobie it is the only way. If you’re used to object relational mapping (ORM) or other forms of query compilers then this may seem strange at first. But: “In data processing it seems, all roads eventually lead back to SQL!”⁵ ;-)
We won’t discuss the benefits or drawbacks here but in general I also lean towards the approach of using the de facto lingua franca for database access because it was made for this and so far no query compiler was able to beat hand crafted SQL in terms of performance. Another benefit is that if you ask a database guru for help, she will be much more able to help you with plain SQL queries than with some meta query which is compiled into something that you have no idea of.

1 override def loadProduct(id: ProductId) = 
2   sql"""SELECT products.id, names.lang_code, names.name 
3         FROM products
4         JOIN names ON products.id = names.product_id
5         WHERE products.id = $id"""
6     .query[(ProductId, LanguageCode, ProductName)]
7     .to[Seq]
8     .transact(tx)

The loadProduct function simply returns all rows for a single product from the database like its Slick counterpart in the impure variant. The parameter will be correctly interpolated by Doobie therefore we don’t need to worry about SQL injections here. We specify the type of the query, instruct Doobie to transform it into a sequence and give it to the transactor.

1 override def loadProducts() =
2   sql"""SELECT products.id, names.lang_code, names.name
3       FROM products
4       JOIN names ON products.id = names.product_id
5       ORDER BY products.id"""
6     .query[(ProductId, LanguageCode, ProductName)]
7     .stream
8     .transact(tx)

Our loadProducts function is equivalent to the first one but it returns the data for all products sorted by product and as a stream using the fs2 library which provides pure functional streaming.

 1 override def saveProduct(p: Product): F[Int] = {
 2   val namesSql = 
 3     "INSERT INTO names (product_id, lang_code, name) VALUES (?, ?, ?)"
 4   val namesValues = p.names.map(t => (p.id, t.lang, t.name))
 5   val program = for {
 6     pi <- sql"INSERT INTO products (id) VALUES(${p.id})".update.run
 7     ni <- Update[(ProductId, LanguageCode, ProductName)](namesSql)
 8             .updateMany(namesValues)
 9   } yield pi + ni
10   program.transact(tx)
11 }

When saving a product we use monadic notation for our program to have it short circuit in the case of failure. Doobie will also put all commands into a database transaction. The function itself will try to create the “master” entry into the products table and save all translations afterwards.

 1 override def updateProduct(p: Product): F[Int] = {
 2   val namesSql =
 3     "INSERT INTO names (product_id, lang_code, name) VALUES (?, ?, ?)"
 4   val namesValues = p.names.map(t => (p.id, t.lang, t.name))
 5   val program = for {
 6     dl <- sql"DELETE FROM names WHERE product_id = ${p.id}".update.run
 7     ts <- Update[(ProductId, LanguageCode, ProductName)](namesSql)
 8             .updateMany(namesValues)
 9   } yield dl + ts
10   program.transact(tx)
11 }

The updateProduct function uses also monadic notation like the saveProduct function we talked about before. The difference is that it first deletes all known translations before saving the given ones.

http4s routes

The routing DSL of http4s differs from the one of Akka-HTTP. Although I like the latter one more it poses no problem to model out a base for our routes.

 1 val productRoutes: HttpRoutes[IO] = HttpRoutes.of[IO] {
 2   case GET -> Root / "product" / id =>
 3     ???
 4   case PUT -> Root / "product" / id =>
 5     ???
 6 }
 7 val productsRoutes: HttpRoutes[IO] = HttpRoutes.of[IO] {
 8   case GET -> Root / "products" =>
 9     ???
10   case POST -> Root / "products" =>
11     ???
12 }

As we can see the DSL is closer to Scala syntax and quite easy to read. But before we move on to the details of each route let’s think about how we can model this a bit more abstract. While it is fine to have our routes bound to IO it would be better to have more flexibility here. We have several options here but for starters we just extract our routes into their own classes like in the following schema.

 1 final class ProductRoutes[F[_]: Sync](repo: Repository[F])
 2   extends Http4sDsl[F] {
 3 
 4   val routes: HttpRoutes[F] = HttpRoutes.of[F] {
 5     case GET -> Root / "product" / UUIDVar(id) =>
 6       ???
 7     case req @ PUT -> Root / "product" / UUIDVar(id) =>
 8       ???
 9   }
10 }
11 
12 final class ProductsRoutes[F[_]: Sync](repo: Repository[F]) 
13   extends Http4sDsl[F] {
14 
15   val routes: HttpRoutes[F] = HttpRoutes.of[F] {
16     case GET -> Root / "products" =>
17       ???
18     case req @ POST -> Root / "products" =>
19       ???
20   }
21 }

So far they only need the repository to access and manipulate data. Now let’s take on the single route implementations.

 1 final class ProductRoutes[F[_]: Sync](repo: Repository[F])
 2   extends Http4sDsl[F] {
 3   implicit def decodeProduct = jsonOf
 4   implicit def encodeProduct = jsonEncoderOf
 5 
 6   val routes: HttpRoutes[F] = HttpRoutes.of[F] {
 7     case GET -> Root / "product" / UUIDVar(id) =>
 8       for {
 9         rows <- repo.loadProduct(id)
10         resp <- Ok(Product.fromDatabase(rows))
11       } yield resp
12     case req @ PUT -> Root / "product" / UUIDVar(id) =>
13       for {
14         p <- req.as[Product]
15         _ <- repo.updateProduct(p)
16         r <- NoContent()
17       } yield r
18   }
19 }

First we need to bring JSON codecs in scope for http4s thus the implicit definitions on top of the file. In the route for loading a single product we simply load the database rows which we pipe through our helper function to construct a proper Product and return that.
The update route (via PUT) transforms the request body into a Product and gives that to the update function of the repository. Finally a NoContent response is returned.

 1 final class ProductsRoutes[F[_]: Sync](repo: Repository[F])
 2   extends Http4sDsl[F] {
 3   implicit def decodeProduct = jsonOf
 4   implicit def encodeProduct = jsonEncoderOf
 5 
 6   val routes: HttpRoutes[F] = HttpRoutes.of[F] {
 7     case GET -> Root / "products" =>
 8       val ps: Stream[F, Product] = repo.loadProducts
 9         .map(cs => Product.fromDatabase(List(cs)))
10         .collect {
11           case Some(p) => p
12         }
13       Ok(ps)
14     case req @ POST -> Root / "products" =>
15       for {
16         p <- req.as[Product]
17         _ <- repo.saveProduct(p)
18         r <- NoContent()
19       } yield r
20   }
21 }

Our first take on the routes for products looks pretty complete already. Again we need implicit definitions for our JSON codecs to be able to serialize and de-serialize our entities. The POST route for creating a product is basically the same as the update route from the previous part. We create a Product from the request body, pass it to the save function of the repository and return a 205 NoContent response.
The GET route for returning all products calls the appropriate repository function which returns a stream which we map over using our helper function. Afterwards we use collect to convert our stream from Option[Product] to a stream of Product which we pass to the Ok function of http4s.

To solve this we need to dive into the fs2 API and leverage it’s power to merge our products back together. So let’s see how we do.

Streaming - Take 1

Because we believe ourselves to be clever we pick the simple sledge hammer approach and just run some accumulator on the stream. So what do we need? A helper function and some code changes on the stream (e.g. in the route).

1 def merge(ps: List[Product])(p: Product): List[Product] =
2   ps.headOption.fold(List(p)) { h =>
3     if (h.id === p.id)
4       h.copy(names = h.names ::: p.names) :: ps.drop(1)
5     else
6       p :: ps
7   }

So this function will take a list (that may be empty) and a product and will merge the top most element (the head) of the list with the given one. It will return an updated list that either contains an updated head element or a new head. Leaving aside the question of who guarantees that the relevant list element will always be the head, we may use it.

1 case GET -> Root / "products" =>
2   val ps: Stream[F, Product] = repo.loadProducts
3     .map(cs => Product.fromDatabase(List(cs)))
4     .collect {
5       case Some(p) => p
6     }
7     .fold(List.empty[Product])((acc, p) => Product.merge(acc)(p))
8   Ok(ps)

Looks so simple, does it? Just a simple fold which uses our accumulator and we should be settled. But life is not that simple…

1 found   : fs2.Stream[F,List[com.wegtam.books.pfhais.pure.models.Product]]
2 required: fs2.Stream[F,com.wegtam.books.pfhais.pure.models.Product]
3          .fold(List.empty[Product])((acc, p) => Product.merge(acc)(p))
4               ^

The compiler complains that we have changed the type of the stream and rightly so. So let’s fix that compiler error.

Let’s take a look again and think about what it means to change a stream of products into a stream of a list of products. It means that we will be building the whole thing in memory! Well if we wanted that we could have skipped streaming at all. So back to the drawing board.

Streaming - Take 2

We need to process our stream of database columns (or products if we use the converter like before) in such a way that all related entities will be grouped into one product and emitted as such. After browsing the documentation of fs2 we stumble upon a function called groupAdjacentBy so we try that one.

 1 case GET -> Root / "products" =>
 2   val ps = repo.loadProducts
 3     .groupAdjacentBy(_._1)
 4     .map {
 5       case (id, rows) => Product.fromDatabase(rows.toList)
 6     }
 7     .collect {
 8       case Some(p) => p
 9     }
10   Ok(ps)

Okay, this does not look complicated and it even compiles - Hooray! :-)
So let’s break it apart piece by piece. The group function of fs2 will partition the input depending on the given function into chunks. A Chunk is used internally by fs2 for all kinds of stuff. You may compare it to a sub-stream of Akka-Streams. However the documentation labels it as: Strict, finite sequence of values that allows index-based random access of elements.
Having our chunks we can map over each one converting it into a list which is then passed to our helper function fromDatabase to create proper products. Last but not least we need to collect our entities to get from an Option[Product] to a stream of Product.

JSON trouble

Now that we have a proper streaming solution we try it out but what do we get when we expect a list of products?

 1 % http :53248/products
 2 HTTP/1.1 200 OK
 3 Content-Type: application/json
 4 Transfer-Encoding: chunked
 5 
 6 {
 7   "id":"8773899b-fcfa-401f-af3e-b188ebb0c00c",
 8   "names":[
 9     {"lang":"de","name":"Erdbeere"},
10     {"lang":"en","name":"Strawberry"}
11   ]
12 }
13 {
14   "id":"983aaf86-abe4-44af-9896-d8f2d2c5f82c",
15   "names":[
16     {"lang":"de","name":"Gurke"},
17     {"lang":"en","name":"Cucumber"}
18   ]
19 }

Well, whatever this is, it is not JSON! It might look like it, but it isn’t. However quite often you can see such things in the wild (read in production).

If we think about it then this sounds like a bug in http4s and indeed we find an issue⁷ for it. Because the underlying problem is not as trivial as it first sounds maybe we should try to work around the issue.
The fs2 API offers concatenation of streams and the nifty intersperse function to insert elements between emitted ones. So let’s give it a try.

 1 case GET -> Root / "products" =>
 2   val prefix = Stream.eval("[".pure[F])
 3   val suffix = Stream.eval("]".pure[F])
 4   val ps = repo.loadProducts
 5     .groupAdjacentBy(_._1)
 6     .map {
 7       case (id, rows) => Product.fromDatabase(rows.toList)
 8     }
 9     .collect {
10       case Some(p) => p
11     }
12     .map(_.asJson.noSpaces)
13     .intersperse(",")
14   @SuppressWarnings(Array("org.wartremover.warts.Any"))
15   val result: Stream[F, String] = prefix ++ ps ++ suffix
16   Ok(result)

First we create streams for the first and last JSON that we need to emit. Please note that we cannot simply use a String here but have to lift it into our HKT F. The usage of pure is okay because we simply lift a fixed value. Then we extend our original stream processing by explicitly converting our products to JSON and inserting the delimiter (a comma) manually using the intersperse function. In the end we simply concatenate our streams and return the result.
Our solution is quite simple, having the downside that we need to suppress a warning from the wartremover⁸ tool. This is somewhat annoying but can happen. If we remove the annotation, we’ll get a compiler error:

1 [error] ... [wartremover:Any] Inferred type containing Any
2 [error] val result: Stream[F, String] = prefix ++ ps ++ suffix
3 [error]                                        ^
4 [error] ... [wartremover:Any] Inferred type containing Any
5 [error] val result: Stream[F, String] = prefix ++ ps ++ suffix
6 [error]                                              ^
7 [error] two errors found

So let’s check if we have succeeded:

 1 % http :53248/products
 2 HTTP/1.1 200 OK
 3 Content-Type: text/plain; charset=UTF-8
 4 Transfer-Encoding: chunked
 5 
 6 [
 7     {
 8         "id": "8773899b-fcfa-401f-af3e-b188ebb0c00c", 
 9         "names": [
10             {
11                 "lang": "de",
12                 "name": "Erdbeere"
13             },
14             {
15                 "lang": "en",
16                 "name": "Strawberry"
17             }
18         ]
19     },
20     {
21         "id": "983aaf86-abe4-44af-9896-d8f2d2c5f82c", 
22         "names": [
23             {
24                 "lang": "de",
25                 "name": "Gurke"
26             },
27             {
28                 "lang": "en",
29                 "name": "Cucumber"
30             }
31         ]
32     }
33 ]

This looks good, so we congratulations: We are done with our routes!

Starting the application

Within our main entry point we simply initialise all needed components and wire them together. We’ll step through each part in this section. The first thing you’ll notice is that we use the IOApp provided by the Cats effect library⁹.

1 object Pure extends IOApp {
2   @SuppressWarnings(Array("org.wartremover.warts.Any"))
3   def run(args: List[String]): IO[ExitCode] = ???
4 }

Yet again we need to suppress a warning from wartremover here. But let’s continue to initialising the database connection.

 1 val migrator: DatabaseMigrator[IO] = new FlywayDatabaseMigrator
 2 
 3 val program = for {
 4   (apiConfig, dbConfig) <- IO {
 5     val cfg = ConfigFactory.load
 6     (loadConfigOrThrow[ApiConfig](cfg, "api"),
 7      loadConfigOrThrow[DatabaseConfig](cfg, "database"))
 8   }
 9   ms <- migrator.migrate(dbConfig.url, dbConfig.user, dbConfig.pass)
10   tx = Transactor
11     .fromDriverManager[IO](dbConfig.driver,
12       dbConfig.url,
13       dbConfig.user,
14       dbConfig.pass)
15   repo = new DoobieRepository(tx)

We create our database migrator explicitly wired to the IO data type. Now we start with a for comprehension in which we load our configuration via pureconfig yet again within an IO. After successful loading of the configuration we continue with migrating the database. Finally we create the transactor needed by Doobie and the database repository.

 1 val program = for {
 2   // ...
 3   productRoutes  = new ProductRoutes(repo)
 4   productsRoutes = new ProductsRoutes(repo)
 5   routes         = productRoutes.routes <+> productsRoutes.routes
 6   httpApp        = Router("/" -> routes).orNotFound
 7   server         = BlazeServerBuilder[IO].bindHttp(apiConfig.port,
 8                      apiConfig.host).withHttpApp(httpApp)
 9   fiber          = server.resource.use(_ => IO(StdIn.readLine())).as(ExitCode.Succes\
10 s)
11 } yield fiber

Here we create our routes via the classes, combine them (via <+> operator) and create the http4s app explicitly using an IO thus wiring our abstract routes to IO. The service will - like the impure one - run until you press enter. But it won’t run yet. ;-)

 1 program.attempt.unsafeRunSync match {
 2   case Left(e) =>
 3     IO {
 4       println("*** An error occured! ***")
 5       if (e != null) {
 6         println(e.getMessage)
 7       }
 8       ExitCode.Error
 9     }
10   case Right(r) => r
11 }

If you remember playing around with MonadError then you’ll recognize the attempt here. We attempt to run our program and execute possible side effects via the unsafeRunSync method from Cats effect. But to provide a proper return type for the IOApp we need to evaluate the return value which is either an error or a proper exit code. In case of an error we print it out on the console (no fancy logging here) and explicitly set an error code as the return value.

As it seems we are done with our pure service! Or are we? Let’s see what we need add to both services if we want to test them.

What about tests?

In the domain of strong static typing (not necessarily functional) you might hear phrases like “It compiles therefore it must be correct thus we don’t need tests!”. While there is a point that certain kinds of tests can be omitted in favour of strong static typing such stances overlook that even a correctly typed program may produce the wrong output. The other extreme (coming from dynamic typed land) is to substitute typing with testing - which is even worse. Remember that testing is usually a probabilistic approach and cannot guarantee the absence of bugs. If you have ever refactored a large code base in both paradigms then you’ll very likely come to esteem a good type system.

However, we need tests, so let’s write some. But before let us think a bit about what kinds of tests we need. :-)
Our service must read and create data in the JSON format. This format should be fixed and changes to it should raise some red flag because: Hey, we just broke our API! Furthermore we want to unleash the power of ScalaCheck¹ to benefit from property based testing. But even when we’re not using that we can still use it to generate test data for us.
Besides the regular unit tests there should be integration tests if a service is written. We can test a lot of things on the unit test side but in the end the integration of all our moving parts is what matters and often (usually on the not so pure side of things) you would have to trick (read mock) a lot to test things in isolation.

Testing the impure service

We will start with writing some data generators using the ScalaCheck library.

Generators

ScalaCheck already provides several generators for primitives but for our data models we have to do some more plumbing. Let’s start with generating a language code.

1 val genLanguageCode: Gen[LanguageCode] = Gen.oneOf(LanguageCodes.all)

Using the Gen.oneOf helper from the library the code becomes dead simple. Generating a UUID is nothing special either.

1 val genUuid: Gen[UUID] = Gen.delay(UUID.randomUUID)
2 )

You might be tempted to use Gen.const here but please don’t because that one will be memorized and thus never change. Another option is using a list of randomly generated UUID values from which we then chose one. That would be sufficient for generators which only generate a single product but if we want to generate lists of them we would have duplicate ids sooner than later.

1 val DefaultProductName: ProductName = "I am a product name!"
2 val genProductName: Gen[ProductName] = for {
3   cs <- Gen.nonEmptyListOf(Gen.alphaNumChar)
4   name = RefType.applyRef[ProductName](cs.mkString)
5            .getOrElse(DefaultProductName)
6 } yield name

So what do we have here? We want to generate a non empty string (because that is a requirement for our ProductName) but we also want to return a properly typed entity. First we let ScalaCheck generate a non empty list of random characters which we give to a utility function of refined. However we need a fallback value in case the validation done by refined fails. Therefore we defined a general default product name beforehand.

1 val genProductName: Gen[ProductName] =
2   Gen.nonEmptyListOf(Gen.alphaNumChar).map(cs => 
3     Refined.unsafeApply(cs.mkString)
4   )

Now that we have generators for language codes and product names we can write a generator for our Translation type.

 1 val genTranslation: Gen[Translation] = for {
 2   c <- genLanguageCode
 3   n <- genProductName
 4 } yield
 5   Translation(
 6     lang = c,
 7     name = n
 8   )
 9 
10 implicit val arbitraryTranslation: Arbitrary[Translation] =
11   Arbitrary(genTranslation)

As we can see the code is also quite simple. Additionally we create an implicit arbitrary value which will be used automatically by the forAll test helper if it is in scope. To be able to generate a Product we will need to provide a non empty list of translations.

1 val genTranslationList: Gen[List[Translation]] = for {
2   ts <- Gen.nonEmptyListOf(genTranslation)
3 } yield ts
4 
5 val genNonEmptyTranslationList: Gen[NonEmptyList[Translation]] = for {
6   t  <- genTranslation
7   ts <- genTranslationList
8   ns = NonEmptyList.fromList(ts)
9 } yield ns.getOrElse(NonEmptyList.of(t))

The first generator will create a non empty list of translations but will be typed as a simple List. Therefore we create a second generator which uses the fromList helper of the non empty list from Cats. Because that helper returns an Option (read is a safe function) we need to fallback to using a simple of function at the end.
With all these in place we can finally create our Product instances.

 1 val genProduct: Gen[Product] = for {
 2   id <- genProductId
 3   ts <- genNonEmptyTranslationList
 4 } yield
 5   Product(
 6     id = id,
 7     names = ts
 8   )
 9 
10 implicit val arbitraryProduct: Arbitrary[Product] =
11   Arbitrary(genProduct)

The code is basically the same as for Translation - the arbitrary implicit included.

Unit Tests

To avoid repeating the construction of our unit test classes we will implement a base class for tests which is quite simple.

1 abstract class BaseSpec extends WordSpec 
2   with MustMatchers with ScalaCheckPropertyChecks {}

Feel free to use other test styles - after all ScalaTest offers a lot² of them. I tend to lean towards the more verbose ones like WordSpec. Maybe that is because I spent a lot of time with RSpec³ in the Ruby world. ;-)

1 import com.wegtam.books.pfhais.impure.models.TypeGenerators._
2 
3 forAll("input") { p: Product =>
4   val rows = p.names.map(t => (p.id, t.lang.value, t.name.value)).toList
5   Product.fromDatabase(rows) must contain(p)
6 }

The code above is a very simple test of our helper function fromDatabase which works in the following way:

1. The forAll will generate a lot of Product entities using the generator.
2. From each entity a list of “rows” is constructed like they would appear in the database.
3. These constructed rows are given to the fromDatabase function.
4. The returned Option must then contain the generated value.

Because we construct the input for the function from a valid generated instance the function must always return a valid output.

Now let’s continue with testing our JSON codec for Product.

Testing a JSON codec

We need our JSON codec to provide several guarantees:

1. It must fail to decode invalid JSON input format (read garbage).
2. It must fail to decode valid JSON input format with invalid data (read wrong semantics).
3. It must succeed to decode completely valid input.
4. It must encode JSON which contains all fields included in the model.
5. It must be able to decode JSON that itself encoded.

The first one is pretty simple and to be honest: You don’t have to write a test for this because that should be guaranteed by the Circe library. Things look a bit different for very simple JSON representations though (read when encoding to numbers or strings).
I’ve seen people arguing about point 5 and there may be applications for it but implementing encoders and decoders in a non-reversible way will make your life way more complicated.

1 forAll("input") { s: String =>
2   decode[Product](s).isLeft must be(true)
3 }

There is not much to say about the test above: It will generate a lot of random strings which will be passed to the decoder which must fail.

1 forAll("id", "names") { (id: String, ns: List[String]) =>
2   val json = """{
3     |"id":""" + id.asJson.noSpaces + """,
4     |"names":""" + ns.asJson.noSpaces + """
5     |}""".stripMargin
6   decode[Product](json).isLeft must be(true)
7 }

This test will generate random instances for id which are all wrong because it must be a UUID and not a string. Also the instances for names will mostly (but maybe not always) be wrong because there might be empty strings or an even empty list. So the decoder is given a valid JSON format but invalid values, therefore it must fail.

 1 forAll("input") { i: Product =>
 2   val json = s"""{
 3     |"id": ${i.id.asJson.noSpaces},
 4     |"names": ${i.names.asJson.noSpaces}
 5     |}""".stripMargin
 6   withClue(s"Unable to decode JSON: $json") {
 7     decode[Product](json) match {
 8       case Left(e)  => fail(e.getMessage)
 9       case Right(v) => v must be(i)
10     }
11   }
12 }

In this case we manually construct a valid JSON input using values from a generated valid Product entity. This is passed to the decoder and the decoder must not only succeed but return an instance equal to the generated one.

1 forAll("input") { i: Product =>
2   val json = i.asJson.noSpaces
3   json must include(s""""id":${i.id.asJson.noSpaces}""")
4   json must include(s""""names":${i.names.asJson.noSpaces}""")
5 }

The test will generate again a lot of entities and we construct a JSON string from each. We then expect the string to include several field names and their correctly encoded values. You might ask why we do not check for more things like: Are these fields the only ones within the JSON string? Well, this would be more cumbersome to test and a JSON containing more fields than we specify won’t matter for the decoder because it will just ignore them.

1 forAll("input") { p: Product =>
2   decode[Product](p.asJson.noSpaces) match {
3     case Left(_)  => fail("Must be able to decode encoded JSON!")
4     case Right(d) => withClue("Must decode the same product!")(d must be(p))
5   }
6 }

Here we encode a generated entity and pass it to the encoder which must return the same entity.

More tests

So this is basically what we do for models. Because we have more than one we will have to write tests for each of the others. I will spare you the JSON tests for Translation but that one also has a helper function called fromUnsafe so let’s take a look at the function.

1 def fromUnsafe(lang: String)(name: String): Option[Translation] =
2   for {
3     l <- RefType.applyRef[LanguageCode](lang).toOption
4     n <- RefType.applyRef[ProductName](name).toOption
5   } yield Translation(lang = l, name = n)

This function simply tries to create a valid Translation entity from unsafe input values using the helpers provided by refined. As we can see it is a total function (read is safe to use). To cover all corner cases we must test it with safe and unsafe input.

 1 forAll("lang", "name") { (l: String, n: String) =>
 2   whenever(
 3     RefType
 4       .applyRef[LanguageCode](l)
 5       .toOption
 6       .isEmpty || RefType.applyRef[ProductName](n).toOption.isEmpty
 7   ) {
 8     Translation.fromUnsafe(l)(n) must be(empty)
 9   }
10 }

Here we generate two random strings which we explicitly check to be invalid using the whenever helper. Finally the function must return an empty Option e.g. None for such values.

1 forAll("input") { t: Translation =>
2   Translation.fromUnsafe(t.lang.value)(t.name.value) must contain(t)
3 }

The test for valid input is very simple because we simply use the values from our automatically generated valid instances. :-)

So far we have no tests for our Repository class which handles all the database work. Neither have we tests for our routes. We have several options for testing here but before can test either of them we have do to some refactoring. For starters we should move our routes out of our main application into separate classes to be able to test them more easily.

Yes, we should. There are of course limits and pros and cons to that but in general this makes sense. Also this has nothing to do with being “impure” or “pure” but with clean structure.

Some refactoring

Moving the routes into separate classes poses no big problem we simply create a ProductRoutes and a ProductsRoutes class which will hold the appropriate routes. As a result our somewhat messy main application code becomes more readable.

 1 def main(args: Array[String]): Unit = {
 2   implicit val system: ActorSystem    = ActorSystem()
 3   implicit val mat: ActorMaterializer = ActorMaterializer()
 4   implicit val ec: ExecutionContext   = system.dispatcher
 5 
 6   val url = ???
 7   val user           = ???
 8   val pass           = ???
 9   val flyway: Flyway = ???
10   val _              = flyway.migrate()
11 
12   val dbConfig: DatabaseConfig[JdbcProfile] =
13     DatabaseConfig.forConfig("database", system.settings.config)
14   val repo = new Repository(dbConfig)
15 
16   val productRoutes  = new ProductRoutes(repo)
17   val productsRoutes = new ProductsRoutes(repo)
18   val routes         = productRoutes.routes ~ productsRoutes.routes
19 
20   val host       = system.settings.config.getString("api.host")
21   val port       = system.settings.config.getInt("api.port")
22   val srv        = Http().bindAndHandle(routes, host, port)
23   val pressEnter = StdIn.readLine()
24   srv.flatMap(_.unbind()).onComplete(_ => system.terminate())
25 }

We simply create our instances from our routing classes and construct our global routes directly from them. This is good but if we want to test the routes in isolation we still have the problem that they are hard-wired to our Repository class which is implemented via Slick. Several options exist to handle this:

1. Use an in-memory test database with according configuration.
2. Abstract further and use a trait instead of the concrete repository implementation.
3. Write integration tests which will require a working database.

Using option 1 is tempting but think about it some more. While the benefit is that we can use our actual implementation and just have to fire up an in-memory database (for example h2), there are also some drawbacks:

1. You have to handle evolutions for the in-memory database.
2. Your evolutions have to be completely portable SQL (read ANSI SQL). Otherwise you’ll have to write each of your evolutions scripts two times (one for production, one for testing).
3. Your code has to be database agnostic. This sounds easier than it is. Even the tools you’re using may use database specific features under the hood.
4. Several features are simply not implemented in some databases. Think of things like cascading deletion via foreign keys.

Taking option 2 is a valid choice but it will result in more code. Also you must pay close attention to the “test repository” implementation to avoid introducing bugs there. Going for the most simple approach is usually feasible. Think of a simple test repository implementation that will just return hard coded values or values passed to it via constructor.

However we will go with option 3 in this case. It has the drawback that you’ll have to provide a real database environment (and maybe more) for testing. But it is as close to production as you can get. Also you will need these either way to test your actual repository implementation, so let’s get going.

Integration Tests

First we need to configure our test database because we do not want to accidentally wipe a production database. For our case we leave everything as is and just change the database name.

 1 api {
 2   host = "localhost"
 3   port = 49152
 4 }
 5 
 6 database {
 7   profile = "slick.jdbc.PostgresProfile$"
 8   db {
 9     connectionPool = "HikariCP"
10     dataSourceClass = "org.postgresql.ds.PGSimpleDataSource"
11     properties {
12       serverName = "localhost"
13       portNumber = "5432"
14       databaseName = "impure_test"
15       user = "impure"
16       password = "secret"
17     }
18     numThreads = 10
19   }
20 }

Another thing we should do is provide a test configuration for our logging framework. We use the logback library and Slick will produce a lot of logging output on the DEBUG level so we should fix that. It is nice to have logging if you need it but it also clutters up your log files. We create a file logback-test.xml in the directory src/it/resources which should look like this:

 1 <?xml version="1.0" encoding="UTF-8"?>
 2 <configuration debug="false">
 3   <appender name="console" class="ch.qos.logback.core.ConsoleAppender">
 4     <filter class="ch.qos.logback.classic.filter.ThresholdFilter">
 5       <level>WARN</level>
 6     </filter>
 7     <encoder>
 8       <pattern>%date %highlight(%-5level) %cyan(%logger{0}) - %msg%n</pattern>
 9     </encoder>
10   </appender>
11 
12   <appender name="async-console" class="ch.qos.logback.classic.AsyncAppender">
13     <appender-ref ref="console"/>
14     <queueSize>5000</queueSize>
15     <discardingThreshold>0</discardingThreshold>
16   </appender>
17 
18   <logger name="com.wegtam.books.pfhais.impure" level="INFO" additivity="false">
19     <appender-ref ref="console"/>
20   </logger>
21 
22   <root>
23     <appender-ref ref="console"/>
24   </root>
25 </configuration>

Due to the nature of integration tests we want to use production or “production like” settings and environment. But really starting our application or service for each test will be quite cumbersome so we should provide a base class for our tests. In this class we will start our service, migrate our database and provide an opportunity to shut it down properly after testing.

Because we do not want to get into trouble when running an Akka-HTTP on the same port, we first create a helper function which will determine a free port number.

1 import java.net.ServerSocket
2 
3 def findAvailablePort(): Int = {
4   val serverSocket = new ServerSocket(0)
5   val freePort     = serverSocket.getLocalPort
6   serverSocket.setReuseAddress(true)
7   serverSocket.close()
8   freePort
9 }

The code is quite simple and very useful for such cases. Please note that it is important to use setReuseAddress because otherwise the found socket will be blocked for a certain amount of time. But now let us continue with our base test class.

 1 abstract class BaseSpec
 2     extends TestKit(
 3       ActorSystem(
 4         "it-test",
 5         ConfigFactory
 6           .parseString(s"api.port=${BaseSpec.findAvailablePort()}")
 7           .withFallback(ConfigFactory.load())
 8       )
 9     )
10     with AsyncWordSpecLike
11     with MustMatchers
12     with ScalaCheckPropertyChecks
13     with BeforeAndAfterAll
14     with BeforeAndAfterEach {
15 
16   implicit val materializer: ActorMaterializer = ActorMaterializer()
17 
18   private val url  = ???
19   private val user = ???
20   private val pass = ???
21   protected val flyway: Flyway = 
22     Flyway.configure().dataSource(url, user, pass).load()
23 
24   override protected def afterAll(): Unit =
25     TestKit.shutdownActorSystem(system, FiniteDuration(5, SECONDS))
26 
27   override protected def beforeAll(): Unit = {
28     val _ = flyway.migrate()
29   }
30 }

As you can see we are using the Akka-Testkit to initialise an actor system. This is useful because there are several helpers available which you might need. We configure the actor system with our free port using the loaded configuration as fallback. Next we globally create an actor materializer which is needed by Akka-HTTP and Akka-Streams. Also we create a globally available Flyway instance to make cleaning and migrating the database easier.
The base class also implements the beforeAll and afterAll methods which will be run before and after all tests. They are used to initially migrate the database and to shut down the actor system properly in the end.

Testing the repository

Now that we our parts in place we can write an integration test for our repository implementation.

First we need to do some things globally for the test scope.

 1 private val dbConfig: DatabaseConfig[JdbcProfile] =
 2   DatabaseConfig.forConfig("database", system.settings.config)
 3 private val repo = new Repository(dbConfig)
 4 
 5 override protected def beforeEach(): Unit = {
 6   flyway.clean()
 7   val _ = flyway.migrate()
 8   super.beforeEach()
 9 }
10 
11 override protected def afterEach(): Unit = {
12   flyway.clean()
13   super.afterEach()
14 }
15 
16 override protected def afterAll(): Unit = {
17   repo.close()
18   super.afterAll()
19 }

We create one Repository instance for all our tests here. The downside is that if one test crashes it then the other will be affected too. On the other hand we avoid running into database connection limits and severe code limbo to ensure closing a repository connection after each test no matter the result.
Also we clean and migrate before each test and clean also after each test. This ensures having a clean environment.

Onwards to the test for loading a single product.

 1 "#loadProduct" when {
 2   "the ID does not exist" must {
 3     "return an empty list of rows" in {
 4       val id = UUID.randomUUID
 5       for {
 6         rows <- repo.loadProduct(id)
 7       } yield {
 8         rows must be(empty)
 9       }
10     }
11   }
12 
13   "the ID exists" must {
14     "return a list with all product rows" in {
15       genProduct.sample match {
16         case None => fail("Could not generate data sample!")
17         case Some(p) =>
18           for {
19             _    <- repo.saveProduct(p)
20             rows <- repo.loadProduct(p.id)
21           } yield {
22             Product.fromDatabase(rows) match {
23               case None => fail("No product created from database rows!")
24               case Some(c) =>
25                 c.id must be(p.id)
26                 c mustEqual p
27             }
28           }
29       }
30     }
31   }
32 }

Loading a non existing product must not produce any result and is simple to test if our database is empty. Testing the loading of a real product is not that much more complicated. We use the ScalaCheck generators to create one, save it and load it again. The loaded product must of course be equal to the saved one.

 1 "#loadProducts" when {
 2   "no products exist" must {
 3     "return an empty stream" in {
 4       val src = Source.fromPublisher(repo.loadProducts())
 5       for {
 6         ps <- src.runWith(Sink.seq)
 7       } yield {
 8         ps must be(empty)
 9       }
10     }
11   }
12 
13   "some products exist" must {
14     "return a stream with all product rows" in {
15       genProducts.sample match {
16         case None => fail("Could not generate data sample!")
17         case Some(ps) =>
18           val expected = ps.flatMap(
19           p => p.names.toNonEmptyList.toList.map(
20             n => (p.id, n.lang, n.name)
21             )
22           )
23           for {
24             _ <- Future.sequence(ps.map(p => repo.saveProduct(p)))
25             src = Source
26               .fromPublisher(repo.loadProducts())
27               // more code omitted here
28             rows <- src.runWith(Sink.seq)
29           } yield {
30             rows must not be (empty)
31             rows.size mustEqual ps.size
32             rows.toList.sorted mustEqual ps.sorted
33           }
34       }
35     }
36   }
37 }

Testing the loading of all products if none exit is trivial like the one for a non existing single product. For the case of multiple products we generate a list of them which we save. Afterwards we load them and use the same transformation logic like in the routes to be able to construct proper Product instances. One thing you might notice is the explicit sorting which is due to the fact that we want to ensure that our product lists are both sorted before comparing them.

 1 "#saveProduct" when {
 2   "the product does not already exist" must {
 3     "save the product to the database" in {
 4       genProduct.sample match {
 5         case None => fail("Could not generate data sample!")
 6         case Some(p) =>
 7           for {
 8             cnts <- repo.saveProduct(p)
 9             rows <- repo.loadProduct(p.id)
10           } yield {
11             withClue("Data missing from database!")(
12               cnts.fold(0)(_ + _) must be(p.names.toNonEmptyList.size + 1))
13             Product.fromDatabase(rows) match {
14               case None => fail("No product created from database rows!")
15               case Some(c) =>
16                 c.id must be(p.id)
17                 c mustEqual p
18             }
19           }
20       }
21     }
22   }
23 
24   "the product does already exist" must {
25     "return an error and not change the database" in {
26       (genProduct.sample, genProduct.sample) match {
27         case (Some(a), Some(b)) =>
28           val p = b.copy(id = a.id)
29           for {
30             cnts <- repo.saveProduct(a)
31             nosv <- repo.saveProduct(p).recover {
32               case _ => 0
33             }
34             rows <- repo.loadProduct(a.id)
35           } yield {
36             withClue("Saving a duplicate product must fail!")(nosv must be(0))
37             Product.fromDatabase(rows) match {
38               case None => fail("No product created from database rows!")
39               case Some(c) =>
40                 c.id must be(a.id)
41                 c mustEqual a
42             }
43           }
44         case _ => fail("Could not create data sample!")
45       }
46     }
47   }
48 }

Here we test the saving which in the first case should simply write the appropriate data into the database. If the product already exists however this should not happen. Our database constraints will ensure that this does not happen (or so we hope ;-)). Slick will throw an exception which we catch in the test code using the recover method from Future to return a zero indicating no affected database rows. In the end we test for this zero and also check if the originally saved product has not been changed.

 1 "#updateProduct" when {
 2   "the product does exist" must {
 3     "update the database" in {
 4       (genProduct.sample, genProduct.sample) match {
 5         case (Some(a), Some(b)) =>
 6           val p = b.copy(id = a.id)
 7           for {
 8             cnts <- repo.saveProduct(a)
 9             upds <- repo.updateProduct(p)
10             rows <- repo.loadProduct(a.id)
11           } yield {
12             withClue("Already existing product was not created!")(
13               cnts.fold(0)(_ + _) must be(a.names.toNonEmptyList.size + 1)
14             )
15             Product.fromDatabase(rows) match {
16               case None => fail("No product created from database rows!")
17               case Some(c) =>
18                 c.id must be(a.id)
19                 c mustEqual p
20             }
21           }
22         case _ => fail("Could not create data sample!")
23       }
24     }
25   }
26 
27   "the product does not exist" must {
28     "return an error and not change the database" in {
29       genProduct.sample match {
30         case None => fail("Could not generate data sample!")
31         case Some(p) =>
32           for {
33             nosv <- repo.updateProduct(p).recover {
34               case _ => 0
35             }
36             rows <- repo.loadProduct(p.id)
37           } yield {
38             withClue("Updating a not existing product must fail!")
39               (nosv must be(0))
40             withClue("Product must not exist in database!")
41               (rows must be(empty))
42           }
43       }
44     }
45   }
46 }

For testing an update we generate two samples, save one to the database, change the id of the other to the one from the first and execute an update. This update should proceed without problems and the data in the database must have been changed correctly.
If the product does not exist then we use the same recover technique like in the saveProduct test.

Congratulations, we have made a check mark on our first integration test using a real database using randomly generated data!

Testing the routes

Regarding the route testing there several options as always. For this one we will define some use cases and then develop some more helper code which will allow us to fire up our routes and do real HTTP requests and then check the database and the responses. We build upon our BaseSpec class and call it BaseUseCaseSpec. In it we will do some more things like define a global base URL which can be used from the tests to make correct requests. Additionally we will write a small actor which simply starts an Akka-HTTP server.

 1 final class BaseUseCaseActor(repo: Repository, mat: ActorMaterializer)
 2   extends Actor with ActorLogging {
 3   import context.dispatcher
 4 
 5   implicit val system: ActorSystem             = context.system
 6   implicit val materializer: ActorMaterializer = mat
 7 
 8   override def receive: Receive = {
 9     case BaseUseCaseActorCmds.Start =>
10       val productRoutes  = new ProductRoutes(repo)
11       val productsRoutes = new ProductsRoutes(repo)
12       val routes         = productRoutes.routes ~ productsRoutes.routes
13       val host           = context.system.settings.config.getString("api.host")
14       val port           = context.system.settings.config.getInt("api.port")
15       val _              = Http().bindAndHandle(routes, host, port)
16     case BaseUseCaseActorCmds.Stop =>
17       context.stop(self)
18   }
19 }
20 
21 object BaseUseCaseActor {
22   def props(repo: Repository, mat: ActorMaterializer): Props =
23     Props(new BaseUseCaseActor(repo, mat))
24 
25   sealed trait BaseUseCaseActorCmds
26 
27   object BaseUseCaseActorCmds {
28     case object Start extends BaseUseCaseActorCmds
29     case object Stop extends BaseUseCaseActorCmds
30   }
31 }

As you can see, the actor is quite simple. Upon receiving the Start command it will initialise the routes and start an Akka-HTTP server. To be able to do this it needs the database repository and an actor materializer which are passed via the constructor. Regarding the BaseUseCaseSpec we will concentrate on the code that differs from the base class.

 1 abstract class BaseUseCaseSpec
 2   extends TestKit(
 3     ActorSystem(
 4       "it-test",
 5       ConfigFactory
 6         .parseString(s"api.port=${BaseUseCaseSpec.findAvailablePort()}")
 7         .withFallback(ConfigFactory.load())
 8     )
 9   )
10   with AsyncWordSpecLike
11   with MustMatchers
12   with ScalaCheckPropertyChecks
13   with BeforeAndAfterAll
14   with BeforeAndAfterEach {
15   // ...
16   final val baseUrl: String = s"""http://${system.settings.config
17     .getString("api.host")}:${system.settings.config
18     .getInt("api.port")}"""
19 
20   protected val dbConfig: DatabaseConfig[JdbcProfile] =
21     DatabaseConfig.forConfig("database", system.settings.config)
22   protected val repo = new Repository(dbConfig)
23 
24   override protected def beforeAll(): Unit = {
25     val _ = flyway.migrate()
26     val a = system.actorOf(BaseUseCaseActor.props(repo, materializer))
27     a ! BaseUseCaseActorCmds.Start
28   }
29   // ...
30 }