dependencies update

.docker/php/cli/Dockerfile

@@ -26,14 +26,14 @@ ENV COMPOSER_CACHE_DIR=/var/.composer
 
 FROM app_setup AS app_worker
 
-RUN install-php-extensions opcache igbinary
+RUN install-php-extensions igbinary
 
 USER $UID
 WORKDIR $WORKDIR
 
 FROM app_setup AS app_devel
 
-RUN install-php-extensions opcache igbinary xdebug
+RUN install-php-extensions igbinary xdebug
 
 USER $UID
 WORKDIR $WORKDIR

.dockerignore

@@ -1,3 +1,5 @@
 .git
 .github
 .docker
+.idea
+.zed

.gitattributes

@@ -1,11 +1,13 @@
-/.*                     export-ignore
-/benchmark/             export-ignore
-/tests/                 export-ignore
-/runtime/               export-ignore
-/*.neon                 export-ignore
-/*.xml                  export-ignore
-/*.xml.dist             export-ignore
-/*.lock                 export-ignore
-docker-compose.yml      export-ignore
-Makefile                export-ignore
-rector.php              export-ignore
+text=auto   eol=lf
+*.phar      binary
+*.php       diff=php
+
+* export-ignore
+/src            -export-ignore
+/src/**         -export-ignore
+/example        -export-ignore
+/example/**     -export-ignore
+/composer.json  -export-ignore
+/CHANGELOG.md   -export-ignore
+/LICENSE        -export-ignore
+/README.md      -export-ignore

.github/workflows/code-check.yml

@@ -21,9 +21,6 @@ jobs:
         with:
           php-version: ${{ matrix.php-version }}
           coverage: none
-        env:
-          fail-fast: true
-          COMPOSER_TOKEN: ${{ secrets.GITHUB_TOKEN }}
 
       - name: Install composer dependencies
         uses: ramsey/composer-install@v3
@@ -35,7 +32,7 @@ jobs:
         run: vendor/bin/phpcs
 
       - name: Psalm
-        run: vendor/bin/psalm --stats --shepherd --output-format=github
+        run: vendor/bin/psalm --php-version=${{ matrix.php-version }} --stats --show-info --no-diff --no-cache
 
       - name: PHPStan
         run: vendor/bin/phpstan analyse

Makefile

@@ -29,6 +29,10 @@ check:
 		composer:latest \
 		composer check
 
+fix:
+	- make phpcbf
+	- make rector
+
 psalm: ## psalm
 	docker run --init -it --rm -u ${USER} -v "$$(pwd):/app" -w /app \
 		ghcr.io/kuaukutsu/php:${PHP_VERSION}-cli \
@@ -84,7 +88,7 @@ publisher:
 
 consumer:
 	USER=$(USER) docker compose -f ./docker-compose.yml run --rm -u $(USER) -w /tests/simulation cli \
-		php worker-with-catch.php --schema=high
+		php worker.php --schema=high
 
 bench: ## bench
 	USER=$(USER) docker compose -f ./docker-compose.yml run --rm -u $(USER) -w / \

composer.json

@@ -18,8 +18,7 @@
     "php": "^8.3",
     "ext-redis": "*",
     "amphp/redis": "^2.0",
-    "kuaukutsu/queue-core": "^0.5.3",
-    "symfony/console": "^7.3"
+    "kuaukutsu/queue-core": "^0.5.3"
   },
   "require-dev": {
     "ext-pcntl": "*",
@@ -31,6 +30,7 @@
     "roave/security-advisories": "dev-latest",
     "slevomat/coding-standard": "^8.7",
     "squizlabs/php_codesniffer": "^3.7",
+    "symfony/console": "^7.3",
     "vimeo/psalm": "^6.13"
   },
   "autoload": {
@@ -47,9 +47,6 @@
     ]
   },
   "config": {
-    "platform": {
-      "php": "8.3.21"
-    },
     "allow-plugins": {
       "dealerdirect/phpcodesniffer-composer-installer": true
     },

rector.php

@@ -1,33 +1,32 @@
 <?php
 
+/**
+ * @noinspection PhpUnhandledExceptionInspection
+ */
+
 declare(strict_types=1);
 
 use Rector\CodeQuality\Rector\Class_\InlineConstructorDefaultToPropertyRector;
-use Rector\Config\RectorConfig;
 use Rector\Php83\Rector\ClassMethod\AddOverrideAttributeToOverriddenMethodsRector;
