wip

src/post/create.ts

@@ -1,8 +1,14 @@
 import parseAcct from '../acct/parse';
-import Post from '../models/post';
-import User from '../models/user';
+import Post, { pack } from '../models/post';
+import User, { isLocalUser, isRemoteUser, IUser } from '../models/user';
+import stream from '../publishers/stream';
+import Following from '../models/following';
+import { createHttp } from '../queue';
+import renderNote from '../remote/activitypub/renderer/note';
+import renderCreate from '../remote/activitypub/renderer/create';
+import context from '../remote/activitypub/renderer/context';
 
-export default async (post, reply, repost, atMentions) => {
+export default async (user: IUser, post, reply, repost, atMentions) => {
 	post.mentions = [];
 
 	function addMention(mentionee) {
@@ -46,5 +52,98 @@ export default async (post, reply, repost, atMentions) => {
 		addMention(_id);
 	}));
 
-	return Post.insert(post);
+	const inserted = await Post.insert(post);
+
+	User.update({ _id: user._id }, {
+		// Increment my posts count
+		$inc: {
+			postsCount: 1
+		},
+
+		$set: {
+			latestPost: post._id
+		}
+	});
+
+	const postObj = await pack(inserted);
+
+	// タイムラインへの投稿
+	if (!post.channelId) {
+		// Publish event to myself's stream
+		stream(post.userId, 'post', postObj);
+
+		// Fetch all followers
+		const followers = await Following.aggregate([{
+			$lookup: {
+				from: 'users',
+				localField: 'followerId',
+				foreignField: '_id',
+				as: 'follower'
+			}
+		}, {
+			$match: {
+				followeeId: post.userId
+			}
+		}], {
+			_id: false
+		});
+
+		const note = await renderNote(user, post);
+		const content = renderCreate(note);
+		content['@context'] = context;
+
+		Promise.all(followers.map(({ follower }) => {
+			if (isLocalUser(follower)) {
+				// Publish event to followers stream
+				stream(follower._id, 'post', postObj);
+			} else {
+				// フォロワーがリモートユーザーかつ投稿者がローカルユーザーなら投稿を配信
+				if (isLocalUser(user)) {
+					createHttp({
+						type: 'deliver',
+						user,
+						content,
+						to: follower.account.inbox
+					}).save();
+				}
+			}
+		}));
+	}
+
+	// チャンネルへの投稿
+	/* TODO
+	if (post.channelId) {
+		promises.push(
+			// Increment channel index(posts count)
+			Channel.update({ _id: post.channelId }, {
+				$inc: {
+					index: 1
+				}
+			}),
+
+			// Publish event to channel
+			promisedPostObj.then(postObj => {
+				publishChannelStream(post.channelId, 'post', postObj);
+			}),
+
+			Promise.all([
+				promisedPostObj,
+
+				// Get channel watchers
+				ChannelWatching.find({
+					channelId: post.channelId,
+					// 削除されたドキュメントは除く
+					deletedAt: { $exists: false }
+				})
+			]).then(([postObj, watches]) => {
+				// チャンネルの視聴者(のタイムライン)に配信
+				watches.forEach(w => {
+					stream(w.userId, 'post', postObj);
+				});
+			})
+		);
+	}*/
+
+	return Promise.all(promises);
+
 };

src/processor/http/deliver-post.ts

