File Coverage

blib/lib/TheSchwartz/JobScheduler.pm

Criterion	Covered	Total	%
statement	179	197	90.8
branch	45	62	72.5
condition	9	18	50.0
subroutine	20	21	95.2
pod	3	3	100.0
total	256	301	85.0

package TheSchwartz::JobScheduler;
## no critic (ControlStructures::ProhibitPostfixControls)
## no critic (Subroutines::RequireArgUnpacking)
## no critic (ControlStructures::ProhibitUnlessBlocks)

use strict;
use warnings;

# ABSTRACT: Lightweight TheSchwartz job dispatcher with maintained database connections

our $VERSION = '0.002'; # VERSION: generated by DZP::OurPkgVersion

use Carp;
use English '-no_match_vars';
use Storable;
use Module::Load qw( load );
use Scalar::Util qw( refaddr );
use Const::Fast;

use Moo;
use Log::Any qw( $log ), hooks => { build_context => [ \&_build_context, ], };
use Log::Any::Adapter::Util;

sub _build_context {

    # my ($level, $category, $data) = @_;
    my %ctx;
    my @caller = Log::Any::Adapter::Util::get_correct_caller();
    $ctx{file} = $caller[1];
    $ctx{line} = $caller[2];
    return %ctx;
}
use Carp::Assert;

use TheSchwartz::JobScheduler::Job;

const my $DEFAULT_OPTION_UNIQKEY => 'no_check';

has databases => (
    is       => 'ro',
    required => 1,
);

has dbh_callback => (
    is       => 'ro',
    required => 0,
);

has _funcmap => (
    is      => 'ro',
    default => sub { {}; },
);

has opts => (
    is      => 'ro',
    default => sub {
        return { handle_uniqkey => $DEFAULT_OPTION_UNIQKEY, };
    },
);

sub insert {
    my ( $self, %args ) = @_;

    # use Data::Dumper;
    # warn Dumper \%args;

    croak q{No job} unless exists $args{'job'};
    my $job = $args{'job'};
    croak q{Argument job not TheSchwartz::JobScheduler::Job} unless $job->isa('TheSchwartz::JobScheduler::Job');

    # my $databases = $args{'databases'};
    my $databases = $self->databases;
    $job->arg( Storable::nfreeze( $job->arg ) ) if ref $job->arg;

    # $log->debugf( 'TheSchwartz::JobScheduler::insert(): databases: %s', $databases );
    foreach my $database_id ( keys %{$databases} ) {

        # $log->debugf( 'TheSchwartz::JobScheduler::insert(): db: %s', $db );

        my $dbh_callback = exists $args{'dbh_callback'} ? $args{'dbh_callback'} : $self->dbh_callback;
        my $dbh          = _get_dbh( $database_id, $dbh_callback );

        # $log->debugf( 'TheSchwartz::JobScheduler::insert(): dbh: %s', $dbh );
        my $prefix = $databases->{$database_id}->{'prefix'} // q{};

        # $log->debugf( 'TheSchwartz::JobScheduler::insert(): prefix: %s', $prefix );

        my $jobid;

        # local $EVAL_ERROR = undef;
        # my $r = eval {
        # $job->funcid( $self->funcname_to_id( $database, $job->funcname ) );
        $job->funcid( $self->funcname_to_id( $dbh, $prefix, $job->funcname ) );
        $job->insert_time(time);

        if ( $job->uniqkey && $self->opts->{'handle_uniqkey'} eq 'acknowledge' ) {
            my $row        = $job->as_hashref;
            my @query_cols = qw( jobid );
            my @where_cols = qw( funcid uniqkey );
            my $sql        = sprintf 'SELECT %s FROM %sjob WHERE funcid = ? AND uniqkey = ?', ( join q{, }, @query_cols ), $prefix;

            my $sth = $dbh->prepare_cached($sql);
            my $i   = 1;
            for my $where_col (@where_cols) {
                $sth->bind_param( $i++, $row->{$where_col}, _bind_param_attr( $dbh, $prefix, $where_col ), );
            }
            $sth->execute();

            # Strange if there would be more than one entry!
            my @job_ids;
            while ( my $ref = $sth->fetchrow_arrayref ) {
                push @job_ids, $ref->[0];
            }

            # $sth->finish;
            # https://metacpan.org/pod/DBI#finish
            # Indicate that no more data will be fetched from this statement
            # handle before it is either executed again or destroyed.
            # You almost certainly do not need to call this method.
            # Adding calls to finish after loop that fetches all rows
            # is a common mistake, don't do it, it can mask genuine
            # problems like uncaught fetch errors.
            return $job_ids[0] if (@job_ids);
        }

        my $row = $job->as_hashref;
        my @col = keys %{$row};
        my $sql = sprintf 'INSERT INTO %sjob (%s) VALUES (%s)', $prefix, ( join q{, }, @col ), ( join q{, }, (q{?}) x @col );

        my $sth = $dbh->prepare_cached($sql);
        my $i   = 1;
        for my $col (@col) {
            $sth->bind_param( $i++, $row->{$col}, _bind_param_attr( $dbh, $prefix, $col ), );
        }
        $sth->execute();

        $jobid = _insert_id( $dbh, $prefix, $sth, "${prefix}job", 'jobid' );

        $log->debugf( 'TheSchwartz::JobScheduler::insert() jobid: %s', $jobid );
        return $jobid if defined $jobid;
    }

    return;
}

