transaction.js

var inherits = require('inherits')
var FSM = require('yafsm')
var once = require('once')

module.exports = begin
begin.Transaction = Transaction

function begin (queryable, options, beginStatement, callback) {
  if (typeof options != 'object') {
    callback = beginStatement;
    beginStatement = options;
    options = {};
  }

  if (undefined === options.autoRollback) {
    options.autoRollback = true;
  }

  if (typeof beginStatement == 'function') {
    callback = beginStatement
    beginStatement = undefined
  }

  if (queryable instanceof Transaction) {
    return beginWithParent(queryable, options, callback)
  }

  var adapter = queryable.adapter;

  var tx = new Transaction({
    adapter:      adapter,
    begin:        beginStatement,
    callback:     callback,
    autoRollback: options.autoRollback
  })

  if (typeof adapter.createQuery != 'function' ||
      typeof queryable.query != 'function') {
    var error = new TypeError(queryable + ' is not a queryable!')
    if (callback) {
      callback(error)
    } else {
      throw error
    }
  }

  if (typeof queryable.acquire == 'function') {
    // it's a pool
    var pool = queryable
    pool.acquire(function (err, conn) {
      if (err) return process.nextTick(function () {
        tx.emit('error', err)
      })
      var release = pool.release.bind(pool, conn)
      tx.on('query', pool.emit.bind(pool, 'query'))
      tx.once('rollback:complete', release)
        .once('commit:complete', release)
        .setConnection(conn)
    })
  }
  else {
    // it's a connection
    tx.setConnection(queryable)
  }

  return tx
}

inherits(Transaction, FSM)
function Transaction(opts) {
  opts = opts || {}
  this.adapter = opts.adapter
  this._connection = null
  this._statements = {
    begin:    opts.begin    || 'BEGIN',
    commit:   opts.commit   || 'COMMIT',
    rollback: opts.rollback || 'ROLLBACK'
  }
  this._queue = []
  this._nestingLevel = opts.nestingLevel || 0
  this._autoRollback = opts.autoRollback;

  this._emitQuery = function (query) { this.emit('query', query) }.bind(this)
  this.handleError = this.handleError.bind(this)

  FSM.call(this, 'disconnected', {
    'disconnected': [ 'connected' ],
    'connected':    [ 'open', 'closed' ],
    'open':         [ 'connected', 'closed' ]
  })

  if (opts.callback) {
    var callback = opts.callback
    this
      .once('error', callback)
      .once('begin:complete', function () {
        this.removeListener('error', callback)
        callback(null, this)
      })
  }
}

Transaction.prototype.handleError = function (err, skipEmit) {
  var self = this
  var rollback = this.rollback.implementations['open']
  if (this.state() !== 'closed' && this._connection && this._autoRollback) {
    rollback.call(this, function (rollbackErr) {
      if (rollbackErr) {
        rollbackErr.previous = err;
        self.emit('error', rollbackErr)
      }
      else if (!skipEmit) {
        self.emit('error', err)
      }
    }, err)
  }
  else if (!skipEmit) self.emit('error', err)
}

Transaction.prototype.query = FSM.method('query', {
  'connected|disconnected': function (text, params, callback) {
    var query = this.adapter.createQuery(text, params, callback)
    this._queue.push(query)
    return query
  },
  'open': function (text, params, callback) {
    var self = this
    var query = this.adapter.createQuery(text, params, callback)
    query.once('error', function (err) {
      self.handleError(err, query.listeners('error').length)
    })
    return this._connection.query(query)
  }
})

;['commit', 'rollback'].forEach(function (methodName) {
  Transaction.prototype[methodName] = FSM.method(methodName, {
    'open': closeVia(methodName),
    'connected|disconnected': function (callback) {
      var fn = this[methodName].implementations['open']
      this._queue.push([fn, [callback]])
      return this
    }
  })
})

Transaction.prototype.setConnection = FSM.method('setConnection', {
  'disconnected': function (connection) {
    var self = this
    var err = self.state('connected')
    if (err) {
      process.nextTick(function () {
        self.emit('error', err)
      })
      return
    }

    self._connection = connection
    self._addConnectionListeners()

    self.emit('begin:start')
    var beginQuery = connection.query(self._statements.begin, function (err) {
      if (err) return self.handleError(err)
      self.emit('begin:complete') // removes error listener
      self._runQueue()
    })

    self.emit('query', beginQuery)
    return self
  }
})

