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

feat(search): index conversation maintenance

Browse Source
This commit is contained in:
rogee
2026-06-07 14:40:13 +08:00
parent 4ba1e6e847
commit 140b17ed72
5 changed files with 194 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/app/bootstrap.go
+1 -1
View File
@@ -679,7 +679,7 @@ func Bootstrap(env string) (*App, error) {
}
searchService := search.NewSearchServiceWithEngine(searchEngine, searchRepo)
searchIndexer := service.NewDurableSearchIndexer(db, workerPool, searchService)
service.RegisterBulkActionSearchIndexer(workerPool, db, searchIndexer)
service.RegisterConversationMaintenanceSearchIndexer(workerPool, db, searchIndexer)
conversationService.SetSearchIndexer(searchIndexer)
messageService.SetSearchIndexer(searchIndexer)
contactService.SetSearchIndexer(searchIndexer)
internal/service/conversation_maintenance_worker.go
+100 -12
View File
@@ -94,14 +94,21 @@ func RegisterConversationMaintenanceJobs(wp *worker.WorkerPool, db *gorm.DB) {
}
func RegisterContactBulkActionSearchIndexer(wp *worker.WorkerPool, db *gorm.DB, indexer SearchIndexer) {
RegisterBulkActionSearchIndexer(wp, db, indexer)
RegisterConversationMaintenanceSearchIndexer(wp, db, indexer)
}
func RegisterBulkActionSearchIndexer(wp *worker.WorkerPool, db *gorm.DB, indexer SearchIndexer) {
RegisterConversationMaintenanceSearchIndexer(wp, db, indexer)
}
func RegisterConversationMaintenanceSearchIndexer(wp *worker.WorkerPool, db *gorm.DB, indexer SearchIndexer) {
if wp == nil || db == nil {
return
}
runner := &conversationMaintenanceRunner{wp: wp, db: db, now: time.Now, searchIndexer: indexer}
wp.Register(TaskTypeConversationReopenSnoozed, runner.performReopenSnoozed)
wp.Register(TaskTypeConversationResolutionForAccount, runner.performResolutionForAccount)
wp.Register(TaskTypeConversationUpdateMessageStatus, runner.performUpdateMessageStatus)
wp.Register(TaskTypeConversationBulkAction, runner.performConversationBulkAction)
wp.Register(TaskTypeContactBulkAction, runner.performContactBulkAction)
}
@@ -254,15 +261,27 @@ func (r *conversationMaintenanceRunner) performReopenSnoozed(ctx context.Context
now := r.now()
nowUnix := now.Unix()
lookbackUnix := now.Add(-scheduledItemsLookback).Unix()
var conversationIDs []uint
if err := r.db.WithContext(ctx).Model(&model.Conversation{}).
Where("status = ?", string(model.ConversationStatusSnoozed)).
Where("snoozed_until BETWEEN ? AND ?", lookbackUnix, nowUnix).
Pluck("id", &conversationIDs).Error; err != nil {
return err
}
if len(conversationIDs) == 0 {
return nil
}
updates := map[string]any{
"status": string(model.ConversationStatusOpen),
"snoozed_until": nil,
"resumed_at": now,
}
return r.db.WithContext(ctx).Model(&model.Conversation{}).
Where("status = ?", string(model.ConversationStatusSnoozed)).
Where("snoozed_until BETWEEN ? AND ?", lookbackUnix, nowUnix).
Updates(updates).Error
if err := r.db.WithContext(ctx).Model(&model.Conversation{}).
Where("id IN ?", conversationIDs).
Updates(updates).Error; err != nil {
return err
}
return r.indexConversationSearchDocumentsByIDs(ctx, conversationIDs)
}
func (r *conversationMaintenanceRunner) performResolutionScheduler(ctx context.Context, job *model.BackgroundJob) error {
@@ -297,15 +316,27 @@ func (r *conversationMaintenanceRunner) performResolutionForAccount(ctx context.
}
cutoff := r.now().Add(-time.Duration(account.AutoResolveDuration) * time.Minute).Unix()
now := r.now()
var conversationIDs []uint
if err := r.db.WithContext(ctx).Model(&model.Conversation{}).
Where("account_id = ? AND status = ? AND contact_id <> 0", account.ID, string(model.ConversationStatusOpen)).
Where("last_activity_at IS NOT NULL AND last_activity_at < ?", cutoff).
Limit(conversationResolutionLimit).
Pluck("id", &conversationIDs).Error; err != nil {
return err
}
if len(conversationIDs) == 0 {
return nil
}
updates := map[string]any{
"status": string(model.ConversationStatusResolved),
"resolved_at": now,
}
return r.db.WithContext(ctx).Model(&model.Conversation{}).
Where("account_id = ? AND status = ? AND contact_id <> 0", account.ID, string(model.ConversationStatusOpen)).
Where("last_activity_at IS NOT NULL AND last_activity_at < ?", cutoff).
Limit(conversationResolutionLimit).
Updates(updates).Error
if err := r.db.WithContext(ctx).Model(&model.Conversation{}).
Where("id IN ?", conversationIDs).
Updates(updates).Error; err != nil {
return err
}
return r.indexConversationSearchDocumentsByIDs(ctx, conversationIDs)
}
func (r *conversationMaintenanceRunner) performUpdateMessageStatus(ctx context.Context, job *model.BackgroundJob) error {
@@ -328,12 +359,27 @@ func (r *conversationMaintenanceRunner) performUpdateMessageStatus(ctx context.C
return fmt.Errorf("load conversation %d for message status update: %w", payload.ConversationID, err)
}
return r.db.WithContext(ctx).Model(&model.Message{}).
var messageIDs []uint
if err := r.db.WithContext(ctx).Model(&model.Message{}).
Where("conversation_id = ?", conversation.ID).
Where("status IN ?", []string{string(model.MessageStatusSent), string(model.MessageStatusDelivered)}).
Where("message_type <> ?", "incoming").
Where("created_at <= ?", time.Unix(payload.Timestamp, 0).UTC()).
Update("status", payload.Status).Error
Pluck("id", &messageIDs).Error; err != nil {
return err
}
if len(messageIDs) == 0 {
return nil
}
if err := r.db.WithContext(ctx).Model(&model.Message{}).
Where("id IN ?", messageIDs).
Update("status", payload.Status).Error; err != nil {
return err
}
if err := r.indexMessageSearchDocuments(ctx, conversation.AccountID, messageIDs); err != nil {
return err
}
return r.indexConversationSearchDocumentsByIDs(ctx, []uint{conversation.ID})
}
func validConversationMessageStatus(status string) bool {
@@ -428,6 +474,48 @@ func (r *conversationMaintenanceRunner) indexConversationSearchDocuments(ctx con
return nil
}
func (r *conversationMaintenanceRunner) indexConversationSearchDocumentsByIDs(ctx context.Context, conversationIDs []uint) error {
if r.searchIndexer == nil || len(conversationIDs) == 0 {
return nil
}
var conversations []model.Conversation
if err := r.db.WithContext(ctx).
Preload("Contact").
Preload("Inbox").
Preload("Assignee").
Preload("Messages", func(db *gorm.DB) *gorm.DB { return db.Order("created_at ASC, id ASC") }).
Where("id IN ?", conversationIDs).
Find(&conversations).Error; err != nil {
return err
}
for i := range conversations {
if err := r.searchIndexer.IndexConversation(ctx, &conversations[i]); err != nil {
return err
}
}
return nil
}
func (r *conversationMaintenanceRunner) indexMessageSearchDocuments(ctx context.Context, accountID uint, messageIDs []uint) error {
if r.searchIndexer == nil || len(messageIDs) == 0 {
return nil
}
var messages []model.Message
if err := r.db.WithContext(ctx).
Preload("Conversation").
Preload("Attachments", func(db *gorm.DB) *gorm.DB { return db.Order("id ASC") }).
Where("account_id = ? AND id IN ?", accountID, messageIDs).
Find(&messages).Error; err != nil {
return err
}
for i := range messages {
if err := r.searchIndexer.IndexMessage(ctx, &messages[i]); err != nil {
return err
}
}
return nil
}
func (r *conversationMaintenanceRunner) performContactBulkAction(ctx context.Context, job *model.BackgroundJob) error {
var payload contactBulkActionJob
if err := json.Unmarshal(job.Payload, &payload); err != nil {
internal/service/conversation_maintenance_worker_test.go
+79
View File
@@ -158,6 +158,46 @@ func TestConversationMaintenanceJobsProcessCampaignSnoozeAndResolution(t *testin
}
}
func TestConversationMaintenanceJobsStatusMaintenanceQueuesSearchIndex(t *testing.T) {
now := time.Date(2026, 6, 5, 20, 30, 0, 0, time.UTC)
db := setupServiceTestDB(t)
wp := worker.NewWorkerPoolWithOptions(db, worker.WithNow(func() time.Time { return now }))
registerConversationMaintenanceJobsWithNow(wp, db, func() time.Time { return now })
delegate := &recordingDurableSearchIndexer{}
searchIndexer := NewDurableSearchIndexer(db, wp, delegate)
RegisterConversationMaintenanceSearchIndexer(wp, db, searchIndexer)
account := createTestAccount(t, db)
account.AutoResolveDuration = 30
if err := db.Save(account).Error; err != nil {
t.Fatalf("save auto resolve account: %v", err)
}
inbox := createTestInbox(t, db, account.ID, "web_widget")
contact := createTestContact(t, db, account.ID)
snoozed := createTestConversation(t, db, account.ID, inbox.ID, contact.ID)
if err := db.Model(snoozed).Updates(map[string]any{"status": string(model.ConversationStatusSnoozed), "snoozed_until": now.Add(-time.Minute).Unix()}).Error; err != nil {
t.Fatalf("snooze conversation: %v", err)
}
oldOpen := createTestConversation(t, db, account.ID, inbox.ID, contact.ID)
if err := db.Model(oldOpen).Update("last_activity_at", now.Add(-45*time.Minute).Unix()).Error; err != nil {
t.Fatalf("set old activity: %v", err)
}
if _, err := wp.Enqueue(context.Background(), TaskTypeConversationReopenSnoozed, nil, worker.WithQueue("low")); err != nil {
t.Fatalf("enqueue reopen: %v", err)
}
processRequiredJob(t, wp, "reopen with index")
processRequiredJob(t, wp, "reopen search index")
assertIndexedConversationStatus(t, delegate, snoozed.ID, string(model.ConversationStatusOpen))
if _, err := wp.Enqueue(context.Background(), TaskTypeConversationResolutionForAccount, conversationResolutionJob{AccountID: account.ID}, worker.WithQueue("low")); err != nil {
t.Fatalf("enqueue resolution: %v", err)
}
processRequiredJob(t, wp, "resolution with index")
processRequiredJob(t, wp, "resolution search index")
assertIndexedConversationStatus(t, delegate, oldOpen.ID, string(model.ConversationStatusResolved))
}
func TestConversationMaintenanceJobsRetryMissingResolutionAccount(t *testing.T) {
now := time.Date(2026, 6, 5, 21, 0, 0, 0, time.UTC)
db := setupServiceTestDB(t)
@@ -219,6 +259,35 @@ func TestConversationMaintenanceJobsUpdateMessageStatus(t *testing.T) {
assertMessageStatus(t, db, afterCutoff.ID, string(model.MessageStatusSent))
}
func TestConversationMaintenanceJobsUpdateMessageStatusQueuesSearchIndex(t *testing.T) {
now := time.Date(2026, 6, 5, 22, 15, 0, 0, time.UTC)
db := setupServiceTestDB(t)
wp := worker.NewWorkerPoolWithOptions(db, worker.WithNow(func() time.Time { return now }))
registerConversationMaintenanceJobsWithNow(wp, db, func() time.Time { return now })
delegate := &recordingDurableSearchIndexer{}
searchIndexer := NewDurableSearchIndexer(db, wp, delegate)
RegisterConversationMaintenanceSearchIndexer(wp, db, searchIndexer)
account := createTestAccount(t, db)
inbox := createTestInbox(t, db, account.ID, "web_widget")
contact := createTestContact(t, db, account.ID)
conversation := createTestConversation(t, db, account.ID, inbox.ID, contact.ID)
cutoff := now.Add(-time.Minute)
message := createConversationMaintenanceMessage(t, db, account.ID, inbox.ID, conversation.ID, "outgoing", string(model.MessageStatusSent), cutoff.Add(-time.Minute))
if _, err := EnqueueConversationMessageStatusUpdate(context.Background(), wp, conversation.ID, cutoff, string(model.MessageStatusRead)); err != nil {
t.Fatalf("enqueue message status update: %v", err)
}
processRequiredJob(t, wp, "message status with index")
processRequiredJob(t, wp, "message search index")
processRequiredJob(t, wp, "conversation search index")
if len(delegate.indexedMessages) != 1 || delegate.indexedMessages[0].ID != message.ID || delegate.indexedMessages[0].Status != string(model.MessageStatusRead) {
t.Fatalf("expected indexed read message, got %#v", delegate.indexedMessages)
}
assertIndexedConversationStatus(t, delegate, conversation.ID, string(model.ConversationStatusOpen))
}
func TestConversationMaintenanceJobsIgnoreInvalidMessageStatus(t *testing.T) {
now := time.Date(2026, 6, 5, 22, 30, 0, 0, time.UTC)
db := setupServiceTestDB(t)
@@ -527,6 +596,16 @@ func assertContactHasLabel(t *testing.T, db *gorm.DB, accountID, contactID uint,
}
}
func assertIndexedConversationStatus(t *testing.T, indexer *recordingDurableSearchIndexer, conversationID uint, status string) {
t.Helper()
for _, conversation := range indexer.indexedConversations {
if conversation.ID == conversationID && conversation.Status == status {
return
}
}
t.Fatalf("expected indexed conversation %d with status %s, got %#v", conversationID, status, indexer.indexedConversations)
}
func assertMessageStatus(t *testing.T, db *gorm.DB, messageID uint, want string) {
t.Helper()
var message model.Message
internal/service/search_indexer_worker_test.go
+4
View File
@@ -12,6 +12,7 @@ import (
type recordingDurableSearchIndexer struct {
indexedConversations []model.Conversation
indexedMessages []model.Message
indexedContacts []uint
indexedContactLabels map[uint][]string
indexedArticles []model.Article
@@ -31,6 +32,9 @@ func (r *recordingDurableSearchIndexer) DeleteConversation(ctx context.Context,
}
func (r *recordingDurableSearchIndexer) IndexMessage(ctx context.Context, message *model.Message) error {
if message != nil {
r.indexedMessages = append(r.indexedMessages, *message)
}
return r.err
}