sub funcname_to_id {
    my ( $self, $dbh, $prefix, $funcname ) = @_;
    $log->debugf( 'TheSchwartz::JobScheduler::funcname_to_id(%s, %s, %s)', $dbh, $prefix, $funcname );

    # my ( $dbh, $prefix ) = ($database->dbh, $database->prefix);

    my $dbid = refaddr $dbh;
    unless ( exists $self->_funcmap->{$dbid} ) {
        my $sth = $dbh->prepare_cached("SELECT funcid, funcname FROM ${prefix}funcmap");
        $sth->execute;
        while ( my $row = $sth->fetchrow_arrayref ) {
            $self->_funcmap->{$dbid}{ $row->[1] } = $row->[0];
        }

        # $sth->finish;
        # See above
    }

    unless ( exists $self->_funcmap->{$dbid}{$funcname} ) {
        ## This might fail in a race condition since funcname is UNIQUE
        my $sth = $dbh->prepare_cached("INSERT INTO ${prefix}funcmap (funcname) VALUES (?)");
        local $EVAL_ERROR = undef;
        my $r = eval { $sth->execute($funcname) };
        if ( !$r ) {
            my $error = $EVAL_ERROR;
            $log->warn( ' Unable to insert the funcname \'%s\'. Error: %s', $funcname, $error );
        }

        my $id = _insert_id( $dbh, $prefix, $sth, "${prefix}funcmap", 'funcid' );

        ## If we got an exception, try to load the record again
        if ($EVAL_ERROR) {
            $sth = $dbh->prepare_cached("SELECT funcid FROM ${prefix}funcmap WHERE funcname = ?");
            $sth->execute($funcname);
            $id = $sth->fetchrow_arrayref->[0]
              or croak "Can't find or create funcname $funcname: $EVAL_ERROR";
        }

        $self->_funcmap->{$dbid}{$funcname} = $id;
    }

    $log->debugf( 'TheSchwartz::JobScheduler::funcname_to_id(): %s', $self->_funcmap->{$dbid}{$funcname} );
    return $self->_funcmap->{$dbid}{$funcname};
}

sub _insert_id {
    my ( $dbh, $prefix, $sth, $table, $col ) = @_;

    # my ( $dbh, $prefix ) = ($database->dbh, $database->prefix);

    my $driver = $dbh->{Driver}{Name};
    if ( $driver eq 'mysql' ) {
        return $dbh->{mysql_insertid};
    }
    elsif ( $driver eq 'Pg' ) {
        return $dbh->last_insert_id( undef, undef, undef, undef, { sequence => ( join q{_}, $table, $col, 'seq' ) } );
    }
    elsif ( $driver eq 'SQLite' ) {
        return $dbh->func('last_insert_rowid');
    }
    else {
        croak "Don't know how to get last insert id for $driver";
    }
}

