Talk federation (#5534)

6 files changed, 125 insertions, 74 deletions

diff --git a/src/client/app/common/views/components/messaging.vue b/src/client/app/common/views/components/messaging.vue
index b21104bf9a..52f55e4333 100644
--- a/src/client/app/common/views/components/messaging.vue
+++ b/src/client/app/common/views/components/messaging.vue
@@ -138,7 +138,7 @@ export default Vue.extend({
 			}
 			this.$root.api('users/search', {
 				query: this.q,
-				localOnly: true,
+				localOnly: false,
 				limit: 10,
 				detail: false
 			}).then(users => {
diff --git a/src/remote/activitypub/models/note.ts b/src/remote/activitypub/models/note.ts
index eb7cd6f81a..17c3721bdb 100644
--- a/src/remote/activitypub/models/note.ts
+++ b/src/remote/activitypub/models/note.ts
@@ -23,6 +23,7 @@ import { genId } from '../../../misc/gen-id';
 import { fetchMeta } from '../../../misc/fetch-meta';
 import { ensure } from '../../../prelude/ensure';
 import { getApLock } from '../../../misc/app-lock';
+import { createMessage } from '../../../services/messages/create';
 
 const logger = apLogger;
 
@@ -223,6 +224,13 @@ export async function createNote(value: string | IObject, resolver?: Resolver, s
 		if (actor.uri) updatePerson(actor.uri);
 	}
 
+	if (note._misskey_talk && visibility === 'specified') {
+		for (const recipient of visibleUsers) {
+			await createMessage(actor, recipient, undefined, text || undefined, (files && files.length > 0) ? files[0] : null);
+			return null;
+		}
+	}
+
 	return await post(actor, {
 		createdAt: note.published ? new Date(note.published) : null,
 		files,
diff --git a/src/remote/activitypub/renderer/note.ts b/src/remote/activitypub/renderer/note.ts
index ca823941c0..d58e1de433 100644
--- a/src/remote/activitypub/renderer/note.ts
+++ b/src/remote/activitypub/renderer/note.ts
@@ -12,7 +12,7 @@ import { Emoji } from '../../../models/entities/emoji';
 import { Poll } from '../../../models/entities/poll';
 import { ensure } from '../../../prelude/ensure';
 
-export default async function renderNote(note: Note, dive = true): Promise<any> {
+export default async function renderNote(note: Note, dive = true, isTalk = false): Promise<any> {
 	const promisedFiles: Promise<DriveFile[]> = note.fileIds.length > 0
 		? DriveFiles.find({ id: In(note.fileIds) })
 		: Promise.resolve([]);
@@ -145,6 +145,10 @@ export default async function renderNote(note: Note, dive = true): Promise<any>
 		}))
 	} : {};
 
+	const asTalk = isTalk ? {
+		_misskey_talk: true
+	} : {};
+
 	return {
 		id: `${config.url}/notes/${note.id}`,
 		type: 'Note',
@@ -160,7 +164,8 @@ export default async function renderNote(note: Note, dive = true): Promise<any>
 		attachment: files.map(renderDocument),
 		sensitive: note.cw != null || files.some(file => file.isSensitive),
 		tag,
-		...asPoll
+		...asPoll,
+		...asTalk
 	};
 }
 
diff --git a/src/remote/activitypub/type.ts b/src/remote/activitypub/type.ts
index df9da42fa8..5670df243d 100644
--- a/src/remote/activitypub/type.ts
+++ b/src/remote/activitypub/type.ts
@@ -75,6 +75,7 @@ export interface INote extends IObject {
 	type: 'Note' | 'Question' | 'Article' | 'Audio' | 'Document' | 'Image' | 'Page' | 'Video';
 	_misskey_content?: string;
 	_misskey_quote?: string;
+	_misskey_talk: boolean;
 }
 
 export interface IQuestion extends IObject {
diff --git a/src/server/api/endpoints/messaging/messages/create.ts b/src/server/api/endpoints/messaging/messages/create.ts
index 7ecbac9fe6..10b82c8f78 100644
--- a/src/server/api/endpoints/messaging/messages/create.ts
+++ b/src/server/api/endpoints/messaging/messages/create.ts
@@ -1,17 +1,12 @@
 import $ from 'cafy';
 import { ID } from '../../../../../misc/cafy-id';
-import { publishMainStream, publishGroupMessagingStream } from '../../../../../services/stream';
-import { publishMessagingStream, publishMessagingIndexStream } from '../../../../../services/stream';
-import pushSw from '../../../../../services/push-notification';
 import define from '../../../define';
 import { ApiError } from '../../../error';
 import { getUser } from '../../../common/getters';
-import { MessagingMessages, DriveFiles, Mutings, UserGroups, UserGroupJoinings } from '../../../../../models';
-import { MessagingMessage } from '../../../../../models/entities/messaging-message';
-import { genId } from '../../../../../misc/gen-id';
+import { MessagingMessages, DriveFiles, UserGroups, UserGroupJoinings } from '../../../../../models';
 import { User } from '../../../../../models/entities/user';
 import { UserGroup } from '../../../../../models/entities/user-group';
-import { Not } from 'typeorm';
+import { createMessage } from '../../../../../services/messages/create';
 
 export const meta = {
 	desc: {
@@ -147,68 +142,5 @@ export default define(meta, async (ps, user) => {
 		throw new ApiError(meta.errors.contentRequired);
 	}
 
-	const message = await MessagingMessages.save({
-		id: genId(),
-		createdAt: new Date(),
-		fileId: file ? file.id : null,
-		recipientId: recipientUser ? recipientUser.id : null,
-		groupId: recipientGroup ? recipientGroup.id : null,
-		text: ps.text ? ps.text.trim() : null,
-		userId: user.id,
-		isRead: false,
-		reads: [] as any[]
-	} as MessagingMessage);
-
-	const messageObj = await MessagingMessages.pack(message);
-
-	if (recipientUser) {
-		// 自分のストリーム
-		publishMessagingStream(message.userId, recipientUser.id, 'message', messageObj);
-		publishMessagingIndexStream(message.userId, 'message', messageObj);
-		publishMainStream(message.userId, 'messagingMessage', messageObj);
-
-		// 相手のストリーム
-		publishMessagingStream(recipientUser.id, message.userId, 'message', messageObj);
-		publishMessagingIndexStream(recipientUser.id, 'message', messageObj);
-		publishMainStream(recipientUser.id, 'messagingMessage', messageObj);
-	} else if (recipientGroup) {
-		// グループのストリーム
-		publishGroupMessagingStream(recipientGroup.id, 'message', messageObj);
-
-		// メンバーのストリーム
-		const joinings = await UserGroupJoinings.find({ userGroupId: recipientGroup.id });
-		for (const joining of joinings) {
-			publishMessagingIndexStream(joining.userId, 'message', messageObj);
-			publishMainStream(joining.userId, 'messagingMessage', messageObj);
-		}
-	}
-
-	// 2秒経っても(今回作成した)メッセージが既読にならなかったら「未読のメッセージがありますよ」イベントを発行する
-	setTimeout(async () => {
-		const freshMessage = await MessagingMessages.findOne(message.id);
-		if (freshMessage == null) return; // メッセージが削除されている場合もある
-
-		if (recipientUser) {
-			if (freshMessage.isRead) return; // 既読
-
-			//#region ただしミュートされているなら発行しない
-			const mute = await Mutings.find({
-				muterId: recipientUser.id,
-			});
-			if (mute.map(m => m.muteeId).includes(user.id)) return;
-			//#endregion
-
-			publishMainStream(recipientUser.id, 'unreadMessagingMessage', messageObj);
-			pushSw(recipientUser.id, 'unreadMessagingMessage', messageObj);
-		} else if (recipientGroup) {
-			const joinings = await UserGroupJoinings.find({ userGroupId: recipientGroup.id, userId: Not(user.id) });
-			for (const joining of joinings) {
-				if (freshMessage.reads.includes(joining.userId)) return; // 既読
-				publishMainStream(joining.userId, 'unreadMessagingMessage', messageObj);
-				pushSw(joining.userId, 'unreadMessagingMessage', messageObj);
-			}
-		}
-	}, 2000);
-
-	return messageObj;
+	return await createMessage(user, recipientUser, recipientGroup, ps.text, file);
 });
diff --git a/src/services/messages/create.ts b/src/services/messages/create.ts
new file mode 100644
index 0000000000..278070aa86
--- /dev/null
+++ b/src/services/messages/create.ts
@@ -0,0 +1,105 @@
+import { User } from '../../models/entities/user';
+import { UserGroup } from '../../models/entities/user-group';
+import { DriveFile } from '../../models/entities/drive-file';
+import { MessagingMessages, UserGroupJoinings, Mutings, Users } from '../../models';
+import { genId } from '../../misc/gen-id';
+import { MessagingMessage } from '../../models/entities/messaging-message';
+import { publishMessagingStream, publishMessagingIndexStream, publishMainStream, publishGroupMessagingStream } from '../stream';
+import pushNotification from '../push-notification';
+import { Not } from 'typeorm';
+import { Note } from '../../models/entities/note';
+import renderNote from '../../remote/activitypub/renderer/note';
+import renderCreate from '../../remote/activitypub/renderer/create';
+import { renderActivity } from '../../remote/activitypub/renderer';
+import { deliver } from '../../queue';
+
+export async function createMessage(user: User, recipientUser: User | undefined, recipientGroup: UserGroup | undefined, text: string | undefined, file: DriveFile | null) {
+	const message = await MessagingMessages.save({
+		id: genId(),
+		createdAt: new Date(),
+		fileId: file ? file.id : null,
+		recipientId: recipientUser ? recipientUser.id : null,
+		groupId: recipientGroup ? recipientGroup.id : null,
+		text: text ? text.trim() : null,
+		userId: user.id,
+		isRead: false,
+		reads: [] as any[]
+	} as MessagingMessage);
+
+	const messageObj = await MessagingMessages.pack(message);
+
+	if (recipientUser) {
+		if (Users.isLocalUser(user)) {
+			// 自分のストリーム
+			publishMessagingStream(message.userId, recipientUser.id, 'message', messageObj);
+			publishMessagingIndexStream(message.userId, 'message', messageObj);
+			publishMainStream(message.userId, 'messagingMessage', messageObj);
+		}
+
+		if (Users.isLocalUser(recipientUser)) {
+			// 相手のストリーム
+			publishMessagingStream(recipientUser.id, message.userId, 'message', messageObj);
+			publishMessagingIndexStream(recipientUser.id, 'message', messageObj);
+			publishMainStream(recipientUser.id, 'messagingMessage', messageObj);
+		}
+	} else if (recipientGroup) {
+		// グループのストリーム
+		publishGroupMessagingStream(recipientGroup.id, 'message', messageObj);
+
+		// メンバーのストリーム
+		const joinings = await UserGroupJoinings.find({ userGroupId: recipientGroup.id });
+		for (const joining of joinings) {
+			publishMessagingIndexStream(joining.userId, 'message', messageObj);
+			publishMainStream(joining.userId, 'messagingMessage', messageObj);
+		}
+	}
+
+	// 2秒経っても(今回作成した)メッセージが既読にならなかったら「未読のメッセージがありますよ」イベントを発行する
+	setTimeout(async () => {
+		const freshMessage = await MessagingMessages.findOne(message.id);
+		if (freshMessage == null) return; // メッセージが削除されている場合もある
+
+		if (recipientUser && Users.isLocalUser(recipientUser)) {
+			if (freshMessage.isRead) return; // 既読
+
+			//#region ただしミュートされているなら発行しない
+			const mute = await Mutings.find({
+				muterId: recipientUser.id,
+			});
+			if (mute.map(m => m.muteeId).includes(user.id)) return;
+			//#endregion
+
+			publishMainStream(recipientUser.id, 'unreadMessagingMessage', messageObj);
+			pushNotification(recipientUser.id, 'unreadMessagingMessage', messageObj);
+		} else if (recipientGroup) {
+			const joinings = await UserGroupJoinings.find({ userGroupId: recipientGroup.id, userId: Not(user.id) });
+			for (const joining of joinings) {
+				if (freshMessage.reads.includes(joining.userId)) return; // 既読
+				publishMainStream(joining.userId, 'unreadMessagingMessage', messageObj);
+				pushNotification(joining.userId, 'unreadMessagingMessage', messageObj);
+			}
+		}
+	}, 2000);
+
+	if (recipientUser && Users.isLocalUser(user) && Users.isRemoteUser(recipientUser)) {
+		const note = {
+			id: message.id,
+			createdAt: message.createdAt,
+			fileIds: message.fileId ? [ message.fileId ] : [],
+			text: message.text,
+			userId: message.userId,
+			visibility: 'specified',
+			mentions: [ recipientUser ].map(u => u.id),
+			mentionedRemoteUsers: JSON.stringify([ recipientUser ].map(u => ({
+				uri: u.uri,
+				username: u.username,
+				host: u.host
+			}))),
+		} as Note;
+
+		const activity = renderActivity(renderCreate(await renderNote(note, false, true), note));
+
+		deliver(user, activity, recipientUser.inbox);
+	}
+	return messageObj;
+}