Working with Queues 📬

<?php
// WHY queues exist. Imagine a signup that sends a welcome email.
// Sending email talks to a remote mail server, which can take 2-3 seconds.

// === The SLOW way: do the work DURING the request ===
function handleSignupBlocking(string $email): string
{
    // sendWelcomeEmail($email);  // <-- user waits here for 2-3 seconds
    return "Account created (after a slow wait)";
}

// === The FAST way: hand the slow work to a queue, respond instantly ===
$queue = [];                                  // pretend this is Redis/RabbitMQ

function handleSignupQueued(string $email, array &$queue): string
{
    // "Enqueue" = drop a small job onto the queue and move on immediately.
    $queue[] = ['type' => 'send_welcome_email', 'to' => $email];
    return "Account created (instantly)";     // user never waits for email
}

echo handleSignupBlocking('alice@example.com') . "\n";
echo handleSignupQueued('bob@example.com', $queue) . "\n";

// The job is now waiting in the queue for a worker to pick up later.
echo "Jobs waiting in queue: " . count($queue) . "\n";
?>

Account created (after a slow wait)
Account created (instantly)
Jobs waiting in queue: 1

<?php
// A "job" is a small, serialisable description of work to do LATER.
// Keep the PAYLOAD tiny — an id, not the whole object — so it survives being
// written to Redis as JSON and read back by a separate worker process.
class SendWelcomeEmail
{
    public function __construct(public int $userId) {}

    // handle() is where the actual work happens, run by the worker.
    public function handle(): void
    {
        // In real code you'd load the user and call your mailer here.
        echo "  -> emailing user #{$this->userId}\n";
    }

    // Turn the job into a string the broker can store...
    public function toJson(): string
    {
        return json_encode(['job' => static::class, 'userId' => $this->userId]);
    }

    // ...and back into a job object on the worker side.
    public static function fromJson(string $raw): self
    {
        $data = json_decode($raw, true);
        return new self($data['userId']);
    }
}

$job = new SendWelcomeEmail(42);
$line = $job->toJson();               // what gets pushed to the queue
echo "On the wire: {$line}\n";

$revived = SendWelcomeEmail::fromJson($line);   // worker rebuilds it
echo "Worker runs the job:\n";
$revived->handle();
?>

On the wire: {"job":"SendWelcomeEmail","userId":42}
Worker runs the job:
  -> emailing user #42

<?php
// 🎯 YOUR TURN — make the producer return INSTANTLY.
// Fill in each blank marked ___ , then run it and check the Output panel.

$queue = [];   // pretend this is Redis

function uploadAvatar(string $file, array &$queue): string
{
    // 1) Push a 'resize_image' job onto the queue instead of resizing now.
    $queue[] = ___;   // 👉 e.g.  ['type' => 'resize_image', 'file' => $file]

    // 2) Return straight away — the user shouldn't wait for the resize.
    return ___;       // 👉 e.g.  "Upload received"
}

echo uploadAvatar('avatar.jpg', $queue) . "\n";
echo "Jobs queued: " . count($queue) . "\n";

// ✅ Expected output:
//    Upload received
//    Jobs queued: 1
?>

Upload received
Jobs queued: 1

<?php
// A complete producer/worker pipeline you can run anywhere. A real system swaps
// the array for Redis, but the LIFECYCLE is identical: push -> pop -> handle.

class Job
{
    public int $attempts = 0;
    public function __construct(
        public string $id,            // a STABLE id, used for idempotency
        public string $type,
        public array $payload,
        public int $maxAttempts = 3,
    ) {}
}

$queue   = [];   // pending jobs (the broker)
$dead    = [];   // dead-letter queue: jobs that exhausted all retries
$done    = [];   // ids we've already completed (idempotency guard)

// === PRODUCER: push jobs, then return to the user instantly ===
function enqueue(array &$queue, Job $job): void {
    $queue[] = $job;
    echo "queued {$job->type} (#{$job->id})\n";
}

enqueue($queue, new Job('e1', 'send_email',   ['to' => 'alice']));
enqueue($queue, new Job('i1', 'resize_image', ['file' => 'avatar.jpg']));
enqueue($queue, new Job('p1', 'charge_card',  ['amount' => 999]));

// Pretend charge_card fails twice (flaky network) then succeeds.
$failsLeft = ['p1' => 2];

// The worker's "do the work" step. Returns true on success.
function handle(Job $job, array &$done, array &$failsLeft): bool {
    // IDEMPOTENCY: if we've already done this exact job id, skip it.
    if (in_array($job->id, $done, true)) {
        echo "  skip {$job->id} (already done)\n";
        return true;
    }
    if (($failsLeft[$job->id] ?? 0) > 0) {
        $failsLeft[$job->id]--;
        return false;                 // simulate a transient failure
    }
    $done[] = $job->id;               // record success so retries are safe
    return true;
}

