Koha/BackgroundJob.pm

   1 package Koha::BackgroundJob;
   2
   3 # This file is part of Koha.
   4 #
   5 # Koha is free software; you can redistribute it and/or modify it
   6 # under the terms of the GNU General Public License as published by
   7 # the Free Software Foundation; either version 3 of the License, or
   8 # (at your option) any later version.
   9 #
  10 # Koha is distributed in the hope that it will be useful, but
  11 # WITHOUT ANY WARRANTY; without even the implied warranty of
  12 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  13 # GNU General Public License for more details.
  14 #
  15 # You should have received a copy of the GNU General Public License
  16 # along with Koha; if not, see <http://www.gnu.org/licenses>.
  17
  18 use Modern::Perl;
  19 use Encode qw();
  20 use JSON;
  21 use Carp qw( croak );
  22 use Net::Stomp;
  23 use Try::Tiny qw( catch try );
  24
  25 use C4::Context;
  26 use Koha::DateUtils qw( dt_from_string );
  27 use Koha::Exceptions;
  28 use Koha::Exceptions::BackgroundJob;
  29
  30 use base qw( Koha::Object );
  31
  32 =head1 NAME
  33
  34 Koha::BackgroundJob - Koha BackgroundJob Object class
  35
  36 This is a base class for BackgroundJob, some methods must be subclassed.
  37
  38 Example of usage:
  39
  40 Producer:
  41 my $job_id = Koha::BackgroundJob->enqueue(
  42     {
  43         job_type => $job_type,
  44         job_size => $job_size,
  45         job_args => $job_args
  46     }
  47 );
  48
  49 Consumer:
  50 Koha::BackgroundJobs->find($job_id)->process;
  51 See also C<misc/background_jobs_worker.pl> for a full example
  52
  53 =head1 API
  54
  55 =head2 Class methods
  56
  57 =head3 connect
  58
  59 Connect to the message broker using default guest/guest credential
  60
  61 =cut
  62
  63 sub connect {
  64     my ( $self );
  65     my $hostname = 'localhost';
  66     my $port = '61613';
  67     my $config = C4::Context->config('message_broker');
  68     my $credentials = {
  69         login => 'guest',
  70         passcode => 'guest',
  71     };
  72     if ($config){
  73         $hostname = $config->{hostname} if $config->{hostname};
  74         $port = $config->{port} if $config->{port};
  75         $credentials->{login} = $config->{username} if $config->{username};
  76         $credentials->{passcode} = $config->{password} if $config->{password};
  77         $credentials->{host} = $config->{vhost} if $config->{vhost};
  78     }
  79     my $stomp = Net::Stomp->new( { hostname => $hostname, port => $port } );
  80     $stomp->connect( $credentials );
  81     return $stomp;
  82 }
  83
  84 =head3 enqueue
  85
  86 Enqueue a new job. It will insert a new row in the DB table and notify the broker that a new job has been enqueued.
  87
  88 C<job_size> is the size of the job
  89 C<job_args> is the arguments of the job. It's a structure that will be JSON encoded.
  90
  91 Return the job_id of the newly created job.
  92
  93 =cut
  94
  95 sub enqueue {
  96     my ( $self, $params ) = @_;
  97
  98     my $job_type    = $self->job_type;
  99     my $job_size    = $params->{job_size};
 100     my $job_args    = $params->{job_args};
 101     my $job_context = $params->{job_context} // C4::Context->userenv;
 102     my $job_queue   = $params->{job_queue}  // 'default';
 103     my $json = $self->json;
 104
 105     my $borrowernumber = (C4::Context->userenv) ? C4::Context->userenv->{number} : undef;
 106     $job_context->{interface} = C4::Context->interface;
 107     my $json_context = $json->encode($job_context);
 108     my $json_args = $json->encode($job_args);
 109
 110     $self->set(
 111         {
 112             status         => 'new',
 113             type           => $job_type,
 114             queue          => $job_queue,
 115             size           => $job_size,
 116             data           => $json_args,
 117             context        => $json_context,
 118             enqueued_on    => dt_from_string,
 119             borrowernumber => $borrowernumber,
 120         }
 121     )->store;
 122
 123     $job_args->{job_id} = $self->id;
 124
 125     my $conn;
 126     try {
 127         $conn = $self->connect;
 128     } catch {
 129         warn "Cannot connect to broker " . $_;
 130     };
 131     return $self->id unless $conn;
 132
 133     $json_args = $json->encode($job_args);
 134     try {
 135         # This namespace is wrong, it must be a vhost instead.
 136         # But to do so it needs to be created on the server => much more work when a new Koha instance is created.
 137         # Also, here we just want the Koha instance's name, but it's not in the config...
 138         # Picking a random id (memcached_namespace) from the config
 139         my $namespace = C4::Context->config('memcached_namespace');
 140         my $encoded_args = Encode::encode_utf8( $json_args ); # FIXME We should better leave this to Net::Stomp?
 141         my $destination = sprintf( "/queue/%s-%s", $namespace, $job_queue );
 142         $conn->send_with_receipt( { destination => $destination, body => $encoded_args, persistent => 'true' } )
 143           or Koha::Exceptions::Exception->throw('Job has not been enqueued');
 144     } catch {
 145         $self->status('failed')->store;
 146         if ( ref($_) eq 'Koha::Exceptions::Exception' ) {
 147             $_->rethrow;
 148         } else {
 149             warn sprintf "The job has not been sent to the message broker: (%s)", $_;
 150         }
 151     };
 152
 153     return $self->id;
 154 }
 155
 156 =head3 process
 157
 158 Process the job!
 159
 160 =cut
 161
 162 sub process {
 163     my ( $self, $args ) = @_;
 164
 165     return {} if ref($self) ne 'Koha::BackgroundJob';
 166
 167     my $derived_class = $self->_derived_class;
 168
 169     $args ||= {};
 170
 171     if ( $self->context ) {
 172         my $context = $self->json->decode($self->context);
 173         C4::Context->_new_userenv(-1);
 174         C4::Context->interface( $context->{interface} );
 175         C4::Context->set_userenv(
 176             $context->{number},       $context->{id},
 177             $context->{cardnumber},   $context->{firstname},
 178             $context->{surname},      $context->{branch},
 179             $context->{branchname},   $context->{flags},
 180             $context->{emailaddress}, undef,
 181             $context->{desk_id},      $context->{desk_name},
 182             $context->{register_id},  $context->{register_name}
 183         );
 184     }
 185     else {
 186         Koha::Logger->get->warn("A background job didn't have context defined (" . $self->id . ")");
 187     }
 188
 189     return $derived_class->process( $args );
 190 }
 191
 192 =head3 start
 193
 194     $self->start;
 195
 196 Marks the job as started.
 197
 198 =cut
 199
 200 sub start {
 201     my ($self) = @_;
 202
 203     Koha::Exceptions::BackgroundJob::InconsistentStatus->throw(
 204         current_status  => $self->status,
 205         expected_status => 'new'
 206     ) unless $self->status eq 'new';
 207
 208     return $self->set(
 209         {
 210             started_on => \'NOW()',
 211             progress   => 0,
 212             status     => 'started',
 213         }
 214     )->store;
 215 }
 216
 217 =head3 step
 218
 219     $self->step;
 220
 221 Makes the job record a step has taken place.
 222
 223 =cut
 224
 225 sub step {
 226     my ($self) = @_;
 227
 228     Koha::Exceptions::BackgroundJob::InconsistentStatus->throw(
 229         current_status  => $self->status,
 230         expected_status => 'started'
 231     ) unless $self->status eq 'started';
 232
 233     # reached the end of the tasks already
 234     Koha::Exceptions::BackgroundJob::StepOutOfBounds->throw()
 235         unless $self->progress < $self->size;
 236
 237     return $self->progress( $self->progress + 1 )->store;
 238 }
 239
 240 =head3 finish
 241
 242     $self->finish;
 243
 244 Makes the job record as finished. If the job status is I<cancelled>, it is kept.
 245
 246 =cut
 247
 248 sub finish {
 249     my ( $self, $data ) = @_;
 250
 251     $self->status('finished') unless $self->status eq 'cancelled' or $self->status eq 'failed';
 252
 253     return $self->set(
 254         {
 255             ended_on => \'NOW()',
 256             data     => $self->json->encode($data),
 257         }
 258     )->store;
 259 }
 260
 261 =head3 json
 262
 263    my $JSON_object = $self->json;
 264
 265 Returns a JSON object with utf8 disabled. Encoding to UTF-8 should be
 266 done later.
 267
 268 =cut
 269
 270 sub json {
 271     my ( $self ) = @_;
 272     $self->{_json} //= JSON->new->utf8(0); # TODO Should we allow_nonref ?
 273     return $self->{_json};
 274 }
 275
 276 =head3 decoded_data
 277
 278     my $job_data = $self->decoded_data;
 279
 280 Returns the decoded JSON contents from $self->data.
 281
 282 =cut
 283
 284 sub decoded_data {
 285     my ($self) = @_;
 286
 287     return $self->data ? $self->json->decode( $self->data ) : undef;
 288 }
 289
 290 =head3 set_encoded_data
 291
 292     $self->set_encoded_data( $data );
 293
 294 Serializes I<$data> as a JSON string and sets the I<data> attribute with it.
 295
 296 =cut
 297
 298 sub set_encoded_data {
 299     my ( $self, $data ) = @_;
 300
 301     return $self->data( $data ? $self->json->encode($data) : undef );
 302 }
 303
 304 =head3 job_type
 305
 306 Return the job type of the job. Must be a string.
 307
 308 =cut
 309
 310 sub job_type { croak "This method must be subclassed" }
 311
 312 =head3 messages
 313
 314 Messages let during the processing of the job.
 315
 316 =cut
 317
 318 sub messages {
 319     my ( $self ) = @_;
 320
 321     my @messages;
 322     my $data_dump = $self->json->decode($self->data);
 323     if ( exists $data_dump->{messages} ) {
 324         @messages = @{ $data_dump->{messages} };
 325     }
 326
 327     return \@messages;
 328 }
 329
 330 =head3 report
 331
 332 Report of the job.
 333
 334 =cut
 335
 336 sub report {
 337     my ( $self ) = @_;
 338
 339     my $data_dump = $self->json->decode($self->data);
 340     return $data_dump->{report} || {};
 341 }
 342
 343 =head3 additional_report
 344
 345 Build additional variables for the job detail view.
 346
 347 =cut
 348
 349 sub additional_report {
 350     my ( $self ) = @_;
 351
 352     return {} if ref($self) ne 'Koha::BackgroundJob';
 353
 354     my $derived_class = $self->_derived_class;
 355
 356     return $derived_class->additional_report;
 357 }
 358
 359 =head3 cancel
 360
 361 Cancel a job.
 362
 363 =cut
 364
 365 sub cancel {
 366     my ( $self ) = @_;
 367     $self->status('cancelled')->store;
 368 }
 369
 370 =head2 Internal methods
 371
 372 =head3 _derived_class
 373
 374 =cut
 375
 376 sub _derived_class {
 377     my ( $self ) = @_;
 378     my $job_type = $self->type;
 379
 380     my $class = $self->type_to_class_mapping->{$job_type};
 381
 382     Koha::Exception->throw($job_type . ' is not a valid job_type')
 383         unless $class;
 384
 385     eval "require $class";
 386     return $class->_new_from_dbic( $self->_result );
 387 }
 388
 389 =head3 type_to_class_mapping
 390
 391     my $mapping = Koha::BackgroundJob->new->type_to_class_mapping;
 392
 393 Returns the available types to class mappings.
 394
 395 =cut
 396
 397 sub type_to_class_mapping {
 398     my ($self) = @_;
 399
 400     my $plugins_mapping = ( C4::Context->config("enable_plugins") ) ? $self->plugin_types_to_classes : {};
 401
 402     return ($plugins_mapping)
 403       ? { %{ $self->core_types_to_classes }, %$plugins_mapping }
 404       : $self->core_types_to_classes;
 405 }
 406
 407 =head3 core_types_to_classes
 408
 409     my $mappings = Koha::BackgroundJob->new->core_types_to_classes
 410
 411 Returns the core background jobs types to class mappings.
 412
 413 =cut
 414
 415 sub core_types_to_classes {
 416     return {
 417         batch_authority_record_deletion     => 'Koha::BackgroundJob::BatchDeleteAuthority',
 418         batch_authority_record_modification => 'Koha::BackgroundJob::BatchUpdateAuthority',
 419         batch_biblio_record_deletion        => 'Koha::BackgroundJob::BatchDeleteBiblio',
 420         batch_biblio_record_modification    => 'Koha::BackgroundJob::BatchUpdateBiblio',
 421         batch_item_record_deletion          => 'Koha::BackgroundJob::BatchDeleteItem',
 422         batch_item_record_modification      => 'Koha::BackgroundJob::BatchUpdateItem',
 423         batch_hold_cancel                   => 'Koha::BackgroundJob::BatchCancelHold',
 424         create_eholdings_from_biblios       => 'Koha::BackgroundJob::CreateEHoldingsFromBiblios',
 425         update_elastic_index                => 'Koha::BackgroundJob::UpdateElasticIndex',
 426         update_holds_queue_for_biblios      => 'Koha::BackgroundJob::BatchUpdateBiblioHoldsQueue',
 427         stage_marc_for_import               => 'Koha::BackgroundJob::StageMARCForImport',
 428         marc_import_commit_batch            => 'Koha::BackgroundJob::MARCImportCommitBatch',
 429         marc_import_revert_batch            => 'Koha::BackgroundJob::MARCImportRevertBatch',
 430     };
 431 }
 432
 433 =head3 plugin_types_to_classes
 434
 435     my $mappings = Koha::BackgroundJob->new->plugin_types_to_classes
 436
 437 Returns the plugin-defined background jobs types to class mappings.
 438
 439 =cut
 440
 441 sub plugin_types_to_classes {
 442     my ($self) = @_;
 443
 444     unless ( exists $self->{_plugin_mapping} ) {
 445         require Koha::Plugins;
 446         my @plugins = Koha::Plugins->new()->GetPlugins( { method => 'background_tasks', } );
 447
 448         foreach my $plugin (@plugins) {
 449
 450             my $tasks    = $plugin->background_tasks;
 451             my $metadata = $plugin->get_metadata;
 452
 453             unless ( $metadata->{namespace} ) {
 454                 Koha::Logger->get->warn(
 455                         q{A plugin includes the 'background_tasks' method, }
 456                       . q{but doesn't provide the required 'namespace' }
 457                       . qq{method ($plugin->{class})} );
 458                 next;
 459             }
 460
 461             my $namespace = $metadata->{namespace};
 462
 463             foreach my $type ( keys %{$tasks} ) {
 464                 my $class = $tasks->{$type};
 465
 466                 # skip if conditions not met
 467                 next unless $type and $class;
 468
 469                 my $key = "plugin_$namespace" . "_$type";
 470
 471                 $self->{_plugin_mapping}->{$key} = $tasks->{$type};
 472             }
 473         }
 474     }
 475
 476     return $self->{_plugin_mapping};
 477 }
 478
 479 =head3 to_api
 480
 481     my $json = $job->to_api;
 482
 483 Overloaded method that returns a JSON representation of the Koha::BackgroundJob object,
 484 suitable for API output.
 485
 486 =cut
 487
 488 sub to_api {
 489     my ( $self, $params ) = @_;
 490
 491     my $json = $self->SUPER::to_api( $params );
 492
 493     $json->{context} = $self->json->decode($self->context)
 494       if defined $self->context;
 495     $json->{data} = $self->decoded_data;
 496
 497     return $json;
 498 }
 499
 500 =head3 to_api_mapping
 501
 502 This method returns the mapping for representing a Koha::BackgroundJob object
 503 on the API.
 504
 505 =cut
 506
 507 sub to_api_mapping {
 508     return {
 509         id             => 'job_id',
 510         borrowernumber => 'patron_id',
 511         ended_on       => 'ended_date',
 512         enqueued_on    => 'enqueued_date',
 513         started_on     => 'started_date',
 514     };
 515 }
 516
 517 =head3 _type
 518
 519 =cut
 520
 521 sub _type {
 522     return 'BackgroundJob';
 523 }
 524
 525 1;