<commit_before>package Minion::Backend::Pg;
use Mojo::Base 'Minion::Backend';

use Carp 'croak';
use Mojo::IOLoop;
use Mojo::Pg 4.0;
use Sys::Hostname 'hostname';

has 'pg';

sub broadcast {
  my ($self, $command, $args, $ids) = (shift, shift, shift || [], shift || []);
  return !!$self->pg->db->query(
    q{update minion_workers set inbox = inbox || $1::jsonb
      where (id = any ($2) or $2 = '{}')}, {json => [[$command, @$args]]}, $ids
  )->rows;
}

sub dequeue {
  my ($self, $id, $wait, $options) = @_;

  if ((my $job = $self->_try($id, $options))) { return $job }
  return undef if Mojo::IOLoop->is_running;

  my $db = $self->pg->db;
  $db->listen('minion.job')->on(notification => sub { Mojo::IOLoop->stop });
  my $timer = Mojo::IOLoop->timer($wait => sub { Mojo::IOLoop->stop });
  Mojo::IOLoop->start;
  $db->unlisten('*') and Mojo::IOLoop->remove($timer);
  undef $db;

  return $self->_try($id, $options);
}

sub enqueue {
  my ($self, $task, $args, $options) = (shift, shift, shift || [], shift || {});

  my $db = $self->pg->db;
  return $db->query(
    "insert into minion_jobs
       (args, attempts, delayed, notes, parents, priority, queue, task)
     values (?, ?, (now() + (interval '1 second' * ?)), ?, ?, ?, ?, ?)
     returning id", {json => $args}, $options->{attempts} // 1,
    $options->{delay} // 0, {json => $options->{notes} || {}},
    $options->{parents} || [], $options->{priority} // 0,
    $options->{queue} // 'default', $task
  )->hash->{id};
}

sub fail_job   { shift->_update(1, @_) }
sub finish_job { shift->_update(0, @_) }

sub list_jobs {
  my ($self, $offset, $limit, $options) = @_;

  my $jobs = $self->pg->db->query(
    'select id, args, attempts,
       array(select id from minion_jobs where parents @> ARRAY[j.id])
         as children, extract(epoch from created) as created,
       extract(epoch from delayed) as delayed,
       extract(epoch from finished) as finished, notes, parents, priority,
       queue, result, extract(epoch from retried) as retried, retries,
       extract(epoch from started) as started, state, task,
       count(*) over() as total, worker
     from minion_jobs as j
     where (id = any ($1) or $1 is null) and (queue = $2 or $2 is null)
       and (state = $3 or $3 is null) and (task = $4 or $4 is null)
     order by id desc
     limit $5 offset $6', @$options{qw(ids queue state task)}, $limit, $offset
  )->expand->hashes->to_array;
  return _total('jobs', $jobs);
}

sub list_locks {
  my ($self, $offset, $limit, $options) = @_;

  my $locks = $self->pg->db->query(
    'select name, extract(epoch from expires) as expires,
       count(*) over() as total from minion_locks
     where (name = $1 or $1 is null) order by expires limit $2 offset $3',
    $options->{name}, $limit, $offset
  )->expand->hashes->to_array;
  return _total('locks', $locks);
}

sub list_workers {
  my ($self, $offset, $limit, $options) = @_;<commit_msg>order list_* methods consistently<commit_after>package Minion::Backend::Pg;
use Mojo::Base 'Minion::Backend';

use Carp 'croak';
use Mojo::IOLoop;
use Mojo::Pg 4.0;
use Sys::Hostname 'hostname';

has 'pg';

sub broadcast {
  my ($self, $command, $args, $ids) = (shift, shift, shift || [], shift || []);
  return !!$self->pg->db->query(
    q{update minion_workers set inbox = inbox || $1::jsonb
      where (id = any ($2) or $2 = '{}')}, {json => [[$command, @$args]]}, $ids
  )->rows;
}

sub dequeue {
  my ($self, $id, $wait, $options) = @_;

  if ((my $job = $self->_try($id, $options))) { return $job }
  return undef if Mojo::IOLoop->is_running;

  my $db = $self->pg->db;
  $db->listen('minion.job')->on(notification => sub { Mojo::IOLoop->stop });
  my $timer = Mojo::IOLoop->timer($wait => sub { Mojo::IOLoop->stop });
  Mojo::IOLoop->start;
  $db->unlisten('*') and Mojo::IOLoop->remove($timer);
  undef $db;

  return $self->_try($id, $options);
}

sub enqueue {
  my ($self, $task, $args, $options) = (shift, shift, shift || [], shift || {});

  my $db = $self->pg->db;
  return $db->query(
    "insert into minion_jobs
       (args, attempts, delayed, notes, parents, priority, queue, task)
     values (?, ?, (now() + (interval '1 second' * ?)), ?, ?, ?, ?, ?)
     returning id", {json => $args}, $options->{attempts} // 1,
    $options->{delay} // 0, {json => $options->{notes} || {}},
    $options->{parents} || [], $options->{priority} // 0,
    $options->{queue} // 'default', $task
  )->hash->{id};
}

sub fail_job   { shift->_update(1, @_) }
sub finish_job { shift->_update(0, @_) }

sub list_jobs {
  my ($self, $offset, $limit, $options) = @_;

  my $jobs = $self->pg->db->query(
    'select id, args, attempts,
       array(select id from minion_jobs where parents @> ARRAY[j.id])
         as children, extract(epoch from created) as created,
       extract(epoch from delayed) as delayed,
       extract(epoch from finished) as finished, notes, parents, priority,
       queue, result, extract(epoch from retried) as retried, retries,
       extract(epoch from started) as started, state, task,
       count(*) over() as total, worker
     from minion_jobs as j
     where (id = any ($1) or $1 is null) and (queue = $2 or $2 is null)
       and (state = $3 or $3 is null) and (task = $4 or $4 is null)
     order by id desc
     limit $5 offset $6', @$options{qw(ids queue state task)}, $limit, $offset
  )->expand->hashes->to_array;
  return _total('jobs', $jobs);
}

sub list_locks {
  my ($self, $offset, $limit, $options) = @_;

  my $locks = $self->pg->db->query(
    'select name, extract(epoch from expires) as expires,
       count(*) over() as total from minion_locks
     where (name = $1 or $1 is null) order by id desc limit $2 offset $3',
    $options->{name}, $limit, $offset
  )->hashes->to_array;
  return _total('locks', $locks);
}

sub list_workers {
  my ($self, $offset, $limit, $options) = @_;
{'fraction_non_alphanumeric': 0.19935197817189632, 'fraction_numerical': 0.008526603001364257, 'mean_word_length': 3.4839449541284404, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 100, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6240356', 'n_tokens_mistral': 2133, 'n_tokens_neox': 2081, 'n_words': 612}