git ssb

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