<?php 

require_once IEM_ADDONS_PATH . "/multithread/config/config.php";
require_once IEM_ADDONS_PATH . "/multithread/api/commandline.php";

class Multithread_API extends API {

    private $db = NULL;
    private static $instance = NULL;
    public $current_multithread_id = 0;
    public $jobid = 0;

    public static function Singleton() {
        if( !self::$instance ) {            
            self::$instance = new Multithread_API();
        }

        return self::$instance;
    } 

    public function __construct() {        
        $this->db = IEM::getDatabase();
    }

    public function log($data, $title=false) {        
        if( IEM_CLI_MODE ) {            
            $title = $title ? $title : "Entry " . time();
            $title = "DEBUG=>" . $title . ": ";
            if( is_bool($data) ) {                
                $data = $data ? "TRUE" : "FALSE";
                echo PHP_EOL . $title . print_r($data, true);
                return NULL;
            }

            echo PHP_EOL . $title . print_r($data, true);
        }

    }

    public function fix_thread_crashes() {        
        $this->log("[FIX THREAD CRASHES]");
        $timenow = $this->GetServerTime();
        $half_hour_ago = $timenow - 2 * 60;
        //$half_hour_ago = $timenow - 2 * 60; // why? ))
        $result = $this->db->Query("SELECT * FROM " . SENDSTUDIO_TABLEPREFIX . "jobs WHERE jobtype='send' AND jobstatus='p'");
        if( $result ) {            
            while( $row = $this->db->Fetch($result) ) {                
                $sq = $this->db->Query("UPDATE " . SENDSTUDIO_TABLEPREFIX . "addon_multithread_slots SET lastupdatetime='0' WHERE jobid=" . $row["jobid"]);
                $this->log($this->db->NumAffected($this->db->NumAffected()), "Update Thread");
                $sq = $this->db->Query("UPDATE " . SENDSTUDIO_TABLEPREFIX . "jobs SET multithread_full=0 WHERE jobid=" . $row["jobid"]);
                $this->log("jobid:" . $row["jobid"] . ", affected:" . $this->db->NumAffected($this->db->NumAffected()), "Update Job");
            }
        }

        $result = $this->db->Query("SELECT a.uid, a.jobid,a.lastupdatetime,b.lastupdatetime as lastupdate_in_job,b.multithread_full FROM " . SENDSTUDIO_TABLEPREFIX . "addon_multithread_slots a LEFT JOIN " . SENDSTUDIO_TABLEPREFIX . "jobs b ON a.jobid=b.jobid WHERE a.lastupdatetime<" . $half_hour_ago . " AND b.multithread_full=1");
        if( $result ) {            
            while( $row = $this->db->Fetch($result) ) {                
                $sq = $this->db->Query("UPDATE " . SENDSTUDIO_TABLEPREFIX . "addon_multithread_slots SET lastupdatetime='0' WHERE uid=" . $row["uid"]);
                $this->log("uid:" . $row["uid"] . ", affected:" . $this->db->NumAffected(), "Update Thread");
                $sq = $this->db->Query("UPDATE " . SENDSTUDIO_TABLEPREFIX . "jobs SET multithread_full=0 WHERE jobid=" . $row["jobid"]);
                $this->log("jobid:" . $row["jobid"] . ", affected:" . $this->db->NumAffected(), "Update Job");                
            }
        }

    }

    public function fix_job_unfinished() {        
        $this->log("[FIX JOB UNFINISHED]");
    }

    public function status() {        
        $this->log("[SYSTEM STATUS]");
        $query = "SELECT b.jobid FROM " . SENDSTUDIO_TABLEPREFIX . "addon_multithread_slots a LEFT JOIN " . SENDSTUDIO_TABLEPREFIX . "jobs b ON a.jobid=b.jobid WHERE (b.jobstatus='w' OR b.jobstatus='i') AND b.jobtype='send'  GROUP BY b.jobid";
        $result = $this->db->Query($query);
        if( $result ) {            
            $jobs = array();
            while( $row = $this->db->Fetch($result) ) {                
                $jobs[$row["jobid"]] = array();
            }

            foreach( $jobs as $i => $v ) {                
                $jobs[$i]["threads"] = array();
                $timenow = $this->GetServerTime();
                $tj = $this->db->Query("SELECT * FROM " . SENDSTUDIO_TABLEPREFIX . "addon_multithread_slots WHERE jobid=" . $i);
                while( $a = $this->db->Fetch($tj) ) {                    
                    $iq = (int) $this->count_queue_by_thread($a["uid"]);
                    $it = (int) $a["emails"];
                    $ts = $it - $iq;
                    $percent = ceil($ts * 100 / $it);
                    $passed = $timenow - (int) ($a["lastupdatetime"]);
                    $ac = MULTITHREAD_LIFETIME < $passed ? "INACTIVE" : "ACTIVE";
                    $jobs[$i]["threads"][] = $a["uid"] . "[" . $ac . "][" . $passed . "s][" . $ts . "/" . $it . "]" . $percent . "%";
                }
            }

            $this->log($jobs, "Status");
            return NULL;
        }

        $this->log("We dont have threads", "Status");
    }

