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
9
var cache = CACHE = {}
10
11
exports.needs = nest({
12
  'config.sync.load': 'first',
13
  'keys.sync.load': 'first',
14
  'sbot.obs.connectionStatus': 'first'
15
})
16
17
exports.gives = {
18
  sbot: {
19
    async: {
20
      get: true,
21
      publish: true,
22
    },
23
    pull: {
24
      log: true,
25
      userFeed: true,
26
      query: true,
27
    },
28
    obs: {
29
      connectionStatus: true
30
    }
31
  }
32
}
33
34
exports.create = function (api) {
35
  const config = api.config.sync.load()
36
  const keys = api.keys.sync.load()
37
38
  var sbot = null
39
  var connectionStatus = Value()
40
41
  var rec = {
42
    sync: () => {},
43
    async: () => {},
44
    source: () => {}
45
  }
46
47
  var rec = Reconnect(function (isConn) {
48
    function notify (value) {
49
      isConn(value); connectionStatus.set(value)
50
    }
51
52
    createClient(keys, {
53
      manifest: require('./manifest.json'),
54
      remote: config.remote,
55
      caps: config.caps
56
    }, function (err, _sbot) {
57
      if (err) {
58
        return notify(err)
59
      }
60
61
      sbot = _sbot
62
      sbot.on('closed', function () {
63
        sbot = null
64
        notify(new Error('closed'))
65
      })
66
67
      notify()
68
    })
69
  })
70
71
  var internal = {
72
    getLatest: rec.async(function (id, cb) {
73
      sbot.getLatest(id, cb)
74
    }),
75
    add: rec.async(function (msg, cb) {
76
      sbot.add(msg, cb)
77
    })
78
  }
79
80
  var feed = createFeed(internal, keys, {remote: true})
81
82
  return {
83
    sbot: {
84
      async: {
85
        get: rec.async(function (key, cb) {
86
          if (typeof cb !== 'function') {
87
            throw new Error('cb must be function')
88
          }
89
          if (CACHE[key]) cb(null, CACHE[key])
90
          else {
91
            sbot.get(key, function (err, value) {
92
              if (err) return cb(err)
93
              cb(null, CACHE[key] = value)
94
            })
95
          }
96
        }),
97
        publish: rec.async((content, cb) => {
98
          if (content.recps) {
99
            content = ssbKeys.box(content, content.recps.map(e => {
100
              return ref.isFeed(e) ? e : e.link
101
            }))
102
          } else if (content.mentions) {
103
            content.mentions.forEach(mention => {
104
              if (ref.isBlob(mention.link)) {
105
                sbot.blobs.push(mention.link, err => {
106
                  if (err) console.error(err)
107
                })
108
              }
109
            })
110
          }
111
112
          feed.add(content, (err, msg) => {
113
            if (err) console.error(err)
114
            else if (!cb) console.log(msg)
115
            cb && cb(err, msg)
116
          })
117
        })
118
      },
119
      pull: {
120
        query: rec.source(query => {
121
          return sbot.query.read(query)
122
        }),
123
        userFeed: rec.source(opts => {
124
          return sbot.createUserStream(opts)
125
        }),
126
        log: rec.source(opts => {
127
          return pull(
128
            sbot.createLogStream(opts),
129
            pull.through(e => {
130
              CACHE[e.key] = CACHE[e.key] || e.value
131
            })
132
          )
133
        })
134
      },
135
      obs: {
136
        connectionStatus: (listener) => connectionStatus(listener)
137
      }
138
    }
139
  }
140
}
141