Workflows

# app/jobs/workflow_run_event_job.rb
class WorkflowRunEventJob < ApplicationJob
  def perform(event)
    # Saved with its reference before the run started, so an event that beats the 201 finds it.
    sync = event["reference"] && OrderSync.find_by(qube_reference: event["reference"])
    return unless sync # a run your app didn't start, such as one from the dashboard

    if event["event"] == "workflow_run.notification"
      # Data, not the run's state: a retry repeats its delivery_id. Never skip one by revision.
      return if sync.deliveries.exists?(delivery_id: event["delivery_id"])

      ImportPage.call(sync, event["name"], event["data"]) # e.g. page, with data["kind"]
      sync.deliveries.create!(delivery_id: event["delivery_id"])
    else
      ApplyRun.call(sync, event)
    end
  end
end

# Any body of the run: a webhook, the 201's or a decision's response `data` (which
# names the id and state `id` and `state`), or a GET.
class ApplyRun
  def self.call(sync, body)
    sync.with_lock do
      # Older than what you have: a repeat, or one that crossed a response.
      next if body["revision"] <= sync.qube_revision

      sync.update!(
        qube_run_id: body["run_id"] || body["id"],
        qube_revision: body["revision"],
        status: body["run_state"] || body["state"],
        question: body["prompt"], # nil once it's answered
        outcome: body["outcome"],
        output: body["output"],
        error: body.dig("error", "message") || body["cancel_reason"]
      )
    end
  end
end

# Starting it: the record first, then the run (Qube.api is the small client in Making requests).
# A reference of its own for every run, not the order's id: revisions count per run, so a second
# run for this order, found by a shared reference, would land on the first run's row and be skipped.
sync = OrderSync.create!(order: order, qube_reference: "order-#{order.id}-#{SecureRandom.uuid}",
                         qube_revision: 0)
run = Qube.api(:post, "/connections/#{connection_id}/workflow-runs",
               { workflow: "add_transaction", input: input,
                 reference: sync.qube_reference, webhook_url: webhook_url })
ApplyRun.call(sync, run)

{
  "event": "workflow_run.notification",
  "timestamp": 1758000000,
  "app_id": "018f0000-0000-7000-8000-000000000003",
  "sandbox": false,
  "connection_id": "018f0000-0000-7000-8000-000000000001",
  "workflow": "pull_changes",
  "workflow_id": "018f0000-0000-7000-8000-000000000004",
  "run_id": "018f0000-0000-7000-8000-000000000000",
  "reference": "nightly-sync-2026-10-11",
  "revision": 5,
  "run_state": "running",
  "links": {"run": "https://qubesync.com/api/v2/connections/…/workflow-runs/…", "decisions": "…", "cancel": "…", "ui": "…"},
  "delivery_id": "018f0000-0000-7000-8000-000000000005",
  "seq": 1,
  "name": "page",
  "state": "pull_lists / send",
  "from": {"workflow": "stream_query", "state": "send", "item_index": 0, "item_count": 3},
  "data": {"kind": "customer", "pass": 1, "page": 1, "more": true,
           "records": [{"list_id": "80000001-1738310400", "full_name": "Acme Co"}]}
}