Class: Nexo::WorkflowJob

Inherits:
ActiveJob::Base
  • Object
show all
Defined in:
lib/nexo/workflow_job.rb

Overview

Executes a queued Workflow run on the host's ActiveJob adapter (Spec 11 R1). Subclasses ::ActiveJob::Base — the host-agnostic base — rather than the host's ApplicationJob, so the job is self-contained. It carries ONLY the run id; the payload lives on the run record (never passed through job args), so no secrets travel through the queue.

Reconstitutes the workflow class and calls Workflow.execute, which the sync Workflow.run shares — so an async run reaches the same done/failed lifecycle, event log, and status notifications. A crashed/retried run is not auto-resumed: a retried job re-runs #call from scratch (Nexo adds no retry_on).

With a second resume_input argument (Spec 13, from Workflow.resume_later) it dispatches the durable resume path instead: it re-enters Workflow.resume (which re-finds the run, guards it is still "suspended" so a race that already resumed fails cleanly, and re-runs #call from the top with the resume input). With no resume argument it takes the original enqueue path, byte-for-byte backward compatible.

Instance Method Summary collapse

Instance Method Details

#perform(run_id, resume_input = nil) ⇒ Object

Runs the job: with no resume_input it finds the run and executes the workflow (the original enqueue path); with a resume_input Hash it takes the durable-resume path via Workflow.resume.



31
32
33
34
35
36
37
38
39
 
# File 'lib/nexo/workflow_job.rb', line 31

def perform(run_id, resume_input = nil)
  if resume_input.nil?
    run = Nexo::RunStore.default.find(run_id)
    klass = Object.const_get(run.workflow_class)
    klass.execute(run, payload: run.payload.transform_keys(&:to_sym))
  else
    Nexo::Workflow.resume(run_id, resume_input.transform_keys(&:to_sym))
  end
end