feat(automation): align message filters

internal/automation/condition_filter.go

@@ -3,8 +3,10 @@ package automation
 import (
 	"context"
 	"fmt"
+	"strconv"
 	"strings"
 
+	"github.com/gochat/gochat/internal/model"
 	"gorm.io/gorm"
 )
 
@@ -29,10 +31,14 @@ func MatchConditions(ctx context.Context, db *gorm.DB, conditions Conditions, co
 	// Separate conditions by type: direct attribute checks vs attribute_changed checks
 	directConditions := Conditions{}
 	changedConditions := Conditions{}
+	messageConditions := Conditions{}
 
 	for _, c := range conditions {
+		attr := conditionAttribute(c)
 		if c.FilterOperator == "attribute_changed" {
 			changedConditions = append(changedConditions, c)
+		} else if isMessageConditionAttribute(attr) {
+			messageConditions = append(messageConditions, c)
 		} else {
 			directConditions = append(directConditions, c)
 		}
@@ -49,6 +55,17 @@ func MatchConditions(ctx context.Context, db *gorm.DB, conditions Conditions, co
 		}
 	}
 
+	// Evaluate message conditions from the triggering message payload.
+	if len(messageConditions) > 0 {
+		matched, err := matchMessageConditions(ctx, db, messageConditions, conv, eventData)
+		if err != nil {
+			return false, err
+		}
+		if !matched {
+			return false, nil
+		}
+	}
+
 	// Evaluate attribute_changed conditions
 	if len(changedConditions) > 0 {
 		matched, err := matchChangedConditions(changedConditions, eventData)
@@ -63,6 +80,251 @@ func MatchConditions(ctx context.Context, db *gorm.DB, conditions Conditions, co
 	return true, nil
 }
 
+func conditionAttribute(c Condition) string {
+	if c.Attribute != "" {
+		return c.Attribute
+	}
+	return c.AttributeKey
+}
+
+func isMessageConditionAttribute(attr string) bool {
+	switch attr {
+	case "content", "message_type", "private_note":
+		return true
+	default:
+		return false
+	}
+}
+
+type messageConditionData struct {
+	Content     string
+	MessageType string
+	Private     bool
+}
+
+func matchMessageConditions(ctx context.Context, db *gorm.DB, conditions Conditions, conv *ConversationForFilter, eventData map[string]interface{}) (bool, error) {
+	message, ok := messageDataForConditions(ctx, db, conv, eventData)
+	if !ok {
+		return false, nil
+	}
+
+	andGroup := Conditions{}
+	orGroup := Conditions{}
+	for _, c := range conditions {
+		if strings.EqualFold(c.QueryOperator, "or") {
+			orGroup = append(orGroup, c)
+		} else {
+			andGroup = append(andGroup, c)
+		}
+	}
+
+	for _, c := range andGroup {
+		matched, err := matchSingleMessageCondition(message, c)
+		if err != nil || !matched {
+			return matched, err
+		}
+	}
+
+	if len(orGroup) == 0 {
+		return true, nil
+	}
+	for _, c := range orGroup {
+		matched, err := matchSingleMessageCondition(message, c)
+		if err != nil {
+			return false, err
+		}
+		if matched {
+			return true, nil
+		}
+	}
+	return false, nil
+}
+
+func matchSingleMessageCondition(message messageConditionData, c Condition) (bool, error) {
+	attr := conditionAttribute(c)
+	op := NormalizeFilterOperator(c.FilterOperator)
+	if len(c.Values) == 0 {
+		return false, nil
+	}
+
+	switch attr {
+	case "content":
+		return matchTextCondition(message.Content, c.Values, op), nil
+	case "private_note":
+		return matchBoolCondition(message.Private, c.Values, op), nil
+	case "message_type":
+		return matchMessageTypeCondition(message.MessageType, c.Values, op), nil
+	default:
+		return false, fmt.Errorf("unsupported message condition attribute: %s", attr)
+	}
+}
+
+func matchTextCondition(actual string, values []string, op string) bool {
+	actual = strings.ToLower(actual)
+	matches := func(value string) bool { return actual == strings.ToLower(value) }
+	contains := func(value string) bool { return strings.Contains(actual, strings.ToLower(value)) }
+	switch op {
+	case "equal":
+		return anyValue(values, matches)
+	case "not_equal":
+		return !anyValue(values, matches)
+	case "contains":
+		return anyValue(values, contains)
+	case "does_not_contain":
+		return !anyValue(values, contains)
+	default:
+		return false
+	}
+}
+
+func matchBoolCondition(actual bool, values []string, op string) bool {
+	matches := func(value string) bool {
+		parsed, ok := parseBoolLike(value)
+		return ok && parsed == actual
+	}
+	switch op {
+	case "equal":
+		return anyValue(values, matches)
+	case "not_equal":
+		return !anyValue(values, matches)
+	default:
+		return false
+	}
+}
+
+func matchMessageTypeCondition(actual string, values []string, op string) bool {
+	actual = normalizeMessageTypeForCondition(actual)
+	matches := func(value string) bool { return actual == normalizeMessageTypeForCondition(value) }
+	switch op {
+	case "equal":
+		return anyValue(values, matches)
+	case "not_equal":
+		return !anyValue(values, matches)
+	default:
+		return false
+	}
+}
+
+func anyValue(values []string, fn func(string) bool) bool {
+	for _, value := range values {
+		if fn(value) {
+			return true
+		}
+	}
+	return false
+}
+
+func parseBoolLike(value string) (bool, bool) {
+	switch strings.ToLower(strings.TrimSpace(value)) {
+	case "true", "1", "t", "yes":
+		return true, true
+	case "false", "0", "f", "no":
+		return false, true
+	default:
+		parsed, err := strconv.ParseBool(value)
+		return parsed, err == nil
+	}
+}
+
+func normalizeMessageTypeForCondition(value string) string {
+	switch strings.ToLower(strings.TrimSpace(value)) {
+	case "0", "incoming":
+		return string(model.MessageTypeIncoming)
+	case "1", "outgoing":
+		return string(model.MessageTypeOutgoing)
+	case "2", "activity":
+		return string(model.MessageTypeActivity)
+	case "3", "template":
+		return string(model.MessageTypeTemplate)
+	default:
+		return strings.ToLower(strings.TrimSpace(value))
+	}
+}
+
+func messageDataForConditions(ctx context.Context, db *gorm.DB, conv *ConversationForFilter, eventData map[string]interface{}) (messageConditionData, bool) {
+	if message, ok := extractMessageFromData(eventData); ok && message != nil {
+		return messageConditionDataFromModel(message), true
+	}
+	if message, ok := messageConditionDataFromMap(eventData); ok {
+		return message, true
+	}
+	if messageID, ok := extractUintFromData(eventData, "message_id"); ok && messageID != 0 && db != nil {
+		var message model.Message
+		query := db.WithContext(ctx).Where("id = ?", messageID)
+		if conv != nil && conv.ID != 0 {
+			query = query.Where("conversation_id = ?", conv.ID)
+		}
+		if err := query.First(&message).Error; err == nil {
+			return messageConditionDataFromModel(&message), true
+		}
+	}
+	return messageConditionData{}, false
+}
+
+func messageConditionDataFromModel(message *model.Message) messageConditionData {
+	content := message.Content
+	attrs := listenerJSONMap(message.ContentAttributes)
+	if processed, ok := attrs["processed_message_content"].(string); ok && processed != "" {
+		content = processed
+	}
+	return messageConditionData{Content: content, MessageType: message.MessageType, Private: message.Private}
+}
+
+func messageConditionDataFromMap(data map[string]interface{}) (messageConditionData, bool) {
+	if data == nil {
+		return messageConditionData{}, false
+	}
+	raw := data
+	if nested, ok := data["message"].(map[string]interface{}); ok {
+		raw = nested
+	}
+	content := firstStringValue(raw, "processed_message_content", "content")
+	messageType := firstStringValue(raw, "message_type")
+	private, hasPrivate := firstBoolValue(raw, "private", "private_note")
+	if content == "" && messageType == "" && !hasPrivate {
+		return messageConditionData{}, false
+	}
+	return messageConditionData{Content: content, MessageType: messageType, Private: private}, true
+}
+
+func firstStringValue(values map[string]interface{}, keys ...string) string {
+	for _, key := range keys {
+		if value, ok := values[key]; ok && value != nil {
+			switch v := value.(type) {
+			case string:
+				return v
+			case fmt.Stringer:
+				return v.String()
+			default:
+				return fmt.Sprintf("%v", v)
+			}
+		}
+	}
+	return ""
+}
+
+func firstBoolValue(values map[string]interface{}, keys ...string) (bool, bool) {
+	for _, key := range keys {
+		if value, ok := values[key]; ok && value != nil {
+			switch v := value.(type) {
+			case bool:
+				return v, true
+			case string:
+				return parseBoolLike(v)
+			case int:
+				return v != 0, true
+			case uint:
+				return v != 0, true
+			case float64:
+				return v != 0, true
+			default:
+				return false, false
+			}
+		}
+	}
+	return false, false
+}
+
 // matchDirectConditions evaluates conditions that directly check conversation attributes.
 // Reference: Chatwoot ConditionsFilterService — builds SQL WHERE clauses from conditions
 func matchDirectConditions(db *gorm.DB, conditions Conditions, conv *ConversationForFilter) (bool, error) {

internal/automation/condition_filter_test.go

@@ -3,6 +3,9 @@ package automation
 import (
 	"context"
 	"testing"
+
+	"github.com/gochat/gochat/internal/model"
+	"gorm.io/datatypes"
 )
 
 // ===========================
@@ -238,6 +241,102 @@ func TestMatchConditions_MixedDirectAndChanged(t *testing.T) {
 	}
 }
 
+func TestMatchConditions_MessageCreatedMessageFilters(t *testing.T) {
+	dbProvider := setupAutomationTestDBProvider(t)
+	accountID, _ := seedTestAccount(dbProvider.DB(), t)
+	inboxID := seedTestInbox(dbProvider.DB(), t, accountID)
+	contactID := seedTestContact(dbProvider.DB(), t, accountID)
+	convID := seedTestConversationWithDetails(dbProvider.DB(), t, accountID, inboxID, contactID, "open", "", "web", 0)
+	conv := &ConversationForFilter{ID: convID, AccountID: accountID, InboxID: inboxID, ContactID: contactID, Status: "open"}
+
+	eventData := map[string]interface{}{
+		"message": &model.Message{
+			AccountID:         accountID,
+			InboxID:           inboxID,
+			ConversationID:    convID,
+			Content:           "Original markdown",
+			ContentAttributes: datatypes.JSON([]byte(`{"processed_message_content":"Need Help With Billing"}`)),
+			MessageType:       "incoming",
+			Private:           false,
+		},
+	}
+
+	conditions := Conditions{
+		{Attribute: "message_type", FilterOperator: "equal_to", Values: []string{"0"}, QueryOperator: "and"},
+		{Attribute: "private_note", FilterOperator: "equal_to", Values: []string{"false"}, QueryOperator: "and"},
+		{Attribute: "content", FilterOperator: "contains", Values: []string{"billing"}, QueryOperator: "and"},
+	}
+
+	matched, err := MatchConditions(context.Background(), dbProvider.DB(), conditions, conv, eventData)
+	if err != nil {
+		t.Fatalf("expected no error, got: %v", err)
+	}
+	if !matched {
+		t.Fatal("expected Chatwoot message filters to match triggering message")
+	}
+}
+
+func TestMatchConditions_MessageFilterRejectsNonMatchingMessage(t *testing.T) {
+	dbProvider := setupAutomationTestDBProvider(t)
+	accountID, _ := seedTestAccount(dbProvider.DB(), t)
+	inboxID := seedTestInbox(dbProvider.DB(), t, accountID)
+	contactID := seedTestContact(dbProvider.DB(), t, accountID)
+	convID := seedTestConversationWithDetails(dbProvider.DB(), t, accountID, inboxID, contactID, "open", "", "web", 0)
+	conv := &ConversationForFilter{ID: convID, AccountID: accountID, InboxID: inboxID, ContactID: contactID, Status: "open"}
+
+	eventData := map[string]interface{}{
+		"message": &model.Message{AccountID: accountID, InboxID: inboxID, ConversationID: convID, Content: "Need help", MessageType: "outgoing", Private: true},
+	}
+	conditions := Conditions{{Attribute: "message_type", FilterOperator: "equal_to", Values: []string{"incoming"}}}
+
+	matched, err := MatchConditions(context.Background(), dbProvider.DB(), conditions, conv, eventData)
+	if err != nil {
+		t.Fatalf("expected no error, got: %v", err)
+	}
+	if matched {
+		t.Fatal("expected message_type condition not to match outgoing message")
+	}
+}
+
+func TestMatchConditions_MessageFiltersSupportOrGroup(t *testing.T) {
+	dbProvider := setupAutomationTestDBProvider(t)
+	accountID, _ := seedTestAccount(dbProvider.DB(), t)
+	inboxID := seedTestInbox(dbProvider.DB(), t, accountID)
+	contactID := seedTestContact(dbProvider.DB(), t, accountID)
+	convID := seedTestConversationWithDetails(dbProvider.DB(), t, accountID, inboxID, contactID, "open", "", "web", 0)
+	conv := &ConversationForFilter{ID: convID, AccountID: accountID, InboxID: inboxID, ContactID: contactID, Status: "open"}
+
+	eventData := map[string]interface{}{
+		"message": map[string]interface{}{"content": "Need billing help", "message_type": "outgoing", "private": false},
+	}
+	conditions := Conditions{
+		{Attribute: "content", FilterOperator: "contains", Values: []string{"pricing"}, QueryOperator: "or"},
+		{Attribute: "message_type", FilterOperator: "equal_to", Values: []string{"1"}, QueryOperator: "or"},
+	}
+
+	matched, err := MatchConditions(context.Background(), dbProvider.DB(), conditions, conv, eventData)
+	if err != nil {
+		t.Fatalf("expected no error, got: %v", err)
+	}
+	if !matched {
+		t.Fatal("expected OR message filter group to match outgoing message_type")
+	}
+}
+
+func TestMatchConditions_MessageFilterWithoutMessageDoesNotMatch(t *testing.T) {
+	dbProvider := setupAutomationTestDBProvider(t)
+	conv := &ConversationForFilter{ID: 1, Status: "open"}
+	conditions := Conditions{{Attribute: "content", FilterOperator: "contains", Values: []string{"help"}}}
+
+	matched, err := MatchConditions(context.Background(), dbProvider.DB(), conditions, conv, map[string]interface{}{})
+	if err != nil {
+		t.Fatalf("expected no error, got: %v", err)
+	}
+	if matched {
+		t.Fatal("expected message condition without triggering message data not to match")
+	}
+}
+
 // ===========================
 // buildConditionClause tests (pure logic, no DB)
 // ===========================
@@ -541,4 +640,4 @@ func TestNormalizeEventName(t *testing.T) {
 			t.Fatalf("normalizeEventName(%q) = %q, expected %q", tt.input, result, tt.expected)
 		}
 	}
-}
+}

