[1.x] Fix trimming direct to storage ingest (#318)

* remove extra closure

* disable lottery for test

* remove only call

* add test to ensure trimming is called

* change comparison to match with database query

* Only collect queries we care about to remove trimming concerns

---------

Co-authored-by: Tim MacDonald <hello@timacdonald.me>

Author: sinnbeck

src/Pulse.php

@@ -311,7 +311,7 @@ public function ingest(): int
             $odds = $this->app->make('config')->get('pulse.ingest.trim.lottery') ?? $this->app->make('config')->get('pulse.ingest.trim_lottery');
 
             Lottery::odds(...$odds)
-                ->winner(fn () => $this->rescue(fn () => $ingest->trim(...)))
+                ->winner(fn () => $this->rescue($ingest->trim(...)))
                 ->choose();
 
             $this->flush();

tests/Feature/PulseTest.php

@@ -34,6 +34,17 @@
     expect($storage->stored[1]->key)->toBe('keep');
 });
 
+it('can trim records', function () {
+    App::instance(Storage::class, $storage = new StorageFake);
+
+    Pulse::record('foo', 'delete', 0, now()->subMonth());
+    Pulse::record('foo', 'keep', 0);
+
+    Pulse::ingest();
+
+    expect($storage->stored)->toHaveCount(1);
+});
+
 it('can lazily capture entries', function () {
     App::instance(Storage::class, $storage = new StorageFake);
 

tests/Feature/Storage/DatabaseStorageTest.php

@@ -123,9 +123,12 @@
 });
 
 it('combines duplicate count aggregates before upserting', function () {
-    Config::set('pulse.ingest.trim.lottery', [0, 1]);
     $queries = collect();
-    DB::listen(fn ($query) => $queries[] = $query);
+    DB::listen(function (QueryExecuted $event) use (&$queries) {
+        if (str_starts_with($event->sql, 'insert')) {
+            $queries[] = $event;
+        }
+    });
 
     Pulse::record('type', 'key1')->count();
     Pulse::record('type', 'key1')->count();
@@ -150,9 +153,12 @@
 });
 
 it('combines duplicate min aggregates before upserting', function () {
-    Config::set('pulse.ingest.trim.lottery', [0, 1]);
     $queries = collect();
-    DB::listen(fn ($query) => $queries[] = $query);
+    DB::listen(function (QueryExecuted $event) use (&$queries) {
+        if (str_starts_with($event->sql, 'insert')) {
+            $queries[] = $event;
+        }
+    });
 
     Pulse::record('type', 'key1', 200)->min();
     Pulse::record('type', 'key1', 100)->min();
@@ -177,9 +183,12 @@
 });
 
 it('combines duplicate max aggregates before upserting', function () {
-    Config::set('pulse.ingest.trim.lottery', [0, 1]);
     $queries = collect();
-    DB::listen(fn ($query) => $queries[] = $query);
+    DB::listen(function (QueryExecuted $event) use (&$queries) {
+        if (str_starts_with($event->sql, 'insert')) {
+            $queries[] = $event;
+        }
+    });
 
     Pulse::record('type', 'key1', 100)->max();
     Pulse::record('type', 'key1', 300)->max();
@@ -204,9 +213,12 @@
 });
 
 it('combines duplicate sum aggregates before upserting', function () {
-    Config::set('pulse.ingest.trim.lottery', [0, 1]);
     $queries = collect();
-    DB::listen(fn ($query) => $queries[] = $query);
+    DB::listen(function (QueryExecuted $event) use (&$queries) {
+        if (str_starts_with($event->sql, 'insert')) {
+            $queries[] = $event;
+        }
+    });
 
     Pulse::record('type', 'key1', 100)->sum();
     Pulse::record('type', 'key1', 300)->sum();
@@ -231,9 +243,12 @@
 });
 
 it('combines duplicate average aggregates before upserting', function () {
-    Config::set('pulse.ingest.trim.lottery', [0, 1]);
     $queries = collect();
-    DB::listen(fn ($query) => $queries[] = $query);
+    DB::listen(function (QueryExecuted $event) use (&$queries) {
+        if (str_starts_with($event->sql, 'insert')) {
+            $queries[] = $event;
+        }
+    });
 
     Pulse::record('type', 'key1', 100)->avg();
     Pulse::record('type', 'key1', 300)->avg();
@@ -466,7 +481,9 @@
 it('collapses values with the same key into a single upsert', function () {
     $bindings = [];
     DB::listen(function (QueryExecuted $event) use (&$bindings) {
-        $bindings = $event->bindings;
+        if (str_starts_with($event->sql, 'insert')) {
+            $bindings = $event->bindings;
+        }
     });
 
     Pulse::set('read_counter', 'post:321', 123);

tests/StorageFake.php

@@ -31,7 +31,7 @@ public function store(Collection $items): void
      */
     public function trim(): void
     {
-        //
+        $this->stored = $this->stored->reject(fn($record) => $record->timestamp <= now()->subWeek()->timestamp);
     }
 
     /**