From d90ae86e1fe94599b48b34b02539f73fb84ebf2a Mon Sep 17 00:00:00 2001 From: Artemiy Solopov Date: Mon, 21 Sep 2026 15:03:45 +0300 Subject: [PATCH] Processing new bid MQTT messages --- app/dao/AuctionsDAO.scala | 13 ++++++++ app/dao/UsersDAO.scala | 4 +++ app/models/models.scala | 2 +- app/mqttClient/MessageProcessorActor.scala | 19 +++++++++-- app/schemas/schemas.scala | 2 +- app/services/auctions/NewBid.scala | 39 ++++++++++++++++++++++ 6 files changed, 75 insertions(+), 4 deletions(-) create mode 100644 app/services/auctions/NewBid.scala diff --git a/app/dao/AuctionsDAO.scala b/app/dao/AuctionsDAO.scala index 2d2735c..409f3c0 100644 --- a/app/dao/AuctionsDAO.scala +++ b/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)) } diff --git a/app/dao/UsersDAO.scala b/app/dao/UsersDAO.scala index 7148960..9d30c8b 100644 --- a/app/dao/UsersDAO.scala +++ b/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)) } diff --git a/app/models/models.scala b/app/models/models.scala index 01f4b25..2a98e95 100644 --- a/app/models/models.scala +++ b/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 { diff --git a/app/mqttClient/MessageProcessorActor.scala b/app/mqttClient/MessageProcessorActor.scala index de2e979..1b3ed4b 100644 --- a/app/mqttClient/MessageProcessorActor.scala +++ b/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() diff --git a/app/schemas/schemas.scala b/app/schemas/schemas.scala index 6dd7318..cf81e4c 100644 --- a/app/schemas/schemas.scala +++ b/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") diff --git a/app/services/auctions/NewBid.scala b/app/services/auctions/NewBid.scala new file mode 100644 index 0000000..be60ef3 --- /dev/null +++ b/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) + } + } +}