<?php

namespace App\Console\Commands;

use Illuminate\Console\Command;
use Illuminate\Queue\QueueManager;
use Illuminate\Queue\Worker;
use Illuminate\Queue\WorkerOptions;
use Illuminate\Support\Facades\Cache;
use Illuminate\Support\Facades\Log;

/**
 * Processes every queued job on the "sqs" connection and then exits.
 *
 * This exists because laravel/framework 5.5 has no drain mode for queue:work:
 * --max-time and --max-jobs arrived in 6.0 and --stop-when-empty in 5.8, so the
 * termination condition has to live in application code. --once is not usable
 * either because a single trigger must clear the whole queue.
 *
 * Launched in the background by TestingController@runSQSJobById, but also safe
 * to run in the foreground.
 */
class SqsDrainCommand extends Command
{
    /**
     * Cache key representing an in-flight drain.
     *
     * Shared with TestingController@runSQSJobById, which acquires this lock
     * before launching the drain. It is declared here, in one place, so the two
     * halves cannot drift apart and silently disable mutual exclusion.
     */
    const LOCK_KEY = 'sqs_drain_running';

    /**
     * Lock time-to-live in MINUTES.
     *
     * Laravel 5.5's Cache::add()/put() take minutes, not seconds (second
     * granularity arrived in 5.8), so 2 == 120s. The TTL only bounds how long a
     * dead drain can block new triggers; a live drain holds the lock
     * indefinitely by refreshing it every HEARTBEAT_SECONDS.
     */
    const LOCK_MINUTES = 2;

    /** Seconds between lock refreshes. 30s against a 120s TTL leaves a 90s margin. */
    const HEARTBEAT_SECONDS = 30;

    /** Consecutive empty polls that mean "queue is empty". */
    const EMPTY_POLL_THRESHOLD = 3;

    /** Consecutive pop() exceptions after which the drain gives up. */
    const POP_FAILURE_THRESHOLD = 5;

    /** Seconds to wait between polls once the queue looks empty. */
    const EMPTY_POLL_SLEEP_SECONDS = 1;

    /** The queue connection this command drains. */
    const CONNECTION_NAME = 'sqs';

    /**
     * @var string
     */
    protected $signature = 'sqs:drain';

    /**
     * @var string
     */
    protected $description = 'Process every queued job on the sqs connection, then exit.';

    /**
     * Unix timestamp of the last lock refresh.
     *
     * @var int
     */
    protected $lastHeartbeatAt = 0;