    public function count_queue_by_thread($thread_id) {        
        $result = $this->db->Fetch($this->db->Query("SELECT COUNT(*) as total FROM " . SENDSTUDIO_TABLEPREFIX . "queues WHERE multithread_id=" . $thread_id));
        return (int) $result["total"];
    }

    public function switch_thread($query) {        
        $query .= " AND multithread_id=" . $this->current_multithread_id . " ";
        $this->log($query, "Requesting subscribers");
    }

    public function tool_queue_info($jobid) {        
        $this->log("[QUEUE INFO]");
        $result = $this->db->Fetch($this->db->Query("SELECT * FROM " . SENDSTUDIO_TABLEPREFIX . "jobs WHERE jobid=" . $jobid . " LIMIT 1"));
        if( $result ) {            
            $this->log($result, "Job found");
            $job_details = unserialize($result["jobdetails"]);
            $this->log($job_details, "Job details");
            $slots = array();
            $sq = $this->db->Query("SELECT * FROM " . SENDSTUDIO_TABLEPREFIX . "addon_multithread_slots WHERE jobid=" . $jobid);
            while( $row = $this->db->Fetch($sq) ) {                
                $timenow = $this->GetServerTime();
                $diff = $timenow - (int) ($row["lastupdatetime"]);
                $status = $diff <= MULTITHREAD_LIFETIME ? "ACTIVE" : "INACTIVE";
                $slots[] = "[" . $row["uid"] . ":" . $row["emails"] . "|" . $row["sent"] . "|" . $row["failed"] . "|" . $diff . "s ago][" . $status . "]";
            }

            $this->log($slots, "Multithread slots");
            if( 0 < (int) ($result["queueid"]) ) {                
                $qs = array();
                $result = $this->db->Query("SELECT COUNt(*) as total, multithread_id FROM " . SENDSTUDIO_TABLEPREFIX . "queues WHERE queueid=" . $result["queueid"] . " AND processed=0 GROUP BY multithread_id");
                while( $row = $this->db->Fetch($result) ) {                    
                    $qs[] = "[uid->" . $row["multithread_id"] . ":" . $row["total"] . "]";
                }
                $this->log($qs, "Emails in queue");
            }
            else {                
                $this->log(0, "Emails in queue");
            }

            $result = $this->db->Fetch($this->db->Query("SELECT * FROM " . SENDSTUDIO_TABLEPREFIX . "addon_multithread_lock WHERE jobid=" . $jobid . " LIMIT 1"));
            $this->log($result, "Multithread lock");
            return NULL;
        }

        $this->log("Job doesnt exist", "Status");
    }

    public function tool_queue_rebuild($jobid) {        
        $this->log("[QUEUE REBUILD]");
        $result = $this->db->Fetch($this->db->Query("SELECT * FROM " . SENDSTUDIO_TABLEPREFIX . "jobs WHERE jobid=" . $jobid . " LIMIT 1"));
        if( $result ) {            
            $sq = $this->db->Query("DELETE FROM " . SENDSTUDIO_TABLEPREFIX . "addon_multithread_slots WHERE jobid=" . $jobid);
            $this->log($this->db->NumAffected(), "Multithread slots deleted");
            $sq = $this->db->Query("DELETE FROM " . SENDSTUDIO_TABLEPREFIX . "addon_multithread_lock WHERE jobid=" . $jobid);
            $this->log($this->db->NumAffected(), "Multithread lock deleted");
            $sq = $this->db->Query("DELETE FROM " . SENDSTUDIO_TABLEPREFIX . "queues WHERE queueid=" . $result["queueid"]);
            $this->log($this->db->NumAffected(), "Job Queue deleted");
            $sq = $this->db->Query("UPDATE " . SENDSTUDIO_TABLEPREFIX . "jobs SET jobstatus='w', lastupdatetime='0', queueid='0' WHERE jobid=" . $jobid . " LIMIT 1");
            $this->log($this->db->NumAffected(), "Job updated");
            return NULL;
        }

        $this->log("Job doesnt exist", "Status");
    }

