refactor email inbox.

2025-11-03 05:23:48 +00:00 · 2024-06-26 01:43:17 +05:30
parent 2bbb161593
commit 6d027c92ab
22 changed files with 493 additions and 435 deletions
--- a/cmd/init.go
+++ b/cmd/init.go
@@ -16,8 +16,8 @@ import (
 	convtag "github.com/abhinavxd/artemis/internal/conversation/tag"
 	"github.com/abhinavxd/artemis/internal/inbox"
 	"github.com/abhinavxd/artemis/internal/inbox/channel/email"
+	"github.com/abhinavxd/artemis/internal/initz"
 	"github.com/abhinavxd/artemis/internal/message"
-	mmodels "github.com/abhinavxd/artemis/internal/message/models"
 	"github.com/abhinavxd/artemis/internal/rbac"
 	"github.com/abhinavxd/artemis/internal/tag"
 	"github.com/abhinavxd/artemis/internal/team"
@@ -170,7 +170,6 @@ func initContactManager(db *sqlx.DB, lo *logf.Logger) *contact.Manager {

 func initMessages(db *sqlx.DB,
 	lo *logf.Logger,
-	incomingMsgQ chan mmodels.IncomingMessage,
 	wsHub *ws.Hub,
 	userMgr *user.Manager,
 	teaMgr *team.Manager,
@@ -179,7 +178,7 @@ func initMessages(db *sqlx.DB,
 	conversationMgr *conversation.Manager,
 	inboxMgr *inbox.Manager,
 	automationEngine *automation.Engine) *message.Manager {
-	mgr, err := message.New(incomingMsgQ,
+	mgr, err := message.New(
 		wsHub,
 		userMgr,
 		teaMgr,
@@ -189,8 +188,10 @@ func initMessages(db *sqlx.DB,
 		conversationMgr,
 		automationEngine,
 		message.Opts{
-			DB: db,
-			Lo: lo,
+			DB:                   db,
+			Lo:                   lo,
+			OutgoingMsgQueueSize: ko.MustInt("message.outgoing_queue_size"),
+			IncomingMsgQueueSize: ko.MustInt("message.incoming_queue_size"),
 		})
 	if err != nil {
 		log.Fatalf("error initializing message manager: %v", err)
@@ -247,31 +248,12 @@ func initAttachmentsManager(db *sqlx.DB, lo *logf.Logger) *attachment.Manager {
 	return mgr
 }

-// initInboxManager initializes the inbox manager and the `active` inboxes.
-func initInboxManager(db *sqlx.DB, lo *logf.Logger, incomingMsgQ chan mmodels.IncomingMessage) *inbox.Manager {
-	mgr, err := inbox.New(lo, db, incomingMsgQ)
+// initInboxManager initializes the inbox manager without registering inboxes.
+func initInboxManager(db *sqlx.DB, lo *logf.Logger) *inbox.Manager {
+	mgr, err := inbox.New(lo, db)
 	if err != nil {
 		log.Fatalf("error initializing inbox manager: %v", err)
 	}
-
-	inboxRecords, err := mgr.GetActiveInboxes()
-	if err != nil {
-		log.Fatalf("error fetching active inboxes %v", err)
-	}
-
-	for _, inboxR := range inboxRecords {
-		switch inboxR.Channel {
-		case "email":
-			log.Printf("initializing `Email` inbox: %s", inboxR.Name)
-			inbox, err := initEmailInbox(inboxR)
-			if err != nil {
-				log.Fatalf("error initializing email inbox %v", err)
-			}
-			mgr.Register(inbox)
-		default:
-			log.Printf("WARNING: Unknown inbox channel: %s", inboxR.Name)
-		}
-	}
 	return mgr
 }

@@ -311,7 +293,7 @@ func initUserFilterMgr(db *sqlx.DB) *filterstore.Manager {
 }

 // initEmailInbox initializes the email inbox.
-func initEmailInbox(inboxRecord inbox.InboxRecord) (inbox.Inbox, error) {
+func initEmailInbox(inboxRecord inbox.InboxRecord, store inbox.MessageStore) (inbox.Inbox, error) {
 	var config email.Config

 	// Load JSON data into Koanf.
@@ -334,9 +316,10 @@ func initEmailInbox(inboxRecord inbox.InboxRecord) (inbox.Inbox, error) {
 	// Set from addr.
 	config.From = inboxRecord.From

-	inbox, err := email.New(email.Opts{
+	inbox, err := email.New(store, email.Opts{
 		ID:     inboxRecord.ID,
 		Config: config,
+		Lo:     initz.Logger(ko.MustString("app.log_level"), ko.MustString("app.env"), "email_inbox"),
 	})

 	if err != nil {
@@ -348,3 +331,25 @@ func initEmailInbox(inboxRecord inbox.InboxRecord) (inbox.Inbox, error) {

 	return inbox, nil
 }
+
+// registerInboxes registers the active inboxes with the inbox manager.
+func registerInboxes(mgr *inbox.Manager, store inbox.MessageStore) {
+	inboxRecords, err := mgr.GetActiveInboxes()
+	if err != nil {
+		log.Fatalf("error fetching active inboxes %v", err)
+	}
+
+	for _, inboxR := range inboxRecords {
+		switch inboxR.Channel {
+		case "email":
+			log.Printf("initializing `Email` inbox: %s", inboxR.Name)
+			inbox, err := initEmailInbox(inboxR, store)
+			if err != nil {
+				log.Fatalf("error initializing email inbox %v", err)
+			}
+			mgr.Register(inbox)
+		default:
+			log.Printf("WARNING: Unknown inbox channel: %s", inboxR.Name)
+		}
+	}
+}
--- a/cmd/main.go
+++ b/cmd/main.go
@@ -6,6 +6,7 @@ import (
 	"os"
 	"os/signal"
 	"syscall"
+	"time"

 	"github.com/abhinavxd/artemis/internal/attachment"
 	"github.com/abhinavxd/artemis/internal/cannedresp"
@@ -15,7 +16,6 @@ import (
 	"github.com/abhinavxd/artemis/internal/inbox"
 	"github.com/abhinavxd/artemis/internal/initz"
 	"github.com/abhinavxd/artemis/internal/message"
-	"github.com/abhinavxd/artemis/internal/message/models"
 	"github.com/abhinavxd/artemis/internal/rbac"
 	"github.com/abhinavxd/artemis/internal/tag"
 	"github.com/abhinavxd/artemis/internal/team"
@@ -60,29 +60,25 @@ func main() {
 	var (
 		shutdownCh = make(chan struct{})
 		ctx, stop  = signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGINT, syscall.SIGTERM)
-
-		// Incoming messages from all inboxes are pushed to this queue.
-		incomingMsgQ = make(chan models.IncomingMessage, ko.MustInt("message.incoming_queue_size"))
-
-		lo = initz.Logger(ko.MustString("app.log_level"), ko.MustString("app.env"), "artemis")
-		rd = initz.Redis(ko)
-		db = initz.DB(ko)
+		lo         = initz.Logger(ko.MustString("app.log_level"), ko.MustString("app.env"), "artemis")
+		rd         = initz.Redis(ko)
+		db         = initz.DB(ko)

 		wsHub              = ws.NewHub()
-		attachmentMgr      = initAttachmentsManager(db, &lo)
-		cntctMgr           = initContactManager(db, &lo)
-		inboxMgr           = initInboxManager(db, &lo, incomingMsgQ)
-		teamMgr            = initTeamMgr(db, &lo)
-		userMgr            = initUserDB(db, &lo)
-		conversationMgr    = initConversations(db, &lo)
-		automationEngine   = initAutomationEngine(db, &lo)
-		msgMgr             = initMessages(db, &lo, incomingMsgQ, wsHub, userMgr, teamMgr, cntctMgr, attachmentMgr, conversationMgr, inboxMgr, automationEngine)
-		autoAssignerEngine = initAutoAssignmentEngine(teamMgr, conversationMgr, msgMgr, &lo)
+		attachmentMgr      = initAttachmentsManager(db, lo)
+		cntctMgr           = initContactManager(db, lo)
+		inboxMgr           = initInboxManager(db, lo)
+		teamMgr            = initTeamMgr(db, lo)
+		userMgr            = initUserDB(db, lo)
+		conversationMgr    = initConversations(db, lo)
+		automationEngine   = initAutomationEngine(db, lo)
+		msgMgr             = initMessages(db, lo, wsHub, userMgr, teamMgr, cntctMgr, attachmentMgr, conversationMgr, inboxMgr, automationEngine)
+		autoAssignerEngine = initAutoAssignmentEngine(teamMgr, conversationMgr, msgMgr, lo)
 	)

 	// Init the app
 	var app = &App{
-		lo:                  &lo,
+		lo:                  lo,
 		cntctMgr:            cntctMgr,
 		inboxMgr:            inboxMgr,
 		userMgr:             userMgr,
@@ -92,25 +88,28 @@ func main() {
 		msgMgr:              msgMgr,
 		constants:           initConstants(),
 		rbac:                initRBACEngine(db),
-		tagMgr:              initTags(db, &lo),
+		tagMgr:              initTags(db, lo),
 		userFilterMgr:       initUserFilterMgr(db),
 		sessMgr:             initSessionManager(rd),
-		cannedRespMgr:       initCannedResponse(db, &lo),
-		conversationTagsMgr: initConversationTags(db, &lo),
+		cannedRespMgr:       initCannedResponse(db, lo),
+		conversationTagsMgr: initConversationTags(db, lo),
 	}

-	automationEngine.SetMsgRecorder(app.msgMgr)
+	// Register all inboxes with the inbox manager.
+	registerInboxes(inboxMgr, msgMgr)
+
+	automationEngine.SetMsgRecorder(msgMgr)
 	automationEngine.SetConvUpdater(conversationMgr)

 	// Start receivers for all active inboxes.
-	inboxMgr.Receive()
+	inboxMgr.Receive(ctx)

 	// Start inserting incoming msgs and dispatch pending outgoing messages.
 	go app.msgMgr.StartDBInserts(ctx, ko.MustInt("message.reader_concurrency"))
 	go app.msgMgr.StartDispatcher(ctx, ko.MustInt("message.dispatch_concurrency"), ko.MustDuration("message.dispatch_read_interval"))

 	// Start automation rule engine.
-	go automationEngine.Serve()
+	go automationEngine.Serve(ctx)

 	// Start conversation auto assigner engine.
 	go autoAssignerEngine.Serve(ctx, ko.MustDuration("autoassigner.assign_interval"))
@@ -133,17 +132,25 @@ func main() {
 		ReadBufferSize:       ko.MustInt("app.server.max_body_size"),
 	}

-	// Goroutine for handling interrupt signals & gracefully shutting down the server.
+	// Handling graceful shutdown with a delay
 	go func() {
+		// Wait for the interruption signal
 		<-ctx.Done()
+
+		log.Printf("\x1b[%dm%s\x1b[0m", 31, "Shutting down the server please wait...")
+
+		// Additional grace period before triggering shutdown
+		time.Sleep(7 * time.Second)
+
+		// Signal to shutdown the server
 		shutdownCh <- struct{}{}
 		stop()
 	}()

-	// Start the HTTP server.
+	// Starting the server and waiting for the shutdown signal
 	log.Printf("🚀 server listening on %s %s", ko.String("app.server.address"), ko.String("app.server.socket"))
 	if err := g.ListenServeAndWaitGracefully(ko.String("app.server.address"), ko.String("server.socket"), s, shutdownCh); err != nil {
 		log.Fatalf("error starting frontend server: %v", err)
 	}
-	log.Println("bye")
+	log.Println("Server shutdown completed")
 }
--- a/frontend/src/assets/styles/main.scss
+++ b/frontend/src/assets/styles/main.scss
@@ -2,8 +2,6 @@
@tailwind components;
@tailwind utilities;

-
-
 // App default font-size.
 // Default: 16px, 15px looked very wide.
 :root {
@@ -24,7 +22,6 @@

 $editorContainerId: 'editor-container';

-// --primary: 217 88.1% 60.4%;

 // Theme.

@@ -39,7 +36,7 @@ $editorContainerId: 'editor-container';
    --popover: 0 0% 100%;
    --popover-foreground: 222.2 84% 4.9%;
  
-    --primary: 221.2 83.2% 53.3%;
+    --primary: 222.2 47.4% 11.2%;
    --primary-foreground: 210 40% 98%;
  
    --secondary: 210 40% 96.1%;
@@ -56,8 +53,8 @@ $editorContainerId: 'editor-container';
  
    --border:214.3 31.8% 91.4%;
    --input:214.3 31.8% 91.4%;
-    --ring:221.2 83.2% 53.3%;
-    --radius: 1rem;
+    --ring:222.2 84% 4.9%;
+    --radius: 0.25rem;
  }
   
  .dark {
@@ -70,7 +67,7 @@ $editorContainerId: 'editor-container';
    --popover:222.2 84% 4.9%;
    --popover-foreground:210 40% 98%;
  
-    --primary:217.2 91.2% 59.8%;
+    --primary:210 40% 98%;
    --primary-foreground:222.2 47.4% 11.2%;
  
    --secondary:217.2 32.6% 17.5%;
@@ -87,18 +84,21 @@ $editorContainerId: 'editor-container';
  
    --border:217.2 32.6% 17.5%;
    --input:217.2 32.6% 17.5%;
-    --ring:224.3 76.3% 48%;
+    --ring:212.7 26.8% 83.9;
  }
 }

-@layer base {
-  * {
-    @apply border-border;
-  }
-  body {
-    @apply bg-background text-foreground;
-  }
-}
+
+
+
+// @layer base {
+//   * {
+//     @apply border-border;
+//   }
+//   body {
+//     @apply bg-background text-foreground;
+//   }
+// }

 // charts
@layer base {
@@ -164,10 +164,9 @@ $editorContainerId: 'editor-container';
  pb-3
  min-w-[30%] max-w-[70%]
  border
-  rounded-md
-  ;
+  rounded-xl;

-  box-shadow: 2px 2px 2px 0px rgba(0,0,0,0.1);
+  box-shadow: 1px 1px 1px 0px rgba(0, 0, 0, 0.1);

  // Making email tables fit.
  table {
@@ -189,10 +188,10 @@ $editorContainerId: 'editor-container';
  background: rgba(0, 0, 0, 0.6);
 }

-[class$='gmail_quote'] {
-  display: none !important;
-}
+// [class$='gmail_quote'] {
+//   display: none !important;
+// }

-blockquote {
-  display: none !important;
-}
+// blockquote {
+//   display: none !important;
+// }
--- a/frontend/src/components/NavBar.vue
+++ b/frontend/src/components/NavBar.vue
@@ -19,7 +19,7 @@ defineProps({
 const route = useRoute();

 const getButtonVariant = (title) => {
-    return route.name === title.toLowerCase() ? "secondary" : "ghost"
+    return route.name === title.toLowerCase() ? "" : "ghost"
 }

 </script>
@@ -64,7 +64,7 @@ const getButtonVariant = (title) => {
                    <Icon :icon="link.icon" class="mr-2 size-5" />
                    {{ link.title }}
                    <span v-if="link.label" :class="cn(
-                        'ml-auto',
+                        'ml-',
                        link.variant === getButtonVariant(link.title)
                        && 'text-background dark:text-white',
                    )">
--- a/frontend/src/components/conversation/ConversationSideBar.vue
+++ b/frontend/src/components/conversation/ConversationSideBar.vue
@@ -220,11 +220,17 @@
          </div>


-          <!-- <div class="flex flex-col gap-1 mb-5">
-            <p class="font-medium">SLA</p>
-            <p v-if="conversationStore.conversation.data.sla_remaining">48 hours remaining</p>
-          </div> -->
-
+          <div class="flex flex-col gap-1 mb-5">
+            <p class="font-medium">
+              First reply at
+            </p>
+            <p v-if="conversationStore.conversation.data.first_reply_at">
+              {{ format(conversationStore.conversation.data.first_reply_at, "PPpp") }}
+            </p>
+            <p v-else>
+              -
+            </p>
+          </div>
        </AccordionContent>
      </AccordionItem>
    </Accordion>
--- a/frontend/src/components/conversation/ReplyBox.vue
+++ b/frontend/src/components/conversation/ReplyBox.vue
@@ -26,13 +26,13 @@ onMounted(() => {
    cannedResponsesStore.fetchAll()
 })

-const sendMessage = async (message) => {
+const sendMessage = (message) => {
    api.sendMessage(conversationStore.conversation.data.uuid, {
        private: message.private,
        message: message.html,
        attachments: JSON.stringify(message.attachments),
    })
-    await api.updateAssigneeLastSeen(conversationStore.conversation.data.uuid)
+    api.updateAssigneeLastSeen(conversationStore.conversation.data.uuid)
 }

 </script>
--- a/frontend/src/components/conversationlist/ConversationList.vue
+++ b/frontend/src/components/conversationlist/ConversationList.vue
@@ -31,15 +31,16 @@
                </div>
            </div>

-            <div class="relative mx-auto my-3">
+            <!-- Search -->
+            <!-- <div class="relative mx-auto my-3">
                <Input id="search" type="text" placeholder="Search message or reference number"
                    class="pl-10 bg-[#F0F2F5]" />
                <span class="absolute start-1 inset-y-0 flex items-center justify-center px-2">
                    <Search class="size-6 text-muted-foreground" />
                </span>
-            </div>
+            </div> -->

-            <div class="flex justify-between">
+            <div class="flex justify-between mt-5">
                <Tabs v-model:model-value="conversationType">
                    <TabsList class="w-full flex justify-evenly">
                        <TabsTrigger value="assigned" class="w-full">
--- a/frontend/src/components/conversationlist/ConversationListItem.vue
+++ b/frontend/src/components/conversationlist/ConversationListItem.vue
@@ -19,7 +19,7 @@
                        {{ conversation.inbox_name }}
                    </p>
                    <p class="text-base font-normal">
-                        {{ conversationStore.getContactFullName (conversation.uuid)}}
+                        {{ conversationStore.getContactFullName(conversation.uuid) }}
                    </p>
                </div>
                <div>
@@ -30,8 +30,8 @@
            </div>
            <div class="pt-2 pr-3">
                <div class="flex justify-between">
-                    <p class="text-gray-800 max-w-xs text-sm dark:text-white text-ellipsis">
-                        {{ conversation.last_message }}
+                    <p class="text-gray-800 max-w-xs text-sm dark:text-white text-ellipsis flex gap-1">
+                        <CheckCheck :size="14" /> {{ conversation.last_message }}
                    </p>
                    <div class="flex items-center justify-center bg-green-500 rounded-full w-[20px] h-[20px]"
                        v-if="conversation.unread_message_count > 0">
@@ -50,7 +50,7 @@ import { useRouter } from 'vue-router'
 import { useConversationStore } from '@/stores/conversation'
 import { formatTime } from '@/utils/datetime'

-import { Mail } from 'lucide-vue-next'
+import { Mail, CheckCheck } from 'lucide-vue-next'
 import { Avatar, AvatarFallback, AvatarImage } from '@/components/ui/avatar'

 const router = useRouter()
--- a/frontend/src/components/dashboard/agent/CountCards.vue
+++ b/frontend/src/components/dashboard/agent/CountCards.vue
@@ -0,0 +1,34 @@
+<template>
+    <div class="flex mt-5 gap-x-7">
+        <Card class="w-1/6" v-for="(value, key) in counts" :key="key">
+            <CardHeader>
+                <CardTitle class="text-4xl">
+                    {{ value }}
+                </CardTitle>
+                <CardDescription>
+                    {{ labels[key] }}
+                </CardDescription>
+            </CardHeader>
+        </Card>
+    </div>
+</template>
+<script setup>
+import {
+    Card,
+    CardDescription,
+    CardHeader,
+    CardTitle,
+} from '@/components/ui/card'
+
+
+defineProps({
+    counts: {
+        type: Object,
+        required: true,
+    },
+    labels: {
+        type: Object,
+        required: true,
+    }
+})
+</script>
--- a/frontend/src/components/message/AgentMessageBubble.vue
+++ b/frontend/src/components/message/AgentMessageBubble.vue
@@ -26,8 +26,9 @@
                <MessageAttachmentPreview :attachments="message.attachments" />
                <Spinner v-if="message.status === 'pending'" />
                <div class="flex items-center space-x-2 mt-2">
-                    <span class="text-slate-500 capitalize text-xs" v-if="message.status != 'pending'">{{
-                        message.status }}</span>
+                    <!-- <span class="text-slate-500 capitalize text-xs" v-if="message.status != 'pending'">{{
+                        message.status }}</span> -->
+                    <CheckCheck :size="14"  v-if="message.status != 'pending'"/>
                    <RotateCcw size="10" @click="retryMessage(message)" class="cursor-pointer"
                        v-if="message.status === 'failed'"></RotateCcw>
                </div>
@@ -68,7 +69,7 @@ import {
    TooltipTrigger
 } from '@/components/ui/tooltip'
 import { Spinner } from '@/components/ui/spinner'
-import { RotateCcw } from 'lucide-vue-next';
+import { RotateCcw, CheckCheck } from 'lucide-vue-next';
 import { Avatar, AvatarFallback, AvatarImage } from '@/components/ui/avatar'
 import MessageAttachmentPreview from "@/components/attachment/MessageAttachmentPreview.vue"

--- a/frontend/src/views/DashboardView.vue
+++ b/frontend/src/views/DashboardView.vue
@@ -1,56 +1,37 @@
 <script setup>
 import { onMounted, ref } from 'vue';
 import { useUserStore } from '@/stores/user'
-import { LineChart } from '@/components/ui/chart-line'
-import { format } from 'date-fns';
+import { format } from 'date-fns'
 import api from '@/api';
-
-import {
-    Card,
-    CardContent,
-    CardDescription,
-    CardFooter,
-    CardHeader,
-    CardTitle,
-} from '@/components/ui/card'
 import { useToast } from '@/components/ui/toast/use-toast'

+import CountCards from '@/components/dashboard/agent/CountCards.vue'
+
 const { toast } = useToast()
 const counts = ref({})
-const data = [
-    {
-        'year': 1970,
-        'Export Growth Rate': 2.04,
-        'Import Growth Rate': 1.53,
-    },
-    {
-        'year': 1971,
-        'Export Growth Rate': 1.96,
-        'Import Growth Rate': 1.58,
-    },
-]
 const userStore = useUserStore()
-
-onMounted(() => {
-    api.getAssigneeStats().then((resp) => {
-        counts.value = resp.data.data;
-    }).catch((err) => {
-        toast({
-            title: 'Uh oh! Something went wrong.',
-            description: err.response.data.message,
-            variant: 'destructive',
-        });
-    });
-})
-
-const labels = {
+const agentCountCardsLabels = {
    total_assigned: "Total Assigned",
    unresolved_count: "Unresolved",
    awaiting_response_count: "Awaiting Response",
    created_today_count: "Created Today"
 };

+onMounted(() => {
+    getStats()
+})

+const getStats = () => {
+    api.getAssigneeStats().then((resp) => {
+        counts.value = resp.data.data
+    }).catch((err) => {
+        toast({
+            title: 'Uh oh! Something went wrong.',
+            description: err.response.data.message,
+            variant: 'destructive',
+        })
+    })
+}
 </script>

 <template>
@@ -61,18 +42,7 @@ const labels = {
                <p class="text-xl text-muted-foreground">🌤️ {{ format(new Date(), "EEEE, MMMM d HH:mm a") }}</p>
            </h4>
        </div>
-        <div class="flex mt-5 gap-x-7">
-            <Card class="w-1/6" v-for="(value, key) in counts" :key="key">
-                <CardHeader>
-                    <CardTitle class="text-4xl">
-                        {{ value }}
-                    </CardTitle>
-                    <CardDescription>
-                        {{ labels[key] }}
-                    </CardDescription>
-                </CardHeader>
-            </Card>
-        </div>
+        <CountCards :counts="counts" :labels="agentCountCardsLabels" />
        <!-- <div class="w-1/2 flex flex-col items-center justify-between">
            <LineChart :data="data" index="year" :categories="['Export Growth Rate', 'Import Growth Rate']"
                :y-formatter="(tick, i) => {
--- a/i18n/en.json
+++ b/i18n/en.json
@@ -1,4 +1,6 @@
 {
-    "_.code": "en",
-    "_.name": "English (en)"
+  "_.code": "en",
+  "_.name": "English (en)",
+  "conversaton.list.empty": "No Conversations Found.",
+  "conversaton.list.empty.filters": "Try adjusting your filters."
 }
--- a/internal/automation/automation.go
+++ b/internal/automation/automation.go
@@ -1,6 +1,7 @@
 package automation

 import (
+	"context"
 	"embed"
 	"fmt"

@@ -71,7 +72,6 @@ func New(opt Opts) (*Engine, error) {
 	return e, nil
 }

-
 func (e *Engine) SetMsgRecorder(msgRecorder MessageRecorder) {
 	e.msgRecorder = msgRecorder
 }
@@ -80,9 +80,14 @@ func (e *Engine) SetConvUpdater(convUpdater ConversationUpdater) {
 	e.convUpdater = convUpdater
 }

-func (e *Engine) Serve() {
-	for conv := range e.conversationQ {
-		e.processConversations(conv)
+func (e *Engine) Serve(ctx context.Context) {
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case conv := <-e.conversationQ:
+			e.processConversations(conv)
+		}
 	}
 }

--- a/internal/conversation/conversations.go
+++ b/internal/conversation/conversations.go
@@ -157,9 +157,9 @@ func (c *Manager) UpdateMeta(convID int, convUUID string, meta map[string]string
 	return nil
 }

-func (c *Manager) UpdateFirstReplyAt(convID int, convUUID string, at time.Time) error {
-	if _, err := c.q.UpdateFirstReplyAt.Exec(convID, convUUID, at); err != nil {
-		c.lo.Error("error updating conversation first reply at", "error", "error")
+func (c *Manager) UpdateFirstReplyAt(convID int, at time.Time) error {
+	if _, err := c.q.UpdateFirstReplyAt.Exec(convID, at); err != nil {
+		c.lo.Error("error updating conversation first reply at", "error", err)
 		return err
 	}
 	return nil
--- a/internal/conversation/models/models.go
+++ b/internal/conversation/models/models.go
@@ -18,6 +18,7 @@ type Conversation struct {
 	ReferenceNumber    null.String `db:"reference_number" json:"reference_number,omitempty"`
 	Priority           null.String `db:"priority" json:"priority"`
 	Status             null.String `db:"status" json:"status"`
+	FirstReplyAt       *time.Time  `db:"first_reply_at" json:"first_reply_at"`
 	AssignedUserID     null.Int    `db:"assigned_user_id" json:"-"`
 	AssignedTeamID     null.Int    `db:"assigned_team_id" json:"-"`
 	AssigneeLastSeenAt *time.Time  `db:"assignee_last_seen_at" json:"assignee_last_seen_at"`
--- a/internal/conversation/queries.sql
+++ b/internal/conversation/queries.sql
@@ -41,6 +41,7 @@ SELECT
    c.status,
    c.uuid,
    c.reference_number,
+    c.first_reply_at,
    ct.uuid AS contact_uuid,
    ct.first_name as first_name,
    ct.last_name as last_name,
@@ -145,4 +146,6 @@ WHERE


 -- name: update-first-reply-at
-UPDATE conversations set first_reply_at = $3 where CASE WHEN $1 > 0 then id = $1 else uuid = $2 end;
+UPDATE conversations
+SET first_reply_at = $2
+WHERE first_reply_at IS NULL AND id = $1;
--- a/internal/inbox/channel/email/email.go
+++ b/internal/inbox/channel/email/email.go
@@ -1,11 +1,19 @@
+// Package email is the email inbox with multiple SMTP servers and IMAP clients.
 package email

 import (
-	"github.com/abhinavxd/artemis/internal/message/models"
+	"context"
+
+	"github.com/abhinavxd/artemis/internal/inbox"
 	"github.com/knadh/smtppool"
+	"github.com/zerodha/logf"
 )

-// Config holds the email channel config.
+const (
+	ChannelEmail = "email"
+)
+
+// Config holds the email inbox config with multiple smtp servers & imap clients.
 type Config struct {
 	SMTP []SMTPConfig `json:"smtp"`
 	IMAP []IMAPConfig `json:"imap"`
@@ -20,12 +28,11 @@ type SMTPConfig struct {
 	TLSType       string            `json:"tls_type"`
 	TLSSkipVerify bool              `json:"tls_skip_verify"`
 	EmailHeaders  map[string]string `json:"email_headers"`
-
 	// SMTP pool options.
 	smtppool.Opt `json:",squash"`
 }

-// IMAP holds imap credentials.
+// IMAPConfig holds imap client credentials & config.
 type IMAPConfig struct {
 	Host         string `json:"host"`
 	Port         int    `json:"port"`
@@ -35,77 +42,72 @@ type IMAPConfig struct {
 	ReadInterval string `json:"read_interval"`
 }

-// Email is the email channel with multiple SMTP servers and IMAP clients.
+// Email is the email inbox with multiple SMTP servers and IMAP clients.
 type Email struct {
-	id          int
-	smtpPools   []*smtppool.Pool
-	imapClients []*IMAP
-	headers     map[string]string
-	from        string
+	id        int
+	smtpPools []*smtppool.Pool
+	imapCfg   []IMAPConfig
+	headers   map[string]string
+	lo        *logf.Logger
+	from      string
+	msgStore  inbox.MessageStore
 }

-// Opts holds the options requierd.
+// Opts holds the options required for the email inbox.
 type Opts struct {
 	ID      int
 	Headers map[string]string
 	Config  Config
+	Lo      *logf.Logger
 }

-// Returns a new instance of the Email inbox.
-func New(opts Opts) (*Email, error) {
+// New returns a new instance of the email inbox.
+func New(store inbox.MessageStore, opts Opts) (*Email, error) {
 	pools, err := NewSmtpPool(opts.Config.SMTP)
-
 	if err != nil {
 		return nil, err
 	}

 	e := &Email{
-		smtpPools:   pools,
-		imapClients: make([]*IMAP, 0, len(opts.Config.IMAP)),
-		headers:     opts.Headers,
-		from:        opts.Config.From,
+		id:        opts.ID,
+		headers:   opts.Headers,
+		from:      opts.Config.From,
+		imapCfg:   opts.Config.IMAP,
+		lo:        opts.Lo,
+		smtpPools: pools,
+		msgStore:  store,
 	}

-	// Initialize the IMAP clients.
-	for _, im := range opts.Config.IMAP {
-		imapClient, err := NewIMAP(im)
-		if err != nil {
-			return nil, err
-		}
-		// Append the IMAP client to the list of IMAP clients.
-		e.imapClients = append(e.imapClients, imapClient)
-	}
-
-	e.id = opts.ID
-
 	return e, nil
 }

-// ID returns the unique identifier of the inbox.
+// Identifier returns the unique identifier of the inbox which is the database id.
 func (e *Email) Identifier() int {
 	return e.id
 }

 // Close closes the email inbox and releases all the resources.
 func (e *Email) Close() error {
-	// Close smtp pool.
 	for _, p := range e.smtpPools {
 		p.Close()
 	}
 	return nil
 }

-func (e *Email) Receive(msgChan chan models.IncomingMessage) error {
-	for _, imap := range e.imapClients {
-		imap.ReadIncomingMessages(e.Identifier(), msgChan)
+// Receive starts receiver for each IMAP client.
+func (e *Email) Receive(ctx context.Context) error {
+	for _, cfg := range e.imapCfg {
+		e.ReadIncomingMessages(ctx, cfg)
 	}
 	return nil
 }

+// FromAddress returns the from address for this inbox.
 func (e *Email) FromAddress() string {
 	return e.from
 }

+// FromAddress returns the channel name for this inbox.
 func (e *Email) Channel() string {
-	return "email"
+	return ChannelEmail
 }
--- a/internal/inbox/channel/email/imap.go
+++ b/internal/inbox/channel/email/imap.go
@@ -1,6 +1,7 @@
 package email

 import (
+	"context"
 	"fmt"
 	"strconv"
 	"strings"
@@ -15,220 +16,220 @@ import (
 	"github.com/emersion/go-imap/v2/imapclient"
 	"github.com/jhillyerd/enmime"
 	"github.com/volatiletech/null/v9"
-	"github.com/zerodha/logf"
 )

-// IMAP represents a wrapper around the IMAP client.
-type IMAP struct {
-	lo           *logf.Logger
-	mailbox      string
-	userName     string
-	readInterval time.Duration
-	cfg          IMAPConfig
-}
+const (
+	DefaultReadInterval = time.Duration(5 * time.Minute)
+)

-func NewIMAP(cfg IMAPConfig) (*IMAP, error) {
-	logger := logf.New(logf.Opts{
-		EnableColor:          true,
-		Level:                logf.DebugLevel,
-		CallerSkipFrameCount: 3,
-		TimestampFormat:      time.RFC3339Nano,
-		EnableCaller:         true,
-	})
+func (e *Email) ReadIncomingMessages(ctx context.Context, cfg IMAPConfig) error {
 	dur, err := time.ParseDuration(cfg.ReadInterval)
-
 	if err != nil {
-		return nil, err
+		e.lo.Warn("could not IMAP read interval, using the default read interval of 5 minutes.", "error", err)
+		dur = DefaultReadInterval
 	}

-	return &IMAP{
-		mailbox:      cfg.Mailbox,
-		lo:           &logger,
-		userName:     cfg.Username,
-		readInterval: dur,
-		cfg:          cfg,
-	}, nil
+	readTicker := time.NewTicker(dur)
+	defer readTicker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		case <-readTicker.C:
+			if err := e.processMailbox(cfg); err != nil {
+				e.lo.Error("error processing mailbox", "error", err)
+			}
+		}
+	}
 }

-func (i *IMAP) ReadIncomingMessages(inboxID int, incomingMsgQ chan<- models.IncomingMessage) error {
-	var (
-		since    = time.Now().Add(time.Hour * 6 * -1)
-		tomorrow = time.Now().Add(time.Hour * 24)
-		t        = time.NewTicker(i.readInterval)
-		c        = &imapclient.Client{}
-	)
-
-	for range t.C {
-		var err error
-
-		c, err = imapclient.DialTLS(i.cfg.Host+":"+fmt.Sprint(i.cfg.Port), &imapclient.Options{})
-		if err != nil {
-			i.lo.Error("error connecting to IMAP server", "imap_username", i.userName, "error", err)
-			continue
-		}
-
-		// Send the login command.
-		cmd := c.Login(i.cfg.Username, i.cfg.Password)
-
-		// Wait for the login command to complete.
-		err = cmd.Wait()
-
-		if err != nil {
-			i.lo.Error("error logging in to the IMAP server", "imap_username", i.userName, "error", err)
-			continue
-		}
-
-		// Select the Mailbox.
-		selectCMD := c.Select(i.mailbox, &imap.SelectOptions{ReadOnly: true})
-
-		_, err = selectCMD.Wait()
-
-		if err != nil {
-			i.lo.Error("error doing imap select", "error", err)
-		}
-
-		i.lo.Debug("fetching emails", "since", since, "to", tomorrow)
-
-		searchCMD := c.Search(&imap.SearchCriteria{
-			Since:  since,
-			Before: tomorrow,
-		},
-			&imap.SearchOptions{
-				ReturnMin:   true,
-				ReturnMax:   true,
-				ReturnAll:   true,
-				ReturnCount: true,
-			},
-		)
-		searchData, err := searchCMD.Wait()
-
-		if err != nil {
-			i.lo.Error("error executing IMAP search command", "imap_username", i.userName, "error", err)
-			continue
-		}
-
-		i.lo.Debug("imap search stats", "count", searchData.Count)
-
-		fetchOptions := &imap.FetchOptions{
-			Envelope:    true,
-			BodySection: []*imap.FetchItemBodySection{{}},
-		}
-
-		seqSet := imap.SeqSet{}
-		seqSet.AddRange(searchData.Min, searchData.Max)
-
-		fetchCmd := c.Fetch(seqSet, fetchOptions)
-
-		for {
-			msg := fetchCmd.Next()
-			if msg == nil {
-				break
-			}
-
-			// Incoming message has message & contact details.
-			incomingMsg := models.IncomingMessage{
-				Message: models.Message{
-					Channel:    "email",
-					SenderType: "contact",
-					Type:       message.TypeIncoming,
-					Meta:       "{}",
-					InboxID:    inboxID,
-					Status:     message.StatusReceived,
-				},
-				Contact: cmodels.Contact{
-					Source: "email",
-				},
-				InboxID: inboxID,
-			}
-			fmt.Println("imap inbox id", inboxID)
-
-			for {
-				fetchItem := msg.Next()
-				if fetchItem == nil {
-					break
-				}
-
-				switch item := fetchItem.(type) {
-				case imapclient.FetchItemDataEnvelope:
-					env := item.Envelope
-					if len(env.From) == 0 {
-						i.lo.Debug("No sender found for email for message id", "source_id", env.MessageID)
-						break
-					}
-
-					// Get contact name.
-					email := env.From[0].Addr()
-					env.From[0].Name = strings.TrimSpace(env.From[0].Name)
-					names := strings.SplitN(env.From[0].Name, " ", 2)
-					if len(names) == 1 && names[0] != "" {
-						incomingMsg.Contact.FirstName, incomingMsg.Contact.LastName = names[0], ""
-					} else if len(names) > 1 && names[0] != "" {
-						incomingMsg.Contact.FirstName, incomingMsg.Contact.LastName = names[0], names[1]
-					} else {
-						incomingMsg.Contact.FirstName = env.From[0].Host
-					}
-
-					incomingMsg.Message.Subject = env.Subject
-					incomingMsg.Message.SourceID = null.StringFrom(env.MessageID)
-					// For contact the source will the unique identifier i.e email.
-					incomingMsg.Contact.SourceID = email
-					incomingMsg.Contact.Email = email
-					incomingMsg.Contact.InboxID = inboxID
-				case imapclient.FetchItemDataBodySection:
-					envel, err := enmime.ReadEnvelope(item.Literal)
-					if err != nil {
-						i.lo.Error("parsing email envelope", "error", err)
-						break
-					}
-					if len(envel.HTML) > 0 {
-						incomingMsg.Message.Content = envel.HTML
-						incomingMsg.Message.ContentType = message.ContentTypeHTML
-					} else if len(envel.Text) > 0 {
-						incomingMsg.Message.Content = envel.Text
-						incomingMsg.Message.ContentType = message.ContentTypeText
-					}
-
-					// Set in reply to and references.
-					inReplyTo := strings.ReplaceAll(strings.ReplaceAll(envel.GetHeader("In-Reply-To"), "<", ""), ">", "")
-					references := strings.Fields(envel.GetHeader("References"))
-					for i, ref := range references {
-						references[i] = strings.Trim(strings.TrimSpace(ref), " <>")
-					}
-
-					incomingMsg.Message.InReplyTo = inReplyTo
-					incomingMsg.Message.References = references
-
-					for _, j := range envel.Attachments {
-						incomingMsg.Message.Attachments = append(incomingMsg.Message.Attachments, amodels.Attachment{
-							Filename:           j.FileName,
-							Header:             j.Header,
-							Content:            j.Content,
-							ContentType:        j.ContentType,
-							ContentID:          j.ContentID,
-							ContentDisposition: attachment.DispositionAttachment,
-							Size:               strconv.Itoa(len(j.Content)),
-						})
-					}
-
-					for _, j := range envel.Inlines {
-						incomingMsg.Message.Attachments = append(incomingMsg.Message.Attachments, amodels.Attachment{
-							Filename:           j.FileName,
-							Header:             j.Header,
-							Content:            j.Content,
-							ContentType:        j.ContentType,
-							ContentID:          j.ContentID,
-							ContentDisposition: attachment.DispositionInline,
-							Size:               strconv.Itoa(len(j.Content)),
-						})
-					}
-					incomingMsgQ <- incomingMsg
-				default:
-					continue
-				}
-			}
-		}
-
-		c.Logout().Wait()
+func (e *Email) processMailbox(cfg IMAPConfig) error {
+	c, err := imapclient.DialTLS(cfg.Host+":"+fmt.Sprint(cfg.Port), &imapclient.Options{})
+	if err != nil {
+		return fmt.Errorf("error connecting to IMAP server: %w", err)
 	}
+	defer c.Logout()
+
+	if err := c.Login(cfg.Username, cfg.Password).Wait(); err != nil {
+		return fmt.Errorf("error logging in to the IMAP server: %w", err)
+	}
+
+	if _, err := c.Select(cfg.Mailbox, &imap.SelectOptions{ReadOnly: true}).Wait(); err != nil {
+		return fmt.Errorf("error selecting mailbox: %w", err)
+	}
+
+	since := time.Now().Add(time.Hour * 12 * -1)
+	before := time.Now().Add(time.Hour * 24)
+
+	searchData, err := e.searchMessages(c, since, before)
+	if err != nil {
+		return fmt.Errorf("error searching messages: %w", err)
+	}
+
+	return e.fetchAndProcessMessages(c, searchData, e.Identifier())
+}
+
+func (e *Email) searchMessages(c *imapclient.Client, since, before time.Time) (*imap.SearchData, error) {
+	searchCMD := c.Search(&imap.SearchCriteria{
+		Since:  since,
+		Before: before,
+	},
+		&imap.SearchOptions{
+			ReturnMin:   true,
+			ReturnMax:   true,
+			ReturnAll:   true,
+			ReturnCount: true,
+		},
+	)
+	return searchCMD.Wait()
+}
+
+func (e *Email) fetchAndProcessMessages(c *imapclient.Client, searchData *imap.SearchData, inboxID int) error {
+	seqSet := imap.SeqSet{}
+	seqSet.AddRange(searchData.Min, searchData.Max)
+
+	// Fetch only envelope.
+	fetchOptions := &imap.FetchOptions{
+		Envelope: true,
+	}
+
+	fetchCmd := c.Fetch(seqSet, fetchOptions)
+
+	for {
+		msg := fetchCmd.Next()
+		if msg == nil {
+			break
+		}
+
+		for fetchItem := msg.Next(); fetchItem != nil; fetchItem = msg.Next() {
+			if item, ok := fetchItem.(imapclient.FetchItemDataEnvelope); ok {
+				if err := e.processEnvelope(c, item.Envelope, msg.SeqNum, inboxID); err != nil {
+					e.lo.Error("error processing envelope", "error", err)
+				}
+			}
+		}
+	}
+
 	return nil
 }
+
+func (e *Email) processEnvelope(c *imapclient.Client, env *imap.Envelope, seqNum uint32, inboxID int) error {
+	if len(env.From) == 0 {
+		e.lo.Debug("no sender found for email", "message_id", env.MessageID)
+		return nil
+	}
+
+	exists, err := e.msgStore.MessageExists(env.MessageID)
+	if exists || err != nil {
+		e.lo.Debug("email message already exists, skipping", "message_id", env.MessageID)
+		return nil
+	}
+
+	incomingMsg := models.IncomingMessage{
+		Message: models.Message{
+			Channel:    e.Channel(),
+			SenderType: message.SenderTypeContact,
+			Type:       message.TypeIncoming,
+			Meta:       "{}",
+			InboxID:    int(inboxID),
+			Status:     message.StatusReceived,
+			Subject:    env.Subject,
+			SourceID:   null.StringFrom(env.MessageID),
+		},
+		Contact: cmodels.Contact{
+			Source:   e.Channel(),
+			SourceID: env.From[0].Addr(),
+			Email:    env.From[0].Addr(),
+			InboxID:  int(inboxID),
+		},
+		InboxID: int(inboxID),
+	}
+	incomingMsg.Contact.FirstName, incomingMsg.Contact.LastName = getContactName(env.From[0])
+
+	fetchOptions := &imap.FetchOptions{
+		BodySection: []*imap.FetchItemBodySection{{}},
+	}
+	seqSet := imap.SeqSet{}
+	seqSet.AddNum(seqNum)
+
+	fullFetchCmd := c.Fetch(seqSet, fetchOptions)
+	fullMsg := fullFetchCmd.Next()
+	if fullMsg == nil {
+		return nil
+	}
+
+	for fullFetchItem := fullMsg.Next(); fullFetchItem != nil; fullFetchItem = fullMsg.Next() {
+		if fullItem, ok := fullFetchItem.(imapclient.FetchItemDataBodySection); ok {
+			return e.processFullMessage(fullItem, &incomingMsg)
+		}
+	}
+
+	return nil
+}
+
+func (e *Email) processFullMessage(item imapclient.FetchItemDataBodySection, incomingMsg *models.IncomingMessage) error {
+	envel, err := enmime.ReadEnvelope(item.Literal)
+	if err != nil {
+		return fmt.Errorf("error parsing email envelope: %w", err)
+	}
+
+	if len(envel.HTML) > 0 {
+		incomingMsg.Message.Content = envel.HTML
+		incomingMsg.Message.ContentType = message.ContentTypeHTML
+	} else if len(envel.Text) > 0 {
+		incomingMsg.Message.Content = envel.Text
+		incomingMsg.Message.ContentType = message.ContentTypeText
+	}
+
+	inReplyTo := strings.ReplaceAll(strings.ReplaceAll(envel.GetHeader("In-Reply-To"), "<", ""), ">", "")
+	references := strings.Fields(envel.GetHeader("References"))
+	for i, ref := range references {
+		references[i] = strings.Trim(strings.TrimSpace(ref), " <>")
+	}
+
+	incomingMsg.Message.InReplyTo = inReplyTo
+	incomingMsg.Message.References = references
+
+	for _, j := range envel.Attachments {
+		incomingMsg.Message.Attachments = append(incomingMsg.Message.Attachments, amodels.Attachment{
+			Filename:           j.FileName,
+			Header:             j.Header,
+			Content:            j.Content,
+			ContentType:        j.ContentType,
+			ContentID:          j.ContentID,
+			ContentDisposition: attachment.DispositionAttachment,
+			Size:               strconv.Itoa(len(j.Content)),
+		})
+	}
+
+	for _, j := range envel.Inlines {
+		incomingMsg.Message.Attachments = append(incomingMsg.Message.Attachments, amodels.Attachment{
+			Filename:           j.FileName,
+			Header:             j.Header,
+			Content:            j.Content,
+			ContentType:        j.ContentType,
+			ContentID:          j.ContentID,
+			ContentDisposition: attachment.DispositionInline,
+			Size:               strconv.Itoa(len(j.Content)),
+		})
+	}
+
+	if err := e.msgStore.ProcessMessage(*incomingMsg); err != nil {
+		return fmt.Errorf("error processing message: %w", err)
+	}
+
+	return nil
+}
+
+func getContactName(imapAddr imap.Address) (string, string) {
+	from := strings.TrimSpace(imapAddr.Name)
+	names := strings.Fields(from)
+	if len(names) == 0 {
+		return imapAddr.Host, ""
+	}
+	if len(names) == 1 {
+		return names[0], ""
+	}
+	return names[0], names[1]
+}
--- a/internal/inbox/channel/email/smtp.go
+++ b/internal/inbox/channel/email/smtp.go
@@ -62,7 +62,7 @@ func NewSmtpPool(configs []SMTPConfig) ([]*smtppool.Pool, error) {
 	return pools, nil
 }

-// Send pushes a message to the server.
+// Send sends an email.
 func (e *Email) Send(m models.Message) error {
 	// If there are more than one SMTP servers, send to a random one from the list.
 	var (
--- a/internal/inbox/inbox.go
+++ b/internal/inbox/inbox.go
@@ -1,9 +1,10 @@
 package inbox

 import (
+	"context"
 	"embed"
 	"encoding/json"
-	"fmt"
+	"errors"

 	"github.com/abhinavxd/artemis/internal/dbutils"
 	"github.com/abhinavxd/artemis/internal/message/models"
@@ -15,16 +16,10 @@ var (
 	// Embedded filesystem
 	//go:embed queries.sql
 	efs embed.FS
+
+	ErrInboxNotFound = errors.New("inbox not found")
 )

-type InboxNotFound struct {
-	ID int
-}
-
-func (e *InboxNotFound) Error() string {
-	return fmt.Sprintf("inbox not found: %d", e.ID)
-}
-
 // Closer provides function for closing a channel.
 type Closer interface {
 	Close() error
@@ -37,7 +32,7 @@ type Identifier interface {

 // MessageHandler defines methods for handling message operations.
 type MessageHandler interface {
-	Receive(chan models.IncomingMessage) error
+	Receive(context.Context) error
 	Send(models.Message) error
 }

@@ -50,6 +45,11 @@ type Inbox interface {
 	Channel() string
 }

+type MessageStore interface {
+	MessageExists(string) (bool, error)
+	ProcessMessage(models.IncomingMessage) error
+}
+
 // Opts contains the options for the initializing the inbox manager.
 type Opts struct {
 	QueueSize   int
@@ -58,10 +58,9 @@ type Opts struct {

 // Manager manages the inbox.
 type Manager struct {
-	queries      queries
-	inboxes      map[int]Inbox
-	incomingMsgQ chan models.IncomingMessage
-	lo           *logf.Logger
+	queries queries
+	inboxes map[int]Inbox
+	lo      *logf.Logger
 }

 // InboxRecord represents a inbox record in DB.
@@ -80,7 +79,7 @@ type queries struct {
 }

 // New returns a new inbox manager.
-func New(lo *logf.Logger, db *sqlx.DB, incomingMsgQ chan models.IncomingMessage) (*Manager, error) {
+func New(lo *logf.Logger, db *sqlx.DB) (*Manager, error) {
 	var q queries

 	// Scan the sql	file into the queries struct.
@@ -89,10 +88,9 @@ func New(lo *logf.Logger, db *sqlx.DB, incomingMsgQ chan models.IncomingMessage)
 	}

 	m := &Manager{
-		lo:           lo,
-		incomingMsgQ: incomingMsgQ,
-		inboxes:      make(map[int]Inbox),
-		queries:      q,
+		lo:      lo,
+		inboxes: make(map[int]Inbox),
+		queries: q,
 	}
 	return m, nil
 }
@@ -106,7 +104,7 @@ func (m *Manager) Register(i Inbox) {
 func (m *Manager) GetInbox(id int) (Inbox, error) {
 	i, ok := m.inboxes[id]
 	if !ok {
-		return nil, &InboxNotFound{ID: id}
+		return nil, ErrInboxNotFound
 	}
 	return i, nil
 }
@@ -122,8 +120,8 @@ func (m *Manager) GetActiveInboxes() ([]InboxRecord, error) {
 }

 // Receive starts receiver for each inbox.
-func (m *Manager) Receive() {
+func (m *Manager) Receive(ctx context.Context) {
 	for _, inb := range m.inboxes {
-		go inb.Receive(m.incomingMsgQ)
+		go inb.Receive(ctx)
 	}
 }
--- a/internal/initz/initz.go
+++ b/internal/initz/initz.go
@@ -67,7 +67,7 @@ func Redis(ko *koanf.Koanf) *redis.Client {
 }

 // Logger initialies a logf logger.
-func Logger(lvl string, env, src string) logf.Logger {
+func Logger(lvl string, env, src string) *logf.Logger {
 	lo := logf.New(logf.Opts{
 		Level:                getLogLevel(lvl),
 		EnableColor:          getColor(env),
@@ -75,7 +75,7 @@ func Logger(lvl string, env, src string) logf.Logger {
 		CallerSkipFrameCount: 3,
 		DefaultFields:        []any{"sc", src},
 	})
-	return lo
+	return &lo
 }

 func getColor(env string) bool {
--- a/internal/message/messages.go
+++ b/internal/message/messages.go
@@ -82,10 +82,10 @@ type Manager struct {
 }

 type Opts struct {
-	DB                     *sqlx.DB
-	Lo                     *logf.Logger
-	OutgoingMsgQueueSize   int
-	OutgoingMsgConcurrency int
+	DB                   *sqlx.DB
+	Lo                   *logf.Logger
+	IncomingMsgQueueSize int
+	OutgoingMsgQueueSize int
 }

 type queries struct {
@@ -100,7 +100,7 @@ type queries struct {
 	MessageExists        *sqlx.Stmt `query:"message-exists"`
 }

-func New(incomingMsgQ chan models.IncomingMessage,
+func New(
 	wsHub *ws.Hub,
 	userMgr *user.Manager,
 	teamMgr *team.Manager,
@@ -126,7 +126,7 @@ func New(incomingMsgQ chan models.IncomingMessage,
 		conversationMgr:        conversationMgr,
 		inboxMgr:               inboxMgr,
 		automationEngine:       automationEngine,
-		incomingMsgQ:           incomingMsgQ,
+		incomingMsgQ:           make(chan models.IncomingMessage, opts.IncomingMsgQueueSize),
 		outgoingMsgQ:           make(chan models.Message, opts.OutgoingMsgQueueSize),
 		outgoingProcessingMsgs: sync.Map{},
 	}, nil
@@ -194,7 +194,6 @@ func (m *Manager) StartDispatcher(ctx context.Context, concurrency int, readInte
 	for {
 		select {
 		case <-ctx.Done():
-			m.lo.Info("context cancelled while sending messages.. Stopping dispatcher.")
 			return
 		case <-dbScanner.C:
 			var (
@@ -254,7 +253,8 @@ func (m *Manager) DispatchWorker() {

 		switch newStatus {
 		case StatusSent:
-			m.conversationMgr.UpdateFirstReplyAt(msg.ConversationID, "", msg.CreatedAt)
+			m.lo.Debug("updating first reply at", "conv_id", msg.ConversationID, "at", msg.CreatedAt)
+			m.conversationMgr.UpdateFirstReplyAt(msg.ConversationID, msg.CreatedAt)
 		}

 		// Broadcast the new message status.
@@ -457,6 +457,29 @@ func (m *Manager) processIncomingMessage(in models.IncomingMessage) error {
 	return nil
 }

+// MessageExists checks if a message with the given messageID exists.
+func (m *Manager) MessageExists(messageID string) (bool, error) {
+	_, err := m.findConversationID([]string{messageID})
+	if err != nil {
+		if errors.Is(err, ErrConversationNotFound) {
+			return false, nil
+		}
+		m.lo.Error("error fetching message from db", "error", err)
+		return false, err
+	}
+	return true, nil
+}
+
+// ProcessMessage enqueues an incoming message for processing.
+func (m *Manager) ProcessMessage(message models.IncomingMessage) error {
+	select {
+	case m.incomingMsgQ <- message:
+		return nil
+	default:
+		return fmt.Errorf("failed to enqueue message: %v", message.Message.ID)
+	}
+}
+
 func (m *Manager) TrimMsg(msg string) string {
 	plain := strings.Trim(strings.TrimSpace(html2text.HTML2Text(msg)), " \t\n\r\v\f")
 	if len(plain) > maxLastMessageLen {
@@ -546,8 +569,8 @@ func (m *Manager) findConversationID(sourceIDs []string) (int, error) {
 		if err == sql.ErrNoRows {
 			return conversationID, ErrConversationNotFound
 		} else {
-			m.lo.Error("error checking for existing message", "error", err)
-			return conversationID, fmt.Errorf("checking for existing message: %w", err)
+			m.lo.Error("error fetching msg from DB", "error", err)
+			return conversationID, err
 		}
 	}
 	return conversationID, nil