git ssb

1
var pull = require('pull-stream')
2
var paramap = require('pull-paramap')
3
var asyncMemo = require('asyncmemo')
4
var issueSchemas = require('./lib/schemas')
5
var multicb = require('multicb')
6
7
function Cache(fn, ssb) {
8
  return asyncMemo(fn)
9
  return function (key, cb) { ac.get(key, cb) }
10
}
11
12
function truncate(str, len) {
13
  return str.length > len ? str.substr(0, len) + '...' : str
14
}
15
16
exports.name = 'issues'
17
18
exports.manifest = {
19
  get: 'async',
20
  list: 'source',
21
  new: 'async',
22
  edit: 'async',
23
  close: 'async',
24
  reopen: 'async',
25
  getMention: 'sync',
26
  isStatusChanged: 'sync'
27
}
28
29
exports.schemas = issueSchemas
30
31
function isStatusChanged(msg, issue) {
32
  var mention = getMention(msg, issue)
33
  return mention ? mention.open : null
34
}
35
36
function getMention(msg, issue) {
37
  var c = msg.value.content
38
  if (msg.key == issue.id || c.issue == issue.id || c.link == issue.id)
39
    if (c.open != null)
40
      return c
41
  if (c.issues) {
42
    var mention
43
    for (var i = 0; i < c.issues.length; i++) {
44
      mention = getMention({value: {
45
        timestamp: msg.value.timestamp,
46
        author: msg.value.author,
47
        content: c.issues[i]
48
      }}, issue)
49
      if (mention)
50
        return mention
51
    }
52
  }
53
}
54
55
function isMsgIdRange(opts) {
56
  return (
57
    (opts.gt && opts.gt[0] === '%') ||
58
    (opts.lt && opts.lt[0] === '%') ||
59
    (opts.gte && opts.gte[0] === '%') ||
60
    (opts.lte && opts.lte[0] === '%')
61
  )
62
}
63
64
function passthrough(read) {
65
  return read
66
}
67
68
function optsToRangeFilter(opts) {
69
  if (!opts || (
70
    opts.gt == null ||
71
    opts.lt == null ||
72
    opts.gte == null ||
73
    opts.lte == null
74
  )) return passthrough
75
  return pull.filter(function (msg) {
76
    return (opts.gt == null || msg.timestamp > opts.gt)
77
        && (opts.lt == null || msg.timestamp < opts.lt)
78
        && (opts.gte == null || msg.timestamp >= opts.gte)
79
        && (opts.lte == null || msg.timestamp <= opts.lte)
80
  })
81
}
82
83
function optsToRange(opts) {
84
  var range = {}
85
  var defined = false
86
  if (opts.gt != null) defined = true, range.$gt = opts.gt
87
  if (opts.lt != null) defined = true, range.$lt = opts.lt
88
  if (opts.gte != null) defined = true, range.$gte = opts.gt
89
  if (opts.lte != null) defined = true, range.$lte = opts.lt
90
  return defined ? range : undefined
91
}
92
93
exports.init = function (ssb) {
94
95
  var ssbGet = asyncMemo(ssb.get)
96
  var liveStreams = []
97
98
  var getIssue = asyncMemo(function (id, cb) {
99
    var issue = {
100
      labels: []
101
    }
102
    var labelCounts = {}
103
    var issueMsg
104
105
    function addLabel(id) {
106
      var count = labelCounts[id] || 0
107
      labelCounts[id] = ++count
108
      if (count === 1) {
109
        var i = issue.labels.indexOf(id)
110
        if (i === -1) issue.labels.push(id)
111
      }
112
    }
113
114
    function removeLabel(id) {
115
      var count = labelCounts[id] || 0
116
      labelCounts[id] = --count
117
      if (count === 0) {
118
        var i = issue.labels.indexOf(id)
119
        if (i > -1) issue.labels.splice(i, 1)
120
      }
121
    }
122
123
    ssbGet(id, function (err, msg) {
124
      msg = {key: id, value: msg}
125
      if (err) return cb(err)
126
      issueMsg = msg
127
      issue.id = msg.key
128
      issue.msg = msg
129
      issue.author = msg.value.author
130
      var c = msg.value.content
131
      issue.project = c.project
132
      issue.text = c.text || c.title || JSON.stringify(msg, null, 2)
133
      issue.created_at = issue.updated_at = msg.value.timestamp
134
      if (Array.isArray(c.labels)) c.labels.forEach(addLabel)
135
      if (c.project)
136
        ssbGet(c.project, gotProjectMsg)
137
      else
138
        getLinks()
139
    })
140
141
    function gotProjectMsg(err, msg) {
142
      if (err) return cb(err)
143
      issue.projectAuthor = msg.author
144
      getLinks()
145
    }
146
147
    function getLinks() {
148
      var now = Date.now()
149
      // compute the result from the past data
150
      pull(
151
        ssb.links({dest: id, reverse: true, values: true,
152
          old: true, live: false, sync: false}),
153
        pull.drain(onOldMsg, onOldEnd)
154
      )
155
      // keep the results up-to-date in the future
156
      var read = ssb.links({dest: id, values: true,
157
        old: false, live: true, sync: false})
158
      liveStreams.push(read)
159
      pull(
160
        read,
161
        pull.drain(onNewMsg, onNewEnd)
162
      )
163
    }
164
165
    function onOldMsg(msg) {
166
      if (!msg.value)
167
        return
168
      var c = msg.value.content
169
170
      // handle updates to issue
171
      if (msg.key == id || c.issue == id || c.link == id) {
172
        if (c.open != null && issue.open == null)
173
          issue.open = c.open
174
        if (c.title != null && issue.title == null)
175
          issue.title = c.title
176
        if (c.labels) {
177
          if (Array.isArray(c.labels.add))
178
            c.labels.add.forEach(addLabel)
179
          if (Array.isArray(c.labels.remove))
180
            c.labels.remove.forEach(removeLabel)
181
        }
182
        if (msg.value.timestamp > issue.updated_at)
183
          issue.updated_at = msg.value.timestamp
184
      }
185
186
      // handle updates via mention
187
      if (c.issues) {
188
        for (var i = 0; i < c.issues.length; i++) {
189
          if (c.type === 'issue-label') {
190
            if (c.issues[i] === id) addLabel(msg.key)
191
          } else {
192
            onOldMsg({value: {
193
              timestamp: msg.value.timestamp,
194
              author: msg.value.author,
195
              content: c.issues[i]
196
            }})
197
          }
198
        }
199
      }
200
201
      checkReady()
202
    }
203
204
    function onNewMsg(msg) {
205
      if (!msg.value)
206
        return
207
      var c = msg.value.content
208
209
      // handle updates to issue
210
      if (msg.key == id || c.issue == id || c.link == id) {
211
        if (c.open != null)
212
          issue.open = c.open
213
        if (c.title != null)
214
          issue.title = c.title
215
        if (msg.value.timestamp > issue.updated_at)
216
          issue.updated_at = msg.value.timestamp
217
        if (c.labels) {
218
          if (Array.isArray(c.labels.add))
219
            c.labels.add.forEach(addLabel)
220
          if (Array.isArray(c.labels.remove))
221
            c.labels.remove.forEach(removeLabel)
222
        }
223
      }
224
225
      // handle updates via mention
226
      if (c.issues) {
227
        for (var i = 0; i < c.issues.length; i++)
228
          onNewMsg({value: {
229
            timestamp: msg.value.timestamp,
230
            author: msg.value.author,
231
            content: c.issues[i]
232
          }})
233
      }
234
    }
235
236
    function checkReady() {
237
      // call back once all the issue properties are set
238
      if (issue.open != null && issue.title != null) {
239
        var _cb = cb
240
        delete cb
241
        _cb(null, issue)
242
      }
243
    }
244
245
    function onOldEnd(err) {
246
      if (err) {
247
        if (cb) cb(err)
248
        else console.error(err)
249
        return
250
      }
251
      // process the root message last
252
      onOldMsg(issueMsg)
253
      // if callback hasn't been called yet, the issue is missing a field
254
      if (cb) {
255
        if (issue.open == null)
256
          issue.open = true
257
        if (issue.title == null)
258
          issue.title = truncate(issue.text.split('\n')[0], 250) || issue.id
259
        checkReady()
260
      }
261
    }
262
263
    function onNewEnd(err) {
264
      if (err) {
265
        if (cb) cb(err)
266
        else console.error(err)
267
      }
268
    }
269
  })
270
271
  function deinit(cb) {
272
    var done = multicb()
273
    // cancel all live streams
274
    liveStreams.forEach(function (read) {
275
      read(true, done())
276
    })
277
    done(cb)
278
  }
279
280
  function listIssues(opts) {
281
    opts.type = 'issue'
282
    return pull(
283
      opts.project && !isMsgIdRange(opts) ? (
284
        ssb.backlinks ? ssb.backlinks.read({
285
          reverse: opts.reverse,
286
          live: opts.live,
287
          query: [{$filter: {
288
            value: {content: {type: opts.type}},
289
            dest: opts.project,
290
            rts: optsToRange(opts)
291
          }}]
292
        }) : pull(
293
          ssb.links({
294
            reverse: opts.reverse,
295
            live: opts.live,
296
            dest: opts.project,
297
            values: true,
298
            rel: 'project'
299
          }),
300
          pull.filter(function (msg) {
301
            return msg.value.content.type === opts.type
302
          }),
303
          optsToRangeFilter(opts)
304
        )
305
      ) : ssb.messagesByType(opts),
306
      pull.unique('key'),
307
      pull.filter(function (msg) {
308
        return (!opts.project || opts.project == msg.value.content.project)
309
          && (!opts.author || opts.author == msg.value.author)
310
      }),
311
      paramap(function (msg, cb) {
312
        getIssue(msg.key, cb)
313
      }, 8),
314
      pull.filter(opts.open != null && function (pr) {
315
        return pr.open == opts.open
316
      })
317
    )
318
  }
319
320
  function editIssue(id, opts, cb) {
321
    var msg
322
    try { ssb.publish(issueSchemas.edit(id, opts), cb) }
323
    catch(e) { return cb(e) }
324
  }
325
326
  function closeIssue(id, cb) {
327
    var msg
328
    try { ssb.publish(issueSchemas.close(id), cb) }
329
    catch(e) { return cb(e) }
330
  }
331
332
  function reopenIssue(id, cb) {
333
    var msg
334
    try { msg = issueSchemas.reopen(id) }
335
    catch(e) { return cb(e) }
336
    ssb.publish(msg, cb)
337
  }
338
339
  function newIssue(opts, cb) {
340
    var msg
341
    try { msg = issueSchemas.new(opts.project, opts.title, opts.text) }
342
    catch(e) { return cb(e) }
343
    ssb.publish(msg, function (err, msg) {
344
      if (err) return cb(err)
345
      getIssue(msg.key, cb)
346
    })
347
  }
348
349
  return {
350
    deinit: deinit,
351
    get: getIssue,
352
    list: listIssues,
353
    new: newIssue,
354
    edit: editIssue,
355
    close: closeIssue,
356
    reopen: reopenIssue,
357
    getMention: getMention,
358
    isStatusChanged: isStatusChanged
359
  }
360
}
361