Session.php

<?php

declare(strict_types=1);

/*
 * This file is part of the Neo4j PHP Client and Driver package.
 *
 * (c) Nagels <https://nagels.tech>
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace Laudis\Neo4j\Bolt;

use Bolt\error\ConnectException as BoltConnectException;
use Exception;
use Laudis\Neo4j\Common\GeneratorHelper;
use Laudis\Neo4j\Common\Neo4jLogger;
use Laudis\Neo4j\Contracts\ConnectionInterface;
use Laudis\Neo4j\Contracts\ConnectionPoolInterface;
use Laudis\Neo4j\Contracts\SessionInterface;
use Laudis\Neo4j\Contracts\TransactionInterface;
use Laudis\Neo4j\Contracts\UnmanagedTransactionInterface;
use Laudis\Neo4j\Databags\Bookmark;
use Laudis\Neo4j\Databags\BookmarkHolder;
use Laudis\Neo4j\Databags\Neo4jError;
use Laudis\Neo4j\Databags\SessionConfiguration;
use Laudis\Neo4j\Databags\Statement;
use Laudis\Neo4j\Databags\SummarizedResult;
use Laudis\Neo4j\Databags\TransactionConfiguration;
use Laudis\Neo4j\Enum\AccessMode;
use Laudis\Neo4j\Exception\Neo4jException;
use Laudis\Neo4j\Formatter\SummarizedResultFormatter;
use Laudis\Neo4j\Neo4j\Neo4jConnectionPool;
use Laudis\Neo4j\NoOpBookmarkManager;
use Laudis\Neo4j\Types\CypherList;
use Psr\Log\LogLevel;
use Throwable;

/**
 * A session using bolt connections.
 */
final class Session implements SessionInterface
{
    private const ROLLBACK_CLASSIFICATIONS = ['ClientError', 'DatabaseError'];

    /** @var list<BoltConnection> */
    private array $usedConnections = [];
    private readonly BookmarkHolder $bookmarkHolder;
    private readonly SessionBookmarkTracker $bookmarkTracker;
    private readonly SessionConfiguration $sessionConfig;

    /**
     * @param ConnectionPool|Neo4jConnectionPool $pool
     */
    public function __construct(
        SessionConfiguration $config,
        private readonly ConnectionPoolInterface $pool,
        /**
         * @psalm-readonly
         */
        private readonly SummarizedResultFormatter $formatter,
    ) {
        $bookmarkManager = $config->getBookmarkManager() ?? NoOpBookmarkManager::instance();
        $this->bookmarkHolder = new BookmarkHolder(Bookmark::from($config->getBookmarks()));
        $this->bookmarkTracker = new SessionBookmarkTracker(
            $this->bookmarkHolder,
            $bookmarkManager,
            $config->getBookmarks(),
        );
        $this->bookmarkHolder->onServerBookmark($this->bookmarkTracker->handleNewBookmark(...));
        $this->sessionConfig = $config->withRoutingBookmarks($this->bookmarkTracker->getRediscoveryBookmarkValues());
    }

    /**
     * @param iterable<Statement> $statements
     *
     * @return CypherList<SummarizedResult>
     */
    public function runStatements(iterable $statements, ?TransactionConfiguration $config = null): CypherList
    {
        $tbr = [];

        $this->getLogger()?->log(LogLevel::INFO, 'Running statements', ['statements' => $statements]);
        $config = $this->mergeTsxConfig($config);

        foreach ($statements as $statement) {
            $tbr[] = $this->executeStatementWithRetry($statement, $config);
        }

        return new CypherList($tbr);
    }

    /**
     * @param iterable<Statement>|null $statements
     */
    public function openTransaction(?iterable $statements = null, ?TransactionConfiguration $config = null): UnmanagedTransactionInterface
    {
        return $this->beginTransaction($statements, $this->mergeTsxConfig($config));
    }

    public function runStatement(Statement $statement, ?TransactionConfiguration $config = null): SummarizedResult
    {
        return $this->runStatements([$statement], $config)->first();
    }

    public function run(string $statement, iterable $parameters = [], ?TransactionConfiguration $config = null): SummarizedResult
    {
        return $this->runStatement(new Statement($statement, $parameters), $config);
    }

