cron_job_expired_workflow.inc

<?php
require_once SQ_CORE_PACKAGE_PATH.'/system/cron/cron_job/cron_job.inc';
require_once SQ_INCLUDE_PATH.'/workflow_manager.inc';
require_once SQ_FUDGE_PATH.'/general/datetime.inc';

class Cron_Job_Expired_WorkFlow extends Cron_Job
{


    function __construct($assetid=0)
    {
        parent::__construct($assetid);

    }//end constructor


    public function create(Array &$link)
    {
        require_once SQ_CORE_PACKAGE_PATH.'/system/system_asset_fns.inc';
        if (!system_asset_fns_create_pre_check($this)) {
            return FALSE;
        }
        $GLOBALS['SQ_SYSTEM']->changeDatabaseConnection('db2');
        $GLOBALS['SQ_SYSTEM']->doTransaction('BEGIN');

        if ($linkid = parent::create($link)) {
            if (!system_asset_fns_create_cleanup($this)) {
                $linkid = FALSE;
            }
        }

        if ($linkid) {
            $GLOBALS['SQ_SYSTEM']->doTransaction('COMMIT');
        } else {
            $GLOBALS['SQ_SYSTEM']->doTransaction('ROLLBACK');
        }

        $GLOBALS['SQ_SYSTEM']->restoreDatabaseConnection();
        return $linkid;

    }//end create()


    protected function _getName($short_name=FALSE)
    {
        return translate('cron_job_expired_workflow');

    }//end _getName()


    public function canClone()
    {
        return FALSE;

    }//end canClone()


    public function canDelete()
    {
        return FALSE;

    }//end canDelete()
    

    protected function _exec(&$msg)
    {
        try {
            if (MatrixDAL::getDbType() === 'oci') {
                $sql = 'SELECT
                            aw.assetid, aw.schemaid, aw.wflow , aw.granted , a.name
                        FROM
                            sq_ast_wflow aw, sq_ast a
                        WHERE
                            aw.assetid = a.assetid
                        AND
                            DBMS_LOB.GETLENGTH(aw.wflow) > 0
                        AND
                            aw.granted = 1';

                $query = MatrixDAL::preparePdoQuery($sql);
                $rows  = MatrixDAL::executePdoAll($query);
            } else {
                $rows = MatrixDAL::executeAll('cron_job_expired_workflow', 'getAllRunningWorkflows');
            }
        } catch (DALException $e) {
            throw new Exception ('Unable to get running workflows due to database error: '.$e->getMessage());
        }
    
        $now = time();
        $cm = $GLOBALS['SQ_SYSTEM']->am->getSystemAsset('cron_manager');
        $last_run = $cm->attr('last_run');

        $ms = $GLOBALS['SQ_SYSTEM']->getMessagingService();

        foreach ($rows as $row) {
            $ms->openQueue();
            
            $workflow = unserialize($row['wflow']);
            $schemaid = $row['schemaid'];
            $wfm = $GLOBALS['SQ_SYSTEM']->getWorkflowManager();
            $step =& $wfm->getCurrentStep($workflow); // need by ref

            if (($step['started'] != 0) && ($step['expiry_time'] != 0)) {
                $exp_time = $step['started'] + $step['expiry_time'];

                if ($exp_time < $now) {
                    // workflow step has expired since the last cron run
                    
                    if ($step['escalate']) {
                        // Escalate the job, at any time
                        if (!$wfm->escalateWorkflow($row['assetid'], $schemaid)) {
                            trigger_localised_error('CRON0052', E_USER_WARNING, $schemaid, $assetid);
                        }
                    } else {
                        // Send a stale message, but only if this is the first
                        // time the cron has run that the workflow has gone stale
                        if ($exp_time > $last_run) {
                            $this->_processExpiredWorkflowStep($row['assetid'], $row['name'], $workflow['schema_name'], $schemaid, $step);
                        }
                    }
                    //break;
                }
            }
            
            // Don't send escalation message for all dependants (See #5034)
            $escalation_wf_msgs = Array();
            foreach($ms->queueContents() as $queued_message) {
                $ms_assetid =  $queued_message->parameters['assetid'];
                if ($ms_assetid) {
                    $parents = $GLOBALS['SQ_SYSTEM']->am->getDependantParents($ms_assetid, '', TRUE, FALSE);
                    foreach($parents as $key => $parentid) {
                        if ($ms_assetid == $parentid) {
                            unset($parents[$key]);
                        }
                    }//end foreach

                    // If asset has no dependant parents or more than 1 dependant parents, send the workflow message
                    if(count($parents) > 1 || empty($parents)) {
                        $escalation_wf_msgs[] = $queued_message;
                    } else {
                        // The asset's only dependant parent is not in workflow, it means workflow was started at this asset
                        $dependant_workflows = $wfm->getSchemaWorkflows($parents[0]);
                        if (empty($dependant_workflows[$schemaid]['steps']) || (isset($dependant_workflows[$schemaid]['complete']) && $dependant_workflows[$schemaid]['complete'])) {
                            $escalation_wf_msgs[] = $queued_message;
                        }//end if
                
                    }//end else
                }//end if
            }//end foreach

            $ms->abortQueue();
            
            // Send all the relevant escalation workflow messages
            $ms->send($escalation_wf_msgs);
            unset($escalation_wf_msgs);
        }   
        unset($rows);

        return SQ_CRON_JOB_COMPLETED;

    }//end _exec()


