rogee/gochat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(search): sync entity writes to indexer

Browse Source
This commit is contained in:
rogee
2026-06-04 20:06:21 +08:00
parent 3fc4859e20
commit dbc5a1a28f
11 changed files with 468 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md
+2 -1
View File
@@ -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.
internal/app/bootstrap.go
+5
View File
@@ -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)
internal/handler/api/v1/article_handler.go
+7 -2
View File
@@ -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})
}
}
internal/search/search_service.go
+55
View File
@@ -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
internal/service/article_service.go
+81 -39
View File
@@ -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)
}
}
}
internal/service/company_service.go
+25 -2
View File
@@ -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
}
}
internal/service/contact_service.go
+24 -1
View File
@@ -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.
internal/service/conversation_service.go
+59 -21
View File
@@ -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
}
}
internal/service/message_service.go
+26 -4
View File
@@ -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
}
}
internal/service/search_indexer.go
+29
View File
@@ -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)
}
}
internal/service/search_indexer_hooks_test.go
+155
View File
@@ -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)
}