    /**
     * Execute the console command.
     *
     * Dependencies are injected from the console container so that nothing
     * request-scoped (request instance, session, request-scoped transaction) is
     * reachable from the drain loop.
     *
     * @param  \Illuminate\Queue\QueueManager  $manager
     * @param  \Illuminate\Queue\Worker  $worker
     * @return int  0 on a clean drain, 1 when the queue could not be polled
     */
    public function handle(QueueManager $manager, Worker $worker)
    {
        $emptyPolls = 0;
        $popFailures = 0;
        $processed = 0;
        $exitCode = 0;
        $startedAt = microtime(true);
        $this->lastHeartbeatAt = time();

        // Everything below sits inside the try so that the finally releases the lock
        // even if setup itself fails. Resolving the connection outside it would leave
        // the lock held until its TTL whenever connection resolution threw.
        try {
            $emptyPollThreshold = $this->emptyPollThreshold();
            $popFailureThreshold = $this->popFailureThreshold();

            $connection = $manager->connection(self::CONNECTION_NAME);
            $queueName = $this->queueName();

            // WorkerOptions on 5.5 is ($delay, $memory, $timeout, $sleep, $maxTries, $force).
            // maxTries = 1 so a failing job is recorded in failed_jobs and deleted rather
            // than released back onto the queue, which would keep this loop alive forever.
            // delay must stay 0: Worker::handleJobException() releases an un-failed job with
            // release($options->delay), and a non-zero delay would hide that message from
            // the empty-poll counting below.
            $options = new WorkerOptions(0, 128, 60, 3, 1, false);

            Log::info('SQS drain started', [
                'connection' => self::CONNECTION_NAME,
                'queue' => $queueName,
                'empty_poll_threshold' => $emptyPollThreshold,
            ]);

            while (true) {
                try {
                    $job = $connection->pop($queueName);
                    $popFailures = 0;
                } catch (\Exception $e) {
                    $popFailures++;

                    Log::error('SQS drain failed to poll the queue', [
                        'message' => $e->getMessage(),
                        'consecutive_failures' => $popFailures,
                    ]);

                    if ($popFailures >= $popFailureThreshold) {
                        $exitCode = 1;
                        break;
                    }

                    $this->pause();
                    $this->heartbeat();
                    continue;
                }

                if (is_null($job)) {
                    $emptyPolls++;

                    if ($emptyPolls >= $emptyPollThreshold) {
                        break;
                    }

                    $this->pause();
                    $this->heartbeat();
                    continue;
                }

                $emptyPolls = 0;
                $processed++;

                try {
                    // Must go through Worker::process() rather than $job->fire():
                    // process() raises JobProcessing/JobProcessed, which the
                    // Queue::before hook in AppServiceProvider listens for. That hook
                    // keys on $event->connectionName === 'sqs', hence the literal
                    // connection name here.
                    $worker->process(self::CONNECTION_NAME, $job, $options);

                    Log::debug('SQS drain processed a job', [
                        'job' => $this->jobName($job),
                        'processed' => $processed,
                    ]);
                } catch (\Exception $e) {
                    // Worker::handleJobException() always rethrows, so this catch is
                    // what keeps one bad message from aborting the whole drain. The job
                    // has already been failed and deleted by FailingJob at maxTries=1.
                    Log::error('SQS drain job failed', [
                        'job' => $this->jobName($job),
                        'processed' => $processed,
                        'message' => $e->getMessage(),
                    ]);
                }

                $this->heartbeat();
            }
        } finally {
            Cache::forget(self::LOCK_KEY);

            Log::info('SQS drain finished', [
                'processed' => $processed,
                'elapsed_seconds' => round(microtime(true) - $startedAt, 2),
                'exit_code' => $exitCode,
            ]);
        }

        return $exitCode;
    }

    /**
     * Wait between polls so an empty queue does not spin the CPU.
     *
     * Extracted as a seam so tests can drain without sleeping in real time.
     *
     * @return void
     */
    protected function pause()
    {
        sleep(self::EMPTY_POLL_SLEEP_SECONDS);
    }

    /**
     * Refresh the lock so a long drain keeps holding it.
     *
     * Rate limited to one cache write per HEARTBEAT_SECONDS so that a drain
     * processing many fast jobs does not write on every iteration.
     *
     * @return void
     */
    protected function heartbeat()
    {
        if ((time() - $this->lastHeartbeatAt) < self::HEARTBEAT_SECONDS) {
            return;
        }

        Cache::put(self::LOCK_KEY, true, self::LOCK_MINUTES);
        $this->lastHeartbeatAt = time();
    }

    /**
     * @return int
     */
    protected function emptyPollThreshold()
    {
        return (int) config('queue.sqs_drain.empty_poll_threshold', self::EMPTY_POLL_THRESHOLD)
            ?: self::EMPTY_POLL_THRESHOLD;
    }

    /**
     * @return int
     */
    protected function popFailureThreshold()
    {
        return (int) config('queue.sqs_drain.pop_failure_threshold', self::POP_FAILURE_THRESHOLD)
            ?: self::POP_FAILURE_THRESHOLD;
    }

    /**
     * @return string|null
     */
    protected function queueName()
    {
        return config('queue.connections.'.self::CONNECTION_NAME.'.queue');
    }

    /**
     * Resolve a loggable name for a job.
     *
     * Deliberately the job class name only. The raw payload is never logged:
     * job bodies can carry sensitive application data.
     *
     * @param  mixed  $job
     * @return string
     */
    protected function jobName($job)
    {
        try {
            return method_exists($job, 'resolveName') ? (string) $job->resolveName() : get_class($job);
        } catch (\Exception $e) {
            return 'unknown';
        }
    }
}