MeshVault/tests/Integration/PostgresRepositoryTest.php

<?php

declare(strict_types=1);

namespace MeshVault\Tests\Integration;

use MeshVault\Repository\PdoImportJobRepository;
use PDO;
use PDOException;
use PHPUnit\Framework\TestCase;

/** Opt in with MESHVAULT_TEST_PG_DSN (and optional PG_USER / PG_PASSWORD with the same prefix). */
final class PostgresRepositoryTest extends TestCase
{
    private string $database;
    private PDO $pdo;
    private PDO $other;

    protected function setUp(): void
    {
        if (!RepositoryDatabase::postgresConfigured()) {
            self::markTestSkipped('Set MESHVAULT_TEST_PG_DSN to run PostgreSQL concurrency tests.');
        }
        // Configured connection/migration failures must be errors, never skips.
        $this->database = tempnam(sys_get_temp_dir(), 'meshvault-pg-');
        $this->pdo = RepositoryDatabase::migrate($this->database);
        $this->other = RepositoryDatabase::connect($this->database);
        foreach ([$this->pdo, $this->other] as $pdo) {
            $pdo->exec("SET lock_timeout = '250ms'");
            $pdo->exec("SET statement_timeout = '5s'");
        }
    }

    protected function tearDown(): void
    {
        foreach (['pdo', 'other'] as $property) {
            if (isset($this->$property) && $this->$property->inTransaction()) {
                $this->$property->rollBack();
            }
            unset($this->$property);
        }
        if (isset($this->database)) {
            RepositoryDatabase::cleanup($this->database);
        }
    }

    public function testMigrationsAndCleanupAreIsolatedToTheGeneratedSchema(): void
    {
        $path = tempnam(sys_get_temp_dir(), 'meshvault-pg-isolated-');
        try {
            $isolated = RepositoryDatabase::migrate($path);
            $schema = $isolated->query('SELECT current_schema()')->fetchColumn();
            $original = $this->pdo->query('SELECT current_schema()')->fetchColumn();
            self::assertMatchesRegularExpression('/^meshvault_test_[a-f0-9]{32}$/', $schema);
            self::assertNotSame($original, $schema);
            self::assertSame($original, $this->other->query('SELECT current_schema()')->fetchColumn());
            self::assertNotSame(
                $this->pdo->query('SELECT pg_backend_pid()')->fetchColumn(),
                $this->other->query('SELECT pg_backend_pid()')->fetchColumn(),
            );
            $jobs = new PdoImportJobRepository($this->pdo);
            $job = $jobs->create('/survives-cleanup', []);
            self::assertSame(0, (int) $isolated->query('SELECT COUNT(*) FROM import_jobs')->fetchColumn());
            unset($isolated);
            RepositoryDatabase::cleanup($path);
            $exists = $this->pdo->prepare('SELECT COUNT(*) FROM pg_namespace WHERE nspname = ?');
            $exists->execute([$schema]);
            self::assertSame(0, (int) $exists->fetchColumn());
            self::assertSame('/survives-cleanup', $jobs->get($job)['source_path']);
            // An unregistered path is never interpreted as a schema identifier.
            RepositoryDatabase::cleanup($schema);
        } finally {
            unset($isolated);
            RepositoryDatabase::cleanup($path);
        }
    }

    public function testOverlappingClaimsSkipLockedJobsAndReturnNullWhenAllAreLocked(): void
    {
        $jobs = new PdoImportJobRepository($this->pdo);
        $other = new PdoImportJobRepository($this->other);
        $first = $jobs->create('/first', []);
        $second = $jobs->create('/second', []);

        // The outer transaction must retain claimNext's row lock after it returns.
        $this->pdo->beginTransaction();
        self::assertSame($first, (int) $jobs->claimNext()['id']);
        self::assertTrue($this->pdo->inTransaction());
        self::assertSame('Pending', $other->get($first)['status']);
        self::assertSame($second, (int) $other->claimNext()['id']);
        self::assertNull($other->claimNext());
        self::assertTrue($this->pdo->inTransaction());
        $this->pdo->commit();
        self::assertSame('Copying', $other->get($first)['status']);
        self::assertSame('Copying', $other->get($second)['status']);
    }

    public function testRolledBackClaimCanBeReclaimedByTheOtherConnection(): void
    {
        $jobs = new PdoImportJobRepository($this->pdo);
        $other = new PdoImportJobRepository($this->other);
        $job = $jobs->create('/rollback', []);
        $this->pdo->beginTransaction();
        self::assertSame($job, (int) $jobs->claimNext()['id']);
        self::assertNull($other->claimNext());
        $this->pdo->rollBack();
        self::assertSame('Pending', $other->get($job)['status']);
        self::assertNull($other->get($job)['started_at']);
        self::assertSame($job, (int) $other->claimNext()['id']);
        self::assertNull($jobs->claimNext());
    }

    public function testTerminalNoOpStillAcquiresExplicitItemReadLock(): void
    {
        $jobs = new PdoImportJobRepository($this->pdo);
        $other = new PdoImportJobRepository($this->other);
        $job = $jobs->create('/terminal', [['source_path' => '/terminal/a', 'size' => 7]]);
        $item = (int) $jobs->items($job)[0]['id'];
        $jobs->finishItem($item, 'Completed');
        $this->pdo->beginTransaction();
        $statement = $this->pdo->prepare('SELECT id FROM import_items WHERE id = ? FOR SHARE');
        $statement->execute([$item]);
        $statement->closeCursor();
        try {
            // Already terminal: no UPDATE occurs, so only SELECT FOR UPDATE can block.
            $other->finishItem($item, 'Completed');
            self::fail('The explicit item read lock must conflict with FOR SHARE.');
        } catch (PDOException $error) {
            self::assertSame('55P03', $error->getCode());
        }
        self::assertFalse($this->other->inTransaction());
        $this->pdo->commit();
        $other->finishItem($item, 'Completed');
        self::assertSame(1, (int) $jobs->get($job)['processed_files']);
    }

    public function testItemLockTimesOutWithoutPublishingProgressAndCanBeRetried(): void
    {
        $jobs = new PdoImportJobRepository($this->pdo);
        $other = new PdoImportJobRepository($this->other);
        $job = $jobs->create('/item', [['source_path' => '/item/a', 'size' => 7]]);
        $jobs->claimNext();
        $item = (int) $jobs->items($job)[0]['id'];
        $this->pdo->beginTransaction();
        $jobs->planItem($item, 'a.stl', str_repeat('a', 64), 7);
        self::assertTrue($this->pdo->inTransaction());
        try {
            $other->finishItem($item, 'Completed');
            self::fail('A concurrent item transition must wait for the item lock.');
        } catch (PDOException $error) {
            self::assertSame('55P03', $error->getCode(), 'Expected PostgreSQL lock_not_available.');
        }
        self::assertFalse($this->other->inTransaction());
        self::assertSame('Pending', $other->items($job)[0]['status']);
        self::assertSame(0, (int) $other->get($job)['processed_files']);
        $this->pdo->commit();
        $other->finishItem($item, 'Completed');
        self::assertSame('Completed', $jobs->items($job)[0]['status']);
        self::assertSame(1, (int) $jobs->get($job)['processed_files']);
        self::assertSame(7, (int) $jobs->get($job)['copied_bytes']);
    }
}