sbot/api/index.js

508 lines
14 KiB
JavaScript
Raw Normal View History

var fs = require('fs')
var pull = require('pull-stream')
var multicb = require('multicb')
var pl = require('pull-level')
var pushable = require('pull-pushable')
var paramap = require('pull-paramap')
var cat = require('pull-cat')
var Notify = require('pull-notify')
var toPull = require('stream-to-pull-stream')
var ref = require('ssb-ref')
var pathlib = require('path')
var u = require('./util')
exports.name = 'patchwork'
exports.version = '1.0.0'
exports.manifest = require('./manifest')
exports.permissions = require('./permissions')
exports.init = function (sbot, opts) {
var api = {}
var phoenixdb = sbot.sublevel('patchwork')
var db = {
isread: phoenixdb.sublevel('isread'),
subscribed: phoenixdb.sublevel('subscribed')
}
var state = {
// indexes (lists of {key:, ts:})
mymsgs: [],
home: u.index(), // also has `.isread`
inbox: u.index(), // also has `.isread` and `.author`
votes: u.index(), // also has `.isread`, `.vote`, and `.votemsg`
myvotes: u.index(), // also has `.vote`
follows: u.index(), // also has `.isread` and `.following`
// views
profiles: {},
sites: {},
names: {}, // ids -> names
ids: {}, // names -> ids
actionItems: {}
}
var processor = require('./processor')(sbot, db, state, emit)
pull(pl.read(sbot.sublevel('log'), { live: true, onSync: onPrehistorySync }), pull.drain(processor))
// track sync state
// - processor does async processing for each message that comes in
// - awaitSync() waits for that processing to finish
// - pinc() on message arrival, pdec() on message processed
// - nP === 0 => all messages processed
var nP = 0, syncCbs = []
function awaitSync (cb) {
if (nP > 0)
syncCbs.push(cb)
else cb()
}
state.pinc = function () { nP++ }
state.pdec = function () {
nP--
if (nP === 0) {
syncCbs.forEach(function (cb) { cb() })
syncCbs.length = 0
}
}
var isPreHistorySynced = false // track so we dont emit events for old messages
// grab for history sync
state.pinc()
function onPrehistorySync () {
console.log('Log history read...')
// when all current items finish, consider prehistory synced (and start emitting)
awaitSync(function () {
console.log('Indexes generated')
isPreHistorySynced = true
})
// release
state.pdec()
}
// events stream
var notify = Notify()
function emit (type, data) {
if (!isPreHistorySynced)
return
var e = data || {}
e.type = type
if (e.type == 'index-change') {
api.getIndexCounts(function (err, counts) {
e.total = counts[e.index]
e.unread = counts[e.index+'Unread']
notify(e)
})
} else
notify(e)
}
// getters
api.createEventStream = function () {
return notify.listen()
}
api.getPaths = function (cb) {
cb(null, {
site: pathlib.join(opts.path, 'publish')
})
}
api.getMyProfile = function (cb) {
awaitSync(function () {
api.getProfile(sbot.id, cb)
})
}
function isInboxFriend (row) {
if (row.author == sbot.id) return true
var p = state.profiles[sbot.id]
if (!p) return false
return p.assignedTo[row.author] && p.assignedTo[row.author].following
}
api.getIndexCounts = function (cb) {
awaitSync(function () {
cb(null, {
inbox: state.inbox.rows.filter(isInboxFriend).length,
inboxUnread: state.inbox.filter(function (row) { return isInboxFriend(row) && row.author != sbot.id && !row.isread }).length,
votes: state.votes.filter(function (row) { return row.vote > 0 }).length,
votesUnread: state.votes.filter(function (row) { return row.vote > 0 && !row.isread }).length,
follows: state.follows.filter(function (row) { return row.following }).length,
followsUnread: state.follows.filter(function (row) { return row.following && !row.isread }).length,
home: state.home.rows.length
})
})
}
api.createInboxStream = indexStreamFn(state.inbox, function (row) {
if (!isInboxFriend(row)) return false
return row.key
})
api.createVoteStream = indexStreamFn(state.votes, function (row) {
if (row.vote <= 0) return false
return row.votemsg
})
api.createMyvoteStream = indexStreamFn(state.myvotes, function (row) {
if (row.vote <= 0) return false
return row.key
})
api.createFollowStream = indexStreamFn(state.follows)
api.createHomeStream = indexStreamFn(state.home)
function indexMarkRead (indexname, key, keyname) {
if (Array.isArray(key)) {
key.forEach(function (k) {
indexMarkRead(indexname, k, keyname)
})
return
}
var index = state[indexname]
var row = index.find(key, keyname)
if (row) {
var wasread = row.isread
row.isread = true
if (!wasread)
emit('index-change', { index: indexname })
}
}
function indexMarkUnread (indexname, key, keyname) {
if (Array.isArray(key)) {
key.forEach(function (k) {
indexMarkUnread(indexname, k, keyname)
})
return
}
var index = state[indexname]
var row = index.find(key, keyname)
if (row) {
var wasread = row.isread
row.isread = false
if (wasread)
emit('index-change', { index: indexname })
}
}
api.markRead = function (key, cb) {
indexMarkRead('inbox', key)
indexMarkRead('votes', key, 'votemsg')
indexMarkRead('follows', key)
if (Array.isArray(key))
db.isread.batch(key.map(function (k) { return { type: 'put', key: k, value: 1 }}), cb)
else
db.isread.put(key, 1, cb)
}
api.markUnread = function (key, cb) {
indexMarkUnread('inbox', key)
indexMarkUnread('votes', key, 'votemsg')
indexMarkUnread('follows', key)
if (Array.isArray(key))
db.isread.batch(key.map(function (k) { return { type: 'del', key: k }}), cb)
else
db.isread.del(key, cb)
}
api.toggleRead = function (key, cb) {
api.isRead(key, function (err, v) {
if (!v) {
api.markRead(key, function (err) {
cb(err, true)
})
} else {
api.markUnread(key, function (err) {
cb(err, false)
})
}
})
}
api.isRead = function (key, cb) {
if (Array.isArray(key)) {
var done = multicb({ pluck: 1 })
key.forEach(function (k, i) {
var cb = done()
db.isread.get(k, function (err, v) { cb(null, !!v) })
})
done(cb)
} else {
db.isread.get(key, function (err, v) {
cb && cb(null, !!v)
})
}
}
api.subscribe = function (key, cb) {
db.subscribed.put(key, 1, cb)
}
api.unsubscribe = function (key, cb) {
db.subscribed.del(key, cb)
}
api.toggleSubscribed = function (key, cb) {
api.isSubscribed(key, function (err, v) {
if (!v) {
api.subscribe(key, function (err) {
cb(err, true)
})
} else {
api.unsubscribe(key, function (err) {
cb(err, false)
})
}
})
}
api.isSubscribed = function (key, cb) {
db.subscribed.get(key, function (err, v) {
cb && cb(null, !!v)
})
}
api.addFileToBlobs = function (path, cb) {
pull(
toPull.source(fs.createReadStream(path)),
sbot.blobs.add(function (err, hash) {
if (err)
cb(err)
else {
var ext = pathlib.extname(path)
if (ext == '.png' || ext == '.jpg' || ext == '.jpeg') {
var res = getImgDim(path)
res.hash = hash
cb(null, res)
} else
cb(null, { hash: hash })
}
})
)
}
api.saveBlobToFile = function (hash, path, cb) {
pull(
sbot.blobs.get(hash),
toPull.sink(fs.createWriteStream(path), cb)
)
}
function getImgDim (path) {
var NativeImage = require('native-image')
var ni = NativeImage.createFromPath(path)
return ni.getSize()
}
var lookupcodeRegex = /(@[a-z0-9\/\+\=]+\.[a-z0-9]+)(?:\[via\])?(.+)?/i
api.useLookupCode = function (code) {
var eventPush = pushable()
// parse and validate the code
var id, addrs
var parts = lookupcodeRegex.exec(code)
var valid = true
if (parts) {
id = parts[1]
addrs = (parts[2]) ? parts[2].split(',') : []
// validate id
if (!ref.isFeedId(id))
valid = false
// parse addresses
addrs = addrs
.map(function (addr) {
addr = addr.split(':')
if (addr.length === 3)
return { host: addr[0], port: +addr[1], key: addr[2] }
})
.filter(Boolean)
} else
valid = false
if (!valid) {
eventPush.push({ type: 'error', message: 'Invalid lookup code' })
eventPush.end()
return eventPush
}
// begin the search!
search(addrs.concat(sbot.gossip.peers()))
function search (peers) {
var peer = peers.pop()
if (!peer)
return eventPush.end()
// connect to the peer
eventPush.push({ type: 'connecting', addr: peer })
sbot.connect(peer, function (err, rpc) {
if (err) {
eventPush.push({ type: 'error', message: 'Failed to connect', err: err })
return search(peers)
}
// try a sync
sync(rpc, function (err, seq) {
if (seq > 0) {
// success!
eventPush.push({ type: 'finished', seq: seq })
eventPush.end()
} else
search(peers) // try next
})
})
}
function sync (rpc, cb) {
// fetch the feed
var seq
eventPush.push({ type: 'syncing', id: id })
pull(
rpc.createHistoryStream({ id: id, keys: false }),
pull.through(function (msg) {
seq = msg.sequence
}),
sbot.createWriteStream(function (err) {
cb(err, seq)
})
)
}
return eventPush
}
api.getSite = function (id, cb) {
awaitSync(function () { cb(null, state.sites[id]) })
}
var sitePathRegex = /(@.*\.ed25519)(.*)/
api.getSiteLink = function (url, cb) {
awaitSync(function () {
// parse url
var parts = sitePathRegex.exec(url)
if (!parts) {
var err = new Error('Not found')
err.notFound = true
return cb(err)
}
var pid = parts[1]
var path = parts[2]
if (path.charAt(0) == '/')
path = path.slice(1) // skip the preceding slash
if (!path)
path = 'index.html' // default asset
// lookup the link
var link = (state.sites[pid]) ? state.sites[pid][path] : null
if (!link) {
var err = new Error('Not found')
err.notFound = true
return cb(err)
}
cb(null, link)
})
}
api.getProfile = function (id, cb) {
awaitSync(function () { cb(null, state.profiles[id]) })
}
api.getAllProfiles = function (cb) {
awaitSync(function () { cb(null, state.profiles) })
}
api.getNamesById = function (cb) {
awaitSync(function () { cb(null, state.names) })
}
api.getName = function (id, cb) {
awaitSync(function () { cb(null, state.names[id]) })
}
api.getIdsByName = function (cb) {
awaitSync(function () { cb(null, state.ids) })
}
api.getActionItems = function (cb) {
awaitSync(function () { cb(null, state.actionItems) })
}
// helper to get an option off an opt function (avoids the `opt || {}` pattern)
function o (opts, k, def) {
return opts && opts[k] !== void 0 ? opts[k] : def
}
// helper to get messages from an index
function indexStreamFn (index, getkey) {
return function (opts) {
// emulate the `ssb.createFeedStream` interface
var lt = o(opts, 'lt')
var lte = o(opts, 'lte')
var gt = o(opts, 'gt')
var gte = o(opts, 'gte')
var limit = o(opts, 'limit')
// lt, lte, gt, gte should look like:
// [msg.value.timestamp, msg.value.author]
// helper to create emittable rows
function lookup (row) {
if (!row) return
var key = (getkey) ? getkey(row) : row.key
if (key) {
var rowcopy = { key: key }
for (var k in row) { // copy index attrs into rowcopy
if (!rowcopy[k]) rowcopy[k] = row[k]
}
return rowcopy
}
}
// helper to fetch rows
function fetch (row, cb) {
sbot.get(row.key, function (err, value) {
// if (err) {
// suppress this error
// the message isnt in the local cache (yet)
// but it got into the index, likely due to a link
// instead of an error, we'll put a null there to indicate the gap
// }
row.value = value
cb(null, row)
})
}
// readstream
var readPush = pushable()
var read = pull(readPush, paramap(fetch))
// await sync, then emit the reads
awaitSync(function () {
var added = 0
for (var i=0; i < index.rows.length; i++) {
var row = index.rows[i]
if (limit && added >= limit)
break
// we're going to only look at timestamp, because that's all that phoenix cares about
var invalid = !!(
(lt && row.ts >= lt[0]) ||
(lte && row.ts > lte[0]) ||
(gt && row.ts <= gt[0]) ||
(gte && row.ts < gte[0])
)
if (invalid)
continue
var r = lookup(row)
if (r) {
readPush.push(r)
added++
}
}
readPush.end()
})
if (opts && opts.live) {
// live stream, concat the live-emitter on the end
index.on('add', onadd)
var livePush = pushable(function () { index.removeListener('add', onadd) })
function onadd (row) { livePush.push(lookup(row)) }
var live = pull(livePush, paramap(fetch))
return cat([read, live])
}
return read
}
}
return api
}