Event processing and rules

Pipeline stages

Events move through three stages. This application owns stage 0 (through the API) and stage 2; stage 1 belongs to the old-rules stack.

Stage Input Work Owner

0, initial

/capture-message, /track

Validate and publish the event to SNS em_events

ApiFunction

1, preprocess

SNS em_events

Store in DynamoDB, forward to Kinesis em_events

old-rules stack

2, postprocess

Kinesis em_events

Run event rules, copy to Firehose (staging only), store in em_events

WorkerFunction

Stage 2 in detail

sequenceDiagram
  autonumber
  participant K as Kinesis em_events
  participant S as WorkerFunction
  participant R as em_event_rules_v2
  participant A as Action registry
  participant H as em_event_rule_logs_v2
  participant E as em_events

  K->>S: batch of up to 500 records
  loop each record
    S->>R: scan all rules (no cache)
    loop each active rule whose filters match
      alt supplier status allows the rule
        S->>A: run each action (email, Slack, task, supplier edit)
        A-->>S: per-action log entry
        S->>H: write history item with executed actions
        S->>R: save last execution
      else supplier status excludes the rule
        S->>H: write history item without actions
      end
    end
    S->>E: store the event
  end

Each record runs every action of the stage in order (rules, Firehose outside production, DynamoDB). A failing action or an undecodable record is logged and skipped, so one bad event never blocks the shard. The function only fails the whole batch on an unexpected error outside that loop, in which case Lambda retries the batch.

Rules are read on every event, so edits in the CRM apply to the next event without a deploy.

Rule matching

A rule matches an event when:

  1. the rule is active, and

  2. every filter of the rule matches. A filter has a key (a selector into the event), an operator, a type and a value.

Operator Type Matches when

eq

keyword

the event value equals the rule value

contains

text

the rule value is contained in the event value, case-insensitive

/constants/rules lists the selectors and operators the CRM can offer, and /simulate/filters and /simulate/rules evaluate filters or rules against a sample event without running actions.

Supplier status gate

When the event targets a supplier, the rule runs only if the supplier’s current status (read from the internal API) is one of the rule’s executable_supplier_statuses. If the supplier can’t be found, the rule runs. A rule that matches but is excluded by status still gets a history entry, without executed actions.

Actions

Slug Name What it does

send_email

Send email

Renders a template through the messaging service and sends it, optionally to the supplier and CC receivers

send_to_slack

Send to Slack

Publishes a notification event to the EventBridge bus pyb-notifications

create_task

Create task

Creates a CRM task through the action tools API

edit_supplier

Edit supplier

Updates supplier fields through the internal API

Every executed action appends an entry to the rule’s history item, including errors, so a failing action is visible in /event_rules_history/…​ without searching logs.

Watchdog

Rules can have a watchdog interval. On every scheduled run, WorkerFunction scans all rules and sets each rule’s alert flag to true when the rule has an interval and hasn’t fired within it, and to false otherwise. The CRM shows the flag on the rule.

CRM events

WorkerFunction also receives events published to em_events_create_event (EventBridge-style detail inside an SNS message). It accepts only these values, as defined in the EventLogger Python SDK:

  • event types: analitics (sic), event, underwriting-event, messaging-result, messaging-log, cli-event

  • target types: supplier, marketplace, compliance, crm-task, general

It then captures the event in EventLogger through the SDK as instance 8001. From there it enters the pipeline at stage 0 like any other event.