diff --git a/cmd/attachment.go b/cmd/attachment.go index 044bcfdb..fc98ac32 100644 --- a/cmd/attachment.go +++ b/cmd/attachment.go @@ -38,7 +38,7 @@ func handleAttachmentUpload(r *fastglue.Request) error { file, err := form.File["files"][0].Open() srcFileName := form.File["files"][0].Filename srcContentType := form.File["files"][0].Header.Get("Content-Type") - srcFileSize := form.File["files"][0].Size + srcFileSize := strconv.FormatInt(form.File["files"][0].Size, 10) srcDisposition := form.Value["disposition"][0] if err != nil { app.lo.Error("reading file into the memory", "error", err) @@ -57,7 +57,7 @@ func handleAttachmentUpload(r *fastglue.Request) error { // Reset the ptr. file.Seek(0, 0) - url, mediaUUID, _, err := app.attachmentMgr.Upload("" /**message uuid**/, srcFileName, srcContentType, srcDisposition, strconv.FormatInt(srcFileSize, 10), file) + url, mediaUUID, _, err := app.attachmentMgr.Upload("" /**message uuid**/, srcFileName, srcContentType, srcDisposition, srcFileSize, file) if err != nil { app.lo.Error("error uploading file", "error", err) return r.SendErrorEnvelope(http.StatusInternalServerError, "Error uploading file", nil, "GeneralException") @@ -68,7 +68,7 @@ func handleAttachmentUpload(r *fastglue.Request) error { "uuid": mediaUUID, "content_type": srcContentType, "name": srcFileName, - "size": strconv.FormatInt(srcFileSize, 10), + "size": srcFileSize, }) } diff --git a/cmd/conversation.go b/cmd/conversation.go index 3938bf4e..4b792046 100644 --- a/cmd/conversation.go +++ b/cmd/conversation.go @@ -125,16 +125,16 @@ func handleUpdateAssignee(r *fastglue.Request) error { userUUID = r.RequestCtx.UserValue("user_uuid").(string) ) - if err := app.conversationMgr.UpdateAssignee(convUUID, assigneeUUID, assigneeType); err != nil { - return r.SendErrorEnvelope(http.StatusInternalServerError, err.Error(), nil, "") - } - - if assigneeType == "agent" { - app.msgMgr.RecordAssigneeUserChange(string(assigneeUUID), convUUID, userUUID) - } - - if assigneeType == "team" { - app.msgMgr.RecordAssigneeTeamChange(string(assigneeUUID), convUUID, userUUID) + if assigneeType == "user" { + if err := app.conversationMgr.UpdateUserAssignee(convUUID, assigneeUUID); err != nil { + return r.SendErrorEnvelope(http.StatusInternalServerError, err.Error(), nil, "") + } + app.msgMgr.RecordAssigneeUserChange(convUUID, string(assigneeUUID), userUUID) + } else if assigneeType == "team" { + if err := app.conversationMgr.UpdateTeamAssignee(convUUID, assigneeUUID); err != nil { + return r.SendErrorEnvelope(http.StatusInternalServerError, err.Error(), nil, "") + } + app.msgMgr.RecordAssigneeTeamChange(convUUID, string(assigneeUUID), userUUID) } return r.SendEnvelope("ok") @@ -188,7 +188,7 @@ func handlAddConversationTags(r *fastglue.Request) error { return r.SendErrorEnvelope(http.StatusInternalServerError, "error adding tags", nil, "") } - if err := app.conversationTagsMgr.AddTags(uuid, tagIDs); err != nil { + if err := app.conversationMgr.AddTags(uuid, tagIDs); err != nil { return r.SendErrorEnvelope(http.StatusInternalServerError, err.Error(), nil, "") } return r.SendEnvelope("ok") diff --git a/cmd/handlers.go b/cmd/handlers.go index b1bd62ee..677c7b7c 100644 --- a/cmd/handlers.go +++ b/cmd/handlers.go @@ -28,7 +28,6 @@ func initHandlers(g *fastglue.Fastglue, hub *ws.Hub) { g.GET("/api/canned-responses", auth(handleGetCannedResponses)) g.GET("/api/attachment/{conversation_uuid}", auth(handleGetAttachment)) g.GET("/api/users/me", auth(handleGetCurrentUser)) - g.GET("/api/users/filters/{page}", auth(handleGetUserFilters)) g.GET("/api/users", auth(handleGetUsers)) g.GET("/api/teams", auth(handleGetTeams)) g.GET("/api/tags", auth(handleGetTags)) diff --git a/cmd/init.go b/cmd/init.go index 61135396..3e222534 100644 --- a/cmd/init.go +++ b/cmd/init.go @@ -8,21 +8,22 @@ import ( "github.com/abhinavxd/artemis/internal/attachment" "github.com/abhinavxd/artemis/internal/attachment/stores/s3" + uauth "github.com/abhinavxd/artemis/internal/auth" "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" - 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" - "github.com/abhinavxd/artemis/internal/rbac" + notifier "github.com/abhinavxd/artemis/internal/notification" + emailnotifier "github.com/abhinavxd/artemis/internal/notification/providers/email" "github.com/abhinavxd/artemis/internal/tag" "github.com/abhinavxd/artemis/internal/team" + "github.com/abhinavxd/artemis/internal/template" "github.com/abhinavxd/artemis/internal/user" - "github.com/abhinavxd/artemis/internal/user/filterstore" "github.com/abhinavxd/artemis/internal/ws" "github.com/go-redis/redis/v8" "github.com/jmoiron/sqlx" @@ -112,8 +113,8 @@ func initUserDB(DB *sqlx.DB, lo *logf.Logger) *user.Manager { return mgr } -func initConversations(db *sqlx.DB, lo *logf.Logger) *conversation.Manager { - c, err := conversation.New(conversation.Opts{ +func initConversations(hub *ws.Hub, db *sqlx.DB, lo *logf.Logger) *conversation.Manager { + c, err := conversation.New(hub, conversation.Opts{ DB: db, Lo: lo, ReferenceNumPattern: ko.String("app.constants.conversation_reference_number_pattern"), @@ -124,17 +125,6 @@ func initConversations(db *sqlx.DB, lo *logf.Logger) *conversation.Manager { return c } -func initConversationTags(db *sqlx.DB, lo *logf.Logger) *convtag.Manager { - mgr, err := convtag.New(convtag.Opts{ - DB: db, - Lo: lo, - }) - if err != nil { - log.Fatalf("error initializing conversation tags: %v", err) - } - return mgr -} - func initTags(db *sqlx.DB, lo *logf.Logger) *tag.Manager { mgr, err := tag.New(tag.Opts{ DB: db, @@ -168,6 +158,14 @@ func initContactManager(db *sqlx.DB, lo *logf.Logger) *contact.Manager { return m } +func initTemplateMgr(db *sqlx.DB) *template.Manager { + m, err := template.New(db) + if err != nil { + log.Fatalf("error initializing template manager: %v", err) + } + return m +} + func initMessages(db *sqlx.DB, lo *logf.Logger, wsHub *ws.Hub, @@ -177,7 +175,9 @@ func initMessages(db *sqlx.DB, attachmentMgr *attachment.Manager, conversationMgr *conversation.Manager, inboxMgr *inbox.Manager, - automationEngine *automation.Engine) *message.Manager { + automationEngine *automation.Engine, + templateManager *template.Manager, +) *message.Manager { mgr, err := message.New( wsHub, userMgr, @@ -187,6 +187,7 @@ func initMessages(db *sqlx.DB, inboxMgr, conversationMgr, automationEngine, + templateManager, message.Opts{ DB: db, Lo: lo, @@ -268,28 +269,36 @@ func initAutomationEngine(db *sqlx.DB, lo *logf.Logger) *automation.Engine { return engine } -func initAutoAssignmentEngine(teamMgr *team.Manager, convMgr *conversation.Manager, msgMgr *message.Manager, lo *logf.Logger) *autoassigner.Engine { - engine, err := autoassigner.New(teamMgr, convMgr, msgMgr, lo) +func initAutoAssignmentEngine(teamMgr *team.Manager, convMgr *conversation.Manager, msgMgr *message.Manager, + notifier notifier.Notifier, hub *ws.Hub, lo *logf.Logger) *autoassigner.Engine { + engine, err := autoassigner.New(teamMgr, convMgr, msgMgr, notifier, hub, lo) if err != nil { log.Fatalf("error initializing auto assignment engine: %v", err) } return engine } -func initRBACEngine(db *sqlx.DB) *rbac.Engine { - engine, err := rbac.New(db) +func initRBACEngine(db *sqlx.DB) *uauth.Engine { + engine, err := uauth.New(db, &logf.Logger{}) if err != nil { log.Fatalf("error initializing rbac enginer: %v", err) } return engine } -func initUserFilterMgr(db *sqlx.DB) *filterstore.Manager { - filterMgr, err := filterstore.New(db) - if err != nil { - log.Fatalf("error initializing user filter manager: %v", err) +func initNotifier(userStore notifier.UserStore, templateRenderer notifier.TemplateRenderer) notifier.Notifier { + var smtpCfg email.SMTPConfig + if err := ko.UnmarshalWithConf("notification.provider.email", &smtpCfg, koanf.UnmarshalConf{Tag: "json"}); err != nil { + log.Fatalf("error unmarshalling email notification provider config: %v", err) } - return filterMgr + notifier, err := emailnotifier.New([]email.SMTPConfig{smtpCfg}, userStore, templateRenderer, emailnotifier.Opts{ + Lo: initz.Logger(ko.MustString("app.log_level"), ko.MustString("app.env"), "email-notifier"), + FromEmail: ko.String("notification.provider.email.email_address"), + }) + if err != nil { + log.Fatalf("error initializing email notifier: %v", err) + } + return notifier } // initEmailInbox initializes the email inbox. @@ -327,7 +336,7 @@ func initEmailInbox(inboxRecord inbox.InboxRecord, store inbox.MessageStore) (in return nil, err } - log.Printf("`%s` inbox: `%s` successfully initalized. %d smtp servers. %d imap clients.", inboxRecord.Channel, inboxRecord.Name, len(config.SMTP), len(config.IMAP)) + log.Printf("`%s` inbox successfully initalized. %d smtp servers. %d imap clients.", inboxRecord.Name, len(config.SMTP), len(config.IMAP)) return inbox, nil } diff --git a/cmd/main.go b/cmd/main.go index 6e797ba3..89a7c54b 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -9,18 +9,16 @@ import ( "time" "github.com/abhinavxd/artemis/internal/attachment" + uauth "github.com/abhinavxd/artemis/internal/auth" "github.com/abhinavxd/artemis/internal/cannedresp" "github.com/abhinavxd/artemis/internal/contact" "github.com/abhinavxd/artemis/internal/conversation" - convtag "github.com/abhinavxd/artemis/internal/conversation/tag" "github.com/abhinavxd/artemis/internal/inbox" "github.com/abhinavxd/artemis/internal/initz" "github.com/abhinavxd/artemis/internal/message" - "github.com/abhinavxd/artemis/internal/rbac" "github.com/abhinavxd/artemis/internal/tag" "github.com/abhinavxd/artemis/internal/team" "github.com/abhinavxd/artemis/internal/user" - "github.com/abhinavxd/artemis/internal/user/filterstore" "github.com/abhinavxd/artemis/internal/ws" "github.com/knadh/koanf/v2" "github.com/valyala/fasthttp" @@ -29,25 +27,31 @@ import ( "github.com/zerodha/logf" ) -var ko = koanf.New(".") +var ( + ko = koanf.New(".") +) + +const ( + // ANSI escape colour codes. + colourRed = "\x1b[31m" + colourGreen = "\x1b[32m" +) // App is the global app context which is passed and injected in the http handlers. type App struct { - constants consts - lo *logf.Logger - cntctMgr *contact.Manager - userMgr *user.Manager - teamMgr *team.Manager - sessMgr *simplesessions.Manager - tagMgr *tag.Manager - msgMgr *message.Manager - rbac *rbac.Engine - userFilterMgr *filterstore.Manager - inboxMgr *inbox.Manager - attachmentMgr *attachment.Manager - cannedRespMgr *cannedresp.Manager - conversationMgr *conversation.Manager - conversationTagsMgr *convtag.Manager + constants consts + lo *logf.Logger + cntctMgr *contact.Manager + userMgr *user.Manager + teamMgr *team.Manager + sessMgr *simplesessions.Manager + tagMgr *tag.Manager + msgMgr *message.Manager + rbac *uauth.Engine + inboxMgr *inbox.Manager + attachmentMgr *attachment.Manager + cannedRespMgr *cannedresp.Manager + conversationMgr *conversation.Manager } func main() { @@ -58,62 +62,65 @@ func main() { initz.Config(ko) var ( - shutdownCh = make(chan struct{}) - ctx, stop = signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGINT, syscall.SIGTERM) - lo = initz.Logger(ko.MustString("app.log_level"), ko.MustString("app.env"), "artemis") - rd = initz.Redis(ko) - db = initz.DB(ko) - + shutdownCh = make(chan struct{}) + ctx, stop = signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGINT, syscall.SIGTERM) + lo = initz.Logger(ko.MustString("app.log_level"), ko.MustString("app.env"), "artemis") + rd = initz.Redis(ko) + db = initz.DB(ko) wsHub = ws.NewHub() + templateMgr = initTemplateMgr(db) attachmentMgr = initAttachmentsManager(db, lo) cntctMgr = initContactManager(db, lo) inboxMgr = initInboxManager(db, lo) teamMgr = initTeamMgr(db, lo) userMgr = initUserDB(db, lo) - conversationMgr = initConversations(db, lo) + notifier = initNotifier(userMgr, templateMgr) + conversationMgr = initConversations(wsHub, db, lo) automationEngine = initAutomationEngine(db, lo) - msgMgr = initMessages(db, lo, wsHub, userMgr, teamMgr, cntctMgr, attachmentMgr, conversationMgr, inboxMgr, automationEngine) - autoAssignerEngine = initAutoAssignmentEngine(teamMgr, conversationMgr, msgMgr, lo) + msgMgr = initMessages(db, lo, wsHub, userMgr, teamMgr, cntctMgr, attachmentMgr, conversationMgr, inboxMgr, automationEngine, templateMgr) + autoAssignerEngine = initAutoAssignmentEngine(teamMgr, conversationMgr, msgMgr, notifier, wsHub, lo) ) - // Init the app - var app = &App{ - lo: lo, - cntctMgr: cntctMgr, - inboxMgr: inboxMgr, - userMgr: userMgr, - teamMgr: teamMgr, - attachmentMgr: attachmentMgr, - conversationMgr: conversationMgr, - msgMgr: msgMgr, - constants: initConstants(), - rbac: initRBACEngine(db), - tagMgr: initTags(db, lo), - userFilterMgr: initUserFilterMgr(db), - sessMgr: initSessionManager(rd), - cannedRespMgr: initCannedResponse(db, lo), - conversationTagsMgr: initConversationTags(db, lo), - } - // Register all inboxes with the inbox manager. registerInboxes(inboxMgr, msgMgr) - automationEngine.SetMsgRecorder(msgMgr) - automationEngine.SetConvUpdater(conversationMgr) + // Set conversation store for the websocket hub. + wsHub.SetConversationStore(conversationMgr) + + // Set stores for the automation engine. + automationEngine.SetMessageStore(msgMgr) + automationEngine.SetConversationStore(conversationMgr) // Start receivers for all active inboxes. 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. + // Start automation rule evaluation engine. go automationEngine.Serve(ctx) // Start conversation auto assigner engine. go autoAssignerEngine.Serve(ctx, ko.MustDuration("autoassigner.assign_interval")) + // Start inserting incoming messages from all active inboxes and dispatch pending outgoing messages. + go msgMgr.StartDBInserts(ctx, ko.MustInt("message.reader_concurrency")) + go msgMgr.StartDispatcher(ctx, ko.MustInt("message.dispatch_concurrency"), ko.MustDuration("message.dispatch_read_interval")) + + // Init the app + var app = &App{ + lo: lo, + cntctMgr: cntctMgr, + inboxMgr: inboxMgr, + userMgr: userMgr, + teamMgr: teamMgr, + attachmentMgr: attachmentMgr, + conversationMgr: conversationMgr, + msgMgr: msgMgr, + constants: initConstants(), + rbac: initRBACEngine(db), + tagMgr: initTags(db, lo), + sessMgr: initSessionManager(rd), + cannedRespMgr: initCannedResponse(db, lo), + } + // Init fastglue http server. g := fastglue.NewGlue() @@ -137,20 +144,18 @@ func main() { // Wait for the interruption signal <-ctx.Done() - log.Printf("\x1b[%dm%s\x1b[0m", 31, "Shutting down the server please wait...") + log.Printf("%sShutting down the server. Please wait.\x1b[0m", colourRed) - // Additional grace period before triggering shutdown - time.Sleep(7 * time.Second) + time.Sleep(5 * time.Second) // Signal to shutdown the server shutdownCh <- struct{}{} stop() }() - // 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")) + log.Printf("%s🚀 server listening on %s %s\x1b[0m", colourGreen, 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("Server shutdown completed") } diff --git a/cmd/messages.go b/cmd/messages.go index 67022bc9..eeadcaa7 100644 --- a/cmd/messages.go +++ b/cmd/messages.go @@ -3,7 +3,6 @@ package main import ( "encoding/json" "net/http" - "time" "github.com/abhinavxd/artemis/internal/attachment/models" "github.com/abhinavxd/artemis/internal/message" @@ -108,13 +107,10 @@ func handleSendMessage(r *fastglue.Request) error { // Update conversation meta with the last message details. trimmedMessage := app.msgMgr.TrimMsg(msg.Content) - app.conversationMgr.UpdateMeta(0, conversationUUID, map[string]string{ - "last_message": trimmedMessage, - "last_message_at": msg.CreatedAt.Format(time.RFC3339), - }) + app.conversationMgr.UpdateLastMessage(0, conversationUUID, trimmedMessage, msg.CreatedAt) // Send WS update. - app.msgMgr.BroadcastNewMsg(msg, trimmedMessage) + app.msgMgr.BroadcastNewConversationMessage(msg, trimmedMessage) return r.SendEnvelope("Message sent") } diff --git a/cmd/users.go b/cmd/users.go index cdb0766e..9885dc18 100644 --- a/cmd/users.go +++ b/cmd/users.go @@ -28,16 +28,3 @@ func handleGetCurrentUser(r *fastglue.Request) error { } return r.SendEnvelope(u) } - -func handleGetUserFilters(r *fastglue.Request) error { - var ( - app = r.Context.(*App) - userID = r.RequestCtx.UserValue("user_id").(int) - page = r.RequestCtx.UserValue("page").(string) - ) - filters, err := app.userFilterMgr.GetFilters(userID, page) - if err != nil { - return r.SendErrorEnvelope(http.StatusInternalServerError, err.Error(), nil, "") - } - return r.SendEnvelope(filters) -} diff --git a/cmd/websocket.go b/cmd/websocket.go index e81d4f82..808bd556 100644 --- a/cmd/websocket.go +++ b/cmd/websocket.go @@ -34,23 +34,9 @@ func handleWS(r *fastglue.Request, hub *ws.Hub) error { ID: userID, Hub: hub, Conn: conn, - Send: make(chan ws.Message, 100000), + Send: make(chan ws.Message, 10000), } - - // Sub this client to all assigned conversations. - convs, err := app.conversationMgr.GetAssignedConversations(userID) - if err != nil { - return - } - // Extract uuids. - uuids := make([]string, len(convs)) - for i, conv := range convs { - uuids[i] = conv.UUID - } - c.SubConv(userID, uuids...) - hub.AddClient(&c) - go c.Listen() c.Serve(2 * time.Second) }) diff --git a/frontend/bun.lockb b/frontend/bun.lockb new file mode 100755 index 00000000..50353a5f Binary files /dev/null and b/frontend/bun.lockb differ diff --git a/frontend/package.json b/frontend/package.json index 2d74a39e..7b4aeb36 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -45,6 +45,7 @@ "tailwindcss-animate": "^1.0.7", "tiptap-extension-resize-image": "^1.1.5", "vue": "^3.4.15", + "vue-draggable-resizable": "^3.0.0", "vue-i18n": "9", "vue-letter": "^0.2.0", "vue-router": "^4.2.5" diff --git a/frontend/src/App.vue b/frontend/src/App.vue index 99ca1bae..f12cd5f0 100644 --- a/frontend/src/App.vue +++ b/frontend/src/App.vue @@ -29,10 +29,10 @@ import { ref, onMounted } from "vue" import { RouterView, useRouter } from 'vue-router' import { cn } from '@/lib/utils' import { useUserStore } from '@/stores/user' -import { initWS } from "./websocket.js" +import { initWS } from "@/websocket.js" import { Toaster } from '@/components/ui/toast' -import NavBar from './components/NavBar.vue' +import NavBar from '@/components/NavBar.vue' import { ResizableHandle, ResizablePanel, @@ -43,7 +43,6 @@ import { } from '@/components/ui/tooltip' - const isCollapsed = ref(false) const navLinks = [ { diff --git a/frontend/src/assets/styles/main.scss b/frontend/src/assets/styles/main.scss index f03f712c..4aae2c0f 100644 --- a/frontend/src/assets/styles/main.scss +++ b/frontend/src/assets/styles/main.scss @@ -174,10 +174,9 @@ $editorContainerId: 'editor-container'; } } -.tiptap-editor-image { - width: 100px; - height: 200px; -} +// .tiptap-editor-image { +// width: 60%; +// } .box { box-shadow: rgb(243, 243, 243) 1px 1px 0px 0px; diff --git a/frontend/src/components/conversation/Conversation.vue b/frontend/src/components/conversation/Conversation.vue index c5a51c61..9887127b 100644 --- a/frontend/src/components/conversation/Conversation.vue +++ b/frontend/src/components/conversation/Conversation.vue @@ -1,8 +1,8 @@ diff --git a/frontend/src/components/conversation/ConversationSideBar.vue b/frontend/src/components/conversation/ConversationSideBar.vue index f8d04563..eb703ee8 100644 --- a/frontend/src/components/conversation/ConversationSideBar.vue +++ b/frontend/src/components/conversation/ConversationSideBar.vue @@ -315,7 +315,7 @@ onMounted(() => { }) const handleAssignedAgentChange = (v) => { - conversationStore.updateAssignee("agent", { + conversationStore.updateAssignee("user", { "assignee_uuid": v.split(":")[0] }) } diff --git a/frontend/src/components/conversation/ReplyBox.vue b/frontend/src/components/conversation/ReplyBox.vue index a95bf459..6ff8bb3d 100644 --- a/frontend/src/components/conversation/ReplyBox.vue +++ b/frontend/src/components/conversation/ReplyBox.vue @@ -1,5 +1,6 @@ diff --git a/frontend/src/constants/conversation.js b/frontend/src/constants/conversation.js index b3f0e110..9dbb7703 100644 --- a/frontend/src/constants/conversation.js +++ b/frontend/src/constants/conversation.js @@ -1,4 +1,5 @@ export const CONVERSATION_PRE_DEFINED_FILTERS = { + ALL: "status_all", STATUS_OPEN: 'status_open', STATUS_PROCESSING: 'status_processing', STATUS_SPAM: 'status_spam', diff --git a/frontend/src/stores/conversation.js b/frontend/src/stores/conversation.js index cd705c4c..fd72ffd4 100644 --- a/frontend/src/stores/conversation.js +++ b/frontend/src/stores/conversation.js @@ -266,22 +266,39 @@ export const useConversationStore = defineStore('conversation', () => { } } } - function updateMessageList (msg) { + + function updateConversationProp (update) { + if (conversation?.data?.uuid === update.uuid) { + conversation.data[update.prop] = update.val + } + } + + function addNewConversation (conv) { + if (!conversationUUIDExists(conv.uuid)) { + conversations.data.push(conv) + } + } + + function conversationUUIDExists (uuid) { + return conversations.data?.find(c => c.uuid === uuid) ? true : false + } + + function updateMessageList (message) { // Check if this conversation is selected and then update messages list. - if (conversation?.data?.uuid === msg.conversation_uuid) { - // Fetch entire msg if the give msg does not exist in the msg list. - if (!messages.data.some(message => message.uuid === msg.uuid)) { - fetchParticipants(msg.conversation_uuid) - fetchMessage(msg.uuid) - updateAssigneeLastSeen(msg.conversation_uuid) + if (conversation?.data?.uuid === message.conversation_uuid) { + // Fetch entire message if the give msg does not exist in the msg list. + if (!messages.data.some(msg => msg.uuid === message.uuid)) { + fetchParticipants(message.conversation_uuid) + fetchMessage(message.uuid) + updateAssigneeLastSeen(message.conversation_uuid) } } } - function updateMessageStatus (uuid, status) { - const message = messages.data.find(m => m.uuid === uuid) - if (message) { - message.status = status + function updateMessageProp (message) { + const existingMessage = messages.data.find(m => m.uuid === message.uuid) + if (existingMessage) { + existingMessage[message.prop] = message.val } } @@ -305,5 +322,5 @@ export const useConversationStore = defineStore('conversation', () => { messages.errorMessage = "" } - return { conversations, conversation, messages, sortedConversations, sortedMessages, getContactFullName, fetchParticipants, fetchNextConversations, updateMessageStatus, updateAssigneeLastSeen, updateMessageList, fetchConversation, fetchConversations, fetchMessages, upsertTags, updateAssignee, updatePriority, updateStatus, updateConversationList, $reset }; + return { conversations, conversation, messages, sortedConversations, sortedMessages, conversationUUIDExists, updateConversationProp, addNewConversation, getContactFullName, fetchParticipants, fetchNextConversations, updateMessageProp, updateAssigneeLastSeen, updateMessageList, fetchConversation, fetchConversations, fetchMessages, upsertTags, updateAssignee, updatePriority, updateStatus, updateConversationList, $reset }; }) diff --git a/frontend/src/views/ConversationView.vue b/frontend/src/views/ConversationView.vue index 5cd35748..e4ebd0e9 100644 --- a/frontend/src/views/ConversationView.vue +++ b/frontend/src/views/ConversationView.vue @@ -18,6 +18,8 @@ diff --git a/frontend/src/views/UserLoginView.vue b/frontend/src/views/UserLoginView.vue index f4562256..d72b3a61 100644 --- a/frontend/src/views/UserLoginView.vue +++ b/frontend/src/views/UserLoginView.vue @@ -9,7 +9,7 @@ -
+
- +
- +
@@ -43,7 +44,7 @@
- Forgot ID or Password? + Forgot Email or Password?
diff --git a/frontend/src/websocket.js b/frontend/src/websocket.js index ce3c2afb..876e38ba 100644 --- a/frontend/src/websocket.js +++ b/frontend/src/websocket.js @@ -1,52 +1,145 @@ -import { useConversationStore } from "./stores/conversation"; +import { useConversationStore } from './stores/conversation' -export function initWS () { +let socket +let reconnectInterval = 1000 // Initial reconnection interval +let maxReconnectInterval = 30000 // Maximum reconnection interval +let reconnectTimeout +let isReconnecting = false +let manualClose = false + +export function initWS() { let convStore = useConversationStore() - // Create a new WebSocket connection to the specified WebSocket URL - const socket = new WebSocket('ws://localhost:9009/api/ws'); + // Initialize the WebSocket connection + function initializeWebSocket() { + socket = new WebSocket('ws://localhost:9009/api/ws') - // Connection opened event - socket.addEventListener('open', function (event) { - // Send a message to the server once the connection is opened - socket.send('Hello, server!!'); - }); - - // Listen for messages from the server - socket.addEventListener('message', function (e) { - console.log('Message from server !', e.data); - if (e.data) { - let event = JSON.parse(e.data) - switch (event.ev) { - case "new_msg": - convStore.updateConversationList(event.d) - convStore.updateMessageList(event.d) - break - case "msg_status_update": - convStore.updateMessageStatus(event.d.uuid, event.d.status) - break - default: - console.log(`Unknown event ${event.ev}`); + // Connection opened event + socket.addEventListener('open', function () { + console.log('WebSocket connection established') + reconnectInterval = 1000 // Reset the reconnection interval + if (reconnectTimeout) { + clearTimeout(reconnectTimeout) // Clear any existing reconnection timeout + reconnectTimeout = null } + }) + + // Listen for messages from the server + socket.addEventListener('message', function (e) { + console.log('Message from server:', e.data) + if (e.data) { + let event = JSON.parse(e.data) + // TODO: move event type to consts. + switch (event.typ) { + case 'new_msg': + convStore.updateConversationList(event.d) + convStore.updateMessageList(event.d) + break + case 'msg_prop_update': + convStore.updateMessageProp(event.d) + break + case 'new_conv': + convStore.addNewConversation(event.d) + break + case 'conv_prop_update': + convStore.updateConversationProp(event.d) + break + default: + console.log(`Unknown event ${event.ev}`) + } + } + }) + + // Handle possible errors + socket.addEventListener('error', function (event) { + console.error('WebSocket error observed:', event) + }) + + // Handle the connection close event + socket.addEventListener('close', function (event) { + console.log('WebSocket connection closed:', event) + if (!manualClose) { + reconnect() + } + }) + } + + // Start the initial WebSocket connection + initializeWebSocket() + + // Reconnect logic + function reconnect() { + if (isReconnecting) return + isReconnecting = true + reconnectTimeout = setTimeout(() => { + initializeWebSocket() + reconnectInterval = Math.min(reconnectInterval * 2, maxReconnectInterval) + isReconnecting = false + }, reconnectInterval) + } + + // Detect network status and handle reconnection + window.addEventListener('online', () => { + if (!isReconnecting && socket.readyState !== WebSocket.OPEN) { + reconnectInterval = 1000 // Reset reconnection interval + reconnect() } - }); + }) +} - // Handle possible errors - socket.addEventListener('error', function (event) { - console.error('WebSocket error observed:', event); - console.log('WebSocket readyState:', socket.readyState); // Log the state of the WebSocket - }); +function waitForWebSocketOpen(socket, callback) { + if (socket.readyState === WebSocket.OPEN) { + callback() + } else { + socket.addEventListener('open', function handler() { + socket.removeEventListener('open', handler) + callback() + }) + } +} +export function sendMessage(message) { + waitForWebSocketOpen(socket, () => { + socket.send(JSON.stringify(message)) + }) +} - // Handle the connection close event - socket.addEventListener('close', function (event) { - console.log('WebSocket connection closed!:', event); - }); +export function subscribeConversations(type, preDefinedFilter) { + let message = { + a: 'conversations_sub', + t: type, + pf: preDefinedFilter + } - socket.onerror = function (event) { - console.error("WebSocket error:", event); - }; - socket.onclose = function (event) { - console.log("WebSocket connection closed:", event); - }; -} \ No newline at end of file + waitForWebSocketOpen(socket, () => { + socket.send(JSON.stringify(message)) + }) +} + +export function subscribeConversation(uuid) { + if (!uuid) { + return + } + let message = { + a: 'conversation_sub', + uuid: uuid + } + + waitForWebSocketOpen(socket, () => { + socket.send(JSON.stringify(message)) + }) +} + +export function unsubscribeConversation(uuid) { + if (!uuid) { + return + } + let message = { + a: 'conversation_unsub', + uuid: uuid + } + + waitForWebSocketOpen(socket, () => { + socket.send(JSON.stringify(message)) + }) +} diff --git a/internal/attachment/attachment.go b/internal/attachment/attachment.go index fe64744b..61510f0c 100644 --- a/internal/attachment/attachment.go +++ b/internal/attachment/attachment.go @@ -8,7 +8,7 @@ import ( "net/textproto" "github.com/abhinavxd/artemis/internal/attachment/models" - "github.com/abhinavxd/artemis/internal/dbutils" + "github.com/abhinavxd/artemis/internal/dbutil" "github.com/jmoiron/sqlx" "github.com/zerodha/logf" ) @@ -56,7 +56,7 @@ func New(opt Opts) (*Manager, error) { // Scan SQL file - if err := dbutils.ScanSQLFile("queries.sql", &q, opt.DB, efs); err != nil { + if err := dbutil.ScanSQLFile("queries.sql", &q, opt.DB, efs); err != nil { return nil, err } return &Manager{ diff --git a/internal/auditlog/auditlog.go b/internal/auditlog/auditlog.go deleted file mode 100644 index 7fd2171e..00000000 --- a/internal/auditlog/auditlog.go +++ /dev/null @@ -1 +0,0 @@ -package auditlog \ No newline at end of file diff --git a/internal/rbac/rbac.go b/internal/auth/auth.go similarity index 57% rename from internal/rbac/rbac.go rename to internal/auth/auth.go index 217036c6..0e17cb9a 100644 --- a/internal/rbac/rbac.go +++ b/internal/auth/auth.go @@ -1,10 +1,11 @@ -package rbac +package auth import ( "embed" - "github.com/abhinavxd/artemis/internal/dbutils" + "github.com/abhinavxd/artemis/internal/dbutil" "github.com/jmoiron/sqlx" + "github.com/zerodha/logf" ) var ( @@ -13,31 +14,30 @@ var ( ) type Engine struct { - q queries + q queries + lo *logf.Logger } type queries struct { HasPermission *sqlx.Stmt `query:"has-permission"` } -func New(db *sqlx.DB) (*Engine, error) { +func New(db *sqlx.DB, lo *logf.Logger) (*Engine, error) { var q queries - - if err := dbutils.ScanSQLFile("queries.sql", &q, db, efs); err != nil { + if err := dbutil.ScanSQLFile("queries.sql", &q, db, efs); err != nil { return nil, err } - return &Engine{ - q: q, + q: q, + lo: lo, }, nil } func (e *Engine) HasPermission(userID int, perm string) (bool, error) { var hasPerm bool - if err := e.q.HasPermission.Get(&hasPerm, userID, perm); err != nil { + e.lo.Error("error fetching user permissions", "user_id", userID, "error", err) return hasPerm, err } - return hasPerm, nil } diff --git a/internal/rbac/queries.sql b/internal/auth/queries.sql similarity index 100% rename from internal/rbac/queries.sql rename to internal/auth/queries.sql diff --git a/internal/autoassigner/autoassigner.go b/internal/autoassigner/autoassigner.go index f65fd5d2..fa312f85 100644 --- a/internal/autoassigner/autoassigner.go +++ b/internal/autoassigner/autoassigner.go @@ -8,8 +8,10 @@ import ( "github.com/abhinavxd/artemis/internal/conversation" "github.com/abhinavxd/artemis/internal/conversation/models" "github.com/abhinavxd/artemis/internal/message" + notifier "github.com/abhinavxd/artemis/internal/notification" "github.com/abhinavxd/artemis/internal/systeminfo" "github.com/abhinavxd/artemis/internal/team" + "github.com/abhinavxd/artemis/internal/ws" "github.com/mr-karan/balance" "github.com/zerodha/logf" ) @@ -17,71 +19,48 @@ import ( const ( roundRobinDefaultWeight = 1 strategyRoundRobin = "round_robin" - strategyLoadBalances = "load_balanced" ) -// Engine handles the assignment of unassigned conversations to agents using a round-robin strategy. +// Engine handles the assignment of unassigned conversations to agents of a tean using a round-robin strategy. type Engine struct { teamRoundRobinBalancer map[int]*balance.Balance - mu sync.Mutex // Mutex to protect the balancer map - convMgr *conversation.Manager - teamMgr *team.Manager - msgMgr *message.Manager - lo *logf.Logger - strategy string + userIDs map[string]int + // Mutex to protect the balancer map + mu sync.Mutex + convMgr *conversation.Manager + teamMgr *team.Manager + msgMgr *message.Manager + lo *logf.Logger + hub *ws.Hub + notifier notifier.Notifier + strategy string } // New creates a new instance of the Engine. -func New(teamMgr *team.Manager, convMgr *conversation.Manager, msgMgr *message.Manager, lo *logf.Logger) (*Engine, error) { - balance, err := populateBalancerPool(teamMgr) +func New(teamMgr *team.Manager, convMgr *conversation.Manager, msgMgr *message.Manager, + notifier notifier.Notifier, hub *ws.Hub, lo *logf.Logger) (*Engine, error) { + var e = Engine{ + notifier: notifier, + strategy: strategyRoundRobin, + convMgr: convMgr, + teamMgr: teamMgr, + msgMgr: msgMgr, + lo: lo, + hub: hub, + mu: sync.Mutex{}, + userIDs: map[string]int{}, + } + balancer, err := e.populateBalancerPool() if err != nil { return nil, err } - return &Engine{ - teamRoundRobinBalancer: balance, - strategy: strategyRoundRobin, - convMgr: convMgr, - teamMgr: teamMgr, - msgMgr: msgMgr, - lo: lo, - mu: sync.Mutex{}, - }, nil -} - -func populateBalancerPool(teamMgr *team.Manager) (map[int]*balance.Balance, error) { - var ( - balancer = make(map[int]*balance.Balance) - teams, err = teamMgr.GetAll() - ) - - if err != nil { - return nil, err - } - - for _, team := range teams { - 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() - } - balancer[team.ID].Add(user.UUID, roundRobinDefaultWeight) - } - } - return balancer, nil + e.teamRoundRobinBalancer = balancer + return &e, nil } func (e *Engine) Serve(ctx context.Context, interval time.Duration) { - // Start updating the balancer pool periodically in a separate goroutine - go e.refreshBalancerPeriodically(ctx, 1*time.Minute) - ticker := time.NewTicker(interval) defer ticker.Stop() - for { select { case <-ctx.Done(): @@ -94,67 +73,11 @@ func (e *Engine) Serve(ctx context.Context, interval time.Duration) { } } -// assignConversations fetches unassigned conversations and assigns them. -func (e *Engine) assignConversations() error { - unassignedConv, err := e.convMgr.GetUnassigned() - if err != nil { - return err - } - - if len(unassignedConv) > 0 { - e.lo.Debug("found unassigned conversations", "count", len(unassignedConv)) - } - - for _, conv := range unassignedConv { - if e.strategy == strategyRoundRobin { - e.roundRobin(conv) - } - } - return nil -} - -// roundRobin fetches an user from the team balancer pool and assigns the conversation to that user. -func (e *Engine) roundRobin(conv models.Conversation) { - pool, ok := e.teamRoundRobinBalancer[conv.AssignedTeamID.Int] - if !ok { - e.lo.Warn("team not found in balancer", "id", conv.AssignedTeamID.Int) - } - userUUID := pool.Get() - e.lo.Debug("fetched user from rr pool for assignment", "user_uuid", userUUID) - if userUUID == "" { - e.lo.Warn("empty user returned from rr pool") - return - } - if err := e.convMgr.UpdateAssignee(conv.UUID, []byte(userUUID), "agent"); err != nil { - e.lo.Error("error updating conversation assignee", "error", err, "conv_uuid", conv.UUID, "user_uuid", userUUID) - return - } - if err := e.msgMgr.RecordAssigneeUserChange(userUUID, conv.UUID, systeminfo.SystemUserUUID); err != nil { - e.lo.Error("error recording conversation user change msg", "error", err, "conv_uuid", conv.UUID, "user_uuid", userUUID) - } -} - -func (e *Engine) refreshBalancerPeriodically(ctx context.Context, updateInterval time.Duration) { - ticker := time.NewTicker(updateInterval) - defer ticker.Stop() - - for { - select { - case <-ctx.Done(): - return - case <-ticker.C: - if err := e.refreshBalancer(); err != nil { - e.lo.Error("Error updating team balancer pool", "error", err) - } - } - } -} - -func (e *Engine) refreshBalancer() error { +func (e *Engine) RefreshBalancer() error { e.mu.Lock() defer e.mu.Unlock() - balancer, err := populateBalancerPool(e.teamMgr) + balancer, err := e.populateBalancerPool() if err != nil { e.lo.Error("Error updating team balancer pool", "error", err) return err @@ -162,3 +85,79 @@ func (e *Engine) refreshBalancer() error { e.teamRoundRobinBalancer = balancer return nil } + +// populateBalancerPool populates the team balancer bool with the team members. +func (e *Engine) populateBalancerPool() (map[int]*balance.Balance, error) { + var ( + balancer = make(map[int]*balance.Balance) + teams, err = e.teamMgr.GetAll() + ) + + if err != nil { + return nil, err + } + + for _, team := range teams { + users, err := e.teamMgr.GetTeamMembers(team.Name) + if err != nil { + return nil, err + } + + // Add the users to team balance map. + for _, user := range users { + if _, ok := balancer[team.ID]; !ok { + balancer[team.ID] = balance.NewBalance() + } + // FIXME: Balancer only supports strings. + balancer[team.ID].Add(user.UUID, roundRobinDefaultWeight) + e.userIDs[user.UUID] = user.ID + } + } + return balancer, nil +} + +// assignConversations fetches unassigned conversations and assigns them. +func (e *Engine) assignConversations() error { + unassignedConversations, err := e.convMgr.GetUnassigned() + if err != nil { + return err + } + + if len(unassignedConversations) > 0 { + e.lo.Debug("found unassigned conversations", "count", len(unassignedConversations)) + } + + for _, conversation := range unassignedConversations { + if e.strategy == strategyRoundRobin { + userUUID := e.getUser(conversation) + if userUUID == "" { + e.lo.Warn("user uuid not found for round robin assignment", "team_id", conversation.AssignedTeamID.Int) + continue + } + + // Update assignee and record the assigne change message. + if err := e.convMgr.UpdateUserAssignee(conversation.UUID, []byte(userUUID)); err != nil { + continue + } + + // Fixme: maybe move to messages? + e.hub.BroadcastConversationAssignment(e.userIDs[userUUID], conversation.UUID, conversation.AvatarURL.String, conversation.FirstName, conversation.LastName, conversation.LastMessage, conversation.InboxName, conversation.LastMessageAt.Time, 1) + + e.msgMgr.RecordAssigneeUserChange(conversation.UUID, userUUID, systeminfo.SystemUserUUID) + + // Send notification to the assignee. + e.notifier.SendAssignedConversationNotification([]string{userUUID}, conversation.UUID) + } + } + return nil +} + +// getUser returns user uuid from the team balancer pool. +func (e *Engine) getUser(conversation models.Conversation) string { + pool, ok := e.teamRoundRobinBalancer[conversation.AssignedTeamID.Int] + if !ok { + e.lo.Warn("team not found in balancer", "id", conversation.AssignedTeamID.Int) + return "" + } + return pool.Get() +} diff --git a/internal/automation/automation.go b/internal/automation/automation.go index 0b05f6b2..3529ac8f 100644 --- a/internal/automation/automation.go +++ b/internal/automation/automation.go @@ -3,11 +3,11 @@ package automation import ( "context" "embed" - "fmt" + "encoding/json" "github.com/abhinavxd/artemis/internal/automation/models" cmodels "github.com/abhinavxd/artemis/internal/conversation/models" - "github.com/abhinavxd/artemis/internal/dbutils" + "github.com/abhinavxd/artemis/internal/dbutil" "github.com/jmoiron/sqlx" "github.com/zerodha/logf" ) @@ -17,29 +17,13 @@ var ( 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 - convUpdater ConversationUpdater - msgRecorder MessageRecorder - conversationQ chan cmodels.Conversation - rules []models.Rule - actions []models.Action -} - -type ConversationUpdater interface { - UpdateAssignee(uuid string, assigneeUUID []byte, assigneeType string) error - UpdateStatus(uuid string, status []byte) error -} - -type MessageRecorder interface { - RecordAssigneeUserChange(updatedValue, convUUID, actorUUID string) error - RecordStatusChange(updatedValue, convUUID, actorUUID string) error + q queries + lo *logf.Logger + conversationStore ConversationStore + messageStore MessageStore + rules []models.Rule + conversationQ chan cmodels.Conversation } type Opts struct { @@ -47,6 +31,20 @@ type Opts struct { Lo *logf.Logger } +type ConversationStore interface { + UpdateTeamAssignee(uuid string, assigneeUUID []byte) error + UpdateStatus(uuid string, status []byte) error +} + +type MessageStore interface { + RecordAssigneeTeamChange(convUUID, value, actorUUID string) error + RecordStatusChange(updatedValue, convUUID, actorUUID string) error +} + +type queries struct { + GetRules *sqlx.Stmt `query:"get-rules"` +} + func New(opt Opts) (*Engine, error) { var ( q queries @@ -55,29 +53,24 @@ func New(opt Opts) (*Engine, error) { conversationQ: make(chan cmodels.Conversation, 10000), } ) - - if err := dbutils.ScanSQLFile("queries.sql", &q, opt.DB, efs); err != nil { + if err := dbutil.ScanSQLFile("queries.sql", &q, opt.DB, efs); err != nil { return nil, err } - - // Fetch rules and actions from the DB. - 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 - + e.rules = e.getRules() return e, nil } -func (e *Engine) SetMsgRecorder(msgRecorder MessageRecorder) { - e.msgRecorder = msgRecorder +func (e *Engine) ReloadRules() { + e.rules = e.getRules() } -func (e *Engine) SetConvUpdater(convUpdater ConversationUpdater) { - e.convUpdater = convUpdater +func (e *Engine) SetMessageStore(messageStore MessageStore) { + e.messageStore = messageStore +} + +func (e *Engine) SetConversationStore(conversationStore ConversationStore) { + e.conversationStore = conversationStore } func (e *Engine) Serve(ctx context.Context) { @@ -85,12 +78,39 @@ func (e *Engine) Serve(ctx context.Context) { select { case <-ctx.Done(): return - case conv := <-e.conversationQ: - e.processConversations(conv) + case conversation := <-e.conversationQ: + e.processConversations(conversation) } } } func (e *Engine) EvaluateRules(c cmodels.Conversation) { - e.conversationQ <- c + select { + case e.conversationQ <- c: + default: + // Queue is full. + e.lo.Warn("EvaluateRules: conversationQ is full, unable to enqueue conversation") + } +} + +func (e *Engine) getRules() []models.Rule { + var rulesJSON []string + err := e.q.GetRules.Select(&rulesJSON) + if err != nil { + e.lo.Error("error fetching automation rules", "error", err) + return nil + } + + var rules []models.Rule + for _, ruleJSON := range rulesJSON { + var rulesBatch []models.Rule + if err := json.Unmarshal([]byte(ruleJSON), &rulesBatch); err != nil { + e.lo.Error("error unmarshalling rule JSON", "error", err) + continue + } + rules = append(rules, rulesBatch...) + } + + e.lo.Debug("fetched rules", "num", len(rules), "rules", rules) + return rules } diff --git a/internal/automation/models/models.go b/internal/automation/models/models.go index e691e278..48cddfb4 100644 --- a/internal/automation/models/models.go +++ b/internal/automation/models/models.go @@ -3,22 +3,28 @@ package models const ( ActionAssignTeam = "assign_team" ActionAssignAgent = "assign_agent" - - RuleTypeNewConversation = "new_conversation" + OperatorAnd = "AND" + OperatorOR = "OR" ) 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"` + GroupOperator string `json:"group_operator" db:"group_operator"` + Groups []RuleGroup `json:"groups" db:"groups"` + Actions []RuleAction `json:"actions" db:"actions"` } -type Action struct { - RuleID int `db:"rule_id"` - Type string `db:"action_type"` - Action string `db:"action"` +type RuleGroup struct { + LogicalOp string `json:"logical_op" db:"logical_op"` + Rules []RuleDetail `json:"rules" db:"rules"` +} + +type RuleDetail struct { + Field string `json:"field" db:"field"` + Operator string `json:"operator" db:"operator"` + Value string `json:"value" db:"value"` +} + +type RuleAction struct { + Type string `json:"action_type" db:"action_type"` + Action string `json:"action" db:"action"` } diff --git a/internal/automation/preprocessor.go b/internal/automation/preprocessor.go deleted file mode 100644 index 93f3bc4c..00000000 --- a/internal/automation/preprocessor.go +++ /dev/null @@ -1,142 +0,0 @@ -package automation - -import ( - "fmt" - "strings" - - "github.com/abhinavxd/artemis/internal/automation/models" - cmodels "github.com/abhinavxd/artemis/internal/conversation/models" - "github.com/abhinavxd/artemis/internal/systeminfo" -) - -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: - if err := e.convUpdater.UpdateAssignee(conv.UUID, []byte(action.Action), "team"); err != nil { - return err - } - if err := e.msgRecorder.RecordAssigneeUserChange(action.Action, conv.UUID, systeminfo.SystemUserUUID); err != nil { - return err - } - case models.ActionAssignAgent: - if err := e.convUpdater.UpdateStatus(conv.UUID, []byte(action.Action)); err != nil { - return err - } - if err := e.msgRecorder.RecordStatusChange(action.Action, conv.UUID, systeminfo.SystemUserUUID); err != nil { - return err - } - default: - return fmt.Errorf("rule action not recognized: %s", action.Type) - } - return nil -} diff --git a/internal/automation/processor.go b/internal/automation/processor.go new file mode 100644 index 00000000..befeccb1 --- /dev/null +++ b/internal/automation/processor.go @@ -0,0 +1,153 @@ +package automation + +import ( + "fmt" + "strings" + + "github.com/abhinavxd/artemis/internal/automation/models" + cmodels "github.com/abhinavxd/artemis/internal/conversation/models" + "github.com/abhinavxd/artemis/internal/systeminfo" +) + +func (e *Engine) processConversations(conversation cmodels.Conversation) { + e.lo.Debug("num rules", "rules", len(e.rules)) + for _, rule := range e.rules { + e.lo.Debug("eval rule", "groups", len(rule.Groups), "rule", rule) + if len(rule.Groups) > 2 { + continue + } + var results []bool + for _, group := range rule.Groups { + e.lo.Debug("evaluating group rule", "logical_op", group.LogicalOp) + result := e.evaluateGroup(group.Rules, group.LogicalOp, conversation) + e.lo.Debug("group evaluation status", "status", result) + results = append(results, result) + } + if evaluateFinalResult(results, rule.GroupOperator) { + e.lo.Debug("rule fully evalauted, executing actions") + // All group rule evaluations successful, execute the actions. + for _, action := range rule.Actions { + e.executeActions(conversation, action) + } + } + } +} + +// evaluateFinalResult +func evaluateFinalResult(results []bool, operator string) bool { + if operator == models.OperatorAnd { + for _, result := range results { + if !result { + return false + } + } + return true + } + if operator == models.OperatorOR { + for _, result := range results { + if result { + return true + } + } + return false + } + return false +} + +// evaluateGroup +func (e *Engine) evaluateGroup(rules []models.RuleDetail, operator string, conversation cmodels.Conversation) bool { + switch operator { + case models.OperatorAnd: + // All conditions within the group must be true + for _, rule := range rules { + if !e.evaluateRule(rule, conversation) { + return false + } + } + return true + case models.OperatorOR: + // At least one condition within the group must be true + for _, rule := range rules { + if e.evaluateRule(rule, conversation) { + return true + } + } + return false + default: + e.lo.Error("invalid group operator", "operator", operator) + } + return false +} + +func (e *Engine) evaluateRule(rule models.RuleDetail, conversation 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 = conversation.Subject + case "content": + conversationValue = conversation.FirstMessage + case "status": + conversationValue = conversation.Status.String + case "priority": + conversationValue = conversation.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": + 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(conversation cmodels.Conversation, action models.RuleAction) { + err := e.applyAction(action, conversation) + if err != nil { + e.lo.Error("error executing rule action", "action", action.Action, "error", err) + } +} + +func (e *Engine) applyAction(action models.RuleAction, conversation cmodels.Conversation) error { + switch action.Type { + case models.ActionAssignTeam: + if err := e.conversationStore.UpdateTeamAssignee(conversation.UUID, []byte(action.Action)); err != nil { + return err + } + if err := e.messageStore.RecordAssigneeTeamChange(conversation.UUID, action.Action, systeminfo.SystemUserUUID); err != nil { + return err + } + case models.ActionAssignAgent: + if err := e.conversationStore.UpdateStatus(conversation.UUID, []byte(action.Action)); err != nil { + return err + } + if err := e.messageStore.RecordStatusChange(action.Action, conversation.UUID, systeminfo.SystemUserUUID); err != nil { + return err + } + default: + return fmt.Errorf("unrecognized rule action: %s", action.Type) + } + return nil +} diff --git a/internal/automation/queries.sql b/internal/automation/queries.sql index 3d2768a7..b509615f 100644 --- a/internal/automation/queries.sql +++ b/internal/automation/queries.sql @@ -1,6 +1,3 @@ -- 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; \ No newline at end of file +select rules +from automation_rules; \ No newline at end of file diff --git a/internal/cannedresp/cannedresp.go b/internal/cannedresp/cannedresp.go index 506f573d..97f1c083 100644 --- a/internal/cannedresp/cannedresp.go +++ b/internal/cannedresp/cannedresp.go @@ -4,7 +4,7 @@ import ( "embed" "fmt" - "github.com/abhinavxd/artemis/internal/dbutils" + "github.com/abhinavxd/artemis/internal/dbutil" "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 := dbutils.ScanSQLFile("queries.sql", &q, opts.DB, efs); err != nil { + if err := dbutil.ScanSQLFile("queries.sql", &q, opts.DB, efs); err != nil { return nil, err } diff --git a/internal/contact/contact.go b/internal/contact/contact.go index 711794fc..e2caef47 100644 --- a/internal/contact/contact.go +++ b/internal/contact/contact.go @@ -4,7 +4,7 @@ import ( "embed" "github.com/abhinavxd/artemis/internal/contact/models" - "github.com/abhinavxd/artemis/internal/dbutils" + "github.com/abhinavxd/artemis/internal/dbutil" "github.com/jmoiron/sqlx" "github.com/zerodha/logf" ) @@ -31,7 +31,7 @@ type queries struct { func New(opts Opts) (*Manager, error) { var q queries - if err := dbutils.ScanSQLFile("queries.sql", &q, opts.DB, efs); err != nil { + if err := dbutil.ScanSQLFile("queries.sql", &q, opts.DB, efs); err != nil { return nil, err } diff --git a/internal/contact/models/models.go b/internal/contact/models/models.go index fe308d05..ef989a48 100644 --- a/internal/contact/models/models.go +++ b/internal/contact/models/models.go @@ -1,16 +1,20 @@ package models -import "time" +import ( + "time" + + "github.com/volatiletech/null/v9" +) type Contact struct { - 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"` - Email string `db:"email" json:"email"` - PhoneNumber *string `db:"phone_number" json:"phone_number"` - AvatarURL *string `db:"avatar_url" json:"avatar_url"` - InboxID int `db:"inbox_id" json:"inbox_id"` - Source string `db:"source" json:"source"` - SourceID string `db:"source_id" json:"source_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"` + Email string `db:"email" json:"email"` + PhoneNumber *string `db:"phone_number" json:"phone_number"` + AvatarURL null.String `db:"avatar_url" json:"avatar_url"` + InboxID int `db:"inbox_id" json:"inbox_id"` + Source string `db:"source" json:"source"` + SourceID string `db:"source_id" json:"source_id"` } diff --git a/internal/conversation/conversations.go b/internal/conversation/conversations.go index 19629ff1..0c2d9ad1 100644 --- a/internal/conversation/conversations.go +++ b/internal/conversation/conversations.go @@ -11,8 +11,9 @@ import ( "time" "github.com/abhinavxd/artemis/internal/conversation/models" - "github.com/abhinavxd/artemis/internal/dbutils" - "github.com/abhinavxd/artemis/internal/stringutils" + "github.com/abhinavxd/artemis/internal/dbutil" + "github.com/abhinavxd/artemis/internal/stringutil" + "github.com/abhinavxd/artemis/internal/ws" "github.com/jmoiron/sqlx" "github.com/lib/pq" "github.com/zerodha/logf" @@ -43,11 +44,23 @@ var ( PriortiyMedium, PriorityHigh, } + + preDefinedFilters = map[string]string{ + "status_open": " c.status = 'Open'", + "status_processing": " c.status = 'Processing'", + "status_spam": " c.status = 'Spam'", + "status_resolved": " c.status = 'Resolved'", + "status_all": " 1=1 ", + } + + assigneeTypeTeam = "team" + assigneeTypeUser = "user" ) type Manager struct { lo *logf.Logger db *sqlx.DB + hub *ws.Hub q queries ReferenceNumPattern string } @@ -66,6 +79,7 @@ type queries struct { GetUnassigned *sqlx.Stmt `query:"get-unassigned"` GetConversationParticipants *sqlx.Stmt `query:"get-conversation-participants"` GetConversations string `query:"get-conversations"` + GetConversationsUUIDs string `query:"get-conversations-uuids"` GetAssignedConversations *sqlx.Stmt `query:"get-assigned-conversations"` GetAssigneeStats *sqlx.Stmt `query:"get-assignee-stats"` InsertConverstionParticipant *sqlx.Stmt `query:"insert-conversation-participant"` @@ -77,15 +91,18 @@ type queries struct { UpdatePriority *sqlx.Stmt `query:"update-priority"` UpdateStatus *sqlx.Stmt `query:"update-status"` UpdateMeta *sqlx.Stmt `query:"update-meta"` + AddTag *sqlx.Stmt `query:"add-tag"` + DeleteTags *sqlx.Stmt `query:"delete-tags"` } -func New(opts Opts) (*Manager, error) { +func New(hub *ws.Hub, opts Opts) (*Manager, error) { var q queries - if err := dbutils.ScanSQLFile("queries.sql", &q, opts.DB, efs); err != nil { + if err := dbutil.ScanSQLFile("queries.sql", &q, opts.DB, efs); err != nil { return nil, err } c := &Manager{ q: q, + hub: hub, db: opts.DB, lo: opts.Lo, ReferenceNumPattern: opts.ReferenceNumPattern, @@ -93,16 +110,17 @@ func New(opts Opts) (*Manager, error) { return c, nil } -func (c *Manager) Create(contactID int, inboxID int, meta []byte) (int, error) { +func (c *Manager) Create(contactID int, inboxID int, meta []byte) (int, string, error) { var ( id int + uuid string refNum, _ = c.generateRefNum(c.ReferenceNumPattern) ) - if err := c.q.InsertConversation.QueryRow(refNum, contactID, StatusOpen, inboxID, meta).Scan(&id); err != nil { + if err := c.q.InsertConversation.QueryRow(refNum, contactID, StatusOpen, inboxID, meta).Scan(&id, &uuid); err != nil { c.lo.Error("inserting new conversation into the DB", "error", err) - return id, err + return id, uuid, err } - return id, nil + return id, uuid, nil } func (c *Manager) Get(uuid string) (models.Conversation, error) { @@ -144,24 +162,33 @@ func (c *Manager) AddParticipant(userID int, convUUID string) error { return nil } -func (c *Manager) UpdateMeta(convID int, convUUID string, meta map[string]string) error { +func (c *Manager) UpdateMeta(conversationID int, conversationUUID string, meta map[string]string) error { metaJSON, err := json.Marshal(meta) if err != nil { c.lo.Error("error marshalling meta", "error", err) return err } - if _, err := c.q.UpdateMeta.Exec(convID, convUUID, metaJSON); err != nil { + if _, err := c.q.UpdateMeta.Exec(conversationID, conversationUUID, metaJSON); err != nil { c.lo.Error("error updating conversation meta", "error", "error") return err } return nil } -func (c *Manager) UpdateFirstReplyAt(convID int, at time.Time) error { - if _, err := c.q.UpdateFirstReplyAt.Exec(convID, at); err != nil { +func (c *Manager) UpdateLastMessage(conversationID int, conversationUUID, lastMessage string, lastMessageAt time.Time) error { + return c.UpdateMeta(conversationID, conversationUUID, map[string]string{ + "last_message": lastMessage, + "last_message_at": lastMessageAt.Format(time.RFC3339), + }) +} + +func (c *Manager) UpdateFirstReplyAt(conversationUUID string, conversationID int, at time.Time) error { + if _, err := c.q.UpdateFirstReplyAt.Exec(conversationID, at); err != nil { c.lo.Error("error updating conversation first reply at", "error", err) return err } + // Send ws update. + c.hub.BroadcastConversationPropertyUpdate(conversationUUID, "first_reply_at", time.Now().Format(time.RFC3339)) return nil } @@ -217,24 +244,10 @@ func (c *Manager) GetConversations(userID int, typ, order, orderBy, predefinedFi qArgs []interface{} cond string // TODO: Remove these hardcoded values. - validOrderBy = map[string]bool{"created_at": true, "priority": true, "status": true, "last_message_at": true} - validOrder = []string{"ASC", "DESC"} - preDefinedFilters = map[string]string{ - "status_open": " c.status = 'Open'", - "status_processing": " c.status = 'Processing'", - "status_spam": " c.status = 'Spam'", - "status_resolved": " c.status = 'Resolved'", - } + validOrderBy = map[string]bool{"created_at": true, "priority": true, "status": true, "last_message_at": true} + validOrder = []string{"ASC", "DESC"} ) - if page <= 0 { - page = 1 - } - - if pageSize <= 0 { - pageSize = 20 - } - switch typ { case "assigned": cond = "AND c.assigned_user_id = $1" @@ -251,11 +264,7 @@ func (c *Manager) GetConversations(userID int, typ, order, orderBy, predefinedFi cond += " AND " + filterClause } - // Calculate offset based on page number and page size - offset := (page - 1) * pageSize - qArgs = append(qArgs, pageSize, offset) - - // Ensure orderBy is valid to prevent SQL injection + // Ensure orderBy is valid. orderByClause := "" if _, ok := validOrderBy[orderBy]; ok { orderByClause = fmt.Sprintf(" ORDER BY %s", orderBy) @@ -269,6 +278,16 @@ func (c *Manager) GetConversations(userID int, typ, order, orderBy, predefinedFi orderByClause += " DESC " } + // Calculate offset based on page number and page size. + if page <= 0 { + page = 1 + } + if pageSize <= 0 || pageSize > 20 { + pageSize = 20 + } + offset := (page - 1) * pageSize + qArgs = append(qArgs, pageSize, offset) + tx, err := c.db.BeginTxx(context.Background(), nil) defer tx.Rollback() if err != nil { @@ -276,7 +295,7 @@ func (c *Manager) GetConversations(userID int, typ, order, orderBy, predefinedFi return conversations, err } - // Include LIMIT, OFFSET, and ORDER BY in the SQL query + // Include LIMIT, OFFSET, and ORDER BY in the SQL query. sqlQuery := fmt.Sprintf("%s %s LIMIT $%d OFFSET $%d", fmt.Sprintf(c.q.GetConversations, cond), orderByClause, len(qArgs)-1, len(qArgs)) if err := tx.Select(&conversations, sqlQuery, qArgs...); err != nil { c.lo.Error("Error fetching conversations", "error", err) @@ -286,6 +305,54 @@ func (c *Manager) GetConversations(userID int, typ, order, orderBy, predefinedFi return conversations, nil } +func (c *Manager) GetConversationUUIDs(userID, page, pageSize int, typ, predefinedFilter string) ([]string, error) { + var ( + conversationUUIDs []string + qArgs []interface{} + cond string + ) + switch typ { + case "assigned": + cond = "AND c.assigned_user_id = $1" + qArgs = append(qArgs, userID) + case "unassigned": + cond = "AND c.assigned_user_id IS NULL AND c.assigned_team_id IN (SELECT team_id FROM team_members WHERE user_id = $1)" + qArgs = append(qArgs, userID) + case "all": + default: + return conversationUUIDs, errors.New("invalid type") + } + + if filterClause, ok := preDefinedFilters[predefinedFilter]; ok { + cond += " AND " + filterClause + } + + tx, err := c.db.BeginTxx(context.Background(), nil) + defer tx.Rollback() + if err != nil { + c.lo.Error("Error preparing get conversation ids query", "error", err) + return conversationUUIDs, err + } + + // Calculate offset based on page number and page size. + if page <= 0 { + page = 1 + } + if pageSize <= 0 || pageSize > 20 { + pageSize = 20 + } + offset := (page - 1) * pageSize + qArgs = append(qArgs, pageSize, offset) + + // Include LIMIT, OFFSET, and ORDER BY in the SQL query. + sqlQuery := fmt.Sprintf("%s LIMIT $%d OFFSET $%d", fmt.Sprintf(c.q.GetConversationsUUIDs, cond), len(qArgs)-1, len(qArgs)) + if err := tx.Select(&conversationUUIDs, sqlQuery, qArgs...); err != nil { + c.lo.Error("Error fetching conversations", "error", err) + return conversationUUIDs, err + } + return conversationUUIDs, nil +} + func (c *Manager) GetAssignedConversations(userID int) ([]models.Conversation, error) { var conversations []models.Conversation if err := c.q.GetAssignedConversations.Select(&conversations, userID); err != nil { @@ -295,18 +362,28 @@ func (c *Manager) GetAssignedConversations(userID int) ([]models.Conversation, e return conversations, nil } +func (c *Manager) UpdateTeamAssignee(uuid string, assigneeUUID []byte) error { + return c.UpdateAssignee(uuid, assigneeUUID, assigneeTypeTeam) +} + +func (c *Manager) UpdateUserAssignee(uuid string, assigneeUUID []byte) error { + return c.UpdateAssignee(uuid, assigneeUUID, assigneeTypeUser) +} + func (c *Manager) UpdateAssignee(uuid string, assigneeUUID []byte, assigneeType string) error { switch assigneeType { - case "agent": + case assigneeTypeUser: if _, err := c.q.UpdateAssignedUser.Exec(uuid, assigneeUUID); err != nil { c.lo.Error("updating conversation assignee", "error", err) return fmt.Errorf("error updating assignee") } - case "team": + c.hub.BroadcastConversationPropertyUpdate(uuid, "assigned_user_uuid", string(assigneeUUID)) + case assigneeTypeTeam: if _, err := c.q.UpdateAssignedTeam.Exec(uuid, assigneeUUID); err != nil { c.lo.Error("updating conversation assignee", "error", err) return fmt.Errorf("error updating assignee") } + c.hub.BroadcastConversationPropertyUpdate(uuid, "assigned_team_uuid", string(assigneeUUID)) default: return errors.New("invalid assignee type") } @@ -314,13 +391,15 @@ func (c *Manager) UpdateAssignee(uuid string, assigneeUUID []byte, assigneeType } func (c *Manager) UpdatePriority(uuid string, priority []byte) error { - if !slices.Contains(priorities, string(priority)) { - return fmt.Errorf("invalid `priority` value %s", priority) + var priorityStr = string(priority) + if !slices.Contains(priorities, string(priorityStr)) { + return fmt.Errorf("invalid `priority` value %s", priorityStr) } if _, err := c.q.UpdatePriority.Exec(uuid, priority); err != nil { c.lo.Error("updating conversation priority", "error", err) return fmt.Errorf("error updating priority") } + c.hub.BroadcastConversationPropertyUpdate(uuid, "priority", priorityStr) return nil } @@ -332,6 +411,7 @@ func (c *Manager) UpdateStatus(uuid string, status []byte) error { c.lo.Error("updating conversation status", "error", err) return fmt.Errorf("error updating status") } + c.hub.BroadcastConversationPropertyUpdate(uuid, "status", string(status)) return nil } @@ -347,11 +427,26 @@ func (c *Manager) GetAssigneeStats(userID int) (models.ConversationCounts, error return counts, nil } +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("error deleting conversation tags", "error", err) + } + + // Add new tags. + for _, tagID := range tagIDs { + if _, err := t.q.AddTag.Exec(convUUID, tagID); err != nil { + t.lo.Error("error adding tags to conversation", "error", err) + } + } + return nil +} + func (c *Manager) generateRefNum(pattern string) (string, error) { if len(pattern) <= 5 { pattern = "01234567890" } - randomNumbers, err := stringutils.RandomNumericString(len(pattern)) + randomNumbers, err := stringutil.RandomNumericString(len(pattern)) if err != nil { return "", err } diff --git a/internal/conversation/models/models.go b/internal/conversation/models/models.go index 971f5583..1a2936b6 100644 --- a/internal/conversation/models/models.go +++ b/internal/conversation/models/models.go @@ -18,7 +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"` + FirstReplyAt null.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"` @@ -32,7 +32,7 @@ type Conversation struct { ContactAvatarURL *string `db:"contact_avatar_url" json:"contact_avatar_url"` AssignedTeamUUID *string `db:"assigned_team_uuid" json:"assigned_team_uuid"` AssignedAgentUUID *string `db:"assigned_user_uuid" json:"assigned_user_uuid"` - LastMessageAt *time.Time `db:"last_message_at" json:"last_message_at"` + LastMessageAt null.Time `db:"last_message_at" json:"last_message_at"` LastMessage string `db:"last_message" json:"last_message"` FirstMessage string `json:"-"` } diff --git a/internal/conversation/queries.sql b/internal/conversation/queries.sql index b22d5566..21e6fe27 100644 --- a/internal/conversation/queries.sql +++ b/internal/conversation/queries.sql @@ -2,7 +2,7 @@ INSERT INTO conversations (reference_number, contact_id, status, inbox_id, meta) VALUES($1, $2, $3, $4, $5) -returning id; +returning id, uuid; -- name: get-conversations @@ -28,6 +28,12 @@ FROM conversations c JOIN inboxes inb on c.inbox_id = inb.id WHERE 1=1 %s +-- name: get-conversations-uuids +SELECT + c.uuid +FROM conversations c +WHERE 1=1 %s + -- name: get-assigned-conversations SELECT uuid from conversations where assigned_user_id = $1; @@ -131,7 +137,27 @@ VALUES($1, (select id from conversations where uuid = $2)); 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; +SELECT + c.updated_at, + c.uuid, + c.assignee_last_seen_at, + c.assigned_team_id, + inb.channel as inbox_channel, + inb.name as inbox_name, + ct.first_name, + ct.last_name, + ct.avatar_url, + COALESCE(c.meta->>'subject', '') as subject, + COALESCE(c.meta->>'last_message', '') as last_message, + COALESCE((c.meta->>'last_message_at')::timestamp, '1970-01-01 00:00:00'::timestamp) as last_message_at, + ( + SELECT COUNT(*) + FROM messages m + WHERE m.conversation_id = c.id AND m.created_at > c.assignee_last_seen_at + ) AS unread_message_count +FROM conversations c + JOIN contacts ct ON c.contact_id = ct.id + JOIN inboxes inb on c.inbox_id = inb.id where assigned_user_id is NULL and assigned_team_id is not null; -- name: get-assignee-stats SELECT @@ -148,4 +174,23 @@ WHERE -- name: update-first-reply-at UPDATE conversations SET first_reply_at = $2 -WHERE first_reply_at IS NULL AND id = $1; \ No newline at end of file +WHERE first_reply_at IS NULL AND id = $1; + +-- name: add-tag +INSERT INTO conversation_tags (conversation_id, tag_id) +VALUES( + ( + SELECT id + from conversations + where uuid = $1 + ), + $2 +) ON CONFLICT DO NOTHING + +-- name: delete-tags +DELETE FROM conversation_tags +WHERE conversation_id = ( + SELECT id + from conversations + where uuid = $1 +) AND tag_id NOT IN (SELECT unnest($2::int[])); diff --git a/internal/conversation/tag/queries.sql b/internal/conversation/tag/queries.sql deleted file mode 100644 index db11d604..00000000 --- a/internal/conversation/tag/queries.sql +++ /dev/null @@ -1,18 +0,0 @@ --- name: add-tag -INSERT INTO conversation_tags (conversation_id, tag_id) -VALUES( - ( - SELECT id - from conversations - where uuid = $1 - ), - $2 -) ON CONFLICT DO NOTHING; - --- name: delete-tags -DELETE FROM conversation_tags -WHERE conversation_id = ( - SELECT id - from conversations - where uuid = $1 -) AND tag_id NOT IN (SELECT unnest($2::int[])); diff --git a/internal/conversation/tag/tag.go b/internal/conversation/tag/tag.go deleted file mode 100644 index 1b47716c..00000000 --- a/internal/conversation/tag/tag.go +++ /dev/null @@ -1,58 +0,0 @@ -package tag - -import ( - "embed" - - "github.com/abhinavxd/artemis/internal/dbutils" - "github.com/jmoiron/sqlx" - "github.com/lib/pq" - "github.com/zerodha/logf" -) - -var ( - //go:embed queries.sql - efs embed.FS -) - -type Manager struct { - q queries - lo *logf.Logger -} - -type Opts struct { - DB *sqlx.DB - Lo *logf.Logger -} - -type queries struct { - AddTag *sqlx.Stmt `query:"add-tag"` - DeleteTags *sqlx.Stmt `query:"delete-tags"` -} - -func New(opts Opts) (*Manager, error) { - var q queries - - if err := dbutils.ScanSQLFile("queries.sql", &q, opts.DB, efs); err != nil { - return nil, err - } - - return &Manager{ - q: q, - lo: opts.Lo, - }, nil -} - -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) - } - - // 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 nil -} diff --git a/internal/dbutils/query_loader.go b/internal/dbutil/query_loader.go similarity index 97% rename from internal/dbutils/query_loader.go rename to internal/dbutil/query_loader.go index 3423bfef..0345a086 100644 --- a/internal/dbutils/query_loader.go +++ b/internal/dbutil/query_loader.go @@ -1,4 +1,4 @@ -package dbutils +package dbutil import ( "io/fs" diff --git a/internal/inbox/channel/email/imap.go b/internal/inbox/channel/email/imap.go index 707ef6c3..4de40f29 100644 --- a/internal/inbox/channel/email/imap.go +++ b/internal/inbox/channel/email/imap.go @@ -122,7 +122,6 @@ func (e *Email) processEnvelope(c *imapclient.Client, env *imap.Envelope, seqNum 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 } diff --git a/internal/inbox/channel/email/smtp.go b/internal/inbox/channel/email/smtp.go index 0d514dc9..6ed8be60 100644 --- a/internal/inbox/channel/email/smtp.go +++ b/internal/inbox/channel/email/smtp.go @@ -13,10 +13,11 @@ import ( ) const ( - headerReturnPath = "Return-Path" - headerMessageID = "Message-ID" - headerReferences = "References" - headerInReplyTo = "In-Reply-To" + headerReturnPath = "Return-Path" + headerMessageID = "Message-ID" + headerReferences = "References" + headerInReplyTo = "In-Reply-To" + dispositionInline = "inline" ) // New returns an SMTP e-mail channels from the given SMTP server configcfg. diff --git a/internal/inbox/inbox.go b/internal/inbox/inbox.go index 9705c99d..1ccae633 100644 --- a/internal/inbox/inbox.go +++ b/internal/inbox/inbox.go @@ -6,7 +6,7 @@ import ( "encoding/json" "errors" - "github.com/abhinavxd/artemis/internal/dbutils" + "github.com/abhinavxd/artemis/internal/dbutil" "github.com/abhinavxd/artemis/internal/message/models" "github.com/jmoiron/sqlx" "github.com/zerodha/logf" @@ -20,7 +20,7 @@ var ( ErrInboxNotFound = errors.New("inbox not found") ) -// Closer provides function for closing a channel. +// Closer provides function for closing an inbox. type Closer interface { Close() error } @@ -83,7 +83,7 @@ func New(lo *logf.Logger, db *sqlx.DB) (*Manager, error) { var q queries // Scan the sql file into the queries struct. - if err := dbutils.ScanSQLFile("queries.sql", &q, db, efs); err != nil { + if err := dbutil.ScanSQLFile("queries.sql", &q, db, efs); err != nil { return nil, err } diff --git a/internal/message/messages.go b/internal/message/messages.go index 47688986..8c50e9a2 100644 --- a/internal/message/messages.go +++ b/internal/message/messages.go @@ -16,13 +16,13 @@ import ( "github.com/abhinavxd/artemis/internal/automation" "github.com/abhinavxd/artemis/internal/contact" "github.com/abhinavxd/artemis/internal/conversation" - "github.com/abhinavxd/artemis/internal/team" - "github.com/abhinavxd/artemis/internal/user" - cmodels "github.com/abhinavxd/artemis/internal/conversation/models" - "github.com/abhinavxd/artemis/internal/dbutils" + "github.com/abhinavxd/artemis/internal/dbutil" "github.com/abhinavxd/artemis/internal/inbox" "github.com/abhinavxd/artemis/internal/message/models" + "github.com/abhinavxd/artemis/internal/team" + "github.com/abhinavxd/artemis/internal/template" + "github.com/abhinavxd/artemis/internal/user" "github.com/abhinavxd/artemis/internal/ws" "github.com/jmoiron/sqlx" "github.com/k3a/html2text" @@ -66,19 +66,20 @@ const ( ) type Manager struct { - q queries - lo *logf.Logger - contactMgr *contact.Manager - attachmentMgr *attachment.Manager - conversationMgr *conversation.Manager - inboxMgr *inbox.Manager - userMgr *user.Manager - teamMgr *team.Manager - automationEngine *automation.Engine - wsHub *ws.Hub - incomingMsgQ chan models.IncomingMessage - outgoingMsgQ chan models.Message - outgoingProcessingMsgs sync.Map + q queries + lo *logf.Logger + contactMgr *contact.Manager + attachmentMgr *attachment.Manager + conversationMgr *conversation.Manager + inboxMgr *inbox.Manager + userMgr *user.Manager + teamMgr *team.Manager + automationEngine *automation.Engine + wsHub *ws.Hub + templateManager *template.Manager + incomingMsgQ chan models.IncomingMessage + outgoingMessageQueue chan models.Message + outgoingProcessingMessages sync.Map } type Opts struct { @@ -109,26 +110,29 @@ func New( inboxMgr *inbox.Manager, conversationMgr *conversation.Manager, automationEngine *automation.Engine, - opts Opts) (*Manager, error) { + templateManager *template.Manager, + opts Opts, +) (*Manager, error) { var q queries - if err := dbutils.ScanSQLFile("queries.sql", &q, opts.DB, efs); err != nil { + if err := dbutil.ScanSQLFile("queries.sql", &q, opts.DB, efs); err != nil { return nil, err } return &Manager{ - q: q, - lo: opts.Lo, - wsHub: wsHub, - userMgr: userMgr, - teamMgr: teamMgr, - contactMgr: contactMgr, - attachmentMgr: attachmentMgr, - conversationMgr: conversationMgr, - inboxMgr: inboxMgr, - automationEngine: automationEngine, - incomingMsgQ: make(chan models.IncomingMessage, opts.IncomingMsgQueueSize), - outgoingMsgQ: make(chan models.Message, opts.OutgoingMsgQueueSize), - outgoingProcessingMsgs: sync.Map{}, + q: q, + lo: opts.Lo, + wsHub: wsHub, + userMgr: userMgr, + teamMgr: teamMgr, + contactMgr: contactMgr, + attachmentMgr: attachmentMgr, + conversationMgr: conversationMgr, + inboxMgr: inboxMgr, + automationEngine: automationEngine, + templateManager: templateManager, + incomingMsgQ: make(chan models.IncomingMessage, opts.IncomingMsgQueueSize), + outgoingMessageQueue: make(chan models.Message, opts.OutgoingMsgQueueSize), + outgoingProcessingMessages: sync.Map{}, }, nil } @@ -208,70 +212,74 @@ func (m *Manager) StartDispatcher(ctx context.Context, concurrency int, readInte // Prepare and push the message to the outgoing queue. for _, msg := range pendingMsgs { - m.outgoingProcessingMsgs.Store(msg.ID, msg.ID) - m.outgoingMsgQ <- msg + var err error + msg.Content, _, err = m.templateManager.RenderDefault(map[string]string{ + "Content": msg.Content, + }) + if err != nil { + m.lo.Error("error rendering message template", "error", err) + m.UpdateMessageStatus(msg.UUID, StatusFailed) + continue + } + m.outgoingProcessingMessages.Store(msg.ID, msg.ID) + m.outgoingMessageQueue <- msg } } } } func (m *Manager) DispatchWorker() { - for msg := range m.outgoingMsgQ { - inbox, err := m.inboxMgr.GetInbox(msg.InboxID) + for message := range m.outgoingMessageQueue { + inbox, err := m.inboxMgr.GetInbox(message.InboxID) if err != nil { - m.lo.Error("error fetching inbox", "error", err, "inbox_id", msg.InboxID) - m.outgoingProcessingMsgs.Delete(msg.ID) + m.lo.Error("error fetching inbox", "error", err, "inbox_id", message.InboxID) + m.outgoingProcessingMessages.Delete(message.ID) continue } - msg.From = inbox.FromAddress() + message.From = inbox.FromAddress() - if err := m.attachAttachments(&msg); err != nil { - m.lo.Error("error attaching attachments to msg", "error", err) - m.outgoingProcessingMsgs.Delete(msg.ID) + if err := m.attachAttachments(&message); err != nil { + m.lo.Error("error attaching attachments to message", "error", err) + m.outgoingProcessingMessages.Delete(message.ID) continue } - msg.To, _ = m.GetToAddress(msg.ConversationID, inbox.Channel()) + message.To, _ = m.GetToAddress(message.ConversationID, inbox.Channel()) if inbox.Channel() == "email" { - msg.InReplyTo, _ = m.GetInReplyTo(msg.ConversationID) - m.lo.Debug("set in reply to for outgoing email message", "in_reply_to", msg.InReplyTo) + message.InReplyTo, _ = m.GetInReplyTo(message.ConversationID) } - err = inbox.Send(msg) + err = inbox.Send(message) var newStatus = StatusSent if err != nil { newStatus = StatusFailed - m.lo.Error("error sending message", "error", err, "inbox_id", msg.InboxID) + m.lo.Error("error sending message", "error", err, "inbox_id", message.InboxID) } - if _, err := m.q.UpdateMessageStatus.Exec(newStatus, msg.UUID); err != nil { - m.lo.Error("error updating message status in DB", "error", err, "inbox_id", msg.InboxID) + if _, err := m.q.UpdateMessageStatus.Exec(newStatus, message.UUID); err != nil { + m.lo.Error("error updating message status in DB", "error", err, "inbox_id", message.InboxID) } switch newStatus { case StatusSent: - m.lo.Debug("updating first reply at", "conv_id", msg.ConversationID, "at", msg.CreatedAt) - m.conversationMgr.UpdateFirstReplyAt(msg.ConversationID, msg.CreatedAt) + m.conversationMgr.UpdateFirstReplyAt(message.ConversationUUID, message.ConversationID, message.CreatedAt) } - // Broadcast the new message status. - m.wsHub.BroadcastMsgStatus(msg.ConversationUUID, map[string]interface{}{ - "uuid": msg.UUID, - "conversation_uuid": msg.ConversationUUID, - "status": newStatus, - }) + // Broadcast message status update to the subscribers. + m.wsHub.BroadcastMessagePropUpdate(message.ConversationUUID, message.UUID, "status" /*message field*/, newStatus) - m.outgoingProcessingMsgs.Delete(msg.ID) + // Remove message from the processing list. + m.outgoingProcessingMessages.Delete(message.ID) } } 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) + m.lo.Error("error fetching to address for msg", "error", err, "conversation_id", convID) return addr, err } return addr, nil @@ -281,10 +289,10 @@ 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 { - m.lo.Error("in reply to not found", "error", err, "conv_id", convID) + m.lo.Error("in reply to not found", "error", err, "conversation_id", convID) return out, nil } - m.lo.Error("error fetching in reply to", "error", err, "conv_id", convID) + m.lo.Error("error fetching in reply to", "error", err, "conversation_id", convID) return out, err } return out, nil @@ -311,37 +319,36 @@ func (m *Manager) InsertWorker(ctx context.Context) { } } -func (m *Manager) RecordAssigneeUserChange(updatedValue, convUUID, actorUUID string) error { - if updatedValue == actorUUID { - return m.RecordActivity(ActivitySelfAssign, updatedValue, convUUID, actorUUID) +func (m *Manager) RecordAssigneeUserChange(conversationUUID, assigneeUUID, actorUUID string) error { + // Self assign. + if assigneeUUID == actorUUID { + return m.RecordActivity(ActivitySelfAssign, assigneeUUID, conversationUUID, actorUUID) } - assignee, err := m.userMgr.GetUser(0, updatedValue) + assignee, err := m.userMgr.GetUser(0, assigneeUUID) if err != nil { - m.lo.Error("Error fetching user to record assignee change", "error", err) + m.lo.Error("Error fetching user to record assignee change", "conversation_uuid", conversationUUID, "actor_uuid", actorUUID, "error", err) return err } - updatedValue = assignee.FullName() - return m.RecordActivity(ActivityAssignedUserChange, updatedValue, convUUID, actorUUID) + return m.RecordActivity(ActivityAssignedUserChange, assignee.FullName() /*new_value*/, conversationUUID, actorUUID) } -func (m *Manager) RecordAssigneeTeamChange(updatedValue, convUUID, actorUUID string) error { - team, err := m.teamMgr.GetTeam(updatedValue) +func (m *Manager) RecordAssigneeTeamChange(conversationUUID, value, actorUUID string) error { + team, err := m.teamMgr.GetTeam(value) if err != nil { return err } - updatedValue = team.Name - return m.RecordActivity(ActivityAssignedTeamChange, updatedValue, convUUID, actorUUID) + return m.RecordActivity(ActivityAssignedTeamChange, team.Name /*new_value*/, conversationUUID, actorUUID) } -func (m *Manager) RecordPriorityChange(updatedValue, convUUID, actorUUID string) error { - return m.RecordActivity(ActivityPriorityChange, updatedValue, convUUID, actorUUID) +func (m *Manager) RecordPriorityChange(updatedValue, conversationUUID, actorUUID string) error { + return m.RecordActivity(ActivityPriorityChange, updatedValue, conversationUUID, actorUUID) } -func (m *Manager) RecordStatusChange(updatedValue, convUUID, actorUUID string) error { - return m.RecordActivity(ActivityStatusChange, updatedValue, convUUID, actorUUID) +func (m *Manager) RecordStatusChange(updatedValue, conversationUUID, actorUUID string) error { + return m.RecordActivity(ActivityStatusChange, updatedValue, conversationUUID, actorUUID) } -func (m *Manager) RecordActivity(activityType, updatedValue, conversationUUID, actorUUID string) error { +func (m *Manager) RecordActivity(activityType, newValue, conversationUUID, actorUUID string) error { var ( actor, err = m.userMgr.GetUser(0, actorUUID) ) @@ -350,7 +357,7 @@ func (m *Manager) RecordActivity(activityType, updatedValue, conversationUUID, a return err } - var content = m.getActivityContent(activityType, updatedValue, actor.FullName()) + var content = m.getActivityContent(activityType, newValue, actor.FullName()) if content == "" { m.lo.Error("Error invalid activity for recording activity", "activity", activityType) return errors.New("invalid activity type for recording activity") @@ -369,8 +376,8 @@ func (m *Manager) RecordActivity(activityType, updatedValue, conversationUUID, a } m.RecordMessage(&msg) - m.BroadcastNewMsg(msg, "") - + m.BroadcastNewConversationMessage(msg, content) + m.conversationMgr.UpdateLastMessage(0, conversationUUID, content, msg.CreatedAt) return nil } @@ -425,28 +432,33 @@ func (m *Manager) processIncomingMessage(in models.IncomingMessage) error { return nil } - if err = m.findOrCreateConversation(&in.Message, in.InboxID, senderID, convMetaJSON); err != nil { - m.lo.Error("error creating conversation", "error", err) + isNewConversation, err := m.findOrCreateConversation(&in.Message, in.InboxID, senderID, convMetaJSON) + if err != nil { return err } if err = m.RecordMessage(&in.Message); err != nil { - m.lo.Error("error inserting conversation message", "error", err) return fmt.Errorf("inserting conversation message: %w", err) } if err := m.uploadAttachments(&in.Message); err != nil { - m.lo.Error("error uploading message attachments", "msg_uuid", in.Message.UUID, "error", err) return fmt.Errorf("uploading message attachments: %w", err) } // Send WS update. if in.Message.ConversationUUID > "" { - m.BroadcastNewMsg(in.Message, "") + var content = "" + if isNewConversation { + content = m.TrimMsg(in.Message.Subject) + } else { + content = m.TrimMsg(in.Message.Content) + } + m.BroadcastNewConversationMessage(in.Message, content) + m.conversationMgr.UpdateLastMessage(in.Message.ConversationID, in.Message.ConversationUUID, content, in.Message.CreatedAt) } - // Evaluate automation rules for this conversation. - if in.Message.NewConversation { + // Evaluate automation rules for this new conversation. + if isNewConversation { m.automationEngine.EvaluateRules(cmodels.Conversation{ UUID: in.Message.ConversationUUID, FirstMessage: in.Message.Content, @@ -518,52 +530,53 @@ func (m *Manager) uploadAttachments(in *models.Message) error { return nil } -func (m *Manager) findOrCreateConversation(in *models.Message, inboxID int, contactID int, meta []byte) error { +func (m *Manager) findOrCreateConversation(in *models.Message, inboxID int, contactID int, meta []byte) (bool, error) { var ( + new bool + err error conversationID int conversationUUID string - newConv bool - err error ) // Search for existing conversation. - if conversationID == 0 && in.InReplyTo > "" { - conversationID, err = m.findConversationID([]string{in.InReplyTo}) - if err != nil && err != ErrConversationNotFound { - return err - } + sourceIDs := in.References + if in.InReplyTo > "" { + sourceIDs = append(sourceIDs, in.InReplyTo) } - if conversationID == 0 && len(in.References) > 0 { - conversationID, err = m.findConversationID(in.References) - if err != nil && err != ErrConversationNotFound { - return err - } + conversationID, err = m.findConversationID(sourceIDs) + if err != nil && err != ErrConversationNotFound { + return new, err } // Conversation not found, create one. if conversationID == 0 { - newConv = true - conversationID, err = m.conversationMgr.Create(contactID, inboxID, meta) + new = true + conversationID, conversationUUID, err = m.conversationMgr.Create(contactID, inboxID, meta) if err != nil || conversationID == 0 { - return fmt.Errorf("inserting conversation: %w", err) + return new, err + } + in.ConversationID = conversationID + in.ConversationUUID = conversationUUID + return new, nil + } + + // Set UUID if not available. + if conversationUUID == "" { + conversationUUID, err = m.conversationMgr.GetUUID(conversationID) + if err != nil { + return new, err } } - - // Fetch & return the UUID of the conversation for UI updates. - conversationUUID, err = m.conversationMgr.GetUUID(conversationID) - if err != nil { - m.lo.Error("Error fetching conversation UUID from id", err) - } - in.ConversationID = conversationID in.ConversationUUID = conversationUUID - in.NewConversation = newConv - - return nil + return new, nil } // findConversationID finds the conversation ID from the message source ID. func (m *Manager) findConversationID(sourceIDs []string) (int, error) { + if len(sourceIDs) == 0 { + return 0, ErrConversationNotFound + } var conversationID int if err := m.q.MessageExists.QueryRow(pq.Array(sourceIDs)).Scan(&conversationID); err != nil { if err == sql.ErrNoRows { @@ -583,6 +596,11 @@ func (m *Manager) attachAttachments(msg *models.Message) error { return err } + // TODO: set attachment headers and replace the inline image src url w + // src="cid:ii_lxxsfhtp0" + // a.Header.Set("Content-Disposition", "inline") + // a.Header.Set("Content-ID", "<"+f.CID+">") + // Fetch the blobs and attach the attachments to the message. for i, att := range attachments { attachments[i].Content, err = m.attachmentMgr.Store.GetBlob(att.UUID) @@ -599,7 +617,7 @@ func (m *Manager) attachAttachments(msg *models.Message) error { // getOutgoingProcessingMsgIDs returns outgoing msg ids currently being processed. func (m *Manager) getOutgoingProcessingMsgIDs() []int { var out = make([]int, 0) - m.outgoingProcessingMsgs.Range(func(key, _ any) bool { + m.outgoingProcessingMessages.Range(func(key, _ any) bool { if k, ok := key.(int); ok { out = append(out, k) } @@ -608,21 +626,6 @@ func (m *Manager) getOutgoingProcessingMsgIDs() []int { return out } -func (m *Manager) BroadcastNewMsg(msg models.Message, trimmedContent string) { - if trimmedContent == "" { - var content = "" - if msg.NewConversation { - content = m.TrimMsg(msg.Subject) - } else { - content = m.TrimMsg(msg.Content) - } - trimmedContent = content - } - m.wsHub.BroadcastNewMsg(msg.ConversationUUID, map[string]interface{}{ - "conversation_uuid": msg.ConversationUUID, - "uuid": msg.UUID, - "last_message": trimmedContent, - "last_message_at": time.Now().Format(time.DateTime), - "private": msg.Private, - }) +func (m *Manager) BroadcastNewConversationMessage(message models.Message, trimmedContent string) { + m.wsHub.BroadcastNewConversationMessage(message.ConversationUUID, trimmedContent, message.UUID, time.Now().Format(time.RFC3339), message.Private) } diff --git a/internal/message/models/models.go b/internal/message/models/models.go index 50491fc9..05beb04d 100644 --- a/internal/message/models/models.go +++ b/internal/message/models/models.go @@ -41,7 +41,6 @@ type Message struct { References []string `json:"-"` InReplyTo string `json:"-"` Headers textproto.MIMEHeader `json:"-"` - NewConversation bool `json:"-"` } // IncomingMessage links a message with the contact information and inbox id. diff --git a/internal/notification/notification.go b/internal/notification/notification.go new file mode 100644 index 00000000..0c3a94d9 --- /dev/null +++ b/internal/notification/notification.go @@ -0,0 +1,22 @@ +package notifier + +// Notifier defines the interface for sending notifications. +type Notifier interface { + SendMessage(userUUIDs []string, subject, content string) error + SendAssignedConversationNotification(userUUIDs []string, convUUID string) error +} + +// TemplateRenderer defines the interface for rendering templates. +type TemplateRenderer interface { + RenderDefault(data interface{}) (subject, content string, err error) +} + +// UserEmailFetcher defines the interfaces for fetchign user email address. +type UserEmailFetcher interface { + GetEmail(id int, uuid string) (string, error) +} + +// UserStore defines the interface for the user store. +type UserStore interface { + UserEmailFetcher +} diff --git a/internal/notification/providers/email/email.go b/internal/notification/providers/email/email.go new file mode 100644 index 00000000..7bafa4f2 --- /dev/null +++ b/internal/notification/providers/email/email.go @@ -0,0 +1,138 @@ +package email + +import ( + "fmt" + "math/rand" + "net/textproto" + + "github.com/abhinavxd/artemis/internal/inbox/channel/email" + "github.com/abhinavxd/artemis/internal/message/models" + notifier "github.com/abhinavxd/artemis/internal/notification" + "github.com/knadh/smtppool" + "github.com/zerodha/logf" +) + +// Notifier handles email notifications. +type Notifier struct { + lo *logf.Logger + from string + smtpPools []*smtppool.Pool + userStore notifier.UserStore + TemplateRenderer notifier.TemplateRenderer +} + +type Opts struct { + Lo *logf.Logger + FromEmail string +} + +// New creates a new instance of email Notifier. +func New(smtpConfig []email.SMTPConfig, userStore notifier.UserStore, TemplateRenderer notifier.TemplateRenderer, opts Opts) (*Notifier, error) { + pools, err := email.NewSmtpPool(smtpConfig) + if err != nil { + return nil, err + } + return &Notifier{ + lo: opts.Lo, + smtpPools: pools, + from: opts.FromEmail, + userStore: userStore, + TemplateRenderer: TemplateRenderer, + }, nil +} + +// SendMessage sends an email using the default template to multiple users. +func (e *Notifier) SendMessage(userUUIDs []string, subject, content string) error { + var recipientEmails []string + for i := 0; i < len(userUUIDs); i++ { + userEmail, err := e.userStore.GetEmail(0, userUUIDs[i]) + if err != nil { + e.lo.Error("error fetching user email for user uuid", "error", err) + return err + } + recipientEmails = append(recipientEmails, userEmail) + } + + // Render with default template. + templateBody, templateSubject, err := e.TemplateRenderer.RenderDefault(map[string]string{ + "Content": content, + }) + if err != nil { + return err + } + + if subject == "" { + subject = templateSubject + } + + m := models.Message{ + Subject: subject, + Content: templateBody, + From: e.from, + To: recipientEmails, + } + + err = e.Send(m) + if err != nil { + e.lo.Error("error sending email notification", "error", err) + return err + } + return nil +} + +func (e *Notifier) SendAssignedConversationNotification(userUUIDs []string, convUUID string) error { + subject := "New conversation assigned to you" + link := fmt.Sprintf("http://localhost:5173/conversations/%s", convUUID) + content := fmt.Sprintf("A new conversation has been assigned to you.
Please review the details and take necessary action by following this link: %s", link) + return e.SendMessage(userUUIDs, subject, content) +} + +// Send sends an email message using one of the SMTP pools. +func (e *Notifier) Send(m models.Message) error { + var ( + ln = len(e.smtpPools) + srv *smtppool.Pool + ) + if ln > 1 { + srv = e.smtpPools[rand.Intn(ln)] + } else { + srv = e.smtpPools[0] + } + + var files []smtppool.Attachment + if m.Attachments != nil { + files = make([]smtppool.Attachment, 0, len(m.Attachments)) + for _, f := range m.Attachments { + a := smtppool.Attachment{ + Filename: f.Filename, + Header: f.Header, + Content: make([]byte, len(f.Content)), + } + copy(a.Content, f.Content) + files = append(files, a) + } + } + + em := smtppool.Email{ + From: m.From, + To: m.To, + Subject: m.Subject, + Attachments: files, + Headers: textproto.MIMEHeader{}, + } + + for k, v := range m.Headers { + em.Headers.Set(k, v[0]) + } + + switch m.ContentType { + case "plain": + em.Text = []byte(m.Content) + default: + em.HTML = []byte(m.Content) + if len(m.AltContent) > 0 { + em.Text = []byte(m.AltContent) + } + } + return srv.Send(em) +} diff --git a/internal/rbac/models/models.go b/internal/rbac/models/models.go deleted file mode 100644 index 2640e7f9..00000000 --- a/internal/rbac/models/models.go +++ /dev/null @@ -1 +0,0 @@ -package models diff --git a/internal/stringutils/strings.go b/internal/stringutil/stringutil.go similarity index 88% rename from internal/stringutils/strings.go rename to internal/stringutil/stringutil.go index d303a016..3a9af1fe 100644 --- a/internal/stringutils/strings.go +++ b/internal/stringutil/stringutil.go @@ -1,4 +1,4 @@ -package stringutils +package stringutil import ( "crypto/rand" @@ -23,7 +23,7 @@ func RandomAlNumString(n int) (string, error) { return string(bytes), nil } -// RandomNumericString generates a random digit string of length n. +// RandomNumericString generates a random digit numeric string of length n. func RandomNumericString(n int) (string, error) { const ( dictionary = "0123456789" diff --git a/internal/tag/tag.go b/internal/tag/tag.go index e38693d4..8af09cef 100644 --- a/internal/tag/tag.go +++ b/internal/tag/tag.go @@ -5,7 +5,7 @@ import ( "fmt" "time" - "github.com/abhinavxd/artemis/internal/dbutils" + "github.com/abhinavxd/artemis/internal/dbutil" "github.com/jmoiron/sqlx" "github.com/zerodha/logf" ) @@ -16,7 +16,7 @@ var ( ) type Tag struct { - ID int `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 := dbutils.ScanSQLFile("queries.sql", &q, opts.DB, efs); err != nil { + if err := dbutil.ScanSQLFile("queries.sql", &q, opts.DB, efs); err != nil { return nil, err } diff --git a/internal/team/team.go b/internal/team/team.go index 72c32c42..07f17dc7 100644 --- a/internal/team/team.go +++ b/internal/team/team.go @@ -6,7 +6,7 @@ import ( "errors" "fmt" - "github.com/abhinavxd/artemis/internal/dbutils" + "github.com/abhinavxd/artemis/internal/dbutil" umodels "github.com/abhinavxd/artemis/internal/user/models" "github.com/jmoiron/sqlx" "github.com/zerodha/logf" @@ -42,7 +42,7 @@ type queries struct { func New(opts Opts) (*Manager, error) { var q queries - if err := dbutils.ScanSQLFile("queries.sql", &q, opts.DB, efs); err != nil { + if err := dbutil.ScanSQLFile("queries.sql", &q, opts.DB, efs); err != nil { return nil, err } diff --git a/internal/template/models/models.go b/internal/template/models/models.go new file mode 100644 index 00000000..5939006d --- /dev/null +++ b/internal/template/models/models.go @@ -0,0 +1,6 @@ +package models + +type Template struct { + Body string `db:"body"` + Subject string `db:"subject"` +} diff --git a/internal/template/queries.sql b/internal/template/queries.sql new file mode 100644 index 00000000..f37565d7 --- /dev/null +++ b/internal/template/queries.sql @@ -0,0 +1,10 @@ +-- name: insert-template +INSERT INTO templates +("name", subject, body, is_default) +VALUES($1, $2, $3, $4); + +-- name: get-template +select id, name, subject, body from templates where name = $1; + +-- name: get-default-template +select id, name, subject, body from templates where is_default is true; \ No newline at end of file diff --git a/internal/template/renderer.go b/internal/template/renderer.go new file mode 100644 index 00000000..d8c09de6 --- /dev/null +++ b/internal/template/renderer.go @@ -0,0 +1,26 @@ +package template + +import ( + "bytes" + "text/template" +) + +// RenderDefault renders the system default template with the data. +func (m *Manager) RenderDefault(data interface{}) (string, string, error) { + templ, err := m.GetDefaultTemplate() + if err != nil { + return "", "", err + } + + tmpl, err := template.New("").Parse(templ.Body) + if err != nil { + return "", "", err + } + + var rendered bytes.Buffer + if err := tmpl.Execute(&rendered, data); err != nil { + return "", "", err + } + + return rendered.String(), templ.Subject, nil +} diff --git a/internal/template/template.go b/internal/template/template.go new file mode 100644 index 00000000..c5bba087 --- /dev/null +++ b/internal/template/template.go @@ -0,0 +1,57 @@ +package template + +import ( + "embed" + + "github.com/abhinavxd/artemis/internal/dbutil" + "github.com/abhinavxd/artemis/internal/template/models" + "github.com/jmoiron/sqlx" +) + +var ( + //go:embed queries.sql + efs embed.FS +) + +type Manager struct { + q queries +} + +type queries struct { + InsertTemplate *sqlx.Stmt `query:"insert-template"` + GetTemplate *sqlx.Stmt `query:"get-template"` + GetDefaultTemplate *sqlx.Stmt `query:"get-default-template"` +} + +func New(db *sqlx.DB) (*Manager, error) { + var q queries + + if err := dbutil.ScanSQLFile("queries.sql", &q, db, efs); err != nil { + return nil, err + } + + return &Manager{q}, nil +} + +func (m *Manager) InsertTemplate(name, subject, body string) error { + if _, err := m.q.InsertTemplate.Exec(name, subject, body); err != nil { + return err + } + return nil +} + +func (m *Manager) GetTemplate(name string) (models.Template, error) { + var template models.Template + if err := m.q.GetTemplate.Get(&template, name); err != nil { + return template, err + } + return template, nil +} + +func (m *Manager) GetDefaultTemplate() (models.Template, error) { + var template models.Template + if err := m.q.GetDefaultTemplate.Get(&template); err != nil { + return template, err + } + return template, nil +} diff --git a/internal/user/filterstore/filterstore.go b/internal/user/filterstore/filterstore.go deleted file mode 100644 index db8b2eba..00000000 --- a/internal/user/filterstore/filterstore.go +++ /dev/null @@ -1,42 +0,0 @@ -package filterstore - -import ( - "embed" - - "github.com/abhinavxd/artemis/internal/dbutils" - "github.com/abhinavxd/artemis/internal/user/filterstore/models" - "github.com/jmoiron/sqlx" -) - -var ( - //go:embed queries.sql - efs embed.FS -) - -type Manager struct { - q queries -} - -type queries struct { - GetFilters *sqlx.Stmt `query:"get-user-filters"` -} - -func New(db *sqlx.DB) (*Manager, error) { - var q queries - - if err := dbutils.ScanSQLFile("queries.sql", &q, db, efs); err != nil { - return nil, err - } - - return &Manager{ - q: q, - }, nil -} - -func (m *Manager) GetFilters(userID int, page string) ([]models.Filter, error) { - var filters []models.Filter - if err := m.q.GetFilters.Select(&filters, userID, page); err != nil { - return filters, err - } - return filters, nil -} diff --git a/internal/user/filterstore/models/models.go b/internal/user/filterstore/models/models.go deleted file mode 100644 index 1e8b0f11..00000000 --- a/internal/user/filterstore/models/models.go +++ /dev/null @@ -1,10 +0,0 @@ -package models - -import "encoding/json" - -type Filter struct { - ID int `db:"id" json:"id"` - UserID int `db:"user_id" json:"user_id"` - Page string `db:"page" json:"page"` - Filters json.RawMessage `db:"filters" json:"filters"` -} diff --git a/internal/user/filterstore/queries.sql b/internal/user/filterstore/queries.sql deleted file mode 100644 index af1e88d0..00000000 --- a/internal/user/filterstore/queries.sql +++ /dev/null @@ -1,2 +0,0 @@ --- name: get-user-filters -SELECT * from user_filters where user_id = $1 and page = $2; \ No newline at end of file diff --git a/internal/user/queries.sql b/internal/user/queries.sql index 695a1ecc..479ee037 100644 --- a/internal/user/queries.sql +++ b/internal/user/queries.sql @@ -1,6 +1,9 @@ -- name: get-users SELECT first_name, last_name, uuid, disabled from users; +-- name: get-email +SELECT email from users where CASE WHEN $1 > 0 THEN id = $1 ELSE uuid = $2 END; + -- name: get-user-by-email select id, email, password, avatar_url, first_name, last_name, uuid from users where email = $1; diff --git a/internal/user/user.go b/internal/user/user.go index c5b20366..6603a4e0 100644 --- a/internal/user/user.go +++ b/internal/user/user.go @@ -7,7 +7,7 @@ import ( "errors" "fmt" - "github.com/abhinavxd/artemis/internal/dbutils" + "github.com/abhinavxd/artemis/internal/dbutil" "github.com/abhinavxd/artemis/internal/user/models" "github.com/jmoiron/sqlx" "github.com/zerodha/logf" @@ -39,6 +39,7 @@ type Opts struct { type queries struct { GetUsers *sqlx.Stmt `query:"get-users"` GetUser *sqlx.Stmt `query:"get-user"` + GetEmail *sqlx.Stmt `query:"get-email"` GetUserByEmail *sqlx.Stmt `query:"get-user-by-email"` SetUserPassword *sqlx.Stmt `query:"set-user-password"` } @@ -46,7 +47,7 @@ type queries struct { func New(opts Opts) (*Manager, error) { var q queries - if err := dbutils.ScanSQLFile("queries.sql", &q, opts.DB, efs); err != nil { + if err := dbutil.ScanSQLFile("queries.sql", &q, opts.DB, efs); err != nil { return nil, err } @@ -110,6 +111,23 @@ func (u *Manager) GetUser(id int, uuid string) (models.User, error) { return user, nil } +func (u *Manager) GetEmail(id int, uuid string) (string, error) { + var uu interface{} + if uuid != "" { + uu = uuid + } + + var email string + if err := u.q.GetEmail.Get(&email, id, uu); err != nil { + if errors.Is(err, sql.ErrNoRows) { + return email, fmt.Errorf("user not found") + } + u.lo.Error("error fetching user from db", "error", err) + return email, fmt.Errorf("fetching user: %w", err) + } + return email, nil +} + func (u *Manager) verifyPassword(pwd []byte, pwdHash string) error { err := bcrypt.CompareHashAndPassword([]byte(pwdHash), pwd) if err != nil { diff --git a/internal/ws/client.go b/internal/ws/client.go index 421eb6cc..3287922b 100644 --- a/internal/ws/client.go +++ b/internal/ws/client.go @@ -3,6 +3,7 @@ package ws import ( "encoding/json" "fmt" + "log" "sync" "time" @@ -10,6 +11,13 @@ import ( "github.com/fasthttp/websocket" ) +const ( + // SubscribeConversations to last 1000 conversations. + // TODO: Move to config. + maxConversationsPagesToSub = 10 + maxConversationsPageSize = 100 +) + type SafeBool struct { flag bool mu sync.Mutex @@ -58,8 +66,6 @@ Loop: } case o, ok := <-c.Send: if !ok { - // Disconnected. - fmt.Println("Client disconnected, breaking serve lop.") break Loop } c.Conn.WriteMessage(o.messageType, o.data) @@ -84,8 +90,6 @@ func (c *Client) Listen() { return } } - - fmt.Println("loop broke closing") c.Hub.RemoveClient(c) c.close() } @@ -99,66 +103,95 @@ func (c *Client) processIncomingMessage(b []byte) { } switch r.Action { - // Sub to conversation updates. - case models.ActionConvSub: - var subR = models.ConvSubUnsubReq{} - if err := json.Unmarshal(b, &subR); err != nil { + case models.ActionConversationsSub: + var req = models.ConversationsSubscribe{} + if err := json.Unmarshal(b, &req); err != nil { return } - c.SubConv(int(c.ID), subR.UUIDs...) - case models.ActionConvUnsub: - var subR = models.ConvSubUnsubReq{} - if err := json.Unmarshal(b, &subR); err != nil { + + // First remove all user conversation subscriptions. + c.RemoveAllUserConversationSubscriptions(c.ID) + + // Add the new subcriptions. + for page := range maxConversationsPagesToSub { + page++ + conversationUUIDs, err := c.Hub.conversationStore.GetConversationUUIDs(c.ID, page, maxConversationsPageSize, req.Type, req.PreDefinedFilter) + if err != nil { + log.Println("error fetching convesation ids", err) + continue + } + c.SubscribeConversations(c.ID, conversationUUIDs) + } + case models.ActionConversationSub: + var req = models.ConversationSubscribe{} + if err := json.Unmarshal(b, &req); err != nil { return } - c.UnsubConv(int(c.ID), subR.UUIDs...) - case models.ActionAssignedConvSub: - // Fetch all assigned conversation & sub. - case models.ActionAssignedConvUnSub: - // Fetch all unassigned conversation and sub. + c.SubscribeConversations(c.ID, []string{req.UUID}) + case models.ActionConversationUnSub: + var req = models.ConversationUnsubscribe{} + if err := json.Unmarshal(b, &req); err != nil { + return + } + c.UnsubscribeConversation(c.ID, req.UUID) + default: + fmt.Println("new incoming ws msg ", string(b)) } } func (c *Client) close() { + c.RemoveAllUserConversationSubscriptions(c.ID) c.Closed.Set(true) close(c.Send) } -func (c *Client) SubConv(userID int, uuids ...string) { - c.Hub.SubMut.Lock() - defer c.Hub.SubMut.Unlock() - - for _, uuid := range uuids { - // Initialize the slice if this is the first subscription for this UUID - if _, ok := c.Hub.Csubs[uuid]; !ok { - c.Hub.Csubs[uuid] = []int{} +func (c *Client) SubscribeConversations(userID int, conversationUUIDs []string) { + for _, conversationUUID := range conversationUUIDs { + // Initialize the slice if it doesn't exist + if c.Hub.ConversationSubs[conversationUUID] == nil { + c.Hub.ConversationSubs[conversationUUID] = []int{} } - // Append the user ID to the slice of subscribed user IDs - c.Hub.Csubs[uuid] = append(c.Hub.Csubs[uuid], userID) - } -} -func (c *Client) UnsubConv(userID int, uuids ...string) { - c.Hub.SubMut.Lock() - defer c.Hub.SubMut.Unlock() - - for _, uuid := range uuids { - currentSubs, ok := c.Hub.Csubs[uuid] - if !ok { - continue // No subscriptions for this UUID - } - j := 0 - for _, sub := range currentSubs { - if sub != userID { - currentSubs[j] = sub - j++ + // Check if userID already exists + exists := false + for _, id := range c.Hub.ConversationSubs[conversationUUID] { + if id == userID { + exists = true + break } } - currentSubs = currentSubs[:j] // Update the slice in-place - if len(currentSubs) == 0 { - delete(c.Hub.Csubs, uuid) // Remove key if no more subscriptions - } else { - c.Hub.Csubs[uuid] = currentSubs + + // Add userID if it doesn't exist + if !exists { + c.Hub.ConversationSubs[conversationUUID] = append(c.Hub.ConversationSubs[conversationUUID], userID) + } + } +} + +func (c *Client) UnsubscribeConversation(userID int, conversationUUID string) { + if userIDs, ok := c.Hub.ConversationSubs[conversationUUID]; ok { + for i, id := range userIDs { + if id == userID { + c.Hub.ConversationSubs[conversationUUID] = append(userIDs[:i], userIDs[i+1:]...) + break + } + } + if len(c.Hub.ConversationSubs[conversationUUID]) == 0 { + delete(c.Hub.ConversationSubs, conversationUUID) + } + } +} + +func (c *Client) RemoveAllUserConversationSubscriptions(userID int) { + for conversationID, userIDs := range c.Hub.ConversationSubs { + for i, id := range userIDs { + if id == userID { + c.Hub.ConversationSubs[conversationID] = append(userIDs[:i], userIDs[i+1:]...) + break + } + } + if len(c.Hub.ConversationSubs[conversationID]) == 0 { + delete(c.Hub.ConversationSubs, conversationID) } } } diff --git a/internal/ws/models/models.go b/internal/ws/models/models.go index 0e7c7da5..8f6815b4 100644 --- a/internal/ws/models/models.go +++ b/internal/ws/models/models.go @@ -1,24 +1,36 @@ package models const ( - ActionConvSub = "c_sub" - ActionConvUnsub = "c_unsub" - ActionAssignedConvSub = "a_c_sub" - ActionAssignedConvUnSub = "a_c_unsub" - - EventNewMsg = "new_msg" - EventMsgStatusUpdate = "msg_status_update" + ActionConversationsSub = "conversations_sub" + ActionConversationSub = "conversation_sub" + ActionConversationUnSub = "conversation_unsub" + MessageTypeNewMessage = "new_msg" + MessageTypeMessagePropUpdate = "msg_prop_update" + MessageTypeNewConversation = "new_conv" + MessageTypeConversationPropertyUpdate = "conv_prop_update" ) type IncomingReq struct { Action string `json:"a"` } +type ConversationsSubscribe struct { + Type string `json:"t"` + PreDefinedFilter string `json:"pf"` +} + +type ConversationSubscribe struct { + UUID string `json:"uuid"` +} +type ConversationUnsubscribe struct { + UUID string `json:"uuid"` +} + type ConvSubUnsubReq struct { UUIDs []string `json:"v"` } -type Event struct { - Type string - Data string +type Message struct { + Type string `json:"typ"` + Data interface{} `json:"d"` } diff --git a/internal/ws/ws.go b/internal/ws/ws.go index 78a63dc5..d91ac5ee 100644 --- a/internal/ws/ws.go +++ b/internal/ws/ws.go @@ -4,27 +4,34 @@ import ( "encoding/json" "fmt" "sync" + "time" "github.com/abhinavxd/artemis/internal/ws/models" "github.com/fasthttp/websocket" ) -// Hub maintains the set of registered clients. +// Hub maintains the set of registered clients and their subscriptions. type Hub struct { clients map[int][]*Client clientsMutex sync.Mutex - // Map of conversation uuid to slice of subbed userids. - Csubs map[string][]int - SubMut sync.Mutex + // Map of conversation uuid to a set of subscribed user IDs. + ConversationSubs map[string][]int + + SubMut sync.Mutex + conversationStore ConversationStore +} + +type ConversationStore interface { + GetConversationUUIDs(userID, page, pageSize int, typ, predefinedFilter string) ([]string, error) } func NewHub() *Hub { return &Hub{ - clients: make(map[int][]*Client, 100000), - clientsMutex: sync.Mutex{}, - Csubs: map[string][]int{}, - SubMut: sync.Mutex{}, + clients: make(map[int][]*Client, 10000), + clientsMutex: sync.Mutex{}, + ConversationSubs: make(map[string][]int), + SubMut: sync.Mutex{}, } } @@ -39,6 +46,10 @@ type PushMessage struct { MaxUsers int `json:"max_users"` } +func (h *Hub) SetConversationStore(store ConversationStore) { + h.conversationStore = store +} + func (h *Hub) AddClient(c *Client) { h.clientsMutex.Lock() defer h.clientsMutex.Unlock() @@ -59,7 +70,7 @@ func (h *Hub) RemoveClient(client *Client) { } } -// ClientAlreadyConnected checks if a user id is already connected or not. +// ClientAlreadyConnected returns true if the client with this id is already connected else returns false. func (h *Hub) ClientAlreadyConnected(id int) bool { h.clientsMutex.Lock() defer h.clientsMutex.Unlock() @@ -75,7 +86,6 @@ func (h *Hub) PushMessage(m PushMessage) { h.clientsMutex.Lock() for _, userID := range m.Users { for _, c := range h.clients[userID] { - fmt.Printf("Pushing msg to %d", userID) c.Conn.WriteMessage(websocket.TextMessage, m.Data) } } @@ -97,52 +107,89 @@ func (h *Hub) PushMessage(m PushMessage) { } } -func (c *Hub) BroadcastNewMsg(convUUID string, msg map[string]interface{}) { - // clientIDs, ok := c.Csubs[convUUID] - // if !ok || len(clientIDs) == 0 { - // return - // } - - clientIDs := []int{1, 2} - - data := map[string]interface{}{ - "ev": models.EventNewMsg, - "d": msg, +func (c *Hub) BroadcastNewConversationMessage(conversationUUID, trimmedMessage, messageUUID, lastMessageAt string, private bool) { + userIDs, ok := c.ConversationSubs[conversationUUID] + if !ok || len(userIDs) == 0 { + return } - // Marshal. - dataB, err := json.Marshal(data) + message := models.Message{ + Type: models.MessageTypeNewMessage, + Data: map[string]interface{}{ + "conversation_uuid": conversationUUID, + "last_message": trimmedMessage, + "uuid": messageUUID, + "last_message_at": lastMessageAt, + "private": private, + }, + } + + c.marshalAndPush(message, userIDs) +} + +func (c *Hub) BroadcastMessagePropUpdate(conversationUUID, messageUUID, prop, value string) { + userIDs, ok := c.ConversationSubs[conversationUUID] + if !ok || len(userIDs) == 0 { + return + } + + message := models.Message{ + Type: models.MessageTypeMessagePropUpdate, + Data: map[string]interface{}{ + "uuid": messageUUID, + "prop": prop, + "val": value, + }, + } + + c.marshalAndPush(message, userIDs) +} + +func (c *Hub) BroadcastConversationAssignment(userID int, conversationUUID string, avatarUrl string, firstName, lastName, lastMessage, inboxName string, lastMessageAt time.Time, unreadMessageCount int) { + message := models.Message{ + Type: models.MessageTypeNewConversation, + Data: map[string]interface{}{ + "uuid": conversationUUID, + "avatar_url": avatarUrl, + "first_name": firstName, + "last_name": lastName, + "last_message": lastMessage, + "last_message_at": time.Now().Format(time.RFC3339), + "inbox_name": inboxName, + "unread_message_count": unreadMessageCount, + }, + } + c.marshalAndPush(message, []int{userID}) +} + +func (c *Hub) BroadcastConversationPropertyUpdate(conversationUUID string, prop, val string) { + userIDs, ok := c.ConversationSubs[conversationUUID] + if !ok || len(userIDs) == 0 { + return + } + + message := models.Message{ + Type: models.MessageTypeConversationPropertyUpdate, + Data: map[string]interface{}{ + "uuid": conversationUUID, + "prop": prop, + "val": val, + }, + } + + c.marshalAndPush(message, userIDs) +} + +func (c *Hub) marshalAndPush(message models.Message, userIDs []int) { + messageB, err := json.Marshal(message) if err != nil { return } - c.PushMessage(PushMessage{ - Data: dataB, - Users: clientIDs, - }) -} - -func (c *Hub) BroadcastMsgStatus(convUUID string, msg map[string]interface{}) { - // clientIDs, ok := c.Csubs[convUUID] - // if !ok || len(clientIDs) == 0 { - // return - // } - - clientIDs := []int{1, 2} - - data := map[string]interface{}{ - "ev": models.EventMsgStatusUpdate, - "d": msg, - } - - // Marshal. - dataB, err := json.Marshal(data) - if err != nil { - return - } + fmt.Println("pushing msg", string(messageB), "type", message.Type, "to_user_ids", userIDs, "connected_userIds", len(c.clients)) c.PushMessage(PushMessage{ - Data: dataB, - Users: clientIDs, + Data: messageB, + Users: userIDs, }) }