Transactions on postgres

Author: davimacedo

src/Adapters/Storage/Postgres/PostgresStorageAdapter.js

@@ -1236,8 +1236,8 @@ export class PostgresStorageAdapter implements StorageAdapter {
   createObject(
     className: string,
     schema: SchemaType,
-    object: any
-    /* transactionalSession: ?any */
+    object: any,
+    transactionalSession: ?any
   ) {
     debug('createObject', className, object);
     let columnsArray = [];
@@ -1366,7 +1366,10 @@ export class PostgresStorageAdapter implements StorageAdapter {
     const qs = `INSERT INTO $1:name (${columnsPattern}) VALUES (${valuesPattern})`;
     const values = [className, ...columnsArray, ...valuesArray];
     debug(qs, values);
-    return this._client
+    const promise = (transactionalSession
+      ? transactionalSession.t
+      : this._client
+    )
       .none(qs, values)
       .then(() => ({ ops: [object] }))
       .catch(error => {
@@ -1386,6 +1389,10 @@ export class PostgresStorageAdapter implements StorageAdapter {
         }
         throw error;
       });
+    if (transactionalSession) {
+      transactionalSession.batch.push(promise);
+    }
+    return promise;
   }
 
   // Remove all objects that match the given Parse Query.
@@ -1394,8 +1401,8 @@ export class PostgresStorageAdapter implements StorageAdapter {
   deleteObjectsByQuery(
     className: string,
     schema: SchemaType,
-    query: QueryType
-    /* transactionalSession: ?any */
+    query: QueryType,
+    transactionalSession: ?any
   ) {
     debug('deleteObjectsByQuery', className, query);
     const values = [className];
@@ -1407,7 +1414,10 @@ export class PostgresStorageAdapter implements StorageAdapter {
     }
     const qs = `WITH deleted AS (DELETE FROM $1:name WHERE ${where.pattern} RETURNING *) SELECT count(*) FROM deleted`;
     debug(qs, values);
-    return this._client
+    const promise = (transactionalSession
+      ? transactionalSession.t
+      : this._client
+    )
       .one(qs, values, a => +a.count)
       .then(count => {
         if (count === 0) {
@@ -1425,6 +1435,10 @@ export class PostgresStorageAdapter implements StorageAdapter {
         }
         // ELSE: Don't delete anything if doesn't exist
       });
+    if (transactionalSession) {
+      transactionalSession.batch.push(promise);
+    }
+    return promise;
   }
   // Return value not currently well specified.
   findOneAndUpdate(
@@ -1449,8 +1463,8 @@ export class PostgresStorageAdapter implements StorageAdapter {
     className: string,
     schema: SchemaType,
     query: QueryType,
-    update: any
-    /* transactionalSession: ?any */
+    update: any,
+    transactionalSession: ?any
   ): Promise<[any]> {
     debug('updateObjectsByQuery', className, query, update);
     const updatePatterns = [];
@@ -1707,7 +1721,14 @@ export class PostgresStorageAdapter implements StorageAdapter {
       where.pattern.length > 0 ? `WHERE ${where.pattern}` : '';
     const qs = `UPDATE $1:name SET ${updatePatterns.join()} ${whereClause} RETURNING *`;
     debug('update: ', qs, values);
-    return this._client.any(qs, values);
+    const promise = (transactionalSession
+      ? transactionalSession.t
+      : this._client
+    ).any(qs, values);
+    if (transactionalSession) {
+      transactionalSession.batch.push(promise);
+    }
+    return promise;
   }
 
   // Hopefully, we can get rid of this. It's only used for config and hooks.
@@ -2359,15 +2380,34 @@ export class PostgresStorageAdapter implements StorageAdapter {
   }
 
   createTransactionalSession(): Promise<any> {
-    return Promise.resolve();
+    return new Promise(resolve => {
+      const transactionalSession = {};
+      transactionalSession.result = this._client.tx(t => {
+        transactionalSession.t = t;
+        transactionalSession.promise = new Promise(resolve => {
+          transactionalSession.resolve = resolve;
+        });
+        transactionalSession.batch = [];
+        resolve(transactionalSession);
+        return transactionalSession.promise;
+      });
+    });
   }
 
-  commitTransactionalSession(): Promise<void> {
-    return Promise.resolve();
+  commitTransactionalSession(transactionalSession: any): Promise<void> {
+    transactionalSession.resolve(
+      transactionalSession.t.batch(transactionalSession.batch)
+    );
+    return transactionalSession.result;
   }
 
-  abortTransactionalSession(): Promise<void> {
-    return Promise.resolve();
+  abortTransactionalSession(transactionalSession: any): Promise<void> {
+    const result = transactionalSession.result.catch();
+    transactionalSession.batch.push(Promise.reject());
+    transactionalSession.resolve(
+      transactionalSession.t.batch(transactionalSession.batch)
+    );
+    return result;
   }
 }