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
  /* multi_ack thin-pack side-band side-band-64k ofs-delta shallow no-progress
67
   * include-tag multi_ack_detailed
68
   * agent=git/2.7.0 */
69
  var sendRefs = receivePackHeader([
70
    agentCap,
71
  ], repo.refs(), repo.symrefs())
72
73
  var lines = pktLine.decode(read, {
74
    onCaps: onCaps,
75
    verbosity: options.verbosity
76
  })
77
  var readWantHave = lines.haves()
78
  var acked
79
  var commonHash
80
  var sendPack
81
  var wants = {}
82
  var shallows = {}
83
  var aborted
84
  var hasWants
85
  var gotHaves
86
87
  function onCaps(caps) {
88
    if (options.verbosity >= 2) {
89
      console.error('client capabilities:', caps)
90
    }
91
  }
92
93
  function readWant(abort, cb) {
94
    if (abort) return
95
    // read upload request (wants list) from client
96
    readWantHave(null, function next(end, want) {
97
      if (end || want.type == 'flush-pkt') {
98
        cb(end || true, cb)
99
        return
100
      }
101
      if (want.type == 'want') {
102
        wants[want.hash] = true
103
        hasWants = true
104
      } else if (want.type == 'shallow') {
105
        shallows[want.hash] = true
106
      } else {
107
        var err = new Error("Unknown thing", want.type, want.hash)
108
        return readWantHave(err, function (e) { cb(e || err) })
109
      }
110
      readWantHave(null, next)
111
    })
112
  }
