Class: Wurk::ApiController

Inherits:
ApplicationController show all
Includes:
ActionController::Live
Defined in:
app/controllers/wurk/api_controller.rb

Overview

JSON APIs consumed by the SolidJS SPA. Action methods stay thin; mapping to the wire shape lives in Wurk::Api::Serializers, and pagination lives in Wurk::Api::Pagination. SSE lives in #stream.

Wire-compat: every payload field reads from the canonical Wurk inspector objects (Stats, Queue, RetrySet, ScheduledSet, DeadSet, ProcessSet, BatchSet, Cron::LoopSet) so dashboards stay aligned with the Redis schema in docs/target/sidekiq-{free,pro,ent}.md.

Constant Summary collapse

STREAM_TICK_SECONDS = 
2.0
STREAM_MAX_DURATION = 
600.0
HISTORY_WINDOW_UNITS = 
{ 's' => 1, 'm' => 60, 'h' => 3600, 'd' => 86_400 }.freeze
DEFAULT_HISTORY_WINDOW = 
24 * 3600
RETRY_ACTIONS =

Per-set action whitelists. Maps the SPA's action name to the SortedEntry/JobSet method. Anything not listed 400s — keeps the bulk/ single dispatchers from reaching arbitrary methods off a request param.

{ 'retry' => :retry, 'delete' => :delete, 'kill' => :kill }.freeze
SCHEDULED_ACTIONS = 
{ 'delete' => :delete, 'add_to_queue' => :add_to_queue }.freeze
DEAD_ACTIONS = 
{ 'retry' => :retry, 'delete' => :delete }.freeze

Instance Method Summary collapse

Instance Method Details

#batch ⇒ Object 



170
171
172
173
174
175
176
177
 
# File 'app/controllers/wurk/api_controller.rb', line 170

def batch
  status = ::Wurk::Batch::Status.new(params[:bid].to_s)
  return render(json: { error: 'unknown batch' }, status: :not_found) unless status.exists?

  render json: status.data.transform_keys(&:to_sym)
rescue ::ArgumentError
  render json: { error: 'unknown batch' }, status: :not_found
end

#batches ⇒ Object 



163
164
165
166
167
168
 
# File 'app/controllers/wurk/api_controller.rb', line 163

def batches
  set = ::Wurk::BatchSet.new
  page = ::Wurk::Api::Pagination.window(params)
  rows = ::Wurk::Api::Pagination.slice(set, page) { |status| status.data.transform_keys(&:to_sym) }
  render json: { total: set.size, page: page[:page], count: page[:count], batches: rows }
end

#clear_queue ⇒ Object

Empties one queue (UNLINK list + drop from the queues set).



73
74
75
76
 
# File 'app/controllers/wurk/api_controller.rb', line 73

def clear_queue
  ::Wurk::Queue.new(params[:name].to_s).clear
  render json: { ok: true }
end

#cron ⇒ Object 



190
191
192
193
 
# File 'app/controllers/wurk/api_controller.rb', line 190

def cron
  now = ::Time.now.to_i
  render json: ::Wurk::Cron::LoopSet.new.map { |lp| ::Wurk::Api::Serializers.cron_row(lp, now) }
end

#cron_history ⇒ Object 



205
206
207
 
# File 'app/controllers/wurk/api_controller.rb', line 205

def cron_history
  render json: { lid: params[:lid].to_s, history: ::Wurk::Web::Enterprise::Periodic.history(params[:lid].to_s) }
end

#dead ⇒ Object 



101
 
# File 'app/controllers/wurk/api_controller.rb', line 101

def dead      = render_sorted_set(::Wurk::DeadSet.new)

#dead_all ⇒ Object 



136
137
138
139
140
141
142
143
144
 
# File 'app/controllers/wurk/api_controller.rb', line 136

def dead_all
  set = ::Wurk::DeadSet.new
  count = case params[:cmd].to_s
          when 'retry'  then set.retry_all
          when 'delete' then clear_set(set)
          else return render(json: { error: 'unknown action' }, status: :bad_request)
          end
  render json: { ok: true, count: count }
end

#dead_bulk ⇒ Object 



134
 
# File 'app/controllers/wurk/api_controller.rb', line 134

def dead_bulk = bulk_entry_action(::Wurk::DeadSet.new, DEAD_ACTIONS)

#dead_job ⇒ Object

--- Dead set mutations --------------------------------------------------



133
 
# File 'app/controllers/wurk/api_controller.rb', line 133

def dead_job  = single_entry_action(::Wurk::DeadSet.new, DEAD_ACTIONS, params[:cmd])

