From 779e8174432fe0c0a361c1cc5991fc3a4365cf97 Mon Sep 17 00:00:00 2001 From: ansuz Date: Mon, 3 Feb 2020 14:20:05 -0500 Subject: [PATCH] stop relying on netflux-server internals * create RPC module from inside historyKeeper * stop passing around netflux-server context * update to use newer netflux-server's formal APIs * manage your own cache of indexes instead of storing things in the netflux context --- lib/api.js | 66 ++++++----- lib/commands/admin-rpc.js | 35 ++---- lib/commands/channel.js | 7 +- lib/historyKeeper.js | 236 ++++++++++++++++++++------------------ lib/rpc.js | 38 ++---- 5 files changed, 185 insertions(+), 197 deletions(-) diff --git a/lib/api.js b/lib/api.js index c6491d5db..b66dcb4de 100644 --- a/lib/api.js +++ b/lib/api.js @@ -4,8 +4,6 @@ const WebSocketServer = require('ws').Server; const NetfluxSrv = require('chainpad-server'); module.exports.create = function (config) { - var rpc; - const log = config.log; const wsConfig = { server: config.httpServer, }; @@ -32,35 +30,45 @@ module.exports.create = function (config) { }); }, 1000 * 60 * 5); // run every five minutes })); - }).nThen(function (w) { - require("./rpc").create(config, w(function (e, _rpc) { - if (e) { - w.abort(); - throw e; - } - rpc = _rpc; - })); }).nThen(function () { - var HK = require('./historyKeeper.js'); - var hkConfig = { - tasks: config.tasks, - rpc: rpc, - store: config.store, - log: log, - }; + // asynchronously create a historyKeeper and RPC together + require('./historyKeeper.js').create(config, function (err, historyKeeper) { + if (err) { throw err; } - var historyKeeper = HK.create(hkConfig); + var log = config.log; - NetfluxSrv.create(new WebSocketServer(wsConfig)) - .on('channelClose', historyKeeper.channelClose) - .on('channelMessage', historyKeeper.channelMessage) - .on('channelOpen', historyKeeper.channelOpen) - .on('sessionClose', function (userId, reason) { - reason = reason; // XXX - }) - .on('error', function (error, label, info) { - info = info; // XXX - }) - .register(historyKeeper.id, historyKeeper.directMessage); + // 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); + }); }); }; diff --git a/lib/commands/admin-rpc.js b/lib/commands/admin-rpc.js index 24a8e4348..763bfb517 100644 --- a/lib/commands/admin-rpc.js +++ b/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'); } diff --git a/lib/commands/channel.js b/lib/commands/channel.js index 052aa3c44..f232ccea8 100644 --- a/lib/commands/channel.js +++ b/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 diff --git a/lib/historyKeeper.js b/lib/historyKeeper.js index 92eb84a5c..30500f180 100644 --- a/lib/historyKeeper.js +++ b/lib/historyKeeper.js @@ -9,6 +9,8 @@ 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; @@ -66,8 +68,8 @@ const isValidValidateKeyString = function (key) { var CHECKPOINT_PATTERN = /^cp\|(([A-Za-z0-9+\/=]+)\|)?/; -module.exports.create = function (cfg) { - const rpc = cfg.rpc; +module.exports.create = function (cfg, cb) { + var rpc; const tasks = cfg.tasks; const store = cfg.store; Log = cfg.log; @@ -75,6 +77,7 @@ 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); @@ -211,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 @@ -233,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 @@ -260,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) { @@ -284,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 @@ -298,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; @@ -313,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) { - ctx.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, @@ -336,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 @@ -347,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; } @@ -362,18 +355,16 @@ 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 @@ -391,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 @@ -414,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); @@ -429,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; } @@ -479,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. @@ -522,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 @@ -600,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); @@ -666,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) - ctx.sendMsg(ctx, user, [seq, 'ACK']); + + Server.send(user.id, [seq, 'ACK']); var channelName = parsed[1]; var config = parsed[2]; var metadata = {}; @@ -736,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... @@ -750,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 - ctx.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(); } - ctx.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}; - ctx.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; @@ -811,21 +800,22 @@ module.exports.create = function (cfg) { } }); } - ctx.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}; - ctx.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 ctx.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; @@ -833,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 ctx.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 ctx.sendMsg(ctx, user, [seq, 'ERROR', 'NO_TXID', HISTORY_KEEPER_ID]); + return void Server.send(user.id, [seq, 'ERROR', 'NO_TXID', HISTORY_KEEPER_ID]); } - ctx.sendMsg(ctx, user, [seq, 'ACK']); + Server.send(user.id, [seq, 'ACK']); return void getOlderHistory(channelName, oldestKnownHash, function (messages) { var toSend = []; if (typeof (desiredMessages) === "number") { @@ -856,64 +846,66 @@ module.exports.create = function (cfg) { } } toSend.forEach(function (msg) { - ctx.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])]); }); - ctx.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) - ctx.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; } - ctx.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]; } - ctx.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); - ctx.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) { - ctx.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 @@ -921,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) { - ctx.sendMsg(ctx, user, output.message); + Server.send(user.id, output.message); //[0, null, 'MSG', user.id, JSON.stringify(output.message)]); }); } @@ -934,10 +926,10 @@ module.exports.create = function (cfg) { } // finally, send a response to the client that sent the RPC - ctx.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) { - ctx.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'])]); } }; @@ -953,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; @@ -967,33 +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, - channelMessage: function (ctx, channel, msgStruct) { - onChannelMessage(ctx, channel, msgStruct); + + 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 (ctx, channelName, user) { - ctx.sendMsg(ctx, user, [ + channelOpen: function (Server, channelName, userId) { + channel_cache[channelName] = {}; + Server.send(userId, [ 0, - HISTORY_KEEPER_ID, // ctx.historyKeeper.id + HISTORY_KEEPER_ID, 'JOIN', channelName ]); }, - directMessage: function (ctx, seq, user, json) { - onDirectMessage(ctx, seq, user, json); + 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); + }); }; diff --git a/lib/rpc.js b/lib/rpc.js index 171fb590f..47ea9976f 100644 --- a/lib/rpc.js +++ b/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);