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