feat/BE: setup websocket and implement broadcast message to user after moving columns

* setup websocket and implement broadcast message to user after moved column

* test: unit test for join websocket room and move column

Author: NhatNguyen1502

.gitignore

@@ -0,0 +1,2 @@
+.idea
+.vscode

backend/app/actors/ProjectActor.scala

@@ -0,0 +1,50 @@
+package actors
+
+import dto.websocket.OutgoingMessage
+import org.apache.pekko.actor.{Actor, ActorRef, Props}
+import play.api.Logger
+import play.api.libs.json.{JsValue, Json}
+
+/**
+  * Actor that manages WebSocket connections for a specific project.
+  */
+object ProjectActor {
+  def props(projectId: Int): Props = Props(new ProjectActor(projectId))
+
+  case class Join(userId: Int, out: ActorRef)
+  case class Leave(userId: Int)
+  case class Broadcast(message: OutgoingMessage)
+}
+
+/**
+  * Actor that manages WebSocket connections for a specific project.
+  * It keeps track of connected users and broadcasts messages to them.
+  *
+  * @param projectId the ID of the project
+  */
+class ProjectActor(projectId: Int) extends Actor {
+  import ProjectActor._
+
+  private var members = Map.empty[Int, ActorRef]
+
+  def receive: Receive = {
+    case Join(userId, out) =>
+      members += userId -> out
+      Logger("actors").info(
+        s"UserId $userId joined project $projectId. Total members: ${members.size}"
+      )
+
+    case Leave(userId) =>
+      members -= userId
+      Logger("actors").info(
+        s"UserId with $userId left project $projectId. Total members: ${members.size}"
+      )
+
+    case Broadcast(msg) =>
+      val js: JsValue = Json.toJson(msg)
+      members.values.foreach(_ ! js)
+      Logger("actors").info(
+        s"Broadcasted message to project $projectId members: ${members.size} users"
+      )
+  }
+}

backend/app/actors/ProjectActorRegistry.scala

@@ -0,0 +1,34 @@
+package actors
+
+import org.apache.pekko.actor.{Actor, ActorRef, Props}
+import dto.websocket.OutgoingMessage
+import modules.ActorNames
+
+object ProjectActorRegistry {
+  def props: Props = Props(new ProjectActorRegistry)
+
+  case class GetProjectActor(projectId: Int)
+  case class BroadcastToProject(projectId: Int, message: OutgoingMessage)
+}
+
+class ProjectActorRegistry extends Actor {
+  import ProjectActorRegistry._
+  import ProjectActor._
+
+  private var projectActors = Map.empty[Int, ActorRef]
+
+  def receive: Receive = {
+    case GetProjectActor(projectId) =>
+      // create new ProjectActor if not exists
+      val actor = projectActors.getOrElse(projectId, {
+        val newActor = context.actorOf(ProjectActor.props(projectId), s"${ActorNames.ProjectActorPrefix}$projectId")
+        projectActors += projectId -> newActor
+        newActor
+      })
+      // return ActorRef for requester
+      sender() ! actor
+
+    case BroadcastToProject(projectId, message) =>
+      projectActors.get(projectId).foreach(_ ! Broadcast(message))
+  }
+}

backend/app/actors/ProjectClientActor.scala