sub list_jobs {
    my ( $self, %args ) = @_;
    $log->debugf( 'TheSchwartz::JobScheduler::list_jobs(%s)', \%args );
    croak q{No search_params} unless exists $args{'search_params'};
    my $search_params = $args{'search_params'};

    croak q{No funcname} unless exists $search_params->{funcname};

    my @options;
    push @options,
      {
        key   => 'run_after',
        op    => '<=',
        value => $search_params->{run_after}
      } if exists $search_params->{run_after};
    push @options,
      {
        key   => 'grabbed_until',
        op    => '<=',
        value => $search_params->{grabbed_until}
      }
      if exists $search_params->{grabbed_until};

    if ( $search_params->{coalesce} ) {
        $search_params->{coalesce_op} ||= q{=};
        push @options,
          {
            key   => 'coalesce',
            op    => $search_params->{coalesce_op},
            value => $search_params->{coalesce}
          };
    }

    my @jobs;
    my $databases = $self->databases;
    foreach my $database_id ( keys %{$databases} ) {
        my $dbh_callback = exists $args{'dbh_callback'} ? $args{'dbh_callback'} : $self->dbh_callback;
        my $dbh          = _get_dbh( $database_id, $dbh_callback );
        my $prefix       = $databases->{$database_id}->{'prefix'} // q{};

        local $EVAL_ERROR = undef;
        my $r = eval {
            my $funcid = $self->funcname_to_id( $dbh, $prefix, $search_params->{funcname} );

            my $sql   = "SELECT * FROM ${prefix}job WHERE funcid = ?";
            my @value = ($funcid);
            for (@options) {
                $sql .= " AND $_->{key} $_->{op} ?";
                push @value, $_->{value};
            }

            my $sth = $dbh->prepare_cached($sql);
            $sth->execute(@value);
            while ( my $ref = $sth->fetchrow_hashref ) {
                $log->debugf( 'TheSchwartz::JobScheduler::list_jobs(): fetch:ref: %s', $ref );

                # my $job_fields = Storable::dclone( $ref );
                # $job->search_params( Storable::nfreeze( $job->search_params ) ) if ref $job->search_params;
                my $arg_tmp = _cond_thaw( $ref->{'arg'} );
                my $job     = TheSchwartz::JobScheduler::Job->new($ref);
                $job->arg($arg_tmp);
                push @jobs, $job;
            }
            1;
        };
        if ( !$r ) {
            my $error = $EVAL_ERROR;
            $log->warn( ' Unable to fetch jobs for funcname \'%s\' (id: %s). Error: %s', $search_params->{funcname}, $error );
        }
    }

    $log->debugf( 'TheSchwartz::JobScheduler::list_jobs(): %s', \@jobs );
    return @jobs;
}

sub _bind_param_attr {
    my ( $dbh, $prefix, $col ) = @_;

    # my ( $dbh, $prefix ) = ( $database->dbh, $database->prefix );

    return if $col ne 'arg';

    my $driver = $dbh->{Driver}{Name};
    if ( $driver eq 'Pg' ) {
        return { pg_type => DBD::Pg::PG_BYTEA() };
    }
    elsif ( $driver eq 'SQLite' ) {
        return DBI::SQL_BLOB();
    }
    return;
}

# Shamelessly copied from TheSchwartz::Job
# Perl::Critic applied
sub _cond_thaw {
    my $data = shift;

    my $magic = eval { Storable::read_magic($data); };
    if (   $magic
        && $magic->{major}
        && $magic->{major} >= 2
        && $magic->{major} <= 5 )    ## no critic (ValuesAndExpressions::ProhibitMagicNumbers)
    {
        my $thawed = eval { Storable::thaw($data) };
        if ($@) {                    ## no critic (Variables::ProhibitPunctuationVars)

            # false alarm... looked like a Storable, but wasn't.
            return $data;
        }
        return $thawed;
    }
    else {
        return $data;
    }
}

