Added transaction helper

README.md

@@ -18,7 +18,8 @@ This plugin will add the `pg` namespace in your Fastify instance, with the follo
 connect: the function to get a connection from the pool
 pool: the pool instance
 Client: a client constructor for a single query
-query: an utility to perform a query without a transaction
+query: a utility to perform a query _without_ a transaction
+transact: a utility to perform multiple queries _with_ a transaction
 ```
 
 Example:
@@ -95,6 +96,31 @@ fastify.listen(3000, err => {
   console.log(`server listening on ${fastify.server.address().port}`)
 })
 ```
+
+Use of `pg.transact`
+```js
+const fastify = require('fastify')
+
+fastify.register(require('fastify-postgres'), {
+  connectionString: 'postgres://postgres@localhost/postgres'
+})
+
+fastify.post('/user/:username', (req, reply) => {
+  fastify.pg.transact(async client => {
+      return client.query('INSERT INTO users(username) VALUES($1) RETURNING id', [req.params.username])
+    },
+    function onResult (err, result) {
+      reply.send(err || result)
+    }
+  )
+})
+
+fastify.listen(3000, err => {
+  if (err) throw err
+  console.log(`server listening on ${fastify.server.address().port}`)
+})
+```
+
 As you can see there is no need to close the client, since is done internally. Promises and async await are supported as well.
 
 ### Native option

index.js

@@ -3,6 +3,50 @@
 const fp = require('fastify-plugin')
 var pg = require('pg')
 
+function transactionUtil (pool, fn, cb) {
+  pool.connect((err, client, done) => {
+    if (err) return cb(err)
+
+    const shouldAbort = (err) => {
+      if (err) {
+        client.query('ROLLBACK', () => {
+          done()
+        })
+      }
+      return !!err
+    }
+
+    client.query('BEGIN', (err) => {
+      if (shouldAbort(err)) return cb(err)
+
+      fn(client).then(res => {
+        client.query('COMMIT', (err) => {
+          done()
+          if (err) {
+            return cb(err)
+          }
+          return cb(null, res)
+        })
+      }).catch(err => {
+        if (shouldAbort(err)) return cb(err)
+      })
+    })
+  })
+}
+
+function transact (fn, cb) {
+  if (!cb) {
+    return new Promise((resolve, reject) => {
+      transactionUtil(this, fn, function (err, res) {
+        if (err) { return reject(err) }
+        return resolve(res)
+      })
+    })
+  }
+
+  return transactionUtil(this, fn, cb)
+}
+
 function fastifyPostgres (fastify, options, next) {
   if (options.native) {
     delete options.native
@@ -21,7 +65,8 @@ function fastifyPostgres (fastify, options, next) {
     connect: pool.connect.bind(pool),
     pool: pool,
     Client: pg.Client,
-    query: pool.query.bind(pool)
+    query: pool.query.bind(pool),
+    transact: transact.bind(pool)
   }
 
   if (name) {

test.js

@@ -253,3 +253,75 @@ test('fastify.pg.test should throw with duplicate connection names', t => {
     t.is(err.message, 'Connection name has already been registered: test')
   })
 })
+
+test('fastify.pg.test use transact util with promise', t => {
+  t.plan(3)
+
+  const fastify = Fastify()
+
+  fastify.register(fastifyPostgres, {
+    name: 'test',
+    connectionString: 'postgres://postgres@localhost/postgres'
+  })
+
+  fastify.ready(err => {
+    t.error(err)
+    fastify.pg.test
+      .query('CREATE TABLE users(id serial PRIMARY KEY, username VARCHAR (50) NOT NULL)')
+      .then(() => {
+        fastify.pg.test
+          .transact(client => { return client.query('INSERT INTO users(username) VALUES($1) RETURNING id', ['brianc']) })
+          .then(result => {
+            t.ok(result.rows[0].id === 1)
+            fastify.pg.test
+              .query('SELECT * FROM users')
+              .then(result => {
+                t.ok(result.rows[0].username === 'brianc')
+              }).catch(err => {
+                t.fail(err)
+                fastify.close()
+              })
+          })
+          .catch(err => {
+            t.fail(err)
+            fastify.close()
+          })
+      })
+      .catch(err => {
+        t.fail(err)
+        fastify.close()
+      })
+  })
+})
+
+test('fastify.pg.test use transact util with callback', t => {
+  t.plan(2)
+
+  const fastify = Fastify()
+
+  fastify.register(fastifyPostgres, {
+    name: 'test',
+    connectionString: 'postgres://postgres@localhost/postgres'
+  })
+
+  fastify.ready(err => {
+    t.error(err)
+    fastify.pg.test
+      .query('CREATE TABLE users2(id serial PRIMARY KEY, username VARCHAR (50) NOT NULL)')
+      .then(() => {
+        fastify.pg.test
+          .transact(client => { return client.query('INSERT INTO users2(username) VALUES($1) RETURNING id', ['brianc']) }, function (err, res) {
+            if (err) {
+              t.fail(err)
+              fastify.close()
+            } else {
+              t.ok(res.rows[0].id === 1)
+            }
+          })
+      })
+      .catch(err => {
+        t.fail(err)
+        fastify.close()
+      })
+  })
+})