// === WORKER LOOP: pop a job, run it, retry or dead-letter on failure ===
echo "\nworker starting...\n";
while (!empty($queue)) {
    $job = array_shift($queue);       // brPop in Redis; pop here
    $job->attempts++;

    if (handle($job, $done, $failsLeft)) {
        echo "  ok {$job->type} (attempt {$job->attempts})\n";
    } elseif ($job->attempts >= $job->maxAttempts) {
        $dead[] = $job;               // give up -> dead-letter for a human
        echo "  DEAD {$job->type} after {$job->attempts} attempts\n";
    } else {
        // BACKOFF: wait longer before each retry (1s, 2s, 4s...). We just
        // print it here; a real worker would re-queue with a delay.
        $backoff = 2 ** ($job->attempts - 1);
        echo "  retry {$job->type} in {$backoff}s\n";
        $queue[] = $job;              // put it back for another go
    }
}

echo "\ndead-letter queue: " . count($dead) . " job(s)\n";
?>

queued send_email (#e1)
queued resize_image (#i1)
queued charge_card (#p1)

worker starting...
  ok send_email (attempt 1)
  ok resize_image (attempt 1)
  retry charge_card in 1s
  retry charge_card in 2s
  ok charge_card (attempt 3)

dead-letter queue: 0 job(s)

<?php
// 🎯 YOUR TURN — finish the retry rule in the worker loop.
// A job should retry while it has attempts left, and dead-letter once it runs out.

$job  = ['type' => 'charge_card', 'attempts' => 0, 'maxAttempts' => 3];
$dead = [];
$fails = 5;   // this job always fails, so it must end up dead-lettered

while (true) {
    $job['attempts']++;
    $succeeded = $fails-- <= 0;          // stays false here — always fails
    if ($succeeded) {
        echo "ok\n";
        break;
    }

    // 👉 Stop retrying once attempts reach maxAttempts: send it to dead-letter.
    if ($job['attempts'] ___ $job['maxAttempts']) {   // 👉 use the >= operator
        $dead[] = $job;
        echo "DEAD after {$job['attempts']} attempts\n";
        break;
    }
    echo "retry {$job['attempts']}\n";
}

echo "dead-letter: " . count($dead) . "\n";

// ✅ Expected output:
//    retry 1
//    retry 2
//    DEAD after 3 attempts
//    dead-letter: 1
?>

retry 1
retry 2
DEAD after 3 attempts
dead-letter: 1

<?php
// The SAME lifecycle, now durable across processes. Needs a Redis server and
// the phpredis extension. The producer runs in your web app; the worker runs
// as a separate long-lived process (see Supervisor below).
$redis = new Redis();
$redis->connect('127.0.0.1', 6379);

// === Producer (inside your web request): push, then respond instantly ===
$redis->lPush('queue:emails', json_encode([
    'id'    => 'welcome-42',          // stable id for idempotency
    'job'   => 'send_welcome_email',
    'to'    => 'alice@example.com',
]));

// === Consumer (worker.php — its own process) ===
while (true) {
    // brPop BLOCKS up to 30s waiting for a job, so the worker doesn't busy-loop
    // burning CPU. It returns [list_name, value] or null on timeout.
    $popped = $redis->brPop(['queue:emails'], 30);
    if (empty($popped)) {
        continue;                     // timed out — loop and wait again
    }

    $data = json_decode($popped[1], true);

    // IDEMPOTENCY: setnx returns false if this id was already processed.
    if (!$redis->setnx("done:{$data['id']}", 1)) {
        continue;                     // duplicate delivery — skip safely
    }

    try {
        sendWelcomeEmail($data['to']);            // your real work
    } catch (Throwable $e) {
        $redis->del("done:{$data['id']}");        // allow a retry
        // DEAD-LETTER: park the raw job for a human to inspect later.
        $redis->lPush('queue:emails:dead', $popped[1]);
        error_log("job failed: {$e->getMessage()}");
    }
}
?>

; /etc/supervisor/conf.d/email-worker.conf
[program:email-worker]
command=php /var/www/worker.php
numprocs=3            ; run 3 parallel workers
process_name=%(program_name)s_%(process_num)02d
autostart=true
autorestart=true     ; restart a worker if it crashes
stopwaitsecs=30      ; let an in-flight job finish before stopping
stderr_logfile=/var/log/email-worker.err.log
stdout_logfile=/var/log/email-worker.out.log