Sqlite3 db migration with co

lagancy-co.js

'use strict'
const co = require('co')
const fs = require('mz/fs')
const sqlite3 = require('../node-sqlite3') // custom version sqlite3

co(function* () {
  try {
    yield fs.unlink('old.db')
    yield fs.unlink('new.db')
  } catch (e) {}

  const oldDb = new sqlite3.Database('old.db')
  yield setup(oldDb)
  yield mock(oldDb)

  const newDb = new sqlite3.Database('new.db')
  yield setup(newDb)
  yield migrate(oldDb, newDb)

}).then(() => {
  console.log('done')
}).catch(err => {
  console.error(err)
})

function* migrate (from, to) {
  const tables = yield t2p(cb => from.all('SELECT name FROM sqlite_master WHERE type="table"', cb))

  for (let i = 0; i < tables.length; i++) {
    const name = tables[i].name
    const rows = yield t2p(cb => from.all(`SELECT * FROM ${name}`, cb)) // todo 分页分批

    if (rows.length) {
      const row0 = rows[0]
      const keys = rowKeys(row0)
      const keyHolder = keys.join(',')
      const valueHolder = keys.map(k => `?`).join(',')

      const stmt = to.prepare(`INSERT INTO ${name} (${keyHolder}) VALUES (${valueHolder})`)
      for (let j = 0; j < rows.length; j++) {
        const values = rowValues(keys, rows[j])
        stmt.run(values)
      }
      yield t2p(cb => stmt.finalize(cb))
    }
  }
}

function rowKeys (row) {
  return Object.keys(row).sort()
}
function rowValues (keys, row) {
  return keys.reduce((m, k) => {
    return m.push(row[k]), m
  }, [])
}

function* setup (db) {
  yield t2p(cb => db.run(`
    CREATE TABLE IF NOT EXISTS lorem (info TEXT, qq INT, happy BOOLEAN);
  `, cb))
}
function* mock (db) {
  const stmt = db.prepare("INSERT INTO lorem VALUES (?, ?, ?)")
  for (let i = 0; i < 10; i++) {
    stmt.run(["Ipsum " + i, i * 100, true])
  }
  yield t2p(cb => stmt.finalize(cb))
}

function t2p (thunk) {
  return new Promise((rs, rj) => {
    // thunk((err) => {
    thunk(function (err) {
      const rest = [].slice.call(arguments, 1)
      if (err) rj(err)
      else rs(rest.length > 1 ? rest : rest[0])
    })
  })
}

migrate-await.js

import { t2p } from '../../src/lib/util'
import sqlite3 from 'sqlite3'
import { fs } from 'mz'

// https://github.com/mapbox/node-sqlite3/wiki/Debugging
// const isDev = process.env.NODE_ENV === 'development'
// const { Database } = isDev ? sqlite3.verbose() : sqlite3
const { Database } = sqlite3

// newDatabaseWithSetup({
//   args: ['data.db', 'dbPass'],
//   setup: async db => {
//     await createUpdateDB(db)
//   }
// })
export async function newDatabaseWithSetup ({ args, setup }) {
  const db = new Database(...args)
  await setup(db)
  return db
}

// newDatabaseWithMigration(
//   argsArr: [
//     ['new.db', 'newPass'],
//     ['old.db', 'oldPass'],
//     ['evenOlder.db', 'evenOlderPass']
//   ],
//   setup: async db => {}
// )
export async function newDatabaseWithMigration ({ argsArr, setup, filter }) {
  const [argsNew, ...argsArrOld] = argsArr
  argsArrOld = argsArrOld.reverse()

  const db = await newDatabaseWithSetup({
    args: argsNew,
    setup: async db => {
      await setup(db, async () => {
        for (const args of argsArrOld) {
          if (await fs.exists(args[0])) {
            const d = new Database(...args)
            await migrate(d, db, filter)
          }
        }
      })
    }
  })
  return db
}

// ON CONFLICT clause
// https://www.sqlite.org/lang_conflict.html
// todo: 单条插入容错？是否需要事务？sqlite3-transaction？
// http://stackoverflow.com/questions/12105198/sqlite-how-to-get-insert-or-ignore-to-work
// unrecognized token: 59481b1557... 表名需要加引号
// Copy table structure to new table in sqlite3 http://stackoverflow.com/questions/12730390/copy-table-structure-to-new-table-in-sqlite3
async function migrate (from, to, filter) {
  /* eslint-disable no-console */
  console.log(`migrating db ${from.filename} => ${to.filename}`)
  const tables = await t2p(cb => from.all('SELECT name, sql FROM sqlite_master WHERE type="table"', cb))

  for (let { name, sql } of tables) {
    sql = sql.replace(/CREATE TABLE/i, 'CREATE TABLE IF NOT EXISTS')
    await t2p(cb => to.exec(sql, cb))

    const condition = filter && filter(name) || ''
    if (condition) console.log(`using condition ${condition}`)
    const rows = await t2p(cb => from.all(`SELECT * FROM "${name}" ${condition}`, cb)) // todo 分页分批

    if (rows.length) {
      const row0 = rows[0]
      const keys = rowKeys(row0)
      const keyHolder = keys.join(',')
      const valueHolder = keys.map(() => `?`).join(',')

      // await t2p(cb => to.exec('BEGIN', cb))
      const stmt = to.prepare(`INSERT OR REPLACE INTO "${name}" (${keyHolder}) VALUES (${valueHolder})`)
      for (const row of rows) {
        const values = rowValues(keys, row)
        stmt.run(values)
      }
      // stmt.finalize()
      // await t2p(cb => to.exec('COMMIT', cb))
      await t2p(cb => stmt.finalize(cb))
    }
    console.log(`migrated table ${name} rows * ${rows.length}`)
  }

  await t2p(cb => from.close(cb))
  const { filename: f } = from
  const nf = await nextFilename(f)
  await fs.rename(f, nf)
  console.log(`migrated db and renamed to ${nf}`)
}

async function nextFilename (file) {
  let i = 1, f = file
  /* eslint-disable no-constant-condition */
  while (true) {
    if (await fs.exists(f)) { // 应该采用二分法
      f = `${file}.${i++}`
    } else {
      break
    }
  }
  return f
}

function rowKeys (row) {
  return Object.keys(row).sort()
}
function rowValues (keys, row) {
  return keys.reduce((m, k) => {
    return m.push(row[k]), m
  }, [])
}