Guides / Workflow Builder

Build a trading-rewards workflow from event ingest

A trading-rewards pipeline takes broker trade events, aggregates them per trader per cycle, and pays a reward for the volume. The workflow's job is the last part: receive one aggregated batch, calculate the reward for each row, and apply it.

The API builds, tests, deploys and monitors that workflow, including the reward calculation: a Code step is a normal step type here, and it can be created, changed, tested and deployed like any other. Gathering the trades, aggregating them, and the decision of what the policy should be are your side.

What you end up with

An event ingest workflow with a batch id path and per-row record id paths.
A read step that resolves each trader from the aggregated row.
A reward write step, in a branch that only runs when the row is eligible.
A live workflow, one canary batch, one run read back, and the per-row result.

Where the reward calculation goes

Most real reward policies are a calculation: exact instrument, category rate, rounding, a cap, a skip reason per row. There are two places to put it, and both are fully supported:

Part	Where it is built
Collector, aggregation, deterministic identities	Your system
Reward calculation	A Code step in the workflow, or your own system before the event is sent
Ingest trigger, read step, router, bulk-update or write step	The API
Deploy, confirming a run, run reads	The API

A Code step in the workflow keeps the policy next to the payout. Create it with POST /{workflowId}/nodes and type: "code", with options.inputs bound to the row ({{<triggerId>.csv}}, {{<readStepId>.Total XP}}) and the source in options.sourceCode.code. Its result is keyed by the names it returns, so the bulk step references {{<codeStepId>.csv}}. It can be changed, duplicated, imported and tested like any other step, and a workflow containing it deploys normally.

What a Code step costs you in review: a dry run does not execute source, so it shows the resolved inputs and the source but not the output, and the step after it cannot run in that dry run (it fails with UNRESOLVED_SIMULATED_REFERENCE). Live-test the Code step on its own with a seed you chose, read the output, and only then dry-run the rest around it. Details are in Code steps.

Your own system keeps the workflow Code-free: send the calculated amount per row, and the workflow's job is delivery. That is simpler to review, and it puts the policy in code you already test.

Either is fine. Choose by where you want the policy to live, not by what the API accepts.

1. Fix the delivery identity and the event contract

Before any call, decide:

The delivery identity. One field per pipeline: a stable broker customer id when the broker supplies one and registration writes the same value, otherwise a normalised email. Never both inside one pipeline.
The aggregation unit. One row per trader per cycle, with the volume facts preserved per instrument. The workflow does not group; grouping has to happen before the batch arrives.
The batch id path and the per-row record id path. These are the two deduplication layers, and they are configuration on the trigger.
The reward rule and its owner. Rates, rounding and caps belong to the workflow's configuration, not to the sender's payload.
The cutoff. Which events are eligible. If a backlog exists before launch, decide explicitly whether it is paid or discarded.

2. Read the catalog, then create the workflow

Take every field key from the catalog rather than guessing one: GET /catalog returns platformActionFields.get and platformActionFields.post, each entry carrying the key the step stores, the label the platform stores beside it, and whether the field is builtin or custom.

GETList step types and trigger typescURL

curl --request GET \
  --url "https://api.returning.ai/v1/workflow-builder/data-workflows/catalog" \
  --header "Authorization: Bearer <API_KEY>"

Create the workflow with an event ingest trigger. The batch id path is read from the root of the delivered event, and record id paths are read from each row:

POSTCreate a workflowcURL

curl --request POST \
  --url "https://api.returning.ai/v1/workflow-builder/data-workflows" \
  --header "Authorization: Bearer <API_KEY>" \
  --header "Content-Type: application/json" \
  --data '{"name":"Trading rewards","trigger":{"type":"event-ingest","name":"Trades","options":{"batchIdPath":"metadata.batch_id","recordIdPaths":["trade_id"]}}}'

Readback: 201, status: draft, trigger.type: event-ingest, the two paths stored under trigger.options, and a revision.

The trigger id (trigger._id) is what the next steps reference. The production ingest id a sender needs is the trigger's id; give the sender that value, plus the endpoint it posts to.

3. Add the steps

A read step that resolves the trader from the aggregated row. total_xp and total_coins are the read keys the catalog returns; a custom field would appear in the same list:

POSTAdd a stepcURL

