Merge pull request #1943 from syuilo/refactor-notes-create

Refactor notes create

src/misc/is-quote.ts

@@ -0,0 +1,5 @@
+import { INote } from '../models/note';
+
+export default function(note: INote): boolean {
+	return note.renoteId != null && (note.text != null || note.poll != null || (note.mediaIds != null && note.mediaIds.length > 0));
+}

src/services/note/create.ts

@@ -21,6 +21,9 @@ import resolveUser from '../../remote/resolve-user';
 import Meta from '../../models/meta';
 import config from '../../config';
 import registerHashtag from '../register-hashtag';
+import isQuote from '../../misc/is-quote';
+import { TextElementMention } from '../../mfm/parse/elements/mention';
+import { TextElementHashtag } from '../../mfm/parse/elements/hashtag';
 
 type Type = 'reply' | 'renote' | 'quote' | 'mention';
 
@@ -30,33 +33,56 @@ type Type = 'reply' | 'renote' | 'quote' | 'mention';
 class NotificationManager {
 	private notifier: IUser;
 	private note: INote;
+	private queue: Array<{
+		target: ILocalUser['_id'];
+		reason: Type;
+	}>;
 
 	constructor(notifier: IUser, note: INote) {
 		this.notifier = notifier;
 		this.note = note;
+		this.queue = [];
 	}
 
-	public async push(notifiee: ILocalUser['_id'], type: Type) {
+	public push(notifiee: ILocalUser['_id'], reason: Type) {
 		// 自分自身へは通知しない
 		if (this.notifier._id.equals(notifiee)) return;
 
-		// ミュート情報を取得
-		const mentioneeMutes = await Mute.find({
-			muterId: notifiee
-		});
+		const exist = this.queue.find(x => x.target.equals(notifiee));
 
-		const mentioneesMutedUserIds = mentioneeMutes.map(m => m.muteeId.toString());
-
-		// 通知される側のユーザーが通知する側のユーザーをミュートしていない限りは通知する
-		if (!mentioneesMutedUserIds.includes(this.notifier._id.toString())) {
-			notify(notifiee, this.notifier._id, type, {
-				noteId: this.note._id
+		if (exist) {
+			// 「メンションされているかつ返信されている」場合は、メンションとしての通知ではなく返信としての通知にする
+			if (reason != 'mention') {
+				exist.reason = reason;
+			}
+		} else {
+			this.queue.push({
+				reason: reason,
+				target: notifiee
 			});
 		}
 	}
+
+	public deliver() {
+		this.queue.forEach(async x => {
+			// ミュート情報を取得
+			const mentioneeMutes = await Mute.find({
+				muterId: x.target
+			});
+
+			const mentioneesMutedUserIds = mentioneeMutes.map(m => m.muteeId.toString());
+
+			// 通知される側のユーザーが通知する側のユーザーをミュートしていない限りは通知する
+			if (!mentioneesMutedUserIds.includes(this.notifier._id.toString())) {
+				notify(x.target, this.notifier._id, x.reason, {
+					noteId: this.note._id
+				});
+			}
+		});
+	}
 }
 
-export default async (user: IUser, data: {
+type Option = {
 	createdAt?: Date;
 	text?: string;
 	reply?: INote;
@@ -70,37 +96,192 @@ export default async (user: IUser, data: {
 	visibleUsers?: IUser[];
 	uri?: string;
 	app?: IApp;
-}, silent = false) => new Promise<INote>(async (res, rej) => {
+};
+
+export default async (user: IUser, data: Option, silent = false) => new Promise<INote>(async (res, rej) => {
 	if (data.createdAt == null) data.createdAt = new Date();
 	if (data.visibility == null) data.visibility = 'public';
 	if (data.viaMobile == null) data.viaMobile = false;
 
-	let tags: string[] = [];
-
-	let tokens: any[] = 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);
-			}
-		});
-	}
-
-	tags = tags.filter(tag => tag.length <= 100);
-
 	if (data.visibleUsers) {
 		data.visibleUsers = data.visibleUsers.filter(x => x != null);
 	}
 
+	// Parse MFM
+	const tokens = data.text ? parse(data.text) : [];
+
+	const tags = extractHashtags(tokens);
+
+	const mentionedUsers = await extractMentionedUsers(tokens);
+
+	const note = await insertNote(user, data, tokens, tags, mentionedUsers);
+
+	res(note);
+
+	if (note == null) {
+		return;
+	}
+
+	// ハッシュタグ登録
+	tags.map(tag => registerHashtag(user, tag));
+
+	// Increment notes count
+	incNotesCount(user);
+
+	// Increment notes count (user)
+	incNotesCountOfUser(user);
+
+	if (data.reply) {
+		saveReply(data.reply, note);
+	}
+
+	if (data.renote) {
+		incRenoteCount(data.renote);
+	}
+
+	if (isQuote(note)) {
+		saveQuote(data.renote, note);
+	}
+
+	// Pack the note
+	const noteObj = await pack(note);
+
+	const nm = new NotificationManager(user, note);
+	const nmRelatedPromises = [];
+
+	createMentionedEvents(mentionedUsers, noteObj, nm);
+
+	const noteActivity = await renderActivity(data, note);
+
+	if (isLocalUser(user)) {
+		deliverNoteToMentionedRemoteUsers(mentionedUsers, user, noteActivity);
+	}
+
+	// If has in reply to note
+	if (data.reply) {
+		// Fetch watchers
+		nmRelatedPromises.push(notifyToWatchersOfReplyee(data.reply, user, nm));
+
+		// この投稿をWatchする
+		if (isLocalUser(user) && user.settings.autoWatch !== false) {
+			watch(user._id, data.reply);
+		}
+
+		// 通知
+		nm.push(data.reply.userId, 'reply');
+	}
+
+	// If it is renote
+	if (data.renote) {
+		// Notify
+		const type = data.text ? 'quote' : 'renote';
+		nm.push(data.renote.userId, type);
+
+		// Fetch watchers
+		nmRelatedPromises.push(notifyToWatchersOfRenotee(data.renote, user, nm, type));
+
+		// この投稿をWatchする
+		if (isLocalUser(user) && user.settings.autoWatch !== false) {
+			watch(user._id, data.renote);
+		}
+
+		// If it is quote renote
+		if (data.text) {
+			// Add mention
+			nm.push(data.renote.userId, 'quote');
+		} else {
+			// Publish event
+			if (!user._id.equals(data.renote.userId)) {
+				event(data.renote.userId, 'renote', noteObj);
+			}
+		}
+	}
+
+	if (!silent) {
+		publish(user, note, noteObj, data.reply, data.renote, data.visibleUsers, noteActivity);
+	}
+
+	Promise.all(nmRelatedPromises).then(() => {
+		nm.deliver();
+	});
+
+	// Register to search database
+	index(note);
+});
+
+async function renderActivity(data: Option, note: INote) {
+	const content = data.renote && data.text == null
+		? renderAnnounce(data.renote.uri ? data.renote.uri : await renderNote(data.renote))
+		: renderCreate(await renderNote(note));
+
+	return packAp(content);
+}
+
+function incRenoteCount(renote: INote) {
+	Note.update({ _id: renote._id }, {
+		$inc: {
+			renoteCount: 1
+		}
+	});
+}
+
+async function publish(user: IUser, note: INote, noteObj: any, reply: INote, renote: INote, visibleUsers: IUser[], noteActivity: any) {
+	if (isLocalUser(user)) {
+		// 投稿がリプライかつ投稿者がローカルユーザーかつリプライ先の投稿の投稿者がリモートユーザーなら配送
+		if (reply && isRemoteUser(reply._user)) {
+			deliver(user, noteActivity, reply._user.inbox);
+		}
+
+		// 投稿がRenoteかつ投稿者がローカルユーザーかつRenote元の投稿の投稿者がリモートユーザーなら配送
+		if (renote && isRemoteUser(renote._user)) {
+			deliver(user, noteActivity, renote._user.inbox);
+		}
+
+		if (['private', 'followers', 'specified'].includes(note.visibility)) {
+			// Publish event to myself's stream
+			stream(note.userId, 'note', await pack(note, user, {
+				detail: true
+			}));
+		} else {
+			// Publish event to myself's stream
+			stream(note.userId, 'note', noteObj);
+
+			// Publish note to local and hybrid timeline stream
+			if (note.visibility != 'home') {
+				publishLocalTimelineStream(noteObj);
+			}
+
+			if (note.visibility == 'public') {
+				publishHybridTimelineStream(null, noteObj);
+			}
+		}
+	}
+
+	// Publish note to global timeline stream
+	if (note.visibility == 'public' && note.replyId == null) {
+		publishGlobalTimelineStream(noteObj);
+	}
+
+	if (note.visibility == 'specified') {
+		visibleUsers.forEach(async (u) => {
+			const n = await pack(note, u, {
+				detail: true
+			});
+			stream(u._id, 'note', n);
+			publishHybridTimelineStream(u._id, n);
+		});
+	}
+
+	if (['public', 'home', 'followers'].includes(note.visibility)) {
+		// フォロワーに配信
+		publishToFollowers(note, noteObj, user, noteActivity);
+	}
+
+	// リストに配信
+	publishToUserLists(note, noteObj);
+}
+
+async function insertNote(user: IUser, data: Option, tokens: ReturnType<typeof parse>, tags: string[], mentionedUsers: IUser[]) {
 	const insert: any = {
 		createdAt: data.createdAt,
 		mediaIds: data.media ? data.media.map(file => file._id) : [],
@@ -133,26 +314,168 @@ export default async (user: IUser, data: {
 
 	if (data.uri != null) insert.uri = data.uri;
 
+	// Append mentions data
+	if (mentionedUsers.length > 0) {
+		insert.mentions = mentionedUsers.map(u => u._id);
+		insert.mentionedRemoteUsers = mentionedUsers.filter(u => isRemoteUser(u)).map(u => ({
+			uri: (u as IRemoteUser).uri,
+			username: u.username,
+			host: u.host
+		}));
+	}
+
 	// 投稿を作成
-	let note: INote;
 	try {
-		note = await Note.insert(insert);
+		return await Note.insert(insert);
 	} catch (e) {
 		// duplicate key error
 		if (e.code === 11000) {
-			return res(null);
+			return null;
 		}
 
 		console.error(e);
-		return rej('something happened');
+		throw 'something happened';
 	}
+}
 
-	res(note);
+function extractHashtags(tokens: ReturnType<typeof parse>): string[] {
+	// Extract hashtags
+	const hashtags = tokens
+		.filter(t => t.type == 'hashtag')
+		.map(t => (t as TextElementHashtag).hashtag)
+		.filter(tag => tag.length <= 100);
 
-	// ハッシュタグ登録
-	tags.map(tag => registerHashtag(user, tag));
+	return [...new Set(hashtags)];
+}
 
-	//#region Increment notes count
+function index(note: INote) {
+	if (note.text == null || config.elasticsearch == null) return;
+
+	es.index({
+		index: 'misskey',
+		type: 'note',
+		id: note._id.toString(),
+		body: {
+			text: note.text
+		}
+	});
+}
+
+async function notifyToWatchersOfRenotee(renote: INote, user: IUser, nm: NotificationManager, type: Type) {
+	const watchers = await NoteWatching.find({
+		noteId: renote._id,
+		userId: { $ne: user._id }
+	}, {
+		fields: {
+			userId: true
+		}
+	});
+
+	watchers.forEach(watcher => {
+		nm.push(watcher.userId, type);
+	});
+}
+
+async function notifyToWatchersOfReplyee(reply: INote, user: IUser, nm: NotificationManager) {
+	const watchers = await NoteWatching.find({
+		noteId: reply._id,
+		userId: { $ne: user._id }
+	}, {
+		fields: {
+			userId: true
+		}
+	});
+
+	watchers.forEach(watcher => {
+		nm.push(watcher.userId, 'reply');
+	});
+}
+
+async function publishToUserLists(note: INote, noteObj: any) {
+	const lists = await UserList.find({
+		userIds: note.userId
+	});
+
+	lists.forEach(list => {
+		publishUserListStream(list._id, 'note', noteObj);
+	});
+}
+
+async function publishToFollowers(note: INote, noteObj: any, user: IUser, noteActivity: any) {
+	const followers = await Following.find({
+		followeeId: note.userId
+	});
+
+	followers.map(async (following) => {
+		const follower = following._follower;
+
+		if (isLocalUser(follower)) {
+			// ストーキングしていない場合
+			if (!following.stalk) {
+				// この投稿が返信ならスキップ
+				if (note.replyId && !note._reply.userId.equals(following.followerId) && !note._reply.userId.equals(note.userId))
+					return;
+			}
+
+			// Publish event to followers stream
+			stream(following.followerId, 'note', noteObj);
+
+			if (isRemoteUser(user) || note.visibility != 'public') {
+				publishHybridTimelineStream(following.followerId, noteObj);
+			}
+		} else {
+			// フォロワーがリモートユーザーかつ投稿者がローカルユーザーなら投稿を配信
+			if (isLocalUser(user)) {
+				deliver(user, noteActivity, follower.inbox);
+			}
+		}
+	});
+}
+
+function deliverNoteToMentionedRemoteUsers(mentionedUsers: IUser[], user: ILocalUser, noteActivity: any) {
+	mentionedUsers.filter(u => isRemoteUser(u)).forEach(async (u) => {
+		deliver(user, noteActivity, (u as IRemoteUser).inbox);
+	});
+}
+
+function createMentionedEvents(mentionedUsers: IUser[], noteObj: any, nm: NotificationManager) {
+	mentionedUsers.filter(u => isLocalUser(u)).forEach(async (u) => {
+		event(u, 'mention', noteObj);
+
+		// Create notification
+		nm.push(u._id, 'mention');
+	});
+}
+
+function saveQuote(renote: INote, note: INote) {
+	Note.update({ _id: renote._id }, {
+		$push: {
+			_quoteIds: note._id
+		},
+
+	});
+}
+
+function saveReply(reply: INote, note: INote) {
+	Note.update({ _id: reply._id }, {
+		$push: {
+			_replyIds: note._id
+		},
+		$inc: {
+			repliesCount: 1
+		}
+	});
+}
+
+function incNotesCountOfUser(user: IUser) {
+	User.update({ _id: user._id }, {
+		$inc: {
+			notesCount: 1
+		}
+	});
+}
+
+function incNotesCount(user: IUser) {
 	if (isLocalUser(user)) {
 		Meta.update({}, {
 			$inc: {
@@ -167,290 +490,23 @@ export default async (user: IUser, data: {
 			}
 		}, { upsert: true });
 	}
-	//#endregion
+}
 
-	// Increment notes count (user)
-	User.update({ _id: user._id }, {
-		$inc: {
-			notesCount: 1
+async function extractMentionedUsers(tokens: ReturnType<typeof parse>): Promise<IUser[]> {
+	if (tokens == null) return [];
+
+	// TODO: Drop dupulicates
+	const mentionTokens = tokens
+		.filter(t => t.type == 'mention') as TextElementMention[];
+
+	// TODO: Drop dupulicates
+	const mentionedUsers = (await Promise.all(mentionTokens.map(async m => {
+		try {
+			return await resolveUser(m.username, m.host);
+		} catch (e) {
+			return null;
 		}
-	});
+	}))).filter(x => x != null);
 
-	if (data.reply) {
-		Note.update({ _id: data.reply._id }, {
-			$push: {
-				_replyIds: note._id
-			}
-		});
-	}
-
-	const isQuote = data.renote && (data.text || data.poll || data.media);
-
-	if (isQuote) {
-		Note.update({ _id: data.renote._id }, {
-			$push: {
-				_quoteIds: note._id
-			}
-		});
-	}
-
-	// Serialize
-	const noteObj = await pack(note);
-
-	const nm = new NotificationManager(user, note);
-
-	const render = async () => {
-		const content = data.renote && data.text == null
-			? renderAnnounce(data.renote.uri ? data.renote.uri : await renderNote(data.renote))
-			: renderCreate(await renderNote(note));
-		return packAp(content);
-	};
-
-	//#region メンション
-	if (data.text) {
-		// TODO: Drop dupulicates
-		const mentionTokens = tokens
-			.filter(t => t.type == 'mention');
-
-		// TODO: Drop dupulicates
-		const mentionedUsers = (await Promise.all(mentionTokens.map(async m => {
-			try {
-				return await resolveUser(m.username, m.host);
-			} catch (e) {
-				return null;
-			}
-		}))).filter(x => x != null);
-
-		// Append mentions data
-		if (mentionedUsers.length > 0) {
-			const set = {
-				mentions: mentionedUsers.map(u => u._id),
-				mentionedRemoteUsers: mentionedUsers.filter(u => isRemoteUser(u)).map(u => ({
-					uri: (u as IRemoteUser).uri,
-					username: u.username,
-					host: u.host
-				}))
-			};
-
-			Note.update({ _id: note._id }, {
-				$set: set
-			});
-
-			Object.assign(note, set);
-		}
-
-		mentionedUsers.filter(u => isLocalUser(u)).forEach(async u => {
-			event(u, 'mention', noteObj);
-
-			// 既に言及されたユーザーに対する返信や引用renoteの場合も無視
-			if (data.reply && data.reply.userId.equals(u._id)) return;
-			if (data.renote && data.renote.userId.equals(u._id)) return;
-
-			// Create notification
-			nm.push(u._id, 'mention');
-		});
-
-		if (isLocalUser(user)) {
-			mentionedUsers.filter(u => isRemoteUser(u)).forEach(async u => {
-				deliver(user, await render(), (u as IRemoteUser).inbox);
-			});
-		}
-	}
-	//#endregion
-
-	if (!silent) {
-		if (isLocalUser(user)) {
-			if (note.visibility == 'private' || note.visibility == 'followers' || note.visibility == 'specified') {
-				// Publish event to myself's stream
-				stream(note.userId, 'note', await pack(note, user, {
-					detail: true
-				}));
-			} else {
-				// Publish event to myself's stream
-				stream(note.userId, 'note', noteObj);
-
-				// Publish note to local and hybrid timeline stream
-				if (note.visibility != 'home') {
-					publishLocalTimelineStream(noteObj);
-				}
-				if (note.visibility == 'public') {
-					publishHybridTimelineStream(null, noteObj);
-				}
-			}
-		}
-
-		// Publish note to global timeline stream
-		if (note.visibility == 'public' && note.replyId == null) {
-			publishGlobalTimelineStream(noteObj);
-		}
-
-		if (note.visibility == 'specified') {
-			data.visibleUsers.forEach(async u => {
-				const n = await pack(note, u, {
-					detail: true
-				});
-				stream(u._id, 'note', n);
-				publishHybridTimelineStream(u._id, n);
-			});
-		}
-
-		if (note.visibility == 'public' || note.visibility == 'home' || note.visibility == 'followers') {
-			// フォロワーに配信
-			Following.find({
-				followeeId: note.userId
-			}).then(followers => {
-				followers.map(async following => {
-					const follower = following._follower;
-
-					if (isLocalUser(follower)) {
-						// ストーキングしていない場合
-						if (!following.stalk) {
-							// この投稿が返信ならスキップ
-							if (note.replyId && !note._reply.userId.equals(following.followerId) && !note._reply.userId.equals(note.userId)) return;
-						}
-
-						// Publish event to followers stream
-						stream(following.followerId, 'note', noteObj);
-
-						if (isRemoteUser(user) || note.visibility != 'public') {
-							publishHybridTimelineStream(following.followerId, noteObj);
-						}
-					} else {
-						//#region AP配送
-						// フォロワーがリモートユーザーかつ投稿者がローカルユーザーなら投稿を配信
-						if (isLocalUser(user)) {
-							deliver(user, await render(), follower.inbox);
-						}
-						//#endergion
-					}
-				});
-			});
-		}
-
-		// リストに配信
-		UserList.find({
-			userIds: note.userId
-		}).then(lists => {
-			lists.forEach(list => {
-				publishUserListStream(list._id, 'note', noteObj);
-			});
-		});
-	}
-
-	//#region リプライとAnnounceのAP配送
-
-	// 投稿がリプライかつ投稿者がローカルユーザーかつリプライ先の投稿の投稿者がリモートユーザーなら配送
-	if (data.reply && isLocalUser(user) && isRemoteUser(data.reply._user)) {
-		deliver(user, await render(), data.reply._user.inbox);
-	}
-
-	// 投稿がRenoteかつ投稿者がローカルユーザーかつRenote元の投稿の投稿者がリモートユーザーなら配送
-	if (data.renote && isLocalUser(user) && isRemoteUser(data.renote._user)) {
-		deliver(user, await render(), data.renote._user.inbox);
-	}
-	//#endergion
-
-	// If has in reply to note
-	if (data.reply) {
-		// Increment replies count
-		Note.update({ _id: data.reply._id }, {
-			$inc: {
-				repliesCount: 1
-			}
-		});
-
-		// Fetch watchers
-		NoteWatching.find({
-			noteId: data.reply._id,
-			userId: { $ne: user._id },
-			// 削除されたドキュメントは除く
-			deletedAt: { $exists: false }
-		}, {
-				fields: {
-					userId: true
-				}
-			}).then(watchers => {
-				watchers.forEach(watcher => {
-					nm.push(watcher.userId, 'reply');
-				});
-			});
-
-		// この投稿をWatchする
-		if (isLocalUser(user) && user.settings.autoWatch !== false) {
-			watch(user._id, data.reply);
-		}
-
-		// 通知
-		nm.push(data.reply.userId, 'reply');
-	}
-
-	// If it is renote
-	if (data.renote) {
-		// Notify
-		const type = data.text ? 'quote' : 'renote';
-		nm.push(data.renote.userId, type);
-
-		// Fetch watchers
-		NoteWatching.find({
-			noteId: data.renote._id,
-			userId: { $ne: user._id }
-		}, {
-				fields: {
-					userId: true
-				}
-			}).then(watchers => {
-				watchers.forEach(watcher => {
-					nm.push(watcher.userId, type);
-				});
-			});
-
-		// この投稿をWatchする
-		if (isLocalUser(user) && user.settings.autoWatch !== false) {
-			watch(user._id, data.renote);
-		}
-
-		// If it is quote renote
-		if (data.text) {
-			// Add mention
-			nm.push(data.renote.userId, 'quote');
-		} else {
-			// Publish event
-			if (!user._id.equals(data.renote.userId)) {
-				event(data.renote.userId, 'renote', noteObj);
-			}
-		}
-
-		//#region TODO: これ重い
-		// 今までで同じ投稿をRenoteしているか
-		//const existRenote = await Note.findOne({
-		//	userId: user._id,
-		//	renoteId: data.renote._id,
-		//	_id: {
-		//		$ne: note._id
-		//	}
-		//});
-		const existRenote: INote | null = null;
-		//#endregion
-
-		if (!existRenote) {
-			// Update renoteee status
-			Note.update({ _id: data.renote._id }, {
-				$inc: {
-					renoteCount: 1
-				}
-			});
-		}
-	}
-
-	// Register to search database
-	if (note.text && config.elasticsearch) {
-		es.index({
-			index: 'misskey',
-			type: 'note',
-			id: note._id.toString(),
-			body: {
-				text: note.text
-			}
-		});
-	}
-});
+	return mentionedUsers;
+}