<?php
 namespace App\Webhook; class QueueManager { static private $instance; private $dbTable; private function __construct() { global $DBTWebhookQueue; $this->dbTable = $DBTWebhookQueue; } static public function getInstance() { if (!self::$instance) { !self::$instance = new self(); } return self::$instance; } public function getRecord(string $entity, int $id, string $action, bool $forUpdate=false) { $db = \DB::getInstance(); $sql = "SELECT * FROM `{$this->dbTable}` WHERE entity=? AND id=? AND action=?"; if ($forUpdate) { $sql .= " FOR UPDATE"; } $record = $db->GetRow($sql, [$entity, $id, $action]); return $record ? : null; } public function insertEntityAction(string $entity, int $id, string $action) { if (!\DB::isInsideTransaction()) { throw new \Exception('You must be inside a transaction when running '.__METHOD__); } $entActionCfg = \App\Config\WebhookQueue::getInstance(); if (!$entActionCfg->get('enabled')) { return false; } if (!$entActionCfg->getActiveEntityAction($entity, $action)) { return false; } $record = $this->getRecord($entity, $id, $action, true); $db = \DB::getInstance(); if (!$record) { $insertSQL = "INSERT INTO `{$this->dbTable}` ( `entity`, `id`, `action`, `lastSentValue`, `toSend`, `addTime`) VALUES (?, ?, ?, ?, ?, ?)"; $insertEx = $db->Execute($insertSQL, [$entity, $id, $action, null, 1, DATETIME]); if (!$insertEx) { throw new \DBException('Could not insert a new record in the webhook queue.'); } } else { if (!$record['toSend']) { $updateSQL = "UPDATE `{$this->dbTable}` SET toSend=?, lastUpdateTime=? WHERE `entity`=? AND `id`=? AND `action`=?"; $updateEx = $db->Execute($updateSQL, [1, DATETIME, $entity, $id, $action]); if (!$updateEx) { throw new \DBException('Could not update a record in the webhook queue.'); } } } } public function confirmEntityAction(string $entity, int $id, string $action, $newSentValue) { if (!\DB::isInsideTransaction()) { throw new \Exception('You must be inside a transaction in order to call '.__METHOD__); } $db = \DB::getInstance(); $updateRecordSQL = "UPDATE `{$this->dbTable}` SET toSend=0, lastSentValue=?, lastSendTime=? WHERE entity=? AND id=? AND action=?"; $updateRecordEx = $db->Execute($updateRecordSQL, [$newSentValue, \Engine\DateTime::create('now')->formatDatetimeIso(), $entity, $id, $action]); if (!$updateRecordEx) { throw new \DBException('Could not confirm entity action in webhook queue.'); } } public function getAllEntitiesToSend($limit=0) { $getRecordsSQL = "SELECT entity, id, action FROM `{$this->dbTable}` WHERE toSend=1 ORDER BY addTime ASC, lastUpdateTime ASC"; if (is_numeric($limit) and $limit>0) { $getRecordsSQL .= " LIMIT ".$limit; } $db = \DB::getInstance(); $recordsList = $db->GetAll($getRecordsSQL); if ($db->ErrorNo()) { throw new \DBException('Could not get webhook entities to be sent.'); } return $recordsList; } public function cleanupOldEntityActions(\Engine\DateTime $datetime) { $db = \DB::getInstance(); $cleanQueueSQL = "DELETE FROM `{$this->dbTable}` WHERE toSend=? AND addTime<=?"; $db->Execute($cleanQueueSQL, [0, $datetime->formatDatetimeIso()]); if ($db->ErrorNo()) { throw new \DBException('Could not clean old entities.'); } } }