db.js

/* globals done */

var pg = require('pg');
var conString = 'postgres://pg_logvent:Bow3CuU7foVHbr@localhost/db_logvent';

module.exports = function(options) {
    'use strict';

    options = options || {};

    var self = this || {};

    var convertTableName = function(tableName) {
        //convert from : to _ syntax
        return tableName.replace(/:/g, '_');
    };

    var tableExists = function (tableName, cb) {
        tableName = convertTableName(tableName);
        var exists = true;

        self.query('select * from information_schema.tables where tableName=\'' + tableName + '\';', [], function(err, result) {
            //does table exist?
            if(result.rowCount === 0) {
                exists = false;
            }

            if(cb) {
                cb(exists);
            }
        });
    };

    self.query = function(text, values, cb) {
        pg.connect(conString, function(err, client, done) {
            client.query(text, values, function(err, result) {
                client.query('COMMIT', done);
                done();

                if (cb) {
                    cb(err, result);
                }
            });
        });
    };

    self.insertEvent = function (tableName, eventId, message) {
        var created = false;

        if(typeof(message.time) === 'number') {
            created = new Date(message.time * 1000);
        }

        tableName = convertTableName(tableName);

        if(!created) {
            self.query('INSERT INTO ' + tableName + ' (eventId, message) VALUES($1, $2)', [eventId, message], function(err, result) {
                if (err) {
                    console.log(err);
                }
            });
        } else {
            self.query('INSERT INTO ' + tableName + ' (eventId, message, created) VALUES($1, $2, $3)', [eventId, message, created], function(err, result) {
                if (err) {
                    console.log(err);
                }
            });
        }
    };

    self.updatePerson = function (tableName, userId, message) {
        tableName = convertTableName(tableName);

        self.query('UPDATE ' + tableName + ' SET message = $2 WHERE userId = $1;',
            [userId, message], function(err, result) {
                if (err) {
                    console.log(err);
                }
                if (result.rowCount === 0) {
                    self.query('INSERT INTO ' + tableName + ' (userId, message) VALUES($1, $2);', [userId, message], function(err, result) {
                        if (err) {
                            console.log(err);
                        }
                    });
                }
            });
    };

    self.createEventTable = function(tableName, cb) {
        tableName = convertTableName(tableName);

        tableExists(tableName, function (exists) {
            if(!exists) {
                self.query('\
                    CREATE TABLE ' + tableName + ' ( \
                        id serial PRIMARY KEY, \
                        eventId character varying(100) NOT NULL,\
                        message json NOT NULL DEFAULT \'{}\'::json,\
                        created timestamp with time zone DEFAULT now()\
                    )\
                    WITH (\
                        OIDS=FALSE\
                    );\
                    ALTER TABLE ' + tableName + ' \
                    OWNER TO pg_logvent;\
                    CREATE INDEX ' + tableName + '_eventId_idx ON ' + tableName + ' USING BTREE (eventId);',
                [], function(err, result) {
                    if (err) {
                        console.log(err);
                    }
                    if (cb) {
                        cb();
                    }
                });
            } else {
                if (cb) {
                    cb();
                }
            }
        });
    };

    self.createPersonTable = function(tableName, cb) {
        tableName = convertTableName(tableName);

        tableExists(tableName, function (exists) {
            if(!exists) {
                self.query('\
                    CREATE TABLE ' + tableName + ' ( \
                        id serial PRIMARY KEY, \
                        userId character varying(100) NOT NULL,\
                        message json NOT NULL DEFAULT \'{}\'::json,\
                        created timestamp with time zone DEFAULT now()\
                    )\
                    WITH (\
                        OIDS=FALSE\
                    );\
                    ALTER TABLE ' + tableName + ' \
                    OWNER TO pg_logvent;\
                    CREATE INDEX ' + tableName + '_userId_idx ON ' + tableName + ' USING BTREE (userId);',
                [], function(err, result) {
                    if (err) {
                        console.log(err);
                    }
                    if (cb) {
                        cb();
                    }
                });
            } else {
                if (cb) {
                    cb();
                }
            }
        });
    };

    return self;
};