index.js

'use strict'
var EventEmitter = require('events').EventEmitter
var sqlite3 = require('sqlite3')
var inherits = require('inherits')
var Readable = require('stream').Readable

var adapter = exports

adapter.name = 'sqlite3'

adapter.verbose = sqlite3.verbose

adapter.createQuery = function(text, values, callback) {
  if (text instanceof SQLite3Query) return text
  return new SQLite3Query(text, values, callback)
}

adapter.createConnection = function(opts, callback) {
  var filename

  if (opts.host) {
    filename = opts.host + (opts.database || '')
  } else {
    filename = opts.database
  }

  if (!filename || filename == '/:memory') filename = ':memory:'

  var mode = Object.keys(opts)
    .map(function(k) {
      return (k == k.toUpperCase() && sqlite3[k]) | 0
    })
    .reduce(function(mode, flag) {
      return mode | flag
    }, 0)

  if (!mode) mode = sqlite3.OPEN_READWRITE | sqlite3.OPEN_CREATE

  var db = new sqlite3.Database(filename, mode)
  db.serialize()
  return new SQLite3Connection(db, callback)
}

inherits(SQLite3Connection, EventEmitter)
function SQLite3Connection(db, callback) {
  EventEmitter.call(this)
  this._db = db
  var self = this
  this._db.on('open', function() {
    self.emit('open')
  })
  this._db.on('close', function() {
    self.emit('end')
  })
  this._db.on('error', function(err) {
    self.emit('error', err)
  })

  if (callback) {
    self.once('error', callback)
    self.once('open', function handleOpenEvent() {
      self.removeListener('error', callback)
      callback(null, self)
    })
  }
}

SQLite3Connection.prototype.adapter = adapter

SQLite3Connection.prototype.query = function(text, values, callback) {
  var query = adapter.createQuery(text, values, callback)

  this.emit('query', query)

  if (query.text.match(/^\s*(insert|update|replace)\s+/i)) {
    runQuery(this._db, query)
  } else {
    eachQuery(this._db, query)
  }

  return query
}

function runQuery(db, query) {
  db.run(query.text, query.values, function(err) {
    query.complete(err, this ? this.changes : 0, this ? this.lastID : -1)
  })
}

function eachQuery(db, query) {
  db.each(query.text, query.values, onRow, function(err, count) {
    query.complete(err, count)
  })

  function onRow(err, row) {
    query.onRow(err, row)
  }
}

SQLite3Connection.prototype.end = function(callback) {
  if (callback) this.on('end', callback)
  this._db.close()
}

inherits(SQLite3Query, Readable)
function SQLite3Query(text, values, callback) {
  Readable.call(this, { objectMode: true })
  this.text = text
  this._fields = null
  this._result = { rows: [] }
  this._errored = false
  if (typeof values == 'function') {
    callback = values
    values = []
  }
  this.values = values || []
  if ((this.callback = callback)) {
    this.on('error', function(error) {
      this.callback(error)
    })
    this.on('data', function(row) {
      this._result.rows.push(row)
    })
  }
}

SQLite3Query.prototype._read = function() {}

SQLite3Query.prototype.onRow = function(err, row) {
  if (this._errored) return
  this._gotData = true
  if (err) {
    this._errored = true
    this.emit('close')
    this.emit('error', err)
    return
  }
  if (!this._result.fields) {
    this._result.fields = Object.keys(row).map(function(name) {
      return { name: name }
    })
    this.emit('fields', this._result.fields)
  }
  this.push(row)
}

SQLite3Query.prototype.complete = function(err, count, lastId, affectedRows) {
  this.push(null)
  if (this._errored) return // we've emitted an error from the row callback
  if (!err && !this._gotData) this.emit('fields', [])
  this.emit('close')
  if (err) return this.emit('error', err)
  this._result.rowCount = count
  this._result.lastInsertId = lastId
  if (affectedRows) {
    this._result.affectedRows = affectedRows
    this._result.changedRows = affectedRows
  }

  if (this.callback) {
    this.callback(null, this._result)
  }
}