xNightR00T File Manager

<?php

require_once dirname(__FILE__) . "/zs_component_upgrader.php";

class JQDBHandler extends ZSComponentUpgrader   {

	private $dbPath;
	private $dbh;
	private $currentDbSchemaVersion = "2018.0.0";

	protected function getSqliteDbName($version = "6.0.0") {
		return "jobqueue.db";
	}

	protected function getMysqlSqlFilenames() {
		return array("jobqueue_mysql_create_database.sql");
	}

	protected function getSqliteSqlFilenames() {
		return array("jobqueue_sqlite_create_database.sql");
	}

	public function getComponent() { return "jobqueue"; }

	public function getSchemaVersion() { return $this->currentDbSchemaVersion; }

	public function getMysqlSchemaVersion() {
		$dbh = $this->getMySqlDbh();
		foreach ($dbh->query("SELECT property_value FROM schema_properties WHERE property = 'JOBQUEUE_SCHEMA_VERSION'") as $row) {
			return $row['property_value'];
		}
		return null;
	}

	public function setMysqlSchemaVersion($dbh, $version) {
		$dbh->exec("REPLACE INTO schema_properties (property, property_value) VALUES ('JOBQUEUE_SCHEMA_VERSION', '{$version}')");
	}

	public function getSqliteSchemaVersion() {
		$dbh = $this->getSqliteDbh();
		foreach ($this->dbh->query('SELECT version FROM version') as $row) {
			return $this->getParsedVersionString($row['version']);
		}

		return null;
	}

	public function setSqliteSchemaVersion($dbh, $version) {
		$dbh->exec("INSERT OR REPLACE INTO version (key, version) VALUES ('SCHEMA_VERSION', '{$version}')");
	}


	function __construct() {

		$this->copySqliteFromPreviousVersions();

		$this->dbPath = get_cfg_var('zend.data_dir') . "/db/jobqueue.db";
		$this->dbh = NULL;

		$db6Path = get_cfg_var('zend.data_dir') . "/db-6.0.0/jobqueue.db";
		if (file_exists($db6Path) && !file_exists($this->dbPath)) {
			message("Using jq db at $db6Path");
			copyWithPerms($db6Path, $this->dbPath);
		}

		$exists = file_exists($this->dbPath);

		$this->dbh = new PDO("sqlite:{$this->dbPath}");
		$this->dbh->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_WARNING);

