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
var dummyCapsRef = {
17
  name: 'capabilities^{}',
18
  hash: '0000000000000000000000000000000000000000'
19
}
20
21
function handleOption(options, name, value) {
22
  switch (name) {
23
    case 'verbosity':
24
      options.verbosity = value|0
25
      return true
26
    case 'progress':
27
      options.progress = !!value && value !== 'false'
28
      return true
29
      /*
30
    case 'followtags':
31
      options.followtags = !!value
32
      return true
33
    case 'depth':
34
      options.depth = depth|0
35
      return true
36
    case 'deepen-relative':
37
      options.deepenRelative = !!value && value !== 'false'
38
      return true
39
      */
40
    default:
41
      console.error('unknown option', name + ': ' + value)
42
      return false
43
  }
44
}
45
46
function capabilitiesSource() {
47
  return pull.once([
48
    'option',
49
    'connect',
50
  ].join('\n') + '\n\n')
51
}
52
53
function optionSource(cmd, options) {
54
  var args = util.split2(cmd)
55
  var msg = handleOption(options, args[0], args[1])
56
  msg = (msg === true) ? 'ok'
57
      : (msg === false) ? 'unsupported'
58
      : 'error ' + msg
59
  return pull.once(msg + '\n')
60
}
61
62
// transform ref objects into lines
63
function listRefs(read) {
64
  var ended
65
  return function (abort, cb) {
66
    if (ended) return cb(ended)
67
    read(abort, function (end, ref) {
68
      ended = end
69
      if (end === true) cb(null, '\n')
70
      if (end) cb(end)
71
      else cb(null,
72
        [ref.value, ref.name].concat(ref.attrs || []).join(' ') + '\n')
73
    })
74
  }
75
}
76
77
// upload-pack: fetch to client
78
// references:
79
//   git/documentation/technical/pack-protocol.txt
80
//   git/documentation/technical/protocol-capabilities.txt
81
function uploadPack(read, repo, options) {
82
  var sendRefs = receivePackHeader([
83
    agentCap,
84
  ], repo.refs(), repo.symrefs())
85
86
  var lines = pktLine.decode(read, {
87
    onCaps: onCaps,
88
    verbosity: options.verbosity
89
  })
90
  var readWantHave = lines.haves()
91
  var acked
92
  var haves = {}
93
  var sendPack
94
  var wants = {}
95
  var shallows = {}
96
  var aborted
97
  var hasWants
98
  var gotHaves
99
100
  function onCaps(caps) {
101
    if (options.verbosity >= 2) {
102
      console.error('client capabilities:', caps)
103
    }
104
  }
105
106
  function readWant(abort, cb) {
107
    if (abort) return
108
    // read upload request (wants list) from client
109
    readWantHave(null, function next(end, want) {
110
      if (end || want.type == 'flush-pkt') {
111
        cb(end || true, cb)
112
        return
113
      }
114
      if (want.type == 'want') {
115
        wants[want.hash] = true
116
        hasWants = true
117
      } else if (want.type == 'shallow') {
118
        shallows[want.hash] = true
119
      } else {
120
        var err = new Error("Unknown thing", want.type, want.hash)
121
        return readWantHave(err, function (e) { cb(e || err) })
122
      }
123
      readWantHave(null, next)
124
    })
125
  }
126
127
  function readHave(abort, cb) {
128
    // Read upload haves (haves list).
129
    // On first obj-id that we have, ACK
130
    // If we have none, NAK.
131
    if (abort) return
132
    if (gotHaves) return cb(true)
133
    readWantHave(null, function next(end, have) {
134
      if (end === true) { // done
135
        gotHaves = true
136
        if (!acked) {
137
          cb(null, 'NAK')
138
        } else {
139
          cb(true)
140
        }
141
      } else if (have.type === 'flush-pkt') {
142
        // found no common object
143
        if (!acked) {
144
          cb(null, 'NAK')
145
        } else {
146
          readWantHave(null, next)
147
        }
148
      } else if (end)
149
        cb(end)
150
      else if (have.type != 'have')
151
        cb(new Error('Unknown have' + JSON.stringify(have)))
152
      else {
153
        haves[have.hash] = true
154
        if (acked) {
155
          readWantHave(null, next)
156
        } else if (repo.hasObjectQuick) {
157
          gotHasObject(null, repo.hasObjectQuick(have.hash))
158
        } else {
159
          repo.hasObjectFromAny(have.hash, gotHasObject)
160
        }
161
      }
162
      function gotHasObject(err, haveIt) {
163
        if (err) return cb(err)
164
        if (!haveIt) return readWantHave(null, next)
165
        acked = true
166
        cb(null, 'ACK ' + have.hash)
167
      }
168
    })
169
  }
170
171
  function readPack(abort, cb) {
172
    if (sendPack) return sendPack(abort, cb)
173
    if (abort || aborted) return console.error('abrt', abort || aborted), cb(abort || aborted)
174
    // send pack file to client
175
    if (!hasWants) return cb(true)
176
    if (options.verbosity >= 2) {
177
      console.error('wants', wants)
178
    }
179
    repo.getPack(wants, haves, options, function (err, _sendPack) {
180
      if (err) return cb(err)
181
      sendPack = _sendPack
182
      sendPack(null, cb)
183
    })
184
  }
185
186
  // Packfile negotiation
187
  return cat([
188
    pktLine.encode(cat([
189
      sendRefs,
190
      pull.once(''),
191
      readWant,
192
      readHave
193
    ])),
194
    readPack
195
  ])
196
}
197
198
// through stream to show a progress bar for objects being read
199
function progressObjects(options) {
200
  // Only show progress bar if it is requested and if it won't interfere with
201
  // the debug output
202
  if (!options.progress || options.verbosity > 1) {
203
    var dummyProgress = function (readObject) { return readObject }
204
    dummyProgress.setNumObjects = function () {}
205
    return dummyProgress
206
  }
207
208
  var numObjects
209
  var size = process.stderr.columns
210
  var bar = new ProgressBar(':percent :bar', {
211
    total: size,
212
    clear: true
213
  })
214
215
  var progress = function (readObject) {
216
    return function (abort, cb) {
217
      readObject(abort, function next(end, object) {
218
        if (end === true) {
219
          bar.terminate()
220
        } else if (!end) {
221
          var name = object.type + ' ' + object.length
222
          bar.tick(size / numObjects)
223
        }
224
225
        cb(end, object)
226
      })
227
    }
228
  }
229
  // TODO: put the num objects in the objects stream as a header object
230
  progress.setNumObjects = function (n) {
231
    numObjects = n
232
  }
233
  return progress
234
}
235
236
// Get a line for each ref that we have. The first line also has capabilities.
237
// Wrap with pktLine.encode.
238
function receivePackHeader(capabilities, refSource, symrefs) {
239
  var hasRefs = false
240
  var first = true
241
  var symrefed = {}
242
  var symrefsObj = {}
243
244
  return cat([
245
    function (end, cb) {
246
      if (end) cb(true)
247
      else if (!symrefs) cb(true)
248
      else pull(
249
        symrefs,
250
        pull.map(function (sym) {
251
          symrefed[sym.ref] = true
252
          symrefsObj[sym.name] = sym.ref
253
          return 'symref=' + sym.name + ':' + sym.ref
254
        }),
255
        pull.collect(function (err, symrefCaps) {
256
          if (err) return cb(err)
257
          capabilities = capabilities.concat(symrefCaps)
258
          cb(true)
259
        })
260
      )
261
    },
262
    pull(
263
      cat([refSource, pull.once(dummyCapsRef)]),
264
      pull.map(function (ref) {
265
        if (ref !== dummyCapsRef) hasRefs = true
266
        else if (hasRefs) return []
267
        else return [ref]
268
269
        // insert symrefs next to the refs that they point to
270
        var out = [ref]
271
        if (ref.name in symrefed)
272
          for (var symrefName in symrefsObj)
273
            if (symrefsObj[symrefName] === ref.name)
274
              out.push({name: symrefName, hash: ref.hash})
275
        return out
276
      }),
277
      pull.flatten(),
278
      pull.map(function (ref) {
279
        var name = ref.name
280
        var value = ref.hash
281
        if (first) {
282
          first = false
283
          name += '\0' + capabilities.join(' ')
284
        }
285
        return value + ' ' + name
286
      })
287
    )
288
  ])
289
}
290
291
// receive-pack: push from client
292
function receivePack(read, repo, options) {
293
  var serverCaps = [
294
    agentCap,
295
    'delete-refs',
296
    'quiet',
297
    'no-thin'
298
  ]
299
  if (repo.setPushOptions) serverCaps.push('push-options')
300
  var sendRefs = receivePackHeader(serverCaps, repo.refs(), null)
301
  var done = multicb({pluck: 1})
302
303
  var clientCaps = {}
304
305
  function onCaps(caps) {
306
    clientCaps = caps
307
    if (options.verbosity >= 2) {
308
      console.error('client capabilities:', caps)
309
    }
310
  }
311
312
  return pktLine.encode(
313
    cat([
314
      // send our refs
315
      sendRefs,
316
      pull.once(''),
317
      function (abort, cb) {
318
        if (abort) return
319
        // receive their refs
320
        var lines = pktLine.decode(read, {
321
          onCaps: onCaps,
322
          verbosity: options.verbosity
323
        })
324
        pull(
325
          lines.updates,
326
          pull.collect(function (err, updates) {
327
            if (err) return cb(err)
328
            if (updates.length === 0) return cb(true)
329
            var usePushOpts = clientCaps['push-options']
330
            pull(
331
              usePushOpts ? lines.pktLineStrs : pull.empty(),
332
              pull.reduce(function (opts, opt) {
333
                opts[opt] = true
334
                return opts
335
              }, {}, pushOptsCb)
336
            )
337
            function pushOptsCb(err, pushOpts) {
338
              if (err) return cb(err)
339
              repo.setPushOptions(pushOpts)
340
              var progress = progressObjects(options)
341
342
              var hasPack = !updates.every(function (update) {
343
                return update.new === null
344
              })
345
              if (repo.uploadPack) {
346
                if (!hasPack) {
347
                  repo.uploadPack(pull.values(updates), pull.empty(), done())
348
                } else {
349
                  var idxCb = done()
350
                  pull(lines.passthrough, indexPack(function (err, idx, packfileFixed) {
351
                    if (err) return idxCb(err)
352
                    repo.uploadPack(pull.values(updates), pull.once({
353
                      pack: pull(
354
                        packfileFixed,
355
                        // for some reason i was getting zero length buffers which
356
                        // were causing muxrpc to fail, so remove them here.
357
                        pull.filter(function (buf) {
358
                          return buf.length
359
                        })
360
                      ),
361
                      idx: idx
362
                    }), idxCb)
363
                  }))
364
                }
365
              } else {
366
                repo.update(pull.values(updates), !hasPack ? pull.empty() : pull(
367
                  lines.passthrough,
368
                  pack.decode({
369
                    verbosity: options.verbosity,
370
                    onHeader: function (numObjects) {
371
                      progress.setNumObjects(numObjects)
372
                    }
373
                  }, repo, done()),
374
                  progress
375
                ), done())
376
              }
377
378
              done(function (err) {
379
                cb(err || true)
380
              })
381
            }
382
          })
383
        )
384
      },
385
      pull.once('unpack ok')
386
    ])
387
  )
388
}
389
390
function prepend(data, read) {
391
  var done
392
  return function (end, cb) {
393
    if (done) {
394
      read(end, cb)
395
    } else {
396
      done = true
397
      cb(null, data)
398
    }
399
  }
400
}
401
402
module.exports = function (repo) {
403
  var ended
404
  var options = {
405
    verbosity: +process.env.GIT_VERBOSITY || 1,
406
    progress: false
407
  }
408
409
  repo = Repo(repo)
410
411
  function handleConnect(cmd, read) {
412
    var args = util.split2(cmd)
413
    switch (args[0]) {
414
      case 'git-upload-pack':
415
        return prepend('\n', uploadPack(read, repo, options))
416
      case 'git-receive-pack':
417
        return prepend('\n', receivePack(read, repo, options))
418
      default:
419
        return pull.error(new Error('Unknown service ' + args[0]))
420
    }
421
  }
422
423
  function handleCommand(line, read) {
424
    var args = util.split2(line)
425
    switch (args[0]) {
426
      case 'capabilities':
427
        return capabilitiesSource()
428
      case 'list':
429
        return listRefs(refSource)
430
      case 'connect':
431
        return handleConnect(args[1], read)
432
      case 'option':
433
        return optionSource(args[1], options)
434
      default:
435
        return pull.error(new Error('Unknown command ' + line))
436
    }
437
  }
438
439
  return function (read) {
440
    var b = buffered()
441
    if (options.verbosity >= 3) {
442
      read = pull.through(function (data) {
443
        console.error('>', JSON.stringify(data.toString('ascii')))
444
      })(read)
445
    }
446
    b(read)
447
448
    var command
449
450
    function getCommand(cb) {
451
      b.lines(null, function next(end, line) {
452
        if (ended = end)
453
          return cb(end)
454
455
        if (line == '')
456
          return b.lines(null, next)
457
458
        if (options.verbosity > 1)
459
          console.error('command:', line)
460
461
        var cmdSource = handleCommand(line, b.passthrough)
462
        cb(null, cmdSource)
463
      })
464
    }
465
466
    return function next(abort, cb) {
467
      if (ended) return cb(ended)
468
469
      if (!command) {
470
        if (abort) return
471
        getCommand(function (end, cmd) {
472
          command = cmd
473
          next(end, cb)
474
        })
475
        return
476
      }
477
478
      command(abort, function (err, data) {
479
        if (err) {
480
          command = null
481
          if (err !== true)
482
            cb(err, data)
483
          else
484
            next(abort, cb)
485
        } else {
486
          if (options.verbosity >= 3) {
487
            console.error('<', JSON.stringify(data))
488
          }
489
          cb(null, data)
490
        }
491
      })
492
    }
493
  }
494
}
495