Processing new bid MQTT messages

app/actors/auctions/AuctionPublisher.scala

@@ -22,7 +22,7 @@ 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 case class AuctionConfirmed(id: Long, responseTopic: String) extends Command
 
   private final val topic = "auctions/created"
 

app/controllers/AuctionsController.scala

@@ -3,23 +3,25 @@ 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}
+import models.Auction
+import caching.{CheckFresh, DigestEtagable, Etagable}
+
+import java.io.DataOutputStream
 
 @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 {
+                                   val auctionPublisher: ActorRef[AuctionPublisher.Command])
+                                  (implicit val ec: ExecutionContext)
+  extends BaseController with CheckFresh {
   def index = authenticatedAction.async { request =>
     for {
       auctions <- auctionsDAO.list_available(request.user)
@@ -27,6 +29,16 @@ class AuctionsController @Inject()(val controllerComponents: ControllerComponent
     } yield Ok(json)
   }
 
+  def show(id: Long) = authenticatedAction.async { implicit request =>
+    auctionsDAO.showAvailable(request.user, id).map { auctionOpt =>
+      auctionOpt.fold(NotFound) { auction =>
+        checkFresh(auction) {
+          Ok(Json.toJson(auction)).withHeaders("etag" -> auction.etag)
+        }
+      }
+    }
+  }
+
   def create = authenticatedAction.async(parse.json) { implicit request =>
     val auctionCreateDataResult = request.body.validate[AuctionCreateData]
     auctionCreateDataResult.fold(
@@ -44,4 +56,18 @@ class AuctionsController @Inject()(val controllerComponents: ControllerComponent
   private case class AuctionCreateData(name: String, startingPrice: BigDecimal)
 
   private given Reads[AuctionCreateData] = Json.reads[AuctionCreateData]
+
+  private given DigestEtagable[Auction] with {
+    override protected def writeDigestData(a: Auction, s: DataOutputStream): Unit = {
+      s.writeLong(a.id.getOrElse(0))
+      s.writeLong(a.authorId)
+      s.writeUTF(a.gid)
+      s.writeUTF(a.name)
+      s.writeLong(a.createdAt.getEpochSecond)
+
+      s.writeLong(a.currentBidderId.getOrElse(0))
+      a.currentBid.foreach(b => s.writeLong((b * 1000).toLong))
+      a.currentBidAt.foreach(ba => s.writeLong(ba.getEpochSecond))
+    }
+  }
 }

app/controllers/HomeController.scala

@@ -1,15 +1,24 @@
 package controllers
 
-import javax.inject._
-import play.api._
-import play.api.mvc._
+import play.api.*
+import _root_.controllers.authentication.{AuthenticatedAction, AuthenticatedRequest}
+import _root_.controllers.caching.{CheckFresh, DigestEtagable}
+import models.User
+import play.api.libs.json.Json
+import play.api.mvc.*
+
+import java.io.DataOutputStream
+import java.nio.charset.StandardCharsets.UTF_8
+import javax.inject.*
 
 /**
  * This controller creates an `Action` to handle HTTP requests to the
  * application's home page.
  */
 @Singleton
-class HomeController @Inject()(val controllerComponents: ControllerComponents) extends BaseController {
+class HomeController @Inject()(val controllerComponents: ControllerComponents,
+                               authenticatedAction: AuthenticatedAction)
+  extends BaseController with Logging with CheckFresh {
 
   /**
    * Create an Action to render an HTML page.
@@ -21,4 +30,20 @@ class HomeController @Inject()(val controllerComponents: ControllerComponents) e
   def index() = Action { implicit request: Request[AnyContent] =>
     Ok(views.html.index())
   }
+
+  def userData = authenticatedAction { implicit request: AuthenticatedRequest[AnyContent] =>
+    val user = request.user
+    checkFresh(user) {
+      // TODO: put the max-age somewhere into config
+      Ok(Json.toJson(user)).withHeaders("Cache-Control" -> "private, max-age=15, must-revalidate")
+    }
+  }
+
+  private given DigestEtagable[models.User] with {
+    override protected def writeDigestData(a: User, s: DataOutputStream): Unit = {
+      s.writeLong(a.id.getOrElse(0))
+      s.write(a.gid.getBytes(UTF_8))
+      s.write(a.username.getBytes(UTF_8))
+    }
+  }
 }

app/controllers/caching/caching.scala

@@ -0,0 +1,57 @@
+package controllers.caching
+
+import play.api.mvc.Results.NotModified
+import play.api.mvc.{Request, Result}
+
+import java.io.{ByteArrayOutputStream, DataOutputStream, OutputStream}
+import java.security.{DigestOutputStream, MessageDigest}
+import java.util.Base64
+import scala.concurrent.{ExecutionContext, Future}
+import scala.util.Using
+
+trait Etagable[A]:
+  extension (a: A) def etag: String
+
+
+trait DigestEtagable[A] extends Etagable[A]:
+  private val base64 = Base64.getUrlEncoder.withoutPadding
+
+  protected def writeDigestData(a: A, s: DataOutputStream): Unit
+
+  extension (a: A) {
+    override def etag: String = {
+      val md = MessageDigest.getInstance("SHA-1")
+      Using(DataOutputStream(DigestOutputStream(OutputStream.nullOutputStream(), md))) { dos =>
+        writeDigestData(a, dos)
+        dos.flush()
+      }
+
+      base64.encodeToString(md.digest())
+    }
+  }
+
+
+trait CheckFresh {
+  protected def checkFreshAsync[A, E: Etagable](entity: E)(onStale: => Future[Result])(using request: Request[A], ec: ExecutionContext): Future[Result] = {
+    compareEtag(request, entity).fold(
+      entityEtag => onStale.map(_.withHeaders("etag" -> entityEtag)),
+      requestEtag => Future.successful(notModified(requestEtag))
+    )
+  }
+
+  protected def checkFresh[A, E: Etagable](entity: E)(onStale: => Result)(using request: Request[A]): Result = {
+    compareEtag(request, entity).fold(
+      entityEtag => onStale.withHeaders("etag" -> entityEtag),
+      requestEtag => notModified(requestEtag)
+    )
+  }
+
+  private def compareEtag[A, E: Etagable](request: Request[A], entity: E) = {
+    val entityEtag = entity.etag
+    request.headers.get("If-None-Match").filter(e => e == entityEtag) match {
+      case Some(etag) => Right(etag)
+      case None => Left(entityEtag)
+    }
+  }
+  private def notModified(etag: String) = NotModified.withHeaders("etag" -> etag)
+}

app/dao/AuctionsDAO.scala

@@ -1,31 +1,35 @@
 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
 
+import models.{Auction, AuctionStatus, User}
+import schemas.Auctions
+import schemas.given
+
 class AuctionsDAO @Inject()(protected val dbConfigProvider: DatabaseConfigProvider)(implicit val ec: ExecutionContext)
   extends HasDatabaseConfigProvider[PostgresProfile] {
 
-  def list_available(user: models.User): Future[Seq[Auction]] = {
+  def list_available(user: User): Future[Seq[Auction]] = {
+    db.run(availableScope(user).result)
+  }
+
+  def showAvailable(user: User, id: Long): Future[Option[Auction]] = {
     db.run {
-      Auctions.filter(row => (row.status === AuctionStatus.Ready) && (row.authorId =!= user.id.get)).result
+      availableScope(user).filter(_.id === id).take(1).result.headOption
     }
   }
 
-  def list_created_by(user: models.User): Future[Seq[Auction]] = {
+  def list_created_by(user: User): Future[Seq[Auction]] = {
     db.run(Auctions.filter(_.authorId === user.id).result)
   }
 
-  def create(name: String, startingPrice: BigDecimal, author: models.User): Future[Auction] = {
+  def create(name: String, startingPrice: BigDecimal, author: User): Future[Auction] = {
     val auction = Auction(
       id = None, name = name,
       gid = Random.alphanumeric.take(24).mkString,
@@ -43,8 +47,25 @@ class AuctionsDAO @Inject()(protected val dbConfigProvider: DatabaseConfigProvid
     db.run((Auctions returning Auctions) += auction)
   }
 
-  def changeStatus(auctionId: Int, status: AuctionStatus): Future[Boolean] = {
+  def changeStatus(auctionId: Long, status: AuctionStatus): Future[Boolean] = {
     db.run(Auctions.filter(_.id === auctionId).map(_.status).update(status))
-    .map(_ == 1)
+      .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

@@ -4,17 +4,15 @@ 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])
+case class User(id: Option[Long], 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
-    )
-  }
+  given writes: Writes[User] = (o: User) => Json.obj(
+    "gid" -> o.gid,
+    "username" -> o.username
+  )
 }
 
-case class AuthToken(id: Option[Int], userId: Int, token: String, createdAt: Instant, expiresAt: Instant)
+case class AuthToken(id: Option[Long], userId: Long, token: String, createdAt: Instant, expiresAt: Instant)
 object AuthToken {
   given writes: OWrites[AuthToken] = (a: AuthToken) => Json.obj(
     "token" -> a.token,
@@ -36,9 +34,9 @@ object AuctionStatus {
   implicit val writes: Writes[AuctionStatus] = (o: AuctionStatus) => JsString(o.toString)
 }
 
-case class Auction(id: Option[Int], name: String, status: AuctionStatus, gid: String, authorId: Int,
+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

@@ -8,7 +8,7 @@ import slick.model.ForeignKeyAction.{Cascade, Restrict}
 import java.time.Instant
 
 class UsersTable(tag: Tag) extends Table[User](tag, "users") {
-  def id = column[Int]("id", O.PrimaryKey, O.AutoInc)
+  def id = column[Long]("id", O.PrimaryKey, O.AutoInc)
   def gid = column[String]("gid", O.Unique)
   def username = column[String]("username", O.Unique)
   def passwordDigest = column[Option[String]]("password_digest")
@@ -20,8 +20,8 @@ lazy val Users = TableQuery[UsersTable]
 
 class AuthTokensTable(tag: Tag) extends Table[AuthToken](tag, "auth_tokens") {
 
-  def id = column[Int]("id", O.PrimaryKey, O.AutoInc)
-  def userId = column[Int]("user_id")
+  def id = column[Long]("id", O.PrimaryKey, O.AutoInc)
+  def userId = column[Long]("user_id")
   def token = column[String]("token")
   def createdAt = column[Instant]("created_at")
   def expiresAt = column[Instant]("expires_at")
@@ -40,14 +40,14 @@ implicit val auctionStatusColumnType: BaseColumnType[AuctionStatus] =
   )
 
 class AuctionsTable(tag: Tag) extends Table[Auction](tag, "auctions") {
-  def id = column[Int]("id", O.PrimaryKey, O.AutoInc)
+  def id = column[Long]("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 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)
+    }
+  }
+}

compose.yaml

@@ -8,3 +8,8 @@ services:
       - db.env
     ports:
       - "5432:5432"
+    volumes:
+      - "db:/var/lib/postgresql"
+
+volumes:
+  db:

conf/application.conf

@@ -8,6 +8,7 @@ play.filters.csrf.header.bypassHeaders {
 }
 play.filters.cors {
     allowedOrigins = ["http://localhost:4321"]
+    exposedHeaders = ["ETag"]
 }
 
 play.modules.enabled += "mqttClient.Module"

conf/routes

@@ -9,10 +9,13 @@ GET     /                           controllers.HomeController.index()
 POST    /login                      controllers.authentication.SessionsController.login()
 DELETE  /logout                     controllers.authentication.SessionsController.logout()
 
+GET /user_data                      controllers.HomeController.userData()
+
 # 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()
+GET     /auctions/:id               controllers.AuctionsController.show(id: Long)
 POST    /auctions                   controllers.AuctionsController.create()
 
 -> /-devel                          controllers.devel.DevRouter