* @author Alexander Pankratov * @copyright 2016-2024 Daniil Gentili * @copyright 2016-2024 Alexander Pankratov * @license https://opensource.org/license/apache-2-0 Apache 2.0 * @link https://github.com/danog/AsyncOrm AsyncOrm documentation */ namespace danog\AsyncOrm\Internal\Driver; use Amp\Postgres\PostgresConnectionPool; use Amp\Sync\LocalKeyedMutex; use danog\AsyncOrm\DbArrayBuilder; use danog\AsyncOrm\Driver\SqlArray; use danog\AsyncOrm\Internal\Serializer\ByteaSerializer; use danog\AsyncOrm\Internal\Serializer\Passthrough; use danog\AsyncOrm\KeyType; use danog\AsyncOrm\Serializer; use danog\AsyncOrm\Settings\PostgresSettings; use danog\AsyncOrm\ValueType; /** * Postgres database backend. * * @internal * @template TKey as array-key * @template TValue * @extends SqlArray */ final class PostgresArray extends SqlArray { /** @var array */ private static array $connections = []; private static ?LocalKeyedMutex $mutex = null; /** * @psalm-suppress MethodSignatureMismatch * @param Serializer $serializer */ public function __construct(DbArrayBuilder $config, Serializer $serializer) { self::$mutex ??= new LocalKeyedMutex; $settings = $config->settings; \assert($settings instanceof PostgresSettings); $dbKey = $settings->getDbIdentifier(); $_ = self::$mutex->acquire($dbKey); if (!isset(self::$connections[$dbKey])) { try { $db = $settings->config->getDatabase(); $user = $settings->config->getUser(); $connection = new PostgresConnectionPool($settings->config->withDatabase(null)); $result = $connection->query("SELECT * FROM pg_database WHERE datname = '{$db}'"); // Replace with getRowCount once it gets fixed if (!\iterator_count($result)) { $connection->query(" CREATE DATABASE {$db} OWNER {$user} ENCODING utf8 "); } $connection->close(); } catch (\Throwable) { } self::$connections[$dbKey] = new PostgresConnectionPool($settings->config, $settings->maxConnections, $settings->idleTimeout); } $connection = self::$connections[$dbKey]; $keyType = match ($config->keyType) { KeyType::STRING_OR_INT => "VARCHAR(255)", KeyType::STRING => "VARCHAR(255)", KeyType::INT => "BIGINT", }; $valueType = match ($config->valueType) { ValueType::INT => "BIGINT", ValueType::STRING => "VARCHAR(255)", ValueType::FLOAT => "FLOAT(53)", ValueType::BOOL => "BOOLEAN", ValueType::SCALAR, ValueType::OBJECT => "BYTEA", }; /** @var Serializer */ $serializer = match ($config->valueType) { ValueType::SCALAR, ValueType::OBJECT => new ByteaSerializer($serializer), default => new Passthrough }; /** @psalm-suppress InvalidArgument */ parent::__construct( $config, $serializer, $connection, "SELECT value FROM \"bytea_{$config->table}\" WHERE key = :index", " INSERT INTO \"bytea_{$config->table}\" (key,value) VALUES (:index, :value) ON CONFLICT (key) DO UPDATE SET value = :value ", " DELETE FROM \"bytea_{$config->table}\" WHERE key = :index ", "SELECT count(key) as count FROM \"bytea_{$config->table}\"", "SELECT key, value FROM \"bytea_{$config->table}\"", "DELETE FROM \"bytea_{$config->table}\"" ); $connection->query(" CREATE TABLE IF NOT EXISTS \"bytea_{$config->table}\" ( \"key\" $keyType PRIMARY KEY NOT NULL, \"value\" $valueType NOT NULL ); "); $result = $connection->query("SELECT * FROM information_schema.columns WHERE table_name='bytea_{$config->table}'"); while ($column = $result->fetchRow()) { ['column_name' => $key, 'data_type' => $type, 'is_nullable' => $null] = $column; \assert(\is_string($key)); \assert(\is_string($type)); $type = \strtoupper($type); if (\str_starts_with($type, 'BIGINT')) { $type = 'BIGINT'; } if ($key === 'key') { $expected = $keyType; } elseif ($key === 'value') { $expected = $valueType; } else { // @codeCoverageIgnoreStart $connection->query("ALTER TABLE \"bytea_{$config->table}\" DROP \"$key\""); continue; // @codeCoverageIgnoreEnd } if ($expected !== $type) { if ($expected === 'BIGINT') { $expected .= " USING $key::bigint"; } $connection->query("ALTER TABLE \"bytea_{$config->table}\" ALTER COLUMN \"$key\" TYPE $expected"); } if ($null !== 'NO') { // @codeCoverageIgnoreStart $connection->query("ALTER TABLE \"bytea_{$config->table}\" ALTER COLUMN \"$key\" SET NOT NULL"); // @codeCoverageIgnoreEnd } } } #[\Override] protected function importFromTable(string $fromTable): void { if ($this->config->table === $fromTable) { return; } $this->db->query(/** @lang PostgreSQL */ " DROP TABLE \"bytea_{$this->config->table}\"; "); $this->db->query(/** @lang PostgreSQL */ " ALTER TABLE \"bytea_$fromTable\" RENAME TO \"bytea_{$this->config->table}\"; "); } }