Files: 04ed909259996b33de1dbd7bfdf406cef94b88dd / sbot / roots.js
8964 bytesRaw
1 | |
2 | var pull = require('pull-stream') |
3 | var FlumeViewLevel = require('flumeview-level') |
4 | var pullCat = require('pull-cat') |
5 | var HLRU = require('hashlru') |
6 | var extend = require('xtend') |
7 | var normalizeChannel = require('ssb-ref').normalizeChannel |
8 | var Defer = require('pull-defer') |
9 | |
10 | // HACK: pull it out of patchcore |
11 | var getRoot = require('patchcore/message/sync/root').create().message.sync.root |
12 | var getTimestamp = require('patchcore/message/sync/timestamp').create().message.sync.timestamp |
13 | |
14 | module.exports = function (ssb, config) { |
15 | var create = FlumeViewLevel(2, function (msg, seq) { |
16 | var result = [ |
17 | [getTimestamp(msg), getRoot(msg) || msg.key] |
18 | ] |
19 | return result |
20 | }) |
21 | |
22 | var index = ssb._flumeUse('patchwork-roots', create) |
23 | |
24 | // cache mostly just to avoid reading the same roots over and over again |
25 | // not really big enough for multiple refresh cycles |
26 | var cache = HLRU(100) |
27 | |
28 | return { |
29 | latest: function ({ids = [ssb.id], onlySubscribedChannels = false}) { |
30 | var stream = Defer.source() |
31 | getFilter((err, filter) => { |
32 | if (err) return stream.abort(err) |
33 | stream.resolve(pull( |
34 | index.read({old: false}), |
35 | |
36 | // BUMP FILTER |
37 | pull.filter(item => { |
38 | if (filter && item.value && item.value) { |
39 | var filterResult = filter(ids, item.value) |
40 | if (filterResult) { |
41 | item.value.filterResult = filterResult |
42 | return true |
43 | } |
44 | } |
45 | }), |
46 | |
47 | // LOOKUP AND ADD ROOTS |
48 | LookupRoots(), |
49 | |
50 | // FILTER ROOTS |
51 | pull.filter(item => { |
52 | var root = item.root || item |
53 | var isPrivate = root.value && typeof root.value.content === 'string' |
54 | |
55 | if (filter && root && root.value && !isPrivate) { |
56 | var filterResult = filter(ids, root) |
57 | if (checkReplyForcesDisplay(item) || shouldShow(filterResult, {onlySubscribedChannels})) { |
58 | root.filterResult = filterResult |
59 | return true |
60 | } |
61 | } |
62 | }) |
63 | )) |
64 | }) |
65 | return stream |
66 | }, |
67 | |
68 | read: function ({ids = [ssb.id], reverse, limit, lt, gt, onlySubscribedChannels = false}) { |
69 | var opts = {reverse, old: true} |
70 | |
71 | // handle markers passed in to lt / gt |
72 | if (lt && typeof lt.timestamp === 'number') lt = lt.timestamp |
73 | if (gt && typeof gt.timestamp === 'number') gt = gt.timestamp |
74 | if (typeof lt === 'number') opts.lt = [lt] |
75 | if (typeof gt === 'number') opts.gt = [gt] |
76 | |
77 | var seen = new Set() |
78 | var included = new Set() |
79 | var marker = {marker: true, timestamp: null} |
80 | |
81 | var stream = Defer.source() |
82 | |
83 | getFilter((err, filter) => { |
84 | if (err) return stream.abort(err) |
85 | stream.resolve(pull( |
86 | // READ ROOTS INDEX |
87 | index.read(opts), |
88 | |
89 | // BUMP FILTER |
90 | pull.filter(item => { |
91 | // keep track of latest timestamp |
92 | marker.timestamp = item.key[0] |
93 | |
94 | if (filter && item.value && item.value) { |
95 | var filterResult = filter(ids, item.value) |
96 | if (filterResult) { |
97 | item.value.filterResult = filterResult |
98 | return true |
99 | } |
100 | } |
101 | }), |
102 | |
103 | // LOOKUP AND ADD ROOTS |
104 | LookupRoots(), |
105 | |
106 | // FILTER ROOTS |
107 | pull.filter(item => { |
108 | var root = item.root || item |
109 | var isPrivate = root.value && typeof root.value.content === 'string' |
110 | |
111 | // skip this item if it has already been included |
112 | if (!included.has(root.key) && filter && root && root.value && !isPrivate) { |
113 | if (checkReplyForcesDisplay(item)) { // include this item if it has matching tags or the author is you |
114 | // update filter result so that we can display the correct bump message |
115 | root.filterResult = extend(item.filterResult, {forced: true}) |
116 | included.add(root.key) |
117 | return true |
118 | } else if (!seen.has(root.key)) { |
119 | seen.add(root.key) |
120 | var filterResult = filter(ids, root) |
121 | if (shouldShow(filterResult, {onlySubscribedChannels})) { |
122 | root.filterResult = filterResult |
123 | included.add(root.key) |
124 | return true |
125 | } |
126 | } |
127 | } |
128 | }), |
129 | |
130 | // MAP ROOT ITEMS |
131 | pull.map(item => { |
132 | var root = item.root || item |
133 | return root |
134 | }) |
135 | )) |
136 | }) |
137 | |
138 | // TRUNCATE |
139 | if (typeof limit === 'number') { |
140 | var count = 0 |
141 | return pullCat([ |
142 | pull( |
143 | stream, |
144 | pull.take(limit), |
145 | pull.through(() => { |
146 | count += 1 |
147 | }) |
148 | ), |
149 | |
150 | // send truncated marker for resuming search |
151 | pull( |
152 | pull.values([marker]), |
153 | pull.filter(() => count === limit) |
154 | ) |
155 | ]) |
156 | } else { |
157 | return stream |
158 | } |
159 | } |
160 | } |
161 | |
162 | function shouldShow (filterResult, {onlySubscribedChannels}) { |
163 | if (filterResult && onlySubscribedChannels && filterResult.hasChannel) { |
164 | return filterResult.matchesChannel || filterResult.matchingTags.length || filterResult.mentionsYou || filterResult.isYours |
165 | } else { |
166 | return !!filterResult |
167 | } |
168 | } |
169 | |
170 | function getThruCache (key, cb) { |
171 | if (cache.has(key)) { |
172 | cb(null, cache.get(key)) |
173 | } else { |
174 | ssb.get(key, (_, value) => { |
175 | var msg = {key, value} |
176 | if (msg.value) { |
177 | cache.set(key, msg) |
178 | } |
179 | cb(null, msg) |
180 | }) |
181 | } |
182 | } |
183 | |
184 | function getFilter (cb) { |
185 | // TODO: rewrite contacts stream |
186 | ssb.friends.get((err, friends) => { |
187 | if (err) return cb(err) |
188 | ssb.patchwork.getSubscriptions((err, subscriptions) => { |
189 | if (err) return cb(err) |
190 | cb(null, function (ids, msg) { |
191 | var type = msg.value.content.type |
192 | if (type === 'vote') return false // filter out likes |
193 | var hasChannel = !!msg.value.content.channel |
194 | var matchesChannel = (type !== 'channel' && checkChannel(subscriptions, ids, msg.value.content.channel)) |
195 | var matchingTags = getMatchingTags(subscriptions, ids, msg.value.content.mentions) |
196 | var isYours = ids.includes(msg.value.author) |
197 | var mentionsYou = getMentionsYou(ids, msg.value.content.mentions) |
198 | |
199 | if (msg.value.author === '@u1Q+EEna+cJ+JGYppltW0hNZPAkJikVtOH5oMYZ3cEA=.ed25519') { |
200 | debugger |
201 | } |
202 | |
203 | var following = checkFollowing(friends, ids, msg.value.author) |
204 | if (isYours || matchesChannel || matchingTags.length || following || mentionsYou) { |
205 | return { |
206 | matchingTags, matchesChannel, isYours, following, mentionsYou, hasChannel |
207 | } |
208 | } |
209 | }) |
210 | }) |
211 | }) |
212 | } |
213 | |
214 | function LookupRoots () { |
215 | return pull.asyncMap((item, cb) => { |
216 | var msg = item.value |
217 | var key = item.key[1] |
218 | if (key === msg.key) { |
219 | // already a root |
220 | return cb(null, msg) |
221 | } |
222 | getThruCache(key, (_, value) => { |
223 | cb(null, extend(msg, { |
224 | root: value |
225 | })) |
226 | }) |
227 | }) |
228 | } |
229 | } |
230 | |
231 | function getMatchingTags (lookup, ids, mentions) { |
232 | if (Array.isArray(mentions)) { |
233 | return mentions.reduce((result, mention) => { |
234 | if (mention && typeof mention.link === 'string' && mention.link.startsWith('#')) { |
235 | if (checkChannel(lookup, ids, mention.link.slice(1))) { |
236 | result.push(normalizeChannel(mention.link.slice(1))) |
237 | } |
238 | } |
239 | return result |
240 | }, []) |
241 | } |
242 | return [] |
243 | } |
244 | |
245 | function getMentionsYou (ids, mentions) { |
246 | if (Array.isArray(mentions)) { |
247 | return mentions.some((mention) => { |
248 | if (mention && typeof mention.link === 'string') { |
249 | return ids.includes(mention.link) |
250 | } |
251 | }) |
252 | } |
253 | } |
254 | |
255 | function checkReplyForcesDisplay (item) { |
256 | var filterResult = item.filterResult || {} |
257 | var matchesTags = filterResult.matchingTags && !!filterResult.matchingTags.length |
258 | return matchesTags || filterResult.isYours |
259 | } |
260 | |
261 | function checkFollowing (lookup, ids, target) { |
262 | // TODO: rewrite contacts index (for some reason the order is different) |
263 | if (!lookup) return false |
264 | // HACK: only lookup the first ID until a method is added to ssb-friends to |
265 | // correctly identify latest info |
266 | var value = ids.slice(0, 1).map(id => lookup[id] && lookup[id][target]) |
267 | return value && value[0] |
268 | } |
269 | |
270 | function checkChannel (lookup, ids, channel) { |
271 | if (!lookup) return false |
272 | channel = normalizeChannel(channel) |
273 | if (channel) { |
274 | var value = mostRecentValue(ids.map(id => lookup[`${id}:${channel}`])) |
275 | return value && value[1] |
276 | } |
277 | } |
278 | |
279 | function mostRecentValue (values, timestampIndex = 0) { |
280 | var mostRecent = null |
281 | values.forEach(value => { |
282 | if (value && (!mostRecent || mostRecent[timestampIndex] < value[timestampIndex])) { |
283 | mostRecent = value |
284 | } |
285 | }) |
286 | return mostRecent |
287 | } |
288 |
Built with git-ssb-web