    public function tool_job_free_threads($jobid) {        
        $this->log("[QUEUE REBUILD]");
        $result = $this->db->Fetch($this->db->Query("SELECT * FROM " . SENDSTUDIO_TABLEPREFIX . "jobs WHERE jobid=" . $jobid . " LIMIT 1"));
        if( $result ) {            
            $sq = $this->db->Query("UPDATE " . SENDSTUDIO_TABLEPREFIX . "addon_multithread_slots SET lastupdatetime='0' WHERE jobid=" . $jobid);
            $this->log($this->db->NumAffected(), "Thread items updated");
            return NULL;
        }

        $this->log("Job doesnt exist", "Status");
    }

    public function tool_job_rebuild_threads($jobid, $threads, $emails) {        
        $this->log("[JOB THREADS REBUILD]");
        $result = $this->db->Fetch($this->db->Query("SELECT * FROM " . SENDSTUDIO_TABLEPREFIX . "jobs WHERE jobid=" . $jobid . " LIMIT 1"));
        if( $result ) {            
            $this->tool_queue_info($jobid);
            $job_details = unserialize($result["jobdetails"]);
            $this->log($job_details, "Job details");
            $sq = $this->db->Query("UPDATE " . SENDSTUDIO_TABLEPREFIX . "jobs SET jobstatus='p', lastupdatetime='0' WHERE jobid=" . $jobid . " LIMIT 1");
            $this->log($this->db->NumAffected(), "Pausing Job");
            sleep(5);
            $this->log("Done", "Sleep to pause the job correctly");
            $sq = $this->db->Query("DELETE FROM " . SENDSTUDIO_TABLEPREFIX . "addon_multithread_slots WHERE jobid=" . $jobid);
            $this->log($this->db->NumAffected(), "Multithread slots deleted");
            $sq = $this->db->Query("UPDATE " . SENDSTUDIO_TABLEPREFIX . "queues SET multithread_id=0 WHERE queueid=" . $result["queueid"]);
            $this->log($this->db->NumAffected(), "Queue set to 0");
            $job_details["multithread"] = $threads;
            $this->split_queue($job_details, $jobid, array( "queueid" => $result["queueid"] ), $emails);
            $sq = $this->db->Query("UPDATE " . SENDSTUDIO_TABLEPREFIX . "jobs SET jobstatus='w', lastupdatetime='0' WHERE jobid=" . $jobid . " LIMIT 1");
            $this->log($this->db->NumAffected(), "Resuming Job");
            return NULL;
        }

        $this->log("Job doesnt exist", "Status");
    }

    public function split_queue($job_details, $jobid=0, $queueinfo=array(), $emails_per_split=MULTITHREAD_MINIMUM_SIZE) {        
        $threads = (int) $job_details["multithread"];
        if( 0 < $threads ) {            
            $GLOBALS["addon_multithread_running"] = true;
            $GLOBALS["addon_multithread_class"] = $this;
        }
        else {            
            return false;
        }

        $this->log($threads, "Threads selected by user");
        $thread_ids = array();
        $i = 0;
        while( $i <= $threads ) {            
            $result = $this->db->InsertQuery("addon_multithread_slots", array( "jobid" => $jobid ));
            $thread_ids[] = $result;
            ++$i;
        }

        $populate = true;
        if( $emails_per_split == MULTITHREAD_MINIMUM_SIZE ) {            
            $batch_size = $emails_per_split;
            $percent_to_split = 2;
            $ss = (int) ($job_details["SendSize"]);
            if( $ss < 10000 ) {                
                $batch_size = 200;
            }
            else {                
                if( $ss < 600000 ) {                    
                    $batch_size = 500;
                }
                else {                    
                    $batch_size = 5000;
                }
            }

        }
        else {            
            $batch_size = $emails_per_split;
        }

        $this->log($batch_size, "Emails balance in thread");
        $this->log($thread_ids, "Threads per default by system");
        $current_thread_in_split = 0;
        while( $populate == true ) {            
            if( count($thread_ids) - 1 < $current_thread_in_split ) {                
                $current_thread_in_split = 0;
            }
            $thread_id = $thread_ids[$current_thread_in_split];
            ++$current_thread_in_split;
            $this->log($thread_id . " with " . $batch_size . " emails", "Populating thread");
            $affected = $this->db->NumAffected($this->db->Query("UPDATE " . SENDSTUDIO_TABLEPREFIX . "queues SET multithread_id='" . $thread_id . "' WHERE queueid='" . $queueinfo["queueid"] . "' AND multithread_id=0 LIMIT " . $batch_size));
            if( $affected == 0 ) {                
                $populate = false;
            }
            else {                
                $this->db->Query("UPDATE " . SENDSTUDIO_TABLEPREFIX . "addon_multithread_slots SET emails=emails+" . $affected . " WHERE uid='" . $thread_id . "'");
                $populate = true;
            }

        }

        $this->db->Query("DELETE FROM " . SENDSTUDIO_TABLEPREFIX . "addon_multithread_slots WHERE jobid=" . $jobid . " AND emails=0");
        $this->log($this->db->NumAffected(), "Deleted unusable threads");
        $created_threads = $this->db->Fetch($this->db->Query("SELECT COUNT(*) as total FROM " . SENDSTUDIO_TABLEPREFIX . "addon_multithread_slots WHERE jobid=" . $jobid));
        $this->db->Query("UPDATE " . SENDSTUDIO_TABLEPREFIX . "jobs SET multithread=" . $created_threads["total"] . " WHERE jobid='" . $jobid . "'");
        $this->log("Job [" . $jobid . "] updated with [" . $created_threads["total"] . "] threads", "Split queues");
    }

