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