無駄なAP deliverをしないように (#5589)
* DeliverManager, note/create * recipe * followers delivers * comment * rename * fix * cleanup
This commit is contained in:
parent
b34b728fbb
commit
44ab428803
|
@ -0,0 +1,131 @@
|
|||
import { Users, Followings } from '../../models';
|
||||
import { ILocalUser, IRemoteUser } from '../../models/entities/user';
|
||||
import { deliver } from '../../queue';
|
||||
|
||||
//#region types
|
||||
interface IRecipe {
|
||||
type: string;
|
||||
}
|
||||
|
||||
interface IFollowersRecipe extends IRecipe {
|
||||
type: 'Followers';
|
||||
}
|
||||
|
||||
interface IDirectRecipe extends IRecipe {
|
||||
type: 'Direct';
|
||||
to: IRemoteUser;
|
||||
}
|
||||
|
||||
const isFollowers = (recipe: any): recipe is IFollowersRecipe =>
|
||||
recipe.type === 'Followers';
|
||||
|
||||
const isDirect = (recipe: any): recipe is IDirectRecipe =>
|
||||
recipe.type === 'Direct';
|
||||
//#endregion
|
||||
|
||||
export default class DeliverManager {
|
||||
private actor: ILocalUser;
|
||||
private activity: any;
|
||||
private recipes: IRecipe[] = [];
|
||||
|
||||
/**
|
||||
* Constructor
|
||||
* @param actor Actor
|
||||
* @param activity Activity to deliver
|
||||
*/
|
||||
constructor(actor: ILocalUser, activity: any) {
|
||||
this.actor = actor;
|
||||
this.activity = activity;
|
||||
}
|
||||
|
||||
/**
|
||||
* Add recipe for followers deliver
|
||||
*/
|
||||
public addFollowersRecipe() {
|
||||
const deliver = {
|
||||
type: 'Followers'
|
||||
} as IFollowersRecipe;
|
||||
|
||||
this.addRecipe(deliver);
|
||||
}
|
||||
|
||||
/**
|
||||
* Add recipe for direct deliver
|
||||
* @param to To
|
||||
*/
|
||||
public addDirectRecipe(to: IRemoteUser) {
|
||||
const recipe = {
|
||||
type: 'Direct',
|
||||
to
|
||||
} as IDirectRecipe;
|
||||
|
||||
this.addRecipe(recipe);
|
||||
}
|
||||
|
||||
/**
|
||||
* Add recipe
|
||||
* @param recipe Recipe
|
||||
*/
|
||||
public addRecipe(recipe: IRecipe) {
|
||||
this.recipes.push(recipe);
|
||||
}
|
||||
|
||||
/**
|
||||
* Execute delivers
|
||||
*/
|
||||
public async execute() {
|
||||
if (!Users.isLocalUser(this.actor)) return;
|
||||
|
||||
const inboxes: string[] = [];
|
||||
|
||||
// build inbox list
|
||||
for (const recipe of this.recipes) {
|
||||
if (isFollowers(recipe)) {
|
||||
// followers deliver
|
||||
const followers = await Followings.find({
|
||||
followeeId: this.actor.id
|
||||
});
|
||||
|
||||
for (const following of followers) {
|
||||
if (Followings.isRemoteFollower(following)) {
|
||||
const inbox = following.followerSharedInbox || following.followerInbox;
|
||||
if (!inboxes.includes(inbox)) inboxes.push(inbox);
|
||||
}
|
||||
}
|
||||
} else if (isDirect(recipe)) {
|
||||
// direct deliver
|
||||
const inbox = recipe.to.inbox;
|
||||
if (inbox && !inboxes.includes(inbox)) inboxes.push(inbox);
|
||||
}
|
||||
}
|
||||
|
||||
// deliver
|
||||
for (const inbox of inboxes) {
|
||||
deliver(this.actor, this.activity, inbox);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
//#region Utilities
|
||||
/**
|
||||
* Deliver activity to followers
|
||||
* @param activity Activity
|
||||
* @param from Followee
|
||||
*/
|
||||
export async function deliverToFollowers(actor: ILocalUser, activity: any) {
|
||||
const manager = new DeliverManager(actor, activity);
|
||||
manager.addFollowersRecipe();
|
||||
await manager.execute();
|
||||
}
|
||||
|
||||
/**
|
||||
* Deliver activity to user
|
||||
* @param activity Activity
|
||||
* @param to Target user
|
||||
*/
|
||||
export async function deliverToUser(actor: ILocalUser, activity: any, to: IRemoteUser) {
|
||||
const manager = new DeliverManager(actor, activity);
|
||||
manager.addDirectRecipe(to);
|
||||
await manager.execute();
|
||||
}
|
||||
//#endregion
|
|
@ -2,13 +2,13 @@ import config from '../../config';
|
|||
import renderAdd from '../../remote/activitypub/renderer/add';
|
||||
import renderRemove from '../../remote/activitypub/renderer/remove';
|
||||
import { renderActivity } from '../../remote/activitypub/renderer';
|
||||
import { deliver } from '../../queue';
|
||||
import { IdentifiableError } from '../../misc/identifiable-error';
|
||||
import { User, ILocalUser } from '../../models/entities/user';
|
||||
import { User } from '../../models/entities/user';
|
||||
import { Note } from '../../models/entities/note';
|
||||
import { Notes, UserNotePinings, Users, Followings } from '../../models';
|
||||
import { Notes, UserNotePinings, Users } from '../../models';
|
||||
import { UserNotePining } from '../../models/entities/user-note-pinings';
|
||||
import { genId } from '../../misc/gen-id';
|
||||
import { deliverToFollowers } from '../../remote/activitypub/deliver-manager';
|
||||
|
||||
/**
|
||||
* 指定した投稿をピン留めします
|
||||
|
@ -82,36 +82,9 @@ export async function deliverPinnedChange(userId: User['id'], noteId: Note['id']
|
|||
|
||||
if (!Users.isLocalUser(user)) return;
|
||||
|
||||
const queue = await CreateRemoteInboxes(user);
|
||||
|
||||
if (queue.length < 1) return;
|
||||
|
||||
const target = `${config.url}/users/${user.id}/collections/featured`;
|
||||
|
||||
const item = `${config.url}/notes/${noteId}`;
|
||||
const content = renderActivity(isAddition ? renderAdd(user, target, item) : renderRemove(user, target, item));
|
||||
for (const inbox of queue) {
|
||||
deliver(user, content, inbox);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* ローカルユーザーのリモートフォロワーのinboxリストを作成する
|
||||
* @param user ローカルユーザー
|
||||
*/
|
||||
async function CreateRemoteInboxes(user: ILocalUser): Promise<string[]> {
|
||||
const followers = await Followings.find({
|
||||
followeeId: user.id
|
||||
});
|
||||
|
||||
const queue: string[] = [];
|
||||
|
||||
for (const following of followers) {
|
||||
if (Followings.isRemoteFollower(following)) {
|
||||
const inbox = following.followerSharedInbox || following.followerInbox;
|
||||
if (!queue.includes(inbox)) queue.push(inbox);
|
||||
}
|
||||
}
|
||||
|
||||
return queue;
|
||||
|
||||
deliverToFollowers(user, content);
|
||||
}
|
||||
|
|
|
@ -1,34 +1,17 @@
|
|||
import renderUpdate from '../../remote/activitypub/renderer/update';
|
||||
import { renderActivity } from '../../remote/activitypub/renderer';
|
||||
import { deliver } from '../../queue';
|
||||
import { Followings, Users } from '../../models';
|
||||
import { Users } from '../../models';
|
||||
import { User } from '../../models/entities/user';
|
||||
import { renderPerson } from '../../remote/activitypub/renderer/person';
|
||||
import { deliverToFollowers } from '../../remote/activitypub/deliver-manager';
|
||||
|
||||
export async function publishToFollowers(userId: User['id']) {
|
||||
const user = await Users.findOne(userId);
|
||||
if (user == null) throw new Error('user not found');
|
||||
|
||||
const followers = await Followings.find({
|
||||
followeeId: user.id
|
||||
});
|
||||
|
||||
const queue: string[] = [];
|
||||
|
||||
// フォロワーがリモートユーザーかつ投稿者がローカルユーザーならUpdateを配信
|
||||
if (Users.isLocalUser(user)) {
|
||||
for (const following of followers) {
|
||||
if (Followings.isRemoteFollower(following)) {
|
||||
const inbox = following.followerSharedInbox || following.followerInbox;
|
||||
if (!queue.includes(inbox)) queue.push(inbox);
|
||||
}
|
||||
}
|
||||
|
||||
if (queue.length > 0) {
|
||||
const content = renderActivity(renderUpdate(await renderPerson(user), user));
|
||||
for (const inbox of queue) {
|
||||
deliver(user, content, inbox);
|
||||
}
|
||||
}
|
||||
deliverToFollowers(user, content);
|
||||
}
|
||||
}
|
||||
|
|
|
@ -1,6 +1,6 @@
|
|||
import es from '../../db/elasticsearch';
|
||||
import { publishMainStream, publishNotesStream } from '../stream';
|
||||
import { deliver } from '../../queue';
|
||||
import DeliverManager from '../../remote/activitypub/deliver-manager';
|
||||
import renderNote from '../../remote/activitypub/renderer/note';
|
||||
import renderCreate from '../../remote/activitypub/renderer/create';
|
||||
import renderAnnounce from '../../remote/activitypub/renderer/announce';
|
||||
|
@ -17,7 +17,7 @@ import extractMentions from '../../misc/extract-mentions';
|
|||
import extractEmojis from '../../misc/extract-emojis';
|
||||
import extractHashtags from '../../misc/extract-hashtags';
|
||||
import { Note, IMentionedRemoteUsers } from '../../models/entities/note';
|
||||
import { Mutings, Users, NoteWatchings, Followings, Notes, Instances, UserProfiles } from '../../models';
|
||||
import { Mutings, Users, NoteWatchings, Notes, Instances, UserProfiles } from '../../models';
|
||||
import { DriveFile } from '../../models/entities/drive-file';
|
||||
import { App } from '../../models/entities/app';
|
||||
import { Not, getConnection, In } from 'typeorm';
|
||||
|
@ -246,12 +246,6 @@ export default async (user: User, data: Option, silent = false) => new Promise<N
|
|||
|
||||
createMentionedEvents(mentionedUsers, note, nm);
|
||||
|
||||
const noteActivity = await renderNoteOrRenoteActivity(data, note);
|
||||
|
||||
if (Users.isLocalUser(user)) {
|
||||
deliverNoteToMentionedRemoteUsers(mentionedUsers, user, noteActivity);
|
||||
}
|
||||
|
||||
const profile = await UserProfiles.findOne(user.id).then(ensure);
|
||||
|
||||
// If has in reply to note
|
||||
|
@ -294,11 +288,42 @@ export default async (user: User, data: Option, silent = false) => new Promise<N
|
|||
}
|
||||
}
|
||||
|
||||
publish(user, note, data.reply, data.renote, noteActivity);
|
||||
|
||||
Promise.all(nmRelatedPromises).then(() => {
|
||||
nm.deliver();
|
||||
});
|
||||
|
||||
//#region AP deliver
|
||||
if (Users.isLocalUser(user)) {
|
||||
(async () => {
|
||||
const noteActivity = await renderNoteOrRenoteActivity(data, note);
|
||||
const dm = new DeliverManager(user, noteActivity);
|
||||
|
||||
// メンションされたリモートユーザーに配送
|
||||
for (const u of mentionedUsers.filter(u => Users.isRemoteUser(u))) {
|
||||
dm.addDirectRecipe(u as IRemoteUser);
|
||||
}
|
||||
|
||||
// 投稿がリプライかつ投稿者がローカルユーザーかつリプライ先の投稿の投稿者がリモートユーザーなら配送
|
||||
if (data.reply && data.reply.userHost !== null) {
|
||||
const u = await Users.findOne(data.reply.userId);
|
||||
if (u && Users.isRemoteUser(u)) dm.addDirectRecipe(u);
|
||||
}
|
||||
|
||||
// 投稿がRenoteかつ投稿者がローカルユーザーかつRenote元の投稿の投稿者がリモートユーザーなら配送
|
||||
if (data.renote && data.renote.userHost !== null) {
|
||||
const u = await Users.findOne(data.renote.userId);
|
||||
if (u && Users.isRemoteUser(u)) dm.addDirectRecipe(u);
|
||||
}
|
||||
|
||||
// フォロワーに配送
|
||||
if (['public', 'home', 'followers'].includes(note.visibility)) {
|
||||
dm.addFollowersRecipe();
|
||||
}
|
||||
|
||||
dm.execute();
|
||||
})();
|
||||
}
|
||||
//#endregion
|
||||
}
|
||||
|
||||
// Register to search database
|
||||
|
@ -320,29 +345,6 @@ function incRenoteCount(renote: Note) {
|
|||
Notes.increment({ id: renote.id }, 'score', 1);
|
||||
}
|
||||
|
||||
async function publish(user: User, note: Note, reply: Note | null | undefined, renote: Note | null | undefined, noteActivity: any) {
|
||||
if (Users.isLocalUser(user)) {
|
||||
// 投稿がリプライかつ投稿者がローカルユーザーかつリプライ先の投稿の投稿者がリモートユーザーなら配送
|
||||
if (reply && reply.userHost !== null) {
|
||||
Users.findOne(reply.userId).then(ensure).then(u => {
|
||||
deliver(user, noteActivity, u.inbox);
|
||||
});
|
||||
}
|
||||
|
||||
// 投稿がRenoteかつ投稿者がローカルユーザーかつRenote元の投稿の投稿者がリモートユーザーなら配送
|
||||
if (renote && renote.userHost !== null) {
|
||||
Users.findOne(renote.userId).then(ensure).then(u => {
|
||||
deliver(user, noteActivity, u.inbox);
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
if (['public', 'home', 'followers'].includes(note.visibility)) {
|
||||
// フォロワーに配信
|
||||
publishToFollowers(note, user, noteActivity);
|
||||
}
|
||||
}
|
||||
|
||||
async function insertNote(user: User, data: Option, tags: string[], emojis: string[], mentionedUsers: User[]) {
|
||||
const insert = new Note({
|
||||
id: genId(data.createdAt!),
|
||||
|
@ -472,34 +474,6 @@ async function notifyToWatchersOfReplyee(reply: Note, user: User, nm: Notificati
|
|||
}
|
||||
}
|
||||
|
||||
async function publishToFollowers(note: Note, user: User, noteActivity: any) {
|
||||
const followers = await Followings.find({
|
||||
followeeId: note.userId
|
||||
});
|
||||
|
||||
const queue: string[] = [];
|
||||
|
||||
for (const following of followers) {
|
||||
if (Followings.isRemoteFollower(following)) {
|
||||
// フォロワーがリモートユーザーかつ投稿者がローカルユーザーなら投稿を配信
|
||||
if (Users.isLocalUser(user)) {
|
||||
const inbox = following.followerSharedInbox || following.followerInbox;
|
||||
if (!queue.includes(inbox)) queue.push(inbox);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
for (const inbox of queue) {
|
||||
deliver(user as any, noteActivity, inbox);
|
||||
}
|
||||
}
|
||||
|
||||
function deliverNoteToMentionedRemoteUsers(mentionedUsers: User[], user: ILocalUser, noteActivity: any) {
|
||||
for (const u of mentionedUsers.filter(u => Users.isRemoteUser(u))) {
|
||||
deliver(user, noteActivity, (u as IRemoteUser).inbox);
|
||||
}
|
||||
}
|
||||
|
||||
async function createMentionedEvents(mentionedUsers: User[], note: Note, nm: NotificationManager) {
|
||||
for (const u of mentionedUsers.filter(u => Users.isLocalUser(u))) {
|
||||
const detailPackedNote = await Notes.pack(note, u, {
|
||||
|
|
|
@ -3,14 +3,14 @@ import renderDelete from '../../remote/activitypub/renderer/delete';
|
|||
import renderAnnounce from '../../remote/activitypub/renderer/announce';
|
||||
import renderUndo from '../../remote/activitypub/renderer/undo';
|
||||
import { renderActivity } from '../../remote/activitypub/renderer';
|
||||
import { deliver } from '../../queue';
|
||||
import renderTombstone from '../../remote/activitypub/renderer/tombstone';
|
||||
import config from '../../config';
|
||||
import { registerOrFetchInstanceDoc } from '../register-or-fetch-instance-doc';
|
||||
import { User } from '../../models/entities/user';
|
||||
import { Note } from '../../models/entities/note';
|
||||
import { Notes, Users, Followings, Instances } from '../../models';
|
||||
import { Notes, Users, Instances } from '../../models';
|
||||
import { notesChart, perUserNotesChart, instanceChart } from '../chart';
|
||||
import { deliverToFollowers } from '../../remote/activitypub/deliver-manager';
|
||||
|
||||
/**
|
||||
* 投稿を削除します。
|
||||
|
@ -49,22 +49,7 @@ export default async function(user: User, note: Note, quiet = false) {
|
|||
? renderUndo(renderAnnounce(renote.uri || `${config.url}/notes/${renote.id}`, note), user)
|
||||
: renderDelete(renderTombstone(`${config.url}/notes/${note.id}`), user));
|
||||
|
||||
const queue: string[] = [];
|
||||
|
||||
const followers = await Followings.find({
|
||||
followeeId: note.userId
|
||||
});
|
||||
|
||||
for (const following of followers) {
|
||||
if (Followings.isRemoteFollower(following)) {
|
||||
const inbox = following.followerSharedInbox || following.followerInbox;
|
||||
if (!queue.includes(inbox)) queue.push(inbox);
|
||||
}
|
||||
}
|
||||
|
||||
for (const inbox of queue) {
|
||||
deliver(user as any, content, inbox);
|
||||
}
|
||||
deliverToFollowers(user, content);
|
||||
}
|
||||
//#endregion
|
||||
|
||||
|
|
|
@ -1,9 +1,9 @@
|
|||
import renderUpdate from '../../../remote/activitypub/renderer/update';
|
||||
import { renderActivity } from '../../../remote/activitypub/renderer';
|
||||
import { deliver } from '../../../queue';
|
||||
import renderNote from '../../../remote/activitypub/renderer/note';
|
||||
import { Users, Notes, Followings } from '../../../models';
|
||||
import { Users, Notes } from '../../../models';
|
||||
import { Note } from '../../../models/entities/note';
|
||||
import { deliverToFollowers } from '../../../remote/activitypub/deliver-manager';
|
||||
|
||||
export async function deliverQuestionUpdate(noteId: Note['id']) {
|
||||
const note = await Notes.findOne(noteId);
|
||||
|
@ -12,26 +12,9 @@ export async function deliverQuestionUpdate(noteId: Note['id']) {
|
|||
const user = await Users.findOne(note.userId);
|
||||
if (user == null) throw new Error('note not found');
|
||||
|
||||
const followers = await Followings.find({
|
||||
followeeId: user.id
|
||||
});
|
||||
|
||||
const queue: string[] = [];
|
||||
|
||||
// フォロワーがリモートユーザーかつ投稿者がローカルユーザーならUpdateを配信
|
||||
if (Users.isLocalUser(user)) {
|
||||
for (const following of followers) {
|
||||
if (Followings.isRemoteFollower(following)) {
|
||||
const inbox = following.followerSharedInbox || following.followerInbox;
|
||||
if (!queue.includes(inbox)) queue.push(inbox);
|
||||
}
|
||||
}
|
||||
|
||||
if (queue.length > 0) {
|
||||
const content = renderActivity(renderUpdate(await renderNote(note, false), user));
|
||||
for (const inbox of queue) {
|
||||
deliver(user, content, inbox);
|
||||
}
|
||||
}
|
||||
deliverToFollowers(user, content);
|
||||
}
|
||||
}
|
||||
|
|
Loading…
Reference in New Issue