git ssb

1
var WebTorrent = require('webtorrent')
2
var electron = require('electron')
3
var parseTorrent = require('parse-torrent')
4
var Path = require('path')
5
var getExt = require('path').extname
6
var fs = require('fs')
7
var ipc = electron.ipcRenderer
8
var watchEvent = require('./lib/watch-event')
9
var rimraf = require('rimraf')
10
var MutantDict = require('@mmckegg/mutant/dict')
11
var MutantStruct = require('@mmckegg/mutant/struct')
12
var convert = require('./lib/convert')
13
var TorrentStatus = require('./models/torrent-status')
14
var Tracker = require('bittorrent-tracker')
15
var magnet = require('magnet-uri')
16
var pull = require('pull-stream')
17
18
console.log = electron.remote.getGlobal('console').log
19
process.exit = electron.remote.app.quit
20
// redirect errors to stderr
21
window.addEventListener('error', function (e) {
22
  e.preventDefault()
23
  console.error(e.error.stack || 'Uncaught ' + e.error)
24
})
25
26
module.exports = function (client, config) {
27
  var seedWhiteList = new Set(config.seedWhiteList ? [].concat(config.seedWhiteList) : [client.id])
28
  var maxSeed = config.maxSeed == null ? 15 : parseInt(config.maxSeed, 10)
29
  var seedInterval = config.seedInterval == null ? 15 : parseInt(config.seedInterval, 10)
30
31
  var announce = config.webtorrent.announceList
32
  var torrentClient = new WebTorrent()
33
  var mediaPath = config.mediaPath
34
  var releases = {}
35
  var prioritizeReleases = []
36
  var paused = []
37
38
  var allTorrentStats = MutantStruct({
39
    downloadSpeed: 0,
40
    uploadSpeed: 0
41
  }, {nextTick: true})
42
43
  var torrentState = MutantDict()
44
45
  setInterval(pollStats, 0.5 * 1000)
46
  setInterval(scrapeInfo, 30 * 1000)
47
  setInterval(seedRarest, 30 * 60 * 1000)
48
49
  seedRarest()
50
  startAutoSeed()
51
52
  torrentClient.on('torrent', function (torrent) {
53
    watchTorrent(torrent.infoHash)
54
  })
55
56
  ipc.on('bg-release', function (ev, id) {
57
    if (releases[id]) {
58
      var release = releases[id]
59
      releases[id] = null
60
      release()
61
    }
62
  })
63
64
  ipc.on('bg-stream-torrent', (ev, id, torrentId) => {
65
    unprioritize(true, () => {
66
      var torrent = torrentClient.get(torrentId)
67
      if (torrent) {
68
        streamTorrent(id, torrentId)
69
      } else {
70
        addTorrent(torrentId, () => {
71
          streamTorrent(id, torrentId)
72
        })
73
      }
74
    })
75
76
    function streamTorrent (id, torrentId) {
77
      var torrent = torrentClient.get(torrentId)
78
      var server = torrent.createServer()
79
      prioritize(torrentId)
80
      server.listen(0, function (err) {
81
        if (err) return ipc.send('bg-response', id, err)
82
        var port = server.address().port
83
        var url = 'http://localhost:' + port + '/0'
84
        ipc.send('bg-response', id, null, url)
85
      })
86
      releases[id] = () => {
87
        server.close()
88
      }
89
    }
90
  })
91
92
  ipc.on('bg-export-torrent', (ev, id, torrentId, filePath) => {
93
    unprioritize(true, () => {
94
      var torrent = torrentClient.get(torrentId)
95
      if (torrent) {
96
        saveFile(id, torrentId, filePath)
97
      } else {
98
        addTorrent(torrentId, () => {
99
          saveFile(id, torrentId, filePath)
100
        })
101
      }
102
    })
103
104
    function saveFile (id, torrentId, exportPath) {
105
      var torrent = torrentClient.get(torrentId)
106
      torrentState.get(torrent.infoHash).saving.set(true)
107
      if (torrent.progress === 1) {
108
        done()
109
      } else {
110
        torrent.once('done', done)
111
      }
112
113
      function done () {
114
        var originalPath = Path.join(getTorrentDataPath(torrent.infoHash), torrent.files[0].path)
115
        convert.export(originalPath, exportPath, (err, info) => {
116
          torrentState.get(torrent.infoHash).saving.set(false)
117
          ipc.send('bg-response', id, err, info)
118
          console.log(info.toString())
119
        })
120
      }
121
    }
122
  })
123
124
  ipc.on('bg-check-torrent', (ev, id, torrentId) => {
125
    var torrent = torrentClient.get(torrentId)
126
    if (torrent) {
127
      ipc.send('bg-response', id, null)
128
    } else {
129
      addTorrent(torrentId, (err) => {
130
        ipc.send('bg-response', id, err)
131
      })
132
    }
133
  })
134
135
  ipc.on('bg-get-all-torrent-state', (ev, id) => {
136
    ipc.send('bg-response', id, torrentState())
137
  })
138
139
  ipc.on('bg-delete-torrent', (ev, id, torrentId) => {
140
    var torrentInfo = parseTorrent(torrentId)
141
    var torrent = torrentClient.get(torrentInfo.infoHash)
142
    if (torrent) {
143
      torrent.destroy()
144
    }
145
146
    fs.unlink(getTorrentPath(torrentInfo.infoHash), function () {
147
      rimraf(getTorrentDataPath(torrentInfo.infoHash), function () {
148
        console.log('Deleted torrent', torrentInfo.infoHash)
149
        ipc.send('bg-response', id)
150
      })
151
    })
152
  })
153
154
  ipc.on('bg-seed-torrent', (ev, id, infoHash) => {
155
    var torrent = torrentClient.get(infoHash)
156
    if (torrent) {
157
      ipc.send('bg-response', id, null, torrent.magnetURI)
158
    } else {
159
      fs.readFile(getTorrentPath(infoHash), function (err, buffer) {
160
        if (err) return ipc.send('bg-response', id, err)
161
        var torrent = parseTorrent(buffer)
162
        torrent.announce = announce.slice()
163
        torrentClient.add(torrent, {
164
          path: getTorrentDataPath(infoHash)
165
        }, function (torrent) {
166
          ipc.send('bg-response', id, null, torrent.magnetURI)
167
        })
168
      })
169
    }
170
  })
171
172
  ipc.send('ipcBackgroundReady', true)
173
174
  // scoped
175
176
  function watchTorrent (infoHash) {
177
    if (!torrentState.has(infoHash)) {
178
      var state = TorrentStatus(infoHash)
179
      torrentState.put(infoHash, state)
180
      state(function (value) {
181
        ipc.send('bg-torrent-status', infoHash, value)
182
      })
183
    }
184
  }
185
186
  function scrapeInfo () {
187
    var keys = torrentState.keys()
188
    getTorrentInfo(keys, (err, info) => {
189
      if (err) return console.log(err)
190
      Object.keys(info).forEach((key) => {
191
        var state = torrentState.get(key)
192
        if (state) {
193
          state.complete.set(info[key].complete)
194
        }
195
      })
196
    })
197
  }
198
199
  function scrapeInfoFor (infoHash) {
200
    getTorrentInfo(infoHash, (err, info) => {
201
      if (err) return console.log(err)
202
      var state = torrentState.get(infoHash)
203
      if (state && info) {
204
        state.complete.set(info.complete)
205
      }
206
    })
207
  }
208
209
  function pollStats () {
210
    torrentState.keys().forEach(refreshTorrentState)
211
    allTorrentStats.downloadSpeed.set(torrentClient.downloadSpeed)
212
    allTorrentStats.uploadSpeed.set(torrentClient.uploadSpeed)
213
  }
214
215
  function refreshTorrentState (infoHash) {
216
    var torrent = torrentClient.get(infoHash)
217
    var state = torrentState.get(infoHash)
218
    if (torrent) {
219
      state.progress.set(torrent.progress)
220
      state.downloadSpeed.set(torrent.downloadSpeed)
221
      state.uploadSpeed.set(torrent.uploadSpeed)
222
      state.uploaded.set(torrent.uploaded)
223
      state.downloaded.set(torrent.downloaded)
224
      state.numPeers.set(torrent.numPeers)
225
      state.seeding.set(true)
226
      state.loading.set(false)
227
    } else {
228
      state.seeding.set(false)
229
    }
230
  }
231
232
  function getTorrentPath (infoHash) {
233
    return `${getTorrentDataPath(infoHash)}.torrent`
234
  }
235
236
  function getTorrentDataPath (infoHash) {
237
    return Path.join(mediaPath, `${infoHash}`)
238
  }
239
240
  function addTorrent (torrentId, cb) {
241
    var torrent = parseTorrent(torrentId)
242
    var torrentPath = getTorrentPath(torrent.infoHash)
243
    torrent.announce = announce.slice()
244
245
    watchTorrent(torrent.infoHash)
246
247
    if (torrentClient.get(torrent.infoHash)) {
248
      cb()
249
    } else {
250
      torrentState.get(torrent.infoHash).loading.set(true)
251
      fs.exists(torrentPath, (exists) => {
252
        torrentClient.add(exists ? torrentPath : torrent, {
253
          path: getTorrentDataPath(torrent.infoHash),
254
          announce
255
        }, function (torrent) {
256
          scrapeInfoFor(torrent.infoHash)
257
          console.log('add torrent', torrent.infoHash)
258
          if (!exists) fs.writeFile(torrentPath, torrent.torrentFile, cb)
259
          else cb()
260
        })
261
      })
262
    }
263
  }
264
265
  function getTorrentInfo (infoHashes, cb) {
266
    if (infoHashes && infoHashes.length) {
267
      Tracker.scrape({
268
        announce: announce[0],
269
        infoHash: infoHashes
270
      }, function (err, info) {
271
        if (err) return cb(err)
272
        if (Array.isArray(infoHashes) && infoHashes.length === 1) info = {[info.infoHash]: info}
273
        cb(null, info)
274
      })
275
    } else {
276
      cb(null, {})
277
    }
278
  }
279
280
  function startAutoSeed () {
281
    client.friends.all((err, graph) => {
282
      if (err) throw err
283
284
      var extendedList = new Set(seedWhiteList)
285
      Array.from(seedWhiteList).forEach((id) => {
286
        var moreIds = graph[id]
287
        if (moreIds) {
288
          Object.keys(moreIds).forEach(x => extendedList.add(x))
289
        }
290
      })
291
292
      console.log(`Seeding torrents from: \n - ${Array.from(extendedList).join('\n - ')}`)
293
294
      pull(
295
        client.createLogStream({ live: true, gt: Date.now() }),
296
        ofType(['ferment/audio', 'ferment/update']),
297
        pull.drain((item) => {
298
          if (item.value && typeof item.value.content.audioSrc === 'string') {
299
            var torrent = magnet.decode(item.value.content.audioSrc)
300
            if (torrent.infoHash) {
301
              if (extendedList.has(item.value.author)) {
302
                fs.exists(Path.join(config.mediaPath, torrent.infoHash + '.torrent'), (exists) => {
303
                  if (!exists) {
304
                    if (!torrentClient.get(torrent.infoHash)) {
305
                      addTorrent(torrent)
306
                      console.log(`Auto seeding torrent ${torrent.infoHash}`)
307
                    }
308
                  }
309
                })
310
              }
311
            }
312
          }
313
        })
314
      )
315
    })
316
  }
317
318
  function seedRarest () {
319
    var i = 0
320
    var items = []
321
    var localTorrents = []
322
    fs.readdir(mediaPath, function (err, entries) {
323
      if (err) throw err
324
      entries.forEach((name) => {
325
        if (getExt(name) === '.torrent') {
326
          localTorrents.push(Path.basename(name, '.torrent'))
327
        }
328
      })
329
330
      // seed rarest torrents first
331
      getTorrentInfo(localTorrents, (err, info) => {
332
        if (err) return console.log(err)
333
        localTorrents.map(infoHash => [infoHash, info[infoHash].complete]).sort((a, b) => {
334
          return (a[1] + Math.random()) - (b[1] + Math.random())
335
        }).slice(0, maxSeed).forEach((item) => {
336
          watchTorrent(item[0])
337
          torrentState.get(item[0]).complete.set(item[1])
338
          items.push(Path.join(mediaPath, `${item[0]}.torrent`))
339
        })
340
        if (items.length) {
341
          next()
342
        }
343
      })
344
    })
345
346
    function next () {
347
      // don't seed all of the torrents at once, roll out slowly to avoid cpu spike
348
      var item = items[i]
349
      setTimeout(function () {
350
        fs.readFile(item, function (err, buffer) {
351
          if (!err) {
352
            var torrent = parseTorrent(buffer)
353
            if (!torrentClient.get(torrent.infoHash)) {
354
              torrent.announce = announce.slice()
355
              torrentClient.add(torrent, {
356
                path: getTorrentDataPath(Path.basename(item, '.torrent'))
357
              }, function (torrent) {
358
                console.log('seeding', torrent.infoHash)
359
                i += 1
360
                if (i < items.length) next()
361
              })
362
            } else {
363
              next()
364
            }
365
          }
366
        })
367
        // wait before seeding next file
368
      }, seedInterval * 1000)
369
    }
370
  }
371
372
  function unprioritize (restart, cb) {
373
    while (prioritizeReleases.length) {
374
      prioritizeReleases.pop()()
375
    }
376
377
    if (paused.length && restart) {
378
      var remaining = paused.length
379
      console.log(`restarting ${paused.length} torrent(s)`)
380
      while (paused.length) {
381
        var torrentFile = paused.pop()
382
        var torrent = parseTorrent(torrentFile)
383
        torrentClient.add(torrent, { path: getTorrentDataPath(torrent.infoHash), announce }, (torrent) => {
384
          remaining -= 1
385
          if (remaining === 0) {
386
            cb && cb()
387
          }
388
        })
389
      }
390
    } else {
391
      cb && cb()
392
    }
393
  }
394
395
  function prioritize (torrentId) {
396
    var torrent = torrentClient.get(torrentId)
397
    torrent.critical(0, Math.floor(torrent.pieces.length / 8))
398
    if (torrent.progress < 0.5) {
399
      torrentClient.torrents.forEach(function (t) {
400
        if (t !== torrent && t.progress < 0.9) {
401
          paused.push(t.torrentFile)
402
          t.destroy()
403
        }
404
      })
405
406
      console.log(`pausing ${paused.length} torrent(s)`)
407
408
      prioritizeReleases.push(watchEvent(torrent, 'download', () => {
409
        if (torrent.progress > 0.8) {
410
          unprioritize(true)
411
        }
412
      }))
413
    }
414
  }
415
}
416
417
function ofType (types) {
418
  types = Array.isArray(types) ? types : [types]
419
  return pull.filter((item) => {
420
    if (item.value) {
421
      return types.includes(item.value.content.type)
422
    } else {
423
      return true
424
    }
425
  })
426
}
427