git ssb

1
var pull = require('pull-stream')
2
var cat = require('pull-cat')
3
var cache = require('pull-cache')
4
var buffered = require('pull-buffered')
5
var Repo = require('pull-git-repo')
6
var pack = require('pull-git-pack')
7
var pktLine = require('./lib/pkt-line')
8
var indexPack = require('pull-git-pack/lib/index-pack')
9
var util = require('./lib/util')
10
var multicb = require('multicb')
11
var ProgressBar = require('progress')
12
13
function handleOption(options, name, value) {
14
  switch (name) {
15
    case 'verbosity':
16
      options.verbosity = +value || 0
17
      return true
18
    case 'progress':
19
      options.progress = !!value && value !== 'false'
20
      return true
21
    default:
22
      console.error('unknown option', name + ': ' + value)
23
      return false
24
  }
25
}
26
27
function capabilitiesSource() {
28
  return pull.once([
29
    'option',
30
    'connect',
31
  ].join('\n') + '\n\n')
32
}
33
34
function optionSource(cmd, options) {
35
  var args = util.split2(cmd)
36
  var msg = handleOption(options, args[0], args[1])
37
  msg = (msg === true) ? 'ok'
38
      : (msg === false) ? 'unsupported'
39
      : 'error ' + msg
40
  return pull.once(msg + '\n')
41
}
42
43
// transform ref objects into lines
44
function listRefs(read) {
45
  var ended
46
  return function (abort, cb) {
47
    if (ended) return cb(ended)
48
    read(abort, function (end, ref) {
49
      ended = end
50
      if (end === true) cb(null, '\n')
51
      if (end) cb(end)
52
      else cb(null,
53
        [ref.value, ref.name].concat(ref.attrs || []).join(' ') + '\n')
54
    })
55
  }
56
}
57
58
// upload-pack: fetch to client
59
function uploadPack(read, repo, options) {
60
  /* multi_ack thin-pack side-band side-band-64k ofs-delta shallow no-progress
61
   * include-tag multi_ack_detailed
62
   * agent=git/2.7.0 */
63
  var sendRefs = receivePackHeader([
64
    'thin-pack',
65
  ], repo.refs(), repo.symrefs(), false)
66
67
  var lines = pktLine.decode(read, options)
68
  var readWantHave = lines.haves()
69
  var acked
70
  var commonHash
71
  var sendPack
72
  var wants = {}
73
  var shallows = {}
74
  var aborted
75
  var hasWants
76
  var gotWants
77
78
  function readWant(abort, cb) {
79
    if (abort) return
80
    // read upload request (wants list) from client
81
    readWantHave(null, function next(end, want) {
82
      if (end || want.type == 'flush-pkt') {
83
        gotWants = true
84
        readHave(end === true ? null : end, cb)
85
        return
86
      }
87
      if (want.type == 'want') {
88
        wants[want.hash] = true
89
        hasWants = true
90
      } else if (want.type == 'shallow') {
91
        shallows[want.hash] = true
92
      } else {
93
        var err = new Error("Unknown thing", want.type, want.hash)
94
        return readWantHave(err, function (e) { cb(e || err) })
95
      }
96
      readWantHave(null, next)
97
    })
98
  }
99
100
  function readHave(abort, cb) {
101
    // Read upload haves (haves list).
102
    // On first obj-id that we have, ACK
103
    // If we have none, NAK.
104
    // TODO: implement multi_ack_detailed
105
    if (abort) return
106
    readWantHave(null, function next(end, have) {
107
      if (end === true) {
108
        cb(true)
109
      } else if (have.type === 'flush-pkt') {
110
        // found no common object
111
        if (!acked) {
112
          cb(null, 'NAK')
113
        } else {
114
          readWantHave(null, next)
115
        }
116
      } else if (end)
117
        cb(end)
118
      else if (have.type != 'have')
119
        cb(new Error('Unknown have' + JSON.stringify(have)))
120
      else if (acked)
121
        readWantHave(null, next)
122
      else
123
        repo.hasObjectFromAny(have.hash, function (err, haveIt) {
124
          if (err) return cb(err)
125
          if (!haveIt)
126
            return readWantHave(null, next)
127
          commonHash = haveIt
128
          acked = true
129
          cb(null, 'ACK ' + have.hash)
130
        })
131
    })
132
  }
133
134
  // Packfile negotiation
135
  return cat([
136
    pktLine.encode(cat([
137
      sendRefs,
138
      pull.once(''),
139
      function (abort, cb) {
140
        if (!gotWants) readWant(abort, cb)
141
        else readHave(abort, cb)
142
      }
143
    ])),
144
145
    function (abort, cb) {
146
      if (abort || aborted) return cb(abort || aborted)
147
      // send pack file to client
148
      if (sendPack)
149
        return sendPack(abort, cb)
150
      if (!hasWants) return cb(true)
151
      getObjects(repo, commonHash, wants, shallows,
152
        function (err, numObjects, readObjects) {
153
          if (err) return cb(err)
154
          var progress = progressObjects(options)
155
          progress.setNumObjects(numObjects)
156
          sendPack = pack.encode(options, numObjects,
157
            progress(readObjects))
158
          cb(null, '')
159
        }
160
      )
161
    }
162
  ])
163
}
164
165
// through stream to show a progress bar for objects being read
166
function progressObjects(options) {
167
  // Only show progress bar if it is requested and if it won't interfere with
168
  // the debug output
169
  if (!options.progress || options.verbosity > 1) {
170
    var dummyProgress = function (readObject) { return readObject }
171
    dummyProgress.setNumObjects = function () {}
172
    return dummyProgress
173
  }
174
175
  var numObjects
176
  var size = process.stderr.columns
177
  var bar = new ProgressBar(':percent :bar', {
178
    total: size,
179
    clear: true
180
  })
181
182
  var progress = function (readObject) {
183
    return function (abort, cb) {
184
      readObject(abort, function next(end, object) {
185
        if (end === true) {
186
          bar.terminate()
187
        } else if (!end) {
188
          var name = object.type + ' ' + object.length
189
          bar.tick(size / numObjects)
190
        }
191
192
        cb(end, object)
193
      })
194
    }
195
  }
196
  // TODO: put the num objects in the objects stream as a header object
197
  progress.setNumObjects = function (n) {
198
    numObjects = n
199
  }
200
  return progress
201
}
202
203
function getObjects(repo, commonHash, heads, shallows, cb) {
204
  // get objects from commonHash to each head, inclusive.
205
  // if commonHash is falsy, use root
206
  var objects = []
207
  var objectsAdded = {}
208
  var done = multicb({pluck: 1})
209
  var ended
210
211
  // walk back from heads until get to commonHash
212
  for (var hash in heads)
213
    addObject(hash, done())
214
215
  // TODO: only add new objects
216
217
  function addObject(hash, cb) {
218
    if (ended) return cb(ended)
219
    if (hash in objectsAdded || hash == commonHash) return cb()
220
    objectsAdded[hash] = true
221
    repo.getObjectFromAny(hash, function (err, object) {
222
      if (err) return cb(err)
223
      if (object.type == 'blob') {
224
        objects.push(object)
225
        cb()
226
      } else {
227
        // object must be read twice, so buffer it
228
        bufferObject(object, function (err, object) {
229
          if (err) return cb(err)
230
          objects.push(object)
231
          var hashes = getObjectLinks(object)
232
          for (var sha1 in hashes)
233
            addObject(sha1, done())
234
          cb()
235
        })
236
      }
237
    })
238
  }
239
240
  done(function (err) {
241
    if (err) return cb(err)
242
    cb(null, objects.length, pull.values(objects))
243
  })
244
}
245
246
function bufferObject(object, cb) {
247
  pull(
248
    object.read,
249
    pull.collect(function (err, bufs) {
250
      if (err) return cb(err)
251
      var buf = Buffer.concat(bufs, object.length)
252
      cb(null, {
253
        type: object.type,
254
        length: object.length,
255
        data: buf,
256
        read: pull.once(buf)
257
      })
258
    })
259
  )
260
}
261
262
// get hashes of git objects linked to from other git objects
263
function getObjectLinks(object, cb) {
264
  switch (object.type) {
265
    case 'blob':
266
      return {}
267
    case 'tree':
268
      return getTreeLinks(object.data)
269
    case 'tag':
270
    case 'commit':
271
      return getCommitOrTagLinks(object.data)
272
  }
273
}
274
275
function getTreeLinks(buf) {
276
  var links = {}
277
  for (var i = 0, j; j = buf.indexOf(0, i, 'ascii') + 1; i = j + 20) {
278
    var hash = buf.slice(j, j + 20).toString('hex')
279
    var mode = parseInt(buf.slice(i, j).toString('ascii'), 8)
280
    if (mode == 0160000) {
281
      // skip link to git commit since it may not be in this repo
282
      continue
283
    }
284
    if (!(hash in links))
285
      links[hash] = true
286
  }
287
  return links
288
}
289
290
function getCommitOrTagLinks(buf) {
291
  var lines = buf.toString('utf8').split('\n')
292
  var links = {}
293
  // iterate until reach blank line (indicating start of commit/tag body)
294
  for (var i = 0; lines[i]; i++) {
295
    var args = lines[i].split(' ')
296
    switch (args[0]) {
297
      case 'tree':
298
      case 'parent':
299
      case 'object':
300
        var hash = args[1]
301
        if (!(hash in links))
302
          links[hash] = true
303
    }
304
  }
305
  return links
306
}
307
308
/*
309
TODO: investigate capabilities
310
report-status delete-refs side-band-64k quiet atomic ofs-delta
311
*/
312
313
// Get a line for each ref that we have. The first line also has capabilities.
314
// Wrap with pktLine.encode.
315
function receivePackHeader(capabilities, refSource, symrefs, usePlaceholder) {
316
  var first = true
317
  var symrefed = {}
318
  var symrefsObj = {}
319
320
  return cat([
321
    function (end, cb) {
322
      if (end) cb(true)
323
      else if (!symrefs) cb(true)
324
      else pull(
325
        symrefs,
326
        pull.map(function (sym) {
327
          symrefed[sym.ref] = true
328
          symrefsObj[sym.name] = sym.ref
329
          return 'symref=' + sym.name + ':' + sym.ref
330
        }),
331
        pull.collect(function (err, symrefCaps) {
332
          if (err) return cb(err)
333
          capabilities = capabilities.concat(symrefCaps)
334
          cb(true)
335
        })
336
      )
337
    },
338
    pull(
339
      refSource,
340
      pull.map(function (ref) {
341
        // insert symrefs next to the refs that they point to
342
        var out = [ref]
343
        if (ref.name in symrefed)
344
          for (var symrefName in symrefsObj)
345
            if (symrefsObj[symrefName] === ref.name)
346
              out.push({name: symrefName, hash: ref.hash})
347
        return out
348
      }),
349
      pull.flatten(),
350
      pull.map(function (ref) {
351
        var name = ref.name
352
        var value = ref.hash
353
        if (first && usePlaceholder) {
354
          first = false
355
          /*
356
          if (end) {
357
            // use placeholder data if there are no refs
358
            value = '0000000000000000000000000000000000000000'
359
            name = 'capabilities^{}'
360
          }
361
          */
362
          name += '\0' + capabilities.join(' ')
363
        }
364
        return value + ' ' + name
365
      })
366
    )
367
  ])
368
}
369
370
// receive-pack: push from client
371
function receivePack(read, repo, options) {
372
  var sendRefs = receivePackHeader([
373
    'delete-refs',
374
    'no-thin',
375
  ], repo.refs(), null, true)
376
  var done = multicb({pluck: 1})
377
378
  return pktLine.encode(
379
    cat([
380
      // send our refs
381
      sendRefs,
382
      pull.once(''),
383
      function (abort, cb) {
384
        if (abort) return
385
        // receive their refs
386
        var lines = pktLine.decode(read, options)
387
        pull(
388
          lines.updates,
389
          pull.collect(function (err, updates) {
390
            if (err) return cb(err)
391
            if (updates.length === 0) return cb(true)
392
            var progress = progressObjects(options)
393
394
            if (repo.uploadPack) {
395
              var idxCb = done()
396
              indexPack(lines.passthrough, function (err, idx, packfileFixed) {
397
                if (err) return idxCb(err)
398
                repo.uploadPack(pull.values(updates), pull.once({
399
                  pack: pull(
400
                    packfileFixed,
401
                    // for some reason i was getting zero length buffers which
402
                    // were causing muxrpc to fail, so remove them here.
403
                    pull.filter(function (buf) {
404
                      return buf.length
405
                    })
406
                  ),
407
                  idx: idx
408
                }), idxCb)
409
              })
410
            } else {
411
              repo.update(pull.values(updates), pull(
412
                lines.passthrough,
413
                pack.decode({
414
                  verbosity: options.verbosity,
415
                  onHeader: function (numObjects) {
416
                    progress.setNumObjects(numObjects)
417
                  }
418
                }, repo, done()),
419
                progress
420
              ), done())
421
            }
422
423
            done(function (err) {
424
              cb(err || true)
425
            })
426
          })
427
        )
428
      },
429
      pull.once('unpack ok')
430
    ])
431
  )
432
}
433
434
function prepend(data, read) {
435
  var done
436
  return function (end, cb) {
437
    if (done) {
438
      read(end, cb)
439
    } else {
440
      done = true
441
      cb(null, data)
442
    }
443
  }
444
}
445
446
module.exports = function (repo) {
447
  var ended
448
  var options = {
449
    verbosity: 1,
450
    progress: false
451
  }
452
453
  repo = Repo(repo)
454
455
  function handleConnect(cmd, read) {
456
    var args = util.split2(cmd)
457
    switch (args[0]) {
458
      case 'git-upload-pack':
459
        return prepend('\n', uploadPack(read, repo, options))
460
      case 'git-receive-pack':
461
        return prepend('\n', receivePack(read, repo, options))
462
      default:
463
        return pull.error(new Error('Unknown service ' + args[0]))
464
    }
465
  }
466
467
  function handleCommand(line, read) {
468
    var args = util.split2(line)
469
    switch (args[0]) {
470
      case 'capabilities':
471
        return capabilitiesSource()
472
      case 'list':
473
        return listRefs(refSource)
474
      case 'connect':
475
        return handleConnect(args[1], read)
476
      case 'option':
477
        return optionSource(args[1], options)
478
      default:
479
        return pull.error(new Error('Unknown command ' + line))
480
    }
481
  }
482
483
  return function (read) {
484
    var b = buffered()
485
    b(read)
486
    var command
487
488
    function getCommand(cb) {
489
      b.lines(null, function next(end, line) {
490
        if (ended = end)
491
          return cb(end)
492
493
        if (line == '')
494
          return b.lines(null, next)
495
496
        if (options.verbosity > 1)
497
          console.error('command:', line)
498
499
        var cmdSource = handleCommand(line, b.passthrough)
500
        cb(null, cmdSource)
501
      })
502
    }
503
504
    return function next(abort, cb) {
505
      if (ended) return cb(ended)
506
507
      if (!command) {
508
        if (abort) return
509
        getCommand(function (end, cmd) {
510
          command = cmd
511
          next(end, cb)
512
        })
513
        return
514
      }
515
516
      command(abort, function (err, data) {
517
        if (err) {
518
          command = null
519
          if (err !== true)
520
            cb(err, data)
521
          else
522
            next(abort, cb)
523
        } else {
524
          cb(null, data)
525
        }
526
      })
527
    }
528
  }
529
}
530