Added Artemis to compose.yaml

.bsp/sbt.json

@@ -1 +0,0 @@
-{"name":"sbt","version":"1.12.13","bspVersion":"2.1.0-M1","languages":["scala"],"argv":["/usr/lib/jvm/java-21-openjdk-amd64/bin/java","-Xms100m","-Xmx100m","-classpath","/home/artemiy/.local/share/JetBrains/IntelliJIdea2026.1/Scala/launcher/sbt-launch.jar","-Dsbt.script=/home/artemiy/.local/bin/sbt","xsbt.boot.Boot","-bsp"]}

.gitignore

@@ -8,3 +8,5 @@ hs_err_pid*
 /project/project/target/
 /project/target
 /target
+
+.bsp

app/actors/auctions/AuctionPublisher.scala

@@ -0,0 +1,103 @@
+package actors.auctions
+
+import com.google.inject.Provides
+import com.hivemq.client.mqtt.datatypes.MqttQos
+import com.hivemq.client.mqtt.mqtt5.Mqtt5AsyncClient
+import com.hivemq.client.mqtt.mqtt5.message.publish.Mqtt5Publish
+import dao.AuctionsDAO
+import models.{Auction, AuctionStatus, User}
+import mqttClient.MqttClientFactory
+import org.apache.pekko.actor.typed.{ActorRef, Behavior}
+import org.apache.pekko.actor.typed.scaladsl.Behaviors
+import play.api.{Configuration, Logging}
+import play.api.libs.concurrent.ActorModule
+import play.api.libs.json.{JsBoolean, JsObject, JsValue, Json}
+
+import java.nio.charset.StandardCharsets
+import javax.inject.Singleton
+import scala.jdk.OptionConverters.RichOptional
+import scala.util.Random
+
+object AuctionPublisher extends ActorModule with Logging {
+  sealed trait Command
+  private case object Ready extends Command
+  case class Publish(auction: Auction, author: User) extends Command
+  private case class AuctionConfirmed(id: Int, responseTopic: String) extends Command
+
+  private final val topic = "auctions/created"
+
+  override type Message = Command
+
+  @Provides
+  @Singleton
+  def apply(configuration: Configuration, auctionsDAO: AuctionsDAO): Behavior[Message] = Behaviors.setup { context =>
+    val client = MqttClientFactory.createClient(configuration, "-auction-publisher")
+    client.connect().whenComplete { (_action, _throwable) =>
+      context.self ! Ready
+    }
+
+    idle(context.self, client, auctionsDAO)
+  }
+
+  private def idle(self: ActorRef[Message], client: Mqtt5AsyncClient, auctionsDAO: AuctionsDAO, waitingPublishes: Seq[Publish] = List()): Behaviors.Receive[Command] = Behaviors.receiveMessage[Command] {
+    case p: Publish =>
+      idle(self, client, auctionsDAO, waitingPublishes :+ p)
+    case Ready =>
+      for { publish <- waitingPublishes } publishAuction(client, publish.auction, publish.author, self)
+      ready(self, client, auctionsDAO)
+    case message @ AuctionConfirmed(_, _) =>
+      logger.warn(s"Received impossible message $message")
+      idle(self, client, auctionsDAO, waitingPublishes)
+  }
+
+  private def ready(self: ActorRef[Message], client: Mqtt5AsyncClient, auctionsDAO: AuctionsDAO): Behaviors.Receive[Command] = Behaviors.receiveMessage[Command] {
+    case Publish(auction, author) =>
+      publishAuction(client, auction, author, self)
+      ready(self, client, auctionsDAO)
+    case AuctionConfirmed(auctionId, responseTopic) =>
+      client.unsubscribeWith().topicFilter(responseTopic).send()
+      auctionsDAO.changeStatus(auctionId, AuctionStatus.Ready)
+      ready(self, client, auctionsDAO)
+    case Ready =>
+      logger.warn("Message Ready received after having already received it before")
+      ready(self, client, auctionsDAO)
+  }
+
+  private def publishAuction(client: Mqtt5AsyncClient, auction: Auction, author: User, self: ActorRef[Command]): Unit = {
+    val payload = Json.toBytes(auction_payload(auction, author))
+    val responseTopic = s"auction-validate/${auction.gid}-${Random.alphanumeric.take(8).mkString}"
+
+    client.subscribeWith()
+      .topicFilter(responseTopic)
+      .callback{ message =>
+        logger.info(s"Receiving message on ${message.getTopic}")
+        logger.info(message.getPayloadAsBytes.mkString("Array(", ", ", ")"))
+        message.getPayload.toScala
+          .map(StandardCharsets.UTF_8.decode(_).toString)
+          .map(Json.parse)
+          .flatMap(v => (v \ "ok").toOption)
+          .map{
+            case b: JsBoolean if b.value => self ! AuctionConfirmed(auction.id.get, responseTopic)
+              // TODO: add error handling
+          }
+      }
+      .send()
+
+    val publish = Mqtt5Publish.builder()
+      .topic(topic)
+      .payload(payload)
+      .responseTopic(responseTopic)
+      .qos(MqttQos.EXACTLY_ONCE)
+      .build()
+    logger.info(s"Publishing auction ${auction.id.getOrElse(-1)}")
+    client.publish(publish)
+  }
+
+  private def auction_payload(auction: Auction, author: User): JsObject = {
+    Json.obj(
+      "gid" -> auction.gid,
+      "author_gid" -> author.gid,
+      "starting_price" -> auction.startingPrice
+    )
+  }
+}

