refactoring nodejs-ssb

* Extracted pipelines
* Extracted caches
* Extracted job queues

still works.

related to #105

Author: soapdog

.jshintrc

@@ -3,8 +3,9 @@
   "quotmark": "double",
   "camelcase": true,
   "browser": true,
+  "semi": false,
   "globals": {
   	"Dexie": true,
   	"_": true
   }
-}
+}

src/core/components/ui/Card.svelte

@@ -2,7 +2,7 @@
   const MessageDropdown = require("../MessageDropdown.svelte")
   const MessageRaw = require("../MessageRaw.svelte")
   const AvatarTile = require("../AvatarTile.svelte")
-  const { isMessageBlured } = require("../../platforms/nodejs-ssb/abusePrevention.js")
+  const { isMessageBlured } = require("../../platforms/common/abusePrevention.js")
 
   export let msg
   export let showRaw = false

src/core/core.js

@@ -1,7 +1,7 @@
 const kernel = require("./kernel/kernel.js")
 const runtimes = require("./runtimes/runtimes.js")
 const queryString = require("query-string")
-const { isMessageHidden } = require("./platforms/nodejs-ssb/abusePrevention.js")
+const { isMessageHidden } = require("./platforms/common/abusePrevention.js")
 
 
 if (window) {

src/core/platforms/common/abusePrevention.js

@@ -0,0 +1,138 @@
+const { getPref, setPref } = require("../../kernel/prefs.js")
+const _ = require("lodash")
+
+const getFilters = () => getPref("filters", [])
+
+const addFilter = filter => {
+  let currentFilters = getFilters()
+
+  filter.action = filter.action || "hide"
+  filter.feed = filter.feed || false
+  filter.channel = filter.channel || false
+  filter.keywords = filter.keywords || false
+  filter.expires = filter.expires || false
+
+  currentFilters.push(filter)
+
+  setPref("filters", currentFilters)
+}
+
+const deleteFilter = filter => {
+  let currentFilters = getFilters()
+
+  setPref(
+    "filters",
+    currentFilters.filter(f => {
+      let keys = Object.keys(filter)
+      let include = false
+      keys.forEach(k => {
+        if (filter[k] !== f[k]) include = true
+      })
+      return include
+    })
+  )
+}
+
+const isMessageBlured = msg => {
+  let currentFilters = getFilters().filter(f => f.action == "blur")
+  if (currentFilters.length > 0) {
+    let res = currentFilters.map(f => isMessageFiltered(msg, f, "blur"))
+    return !res.some(r => r)
+  } else {
+    return false
+  }
+}
+
+const isChannelFiltered = channel => {
+  let currentFilters = getFilters().filter(f => f.action == "hide")
+  if (currentFilters.length > 0) {
+    let res = currentFilters.map(filter => {
+      if (channel && channel.startsWith("#")) {
+        channel = channel.slice(1)
+      }
+
+      if (filter.expires) {
+        let expirationDate = new Date(filter.expires)
+        let today = new Date()
+
+        if (today > expirationDate) {
+          return true
+        }
+      }
+
+      if (channel == filter.channel) {
+        return false
+      }
+
+      return true
+    })
+
+    return !res.some(r => r)
+  } else {
+    return false
+  }
+}
+
+const isMessageHidden = msg => {
+  let currentFilters = getFilters().filter(f => f.action == "hide")
+  if (currentFilters.length > 0) {
+    let res = currentFilters.map(f => isMessageFiltered(msg, f, "hide"))
+    res = !res.some(r => !r)
+    return res
+  } else {
+    return true // true because it is used by a pull.filter()
+  }
+}
+
+const isMessageFiltered = (msg, filter, action) => {
+  let filterResults = []
+  if (filter.action !== action) {
+    return true
+  }
+
+  if (filter.expires) {
+    let expirationDate = new Date(filter.expires)
+    let today = new Date()
+
+    if (today > expirationDate) {
+      return true
+    }
+  }
+
+  if (filter.feed) {
+    if (filter.feed == msg.value.author) {
+      filterResults.push(true)
+    } else {
+      filterResults.push(false)
+    }
+  }
+
+  if (filter.channel) {
+    if (msg.value.content.channel && filter.channel == msg.value.content.channel) {
+      filterResults.push(true)
+    } else {
+      filterResults.push(false)
+    }
+  }
+
+  if (filter.keywords.length > 0 && msg.value.content.type == "post" && msg.value.content.text) {
+    let keywords = filter.keywords
+    let content = msg.value.content.text.toLowerCase()
+
+    let res = keywords.map(k => content.includes(k.toLowerCase())).some(r => r)
+    if (res) console.log("filtered due to keywords")
+    filterResults.push(res)
+  }
+
+  return !filterResults.some(n => n == true)
+}
+
+module.exports = {
+  getFilters,
+  isMessageBlured,
+  isMessageFiltered,
+  isMessageHidden,
+  isChannelFiltered,
+  addFilter,
+  deleteFilter,
+}

src/core/platforms/common/cache.js

@@ -0,0 +1,55 @@
+
+const caches = {}
+const cacheResult = (kind, msgId, value) => {
+  let key = `cache-${kind}-${msgId}`
+  caches[key] = {
+    time: Date.now(),
+    value,
+  }
+}
+const invalidateCacheResult = (kind, msgId) => {
+  let key = `cache-${kind}-${msgId}`
+  delete caches[key]
+}
+
+const resultFromCache = (kind, msgId, falseIfOlderThan) => {
+  let key = `cache-${kind}-${msgId}`
+  let currentDate = Date.now()
+  if (caches.hasOwnProperty(key)) {
+    let expiryDate = caches[key].time + falseIfOlderThan * 1000
+    if (expiryDate > currentDate) {
+      return caches[key].value
+    }
+  }
+  // console.log("no cached result for", key)
+  return false
+}
+
+
+const getMsgCache = (id) => {
+  let data = sessionStorage.getItem(id)
+  if (data) {
+    try {
+      return JSON.parse(data)
+    } catch (n) {
+      sessionStorage.removeItem(id)
+      return false
+    }
+  } else {
+    return false
+  }
+}
+
+const setMsgCache = (id, data) => {
+  sessionStorage.setItem(id, JSON.stringify(data))
+}
+
+
+module.exports = {
+  caches,
+  cacheResult,
+  resultFromCache,
+  invalidateCacheResult,
+  getMsgCache,
+  setMsgCache
+}

src/core/platforms/common/pipelines.js

@@ -0,0 +1,15 @@
+
+const pipelines = {
+  threadPipes: new Set(),
+  messagePipes: new Set(),
+  thread: {
+    use: (func) => pipelines.threadPipes.add(func),
+    get: () => [...pipelines.threadPipes].map((p) => p.apply(p)),
+  },
+  message: {
+    use: (func) => pipelines.messagePipes.add(func),
+    get: () => [...pipelines.messagePipes].map((p) => p.apply(p)),
+  },
+}
+
+module.exports = pipelines

src/core/platforms/common/queues.js

@@ -0,0 +1,57 @@
+const { resultFromCache, cacheResult } = require("./cache.js")
+
+const workQueue = {}
+
+let parallelJobs = 0
+
+const maxParallelJobs = 5
+
+const enqueue = (kind, msgId, falseIfOlderThan, work, cb) => {
+  let key = `queue-${kind}-${msgId}`
+  let possibleResult = resultFromCache(kind, msgId, falseIfOlderThan)
+
+  if (possibleResult) {
+    cb(possibleResult)
+  } else {
+    if (workQueue.hasOwnProperty(key)) {
+      // is in the queue, should wait for result.
+      setTimeout(() => {
+        enqueue(kind, msgId, falseIfOlderThan, work, cb)
+      }, (falseIfOlderThan * 1000) / 2)
+    } else {
+      // not in the queue, should enqueue and block for  result.
+      workQueue[key] = { work, cb, kind, msgId }
+    }
+    processQueue()
+  }
+}
+
+const processQueue = () => {
+  if (Object.keys(workQueue).length > 0 && parallelJobs < maxParallelJobs) {
+    // entries in the queue.
+    let jobs = Object.keys(workQueue).slice(0, maxParallelJobs)
+    parallelJobs += jobs.length
+
+    jobs.forEach(async (job) => {
+      let work = workQueue[job]
+      try {
+        // console.log("starting job...", job)
+        let res = await work.work()
+        cacheResult(work.kind, work.msgId, res)
+        delete workQueue[job]
+        work.cb(res)
+        console.log("remaining jobs", Object.keys(workQueue).length)
+        setTimeout(processQueue, 10)
+      } catch (n) {
+        console.log("work", work)
+        console.error("error with work", n)
+      }
+      parallelJobs = parallelJobs >= 0 ? parallelJobs - 1 : 0
+    })
+  }
+}
+
+module.exports = {
+  enqueue,
+  processQueue
+}