    public function check_thread_finished($jobid, $queueid) {
        $query = "SELECT recipient FROM " . SENDSTUDIO_TABLEPREFIX . "queues WHERE processed=0 AND queueid=" . $queueid . " AND multithread_id=" . $this->current_multithread_id . " LIMIT 1";
        $thread_result = $this->db->Fetch($this->db->Query($query));
        if( !$thread_result ) {            
            $this->log(!$thread_result ? "We dont have mails in queue for thread {" . $this->current_multithread_id . "}" : "We have queue for thread {" . $this->current_multithread_id . "}", "Check thread finished");
            $this->db->Query("DELETE FROM " . SENDSTUDIO_TABLEPREFIX . "addon_multithread_slots WHERE uid=" . $this->current_multithread_id);
            $this->log($this->db->NumAffected(), "Deleting thread [" . $this->current_multithread_id . "] in multithread slots");
            $this->db->Query("DELETE FROM " . SENDSTUDIO_TABLEPREFIX . "queues WHERE multithread_id=" . $this->current_multithread_id);
            $this->log($this->db->NumAffected(), "Deleting queues with [" . $this->current_multithread_id . "] and processed = 1");
            $query = "SELECT * FROM " . SENDSTUDIO_TABLEPREFIX . "jobs WHERE jobid=" . $jobid;
            $result = $this->db->Fetch($this->db->Query($query));
            if( $result ) {                
                $job_details = unserialize($result["jobdetails"]);
                $job_details["EmailResults"]["success"] = (int) ($result["multithread_sent"]);
                $job_details["EmailResults"]["failure"] = (int) ($result["multithread_failed"]);
                $query = "UPDATE [|PREFIX|]jobs SET multithread_sent=multithread_sent+" . $GLOBALS["multithread_sent"] . ", multithread_failed=multithread_failed+" . $GLOBALS["multithread_failed"] . ", jobdetails='" . $this->db->Quote(serialize($job_details)) . "' WHERE jobid=" . $jobid;
                $this->log($query, "Updating JobDetails");
                $this->log($job_details, "Updating the jobs mail info at last time");
                $result = $this->db->Query($query);
            }

        }

        $result = $this->db->Fetch($this->db->Query("SELECT recipient FROM " . SENDSTUDIO_TABLEPREFIX . "queues WHERE queueid=" . $queueid . " LIMIT 1"));
        if( !$result ) {            
            return true;
        }

        return false;
    }

    public function orphan_job($jobid, $queueid) {        
        $row = $this->db->Fetch($this->db->Query("SELECT * FROM " . SENDSTUDIO_TABLEPREFIX . "jobs WHERE jobid=" . $jobid));
        if( $row ) {            
            if( 0 < $queueid ) {                
                $q = $this->db->Fetch($this->db->Query("SELECT COUNT(*) as total FROM " . SENDSTUDIO_TABLEPREFIX . "queues WHERE queueid=" . $queueid));
                if( (int) ($q["total"]) == 0 ) {                
                    if( (int) ($row["jobtime"]) < (int) ($row["lastupdatetime"]) ) {                        
                        return true;
                    }
                }
            }
        }

        return false;
    }

