some more commits.

cmd/conversation.go

@@ -70,7 +70,7 @@ func handleUpdateAssignee(r *fastglue.Request) error {
 		uuid         = r.RequestCtx.UserValue("conversation_uuid").(string)
 		assigneeType = r.RequestCtx.UserValue("assignee_type").(string)
 		userUUID     = r.RequestCtx.UserValue("user_uuid").(string)
-		userID       = r.RequestCtx.UserValue("user_id").(int64)
+		userID       = r.RequestCtx.UserValue("user_id").(int)
 	)
 
 	if err := app.conversationMgr.UpdateAssignee(uuid, assigneeUUID, assigneeType); err != nil {
@@ -115,7 +115,7 @@ func handleUpdatePriority(r *fastglue.Request) error {
 		priority = p.Peek("priority")
 		uuid     = r.RequestCtx.UserValue("conversation_uuid").(string)
 		userUUID = r.RequestCtx.UserValue("user_uuid").(string)
-		userID   = r.RequestCtx.UserValue("user_id").(int64)
+		userID   = r.RequestCtx.UserValue("user_id").(int)
 	)
 	if err := app.conversationMgr.UpdatePriority(uuid, priority); err != nil {
 		return r.SendErrorEnvelope(http.StatusInternalServerError, err.Error(), nil, "")
@@ -139,7 +139,7 @@ func handleUpdateStatus(r *fastglue.Request) error {
 		status   = p.Peek("status")
 		uuid     = r.RequestCtx.UserValue("conversation_uuid").(string)
 		userUUID = r.RequestCtx.UserValue("user_uuid").(string)
-		userID   = r.RequestCtx.UserValue("user_id").(int64)
+		userID   = r.RequestCtx.UserValue("user_id").(int)
 	)
 	if err := app.conversationMgr.UpdateStatus(uuid, status); err != nil {
 		return r.SendErrorEnvelope(http.StatusInternalServerError, err.Error(), nil, "")

cmd/init.go

@@ -8,6 +8,8 @@ import (
 
 	"github.com/abhinavxd/artemis/internal/attachment"
 	"github.com/abhinavxd/artemis/internal/attachment/stores/s3"
+	"github.com/abhinavxd/artemis/internal/autoassigner"
+	"github.com/abhinavxd/artemis/internal/automation"
 	"github.com/abhinavxd/artemis/internal/cannedresp"
 	"github.com/abhinavxd/artemis/internal/contact"
 	"github.com/abhinavxd/artemis/internal/conversation"
@@ -164,8 +166,9 @@ func initContactManager(db *sqlx.DB, lo *logf.Logger) *contact.Manager {
 	return m
 }
 
-func initMessages(db *sqlx.DB, lo *logf.Logger, incomingMsgQ chan mmodels.IncomingMessage, wsHub *ws.Hub, contactMgr *contact.Manager, attachmentMgr *attachment.Manager, conversationMgr *conversation.Manager, inboxMgr *inbox.Manager) *message.Manager {
+func initMessages(db *sqlx.DB, lo *logf.Logger, incomingMsgQ chan mmodels.IncomingMessage, wsHub *ws.Hub, contactMgr *contact.Manager, attachmentMgr *attachment.Manager,
-	mgr, err := message.New(incomingMsgQ, wsHub, contactMgr, attachmentMgr, inboxMgr, conversationMgr, message.Opts{
+	conversationMgr *conversation.Manager, inboxMgr *inbox.Manager, automationEngine *automation.Engine) *message.Manager {
+	mgr, err := message.New(incomingMsgQ, wsHub, contactMgr, attachmentMgr, inboxMgr, conversationMgr, automationEngine, message.Opts{
 		DB: db,
 		Lo: lo,
 	})
@@ -252,6 +255,25 @@ func initInboxManager(db *sqlx.DB, lo *logf.Logger, incomingMsgQ chan mmodels.In
 	return mgr
 }
 
+func initAutomationEngine(convMgr *conversation.Manager, db *sqlx.DB, lo *logf.Logger) *automation.Engine {
+	engine, err := automation.New(convMgr, automation.Opts{
+		DB: db,
+		Lo: lo,
+	})
+	if err != nil {
+		log.Fatalf("error initializing automation engine: %v", err)
+	}
+	return engine
+}
+
+func initAutoAssignmentEngine(teamMgr *team.Manager, userMgr *user.Manager, convMgr *conversation.Manager, lo *logf.Logger) *autoassigner.Engine {
+	engine, err := autoassigner.New(teamMgr, userMgr, convMgr, lo)
+	if err != nil {
+		log.Fatalf("error initializing auto assignment engine: %v", err)
+	}
+	return engine
+}
+
 // initEmailInbox initializes the email inbox.
 func initEmailInbox(inboxRecord inbox.InboxRecord) (inbox.Inbox, error) {
 	var config email.Config

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"
@@ -50,7 +51,7 @@ func main() {
 	// Load command line flags into Koanf.
 	initFlags()
 
-	// Load the config file into Koanf.
+	// Load the config files into Koanf.
 	initz.Config(ko)
 
 	var (
@@ -64,27 +65,30 @@ func main() {
 		rd = initz.Redis(ko)
 		db = initz.DB(ko)
 
-		attachmentMgr   = initAttachmentsManager(db, &lo)
+		attachmentMgr      = initAttachmentsManager(db, &lo)
-		cntctMgr        = initContactManager(db, &lo)
+		cntctMgr           = initContactManager(db, &lo)
-		conversationMgr = initConversations(db, &lo)
+		conversationMgr    = initConversations(db, &lo)
-		inboxMgr        = initInboxManager(db, &lo, incomingMsgQ)
+		inboxMgr           = initInboxManager(db, &lo, incomingMsgQ)
+		automationEngine   = initAutomationEngine(conversationMgr, db, &lo)
+		teamMgr            = initTeamMgr(db, &lo)
+		userMgr            = initUserDB(db, &lo)
+		autoAssignerEngine = initAutoAssignmentEngine(teamMgr, userMgr, conversationMgr, &lo)
 
-		// Websocket hub.
+		// Init Websocket hub.
 		wsHub = ws.NewHub()
 	)
 
-	// Init the app.
+	// Init the app
 	var app = &App{
-		lo:                  &lo,
+		lo:              &lo,
-		cntctMgr:            cntctMgr,
+		cntctMgr:        cntctMgr,
-		inboxMgr:            inboxMgr,
+		inboxMgr:        inboxMgr,
-		attachmentMgr:       attachmentMgr,
+		attachmentMgr:   attachmentMgr,
-		conversationMgr:     conversationMgr,
+		conversationMgr: conversationMgr,
-		constants:           initConstants(),
+		constants:       initConstants(),
-		msgMgr:              initMessages(db, &lo, incomingMsgQ, wsHub, cntctMgr, attachmentMgr, conversationMgr, inboxMgr),
+		msgMgr:          initMessages(db, &lo, incomingMsgQ, wsHub, cntctMgr, attachmentMgr, conversationMgr, inboxMgr, automationEngine),
-		sessMgr:             initSessionManager(rd),
+		sessMgr:         initSessionManager(rd),
-		userMgr:             initUserDB(db, &lo),
+
-		teamMgr:             initTeamMgr(db, &lo),
 		tagMgr:              initTags(db, &lo),
 		cannedRespMgr:       initCannedResponse(db, &lo),
 		conversationTagsMgr: initConversationTags(db, &lo),
@@ -93,10 +97,16 @@ func main() {
 	// Start receivers for all active inboxes.
 	inboxMgr.Receive()
 
-	// Start message inserter and dispatchers.
+	// Start incoming msg inserter and outgoing msg dispatchers.
 	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()
+
+	// Start auto assigner enginer.
+	go autoAssignerEngine.Serve(ctx, 10*time.Second)
+
 	// Init fastglue http server.
 	g := fastglue.NewGlue()
 
@@ -123,7 +133,7 @@ func main() {
 	}()
 
 	// Start the HTTP server.
-	log.Printf("server listening on %s %s", ko.String("app.server.address"), ko.String("app.server.socket"))
+	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)
 	}

cmd/messages.go

@@ -95,7 +95,7 @@ func handleSendMessage(r *fastglue.Request) error {
 	_, _, err := app.msgMgr.RecordMessage(
 		mmodels.Message{
 			ConversationUUID: conversationUUID,
-			SenderID:         int64(userID),
+			SenderID:         userID,
 			Type:             message.TypeOutgoing,
 			SenderType:       "user",
 			Status:           status,

frontend/src/App.vue

@@ -1,4 +1,5 @@
 <template>
+  <Toaster />
   <TooltipProvider :delay-duration=200>
     <div class="bg-background text-foreground">
       <div v-if="$route.path !== '/login'">
@@ -20,6 +21,7 @@
         <RouterView />
       </div>
     </div>
+
   </TooltipProvider>
 </template>
 
@@ -31,6 +33,7 @@ import { useUserStore } from '@/stores/user'
 import { initWS } from "./websocket.js"
 import api from '@/api';
 
+import { Toaster } from '@/components/ui/toast'
 import NavBar from './components/NavBar.vue'
 import {
   ResizableHandle,
@@ -42,6 +45,7 @@ import {
 } from '@/components/ui/tooltip'
 
 
+
 // State
 const isCollapsed = ref(false)
 const navLinks = [

frontend/src/assets/styles/main.scss

@@ -2,10 +2,12 @@
 @tailwind components;
 @tailwind utilities;
 
+
+
 // App default font-size.
 // Default: 16px, 15px looked very wide.
 :root {
-  font-size: 14px;
+  font-size: 15px;
 }
 
 .tab-container-default {
@@ -25,66 +27,67 @@ $editorContainerId: 'editor-container';
 // --primary: 217 88.1% 60.4%;
 
 // Theme.
+
 @layer base {
   :root {
     --background: 0 0% 100%;
-    --foreground: 222.2 84% 4.9%;
+    --foreground: 0 0% 3.9%;
-
+  
     --card: 0 0% 100%;
-    --card-foreground: 222.2 84% 4.9%;
+    --card-foreground: 0 0% 3.9%;
-
+  
     --popover: 0 0% 100%;
-    --popover-foreground: 222.2 84% 4.9%;
+    --popover-foreground: 0 0% 3.9%;
-
+  
-    --primary: 221.2 83.2% 53.3%;
+    --primary: 0 72.2% 50.6%;
-    --primary-foreground: 210 40% 98%;
+    --primary-foreground: 0 85.7% 97.3%;
-
+  
-    --secondary: 210 40% 96.1%;
+    --secondary: 0 0% 96.1%;
-    --secondary-foreground: 222.2 47.4% 11.2%;
+    --secondary-foreground: 0 0% 9%;
-
+  
-    --muted: 210 40% 96.1%;
+    --muted: 0 0% 96.1%;
-    --muted-foreground: 215.4 16.3% 46.9%;
+    --muted-foreground: 0 0% 45.1%;
-
+  
-    --accent: 210 40% 96.1%;
+    --accent: 0 0% 96.1%;
-    --accent-foreground: 222.2 47.4% 11.2%;
+    --accent-foreground: 0 0% 9%;
-
+  
     --destructive: 0 84.2% 60.2%;
-    --destructive-foreground: 210 40% 98%;
+    --destructive-foreground: 0 0% 98%;
-
+  
-    --border: 214.3 31.8% 91.4%;
+    --border:0 0% 89.8%;
-    --input: 214.3 31.8% 91.4%;
+    --input:0 0% 89.8%;
-    --ring: 221.2 83.2% 53.3%;
+    --ring:0 72.2% 50.6%;
     --radius: 0.5rem;
   }
-
+   
   .dark {
-    --background: 222.2 84% 4.9%;
+    --background:0 0% 3.9%;
-    --foreground: 210 40% 98%;
+    --foreground:0 0% 98%;
-
+  
-    --card: 222.2 84% 4.9%;
+    --card:0 0% 3.9%;
-    --card-foreground: 210 40% 98%;
+    --card-foreground:0 0% 98%;
-
+  
-    --popover: 222.2 84% 4.9%;
+    --popover:0 0% 3.9%;
-    --popover-foreground: 210 40% 98%;
+    --popover-foreground:0 0% 98%;
-
+  
-    --primary: 217.2 91.2% 59.8%;
+    --primary:0 72.2% 50.6%;
-    --primary-foreground: 222.2 47.4% 11.2%;
+    --primary-foreground:0 85.7% 97.3%;
-
+  
-    --secondary: 217.2 32.6% 17.5%;
+    --secondary:0 0% 14.9%;
-    --secondary-foreground: 210 40% 98%;
+    --secondary-foreground:0 0% 98%;
-
+  
-    --muted: 217.2 32.6% 17.5%;
+    --muted:0 0% 14.9%;
-    --muted-foreground: 215 20.2% 65.1%;
+    --muted-foreground:0 0% 63.9%;
-
+  
-    --accent: 217.2 32.6% 17.5%;
+    --accent:0 0% 14.9%;
-    --accent-foreground: 210 40% 98%;
+    --accent-foreground:0 0% 98%;
-
+  
-    --destructive: 0 62.8% 30.6%;
+    --destructive:0 62.8% 30.6%;
-    --destructive-foreground: 210 40% 98%;
+    --destructive-foreground:0 0% 98%;
-
+  
-    --border: 217.2 32.6% 17.5%;
+    --border:0 0% 14.9%;
-    --input: 217.2 32.6% 17.5%;
+    --input:0 0% 14.9%;
-    --ring: 224.3 76.3% 48%;
+    --ring:0 72.2% 50.6%;
   }
 }
 
@@ -177,8 +180,8 @@ $editorContainerId: 'editor-container';
 }
 
 .box {
-  box-shadow: rgb(243, 243, 243) 2px 2px 0px 0px;
+  box-shadow: rgb(243, 243, 243) 1px 1px 0px 0px;
-  -webkit-box-shadow: rgb(243, 243, 243) 2px 2px 0px 0px;
+  -webkit-box-shadow: rgb(243, 243, 243) 1px 1px 0px 0px;
 }
 
 [id^='radix-vue-splitter-resize-handle'] {

frontend/src/components/AgentMessageBubble.vue

@@ -3,7 +3,7 @@
 
         <div class="pr-[47px] mb-1">
             <p class="text-muted-foreground text-sm">
-                {{ getFullName(message) }}
+                {{ getFullName }}
             </p>
         </div>
 
@@ -26,7 +26,7 @@
                 <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>
+                        message.status }}</span>
                     <RotateCcw size="10" @click="retryMessage(message)" class="cursor-pointer"
                         v-if="message.status === 'failed'"></RotateCcw>
                 </div>
@@ -34,7 +34,7 @@
             <Avatar class="cursor-pointer">
                 <AvatarImage :src=getAvatar />
                 <AvatarFallback>
-                    {{ avatarFallback(message) }}
+                    {{ avatarFallback }}
                 </AvatarFallback>
             </Avatar>
 
@@ -56,9 +56,10 @@
 </template>
 
 <script setup>
+import { computed } from 'vue'
 import { format } from 'date-fns'
 import { useConversationStore } from '@/stores/conversation'
-import api from '@/api';
+import api from '@/api'
 
 import {
     Tooltip,
@@ -71,34 +72,26 @@ import { Avatar, AvatarFallback, AvatarImage } from '@/components/ui/avatar'
 import MessageAttachmentPreview from "./MessageAttachmentPreview.vue"
 
 
-defineProps({
+const props = defineProps({
     message: Object,
 })
 const convStore = useConversationStore()
 
-const getAvatar = (msg) => {
+const participant = computed(() => {
-    if (msg.sender_uuid && convStore.conversation.participants) {
+    return convStore.conversation?.participants[props.message.sender_uuid] || {};
-        let participant = convStore.conversation.participants[msg.sender_uuid]
+});
-        return participant.avatar_url ? participant.avatar_url : ''
-    }
-    return ''
-}
 
-const getFullName = (msg) => {
+const getFullName = computed(() => {
-    if (msg.sender_uuid && convStore.conversation.participants) {
+    return `${participant.value?.first_name} ${participant.value.last_name}`;
-        let participant = convStore.conversation.participants[msg.sender_uuid]
+});
-        return participant.first_name + ' ' + participant.last_name
-    }
-    return ''
-}
 
-const avatarFallback = (msg) => {
+const getAvatar = computed(() => {
-    if (msg.sender_uuid && convStore.conversation.participants) {
+    return participant.value.avatar_url || '';
-        let participant = convStore.conversation.participants[msg.sender_uuid]
+});
-        return participant.first_name.toUpperCase().substring(0, 2)
+
-    }
+const avatarFallback = computed(() => {
-    return ''
+    return participant.value?.first_name.toUpperCase().substring(0, 2);
-}
+});
 
 const retryMessage = (msg) => {
     api.retryMessage(msg.uuid)

frontend/src/components/ConversationPlaceholder.vue

@@ -1,7 +1,7 @@
 <template>
     <h1 class="h-screen flex items-center justify-center">
         <div class="flex flex-row items-center justify-center">
-            <p>Select a conversation from the left panel.</p>
+            <p>Select a conversation.</p>
         </div>
     </h1>
 </template>

frontend/src/components/ReplyBox.vue

@@ -22,7 +22,6 @@ import { useCannedResponses } from '@/stores/canned_responses'
 const conversationStore = useConversationStore()
 const cannedResponsesStore = useCannedResponses()
 
-
 onMounted(() => {
     cannedResponsesStore.fetchAll()
 })
@@ -34,7 +33,7 @@ const sendMessage = async (message) => {
         message: message.html,
         attachments: JSON.stringify(message.attachments),
     })
-    conversationStore.fetchMessages(conversationStore.conversation.data.uuid)
+    api.updateAssigneeLastSeen(conversationStore.conversation.data.uuid)
 }
 
 </script>

frontend/src/components/ui/toast/ToastProvider.vue

@@ -1,7 +1,7 @@
 <script setup lang="ts">
 import { ToastProvider, type ToastProviderProps } from 'radix-vue'
 
-const props = defineProps<ToastProviderProps>()
+const props =  defineProps<ToastProviderProps>()
 </script>
 
 <template>

frontend/src/stores/conversation.js

@@ -160,10 +160,14 @@ export const useConversationStore = defineStore('conversation', () => {
 
     // Websocket updates.
     function updateConversationList (msg) {
-        const conversation = conversations.value.data.find(c => c.uuid === msg.conversation_uuid);
+        const updatedConversation = conversations.value.data.find(c => c.uuid === msg.conversation_uuid);
-        if (conversation) {
+        if (updatedConversation) {
-            conversation.last_message = msg.last_message;
+            updatedConversation.last_message = msg.last_message;
-            conversation.last_message_at = msg.last_message_at;
+            updatedConversation.last_message_at = msg.last_message_at;
+            // If updated conversation is open do not increment the count.
+            if (updatedConversation.uuid !== conversation.value.data.uuid) {
+                updatedConversation.unread_message_count += 1
+            }
         }
     }
     function updateMessageList (msg) {

go.mod

@@ -18,6 +18,7 @@ require (
 	github.com/knadh/koanf/v2 v2.1.1
 	github.com/knadh/smtppool v1.1.0
 	github.com/lib/pq v1.10.9
+	github.com/mr-karan/balance v0.0.0-20230131075323-e0d55eb3e4b9
 	github.com/rhnvrm/simples3 v0.8.3
 	github.com/spf13/pflag v1.0.5
 	github.com/valyala/fasthttp v1.54.0

go.sum

@@ -120,6 +120,8 @@ github.com/mitchellh/copystructure v1.2.0 h1:vpKXTN4ewci03Vljg/q9QvCGUDttBOGBIa1
 github.com/mitchellh/copystructure v1.2.0/go.mod h1:qLl+cE2AmVv+CoeAwDPye/v+N2HKCj9FbZEVFJRxO9s=
 github.com/mitchellh/reflectwalk v1.0.2 h1:G2LzWKi524PWgd3mLHV8Y5k7s6XUvT0Gef6zxSIeXaQ=
 github.com/mitchellh/reflectwalk v1.0.2/go.mod h1:mSTlrgnPZtwu0c4WaC2kGObEpuNDbx0jmZXqmk4esnw=
+github.com/mr-karan/balance v0.0.0-20230131075323-e0d55eb3e4b9 h1:mQECODpWykYPwBg+ELr04Lwm/kSP6+2LPWmvVTwrTPo=
+github.com/mr-karan/balance v0.0.0-20230131075323-e0d55eb3e4b9/go.mod h1:YMjMm+2l1ye+v1MeuUJ1QPxXKzWp+x8iqg4vWuKB3Ao=
 github.com/nxadm/tail v1.4.4 h1:DQuhQpB1tVlglWS2hLQ5OV6B5r8aGxSrPc5Qo6uTN78=
 github.com/nxadm/tail v1.4.4/go.mod h1:kenIhsEOeOJmVchQTgglprH7qJGnHDVpk1VPCcaMI8A=
 github.com/olekukonko/tablewriter v0.0.5 h1:P2Ga83D34wi1o9J6Wh1mRuqd4mF/x/lgBS7N7AbDhec=

internal/attachment/attachment.go

@@ -8,7 +8,7 @@ import (
 	"net/textproto"
 
 	"github.com/abhinavxd/artemis/internal/attachment/models"
-	"github.com/abhinavxd/artemis/internal/utils"
+	"github.com/abhinavxd/artemis/internal/dbutils"
 	"github.com/jmoiron/sqlx"
 	"github.com/zerodha/logf"
 )
@@ -55,7 +55,8 @@ func New(opt Opts) (*Manager, error) {
 	var q queries
 
 	// Scan SQL file
-	if err := utils.ScanSQLFile("queries.sql", &q, opt.DB, efs); err != nil {
+
+	if err := dbutils.ScanSQLFile("queries.sql", &q, opt.DB, efs); err != nil {
 		return nil, err
 	}
 	return &Manager{
@@ -93,7 +94,7 @@ func (m *Manager) Upload(msgUUID, fileName, contentType, contentDisposition, fil
 }
 
 // AttachMessage attaches given attachments to a message.
-func (m *Manager) AttachMessage(attachments models.Attachments, msgID int64) error {
+func (m *Manager) AttachMessage(attachments models.Attachments, msgID int) error {
 	var err error
 	for _, attachment := range attachments {
 		if attachment.UUID == "" {
@@ -111,7 +112,7 @@ func (m *Manager) AttachMessage(attachments models.Attachments, msgID int64) err
 	return nil
 }
 
-func (m *Manager) GetMessageAttachments(msgID int64) (models.Attachments, error) {
+func (m *Manager) GetMessageAttachments(msgID int) (models.Attachments, error) {
 	var attachments models.Attachments
 	if err := m.queries.GetMessageAttachments.Select(&attachments, msgID); err != nil {
 		m.lo.Error("error fetching message attachments", "error", err)

internal/auditlog/auditlog.go

@@ -0,0 +1 @@
+package auditlog

internal/autoassigner/autoassigner.go

@@ -0,0 +1,96 @@
+package autoassigner
+
+import (
+	"context"
+	"strconv"
+	"time"
+
+	"github.com/abhinavxd/artemis/internal/conversation"
+	"github.com/abhinavxd/artemis/internal/team"
+	"github.com/abhinavxd/artemis/internal/user"
+	"github.com/mr-karan/balance"
+	"github.com/zerodha/logf"
+)
+
+const (
+	roundRobinDefaultWeight = 1
+)
+
+// Engine handles the assignment of unassigned conversations to agents using a round-robin strategy.
+type Engine struct {
+	// Smooth Weighted Round Robin.
+	teamRoundRobinBalancer map[int]*balance.Balance
+	convMgr                *conversation.Manager
+	userMgr                *user.Manager
+	teamMgr                *team.Manager
+	lo                     *logf.Logger
+}
+
+// New creates a new instance of the Engine.
+func New(teamMgr *team.Manager, userMgr *user.Manager, convMgr *conversation.Manager, lo *logf.Logger) (*Engine, error) {
+	// Get all teams and add users of each them to their respective round robin balancer.
+	teams, err := teamMgr.GetAll()
+	if err != nil {
+		return nil, err
+	}
+
+	var balancer = make(map[int]*balance.Balance)
+	for _, team := range teams {
+		// Fetch all users in the team.
+		users, err := teamMgr.GetTeamMembers(team.Name)
+		if err != nil {
+			return nil, err
+		}
+
+		// Now add the users to team balance map.
+		for _, user := range users {
+			if _, ok := balancer[team.ID]; !ok {
+				balancer[team.ID] = balance.NewBalance()
+			} else {
+				balancer[team.ID].Add(strconv.Itoa(user.ID), roundRobinDefaultWeight)
+			}
+		}
+	}
+	return &Engine{
+		teamRoundRobinBalancer: balancer,
+		userMgr:                userMgr,
+		teamMgr:                teamMgr,
+		lo:                     lo,
+	}, nil
+}
+
+// AssignConversations processes unassigned conversations and assigns them to agents.
+func (e *Engine) AssignConversations() error {
+	unassignedConv, err := e.convMgr.GetUnassigned()
+	if err != nil {
+		return err
+	}
+
+	for _, conv := range unassignedConv {
+		// Fetch an agent from the team balancer pool and assign.
+		pool, ok := e.teamRoundRobinBalancer[conv.AssignedTeamID.Int]
+		if !ok {
+			continue
+		}
+		userID := pool.Get()
+		
+		if userID == "" {
+			continue
+		}
+
+		e.convMgr.UpdateAssignee(conv.UUID, []byte("88be466f-adf3-427e-af6a-88df2d3fbb01"), "agent")
+	}
+	return nil
+}
+
+func (e *Engine) Serve(ctx context.Context, interval time.Duration) {
+	ticker := time.NewTicker(interval)
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			e.AssignConversations()
+		}
+	}
+}

internal/automation/automation.go

@@ -0,0 +1,73 @@
+package automation
+
+import (
+	"embed"
+	"fmt"
+
+	"github.com/abhinavxd/artemis/internal/automation/models"
+	"github.com/abhinavxd/artemis/internal/conversation"
+	cmodels "github.com/abhinavxd/artemis/internal/conversation/models"
+	"github.com/abhinavxd/artemis/internal/dbutils"
+	"github.com/jmoiron/sqlx"
+	"github.com/zerodha/logf"
+)
+
+var (
+	//go:embed queries.sql
+	efs embed.FS
+)
+
+type queries struct {
+	GetNewConversationRules *sqlx.Stmt `query:"get-rules"`
+	GetRuleActions          *sqlx.Stmt `query:"get-rule-actions"`
+}
+
+type Engine struct {
+	q                queries
+	lo               *logf.Logger
+	convMgr          *conversation.Manager
+	newConversationQ chan cmodels.Conversation
+	rules            []models.Rule
+	actions          []models.Action
+}
+
+type Opts struct {
+	DB *sqlx.DB
+	Lo *logf.Logger
+}
+
+func New(convMgr *conversation.Manager, opt Opts) (*Engine, error) {
+	var (
+		q queries
+		e = &Engine{
+			lo:               opt.Lo,
+			convMgr:          convMgr,
+			newConversationQ: make(chan cmodels.Conversation, 10000),
+		}
+	)
+
+	if err := dbutils.ScanSQLFile("queries.sql", &q, opt.DB, efs); err != nil {
+		return nil, err
+	}
+
+	// Fetch applicable rules & actions.
+	if err := q.GetNewConversationRules.Select(&e.rules); err != nil {
+		return nil, fmt.Errorf("fetching rules: %w", err)
+	}
+	if err := q.GetRuleActions.Select(&e.actions); err != nil {
+		return nil, fmt.Errorf("fetching rule actions: %w", err)
+	}
+	e.q = q
+
+	return e, nil
+}
+
+func (e *Engine) Serve() {
+	for conv := range e.newConversationQ {
+		e.processConversations(conv)
+	}
+}
+
+func (e *Engine) ProcessConversation(c cmodels.Conversation) {
+	e.newConversationQ <- c
+}

internal/automation/models/models.go

@@ -0,0 +1,24 @@
+package models
+
+const (
+	ActionAssignTeam  = "assign_team"
+	ActionAssignAgent = "assign_agent"
+
+	RuleTypeNewConversation = "new_conversation"
+)
+
+type Rule struct {
+	ID        int    `db:"id"`
+	Type      string `db:"type"`
+	Field     string `db:"field"`
+	Operator  string `db:"operator"`
+	Value     string `db:"value"`
+	GroupID   int    `db:"group_id"`
+	LogicalOp string `db:"logical_op"`
+}
+
+type Action struct {
+	RuleID int    `db:"rule_id"`
+	Type   string `db:"action_type"`
+	Action string `db:"action"`
+}

internal/automation/preprocessor.go

@@ -0,0 +1,130 @@
+package automation
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/abhinavxd/artemis/internal/automation/models"
+	cmodels "github.com/abhinavxd/artemis/internal/conversation/models"
+)
+
+func (e *Engine) processConversations(conv cmodels.Conversation) {
+	var (
+		groupRules    = make(map[int][]models.Rule)
+		groupOperator = make(map[int]string)
+	)
+
+	// Group rules by RuleID and their logical operators.
+	for _, rule := range e.rules {
+		groupRules[rule.GroupID] = append(groupRules[rule.GroupID], rule)
+		groupOperator[rule.GroupID] = rule.LogicalOp
+	}
+
+	fmt.Printf("%+v \n", e.actions)
+	fmt.Printf("%+v \n", e.rules)
+
+	// Evaluate rules grouped by RuleID
+	for groupID, rules := range groupRules {
+		e.lo.Debug("evaluating group rule", "group_id", groupID, "operator", groupOperator[groupID])
+		if e.evaluateGroup(rules, groupOperator[groupID], conv) {
+			for _, action := range e.actions {
+				if action.RuleID == rules[0].ID {
+					e.executeActions(conv)
+				}
+			}
+		}
+	}
+}
+
+// Helper function to evaluate a group of rules
+func (e *Engine) evaluateGroup(rules []models.Rule, operator string, conv cmodels.Conversation) bool {
+	switch operator {
+	case "AND":
+		// All conditions within the group must be true
+		for _, rule := range rules {
+			if !e.evaluateRule(rule, conv) {
+				e.lo.Debug("rule evaluation was not success", "id", rule.ID)
+				return false
+			}
+		}
+		e.lo.Debug("all AND rules are success")
+		return true
+	case "OR":
+		// At least one condition within the group must be true
+		for _, rule := range rules {
+			if e.evaluateRule(rule, conv) {
+				e.lo.Debug("OR rules are success", "id", rule.ID)
+				return true
+			}
+		}
+		return false
+	default:
+		e.lo.Error("invalid group operator", "operator", operator)
+	}
+	return false
+}
+
+func (e *Engine) evaluateRule(rule models.Rule, conv cmodels.Conversation) bool {
+	var (
+		conversationValue string
+		conditionMet      bool
+	)
+
+	// Extract the value from the conversation based on the rule's field
+	switch rule.Field {
+	case "subject":
+		conversationValue = conv.Subject
+	case "content":
+		conversationValue = conv.FirstMessage
+	case "status":
+		conversationValue = conv.Status.String
+	case "priority":
+		conversationValue = conv.Priority.String
+	default:
+		e.lo.Error("rule field not recognized", "field", rule.Field)
+		return false
+	}
+
+	// Lower case the value.
+	conversationValue = strings.ToLower(conversationValue)
+
+	// Compare the conversation value with the rule's value based on the operator
+	switch rule.Operator {
+	case "equals":
+		conditionMet = conversationValue == rule.Value
+	case "not equal":
+		conditionMet = conversationValue != rule.Value
+	case "contains":
+		e.lo.Debug("eval rule", "field", rule.Field, "conv_val", conversationValue, "rule_val", rule.Value)
+		conditionMet = strings.Contains(conversationValue, rule.Value)
+	case "startsWith":
+		conditionMet = strings.HasPrefix(conversationValue, rule.Value)
+	case "endsWith":
+		conditionMet = strings.HasSuffix(conversationValue, rule.Value)
+	default:
+		e.lo.Error("logical operator not recognized for evaluating rules", "operator", rule.Operator)
+		return false
+	}
+
+	return conditionMet
+}
+
+func (e *Engine) executeActions(conv cmodels.Conversation) {
+	for _, action := range e.actions {
+		err := e.processAction(action, conv)
+		if err != nil {
+			e.lo.Error("error executing rule action", "action", action.Action, "error", err)
+		}
+	}
+}
+
+func (e *Engine) processAction(action models.Action, conv cmodels.Conversation) error {
+	switch action.Type {
+	case models.ActionAssignTeam:
+		return e.convMgr.UpdateAssignee(conv.UUID, []byte(action.Action), "team")
+	case models.ActionAssignAgent:
+		return e.convMgr.UpdateStatus(conv.UUID, []byte(action.Action))
+	default:
+		return fmt.Errorf("rule action not recognized: %s", action.Type)
+	}
+}

internal/automation/queries.sql

@@ -0,0 +1,6 @@
+-- name: get-rules
+select er.id, er.type, ec.field, ec."operator", ec.value, ec.group_id, ecg.logical_op from engine_rules er inner join engine_conditions ec on ec.rule_id = er.id  
+inner join engine_condition_groups ecg on ecg.id = ec.group_id;
+
+-- name: get-rule-actions
+select rule_id, action_type, action from engine_actions;

internal/cannedresp/cannedresp.go

@@ -4,7 +4,7 @@ import (
 	"embed"
 	"fmt"
 
-	"github.com/abhinavxd/artemis/internal/utils"
+	"github.com/abhinavxd/artemis/internal/dbutils"
 	"github.com/jmoiron/sqlx"
 	"github.com/zerodha/logf"
 )
@@ -37,7 +37,7 @@ type queries struct {
 func New(opts Opts) (*Manager, error) {
 	var q queries
 
-	if err := utils.ScanSQLFile("queries.sql", &q, opts.DB, efs); err != nil {
+	if err := dbutils.ScanSQLFile("queries.sql", &q, opts.DB, efs); err != nil {
 		return nil, err
 	}
 

internal/contact/contact.go

@@ -2,10 +2,9 @@ package contact
 
 import (
 	"embed"
-	"fmt"
 
 	"github.com/abhinavxd/artemis/internal/contact/models"
-	"github.com/abhinavxd/artemis/internal/utils"
+	"github.com/abhinavxd/artemis/internal/dbutils"
 	"github.com/jmoiron/sqlx"
 	"github.com/zerodha/logf"
 )
@@ -32,7 +31,7 @@ type queries struct {
 func New(opts Opts) (*Manager, error) {
 	var q queries
 
-	if err := utils.ScanSQLFile("queries.sql", &q, opts.DB, efs); err != nil {
+	if err := dbutils.ScanSQLFile("queries.sql", &q, opts.DB, efs); err != nil {
 		return nil, err
 	}
 
@@ -42,9 +41,8 @@ func New(opts Opts) (*Manager, error) {
 	}, nil
 }
 
-func (m *Manager) Upsert(con models.Contact) (int64, error) {
+func (m *Manager) Upsert(con models.Contact) (int, error) {
-	fmt.Println("con em", con.Email)
+	var contactID int
-	var contactID int64
 	if err := m.q.InsertContact.QueryRow(con.Source, con.SourceID, con.InboxID,
 		con.FirstName, con.LastName, con.Email, con.PhoneNumber, con.AvatarURL).Scan(&contactID); err != nil {
 		m.lo.Error("inserting contact", "error", err)

internal/contact/models/models.go

@@ -3,7 +3,7 @@ package models
 import "time"
 
 type Contact struct {
-	ID          int64     `db:"id" json:"id"`
+	ID          int     `db:"id" json:"id"`
 	CreatedAt   time.Time `db:"created_at" json:"created_at"`
 	FirstName   string    `db:"first_name" json:"first_name"`
 	LastName    string    `db:"last_name" json:"last_name"`

internal/conversation/conversations.go

@@ -7,7 +7,8 @@ import (
 	"slices"
 
 	"github.com/abhinavxd/artemis/internal/conversation/models"
-	"github.com/abhinavxd/artemis/internal/utils"
+	"github.com/abhinavxd/artemis/internal/dbutils"
+	"github.com/abhinavxd/artemis/internal/stringutils"
 	"github.com/jmoiron/sqlx"
 	"github.com/lib/pq"
 	"github.com/zerodha/logf"
@@ -50,6 +51,7 @@ type queries struct {
 	GetUUID                      *sqlx.Stmt `query:"get-uuid"`
 	GetInboxID                   *sqlx.Stmt `query:"get-inbox-id"`
 	GetConversation              *sqlx.Stmt `query:"get-conversation"`
+	GetUnassigned                *sqlx.Stmt `query:"get-unassigned"`
 	GetConversationParticipants  *sqlx.Stmt `query:"get-conversation-participants"`
 	GetConversations             *sqlx.Stmt `query:"get-conversations"`
 	GetAssignedConversations     *sqlx.Stmt `query:"get-assigned-conversations"`
@@ -64,7 +66,7 @@ type queries struct {
 
 func New(opts Opts) (*Manager, error) {
 	var q queries
-	if err := utils.ScanSQLFile("queries.sql", &q, opts.DB, efs); err != nil {
+	if err := dbutils.ScanSQLFile("queries.sql", &q, opts.DB, efs); err != nil {
 		return nil, err
 	}
 	c := &Manager{
@@ -76,9 +78,9 @@ func New(opts Opts) (*Manager, error) {
 	return c, nil
 }
 
-func (c *Manager) Create(contactID int64, inboxID int, meta string) (int64, error) {
+func (c *Manager) Create(contactID int, inboxID int, meta string) (int, error) {
 	var (
-		id        int64
+		id        int
 		refNum, _ = c.generateRefNum(c.ReferenceNumPattern)
 	)
 	if err := c.q.InsertConversation.QueryRow(refNum, contactID, StatusOpen, inboxID, meta).Scan(&id); err != nil {
@@ -127,8 +129,18 @@ func (c *Manager) AddParticipant(userID int, convUUID string) error {
 	return nil
 }
 
-func (c *Manager) GetID(uuid string) (int64, error) {
+func (c *Manager) GetUnassigned() ([]models.Conversation, error) {
-	var id int64
+	var conv []models.Conversation
+	if err := c.q.GetUnassigned.Get(&conv); err != nil {
+		if err != sql.ErrNoRows {
+			return conv, fmt.Errorf("conversation not found")
+		}
+	}
+	return conv, nil
+}
+
+func (c *Manager) GetID(uuid string) (int, error) {
+	var id int
 	if err := c.q.GetID.QueryRow(uuid).Scan(&id); err != nil {
 		if err == sql.ErrNoRows {
 			return id, fmt.Errorf("conversation not found: %w", err)
@@ -139,7 +151,7 @@ func (c *Manager) GetID(uuid string) (int64, error) {
 	return id, nil
 }
 
-func (c *Manager) GetUUID(id int64) (string, error) {
+func (c *Manager) GetUUID(id int) (string, error) {
 	var uuid string
 	if err := c.q.GetUUID.QueryRow(id).Scan(&uuid); err != nil {
 		if err == sql.ErrNoRows {
@@ -223,11 +235,10 @@ func (c *Manager) generateRefNum(pattern string) (string, error) {
 	if len(pattern) <= 5 {
 		pattern = "01234567890"
 	}
-	randomNumbers, err := utils.GenerateRandomNumericString(len(pattern))
+	randomNumbers, err := stringutils.RandomNumericString(len(pattern))
 	if err != nil {
 		return "", err
 	}
-
 	result := []byte(pattern)
 	randomIndex := 0
 	for i := range result {
@@ -236,6 +247,5 @@ func (c *Manager) generateRefNum(pattern string) (string, error) {
 			randomIndex++
 		}
 	}
-
 	return string(result), nil
 }

internal/conversation/models/models.go

@@ -5,23 +5,26 @@ import (
 	"time"
 
 	"github.com/abhinavxd/artemis/internal/contact/models"
+	"github.com/volatiletech/null/v9"
 )
 
 type Conversation struct {
-	ID                 int64      `db:"id" json:"-"`
+	ID                 int         `db:"id" json:"-"`
-	CreatedAt          time.Time  `db:"created_at" json:"created_at"`
+	CreatedAt          time.Time   `db:"created_at" json:"created_at"`
-	UpdatedAt          time.Time  `db:"updated_at" json:"updated_at"`
+	UpdatedAt          time.Time   `db:"updated_at" json:"updated_at"`
-	UUID               string     `db:"uuid" json:"uuid"`
+	UUID               string      `db:"uuid" json:"uuid"`
-	ClosedAt           *time.Time `db:"closed_at" json:"closed_at,omitempty"`
+	ClosedAt           null.Time   `db:"closed_at" json:"closed_at,omitempty"`
-	ResolvedAt         *time.Time `db:"resolved_at" json:"resolved_at,omitempty"`
+	ResolvedAt         null.Time   `db:"resolved_at" json:"resolved_at,omitempty"`
-	ReferenceNumber    *string    `db:"reference_number" json:"reference_number,omitempty"`
+	ReferenceNumber    null.String `db:"reference_number" json:"reference_number,omitempty"`
-	Priority           *string    `db:"priority" json:"priority"`
+	Priority           null.String `db:"priority" json:"priority"`
-	Status             *string    `db:"status" json:"status"`
+	Status             null.String `db:"status" json:"status"`
-	AssigneeLastSeenAt *time.Time `db:"assignee_last_seen_at" json:"assignee_last_seen_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"`
 	models.Contact
-
+	// Psuedo fields.
-	// Fields not in schema.
+	FirstMessage       string
+	Subject            string           `db:"subject" json:"subject"`
 	UnreadMessageCount int              `db:"unread_message_count" json:"unread_message_count"`
 	InboxName          string           `db:"inbox_name" json:"inbox_name"`
 	InboxChannel       string           `db:"inbox_channel" json:"inbox_channel"`

internal/conversation/queries.sql

@@ -34,7 +34,7 @@ SELECT
     (
         SELECT COUNT(*)
         FROM messages m
-        WHERE m.conversation_id = c.id AND m.created_at > c.assignee_last_seen_at
+        WHERE m.conversation_id = c.id AND m.created_at > c.assignee_last_seen_at AND m.type = 'incoming'
     ) AS unread_message_count
 FROM conversations c
     JOIN contacts ct ON c.contact_id = ct.id
@@ -137,4 +137,7 @@ INSERT INTO conversation_participants
 VALUES($1, (select id from conversations where uuid = $2));
 
 -- name: get-assigned-uuids
-select uuids from conversations where assigned_user_id = $1;
+select uuids from conversations where assigned_user_id = $1;
+
+-- name: get-unassigned
+SELECT id, uuid, assigned_team_id from conversations where assigned_user_id is NULL and assigned_team_id is not null;

internal/conversation/tag/tag.go

@@ -2,9 +2,8 @@ package tag
 
 import (
 	"embed"
-	"fmt"
 
-	"github.com/abhinavxd/artemis/internal/utils"
+	"github.com/abhinavxd/artemis/internal/dbutils"
 	"github.com/jmoiron/sqlx"
 	"github.com/lib/pq"
 	"github.com/zerodha/logf"
@@ -33,7 +32,7 @@ type queries struct {
 func New(opts Opts) (*Manager, error) {
 	var q queries
 
-	if err := utils.ScanSQLFile("queries.sql", &q, opts.DB, efs); err != nil {
+	if err := dbutils.ScanSQLFile("queries.sql", &q, opts.DB, efs); err != nil {
 		return nil, err
 	}
 
@@ -47,14 +46,12 @@ func (t *Manager) AddTags(convUUID string, tagIDs []int) error {
 	// Delete tags that have been removed.
 	if _, err := t.q.DeleteTags.Exec(convUUID, pq.Array(tagIDs)); err != nil {
 		t.lo.Error("inserting tag for conversation", "error", err, "converastion_uuid", convUUID, "tag_id", tagIDs)
-		return fmt.Errorf("error updating tags")
 	}
 
 	// Add new tags one by one.
 	for _, tagID := range tagIDs {
 		if _, err := t.q.AddTag.Exec(convUUID, tagID); err != nil {
 			t.lo.Error("inserting tag for conversation", "error", err, "converastion_uuid", convUUID, "tag_id", tagID)
-			return fmt.Errorf("error updating tags")
 		}
 	}
 	return nil

internal/dbutils/query_loader.go

@@ -1,4 +1,4 @@
-package utils
+package dbutils
 
 import (
 	"io/fs"

internal/inbox/channel/email/imap.go

@@ -100,7 +100,6 @@ func (i *IMAP) ReadIncomingMessages(inboxID int, incomingMsgQ chan<- models.Inco
 				ReturnCount: true,
 			},
 		)
-
 		searchData, err := searchCMD.Wait()
 
 		if err != nil {

internal/inbox/inbox.go

@@ -5,9 +5,8 @@ import (
 	"encoding/json"
 	"fmt"
 
+	"github.com/abhinavxd/artemis/internal/dbutils"
 	"github.com/abhinavxd/artemis/internal/message/models"
-	"github.com/abhinavxd/artemis/internal/utils"
-
 	"github.com/jmoiron/sqlx"
 	"github.com/zerodha/logf"
 )
@@ -85,7 +84,7 @@ func New(lo *logf.Logger, db *sqlx.DB, incomingMsgQ chan models.IncomingMessage)
 	var q queries
 
 	// Scan the sql	file into the queries struct.
-	if err := utils.ScanSQLFile("queries.sql", &q, db, efs); err != nil {
+	if err := dbutils.ScanSQLFile("queries.sql", &q, db, efs); err != nil {
 		return nil, err
 	}
 

internal/message/messages.go

@@ -12,11 +12,13 @@ import (
 	"time"
 
 	"github.com/abhinavxd/artemis/internal/attachment"
+	"github.com/abhinavxd/artemis/internal/automation"
 	"github.com/abhinavxd/artemis/internal/contact"
 	"github.com/abhinavxd/artemis/internal/conversation"
+	cmodels "github.com/abhinavxd/artemis/internal/conversation/models"
+	"github.com/abhinavxd/artemis/internal/dbutils"
 	"github.com/abhinavxd/artemis/internal/inbox"
 	"github.com/abhinavxd/artemis/internal/message/models"
-	"github.com/abhinavxd/artemis/internal/utils"
 	"github.com/abhinavxd/artemis/internal/ws"
 	"github.com/jmoiron/sqlx"
 	"github.com/k3a/html2text"
@@ -66,6 +68,7 @@ type Manager struct {
 	attachmentMgr          *attachment.Manager
 	conversationMgr        *conversation.Manager
 	inboxMgr               *inbox.Manager
+	automationEngine       *automation.Engine
 	wsHub                  *ws.Hub
 	incomingMsgQ           chan models.IncomingMessage
 	outgoingMsgQ           chan models.Message
@@ -92,10 +95,11 @@ type queries struct {
 	MessageExists        *sqlx.Stmt `query:"message-exists"`
 }
 
-func New(incomingMsgQ chan models.IncomingMessage, wsHub *ws.Hub, contactMgr *contact.Manager, attachmentMgr *attachment.Manager, inboxMgr *inbox.Manager, conversationMgr *conversation.Manager, opts Opts) (*Manager, error) {
+func New(incomingMsgQ chan models.IncomingMessage, wsHub *ws.Hub, contactMgr *contact.Manager, attachmentMgr *attachment.Manager, inboxMgr *inbox.Manager,
+	conversationMgr *conversation.Manager, automationEngine *automation.Engine, opts Opts) (*Manager, error) {
 	var q queries
 
-	if err := utils.ScanSQLFile("queries.sql", &q, opts.DB, efs); err != nil {
+	if err := dbutils.ScanSQLFile("queries.sql", &q, opts.DB, efs); err != nil {
 		return nil, err
 	}
 	return &Manager{
@@ -106,6 +110,7 @@ func New(incomingMsgQ chan models.IncomingMessage, wsHub *ws.Hub, contactMgr *co
 		attachmentMgr:          attachmentMgr,
 		conversationMgr:        conversationMgr,
 		inboxMgr:               inboxMgr,
+		automationEngine:       automationEngine,
 		incomingMsgQ:           incomingMsgQ,
 		outgoingMsgQ:           make(chan models.Message, opts.OutgoingMsgQueueSize),
 		outgoingProcessingMsgs: sync.Map{},
@@ -139,9 +144,9 @@ func (m *Manager) UpdateMessageStatus(uuid string, status string) error {
 }
 
 // RecordMessage inserts a message and attaches the attachments to the message.
-func (m *Manager) RecordMessage(msg models.Message) (int64, string, error) {
+func (m *Manager) RecordMessage(msg models.Message) (int, string, error) {
 	var (
-		msgID          int64
+		msgID          int
 		msgUUID        string
 		query          *sqlx.Stmt
 		convIdentifier interface{}
@@ -172,7 +177,7 @@ func (m *Manager) RecordMessage(msg models.Message) (int64, string, error) {
 	return msgID, msgUUID, nil
 }
 
-func (m *Manager) RecordActivity(activityType, value, conversationUUID, userName string, userID int64) error {
+func (m *Manager) RecordActivity(activityType, value, conversationUUID, userName string, userID int) error {
 	var content = m.getActivityContent(activityType, value, userName)
 	if content == "" {
 		m.lo.Error("invalid activity for inserting message", "activity", activityType)
@@ -214,7 +219,7 @@ func (m *Manager) StartDispatcher(ctx context.Context, concurrency int, readInte
 		case <-dbScanner.C:
 			var (
 				pendingMsgs = []models.Message{}
-				msgIDs      = m.getProcessingMsgIDs()
+				msgIDs      = m.getOutgoingProcessingMsgIDs()
 			)
 
 			// Skip the currently processing msg ids.
@@ -222,8 +227,6 @@ func (m *Manager) StartDispatcher(ctx context.Context, concurrency int, readInte
 				m.lo.Error("error fetching pending messages from db", "error", err)
 			}
 
-			fmt.Printf("pendings msg %+v \n", pendingMsgs)
-
 			// Prepare and push the message to the outgoing queue.
 			for _, msg := range pendingMsgs {
 				m.outgoingProcessingMsgs.Store(msg.ID, msg.ID)
@@ -254,7 +257,7 @@ func (m *Manager) DispatchWorker() {
 
 		if inbox.Channel() == "email" {
 			msg.InReplyTo, _ = m.GetInReplyTo(msg.ConversationID)
-			m.lo.Debug("set in reply to", "in_reply_to", msg.InReplyTo)
+			m.lo.Debug("set in reply to for outgoing email message", "in_reply_to", msg.InReplyTo)
 		}
 
 		err = inbox.Send(msg)
@@ -279,7 +282,7 @@ func (m *Manager) DispatchWorker() {
 	}
 }
 
-func (m *Manager) GetToAddress(convID int64, channel string) ([]string, error) {
+func (m *Manager) GetToAddress(convID int, channel string) ([]string, error) {
 	var addr []string
 	if err := m.q.GetToAddress.Select(&addr, convID, channel); err != nil {
 		m.lo.Error("error fetching to address for msg", "error", err, "conv_id", convID)
@@ -288,7 +291,7 @@ func (m *Manager) GetToAddress(convID int64, channel string) ([]string, error) {
 	return addr, nil
 }
 
-func (m *Manager) GetInReplyTo(convID int64) (string, error) {
+func (m *Manager) GetInReplyTo(convID int) (string, error) {
 	var out string
 	if err := m.q.GetInReplyTo.Get(&out, convID); err != nil {
 		if err == sql.ErrNoRows {
@@ -339,13 +342,11 @@ func (m *Manager) processIncomingMessage(in models.IncomingMessage) error {
 	}
 
 	// This message already exists?
-	m.lo.Debug("searching for message with id", "source_id", in.Message.SourceID.String)
 	conversationID, err := m.findConversationID([]string{in.Message.SourceID.String})
 	if err != nil && err != ErrConversationNotFound {
 		return err
 	}
 	if conversationID > 0 {
-		m.lo.Debug("conversation already exists for message", "source_id", in.Message.SourceID.String)
 		return nil
 	}
 
@@ -367,7 +368,7 @@ func (m *Manager) processIncomingMessage(in models.IncomingMessage) error {
 		return fmt.Errorf("uploading message attachments: %w", err)
 	}
 
-	// WS update.
+	// Send WS update.
 	cuuid, err := m.conversationMgr.GetUUID(in.Message.ConversationID)
 	if err != nil {
 		m.lo.Error("error fetching uuid for conversation", "conversation_id", in.Message.ConversationID, "error", err)
@@ -390,6 +391,15 @@ func (m *Manager) processIncomingMessage(in models.IncomingMessage) error {
 		})
 	}
 
+	// Pass this conversation for evaluating automation rules.
+	if newConv {
+		m.automationEngine.ProcessConversation(cmodels.Conversation{
+			UUID:         cuuid,
+			FirstMessage: in.Message.Content,
+			Subject:      in.Message.Subject,
+		})
+	}
+
 	return nil
 }
 
@@ -404,9 +414,9 @@ func (m *Manager) TrimMsg(msg string) string {
 
 func (m *Manager) uploadAttachments(in *models.Message) error {
 	var (
-		inlineAttachments = false
+		hasInline = false
-		msgID             = in.ID
+		msgID     = in.ID
-		msgUUID           = in.UUID
+		msgUUID   = in.UUID
 	)
 	for _, att := range in.Attachments {
 		reader := bytes.NewReader(att.Content)
@@ -416,13 +426,13 @@ func (m *Manager) uploadAttachments(in *models.Message) error {
 			return errors.New("error uploading attachments for incoming message")
 		}
 		if att.ContentDisposition == attachment.DispositionInline {
-			inlineAttachments = true
+			hasInline = true
 			in.Content = strings.ReplaceAll(in.Content, "cid:"+att.ContentID, url)
 		}
 	}
 
 	// Update the msg content the `cid:content_id` urls have been replaced.
-	if inlineAttachments {
+	if hasInline {
 		if _, err := m.q.UpdateMessageContent.Exec(in.Content, msgID); err != nil {
 			m.lo.Error("error updating message content", "message_uuid", msgUUID)
 			return fmt.Errorf("updating msg content: %w", err)
@@ -431,23 +441,21 @@ func (m *Manager) uploadAttachments(in *models.Message) error {
 	return nil
 }
 
-func (m *Manager) findOrCreateConversation(in *models.Message, inboxID int, contactID int64, conversationMeta string) (int64, bool, error) {
+func (m *Manager) findOrCreateConversation(in *models.Message, inboxID int, contactID int, conversationMeta string) (int, bool, error) {
 	var (
-		conversationID int64
+		conversationID int
 		newConv        bool
 		err            error
 	)
 
 	// Search for existing conversation.
 	if conversationID == 0 && in.InReplyTo > "" {
-		m.lo.Debug("searching for message with id", "source_id", in.InReplyTo)
 		conversationID, err = m.findConversationID([]string{in.InReplyTo})
 		if err != nil && err != ErrConversationNotFound {
 			return conversationID, newConv, err
 		}
 	}
 	if conversationID == 0 && len(in.References) > 0 {
-		m.lo.Debug("searching for message with id", "source_id", in.References)
 		conversationID, err = m.findConversationID(in.References)
 		if err != nil && err != ErrConversationNotFound {
 			return conversationID, newConv, err
@@ -485,8 +493,8 @@ func (m *Manager) getActivityContent(activityType, value, userName string) strin
 }
 
 // findConversationID finds the conversation ID from the message source ID.
-func (m *Manager) findConversationID(sourceIDs []string) (int64, error) {
+func (m *Manager) findConversationID(sourceIDs []string) (int, error) {
-	var conversationID int64
+	var conversationID int
 	if err := m.q.MessageExists.QueryRow(pq.Array(sourceIDs)).Scan(&conversationID); err != nil {
 		if err == sql.ErrNoRows {
 			return conversationID, ErrConversationNotFound
@@ -518,11 +526,11 @@ func (m *Manager) attachAttachments(msg *models.Message) error {
 	return nil
 }
 
-// getProcessingMsgIDs returns outgoing msg ids currently being processed.
+// getOutgoingProcessingMsgIDs returns outgoing msg ids currently being processed.
-func (m *Manager) getProcessingMsgIDs() []int64 {
+func (m *Manager) getOutgoingProcessingMsgIDs() []int {
-	var out = make([]int64, 0)
+	var out = make([]int, 0)
 	m.outgoingProcessingMsgs.Range(func(key, _ any) bool {
-		if k, ok := key.(int64); ok {
+		if k, ok := key.(int); ok {
 			out = append(out, k)
 		}
 		return true

internal/message/models/models.go

@@ -12,18 +12,18 @@ import (
 
 // Message represents a message in the database.
 type Message struct {
-	ID             int64              `db:"id" json:"id"`
+	ID             int                `db:"id" json:"id"`
 	CreatedAt      time.Time          `db:"created_at" json:"created_at"`
 	UpdatedAt      time.Time          `db:"updated_at" json:"updated_at"`
 	UUID           string             `db:"uuid" json:"uuid"`
 	Type           string             `db:"type" json:"type"`
 	Status         string             `db:"status" json:"status"`
-	ConversationID int64              `db:"conversation_id" json:"conversation_id"`
+	ConversationID int                `db:"conversation_id" json:"conversation_id"`
 	Content        string             `db:"content" json:"content"`
 	ContentType    string             `db:"content_type" json:"content_type"`
 	Private        bool               `db:"private" json:"private"`
 	SourceID       null.String        `db:"source_id" json:"-"`
-	SenderID       int64              `db:"sender_id" json:"sender_id"`
+	SenderID       int                `db:"sender_id" json:"sender_id"`
 	SenderType     string             `db:"sender_type" json:"sender_type"`
 	InboxID        int                `db:"inbox_id" json:"-"`
 	Meta           string             `db:"meta" json:"meta"`

internal/stringutils/strings.go

@@ -0,0 +1,43 @@
+package stringutils
+
+import (
+	"crypto/rand"
+)
+
+// RandomAlNumString generates a random alphanumeric string of length n.
+func RandomAlNumString(n int) (string, error) {
+	const (
+		dictionary = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
+	)
+
+	var bytes = make([]byte, n)
+
+	if _, err := rand.Read(bytes); err != nil {
+		return "", err
+	}
+
+	for k, v := range bytes {
+		bytes[k] = dictionary[v%byte(len(dictionary))]
+	}
+
+	return string(bytes), nil
+}
+
+// RandomNumericString generates a random digit string of length n.
+func RandomNumericString(n int) (string, error) {
+	const (
+		dictionary = "0123456789"
+	)
+
+	var bytes = make([]byte, n)
+
+	if _, err := rand.Read(bytes); err != nil {
+		return "", err
+	}
+
+	for k, v := range bytes {
+		bytes[k] = dictionary[v%byte(len(dictionary))]
+	}
+
+	return string(bytes), nil
+}

internal/tag/tag.go

@@ -5,7 +5,7 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/abhinavxd/artemis/internal/utils"
+	"github.com/abhinavxd/artemis/internal/dbutils"
 	"github.com/jmoiron/sqlx"
 	"github.com/zerodha/logf"
 )
@@ -16,7 +16,7 @@ var (
 )
 
 type Tag struct {
-	ID        int64     `db:"id" json:"id"`
+	ID        int     `db:"id" json:"id"`
 	CreatedAt time.Time `db:"created_at" json:"created_at"`
 	Name      string    `db:"name" json:"name"`
 }
@@ -40,7 +40,7 @@ type queries struct {
 func New(opts Opts) (*Manager, error) {
 	var q queries
 
-	if err := utils.ScanSQLFile("queries.sql", &q, opts.DB, efs); err != nil {
+	if err := dbutils.ScanSQLFile("queries.sql", &q, opts.DB, efs); err != nil {
 		return nil, err
 	}
 

internal/team/queries.sql

@@ -2,4 +2,12 @@
 SELECT name, uuid from teams where disabled is not true;
 
 -- name: get-team
-SELECT name, uuid from teams where disabled is not true and uuid = $1;
+SELECT name, uuid from teams where disabled is not true and uuid = $1;
+
+-- name: get-team-members
+SELECT u.id, t.id as team_id
+FROM users u
+JOIN team_members tm ON tm.user_id = u.id
+JOIN teams t ON t.id = tm.team_id
+WHERE t.name = $1;
+

internal/team/team.go

@@ -6,7 +6,8 @@ import (
 	"errors"
 	"fmt"
 
-	"github.com/abhinavxd/artemis/internal/utils"
+	"github.com/abhinavxd/artemis/internal/dbutils"
+	umodels "github.com/abhinavxd/artemis/internal/user/models"
 	"github.com/jmoiron/sqlx"
 	"github.com/zerodha/logf"
 )
@@ -17,7 +18,7 @@ var (
 )
 
 type Team struct {
-	ID   string `db:"id" json:"-"`
+	ID   int    `db:"id" json:"-"`
 	Name string `db:"name" json:"name"`
 	UUID string `db:"uuid" json:"uuid"`
 }
@@ -33,14 +34,15 @@ type Opts struct {
 }
 
 type queries struct {
-	GetTeams *sqlx.Stmt `query:"get-teams"`
+	GetTeams       *sqlx.Stmt `query:"get-teams"`
-	GetTeam  *sqlx.Stmt `query:"get-team"`
+	GetTeam        *sqlx.Stmt `query:"get-team"`
+	GetTeamMembers *sqlx.Stmt `query:"get-team-members"`
 }
 
 func New(opts Opts) (*Manager, error) {
 	var q queries
 
-	if err := utils.ScanSQLFile("queries.sql", &q, opts.DB, efs); err != nil {
+	if err := dbutils.ScanSQLFile("queries.sql", &q, opts.DB, efs); err != nil {
 		return nil, err
 	}
 
@@ -73,3 +75,15 @@ func (u *Manager) GetTeam(uuid string) (Team, error) {
 	}
 	return team, nil
 }
+
+func (u *Manager) GetTeamMembers(name string) ([]umodels.User, error) {
+	var users []umodels.User
+	if err := u.q.GetTeamMembers.Select(&users, name); err != nil {
+		if errors.Is(sql.ErrNoRows, err) {
+			return users, nil
+		}
+		u.lo.Error("error fetching team members from db", "team_name", name, "error", err)
+		return users, fmt.Errorf("fetching team members: %w", err)
+	}
+	return users, nil
+}

internal/user/models/models.go

@@ -1,7 +1,7 @@
 package models
 
 type User struct {
-	ID        int64   `db:"id" json:"-"`
+	ID        int     `db:"id" json:"-"`
 	UUID      string  `db:"uuid" json:"uuid"`
 	FirstName string  `db:"first_name" json:"first_name"`
 	LastName  string  `db:"last_name" json:"last_name"`
@@ -9,6 +9,7 @@ type User struct {
 	AvatarURL *string `db:"avatar_url" json:"avatar_url"`
 	Disabled  string  `db:"disabled" json:"disabled"`
 	Password  string  `db:"password" json:"-"`
+	TeamID    int     `db:"team_id" json:"-"`
 }
 
 func (u *User) FullName() string {

internal/user/user.go

@@ -1,4 +1,4 @@
-// Package user provides functions to login, logout and fetch user details.
+// Package user handles user login, logout and provides functions to fetch user details.
 package user
 
 import (
@@ -7,8 +7,8 @@ import (
 	"errors"
 	"fmt"
 
+	"github.com/abhinavxd/artemis/internal/dbutils"
 	"github.com/abhinavxd/artemis/internal/user/models"
-	"github.com/abhinavxd/artemis/internal/utils"
 	"github.com/jmoiron/sqlx"
 	"github.com/zerodha/logf"
 	"golang.org/x/crypto/bcrypt"
@@ -46,7 +46,7 @@ type queries struct {
 func New(opts Opts) (*Manager, error) {
 	var q queries
 
-	if err := utils.ScanSQLFile("queries.sql", &q, opts.DB, efs); err != nil {
+	if err := dbutils.ScanSQLFile("queries.sql", &q, opts.DB, efs); err != nil {
 		return nil, err
 	}
 

internal/utils/utils.go

@@ -1,127 +0,0 @@
-package utils
-
-import (
-	"crypto/rand"
-	"fmt"
-	"net/textproto"
-	"path/filepath"
-	"regexp"
-	"strings"
-	"time"
-
-	"golang.org/x/crypto/bcrypt"
-)
-
-var (
-	reSpaces = regexp.MustCompile(`[\s]+`)
-)
-
-// RandomAlNumString generates a random alphanumeric string of length n.
-func RandomAlNumString(n int) (string, error) {
-	const (
-		dictionary = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
-	)
-
-	var bytes = make([]byte, n)
-
-	if _, err := rand.Read(bytes); err != nil {
-		return "", err
-	}
-
-	for k, v := range bytes {
-		bytes[k] = dictionary[v%byte(len(dictionary))]
-	}
-
-	return string(bytes), nil
-}
-
-// GenerateRandomNumericString generates a random digit string of length n.
-func GenerateRandomNumericString(n int) (string, error) {
-	const (
-		dictionary = "0123456789"
-	)
-
-	var bytes = make([]byte, n)
-
-	if _, err := rand.Read(bytes); err != nil {
-		return "", err
-	}
-
-	for k, v := range bytes {
-		bytes[k] = dictionary[v%byte(len(dictionary))]
-	}
-
-	return string(bytes), nil
-}
-
-// GeneratePassword generates a secure password of specified length.
-func GeneratePassword(len int) ([]byte, error) {
-	randomString, err := RandomAlNumString(len)
-
-	if err != nil {
-		return nil, err
-	}
-
-	hashedPassword, err := bcrypt.GenerateFromPassword([]byte(randomString), bcrypt.DefaultCost)
-	if err != nil {
-		return nil, err
-	}
-
-	return hashedPassword, nil
-}
-
-// InArray checks if an element of type T is present in a slice of type T.
-func InArray[T comparable](val T, vals []T) bool {
-	for _, v := range vals {
-		if v == val {
-			return true
-		}
-	}
-	return false
-}
-
-// MakeFilename makes a filename from the given string.
-func MakeFilename(fName string) string {
-	name := strings.TrimSpace(fName)
-	if name == "" {
-		name, _ = RandomAlNumString(10)
-	}
-	// replace whitespace with "-"
-	name = reSpaces.ReplaceAllString(name, "-")
-	return filepath.Base(name)
-}
-
-// MakeAttachmentHeader
-func MakeAttachmentHeader(filename, encoding, contentType string) textproto.MIMEHeader {
-	if encoding == "" {
-		encoding = "base64"
-	}
-	if contentType == "" {
-		contentType = "application/octet-stream"
-	}
-
-	h := textproto.MIMEHeader{}
-	h.Set("Content-Disposition", "attachment; filename="+filename)
-	h.Set("Content-Type", fmt.Sprintf("%s; name=\""+filename+"\"", contentType))
-	h.Set("Content-Transfer-Encoding", encoding)
-	return h
-}
-
-// SplitName splits a full name into first name and last name.
-func SplitName(fullName string) (firstName string, lastName string) {
-	parts := strings.Fields(fullName)
-	if len(parts) > 1 {
-		lastName = parts[len(parts)-1]
-		firstName = strings.Join(parts[:len(parts)-1], " ")
-	} else if len(parts) == 1 {
-		firstName = parts[0]
-	}
-	return firstName, lastName
-}
-
-// BackoffDelay introduces a delay between actions with backoff behavior.
-func BackoffDelay(try int, dur time.Duration) {
-	if try > 0 {
-		<-time.After(time.Duration(try) * time.Duration(dur))
-	}
-}

package-lock.json

@@ -0,0 +1,25 @@
+{
+  "name": "artemis",
+  "lockfileVersion": 3,
+  "requires": true,
+  "packages": {
+    "": {
+      "devDependencies": {
+        "typescript": "^5.4.5"
+      }
+    },
+    "node_modules/typescript": {
+      "version": "5.4.5",
+      "resolved": "https://registry.npmjs.org/typescript/-/typescript-5.4.5.tgz",
+      "integrity": "sha512-vcI4UpRgg81oIRUFwR0WSIHKt11nJ7SAVlYNIu+QpqeyXP+gpQJy/Z4+F0aGxSE4MqwjyXvW/TzgkLAx2AGHwQ==",
+      "dev": true,
+      "bin": {
+        "tsc": "bin/tsc",
+        "tsserver": "bin/tsserver"
+      },
+      "engines": {
+        "node": ">=14.17"
+      }
+    }
+  }
+}

package.json

@@ -0,0 +1,5 @@
+{
+  "devDependencies": {
+    "typescript": "^5.4.5"
+  }
+}