Initial version

Author: stuartpb

app/controllers/messages/create.js

@@ -0,0 +1,38 @@
+var express = require('express');
+var serveStatic = require('serve-static');
+var bodyParser = require('body-parser');
+var session = require('express-session');
+var RedisStore = require('connect-redis')(session);
+
+function appCtor(cfg, pool) {
+  var app = express();
+
+  app.set('trust proxy', true);
+  app.set('view engine', 'jade');
+  app.set('views', __dirname + '/views');
+  app.use(serveStatic(__dirname + '/static'));
+  app.use(session({
+    store: new RedisStore({
+      host: cfg.redis.hostname,
+      port: cfg.redis.port}),
+    secret: cfg.env.SECRET_KEY_BASE,
+    resave: false,
+    saveUninitialized: false
+  }));
+  app.use(function (req, res, next) {
+    if (!req.session) {
+      return next(new Error("couldn't find sessions"));
+    }
+    else return next();
+  });
+  app.use(bodyParser.urlencoded({ extended: false }));
+
+  app.get('/', function(req, res) {
+    return res.render('index');
+  });
+  app.get('/m', function(req, res) {
+    return res.render('messaging');
+  });
+
+  return app;
+}

app/controllers/messages/delete.js

@@ -0,0 +1,38 @@
+var r = require('rethinkdb');
+var endex = require('endex');
+var Promise = require('bluebird');
+
+module.exports = function poolCtor(cfg) {
+  var pool = {};
+  var serverReportError = console.error.bind(console);
+
+  var conn;
+
+  function runQueryNormally(query) {
+    return query.run(conn);
+  }
+
+  var connPromise = r.connect(cfg.rethinkdb).then(function(connection) {
+    conn = connection;
+    pool.runQuery = runQueryNormally;
+    return endex(r).db(cfg.rethinkdb && cfg.rethinkdb.db || 'dingroll')
+      .table('users')
+        .index('username')
+      .table('messages')
+        .index('sender')
+        .index('target')
+        .index('creation')
+        .index('mentions', {multi: true})
+        .index('hashtags', {multi: true})
+      .run(connection);
+  }).catch(serverReportError);
+
+  pool.runQuery = function queueQueryRun(query) {
+    return new Promise(function(resolve, reject) {
+      connPromise.then(function(conn){
+        query.run(conn).then(resolve, reject);
+        return conn;
+      });
+    });
+  };
+};

app/controllers/messages/edit.js

@@ -0,0 +1,53 @@
+var r = require('rethinkdb');
+
+function socketAppCtor(cfg, pool) { return function socketApp(socket) {
+
+  // just listen for changes in sidebar filter status -
+  // report every new message in active filters
+  var filters;
+  function reportError(err){
+    socket.write({
+      type: 'error',
+      err: err
+    });
+  }
+
+  function haveMessage(message){
+    socket.write({
+      type: 'haveMessage',
+      message: message
+    });
+  }
+
+  var adHocFilterCursor;
+
+  function setAdHocFilter(filter) {
+    if (adHocFilterCursor) adHocFilterCursor.close();
+    adHocFilterCursor = pool.runQuery(r.table('messages')
+      .filter(r.row('body').match(filter)));
+    adHocFilterCursor.each(function(err, message){
+      if (err) return reportError(err);
+      haveMessage(message);
+    }, function onFinished(){
+      adHocFilterCursor = pool.runQuery(r.table('messages')
+        .filter(r.row('body').match(filter)).changes());
+      adHocFilterCursor.each(function(err, changes){
+        if (err) return reportError(err);
+        haveMessage(changes.new_val);
+      });
+    });
+  }
+
+  socket.on('data', function(data){
+    if (data.type == 'error') return console.error(data);
+    else if (data.type == 'createMessage') {
+      pool.runQuery(r.table('messages').insert(data.message));
+    } else if (data.type == 'setAdHocFilter') {
+      setAdHocFilter(data.filter);
+    } else {
+      console.error('Unrecognized message type', data);
+    }
+  });
+}}
+
+module.exports = socketAppCtor;