git ssb

1
// read stream to get events
2
// for each item, check to see if already rendered root
3
// accept prioritized list (render these first)
4
5
var pull = require('pull-stream')
6
var nest = require('depnest')
7
var HLRU = require('hashlru')
8
9
exports.needs = nest({
10
  'backlinks.obs.for': 'first',
11
  'sbot.async.get': 'first',
12
  'message.sync.isBlocked': 'first',
13
  'message.sync.root': 'first',
14
  'message.sync.unbox': 'first',
15
  'feed.pull.withReplies': 'first'
16
})
17
18
exports.gives = nest('feed.pull.rollup', true)
19
20
exports.create = function (api) {
21
  // cache mostly just to avoid reading the same roots over and over again
22
  // not really big enough for multiple refresh cycles
23
  var cache = HLRU(100)
24
25
  return nest('feed.pull.rollup', function (rootFilter) {
26
    var seen = new Set()
27
    return pull(
28
      pull.map(msg => {
29
        if (msg.value) {
30
          var root = api.message.sync.root(msg)
31
          if (!root) {
32
            // already a root, pass thru!
33
            return msg
34
          } else {
35
            return root
36
          }
37
        }
38
      }),
39
40
      // UNIQUE
41
      pull.filter(idOrMsg => {
42
        if (idOrMsg) {
43
          if (idOrMsg.key) idOrMsg = idOrMsg.key
44
          if (typeof idOrMsg === 'string') {
45
            var key = idOrMsg
46
            if (!seen.has(key)) {
47
              seen.add(key)
48
              return true
49
            }
50
          }
51
        }
52
      }),
53
54
      // LOOKUP (if needed)
55
      pull.asyncMap((keyOrMsg, cb) => {
56
        if (keyOrMsg.value) {
57
          cb(null, keyOrMsg)
58
        } else {
59
          var key = keyOrMsg
60
          if (cache.has(key)) {
61
            cb(null, cache.get(key))
62
          } else {
63
            api.sbot.async.get(key, (_, value) => {
64
              var msg = {key, value}
65
              if (msg.value) {
66
                cache.set(key, msg)
67
              }
68
              cb(null, msg)
69
            })
70
          }
71
        }
72
      }),
73
74
      // UNBOX (if needed)
75
      pull.map(msg => {
76
        if (msg.value && typeof msg.value.content === 'string') {
77
          var unboxed = api.message.sync.unbox(msg)
78
          if (unboxed) return unboxed
79
        }
80
        return msg
81
      }),
82
83
      // FILTER
84
      pull.filter(msg => msg && msg.value),
85
      pull.filter(rootFilter || (() => true)),
86
      pull.filter(msg => !api.message.sync.isBlocked(msg)),
87
88
      // ADD REPLIES
89
      api.feed.pull.withReplies()
90
    )
91
  })
92
}
93