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