mirrors/barkey
1
0
Fork 0
mirror of https://codeberg.org/yeentown/barkey.git synced 2025-11-03 23:14:13 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

refactor: SystemWebhook/UserWebhookの配信処理呼び出し部分の改善 (#15035)

Browse source 
* UserWebhook側の対処

* SystemWebhook側の対処

* fix test
This commit is contained in:
おさむのひと 2025-01-14 20:14:02 +09:00 • committed by GitHub
parent 5445b023e5
commit d2e22f9050
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
12 changed files with 258 additions and 140 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

32
packages/backend/src/core/AbuseReportNotificationService.ts
View file

@ -160,22 +160,22 @@ export class AbuseReportNotificationService implements OnApplicationShutdown {
};
});
const recipientWebhookIds = await this.fetchWebhookRecipients()
.then(it => it
.filter(it => it.isActive && it.systemWebhookId && it.method === 'webhook')
.map(it => it.systemWebhookId)
.filter(x => x != null));
for (const webhookId of recipientWebhookIds) {
await Promise.all(
convertedReports.map(it => {
return this.systemWebhookService.enqueueSystemWebhook(
webhookId,
type,
it,
);
}),
);
}
const inactiveRecipients = await this.fetchWebhookRecipients()
.then(it => it.filter(it => !it.isActive));
const withoutWebhookIds = inactiveRecipients
.map(it => it.systemWebhookId)
.filter(x => x != null);
return Promise.all(
convertedReports.map(it => {
return this.systemWebhookService.enqueueSystemWebhook(
type,
it,
{
excludes: withoutWebhookIds,
},
);
}),
);
}
/**

33
packages/backend/src/core/NoteCreateService.ts
View file

@ -614,14 +614,7 @@ export class NoteCreateService implements OnApplicationShutdown {
this.roleService.addNoteToRoleTimeline(noteObj);
this.webhookService.getActiveWebhooks().then(webhooks => {
webhooks = webhooks.filter(x => x.userId === user.id && x.on.includes('note'));
for (const webhook of webhooks) {
this.queueService.userWebhookDeliver(webhook, 'note', {
note: noteObj,
});
}
});
this.webhookService.enqueueUserWebhook(user.id, 'note', { note: noteObj });
const nm = new NotificationManager(this.mutingsRepository, this.notificationService, user, note);
@ -641,13 +634,7 @@ export class NoteCreateService implements OnApplicationShutdown {
if (!isThreadMuted) {
nm.push(data.reply.userId, 'reply');
this.globalEventService.publishMainStream(data.reply.userId, 'reply', noteObj);
const webhooks = (await this.webhookService.getActiveWebhooks()).filter(x => x.userId === data.reply!.userId && x.on.includes('reply'));
for (const webhook of webhooks) {
this.queueService.userWebhookDeliver(webhook, 'reply', {
note: noteObj,
});
}
this.webhookService.enqueueUserWebhook(data.reply.userId, 'reply', { note: noteObj });
}
}
}
@ -664,13 +651,7 @@ export class NoteCreateService implements OnApplicationShutdown {
// Publish event
if ((user.id !== data.renote.userId) && data.renote.userHost === null) {
this.globalEventService.publishMainStream(data.renote.userId, 'renote', noteObj);
const webhooks = (await this.webhookService.getActiveWebhooks()).filter(x => x.userId === data.renote!.userId && x.on.includes('renote'));
for (const webhook of webhooks) {
this.queueService.userWebhookDeliver(webhook, 'renote', {
note: noteObj,
});
}
this.webhookService.enqueueUserWebhook(data.renote.userId, 'renote', { note: noteObj });
}
}
@ -796,13 +777,7 @@ export class NoteCreateService implements OnApplicationShutdown {
});
this.globalEventService.publishMainStream(u.id, 'mention', detailPackedNote);
const webhooks = (await this.webhookService.getActiveWebhooks()).filter(x => x.userId === u.id && x.on.includes('mention'));
for (const webhook of webhooks) {
this.queueService.userWebhookDeliver(webhook, 'mention', {
note: detailPackedNote,
});
}
this.webhookService.enqueueUserWebhook(u.id, 'mention', { note: detailPackedNote });
// Create notification
nm.push(u.id, 'mention');

31
packages/backend/src/core/SystemWebhookService.ts
View file

@ -50,7 +50,6 @@ export type SystemWebhookPayload<T extends SystemWebhookEventType> =
@Injectable()
export class SystemWebhookService implements OnApplicationShutdown {
private logger: Logger;
private activeSystemWebhooksFetched = false;
private activeSystemWebhooks: MiSystemWebhook[] = [];
@ -62,11 +61,9 @@ export class SystemWebhookService implements OnApplicationShutdown {
private idService: IdService,
private queueService: QueueService,
private moderationLogService: ModerationLogService,
private loggerService: LoggerService,
private globalEventService: GlobalEventService,
) {
this.redisForSub.on('message', this.onMessage);
this.logger = this.loggerService.getLogger('webhook');
}
@bindThis
@ -193,28 +190,24 @@ export class SystemWebhookService implements OnApplicationShutdown {
/**
* SystemWebhook をWebhook配送キューに追加する
* @see QueueService.systemWebhookDeliver
* // TODO: contentの型を厳格化する
*/
@bindThis
public async enqueueSystemWebhook<T extends SystemWebhookEventType>(
webhook: MiSystemWebhook | MiSystemWebhook['id'],
type: T,
content: SystemWebhookPayload<T>,
opts?: {
excludes?: MiSystemWebhook['id'][];
},
) {
const webhookEntity = typeof webhook === 'string'
? (await this.fetchActiveSystemWebhooks()).find(a => a.id === webhook)
: webhook;
if (!webhookEntity || !webhookEntity.isActive) {
this.logger.info(`SystemWebhook is not active or not found : ${webhook}`);
return;
}
if (!webhookEntity.on.includes(type)) {
this.logger.info(`SystemWebhook ${webhookEntity.id} is not listening to ${type}`);
return;
}
return this.queueService.systemWebhookDeliver(webhookEntity, type, content);
const webhooks = await this.fetchActiveSystemWebhooks()
.then(webhooks => {
return webhooks.filter(webhook => !opts?.excludes?.includes(webhook.id) && webhook.on.includes(type));
});
return Promise.all(
webhooks.map(webhook => {
return this.queueService.systemWebhookDeliver(webhook, type, content);
}),
);
}
@bindThis