sub _get_dbh {
    my ( $database_id, $dbh_callback ) = @_;

    # my $cb = $dbh_callback;
    my $dbh;
    if ( ref $dbh_callback eq 'CODE' ) {
        $dbh = &{$dbh_callback}($database_id);
    }
    else {
        my $dbh_callback_code = $dbh_callback;
        my ( $module, $creator ) = split qr/\-\>/msx, $dbh_callback_code;
        local $EVAL_ERROR = undef;
        my $r = eval { load $module };
        if ($EVAL_ERROR) {
            croak q{Cannot load dbh_callback module '}, $module, q{'};
        }
        my $callback;
        local $EVAL_ERROR = undef;
        $r = eval { $callback = $module->$creator() };
        if ($EVAL_ERROR) {
            croak q{Cannot instantiate dbh_callback module '}, "$module->$creator", q{'};
        }
        local $EVAL_ERROR = undef;
        $r = eval { $dbh = $callback->dbh($database_id) };
        if ($EVAL_ERROR) {
            croak q{Cannot get dbh from callback '}, "$module->$creator->dbh( $database_id )", q{'};
        }
    }

    return $dbh;
}

1;

__END__

=pod

=encoding UTF-8

=head1 NAME

TheSchwartz::JobScheduler - Lightweight TheSchwartz job dispatcher with maintained database connections

=head1 VERSION

version 0.002

=head1 SYNOPSIS

    use TheSchwartz::JobScheduler;
    my @databases = (
        { id => 'db_1', prefix => 'theschwartz_schema.', },
        { id => 'db_2', prefix => 'theschwartz_schema.', },
    );
    use Database::ManagedHandle;
    sub get_dbh {
        my ($db_id) = @_;
        my $mh1 = Database::ManagedHandle->instance;
        return $mh1->dbh( $db_id );
    }

    my $client = TheSchwartz::JobScheduler->new(
        databases => \@databases,
        dbh_callback => \&get_dbh,
        );
    my $job_id = $client->insert(
        job => TheSchwartz::JobScheduler::Job->new(
                    funcname => 'fetch',
                    arg      => {type => 'site', url => 'https://example.com/'},
                    ),
    );

    my $job1 = TheSchwartz::JobScheduler::Job->new;
    $job1->funcname("WorkerName");
    $job1->arg({ foo => "bar" });
    $job1->uniqkey("uniqkey");
    $job1->run_after( time + 60 );
    $client->insert( job => $job1 );
    my $job2 = TheSchwartz::JobScheduler::Job->new(
        funcname => 'WorkerName',
        arg => { foo => 'baz' },
        );
    $client->insert( job => $job2 );

    my @jobs = $client->list_jobs( search_params => { funcname => 'funcname' }, );
    for my $job (@jobs) {
        print $job->jobid;
    }

=head1 DESCRIPTION

TheSchwartz::JobScheduler is an interface to insert a new job into
TheSchwartz job queue (maintained by a database).

The rationale behind this module is using it in a long running web service,
for instance, in L<Dancer2>. Because the database connections cannot
be relied to stay open indefinitely, we get a new database handle
for each operation.

This module is solely created for the purpose of injecting a new job
from web servers without loading additional TheSchwartz and
Data::ObjectDriver modules onto your system. Your TheSchwartz job worker
processes will still need to be implemented using the full featured
L<TheSchwartz::Worker> module.

=head2 Configuration: Databases and Their Handles

L<TheSchwartz> can use several different databases simultaneously,
for instance, to share load and distribute jobs safely to only
those workers who could, in turn, demand restricted access.
This makes TheSchwartz very decentralized.

If your setup is reasonably simple, for instance, a webapp,
e.g. L<Dancer2>, and L<TheSchwartz>
as a worker system executing long running tasks which would
disrupt the webapp, then perhaps you only use one database.
In that case, you can consider using the same database handle
in both webapp and TheSchwartz. If you use database transactions
to ensure an atomized commit, you can involve TheSchwartz::JobScheduler
in the same transaction. If your transaction fails after worker task
is inserted, then also the worker task gets cancelled (rollbacked).

If, however, your TheSchwartz system is complex or otherwise separate
from the systems which create the tasks, or you simply use more than one
database in TheSchwartz, you cannot share your other database handles
with TheSchwartz::JobScheduler. Scheduler might need to access all databases
in sequence to place the task in the right one. Besides this,
TheSchwartz::JobScheduler is prepared for the possibility of one or more
databases being off-line. It loops through all the databases
until it gets a working database handle.

