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