From aa0deedd48e5126ca51e8daa2923f52259eaf272 Mon Sep 17 00:00:00 2001 From: Abhinav Raut Date: Wed, 5 Aug 2026 18:01:20 +0530 Subject: [PATCH] fix automation rule feedback loops and scope the starts with operator An automation rule that listens on an event and then writes the same field re-fires its own event, so the rule matches again and loops forever. Status already returned early on a no-op write, but priority and user assignment did not, so those two could spin a worker doing a DB write per lap. Add the same unchanged-value guard to both. Suppression while the engine applies actions was a plain flag in a sync.Map, so with more than one worker on the same conversation the first one to finish deleted the key while the other was still applying actions. Make it a refcount so each worker releases only its own claim. The starts with operator is only implemented by the automation evaluator, not the SQL filter builder, so give automation text fields their own operator list instead of adding it to the shared one. Also switch the notify action to its own recipients field type, fix the snooze duration hint since the backend only takes Go duration units, trim and lowercase notify recipient entries, and treat a missing previous value as no match rather than an empty string. --- .../src/composables/useConversationFilters.js | 12 ++--- .../apps/main/src/constants/filterConfig.js | 10 ++++ .../features/admin/automation/ActionBox.vue | 11 ++-- frontend/apps/main/src/stores/webhook.js | 6 +-- i18n/en-US.json | 2 +- internal/automation/automation.go | 54 ++++++++++++++----- internal/automation/evaluator.go | 16 ++++-- internal/automation/evaluator_test.go | 2 +- internal/conversation/conversation.go | 47 ++++++++++++---- internal/webhook/webhook.go | 2 +- 10 files changed, 118 insertions(+), 44 deletions(-) diff --git a/frontend/apps/main/src/composables/useConversationFilters.js b/frontend/apps/main/src/composables/useConversationFilters.js index 0a1df7d2..2f354901 100644 --- a/frontend/apps/main/src/composables/useConversationFilters.js +++ b/frontend/apps/main/src/composables/useConversationFilters.js @@ -29,11 +29,11 @@ export function useConversationFilters () { } const customAttributeDataTypeToFieldOperators = { - 'text': FIELD_OPERATORS.TEXT, + 'text': FIELD_OPERATORS.TEXT_AUTOMATION, 'number': FIELD_OPERATORS.NUMBER, 'checkbox': FIELD_OPERATORS.BOOLEAN, 'date': FIELD_OPERATORS.DATE, - 'link': FIELD_OPERATORS.TEXT, + 'link': FIELD_OPERATORS.TEXT_AUTOMATION, 'list': FIELD_OPERATORS.SELECT, } @@ -164,17 +164,17 @@ export function useConversationFilters () { contact_email: { label: t('globals.terms.email'), type: FIELD_TYPE.TEXT, - operators: FIELD_OPERATORS.TEXT + operators: FIELD_OPERATORS.TEXT_AUTOMATION }, content: { label: t('globals.terms.content'), type: FIELD_TYPE.TEXT, - operators: FIELD_OPERATORS.TEXT + operators: FIELD_OPERATORS.TEXT_AUTOMATION }, subject: { label: t('globals.terms.subject'), type: FIELD_TYPE.TEXT, - operators: FIELD_OPERATORS.TEXT + operators: FIELD_OPERATORS.TEXT_AUTOMATION }, status: { label: t('globals.terms.status'), @@ -336,7 +336,7 @@ export function useConversationFilters () { }, notify: { label: t('actions.notify'), - type: FIELD_TYPE.TAG + type: FIELD_TYPE.RECIPIENTS }, snooze: { label: t('actions.snooze'), diff --git a/frontend/apps/main/src/constants/filterConfig.js b/frontend/apps/main/src/constants/filterConfig.js index 8f9175ec..c60b79ca 100644 --- a/frontend/apps/main/src/constants/filterConfig.js +++ b/frontend/apps/main/src/constants/filterConfig.js @@ -16,6 +16,7 @@ export const FIELD_TYPE = { BOOLEAN: 'boolean', DATE: 'date', WEBHOOK: 'webhook', + RECIPIENTS: 'recipients', } export const OPERATOR = { @@ -46,6 +47,15 @@ export const FIELD_OPERATORS = { SELECT: [OPERATOR.EQUALS, OPERATOR.NOT_EQUALS, OPERATOR.SET, OPERATOR.NOT_SET], BOOLEAN: [OPERATOR.EQUALS, OPERATOR.NOT_EQUALS], TEXT: [ + OPERATOR.EQUALS, + OPERATOR.NOT_EQUALS, + OPERATOR.SET, + OPERATOR.NOT_SET, + OPERATOR.CONTAINS, + OPERATOR.NOT_CONTAINS + ], + // "starts with" is only implemented by the automation evaluator, not the SQL filter builder. + TEXT_AUTOMATION: [ OPERATOR.EQUALS, OPERATOR.NOT_EQUALS, OPERATOR.SET, diff --git a/frontend/apps/main/src/features/admin/automation/ActionBox.vue b/frontend/apps/main/src/features/admin/automation/ActionBox.vue index dfcd6669..e5f06fc7 100644 --- a/frontend/apps/main/src/features/admin/automation/ActionBox.vue +++ b/frontend/apps/main/src/features/admin/automation/ActionBox.vue @@ -34,11 +34,7 @@
-
+
{ const webhooks = ref([]) diff --git a/i18n/en-US.json b/i18n/en-US.json index 0883dcab..04234966 100644 --- a/i18n/en-US.json +++ b/i18n/en-US.json @@ -1163,7 +1163,7 @@ "placeholders.selectType": "Select type", "placeholders.selectValue": "Select value", "placeholders.selectWebhook": "Select webhook", - "placeholders.snoozeDuration": "e.g. 3h, 2h30m, 1d", + "placeholders.snoozeDuration": "e.g. 30m, 3h, 2h30m", "placeholders.startConversation": "Start conversation", "placeholders.tellUsAboutYourself": "Tell us about yourself", "placeholders.webhookEventName": "e.g. priority.escalated", diff --git a/internal/automation/automation.go b/internal/automation/automation.go index 327c6fbb..653b2b26 100644 --- a/internal/automation/automation.go +++ b/internal/automation/automation.go @@ -47,17 +47,20 @@ type ConversationTask struct { } type Engine struct { - rules []models.Rule - rulesMu sync.RWMutex - q queries - lo *logf.Logger - i18n *i18n.I18n - conversationStore conversationStore - taskQueue chan ConversationTask - closed bool - closedMu sync.RWMutex - wg sync.WaitGroup - suppressAutomation sync.Map + rules []models.Rule + rulesMu sync.RWMutex + q queries + lo *logf.Logger + i18n *i18n.I18n + conversationStore conversationStore + taskQueue chan ConversationTask + closed bool + closedMu sync.RWMutex + wg sync.WaitGroup + + // Refcounted per conversation as multiple workers can apply actions on the same conversation concurrently. + suppressed map[string]int + suppressedMu sync.Mutex } type Opts struct { @@ -300,7 +303,7 @@ func (e *Engine) EvaluateConversationUpdateRules(conversation cmodels.Conversati e.lo.Error("error evaluating conversation update rules: eventType is empty") return } - if _, suppressed := e.suppressAutomation.Load(conversation.UUID); suppressed { + if e.isSuppressed(conversation.UUID) { e.lo.Debug("automation suppressed for conversation, skipping update rule evaluation", "uuid", conversation.UUID, "event_type", eventType) return } @@ -379,6 +382,33 @@ func (e *Engine) handleTimeTrigger() { } } +// suppress marks a conversation as having automation actions in flight. +func (e *Engine) suppress(conversationUUID string) { + e.suppressedMu.Lock() + defer e.suppressedMu.Unlock() + if e.suppressed == nil { + e.suppressed = map[string]int{} + } + e.suppressed[conversationUUID]++ +} + +// unsuppress releases one in-flight claim on a conversation. +func (e *Engine) unsuppress(conversationUUID string) { + e.suppressedMu.Lock() + defer e.suppressedMu.Unlock() + if e.suppressed[conversationUUID] <= 1 { + delete(e.suppressed, conversationUUID) + return + } + e.suppressed[conversationUUID]-- +} + +func (e *Engine) isSuppressed(conversationUUID string) bool { + e.suppressedMu.Lock() + defer e.suppressedMu.Unlock() + return e.suppressed[conversationUUID] > 0 +} + // queryRules fetches automation rules from the database. func (e *Engine) queryRules() []models.Rule { var ( diff --git a/internal/automation/evaluator.go b/internal/automation/evaluator.go index 28dc4911..b80c7659 100644 --- a/internal/automation/evaluator.go +++ b/internal/automation/evaluator.go @@ -35,13 +35,13 @@ func (e *Engine) evalConversationRules(rules []models.Rule, conversation cmodels if evaluateFinalResult(groupEvalResults, rule.GroupOperator) { e.lo.Debug("all rules within groups evaluated successfully, executing actions", "conversation_uuid", conversation.UUID) - e.suppressAutomation.Store(conversation.UUID, struct{}{}) + e.suppress(conversation.UUID) for _, action := range rule.Actions { if err := e.conversationStore.ApplyAction(action, conversation, umodels.User{}); err != nil { e.lo.Error("error applying action on conversation", "action", action, "conversation_uuid", conversation.UUID, "error", err) } } - e.suppressAutomation.Delete(conversation.UUID) + e.unsuppress(conversation.UUID) if rule.ExecutionMode == models.ExecutionModeFirstMatch { e.lo.Debug("automation is first match rule execution mode, breaking out of rule evaluation", "conversation_uuid", conversation.UUID) break @@ -153,7 +153,13 @@ func (e *Engine) evaluateRule(rule models.RuleDetail, conversation cmodels.Conve valueToCompare = strconv.Itoa(conversation.InboxID) case models.ConversationPreviousStatus, models.ConversationPreviousPriority, models.ConversationPreviousAssignedUser, models.ConversationPreviousAssignedTeam: - valueToCompare = previousValues[rule.Field] + // An absent key is not the same as an empty previous value. + previous, ok := previousValues[rule.Field] + if !ok { + e.lo.Debug("no previous value available for field, skipping rule", "field", rule.Field, "conversation_uuid", conversation.UUID) + return false + } + valueToCompare = previous default: e.lo.Error("error unrecognized conversation field", "field", rule.Field, "field_type", rule.FieldType, "conversation_uuid", conversation.UUID) return false @@ -230,7 +236,7 @@ func (e *Engine) evaluateRule(rule models.RuleDetail, conversation cmodels.Conve for _, ruleValue := range ruleValues { // Normalize rule value by collapsing multiple spaces normalizedRuleValue := strings.Join(strings.Fields(ruleValue), " ") - + // Respect CaseSensitiveMatch flag if rule.CaseSensitiveMatch { if strings.Contains(normalizedInputText, normalizedRuleValue) { @@ -256,7 +262,7 @@ func (e *Engine) evaluateRule(rule models.RuleDetail, conversation cmodels.Conve for _, ruleValue := range ruleValues { // Normalize rule value by collapsing multiple spaces normalizedRuleValue := strings.Join(strings.Fields(ruleValue), " ") - + // Respect CaseSensitiveMatch flag if rule.CaseSensitiveMatch { if strings.Contains(normalizedInputText, normalizedRuleValue) { diff --git a/internal/automation/evaluator_test.go b/internal/automation/evaluator_test.go index 443d2d49..246f4dc8 100644 --- a/internal/automation/evaluator_test.go +++ b/internal/automation/evaluator_test.go @@ -1594,4 +1594,4 @@ func TestTriggerWebhookAction_PassedThrough(t *testing.T) { mockStore.AssertExpectations(t) assert.Equal(t, 1, mockStore.callCount) -} \ No newline at end of file +} diff --git a/internal/conversation/conversation.go b/internal/conversation/conversation.go index 23553c61..7d778632 100644 --- a/internal/conversation/conversation.go +++ b/internal/conversation/conversation.go @@ -732,12 +732,21 @@ func (c *Manager) UpdateConversationWaitingSince(conversationUUID string, at *ti // UpdateConversationUserAssignee sets the assignee of a conversation to a specifc user. func (c *Manager) UpdateConversationUserAssignee(uuid string, assigneeID int, actor umodels.User) error { - previousConversation, _ := c.GetConversation(0, uuid, "") + previousConversation, err := c.GetConversation(0, uuid, "") + if err != nil { + return err + } previousUserID := "" if previousConversation.AssignedUserID.Valid { previousUserID = strconv.Itoa(previousConversation.AssignedUserID.Int) } + // Return early on a no-op write, else automation rules acting on the user assigned event re-trigger themselves. + if previousConversation.AssignedUserID.Valid && previousConversation.AssignedUserID.Int == assigneeID { + c.lo.Debug("no assignee update: conversation assignee unchanged", "uuid", uuid, "assignee_id", assigneeID) + return nil + } + if err := c.updateAssignee(uuid, assigneeID, models.AssigneeTypeUser); err != nil { return envelope.NewError(envelope.GeneralError, c.i18n.T("globals.messages.somethingWentWrong"), nil) } @@ -765,7 +774,9 @@ func (c *Manager) ClaimUnassignedConversation(uuid string, assigneeID, expectedT } c.broadcastReassignment(uuid, prev, prevErr) - return c.afterUserAssignedHooks(uuid, assigneeID, actor, nil) + return c.afterUserAssignedHooks(uuid, assigneeID, actor, map[string]string{ + amodels.ConversationPreviousAssignedUser: "", + }) } // afterUserAssignedHooks runs the side-effects shared by every user-assignment path. @@ -864,8 +875,12 @@ func (c *Manager) UpdateConversationTeamAssignee(uuid string, teamID int, actor } } + previousTeamID := "" + if previousAssignedTeamID > 0 { + previousTeamID = strconv.Itoa(previousAssignedTeamID) + } c.automation.EvaluateConversationUpdateRules(conversation, amodels.EventConversationTeamAssigned, map[string]string{ - amodels.ConversationPreviousAssignedTeam: strconv.Itoa(previousAssignedTeamID), + amodels.ConversationPreviousAssignedTeam: previousTeamID, }) } @@ -893,7 +908,10 @@ func (c *Manager) broadcastReassignment(uuid string, prev models.ConversationLis // UpdateConversationPriority updates the priority of a conversation. func (c *Manager) UpdateConversationPriority(uuid string, priorityID int, priority string, actor umodels.User) error { - previousConversation, _ := c.GetConversation(0, uuid, "") + previousConversation, err := c.GetConversation(0, uuid, "") + if err != nil { + return err + } previousPriorityID := "" if previousConversation.PriorityID.Valid { previousPriorityID = strconv.Itoa(previousConversation.PriorityID.Int) @@ -906,6 +924,13 @@ func (c *Manager) UpdateConversationPriority(uuid string, priorityID int, priori } priority = p.Name } + + // Return early on a no-op write, else automation rules acting on the priority change event re-trigger themselves. + if previousConversation.Priority.String == priority { + c.lo.Debug("no priority update: conversation priority unchanged", "uuid", uuid, "priority", priority) + return nil + } + if _, err := c.q.UpdateConversationPriority.Exec(uuid, priority); err != nil { c.lo.Error("error updating conversation priority", "error", err) return envelope.NewError(envelope.GeneralError, c.i18n.T("globals.messages.somethingWentWrong"), nil) @@ -1477,7 +1502,10 @@ func (m *Manager) ApplyAction(action amodels.RuleAction, conv models.Conversatio case amodels.ActionSendCSAT: return m.SendCSATReply(user.ID, conv) case amodels.ActionSnooze: - return m.UpdateConversationStatus(conv.UUID, 0, models.StatusSnoozed, action.Value[0], user) + if len(action.Value) == 0 || strings.TrimSpace(action.Value[0]) == "" { + return fmt.Errorf("snooze action requires a duration") + } + return m.UpdateConversationStatus(conv.UUID, 0, models.StatusSnoozed, strings.TrimSpace(action.Value[0]), user) case amodels.ActionTriggerWebhook: if len(action.Value) < 2 { return fmt.Errorf("trigger webhook action requires a webhook and an event name") @@ -1572,12 +1600,13 @@ func (m *Manager) resolveNotifyRecipients(entries []string, conv models.Conversa } func parseNotifyRecipient(entry string) (string, int) { - parts := strings.SplitN(entry, ":", 2) + parts := strings.SplitN(strings.TrimSpace(entry), ":", 2) + kind := strings.ToLower(strings.TrimSpace(parts[0])) if len(parts) == 1 { - return parts[0], 0 + return kind, 0 } - id, _ := strconv.Atoi(parts[1]) - return parts[0], id + id, _ := strconv.Atoi(strings.TrimSpace(parts[1])) + return kind, id } // RemoveConversationAssignee removes assigned user from a conversation. diff --git a/internal/webhook/webhook.go b/internal/webhook/webhook.go index 7bf07e97..47d62a68 100644 --- a/internal/webhook/webhook.go +++ b/internal/webhook/webhook.go @@ -66,7 +66,7 @@ type Opts struct { type DeliveryTask struct { Event models.WebhookEvent Payload any - // WebhookID targets a single webhook instead of fanning out to event subscribers. + // Zero means fan out to all subscribers of Event. WebhookID int }