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