git ssb

1
var pull = require('pull-stream')
2
var pullPushable = require('pull-pushable')
3
var pullNext = require('pull-next')
4
var SortedArray = require('sorted-array-functions')
5
6
module.exports = FeedSummary
7
8
function FeedSummary (source, opts, cb) {
9
  var bumpFilter = opts && opts.bumpFilter
10
  var windowSize = opts && opts.windowSize || 1000
11
  var last = null
12
  var returned = false
13
  var done = false
14
  return pullNext(() => {
15
    if (!done) {
16
      var next = {reverse: true, limit: windowSize, live: false}
17
      if (last) {
18
        next.lt = last.timestamp || last.value.sequence
19
      }
20
      var pushable = pullPushable()
21
      pull(
22
        source(next),
23
        pull.collect((err, values) => {
24
          if (err) throw err
25
          if (!values.length) {
26
            done = true
27
          } else {
28
            var fromTime = last && last.timestamp || Date.now()
29
            groupMessages(values, fromTime, bumpFilter).forEach(v => pushable.push(v))
30
            last = values[values.length - 1]
31
          }
32
          pushable.end()
33
          if (!returned) cb && cb()
34
          returned = true
35
        })
36
      )
37
    }
38
    return pushable
39
  })
40
}
41
42
function groupMessages (messages, fromTime, bumpFilter) {
43
  var follows = {}
44
  var messageUpdates = {}
45
  reverseForEach(messages, function (msg) {
46
    if (!msg.value) return
47
    var c = msg.value.content
48
    if (c.type === 'contact') {
49
      updateContact(msg, follows)
50
    } else if (c.type === 'vote') {
51
      if (c.vote && c.vote.link) {
52
        // only show digs of posts added in the current window
53
        // and only for the main post
54
        const group = messageUpdates[c.vote.link]
55
        if (group) {
56
          if (c.vote.value > 0) {
57
            group.lastUpdateType = 'dig'
58
            group.digs.add(msg.value.author)
59
            // only bump when filter passes
60
            if (!bumpFilter || bumpFilter(msg, group)) {
61
              group.updated = msg.timestamp
62
            }
63
          } else {
64
            group.digs.delete(msg.value.author)
65
            if (group.lastUpdateType === 'dig' && !group.digs.size && !group.replies.length) {
66
              group.lastUpdateType = 'reply'
67
            }
68
          }
69
        }
70
      }
71
    } else {
72
      if (c.root) {
73
        const group = ensureMessage(c.root, messageUpdates)
74
        group.fromTime = fromTime
75
        group.lastUpdateType = 'reply'
76
        group.repliesFrom.add(msg.value.author)
77
        //SortedArray.add(group.replies, msg, compareUserTimestamp)
78
        group.replies.push(msg)
79
        group.channel = group.channel || msg.value.content.channel
80
81
        // only bump when filter passes
82
        if (!bumpFilter || bumpFilter(msg, group)) {
83
          group.updated = msg.timestamp
84
        }
85
      } else {
86
        const group = ensureMessage(msg.key, messageUpdates)
87
        group.fromTime = fromTime
88
        group.lastUpdateType = 'post'
89
        group.updated = msg.timestamp
90
        group.author = msg.value.author
91
        group.channel = msg.value.content.channel
92
        group.message = msg
93
      }
94
    }
95
  })
96
97
  var result = []
98
  Object.keys(follows).forEach((key) => {
99
    if (follows[key].updated) {
100
      SortedArray.add(result, follows[key], compareUpdated)
101
    }
102
  })
103
  Object.keys(messageUpdates).forEach((key) => {
104
    if (messageUpdates[key].updated) {
105
      SortedArray.add(result, messageUpdates[key], compareUpdated)
106
    }
107
  })
108
  return result
109
}
110
111
function compareUpdated (a, b) {
112
  return b.updated - a.updated
113
}
114
115
function reverseForEach (items, fn) {
116
  var i = items.length - 1
117
  var start = Date.now()
118
  nextBatch()
119
120
  function nextBatch () {
121
    while (i >= 0 && Date.now() - start < 10) {
122
      fn(items[i], i)
123
      i -= 1
124
    }
125
126
    if (i > 0) {
127
      setImmediate(nextBatch)
128
    }
129
  }
130
}
131
132
function updateContact (msg, groups) {
133
  var c = msg.value.content
134
  var id = msg.value.author
135
  var group = groups[id]
136
  if (c.contact) {
137
    if (c.following) {
138
      if (!group) {
139
        group = groups[id] = {
140
          type: 'follow',
141
          lastUpdateType: null,
142
          contacts: new Set(),
143
          updated: 0,
144
          author: id,
145
          id: id
146
        }
147
      }
148
      group.contacts.add(c.contact)
149
      group.updated = msg.timestamp
150
    } else {
151
      if (group) {
152
        group.contacts.delete(c.contact)
153
        if (!group.contacts.size) {
154
          delete groups[id]
155
        }
156
      }
157
    }
158
  }
159
}
160
161
function ensureMessage (id, groups) {
162
  var group = groups[id]
163
  if (!group) {
164
    group = groups[id] = {
165
      type: 'message',
166
      repliesFrom: new Set(),
167
      replies: [],
168
      message: null,
169
      messageId: id,
170
      digs: new Set(),
171
      updated: 0
172
    }
173
  }
174
  return group
175
}
176