Merge branch 'consolidated-api' into trim-history

lib/api.js

@@ -0,0 +1,74 @@
+/* jshint esversion: 6 */
+const nThen = require("nthen");
+const WebSocketServer = require('ws').Server;
+const NetfluxSrv = require('chainpad-server');
+
+module.exports.create = function (config) {
+    const wsConfig = {
+        server: config.httpServer,
+    };
+
+    nThen(function (w) {
+        require('../storage/file').create(config, w(function (_store) {
+            config.store = _store;
+        }));
+    }).nThen(function (w) {
+        require("../storage/tasks").create(config, w(function (e, tasks) {
+            if (e) {
+                throw e;
+            }
+            config.tasks = tasks;
+            if (config.disableIntegratedTasks) { return; }
+
+            // XXX support stopping this interval
+            setInterval(function () {
+                tasks.runAll(function (err) {
+                    if (err) {
+                        // either TASK_CONCURRENCY or an error with tasks.list
+                        // in either case it is already logged.
+                    }
+                });
+            }, 1000 * 60 * 5); // run every five minutes
+        }));
+    }).nThen(function () {
+        // asynchronously create a historyKeeper and RPC together
+        require('./historyKeeper.js').create(config, function (err, historyKeeper) {
+            if (err) { throw err; }
+
+            var log = config.log;
+
+            // spawn ws server and attach netflux event handlers
+            NetfluxSrv.create(new WebSocketServer(wsConfig))
+                .on('channelClose', historyKeeper.channelClose)
+                .on('channelMessage', historyKeeper.channelMessage)
+                .on('channelOpen', historyKeeper.channelOpen)
+                .on('sessionClose', function (userId, reason) {
+                    if (['BAD_MESSAGE', 'SOCKET_ERROR', 'SEND_MESSAGE_FAIL_2'].indexOf(reason) !== -1) {
+                        return void log.error('SESSION_CLOSE_WITH_ERROR', {
+                            userId: userId,
+                            reason: reason,
+                        });
+                    }
+                    log.verbose('SESSION_CLOSE_ROUTINE', {
+                        userId: userId,
+                        reason: reason,
+                    });
+                })
+                .on('error', function (error, label, info) {
+                    if (!error) { return; }
+                    /* labels:
+                        SEND_MESSAGE_FAIL, SEND_MESSAGE_FAIL_2, FAIL_TO_DISCONNECT,
+                        FAIL_TO_TERMINATE, HANDLE_CHANNEL_LEAVE, NETFLUX_BAD_MESSAGE,
+                        NETFLUX_WEBSOCKET_ERROR
+                    */
+                    log.error(label, {
+                        code: error.code,
+                        message: error.message,
+                        stack: error.stack,
+                        info: info,
+                    });
+                })
+                .register(historyKeeper.id, historyKeeper.directMessage);
+        });
+    });
+};

lib/commands/admin-rpc.js

@@ -6,25 +6,15 @@ var Fs = require("fs");
 
 var Admin = module.exports;
 
