summaryrefslogtreecommitdiff
path: root/src/queue/processors
diff options
context:
space:
mode:
authorMeiMei <30769358+mei23@users.noreply.github.com>2021-05-08 18:56:21 +0900
committerGitHub <noreply@github.com>2021-05-08 18:56:21 +0900
commit591a5c277cdf197f043e5b24743dbdb346e65c4f (patch)
treeb84126eec671a7a1f1172b472cd9143060357f00 /src/queue/processors
parentCreate get-index-stats.ts (diff)
downloadsharkey-591a5c277cdf197f043e5b24743dbdb346e65c4f.tar.gz
sharkey-591a5c277cdf197f043e5b24743dbdb346e65c4f.tar.bz2
sharkey-591a5c277cdf197f043e5b24743dbdb346e65c4f.zip
Add queue types (#7504)
Diffstat (limited to 'src/queue/processors')
-rw-r--r--src/queue/processors/db/delete-drive-files.ts3
-rw-r--r--src/queue/processors/db/export-blocking.ts5
-rw-r--r--src/queue/processors/db/export-following.ts5
-rw-r--r--src/queue/processors/db/export-mute.ts5
-rw-r--r--src/queue/processors/db/export-notes.ts9
-rw-r--r--src/queue/processors/db/export-user-lists.ts5
-rw-r--r--src/queue/processors/db/import-following.ts3
-rw-r--r--src/queue/processors/db/import-user-lists.ts3
-rw-r--r--src/queue/processors/db/index.ts7
-rw-r--r--src/queue/processors/deliver.ts3
-rw-r--r--src/queue/processors/inbox.ts4
-rw-r--r--src/queue/processors/object-storage/clean-remote-files.ts2
-rw-r--r--src/queue/processors/object-storage/delete-file.ts3
-rw-r--r--src/queue/processors/object-storage/index.ts5
14 files changed, 37 insertions, 25 deletions
diff --git a/src/queue/processors/db/delete-drive-files.ts b/src/queue/processors/db/delete-drive-files.ts
index a2fd9050a9..874623204b 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<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);
diff --git a/src/queue/processors/db/export-blocking.ts b/src/queue/processors/db/export-blocking.ts
index 9bbc9b2f12..001b50a22c 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<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);
diff --git a/src/queue/processors/db/export-following.ts b/src/queue/processors/db/export-following.ts
index 79df2298d8..c1ccb7af4c 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<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);
diff --git a/src/queue/processors/db/export-mute.ts b/src/queue/processors/db/export-mute.ts
index c10556f882..55d45cc29c 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<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);
diff --git a/src/queue/processors/db/export-notes.ts b/src/queue/processors/db/export-notes.ts
index f76a47aacd..2d09c0d201 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<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);
diff --git a/src/queue/processors/db/export-user-lists.ts b/src/queue/processors/db/export-user-lists.ts
index b6c527fc78..3f793e064f 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<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);
diff --git a/src/queue/processors/db/import-following.ts b/src/queue/processors/db/import-following.ts
index 554337849e..55c0aaa9f9 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<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);
diff --git a/src/queue/processors/db/import-user-lists.ts b/src/queue/processors/db/import-user-lists.ts
index 2fe023da7d..d316b95ff0 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<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);
diff --git a/src/queue/processors/db/index.ts b/src/queue/processors/db/index.ts
index 921cdf7ab1..b56b7bfa2c 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<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);
}
}
diff --git a/src/queue/processors/deliver.ts b/src/queue/processors/deliver.ts
index b167154fcd..f9c53fc8f1 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<Instance[]>(1000 * 60 * 60);
-export default async (job: Bull.Job) => {
+export default async (job: Bull.Job<DeliverJobData>) => {
const { host } = new URL(job.data.to);
// ブロックしてたら中断
diff --git a/src/queue/processors/inbox.ts b/src/queue/processors/inbox.ts
index 7c746eb25d..2ef19777f1 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<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
diff --git a/src/queue/processors/object-storage/clean-remote-files.ts b/src/queue/processors/object-storage/clean-remote-files.ts
index 7b34892e1f..a922755f4d 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<void> {
+export default async function cleanRemoteFiles(job: Bull.Job<{}>, done: any): Promise<void> {
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 f899df7d2e..31050998af 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<ObjectStorageFileJobData>) => {
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 33ef665b38..0d9570e179 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<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);
}
}