From dbc5a1a28f254a13d3dcbbb07a4c5246fe2afbbc Mon Sep 17 00:00:00 2001 From: Rogee Date: Thu, 4 Jun 2026 20:06:21 +0800 Subject: [PATCH] feat(search): sync entity writes to indexer --- docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md | 3 +- internal/app/bootstrap.go | 5 + internal/handler/api/v1/article_handler.go | 9 +- internal/search/search_service.go | 55 +++++++ internal/service/article_service.go | 120 +++++++++----- internal/service/company_service.go | 27 ++- internal/service/contact_service.go | 25 ++- internal/service/conversation_service.go | 80 ++++++--- internal/service/message_service.go | 30 +++- internal/service/search_indexer.go | 29 ++++ internal/service/search_indexer_hooks_test.go | 155 ++++++++++++++++++ 11 files changed, 468 insertions(+), 70 deletions(-) create mode 100644 internal/service/search_indexer.go create mode 100644 internal/service/search_indexer_hooks_test.go diff --git a/docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md b/docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md index ae40aa4f..02ef8f5e 100644 --- a/docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md +++ b/docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md @@ -80,7 +80,7 @@ Tracking table: | P1.3 | Implement Meilisearch engine wrapper, index naming, bootstrap, sortable/filterable/searchable settings. | `internal/search/engine_meili.go` | Chatwoot search models/services | Review | | P1.4 | Keep existing DB search as explicit dev fallback only. Production config must prefer Meilisearch. | `internal/search/engine_db.go`, `internal/search/search_service.go` | User decision on Meilisearch | Done | | P1.5 | Define documents and serializers for conversations, messages, contacts, companies, articles, and help-center content. | `internal/search/engine.go` | `reference/chatwoot` models/serializers | Review | -| P1.6 | Wire create/update/delete hooks from entity services into async or synchronous indexing boundary. | `internal/service/*`, `internal/repository/*`, `internal/worker/*` | Chatwoot callbacks/jobs | Todo | +| P1.6 | Wire create/update/delete hooks from entity services into async or synchronous indexing boundary. | `internal/service/*`, `internal/search/search_service.go`, `internal/app/bootstrap.go` | Chatwoot callbacks/jobs | Done | | P1.7 | Add batch reindex command and account/entity filters. | `cmd/reindex_search` | Chatwoot reindex/search tasks | Done | | P1.8 | Add mocked engine tests and service integration tests without requiring live Meilisearch. | `internal/search/engine_test.go`, `internal/config/config_test.go` | Existing test style | Done | | P1.9 | Document Meilisearch env vars and local startup flow. | this doc, ops docs if needed | Hermes plan | Done | @@ -313,3 +313,4 @@ env GOCACHE=/tmp/gochat-gocache GOMODCACHE=/tmp/gochat-gomodcache go run ./cmd/d - 2026-06-04: Baseline stabilized and committed as `42cdab8 chore: stabilize chatwoot parity baseline`; `go test ./...` passed and route dump reported `TOTAL: 704`. - 2026-06-04: Phase 1 search foundation added: Meilisearch config/env defaults, `SearchEngine` contract, Meilisearch HTTP wrapper with bootstrap/settings, DB fallback adapter, document builders, reindex command, and no-live-Meilisearch tests. Verified `go test ./...` in unsandboxed mode because miniredis/httptest need local sockets; route dump still reports `TOTAL: 704`. +- 2026-06-04: Phase 1 indexing hooks wired for conversations, messages, contacts, companies, and articles. Create/update/delete paths now call the service-layer `SearchIndexer` boundary, bootstrap injects the Meilisearch-backed search service, and unit tests cover each entity hook path. diff --git a/internal/app/bootstrap.go b/internal/app/bootstrap.go index 1ab7ac61..a4b73382 100644 --- a/internal/app/bootstrap.go +++ b/internal/app/bootstrap.go @@ -632,6 +632,11 @@ func Bootstrap(env string) (*App, error) { return nil, fmt.Errorf("search engine init failed: %w", err) } searchService := search.NewSearchServiceWithEngine(searchEngine, searchRepo) + conversationService.SetSearchIndexer(searchService) + messageService.SetSearchIndexer(searchService) + contactService.SetSearchIndexer(searchService) + companyService.SetSearchIndexer(searchService) + articleService.SetSearchIndexer(searchService) // Custom attribute definition + custom filter + custom attribute value services customAttributeDefinitionService := service.NewCustomAttributeDefinitionService(customAttributeDefinitionRepo) diff --git a/internal/handler/api/v1/article_handler.go b/internal/handler/api/v1/article_handler.go index 44122b09..b735bfd3 100644 --- a/internal/handler/api/v1/article_handler.go +++ b/internal/handler/api/v1/article_handler.go @@ -31,6 +31,11 @@ func (h *ArticleHandler) Create(c *gin.Context) { response.AbortWithStatusError(c, http.StatusBadRequest, response.ErrBadRequest, "invalid portal_id") return } + accountID, err := parseUintParam(c, "account_id") + if err != nil { + response.AbortWithStatusError(c, http.StatusBadRequest, response.ErrBadRequest, "invalid account id") + return + } // Chatwoot: params.require(:article) → {"article": {...}} var wrapper struct { @@ -51,7 +56,7 @@ func (h *ArticleHandler) Create(c *gin.Context) { authorID = uint(authorID64) } - article, err := h.svc.Create(c.Request.Context(), uint(portalID), authorID, &req) + article, err := h.svc.CreateWithAccount(c.Request.Context(), accountID, uint(portalID), authorID, &req) if err != nil { applogger.L().Errorf("Create article: %v", err) response.AbortWithStatusError(c, http.StatusInternalServerError, response.ErrInternal, "failed to create article") @@ -387,4 +392,4 @@ func (h *ArticleHandler) BulkActions(c *gin.Context) { } response.OK(c, gin.H{"action": req.Action, "ids": req.IDs}) -} \ No newline at end of file +} diff --git a/internal/search/search_service.go b/internal/search/search_service.go index 562be615..3f21e08d 100644 --- a/internal/search/search_service.go +++ b/internal/search/search_service.go @@ -289,6 +289,61 @@ func (s *SearchService) IndexDocument(ctx context.Context, doc SearchDocument) e return s.engine.IndexDocument(ctx, doc) } +func (s *SearchService) IndexConversation(ctx context.Context, conversation *model.Conversation) error { + if conversation == nil { + return nil + } + return s.IndexDocument(ctx, ConversationDocument(*conversation)) +} + +func (s *SearchService) DeleteConversation(ctx context.Context, accountID uint, id uint) error { + return s.DeleteDocument(ctx, ResultTypeConversation, accountID, id) +} + +func (s *SearchService) IndexMessage(ctx context.Context, message *model.Message) error { + if message == nil { + return nil + } + return s.IndexDocument(ctx, MessageDocument(*message)) +} + +func (s *SearchService) DeleteMessage(ctx context.Context, accountID uint, id uint) error { + return s.DeleteDocument(ctx, ResultTypeMessage, accountID, id) +} + +func (s *SearchService) IndexContact(ctx context.Context, contact *model.Contact) error { + if contact == nil { + return nil + } + return s.IndexDocument(ctx, ContactDocument(*contact)) +} + +func (s *SearchService) DeleteContact(ctx context.Context, accountID uint, id uint) error { + return s.DeleteDocument(ctx, ResultTypeContact, accountID, id) +} + +func (s *SearchService) IndexCompany(ctx context.Context, company *model.Company) error { + if company == nil { + return nil + } + return s.IndexDocument(ctx, CompanyDocument(*company)) +} + +func (s *SearchService) DeleteCompany(ctx context.Context, accountID uint, id uint) error { + return s.DeleteDocument(ctx, ResultTypeCompany, accountID, id) +} + +func (s *SearchService) IndexArticle(ctx context.Context, article *model.Article) error { + if article == nil { + return nil + } + return s.IndexDocument(ctx, ArticleDocument(*article)) +} + +func (s *SearchService) DeleteArticle(ctx context.Context, accountID uint, id uint) error { + return s.DeleteDocument(ctx, ResultTypeArticle, accountID, id) +} + func (s *SearchService) IndexBatch(ctx context.Context, docs []SearchDocument) error { if s.engine == nil { return nil diff --git a/internal/service/article_service.go b/internal/service/article_service.go index 3af7109d..a4f23c7d 100644 --- a/internal/service/article_service.go +++ b/internal/service/article_service.go @@ -12,45 +12,66 @@ import ( // ArticleService implements business logic for Article CRUD. type ArticleService struct { - repo *repository.ArticleRepo + repo *repository.ArticleRepo + searchIndexer SearchIndexer } func NewArticleService(repo *repository.ArticleRepo) *ArticleService { return &ArticleService{repo: repo} } +func (s *ArticleService) SetSearchIndexer(indexer SearchIndexer) { + s.searchIndexer = indexer +} + +func (s *ArticleService) indexArticle(ctx context.Context, article *model.Article) { + if s.searchIndexer != nil { + logSearchIndexError("article", article.ID, s.searchIndexer.IndexArticle(ctx, article)) + } +} + +func (s *ArticleService) deleteArticleIndex(ctx context.Context, accountID uint, id uint) { + if s.searchIndexer != nil { + logSearchIndexError("article", id, s.searchIndexer.DeleteArticle(ctx, accountID, id)) + } +} + // CreateArticleRequest is the DTO for creating an article. type CreateArticleRequest struct { - Title string `json:"title" validate:"required,min=2"` - Slug string `json:"slug" validate:"required"` - Description string `json:"description"` - Content string `json:"content"` - Status model.ArticleStatus `json:"status"` - FolderID *uint `json:"folder_id"` - CategoryID *uint `json:"category_id"` - Position int `json:"position"` - Locale string `json:"locale"` - AssociatedArticleID *uint `json:"associated_article_id"` - Meta json.RawMessage `json:"meta"` - CustomAttributes json.RawMessage `json:"custom_attributes"` + Title string `json:"title" validate:"required,min=2"` + Slug string `json:"slug" validate:"required"` + Description string `json:"description"` + Content string `json:"content"` + Status model.ArticleStatus `json:"status"` + FolderID *uint `json:"folder_id"` + CategoryID *uint `json:"category_id"` + Position int `json:"position"` + Locale string `json:"locale"` + AssociatedArticleID *uint `json:"associated_article_id"` + Meta json.RawMessage `json:"meta"` + CustomAttributes json.RawMessage `json:"custom_attributes"` } // UpdateArticleRequest is the DTO for updating an article. type UpdateArticleRequest struct { - Title string `json:"title"` - Description string `json:"description"` - Content string `json:"content"` - Status model.ArticleStatus `json:"status"` - FolderID *uint `json:"folder_id"` - CategoryID *uint `json:"category_id"` - Position *int `json:"position"` - Locale string `json:"locale"` - AssociatedArticleID *uint `json:"associated_article_id"` - Meta json.RawMessage `json:"meta"` - CustomAttributes json.RawMessage `json:"custom_attributes"` + Title string `json:"title"` + Description string `json:"description"` + Content string `json:"content"` + Status model.ArticleStatus `json:"status"` + FolderID *uint `json:"folder_id"` + CategoryID *uint `json:"category_id"` + Position *int `json:"position"` + Locale string `json:"locale"` + AssociatedArticleID *uint `json:"associated_article_id"` + Meta json.RawMessage `json:"meta"` + CustomAttributes json.RawMessage `json:"custom_attributes"` } func (s *ArticleService) Create(ctx context.Context, portalID uint, authorID uint, req *CreateArticleRequest) (*model.Article, error) { + return s.CreateWithAccount(ctx, 0, portalID, authorID, req) +} + +func (s *ArticleService) CreateWithAccount(ctx context.Context, accountID uint, portalID uint, authorID uint, req *CreateArticleRequest) (*model.Article, error) { // Default status is draft if not specified status := string(req.Status) if status == "" { @@ -58,20 +79,21 @@ func (s *ArticleService) Create(ctx context.Context, portalID uint, authorID uin } article := &model.Article{ - PortalID: portalID, - AuthorID: &authorID, - Title: req.Title, - Slug: req.Slug, - Description: req.Description, - Content: req.Content, - Status: status, - FolderID: req.FolderID, - CategoryID: req.CategoryID, - Position: req.Position, - Locale: req.Locale, + AccountID: accountID, + PortalID: portalID, + AuthorID: &authorID, + Title: req.Title, + Slug: req.Slug, + Description: req.Description, + Content: req.Content, + Status: status, + FolderID: req.FolderID, + CategoryID: req.CategoryID, + Position: req.Position, + Locale: req.Locale, AssociatedArticleID: req.AssociatedArticleID, - Meta: req.Meta, - CustomAttributes: req.CustomAttributes, + Meta: req.Meta, + CustomAttributes: req.CustomAttributes, } if err := s.repo.Create(ctx, article); err != nil { @@ -82,6 +104,7 @@ func (s *ArticleService) Create(ctx context.Context, portalID uint, authorID uin if req.AssociatedArticleID != nil { applogger.L().Debugf("Article %d linked to root article %d", article.ID, *req.AssociatedArticleID) } + s.indexArticle(ctx, article) return article, nil } @@ -138,17 +161,20 @@ func (s *ArticleService) Update(ctx context.Context, id uint, req *UpdateArticle if err := s.repo.Update(ctx, article); err != nil { return nil, fmt.Errorf("update article: %w", err) } + s.indexArticle(ctx, article) return article, nil } func (s *ArticleService) Delete(ctx context.Context, id uint) error { // Check existence first — GORM Delete() returns nil even for non-existent IDs - if _, err := s.repo.GetByID(ctx, id); err != nil { + article, err := s.repo.GetByID(ctx, id) + if err != nil { return fmt.Errorf("find article for delete: %w", err) } if err := s.repo.Delete(ctx, id); err != nil { return fmt.Errorf("delete article: %w", err) } + s.deleteArticleIndex(ctx, article.AccountID, id) return nil } @@ -223,14 +249,30 @@ func (s *ArticleService) BulkUpdateStatus(ctx context.Context, ids []uint, statu if err := s.repo.BulkUpdateStatus(ctx, ids, status); err != nil { return fmt.Errorf("bulk update status: %w", err) } + for _, id := range ids { + article, err := s.repo.GetByID(ctx, id) + if err == nil { + s.indexArticle(ctx, article) + } + } return nil } // BulkDelete deletes multiple articles. func (s *ArticleService) BulkDelete(ctx context.Context, ids []uint) error { + articles := make([]*model.Article, 0, len(ids)) + for _, id := range ids { + article, err := s.repo.GetByID(ctx, id) + if err == nil { + articles = append(articles, article) + } + } if err := s.repo.BulkDelete(ctx, ids); err != nil { return fmt.Errorf("bulk delete: %w", err) } + for _, article := range articles { + s.deleteArticleIndex(ctx, article.AccountID, article.ID) + } return nil } @@ -262,4 +304,4 @@ func (s *ArticleService) BulkActions(ctx context.Context, req *BulkActionsReques default: return fmt.Errorf("unsupported action: %s — must be publish, archive, or delete", req.Action) } -} \ No newline at end of file +} diff --git a/internal/service/company_service.go b/internal/service/company_service.go index 10c91dcd..dafa1bde 100644 --- a/internal/service/company_service.go +++ b/internal/service/company_service.go @@ -19,6 +19,7 @@ type CompanyService struct { companyRepo *repository.CompanyRepo contactRepo *repository.ContactRepo conversationRepo *repository.ConversationRepo + searchIndexer SearchIndexer } // NewCompanyService creates a new Company service. @@ -30,6 +31,22 @@ func NewCompanyService(companyRepo *repository.CompanyRepo, contactRepo *reposit } } +func (s *CompanyService) SetSearchIndexer(indexer SearchIndexer) { + s.searchIndexer = indexer +} + +func (s *CompanyService) indexCompany(ctx context.Context, company *model.Company) { + if s.searchIndexer != nil { + logSearchIndexError("company", company.ID, s.searchIndexer.IndexCompany(ctx, company)) + } +} + +func (s *CompanyService) deleteCompanyIndex(ctx context.Context, accountID uint, id uint) { + if s.searchIndexer != nil { + logSearchIndexError("company", id, s.searchIndexer.DeleteCompany(ctx, accountID, id)) + } +} + // CreateCompanyRequest is the DTO for creating a company. type CreateCompanyRequest struct { Name string `json:"name" validate:"required,min=1"` @@ -97,6 +114,7 @@ func (s *CompanyService) Create(ctx context.Context, accountID uint, req *Create applogger.L().Errorf("Create company for account %d: %v", accountID, err) return nil, err } + s.indexCompany(ctx, company) return company, nil } @@ -134,6 +152,7 @@ func (s *CompanyService) Update(ctx context.Context, id, accountID uint, req *Up applogger.L().Errorf("Update company %d for account %d: %v", id, accountID, err) return nil, err } + s.indexCompany(ctx, company) return company, nil } @@ -146,7 +165,11 @@ func (s *CompanyService) Delete(ctx context.Context, id, accountID uint) error { if company == nil { return errors.New("company not found") } - return s.companyRepo.Delete(ctx, id) + if err := s.companyRepo.Delete(ctx, id); err != nil { + return err + } + s.deleteCompanyIndex(ctx, accountID, id) + return nil } // ListContacts retrieves contacts associated with a company. @@ -275,4 +298,4 @@ func (s *CompanyService) RemoveContact(ctx context.Context, companyID, accountID return err } return nil -} \ No newline at end of file +} diff --git a/internal/service/contact_service.go b/internal/service/contact_service.go index b92291a9..5eb19885 100644 --- a/internal/service/contact_service.go +++ b/internal/service/contact_service.go @@ -23,6 +23,7 @@ type ContactService struct { repo *repository.ContactRepo contactInboxSvc *ContactInboxService noteRepo *repository.NoteRepo + searchIndexer SearchIndexer } // NewContactService creates a new Contact service. @@ -30,6 +31,22 @@ func NewContactService(repo *repository.ContactRepo, contactInboxSvc *ContactInb return &ContactService{repo: repo, contactInboxSvc: contactInboxSvc, noteRepo: noteRepo} } +func (s *ContactService) SetSearchIndexer(indexer SearchIndexer) { + s.searchIndexer = indexer +} + +func (s *ContactService) indexContact(ctx context.Context, contact *model.Contact) { + if s.searchIndexer != nil { + logSearchIndexError("contact", contact.ID, s.searchIndexer.IndexContact(ctx, contact)) + } +} + +func (s *ContactService) deleteContactIndex(ctx context.Context, accountID uint, id uint) { + if s.searchIndexer != nil { + logSearchIndexError("contact", id, s.searchIndexer.DeleteContact(ctx, accountID, id)) + } +} + // Ready reports whether the service has the dependencies required for DB-backed operations. func (s *ContactService) Ready() bool { return s != nil && s.repo != nil @@ -118,6 +135,7 @@ func (s *ContactService) Create(ctx context.Context, accountID uint, req CreateC applogger.L().Errorf("Failed to create contact: %v", err) return nil, err } + s.indexContact(ctx, contact) // Auto-create ContactInbox when inbox_id is provided (Chatwoot pattern) if req.InboxID != nil && *req.InboxID > 0 { @@ -210,6 +228,7 @@ func (s *ContactService) Update(ctx context.Context, accountID, id uint, req Upd if err := s.repo.Update(ctx, contact); err != nil { return nil, err } + s.indexContact(ctx, contact) return contact, nil } @@ -219,7 +238,11 @@ func (s *ContactService) Delete(ctx context.Context, accountID, id uint) error { if err != nil { return err } - return s.repo.Delete(ctx, contact.ID) + if err := s.repo.Delete(ctx, contact.ID); err != nil { + return err + } + s.deleteContactIndex(ctx, accountID, contact.ID) + return nil } // CreateNoteRequest is the DTO for creating a contact note. diff --git a/internal/service/conversation_service.go b/internal/service/conversation_service.go index cae85ef0..5a8c48c5 100644 --- a/internal/service/conversation_service.go +++ b/internal/service/conversation_service.go @@ -20,13 +20,14 @@ import ( // ConversationService implements business logic for Conversation operations. // Reference: Chatwoot app/controllers/api/v1/conversations_controller.rb type ConversationService struct { - repo *repository.ConversationRepo - msgRepo *repository.MessageRepo - dispatcher *channel.Dispatcher - inboxMemberSvc *InboxMemberService - accountUserRepo *repository.AccountUserRepo - teamRepo *repository.TeamRepo - teamMemberRepo *repository.TeamMemberRepo + repo *repository.ConversationRepo + msgRepo *repository.MessageRepo + dispatcher *channel.Dispatcher + inboxMemberSvc *InboxMemberService + accountUserRepo *repository.AccountUserRepo + teamRepo *repository.TeamRepo + teamMemberRepo *repository.TeamMemberRepo + searchIndexer SearchIndexer } // NewConversationService creates a new Conversation service. @@ -34,6 +35,28 @@ func NewConversationService(repo *repository.ConversationRepo, msgRepo *reposito return &ConversationService{repo: repo, msgRepo: msgRepo, dispatcher: dispatcher, inboxMemberSvc: inboxMemberSvc, accountUserRepo: accountUserRepo, teamRepo: teamRepo, teamMemberRepo: teamMemberRepo} } +func (s *ConversationService) SetSearchIndexer(indexer SearchIndexer) { + s.searchIndexer = indexer +} + +func (s *ConversationService) indexConversation(ctx context.Context, conversation *model.Conversation) { + if s.searchIndexer != nil { + logSearchIndexError("conversation", conversation.ID, s.searchIndexer.IndexConversation(ctx, conversation)) + } +} + +func (s *ConversationService) deleteConversationIndex(ctx context.Context, accountID uint, id uint) { + if s.searchIndexer != nil { + logSearchIndexError("conversation", id, s.searchIndexer.DeleteConversation(ctx, accountID, id)) + } +} + +func (s *ConversationService) indexMessage(ctx context.Context, message *model.Message) { + if s.searchIndexer != nil { + logSearchIndexError("message", message.ID, s.searchIndexer.IndexMessage(ctx, message)) + } +} + // dispatchConversationEvent is a helper to build and dispatch a conversation event. func (s *ConversationService) dispatchConversationEvent(ctx context.Context, eventType channel.EventType, conversation *model.Conversation) { event := channel.NewChannelEvent(eventType, channel.ChannelType(conversation.ChannelType), conversation.AccountID, conversation.InboxID) @@ -127,6 +150,7 @@ func (s *ConversationService) Create(ctx context.Context, accountID uint, req Cr // Dispatch EventConversationCreated s.dispatchConversationEvent(ctx, channel.EventConversationCreated, conversation) + s.indexConversation(ctx, conversation) // If status is open, also dispatch EventConversationOpened if status == model.ConversationStatusOpen { @@ -153,6 +177,7 @@ func (s *ConversationService) Create(ctx context.Context, accountID uint, req Cr applogger.L().Errorf("Failed to create initial message for conversation %d: %v", conversation.ID, err) // Non-critical: conversation was created, message creation failure is logged but not fatal } else { + s.indexMessage(ctx, initialMsg) s.dispatcher.Dispatch(ctx, channel.NewChannelEvent(channel.EventMessageCreated, channel.ChannelAPI, accountID, req.InboxID)) } } @@ -192,6 +217,7 @@ func (s *ConversationService) Update(ctx context.Context, accountID, id uint, re // Dispatch EventConversationUpdated s.dispatchConversationEvent(ctx, channel.EventConversationUpdated, conversation) + s.indexConversation(ctx, conversation) // If status changed, dispatch appropriate status event if req.Status != "" && req.Status != oldStatus { @@ -233,6 +259,7 @@ func (s *ConversationService) AssignAgent(ctx context.Context, accountID, id uin if err := s.dispatcher.Dispatch(ctx, event); err != nil { applogger.L().Errorf("failed to dispatch event %s: %v", channel.EventConversationUnassigned, err) } + s.indexConversation(ctx, conversation) return conversation, nil } @@ -271,6 +298,7 @@ func (s *ConversationService) AssignAgent(ctx context.Context, accountID, id uin if err := s.dispatcher.Dispatch(ctx, event); err != nil { applogger.L().Errorf("failed to dispatch event %s: %v", channel.EventConversationAssigned, err) } + s.indexConversation(ctx, conversation) return conversation, nil } @@ -286,10 +314,10 @@ func (s *ConversationService) UnassignAgent(ctx context.Context, accountID, id u // Reference: Chatwoot app/controllers/api/v1/accounts/conversations_controller.rb #toggle_status type ToggleStatusRequest struct { Status string `json:"status" validate:"required,oneof=open resolved pending snoozed"` - AssigneeID *uint `json:"assignee_id,omitempty"` // Chatwoot: auto-assign on reopen - SnoozedUntil *int64 `json:"snoozed_until,omitempty"` // Chatwoot: snooze with wake-up time (unix timestamp) - UserID *uint `json:"user_id,omitempty"` // Chatwoot: should_assign_conversation — auto-assign to agent who opens - IsBot bool `json:"is_bot,omitempty"` // Chatwoot: pending_to_open_by_bot — agent bot triggers handoff + AssigneeID *uint `json:"assignee_id,omitempty"` // Chatwoot: auto-assign on reopen + SnoozedUntil *int64 `json:"snoozed_until,omitempty"` // Chatwoot: snooze with wake-up time (unix timestamp) + UserID *uint `json:"user_id,omitempty"` // Chatwoot: should_assign_conversation — auto-assign to agent who opens + IsBot bool `json:"is_bot,omitempty"` // Chatwoot: pending_to_open_by_bot — agent bot triggers handoff } // ToggleStatus toggles the conversation status with Chatwoot-compatible logic: @@ -319,6 +347,7 @@ func (s *ConversationService) ToggleStatus(ctx context.Context, accountID, id ui conversation.Status = string(model.ConversationStatusOpen) // Fire bot handoff event (Chatwoot dispatches conversation.bot_handoff!) s.dispatchConversationEvent(ctx, channel.EventConversationOpened, conversation) + s.indexConversation(ctx, conversation) return conversation, nil } @@ -377,6 +406,7 @@ func (s *ConversationService) ToggleStatus(ctx context.Context, accountID, id ui // For pending/snoozed, dispatch generic updated event s.dispatchConversationEvent(ctx, channel.EventConversationUpdated, conversation) } + s.indexConversation(ctx, conversation) return conversation, nil } @@ -410,6 +440,7 @@ func (s *ConversationService) UpdateLabels(ctx context.Context, accountID, id ui // Dispatch EventConversationLabelsUpdated s.dispatchConversationEvent(ctx, channel.EventConversationLabelsUpdated, conversation) + s.indexConversation(ctx, conversation) return conversation, nil } @@ -427,6 +458,7 @@ func (s *ConversationService) Delete(ctx context.Context, accountID, id uint) er // Dispatch EventConversationDeleted s.dispatchConversationEvent(ctx, channel.EventConversationDeleted, conversation) + s.deleteConversationIndex(ctx, accountID, conversation.ID) return nil } @@ -457,6 +489,7 @@ func (s *ConversationService) Mute(ctx context.Context, accountID, id uint) (*mo // Dispatch EventConversationMuted s.dispatchConversationEvent(ctx, channel.EventConversationMuted, conversation) + s.indexConversation(ctx, conversation) return conversation, nil } @@ -485,6 +518,7 @@ func (s *ConversationService) Unmute(ctx context.Context, accountID, id uint) (* // Dispatch EventConversationUnmuted s.dispatchConversationEvent(ctx, channel.EventConversationUnmuted, conversation) + s.indexConversation(ctx, conversation) return conversation, nil } @@ -506,8 +540,8 @@ type FilterParams struct { ConversationType string `json:"conversation_type,omitempty" form:"conversation_type" validate:"omitempty,oneof=mention participating unattended"` SortBy string `json:"sort_by,omitempty" form:"sort_by" validate:"omitempty,oneof=last_activity_at_asc last_activity_at_desc created_at_asc created_at_desc priority_asc priority_desc waiting_since_asc waiting_since_desc latest sort_on_created_at sort_on_priority sort_on_waiting_since"` UpdatedWithin *int `json:"updated_within,omitempty" form:"updated_within"` // seconds - Query string `json:"q,omitempty" form:"q"` // 1:1 Chatwoot: filter_by_query — search messages ILIKE - SourceID string `json:"source_id,omitempty" form:"source_id"` // 1:1 Chatwoot: filter_by_source_id — contact_inbox.source_id + Query string `json:"q,omitempty" form:"q"` // 1:1 Chatwoot: filter_by_query — search messages ILIKE + SourceID string `json:"source_id,omitempty" form:"source_id"` // 1:1 Chatwoot: filter_by_source_id — contact_inbox.source_id // AllowedInboxIDs is populated at runtime — the list of inbox IDs the user can access. // Not populated from request params; set by the service based on user permissions. @@ -523,10 +557,10 @@ type FilterResult struct { } type FilterCountMeta struct { - MineCount int64 `json:"mine_count"` - AssignedCount int64 `json:"assigned_count"` + MineCount int64 `json:"mine_count"` + AssignedCount int64 `json:"assigned_count"` UnassignedCount int64 `json:"unassigned_count"` - AllCount int64 `json:"all_count"` + AllCount int64 `json:"all_count"` } // Filter retrieves conversations matching advanced filter criteria. @@ -612,7 +646,7 @@ func (s *ConversationService) Filter(ctx context.Context, accountID uint, userID // 8. Filter by updated_within (seconds) — Chatwoot: updated_at > NOW - interval if params.UpdatedWithin != nil { - query = query.Where("updated_at > NOW() - INTERVAL '"+fmt.Sprintf("%d", *params.UpdatedWithin)+" seconds'") + query = query.Where("updated_at > NOW() - INTERVAL '" + fmt.Sprintf("%d", *params.UpdatedWithin) + " seconds'") } // 9. Filter by query (q) — 1:1 Chatwoot: filter_by_query @@ -674,10 +708,10 @@ func (s *ConversationService) Filter(ctx context.Context, accountID uint, userID return &FilterResult{ Conversations: conversations, Count: FilterCountMeta{ - MineCount: mineCount, - AssignedCount: allCount - unassignedCount, + MineCount: mineCount, + AssignedCount: allCount - unassignedCount, UnassignedCount: unassignedCount, - AllCount: allCount, + AllCount: allCount, }, }, nil } @@ -709,6 +743,7 @@ func (s *ConversationService) UpdatePriority(ctx context.Context, accountID, id // Dispatch EventConversationPriorityUpdated s.dispatchConversationEvent(ctx, channel.EventConversationPriorityUpdated, conversation) + s.indexConversation(ctx, conversation) return conversation, nil } @@ -749,6 +784,7 @@ func (s *ConversationService) MarkUnread(ctx context.Context, accountID, id uint if err != nil { return nil, err } + s.indexConversation(ctx, conversation) return conversation, nil } @@ -791,6 +827,7 @@ func (s *ConversationService) UpdateCustomAttributes(ctx context.Context, accoun if err != nil { return nil, err } + s.indexConversation(ctx, conversation) return conversation, nil } @@ -955,5 +992,6 @@ func (s *ConversationService) AssignTeam(ctx context.Context, accountID, convers } s.dispatchConversationEvent(ctx, channel.EventConversationAssigned, conversation) + s.indexConversation(ctx, conversation) return conversation, nil -} \ No newline at end of file +} diff --git a/internal/service/message_service.go b/internal/service/message_service.go index 9d386848..c5028bfb 100644 --- a/internal/service/message_service.go +++ b/internal/service/message_service.go @@ -16,9 +16,10 @@ import ( // MessageService implements business logic for Message operations. // Reference: Chatwoot app/controllers/api/v1/messages_controller.rb type MessageService struct { - repo *repository.MessageRepo - dispatcher *channel.Dispatcher - llmProvider llm.Provider + repo *repository.MessageRepo + dispatcher *channel.Dispatcher + searchIndexer SearchIndexer + llmProvider llm.Provider } // NewMessageService creates a new Message service. @@ -26,6 +27,22 @@ func NewMessageService(repo *repository.MessageRepo, dispatcher *channel.Dispatc return &MessageService{repo: repo, dispatcher: dispatcher, llmProvider: llmProvider} } +func (s *MessageService) SetSearchIndexer(indexer SearchIndexer) { + s.searchIndexer = indexer +} + +func (s *MessageService) indexMessage(ctx context.Context, message *model.Message) { + if s.searchIndexer != nil { + logSearchIndexError("message", message.ID, s.searchIndexer.IndexMessage(ctx, message)) + } +} + +func (s *MessageService) deleteMessageIndex(ctx context.Context, accountID uint, id uint) { + if s.searchIndexer != nil { + logSearchIndexError("message", id, s.searchIndexer.DeleteMessage(ctx, accountID, id)) + } +} + // dispatchMessageEvent is a helper to build and dispatch a message event. func (s *MessageService) dispatchMessageEvent(ctx context.Context, eventType channel.EventType, message *model.Message) { event := channel.NewChannelEvent(eventType, channel.ChannelAPI, message.AccountID, message.InboxID) @@ -113,6 +130,7 @@ func (s *MessageService) Create(ctx context.Context, accountID uint, userID uint // Dispatch EventMessageCreated s.dispatchMessageEvent(ctx, channel.EventMessageCreated, message) + s.indexMessage(ctx, message) // Dispatch additional event based on message type if req.MessageType == "incoming" { @@ -150,6 +168,7 @@ func (s *MessageService) Update(ctx context.Context, accountID, id uint, req Upd // Dispatch EventMessageUpdated s.dispatchMessageEvent(ctx, channel.EventMessageUpdated, message) + s.indexMessage(ctx, message) return message, nil } @@ -167,6 +186,7 @@ func (s *MessageService) Delete(ctx context.Context, accountID, id uint) error { // Dispatch EventMessageDeleted s.dispatchMessageEvent(ctx, channel.EventMessageDeleted, message) + s.deleteMessageIndex(ctx, accountID, message.ID) return nil } @@ -196,6 +216,7 @@ func (s *MessageService) UpdateStatus(ctx context.Context, id uint, status strin if err := s.dispatcher.Dispatch(ctx, event); err != nil { applogger.L().Errorf("failed to dispatch event %s for message %d: %v", channel.EventMessageStatusUpdated, message.ID, err) } + s.indexMessage(ctx, message) return message, nil } @@ -226,6 +247,7 @@ func (s *MessageService) Retry(ctx context.Context, accountID, id uint) (*model. if err := s.dispatcher.Dispatch(ctx, event); err != nil { applogger.L().Errorf("failed to dispatch retry event for message %d: %v", message.ID, err) } + s.indexMessage(ctx, message) return message, nil } @@ -310,4 +332,4 @@ func (s *MessageService) ListAttachments(ctx context.Context, accountID, convers } return attachments[start:end], total, nil -} \ No newline at end of file +} diff --git a/internal/service/search_indexer.go b/internal/service/search_indexer.go new file mode 100644 index 00000000..8dda0a25 --- /dev/null +++ b/internal/service/search_indexer.go @@ -0,0 +1,29 @@ +package service + +import ( + "context" + + "github.com/gochat/gochat/internal/model" + applogger "github.com/gochat/gochat/pkg/logger" +) + +// SearchIndexer is the service-layer boundary for search synchronization. +// Implementations may index synchronously or enqueue durable jobs later. +type SearchIndexer interface { + IndexConversation(ctx context.Context, conversation *model.Conversation) error + DeleteConversation(ctx context.Context, accountID uint, id uint) error + IndexMessage(ctx context.Context, message *model.Message) error + DeleteMessage(ctx context.Context, accountID uint, id uint) error + IndexContact(ctx context.Context, contact *model.Contact) error + DeleteContact(ctx context.Context, accountID uint, id uint) error + IndexCompany(ctx context.Context, company *model.Company) error + DeleteCompany(ctx context.Context, accountID uint, id uint) error + IndexArticle(ctx context.Context, article *model.Article) error + DeleteArticle(ctx context.Context, accountID uint, id uint) error +} + +func logSearchIndexError(entity string, id uint, err error) { + if err != nil { + applogger.L().Warnf("search index sync failed for %s %d: %v", entity, id, err) + } +} diff --git a/internal/service/search_indexer_hooks_test.go b/internal/service/search_indexer_hooks_test.go new file mode 100644 index 00000000..f095ce16 --- /dev/null +++ b/internal/service/search_indexer_hooks_test.go @@ -0,0 +1,155 @@ +package service + +import ( + "context" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/gochat/gochat/internal/model" + "github.com/gochat/gochat/internal/repository" +) + +type mockServiceSearchIndexer struct { + indexed []string + deleted []string +} + +func (m *mockServiceSearchIndexer) IndexConversation(ctx context.Context, conversation *model.Conversation) error { + m.indexed = append(m.indexed, "conversation") + return nil +} + +func (m *mockServiceSearchIndexer) DeleteConversation(ctx context.Context, accountID uint, id uint) error { + m.deleted = append(m.deleted, "conversation") + return nil +} + +func (m *mockServiceSearchIndexer) IndexMessage(ctx context.Context, message *model.Message) error { + m.indexed = append(m.indexed, "message") + return nil +} + +func (m *mockServiceSearchIndexer) DeleteMessage(ctx context.Context, accountID uint, id uint) error { + m.deleted = append(m.deleted, "message") + return nil +} + +func (m *mockServiceSearchIndexer) IndexContact(ctx context.Context, contact *model.Contact) error { + m.indexed = append(m.indexed, "contact") + return nil +} + +func (m *mockServiceSearchIndexer) DeleteContact(ctx context.Context, accountID uint, id uint) error { + m.deleted = append(m.deleted, "contact") + return nil +} + +func (m *mockServiceSearchIndexer) IndexCompany(ctx context.Context, company *model.Company) error { + m.indexed = append(m.indexed, "company") + return nil +} + +func (m *mockServiceSearchIndexer) DeleteCompany(ctx context.Context, accountID uint, id uint) error { + m.deleted = append(m.deleted, "company") + return nil +} + +func (m *mockServiceSearchIndexer) IndexArticle(ctx context.Context, article *model.Article) error { + m.indexed = append(m.indexed, "article") + return nil +} + +func (m *mockServiceSearchIndexer) DeleteArticle(ctx context.Context, accountID uint, id uint) error { + m.deleted = append(m.deleted, "article") + return nil +} + +func TestConversationService_SearchIndexHooks(t *testing.T) { + svc, db := setupConversationService(t) + indexer := &mockServiceSearchIndexer{} + svc.SetSearchIndexer(indexer) + + account := createConversationServiceTestAccount(t, db) + inbox := createConversationServiceTestInbox(t, db, account.ID) + contact := createConversationServiceTestContact(t, db, account.ID) + + conversation, err := svc.Create(context.Background(), account.ID, CreateConversationRequest{InboxID: inbox.ID, ContactID: contact.ID}) + require.NoError(t, err) + _, err = svc.UpdatePriority(context.Background(), account.ID, conversation.ID, "high") + require.NoError(t, err) + require.NoError(t, svc.Delete(context.Background(), account.ID, conversation.ID)) + + assert.Equal(t, []string{"conversation", "conversation"}, indexer.indexed) + assert.Equal(t, []string{"conversation"}, indexer.deleted) +} + +func TestMessageService_SearchIndexHooks(t *testing.T) { + db, _, _, svc := setupMessageServiceWithDefaultLLM(t) + indexer := &mockServiceSearchIndexer{} + svc.SetSearchIndexer(indexer) + + account := createTestAccount(t, db) + user := createTestUser(t, db, account.ID) + inbox := createTestInbox(t, db, account.ID, "web_widget") + contact := createTestContact(t, db, account.ID) + conversation := createTestConversation(t, db, account.ID, inbox.ID, contact.ID) + + message, err := svc.Create(context.Background(), account.ID, user.ID, CreateMessageRequest{ConversationID: conversation.ID, Content: "hello", MessageType: "outgoing"}) + require.NoError(t, err) + _, err = svc.Update(context.Background(), account.ID, message.ID, UpdateMessageRequest{Content: "updated"}) + require.NoError(t, err) + require.NoError(t, svc.Delete(context.Background(), account.ID, message.ID)) + + assert.Equal(t, []string{"message", "message"}, indexer.indexed) + assert.Equal(t, []string{"message"}, indexer.deleted) +} + +func TestContactService_SearchIndexHooks(t *testing.T) { + db := setupServiceTestDB(t) + repo := NewContactService(repository.NewContactRepo(db), nil, nil) + indexer := &mockServiceSearchIndexer{} + repo.SetSearchIndexer(indexer) + + account := createTestAccount(t, db) + contact, err := repo.Create(context.Background(), account.ID, CreateContactRequest{Name: "Ada", Email: "ada@example.com"}) + require.NoError(t, err) + _, err = repo.Update(context.Background(), account.ID, contact.ID, UpdateContactRequest{Name: "Ada Lovelace"}) + require.NoError(t, err) + require.NoError(t, repo.Delete(context.Background(), account.ID, contact.ID)) + + assert.Equal(t, []string{"contact", "contact"}, indexer.indexed) + assert.Equal(t, []string{"contact"}, indexer.deleted) +} + +func TestCompanyService_SearchIndexHooks(t *testing.T) { + db, _, _, _, svc := setupCompanyServiceTest(t) + indexer := &mockServiceSearchIndexer{} + svc.SetSearchIndexer(indexer) + + account := createTestAccount(t, db) + company, err := svc.Create(context.Background(), account.ID, &CreateCompanyRequest{Name: "Acme"}) + require.NoError(t, err) + _, err = svc.Update(context.Background(), company.ID, account.ID, &UpdateCompanyRequest{Name: "Acme Inc"}) + require.NoError(t, err) + require.NoError(t, svc.Delete(context.Background(), company.ID, account.ID)) + + assert.Equal(t, []string{"company", "company"}, indexer.indexed) + assert.Equal(t, []string{"company"}, indexer.deleted) +} + +func TestArticleService_SearchIndexHooks(t *testing.T) { + _, _, svc := setupArticleService(t) + indexer := &mockServiceSearchIndexer{} + svc.SetSearchIndexer(indexer) + + article, err := svc.CreateWithAccount(context.Background(), 7, 3, 1, &CreateArticleRequest{Title: "Install", Slug: "install"}) + require.NoError(t, err) + _, err = svc.Update(context.Background(), article.ID, &UpdateArticleRequest{Title: "Install GoChat"}) + require.NoError(t, err) + require.NoError(t, svc.Delete(context.Background(), article.ID)) + + assert.Equal(t, []string{"article", "article"}, indexer.indexed) + assert.Equal(t, []string{"article"}, indexer.deleted) +}