curl --request POST \
  --url "https://api.returning.ai/v1/workflow-builder/data-workflows/65f1a2b3c4d5e6f708192a3b/nodes" \
  --header "Authorization: Bearer <API_KEY>" \
  --header "Content-Type: application/json" \
  --header "If-Match: 162af81f5de4ff2b557a4087de9a9a73a0ed4e621f737787e35a93614cb50d1c" \
  --data '{"type":"platform-action","name":"Read trader","position":{"after":"trigger"},"options":{"action":"get-user-fields","selectUser":{"type":"email","value":["{{65f1a2b3c4d5e6f708192a41.email}}"]},"fields":[{"label":"Total XP","field":"total_xp"},{"label":"Total Coins","field":"total_coins"}]}}'

A router that keeps ineligible rows out of the reward step. The condition reads the earlier step's output, so a trader the read step could not resolve never reaches the reward:

POSTAdd a stepcURL

curl --request POST \
  --url "https://api.returning.ai/v1/workflow-builder/data-workflows/65f1a2b3c4d5e6f708192a3b/nodes" \
  --header "Authorization: Bearer <API_KEY>" \
  --header "Content-Type: application/json" \
  --header "If-Match: 68f78253046393414dec404e5442c140bb5db0393a7d1c389ff57bba71585501" \
  --data '{"type":"router","name":"Trader eligible?","position":{"after":"65f1a2b3c4d5e6f708192a3c"},"options":{"executePaths":"first_matching","branches":[{"branchName":"XP readable","branchType":"CONDITION","conditions":[{"id":"group-1","operator":"AND","nestedGroups":[{"id":"nested-1","operator":"AND","conditions":[{"id":"condition-1","key":["{{65f1a2b3c4d5e6f708192a3c.Total XP}}"],"type":"number","operator":"greater_than_or_equal","value":["0"]}]}]}]},{"branchName":"Otherwise","branchType":"FALLBACK","conditions":[]}]}}'

The reward step. xp is the write key from the catalog's post list. Two shapes are supported:

A per-trader write with post-user-fields, which is simple and needs no file:

POSTAdd a stepcURL

curl --request POST \
  --url "https://api.returning.ai/v1/workflow-builder/data-workflows/65f1a2b3c4d5e6f708192a3b/nodes" \
  --header "Authorization: Bearer <API_KEY>" \
  --header "Content-Type: application/json" \
  --header "If-Match: 9ab88dc5d9fdaada545d24e2d76d58cd758c7b99eb905df867d90cdda294e12f" \
  --data '{"type":"platform-action","name":"Award volume XP","position":{"routerNodeId":"65f1a2b3c4d5e6f708192a3e","branchIndex":0},"options":{"action":"post-user-fields","selectUser":{"type":"email","value":["{{65f1a2b3c4d5e6f708192a41.email}}"]},"fields":[{"label":"XP","field":"xp","logic":"add","value":["{{65f1a2b3c4d5e6f708192a41.volume_xp}}"]}]}}'

A bulk update from a CSV the workflow builds, which is the shape a high-volume pipeline needs:

POSTAdd a stepcURL

curl --request POST \
  --url "https://api.returning.ai/v1/workflow-builder/data-workflows/65f1a2b3c4d5e6f708192a3b/nodes" \
  --header "Authorization: Bearer <API_KEY>" \
  --header "Content-Type: application/json" \
  --header "If-Match: 9ab88dc5d9fdaada545d24e2d76d58cd758c7b99eb905df867d90cdda294e12f" \
  --data '{"type":"users-bulk-update","name":"Apply volume rewards","position":{"routerNodeId":"65f1a2b3c4d5e6f708192a3e","branchIndex":0},"options":{"payloadFormat":"csv","processingMode":"standard","identifier":"email","logName":["Trading rewards batch"],"csv":["{{65f1a2b3c4d5e6f708192a41.csv}}"]}}'

The bulk step queues a separate job, so the workflow's run status is not the reward's outcome. Read the job with the platform's bulk update endpoints (Get bulk update job status and Get bulk update job details), and keep its log name unique per batch so a job can be matched to the batch that produced it.

4. Validate and dry-run

POSTValidate a stored workflowcURL

curl --request POST \
  --url "https://api.returning.ai/v1/workflow-builder/data-workflows/65f1a2b3c4d5e6f708192a3b/validate" \
  --header "Authorization: Bearer <API_KEY>" \
  --header "Content-Type: application/json" \
  --data '{}'

POSTTest a whole workflowcURL

