Merge pull request #1397 from syuilo/refactor

Refactor

src/client/app/init.ts

@@ -14,7 +14,7 @@ import ElementLocaleJa from 'element-ui/lib/locale/lang/ja';
 import App from './app.vue';
 import checkForUpdate from './common/scripts/check-for-update';
 import MiOS, { API } from './common/mios';
-import { version, codename, hostname, lang } from './config';
+import { version, codename, lang } from './config';
 
 let elementLocale;
 switch (lang) {
@@ -60,10 +60,6 @@ console.info(
 window.clearTimeout((window as any).mkBootTimer);
 delete (window as any).mkBootTimer;
 
-if (hostname != 'localhost') {
-	document.domain = hostname;
-}
-
 //#region Set lang attr
 const html = document.documentElement;
 html.setAttribute('lang', lang);

src/following/distribute.ts

@@ -1,42 +0,0 @@
-import User, { pack as packUser } from '../models/user';
-import FollowingLog from '../models/following-log';
-import FollowedLog from '../models/followed-log';
-import event from '../publishers/stream';
-import notify from '../publishers/notify';
-
-export default async (follower, followee) => Promise.all([
-	// Increment following count
-	User.update(follower._id, {
-		$inc: {
-			followingCount: 1
-		}
-	}),
-
-	FollowingLog.insert({
-		createdAt: new Date(),
-		userId: followee._id,
-		count: follower.followingCount + 1
-	}),
-
-	// Increment followers count
-	User.update({ _id: followee._id }, {
-		$inc: {
-			followersCount: 1
-		}
-	}),
-
-	FollowedLog.insert({
-		createdAt: new Date(),
-		userId: follower._id,
-		count: followee.followersCount + 1
-	}),
-
-	followee.host === null && Promise.all([
-		// Notify
-		notify(followee.id, follower.id, 'follow'),
-
-		// Publish follow event
-		packUser(follower, followee)
-			.then(packed => event(followee._id, 'followed', packed))
-	])
-]);

src/index.ts

@@ -30,8 +30,12 @@ const ev = new Xev();
 
 process.title = 'Misskey';
 
+if (process.env.NODE_ENV != 'production') {
+	process.env.DEBUG = 'misskey:*';
+}
+
 // https://github.com/Automattic/kue/issues/822
-require('events').EventEmitter.prototype._maxListeners = 256;
+require('events').EventEmitter.prototype._maxListeners = 512;
 
 // Start app
 main();
@@ -99,7 +103,7 @@ async function workerMain(opt) {
 
 	if (!opt['only-server']) {
 		// start processor
-		require('./queue').process();
+		require('./queue').default();
 	}
 
 	// Send a 'ready' message to parent process

src/models/post-watching.ts

@@ -2,6 +2,7 @@ import * as mongo from 'mongodb';
 import db from '../db/mongodb';
 
 const PostWatching = db.get<IPostWatching>('postWatching');
+PostWatching.createIndex(['userId', 'postId'], { unique: true });
 export default PostWatching;
 
 export interface IPostWatching {

src/models/post.ts

@@ -27,6 +27,7 @@ export type IPost = {
 	_id: mongo.ObjectID;
 	channelId: mongo.ObjectID;
 	createdAt: Date;
+	deletedAt: Date;
 	mediaIds: mongo.ObjectID[];
 	replyId: mongo.ObjectID;
 	repostId: mongo.ObjectID;
@@ -52,6 +53,20 @@ export type IPost = {
 		speed: number;
 	};
 	uri: string;
+
+	_reply?: {
+		userId: mongo.ObjectID;
+	};
+	_repost?: {
+		userId: mongo.ObjectID;
+	};
+	_user: {
+		host: string;
+		hostLower: string;
+		account: {
+			inbox?: string;
+		};
+	};
 };
 
 /**

src/post/create.ts

@@ -1,40 +0,0 @@
-import Post from '../models/post';
-
-export default async (post, reply, repost, mentions) => {
-	post.mentions = [];
-
-	function addMention(mentionee) {
-		// Reject if already added
-		if (post.mentions.some(x => x.equals(mentionee))) return;
-
-		// Add mention
-		post.mentions.push(mentionee);
-	}
-
-	if (reply) {
-		// Add mention
-		addMention(reply.userId);
-		post.replyId = reply._id;
-		post._reply = { userId: reply.userId };
-	} else {
-		post.replyId = null;
-		post._reply = null;
-	}
-
-	if (repost) {
-		if (post.text) {
-			// Add mention
-			addMention(repost.userId);
-		}
-
-		post.repostId = repost._id;
-		post._repost = { userId: repost.userId };
-	} else {
-		post.repostId = null;
-		post._repost = null;
-	}
-
-	await Promise.all(mentions.map(({ _id }) => addMention(_id)));
-
-	return Post.insert(post);
-};

src/post/distribute.ts

@@ -1,274 +0,0 @@
-import Channel from '../models/channel';
-import ChannelWatching from '../models/channel-watching';
-import Following from '../models/following';
-import Mute from '../models/mute';
-import Post, { pack } from '../models/post';
-import Watching from '../models/post-watching';
-import User, { isLocalUser } from '../models/user';
-import stream, { publishChannelStream } from '../publishers/stream';
-import notify from '../publishers/notify';
-import pushSw from '../publishers/push-sw';
-import { createHttp } from '../queue';
-import watch from './watch';
-
-export default async (user, mentions, post) => {
-	const promisedPostObj = pack(post);
-	const promises = [
-		User.update({ _id: user._id }, {
-			// Increment my posts count
-			$inc: {
-				postsCount: 1
-			},
-
-			$set: {
-				latestPost: post._id
-			}
-		}),
-	] as Array<Promise<any>>;
-
-	function addMention(promisedMentionee, reason) {
-		// Publish event
-		promises.push(promisedMentionee.then(mentionee => {
-			if (user._id.equals(mentionee)) {
-				return Promise.resolve();
-			}
-
-			return Promise.all([
-				promisedPostObj,
-				Mute.find({
-					muterId: mentionee,
-					deletedAt: { $exists: false }
-				})
-			]).then(([postObj, mentioneeMutes]) => {
-				const mentioneesMutedUserIds = mentioneeMutes.map(m => m.muteeId.toString());
-				if (mentioneesMutedUserIds.indexOf(user._id.toString()) == -1) {
-					stream(mentionee, reason, postObj);
-					pushSw(mentionee, reason, postObj);
-				}
-			});
-		}));
-	}
-
-	// タイムラインへの投稿
-	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 new Promise((resolve, reject) => {
-					createHttp({
-						type: 'deliverPost',
-						fromId: user._id,
-						toId: following.followerId,
-						postId: post._id
-					}).save(error => {
-						if (error) {
-							reject(error);
-						} else {
-							resolve();
-						}
-					});
-				});
-			})))
-		);
-	}
-
-	// チャンネルへの投稿
-	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);
-				});
-			})
-		);
-	}
-
-	// If has in reply to post
-	if (post.replyId) {
-		promises.push(
-			// Increment replies count
-			Post.update({ _id: post.replyId }, {
-				$inc: {
-					repliesCount: 1
-				}
-			}),
-
-			// 自分自身へのリプライでない限りは通知を作成
-			promisedPostObj.then(({ reply }) => {
-				return notify(reply.userId, user._id, 'reply', {
-					postId: post._id
-				});
-			}),
-
-			// Fetch watchers
-			Watching
-				.find({
-					postId: post.replyId,
-					userId: { $ne: user._id },
-					// 削除されたドキュメントは除く
-					deletedAt: { $exists: false }
-				}, {
-					fields: {
-						userId: true
-					}
-				})
-				.then(watchers => {
-					watchers.forEach(watcher => {
-						notify(watcher.userId, user._id, 'reply', {
-							postId: post._id
-						});
-					});
-				})
-		);
-
-		// Add mention
-		addMention(promisedPostObj.then(({ reply }) => reply.userId), 'reply');
-
-		// この投稿をWatchする
-		if (user.account.settings.autoWatch !== false) {
-			promises.push(promisedPostObj.then(({ reply }) => {
-				return watch(user._id, reply);
-			}));
-		}
-	}
-
-	// If it is repost
-	if (post.repostId) {
-		const type = post.text ? 'quote' : 'repost';
-
-		promises.push(
-			promisedPostObj.then(({ repost }) => Promise.all([
-				// Notify
-				notify(repost.userId, user._id, type, {
-					postId: post._id
-				}),
-
-				// この投稿をWatchする
-				// TODO: ユーザーが「Repostしたときに自動でWatchする」設定を
-				//       オフにしていた場合はしない
-				watch(user._id, repost)
-			])),
-
-			// Fetch watchers
-			Watching
-				.find({
-					postId: post.repostId,
-					userId: { $ne: user._id },
-					// 削除されたドキュメントは除く
-					deletedAt: { $exists: false }
-				}, {
-					fields: {
-						userId: true
-					}
-				})
-				.then(watchers => {
-					watchers.forEach(watcher => {
-						notify(watcher.userId, user._id, type, {
-							postId: post._id
-						});
-					});
-				})
-		);
-
-		// If it is quote repost
-		if (post.text) {
-			// Add mention
-			addMention(promisedPostObj.then(({ repost }) => repost.userId), 'quote');
-		} else {
-			promises.push(promisedPostObj.then(postObj => {
-				// Publish event
-				if (!user._id.equals(postObj.repost.userId)) {
-					stream(postObj.repost.userId, 'repost', postObj);
-				}
-			}));
-		}
-
-		// 今までで同じ投稿をRepostしているか
-		const existRepost = await Post.findOne({
-			userId: user._id,
-			repostId: post.repostId,
-			_id: {
-				$ne: post._id
-			}
-		});
-
-		if (!existRepost) {
-			// Update repostee status
-			promises.push(Post.update({ _id: post.repostId }, {
-				$inc: {
-					repostCount: 1
-				}
-			}));
-		}
-	}
-
-	// Resolve all mentions
-	await promisedPostObj.then(({ reply, repost }) => Promise.all(mentions.map(async mention => {
-		// 既に言及されたユーザーに対する返信や引用repostの場合も無視
-		if (reply && reply.userId.equals(mention)) return;
-		if (repost && repost.userId.equals(mention)) return;
-
-		// Add mention
-		addMention(mention, 'mention');
-
-		// Create notification
-		await notify(mention, user._id, 'mention', {
-			postId: post._id
-		});
-	})));
-
-	await Promise.all(promises);
-
-	return promisedPostObj;
-};

src/queue/index.ts

@@ -1,6 +1,6 @@
 import { createQueue } from 'kue';
+
 import config from '../config';
-import db from './processors/db';
 import http from './processors/http';
 
 const queue = createQueue({
@@ -18,17 +18,19 @@ export function createHttp(data) {
 		.backoff({ delay: 16384, type: 'exponential' });
 }
 
-export function createDb(data) {
+export function deliver(user, content, to) {
-	return queue.create('db', data);
+	return createHttp({
+		type: 'deliver',
+		user,
+		content,
+		to
+	});
 }
 
-export function process() {
+export default function() {
-	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

src/queue/processors/db/delete-post-dependents.ts

@@ -1,22 +0,0 @@
-import Favorite from '../../../models/favorite';
-import Notification from '../../../models/notification';
-import PollVote from '../../../models/poll-vote';
-import PostReaction from '../../../models/post-reaction';
-import PostWatching from '../../../models/post-watching';
-import Post from '../../../models/post';
-
-export default ({ data }, done) => Promise.all([
-	Favorite.remove({ postId: data._id }),
-	Notification.remove({ postId: data._id }),
-	PollVote.remove({ postId: data._id }),
-	PostReaction.remove({ postId: data._id }),
-	PostWatching.remove({ postId: data._id }),
-	Post.find({ repostId: data._id }).then(reposts => Promise.all([
-		Notification.remove({
-			postId: {
-				$in: reposts.map(({ _id }) => _id)
-			}
-		}),
-		Post.remove({ repostId: data._id })
-	]))
-]).then(() => done(), done);

src/queue/processors/db/index.ts

@@ -1,7 +0,0 @@
-import deletePostDependents from './delete-post-dependents';
-
-const handlers = {
-  deletePostDependents
-};
-
-export default (job, done) => handlers[job.data.type](job, done);

src/queue/processors/http/deliver-post.ts

@@ -1,27 +0,0 @@
-import Post from '../../../models/post';
-import User, { IRemoteUser } from '../../../models/user';
-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 ({ data }, done) => {
-	try {
-		const promisedTo = User.findOne({ _id: data.toId }) as Promise<IRemoteUser>;
-		const [from, post] = await Promise.all([
-			User.findOne({ _id: data.fromId }),
-			Post.findOne({ _id: data.postId })
-		]);
-		const note = await renderNote(from, post);
-		const to = await promisedTo;
-		const create = renderCreate(note);
-
-		create['@context'] = context;
-
-		await request(from, to.account.inbox, create);
-	} catch (error) {
-		done(error);
-	}
-
-	done();
-};

src/queue/processors/http/deliver.ts

@@ -0,0 +1,19 @@
+import * as kue from 'kue';
+
+import request from '../../../remote/request';
+
+export default async (job: kue.Job, done): Promise<void> => {
+	try {
+		await request(job.data.user, job.data.to, job.data.content);
+		done();
+	} catch (res) {
+		if (res.statusCode >= 400 && res.statusCode < 500) {
+			// HTTPステータスコード4xxはクライアントエラーであり、それはつまり
+			// 何回再送しても成功することはないということなのでエラーにはしないでおく
+			done();
+		} else {
+			console.warn(`deliver failed: ${res.statusMessage}`);
+			done(new Error(res.statusMessage));
+		}
+	}
+};

src/queue/processors/http/follow.ts

@@ -1,66 +0,0 @@
-import User, { isLocalUser, isRemoteUser, pack as packUser } from '../../../models/user';
-import Following from '../../../models/following';
-import FollowingLog from '../../../models/following-log';
-import FollowedLog from '../../../models/followed-log';
-import event from '../../../publishers/stream';
-import notify from '../../../publishers/notify';
-import context from '../../../remote/activitypub/renderer/context';
-import render from '../../../remote/activitypub/renderer/follow';
-import request from '../../../remote/request';
-
-export default ({ data }, done) => Following.findOne({ _id: data.following }).then(async ({ followerId, followeeId }) => {
-	const [follower, followee] = await Promise.all([
-		User.findOne({ _id: followerId }),
-		User.findOne({ _id: followeeId })
-	]);
-
-	if (isLocalUser(follower) && isRemoteUser(followee)) {
-		const rendered = render(follower, followee);
-		rendered['@context'] = context;
-
-		await request(follower, followee.account.inbox, rendered);
-	}
-
-	return [follower, followee];
-}).then(([follower, followee]) => Promise.all([
-	// Increment following count
-	User.update(follower._id, {
-		$inc: {
-			followingCount: 1
-		}
-	}),
-
-	FollowingLog.insert({
-		createdAt: data.following.createdAt,
-		userId: follower._id,
-		count: follower.followingCount + 1
-	}),
-
-	// Increment followers count
-	User.update({ _id: followee._id }, {
-		$inc: {
-			followersCount: 1
-		}
-	}),
-
-	FollowedLog.insert({
-		createdAt: data.following.createdAt,
-		userId: follower._id,
-		count: followee.followersCount + 1
-	}),
-
-	// Publish follow event
-	isLocalUser(follower) && packUser(followee, follower)
-		.then(packed => event(follower._id, 'follow', packed)),
-
-	isLocalUser(followee) && Promise.all([
-		packUser(follower, followee)
-			.then(packed => event(followee._id, 'followed', packed)),
-
-		// Notify
-		isLocalUser(followee) && notify(followee._id, follower._id, 'follow')
-	])
-]).then(() => done(), error => {
-	done();
-	throw error;
-}), done);

src/queue/processors/http/index.ts

@@ -1,17 +1,20 @@
-import deliverPost from './deliver-post';
+import deliver from './deliver';
-import follow from './follow';
-import performActivityPub from './perform-activitypub';
 import processInbox from './process-inbox';
 import reportGitHubFailure from './report-github-failure';
-import unfollow from './unfollow';
 
 const handlers = {
-  deliverPost,
+	deliver,
-  follow,
+	processInbox,
-  performActivityPub,
+	reportGitHubFailure
-  processInbox,
-  reportGitHubFailure,
-  unfollow
 };
 
-export default (job, done) => handlers[job.data.type](job, done);
+export default (job, done) => {
+	const handler = handlers[job.data.type];
+
+	if (handler) {
+		handler(job, done);
+	} else {
+		console.error(`Unknown job: ${job.data.type}`);
+		done();
+	}
+};

src/queue/processors/http/perform-activitypub.ts

@@ -1,8 +0,0 @@
-import User from '../../../models/user';
-import act from '../../../remote/activitypub/act';
-import Resolver from '../../../remote/activitypub/resolver';
-
-export default ({ data }, done) => User.findOne({ _id: data.actor })
-	.then(actor => act(new Resolver(), actor, data.outbox))
-	.then(Promise.all)
-	.then(() => done(), done);

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

@@ -1,44 +1,66 @@
+import * as kue from 'kue';
+import * as debug from 'debug';
+
 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 }, done) => {
+const log = debug('misskey:queue:inbox');
-	try {
-		const keyIdLower = data.signature.keyId.toLowerCase();
-		let user;
 
-		if (keyIdLower.startsWith('acct:')) {
+// ユーザーのinboxにアクティビティが届いた時の処理
-			const { username, host } = parseAcct(keyIdLower.slice('acct:'.length));
+export default async (job: kue.Job, done): Promise<void> => {
-			if (host === null) {
+	const signature = job.data.signature;
-				done();
+	const activity = job.data.activity;
-				return;
-			}
 
-			user = await User.findOne({ usernameLower: username, hostLower: host }) as IRemoteUser;
+	//#region Log
-		} else {
+	const info = Object.assign({}, activity);
-			user = await User.findOne({
+	delete info['@context'];
-				host: { $ne: null },
+	delete info['signature'];
-				'account.publicKey.id': data.signature.keyId
+	log(info);
-			}) as IRemoteUser;
+	//#endregion
 
-			if (user === null) {
+	const keyIdLower = signature.keyId.toLowerCase();
-				user = await resolvePerson(new Resolver(), data.signature.keyId);
+	let user;
-			}
-		}
 
-		if (user === null || !verifySignature(data.signature, user.account.publicKey.publicKeyPem)) {
+	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();
 			return;
 		}
 
-		await Promise.all(await act(new Resolver(), user, data.inbox, true));
+		user = await User.findOne({ usernameLower: username, hostLower: host }) as IRemoteUser;
-	} catch (error) {
+	} else {
-		done(error);
+		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;
 	}
 
-	done();
+	if (!verifySignature(signature, user.account.publicKey.publicKeyPem)) {
+		console.warn('signature verification failed');
+		done();
+		return;
+	}
+
+	// アクティビティを処理
+	try {
+		await act(user, activity);
+		done();
+	} catch (e) {
+		done(e);
+	}
 };

src/queue/processors/http/report-github-failure.ts

@@ -1,31 +1,24 @@
 import * as request from 'request-promise-native';
 import User from '../../../models/user';
-const createPost = require('../../../server/api/endpoints/posts/create');
+import createPost from '../../../services/post/create';
 
-export default async ({ data }, done) => {
+export default async ({ data }) => {
-	try {
+	const asyncBot = User.findOne({ _id: data.userId });
-		const asyncBot = User.findOne({ _id: data.userId });
 
-		// Fetch parent status
+	// Fetch parent status
-		const parentStatuses = await request({
+	const parentStatuses = await request({
-			url: `${data.parentUrl}/statuses`,
+		url: `${data.parentUrl}/statuses`,
-			headers: {
+		headers: {
-				'User-Agent': 'misskey'
+			'User-Agent': 'misskey'
-			},
+		},
-			json: true
+		json: true
-		});
+	});
 
-		const parentState = parentStatuses[0].state;
+	const parentState = parentStatuses[0].state;
-		const stillFailed = parentState == 'failure' || parentState == 'error';
+	const stillFailed = parentState == 'failure' || parentState == 'error';
-		const text = stillFailed ?
+	const text = stillFailed ?
-			`**⚠️BUILD STILL FAILED⚠️**: ?[${data.message}](${data.htmlUrl})` :
+		`**⚠️BUILD STILL FAILED⚠️**: ?[${data.message}](${data.htmlUrl})` :
-			`**🚨BUILD FAILED🚨**: →→→?[${data.message}](${data.htmlUrl})←←←`;
+		`**🚨BUILD FAILED🚨**: →→→?[${data.message}](${data.htmlUrl})←←←`;
 
-		createPost({ text }, await asyncBot);
+	createPost(await asyncBot, { text });
-	} catch (error) {
-		done(error);
-		return;
-	}
-
-	done();
 };

src/queue/processors/http/unfollow.ts

@@ -1,71 +0,0 @@
-import FollowedLog from '../../../models/followed-log';
-import Following from '../../../models/following';
-import FollowingLog from '../../../models/following-log';
-import User, { isLocalUser, isRemoteUser, pack as packUser } from '../../../models/user';
-import stream from '../../../publishers/stream';
-import renderFollow from '../../../remote/activitypub/renderer/follow';
-import renderUndo from '../../../remote/activitypub/renderer/undo';
-import context from '../../../remote/activitypub/renderer/context';
-import request from '../../../remote/request';
-
-export default async ({ data }, done) => {
-	const following = await Following.findOne({ _id: data.id });
-	if (following === null) {
-		done();
-		return;
-	}
-
-	let follower;
-	let followee;
-
-	try {
-		[follower, followee] = await Promise.all([
-			User.findOne({ _id: following.followerId }),
-			User.findOne({ _id: following.followeeId })
-		]);
-
-		if (isLocalUser(follower) && isRemoteUser(followee)) {
-			const undo = renderUndo(renderFollow(follower, followee));
-			undo['@context'] = context;
-
-			await request(follower, followee.account.inbox, undo);
-		}
-	} catch (error) {
-		done(error);
-		return;
-	}
-
-	try {
-		await Promise.all([
-			// Delete following
-			Following.findOneAndDelete({ _id: data.id }),
-
-			// Decrement following count
-			User.update({ _id: follower._id }, { $inc: { followingCount: -1 } }),
-			FollowingLog.insert({
-				createdAt: new Date(),
-				userId: follower._id,
-				count: follower.followingCount - 1
-			}),
-
-			// Decrement followers count
-			User.update({ _id: followee._id }, { $inc: { followersCount: -1 } }),
-			FollowedLog.insert({
-				createdAt: new Date(),
-				userId: followee._id,
-				count: followee.followersCount - 1
-			})
-		]);
-
-		if (isLocalUser(follower)) {
-			return;
-		}
-
-		const promisedPackedUser = packUser(followee, follower);
-
-		// Publish follow event
-		stream(follower._id, 'unfollow', promisedPackedUser);
-	} finally {
-		done();
-	}
-};

src/remote/activitypub/act/create.ts

@@ -1,10 +0,0 @@
-import create from '../create';
-import Resolver from '../resolver';
-
-export default (resolver: Resolver, actor, activity, distribute) => {
-	if ('actor' in activity && actor.account.uri !== activity.actor) {
-		throw new Error();
-	}
-
-	return create(resolver, actor, activity.object, distribute);
-};

src/remote/activitypub/act/create/image.ts

@@ -0,0 +1,18 @@
+import * as debug from 'debug';
+
+import uploadFromUrl from '../../../../services/drive/upload-from-url';
+import { IRemoteUser } from '../../../../models/user';
+import { IDriveFile } from '../../../../models/drive-file';
+
+const log = debug('misskey:activitypub');
+
+export default async function(actor: IRemoteUser, image): Promise<IDriveFile> {
+	if ('attributedTo' in image && actor.account.uri !== image.attributedTo) {
+		log(`invalid image: ${JSON.stringify(image, null, 2)}`);
+		throw new Error('invalid image');
+	}
+
+	log(`Creating the Image: ${image.id}`);
+
+	return await uploadFromUrl(image.url, actor);
+}

src/remote/activitypub/act/create/index.ts

@@ -0,0 +1,44 @@
+import * as debug from 'debug';
+
+import Resolver from '../../resolver';
+import { IRemoteUser } from '../../../../models/user';
+import createNote from './note';
+import createImage from './image';
+import { ICreate } from '../../type';
+
+const log = debug('misskey:activitypub');
+
+export default async (actor: IRemoteUser, activity: ICreate): Promise<void> => {
+	if ('actor' in activity && actor.account.uri !== activity.actor) {
+		throw new Error('invalid actor');
+	}
+
+	const uri = activity.id || activity;
+
+	log(`Create: ${uri}`);
+
+	const resolver = new Resolver();
+
+	let object;
+
+	try {
+		object = await resolver.resolve(activity.object);
+	} catch (e) {
+		log(`Resolution failed: ${e}`);
+		throw e;
+	}
+
+	switch (object.type) {
+	case 'Image':
+		createImage(actor, object);
+		break;
+
+	case 'Note':
+		createNote(resolver, actor, object);
+		break;
+
+	default:
+		console.warn(`Unknown type: ${object.type}`);
+		break;
+	}
+};

src/remote/activitypub/act/create/note.ts

@@ -0,0 +1,89 @@
+import { JSDOM } from 'jsdom';
+import * as debug from 'debug';
+
+import Resolver from '../../resolver';
+import Post, { IPost } from '../../../../models/post';
+import createPost from '../../../../services/post/create';
+import { IRemoteUser } from '../../../../models/user';
+import resolvePerson from '../../resolve-person';
+import createImage from './image';
+import config from '../../../../config';
+
+const log = debug('misskey:activitypub');
+
+/**
+ * 投稿作成アクティビティを捌きます
+ */
+export default async function createNote(resolver: Resolver, actor: IRemoteUser, note, silent = false): Promise<IPost> {
+	if (typeof note.id !== 'string') {
+		log(`invalid note: ${JSON.stringify(note, null, 2)}`);
+		throw new Error('invalid note');
+	}
+
+	// 既に同じURIを持つものが登録されていないかチェックし、登録されていたらそれを返す
+	const exist = await Post.findOne({ uri: note.id });
+	if (exist) {
+		return exist;
+	}
+
+	log(`Creating the Note: ${note.id}`);
+
+	//#region Visibility
+	let visibility = 'public';
+	if (!note.to.includes('https://www.w3.org/ns/activitystreams#Public')) visibility = 'unlisted';
+	if (note.cc.length == 0) visibility = 'private';
+	// TODO
+	if (visibility != 'public') throw new Error('unspported visibility');
+	//#endergion
+
+	//#region 添付メディア
+	const media = [];
+	if ('attachment' in note && note.attachment != null) {
+		// TODO: attachmentは必ずしもImageではない
+		// TODO: attachmentは必ずしも配列ではない
+		// TODO: ループの中でawaitはすべきでない
+		note.attachment.forEach(async media => {
+			const created = await createImage(note.actor, media);
+			media.push(created);
+		});
+	}
+	//#endregion
+
+	//#region リプライ
+	let reply = null;
+	if ('inReplyTo' in note && note.inReplyTo != null) {
+		// リプライ先の投稿がMisskeyに登録されているか調べる
+		const uri: string = note.inReplyTo.id || note.inReplyTo;
+		const inReplyToPost = uri.startsWith(config.url + '/')
+			? await Post.findOne({ _id: uri.split('/').pop() })
+			: await Post.findOne({ uri });
+
+		if (inReplyToPost) {
+			reply = inReplyToPost;
+		} else {
+			// 無かったらフェッチ
+			const inReplyTo = await resolver.resolve(note.inReplyTo) as any;
+
+			// リプライ先の投稿の投稿者をフェッチ
+			const actor = await resolvePerson(inReplyTo.attributedTo) as IRemoteUser;
+
+			// TODO: silentを常にtrueにしてはならない
+			reply = await createNote(resolver, actor, inReplyTo);
+		}
+	}
+	//#endregion
+
+	const { window } = new JSDOM(note.content);
+
+	return await createPost(actor, {
+		createdAt: new Date(note.published),
+		media,
+		reply,
+		repost: undefined,
+		text: window.document.body.textContent,
+		viaMobile: false,
+		geo: undefined,
+		visibility,
+		uri: note.id
+	});
+}

src/remote/activitypub/act/delete.ts

@@ -1,21 +0,0 @@
-import create from '../create';
-import deleteObject from '../delete';
-
-export default async (resolver, actor, activity) => {
-	if ('actor' in activity && actor.account.uri !== activity.actor) {
-		throw new Error();
-	}
-
-	const results = await create(resolver, actor, activity.object);
-
-	await Promise.all(results.map(async promisedResult => {
-		const result = await promisedResult;
-		if (result === null) {
-			return;
-		}
-
-		await deleteObject(result);
-	}));
-
-	return null;
-};

src/remote/activitypub/act/delete/index.ts

@@ -0,0 +1,36 @@
+import Resolver from '../../resolver';
+import deleteNote from './note';
+import Post from '../../../../models/post';
+import { IRemoteUser } from '../../../../models/user';
+
+/**
+ * 削除アクティビティを捌きます
+ */
+export default async (actor: IRemoteUser, activity): Promise<void> => {
+	if ('actor' in activity && actor.account.uri !== activity.actor) {
+		throw new Error('invalid actor');
+	}
+
+	const resolver = new Resolver();
+
+	const object = await resolver.resolve(activity.object);
+
+	const uri = (object as any).id;
+
+	switch (object.type) {
+	case 'Note':
+		deleteNote(actor, uri);
+		break;
+
+	case 'Tombstone':
+		const post = await Post.findOne({ uri });
+		if (post != null) {
+			deleteNote(actor, uri);
+		}
+		break;
+
+	default:
+		console.warn(`Unknown type: ${object.type}`);
+		break;
+	}
+};

src/remote/activitypub/act/delete/note.ts

@@ -0,0 +1,30 @@
+import * as debug from 'debug';
+
+import Post from '../../../../models/post';
+import { IRemoteUser } from '../../../../models/user';
+
+const log = debug('misskey:activitypub');
+
+export default async function(actor: IRemoteUser, uri: string): Promise<void> {
+	log(`Deleting the Note: ${uri}`);
+
+	const post = await Post.findOne({ uri });
+
+	if (post == null) {
+		throw new Error('post not found');
+	}
+
+	if (!post.userId.equals(actor._id)) {
+		throw new Error('投稿を削除しようとしているユーザーは投稿の作成者ではありません');
+	}
+
+	Post.update({ _id: post._id }, {
+		$set: {
+			deletedAt: new Date(),
+			text: null,
+			textHtml: null,
+			mediaIds: [],
+			poll: null
+		}
+	});
+}

src/remote/activitypub/act/follow.ts

@@ -1,17 +1,12 @@
-import { MongoError } from 'mongodb';
 import parseAcct from '../../../acct/parse';
-import Following, { IFollowing } from '../../../models/following';
+import User, { IRemoteUser } from '../../../models/user';
-import User from '../../../models/user';
 import config from '../../../config';
-import { createHttp } from '../../../queue';
+import follow from '../../../services/following/create';
-import context from '../renderer/context';
+import { IFollow } from '../type';
-import renderAccept from '../renderer/accept';
-import request from '../../request';
-import Resolver from '../resolver';
 
-export default async (resolver: Resolver, actor, activity, distribute) => {
+export default async (actor: IRemoteUser, activity: IFollow): Promise<void> => {
 	const prefix = config.url + '/@';
-	const id = activity.object.id || activity.object;
+	const id = typeof activity == 'string' ? activity : activity.id;
 
 	if (!id.startsWith(prefix)) {
 		return null;
@@ -27,52 +22,5 @@ export default async (resolver: Resolver, actor, activity, distribute) => {
 		throw new Error();
 	}
 
-	if (!distribute) {
+	await follow(actor, followee, activity);
-		const { _id } = await Following.findOne({
-			followerId: actor._id,
-			followeeId: followee._id
-		});
-
-		return {
-			resolver,
-			object: { $ref: 'following', $id: _id }
-		};
-	}
-
-	const promisedFollowing = Following.insert({
-		createdAt: new Date(),
-		followerId: actor._id,
-		followeeId: followee._id
-	}).then(following => new Promise((resolve, reject) => {
-		createHttp({
-			type: 'follow',
-			following: following._id
-		}).save(error => {
-			if (error) {
-				reject(error);
-			} else {
-				resolve(following);
-			}
-		});
-	}) as Promise<IFollowing>, async error => {
-		// duplicate key error
-		if (error instanceof MongoError && error.code === 11000) {
-			return Following.findOne({
-				followerId: actor._id,
-				followeeId: followee._id
-			});
-		}
-
-		throw error;
-	});
-
-	const accept = renderAccept(activity);
-	accept['@context'] = context;
-
-	await request(followee, actor.account.inbox, accept);
-
-	return promisedFollowing.then(({ _id }) => ({
-		resolver,
-		object: { $ref: 'following', $id: _id }
-	}));
 };

src/remote/activitypub/act/index.ts

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

src/remote/activitypub/act/undo/follow.ts

@@ -0,0 +1,26 @@
+import parseAcct from '../../../../acct/parse';
+import User, { IRemoteUser } from '../../../../models/user';
+import config from '../../../../config';
+import unfollow from '../../../../services/following/delete';
+import { IFollow } from '../../type';
+
+export default async (actor: IRemoteUser, activity: IFollow): Promise<void> => {
+	const prefix = config.url + '/@';
+	const id = typeof activity == 'string' ? activity : activity.id;
+
+	if (!id.startsWith(prefix)) {
+		return null;
+	}
+
+	const { username, host } = parseAcct(id.slice(prefix.length));
+	if (host !== null) {
+		throw new Error();
+	}
+
+	const followee = await User.findOne({ username, host });
+	if (followee === null) {
+		throw new Error();
+	}
+
+	await unfollow(actor, followee, activity);
+};

src/remote/activitypub/act/undo/index.ts

@@ -1,27 +1,37 @@
-import act from '../../act';
+import * as debug from 'debug';
-import deleteObject from '../../delete';
+
-import unfollow from './unfollow';
+import { IRemoteUser } from '../../../../models/user';
+import { IUndo } from '../../type';
+import unfollow from './follow';
 import Resolver from '../../resolver';
 
-export default async (resolver: Resolver, actor, activity): Promise<void> => {
+const log = debug('misskey:activitypub');
+
+export default async (actor: IRemoteUser, activity: IUndo): Promise<void> => {
 	if ('actor' in activity && actor.account.uri !== activity.actor) {
-		throw new Error();
+		throw new Error('invalid actor');
 	}
 
-	const results = await act(resolver, actor, activity.object);
+	const uri = activity.id || activity;
 
-	await Promise.all(results.map(async promisedResult => {
+	log(`Undo: ${uri}`);
-		const result = await promisedResult;
 
-		if (result === null || await deleteObject(result) !== null) {
+	const resolver = new Resolver();
-			return;
-		}
 
-		switch (result.object.$ref) {
+	let object;
-		case 'following':
+
-			await unfollow(result.object);
+	try {
-		}
+		object = await resolver.resolve(activity.object);
-	}));
+	} catch (e) {
+		log(`Resolution failed: ${e}`);
+		throw e;
+	}
+
+	switch (object.type) {
+		case 'Follow':
+			unfollow(actor, object);
+			break;
+	}
 
 	return null;
 };

src/remote/activitypub/act/undo/unfollow.ts

@@ -1,11 +0,0 @@
-import { createHttp } from '../../../../queue';
-
-export default ({ $id }) => new Promise((resolve, reject) => {
-	createHttp({ type: 'unfollow', id: $id }).save(error => {
-		if (error) {
-			reject(error);
-		} else {
-			resolve();
-		}
-	});
-});

src/remote/activitypub/create.ts

@@ -1,222 +0,0 @@
-import { JSDOM } from 'jsdom';
-import { ObjectID } from 'mongodb';
-import parseAcct from '../../acct/parse';
-import config from '../../config';
-import DriveFile from '../../models/drive-file';
-import Post from '../../models/post';
-import User from '../../models/user';
-import { IRemoteUser } from '../../models/user';
-import uploadFromUrl from '../../drive/upload-from-url';
-import createPost from '../../post/create';
-import distributePost from '../../post/distribute';
-import resolvePerson from './resolve-person';
-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 { window } = new JSDOM(note.content);
-		const mentions = [];
-		const tags = [];
-
-		for (const { href, name, type } of note.tags) {
-			switch (type) {
-			case 'Hashtag':
-				if (name.startsWith('#')) {
-					tags.push(name.slice(1));
-				}
-				break;
-
-			case 'Mention':
-				mentions.push(resolvePerson(resolver, href));
-				break;
-			}
-		}
-
-		const [mediaIds, reply] = await Promise.all([
-			'attachment' in note && this.create(resolver, note.attachment)
-				.then(collection => Promise.all(collection))
-				.then(collection => collection
-					.filter(media => media !== null && media.object.$ref === 'driveFiles.files')
-					.map(({ object }: IResult) => object.$id)),
-
-			'inReplyTo' in note && this.create(resolver, note.inReplyTo)
-				.then(collection => Promise.all(collection.map(promise => promise.then(result => {
-					if (result !== null && result.object.$ref === 'posts') {
-						throw result.object;
-					}
-				}, () => { }))))
-				.then(() => null, ({ $id }) => Post.findOne({ _id: $id }))
-		]);
-
-		const inserted = await createPost({
-			channelId: undefined,
-			index: undefined,
-			createdAt: new Date(note.published),
-			mediaIds,
-			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,
-			tags
-		}, reply, null, await Promise.all(mentions));
-
-		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;
-			const localPrefix = config.url + '/@';
-
-			if (uri.startsWith(localPrefix)) {
-				const [acct, id] = uri.slice(localPrefix).split('/', 2);
-				const user = await User.aggregate([
-					{
-						$match: parseAcct(acct)
-					},
-					{
-						$lookup: {
-							from: 'posts',
-							localField: '_id',
-							foreignField: 'userId',
-							as: 'post'
-						}
-					},
-					{
-						$match: {
-							post: { _id: id }
-						}
-					}
-				]);
-
-				if (user === null || user.posts.length <= 0) {
-					throw new Error();
-				}
-
-				return {
-					resolver: collection.resolver,
-					object: {
-						$ref: 'posts',
-						id
-					}
-				};
-			}
-
-			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/delete/index.ts

@@ -1,10 +0,0 @@
-import deletePost from './post';
-
-export default async ({ object }) => {
-	switch (object.$ref) {
-	case 'posts':
-		return deletePost(object);
-	}
-
-	return null;
-};

src/remote/activitypub/delete/post.ts

@@ -1,13 +0,0 @@
-import Post from '../../../models/post';
-import { createDb } from '../../../queue';
-
-export default async ({ $id }) => {
-	const promisedDeletion = Post.findOneAndDelete({ _id: $id });
-
-	await new Promise((resolve, reject) => createDb({
-		type: 'deletePostDependents',
-		id: $id
-	}).delay(65536).save(error => error ? reject(error) : resolve()));
-
-	return promisedDeletion;
-};

src/remote/activitypub/renderer/note.ts

@@ -2,11 +2,14 @@ import renderDocument from './document';
 import renderHashtag from './hashtag';
 import config from '../../../config';
 import DriveFile from '../../../models/drive-file';
-import Post from '../../../models/post';
+import Post, { IPost } from '../../../models/post';
-import User from '../../../models/user';
+import User, { IUser } from '../../../models/user';
+
+export default async (user: IUser, post: IPost) => {
+	const promisedFiles = post.mediaIds
+		? DriveFile.find({ _id: { $in: post.mediaIds } })
+		: Promise.resolve([]);
 
-export default async (user, post) => {
-	const promisedFiles = DriveFile.find({ _id: { $in: post.mediaIds } });
 	let inReplyTo;
 
 	if (post.replyId) {
@@ -16,11 +19,11 @@ export default async (user, post) => {
 
 		if (inReplyToPost !== null) {
 			const inReplyToUser = await User.findOne({
-				_id: post.userId,
+				_id: inReplyToPost.userId,
 			});
 
 			if (inReplyToUser !== null) {
-				inReplyTo = `${config.url}@${inReplyToUser.username}/${inReplyToPost._id}`;
+				inReplyTo = inReplyToPost.uri || `${config.url}/@${inReplyToUser.username}/${inReplyToPost._id}`;
 			}
 		}
 	} else {
@@ -39,6 +42,6 @@ export default async (user, post) => {
 		cc: `${attributedTo}/followers`,
 		inReplyTo,
 		attachment: (await promisedFiles).map(renderDocument),
-		tag: post.tags.map(renderHashtag)
+		tag: (post.tags || []).map(renderHashtag)
 	};
 };

src/remote/activitypub/resolve-person.ts

@@ -3,15 +3,12 @@ import { toUnicode } from 'punycode';
 import parseAcct from '../../acct/parse';
 import config from '../../config';
 import User, { validateUsername, isValidName, isValidDescription } from '../../models/user';
-import { createHttp } from '../../queue';
 import webFinger from '../webfinger';
-import create from './create';
+import Resolver from './resolver';
+import uploadFromUrl from '../../services/drive/upload-from-url';
+import { isCollectionOrOrderedCollection } from './type';
 
-async function isCollection(collection) {
+export default async (value, verifier?: string) => {
-	return ['Collection', 'OrderedCollection'].includes(collection.type);
-}
-
-export default async (parentResolver, value, verifier?: string) => {
 	const id = value.id || value;
 	const localPrefix = config.url + '/@';
 
@@ -19,34 +16,35 @@ export default async (parentResolver, value, verifier?: string) => {
 		return User.findOne(parseAcct(id.slice(localPrefix)));
 	}
 
-	const { resolver, object } = await parentResolver.resolveOne(value);
+	const resolver = new Resolver();
+
+	const object = await resolver.resolve(value) as any;
 
 	if (
-		object === null ||
+		object == null ||
-		object.id !== id ||
 		object.type !== 'Person' ||
 		typeof object.preferredUsername !== 'string' ||
 		!validateUsername(object.preferredUsername) ||
-		!isValidName(object.name) ||
+		!isValidName(object.name == '' ? null : object.name) ||
 		!isValidDescription(object.summary)
 	) {
-		throw new Error();
+		throw new Error('invalid person');
 	}
 
-	const [followers, following, outbox, finger] = await Promise.all([
+	const [followersCount = 0, followingCount = 0, postsCount = 0, finger] = await Promise.all([
-		resolver.resolveOne(object.followers).then(
+		resolver.resolve(object.followers).then(
-			resolved => isCollection(resolved.object) ? resolved.object : null,
+			resolved => isCollectionOrOrderedCollection(resolved) ? resolved.totalItems : undefined,
-			() => null
+			() => undefined
 		),
-		resolver.resolveOne(object.following).then(
+		resolver.resolve(object.following).then(
-			resolved => isCollection(resolved.object) ? resolved.object : null,
+			resolved => isCollectionOrOrderedCollection(resolved) ? resolved.totalItems : undefined,
-			() => null
+			() => undefined
 		),
-		resolver.resolveOne(object.outbox).then(
+		resolver.resolve(object.outbox).then(
-			resolved => isCollection(resolved.object) ? resolved.object : null,
+			resolved => isCollectionOrOrderedCollection(resolved) ? resolved.totalItems : undefined,
-			() => null
+			() => undefined
 		),
-		webFinger(id, verifier),
+		webFinger(id, verifier)
 	]);
 
 	const host = toUnicode(finger.subject.replace(/^.*?@/, ''));
@@ -57,12 +55,12 @@ export default async (parentResolver, value, verifier?: string) => {
 	const user = await User.insert({
 		avatarId: null,
 		bannerId: null,
-		createdAt: Date.parse(object.published),
+		createdAt: Date.parse(object.published) || null,
 		description: summaryDOM.textContent,
-		followersCount: followers ? followers.totalItem || 0 : 0,
+		followersCount,
-		followingCount: following ? following.totalItem || 0 : 0,
+		followingCount,
+		postsCount,
 		name: object.name,
-		postsCount: outbox ? outbox.totalItem || 0 : 0,
 		driveCapacity: 1024 * 1024 * 8, // 8MiB
 		username: object.preferredUsername,
 		usernameLower: object.preferredUsername.toLowerCase(),
@@ -78,34 +76,14 @@ export default async (parentResolver, value, verifier?: string) => {
 		},
 	});
 
-	createHttp({
+	const [avatarId, bannerId] = (await Promise.all([
-		type: 'performActivityPub',
-		actor: user._id,
-		outbox
-	}).save();
-
-	const [avatarId, bannerId] = await Promise.all([
 		object.icon,
 		object.image
-	].map(async value => {
+	].map(img =>
-		if (value === undefined) {
+		img == null
-			return null;
+			? Promise.resolve(null)
-		}
+			: uploadFromUrl(img.url, user)
-
+	))).map(file => file != null ? file._id : null);
-		try {
-			const created = await create(resolver, user, value);
-
-			await Promise.all(created.map(asyncCreated => asyncCreated.then(created => {
-				if (created !== null && created.object.$ref === 'driveFiles.files') {
-					throw created.object.$id;
-				}
-			}, () => {})));
-
-			return null;
-		} catch (id) {
-			return id;
-		}
-	}));
 
 	User.update({ _id: user._id }, { $set: { avatarId, bannerId } });
 

src/remote/activitypub/resolver.ts

@@ -1,20 +1,51 @@
-const request = require('request-promise-native');
+import * as request from 'request-promise-native';
+import * as debug from 'debug';
+import { IObject } from './type';
+
+const log = debug('misskey:activitypub:resolver');
 
 export default class Resolver {
-	private requesting: Set<string>;
+	private history: Set<string>;
 
-	constructor(iterable?: Iterable<string>) {
+	constructor() {
-		this.requesting = new Set(iterable);
+		this.history = new Set();
 	}
 
-	private async resolveUnrequestedOne(value) {
+	public async resolveCollection(value) {
-		if (typeof value !== 'string') {
+		const collection = typeof value === 'string'
-			return { resolver: this, object: value };
+			? 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}`);
 		}
 
-		const resolver = new Resolver(this.requesting);
+		return collection;
+	}
 
-		resolver.requesting.add(value);
+	public async resolve(value): Promise<IObject> {
+		if (value == null) {
+			throw new Error('resolvee is null (or undefined)');
+		}
+
+		if (typeof value !== 'string') {
+			return value;
+		}
+
+		if (this.history.has(value)) {
+			throw new Error('cannot resolve already resolved one');
+		}
+
+		this.history.add(value);
 
 		const object = await request({
 			url: value,
@@ -29,41 +60,11 @@ 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 };
+		log(`resolved: ${JSON.stringify(object, null, 2)}`);
-	}
 
-	public async resolveCollection(value) {
+		return object;
-		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);
 	}
 }

src/remote/activitypub/type.ts

@@ -1,3 +1,48 @@
-export type IObject = {
+export type Object = { [x: string]: any };
+
+export interface IObject {
+	'@context': string | object | any[];
 	type: string;
-};
+	id?: string;
+	summary?: string;
+}
+
+export interface IActivity extends IObject {
+	//type: 'Activity';
+	actor: IObject | string;
+	object: IObject | string;
+	target?: IObject | string;
+}
+
+export interface ICollection extends IObject {
+	type: 'Collection';
+	totalItems: number;
+	items: IObject | string | IObject[] | string[];
+}
+
+export interface IOrderedCollection extends IObject {
+	type: 'OrderedCollection';
+	totalItems: number;
+	orderedItems: IObject | string | IObject[] | string[];
+}
+
+export const isCollection = (object: IObject): object is ICollection =>
+	object.type === 'Collection';
+
+export const isOrderedCollection = (object: IObject): object is IOrderedCollection =>
+	object.type === 'OrderedCollection';
+
+export const isCollectionOrOrderedCollection = (object: IObject): object is ICollection | IOrderedCollection =>
+	isCollection(object) || isOrderedCollection(object);
+
+export interface ICreate extends IActivity {
+	type: 'Create';
+}
+
+export interface IUndo extends IActivity {
+	type: 'Undo';
+}
+
+export interface IFollow extends IActivity {
+	type: 'Follow';
+}

src/remote/request.ts

@@ -1,9 +1,15 @@
 import { request } from 'https';
 import { sign } from 'http-signature';
 import { URL } from 'url';
+import * as debug from 'debug';
+
 import config from '../config';
 
+const log = debug('misskey:activitypub:deliver');
+
 export default ({ account, username }, url, object) => new Promise((resolve, reject) => {
+	log(`--> ${url}`);
+
 	const { protocol, hostname, port, pathname, search } = new URL(url);
 
 	const req = request({
@@ -14,6 +20,8 @@ export default ({ account, username }, url, object) => new Promise((resolve, rej
 		path: pathname + search,
 	}, res => {
 		res.on('end', () => {
+			log(`${url} --> ${res.statusCode}`);
+
 			if (res.statusCode >= 200 && res.statusCode < 300) {
 				resolve();
 			} else {

src/remote/resolve-user.ts

@@ -1,7 +1,6 @@
 import { toUnicode, toASCII } from 'punycode';
 import User from '../models/user';
 import resolvePerson from './activitypub/resolve-person';
-import Resolver from './activitypub/resolver';
 import webFinger from './webfinger';
 
 export default async (username, host, option) => {
@@ -17,10 +16,10 @@ export default async (username, host, option) => {
 		const finger = await webFinger(acctLower, acctLower);
 		const self = finger.links.find(link => link.rel && link.rel.toLowerCase() === 'self');
 		if (!self) {
-			throw new Error();
+			throw new Error('self link not found');
 		}
 
-		user = await resolvePerson(new Resolver(), self.href, acctLower);
+		user = await resolvePerson(self.href, acctLower);
 	}
 
 	return user;

src/server/activitypub/inbox.ts

@@ -3,9 +3,7 @@ import * as express from 'express';
 import { parseRequest } from 'http-signature';
 import { createHttp } from '../../queue';
 
-const app = express();
+const app = express.Router();
-
-app.disable('x-powered-by');
 
 app.post('/@:user/inbox', bodyParser.json({
 	type() {
@@ -24,7 +22,7 @@ app.post('/@:user/inbox', bodyParser.json({
 
 	createHttp({
 		type: 'processInbox',
-		inbox: req.body,
+		activity: req.body,
 		signature,
 	}).save();
 

src/server/activitypub/outbox.ts

@@ -6,8 +6,7 @@ import config from '../../config';
 import Post from '../../models/post';
 import withUser from './with-user';
 
-const app = express();
+const app = express.Router();
-app.disable('x-powered-by');
 
 app.get('/@:user/outbox', withUser(username => {
 	return `${config.url}/@${username}/inbox`;

src/server/activitypub/post.ts

@@ -5,8 +5,7 @@ import parseAcct from '../../acct/parse';
 import Post from '../../models/post';
 import User from '../../models/user';
 
-const app = express();
+const app = express.Router();
-app.disable('x-powered-by');
 
 app.get('/@:user/:post', async (req, res, next) => {
 	const accepted = req.accepts(['html', 'application/activity+json', 'application/ld+json']);

src/server/activitypub/publickey.ts

@@ -4,8 +4,7 @@ import render from '../../remote/activitypub/renderer/key';
 import config from '../../config';
 import withUser from './with-user';
 
-const app = express();
+const app = express.Router();
-app.disable('x-powered-by');
 
 app.get('/@:user/publickey', withUser(username => {
 	return `${config.url}/@${username}/publickey`;

src/server/activitypub/user.ts

@@ -11,8 +11,7 @@ const respond = withUser(username => `${config.url}/@${username}`, (user, req, r
 	res.json(rendered);
 });
 
-const app = express();
+const app = express.Router();
-app.disable('x-powered-by');
 
 app.get('/@:user', (req, res, next) => {
 	const accepted = req.accepts(['html', 'application/activity+json', 'application/ld+json']);

src/server/api/endpoints/following/create.ts

@@ -4,7 +4,7 @@
 import $ from 'cafy';
 import User from '../../../../models/user';
 import Following from '../../../../models/following';
-import { createHttp } from '../../../../queue';
+import create from '../../../../services/following/create';
 
 /**
  * Follow a user
@@ -50,15 +50,8 @@ module.exports = (params, user) => new Promise(async (res, rej) => {
 	}
 
 	// Create following
-	const { _id } = await Following.insert({
+	create(follower, followee);
-		createdAt: new Date(),
-		followerId: follower._id,
-		followeeId: followee._id
-	});
-
-	createHttp({ type: 'follow', following: _id }).save();
 
 	// Send response
 	res();
-
 });

src/server/api/endpoints/posts/create.ts

@@ -3,17 +3,12 @@
  */
 import $ from 'cafy';
 import deepEqual = require('deep-equal');
-import parseAcct from '../../../../acct/parse';
+import Post, { IPost, isValidText, isValidCw, pack } from '../../../../models/post';
-import renderAcct from '../../../../acct/render';
+import { ILocalUser } from '../../../../models/user';
-import config from '../../../../config';
-import html from '../../../../text/html';
-import parse from '../../../../text/parse';
-import Post, { IPost, isValidText, isValidCw } from '../../../../models/post';
-import User, { ILocalUser } from '../../../../models/user';
 import Channel, { IChannel } from '../../../../models/channel';
 import DriveFile from '../../../../models/drive-file';
-import create from '../../../../post/create';
+import create from '../../../../services/post/create';
-import distribute from '../../../../post/distribute';
+import { IApp } from '../../../../models/app';
 
 /**
  * Create a post
@@ -23,7 +18,7 @@ import distribute from '../../../../post/distribute';
  * @param {any} app
  * @return {Promise<any>}
  */
-module.exports = (params, user: ILocalUser, app) => new Promise(async (res, rej) => {
+module.exports = (params, user: ILocalUser, app: IApp) => new Promise(async (res, rej) => {
 	// Get 'visibility' parameter
 	const [visibility = 'public', visibilityErr] = $(params.visibility).optional.string().or(['public', 'unlisted', 'private', 'direct']).$;
 	if (visibilityErr) return rej('invalid visibility');
@@ -231,85 +226,26 @@ module.exports = (params, user: ILocalUser, app) => new Promise(async (res, rej)
 		}
 	}
 
-	let tokens = null;
-	if (text) {
-		// Analyze
-		tokens = parse(text);
-
-		// Extract hashtags
-		const hashtags = tokens
-			.filter(t => t.type == 'hashtag')
-			.map(t => t.hashtag);
-
-		hashtags.forEach(tag => {
-			if (tags.indexOf(tag) == -1) {
-				tags.push(tag);
-			}
-		});
-	}
-
-	let atMentions = [];
-
-	// If has text content
-	if (text) {
-		/*
-				// Extract a hashtags
-				const hashtags = tokens
-					.filter(t => t.type == 'hashtag')
-					.map(t => t.hashtag)
-					// Drop dupulicates
-					.filter((v, i, s) => s.indexOf(v) == i);
-
-				// ハッシュタグをデータベースに登録
-				registerHashtags(user, hashtags);
-		*/
-		// Extract an '@' mentions
-		atMentions = tokens
-			.filter(t => t.type == 'mention')
-			.map(renderAcct)
-			// Drop dupulicates
-			.filter((v, i, s) => s.indexOf(v) == i)
-			// Fetch mentioned user
-			// SELECT _id
-			.map(mention => User.findOne(parseAcct(mention), { _id: true }));
-	}
-
 	// 投稿を作成
-	const post = await create({
+	const post = await create(user, {
 		createdAt: new Date(),
-		channelId: channel ? channel._id : undefined,
+		media: files,
-		index: channel ? channel.index + 1 : undefined,
-		mediaIds: files ? files.map(file => file._id) : [],
 		poll: poll,
 		text: text,
-		textHtml: tokens === null ? null : html(tokens),
+		reply,
+		repost,
 		cw: cw,
 		tags: tags,
-		userId: user._id,
+		app: app,
-		appId: app ? app._id : null,
 		viaMobile: viaMobile,
 		visibility,
 		geo
-	}, reply, repost, await Promise.all(atMentions));
+	});
 
-	const postObj = await distribute(user, post.mentions, post);
+	const postObj = await pack(post, user);
 
 	// Reponse
 	res({
 		createdPost: postObj
 	});
-
-	// Register to search database
-	if (post.text && config.elasticsearch.enable) {
-		const es = require('../../../db/elasticsearch');
-
-		es.index({
-			index: 'misskey',
-			type: 'post',
-			id: post._id.toString(),
-			body: {
-				text: post.text
-			}
-		});
-	}
 });

src/server/api/endpoints/users/show.ts

@@ -37,7 +37,8 @@ module.exports = (params, me) => new Promise(async (res, rej) => {
 	if (typeof host === 'string') {
 		try {
 			user = await resolveRemoteUser(username, host, cursorOption);
-		} catch (exception) {
+		} catch (e) {
+			console.warn(`failed to resolve remote user: ${e}`);
 			return rej('failed to resolve remote user');
 		}
 	} else {

src/server/web/index.ts

@@ -11,7 +11,7 @@ import * as bodyParser from 'body-parser';
 import * as favicon from 'serve-favicon';
 import * as compression from 'compression';
 
-const client = `${__dirname}/../../client/`;
+const client = path.resolve(`${__dirname}/../../client/`);
 
 // Create server
 const app = express();

src/server/webfinger.ts

@@ -1,11 +1,12 @@
+import * as express from 'express';
+
 import config from '../config';
 import parseAcct from '../acct/parse';
 import User from '../models/user';
-const express = require('express');
 
 const app = express();
 
-app.get('/.well-known/webfinger', async (req, res) => {
+app.get('/.well-known/webfinger', async (req: express.Request, res: express.Response) => {
 	if (typeof req.query.resource !== 'string') {
 		return res.sendStatus(400);
 	}
@@ -34,13 +35,15 @@ app.get('/.well-known/webfinger', async (req, res) => {
 
 	return res.json({
 		subject: `acct:${user.username}@${config.host}`,
-		links: [
+		links: [{
-			{
+			rel: 'self',
-				rel: 'self',
+			type: 'application/activity+json',
-				type: 'application/activity+json',
+			href: `${config.url}/@${user.username}`
-				href: `${config.url}/@${user.username}`
+		}, {
-			}
+			rel: 'http://webfinger.net/rel/profile-page',
-		]
+			type: 'text/html',
+			href: `${config.url}/@${user.username}`
+		}]
 	});
 });
 

src/services/drive/add-file.ts

@@ -10,12 +10,12 @@ import * as debug from 'debug';
 import fileType = require('file-type');
 import prominence = require('prominence');
 
-import DriveFile, { IMetadata, getGridFSBucket } from '../models/drive-file';
+import DriveFile, { IMetadata, getGridFSBucket } from '../../models/drive-file';
-import DriveFolder from '../models/drive-folder';
+import DriveFolder from '../../models/drive-folder';
-import { pack } from '../models/drive-file';
+import { pack } from '../../models/drive-file';
-import event, { publishDriveStream } from '../publishers/stream';
+import event, { publishDriveStream } from '../../publishers/stream';
-import getAcct from '../acct/render';
+import getAcct from '../../acct/render';
-import config from '../config';
+import config from '../../config';
 
 const gm = _gm.subClass({
 	imageMagick: true

src/services/drive/upload-from-url.ts

@@ -1,19 +1,23 @@
 import * as URL from 'url';
-import { IDriveFile, validateFileName } from '../models/drive-file';
+import { IDriveFile, validateFileName } from '../../models/drive-file';
 import create from './add-file';
 import * as debug from 'debug';
 import * as tmp from 'tmp';
 import * as fs from 'fs';
 import * as request from 'request';
 
-const log = debug('misskey:common:drive:upload_from_url');
+const log = debug('misskey:drive:upload-from-url');
 
 export default async (url, user, folderId = null, uri = null): Promise<IDriveFile> => {
+	log(`REQUESTED: ${url}`);
+
 	let name = URL.parse(url).pathname.split('/').pop();
 	if (!validateFileName(name)) {
 		name = null;
 	}
 
+	log(`name: ${name}`);
+
 	// Create temp file
 	const path = await new Promise((res: (string) => void, rej) => {
 		tmp.file((e, path) => {
@@ -37,6 +41,8 @@ export default async (url, user, folderId = null, uri = null): Promise<IDriveFil
 
 	const driveFile = await create(user, path, name, null, folderId, false, uri);
 
+	log(`created: ${driveFile._id}`);
+
 	// clean-up
 	fs.unlink(path, (e) => {
 		if (e) log(e.stack);

src/services/following/create.ts

@@ -0,0 +1,72 @@
+import User, { isLocalUser, isRemoteUser, pack as packUser, IUser } from '../../models/user';
+import Following from '../../models/following';
+import FollowingLog from '../../models/following-log';
+import FollowedLog from '../../models/followed-log';
+import event from '../../publishers/stream';
+import notify from '../../publishers/notify';
+import context from '../../remote/activitypub/renderer/context';
+import renderFollow from '../../remote/activitypub/renderer/follow';
+import renderAccept from '../../remote/activitypub/renderer/accept';
+import { deliver } from '../../queue';
+
+export default async function(follower: IUser, followee: IUser, activity?) {
+	const following = await Following.insert({
+		createdAt: new Date(),
+		followerId: follower._id,
+		followeeId: followee._id
+	});
+
+	//#region Increment following count
+	User.update({ _id: follower._id }, {
+		$inc: {
+			followingCount: 1
+		}
+	});
+
+	FollowingLog.insert({
+		createdAt: following.createdAt,
+		userId: follower._id,
+		count: follower.followingCount + 1
+	});
+	//#endregion
+
+	//#region Increment followers count
+	User.update({ _id: followee._id }, {
+		$inc: {
+			followersCount: 1
+		}
+	});
+	FollowedLog.insert({
+		createdAt: following.createdAt,
+		userId: followee._id,
+		count: followee.followersCount + 1
+	});
+	//#endregion
+
+	// Publish follow event
+	if (isLocalUser(follower)) {
+		packUser(followee, follower).then(packed => event(follower._id, 'follow', packed));
+	}
+
+	// Publish followed event
+	if (isLocalUser(followee)) {
+		packUser(follower, followee).then(packed => event(followee._id, 'followed', packed)),
+
+		// 通知を作成
+		notify(followee._id, follower._id, 'follow');
+	}
+
+	if (isLocalUser(follower) && isRemoteUser(followee)) {
+		const content = renderFollow(follower, followee);
+		content['@context'] = context;
+
+		deliver(follower, content, followee.account.inbox).save();
+	}
+
+	if (isRemoteUser(follower) && isLocalUser(followee)) {
+		const content = renderAccept(activity);
+		content['@context'] = context;
+
+		deliver(followee, content, follower.account.inbox).save();
+	}
+}

src/services/following/delete.ts

@@ -0,0 +1,64 @@
+import User, { isLocalUser, isRemoteUser, pack as packUser, IUser } from '../../models/user';
+import Following from '../../models/following';
+import FollowingLog from '../../models/following-log';
+import FollowedLog from '../../models/followed-log';
+import event from '../../publishers/stream';
+import context from '../../remote/activitypub/renderer/context';
+import renderFollow from '../../remote/activitypub/renderer/follow';
+import renderUndo from '../../remote/activitypub/renderer/undo';
+import { deliver } from '../../queue';
+
+export default async function(follower: IUser, followee: IUser, activity?) {
+	const following = await Following.findOne({
+		followerId: follower._id,
+		followeeId: followee._id
+	});
+
+	if (following == null) {
+		console.warn('フォロー解除がリクエストされましたがフォローしていませんでした');
+		return;
+	}
+
+	Following.remove({
+		_id: following._id
+	});
+
+	//#region Decrement following count
+	User.update({ _id: follower._id }, {
+		$inc: {
+			followingCount: -1
+		}
+	});
+
+	FollowingLog.insert({
+		createdAt: following.createdAt,
+		userId: follower._id,
+		count: follower.followingCount - 1
+	});
+	//#endregion
+
+	//#region Decrement followers count
+	User.update({ _id: followee._id }, {
+		$inc: {
+			followersCount: -1
+		}
+	});
+	FollowedLog.insert({
+		createdAt: following.createdAt,
+		userId: followee._id,
+		count: followee.followersCount - 1
+	});
+	//#endregion
+
+	// Publish unfollow event
+	if (isLocalUser(follower)) {
+		packUser(followee, follower).then(packed => event(follower._id, 'unfollow', packed));
+	}
+
+	if (isLocalUser(follower) && isRemoteUser(followee)) {
+		const content = renderUndo(renderFollow(follower, followee));
+		content['@context'] = context;
+
+		deliver(follower, content, followee.account.inbox).save();
+	}
+}

src/services/post/create.ts

@@ -0,0 +1,358 @@
+import Post, { pack, IPost } from '../../models/post';
+import User, { isLocalUser, IUser, isRemoteUser } from '../../models/user';
+import stream from '../../publishers/stream';
+import Following from '../../models/following';
+import { deliver } from '../../queue';
+import renderNote from '../../remote/activitypub/renderer/note';
+import renderCreate from '../../remote/activitypub/renderer/create';
+import context from '../../remote/activitypub/renderer/context';
+import { IDriveFile } from '../../models/drive-file';
+import notify from '../../publishers/notify';
+import PostWatching from '../../models/post-watching';
+import watch from './watch';
+import Mute from '../../models/mute';
+import pushSw from '../../publishers/push-sw';
+import event from '../../publishers/stream';
+import parse from '../../text/parse';
+import html from '../../text/html';
+import { IApp } from '../../models/app';
+
+export default async (user: IUser, data: {
+	createdAt?: Date;
+	text?: string;
+	reply?: IPost;
+	repost?: IPost;
+	media?: IDriveFile[];
+	geo?: any;
+	poll?: any;
+	viaMobile?: boolean;
+	tags?: string[];
+	cw?: string;
+	visibility?: string;
+	uri?: string;
+	app?: IApp;
+}, silent = false) => new Promise<IPost>(async (res, rej) => {
+	if (data.createdAt == null) data.createdAt = new Date();
+	if (data.visibility == null) data.visibility = 'public';
+
+	const tags = data.tags || [];
+
+	let tokens = null;
+
+	if (data.text) {
+		// Analyze
+		tokens = parse(data.text);
+
+		// Extract hashtags
+		const hashtags = tokens
+			.filter(t => t.type == 'hashtag')
+			.map(t => t.hashtag);
+
+		hashtags.forEach(tag => {
+			if (tags.indexOf(tag) == -1) {
+				tags.push(tag);
+			}
+		});
+	}
+
+	const insert: any = {
+		createdAt: data.createdAt,
+		mediaIds: data.media ? data.media.map(file => file._id) : [],
+		replyId: data.reply ? data.reply._id : null,
+		repostId: data.repost ? data.repost._id : null,
+		text: data.text,
+		textHtml: tokens === null ? null : html(tokens),
+		poll: data.poll,
+		cw: data.cw,
+		tags,
+		userId: user._id,
+		viaMobile: data.viaMobile,
+		geo: data.geo || null,
+		appId: data.app ? data.app._id : null,
+		visibility: data.visibility,
+
+		// 以下非正規化データ
+		_reply: data.reply ? { userId: data.reply.userId } : null,
+		_repost: data.repost ? { userId: data.repost.userId } : null,
+		_user: {
+			host: user.host,
+			hostLower: user.hostLower,
+			account: isLocalUser(user) ? {} : {
+				inbox: user.account.inbox
+			}
+		}
+	};
+
+	if (data.uri != null) insert.uri = data.uri;
+
+	// 投稿を作成
+	const post = await Post.insert(insert);
+
+	res(post);
+
+	User.update({ _id: user._id }, {
+		// Increment posts count
+		$inc: {
+			postsCount: 1
+		},
+		// Update latest post
+		$set: {
+			latestPost: post
+		}
+	});
+
+	// Serialize
+	const postObj = await pack(post);
+
+	// タイムラインへの投稿
+	if (post.channelId == null) {
+		// Publish event to myself's stream
+		if (isLocalUser(user)) {
+			stream(post.userId, 'post', postObj);
+		}
+
+		// Fetch all followers
+		const followers = await Following.aggregate([{
+			$lookup: {
+				from: 'users',
+				localField: 'followerId',
+				foreignField: '_id',
+				as: 'user'
+			}
+		}, {
+			$match: {
+				followeeId: post.userId
+			}
+		}], {
+			_id: false
+		});
+
+		if (!silent) {
+			const note = await renderNote(user, post);
+			const content = renderCreate(note);
+			content['@context'] = context;
+
+			// 投稿がリプライかつ投稿者がローカルユーザーかつリプライ先の投稿の投稿者がリモートユーザーなら配送
+			if (data.reply && isLocalUser(user) && isRemoteUser(data.reply._user)) {
+				deliver(user, content, data.reply._user.account.inbox).save();
+			}
+
+			Promise.all(followers.map(follower => {
+				follower = follower.user[0];
+
+				if (isLocalUser(follower)) {
+					// Publish event to followers stream
+					stream(follower._id, 'post', postObj);
+				} else {
+					// フォロワーがリモートユーザーかつ投稿者がローカルユーザーなら投稿を配信
+					if (isLocalUser(user)) {
+						deliver(user, content, 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);
+				});
+			})
+		);
+	}*/
+
+	const mentions = [];
+
+	async function addMention(mentionee, reason) {
+		// Reject if already added
+		if (mentions.some(x => x.equals(mentionee))) return;
+
+		// Add mention
+		mentions.push(mentionee);
+
+		// Publish event
+		if (!user._id.equals(mentionee)) {
+			const mentioneeMutes = await Mute.find({
+				muter_id: mentionee,
+				deleted_at: { $exists: false }
+			});
+			const mentioneesMutedUserIds = mentioneeMutes.map(m => m.muteeId.toString());
+			if (mentioneesMutedUserIds.indexOf(user._id.toString()) == -1) {
+				event(mentionee, reason, postObj);
+				pushSw(mentionee, reason, postObj);
+			}
+		}
+	}
+
+	// If has in reply to post
+	if (data.reply) {
+		// Increment replies count
+		Post.update({ _id: data.reply._id }, {
+			$inc: {
+				repliesCount: 1
+			}
+		});
+
+		// (自分自身へのリプライでない限りは)通知を作成
+		notify(data.reply.userId, user._id, 'reply', {
+			postId: post._id
+		});
+
+		// Fetch watchers
+		PostWatching.find({
+			postId: data.reply._id,
+			userId: { $ne: user._id },
+			// 削除されたドキュメントは除く
+			deletedAt: { $exists: false }
+		}, {
+			fields: {
+				userId: true
+			}
+		}).then(watchers => {
+			watchers.forEach(watcher => {
+				notify(watcher.userId, user._id, 'reply', {
+					postId: post._id
+				});
+			});
+		});
+
+		// この投稿をWatchする
+		if (isLocalUser(user) && user.account.settings.autoWatch !== false) {
+			watch(user._id, data.reply);
+		}
+
+		// Add mention
+		addMention(data.reply.userId, 'reply');
+	}
+
+	// If it is repost
+	if (data.repost) {
+		// Notify
+		const type = data.text ? 'quote' : 'repost';
+		notify(data.repost.userId, user._id, type, {
+			post_id: post._id
+		});
+
+		// Fetch watchers
+		PostWatching.find({
+			postId: data.repost._id,
+			userId: { $ne: user._id },
+			// 削除されたドキュメントは除く
+			deletedAt: { $exists: false }
+		}, {
+			fields: {
+				userId: true
+			}
+		}).then(watchers => {
+			watchers.forEach(watcher => {
+				notify(watcher.userId, user._id, type, {
+					postId: post._id
+				});
+			});
+		});
+
+		// この投稿をWatchする
+		if (isLocalUser(user) && user.account.settings.autoWatch !== false) {
+			watch(user._id, data.repost);
+		}
+
+		// If it is quote repost
+		if (data.text) {
+			// Add mention
+			addMention(data.repost.userId, 'quote');
+		} else {
+			// Publish event
+			if (!user._id.equals(data.repost.userId)) {
+				event(data.repost.userId, 'repost', postObj);
+			}
+		}
+
+		// 今までで同じ投稿をRepostしているか
+		const existRepost = await Post.findOne({
+			userId: user._id,
+			repostId: data.repost._id,
+			_id: {
+				$ne: post._id
+			}
+		});
+
+		if (!existRepost) {
+			// Update repostee status
+			Post.update({ _id: data.repost._id }, {
+				$inc: {
+					repostCount: 1
+				}
+			});
+		}
+	}
+
+	// If has text content
+	if (data.text) {
+		// Extract an '@' mentions
+		const atMentions = tokens
+			.filter(t => t.type == 'mention')
+			.map(m => m.username)
+			// Drop dupulicates
+			.filter((v, i, s) => s.indexOf(v) == i);
+
+		// Resolve all mentions
+		await Promise.all(atMentions.map(async mention => {
+			// Fetch mentioned user
+			// SELECT _id
+			const mentionee = await User
+				.findOne({
+					usernameLower: mention.toLowerCase()
+				}, { _id: true });
+
+			// When mentioned user not found
+			if (mentionee == null) return;
+
+			// 既に言及されたユーザーに対する返信や引用repostの場合も無視
+			if (data.reply && data.reply.userId.equals(mentionee._id)) return;
+			if (data.repost && data.repost.userId.equals(mentionee._id)) return;
+
+			// Add mention
+			addMention(mentionee._id, 'mention');
+
+			// Create notification
+			notify(mentionee._id, user._id, 'mention', {
+				post_id: post._id
+			});
+		}));
+	}
+
+	// Append mentions data
+	if (mentions.length > 0) {
+		Post.update({ _id: post._id }, {
+			$set: {
+				mentions
+			}
+		});
+	}
+});

src/services/post/watch.ts

@@ -1,5 +1,5 @@
 import * as mongodb from 'mongodb';
-import Watching from '../models/post-watching';
+import Watching from '../../models/post-watching';
 
 export default async (me: mongodb.ObjectID, post: object) => {
 	// 自分の投稿はwatchできない