    public function get_thread($jobid) {        
        $this->jobid = $jobid;
        $timenow = $this->GetServerTime();
        $half_hour_ago = $timenow - MULTITHREAD_LIFETIME;
        $data = array();
        $row = $this->db->Fetch($this->db->Query("SELECT uid FROM " . SENDSTUDIO_TABLEPREFIX . "addon_multithread_slots WHERE jobid=" . $jobid . " AND lastupdatetime < " . $half_hour_ago . " LIMIT 1"));
        if( !$row ) {            
            $this->db->Query("UPDATE " . SENDSTUDIO_TABLEPREFIX . "jobs SET multithread_full=1 WHERE jobid='" . $jobid . "'");
            $this->log("(The system will ignore the job [" . $jobid . "] until is completed", "Mark Job as Full Slots");
            $wht = $this->db->Fetch($this->db->Query("SELECT uid FROM " . SENDSTUDIO_TABLEPREFIX . "addon_multithread_slots WHERE jobid=" . $jobid . " LIMIT 1"));
            if( $wht ) {                
                $this->log("I will kill you CRON, there is no threads for you", "Threads-info");
                exit ();
            }
            else {                
                $this->log("We dont have threads in threads table, lets IEM handle the process", "Threads-info");
                $data["finish_job"] = true;
                $data["thread_id"] = false;
                $query = "SELECT * FROM " . SENDSTUDIO_TABLEPREFIX . "jobs WHERE jobid=" . $jobid;
                $result = $this->db->Fetch($this->db->Query($query));
                if( $result ) {                    
                    $job_details = unserialize($result["jobdetails"]);
                    $job_details["EmailResults"]["success"] = (int) ($result["multithread_sent"]);
                    $job_details["EmailResults"]["failure"] = (int) ($result["multithread_failed"]);
                    $query = "UPDATE [|PREFIX|]jobs SET multithread_sent=multithread_sent+" . $GLOBALS["multithread_sent"] . ", multithread_failed=multithread_failed+" . $GLOBALS["multithread_failed"] . ", jobdetails='" . $this->db->Quote(serialize($job_details)) . "' WHERE jobid=" . $jobid;
                    $this->log($job_details, "Updating the jobs mail info at last time");
                    $result = $this->db->Query($query);
                }
            }

        }
        else {            
            $this->current_multithread_id = (int) ($row["uid"]);
            $this->log($row, "Thread-info");
            $this->update_thread(NULL);
            $data["finish_job"] = false;
            $data["thread_id"] = $this->current_multithread_id;
            $this->cron_shoot();
        }

        return $data;
    }

    public function cron_shoot() {        
        if( MULTITHREAD_ALL_IN_ONE == true ) {            
            $options = CommandLine::parseArgs($_SERVER["argv"]);
            if( isset( $options["multithread-shot"] ) ) {                
                return false;
            }

            $timenow = $this->GetServerTime();
            $half_hour_ago = $timenow - MULTITHREAD_LIFETIME;
            $threads = $this->db->Query("SELECT * FROM " . SENDSTUDIO_TABLEPREFIX . "addon_multithread_slots WHERE jobid=" . $this->jobid . " AND lastupdatetime < " . $half_hour_ago);
            $crons = 0;
            $cronsuid = array();
            while( $row = $this->db->Fetch($threads) ) {                
                ++$crons;
                $cronsuid[$crons - 1] = $row["uid"];
            }

            $crons = MULTITHREAD_THREADS_START == 0 ? $crons : MULTITHREAD_THREADS_START;
            $i = 0;
            while( $i <= $crons ) {                
                $cron_path = MULTITHREAD_PHP_PATH . " " . dirname(dirname(dirname(dirname("")))) . "/cron/cron.php --multithread-shot=" . $cronsuid[$i] . " > /dev/null 2>&1 &";
                $this->log($cron_path, "Cron Thread");
                exec($cron_path);
                sleep(2);
                ++$i;                
            }

        }

    }

    public function update_thread($db=NULL, $thread=NULL) {        
        $db = $db ? $db : $this->db;
        $timenow = $this->GetServerTime();
        $thread = $thread ? $thread : $this->current_multithread_id;
        $db->Query("UPDATE " . SENDSTUDIO_TABLEPREFIX . "addon_multithread_slots SET lastupdatetime='" . $timenow . "' WHERE uid='" . $thread . "'");
    }

