git ssb

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