feat(messaging): finish implementation

backend/api_gateway/src/main/kotlin/com/linkedout/backend/controller/EmployerController.kt

@@ -1,21 +1,50 @@
 package com.linkedout.backend.controller
 
+import com.linkedout.backend.dto.messaging.SendMessageDto
 import com.linkedout.backend.model.Employer
+import com.linkedout.backend.model.Message
+import com.linkedout.backend.model.MessageChannel
 import com.linkedout.backend.service.EmployerService
+import com.linkedout.backend.service.MessageChannelService
+import com.linkedout.backend.service.MessageService
 import org.springframework.http.server.reactive.ServerHttpRequest
 import org.springframework.web.bind.annotation.GetMapping
 import org.springframework.web.bind.annotation.PathVariable
+import org.springframework.web.bind.annotation.PostMapping
+import org.springframework.web.bind.annotation.RequestBody
 import org.springframework.web.bind.annotation.RequestMapping
 import org.springframework.web.bind.annotation.RestController
 import reactor.core.publisher.Mono
+import java.security.Principal
 
 @RestController
 @RequestMapping("/api/v1/employers")
 class EmployerController(
-    private val employerService: EmployerService
+    private val employerService: EmployerService,
+    private val messageChannelService: MessageChannelService,
+    private val messageService: MessageService
 ) {
     @GetMapping("/{employerId}")
     open fun getEmployerById(request: ServerHttpRequest, @PathVariable employerId: String): Mono<Employer> {
         return Mono.just(employerService.findOne(request.id, employerId))
     }
+
+    @GetMapping("/{employerId}/messaging")
+    open fun getEmployerMessaging(
+        request: ServerHttpRequest,
+        principal: Principal,
+        @PathVariable employerId: String
+    ): Mono<MessageChannel> {
+        return Mono.just(messageChannelService.findOneChannelOfUserWithEmployer(request.id, principal.name, employerId))
+    }
+
+    @PostMapping("/{employerId}/messaging")
+    open fun sendMessage(
+        request: ServerHttpRequest,
+        principal: Principal,
+        @PathVariable employerId: String,
+        @RequestBody body: SendMessageDto
+    ): Mono<Message> {
+        return Mono.just(messageService.sendMessageToEmployer(request.id, principal.name, employerId, body.content))
+    }
 }

backend/api_gateway/src/main/kotlin/com/linkedout/backend/controller/MessagingController.kt

@@ -1,12 +1,15 @@
 package com.linkedout.backend.controller
 
+import com.linkedout.backend.dto.messaging.SendMessageDto
 import com.linkedout.backend.model.Message
 import com.linkedout.backend.model.MessageChannel
 import com.linkedout.backend.service.MessageChannelService
 import com.linkedout.backend.service.MessageService
 import org.springframework.http.server.reactive.ServerHttpRequest
 import org.springframework.web.bind.annotation.GetMapping
 import org.springframework.web.bind.annotation.PathVariable
+import org.springframework.web.bind.annotation.PostMapping
+import org.springframework.web.bind.annotation.RequestBody
 import org.springframework.web.bind.annotation.RequestMapping
 import org.springframework.web.bind.annotation.RestController
 import reactor.core.publisher.Flux
@@ -41,4 +44,14 @@ class MessagingController(
     ): Flux<Message> {
         return Flux.fromIterable(messageService.findAllMessagesOfUserInChannel(request.id, principal.name, channelId))
     }
+
+    @PostMapping("/{channelId}/messages")
+    open fun sendMessage(
+        request: ServerHttpRequest,
+        principal: Principal,
+        @PathVariable channelId: String,
+        @RequestBody body: SendMessageDto
+    ): Mono<Message> {
+        return Mono.just(messageService.sendMessage(request.id, principal.name, channelId, body.content))
+    }
 }

backend/api_gateway/src/main/kotlin/com/linkedout/backend/dto/messaging/SendMessageDto.kt

@@ -0,0 +1,5 @@
+package com.linkedout.backend.dto.messaging
+
+data class SendMessageDto(
+    val content: String
+)

backend/api_gateway/src/main/kotlin/com/linkedout/backend/service/MessageChannelService.kt

