From 245b08b624573cf4f6cb3192995d5d4f319cbce2 Mon Sep 17 00:00:00 2001 From: MeiMei <30769358+mei23@users.noreply.github.com> Date: Tue, 29 Oct 2019 06:01:14 +0900 Subject: [PATCH] Talk federation (#5534) --- .../app/common/views/components/messaging.vue | 2 +- src/remote/activitypub/models/note.ts | 8 ++ src/remote/activitypub/renderer/note.ts | 9 +- src/remote/activitypub/type.ts | 1 + .../endpoints/messaging/messages/create.ts | 74 +----------- src/services/messages/create.ts | 105 ++++++++++++++++++ 6 files changed, 125 insertions(+), 74 deletions(-) create mode 100644 src/services/messages/create.ts diff --git a/src/client/app/common/views/components/messaging.vue b/src/client/app/common/views/components/messaging.vue index b21104bf9..52f55e433 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 eb7cd6f81..17c3721bd 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 ca823941c..d58e1de43 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 { +export default async function renderNote(note: Note, dive = true, isTalk = false): Promise { const promisedFiles: Promise = 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 })) } : {}; + 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 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 df9da42fa..5670df243 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 7ecbac9fe..10b82c8f7 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 000000000..278070aa8 --- /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; +}