Class: Shikibu::App

Inherits:
Object
  • Object
show all
Defined in:
lib/shikibu/app.rb

Overview

Main application class for Shikibu Manages configuration, lifecycle, and workflow execution

Examples:

app = Shikibu::App.new(
  database_url: 'sqlite://shikibu.db',
  service_name: 'my-service',
  auto_migrate: true
)

app.register OrderWorkflow
app.start

instance_id = app.start_workflow(OrderWorkflow, order_id: '123', amount: 99.99)
result = app.get_result(instance_id)

app.shutdown

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(database_url:, service_name: 'shikibu', auto_migrate: false, hooks: nil, use_listen_notify: nil, message_retention_days: 7, outbox_enabled: false, broker_url: nil, outbox_poll_interval: 1.0, outbox_max_retries: 3, outbox_max_age_hours: nil) ⇒ App

Create a new Shikibu application

Raises:

  • (ArgumentError) 


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
 
# File 'lib/shikibu/app.rb', line 40

def initialize(database_url:, service_name: 'shikibu', auto_migrate: false, hooks: nil,
               use_listen_notify: nil, message_retention_days: 7,
               outbox_enabled: false, broker_url: nil, outbox_poll_interval: 1.0,
               outbox_max_retries: 3, outbox_max_age_hours: nil)
  @database_url = database_url
  @service_name = service_name
  @auto_migrate = auto_migrate
  @hooks = hooks
  @use_listen_notify = use_listen_notify
  @message_retention_days = message_retention_days
  @outbox_enabled = outbox_enabled
  @broker_url = broker_url
  @outbox_poll_interval = outbox_poll_interval
  @outbox_max_retries = outbox_max_retries
  @outbox_max_age_hours = outbox_max_age_hours
  @worker_id = Locking.generate_worker_id(service_name)
  @registered_workflows = {}
  @running = false
  @worker = nil
  @notify_listener = nil

  # Validate outbox configuration
  raise ArgumentError, 'broker_url is required when outbox_enabled is true' if @outbox_enabled && @broker_url.nil?

  # Initialize storage
  @storage = Storage::SequelStorage.new(database_url, auto_migrate: auto_migrate)

  # Initialize notify listener for PostgreSQL
  setup_notify_listener if should_enable_listen_notify?

  # Initialize replay engine
  @replay_engine = ReplayEngine.new(
    storage: @storage,
    worker_id: @worker_id,
    hooks: @hooks
  )
end

Instance Attribute Details

#broker_url ⇒ Object (readonly)

Returns the value of attribute broker_url.



25
26
27
 
# File 'lib/shikibu/app.rb', line 25

def broker_url
  @broker_url
end

#hooks ⇒ Object (readonly)

Returns the value of attribute hooks.



25
26
27
 
# File 'lib/shikibu/app.rb', line 25

def hooks
  @hooks
end

#message_retention_days ⇒ Object (readonly)

Returns the value of attribute message_retention_days.



25
26
27
 
# File 'lib/shikibu/app.rb', line 25

def message_retention_days
  @message_retention_days
end

#notify_listener ⇒ Object (readonly)

Access the notify listener



261
262
263
 
# File 'lib/shikibu/app.rb', line 261

def notify_listener
  @notify_listener
end

#outbox_enabled ⇒ Object (readonly)

Returns the value of attribute outbox_enabled.



25
26
27
 
# File 'lib/shikibu/app.rb', line 25

def outbox_enabled
  @outbox_enabled
end

#outbox_max_age_hours ⇒ Object (readonly)

Returns the value of attribute outbox_max_age_hours.



25
26
27
 
# File 'lib/shikibu/app.rb', line 25

def outbox_max_age_hours
  @outbox_max_age_hours
end

#outbox_max_retries ⇒ Object (readonly)

Returns the value of attribute outbox_max_retries.



25
26
27
 
# File 'lib/shikibu/app.rb', line 25

def outbox_max_retries
  @outbox_max_retries
end

#outbox_poll_interval ⇒ Object (readonly)

Returns the value of attribute outbox_poll_interval.



25
26
27
 
# File 'lib/shikibu/app.rb', line 25

def outbox_poll_interval
  @outbox_poll_interval
end

#service_name ⇒ Object (readonly)

Returns the value of attribute service_name.



25
26
27
 
# File 'lib/shikibu/app.rb', line 25

def service_name
  @service_name
end

#storage ⇒ Object (readonly)

Returns the value of attribute storage.



25
26
27
 
# File 'lib/shikibu/app.rb', line 25

def storage
  @storage
end

#worker ⇒ Object (readonly)

Access the worker



119
120
121
 
# File 'lib/shikibu/app.rb', line 119

def worker
  @worker
end

#worker_id ⇒ Object (readonly)

Returns the value of attribute worker_id.



25
26
27
 
# File 'lib/shikibu/app.rb', line 25

def worker_id
  @worker_id
end

Instance Method Details

#cancel_workflow(instance_id, reason: nil) ⇒ Object

Cancel a workflow

Raises:



190
191
192
193
194
195
196
197
198
199
200
 
# File 'lib/shikibu/app.rb', line 190

