michaelschiemer/src/Framework/Database/ReadWriteConnection.php

<?php

declare(strict_types=1);

namespace App\Framework\Database;

use App\Framework\Database\Config\ReadWriteConfig;
use App\Framework\Database\Exception\DatabaseException;
use App\Framework\Database\ReadWrite\MasterSlaveRouter;
use App\Framework\Database\ReadWrite\ReplicationLagDetector;
use App\Framework\Database\ValueObjects\SqlQuery;
use App\Framework\DateTime\Clock;

/**
 * Class ReadWriteConnection provides a mechanism to handle separate writing and reading database connections.
 * It ensures that write-heavy operations are directed to a designated write connection,
 * while read operations are distributed among a pool of read connections.
 *
 * Enhanced with load balancing, health checks, and lag detection.
 */
final class ReadWriteConnection implements ConnectionInterface
{
    private ConnectionInterface $writeConnection;

    private array $readConnections;

    private bool $forceWrite = false;

    private ?MasterSlaveRouter $router = null;

    public function __construct(
        ConnectionInterface $writeConnection,
        array $readConnections,
        private readonly ?ReadWriteConfig $config = null,
        private readonly ?Clock $clock = null,
        private readonly ?ReplicationLagDetector $lagDetector = null
    ) {
        $this->writeConnection = $writeConnection;
        $this->readConnections = $readConnections;

        if (empty($this->readConnections)) {
            throw new DatabaseException('At least one read connection is required');
        }

        // Initialize advanced router if dependencies are provided
        if ($this->config && $this->clock && $this->lagDetector) {
            $this->router = new MasterSlaveRouter(
                $writeConnection,
                $readConnections,
                $this->config,
                $this->clock,
                $this->lagDetector
            );
        }
    }

    public function execute(SqlQuery $query): int
    {
        $this->forceWrite = true;

        return $this->writeConnection->execute($query);
    }

    public function query(SqlQuery $query): ResultInterface
    {
        $connection = $this->getConnection($query->sql);

        return $connection->query($query);
    }

    public function queryOne(SqlQuery $query): ?array
    {
        $connection = $this->getConnection($query->sql);

        return $connection->queryOne($query);
    }

    public function queryColumn(SqlQuery $query): array
    {
        $connection = $this->getConnection($query->sql);

        return $connection->queryColumn($query);
    }

    public function queryScalar(SqlQuery $query): mixed
    {
        $connection = $this->getConnection($query->sql);

        return $connection->queryScalar($query);
    }

    public function beginTransaction(): void
    {
        $this->forceWrite = true;
        $this->writeConnection->beginTransaction();
    }

    public function commit(): void
    {
        $this->writeConnection->commit();
        $this->forceWrite = false;
    }

    public function rollback(): void
    {
        $this->writeConnection->rollback();
        $this->forceWrite = false;
    }

    public function inTransaction(): bool
    {
        return $this->writeConnection->inTransaction();
    }

    public function lastInsertId(): string
    {
        return $this->writeConnection->lastInsertId();
    }

    public function getPdo(): \PDO
    {
        return $this->writeConnection->getPdo();
    }

    public function forceWriteConnection(): void
    {
        $this->forceWrite = true;
    }

    public function resetConnectionMode(): void
    {
        $this->forceWrite = false;
        $this->router?->resetStickyConnection();
    }

    private function shouldUseWriteConnection(string $sql): bool
    {
        if ($this->forceWrite || $this->inTransaction()) {
            return true;
        }

        $sql = trim(strtoupper($sql));
        $writeOperations = ['INSERT', 'UPDATE', 'DELETE', 'CREATE', 'ALTER', 'DROP', 'TRUNCATE'];

        return array_any($writeOperations, fn ($operation) => str_starts_with($sql, $operation));

    }

    /**
     * Get appropriate connection for SQL query
     */
    private function getConnection(string $sql): ConnectionInterface
    {
        if ($this->router) {
            return $this->router->route($sql, $this->forceWrite);
        }

        // Fallback to simple routing
        if ($this->shouldUseWriteConnection($sql)) {
            return $this->writeConnection;
        }

        return $this->getReadConnection();
    }

    /**
     * Simple round-robin read connection selection (fallback)
     */
    private function getReadConnection(): ConnectionInterface
    {
        static $currentReadIndex = 0;

        $connection = $this->readConnections[$currentReadIndex];
        $currentReadIndex = ($currentReadIndex + 1) % count($this->readConnections);

        return $connection;
    }

    /**
     * Get read/write statistics
     */
    public function getStatistics(): array
    {
        if ($this->router) {
            return $this->router->getStatistics();
        }

        return [
            'router' => 'simple',
            'master_connection' => 'active',
            'replica_count' => count($this->readConnections),
            'load_balancing_strategy' => 'round_robin',
        ];
    }

    /**
     * Get write connection (for direct access if needed)
     */
    public function getWriteConnection(): ConnectionInterface
    {
        return $this->writeConnection;
    }

    /**
     * Get read connections
     */
    public function getReadConnections(): array
    {
        return $this->readConnections;
    }

    /**
     * Check if advanced routing is enabled
     */
    public function hasAdvancedRouting(): bool
    {
        return $this->router !== null;
    }
}