EventSources

A flow’s source field names an EventSource. EventSources are the channels that put messages on the SQS queues; flow_worker matches each queued message to the flows whose source and event_type equal the envelope values and runs them.

EventSources are not part of the connector config blob. They live in their own lifecycle table and are authored separately via a dedicated API. The config-v3 schema reference does not cover them; this page does.

The API

EventSourceCreate body fields

The name regex allows lowercase letters, digits, and underscores only. Hyphens are not allowed.

Minimal create example

curl -X POST https://app.plugsync.com/api/connectors/<connector_id>/sources \
  -H "Authorization: Bearer <token>" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "shop_webhook",
    "type": "webhook",
    "config": {
      "inbound": {
        "url_template": "/api/webhooks/{connector_id}/shop_webhook",
        "verify_signature": "hmac_sha256",
        "signature_header": "x-shop-signature",
        "event_type_path": "$.event"
      }
    },
    "secret": "your-hmac-secret"
  }'

Source types

The six valid values of type:

Inbound channels

webhook

Receives POST /api/webhooks/{connector_id}/{name}. The signature is verified before the body is parsed; requests with a missing or wrong signature return 401.

config.inbound keys:

The secret is never stored in plaintext and never returned by the API. Supply it at create time via the secret field; rotate it with POST .../rotate-secret.

Example from _examples/inbound-webhook-to-hubspot.sources.json:

[
  {
    "name": "shop_webhook",
    "type": "webhook",
    "config": {
      "inbound": {
        "url_template": "/api/webhooks/{connector_id}/shop_webhook",
        "verify_signature": "hmac_sha256",
        "signature_header": "x-shop-signature",
        "event_type_path": "$.event"
      }
    }
  }
]

event_api

The Event API source (source_name: "event_api") is the push channel for enterprise integrations. External systems call POST /api/v1/events (authenticated with a connector API key) and Plugsync enqueues the payload directly. No config keys are required; create the source with config: {}.

The event type is taken from the request body, not from a config key. Flows matching this source receive $.payload.* populated from the fields the caller sent.

The source’s name must be the literal string event_api. POST /api/v1/events always stamps the outgoing SQS envelope with source_name: "event_api", regardless of what name you gave the EventSource row - there is no per-connector aliasing. POST .../sources rejects a type: "event_api" creation under any other name up front, with the error: “event_api sources must be named ‘event_api’; the ingest endpoint stamps source_name=‘event_api’, so any other name never matches a flow”. The same reasoning makes event_api sources non-renameable (PATCH .../sources/event_api with a name change 422s) and reserves event_api as a rename target for any other source. A flow’s source field must therefore also read literally "event_api" to ever match traffic from this endpoint.

poller

source_worker issues a GET to config.inbound.url on every tick, extracts items, and enqueues each one as a separate inbound event.

config.inbound keys:

The poller runs every TICK_INTERVAL_SECONDS (30 seconds). Items are emitted to inbound-sync.fifo.

scheduled

Emits an empty-payload tick on every tick of source_worker. Use this to drive periodic flows (nightly reconciliation, heartbeats, etc.).

config.inbound key:

The scheduled source emits to inbound-sync.fifo. The flow receives $.payload as {}.

The HubSpot journal source

Type hubspot, config: {}.

source_worker polls the HubSpot journal v4 API. It persists its read position in state.journal_offset so each poll starts from where the last one left off. Events are emitted to outbound-sync.fifo.

Event type format: {objectTypeId}.{action} lowercased. Examples:

