Added transaction helper

.travis.yml

@@ -8,6 +8,9 @@ node_js:
 services:
   - postgresql
 
+before_script:
+  - psql -d postgres -c 'CREATE TABLE users(id serial PRIMARY KEY, username VARCHAR (50) NOT NULL);' -U postgres
+
 notifications:
   email:
     on_success: never

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,54 @@ 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 => {
+    try {
+      const id = await client.query('INSERT INTO users(username) VALUES($1) RETURNING id', [req.params.username])
+      reply.send(id)
+    } catch (err) {
+      reply.send(err)
+    }
+  })
+})
+
+/* or with a transaction callback
+
+fastify.pg.transact(client => {
+    return client.query('INSERT INTO users(username) VALUES($1) RETURNING id', [req.params.username])
+  },
+  function onResult (err, result) {
+    reply.send(err || result)
+  }
+})
+
+*/
+
+/* or with a commit callback
+
+fastify.pg.transact((client, commit) => { 
+  client.query('INSERT INTO users(username) VALUES($1) RETURNING id', [req.params.username], (err, id) => {
+    commit(err, id)
+  });
+})
+
+*/
+
+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,58 @@
 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
+    }
+
+    const commit = (err, res) => {
+      if (shouldAbort(err)) return cb(err)
+
+      client.query('COMMIT', (err) => {
+        done()
+        if (err) {
+          return cb(err)
+        }
+        return cb(null, res)
+      })
+    }
+
+    client.query('BEGIN', (err) => {
+      if (shouldAbort(err)) return cb(err)
+
+      const promise = fn(client, commit)
+
+      if (promise && typeof promise.then === 'function') {
+        promise.then(
+          (res) => commit(null, res),
+          (e) => commit(e))
+      }
+    })
+  })
+}
+
+function transact (fn, cb) {
+  if (cb && typeof cb === 'function') {
+    return transactionUtil(this, fn, cb)
+  }
+
+  return new Promise((resolve, reject) => {
+    transactionUtil(this, fn, function (err, res) {
+      if (err) { return reject(err) }
+      return resolve(res)
+    })
+  })
+}
+
 function fastifyPostgres (fastify, options, next) {
   if (options.native) {
     delete options.native
@@ -21,7 +73,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,97 @@ 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()
+  t.tearDown(fastify.close.bind(fastify))
+
+  fastify.register(fastifyPostgres, {
+    name: 'test',
+    connectionString: 'postgres://postgres@localhost/postgres'
+  })
+
+  fastify.ready(err => {
+    t.error(err)
+    fastify.pg.test
+      .transact(client => client.query('INSERT INTO users(username) VALUES($1) RETURNING id', ['with-promise']))
+      .then(result => {
+        t.equals(result.rows.length, 1)
+        fastify.pg.test
+          .query(`SELECT * FROM users WHERE username = 'with-promise'`)
+          .then(result => {
+            t.ok(result.rows[0].username === 'with-promise')
+          }).catch(err => {
+            t.fail(err)
+          })
+      })
+      .catch(err => {
+        t.fail(err)
+      })
+  })
+})
+
+test('fastify.pg.test use transact util with callback', t => {
+  t.plan(4)
+
+  const fastify = Fastify()
+  t.tearDown(fastify.close.bind(fastify))
+
+  fastify.register(fastifyPostgres, {
+    name: 'test',
+    connectionString: 'postgres://postgres@localhost/postgres'
+  })
+
+  fastify.ready(err => {
+    t.error(err)
+
+    fastify.pg.test
+      .transact(client => client.query('INSERT INTO users(username) VALUES($1) RETURNING id', ['with-callback']), function (err, res) {
+        t.error(err)
+        t.equals(res.rows.length, 1)
+
+        fastify.pg.test
+          .query(`SELECT * FROM users WHERE username = 'with-callback'`)
+          .then(result => {
+            t.ok(result.rows[0].username === 'with-callback')
+          }).catch(err => {
+            t.fail(err)
+          })
+      })
+  })
+})
+
+test('fastify.pg.test use transact util with commit callback', t => {
+  t.plan(4)
+
+  const fastify = Fastify()
+  t.tearDown(fastify.close.bind(fastify))
+
+  fastify.register(fastifyPostgres, {
+    name: 'test',
+    connectionString: 'postgres://postgres@localhost/postgres'
+  })
+
+  fastify.ready(err => {
+    t.error(err)
+
+    fastify.pg.test.transact((client, commit) => {
+      client.query('INSERT INTO users(username) VALUES($1) RETURNING id', ['commit-callback'], (err, id) => {
+        commit(err, id)
+      })
+    }, function (err, res) {
+      t.error(err)
+      t.equals(res.rows.length, 1)
+
+      fastify.pg.test
+        .query(`SELECT * FROM users WHERE username = 'commit-callback'`)
+        .then(result => {
+          t.ok(result.rows[0].username === 'commit-callback')
+        }).catch(err => {
+          t.fail(err)
+        })
+    })
+  })
+})