git ssb

1
#!/usr/bin/env node
2
3
var fs = require('fs')
4
var path = require('path')
5
var URL = require('url')
6
var http = require('http')
7
var https = require('https')
8
var crypto = require('crypto')
9
var readline = require('readline')
10
var os = require('os')
11
12
var ssbClient = require('ssb-client')
13
var pull = require('pull-stream')
14
15
var pkg = require('./package')
16
17
var userAgentBase = pkg.name + '/' + pkg.version
18
var userAgentContact
19
var userAgentBot = false
20
21
function estimateMessageSize(content) {
22
  var draftMsg = {
23
    key: '%0000000000000000000000000000000000000000000=.sha256',
24
    value: {
25
      previous: '%0000000000000000000000000000000000000000000=.sha256',
26
      author: '@0000000000000000000000000000000000000000000=.ed25519',
27
      sequence: 100000,
28
      timestamp: 1000000000000.0001,
29
      hash: 'sha256',
30
      content: content,
31
      signature: '00000000000000000000000000000000000000000000000000000000000000000000000000000000000000==.sig.ed25519'
32
    }
33
  }
34
  return JSON.stringify(draftMsg, null, 2).length
35
}
36
37
function getJson(url, cb) {
38
  var opts = URL.parse(url)
39
  opts.headers = {
40
    'User-Agent': userAgentBase
41
      + (userAgentContact ? ' (' + userAgentContact + ')' : '')
42
      + (userAgentBot ? ' bot' : '')
43
  }
44
  var h = opts.protocol === 'https:' ? https : http
45
  h.get(opts, function (res) {
46
    if (res.statusCode !== 200) {
47
      console.error(res.headers, url)
48
      return cb(new Error('HTTP ' + res.statusCode + ' ' + res.statusMessage))
49
    }
50
    var bufs = []
51
    res.on('data', function (buf) {
52
      bufs.push(buf)
53
    })
54
    res.on('end', function () {
55
      res.removeListener('error', cb)
56
      var buf = Buffer.concat(bufs)
57
      bufs = null
58
      var data
59
      try {
60
        data = JSON.parse(buf.toString('utf8'))
61
      } catch(e) {
62
        return cb(e)
63
      }
64
      cb(null, data)
65
    })
66
    res.on('error', cb)
67
  })
68
}
69
70
function publishDrafts(sbot, drafts, cb) {
71
  var draftIdIndex = {}
72
  drafts.forEach(function (draft, i) {
73
    draftIdIndex[draft.draftId] = i
74
  })
75
  var ids = []
76
77
  function replaceDraftIds(obj) {
78
    if (typeof obj === 'string') {
79
      var i = draftIdIndex[obj]
80
      if (typeof i === 'number') {
81
        var id = ids[i]
82
        if (!id) throw new ReferenceError('draft referernces unknown message')
83
        return id
84
      }
85
    } else if (Array.isArray(obj)) {
86
      return obj.map(replaceDraftIds)
87
    } else if (obj !== null && typeof obj === 'object') {
88
      var o = {}
89
      for (var k in obj) o[k] = replaceDraftIds(obj[k])
90
      return o
91
    }
92
    return obj
93
  }
94
95
  pull(
96
    pull.values(drafts),
97
    pull.asyncMap(function (draft, cb) {
98
      var content = replaceDraftIds(draft.content)
99
      sbot.publish(content, function (err, msg) {
100
        if (err) return cb(err)
101
        ids.push(msg.key)
102
        cb(null, msg)
103
      })
104
    }),
105
    pull.collect(cb)
106
  )
107
}
108
109
var args = process.argv.slice(2)
110
var yes = false
111
var dry = false
112
var help = false
113
var urls = []
114
args.forEach(function (arg) {
115
  if (arg[0] === '-') switch (arg) {
116
    case '-n': return dry = true
117
    case '-y': return yes = true
118
    case '-h': return help = true
119
    default: throw 'Unknown argument: ' + arg
120
  } else urls.push(arg)
121
})
122
123
if (help) {
124
  process.stdout.write(fs.readFileSync(path.join(__dirname, 'usage.txt')))
125
  process.exit(0)
126
}
127
128
ssbClient(function (err, sbot, config) {
129
  if (err) throw err
130
  var conf = config.wikimedia || {}
131
  userAgentContact = conf.contact
132
  userAgentBot = conf.bot
133
134
  if (urls.length === 0) {
135
    var pagesFile = path.join(config.path, 'wikimedia-pages.txt')
136
    var pagesData = fs.readFileSync(pagesFile, 'utf8')
137
    urls = pagesData.split('\n').filter(RegExp.prototype.test.bind(/[^#]/))
138
    if (!urls.length) {
139
      console.log('No pages to sync.')
140
      return sbot.close()
141
    }
142
  }
143
144
  var pagesInfo = urls.map(function (page) {
145
    var m = /^(https?:\/\/.*?)(\/wiki)?\/(.*)$/.exec(page)
146
    if (!m) throw 'Unable to parse page URL ' + page
147
    return {
148
      site: m[1] + '/',
149
      api: m[1] + (m[2] ? '/w' : '/wiki') + '/api.php',
150
      title: m[3]
151
    }
152
  })
153
  var pagesInfoByApi = {}
154
  pagesInfo.forEach(function (pageInfo) {
155
    var infos = pagesInfoByApi[pageInfo.api] || (pagesInfoByApi[pageInfo.api] = [])
156
    infos.push(pageInfo)
157
  })
158
  console.log('Normalizing titles...')
159
  var waiting = 0
160
  for (var api in pagesInfoByApi) (function (api) {
161
    var pagesInfoForApi = pagesInfoByApi[api]
162
    var pagesInfoForApiByTitle = {}
163
    var titles = pagesInfoForApi.map(function (info) {
164
      pagesInfoForApiByTitle[info.title] = info
165
      return info.title
166
    })
167
    var url = api + '?format=json&action=query' +
168
      '&titles=' + encodeURIComponent('\x1f' + titles.join('\x1f')) +
169
      '&' // trailing & needed for some reason
170
    waiting++
171
    getJson(url, function (err, data) {
172
      if (err) throw err
173
      if (data.warnings) console.trace('Warnings:', data.warnings)
174
      if (data.query.normalized) data.query.normalized.forEach(function (norm) {
175
        var info = pagesInfoForApiByTitle[norm.from]
176
        if (!info) {
177
          console.error(JSON.stringify({titles: titles, response: data}, 0, 2))
178
          throw new Error('Unexpected title in server response')
179
        }
180
        // console.log('Normalized title', norm.from, norm.to)
181
        info.title = norm.to
182
      })
183
      if (!--waiting) next()
184
    })
185
  }(api))
186
187
  function next() {
188
    console.log('Getting revisions...')
189
    var userHashes = {}
190
    pull(
191
      pull.values(pagesInfo),
192
      pull.asyncMap(function (pageInfo, cb) {
193
        // Calculate blob id for page URL + title, for linking
194
        pull(
195
          pull.once(pageInfo.site + '\t' + pageInfo.title),
196
          sbot.blobs.add(function (err, hash) {
197
            pageInfo.hash = hash
198
            cb(null, pageInfo)
199
          })
200
        )
201
      }),
202
      pull.asyncMap(function (pageInfo, cb) {
203
        // Get previous messages for this page.
204
        // Simple solution: find the revision with latest timestamp.
205
        var maxRevTs = ''
206
        var maxRevMsgId
207
        pull(
208
          sbot.links({
209
            dest: pageInfo.hash,
210
            rel: 'pageId',
211
            values: true,
212
            meta: false
213
          }),
214
          pull.filter(function (msg) {
215
            var c = msg && msg.value && msg.value.content
216
            return c
217
              && c.type === 'wikimedia/revisions'
218
              && c.site === pageInfo.site
219
              && c.title === pageInfo.title
220
          }),
221
          pull.drain(function (msg) {
222
            var c = msg && msg.value && msg.value.content
223
            var revs = Array.isArray(c.revisions) && c.revisions
224
            if (revs) revs.forEach(function (rev) {
225
              if (rev && rev.timestamp > maxRevTs) {
226
                maxRevTs = rev.timestamp
227
                maxRevMsgId == msg.key
228
              }
229
            })
230
          }, function (err) {
231
            if (err) return cb(err)
232
            pageInfo.latestMsgId = maxRevMsgId
233
            pageInfo.latestRevTs = maxRevTs
234
            cb(null, pageInfo)
235
          })
236
        )
237
      }),
238
      pull.map(function (pageInfo) {
239
        // Get new revisions.
240
        var rvcontinue, rvdone
241
        var rvstart = pageInfo.latestRevTs
242
        var prevId = pageInfo.latestMsgId
243
        var aborted
244
        var revisions = pull(
245
          function (abort, cb) {
246
            if (aborted = abort) return cb(abort)
247
            if (rvdone) return cb(true)
248
249
            console.log('Getting revisions for', pageInfo.title + '...',
250
              rvstart || '', rvcontinue || '')
251
            var url = api + '?format=json&action=query&prop=revisions&rvslots=*'
252
              + '&titles=' + encodeURIComponent(pageInfo.title)
253
              + '&rvprop=ids|timestamp|comment|user|sha1|size|slotsha1|slotsize|content|roles|flags|tags'
254
              + '&rvdir=newer'
255
              + (rvcontinue ? '&rvcontinue=' + rvcontinue : '')
256
              + (rvstart ? '&rvstart=' + rvstart : '')
257
              + '&rvlimit=50'
258
            getJson(url, function (err, data) {
259
              if (aborted) return err && console.trace(err)
260
              if (err) return cb(err)
261
262
              var warnings = data.warnings
263
              if (warnings) {
264
                if (warnings.main) {
265
                  if (warnings.main['*'] === 'Unrecognized parameter: rvslots.') {
266
                    delete warnings.main['*']
267
                    if (Object.keys(warnings.main).length === 0) {
268
                      delete warnings.main
269
                    }
270
                  }
271
                }
272
                if (warnings.revisions) {
273
                  if (warnings.revisions['*'] === 'Unrecognized values for parameter "rvprop": slotsha1, slotsize, roles.') {
274
                    delete warnings.revisions['*']
275
                    if (Object.keys(warnings.revisions).length === 0) {
276
                      delete warnings.revisions
277
                    }
278
                  }
279
                }
280
                if (Object.keys(warnings).length > 0) {
281
                  console.trace('Warnings:', warnings)
282
                }
283
              }
284
285
              rvcontinue = data.continue && data.continue.rvcontinue
286
              if (!rvcontinue) rvdone = true
287
              var page
288
              if (data.query) for (var pageid in data.query.pages) {
289
                page = data.query.pages[pageid]
290
                if (page.title === pageInfo.title) break
291
                else page = null
292
              }
293
              if (!page) {
294
                console.trace(data.query.pages, pageInfo)
295
                return cb(new Error('Unable to find page'))
296
              }
297
              var revs = page.revisions || []
298
              console.log('Got ' + revs.length + ' revisions')
299
              cb(null, revs)
300
            })
301
          },
302
          pull.flatten(),
303
304
          pull.through(function (rev) {
305
            if (!rev.slots) {
306
               // old API does not use slots.
307
               // Transform result to be forward-compatible.
308
              rev.slots = {
309
                main: {
310
                  size: rev.size,
311
                  sha1: rev.sha1,
312
                  contentmodel: rev.contentmodel,
313
                  contentformat: rev.contentformat,
314
                  '*': rev['*']
315
                }
316
              }
317
              delete rev.contentmodel
318
              delete rev.contentformat
319
              delete rev['*']
320
            }
321
            // duplicate values supplied in new API in slotsize and slotsha1
322
            delete rev.sha1
323
            delete rev.size
324
          }),
325
326
          pull.asyncMap(function (rev, cb) {
327
            // Calculate blob id for user page URL + title, for linking
328
            var hash = userHashes[rev.user]
329
            if (hash) {
330
              rev.userId = hash
331
              return cb(null, rev)
332
            }
333
            pull(
334
              pull.once(pageInfo.site + '\tUser:' + rev.user),
335
              sbot.blobs.add(function (err, hash) {
336
                rev.userId = userHashes[rev.user] = hash
337
                cb(null, rev)
338
              })
339
            )
340
          }),
341
342
          pull.asyncMap(function (rev, cb) {
343
            var waiting = 0
344
            for (var slot in rev.slots) (function (slot) {
345
              waiting++
346
              var slotInfo = rev.slots[slot]
347
              var content = slotInfo['*']
348
              if (!content) {
349
                console.trace(slotInfo)
350
                return cb(new Error('Missing content'))
351
              }
352
              var sha1 = crypto.createHash('sha1').update(content).digest('hex')
353
              if (sha1 !== slotInfo.sha1) {
354
                console.trace(slotInfo, sha1)
355
                return cb(new Error('Mismatched content sha1'))
356
              }
357
              pull(
358
                pull.once(content),
359
                sbot.blobs.add(function (err, hash) {
360
                  if (err) return cb(err)
361
                  slotInfo.link = hash
362
                  delete slotInfo['*']
363
                  if (!--waiting) cb(null, rev)
364
                })
365
              )
366
            }(slot))
367
          })
368
        )
369
370
        var queuedRevisions = []
371
        var ended
372
        function cbDraft(content, cb) {
373
          if (!content.revisions.length) {
374
            console.log('No revisions for', pageInfo.title)
375
            return cb(true)
376
          }
377
          console.log('Prepared a message',
378
            'with', content.revisions.length, 'revisions',
379
            'for', pageInfo.title)
380
          prevId = '%' + crypto.createHash('sha256').update(JSON.stringify(content)).digest('base64') + '.draft6'
381
          cb(null, {
382
            draftId: prevId,
383
            content: content
384
          })
385
        }
386
387
        return function (abort, cb) {
388
          if (abort) return revisions(abort, cb)
389
          if (ended) return cb(true)
390
          var content = {
391
            type: 'wikimedia/revisions',
392
            site: pageInfo.site,
393
            title: pageInfo.title,
394
            pageId: pageInfo.hash,
395
            parents: prevId ? [prevId] : undefined,
396
            revisions: queuedRevisions.splice(0)
397
          }
398
          revisions(null, function next(end, revision) {
399
            if (ended = end) return cbDraft(content, cb)
400
            content.revisions.push(revision)
401
            if (estimateMessageSize(content) > 8192) {
402
              queuedRevisions.push(content.revisions.pop())
403
              // console.log('filled msg for ', pageInfo.title, ' with ', content.revisions.length, 'revisions')
404
              return cbDraft(content, cb)
405
            }
406
            revisions(null, next)
407
          })
408
        }
409
      }),
410
      pull.flatten(),
411
412
      pull.collect(function (err, drafts) {
413
        if (err) throw err
414
        if (dry) {
415
          console.log(JSON.stringify(drafts, 0, 2))
416
          return sbot.close()
417
        }
418
        if (!drafts.length) {
419
          console.log('No messages to publish.')
420
          return sbot.close()
421
        }
422
        if (yes) return confirmed(true)
423
        var rl = readline.createInterface({
424
          input: process.stdin,
425
          output: process.stdout
426
        })
427
        rl.question('Publish ' + drafts.length + ' messages? [Y/n] ', function (answer) {
428
          rl.close()
429
          confirmed(!/^n/i.test(answer))
430
        })
431
        function confirmed(yes) {
432
          if (!yes) return sbot.close()
433
          publishDrafts(sbot, drafts, function (err, msgs) {
434
            if (err) throw err
435
            console.log('Published:\n' + msgs.map(function (msg) {
436
              return msg.key
437
            }.join('\n')))
438
            sbot.close()
439
          })
440
        }
441
      })
442
    )
443
  }
444
})
445