Enriched payload shape: source_worker fetches the object from the HubSpot CRM v3 API (GET /crm/v3/objects/{object_type}/{object_id}) before enqueuing, so flows always have object.id available. It does NOT fetch every property by default. The CRM v3 GET only returns the properties named in its properties query param; source_worker names none unless the source’s enrich_properties config key lists them (see below). With enrich_properties empty or absent, HubSpot falls back to its own default minimal set: in practice this has shown up in production as just hs_object_id, createdate, hs_lastmodifieddate, none of a portal’s custom business properties and not even built-ins like hubspot_owner_id. A flow reading $.payload.object.properties.vat_number when vat_number was never named in enrich_properties gets None silently, on every event, with no error anywhere (issue #772).

The payload delivered to flow_worker (and available as $.payload.* in step expressions) is:

{
  "event": { "...raw journal event fields..." },
  "object": {
    "id": "12345",
    "properties": { "hs_object_id": "12345", "createdate": "...", "hs_lastmodifieddate": "..." }
  }
}

Steps that read HubSpot properties therefore use $.payload.object.properties.*, but only for properties actually requested, per enrich_properties below.

Reacting to which property changed (event.propertyChanges)

$.payload.event is the raw journal event HubSpot emitted, not an enrichment-time snapshot, and it carries HubSpot’s own propertyChanges: a dict {"<property_name>": "<new_value>", ...} listing exactly the properties THIS event reports as changed. On an update event that is the actual change set; on a created event it is the properties that were given a value at creation time; a delete event carries no properties at all (there is nothing left to report).

This is the field to gate a step on a specific property having changed, rather than on the object having changed at all - object.properties above is enriched independently and does not tell you which of its keys this particular event actually touched:

{"id": "sync_vat", "action": "post", "target": "erp_outbound",
 "entity": "company", "url": "/api/companies/vat-changed",
 "when": {"exists": "$.payload.event.propertyChanges.vat_number"},
 "body": {"vat_number": "$.payload.object.properties.vat_number"}}

The when gate above only lets the step run for journal events whose own change set included vat_number; an unrelated update to the same company (e.g. only city changed) skips it. See when conditions for the full condition DSL, and narrow_outbound_projection for a flow-level flag built on this same field.

Reacting to a deletion ({objectTypeId}.delete)

Deleting an object in HubSpot emits a journal event like any other change, so a flow reacts to it by declaring the mechanical event_type {objectTypeId}.delete (or a wildcard {objectTypeId}.* covering it):

{"name": "company_deleted_to_erp", "source": "hubspot_journal",
 "event_type": "0-2.delete",
 "steps": [{"id": "drop", "action": "plugin", "plugin": "erp_ops"}]}

A deletion is the one event whose object no longer exists, which changes what the payload can carry:

• object is always null. Enrichment fetches the object from the CRM API, and that GET necessarily 404s for a deleted record. Any step reading $.payload.object.properties.* gets None on a delete event.
• event.propertyChanges is empty: there is nothing left to report.
• $.payload.external_id carries the target-side id, resolved from the connector’s identity map by the HubSpot object id. It is added only on delete events, and it is the field a plugin step should key on: it is the only identifier of the record that survives the deletion. (The declarative delete step does not read it - it resolves the id itself; see the last limit below.)

{
  "event": {"objectTypeId": "0-2", "objectId": 4512, "action": "DELETE",
            "occurredAt": 1753372800000},
  "object": null,
  "external_id": "ERP-1042"
}

Limits worth knowing:

• Unmapped objects are skipped, silently by design. If the connector’s identity map holds no row for the deleted HubSpot id (the object was never synced by this connector, or its mapping was already removed), no event is emitted: there is nothing to delete on the target side. The skip is logged as journal_delete_unmapped_skip with the connector, object type and object id, so a genuinely lost mapping is still diagnosable.
• The subscription must cover DELETE. HubSpot only journals what the portal’s subscription asks for, and the subscription is derived from the flows: without a flow declaring {objectTypeId}.delete (or the wildcard), HubSpot never records the deletion in the first place. This holds even on a source that declares explicit triggers (see hubspot-triggers-to-rest): a trigger fires on a property change and a deletion carries none, so real deletions are always routed by the flows, never by a trigger, whatever the trigger’s own event_type is named.
• Adding that .delete flow costs no extra journal volume. A subscription’s property filter applies to UPDATE only: HubSpot delivers every CREATE and every DELETE regardless of the listed properties, and merely projects the entry’s propertyChanges onto them (measured on a staging portal for all three actions). So a DELETE joining a property-filtered UPDATE subscription on the same object type keeps that filter intact, and the per-portal journal stays as narrow as it was.
• The fallback poller does not report deletions. The per-portal journal consumer is the only path that emits them. A connector using a poller source instead is polling a plain REST endpoint, which has no notion of a HubSpot deletion at all.
• A delete step resolves its own id, and needs pointing at the event. The declarative delete action against a REST target does not read $.payload.external_id: it resolves the external id from the identity map itself, keyed by the HubSpot id it reads through the entity schema’s identity_source_path. Its default chain ($.payload.object.id, then $.payload.objectId) resolves neither field on a delete event, since object is null and the id lives under event, so the target schema must declare "identity_source_path": "$.payload.event.objectId" explicitly. Forgetting it is caught at publish: the response carries the warning delete_rest_identity_source_path_unresolvable, naming the path to declare. A plugin step has no such constraint: it receives the whole payload, external_id included.

Example from _examples/outbound-hubspot-journal-to-api.sources.json:

[
  {
    "name": "hubspot_journal",
    "type": "hubspot",
    "config": {}
  }
]

enrich_properties

config.enrich_properties is a list of HubSpot internal property names to fetch when enriching a journal event. Any custom property (or non-default built-in) a flow’s steps read via $.payload.object.properties.<name> must be named here: nothing is inferred from the flows automatically.

Full example: a company sync flow that pushes vat_number, pec, and hubspot_owner_id out to an external ERP on every HubSpot company update.

[
  {
    "name": "hubspot_journal",
    "type": "hubspot",
    "config": {
      "enrich_properties": ["vat_number", "pec", "hubspot_owner_id"]
    }
  }
]

{
  "name": "hubspot_company_to_erp",
  "source": "hubspot_journal",
  "event_type": "0-2.update",
  "steps": [
    {
      "id": "push_erp",
      "action": "post",
      "target": "erp_outbound",
      "entity": "company",
      "url": "/api/companies/upsert",
      "body": {
        "vat_number": "$.payload.object.properties.vat_number",
        "pec": "$.payload.object.properties.pec",
        "owner_id": "$.payload.object.properties.hubspot_owner_id"
      }
    }
  ]
}

Set or change it after creation with:

curl -X PATCH https://app.plugsync.com/api/connectors/<connector_id>/sources/hubspot_journal \
  -H "Authorization: Bearer <token>" \
  -H "Content-Type: application/json" \
  -d '{"config": {"enrich_properties": ["vat_number", "pec", "hubspot_owner_id"]}}'

It is also editable from the dashboard: open the connector’s Sources panel, select the hubspot source, and use the Enrich properties field (add/remove chips, then Save trigger routing, the same save button also persists enrich_properties).

Gotcha on a portal shared by multiple connectors: source_worker polls the journal once per HubSpot portal, not once per connector, and enriches with the union of every connector’s enrich_properties bound to that portal. A single misconfigured connector cannot narrow what another connector on the same portal receives, but it also means the underlying HubSpot GET always requests the union: harmless for correctness, just worth knowing when reasoning about the API call’s properties param from a HubSpot rate-limit or audit-log angle.

Publish-time hint: GET /api/connectors/{connector_id}/preflight (warning-only draft validation, rendered in the dashboard’s Review-changes drawer) flags a flow that reads $.payload.object.properties.<name> for a <name> missing from its source’s enrich_properties with a warning (enrich_properties_gap). It is informational only, it never blocks publish, but it is the fastest way to catch this before the field silently comes back empty in production.

Outbound targets

Type rest. Not polled by source_worker; used as the target of a post or upsert step to send data to an external API.

config.outbound keys:

At runtime flow_worker resolves the credential and injects the appropriate Authorization (or X-API-Key) header on every outbound request, merged with default_headers if set. The credential value is never logged.

Example from _examples/outbound-hubspot-journal-to-api.sources.json:

{
  "name": "crm_api",
  "type": "rest",
  "config": {
    "outbound": {
      "base_url": "https://api.example-crm.com",
      "credential_name": "crm_api_key",
      "auth_type": "bearer"
    }
  }
}

OAuth2 client_credentials destinations

Some target APIs sit behind an OAuth2 client_credentials token endpoint (e.g. Keycloak) instead of a static API key. Create a credential of type oauth2_client_credentials with token_url, client_id, client_secret, and an optional scope:

POST /api/credentials
{
  "name": "crm_keycloak",
  "credential_type": "oauth2_client_credentials",
  "data": {
    "token_url": "https://idp.example.com/realms/x/protocol/openid-connect/token",
    "client_id": "svc-client",
    "client_secret": "svc-secret",
    "scope": "api.write"
  }
}

Bind it to the source the same way as any other credential:

{
  "name": "crm_api",
  "type": "rest",
  "config": {
    "outbound": {
      "base_url": "https://api.example-crm.com",
      "credential_name": "crm_keycloak",
      "default_headers": {"X-TenantId": "MASTER"}
    }
  }
}

flow_worker POSTs token_url form-urlencoded (grant_type=client_credentials, plus scope if configured) to obtain a Bearer access token, and caches it in-process until it is close to expiry (from the response’s expires_in) — the token endpoint is hit once per credential per TTL window, never once per outbound call. A token endpoint failure is classified transient (429/5xx, worth a retry), permanent (other 4xx), or fatal (401/403 — the configured client_id/client_secret is wrong), the same taxonomy used elsewhere in the sync engine for outbound HTTP errors. client_secret is stored Fernet-encrypted like every other credential; neither the client secret nor the fetched access token is ever logged.

A post step that uses this source:

{
  "id": "push_api",
  "action": "post",
  "target": "crm_api",
  "entity": "company",
  "url": "/api/companies/upsert",
  "body": "$.payload.object.properties"
}

The step url is appended to base_url, so the actual request goes to https://api.example-crm.com/api/companies/upsert.

Referencing a source from a flow

A flow references a source in two places:

1. flow.source - the EventSource name whose events trigger the flow.
2. step.target (on post and upsert steps targeting non-HubSpot endpoints) - the EventSource name of the outbound rest source to POST to.

Both must equal the name of an existing EventSource on the same connector. Deleting a source that is referenced by an active flow returns 422; the error lists the active version number and every referencing flow/step.

Renaming a source (PATCH with {"name": "new_name"}) follows the same rule: it is honored only when no flow of the active published version references the source (those match events by name at runtime), and returns 422 listing the blockers otherwise. References inside the draft are cascaded to the new name automatically: flow.source, step.target, step.from_schema/step.to_schema, entity schema keys, and captured payload samples. event_api sources cannot be renamed, and the reserved names event_api and hubspot are refused as new names. Note for webhook sources: the delivery URL contains the source name, so the external system must switch to the new URL after the rename.

To swap a source atomically while staying inside the plan’s source limit, publish the repointed flows with the replacement declared in new_sources and the source it replaces in delete_sources on the same POST /publish call: the deletion happens after the new version goes active, and the limit is checked on the final set of the operation.

For the full flow authoring syntax see authoring-a-flow.md.