<?php

declare(strict_types=1);

namespace Apexl\Io\Jobs\Tests;

use Apexl\Io\Collections\LoaderConditionCollection;
use Apexl\Io\Dtos\LoaderCondition;
use Apexl\Io\Jobs\Entities\QueueJob;
use Apexl\Io\Jobs\Factories\QueueJobFactory;
use Apexl\Io\TestInfrastructure\FeatureTest;
use Faker\Generator;
use PHPUnit\Framework\Attributes\Test;

/**
 * Two enqueues of the same handler racing for a sequence number.
 *
 * QueueJobFactory::nextSequence() is an unlocked SELECT MAX(sequence)+1 and
 * the insert that follows is a separate statement, so two processes read the
 * same number and both write it. While (sequence, handler) was UNIQUE that
 * collision destroyed a job: under apexl/vault <=3 every insert was emitted as
 * INSERT ... ON DUPLICATE KEY UPDATE, so the second write silently overwrote
 * the first — contents and all, no exception, nothing in the log. Under vault
 * >=4 the insert is strict and throws at enqueue instead. Either way a job
 * that was handed to the queue never ran.
 *
 * The interleaving is reproduced by hand rather than by threads: both reads
 * happen before either write, which is exactly the ordering that loses a job,
 * and it happens on every run instead of under load. A genuinely concurrent
 * version would need a second connection, and the harness wraps each test in a
 * transaction the second connection could not see into.
 */
final class QueueJobSequenceCollisionTest extends FeatureTest
{
    protected Generator $faker;

    private function factory(): QueueJobFactory
    {
        return \Apexl\Io\container()->get(QueueJobFactory::class);
    }

    /** @return QueueJob[] every job stored against $handler, in insertion order */
    private function jobsFor(string $handler): array
    {
        $cond = new LoaderConditionCollection();
        $cond->add(LoaderCondition::equals('handler', $handler));

        $jobs = QueueJob::findMultiple(conditions: $cond)->toArray();
        usort($jobs, static fn (QueueJob $a, QueueJob $b): int => (int) $a->id <=> (int) $b->id);

        return $jobs;
    }

    #[Test]
    public function twoEnqueuesThatComputeTheSameSequenceBothSurvive(): void
    {
        $handler = 'RacingHandler\\' . bin2hex(random_bytes(6));

        // Both processes read the max before either has written.
        $first  = $this->factory()->nextSequence($handler);
        $second = $this->factory()->nextSequence($handler);
        self::assertSame($first, $second, 'The race being reproduced: one number, two claimants.');

        $a = $this->factory()->make($handler, ['job' => 'a'], sequence: $first);
        $b = $this->factory()->make($handler, ['job' => 'b'], sequence: $second);

        $stored = $this->jobsFor($handler);

        self::assertCount(2, $stored, 'Neither job may be overwritten by the other.');
        self::assertNotSame((int) $a->id, (int) $b->id);
        self::assertSame(
            [['job' => 'a'], ['job' => 'b']],
            array_map(static fn (QueueJob $j): mixed => $j->contents, $stored),
            'Each job keeps its own contents — the overwrite swapped them.',
        );
        self::assertSame(
            [$first, $first],
            array_map(static fn (QueueJob $j): int => (int) $j->sequence, $stored),
            'They share the sequence; id is what tells them apart.',
        );
    }

    /**
     * The finder orders by (sequence, id), so two jobs sharing a number still
     * run in arrival order rather than an arbitrary one.
     */
    #[Test]
    public function collidingJobsStillRunInArrivalOrder(): void
    {
        $handler = 'RacingHandler\\' . bin2hex(random_bytes(6));
        $seq     = $this->factory()->nextSequence($handler);

        $first  = $this->factory()->make($handler, ['job' => 'first'], sequence: $seq);
        $second = $this->factory()->make($handler, ['job' => 'second'], sequence: $seq);

        self::assertSame(
            [(int) $first->id, (int) $second->id],
            array_map(static fn (QueueJob $j): int => (int) $j->id, $this->jobsFor($handler)),
        );
        self::assertLessThan((int) $second->id, (int) $first->id);
    }

    /** Uncontended allocation is unchanged: still one more than the handler's highest. */
    #[Test]
    public function sequentialEnqueuesStillIncrement(): void
    {
        $handler = 'OrderlyHandler\\' . bin2hex(random_bytes(6));

        $one   = $this->factory()->make($handler, ['n' => 1]);
        $two   = $this->factory()->make($handler, ['n' => 2]);
        $three = $this->factory()->make($handler, ['n' => 3]);

        self::assertSame(
            [(int) $one->sequence + 1, (int) $one->sequence + 2],
            [(int) $two->sequence, (int) $three->sequence],
        );
    }

    /** Sequences are per handler, so a busy handler never pushes a quiet one's numbering along. */
    #[Test]
    public function sequencesAreAllocatedPerHandler(): void
    {
        $busy  = 'BusyHandler\\' . bin2hex(random_bytes(6));
        $quiet = 'QuietHandler\\' . bin2hex(random_bytes(6));

        $this->factory()->make($busy, ['n' => 1]);
        $this->factory()->make($busy, ['n' => 2]);

        self::assertSame(1, (int) $this->factory()->make($quiet, ['n' => 1])->sequence);
    }
}
