Class: Textus::Infra::Store::WorkflowQueue

Inherits:
Base
  • Object
show all
Defined in:
lib/textus/infra/store/workflow_queue.rb

Instance Method Summary collapse

Methods inherited from Base

#initialize

Constructor Details

This class inherits a constructor from Textus::Infra::Store::Base

Instance Method Details

#complete(event_id) ⇒ Object 



37
38
39
 
# File 'lib/textus/infra/store/workflow_queue.rb', line 37

def complete(event_id)
  table(:workflow_events).where(id: event_id).update(state: "done")
end

#fail(event_id, retry_count:, error:) ⇒ Object 



41
42
43
 
# File 'lib/textus/infra/store/workflow_queue.rb', line 41

def fail(event_id, retry_count:, error:)
  table(:workflow_events).where(id: event_id).update(state: "failed", retry_count:, error:)
end

#mark_processing(event_id) ⇒ Object 



33
34
35
 
# File 'lib/textus/infra/store/workflow_queue.rb', line 33

def mark_processing(event_id)
  table(:workflow_events).where(id: event_id).update(state: "processing")
end

#pending?(event_type:, key:) ⇒ Boolean

Returns:

  • (Boolean) 


26
27
28
29
30
31
 
# File 'lib/textus/infra/store/workflow_queue.rb', line 26

def pending?(event_type:, key:)
  table(:workflow_events)
    .where(event_type: event_type.to_s, key: key)
    .where(state: %w[pending processing])
    .any?
end

#pop ⇒ Object 



15
16
17
18
19
20
21
22
23
24
 
# File 'lib/textus/infra/store/workflow_queue.rb', line 15

def pop
  row = table(:workflow_events)
        .where(state: "pending")
        .where(Sequel.|({ scheduled_at: nil }, Sequel.lit("scheduled_at <= datetime('now')")))
        .order(:created_at, :id)
        .limit(1)
        .for_update
        .first
  result(row)
end

#push(event_type:, workflow_type:, key:, lane:, payload: nil, created_at: Time.now.utc, scheduled_at: nil, state: "pending", retry_count: 0, max_attempts: 3) ⇒ Object 



5
6
7
8
9
10
11
12
13
 
# File 'lib/textus/infra/store/workflow_queue.rb', line 5

def push(event_type:, workflow_type:, key:, lane:, payload: nil,
         created_at: Time.now.utc, scheduled_at: nil,
         state: "pending", retry_count: 0, max_attempts: 3)
  table(:workflow_events).insert(
    event_type: event_type.to_s, workflow_type:, key:, lane:,
    payload: payload&.to_json, created_at:, scheduled_at:,
    state:, retry_count:, max_attempts:
  )
end

#reschedule(event_id, retry_count:, delay:, error:) ⇒ Object 



45
46
47
48
49
50
 
# File 'lib/textus/infra/store/workflow_queue.rb', line 45

def reschedule(event_id, retry_count:, delay:, error:)
  table(:workflow_events).where(id: event_id).update(
    state: "pending", retry_count:, error:,
    scheduled_at: Sequel.lit("datetime('now', '+#{delay.to_i} seconds')")
  )
end