feat: scheduled job management — run now, reschedule, run all now

CHANGELOG.md

@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- Scheduled job management — "Run Now" and offset buttons (+1h / +24h / +7d) on each scheduled job row; Turbo Stream removes the row on run-now and updates the scheduled-at cell on offset reschedule; "Run All Now (N)" header button back-dates all matching scheduled executions; backed by `ScheduledJobsController` using standard CRUD (`update` for single, `create` for bulk via `run_all_now` collection route)
+
 ## [0.2.0] - 2026-05-25
 
 ### Added

README.md

@@ -11,6 +11,7 @@ A mountable Rails engine that provides a unified web dashboard for the full [Sol
 
 - **Overview dashboard** with live counts across all three Solid Stack components; cards are clickable and link directly to each section
 - **Solid Queue** — browse jobs by status (ready, scheduled, claimed, blocked) with filtering by job class, queue name, priority, and time period; manage failed jobs (retry / discard / bulk retry / bulk discard), pause/resume queues, and inspect worker processes; **Bulk selection** checkbox-selects individual jobs for discard or retry; **Discard All** bulk-discards every job matching the current filters in one request; **CSV export** downloads jobs or failed jobs as a CSV file respecting active filters
+- **Scheduled job management** — "Run Now" and offset buttons (+1h / +24h / +7d) per row update the scheduled time inline via Turbo Stream; "Run All Now (N)" in the header back-dates all matching executions at once
 - **Job detail page** — drill into any job to see full arguments (pretty-printed JSON), queue, priority, enqueued time, Active Job ID, concurrency key, scheduled/blocked-until metadata, and a Discard button
 - **Failed job detail page** — drill into any failed job to see the full error, backtrace, and an inline JSON argument editor; submit to update arguments and retry in one action
 - **Solid Cache** — entry count and total byte size at a glance

app/controllers/solid_stack_web/scheduled_jobs_controller.rb

@@ -0,0 +1,52 @@
+module SolidStackWeb
+  class ScheduledJobsController < ApplicationController
+    def create
+      @period  = params[:period].presence_in(PERIOD_DURATIONS.keys)
+      job_ids  = scheduled_scope.pluck("solid_queue_jobs.id")
+      SolidQueue::ScheduledExecution.where(job_id: job_ids).update_all(scheduled_at: 1.second.ago)
+      SolidQueue::Job.where(id: job_ids).update_all(scheduled_at: 1.second.ago)
+      redirect_to jobs_path(status: "scheduled", period: @period),
+                  notice: "#{job_ids.size} #{"job".pluralize(job_ids.size)} scheduled to run immediately."
+    rescue => e
+      redirect_to jobs_path(status: "scheduled", period: @period),
+                  alert: "Could not run jobs: #{e.message}"
+    end
+
+    def update
+      @execution = SolidQueue::ScheduledExecution.find(params[:id])
+      @period    = params[:period].presence_in(PERIOD_DURATIONS.keys)
+      @run_now   = params[:offset] == "now"
+      new_time   = resolve_new_time(@execution, params[:offset])
+
+      @execution.update!(scheduled_at: new_time)
+      @execution.job.update!(scheduled_at: new_time)
+
+      respond_to do |format|
+        format.turbo_stream
+        format.html do
+          notice = @run_now ? "Job scheduled to run immediately." : "Job rescheduled by +#{params[:offset]}."
+          redirect_to jobs_path(status: "scheduled", period: @period), notice: notice
+        end
+      end
+    rescue ArgumentError => e
+      redirect_to jobs_path(status: "scheduled"), alert: e.message
+    rescue => e
+      redirect_to jobs_path(status: "scheduled"), alert: "Could not reschedule job: #{e.message}"
+    end
+
+    private
+
+    def scheduled_scope
+      scope = SolidQueue::ScheduledExecution.joins(:job)
+      scope = scope.where("solid_queue_jobs.created_at >= ?", PERIOD_DURATIONS[@period].ago) if @period.present?
+      scope
+    end
+
+    def resolve_new_time(execution, offset)
+      return 1.second.ago if offset == "now"
+      raise ArgumentError, "Invalid offset." unless PERIOD_DURATIONS.key?(offset)
+
+      execution.scheduled_at + PERIOD_DURATIONS[offset]
+    end
+  end
+end

app/views/solid_stack_web/jobs/index.html.erb

@@ -3,6 +3,14 @@
   <div class="sqw-header-actions">
     <%= link_to "Export CSV", jobs_path(format: :csv, status: @status, q: @search, queue: @queue, period: @period, priority: @priority),
           class: "sqw-btn sqw-btn--muted sqw-btn--sm", data: { turbo: false } %>
+    <% if @status == "scheduled" && @executions&.any? %>
+      <%= button_to "Run All Now (#{@pagy.count})",
+            run_all_now_scheduled_jobs_path(period: @period),
+            method: :post,
+            class: "sqw-btn sqw-btn--sm",
+            data: { turbo_confirm: "Run all #{@pagy.count} scheduled jobs immediately?",
+                    turbo_frame: "_top" } %>
+    <% end %>
     <% if SolidStackWeb::Job::DISCARDABLE.include?(@status) && @executions&.any? %>
       <%= button_to "Discard All (#{@pagy.count})",
             discard_all_jobs_path(status: @status, q: @search, queue: @queue, period: @period, priority: @priority),
@@ -111,9 +119,22 @@
               <td><%= execution.job.priority %></td>
               <td class="sqw-muted"><%= execution.created_at.strftime("%b %d %H:%M") %></td>
               <% if @status == "scheduled" %>
-                <td class="sqw-muted"><%= execution.scheduled_at&.strftime("%b %d %H:%M") %></td>
+                <td id="scheduled_at_<%= execution.id %>" class="sqw-muted"><%= execution.scheduled_at&.strftime("%b %d %H:%M") %></td>
               <% end %>
               <td class="sqw-actions">
+                <% if @status == "scheduled" %>
+                  <%= button_to "Run Now", scheduled_job_path(execution),
+                        method: :patch,
+                        params: { offset: "now", period: @period },
+                        class: "sqw-btn sqw-btn--sm",
+                        data: { turbo_confirm: "Run this job immediately?" } %>
+                  <% %w[1h 24h 7d].each do |offset| %>
+                    <%= button_to "+#{offset}", scheduled_job_path(execution),
+                          method: :patch,
+                          params: { offset: offset, period: @period },
+                          class: "sqw-btn sqw-btn--muted sqw-btn--sm" %>
+                  <% end %>
+                <% end %>
                 <% if %w[ready scheduled blocked].include?(@status) %>
                   <%= button_to "Discard", job_path(execution, status: @status, q: @search, queue: @queue, period: @period, priority: @priority),
                         method: :delete, class: "sqw-btn sqw-btn--danger sqw-btn--sm",

app/views/solid_stack_web/scheduled_jobs/update.turbo_stream.erb

@@ -0,0 +1,9 @@
+<% if @run_now %>
+  <%= turbo_stream.remove "execution_#{@execution.id}" %>
+<% else %>
+  <%= turbo_stream.replace "scheduled_at_#{@execution.id}" do %>
+    <td id="scheduled_at_<%= @execution.id %>" class="sqw-muted">
+      <%= @execution.scheduled_at.strftime("%b %d %H:%M") %>
+    </td>
+  <% end %>
+<% end %>

config/routes.rb

@@ -4,6 +4,12 @@
   resource :job_selection,        path: "jobs/selection",        only: [:destroy],         controller: "jobs/selections"
   resource :failed_job_selection, path: "failed_jobs/selection", only: [:create, :destroy], controller: "failed_jobs/selections"
 
+  resources :scheduled_jobs, only: [:update] do
+    collection do
+      post :run_all_now, action: :create
+    end
+  end
+
   resources :jobs, only: [:index, :show, :destroy] do
     collection do
       post :discard_all, action: :destroy

spec/requests/solid_stack_web/scheduled_jobs_spec.rb

@@ -0,0 +1,210 @@
+require "rails_helper"
+
+RSpec.describe "ScheduledJobs", type: :request do
+  let(:engine_root) { "/solid_stack" }
+
+  def create_scheduled(class_name: "ScheduledJob", queue_name: "default", scheduled_at: 2.hours.from_now)
+    SolidQueue::Job.skip_callback(:create, :after, :prepare_for_execution)
+    job = SolidQueue::Job.create!(
+      class_name:, queue_name:, priority: 0, scheduled_at:,
+      arguments: { "executions" => 0, "exception_executions" => {} }
+    )
+    SolidQueue::ScheduledExecution.create!(
+      job: job, queue_name:, priority: 0, scheduled_at:
+    )
+    SolidQueue::Job.set_callback(:create, :after, :prepare_for_execution)
+    job
+  end
+
+  describe "POST /scheduled_jobs/run_all_now" do
+    it "back-dates all scheduled executions and redirects with notice" do
+      create_scheduled
+
+      post "#{engine_root}/scheduled_jobs/run_all_now"
+
+      expect(response).to redirect_to("#{engine_root}/jobs?status=scheduled")
+      follow_redirect!
+      expect(response.body).to include("scheduled to run immediately")
+    end
+
+    it "sets scheduled_at to the past for each execution" do
+      job = create_scheduled
+
+      post "#{engine_root}/scheduled_jobs/run_all_now"
+
+      expect(job.scheduled_execution.reload.scheduled_at).to be <= Time.current
+    end
+
+    it "also updates the underlying job's scheduled_at" do
+      job = create_scheduled
+
+      post "#{engine_root}/scheduled_jobs/run_all_now"
+
+      expect(job.reload.scheduled_at).to be <= Time.current
+    end
+
+    it "includes the count in the notice" do
+      create_scheduled
+
+      post "#{engine_root}/scheduled_jobs/run_all_now"
+      follow_redirect!
+
+      expect(response.body).to include("1 job scheduled to run immediately")
+    end
+
+    it "pluralises correctly for multiple jobs" do
+      create_scheduled(class_name: "JobA")
+      create_scheduled(class_name: "JobB")
+
+      post "#{engine_root}/scheduled_jobs/run_all_now"
+      follow_redirect!
+
+      expect(response.body).to include("2 jobs scheduled to run immediately")
+    end
+
+    it "preserves period in the redirect" do
+      post "#{engine_root}/scheduled_jobs/run_all_now", params: { period: "24h" }
+
+      expect(response).to redirect_to("#{engine_root}/jobs?period=24h&status=scheduled")
+    end
+
+    it "redirects with alert when an error occurs" do
+      allow(SolidQueue::ScheduledExecution).to receive(:joins).and_raise(RuntimeError, "db error")
+
+      post "#{engine_root}/scheduled_jobs/run_all_now"
+
+      expect(response).to redirect_to("#{engine_root}/jobs?status=scheduled")
+      expect(flash[:alert]).to include("Could not run jobs")
+    end
+  end
+
+  describe "PATCH /scheduled_jobs/:id" do
+    context "with offset=now" do
+      it "sets scheduled_at to the past and redirects" do
+        job = create_scheduled
+        execution = job.scheduled_execution
+
+        patch "#{engine_root}/scheduled_jobs/#{execution.id}", params: { offset: "now" }
+
+        expect(response).to redirect_to("#{engine_root}/jobs?status=scheduled")
+        expect(execution.reload.scheduled_at).to be <= Time.current
+      end
+
+      it "also updates the job's scheduled_at" do
+        job = create_scheduled
+        execution = job.scheduled_execution
+
+        patch "#{engine_root}/scheduled_jobs/#{execution.id}", params: { offset: "now" }
+
+        expect(job.reload.scheduled_at).to be <= Time.current
+      end
+
+      it "removes the row via turbo stream" do
+        job = create_scheduled
+        execution = job.scheduled_execution
+
+        patch "#{engine_root}/scheduled_jobs/#{execution.id}",
+              params: { offset: "now" },
+              headers: { "Accept" => "text/vnd.turbo-stream.html" }
+
+        expect(response.media_type).to eq("text/vnd.turbo-stream.html")
+        expect(response.body).to include("execution_#{execution.id}")
+        expect(response.body).to include("remove")
+      end
+    end
+
+    context "with offset=1h" do
+      it "postpones scheduled_at by 1 hour and redirects" do
+        job = create_scheduled
+        execution = job.scheduled_execution
+        original = execution.scheduled_at
+
+        patch "#{engine_root}/scheduled_jobs/#{execution.id}", params: { offset: "1h" }
+
+        expect(response).to redirect_to("#{engine_root}/jobs?status=scheduled")
+        expect(execution.reload.scheduled_at).to be_within(5.seconds).of(original + 1.hour)
+      end
+
+      it "updates the cell via turbo stream" do
+        job = create_scheduled
+        execution = job.scheduled_execution
+
+        patch "#{engine_root}/scheduled_jobs/#{execution.id}",
+              params: { offset: "1h" },
+              headers: { "Accept" => "text/vnd.turbo-stream.html" }
+
+        expect(response.media_type).to eq("text/vnd.turbo-stream.html")
+        expect(response.body).to include("scheduled_at_#{execution.id}")
+        expect(response.body).to include("replace")
+      end
+    end
+
+    context "with offset=24h" do
+      it "postpones scheduled_at by 24 hours" do
+        job = create_scheduled
+        execution = job.scheduled_execution
+        original = execution.scheduled_at
+
+        patch "#{engine_root}/scheduled_jobs/#{execution.id}", params: { offset: "24h" }
+
+        expect(execution.reload.scheduled_at).to be_within(5.seconds).of(original + 24.hours)
+      end
+    end
+
+    context "with offset=7d" do
+      it "postpones scheduled_at by 7 days" do
+        job = create_scheduled
+        execution = job.scheduled_execution
+        original = execution.scheduled_at
+
+        patch "#{engine_root}/scheduled_jobs/#{execution.id}", params: { offset: "7d" }
+
+        expect(execution.reload.scheduled_at).to be_within(5.seconds).of(original + 7.days)
+      end
+    end
+
+    context "with an invalid offset" do
+      it "redirects with an alert" do
+        job = create_scheduled
+        execution = job.scheduled_execution
+
+        patch "#{engine_root}/scheduled_jobs/#{execution.id}", params: { offset: "bogus" }
+
+        expect(response).to redirect_to("#{engine_root}/jobs?status=scheduled")
+        follow_redirect!
+        expect(response.body).to include("Invalid offset")
+      end
+
+      it "does not modify scheduled_at" do
+        job = create_scheduled
+        execution = job.scheduled_execution
+        original = execution.scheduled_at
+
+        patch "#{engine_root}/scheduled_jobs/#{execution.id}", params: { offset: "bogus" }
+
+        expect(execution.reload.scheduled_at).to be_within(1.second).of(original)
+      end
+    end
+
+    context "with a missing execution" do
+      it "redirects with an alert" do
+        patch "#{engine_root}/scheduled_jobs/0", params: { offset: "now" }
+
+        expect(response).to redirect_to("#{engine_root}/jobs?status=scheduled")
+        follow_redirect!
+        expect(response.body).to include("Could not reschedule job")
+      end
+    end
+
+    context "when period param is present" do
+      it "preserves period in the redirect" do
+        job = create_scheduled
+        execution = job.scheduled_execution
+
+        patch "#{engine_root}/scheduled_jobs/#{execution.id}", params: { offset: "now", period: "24h" }
+
+        expect(response).to redirect_to("#{engine_root}/jobs?period=24h&status=scheduled")
+      end
+    end
+  end
+end