    protected function _processExpiredWorkflowStep($assetid, $asset_name, $schema_name, $schemaid, $step_data)
    {
        $wfm = $GLOBALS['SQ_SYSTEM']->getWorkflowManager();
        if ($wfm->silentWorkflowParty($assetid)) {
            return;
        }
        
        $ms = $GLOBALS['SQ_SYSTEM']->getMessagingService();
        $subject = 'Stale Workflow Step';
        $hrefs = $GLOBALS['SQ_SYSTEM']->am->getAssetBackendHref(Array($assetid => 'workflow'), FALSE);
        $href = str_replace('./', '', current($hrefs));
        $root_urls = explode("\n", SQ_CONF_SYSTEM_ROOT_URLS);
        $url = 'http://'.current($root_urls).'/'.$href;
        $type = 'asset.workflow.stale';
        $admins = $GLOBALS['SQ_SYSTEM']->am->getPermission($assetid, SQ_PERMISSION_ADMIN, TRUE, FALSE, TRUE);

        $schema_workflows = $wfm->getSchemaWorkflows($assetid);
        $workflow =& $schema_workflows[$schemaid];
        $schema_asset = $GLOBALS['SQ_SYSTEM']->am->getAsset($schemaid);
        $asset = $GLOBALS['SQ_SYSTEM']->am->getAsset($assetid);

        // don't automatically escalate, just notify admins
        $msg_reps = Array(
                        'asset_name'    => $asset_name,
                        'assetid'       => $assetid,
                        'step_id'       => implode('.', $workflow['current_step']),
                        'step_name'     => $step_data['step_name'],
                        'schema'        => $schema_name,
                        'started_time'  => easy_datetime($step_data['started']),
                        'expiry_time'   => easy_time_total($step_data['expiry_time']),
                        'workflow_url'  => $url,
                        'asset_url'     => $asset->getDependantParentsURL(),                        
                    );

        // If there are no admins, send it to system admins instead
        if (empty($admins)) $admins = Array(0);
        $msg = $ms->newMessage($admins, $type, $msg_reps);
        $msg->parameters['assetid'] = $assetid;
        
        // Does this step have a custom message for stale workflow?
        if (isset($step_data['message_stale']) && trim($step_data['message_stale']) !== '' ) {
            $msg->body = trim($step_data['message_stale']);
        }

        // Does this step have a custom subject for stale workflow?
        if (isset($step_data['subject_stale']) && trim($step_data['subject_stale']) !== '' ) {
            $subject = trim(array_get_index($step_data, 'subject_stale', 'Stale Workflow Step'));
            $msg->subject = (!empty($subject)) ? $subject : 'Stale Workflow Step';
        }//end if

        if (!is_null($schema_asset)) {
            if ($schema_asset->attr('schema_reply_to_email_address') != '') {
                $msg->parameters['reply_to'] = $schema_asset->attr('schema_reply_to_email_address');
            } else {
                $msg->parameters['reply_to'] = $msg->from;
            }
            if ($schema_asset->attr('schema_from_email_address') != '') {
                $msg->from = $schema_asset->attr('schema_from_email_address');
            }
        }

        $wfm->addMessageReplacements($asset, $msg->subject, $msg->body, $msg->replacements);
        $ms->enqueueMessage($msg);
        
        $GLOBALS['SQ_SYSTEM']->am->forgetAsset($asset);

    }//end _processExpiredWorkflowStep()


    public function run()
    {
        $res = parent::run();
        if ($res & SQ_CRON_JOB_REMOVE) {
            $res = $res - SQ_CRON_JOB_REMOVE;
        }
        return $res;

    }//end run()


}//end class



?>