Trigger / Workflow Automation Engine

Overview

The trigger engine is the heart of the entire automation system. The idea is simple: "if event X happens to a customer and they satisfy condition Y, do Z." These rules live in the trigger_rule table and can be attached to a workflow.

The trigger-worker profile runs consumers on eight queues, one per source event type, but they all converge on the same pipeline: filter rules → check dedup and cooldown → evaluate conditions → dispatch the action, or schedule it for later.

Business Flow

The eight source types

Queue	Source type	Where the event comes from
attribute_change	attribute_change, batched	pg-listener via pg_notify('attribute_changed') and the import path
attribute_change_realtime	attribute_change, per user	the realtime path
audience_membership_trigger	audience_membership	the audience consumer, on member add or removal
friend_track_event_trigger	friend_track_event	pg-listener via friend_track_event_inserted
form_submitted_trigger	form_submitted	pg-listener, when client-api saves a form
campaign_click_trigger	campaign_click	pg-listener via a trigger on the tracking_log table
scheduled_trigger_fire	scheduled	the cron scanner on the cron-scheduler profile
scheduled_action_fire	—	the cron scanner, for actions that have come due

Pipeline for a single message

1. Source-type gate — HasActiveSourceType(lineOaId, organizationId, sourceType) consults a cache (60-second TTL) to check whether the OA has any enabled rule of this type. If not, the message is acknowledged immediately without touching the database, which stops event floods from doing pointless work.
2. Load rules — getTriggerRules (60-second cache), then filter down to rules matching both the source type and the target, for example requiring sourceConfig.audienceId to match the audience that changed.
3. isKeyRelevant — for attribute changes, skip immediately if the changed key has nothing to do with the rule's conditions.
4. Dedup and cooldown — shouldSkipByDedup uses Redis to stop the same rule from firing repeatedly for the same user, honouring frequency (for example, once ever) and cooldownSeconds.
5. Evaluate conditions — evaluateConditions and evaluateCondition
   • a full set of comparison operators is supported, including day-count date comparisons where roundDiffDays mirrors JavaScript's Math.round behaviour so results match the legacy system
   • split tests are supported through evaluateSplitTest, which partitions users deterministically from their userId
6. Execute the action — executeAction(rule, user, triggerDepth)
   • if actionConfig.delay is set, the action is not run immediately; ScheduleAction records it into the scheduled_action table instead (see Delayed / Scheduled Actions)
   • otherwise, an ActionExecutePayload is published onto action_execute, or onto web_request_execute when actionType = web_request
   • the constant maxTriggerDepth = 5 guards against an action that triggers further triggers and spirals into an infinite loop
7. Write the log — record the rule id, user, triggerOn, status, and action type/config into trigger_log.
8. Most handlers always return nil because the service handles its own errors and logging; only a malformed payload reaches the DLQ.

scheduled_trigger_fire

The cron finds rules with source_type='scheduled' that have come due, evaluating the schedule per rule in the Asia/Bangkok timezone, and publishes them onto this queue. The consumer then expands each into per-user work in batches of 500 (scheduledFireBatchSize) and runs them through the same pipeline.

Key Files & Functions

• internal/trigger/consumer.go — Consumer.Register() binds all eight queues, with the handlers HandleAttributeChange, HandleRealtimeAttributeChange, HandleAudienceMembershipTrigger, HandleFriendTrackEventTrigger, HandleFormSubmittedTrigger, HandleCampaignClickTrigger, HandleScheduledTriggerFire, and HandleScheduledActionFire
• internal/trigger/service.go
  • TriggerService.EvaluateTriggers(), EvaluateTriggersForBatch(), processAttributeRule()
  • HasActiveSourceType(), getTriggerRules(), shouldSkipByDedup(), isKeyRelevant()
  • evaluateCondition(), evaluateConditions(), evaluateSplitTest(), getUserValue()
  • executeAction(), getLineUser(), mapLineUser()
  • the constants cacheTTLSecs = 60, maxTriggerDepth = 5, scheduledFireBatchSize = 500
• internal/trigger/service_events.go
  • EvaluateAudienceTriggers(), EvaluateFriendTrackTriggers(), EvaluateFormSubmittedTriggers(), EvaluateCampaignClickTriggers()
  • ProcessScheduledTriggerFire(), ProcessScheduledActionFire(), handleScheduledActionExpired()
• internal/trigger/repository.go — TriggerRuleRepository and TriggerLogRepository.LogTrigger()
• internal/payloads/payloads.go — ActionExecutePayload, the cross-domain contract
• cmd/worker/main.go — runTriggerWorker(), which cross-wires TriggerService and ScheduledActionService
• Queues: consumes all eight queues above (profile trigger-worker) and publishes to action_execute and web_request_execute

Connections to Other Services

• Receives jobs from: pg-listener (five channels), audience processing, line user CSV import, and the cron scanner on the cron-scheduler profile
• Tables: trigger_rule (the rules themselves), trigger_log (firing records), line_user (the user data conditions are evaluated against), scheduled_action (delayed actions), and workflow (rule grouping)
• Redis: caches for rules and the source-type gate (60-second TTL), plus dedup and cooldown keys per (rule, user) pair
• RabbitMQ: publishes onward to Workflow Action Execution and External API Calls from Workflows
• ENV toggles: ENABLE_TRIGGER_EVALUATION and ENABLE_TRIGGER_SOURCE_TYPE_CACHE
• CMS side: the cms-api-go workflow and trigger-rule domains are what create and edit these rules