@@ -1,93 +0,0 @@
-import Channel from '../../models/channel';
-import Following from '../../models/following';
-import ChannelWatching from '../../models/channel-watching';
-import Post, { pack } from '../../models/post';
-import User, { isLocalUser } from '../../models/user';
-import stream, { publishChannelStream } from '../../publishers/stream';
-import context from '../../remote/activitypub/renderer/context';
-import renderCreate from '../../remote/activitypub/renderer/create';
-import renderNote from '../../remote/activitypub/renderer/note';
-import request from '../../remote/request';
-
-export default ({ data }) => Post.findOne({ _id: data.id }).then(post => {
-	const promisedPostObj = pack(post);
-	const promises = [];
-
-	// タイムラインへの投稿
-	if (!post.channelId) {
-		promises.push(
-			// Publish event to myself's stream
-			promisedPostObj.then(postObj => {
-				stream(post.userId, 'post', postObj);
-			}),
-
-			Promise.all([
-				User.findOne({ _id: post.userId }),
-
-				// Fetch all followers
-				Following.aggregate([{
-					$lookup: {
-						from: 'users',
-						localField: 'followerId',
-						foreignField: '_id',
-						as: 'follower'
-					}
-				}, {
-					$match: {
-						followeeId: post.userId
-					}
-				}], {
-					_id: false
-				})
-			]).then(([user, followers]) => Promise.all(followers.map(following => {
-				if (isLocalUser(following.follower)) {
-					// Publish event to followers stream
-					return promisedPostObj.then(postObj => {
-						stream(following.followerId, 'post', postObj);
-					});
-				}
-
-				return renderNote(user, post).then(note => {
-					const create = renderCreate(note);
-					create['@context'] = context;
-					return request(user, following.follower[0].account.inbox, create);
-				});
-			})))
-		);
-	}
-
-	// チャンネルへの投稿
-	if (post.channelId) {
-		promises.push(
-			// Increment channel index(posts count)
-			Channel.update({ _id: post.channelId }, {
-				$inc: {
-					index: 1
-				}
-			}),
-
-			// Publish event to channel
-			promisedPostObj.then(postObj => {
-				publishChannelStream(post.channelId, 'post', postObj);
-			}),
-
-			Promise.all([
-				promisedPostObj,
-
-				// Get channel watchers
-				ChannelWatching.find({
-					channelId: post.channelId,
-					// 削除されたドキュメントは除く
-					deletedAt: { $exists: false }
-				})
-			]).then(([postObj, watches]) => {
-				// チャンネルの視聴者(のタイムライン)に配信
-				watches.forEach(w => {
-					stream(w.userId, 'post', postObj);
-				});
-			})
-		);
-	}
-
-	return Promise.all(promises);
-});

src/processor/http/process-inbox.ts

@@ -1,39 +0,0 @@
-import { verifySignature } from 'http-signature';
-import parseAcct from '../../acct/parse';
-import User, { IRemoteUser } from '../../models/user';
-import act from '../../remote/activitypub/act';
-import resolvePerson from '../../remote/activitypub/resolve-person';
-import Resolver from '../../remote/activitypub/resolver';
-
-export default async ({ data }): Promise<void> => {
-	const keyIdLower = data.signature.keyId.toLowerCase();
-	let user;
-
-	if (keyIdLower.startsWith('acct:')) {
-		const { username, host } = parseAcct(keyIdLower.slice('acct:'.length));
-		if (host === null) {
-			throw 'request was made by local user';
-		}
-
-		user = await User.findOne({ usernameLower: username, hostLower: host }) as IRemoteUser;
-	} else {
-		user = await User.findOne({
-			host: { $ne: null },
-			'account.publicKey.id': data.signature.keyId
-		}) as IRemoteUser;
-
-		if (user === null) {
-			user = await resolvePerson(data.signature.keyId);
-		}
-	}
-
-	if (user === null) {
-		throw 'failed to resolve user';
-	}
-
-	if (!verifySignature(data.signature, user.account.publicKey.publicKeyPem)) {
-		throw 'signature verification failed';
-	}
-
-	await Promise.all(await act(new Resolver(), user, data.inbox, true));
-};

src/queue.ts

@@ -1,10 +0,0 @@
-import { createQueue } from 'kue';
-import config from './config';
-
-export default createQueue({
-	redis: {
-		port: config.redis.port,
-		host: config.redis.host,
-		auth: config.redis.pass
-	}
-});

src/queue/index.ts

@@ -0,0 +1,37 @@
+import { createQueue } from 'kue';
+import config from '../config';
+import db from './processors/db';
+import http from './processors/http';
+
+const queue = createQueue({
+	redis: {
+		port: config.redis.port,
+		host: config.redis.host,
+		auth: config.redis.pass
+	}
+});
+
+export function createHttp(data) {
+	return queue
+		.create('http', data)
+		.attempts(16)
+		.backoff({ delay: 16384, type: 'exponential' });
+}
+
+export function createDb(data) {
+	return queue.create('db', data);
+}
+
+export function process() {
+	queue.process('db', db);
+
+	/*
+		256 is the default concurrency limit of Mozilla Firefox and Google
+		Chromium.
+		a8af215e691f3a2205a3758d2d96e9d328e100ff - chromium/src.git - Git at Google
+		https://chromium.googlesource.com/chromium/src.git/+/a8af215e691f3a2205a3758d2d96e9d328e100ff
+		Network.http.max-connections - MozillaZine Knowledge Base
+		http://kb.mozillazine.org/Network.http.max-connections
+	*/
+	queue.process('http', 256, http);
+}

src/queue/processors/http/deliver.ts

