Improve conflict handling.

Model/ConflictException.php

@@ -0,0 +1,7 @@
+<?php
+namespace SnowIO\IdempotentAPI\Model;
+
+class ConflictException extends \RuntimeException
+{
+
+}

Model/DispatchPlugin.php

@@ -16,22 +16,22 @@ class DispatchPlugin
     private $response;
     private $lockService;
     private $resourceConnection;
-    private $modificationTimeRepo;
+    private $messageGroupRepository;
     private $errorProcessor;
 
     public function __construct(
         Request $request,
         Response $response,
         LockService $lockService,
         ResourceConnection $resourceConnection,
-        ResourceModificationTimeRepository $resourceTimestampRepository,
+        WebApiMessageGroupRepository $messageGroupRepository,
         ErrorProcessor $errorProcessor
     ) {
         $this->request = $request;
         $this->response = $response;
         $this->lockService = $lockService;
         $this->resourceConnection = $resourceConnection;
-        $this->modificationTimeRepo = $resourceTimestampRepository;
+        $this->messageGroupRepository = $messageGroupRepository;
         $this->errorProcessor = $errorProcessor;
     }
 
@@ -47,7 +47,7 @@ public function aroundDispatch(
         $messageGroupId = $this->request->getHeader('X-Message-Group-ID', $default = false);
         $messageTimestamp = $this->request->getHeader('X-Message-Timestamp', $default = false);
 
-        if ($messageGroupId === false) {
+        if ($messageGroupId === false || $messageTimestamp === false) {
             return $proceed($request);
         }
 
@@ -57,17 +57,13 @@ public function aroundDispatch(
         }
 
         try {
-            if ($messageTimestamp !== false &&
-                $lastModificationTime = $this->modificationTimeRepo->getLastModificationTime($messageGroupId)
-            ) {
-                if (!$this->isUnmodifiedSince($lastModificationTime, $messageTimestamp)) {
-                    $this->response->setStatusCode(412);
-                    return $this->response;
-                }
-                $updateTimestamp = $messageTimestamp;
-            } else {
-                $lastModificationTime = null;
-                $updateTimestamp = \time();
+            $messageGroup = $this->messageGroupRepository->getMessageGroup($messageGroupId);
+            $lastTimestamp = $messageGroup['timestamp'] ?? null;
+            $lastVersion = $messageGroup['version'] ?? null;
+
+            if ($lastTimestamp !== null && $messageTimestamp < $lastTimestamp) {
+                $this->response->setStatusCode(412);
+                return $this->response;
             }
 
             $connection = $this->resourceConnection->getConnection();
@@ -83,17 +79,21 @@ public function aroundDispatch(
                     return $response;
                 }
 
-                $this->modificationTimeRepo->updateModificationTime(
+                $this->messageGroupRepository->updateModificationTime(
                     $messageGroupId,
-                    $updateTimestamp,
-                    $lastModificationTime
+                    $messageTimestamp,
+                    $lastVersion,
+                    $lastTimestamp
                 );
                 $connection->commit();
                 return $response;
             } catch (\Throwable $e) {
                 $connection->rollBack();
                 throw $e;
             }
+        } catch (ConflictException $e) {
+            $this->response->setStatusCode(409);
+            return $this->response;
         } catch (\Exception $e) {
             $e = $this->errorProcessor->maskException($e);
             $this->response->setStatusCode($e->getHttpCode());
@@ -105,9 +105,4 @@ public function aroundDispatch(
             $this->lockService->releaseLock("idempotent_api.$messageGroupId");
         }
     }
-
-    private function isUnmodifiedSince(int $modificationTime, int $expectedTime)
-    {
-        return $modificationTime <= $expectedTime;
-    }
 }

Model/WebApiMessageGroupRepository.php

@@ -0,0 +1,55 @@
+<?php
+
+namespace SnowIO\IdempotentAPI\Model;
+
+use Magento\Framework\Model\ResourceModel\Db\Context;
+
+class WebApiMessageGroupRepository
+{
+    private $dbConnection;
+
+    public function __construct(Context $dbContext, $connectionName = null)
+    {
+        $this->dbConnection = $dbContext->getResources()->getConnection($connectionName);
+    }
+
+    public function getMessageGroup(string $id)
+    {
+        $id = md5($id);
+        $select = $this->dbConnection->select()
+            ->from(['t' => $this->dbConnection->getTableName('webapi_message_group')], ['timestamp', 'version'])
+            ->where('t.id = ?', $id);
+        $result = $this->dbConnection->fetchAssoc($select);
+
+        return array_shift($result);
+    }
+
+    public function updateModificationTime(
+        string $id,
+        string $modificationTime,
+        int $expectedVersion = null,
+        string $expectedModificationTime = null
+    ) {
+        $id = md5($id);
+
+        if ($expectedModificationTime !== null) {
+            $rowsAffected = $this->dbConnection->update(
+                $this->dbConnection->getTableName('webapi_message_group'),
+                [
+                    'timestamp' => $modificationTime,
+                    'version' => $expectedVersion + 1
+                ],
+                ['id = ?' => $id, 'timestamp = ?' => $expectedModificationTime, 'version = ?' => $expectedVersion]
+            );
+        } else {
+            $rowsAffected = $this->dbConnection->insert(
+                $this->dbConnection->getTableName('webapi_message_group'),
+                ['id' => $id, 'timestamp' => $modificationTime, 'version' => 1]
+            );
+        }
+
+        if ($rowsAffected === 0) {
+            throw new ConflictException;
+        }
+    }
+}

Setup/InstallSchema.php

@@ -22,19 +22,25 @@ public function install(SchemaSetupInterface $setup, ModuleContextInterface $con
         $installer = $setup;
         $installer->startSetup();
         $table = $installer->getConnection()->newTable(
-            $installer->getTable('webapi_resource_modification_log')
+            $installer->getTable('webapi_message_group')
         )->addColumn(
-            'identifier',
+            'id',
             Table::TYPE_TEXT,
             255,
             ['primary' => true, 'nullable' => false],
-            'Resource Identifier'
+            'Message Group ID'
         )->addColumn(
             'timestamp',
-            Table::TYPE_DECIMAL,
-            null,
+            Table::TYPE_TEXT,
+            255,
             ['unsigned' => true, 'nullable' => false],
-            'Timestamp'
+            'Message Timestamp'
+        )->addColumn(
+            'version',
+            Table::TYPE_INTEGER,
+            255,
+            ['unsigned' => true, 'nullable' => false, 'default' => 1],
+            'Version'
         );
         $installer->getConnection()->createTable($table);
         $installer->endSetup();

Test/Unit/Model/DispatchPluginTest.php

@@ -11,7 +11,7 @@
 use PHPUnit_Framework_MockObject_MockObject;
 use PHPUnit_Framework_TestCase;
 use SnowIO\IdempotentAPI\Model\DispatchPlugin;
-use SnowIO\IdempotentAPI\Model\ResourceModificationTimeRepository;
+use SnowIO\IdempotentAPI\Model\WebApiMessageGroupRepository;
 use SnowIO\Lock\Api\LockService;
 use Magento\Framework\Webapi\Response;
 use \Magento\Framework\App\RequestInterface;
@@ -20,7 +20,7 @@
 
 class DispatchPluginTest extends PHPUnit_Framework_TestCase
 {
-    /** @var  ResourceModificationTimeRepository | PHPUnit_Framework_MockObject_MockObject */
+    /** @var  WebApiMessageGroupRepository | PHPUnit_Framework_MockObject_MockObject */
     private $mockResourceTimestampRespository;
 
     /** @var  LockService | PHPUnit_Framework_MockObject_MockObject */
@@ -56,7 +56,7 @@ public function __construct($name = null, array $data = array(), $dataName = '')
 
     public function setUp()
     {
-        $this->mockResourceTimestampRespository = $this->getMockBuilder(ResourceModificationTimeRepository::class)
+        $this->mockResourceTimestampRespository = $this->getMockBuilder(WebApiMessageGroupRepository::class)
             ->disableOriginalConstructor()->setMethods([
                 'getLastModificationTime',
                 'updateModificationTime'