internal/automation/condition_validator.go

@@ -157,6 +157,10 @@ func ValidateConditions(conditions Conditions) error {
 			}
 		}
 
+		if err := validateAttributeOperator(c.Attribute, filterOperator); err != nil {
+			return fmt.Errorf("condition[%d]: %w", i, err)
+		}
+
 		// is_present / is_not_present should not need values
 		if filterOperator == "is_present" || filterOperator == "is_not_present" {
 			// Values are optional for presence checks — not enforced
@@ -168,6 +172,20 @@ func ValidateConditions(conditions Conditions) error {
 	return nil
 }
 
+func validateAttributeOperator(attribute string, operator string) error {
+	switch attribute {
+	case "message_type", "private_note":
+		if operator != "equal" && operator != "not_equal" {
+			return fmt.Errorf("attribute '%s' does not support filter_operator '%s'", attribute, operator)
+		}
+	case "content":
+		if operator != "equal" && operator != "not_equal" && operator != "contains" && operator != "does_not_contain" {
+			return fmt.Errorf("attribute 'content' does not support filter_operator '%s'", operator)
+		}
+	}
+	return nil
+}
+
 // ValidateActions checks that all actions use valid action types.
 // Reference: Chatwoot AutomationRule action_name validation
 func ValidateActions(actions Actions) error {

internal/automation/condition_validator_test.go

@@ -183,6 +183,30 @@ func TestValidateConditions_ContainsOperator(t *testing.T) {
 	}
 }
 
+func TestValidateConditions_ChatwootMessageFilterOperators(t *testing.T) {
+	valid := Conditions{
+		{Attribute: "message_type", FilterOperator: "equal_to", Values: []string{"0"}},
+		{Attribute: "private_note", FilterOperator: "not_equal_to", Values: []string{"true"}},
+		{Attribute: "content", FilterOperator: "does_not_contain", Values: []string{"spam"}},
+	}
+	if err := ValidateConditions(valid); err != nil {
+		t.Fatalf("expected Chatwoot message filter operators to validate, got: %v", err)
+	}
+}
+
+func TestValidateConditions_RejectsUnsupportedMessageFilterOperators(t *testing.T) {
+	tests := []Condition{
+		{Attribute: "message_type", FilterOperator: "contains", Values: []string{"incoming"}},
+		{Attribute: "private_note", FilterOperator: "contains", Values: []string{"true"}},
+		{Attribute: "content", FilterOperator: "is_present", Values: []string{}},
+	}
+	for _, condition := range tests {
+		if err := ValidateConditions(Conditions{condition}); err == nil {
+			t.Fatalf("expected invalid operator for %s/%s", condition.Attribute, condition.FilterOperator)
+		}
+	}
+}
+
 func TestValidateConditions_AttributeChangedOperator(t *testing.T) {
 	conditions := Conditions{
 		{
@@ -211,4 +235,4 @@ func TestValidateConditions_DoesNotContainOperator(t *testing.T) {
 	if err != nil {
 		t.Fatalf("expected no error for does_not_contain operator, got: %v", err)
 	}
-}
+}

internal/automation/listener_test.go

@@ -303,6 +303,55 @@ func TestAutomationRuleListener_OnEvent_DispatchesAllChatwootReferenceEvents(t *
 	}
 }
 
+func TestAutomationRuleListener_OnEvent_MessageCreatedAppliesMessageFilters(t *testing.T) {
+	dbProvider := setupAutomationTestDBProvider(t)
+	db := dbProvider.DB()
+
+	accountID, _ := seedTestAccount(db, t)
+	inboxID := seedTestInbox(db, t, accountID)
+	contactID := seedTestContact(db, t, accountID)
+	conversationID := seedTestConversation(db, t, accountID, inboxID, contactID)
+
+	matchingRule := &AutomationRule{
+		AccountID: accountID,
+		EventName: "message_created",
+		Name:      "incoming billing message",
+		Conditions: Conditions{
+			{Attribute: "message_type", FilterOperator: "equal_to", Values: []string{"0"}, QueryOperator: "and"},
+			{Attribute: "private_note", FilterOperator: "equal_to", Values: []string{"false"}, QueryOperator: "and"},
+			{Attribute: "content", FilterOperator: "contains", Values: []string{"billing"}, QueryOperator: "and"},
+		},
+		Actions: Actions{},
+		Active:  true,
+	}
+	nonMatchingRule := &AutomationRule{
+		AccountID: accountID,
+		EventName: "message_created",
+		Name:      "private note only",
+		Conditions: Conditions{
+			{Attribute: "private_note", FilterOperator: "equal_to", Values: []string{"true"}},
+		},
+		Actions: Actions{},
+		Active:  true,
+	}
+	ruleSvc := NewAutomationRuleService(dbProvider)
+	require.NoError(t, ruleSvc.Create(context.Background(), matchingRule))
+	require.NoError(t, ruleSvc.Create(context.Background(), nonMatchingRule))
+
+	message := &model.Message{AccountID: accountID, InboxID: inboxID, ConversationID: conversationID, MessageType: "incoming", ContentType: "text", Content: "I need Billing help", Private: false}
+	event := &channel.ChannelEvent{Type: channel.EventMessageCreated, AccountID: accountID, InboxID: inboxID, Data: map[string]interface{}{"message": message}}
+
+	require.NoError(t, NewAutomationRuleListener(dbProvider).OnEvent(context.Background(), event))
+
+	var matchedCount int64
+	require.NoError(t, db.Model(&AutomationExecution{}).Where("rule_id = ? AND status = ?", matchingRule.ID, ExecutionStatusSuccess).Count(&matchedCount).Error)
+	assert.Equal(t, int64(1), matchedCount)
+
+	var skippedCount int64
+	require.NoError(t, db.Model(&AutomationExecution{}).Where("rule_id = ? AND status = ?", nonMatchingRule.ID, ExecutionStatusSkipped).Count(&skippedCount).Error)
+	assert.Equal(t, int64(1), skippedCount)
+}
+
 func TestAutomationRuleListener_OnEvent_UsesChangedAttributesForConditions(t *testing.T) {
 	dbProvider := setupAutomationTestDBProvider(t)
 	db := dbProvider.DB()