Implement channel archiving

index.js

@@ -12,6 +12,7 @@ var createMessagesView = require('./views/messages')
 var createTopicsView = require('./views/topics')
 var createUsersView = require('./views/users')
 var createModerationView = require('./views/moderation')
+var createArchivingView = require('./views/channel-archiving')
 var swarm = require('./swarm')
 
 var DATABASE_VERSION = 1
@@ -22,6 +23,7 @@ var TOPICS = 't'
 var USERS = 'u'
 var MODERATION_AUTH = 'mx'
 var MODERATION_INFO = 'my'
+var ARCHIVES = 'a' 
 
 module.exports = Cabal
 module.exports.databaseVersion = DATABASE_VERSION
@@ -96,13 +98,16 @@ function Cabal (storage, key, opts) {
     sublevel(this.db, MODERATION_AUTH, { valueEncoding: json }),
     sublevel(this.db, MODERATION_INFO, { valueEncoding: json })
   ))
+  this.kcore.use('archives', createArchivingView(
+    sublevel(this.db, ARCHIVES, { valueEncoding: json })))
 
   this.messages = this.kcore.api.messages
   this.channels = this.kcore.api.channels
   this.memberships = this.kcore.api.memberships
   this.topics = this.kcore.api.topics
   this.users = this.kcore.api.users
   this.moderation = this.kcore.api.moderation
+  this.archives = this.kcore.api.archives
 }
 
 inherits(Cabal, events.EventEmitter)

views/channel-archiving.js

@@ -0,0 +1,121 @@
+const EventEmitter = require('events').EventEmitter
+const pump = require('pump')
+const Writable = require('readable-stream').Writable
+const makeView = require('kappa-view')
+
+/*
+view data structure, the value (1) doesn't matter
+archive!<channelname>: '<pubkey of archiver>:<reason string>'
+archive!default: 'feed..b01:unused channel'
+
+ARCHIVE msg: leveldb PUT
+UNARCHIVE msg: leveldb DEL
+
+message schema: 
+{ 
+  channel: <channel>,
+  type: 'channel/archive' or 'channel/unarchive',
+  key: <pubkey of archiver>,
+  reason: <optional string reason for archiving>
+}
+*/
+
+function getAuthorizedKeys (kcore, cb) {
+  return Promise.all([getKeys('admin'), getKeys('mod')]).then(res => cb(res[0].concat(res[1]).map(row => row.id)))
+
+  // collect pubkeys for cabal-wide mods or admins
+  function getKeys (flag) {
+    return new Promise((resolve, reject) => {
+      function processResult (err, result) {
+        if (err) resolve([])
+        resolve(result)
+      }
+      kcore.api.moderation.listByFlag({ flag, channel: '@' }, processResult)
+    })
+  }
+}
+
+module.exports = function (lvl) {
+  var events = new EventEmitter()
+
+  return makeView(lvl, function (db) {
+    return {
+      maxBatch: 500,
+      map: function (msgs, next) {
+        // 1. go over each msg
+        // 2. check if it's an archive/unarchive msg (skip if not)
+        // 3. accumulate a levelup PUT or DEL operation for it
+        // 4. write it all to leveldb as a batch
+        const ops = []
+        const seen = {}
+        msgs.forEach(function (msg) {
+          if (!sanitize(msg)) { return }
+          const channel = msg.value.content.channel
+          const reason = msg.value.content.reason || ''
+          const key = msg.key
+          const pair = channel+key // a composite key, to know if we have seen this pair
+          if (msg.value.type === 'channel/archive') {
+            ops.push({
+              type: 'put',
+              key: `archive!${channel}!${key}`,
+              value: `${reason}`
+            })
+
+            if (!seen[pair]) events.emit('archive', channel, reason, key)
+            seen[pair] = true
+          } else if (msg.value.type === 'channel/unarchive') {
+            ops.push({
+              type: 'del', 
+              key: `archive!${channel}!${key}`,
+            })
+            if (seen[pair]) { delete seen[pair] }
+            events.emit('unarchive', channel, reason, key)
+          }
+        })
+        if (ops.length) db.batch(ops, next)
+        else next()
+      },
+      api: {
+        // get the list of currently archived channels 
+        // TODO: include stored values (use lvl.createValueStream() somehow?)
+        get: function (core, cb) {
+          this.ready(function () {
+            // query mod view to determine if archiver is either a mod or an admin 
+            getAuthorizedKeys(core, authorizedKeys => {
+              const channels = []
+              db.createKeyStream({
+                gt: 'archive!!',
+                lt: 'archive!~'
+              })
+                .on('data', function (row) {
+                  // structure of `row`: archive!<channel>!<key>
+                  const pieces = row.split('!')
+                  const channel = pieces[1]
+                  const key = pieces[2]
+                  if (authorizedKeys.indexOf(key) >= 0) {
+                    channels.push(channel)
+                  }
+                })
+                .once('end', function () {
+                  cb(null, channels)
+                })
+                .once('error', cb)
+            })
+          })
+        },
+        events: events
+      }
+    }
+  })
+}
+
+// Either returns a well-formed chat message, or null.
+function sanitize (msg) {
+  if (typeof msg !== 'object') return null
+  if (typeof msg.value !== 'object') return null
+  if (typeof msg.value.content !== 'object') return null
+  if (typeof msg.value.timestamp !== 'number') return null
+  if (typeof msg.value.type !== 'string') return null
+  if (typeof msg.value.content.channel !== 'string') return null
+  return msg
+}