@@ -14,7 +14,8 @@ class MessageChannelService(
     private val natsService: NatsService,
     private val employerService: EmployerService,
     @Value("\${app.services.messageChannels.subjects.findAllOfUser}") private val findAllOfUsersSubject: String,
-    @Value("\${app.services.messageChannels.subjects.findOneOfUser}") private val findOneOfUserSubject: String
+    @Value("\${app.services.messageChannels.subjects.findOneOfUser}") private val findOneOfUserSubject: String,
+    @Value("\${app.services.messageChannels.subjects.findOneOfUserWithEmployer}") private val findOneOfUserWithEmployerSubject: String
 ) {
     fun findAllChannelsOfUser(requestId: String, userId: String): List<MessageChannel> {
         // Request message channels from the messaging service
@@ -75,4 +76,33 @@ class MessageChannelService(
             getUserMessageChannelByIdResponse.messageChannel.lastMessage
         )
     }
+
+    fun findOneChannelOfUserWithEmployer(requestId: String, userId: String, employerId: String): MessageChannel {
+        // Request message channel from the messaging service
+        val request = RequestResponseFactory.newRequest(requestId)
+            .setGetUserMessageChannelWithEmployerRequest(
+                Messaging.GetUserMessageChannelWithEmployerRequest.newBuilder()
+                    .setUserId(userId)
+                    .setEmployerId(employerId)
+            )
+            .build()
+
+        val response = natsService.requestWithReply(findOneOfUserWithEmployerSubject, request)
+
+        // Handle the response
+        if (!response.hasGetUserMessageChannelWithEmployerResponse()) {
+            throw Exception("Invalid response")
+        }
+
+        val getUserMessageChannelByIdResponse = response.getUserMessageChannelWithEmployerResponse
+
+        // Get the employer from the employer service
+        val employer = employerService.findOne(requestId, getUserMessageChannelByIdResponse.messageChannel.employerId)
+
+        return MessageChannel(
+            getUserMessageChannelByIdResponse.messageChannel.id,
+            employer,
+            getUserMessageChannelByIdResponse.messageChannel.lastMessage
+        )
+    }
 }

backend/api_gateway/src/main/kotlin/com/linkedout/backend/service/MessageService.kt

@@ -3,6 +3,7 @@ package com.linkedout.backend.service
 import com.linkedout.backend.model.Message
 import com.linkedout.common.service.NatsService
 import com.linkedout.common.utils.RequestResponseFactory
+import com.linkedout.proto.models.MessageOuterClass
 import com.linkedout.proto.services.Messaging
 import org.springframework.beans.factory.annotation.Value
 import org.springframework.stereotype.Service
@@ -12,7 +13,9 @@ import java.util.*
 @Service
 class MessageService(
     private val natsService: NatsService,
-    @Value("\${app.services.messages.subjects.findAllOfUserInChannel}") private val findAllOfUserInChannelSubject: String
+    @Value("\${app.services.messages.subjects.findAllOfUserInChannel}") private val findAllOfUserInChannelSubject: String,
+    @Value("\${app.services.messages.subjects.sendMessage}") private val sendMessageSubject: String,
+    @Value("\${app.services.messages.subjects.sendMessageToEmployer}") private val sendMessageToEmployerSubject: String
 ) {
     fun findAllMessagesOfUserInChannel(requestId: String, userId: String, channelId: String): List<Message> {
         // Request messages from the messaging service
@@ -46,4 +49,67 @@ class MessageService(
             )
         }
     }
+
+    fun sendMessage(requestId: String, userId: String, channelId: String, content: String): Message {
+        // Send the message using the messaging service
+        val request = RequestResponseFactory.newRequest(requestId)
+            .setSendMessageRequest(
+                Messaging.SendMessageRequest.newBuilder()
+                    .setUserId(userId)
+                    .setMessageChannelId(channelId)
+                    .setContent(content)
+                    .setDirection(MessageOuterClass.Message.Direction.ToEmployer)
+            )
+            .build()
+
+        val response = natsService.requestWithReply(sendMessageSubject, request)
+
+        // Handle the response
+        if (!response.hasSendMessageResponse()) {
+            throw Exception("Invalid response")
+        }
+
+        val sendMessageResponse = response.sendMessageResponse
+        val sentAt = Date(sendMessageResponse.message.sentAt)
+
+        return Message(
+            sendMessageResponse.message.id,
+            sendMessageResponse.messageChannelId,
+            sendMessageResponse.employerId,
+            sendMessageResponse.message.directionValue,
+            DateTimeFormatter.ISO_INSTANT.format(sentAt.toInstant()),
+            sendMessageResponse.message.content
+        )
+    }
+
+    fun sendMessageToEmployer(requestId: String, userId: String, employerId: String, content: String): Message {
+        // Send the message using the messaging service
+        val request = RequestResponseFactory.newRequest(requestId)
+            .setSendMessageToEmployerRequest(
+                Messaging.SendMessageToEmployerRequest.newBuilder()
+                    .setUserId(userId)
+                    .setEmployerId(employerId)
+                    .setContent(content)
+            )
+            .build()
+
+        val response = natsService.requestWithReply(sendMessageToEmployerSubject, request)
+
+        // Handle the response
+        if (!response.hasSendMessageToEmployerResponse()) {
+            throw Exception("Invalid response")
+        }
+
+        val sendMessageToEmployerResponse = response.sendMessageToEmployerResponse
+        val sentAt = Date(sendMessageToEmployerResponse.message.sentAt)
+
+        return Message(
+            sendMessageToEmployerResponse.message.id,
+            sendMessageToEmployerResponse.messageChannelId,
+            sendMessageToEmployerResponse.employerId,
+            sendMessageToEmployerResponse.message.directionValue,
+            DateTimeFormatter.ISO_INSTANT.format(sentAt.toInstant()),
+            sendMessageToEmployerResponse.message.content
+        )
+    }
 }

