Regroup improvements

src/Commands/RegroupCommand.php

@@ -7,16 +7,17 @@
 use Illuminate\Console\ConfirmableTrait;
 use Illuminate\Database\DatabaseManager;
 use Illuminate\Support\Facades\Config;
-use Laravel\Pulse\Contracts\Grouping;
+use Laravel\Pulse\Contracts\Groupable;
+use Laravel\Pulse\Contracts\Storage;
+use Laravel\Pulse\Pulse;
 use Laravel\Pulse\Queries\Concerns\InteractsWithConnection;
 use ReflectionClass;
 use Symfony\Component\Console\Attribute\AsCommand;
 
 #[AsCommand(name: 'pulse:regroup')]
 class RegroupCommand extends Command
 {
-    use ConfirmableTrait;
-    use InteractsWithConnection;
+    use ConfirmableTrait, InteractsWithConnection;
 
     /**
      * The command's signature.
@@ -32,53 +33,23 @@ class RegroupCommand extends Command
      */
     public $description = 'Re-apply grouping for supporting recorders.';
 
-    /**
-     * Create a new command instance.
-     */
-    public function __construct(
-        protected Repository $config,
-        protected DatabaseManager $db
-    ) {
-        parent::__construct();
-    }
-
     /**
      * Handle the command.
      */
-    public function handle(): int
+    public function handle(
+        Pulse $pulse,
+        Storage $storage,
+    ): int
     {
         if (! $this->confirmToProceed()) {
             return Command::FAILURE;
         }
 
-        collect(array_keys(Config::get('pulse.recorders')))
-            ->filter(fn ($recorder) => (new ReflectionClass($recorder))->implementsInterface(Grouping::class)) // @phpstan-ignore argument.type
-            ->map(fn ($recorder) => app($recorder)) // @phpstan-ignore argument.type
-            ->each(function ($recorder) {
-                $this->info("Re-grouping {$recorder->table}...");
-
-                $this->connection()->query()
-                    ->from($recorder->table)
-                    ->select($recorder->groupColumn())
-                    ->distinct()
-                    ->pluck($recorder->groupColumn())
-                    ->each(function ($value) use ($recorder) {
-                        $newValue = $recorder->group($value)();
-
-                        if ($newValue === $value) {
-                            return;
-                        }
-
-                        $this->info(" - [{$value}] => [{$newValue}]");
+        $pulse->recorders()
+            ->filter(fn ($recorder) => class_implements($recorder, Groupable::class))
+            ->pipe($storage->regroup(...));
 
-                        $this->connection()->query()
-                            ->from($recorder->table)
-                            ->where($recorder->groupColumn(), $value)
-                            ->update([
-                                $recorder->groupColumn() => $newValue,
-                            ]);
-                    });
-            });
+        $this->components->info('Recorders regrouped.');
 
         return Command::SUCCESS;
     }

src/Contracts/Grouping.php → src/Contracts/Groupable.php

@@ -4,7 +4,7 @@
 
 use Closure;
 
-interface Grouping
+interface Groupable
 {
     /**
      * Return a closure that groups the given value.

src/Pulse.php

@@ -261,6 +261,16 @@ protected function shouldRecord(Entry|Update $entry): bool
         return $this->filters->every(fn (callable $filter) => $filter($entry));
     }
 
+    /**
+     * Get the recorders.
+     *
+     * @return \Illuminate\Support\Collection<int, object>
+     */
+    public function recorders(): Collection
+    {
+        return $this->recorders;
+    }
+
     /**
      * Get the tables used by the recorders.
      *

src/Recorders/CacheInteractions.php

@@ -7,14 +7,14 @@
 use Illuminate\Cache\Events\CacheHit;
 use Illuminate\Cache\Events\CacheMissed;
 use Illuminate\Config\Repository;
-use Laravel\Pulse\Contracts\Grouping;
+use Laravel\Pulse\Contracts\Groupable;
 use Laravel\Pulse\Entry;
 use Laravel\Pulse\Pulse;
 
 /**
  * @internal
  */
-class CacheInteractions implements Grouping
+class CacheInteractions implements Groupable
 {
     use Concerns\Ignores;
     use Concerns\Sampling;

src/Recorders/OutgoingRequests.php

@@ -9,7 +9,7 @@
 use Illuminate\Foundation\Application;
 use Illuminate\Http\Client\Factory as HttpFactory;
 use Laravel\Pulse\Concerns\ConfiguresAfterResolving;
-use Laravel\Pulse\Contracts\Grouping;
+use Laravel\Pulse\Contracts\Groupable;
 use Laravel\Pulse\Entry;
 use Laravel\Pulse\Pulse;
 use Psr\Http\Message\RequestInterface;
@@ -19,7 +19,7 @@
 /**
  * @internal
  */
-class OutgoingRequests implements Grouping
+class OutgoingRequests implements Groupable
 {
     use Concerns\Ignores;
     use Concerns\Sampling;

src/Storage/Database.php

@@ -8,6 +8,7 @@
 use Illuminate\Database\Connection;
 use Illuminate\Database\DatabaseManager;
 use Illuminate\Support\Collection;
+use Laravel\Pulse\Contracts\Groupable;
 use Laravel\Pulse\Contracts\Storage;
 use Laravel\Pulse\Entry;
 use Laravel\Pulse\Update;
@@ -93,6 +94,38 @@ public function purge(Collection $tables): void
             ->truncate());
     }
 
+    /*
+     * Regroup the recorders.
+     *
+     * @param  \Illuminate\Support\Collection<int, \Laravel\Pulse\Contracts\Groupable>  $recorders
+     */
+    public function regroup(Collection $recorders): void
+    {
+        $recorders->each(function (Groupable $recorder) {
+            $this->connection()
+                ->table($recorder->table)
+                ->select($recorder->groupColumn())
+                ->distinct()
+                ->orderBy('date')
+                ->each(function ($record) use ($recorder) {
+                    $value = $record->{$recorder->groupColumn()};
+
+                    $newValue = $recorder->group($value)();
+
+                    if ($newValue === $value) {
+                        return;
+                    }
+
+                    $this->connection()
+                        ->table($recorder->table)
+                        ->where($recorder->groupColumn(), $value)
+                        ->update([
+                            $recorder->groupColumn() => $newValue,
+                        ]);
+                });
+        });
+    }
+
     /**
      * The interval to trim the storage to.
      */