CRM/Mailing/BAO/MailingJob.php

   1 <?php
   2 /*
   3  +--------------------------------------------------------------------+
   4  | Copyright CiviCRM LLC. All rights reserved.                        |
   5  |                                                                    |
   6  | This work is published under the GNU AGPLv3 license with some      |
   7  | permitted exceptions and without any warranty. For full license    |
   8  | and copyright information, see https://civicrm.org/licensing       |
   9  +--------------------------------------------------------------------+
  10  */
  11 use Civi\Api4\ActivityContact;
  12
  13 /**
  14  *
  15  * @package CRM
  16  * @copyright CiviCRM LLC https://civicrm.org/licensing
  17  */
  18
  19 require_once 'Mail.php';
  20
  21 /**
  22  * Class CRM_Mailing_BAO_MailingJob
  23  */
  24 class CRM_Mailing_BAO_MailingJob extends CRM_Mailing_DAO_MailingJob {
  25   const MAX_CONTACTS_TO_PROCESS = 1000;
  26
  27   /**
  28    * (Dear God Why) Keep a global count of mails processed within the current
  29    * request.
  30    *
  31    * @var int
  32    */
  33   public static $mailsProcessed = 0;
  34
  35   /**
  36    * Create mailing job.
  37    *
  38    * @param array $params
  39    *
  40    * @return \CRM_Mailing_BAO_MailingJob
  41    * @throws \CRM_Core_Exception
  42    */
  43   public static function create($params) {
  44     if (empty($params['id']) && empty($params['mailing_id'])) {
  45       throw new CRM_Core_Exception("Failed to create job: Unknown mailing ID");
  46     }
  47     $op = empty($params['id']) ? 'create' : 'edit';
  48     CRM_Utils_Hook::pre($op, 'MailingJob', CRM_Utils_Array::value('id', $params), $params);
  49
  50     $jobDAO = new CRM_Mailing_BAO_MailingJob();
  51     $jobDAO->copyValues($params);
  52     $jobDAO->save();
  53     if (!empty($params['mailing_id']) && empty('is_calling_function_updated_to_reflect_deprecation')) {
  54       CRM_Mailing_BAO_Mailing::getRecipients($params['mailing_id']);
  55     }
  56     CRM_Utils_Hook::post($op, 'MailingJob', $jobDAO->id, $jobDAO);
  57     return $jobDAO;
  58   }
  59
  60   /**
  61    * Initiate all pending/ready jobs.
  62    *
  63    * @param array $testParams
  64    * @param string|null $mode
  65    *   Either 'sms' or null
  66    *
  67    * @return bool|null
  68    */
  69   public static function runJobs($testParams = NULL, $mode = NULL) {
  70     $job = new CRM_Mailing_BAO_MailingJob();
  71
  72     $jobTable = CRM_Mailing_DAO_MailingJob::getTableName();
  73     $mailingTable = CRM_Mailing_DAO_Mailing::getTableName();
  74     $mailerBatchLimit = Civi::settings()->get('mailerBatchLimit');
  75
  76     if (!empty($testParams)) {
  77       $query = "
  78       SELECT *
  79         FROM $jobTable
  80        WHERE id = {$testParams['job_id']}";
  81       $job->query($query);
  82     }
  83     else {
  84       $currentTime = date('YmdHis');
  85       $mailingACL = CRM_Mailing_BAO_Mailing::mailingACL('m');
  86       $domainID = CRM_Core_Config::domainID();
  87
  88       $modeClause = 'AND m.sms_provider_id IS NULL';
  89       if ($mode == 'sms') {
  90         $modeClause = 'AND m.sms_provider_id IS NOT NULL';
  91       }
  92
  93       // Select the first child job that is scheduled
  94       // CRM-6835
  95       $query = "
  96       SELECT   j.*
  97         FROM   $jobTable     j,
  98            $mailingTable m
  99        WHERE   m.id = j.mailing_id AND m.domain_id = {$domainID}
 100                      {$modeClause}
 101          AND   j.is_test = 0
 102          AND   ( ( j.start_date IS null
 103          AND       j.scheduled_date <= $currentTime
 104          AND       j.status = 'Scheduled' )
 105                 OR     ( j.status = 'Running'
 106          AND       j.end_date IS null ) )
 107          AND (j.job_type = 'child')
 108          AND   {$mailingACL}
 109       ORDER BY j.scheduled_date ASC,
 110         j.id
 111       ";
 112
 113       $job->query($query);
 114     }
 115
 116     while ($job->fetch()) {
 117       // still use job level lock for each child job
 118       $lock = Civi::lockManager()->acquire("data.mailing.job.{$job->id}");
 119       if (!$lock->isAcquired()) {
 120         continue;
 121       }
 122
 123       // for test jobs we do not change anything, since its on a short-circuit path
 124       if (empty($testParams)) {
 125         // we've got the lock, but while we were waiting and processing
 126         // other emails, this job might have changed under us
 127         // lets get the job status again and check
 128         $job->status = CRM_Core_DAO::getFieldValue(
 129           'CRM_Mailing_DAO_MailingJob',
 130           $job->id,
 131           'status',
 132           'id',
 133           TRUE
 134         );
 135
 136         if (
 137           $job->status != 'Running' &&
 138           $job->status != 'Scheduled'
 139         ) {
 140           // this includes Cancelled and other statuses, CRM-4246
 141           $lock->release();
 142           continue;
 143         }
 144       }
 145
 146       /* Queue up recipients for the child job being launched */
 147
 148       if ($job->status != 'Running') {
 149         $transaction = new CRM_Core_Transaction();
 150
 151         // have to queue it up based on the offset and limits
 152         // get the parent ID, and limit and offset
 153         $job->queue($testParams);
 154
 155         // Update to show job has started.
 156         self::create([
 157           'id' => $job->id,
 158           'start_date' => date('YmdHis'),
 159           'status' => 'Running',
 160         ]);
 161
 162         $transaction->commit();
 163       }
 164
 165       // Get the mailer
 166       if ($mode === NULL) {
 167         $mailer = \Civi::service('pear_mail');
 168       }
 169       elseif ($mode == 'sms') {
 170         $mailer = CRM_SMS_Provider::singleton(['mailing_id' => $job->mailing_id]);
 171       }
 172
 173       // Compose and deliver each child job
 174       if (\CRM_Utils_Constant::value('CIVICRM_FLEXMAILER_HACK_DELIVER')) {
 175         $isComplete = Civi\Core\Resolver::singleton()->call(CIVICRM_FLEXMAILER_HACK_DELIVER, [$job, $mailer, $testParams]);
 176       }
 177       else {
 178         $isComplete = $job->deliver($mailer, $testParams);
 179       }
 180
 181       CRM_Utils_Hook::post('create', 'CRM_Mailing_DAO_Spool', $job->id, $isComplete);
 182
 183       // Mark the child complete
 184       if ($isComplete) {
 185         // Finish the job.
 186
 187         $transaction = new CRM_Core_Transaction();
 188         self::create(['id' => $job->id, 'end_date' => date('YmdHis'), 'status' => 'Complete']);
 189         $transaction->commit();
 190
 191         // don't mark the mailing as complete
 192       }
 193
 194       // Release the child joblock
 195       $lock->release();
 196
 197       if ($testParams) {
 198         return $isComplete;
 199       }
 200
 201       // CRM-17629: Stop processing jobs if mailer batch limit reached
 202       if ($mailerBatchLimit > 0 && self::$mailsProcessed >= $mailerBatchLimit) {
 203         break;
 204       }
 205
 206     }
 207   }
 208
 209   /**
 210    * Post process to determine if the parent job
 211    * as well as the mailing is complete after the run.
 212    * @param string|null $mode
 213    *   Either 'sms' or null
 214    */
 215   public static function runJobs_post($mode = NULL) {
 216
 217     $job = new CRM_Mailing_BAO_MailingJob();
 218
 219     $mailing = new CRM_Mailing_BAO_Mailing();
 220
 221     $config = CRM_Core_Config::singleton();
 222     $jobTable = CRM_Mailing_DAO_MailingJob::getTableName();
 223     $mailingTable = CRM_Mailing_DAO_Mailing::getTableName();
 224
 225     $currentTime = date('YmdHis');
 226     $mailingACL = CRM_Mailing_BAO_Mailing::mailingACL('m');
 227     $domainID = CRM_Core_Config::domainID();
 228
 229     $query = "
 230                 SELECT   j.*
 231                   FROM   $jobTable     j,
 232                                  $mailingTable m
 233                  WHERE   m.id = j.mailing_id AND m.domain_id = {$domainID}
 234                    AND   j.is_test = 0
 235                    AND       j.scheduled_date <= $currentTime
 236                    AND       j.status = 'Running'
 237                    AND       j.end_date IS null
 238                    AND       (j.job_type != 'child' OR j.job_type is NULL)
 239                 ORDER BY j.scheduled_date,
 240                                  j.start_date";
 241
 242     $job->query($query);
 243
 244     // For each parent job that is running, let's look at their child jobs
 245     while ($job->fetch()) {
 246
 247       $child_job = new CRM_Mailing_BAO_MailingJob();
 248
 249       $child_job_sql = "
 250             SELECT count(j.id)
 251                         FROM civicrm_mailing_job j, civicrm_mailing m
 252                         WHERE m.id = j.mailing_id
 253                         AND j.job_type = 'child'
 254                         AND j.parent_id = %1
 255             AND j.status <> 'Complete'";
 256       $params = [1 => [$job->id, 'Integer']];
 257
 258       $anyChildLeft = CRM_Core_DAO::singleValueQuery($child_job_sql, $params);
 259
 260       // all of the child jobs are complete, update
 261       // the parent job as well as the mailing status
 262       if (!$anyChildLeft) {
 263
 264         $transaction = new CRM_Core_Transaction();
 265
 266         $saveJob = new CRM_Mailing_DAO_MailingJob();
 267         $saveJob->id = $job->id;
 268         $saveJob->end_date = date('YmdHis');
 269         $saveJob->status = 'Complete';
 270         $saveJob->save();
 271
 272         $mailing->reset();
 273         $mailing->id = $job->mailing_id;
 274         $mailing->is_completed = TRUE;
 275         $mailing->save();
 276         $transaction->commit();
 277
 278         // CRM-17763
 279         CRM_Utils_Hook::postMailing($job->mailing_id);
 280       }
 281     }
 282   }
 283
 284   /**
 285    * before we run jobs, we need to split the jobs
 286    * @param int $offset
 287    * @param string|null $mode
 288    *   Either 'sms' or null
 289    */
 290   public static function runJobs_pre($offset = 200, $mode = NULL) {
 291     $job = new CRM_Mailing_BAO_MailingJob();
 292
 293     $jobTable = CRM_Mailing_DAO_MailingJob::getTableName();
 294     $mailingTable = CRM_Mailing_DAO_Mailing::getTableName();
 295
 296     $currentTime = date('YmdHis');
 297     $mailingACL = CRM_Mailing_BAO_Mailing::mailingACL('m');
 298
 299     $workflowClause = CRM_Mailing_BAO_MailingJob::workflowClause();
 300
 301     $domainID = CRM_Core_Config::domainID();
 302
 303     $modeClause = 'AND m.sms_provider_id IS NULL';
 304     if ($mode == 'sms') {
 305       $modeClause = 'AND m.sms_provider_id IS NOT NULL';
 306     }
 307
 308     // Select all the mailing jobs that are created from
 309     // when the mailing is submitted or scheduled.
 310     $query = "
 311     SELECT   j.*
 312       FROM   $jobTable     j,
 313          $mailingTable m
 314      WHERE   m.id = j.mailing_id AND m.domain_id = {$domainID}
 315                  $workflowClause
 316                  $modeClause
 317        AND   j.is_test = 0
 318        AND   ( ( j.start_date IS null
 319        AND       j.scheduled_date <= $currentTime
 320        AND       j.status = 'Scheduled'
 321        AND       j.end_date IS null ) )
 322        AND ((j.job_type is NULL) OR (j.job_type <> 'child'))
 323     ORDER BY j.scheduled_date,
 324          j.start_date";
 325
 326     $job->query($query);
 327
 328     // For each of the "Parent Jobs" we find, we split them into
 329     // X Number of child jobs
 330     while ($job->fetch()) {
 331       // still use job level lock for each child job
 332       $lock = Civi::lockManager()->acquire("data.mailing.job.{$job->id}");
 333       if (!$lock->isAcquired()) {
 334         continue;
 335       }
 336
 337       // Re-fetch the job status in case things
 338       // changed between the first query and now
 339       // to avoid race conditions
 340       $job->status = CRM_Core_DAO::getFieldValue(
 341         'CRM_Mailing_DAO_MailingJob',
 342         $job->id,
 343         'status',
 344         'id',
 345         TRUE
 346       );
 347       if ($job->status != 'Scheduled') {
 348         $lock->release();
 349         continue;
 350       }
 351
 352       $transaction = new CRM_Core_Transaction();
 353
 354       $job->split_job($offset);
 355
 356       // Update the status of the parent job
 357       self::create(['id' => $job->id, 'start_date' => date('YmdHis'), 'status' => 'Running']);
 358       $transaction->commit();
 359
 360       // Release the job lock
 361       $lock->release();
 362     }
 363   }
 364
 365   /**
 366    * Split the parent job into n number of child job based on an offset.
 367    * If null or 0 , we create only one child job
 368    * @param int $offset
 369    */
 370   public function split_job($offset = 200) {
 371     $recipient_count = CRM_Mailing_BAO_Recipients::mailingSize($this->mailing_id);
 372
 373     $jobTable = CRM_Mailing_DAO_MailingJob::getTableName();
 374
 375     $dao = new CRM_Core_DAO();
 376
 377     $sql = "
 378 INSERT INTO civicrm_mailing_job
 379 (`mailing_id`, `scheduled_date`, `status`, `job_type`, `parent_id`, `job_offset`, `job_limit`)
 380 VALUES (%1, %2, %3, %4, %5, %6, %7)
 381 ";
 382     $params = [
 383       1 => [$this->mailing_id, 'Integer'],
 384       2 => [$this->scheduled_date, 'String'],
 385       3 => ['Scheduled', 'String'],
 386       4 => ['child', 'String'],
 387       5 => [$this->id, 'Integer'],
 388       6 => [0, 'Integer'],
 389       7 => [$recipient_count, 'Integer'],
 390     ];
 391
 392     // create one child job if the mailing size is less than the offset
 393     // probably use a CRM_Mailing_DAO_MailingJob( );
 394     if (empty($offset) ||
 395       $recipient_count <= $offset
 396     ) {
 397       CRM_Core_DAO::executeQuery($sql, $params);
 398     }
 399     else {
 400       // Creating 'child jobs'
 401       $scheduled_unixtime = strtotime($this->scheduled_date);
 402       for ($i = 0, $s = 0; $i < $recipient_count; $i = $i + $offset, $s++) {
 403         $params[2][0] = date('Y-m-d H:i:s', $scheduled_unixtime + $s);
 404         $params[6][0] = $i;
 405         $params[7][0] = $offset;
 406         CRM_Core_DAO::executeQuery($sql, $params);
 407       }
 408     }
 409
 410   }
 411
 412   /**
 413    * @param array $testParams
 414    */
 415   public function queue($testParams = NULL) {
 416     $mailing = new CRM_Mailing_BAO_Mailing();
 417     $mailing->id = $this->mailing_id;
 418     if (!empty($testParams)) {
 419       $mailing->getTestRecipients($testParams);
 420     }
 421     else {
 422       // We are still getting all the recipients from the parent job
 423       // so we don't mess with the include/exclude logic.
 424       $recipients = CRM_Mailing_BAO_Recipients::mailingQuery($this->mailing_id, $this->job_offset, $this->job_limit);
 425
 426       // FIXME: this is not very smart, we should move this to one DB call
 427       // INSERT INTO ... SELECT FROM ..
 428       // the thing we need to figure out is how to generate the hash automatically
 429       $now = time();
 430       $params = [];
 431       $count = 0;
 432       // dev/core#1768 Get the mail sync interval.
 433       $mail_sync_interval = Civi::settings()->get('civimail_sync_interval');
 434       while ($recipients->fetch()) {
 435         // CRM-18543: there are situations when both the email and phone are null.
 436         // Skip the recipient in this case.
 437         if (empty($recipients->email_id) && empty($recipients->phone_id)) {
 438           continue;
 439         }
 440
 441         if ($recipients->phone_id) {
 442           $recipients->email_id = "null";
 443         }
 444         else {
 445           $recipients->phone_id = "null";
 446         }
 447
 448         $params[] = [
 449           $this->id,
 450           $recipients->email_id,
 451           $recipients->contact_id,
 452           $recipients->phone_id,
 453         ];
 454         $count++;
 455         // dev/core#1768 Mail sync interval is now configurable.
 456         if ($count % $mail_sync_interval == 0) {
 457           CRM_Mailing_Event_BAO_Queue::bulkCreate($params, $now);
 458           $count = 0;
 459           $params = [];
 460         }
 461       }
 462
 463       if (!empty($params)) {
 464         CRM_Mailing_Event_BAO_Queue::bulkCreate($params, $now);
 465       }
 466     }
 467   }
 468
 469   /**
 470    * Send the mailing.
 471    *
 472    * @deprecated
 473    *   This is used by CiviMail but will be made redundant by FlexMailer.
 474    * @param object $mailer
 475    *   A Mail object to send the messages.
 476    *
 477    * @param array $testParams
 478    * @return bool
 479    */
 480   public function deliver(&$mailer, $testParams = NULL) {
 481     if (\Civi::settings()->get('experimentalFlexMailerEngine')) {
 482       throw new \RuntimeException("Cannot use legacy deliver() when experimentalFlexMailerEngine is enabled");
 483     }
 484
 485     $mailing = new CRM_Mailing_BAO_Mailing();
 486     $mailing->id = $this->mailing_id;
 487     $mailing->find(TRUE);
 488
 489     $config = NULL;
 490
 491     if ($config == NULL) {
 492       $config = CRM_Core_Config::singleton();
 493     }
 494
 495     if (property_exists($mailing, 'language') && $mailing->language && $mailing->language != CRM_Core_I18n::getLocale()) {
 496       $swapLang = CRM_Utils_AutoClean::swap('global://dbLocale?getter', 'call://i18n/setLocale', $mailing->language);
 497     }
 498
 499     $job_date = $this->scheduled_date;
 500     $fields = [];
 501
 502     if (!empty($testParams)) {
 503       $mailing->subject = ts('[CiviMail Draft]') . ' ' . $mailing->subject;
 504     }
 505
 506     CRM_Mailing_BAO_Mailing::tokenReplace($mailing);
 507
 508     // get and format attachments
 509     $attachments = CRM_Core_BAO_File::getEntityFile('civicrm_mailing', $mailing->id);
 510
 511     if (defined('CIVICRM_MAIL_SMARTY') && CIVICRM_MAIL_SMARTY) {
 512       CRM_Core_Smarty::registerStringResource();
 513     }
 514
 515     // CRM-12376
 516     // This handles the edge case scenario where all the mails
 517     // have been delivered in prior jobs.
 518     $isDelivered = TRUE;
 519
 520     // make sure that there's no more than $mailerBatchLimit mails processed in a run
 521     $mailerBatchLimit = Civi::settings()->get('mailerBatchLimit');
 522     $eq = self::findPendingTasks($this->id, $mailing->sms_provider_id ? 'sms' : 'email');
 523     while ($eq->fetch()) {
 524       if ($mailerBatchLimit > 0 && self::$mailsProcessed >= $mailerBatchLimit) {
 525         if (!empty($fields)) {
 526           $this->deliverGroup($fields, $mailing, $mailer, $job_date, $attachments);
 527         }
 528         return FALSE;
 529       }
 530       self::$mailsProcessed++;
 531
 532       $fields[] = [
 533         'id' => $eq->id,
 534         'hash' => $eq->hash,
 535         'contact_id' => $eq->contact_id,
 536         'email' => $eq->email,
 537         'phone' => $eq->phone,
 538       ];
 539       if (count($fields) == self::MAX_CONTACTS_TO_PROCESS) {
 540         $isDelivered = $this->deliverGroup($fields, $mailing, $mailer, $job_date, $attachments);
 541         if (!$isDelivered) {
 542           return $isDelivered;
 543         }
 544         $fields = [];
 545       }
 546     }
 547
 548     if (!empty($fields)) {
 549       $isDelivered = $this->deliverGroup($fields, $mailing, $mailer, $job_date, $attachments);
 550     }
 551     return $isDelivered;
 552   }
 553
 554   /**
 555    * @deprecated
 556    *   This is used by CiviMail but will be made redundant by FlexMailer.
 557    * @param array $fields
 558    *   List of intended recipients.
 559    *   Each recipient is an array with keys 'hash', 'contact_id', 'email', etc.
 560    * @param $mailing
 561    * @param $mailer
 562    * @param $job_date
 563    * @param $attachments
 564    *
 565    * @return bool|null
 566    * @throws Exception
 567    */
 568   public function deliverGroup(&$fields, &$mailing, &$mailer, &$job_date, &$attachments) {
 569     static $smtpConnectionErrors = 0;
 570
 571     if (!is_object($mailer) || empty($fields)) {
 572       throw new CRM_Core_Exception('Either mailer is not an object or we don\'t have recipients to send to in this group');
 573     }
 574
 575     // get the return properties
 576     $returnProperties = $mailing->getReturnProperties();
 577     $params = $targetParams = $deliveredParams = [];
 578     $count = 0;
 579     // dev/core#1768 Get the mail sync interval.
 580     $mail_sync_interval = Civi::settings()->get('civimail_sync_interval');
 581     $retryGroup = FALSE;
 582
 583     // CRM-15702: Sending bulk sms to contacts without e-mail address fails.
 584     // Solution is to skip checking for on hold
 585     //do include a statement to check wether e-mail address is on hold
 586     $skipOnHold = TRUE;
 587     if ($mailing->sms_provider_id) {
 588       //do not include a statement to check wether e-mail address is on hold
 589       $skipOnHold = FALSE;
 590     }
 591
 592     foreach ($fields as $key => $field) {
 593       $params[] = $field['contact_id'];
 594     }
 595
 596     [$details] = CRM_Utils_Token::getTokenDetails(
 597       $params,
 598       $returnProperties,
 599       $skipOnHold, TRUE, NULL,
 600       $mailing->getFlattenedTokens(),
 601       get_class($this),
 602       $this->id
 603     );
 604
 605     foreach ($fields as $key => $field) {
 606       $contactID = $field['contact_id'];
 607       if (!array_key_exists($contactID, $details)) {
 608         $details[$contactID] = [];
 609       }
 610
 611       // Compose the mailing.
 612       $recipient = $replyToEmail = NULL;
 613       $replyValue = strcmp($mailing->replyto_email, $mailing->from_email);
 614       if ($replyValue) {
 615         $replyToEmail = $mailing->replyto_email;
 616       }
 617
 618       $message = $mailing->compose(
 619         $this->id, $field['id'], $field['hash'],
 620         $field['contact_id'], $field['email'],
 621         $recipient, FALSE, $details[$contactID], $attachments,
 622         FALSE, NULL, $replyToEmail
 623       );
 624       if (empty($message)) {
 625         // lets keep the message in the queue
 626         // most likely a permissions related issue with smarty templates
 627         // or a bad contact id? CRM-9833
 628         continue;
 629       }
 630
 631       // Send the mailing.
 632
 633       $body = $message->get();
 634       $headers = $message->headers();
 635
 636       if ($mailing->sms_provider_id) {
 637         $provider = CRM_SMS_Provider::singleton(['mailing_id' => $mailing->id]);
 638         $body = $provider->getMessage($message, $field['contact_id'], $details[$contactID]);
 639         $headers = $provider->getRecipientDetails($field, $details[$contactID]);
 640       }
 641
 642       // make $recipient actually be the *encoded* header, so as not to baffle Mail_RFC822, CRM-5743
 643       $recipient = $headers['To'];
 644       $result = NULL;
 645
 646       // disable error reporting on real mailings (but leave error reporting for tests), CRM-5744
 647       if ($job_date) {
 648         $errorScope = CRM_Core_TemporaryErrorScope::ignoreException();
 649       }
 650
 651       $result = $mailer->send($recipient, $headers, $body, $this->id);
 652
 653       if ($job_date) {
 654         unset($errorScope);
 655       }
 656
 657       if (is_a($result, 'PEAR_Error') && !$mailing->sms_provider_id) {
 658         // CRM-9191
 659         $message = $result->getMessage();
 660         if ($this->isTemporaryError($message)) {
 661           // lets log this message and code
 662           $code = $result->getCode();
 663           CRM_Core_Error::debug_log_message("SMTP Socket Error or failed to set sender error. Message: $message, Code: $code");
 664
 665           // these are socket write errors which most likely means smtp connection errors
 666           // lets skip them and reconnect.
 667           $smtpConnectionErrors++;
 668           if ($smtpConnectionErrors <= 5) {
 669             $mailer->disconnect();
 670             $retryGroup = TRUE;
 671             continue;
 672           }
 673
 674           // seems like we have too many of them in a row, we should
 675           // write stuff to disk and abort the cron job
 676           $this->writeToDB(
 677             $deliveredParams,
 678             $targetParams,
 679             $mailing,
 680             $job_date
 681           );
 682
 683           CRM_Core_Error::debug_log_message("Too many SMTP Socket Errors. Exiting");
 684           CRM_Utils_System::civiExit();
 685         }
 686
 687         // Register the bounce event.
 688
 689         $params = [
 690           'event_queue_id' => $field['id'],
 691           'job_id' => $this->id,
 692           'hash' => $field['hash'],
 693         ];
 694         $params = array_merge($params,
 695           CRM_Mailing_BAO_BouncePattern::match($result->getMessage())
 696         );
 697         CRM_Mailing_Event_BAO_Bounce::create($params);
 698       }
 699       elseif (is_a($result, 'PEAR_Error') && $mailing->sms_provider_id) {
 700         // Handle SMS errors: CRM-15426
 701         $job_id = intval($this->id);
 702         $mailing_id = intval($mailing->id);
 703         CRM_Core_Error::debug_log_message("Failed to send SMS message. Vars: mailing_id: ${mailing_id}, job_id: ${job_id}. Error message follows.");
 704         CRM_Core_Error::debug_log_message($result->getMessage());
 705       }
 706       else {
 707         // Register the delivery event.
 708         $deliveredParams[] = $field['id'];
 709         $targetParams[] = $field['contact_id'];
 710
 711         $count++;
 712         // dev/core#1768 Mail sync interval is now configurable.
 713         if ($count % $mail_sync_interval == 0) {
 714           $this->writeToDB(
 715             $deliveredParams,
 716             $targetParams,
 717             $mailing,
 718             $job_date
 719           );
 720           $count = 0;
 721
 722           // hack to stop mailing job at run time, CRM-4246.
 723           // to avoid making too many DB calls for this rare case
 724           // lets do it when we snapshot
 725           $status = CRM_Core_DAO::getFieldValue(
 726             'CRM_Mailing_DAO_MailingJob',
 727             $this->id,
 728             'status',
 729             'id',
 730             TRUE
 731           );
 732
 733           if ($status != 'Running') {
 734             return FALSE;
 735           }
 736         }
 737       }
 738
 739       unset($result);
 740
 741       // seems like a successful delivery or bounce, lets decrement error count
 742       // only if we have smtp connection errors
 743       if ($smtpConnectionErrors > 0) {
 744         $smtpConnectionErrors--;
 745       }
 746
 747       // If we have enabled the Throttle option, this is the time to enforce it.
 748       $mailThrottleTime = Civi::settings()->get('mailThrottleTime');
 749       if (!empty($mailThrottleTime)) {
 750         usleep((int ) $mailThrottleTime);
 751       }
 752     }
 753
 754     $result = $this->writeToDB(
 755       $deliveredParams,
 756       $targetParams,
 757       $mailing,
 758       $job_date
 759     );
 760
 761     if ($retryGroup) {
 762       return FALSE;
 763     }
 764
 765     return $result;
 766   }
 767
 768   /**
 769    * Determine if an SMTP error is temporary or permanent.
 770    *
 771    * @param string $message
 772    *   PEAR error message.
 773    * @return bool
 774    *   TRUE - Temporary/retriable error
 775    *   FALSE - Permanent/non-retriable error
 776    */
 777   protected function isTemporaryError($message) {
 778     // SMTP response code is buried in the message.
 779     $code = preg_match('/ \(code: (.+), response: /', $message, $matches) ? $matches[1] : '';
 780
 781     if (strpos($message, 'Failed to write to socket') !== FALSE) {
 782       return TRUE;
 783     }
 784
 785     // Register 5xx SMTP response code (permanent failure) as bounce.
 786     if (isset($code[0]) && $code[0] === '5') {
 787       return FALSE;
 788     }
 789
 790     if (strpos($message, 'Failed to set sender') !== FALSE) {
 791       return TRUE;
 792     }
 793
 794     if (strpos($message, 'Failed to add recipient') !== FALSE) {
 795       return TRUE;
 796     }
 797
 798     if (strpos($message, 'Failed to send data') !== FALSE) {
 799       return TRUE;
 800     }
 801
 802     return FALSE;
 803   }
 804
 805   /**
 806    * Cancel a mailing.
 807    *
 808    * @param int $mailingId
 809    *   The id of the mailing to be canceled.
 810    */
 811   public static function cancel($mailingId) {
 812     $sql = "
 813 SELECT *
 814 FROM   civicrm_mailing_job
 815 WHERE  mailing_id = %1
 816 AND    is_test = 0
 817 AND    ( ( job_type IS NULL ) OR
 818            job_type <> 'child' )
 819 ";
 820     $params = [1 => [$mailingId, 'Integer']];
 821     $job = CRM_Core_DAO::executeQuery($sql, $params);
 822     if ($job->fetch() &&
 823       in_array($job->status, ['Scheduled', 'Running', 'Paused'])
 824     ) {
 825
 826       self::create(['id' => $job->id, 'end_date' => date('YmdHis'), 'status' => 'Canceled']);
 827
 828       // also cancel all child jobs
 829       $sql = "
 830 UPDATE civicrm_mailing_job
 831 SET    status = 'Canceled',
 832        end_date = %2
 833 WHERE  parent_id = %1
 834 AND    is_test = 0
 835 AND    job_type = 'child'
 836 AND    status IN ( 'Scheduled', 'Running', 'Paused' )
 837 ";
 838       $params = [
 839         1 => [$job->id, 'Integer'],
 840         2 => [date('YmdHis'), 'Timestamp'],
 841       ];
 842       CRM_Core_DAO::executeQuery($sql, $params);
 843     }
 844   }
 845
 846   /**
 847    * Pause a mailing
 848    *
 849    * @param int $mailingID
 850    *   The id of the mailing to be paused.
 851    */
 852   public static function pause($mailingID) {
 853     $sql = "
 854       UPDATE civicrm_mailing_job
 855       SET status = 'Paused'
 856       WHERE mailing_id = %1
 857       AND is_test = 0
 858       AND status IN ('Scheduled', 'Running')
 859     ";
 860     CRM_Core_DAO::executeQuery($sql, [1 => [$mailingID, 'Integer']]);
 861   }
 862
 863   /**
 864    * Resume a mailing
 865    *
 866    * @param int $mailingID
 867    *   The id of the mailing to be resumed.
 868    */
 869   public static function resume($mailingID) {
 870     $sql = "
 871       UPDATE civicrm_mailing_job
 872       SET status = 'Scheduled'
 873       WHERE mailing_id = %1
 874       AND is_test = 0
 875       AND start_date IS NULL
 876       AND status = 'Paused'
 877     ";
 878     CRM_Core_DAO::executeQuery($sql, [1 => [$mailingID, 'Integer']]);
 879
 880     $sql = "
 881       UPDATE civicrm_mailing_job
 882       SET status = 'Running'
 883       WHERE mailing_id = %1
 884       AND is_test = 0
 885       AND start_date IS NOT NULL
 886       AND status = 'Paused'
 887     ";
 888     CRM_Core_DAO::executeQuery($sql, [1 => [$mailingID, 'Integer']]);
 889   }
 890
 891   /**
 892    * Return a translated status enum string.
 893    *
 894    * @param string $status
 895    *   The status enum.
 896    *
 897    * @return string
 898    *   The translated version
 899    */
 900   public static function status($status) {
 901     static $translation = NULL;
 902
 903     if (empty($translation)) {
 904       $translation = [
 905         'Scheduled' => ts('Scheduled'),
 906         'Running' => ts('Running'),
 907         'Complete' => ts('Complete'),
 908         'Paused' => ts('Paused'),
 909         'Canceled' => ts('Canceled'),
 910       ];
 911     }
 912     return CRM_Utils_Array::value($status, $translation, ts('Not scheduled'));
 913   }
 914
 915   /**
 916    * Return a workflow clause for use in SQL queries,
 917    * to only process jobs that are approved.
 918    *
 919    * @return string
 920    *   For use in a WHERE clause
 921    */
 922   public static function workflowClause() {
 923     // add an additional check and only process
 924     // jobs that are approved
 925     if (CRM_Mailing_Info::workflowEnabled()) {
 926       $approveOptionID = CRM_Core_PseudoConstant::getKey('CRM_Mailing_BAO_Mailing', 'approval_status_id', 'Approved');
 927       if ($approveOptionID) {
 928         return " AND m.approval_status_id = $approveOptionID ";
 929       }
 930     }
 931     return '';
 932   }
 933
 934   /**
 935    * @param array $deliveredParams
 936    * @param array $targetParams
 937    * @param $mailing
 938    * @param $job_date
 939    *
 940    * @return bool
 941    * @throws CRM_Core_Exception
 942    * @throws Exception
 943    */
 944   public function writeToDB(
 945     &$deliveredParams,
 946     &$targetParams,
 947     &$mailing,
 948     $job_date
 949   ) {
 950     static $activityTypeID = NULL;
 951     static $writeActivity = NULL;
 952
 953     if (!empty($deliveredParams)) {
 954       CRM_Mailing_Event_BAO_Delivered::bulkCreate($deliveredParams);
 955       $deliveredParams = [];
 956     }
 957
 958     if ($writeActivity === NULL) {
 959       $writeActivity = Civi::settings()->get('write_activity_record');
 960     }
 961
 962     if (!$writeActivity) {
 963       return TRUE;
 964     }
 965
 966     $result = TRUE;
 967     if (!empty($targetParams) && !empty($mailing->scheduled_id)) {
 968       if (!$activityTypeID) {
 969         if ($mailing->sms_provider_id) {
 970           $mailing->subject = $mailing->name;
 971           $activityTypeID = CRM_Core_PseudoConstant::getKey('CRM_Activity_BAO_Activity', 'activity_type_id', 'Mass SMS'
 972           );
 973         }
 974         else {
 975           $activityTypeID = CRM_Core_PseudoConstant::getKey('CRM_Activity_BAO_Activity', 'activity_type_id', 'Bulk Email');
 976         }
 977         if (!$activityTypeID) {
 978           throw new CRM_Core_Exception(ts('No relevant activity type found when recording Mailing Event delivered Activity'));
 979         }
 980       }
 981
 982       $activity = [
 983         'source_contact_id' => $mailing->scheduled_id,
 984         'activity_type_id' => $activityTypeID,
 985         'source_record_id' => $this->mailing_id,
 986         'activity_date_time' => $job_date,
 987         'subject' => $mailing->subject,
 988         'status_id' => 'Completed',
 989         'campaign_id' => $mailing->campaign_id,
 990       ];
 991
 992       //check whether activity is already created for this mailing.
 993       //if yes then create only target contact record.
 994       $query = "
 995 SELECT id
 996 FROM   civicrm_activity
 997 WHERE  civicrm_activity.activity_type_id = %1
 998 AND    civicrm_activity.source_record_id = %2
 999 ";
1000
1001       $queryParams = [
1002         1 => [$activityTypeID, 'Integer'],
1003         2 => [$this->mailing_id, 'Integer'],
1004       ];
1005       $activityID = CRM_Core_DAO::singleValueQuery($query, $queryParams);
1006       $targetRecordID = CRM_Core_PseudoConstant::getKey('CRM_Activity_BAO_ActivityContact', 'record_type_id', 'Activity Targets');
1007
1008       $activityTargets = [];
1009       foreach ($targetParams as $id) {
1010         $activityTargets[$id] = ['contact_id' => (int) $id];
1011       }
1012       if ($activityID) {
1013         $activity['id'] = $activityID;
1014
1015         // CRM-9519
1016         if (CRM_Core_BAO_Email::isMultipleBulkMail()) {
1017           // make sure we don't attempt to duplicate the target activity
1018           // @todo - we don't have to do one contact at a time....
1019           foreach ($activityTargets as $key => $target) {
1020             $sql = "
1021 SELECT id
1022 FROM   civicrm_activity_contact
1023 WHERE  activity_id = $activityID
1024 AND    contact_id = {$target['contact_id']}
1025 AND    record_type_id = $targetRecordID
1026 ";
1027             if (CRM_Core_DAO::singleValueQuery($sql)) {
1028               unset($activityTargets[$key]);
1029             }
1030           }
1031         }
1032       }
1033
1034       try {
1035         $activity = civicrm_api3('Activity', 'create', $activity);
1036         ActivityContact::save(FALSE)->setRecords($activityTargets)->setDefaults(['activity_id' => $activity['id'], 'record_type_id' => $targetRecordID])->execute();
1037       }
1038       catch (Exception $e) {
1039         $result = FALSE;
1040       }
1041
1042       $targetParams = [];
1043     }
1044
1045     return $result;
1046   }
1047
1048   /**
1049    * Search the mailing-event queue for a list of pending delivery tasks.
1050    *
1051    * @param int $jobId
1052    * @param string $medium
1053    *   Ex: 'email' or 'sms'.
1054    *
1055    * @return \CRM_Mailing_Event_BAO_Queue
1056    *   A query object whose rows provide ('id', 'contact_id', 'hash') and ('email' or 'phone').
1057    */
1058   public static function findPendingTasks($jobId, $medium) {
1059     $eq = new CRM_Mailing_Event_BAO_Queue();
1060     $queueTable = CRM_Mailing_Event_BAO_Queue::getTableName();
1061     $emailTable = CRM_Core_BAO_Email::getTableName();
1062     $phoneTable = CRM_Core_BAO_Phone::getTableName();
1063     $contactTable = CRM_Contact_BAO_Contact::getTableName();
1064     $deliveredTable = CRM_Mailing_Event_BAO_Delivered::getTableName();
1065     $bounceTable = CRM_Mailing_Event_BAO_Bounce::getTableName();
1066
1067     $query = "  SELECT      $queueTable.id,
1068                                 $emailTable.email as email,
1069                                 $queueTable.contact_id,
1070                                 $queueTable.hash,
1071                                 NULL as phone
1072                     FROM        $queueTable
1073                     INNER JOIN  $emailTable
1074                             ON  $queueTable.email_id = $emailTable.id
1075                     INNER JOIN  $contactTable
1076                             ON  $contactTable.id = $emailTable.contact_id
1077                     LEFT JOIN   $deliveredTable
1078                             ON  $queueTable.id = $deliveredTable.event_queue_id
1079                     LEFT JOIN   $bounceTable
1080                             ON  $queueTable.id = $bounceTable.event_queue_id
1081                     WHERE       $queueTable.job_id = " . $jobId . "
1082                         AND     $deliveredTable.id IS null
1083                         AND     $bounceTable.id IS null
1084                         AND    $contactTable.is_opt_out = 0";
1085
1086     if ($medium === 'sms') {
1087       $query = "
1088                     SELECT      $queueTable.id,
1089                                 $phoneTable.phone as phone,
1090                                 $queueTable.contact_id,
1091                                 $queueTable.hash,
1092                                 NULL as email
1093                     FROM        $queueTable
1094                     INNER JOIN  $phoneTable
1095                             ON  $queueTable.phone_id = $phoneTable.id
1096                     INNER JOIN  $contactTable
1097                             ON  $contactTable.id = $phoneTable.contact_id
1098                     LEFT JOIN   $deliveredTable
1099                             ON  $queueTable.id = $deliveredTable.event_queue_id
1100                     LEFT JOIN   $bounceTable
1101                             ON  $queueTable.id = $bounceTable.event_queue_id
1102                     WHERE       $queueTable.job_id = " . $jobId . "
1103                         AND     $deliveredTable.id IS null
1104                         AND     $bounceTable.id IS null
1105                         AND    ( $contactTable.is_opt_out = 0
1106                         OR       $contactTable.do_not_sms = 0 )";
1107     }
1108     $eq->query($query);
1109     return $eq;
1110   }
1111
1112   /**
1113    * Delete the mailing job.
1114    *
1115    * @param int $id
1116    * @deprecated
1117    * @return bool
1118    */
1119   public static function del($id) {
1120     return (bool) self::deleteRecord(['id' => $id]);
1121   }
1122
1123 }