Move updates to concern of the storage driver

src/Commands/CheckCommand.php

@@ -48,7 +48,7 @@ public function handle(
         while (true) {
             $now = new CarbonImmutable();
 
-            if ($cache->get('laravel:pulse:restart') !== $lastRestart) {
+            if ($lastRestart !== $cache->get('laravel:pulse:restart')) {
                 return self::SUCCESS;
             }
 

src/Commands/WorkCommand.php

@@ -44,7 +44,7 @@ public function handle(
         while (true) {
             $now = new CarbonImmutable;
 
-            if ($cache->get('laravel:pulse:restart') !== $lastRestart) {
+            if ($lastRestart !== $cache->get('laravel:pulse:restart')) {
                 return self::SUCCESS;
             }
 

src/Entries/SlowJobFinished.php

@@ -2,8 +2,6 @@
 
 namespace Laravel\Pulse\Entries;
 
-use Illuminate\Database\Connection;
-
 /**
  * @internal
  */
@@ -19,19 +17,6 @@ public function __construct(
         //
     }
 
-    /**
-     * Perform the update.
-     */
-    public function perform(Connection $db): void
-    {
-        $db->table($this->table())
-            ->where('job_uuid', $this->jobUuid)
-            ->update([
-                'slowest' => $db->raw("COALESCE(GREATEST(`slowest`,{$this->duration}),{$this->duration})"),
-                'slow' => $db->raw('`slow` + 1'),
-            ]);
-    }
-
     /**
      * The update's table.
      */

src/Entries/Update.php

@@ -2,20 +2,13 @@
 
 namespace Laravel\Pulse\Entries;
 
-use Illuminate\Database\Connection;
-
 abstract class Update
 {
     /**
      * The update's table.
      */
     abstract public function table(): string;
 
-    /**
-     * Perform the update.
-     */
-    abstract public function perform(Connection $db): void;
-
     /**
      * Resolve the update for ingest and storage.
      */

src/Ingests/Redis.php

@@ -90,7 +90,7 @@ protected function trimAfter(): Interval
     /**
      * Get the redis connection.
      */
-    protected function connection(): RedisAdapter
+    public function connection(): RedisAdapter
     {
         return new RedisAdapter($this->config, $this->manager->connection(
             $this->config->get('pulse.ingest.redis.connection')

src/PulseServiceProvider.php

@@ -52,6 +52,8 @@ public function register(): void
         }
 
         $this->app->singleton(Pulse::class);
+        $this->app->singleton(DatabaseStorage::class);
+        $this->app->singleton(RedisIngest::class);
 
         $this->app->bind(Storage::class, DatabaseStorage::class);
 

src/Storage/Database.php

@@ -4,16 +4,26 @@
 
 use Carbon\CarbonImmutable;
 use Carbon\CarbonInterval as Interval;
+use Closure;
 use Illuminate\Config\Repository;
 use Illuminate\Database\Connection;
 use Illuminate\Database\DatabaseManager;
 use Illuminate\Support\Collection;
+use Illuminate\Support\Traits\ReflectsClosures;
 use Laravel\Pulse\Contracts\Storage;
 use Laravel\Pulse\Entries\Entry;
+use Laravel\Pulse\Entries\SlowJobFinished;
 use Laravel\Pulse\Entries\Update;
 
 class Database implements Storage
 {
+    use ReflectsClosures;
+
+    /**
+     * Additional storage update handlers.
+     */
+    protected array $updateHandlers = [];
+
     /**
      * Create a new Database storage instance.
      */
@@ -24,6 +34,20 @@ public function __construct(
         //
     }
 
+    /**
+     * Handle the update using the closure.
+     *
+     * @param  (callable(Update): void)  $callback
+     */
+    public function handleUpdateUsing($callback): self
+    {
+        foreach ($this->firstClosureParameterTypes(Closure::fromCallable($callback)) as $class) {
+            $this->updateHandlers[$class] = $callback;
+        }
+
+        return $this;
+    }
+
     /**
      * Store the entries and updates.
      *
@@ -43,7 +67,30 @@ public function store(Collection $items): void
                     ->map(fn (Collection $inserts) => $inserts->pluck('attributes')->all())
                     ->each($this->connection()->table($table)->insert(...)));
 
-            $updates->each(fn (Update $update) => $update->perform($this->connection()));
+            $this->perform($updates);
+        });
+    }
+
+    /**
+     * Perform the given updates.
+     */
+    protected function perform($updates)
+    {
+        $updates->each(function (Update $update) {
+            if ($this->updateHandlers[$update::class] ?? false) {
+                $this->updateHandlers[$update::class]($update);
+
+                return;
+            }
+
+            if ($update instanceof SlowJobFinished) {
+                $this->connection()->table($update->table())
+                    ->where('job_uuid', $update->jobUuid)
+                    ->update([
+                        'slowest' => $this->connection()->raw("COALESCE(GREATEST(`slowest`,{$update->duration}),{$update->duration})"),
+                        'slow' => $this->connection()->raw('`slow` + 1'),
+                    ]);
+            }
         });
     }
 
@@ -69,7 +116,7 @@ protected function trimAfter(): Interval
     /**
      * Get the database connection.
      */
-    protected function connection(): Connection
+    public function connection(): Connection
     {
         return $this->manager->connection(
             $this->config->get('pulse.storage.database.connection') ?? $this->config->get('database.default')

tests/Feature/DatabaseStorageTest.php

@@ -0,0 +1,41 @@
+<?php
+
+use Illuminate\Support\Facades\DB;
+use Laravel\Pulse\Entries\Entry;
+use Laravel\Pulse\Entries\SlowJobFinished;
+use Laravel\Pulse\Storage\Database;
+
+it('performs slow job updates', function () {
+    $storage = App::make(Database::class);
+    $update = new SlowJobFinished('job-uuid', 321);
+
+    $storage->store(collect([
+        new Entry('pulse_jobs', [
+            'date' => now()->toDateTimeString(),
+            'job' => 'MyJob',
+            'job_uuid' => 'job-uuid',
+            'user_id' => '55',
+        ]),
+    ]));
+
+    $storage->store(collect([
+        new SlowJobFinished('job-uuid', 456),
+    ]));
+
+    $jobs = DB::table('pulse_jobs')->get();
+    $this->assertCount(1, $jobs);
+    $this->assertSame(1, $jobs[0]->slow);
+    $this->assertSame(456, $jobs[0]->slowest);
+});
+
+it('allows custom update handlers', function () {
+    $captured = null;
+    App::make(Database::class)->handleUpdateUsing(function (SlowJobFinished $update) use (&$captured) {
+        $captured = $update;
+    });
+
+    $update = new SlowJobFinished('job-uuid', 321);
+    App::make(Database::class)->store(collect([$update])); // resolve again to ensure that the handlers persist across different resolves from container.
+
+    $this->assertSame($update, $captured);
+});