Atomic graph mutation (#124)

* Fix schema use with graph

* Add $startTransaction params

* allow for atomic graph upsert/insert
* allow for atomic multi create

* Fix $startTransaction (atomic)

* Is now $atomic
* Better testing
* README infos

Co-authored-by: Dekel Barzilay <[email protected]>

Author: Thomas-git

README.md

@@ -246,6 +246,8 @@ Note that all this eager related options are optional.
   [`transaction`](https://vincit.github.io/objection.js/api/objection/#transaction)
   documentation.
 
+- **`$atomic`** - when `true` ensure that multi create or graph insert/upsert success or fail all at once. Under the hood, automaticaly create a transaction and commit on success or rollback on partial or total failure. __Ignored__ if you added your own `transaction` object in params.
+
 - **`mergeAllowEager`** - Will merge the given expression to the existing expression from the `allowEager` service option. 
   See [`allowGraph`](https://vincit.github.io/objection.js/api/query-builder/eager-methods.html#allowgraph)
   documentation.

src/index.js

@@ -314,12 +314,46 @@ class Service extends AdapterService {
     return null;
   }
 
+  async _createTransaction (params) {
+    if (!params.transaction && params.$atomic) {
+      delete params.$atomic;
+      params.transaction = params.transaction || {};
+      params.transaction.trx = await this.Model.startTransaction();
+      return params.transaction;
+    }
+    return null;
+  }
+
+  _commitTransaction (transaction) {
+    return async (data) => {
+      if (transaction) {
+        await transaction.trx.commit();
+      }
+      return data;
+    };
+  }
+
+  _rollbackTransaction (transaction) {
+    return async (err) => {
+      if (transaction) {
+        await transaction.trx.rollback();
+      }
+      throw err;
+    };
+  }
+
   _createQuery (params = {}) {
     const trx = params.transaction ? params.transaction.trx : null;
     const schema = params.schema || this.schema;
     const query = this.Model.query(trx);
-
-    return schema ? query.withSchema(schema) : query;
+    if (schema) {
+      query.context({
+        onBuild (builder) {
+          builder.withSchema(schema);
+        }
+      });
+    }
+    return query;
   }
 
   _selectQuery (q, $select) {
@@ -587,7 +621,8 @@ class Service extends AdapterService {
    * @param {object} data
    * @param {object} params
    */
-  _create (data, params) {
+  async _create (data, params) {
+    const transaction = await this._createTransaction(params);
     const create = (data, params) => {
       const q = this._createQuery(params);
       const allowedUpsert = this.mergeRelations(this.allowedUpsert, params.mergeAllowUpsert);
@@ -604,6 +639,7 @@ class Service extends AdapterService {
       } else {
         q.insert(data, this.id);
       }
+
       return q
         .then(row => {
           if (params.query && params.query.$noSelect) { return data; }
@@ -628,10 +664,10 @@ class Service extends AdapterService {
     };
 
     if (Array.isArray(data)) {
-      return Promise.all(data.map(current => create(current, params)));
+      return Promise.all(data.map(current => create(current, params))).then(this._commitTransaction(transaction), this._rollbackTransaction(transaction));
     }
 
-    return create(data, params);
+    return create(data, params).then(this._commitTransaction(transaction), this._rollbackTransaction(transaction));
   }
 
   /**
@@ -648,13 +684,17 @@ class Service extends AdapterService {
         // that we can fill any existing keys that the
         // client isn't updating with null;
         return this.Model.fetchTableMetadata()
-          .then(meta => {
+          .then(async meta => {
             let newObject = Object.assign({}, data);
+            let transaction = null;
 
             const allowedUpsert = this.mergeRelations(this.allowedUpsert, params.mergeAllowUpsert);
+
             if (allowedUpsert) {
               // Ensure the object we fetched is the one we update
               this._checkUpsertId(id, newObject);
+              // Create transaction if needed
+              transaction = await this._createTransaction(params);
             }
 
             for (const key of meta.columns) {
@@ -666,7 +706,7 @@ class Service extends AdapterService {
             if (allowedUpsert) {
               return this._createQuery(params)
                 .allowGraph(allowedUpsert)
-                .upsertGraphAndFetch(newObject, this.upsertGraphOptions);
+                .upsertGraphAndFetch(newObject, this.upsertGraphOptions).then(this._commitTransaction(transaction), this._rollbackTransaction(transaction));
             }
 
             // NOTE (EK): Delete id field so we don't update it
@@ -710,11 +750,13 @@ class Service extends AdapterService {
       this._checkUpsertId(id, dataCopy);
 
       // Get object first to ensure it satisfy user query
-      return this._get(id, params).then(() => {
+      return this._get(id, params).then(async () => {
+        // Create transaction if needed
+        const transaction = await this._createTransaction(params);
         return this._createQuery(params)
           .allowGraph(allowedUpsert)
           .upsertGraphAndFetch(dataCopy, this.upsertGraphOptions)
-          .then(this._selectFields(params, data));
+          .then(this._selectFields(params, data)).then(this._commitTransaction(transaction), this._rollbackTransaction(transaction));
       });
     }
 

test/index.test.js

@@ -15,7 +15,7 @@ import PeopleRoomsCustomIdSeparator from './people-rooms-custom-id-separator';
 import Company from './company';
 import Employee from './employee';
 import Client from './client';
-import { Model } from 'objection';
+import { Model, UniqueViolationError } from 'objection';
 
 const testSuite = adapterTests([
   '.options',
@@ -238,6 +238,7 @@ function clean (done) {
               table.json('jsonArray');
               table.jsonb('jsonbObject');
               table.jsonb('jsonbArray');
+              table.unique('name');
             });
           });
         });
@@ -260,6 +261,7 @@ function clean (done) {
             table.increments('id');
             table.integer('companyId');
             table.string('name');
+            table.unique('name');
           })
           .then(() => done());
       });
@@ -1063,18 +1065,18 @@ describe('Feathers Objection Service', () => {
       return companies
         .create([
           {
-            name: 'Google',
+            name: 'Facebook',
             clients: [
               {
-                name: 'Dan Davis'
+                name: 'Danny Lapierre'
               },
               {
-                name: 'Ken Patrick'
+                name: 'Kirck Filty'
               }
             ]
           },
           {
-            name: 'Apple'
+            name: 'Yubico'
           }
         ]).then(() => {
           companies.createUseUpsertGraph = false;
@@ -1844,6 +1846,157 @@ describe('Feathers Objection Service', () => {
         });
       });
     });
+
+    it('works with atomic', () => {
+      return people.create({ name: 'Rollback' }, { transaction, $atomic: true }).then(() => {
+        expect(transaction.trx.isCompleted()).to.equal(false); // Atomic must be ignored and transaction still running
+        return transaction.trx.rollback().then(() => {
+          return people.find({ query: { name: 'Rollback' } }).then((data) => {
+            expect(data.length).to.equal(0);
+          });
+        });
+      });
+    });
+  });
+
+  describe('Atomic Transactions', () => {
+    before(async () => {
+      await companies
+        .create([
+          {
+            name: 'Google',
+            clients: [
+              {
+                name: 'Dan Davis'
+              },
+              {
+                name: 'Ken Patrick'
+              }
+            ]
+          },
+          {
+            name: 'Apple'
+          }
+        ]);
+    });
+
+    after(async () => {
+      await clients.remove(null);
+      await companies.remove(null);
+    });
+
+    it('Rollback on sub insert failure', () => {
+      // Dan Davis already exists
+      return companies.create({ name: 'Compaq', clients: [{ name: 'Dan Davis' }] }, { $atomic: true }).catch((error) => {
+        expect(error instanceof errors.GeneralError).to.be.ok;
+        expect(error.message).to.match(/SQLITE_CONSTRAINT: UNIQUE/);
+        return companies.find({ query: { name: 'Compaq', $eager: 'clients' } }).then(
+          (data) => {
+            expect(data.length).to.equal(0);
+          });
+      });
+    });
+
+    it('Rollback on multi insert failure', () => {
+      // Google already exists
+      return companies.create([{ name: 'Google' }, { name: 'Compaq' }], { $atomic: true }).catch((error) => {
+        expect(error instanceof errors.GeneralError).to.be.ok;
+        expect(error.message).to.match(/SQLITE_CONSTRAINT: UNIQUE/);
+        return companies.find({ query: { name: 'Compaq' } }).then(
+          (data) => {
+            expect(data.length).to.equal(0);
+          });
+      });
+    });
+
+    it('Rollback on update failure', () => {
+      // Dan Davis appears twice, so clients must stay as it is
+      return companies.find({ query: { name: 'Google' } }).then(data => {
+        return companies.update(data[0].id, {
+          name: 'Google',
+          clients: [
+            {
+              name: 'Dan Davis'
+            },
+            {
+              name: 'Dan Davis'
+            },
+            {
+              name: 'Kirk Maelström'
+            }
+          ]
+        }, { $atomic: true }).catch((error) => {
+          expect(error instanceof errors.GeneralError).to.be.ok;
+          expect(error.message).to.match(/SQLITE_CONSTRAINT: UNIQUE/);
+          return companies.find({ query: { name: 'Google', $eager: 'clients' } }).then(
+            (data) => {
+              expect(data.length).to.equal(1);
+              expect(data[0].clients.length).to.equal(2);
+              expect(data[0].clients[0].name).to.equal('Dan Davis');
+              expect(data[0].clients[1].name).to.equal('Ken Patrick');
+            });
+        });
+      });
+    });
+
+    it('Rollback on patch failure', () => {
+      // Dan Davis appears twice, so clients must stay as it is
+      return companies.find({ query: { name: 'Google' } }).then(data => {
+        return companies.patch(data[0].id, {
+          clients: [
+            {
+              name: 'Dan Davis'
+            },
+            {
+              name: 'Dan Davis'
+            },
+            {
+              name: 'Kirk Maelström'
+            }
+          ]
+        }, { $atomic: true }).catch((error) => {
+          expect(error instanceof UniqueViolationError).to.be.ok;
+          expect(error.message).to.match(/SQLITE_CONSTRAINT: UNIQUE/);
+          return companies.find({ query: { name: 'Google', $eager: 'clients' } }).then(
+            (data) => {
+              expect(data.length).to.equal(1);
+              expect(data[0].clients.length).to.equal(2);
+              expect(data[0].clients[0].name).to.equal('Dan Davis');
+              expect(data[0].clients[1].name).to.equal('Ken Patrick');
+            });
+        });
+      });
+    });
+
+    it('Commit on patch success', () => {
+      // Dan Davis appears twice, so clients must stay as it is
+      return companies.find({ query: { name: 'Google' } }).then(data => {
+        return companies.patch(data[0].id, {
+          clients: [
+            {
+              name: 'Dan David'
+            },
+            {
+              name: 'Dan Davis'
+            },
+            {
+              name: 'Kirk Maelström'
+            }
+          ]
+        }, { $atomic: true }).catch((error) => {
+          expect(error instanceof UniqueViolationError).to.be.ok;
+          expect(error.message).to.match(/SQLITE_CONSTRAINT: UNIQUE/);
+          return companies.find({ query: { name: 'Google', $eager: 'clients' } }).then(
+            (data) => {
+              expect(data.length).to.equal(1);
+              expect(data[0].clients.length).to.equal(3);
+              expect(data[0].clients[0].name).to.equal('Dan David');
+              expect(data[0].clients[0].name).to.equal('Dan Davis');
+              expect(data[0].clients[1].name).to.equal('Kirk Maelström');
+            });
+        });
+      });
+    });
   });
 
   describe('$noSelect', () => {