Fix #6923 - Scheduling and delivery of press release to wordpress
[Pman.Core] / Notify.php
1 <?php
2 require_once 'Pman.php';
3
4 /**
5  * notification script runner
6  *
7  * This does not actualy send stuf out, it only starts the NotifySend/{id}
8  * which does the actuall notifcations.
9  *
10  * It manages a pool of notifiers.
11  * 
12  * 
13  */
14
15
16 class Pman_Core_Notify extends Pman
17 {
18     
19     static $cli_desc = "Runs the notification queue (usually from cron)
20                         Normally used to sends out emails to anyone in the notification list.
21     
22                         /etc/cron.d/pman-core-notify
23                         * *  * * *     www-data     /usr/bin/php /home/gitlive/web.mtrack/admin.php  Core/Notify > /dev/null
24     
25 ";
26     
27     static $cli_opts = array(
28         'debug' => array(
29             'desc' => 'Turn on debugging (see DataObjects debugLevel )',
30             'default' => 0,
31             'short' => 'v',
32             'min' => 1,
33             'max' => 1,
34             
35         ),
36         'list' => array(
37             'desc' => 'List message to send, do not send them..',
38             'default' => 0,
39             'short' => 'l',
40             'min' => 0,
41             'max' => 0,
42             
43         ),
44         'old' => array(
45             'desc' => 'Show old messages.. (and new messages...)',
46             'default' => 0,
47             'short' => 'o',
48             'min' => 0,
49             'max' => 0,
50             
51         ),
52         'force' => array(
53             'desc' => 'Force redelivery, even if it has been sent before or not queued...',
54             'default' => 0,
55             'short' => 'f',
56             'min' => 0,
57             'max' => 0,
58         ),
59        /* removed - use GenerateNotifcations.php hooked classes
60          'generate' =>  'Generate notifications for a table, eg. cash_invoice',
61             
62         ),
63         */
64          'limit' => array(
65             'desc' => 'Limit search for no. to send to ',
66             'default' => 1000,
67             'short' => 'L',
68             'min' => 0,
69             'max' => 999,
70         ),
71         'dryrun' => array(
72             'desc' => 'Dry run - do not send.',
73             'default' => 0,
74             'short' => 'D',
75             'min' => 0,
76             'max' => 0,
77         ),
78         'poolsize' => array(
79             'desc' => 'Pool size',
80             'default' => 10,
81             'short' => 'P',
82             'min' => 0,
83             'max' => 100,
84         ),
85     );
86     /**
87      * @var $nice_level Unix 'nice' level to stop it jamming server up.
88      */
89     var $nice_level = false;
90     /**
91      * @var $max_pool_size maximum runners to start at once.
92      */
93     var $max_pool_size = 10;
94     /**
95      * @var $max_to_domain maximum connections to make to a single domain
96      */
97     var $max_to_domain = 10;
98     
99     /**
100      * @var $maxruntime - maximum seconds a child is allowed to run - defaut 2 minutes
101      */
102     var $maxruntime = 120;
103     
104     /**
105     * @var {Boolean} log_events - default true if events should be logged.
106     */
107     var $log_events = true;
108     /**
109     * @var {Number} try_again_minutes how long after failing to try again default = 30 if max runtime fails
110     */
111     var $try_again_minutes = 30;
112     
113     /**
114      * @var {String} table - the table that the class will query for notification events
115      */
116     var $table = 'core_notify';
117     /**
118      * @var {String} target - the application that will run for each Row in the table (eg. Pman/Core/NotifySend)
119      */
120     var $target = 'Core/NotifySend';
121     
122     
123     
124     var $evtype = ''; // any notification...
125                     // this script should only handle EMAIL notifications..
126                     
127                     
128     var $force = false;
129     function getAuth()
130     {
131         $ff = HTML_FlexyFramework::get();
132         if (!$ff->cli) {
133             die("access denied");
134         }
135         HTML_FlexyFramework::ensureSingle($_SERVER["SCRIPT_NAME"] .'|'. __FILE__, $this);
136         return true;
137     }
138     
139     var $pool = array();
140     
141     function parseArgs(&$opts)
142     {
143         if ($opts['debug']) {
144             DB_DataObject::debugLevel($opts['debug']);
145             print_r($opts);
146         }
147         $this->opts = $opts;
148         if (!empty($opts['poolsize'])) {
149             $this->max_pool_size = $opts['poolsize'];
150         }
151         
152         if (empty($opts['limit'])) {
153             $opts['limit'] = '1000'; // not sure why it's not picking up the defautl..
154         }
155         
156         if (!empty($opts['old'])) {
157             $opts['list'] = 1; // force listing..
158         }
159         
160         $this->force = empty($opts['force']) ? 0 : 1;
161      
162         if (!empty($opts['send-to'])) {
163             $this->send_to = $opts['send-to'];
164         }
165     }
166     
167     
168     function get($r,$opts=array())    
169     {
170         $this->parseArgs($opts); 
171          
172         //date_default_timezone_set('UTC');
173         
174         
175         $this->generateNotifications();
176         
177          
178         //DB_DataObject::debugLevel(1);
179         $w = DB_DataObject::factory($this->table);
180         $total = 0;
181         
182         if (!empty($opts['old'])) {
183             // show old and new...
184             
185             $w->orderBy('act_when DESC'); // latest first
186             $w->limit($opts['limit']); // we can run 
187             $total = min($w->count(), $opts['limit']);
188         } else {   
189             // standard
190             
191             //$w->whereAdd('act_when > sent'); // eg.. sent is not valid..
192             $w->whereAdd("sent < '1970-01-01' OR sent IS NULL"); // eg.. sent is not valid..
193             
194             $w->whereAdd('act_start > NOW() - INTERVAL 14 DAY'); // ignore the ones stuck in the queue
195             if (!$this->force) {
196                 $w->whereAdd('act_when < NOW()'); // eg.. not if future..
197             }
198     
199             $w->orderBy('act_when ASC'); // oldest first.
200             $total = min($w->count(), $opts['limit']);
201             $this->logecho("QUEUE is {$w->count()} only running " . ((int) $opts['limit']));
202             
203             $w->limit($opts['limit']); // we can run 1000 ...
204         }
205         
206         if (!empty($this->evtype)) {
207             $w->evtype = $this->evtype;
208         }
209         
210         $w->autoJoin();
211         $w->find();
212         
213         $ar = array(); // $w->fetchAll();
214         
215         if (!empty($opts['list'])) {
216             
217             
218             while ($w->fetch()) { 
219                 $o = $w->object();
220                 
221                 
222                 $this->logecho("{$w->id} : {$w->person()->email} email    : ".
223                         $o->toEventString()."    ". $w->status()  );
224             }
225             exit;
226         }
227         
228         //echo "BATCH SIZE: ".  count($ar) . "\n";
229         $pushed = array();
230         $requeue = array();
231         while (true) {
232             if ($w->fetch()) {
233                 $ar[] = clone($w);
234                 $total--;
235             }
236             
237             $this->logecho("BATCH SIZE: ".  (count($ar) + $total) );
238             
239             if (empty($ar)) {
240                 $this->logecho("COMPLETED MAIN QUEUE - running deleted");
241                 
242                 if (empty($pushed)) {
243                     break;
244                 }
245                 $ar = $pushed;
246                 $pushed = false;
247                 continue;
248             }
249             
250             
251             $p = array_shift($ar);
252             if (!$this->poolfree()) {
253                 array_unshift($ar,$p); /// put it back on..
254                 sleep(3);
255                 continue;
256             }
257             $email = $p->person() ? $p->person()->email : $p->to_email;
258             
259             if ($this->poolHasDomain($email) > $this->max_to_domain) {
260                 
261                 if ($pushed === false) {
262                     // we only try once to requeue..
263                     $requeue[] = $p;
264                     continue;
265                 }
266                 $pushed[] = $p;
267                 
268                 
269                 //sleep(3);
270                 continue;
271             }
272             
273             
274             $this->run($p->id,$email);
275             
276             
277             
278         }
279         
280         // we should have a time limit here...
281         while(count($this->pool)) {
282             $this->poolfree();
283              sleep(3);
284         }
285          
286         foreach($requeue as $p) {
287             $pp = clone($p);
288             $p->act_when = $p->sqlValue('NOW + INTERVAL 1 MINUTE');
289             $p->update($pp);
290             
291         }
292         
293         
294         $this->logecho("DONE");
295         exit;
296     }
297     
298     function generateNotifications()
299     {
300         // this should check each module for 'GenerateNotifications.php' class..
301         //and run it if found..
302         $ff = HTML_FlexyFramework::get();
303        
304         $disabled = explode(',', $ff->disable);
305
306         $modules = array_reverse($this->modulesList());
307         
308         // move 'project' one to the end...
309         
310         foreach ($modules as $module){
311             if(in_array($module, $disabled)){
312                 continue;
313             }
314             $file = $this->rootDir. "/Pman/$module/GenerateNotifications.php";
315             if(!file_exists($file)){
316                 continue;
317             }
318             
319             require_once $file;
320             $class = "Pman_{$module}_GenerateNotifications";
321             $x = new $class;
322             if(!method_exists($x, 'generate')){
323                 continue;
324             };
325             //echo "$module\n";
326             $x->generate($this);
327         }
328                 
329     
330     }
331     
332     
333     
334     function run($id, $email='', $cmdOpts="")
335     {
336         
337         static $renice = false;
338         if (!$renice) {
339             require_once 'System.php';
340             $renice = System::which('renice');
341         }
342         
343         // phpinfo();exit;
344         
345         
346         $tn =  $this->tempName('stdout', true);
347         $descriptorspec = array(
348             0 => array("pipe", 'r'),  // stdin is a pipe that the child will read from
349             1 => array("file", $tn, 'w'),  // stdout is a pipe that the child will write to
350             2 => array("pipe", "w") // stderr is a file to write to
351          );
352         
353         static $php = false;
354         if (!$php) {
355             require_once 'System.php';
356             $php = System::which('php');
357         }
358         
359         $sn =  $_SERVER["SCRIPT_NAME"];
360         
361         $cwd = $sn[0] == '/' ? dirname($sn) : dirname(realpath(getcwd() . '/'. $sn)); // same as run on.. (so script should end up being same relatively..)
362         $app = $cwd . '/' . basename($_SERVER["SCRIPT_NAME"]) . '  ' . $this->target . '/'. $id;
363         if ($this->force) {
364             $app .= ' -f';
365         }
366         if (!empty($this->send_to)) {
367             $app .= ' --sent-to='.escapeshellarg($this->send_to);
368         }
369         $cmd = 'exec ' . $php . ' ' . $app . ' ' . $cmdOpts; //. ' &';
370         
371        
372         $pipe = array();
373         $this->logecho("call proc_open $cmd");
374         
375         
376         if ($this->max_pool_size === 1) {
377             passthru($cmd);
378             return;
379         }
380         
381         
382         if (!empty($this->opts['dryrun'])) {
383             $this->logecho("DRY RUN");
384             return;
385         }
386         
387         $p = proc_open($cmd, $descriptorspec, $pipes, $cwd );
388         $info =  proc_get_status($p);
389         
390         if ($this->nice_level !== false) { 
391             $rcmd = "$renice {$this->nice_level} {$info['pid']}";
392             `$rcmd`;
393         } 
394         $this->pool[] = array(
395                 'proc' => $p,
396                 'pid' => $info['pid'],
397                 'out' => $tn,
398                 'cmd' => $cmd,
399                 'email' => $email,
400                 'pipes' => $pipes,
401                 'notify_id' => $id,
402                 'started' => time()
403             
404                 
405         );
406         $this->logecho("RUN ({$info['pid']}) $cmd ");
407     }
408     
409     function poolfree()
410     {
411         $pool = array();
412         clearstatcache();
413          
414         foreach($this->pool as $p) {
415              
416             //echo "CHECK PID: " . $p['pid'] . "\n";
417             $info =  proc_get_status($p['proc']);
418             //var_dump($info);
419             
420             // update if necessday.
421             if ($info['pid'] && $p['pid'] != $info['pid']) {
422                 $this->logecho("CHANING PID FROM " . $p['pid']  .  "  TO ". $info['pid']);
423                 $p['pid'] = $info['pid'];
424             }
425             
426             //echo @file_get_contents('/proc/'. $p['pid'] .'/cmdline') . "\n";
427             
428             if ($info['running']) {
429             
430                 //if (file_exists('/proc/'.$p['pid'])) {
431                 $runtime = time() - $p['started'];
432                 //echo "RUNTIME ({$p['pid']}): $runtime\n";
433                 if ($runtime > $this->maxruntime) {
434                     
435                     proc_terminate($p['proc'], 9);
436                     //fclose($p['pipes'][1]);
437                     fclose($p['pipes'][0]);
438                     fclose($p['pipes'][2]);
439                     $this->logecho("TERMINATING: ({$p['pid']}) " . $p['cmd'] . " : " . file_get_contents($p['out']));
440                     @unlink($p['out']);
441                     
442                     // schedule again
443                     $w = DB_DataObject::factory($this->table);
444                     $w->get($p['notify_id']);
445                     $ww = clone($w);
446                     if ($this->log_events) {
447                         $this->addEvent('NOTIFY', $w, 'TERMINATED - TIMEOUT');
448                     }
449                     $w->act_when = date('Y-m-d H:i:s', strtotime("NOW + {$this->try_again_minutes} MINUTES"));
450                     $w->update($ww);
451                     
452                     continue;
453                 }
454                 
455                 $pool[] = $p;
456                 continue;
457             }
458             fclose($p['pipes'][0]);
459             fclose($p['pipes'][2]);
460             //echo "CLOSING: ({$p['pid']}) " . $p['cmd'] . " : " . file_get_contents($p['out']) . "\n";
461             //fclose($p['pipes'][1]);
462             
463             proc_close($p['proc']);
464             
465             
466             //clearstatcache();
467             //if (file_exists('/proc/'.$p['pid'])) {
468             //    $pool[] = $p;
469             //    continue;
470             //}
471             $this->logecho("ENDED: ({$p['pid']}) " .  $p['cmd'] . " : " . file_get_contents($p['out']) );
472             @unlink($p['out']);
473             //unlink($p['out']);
474         }
475         $this->logecho("POOL SIZE: ". count($pool) );
476         $this->pool = $pool;
477         if (count($pool) < $this->max_pool_size) {
478             return true;
479         }
480         return false;
481         
482     }
483     /**
484      * see if pool is already trying to deliver to this domain.?
485      * -- if so it get's pushed to the end of the queue.
486      *
487      */
488     function poolHasDomain($email)
489     {
490         $ret = 0;
491         $ea = explode('@',$email);
492         $dom = strtolower(array_pop($ea));
493         foreach($this->pool as $p) {
494             $ea = explode('@',$p['email']);
495             $mdom = strtolower(array_pop($ea));
496             if ($mdom == $dom) {
497                 $ret++;
498             }
499         }
500         return $ret;
501         
502     }
503
504     function output()
505     {
506         $this->logecho("DONE");
507         exit;
508     }
509     function logecho($str)
510     {
511         echo date("Y-m-d H:i:s - ") . $str . "\n";
512     }
513 }