-use Rector\Set\ValueObject\SetList;
 
-return static function (RectorConfig $rectorConfig): void {
-    $rectorConfig->paths(
+return Rector\Config\RectorConfig::configure()
+    ->withPaths(
         [
             __DIR__ . '/src',
             __DIR__ . '/tests',
         ]
-    );
-
-    // register a single rule
-    $rectorConfig->rule(InlineConstructorDefaultToPropertyRector::class);
-    $rectorConfig->ruleWithConfiguration(
+    )
+    ->withParallel()
+    ->withCache('/tmp/var/rector')
+    ->withPhpSets()
+    ->withRules(
+        [
+            InlineConstructorDefaultToPropertyRector::class,
+        ]
+    )
+    ->withConfiguredRule(
         AddOverrideAttributeToOverriddenMethodsRector::class,
         [
             'allow_override_empty_method' => true,
         ]
     );
-
-    // define sets of rules
-    $rectorConfig->sets([
-        SetList::PHP_81,
-        SetList::PHP_82,
-        SetList::PHP_83,
-    ]);
-};

src/Consumer.php

@@ -9,7 +9,6 @@
 use Throwable;
 use Amp\Redis\RedisClient;
 use Revolt\EventLoop;
-use kuaukutsu\queue\core\exception\QueueConsumeException;
 use kuaukutsu\queue\core\handler\HandlerInterface;
 use kuaukutsu\queue\core\ConsumerInterface;
 use kuaukutsu\queue\core\SchemaInterface;
@@ -44,10 +43,7 @@
         $this->handler = new TaskHandler($handler, $catch);
     }
 
