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