Add queue types (#7504)

src/queue/index.ts

@@ -1,7 +1,6 @@
 import * as httpSignature from 'http-signature';
 
 import config from '@/config';
-import { User } from '../models/entities/user';
 import { program } from '../argv';
 
 import processDeliver from './processors/deliver';
@@ -11,14 +10,9 @@ import procesObjectStorage from './processors/object-storage';
 import { queueLogger } from './logger';
 import { DriveFile } from '../models/entities/drive-file';
 import { getJobInfo } from './get-job-info';
-import { IActivity } from '../remote/activitypub/type';
 import { dbQueue, deliverQueue, inboxQueue, objectStorageQueue } from './queues';
-
+import { ThinUser } from './types';
-export type InboxJobData = {
+import { IActivity } from '@/remote/activitypub/type';
-	activity: IActivity,
-	/** HTTP-Signature */
-	signature: httpSignature.IParsedSignature
-};
 
 function renderError(e: Error): any {
 	return {
@@ -65,8 +59,9 @@ objectStorageQueue
 	.on('error', (job: any, err: Error) => objectStorageLogger.error(`error ${err}`, { job, e: renderError(err) }))
 	.on('stalled', (job) => objectStorageLogger.warn(`stalled id=${job.id}`));
 
-export function deliver(user: { id: User['id']; host: null; }, content: any, to: any) {
+export function deliver(user: ThinUser, content: unknown, to: string | null) {
 	if (content == null) return null;
+	if (to == null) return null;
 
 	const data = {
 		user,
@@ -85,7 +80,7 @@ export function deliver(user: { id: User['id']; host: null; }, content: any, to:
 	});
 }
 
-export function inbox(activity: any, signature: httpSignature.IParsedSignature) {
+export function inbox(activity: IActivity, signature: httpSignature.IParsedSignature) {
 	const data = {
 		activity: activity,
 		signature
@@ -102,7 +97,7 @@ export function inbox(activity: any, signature: httpSignature.IParsedSignature)
 	});
 }
 
-export function createDeleteDriveFilesJob(user: { id: User['id'] }) {
+export function createDeleteDriveFilesJob(user: ThinUser) {
 	return dbQueue.add('deleteDriveFiles', {
 		user: user
 	}, {
@@ -111,7 +106,7 @@ export function createDeleteDriveFilesJob(user: { id: User['id'] }) {
 	});
 }
 
-export function createExportNotesJob(user: { id: User['id'] }) {
+export function createExportNotesJob(user: ThinUser) {
 	return dbQueue.add('exportNotes', {
 		user: user
 	}, {
@@ -120,7 +115,7 @@ export function createExportNotesJob(user: { id: User['id'] }) {
 	});
 }
 
-export function createExportFollowingJob(user: { id: User['id'] }) {
+export function createExportFollowingJob(user: ThinUser) {
 	return dbQueue.add('exportFollowing', {
 		user: user
 	}, {
@@ -129,7 +124,7 @@ export function createExportFollowingJob(user: { id: User['id'] }) {
 	});
 }
 
-export function createExportMuteJob(user: { id: User['id'] }) {
+export function createExportMuteJob(user: ThinUser) {
 	return dbQueue.add('exportMute', {
 		user: user
 	}, {
@@ -138,7 +133,7 @@ export function createExportMuteJob(user: { id: User['id'] }) {
 	});
 }
 
-export function createExportBlockingJob(user: { id: User['id'] }) {
+export function createExportBlockingJob(user: ThinUser) {
 	return dbQueue.add('exportBlocking', {
 		user: user
 	}, {
@@ -147,7 +142,7 @@ export function createExportBlockingJob(user: { id: User['id'] }) {
 	});
 }
 
-export function createExportUserListsJob(user: { id: User['id'] }) {
+export function createExportUserListsJob(user: ThinUser) {
 	return dbQueue.add('exportUserLists', {
 		user: user
 	}, {
@@ -156,7 +151,7 @@ export function createExportUserListsJob(user: { id: User['id'] }) {
 	});
 }
 
-export function createImportFollowingJob(user: { id: User['id'] }, fileId: DriveFile['id']) {
+export function createImportFollowingJob(user: ThinUser, fileId: DriveFile['id']) {
 	return dbQueue.add('importFollowing', {
 		user: user,
 		fileId: fileId
@@ -166,7 +161,7 @@ export function createImportFollowingJob(user: { id: User['id'] }, fileId: Drive
 	});
 }
 
-export function createImportUserListsJob(user: { id: User['id'] }, fileId: DriveFile['id']) {
+export function createImportUserListsJob(user: ThinUser, fileId: DriveFile['id']) {
 	return dbQueue.add('importUserLists', {
 		user: user,
 		fileId: fileId

src/queue/initialize.ts

@@ -1,8 +1,8 @@
-import * as Queue from 'bull';
+import * as Bull from 'bull';
 import config from '@/config';
 
-export function initialize(name: string, limitPerSec = -1) {
+export function initialize<T>(name: string, limitPerSec = -1) {
-	return new Queue(name, {
+	return new Bull<T>(name, {
 		redis: {
 			port: config.redis.port,
 			host: config.redis.host,

src/queue/processors/db/delete-drive-files.ts

@@ -4,10 +4,11 @@ import { queueLogger } from '../../logger';
 import { deleteFileSync } from '../../../services/drive/delete-file';
 import { Users, DriveFiles } from '../../../models';
 import { MoreThan } from 'typeorm';
+import { DbUserJobData } from '@/queue/types';
 
 const logger = queueLogger.createSubLogger('delete-drive-files');
 
-export async function deleteDriveFiles(job: Bull.Job, done: any): Promise<void> {
+export async function deleteDriveFiles(job: Bull.Job<DbUserJobData>, done: any): Promise<void> {
 	logger.info(`Deleting drive files of ${job.data.user.id} ...`);
 
 	const user = await Users.findOne(job.data.user.id);

src/queue/processors/db/export-blocking.ts

@@ -8,10 +8,11 @@ import dateFormat = require('dateformat');
 import { getFullApAccount } from '@/misc/convert-host';
 import { Users, Blockings } from '../../../models';
 import { MoreThan } from 'typeorm';
+import { DbUserJobData } from '@/queue/types';
 
 const logger = queueLogger.createSubLogger('export-blocking');
 
-export async function exportBlocking(job: Bull.Job, done: any): Promise<void> {
+export async function exportBlocking(job: Bull.Job<DbUserJobData>, done: any): Promise<void> {
 	logger.info(`Exporting blocking of ${job.data.user.id} ...`);
 
 	const user = await Users.findOne(job.data.user.id);
@@ -61,7 +62,7 @@ export async function exportBlocking(job: Bull.Job, done: any): Promise<void> {
 			}
 
 			const content = getFullApAccount(u.username, u.host);
-			await new Promise((res, rej) => {
+			await new Promise<void>((res, rej) => {
 				stream.write(content + '\n', err => {
 					if (err) {
 						logger.error(err);

src/queue/processors/db/export-following.ts

@@ -8,10 +8,11 @@ import dateFormat = require('dateformat');
 import { getFullApAccount } from '@/misc/convert-host';
 import { Users, Followings } from '../../../models';
 import { MoreThan } from 'typeorm';
+import { DbUserJobData } from '@/queue/types';
 
 const logger = queueLogger.createSubLogger('export-following');
 
-export async function exportFollowing(job: Bull.Job, done: any): Promise<void> {
+export async function exportFollowing(job: Bull.Job<DbUserJobData>, done: any): Promise<void> {
 	logger.info(`Exporting following of ${job.data.user.id} ...`);
 
 	const user = await Users.findOne(job.data.user.id);
@@ -61,7 +62,7 @@ export async function exportFollowing(job: Bull.Job, done: any): Promise<void> {
 			}
 
 			const content = getFullApAccount(u.username, u.host);
-			await new Promise((res, rej) => {
+			await new Promise<void>((res, rej) => {
 				stream.write(content + '\n', err => {
 					if (err) {
 						logger.error(err);

src/queue/processors/db/export-mute.ts

@@ -8,10 +8,11 @@ import dateFormat = require('dateformat');
 import { getFullApAccount } from '@/misc/convert-host';
 import { Users, Mutings } from '../../../models';
 import { MoreThan } from 'typeorm';
+import { DbUserJobData } from '@/queue/types';
 
 const logger = queueLogger.createSubLogger('export-mute');
 
-export async function exportMute(job: Bull.Job, done: any): Promise<void> {
+export async function exportMute(job: Bull.Job<DbUserJobData>, done: any): Promise<void> {
 	logger.info(`Exporting mute of ${job.data.user.id} ...`);
 
 	const user = await Users.findOne(job.data.user.id);
@@ -61,7 +62,7 @@ export async function exportMute(job: Bull.Job, done: any): Promise<void> {
 			}
 
 			const content = getFullApAccount(u.username, u.host);
-			await new Promise((res, rej) => {
+			await new Promise<void>((res, rej) => {
 				stream.write(content + '\n', err => {
 					if (err) {
 						logger.error(err);

src/queue/processors/db/export-notes.ts

@@ -9,10 +9,11 @@ import { Users, Notes, Polls } from '../../../models';
 import { MoreThan } from 'typeorm';
 import { Note } from '../../../models/entities/note';
 import { Poll } from '../../../models/entities/poll';
+import { DbUserJobData } from '@/queue/types';
 
 const logger = queueLogger.createSubLogger('export-notes');
 
-export async function exportNotes(job: Bull.Job, done: any): Promise<void> {
+export async function exportNotes(job: Bull.Job<DbUserJobData>, done: any): Promise<void> {
 	logger.info(`Exporting notes of ${job.data.user.id} ...`);
 
 	const user = await Users.findOne(job.data.user.id);
@@ -33,7 +34,7 @@ export async function exportNotes(job: Bull.Job, done: any): Promise<void> {
 
 	const stream = fs.createWriteStream(path, { flags: 'a' });
 
-	await new Promise((res, rej) => {
+	await new Promise<void>((res, rej) => {
 		stream.write('[', err => {
 			if (err) {
 				logger.error(err);
@@ -72,7 +73,7 @@ export async function exportNotes(job: Bull.Job, done: any): Promise<void> {
 				poll = await Polls.findOneOrFail({ noteId: note.id });
 			}
 			const content = JSON.stringify(serialize(note, poll));
-			await new Promise((res, rej) => {
+			await new Promise<void>((res, rej) => {
 				stream.write(exportedNotesCount === 0 ? content : ',\n' + content, err => {
 					if (err) {
 						logger.error(err);
@@ -92,7 +93,7 @@ export async function exportNotes(job: Bull.Job, done: any): Promise<void> {
 		job.progress(exportedNotesCount / total);
 	}
 
-	await new Promise((res, rej) => {
+	await new Promise<void>((res, rej) => {
 		stream.write(']', err => {
 			if (err) {
 				logger.error(err);

src/queue/processors/db/export-user-lists.ts

@@ -8,10 +8,11 @@ import dateFormat = require('dateformat');
 import { getFullApAccount } from '@/misc/convert-host';
 import { Users, UserLists, UserListJoinings } from '../../../models';
 import { In } from 'typeorm';
+import { DbUserJobData } from '@/queue/types';
 
 const logger = queueLogger.createSubLogger('export-user-lists');
 
-export async function exportUserLists(job: Bull.Job, done: any): Promise<void> {
+export async function exportUserLists(job: Bull.Job<DbUserJobData>, done: any): Promise<void> {
 	logger.info(`Exporting user lists of ${job.data.user.id} ...`);
 
 	const user = await Users.findOne(job.data.user.id);
@@ -45,7 +46,7 @@ export async function exportUserLists(job: Bull.Job, done: any): Promise<void> {
 		for (const u of users) {
 			const acct = getFullApAccount(u.username, u.host);
 			const content = `${list.name},${acct}`;
-			await new Promise((res, rej) => {
+			await new Promise<void>((res, rej) => {
 				stream.write(content + '\n', err => {
 					if (err) {
 						logger.error(err);

src/queue/processors/db/import-following.ts

@@ -7,10 +7,11 @@ import { resolveUser } from '../../../remote/resolve-user';
 import { downloadTextFile } from '@/misc/download-text-file';
 import { isSelfHost, toPuny } from '@/misc/convert-host';
 import { Users, DriveFiles } from '../../../models';
+import { DbUserImportJobData } from '@/queue/types';
 
 const logger = queueLogger.createSubLogger('import-following');
 
-export async function importFollowing(job: Bull.Job, done: any): Promise<void> {
+export async function importFollowing(job: Bull.Job<DbUserImportJobData>, done: any): Promise<void> {
 	logger.info(`Importing following of ${job.data.user.id} ...`);
 
 	const user = await Users.findOne(job.data.user.id);

src/queue/processors/db/import-user-lists.ts

@@ -8,10 +8,11 @@ import { downloadTextFile } from '@/misc/download-text-file';
 import { isSelfHost, toPuny } from '@/misc/convert-host';
 import { DriveFiles, Users, UserLists, UserListJoinings } from '../../../models';
 import { genId } from '@/misc/gen-id';
+import { DbUserImportJobData } from '@/queue/types';
 
 const logger = queueLogger.createSubLogger('import-user-lists');
 
-export async function importUserLists(job: Bull.Job, done: any): Promise<void> {
+export async function importUserLists(job: Bull.Job<DbUserImportJobData>, done: any): Promise<void> {
 	logger.info(`Importing user lists of ${job.data.user.id} ...`);
 
 	const user = await Users.findOne(job.data.user.id);

src/queue/processors/db/index.ts

@@ -1,4 +1,5 @@
 import * as Bull from 'bull';
+import { DbJobData } from '@/queue/types';
 import { deleteDriveFiles } from './delete-drive-files';
 import { exportNotes } from './export-notes';
 import { exportFollowing } from './export-following';
@@ -17,10 +18,10 @@ const jobs = {
 	exportUserLists,
 	importFollowing,
 	importUserLists
-} as any;
+} as Record<string, Bull.ProcessCallbackFunction<DbJobData> | Bull.ProcessPromiseFunction<DbJobData>>;
 
-export default function(dbQueue: Bull.Queue) {
+export default function(dbQueue: Bull.Queue<DbJobData>) {
 	for (const [k, v] of Object.entries(jobs)) {
-		dbQueue.process(k, v as any);
+		dbQueue.process(k, v);
 	}
 }

src/queue/processors/deliver.ts

@@ -10,6 +10,7 @@ import { fetchMeta } from '@/misc/fetch-meta';
 import { toPuny } from '@/misc/convert-host';
 import { Cache } from '@/misc/cache';
 import { Instance } from '../../models/entities/instance';
+import { DeliverJobData } from '../types';
 
 const logger = new Logger('deliver');
 
@@ -17,7 +18,7 @@ let latest: string | null = null;
 
 const suspendedHostsCache = new Cache<Instance[]>(1000 * 60 * 60);
 
-export default async (job: Bull.Job) => {
+export default async (job: Bull.Job<DeliverJobData>) => {
 	const { host } = new URL(job.data.to);
 
 	// ブロックしてたら中断

src/queue/processors/inbox.ts

@@ -10,7 +10,7 @@ import { fetchMeta } from '@/misc/fetch-meta';
 import { toPuny, extractDbHost } from '@/misc/convert-host';
 import { getApId } from '../../remote/activitypub/type';
 import { fetchInstanceMetadata } from '../../services/fetch-instance-metadata';
-import { InboxJobData } from '..';
+import { InboxJobData } from '../types';
 import DbResolver from '../../remote/activitypub/db-resolver';
 import { resolvePerson } from '../../remote/activitypub/models/person';
 import { LdSignature } from '../../remote/activitypub/misc/ld-signature';
@@ -23,7 +23,7 @@ export default async (job: Bull.Job<InboxJobData>): Promise<string> => {
 	const activity = job.data.activity;
 
 	//#region Log
-	const info = Object.assign({}, activity);
+	const info = Object.assign({}, activity) as any;
 	delete info['@context'];
 	logger.debug(JSON.stringify(info, null, 2));
 	//#endregion

src/queue/processors/object-storage/clean-remote-files.ts

@@ -7,7 +7,7 @@ import { MoreThan, Not, IsNull } from 'typeorm';
 
 const logger = queueLogger.createSubLogger('clean-remote-files');
 
-export default async function cleanRemoteFiles(job: Bull.Job, done: any): Promise<void> {
+export default async function cleanRemoteFiles(job: Bull.Job<{}>, done: any): Promise<void> {
 	logger.info(`Deleting cached remote files...`);
 
 	let deletedCount = 0;

src/queue/processors/object-storage/delete-file.ts

@@ -1,7 +1,8 @@
+import { ObjectStorageFileJobData } from '@/queue/types';
 import * as Bull from 'bull';
 import { deleteObjectStorageFile } from '../../../services/drive/delete-file';
 
-export default async (job: Bull.Job) => {
+export default async (job: Bull.Job<ObjectStorageFileJobData>) => {
 	const key: string = job.data.key;
 
 	await deleteObjectStorageFile(key);

src/queue/processors/object-storage/index.ts

@@ -1,14 +1,15 @@
 import * as Bull from 'bull';
+import { ObjectStorageJobData } from '@/queue/types';
 import deleteFile from './delete-file';
 import cleanRemoteFiles from './clean-remote-files';
 
 const jobs = {
 	deleteFile,
 	cleanRemoteFiles,
-} as any;
+} as Record<string, Bull.ProcessCallbackFunction<ObjectStorageJobData> | Bull.ProcessPromiseFunction<ObjectStorageJobData>>;
 
 export default function(q: Bull.Queue) {
 	for (const [k, v] of Object.entries(jobs)) {
-		q.process(k, 16, v as any);
+		q.process(k, 16, v);
 	}
 }

src/queue/queues.ts

@@ -1,7 +1,8 @@
 import config from '@/config';
 import { initialize as initializeQueue } from './initialize';
+import { DeliverJobData, InboxJobData, DbJobData, ObjectStorageJobData } from './types';
 
-export const deliverQueue = initializeQueue('deliver', config.deliverJobPerSec || 128);
+export const deliverQueue = initializeQueue<DeliverJobData>('deliver', config.deliverJobPerSec || 128);
-export const inboxQueue = initializeQueue('inbox', config.inboxJobPerSec || 16);
+export const inboxQueue = initializeQueue<InboxJobData>('inbox', config.inboxJobPerSec || 16);
-export const dbQueue = initializeQueue('db');
+export const dbQueue = initializeQueue<DbJobData>('db');
-export const objectStorageQueue = initializeQueue('objectStorage');
+export const objectStorageQueue = initializeQueue<ObjectStorageJobData>('objectStorage');

src/queue/types.ts

@@ -0,0 +1,39 @@
+import { DriveFile } from '@/models/entities/drive-file';
+import { User } from '@/models/entities/user';
+import { IActivity } from '@/remote/activitypub/type';
+import * as httpSignature from 'http-signature';
+
+export type DeliverJobData = {
+	/** Actor */
+	user: ThinUser;
+	/** Activity */
+	content: unknown;
+	/** inbox URL to deliver */
+	to: string;
+};
+
+export type InboxJobData = {
+	activity: IActivity;
+	signature: httpSignature.IParsedSignature;
+};
+
+export type DbJobData = DbUserJobData | DbUserImportJobData;
+
+export type DbUserJobData = {
+	user: ThinUser;
+};
+
+export type DbUserImportJobData = {
+	user: ThinUser;
+	fileId: DriveFile['id'];
+};
+
+export type ObjectStorageJobData = ObjectStorageFileJobData | {};
+
+export type ObjectStorageFileJobData = {
+	key: string;
+};
+
+export type ThinUser = {
+	id: User['id'];
+};