migrate-ds-postgres-connector.js

'use strict'

const { Connector } = require('@deepstream/storage-postgres')
const async = require('async')

// the connection data for the original database
const postgresOptionsFrom = {
  user: '',
  database: '',
  password: '',
  host: '',
  port: 5432, // postgres default post
  schema: 'ds', // schema defaults to ds. Will be created if it doesn't exist
  max: 10, // #concurrent connections
  idleTimeoutMillis: 30000, // timeout after which connection will be cut
  writeInterval: 200, // amout of milliseconds during which writes will be
  useJsonb: true, // store values as searchable binary JSON (slower)
  notifications: {
    CREATE_TABLE: false, // Get notified when tables are created
    DESTROY_TABLE: false, // Get notified when tables are dropped
    INSERT: false, // Get notified when records are created
    UPDATE: false, // Get notified when records are updated
    DELETE: false // Get notified when records are deleted
  }
}

// the connection data for the destination database with the new table columns
const postgresOptionsTo = {
  user: '',
  database: '',
  password: '',
  host: '',
  port: 5432, // postgres default post
  schema: 'ds', // schema defaults to ds. Will be created if it doesn't exist
  max: 10, // #concurrent connections
  idleTimeoutMillis: 30000, // timeout after which connection will be cut
  writeInterval: 200, // amout of milliseconds during which writes will be
  useJsonb: true, // store values as searchable binary JSON (slower)
  notifications: {
    CREATE_TABLE: false, // Get notified when tables are created
    DESTROY_TABLE: false, // Get notified when tables are dropped
    INSERT: false, // Get notified when records are created
    UPDATE: false, // Get notified when records are updated
    DELETE: false // Get notified when records are deleted
  }
}

const connectorFrom = new Connector(postgresOptionsFrom)
const connectorTo = new Connector(postgresOptionsTo)
// start connector
connectorFrom.init()
connectorTo.init()

/*
  SQL queries
 */
// set the table name to migrate
const table = 'user_audit_log'
connectorFrom.query(`select * from ${postgresOptionsFrom.schema}.${table};`, (err, result) => {
  if (err) {
    console.error(err)
    return
  }
  if (result.rows.length === 0) {
    console.log('NO DATA')
    return
  }
  async.each(result.rows, (value, cb) => {
    if (!value.val._v) return cb()
    connectorTo.set(`${table}/${value.id}`, value.val._v, value.val._d, (err) => {
      if (err) return cb(err)
      return cb()
    })
  }, (err) => {
    if (err) console.error(err)
    else console.log('DONE')
  })
})