@@ -0,0 +1,72 @@
+package actors
+
+import org.apache.pekko.actor.{Actor, ActorRef, Props}
+import org.apache.pekko.pattern.{ask, pipe}
+import org.apache.pekko.util.Timeout
+
+import scala.concurrent.ExecutionContext
+import scala.concurrent.duration._
+
+/**
+ * Actor that manages a WebSocket connection for a user in a specific project.
+ */
+object ProjectClientActor {
+  def props(out: ActorRef,
+            userId: Int,
+            projectId: Int,
+            registry: ActorRef): Props =
+    Props(new ProjectClientActor(out, userId, projectId, registry))
+
+  /**
+   * Message indicating that the ProjectActor has been found in the registry.
+   * @param projectRef the ActorRef of the ProjectActor
+   */
+  private case class RegistryFound(projectRef: ActorRef)
+}
+
+/**
+ * Actor that manages a WebSocket connection for a user in a specific project.
+ * It registers the user with the ProjectActor upon creation and deregisters
+ * upon termination.
+ *
+ * @param out      the ActorRef to send messages to the WebSocket
+ * @param userId   the ID of the user
+ * @param projectId the ID of the project
+ * @param registry the ActorRef of the ProjectActorRegistry
+ */
+class ProjectClientActor(out: ActorRef,
+                         userId: Int,
+                         projectId: Int,
+                         registry: ActorRef)
+    extends Actor {
+  import ProjectActor._
+  import ProjectActorRegistry._
+  import ProjectClientActor._
+
+  // Execution context and timeout for ask pattern
+  implicit val ec: ExecutionContext = context.dispatcher
+  implicit val timeout: Timeout = Timeout(3.seconds)
+
+  // Reference to the ProjectActor, once found
+  private var projectRefOpt: Option[ActorRef] = None
+
+  // On start, ask the registry for the ProjectActor
+  override def preStart(): Unit = {
+    (registry ? GetProjectActor(projectId))
+      .mapTo[ActorRef]
+      .map(RegistryFound) pipeTo self
+  }
+
+  // On stop, inform the ProjectActor that the user is leaving
+  override def postStop(): Unit = {
+    projectRefOpt.foreach(_ ! Leave(userId))
+    super.postStop()
+  }
+
+  // Handle incoming messages
+  def receive: Receive = {
+    case RegistryFound(projectRef) =>
+      projectRefOpt = Some(projectRef)
+      projectRef ! Join(userId, out)
+  }
+}

backend/app/controllers/AuthenticatedAction.scala

@@ -1,5 +1,7 @@
 package controllers
 
+import org.apache.pekko.stream.scaladsl.Flow
+import play.api.libs.json.JsValue
 import play.api.mvc._
 import services.{CookieService, JwtService, UserToken}
 
@@ -62,3 +64,35 @@ class AuthenticatedActionWithUser @Inject()(
         }
     }
 }
+
+class AuthenticatedWebSocket @Inject()(
+                                          jwtService: JwtService,
+                                          cookieService: CookieService
+                                      )(implicit ec: ExecutionContext) {
+
+  def apply(
+    block: UserToken ⇒ Future[Either[Result, Flow[JsValue, JsValue, _]]]
+  ): WebSocket = {
+    WebSocket.acceptOrResult[JsValue, JsValue] { requestHeader ⇒
+      cookieService.getTokenFromRequest(requestHeader) match {
+        case Some(token) ⇒
+          jwtService.validateToken(token) match {
+            case Success(userToken) ⇒
+              block(userToken)
+            case Failure(ex) ⇒
+              Future.successful(
+                Left(
+                  Results
+                    .Unauthorized(s"Invalid token: ${ex.getMessage}")
+                    .withCookies(cookieService.createExpiredAuthCookie())
+                )
+              )
+          }
+        case None ⇒
+          Future.successful(
+            Left(Results.Unauthorized("No authentication token found"))
+          )
+      }
+    }
+  }
+}

backend/app/controllers/ColumnController.scala

@@ -112,15 +112,15 @@ class ColumnController @Inject()(
       }
     }
 
