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