Scala3 migration (#196)

* Upgrade scala 3

docker-compose.yml

@@ -1,10 +1,10 @@
 version: '2'
 services:
-  mongo:
+  nio_mongo:
     image: mongo:3.4.3
     ports:
       - 27017:27017
-  zookeeper:
+  nio_zookeeper:
     image: confluentinc/cp-zookeeper:5.2.3
     ports:
      - 32182:32181
@@ -14,12 +14,12 @@ services:
     extra_hosts:
       - "moby:127.0.0.1"
       - "localhost: 127.0.0.1"
-  kafka:
+  nio_kafka:
     image: confluentinc/cp-kafka:5.2.3
     ports:
      - 29092:29092
     depends_on:
-      - zookeeper
+      - nio_zookeeper
     environment:
       KAFKA_BROKER_ID: 1
       KAFKA_ZOOKEEPER_CONNECT: zookeeper:32181
@@ -30,10 +30,26 @@ services:
     extra_hosts:
       - "moby:127.0.0.1"
       - "localhost: 127.0.0.1"
-  s3server:
+  nio_s3server:
     image: scality/s3server
     ports:
       - 8000:8000
     environment:
       - "SCALITY_ACCESS_KEY_ID=newAccessKey"
-      - "SCALITY_SECRET_ACCESS_KEY=newSecretKey"
+      - "SCALITY_SECRET_ACCESS_KEY=newSecretKey"
+  nio_akhq:
+    image: tchiotludo/akhq
+    ports:
+      - 9005:8080
+    environment:
+      AKHQ_CONFIGURATION: |
+        akhq:
+          connections:
+            docker-kafka-server:
+              properties:
+                bootstrap.servers: "nio_kafka:9092"
+    depends_on:
+      - nio_kafka
+    extra_hosts:
+      - "moby:127.0.0.1"
+      - "localhost: 127.0.0.1"

nio-provider/app/configuration/Configuration.scala

@@ -1,18 +1,22 @@
 package configuration
 
 import play.api.Configuration
-import pureconfig._
+import pureconfig.*
+import pureconfig.generic.derivation.default.*
+import pureconfig.generic.semiauto.*
 import pureconfig.generic.ProductHint
 
 import scala.concurrent.duration.FiniteDuration
 
 object NioConfiguration {
-  import pureconfig.generic.auto._
-  implicit def hint[T]: ProductHint[T] =
-    ProductHint[T](ConfigFieldMapping(CamelCase, CamelCase))
 
-  def apply(config: Configuration): NioConfiguration =
+  implicit def hint[T]: ProductHint[T] = ProductHint[T](ConfigFieldMapping(CamelCase, CamelCase))
+
+  def apply(config: Configuration): NioConfiguration = {
+    given ConfigReader[NioConfiguration] = deriveReader
+    given ConfigWriter[NioConfiguration] = deriveWriter
     ConfigSource.fromConfig(config.underlying).at("nio").loadOrThrow[NioConfiguration]
+  }
 }
 
 case class NioConfiguration(websocketHost: String, filter: Otoroshi, kafka: KafkaConfig, nio: NioConfig)

nio-provider/app/controllers/HomeController.scala

@@ -4,7 +4,7 @@ import org.apache.pekko.actor.ActorSystem
 import auth.AuthActionWithEmail
 import configuration.Env
 import messaging.KafkaMessageBroker
-import play.api.mvc.{AbstractController, ControllerComponents}
+import play.api.mvc.{AbstractController, Action, AnyContent, ControllerComponents}
 
 import scala.concurrent.ExecutionContext
 
@@ -16,13 +16,13 @@ class HomeController(val AuthAction: AuthActionWithEmail,
                      implicit val ec: ExecutionContext)
     extends AbstractController(cc) {
 
-  def index() = AuthAction { implicit req =>
+  def index(): Action[AnyContent] = AuthAction { implicit req =>
     Ok(views.html.index(env, req.email, env.config.websocketHost))
   }
 
-  def indexOther() = index()
+  def indexOther(): Action[AnyContent] = index()
 
-  def otherRoutes(route: String) = AuthAction { implicit req =>
+  def otherRoutes(route: String): Action[AnyContent] = AuthAction { implicit req =>
     Ok(views.html.index(env, req.email, env.config.websocketHost))
   }
 

nio-provider/app/controllers/UserDataController.scala

@@ -35,7 +35,7 @@ class UserDataController(
   def uploadFile(tenant: String, orgKey: String, userId: String, name: String): Action[MultipartFormData[Files.TemporaryFile]] =
     AuthAction.async(parse.multipartFormData) { implicit req =>
       NioLogger.info(s"upload file $name")
-      val src: Source[ByteString, _] =
+      val src: Source[ByteString, ?] =
         StreamConverters.fromInputStream { () =>
           new FileInputStream(req.body.files.head.ref)
         }

nio-provider/app/models/UserExtractTask.scala

@@ -39,18 +39,18 @@ object UserExtractTask {
       } and
       (__ \ "uploadStartedAt").readNullable[LocalDateTime] and
       (__ \ "endedAt").readNullable[LocalDateTime]
-  )(UserExtractTask.apply _)
+  )(UserExtractTask.apply)
 
   implicit val userExtractTaskWrites: Writes[UserExtractTask] = (
-    (JsPath \ "_id").write[String] and
+    (JsPath \   "_id").write[String] and
       (JsPath \ "tenant").write[String] and
       (JsPath \ "orgKey").write[String] and
       (JsPath \ "userId").write[String] and
       (JsPath \ "email").write[String] and
       (JsPath \ "startedAt").write[LocalDateTime] and
       (JsPath \ "uploadStartedAt").writeNullable[LocalDateTime] and
       (JsPath \ "endedAt").writeNullable[LocalDateTime]
-  )(unlift(UserExtractTask.unapply))
+  )(u => (u._id, u.tenant, u.orgKey, u.userId, u.email, u.startedAt, u.uploadStartedAt, u.endedAt))
 
   implicit val userExtractTaskOWrites: OWrites[UserExtractTask] = (
     (JsPath \ "_id").write[String] and
@@ -61,7 +61,7 @@ object UserExtractTask {
       (JsPath \ "startedAt").write[LocalDateTime] and
       (JsPath \ "uploadStartedAt").writeNullable[LocalDateTime] and
       (JsPath \ "endedAt").writeNullable[LocalDateTime]
-  )(unlift(UserExtractTask.unapply))
+  )(u => (u._id, u.tenant, u.orgKey, u.userId, u.email, u.startedAt, u.uploadStartedAt, u.endedAt))
 
   implicit val format: Format[UserExtractTask]   =
     Format(userExtractTaskReads, userExtractTaskWrites)

nio-provider/app/models/events.scala

@@ -7,11 +7,13 @@ import utils.DateUtils
 import scala.collection.Seq
 
 object EventType extends Enumeration {
-  type WeekDay = Value
+  type EventType = Value
   val UserExtractTaskAsked, UserExtractTaskCompleted, Unknown = Value
 
   def from(name: String): Value =
     values.find(_.toString.toLowerCase == name.toLowerCase()).getOrElse(Unknown)
+
+  implicit val format: Format[EventType] = Json.formatEnum(this)
 }
 
 object NioEvent {

nio-provider/app/service/NioService.scala

@@ -24,8 +24,9 @@ class NioService(env: Env, wSClient: WSClient)(
                       orgKey: String,
                       userId: String,
                       name: String,
-                      src: Source[ByteString, _],
+                      src: Source[ByteString, ?],
                       contentTypeHeader: Option[String]): Future[JsValue] = {
+    import play.api.libs.ws.bodyWritableOf_Multipart
     wSClient
       .url(
         s"${env.config.nio.url}/api/$tenant/organisations/$orgKey/users/$userId/_files/$name")

nio-provider/app/utils/Results.scala

@@ -43,7 +43,7 @@ object Result {
 
     def fromJsError(jsError: Seq[(JsPath, Seq[JsonValidationError])]): AppErrors = {
       val fieldErrors = jsError.map { case (k, v) =>
-        (k.toJsonString, v.map(err => ErrorMessage(err.message, err.args.map(_.toString): _*)).toList)
+        (k.toJsonString, v.map(err => ErrorMessage(err.message, err.args.map(_.toString)*)).toList)
       }.toMap
       AppErrors(fieldErrors = fieldErrors)
     }

nio-provider/build.sbt

@@ -25,7 +25,9 @@ libraryDependencies ++= Seq(
   "org.apache.pekko"         %% "pekko-connectors-kafka"        % pekkoKafka,
   "de.svenkubiak"             % "jBCrypt"            % "0.4.1", //  ISC/BSD
   "com.auth0"                 % "java-jwt"           % javaJwt, // MIT license
-  "com.github.pureconfig"    %% "pureconfig"         % pureConfig, // Apache 2.0
+//  "com.github.pureconfig"    %% "pureconfig"         % pureConfig, // Apache 2.0
+  "com.github.pureconfig"    %% "pureconfig-core"          % pureConfig, // Apache 2.0
+  "com.github.pureconfig"    %% "pureconfig-generic-scala3" % pureConfig, // Apache 2.0
   "org.scalactic"            %% "scalactic"          % scalaticVersion, // Apache 2.0
   "org.webjars"               % "swagger-ui"         % "3.12.1",
   "org.typelevel"            %% "cats-core"          % catsVersion, // MIT

nio-server/app/configuration/Configuration.scala

@@ -1,19 +1,13 @@
 package configuration
 
 import play.api.Configuration
-import pureconfig._
 import pureconfig.generic.ProductHint
-import pureconfig.generic.auto._
+import pureconfig.*
+import pureconfig.generic.derivation.default.*
+import pureconfig.generic.semiauto.*
 
 import scala.concurrent.duration.FiniteDuration
 
-object NioConfiguration {
-
-  implicit def hint[T]: ProductHint[T] = ProductHint[T](ConfigFieldMapping(CamelCase, CamelCase))
-
-  def apply(config: Configuration): NioConfiguration =
-    ConfigSource.fromConfig(config.underlying).at("nio").loadOrThrow[NioConfiguration]
-}
 
 case class NioConfiguration(
     baseUrl: String,
@@ -89,23 +83,9 @@ case class KafkaConfig(
 
 case class Location(location: Option[String])
 
-object TenantConfiguration {
-  implicit def hint[T]: ProductHint[T] =
-    ProductHint[T](ConfigFieldMapping(CamelCase, CamelCase))
-
-  def apply(config: Configuration): TenantConfiguration =
-    ConfigSource.fromConfig(config.underlying).at("tenant").loadOrThrow[TenantConfiguration]
-}
 
 case class TenantConfiguration(admin: AdminConfig)
 
-object HealthCheckConfiguration {
-  implicit def hint[T]: ProductHint[T] =
-    ProductHint[T](ConfigFieldMapping(CamelCase, CamelCase))
-
-  def apply(config: Configuration): HealthCheckConfiguration =
-    ConfigSource.fromConfig(config.underlying).at("healthcheck").loadOrThrow[HealthCheckConfiguration]
-}
 
 case class HealthCheckConfiguration(secret: String, header: String)
 
@@ -127,3 +107,36 @@ case class MailGunConfig(apiKey: String, endpoint: String, from: String)
 case class CatchUpEventsConfig(strategy: String, delay: FiniteDuration, interval: FiniteDuration)
 
 case class Db(batchSize: Int)
+
+object NioConfiguration {
+
+  implicit def hint[T]: ProductHint[T] = ProductHint[T](ConfigFieldMapping(CamelCase, CamelCase))
+
+  def apply(config: Configuration): NioConfiguration = {
+    given ConfigReader[NioConfiguration] = deriveReader
+    given ConfigWriter[NioConfiguration] = deriveWriter
+    ConfigSource.fromConfig(config.underlying).at("nio").loadOrThrow[NioConfiguration]
+  }
+}
+
+
+object TenantConfiguration {
+  implicit def hint[T]: ProductHint[T] = ProductHint[T](ConfigFieldMapping(CamelCase, CamelCase))
+
+  def apply(config: Configuration): TenantConfiguration = {
+    given ConfigReader[TenantConfiguration] = deriveReader
+    given ConfigWriter[TenantConfiguration] = deriveWriter
+    ConfigSource.fromConfig(config.underlying).at("tenant").loadOrThrow[TenantConfiguration]
+  }
+}
+
+
+object HealthCheckConfiguration {
+  implicit def hint[T]: ProductHint[T] = ProductHint[T](ConfigFieldMapping(CamelCase, CamelCase))
+
+  def apply(config: Configuration): HealthCheckConfiguration = {
+    given ConfigReader[HealthCheckConfiguration] = deriveReader
+    given ConfigWriter[HealthCheckConfiguration] = deriveWriter
+    ConfigSource.fromConfig(config.underlying).at("healthcheck").loadOrThrow[HealthCheckConfiguration]
+  }
+}

nio-server/app/controllers/Auth0Controller.scala

@@ -91,7 +91,8 @@ class Auth0Controller(env: Env, wsClient: WSClient, cc: ControllerComponents)(im
     }
   }
 
-  private def getToken(code: String, sessionId: String): Future[Either[AppErrors, (String, String)]] =
+  private def getToken(code: String, sessionId: String): Future[Either[AppErrors, (String, String)]] = {
+    import play.api.libs.ws.writeableOf_JsValue
     wsClient
       .url(s"https://${auth0Config.domain}/oauth/token")
       .addHttpHeaders(HeaderNames.ACCEPT -> MimeTypes.JSON)
@@ -118,6 +119,7 @@ class Auth0Controller(env: Env, wsClient: WSClient, cc: ControllerComponents)(im
             Left(AppErrors(Seq(ErrorMessage("tokens.not.send"))))
         }
       }
+  }
 
   private def getUser(accessToken: String): Future[JsValue] =
     wsClient

nio-server/app/controllers/ConsentController.scala

@@ -273,7 +273,7 @@ class ConsentController(
 
   val newLineSplit: Flow[ByteString, ByteString, NotUsed] = Framing.delimiter(ByteString("\n"), 10000, allowTruncation = true)
   val toJson: Flow[ByteString, JsValue, NotUsed] = Flow[ByteString] via newLineSplit map (_.utf8String) filterNot (_.isEmpty) map (l => Json.parse(l))
-  private def ndJson(implicit ec: ExecutionContext): BodyParser[Source[JsValue, _]] = BodyParser(_ => Accumulator.source[ByteString].map(s => Right(s.via(toJson)))(ec))
+  private def ndJson(implicit ec: ExecutionContext): BodyParser[Source[JsValue, ?]] = BodyParser(_ => Accumulator.source[ByteString].map(s => Right(s.via(toJson)))(ec))
 
   private object ImportError {
     implicit val format: OFormat[ImportError] = Json.format[ImportError]
@@ -314,7 +314,7 @@ class ConsentController(
     }
 
 
-  def batchImport(tenant: String, orgKey: String): Action[Source[JsValue, _]] = AuthAction.async(ndJson) { implicit req =>
+  def batchImport(tenant: String, orgKey: String): Action[Source[JsValue, ?]] = AuthAction.async(ndJson) { implicit req =>
     val result: Future[JsValue] = req.body
       .map(json => ((json \ "userId").validate[String].getOrElse(""), json))
       .via(sharding(10, Flow[(String, JsValue)].mapAsync(1) { case (_, json) =>
@@ -326,14 +326,14 @@ class ConsentController(
           }
         )
       }))
-      .fold(ImportResult()){ (acc, elt) => acc combine elt }
+      .fold(ImportResult()){ (acc, elt) => acc.combine(elt) }
       .map { importResult => Json.toJson(importResult) }
       .runWith(Sink.head)
 
     result.map { json => Ok(json) }
   }
 
-  private def handleImportPatch(tenant: String, orgKey: String, req: SecuredAuthContext[Source[JsValue, _]], json: JsValue, userId: String, patchCommand: PartialConsentFact): Future[ImportResult] = {
+  private def handleImportPatch(tenant: String, orgKey: String, req: SecuredAuthContext[Source[JsValue, ?]], json: JsValue, userId: String, patchCommand: PartialConsentFact): Future[ImportResult] = {
     (for {
       _ <- if (patchCommand.userId.isDefined && !patchCommand.userId.contains(userId)) IO.error(ImportResult.error("error.userId.is.immutable", command = json))
       else IO.succeed(patchCommand)
@@ -370,7 +370,7 @@ class ConsentController(
     } yield result).merge
   }
 
-  private def handleImportUpdate(tenant: String, orgKey: String, req: SecuredAuthContext[Source[JsValue, _]], json: JsValue, userId: String, consentFact: ConsentFact): Future[ImportResult] = {
+  private def handleImportUpdate(tenant: String, orgKey: String, req: SecuredAuthContext[Source[JsValue, ?]], json: JsValue, userId: String, consentFact: ConsentFact): Future[ImportResult] = {
     if (consentFact.userId != userId) {
       NioLogger.error(s"error.userId.is.immutable : userId in path $userId // userId on body ${consentFact.userId}")
 

nio-server/app/controllers/ExtractionController.scala

@@ -45,7 +45,7 @@ class ExtractionController(
 
   implicit val mat: Materializer = Materializer(system)
 
-  val fileBodyParser: BodyParser[Source[ByteString, _]]            = BodyParser { _ =>
+  val fileBodyParser: BodyParser[Source[ByteString, ?]]            = BodyParser { _ =>
     Accumulator.source[ByteString].map(s => Right(s))
   }
   implicit val readable: ReadableEntity[AppIds]                    = AppIds
@@ -166,7 +166,7 @@ class ExtractionController(
     }
 
   private def upload(tenant: String, task: ExtractionTask, appId: String, name: String)(implicit
-      req: ReqWithExtractionTask[Source[ByteString, _]]
+      req: ReqWithExtractionTask[Source[ByteString, ?]]
   ): Future[String] =
     req.body
       .via(
@@ -184,7 +184,7 @@ class ExtractionController(
       }
 
   private def oldUpload(tenant: String, task: ExtractionTask, appId: String, name: String)(implicit
-      req: ReqWithExtractionTask[Source[ByteString, _]]
+      req: ReqWithExtractionTask[Source[ByteString, ?]]
   ): Future[String] = {
     val uploadKey =
       s"$tenant/${task.orgKey}/${task.userId}/${task._id}/$appId/$name"