#delete_queue_job ⇒ Object

Removes a single job from a queue by jid. LREM matches exact bytes, so we locate the record (Queue#find_job) and let it delete its own value.



80
81
82
83
84
85
 
# File 'app/controllers/wurk/api_controller.rb', line 80

def delete_queue_job
  record = ::Wurk::Queue.new(params[:name].to_s).find_job(params[:jid].to_s)
  return render(json: { error: 'unknown job' }, status: :not_found) unless record

  render json: { ok: true, deleted: record.delete }
end

#enqueue_cron ⇒ Object 



198
199
200
201
202
203
 
# File 'app/controllers/wurk/api_controller.rb', line 198

def enqueue_cron
  jid = ::Wurk::Web::Enterprise::Periodic.enqueue_now(params[:lid].to_s)
  return render(json: { error: 'unknown loop' }, status: :not_found) if jid.nil?

  render json: { ok: true, jid: jid }
end

#history ⇒ Object

Cluster-total throughput/failures time-series for the dashboard charts. :bucket is 1m/5m/1h; ?window=24h (s/m/h/d suffix) is clamped to the bucket's retention. Recharts-ready array under series.



230
231
232
233
234
235
236
 
# File 'app/controllers/wurk/api_controller.rb', line 230

def history
  window = parse_window(params[:window])
  series = ::Wurk::Web::Enterprise::Historical.history(params[:bucket].to_s, window: window)
  render json: { bucket: params[:bucket].to_s, window: window, series: series.map { |row| ::Wurk::Api::Serializers.history_point(row) } }
rescue ::ArgumentError => e
  render json: { error: e.message }, status: :bad_request
end

#history_snapshots ⇒ Object

Ent §5.3 Historical snapshots from the capped history:metrics stream. ?limit=N (default 1000) most-recent points, oldest→newest, each {at:, processed:, failures:, …}. Fields are read generically so a migrated Sidekiq Ent stream renders as-is.



242
243
244
245
246
247
 
# File 'app/controllers/wurk/api_controller.rb', line 242

def history_snapshots
  limit = ::Wurk::Api::Pagination.clamp_int(
    params[:limit], 1, ::Wurk::History::STREAM_CAP, ::Wurk::History::STREAM_DEFAULT_LIMIT
  )
  render json: { snapshots: ::Wurk::Web::Enterprise::Historical.snapshots(limit: limit) }
end

#limiters ⇒ Object 



179
180
181
182
183
 
# File 'app/controllers/wurk/api_controller.rb', line 179

def limiters
  names = ::Wurk::Web::Enterprise::Limits.list(filter: params[:substr])
  page = ::Wurk::Api::Pagination.window(params)
  render json: { total: names.size, page: page[:page], count: page[:count], limiters: limiter_rows(names, page) }
end

#meta ⇒ Object

Boot-time flags the SPA reads once to shape the UI (e.g. hide destructive actions and show the read-only banner). Always a GET, so it stays reachable while read-only mode blocks mutations.



44
45
46
47
48
49
50
51
52
 
# File 'app/controllers/wurk/api_controller.rb', line 44

def meta
  config = ::Wurk::Web.config
  render json: {
    version: ::Wurk::VERSION,
    read_only: config.read_only? || !mutations_authorized?(config),
    read_only_message: config.read_only_message,
    custom_tabs: config.custom_tabs
  }
end

#metrics ⇒ Object 



209
210
211
212
213
214
215
 
# File 'app/controllers/wurk/api_controller.rb', line 209

def metrics
  minutes = ::Wurk::Api::Pagination.clamp_int(params[:minutes], 1, ::Wurk::Metrics::Query::MAX_MINUTES, 60)
  rows = ::Wurk::Web::Enterprise::Historical.top(minutes: minutes, class_filter: params[:substr])
  render json: { minutes: minutes, top_jobs: rows.map { |(klass, totals)| ::Wurk::Api::Serializers.metric_row(klass, totals) } }
rescue ::Wurk::Metrics::Query::WindowTooWide => e
  render json: { error: e.message }, status: :bad_request
end

#metrics_for_job ⇒ Object 



217
218
219
220
221
222
223
224
225
 
# File 'app/controllers/wurk/api_controller.rb', line 217

def metrics_for_job
  klass = params[:klass].to_s
  minutes, hours = metrics_window(params)
  rows = ::Wurk::Web::Enterprise::Historical.for_job(klass, minutes: minutes, hours: hours)
  series = rows.map { |row| row.merge(at: row[:at].to_f) }
  render json: { klass: klass, minutes: minutes, hours: hours, series: series }
rescue ::ArgumentError, ::Wurk::Metrics::Query::WindowTooWide => e
  render json: { error: e.message }, status: :bad_request
end

#pause_cron ⇒ Object 



195
 
# File 'app/controllers/wurk/api_controller.rb', line 195

def pause_cron   = render_cron_action(::Wurk::Web::Enterprise::Periodic.pause(params[:lid].to_s))

#pause_queue ⇒ Object

Pause/unpause a queue (Pro §6, §10.1). Idempotent; returns the resulting state so the SPA can update its toggle without a refetch round-trip.



89
90
91
92
 
# File 'app/controllers/wurk/api_controller.rb', line 89

def pause_queue
  ::Wurk::Queue.new(params[:name].to_s).pause!
  render json: { ok: true, paused: true }
end

#processes ⇒ Object 



146
147
148
 
# File 'app/controllers/wurk/api_controller.rb', line 146

def processes
  render json: ::Wurk::ProcessSet.new.map { |p| ::Wurk::Api::Serializers.process_row(p) }
end

#profiles ⇒ Object

Profiles list (v8.0+). The SPA links each row to /profiles/:key (view) and /profiles/:key/data (raw blob). Newest first.



276
277
278
279
 
# File 'app/controllers/wurk/api_controller.rb', line 276

def profiles
  records = ::Wurk::ProfileSet.new.map { |rec| ::Wurk::Api::Serializers.profile_record(rec) }
  render json: records.sort_by { |r| -(r[:started_at] || 0) }
end

#queue ⇒ Object 



62
63
64
65
66
67
68
69
70
 
# File 'app/controllers/wurk/api_controller.rb', line 62

def queue
  q = ::Wurk::Queue.new(params[:name].to_s)
  page = ::Wurk::Api::Pagination.window(params)
  jobs = ::Wurk::Api::Pagination.slice(q, page) { |rec| ::Wurk::Api::Serializers.job_record(rec) }
  render json: {
    name: q.name, size: q.size, latency: q.latency, paused: q.paused?,
    page: page[:page], count: page[:count], jobs: jobs
  }
end

#queue_history ⇒ Object

Per-queue size/latency gauge time-series for the Metrics/Historical tab. :bucket is 1m/5m/1h; ?window=24h (s/m/h/d) is clamped to the bucket's retention; optional ?queue=<name> narrows to one queue. Each queue's points are Recharts-ready.



253
254
255
256
257
258
259
260
261
262
263
264
 
# File 'app/controllers/wurk/api_controller.rb', line 253

def queue_history
  window = parse_window(params[:window])
  queues = params[:queue].present? ? [params[:queue].to_s] : nil
  series = ::Wurk::Web::Enterprise::Historical.queue_history(params[:bucket].to_s, window: window, queues: queues)
  render json: {
    bucket: params[:bucket].to_s,
    window: window,
    queues: series.map { |row| ::Wurk::Api::Serializers.queue_history_series(row) }
  }
rescue ::ArgumentError => e
  render json: { error: e.message }, status: :bad_request
end

#queues ⇒ Object 



58
59
60
 
# File 'app/controllers/wurk/api_controller.rb', line 58

def queues
  render json: ::Wurk::Stats.new.queue_summaries.map { |q| ::Wurk::Api::Serializers.queue_summary(q) }
end

#quiet_process ⇒ Object

Busy-page controls: SIGTSTP (quiet — drop fetch, drain in-flight) and SIGTERM (stop — graceful shutdown). Both are async; the target notices on its next heartbeat (≤10s). identity absent or "all" signals every live process.



160
 
# File 'app/controllers/wurk/api_controller.rb', line 160

def quiet_process = signal_processes(:quiet!)

#reset_limiter ⇒ Object 



185
186
187
188
 
# File 'app/controllers/wurk/api_controller.rb', line 185

def reset_limiter
  ::Wurk::Web::Enterprise::Limits.reset(params[:name].to_s)
  render json: { ok: true }
end

#retries ⇒ Object 



99
 
# File 'app/controllers/wurk/api_controller.rb', line 99

def retries   = render_sorted_set(::Wurk::RetrySet.new)

#retries_all ⇒ Object 



107
108
109
110
111
112
113
114
115
116
 
# File 'app/controllers/wurk/api_controller.rb', line 107

def retries_all
  set = ::Wurk::RetrySet.new
  count = case params[:cmd].to_s
          when 'retry'  then set.retry_all
          when 'kill'   then set.kill_all
          when 'delete' then clear_set(set)
          else return render(json: { error: 'unknown action' }, status: :bad_request)
          end
  render json: { ok: true, count: count }
end

#retries_bulk ⇒ Object 



105
 
# File 'app/controllers/wurk/api_controller.rb', line 105

def retries_bulk = bulk_entry_action(::Wurk::RetrySet.new, RETRY_ACTIONS)

#retry_job ⇒ Object

--- Retry set mutations -------------------------------------------------



104
 
# File 'app/controllers/wurk/api_controller.rb', line 104

def retry_job    = single_entry_action(::Wurk::RetrySet.new, RETRY_ACTIONS, params[:cmd])

#scheduled ⇒ Object 



100
 
# File 'app/controllers/wurk/api_controller.rb', line 100

def scheduled = render_sorted_set(::Wurk::ScheduledSet.new)

#scheduled_all ⇒ Object 



122
123
124
125
126
127
128
129
130
 
# File 'app/controllers/wurk/api_controller.rb', line 122

def scheduled_all
  set = ::Wurk::ScheduledSet.new
  count = case params[:cmd].to_s
          when 'delete'       then clear_set(set)
          when 'add_to_queue' then drain_set(set, :add_to_queue)
          else return render(json: { error: 'unknown action' }, status: :bad_request)
          end
  render json: { ok: true, count: count }
end

#scheduled_bulk ⇒ Object 



120
 
# File 'app/controllers/wurk/api_controller.rb', line 120

def scheduled_bulk = bulk_entry_action(::Wurk::ScheduledSet.new, SCHEDULED_ACTIONS)

#scheduled_job ⇒ Object

--- Scheduled set mutations ---------------------------------------------



119
 
# File 'app/controllers/wurk/api_controller.rb', line 119

def scheduled_job  = single_entry_action(::Wurk::ScheduledSet.new, SCHEDULED_ACTIONS, params[:cmd])

#search ⇒ Object 



266
267
268
269
270
271
272
 
# File 'app/controllers/wurk/api_controller.rb', line 266

def search
  substr = params[:substr].to_s
  return render(json: { substr: substr, total: 0, hits: [] }) if substr.empty?

  hits = ::Wurk::Web::Search.new(substr, kinds: parse_search_kinds(params), limit: parse_search_limit(params)).to_a
  render json: { substr: substr, total: hits.size, hits: hits }
end

#stats ⇒ Object 



54
55
56
 
# File 'app/controllers/wurk/api_controller.rb', line 54

def stats
  render json: ::Wurk::Api::Serializers.stats_payload(::Wurk::Stats.new)
end

#stop_process ⇒ Object 



161
 
# File 'app/controllers/wurk/api_controller.rb', line 161

def stop_process  = signal_processes(:stop!)

#stream ⇒ Object

SSE: one event: stats per tick with a fresh Stats snapshot. Caps at STREAM_MAX_DURATION so a stale browser tab can't tie a Rails worker forever — the client reconnects automatically when the stream closes.

?max_duration= and ?tick= are test/debug knobs; the SPA never sets them. ?max_duration=0 emits one tick and closes.



287
288
289
290
291
292
293
294
 
# File 'app/controllers/wurk/api_controller.rb', line 287

def stream
  stream_headers!
  clamp = ::Wurk::Api::Pagination.method(:clamp_float)
  tick = clamp.call(params[:tick], 0.0, STREAM_TICK_SECONDS, STREAM_TICK_SECONDS)
  max_dur = clamp.call(params[:max_duration], 0.0, STREAM_MAX_DURATION, STREAM_MAX_DURATION)
  sse = ::ActionController::Live::SSE.new(response.stream, retry: (STREAM_TICK_SECONDS * 1000).to_i)
  drive_stream(sse, tick, max_dur)
end

#unpause_cron ⇒ Object 



196
 
# File 'app/controllers/wurk/api_controller.rb', line 196

def unpause_cron = render_cron_action(::Wurk::Web::Enterprise::Periodic.unpause(params[:lid].to_s))

#unpause_queue ⇒ Object 



94
95
96
97
 
# File 'app/controllers/wurk/api_controller.rb', line 94

def unpause_queue
  ::Wurk::Queue.new(params[:name].to_s).unpause!
  render json: { ok: true, paused: false }
end

#workers ⇒ Object

Currently-executing jobs across the cluster (WorkSet), oldest first. The Busy page's process-detail modal filters client-side by process_id.



152
153
154
 
# File 'app/controllers/wurk/api_controller.rb', line 152

def workers
  render json: ::Wurk::WorkSet.new.map { |pid, tid, work| ::Wurk::Api::Serializers.work_row(pid, tid, work) }
end