Transaction support

.github/workflows/build-ci.yml

@@ -44,10 +44,6 @@ jobs:
                     - '8.0'
                     - '8.1'
         services:
-            mongo:
-                image: mongo:${{ matrix.mongodb }}
-                ports:
-                    - 27017:27017
             mysql:
                 image: mysql:5.7
                 ports:
@@ -59,6 +55,13 @@ jobs:
 
         steps:
             -   uses: actions/checkout@v2
+            -   name: Create MongoDB Replica Set
+                run: |
+                    docker run --name mongodb -p 27017:27017 -e MONGO_INITDB_DATABASE=unittest --detach mongo:${{ matrix.mongodb }} mongod --replSet rs
+                    until docker exec --tty mongodb mongo 127.0.0.1:27017 --eval "db.runCommand({ ping: 1 })"; do
+                    sleep 1
+                    done
+                    sudo docker exec --tty mongodb mongo 127.0.0.1:27017 --eval "rs.initiate({\"_id\":\"rs\",\"members\":[{\"_id\":0,\"host\":\"127.0.0.1:27017\" }]})"
             -   name: "Installing php"
                 uses: shivammathur/setup-php@v2
                 with:
@@ -88,7 +91,7 @@ jobs:
                 run: |
                     ./vendor/bin/phpunit --coverage-clover coverage.xml
                 env:
-                    MONGODB_URI: 'mongodb://127.0.0.1/'
+                    MONGODB_URI: 'mongodb://127.0.0.1/?replicaSet=rs'
                     MYSQL_HOST: 0.0.0.0
                     MYSQL_PORT: 3307
             -   uses: codecov/codecov-action@v1

README.md