    public function update_application($_this, $jobid, $jobdetails, $statid, $emails_sent=0) {        
        if( !isset( $GLOBALS["mt_emails_sent"] ) ) {            
            $GLOBALS["mt_emails_sent"] = 0;
        }

        $GLOBALS["mt_emails_sent"] = $GLOBALS["mt_emails_sent"] + $emails_sent;
        if( $emails_sent == 0 || $GLOBALS["mt_emails_sent"] % MULTITHREAD_COUNTER_UPDATE_JOB_THREAD == 0 ) {            
            $this->update_thread($_this->Db);
        }

        if( $emails_sent == 0 || $GLOBALS["mt_emails_sent"] % MULTITHREAD_COUNTER_UPDATE_JOB_ALIVE == 0 ) {            
            $query = "UPDATE " . SENDSTUDIO_TABLEPREFIX . "jobs SET lastupdatetime='" . $this->GetServerTime() . "' WHERE jobid='" . (int) ($jobid) . "'";
            $_this->Db->Query($query);
        }

        if( $emails_sent == 0 || $GLOBALS["mt_emails_sent"] % MULTITHREAD_COUNTER_UPDATE_STATS_NEWSLETTERS == 0 ) {            
            $query = "UPDATE " . SENDSTUDIO_TABLEPREFIX . "stats_newsletters SET multipartrecipients=multipartrecipients + " . $GLOBALS["recipient_m"] . ", htmlrecipients=htmlrecipients + " . $GLOBALS["recipient_h"] . ", textrecipients=textrecipients + " . $GLOBALS["recipient_t"] . " WHERE statid='" . $statid . "'";
            $_this->Db->Query($query);
            $GLOBALS["recipient_m"] = 0;
            $GLOBALS["recipient_h"] = 0;
            $GLOBALS["recipient_t"] = 0;
        }

        if( $emails_sent == 0 || $GLOBALS["mt_emails_sent"] % MULTITHREAD_COUNTER_UPDATE_JOB_DETAILS == 0 ) {            
            $query = "SELECT multithread_sent,multithread_failed FROM " . SENDSTUDIO_TABLEPREFIX . "jobs WHERE jobid=" . $jobid;
            $result = $_this->Db->Fetch($_this->Db->Query($query));
            if( $result ) {                
                $job_details = $jobdetails;
                $job_details["EmailResults"]["success"] = (int) ($result["multithread_sent"]);
                $job_details["EmailResults"]["failure"] = (int) ($result["multithread_failed"]);
                $query = "UPDATE [|PREFIX|]jobs SET multithread_sent=multithread_sent+" . $GLOBALS["multithread_sent"] . ", multithread_failed=multithread_failed+" . $GLOBALS["multithread_failed"] . ", jobdetails='" . $_this->Db->Quote(serialize($job_details)) . "' WHERE jobid=" . $jobid;
                $this->log($GLOBALS["multithread_sent"], "Updating with");
                $result = $_this->Db->Query($query);
                $GLOBALS["multithread_sent"] = 0;
                $GLOBALS["multithread_failed"] = 0;
            }

        }

    }

    public function locked_creating_thread($jobid) {        
        $result = $this->db->Fetch($this->db->Query("SELECT * FROM " . SENDSTUDIO_TABLEPREFIX . "addon_multithread_lock WHERE jobid=" . $jobid . " AND finished=0 LIMIT 1"));
        return $result ? true : false;
    }

    public function lock_creating_thread($jobid) {        
        $timenow = $this->GetServerTime();
        $result = $this->db->InsertQuery("addon_multithread_lock", array( "jobid" => $jobid , "started" => $timenow ));
    }

    public function unlock_creating_thread($jobid) {        
        $timenow = $this->GetServerTime();
        $this->db->Query("UPDATE " . SENDSTUDIO_TABLEPREFIX . "addon_multithread_lock SET finished='" . $timenow . "' WHERE jobid='" . $jobid . "'");
    }

