git ssb

1
'use strict';
2
3
var join      = require('path').join
4
var EventEmitter = require('events')
5
//var Obv       = require('obv')
6
7
var pull      = require('pull-stream')
8
var timestamp = require('monotonic-timestamp')
9
var explain   = require('explain-error')
10
//var createFeed = require('ssb-feed')
11
var ref       = require('ssb-ref')
12
var ssbKeys   = require('ssb-keys')
13
var Notify    = require('pull-notify')
14
15
var isFeedId = ref.isFeedId
16
var isMsgId  = ref.isMsgId
17
var isBlobId = ref.isBlobId
18
19
var u         = require('./util')
20
var stdopts   = u.options
21
var Format    = u.Format
22
//53 bit integer
23
var MAX_INT  = 0x1fffffffffffff
24
25
function isString (s) {
26
  return 'string' === typeof s
27
}
28
29
var isArray = Array.isArray
30
31
function isObject (o) {
32
  return o && 'object' === typeof o && !Array.isArray(o)
33
}
34
35
function getVMajor () {
36
  var version = require('./package.json').version
37
  return (version.split('.')[0])|0
38
}
39
40
function errorCB (err) {
41
  if(err) throw err
42
}
43
44
module.exports = function (_db, opts, keys, path) {
45
  path = path || _db.location
46
47
  keys = keys || ssbKeys.generate()
48
49
  var db = require('./db')(join(opts.path || path, 'flume'), keys, opts)
50
51
  //legacy database
52
  if(_db) require('./legacy')(_db, db)
53
  else db.ready.set(true)
54
55
  db.sublevel = function (a, b) {
56
    return _db.sublevel(a, b)
57
  }
58
59
  //UGLY HACK, but...
60
  //fairly sure that something up the stack expects ssb to be an event emitter.
61
  db.__proto__ = new EventEmitter()
62
63
  db.opts = opts
64
65
  var _get = db.get
66
67
  db.get = function (key, cb) {
68
    let isPrivate = false
69
    let unbox
70
    if (typeof key === 'object') {
71
      isPrivate = key.private === true
72
      unbox = key.unbox
73
      key = key.id
74
    }
75
76
    if(ref.isMsg(key)) {
77
      return db.keys.get(key, function (err, data) {
78
        if (err) return cb(err)
79
80
        if (isPrivate && unbox) {
81
          data = db.unbox(data, unbox)
82
        }
83
84
        let result
85
86
        if (isPrivate) {
87
          result = data.value
88
        } else {
89
          result = u.originalValue(data.value)
90
        }
91
92
        cb(null, result)
93
      })
94
    } else if(ref.isMsgLink(key)) {
95
      var link = ref.parseLink(key)
96
      return db.get({
97
        id: link.link,
98
        private: true,
99
        unbox: link.query.unbox.replace(/\s/g, '+')
100
      }, cb)
101
    } else if (Number.isInteger(key)) {
102
      _get(key, cb) // seq
103
    } else {
104
      throw new Error('ssb-db.get: key *must* be a ssb message id or a flume offset')
105
    }
106
  }
107
108
  db.add = function (msg, cb) {
109
    db.queue(msg, function (err, data) {
110
      if(err) cb(err)
111
      else db.flush(function () { cb(null, data) })
112
    })
113
  }
114
115
  db.createFeed = function (keys) {
116
    if(!keys) keys = ssbKeys.generate()
117
    function add (content, cb) {
118
      //LEGACY: hacks to support add as a continuable
119
      if(!cb)
120
        return function (cb) { add (content, cb) }
121
122
      db.append({content: content, keys: keys}, cb)
123
    }
124
    return {
125
      add: add, publish: add,
126
      id: keys.id, keys: keys
127
    }
128
  }
129
130
  db.createRawLogStream = function (opts) {
131
    return db.stream(opts)
132
  }
133
134
  //pull in the features that are needed to pass the tests
135
  //and that sbot, etc uses but are slow.
136
  require('./extras')(db, opts, keys)
137
138
  //writeStream - used in (legacy) replication.
139
  db.createWriteStream = function (cb) {
140
    cb = cb || errorCB
141
    return pull(
142
      pull.asyncMap(function (data, cb) {
143
        db.queue(data, function (err, msg) {
144
          if(err) {
145
            db.emit('invalid', err, msg)
146
          }
147
          setImmediate(cb)
148
        })
149
      }),
150
      pull.drain(null, function (err) {
151
        if(err) return cb(err)
152
        db.flush(cb)
153
      })
154
    )
155
  }
156
157
  //should be private
158
  db.createHistoryStream = db.clock.createHistoryStream
159
160
  //called with [id, seq] or "<id>:<seq>"
161
  db.getAtSequence = function (seqid, cb) {
162
    //will NOT expose private plaintext
163
    db.clock.get(isString(seqid) ? seqid.split(':') : seqid, function (err, value) {
164
      if(err) cb(err)
165
      else cb(null, u.originalData(value))
166
    })
167
  }
168
169
  db.getVectorClock = function (_, cb) {
170
    if(!cb) cb = _
171
    db.last.get(function (err, h) {
172
      if(err) return cb(err)
173
      var clock = {}
174
      for(var k in h)
175
        clock[k] = h[k].sequence
176
      cb(null, clock)
177
    })
178
  }
179
180
  if(_db) {
181
    var close = db.close
182
    db.close = function (cb) {
183
      var n = 2
184
      _db.close(next); close(next)
185
186
      function next (err) {
187
        if(err && n>0) {
188
          n = -1
189
          return cb(err)
190
        }
191
        if(--n) return
192
        cb()
193
      }
194
    }
195
  }
196
197
  return db
198
}
199