package actors.auctions

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 dao.AuctionsDAO
import models.{Auction, AuctionStatus, User}
import mqttClient.MqttClientFactory
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.{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"

  override type Message = Command

  @Provides
  @Singleton
  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(context.self, client, auctionsDAO)
  }

  private def idle(self: ActorRef[Message], client: Mqtt5AsyncClient, auctionsDAO: AuctionsDAO, waitingPublishes: Seq[Publish] = List()): Behaviors.Receive[Command] = Behaviors.receiveMessage[Command] {
    case p: Publish =>
      idle(self, client, auctionsDAO, waitingPublishes :+ p)
    case Ready =>
      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(self: ActorRef[Message], client: Mqtt5AsyncClient, auctionsDAO: AuctionsDAO): Behaviors.Receive[Command] = Behaviors.receiveMessage[Command] {
    case Publish(auction, author) =>
      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(self, client, auctionsDAO)
  }

  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)}")
    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
    )
  }
}