CodeIgniter 4 CLI Queue Advanced

Tanpa Redis, Tanpa Laravel — Bikin Queue System Sendiri di CI4 Pakai Spark Command

Khalim CLI & Architecture 14 menit baca

Kirim email verifikasi, proses invoice, generate laporan PDF besar — semua ini gak boleh bikin user nunggu di browser. Solusinya: queue. Tapi banyak developer CI4 langsung nyerah karena ngerasa butuh Redis, RabbitMQ, atau pindah ke Laravel.

Kenyataannya? Kamu bisa bikin queue system yang solid di CI4 dengan modal tabel database dan Spark CLI Command. Ini yang gue pakai di beberapa project production skala menengah — dan works.

Cara Kerjanya

TRIGGER

Controller / Event

→

JOBS TABLE

Database queue

→

SPARK WORKER

php spark queue:work

→

EXECUTED

Job selesai / retry

Controller push job ke tabel database. Spark worker jalan di background, ambil job satu per satu, eksekusi, lalu tandai selesai. Kalau gagal, otomatis di-retry sampai batas maksimum.

Implementasi

01 Buat tabel jobs sebagai queue storage

Semua job yang di-queue disimpan di tabel ini. Field payload nyimpen serialized data, attempts nyatat berapa kali udah dicoba, dan available_at buat delay job.

bash

php spark make:migration CreateJobsTable

app/Database/Migrations/..._CreateJobsTable.php

public function up()
{
    $this->forge->addField([
        'id'           => ['type' => 'BIGINT', 'auto_increment' => true],
        'queue'        => ['type' => 'VARCHAR', 'constraint' => 100, 'default' => 'default'],
        'payload'      => ['type' => 'LONGTEXT'],
        'attempts'     => ['type' => 'TINYINT', 'default' => 0],
        'max_attempts' => ['type' => 'TINYINT', 'default' => 3],
        'status'       => ['type' => 'ENUM', 'constraint' => ['pending', 'processing', 'done', 'failed'], 'default' => 'pending'],
        'available_at' => ['type' => 'DATETIME'],
        'created_at'   => ['type' => 'DATETIME', 'null' => true],
        'failed_at'    => ['type' => 'DATETIME', 'null' => true],
        'error'        => ['type' => 'TEXT', 'null' => true],
    ]);
    $this->forge->addPrimaryKey('id');
    $this->forge->addKey(['queue', 'status', 'available_at']);
    $this->forge->createTable('jobs');
}

bash

php spark migrate

02 Buat interface JobInterface sebagai kontrak job

Setiap job class harus implement interface ini. Satu method: handle(). Simpel dan konsisten — worker gak perlu tau jenis job-nya, cukup panggil handle().

app/Interfaces/JobInterface.php

namespace App\Interfaces;

interface JobInterface
{
    public function handle(): void;
}

03 Buat Job class — contoh: kirim email verifikasi

Job class cuma bertanggung jawab melakukan satu hal. Data yang dibutuhkan di-pass via constructor dan disimpan sebagai property — ini yang akan di-serialize ke database.

app/Jobs/SendVerificationEmailJob.php

namespace App\Jobs;

use App\Interfaces\JobInterface;

class SendVerificationEmailJob implements JobInterface
{
    public function __construct(
        private readonly string $email,
        private readonly string $name,
        private readonly string $token,
    ) {}

    public function handle(): void
    {
        $emailSvc = service('email');

        $emailSvc
            ->setTo($this->email)
            ->setSubject('Verifikasi Akun Kamu')
            ->setMessage(view('emails/verification', [
                'name'  => $this->name,
                'token' => $this->token,
            ]));

        if (!$emailSvc->send()) {
            throw new \RuntimeException($emailSvc->printDebugger(['headers']));
        }
    }
}

💡

Kalau handle() throw Exception, worker otomatis increment attempts dan reschedule job. Setelah max_attempts tercapai, status jadi failed.

04 Buat Queue dispatcher — satu baris push job

Kelas ini yang dipanggil dari Controller atau Service untuk push job ke database. Payload-nya JSON dari serialized job object.

app/Libraries/Queue.php

namespace App\Libraries;

use App\Interfaces\JobInterface;

class Queue
{
    private $db;

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

    public function push(
        JobInterface $job,
        string $queue = 'default',
        int $delaySeconds = 0
    ): int {
        $payload = json_encode([
            'class' => get_class($job),
            'data'  => serialize($job),
        ]);

        $this->db->table('jobs')->insert([
            'queue'        => $queue,
            'payload'      => $payload,
            'status'       => 'pending',
            'max_attempts' => 3,
            'available_at' => date('Y-m-d H:i:s', time() + $delaySeconds),
            'created_at'   => date('Y-m-d H:i:s'),
        ]);

        return $this->db->insertID();
    }
}

Push job dari mana saja dengan satu baris:

php

(new Queue())->push(
    new SendVerificationEmailJob($email, $name, $token)
);

// Dengan delay 60 detik & queue khusus
(new Queue())->push(
    new GenerateInvoiceJob($orderId),
    queue: 'invoices',
    delaySeconds: 60
);

05 Buat Spark Command — php spark queue:work

