git ssb

1
var pull = require('pull-stream')
2
var defer = require('pull-defer')
3
var { Value, onceTrue, watch, Set: MutantSet } = require('mutant')
4
var ref = require('ssb-ref')
5
var Reconnect = require('pull-reconnect')
6
var createClient = require('ssb-client')
7
var createFeed = require('ssb-feed')
8
var nest = require('depnest')
9
var ssbKeys = require('ssb-keys')
10
var flat = require('flat')
11
12
exports.needs = nest({
13
  'config.sync.load': 'first',
14
  'keys.sync.load': 'first',
15
  'sbot.obs.connectionStatus': 'first',
16
  'sbot.hook.publish': 'map'
17
})
18
19
exports.gives = {
20
  sbot: {
21
    sync: {
22
      cache: true
23
    },
24
    async: {
25
      get: true,
26
      publish: true,
27
      addBlob: true,
28
      gossipConnect: true,
29
      friendsGet: true
30
    },
31
    pull: {
32
      log: true,
33
      userFeed: true,
34
      messagesByType: true,
35
      feed: true,
36
      links: true,
37
      backlinks: true,
38
      stream: true
39
    },
40
    obs: {
41
      connectionStatus: true,
42
      connection: true,
43
      connectedPeers: true,
44
      localPeers: true
45
    }
46
  }
47
}
48
49
exports.create = function (api) {
50
  const config = api.config.sync.load()
51
  const keys = api.keys.sync.load()
52
  var cache = {}
53
54
  var sbot = null
55
  var connection = Value()
56
  var connectionStatus = Value()
57
  var connectedPeers = MutantSet()
58
  var localPeers = MutantSet()
59
60
  var rec = Reconnect(function (isConn) {
61
    function notify (value) {
62
      isConn(value); connectionStatus.set(value)
63
    }
64
65
    var opts = {
66
      path: config.path,
67
      remote: config.remote,
68
      host: config.host,
69
      port: config.port,
70
      key: config.key,
71
      appKey: config.caps.shs,
72
      timers: config.timers
73
    }
74
75
    createClient(keys, opts, function (err, _sbot) {
76
      if (err) {
77
        return notify(err)
78
      }
79
80
      sbot = _sbot
81
      sbot.on('closed', function () {
82
        sbot = null
83
        connection.set(null)
84
        notify(new Error('closed'))
85
      })
86
87
      connection.set(sbot)
88
      notify()
89
    })
90
  })
91
92
  var internal = {
93
    getLatest: rec.async(function (id, cb) {
94
      sbot.getLatest(id, cb)
95
    }),
96
    add: rec.async(function (msg, cb) {
97
      sbot.add(msg, cb)
98
    })
99
  }
100
101
  watch(connection, (sbot) => {
102
    if (sbot) {
103
      sbot.gossip.peers((err, peers) => {
104
        if (err) return console.error(err)
105
        connectedPeers.set(peers.filter(x => x.state === 'connected').map(x => x.key))
106
        localPeers.set(peers.filter(x => x.source === 'local').map(x => x.key))
107
      })
108
      pull(
109
        sbot.gossip.changes(),
110
        pull.drain(data => {
111
          if (data.peer) {
112
            if (data.type === 'remove') {
113
              connectedPeers.delete(data.peer.key)
114
              localPeers.delete(data.peer.key)
115
            } else {
116
              if (data.peer.source === 'local') {
117
                localPeers.add(data.peer.key)
118
              }
119
              if (data.peer.state === 'connected') {
120
                connectedPeers.add(data.peer.key)
121
              } else {
122
                connectedPeers.delete(data.peer.key)
123
              }
124
            }
125
          }
126
        })
127
      )
128
    }
129
  })
130
131
  var feed = createFeed(internal, keys, {remote: true})
132
133
  return {
134
    sbot: {
135
      sync: {
136
        cache: () => cache
137
      },
138
      async: {
139
        get: rec.async(function (key, cb) {
140
          if (typeof cb !== 'function') {
141
            throw new Error('cb must be function')
142
          }
143
          if (cache[key]) cb(null, cache[key])
144
          else {
145
            sbot.get(key, function (err, value) {
146
              if (err) return cb(err)
147
              runHooks({key, value})
148
              cb(null, value)
149
            })
150
          }
151
        }),
152
        publish: rec.async((content, cb) => {
153
          if (content.recps) {
154
            content = ssbKeys.box(content, content.recps.map(e => {
155
              return ref.isFeed(e) ? e : e.link
156
            }))
157
          } else {
158
            var flatContent = flat(content)
159
            Object.keys(flatContent).forEach(key => {
160
              var val = flatContent[key]
161
              if (ref.isBlob(val)) {
162
                sbot.blobs.push(val, err => {
163
                  if (err) console.error(err)
164
                })
165
              }
166
            })
167
          }
168
169
          if (sbot) {
170
            // instant updating of interface (just incase sbot is busy)
171
            runHooks({
172
              publishing: true,
173
              timestamp: Date.now(),
174
              value: {
175
                timestamp: Date.now(),
176
                author: keys.id,
177
                content
178
              }
179
            })
180
          }
181
182
          feed.add(content, (err, msg) => {
183
            if (err) console.error(err)
184
            else if (!cb) console.log(msg)
185
            cb && cb(err, msg)
186
          })
187
        }),
188
        addBlob: rec.async((stream, cb) => {
189
          return pull(
190
            stream,
191
            sbot.blobs.add(cb)
192
          )
193
        }),
194
        gossipConnect: rec.async(function (opts, cb) {
195
          sbot.gossip.connect(opts, cb)
196
        }),
197
        friendsGet: rec.async(function (opts, cb) {
198
          sbot.friends.get(opts, cb)
199
        })
200
      },
201
      pull: {
202
        backlinks: rec.source(query => {
203
          return sbot.backlinks.read(query)
204
        }),
205
        userFeed: rec.source(opts => {
206
          return sbot.createUserStream(opts)
207
        }),
208
        messagesByType: rec.source(opts => {
209
          return sbot.messagesByType(opts)
210
        }),
211
        feed: rec.source(function (opts) {
212
          return pull(
213
            sbot.createFeedStream(opts),
214
            pull.through(runHooks)
215
          )
216
        }),
217
        log: rec.source(opts => {
218
          return pull(
219
            sbot.createLogStream(opts),
220
            pull.through(runHooks)
221
          )
222
        }),
223
        links: rec.source(function (query) {
224
          return sbot.links(query)
225
        }),
226
        stream: function (fn) {
227
          var stream = defer.source()
228
          onceTrue(connection, function (connection) {
229
            stream.resolve(fn(connection))
230
          })
231
          return stream
232
        }
233
      },
234
      obs: {
235
        connectionStatus: (listener) => connectionStatus(listener),
236
        connection,
237
        connectedPeers: () => connectedPeers,
238
        localPeers: () => localPeers
239
      }
240
    }
241
  }
242
243
  // scoped
244
245
  function runHooks (msg) {
246
    if (msg.publishing) {
247
      api.sbot.hook.publish(msg)
248
    } else if (!cache[msg.key]) {
249
      // cache[msg.key] = msg.value
250
      // api.sbot.hook.feed(msg)
251
    }
252
  }
253
}
254