diff --git a/app/actors/auctions/AuctionPublisher.scala b/app/actors/auctions/AuctionPublisher.scala index e539191..16c89bd 100644 --- a/app/actors/auctions/AuctionPublisher.scala +++ b/app/actors/auctions/AuctionPublisher.scala @@ -4,20 +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, User} +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.{JsObject, 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, author: User) extends Command + private case class AuctionConfirmed(id: Int, responseTopic: String) extends Command private final val topic = "auctions/created" @@ -25,38 +30,63 @@ 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[Publish] = 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 p: Publish => - idle(client, waitingPublishes :+ p) + idle(self, client, auctionsDAO, waitingPublishes :+ p) case Ready => - for { publish <- waitingPublishes } publishAuction(client, publish.auction, publish.author) - ready(client) + 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(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, author) => - publishAuction(client, auction, author) - ready(client) + 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(client) + ready(self, client, auctionsDAO) } - private def publishAuction(client: Mqtt5AsyncClient, auction: Auction, author: User): Unit = { + 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)}") diff --git a/app/dao/AuctionsDAO.scala b/app/dao/AuctionsDAO.scala index 2c1d68d..7b1c5b6 100644 --- a/app/dao/AuctionsDAO.scala +++ b/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) + } } diff --git a/app/dao/AuthTokensDAO.scala b/app/dao/AuthTokensDAO.scala index fef04f5..bdf8835 100644 --- a/app/dao/AuthTokensDAO.scala +++ b/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) ) diff --git a/app/models/models.scala b/app/models/models.scala index 806e6fd..82b5575 100644 --- a/app/models/models.scala +++ b/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) diff --git a/app/schemas/schemas.scala b/app/schemas/schemas.scala index 3396f54..8f96afe 100644 --- a/app/schemas/schemas.scala +++ b/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) } diff --git a/app/utils/Random.scala b/app/utils/Random.scala deleted file mode 100644 index 1e71d2b..0000000 --- a/app/utils/Random.scala +++ /dev/null @@ -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()) - } -} diff --git a/conf/evolutions/default/4.sql b/conf/evolutions/default/4.sql new file mode 100644 index 0000000..a1ec49f --- /dev/null +++ b/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; \ No newline at end of file