Refactor subscription engine

docker-compose.yaml

@@ -1,4 +1,15 @@
+# docker run --rm -it --volume $PWD:/app --net="host" -w /app ghcr.io/patchlevel/php:8.5
 services:
+  php:
+    image: ghcr.io/patchlevel/php:8.5
+    volumes:
+      - .:/app
+    working_dir: /app
+    network_mode: host
+    tty: true
+    stdin_open: true
+    command: sleep infinity
+
   postgres:
     image: postgres:alpine
     environment:
@@ -13,4 +24,4 @@ services:
       - MYSQL_ALLOW_EMPTY_PASSWORD="yes"
       - MYSQL_DATABASE=eventstore
     ports:
-      - 3306:3306
+      - 3306:3306

src/Subscription/Engine/Command/Boot.php

@@ -0,0 +1,21 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Subscription\Engine\Command;
+
+final class Boot extends Command
+{
+    /**
+     * @param list<string>|null $ids
+     * @param list<string>|null $groups
+     * @param positive-int|null $limit
+     */
+    public function __construct(
+        array|null $ids = null,
+        array|null $groups = null,
+        public readonly int|null $limit = null,
+    ) {
+        parent::__construct($ids, $groups);
+    }
+}

src/Subscription/Engine/Command/Command.php

@@ -0,0 +1,18 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Subscription\Engine\Command;
+
+abstract class Command
+{
+    /**
+     * @param list<string>|null $ids
+     * @param list<string>|null $groups
+     */
+    public function __construct(
+        public readonly array|null $ids = null,
+        public readonly array|null $groups = null,
+    ) {
+    }
+}

src/Subscription/Engine/Command/Pause.php

@@ -0,0 +1,9 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Subscription\Engine\Command;
+
+final class Pause extends Command
+{
+}

src/Subscription/Engine/Command/Reactivate.php

@@ -0,0 +1,9 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Subscription\Engine\Command;
+
+final class Reactivate extends Command
+{
+}

src/Subscription/Engine/Command/Refresh.php

@@ -0,0 +1,9 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Subscription\Engine\Command;
+
+final class Refresh extends Command
+{
+}

src/Subscription/Engine/Command/Remove.php

@@ -0,0 +1,9 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Subscription\Engine\Command;
+
+final class Remove extends Command
+{
+}

src/Subscription/Engine/Command/Run.php

@@ -0,0 +1,21 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Subscription\Engine\Command;
+
+final class Run extends Command
+{
+    /**
+     * @param list<string>|null $ids
+     * @param list<string>|null $groups
+     * @param positive-int|null $limit
+     */
+    public function __construct(
+        array|null $ids = null,
+        array|null $groups = null,
+        public readonly int|null $limit = null,
+    ) {
+        parent::__construct($ids, $groups);
+    }
+}

src/Subscription/Engine/Command/Setup.php

@@ -0,0 +1,20 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Subscription\Engine\Command;
+
+final class Setup extends Command
+{
+    /**
+     * @param list<string>|null $ids
+     * @param list<string>|null $groups
+     */
+    public function __construct(
+        array|null $ids = null,
+        array|null $groups = null,
+        public readonly bool $skipBooting = false,
+    ) {
+        parent::__construct($ids, $groups);
+    }
+}

src/Subscription/Engine/Command/Teardown.php

@@ -0,0 +1,9 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Subscription\Engine\Command;
+
+final class Teardown extends Command
+{
+}

src/Subscription/Engine/Event/OnCommand.php

@@ -0,0 +1,15 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Subscription\Engine\Event;
+
+use Patchlevel\EventSourcing\Subscription\Engine\Command\Command;
+
+final class OnCommand
+{
+    public function __construct(
+        public readonly Command $command,
+    ) {
+    }
+}

src/Subscription/Engine/Event/OnHandleMessage.php

@@ -0,0 +1,17 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Subscription\Engine\Event;
+
+use Patchlevel\EventSourcing\Message\Message;
+use Patchlevel\EventSourcing\Subscription\Subscription;
+
+final class OnHandleMessage
+{
+    public function __construct(
+        public readonly Subscription $subscription,
+        public readonly Message $message,
+    ) {
+    }
+}

src/Subscription/Engine/Event/OnHandleMessageError.php

@@ -0,0 +1,21 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Subscription\Engine\Event;
+
+use Patchlevel\EventSourcing\Message\Message;
+use Patchlevel\EventSourcing\Subscription\Subscription;
+use Throwable;
+
+final class OnHandleMessageError
+{
+    public function __construct(
+        public readonly Subscription $subscription,
+        public readonly Throwable $throwable,
+        public readonly Message $message,
+        public readonly int $index,
+        public bool $transitionToFailed = false,
+    ) {
+    }
+}

src/Subscription/Engine/Event/OnHandleMessageSuccess.php

@@ -0,0 +1,19 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Subscription\Engine\Event;
+
+use Patchlevel\EventSourcing\Message\Message;
+use Patchlevel\EventSourcing\Subscription\Subscription;
+
+final class OnHandleMessageSuccess
+{
+    public function __construct(
+        public readonly Subscription $subscription,
+        public readonly Message $message,
+        public readonly int $index,
+        public bool $shouldChangePosition = true,
+    ) {
+    }
+}

src/Subscription/Engine/Event/OnProcessingFinished.php

@@ -0,0 +1,21 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Subscription\Engine\Event;
+
+use Patchlevel\EventSourcing\Subscription\Engine\Command\Command;
+
+final class OnProcessingFinished
+{
+    public const REASON_STREAM_ENDED = 'stream-ended';
+    public const REASON_LIMIT_REACHED = 'limit-reached';
+
+    /** @param self::REASON_* $reason */
+    public function __construct(
+        public readonly Command $command,
+        public readonly string $reason,
+        public readonly int $processed,
+    ) {
+    }
+}

src/Subscription/Engine/Event/OnResult.php

@@ -0,0 +1,17 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Subscription\Engine\Event;
+
+use Patchlevel\EventSourcing\Subscription\Engine\Command\Command;
+use Patchlevel\EventSourcing\Subscription\Engine\Result;
+
+final class OnResult
+{
+    public function __construct(
+        public readonly Command $command,
+        public readonly Result $result,
+    ) {
+    }
+}