Class: Wurk::ApiController

Inherits:
ApplicationController show all
Includes:
ActionController::Live, SameOriginGuard, SseStreaming, StreamConcurrencyGuard
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 = 
120.0
STREAM_MIN_TICK_SECONDS =

Positive floor for ?tick=: a zero/negative tick would sleep 0 the SSE loop into a tight Redis-read/write spin (Live runs it in a spawned thread) for up to STREAM_MAX_DURATION. Clamp before it reaches drive_stream.

0.1
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

Constants included from StreamConcurrencyGuard

StreamConcurrencyGuard::MAX_CONCURRENT_STREAMS, StreamConcurrencyGuard::RETRY_AFTER_SECONDS

Constants included from SameOriginGuard

SameOriginGuard::SAFE_METHODS

Instance Method Summary collapse

Methods included from StreamConcurrencyGuard

acquire, release

Instance Method Details

#batch ⇒ Object 



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

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 



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

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).



77
78
79
80
 
# File 'app/controllers/wurk/api_controller.rb', line 77

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

#cron ⇒ Object 



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

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 



211
212
213
 
# File 'app/controllers/wurk/api_controller.rb', line 211

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

#dead ⇒ Object 



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

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

#dead_all ⇒ Object 



140
141
142
143
144
145
146
147
148
 
# File 'app/controllers/wurk/api_controller.rb', line 140

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 



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

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

#dead_job ⇒ Object

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



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

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 — one Lua round-trip (Fast::QueueExt#delete_job) instead of a paged find_job walk.



84
85
86
87
88
89
 
# File 'app/controllers/wurk/api_controller.rb', line 84

def delete_queue_job
  removed = ::Wurk::Queue.new(params[:name].to_s).delete_job(params[:jid].to_s)
  return render(json: { error: 'unknown job' }, status: :not_found) if removed.zero?

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

#enqueue_cron ⇒ Object 



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

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.



236
237
238
239
240
241
242
 
# File 'app/controllers/wurk/api_controller.rb', line 236

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.



248
249
250
251
252
253
 
# File 'app/controllers/wurk/api_controller.rb', line 248

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 



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

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.



48
49
50
51
52
53
54
55
56
 
# File 'app/controllers/wurk/api_controller.rb', line 48

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 



215
216
217
218
219
220
221
 
# File 'app/controllers/wurk/api_controller.rb', line 215

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 



223
224
225
226
227
228
229
230
231
 
# File 'app/controllers/wurk/api_controller.rb', line 223

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 



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

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.



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

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

#processes ⇒ Object 



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

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

#profiles ⇒ Object

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



283
284
285
286
 
# File 'app/controllers/wurk/api_controller.rb', line 283

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 



66
67
68
69
70
71
72
73
74
 
# File 'app/controllers/wurk/api_controller.rb', line 66

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.



259
260
261
262
263
264
265
266
267
268
269
270
 
# File 'app/controllers/wurk/api_controller.rb', line 259

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 



62
63
64
 
# File 'app/controllers/wurk/api_controller.rb', line 62

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.



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

def quiet_process = signal_processes(:quiet!)

#reset_limiter ⇒ Object 



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

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

#retries ⇒ Object 



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

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

#retries_all ⇒ Object 



111
112
113
114
115
116
117
118
119
120
 
# File 'app/controllers/wurk/api_controller.rb', line 111

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 



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

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

#retry_job ⇒ Object

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



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

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

#scheduled ⇒ Object 



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

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

#scheduled_all ⇒ Object 



126
127
128
129
130
131
132
133
134
 
# File 'app/controllers/wurk/api_controller.rb', line 126

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 



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

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

#scheduled_job ⇒ Object

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



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

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

#search ⇒ Object 



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

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

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

#stats ⇒ Object 



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

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

#stop_process ⇒ Object 



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

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. Bounded per process by StreamConcurrencyGuard (503 + Retry-After past the cap) so a burst of tabs can't pin every Puma thread.

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



296
297
298
299
300
301
302
303
304
305
 
# File 'app/controllers/wurk/api_controller.rb', line 296

def stream
  with_stream_slot do
    stream_headers!
    clamp = ::Wurk::Api::Pagination.method(:clamp_float)
    tick = clamp.call(params[:tick], STREAM_MIN_TICK_SECONDS, 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
end

#unpause_cron ⇒ Object 



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

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

#unpause_queue ⇒ Object 



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

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.



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

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