git ssb

1
var pull = require('pull-stream')
2
var ref = require('ssb-ref')
3
var Reconnect = require('pull-reconnect')
4
var createClient = require('ssb-client')
5
var createFeed = require('ssb-feed')
6
var nest = require('depnest')
7
var Value = require('mutant/value')
8
var ssbKeys = require('ssb-keys')
9
10
exports.needs = nest({
11
  'config.sync.load': 'first',
12
  'keys.sync.load': 'first',
13
  'sbot.obs.connectionStatus': 'first',
14
  'sbot.hook.feed': 'map'
15
})
16
17
exports.gives = {
18
  sbot: {
19
    sync: {
20
      cache: true
21
    },
22
    async: {
23
      get: true,
24
      publish: true,
25
      addBlob: true,
26
      gossipConnect: true
27
    },
28
    pull: {
29
      log: true,
30
      userFeed: true,
31
      query: true,
32
      feed: true,
33
      links: true,
34
      search: true,
35
      replicateProgress: true,
36
      queryProgress: true
37
    },
38
    obs: {
39
      connectionStatus: true,
40
      connectedPeers: true,
41
      localPeers: true
42
    }
43
  }
44
}
45
46
exports.create = function (api) {
47
  const config = api.config.sync.load()
48
  const keys = api.keys.sync.load()
49
  var cache = {}
50
51
  var sbot = null
52
  var connectionStatus = Value()
53
  var connectedPeers = Value([])
54
  var localPeers = Value([])
55
56
  setInterval(refreshPeers, 5e3)
57
58
  var rec = Reconnect(function (isConn) {
59
    function notify (value) {
60
      isConn(value); connectionStatus.set(value)
61
    }
62
63
    createClient(keys, config, function (err, _sbot) {
64
      if (err) {
65
        return notify(err)
66
      }
67
68
      sbot = _sbot
69
      sbot.on('closed', function () {
70
        sbot = null
71
        notify(new Error('closed'))
72
      })
73
74
      notify()
75
    })
76
  })
77
78
  var internal = {
79
    getLatest: rec.async(function (id, cb) {
80
      sbot.getLatest(id, cb)
81
    }),
82
    add: rec.async(function (msg, cb) {
83
      sbot.add(msg, cb)
84
    })
85
  }
86
87
  var feed = createFeed(internal, keys, {remote: true})
88
89
  return {
90
    sbot: {
91
      sync: {
92
        cache: () => cache
93
      },
94
      async: {
95
        get: rec.async(function (key, cb) {
96
          if (typeof cb !== 'function') {
97
            throw new Error('cb must be function')
98
          }
99
          if (cache[key]) cb(null, cache[key])
100
          else {
101
            sbot.get(key, function (err, value) {
102
              if (err) return cb(err)
103
              runHooks({key, value})
104
              cb(null, value)
105
            })
106
          }
107
        }),
108
        publish: rec.async((content, cb) => {
109
          if (content.recps) {
110
            content = ssbKeys.box(content, content.recps.map(e => {
111
              return ref.isFeed(e) ? e : e.link
112
            }))
113
          } else if (content.mentions) {
114
            content.mentions.forEach(mention => {
115
              if (ref.isBlob(mention.link)) {
116
                sbot.blobs.push(mention.link, err => {
117
                  if (err) console.error(err)
118
                })
119
              }
120
            })
121
          }
122
123
          feed.add(content, (err, msg) => {
124
            if (err) console.error(err)
125
            else if (!cb) console.log(msg)
126
            cb && cb(err, msg)
127
          })
128
        }),
129
        addBlob: rec.async((stream, cb) => {
130
          return pull(
131
            stream,
132
            Hash(function (err, id) {
133
              if (err) return cb(err)
134
              // completely UGLY hack to tell when the blob has been sucessfully written...
135
              var start = Date.now()
136
              var n = 5
137
              next()
138
139
              function next () {
140
                setTimeout(function () {
141
                  sbot.blobs.has(id, function (_, has) {
142
                    if (has) return cb(null, id)
143
                    if (n--) next()
144
                    else cb(new Error('write failed'))
145
                  })
146
                }, Date.now() - start)
147
              }
148
            }),
149
            sbot.blobs.add()
150
          )
151
        }),
152
        gossipConnect: rec.async(function (opts, cb) {
153
          sbot.gossip.connect(opts, cb)
154
        })
155
      },
156
      pull: {
157
        query: rec.source(query => {
158
          return sbot.query.read(query)
159
        }),
160
        userFeed: rec.source(opts => {
161
          return sbot.createUserStream(opts)
162
        }),
163
        feed: rec.source(function (opts) {
164
          return pull(
165
            sbot.createFeedStream(opts),
166
            pull.through(runHooks)
167
          )
168
        }),
169
        log: rec.source(opts => {
170
          return pull(
171
            sbot.createLogStream(opts),
172
            pull.through(runHooks)
173
          )
174
        }),
175
        links: rec.source(function (query) {
176
          return sbot.links(query)
177
        }),
178
        search: rec.source(function (opts) {
179
          return sbot.fulltext.search(opts)
180
        }),
181
        replicateProgress: rec.source(function (opts) {
182
          return sbot.replicate.changes()
183
        }),
184
        queryProgress: rec.source(function (opts) {
185
          return sbot.query.progress()
186
        })
187
      },
188
      obs: {
189
        connectionStatus: (listener) => connectionStatus(listener),
190
        connectedPeers: () => connectedPeers,
191
        localPeers: () => localPeers
192
      }
193
    }
194
  }
195
196
  // scoped
197
198
  function runHooks (msg) {
199
    if (!cache[msg.key]) {
200
      cache[msg.key] = msg.value
201
      api.sbot.hook.feed(msg)
202
    }
203
  }
204
205
  function refreshPeers () {
206
    if (sbot) {
207
      sbot.gossip.peers((err, peers) => {
208
        if (err) throw console.log(err)
209
        connectedPeers.set(peers.filter(x => x.state === 'connected').map(x => x.key))
210
        localPeers.set(peers.filter(x => x.source === 'local').map(x => x.key))
211
      })
212
    }
213
  }
214
}
215
216
function Hash (onHash) {
217
  var buffers = []
218
  return pull.through(function (data) {
219
    buffers.push(typeof data === 'string'
220
      ? new Buffer(data, 'utf8')
221
      : data
222
    )
223
  }, function (err) {
224
    if (err && !onHash) throw err
225
    var b = buffers.length > 1 ? Buffer.concat(buffers) : buffers[0]
226
    var h = '&' + ssbKeys.hash(b)
227
    onHash && onHash(err, h)
228
  })
229
}
230