app/controllers/AuctionsController.scala

@@ -0,0 +1,47 @@
+package controllers
+
+import actors.auctions.AuctionPublisher
+import controllers.authentication.AuthenticatedAction
+import dao.{AuctionsDAO, AuthTokensDAO}
+import models.AuctionStatus.Ready
+import models.{Auction, User}
+import org.apache.pekko.actor.typed.ActorRef
+import play.api.libs.functional.syntax.toFunctionalBuilderOps
+import play.api.libs.json.*
+import play.api.mvc.{BaseController, ControllerComponents}
+
+import javax.inject.{Inject, Singleton}
+import scala.concurrent.{ExecutionContext, Future}
+
+@Singleton
+class AuctionsController @Inject()(val controllerComponents: ControllerComponents,
+                                   val auctionsDAO: AuctionsDAO,
+                                   val authTokensDAO: AuthTokensDAO,
+                                   authenticatedAction: AuthenticatedAction,
+                                   val auctionPublisher: ActorRef[AuctionPublisher.Command],
+                                   implicit val ec: ExecutionContext) extends BaseController {
+  def index = authenticatedAction.async { request =>
+    for {
+      auctions <- auctionsDAO.list_available(request.user)
+      json = Json.toJson(auctions)
+    } yield Ok(json)
+  }
+
+  def create = authenticatedAction.async(parse.json) { implicit request =>
+    val auctionCreateDataResult = request.body.validate[AuctionCreateData]
+    auctionCreateDataResult.fold(
+      errors => Future.successful(BadRequest(JsError.toJson(errors))),
+      auctionCreateData => {
+        for {
+          auction <- auctionsDAO.create(auctionCreateData.name, auctionCreateData.startingPrice, request.user)
+          json = Json.toJson(auction)
+          _ = auctionPublisher ! AuctionPublisher.Publish(auction, request.user)
+        } yield Created(json)
+      }
+    )
+  }
+
+  private case class AuctionCreateData(name: String, startingPrice: BigDecimal)
+
+  private given Reads[AuctionCreateData] = Json.reads[AuctionCreateData]
+}

app/controllers/authentication/AuthenticatedAction.scala

@@ -2,34 +2,39 @@ package controllers.authentication
 
 import dao.AuthTokensDAO
 import jakarta.inject.Inject
-import models.AuthToken
+import models.{AuthToken, User}
 import play.api.libs.json.*
 import play.api.mvc.Results.Unauthorized
-import play.api.mvc.{ActionRefiner, Request, Result, WrappedRequest}
+import play.api.mvc.{ActionBuilder, ActionRefiner, AnyContent, BodyParsers, Request, Result, WrappedRequest}
 
 import scala.concurrent.{ExecutionContext, Future}
 
