<?php

namespace Tests\Unit;

use App\Console\Commands\SqsDrainCommand;
use Illuminate\Queue\QueueManager;
use Illuminate\Queue\Worker;
use Illuminate\Queue\WorkerOptions;
use Illuminate\Support\Facades\Cache;
use Tests\TestCase;

/**
 * Feature: async-sqs-job-trigger
 *
 * Drain loop properties, driven by a scripted fake queue connection so no AWS
 * call and no real sleep is involved.
 */
class SqsDrainCommandTest extends TestCase
{
    protected function setUp()
    {
        parent::setUp();
        Cache::forget(SqsDrainCommand::LOCK_KEY);
    }

    protected function tearDown()
    {
        Cache::forget(SqsDrainCommand::LOCK_KEY);
        parent::tearDown();
    }

    /**
     * The command's executable source with comments stripped, so that assertions
     * about what the code calls are not satisfied or broken by prose in comments.
     *
     * @return string
     */
    private function commandCode()
    {
        return php_strip_whitespace((new \ReflectionClass(SqsDrainCommand::class))->getFileName());
    }

    /**
     * Run a drain over a scripted sequence of pop() results.
     *
     * Each script entry is either a FakeJob, null (empty poll), or an Exception
     * (a poll that throws). The script answers "empty" once exhausted so no test
     * can hang.
     *
     * @return array
     */
    private function drain(array $script, array $failingJobs = [])
    {
        $connection = new FakeSqsConnection($script);
        $worker = new FakeWorker($failingJobs);

        $manager = \Mockery::mock(QueueManager::class);
        $manager->shouldReceive('connection')->with('sqs')->andReturn($connection);

        $command = new SqsDrainCommandSpy();
        $exit = $command->handle($manager, $worker);

        return ['exit' => $exit, 'worker' => $worker, 'connection' => $connection, 'command' => $command];
    }

    /**
     * Feature: async-sqs-job-trigger, Property 10: The drain loop terminates exactly
     * at the empty-poll threshold; a job found and a job that throws both reset the
     * consecutive-empty count to zero.
     */
    public function testLoopTerminatesExactlyAtTheEmptyPollThreshold()
    {
        $threshold = SqsDrainCommand::EMPTY_POLL_THRESHOLD;

        // Fewer than the threshold of empties interleaved with a job must not stop.
        $script = [];
        for ($i = 0; $i < $threshold - 1; $i++) {
            $script[] = null;
        }
        $script[] = new FakeJob('App\Jobs\Interleaved');
        for ($i = 0; $i < $threshold; $i++) {
            $script[] = null;
        }

        $result = $this->drain($script);

        $this->assertSame(1, $result['worker']->processedCount, 'The interleaved job must be processed.');
        $this->assertSame(0, $result['exit']);

        // The loop must stop on the threshold-th consecutive empty, not later.
        $expectedPolls = ($threshold - 1) + 1 + $threshold;
        $this->assertSame($expectedPolls, $result['connection']->popCount);
    }

    /**
     * Feature: async-sqs-job-trigger, Property 14: A clean drain exits zero.
     */
    public function testCleanDrainExitsZero()
    {
        $result = $this->drain([
            new FakeJob('App\Jobs\A'),
            new FakeJob('App\Jobs\B'),
            new FakeJob('App\Jobs\C'),
            null, null, null,
        ]);

        $this->assertSame(0, $result['exit']);
        $this->assertSame(3, $result['worker']->processedCount);
    }

    /**
     * Feature: async-sqs-job-trigger, Property 12: A failing job never halts the
     * drain - the loop logs and continues to the next poll.
     */
    public function testFailingJobDoesNotHaltTheDrain()
    {
        $result = $this->drain(
            [
                new FakeJob('App\Jobs\Before'),
                new FakeJob('App\Jobs\Poison'),
                new FakeJob('App\Jobs\After'),
                null, null, null,
            ],
            ['App\Jobs\Poison']
        );

        $this->assertSame(0, $result['exit'], 'A failing job must not change the exit status.');
        $this->assertSame(3, $result['worker']->processedCount, 'Jobs after the failure must still run.');
        $this->assertContains('App\Jobs\After', $result['worker']->processedNames);
    }

