Auction status change and confirmation

app/actors/auctions/AuctionPublisher.scala

@@ -4,21 +4,25 @@ 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 models.Auction
+import dao.AuctionsDAO
+import models.{Auction, AuctionStatus, User}
 import mqttClient.MqttClientFactory
-import org.apache.pekko.actor.typed.Behavior
+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.Json
+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) 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"
 
@@ -26,41 +30,74 @@ object AuctionPublisher extends ActorModule with Logging {
 
   @Provides
   @Singleton
-  def apply(configuration: Configuration): Behavior[Message] = Behaviors.setup { context =>
+  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(client)
+    idle(context.self, client, auctionsDAO)
   }
 
-  private def idle(client: Mqtt5AsyncClient, waitingPublishes: Seq[Auction] = List()): Behaviors.Receive[Command] = Behaviors.receiveMessage[Command] {
+  private def idle(self: ActorRef[Message], client: Mqtt5AsyncClient, auctionsDAO: AuctionsDAO, waitingPublishes: Seq[Publish] = List()): Behaviors.Receive[Command] = Behaviors.receiveMessage[Command] {
-    case Publish(auction) =>
+    case p: Publish =>
-      idle(client, waitingPublishes :+ auction)
+      idle(self, client, auctionsDAO, waitingPublishes :+ p)
     case Ready =>
-      for { auction <- waitingPublishes } publishAuction(client, auction)
+      for { publish <- waitingPublishes } publishAuction(client, publish.auction, publish.author, self)
-      ready(client)
+      ready(self, client, auctionsDAO)
+    case message @ AuctionConfirmed(_, _) =>
+      logger.warn(s"Received impossible message $message")
+      idle(self, client, auctionsDAO, waitingPublishes)
   }
 
-  private def ready(client: Mqtt5AsyncClient): Behaviors.Receive[Command] = Behaviors.receiveMessage[Command] {
+  private def ready(self: ActorRef[Message], client: Mqtt5AsyncClient, auctionsDAO: AuctionsDAO): Behaviors.Receive[Command] = Behaviors.receiveMessage[Command] {
-    case Publish(auction) =>
+    case Publish(auction, author) =>
-      publishAuction(client, auction)
+      publishAuction(client, auction, author, self)
-      ready(client)
+      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 ")
+      logger.warn("Message Ready received after having already received it before")
-      ready(client)
+      ready(self, client, auctionsDAO)
   }
 
-  private def publishAuction(client: Mqtt5AsyncClient, auction: Auction): Unit = {
+  private def publishAuction(client: Mqtt5AsyncClient, auction: Auction, author: User, self: ActorRef[Command]): Unit = {
-    val payload = Json.toBytes(Json.toJson(auction))
+    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

@@ -28,7 +28,7 @@ class AuctionsController @Inject()(val controllerComponents: ControllerComponent
           auction <- auctionsDAO.create(auctionCreateData.name, auctionCreateData.startingPrice, request.user)
           json = Json.toJson(auction)
           response = Created(json)
-          _ = auctionPublisher ! AuctionPublisher.Publish(auction)
+          _ = auctionPublisher ! AuctionPublisher.Publish(auction, request.user)
         } yield response
       }
     )

app/dao/AuctionsDAO.scala

@@ -1,15 +1,16 @@
 package dao
 
-import models.Auction
+import models.{Auction, AuctionStatus}
 import schemas.Auctions
+import schemas.auctionStatusColumnType
 import play.api.db.slick.{DatabaseConfigProvider, HasDatabaseConfigProvider}
 import slick.jdbc.PostgresProfile
 import slick.jdbc.PostgresProfile.api.*
-import utils.Random
 
 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] {
@@ -17,7 +18,8 @@ class AuctionsDAO @Inject()(protected val dbConfigProvider: DatabaseConfigProvid
   def create(name: String, startingPrice: BigDecimal, author: models.User): Future[Auction] = {
     val auction = Auction(
       id = None, name = name,
-      gid = Random.randomBase64(24),
+      gid = Random.alphanumeric.take(24).mkString,
+      status = AuctionStatus.Created,
       authorId = author.id.get,
       startingPrice = startingPrice,
       currentBid = None, currentBidderId = None, currentBidAt = None,
@@ -30,4 +32,9 @@ class AuctionsDAO @Inject()(protected val dbConfigProvider: DatabaseConfigProvid
   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

@@ -12,9 +12,9 @@ import java.time.{Duration, Instant}
 import scala.concurrent.{ExecutionContext, Future}
 import models.AuthToken
 import schemas.AuthTokens
-import utils.Random
 
 import scala.language.postfixOps
+import scala.util.Random
 
 class AuthTokensDAO @Inject()(protected val dbConfigProvider: DatabaseConfigProvider)(implicit val ec: ExecutionContext) extends HasDatabaseConfigProvider[PostgresProfile] {
   private final lazy val tokenValidDuration = Duration.ofDays(30)
@@ -28,7 +28,7 @@ class AuthTokensDAO @Inject()(protected val dbConfigProvider: DatabaseConfigProv
     val token = AuthToken(
       id = None,
       userId = user.id.get,
-      token = Random.randomBase64(12),
+      token = Random.alphanumeric.take(12).mkString,
       createdAt = Instant.now,
       expiresAt = Instant.now.plus(tokenValidDuration)
     )

app/models/models.scala

@@ -8,7 +8,17 @@ case class User(id: Option[Int], username: String, gid: String, passwordDigest:
 
 case class AuthToken(id: Option[Int], userId: Int, token: String, createdAt: Instant, expiresAt: Instant)
 
-case class Auction(id: Option[Int], name: String, gid: String, authorId: Int,
+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)})
+}
+
+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)

app/schemas/schemas.scala

@@ -33,9 +33,16 @@ 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")
@@ -44,7 +51,7 @@ class AuctionsTable(tag: Tag) extends Table[Auction](tag, "auctions") {
   def currentBidAt = column[Option[Instant]]("current_bid_at")
   def createdAt = column[Instant]("created_at")
 
-  override def * : ProvenShape[Auction] = (id.?, name, gid, authorId, startingPrice, currentBid, currentBidderId, currentBidAt, createdAt).mapTo[Auction]
+  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)
 }

app/utils/Random.scala

@@ -1,16 +0,0 @@
-package utils
-
-import java.nio.ByteBuffer
-import java.security.SecureRandom
-import java.util.Base64
-
-object Random {
-  private lazy val random = SecureRandom()
-  private lazy val base64 = Base64.getEncoder
-  
-  def randomBase64(bytes: Int): String = {
-    val tokenRaw = ByteBuffer.allocate(bytes)
-    random.nextBytes(tokenRaw.array())
-    base64.encodeToString(tokenRaw.array())
-  }
-}

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;