System architecture

EventLogger Rules is one AWS SAM stack (template.yaml) per account with two Lambda functions built from the same src/ code: ApiFunction serves HTTP requests, and WorkerFunction handles every background trigger (Kinesis, the CRM SNS topic and the watchdog schedule).

Architectural principles

Reuse, don’t own, shared infrastructure

The DynamoDB tables, the Kinesis stream and the SNS topics already exist and are shared with the rest of EventLogger. The template references them by name and never creates them, so deleting the stack can’t touch live data.

One switch for every consumer

The Kinesis mapping, the watchdog schedule and the CRM SNS subscription are all controlled by the EnableTriggers parameter. A stack can be deployed with a working API while consuming nothing, which is how a new deployment runs side by side with an old one. See EnableTriggers.

Nothing happens at import time

Settings, AWS clients and the Flask app are created on first use behind lru_cache accessors. The one exception is pyb-flask-utils, which loads common/crm when its secret manager module is imported.

Requests and background work are separate

The API keeps a 30 s timeout and a role without Kinesis or Firehose access. Background work gets 900 s and 1024 MB, and its errors and throttles never show up as API errors.

Component topology

flowchart LR
  subgraph clients ["Clients"]
    SDK(["SDKs and tracking pixels"])
    CRM(["CRM"])
  end

  subgraph stack ["SAM stack eventlogger-rules-dev / -prod"]
    API["RulesApi + ApiFunction"]
    WORKER["WorkerFunction"]
  end

  subgraph old ["Old-rules stack (release-P312)"]
    STAGE1["Stage 1: preprocess"]
  end

  SNS[("SNS em_events")]
  KIN[("Kinesis em_events")]
  CRMTOPIC[("SNS em_events_create_event")]
  SCHED(["Watchdog schedule"])
  DDB[("DynamoDB tables")]
  EL(["EventLogger API"])
  EXT(["Slack, email, tasks, suppliers"])

  SDK -->|"/capture-message, /track"| API
  CRM -->|"rules API"| API
  API -->|"stage 0"| SNS
  API --> DDB
  SNS --> STAGE1 --> KIN
  KIN -->|"batch of 500"| WORKER
  CRMTOPIC --> WORKER
  SCHED --> WORKER
  WORKER -->|"rule actions"| EXT
  WORKER --> DDB
  WORKER -->|"CRM events"| EL

Stage 1 of the event pipeline (SNS to DynamoDB and Kinesis) belongs to the old-rules stack, which keeps running. This stack starts at the Kinesis stream: it doesn’t write to Kinesis.

Functions

ApiFunction (handlers/api.py) is behind RulesApi (ANY / and ANY /{proxy+}) with a 30 s timeout. API Gateway REST events are translated to WSGI for the Flask app. CORS, including preflight, is answered by flask-cors for https://payability.com and its subdomains only, see Security.

WorkerFunction (handlers/worker.py) runs with 900 s and 1024 MB. Its entry point looks at the shape of the event and hands it to the matching code:

Trigger Recognised by Runs

Kinesis em_events[_staging], batch size 500

Records[0].eventSource == "aws:kinesis"

Stage 2 of the pipeline (pipeline/stage_2_postprocess.py)

SNS em_events_create_event[_staging]

Records[0].EventSource == "aws:sns"

handlers/crm_event.py: forwards allowed CRM events to EventLogger as instance 8001

Schedule, rate(15 minutes) in production and rate(2 minutes) in staging

source == "aws.events" and detail-type == "Scheduled Event"

handlers/watchdog.py

Any other event raises an error. The Kinesis mapping and the SNS subscription exist, and the schedule is ENABLED, only with EnableTriggers=true.

Data stores

Store Use

em_event_rules_v2

Event rules: filters, actions, watchdog settings, last execution

em_event_rule_logs_v2

Rule execution history, one item per matched rule and event

em_action_template

Reusable action templates

em_events

Stored events, written by stage 2 and read by the tracking endpoints

SNS em_events

Entry point for new events (stage 0)

Firehose em_events_es_staging

Copy of every event, staging only

EventBridge bus pyb-notifications

Slack notifications from the Slack action

Tables carry the staging suffix in the staging account. Items keep the format the production tables already use: integer timestamps, a marshalled data attribute and created_by* fields.

Search is disabled in both environments (ES_EVENTS_URL is empty), so /events/stream answers 503.

Request path

sequenceDiagram
  autonumber
  participant C as Client
  participant G as api.payability.com
  participant L as ApiFunction
  participant F as Flask app
  participant I as Internal API
  participant D as DynamoDB

  C->>G: GET /eventlogger_v2/event_rules/ (X-Session-Token, X-InstanceId)
  G->>L: REST proxy event (path keeps the base path)
  L->>F: WSGI request
  F->>I: GET /token/validate
  I-->>F: login_id
  F->>F: check instance authorization
  F->>D: query em_event_rules_v2
  D-->>F: items
  F-->>L: JSON response
  L-->>G: statusCode, headers, body
  G-->>C: 200

The base path stays in the request path on the custom domain, which is why production routes are registered under /eventlogger_v2. On an execute-api hostname the adapter adds the stage to the script root instead, so generated URLs work in both places.

Observability

  • Logs go to CloudWatch Logs, one log group per function: /aws/lambda/<stack>-api and /aws/lambda/<stack>-worker. The template creates them and points each function at its group, so their retention (90 days in production, 30 in staging) is set there and they are deleted with the stack. LOG_LEVEL is WARNING in both environments.

  • Sentry is initialised for production and staging, tagged with app=eventlogger-new-rules and component (api or worker). Worker events also carry trigger (kinesis, crm_event or watchdog).

  • X-Ray tracing is active on every function.