Transaction.prototype._runQueue = function () {
  var self = this
  return next()

  function next (err, skipEmit) {
    if (err) {
      self.handleError(err, skipEmit)
    }
    if (!self._queue.length) {
      if (self.state() !== 'closed' && (err = self.state('open'))) {
        self.handleError(err)
      }
      return
    }

    var task = self._queue.shift()

    if (Array.isArray(task)) {
      runFunctionCall(self, task, next)
    } else if (task instanceof Transaction) {
      runChildTransaction(self, task)
    } else {
      runQueuedQuery(self, task, next)
    }
  }
}

function runFunctionCall (ctx, fnAndArgs, next) {
  var fn = fnAndArgs[0]
    , args = fnAndArgs[1]
    , last = args[args.length - 1]

  if (typeof last == 'function') {
    args[args.length - 1] = function (err) {
      if (err) return last(err)
      last.apply(this, arguments)
      next()
    }
  } else {
    args.push(next)
  }

  return fn.apply(ctx, args)
}

function runQueuedQuery (self, query, next) {
  if (self.state() == 'closed') {
    self.query(query, function (err) {
      query.emit('error', err)
      next()
    })
    return
  }
  self._connection.query(query)
  var onext = once(next) // ensure we only call `next` once
  query.once('error', function (err) {
    onext(err, this.listeners('error').length)
  })
  query.once('close', function () {
    // let 'error' events have a chance to call `next` first
    process.nextTick(onext)
  })
}

function beginWithParent (parent, options, callback) {
  var child = createChildTransaction(parent, options, callback)
  switch (parent.state()) {
    case 'disconnected':
    case 'connected':
      parent._queue.push(child)
      break
    case 'open':
      runChildTransaction(parent, child)
      break
    case 'closed':
      var error = new Error("Cannot start child transaction on parent in state 'closed'")
      process.nextTick(function () {
        // callback is already attached to error event
        child.emit('error', error)
      })
  }
  return child
}

function createChildTransaction (parent, options, callback) {
  var nestingLevel = parent._nestingLevel + 1
  var savepointName = 'sp_' + nestingLevel

  var child = new Transaction({
    adapter:      parent.adapter,
    nestingLevel: nestingLevel,
    callback:     callback,
    begin:        'SAVEPOINT '         + savepointName,
    commit:       'RELEASE SAVEPOINT ' + savepointName,
    rollback:     'ROLLBACK TO '       + savepointName,
    autoRollback: options.autoRollback,
  })

  child
    .on('query', parent._emitQuery)
    .once('connected', function () {
      parent._removeConnectionListeners()
      parent.state('connected')
    })
    .once('close', function () {
      parent._addConnectionListeners()
      parent._runQueue()
    })

  return child
}

function runChildTransaction (parent, child) {
  // Child transaction
  child.setConnection(parent._connection)
  child.on('error', function (err) {
    if (child.listeners('error').length == 1) {
      // if a child transaction errors, and the parent is the only
      // listener, it should re-emit the error, but *not* roll back
      parent.emit('error', err)
    }
  })
}

Transaction.prototype._addConnectionListeners = function () {
  this._connection.on('error', this.handleError)
  this._connection.on('query', this._emitQuery)
}

Transaction.prototype._removeConnectionListeners = function () {
  this._connection.removeListener('error', this.handleError)
  this._connection.removeListener('query', this._emitQuery)
}

function closeVia (action) {
  return function (callback) {
    var self = this
    if (self.state() == 'closed') {
      return
    }
    var err = self.state('closed')
    if (err) {
      return self.handleError(err, callback)
    }
    self.emit(action + ':start')
    var q = self._connection.query(self._statements[action], function (err) {
      self._removeConnectionListeners()
      self._connection = null
      if (err) {
        self.handleError(new CloseFailedError(err, action), callback)
      } else {
        self.emit(action + ':complete')
        self.emit('close')
        if (callback) callback()
      }
    })
    return self
  }
}

inherits(CloseFailedError, Error)
function CloseFailedError(err, action) {
  Error.captureStackTrace(this, CloseFailedError)
  this.name = action + ' failed'
  this.message = err.message
  this.previous = null;
}