Inserts TMP data into a POSTGIS database

tmps-to-postgis.js

#!/usr/bin/env node

'use strict';

var pg            = require('pg')
  , geojsonStream = require('geojson-stream')
  , fs            = require('fs')
  , stdout        = require('stdout-stream')
  , through2      = require('through2')
  , Q             = require('q')
  , chalk         = require('chalk')

var queries = {
  update: 'UPDATE tmps SET type=$2, geom=ST_GeomFromGeoJSON($3) WHERE id=$1;',
  insert: 'INSERT INTO tmps (id, type, geom) ' +
        'SELECT $1, $2, ST_GeomFromGeoJSON($3)' +
        'WHERE NOT EXISTS (SELECT 1 FROM tmps WHERE id=$1);'
}

var client = new pg.Client({
  user: 'digitalsadhu',
  database: 'tmps'
})

function error(err) {
  console.log(chalk.red(err.message))
  console.trace()
  process.exit(1)
}

function performQuery(client, query, values) {
  var deferred = Q.defer()
  client.query({ text: query, values: values },
    function (err, result) {
      if (err) return deferred.reject(err)
      deferred.resolve(result)
    })
  return deferred.promise
}

function insertGeojson(client) {
  return through2.obj(function (geojson, enc, cb) {
    var values = [geojson.properties.id, process.argv[3], geojson.geometry]

    performQuery(client, queries.update, values)
      .then(performQuery.bind(this, client, queries.insert, values))
      .then(this.push.bind(this, geojson))
      .then(cb.bind(this, null))
      .fail(function (err) { error(err) })
  })
}

client.connect(function(err) {
  if(err) return console.error('could not connect to postgres', err)

  fs.createReadStream(process.argv[2])
    .pipe(geojsonStream.parse())
    .pipe(insertGeojson(client))
    .pipe(geojsonStream.stringify())
    .on('end', function () { process.exit(0) })
    .pipe(stdout)
})