allow to get the index of the stream

src/Store/ArrayStream.php

@@ -40,6 +40,11 @@ public function position(): int
         return $this->position;
     }
 
+    public function index(): int
+    {
+        return $this->position;
+    }
+
     public function current(): Message|null
     {
         return $this->iterator->current() ?: null;

src/Store/DoctrineDbalStoreStream.php

@@ -22,6 +22,8 @@ final class DoctrineDbalStoreStream implements Stream, IteratorAggregate
 
     private int $position;
 
+    private int|null $index;
+
     public function __construct(
         private readonly Result $result,
         EventSerializer $serializer,
@@ -30,6 +32,7 @@ public function __construct(
     ) {
         $this->generator = $this->buildGenerator($result, $serializer, $aggregateRootRegistry, $platform);
         $this->position = 0;
+        $this->index = null;
     }
 
     public function close(): void
@@ -47,6 +50,15 @@ public function position(): int
         return $this->position;
     }
 
+    public function index(): int
+    {
+        if (!$this->index) {
+            throw new InvalidIndex();
+        }
+
+        return $this->index;
+    }
+
     /** @return Traversable<Message> */
     public function getIterator(): Traversable
     {
@@ -60,9 +72,10 @@ private function buildGenerator(
         AggregateRootRegistry $aggregateRootRegistry,
         AbstractPlatform $platform,
     ): Generator {
-        /** @var array{aggregate: string, aggregate_id: string, playhead: int|string, event: string, payload: string, recorded_on: string, custom_headers: string} $data */
+        /** @var array{id: string, aggregate: string, aggregate_id: string, playhead: int|string, event: string, payload: string, recorded_on: string, custom_headers: string} $data */
         foreach ($result->iterateAssociative() as $data) {
             $this->position++;
+            $this->index = $data['id'];
 
             $event = $serializer->deserialize(new SerializedEvent($data['event'], $data['payload']));
 

src/Store/InvalidIndex.php

@@ -0,0 +1,15 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Store;
+
+use RuntimeException;
+
+final class InvalidIndex extends RuntimeException
+{
+    public function __construct()
+    {
+        parent::__construct('invalid index');
+    }
+}

src/Store/Stream.php

@@ -15,4 +15,6 @@ public function close(): void;
     public function current(): Message|null;
 
     public function position(): int;
+
+    public function index(): int;
 }

tests/Unit/Store/ArrayStreamTest.php

@@ -44,11 +44,13 @@ public function testWithMessages(): void
 
         $stream = new ArrayStream($messages);
 
+        self::assertSame(0, $stream->index());
         self::assertSame(0, $stream->position());
         self::assertSame($message, $stream->current());
 
         $array = iterator_to_array($stream);
 
+        self::assertSame(1, $stream->index());
         self::assertSame(1, $stream->position());
         self::assertSame(null, $stream->current());
 
@@ -80,10 +82,12 @@ public function testPosition(): void
 
         $stream = new ArrayStream($messages);
 
+        self::assertSame(0, $stream->index());
         self::assertSame(0, $stream->position());
 
         iterator_to_array($stream);
 
+        self::assertSame(3, $stream->index());
         self::assertSame(3, $stream->position());
     }
 }

tests/Unit/Store/DoctrineDbalStoreTest.php

@@ -84,6 +84,7 @@ public function testLoadWithOneEvent(): void
         $result->iterateAssociative()->willReturn(new ArrayIterator(
             [
                 [
+                    'id' => 1,
                     'aggregate' => 'profile',
                     'aggregate_id' => '1',
                     'playhead' => '1',
@@ -136,6 +137,7 @@ public function testLoadWithOneEvent(): void
 
         $message = $stream->current();
 
+        self::assertSame(1, $stream->index());
         self::assertInstanceOf(Message::class, $message);
         self::assertInstanceOf(ProfileCreated::class, $message->event());
         self::assertSame('1', $message->aggregateId());