-var getActiveSessions = function (Env, ctx, cb) {
-    var total = ctx.users ? Object.keys(ctx.users).length : '?';
-
-    var ips = [];
-    Object.keys(ctx.users).forEach(function (u) {
-        var user = ctx.users[u];
-        var socket = user.socket;
-        var req = socket.upgradeReq;
-        var conn = req && req.connection;
-        var ip = (req && req.headers && req.headers['x-forwarded-for']) || (conn && conn.remoteAddress);
-        if (ip && ips.indexOf(ip) === -1) {
-            ips.push(ip);
-        }
-    });
-
-    cb (void 0, [total, ips.length]);
+var getActiveSessions = function (Env, Server, cb) {
+    var stats = Server.getSessionStats();
+    cb(void 0, [
+        stats.total,
+        stats.unique
+    ]);
 };
 
-var shutdown = function (Env, ctx, cb) {
+var shutdown = function (Env, Server, cb) {
     return void cb('E_NOT_IMPLEMENTED');
     //clearInterval(Env.sessionExpirationInterval);
     // XXX set a flag to prevent incoming database writes
@@ -91,19 +81,18 @@ var getDiskUsage = function (Env, cb) {
     });
 };
 
-
-
-Admin.command = function (Env, ctx, publicKey, data, cb) {
+Admin.command = function (Env, Server, publicKey, data, cb) {
     var admins = Env.admins;
     if (admins.indexOf(publicKey) === -1) {
         return void cb("FORBIDDEN");
     }
+
     // Handle commands here
     switch (data[0]) {
         case 'ACTIVE_SESSIONS':
-            return getActiveSessions(Env, ctx, cb);
+            return getActiveSessions(Env, Server, cb);
         case 'ACTIVE_PADS':
-            return cb(void 0, ctx.channels ? Object.keys(ctx.channels).length : '?');
+            return cb(void 0, Server.getActiveChannelCount());
         case 'REGISTERED_USERS':
             return getRegisteredUsers(Env, cb);
         case 'DISK_USAGE':
@@ -112,7 +101,7 @@ Admin.command = function (Env, ctx, publicKey, data, cb) {
             Env.flushCache();
             return cb(void 0, true);
         case 'SHUTDOWN':
-            return shutdown(Env, ctx, cb);
+            return shutdown(Env, Server, cb);
         default:
             return cb('UNHANDLED_ADMIN_COMMAND');
     }

lib/commands/channel.js

@@ -147,7 +147,7 @@ Channel.isNewChannel = function (Env, channel, cb) {
 
     Otherwise behaves the same as sending to a channel
 */
-Channel.writePrivateMessage = function (Env, args, nfwssCtx, cb) {
+Channel.writePrivateMessage = function (Env, args, Server, cb) {
     var channelId = args[0];
     var msg = args[1];
 
@@ -161,7 +161,7 @@ Channel.writePrivateMessage = function (Env, args, nfwssCtx, cb) {
 
     // We expect a modern netflux-websocket-server instance
     // if this API isn't here everything will fall apart anyway
-    if (!(nfwssCtx && nfwssCtx.historyKeeper && typeof(nfwssCtx.historyKeeper.onChannelMessage) === 'function')) {
+    if (!(Server && typeof(Server.send) === 'function')) {
         return void cb("NOT_IMPLEMENTED");
     }
 
@@ -180,8 +180,9 @@ Channel.writePrivateMessage = function (Env, args, nfwssCtx, cb) {
         msg // the actual message content. Generally a string
     ];
 
+    // XXX this API doesn't exist anymore...
     // store the message and do everything else that is typically done when going through historyKeeper
-    nfwssCtx.historyKeeper.onChannelMessage(nfwssCtx, channelStruct, fullMessage);
+    Env.historyKeeper.onChannelMessage(Server, channelStruct, fullMessage);
 
     // call back with the message and the target channel.
     // historyKeeper will take care of broadcasting it if anyone is in the channel

lib/historyKeeper.js

@@ -1,6 +1,5 @@
 /* jshint esversion: 6 */
 /* global Buffer */
-;(function () { 'use strict';
 
 const nThen = require('nthen');
 const Nacl = require('tweetnacl/nacl-fast');
@@ -10,7 +9,11 @@ const Meta = require("./metadata");
 const WriteQueue = require("./write-queue");
 const BatchRead = require("./batch-read");
 
+const RPC = require("./rpc");
+
 const Extras = require("./hk-util.js");
+const STANDARD_CHANNEL_LENGTH = Extras.STANDARD_CHANNEL_LENGTH;
+const EPHEMERAL_CHANNEL_LENGTH = Extras.EPHEMERAL_CHANNEL_LENGTH;
 
 let Log;
 const now = function () { return (new Date()).getTime(); };
@@ -63,8 +66,10 @@ const isValidValidateKeyString = function (key) {
     }
 };
 
-module.exports.create = function (cfg) {
-    const rpc = cfg.rpc;
+var CHECKPOINT_PATTERN = /^cp\|(([A-Za-z0-9+\/=]+)\|)?/;
+
+module.exports.create = function (cfg, cb) {
+    var rpc;
     const tasks = cfg.tasks;
     const store = cfg.store;
     Log = cfg.log;
@@ -72,18 +77,11 @@ module.exports.create = function (cfg) {
     Log.silly('HK_LOADING', 'LOADING HISTORY_KEEPER MODULE');
 
     const metadata_cache = {};
+    const channel_cache = {};
     const HISTORY_KEEPER_ID = Crypto.randomBytes(8).toString('hex');
 
     Log.verbose('HK_ID', 'History keeper ID: ' + HISTORY_KEEPER_ID);
 
-    let sendMsg = function () {};
-    let STANDARD_CHANNEL_LENGTH, EPHEMERAL_CHANNEL_LENGTH;
-    const setConfig = function (config) {
-        STANDARD_CHANNEL_LENGTH = config.STANDARD_CHANNEL_LENGTH;
-        EPHEMERAL_CHANNEL_LENGTH = config.EPHEMERAL_CHANNEL_LENGTH;
-        sendMsg = config.sendMsg;
-    };
-
     /*  computeIndex
         can call back with an error or a computed index which includes:
             * cpIndex:
@@ -216,8 +214,9 @@ module.exports.create = function (cfg) {
         if the channel exists but its index does not then it caches the index
     */
     const batchIndexReads = BatchRead("HK_GET_INDEX");
-    const getIndex = (ctx, channelName, cb) => {
-        const chan = ctx.channels[channelName];
+    const getIndex = (channelName, cb) => {
+        const chan = channel_cache[channelName];
+
         // if there is a channel in memory and it has an index cached, return it
         if (chan && chan.index) {
             // enforce async behaviour
@@ -238,15 +237,7 @@ module.exports.create = function (cfg) {
         });
     };
 
-    /*::
-    type cp_index_item = {
-        offset: number,
-        line: number
-    }
-    */
-
     /*  storeMessage
-        * ctx
         * channel id
         * the message to store
         * whether the message is a checkpoint
@@ -265,7 +256,7 @@ module.exports.create = function (cfg) {
     */
     const queueStorage = WriteQueue();
 
-    const storeMessage = function (ctx, channel, msg, isCp, optionalMessageHash) {
+    const storeMessage = function (channel, msg, isCp, optionalMessageHash) {
         const id = channel.id;
 
         queueStorage(id, function (next) {
@@ -289,7 +280,7 @@ module.exports.create = function (cfg) {
                     }
                 }));
             }).nThen((waitFor) => {
-                getIndex(ctx, id, waitFor((err, index) => {
+                getIndex(id, waitFor((err, index) => {
                     if (err) {
                         Log.warn("HK_STORE_MESSAGE_INDEX", err.stack);
                         // non-critical, we'll be able to get the channel index later
@@ -303,10 +294,10 @@ module.exports.create = function (cfg) {
                                 delete index.offsetByHash[k];
                             }
                         }
-                        index.cpIndex.push(({
+                        index.cpIndex.push({
                             offset: index.size,
                             line: ((index.line || 0) + 1)
-                        } /*:cp_index_item*/));
+                        });
                     }
                     if (optionalMessageHash) { index.offsetByHash[optionalMessageHash] = index.size; }
                     index.size += msgBin.length;
@@ -318,21 +309,10 @@ module.exports.create = function (cfg) {
         });
     };
 
-    /*  historyKeeperBroadcast
-        * uses API from the netflux server to send messages to every member of a channel
-        * sendMsg runs in a try-catch and drops users if sending a message fails
-    */
-    const historyKeeperBroadcast = function (ctx, channel, msg) {
-        let chan = ctx.channels[channel] || (([] /*:any*/) /*:Chan_t*/);
-        chan.forEach(function (user) {
-            sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify(msg)]);
-        });
-    };
-
     /*  expireChannel is here to clean up channels that should have been removed
         but for some reason are still present
     */
-    const expireChannel = function (ctx, channel) {
+    const expireChannel = function (channel) {
         return void store.archiveChannel(channel, function (err) {
             Log.info("ARCHIVAL_CHANNEL_BY_HISTORY_KEEPER_EXPIRATION", {
                 channelId: channel,
@@ -341,6 +321,14 @@ module.exports.create = function (cfg) {
         });
     };
 
+    /*  dropChannel
+        * cleans up memory structures which are managed entirely by the historyKeeper
+    */
+    const dropChannel = function (chanName) {
+        delete metadata_cache[chanName];
+        delete channel_cache[chanName];
+    };
+
     /*  checkExpired
         * synchronously returns true or undefined to indicate whether the channel is expired
           * according to its metadata
@@ -352,7 +340,7 @@ module.exports.create = function (cfg) {
 
         FIXME the boolean nature of this API should be separated from its side effects
     */
-    const checkExpired = function (ctx, channel) {
+    const checkExpired = function (Server, channel) {
         if (!(channel && channel.length === STANDARD_CHANNEL_LENGTH)) { return false; }
         let metadata = metadata_cache[channel];
         if (!(metadata && typeof(metadata.expire) === 'number')) { return false; }
@@ -367,26 +355,22 @@ module.exports.create = function (cfg) {
         // there may have been a problem with scheduling tasks
         // or the scheduled tasks may not be running
         // so trigger a removal from here
-        if (pastDue >= ONE_DAY) { expireChannel(ctx, channel); }
+        if (pastDue >= ONE_DAY) { expireChannel(channel); }
 
         // close the channel
         store.closeChannel(channel, function () {
-            historyKeeperBroadcast(ctx, channel, {
+            // XXX make sure that clients actually disconnect when we broadcast an error
+            Server.channelBroadcast(channel, {
                 error: 'EEXPIRED',
                 channel: channel
-            });
-            // remove it from any caches after you've told anyone in the channel
-            // that it has expired
-            delete ctx.channels[channel];
-            delete metadata_cache[channel];
+            }, HISTORY_KEEPER_ID);
+            dropChannel(channel);
         });
 
         // return true to indicate that it has expired
         return true;
     };
 
-    var CHECKPOINT_PATTERN = /^cp\|(([A-Za-z0-9+\/=]+)\|)?/;
-
     /*  onChannelMessage
         Determine what we should store when a message a broadcasted to a channel"
 
@@ -398,7 +382,7 @@ module.exports.create = function (cfg) {
         * adds timestamps to incoming messages
         * writes messages to the store
     */
-    const onChannelMessage = function (ctx, channel, msgStruct) {
+    const onChannelMessage = function (Server, channel, msgStruct) {
         // TODO our usage of 'channel' here looks prone to errors
         // we only use it for its 'id', but it can contain other stuff
         // also, we're using this RPC from both the RPC and Netflux-server
@@ -421,7 +405,7 @@ module.exports.create = function (cfg) {
         let metadata;
         nThen(function (w) {
             // getIndex (and therefore the latest metadata)
-            getIndex(ctx, channel.id, w(function (err, index) {
+            getIndex(channel.id, w(function (err, index) {
                 if (err) {
                     w.abort();
                     return void Log.error('CHANNEL_MESSAGE_ERROR', err);
@@ -436,7 +420,7 @@ module.exports.create = function (cfg) {
                 metadata = index.metadata;
 
                 // don't write messages to expired channels
-                if (checkExpired(ctx, channel)) { return void w.abort(); }
+                if (checkExpired(Server, channel)) { return void w.abort(); }
 
                 // if there's no validateKey present skip to the next block
                 if (!metadata.validateKey) { return; }
@@ -486,20 +470,10 @@ module.exports.create = function (cfg) {
             msgStruct.push(now());
 
             // storeMessage
-            storeMessage(ctx, channel, JSON.stringify(msgStruct), isCp, getHash(msgStruct[4], Log));
+            storeMessage(channel, JSON.stringify(msgStruct), isCp, getHash(msgStruct[4], Log));
         });
     };
 
-    /*  dropChannel
-        * exported as API
-          * used by chainpad-server/NetfluxWebsocketSrv.js
-        * cleans up memory structures which are managed entirely by the historyKeeper
-          * the netflux server manages other memory in ctx.channels
-    */
-    const dropChannel = function (chanName) {
-        delete metadata_cache[chanName];
-    };
-
     /*  getHistoryOffset
         returns a number representing the byte offset from the start of the log
         for whatever history you're seeking.
@@ -529,12 +503,12 @@ module.exports.create = function (cfg) {
           * -1 if you didn't find it
 
     */
-    const getHistoryOffset = (ctx, channelName, lastKnownHash, cb /*:(e:?Error, os:?number)=>void*/) => {
+    const getHistoryOffset = (channelName, lastKnownHash, cb) => {
         // lastKnownhash === -1 means we want the complete history
         if (lastKnownHash === -1) { return void cb(null, 0); }
         let offset = -1;
         nThen((waitFor) => {
-            getIndex(ctx, channelName, waitFor((err, index) => {
+            getIndex(channelName, waitFor((err, index) => {
                 if (err) { waitFor.abort(); return void cb(err); }
 
                 // check if the "hash" the client is requesting exists in the index
@@ -607,10 +581,10 @@ module.exports.create = function (cfg) {
         * GET_HISTORY
 
     */
-    const getHistoryAsync = (ctx, channelName, lastKnownHash, beforeHash, handler, cb) => {
+    const getHistoryAsync = (channelName, lastKnownHash, beforeHash, handler, cb) => {
         let offset = -1;
         nThen((waitFor) => {
-            getHistoryOffset(ctx, channelName, lastKnownHash, waitFor((err, os) => {
+            getHistoryOffset(channelName, lastKnownHash, waitFor((err, os) => {
                 if (err) {
                     waitFor.abort();
                     return void cb(err);
@@ -673,42 +647,50 @@ module.exports.create = function (cfg) {
     /*  onChannelCleared
         * broadcasts to all clients in a channel if that channel is deleted
     */
-    const onChannelCleared = function (ctx, channel) {
-        historyKeeperBroadcast(ctx, channel, {
+    const onChannelCleared = function (Server, channel) {
+        Server.channelBroadcast(channel, {
             error: 'ECLEARED',
             channel: channel
-        });
+        }, HISTORY_KEEPER_ID);
     };
+
     // When a channel is removed from datastore, broadcast a message to all its connected users
-    const onChannelDeleted = function (ctx, channel) {
+    const onChannelDeleted = function (Server, channel) {
         store.closeChannel(channel, function () {
-            historyKeeperBroadcast(ctx, channel, {
+            Server.channelBroadcast(channel, {
                 error: 'EDELETED',
                 channel: channel
-            });
+            }, HISTORY_KEEPER_ID);
         });
-        delete ctx.channels[channel];
+
+        delete channel_cache[channel];
+        Server.clearChannel(channel);
         delete metadata_cache[channel];
     };
     // Check if the selected channel is expired
     // If it is, remove it from memory and broadcast a message to its members
 
-    const onChannelMetadataChanged = function (ctx, channel, metadata) {
-        if (channel && metadata_cache[channel] && typeof (metadata) === "object") {
-            Log.silly('SET_METADATA_CACHE', 'Channel '+ channel +', metadata: '+ JSON.stringify(metadata));
-            metadata_cache[channel] = metadata;
-            if (ctx.channels[channel] && ctx.channels[channel].index) {
-                ctx.channels[channel].index.metadata = metadata;
-            }
-            historyKeeperBroadcast(ctx, channel, metadata);
+    const onChannelMetadataChanged = function (Server, channel, metadata) {
+        if (!(channel && metadata_cache[channel] && typeof (metadata) === "object")) { return; }
+        Log.silly('SET_METADATA_CACHE', {
+            channel: channel,
+            metadata: JSON.stringify(metadata),
+        });
+
+        metadata_cache[channel] = metadata;
+
+        if (channel_cache[channel] && channel_cache[channel].index) {
+            channel_cache[channel].index.metadata = metadata;
         }
+        Server.channelBroadcast(channel, metadata, HISTORY_KEEPER_ID);
     };
 
-    const handleGetHistory = function (ctx, seq, user, parsed) {
+    const handleGetHistory = function (Server, seq, user, parsed) {
         // parsed[1] is the channel id
         // parsed[2] is a validation key or an object containing metadata (optionnal)
         // parsed[3] is the last known hash (optionnal)
-        sendMsg(ctx, user, [seq, 'ACK']);
+
+        Server.send(user.id, [seq, 'ACK']);
         var channelName = parsed[1];
         var config = parsed[2];
         var metadata = {};
@@ -743,7 +725,7 @@ module.exports.create = function (cfg) {
                 unfortunately, we can't just serve it blindly, since then young channels will
                 send the metadata twice, so let's do a quick check of what we're going to serve...
             */
-            getIndex(ctx, channelName, waitFor((err, index) => {
+            getIndex(channelName, waitFor((err, index) => {
                 /*  if there's an error here, it should be encountered
                     and handled by the next nThen block.
                     so, let's just fall through...
@@ -757,29 +739,29 @@ module.exports.create = function (cfg) {
                 if (!index || !index.metadata) { return void w(); }
                 // And then check if the channel is expired. If it is, send the error and abort
                 // FIXME this is hard to read because 'checkExpired' has side effects
-                if (checkExpired(ctx, channelName)) { return void waitFor.abort(); }
+                if (checkExpired(Server, channelName)) { return void waitFor.abort(); }
                 // always send metadata with GET_HISTORY requests
-                sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify(index.metadata)], w);
+                Server.send(user.id, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify(index.metadata)], w);
             }));
         }).nThen(() => {
             let msgCount = 0;
 
             // TODO compute lastKnownHash in a manner such that it will always skip past the metadata line?
-            getHistoryAsync(ctx, channelName, lastKnownHash, false, (msg, readMore) => {
+            getHistoryAsync(channelName, lastKnownHash, false, (msg, readMore) => {
                 if (!msg) { return; }
                 msgCount++;
                 // avoid sending the metadata message a second time
                 if (isMetadataMessage(msg) && metadata_cache[channelName]) { return readMore(); }
-                sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify(msg)], readMore);
+                Server.send(user.id, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify(msg)], readMore);
             }, (err) => {
                 if (err && err.code !== 'ENOENT') {
                     if (err.message !== 'EINVAL') { Log.error("HK_GET_HISTORY", err); }
                     const parsedMsg = {error:err.message, channel: channelName};
-                    sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify(parsedMsg)]);
+                    Server.send(user.id, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify(parsedMsg)]);
                     return;
                 }
 
-                const chan = ctx.channels[channelName];
+                const chan = channel_cache[channelName];
 
                 if (msgCount === 0 && !metadata_cache[channelName] && chan && chan.indexOf(user) > -1) {
                     metadata_cache[channelName] = metadata;
@@ -818,21 +800,22 @@ module.exports.create = function (cfg) {
                             }
                         });
                     }
-                    sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify(metadata)]);
+                    Server.send(user.id, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify(metadata)]);
                 }
 
                 // End of history message:
                 let parsedMsg = {state: 1, channel: channelName};
-                sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify(parsedMsg)]);
+
+                Server.send(user.id, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify(parsedMsg)]);
             });
         });
     };
 
-    const handleGetHistoryRange = function (ctx, seq, user, parsed) {
+    const handleGetHistoryRange = function (Server, seq, user, parsed) {
         var channelName = parsed[1];
         var map = parsed[2];
         if (!(map && typeof(map) === 'object')) {
-            return void sendMsg(ctx, user, [seq, 'ERROR', 'INVALID_ARGS', HISTORY_KEEPER_ID]);
+            return void Server.send(user.id, [seq, 'ERROR', 'INVALID_ARGS', HISTORY_KEEPER_ID]);
         }
 
         var oldestKnownHash = map.from;
@@ -840,14 +823,14 @@ module.exports.create = function (cfg) {
         var desiredCheckpoint = map.cpCount;
         var txid = map.txid;
         if (typeof(desiredMessages) !== 'number' && typeof(desiredCheckpoint) !== 'number') {
-            return void sendMsg(ctx, user, [seq, 'ERROR', 'UNSPECIFIED_COUNT', HISTORY_KEEPER_ID]);
+            return void Server.send(user.id, [seq, 'ERROR', 'UNSPECIFIED_COUNT', HISTORY_KEEPER_ID]);
         }
 
         if (!txid) {
-            return void sendMsg(ctx, user, [seq, 'ERROR', 'NO_TXID', HISTORY_KEEPER_ID]);
+            return void Server.send(user.id, [seq, 'ERROR', 'NO_TXID', HISTORY_KEEPER_ID]);
         }
 
-        sendMsg(ctx, user, [seq, 'ACK']);
+        Server.send(user.id, [seq, 'ACK']);
         return void getOlderHistory(channelName, oldestKnownHash, function (messages) {
             var toSend = [];
             if (typeof (desiredMessages) === "number") {
@@ -863,64 +846,66 @@ module.exports.create = function (cfg) {
                 }
             }
             toSend.forEach(function (msg) {
-                sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id,
+                Server.send(user.id, [0, HISTORY_KEEPER_ID, 'MSG', user.id,
                     JSON.stringify(['HISTORY_RANGE', txid, msg])]);
             });
 
-            sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id,
+            Server.send(user.id, [0, HISTORY_KEEPER_ID, 'MSG', user.id,
                 JSON.stringify(['HISTORY_RANGE_END', txid, channelName])
             ]);
         });
     };
 
-    const handleGetFullHistory = function (ctx, seq, user, parsed) {
+    const handleGetFullHistory = function (Server, seq, user, parsed) {
         // parsed[1] is the channel id
         // parsed[2] is a validation key (optionnal)
         // parsed[3] is the last known hash (optionnal)
-        sendMsg(ctx, user, [seq, 'ACK']);
+
+        Server.send(user.id, [seq, 'ACK']);
 
         // FIXME should we send metadata here too?
         // none of the clientside code which uses this API needs metadata, but it won't hurt to send it (2019-08-22)
-        return void getHistoryAsync(ctx, parsed[1], -1, false, (msg, readMore) => {
+        return void getHistoryAsync(parsed[1], -1, false, (msg, readMore) => {
             if (!msg) { return; }
-            sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify(['FULL_HISTORY', msg])], readMore);
+            Server.send(user.id, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify(['FULL_HISTORY', msg])], readMore);
         }, (err) => {
             let parsedMsg = ['FULL_HISTORY_END', parsed[1]];
             if (err) {
                 Log.error('HK_GET_FULL_HISTORY', err.stack);
                 parsedMsg = ['ERROR', parsed[1], err.message];
             }
-            sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify(parsedMsg)]);
+            Server.send(user.id, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify(parsedMsg)]);
         });
     };
 
-    const handleRPC = function (ctx, seq, user, parsed) {
+    const handleRPC = function (Server, seq, user, parsed) {
         if (typeof(rpc) !== 'function') { return; }
 
         /* RPC Calls...  */
         var rpc_call = parsed.slice(1);
 
-        sendMsg(ctx, user, [seq, 'ACK']);
+        // XXX ensure user is guaranteed to have 'id'
+        Server.send(user.id, [seq, 'ACK']);
         try {
             // slice off the sequence number and pass in the rest of the message
-            rpc(ctx, rpc_call, function (err, output) {
+            rpc(Server, rpc_call, function (err, output) {
                 if (err) {
-                    sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify([parsed[0], 'ERROR', err])]);
+                    Server.send(user.id, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify([parsed[0], 'ERROR', err])]);
                     return;
                 }
                 var msg = rpc_call[0].slice();
                 if (msg[3] === 'REMOVE_OWNED_CHANNEL') {
-                    onChannelDeleted(ctx, msg[4]);
+                    onChannelDeleted(Server, msg[4]);
                 }
                 if (msg[3] === 'CLEAR_OWNED_CHANNEL') {
-                    onChannelCleared(ctx, msg[4]);
+                    onChannelCleared(Server, msg[4]);
                 }
 
                 if (msg[3] === 'SET_METADATA') { // or whatever we call the RPC????
                     // make sure we update our cache of metadata
                     // or at least invalidate it and force other mechanisms to recompute its state
                     // 'output' could be the new state as computed by rpc
-                    onChannelMetadataChanged(ctx, msg[4].channel, output[1]);
+                    onChannelMetadataChanged(Server, msg[4].channel, output[1]);
                 }
 
                 // unauthenticated RPC calls have a different message format
@@ -928,10 +913,10 @@ module.exports.create = function (cfg) {
                     // this is an inline reimplementation of historyKeeperBroadcast
                     // because if we use that directly it will bypass signature validation
                     // which opens up the user to malicious behaviour
-                    let chan = ctx.channels[output.channel];
+                    let chan = channel_cache[output.channel];
                     if (chan && chan.length) {
                         chan.forEach(function (user) {
-                            sendMsg(ctx, user, output.message);
+                            Server.send(user.id, output.message);
                             //[0, null, 'MSG', user.id, JSON.stringify(output.message)]);
                         });
                     }
@@ -941,10 +926,10 @@ module.exports.create = function (cfg) {
                 }
 
                 // finally, send a response to the client that sent the RPC
-                sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify([parsed[0]].concat(output))]);
+                Server.send(user.id, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify([parsed[0]].concat(output))]);
             });
         } catch (e) {
-            sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify([parsed[0], 'ERROR', 'SERVER_ERROR'])]);
+            Server.send(user.id, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify([parsed[0], 'ERROR', 'SERVER_ERROR'])]);
         }
     };
 
@@ -960,7 +945,7 @@ module.exports.create = function (cfg) {
           * check if it's expired and execute all the associated side-effects
           * routes queries to the appropriate handlers
     */
-    const onDirectMessage = function (ctx, seq, user, json) {
+    const onDirectMessage = function (Server, seq, user, json) {
         Log.silly('HK_MESSAGE', json);
 
         let parsed;
@@ -974,23 +959,49 @@ module.exports.create = function (cfg) {
         // If the requested history is for an expired channel, abort
         // Note the if we don't have the keys for that channel in metadata_cache, we'll
         // have to abort later (once we know the expiration time)
-        if (checkExpired(ctx, parsed[1])) { return; }
+        if (checkExpired(Server, parsed[1])) { return; }
 
         // look up the appropriate command in the map of commands or fall back to RPC
         var command = directMessageCommands[parsed[0]] || handleRPC;
 
         // run the command with the standard function signature
-        command(ctx, seq, user, parsed);
+        command(Server, seq, user, parsed);
     };
 
-    return {
+    cfg.historyKeeper = {
         id: HISTORY_KEEPER_ID,
-        setConfig: setConfig,
-        onChannelMessage: onChannelMessage,
-        dropChannel: dropChannel,
-        checkExpired: checkExpired,
-        onDirectMessage: onDirectMessage,
+
+        channelMessage: function (Server, channel, msgStruct) {
+            // netflux-server emits 'channelMessage' events whenever someone broadcasts to a channel
+            // historyKeeper stores these messages if the channel id indicates that they are
+            // a channel type with permanent history
+            onChannelMessage(Server, channel, msgStruct);
+        },
+        channelClose: function (channelName) {
+            // netflux-server emits 'channelClose' events whenever everyone leaves a channel
+            // we drop cached metadata and indexes at the same time
+            dropChannel(channelName);
+        },
+        channelOpen: function (Server, channelName, userId) {
+            channel_cache[channelName] = {};
+            Server.send(userId, [
+                0,
+                HISTORY_KEEPER_ID,
+                'JOIN',
+                channelName
+            ]);
+        },
+        directMessage: function (Server, seq, user, json) {
+            // netflux-server allows you to register an id with a handler
+            // this handler is invoked every time someone sends a message to that id
+            onDirectMessage(Server, seq, user, json);
+        },
     };
-};
 
-}());
+    RPC.create(cfg, function (err, _rpc) {
+        if (err) { throw err; }
+
+        rpc = _rpc;
+        cb(void 0, cfg.historyKeeper);
+    });
+};

lib/hk-util.js

@@ -22,3 +22,12 @@ HK.getHash = function (msg, Log) {
     return msg.slice(0,64);
 };
 
+// historyKeeper should explicitly store any channel
+// with a 32 character id
+HK.STANDARD_CHANNEL_LENGTH = 32;
+
+// historyKeeper should not store messages sent to any channel
+// with a 34 character id
+HK.EPHEMERAL_CHANNEL_LENGTH = 34;
+
+

lib/rpc.js

@@ -24,7 +24,6 @@ const UNAUTHENTICATED_CALLS = [
     'GET_MULTIPLE_FILE_SIZE',
     'IS_CHANNEL_PINNED',
     'IS_NEW_CHANNEL',
-    'GET_HISTORY_OFFSET',
     'GET_DELETED_PADS',
     'WRITE_PRIVATE_MESSAGE',
 ];
@@ -66,25 +65,9 @@ var isUnauthenticateMessage = function (msg) {
     return msg && msg.length === 2 && isUnauthenticatedCall(msg[0]);
 };
 
-var handleUnauthenticatedMessage = function (Env, msg, respond, nfwssCtx) {
+var handleUnauthenticatedMessage = function (Env, msg, respond, Server) {
     Env.Log.silly('LOG_RPC', msg[0]);
     switch (msg[0]) {
-        case 'GET_HISTORY_OFFSET': { // XXX not actually used anywhere?
-            if (typeof(msg[1]) !== 'object' || typeof(msg[1].channelName) !== 'string') {
-                return respond('INVALID_ARG_FORMAT', msg);
-            }
-            const msgHash = typeof(msg[1].msgHash) === 'string' ? msg[1].msgHash : undefined;
-            nfwssCtx.getHistoryOffset(nfwssCtx, msg[1].channelName, msgHash, (e, ret) => {
-                if (e) {
-                    if (e.code !== 'ENOENT') {
-                        Env.WARN(e.stack, msg);
-                    }
-                    return respond(e.message);
-                }
-                respond(e, [null, ret, null]);
-            });
-            break;
-        }
         case 'GET_FILE_SIZE':
             return void Pinning.getFileSize(Env, msg[1], function (e, size) {
                 Env.WARN(e, msg[1]);
@@ -120,7 +103,7 @@ var handleUnauthenticatedMessage = function (Env, msg, respond, nfwssCtx) {
                 respond(e, [null, isNew, null]);
             });
         case 'WRITE_PRIVATE_MESSAGE':
-            return void Channel.writePrivateMessage(Env, msg[1], nfwssCtx, function (e, output) {
+            return void Channel.writePrivateMessage(Env, msg[1], Server, function (e, output) {
                 respond(e, output);
             });
         default:
@@ -134,7 +117,7 @@ var handleAuthenticatedMessage = function (Env, map) {
     var safeKey = map.safeKey;
     var publicKey = map.publicKey;
     var Respond = map.Respond;
-    var ctx = map.ctx;
+    var Server = map.Server;
 
     Env.Log.silly('LOG_RPC', msg[0]);
     switch (msg[0]) {
@@ -265,7 +248,7 @@ var handleAuthenticatedMessage = function (Env, map) {
                 Respond(e);
             });
         case 'ADMIN':
-            return void Admin.command(Env, ctx, safeKey, msg[1], function (e, result) { // XXX SPECIAL
+            return void Admin.command(Env, Server, safeKey, msg[1], function (e, result) { // XXX SPECIAL
                 if (e) {
                     Env.WARN(e, result);
                     return void Respond(e);
@@ -285,7 +268,7 @@ var handleAuthenticatedMessage = function (Env, map) {
     }
 };
 
-var rpc = function (Env, ctx, data, respond) {
+var rpc = function (Env, Server, data, respond) {
     if (!Array.isArray(data)) {
         Env.Log.debug('INVALID_ARG_FORMET', data);
         return void respond('INVALID_ARG_FORMAT');
@@ -304,7 +287,7 @@ var rpc = function (Env, ctx, data, respond) {
     }
 
     if (isUnauthenticateMessage(msg)) {
-        return handleUnauthenticatedMessage(Env, msg, respond, ctx);
+        return handleUnauthenticatedMessage(Env, msg, respond);
     }
 
     var signature = msg.shift();
@@ -369,7 +352,7 @@ var rpc = function (Env, ctx, data, respond) {
         safeKey: safeKey,
         publicKey: publicKey,
         Respond: Respond,
-        ctx: ctx,
+        Server: Server,
     });
 };
 
@@ -394,6 +377,7 @@ RPC.create = function (config, cb) {
     };
 
     var Env = {
+        historyKeeper: config.historyKeeper,
         defaultStorageLimit: config.defaultStorageLimit,
         maxUploadSize: config.maxUploadSize || (20 * 1024 * 1024),
         Sessions: {},
@@ -465,11 +449,9 @@ RPC.create = function (config, cb) {
             Env.blobStore = blob;
         }));
     }).nThen(function () {
-        // XXX it's ugly that we pass ctx and Env separately
-        // when they're effectively the same thing...
-        cb(void 0, function (ctx, data, respond) {
+        cb(void 0, function (Server, data, respond) {
             try {
-                return rpc(Env, ctx, data, respond);
+                return rpc(Env, Server, data, respond);
             } catch (e) {
                 console.log("Error from RPC with data " + JSON.stringify(data));
                 console.log(e.stack);

server.js

@@ -4,17 +4,12 @@
 var Express = require('express');
 var Http = require('http');
 var Fs = require('fs');
-var WebSocketServer = require('ws').Server;
-var NetfluxSrv = require('chainpad-server/NetfluxWebsocketSrv');
 var Package = require('./package.json');
 var Path = require("path");
 var nThen = require("nthen");
 
 var config = require("./lib/load-config");
 
-// support multiple storage back ends
-var Storage = require('./storage/file');
-
 var app = Express();
 
 // mode can be FRESH (default), DEV, or PACKAGE
@@ -69,11 +64,9 @@ var setHeaders = (function () {
     if (Object.keys(headers).length) {
         return function (req, res) {
             const h = [
-                    /^\/pad(2)?\/inner\.html.*/,
+                    /^\/pad\/inner\.html.*/,
                     /^\/common\/onlyoffice\/.*\/index\.html.*/,
-                    /^\/sheet\/inner\.html.*/,
-                    /^\/ooslide\/inner\.html.*/,
-                    /^\/oodoc\/inner\.html.*/,
+                    /^\/(sheet|ooslide|oodoc)\/inner\.html.*/,
                 ].some((regex) => {
                     return regex.test(req.url)
                 }) ? padHeaders : headers;
@@ -117,11 +110,6 @@ app.use(function (req, res, next) {
 
 app.use(Express.static(__dirname + '/www'));
 
-Fs.exists(__dirname + "/customize", function (e) {
-    if (e) { return; }
-    console.log("Cryptpad is customizable, see customize.dist/readme.md for details");
-});
-
 // FIXME I think this is a regression caused by a recent PR
 // correct this hack without breaking the contributor's intended behaviour.
 
@@ -207,80 +195,36 @@ app.use(function (req, res, next) {
 
 var httpServer = Http.createServer(app);
 
-httpServer.listen(config.httpPort,config.httpAddress,function(){
-    var host = config.httpAddress;
-    var hostName = !host.indexOf(':') ? '[' + host + ']' : host;
-
-    var port = config.httpPort;
-    var ps = port === 80? '': ':' + port;
-
-    console.log('[%s] server available http://%s%s', new Date().toISOString(), hostName, ps);
-});
-if (config.httpSafePort) {
-    Http.createServer(app).listen(config.httpSafePort, config.httpAddress);
-}
-
-var wsConfig = { server: httpServer };
+nThen(function (w) {
+    Fs.exists(__dirname + "/customize", w(function (e) {
+        if (e) { return; }
+        console.log("Cryptpad is customizable, see customize.dist/readme.md for details");
+    }));
+}).nThen(function (w) {
+    httpServer.listen(config.httpPort,config.httpAddress,function(){
+        var host = config.httpAddress;
+        var hostName = !host.indexOf(':') ? '[' + host + ']' : host;
 
-var rpc;
-var historyKeeper;
+        var port = config.httpPort;
+        var ps = port === 80? '': ':' + port;
 
-var log;
+        console.log('[%s] server available http://%s%s', new Date().toISOString(), hostName, ps);
+    });
 
-// Initialize logging, the the store, then tasks, then rpc, then history keeper and then start the server
-var nt = nThen(function (w) {
-    // set up logger
-    var Logger = require("./lib/log");
-    //console.log("Loading logging module");
-    Logger.create(config, w(function (_log) {
-        log = config.log = _log;
-    }));
-}).nThen(function (w) {
-    if (config.externalWebsocketURL) {
-        // if you plan to use an external websocket server
-        // then you don't need to load any API services other than the logger.
-        // Just abort.
-        w.abort();
-        return;
+    if (config.httpSafePort) {
+        Http.createServer(app).listen(config.httpSafePort, config.httpAddress, w());
     }
-    Storage.create(config, w(function (_store) {
-        config.store = _store;
-    }));
-}).nThen(function (w) {
-    var Tasks = require("./storage/tasks");
-    Tasks.create(config, w(function (e, tasks) {
-        if (e) {
-            throw e;
-        }
-        config.tasks = tasks;
-        if (config.disableIntegratedTasks) { return; }
-        setInterval(function () {
-            tasks.runAll(function (err) {
-                if (err) {
-                    // either TASK_CONCURRENCY or an error with tasks.list
-                    // in either case it is already logged.
-                }
-            });
-        }, 1000 * 60 * 5); // run every five minutes
-    }));
-}).nThen(function (w) {
-    require("./lib/rpc").create(config, w(function (e, _rpc) {
-        if (e) {
-            w.abort();
-            throw e;
-        }
-        rpc = _rpc;
-    }));
 }).nThen(function () {
-    var HK = require('./lib/historyKeeper.js');
-    var hkConfig = {
-        tasks: config.tasks,
-        rpc: rpc,
-        store: config.store,
-        log: log,
-    };
-    historyKeeper = HK.create(hkConfig);
-}).nThen(function () {
-    var wsSrv = new WebSocketServer(wsConfig);
-    NetfluxSrv.run(wsSrv, config, historyKeeper);
+    var wsConfig = { server: httpServer };
+
+    // Initialize logging then start the API server
+    require("./lib/log").create(config, function (_log) {
+        config.log = _log;
+        config.httpServer = httpServer;
+
+        if (config.externalWebsocketURL) { return; }
+        require("./lib/api").create(config);
+    });
 });
+
+

storage/file.js

@@ -955,7 +955,9 @@ var trimChannel = function (env, channelName, hash, _cb) {
         var handler = function (msgObj, readMore, abort) {
             if (ABORT) { return void abort(); }
             // the first message might be metadata... ignore it if so
-            if (i++ === 0 && msgObj.buff.indexOf('{') === 0) { return; }
+            if (i++ === 0 && msgObj.buff.indexOf('{') === 0) {
+                return readMore();
+            }
 
             if (retain) {
                 // if this flag is set then you've already found
@@ -973,6 +975,9 @@ var trimChannel = function (env, channelName, hash, _cb) {
             if (msgHash === hash) {
                 // everything from this point on should be retained
                 retain = true;
+                return void tempStream.write(msgObj.buff, function () {
+                    readMore();
+                });
             }
         };