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