From 0965a23c03f2d53db203ea9d9bda264b06ebd238 Mon Sep 17 00:00:00 2001 From: MeiMei <30769358+mei23@users.noreply.github.com> Date: Sat, 8 May 2021 18:56:21 +0900 Subject: [PATCH] Add queue types (#7504) --- src/queue/index.ts | 31 +++++++-------- src/queue/initialize.ts | 6 +-- src/queue/processors/db/delete-drive-files.ts | 3 +- src/queue/processors/db/export-blocking.ts | 5 ++- src/queue/processors/db/export-following.ts | 5 ++- src/queue/processors/db/export-mute.ts | 5 ++- src/queue/processors/db/export-notes.ts | 9 +++-- src/queue/processors/db/export-user-lists.ts | 5 ++- src/queue/processors/db/import-following.ts | 3 +- src/queue/processors/db/import-user-lists.ts | 3 +- src/queue/processors/db/index.ts | 7 ++-- src/queue/processors/deliver.ts | 3 +- src/queue/processors/inbox.ts | 4 +- .../object-storage/clean-remote-files.ts | 2 +- .../processors/object-storage/delete-file.ts | 3 +- src/queue/processors/object-storage/index.ts | 5 ++- src/queue/queues.ts | 9 +++-- src/queue/types.ts | 39 +++++++++++++++++++ 18 files changed, 97 insertions(+), 50 deletions(-) create mode 100644 src/queue/types.ts diff --git a/src/queue/index.ts b/src/queue/index.ts index 201b66379..01dd020d9 100644 --- a/src/queue/index.ts +++ b/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'; - -export type InboxJobData = { - activity: IActivity, - /** HTTP-Signature */ - signature: httpSignature.IParsedSignature -}; +import { ThinUser } from './types'; +import { IActivity } from '@/remote/activitypub/type'; 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 diff --git a/src/queue/initialize.ts b/src/queue/initialize.ts index 941fe4bc3..4c0e5f9d8 100644 --- a/src/queue/initialize.ts +++ b/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) { - return new Queue(name, { +export function initialize(name: string, limitPerSec = -1) { + return new Bull(name, { redis: { port: config.redis.port, host: config.redis.host, diff --git a/src/queue/processors/db/delete-drive-files.ts b/src/queue/processors/db/delete-drive-files.ts index a2fd9050a..874623204 100644 --- a/src/queue/processors/db/delete-drive-files.ts +++ b/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 { +export async function deleteDriveFiles(job: Bull.Job, done: any): Promise { logger.info(`Deleting drive files of ${job.data.user.id} ...`); const user = await Users.findOne(job.data.user.id); diff --git a/src/queue/processors/db/export-blocking.ts b/src/queue/processors/db/export-blocking.ts index 9bbc9b2f1..001b50a22 100644 --- a/src/queue/processors/db/export-blocking.ts +++ b/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 { +export async function exportBlocking(job: Bull.Job, done: any): Promise { 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 { } const content = getFullApAccount(u.username, u.host); - await new Promise((res, rej) => { + await new Promise((res, rej) => { stream.write(content + '\n', err => { if (err) { logger.error(err); diff --git a/src/queue/processors/db/export-following.ts b/src/queue/processors/db/export-following.ts index 79df2298d..c1ccb7af4 100644 --- a/src/queue/processors/db/export-following.ts +++ b/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 { +export async function exportFollowing(job: Bull.Job, done: any): Promise { 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 { } const content = getFullApAccount(u.username, u.host); - await new Promise((res, rej) => { + await new Promise((res, rej) => { stream.write(content + '\n', err => { if (err) { logger.error(err); diff --git a/src/queue/processors/db/export-mute.ts b/src/queue/processors/db/export-mute.ts index c10556f88..55d45cc29 100644 --- a/src/queue/processors/db/export-mute.ts +++ b/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 { +export async function exportMute(job: Bull.Job, done: any): Promise { 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 { } const content = getFullApAccount(u.username, u.host); - await new Promise((res, rej) => { + await new Promise((res, rej) => { stream.write(content + '\n', err => { if (err) { logger.error(err); diff --git a/src/queue/processors/db/export-notes.ts b/src/queue/processors/db/export-notes.ts index f76a47aac..2d09c0d20 100644 --- a/src/queue/processors/db/export-notes.ts +++ b/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 { +export async function exportNotes(job: Bull.Job, done: any): Promise { 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 { const stream = fs.createWriteStream(path, { flags: 'a' }); - await new Promise((res, rej) => { + await new Promise((res, rej) => { stream.write('[', err => { if (err) { logger.error(err); @@ -72,7 +73,7 @@ export async function exportNotes(job: Bull.Job, done: any): Promise { poll = await Polls.findOneOrFail({ noteId: note.id }); } const content = JSON.stringify(serialize(note, poll)); - await new Promise((res, rej) => { + await new Promise((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 { job.progress(exportedNotesCount / total); } - await new Promise((res, rej) => { + await new Promise((res, rej) => { stream.write(']', err => { if (err) { logger.error(err); diff --git a/src/queue/processors/db/export-user-lists.ts b/src/queue/processors/db/export-user-lists.ts index b6c527fc7..3f793e064 100644 --- a/src/queue/processors/db/export-user-lists.ts +++ b/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 { +export async function exportUserLists(job: Bull.Job, done: any): Promise { 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 { 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((res, rej) => { stream.write(content + '\n', err => { if (err) { logger.error(err); diff --git a/src/queue/processors/db/import-following.ts b/src/queue/processors/db/import-following.ts index 554337849..55c0aaa9f 100644 --- a/src/queue/processors/db/import-following.ts +++ b/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 { +export async function importFollowing(job: Bull.Job, done: any): Promise { logger.info(`Importing following of ${job.data.user.id} ...`); const user = await Users.findOne(job.data.user.id); diff --git a/src/queue/processors/db/import-user-lists.ts b/src/queue/processors/db/import-user-lists.ts index 2fe023da7..d316b95ff 100644 --- a/src/queue/processors/db/import-user-lists.ts +++ b/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 { +export async function importUserLists(job: Bull.Job, done: any): Promise { logger.info(`Importing user lists of ${job.data.user.id} ...`); const user = await Users.findOne(job.data.user.id); diff --git a/src/queue/processors/db/index.ts b/src/queue/processors/db/index.ts index 921cdf7ab..b56b7bfa2 100644 --- a/src/queue/processors/db/index.ts +++ b/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 | Bull.ProcessPromiseFunction>; -export default function(dbQueue: Bull.Queue) { +export default function(dbQueue: Bull.Queue) { for (const [k, v] of Object.entries(jobs)) { - dbQueue.process(k, v as any); + dbQueue.process(k, v); } } diff --git a/src/queue/processors/deliver.ts b/src/queue/processors/deliver.ts index b167154fc..f9c53fc8f 100644 --- a/src/queue/processors/deliver.ts +++ b/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(1000 * 60 * 60); -export default async (job: Bull.Job) => { +export default async (job: Bull.Job) => { const { host } = new URL(job.data.to); // ブロックしてたら中断 diff --git a/src/queue/processors/inbox.ts b/src/queue/processors/inbox.ts index 7c746eb25..2ef19777f 100644 --- a/src/queue/processors/inbox.ts +++ b/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): Promise => { 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 diff --git a/src/queue/processors/object-storage/clean-remote-files.ts b/src/queue/processors/object-storage/clean-remote-files.ts index 7b34892e1..a922755f4 100644 --- a/src/queue/processors/object-storage/clean-remote-files.ts +++ b/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 { +export default async function cleanRemoteFiles(job: Bull.Job<{}>, done: any): Promise { logger.info(`Deleting cached remote files...`); let deletedCount = 0; diff --git a/src/queue/processors/object-storage/delete-file.ts b/src/queue/processors/object-storage/delete-file.ts index f899df7d2..31050998a 100644 --- a/src/queue/processors/object-storage/delete-file.ts +++ b/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) => { const key: string = job.data.key; await deleteObjectStorageFile(key); diff --git a/src/queue/processors/object-storage/index.ts b/src/queue/processors/object-storage/index.ts index 33ef665b3..0d9570e17 100644 --- a/src/queue/processors/object-storage/index.ts +++ b/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 | Bull.ProcessPromiseFunction>; 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); } } diff --git a/src/queue/queues.ts b/src/queue/queues.ts index 819bcf31d..5e2754b83 100644 --- a/src/queue/queues.ts +++ b/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 inboxQueue = initializeQueue('inbox', config.inboxJobPerSec || 16); -export const dbQueue = initializeQueue('db'); -export const objectStorageQueue = initializeQueue('objectStorage'); +export const deliverQueue = initializeQueue('deliver', config.deliverJobPerSec || 128); +export const inboxQueue = initializeQueue('inbox', config.inboxJobPerSec || 16); +export const dbQueue = initializeQueue('db'); +export const objectStorageQueue = initializeQueue('objectStorage'); diff --git a/src/queue/types.ts b/src/queue/types.ts new file mode 100644 index 000000000..a782fc6b9 --- /dev/null +++ b/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']; +};