8
packages/backend/src/core/UserBlockingService.ts
View file

@ -118,13 +118,7 @@ export class UserBlockingService implements OnModuleInit {
schema: 'UserDetailedNotMe',
}).then(async packed => {
this.globalEventService.publishMainStream(follower.id, 'unfollow', packed);
const webhooks = (await this.webhookService.getActiveWebhooks()).filter(x => x.userId === follower.id && x.on.includes('unfollow'));
for (const webhook of webhooks) {
this.queueService.userWebhookDeliver(webhook, 'unfollow', {
user: packed,
});
}
this.webhookService.enqueueUserWebhook(follower.id, 'unfollow', { user: packed });
});
}

32
packages/backend/src/core/UserFollowingService.ts
View file

@ -333,13 +333,7 @@ export class UserFollowingService implements OnModuleInit {
schema: 'UserDetailedNotMe',
}).then(async packed => {
this.globalEventService.publishMainStream(follower.id, 'follow', packed);
const webhooks = (await this.webhookService.getActiveWebhooks()).filter(x => x.userId === follower.id && x.on.includes('follow'));
for (const webhook of webhooks) {
this.queueService.userWebhookDeliver(webhook, 'follow', {
user: packed,
});
}
this.webhookService.enqueueUserWebhook(follower.id, 'follow', { user: packed });
});
}
@ -347,13 +341,7 @@ export class UserFollowingService implements OnModuleInit {
if (this.userEntityService.isLocalUser(followee)) {
this.userEntityService.pack(follower.id, followee).then(async packed => {
this.globalEventService.publishMainStream(followee.id, 'followed', packed);
const webhooks = (await this.webhookService.getActiveWebhooks()).filter(x => x.userId === followee.id && x.on.includes('followed'));
for (const webhook of webhooks) {
this.queueService.userWebhookDeliver(webhook, 'followed', {
user: packed,
});
}
this.webhookService.enqueueUserWebhook(followee.id, 'followed', { user: packed });
});
// 通知を作成
@ -400,13 +388,7 @@ export class UserFollowingService implements OnModuleInit {
schema: 'UserDetailedNotMe',
}).then(async packed => {
this.globalEventService.publishMainStream(follower.id, 'unfollow', packed);
const webhooks = (await this.webhookService.getActiveWebhooks()).filter(x => x.userId === follower.id && x.on.includes('unfollow'));
for (const webhook of webhooks) {
this.queueService.userWebhookDeliver(webhook, 'unfollow', {
user: packed,
});
}
this.webhookService.enqueueUserWebhook(follower.id, 'unfollow', { user: packed });
});
}
@ -744,13 +726,7 @@ export class UserFollowingService implements OnModuleInit {
});
this.globalEventService.publishMainStream(follower.id, 'unfollow', packedFollowee);
const webhooks = (await this.webhookService.getActiveWebhooks()).filter(x => x.userId === follower.id && x.on.includes('unfollow'));
for (const webhook of webhooks) {
this.queueService.userWebhookDeliver(webhook, 'unfollow', {
user: packedFollowee,
});
}
this.webhookService.enqueueUserWebhook(follower.id, 'unfollow', { user: packedFollowee });
}
@bindThis