		if (!$exists ) {
			message("Creating JQ database in {$this->dbPath}");
			$queries = file_get_contents($_GET['zsDir'] . "/share/jobqueue_sqlite_create_database.sql");

			try {
				$this->dbh->beginTransaction();
				execQueries($this->dbh, $queries);
				$this->dbh->commit();
			} catch (PDOException $exc) {
				ERROR($exc->getMessage());
				$this->dbh->rollback();
			}

			chmod($this->dbPath, 0664);

		} else {
			message("JQ database exists in {$this->dbPath}");
		}
	}

	/**
	 * * @param PDO $db
	 */
	protected function updateDefaultQueueByIni($db) {

		message("Updating Job Queue default queue settings according to jqd.ini");

		$jqdIni = parse_ini_file($_GET['zsDir'] . "/etc/jqd.ini");

		$stmt = $db->prepare("UPDATE jobqueue_queue SET max_http_jobs=:max_http_jobs, http_connection_timeout=:http_connection_timeout, http_job_timeout=:http_job_timeout, http_job_retry_count=:http_job_retry_count, http_job_retry_timeout=:http_job_retry_timeout WHERE id = 1");
		$stmt->bindValue(":max_http_jobs", $jqdIni['zend_jobqueue.max_http_jobs']);
		$stmt->bindValue(":http_connection_timeout", $jqdIni['zend_jobqueue.connection_timeout']);
		$stmt->bindValue(":http_job_timeout", $jqdIni['zend_jobqueue.http_job_timeout']);
		$stmt->bindValue(":http_job_retry_count", $jqdIni['zend_jobqueue.http_job_retry_count']);
		$stmt->bindValue(":http_job_retry_timeout", $jqdIni['zend_jobqueue.http_job_retry_timeout']);

		$stmt->execute();
	}

	public function upgrade() {

		if ($this->dbh) {

			message("Checking if sqlite upgrade is needed...");
			$dbCurrentVersion = "";
			foreach ($this->dbh->query('SELECT version FROM version') as $row) {
				$dbCurrentVersion = $row['version'];
			}
			message("Found current version $dbCurrentVersion");

			try {
				//$this->dbh->beginTransaction();

				$nodeId = get_cfg_var('zend.node_id');

				if (version_compare($dbCurrentVersion, "9.1.0", "<")) {

					message("Upgrading to 9.1.0");

					$queries = "
					CREATE TABLE IF NOT EXISTS jobqueue_job_retries (
						id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
						job_id INTEGER NOT NULL,
						status INTEGER NOT NULL,
						creation_time timestamp NULL DEFAULT NULL,
						schedule_time timestamp NULL DEFAULT NULL,
						start_time timestamp NULL DEFAULT NULL,
						end_time timestamp NULL DEFAULT NULL
					);
					CREATE INDEX IF NOT EXISTS IDX_JOBQUEUE_RETRIES ON jobqueue_job_retries (job_id, status);

					CREATE INDEX IF NOT EXISTS IDX_JOBQUEUE_JOBS_LIST ON jobqueue_job (schedule_id, queue_id ASC);

					UPDATE jobqueue_job SET status = 13 WHERE status = 5; /* update failed jobs with http error status */

					";

					execQueries($this->dbh, $queries);
				}

				if (version_compare($dbCurrentVersion, "9.1.2", "<")) {

					message("Upgrading to 9.1.3 - adding indexes");

					$queries = "
					CREATE INDEX IF NOT EXISTS IDX_JOBQUEUE_SCHED_STIME on jobqueue_job(schedule_id, start_time DESC);
					CREATE INDEX IF NOT EXISTS IDX_JOBQUEUE_SCHED_SCHTIME on jobqueue_job(schedule_id, schedule_time DESC);
					";

					execQueries($this->dbh, $queries);
				}

				if (version_compare($dbCurrentVersion, "2018.0.0", "<")) {

					message("Upgrading to 2018.0.0");

					$queries = "
					CREATE INDEX IF NOT EXISTS IDX_JOBQUEUE_RETRIES_END ON jobqueue_job_retries (end_time);
					ALTER TABLE jobqueue_queue ADD cli_job_timeout INTEGER;
					UPDATE jobqueue_queue SET cli_job_timeout = 1800;

					ALTER TABLE jobqueue_job ADD progress INTEGER DEFAULT NULL;
					ALTER TABLE jobqueue_job ADD progress_message VARCHAR(1024) DEFAULT NULL;

					INSERT OR REPLACE INTO version (key, version) VALUES ('SCHEMA_VERSION', '{$this->currentDbSchemaVersion}');

					";

					execQueries($this->dbh, $queries);
				}

			} catch (PDOException $exc) {
				ERROR($exc->getMessage());
				//$this->dbh->rollback();
			}
		}

		try {
			message('Checking if MYSQL upgrade is in needed...');
			$dbIni = parse_ini_file($_GET['zsDir'] . "/etc/zend_database.ini");

			$oldDbIni = NULL;
			if (file_exists($_GET['zsDir'] . "/etc-5.6.0/monitor_node.ini")) {
				$oldDbIni = parse_ini_file($_GET['zsDir'] . "/etc-5.6.0/monitor_node.ini");
			}


			$dbType = strtolower($dbIni['zend.database.type']);
			if (strstr($dbType, "mysql")) {
				message("PE with MYSQL installation detected");
				$dbHost     = $dbIni['zend.database.host_name'];
				$dbPort     = $dbIni['zend.database.port'];
				$dbUsername = $dbIni['zend.database.user'];
				$dbPassword = $dbIni['zend.database.password'];
				$dbName     = $dbIni['zend.database.name'];
				$mysqlDbh = new PDO("mysql:host=$dbHost;port=$dbPort;dbname=$dbName",$dbUsername, $dbPassword);
				$mysqlDbh->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_WARNING);

				if ($oldDbIni) {
					$dbHostOld     = $dbIni['zend_monitor.database.host_name'];
					$dbPortOld     = $dbIni['zend_monitor.database.port'];
					$dbUsernameOld = $dbIni['zend_monitor.database.user'];
					$dbPasswordOld = $dbIni['zend_monitor.database.password'];
					$dbNameOld     = $dbIni['zend_monitor.database.name'];

				}


				$res = $mysqlDbh->query("SELECT property_value FROM schema_properties WHERE property = 'JOBQUEUE_SCHEMA_VERSION'");
				$jqSchemaVersion = false;
				foreach ($res as $row) {
					$jqSchemaVersion = $row['property_value'];
					break;
				}
				if (!$jqSchemaVersion) {
					message("Job Queue tables not found in MySQL db, will create...");
					$mysqlCreationQueries = file_get_contents($_GET['zsDir'] . "/share/jobqueue_mysql_create_database.sql");

					$mysqlCreationQueries .= "
						REPLACE INTO schema_properties VALUES('JOBQUEUE_SCHEMA_VERSION', '{$this->currentDbSchemaVersion}');";

					try {
						$mysqlDbh->beginTransaction();
						execQueries($mysqlDbh, $mysqlCreationQueries);
						$mysqlDbh->commit();
					} catch (PDOException $exc) {
						ERROR($exc->getMessage());
						$mysqlDbh->rollback();
					}
				} else {
					message("Job Queue tables already created. will upgrade to $this->currentDbSchemaVersion");
					message("Upgrading from schema version $jqSchemaVersion");
					try {
						$mysqlDbh->beginTransaction();

						$queries = "";

						if (version_compare($jqSchemaVersion, "9.0.0", "<")) {
							$queries =
							"CREATE INDEX IDX_QUEUE_STATUS ON jobqueue_queue (status);
							 DROP INDEX IDX_JOBQUEUE_NEXT_JOB ON jobqueue_job;
							 CREATE INDEX IDX_JOBQUEUE_NEXT_JOB ON jobqueue_job (queue_id,status,priority);
							 CREATE INDEX IDX_JOBQUEUE_JOBS_LIST ON jobqueue_job (schedule_id, queue_id ASC);
							";

							execQueries($mysqlDbh, $queries);

						}

						if (version_compare($jqSchemaVersion, "9.1.0", "<")) {

						    $queries =
						    "
							CREATE TABLE jobqueue_job_retries (
								id INTEGER NOT NULL PRIMARY KEY AUTO_INCREMENT,
								job_id INTEGER NOT NULL,
								status INTEGER NOT NULL,
								creation_time timestamp NULL DEFAULT NULL,
								schedule_time timestamp NULL DEFAULT NULL,
								start_time timestamp NULL DEFAULT NULL,
								end_time timestamp NULL DEFAULT NULL
							);
							CREATE INDEX IDX_JOBQUEUE_RETRIES ON jobqueue_job_retries (job_id, status);

							CREATE INDEX IDX_JOBQUEUE_JOBS_LIST ON jobqueue_job (schedule_id, queue_id ASC);

							UPDATE jobqueue_job SET status = 13 WHERE status = 5; /* update failed jobs with http error status */
							";

						    message("JQ upgrading to 9.1. Running queries : ($queries)");
							execQueries($mysqlDbh, $queries);
						}

						if (version_compare($jqSchemaVersion, "2018.0.0", "<")) {

							message("JQ upgrading to 2018.0.0");

						    $queries =
						    "
							SET SESSION sql_mode = '';

							CREATE INDEX IDX_JOBQUEUE_RETRIES_END ON jobqueue_job_retries (end_time);
							UPDATE jobqueue_job SET schedule_time=creation_time WHERE schedule_id = 0 AND schedule_time = '0000-00-00 00:00:00';
							ALTER TABLE jobqueue_queue ADD cli_job_timeout INTEGER;
							UPDATE jobqueue_queue SET cli_job_timeout = 1800;
							ALTER TABLE jobqueue_job ADD progress INTEGER DEFAULT NULL;
							ALTER TABLE jobqueue_job ADD progress_message VARCHAR(1024) DEFAULT NULL;
							";
							message("JQ upgrading to 2018.0.0. Running queries : ($queries)");

							execQueries($mysqlDbh, $queries);
						}

						$queries = "REPLACE INTO schema_properties VALUES('JOBQUEUE_SCHEMA_VERSION', '{$this->currentDbSchemaVersion}')";
						execQueries($mysqlDbh, $queries);

						if (version_compare($jqSchemaVersion, "8.5.0", "<")) {
							$this->updateDefaultQueueByIni($mysqlDbh);
						}

						$mysqlDbh->commit();
					} catch (PDOException $exc) {
						ERROR($exc->getMessage());
						$mysqlDbh->rollback();
					}
				}

			} else {
				message("Non MYSQL configuration detected - found " . $dbType. ".");
			}

		} catch (Exception $exc) {
			ERROR($exc->getMessage());
		}
	}


	protected function upgradeSqlite() {

	}

	protected function upgradeMysql() {


	}
}