-    /**
-     * @throws QueueConsumeException
-     */
     #[Override]
     public function consume(SchemaInterface $schema): void
     {
         $string = new RedisString($this->redis, $schema);
@@ -69,7 +65,7 @@
         );
     }
 
     #[Override]
     public function disconnect(): void
     {
         if ($this->ctx instanceof Context) {

src/Publisher.php

@@ -29,11 +29,7 @@
     {
     }
 
-    /**
-     * @return non-empty-string
-     * @throws QueuePublishException
-     */
     #[Override]
     public function push(SchemaInterface $schema, QueueTask $task, ?QueueContext $context = null): string
     {
         $string = new RedisString($this->redis, $schema);
@@ -54,7 +50,7 @@
         return $task->getUuid();
     }
 
     #[Override]
     public function pushBatch(SchemaInterface $schema, iterable $taskBatch, ?QueueContext $context = null): array
     {
         $groupAwait = [];

src/internal/Context.php

@@ -28,22 +28,18 @@ final class Context
     /**
      * @var non-empty-string[]
      */
-    private array $identityList = [];
+    private array $ackList = [];
 
     /**
      * @var non-empty-string[]
      */
     private array $payloadList = [];
 
-    /**
-     * @param non-negative-int $maxExceededAttempts
-     */
     public function __construct(
         public readonly SchemaInterface $schema,
         private readonly RedisConsume $streamGroup,
         private readonly RedisString $string,
         private readonly EventDispatcher $eventDispatcher,
-        public readonly int $maxExceededAttempts = 3,
     ) {
     }
 
@@ -91,16 +87,16 @@ static function () use ($fn, $name, $event): void {
      */
     public function setAck(string $identity, string $payloadUuid): void
     {
-        $this->identityList[] = $identity;
+        $this->ackList[] = $identity;
         $this->payloadList[] = $payloadUuid;
     }
 
     public function sendAck(): void
     {
-        if ($this->identityList !== []) {
-            $this->streamGroup->ack($this->identityList[0], ...array_slice($this->identityList, 1));
-            $this->trigger(Event::MessageAck, new MessageAckEvent($this->identityList));
-            $this->identityList = [];
+        if ($this->ackList !== []) {
+            $this->streamGroup->ack($this->ackList[0], ...array_slice($this->ackList, 1));
+            $this->trigger(Event::MessageAck, new MessageAckEvent($this->ackList));
+            $this->ackList = [];
         }
 
         if ($this->payloadList !== []) {

src/internal/stream/RedisConsume.php

@@ -170,13 +170,13 @@ public function pending(string $identity): array
             1,
         );
 
-        if (is_array($result)) {
+        if (is_array($result) && is_array($result[0])) {
             return [
-                /**
-                 * @phpstan-ignore offsetAccess.nonOffsetAccessible,cast.string
-                 */
+                /** @phpstan-ignore cast.string */
                 'consumer' => (string)($result[0][1] ?? $template['consumer']),
+                /** @phpstan-ignore cast.int */
                 'elapsedMilliseconds' => (int)($result[0][2] ?? $template['elapsedMilliseconds']),
+                /** @phpstan-ignore cast.int */
                 'deliveryCount' => (int)($result[0][3] ?? $template['deliveryCount']),
             ];
         }

src/internal/workflow/WorkflowCatch.php

@@ -22,8 +22,13 @@
  */
 final readonly class WorkflowCatch
 {
-    public function __construct(private RedisConsume $stream)
-    {
+    /**
+     * @param non-negative-int $maxExceededAttempts
+     */
+    public function __construct(
+        private RedisConsume $stream,
+        private int $maxExceededAttempts = 3,
+    ) {
     }
 
     /**
@@ -51,7 +56,7 @@ public function __invoke(Context $ctx, string $identity, Payload $payload, Throw
             return;
         }
 
-        if ($ctx->maxExceededAttempts === 0) {
+        if ($this->maxExceededAttempts === 0) {
             $ctx->trigger(
                 Event::MessageHandleError,
                 new MessageErrorEvent($payload, $exception),
@@ -67,7 +72,7 @@ public function __invoke(Context $ctx, string $identity, Payload $payload, Throw
             new MessageErrorEvent($payload, $exception, $attempts),
         );
 
-        if ($attempts >= $ctx->maxExceededAttempts) {
+        if ($attempts >= $this->maxExceededAttempts) {
             $this->dlq(
                 $ctx,
                 $identity,
@@ -153,10 +158,9 @@ private function incrAttempt(Context $ctx, Payload $payload, int $currentAttempt
     private function attempts(RedisConsume $command, Context $ctx, string $identity): int
     {
         /**
-         * @param non-empty-string $identity
-         * @return positive-int
+         * @psalm-var Closure(RedisConsume, Context, non-empty-string): positive-int $fn
          */
-        $fn = static function (RedisConsume $command, Context $ctx, string $identity): int {
+        static $fn = static function (RedisConsume $command, Context $ctx, string $identity): int {
             /** @var non-empty-string $identity */
             try {
                 $pending = $command->pending($identity);

src/internal/workflow/WorkflowClaim.php

@@ -4,6 +4,7 @@
 
 namespace kuaukutsu\poc\queue\stream\internal\workflow;
 
+use Closure;
 use Amp\CancelledException;
 use Amp\TimeoutCancellation;
 use kuaukutsu\poc\queue\stream\event\Event;
@@ -29,7 +30,17 @@ public function __construct(
     public function __invoke(Context $ctx, WorkflowCatch $catch): void
     {
         $action = $this->action->run(...);
-        $workflow = static function (string $identity, Payload $payload) use ($action, $catch, $ctx): void {
+
+        /**
+         * @psalm-var Closure(Context, string, Payload, callable, callable): void $workflow
+         */
+        static $workflow = static function (
+            Context $ctx,
+            string $identity,
+            Payload $payload,
+            callable $action,
+            callable $catch,
+        ): void {
             /** @var non-empty-string $identity */
             if ($action($catch(...), $ctx, $identity, $payload)) {
                 $ctx->setAck($identity, $payload->uuid);
@@ -40,7 +51,7 @@ public function __invoke(Context $ctx, WorkflowCatch $catch): void
         while (true) {
             $list = [];
             foreach ($this->autoclaim($this->stream, $lastIdentity) as $identity => $payload) {
-                $list[] = async($workflow(...), $identity, $payload);
+                $list[] = async($workflow(...), $ctx, $identity, $payload, $action, $catch);
                 $lastIdentity = $identity;
             }
 
@@ -67,9 +78,10 @@ public function __invoke(Context $ctx, WorkflowCatch $catch): void
     private function autoclaim(RedisConsume $command, string $lastIdentity): iterable
     {
         /**
-         * @return iterable<non-empty-string, Payload>
+         * @psalm-var Closure(RedisConsume, string): iterable<non-empty-string, Payload> $fn
+         * @phpstan-ignore varTag.nativeType
          */
-        $fn = static function (RedisConsume $command, string $lastIdentity): iterable {
+        static $fn = static function (RedisConsume $command, string $lastIdentity): iterable {
             $batch = $command->autoclaim($lastIdentity);
             if ($batch === []) {
                 return;