feat: switch to db queue as default

Author: imorland

framework/core/migrations/2025_10_11_000000_create_queue_jobs.php

@@ -0,0 +1,17 @@
+<?php
+
+use Flarum\Database\Migration;
+use Illuminate\Database\Schema\Blueprint;
+
+return Migration::createTableIfNotExists(
+    'queue_jobs',
+    function (Blueprint $table) {
+        $table->bigIncrements('id');
+        $table->string('queue')->index();
+        $table->longText('payload');
+        $table->unsignedTinyInteger('attempts');
+        $table->unsignedInteger('reserved_at')->nullable();
+        $table->unsignedInteger('available_at');
+        $table->unsignedInteger('created_at');
+    }
+);

framework/core/migrations/2025_10_11_000001_create_queue_failed_jobs.php

@@ -0,0 +1,17 @@
+<?php
+
+use Flarum\Database\Migration;
+use Illuminate\Database\Schema\Blueprint;
+
+return Migration::createTableIfNotExists(
+    'queue_failed_jobs',
+    function (Blueprint $table) {
+        $table->id();
+        $table->string('uuid')->unique();
+        $table->text('connection')->nullable();
+        $table->text('queue');
+        $table->longText('payload');
+        $table->longText('exception');
+        $table->timestamp('failed_at')->useCurrent();
+    }
+);

framework/core/src/Console/ConsoleServiceProvider.php

@@ -19,6 +19,8 @@
 use Flarum\Foundation\Console\AssetsPublishCommand;
 use Flarum\Foundation\Console\CacheClearCommand;
 use Flarum\Foundation\Console\InfoCommand;
+use Flarum\Foundation\ContainerUtil;
+use Flarum\Queue\Console\DatabaseWorkerArgs;
 use Flarum\Settings\SettingsRepositoryInterface;
 use Illuminate\Console\Events\CommandFinished;
 use Illuminate\Console\Scheduling\CacheEventMutex;
@@ -74,8 +76,8 @@ public function register(): void
             ];
         });
 
