view release on metacpan or  search on metacpan

lib/Minion/Backend/SQLite.pm  view on Meta::CPAN

package Minion::Backend::SQLite;
use Mojo::Base 'Minion::Backend';

use Carp 'croak';
use DBD::SQLite::Constants ':result_codes';
use List::Util 'min';
use Minion::Util 'next_cron_time';
use Mojo::SQLite;
use Mojo::Util 'steady_time';
use Sys::Hostname 'hostname';
use Time::HiRes 'usleep';

our $VERSION = 'v6.0.0';

has dequeue_interval => 0.5;
has 'sqlite';

sub new {
  my $self = shift->SUPER::new(sqlite => Mojo::SQLite->new(@_));
  $self->sqlite->auto_migrate(1)->migrations->name('minion')->from_data;
  return $self;
}

sub broadcast {
  my ($self, $command, $args, $ids) = (shift, shift, shift || [], shift || []);
  my $ids_in = join ',', ('?')x@$ids;
  return !!$self->sqlite->db->query(
    q{update minion_workers set inbox =
      json_set(inbox, '$[' || json_array_length(inbox) || ']', json(?))} .
      (@$ids ? " where id in ($ids_in)" : ''),
      {json => [$command, @$args]}, @$ids
  )->rows;
}

sub dequeue {
  my ($self, $id, $wait, $options) = @_;
  my $job = $self->_try($id, $options);
  unless ($job) {
    my $int = $self->dequeue_interval;
    my $end = steady_time + $wait;
    my $remaining = $wait;
    usleep(min($int, $remaining) * 1000000)
      until ($remaining = $end - steady_time) <= 0
      or $job = $self->_try($id, $options);
  }
  return $job || $self->_try($id, $options);
}

sub dispatch_schedules {
  my $self = shift;

  my $db = $self->sqlite->db;
  my $timeout = $db->dbh->sqlite_busy_timeout // 0;
  $db->dbh->sqlite_busy_timeout(1000);
  my ($tx, $errored, $error);
  {
    local $@;
    unless (eval { $tx = $db->begin('immediate'); 1 }) {
      my $errcode = $db->dbh->err;
      if ($errcode and $errcode == SQLITE_BUSY) {
        $db->dbh->sqlite_busy_timeout($timeout);
        $db->dbh->rollback; # clear internal transaction state
        return [];
      } else {
        $errored = 1;
        $error = $@;
      }
    }
  }
  $db->dbh->sqlite_busy_timeout($timeout);
  die $error if $errored;

  my $due = $db->query(
    q{select id, args, attempts, cron, expire, lax, name, strftime('%s',next_run) as next_run,
        strftime('%s','now') as now, notes, priority, queue, task
      from minion_schedules where next_run <= datetime('now') and not paused}
  )->expand(json => [qw(args notes)])->hashes;

  my @dispatched;
  for my $s (@$due) {
    my $job_id = $self->_enqueue(
      $db,
      $s->{task},
      $s->{args},
      {
        attempts => $s->{attempts},
        expire   => $s->{expire},
        lax      => $s->{lax},
        notes    => $s->{notes},
        priority => $s->{priority},
        queue    => $s->{queue}
      }
    );
    my $basis = $s->{next_run} > $s->{now} ? $s->{next_run} : $s->{now};
    my $next  = next_cron_time $s->{cron}, $basis;
    $db->query(q{update minion_schedules set last_job = ?, last_run = datetime('now'),
      next_run = datetime(?,'unixepoch') where id = ?},
      $job_id, $next, $s->{id});
    push @dispatched, {id => $s->{id}, name => $s->{name}, job => $job_id};
  }
  $tx->commit;

  return \@dispatched;
}

sub enqueue {
  my $self = shift;
  return $self->_enqueue($self->sqlite->db, @_);
}

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

sub history {
  my $self = shift;

  my $db = $self->sqlite->db;
  my $steps = $db->query(
    q{with recursive generate_series(ts) as (
        select datetime('now','-23 hours')
        union all
        select datetime(ts,'+1 hour') from generate_series
        where datetime(ts,'+1 hour') <= datetime('now')
      ) select ts, strftime('%s',ts) as epoch,
        strftime('%d',ts,'localtime') as day,
        strftime('%H',ts,'localtime') as hour
      from generate_series order by epoch})->hashes;

  my $counts = $db->query(
    q{select strftime('%d',finished,'localtime') as day,