prism/test/test_cli_jobs.rb

require 'minitest/autorun'
require 'prism'
require 'tmpdir'
require 'fileutils'
require 'stringio'

class CliJobsEchoJob
  def self.log
    @log ||= []
  end

  def perform(value)
    self.class.log << value
  end
end

class CliJobsAlwaysFailsJob
  def perform
    raise 'boom'
  end
end

class CliJobsTest < Minitest::Test
  def setup
    @dir = Dir.mktmpdir('prism-cli-jobs')
    @cli = Prism::Cli.new(@dir)
    @queue_path = File.join(@dir, 'db', 'jobs.pstore')
    CliJobsEchoJob.log.clear
  end

  def teardown
    FileUtils.remove_entry(@dir)
  end

  def queue
    Prism::Jobs::Queue.new(@queue_path)
  end

  def test_jobs_work_once_drains_ready_jobs_and_returns_a_single_cycle
    queue.enqueue(job_class: 'CliJobsEchoJob', args: ['a'])
    queue.enqueue(job_class: 'CliJobsEchoJob', args: ['b'])

    out = StringIO.new
    cycles = @cli.jobs_work(['--once', '--verbose'], out: out)

    assert_equal 1, cycles
    assert_equal %w[a b], CliJobsEchoJob.log.sort
    assert_equal 0, queue.size
    assert_match(/processed 2 job\(s\)/, out.string)
  end

  def test_jobs_work_respects_custom_file_option
    custom_path = File.join(@dir, 'custom_jobs.pstore')
    Prism::Jobs::Queue.new(custom_path).enqueue(job_class: 'CliJobsEchoJob', args: ['custom'])

    @cli.jobs_work(['--once', '--file=custom_jobs.pstore'])

    assert_equal ['custom'], CliJobsEchoJob.log
    assert_equal 0, Prism::Jobs::Queue.new(custom_path).size
  end

  def test_jobs_work_scopes_to_queue_name
    queue.enqueue(job_class: 'CliJobsEchoJob', args: ['default'])
    queue.enqueue(job_class: 'CliJobsEchoJob', args: ['mailers'], queue_name: 'mailers')

    @cli.jobs_work(['--once', '--queue=mailers'])

    assert_equal ['mailers'], CliJobsEchoJob.log
    assert_equal 1, queue.size
  end

  def test_jobs_work_stops_after_max_cycles_without_sleeping_forever
    cycles = @cli.jobs_work(['--max-cycles=3', '--interval=0'])

    assert_equal 3, cycles
  end

  def test_jobs_work_does_not_crash_on_failing_jobs_and_reports_via_error_output
    queue.enqueue(job_class: 'CliJobsAlwaysFailsJob', args: [], max_attempts: 1)

    _out, err = capture_io { @cli.jobs_work(['--once']) }

    assert_match(/CliJobsAlwaysFailsJob.*failed: boom/, err)
    assert_equal 1, queue.dead_letter_size
  end

  def test_jobs_work_requires_extra_files_before_running
    job_file = File.join(@dir, 'extra_job.rb')
    File.write(job_file, <<~RUBY)
      class CliJobsRequiredJob
        def perform(marker_path)
          File.write(marker_path, 'ran')
        end
      end
    RUBY
    marker_path = File.join(@dir, 'marker.txt')
    queue.enqueue(job_class: 'CliJobsRequiredJob', args: [marker_path])

    @cli.jobs_work(['--once', '--require=extra_job.rb'])

    assert_equal 'ran', File.read(marker_path)
  end

  def test_jobs_work_rejects_unknown_options
    assert_raises(Prism::Cli::UsageError) { @cli.jobs_work(['--bogus']) }
  end

  def test_run_dispatches_jobs_work_through_the_public_cli_entrypoint
    queue.enqueue(job_class: 'CliJobsEchoJob', args: ['via-run'])

    @cli.run(['jobs', 'work', '--once'])

    assert_equal ['via-run'], CliJobsEchoJob.log
  end

  def test_run_rejects_unknown_jobs_subcommand
    assert_raises(Prism::Cli::UsageError) { @cli.run(['jobs', 'bogus']) }
  end

  def test_jobs_work_writes_its_own_pid_while_running_and_removes_it_on_exit
    pidfile = File.join(@dir, 'jobs.pid')

    thread = Thread.new { @cli.jobs_work(['--max-cycles=3', '--interval=0.2', "--pidfile=#{pidfile}"]) }
    sleep 0.05 until File.exist?(pidfile)
    assert_equal Process.pid.to_s, File.read(pidfile).strip

    thread.join

    refute File.exist?(pidfile), 'pidfile should be removed once the worker stops'
  end

  def test_jobs_work_refuses_to_start_while_a_live_process_holds_the_pidfile
    pidfile = File.join(@dir, 'jobs.pid')
    # A genuinely different, currently-alive OS process (a plain child that
    # sleeps) - the same scenario a second real daemon would face, unlike
    # writing our own pid, which the cleanup step would then be entitled to
    # remove because it would truly be ours.
    other_pid = Process.spawn('sleep', '5')
    File.write(pidfile, other_pid.to_s)

    error = assert_raises(Prism::Cli::UsageError) { @cli.jobs_work(['--once', "--pidfile=#{pidfile}"]) }

    assert_match(/already running/, error.message)
    assert_equal other_pid.to_s, File.read(pidfile).strip, "the other process's pidfile must be left untouched"
  ensure
    Process.kill('TERM', other_pid)
    Process.wait(other_pid)
  end

  def test_jobs_work_reclaims_a_stale_pidfile_left_by_a_dead_process
    pidfile = File.join(@dir, 'jobs.pid')
    dead_pid = fork { exit }
    Process.wait(dead_pid)
    File.write(pidfile, dead_pid.to_s)

    queue.enqueue(job_class: 'CliJobsEchoJob', args: ['after-stale-pid'])
    @cli.jobs_work(['--once', "--pidfile=#{pidfile}"])

    assert_equal ['after-stale-pid'], CliJobsEchoJob.log
    refute File.exist?(pidfile)
  end

  # Real end-to-end daemon smoke test: runs `bin/prism jobs work` as its own
  # OS process (no --once, no --max-cycles), lets it poll and drain a job,
  # then sends it SIGINT the same way an operator or init system would and
  # confirms it shuts down cleanly rather than needing to be killed.
  def test_daemon_process_drains_jobs_and_shuts_down_cleanly_on_sigint
    job_file = File.join(@dir, 'daemon_job.rb')
    marker_path = File.join(@dir, 'daemon_marker.txt')
    File.write(job_file, <<~RUBY)
      class CliJobsDaemonJob
        def perform(marker_path)
          File.write(marker_path, 'ran')
        end
      end
    RUBY
    queue.enqueue(job_class: 'CliJobsDaemonJob', args: [marker_path])

    bin = File.expand_path('../bin/prism', __dir__)
    pid = Process.spawn(
      bin, 'jobs', 'work', '--interval=0.05', '--require=daemon_job.rb',
      chdir: @dir, out: File::NULL, err: File::NULL
    )

    deadline = Time.now + 5
    sleep 0.05 until File.exist?(marker_path) || Time.now > deadline
    assert File.exist?(marker_path), 'daemon did not process the job within 5s'
    assert_equal 'ran', File.read(marker_path)

    Process.kill('INT', pid)
    _, status = Process.wait2(pid)
    assert status.success?, "daemon did not exit cleanly after SIGINT (status: #{status})"
  end

  # Real horizontal-scaling scenario: two independent `bin/prism jobs work`
  # daemons (separate OS processes, no pidfile since neither is meant to be
  # "the only one") polling the SAME queue file. Jobs::Queue#claim_next! is
  # already proven race-free at the storage layer (test_jobs_queue.rb spawns
  # 4 bare claimer processes), but that does not by itself prove the CLI
  # daemon loop (poll -> work_off -> sleep -> repeat, plus its own error
  # handling/verbose logging) is safe to run in more than one copy - this
  # drives the actual `bin/prism jobs work` entrypoint twice and checks both
  # that every job runs exactly once AND that work genuinely got split
  # between the two processes rather than one daemon draining the whole
  # queue before the other gets a chance to poll.
  def test_two_real_daemon_processes_share_the_same_queue_without_dropping_or_duplicating_jobs
    job_count = 60
    results_dir = File.join(@dir, 'daemon_results')
    FileUtils.mkdir_p(results_dir)

    job_file = File.join(@dir, 'multi_daemon_job.rb')
    File.write(job_file, <<~RUBY)
      class CliJobsMultiDaemonJob
        def perform(id, results_dir)
          sleep 0.02
          File.write(File.join(results_dir, "job-\#{id}.done"), Process.pid.to_s)
        end
      end
    RUBY

    job_count.times { |i| queue.enqueue(job_class: 'CliJobsMultiDaemonJob', args: [i, results_dir]) }

    bin = File.expand_path('../bin/prism', __dir__)
    pids = 2.times.map do
      Process.spawn(
        bin, 'jobs', 'work', '--interval=0.01', '--require=multi_daemon_job.rb',
        chdir: @dir, out: File::NULL, err: File::NULL
      )
    end

    deadline = Time.now + 10
    sleep 0.05 while queue.size.positive? && Time.now < deadline

    pids.each { |pid| Process.kill('INT', pid) }
    pids.each { |pid| Process.wait2(pid) }

    assert_equal 0, queue.size, 'both daemons together should have drained the whole queue within the deadline'

    result_files = Dir.glob(File.join(results_dir, 'job-*.done'))
    assert_equal job_count, result_files.size, 'every job should have run exactly once (no drops, no duplicates)'
    expected_names = (0...job_count).map { |i| "job-#{i}.done" }.sort
    assert_equal expected_names, result_files.map { |f| File.basename(f) }.sort

    processing_pids = result_files.map { |f| File.read(f).strip }.uniq
    assert_equal pids.map(&:to_s).sort, processing_pids.sort,
                 'both daemon processes should have actually processed jobs, not just one of them'
  end
end