git ssb

1
var nest = require('depnest')
2
var pull = require('pull-stream')
3
var MutantPullReduce = require('mutant-pull-reduce')
4
5
exports.needs = nest({
6
  'sbot.pull.backlinks': 'first'
7
})
8
9
exports.gives = nest('backlinks.obs.filter', true)
10
11
/**
12
 * sbot.obs.filter returns an observable list of messages that link
13
 * back to the message with the given message ID (@id). Only messages that
14
 * pass the filter are added to the list.
15
 *
16
 * When a message arrives, if a filter function is given in the options (opts.filter)
17
 * and passing it to the filter function does not result in it returning
18
 * 'true' the message is not added to the observable list.
19
 *
20
 * A 'sync' observable property is also added to the returned observable
21
 * which is 'true' when all previously seen messages are caught up with.
22
 *
23
 * Note: Unlike backlinks.obs.for this does not cache the observable for
24
 * callers that supply the same arguments.
25
 */
26
exports.create = function (api) {
27
  function pullFilterReduceObs (id, opts) {
28
    if (!id || typeof (id) !== 'string') {
29
      throw new Error('id must be a string.')
30
    }
31
32
    var sbotFilter = {
33
      $filter: {
34
        dest: id
35
      }
36
    }
37
38
    var msgBacklinks = api.sbot.pull.backlinks({
39
      query: [sbotFilter],
40
      index: 'DTA', // use asserted timestamps
41
      live: true
42
    })
43
44
    // If a filter function is supplied in the options, we use it to filter
45
    // the links stream, otherwise we use all the messages from the stream
46
    var filterFunction = opts && opts.filter ? opts.filter : () => true
47
48
    var filteredBacklinks = pull(
49
      msgBacklinks,
50
      pull.filter(filterFunction)
51
    )
52
53
    var backlinksObs = MutantPullReduce(filteredBacklinks, (state, msg) => {
54
      state.push(msg)
55
      return state
56
    }, {
57
      startValue: [],
58
      nextTick: true,
59
      sync: true
60
    })
61
62
    return backlinksObs
63
  }
64
65
  return nest({
66
    'backlinks.obs.filter': (id, opts) => pullFilterReduceObs(id, opts)
67
  })
68
}
69