feat(search): queue index updates durably
This commit is contained in:
@@ -641,13 +641,14 @@ 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)
|
||||
searchIndexer := service.NewDurableSearchIndexer(db, workerPool, searchService)
|
||||
conversationService.SetSearchIndexer(searchIndexer)
|
||||
messageService.SetSearchIndexer(searchIndexer)
|
||||
contactService.SetSearchIndexer(searchIndexer)
|
||||
contactService.SetSearchReader(searchService)
|
||||
companyService.SetSearchIndexer(searchService)
|
||||
companyService.SetSearchIndexer(searchIndexer)
|
||||
companyService.SetSearchReader(searchService)
|
||||
articleService.SetSearchIndexer(searchService)
|
||||
articleService.SetSearchIndexer(searchIndexer)
|
||||
|
||||
// Custom attribute definition + custom filter + custom attribute value services
|
||||
customAttributeDefinitionService := service.NewCustomAttributeDefinitionService(customAttributeDefinitionRepo)
|
||||
|
||||
@@ -0,0 +1,204 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
"github.com/gochat/gochat/internal/model"
|
||||
"github.com/gochat/gochat/internal/worker"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
const TaskTypeSearchIndex = "search:index"
|
||||
|
||||
type searchIndexJob struct {
|
||||
Operation string `json:"operation"`
|
||||
Entity string `json:"entity"`
|
||||
AccountID uint `json:"account_id"`
|
||||
ID uint `json:"id"`
|
||||
}
|
||||
|
||||
// DurableSearchIndexer defers Meilisearch write-side synchronization to the
|
||||
// durable worker while keeping read paths pointed at the real SearchService.
|
||||
type DurableSearchIndexer struct {
|
||||
db *gorm.DB
|
||||
worker *worker.WorkerPool
|
||||
delegate SearchIndexer
|
||||
}
|
||||
|
||||
func NewDurableSearchIndexer(db *gorm.DB, wp *worker.WorkerPool, delegate SearchIndexer) *DurableSearchIndexer {
|
||||
indexer := &DurableSearchIndexer{db: db, worker: wp, delegate: delegate}
|
||||
if wp != nil {
|
||||
wp.Register(TaskTypeSearchIndex, indexer.perform)
|
||||
}
|
||||
return indexer
|
||||
}
|
||||
|
||||
func (i *DurableSearchIndexer) IndexConversation(ctx context.Context, conversation *model.Conversation) error {
|
||||
if conversation == nil {
|
||||
return nil
|
||||
}
|
||||
return i.enqueueOrIndex(ctx, "conversation", conversation.AccountID, conversation.ID, func() error {
|
||||
return i.delegate.IndexConversation(ctx, conversation)
|
||||
})
|
||||
}
|
||||
|
||||
func (i *DurableSearchIndexer) DeleteConversation(ctx context.Context, accountID uint, id uint) error {
|
||||
return i.enqueueOrDelete(ctx, "conversation", accountID, id, func() error { return i.delegate.DeleteConversation(ctx, accountID, id) })
|
||||
}
|
||||
|
||||
func (i *DurableSearchIndexer) IndexMessage(ctx context.Context, message *model.Message) error {
|
||||
if message == nil {
|
||||
return nil
|
||||
}
|
||||
return i.enqueueOrIndex(ctx, "message", message.AccountID, message.ID, func() error {
|
||||
return i.delegate.IndexMessage(ctx, message)
|
||||
})
|
||||
}
|
||||
|
||||
func (i *DurableSearchIndexer) DeleteMessage(ctx context.Context, accountID uint, id uint) error {
|
||||
return i.enqueueOrDelete(ctx, "message", accountID, id, func() error { return i.delegate.DeleteMessage(ctx, accountID, id) })
|
||||
}
|
||||
|
||||
func (i *DurableSearchIndexer) IndexContact(ctx context.Context, contact *model.Contact) error {
|
||||
if contact == nil {
|
||||
return nil
|
||||
}
|
||||
return i.enqueueOrIndex(ctx, "contact", contact.AccountID, contact.ID, func() error {
|
||||
return i.delegate.IndexContact(ctx, contact)
|
||||
})
|
||||
}
|
||||
|
||||
func (i *DurableSearchIndexer) DeleteContact(ctx context.Context, accountID uint, id uint) error {
|
||||
return i.enqueueOrDelete(ctx, "contact", accountID, id, func() error { return i.delegate.DeleteContact(ctx, accountID, id) })
|
||||
}
|
||||
|
||||
func (i *DurableSearchIndexer) IndexCompany(ctx context.Context, company *model.Company) error {
|
||||
if company == nil {
|
||||
return nil
|
||||
}
|
||||
return i.enqueueOrIndex(ctx, "company", company.AccountID, company.ID, func() error {
|
||||
return i.delegate.IndexCompany(ctx, company)
|
||||
})
|
||||
}
|
||||
|
||||
func (i *DurableSearchIndexer) DeleteCompany(ctx context.Context, accountID uint, id uint) error {
|
||||
return i.enqueueOrDelete(ctx, "company", accountID, id, func() error { return i.delegate.DeleteCompany(ctx, accountID, id) })
|
||||
}
|
||||
|
||||
func (i *DurableSearchIndexer) IndexArticle(ctx context.Context, article *model.Article) error {
|
||||
if article == nil {
|
||||
return nil
|
||||
}
|
||||
return i.enqueueOrIndex(ctx, "article", article.AccountID, article.ID, func() error {
|
||||
return i.delegate.IndexArticle(ctx, article)
|
||||
})
|
||||
}
|
||||
|
||||
func (i *DurableSearchIndexer) DeleteArticle(ctx context.Context, accountID uint, id uint) error {
|
||||
return i.enqueueOrDelete(ctx, "article", accountID, id, func() error { return i.delegate.DeleteArticle(ctx, accountID, id) })
|
||||
}
|
||||
|
||||
func (i *DurableSearchIndexer) enqueueOrIndex(ctx context.Context, entity string, accountID, id uint, fallback func() error) error {
|
||||
return i.enqueueOrRun(ctx, searchIndexJob{Operation: "index", Entity: entity, AccountID: accountID, ID: id}, fallback)
|
||||
}
|
||||
|
||||
func (i *DurableSearchIndexer) enqueueOrDelete(ctx context.Context, entity string, accountID, id uint, fallback func() error) error {
|
||||
return i.enqueueOrRun(ctx, searchIndexJob{Operation: "delete", Entity: entity, AccountID: accountID, ID: id}, fallback)
|
||||
}
|
||||
|
||||
func (i *DurableSearchIndexer) enqueueOrRun(ctx context.Context, payload searchIndexJob, fallback func() error) error {
|
||||
if i.delegate == nil {
|
||||
return nil
|
||||
}
|
||||
if i.worker == nil {
|
||||
return fallback()
|
||||
}
|
||||
_, err := i.worker.Enqueue(ctx, TaskTypeSearchIndex, payload, worker.WithQueue("search"), worker.WithMaxAttempts(3))
|
||||
return err
|
||||
}
|
||||
|
||||
func (i *DurableSearchIndexer) perform(ctx context.Context, job *model.BackgroundJob) error {
|
||||
if i.delegate == nil {
|
||||
return nil
|
||||
}
|
||||
var payload searchIndexJob
|
||||
if err := json.Unmarshal(job.Payload, &payload); err != nil {
|
||||
return fmt.Errorf("unmarshal search index job: %w", err)
|
||||
}
|
||||
if payload.ID == 0 || payload.Entity == "" || payload.Operation == "" {
|
||||
return fmt.Errorf("invalid search index job payload: %#v", payload)
|
||||
}
|
||||
switch payload.Operation {
|
||||
case "index":
|
||||
return i.performIndex(ctx, payload)
|
||||
case "delete":
|
||||
return i.performDelete(ctx, payload)
|
||||
default:
|
||||
return fmt.Errorf("unsupported search index operation %q", payload.Operation)
|
||||
}
|
||||
}
|
||||
|
||||
func (i *DurableSearchIndexer) performIndex(ctx context.Context, payload searchIndexJob) error {
|
||||
switch payload.Entity {
|
||||
case "conversation":
|
||||
var item model.Conversation
|
||||
if err := i.load(ctx, payload, &item); err != nil {
|
||||
return err
|
||||
}
|
||||
return i.delegate.IndexConversation(ctx, &item)
|
||||
case "message":
|
||||
var item model.Message
|
||||
if err := i.load(ctx, payload, &item); err != nil {
|
||||
return err
|
||||
}
|
||||
return i.delegate.IndexMessage(ctx, &item)
|
||||
case "contact":
|
||||
var item model.Contact
|
||||
if err := i.load(ctx, payload, &item); err != nil {
|
||||
return err
|
||||
}
|
||||
return i.delegate.IndexContact(ctx, &item)
|
||||
case "company":
|
||||
var item model.Company
|
||||
if err := i.load(ctx, payload, &item); err != nil {
|
||||
return err
|
||||
}
|
||||
return i.delegate.IndexCompany(ctx, &item)
|
||||
case "article":
|
||||
var item model.Article
|
||||
if err := i.load(ctx, payload, &item); err != nil {
|
||||
return err
|
||||
}
|
||||
return i.delegate.IndexArticle(ctx, &item)
|
||||
default:
|
||||
return fmt.Errorf("unsupported search index entity %q", payload.Entity)
|
||||
}
|
||||
}
|
||||
|
||||
func (i *DurableSearchIndexer) load(ctx context.Context, payload searchIndexJob, dest any) error {
|
||||
err := i.db.WithContext(ctx).Where("id = ? AND account_id = ?", payload.ID, payload.AccountID).First(dest).Error
|
||||
if errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
return i.performDelete(ctx, payload)
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
func (i *DurableSearchIndexer) performDelete(ctx context.Context, payload searchIndexJob) error {
|
||||
switch payload.Entity {
|
||||
case "conversation":
|
||||
return i.delegate.DeleteConversation(ctx, payload.AccountID, payload.ID)
|
||||
case "message":
|
||||
return i.delegate.DeleteMessage(ctx, payload.AccountID, payload.ID)
|
||||
case "contact":
|
||||
return i.delegate.DeleteContact(ctx, payload.AccountID, payload.ID)
|
||||
case "company":
|
||||
return i.delegate.DeleteCompany(ctx, payload.AccountID, payload.ID)
|
||||
case "article":
|
||||
return i.delegate.DeleteArticle(ctx, payload.AccountID, payload.ID)
|
||||
default:
|
||||
return fmt.Errorf("unsupported search index entity %q", payload.Entity)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,145 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/gochat/gochat/internal/model"
|
||||
"github.com/gochat/gochat/internal/worker"
|
||||
)
|
||||
|
||||
type recordingDurableSearchIndexer struct {
|
||||
indexedContacts []uint
|
||||
deletedContacts []uint
|
||||
err error
|
||||
}
|
||||
|
||||
func (r *recordingDurableSearchIndexer) IndexConversation(ctx context.Context, conversation *model.Conversation) error {
|
||||
return r.err
|
||||
}
|
||||
|
||||
func (r *recordingDurableSearchIndexer) DeleteConversation(ctx context.Context, accountID uint, id uint) error {
|
||||
return r.err
|
||||
}
|
||||
|
||||
func (r *recordingDurableSearchIndexer) IndexMessage(ctx context.Context, message *model.Message) error {
|
||||
return r.err
|
||||
}
|
||||
|
||||
func (r *recordingDurableSearchIndexer) DeleteMessage(ctx context.Context, accountID uint, id uint) error {
|
||||
return r.err
|
||||
}
|
||||
|
||||
func (r *recordingDurableSearchIndexer) IndexContact(ctx context.Context, contact *model.Contact) error {
|
||||
r.indexedContacts = append(r.indexedContacts, contact.ID)
|
||||
return r.err
|
||||
}
|
||||
|
||||
func (r *recordingDurableSearchIndexer) DeleteContact(ctx context.Context, accountID uint, id uint) error {
|
||||
r.deletedContacts = append(r.deletedContacts, id)
|
||||
return r.err
|
||||
}
|
||||
|
||||
func (r *recordingDurableSearchIndexer) IndexCompany(ctx context.Context, company *model.Company) error {
|
||||
return r.err
|
||||
}
|
||||
|
||||
func (r *recordingDurableSearchIndexer) DeleteCompany(ctx context.Context, accountID uint, id uint) error {
|
||||
return r.err
|
||||
}
|
||||
|
||||
func (r *recordingDurableSearchIndexer) IndexArticle(ctx context.Context, article *model.Article) error {
|
||||
return r.err
|
||||
}
|
||||
|
||||
func (r *recordingDurableSearchIndexer) DeleteArticle(ctx context.Context, accountID uint, id uint) error {
|
||||
return r.err
|
||||
}
|
||||
|
||||
func TestDurableSearchIndexerQueuesAndReplaysContactIndex(t *testing.T) {
|
||||
db := setupServiceTestDB(t)
|
||||
account := createTestAccount(t, db)
|
||||
contact := &model.Contact{AccountID: account.ID, Name: "Index Me", Email: "index@example.com"}
|
||||
if err := db.Create(contact).Error; err != nil {
|
||||
t.Fatalf("create contact: %v", err)
|
||||
}
|
||||
delegate := &recordingDurableSearchIndexer{}
|
||||
wp := worker.NewWorkerPoolWithOptions(db, worker.WithNow(func() time.Time { return time.Date(2026, 6, 5, 14, 0, 0, 0, time.UTC) }))
|
||||
indexer := NewDurableSearchIndexer(db, wp, delegate)
|
||||
|
||||
if err := indexer.IndexContact(context.Background(), contact); err != nil {
|
||||
t.Fatalf("queue contact index: %v", err)
|
||||
}
|
||||
if len(delegate.indexedContacts) != 0 {
|
||||
t.Fatalf("contact indexed synchronously before worker replay")
|
||||
}
|
||||
var count int64
|
||||
if err := db.Model(&model.BackgroundJob{}).Where("job_type = ? AND queue = ? AND status = ?", TaskTypeSearchIndex, "search", model.BackgroundJobStatusQueued).Count(&count).Error; err != nil {
|
||||
t.Fatalf("count search jobs: %v", err)
|
||||
}
|
||||
if count != 1 {
|
||||
t.Fatalf("expected one queued search job, got %d", count)
|
||||
}
|
||||
processed, err := wp.ProcessOne(context.Background())
|
||||
if err != nil || !processed {
|
||||
t.Fatalf("process search job: processed=%v err=%v", processed, err)
|
||||
}
|
||||
if len(delegate.indexedContacts) != 1 || delegate.indexedContacts[0] != contact.ID {
|
||||
t.Fatalf("expected durable contact index replay, got %#v", delegate.indexedContacts)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDurableSearchIndexerRetriesDelegateFailure(t *testing.T) {
|
||||
db := setupServiceTestDB(t)
|
||||
account := createTestAccount(t, db)
|
||||
contact := &model.Contact{AccountID: account.ID, Name: "Retry Me"}
|
||||
if err := db.Create(contact).Error; err != nil {
|
||||
t.Fatalf("create contact: %v", err)
|
||||
}
|
||||
delegate := &recordingDurableSearchIndexer{err: errors.New("meili unavailable")}
|
||||
wp := worker.NewWorkerPoolWithOptions(db, worker.WithNow(func() time.Time { return time.Date(2026, 6, 5, 14, 15, 0, 0, time.UTC) }), worker.WithBackoff(func(attempt int) time.Duration { return time.Minute }))
|
||||
indexer := NewDurableSearchIndexer(db, wp, delegate)
|
||||
|
||||
if err := indexer.IndexContact(context.Background(), contact); err != nil {
|
||||
t.Fatalf("queue contact index: %v", err)
|
||||
}
|
||||
processed, err := wp.ProcessOne(context.Background())
|
||||
if err == nil || !processed {
|
||||
t.Fatalf("expected retryable worker error, processed=%v err=%v", processed, err)
|
||||
}
|
||||
var job model.BackgroundJob
|
||||
if err := db.Where("job_type = ?", TaskTypeSearchIndex).First(&job).Error; err != nil {
|
||||
t.Fatalf("load search job: %v", err)
|
||||
}
|
||||
if job.Status != model.BackgroundJobStatusRetrying || job.LastError == "" {
|
||||
t.Fatalf("expected retrying search job with error, got status=%s last_error=%q", job.Status, job.LastError)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDurableSearchIndexerDeletesWhenIndexedRecordDisappears(t *testing.T) {
|
||||
db := setupServiceTestDB(t)
|
||||
account := createTestAccount(t, db)
|
||||
contact := &model.Contact{AccountID: account.ID, Name: "Delete Before Replay"}
|
||||
if err := db.Create(contact).Error; err != nil {
|
||||
t.Fatalf("create contact: %v", err)
|
||||
}
|
||||
delegate := &recordingDurableSearchIndexer{}
|
||||
wp := worker.NewWorkerPoolWithOptions(db, worker.WithNow(func() time.Time { return time.Date(2026, 6, 5, 14, 30, 0, 0, time.UTC) }))
|
||||
indexer := NewDurableSearchIndexer(db, wp, delegate)
|
||||
|
||||
if err := indexer.IndexContact(context.Background(), contact); err != nil {
|
||||
t.Fatalf("queue contact index: %v", err)
|
||||
}
|
||||
if err := db.Delete(contact).Error; err != nil {
|
||||
t.Fatalf("delete contact before replay: %v", err)
|
||||
}
|
||||
processed, err := wp.ProcessOne(context.Background())
|
||||
if err != nil || !processed {
|
||||
t.Fatalf("process search job: processed=%v err=%v", processed, err)
|
||||
}
|
||||
if len(delegate.deletedContacts) != 1 || delegate.deletedContacts[0] != contact.ID {
|
||||
t.Fatalf("expected missing record to delete search document, got %#v", delegate.deletedContacts)
|
||||
}
|
||||
}
|
||||
@@ -71,6 +71,7 @@ func setupServiceTestDB(t *testing.T) *gorm.DB {
|
||||
&model.ContactExport{},
|
||||
&model.DataImport{},
|
||||
&model.Notification{},
|
||||
&model.BackgroundJob{},
|
||||
); err != nil {
|
||||
t.Fatalf("failed to auto-migrate models: %v", err)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user