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