diff --git a/.gitignore b/.gitignore
index b1dd6310c..a05ef34c7 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1,6 +1,7 @@
 /.bundle/
 /doc/
 /log/*.log
+/log/*.log.*
 /pkg/
 /tmp/
 /test/dummy/db/*.sqlite3
diff --git a/Gemfile b/Gemfile
index 5f929b7bc..30b9304dc 100644
--- a/Gemfile
+++ b/Gemfile
@@ -3,3 +3,17 @@ git_source(:github) { |repo| "https://github.com/#{repo}.git" }
 
 # Specify your gem's dependencies in solid_queue.gemspec.
 gemspec
+
+# https://github.com/brianmario/mysql2/issues/1345#issuecomment-1963508372
+gem "mysql2"
+gem "pg"
+gem "sqlite3"
+gem "pry-byebug"
+
+gem "awesome_print"
+gem "good_job"
+# gem "sidekiq", "~> 6.5"
+gem "sidekiq", "~> 7.2"
+gem "sidekiq-pro", "~> 7.2", source: "https://enterprise.contribsys.com/"
+# gem "sidekiq-batch"
+gem "after_commit_everywhere"
diff --git a/Gemfile.lock b/Gemfile.lock
index 0f1a754f7..b4e37024e 100644
--- a/Gemfile.lock
+++ b/Gemfile.lock
@@ -9,6 +9,12 @@ PATH
       railties (>= 7.1)
       thor (~> 1.3.1)
 
+GEM
+  remote: https://enterprise.contribsys.com/
+  specs:
+    sidekiq-pro (7.2.0)
+      sidekiq (>= 7.2.0, < 8)
+
 GEM
   remote: https://rubygems.org/
   specs:
@@ -47,10 +53,19 @@ GEM
       minitest (>= 5.1)
       mutex_m
       tzinfo (~> 2.0)
+    after_commit_everywhere (1.3.1)
+      activerecord (>= 4.2)
+      activesupport
     ast (2.4.2)
+    awesome_print (1.9.2)
+    base64 (0.1.1)
     base64 (0.2.0)
+    bigdecimal (3.1.4)
     bigdecimal (3.1.8)
+    builder (3.2.4)
     builder (3.3.0)
+    byebug (11.1.3)
+    coderay (1.1.3)
     concurrent-ruby (1.3.4)
     connection_pool (2.4.1)
     crass (1.0.6)
@@ -66,8 +81,16 @@ GEM
       raabro (~> 1.4)
     globalid (1.2.1)
       activesupport (>= 6.1)
-    i18n (1.14.5)
+    good_job (3.24.0)
+      activejob (>= 6.0.0)
+      activerecord (>= 6.0.0)
+      concurrent-ruby (>= 1.0.2)
+      fugit (>= 1.1)
+      railties (>= 6.0.0)
+      thor (>= 0.14.1)
+    i18n (1.14.1)
       concurrent-ruby (~> 1.0)
+    i18n (1.14.5)
     io-console (0.6.0)
     irb (1.6.2)
       reline (>= 0.3.0)
@@ -76,12 +99,15 @@ GEM
     loofah (2.22.0)
       crass (~> 1.0.2)
       nokogiri (>= 1.12.0)
+    method_source (1.0.0)
     mini_portile2 (2.8.1)
     minitest (5.24.0)
     mocha (2.1.0)
       ruby2_keywords (>= 0.0.5)
+    mutex_m (0.1.2)
     mutex_m (0.2.0)
     mysql2 (0.5.4)
+    mysql2 (0.5.6)
     nio4r (2.7.0)
     nokogiri (1.16.6-arm64-darwin)
       racc (~> 1.4)
@@ -94,6 +120,12 @@ GEM
       ast (~> 2.4.1)
       racc
     pg (1.5.4)
+    pry (0.14.2)
+      coderay (~> 1.1)
+      method_source (~> 1.0)
+    pry-byebug (3.10.1)
+      byebug (~> 11.0)
+      pry (>= 0.13, < 0.15)
     puma (6.4.2)
       nio4r (~> 2.0)
     raabro (1.4.0)
@@ -123,6 +155,8 @@ GEM
       zeitwerk (~> 2.6)
     rainbow (3.1.1)
     rake (13.0.6)
+    redis-client (0.20.0)
+      connection_pool
     regexp_parser (2.9.0)
     reline (0.3.2)
       io-console (~> 0.5)
@@ -159,6 +193,11 @@ GEM
       rubocop-rails
     ruby-progressbar (1.13.0)
     ruby2_keywords (0.0.5)
+    sidekiq (7.2.1)
+      concurrent-ruby (< 2)
+      connection_pool (>= 2.3.0)
+      rack (>= 2.2.4)
+      redis-client (>= 0.19.0)
     sqlite3 (1.5.4)
       mini_portile2 (~> 2.8.0)
     strscan (3.1.0)
@@ -178,12 +217,18 @@ PLATFORMS
   x86_64-linux
 
 DEPENDENCIES
+  after_commit_everywhere
+  awesome_print
   debug
+  good_job
   mocha
   mysql2
   pg
+  pry-byebug
   puma
   rubocop-rails-omakase
+  sidekiq (~> 7.2)
+  sidekiq-pro (~> 7.2)!
   solid_queue!
   sqlite3
 
diff --git a/README.md b/README.md
index 0f9a19a08..d5a250bd7 100644
--- a/README.md
+++ b/README.md
@@ -1,3 +1,6 @@
+SolidQueue::Supervisor.start(mode: :work)
+SolidQueue::Supervisor.start(mode: :dispatch)
+
 # Solid Queue
 
 Solid Queue is a DB-based queuing backend for [Active Job](https://edgeguides.rubyonrails.org/active_job_basics.html), designed with simplicity and performance in mind.
@@ -335,6 +338,17 @@ class ApplicationMailer < ActionMailer::Base
     Rails.error.report(exception)
     raise exception
   end
+```
+
+## Batch jobs
+
+```rb
+SolidQueue::JobBatch.enqueue(on_finish: BatchCompletionJob) do
+  5.times.map { |i| SleepyJob.perform_later(i) }
+end
+
+SolidQueue::JobBatch.enqueue(on_success: BatchCompletionJob) do
+  5.times.map { |i| SleepyJob.perform_later(i) }
 end
 ```
 
diff --git a/app/models/solid_queue/claimed_execution.rb b/app/models/solid_queue/claimed_execution.rb
index d4abf45ab..b8a1784fc 100644
--- a/app/models/solid_queue/claimed_execution.rb
+++ b/app/models/solid_queue/claimed_execution.rb
@@ -65,6 +65,8 @@ def perform
     else
       failed_with(result.error)
     end
+
+    job.job_batch.touch(:changed_at, :last_changed_at) if job.batch_id.present?
   ensure
     job.unblock_next_blocked_job
   end
diff --git a/app/models/solid_queue/client_middleware.rb b/app/models/solid_queue/client_middleware.rb
new file mode 100644
index 000000000..9b7dbeaa1
--- /dev/null
+++ b/app/models/solid_queue/client_middleware.rb
@@ -0,0 +1,40 @@
+# frozen_string_literal: true
+
+module SolidQueue
+  class ClientMiddleware
+    def call(_worker_class, job, _queue, _redis_pool)
+      # {
+      #   "retry"=>true,
+      #   "queue"=>"default",
+      #   "class"=>"ActiveJob::QueueAdapters::SidekiqAdapter::JobWrapper",
+      #   "wrapped"=>SolidQueue::WorkflowTest::SidekiqJob,
+      #   "args"=>
+      #     [{"job_class"=>"SolidQueue::WorkflowTest::SidekiqJob",
+      #       "job_id"=>"e5db262e-2c49-4556-a2e4-78c009105e5d",
+      #       "provider_job_id"=>nil,
+      #       "queue_name"=>"default",
+      #       "priority"=>nil,
+      #       "arguments"=>["world"],
+      #       "executions"=>0,
+      #       "exception_executions"=>{},
+      #       "locale"=>"en",
+      #       "timezone"=>"UTC",
+      #       "enqueued_at"=>"2024-02-14T16:29:36.752639000Z",
+      #       "scheduled_at"=>nil,
+      #       "batch_id"=>nil}],
+      #   "jid"=>"185f9fb47e6610a31a576f72",
+      #   "created_at"=>1707928176.753821
+      # }
+      # job["args"].first["batch_id"] = JobBatch.current_batch_id
+      # process_queue_job = job["args"].first
+      # job["process_queue_id"] = process_queue_job.id
+      # job["args"] = process_queue_job.args
+      # update_jid(process_queue_job)
+      if SolidQueue::Workflow.current_workflow.present?
+        job["workflow_id"] = SolidQueue::Workflow.current_workflow.id
+      end
+
+      yield
+    end
+  end
+end
diff --git a/app/models/solid_queue/job.rb b/app/models/solid_queue/job.rb
index 8574c1ec9..f21d313ce 100644
--- a/app/models/solid_queue/job.rb
+++ b/app/models/solid_queue/job.rb
@@ -8,6 +8,8 @@ class EnqueueError < StandardError; end
 
     serialize :arguments, coder: JSON
 
+    belongs_to :job_batch, foreign_key: :batch_id, optional: true
+
     class << self
       def enqueue_all(active_jobs)
         active_jobs_by_job_id = active_jobs.index_by(&:job_id)
@@ -53,6 +55,7 @@ def create_all_from_active_jobs(active_jobs)
         end
 
         def attributes_from_active_job(active_job)
+          active_job.batch_id = JobBatch.current_batch_id || active_job.batch_id
           {
             queue_name: active_job.queue_name || DEFAULT_QUEUE_NAME,
             active_job_id: active_job.job_id,
@@ -60,7 +63,8 @@ def attributes_from_active_job(active_job)
             scheduled_at: active_job.scheduled_at,
             class_name: active_job.class.name,
             arguments: active_job.serialize,
-            concurrency_key: active_job.concurrency_key
+            concurrency_key: active_job.concurrency_key,
+            batch_id: active_job.batch_id
           }
         end
     end
diff --git a/app/models/solid_queue/job/executable.rb b/app/models/solid_queue/job/executable.rb
index e2146a676..2222f95ed 100644
--- a/app/models/solid_queue/job/executable.rb
+++ b/app/models/solid_queue/job/executable.rb
@@ -76,7 +76,7 @@ def dispatch_bypassing_concurrency_limits
       end
 
       def finished!
-        if SolidQueue.preserve_finished_jobs?
+        if SolidQueue.preserve_finished_jobs? || batch_id.present?
           touch(:finished_at)
         else
           destroy!
diff --git a/app/models/solid_queue/job/workflow_callback_job.rb b/app/models/solid_queue/job/workflow_callback_job.rb
new file mode 100644
index 000000000..7a81ffdae
--- /dev/null
+++ b/app/models/solid_queue/job/workflow_callback_job.rb
@@ -0,0 +1,50 @@
+# frozen_string_literal: true
+
+module SolidQueue
+  class Job
+    class WorkflowCallbackJob < ActiveJob::Base
+      def perform(batch, workflow, workflow_level_id)
+        # puts "checking batch for workflow_level_id #{workflow_level_id}"
+        # apply job finished to nodes
+        jobs = batch.jobs
+        jobs_by_job_id = jobs.index_by(&:active_job_id)
+
+        ActiveRecord::Base.transaction do
+          workflow.workflow_nodes.where(active_job_id: jobs.pluck(:active_job_id)).each do |node|
+            # job = jobs.find { |j| j.active_job_id == node.active_job_id }
+            jobs = Array.wrap(jobs_by_job_id[node.active_job_id])
+            if jobs.any?(&:finished?)
+              # node.update!(finished_at: job.finished_at)
+              node.update!(finished_at: Time.current)
+            end
+          end
+        end
+
+        # finish workflow if all nodes are done
+        workflow.with_lock do
+          break if workflow.batch_id?
+          break if workflow.workflow_nodes.incomplete.any?
+          puts "and here we are..."
+
+          active_job = SolidQueue::Job::WorkflowMonitorJob.perform_later(workflow)
+          workflow.update!(batch_id: active_job.job_id)
+          puts "#{workflow.id}: and here we are... #{workflow.batch_id} #{workflow_level_id} #{workflow.workflow_nodes.incomplete.count}"
+        rescue => e
+          binding.pry
+          raise
+        end
+
+        return if workflow.batch_id?
+
+        workflow.workflow_nodes.incomplete.where(parent_workflow_level_id: workflow_level_id).each do |next_node|
+          jobs_to_insert = Workflow.prep_level(workflow, next_node.workflow_level_id)
+          Workflow.enqueue_solid_queue(workflow, next_node.workflow_level_id, jobs_to_insert)
+        end
+      rescue => e
+        puts "Error: #{e.message}"
+        puts e.backtrace
+        raise
+      end
+    end
+  end
+end
diff --git a/app/models/solid_queue/job/workflow_good_job_callback_job.rb b/app/models/solid_queue/job/workflow_good_job_callback_job.rb
new file mode 100644
index 000000000..67e8cc1bd
--- /dev/null
+++ b/app/models/solid_queue/job/workflow_good_job_callback_job.rb
@@ -0,0 +1,86 @@
+# frozen_string_literal: true
+
+module SolidQueue
+  class Job
+    class WorkflowGoodJobCallbackJob < ActiveJob::Base
+      def perform(batch, params)
+        workflow = batch.properties[:workflow]
+        workflow_level_id = batch.properties[:workflow_level_id]
+
+        ActiveRecord::Base.transaction do
+          # Mark nodes a finished or failed
+          any_discarded = mark_node_completion(workflow, batch)
+
+          # If any jobs in the level were discarded, discard the level and all its descendents
+          discard_descendents(workflow, workflow_level_id) if any_discarded
+        end
+
+        attempt_finish(workflow, workflow_level_id)
+
+        return if workflow.batch_id?
+
+        # enqueue next level of jobs
+        enqueue_next_levels(workflow, workflow_level_id)
+      rescue StandardError => e
+        puts "Error: #{e.message}"
+        puts e.backtrace
+        raise
+      end
+
+      def attempt_finish(workflow, workflow_level_id)
+        # Finish workflow if all nodes are done
+        workflow.with_lock do
+          return if workflow.batch_id?
+          return if workflow.workflow_nodes.incomplete.any?
+
+          active_job = SolidQueue::Job::WorkflowMonitorJob.perform_later(workflow)
+          workflow.update!(batch_id: active_job.job_id)
+          puts "#{workflow.id}: and here we are... #{workflow.batch_id} #{workflow_level_id} #{workflow.workflow_nodes.incomplete.count}"
+        rescue StandardError => e
+          puts "Error: #{e.message}"
+          puts e.backtrace
+          raise
+        end
+      end
+
+      def mark_node_completion(workflow, batch)
+        active_jobs = batch.active_jobs
+        good_jobs = GoodJob::Job.where(active_job_id: active_jobs.map(&:job_id)) # only get the last run
+        jobs_by_job_id = good_jobs.index_by(&:active_job_id)
+        any_discarded = false
+
+        workflow.workflow_nodes.where(active_job_id: jobs_by_job_id.keys).each do |node|
+          next unless node.incomplete?
+
+          jobs = Array.wrap(jobs_by_job_id[node.active_job_id])
+          if jobs.any?(&:discarded?)
+            any_discarded = true
+            # node.update!(discarded_at: job.discarded_at)
+            node.update!(discarded_at: Time.current)
+          elsif jobs.any?(&:finished?)
+            # node.update!(finished_at: job.finished_at)
+            node.update!(finished_at: Time.current)
+          end
+        end
+
+        any_discarded
+      end
+
+      def discard_descendents(workflow, workflow_level_id)
+        # get descendents and discard them
+        child_nodes = WorkflowNode.descendants(workflow, workflow_level_id)
+        SolidQueue::WorkflowNode
+          .where(workflow: workflow, workflow_level_id: child_nodes.pluck(:workflow_level_id))
+          .update_all(discarded_at: Time.current)
+      end
+
+      def enqueue_next_levels(workflow, workflow_level_id)
+        # ActiveRecord::Base.transaction - causes a deadlock in #prep_node
+        workflow.workflow_nodes.incomplete.where(parent_workflow_level_id: workflow_level_id).each do |next_node|
+          jobs_to_insert = Workflow.prep_level(workflow, next_node.workflow_level_id)
+          Workflow.enqueue_good_job(workflow, next_node.workflow_level_id, jobs_to_insert)
+        end
+      end
+    end
+  end
+end
diff --git a/app/models/solid_queue/job/workflow_monitor_job.rb b/app/models/solid_queue/job/workflow_monitor_job.rb
new file mode 100644
index 000000000..633f364b6
--- /dev/null
+++ b/app/models/solid_queue/job/workflow_monitor_job.rb
@@ -0,0 +1,61 @@
+# frozen_string_literal: true
+
+module SolidQueue
+  class Job
+    class WorkflowMonitorJob < ActiveJob::Base
+      retry_on StandardError, wait: 5.seconds, attempts: 5
+
+      class GoodJobJob < ApplicationJob
+        self.queue_adapter = :good_job
+
+        def perform(arg)
+          # puts "Hi #{arg}!"
+          # raise "doh" if arg == "6"
+        end
+      end
+
+      # def perform(batch, workflow)
+      def perform(workflow)
+        # check for finish levels and jobs
+        #
+        puts "YOYOYOYOYOYOYO"
+
+        # SolidQueue::Workflow.enqueue do |w|
+        #   w.run(GoodJobJob.new("1").set(wait_until: 1.second.from_now)) do
+        #     w.run(GoodJobJob.new("2")) do
+        #       w.run(GoodJobJob.new("3").set(wait_until: 2.seconds.from_now))
+        #       w.run(GoodJobJob.new("4"))
+        #     end
+        #     w.run(GoodJobJob.new("5")) do
+        #       w.run(GoodJobJob.new("6")) do
+        #         w.run(GoodJobJob.new("7"))
+        #         w.run(GoodJobJob.new("7.5"))
+        #         w.run(GoodJobJob.new("7.75"))
+        #       end
+        #       w.run(GoodJobJob.new("6.5"))
+        #     end
+        #     w.run(GoodJobJob.new("8"), GoodJobJob.new("9"), GoodJobJob.new("10")) do
+        #       w.run(GoodJobJob.new("11"), GoodJobJob.new("12"), GoodJobJob.new("13"), GoodJobJob.new("14")) do
+        #         w.run(GoodJobJob.new("15"))
+        #         w.run(GoodJobJob.new("16"))
+        #         w.run(GoodJobJob.new("17"))
+        #         w.run(GoodJobJob.new("18")) do
+        #           w.run(GoodJobJob.new("19").set(wait_until: 3.seconds.from_now)) do
+        #             w.run(GoodJobJob.new("20"))
+        #           end
+        #           w.run(GoodJobJob.new("21"))
+        #           w.run(GoodJobJob.new("22"))
+        #           w.run(GoodJobJob.new("23"))
+        #         end
+        #       end
+        #     end
+        #   end
+        # end
+      end
+    end
+  end
+end
+
+# or enqueue a batch job at start of workflow
+#   every child workflow batch job is part of it
+#   every child workflow batch job enqueues the next level of jobs so it never finishes until they're all done
diff --git a/app/models/solid_queue/job/workflow_sidekiq_batch_callback.rb b/app/models/solid_queue/job/workflow_sidekiq_batch_callback.rb
new file mode 100644
index 000000000..a5f7c2815
--- /dev/null
+++ b/app/models/solid_queue/job/workflow_sidekiq_batch_callback.rb
@@ -0,0 +1,22 @@
+# frozen_string_literal: true
+
+module SolidQueue
+  class Job
+    class WorkflowSidekiqBatchCallback
+      # def on_complete(status, options)
+      #   # options["workflow_id"]
+      #   # options["workflow_level_id"]
+      #   puts "", "", "on_complete #{options} #{status}", "", ""
+      # end
+
+      def on_success(status, options)
+        puts "", "", "on_success #{options} #{status}", "", ""
+      end
+
+      def on_death(status, options)
+        binding.pry
+        puts "", "", "on_death #{options} #{status}"
+      end
+    end
+  end
+end
diff --git a/app/models/solid_queue/job/workflow_sidekiq_monitor_job.rb b/app/models/solid_queue/job/workflow_sidekiq_monitor_job.rb
new file mode 100644
index 000000000..13d0102c9
--- /dev/null
+++ b/app/models/solid_queue/job/workflow_sidekiq_monitor_job.rb
@@ -0,0 +1,59 @@
+# frozen_string_literal: true
+
+module SolidQueue
+  class Job
+    class WorkflowSidekiqMonitorJob < ActiveJob::Base
+      retry_on StandardError, wait: 5.seconds, attempts: 5
+
+      self.queue_adapter = :sidekiq
+
+      # def perform(batch, workflow)
+      def perform(workflow)
+        # workflow.changed_at?
+
+        loop do
+          puts "Checking the sidekiq workflow #{workflow.id}..."
+          workflow.with_lock do
+            return if workflow.batch_id?
+            unless workflow.workflow_nodes.incomplete.any?
+              workflow.update!(batch_id: 12345)
+
+              begin
+                puts "SolidQueue::Job::WorkflowMonitorJob #{workflow.id}..."
+                SolidQueue::Job::WorkflowMonitorJob.perform_later(workflow)
+              rescue => e
+                binding.pry
+                raise
+              end
+
+              return
+            end
+
+            # return if workflow.batch_id?
+            # return if workflow.batch_id? && workflow.workflow_levels.where(started_at: nil).any?
+
+            workflow.workflow_levels.where(finished_at: nil).where.not(started_at: nil).each do |workflow_level|
+              if workflow.workflow_nodes.where(workflow_level_id: workflow_level.workflow_level_id).incomplete.none?
+                workflow_level.update!(finished_at: Time.current)
+              end
+            end
+
+            workflow.workflow_levels.where(started_at: nil).each do |next_level|
+              puts "Checking the sidekiq workflow #{next_level.id}..."
+              parent_level = workflow.workflow_levels.find_by!(workflow_level_id: next_level.parent_workflow_level_id)
+              next if parent_level.finished_at.blank?
+
+              Workflow.enqueue_level(workflow, next_level)
+            end
+          end
+
+          sleep 1
+        end
+      end
+    end
+  end
+end
+
+# or enqueue a batch job at start of workflow
+#   every child workflow batch job is part of it
+#   every child workflow batch job enqueues the next level of jobs so it never finishes until they're all done
diff --git a/app/models/solid_queue/job_batch.rb b/app/models/solid_queue/job_batch.rb
new file mode 100644
index 000000000..eb41f7de4
--- /dev/null
+++ b/app/models/solid_queue/job_batch.rb
@@ -0,0 +1,163 @@
+# frozen_string_literal: true
+
+module SolidQueue
+  class JobBatch < Record
+    belongs_to :job, foreign_key: :job_id, optional: true
+    belongs_to :parent_job_batch, foreign_key: :parent_job_batch_id, class_name: "SolidQueue::JobBatch", optional: true
+    has_many :jobs, foreign_key: :batch_id
+    has_many :children, foreign_key: :parent_job_batch_id, class_name: "SolidQueue::JobBatch"
+
+    serialize :on_finish_active_job, coder: JSON
+    serialize :on_success_active_job, coder: JSON
+    serialize :on_failure_active_job, coder: JSON
+
+    scope :incomplete, -> {
+      where(finished_at: nil).where("changed_at IS NOT NULL OR last_changed_at < ?", 1.hour.ago)
+    }
+    scope :finished, -> { where.not(finished_at: nil) }
+
+    class << self
+      def current_batch_id
+        ActiveSupport::IsolatedExecutionState[:current_batch_id]
+      end
+
+      def enqueue(attributes = {})
+        job_batch = nil
+        transaction do
+          job_batch = create!(batch_attributes(attributes))
+          wrap_in_batch_context(job_batch.id) do
+            yield job_batch
+          end
+        end
+
+        job_batch
+      end
+
+      def dispatch_finished_batches
+        incomplete.order(:id).pluck(:id).each do |id|
+          transaction do
+            where(id: id).includes(:children, :jobs).non_blocking_lock.each(&:finish)
+          end
+        end
+      end
+
+      def wrap_in_batch_context(batch_id)
+        previous_batch_id = current_batch_id.presence || nil
+        ActiveSupport::IsolatedExecutionState[:current_batch_id] = batch_id
+        yield
+      ensure
+        ActiveSupport::IsolatedExecutionState[:current_batch_id] = previous_batch_id
+      end
+
+      private
+
+        def batch_attributes(attributes)
+          on_finish_klass = attributes.delete(:on_finish)
+          on_success_klass = attributes.delete(:on_success)
+          on_failure_klass = attributes.delete(:on_failure)
+
+          if on_finish_klass.present?
+            attributes[:on_finish_active_job] = as_active_job(on_finish_klass).serialize
+          end
+
+          if on_success_klass.present?
+            attributes[:on_success_active_job] = as_active_job(on_success_klass).serialize
+          end
+
+          if on_failure_klass.present?
+            attributes[:on_failure_active_job] = as_active_job(on_failure_klass).serialize
+          end
+
+          attributes[:parent_job_batch_id] = current_batch_id if current_batch_id.present?
+
+          attributes
+        end
+
+        def as_active_job(active_job_klass)
+          active_job_klass.is_a?(ActiveJob::Base) ? active_job_klass : active_job_klass.new
+        end
+    end
+
+    # Instance-level enqueue
+    def enqueue(attributes = {})
+      raise "You cannot enqueue a batch that is already finished" if finished?
+
+      transaction do
+        self.class.wrap_in_batch_context(id) do
+          yield self
+        end
+      end
+
+      self
+    end
+
+    def finished?
+      finished_at.present?
+    end
+
+    def finish
+      return if finished?
+      reset_changed_at
+
+      all_jobs_succeeded = true
+      attrs = {}
+      jobs.find_each do |next_job|
+        # SolidQueue does treats `discard_on` differently than failures. The job will report as being :finished,
+        #   and there is no record of the failure.
+        # GoodJob would report a discard as an error. It's possible we should do that in the future?
+        if fire_failure_job?(next_job)
+          perform_completion_job(:on_failure_active_job, attrs)
+          update!(attrs)
+        end
+
+        status = next_job.status
+        all_jobs_succeeded = all_jobs_succeeded && status != :failed
+        return unless status.in?([ :finished, :failed ])
+      end
+
+      children.find_each do |child|
+        return unless child.finished?
+      end
+
+      if on_finish_active_job.present?
+        perform_completion_job(:on_finish_active_job, attrs)
+      end
+
+      if on_success_active_job.present? && all_jobs_succeeded
+        perform_completion_job(:on_success_active_job, attrs)
+      end
+
+      transaction do
+        parent_job_batch.touch(:changed_at, :last_changed_at) if parent_job_batch_id.present?
+        update!({ finished_at: Time.zone.now }.merge(attrs))
+      end
+    end
+
+    private
+
+      def fire_failure_job?(job)
+        return false if on_failure_active_job.blank? || job.failed_execution.blank?
+        job = ActiveJob::Base.deserialize(on_failure_active_job)
+        job.provider_job_id.blank?
+      end
+
+      def perform_completion_job(job_field, attrs)
+        active_job = ActiveJob::Base.deserialize(send(job_field))
+        active_job.send(:deserialize_arguments_if_needed)
+        active_job.arguments = [ self ] + Array.wrap(active_job.arguments)
+        self.class.wrap_in_batch_context(id) do
+          ActiveJob.perform_all_later([ active_job ])
+        end
+        active_job.provider_job_id = Job.find_by(active_job_id: active_job.job_id).id
+        attrs[job_field] = active_job.serialize
+      end
+
+      def reset_changed_at
+        if changed_at.blank? && last_changed_at.present?
+          update_columns(last_changed_at: Time.zone.now) # wait another hour before we check again
+        else
+          update_columns(changed_at: nil) # clear out changed_at so we ignore this until the next job finishes
+        end
+      end
+  end
+end
diff --git a/app/models/solid_queue/server_middleware.rb b/app/models/solid_queue/server_middleware.rb
new file mode 100644
index 000000000..838d16e4e
--- /dev/null
+++ b/app/models/solid_queue/server_middleware.rb
@@ -0,0 +1,46 @@
+# frozen_string_literal: true
+
+module SolidQueue
+  class ServerMiddleware
+    def call(worker, job, _queue)
+      # {
+      #   "retry"=>true,
+      #   "queue"=>"default",
+      #   "wrapped"=>"SolidQueue::WorkflowTest::SidekiqJob",
+      #   "args"=>
+      #     [{"job_class"=>"SolidQueue::WorkflowTest::SidekiqJob",
+      #       "job_id"=>"0f408431-04f1-4744-a2d4-575f4a59feb5",
+      #       "provider_job_id"=>nil,
+      #       "queue_name"=>"default",
+      #       "priority"=>nil,
+      #       "arguments"=>["world"],
+      #       "executions"=>0,
+      #       "exception_executions"=>{},
+      #       "locale"=>"en",
+      #       "timezone"=>"UTC",
+      #       "enqueued_at"=>"2024-02-14T16:02:53.234564000Z",
+      #       "scheduled_at"=>nil,
+      #       "batch_id"=>nil}],
+      #   "class"=>"ActiveJob::QueueAdapters::SidekiqAdapter::JobWrapper",
+      #   "jid"=>"f964458a2f10e4b38edd364f",
+      #   "created_at"=>1707926573.235855,
+      #   "enqueued_at"=>1707926643.688022
+      # }
+      if job["workflow_id"].present?
+        active_job_id = job["args"].first["job_id"]
+        workflow = SolidQueue::Workflow.find(job["workflow_id"])
+        begin
+          yield
+        ensure
+          workflow.workflow_nodes.find_by(active_job_id: active_job_id)&.update!(finished_at: Time.current)
+        end
+      else
+        begin
+          yield
+        ensure
+          puts "the job finished in the middleware #{$!}"
+        end
+      end
+    end
+  end
+end
diff --git a/app/models/solid_queue/workflow.rb b/app/models/solid_queue/workflow.rb
new file mode 100644
index 000000000..1ed242885
--- /dev/null
+++ b/app/models/solid_queue/workflow.rb
@@ -0,0 +1,234 @@
+# frozen_string_literal: true
+
+require "tsort"
+require "after_commit_everywhere"
+
+module SolidQueue
+  class Workflow < Record
+    belongs_to :job_batch, foreign_key: :batch_id, optional: true
+    has_many :workflow_nodes, foreign_key: :workflow_id, dependent: :destroy
+
+    class TopologicalHash < Hash
+      include TSort
+
+      alias tsort_each_node each_key
+
+      def tsort_each_child(node, &block)
+        fetch(node).each(&block)
+      end
+    end
+
+    # TODO:
+    # Could we just have one batch for the whole workflow
+    #  it has every job in it, but one of those jobs checks the batch, then re-enqueues for a second later
+    #    and it can handle managing the levels each second
+    #    and we mark a level as finished so we can query less
+    #  how do we handle things on the same level with different rules? still an issue
+
+    class << self
+      def current_workflow
+        ActiveSupport::IsolatedExecutionState[:workflow]
+      end
+
+      def enqueue
+        transaction do
+          workflow = create!
+
+          workflow.with_context { yield workflow }
+
+          # dependencies = workflow.built_levels.values.flatten
+          # t_hash = TopologicalHash.new
+          # dependencies.each do |node|
+          #   t_hash[node.workflow_level_id] ||= []
+          #   if node.parent_workflow_level_id.present?
+          #     t_hash[node.workflow_level_id] << node.parent_workflow_level_id
+          #   end
+          # end
+          # t_hash.sort # validates the graph is acyclic
+
+          SolidQueue::WorkflowNode.insert_all workflow.built_nodes.map { |wn|
+            wn.attributes.except("id").merge("created_at" => Time.current, "updated_at" => Time.current)
+          }
+
+          workflow.workflow_nodes.where(parent_workflow_level_id: nil).each do |workflow_node|
+            jobs_to_insert = [prep_node(workflow_node)]
+
+            # jobs_to_insert = prep_level(workflow, workflow_node.workflow_level_id)
+
+            # Sidekiq
+            # enqueue_level(workflow, workflow_node.workflow_level_id)
+
+            # Sidekiq Pro happens in after commit
+            # enqueue_sidekiq_pro(workflow, workflow_node.workflow_level_id, jobs_to_insert)
+
+            # Only actual SolidQueue specific code
+            enqueue_solid_queue(workflow, workflow_node.workflow_level_id, jobs_to_insert)
+
+            # GoodJob
+            # enqueue_good_job(workflow, workflow_node.workflow_level_id, jobs_to_insert)
+          end
+
+          # Sidekiq
+          # AfterCommitEverywhere.after_commit do
+          #   SolidQueue::Job::WorkflowSidekiqMonitorJob.set(wait: 1.second).perform_later(workflow)
+          # end
+
+          workflow
+        end
+      rescue => e
+        puts "Error: #{e.message}"
+        puts e.backtrace
+        raise
+      end
+
+      def parent_level_id
+        ActiveSupport::IsolatedExecutionState[:parent_level_id]
+      end
+
+      def prep_level(workflow, workflow_level_id)
+        # puts "prepping level #{workflow_level_id}"
+        level_nodes = workflow.workflow_nodes.where(workflow_level_id: workflow_level_id)
+
+        jobs_to_insert = []
+        level_nodes.each do |level_node|
+          jobs_to_insert << prep_node(level_node)
+        end
+
+        jobs_to_insert.compact
+      end
+
+      def prep_node(workflow_node)
+        return if workflow_node.started_at.present? # should group by level first instead of iterating each on level
+
+        workflow_node.update!(started_at: Time.current)
+        ActiveJob::Base.deserialize(workflow_node.active_job).tap do |active_job|
+          active_job.send(:deserialize_arguments_if_needed)
+        end
+      end
+
+      def enqueue_good_job(workflow, workflow_level_id, jobs_to_insert)
+        workflow.with_context do
+          GoodJob::Batch.enqueue(
+            jobs_to_insert,
+            on_finish: ::SolidQueue::Job::WorkflowGoodJobCallbackJob,
+            workflow: workflow,
+            workflow_level_id: workflow_level_id
+            # workflow_level: workflow_level
+          )
+        end
+      end
+
+      def enqueue_solid_queue(workflow, workflow_level_id, jobs_to_insert)
+        workflow.with_context do
+          JobBatch.enqueue(on_finish: SolidQueue::Job::WorkflowCallbackJob.new(workflow, workflow_level_id)) do
+            ActiveJob.perform_all_later(jobs_to_insert) if jobs_to_insert.any?
+          end
+        end
+      end
+
+      def enqueue_sidekiq_pro(workflow, workflow_level_id, jobs_to_insert)
+        AfterCommitEverywhere.after_commit do
+          workflow.with_context do
+            # Sidekiq Pro
+            batch = Sidekiq::Batch.new
+            %i[success death].each do |event|
+              batch.on(
+                event,
+                SolidQueue::Job::WorkflowSidekiqBatchCallback,
+                workflow_id: workflow.id,
+                workflow_level_id: workflow_level_id
+                # workflow_level_id: workflow_level.workflow_level_id
+              )
+            end
+            batch.jobs do
+              ActiveJob.perform_all_later(jobs_to_insert) if jobs_to_insert.any?
+            end
+
+            # If rails >= 7.1
+            # ActiveJob.perform_all_later(jobs_to_insert) if jobs_to_insert.any?
+            # If rails < 7.1
+            # jobs_to_insert.group_by(&:queue_adapter).each do |queue_adapter, jobs|
+            #   jobs.each do |job|
+            #     if job.scheduled_at
+            #       queue_adapter.enqueue_at(job, job.scheduled_at.to_f)
+            #     else
+            #       queue_adapter.enqueue(job)
+            #     end
+            #   end
+            # end
+          end
+        end
+      end
+    end
+
+    attr_reader :built_nodes, :built_levels
+
+    def with_context
+      begin
+        was = ActiveSupport::IsolatedExecutionState[:workflow]
+        ActiveSupport::IsolatedExecutionState[:workflow] = self
+        yield self
+      ensure
+        ActiveSupport::IsolatedExecutionState[:workflow] = was
+      end
+    end
+
+    def run(*active_jobs)
+      @built_nodes ||= []
+      @built_levels ||= []
+      workflow_level_id = SecureRandom.uuid
+
+      if Workflow.parent_level_id.blank?
+        # @built_levels << workflow_levels.build(
+        #   workflow: self,
+        #   workflow_level_id: workflow_level_id
+        # )
+
+        active_jobs.each do |active_job|
+          @built_nodes << workflow_nodes.build(
+            workflow: self,
+            active_job_id: active_job.job_id,
+            active_job: active_job.serialize,
+            workflow_level_id: workflow_level_id
+          )
+        end
+
+        begin
+          was = ActiveSupport::IsolatedExecutionState[:parent_level_id]
+          ActiveSupport::IsolatedExecutionState[:parent_level_id] = workflow_level_id
+          yield if block_given?
+        ensure
+          ActiveSupport::IsolatedExecutionState[:parent_level_id] = was
+        end
+
+
+      elsif Workflow.parent_level_id.present?
+        # @built_levels << workflow_levels.build(
+        #   workflow: self,
+        #   workflow_level_id: workflow_level_id,
+        #   parent_workflow_level_id: Workflow.parent_level_id
+        # )
+
+        active_jobs.each do |active_job|
+          @built_nodes << workflow_nodes.build(
+            workflow: self,
+            active_job_id: active_job.job_id,
+            active_job: active_job.serialize,
+            workflow_level_id: workflow_level_id,
+            parent_workflow_level_id: Workflow.parent_level_id
+          )
+        end
+
+        begin
+          was = ActiveSupport::IsolatedExecutionState[:parent_level_id]
+          ActiveSupport::IsolatedExecutionState[:parent_level_id] = workflow_level_id
+          yield if block_given?
+        ensure
+          ActiveSupport::IsolatedExecutionState[:parent_level_id] = was
+        end
+      end
+
+      active_jobs
+    end
+  end
+end
diff --git a/app/models/solid_queue/workflow_node.rb b/app/models/solid_queue/workflow_node.rb
new file mode 100644
index 000000000..15bf6d51a
--- /dev/null
+++ b/app/models/solid_queue/workflow_node.rb
@@ -0,0 +1,33 @@
+# frozen_string_literal: true
+
+module SolidQueue
+  class WorkflowNode < Record
+    belongs_to :workflow
+
+    serialize :active_job, coder: JSON
+
+    scope :incomplete, -> { where(finished_at: nil, discarded_at: nil) }
+
+    def self.descendants(workflow, workflow_level_id)
+      workflow_id = workflow.id
+      sql = <<-SQL
+        WITH RECURSIVE descendants AS (
+          SELECT id, workflow_level_id, parent_workflow_level_id, finished_at, discarded_at
+          FROM solid_queue_workflow_nodes
+          WHERE workflow_level_id = :workflow_level_id AND workflow_id = :workflow_id
+          UNION ALL
+          SELECT n.id, n.workflow_level_id, n.parent_workflow_level_id, n.finished_at, n.discarded_at
+          FROM solid_queue_workflow_nodes n
+          JOIN descendants d ON n.parent_workflow_level_id = d.workflow_level_id
+        )
+        SELECT * FROM descendants;
+      SQL
+
+      WorkflowNode.find_by_sql([sql, workflow_level_id: workflow_level_id, workflow_id: workflow_id])
+    end
+
+    def incomplete?
+      finished_at.blank? && discarded_at.blank?
+    end
+  end
+end
diff --git a/db/migrate/20240131013203_create_solid_queue_batch_table.rb b/db/migrate/20240131013203_create_solid_queue_batch_table.rb
new file mode 100644
index 000000000..91b76ee84
--- /dev/null
+++ b/db/migrate/20240131013203_create_solid_queue_batch_table.rb
@@ -0,0 +1,21 @@
+class CreateSolidQueueBatchTable < ActiveRecord::Migration[7.1]
+  def change
+    create_table :solid_queue_job_batches do |t|
+      t.references :parent_job_batch, index: true # FIXME: foreign key
+      t.text :on_finish_active_job
+      t.text :on_success_active_job
+      t.text :on_failure_active_job
+      t.datetime :finished_at
+      t.datetime :changed_at
+      t.datetime :last_changed_at
+      t.timestamps
+
+      t.index [ :finished_at ]
+      t.index [ :changed_at ]
+      t.index [ :last_changed_at ]
+    end
+
+    add_reference :solid_queue_jobs, :batch, index: true
+    add_foreign_key :solid_queue_jobs, :solid_queue_job_batches, column: :batch_id, on_delete: :cascade
+  end
+end
diff --git a/db/migrate/20240203164236_create_solid_queue_workflows.rb b/db/migrate/20240203164236_create_solid_queue_workflows.rb
new file mode 100644
index 000000000..a310deb86
--- /dev/null
+++ b/db/migrate/20240203164236_create_solid_queue_workflows.rb
@@ -0,0 +1,23 @@
+class CreateSolidQueueWorkflows < ActiveRecord::Migration[7.1]
+  def change
+    create_table :solid_queue_workflows do |t|
+      # t.bigint :batch_id, index: true
+      t.string :batch_id, index: true
+      t.timestamps
+    end
+
+    create_table :solid_queue_workflow_nodes do |t|
+      t.references :workflow, null: false, foreign_key: { to_table: :solid_queue_workflows }
+      t.string :active_job_id, null: false
+      t.text :active_job, null: false
+      t.string :workflow_level_id, null: false
+      t.string :parent_workflow_level_id
+      t.datetime :started_at, index: true
+      t.datetime :finished_at, index: true
+      t.datetime :discarded_at, index: true
+      t.timestamps
+    end
+
+    add_index :solid_queue_workflow_levels, %i[workflow_level_id], unique: true
+  end
+end
diff --git a/db/migrate/20240219040006_create_good_jobs.rb b/db/migrate/20240219040006_create_good_jobs.rb
new file mode 100644
index 000000000..6937b5a0a
--- /dev/null
+++ b/db/migrate/20240219040006_create_good_jobs.rb
@@ -0,0 +1,91 @@
+# frozen_string_literal: true
+
+class CreateGoodJobs < ActiveRecord::Migration[7.1]
+  def change
+    # Uncomment for Postgres v12 or earlier to enable gen_random_uuid() support
+    # enable_extension 'pgcrypto'
+
+    # create_table :good_jobs, id: :uuid do |t|
+    #   t.text :queue_name
+    #   t.integer :priority
+    #   t.jsonb :serialized_params
+    #   t.datetime :scheduled_at
+    #   t.datetime :performed_at
+    #   t.datetime :finished_at
+    #   t.text :error
+
+    #   t.timestamps
+
+    #   t.uuid :active_job_id
+    #   t.text :concurrency_key
+    #   t.text :cron_key
+    #   t.uuid :retried_good_job_id
+    #   t.datetime :cron_at
+
+    #   t.uuid :batch_id
+    #   t.uuid :batch_callback_id
+
+    #   t.boolean :is_discrete
+    #   t.integer :executions_count
+    #   t.text :job_class
+    #   t.integer :error_event, limit: 2
+    #   t.text :labels, array: true
+    # end
+
+    # create_table :good_job_batches, id: :uuid do |t|
+    #   t.timestamps
+    #   t.text :description
+    #   t.jsonb :serialized_properties
+    #   t.text :on_finish
+    #   t.text :on_success
+    #   t.text :on_discard
+    #   t.text :callback_queue_name
+    #   t.integer :callback_priority
+    #   t.datetime :enqueued_at
+    #   t.datetime :discarded_at
+    #   t.datetime :finished_at
+    # end
+
+    # create_table :good_job_executions, id: :uuid do |t|
+    #   t.timestamps
+
+    #   t.uuid :active_job_id, null: false
+    #   t.text :job_class
+    #   t.text :queue_name
+    #   t.jsonb :serialized_params
+    #   t.datetime :scheduled_at
+    #   t.datetime :finished_at
+    #   t.text :error
+    #   t.integer :error_event, limit: 2
+    # end
+
+    # create_table :good_job_processes, id: :uuid do |t|
+    #   t.timestamps
+    #   t.jsonb :state
+    # end
+
+    # create_table :good_job_settings, id: :uuid do |t|
+    #   t.timestamps
+    #   t.text :key
+    #   t.jsonb :value
+    #   t.index :key, unique: true
+    # end
+
+    # add_index :good_jobs, :scheduled_at, where: "(finished_at IS NULL)", name: :index_good_jobs_on_scheduled_at
+    # add_index :good_jobs, [:queue_name, :scheduled_at], where: "(finished_at IS NULL)", name: :index_good_jobs_on_queue_name_and_scheduled_at
+    # add_index :good_jobs, [:active_job_id, :created_at], name: :index_good_jobs_on_active_job_id_and_created_at
+    # add_index :good_jobs, :concurrency_key, where: "(finished_at IS NULL)", name: :index_good_jobs_on_concurrency_key_when_unfinished
+    # add_index :good_jobs, [:cron_key, :created_at], where: "(cron_key IS NOT NULL)", name: :index_good_jobs_on_cron_key_and_created_at_cond
+    # add_index :good_jobs, [:cron_key, :cron_at], where: "(cron_key IS NOT NULL)", unique: true, name: :index_good_jobs_on_cron_key_and_cron_at_cond
+    # add_index :good_jobs, [:finished_at], where: "retried_good_job_id IS NULL AND finished_at IS NOT NULL", name: :index_good_jobs_jobs_on_finished_at
+    # add_index :good_jobs, [:priority, :created_at], order: { priority: "DESC NULLS LAST", created_at: :asc },
+    #   where: "finished_at IS NULL", name: :index_good_jobs_jobs_on_priority_created_at_when_unfinished
+    # add_index :good_jobs, [:priority, :created_at], order: { priority: "ASC NULLS LAST", created_at: :asc },
+    #   where: "finished_at IS NULL", name: :index_good_job_jobs_for_candidate_lookup
+    # add_index :good_jobs, [:batch_id], where: "batch_id IS NOT NULL"
+    # add_index :good_jobs, [:batch_callback_id], where: "batch_callback_id IS NOT NULL"
+    # add_index :good_jobs, :labels, using: :gin, where: "(labels IS NOT NULL)", name: :index_good_jobs_on_labels
+
+    # add_index :good_job_executions, [:active_job_id, :created_at], name: :index_good_job_executions_on_active_job_id_and_created_at
+  end
+end
diff --git a/lib/active_job/job_batch_id.rb b/lib/active_job/job_batch_id.rb
new file mode 100644
index 000000000..28b42e86a
--- /dev/null
+++ b/lib/active_job/job_batch_id.rb
@@ -0,0 +1,35 @@
+# frozen_string_literal: true
+
+# Inspired by active_job/core.rb docs
+# https://github.com/rails/rails/blob/1c2529b9a6ba5a1eff58be0d0373d7d9d401015b/activejob/lib/active_job/core.rb#L136
+module ActiveJob
+  module JobBatchId
+    extend ActiveSupport::Concern
+
+    included do
+      attr_accessor :batch_id, :bid
+
+      after_discard do |job, error|
+        # grab the workflow node and mark it as discarded?
+      end
+    end
+
+    def serialize
+      super.merge("batch_id" => batch_id, "bid" => Thread.current[:sidekiq_batch]&.bid)
+    end
+
+    def deserialize(job_data)
+      super
+      self.batch_id = job_data["batch_id"]
+      self.bid = job_data["bid"]
+    end
+
+    def batch
+      @batch ||= SolidQueue::JobBatch.find_by(id: batch_id)
+    end
+
+    def sidekiq_batch
+      @sidekiq_batch ||= Sidekiq::Batch.new(bid)
+    end
+  end
+end
diff --git a/lib/solid_queue.rb b/lib/solid_queue.rb
index e7070d265..4e7bc6dda 100644
--- a/lib/solid_queue.rb
+++ b/lib/solid_queue.rb
@@ -5,6 +5,7 @@
 
 require "active_job"
 require "active_job/queue_adapters"
+require "active_job/job_batch_id"
 
 require "active_support"
 require "active_support/core_ext/numeric/time"
diff --git a/lib/solid_queue/dispatcher.rb b/lib/solid_queue/dispatcher.rb
index a22a82d8c..d8e7f5656 100644
--- a/lib/solid_queue/dispatcher.rb
+++ b/lib/solid_queue/dispatcher.rb
@@ -30,6 +30,7 @@ def poll
       def dispatch_next_batch
         with_polling_volume do
           ScheduledExecution.dispatch_next_batch(batch_size)
+          SolidQueue::JobBatch.dispatch_finished_batches
         end
       end
 
diff --git a/lib/solid_queue/engine.rb b/lib/solid_queue/engine.rb
index d10997c74..452ae4457 100644
--- a/lib/solid_queue/engine.rb
+++ b/lib/solid_queue/engine.rb
@@ -35,6 +35,7 @@ class Engine < ::Rails::Engine
     initializer "solid_queue.active_job.extensions" do
       ActiveSupport.on_load :active_job do
         include ActiveJob::ConcurrencyControls
+        include ActiveJob::JobBatchId
       end
     end
   end
diff --git a/test/dummy/app/jobs/batch_completion_job.rb b/test/dummy/app/jobs/batch_completion_job.rb
new file mode 100644
index 000000000..1ac3fd91f
--- /dev/null
+++ b/test/dummy/app/jobs/batch_completion_job.rb
@@ -0,0 +1,7 @@
+class BatchCompletionJob < ApplicationJob
+  queue_as :background
+
+  def perform(batch, arg1, arg2)
+    Rails.logger.info "#{batch.jobs.size} jobs completed! #{arg1} #{arg2}"
+  end
+end
diff --git a/test/dummy/app/jobs/good_job_callback_job.rb b/test/dummy/app/jobs/good_job_callback_job.rb
new file mode 100644
index 000000000..9794462e7
--- /dev/null
+++ b/test/dummy/app/jobs/good_job_callback_job.rb
@@ -0,0 +1,15 @@
+class GoodJobCallbackJob < ApplicationJob
+  self.queue_adapter = :good_job
+
+  # Callback jobs must accept a `batch` and `options` argument
+  def perform(batch, params)
+    puts "", "", "", "", "", "", "ok!", "", "", "", "", "", ""
+
+    binding.pry
+    # The batch object will contain the Batch's properties, which are mutable
+    batch.properties[:user] # => <User id: 1, ...>
+
+    # Params is a hash containing additional context (more may be added in the future)
+    params[:event] # => :finish, :success, :discard
+  end
+end
diff --git a/test/dummy/app/jobs/nice_job.rb b/test/dummy/app/jobs/nice_job.rb
new file mode 100644
index 000000000..c84de945e
--- /dev/null
+++ b/test/dummy/app/jobs/nice_job.rb
@@ -0,0 +1,14 @@
+class NiceJob < ApplicationJob
+  retry_on Exception, wait: 1.second, attempts: 3
+  # discard_on Oops
+
+  sidekiq_options retry: 2
+
+  self.queue_adapter = :good_job
+
+  def perform(arg)
+    # raise Oops # This triggers and on_success...
+    puts "Hi #{arg}, from good job!", "", ""
+    # raise "eek"
+  end
+end
diff --git a/test/dummy/app/jobs/sleepy_job.rb b/test/dummy/app/jobs/sleepy_job.rb
new file mode 100644
index 000000000..dd105cdc0
--- /dev/null
+++ b/test/dummy/app/jobs/sleepy_job.rb
@@ -0,0 +1,10 @@
+class SleepyJob < ApplicationJob
+  queue_as :background
+
+  retry_on Exception, wait: 30.seconds, attempts: 5
+
+  def perform(seconds_to_sleep)
+    Rails.logger.info "Feeling #{seconds_to_sleep} seconds sleepy..."
+    sleep seconds_to_sleep
+  end
+end
diff --git a/test/dummy/config/application.rb b/test/dummy/config/application.rb
index 502f70cb7..fb2240e1c 100644
--- a/test/dummy/config/application.rb
+++ b/test/dummy/config/application.rb
@@ -28,6 +28,7 @@ class Application < Rails::Application
     # config.eager_load_paths << Rails.root.join("extras")
 
     config.active_job.queue_adapter = :solid_queue
+    # config.active_job.queue_adapter = :good_job
 
     if ENV["SEPARATE_CONNECTION"] && ENV["TARGET_DB"] != "sqlite"
       config.solid_queue.connects_to = { database: { writing: :primary, reading: :replica } }
diff --git a/test/dummy/config/database.yml b/test/dummy/config/database.yml
index 85f80b365..3a1ca5b7c 100644
--- a/test/dummy/config/database.yml
+++ b/test/dummy/config/database.yml
@@ -18,7 +18,7 @@ default: &default
   adapter: postgresql
   encoding: unicode
   username: postgres
-  pool: 5
+  pool: 25
   host: "127.0.0.1"
   port: 55432
   gssencmode: disable # https://github.com/ged/ruby-pg/issues/311
diff --git a/test/dummy/config/environments/development.rb b/test/dummy/config/environments/development.rb
index 4d9f1d465..2f7ed8614 100644
--- a/test/dummy/config/environments/development.rb
+++ b/test/dummy/config/environments/development.rb
@@ -61,4 +61,7 @@
 
   logger = ActiveSupport::Logger.new(STDOUT)
   config.solid_queue.on_thread_error = ->(exception) { logger.error("#{exception.class.name}: #{exception.message}\n#{(exception.backtrace || caller)&.join("\n")}") }
+
+  # config.solid_queue.logger = ActiveSupport::Logger.new(nil)
+  # config.log_level = ENV.fetch("LOG_LEVEL", "debug").to_sym
 end
diff --git a/test/dummy/config/environments/test.rb b/test/dummy/config/environments/test.rb
index df8320265..d5252c9c8 100644
--- a/test/dummy/config/environments/test.rb
+++ b/test/dummy/config/environments/test.rb
@@ -53,6 +53,11 @@
   logger = ActiveSupport::Logger.new(STDOUT)
   config.solid_queue.on_thread_error = ->(exception) { logger.error("#{exception.class.name}: #{exception.message}\n#{(exception.backtrace || caller)&.join("\n")}") }
   config.solid_queue.logger = ActiveSupport::Logger.new(nil)
+  # config.log_level = ENV.fetch("LOG_LEVEL", "debug").to_sym
+  # config.solid_queue.on_thread_error = ->(exception) { logger.error("#{exception.class.name}: #{exception.message}\n#{exception.backtrace.join("\n")}") }
+  # # config.solid_queue.logger = ActiveSupport::Logger.new(nil)
 
   config.solid_queue.shutdown_timeout = 2.seconds
+
+  config.good_job.execution_mode = :async
 end
diff --git a/test/dummy/config/initializers/solid_queue_record.rb b/test/dummy/config/initializers/solid_queue_record.rb
index 72e237fe7..c4f5f62fa 100644
--- a/test/dummy/config/initializers/solid_queue_record.rb
+++ b/test/dummy/config/initializers/solid_queue_record.rb
@@ -4,3 +4,156 @@
   raise "Expected to run on SolidQueue::Record, got #{self.inspect}" unless self == SolidQueue::Record
   Rails.application.config.x.solid_queue_record_hook_ran = true
 end
+
+# FIXME: load these automatically like acidic job? sidekiq workflow? one of those does it
+#        and it's great
+require "sidekiq"
+require_relative "../../../../app/models/solid_queue/server_middleware.rb"
+require_relative "../../../../app/models/solid_queue/client_middleware.rb"
+
+# From sidekiq WIKI:
+# (https://github.com/mperham/sidekiq/wiki/Middleware#sometimes-client-side-middleware-should-be-registered-
+# in-both-places)
+# Sometimes client-side middleware should be registered in both places.
+# The jobs running in the Sidekiq server can themselves push new jobs to Sidekiq, thus acting as clients.
+# You must configure your client middleware within the configure_server block also in that case
+Sidekiq.configure_client do |config|
+  config.client_middleware do |chain|
+    chain.add ::SolidQueue::ClientMiddleware
+  end
+end
+
+Sidekiq.configure_server do |config|
+  config.client_middleware do |chain|
+    chain.add ::SolidQueue::ClientMiddleware
+  end
+
+  config.server_middleware do |chain|
+    chain.add ::SolidQueue::ServerMiddleware
+  end
+
+  # https://github.com/sidekiq/sidekiq/issues/4496#issuecomment-677838552
+  config.death_handlers << -> (job, exception) do
+    # worker = job["wrapped"].safe_constantize
+    # worker&.sidekiq_retries_exhausted_block&.call(job, exception)
+  end
+end
+
+# require 'sidekiq_adapter_extension'
+
+# module SidekiqAdapterExtension
+#   # included do
+#   #   class << self
+#   #     alias_method :original_enqueue, :enqueue
+
+#   #     def enqueue(job)
+#   #       # Custom logic before enqueue
+#   #       Rails.logger.info "Custom logic before enqueue"
+
+#   #       original_enqueue(job)
+
+#   #       # Custom logic after enqueue
+#   #       Rails.logger.info "Custom logic after enqueue"
+#   #     end
+#   #   end
+#   # end
+#   def enqueue(job) # :nodoc:
+#     puts "we dun did it!!!!", "", "", "", "", ""
+#     super
+#   end
+
+#   def enqueue_at(job, timestamp) # :nodoc:
+#     puts "we dun did it", "", "", "", "", ""
+#     super
+#   end
+
+#   def enqueue_all(jobs) # :nodoc:
+#     puts "we dun did it ALL", "", "", "", "", ""
+#     super
+#   end
+# end
+
+# ActiveSupport.on_load(:active_job) do
+#   ActiveJob::QueueAdapters::SidekiqAdapter.include(SidekiqAdapterExtension)
+#   binding.pry
+# end
+
+Rails.application.config.to_prepare do
+  unless defined?(OriginalAdapter)
+    OriginalAdapter = ActiveJob::QueueAdapters::SidekiqAdapter
+  end
+
+  module ActiveJob
+    module QueueAdapters
+      class SidekiqAdapter
+        def enqueue(job) # :nodoc:
+          # OriginalAdapter.new.enqueue(job)
+          job.provider_job_id = JobWrapper.set(
+            wrapped: job.class,
+            queue: job.queue_name
+          ).perform_async(job.serialize)
+        end
+
+        def enqueue_at(job, timestamp) # :nodoc:
+          # Sidekiq::Batch.current?
+          if Thread.current[:sidekiq_batch].blank? && job.bid.present?
+            batch = Sidekiq::Batch.new(job.bid)
+            batch.jobs do
+              # OriginalAdapter.new.enqueue_at(job, timestamp)
+              job.provider_job_id = JobWrapper.set(
+                wrapped: job.class,
+                queue: job.queue_name,
+              ).perform_at(timestamp, job.serialize)
+            end
+          else
+            job.provider_job_id = JobWrapper.set(
+              wrapped: job.class,
+              queue: job.queue_name,
+            ).perform_at(timestamp, job.serialize)
+          end
+        end
+
+        def enqueue_all(jobs) # :nodoc:
+          puts "WE ENQUEUED ALL #{jobs.map(&:job_id)}!"
+          # OriginalAdapter.new.enqueue_all(jobs)
+          enqueued_count = 0
+          jobs.group_by(&:class).each do |job_class, same_class_jobs|
+            same_class_jobs.group_by(&:queue_name).each do |queue, same_class_and_queue_jobs|
+              immediate_jobs, scheduled_jobs = same_class_and_queue_jobs.partition { |job| job.scheduled_at.nil? }
+
+              if immediate_jobs.any?
+                jids = Sidekiq::Client.push_bulk(
+                  "class" => JobWrapper,
+                  "wrapped" => job_class,
+                  "queue" => queue,
+                  "args" => immediate_jobs.map { |job| [job.serialize] },
+                )
+                enqueued_count += jids.compact.size
+              end
+
+              if scheduled_jobs.any?
+                jids = Sidekiq::Client.push_bulk(
+                  "class" => JobWrapper,
+                  "wrapped" => job_class,
+                  "queue" => queue,
+                  "args" => scheduled_jobs.map { |job| [job.serialize] },
+                  "at" => scheduled_jobs.map { |job| job.scheduled_at&.to_f }
+                )
+                enqueued_count += jids.compact.size
+              end
+            end
+          end
+          enqueued_count
+        end
+
+        # class JobWrapper # :nodoc:
+        #   include Sidekiq::Worker
+
+        #   def perform(job_data)
+        #     Base.execute job_data.merge("provider_job_id" => jid)
+        #   end
+        # end
+      end
+    end
+  end
+end
diff --git a/test/dummy/config/queue.yml b/test/dummy/config/queue.yml
index cce3e9cdb..ba036c27f 100644
--- a/test/dummy/config/queue.yml
+++ b/test/dummy/config/queue.yml
@@ -1,7 +1,7 @@
 default: &default
   workers:
     - queues: background
-      threads: 3
+      threads: 22
     - queues: default
       threads: 5
   dispatchers:
diff --git a/test/dummy/db/schema.rb b/test/dummy/db/schema.rb
index a52fb8202..6f42f9086 100644
--- a/test/dummy/db/schema.rb
+++ b/test/dummy/db/schema.rb
@@ -46,6 +46,22 @@
     t.index ["job_id"], name: "index_solid_queue_failed_executions_on_job_id", unique: true
   end
 
+  create_table "solid_queue_job_batches", charset: "utf8mb4", collation: "utf8mb4_0900_ai_ci", force: :cascade do |t|
+    t.bigint "parent_job_batch_id"
+    t.text "on_finish_active_job"
+    t.text "on_success_active_job"
+    t.text "on_failure_active_job"
+    t.datetime "finished_at"
+    t.datetime "changed_at"
+    t.datetime "last_changed_at"
+    t.datetime "created_at", null: false
+    t.datetime "updated_at", null: false
+    t.index ["changed_at"], name: "index_solid_queue_job_batches_on_changed_at"
+    t.index ["finished_at"], name: "index_solid_queue_job_batches_on_finished_at"
+    t.index ["last_changed_at"], name: "index_solid_queue_job_batches_on_last_changed_at"
+    t.index ["parent_job_batch_id"], name: "index_solid_queue_job_batches_on_parent_job_batch_id"
+  end
+
   create_table "solid_queue_jobs", charset: "utf8mb4", collation: "utf8mb4_0900_ai_ci", force: :cascade do |t|
     t.string "queue_name", null: false
     t.string "class_name", null: false
@@ -57,7 +73,9 @@
     t.string "concurrency_key"
     t.datetime "created_at", null: false
     t.datetime "updated_at", null: false
+    t.bigint "batch_id"
     t.index ["active_job_id"], name: "index_solid_queue_jobs_on_active_job_id"
+    t.index ["batch_id"], name: "index_solid_queue_jobs_on_batch_id"
     t.index ["class_name"], name: "index_solid_queue_jobs_on_class_name"
     t.index ["finished_at"], name: "index_solid_queue_jobs_on_finished_at"
     t.index ["queue_name", "finished_at"], name: "index_solid_queue_jobs_for_filtering"
@@ -140,10 +158,53 @@
     t.index ["key"], name: "index_solid_queue_semaphores_on_key", unique: true
   end
 
+  create_table "solid_queue_workflow_levels", charset: "utf8mb4", collation: "utf8mb4_0900_ai_ci", force: :cascade do |t|
+    t.bigint "workflow_id", null: false
+    t.string "workflow_level_id", null: false
+    t.string "parent_workflow_level_id"
+    t.datetime "started_at"
+    t.datetime "finished_at"
+    t.datetime "discarded_at"
+    t.datetime "created_at", null: false
+    t.datetime "updated_at", null: false
+    t.index ["discarded_at"], name: "index_solid_queue_workflow_levels_on_discarded_at"
+    t.index ["finished_at"], name: "index_solid_queue_workflow_levels_on_finished_at"
+    t.index ["started_at"], name: "index_solid_queue_workflow_levels_on_started_at"
+    t.index ["workflow_id"], name: "index_solid_queue_workflow_levels_on_workflow_id"
+    t.index ["workflow_level_id"], name: "index_solid_queue_workflow_levels_on_workflow_level_id", unique: true
+  end
+
+  create_table "solid_queue_workflow_nodes", charset: "utf8mb4", collation: "utf8mb4_0900_ai_ci", force: :cascade do |t|
+    t.bigint "workflow_id", null: false
+    t.string "active_job_id", null: false
+    t.text "active_job", null: false
+    t.string "workflow_level_id", null: false
+    t.string "parent_workflow_level_id"
+    t.datetime "started_at"
+    t.datetime "finished_at"
+    t.datetime "discarded_at"
+    t.datetime "created_at", null: false
+    t.datetime "updated_at", null: false
+    t.index ["discarded_at"], name: "index_solid_queue_workflow_nodes_on_discarded_at"
+    t.index ["finished_at"], name: "index_solid_queue_workflow_nodes_on_finished_at"
+    t.index ["started_at"], name: "index_solid_queue_workflow_nodes_on_started_at"
+    t.index ["workflow_id"], name: "index_solid_queue_workflow_nodes_on_workflow_id"
+  end
+
+  create_table "solid_queue_workflows", charset: "utf8mb4", collation: "utf8mb4_0900_ai_ci", force: :cascade do |t|
+    t.string "batch_id"
+    t.datetime "created_at", null: false
+    t.datetime "updated_at", null: false
+    t.index ["batch_id"], name: "index_solid_queue_workflows_on_batch_id"
+  end
+
   add_foreign_key "solid_queue_blocked_executions", "solid_queue_jobs", column: "job_id", on_delete: :cascade
   add_foreign_key "solid_queue_claimed_executions", "solid_queue_jobs", column: "job_id", on_delete: :cascade
   add_foreign_key "solid_queue_failed_executions", "solid_queue_jobs", column: "job_id", on_delete: :cascade
+  add_foreign_key "solid_queue_jobs", "solid_queue_job_batches", column: "batch_id", on_delete: :cascade
   add_foreign_key "solid_queue_ready_executions", "solid_queue_jobs", column: "job_id", on_delete: :cascade
   add_foreign_key "solid_queue_recurring_executions", "solid_queue_jobs", column: "job_id", on_delete: :cascade
   add_foreign_key "solid_queue_scheduled_executions", "solid_queue_jobs", column: "job_id", on_delete: :cascade
+  add_foreign_key "solid_queue_workflow_levels", "solid_queue_workflows", column: "workflow_id"
+  add_foreign_key "solid_queue_workflow_nodes", "solid_queue_workflows", column: "workflow_id"
 end
diff --git a/test/models/solid_queue/job_batch_test.rb b/test/models/solid_queue/job_batch_test.rb
new file mode 100644
index 000000000..e49f59c27
--- /dev/null
+++ b/test/models/solid_queue/job_batch_test.rb
@@ -0,0 +1,66 @@
+require "test_helper"
+
+class SolidQueue::JobBatchTest < ActiveSupport::TestCase
+  self.use_transactional_tests = false
+
+  teardown do
+    SolidQueue::Job.destroy_all
+    SolidQueue::JobBatch.destroy_all
+  end
+
+  class BatchWithArgumentsJob < ApplicationJob
+    def perform(batch, arg1, arg2)
+      Rails.logger.info "Hi #{batch.id}, #{arg1}, #{arg2}!"
+    end
+  end
+
+  class NiceJob < ApplicationJob
+    retry_on Exception, wait: 1.second
+
+    def perform(arg)
+      Rails.logger.info "Hi #{arg}!"
+    end
+  end
+
+  test "batch will be completed on success" do
+    batch = SolidQueue::JobBatch.enqueue(on_finish: BatchCompletionJob) { }
+    assert_not_nil batch.on_finish_active_job
+    assert_equal BatchCompletionJob.name, batch.on_finish_active_job["job_class"]
+  end
+
+  test "batch will be completed on finish" do
+    batch = SolidQueue::JobBatch.enqueue(on_success: BatchCompletionJob) { }
+    assert_not_nil batch.on_success_active_job
+    assert_equal BatchCompletionJob.name, batch.on_success_active_job["job_class"]
+  end
+
+  test "sets the batch_id on jobs created inside of the enqueue block" do
+    batch = SolidQueue::JobBatch.enqueue(on_finish: BatchCompletionJob) do
+      NiceJob.perform_later("world")
+      NiceJob.perform_later("people")
+    end
+
+    assert_equal 2, SolidQueue::Job.count
+    assert_equal [ batch.id ] * 2, SolidQueue::Job.last(2).map(&:batch_id)
+  end
+
+  test "batch id is present inside the block" do
+    assert_nil SolidQueue::JobBatch.current_batch_id
+    SolidQueue::JobBatch.enqueue(on_finish: BatchCompletionJob) do
+      assert_not_nil SolidQueue::JobBatch.current_batch_id
+    end
+    assert_nil SolidQueue::JobBatch.current_batch_id
+  end
+
+  test "allow arguments and options for callbacks" do
+    SolidQueue::JobBatch.enqueue(
+      on_finish: BatchWithArgumentsJob.new(1, 2).set(queue: :batch),
+    ) do
+      NiceJob.perform_later("world")
+    end
+
+    assert_not_nil SolidQueue::JobBatch.last.on_finish_active_job["arguments"]
+    assert_equal SolidQueue::JobBatch.last.on_finish_active_job["arguments"], [ 1, 2 ]
+    assert_equal SolidQueue::JobBatch.last.on_finish_active_job["queue_name"], "batch"
+  end
+end
diff --git a/test/models/solid_queue/workflow_test.rb b/test/models/solid_queue/workflow_test.rb
new file mode 100644
index 000000000..8d0dcccee
--- /dev/null
+++ b/test/models/solid_queue/workflow_test.rb
@@ -0,0 +1,267 @@
+require "test_helper"
+
+class SolidQueue::WorkflowTest < ActiveSupport::TestCase
+  self.use_transactional_tests = false
+
+  setup do
+    @pid = run_supervisor_as_fork(mode: :all)
+    wait_for_registered_processes(3, timeout: 3.second)
+  end
+
+  teardown do
+    terminate_process(@pid) if process_exists?(@pid)
+
+    SolidQueue::JobBatch.destroy_all
+    SolidQueue::WorkflowNode.destroy_all
+    SolidQueue::Workflow.destroy_all
+    SolidQueue::Process.destroy_all
+    SolidQueue::Job.destroy_all
+    RedisClient.new.call("flushdb")
+  end
+
+  class NiceJob < ApplicationJob
+    # retry_on Exception, wait: 1.second
+
+    def perform(arg)
+      puts "Hi #{arg}!"
+    end
+  end
+
+  class Oops < StandardError; end
+
+  class SidekiqJob < ApplicationJob
+    retry_on Exception, wait: 1.second, attempts: 3
+    discard_on Oops
+
+    sidekiq_options retry: 2
+
+    self.queue_adapter = :sidekiq
+
+    def perform(arg)
+      # raise Oops # This triggers and on_success...
+      puts "Hi #{arg}, from sidekiq!", "", ""
+      raise "eek"
+    end
+  end
+
+  # class GoodJobJob < ApplicationJob
+  #   retry_on Exception, wait: 1.second, attempts: 3
+  #   discard_on Oops
+
+  #   sidekiq_options retry: 2
+
+  #   self.queue_adapter = :good_job
+
+  #   def perform(arg)
+  #     puts "Hi #{arg}, from good job!", "", ""
+  #     # raise Oops
+  #     raise "eek"
+  #   end
+  # end
+
+  class GoodJobJob < ApplicationJob
+    self.queue_adapter = :good_job
+
+    def perform(arg)
+      puts "Hi #{arg}!"
+      # raise "doh" if arg == "6"
+    end
+  end
+
+  class GoodBatchCallbackJob < ApplicationJob
+    self.queue_adapter = :good_job
+
+    def perform(batch, params)
+      puts "Here's how many jobs we ran: #{batch.active_jobs.size}"
+    end
+  end
+
+  class MyCallback
+    def on_complete(status, options)
+      puts "Uh oh, batch has failures" if status.failures != 0
+    end
+
+    def on_success(status, options)
+      puts "#{options['uid']}'s batch succeeded.  Kudos!"
+    end
+  end
+
+  # When these jobs have finished, it will enqueue your `MyBatchCallbackJob.perform_later(batch, options)`
+  class MyBatchCallbackJob < ApplicationJob
+    self.queue_adapter = :good_job
+
+    # Callback jobs must accept a `batch` and `options` argument
+    def perform(batch, params)
+      ap params
+      ap GoodJob::Job.where(active_job_id: batch.active_jobs.map(&:job_id))
+    end
+  end
+
+  # test "good job batches" do
+  #   Thread.new do
+  #     capsule = GoodJob.capsule
+  #     capsule.start
+  #   end
+
+  #   # loop do
+  #   SolidQueue::Workflow.enqueue do |w|
+  #     w.run(GoodJobJob.new("1").set(wait_until: 1.second.from_now)) do
+  #       w.run(GoodJobJob.new("2")) do
+  #         w.run(GoodJobJob.new("3").set(wait_until: 2.seconds.from_now))
+  #         w.run(GoodJobJob.new("4"))
+  #       end
+  #       w.run(GoodJobJob.new("5")) do
+  #         w.run(GoodJobJob.new("6")) do
+  #           w.run(GoodJobJob.new("7"))
+  #           w.run(GoodJobJob.new("7.5"))
+  #           w.run(GoodJobJob.new("7.75"))
+  #           w.run(GoodJobJob.new("7.9")) do
+  #             w.run(GoodJobJob.new("7.95"))
+  #             w.run(GoodJobJob.new("7.97"))
+  #             w.run(GoodJobJob.new("7.99"))
+  #           end
+  #         end
+  #         w.run(GoodJobJob.new("6.5"))
+  #       end
+  #       w.run(GoodJobJob.new("8"), GoodJobJob.new("9"), GoodJobJob.new("10")) do
+  #         w.run(GoodJobJob.new("11"), GoodJobJob.new("12"), GoodJobJob.new("13"), GoodJobJob.new("14")) do
+  #           w.run(GoodJobJob.new("15")) do
+  #             w.run(GoodJobJob.new("15.1")) do
+  #               w.run(GoodJobJob.new("15.2"))
+  #             end
+  #           end
+  #           w.run(GoodJobJob.new("16"))
+  #           w.run(GoodJobJob.new("17"))
+  #           w.run(GoodJobJob.new("18")) do
+  #             w.run(GoodJobJob.new("19").set(wait_until: 3.seconds.from_now)) do
+  #               w.run(GoodJobJob.new("20"))
+  #             end
+  #             w.run(GoodJobJob.new("21"))
+  #             w.run(GoodJobJob.new("22"))
+  #             w.run(GoodJobJob.new("23"))
+  #           end
+  #         end
+  #       end
+  #     end
+  #   end
+  #   # sleep 5
+  #   # end
+
+  #   sleep 120
+  # end
+
+  # test "sidekiq batches" do
+  #   $sidekiq_instance = Sidekiq.configure_embed do |config|
+  #     config.queues = %w[background default realtime reports]
+  #     config.concurrency = 1
+  #   end
+  #   $sidekiq_instance.run
+
+  #   batch = Sidekiq::Batch.new
+  #   %i[success death].each do |event|
+  #     batch.on(
+  #       event,
+  #       SolidQueue::Job::WorkflowSidekiqBatchCallback,
+  #       stuff: 123,
+  #       and_junk: 234
+  #     )
+  #   end
+  #   batch.jobs do
+  #     ActiveJob.perform_all_later([
+  #       SidekiqJob.new("1"),
+  #       SidekiqJob.new("2"),
+  #       SidekiqJob.new("3")
+  #     ])
+  #   end
+
+  #   sleep 120
+  # end
+
+  test "sets the batch_id on jobs created inside of the enqueue block" do
+    SolidQueue::Workflow.enqueue do |w|
+      w.run(NiceJob.new("1").set(wait_until: 1.second.from_now)) do
+        w.run(NiceJob.new("2")) do
+          w.run(NiceJob.new("3").set(wait_until: 2.seconds.from_now))
+          w.run(NiceJob.new("4"))
+        end
+        w.run(NiceJob.new("5")) do
+          w.run(NiceJob.new("6")) do
+            w.run(NiceJob.new("7"))
+            w.run(NiceJob.new("7.5"))
+            w.run(NiceJob.new("7.75"))
+            w.run(NiceJob.new("7.9")) do
+              w.run(NiceJob.new("7.95"))
+              w.run(NiceJob.new("7.97"))
+              w.run(NiceJob.new("7.99"))
+            end
+          end
+          w.run(NiceJob.new("6.5"))
+        end
+        w.run(NiceJob.new("8"), NiceJob.new("9"), NiceJob.new("10")) do
+          w.run(NiceJob.new("11"), NiceJob.new("12"), NiceJob.new("13"), NiceJob.new("14")) do
+            w.run(NiceJob.new("15")) do
+              w.run(NiceJob.new("15.1")) do
+                w.run(NiceJob.new("15.2"))
+              end
+            end
+            w.run(NiceJob.new("16"))
+            w.run(NiceJob.new("17"))
+            w.run(NiceJob.new("18")) do
+              w.run(NiceJob.new("19").set(wait_until: 3.seconds.from_now)) do
+                w.run(NiceJob.new("20"))
+              end
+              w.run(NiceJob.new("21"))
+              w.run(NiceJob.new("22"))
+              w.run(NiceJob.new("23"))
+            end
+          end
+        end
+      end
+    end
+
+    # SolidQueue::Workflow.enqueue do |w|
+    #   w.run(NiceJob.new("1").set(wait_until: 1.second.from_now)) do
+    #     w.run(NiceJob.new("2")) do
+    #       w.run(NiceJob.new("3").set(wait_until: 2.seconds.from_now))
+    #       w.run(NiceJob.new("4"))
+    #     end
+    #     w.run(NiceJob.new("5")) do
+    #       w.run(NiceJob.new("6")) do
+    #         w.run(NiceJob.new("7"))
+    #       end
+    #     end
+    #     w.run(NiceJob.new("8"), NiceJob.new("9"), NiceJob.new("10")) do
+    #       w.run(NiceJob.new("11"), NiceJob.new("12"), NiceJob.new("13"), NiceJob.new("14")) do
+    #         w.run(NiceJob.new("15"))
+    #         w.run(NiceJob.new("16"))
+    #         w.run(NiceJob.new("17"))
+    #         w.run(NiceJob.new("18")) do
+    #           w.run(NiceJob.new("19").set(wait_until: 3.seconds.from_now))
+    #           w.run(NiceJob.new("20"))
+    #           w.run(NiceJob.new("21"))
+    #           w.run(NiceJob.new("22"))
+    #         end
+    #       end
+    #     end
+    #   end
+    # end
+    DiscardJob.perform_later("123")
+
+    sleep 10
+
+    binding.pry
+  end
+
+  class DiscardJob < ApplicationJob
+    # retry_on Exception, wait: 1.second, attempts: 3
+    discard_on Oops
+
+    # has_one :ready_execution
+    # has_one :claimed_execution
+    # has_one :failed_execution
+    def perform(arg)
+      puts "Hi #{arg}, time to discard!"
+      raise Oops # This triggers and on_success...
+    end
+  end
+end
diff --git a/test/test_helpers/jobs_test_helper.rb b/test/test_helpers/jobs_test_helper.rb
index d0833fcf2..f73458f01 100644
--- a/test/test_helpers/jobs_test_helper.rb
+++ b/test/test_helpers/jobs_test_helper.rb
@@ -9,6 +9,14 @@ def wait_for_jobs_to_finish_for(timeout = 1.second, except: [])
     end
   end
 
+  def wait_for_job_batches_to_finish_for(timeout = 1.second)
+    wait_while_with_timeout(timeout) do
+      skip_active_record_query_cache do
+        SolidQueue::JobBatch.where(finished_at: nil).any?
+      end
+    end
+  end
+
   def assert_no_unfinished_jobs
     skip_active_record_query_cache do
       assert SolidQueue::Job.where(finished_at: nil).none?