flows[] reference

A flow is a (source, event type) pair plus an ordered sequence of steps. When the flow worker receives a message it loads the connector’s published config, matches every flow whose source and event_type equal the message envelope, and runs each matching flow’s steps in order.

Fields

Validators

• Step ids must be unique within a flow. Violation: step ids within a flow must be unique.

Worked example: event_type_from

When the inbound payload carries the event type itself (Compass does this with {"event": "upsert_company", ...}), you can record where the type lives via event_type_from:

{
  "name": "compass_company_upsert",
  "source": "compass_inbound",
  "event_type": "upsert_company",
  "event_type_from": "$.event",
  "steps": [
    {"id": "upsert_hubspot", "action": "upsert", "target": "hubspot",
     "entity": "company", "data": "$.payload"}
  ]
}

Today the producer (the inbound API endpoint or source_worker) is responsible for placing the resolved event type on the SQS envelope as event_type. The flow worker then matches flow.event_type == envelope.event_type (strict equality) and never reads event_type_from at runtime. The field is persisted on the flow definition and projected into event_flow_manifests as a forward-compat hint; no current producer resolves the JSONPath automatically.

Dispatch model

The flow worker consumes from two SQS FIFO queues (inbound-sync.fifo and outbound-sync.fifo). For each message it:

1. Loads the connector’s active config from DynamoDB (the worker hot-read replica written by the outbox dispatcher; see draft-publish.md).
2. Walks config.flows[], picks every flow where source == msg.source_name and event_type == msg.event_type.
3. Runs the matched flows in declared order.

Steps within a flow are sequential; flows themselves are independent and order between them is not guaranteed beyond declaration order.