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