Database handles are provided by the calling program.
This allows the caller to use any available system to provide
the handles. If TheSchwartz::JobScheduler receives an C<undef>
instead of a database handle, it tries the next database.
If there is no working database handles, it croaks.

Database configuration does not need database addresses, dns:s
or usernames and passwords. Because TheSchwartz::JobScheduler
gets the database handle from outside, it only needs to know
a database id to separate between databases and a possible
prefix for each database. Prefix is prepended to every
database table and sequence name. If your database uses a different schema
than the default one for TheSchwartz tables, use C<prefix>
to solve this.

    my %dbs = (
        db_1 => [ 'dbi:SQLite:...', undef, undef, {} ],
        db_2 => [ 'dbi:SQLite:...', undef, undef, {} ],
    );
    sub get_dbh {
        my ($id) = @_;
        my @connection_info = @{ $dbs->{ $id } };
        return DBI->connect( @connection_info );
    };
    my %databases = (
        db_1 => { prefix => 'theschwartz_schema.', dbh_callback => \&get_dbh, },
        db_2 => { prefix => 'another_schema.', dbh_callback => \&get_dbh, },
    );
    use TheSchwartz::JobScheduler;
    my $scheduler = TheSchwartz::JobScheduler->new(
        databases => \%databases,
        dbh_callback => \&get_dbh,
        );

