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