wip

src/Commands/CheckCommand.php

@@ -76,6 +76,7 @@ public function handle(): int
                 ])->toJson(),
             ];
 
+            // TODO: do this via the injest
             DB::table('pulse_servers')->insert($stats);
 
             $this->line('<fg=gray>[system stats]</> '.json_encode($stats));

src/Commands/RestartCommand.php

@@ -31,7 +31,7 @@ class RestartCommand extends Command
     /**
      * Handle the command.
      */
-    public function handle(Redis $redisIngest, Database $databaseIngest): void
+    public function handle(): void
     {
         Cache::forever('illuminate:pulse:restart', $this->currentTime());
 

src/Commands/WorkCommand.php

@@ -37,14 +37,14 @@ public function handle(Redis $redisIngest, Database $databaseIngest): int
         $lastTrimmedDatabaseAt = (new CarbonImmutable)->startOfMinute();
 
         while (true) {
+            $now = new CarbonImmutable;
+
             if (Cache::get('illuminate:pulse:restart') !== $lastRestart) {
                 $this->comment('Pulse restart requested. Exiting at '.$now->toDateTimeString());
 
                 return self::SUCCESS;
             }
 
-            $now = new CarbonImmutable;
-
             if ($now->subMinute()->greaterThan($lastTrimmedDatabaseAt)) {
                 $this->comment('Trimming the database at '.$now->toDateTimeString());
 
@@ -59,9 +59,9 @@ public function handle(Redis $redisIngest, Database $databaseIngest): int
                 $this->comment('Queue finished processing. Sleeping at '.$now->toDateTimeString());
 
                 Sleep::for(1)->second();
+            } else {
+                $this->comment('Processed ['.$processed.'] entries at '.$now->toDateTimeString());
             }
-
-            $this->comment('Processed ['.$processed.'] entries at '.$now->toDateTimeString());
         }
     }
 }

src/Redis.php

@@ -50,23 +50,24 @@ public function xrange($key, $start, $end, $count = null)
     /**
      * Trim the stream.
      */
-    public function xtrim($key, $strategy, $stratgyModifier, $threshold)
+    public function xtrim($key, $strategy, $strategyModifier, $threshold)
     {
         $prefix = config('database.redis.options.prefix');
 
         if ($this->isPhpRedis()) {
             // PHP Redis does not support the minid strategy.
-            return $this->client()->rawCommand('XTRIM', $prefix.$key, $strategy, $stratgyModifier, $threshold);
+            return $this->client()->rawCommand('XTRIM', $prefix.$key, $strategy, $strategyModifier, $threshold);
         }
 
-        return $this->client()->xtrim($key, [$strategy, $stratgyModifier], $threshold);
+        return $this->client()->xtrim($key, [$strategy, $strategyModifier], $threshold);
     }
 
     /**
      * Run commands in a pipeline.
      */
     public function pipeline(callable $closure): array
     {
+        // TODO explain this code - lol
         // ensure we run against a connection...
         return $this->connection->pipeline(function ($redis) use ($closure) {
             $closure(new self(client: $redis));
@@ -78,6 +79,7 @@ public function pipeline(callable $closure): array
      */
     public function streamIdAt(CarbonImmutable $timestamp): string
     {
+        // TODO: pass through intervals rather than date instances everywhere
         $diff = (new CarbonImmutable)->diffInMilliseconds($timestamp);
 
         $redisTime = $this->client()->time();