class Riffer::Rig::Runtime::Jobs

# File lib/riffer/rig/runtime/jobs.rb, line 28
def initialize(inbox, clock: -> { Process.clock_gettime(Process::CLOCK_MONOTONIC) })
  @inbox = inbox
  @clock = clock
  @jobs = {}
  @groups = []
  @logs = {}
  @directory = nil
  @mutex = Mutex.new
  @watchers = ThreadGroup.new
end

# File lib/riffer/rig/runtime/jobs.rb, line 101
def consume
  completions = @inbox.drain
  @mutex.synchronize { completions.each { |completion| @jobs.delete(completion.job_id) } }
  completions
end

# File lib/riffer/rig/runtime/jobs.rb, line 87
def kill(id)
  @mutex.synchronize do
    @jobs.fetch(id, nil)&.tap do |job|
      # A completed job's pid is already reaped and may belong to another process.
      next unless job.active?

      signal_group(job.pid, 'TERM')
      # A stopped process holds TERM pending until it is continued.
      signal_group(job.pid, 'CONT')
    end
  end
end

# File lib/riffer/rig/runtime/jobs.rb, line 64
def list
  @mutex.synchronize { @jobs.values }
end

# File lib/riffer/rig/runtime/jobs.rb, line 71
def read(id, limit)
  @mutex.synchronize do
    job = @jobs.fetch(id, nil)
    return unless job

    output, written = @logs.fetch(id).read(job.offset, limit)
    # Output past the read limit is dropped rather than queued, so a job writing
    # faster than it is read cannot build a backlog.
    advanced = job.advance(written - job.offset)
    @jobs[id] = advanced
    [advanced, output]
  end
end

# File lib/riffer/rig/runtime/jobs.rb, line 43
def register(pid, command, output)
  id = SecureRandom.hex(4)
  path = File.join(directory, id)
  log = Riffer::Rig::Runtime::JobLog.new(output, path)
  job = Riffer::Rig::Runtime::Job.new(
    id: id,
    pid: pid,
    command: command,
    output_path: path,
    started_at: @clock.call
  )
  @mutex.synchronize do
    @jobs[id] = job
    @groups << pid
    @logs[id] = log
  end
  @watchers.add(Thread.new { watch(job, log) })
  job
end

# File lib/riffer/rig/runtime/jobs.rb, line 109
def shutdown(grace = SHUTDOWN_GRACE_SECONDS)
  kill_outstanding('TERM')
  # A stopped process holds TERM pending until it is continued.
  kill_outstanding('CONT')
  deadline = @clock.call + grace
  @watchers.list.each { |watcher| watcher.join([deadline - @clock.call, 0.0].max) }
  # A job that ignores TERM would otherwise hold its watcher, and this join, forever.
  kill_outstanding('KILL')
  # A killed process stays in its group until its parent reaps it. Once the
  # leader is gone that parent is not this process and may never do so, so
  # the watchers are released rather than left polling.
  @mutex.synchronize { @groups.clear }
  @watchers.list.each(&:join)
  logs, directory = @mutex.synchronize { [@logs.values, @directory] }
  # A process that left its job's group can hold the pipe open indefinitely.
  logs.each(&:close)
  FileUtils.rm_rf(directory) if directory
end