Ini jantungnya. Worker command ini yang jalan terus di background, ambil job pending, eksekusi, dan handle retry kalau gagal.

bash

php spark make:command QueueWork

app/Commands/QueueWork.php

namespace App\Commands;

use CodeIgniter\CLI\BaseCommand;
use CodeIgniter\CLI\CLI;

class QueueWork extends BaseCommand
{
    protected $group   = 'Queue';
    protected $name    = 'queue:work';
    protected $description = 'Process pending jobs from the queue';

    protected $arguments = [
        'queue' => 'Nama queue (default: default)',
    ];
    protected $options = [
        '--sleep' => 'Detik jeda antar poll (default: 3)',
        '--once'  => 'Proses satu job lalu berhenti',
    ];

    public function run(array $params)
    {
        $queue = $params[0] ?? 'default';
        $sleep = (int) CLI::getOption('sleep') ?: 3;
        $once  = CLI::getOption('once');

        CLI::write("[Queue Worker] Listening on: $queue", 'green');

        while (true) {
            $job = $this->fetchNextJob($queue);

            if (!$job) {
                if ($once) break;
                sleep($sleep);
                continue;
            }

            $this->processJob($job);

            if ($once) break;
        }
    }

    private function fetchNextJob(string $queue): ?object
    {
        $db = db_connect();
        $now = date('Y-m-d H:i:s');

        $job = $db->table('jobs')
            ->where('queue', $queue)
            ->where('status', 'pending')
            ->where('available_at <=', $now)
            ->orderBy('id', 'ASC')
            ->limit(1)
            ->get()->getRow();

        if (!$job) return null;

        // Lock job supaya worker lain tidak ambil job yang sama
        $db->table('jobs')
            ->where('id', $job->id)
            ->update(['status' => 'processing']);

        return $job;
    }

    private function processJob(object $row): void
    {
        $db      = db_connect();
        $payload = json_decode($row->payload, true);

        CLI::write("  → Processing job #{$row->id}: {$payload['class']}");

        try {
            // Unserialize job object dan jalankan
            $job = unserialize($payload['data']);
            $job->handle();

            $db->table('jobs')
                ->where('id', $row->id)
                ->update(['status' => 'done']);

            CLI::write("  ✓ Done", 'green');

        } catch (\Throwable $e) {
            $attempts = $row->attempts + 1;
            $failed   = $attempts >= $row->max_attempts;

            $db->table('jobs')->where('id', $row->id)->update([
                'status'       => $failed ? 'failed' : 'pending',
                'attempts'     => $attempts,
                'available_at' => date('Y-m-d H:i:s', time() + 60), // retry 60 detik
                'failed_at'    => $failed ? date('Y-m-d H:i:s') : null,
                'error'        => $e->getMessage(),
            ]);

            CLI::write("  ✗ Failed: " . $e->getMessage(), 'red');
        }
    }
}

06 Jalankan worker & lihat output-nya

Worker bisa dijalankan langsung dari terminal. Untuk production, gunakan supervisor biar otomatis restart kalau crash.

bash

# Jalankan worker default queue
php spark queue:work

# Queue spesifik dengan sleep 5 detik
php spark queue:work invoices --sleep=5

# Proses satu job lalu stop (bagus buat cron)
php spark queue:work --once

terminal — php spark queue:work

[Queue Worker] Listening on: default
  → Processing job #1: App\Jobs\SendVerificationEmailJob
  ✓ Done
  → Processing job #2: App\Jobs\GenerateInvoiceJob
  ✓ Done
  → Processing job #3: App\Jobs\SendVerificationEmailJob
  ✗ Failed: Connection refused (attempt 1/3, retry in 60s)
  ...

07 Setup Supervisor buat production

Di production, worker harus jalan terus bahkan setelah server restart atau crash. supervisor adalah process monitor yang handle ini secara otomatis.

bash

sudo apt install supervisor

/etc/supervisor/conf.d/ci4-worker.conf

[program:ci4-queue-worker]
command=php /var/www/html/spark queue:work
directory=/var/www/html
autostart=true
autorestart=true
numprocs=2
redirect_stderr=true
stdout_logfile=/var/log/ci4-worker.log

bash

sudo supervisorctl reread
sudo supervisorctl update
sudo supervisorctl start ci4-queue-worker:*

✅

Setting numprocs=2 menjalankan 2 worker sekaligus — berguna kalau volume job tinggi. Pastikan tabel jobs pakai InnoDB agar locking row-level bekerja dengan benar.

Penutup

Queue system ini bukan pengganti Redis atau SQS untuk skala enterprise — tapi untuk project menengah dengan ratusan hingga ribuan job per hari, ini lebih dari cukup dan zero external dependency. Kamu bisa extend ini dengan priority queue, dead-letter logging, atau dashboard monitoring sederhana. Di artikel berikutnya gue bakal bahas CI4 Event & Hook — cara bikin sistem yang benar-benar decoupled tanpa satu class pun tau soal class lain.

CI4 Custom Console Command + Queue Simulation

Tanpa Redis, Tanpa Laravel — Bikin Queue System Sendiri di CI4 Pakai Spark Command