git ssb

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