add tombstone handler support

config/ksql.php

@@ -1,7 +1,7 @@
 <?php
 
-use ZiffMedia\LaravelKsql\DiscoverResources;
 use ZiffMedia\Ksql\ContentType;
+use ZiffMedia\LaravelKsql\DiscoverResources;
 
 return [
     'endpoint' => env('KSQL_ENDPOINT'),

src/Client.php

@@ -5,6 +5,7 @@
 use ZiffMedia\Ksql\Client as KsqlClient;
 use ZiffMedia\Ksql\PullQuery;
 use ZiffMedia\Ksql\ResultRow;
+use ZiffMedia\Ksql\TombstoneRow;
 
 class Client extends KsqlClient
 {
@@ -18,7 +19,11 @@ public function streamAndEmit(array|PushQuery $query): void
         }
 
         $handler = function (ResultRow $row) {
-            event($row->query->event, $row);
+            if ($row instanceof TombstoneRow) {
+                event($row->query->tombstoneEvent, $row);
+            } else {
+                event($row->query->event, $row);
+            }
         };
 
         foreach ($query as $q) {

src/ConsumerCommand.php

@@ -41,6 +41,7 @@ public function handle()
             if ($resource->shouldConsume) {
                 $query = new PushQuery($resource->getKeyName(), $resource->getKsqlStreamQuery(), fn () => null, $resource->offset);
                 $query->event = $resource->getEventName();
+                $query->tombstoneEvent = $resource->getTombstoneEventName();
                 $streamQueries[] = $query;
             }
         }

src/KsqlResource.php

@@ -8,6 +8,7 @@
 use Illuminate\Support\Str;
 use ZiffMedia\Ksql\Offset;
 use ZiffMedia\Ksql\ResultRow;
+use ZiffMedia\Ksql\TombstoneRow;
 
 class KsqlResource
 {
@@ -32,6 +33,10 @@ public function handle(ResultRow $data): void
     {
     }
 
+    public function handleTombstone(TombstoneRow $data): void
+    {
+    }
+
     public function getKsqlStreamQuery(): string
     {
         return sprintf('SELECT * FROM %s EMIT CHANGES;', $this->ksqlTable);
@@ -60,6 +65,11 @@ public function getEventName(): string
         return 'ksql.'.$this->getKeyName();
     }
 
+    public function getTombstoneEventName(): string
+    {
+        return $this->getEventName().'.tombstone';
+    }
+
     public function getCatchupQuery(): string
     {
         $latestModel = $this->getLatestModel();

src/KsqlServiceProvider.php

@@ -36,6 +36,7 @@ public function register()
                 config('ksql.auth.password')
             );
             $client->setAcceptContentType(config('ksql.client_content_type', ContentType::V1_DELIMITTED));
+
             return $client;
         });
     }
@@ -59,6 +60,7 @@ private function discoverResources()
             $instance = new $className;
             app(ResourceManager::class)->{$instance->getKeyName()} = $instance;
             Event::listen($instance->getEventName(), [$instance, 'handle']);
+            Event::listen($instance->getTombstoneEventName(), [$instance, 'handleTombstone']);
         });
     }
 }

src/PushQuery.php

@@ -7,4 +7,6 @@
 class PushQuery extends KsqlPushQuery
 {
     public string $event;
+
+    public string $tombstoneEvent;
 }

tests/KsqlResourceTest.php

@@ -122,3 +122,8 @@ public function getCatchupQuery(): string
     $kr = new KsqlResource();
     expect($kr->getEventName())->toBe('ksql.ksql_resource');
 });
+
+test('it should generate tombstone event names', function () {
+    $kr = new KsqlResource();
+    expect($kr->getTombstoneEventName())->toBe('ksql.ksql_resource.tombstone');
+});