    /**
     * @param array<string, mixed> $parameters
     */
    public function executeQuery(string $cypher, array $parameters = []): SummarizedResult
    {
        $statement = new Statement($cypher, $parameters);

        return $this->retry(
            static function (TransactionInterface $tx) use ($statement): SummarizedResult {
                $result = $tx->runStatement($statement);
                $result->preload();

                return $result;
            },
            $this->mergeTsxConfig(null),
            $this->sessionConfig,
            TelemetryAPIEnum::EXECUTE_QUERY,
        );
    }

    public function writeTransaction(callable $tsxHandler, ?TransactionConfiguration $config = null)
    {
        $this->getLogger()?->log(LogLevel::INFO, 'Beginning write transaction', ['config' => $config]);
        $config = $this->mergeTsxConfig($config);

        return $this->retry(
            $tsxHandler,
            $config,
            $this->sessionConfig->withAccessMode(AccessMode::WRITE()),
            TelemetryAPIEnum::TRANSACTION_FUNCTION
        );
    }

    public function readTransaction(callable $tsxHandler, ?TransactionConfiguration $config = null)
    {
        $this->getLogger()?->log(LogLevel::INFO, 'Beginning read transaction', ['config' => $config]);
        $config = $this->mergeTsxConfig($config);

        return $this->retry(
            $tsxHandler,
            $config,
            $this->sessionConfig->withAccessMode(AccessMode::READ()),
            TelemetryAPIEnum::TRANSACTION_FUNCTION
        );
    }

    /**
     * @template U
     *
     * @param callable(TransactionInterface):U $tsxHandler
     *
     * @return U
     */
    private function retry(
        callable $tsxHandler,
        TransactionConfiguration $config,
        SessionConfiguration $sessionConfig,
        TelemetryAPIEnum $telemetryApi,
    ) {
        $transaction = null;
        $maxRetries = 3;
        $error = null;
        while ($maxRetries > 0) {
            --$maxRetries;
            try {
                if ($transaction === null) {
                    $transaction = $this->startTransaction($config, $sessionConfig, $telemetryApi);
                }
                $tbr = $tsxHandler($transaction);

                $transaction->commit();

                return $tbr;
            } catch (Neo4jException $e) {
                if ($e->getClassification() === 'TransientError' && $transaction instanceof BoltUnmanagedTransaction) {
                    usleep(100_000); // transient errors mean we have to retry later. For now we wait 100 ms, later we'll make this configurable, non-blocking
                    $error = $e;

                    continue;
                }

                $transaction = null;
                if ($e->getTitle() === 'NotALeader' || $e->getNeo4jCode() === 'Neo.ClientError.Cluster.NotALeader') {
                    if ($this->pool instanceof Neo4jConnectionPool) {
                        $this->pool->clearRoutingTable($this->sessionConfig);
                    }
                    $error = $e;

                    continue;
                }

                throw $e;
            } catch (Throwable $e) {
                if ($this->isConnectionError($e)) {
                    $transaction = null;
                    $error = $e;

                    continue;
                }

                throw $e;
            }
        }

        throw new Exception('Failed to execute transaction', 0, $error);
    }

    /**
     * Check if the exception is a connection-related error (network/socket/timeout).
     * NotALeader is a routing error, not a connection error - the connection is fine.
     */
    private function isConnectionError(Throwable $e): bool
    {
        if ($e instanceof BoltConnectException) {
            return true;
        }

        $message = strtolower($e->getMessage());

        return str_contains($message, 'interrupted system call')
            || str_contains($message, 'broken pipe')
            || str_contains($message, 'connection reset')
            || str_contains($message, 'connection timeout')
            || str_contains($message, 'connection closed')
            || str_contains($message, 'connection refused')
            || str_contains($message, 'i/o error');
    }

    /**
     * Check if the exception should trigger a routing table clear.
     */
    private function shouldClearRoutingTable(Neo4jException $e): bool
    {
        $message = strtolower($e->getMessage());
        $title = $e->getTitle();

        return str_contains($message, 'interrupted system call')
            || str_contains($message, 'broken pipe')
            || str_contains($message, 'connection reset')
            || str_contains($message, 'connection timeout')
            || str_contains($message, 'connection closed')
            || $e->getNeo4jCode() === 'Neo.ClientError.Cluster.NotALeader'
            || $title === 'NotALeader';
    }