-class AuthenticatedRequest[A](val token: AuthToken, request: Request[A]) extends WrappedRequest[A](request)
+class AuthenticatedRequest[A](val token: AuthToken, val user: User, request: Request[A]) extends WrappedRequest[A](request)
-
-class AuthenticatedAction @Inject()(authTokensDao: AuthTokensDAO)(implicit val executionContext: ExecutionContext) extends ActionRefiner[Request, AuthenticatedRequest] {
-  private final val tokenPrefix = "Token"
 
+class AuthenticatedAction @Inject()(val parser: BodyParsers.Default, authTokensDao: AuthTokensDAO)(implicit val executionContext: ExecutionContext)
+  extends ActionRefiner[Request, AuthenticatedRequest]
+  with ActionBuilder[AuthenticatedRequest, AnyContent]{
   override protected def refine[A](request: Request[A]): Future[Either[Result, AuthenticatedRequest[A]]] = {
-    val unauthenticatedRequestError = Left(Unauthorized(Json.obj("error" -> "Unauthenticated")))
     val authHeader = request.headers.get("Authorization")
-      .flatMap { header =>
+      .map(_.split(" ", 2)(1))
-        if header.startsWith(tokenPrefix) then Some(header.substring(tokenPrefix.length).trim)
-        else None
-      }
       .filter(_.nonEmpty)
 
     authHeader match {
       case Some(token) =>
-        authTokensDao.find(token).map {
+        authTokensDao.find(token).flatMap {
-          case Some(authToken) => Right(AuthenticatedRequest[A](authToken, request))
+          case Some(authToken) => findUser(request, authToken)
-          case None => unauthenticatedRequestError
+          case None => Future.successful(unauthenticatedRequestError)
         }
       case None => Future.successful(unauthenticatedRequestError)
     }
   }
+
+  private val unauthenticatedRequestError = Left(Unauthorized(Json.obj("error" -> "Unauthenticated")))
+
+  private def findUser[A](request: Request[A], token: AuthToken) = {
+    authTokensDao.getUser(token).map {
+      case Some(user) => Right(AuthenticatedRequest(token, user, request))
+      case None => unauthenticatedRequestError
+    }
+  }
 }

app/controllers/authentication/SessionsController.scala

@@ -5,46 +5,49 @@ import jakarta.inject.{Inject, Singleton}
 import models.AuthToken
 import play.api.libs.json.*
 import play.api.libs.functional.syntax.*
-import play.api.mvc.{AbstractController, Action, AnyContent, ControllerComponents}
+import play.api.mvc.{AbstractController, ControllerComponents}
 
 import scala.concurrent.{ExecutionContext, Future}
 
+object SessionsController {
   case class LoginData(username: String, password: String)
-implicit val loginDataReads: Reads[LoginData] = (
+  given Reads[LoginData] = Json.reads[LoginData]
-  (JsPath \ "username").read[String] and
+}
-    (JsPath \ "password").read[String]
-)(LoginData.apply)
-implicit val tokenWrites: Writes[AuthToken] = (token: AuthToken) => Json.obj(
-  "token" -> token.token,
-  "createdAt" -> token.createdAt,
-  "expiresAt" -> token.expiresAt
-)
 
 
 @Singleton
-class SessionsController @Inject()(controllerComponents: ControllerComponents,
+class SessionsController @Inject()(
-                                   usersDAO: UsersDAO, authTokensDAO: AuthTokensDAO,
+                                    controllerComponents: ControllerComponents,
-                                   authenticatedAction: AuthenticatedAction,
+                                    usersDAO: UsersDAO,
-                                   implicit val ec: ExecutionContext) extends AbstractController(controllerComponents) {
+                                    authTokensDAO: AuthTokensDAO,
+                                    authenticatedAction: AuthenticatedAction)
+                                  (implicit val ec: ExecutionContext)
+  extends AbstractController(controllerComponents) {
+  import SessionsController.{LoginData, given}
+
   def login() = Action.async(parse.json) { implicit request =>
     val loginDataResult = request.body.validate[LoginData]
     loginDataResult.fold(
-      errors => {
+      errors => Future.successful(BadRequest(JsError.toJson(errors))),
-        Future.successful(BadRequest(Json.obj("message" -> JsError.toJson(errors))))
-      },
       loginData => {
         usersDAO.authenticate(loginData.username, loginData.password).flatMap { userOption =>
           userOption.fold(
             Future.successful(UnprocessableEntity(Json.obj("message" -> "Invalid authentication")))
           ){ user =>
-            authTokensDAO.createToken(user).map(token => Ok(Json.toJson(token)))
+            authTokensDAO.createToken(user).map { token =>
+              val json = Json.toJsObject(token) ++ Json.obj(
+                "userId" -> user.id.get,
+                "userGid" -> user.gid
+              )
+              Ok(json)
+            }
           }
         }
       }
     )
   }
 
-  def logout = (Action andThen authenticatedAction).async { request =>
+  def logout = authenticatedAction.async { request =>
     authTokensDAO.destroy(request.token).map{ _ => Ok("") }
   }
 }

app/controllers/devel/DevActionsController.scala

@@ -0,0 +1,32 @@
+package controllers.devel
+
+import dao.UsersDAO
+import play.api.libs.functional.syntax.toFunctionalBuilderOps
+import play.api.libs.json.{JsError, JsPath, Json, Reads}
+import play.api.mvc.*
+
+import javax.inject.Inject
+import scala.concurrent.{ExecutionContext, Future}
+
+class DevActionsController @Inject()(val controllerComponents: ControllerComponents,
+                                     val usersDAO: UsersDAO)
+                                    (implicit val ec: ExecutionContext)
+  extends BaseController {
+
+  private case class UserData(username: String, password: String)
+  private implicit val userDataReads: Reads[UserData] = (
+    (JsPath \ "username").read[String] and (JsPath \ "password").read[String]
+  )(UserData.apply)
+
+  def createUser = Action.async(parse.json) { implicit request =>
+    request.body.validate[UserData].fold(
+      errors => Future.successful(BadRequest(JsError.toJson(errors))),
+      userData => usersDAO.create(userData.username, userData.password)
+        .map {
+          case Left(UsersDAO.UsernameTaken) => BadRequest(Json.obj("error" -> "Username exists"))
+          case Left(_ex) => BadRequest(Json.obj("error" -> "User creation error"))
+          case Right(user) => Ok(Json.toJson(user))
+        }
+    )
+  }
+}

app/controllers/devel/DevRouter.scala

@@ -0,0 +1,24 @@
+package controllers.devel
+
+import play.api.Environment
+import play.api.Mode.Dev
+import play.api.mvc.*
+import play.api.routing.Router.Routes
+import play.api.routing.SimpleRouter
+import play.api.routing.sird.{POST, UrlContext}
+
+import javax.inject.Inject
+
+class DevRouter @Inject()(environment: Environment, controller: DevActionsController) extends SimpleRouter {
+  override def routes: Routes = {
+    if (environment.mode == Dev) {
+      devRoutes
+    } else {
+      PartialFunction.empty
+    }
+  }
+
+  private val devRoutes: PartialFunction[RequestHeader, Handler] = {
+    case POST(p"/users") => controller.createUser()
+  }
+}

app/controllers/filters/LoggingFilter.scala

@@ -17,7 +17,7 @@ class LoggingFilter @Inject()(implicit val mat: Materializer, ec: ExecutionConte
       val endTime = System.currentTimeMillis()
       val requestTime = endTime - startTime
 
-      logger.info(s"${requestHeader.method} ${requestHeader.uri} ${result.header.status} $requestTime")
+      logger.info(s"${requestHeader.method} ${requestHeader.uri} ${result.header.status} ${requestTime}ms")
 
       result
     }

app/dao/AuctionsDAO.scala

@@ -0,0 +1,50 @@
+package dao
+
+import models.{Auction, AuctionStatus}
+import schemas.Auctions
+import schemas.given
+import play.api.db.slick.{DatabaseConfigProvider, HasDatabaseConfigProvider}
+import slick.jdbc.PostgresProfile
+import slick.jdbc.PostgresProfile.api.*
+
+import java.time.Instant
+import javax.inject.Inject
+import scala.concurrent.{ExecutionContext, Future}
+import scala.util.Random
+
+class AuctionsDAO @Inject()(protected val dbConfigProvider: DatabaseConfigProvider)(implicit val ec: ExecutionContext)
+  extends HasDatabaseConfigProvider[PostgresProfile] {
+
+  def list_available(user: models.User): Future[Seq[Auction]] = {
+    db.run {
+      Auctions.filter(row => (row.status === AuctionStatus.Ready) && (row.authorId =!= user.id.get)).result
+    }
+  }
+
+  def list_created_by(user: models.User): Future[Seq[Auction]] = {
+    db.run(Auctions.filter(_.authorId === user.id).result)
+  }
+
+  def create(name: String, startingPrice: BigDecimal, author: models.User): Future[Auction] = {
+    val auction = Auction(
+      id = None, name = name,
+      gid = Random.alphanumeric.take(24).mkString,
+      status = AuctionStatus.Created,
+      authorId = author.id.get,
+      startingPrice = startingPrice,
+      currentBid = None, currentBidderId = None, currentBidAt = None,
+      createdAt = Instant.now()
+    )
+
+    create(auction)
+  }
+
+  def create(auction: Auction) : Future[Auction] = {
+    db.run((Auctions returning Auctions) += auction)
+  }
+
+  def changeStatus(auctionId: Int, status: AuctionStatus): Future[Boolean] = {
+    db.run(Auctions.filter(_.id === auctionId).map(_.status).update(status))
+    .map(_ == 1)
+  }
+}

app/dao/AuthTokensDAO.scala

@@ -14,10 +14,9 @@ import models.AuthToken
 import schemas.AuthTokens
 
 import scala.language.postfixOps
+import scala.util.Random
 
 class AuthTokensDAO @Inject()(protected val dbConfigProvider: DatabaseConfigProvider)(implicit val ec: ExecutionContext) extends HasDatabaseConfigProvider[PostgresProfile] {
-  private lazy val random = SecureRandom()
-  private lazy val base64 = Base64.getEncoder
   private final lazy val tokenValidDuration = Duration.ofDays(30)
 
   def find(token: String): Future[Option[AuthToken]] = {
@@ -26,12 +25,10 @@ class AuthTokensDAO @Inject()(protected val dbConfigProvider: DatabaseConfigProv
   }
 
   def createToken(user: models.User): Future[AuthToken] = {
-    val tokenRaw = ByteBuffer.allocate(12)
-    random.nextBytes(tokenRaw.array())
     val token = AuthToken(
       id = None,
       userId = user.id.get,
-      token = base64.encodeToString(tokenRaw.array()),
+      token = Random.alphanumeric.take(12).mkString,
       createdAt = Instant.now,
       expiresAt = Instant.now.plus(tokenValidDuration)
     )
@@ -39,6 +36,11 @@ class AuthTokensDAO @Inject()(protected val dbConfigProvider: DatabaseConfigProv
     db.run((AuthTokens returning AuthTokens.map(_.id)) += token).map { tokenId => token.copy(id = Some(tokenId)) }
   }
 
+  def getUser(token: AuthToken): Future[Option[models.User]] = {
+    val query = schemas.Users.filter(_.id === token.userId).take(1).result.headOption
+    db.run(query)
+  }
+
   def destroy(token: AuthToken): Future[Unit] = {
     val q = AuthTokens.filter(_.token === token.token).delete
     db.run(q).map{ _ => () }

app/dao/UsersDAO.scala

@@ -1,5 +1,6 @@
 package dao
 
+import dao.UsersDAO.{UnexpectedError, UsernameTaken}
 import de.mkammerer.argon2.Argon2Factory
 import execution.CryptoExecutionContext
 import jakarta.inject.Inject
@@ -9,23 +10,70 @@ import slick.jdbc.PostgresProfile.api.*
 
 import scala.concurrent.{ExecutionContext, Future}
 import models.User
+import org.postgresql.util.{PSQLException, PSQLState}
 import schemas.Users
 
-class UsersDAO @Inject()(protected val dbConfigProvider: DatabaseConfigProvider, private val cryptoEC: CryptoExecutionContext)(implicit val ec: ExecutionContext)
+import java.nio.charset.StandardCharsets.UTF_8
+import scala.util.Random
+
+class UsersDAO @Inject()(protected val dbConfigProvider: DatabaseConfigProvider, private val cryptoEC: CryptoExecutionContext)
+                        (using ec: ExecutionContext)
   extends HasDatabaseConfigProvider[PostgresProfile] {
   private val argon2 = Argon2Factory.create()
 
+  private object ArgonParameters {
+    val iterations = 3
+    val parallelism = 1
+    val memory = 65536
+  }
+
+  private final val dummyDigest = "$argon2i$v=19$m=65536,t=3,p=1$EB9QibLwzqkO7ne2bJoU/Q$ADkdyweXKylJORvWXbBVekURYaazkW/OplTq2wdN1So"
+
   def authenticate(username: String, password: String): Future[Option[User]] = {
     val query = Users.filter(_.username === username).take(1).result.headOption
-
     val userFuture = db.run(query)
 
     userFuture.map { userOption =>
       for {
         user <- userOption
         digest <- user.passwordDigest
-        if argon2.verify(digest, password.getBytes("UTF-8"))
+      } yield (user, digest)
-      } yield user
+    }.map {
-    }(using cryptoEC)
+        case Some((user, digest)) => (Some(user), digest)
+        case None => (None, dummyDigest)
+    }.flatMap { (userOpt, digest) =>
+      for {
+        validHash <- Future(argon2.verify(digest, password.getBytes(UTF_8)))(using cryptoEC)
+        opt = userOpt.flatMap(u => if validHash then Some(u) else None)
+      } yield opt
     }
   }
+
+  def create(username: String, password: String): Future[Either[UsersDAO.RegistrationError, User]] = {
+    Future(argonHash(password))(using cryptoEC)
+      .flatMap{ passwordDigest =>
+        val gid = Random.alphanumeric.take(12).mkString
+        val user = User(
+          id = None,
+          gid = gid,
+          username = username,
+          passwordDigest = Some(passwordDigest)
+        )
+
+        db.run((Users returning Users) += user).map(Right(_))
+      }.recover {
+        case ex: PSQLException if ex.getSQLState == PSQLState.UNIQUE_VIOLATION.getState =>
+          Left(UsernameTaken)
+        case ex: Throwable =>
+          Left(UnexpectedError(ex))
+      }
+  }
+
+  private def argonHash(data: String) = argon2.hash(ArgonParameters.iterations, ArgonParameters.memory, ArgonParameters.parallelism, data.getBytes(UTF_8))
+}
+
+object UsersDAO {
+  sealed trait RegistrationError
+  case object UsernameTaken extends RegistrationError
+  case class UnexpectedError(cause: Throwable) extends RegistrationError
+}

app/models/models.scala

@@ -1,7 +1,46 @@
 package models
 
+import play.api.libs.json.{JsString, JsValue, Json, OWrites, Writes}
+
 import java.time.Instant
 
 case class User(id: Option[Int], username: String, gid: String, passwordDigest: Option[String])
+object User {
+  given writes: Writes[User] = new Writes[User] {
+    override def writes(o: User): JsValue = Json.obj(
+      "gid" -> o.gid,
+      "username" -> o.username
+    )
+  }
+}
 
 case class AuthToken(id: Option[Int], userId: Int, token: String, createdAt: Instant, expiresAt: Instant)
+object AuthToken {
+  given writes: OWrites[AuthToken] = (a: AuthToken) => Json.obj(
+    "token" -> a.token,
+    "userId" -> a.userId,
+    "createdAt" -> a.createdAt,
+    "expiresAt" -> a.expiresAt
+  )
+}
+
+enum AuctionStatus(val value: Short) {
+  case Created extends AuctionStatus(0)
+  case Ready extends AuctionStatus(10)
+  case Finished extends AuctionStatus(10000)
+}
+
+object AuctionStatus {
+  lazy val byValue: Map[Short, AuctionStatus] = Map.from(AuctionStatus.values.map{it => (it.value, it)})
+
+  implicit val writes: Writes[AuctionStatus] = (o: AuctionStatus) => JsString(o.toString)
+}
+
+case class Auction(id: Option[Int], name: String, status: AuctionStatus, gid: String, authorId: Int,
+                   startingPrice: BigDecimal, currentBid: Option[BigDecimal],
+                   currentBidderId: Option[Int], currentBidAt: Option[Instant],
+                   createdAt: Instant)
+
+object Auction {
+  implicit val writes: OWrites[Auction] = Json.writes[Auction]
+}

app/mqttClient/MessageProcessorActor.scala

@@ -0,0 +1,46 @@
+package mqttClient
+
+import com.google.inject.Provides
+import com.hivemq.client.mqtt.mqtt5.Mqtt5AsyncClient
+import com.hivemq.client.mqtt.mqtt5.message.publish.Mqtt5Publish
+import jakarta.inject.{Inject, Singleton}
+import org.apache.pekko.actor.typed.{ActorRef, Behavior}
+import org.apache.pekko.actor.typed.scaladsl.Behaviors
+import play.api.Logging
+import play.api.libs.concurrent.ActorModule
+
+import java.nio.charset.StandardCharsets
+
+object MessageProcessorActor extends ActorModule with Logging {
+  sealed trait Command
+  case object Stop extends Command
+  private case class MqttMessageReceived(publish: Mqtt5Publish) extends Command
+
+  override type Message = Command
+
+  @Singleton
+  class Reg @Inject()(private val mqttMessageProcessorActor: ActorRef[Command])
+
+  @Provides
+  def create(client: Mqtt5AsyncClient): Behavior[Command] = Behaviors.setup { context =>
+    client.connect().whenComplete { (_action, _throwable) =>
+      client.subscribeWith()
+        .topicFilter("test/+")
+        .callback{ (publish) =>
+          context.self ! MqttMessageReceived(publish)
+        }
+        .send()
+    }
+
+    Behaviors.receiveMessage[Command] {
+      case MqttMessageReceived(publish) =>
+        val topic = publish.getTopic
+        val payload = String(publish.getPayloadAsBytes, StandardCharsets.UTF_8)
+        logger.info(s"Received message $topic $payload")
+        Behaviors.same
+      case Stop => Behaviors.stopped{ () =>
+        client.disconnect()
+      }
+    }
+  }
+}

app/mqttClient/Module.scala

@@ -0,0 +1,18 @@
+package mqttClient
+
+import actors.auctions.AuctionPublisher
+import com.google.inject.{AbstractModule, Provides, Singleton}
+import com.hivemq.client.mqtt.mqtt5.{Mqtt5AsyncClient, Mqtt5Client}
+import play.api.libs.concurrent.PekkoGuiceSupport
+import play.api.{Configuration, Logging}
+
+class Module extends AbstractModule with PekkoGuiceSupport with Logging {
+  override def configure(): Unit = {
+    bindTypedActor(MessageProcessorActor, "mqtt-message-processor-actor")
+    bind(classOf[MessageProcessorActor.Reg]).asEagerSingleton()
+    bindTypedActor(AuctionPublisher, "auction-publisher") // TODO move somewhere else
+  }
+
+  @Provides @Singleton
+  def mqtt5Client(config: Configuration): Mqtt5AsyncClient = MqttClientFactory.createClient(config)
+}

app/mqttClient/MqttClientFactory.scala

@@ -0,0 +1,27 @@
+package mqttClient
+
+import com.hivemq.client.mqtt.mqtt5.{Mqtt5AsyncClient, Mqtt5Client}
+import play.api.Configuration
+
+import java.nio.charset.StandardCharsets
+
+object MqttClientFactory {
+  def createClient(configuration: Configuration, suffix: String = ""): Mqtt5AsyncClient = {
+    val mqttConfig = configuration.get[Configuration]("mqtt")
+
+    var builder = Mqtt5Client.builder()
+      .identifier(mqttConfig.get[String]("clientId") + suffix)
+      .serverHost(mqttConfig.get[String]("host"))
+      .serverPort(mqttConfig.get("port"))
+
+    mqttConfig.getOptional[String]("username").foreach { username =>
+      val password = mqttConfig.get[String]("password").getBytes(StandardCharsets.US_ASCII)
+      builder = builder.simpleAuth()
+        .username(username)
+        .password(password)
+        .applySimpleAuth()
+    }
+
+    builder.buildAsync()
+  }
+}

app/schemas/schemas.scala

@@ -1,9 +1,9 @@
 package schemas
 
-import slick.jdbc.PostgresProfile.api._
+import slick.jdbc.PostgresProfile.api.*
-import models._
+import models.*
 import slick.lifted.ProvenShape
-import slick.model.ForeignKeyAction.Cascade
+import slick.model.ForeignKeyAction.{Cascade, Restrict}
 
 import java.time.Instant
 
@@ -32,3 +32,28 @@ class AuthTokensTable(tag: Tag) extends Table[AuthToken](tag, "auth_tokens") {
 }
 
 lazy val AuthTokens = TableQuery[AuthTokensTable]
+
+implicit val auctionStatusColumnType: BaseColumnType[AuctionStatus] =
+  MappedColumnType.base[AuctionStatus, Short](
+    enumValue => enumValue.value,
+    intValue => AuctionStatus.byValue(intValue)
+  )
+
+class AuctionsTable(tag: Tag) extends Table[Auction](tag, "auctions") {
+  def id = column[Int]("id", O.PrimaryKey, O.AutoInc)
+  def name = column[String]("name")
+  def status = column[AuctionStatus]("status")
+  def gid = column[String]("gid", O.Unique)
+  def authorId = column[Int]("author_id")
+  def startingPrice = column[BigDecimal]("starting_price", O.SqlType("DECIMAL(20, 3)"))
+  def currentBid = column[Option[BigDecimal]]("current_bid", O.SqlType("DECIMAL(20, 3)"))
+  def currentBidderId = column[Option[Int]]("current_bidder_id")
+  def currentBidAt = column[Option[Instant]]("current_bid_at")
+  def createdAt = column[Instant]("created_at")
+
+  override def * : ProvenShape[Auction] = (id.?, name, status, gid, authorId, startingPrice, currentBid, currentBidderId, currentBidAt, createdAt).mapTo[Auction]
+
+  def author = foreignKey("users", authorId, Users)(_.id, onDelete = Restrict)
+}
+
+lazy val Auctions = TableQuery[AuctionsTable]

build.sbt

@@ -21,6 +21,9 @@ libraryDependencies += "org.playframework" %% "play-slick-evolutions" % "6.2.0"
 // Source: https://mvnrepository.com/artifact/de.mkammerer/argon2-jvm
 libraryDependencies += "de.mkammerer" % "argon2-jvm" % "2.12"
 
+// Source: https://mvnrepository.com/artifact/com.hivemq/hivemq-mqtt-client
+libraryDependencies += "com.hivemq" % "hivemq-mqtt-client" % "1.3.17"
+
 // Adds additional packages into Twirl
 //TwirlKeys.templateImports += "me.artemis.controllers._"
 

compose.yaml

@@ -1,4 +1,6 @@
 name: 'neon-vertigo-api'
+include:
+  - ../shared/compose.shared.yml
 services:
   db:
     image: "postgres:18"

conf/application.conf

@@ -2,6 +2,16 @@
 
 play.http.filters=controllers.filters.Filters
 
+play.filters.enabled += "play.filters.cors.CORSFilter"
+play.filters.csrf.header.bypassHeaders {
+    X-Requested-With = "*"
+}
+play.filters.cors {
+    allowedOrigins = ["http://localhost:4321"]
+}
+
+play.modules.enabled += "mqttClient.Module"
+
 slick.dbs.default = {
     profile = "slick.jdbc.PostgresProfile$"
     db = {
@@ -26,5 +36,20 @@ pekko {
             }
             throughput = 1
         }
+
+        mqtt-dispatcher {
+            type = Dispatcher
+            executor = "fork-join-executor"
+            fork-join-executor {
             }
         }
+    }
+}
+
+mqtt {
+    host = "localhost"
+    port = 1883
+    clientId = "neon-vertigo-api"
+    username = "artemis"
+    password = "artemis"
+}

conf/evolutions/default/3.sql

@@ -2,13 +2,16 @@
 CREATE TABLE auctions
 (
     id                BIGINT PRIMARY KEY GENERATED BY DEFAULT AS IDENTITY,
+    name              VARCHAR                      NOT NULL,
     gid               VARCHAR UNIQUE               NOT NULL,
-    author_id         BIGINT         NOT NULL REFERENCES users (id) ON DELETE RESTRICT,
+    author_id         BIGINT REFERENCES users (id) NOT NULL,
-    description       VARCHAR,
+    starting_price    DECIMAL(20, 3)               NOT NULL,
-    starting_price    DECIMAL(12, 3) NOT NULL,
+    current_bid       DECIMAL(20, 3),
-    current_bid       DECIMAL(12, 3),
+    current_bidder_id BIGINT REFERENCES users (id),
-    current_bidder_id BIGINT         REFERENCES users (id) ON DELETE SET NULL
+    current_bid_at    TIMESTAMP,
+    created_at        TIMESTAMP                    NOT NULL
 );
 
 -- !Downs
-DROP TABLE auctions;
+
+DROP TABLE auctions

conf/evolutions/default/4.sql

@@ -0,0 +1,7 @@
+-- !Ups
+ALTER TABLE auctions
+ADD COLUMN status SMALLINT NOT NULL DEFAULT 0;
+
+-- !Downs
+ALTER TABLE auctions
+DROP COLUMN status;

conf/logback.xml

@@ -41,6 +41,8 @@
 
   <logger name="play" level="INFO"/>
   <logger name="application" level="DEBUG"/>
+  <logger name="mqttClient" level="DEBUG"/>
+  <logger name="actors" level="INFO"/>
 
   <root level="WARN">
     <appender-ref ref="ASYNCFILE"/>

conf/routes

@@ -7,9 +7,12 @@
 GET     /                           controllers.HomeController.index()
 
 POST    /login                      controllers.authentication.SessionsController.login()
-# TODO: enable it later
-+nocsrf
 DELETE  /logout                     controllers.authentication.SessionsController.logout()
 
 # Map static resources from the /public folder to the /assets URL path
 GET     /assets/*file               controllers.Assets.versioned(path="/public", file: Asset)
+
+GET     /auctions                   controllers.AuctionsController.index()
+POST    /auctions                   controllers.AuctionsController.create()
+
+-> /-devel                          controllers.devel.DevRouter