class JQDbCreator_2019_0_0 extends JQDBHandler
{
    const SCHEMA_VERSION = "2019.0.0";

	public function getSchemaVersion() { return self::SCHEMA_VERSION; }

	public function upgrade() {
        $this->upgradeSqlite();

        if ($this->isClusterUpgrade()) {
            $this->upgradeMysql();
        }
    }

	public function upgradeSqlite() {
		message("Upgrading Job Queue SQLITE schema to " . self::SCHEMA_VERSION);
		$dbh = $this->getSqliteDbh();

		try {
			$dbh->exec("DROP INDEX IDX_JOBQUEUE_SCHEDULE_NAME");
		} catch (Exception $ex) {}

		$dbh->exec("
			CREATE UNIQUE INDEX IF NOT EXISTS IDX_JOBQUEUE_SCHEDULE_NAME_STATUS ON jobqueue_schedule (name, status);
		");

		$this->setSqliteSchemaVersion( $dbh, self::SCHEMA_VERSION );
	}

	public function upgradeMysql() {
		message("Upgrading Job Queue MYSQL schema to " . self::SCHEMA_VERSION);
		$dbh = $this->getMySqlDbh();
		try {
			$dbh->exec("DROP INDEX IDX_JOBQUEUE_SCHEDULE_NAME ON jobqueue_schedule");
		} catch (Exception $ex) {}

		$dbh->exec("
			CREATE UNIQUE INDEX IDX_JOBQUEUE_SCHEDULE_NAME_STATUS ON jobqueue_schedule (name, status);
		");
		$this->setMysqlSchemaVersion($dbh, self::SCHEMA_VERSION);
	}
}