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