FreeSTAR-Status-Engine/classes/queue.php

<?php

/**
 * Class for creating and managing the queue system
 */
class Queue
{
    public $task_id;
    public $type_id;
    public $status;
    public $template_data1;    // i.e. Subject for email
    public $template_data2;    // i.e. HTML email body
    public $create_time;
    public $completed_time;
    public $num_errors;
    public $user_id;
    public $all_type_id = array('notify_telegram' => 1,
                                'notify_email'    => 2);

    public $all_status = array('populating' => 1,
                               'ready'      => 2,
                               'processing' => 3,
                               'completed'  => 4,
                               'failed'     => 5);

    public function add_task() {
        global $mysqli;
        $stmt = $mysqli->prepare("INSERT INTO queue_task (type_id, status, template_data1, template_data2, created_time, user_id) VALUES (?,?,?,?,?,?)");
        if ( false===$stmt ) {
            //die('prepare() failed: ' . htmlspecialchars($mysqli->error));
            echo $mysqli->errno();

        }
        #if ( false === $stmt ) { syslog(1, "Error :". $mysqli->error); }
        $now = time();
        $res = $stmt->bind_param("iissii", $this->type_id, $this->status, $this->template_data1, $this->template_data2, $now, $this->user_id);
        if ( false === $res ) {
        echo "error";
        die();
        }
        $stmt->execute();
        $query = $stmt->get_result();
        print $query;
        $this->task_id = $mysqli->insert_id;
        return $this->task_id;
    }

    /**
     * Remove task from the queue
     * @return void
     */
    public function delete_task($task_id){
        global $mysqli;
        $stmt = $mysqli->prepare("DELETE FROM queue_task WHERE id = ?");
        $stmt->bind_param("i", $task_id);
        $stmt->execute();
    }

    /**
     * Update status for given task
     * @param int $new_status The new current status of the task. Must be selected from the $all_status array.
     * @return void
     */
    public function set_task_status($new_status) {
        global $mysqli;
        $stmt = $mysqli->prepare("UPDATE queue_task SET status = ? WHERE id = ?");
        $stmt->bind_param("ii", $new_status, $this->task_id);
        $stmt->execute();
        $this->status = $new_status;
    }

    /**
     * Add notification queue data for given task
     * @param array $arr_data Array filled with subscriber_id
     * @return void
     */
    public function add_notification($arr_data) {
        global $mysqli;

        //Default status = 1, retres = 0, task_id = $this->task_id

        // Build query manually since mysqli doesn't cater well for multi insert..
        $count     = count($arr_data);  // Let's find number of elements
        $counter   = 0;
        $query     = '';
        $seperator = ',';
        $sub_query = '(%d, %d, %d ,%d)%s';

        foreach ($arr_data as $value) {
            $counter++;
            if ($counter == $count)  { $seperator = ''; }  // Make sure last character for SQL query is correct
            $query .= sprintf($sub_query, $this->task_id, 1, $value, 0, $seperator);
        }
        $sql = "INSERT INTO queue_notify (task_id, status, subscriber_id, retries) VALUES ". $query;

        $mysqli->query($sql);

        $this->set_task_status($this->all_status['ready']); // Make task available for release
    }

    public function update_notification_retries($task_id, $subscriber_id) {
        global $mysqli;
        $stmt = $mysqli->prepare("UPDATE queue_notify SET retries = retries+1 WHERE task_id = ? AND subscriber_id = ?");
        $stmt->bind_param("ii", $task_id, $subscriber_id);
        $stmt->execute();
    }

    public function delete_notification($task_id, $subscriber_id) {
        global $mysqli;
        $stmt = $mysqli->prepare("DELETE FROM queue_notify WHERE task_id = ? AND subscriber_id = ?");
        $stmt->bind_param("ii", $task_id, $subscriber_id);
        $stmt->execute();

    }

    // TODO: Fix max attempts for notifications
    public function process_queue(){
      global $mysqli;
      $stmt = $mysqli->query("SELECT qn.id, qn.task_id, qn.status, qn.subscriber_id, qn.retries, sub.firstname, sub.userID, sub.token FROM queue_notify AS qn INNER JOIN subscribers AS sub ON qn.subscriber_id = sub.subscriberID WHERE qn.status NOT LIKE 2 AND sub.active=1");
      while ( $result = $stmt->fetch_assoc() ) {

          $i = 2;
          $stmt2 = $mysqli->prepare("SELECT * FROM queue_task WHERE id = ? AND status = ?");
          $stmt2->bind_param("ii", $result['task_id'], $i);
          $stmt2->execute();
          $tmp = $stmt2->get_result();
          $result2 = $tmp->fetch_assoc();
          $typeID = $result2['type_id'];

          // Handle telegram
          if ($typeID == 1) {
              $msg = str_replace("#s", $result['firstname'], $result2['template_data2']);
              if ( ! Notification::submit_queue_telegram($result['userID'], $result['firstname'], $msg) ) {
                Queue::update_notification_retries($result['task_id'], $result['subscriber_id']); // Sent
              } else {
                Queue::delete_notification($result['task_id'], $result['subscriber_id']); // Failed
              }
          }

          // Handle email
          if ($typeID == 2) {
              $msg = str_replace("%token%", $result['token'], $result2['template_data2']);
              if ( ! Notification::submit_queue_email($result['userID'], $result2['template_data1'], $msg) ) {
                Queue::update_notification_retries($result['task_id'], $result['subscriber_id']); // Sent

              } else {
                Queue::delete_notification($result['task_id'], $result['subscriber_id']); // Failed
              }
          }
      }

      // Check if queue log is empty and if so delete the queue_task
      $stmt = $mysqli->query("SELECT id, (SELECT COUNT(*) FROM queue_notify AS qn WHERE qn.task_id = queue_task.id) AS count FROM queue_task");
      while ( $result = $stmt->fetch_assoc() ) {
        if ( $result['count'] == 0 ) {
          Queue::delete_task($result['id']);
        }
      }

    }
}