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
      caps: config.caps,
74
      friends: config.friends
75
    }
76
77
    createClient(keys, opts, function (err, _sbot) {
78
      if (err) {
79
        return notify(err)
80
      }
81
82
      sbot = _sbot
83
      sbot.on('closed', function () {
84
        sbot = null
85
        connection.set(null)
86
        notify(new Error('closed'))
87
      })
88
89
      connection.set(sbot)
90
      notify()
91
    })
92
  })
93
94
  var internal = {
95
    getLatest: rec.async(function (id, cb) {
96
      sbot.getLatest(id, cb)
97
    }),
98
    add: rec.async(function (msg, cb) {
99
      sbot.add(msg, cb)
100
    })
101
  }
102
103
  watch(connection, (sbot) => {
104
    if (sbot) {
105
      sbot.gossip.peers((err, peers) => {
106
        if (err) return console.error(err)
107
        connectedPeers.set(peers.filter(x => x.state === 'connected').map(x => x.key))
108
        localPeers.set(peers.filter(x => x.source === 'local').map(x => x.key))
109
      })
110
      pull(
111
        sbot.gossip.changes(),
112
        pull.drain(data => {
113
          if (data.peer) {
114
            if (data.type === 'remove' || data.type === 'disconnect') {
115
              connectedPeers.delete(data.peer.key)
116
              localPeers.delete(data.peer.key)
117
            } else {
118
              if (data.peer.source === 'local') {
119
                localPeers.add(data.peer.key)
120
              }
121
              if (data.peer.state === 'connected') {
122
                connectedPeers.add(data.peer.key)
123
              } else {
124
                connectedPeers.delete(data.peer.key)
125
              }
126
            }
127
          }
128
        })
129
      )
130
    }
131
  })
132
133
  var feed = createFeed(internal, keys, {remote: true})
134
135
  return {
136
    sbot: {
137
      sync: {
138
        cache: () => cache
139
      },
140
      async: {
141
        get: rec.async(function (key, cb) {
142
          if (typeof cb !== 'function') {
143
            throw new Error('cb must be function')
144
          }
145
          if (cache[key]) cb(null, cache[key])
146
          else {
147
            sbot.get(key, function (err, value) {
148
              if (err) return cb(err)
149
              runHooks({key, value})
150
              cb(null, value)
151
            })
152
          }
153
        }),
154
        publish: rec.async((content, cb) => {
155
          if (content.recps) {
156
            content = ssbKeys.box(content, content.recps.map(e => {
157
              return ref.isFeed(e) ? e : e.link
158
            }))
159
          } else {
160
            var flatContent = flat(content)
161
            Object.keys(flatContent).forEach(key => {
162
              var val = flatContent[key]
163
              if (ref.isBlob(val)) {
164
                sbot.blobs.push(val, err => {
165
                  if (err) console.error(err)
166
                })
167
              }
168
            })
169
          }
170
171
          if (sbot) {
172
            // instant updating of interface (just incase sbot is busy)
173
            runHooks({
174
              publishing: true,
175
              timestamp: Date.now(),
176
              value: {
177
                timestamp: Date.now(),
178
                author: keys.id,
179
                content
180
              }
181
            })
182
          }
183
184
          feed.add(content, (err, msg) => {
185
            if (err) console.error(err)
186
            else if (!cb) console.log(msg)
187
            cb && cb(err, msg)
188
          })
189
        }),
190
        addBlob: rec.async((stream, cb) => {
191
          return pull(
192
            stream,
193
            sbot.blobs.add(cb)
194
          )
195
        }),
196
        gossipConnect: rec.async(function (opts, cb) {
197
          sbot.gossip.connect(opts, cb)
198
        }),
199
        friendsGet: rec.async(function (opts, cb) {
200
          sbot.friends.get(opts, cb)
201
        })
202
      },
203
      pull: {
204
        backlinks: rec.source(query => {
205
          return sbot.backlinks.read(query)
206
        }),
207
        userFeed: rec.source(opts => {
208
          return sbot.createUserStream(opts)
209
        }),
210
        messagesByType: rec.source(opts => {
211
          return sbot.messagesByType(opts)
212
        }),
213
        feed: rec.source(function (opts) {
214
          return pull(
215
            sbot.createFeedStream(opts),
216
            pull.through(runHooks)
217
          )
218
        }),
219
        log: rec.source(opts => {
220
          return pull(
221
            sbot.createLogStream(opts),
222
            pull.through(runHooks)
223
          )
224
        }),
225
        links: rec.source(function (query) {
226
          return sbot.links(query)
227
        }),
228
        stream: function (fn) {
229
          var stream = defer.source()
230
          onceTrue(connection, function (connection) {
231
            stream.resolve(fn(connection))
232
          })
233
          return stream
234
        }
235
      },
236
      obs: {
237
        connectionStatus: (listener) => connectionStatus(listener),
238
        connection,
239
        connectedPeers: () => connectedPeers,
240
        localPeers: () => localPeers
241
      }
242
    }
243
  }
244
245
  // scoped
246
247
  function runHooks (msg) {
248
    if (msg.publishing) {
249
      api.sbot.hook.publish(msg)
250
    } else if (!cache[msg.key]) {
251
      // cache[msg.key] = msg.value
252
      // api.sbot.hook.feed(msg)
253
    }
254
  }
255
}
256