    public function fetch_job($_this) {        
        $result = $this->db->Query("SELECT multithread_full FROM " . SENDSTUDIO_TABLEPREFIX . "jobs LIMIT 1");
        if( !$result ) {            
            $result = $this->db->Query("ALTER TABLE `" . SENDSTUDIO_TABLEPREFIX . "jobs` ADD `multithread_full` INT( 11 ) NOT NULL DEFAULT '0'");
        }

        $result = $this->db->Query("SELECT multithread_sent FROM " . SENDSTUDIO_TABLEPREFIX . "jobs LIMIT 1");
        if( !$result ) {            
            $result = $this->db->Query("ALTER TABLE `" . SENDSTUDIO_TABLEPREFIX . "jobs` ADD `multithread_sent` INT( 11 ) NOT NULL DEFAULT '0'");
        }

        $result = $this->db->Query("SELECT multithread_failed FROM " . SENDSTUDIO_TABLEPREFIX . "jobs LIMIT 1");
        if( !$result ) {            
            $result = $this->db->Query("ALTER TABLE `" . SENDSTUDIO_TABLEPREFIX . "jobs` ADD `multithread_failed` INT( 11 ) NOT NULL DEFAULT '0'");
        }

        $GLOBALS["multithread_sent"] = 0;
        $GLOBALS["multithread_failed"] = 0;
        $GLOBALS["recipient_m"] = 0;
        $GLOBALS["recipient_h"] = 0;
        $GLOBALS["recipient_t"] = 0;
        $this->fix_thread_crashes();
        $this->current_thread = 0;
        $timenow = $_this->GetServerTime();
        $half_hour_ago = $timenow - MULTITHREAD_COUNTER_CRASHED_JOB * 60;
        $query = "SELECT * FROM " . SENDSTUDIO_TABLEPREFIX . "jobs WHERE jobtype='send' AND (";
        $query .= " (jobstatus ='w' AND jobtime < " . $timenow . ") OR ";
        $query .= " (jobstatus='r' AND jobtime < " . $timenow . ") OR ";
        $query .= " (jobstatus='i' AND jobtime < " . $timenow . " AND lastupdatetime < " . $half_hour_ago . ")";
        $query .= ") AND (approved > 0)";
        $query .= " OR (multithread>0 AND jobtype='send' AND multithread_full=0 AND (jobstatus ='w' OR jobstatus ='i' OR jobstatus ='r')) ";
        $query .= " ORDER BY jobtime ASC LIMIT 1";
        echo "Get new Job to work: " . $query . PHP_EOL;
        $result = $_this->Db->Query($query);
        if( !$result ) {            
            return false;
        }

        $row = $_this->Db->Fetch($result);
        if( empty( $row ) ) {            
            return false;
        }

        if( 0 < (int) ($row["multithread"]) ) {            
            $GLOBALS["addon_multithread_class"] = $this;
        }

        $query = "SELECT COUNT(*) AS count FROM " . SENDSTUDIO_TABLEPREFIX . "jobs_lists WHERE jobid='" . $row["jobid"] . "'";
        $result = $_this->Db->Query($query);
        $count_check = $_this->Db->FetchOne($result, "count");
        if( $count_check <= 0 ) {            
            echo "*** Found an orphaned job: " . $row["jobid"] . " *** - No entries in the jobs_lists table. Please contact your administrator." . "\n";
            $_this->PauseJob($row["jobid"]);
            return false;
        }

        $_this->jobstatus = $row["jobstatus"];
        $_this->jobdetails = unserialize($row["jobdetails"]);
        $_this->jobowner = $row["ownerid"];
        if( isset( $_this->jobdetails["multithread"] ) ) {
            //
        }

        if( $row["jobstatus"] == "i" ) {
            //
        }

        if( $row["jobstatus"] == "r" ) {
            
            $this->db->StartTransaction();
            $start_ok = $_this->StartJob($row["jobid"]);
            $this->log($start_ok, "Job started");
            if( !$start_ok ) {                
                $this->db->RollbackTransaction();
                return false;
            }

            $resend_ok = $_this->ResendJob_Setup($row["jobid"]);
            $this->log($resend_ok, "Resend Job Setup");
            if( !$resend_ok ) {                
                $this->db->RollbackTransaction();
                return false;
            }

            $this->db->CommitTransaction();
            $lock = $this->lock_creating_thread($row["jobid"]);
            $this->log($lock, "Locking job");
            $queueinfo = array( "queueid" => $row["queueid"] , "queuetype" => "send" , "ownerid" => $row["ownerid"] );
            $this->split_queue($_this->jobdetails, $row["jobid"], $queueinfo);
            $this->unlock_creating_thread($row["jobid"]);
        }

        return $row["jobid"];
    }