    /**
     * Execute a statement with automatic retry on connection errors.
     * Retries up to 3 times on connection failures, clearing routing table between attempts.
     *
     * @param Statement                $statement The statement to execute
     * @param TransactionConfiguration $config    Transaction configuration
     *
     * @return SummarizedResult The result of the statement
     */

    /**
     * Execute instant transaction (session.run) with automatic retry on connection/routing errors.
     *
     * PURPOSE:
     * - Handles transient failures transparently to user: connection timeouts, server unavailable, etc.
     * - Supports cluster failover: when server goes down, clears routing table and retries on different node
     * - Distinguishes errors: retries on connection/routing issues but fails immediately on client errors (syntax, auth)
     * - Improves reliability: 3 retry attempts with fresh routing table each time = high availability
     *
     * EXAMPLE: User calls session.run("CREATE (n)") during cluster failover:
     *   Attempt 1: Node A is leader → "NotALeader" (stepping down) → Clear routing table
     *   Attempt 2: Node B elected leader → "Connection timeout" (election in progress) → Retry
     *   Attempt 3: Cluster stable → Query succeeds
     *   User sees: Query succeeded transparently (no exception, no manual retry needed)
     *
     * WHY THIS IS CRITICAL FOR DRIVERS:
     * - All Neo4j drivers (Java, Python, JavaScript) have this pattern
     * - Without it: user must manually retry or wrap every session.run() in try-catch
     * - With it: driver handles recovery automatically = better UX and reliability
     */
    private function executeStatementWithRetry(Statement $statement, TransactionConfiguration $config): SummarizedResult
    {    // Retry instant transactions up to 3 times on connection/routing errors; catch distinguishes retryable errors from client errors (syntax, auth) and clears routing table for cluster failover.
        $maxRetries = 3;
        $retries = 0;

        while ($retries < $maxRetries) {
            try {
                return $this->beginInstantTransaction($this->sessionConfig, $config)->runStatement($statement);
            } catch (Neo4jException $e) {
                if (!$this->shouldClearRoutingTable($e)) {
                    throw $e;
                }
                $this->handleStatementRetry($retries, $maxRetries, $e);
            } catch (Throwable $e) {
                if (!$this->isConnectionError($e)) {
                    throw $e;
                }
                $this->handleStatementRetry($retries, $maxRetries, $e);
            }
        }

        throw new Neo4jException([Neo4jError::fromMessageAndCode('Neo.ClientError.General', 'Statement execution failed after maximum retries')]);
    }

    /**
     * Handle retry logic for statement execution - clear routing and increment counter.
     * Throws the exception if max retries exceeded.
     */
    private function handleStatementRetry(int &$retries, int $maxRetries, Throwable $e): void
    {
        $this->getLogger()?->log(LogLevel::WARNING, 'Connection error in instant transaction, retrying', [
            'error' => $e->getMessage(),
            'retry' => $retries + 1,
        ]);

        $this->pool->close();

        ++$retries;
        if ($retries >= $maxRetries) {
            throw $e;
        }
    }

    public function transaction(callable $tsxHandler, ?TransactionConfiguration $config = null)
    {
        return $this->writeTransaction($tsxHandler, $config);
    }

    /**
     * @param iterable<Statement> $statements
     */
    public function beginTransaction(?iterable $statements = null, ?TransactionConfiguration $config = null): UnmanagedTransactionInterface
    {
        $this->getLogger()?->log(LogLevel::INFO, 'Beginning transaction', ['statements' => $statements, 'config' => $config]);
        $config = $this->mergeTsxConfig($config);
        $tsx = $this->startTransaction($config, $this->sessionConfig, TelemetryAPIEnum::UNMANAGED_TRANSACTION);

        $tsx->runStatements($statements ?? []);

        return $tsx;
    }

    public function beginReadTransaction(?TransactionConfiguration $config = null): UnmanagedTransactionInterface
    {
        $config = $this->mergeTsxConfig($config);

        return $this->startTransaction($config, $this->sessionConfig->withAccessMode(AccessMode::READ()), TelemetryAPIEnum::TRANSACTION_FUNCTION);
    }

