From 3d6d513ba97ef599dab88bf8f2af488f3abed252 Mon Sep 17 00:00:00 2001
From: Abhinav Raut <abhinav.raut@zerodha.com>
Date: Mon, 29 Jul 2024 04:03:19 +0530
Subject: [PATCH] more commits.

---
 cmd/handlers.go                               |  3 -
 .../conversation/ConversationSideBar.vue      |  1 -
 .../src/components/message/MessageList.vue    | 33 +++++---
 frontend/src/stores/conversation.js           | 18 +++--
 frontend/src/websocket.js                     |  3 -
 internal/conversation/conversation.go         | 80 +++++++++----------
 internal/conversation/message.go              |  5 +-
 internal/conversation/queries.sql             |  2 +-
 internal/ws/ws.go                             |  4 +-
 9 files changed, 80 insertions(+), 69 deletions(-)

diff --git a/cmd/handlers.go b/cmd/handlers.go
index ccbb4448..10dffcfd 100644
--- a/cmd/handlers.go
+++ b/cmd/handlers.go
@@ -21,7 +21,6 @@ func initHandlers(g *fastglue.Fastglue, hub *ws.Hub) {
 	g.GET("/api/conversations/all", auth(handleGetAllConversations, "conversation:all"))
 	g.GET("/api/conversations/team", auth(handleGetTeamConversations, "conversation:team"))
 	g.GET("/api/conversations/assigned", auth(handleGetAssignedConversations, "conversation:assigned"))
-
 	g.GET("/api/conversations/{uuid}", auth(handleGetConversation))
 	g.GET("/api/conversations/{uuid}/participants", auth(handleGetConversationParticipants))
 	g.PUT("/api/conversations/{uuid}/last-seen", auth(handleUpdateAssigneeLastSeen))
@@ -30,8 +29,6 @@ func initHandlers(g *fastglue.Fastglue, hub *ws.Hub) {
 	g.PUT("/api/conversations/{uuid}/priority", auth(handleUpdatePriority, "conversation:edit_priority"))
 	g.PUT("/api/conversations/{uuid}/status", auth(handleUpdateStatus, "conversation:edit_status"))
 	g.POST("/api/conversations/{uuid}/tags", auth(handleAddConversationTags))
-
-	// Message.
 	g.GET("/api/conversations/{uuid}/messages", auth(handleGetMessages))
 	g.GET("/api/message/{uuid}/retry", auth(handleRetryMessage))
 	g.GET("/api/message/{uuid}", auth(handleGetMessage))
diff --git a/frontend/src/components/conversation/ConversationSideBar.vue b/frontend/src/components/conversation/ConversationSideBar.vue
index bfbd75cd..5503d4ba 100644
--- a/frontend/src/components/conversation/ConversationSideBar.vue
+++ b/frontend/src/components/conversation/ConversationSideBar.vue
@@ -41,7 +41,6 @@
                       ? (() => {
                           console.log(' ->', conversationStore.conversation.data.assigned_user_id , 'agents ', agents)
                           const agent = agents.find(agent => agent.id === conversationStore.conversation.data.assigned_user_id);
-                          console.log('Foud agent ', agent)
                           return agent ? `${agent.first_name} ${agent.last_name}` : "Select agent...";
                         })()
                       : "Select agent..."
diff --git a/frontend/src/components/message/MessageList.vue b/frontend/src/components/message/MessageList.vue
index 00c4934e..2462f372 100644
--- a/frontend/src/components/message/MessageList.vue
+++ b/frontend/src/components/message/MessageList.vue
@@ -1,9 +1,13 @@
 <template>
     <div ref="threadEl" class="overflow-y-scroll">
-        <div class="text-center mt-3" v-if="conversationStore.messages.hasMore">
-            <Button variant="secondary" @click="conversationStore.fetchNextMessages">Fetch more</Button>
+        <div class="text-center mt-3" v-if="conversationStore.messages.hasMore && !conversationStore.messages.loading">
+            <Button variant="ghost" @click="conversationStore.fetchNextMessages">
+                <RefreshCw size="17" class="mr-2" />
+                Load more
+            </Button>
         </div>
-        <div v-for="message in conversationStore.sortedMessages" :key="message.uuid" :class="message.type === 'activity' ? 'm-4' : 'm-6'">
+        <div v-for="message in conversationStore.sortedMessages" :key="message.uuid"
+            :class="message.type === 'activity' ? 'm-4' : 'm-6'">
             <div v-if="!message.private">
                 <ContactMessageBubble :message="message" v-if="message.type === 'incoming'" />
                 <AgentMessageBubble :message="message" v-if="message.type === 'outgoing'" />
@@ -19,30 +23,37 @@
 </template>
 
 <script setup>
-import { ref, onMounted } from 'vue';
+import { ref, onMounted, watch } from 'vue';
 import ContactMessageBubble from "./ContactMessageBubble.vue";
 import ActivityMessageBubble from "./ActivityMessageBubble.vue";
 import AgentMessageBubble from "./AgentMessageBubble.vue";
 import { useConversationStore } from '@/stores/conversation'
 import { Button } from '@/components/ui/button';
+import { RefreshCw } from 'lucide-vue-next';
 
 const conversationStore = useConversationStore()
 const threadEl = ref(null);
 
 
 const scrollToBottom = () => {
-    const thread = threadEl.value;
-    if (thread) {
-        thread.scrollTop = thread.scrollHeight;
-    }
+    setTimeout(() => {
+        console.log("Scrolling bottom")
+        const thread = threadEl.value;
+        if (thread) {
+            thread.scrollTop = thread.scrollHeight;
+        }
+    }, 0)
 };
 
+
 onMounted(() => {
-    setTimeout(() => {
-        scrollToBottom();
-    }, 0);
+    scrollToBottom();
 });
 
+// On conversation change scroll to the bottom
+watch(() => conversationStore.conversation.data, () => {
+    scrollToBottom();
+});
 
 const isPrivateNote = (message) => {
     return message.type === "outgoing" && message.private;
diff --git a/frontend/src/stores/conversation.js b/frontend/src/stores/conversation.js
index 706a82be..c107b93d 100644
--- a/frontend/src/stores/conversation.js
+++ b/frontend/src/stores/conversation.js
@@ -58,15 +58,14 @@ export const useConversationStore = defineStore('conversation', () => {
         );
     })
 
-    // Computed property to sort message by created_at
+    // Computed property to sort messages by created_at
     const sortedMessages = computed(() => {
         if (!messages.data) {
             return [];
         }
-        return [...messages.data];
+        return [...messages.data].sort((a, b) => new Date(a.created_at) - new Date(b.created_at));
     });
 
-
     const getContactFullName = (uuid) => {
         const conv = conversations.data.find(conv => conv.uuid === uuid);
         return conv ? `${conv.first_name} ${conv.last_name}` : '';
@@ -84,7 +83,10 @@ export const useConversationStore = defineStore('conversation', () => {
         try {
             const resp = await api.getConversation(uuid);
             conversation.data = resp.data.data
+            // mark this conversation as read.
             markAsRead(uuid)
+            // reset messages state on new conversation fetch.
+            resetMessages()
         } catch (error) {
             conversation.errorMessage = handleHTTPError(error).message;
         } finally {
@@ -131,9 +133,6 @@ export const useConversationStore = defineStore('conversation', () => {
 
             // Add new messages to the messages data
             messages.data.unshift(...newMessages);
-
-            // Sort messages immediately after updating the state
-            messages.data.sort((a, b) => new Date(a.created_at) - new Date(b.created_at));
         } catch (error) {
             toast({
                 title: 'Could not fetch messages, Please try again.',
@@ -355,9 +354,16 @@ export const useConversationStore = defineStore('conversation', () => {
         conversation.errorMessage = ""
 
         // Reset messages state
+        resetMessages()
+    }
+
+    function resetMessages () {
         messages.data = []
         messages.loading = false
+        messages.page = 1
+        messages.hasMore = true
         messages.errorMessage = ""
+        seenMessageUUIDs = new Set()
     }
 
     return { conversations, conversation, messages, sortedConversations, sortedMessages, conversationUUIDExists, updateConversationProp, addNewConversation, getContactFullName, fetchParticipants, fetchNextMessages, fetchNextConversations, updateMessageProp, updateAssigneeLastSeen, updateMessageList, fetchConversation, fetchConversations, fetchMessages, upsertTags, updateAssignee, updatePriority, updateStatus, updateConversationList, $reset };
diff --git a/frontend/src/websocket.js b/frontend/src/websocket.js
index 876e38ba..4e870991 100644
--- a/frontend/src/websocket.js
+++ b/frontend/src/websocket.js
@@ -16,7 +16,6 @@ export function initWS() {
 
         // Connection opened event
         socket.addEventListener('open', function () {
-            console.log('WebSocket connection established')
             reconnectInterval = 1000 // Reset the reconnection interval
             if (reconnectTimeout) {
                 clearTimeout(reconnectTimeout) // Clear any existing reconnection timeout
@@ -26,7 +25,6 @@ export function initWS() {
 
         // Listen for messages from the server
         socket.addEventListener('message', function (e) {
-            console.log('Message from server:', e.data)
             if (e.data) {
                 let event = JSON.parse(e.data)
                 // TODO: move event type to consts.
@@ -57,7 +55,6 @@ export function initWS() {
 
         // Handle the connection close event
         socket.addEventListener('close', function (event) {
-            console.log('WebSocket connection closed:', event)
             if (!manualClose) {
                 reconnect()
             }
diff --git a/internal/conversation/conversation.go b/internal/conversation/conversation.go
index ef9ebe80..a7cd8ef9 100644
--- a/internal/conversation/conversation.go
+++ b/internal/conversation/conversation.go
@@ -138,14 +138,14 @@ func New(
 
 type queries struct {
 	// Conversation queries.
-	GetInReplyTo                       *sqlx.Stmt `query:"get-in-reply-to"`
+	GetLatestReceivedMessageSourceID   *sqlx.Stmt `query:"get-latest-received-message-source-id"`
 	GetToAddress                       *sqlx.Stmt `query:"get-to-address"`
 	GetConversationID                  *sqlx.Stmt `query:"get-conversation-id"`
 	GetConversationUUID                *sqlx.Stmt `query:"get-conversation-uuid"`
 	GetConversation                    *sqlx.Stmt `query:"get-conversation"`
 	GetRecentConversations             *sqlx.Stmt `query:"get-recent-conversations"`
 	GetUnassignedConversations         *sqlx.Stmt `query:"get-unassigned-conversations"`
-	GetConversations                   string     `query:"get-conversations-conversations"`
+	GetConversations                   string     `query:"get-conversations"`
 	GetConversationsUUIDs              string     `query:"get-conversations-uuids"`
 	GetConversationParticipants        *sqlx.Stmt `query:"get-conversation-participants"`
 	GetAssignedConversations           *sqlx.Stmt `query:"get-assigned-conversations"`
@@ -163,7 +163,7 @@ type queries struct {
 	AddConversationTag                 *sqlx.Stmt `query:"add-conversation-tag"`
 	DeleteConversationTags             *sqlx.Stmt `query:"delete-conversation-tags"`
 
-	// Message queries
+	// Message queries.
 	GetMessage           *sqlx.Stmt `query:"get-message"`
 	GetMessages          string     `query:"get-messages"`
 	GetPendingMessages   *sqlx.Stmt `query:"get-pending-messages"`
@@ -247,39 +247,6 @@ func (c *Manager) AddConversationParticipant(userID int, conversationUUID string
 	return nil
 }
 
-// UpdateConversationMeta updates the metadata of a conversation.
-func (c *Manager) UpdateConversationMeta(conversationID int, conversationUUID string, meta map[string]string) error {
-	metaJSON, err := json.Marshal(meta)
-	if err != nil {
-		c.lo.Error("error marshalling meta", "meta", meta, "error", err)
-		return err
-	}
-	if _, err := c.q.UpdateConversationMeta.Exec(conversationID, conversationUUID, metaJSON); err != nil {
-		c.lo.Error("error updating conversation meta", "error", "error")
-		return err
-	}
-	return nil
-}
-
-// UpdateConversationLastMessage updates the last message details in the conversation meta.
-func (c *Manager) UpdateConversationLastMessage(conversationID int, conversationUUID, lastMessage string, lastMessageAt time.Time) error {
-	return c.UpdateConversationMeta(conversationID, conversationUUID, map[string]string{
-		"last_message":    lastMessage,
-		"last_message_at": lastMessageAt.Format(time.RFC3339),
-	})
-}
-
-// UpdateConversationFirstReplyAt updates the first reply timestamp for a conversation.
-func (c *Manager) UpdateConversationFirstReplyAt(conversationUUID string, conversationID int, at time.Time) error {
-	if _, err := c.q.UpdateConversationFirstReplyAt.Exec(conversationID, at); err != nil {
-		c.lo.Error("error updating conversation first reply at", "error", err)
-		return err
-	}
-	// Send ws update.
-	c.wsHub.BroadcastConversationPropertyUpdate(conversationUUID, "first_reply_at", time.Now().Format(time.RFC3339))
-	return nil
-}
-
 // GetUnassignedConversations retrieves unassigned conversations.
 func (c *Manager) GetUnassignedConversations() ([]models.Conversation, error) {
 	var conv []models.Conversation
@@ -387,6 +354,39 @@ func (c *Manager) GetConversationUUIDs(userID, page, pageSize int, typ, filter s
 	return ids, nil
 }
 
+// UpdateConversationMeta updates the metadata of a conversation.
+func (c *Manager) UpdateConversationMeta(conversationID int, conversationUUID string, meta map[string]string) error {
+	metaJSON, err := json.Marshal(meta)
+	if err != nil {
+		c.lo.Error("error marshalling meta", "meta", meta, "error", err)
+		return err
+	}
+	if _, err := c.q.UpdateConversationMeta.Exec(conversationID, conversationUUID, metaJSON); err != nil {
+		c.lo.Error("error updating conversation meta", "error", "error")
+		return err
+	}
+	return nil
+}
+
+// UpdateConversationLastMessage updates the last message details in the conversation meta.
+func (c *Manager) UpdateConversationLastMessage(conversationID int, conversationUUID, lastMessage string, lastMessageAt time.Time) error {
+	return c.UpdateConversationMeta(conversationID, conversationUUID, map[string]string{
+		"last_message":    lastMessage,
+		"last_message_at": lastMessageAt.Format(time.RFC3339),
+	})
+}
+
+// UpdateConversationFirstReplyAt updates the first reply timestamp for a conversation.
+func (c *Manager) UpdateConversationFirstReplyAt(conversationUUID string, conversationID int, at time.Time) error {
+	if _, err := c.q.UpdateConversationFirstReplyAt.Exec(conversationID, at); err != nil {
+		c.lo.Error("error updating conversation first reply at", "error", err)
+		return err
+	}
+	// Broadcast update to all subscribers.
+	c.wsHub.BroadcastConversationPropertyUpdate(conversationUUID, "first_reply_at", at.Format(time.RFC3339))
+	return nil
+}
+
 // UpdateConversationUserAssignee sets the assignee of a conversation to a specifc user.
 func (c *Manager) UpdateConversationUserAssignee(uuid string, assigneeID int, actor umodels.User) error {
 	if err := c.UpdateAssignee(uuid, assigneeID, models.AssigneeTypeUser); err != nil {
@@ -529,7 +529,7 @@ func (c *Manager) generateConversationsListQuery(userID int, baseQuery, typ, ord
 	case models.AllConversations:
 		// No conditions.
 	default:
-		return "", nil, errors.New("invalid type of conversation")
+		return "", nil, fmt.Errorf("invalid conversation type %s", typ)
 	}
 
 	if filterClause, ok := models.ValidFilters[filter]; ok {
@@ -571,10 +571,10 @@ func (m *Manager) GetToAddress(conversationID int, channel string) ([]string, er
 	return addr, nil
 }
 
-// GetInReplyTo retrieves the `In-Reply-To` for given conversation.
-func (m *Manager) GetInReplyTo(conversationID int) (string, error) {
+// GetLatestReceivedMessageSourceID returns the last received message source ID.
+func (m *Manager) GetLatestReceivedMessageSourceID(conversationID int) (string, error) {
 	var out string
-	if err := m.q.GetInReplyTo.Get(&out, conversationID); err != nil {
+	if err := m.q.GetLatestReceivedMessageSourceID.Get(&out, conversationID); err != nil {
 		m.lo.Error("error fetching `in reply to`", "error", err, "conversation_id", conversationID)
 		return out, err
 	}
diff --git a/internal/conversation/message.go b/internal/conversation/message.go
index b443ce21..211ddc04 100644
--- a/internal/conversation/message.go
+++ b/internal/conversation/message.go
@@ -149,7 +149,7 @@ func (m *Manager) MessageDispatchWorker(ctx context.Context) {
 			// Get from, to addresses and inReplyTo.
 			message.From = inbox.FromAddress()
 			message.To, _ = m.GetToAddress(message.ConversationID, inbox.Channel())
-			message.InReplyTo, _ = m.GetInReplyTo(message.ConversationID)
+			message.InReplyTo, _ = m.GetLatestReceivedMessageSourceID(message.ConversationID)
 
 			// Send.
 			err = inbox.Send(message)
@@ -182,6 +182,7 @@ func (m *Manager) GetConversationMessages(conversationUUID string, page, pageSiz
 		messages = make([]models.Message, 0)
 		qArgs    []interface{}
 	)
+
 	qArgs = append(qArgs, conversationUUID)
 	query, qArgs, err := m.generateMessagesQuery(m.q.GetMessages, qArgs, page, pageSize)
 	if err != nil {
@@ -190,7 +191,7 @@ func (m *Manager) GetConversationMessages(conversationUUID string, page, pageSiz
 	}
 
 	tx, err := m.db.BeginTxx(context.Background(), nil)
-	defer tx.Commit()
+	defer tx.Rollback()
 	if err != nil {
 		m.lo.Error("error preparing get messages query", "error", err)
 		return messages, envelope.NewError(envelope.GeneralError, "Error fetching messages", nil)
diff --git a/internal/conversation/queries.sql b/internal/conversation/queries.sql
index 718dafbf..b8440489 100644
--- a/internal/conversation/queries.sql
+++ b/internal/conversation/queries.sql
@@ -228,7 +228,7 @@ WHERE conversation_id = (
 -- name: get-to-address
 SELECT cm.source_id from conversations c inner join contact_methods cm on cm.contact_id = c.contact_id where c.id = $1 and cm.source = $2;
 
--- name: get-in-reply-to
+-- name: get-latest-received-message-source-id
 SELECT source_id
 FROM messages
 WHERE conversation_id = $1 and status = 'received'
diff --git a/internal/ws/ws.go b/internal/ws/ws.go
index 33619b34..4f26464e 100644
--- a/internal/ws/ws.go
+++ b/internal/ws/ws.go
@@ -110,7 +110,7 @@ func (h *Hub) PushMessage(msg PushMessage) {
 }
 
 // BroadcastNewConversationMessage broadcasts a new conversation message.
-func (h *Hub) BroadcastNewConversationMessage(conversationUUID, trimmedMessage, messageUUID, lastMessageAt string, private bool) {
+func (h *Hub) BroadcastNewConversationMessage(conversationUUID, content, messageUUID, lastMessageAt string, private bool) {
 	userIDs, ok := h.conversationSubs[conversationUUID]
 	if !ok || len(userIDs) == 0 {
 		return
@@ -120,7 +120,7 @@ func (h *Hub) BroadcastNewConversationMessage(conversationUUID, trimmedMessage,
 		Type: models.MessageTypeNewMessage,
 		Data: map[string]interface{}{
 			"conversation_uuid": conversationUUID,
-			"last_message":      trimmedMessage,
+			"last_message":      content,
 			"uuid":              messageUUID,
 			"last_message_at":   lastMessageAt,
 			"private":           private,