@@ -0,0 +1,17 @@
+import * as kue from 'kue';
+
+import Channel from '../../models/channel';
+import Following from '../../models/following';
+import ChannelWatching from '../../models/channel-watching';
+import Post, { pack } from '../../models/post';
+import User, { isLocalUser } from '../../models/user';
+import stream, { publishChannelStream } from '../../publishers/stream';
+import context from '../../remote/activitypub/renderer/context';
+import renderCreate from '../../remote/activitypub/renderer/create';
+import renderNote from '../../remote/activitypub/renderer/note';
+import request from '../../remote/request';
+
+export default async (job: kue.Job, done): Promise<void> => {
+
+	request(user, following.follower[0].account.inbox, create);
+}

src/queue/processors/http/process-inbox.ts

@@ -0,0 +1,55 @@
+import * as kue from 'kue';
+
+import { verifySignature } from 'http-signature';
+import parseAcct from '../../acct/parse';
+import User, { IRemoteUser } from '../../models/user';
+import act from '../../remote/activitypub/act';
+import resolvePerson from '../../remote/activitypub/resolve-person';
+import Resolver from '../../remote/activitypub/resolver';
+
+// ユーザーのinboxにアクティビティが届いた時の処理
+export default async (job: kue.Job, done): Promise<void> => {
+	const signature = job.data.signature;
+	const activity = job.data.activity;
+
+	const keyIdLower = signature.keyId.toLowerCase();
+	let user;
+
+	if (keyIdLower.startsWith('acct:')) {
+		const { username, host } = parseAcct(keyIdLower.slice('acct:'.length));
+		if (host === null) {
+			console.warn(`request was made by local user: @${username}`);
+			done();
+		}
+
+		user = await User.findOne({ usernameLower: username, hostLower: host }) as IRemoteUser;
+	} else {
+		user = await User.findOne({
+			host: { $ne: null },
+			'account.publicKey.id': signature.keyId
+		}) as IRemoteUser;
+
+		// アクティビティを送信してきたユーザーがまだMisskeyサーバーに登録されていなかったら登録する
+		if (user === null) {
+			user = await resolvePerson(signature.keyId);
+		}
+	}
+
+	if (user === null) {
+		done(new Error('failed to resolve user'));
+		return;
+	}
+
+	if (!verifySignature(signature, user.account.publicKey.publicKeyPem)) {
+		done(new Error('signature verification failed'));
+		return;
+	}
+
+	// アクティビティを処理
+	try {
+		await act(new Resolver(), user, activity);
+		done();
+	} catch (e) {
+		done(e);
+	}
+};

src/remote/activitypub/act/create.ts

@@ -1,10 +1,92 @@
-import create from '../create';
-import Resolver from '../resolver';
+import { JSDOM } from 'jsdom';
+const createDOMPurify = require('dompurify');
 