    /**
     * Feature: async-sqs-job-trigger, Property 13: Repeated poll failures terminate
     * the drain non-zero.
     */
    public function testRepeatedPollFailuresTerminateNonZero()
    {
        $threshold = SqsDrainCommand::POP_FAILURE_THRESHOLD;

        $script = [];
        for ($i = 0; $i < $threshold; $i++) {
            $script[] = new \RuntimeException('sqs unreachable');
        }

        $result = $this->drain($script);

        $this->assertSame(1, $result['exit'], 'Exhausting the poll-failure threshold must exit non-zero.');
        $this->assertSame($threshold, $result['connection']->popCount);
    }

    /**
     * Feature: async-sqs-job-trigger, Property 13 (reset): a poll that completes
     * without raising resets the consecutive-failure count, so interleaved failures
     * below the threshold do not abort the drain.
     */
    public function testSuccessfulPollResetsTheFailureCount()
    {
        $threshold = SqsDrainCommand::POP_FAILURE_THRESHOLD;

        $script = [];
        for ($i = 0; $i < $threshold - 1; $i++) {
            $script[] = new \RuntimeException('transient');
        }
        $script[] = new FakeJob('App\Jobs\Recovered');
        for ($i = 0; $i < $threshold - 1; $i++) {
            $script[] = new \RuntimeException('transient again');
        }
        $script[] = null;
        $script[] = null;
        $script[] = null;

        $result = $this->drain($script);

        $this->assertSame(0, $result['exit'], 'Failures below the threshold must not abort the drain.');
        $this->assertSame(1, $result['worker']->processedCount);
    }

    /**
     * Feature: async-sqs-job-trigger, Property 11: Queue emptiness is decided only
     * by consecutive empty polls, never by the reported queue size.
     */
    public function testEmptinessIgnoresReportedQueueSize()
    {
        // size() lies in both directions; the drain must not consult it at all.
        foreach ([0, 999] as $reportedSize) {
            $connection = new FakeSqsConnection([null, null, null], $reportedSize);
            $worker = new FakeWorker();

            $manager = \Mockery::mock(QueueManager::class);
            $manager->shouldReceive('connection')->with('sqs')->andReturn($connection);

            $exit = (new SqsDrainCommandSpy())->handle($manager, $worker);

            $this->assertSame(0, $exit);
            $this->assertSame(0, $connection->sizeCount, 'size() must never be called.');
        }

        $this->assertNotContains('->size(', $this->commandCode());
    }

    /**
     * Feature: async-sqs-job-trigger, Property 16: Every job is processed through the
     * worker pipeline with the connection name 'sqs', so queue events are dispatched
     * and the Queue::before hook in AppServiceProvider still fires.
     */
    public function testJobsAreProcessedThroughTheWorkerOnTheSqsConnection()
    {
        $result = $this->drain([new FakeJob('App\Jobs\X'), null, null, null]);

        $this->assertSame(['sqs'], array_unique($result['worker']->connectionNames));

        // maxTries must be 1 so a failing job is failed and deleted, not requeued.
        $this->assertSame(1, $result['worker']->options->maxTries);
        $this->assertSame(0, $result['worker']->options->delay);

        // fire() must not be called directly: that would skip JobProcessing/JobProcessed.
        $this->assertNotContains('->fire()', $this->commandCode());
    }