9
packages/backend/src/core/UserService.ts
View file

@ -63,13 +63,6 @@ export class UserService {
@bindThis
public async notifySystemWebhook(user: MiUser, type: 'userCreated') {
const packedUser = await this.userEntityService.pack(user, null, { schema: 'UserLite' });
const recipientWebhookIds = await this.systemWebhookService.fetchSystemWebhooks({ isActive: true, on: [type] });
for (const webhookId of recipientWebhookIds) {
await this.systemWebhookService.enqueueSystemWebhook(
webhookId,
type,
packedUser,
);
}
return this.systemWebhookService.enqueueSystemWebhook(type, packedUser);
}
}

25
packages/backend/src/core/UserWebhookService.ts
View file

@ -5,13 +5,14 @@
import { Inject, Injectable } from '@nestjs/common';
import * as Redis from 'ioredis';
import { type WebhooksRepository } from '@/models/_.js';
import { MiUser, type WebhooksRepository } from '@/models/_.js';
import { MiWebhook, WebhookEventTypes } from '@/models/Webhook.js';
import { DI } from '@/di-symbols.js';
import { bindThis } from '@/decorators.js';
import { GlobalEvents } from '@/core/GlobalEventService.js';
import type { OnApplicationShutdown } from '@nestjs/common';
import type { Packed } from '@/misc/json-schema.js';
import { QueueService } from '@/core/QueueService.js';
import type { OnApplicationShutdown } from '@nestjs/common';
export type UserWebhookPayload<T extends WebhookEventTypes> =
T extends 'note' | 'reply' | 'renote' |'mention' ? {
@ -34,6 +35,7 @@ export class UserWebhookService implements OnApplicationShutdown {
private redisForSub: Redis.Redis,
@Inject(DI.webhooksRepository)
private webhooksRepository: WebhooksRepository,
private queueService: QueueService,
) {
this.redisForSub.on('message', this.onMessage);
}
@ -75,6 +77,25 @@ export class UserWebhookService implements OnApplicationShutdown {
return query.getMany();
}
/**
* UserWebhook をWebhook配送キューに追加する
* @see QueueService.userWebhookDeliver
*/
@bindThis
public async enqueueUserWebhook<T extends WebhookEventTypes>(
userId: MiUser['id'],
type: T,
content: UserWebhookPayload<T>,
) {
const webhooks = await this.getActiveWebhooks()
.then(webhooks => webhooks.filter(webhook => webhook.userId === userId && webhook.on.includes(type)));
return Promise.all(
webhooks.map(webhook => {
return this.queueService.userWebhookDeliver(webhook, type, content);
}),
);
}
@bindThis
private async onMessage(_: string, data: string): Promise<void> {
const obj = JSON.parse(data);

26
packages/backend/src/queue/processors/CheckModeratorsActivityProcessorService.ts
View file

@ -231,15 +231,10 @@ export class CheckModeratorsActivityProcessorService {
// -- SystemWebhook
const systemWebhooks = await this.systemWebhookService.fetchActiveSystemWebhooks()
.then(it => it.filter(it => it.on.includes('inactiveModeratorsWarning')));
for (const systemWebhook of systemWebhooks) {
this.systemWebhookService.enqueueSystemWebhook(
systemWebhook,
'inactiveModeratorsWarning',
{ remainingTime: remainingTime },
);
}
return this.systemWebhookService.enqueueSystemWebhook(
'inactiveModeratorsWarning',
{ remainingTime: remainingTime },
);
}
@bindThis
@ -269,15 +264,10 @@ export class CheckModeratorsActivityProcessorService {
// -- SystemWebhook
const systemWebhooks = await this.systemWebhookService.fetchActiveSystemWebhooks()
.then(it => it.filter(it => it.on.includes('inactiveModeratorsInvitationOnlyChanged')));
for (const systemWebhook of systemWebhooks) {
this.systemWebhookService.enqueueSystemWebhook(
systemWebhook,
'inactiveModeratorsInvitationOnlyChanged',
{},
);
}
return this.systemWebhookService.enqueueSystemWebhook(
'inactiveModeratorsInvitationOnlyChanged',
{},
);
}
@bindThis

50
packages/backend/test/unit/AbuseReportNotificationService.ts
View file

@ -3,13 +3,14 @@
* SPDX-License-Identifier: AGPL-3.0-only
*/
import { jest } from '@jest/globals';
import { describe, jest } from '@jest/globals';
import { Test, TestingModule } from '@nestjs/testing';
import { randomString } from '../utils.js';
import { AbuseReportNotificationService } from '@/core/AbuseReportNotificationService.js';
import {
AbuseReportNotificationRecipientRepository,
MiAbuseReportNotificationRecipient,
MiAbuseUserReport,
MiSystemWebhook,
MiUser,
SystemWebhooksRepository,
@ -112,7 +113,10 @@ describe('AbuseReportNotificationService', () => {
provide: SystemWebhookService, useFactory: () => ({ enqueueSystemWebhook: jest.fn() }),
},
{
provide: UserEntityService, useFactory: () => ({ pack: (v: any) => v }),
provide: UserEntityService, useFactory: () => ({
pack: (v: any) => Promise.resolve(v),
packMany: (v: any) => Promise.resolve(v),
}),
},
{
provide: EmailService, useFactory: () => ({ sendEmail: jest.fn() }),
@ -344,4 +348,46 @@ describe('AbuseReportNotificationService', () => {
expect(recipients).toEqual([recipient3]);
});
});
describe('notifySystemWebhook', () => {
test('非アクティブな通報通知はWebhook送信から除外される', async () => {
const recipient1 = await createRecipient({
method: 'webhook',
systemWebhookId: systemWebhook1.id,
isActive: true,
});
const recipient2 = await createRecipient({
method: 'webhook',
systemWebhookId: systemWebhook2.id,
isActive: false,
});
const reports: MiAbuseUserReport[] = [
{
id: idService.gen(),
targetUserId: alice.id,
targetUser: alice,
reporterId: bob.id,
reporter: bob,
assigneeId: null,
assignee: null,
resolved: false,
forwarded: false,
comment: 'test',
moderationNote: '',
resolvedAs: null,
targetUserHost: null,
reporterHost: null,
},
];
await service.notifySystemWebhook(reports, 'abuseReport');
// 実際に除外されるかはSystemWebhookService側で確認する.
// ここでは非アクティブな通報通知を除外設定できているかを確認する
expect(webhookService.enqueueSystemWebhook).toHaveBeenCalledTimes(1);
expect(webhookService.enqueueSystemWebhook.mock.calls[0][0]).toBe('abuseReport');
expect(webhookService.enqueueSystemWebhook.mock.calls[0][2]).toEqual({ excludes: [systemWebhook2.id] });
});
});
});

49
packages/backend/test/unit/SystemWebhookService.ts
View file

@ -314,9 +314,10 @@ describe('SystemWebhookService', () => {
isActive: true,
on: ['abuseReport'],
});
await service.enqueueSystemWebhook(webhook.id, 'abuseReport', { foo: 'bar' } as any);
await service.enqueueSystemWebhook('abuseReport', { foo: 'bar' } as any);
expect(queueService.systemWebhookDeliver).toHaveBeenCalled();
expect(queueService.systemWebhookDeliver).toHaveBeenCalledTimes(1);
expect(queueService.systemWebhookDeliver.mock.calls[0][0] as MiSystemWebhook).toEqual(webhook);
});
test('非アクティブなWebhookはキューに追加されない', async () => {
@ -324,7 +325,7 @@ describe('SystemWebhookService', () => {
isActive: false,
on: ['abuseReport'],
});
await service.enqueueSystemWebhook(webhook.id, 'abuseReport', { foo: 'bar' } as any);
await service.enqueueSystemWebhook('abuseReport', { foo: 'bar' } as any);
expect(queueService.systemWebhookDeliver).not.toHaveBeenCalled();
});
@ -338,11 +339,49 @@ describe('SystemWebhookService', () => {
isActive: true,
on: ['abuseReportResolved'],
});
await service.enqueueSystemWebhook(webhook1.id, 'abuseReport', { foo: 'bar' } as any);
await service.enqueueSystemWebhook(webhook2.id, 'abuseReport', { foo: 'bar' } as any);
await service.enqueueSystemWebhook('abuseReport', { foo: 'bar' } as any);
expect(queueService.systemWebhookDeliver).not.toHaveBeenCalled();
});
test('混在した時、有効かつ許可されたイベント種別のみ', async () => {
const webhook1 = await createWebhook({
isActive: true,
on: ['abuseReport'],
});
const webhook2 = await createWebhook({
isActive: true,
on: ['abuseReportResolved'],
});
const webhook3 = await createWebhook({
isActive: false,
on: ['abuseReport'],
});
const webhook4 = await createWebhook({
isActive: false,
on: ['abuseReportResolved'],
});
await service.enqueueSystemWebhook('abuseReport', { foo: 'bar' } as any);
expect(queueService.systemWebhookDeliver).toHaveBeenCalledTimes(1);
expect(queueService.systemWebhookDeliver.mock.calls[0][0] as MiSystemWebhook).toEqual(webhook1);
});
test('除外指定した場合は送信されない', async () => {
const webhook1 = await createWebhook({
isActive: true,
on: ['abuseReport'],
});
const webhook2 = await createWebhook({
isActive: true,
on: ['abuseReport'],
});
await service.enqueueSystemWebhook('abuseReport', { foo: 'bar' } as any, { excludes: [webhook2.id] });
expect(queueService.systemWebhookDeliver).toHaveBeenCalledTimes(1);
expect(queueService.systemWebhookDeliver.mock.calls[0][0] as MiSystemWebhook).toEqual(webhook1);
});
});
describe('fetchActiveSystemWebhooks', () => {

91
packages/backend/test/unit/UserWebhookService.ts
View file

@ -1,4 +1,3 @@
/*
* SPDX-FileCopyrightText: syuilo and misskey-project
* SPDX-License-Identifier: AGPL-3.0-only
@ -71,7 +70,7 @@ describe('UserWebhookService', () => {
LoggerService,
GlobalEventService,
{
provide: QueueService, useFactory: () => ({ systemWebhookDeliver: jest.fn() }),
provide: QueueService, useFactory: () => ({ userWebhookDeliver: jest.fn() }),
},
],
})
@ -242,4 +241,92 @@ describe('UserWebhookService', () => {
});
});
});
describe('アプリを毎回作り直す必要があるグループ', () => {
beforeEach(async () => {
await beforeAllImpl();
await beforeEachImpl();
});
afterEach(async () => {
await afterEachImpl();
await afterAllImpl();
});
describe('enqueueUserWebhook', () => {
test('キューに追加成功', async () => {
const webhook = await createWebhook({
active: true,
on: ['note'],
});
await service.enqueueUserWebhook(webhook.userId, 'note', { foo: 'bar' } as any);
expect(queueService.userWebhookDeliver).toHaveBeenCalledTimes(1);
expect(queueService.userWebhookDeliver.mock.calls[0][0] as MiWebhook).toEqual(webhook);
});
test('非アクティブなWebhookはキューに追加されない', async () => {
const webhook = await createWebhook({
active: false,
on: ['note'],
});
await service.enqueueUserWebhook(webhook.userId, 'note', { foo: 'bar' } as any);
expect(queueService.userWebhookDeliver).not.toHaveBeenCalled();
});
test('未許可のイベント種別が渡された場合はWebhookはキューに追加されない', async () => {
const webhook1 = await createWebhook({
active: true,
on: [],
});
const webhook2 = await createWebhook({
active: true,
on: ['note'],
});
await service.enqueueUserWebhook(webhook1.userId, 'renote', { foo: 'bar' } as any);
await service.enqueueUserWebhook(webhook2.userId, 'renote', { foo: 'bar' } as any);
expect(queueService.userWebhookDeliver).not.toHaveBeenCalled();
});
test('ユーザIDが異なるWebhookはキューに追加されない', async () => {
const webhook = await createWebhook({
active: true,
on: ['note'],
});
await service.enqueueUserWebhook(idService.gen(), 'note', { foo: 'bar' } as any);
expect(queueService.userWebhookDeliver).not.toHaveBeenCalled();
});
test('混在した時、有効かつ許可されたイベント種別のみ', async () => {
const userId = root.id;
const webhook1 = await createWebhook({
userId,
active: true,
on: ['note'],
});
const webhook2 = await createWebhook({
userId,
active: true,
on: ['renote'],
});
const webhook3 = await createWebhook({
userId,
active: false,
on: ['note'],
});
const webhook4 = await createWebhook({
userId,
active: false,
on: ['renote'],
});
await service.enqueueUserWebhook(userId, 'note', { foo: 'bar' } as any);
expect(queueService.userWebhookDeliver).toHaveBeenCalledTimes(1);
expect(queueService.userWebhookDeliver.mock.calls[0][0] as MiWebhook).toEqual(webhook1);
});
});
});
});

12
packages/backend/test/unit/queue/processors/CheckModeratorsActivityProcessorService.ts
View file

@ -18,6 +18,7 @@ import { QueueLoggerService } from '@/queue/QueueLoggerService.js';
import { EmailService } from '@/core/EmailService.js';
import { SystemWebhookService } from '@/core/SystemWebhookService.js';
import { AnnouncementService } from '@/core/AnnouncementService.js';
import { SystemWebhookEventType } from '@/models/SystemWebhook.js';
const baseDate = new Date(Date.UTC(2000, 11, 15, 12, 0, 0));
@ -334,9 +335,10 @@ describe('CheckModeratorsActivityProcessorService', () => {
mockModeratorRole([user1]);
await service.notifyInactiveModeratorsWarning({ time: 1, asDays: 0, asHours: 0 });
expect(systemWebhookService.enqueueSystemWebhook).toHaveBeenCalledTimes(2);
expect(systemWebhookService.enqueueSystemWebhook.mock.calls[0][0]).toEqual(systemWebhook1);
expect(systemWebhookService.enqueueSystemWebhook.mock.calls[1][0]).toEqual(systemWebhook2);
// typeとactiveによる絞り込みが機能しているかはSystemWebhookServiceのテストで確認する.
// ここでは呼び出されているか、typeが正しいかのみを確認する
expect(systemWebhookService.enqueueSystemWebhook).toHaveBeenCalledTimes(1);
expect(systemWebhookService.enqueueSystemWebhook.mock.calls[0][0] as SystemWebhookEventType).toEqual('inactiveModeratorsWarning');
});
});
@ -372,8 +374,10 @@ describe('CheckModeratorsActivityProcessorService', () => {
mockModeratorRole([user1]);
await service.notifyChangeToInvitationOnly();
// typeとactiveによる絞り込みが機能しているかはSystemWebhookServiceのテストで確認する.
// ここでは呼び出されているか、typeが正しいかのみを確認する
expect(systemWebhookService.enqueueSystemWebhook).toHaveBeenCalledTimes(1);
expect(systemWebhookService.enqueueSystemWebhook.mock.calls[0][0]).toEqual(systemWebhook2);
expect(systemWebhookService.enqueueSystemWebhook.mock.calls[0][0] as SystemWebhookEventType).toEqual('inactiveModeratorsInvitationOnlyChanged');
});
});
});