def cancel_workflow(instance_id, reason: nil) # rubocop:disable Lint/UnusedMethodArgument
  instance = @storage.get_instance(instance_id)
  raise WorkflowNotFoundError, instance_id unless instance

  # Only cancel if not already terminal
  terminal_statuses = [Status::COMPLETED, Status::FAILED, Status::CANCELLED]
  return false if terminal_statuses.include?(instance[:status])

  @storage.update_instance_status(instance_id, Status::CANCELLED)
  true
end

#get_result(instance_id) ⇒ Hash

Get workflow result

Raises:



154
155
156
157
158
159
160
161
162
163
 
# File 'lib/shikibu/app.rb', line 154

def get_result(instance_id)
  instance = @storage.get_instance(instance_id)
  raise WorkflowNotFoundError, instance_id unless instance

  {
    status: instance[:status],
    output: instance[:output_data],
    error: instance[:status] == Status::FAILED ? 'Workflow failed' : nil
  }
end

#get_status(instance_id) ⇒ String

Get workflow status

Raises:



180
181
182
183
184
185
 
# File 'lib/shikibu/app.rb', line 180

def get_status(instance_id)
  instance = @storage.get_instance(instance_id)
  raise WorkflowNotFoundError, instance_id unless instance

  instance[:status]
end

#get_typed_result(instance_id, workflow_class) ⇒ Object?

Get workflow result with typed output deserialization

Raises:



170
171
172
173
174
175
 
# File 'lib/shikibu/app.rb', line 170

def get_typed_result(instance_id, workflow_class)
  result = get_result(instance_id)
  return nil unless result[:output]

  workflow_class.deserialize_output(result[:output])
end

#list_workflows(limit: 100, offset: 0, status: nil, workflow_name: nil) ⇒ Array<Hash>

List workflow instances



208
209
210
211
212
213
214
215
 
# File 'lib/shikibu/app.rb', line 208

def list_workflows(limit: 100, offset: 0, status: nil, workflow_name: nil)
  @storage.list_instances(
    limit: limit,
    offset: offset,
    status_filter: status,
    workflow_name: workflow_name
  )
end

#outbox_enabled? ⇒ Boolean

Check if outbox relayer is enabled



79
80
81
 
# File 'lib/shikibu/app.rb', line 79

def outbox_enabled?
  @outbox_enabled
end

#register(workflow_class) ⇒ Object

Register a workflow class



85
86
87
88
89
 
# File 'lib/shikibu/app.rb', line 85

def register(workflow_class)
  name = workflow_class.workflow_name
  @registered_workflows[name] = workflow_class
  Shikibu.register_workflow(workflow_class)
end

#registered_workflows ⇒ Object

Get registered workflows



256
257
258
 
# File 'lib/shikibu/app.rb', line 256

def registered_workflows
  @registered_workflows.dup
end

#resume_workflow(instance_id) ⇒ Object

Resume a workflow



144
145
146
147
148
149
 
# File 'lib/shikibu/app.rb', line 144

def resume_workflow(instance_id)
  @replay_engine.resume_workflow(instance_id)
rescue WaitForTimerSignal, WaitForChannelSignal
  # Workflow still suspended
  nil
end

#running? ⇒ Boolean

Check if app is running



122
123
124
 
# File 'lib/shikibu/app.rb', line 122

def running?
  @running
end

#send_event(event_type, data, metadata: nil, target_instance_id: nil) ⇒ Object

Send an event to waiting workflows



222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
 
# File 'lib/shikibu/app.rb', line 222

def send_event(event_type, data, metadata: nil, target_instance_id: nil)
  full_metadata = (metadata || {}).merge(
    source_service: @service_name,
    published_at: Time.now.iso8601
  )

  # Publish message to channel for persistence
  message_id = @storage.publish_message(
    channel: event_type,
    data: data,
    metadata: full_metadata
  )

  # Deliver to subscribers
  if target_instance_id
    # Point-to-Point: deliver only to specific instance
    @storage.deliver_channel_message(
      instance_id: target_instance_id,
      channel: event_type,
      message_id: message_id,
      data: data,
      metadata: full_metadata,
      worker_id: @worker_id
    )
  else
    # Broadcast: wake all waiting subscribers (Worker will handle delivery)
    @notify_listener&.notify(Notify::Channel::CHANNEL_MESSAGE, { channel: event_type })
  end

  # Call hook
  hooks&.on_event_sent&.call(event_type, @service_name, data)
end

#shutdown ⇒ Object

Gracefully shutdown the application



109
110
111
112
113
114
115
116
 
# File 'lib/shikibu/app.rb', line 109

def shutdown
  return unless @running

  @running = false
  @worker&.stop
  @notify_listener&.stop
  @storage.close
end

#start ⇒ Object

Start the background worker



92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
 
# File 'lib/shikibu/app.rb', line 92

def start
  return if @running

  @running = true

  # Start notify listener before worker
  @notify_listener&.start

  @worker = Worker.new(self)

  # Register notification handlers
  register_notify_handlers if @notify_listener

  @worker.start
end

#start_workflow(workflow_class, instance_id: nil, **input) ⇒ String

Start a new workflow



131
132
133
134
135
136
137
138
139
140
 
# File 'lib/shikibu/app.rb', line 131

def start_workflow(workflow_class, instance_id: nil, **input)
  instance_id ||= SecureRandom.uuid

  @replay_engine.start_workflow(workflow_class, instance_id: instance_id, **input)

  instance_id
rescue WaitForTimerSignal, WaitForChannelSignal
  # Workflow suspended, return instance_id
  instance_id
end