    /**
     * Feature: async-sqs-job-trigger, Property 15: Thresholds come from
     * configuration with defaults of 3 and 5.
     */
    public function testThresholdsComeFromConfigurationWithDefaults()
    {
        $this->assertSame(3, SqsDrainCommand::EMPTY_POLL_THRESHOLD);
        $this->assertSame(5, SqsDrainCommand::POP_FAILURE_THRESHOLD);

        // Default (nothing configured).
        config(['queue.sqs_drain' => null]);
        $command = new SqsDrainCommandSpy();
        $this->assertSame(3, $command->exposedEmptyPollThreshold());
        $this->assertSame(5, $command->exposedPopFailureThreshold());

        // Configured override is honoured.
        config(['queue.sqs_drain.empty_poll_threshold' => 7, 'queue.sqs_drain.pop_failure_threshold' => 2]);
        $this->assertSame(7, $command->exposedEmptyPollThreshold());
        $this->assertSame(2, $command->exposedPopFailureThreshold());

        // empty_poll_threshold is 7 here, so the drain must poll 7 times before stopping.
        $result = $this->drain([null, null]);
        $this->assertSame(7, $result['connection']->popCount, 'The configured empty-poll threshold must be honoured.');

        // And the configured poll-failure threshold of 2 must abort after 2 throws.
        $result = $this->drain([new \RuntimeException('a'), new \RuntimeException('b')]);
        $this->assertSame(1, $result['exit']);
        $this->assertSame(2, $result['connection']->popCount);

        config(['queue.sqs_drain' => null]);
    }

    /**
     * Feature: async-sqs-job-trigger, Property 9: the lock is released when the drain
     * ends - on a clean drain, on the poll-failure abort, and on an unexpected throw
     * during setup.
     */
    public function testLockIsReleasedOnEveryExitPath()
    {
        Cache::put(SqsDrainCommand::LOCK_KEY, true, SqsDrainCommand::LOCK_MINUTES);
        $this->drain([null, null, null]);
        $this->assertFalse(Cache::has(SqsDrainCommand::LOCK_KEY), 'Clean drain must release the lock.');

        Cache::put(SqsDrainCommand::LOCK_KEY, true, SqsDrainCommand::LOCK_MINUTES);
        $failures = array_fill(0, SqsDrainCommand::POP_FAILURE_THRESHOLD, new \RuntimeException('down'));
        $this->drain($failures);
        $this->assertFalse(Cache::has(SqsDrainCommand::LOCK_KEY), 'Aborted drain must release the lock.');

        // Setup itself failing must still release, which is why connection resolution
        // lives inside the try block rather than before it.
        Cache::put(SqsDrainCommand::LOCK_KEY, true, SqsDrainCommand::LOCK_MINUTES);
        $manager = \Mockery::mock(QueueManager::class);
        $manager->shouldReceive('connection')->with('sqs')->andThrow(new \RuntimeException('resolve failed'));

        try {
            (new SqsDrainCommandSpy())->handle($manager, new FakeWorker());
        } catch (\Exception $e) {
            // expected
        }

        $this->assertFalse(
            Cache::has(SqsDrainCommand::LOCK_KEY),
            'An unexpected throw must still release the lock.'
        );
    }

    /**
     * Feature: async-sqs-job-trigger, Property 19: Every drain run is traceable -
     * info at start, debug per job, info at the end with count and elapsed time.
     *
     * Feature: async-sqs-job-trigger, Property 20: Credentials never reach the logs,
     * and neither do raw job bodies.
     */
    public function testDrainRunIsTraceableAndCredentialFree()
    {
        $entries = [];
        $collect = function ($level) use (&$entries) {
            return function ($message, $context = []) use ($level, &$entries) {
                $entries[] = ['level' => $level, 'message' => $message, 'context' => $context];
            };
        };

        \Illuminate\Support\Facades\Log::shouldReceive('info')->andReturnUsing($collect('info'));
        \Illuminate\Support\Facades\Log::shouldReceive('debug')->andReturnUsing($collect('debug'));
        \Illuminate\Support\Facades\Log::shouldReceive('error')->andReturnUsing($collect('error'));

        $this->drain([new FakeJob('App\Jobs\Traced'), null, null, null]);

        $start = $entries[0];
        $this->assertSame('info', $start['level']);
        $this->assertSame('sqs', $start['context']['connection']);
        $this->assertArrayHasKey('empty_poll_threshold', $start['context']);

        $debug = array_values(array_filter($entries, function ($entry) {
            return $entry['level'] === 'debug';
        }));
        $this->assertSame('App\Jobs\Traced', $debug[0]['context']['job']);
        $this->assertSame(1, $debug[0]['context']['processed']);

        $finish = end($entries);
        $this->assertSame('info', $finish['level']);
        $this->assertSame(1, $finish['context']['processed']);
        $this->assertArrayHasKey('elapsed_seconds', $finish['context']);

        // No raw job bodies and no credentials.
        $serialised = json_encode($entries);
        $this->assertNotContains('raw-body', $serialised);

        foreach ([config('queue.connections.sqs.key'), config('queue.connections.sqs.secret')] as $credential) {
            if (is_string($credential) && strlen($credential) > 4) {
                $this->assertNotContains($credential, $serialised);
            }
        }
    }

