Update

packages/mango/Bundle/MangoBundle.php

@@ -20,7 +20,6 @@
 use Manyou\Mango\Doctrine\Type\UuidType;
 use Manyou\Mango\HttpKernel\AsDtoInitializer;
 use Manyou\Mango\Scheduler\Messenger\RecurringScheduleMiddleware;
-use Manyou\Mango\Scheduler\Messenger\ScheduledMessageMiddleware;
 use Manyou\Mango\TaskQueue\Doctrine\Type\TaskStatusType;
 use Manyou\Mango\TaskQueue\Messenger\Middleware\TaskQueueMiddware;
 use ReflectionClass;
@@ -62,14 +61,9 @@ static function (ChildDefinition $definition, AsDtoInitializer $attribute, Refle
             priority: 1,
         );
 
-        $container->addCompilerPass(
-            new MessengerMiddlewarePass(['id' => ScheduledMessageMiddleware::class]),
-            priority: 2,
-        );
-
         $container->addCompilerPass(
             new MessengerMiddlewarePass(['id' => RecurringScheduleMiddleware::class]),
-            priority: 3,
+            priority: 2,
         );
 
         $container->addCompilerPass(

packages/mango/Doctrine/SchemaProvider.php

@@ -47,18 +47,26 @@ public function getConnection(): Connection
     public function transactional(Closure $func)
     {
         $this->connection->beginTransaction();
+        $commit = true;
+
         try {
-            $res = $func($this);
-            $this->connection->commit();
+            $res = $func($this, $commit);
+            if ($commit) {
+                $this->connection->commit();
+            }
 
             return $res;
         } catch (Throwable $e) {
-            try {
-                $this->connection->rollBack();
-            } catch (PDOException) {
-            }
+            $commit = false;
 
             throw $e;
+        } finally {
+            if (! $commit) {
+                try {
+                    $this->connection->rollBack();
+                } catch (PDOException) {
+                }
+            }
         }
     }
 

packages/mango/Doctrine/Type/BackedEnumType.php

@@ -6,6 +6,7 @@
 
 use BackedEnum;
 use Doctrine\DBAL\Platforms\AbstractPlatform;
+use Doctrine\DBAL\Platforms\PostgreSQLPlatform;
 use Doctrine\DBAL\Types\ConversionException;
 use LogicException;
 
@@ -19,6 +20,8 @@ trait BackedEnumType
         convertToDatabaseValue as doConvertToDatabaseValue;
     }
 
+    abstract private function getName(): string;
+
     abstract private function getEnumClass(): string;
 
     private function getEnums(): array
@@ -56,4 +59,13 @@ public function convertToPHPValue($value, AbstractPlatform $platform): ?BackedEn
 
         return $this->getEnumClass()::from($value);
     }
+
+    public function getMappedDatabaseTypes(AbstractPlatform $platform): array
+    {
+        if ($platform instanceof PostgreSQLPlatform) {
+            return [$this->getName()];
+        }
+
+        return [];
+    }
 }

packages/mango/Scheduler/Doctrine/Table/ScheduledMessagesTable.php

@@ -8,7 +8,6 @@
 use Doctrine\DBAL\Types\Types;
 use Manyou\Mango\Doctrine\Contract\TableProvider;
 use Manyou\Mango\Doctrine\Table;
-use Manyou\Mango\Doctrine\Type\UsDateTimeImmutableType;
 
 class ScheduledMessagesTable implements TableProvider
 {
@@ -18,11 +17,9 @@ public function __invoke(Schema $schema): Table
     {
         $table = new Table($schema, self::NAME);
         $table->addColumn('key', Types::STRING);
-        $table->addColumn('envelope', Types::JSON)->setPlatformOptions(['jsonb' => true]);
-        $table->addColumn('available_at', Types::DATETIME_IMMUTABLE, alias: 'availableAt');
-        $table->addColumn('last_dispatched_at', UsDateTimeImmutableType::NAME, ['notnull' => false], alias: 'lastDispatchedAt');
+        $table->addColumn('message_id', Types::BIGINT, alias: 'messageId');
         $table->setPrimaryKey(['key']);
-        $table->addIndex(['available_at']);
+        $table->addUniqueIndex(['message_id']);
 
         return $table;
     }

packages/mango/Scheduler/Messenger/ScheduleMessageStamp.php

@@ -0,0 +1,46 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Manyou\Mango\Scheduler\Messenger;
+
+use Symfony\Component\Messenger\Stamp\NonSendableStampInterface;
+
+class ScheduleMessageStamp implements NonSendableStampInterface
+{
+    private function __construct(
+        public readonly string $key,
+        public readonly string $action,
+    ) {
+    }
+
+    public static function insert(string $key): self
+    {
+        return new self($key, 'insert');
+    }
+
+    public static function update(string $key): self
+    {
+        return new self($key, 'update');
+    }
+
+    public static function upsert(string $key): self
+    {
+        return new self($key, 'upsert');
+    }
+
+    public function isUpdate(): bool
+    {
+        return 'update' === $this->action;
+    }
+
+    public function isUpsert(): bool
+    {
+        return 'upsert' === $this->action;
+    }
+
+    public function isInsert(): bool
+    {
+        return 'insert' === $this->action;
+    }
+}