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

feat(search): index csat surveys

Browse Source
This commit is contained in:
rogee
2026-06-07 15:36:51 +08:00
parent 8eabadc3fc
commit 0aff792b40
6 changed files with 139 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/app/bootstrap.go
+2 -1
View File
@@ -681,6 +681,7 @@ func Bootstrap(env string) (*App, error) {
articleService.SetSearchIndexer(searchIndexer)
automationRuleService.SetSearchIndexer(searchIndexer)
macroService.SetSearchIndexer(searchIndexer)
csatSurveyService.SetSearchIndexer(searchIndexer)
botRuleService.SetSearchIndexer(searchIndexer)
facebookWebhookHandler.WithSearchIndexer(searchIndexer)
telegramWebhookHandler.WithSearchIndexer(searchIndexer)
@@ -693,7 +694,7 @@ func Bootstrap(env string) (*App, error) {
channelDispatcher.Register(automation.NewAgentBotRuleListenerWithSearchIndexer(&dbProvider{db: db}, searchIndexer))
automation.RegisterAutomationRuleListenerWithWorkerAndSearchIndexer(channelDispatcher, &dbProvider{db: db}, workerPool, searchIndexer)
// M6: Register CSAT survey listener on the dispatcher for conversation.resolved events
channelDispatcher.Register(automation.NewCsatSurveyListenerWithWorker(&dbProvider{db: db}, workerPool))
channelDispatcher.Register(automation.NewCsatSurveyListenerWithWorkerAndSearchIndexer(&dbProvider{db: db}, workerPool, searchIndexer))
// Custom attribute definition + custom filter + custom attribute value services
customAttributeDefinitionService := service.NewCustomAttributeDefinitionService(customAttributeDefinitionRepo)
internal/automation/csat_survey_listener.go
+23 -4
View File
@@ -17,8 +17,9 @@ const TaskTypeCsatSurveySend = "csat:survey_send"
// CsatSurveyListener sends CSAT surveys when conversations are resolved.
// Reference: Chatwoot AutomationRuleListener triggers SendCSATSurvey action on conversation.resolved
type CsatSurveyListener struct {
db DBProvider
worker *worker.WorkerPool
db DBProvider
worker *worker.WorkerPool
searchIndexer CsatSurveySearchIndexer
}
// NewCsatSurveyListener creates a new CsatSurveyListener.
@@ -39,6 +40,16 @@ func (l *CsatSurveyListener) SetWorkerPool(wp *worker.WorkerPool) {
}
}
func (l *CsatSurveyListener) SetSearchIndexer(indexer CsatSurveySearchIndexer) {
l.searchIndexer = indexer
}
func NewCsatSurveyListenerWithWorkerAndSearchIndexer(db DBProvider, wp *worker.WorkerPool, indexer CsatSurveySearchIndexer) *CsatSurveyListener {
listener := NewCsatSurveyListenerWithWorker(db, wp)
listener.SetSearchIndexer(indexer)
return listener
}
// Name returns the listener name for registration.
func (l *CsatSurveyListener) Name() string {
return "csat_survey_listener"
@@ -66,7 +77,9 @@ func (l *CsatSurveyListener) OnEvent(ctx context.Context, event *channel.Channel
}
func (l *CsatSurveyListener) sendSurvey(ctx context.Context, conversationID uint) error {
message, err := NewCsatSurveyService(l.db).SendSurveyForConversationID(ctx, conversationID)
service := NewCsatSurveyService(l.db)
service.SetSearchIndexer(l.searchIndexer)
message, err := service.SendSurveyForConversationID(ctx, conversationID)
if err != nil {
applogger.L().Errorf("csat_survey_listener: failed to send CSAT survey for conversation %d: %v", conversationID, err)
return err
@@ -117,7 +130,13 @@ func (l *CsatSurveyListener) onMessageUpdated(ctx context.Context, event *channe
return nil
}
_, err := ApplyCsatSubmission(ctx, l.db.DB(), &message, &conversation, submittedValues)
return err
if err != nil {
return err
}
service := NewCsatSurveyService(l.db)
service.SetSearchIndexer(l.searchIndexer)
service.indexCsatSearchDocuments(ctx, &message, &conversation)
return nil
}
func eventUint(event *channel.ChannelEvent, key string) (uint, bool) {
internal/automation/csat_survey_listener_test.go
+24
View File
@@ -18,7 +18,9 @@ func TestCsatSurveyListener_ResolvedConversationSendsOneSurveyMessage(t *testing
db := dbProvider.DB()
accountID, _ := seedTestAccount(db, t)
conversation := seedCsatListenerConversation(t, db, accountID, true)
indexer := &recordingActionSearchIndexer{}
listener := NewCsatSurveyListener(dbProvider)
listener.SetSearchIndexer(indexer)
event := &channel.ChannelEvent{Type: channel.EventConversationResolved, ConversationID: conversation.ID}
if err := listener.OnEvent(context.Background(), event); err != nil {
@@ -56,6 +58,12 @@ func TestCsatSurveyListener_ResolvedConversationSendsOneSurveyMessage(t *testing
if responseCount != 0 {
t.Fatalf("expected no response before customer submits rating, got %d", responseCount)
}
if len(indexer.messages) != 1 || indexer.messages[0].ID != messages[0].ID {
t.Fatalf("expected one indexed CSAT message %d, got %#v", messages[0].ID, indexer.messages)
}
if len(indexer.conversations) != 1 || indexer.conversations[0].ID != conversation.ID {
t.Fatalf("expected one indexed CSAT conversation %d, got %#v", conversation.ID, indexer.conversations)
}
}
func TestCsatSurveyListener_ResolvedConversationSkipsDisabledInbox(t *testing.T) {
@@ -85,7 +93,9 @@ func TestCsatSurveyListener_ResolvedConversationQueuesDurableSurveySend(t *testi
accountID, _ := seedTestAccount(db, t)
conversation := seedCsatListenerConversation(t, db, accountID, true)
wp := worker.NewWorkerPoolWithOptions(db, worker.WithNow(func() time.Time { return time.Date(2026, 6, 5, 13, 30, 0, 0, time.UTC) }))
indexer := &recordingActionSearchIndexer{}
listener := NewCsatSurveyListenerWithWorker(dbProvider, wp)
listener.SetSearchIndexer(indexer)
event := &channel.ChannelEvent{Type: channel.EventConversationResolved, ConversationID: conversation.ID}
if err := listener.OnEvent(context.Background(), event); err != nil {
@@ -119,6 +129,12 @@ func TestCsatSurveyListener_ResolvedConversationQueuesDurableSurveySend(t *testi
if messageCount != 1 {
t.Fatalf("expected one durable csat message, got %d", messageCount)
}
if len(indexer.messages) != 1 {
t.Fatalf("expected durable CSAT replay to index one message, got %d", len(indexer.messages))
}
if len(indexer.conversations) != 1 || indexer.conversations[0].ID != conversation.ID {
t.Fatalf("expected durable CSAT replay to index conversation %d, got %#v", conversation.ID, indexer.conversations)
}
}
func TestCsatSurveyListener_MessageUpdatedBuildsResponse(t *testing.T) {
@@ -144,7 +160,9 @@ func TestCsatSurveyListener_MessageUpdatedBuildsResponse(t *testing.T) {
if err := db.Create(message).Error; err != nil {
t.Fatalf("failed to create message: %v", err)
}
indexer := &recordingActionSearchIndexer{}
listener := NewCsatSurveyListener(dbProvider)
listener.SetSearchIndexer(indexer)
event := &channel.ChannelEvent{Type: channel.EventMessageUpdated, Data: map[string]any{"message_id": message.ID}}
if err := listener.OnEvent(context.Background(), event); err != nil {
@@ -161,6 +179,12 @@ func TestCsatSurveyListener_MessageUpdatedBuildsResponse(t *testing.T) {
if response.AssignedAgentID == nil || *response.AssignedAgentID != userID {
t.Fatalf("expected assigned agent %d, got %v", userID, response.AssignedAgentID)
}
if len(indexer.messages) != 1 || indexer.messages[0].ID != message.ID {
t.Fatalf("expected updated CSAT message %d to be indexed, got %#v", message.ID, indexer.messages)
}
if len(indexer.conversations) != 1 || indexer.conversations[0].ID != conversation.ID {
t.Fatalf("expected updated CSAT conversation %d to be indexed, got %#v", conversation.ID, indexer.conversations)
}
}
func seedCsatListenerConversation(t *testing.T, db *gorm.DB, accountID uint, csatEnabled bool) *model.Conversation {
internal/automation/csat_survey_service.go
+32 -1
View File
@@ -10,6 +10,7 @@ import (
"time"
"github.com/gochat/gochat/internal/model"
applogger "github.com/gochat/gochat/pkg/logger"
"gorm.io/datatypes"
"gorm.io/gorm"
)
@@ -17,7 +18,15 @@ import (
// CsatSurveyService provides CRUD + metrics for CSAT survey responses.
// Reference: Chatwoot CsatSurveys::ResponseBuilder + CsatSurveyResponse API
type CsatSurveyService struct {
db DBProvider
db DBProvider
searchIndexer CsatSurveySearchIndexer
}
// CsatSurveySearchIndexer is the narrow Meilisearch synchronization boundary
// needed by CSAT survey message creation and submission updates.
type CsatSurveySearchIndexer interface {
IndexConversation(ctx context.Context, conversation *model.Conversation) error
IndexMessage(ctx context.Context, message *model.Message) error
}
// NewCsatSurveyService creates a new CsatSurveyService.
@@ -25,6 +34,10 @@ func NewCsatSurveyService(db DBProvider) *CsatSurveyService {
return &CsatSurveyService{db: db}
}
func (s *CsatSurveyService) SetSearchIndexer(indexer CsatSurveySearchIndexer) {
s.searchIndexer = indexer
}
// Ready reports whether the service has a DB provider configured.
func (s *CsatSurveyService) Ready() bool {
return s != nil && s.db != nil
@@ -118,6 +131,7 @@ func (s *CsatSurveyService) SubmitPublicSurveyByConversationUUID(ctx context.Con
if _, err := ApplyCsatSubmission(ctx, s.db.DB(), message, conversation, submittedValues); err != nil {
return nil, err
}
s.indexCsatSearchDocuments(ctx, message, conversation)
return s.publicCsatPayload(ctx, message, conversation, inbox, account)
}
@@ -179,9 +193,26 @@ func (s *CsatSurveyService) SendSurveyForConversation(ctx context.Context, conve
if err := s.db.DB().WithContext(ctx).Create(message).Error; err != nil {
return nil, err
}
s.indexCsatSearchDocuments(ctx, message, conversation)
return message, nil
}
func (s *CsatSurveyService) indexCsatSearchDocuments(ctx context.Context, message *model.Message, conversation *model.Conversation) {
if s.searchIndexer == nil {
return
}
if message != nil {
if err := s.searchIndexer.IndexMessage(ctx, message); err != nil {
applogger.L().Warnf("search index sync failed for csat message %d: %v", message.ID, err)
}
}
if conversation != nil {
if err := s.searchIndexer.IndexConversation(ctx, conversation); err != nil {
applogger.L().Warnf("search index sync failed for csat conversation %d: %v", conversation.ID, err)
}
}
}
// ListByAccount retrieves CSAT survey responses for an account with optional filters.
func (s *CsatSurveyService) ListByAccount(ctx context.Context, accountID uint, filter CsatListFilter) ([]CsatSurveyResponse, int, error) {
var responses []CsatSurveyResponse
internal/automation/csat_survey_service_test.go
+52
View File
@@ -2,7 +2,11 @@ package automation
import (
"context"
"encoding/json"
"testing"
"github.com/gochat/gochat/internal/model"
"gorm.io/datatypes"
)
func TestCsatSurveyService_Create(t *testing.T) {
@@ -305,3 +309,51 @@ func TestCsatSurveyService_Metrics(t *testing.T) {
t.Fatalf("expected distribution {1:1, 3:1, 5:1}, got %v", metrics.RatingsCount)
}
}
func TestCsatSurveyService_SubmitPublicSurveyIndexesUpdatedMessage(t *testing.T) {
dbProvider := setupAutomationTestDBProvider(t)
db := dbProvider.DB()
accountID, _ := seedTestAccount(db, t)
conversation := seedCsatListenerConversation(t, db, accountID, true)
conversation.UUID = "public-csat-search-index"
if err := db.Save(conversation).Error; err != nil {
t.Fatalf("failed to set conversation uuid: %v", err)
}
message := &model.Message{
ConversationID: conversation.ID,
AccountID: conversation.AccountID,
InboxID: conversation.InboxID,
Content: "Rate this chat",
ContentType: "input_csat",
MessageType: "template",
Status: "sent",
ContentAttributes: datatypes.JSON(`{}`),
}
if err := db.Create(message).Error; err != nil {
t.Fatalf("failed to create CSAT message: %v", err)
}
indexer := &recordingActionSearchIndexer{}
svc := NewCsatSurveyService(dbProvider)
svc.SetSearchIndexer(indexer)
_, err := svc.SubmitPublicSurveyByConversationUUID(context.Background(), conversation.UUID, map[string]any{
"csat_survey_response": map[string]any{"rating": 4, "feedback_message": "Helpful"},
})
if err != nil {
t.Fatalf("submit public survey: %v", err)
}
if len(indexer.messages) != 1 || indexer.messages[0].ID != message.ID {
t.Fatalf("expected submitted CSAT message %d to be indexed, got %#v", message.ID, indexer.messages)
}
var attrs map[string]any
if err := json.Unmarshal(indexer.messages[0].ContentAttributes, &attrs); err != nil {
t.Fatalf("decode indexed message attributes: %v", err)
}
if _, ok := attrs["submitted_values"]; !ok {
t.Fatalf("expected indexed CSAT message to include submitted_values, got %#v", attrs)
}
if len(indexer.conversations) != 1 || indexer.conversations[0].ID != conversation.ID {
t.Fatalf("expected submitted CSAT conversation %d to be indexed, got %#v", conversation.ID, indexer.conversations)
}
}