    public function process_job($_this, $jobid=0) {
        
        $this->log("Working with job=> " . $jobid);
        if( !$_this->StartJob($jobid) ) {
            //
        }

        $user = GetUser($_this->jobowner);
        IEM::userLogin($_this->jobowner, false);
        $queueid = false;
        if( !($queueid = $_this->GetJobQueue($jobid)) ) {            
            if( $this->locked_creating_thread($jobid) ) {                
                $this->log("Job locked, we are creating the queue, die CRON");
                exit;
            }

            $this->lock_creating_thread($jobid);
            $sendqueue = $_this->CreateQueue("send");
            $queueok = $_this->JobQueue($jobid, $sendqueue);
            $send_criteria = $_this->jobdetails["SendCriteria"];
            $original_queuesize = $_this->jobdetails["SendSize"];
            $queueinfo = array( "queueid" => $sendqueue , "queuetype" => "send" , "ownerid" => $_this->jobowner );
            if( isset( $_this->jobdetails["Segments"] ) && is_array($_this->jobdetails["Segments"]) ) {                
                $_this->Subscriber_API->GetSubscribersFromSegment($_this->jobdetails["Segments"], false, $queueinfo, "nosort");
            }
            else {                
                $_this->Subscriber_API->GetSubscribers($send_criteria, array(), false, $queueinfo, $sendqueue);
            }

            $this->log("Removing duplicates");
            $_this->Subscriber_API->RemoveDuplicatesInQueue($sendqueue, "send", $_this->jobdetails["Lists"]);
            $this->log("Removed Banned Emails");
            $_this->Subscriber_API->RemoveBannedEmails($_this->jobdetails["Lists"], $sendqueue, "send");
            $this->log("Remev Unsubscribed Emails");
            $_this->Subscriber_API->RemoveUnsubscribedEmails($_this->jobdetails["Lists"], $sendqueue, "send");
            $this->log("Splitting Queues in Threads");
            $this->split_queue($_this->jobdetails, $jobid, $queueinfo);
            $this->log("Unlocking Job");
            $this->unlock_creating_thread($jobid);
            $queueid = $sendqueue;
            $newsletterstats = $_this->jobdetails;
            $newsletterstats["Job"] = $jobid;
            $newsletterstats["Queue"] = $sendqueue;
            $newsletterstats["SentBy"] = $queueinfo["ownerid"];
            $real_queuesize = $_this->Subscriber_API->QueueSize($queueid, "send");
            $newsletterstats["SendSize"] = $real_queuesize;
            $this->log("Saving Newsletter Stats");
            $statid = $_this->Stats_API->SaveNewsletterStats($newsletterstats);
            //$tempAPIFile = dirname("") . "/../../../functions/api/module_trackerfactory.php"; // LOL
            $tempAPIFile = dirname(__FILE__) . "/../../../functions/api/module_trackerfactory.php";
            if( is_file($tempAPIFile) ) {                
                $this->log($tempAPIFile, "Inside Tracking Factory");
                require_once $tempAPIFile;
                $temp = array_merge($_this->jobdetails, array( "statid" => $statid , "stattype" => "newsletter" , "newsletterid" => $_this->jobdetails["Newsletter"] ));
                $status = module_Tracker::ParseOptionsForAllTracker($temp);
            }

            $_this->Stats_API->UpdateUserStats($queueinfo["ownerid"], $jobid, $statid);
            $check_stats = $_this->Stats_API->ReCheckUserStats($user, $original_queuesize, $real_queuesize, AdjustTime());
            $not_ok_to_send_reason = $check_stats[1];
            $ok_to_send = $check_stats[0];
            if( !$ok_to_send ) {
                
                trigger_error("Multithread_API" . "::" . "Multithread_API::process_job" . " -- " . GetLang($not_ok_to_send_reason), 512);
                $_this->PauseJob($jobid);
                $_this->UnapproveJob($jobid);
                IEM::userLogout();
                return false;
            }

            API_USERS::creditEvaluateWarnings($user->GetNewAPI());
        }

        $_this->statid = $_this->LoadStats($jobid);
        if( empty( $_this->statid) ) {            
            echo "Multithread_API" . "::" . "Multithread_API::process_job" . " -- Cannot find statid. Previous preliminary job process did not get finalized (either CRON died, or it hasn't finished processing the job). Ignoring this job: jobid " . $jobid . "." . PHP_EOL;
            trigger_error("Multithread_API" . "::" . "Multithread_API::process_job" . " -- Cannot find statid. Previous preliminary job process did not get finalized (either CRON died, or it hasn't finished processing the job). Ignoring this job: jobid " . $jobid . ".", 1024);
            IEM::userLogout();
            return false;
        }

        $_this->queuesize = $_this->jobdetails["SendSize"];
        $finished = $_this->ActionJob($jobid, $queueid);
        if( $finished ) {
            
            $_this->jobstatus = "c";
            $_this->FinishJob($jobid);
        }

        IEM::userLogout();
        return true;
    }

    public function iem_show_threads_in_email_campaign() {        
        $GLOBALS["THREAD_SEND_SELECT_THREAD"] = $this->template_system->ParseTemplate("iem_thread_send_select_thread", true);
    }

    public function iem_email_campaign_save_thread($send_details, $data) {        
        $multithread = isset( $data["multithread"] ) && 0 < (int) ($data["multithread"]) ? (int) ($data["multithread"]) : 0;
        $send_details["multithread"] = $multithread;
        $this->template_system->Assign("multithread", $multithread);
        $GLOBALS["THREAD_SEND_SELECT_THREAD_CONFIRM"] = $this->template_system->ParseTemplate("iem_multithread_send_select_multithread_confirm", true);
    }

}

?>