Module: ActiveJob::Notificare::Projection

Defined in:
lib/active_job/notificare/projection.rb

Constant Summary collapse

SUBSCRIPTIONS = 
[]

Class Method Summary collapse

Class Method Details

.subscribe! ⇒ Object 



6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
 
# File 'lib/active_job/notificare/projection.rb', line 6

def self.subscribe!
  SUBSCRIPTIONS << ActiveSupport::Notifications.subscribe("enqueue.active_job") do |event|
    job = event.payload[:job]
    next unless tracks_progress?(job)

    begin
      Execution.find_or_create_by!(job_id: job.job_id) do |e|
        e.job_class = job.class.name
        e.status = "enqueued"
      end
    rescue ActiveRecord::RecordNotUnique
      # Two concurrent projections for the same job_id — uniqueness constraint caught the
      # duplicate; the other thread won, so just find the existing row.
      Execution.find_by!(job_id: job.job_id)
    end
  end

  SUBSCRIPTIONS << ActiveSupport::Notifications.subscribe("perform_start.active_job") do |event|
    job = event.payload[:job]
    next unless tracks_progress?(job)

    execution = Execution.find_by(job_id: job.job_id)
    next unless execution

    if execution.running?
      # Resume path (ERD §9 case 3): worker was killed before perform.active_job fired,
      # leaving status as running. Preserve progress_current and started_at.
      # No continuation_state column — Continuation owns that (ERD §6).
      execution.update!(error: nil) if execution.error.present?
    else
      execution.update!(status: "running", started_at: Time.current)
    end
  end

  # Mirror ActiveJob::Continuation's current step name onto the execution row.
  SUBSCRIPTIONS << ActiveSupport::Notifications.subscribe("step_started.active_job") do |event|
    job = event.payload[:job]
    next unless tracks_progress?(job)

    step = event.payload[:step]
    Execution.find_by(job_id: job.job_id)&.update!(current_step: step.name.to_s)
  end

  # ActiveJob::Continuation fires `step.active_job` (not `step_completed`) after each step's
  # block finishes (whether successfully or with an exception). Only write a notification
  # when the step completed without error and was not interrupted.
  SUBSCRIPTIONS << ActiveSupport::Notifications.subscribe("step.active_job") do |event|
    next if event.payload[:exception_object]
    next if event.payload[:interrupted]

    job = event.payload[:job]
    next unless tracks_progress?(job)

    step = event.payload[:step]
    notify_event = job.respond_to?(:notificare_step_notify_for) ? job.notificare_step_notify_for(step.name) : nil
    write_step_notification(job, step.name, notify_event) if notify_event
  end

  SUBSCRIPTIONS << ActiveSupport::Notifications.subscribe("perform.active_job") do |event|
    job = event.payload[:job]
    next unless tracks_progress?(job)

    execution = Execution.find_by(job_id: job.job_id)
    next unless execution

    if (exception = event.payload[:exception_object])
      execution.update!(status: "failed", completed_at: Time.current, error: exception.message)
      write_lifecycle_notification(job, :failed, exception.message) if notifies_on?(job, :failed)
    else
      execution.update!(status: "completed", completed_at: Time.current)
      write_lifecycle_notification(job, :completed) if notifies_on?(job, :completed)
    end
  end
end

.unsubscribe! ⇒ Object 



81
82
83
84
 
# File 'lib/active_job/notificare/projection.rb', line 81

def self.unsubscribe!
  SUBSCRIPTIONS.each { |s| ActiveSupport::Notifications.unsubscribe(s) }
  SUBSCRIPTIONS.clear
end