curl --request POST \
  --url "https://api.returning.ai/v1/workflow-builder/data-workflows/65f1a2b3c4d5e6f708192a3b/test" \
  --header "Authorization: Bearer <API_KEY>" \
  --header "Content-Type: application/json" \
  --data '{"mode":"dry-run","triggerPayload":{"metadata":{"batch_id":"batch-0001"},"data":[{"email":"trader@example.com","trade_id":"T-1","volume_xp":"250"}]}}'

Readback: the trace names the branch taken, and the reward step is simulated with the resolved member, field and amount. sideEffects: "none" proves nothing was paid.

If the workflow carries a Code step, the dry run is a partial walk: the Code step is simulated with its source and no output, and every step after it fails with UNRESOLVED_SIMULATED_REFERENCE. Test the Code step live on its own first, then read the trace here as evidence for everything around it.

Run at least two dry runs: one eligible row and one row that must not pay.

5. Deploy and prove one canary cycle

POSTDeploy a workflowcURL

curl --request POST \
  --url "https://api.returning.ai/v1/workflow-builder/data-workflows/65f1a2b3c4d5e6f708192a3b/deploy" \
  --header "Authorization: Bearer <API_KEY>" \
  --header "If-Match: 68f78253046393414dec404e5442c140bb5db0393a7d1c389ff57bba71585501"

Then send one bounded batch from your collector, with a fresh batch id and fresh record ids, and read it back in three places:

GET /{workflowId}/runs for the run, then GET /runs/{runId} for the per-step result.
The bulk update job, when the reward step is a bulk step: status, then per-row detail, then the member's own field value.
Your own ledger: the batch should be acknowledged once, and a replay of the same batch must not pay again.

success on the run is not the reward. The member's field value, or the bulk job's per-row result, is the reward. Say "paid" only when you have one of those.

6. Failure handling

What you see	What it means	What to do
Ingest rejected with a batch id path message	The batch id was not at the configured path in the root of the event	Fix the sender's envelope; keep the same batch id only for a byte-identical retry.
Ingest `skipped`	A duplicate batch id, or every row was a duplicate	Do not resend under a new batch id unless the data really is new.
Run `error`	A step failed; the run detail names it	Fix the step, then decide about the batch.
Run `success`, member unchanged	The reward step was skipped or the amount was zero	Read the run detail and the branch that ran.
Bulk job `Failed` with every row `success`	The job failed as a whole; nothing was applied	Check the CSV header against the community's identifier field, then resend as a new batch.
Two awards for one row	The row reached the reward step twice	Fix the record id path or the sender's aggregation before the next cycle.

Retrying a failed run re-executes the whole workflow, so a retry after a partial payment can pay twice. Read the member's value before retrying. A new batch or aggregate id alone is not a safe financial correction: on a trigger with recordIdPaths, a new batch to the same workflow only pays rows whose record ids are new (a duplicated workflow is a new workflow, so its first batch pays every row again), so correct an amount already paid through your approved correction or compensation path, as described in Trading volume and rewards.

7. Cleanup

Undeploy, then delete, exactly as in the webhook playbook. Keep the batch ids and job names from your canary in your own record: they are the evidence that the pipeline worked.

What the API does not cover

The collector, its cursor and its overlap window.
Aggregation by the delivery identity, and the choice between a customer id and an email.
Judging whether the reward policy is right. A Code step runs what you wrote; it does not tell you whether the rule is the one you meant.
The reward policy decision itself. Sending a calculated amount from your own system is the API-only alternative.
A recurring schedule for the collector.

Endpoints on this page

List step types and trigger types: GET /v1/workflow-builder/data-workflows/catalog
Create a workflow: POST /v1/workflow-builder/data-workflows
Add a step: POST /v1/workflow-builder/data-workflows/{workflowId}/nodes
Validate a stored workflow: POST /v1/workflow-builder/data-workflows/{workflowId}/validate
Test a whole workflow: POST /v1/workflow-builder/data-workflows/{workflowId}/test
Deploy a workflow: POST /v1/workflow-builder/data-workflows/{workflowId}/deploy
Get bulk update job status: GET /v1/users/bulk-update/{id}/status
Get bulk update job details: GET /v1/users/bulk-update/{id}/details
List the runs of a workflow: GET /v1/workflow-builder/data-workflows/{workflowId}/runs
Get one run: GET /v1/workflow-builder/data-workflows/runs/{runId}