include/queue.php

   1 <?php
   2
   3 use Friendica\Core\Config;
   4
   5 require_once 'include/queue_fn.php';
   6 require_once 'include/dfrn.php';
   7 require_once 'include/datetime.php';
   8 require_once 'include/items.php';
   9 require_once 'include/bbcode.php';
  10 require_once 'include/socgraph.php';
  11 require_once 'include/cache.php';
  12
  13 function queue_run(&$argv, &$argc) {
  14         global $a;
  15
  16         if ($argc > 1) {
  17                 $queue_id = intval($argv[1]);
  18         } else {
  19                 $queue_id = 0;
  20         }
  21
  22         $cachekey_deadguy = 'queue_run:deadguy:';
  23         $cachekey_server = 'queue_run:server:';
  24
  25         if (!$queue_id) {
  26
  27                 logger('queue: start');
  28
  29                 // Handling the pubsubhubbub requests
  30                 proc_run(array('priority' => PRIORITY_HIGH, 'dont_fork' => true), 'include/pubsubpublish.php');
  31
  32                 $r = q("SELECT `queue`.*, `contact`.`name`, `contact`.`uid` FROM `queue`
  33                         INNER JOIN `contact` ON `queue`.`cid` = `contact`.`id`
  34                         WHERE `queue`.`created` < UTC_TIMESTAMP() - INTERVAL 3 DAY");
  35
  36                 if (dbm::is_result($r)) {
  37                         foreach ($r as $rr) {
  38                                 logger('Removing expired queue item for ' . $rr['name'] . ', uid=' . $rr['uid']);
  39                                 logger('Expired queue data: ' . $rr['content'], LOGGER_DATA);
  40                         }
  41                         q("DELETE FROM `queue` WHERE `created` < UTC_TIMESTAMP() - INTERVAL 3 DAY");
  42                 }
  43
  44                 /*
  45                  * For the first 12 hours we'll try to deliver every 15 minutes
  46                  * After that, we'll only attempt delivery once per hour.
  47                  */
  48                 $r = q("SELECT `id` FROM `queue` WHERE ((`created` > UTC_TIMESTAMP() - INTERVAL 12 HOUR AND `last` < UTC_TIMESTAMP() - INTERVAL 15 MINUTE) OR (`last` < UTC_TIMESTAMP() - INTERVAL 1 HOUR)) ORDER BY `cid`, `created`");
  49
  50                 call_hooks('queue_predeliver', $a, $r);
  51
  52                 if (dbm::is_result($r)) {
  53                         foreach ($r as $q_item) {
  54                                 logger('Call queue for id '.$q_item['id']);
  55                                 proc_run(array('priority' => PRIORITY_LOW, 'dont_fork' => true), "include/queue.php", $q_item['id']);
  56                         }
  57                 }
  58                 return;
  59         }
  60
  61
  62         // delivering
  63
  64         require_once 'include/salmon.php';
  65         require_once 'include/diaspora.php';
  66
  67         $r = q("SELECT * FROM `queue` WHERE `id` = %d LIMIT 1",
  68                 intval($queue_id));
  69
  70         if (!dbm::is_result($r)) {
  71                 return;
  72         }
  73
  74         $q_item = $r[0];
  75
  76         $c = q("SELECT * FROM `contact` WHERE `id` = %d LIMIT 1",
  77                 intval($q_item['cid'])
  78         );
  79
  80         if (!dbm::is_result($c)) {
  81                 remove_queue_item($q_item['id']);
  82                 return;
  83         }
  84
  85         $dead = Cache::get($cachekey_deadguy.$c[0]['notify']);
  86
  87         if (!is_null($dead) && $dead) {
  88                 logger('queue: skipping known dead url: '.$c[0]['notify']);
  89                 update_queue_time($q_item['id']);
  90                 return;
  91         }
  92
  93         $server = poco_detect_server($c[0]['url']);
  94
  95         if ($server != "") {
  96                 $vital = Cache::get($cachekey_server.$server);
  97
  98                 if (is_null($vital)) {
  99                         logger("Check server ".$server." (".$c[0]["network"].")");
 100
 101                         $vital = poco_check_server($server, $c[0]["network"], true);
 102                         Cache::set($cachekey_server.$server, $vital, CACHE_QUARTER_HOUR);
 103                 }
 104
 105                 if (!is_null($vital) && !$vital) {
 106                         logger('queue: skipping dead server: '.$server);
 107                         update_queue_time($q_item['id']);
 108                         return;
 109                 }
 110         }
 111
 112         $u = q("SELECT `user`.*, `user`.`pubkey` AS `upubkey`, `user`.`prvkey` AS `uprvkey`
 113                 FROM `user` WHERE `uid` = %d LIMIT 1",
 114                 intval($c[0]['uid'])
 115         );
 116         if (!dbm::is_result($u)) {
 117                 remove_queue_item($q_item['id']);
 118                 return;
 119         }
 120
 121         $data      = $q_item['content'];
 122         $public    = $q_item['batch'];
 123         $contact   = $c[0];
 124         $owner     = $u[0];
 125
 126         $deliver_status = 0;
 127
 128         switch ($contact['network']) {
 129                 case NETWORK_DFRN:
 130                         logger('queue: dfrndelivery: item '.$q_item['id'].' for '.$contact['name'].' <'.$contact['url'].'>');
 131                         $deliver_status = dfrn::deliver($owner, $contact, $data);
 132
 133                         if ($deliver_status == (-1)) {
 134                                 update_queue_time($q_item['id']);
 135                                 Cache::set($cachekey_deadguy.$contact['notify'], true, CACHE_QUARTER_HOUR);
 136                         } else {
 137                                 remove_queue_item($q_item['id']);
 138                         }
 139                         break;
 140                 case NETWORK_OSTATUS:
 141                         if ($contact['notify']) {
 142                                 logger('queue: slapdelivery: item '.$q_item['id'].' for '.$contact['name'].' <'.$contact['url'].'>');
 143                                 $deliver_status = slapper($owner, $contact['notify'], $data);
 144
 145                                 if ($deliver_status == (-1)) {
 146                                         update_queue_time($q_item['id']);
 147                                         Cache::set($cachekey_deadguy.$contact['notify'], true, CACHE_QUARTER_HOUR);
 148                                 } else {
 149                                         remove_queue_item($q_item['id']);
 150                                 }
 151                         }
 152                         break;
 153                 case NETWORK_DIASPORA:
 154                         if ($contact['notify']) {
 155                                 logger('queue: diaspora_delivery: item '.$q_item['id'].' for '.$contact['name'].' <'.$contact['url'].'>');
 156                                 $deliver_status = Diaspora::transmit($owner, $contact, $data, $public, true);
 157
 158                                 if ($deliver_status == (-1)) {
 159                                         update_queue_time($q_item['id']);
 160                                         Cache::set($cachekey_deadguy.$contact['notify'], true, CACHE_QUARTER_HOUR);
 161                                 } else {
 162                                         remove_queue_item($q_item['id']);
 163                                 }
 164                         }
 165                         break;
 166
 167                 default:
 168                         $params = array('owner' => $owner, 'contact' => $contact, 'queue' => $q_item, 'result' => false);
 169                         call_hooks('queue_deliver', $a, $params);
 170
 171                         if ($params['result']) {
 172                                 remove_queue_item($q_item['id']);
 173                         } else {
 174                                 update_queue_time($q_item['id']);
 175                         }
 176                         break;
 177
 178         }
 179         logger('Deliver status '.(int)$deliver_status.' for item '.$q_item['id'].' to '.$contact['name'].' <'.$contact['url'].'>');
 180
 181         return;
 182 }