kiri-databases/Command.php

<?php
/**
 * Created by PhpStorm.
 * User: whwyy
 * Date: 2018/3/30 0030
 * Time: 15:23
 */
declare(strict_types=1);

namespace Database;


use Exception;
use Kiri\Abstracts\Component;
use Kiri\Di\Container;
use PDO;
use Throwable;

/**
 * Class Command
 * @package Database
 */
class Command extends Component
{

	/** @var int 数据库操作最大重试次数 */
	private const MAX_RETRIES = 2;

	public Connection $connection;
	public ?string    $sql    = '';
	public array      $params = [];
	
	/**
	 * @param array $params
	 * @throws
	 */
	public function __construct(array $params = [])
	{
		parent::__construct();
		Container::configure($this, $params);
	}
	
	
	/**
	 * @return bool
	 * @throws
	 */
	public function incrOrDecr(): bool
	{
		return (bool)$this->_prepare();
	}
	
	
	/**
	 * @return bool|array
	 * @throws
	 */
	public function all(): bool|array
	{
		return $this->search('fetchAll');
	}
	
	/**
	 * @return array|bool|null
	 * @throws
	 */
	public function one(): array|null|bool
	{
		return $this->search('fetch');
	}
	
	/**
	 * @return bool
	 * @throws Exception
	 */
	public function exists(): bool
	{
		$data = $this->search('fetch');
		if (!$data) {
			return false;
		}
		return true;
	}
	
	/**
	 * @return mixed
	 * @throws
	 */
	public function fetchColumn(): mixed
	{
		return $this->search('fetchColumn');
	}
	
	/**
	 * @return mixed
	 * @throws
	 */
	public function rowCount(): int
	{
		$data = $this->search('fetch');
		
		return !$data ? 0 : +current($data);
	}
	
	
	/**
	 * 执行查询类 SQL 操作，内置断连重试机制
	 * 当数据库连接断开时自动重试，断开的连接不再归还池中防止污染连接池
	 * @param string $method PDO 结果获取方法名
	 * @return mixed
	 * @throws
	 */
	protected function search(string $method): mixed
	{
		$retryCount = 0;

		while ($retryCount <= self::MAX_RETRIES) {
			$client = $this->connection->getConnection();
			try {
				$startTime = microtime(true);
				if (($prepare = $client->prepare($this->sql)) === false) {
					throw new Exception('(' . $prepare->errorInfo()[0] . ')' . $client->errorInfo()[2]);
				}
				
				$prepare->execute($this->params);
				
				$result = $method == 'rowCount' ? $prepare->rowCount() : $prepare->{$method}(PDO::FETCH_ASSOC);
				$prepare->closeCursor();
				
				$this->connection->println($startTime, microtime(true), $this->sql, $this->params);
				
				$this->connection->release($client);
				return $result;
			} catch (Throwable $throwable) {
				$this->getLogger()->json_log($throwable);
				
				if ($this->isRefresh($throwable) && $retryCount < self::MAX_RETRIES) {
					$retryCount++;
					$this->connection->connections->abandon($this->connection->getName());
					continue;
				}
				
				$this->connection->release($client);
				
				$errorMsg = $throwable->getMessage() . PHP_EOL . '                 Sql: ' . $this->sql . '.' . json_encode($this->params);
				
				return $this->getLogger()->logCategory($errorMsg . PHP_EOL, 'mysql');
			}
		}

		$errorMsg = 'Max retry exceeded for SQL: ' . $this->sql;
		return $this->getLogger()->logCategory($errorMsg, 'mysql');
	}
	
	
	/**
	 * @return bool
	 * @throws
	 */
	public function flush(): bool
	{
		return (bool)$this->_prepare();
	}
	
	
	/**
	 * 执行写入类 SQL 操作，内置断连重试机制
	 * 当数据库连接断开时自动重试，断开的连接不再归还池中防止污染连接池
	 * @return int|bool
	 * @throws
	 */
	private function _prepare(): int|bool
	{
		$retryCount = 0;

		while ($retryCount <= self::MAX_RETRIES) {
			$client = $this->connection->getConnection();
			try {
				$startTime = microtime(true);
				if (($prepare = $client->prepare($this->sql)) === false) {
					throw new Exception('(' . $prepare->errorInfo()[0] . ')' . $prepare->errorInfo()[2]);
				}
				if ($prepare->execute($this->params) === false) {
					throw new Exception('(' . $prepare->errorInfo()[0] . ')' . $prepare->errorInfo()[2]);
				}
				
				$prepare->closeCursor();
				
				$result = $client->lastInsertId();
				
				$this->connection->println($startTime, microtime(true), $this->sql, $this->params);
				if (str_starts_with($this->sql, 'DELETE')) {
					$this->connection->release($client);
					return $prepare->rowCount();
				}
				
				$this->connection->release($client);
				return $result == 0 ? $prepare->rowCount() : (int)$result;
			} catch (Throwable $throwable) {
				$this->getLogger()->json_log($throwable);
				
				if ($this->isRefresh($throwable) && $retryCount < self::MAX_RETRIES) {
					$retryCount++;
					$this->connection->connections->abandon($this->connection->getName());
					continue;
				}
				
				$this->connection->release($client);
				
				$errorMsg = $throwable->getMessage() . PHP_EOL . '                 Sql: ' . $this->sql . '.' . json_encode($this->params, JSON_UNESCAPED_UNICODE);
				
				return $this->getLogger()->logCategory($errorMsg . PHP_EOL, 'mysql');
			}
		}

		$errorMsg = 'Max retry exceeded for SQL: ' . $this->sql;
		return $this->getLogger()->logCategory($errorMsg, 'mysql');
	}
	
	
	/**
	 * @param Throwable $throwable
	 * @return bool
	 */
	protected function isRefresh(Throwable $throwable): bool
	{
		$message = $throwable->getMessage();
		
		// MySQL 错误处理
		if (str_contains($message, 'MySQL server has gone away')) {
			return true;
		}
		if (str_contains($message, 'Send of 14 bytes failed with errno=32 Broken pipe')) {
			return true;
		}
		if (str_contains($message, 'Lost connection to MySQL server during query')) {
			return true;
		}
		
		// PostgreSQL 错误处理
		if (str_contains($message, 'server closed the connection unexpectedly')) {
			return true;
		}
		if (str_contains($message, 'Connection refused')) {
			return true;
		}
		if (str_contains($message, 'Broken pipe')) {
			return true;
		}
		if (str_contains($message, 'connection to server was lost')) {
			return true;
		}
		
		return false;
	}
	
	
	/**
	 * @return bool
	 * @throws
	 */
	public function delete(): bool
	{
		return $this->_prepare();
	}
	
	
	/**
	 * @return int|bool
	 */
	public function exec(): int|bool
	{
		return $this->_prepare();
	}
	
	/**
	 * @param array $data
	 * @return $this
	 */
	public function bindValues(array $data = []): static
	{
		if (count($data) > 0) {
			$this->params = array_merge($this->params, $data);
		}
		return $this;
	}
	
}