feat: アンケート終了通知

Resolve #4664

packages/backend/src/queue/index.ts

@@ -8,10 +8,11 @@ import processInbox from './processors/inbox.js';
 import processDb from './processors/db/index.js';
 import processObjectStorage from './processors/object-storage/index.js';
 import processSystemQueue from './processors/system/index.js';
+import { endedPollNotification } from './processors/ended-poll-notification.js';
 import { queueLogger } from './logger.js';
 import { DriveFile } from '@/models/entities/drive-file.js';
 import { getJobInfo } from './get-job-info.js';
-import { systemQueue, dbQueue, deliverQueue, inboxQueue, objectStorageQueue } from './queues.js';
+import { systemQueue, dbQueue, deliverQueue, inboxQueue, objectStorageQueue, endedPollNotificationQueue } from './queues.js';
 import { ThinUser } from './types.js';
 import { IActivity } from '@/remote/activitypub/type.js';
 
@@ -255,6 +256,7 @@ export default function() {
 
 	deliverQueue.process(config.deliverJobConcurrency || 128, processDeliver);
 	inboxQueue.process(config.inboxJobConcurrency || 16, processInbox);
+	endedPollNotificationQueue.process(endedPollNotification);
 	processDb(dbQueue);
 	processObjectStorage(objectStorageQueue);
 

packages/backend/src/queue/processors/ended-poll-notification.ts

@@ -0,0 +1,33 @@
+import Bull from 'bull';
+import { In } from 'typeorm';
+import { Notes, Polls, PollVotes } from '@/models/index.js';
+import { queueLogger } from '../logger.js';
+import { EndedPollNotificationJobData } from '@/queue/types.js';
+import { createNotification } from '@/services/create-notification.js';
+
+const logger = queueLogger.createSubLogger('ended-poll-notification');
+
+export async function endedPollNotification(job: Bull.Job<EndedPollNotificationJobData>, done: any): Promise<void> {
+	const note = await Notes.findOne(job.data.noteId);
+	if (note == null || !note.hasPoll) {
+		done();
+		return;
+	}
+
+	const votes = await PollVotes.createQueryBuilder('vote')
+		.select('vote.userId')
+		.where('vote.noteId = :noteId', { noteId: note.id })
+		.innerJoinAndSelect('vote.user', 'user')
+		.andWhere('user.host IS NULL')
+		.getMany();
+
+	const userIds = [...new Set([note.userId, ...votes.map(v => v.userId)])];
+
+	for (const userId of userIds) {
+		createNotification(userId, 'pollEnded', {
+			noteId: note.id,
+		});
+	}
+
+	done();
+}

packages/backend/src/queue/queues.ts

@@ -1,8 +1,9 @@
 import config from '@/config/index.js';
 import { initialize as initializeQueue } from './initialize.js';
-import { DeliverJobData, InboxJobData, DbJobData, ObjectStorageJobData } from './types.js';
+import { DeliverJobData, InboxJobData, DbJobData, ObjectStorageJobData, EndedPollNotificationJobData } from './types.js';
 
 export const systemQueue = initializeQueue<Record<string, unknown>>('system');
+export const endedPollNotificationQueue = initializeQueue<EndedPollNotificationJobData>('endedPollNotification');
 export const deliverQueue = initializeQueue<DeliverJobData>('deliver', config.deliverJobPerSec || 128);
 export const inboxQueue = initializeQueue<InboxJobData>('inbox', config.inboxJobPerSec || 16);
 export const dbQueue = initializeQueue<DbJobData>('db');

packages/backend/src/queue/types.ts

@@ -1,4 +1,5 @@
 import { DriveFile } from '@/models/entities/drive-file.js';
+import { Note } from '@/models/entities/note';
 import { User } from '@/models/entities/user.js';
 import { IActivity } from '@/remote/activitypub/type.js';
 import httpSignature from 'http-signature';
@@ -41,6 +42,10 @@ export type ObjectStorageFileJobData = {
 	key: string;
 };
 
+export type EndedPollNotificationJobData = {
+	noteId: Note['id'];
+};
+
 export type ThinUser = {
 	id: User['id'];
 };