-        $this->container->singleton('flarum.console.scheduled', function () {
-            return [];
+        $this->container->singleton('flarum.console.scheduled', function (Container $container) {
+            return [array_merge($container->make('flarum.queue.schedule', []))];
         });
     }
 

framework/core/src/Queue/Console/DatabaseWorkerArgs.php

@@ -0,0 +1,41 @@
+<?php
+
+namespace Flarum\Queue\Console;
+
+use Flarum\Settings\SettingsRepositoryInterface;
+
+class DatabaseWorkerArgs
+{
+    public function __construct(protected SettingsRepositoryInterface $settings)
+    {
+    }
+
+    public function args(): array
+    {
+        $args = [
+            '--stop-when-empty',
+        ];
+
+        if ($retries = $this->settings->get('database-queue.retries')) {
+            $args['--tries'] = $retries;
+        }
+
+        if ($memory = $this->settings->get('database-queue.memory')) {
+            $args['--memory'] = $memory;
+        }
+
+        if ($timeout = $this->settings->get('database-queue.timeout')) {
+            $args['--timeout'] = $timeout;
+        }
+
+        if ($rest = $this->settings->get('database-queue.rest')) {
+            $args['--rest'] = $rest;
+        }
+
+        if ($backoff = $this->settings->get('database-queue.backoff')) {
+            $args['--backoff'] = $backoff;
+        }
+
+        return $args;
+    }
+}

framework/core/src/Queue/Console/WorkCommand.php

@@ -0,0 +1,29 @@
+<?php
+
+namespace Flarum\Queue\Console;
+
+use Carbon\Carbon;
+use Flarum\Settings\SettingsRepositoryInterface;
+use Illuminate\Contracts\Cache\Repository as Cache;
+use Illuminate\Queue\Worker;
+
+class WorkCommand extends \Illuminate\Queue\Console\WorkCommand
+{
+    public function __construct(Worker $worker, Cache $cache, protected SettingsRepositoryInterface $settings)
+    {
+        parent::__construct($worker, $cache);
+    }
+
+    public function handle()
+    {
+        $this->settings->set('database_queue.working', Carbon::now()->toIso8601String());
+
+        try {
+            parent::handle();
+        } catch (\Exception $e) {
+            $this->settings->delete('database_queue.working');
+
+            throw $e;
+        }
+    }
+}

framework/core/src/Queue/DatabaseUuidFailedJobProvider.php

@@ -0,0 +1,24 @@
+<?php
+
+namespace Flarum\Queue;
+
+use Illuminate\Database\ConnectionInterface;
+use Illuminate\Database\ConnectionResolverInterface;
+
+class DatabaseUuidFailedJobProvider extends \Illuminate\Queue\Failed\DatabaseUuidFailedJobProvider
+{
+    public function __construct(ConnectionResolverInterface $resolver, $database, $table, protected ConnectionInterface $connection)
+    {
+        parent::__construct($resolver, $database, $table);
+    }
+
+    /**
+     * Get a new query builder instance for the table.
+     *
+     * @return \Illuminate\Database\Query\Builder
+     */
+    protected function getTable()
+    {
+        return $this->connection->table($this->table);
+    }
+}

framework/core/src/Queue/QueueServiceProvider.php

@@ -11,9 +11,11 @@
 
 use Flarum\Foundation\AbstractServiceProvider;
 use Flarum\Foundation\Config;
+use Flarum\Foundation\ContainerUtil;
 use Flarum\Foundation\ErrorHandling\Registry;
 use Flarum\Foundation\ErrorHandling\Reporter;
 use Flarum\Foundation\Paths;
+use Flarum\Queue\Console\DatabaseWorkerArgs;
 use Illuminate\Container\Container as ContainerImplementation;
 use Illuminate\Contracts\Cache\Factory as CacheFactory;
 use Illuminate\Contracts\Cache\Repository;
@@ -24,8 +26,8 @@
 use Illuminate\Contracts\Queue\Queue;
 use Illuminate\Queue\Connectors\ConnectorInterface;
 use Illuminate\Queue\Console as Commands;
+use Illuminate\Queue\DatabaseQueue;
 use Illuminate\Queue\Events\JobFailed;
-use Illuminate\Queue\Failed\NullFailedJobProvider;
 use Illuminate\Queue\Listener as QueueListener;
 use Illuminate\Queue\SyncQueue;
 use Illuminate\Queue\Worker;
@@ -39,7 +41,7 @@ class QueueServiceProvider extends AbstractServiceProvider
         Commands\ListFailedCommand::class,
         Commands\RestartCommand::class,
         Commands\RetryCommand::class,
-        Commands\WorkCommand::class,
+        Console\WorkCommand::class,
     ];
 
     public function register(): void
@@ -55,7 +57,10 @@ public function register(): void
         // Extensions can override this binding if they want to make Flarum use
         // a different queuing backend.
         $this->container->singleton('flarum.queue.connection', function (ContainerImplementation $container) {
-            $queue = new SyncQueue;
+            $queue = new DatabaseQueue(
+                $this->container->make('db.connection'),
+                'queue_jobs'
+            );
             $queue->setContainer($container);
 
             return $queue;
@@ -117,7 +122,27 @@ public function __call(string $name, ?array $arguments): mixed
         });
 
         $this->container->singleton('queue.failer', function () {
-            return new NullFailedJobProvider();
+            /** @var Config $config */
+            $config = $this->container->make(Config::class);
+
+            return new DatabaseUuidFailedJobProvider(
+                $this->container->make('db'),
+                $config->offsetGet('database.database'),
+                'queue_failed_jobs',
+                $this->container->make('flarum.db')
+            );
+        });
+
+        // By default, we use the DatabaseQueue, which requires a schedule to run. Other queue drivers
+        // may not need this, so we allow extensions to override this binding.
+        $this->container->singleton('flarum.queue.schedule', function (Container $container): array {
+            return [
+                'command' => 'queue:work',
+                'callback' => ContainerUtil::wrapCallback(function ($event) {
+                    $event->everyMinute();
+                }, $container),
+                'args' => ($container->make(DatabaseWorkerArgs::class))->args(),
+            ];
         });
 
         $this->container->alias('flarum.queue.connection', Queue::class);