-  /** PATCH /columns/:columnId/position */
-  def updatePosition(columnId: Int): Action[JsValue] =
+  /** PATCH /projects/:projectId/columns/:columnId/position */
+  def updatePosition(projectId: Int, columnId: Int): Action[JsValue] =
     authenticatedActionWithUser.async(parse.json) { request =>
       implicit val messages: Messages = request.messages
       val updatedBy = request.userToken.userId
       handleJsonValidation[UpdateColumnPositionRequest](request.body) {
-        updatePositionDto =>
+        updateColumnPositionDto =>
           columnService
-            .updatePosition(columnId, updatePositionDto, updatedBy)
+            .updatePosition(projectId, columnId, updateColumnPositionDto, updatedBy)
             .map { _ =>
               Ok(
                 Json.toJson(

backend/app/controllers/WebSocketController.scala

@@ -0,0 +1,37 @@
+package controllers
+
+import modules.ActorNames
+import org.apache.pekko.actor.{ActorRef, ActorSystem}
+import org.apache.pekko.stream.Materializer
+import play.api.libs.json.JsValue
+import play.api.libs.streams.ActorFlow
+import play.api.mvc._
+import services.ProjectService
+
+import javax.inject._
+import scala.concurrent.ExecutionContext
+
+@Singleton
+class WebSocketController @Inject()(
+                                     cc: ControllerComponents,
+                                     authenticatedWebSocket: AuthenticatedWebSocket,
+                                     @Named(ActorNames.ProjectActorRegistry) projectActorRegistry: ActorRef,
+                                     projectService: ProjectService
+                                   )(implicit system: ActorSystem, mat: Materializer, ec: ExecutionContext)
+  extends AbstractController(cc) {
+
+  def joinProject(projectId: Int): WebSocket = authenticatedWebSocket { userToken =>
+    projectService.isUserInActiveProject(userToken.userId, projectId).map { exists =>
+      if (exists) {
+        Right(
+          ActorFlow.actorRef[JsValue, JsValue] { out =>
+            actors.ProjectClientActor.props(out, userToken.userId, projectId, projectActorRegistry)
+          }
+        )
+      } else {
+        Left(Results.Forbidden("User is not a member of this project"))
+      }
+    }
+  }
+}
+

backend/app/dto/websocket/OutgoingMessage.scala

@@ -0,0 +1,12 @@
+package dto.websocket
+
+import dto.websocket.board.BoardMessage
+import play.api.libs.json.Writes
+
+trait OutgoingMessage
+
+object OutgoingMessage {
+  implicit val writes: Writes[OutgoingMessage] = {
+    case bm: BoardMessage => BoardMessage.writes.writes(bm)
+  }
+}

backend/app/dto/websocket/board/BoardMessage.scala

@@ -0,0 +1,19 @@
+package dto.websocket.board
+
+import play.api.libs.json._
+import dto.websocket.OutgoingMessage
+
+sealed trait BoardMessage extends OutgoingMessage
+
+case class ColumnMoved(columnId: Int, newPosition: Int) extends BoardMessage
+case class TaskMoved(taskId: Int, fromColumnId: Int, toColumnId: Int, newPosition: Int) extends BoardMessage
+
+object BoardMessage {
+  implicit val columnMovedFormat: OFormat[ColumnMoved] = Json.format[ColumnMoved]
+  implicit val taskMovedFormat: OFormat[TaskMoved]     = Json.format[TaskMoved]
+
+  implicit val writes: Writes[BoardMessage] = Writes {
+    case cm: ColumnMoved => Json.obj("type" -> "columnMoved") ++ Json.toJsObject(cm)
+    case tm: TaskMoved   => Json.obj("type" -> "taskMoved") ++ Json.toJsObject(tm)
+  }
+}

backend/app/modules/ActorsModule.scala

@@ -0,0 +1,28 @@
+package modules
+
+import actors.ProjectActorRegistry
+import com.google.inject.name.Named
+import com.google.inject.{AbstractModule, Provides, Singleton}
+import org.apache.pekko.actor.{ActorRef, ActorSystem, Props}
+import play.api.Logger
+
+/**
+ * Constants for actor names to ensure consistency across the application.
+ */
+object ActorNames {
+  final val ProjectActorRegistry = "project-actor-registry"
+  final val ProjectActorPrefix = "project-actor-"
+}
+
+/**
+ * Module to provide actor instances.
+ */
+class ActorsModule extends AbstractModule {
+  @Provides
+  @Singleton
+  @Named(ActorNames.ProjectActorRegistry)
+  def provideProjectActorRegistry(system: ActorSystem): ActorRef = {
+    Logger("actors").info("Creating ProjectActorRegistry actor")
+    system.actorOf(Props[ProjectActorRegistry], ActorNames.ProjectActorRegistry)
+  }
+}