backend/api_gateway/src/main/resources/application.yml

@@ -26,9 +26,12 @@ app:
       subjects:
         findAllOfUser: messaging.findAllChannelsOfUser
         findOneOfUser: messaging.findOneChannelOfUser
+        findOneOfUserWithEmployer: messaging.findOneChannelOfUserWithEmployer
     messages:
       subjects:
         findAllOfUserInChannel: messaging.findAllMessagesOfUserInChannel
+        sendMessage: messaging.sendMessage
+        sendMessageToEmployer: messaging.sendMessageToEmployer
     employer:
       subjects:
         findOne: employer.findOne

backend/messaging/src/main/kotlin/com/linkedout/messaging/function/messageChannels/GetChannelOfUser.kt

@@ -18,11 +18,10 @@ class GetChannelOfUser(private val messageChannelService: MessageChannelService)
         val request = t.getUserMessageChannelRequest
         val responseMono = messageChannelService.findOneWithSeasonworkerId(UUID.fromString(request.userId), UUID.fromString(request.messageChannelId))
             .map { messageChannel ->
-                // TODO: Get the last message
                 MessageChannelOuterClass.MessageChannel.newBuilder()
                     .setId(messageChannel.id.toString())
                     .setEmployerId(messageChannel.employerId.toString())
-                    .setLastMessage("<TODO>")
+                    .setLastMessage(messageChannel.lastMessage ?: "")
                     .build()
             }
             .map { messageChannel ->

backend/messaging/src/main/kotlin/com/linkedout/messaging/function/messageChannels/GetChannelOfUserWithEmployer.kt

@@ -0,0 +1,45 @@
+package com.linkedout.messaging.function.messageChannels
+
+import com.linkedout.common.utils.RequestResponseFactory
+import com.linkedout.messaging.service.MessageChannelService
+import com.linkedout.proto.RequestOuterClass.Request
+import com.linkedout.proto.ResponseOuterClass.Response
+import com.linkedout.proto.models.MessageChannelOuterClass
+import com.linkedout.proto.services.Messaging
+import org.springframework.http.HttpStatus
+import org.springframework.stereotype.Component
+import java.util.*
+import java.util.function.Function
+
+@Component
+class GetChannelOfUserWithEmployer(private val messageChannelService: MessageChannelService) : Function<Request, Response> {
+    override fun apply(t: Request): Response {
+        // Get the message channel from the database
+        val request = t.getUserMessageChannelWithEmployerRequest
+        val responseMono = messageChannelService.findOneWithSeasonworkerIdAndEmployerId(UUID.fromString(request.userId), UUID.fromString(request.employerId))
+            .map { messageChannel ->
+                MessageChannelOuterClass.MessageChannel.newBuilder()
+                    .setId(messageChannel.id.toString())
+                    .setEmployerId(messageChannel.employerId.toString())
+                    .setLastMessage(messageChannel.lastMessage ?: "")
+                    .build()
+            }
+            .map { messageChannel ->
+                Messaging.GetUserMessageChannelWithEmployerResponse.newBuilder()
+                    .setMessageChannel(messageChannel)
+                    .build()
+            }
+
+        // Block until the response is ready
+        val response = try {
+            responseMono.block()
+        } catch (e: Exception) {
+            return RequestResponseFactory.newFailedResponse(e.message ?: "Unknown error").build()
+        }
+            ?: return RequestResponseFactory.newFailedResponse("Message channel not found", HttpStatus.NOT_FOUND).build()
+
+        return RequestResponseFactory.newSuccessfulResponse()
+            .setGetUserMessageChannelWithEmployerResponse(response)
+            .build()
+    }
+}

backend/messaging/src/main/kotlin/com/linkedout/messaging/function/messageChannels/GetChannelsOfUser.kt

@@ -16,11 +16,10 @@ class GetChannelsOfUser(private val messageChannelService: MessageChannelService
         // Get the message channels from the database
         val responseMono = messageChannelService.findAllWithSeasonworkerId(UUID.fromString(t.getUserMessageChannelsRequest.userId))
             .map { messageChannel ->
-                // TODO: Get the last message
                 MessageChannel.newBuilder()
                     .setId(messageChannel.id.toString())
                     .setEmployerId(messageChannel.employerId.toString())
-                    .setLastMessage("<TODO>")
+                    .setLastMessage(messageChannel.lastMessage ?: "")
                     .build()
             }
             .reduce(GetUserMessageChannelsResponse.newBuilder()) { builder, messageChannel ->

backend/messaging/src/main/kotlin/com/linkedout/messaging/function/messages/GetMessagesOfUser.kt

@@ -3,6 +3,7 @@ package com.linkedout.messaging.function.messages
 import com.linkedout.common.utils.RequestResponseFactory
 import com.linkedout.messaging.service.MessageChannelService
 import com.linkedout.messaging.service.MessageService
+import com.linkedout.messaging.utils.MessageDirection
 import com.linkedout.proto.RequestOuterClass.Request
 import com.linkedout.proto.ResponseOuterClass.Response
 import com.linkedout.proto.models.MessageOuterClass.Message
@@ -37,7 +38,7 @@ class GetMessagesOfUser(
             .map { message ->
                 Message.newBuilder()
                     .setId(message.id.toString())
-                    .setDirectionValue(message.direction)
+                    .setDirection(MessageDirection.toProto(message.direction))
                     .setSentAt(message.created.toEpochSecond(ZoneOffset.UTC) * 1000)
                     .setContent(message.message)
                     .build()

backend/messaging/src/main/kotlin/com/linkedout/messaging/function/messages/SendMessage.kt

@@ -0,0 +1,71 @@
+package com.linkedout.messaging.function.messages
+
+import com.linkedout.common.utils.RequestResponseFactory
+import com.linkedout.messaging.service.MessageChannelService
+import com.linkedout.messaging.service.MessageService
+import com.linkedout.messaging.utils.MessageDirection
+import com.linkedout.proto.RequestOuterClass.Request
+import com.linkedout.proto.ResponseOuterClass.Response
+import com.linkedout.proto.models.MessageOuterClass
+import com.linkedout.proto.services.Messaging.SendMessageResponse
+import org.springframework.http.HttpStatus
+import org.springframework.stereotype.Component
+import java.time.ZoneOffset
+import java.util.*
+import java.util.function.Function
+
+@Component
+class SendMessage(
+    private val messageService: MessageService,
+    private val messageChannelService: MessageChannelService
+) : Function<Request, Response> {
+    override fun apply(t: Request): Response {
+        // Extract the request
+        val request = t.sendMessageRequest
+        val userId = UUID.fromString(request.userId)
+        val messageChannelId = UUID.fromString(request.messageChannelId)
+
+        // Get the message channel from the database
+        val messageChannel = try {
+            messageChannelService.findOneWithSeasonworkerId(userId, messageChannelId)
+                .block()
+        } catch (e: Exception) {
+            return RequestResponseFactory.newFailedResponse(e.message ?: "Unknown error").build()
+        }
+            ?: return RequestResponseFactory.newFailedResponse("Message channel not found").build()
+
+        // Insert the message into the database
+        val responseMono = try {
+            messageService.saveMessage(userId, messageChannelId, request.content, MessageDirection.fromProto(request.direction))
+                .map { message ->
+                    MessageOuterClass.Message.newBuilder()
+                        .setId(message.id.toString())
+                        .setDirection(MessageDirection.toProto(message.direction))
+                        .setSentAt(message.created.toEpochSecond(ZoneOffset.UTC) * 1000)
+                        .setContent(message.message)
+                        .build()
+                }
+                .map { builder ->
+                    SendMessageResponse.newBuilder()
+                        .setMessageChannelId(messageChannelId.toString())
+                        .setEmployerId(messageChannel.employerId.toString())
+                        .setMessage(builder)
+                        .build()
+                }
+        } catch (e: Exception) {
+            return RequestResponseFactory.newFailedResponse(e.message ?: "Unknown error").build()
+        }
+
+        // Block until the response is ready
+        val response = try {
+            responseMono.block()
+        } catch (e: Exception) {
+            return RequestResponseFactory.newFailedResponse(e.message ?: "Unknown error").build()
+        }
+            ?: return RequestResponseFactory.newFailedResponse("Unable to send the message", HttpStatus.INTERNAL_SERVER_ERROR).build()
+
+        return RequestResponseFactory.newSuccessfulResponse()
+            .setSendMessageResponse(response)
+            .build()
+    }
+}