    public function beginWriteTransaction(?TransactionConfiguration $config = null): UnmanagedTransactionInterface
    {
        $config = $this->mergeTsxConfig($config);

        return $this->startTransaction($config, $this->sessionConfig->withAccessMode(AccessMode::WRITE()), TelemetryAPIEnum::TRANSACTION_FUNCTION);
    }

    /**
     * @return UnmanagedTransactionInterface
     */
    private function beginInstantTransaction(
        SessionConfiguration $config,
        TransactionConfiguration $tsxConfig,
    ): TransactionInterface {
        $this->getLogger()?->log(LogLevel::INFO, 'Starting instant transaction', ['config' => $tsxConfig]);
        $connection = $this->acquireConnection($tsxConfig, $config);

        /** @var ConnectionPoolInterface<ConnectionInterface>|null $pool */
        $pool = $this->pool;

        return new BoltUnmanagedTransaction(
            $this->sessionConfig->getDatabase(),
            $this->formatter,
            $connection,
            $this->sessionConfig,
            $tsxConfig,
            $this->bookmarkHolder,
            new BoltMessageFactory($connection, $this->getLogger()),
            true,
            TelemetryAPIEnum::SESSION_RUN,
            $pool,
            false,
            $this->bookmarkTracker,
        );
    }

    /**
     * @throws Exception
     */
    private function acquireConnection(TransactionConfiguration $config, SessionConfiguration $sessionConfig): BoltConnection
    {
        $this->getLogger()?->log(LogLevel::INFO, 'Acquiring connection', ['config' => $config, 'sessionConfig' => $sessionConfig]);
        $connectionGenerator = $this->pool->acquire($sessionConfig);
        /**
         * @var BoltConnection $connection
         *
         * @psalm-suppress UnnecessaryVarAnnotation
         */
        $connection = GeneratorHelper::getReturnFromGenerator($connectionGenerator);

        // We try and let the server do the timeout management.
        // Since the client should not run indefinitely, we just add the client side by two, just in case
        $timeout = $config->getTimeout();
        if ($timeout !== null) {
            $timeout = ($timeout < 30) ? 30 : $timeout;
            $connection->setTimeout($timeout + 2);
        }
        $this->usedConnections[] = $connection;

        return $connection;
    }

    private function startTransaction(TransactionConfiguration $config, SessionConfiguration $sessionConfig, TelemetryAPIEnum $telemetryApi): UnmanagedTransactionInterface
    {
        $this->getLogger()?->log(LogLevel::INFO, 'Starting transaction', ['config' => $config, 'sessionConfig' => $sessionConfig]);
        $connection = $this->acquireConnection($config, $sessionConfig);

        // Defer BEGIN to first run/commit/rollback. The driver does not support OPT_EAGER_TX_BEGIN,
        // so BEGIN is sent on first run/commit/rollback. This matches test_disconnect_on_tx_begin,
        // which expects the error at "after run" when the stub disconnects on BEGIN.

        /** @var ConnectionPoolInterface<ConnectionInterface>|null $pool */
        $pool = $this->pool;

        return new BoltUnmanagedTransaction(
            $this->sessionConfig->getDatabase(),
            $this->formatter,
            $connection,
            $this->sessionConfig,
            $config,
            $this->bookmarkHolder,
            new BoltMessageFactory($connection, $this->getLogger()),
            false,
            $telemetryApi,
            $pool,
            false,
            $this->bookmarkTracker,
        );
    }

    /**
     * Clean up a connection that failed during BEGIN or other initialization.
     * Resets the connection if it's in FAILED state and releases it back to the pool.
     */
    private function cleanupFailedConnection(BoltConnection $connection): void
    {
        if ($connection->getServerState() === 'FAILED') {
            $connection->reset();
        }
        // Release connection back to pool for reuse
        $this->pool->release($connection);
    }

    private function mergeTsxConfig(?TransactionConfiguration $config): TransactionConfiguration
    {
        return TransactionConfiguration::default()->merge($config);
    }

    public function getLastBookmark(): Bookmark
    {
        return $this->bookmarkHolder->getBookmark();
    }

    public function close(): void
    {
        foreach ($this->usedConnections as $connection) {
            $connection->discardUnconsumedResults();
        }
        $this->usedConnections = [];
    }

    private function getLogger(): ?Neo4jLogger
    {
        return $this->pool->getLogger();
    }
}