summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorNamekuji <11836635+nmkj-io@users.noreply.github.com>2023-04-11 20:13:58 -0400
committerGitHub <noreply@github.com>2023-04-12 09:13:58 +0900
commitda833222005ddf712db2c40bcf3848874f7ee3f2 (patch)
treea8b45580a94a212fa5f05e056fcc59c9db7a7b73
parentUpdate CHANGELOG.md (#10591) (diff)
downloadsharkey-da833222005ddf712db2c40bcf3848874f7ee3f2.tar.gz
sharkey-da833222005ddf712db2c40bcf3848874f7ee3f2.tar.bz2
sharkey-da833222005ddf712db2c40bcf3848874f7ee3f2.zip
feat: queueing bulk follow/unfollow and block/unblock (#10544)
* wrap follow/unfollow and block/unblock as job queue * create import job to follow in each iteration * make relationship jobs concurrent * replace to job queue if called repeatedly * use addBulk to import * omit stream when importing * fix job caller * use ThinUser instead of User to reduce redis memory consumption * createImportFollowingToDbJobの呼び出し方を変える, 型補強 * Force ThinUser * オブジェクト操作のみのメソッド名はgenerate...Data * Force ThinUser in generateRelationshipJobData * silent bulk unfollow at admin api endpoint --------- Co-authored-by: tamaina <tamaina@hotmail.co.jp>
-rw-r--r--packages/backend/src/core/QueueModule.ts13
-rw-r--r--packages/backend/src/core/QueueService.ts113
-rw-r--r--packages/backend/src/core/UserBlockingService.ts16
-rw-r--r--packages/backend/src/core/UserFollowingService.ts9
-rw-r--r--packages/backend/src/core/UserListService.ts8
-rw-r--r--packages/backend/src/queue/DbQueueProcessorsService.ts2
-rw-r--r--packages/backend/src/queue/QueueProcessorModule.ts4
-rw-r--r--packages/backend/src/queue/QueueProcessorService.ts40
-rw-r--r--packages/backend/src/queue/RelationshipQueueProcessorsService.ts26
-rw-r--r--packages/backend/src/queue/processors/DeleteDriveFilesProcessorService.ts4
-rw-r--r--packages/backend/src/queue/processors/ExportBlockingProcessorService.ts4
-rw-r--r--packages/backend/src/queue/processors/ExportFavoritesProcessorService.ts4
-rw-r--r--packages/backend/src/queue/processors/ExportFollowingProcessorService.ts4
-rw-r--r--packages/backend/src/queue/processors/ExportMutingProcessorService.ts4
-rw-r--r--packages/backend/src/queue/processors/ExportNotesProcessorService.ts4
-rw-r--r--packages/backend/src/queue/processors/ExportUserListsProcessorService.ts4
-rw-r--r--packages/backend/src/queue/processors/ImportBlockingProcessorService.ts79
-rw-r--r--packages/backend/src/queue/processors/ImportFollowingProcessorService.ts74
-rw-r--r--packages/backend/src/queue/processors/ImportMutingProcessorService.ts6
-rw-r--r--packages/backend/src/queue/processors/RelationshipProcessorService.ts68
-rw-r--r--packages/backend/src/queue/types.ts39
-rw-r--r--packages/backend/src/server/api/endpoints/admin/federation/remove-all-following.ts8
-rw-r--r--packages/backend/src/server/api/endpoints/admin/suspend-user.ts35
23 files changed, 400 insertions, 168 deletions
diff --git a/packages/backend/src/core/QueueModule.ts b/packages/backend/src/core/QueueModule.ts
index 8733a7d7eb..bac85d7a15 100644
--- a/packages/backend/src/core/QueueModule.ts
+++ b/packages/backend/src/core/QueueModule.ts
@@ -3,7 +3,7 @@ import Bull from 'bull';
import { DI } from '@/di-symbols.js';
import type { Config } from '@/config.js';
import type { Provider } from '@nestjs/common';
-import type { DeliverJobData, InboxJobData, DbJobData, ObjectStorageJobData, EndedPollNotificationJobData, WebhookDeliverJobData } from '../queue/types.js';
+import type { DeliverJobData, InboxJobData, DbJobData, ObjectStorageJobData, EndedPollNotificationJobData, WebhookDeliverJobData, RelationshipJobData, DbJobMap } from '../queue/types.js';
function q<T>(config: Config, name: string, limitPerSec = -1) {
return new Bull<T>(name, {
@@ -41,7 +41,8 @@ export type SystemQueue = Bull.Queue<Record<string, unknown>>;
export type EndedPollNotificationQueue = Bull.Queue<EndedPollNotificationJobData>;
export type DeliverQueue = Bull.Queue<DeliverJobData>;
export type InboxQueue = Bull.Queue<InboxJobData>;
-export type DbQueue = Bull.Queue<DbJobData>;
+export type DbQueue = Bull.Queue<DbJobData<keyof DbJobMap>>;
+export type RelationshipQueue = Bull.Queue<RelationshipJobData>;
export type ObjectStorageQueue = Bull.Queue<ObjectStorageJobData>;
export type WebhookDeliverQueue = Bull.Queue<WebhookDeliverJobData>;
@@ -75,6 +76,12 @@ const $db: Provider = {
inject: [DI.config],
};
+const $relationship: Provider = {
+ provide: 'queue:relationship',
+ useFactory: (config: Config) => q(config, 'relationship'),
+ inject: [DI.config],
+};
+
const $objectStorage: Provider = {
provide: 'queue:objectStorage',
useFactory: (config: Config) => q(config, 'objectStorage'),
@@ -96,6 +103,7 @@ const $webhookDeliver: Provider = {
$deliver,
$inbox,
$db,
+ $relationship,
$objectStorage,
$webhookDeliver,
],
@@ -105,6 +113,7 @@ const $webhookDeliver: Provider = {
$deliver,
$inbox,
$db,
+ $relationship,
$objectStorage,
$webhookDeliver,
],
diff --git a/packages/backend/src/core/QueueService.ts b/packages/backend/src/core/QueueService.ts
index 498ceced7a..375ac49911 100644
--- a/packages/backend/src/core/QueueService.ts
+++ b/packages/backend/src/core/QueueService.ts
@@ -6,9 +6,10 @@ import type { Webhook, webhookEventTypes } from '@/models/entities/Webhook.js';
import type { Config } from '@/config.js';
import { DI } from '@/di-symbols.js';
import { bindThis } from '@/decorators.js';
-import type { DbQueue, DeliverQueue, EndedPollNotificationQueue, InboxQueue, ObjectStorageQueue, SystemQueue, WebhookDeliverQueue } from './QueueModule.js';
-import type { ThinUser } from '../queue/types.js';
+import type { DbQueue, DeliverQueue, EndedPollNotificationQueue, InboxQueue, ObjectStorageQueue, RelationshipQueue, SystemQueue, WebhookDeliverQueue } from './QueueModule.js';
+import type { DbJobData, RelationshipJobData, ThinUser } from '../queue/types.js';
import type httpSignature from '@peertube/http-signature';
+import Bull from 'bull';
@Injectable()
export class QueueService {
@@ -21,6 +22,7 @@ export class QueueService {
@Inject('queue:deliver') public deliverQueue: DeliverQueue,
@Inject('queue:inbox') public inboxQueue: InboxQueue,
@Inject('queue:db') public dbQueue: DbQueue,
+ @Inject('queue:relationship') public relationshipQueue: RelationshipQueue,
@Inject('queue:objectStorage') public objectStorageQueue: ObjectStorageQueue,
@Inject('queue:webhookDeliver') public webhookDeliverQueue: WebhookDeliverQueue,
) {}
@@ -56,7 +58,7 @@ export class QueueService {
activity: activity,
signature,
};
-
+
return this.inboxQueue.add(data, {
attempts: this.config.inboxJobMaxAttempts ?? 8,
timeout: 5 * 60 * 1000, // 5min
@@ -71,7 +73,7 @@ export class QueueService {
@bindThis
public createDeleteDriveFilesJob(user: ThinUser) {
return this.dbQueue.add('deleteDriveFiles', {
- user: user,
+ user: { id: user.id },
}, {
removeOnComplete: true,
removeOnFail: true,
@@ -81,7 +83,7 @@ export class QueueService {
@bindThis
public createExportCustomEmojisJob(user: ThinUser) {
return this.dbQueue.add('exportCustomEmojis', {
- user: user,
+ user: { id: user.id },
}, {
removeOnComplete: true,
removeOnFail: true,
@@ -91,7 +93,7 @@ export class QueueService {
@bindThis
public createExportNotesJob(user: ThinUser) {
return this.dbQueue.add('exportNotes', {
- user: user,
+ user: { id: user.id },
}, {
removeOnComplete: true,
removeOnFail: true,
@@ -101,7 +103,7 @@ export class QueueService {
@bindThis
public createExportFavoritesJob(user: ThinUser) {
return this.dbQueue.add('exportFavorites', {
- user: user,
+ user: { id: user.id },
}, {
removeOnComplete: true,
removeOnFail: true,
@@ -111,7 +113,7 @@ export class QueueService {
@bindThis
public createExportFollowingJob(user: ThinUser, excludeMuting = false, excludeInactive = false) {
return this.dbQueue.add('exportFollowing', {
- user: user,
+ user: { id: user.id },
excludeMuting,
excludeInactive,
}, {
@@ -123,7 +125,7 @@ export class QueueService {
@bindThis
public createExportMuteJob(user: ThinUser) {
return this.dbQueue.add('exportMuting', {
- user: user,
+ user: { id: user.id },
}, {
removeOnComplete: true,
removeOnFail: true,
@@ -133,7 +135,7 @@ export class QueueService {
@bindThis
public createExportBlockingJob(user: ThinUser) {
return this.dbQueue.add('exportBlocking', {
- user: user,
+ user: { id: user.id },
}, {
removeOnComplete: true,
removeOnFail: true,
@@ -143,7 +145,7 @@ export class QueueService {
@bindThis
public createExportUserListsJob(user: ThinUser) {
return this.dbQueue.add('exportUserLists', {
- user: user,
+ user: { id: user.id },
}, {
removeOnComplete: true,
removeOnFail: true,
@@ -153,7 +155,7 @@ export class QueueService {
@bindThis
public createImportFollowingJob(user: ThinUser, fileId: DriveFile['id']) {
return this.dbQueue.add('importFollowing', {
- user: user,
+ user: { id: user.id },
fileId: fileId,
}, {
removeOnComplete: true,
@@ -162,9 +164,15 @@ export class QueueService {
}
@bindThis
+ public createImportFollowingToDbJob(user: ThinUser, targets: string[]) {
+ const jobs = targets.map(rel => this.generateToDbJobData('importFollowingToDb', { user, target: rel }));
+ return this.dbQueue.addBulk(jobs);
+ }
+
+ @bindThis
public createImportMutingJob(user: ThinUser, fileId: DriveFile['id']) {
return this.dbQueue.add('importMuting', {
- user: user,
+ user: { id: user.id },
fileId: fileId,
}, {
removeOnComplete: true,
@@ -175,7 +183,7 @@ export class QueueService {
@bindThis
public createImportBlockingJob(user: ThinUser, fileId: DriveFile['id']) {
return this.dbQueue.add('importBlocking', {
- user: user,
+ user: { id: user.id },
fileId: fileId,
}, {
removeOnComplete: true,
@@ -184,9 +192,31 @@ export class QueueService {
}
@bindThis
+ public createImportBlockingToDbJob(user: ThinUser, targets: string[]) {
+ const jobs = targets.map(rel => this.generateToDbJobData('importBlockingToDb', { user, target: rel }));
+ return this.dbQueue.addBulk(jobs);
+ }
+
+ @bindThis
+ private generateToDbJobData<T extends 'importFollowingToDb' | 'importBlockingToDb', D extends DbJobData<T>>(name: T, data: D): {
+ name: string,
+ data: D,
+ opts: Bull.JobOptions,
+ } {
+ return {
+ name,
+ data,
+ opts: {
+ removeOnComplete: true,
+ removeOnFail: true,
+ },
+ };
+ }
+
+ @bindThis
public createImportUserListsJob(user: ThinUser, fileId: DriveFile['id']) {
return this.dbQueue.add('importUserLists', {
- user: user,
+ user: { id: user.id },
fileId: fileId,
}, {
removeOnComplete: true,
@@ -197,7 +227,7 @@ export class QueueService {
@bindThis
public createImportCustomEmojisJob(user: ThinUser, fileId: DriveFile['id']) {
return this.dbQueue.add('importCustomEmojis', {
- user: user,
+ user: { id: user.id },
fileId: fileId,
}, {
removeOnComplete: true,
@@ -208,7 +238,7 @@ export class QueueService {
@bindThis
public createDeleteAccountJob(user: ThinUser, opts: { soft?: boolean; } = {}) {
return this.dbQueue.add('deleteAccount', {
- user: user,
+ user: { id: user.id },
soft: opts.soft,
}, {
removeOnComplete: true,
@@ -217,6 +247,51 @@ export class QueueService {
}
@bindThis
+ public createFollowJob(followings: { from: ThinUser, to: ThinUser, requestId?: string, silent?: boolean }[]) {
+ const jobs = followings.map(rel => this.generateRelationshipJobData('follow', rel));
+ return this.relationshipQueue.addBulk(jobs);
+ }
+
+ @bindThis
+ public createUnfollowJob(followings: { from: ThinUser, to: ThinUser, requestId?: string }[]) {
+ const jobs = followings.map(rel => this.generateRelationshipJobData('unfollow', rel));
+ return this.relationshipQueue.addBulk(jobs);
+ }
+
+ @bindThis
+ public createBlockJob(blockings: { from: ThinUser, to: ThinUser, silent?: boolean }[]) {
+ const jobs = blockings.map(rel => this.generateRelationshipJobData('block', rel));
+ return this.relationshipQueue.addBulk(jobs);
+ }
+
+ @bindThis
+ public createUnblockJob(blockings: { from: ThinUser, to: ThinUser, silent?: boolean }[]) {
+ const jobs = blockings.map(rel => this.generateRelationshipJobData('unblock', rel));
+ return this.relationshipQueue.addBulk(jobs);
+ }
+
+ @bindThis
+ private generateRelationshipJobData(name: 'follow' | 'unfollow' | 'block' | 'unblock', data: RelationshipJobData): {
+ name: string,
+ data: RelationshipJobData,
+ opts: Bull.JobOptions,
+ } {
+ return {
+ name,
+ data: {
+ from: { id: data.from.id },
+ to: { id: data.to.id },
+ silent: data.silent,
+ requestId: data.requestId,
+ },
+ opts: {
+ removeOnComplete: true,
+ removeOnFail: true,
+ },
+ };
+ }
+
+ @bindThis
public createDeleteObjectStorageFileJob(key: string) {
return this.objectStorageQueue.add('deleteFile', {
key: key,
@@ -246,7 +321,7 @@ export class QueueService {
createdAt: Date.now(),
eventId: uuid(),
};
-
+
return this.webhookDeliverQueue.add(data, {
attempts: 4,
timeout: 1 * 60 * 1000, // 1min
@@ -264,7 +339,7 @@ export class QueueService {
//deliverLogger.succ(`Cleaned ${jobs.length} ${status} jobs`);
});
this.deliverQueue.clean(0, 'delayed');
-
+
this.inboxQueue.once('cleaned', (jobs, status) => {
//inboxLogger.succ(`Cleaned ${jobs.length} ${status} jobs`);
});
diff --git a/packages/backend/src/core/UserBlockingService.ts b/packages/backend/src/core/UserBlockingService.ts
index b3e306346e..3ca22f8bbc 100644
--- a/packages/backend/src/core/UserBlockingService.ts
+++ b/packages/backend/src/core/UserBlockingService.ts
@@ -24,7 +24,7 @@ export class UserBlockingService implements OnModuleInit {
constructor(
private moduleRef: ModuleRef,
-
+
@Inject(DI.followRequestsRepository)
private followRequestsRepository: FollowRequestsRepository,
@@ -54,12 +54,12 @@ export class UserBlockingService implements OnModuleInit {
}
@bindThis
- public async block(blocker: User, blockee: User) {
+ public async block(blocker: User, blockee: User, silent = false) {
await Promise.all([
- this.cancelRequest(blocker, blockee),
- this.cancelRequest(blockee, blocker),
- this.userFollowingService.unfollow(blocker, blockee),
- this.userFollowingService.unfollow(blockee, blocker),
+ this.cancelRequest(blocker, blockee, silent),
+ this.cancelRequest(blockee, blocker, silent),
+ this.userFollowingService.unfollow(blocker, blockee, silent),
+ this.userFollowingService.unfollow(blockee, blocker, silent),
this.removeFromList(blockee, blocker),
]);
@@ -89,7 +89,7 @@ export class UserBlockingService implements OnModuleInit {
}
@bindThis
- private async cancelRequest(follower: User, followee: User) {
+ private async cancelRequest(follower: User, followee: User, silent = false) {
const request = await this.followRequestsRepository.findOneBy({
followeeId: followee.id,
followerId: follower.id,
@@ -110,7 +110,7 @@ export class UserBlockingService implements OnModuleInit {
}).then(packed => this.globalEventService.publishMainStream(followee.id, 'meUpdated', packed));
}
- if (this.userEntityService.isLocalUser(follower)) {
+ if (this.userEntityService.isLocalUser(follower) && !silent) {
this.userEntityService.pack(followee, follower, {
detail: true,
}).then(async packed => {
diff --git a/packages/backend/src/core/UserFollowingService.ts b/packages/backend/src/core/UserFollowingService.ts
index d7bb8f3920..dacaa7263a 100644
--- a/packages/backend/src/core/UserFollowingService.ts
+++ b/packages/backend/src/core/UserFollowingService.ts
@@ -43,7 +43,7 @@ export class UserFollowingService implements OnModuleInit {
constructor(
private moduleRef: ModuleRef,
-
+
@Inject(DI.usersRepository)
private usersRepository: UsersRepository,
@@ -79,7 +79,7 @@ export class UserFollowingService implements OnModuleInit {
}
@bindThis
- public async follow(_follower: { id: User['id'] }, _followee: { id: User['id'] }, requestId?: string): Promise<void> {
+ public async follow(_follower: { id: User['id'] }, _followee: { id: User['id'] }, requestId?: string, silent = false): Promise<void> {
const [follower, followee] = await Promise.all([
this.usersRepository.findOneByOrFail({ id: _follower.id }),
this.usersRepository.findOneByOrFail({ id: _followee.id }),
@@ -139,7 +139,7 @@ export class UserFollowingService implements OnModuleInit {
}
}
- await this.insertFollowingDoc(followee, follower);
+ await this.insertFollowingDoc(followee, follower, silent);
if (this.userEntityService.isRemoteUser(follower) && this.userEntityService.isLocalUser(followee)) {
const content = this.apRendererService.addContext(this.apRendererService.renderAccept(this.apRendererService.renderFollow(follower, followee, requestId), followee));
@@ -155,6 +155,7 @@ export class UserFollowingService implements OnModuleInit {
follower: {
id: User['id']; host: User['host']; uri: User['host']; inbox: User['inbox']; sharedInbox: User['sharedInbox']
},
+ silent = false,
): Promise<void> {
if (follower.id === followee.id) return;
@@ -233,7 +234,7 @@ export class UserFollowingService implements OnModuleInit {
this.perUserFollowingChart.update(follower, followee, true);
// Publish follow event
- if (this.userEntityService.isLocalUser(follower)) {
+ if (this.userEntityService.isLocalUser(follower) && !silent) {
this.userEntityService.pack(followee.id, follower, {
detail: true,
}).then(async packed => {
diff --git a/packages/backend/src/core/UserListService.ts b/packages/backend/src/core/UserListService.ts
index bc726a1feb..08cc907ebf 100644
--- a/packages/backend/src/core/UserListService.ts
+++ b/packages/backend/src/core/UserListService.ts
@@ -11,6 +11,7 @@ import { UserEntityService } from '@/core/entities/UserEntityService.js';
import { ProxyAccountService } from '@/core/ProxyAccountService.js';
import { bindThis } from '@/decorators.js';
import { RoleService } from '@/core/RoleService.js';
+import { QueueService } from '@/core/QueueService.js';
@Injectable()
export class UserListService {
@@ -29,6 +30,7 @@ export class UserListService {
private roleService: RoleService,
private globalEventService: GlobalEventService,
private proxyAccountService: ProxyAccountService,
+ private queueService: QueueService,
) {
}
@@ -47,14 +49,14 @@ export class UserListService {
userId: target.id,
userListId: list.id,
} as UserListJoining);
-
+
this.globalEventService.publishUserListStream(list.id, 'userAdded', await this.userEntityService.pack(target));
-
+
// このインスタンス内にこのリモートユーザーをフォローしているユーザーがいなくても投稿を受け取るためにダミーのユーザーがフォローしたということにする
if (this.userEntityService.isRemoteUser(target)) {
const proxy = await this.proxyAccountService.fetch();
if (proxy) {
- this.userFollowingService.follow(proxy, target);
+ this.queueService.createFollowJob([{ from: { id: proxy.id }, to: { id: target.id } }]);
}
}
}
diff --git a/packages/backend/src/queue/DbQueueProcessorsService.ts b/packages/backend/src/queue/DbQueueProcessorsService.ts
index 910e2bea45..233a36dd04 100644
--- a/packages/backend/src/queue/DbQueueProcessorsService.ts
+++ b/packages/backend/src/queue/DbQueueProcessorsService.ts
@@ -52,8 +52,10 @@ export class DbQueueProcessorsService {
q.process('exportBlocking', (job, done) => this.exportBlockingProcessorService.process(job, done));
q.process('exportUserLists', (job, done) => this.exportUserListsProcessorService.process(job, done));
q.process('importFollowing', (job, done) => this.importFollowingProcessorService.process(job, done));
+ q.process('importFollowingToDb', (job) => this.importFollowingProcessorService.processDb(job));
q.process('importMuting', (job, done) => this.importMutingProcessorService.process(job, done));
q.process('importBlocking', (job, done) => this.importBlockingProcessorService.process(job, done));
+ q.process('importBlockingToDb', (job) => this.importBlockingProcessorService.processDb(job));
q.process('importUserLists', (job, done) => this.importUserListsProcessorService.process(job, done));
q.process('importCustomEmojis', (job, done) => this.importCustomEmojisProcessorService.process(job, done));
q.process('deleteAccount', (job) => this.deleteAccountProcessorService.process(job));
diff --git a/packages/backend/src/queue/QueueProcessorModule.ts b/packages/backend/src/queue/QueueProcessorModule.ts
index 6a8f35cdda..4db9b38547 100644
--- a/packages/backend/src/queue/QueueProcessorModule.ts
+++ b/packages/backend/src/queue/QueueProcessorModule.ts
@@ -4,6 +4,7 @@ import { GlobalModule } from '@/GlobalModule.js';
import { QueueLoggerService } from './QueueLoggerService.js';
import { QueueProcessorService } from './QueueProcessorService.js';
import { DbQueueProcessorsService } from './DbQueueProcessorsService.js';
+import { RelationshipQueueProcessorsService } from './RelationshipQueueProcessorsService.js';
import { ObjectStorageQueueProcessorsService } from './ObjectStorageQueueProcessorsService.js';
import { DeliverProcessorService } from './processors/DeliverProcessorService.js';
import { EndedPollNotificationProcessorService } from './processors/EndedPollNotificationProcessorService.js';
@@ -32,6 +33,7 @@ import { ResyncChartsProcessorService } from './processors/ResyncChartsProcessor
import { TickChartsProcessorService } from './processors/TickChartsProcessorService.js';
import { AggregateRetentionProcessorService } from './processors/AggregateRetentionProcessorService.js';
import { ExportFavoritesProcessorService } from './processors/ExportFavoritesProcessorService.js';
+import { RelationshipProcessorService } from './processors/RelationshipProcessorService.js';
@Module({
imports: [
@@ -61,9 +63,11 @@ import { ExportFavoritesProcessorService } from './processors/ExportFavoritesPro
DeleteAccountProcessorService,
DeleteFileProcessorService,
CleanRemoteFilesProcessorService,
+ RelationshipProcessorService,
SystemQueueProcessorsService,
ObjectStorageQueueProcessorsService,
DbQueueProcessorsService,
+ RelationshipQueueProcessorsService,
WebhookDeliverProcessorService,
EndedPollNotificationProcessorService,
DeliverProcessorService,
diff --git a/packages/backend/src/queue/QueueProcessorService.ts b/packages/backend/src/queue/QueueProcessorService.ts
index 8457747cb0..706110f6fc 100644
--- a/packages/backend/src/queue/QueueProcessorService.ts
+++ b/packages/backend/src/queue/QueueProcessorService.ts
@@ -13,6 +13,7 @@ import { EndedPollNotificationProcessorService } from './processors/EndedPollNot
import { DeliverProcessorService } from './processors/DeliverProcessorService.js';
import { InboxProcessorService } from './processors/InboxProcessorService.js';
import { QueueLoggerService } from './QueueLoggerService.js';
+import { RelationshipQueueProcessorsService } from './RelationshipQueueProcessorsService.js';
@Injectable()
export class QueueProcessorService {
@@ -27,6 +28,7 @@ export class QueueProcessorService {
private systemQueueProcessorsService: SystemQueueProcessorsService,
private objectStorageQueueProcessorsService: ObjectStorageQueueProcessorsService,
private dbQueueProcessorsService: DbQueueProcessorsService,
+ private relationshipQueueProcessorsService: RelationshipQueueProcessorsService,
private webhookDeliverProcessorService: WebhookDeliverProcessorService,
private endedPollNotificationProcessorService: EndedPollNotificationProcessorService,
private deliverProcessorService: DeliverProcessorService,
@@ -52,14 +54,15 @@ export class QueueProcessorService {
};
}
}
-
+
const systemLogger = this.logger.createSubLogger('system');
const deliverLogger = this.logger.createSubLogger('deliver');
const webhookLogger = this.logger.createSubLogger('webhook');
const inboxLogger = this.logger.createSubLogger('inbox');
const dbLogger = this.logger.createSubLogger('db');
+ const relationshipLogger = this.logger.createSubLogger('relationship');
const objectStorageLogger = this.logger.createSubLogger('objectStorage');
-
+
this.queueService.systemQueue
.on('waiting', (jobId) => systemLogger.debug(`waiting id=${jobId}`))
.on('active', (job) => systemLogger.debug(`active id=${job.id}`))
@@ -67,7 +70,7 @@ export class QueueProcessorService {
.on('failed', (job, err) => systemLogger.warn(`failed(${err}) id=${job.id}`, { job, e: renderError(err) }))
.on('error', (job: any, err: Error) => systemLogger.error(`error ${err}`, { job, e: renderError(err) }))
.on('stalled', (job) => systemLogger.warn(`stalled id=${job.id}`));
-
+
this.queueService.deliverQueue
.on('waiting', (jobId) => deliverLogger.debug(`waiting id=${jobId}`))
.on('active', (job) => deliverLogger.debug(`active ${getJobInfo(job, true)} to=${job.data.to}`))
@@ -75,7 +78,7 @@ export class QueueProcessorService {
.on('failed', (job, err) => deliverLogger.warn(`failed(${err}) ${getJobInfo(job)} to=${job.data.to}`))
.on('error', (job: any, err: Error) => deliverLogger.error(`error ${err}`, { job, e: renderError(err) }))
.on('stalled', (job) => deliverLogger.warn(`stalled ${getJobInfo(job)} to=${job.data.to}`));
-
+
this.queueService.inboxQueue
.on('waiting', (jobId) => inboxLogger.debug(`waiting id=${jobId}`))
.on('active', (job) => inboxLogger.debug(`active ${getJobInfo(job, true)}`))
@@ -83,7 +86,7 @@ export class QueueProcessorService {
.on('failed', (job, err) => inboxLogger.warn(`failed(${err}) ${getJobInfo(job)} activity=${job.data.activity ? job.data.activity.id : 'none'}`, { job, e: renderError(err) }))
.on('error', (job: any, err: Error) => inboxLogger.error(`error ${err}`, { job, e: renderError(err) }))
.on('stalled', (job) => inboxLogger.warn(`stalled ${getJobInfo(job)} activity=${job.data.activity ? job.data.activity.id : 'none'}`));
-
+
this.queueService.dbQueue
.on('waiting', (jobId) => dbLogger.debug(`waiting id=${jobId}`))
.on('active', (job) => dbLogger.debug(`active id=${job.id}`))
@@ -91,7 +94,15 @@ export class QueueProcessorService {
.on('failed', (job, err) => dbLogger.warn(`failed(${err}) id=${job.id}`, { job, e: renderError(err) }))
.on('error', (job: any, err: Error) => dbLogger.error(`error ${err}`, { job, e: renderError(err) }))
.on('stalled', (job) => dbLogger.warn(`stalled id=${job.id}`));
-
+
+ this.queueService.relationshipQueue
+ .on('waiting', (jobId) => relationshipLogger.debug(`waiting id=${jobId}`))
+ .on('active', (job) => relationshipLogger.debug(`active id=${job.id}`))
+ .on('completed', (job, result) => relationshipLogger.debug(`completed(${result}) id=${job.id}`))
+ .on('failed', (job, err) => relationshipLogger.warn(`failed(${err}) id=${job.id}`, { job, e: renderError(err) }))
+ .on('error', (job: any, err: Error) => relationshipLogger.error(`error ${err}`, { job, e: renderError(err) }))
+ .on('stalled', (job) => relationshipLogger.warn(`stalled id=${job.id}`));
+
this.queueService.objectStorageQueue
.on('waiting', (jobId) => objectStorageLogger.debug(`waiting id=${jobId}`))
.on('active', (job) => objectStorageLogger.debug(`active id=${job.id}`))
@@ -99,7 +110,7 @@ export class QueueProcessorService {
.on('failed', (job, err) => objectStorageLogger.warn(`failed(${err}) id=${job.id}`, { job, e: renderError(err) }))
.on('error', (job: any, err: Error) => objectStorageLogger.error(`error ${err}`, { job, e: renderError(err) }))
.on('stalled', (job) => objectStorageLogger.warn(`stalled id=${job.id}`));
-
+
this.queueService.webhookDeliverQueue
.on('waiting', (jobId) => webhookLogger.debug(`waiting id=${jobId}`))
.on('active', (job) => webhookLogger.debug(`active ${getJobInfo(job, true)} to=${job.data.to}`))
@@ -107,26 +118,27 @@ export class QueueProcessorService {
.on('failed', (job, err) => webhookLogger.warn(`failed(${err}) ${getJobInfo(job)} to=${job.data.to}`))
.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}`));
-
+
this.queueService.deliverQueue.process(this.config.deliverJobConcurrency ?? 128, (job) => this.deliverProcessorService.process(job));
this.queueService.inboxQueue.process(this.config.inboxJobConcurrency ?? 16, (job) => this.inboxProcessorService.process(job));
this.queueService.endedPollNotificationQueue.process((job, done) => this.endedPollNotificationProcessorService.process(job, done));
this.queueService.webhookDeliverQueue.process(64, (job) => this.webhookDeliverProcessorService.process(job));
this.dbQueueProcessorsService.start(this.queueService.dbQueue);
+ this.relationshipQueueProcessorsService.start(this.queueService.relationshipQueue);
this.objectStorageQueueProcessorsService.start(this.queueService.objectStorageQueue);
-
+
this.queueService.systemQueue.add('tickCharts', {
}, {
repeat: { cron: '55 * * * *' },
removeOnComplete: true,
});
-
+
this.queueService.systemQueue.add('resyncCharts', {
}, {
repeat: { cron: '0 0 * * *' },
removeOnComplete: true,
});
-
+
this.queueService.systemQueue.add('cleanCharts', {
}, {
repeat: { cron: '0 0 * * *' },
@@ -138,19 +150,19 @@ export class QueueProcessorService {
repeat: { cron: '0 0 * * *' },
removeOnComplete: true,
});
-
+
this.queueService.systemQueue.add('clean', {
}, {
repeat: { cron: '0 0 * * *' },
removeOnComplete: true,
});
-
+
this.queueService.systemQueue.add('checkExpiredMutings', {
}, {
repeat: { cron: '*/5 * * * *' },
removeOnComplete: true,
});
-
+
this.systemQueueProcessorsService.start(this.queueService.systemQueue);
}
}
diff --git a/packages/backend/src/queue/RelationshipQueueProcessorsService.ts b/packages/backend/src/queue/RelationshipQueueProcessorsService.ts
new file mode 100644
index 0000000000..af086fa4e7
--- /dev/null
+++ b/packages/backend/src/queue/RelationshipQueueProcessorsService.ts
@@ -0,0 +1,26 @@
+import { Inject, Injectable } from '@nestjs/common';
+import { bindThis } from '@/decorators.js';
+import { RelationshipProcessorService } from './processors/RelationshipProcessorService.js';
+import type Bull from 'bull';
+import { DI } from '@/di-symbols.js';
+import type { Config } from '@/config.js';
+
+@Injectable()
+export class RelationshipQueueProcessorsService {
+ constructor(
+ @Inject(DI.config)
+ private config: Config,
+
+ private relationshipProcessorService: RelationshipProcessorService,
+ ) {
+ }
+
+ @bindThis
+ public start(q: Bull.Queue): void {
+ const maxJobs = (this.config.deliverJobConcurrency ?? 128) / 4; // conservative?
+ q.process('follow', maxJobs, (job) => this.relationshipProcessorService.processFollow(job));
+ q.process('unfollow', maxJobs, (job) => this.relationshipProcessorService.processUnfollow(job));
+ q.process('block', maxJobs, (job) => this.relationshipProcessorService.processBlock(job));
+ q.process('unblock', maxJobs, (job) => this.relationshipProcessorService.processUnblock(job));
+ }
+}
diff --git a/packages/backend/src/queue/processors/DeleteDriveFilesProcessorService.ts b/packages/backend/src/queue/processors/DeleteDriveFilesProcessorService.ts
index fa0c1733f6..604497cf54 100644
--- a/packages/backend/src/queue/processors/DeleteDriveFilesProcessorService.ts
+++ b/packages/backend/src/queue/processors/DeleteDriveFilesProcessorService.ts
@@ -7,7 +7,7 @@ import type Logger from '@/logger.js';
import { DriveService } from '@/core/DriveService.js';
import { QueueLoggerService } from '../QueueLoggerService.js';
import type Bull from 'bull';
-import type { DbUserJobData } from '../types.js';
+import type { DbJobDataWithUser } from '../types.js';
import { bindThis } from '@/decorators.js';
@Injectable()
@@ -31,7 +31,7 @@ export class DeleteDriveFilesProcessorService {
}
@bindThis
- public async process(job: Bull.Job<DbUserJobData>, done: () => void): Promise<void> {
+ public async process(job: Bull.Job<DbJobDataWithUser>, done: () => void): Promise<void> {
this.logger.info(`Deleting drive files of ${job.data.user.id} ...`);
const user = await this.usersRepository.findOneBy({ id: job.data.user.id });
diff --git a/packages/backend/src/queue/processors/ExportBlockingProcessorService.ts b/packages/backend/src/queue/processors/ExportBlockingProcessorService.ts
index 7f2c2d08b5..a020006732 100644
--- a/packages/backend/src/queue/processors/ExportBlockingProcessorService.ts
+++ b/packages/backend/src/queue/processors/ExportBlockingProcessorService.ts
@@ -11,7 +11,7 @@ import { createTemp } from '@/misc/create-temp.js';
import { UtilityService } from '@/core/UtilityService.js';
import { QueueLoggerService } from '../QueueLoggerService.js';
import type Bull from 'bull';
-import type { DbUserJobData } from '../types.js';
+import type { DbJobDataWithUser } from '../types.js';
import { bindThis } from '@/decorators.js';
@Injectable()
@@ -36,7 +36,7 @@ export class ExportBlockingProcessorService {
}
@bindThis
- public async process(job: Bull.Job<DbUserJobData>, done: () => void): Promise<void> {
+ public async process(job: Bull.Job<DbJobDataWithUser>, done: () => void): Promise<void> {
this.logger.info(`Exporting blocking of ${job.data.user.id} ...`);
const user = await this.usersRepository.findOneBy({ id: job.data.user.id });
diff --git a/packages/backend/src/queue/processors/ExportFavoritesProcessorService.ts b/packages/backend/src/queue/processors/ExportFavoritesProcessorService.ts
index e9330772b9..daefcdf2f5 100644
--- a/packages/backend/src/queue/processors/ExportFavoritesProcessorService.ts
+++ b/packages/backend/src/queue/processors/ExportFavoritesProcessorService.ts
@@ -13,7 +13,7 @@ import type { Note } from '@/models/entities/Note.js';
import { bindThis } from '@/decorators.js';
import { QueueLoggerService } from '../QueueLoggerService.js';
import type Bull from 'bull';
-import type { DbUserJobData } from '../types.js';
+import type { DbJobDataWithUser } from '../types.js';
@Injectable()
export class ExportFavoritesProcessorService {
@@ -42,7 +42,7 @@ export class ExportFavoritesProcessorService {
}
@bindThis
- public async process(job: Bull.Job<DbUserJobData>, done: () => void): Promise<void> {
+ public async process(job: Bull.Job<DbJobDataWithUser>, done: () => void): Promise<void> {
this.logger.info(`Exporting favorites of ${job.data.user.id} ...`);
const user = await this.usersRepository.findOneBy({ id: job.data.user.id });
diff --git a/packages/backend/src/queue/processors/ExportFollowingProcessorService.ts b/packages/backend/src/queue/processors/ExportFollowingProcessorService.ts
index 064b126e44..59443de57f 100644
--- a/packages/backend/src/queue/processors/ExportFollowingProcessorService.ts
+++ b/packages/backend/src/queue/processors/ExportFollowingProcessorService.ts
@@ -12,7 +12,7 @@ import type { Following } from '@/models/entities/Following.js';
import { UtilityService } from '@/core/UtilityService.js';
import { QueueLoggerService } from '../QueueLoggerService.js';
import type Bull from 'bull';
-import type { DbUserJobData } from '../types.js';
+import type { DbExportFollowingData } from '../types.js';
import { bindThis } from '@/decorators.js';
@Injectable()
@@ -40,7 +40,7 @@ export class ExportFollowingProcessorService {
}
@bindThis
- public async process(job: Bull.Job<DbUserJobData>, done: () => void): Promise<void> {
+ public async process(job: Bull.Job<DbExportFollowingData>, done: () => void): Promise<void> {
this.logger.info(`Exporting following of ${job.data.user.id} ...`);
const user = await this.usersRepository.findOneBy({ id: job.data.user.id });
diff --git a/packages/backend/src/queue/processors/ExportMutingProcessorService.ts b/packages/backend/src/queue/processors/ExportMutingProcessorService.ts
index 94c7ea8a46..a2a718b892 100644
--- a/packages/backend/src/queue/processors/ExportMutingProcessorService.ts
+++ b/packages/backend/src/queue/processors/ExportMutingProcessorService.ts
@@ -11,7 +11,7 @@ import { createTemp } from '@/misc/create-temp.js';
import { UtilityService } from '@/core/UtilityService.js';
import { QueueLoggerService } from '../QueueLoggerService.js';
import type Bull from 'bull';
-import type { DbUserJobData } from '../types.js';
+import type { DbJobDataWithUser } from '../types.js';
import { bindThis } from '@/decorators.js';
@Injectable()
@@ -39,7 +39,7 @@ export class ExportMutingProcessorService {
}
@bindThis
- public async process(job: Bull.Job<DbUserJobData>, done: () => void): Promise<void> {
+ public async process(job: Bull.Job<DbJobDataWithUser>, done: () => void): Promise<void> {
this.logger.info(`Exporting muting of ${job.data.user.id} ...`);
const user = await this.usersRepository.findOneBy({ id: job.data.user.id });
diff --git a/packages/backend/src/queue/processors/ExportNotesProcessorService.ts b/packages/backend/src/queue/processors/ExportNotesProcessorService.ts
index 2f74dd63cc..1aa20d6f1d 100644
--- a/packages/backend/src/queue/processors/ExportNotesProcessorService.ts
+++ b/packages/backend/src/queue/processors/ExportNotesProcessorService.ts
@@ -13,7 +13,7 @@ import type { Note } from '@/models/entities/Note.js';
import { bindThis } from '@/decorators.js';
import { QueueLoggerService } from '../QueueLoggerService.js';
import type Bull from 'bull';
-import type { DbUserJobData } from '../types.js';
+import type { DbJobDataWithUser } from '../types.js';
@Injectable()
export class ExportNotesProcessorService {
@@ -39,7 +39,7 @@ export class ExportNotesProcessorService {
}
@bindThis
- public async process(job: Bull.Job<DbUserJobData>, done: () => void): Promise<void> {
+ public async process(job: Bull.Job<DbJobDataWithUser>, done: () => void): Promise<void> {
this.logger.info(`Exporting notes of ${job.data.user.id} ...`);
const user = await this.usersRepository.findOneBy({ id: job.data.user.id });
diff --git a/packages/backend/src/queue/processors/ExportUserListsProcessorService.ts b/packages/backend/src/queue/processors/ExportUserListsProcessorService.ts
index 6400161b8c..ce8ed2f5e8 100644
--- a/packages/backend/src/queue/processors/ExportUserListsProcessorService.ts
+++ b/packages/backend/src/queue/processors/ExportUserListsProcessorService.ts
@@ -11,7 +11,7 @@ import { createTemp } from '@/misc/create-temp.js';
import { UtilityService } from '@/core/UtilityService.js';
import { QueueLoggerService } from '../QueueLoggerService.js';
import type Bull from 'bull';
-import type { DbUserJobData } from '../types.js';
+import type { DbJobDataWithUser } from '../types.js';
import { bindThis } from '@/decorators.js';
@Injectable()
@@ -39,7 +39,7 @@ export class ExportUserListsProcessorService {
}
@bindThis
- public async process(job: Bull.Job<DbUserJobData>, done: () => void): Promise<void> {
+ public async process(job: Bull.Job<DbJobDataWithUser>, done: () => void): Promise<void> {
this.logger.info(`Exporting user lists of ${job.data.user.id} ...`);
const user = await this.usersRepository.findOneBy({ id: job.data.user.id });
diff --git a/packages/backend/src/queue/processors/ImportBlockingProcessorService.ts b/packages/backend/src/queue/processors/ImportBlockingProcessorService.ts
index b8d9b3a52d..3f075b02d2 100644
--- a/packages/backend/src/queue/processors/ImportBlockingProcessorService.ts
+++ b/packages/backend/src/queue/processors/ImportBlockingProcessorService.ts
@@ -1,38 +1,31 @@
import { Inject, Injectable } from '@nestjs/common';
import { IsNull } from 'typeorm';
import { DI } from '@/di-symbols.js';
-import type { UsersRepository, BlockingsRepository, DriveFilesRepository } from '@/models/index.js';
-import type { Config } from '@/config.js';
+import type { UsersRepository, DriveFilesRepository } from '@/models/index.js';
import type Logger from '@/logger.js';
import * as Acct from '@/misc/acct.js';
import { RemoteUserResolveService } from '@/core/RemoteUserResolveService.js';
-import { UserBlockingService } from '@/core/UserBlockingService.js';
import { DownloadService } from '@/core/DownloadService.js';
import { UtilityService } from '@/core/UtilityService.js';
import { QueueLoggerService } from '../QueueLoggerService.js';
import type Bull from 'bull';
-import type { DbUserImportJobData } from '../types.js';
+import type { DbUserImportJobData, DbUserImportToDbJobData } from '../types.js';
import { bindThis } from '@/decorators.js';
+import { QueueService } from '@/core/QueueService.js';
@Injectable()
export class ImportBlockingProcessorService {
private logger: Logger;
constructor(
- @Inject(DI.config)
- private config: Config,
-
@Inject(DI.usersRepository)
private usersRepository: UsersRepository,
- @Inject(DI.blockingsRepository)
- private blockingsRepository: BlockingsRepository,
-
@Inject(DI.driveFilesRepository)
private driveFilesRepository: DriveFilesRepository,
+ private queueService: QueueService,
private utilityService: UtilityService,
- private userBlockingService: UserBlockingService,
private remoteUserResolveService: RemoteUserResolveService,
private downloadService: DownloadService,
private queueLoggerService: QueueLoggerService,
@@ -59,46 +52,50 @@ export class ImportBlockingProcessorService {
}
const csv = await this.downloadService.downloadTextFile(file.url);
+ const targets = csv.trim().split('\n');
+ this.queueService.createImportBlockingToDbJob({ id: user.id }, targets);
- let linenum = 0;
+ this.logger.succ('Import jobs created');
+ done();
+ }
- for (const line of csv.trim().split('\n')) {
- linenum++;
+ @bindThis
+ public async processDb(job: Bull.Job<DbUserImportToDbJobData>): Promise<void> {
+ const line = job.data.target;
+ const user = job.data.user;
- try {
- const acct = line.split(',')[0].trim();
- const { username, host } = Acct.parse(acct);
+ try {
+ const acct = line.split(',')[0].trim();
+ const { username, host } = Acct.parse(acct);
- let target = this.utilityService.isSelfHost(host!) ? await this.usersRepository.findOneBy({
- host: IsNull(),
- usernameLower: username.toLowerCase(),
- }) : await this.usersRepository.findOneBy({
- host: this.utilityService.toPuny(host!),
- usernameLower: username.toLowerCase(),
- });
+ if (!host) return;
- if (host == null && target == null) continue;
+ let target = this.utilityService.isSelfHost(host) ? await this.usersRepository.findOneBy({
+ host: IsNull(),
+ usernameLower: username.toLowerCase(),
+ }) : await this.usersRepository.findOneBy({
+ host: this.utilityService.toPuny(host),
+ usernameLower: username.toLowerCase(),
+ });
- if (target == null) {
- target = await this.remoteUserResolveService.resolveUser(username, host);
- }
+ if (host == null && target == null) return;
- if (target == null) {
- throw `cannot resolve user: @${username}@${host}`;
- }
+ if (target == null) {
+ target = await this.remoteUserResolveService.resolveUser(username, host);
+ }
- // skip myself
- if (target.id === job.data.user.id) continue;
+ if (target == null) {
+ throw `Unable to resolve user: @${username}@${host}`;
+ }
- this.logger.info(`Block[${linenum}] ${target.id} ...`);
+ // skip myself
+ if (target.id === job.data.user.id) return;
- await this.userBlockingService.block(user, target);
- } catch (e) {
- this.logger.warn(`Error in line:${linenum} ${e}`);
- }
- }
+ this.logger.info(`Block ${target.id} ...`);
- this.logger.succ('Imported');
- done();
+ this.queueService.createBlockJob([{ from: { id: user.id }, to: { id: target.id }, silent: true }]);
+ } catch (e) {
+ this.logger.warn(`Error: ${e}`);
+ }
}
}
diff --git a/packages/backend/src/queue/processors/ImportFollowingProcessorService.ts b/packages/backend/src/queue/processors/ImportFollowingProcessorService.ts
index 037a6f2456..aa5cf12c50 100644
--- a/packages/backend/src/queue/processors/ImportFollowingProcessorService.ts
+++ b/packages/backend/src/queue/processors/ImportFollowingProcessorService.ts
@@ -2,34 +2,30 @@ import { Inject, Injectable } from '@nestjs/common';
import { IsNull } from 'typeorm';
import { DI } from '@/di-symbols.js';
import type { UsersRepository, DriveFilesRepository } from '@/models/index.js';
-import type { Config } from '@/config.js';
import type Logger from '@/logger.js';
import * as Acct from '@/misc/acct.js';
import { RemoteUserResolveService } from '@/core/RemoteUserResolveService.js';
import { DownloadService } from '@/core/DownloadService.js';
-import { UserFollowingService } from '@/core/UserFollowingService.js';
import { UtilityService } from '@/core/UtilityService.js';
import { QueueLoggerService } from '../QueueLoggerService.js';
import type Bull from 'bull';
-import type { DbUserImportJobData } from '../types.js';
+import type { DbUserImportJobData, DbUserImportToDbJobData } from '../types.js';
import { bindThis } from '@/decorators.js';
+import { QueueService } from '@/core/QueueService.js';
@Injectable()
export class ImportFollowingProcessorService {
private logger: Logger;
constructor(
- @Inject(DI.config)
- private config: Config,
-
@Inject(DI.usersRepository)
private usersRepository: UsersRepository,
@Inject(DI.driveFilesRepository)
private driveFilesRepository: DriveFilesRepository,
+ private queueService: QueueService,
private utilityService: UtilityService,
- private userFollowingService: UserFollowingService,
private remoteUserResolveService: RemoteUserResolveService,
private downloadService: DownloadService,
private queueLoggerService: QueueLoggerService,
@@ -56,46 +52,50 @@ export class ImportFollowingProcessorService {
}
const csv = await this.downloadService.downloadTextFile(file.url);
+ const targets = csv.trim().split('\n');
+ this.queueService.createImportFollowingToDbJob({ id: user.id }, targets);
- let linenum = 0;
+ this.logger.succ('Import jobs created');
+ done();
+ }
+
+ @bindThis
+ public async processDb(job: Bull.Job<DbUserImportToDbJobData>): Promise<void> {
+ const line = job.data.target;
+ const user = job.data.user;
- for (const line of csv.trim().split('\n')) {
- linenum++;
+ try {
+ const acct = line.split(',')[0].trim();
+ const { username, host } = Acct.parse(acct);
- try {
- const acct = line.split(',')[0].trim();
- const { username, host } = Acct.parse(acct);
+ if (!host) return;
- let target = this.utilityService.isSelfHost(host!) ? await this.usersRepository.findOneBy({
- host: IsNull(),
- usernameLower: username.toLowerCase(),
- }) : await this.usersRepository.findOneBy({
- host: this.utilityService.toPuny(host!),
- usernameLower: username.toLowerCase(),
- });
+ let target = this.utilityService.isSelfHost(host) ? await this.usersRepository.findOneBy({
+ host: IsNull(),
+ usernameLower: username.toLowerCase(),
+ }) : await this.usersRepository.findOneBy({
+ host: this.utilityService.toPuny(host),
+ usernameLower: username.toLowerCase(),
+ });
- if (host == null && target == null) continue;
+ if (host == null && target == null) return;
- if (target == null) {
- target = await this.remoteUserResolveService.resolveUser(username, host);
- }
+ if (target == null) {
+ target = await this.remoteUserResolveService.resolveUser(username, host);
+ }
- if (target == null) {
- throw `cannot resolve user: @${username}@${host}`;
- }
+ if (target == null) {
+ throw `Unable to resolve user: @${username}@${host}`;
+ }
- // skip myself
- if (target.id === job.data.user.id) continue;
+ // skip myself
+ if (target.id === job.data.user.id) return;
- this.logger.info(`Follow[${linenum}] ${target.id} ...`);
+ this.logger.info(`Follow ${target.id} ...`);
- this.userFollowingService.follow(user, target);
- } catch (e) {
- this.logger.warn(`Error in line:${linenum} ${e}`);
- }
+ this.queueService.createFollowJob([{ from: user, to: { id: target.id }, silent: true }]);
+ } catch (e) {
+ this.logger.warn(`Error: ${e}`);
}
-
- this.logger.succ('Imported');
- done();
}
}
diff --git a/packages/backend/src/queue/processors/ImportMutingProcessorService.ts b/packages/backend/src/queue/processors/ImportMutingProcessorService.ts
index 83d382057b..379994ee79 100644
--- a/packages/backend/src/queue/processors/ImportMutingProcessorService.ts
+++ b/packages/backend/src/queue/processors/ImportMutingProcessorService.ts
@@ -66,11 +66,13 @@ export class ImportMutingProcessorService {
const acct = line.split(',')[0].trim();
const { username, host } = Acct.parse(acct);
- let target = this.utilityService.isSelfHost(host!) ? await this.usersRepository.findOneBy({
+ if (!host) continue;
+
+ let target = this.utilityService.isSelfHost(host) ? await this.usersRepository.findOneBy({
host: IsNull(),
usernameLower: username.toLowerCase(),
}) : await this.usersRepository.findOneBy({
- host: this.utilityService.toPuny(host!),
+ host: this.utilityService.toPuny(host),
usernameLower: username.toLowerCase(),
});
diff --git a/packages/backend/src/queue/processors/RelationshipProcessorService.ts b/packages/backend/src/queue/processors/RelationshipProcessorService.ts
new file mode 100644
index 0000000000..a5006dcf03
--- /dev/null
+++ b/packages/backend/src/queue/processors/RelationshipProcessorService.ts
@@ -0,0 +1,68 @@
+import { Inject, Injectable } from '@nestjs/common';
+import type Bull from 'bull';
+
+import { UserFollowingService } from '@/core/UserFollowingService.js';
+import { UserBlockingService } from '@/core/UserBlockingService.js';
+import { bindThis } from '@/decorators.js';
+import type Logger from '@/logger.js';
+
+import { QueueLoggerService } from '../QueueLoggerService.js';
+import { RelationshipJobData } from '../types.js';
+import type { UsersRepository } from '@/models/index.js';
+import { DI } from '@/di-symbols.js';
+
+@Injectable()
+export class RelationshipProcessorService {
+ private logger: Logger;
+
+ constructor(
+ @Inject(DI.usersRepository)
+ private usersRepository: UsersRepository,
+
+ private queueLoggerService: QueueLoggerService,
+ private userFollowingService: UserFollowingService,
+ private userBlockingService: UserBlockingService,
+ ) {
+ this.logger = this.queueLoggerService.logger.createSubLogger('follow-block');
+ }
+
+ @bindThis
+ public async processFollow(job: Bull.Job<RelationshipJobData>): Promise<string> {
+ this.logger.info(`${job.data.from.id} is trying to follow ${job.data.to.id}`);
+ await this.userFollowingService.follow(job.data.from, job.data.to, job.data.requestId, job.data.silent);
+ return 'ok';
+ }
+
+ @bindThis
+ 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 }),
+ ]);
+ await this.userFollowingService.unfollow(follower, followee, job.data.silent);
+ return 'ok';
+ }
+
+ @bindThis
+ 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 }),
+ ]);
+ await this.userBlockingService.block(blockee, blocker, job.data.silent);
+ return 'ok';
+ }
+
+ @bindThis
+ 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 }),
+ ]);
+ await this.userBlockingService.unblock(blockee, blocker);
+ return 'ok';
+ }
+}
diff --git a/packages/backend/src/queue/types.ts b/packages/backend/src/queue/types.ts
index 5d650c6864..23c973d449 100644
--- a/packages/backend/src/queue/types.ts
+++ b/packages/backend/src/queue/types.ts
@@ -21,9 +21,39 @@ export type InboxJobData = {
signature: httpSignature.IParsedSignature;
};
-export type DbJobData = DbUserJobData | DbUserImportJobData | DbUserDeleteJobData;
+export type RelationshipJobData = {
+ from: ThinUser;
+ to: ThinUser;
+ silent?: boolean;
+ requestId?: string;
+}
-export type DbUserJobData = {
+export type DbJobData<T extends keyof DbJobMap> = DbJobMap[T];
+
+export type DbJobMap = {
+ deleteDriveFiles: DbJobDataWithUser;
+ exportCustomEmojis: DbJobDataWithUser;
+ exportNotes: DbJobDataWithUser;
+ exportFavorites: DbJobDataWithUser;
+ exportFollowing: DbExportFollowingData;
+ exportMuting: DbJobDataWithUser;
+ exportBlocking: DbJobDataWithUser;
+ exportUserLists: DbJobDataWithUser;
+ importFollowing: DbUserImportJobData;
+ importFollowingToDb: DbUserImportToDbJobData;
+ importMuting: DbUserImportJobData;
+ importBlocking: DbUserImportJobData;
+ importBlockingToDb: DbUserImportToDbJobData;
+ importUserLists: DbUserImportJobData;
+ importCustomEmojis: DbUserImportJobData;
+ deleteAccount: DbUserDeleteJobData;
+}
+
+export type DbJobDataWithUser = {
+ user: ThinUser;
+}
+
+export type DbExportFollowingData = {
user: ThinUser;
excludeMuting: boolean;
excludeInactive: boolean;
@@ -39,6 +69,11 @@ export type DbUserImportJobData = {
fileId: DriveFile['id'];
};
+export type DbUserImportToDbJobData = {
+ user: ThinUser;
+ target: string;
+};
+
export type ObjectStorageJobData = ObjectStorageFileJobData | Record<string, unknown>;
export type ObjectStorageFileJobData = {
diff --git a/packages/backend/src/server/api/endpoints/admin/federation/remove-all-following.ts b/packages/backend/src/server/api/endpoints/admin/federation/remove-all-following.ts
index b073209a5b..54ce095488 100644
--- a/packages/backend/src/server/api/endpoints/admin/federation/remove-all-following.ts
+++ b/packages/backend/src/server/api/endpoints/admin/federation/remove-all-following.ts
@@ -1,8 +1,8 @@
import { Inject, Injectable } from '@nestjs/common';
import { Endpoint } from '@/server/api/endpoint-base.js';
import type { FollowingsRepository, UsersRepository } from '@/models/index.js';
-import { UserFollowingService } from '@/core/UserFollowingService.js';
import { DI } from '@/di-symbols.js';
+import { QueueService } from '@/core/QueueService.js';
export const meta = {
tags: ['admin'],
@@ -29,7 +29,7 @@ export default class extends Endpoint<typeof meta, typeof paramDef> {
@Inject(DI.notesRepository)
private followingsRepository: FollowingsRepository,
- private userFollowingService: UserFollowingService,
+ private queueService: QueueService,
) {
super(meta, paramDef, async (ps, me) => {
const followings = await this.followingsRepository.findBy({
@@ -41,9 +41,7 @@ export default class extends Endpoint<typeof meta, typeof paramDef> {
this.usersRepository.findOneByOrFail({ id: f.followeeId }),
])));
- for (const pair of pairs) {
- this.userFollowingService.unfollow(pair[0], pair[1]);
- }
+ this.queueService.createUnfollowJob(pairs.map(p => ({ to: p[0], from: p[1], silent: true })));
});
}
}
diff --git a/packages/backend/src/server/api/endpoints/admin/suspend-user.ts b/packages/backend/src/server/api/endpoints/admin/suspend-user.ts
index 3c99225272..eabbceac0e 100644
--- a/packages/backend/src/server/api/endpoints/admin/suspend-user.ts
+++ b/packages/backend/src/server/api/endpoints/admin/suspend-user.ts
@@ -1,15 +1,15 @@
+import { IsNull, Not } from 'typeorm';
import { Inject, Injectable } from '@nestjs/common';
import { Endpoint } from '@/server/api/endpoint-base.js';
import type { UsersRepository, FollowingsRepository } from '@/models/index.js';
import type { User } from '@/models/entities/User.js';
-import { GlobalEventService } from '@/core/GlobalEventService.js';
+import type { RelationshipJobData } from '@/queue/types.js';
import { ModerationLogService } from '@/core/ModerationLogService.js';
import { UserSuspendService } from '@/core/UserSuspendService.js';
-import { UserFollowingService } from '@/core/UserFollowingService.js';
import { DI } from '@/di-symbols.js';
-import { UserEntityService } from '@/core/entities/UserEntityService.js';
import { bindThis } from '@/decorators.js';
import { RoleService } from '@/core/RoleService.js';
+import { QueueService } from '@/core/QueueService.js';
export const meta = {
tags: ['admin'],
@@ -36,12 +36,10 @@ export default class extends Endpoint<typeof meta, typeof paramDef> {
@Inject(DI.followingsRepository)
private followingsRepository: FollowingsRepository,
- private userEntityService: UserEntityService,
- private userFollowingService: UserFollowingService,
private userSuspendService: UserSuspendService,
private roleService: RoleService,
private moderationLogService: ModerationLogService,
- private globalEventService: GlobalEventService,
+ private queueService: QueueService,
) {
super(meta, paramDef, async (ps, me) => {
const user = await this.usersRepository.findOneBy({ id: ps.userId });
@@ -71,20 +69,23 @@ export default class extends Endpoint<typeof meta, typeof paramDef> {
@bindThis
private async unFollowAll(follower: User) {
- const followings = await this.followingsRepository.findBy({
- followerId: follower.id,
+ const followings = await this.followingsRepository.find({
+ where: {
+ followerId: follower.id,
+ followeeId: Not(IsNull()),
+ },
});
-
+
+ const jobs: RelationshipJobData[] = [];
for (const following of followings) {
- const followee = await this.usersRepository.findOneBy({
- id: following.followeeId,
- });
-
- if (followee == null) {
- throw `Cant find followee ${following.followeeId}`;
+ if (following.followeeId && following.followerId) {
+ jobs.push({
+ from: { id: following.followerId },
+ to: { id: following.followeeId },
+ silent: true,
+ });
}
-
- await this.userFollowingService.unfollow(follower, followee, true);
}
+ this.queueService.createUnfollowJob(jobs);
}
}