458 lines
16 KiB
Go
458 lines
16 KiB
Go
package automation
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"strconv"
|
|
"time"
|
|
|
|
"github.com/gochat/gochat/internal/model"
|
|
"github.com/gochat/gochat/internal/worker"
|
|
applogger "github.com/gochat/gochat/pkg/logger"
|
|
"gorm.io/gorm"
|
|
)
|
|
|
|
// DBProvider provides database access — implemented by app.App to break the import cycle.
|
|
type DBProvider interface {
|
|
DB() *gorm.DB
|
|
}
|
|
|
|
// AutomationRuleService provides CRUD + condition matching + action execution for automation rules.
|
|
// Reference: Chatwoot AutomationRules::ActionService + ConditionsFilterService pattern
|
|
type AutomationRuleService struct {
|
|
db DBProvider
|
|
worker *worker.WorkerPool
|
|
searchIndexer ActionSearchIndexer
|
|
}
|
|
|
|
// NewAutomationRuleService creates a new AutomationRuleService.
|
|
func NewAutomationRuleService(db DBProvider) *AutomationRuleService {
|
|
return &AutomationRuleService{db: db}
|
|
}
|
|
|
|
func NewAutomationRuleServiceWithWorker(db DBProvider, wp *worker.WorkerPool) *AutomationRuleService {
|
|
s := NewAutomationRuleService(db)
|
|
s.SetWorkerPool(wp)
|
|
return s
|
|
}
|
|
|
|
func (s *AutomationRuleService) SetWorkerPool(wp *worker.WorkerPool) {
|
|
s.worker = wp
|
|
RegisterActionDeliveryJobs(wp, s.db)
|
|
}
|
|
|
|
func (s *AutomationRuleService) SetSearchIndexer(indexer ActionSearchIndexer) {
|
|
s.searchIndexer = indexer
|
|
}
|
|
|
|
// GetByID retrieves an automation rule by ID.
|
|
func (s *AutomationRuleService) GetByID(ctx context.Context, id uint) (*AutomationRule, error) {
|
|
var rule AutomationRule
|
|
if err := s.db.DB().WithContext(ctx).First(&rule, id).Error; err != nil {
|
|
return nil, err
|
|
}
|
|
_ = s.hydrateAutomationRuleFiles(ctx, &rule)
|
|
return &rule, nil
|
|
}
|
|
|
|
// GetByIDForAccount retrieves an automation rule scoped to an account.
|
|
func (s *AutomationRuleService) GetByIDForAccount(ctx context.Context, accountID, id uint) (*AutomationRule, error) {
|
|
var rule AutomationRule
|
|
if err := s.db.DB().WithContext(ctx).
|
|
Where("account_id = ?", accountID).
|
|
First(&rule, id).Error; err != nil {
|
|
return nil, err
|
|
}
|
|
_ = s.hydrateAutomationRuleFiles(ctx, &rule)
|
|
return &rule, nil
|
|
}
|
|
|
|
// ListByAccount retrieves all automation rules for an account.
|
|
func (s *AutomationRuleService) ListByAccount(ctx context.Context, accountID uint) ([]AutomationRule, error) {
|
|
var rules []AutomationRule
|
|
if err := s.db.DB().WithContext(ctx).
|
|
Where("account_id = ?", accountID).
|
|
Order("id ASC").
|
|
Find(&rules).Error; err != nil {
|
|
return nil, err
|
|
}
|
|
for i := range rules {
|
|
_ = s.hydrateAutomationRuleFiles(ctx, &rules[i])
|
|
}
|
|
return rules, nil
|
|
}
|
|
|
|
// ListActiveByAccountAndEvent retrieves active automation rules for a given account and event.
|
|
// Reference: Chatwoot AutomationRuleListener loads rules matching event_name
|
|
func (s *AutomationRuleService) ListActiveByAccountAndEvent(ctx context.Context, accountID uint, eventName string) ([]AutomationRule, error) {
|
|
var rules []AutomationRule
|
|
if err := s.db.DB().WithContext(ctx).
|
|
Where("account_id = ? AND event_name = ? AND active = ?", accountID, eventName, true).
|
|
Order("id ASC").
|
|
Find(&rules).Error; err != nil {
|
|
return nil, err
|
|
}
|
|
return rules, nil
|
|
}
|
|
|
|
// Create creates a new automation rule.
|
|
// New rules default to active unless explicitly set to inactive.
|
|
func (s *AutomationRuleService) Create(ctx context.Context, rule *AutomationRule) error {
|
|
rule.Conditions = NormalizeConditions(rule.Conditions)
|
|
// Validate conditions before saving
|
|
if err := ValidateConditions(rule.Conditions); err != nil {
|
|
return fmt.Errorf("invalid conditions: %w", err)
|
|
}
|
|
if err := s.normalizeAutomationRuleAttachmentActions(ctx, rule.AccountID, &rule.Actions); err != nil {
|
|
return err
|
|
}
|
|
// Validate actions before saving
|
|
if err := ValidateActions(rule.Actions); err != nil {
|
|
return fmt.Errorf("invalid actions: %w", err)
|
|
}
|
|
// Use Select to force all fields including zero-value bool Active=false
|
|
// Without Select, GORM skips zero-value fields and uses column defaults.
|
|
if err := s.db.DB().WithContext(ctx).Select(
|
|
"AccountID", "EventName", "Name", "Description",
|
|
"Conditions", "Actions", "Active",
|
|
).Create(rule).Error; err != nil {
|
|
return err
|
|
}
|
|
return s.hydrateAutomationRuleFiles(ctx, rule)
|
|
}
|
|
|
|
// Update updates an existing automation rule.
|
|
func (s *AutomationRuleService) Update(ctx context.Context, rule *AutomationRule) error {
|
|
rule.Conditions = NormalizeConditions(rule.Conditions)
|
|
// Validate conditions before saving
|
|
if err := ValidateConditions(rule.Conditions); err != nil {
|
|
return fmt.Errorf("invalid conditions: %w", err)
|
|
}
|
|
if err := s.normalizeAutomationRuleAttachmentActions(ctx, rule.AccountID, &rule.Actions); err != nil {
|
|
return err
|
|
}
|
|
// Validate actions before saving
|
|
if err := ValidateActions(rule.Actions); err != nil {
|
|
return fmt.Errorf("invalid actions: %w", err)
|
|
}
|
|
return s.db.DB().WithContext(ctx).Save(rule).Error
|
|
}
|
|
|
|
// UpdateForAccount updates an existing automation rule within account scope.
|
|
func (s *AutomationRuleService) UpdateForAccount(ctx context.Context, accountID uint, rule *AutomationRule) error {
|
|
var existing AutomationRule
|
|
if err := s.db.DB().WithContext(ctx).
|
|
Where("account_id = ?", accountID).
|
|
First(&existing, rule.ID).Error; err != nil {
|
|
return err
|
|
}
|
|
|
|
existing.Name = rule.Name
|
|
existing.Description = rule.Description
|
|
existing.EventName = rule.EventName
|
|
existing.Conditions = NormalizeConditions(rule.Conditions)
|
|
if rule.Actions != nil {
|
|
if err := s.normalizeAutomationRuleAttachmentActions(ctx, accountID, &rule.Actions); err != nil {
|
|
return err
|
|
}
|
|
existing.Actions = rule.Actions
|
|
}
|
|
existing.Active = rule.Active
|
|
|
|
return s.Update(ctx, &existing)
|
|
}
|
|
|
|
func (s *AutomationRuleService) normalizeAutomationRuleAttachmentActions(ctx context.Context, accountID uint, actions *Actions) error {
|
|
if actions == nil {
|
|
return nil
|
|
}
|
|
for i := range *actions {
|
|
action := &(*actions)[i]
|
|
if action.ActionName != "send_attachment" {
|
|
continue
|
|
}
|
|
if action.ActionParams == nil {
|
|
return fmt.Errorf("invalid attachment")
|
|
}
|
|
upload, err := s.findAutomationRuleUpload(ctx, accountID, firstAutomationRuleBlobID(action.ActionParams))
|
|
if err != nil {
|
|
return fmt.Errorf("invalid attachment")
|
|
}
|
|
action.ActionParams["blob_id"] = upload.ID
|
|
delete(action.ActionParams, "attachment_url")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func firstAutomationRuleBlobID(params map[string]interface{}) interface{} {
|
|
if value := params["blob_id"]; value != nil {
|
|
return value
|
|
}
|
|
return params["attachment_url"]
|
|
}
|
|
|
|
func (s *AutomationRuleService) findAutomationRuleUpload(ctx context.Context, accountID uint, value interface{}) (*model.DirectUpload, error) {
|
|
switch v := value.(type) {
|
|
case string:
|
|
if parsed, err := strconv.ParseUint(v, 10, 64); err == nil {
|
|
return s.findAutomationRuleUploadByID(ctx, accountID, uint(parsed))
|
|
}
|
|
var upload model.DirectUpload
|
|
if err := s.db.DB().WithContext(ctx).Where("account_id = ? AND upload_uuid = ?", accountID, v).First(&upload).Error; err != nil {
|
|
return nil, err
|
|
}
|
|
return &upload, nil
|
|
case float64:
|
|
return s.findAutomationRuleUploadByID(ctx, accountID, uint(v))
|
|
case int:
|
|
return s.findAutomationRuleUploadByID(ctx, accountID, uint(v))
|
|
case uint:
|
|
return s.findAutomationRuleUploadByID(ctx, accountID, v)
|
|
case json.Number:
|
|
parsed, err := strconv.ParseUint(string(v), 10, 64)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return s.findAutomationRuleUploadByID(ctx, accountID, uint(parsed))
|
|
default:
|
|
return nil, gorm.ErrRecordNotFound
|
|
}
|
|
}
|
|
|
|
func (s *AutomationRuleService) findAutomationRuleUploadByID(ctx context.Context, accountID, id uint) (*model.DirectUpload, error) {
|
|
var upload model.DirectUpload
|
|
if err := s.db.DB().WithContext(ctx).Where("account_id = ? AND id = ?", accountID, id).First(&upload).Error; err != nil {
|
|
return nil, err
|
|
}
|
|
return &upload, nil
|
|
}
|
|
|
|
func (s *AutomationRuleService) hydrateAutomationRuleFiles(ctx context.Context, rule *AutomationRule) error {
|
|
ids := macroAttachmentBlobIDs(rule.Actions)
|
|
if len(ids) == 0 {
|
|
rule.Files = nil
|
|
return nil
|
|
}
|
|
var uploads []model.DirectUpload
|
|
if err := s.db.DB().WithContext(ctx).Where("account_id = ? AND id IN ?", rule.AccountID, ids).Order("id ASC").Find(&uploads).Error; err != nil {
|
|
return err
|
|
}
|
|
files := make([]AutomationRuleFile, 0, len(uploads))
|
|
for _, upload := range uploads {
|
|
files = append(files, AutomationRuleFile{ID: upload.ID, AutomationRuleID: rule.ID, FileType: upload.MimeType, AccountID: upload.AccountID, FileURL: upload.FileURL, BlobID: upload.ID, Filename: upload.OriginalName})
|
|
}
|
|
rule.Files = files
|
|
return nil
|
|
}
|
|
|
|
// Delete soft-deletes an automation rule.
|
|
func (s *AutomationRuleService) Delete(ctx context.Context, id uint) error {
|
|
return s.db.DB().WithContext(ctx).Delete(&AutomationRule{}, id).Error
|
|
}
|
|
|
|
// DeleteForAccount deletes an automation rule scoped to an account.
|
|
func (s *AutomationRuleService) DeleteForAccount(ctx context.Context, accountID, id uint) error {
|
|
rule, err := s.GetByIDForAccount(ctx, accountID, id)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return s.db.DB().WithContext(ctx).Delete(rule).Error
|
|
}
|
|
|
|
// Clone duplicates an automation rule, creating a new copy with "(copy)" suffix and Active=false.
|
|
// Reference: Chatwoot AutomationRulesController#clone
|
|
func (s *AutomationRuleService) Clone(ctx context.Context, id uint) (*AutomationRule, error) {
|
|
original, err := s.GetByID(ctx, id)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
cloned := &AutomationRule{
|
|
AccountID: original.AccountID,
|
|
EventName: original.EventName,
|
|
Name: original.Name + " (copy)",
|
|
Description: original.Description,
|
|
Conditions: original.Conditions,
|
|
Actions: original.Actions,
|
|
Active: false,
|
|
}
|
|
|
|
if err := s.db.DB().WithContext(ctx).Create(cloned).Error; err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return cloned, nil
|
|
}
|
|
|
|
// CloneForAccount duplicates an automation rule scoped to an account.
|
|
func (s *AutomationRuleService) CloneForAccount(ctx context.Context, accountID, id uint) (*AutomationRule, error) {
|
|
original, err := s.GetByIDForAccount(ctx, accountID, id)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
cloned := &AutomationRule{
|
|
AccountID: original.AccountID,
|
|
EventName: original.EventName,
|
|
Name: original.Name + " (copy)",
|
|
Description: original.Description,
|
|
Conditions: original.Conditions,
|
|
Actions: original.Actions,
|
|
Active: false,
|
|
}
|
|
|
|
if err := s.db.DB().WithContext(ctx).Create(cloned).Error; err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return cloned, nil
|
|
}
|
|
|
|
// ToggleActive toggles the active state of an automation rule and sets the corresponding timestamp.
|
|
func (s *AutomationRuleService) ToggleActive(ctx context.Context, id uint, active bool) error {
|
|
now := time.Now()
|
|
updates := map[string]interface{}{
|
|
"active": active,
|
|
}
|
|
if active {
|
|
updates["active_at"] = &now
|
|
updates["inactive_at"] = nil
|
|
} else {
|
|
updates["inactive_at"] = &now
|
|
updates["active_at"] = nil
|
|
}
|
|
return s.db.DB().WithContext(ctx).
|
|
Model(&AutomationRule{}).
|
|
Where("id = ?", id).
|
|
Updates(updates).Error
|
|
}
|
|
|
|
// ToggleActiveForAccount toggles the active state within account scope.
|
|
func (s *AutomationRuleService) ToggleActiveForAccount(ctx context.Context, accountID, id uint, active bool) error {
|
|
if _, err := s.GetByIDForAccount(ctx, accountID, id); err != nil {
|
|
return err
|
|
}
|
|
return s.ToggleActive(ctx, id, active)
|
|
}
|
|
|
|
// MatchAndExecute checks all active rules for the given event and account,
|
|
// runs condition filters for each rule, and executes actions for matching rules.
|
|
// Reference: Chatwoot AutomationRuleListener — loads rules, runs ConditionsFilterService, then ActionService
|
|
func (s *AutomationRuleService) MatchAndExecute(ctx context.Context, accountID uint, eventName string, conversationID uint, eventData map[string]interface{}) error {
|
|
rules, err := s.ListActiveByAccountAndEvent(ctx, accountID, eventName)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to load automation rules: %w", err)
|
|
}
|
|
|
|
if len(rules) == 0 {
|
|
applogger.L().Debugf("no active automation rules for account %d event %s", accountID, eventName)
|
|
return nil
|
|
}
|
|
|
|
// Load the conversation for condition matching
|
|
conversation, err := s.loadConversation(ctx, conversationID)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to load conversation %d: %w", conversationID, err)
|
|
}
|
|
|
|
actionSvc := NewActionServiceWithWorker(s.db, s.worker)
|
|
actionSvc.SetSearchIndexer(s.searchIndexer)
|
|
logSvc := NewExecutionLogService(s.db)
|
|
|
|
for _, rule := range rules {
|
|
matched, err := MatchConditions(ctx, s.db.DB(), rule.Conditions, conversation, eventData)
|
|
if err != nil {
|
|
applogger.L().Errorf("condition matching failed for rule %d: %v", rule.ID, err)
|
|
if logErr := logSvc.LogRuleExecutionWithResults(ctx, accountID, rule.ID, conversationID, eventName, ExecutionStatusFailed, 0, 0, fmt.Sprintf("condition matching failed: %v", err), nil); logErr != nil {
|
|
applogger.L().Warnf("failed to log automation condition error for rule %d: %v", rule.ID, logErr)
|
|
}
|
|
continue
|
|
}
|
|
if !matched {
|
|
applogger.L().Debugf("rule %d did not match conditions for conversation %d", rule.ID, conversationID)
|
|
if logErr := logSvc.LogRuleExecutionWithResults(ctx, accountID, rule.ID, conversationID, eventName, ExecutionStatusSkipped, 0, 0, "conditions_not_matched", nil); logErr != nil {
|
|
applogger.L().Warnf("failed to log skipped automation rule %d: %v", rule.ID, logErr)
|
|
}
|
|
continue
|
|
}
|
|
|
|
applogger.L().Infof("automation rule %d matched for conversation %d, executing %d actions", rule.ID, conversationID, len(rule.Actions))
|
|
|
|
// Stamp automation_rule_id on actions for audit trail
|
|
// Reference: Chatwoot AutomationRules::ActionService stamps automation_rule_id
|
|
actionsExecuted := 0
|
|
actionsFailed := 0
|
|
var firstErrorMsg string
|
|
actionResults := make([]ActionExecutionResult, 0, len(rule.Actions))
|
|
for _, action := range rule.Actions {
|
|
actionForExecution := actionWithAutomationContext(action, eventName)
|
|
result, err := actionSvc.ExecuteWithResult(ctx, accountID, conversationID, actionForExecution, ActionSourceAutomation, rule.ID)
|
|
if err != nil {
|
|
applogger.L().Errorf("action %s failed for rule %d on conversation %d: %v", action.ActionName, rule.ID, conversationID, err)
|
|
actionsFailed++
|
|
if firstErrorMsg == "" {
|
|
firstErrorMsg = err.Error()
|
|
}
|
|
actionResults = append(actionResults, result)
|
|
// Continue executing remaining actions (Chatwoot pattern)
|
|
} else {
|
|
actionsExecuted++
|
|
actionResults = append(actionResults, result)
|
|
}
|
|
}
|
|
|
|
// Log execution for audit trail
|
|
status := ExecutionStatusSuccess
|
|
if actionsFailed > 0 && actionsExecuted > 0 {
|
|
status = ExecutionStatusPartial
|
|
} else if actionsFailed > 0 && actionsExecuted == 0 {
|
|
status = ExecutionStatusFailed
|
|
}
|
|
if logErr := logSvc.LogRuleExecutionWithResults(ctx, accountID, rule.ID, conversationID, eventName, status, actionsExecuted, actionsFailed, firstErrorMsg, actionResults); logErr != nil {
|
|
applogger.L().Warnf("failed to log automation execution for rule %d: %v", rule.ID, logErr)
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func actionWithAutomationContext(action Action, eventName string) Action {
|
|
params := map[string]interface{}{}
|
|
for key, value := range action.ActionParams {
|
|
params[key] = value
|
|
}
|
|
params["_event_name"] = eventName
|
|
return Action{ActionName: action.ActionName, ActionParams: params}
|
|
}
|
|
|
|
// loadConversation loads a conversation from the database.
|
|
func (s *AutomationRuleService) loadConversation(ctx context.Context, conversationID uint) (*ConversationForFilter, error) {
|
|
var conv ConversationForFilter
|
|
if err := s.db.DB().WithContext(ctx).
|
|
Table("conversations").
|
|
Where("id = ?", conversationID).
|
|
First(&conv).Error; err != nil {
|
|
return nil, err
|
|
}
|
|
return &conv, nil
|
|
}
|
|
|
|
// ConversationForFilter is a lightweight struct used for condition matching.
|
|
// It mirrors the conversation model fields needed by ConditionsFilterService.
|
|
type ConversationForFilter struct {
|
|
ID uint
|
|
AccountID uint
|
|
InboxID uint
|
|
ContactID uint
|
|
AssigneeID *uint
|
|
Status string
|
|
Priority string
|
|
ChannelType string
|
|
LastMessageAt *int64
|
|
Labels []string `gorm:"type:text;serializer:json"` // populated from conversation_labels join if needed
|
|
}
|
|
|
|
// TableName overrides for GORM queries.
|
|
func (ConversationForFilter) TableName() string { return "conversations" }
|