@@ -37,6 +37,7 @@ This package adds functionalities to the Eloquent model and Query builder for Mo
   - [Query Builder](#query-builder)
     - [Basic Usage](#basic-usage-2)
     - [Available operations](#available-operations)
+  - [Transactions](#transactions)
   - [Schema](#schema)
     - [Basic Usage](#basic-usage-3)
     - [Geospatial indexes](#geospatial-indexes)
@@ -968,6 +969,52 @@ If you are familiar with [Eloquent Queries](http://laravel.com/docs/queries), th
 ### Available operations
 To see the available operations, check the [Eloquent](#eloquent) section.
 
+Transactions
+------------
+Transactions require MongoDB version ^4.0 as well as deployment of replica set or sharded clusters. You can find more information [in the MongoDB docs](https://docs.mongodb.com/manual/core/transactions/)
+
+### Basic Usage
+
+```php
+DB::transaction(function () {
+    User::create(['name' => 'john', 'age' => 19, 'title' => 'admin', 'email' => '[email protected]']);
+    DB::collection('users')->where('name', 'john')->update(['age' => 20]);
+    DB::collection('users')->where('name', 'john')->delete();
+});
+```
+
+```php
+// begin a transaction
+DB::beginTransaction();
+User::create(['name' => 'john', 'age' => 19, 'title' => 'admin', 'email' => '[email protected]']);
+DB::collection('users')->where('name', 'john')->update(['age' => 20]);
+DB::collection('users')->where('name', 'john')->delete();
+
+// commit changes
+DB::commit();
+```
+
+To abort a transaction, call the `rollBack` method at any point during the transaction:
+```php
+DB::beginTransaction();
+User::create(['name' => 'john', 'age' => 19, 'title' => 'admin', 'email' => '[email protected]']);
+
+// Abort the transaction, discarding any data created as part of it
+DB::rollBack();
+```
+
+**NOTE:** Transactions in MongoDB cannot be nested. DB::beginTransaction() function will start new transactions in a new created or existing session and will raise the RuntimeException when transactions already exist. See more in MongoDB official docs [Transactions and Sessions](https://www.mongodb.com/docs/manual/core/transactions/#transactions-and-sessions)
+```php
+DB::beginTransaction();
+User::create(['name' => 'john', 'age' => 20, 'title' => 'admin']);
+
+// This call to start a nested transaction will raise a RuntimeException
+DB::beginTransaction();
+DB::collection('users')->where('name', 'john')->update(['age' => 20]);
+DB::commit();
+DB::rollBack();
+```
+
 Schema
 ------
 The database driver also has (limited) schema builder support. You can easily manipulate collections and set indexes.

phpunit.xml.dist

@@ -19,6 +19,9 @@
             <file>tests/QueryBuilderTest.php</file>
             <file>tests/QueryTest.php</file>
         </testsuite>
+        <testsuite name="transaction">
+            <file>tests/TransactionTest.php</file>
+        </testsuite>
         <testsuite name="model">
             <file>tests/ModelTest.php</file>
             <file>tests/RelationsTest.php</file>
@@ -36,7 +39,7 @@
     </testsuites>
     <php>
         <env name="MONGODB_URI" value="mongodb://127.0.0.1/" />
-        <env name="MONGO_DATABASE" value="unittest"/>
+        <env name="MONGODB_DATABASE" value="unittest"/>
         <env name="MYSQL_HOST" value="mysql"/>
         <env name="MYSQL_PORT" value="3306"/>
         <env name="MYSQL_DATABASE" value="unittest"/>

src/Concerns/ManagesTransactions.php

@@ -0,0 +1,116 @@
+<?php
+
+namespace Jenssegers\Mongodb\Concerns;
+
+use Closure;
+use MongoDB\Client;
+use MongoDB\Driver\Exception\RuntimeException;
+use MongoDB\Driver\Session;
+use function MongoDB\with_transaction;
+use Throwable;
+
+/**
+ * @see https://docs.mongodb.com/manual/core/transactions/
+ */
+trait ManagesTransactions
+{
+    protected ?Session $session = null;
+
+    protected $transactions = 0;
+
+    /**
+     * @return Client
+     */
+    abstract public function getMongoClient();
+
+    public function getSession(): ?Session
+    {
+        return $this->session;
+    }
+
+    private function getSessionOrCreate(): Session
+    {
+        if ($this->session === null) {
+            $this->session = $this->getMongoClient()->startSession();
+        }
+
+        return $this->session;
+    }
+
+    private function getSessionOrThrow(): Session
+    {
+        $session = $this->getSession();
+
+        if ($session === null) {
+            throw new RuntimeException('There is no active session.');
+        }
+
+        return $session;
+    }
+
+    /**
+     * Starts a transaction on the active session. An active session will be created if none exists.
+     */
+    public function beginTransaction(array $options = []): void
+    {
+        $this->getSessionOrCreate()->startTransaction($options);
+        $this->transactions = 1;
+    }
+
+    /**
+     * Commit transaction in this session.
+     */
+    public function commit(): void
+    {
+        $this->getSessionOrThrow()->commitTransaction();
+        $this->transactions = 0;
+    }
+
+    /**
+     * Abort transaction in this session.
+     */
+    public function rollBack($toLevel = null): void
+    {
+        $this->getSessionOrThrow()->abortTransaction();
+        $this->transactions = 0;
+    }
+
+    /**
+     * Static transaction function realize the with_transaction functionality provided by MongoDB.
+     *
+     * @param  int  $attempts
+     */
+    public function transaction(Closure $callback, $attempts = 1, array $options = []): mixed
+    {
+        $attemptsLeft = $attempts;
+        $callbackResult = null;
+        $throwable = null;
+
+        $callbackFunction = function (Session $session) use ($callback, &$attemptsLeft, &$callbackResult, &$throwable) {
+            $attemptsLeft--;
+
+            if ($attemptsLeft < 0) {
+                $session->abortTransaction();
+
+                return;
+            }
+
+            // Catch, store and re-throw any exception thrown during execution
+            // of the callable. The last exception is re-thrown if the transaction
+            // was aborted because the number of callback attempts has been exceeded.
+            try {
+                $callbackResult = $callback($this);
+            } catch (Throwable $throwable) {
+                throw $throwable;
+            }
+        };
+
+        with_transaction($this->getSessionOrCreate(), $callbackFunction, $options);
+
+        if ($attemptsLeft < 0 && $throwable) {
+            throw $throwable;
+        }
+
+        return $callbackResult;
+    }
+}

src/Connection.php

@@ -5,11 +5,14 @@
 use Illuminate\Database\Connection as BaseConnection;
 use Illuminate\Support\Arr;
 use InvalidArgumentException;
+use Jenssegers\Mongodb\Concerns\ManagesTransactions;
 use MongoDB\Client;
 use MongoDB\Database;
 
 class Connection extends BaseConnection
 {
+    use ManagesTransactions;
+
     /**
      * The MongoDB database handler.
      *

src/Query/Builder.php

@@ -346,6 +346,8 @@ public function getFresh($columns = [], $returnLazy = false)
                 $options = array_merge($options, $this->options);
             }
 
+            $options = $this->inheritConnectionOptions($options);
+
             // Execute aggregation
             $results = iterator_to_array($this->collection->aggregate($pipeline, $options));
 
@@ -356,12 +358,10 @@ public function getFresh($columns = [], $returnLazy = false)
             // Return distinct results directly
             $column = isset($this->columns[0]) ? $this->columns[0] : '_id';
 
+            $options = $this->inheritConnectionOptions();
+
             // Execute distinct
-            if ($wheres) {
-                $result = $this->collection->distinct($column, $wheres);
-            } else {
-                $result = $this->collection->distinct($column);
-            }
+            $result = $this->collection->distinct($column, $wheres ?: [], $options);
 
             return new Collection($result);
         } // Normal query
@@ -407,6 +407,8 @@ public function getFresh($columns = [], $returnLazy = false)
                 $options = array_merge($options, $this->options);
             }
 
+            $options = $this->inheritConnectionOptions($options);
+
             // Execute query and get MongoCursor
             $cursor = $this->collection->find($wheres, $options);
 
@@ -581,8 +583,9 @@ public function insert(array $values)
             $values = [$values];
         }
 
-        // Batch insert
-        $result = $this->collection->insertMany($values);
+        $options = $this->inheritConnectionOptions();
+
+        $result = $this->collection->insertMany($values, $options);
 
         return 1 == (int) $result->isAcknowledged();
     }
@@ -592,7 +595,9 @@ public function insert(array $values)
      */
     public function insertGetId(array $values, $sequence = null)
     {
-        $result = $this->collection->insertOne($values);
+        $options = $this->inheritConnectionOptions();
+
+        $result = $this->collection->insertOne($values, $options);
 
         if (1 == (int) $result->isAcknowledged()) {
             if ($sequence === null) {
@@ -614,6 +619,8 @@ public function update(array $values, array $options = [])
             $values = ['$set' => $values];
         }
 
+        $options = $this->inheritConnectionOptions($options);
+
         return $this->performUpdate($values, $options);
     }
 
@@ -635,6 +642,8 @@ public function increment($column, $amount = 1, array $extra = [], array $option
             $query->orWhereNotNull($column);
         });
 
+        $options = $this->inheritConnectionOptions($options);
+
         return $this->performUpdate($query, $options);
     }
 
@@ -696,7 +705,10 @@ public function delete($id = null)
         }
 
         $wheres = $this->compileWheres();
-        $result = $this->collection->DeleteMany($wheres);
+        $options = $this->inheritConnectionOptions();
+
+        $result = $this->collection->deleteMany($wheres, $options);
+
         if (1 == (int) $result->isAcknowledged()) {
             return $result->getDeletedCount();
         }
@@ -721,7 +733,8 @@ public function from($collection, $as = null)
      */
     public function truncate(): bool
     {
-        $result = $this->collection->deleteMany([]);
+        $options = $this->inheritConnectionOptions();
+        $result = $this->collection->deleteMany([], $options);
 
         return 1 === (int) $result->isAcknowledged();
     }
@@ -855,6 +868,8 @@ protected function performUpdate($query, array $options = [])
             $options['multiple'] = true;
         }
 
+        $options = $this->inheritConnectionOptions($options);
+
         $wheres = $this->compileWheres();
         $result = $this->collection->UpdateMany($wheres, $query, $options);
         if (1 == (int) $result->isAcknowledged()) {
@@ -1249,6 +1264,18 @@ public function options(array $options)
         return $this;
     }
 
+    /**
+     * Apply the connection's session to options if it's not already specified.
+     */
+    private function inheritConnectionOptions(array $options = []): array
+    {
+        if (! isset($options['session']) && ($session = $this->connection->getSession())) {
+            $options['session'] = $session;
+        }
+
+        return $options;
+    }
+
     /**
      * @inheritdoc
      */