113
114
  function readHave(abort, cb) {
115
    // Read upload haves (haves list).
116
    // On first obj-id that we have, ACK
117
    // If we have none, NAK.
118
    // TODO: implement multi_ack_detailed
119
    // FIXME!
120
    if (abort) return
121
    if (gotHaves) return cb(true)
122
    readWantHave(null, function next(end, have) {
123
      if (end === true) {
124
        gotHaves = true
125
        if (!acked) {
126
          cb(null, 'NAK')
127
        } else {
128
          cb(true)
129
        }
130
      } else if (have.type === 'flush-pkt') {
131
        // found no common object
132
        if (!acked) {
133
          cb(null, 'NAK')
134
        } else {
135
          readWantHave(null, next)
136
        }
137
      } else if (end)
138
        cb(end)
139
      else if (have.type != 'have')
140
        cb(new Error('Unknown have' + JSON.stringify(have)))
141
      else if (acked)
142
        readWantHave(null, next)
143
      else
144
        repo.hasObjectFromAny(have.hash, function (err, haveIt) {
145
          if (err) return cb(err)
146
          if (!haveIt)
147
            return readWantHave(null, next)
148
          commonHash = haveIt
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('common', commonHash, 'wants', wants)
162
    }
163
    // TODO: show progress during getObjects
164
    getObjects(repo, commonHash, wants, shallows,
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 >= 1) {
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, commonHash, heads, shallows, cb) {
229
  // get objects from commonHash to each head, inclusive.
230
  // if commonHash is falsy, use root
231
  var objects = []
232
  var objectsAdded = {}
233
  var done = multicb({pluck: 1})
234
  var ended
235
236
  // walk back from heads until get to commonHash
237
  for (var hash in heads)
238
    addObject(hash, done())
239
240
  // TODO: only add new objects
241
242
  function addObject(hash, cb) {
243
    if (ended) return cb(ended)
244
    if (hash in objectsAdded || hash == commonHash) return cb()
245
    objectsAdded[hash] = true
246
    repo.getObjectFromAny(hash, function (err, object) {
247
      if (err) return cb(err)
248
      if (object.type == 'blob') {
249
        objects.push(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.push(object)
256
          var hashes = getObjectLinks(object)
257
          for (var sha1 in hashes)
258
            addObject(sha1, done())
259
          cb()
260
        })
261
      }
262
    })
263
  }
264
265
  done(function (err) {
266
    if (err) return cb(err)
267
    // console.error(objects.reduce(function (n, obj) { return obj.length + n}, 0) + ' bytes')
268
    cb(null, objects.length, pull.values(objects))
269
  })
270
}
271
272
function bufferObject(object, cb) {
273
  pull(
274
    object.read,
275
    pull.collect(function (err, bufs) {
276
      if (err) return cb(err)
277
      var buf = Buffer.concat(bufs, object.length)
278
      cb(null, {
279
        type: object.type,
280
        length: object.length,
281
        data: buf,
282
        read: pull.once(buf)
283
      })
284
    })
285
  )
286
}
287
288
// get hashes of git objects linked to from other git objects
289
function getObjectLinks(object, cb) {
290
  switch (object.type) {
291
    case 'blob':
292
      return {}
293
    case 'tree':
294
      return getTreeLinks(object.data)
295
    case 'tag':
296
    case 'commit':
297
      return getCommitOrTagLinks(object.data)
298
  }
299
}
300
301
function getTreeLinks(buf) {
302
  var links = {}
303
  for (var i = 0, j; j = buf.indexOf(0, i, 'ascii') + 1; i = j + 20) {
304
    var hash = buf.slice(j, j + 20).toString('hex')
305
    var mode = parseInt(buf.slice(i, j).toString('ascii'), 8)
306
    if (mode == 0160000) {
307
      // skip link to git commit since it may not be in this repo
308
      continue
309
    }
310
    if (!(hash in links))
311
      links[hash] = true
312
  }
313
  return links
314
}
315
316
function getCommitOrTagLinks(buf) {
317
  var lines = buf.toString('utf8').split('\n')
318
  var links = {}
319
  // iterate until reach blank line (indicating start of commit/tag body)
320
  for (var i = 0; lines[i]; i++) {
321
    var args = lines[i].split(' ')
322
    switch (args[0]) {
323
      case 'tree':
324
      case 'parent':
325
      case 'object':
326
        var hash = args[1]
327
        if (!(hash in links))
328
          links[hash] = true
329
    }
330
  }
331
  return links
332
}
333
334
/*
335
TODO: investigate capabilities
336
report-status delete-refs side-band-64k quiet atomic ofs-delta
337
*/
338
339
// Get a line for each ref that we have. The first line also has capabilities.
340
// Wrap with pktLine.encode.
341
function receivePackHeader(capabilities, refSource, symrefs) {
342
  var first = true
343
  var symrefed = {}
344
  var symrefsObj = {}
345
346
  return cat([
347
    function (end, cb) {
348
      if (end) cb(true)
349
      else if (!symrefs) cb(true)
350
      else pull(
351
        symrefs,
352
        pull.map(function (sym) {
353
          symrefed[sym.ref] = true
354
          symrefsObj[sym.name] = sym.ref
355
          return 'symref=' + sym.name + ':' + sym.ref
356
        }),
357
        pull.collect(function (err, symrefCaps) {
358
          if (err) return cb(err)
359
          capabilities = capabilities.concat(symrefCaps)
360
          cb(true)
361
        })
362
      )
363
    },
364
    pull(
365
      refSource,
366
      pull.map(function (ref) {
367
        // insert symrefs next to the refs that they point to
368
        var out = [ref]
369
        if (ref.name in symrefed)
370
          for (var symrefName in symrefsObj)
371
            if (symrefsObj[symrefName] === ref.name)
372
              out.push({name: symrefName, hash: ref.hash})
373
        return out
374
      }),
375
      pull.flatten(),
376
      pull.map(function (ref) {
377
        var name = ref.name
378
        var value = ref.hash
379
        if (first) {
380
          first = false
381
          /*
382
          if (end) {
383
            // use placeholder data if there are no refs
384
            value = '0000000000000000000000000000000000000000'
385
            name = 'capabilities^{}'
386
          }
387
          */
388
          name += '\0' + capabilities.join(' ')
389
        }
390
        return value + ' ' + name
391
      })
392
    )
393
  ])
394
}
395
396
// receive-pack: push from client
397
function receivePack(read, repo, options) {
398
  var sendRefs = receivePackHeader([
399
    agentCap,
400
    'delete-refs',
401
    'quiet',
402
    'no-thin',
403
  ], repo.refs(), null)
404
  var done = multicb({pluck: 1})
405
406
  function onCaps(caps) {
407
    if (options.verbosity >= 2) {
408
      console.error('client capabilities:', caps)
409
    }
410
  }
411
412
  return pktLine.encode(
413
    cat([
414
      // send our refs
415
      sendRefs,
416
      pull.once(''),
417
      function (abort, cb) {
418
        if (abort) return
419
        // receive their refs
420
        var lines = pktLine.decode(read, {
421
          onCaps: onCaps,
422
          verbosity: options.verbosity
423
        })
424
        pull(
425
          lines.updates,
426
          pull.collect(function (err, updates) {
427
            if (err) return cb(err)
428
            if (updates.length === 0) return cb(true)
429
            var progress = progressObjects(options)
430
431
            var hasPack = !updates.every(function (update) {
432
              return update.new === null
433
            })
434
            if (!hasPack) {
435
              return repo.update(pull.values(updates), pull.empty(), done())
436
            }
437
438
            if (repo.uploadPack) {
439
              var idxCb = done()
440
              indexPack(lines.passthrough, function (err, idx, packfileFixed) {
441
                if (err) return idxCb(err)
442
                repo.uploadPack(pull.values(updates), pull.once({
443
                  pack: pull(
444
                    packfileFixed,
445
                    // for some reason i was getting zero length buffers which
446
                    // were causing muxrpc to fail, so remove them here.
447
                    pull.filter(function (buf) {
448
                      return buf.length
449
                    })
450
                  ),
451
                  idx: idx
452
                }), idxCb)
453
              })
454
            } else {
455
              repo.update(pull.values(updates), pull(
456
                lines.passthrough,
457
                pack.decode({
458
                  verbosity: options.verbosity,
459
                  onHeader: function (numObjects) {
460
                    progress.setNumObjects(numObjects)
461
                  }
462
                }, repo, done()),
463
                progress
464
              ), done())
465
            }
466
467
            done(function (err) {
468
              cb(err || true)
469
            })
470
          })
471
        )
472
      },
473
      pull.once('unpack ok')
474
    ])
475
  )
476
}
477
478
function prepend(data, read) {
479
  var done
480
  return function (end, cb) {
481
    if (done) {
482
      read(end, cb)
483
    } else {
484
      done = true
485
      cb(null, data)
486
    }
487
  }
488
}
489
490
module.exports = function (repo) {
491
  var ended
492
  var options = {
493
    verbosity: +process.env.GIT_VERBOSITY || 1,
494
    progress: false
495
  }
496
497
  repo = Repo(repo)
498
499
  function handleConnect(cmd, read) {
500
    var args = util.split2(cmd)
501
    switch (args[0]) {
502
      case 'git-upload-pack':
503
        return prepend('\n', uploadPack(read, repo, options))
504
      case 'git-receive-pack':
505
        return prepend('\n', receivePack(read, repo, options))
506
      default:
507
        return pull.error(new Error('Unknown service ' + args[0]))
508
    }
509
  }
510
511
  function handleCommand(line, read) {
512
    var args = util.split2(line)
513
    switch (args[0]) {
514
      case 'capabilities':
515
        return capabilitiesSource()
516
      case 'list':
517
        return listRefs(refSource)
518
      case 'connect':
519
        return handleConnect(args[1], read)
520
      case 'option':
521
        return optionSource(args[1], options)
522
      default:
523
        return pull.error(new Error('Unknown command ' + line))
524
    }
525
  }
526
527
  return function (read) {
528
    var b = buffered()
529
    if (options.verbosity >= 3) {
530
      read = pull.through(function (data) {
531
        console.error('>', JSON.stringify(data.toString('ascii')))
532
      })(read)
533
    }
534
    b(read)
535
536
    var command
537
538
    function getCommand(cb) {
539
      b.lines(null, function next(end, line) {
540
        if (ended = end)
541
          return cb(end)
542
543
        if (line == '')
544
          return b.lines(null, next)
545
546
        if (options.verbosity > 1)
547
          console.error('command:', line)
548
549
        var cmdSource = handleCommand(line, b.passthrough)
550
        cb(null, cmdSource)
551
      })
552
    }
553
554
    return function next(abort, cb) {
555
      if (ended) return cb(ended)
556
557
      if (!command) {
558
        if (abort) return
559
        getCommand(function (end, cmd) {
560
          command = cmd
561
          next(end, cb)
562
        })
563
        return
564
      }
565
566
      command(abort, function (err, data) {
567
        if (err) {
568
          command = null
569
          if (err !== true)
570
            cb(err, data)
571
          else
572
            next(abort, cb)
573
        } else {
574
          if (options.verbosity >= 3) {
575
            console.error('<', JSON.stringify(data))
576
          }
577
          cb(null, data)
578
        }
579
      })
580
    }
581
  }
582
}
583