-export default (resolver: Resolver, actor, activity, distribute) => {
+import Resolver from '../resolver';
+import DriveFile from '../../../models/drive-file';
+import Post from '../../../models/post';
+import uploadFromUrl from '../../../drive/upload-from-url';
+import createPost from '../../../post/create';
+
+export default async (resolver: Resolver, actor, activity): Promise<void> => {
 	if ('actor' in activity && actor.account.uri !== activity.actor) {
-		throw new Error();
+		throw new Error('invalid actor');
 	}
 
-	return create(resolver, actor, activity.object, distribute);
+	const uri = activity.id || activity;
+
+	try {
+		await Promise.all([
+			DriveFile.findOne({ 'metadata.uri': uri }).then(file => {
+				if (file !== null) {
+					throw new Error();
+				}
+			}, () => {}),
+			Post.findOne({ uri }).then(post => {
+				if (post !== null) {
+					throw new Error();
+				}
+			}, () => {})
+		]);
+	} catch (object) {
+		throw new Error(`already registered: ${uri}`);
+	}
+
+	const object = await resolver.resolve(activity);
+
+	switch (object.type) {
+	case 'Image':
+		createImage(resolver, object);
+		break;
+
+	case 'Note':
+		createNote(resolver, object);
+		break;
+	}
+
+	///
+
+	async function createImage(resolver: Resolver, image) {
+		if ('attributedTo' in image && actor.account.uri !== image.attributedTo) {
+			throw new Error('invalid image');
+		}
+
+		return await uploadFromUrl(image.url, actor);
+	}
+
+	async function createNote(resolver: Resolver, note) {
+		if (
+			('attributedTo' in note && actor.account.uri !== note.attributedTo) ||
+			typeof note.id !== 'string'
+		) {
+			throw new Error('invalid note');
+		}
+
+		const mediaIds = [];
+
+		if ('attachment' in note) {
+			note.attachment.forEach(async media => {
+				const created = await createImage(resolver, media);
+				mediaIds.push(created._id);
+			});
+		}
+
+		const { window } = new JSDOM(note.content);
+
+		await createPost(actor, {
+			channelId: undefined,
+			index: undefined,
+			createdAt: new Date(note.published),
+			mediaIds,
+			replyId: undefined,
+			repostId: undefined,
+			poll: undefined,
+			text: window.document.body.textContent,
+			textHtml: note.content && createDOMPurify(window).sanitize(note.content),
+			userId: actor._id,
+			appId: null,
+			viaMobile: false,
+			geo: undefined,
+			uri: note.id
+		}, null, null, []);
+	}
 };

src/remote/activitypub/act/index.ts

@@ -2,35 +2,29 @@ import create from './create';
 import performDeleteActivity from './delete';
 import follow from './follow';
 import undo from './undo';
-import createObject from '../create';
 import Resolver from '../resolver';
+import { IObject } from '../type';
 
-export default async (parentResolver: Resolver, actor, value, distribute?: boolean) => {
-	const collection = await parentResolver.resolveCollection(value);
-
-	return collection.object.map(async element => {
-		const { resolver, object } = await collection.resolver.resolveOne(element);
-		const created = await (await createObject(resolver, actor, [object], distribute))[0];
-
-		if (created !== null) {
-			return created;
-		}
-
-		switch (object.type) {
+export default async (parentResolver: Resolver, actor, activity: IObject): Promise<void> => {
+	switch (activity.type) {
 	case 'Create':
-			return create(resolver, actor, object, distribute);
+		await create(parentResolver, actor, activity);
+		break;
 
 	case 'Delete':
-			return performDeleteActivity(resolver, actor, object);
+		await performDeleteActivity(parentResolver, actor, activity);
+		break;
 
 	case 'Follow':
-			return follow(resolver, actor, object, distribute);
+		await follow(parentResolver, actor, activity);
+		break;
 
 	case 'Undo':
-			return undo(resolver, actor, object);
+		await undo(parentResolver, actor, activity);
+		break;
 
 	default:
+		console.warn(`unknown activity type: ${activity.type}`);
 		return null;
 	}
-	});
 };

src/remote/activitypub/create.ts

@@ -1,158 +0,0 @@
-import { JSDOM } from 'jsdom';
-import { ObjectID } from 'mongodb';
-import config from '../../config';
-import DriveFile from '../../models/drive-file';
-import Post from '../../models/post';
-import { IRemoteUser } from '../../models/user';
-import uploadFromUrl from '../../drive/upload-from-url';
-import createPost from '../../post/create';
-import distributePost from '../../post/distribute';
-import Resolver from './resolver';
-const createDOMPurify = require('dompurify');
-
-type IResult = {
-	resolver: Resolver;
-	object: {
-		$ref: string;
-		$id: ObjectID;
-	};
-};
-
-class Creator {
-	private actor: IRemoteUser;
-	private distribute: boolean;
-
-	constructor(actor, distribute) {
-		this.actor = actor;
-		this.distribute = distribute;
-	}
-
-	private async createImage(resolver: Resolver, image) {
-		if ('attributedTo' in image && this.actor.account.uri !== image.attributedTo) {
-			throw new Error();
-		}
-
-		const { _id } = await uploadFromUrl(image.url, this.actor, image.id || null);
-		return {
-			resolver,
-			object: { $ref: 'driveFiles.files', $id: _id }
-		};
-	}
-
-	private async createNote(resolver: Resolver, note) {
-		if (
-			('attributedTo' in note && this.actor.account.uri !== note.attributedTo) ||
-			typeof note.id !== 'string'
-		) {
-			throw new Error();
-		}
-
-		const mediaIds = 'attachment' in note &&
-			(await Promise.all(await this.create(resolver, note.attachment)))
-				.filter(media => media !== null && media.object.$ref === 'driveFiles.files')
-				.map(({ object }) => object.$id);
-
-		const { window } = new JSDOM(note.content);
-
-		const inserted = await createPost({
-			channelId: undefined,
-			index: undefined,
-			createdAt: new Date(note.published),
-			mediaIds,
-			replyId: undefined,
-			repostId: undefined,
-			poll: undefined,
-			text: window.document.body.textContent,
-			textHtml: note.content && createDOMPurify(window).sanitize(note.content),
-			userId: this.actor._id,
-			appId: null,
-			viaMobile: false,
-			geo: undefined,
-			uri: note.id
-		}, null, null, []);
-
-		const promises = [];
-
-		if (this.distribute) {
-			promises.push(distributePost(this.actor, inserted.mentions, inserted));
-		}
-
-		// Register to search database
-		if (note.content && config.elasticsearch.enable) {
-			const es = require('../../db/elasticsearch');
-
-			promises.push(new Promise((resolve, reject) => {
-				es.index({
-					index: 'misskey',
-					type: 'post',
-					id: inserted._id.toString(),
-					body: {
-						text: window.document.body.textContent
-					}
-				}, resolve);
-			}));
-		}
-
-		await Promise.all(promises);
-
-		return {
-			resolver,
-			object: { $ref: 'posts', id: inserted._id }
-		};
-	}
-
-	public async create(parentResolver: Resolver, value): Promise<Array<Promise<IResult>>> {
-		const collection = await parentResolver.resolveCollection(value);
-
-		return collection.object.map(async element => {
-			const uri = element.id || element;
-
-			try {
-				await Promise.all([
-					DriveFile.findOne({ 'metadata.uri': uri }).then(file => {
-						if (file === null) {
-							return;
-						}
-
-						throw {
-							$ref: 'driveFile.files',
-							$id: file._id
-						};
-					}, () => {}),
-					Post.findOne({ uri }).then(post => {
-						if (post === null) {
-							return;
-						}
-
-						throw {
-							$ref: 'posts',
-							$id: post._id
-						};
-					}, () => {})
-				]);
-			} catch (object) {
-				return {
-					resolver: collection.resolver,
-					object
-				};
-			}
-
-			const { resolver, object } = await collection.resolver.resolveOne(element);
-
-			switch (object.type) {
-			case 'Image':
-				return this.createImage(resolver, object);
-
-			case 'Note':
-				return this.createNote(resolver, object);
-			}
-
-			return null;
-		});
-	}
-}
-
-export default (resolver: Resolver, actor, value, distribute?: boolean) => {
-	const creator = new Creator(actor, distribute);
-	return creator.create(resolver, value);
-};

src/remote/activitypub/resolver.ts

@@ -1,20 +1,45 @@
+import { IObject } from "./type";
+
 const request = require('request-promise-native');
 
 export default class Resolver {
-	private requesting: Set<string>;
+	private history: Set<string>;
 
-	constructor(iterable?: Iterable<string>) {
-		this.requesting = new Set(iterable);
+	constructor() {
+		this.history = new Set();
 	}
 
-	private async resolveUnrequestedOne(value) {
+	public async resolveCollection(value) {
+		const collection = typeof value === 'string'
+			? await this.resolve(value)
+			: value;
+
+		switch (collection.type) {
+		case 'Collection':
+			collection.objects = collection.object.items;
+			break;
+
+		case 'OrderedCollection':
+			collection.objects = collection.object.orderedItems;
+			break;
+
+		default:
+			throw new Error(`unknown collection type: ${collection.type}`);
+		}
+
+		return collection;
+	}
+
+	public async resolve(value): Promise<IObject> {
 		if (typeof value !== 'string') {
-			return { resolver: this, object: value };
+			return value;
 		}
 
-		const resolver = new Resolver(this.requesting);
+		if (this.history.has(value)) {
+			throw new Error('cannot resolve already resolved one');
+		}
 
-		resolver.requesting.add(value);
+		this.history.add(value);
 
 		const object = await request({
 			url: value,
@@ -29,41 +54,9 @@ export default class Resolver {
 				!object['@context'].includes('https://www.w3.org/ns/activitystreams') :
 				object['@context'] !== 'https://www.w3.org/ns/activitystreams'
 		)) {
-			throw new Error();
+			throw new Error('invalid response');
 		}
 
-		return { resolver, object };
-	}
-
-	public async resolveCollection(value) {
-		const resolved = typeof value === 'string' ?
-			await this.resolveUnrequestedOne(value) :
-			{ resolver: this, object: value };
-
-		switch (resolved.object.type) {
-		case 'Collection':
-			resolved.object = resolved.object.items;
-			break;
-
-		case 'OrderedCollection':
-			resolved.object = resolved.object.orderedItems;
-			break;
-
-		default:
-			if (!Array.isArray(value)) {
-				resolved.object = [resolved.object];
-			}
-			break;
-		}
-
-		return resolved;
-	}
-
-	public resolveOne(value) {
-		if (this.requesting.has(value)) {
-			throw new Error();
-		}
-
-		return this.resolveUnrequestedOne(value);
+		return object;
 	}
 }

src/server/activitypub/inbox.ts

@@ -24,7 +24,7 @@ app.post('/@:user/inbox', bodyParser.json({
 
 	queue.create('http', {
 		type: 'processInbox',
-		inbox: req.body,
+		activity: req.body,
 		signature,
 	}).save();