feat: save tids to individual user inboxes based on recipient list, new /world/all route

src/activitypub/mocks.js

@@ -88,6 +88,8 @@ Mocks.post = async (objects) => {
 			content,
 			sourceContent,
 			inReplyTo: toPid,
+			to,
+			cc,
 		} = object;
 
 		const timestamp = new Date(published).getTime();
@@ -106,6 +108,7 @@ Mocks.post = async (objects) => {
 
 			edited,
 			editor: edited ? uid : undefined,
+			_activitypub: { to, cc },
 		};
 
 		return payload;

src/activitypub/notes.js

@@ -3,10 +3,11 @@
 const winston = require('winston');
 
 const db = require('../database');
+const user = require('../user');
 const topics = require('../topics');
 const posts = require('../posts');
 const utils = require('../utils');
-const pubsub = require('../pubsub');
+// const pubsub = require('../pubsub');
 const slugify = require('../slugify');
 
 const activitypub = module.parent.exports;
@@ -22,7 +23,8 @@ Notes.resolveId = async (uid, id) => {
 Notes.assert = async (uid, input, options = {}) => {
 	// Ensures that each note has been saved to the database
 	await Promise.all(input.map(async (item) => {
-		const id = activitypub.helpers.isUri(item) ? item : item.pid;
+		let id = activitypub.helpers.isUri(item) ? item : item.pid;
+		id = await Notes.resolveId(uid, id);
 		const key = `post:${id}`;
 		const exists = await db.exists(key);
 		winston.verbose(`[activitypub/notes.assert] Asserting note id ${id}`);
@@ -38,15 +40,48 @@ Notes.assert = async (uid, input, options = {}) => {
 				postData = item;
 			}
 
-			await db.setObject(key, postData);
+			// Parse ActivityPub-specific data
+			const { to, cc } = postData._activitypub;
+			let recipients = new Set([...to, ...cc]);
+			recipients = await Notes.getLocalRecipients(recipients);
+			if (recipients.size > 0) {
+				await db.setAdd(`post:${id}:recipients`, Array.from(recipients));
+			}
+
+			const hash = { ...postData };
+			delete hash._activitypub;
+			await db.setObject(key, hash);
 			winston.verbose(`[activitypub/notes.assert] Note ${id} saved.`);
 		}
 
-		if (options.update === true) {
-			require('../posts/cache').del(String(id));
-			pubsub.publish('post:edit', String(id));
+		// odd circular modular dependency issue here...
+		// if (options.update === true) {
+		// 	require('../posts/cache').del(String(id));
+		// 	pubsub.publish('post:edit', String(id));
+		// }
+	}));
+};
+
+Notes.getLocalRecipients = async (recipients) => {
+	const uids = new Set();
+	await Promise.all(Array.from(recipients).map(async (recipient) => {
+		const { type, id } = await activitypub.helpers.resolveLocalId(recipient);
+		if (type === 'user' && await user.exists(id)) {
+			uids.add(parseInt(id, 10));
+			return;
+		}
+
+		const followedUid = await db.getObjectField('followersUrl:uid', recipient);
+		if (followedUid) {
+			const followers = await db.getSortedSetMembers(`followersRemote:${followedUid}`);
+			if (followers.length) {
+				uids.add(...followers.map(uid => parseInt(uid, 10)));
+			}
+			// return;
 		}
 	}));
+
+	return uids;
 };
 
 Notes.getParentChain = async (uid, input) => {
@@ -161,6 +196,7 @@ Notes.assertTopic = async (uid, id) => {
 	await Promise.all([ // must be done after .assert()
 		Notes.assertParentChain(chain, tid),
 		Notes.updateTopicCounts(tid),
+		Notes.syncUserInboxes(tid),
 		topics.updateLastPostTimeFromLastPid(tid),
 		topics.updateTeaser(tid),
 	]);
@@ -182,6 +218,18 @@ Notes.updateTopicCounts = async function (tid) {
 	});
 };
 
+Notes.syncUserInboxes = async function (tid) {
+	const pids = await db.getSortedSetMembers(`tid:${tid}:posts`);
+	const recipients = await db.getSetsMembers(pids.map(id => `post:${id}:recipients`));
+	const uids = recipients.reduce((set, uids) => new Set([...set, ...uids.map(u => parseInt(u, 10))]), new Set());
+	const cid = await topics.getTopicField(tid, 'cid');
+	const keys = Array.from(uids).map(uid => `uid:${uid}:inbox`);
+	const score = await db.sortedSetScore(`cid:${cid}:tids`, tid);
+
+	winston.verbose(`[activitypub/syncUserInboxes] Syncing tid ${tid} with ${uids.length} inboxes`);
+	await db.sortedSetsAdd(keys, keys.map(() => score), tid);
+};
+
 Notes.getTopicPosts = async (tid, uid, start, stop) => {
 	const pids = await db.getSortedSetRange(`tid:${tid}:posts`, start, stop);
 	return await posts.getPostsByPids(pids, uid);

src/controllers/activitypub/topics.js

@@ -15,10 +15,20 @@ controller.list = async function (req, res) {
 	const start = Math.max(0, (page - 1) * topicsPerPage);
 	const stop = start + topicsPerPage - 1;
 
-	const tids = await db.getSortedSetRevRange('cid:-1:tids', start, stop);
+	const sets = ['cid:-1:tids', `uid:${req.uid}:inbox`];
+	if (req.params.filter === 'all') {
+		sets.pop();
+	}
+
+	const tids = await db.getSortedSetRevIntersect({
+		sets,
+		start,
+		stop,
+		weights: sets.map((s, index) => (index ? 0 : 1)),
+	});
 
 	const data = {};
-	data.topicCount = await db.sortedSetCard('cid:-1:tids');
+	data.topicCount = await db.sortedSetIntersectCard(sets);
 	data.topics = await topics.getTopicsByTids(tids, { uid: req.uid });
 	topics.calculateTopicIndices(data.topics, start);
 

src/routes/activitypub.js

@@ -3,7 +3,7 @@
 const helpers = require('./helpers');
 
 module.exports = function (app, middleware, controllers) {
-	helpers.setupPageRoute(app, '/world', [middleware.activitypub.enabled], controllers.activitypub.topics.list);
+	helpers.setupPageRoute(app, '/world/:filter?', [middleware.activitypub.enabled], controllers.activitypub.topics.list);
 
 	/**
 	 * These controllers only respond if the sender is making an json+activitypub style call (i.e. S2S-only)