diff options
| -rwxr-xr-x | channels-v2.js | 285 | ||||
| -rw-r--r-- | pull-many-v2.js | 95 |
2 files changed, 380 insertions, 0 deletions
diff --git a/channels-v2.js b/channels-v2.js new file mode 100755 index 0000000..5e96694 --- /dev/null +++ b/channels-v2.js @@ -0,0 +1,285 @@ +const fs = require("fs"); +const home = require("os").homedir(); + +var argv = require("minimist")(process.argv.slice(2)); +var clientFactory = require("ssb-client"); +var path = require("path"); +var pull = require("pull-stream"); +var many = require("./pull-many-v2"); + +// Change to 'true' to get debugging output +DEBUG = false + +DEFAULT_CHANNEL_METADATA = { + lastMsgTimestamp: 0 +} +DEFAULT_CHANNEL_OBJECT = { + messages: [] +} +DEFAULT_CHANNEL_OBJECT_SIZE = 15 +GRAPH_TYPE = "follow" +MESSAGE_SEPERATOR = "---------------" +MESSAGE_SEPERATOR_COLOR = "\x1b[33m" +MESSAGE_COLOR = "\x1b[0m" +SBOT_ROOT = path.join(home, ".ssb"); +SBOT_CHANNEL_DIR = path.join(SBOT_ROOT, "channels"); +SBOT_CHANNEL_DATA = path.join(SBOT_CHANNEL_DIR, ".data"); + +class StreamData { + constructor() { + this.finished = false, + this.oldestTimestampSeen = Number.MAX_SAFE_INTEGER, + this.waitingMessages = [] + + this.markFinished = function() { + this.finished = true, + this.oldestTimestampSeen = 0 + } + } +} + +if (!fs.existsSync(SBOT_ROOT)) { + debug("no ~/.ssb folder detected, creating it..."); + fs.mkdirSync(SBOT_ROOT); +} + +if (!fs.existsSync(SBOT_CHANNEL_DIR)) { + debug("no channels folder detected, creating it..."); + fs.mkdirSync(SBOT_CHANNEL_DIR); +} + +if (!fs.existsSync(SBOT_CHANNEL_DATA)) { + debug("no channel metadata file found, creating it..."); + fs.writeFileSync(SBOT_CHANNEL_DATA, "{}"); +} + +debug("Creating client..."); +clientFactory(function (err, client) { + if(err) { + console.log("[FATAL] Error when starting scuttlebot: " + err); + throw err; + } + debug("Successfully created client"); + + + main(client); +}); + +function main(client) { + // var metadata = JSON.parse(fs.readFileSync(SBOT_CHANNEL_DATA)); + + if(!argv._ || argv._.length == 0) { + showHelp(metadata); + client.close(true, () => {}); + process.exit(0); + } + + var channelName = "#" + argv._[0]; + // var trackedChannelData = getChannelData(channelName, metadata); // only get data for the first channel given (easy to change this if needed) + // var trackedChannelMessages = getChannelMessages(channelName); + var trackedChannelMessages = DEFAULT_CHANNEL_OBJECT; + // debug("fetched " + trackedChannelMessages["messages"].length + " messages from cache"); + + client.friends.hops({ + dunbar: Number.MAX_SAFE_INTEGER, + max: 3 + }, + function(err, hops) { + if(err) { + console.log("[FATAL] Could not retreive friends list (is ssb-friends installed?)") + client.close(true, () => {}); + process.exit(4); + } + + fetchChannelMessagesFrom(client, channelName, trackedChannelMessages, Object.keys(hops)); + } + ) +} + +function fetchChannelMessagesFrom(client, channelName, trackedChannelMessages, hops) { + // So it seems like Scuttlebutt IDs are case insensitive + // This means that you can accidentially capitalize/uncapitalize letters when following someone, and the follow will still go through + // However, the typo will remain in your friend graph and ruin any attempt made to query for that ID in its normal capitalization + // To avoid this skullfuckery, we just use a list and map lower() onto it beforehand + var followedIds = hops.map(id => id.toLowerCase()); + + debug("Fetching messages from IDs in " + JSON.stringify(followedIds)); + + var streamController = { + assembledMessages: [], + streamData: { + hashtagStream: new StreamData(), + channelStream: new StreamData() + }, + pushMessage: function(source, newMsg) { + var streamData = source == "hashtag" ? this.streamData.hashtagStream : this.streamData.channelStream; + streamData.oldestTimestampSeen = newMsg.value.timestamp; + streamData.waitingMessages.push(newMsg); + + this.printNewlySafeMessages(); + }, + printNewlySafeMessages: function() { + var newlySafeMessages = []; + var oldestSafeTimestamp = Math.max(...Object.values(this.streamData).map(datum => datum.oldestTimestampSeen)); + debug("oldest safe timestamp:" + oldestSafeTimestamp); + + for(var streamDataIndex in this.streamData) { + var streamDatum = this.streamData[streamDataIndex] + debug("length of waiting messages: " + streamDatum.waitingMessages.length); + if(streamDatum.waitingMessages.length) { + debug("timestamp of first waiting message: " + streamDatum.waitingMessages[0].value.timestamp); + } + while(streamDatum.waitingMessages.length && streamDatum.waitingMessages[0].value.timestamp >= oldestSafeTimestamp) { + var safeMsg = streamDatum.waitingMessages.shift(); // pop the newest waiting message + newlySafeMessages.push(safeMsg); + } + } + + // in general, newlySafeMessages might not be in order, we should sort before appending + newlySafeMessages.sort((msg1, msg2) => (msg1.value.timestamp > msg2.value.timestamp ? -1 : 1)); + this.assembledMessages = this.assembledMessages.concat(newlySafeMessages); + for(var msgIndex in newlySafeMessages) { + printMsg(newlySafeMessages[msgIndex]); + } + debug("***************"); + }, + finish: function() { + for(var datumIndex in this.streamData) { + this.streamData[datumIndex].oldestTimestampSeen = 0; + } + this.printNewlySafeMessages(); + } + } + + var hashtagStream = createHashtagStream(client, channelName); + var channelStream = createChannelStream(client, channelName); + var stream = many([hashtagStream, channelStream]); + + pull(stream, pull.filter(function(msg) { + return followedIds.includes(msg.data.value.author.toLowerCase()); + }), pull.filter(function(msg) { + var hasHashtag = msg.data.value.content && msg.data.value.content.text && typeof(msg.data.value.content.text) == "string" && msg.data.value.content.text.includes(channelName); + if(msg.source == "hashtag") { + return hasHashtag; + } + else { + return !hasHashtag; // prevents us from double-counting messages with both the hashtag and the channel header + } + }), pull.drain(function(msg) { + streamController.pushMessage(msg.source, msg.data); + }, function() { + streamController.finish(); + client.close(true, () => {}); + process.exit(0); + })); +} + +function saveMessages(client, taskData, channelName, trackedChannelMessages, metadata) { + debug("found " + taskData.msgsFound + " new messages"); + debug(JSON.stringify(taskData)); + console.log("Found " + trackedChannelMessages["messages"].length + " messages:"); + + updateChannelMessages(channelName, trackedChannelMessages, metadata, taskData.lastTimestamp); + +} + +function showHelp(metadata) { + console.log("No channel provided. To get messages for a channel, run 'node channels.js example'"); + console.log("(note that 'node channels.js #example' will not work, because # is a protected character)"); + + var trackedChannels = getTrackedChannels(metadata); + if(trackedChannels.length) { + console.log("Currently tracked channels are:"); + for(const channelNameIndex in trackedChannels) { + var stats = fs.statSync(path.join(SBOT_CHANNEL_DIR, trackedChannels[channelNameIndex])); + if(stats.size > DEFAULT_CHANNEL_OBJECT_SIZE) { // don't say we're tracking empty channels -- if they were interesting, they'd have stuff in them + console.log(trackedChannels[channelNameIndex]); + } + } + } + else { + console.log("No channels are currently tracked."); + } +} + +function getTrackedChannels(data) { + return Object.keys(data); +} + +function getChannelData(channelName, data) { + if(!data[channelName]) { + data[channelName] = { + lastMsgTimestamp: 0, + } + } + + return data[channelName]; +} + +function getChannelMessages(channelName) { + var messageCachePath = path.join(SBOT_CHANNEL_DIR, channelName); + if (!fs.existsSync(messageCachePath)) { + return DEFAULT_CHANNEL_OBJECT; + } + + var raw = fs.readFileSync(messageCachePath, "utf8"); + return JSON.parse(raw); +} + +function createHashtagStream(client, channelName) { + var search = client.search && client.search.query; + if(!search) { + console.log("[FATAL] ssb-search plugin must be installed to us channels"); + process.exit(5); + } + + var query = search({ + query: channelName + }); + + query.streamName = "hashtag"; // mark the stream object so we can tell which stream an object came from + + return query; +} + +function createChannelStream(client, channelName) { + var query = client.query.read({ + query: [{ + "$filter": { + value: { + content: { + channel: channelName.substring(1) + } + } + } + }], + reverse: true + }); + + query.streamName = "channel"; // mark the stream object so we can tell which stream a message came from later + + return query; +} + +function printMsg(msg) { + console.log(MESSAGE_SEPERATOR_COLOR, MESSAGE_SEPERATOR); + console.log(MESSAGE_COLOR, msg.value.content.text); +} + +function updateChannelMessages(channelName, channelJson, metadata, latestTimestamp) { + fs.writeFileSync(path.join(SBOT_CHANNEL_DIR, channelName), JSON.stringify(channelJson)); + + if(metadata.channelName == undefined) { + metadata[channelName] = DEFAULT_CHANNEL_METADATA; + } + + metadata[channelName].lastMsgTimestamp = latestTimestamp; // once we've successfully updated messages, update the most recent message ID + fs.writeFileSync(SBOT_CHANNEL_DATA, JSON.stringify(metadata)); +} + +function debug(message) { + if(DEBUG) { + var timestamp = new Date(); + console.log("[" + timestamp.toISOString() + "] " + message); + } +} diff --git a/pull-many-v2.js b/pull-many-v2.js new file mode 100644 index 0000000..beea621 --- /dev/null +++ b/pull-many-v2.js @@ -0,0 +1,95 @@ +// All credit to https://github.com/pull-stream/pull-many/blob/master/index.js +// Edits have been made to distinguish the original stream that each object came from + +module.exports = function (ary) { + + var capped = !!ary + var inputs = (ary || []).map(create), i = 0, abort, cb + + function create (stream) { + return {ready: false, reading: false, ended: false, read: stream, data: null} + } + + function check () { + if(!cb) return + clean() + var l = inputs.length + var _cb = cb + if(l === 0 && (abort || capped)) { + cb = null; _cb(abort || true) + return + } + + //scan the inputs to check whether there is one we can use. + for(var j = 0; j < l; j++) { + var current = inputs[(i + j) % l] + if(current.ready && !current.ended) { + var data = { // [EDIT] keep track of which exact source the data came from + data: current.data, + source: current.read.streamName + } + current.ready = false + current.data = null + i ++; cb = null + return _cb(null, data) + } + } + } + + function clean () { + var l = inputs.length + //iterate backwards so that we can remove items. + while(l--) { + if(inputs[l].ended) + inputs.splice(l, 1) + } + } + + function next () { + var l = inputs.length + while(l--) + (function (current) { + //read the next item if we aren't already + if(l > inputs.length) throw new Error('this should never happen') + if(current.reading || current.ended || current.ready) return + current.reading = true + var sync = true + current.read(abort, function next (end, data) { + current.data = data + current.ready = true + current.reading = false + + if(end === true || abort) current.ended = true + else if(end) abort = current.ended = end + //check whether we need to abort this stream. + if(abort && !end) current.read(abort, next) + if(!sync) check() + }) + sync = false + })(inputs[l]) + + //scan the feed + check() + } + + function read (_abort, _cb) { + abort = abort || _abort; cb = _cb; next() + } + + read.add = function (stream) { + if(!stream) { + //the stream will now end when all the streams end. + capped = true + //we just changed state, so we may need to cb + return next() + } + inputs.push(create(stream)) + next() + } + + read.cap = function (err) { + read.add(null) + } + + return read +} |