    /**
     * Feature: async-sqs-job-trigger, Property 21: The drain uses no request-scoped
     * state - every dependency is injected from the console container.
     */
    public function testDrainUsesNoRequestScopedState()
    {
        $handle = new \ReflectionMethod(SqsDrainCommand::class, 'handle');
        $types = [];

        foreach ($handle->getParameters() as $parameter) {
            $types[] = $parameter->getClass()->getName();
        }

        $this->assertSame([QueueManager::class, Worker::class], $types);

        foreach (['request(', 'Request $', 'session(', '$_GET', '$_POST'] as $forbidden) {
            $this->assertNotContains($forbidden, $this->commandCode());
        }
    }
}

/**
 * Removes the real sleep so the loop runs at full speed in tests.
 */
class SqsDrainCommandSpy extends SqsDrainCommand
{
    public $pauseCount = 0;

    protected function pause()
    {
        $this->pauseCount++;
    }

    public function exposedEmptyPollThreshold()
    {
        return $this->emptyPollThreshold();
    }

    public function exposedPopFailureThreshold()
    {
        return $this->popFailureThreshold();
    }
}

/**
 * A queue connection whose pop() replays a script.
 */
class FakeSqsConnection
{
    public $popCount = 0;
    public $sizeCount = 0;

    private $script;
    private $reportedSize;

    public function __construct(array $script, $reportedSize = 0)
    {
        $this->script = $script;
        $this->reportedSize = $reportedSize;
    }

    public function pop($queue = null)
    {
        $this->popCount++;

        $entry = array_shift($this->script);

        // Once the script runs out, keep answering "empty" so no test can hang.
        if (empty($this->script)) {
            $this->script = [null];
        }

        if ($entry instanceof \Exception) {
            throw $entry;
        }

        return $entry;
    }

    public function size($queue = null)
    {
        $this->sizeCount++;

        return $this->reportedSize;
    }
}

class FakeJob
{
    private $name;

    public function __construct($name)
    {
        $this->name = $name;
    }

    public function resolveName()
    {
        return $this->name;
    }

    public function getRawBody()
    {
        return 'raw-body';
    }
}

/**
 * Records process() calls and throws for nominated job names, mirroring the real
 * Worker::process()'s rethrow behaviour.
 */
class FakeWorker extends Worker
{
    public $processedCount = 0;
    public $processedNames = [];
    public $connectionNames = [];
    public $options;

    private $failingJobs;

    public function __construct(array $failingJobs = [])
    {
        // Deliberately not calling parent::__construct(): this fake never touches
        // the real manager, dispatcher or exception handler.
        $this->failingJobs = $failingJobs;
    }

    public function process($connectionName, $job, WorkerOptions $options)
    {
        $this->processedCount++;
        $this->processedNames[] = $job->resolveName();
        $this->connectionNames[] = $connectionName;
        $this->options = $options;

        if (in_array($job->resolveName(), $this->failingJobs, true)) {
            // Worker::handleJobException() always rethrows, so the real worker does too.
            throw new \RuntimeException('job failed');
        }
    }
}