Revert "Merge branch 'refactor-deliver-actor' into mk.absturztau.be"

This reverts commit 76b4e4ee08, reversing
changes made to 22f2ac427f.

packages/backend/src/queue/index.ts

@@ -83,11 +83,14 @@ webhookDeliverQueue
 	.on('error', (job: any, err: Error) => webhookLogger.error(`error ${err}`, { job, e: renderError(err) }))
 	.on('stalled', (job) => webhookLogger.warn(`stalled ${getJobInfo(job)} to=${job.data.to}`));
 
-export function deliver(content: unknown, to: string | null) {
+export function deliver(user: ThinUser, content: unknown, to: string | null) {
 	if (content == null) return null;
 	if (to == null) return null;
 
 	const data = {
+		user: {
+			id: user.id,
+		},
 		content,
 		to,
 	};

packages/backend/src/queue/processors/deliver.ts

@@ -9,7 +9,6 @@ import { toPuny } from '@/misc/convert-host.js';
 import { StatusError } from '@/misc/fetch.js';
 import { shouldSkipInstance } from '@/misc/skipped-instances.js';
 import { DeliverJobData } from '@/queue/types.js';
-import { DbResolver } from '@/remote/activitypub/db-resolver.js';
 
 const logger = new Logger('deliver');
 
@@ -19,20 +18,13 @@ export default async (job: Bull.Job<DeliverJobData>) => {
 
 	if (await shouldSkipInstance(puny)) return 'skip';
 
-	// get user/actor for signing
-	const userUri = job.data.content.actor;
-	if (userUri == null) return 'error: missing actor';
-	const user = await new DbResolver().getUserFromApId(userUri);
-	if (user == null) return 'error: actor not found';
-	if (user.host != null) return 'error: actor not local';
-
 	try {
 		if (Array.isArray(job.data.content)) {
 			await Promise.all(
-				job.data.content.map(x => request(user, job.data.to, x))
+				job.data.content.map(x => request(job.data.user, job.data.to, x))
 			);
 		} else {
-			await request(user, job.data.to, job.data.content);
+			await request(job.data.user, job.data.to, job.data.content);
 		}
 
 		// Update stats

packages/backend/src/queue/types.ts

@@ -6,8 +6,10 @@ import { Webhook } from '@/models/entities/webhook.js';
 import { IActivity } from '@/remote/activitypub/type.js';
 
 export type DeliverJobData = {
-	/** Activity, containing the actor URI */
-	content: IActivity;
+	/** Actor */
+	user: ThinUser;
+	/** Activity */
+	content: unknown;
 	/** inbox URL to deliver */
 	to: string;
 };

packages/backend/src/remote/activitypub/deliver-manager.ts

@@ -14,7 +14,6 @@ interface IEveryoneRecipe extends IRecipe {
 
 interface IFollowersRecipe extends IRecipe {
 	type: 'Followers';
-	followee: ILocalUser;
 }
 
 interface IDirectRecipe extends IRecipe {
@@ -33,24 +32,26 @@ const isDirect = (recipe: any): recipe is IDirectRecipe =>
 //#endregion
 
 export class DeliverManager {
+	private actor: { id: User['id']; host: null; };
 	private activity: any;
 	private recipes: IRecipe[] = [];
 
 	/**
 	 * Constructor
+	 * @param actor Actor
 	 * @param activity Activity to deliver
 	 */
-	constructor(activity: any) {
+	constructor(actor: { id: User['id']; host: null; }, activity: any) {
+		this.actor = actor;
 		this.activity = activity;
 	}
 
 	/**
 	 * Add recipe for followers deliver
 	 */
-	public addFollowersRecipe(followee: ILocalUser) {
+	public addFollowersRecipe() {
 		const deliver = {
 			type: 'Followers',
-			followee,
 		} as IFollowersRecipe;
 
 		this.addRecipe(deliver);
@@ -88,6 +89,8 @@ export class DeliverManager {
 	 * Execute delivers
 	 */
 	public async execute() {
+		if (!Users.isLocalUser(this.actor)) return;
+
 		const inboxes = new Set<string>();
 
 		/*
@@ -113,25 +116,21 @@ export class DeliverManager {
 			}
 		}
 
-		await Promise.all(
-			this.recipes.filter(isFollowers)
-			.map(recipe => {
+		if (this.recipes.some(r => isFollowers(r))) {
 			// followers deliver
-				return Followings.createQueryBuilder('followings')
+			const followers = await Followings.createQueryBuilder('followings')
 				// return either the shared inbox (if available) or the individual inbox
 				.select('COALESCE(followings.followerSharedInbox, followings.followerInbox)', 'inbox')
 				// so we don't have to make our inboxes Set work as hard
 				.distinct(true)
 				// ...for the specific actors followers
-					.where('followings.followeeId = :actorId', { actorId: recipe.followee.id })
+				.where('followings.followeeId = :actorId', { actorId: this.actor.id })
 				// don't deliver to ourselves
 				.andWhere('followings.followerHost IS NOT NULL')
-					.getRawMany()
-					.then(followers =>
-						followers.forEach(({ inbox }) => inboxes.add(inbox))
-					);
-			})
-		);
+				.getRawMany();
+
+			followers.forEach(({ inbox }) => inboxes.add(inbox));
+		}
 
 		this.recipes.filter((recipe): recipe is IDirectRecipe =>
 			// followers recipes have already been processed
@@ -156,7 +155,7 @@ export class DeliverManager {
 			// skip instances as indicated
 			if (instancesToSkip.includes(new URL(inbox).host)) continue;
 
-			deliver(this.activity, inbox);
+			deliver(this.actor, this.activity, inbox);
 		}
 	}
 }
@@ -167,9 +166,9 @@ export class DeliverManager {
  * @param activity Activity
  * @param from Followee
  */
-export async function deliverToFollowers(actor: ILocalUser, activity: any) {
-	const manager = new DeliverManager(activity);
-	manager.addFollowersRecipe(actor);
+export async function deliverToFollowers(actor: { id: ILocalUser['id']; host: null; }, activity: any) {
+	const manager = new DeliverManager(actor, activity);
+	manager.addFollowersRecipe();
 	await manager.execute();
 }
 
@@ -178,8 +177,8 @@ export async function deliverToFollowers(actor: ILocalUser, activity: any) {
  * @param activity Activity
  * @param to Target user
  */
-export async function deliverToUser(activity: any, to: IRemoteUser) {
-	const manager = new DeliverManager(activity);
+export async function deliverToUser(actor: { id: ILocalUser['id']; host: null; }, activity: any, to: IRemoteUser) {
+	const manager = new DeliverManager(actor, activity);
 	manager.addDirectRecipe(to);
 	await manager.execute();
 }

packages/backend/src/remote/activitypub/renderer/read.ts

@@ -0,0 +1,9 @@
+import config from '@/config/index.js';
+import { User } from '@/models/entities/user.js';
+import { MessagingMessage } from '@/models/entities/messaging-message.js';
+
+export const renderReadActivity = (user: { id: User['id'] }, message: MessagingMessage) => ({
+	type: 'Read',
+	actor: `${config.url}/users/${user.id}`,
+	object: message.uri,
+});

packages/backend/src/server/api/common/read-messaging-message.ts

@@ -7,6 +7,10 @@ import { MessagingMessages, UserGroupJoinings, Users } from '@/models/index.js';
 import { IdentifiableError } from '@/misc/identifiable-error.js';
 import { UserGroup } from '@/models/entities/user-group.js';
 import { toArray } from '@/prelude/array.js';
+import { renderReadActivity } from '@/remote/activitypub/renderer/read.js';
+import { renderActivity } from '@/remote/activitypub/renderer/index.js';
+import { deliver } from '@/queue/index.js';
+import orderedCollection from '@/remote/activitypub/renderer/ordered-collection.js';
 
 /**
  * Mark messages as read
@@ -129,3 +133,18 @@ export async function readGroupMessagingMessage(
 		}
 	}
 }
+
+export async function deliverReadActivity(user: { id: User['id']; host: null; }, recipient: IRemoteUser, messages: MessagingMessage | MessagingMessage[]) {
+	const contents = toArray(messages)
+		.filter(x => x.uri)
+		.map(x => renderReadActivity(user, x));
+
+	if (contents.length > 1) {
+		const collection = orderedCollection(null, contents.length, undefined, undefined, contents);
+		deliver(user, renderActivity(collection), recipient.inbox);
+	} else {
+		for (const content of contents) {
+			deliver(user, renderActivity(content), recipient.inbox);
+		}
+	}
+}

packages/backend/src/server/api/endpoints/admin/reports/resolve.ts

@@ -29,7 +29,7 @@ export default define(meta, paramDef, async (ps, me) => {
 		const actor = await getInstanceActor();
 		const targetUser = await Users.findOneByOrFail({ id: report.targetUserId });
 
-		deliver(renderActivity(renderFlag(actor, report)), targetUser.inbox);
+		deliver(actor, renderActivity(renderFlag(actor, report)), targetUser.inbox);
 	}
 
 	await AbuseUserReports.update(report.id, {

packages/backend/src/server/api/endpoints/messaging/messages.ts

@@ -4,7 +4,7 @@ import define from '@/server/api/define.js';
 import { ApiError } from '@/server/api/error.js';
 import { getUser } from '@/server/api/common/getters.js';
 import { makePaginationQuery } from '@/server/api/common/make-pagination-query.js';
-import { readUserMessagingMessage, readGroupMessagingMessage } from '@/server/api/common/read-messaging-message.js';
+import { readUserMessagingMessage, readGroupMessagingMessage, deliverReadActivity } from '@/server/api/common/read-messaging-message.js';
 
 export const meta = {
 	tags: ['messaging'],
@@ -75,6 +75,11 @@ export default define(meta, paramDef, async (ps, user) => {
 		// Mark all as read
 		if (ps.markAsRead) {
 			readUserMessagingMessage(user.id, recipient.id, messages.filter(m => m.recipientId === user.id).map(x => x.id));
+
+			// リモートユーザーとのメッセージだったら既読配信
+			if (Users.isLocalUser(user) && Users.isRemoteUser(recipient)) {
+				deliverReadActivity(user, recipient, messages);
+			}
 		}
 
 		return await Promise.all(messages.map(message => MessagingMessages.pack(message, user, {

packages/backend/src/server/api/endpoints/notes/polls/vote.ts

@@ -133,7 +133,7 @@ export default define(meta, paramDef, async (ps, user) => {
 	if (note.userHost != null) {
 		const pollOwner = await Users.findOneByOrFail({ id: note.userId }) as IRemoteUser;
 
-		deliver(renderActivity(await renderVote(user, vote, note, poll, pollOwner)), pollOwner.inbox);
+		deliver(user, renderActivity(await renderVote(user, vote, note, poll, pollOwner)), pollOwner.inbox);
 	}
 
 	// リモートフォロワーにUpdate配信

packages/backend/src/server/api/stream/channels/messaging.ts

@@ -1,7 +1,7 @@
 import { UserGroupJoinings, Users, MessagingMessages } from '@/models/index.js';
 import { User, ILocalUser, IRemoteUser } from '@/models/entities/user.js';
 import { UserGroup } from '@/models/entities/user-group.js';
-import { readUserMessagingMessage, readGroupMessagingMessage } from '@/server/api/common/read-messaging-message.js';
+import { readUserMessagingMessage, readGroupMessagingMessage, deliverReadActivity } from '@/server/api/common/read-messaging-message.js';
 import Channel from '@/server/api/stream/channel.js';
 import { StreamMessages } from '@/server/api/stream/types.js';
 
@@ -69,6 +69,13 @@ export default class extends Channel {
 			case 'read':
 				if (this.otherpartyId) {
 					readUserMessagingMessage(this.user!.id, this.otherpartyId, [body.id]);
+
+					// リモートユーザーからのメッセージだったら既読配信
+					if (Users.isLocalUser(this.user!) && Users.isRemoteUser(this.otherparty!)) {
+						MessagingMessages.findOneBy({ id: body.id }).then(message => {
+							if (message) deliverReadActivity(this.user as ILocalUser, this.otherparty as IRemoteUser, message);
+						});
+					}
 				} else if (this.groupId) {
 					readGroupMessagingMessage(this.user!.id, this.groupId, [body.id]);
 				}

packages/backend/src/services/blocking/create.ts

@@ -34,7 +34,7 @@ export default async function(blocker: User, blockee: User): Promise<void> {
 
 	if (Users.isLocalUser(blocker) && Users.isRemoteUser(blockee) && blocker.federateBlocks) {
 		const content = renderActivity(renderBlock(blocking));
-		deliver(content, blockee.inbox);
+		deliver(blocker, content, blockee.inbox);
 	}
 }
 
@@ -78,13 +78,13 @@ async function cancelRequest(follower: User, followee: User): Promise<void> {
 	// Send Undo Follow if followee is remote
 	if (Users.isLocalUser(follower) && Users.isRemoteUser(followee)) {
 		const content = renderActivity(renderUndo(renderFollow(follower, followee), follower));
-		deliver(content, followee.inbox);
+		deliver(follower, content, followee.inbox);
 	}
 
 	// Send Reject if follower is remote
 	if (Users.isRemoteUser(follower) && Users.isLocalUser(followee)) {
 		const content = renderActivity(renderReject(renderFollow(follower, followee, request.requestId!), followee));
-		deliver(content, follower.inbox);
+		deliver(followee, content, follower.inbox);
 	}
 }
 
@@ -125,7 +125,7 @@ async function unFollow(follower: User, followee: User): Promise<void> {
 	// Send Undo Follow if follower is remote
 	if (Users.isLocalUser(follower) && Users.isRemoteUser(followee)) {
 		const content = renderActivity(renderUndo(renderFollow(follower, followee), follower));
-		deliver(content, followee.inbox);
+		deliver(follower, content, followee.inbox);
 	}
 }
 

packages/backend/src/services/blocking/delete.ts

@@ -29,6 +29,6 @@ export default async function(blocker: CacheableUser, blockee: CacheableUser) {
 	// deliver if remote bloking
 	if (Users.isLocalUser(blocker) && Users.isRemoteUser(blockee)) {
 		const content = renderActivity(renderUndo(renderBlock(blocking), blocker));
-		deliver(content, blockee.inbox);
+		deliver(blocker, content, blockee.inbox);
 	}
 }

packages/backend/src/services/following/create.ts

@@ -146,7 +146,7 @@ export default async function(_follower: { id: User['id'] }, _followee: { id: Us
 	if (Users.isRemoteUser(follower) && Users.isLocalUser(followee) && blocked) {
 		// リモートフォローを受けてブロックしていた場合は、エラーにするのではなくRejectを送り返しておしまい。
 		const content = renderActivity(renderReject(renderFollow(follower, followee, requestId), followee));
-		deliver(content, follower.inbox);
+		deliver(followee , content, follower.inbox);
 		return;
 	} else if (Users.isRemoteUser(follower) && Users.isLocalUser(followee) && blocking) {
 		// リモートフォローを受けてブロックされているはずの場合だったら、ブロック解除しておく。
@@ -195,6 +195,6 @@ export default async function(_follower: { id: User['id'] }, _followee: { id: Us
 
 	if (Users.isRemoteUser(follower) && Users.isLocalUser(followee)) {
 		const content = renderActivity(renderAccept(renderFollow(follower, followee, requestId), followee));
-		deliver(content, follower.inbox);
+		deliver(followee, content, follower.inbox);
 	}
 }

packages/backend/src/services/following/delete.ts

@@ -52,13 +52,13 @@ export default async function(follower: { id: User['id']; host: User['host']; ur
 
 	if (Users.isLocalUser(follower) && Users.isRemoteUser(followee)) {
 		const content = renderActivity(renderUndo(renderFollow(follower, followee), follower));
-		deliver(content, followee.inbox);
+		deliver(follower, content, followee.inbox);
 	}
 
 	if (Users.isLocalUser(followee) && Users.isRemoteUser(follower)) {
 		// local user has null host
 		const content = renderActivity(renderReject(renderFollow(follower, followee), followee));
-		deliver(content, follower.inbox);
+		deliver(followee, content, follower.inbox);
 	}
 }
 

packages/backend/src/services/following/reject.ts

@@ -99,7 +99,7 @@ async function deliverReject(followee: Local, follower: Remote): Promise<void> {
 	});
 
 	const content = renderActivity(renderReject(renderFollow(follower, followee, request?.requestId || undefined), followee));
-	deliver(content, follower.inbox);
+	deliver(followee, content, follower.inbox);
 }
 
 /**

packages/backend/src/services/following/requests/accept.ts

@@ -27,7 +27,7 @@ export async function acceptFollowRequest(followee: User, follower: User): Promi
 
 	if (Users.isRemoteUser(follower) && Users.isLocalUser(followee)) {
 		const content = renderActivity(renderAccept(renderFollow(follower, followee, request.requestId!), followee));
-		deliver(content, follower.inbox);
+		deliver(followee, content, follower.inbox);
 	}
 
 	Users.pack(followee.id, followee, {

packages/backend/src/services/following/requests/cancel.ts

@@ -17,7 +17,7 @@ export async function cancelFollowRequest(followee: User, follower: User): Promi
 		const content = renderActivity(renderUndo(renderFollow(follower, followee), follower));
 
 		if (Users.isLocalUser(follower)) {
-			deliver(content, followee.inbox);
+			deliver(follower, content, followee.inbox);
 		}
 	}
 

packages/backend/src/services/following/requests/create.ts

@@ -64,6 +64,6 @@ export async function createFollowRequest(follower: User, followee: User, reques
 
 	if (Users.isLocalUser(follower) && Users.isRemoteUser(followee)) {
 		const content = renderActivity(renderFollow(follower, followee));
-		deliver(content, followee.inbox);
+		deliver(follower, content, followee.inbox);
 	}
 }

packages/backend/src/services/messages/create.ts

@@ -97,7 +97,7 @@ export async function createMessage(user: { id: User['id']; host: User['host'];
 
 		const activity = renderActivity(renderCreate(await renderNote(note, false, true), note));
 
-		deliver(activity, recipientUser.inbox);
+		deliver(user, activity, recipientUser.inbox);
 	}
 	return messageObj;
 }

packages/backend/src/services/messages/delete.ts

@@ -22,7 +22,7 @@ async function postDeleteMessage(message: MessagingMessage): Promise<void> {
 
 		if (Users.isLocalUser(user) && Users.isRemoteUser(recipient)) {
 			const activity = renderActivity(renderDelete(renderTombstone(`${config.url}/notes/${message.id}`), user));
-			deliver(activity, recipient.inbox);
+			deliver(user, activity, recipient.inbox);
 		}
 	} else if (message.groupId) {
 		publishGroupMessagingStream(message.groupId, 'deleted', message.id);

packages/backend/src/services/note/create.ts

@@ -432,7 +432,7 @@ export default async (user: { id: User['id']; username: User['username']; host:
 		if (Users.isLocalUser(user) && !data.localOnly) {
 			(async () => {
 				const noteActivity = renderActivity(await renderNoteOrRenoteActivity(note));
-				const dm = new DeliverManager(noteActivity);
+				const dm = new DeliverManager(user, noteActivity);
 
 				// Delivered to remote users who have been mentioned
 				for (const u of mentionedUsers.filter(u => Users.isRemoteUser(u))) {
@@ -453,7 +453,7 @@ export default async (user: { id: User['id']; username: User['username']; host:
 
 				// Deliver to followers
 				if (['public', 'home', 'followers'].includes(note.visibility)) {
-					dm.addFollowersRecipe(user);
+					dm.addFollowersRecipe();
 				}
 
 				if (['public'].includes(note.visibility)) {

packages/backend/src/services/note/delete.ts

@@ -60,8 +60,8 @@ export async function deleteNotes(notes: Note[], user?: User): Promise<void> {
 
 	// Compute addressing information.
 	// Since we do not send any actual content, we send all note deletions to everyone.
-	const manager = new DeliverManager(content);
-	manager.addFollowersRecipe(fetchedUser);
+	const manager = new DeliverManager(fetchedUser, content);
+	manager.addFollowersRecipe();
 	manager.addEveryone();
 	// Check mentioned users, since not all may have a shared inbox.
 	await Promise.all(

packages/backend/src/services/note/reaction/create.ts

@@ -130,14 +130,14 @@ export async function createReaction(user: { id: User['id']; host: User['host'];
 	//#region 配信
 	if (Users.isLocalUser(user) && !note.localOnly) {
 		const content = renderActivity(await renderLike(record, note));
-		const dm = new DeliverManager(content);
+		const dm = new DeliverManager(user, content);
 		if (note.userHost !== null) {
 			const reactee = await Users.findOneBy({ id: note.userId });
 			dm.addDirectRecipe(reactee as IRemoteUser);
 		}
 
 		if (['public', 'home', 'followers'].includes(note.visibility)) {
-			dm.addFollowersRecipe(user);
+			dm.addFollowersRecipe();
 		} else if (note.visibility === 'specified') {
 			const visibleUsers = await Promise.all(note.visibleUserIds.map(id => Users.findOneBy({ id })));
 			for (const u of visibleUsers.filter(u => u && Users.isRemoteUser(u))) {

packages/backend/src/services/note/reaction/delete.ts

@@ -46,12 +46,12 @@ export async function deleteReaction(user: { id: User['id']; host: User['host'];
 	//#region 配信
 	if (Users.isLocalUser(user) && !note.localOnly) {
 		const content = renderActivity(renderUndo(await renderLike(exist, note), user));
-		const dm = new DeliverManager(content);
+		const dm = new DeliverManager(user, content);
 		if (note.userHost !== null) {
 			const reactee = await Users.findOneBy({ id: note.userId });
 			dm.addDirectRecipe(reactee as IRemoteUser);
 		}
-		dm.addFollowersRecipe(user);
+		dm.addFollowersRecipe();
 		dm.execute();
 	}
 	//#endregion

packages/backend/src/services/relay.ts

@@ -36,8 +36,9 @@ export async function addRelay(inbox: string): Promise<Relay> {
 	}).then(x => Relays.findOneByOrFail(x.identifiers[0]));
 
 	const relayActor = await getRelayActor();
-	const activity = renderActivity(renderFollowRelay(relay, relayActor));
-	deliver(activity, relay.inbox);
+	const follow = renderFollowRelay(relay, relayActor);
+	const activity = renderActivity(follow);
+	deliver(relayActor, activity, relay.inbox);
 
 	return relay;
 }
@@ -52,14 +53,17 @@ export async function removeRelay(inbox: string): Promise<void> {
 	}
 
 	const relayActor = await getRelayActor();
-	const activity = renderActivity(renderUndo(renderFollowRelay(relay, relayActor), relayActor));
-	deliver(activity, relay.inbox);
+	const follow = renderFollowRelay(relay, relayActor);
+	const undo = renderUndo(follow, relayActor);
+	const activity = renderActivity(undo);
+	deliver(relayActor, activity, relay.inbox);
 
 	await Relays.delete(relay.id);
 }
 
 export async function listRelay(): Promise<Relay[]> {
-	return await Relays.find();
+	const relays = await Relays.find();
+	return relays;
 }
 
 export async function relayAccepted(id: string): Promise<string> {
@@ -90,7 +94,7 @@ export async function deliverToRelays(user: { id: User['id']; host: null; }, act
 	const signed = await attachLdSignature(copy, user);
 
 	for (const relay of relays) {
-		deliver(signed, relay.inbox);
+		deliver(user, signed, relay.inbox);
 	}
 }
 
@@ -108,6 +112,6 @@ export async function deliverMultipleToRelays(user: User, activities: any[]): Pr
 	}));
 
 	for (const relay of relays) {
-		deliver(content, relay.inbox);
+		deliver(user, content, relay.inbox);
 	}
 }

packages/backend/src/services/suspend-user.ts

@@ -13,7 +13,7 @@ export async function doPostSuspend(user: { id: User['id']; host: User['host'] }
 		const content = renderActivity(renderDelete(`${config.url}/users/${user.id}`, user));
 
 		// deliver to all of known network
-		const dm = new DeliverManager(content);
+		const dm = new DeliverManager(user, content);
 		dm.addEveryone();
 		await dm.execute();
 	}

packages/backend/src/services/unsuspend-user.ts

@@ -32,7 +32,7 @@ export async function doPostUnsuspend(user: User): Promise<void> {
 		}
 
 		for (const inbox of queue) {
-			deliver(content, inbox);
+			deliver(user as any, content, inbox);
 		}
 	}
 }