Processing new bid MQTT messages

app/dao/AuctionsDAO.scala

@@ -52,6 +52,19 @@ class AuctionsDAO @Inject()(protected val dbConfigProvider: DatabaseConfigProvid
       .map(_ == 1)
   }
 
+  def find_by_gid(gid: String): Future[Option[Auction]] = {
+    db.run(Auctions.filter(_.gid === gid).take(1).result.headOption)
+  }
+
+  def newBid(auction: Auction, bidder: User, bid: BigDecimal, timestamp: Instant): Future[Option[Auction]] = {
+    db.run {
+      Auctions.filter(_.id === auction.id.get).map{ it => (it.currentBid, it.currentBidAt, it.currentBidderId) }
+        .update((Some(bid), Some(timestamp), Some(bidder.id.get)))
+        .andThen { Auctions.filter(_.id === auction.id.get).result.headOption }
+        .transactionally
+    }
+  }
+
   private def availableScope(user: User) = {
     Auctions.filter(row => (row.status === AuctionStatus.Ready) && (row.authorId =!= user.id.get))
   }

app/dao/UsersDAO.scala

@@ -69,6 +69,10 @@ class UsersDAO @Inject()(protected val dbConfigProvider: DatabaseConfigProvider,
       }
   }
 
+  def find_by_gid(gid: String): Future[Option[User]] = {
+    db.run(Users.filter(_.gid === gid).take(1).result.headOption)
+  }
+
   private def argonHash(data: String) = argon2.hash(ArgonParameters.iterations, ArgonParameters.memory, ArgonParameters.parallelism, data.getBytes(UTF_8))
 }
 

app/models/models.scala

@@ -36,7 +36,7 @@ object AuctionStatus {
 
 case class Auction(id: Option[Long], name: String, status: AuctionStatus, gid: String, authorId: Long,
                    startingPrice: BigDecimal, currentBid: Option[BigDecimal],
-                   currentBidderId: Option[Int], currentBidAt: Option[Instant],
+                   currentBidderId: Option[Long], currentBidAt: Option[Instant],
                    createdAt: Instant)
 
 object Auction {

app/mqttClient/MessageProcessorActor.scala

@@ -8,8 +8,10 @@ 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 play.api.libs.json._
 
 import java.nio.charset.StandardCharsets
+import scala.util.matching.Regex
 
 object MessageProcessorActor extends ActorModule with Logging {
   sealed trait Command
@@ -18,14 +20,16 @@ object MessageProcessorActor extends ActorModule with Logging {
 
   override type Message = Command
 
+  private final val newBidPattern: Regex = """notify/new_bid/(\w+)""".r
+
   @Singleton
   class Reg @Inject()(private val mqttMessageProcessorActor: ActorRef[Command])
 
   @Provides
-  def create(client: Mqtt5AsyncClient): Behavior[Command] = Behaviors.setup { context =>
+  def create(client: Mqtt5AsyncClient, newBidService: services.auctions.NewBid): Behavior[Command] = Behaviors.setup { context =>
     client.connect().whenComplete { (_action, _throwable) =>
       client.subscribeWith()
-        .topicFilter("test/+")
+        .topicFilter("notify/new_bid/+")
         .callback{ (publish) =>
           context.self ! MqttMessageReceived(publish)
         }
@@ -37,6 +41,17 @@ object MessageProcessorActor extends ActorModule with Logging {
         val topic = publish.getTopic
         val payload = String(publish.getPayloadAsBytes, StandardCharsets.UTF_8)
         logger.info(s"Received message $topic $payload")
+
+        topic.toString match {
+          case newBidPattern(auctionGid) =>
+            Json.parse(payload).validate[services.auctions.NewBid.Request] match {
+              case JsSuccess(request, _) => newBidService.perform(auctionGid, request)
+              case JsError(errors) =>
+                logger.warn(errors.toString())
+            }
+            println(s"New bid $auctionGid")
+        }
+
         Behaviors.same
       case Stop => Behaviors.stopped{ () =>
         client.disconnect()

app/schemas/schemas.scala

@@ -47,7 +47,7 @@ class AuctionsTable(tag: Tag) extends Table[Auction](tag, "auctions") {
   def authorId = column[Long]("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 currentBidderId = column[Option[Long]]("current_bidder_id")
   def currentBidAt = column[Option[Instant]]("current_bid_at")
   def createdAt = column[Instant]("created_at")
 

app/services/auctions/NewBid.scala

@@ -0,0 +1,39 @@
+package services.auctions
+
+import dao.{AuctionsDAO, UsersDAO}
+import models.Auction
+import play.api.libs.functional.syntax.toFunctionalBuilderOps
+import play.api.libs.json.{JsPath, JsResult, Reads}
+
+import java.time.Instant
+import javax.inject.Inject
+import scala.concurrent.{ExecutionContext, Future}
+
+object NewBid {
+  case class Request(bid: BigDecimal, bidderGid: String, timestamp: Instant)
+
+  given Reads[Request] = (
+    (JsPath \ "bid").read[BigDecimal] and
+      (JsPath \ "bidder_gid").read[String] and
+      (JsPath \ "timestamp").read[Instant]
+    )(Request.apply)
+}
+
+class NewBid @Inject()(auctionsDao: AuctionsDAO, usersDAO: UsersDAO)(using ec: ExecutionContext) {
+  import NewBid.Request
+
+  def perform(auctionGid: String, request: Request): Future[Option[Auction]] = {
+    val dataFuture = for {
+      auction <- auctionsDao.find_by_gid(auctionGid)
+      bidder <- usersDAO.find_by_gid(request.bidderGid)
+    } yield (auction, bidder)
+
+    dataFuture.flatMap {
+      case (Some(auction), Some(bidder)) =>
+        auctionsDao.newBid(auction, bidder, bid = request.bid, timestamp = request.timestamp)
+      case (_, _) =>
+        // One of them (or both) aren't found
+        Future.successful(None)
+    }
+  }
+}