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
6
function Cache(fn, ssb) {
7
  return asyncMemo(fn)
8
  return function (key, cb) { ac.get(key, cb) }
9
}
10
11
function isUpdateValid(issue, msg) {
12
  return msg.value.author == issue.author
13
      && msg.value.author == issue.projectAuthor
14
}
15
16
exports.name = 'issues'
17
18
exports.manifest = {
19
  get: 'async',
20
  createFeedStream: 'source',
21
  new: 'async',
22
  edit: 'async',
23
  close: 'async',
24
  reopen: 'async'
25
}
26
27
exports.schemas = issueSchemas
28
29
exports.init = function (ssb) {
30
31
  var ssbGet = asyncMemo(ssb.get)
32
33
  var getIssue = asyncMemo(function (id, cb) {
34
    var issue = {}
35
    var issueMsg
36
37
    if (id && id.value && id.key) {
38
      var msg = id
39
      id = id.key
40
      gotIssueMsg(null, msg)
41
    } else {
42
      ssbGet(id, gotIssueMsg)
43
    }
44
45
    function gotIssueMsg(err, msg) {
46
      if (err) return cb(err)
47
      issueMsg = msg
48
      issue.id = msg.key
49
      issue.author = msg.value.author
50
      var c = msg.value.content
51
      issue.project = c.project
52
      issue.text = c.text
53
      issue.created_at = issue.updated_at = msg.value.timestamp
54
      if (c.project)
55
        ssbGet(c.project, gotProjectMsg)
56
      else
57
        getLinks()
58
    }
59
60
    function gotProjectMsg(err, msg) {
61
      if (err) return cb(err)
62
      issue.projectAuthor = msg.author
63
      getLinks()
64
    }
65
66
    function getLinks() {
67
      var now = Date.now()
68
      // compute the result from the past data
69
      pull(
70
        ssb.links({dest: id, lt: now, reverse: true, values: true}),
71
        pull.drain(onOldMsg, onOldEnd)
72
      )
73
      // keep the results up-to-date in the future
74
      pull(
75
        ssb.links({dest: id, gte: now, values: true, live: true}),
76
        pull.drain(onNewMsg, onNewEnd)
77
      )
78
    }
79
80
    function onOldMsg(msg) {
81
      if (!isUpdateValid(issue, msg))
82
        return
83
      var c = msg.value.content
84
85
      // handle updates to issue
86
      if (msg.key == id || c.issue == id || c.link == id) {
87
        if (c.open != null && issue.open == null)
88
          issue.open = c.open
89
        if (c.title != null && issue.title == null)
90
          issue.title = c.title
91
        if (msg.value.timestamp > issue.updated_at)
92
          issue.updated_at = msg.value.timestamp
93
      }
94
95
      // handle updates via mention
96
      if (c.issues) {
97
        for (var i = 0; i < c.issues.length; i++)
98
          onOldMsg({value: {
99
            timestamp: msg.value.timestamp,
100
            author: msg.value.author,
101
            content: c.issues[i]
102
          }})
103
      }
104
105
      checkReady()
106
    }
107
108
    function onNewMsg(msg) {
109
      if (!isUpdateValid(issue, msg))
110
        return
111
      var c = msg.value.content
112
113
      // handle updates to issue
114
      if (msg.key == id || c.issue == id || c.link == id) {
115
        if (c.open != null)
116
          issue.open = c.open
117
        if (c.title != null)
118
          issue.title = c.title
119
        if (msg.value.timestamp > issue.updated_at)
120
          issue.updated_at = msg.value.timestamp
121
      }
122
123
      // handle updates via mention
124
      if (c.issues) {
125
        for (var i = 0; i < c.issues.length; i++)
126
          onNewMsg({value: {
127
            timestamp: msg.value.timestamp,
128
            author: msg.value.author,
129
            content: c.issues[i]
130
          }})
131
      }
132
    }
133
134
    function checkReady() {
135
      // call back once all the issue properties are set
136
      if (issue.open != null && issue.title != null) {
137
        var _cb = cb
138
        delete cb
139
        _cb(null, issue)
140
      }
141
    }
142
143
    function onOldEnd(err) {
144
      if (err) {
145
        if (cb) cb(err)
146
        else console.error(err)
147
        return
148
      }
149
      // process the root message last
150
      onOldMsg(issueMsg)
151
      // if callback hasn't been called yet, the issue is missing a field
152
      if (cb) {
153
        if (issue.open == null)
154
          issue.open = true
155
        if (issue.title == null)
156
          issue.title = issue.id
157
        checkReady()
158
      }
159
    }
160
161
    function onNewEnd(err) {
162
      if (err) {
163
        if (cb) cb(err)
164
        else console.error(err)
165
      }
166
    }
167
  })
168
169
  function createFeedStream(opts) {
170
    opts.type = 'issue'
171
    delete opts.limit
172
    return pull(
173
      // TODO: use links2 for this
174
      ssb.messagesByType(opts),
175
      pull.filter(function (msg) {
176
        return (!opts.project || opts.project == msg.value.content.project)
177
          && (!opts.author || opts.author == msg.value.author)
178
      }),
179
      paramap(getIssue, 8)
180
    )
181
  }
182
183
  function editIssue(id, opts, cb) {
184
    ssb.publish(issueSchemas.edit(id, opts), cb)
185
  }
186
187
  function closeIssue(id, cb) {
188
    ssb.publish(issueSchemas.close(id), cb)
189
  }
190
191
  function reopenIssue(id, cb) {
192
    ssb.publish(issueSchemas.reopen(id), cb)
193
  }
194
195
  function newIssue(opts, cb) {
196
    var msg = issueSchemas.new(opts.project, opts.title, opts.text)
197
    ssb.publish(msg, function (err, msg) {
198
      if (err) return cb(err)
199
      getIssue(msg, cb)
200
    })
201
  }
202
203
  return {
204
    get: getIssue,
205
    createFeedStream: createFeedStream,
206
    new: newIssue,
207
    edit: editIssue,
208
    close: closeIssue,
209
    reopen: reopenIssue
210
  }
211
}
212