In the following example the calling
program is using L<Database::ManagedHandle>, a module
which makes certain that a database handle is always usable.

    # First create a Database::ManagedHandle config class
    # See Database::ManagedHandle for instructions
    # Then just use it:
    my %databases = (
        db_1 => {
            prefix => 'theschwartz_schema.',
        },
        db_2 => {
        {
            prefix => 'another_schema.',
        },
    );
    use TheSchwartz::JobScheduler;
    my $scheduler = TheSchwartz::JobScheduler->new(
        databases => \%databases,
        dbh_callback => 'Database::ManagedHandle->instance',
    );

=head2 DBH Callback

The item C<dbh_callback> can be either a CODE reference, i.e. a subroutine,
or a string which when executed with C<eval> will produce an object.
This object must have at least one method: C<dbh()>. This method,
when called, must return either a C<DBI::db> object (such as created by
C<< DBI->connect >>, or an C<undef>.

    use TheSchwartz::JobScheduler;
    my $scheduler = TheSchwartz::JobScheduler->new(
        databases => \%databases,
        dbh_callback => 'Database::ManagedHandle->instance',
    );

You can specify C<dbh_callback> either when creating the client object
or when calling C<insert()> or C<list_jobs()>. If you are using
TheSchwartz::JobScheduler as part of another system, for example,
a web service, you will probably want to share one opened database handle
because that will allow you to include TheSchwartz::JobScheduler into
a transaction.

    my %databases = (
        db_1 => { prefix => 'theschwartz_schema.', },
        db_2 => { prefix => 'another_schema.', },
    );
    sub get_dbh {
        my ($id) = @_;
        my @connection_info = @{ $databases{ $id } };
        return DBI->connect( @connection_info );
    };
    use TheSchwartz::JobScheduler;
    my $scheduler = TheSchwartz::JobScheduler->new(
        databases => \%databases,
    );
    my $job = TheSchwartz::JobScheduler::Job->new(
        funcname => 'my_func',
        );
    $scheduler->insert(
        job => $job,
        dbh_callback => $get_dbh,
        );

=head2 Uniqkey

The C<uniqkey> field is an arbitrary string identifier
used to prevent applications from posting duplicate jobs.
At most one with the same uniqkey value can be posted
to a single TheSchwartz database.

There are, however, valid situations when inserting the same
job and uniqkey would make sense. For instance, in a case
when several different actions one after another but independent of each other
would result in the same job being required to run.

Note, the job arguments do not enter into the uniqueness consideration,
only job name and unique key (C<funcid> and C<uniqkey> fields).

Depending on the database and whether uniqueness is protected with
database constraints, such as primary keys, trying to insert another
job with the same C<uniqkey> can cause an error, the previous row being
rewritten with new content and new arguments, or another row being created.

User can choose how to deal with this situation.
When instantiating C<TheSchwartz::JobScheduler>, user can define
the additional option C<handle_uniqkey> with any of the
following values:

=over 8

=item B<no_check>

This option does not do any checking on the condition. If the database
is configured to not allow an insert operation, it will throw
an exception. User must be prepared for this, for instance,
by enclosing the operation in C<eval>.

This is the default setting.

=item B<overwrite>

Update the fields C<arg>, C<insert_time>, C<run_after>, C<grabbed_until>,
C<priority> and C<coalesce>, and return the existing entry's C<jobid>.
This setting will create a slight overhead.

Not yet implemented.

=item B<acknowledge>

If there is already a matching entry (C<funcid> and C<uniqkey> fields),
no change will be made. The C<jobid> of the existing entry will be returned.
This setting will create a slight overhead.

=back

B<N.B. This option is used only when TheSchwartz::JobScheduler::Job has
set the field C<uniqkey>.
If you don't use uniqkey, this problem will never arise.>

B<N.B.2. Using either c<overwrite> or C<acknowledge> is the recommended
value. Only in situations which require extreme throughput, should you
consider other alternatives for this problem.>

    # Depending on the database table settings,
    # this will either throw an exception or
    # it will pass and result with invalid table data.
    my $scheduler = TheSchwartz::JobScheduler->new(
        databases => \%databases,
        dbh_callback => 'Database::ManagedHandle->instance',
        opts => {
            handle_uniqkey => 'no_check',
        },
    );
    my $job = TheSchwartz::JobScheduler::Job->new(
        funcname => 'Test::uniqkey',
        arg      => { an_item => 'value A' },
        uniqkey  => 'UNIQUE_STR_A',
        );
    $scheduler->insert( $job );
    $job = TheSchwartz::JobScheduler::Job->new(
        funcname => 'Test::uniqkey',
        arg      => { an_item => 'value B' },
        uniqkey  => 'UNIQUE_STR_A',
        );
    $scheduler->insert( $job );

=head2 Logging

TheSchwartz::JobScheduler uses the excellent L<Log::Any> to produce logging messages.

The easiest way to get the logging messages printed is to add the following line
in the preamble of your program:

    use Log::Any::Adapter ('Stdout', log_level => 'debug' );

Alternative, you can do this on the command line:

    perl '-MLog::Any::Adapter(Stdout, log_level=>trace)'

=head2 databases

The databases used by TheSchwartz.

Please see above L</"Configuration: Databases and Their Handles">.

=head2 dbh_callback

Callback for TheSchwartz::JobScheduler to get a database handle.

Please see above L</"Configuration: Databases and Their Handles">.

=head2 opts

Additional options for controlling other features, including uniqkey.

Please see above L<Uniqkey>.

Example:

    my $scheduler = TheSchwartz::JobScheduler->new(
        databases => \@databases,
        dbh_callback => 'Database::ManagedHandle->instance',
        opts => {
            handle_uniqkey => 'no_check',
        },
    );

=head2 insert

Return a list of active jobs collected from all accessible databases.
Create a job.

Parameters: job (TheSchwartz::JobScheduler::Job)

    my @jobs = $client->insert(
        job => TheSchwartz::JobScheduler::Job->new(
                    funcname => 'fetch',
                    arg      => {type => 'site', url => 'https://example.com/'},
                    ),
    );

=head2 funcname_to_id

Fetch function id from database. If not exists, then insert.

=head2 list_jobs

Return a list of active jobs collected from all accessible databases.

Parameters: A hash containing named parameters.

    my @jobs = $client->list_jobs(
        search_params => { funcname => 'fetch_webpage'},
    );

=begin Pod::Coverage




=end Pod::Coverage

=for stopwords TheSchwartz DBI Uniqkey uniqkey webapp

=head1 THANKS

This module is very much inspired by L<TheSchwartz::Simple>.

=head1 SEE ALSO

=over 8

=item L<TheSchwartz>

=item L<TheSchwartz::Simple>

=back

=head1 AUTHOR

Mikko Koivunalho <[e-mail]>

=head1 COPYRIGHT AND LICENSE

This software is copyright (c) 2023 by Mikko Koivunalho.

This is free software; you can redistribute it and/or modify it under
the same terms as the Perl 5 programming language system itself.

=cut