merge: Fix account migrations (!1229)

View MR for information: https://activitypub.software/TransFem-org/Sharkey/-/merge_requests/1229

Approved-by: dakkar <dakkar@thenautilus.net>
Approved-by: Marie <github@yuugi.dev>

packages/backend/src/core/AccountMoveService.ts

@@ -9,6 +9,7 @@ import { IsNull, In, MoreThan, Not } from 'typeorm';
 import { bindThis } from '@/decorators.js';
 import { DI } from '@/di-symbols.js';
 import type { MiLocalUser, MiRemoteUser, MiUser } from '@/models/User.js';
+import { isLocalUser } from '@/models/User.js';
 import type { BlockingsRepository, FollowingsRepository, InstancesRepository, MiMeta, MutingsRepository, UserListMembershipsRepository, UsersRepository, NoteScheduleRepository, MiNoteSchedule } from '@/models/_.js';
 import type { RelationshipJobData, ThinUser } from '@/queue/types.js';
 
@@ -29,13 +30,25 @@ import { AntennaService } from '@/core/AntennaService.js';
 import { CacheService } from '@/core/CacheService.js';
 import { UserListService } from '@/core/UserListService.js';
 import { TimeService } from '@/global/TimeService.js';
+import { InternalEventService } from '@/global/InternalEventService.js';
+import { LoggerService } from '@/core/LoggerService.js';
+import type Logger from '@/logger.js';
+import { renderInlineError } from '@/misc/render-inline-error.js';
+import { IdentifiableError } from '@/misc/identifiable-error.js';
+import type { Packed } from '@/misc/json-schema.js';
+import type { Config } from '@/config.js';
 
 @Injectable()
 export class AccountMoveService {
+	private readonly logger: Logger;
+
 	constructor(
 		@Inject(DI.meta)
 		private meta: MiMeta,
 
+		@Inject(DI.config)
+		private readonly config: Config,
+
 		@Inject(DI.usersRepository)
 		private usersRepository: UsersRepository,
 
@@ -74,7 +87,27 @@ export class AccountMoveService {
 		private readonly cacheService: CacheService,
 		private readonly userListService: UserListService,
 		private readonly timeService: TimeService,
+		private readonly internalEventService: InternalEventService,
+		private readonly loggerService: LoggerService,
 	) {
+		this.logger = this.loggerService.getLogger('account-move');
+	}
+
+	@bindThis
+	public async restartMigration(src: MiUser): Promise<void> {
+		if (!src.movedToUri) {
+			throw new IdentifiableError('ddcf173a-00f2-4aa4-ba12-cddd131bacf4', `Can't restart migrated for user ${src.id}: user has not migrated`);
+		}
+
+		const dst = await this.apPersonService.resolvePerson(src.movedToUri);
+		this.logger.info(`Restarting migration from ${src.id} (@${src.usernameLower}@${src.host ?? this.config.host}) to ${dst.id} (@${dst.usernameLower}@${dst.host ?? this.config.host})`);
+
+		if (isLocalUser(src)) {
+			// This calls createMoveJob at the end
+			await this.moveFromLocal(src, dst);
+		} else {
+			await this.queueService.createMoveJob(src, dst);
+		}
 	}
 
 	/**
@@ -83,7 +116,7 @@ export class AccountMoveService {
 	 * After delivering Move activity, its local followers unfollow the old account and then follow the new one.
 	 */
 	@bindThis
-	public async moveFromLocal(src: MiLocalUser, dst: MiLocalUser | MiRemoteUser): Promise<unknown> {
+	public async moveFromLocal(src: MiLocalUser, dst: MiLocalUser | MiRemoteUser): Promise<Packed<'MeDetailed'>> {
 		const srcUri = this.userEntityService.getUserUri(src);
 		const dstUri = this.userEntityService.getUserUri(dst);
 
@@ -96,7 +129,7 @@ export class AccountMoveService {
 		Object.assign(src, update);
 
 		// Update cache
-		this.globalEventService.publishInternalEvent('localUserUpdated', src);
+		await this.internalEventService.emit('localUserUpdated', { id: src.id });
 
 		const srcPerson = await this.apRendererService.renderPerson(src);
 		const updateAct = this.apRendererService.addContext(this.apRendererService.renderUpdate(srcPerson, src));
@@ -114,12 +147,12 @@ export class AccountMoveService {
 
 		// Unfollow after 24 hours
 		const followings = await this.cacheService.userFollowingsCache.fetch(src.id);
-		this.queueService.createDelayedUnfollowJob(Array.from(followings.keys()).map(followeeId => ({
+		await this.queueService.createDelayedUnfollowJob(Array.from(followings.keys()).map(followeeId => ({
 			from: { id: src.id },
 			to: { id: followeeId },
 		})), process.env.NODE_ENV === 'test' ? 10000 : 1000 * 60 * 60 * 24);
 
-		await this.postMoveProcess(src, dst);
+		await this.queueService.createMoveJob(src, dst);
 
 		return iObj;
 	}
@@ -127,18 +160,18 @@ export class AccountMoveService {
 	@bindThis
 	public async postMoveProcess(src: MiUser, dst: MiUser): Promise<void> {
 		// Copy blockings and mutings, and update lists
-		try {
-			await Promise.all([
-				this.copyBlocking(src, dst),
-				this.copyMutings(src, dst),
-				this.deleteScheduledNotes(src),
-				this.copyRoles(src, dst),
-				this.updateLists(src, dst),
-				this.antennaService.onMoveAccount(src, dst),
-			]);
-		} catch {
-			/* skip if any error happens */
-		}
+		await this.copyBlocking(src, dst)
+			.catch(err => this.logger.warn(`Error copying blockings in migration ${src.id} (@${src.usernameLower}@${src.host ?? this.config.host}) to ${dst.id} (@${dst.usernameLower}@${dst.host ?? this.config.host}): ${renderInlineError(err)}`));
+		await this.copyMutings(src, dst)
+			.catch(err => this.logger.warn(`Error copying mutings in migration ${src.id} (@${src.usernameLower}@${src.host ?? this.config.host}) to ${dst.id} (@${dst.usernameLower}@${dst.host ?? this.config.host}): ${renderInlineError(err)}`));
+		await this.deleteScheduledNotes(src)
+			.catch(err => this.logger.warn(`Error deleting scheduled notes in migration ${src.id} (@${src.usernameLower}@${src.host ?? this.config.host}) to ${dst.id} (@${dst.usernameLower}@${dst.host ?? this.config.host}): ${renderInlineError(err)}`));
+		await this.copyRoles(src, dst)
+			.catch(err => this.logger.warn(`Error copying roles in migration ${src.id} (@${src.usernameLower}@${src.host ?? this.config.host}) to ${dst.id} (@${dst.usernameLower}@${dst.host ?? this.config.host}): ${renderInlineError(err)}`));
+		await this.updateLists(src, dst)
+			.catch(err => this.logger.warn(`Error updating lists in migration ${src.id} (@${src.usernameLower}@${src.host ?? this.config.host}) to ${dst.id} (@${dst.usernameLower}@${dst.host ?? this.config.host}): ${renderInlineError(err)}`));
+		await this.antennaService.onMoveAccount(src, dst)
+			.catch(err => this.logger.warn(`Error updating antennas in migration ${src.id} (@${src.usernameLower}@${src.host ?? this.config.host}) to ${dst.id} (@${dst.usernameLower}@${dst.host ?? this.config.host}): ${renderInlineError(err)}`));
 
 		// follow the new account
 		const proxy = await this.systemAccountService.fetch('proxy');
@@ -153,29 +186,33 @@ export class AccountMoveService {
 		// Decrease following count instead of unfollowing.
 		try {
 			await this.adjustFollowingCounts(followJobs.map(job => job.from.id), src);
-		} catch {
+		} catch (err) {
 			/* skip if any error happens */
+			this.logger.warn(`Non-fatal exception in migration from ${src.id} (@${src.usernameLower}@${src.host ?? this.config.host}) to ${dst.id} (@${dst.usernameLower}@${dst.host ?? this.config.host}): ${renderInlineError(err)}`);
 		}
 
 		// Should be queued because this can cause a number of follow per one move.
-		this.queueService.createFollowJob(followJobs);
+		await this.queueService.createFollowJob(followJobs);
 	}
 
 	@bindThis
 	public async copyBlocking(src: ThinUser, dst: ThinUser): Promise<void> {
 		// Followers shouldn't overlap with blockers, but the destination account, different from the blockee (i.e., old account), may have followed the local user before moving.
 		// So block the destination account here.
-		const srcBlockings = await this.blockingsRepository.findBy({ blockeeId: src.id });
-		const dstBlockings = await this.blockingsRepository.findBy({ blockeeId: dst.id });
-		const blockerIds = dstBlockings.map(blocking => blocking.blockerId);
+		const [srcBlockers, dstBlockers, dstFollowers] = await Promise.all([
+			this.cacheService.userBlockedCache.fetch(src.id),
+			this.cacheService.userBlockedCache.fetch(dst.id),
+			this.cacheService.userFollowersCache.fetch(dst.id),
+		]);
 		// reblock the destination account
 		const blockJobs: RelationshipJobData[] = [];
-		for (const blocking of srcBlockings) {
-			if (blockerIds.includes(blocking.blockerId)) continue; // skip if already blocked
-			blockJobs.push({ from: { id: blocking.blockerId }, to: { id: dst.id } });
+		for (const blockerId of srcBlockers) {
+			if (dstBlockers.has(blockerId)) continue; // skip if already blocked
+			if (dstFollowers.has(blockerId)) continue; // skip if already following
+			blockJobs.push({ from: { id: blockerId }, to: { id: dst.id } });
 		}
 		// no need to unblock the old account because it may be still functional
-		this.queueService.createBlockJob(blockJobs);
+		await this.queueService.createBlockJob(blockJobs);
 	}
 
 	@bindThis
@@ -188,9 +225,12 @@ export class AccountMoveService {
 		if (oldMutings.length === 0) return;
 
 		// Check if the destination account is already indefinitely muted by the muter
-		const existingMutingsMuterUserIds = await this.mutingsRepository.findBy(
-			{ muteeId: dst.id, expiresAt: IsNull() },
-		).then(mutings => mutings.map(muting => muting.muterId));
+		const [existingMutingsMuterUserIds, dstFollowers] = await Promise.all([
+			this.mutingsRepository.findBy(
+				{ muteeId: dst.id, expiresAt: IsNull() },
+			).then(mutings => mutings.map(muting => muting.muterId)),
+			this.cacheService.userFollowersCache.fetch(dst.id),
+		]);
 
 		const newMutings: Map<string, { muterId: string; muteeId: string; expiresAt: Date | null; }> = new Map();
 
@@ -204,6 +244,7 @@ export class AccountMoveService {
 		};
 		for (const muting of oldMutings) {
 			if (existingMutingsMuterUserIds.includes(muting.muterId)) continue; // skip if already muted indefinitely
+			if (dstFollowers.has(muting.muterId)) continue; // skip if already following
 			newMutings.set(genId(), {
 				...muting,
 				muteeId: dst.id,
@@ -304,22 +345,30 @@ export class AccountMoveService {
 		if (localFollowerIds.length === 0) return;
 
 		// Set the old account's following and followers counts to 0.
+		// TODO use CollapsedQueueService when merged
 		await this.usersRepository.update({ id: oldAccount.id }, { followersCount: 0, followingCount: 0 });
+		await this.internalEventService.emit(oldAccount.host == null ? 'localUserUpdated' : 'remoteUserUpdated', { id: oldAccount.id });
 
 		// Decrease following counts of local followers by 1.
+		// TODO use CollapsedQueueService when merged
 		await this.usersRepository.decrement({ id: In(localFollowerIds) }, 'followingCount', 1);
+		await this.internalEventService.emit('usersUpdated', { ids: localFollowerIds });
 
 		// Decrease follower counts of local followees by 1.
 		const oldFollowings = await this.cacheService.userFollowingsCache.fetch(oldAccount.id);
-		if (oldFollowings.size > 0) {
-			await this.usersRepository.decrement({ id: In(Array.from(oldFollowings.keys())) }, 'followersCount', 1);
+		const oldFolloweeIds = Array.from(oldFollowings.keys());
+		if (oldFolloweeIds.length > 0) {
+			// TODO use CollapsedQueueService when merged
+			await this.usersRepository.decrement({ id: In(oldFolloweeIds) }, 'followersCount', 1);
+			await this.internalEventService.emit('usersUpdated', { ids: oldFolloweeIds });
 		}
 
 		// Update instance stats by decreasing remote followers count by the number of local followers who were following the old account.
 		if (this.meta.enableStatsForFederatedInstances) {
 			if (this.userEntityService.isRemoteUser(oldAccount)) {
-				this.federatedInstanceService.fetchOrRegister(oldAccount.host).then(async i => {
-					this.instancesRepository.decrement({ id: i.id }, 'followersCount', localFollowerIds.length);
+				// TODO use CollapsedQueueService when merged
+				await this.federatedInstanceService.fetchOrRegister(oldAccount.host).then(async i => {
+					await this.instancesRepository.decrement({ id: i.id }, 'followersCount', localFollowerIds.length);
 					if (this.meta.enableChartsForFederatedInstances) {
 						this.instanceChart.updateFollowers(i.host, false);
 					}

packages/backend/src/core/CacheService.ts

@@ -261,7 +261,7 @@ export class CacheService implements OnApplicationShutdown {
 		});
 
 		this.userMutingsCache = this.cacheManagementService.createQuantumKVCache<Set<string>>('userMutings', {
-			lifetime: 1000 * 60 * 30, // 3m (workaround for mute expiration)
+			lifetime: 1000 * 60 * 3, // 3m (workaround for mute expiration)
 			fetcher: async muterId => {
 				const mutings = await this.mutingsRepository.find({ where: { muterId: muterId }, select: ['muteeId'] });
 				return new Set(mutings.map(muting => muting.muteeId));
@@ -283,7 +283,7 @@ export class CacheService implements OnApplicationShutdown {
 		});
 
 		this.userMutedCache = this.cacheManagementService.createQuantumKVCache<Set<string>>('userMuted', {
-			lifetime: 1000 * 60 * 30, // 3m (workaround for mute expiration)
+			lifetime: 1000 * 60 * 3, // 3m (workaround for mute expiration)
 			fetcher: async muteeId => {
 				const mutings = await this.mutingsRepository.find({ where: { muteeId }, select: ['muterId'] });
 				return new Set(mutings.map(muting => muting.muterId));

packages/backend/src/core/NoteDeleteService.ts

@@ -27,6 +27,7 @@ import { LatestNoteService } from '@/core/LatestNoteService.js';
 import { ApLogService } from '@/core/ApLogService.js';
 import type Logger from '@/logger.js';
 import { TimeService } from '@/global/TimeService.js';
+import { trackPromise } from '@/misc/promise-tracker.js';
 import { LoggerService } from '@/core/LoggerService.js';
 
 @Injectable()
@@ -108,7 +109,7 @@ export class NoteDeleteService {
 					? this.apRendererService.renderUndo(this.apRendererService.renderAnnounce(renote.uri ?? `${this.config.url}/notes/${renote.id}`, note), user)
 					: this.apRendererService.renderDelete(this.apRendererService.renderTombstone(`${this.config.url}/notes/${note.id}`), user));
 
-				this.deliverToConcerned(user, note, content);
+				trackPromise(this.deliverToConcerned(user, note, content));
 			}
 
 			// also deliver delete activity to cascaded notes
@@ -117,7 +118,7 @@ export class NoteDeleteService {
 				if (!cascadingNote.user) continue;
 				if (!isLocalUser(cascadingNote.user)) continue;
 				const content = this.apRendererService.addContext(this.apRendererService.renderDelete(this.apRendererService.renderTombstone(`${this.config.url}/notes/${cascadingNote.id}`), cascadingNote.user));
-				this.deliverToConcerned(cascadingNote.user, cascadingNote, content);
+				trackPromise(this.deliverToConcerned(cascadingNote.user, cascadingNote, content));
 			}
 			//#endregion
 

packages/backend/src/core/NotePiningService.ts

@@ -84,7 +84,7 @@ export class NotePiningService {
 
 		// Deliver to remote followers
 		if (this.userEntityService.isLocalUser(user) && !note.localOnly && ['public', 'home'].includes(note.visibility)) {
-			this.deliverPinnedChange(user, note.id, true);
+			await this.deliverPinnedChange(user, note.id, true);
 		}
 	}
 
@@ -105,14 +105,14 @@ export class NotePiningService {
 			throw new IdentifiableError('b302d4cf-c050-400a-bbb3-be208681f40c', `Note ${noteId} does not exist`);
 		}
 
-		this.userNotePiningsRepository.delete({
+		await this.userNotePiningsRepository.delete({
 			userId: user.id,
 			noteId: note.id,
 		});
 
 		// Deliver to remote followers
 		if (this.userEntityService.isLocalUser(user) && !note.localOnly && ['public', 'home'].includes(note.visibility)) {
-			this.deliverPinnedChange(user, noteId, false);
+			await this.deliverPinnedChange(user, noteId, false);
 		}
 	}
 

packages/backend/src/core/QueueService.ts

@@ -763,7 +763,13 @@ export class QueueService implements OnModuleInit {
 	}
 
 	@bindThis
-	private generateRelationshipJobData(name: 'follow' | 'unfollow' | 'block' | 'unblock', data: RelationshipJobData, opts: Bull.JobsOptions = {}): {
+	public createMoveJob(from: ThinUser, to: ThinUser) {
+		const job = this.generateRelationshipJobData('move', { from, to });
+		return this.relationshipQueue.add(job.name, job.data, job.opts);
+	}
+
+	@bindThis
+	private generateRelationshipJobData(name: 'follow' | 'unfollow' | 'block' | 'unblock' | 'move', data: RelationshipJobData, opts: Bull.JobsOptions = {}): {
 		name: string,
 		data: RelationshipJobData,
 		opts: Bull.JobsOptions,

packages/backend/src/core/RelayService.ts

@@ -16,9 +16,15 @@ import { deepClone } from '@/misc/clone.js';
 import { bindThis } from '@/decorators.js';
 import { SystemAccountService } from '@/core/SystemAccountService.js';
 import { CacheManagementService, ManagedMemorySingleCache } from '@/global/CacheManagementService.js';
+import type { IActivity } from '@/core/activitypub/type.js';
+import { LoggerService } from '@/core/LoggerService.js';
+import type Logger from '@/logger.js';
+import { renderInlineError } from '@/misc/render-inline-error.js';
+import type { Signed } from '@/core/activitypub/JsonLdService.js';
 
 @Injectable()
 export class RelayService {
+	private readonly logger: Logger;
 	private readonly relaysCache: ManagedMemorySingleCache<MiRelay[]>;
 
 	constructor(
@@ -29,9 +35,11 @@ export class RelayService {
 		private queueService: QueueService,
 		private systemAccountService: SystemAccountService,
 		private apRendererService: ApRendererService,
+		private readonly loggerService: LoggerService,
 
 		cacheManagementService: CacheManagementService,
 	) {
+		this.logger = this.loggerService.getLogger('relay');
 		this.relaysCache = cacheManagementService.createMemorySingleCache<MiRelay[]>('relay', 1000 * 60 * 10); // 10m
 	}
 
@@ -106,10 +114,19 @@ export class RelayService {
 		const copy = deepClone(activity);
 		if (!copy.to) copy.to = ['https://www.w3.org/ns/activitystreams#Public'];
 
-		const signed = await this.apRendererService.attachLdSignature(copy, user);
+		const signed = await this.signActivity(copy, user);
 
 		for (const relay of relays) {
 			this.queueService.deliver(user, signed, relay.inbox, false);
 		}
 	}
+
+	private async signActivity<T extends IActivity>(activity: T, user: { id: MiUser['id']; host: null; }): Promise<T | Signed<T>> {
+		try {
+			return await this.apRendererService.attachLdSignature(activity, user);
+		} catch (err) {
+			this.logger.warn(`Error signing activity ${activity.id}: ${renderInlineError(err)}`);
+			return activity;
+		}
+	}
 }

packages/backend/src/core/WebhookTestService.ts

@@ -494,7 +494,7 @@ export class WebhookTestService {
 			url: null,
 			uri: null,
 			movedTo: null,
-			alsoKnownAs: [],
+			movedToUri: null,
 			createdAt: this.timeService.date.toISOString(),
 			updatedAt: user.updatedAt?.toISOString() ?? null,
 			lastFetchedAt: user.lastFetchedAt?.toISOString() ?? null,

packages/backend/src/core/activitypub/ApRendererService.ts

@@ -36,7 +36,7 @@ import { CacheService } from '@/core/CacheService.js';
 import { isPureRenote, isQuote, isRenote } from '@/misc/is-renote.js';
 import { FederatedInstanceService } from '@/core/FederatedInstanceService.js';
 import { TimeService } from '@/global/TimeService.js';
-import { JsonLdService } from './JsonLdService.js';
+import { JsonLdService, type Signed } from './JsonLdService.js';
 import { ApMfmService } from './ApMfmService.js';
 import { CONTEXT } from './misc/contexts.js';
 import { getApId, ILink, IOrderedCollection, IOrderedCollectionPage } from './type.js';
@@ -804,7 +804,7 @@ export class ApRendererService {
 	}
 
 	@bindThis
-	public async attachLdSignature(activity: any, user: { id: MiUser['id']; host: null; }): Promise<IActivity> {
+	public async attachLdSignature<T extends IActivity>(activity: T, user: { id: MiUser['id']; host: null; }): Promise<T | Signed<T>> {
 		// Linked Data signatures are cryptographic signatures attached to each activity to provide proof of authenticity.
 		// When using authorized fetch, this is often undesired as any signed activity can be forwarded to a blocked instance by relays and other instances.
 		// This setting allows admins to disable LD signatures for increased privacy, at the expense of fewer relayed activities and additional inbound fetch (GET) requests.

packages/backend/src/core/activitypub/misc/contexts.ts

@@ -523,6 +523,10 @@ const activitystreams = {
 			'@id': 'as:alsoKnownAs',
 			'@type': '@id',
 		},
+		'movedTo': {
+			'@id': 'as:movedTo',
+			'@type': '@id',
+		},
 	},
 } satisfies JsonLd;
 
@@ -584,6 +588,7 @@ const extension_context_definition = {
 	backgroundUrl: 'sharkey:backgroundUrl',
 	listenbrainz: 'sharkey:listenbrainz',
 	enableRss: 'sharkey:enableRss',
+	noindex: 'sharkey:noindex',
 	// vcard
 	vcard: 'http://www.w3.org/2006/vcard/ns#',
 } satisfies Context;

packages/backend/src/core/activitypub/models/ApNoteService.ts

@@ -32,6 +32,7 @@ import { renderInlineError } from '@/misc/render-inline-error.js';
 import { extractMediaFromHtml } from '@/core/activitypub/misc/extract-media-from-html.js';
 import { extractMediaFromMfm } from '@/core/activitypub/misc/extract-media-from-mfm.js';
 import { getContentByType } from '@/core/activitypub/misc/get-content-by-type.js';
+import { trackPromise } from '@/misc/promise-tracker.js';
 import { CustomEmojiService, encodeEmojiKey, isValidEmojiName } from '@/core/CustomEmojiService.js';
 import { TimeService } from '@/global/TimeService.js';
 import { getOneApId, getApId, validPost, isEmoji, getApType, isApObject, isDocument, IApDocument, isLink } from '../type.js';
@@ -300,7 +301,7 @@ export class ApNoteService implements OnModuleInit {
 					await this.pollService.vote(actor, reply, index);
 
 					// リモートフォロワーにUpdate配信
-					this.pollService.deliverQuestionUpdate(reply);
+					trackPromise(this.pollService.deliverQuestionUpdate(reply));
 				}
 				return null;
 			};
@@ -479,7 +480,7 @@ export class ApNoteService implements OnModuleInit {
 					await this.pollService.vote(actor, reply, index);
 
 					// リモートフォロワーにUpdate配信
-					this.pollService.deliverQuestionUpdate(reply);
+					trackPromise(this.pollService.deliverQuestionUpdate(reply));
 				}
 				return null;
 			};

packages/backend/src/core/activitypub/models/ApPersonService.ts

@@ -46,6 +46,7 @@ import { verifyFieldLinks } from '@/misc/verify-field-link.js';
 import { isRetryableError } from '@/misc/is-retryable-error.js';
 import { renderInlineError } from '@/misc/render-inline-error.js';
 import { IdentifiableError } from '@/misc/identifiable-error.js';
+import { QueueService } from '@/core/QueueService.js';
 import { getApId, getApType, isActor, isCollection, isCollectionOrOrderedCollection, isPropertyValue } from '../type.js';
 import { ApLoggerService } from '../ApLoggerService.js';
 import { extractApHashtags } from './tag.js';
@@ -121,6 +122,7 @@ export class ApPersonService implements OnModuleInit {
 		private readonly apUtilityService: ApUtilityService,
 		private readonly idService: IdService,
 		private readonly timeService: TimeService,
+		private readonly queueService: QueueService,
 
 		apLoggerService: ApLoggerService,
 	) {
@@ -988,7 +990,7 @@ export class ApPersonService implements OnModuleInit {
 			return 'skip: alsoKnownAs does not include from.uri';
 		}
 
-		await this.accountMoveService.postMoveProcess(src, dst);
+		await this.queueService.createMoveJob(src, dst);
 
 		return 'ok';
 	}

packages/backend/src/core/entities/UserEntityService.ts

@@ -409,6 +409,30 @@ export class UserEntityService implements OnModuleInit {
 		);
 	}
 
+	@bindThis
+	public async resolveAlsoKnownAs(user: MiUser): Promise<{ uri: string, id: string | null }[] | null> {
+		if (!user.alsoKnownAs) {
+			return null;
+		}
+
+		const alsoKnownAs: { uri: string, id: string | null }[] = [];
+		for (const uri of new Set(user.alsoKnownAs)) {
+			try {
+				const resolved = await this.apPersonService.resolvePerson(uri);
+				alsoKnownAs.push({ uri, id: resolved.id });
+			} catch {
+				// ignore errors - we expect some users to be deleted or unavailable
+				alsoKnownAs.push({ uri, id: null });
+			}
+		}
+
+		if (alsoKnownAs.length < 1) {
+			return null;
+		}
+
+		return alsoKnownAs;
+	}
+
 	@bindThis
 	public getIdenticonUrl(user: MiUser): string {
 		if ((user.host == null || user.host === this.config.host) && user.username.includes('.') && this.meta.iconUrl) { // ローカルのシステムアカウントの場合
@@ -551,6 +575,10 @@ export class UserEntityService implements OnModuleInit {
 		let fetchPoliciesPromise: Promise<RolePolicies> | null = null;
 		const fetchPolicies = () => fetchPoliciesPromise ??= this.roleService.getUserPolicies(user);
 
+		// This has a cache so it's fine to await here
+		const alsoKnownAs = await this.resolveAlsoKnownAs(user);
+		const alsoKnownAsIds = alsoKnownAs?.map(aka => aka.id).filter(id => id != null) ?? null;
+
 		const bypassSilence = isMe || (myFollowings ? myFollowings.has(user.id) : false);
 
 		const packed = {
@@ -617,10 +645,8 @@ export class UserEntityService implements OnModuleInit {
 				url: profile!.url,
 				uri: user.uri,
 				movedTo: user.movedToUri ? Promise.resolve(opts.userIdsByUri?.get(user.movedToUri) ?? this.apPersonService.resolvePerson(user.movedToUri).then(user => user.id).catch(() => null)) : null,
-				alsoKnownAs: user.alsoKnownAs
-					? Promise.all(user.alsoKnownAs.map(uri => Promise.resolve(opts.userIdsByUri?.get(uri) ?? this.apPersonService.fetchPerson(uri).then(user => user?.id).catch(() => null))))
-						.then(xs => xs.length === 0 ? null : xs.filter(x => x != null))
-					: null,
+				movedToUri: user.movedToUri,
+				// alsoKnownAs moved from packedUserDetailedNotMeOnly for privacy
 				bannerUrl: user.bannerId == null ? null : user.bannerUrl,
 				bannerBlurhash: user.bannerId == null ? null : user.bannerBlurhash,
 				backgroundUrl: user.backgroundId == null ? null : user.backgroundUrl,
@@ -711,6 +737,9 @@ export class UserEntityService implements OnModuleInit {
 				defaultCW: profile!.defaultCW,
 				defaultCWPriority: profile!.defaultCWPriority,
 				allowUnsignedFetch: user.allowUnsignedFetch,
+				// alsoKnownAs moved from packedUserDetailedNotMeOnly for privacy
+				alsoKnownAs: alsoKnownAsIds,
+				skAlsoKnownAs: alsoKnownAs,
 			} : {}),
 
 			...(opts.includeSecrets ? {

packages/backend/src/models/json-schema/user.ts

@@ -309,19 +309,16 @@ export const packedUserDetailedNotMeOnlySchema = {
 			nullable: true, optional: false,
 		},
 		movedTo: {
+			type: 'string',
+			format: 'id',
+			nullable: true, optional: false,
+		},
+		movedToUri: {
 			type: 'string',
 			format: 'uri',
 			nullable: true, optional: false,
 		},
-		alsoKnownAs: {
-			type: 'array',
-			nullable: true, optional: false,
-			items: {
-				type: 'string',
-				format: 'id',
-				nullable: false, optional: false,
-			},
-		},
+		// alsoKnownAs moved to packedUserDetailedNotMeOnly for privacy
 		bannerUrl: {
 			type: 'string',
 			format: 'url',
@@ -814,6 +811,36 @@ export const packedMeDetailedOnlySchema = {
 			enum: userUnsignedFetchOptions,
 			nullable: false, optional: false,
 		},
+		// alsoKnownAs moved from packedUserDetailedNotMeOnly for privacy
+		alsoKnownAs: {
+			type: 'array',
+			nullable: true, optional: false,
+			items: {
+				type: 'string',
+				format: 'id',
+				nullable: false, optional: false,
+			},
+		},
+		skAlsoKnownAs: {
+			type: 'array',
+			nullable: true, optional: false,
+			items: {
+				type: 'object',
+				nullable: false, optional: false,
+				properties: {
+					uri: {
+						type: 'string',
+						format: 'uri',
+						nullable: false, optional: false,
+					},
+					id: {
+						type: 'string',
+						format: 'id',
+						nullable: true, optional: false,
+					},
+				},
+			},
+		},
 	},
 } as const;
 

packages/backend/src/queue/QueueProcessorService.ts

@@ -429,6 +429,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
 					case 'unfollow': return this.relationshipProcessorService.processUnfollow(job);
 					case 'block': return this.relationshipProcessorService.processBlock(job);
 					case 'unblock': return this.relationshipProcessorService.processUnblock(job);
+					case 'move': return this.relationshipProcessorService.processMove(job);
 					default: throw new Error(`unrecognized job type ${job.name} for relationship`);
 				}
 			};

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

@@ -7,6 +7,8 @@ import { Inject, Injectable } from '@nestjs/common';
 
 import { UserFollowingService } from '@/core/UserFollowingService.js';
 import { UserBlockingService } from '@/core/UserBlockingService.js';
+import { CacheService } from '@/core/CacheService.js';
+import { AccountMoveService } from '@/core/AccountMoveService.js';
 import { bindThis } from '@/decorators.js';
 import type Logger from '@/logger.js';
 
@@ -28,8 +30,10 @@ export class RelationshipProcessorService {
 		private queueLoggerService: QueueLoggerService,
 		private userFollowingService: UserFollowingService,
 		private userBlockingService: UserBlockingService,
+		private readonly cacheService: CacheService,
+		private readonly accountMoveService: AccountMoveService,
 	) {
-		this.logger = this.queueLoggerService.logger.createSubLogger('follow-block');
+		this.logger = this.queueLoggerService.logger.createSubLogger('relationship');
 	}
 
 	@bindThis
@@ -47,8 +51,8 @@ export class RelationshipProcessorService {
 	public async processUnfollow(job: Bull.Job<RelationshipJobData>): Promise<string> {
 		this.logger.info(`${job.data.from.id} is trying to unfollow ${job.data.to.id}`);
 		const [follower, followee] = await Promise.all([
-			this.usersRepository.findOneByOrFail({ id: job.data.from.id }),
-			this.usersRepository.findOneByOrFail({ id: job.data.to.id }),
+			this.cacheService.findUserById(job.data.from.id),
+			this.cacheService.findUserById(job.data.to.id),
 		]) as [MiLocalUser | MiRemoteUser, MiLocalUser | MiRemoteUser];
 		await this.userFollowingService.unfollow(follower, followee, job.data.silent);
 		return 'ok';
@@ -58,8 +62,8 @@ export class RelationshipProcessorService {
 	public async processBlock(job: Bull.Job<RelationshipJobData>): Promise<string> {
 		this.logger.info(`${job.data.from.id} is trying to block ${job.data.to.id}`);
 		const [blockee, blocker] = await Promise.all([
-			this.usersRepository.findOneByOrFail({ id: job.data.from.id }),
-			this.usersRepository.findOneByOrFail({ id: job.data.to.id }),
+			this.cacheService.findUserById(job.data.from.id),
+			this.cacheService.findUserById(job.data.to.id),
 		]);
 		await this.userBlockingService.block(blockee, blocker, job.data.silent);
 		return 'ok';
@@ -69,10 +73,20 @@ export class RelationshipProcessorService {
 	public async processUnblock(job: Bull.Job<RelationshipJobData>): Promise<string> {
 		this.logger.info(`${job.data.from.id} is trying to unblock ${job.data.to.id}`);
 		const [blockee, blocker] = await Promise.all([
-			this.usersRepository.findOneByOrFail({ id: job.data.from.id }),
-			this.usersRepository.findOneByOrFail({ id: job.data.to.id }),
+			this.cacheService.findUserById(job.data.from.id),
+			this.cacheService.findUserById(job.data.to.id),
 		]);
 		await this.userBlockingService.unblock(blockee, blocker);
 		return 'ok';
 	}
+
+	public async processMove(job: Bull.Job<RelationshipJobData>): Promise<string> {
+		this.logger.info(`${job.data.from.id} is trying to migrate to ${job.data.to.id}`);
+		const [src, dst] = await Promise.all([
+			this.cacheService.findUserById(job.data.from.id),
+			this.cacheService.findUserById(job.data.to.id),
+		]);
+		await this.accountMoveService.postMoveProcess(src, dst);
+		return 'ok';
+	}
 }

packages/backend/src/server/api/endpoint-list.ts

@@ -87,6 +87,7 @@ export * as 'admin/relays/add' from './endpoints/admin/relays/add.js';
 export * as 'admin/relays/list' from './endpoints/admin/relays/list.js';
 export * as 'admin/relays/remove' from './endpoints/admin/relays/remove.js';
 export * as 'admin/reset-password' from './endpoints/admin/reset-password.js';
+export * as 'admin/restart-migration' from './endpoints/admin/restart-migration.js';
 export * as 'admin/resolve-abuse-user-report' from './endpoints/admin/resolve-abuse-user-report.js';
 export * as 'admin/roles/annotate-condition' from './endpoints/admin/roles/annotate-condition.js';
 export * as 'admin/roles/assign' from './endpoints/admin/roles/assign.js';

packages/backend/src/server/api/endpoints/admin/restart-migration.ts

@@ -0,0 +1,67 @@
+/*
+ * SPDX-FileCopyrightText: hazelnoot and other Sharkey contributors
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+import { Injectable } from '@nestjs/common';
+import { Endpoint } from '@/server/api/endpoint-base.js';
+import { CacheService } from '@/core/CacheService.js';
+import { ModerationLogService } from '@/core/ModerationLogService.js';
+import { AccountMoveService } from '@/core/AccountMoveService.js';
+import { ApiError } from '@/server/api/error.js';
+import { IdentifiableError } from '@/misc/identifiable-error.js';
+
+export const meta = {
+	tags: ['admin'],
+
+	requireCredential: true,
+	requireAdmin: true,
+	kind: 'write:admin:restart-migration',
+
+	errors: {
+		accountHasNotMigrated: {
+			message: 'Account has not migrated anywhere.',
+			code: 'ACCOUNT_HAS_NOT_MIGRATED',
+			id: 'ddcf173a-00f2-4aa4-ba12-cddd131bacf4',
+		},
+	},
+
+	res: {},
+} as const;
+
+export const paramDef = {
+	type: 'object',
+	properties: {
+		userId: { type: 'string', format: 'misskey:id' },
+	},
+	required: ['userId'],
+} as const;
+
+@Injectable()
+export default class extends Endpoint<typeof meta, typeof paramDef> { // eslint-disable-line import/no-default-export
+	constructor(
+		private readonly cacheService: CacheService,
+		private readonly moderationLogService: ModerationLogService,
+		private readonly accountMoveService: AccountMoveService,
+	) {
+		super(meta, paramDef, async (ps, me) => {
+			try {
+				const user = await this.cacheService.findUserById(ps.userId);
+				await this.accountMoveService.restartMigration(user);
+
+				await this.moderationLogService.log(me, 'restartMigration', {
+					userId: user.id,
+					userUsername: user.username,
+					userHost: user.host,
+				});
+			} catch (err) {
+				// TODO allow this mapping stuff to be defined in the meta
+				if (err instanceof IdentifiableError && err.id === 'ddcf173a-00f2-4aa4-ba12-cddd131bacf4') {
+					throw new ApiError(meta.errors.accountHasNotMigrated);
+				} else {
+					throw err;
+				}
+			}
+		});
+	}
+}

packages/backend/src/server/api/endpoints/admin/show-user.ts

@@ -13,6 +13,8 @@ import { IdService } from '@/core/IdService.js';
 import { notificationRecieveConfig } from '@/models/json-schema/user.js';
 import { isSystemAccount } from '@/misc/is-system-account.js';
 import { CacheService } from '@/core/CacheService.js';
+import { UserEntityService } from '@/core/entities/UserEntityService.js';
+import { ApPersonService } from '@/core/activitypub/models/ApPersonService.js';
 
 export const meta = {
 	tags: ['admin'],
@@ -225,6 +227,46 @@ export const meta = {
 				type: 'string',
 				optional: false, nullable: true,
 			},
+			movedAt: {
+				type: 'string',
+				optional: true, nullable: true,
+			},
+			movedTo: {
+				type: 'object',
+				optional: true, nullable: true,
+				properties: {
+					uri: {
+						type: 'string',
+						format: 'uri',
+						nullable: false, optional: false,
+					},
+					user: {
+						type: 'object',
+						ref: 'UserDetailed',
+						nullable: true, optional: true,
+					},
+				},
+			},
+			alsoKnownAs: {
+				type: 'array',
+				nullable: true, optional: true,
+				items: {
+					type: 'object',
+					nullable: false, optional: false,
+					properties: {
+						uri: {
+							type: 'string',
+							format: 'uri',
+							nullable: false, optional: false,
+						},
+						user: {
+							type: 'object',
+							ref: 'UserDetailed',
+							nullable: true, optional: true,
+						},
+					},
+				},
+			},
 		},
 	},
 } as const;
@@ -253,6 +295,8 @@ export default class extends Endpoint<typeof meta, typeof paramDef> { // eslint-
 		private roleEntityService: RoleEntityService,
 		private idService: IdService,
 		private readonly cacheService: CacheService,
+		private readonly apPersonService: ApPersonService,
+		private readonly userEntityService: UserEntityService,
 	) {
 		super(meta, paramDef, async (ps, me) => {
 			const [user, profile] = await Promise.all([
@@ -280,6 +324,22 @@ export default class extends Endpoint<typeof meta, typeof paramDef> { // eslint-
 
 			const followStats = await this.cacheService.getFollowStats(user.id);
 
+			const movedAt = user.movedAt?.toISOString();
+
+			const movedToUser = user.movedToUri ? await this.apPersonService.resolvePerson(user.movedToUri) : null;
+			const movedTo = user.movedToUri ? {
+				uri: user.movedToUri,
+				user: movedToUser ? await this.userEntityService.pack(movedToUser, me, { schema: 'UserDetailed' }) : undefined,
+			} : null;
+
+			// This is kinda heavy, but it's an admin endpoint so ok.
+			const aka = await this.userEntityService.resolveAlsoKnownAs(user);
+			const akaUsers = aka ? await this.userEntityService.packMany(aka.map(aka => aka.id).filter(id => id != null), me, { schema: 'UserDetailed' }) : [];
+			const alsoKnownAs = aka?.map(aka => ({
+				uri: aka.uri,
+				user: aka.id ? akaUsers.find(u => u.id === aka.id) : undefined,
+			}));
+
 			return {
 				email: profile.email,
 				emailVerified: profile.emailVerified,
@@ -318,6 +378,9 @@ export default class extends Endpoint<typeof meta, typeof paramDef> { // eslint-
 					totalFollowers: Math.max(user.followersCount, followStats.localFollowers + followStats.remoteFollowers),
 					totalFollowing: Math.max(user.followingCount, followStats.localFollowing + followStats.remoteFollowing),
 				},
+				movedAt,
+				movedTo,
+				alsoKnownAs,
 			};
 		});
 	}

packages/backend/src/server/api/endpoints/i/update.ts

@@ -35,6 +35,7 @@ import { AvatarDecorationService } from '@/core/AvatarDecorationService.js';
 import { notificationRecieveConfig } from '@/models/json-schema/user.js';
 import { userUnsignedFetchOptions } from '@/const.js';
 import { renderInlineError } from '@/misc/render-inline-error.js';
+import { trackPromise } from '@/misc/promise-tracker.js';
 import { ApiLoggerService } from '../../ApiLoggerService.js';
 import { ApiError } from '../../error.js';
 
@@ -640,12 +641,12 @@ export default class extends Endpoint<typeof meta, typeof paramDef> { // eslint-
 
 			// 鍵垢を解除したとき、溜まっていたフォローリクエストがあるならすべて承認
 			if (user.isLocked && ps.isLocked === false) {
-				this.userFollowingService.acceptAllFollowRequests(user);
+				await this.userFollowingService.acceptAllFollowRequests(user);
 			}
 
 			// フォロワーにUpdateを配信
 			if (this.userNeedsPublishing(user, updates) || this.profileNeedsPublishing(profile, updatedProfile)) {
-				this.accountUpdateService.publishToFollowers(user);
+				trackPromise(this.accountUpdateService.publishToFollowers(user));
 			}
 
 			return iObj;

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

@@ -172,11 +172,11 @@ export default class extends Endpoint<typeof meta, typeof paramDef> { // eslint-
 			if (note.userHost != null) {
 				const pollOwner = await this.usersRepository.findOneByOrFail({ id: note.userId }) as MiRemoteUser;
 
-				this.queueService.deliver(me, this.apRendererService.addContext(await this.apRendererService.renderVote(me, vote, note, poll, pollOwner)), pollOwner.inbox, false);
+				this.queueService.deliver(me, this.apRendererService.addContext(this.apRendererService.renderVote(me, vote, note, poll, pollOwner)), pollOwner.inbox, false);
 			}
 
 			// リモートフォロワーにUpdate配信
-			this.pollService.deliverQuestionUpdate(note);
+			await this.pollService.deliverQuestionUpdate(note);
 		});
 	}
 }

packages/backend/src/types.ts

@@ -108,6 +108,7 @@ export const moderationLogTypes = [
 	'deleteGlobalAnnouncement',
 	'deleteUserAnnouncement',
 	'resetPassword',
+	'restartMigration',
 	'setMandatoryCW',
 	'setMandatoryCWForNote',
 	'setMandatoryCWForInstance',
@@ -290,6 +291,11 @@ export type ModerationLogPayloads = {
 		userUsername: string;
 		userHost: string | null;
 	};
+	restartMigration: {
+		userId: string;
+		userUsername: string;
+		userHost: string | null;
+	};
 	setMandatoryCW: {
 		newCW: string | null;
 		oldCW: string | null;