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
    },
26
    pull: {
27
      log: true,
28
      userFeed: true,
29
      query: true,
30
      feed: true
31
    },
32
    obs: {
33
      connectionStatus: true,
34
      connectedPeers: true,
35
      localPeers: true
36
    }
37
  }
38
}
39
40
exports.create = function (api) {
41
  const config = api.config.sync.load()
42
  const keys = api.keys.sync.load()
43
  var cache = {}
44
45
  var sbot = null
46
  var connectionStatus = Value()
47
  var connectedPeers = Value([])
48
  var localPeers = Value([])
49
50
  setInterval(refreshPeers, 5e3)
51
52
  var rec = {
53
    sync: () => {},
54
    async: () => {},
55
    source: () => {}
56
  }
57
58
  var rec = Reconnect(function (isConn) {
59
    function notify (value) {
60
      isConn(value); connectionStatus.set(value)
61
    }
62
63
    createClient(keys, {
64
      manifest: require('./manifest.json'),
65
      remote: config.remote,
66
      caps: config.caps
67
    }, function (err, _sbot) {
68
      if (err) {
69
        return notify(err)
70
      }
71
72
      sbot = _sbot
73
      sbot.on('closed', function () {
74
        sbot = null
75
        notify(new Error('closed'))
76
      })
77
78
      notify()
79
    })
80
  })
81
82
  var internal = {
83
    getLatest: rec.async(function (id, cb) {
84
      sbot.getLatest(id, cb)
85
    }),
86
    add: rec.async(function (msg, cb) {
87
      sbot.add(msg, cb)
88
    })
89
  }
90
91
  var feed = createFeed(internal, keys, {remote: true})
92
93
  return {
94
    sbot: {
95
      sync: {
96
        cache: () => cache
97
      },
98
      async: {
99
        get: rec.async(function (key, cb) {
100
          if (typeof cb !== 'function') {
101
            throw new Error('cb must be function')
102
          }
103
          if (cache[key]) cb(null, cache[key])
104
          else {
105
            sbot.get(key, function (err, value) {
106
              if (err) return cb(err)
107
              runHooks({key, value})
108
              cb(null, value)
109
            })
110
          }
111
        }),
112
        publish: rec.async((content, cb) => {
113
          if (content.recps) {
114
            content = ssbKeys.box(content, content.recps.map(e => {
115
              return ref.isFeed(e) ? e : e.link
116
            }))
117
          } else if (content.mentions) {
118
            content.mentions.forEach(mention => {
119
              if (ref.isBlob(mention.link)) {
120
                sbot.blobs.push(mention.link, err => {
121
                  if (err) console.error(err)
122
                })
123
              }
124
            })
125
          }
126
127
          feed.add(content, (err, msg) => {
128
            if (err) console.error(err)
129
            else if (!cb) console.log(msg)
130
            cb && cb(err, msg)
131
          })
132
        })
133
      },
134
      pull: {
135
        query: rec.source(query => {
136
          return sbot.query.read(query)
137
        }),
138
        userFeed: rec.source(opts => {
139
          return sbot.createUserStream(opts)
140
        }),
141
        feed: rec.source(function (opts) {
142
          return pull(
143
            sbot.createFeedStream(opts),
144
            pull.through(runHooks)
145
          )
146
        }),
147
        log: rec.source(opts => {
148
          return pull(
149
            sbot.createLogStream(opts),
150
            pull.through(runHooks)
151
          )
152
        })
153
      },
154
      obs: {
155
        connectionStatus: (listener) => connectionStatus(listener),
156
        connectedPeers: () => connectedPeers,
157
        localPeers: () => localPeers
158
      }
159
    }
160
  }
161
162
  // scoped
163
164
  function runHooks (msg) {
165
    if (!cache[msg.key]) {
166
      cache[msg.key] = msg.value
167
      api.sbot.hook.feed(msg)
168
    }
169
  }
170
171
  function refreshPeers () {
172
    if (connectionStatus() === null) { // wat!?
173
      sbot.gossip.peers((err, peers) => {
174
        if (err) throw console.log(err)
175
        connectedPeers.set(peers.filter(x => x.state === 'connected').map(x => x.key))
176
      })
177
      if (sbot.local && sbot.local.list) {
178
        sbot.local.list((err, keys) => {
179
          if (!err) {
180
            localPeers.set(keys)
181
          }
182
        })
183
      }
184
    }
185
  }
186
}
187