limepotato/jormungandr-bite
1
0
Fork 0
mirror of https://iceshrimp.dev/limepotato/jormungandr-bite.git synced 2024-11-13 13:37:31 -07:00
Code Issues Projects Releases Packages Wiki Activity Actions

Add queue types (#7504)

Browse source
This commit is contained in:
MeiMei 2021-05-08 18:56:21 +09:00 committed by GitHub
parent befe913517
commit 0965a23c03
18 changed files with 97 additions and 50 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

31
src/queue/index.ts
View file

@ -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

6
src/queue/initialize.ts
View file

@ -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<T>(name: string, limitPerSec = -1) {
return new Bull<T>(name, {
redis: {
port: config.redis.port,
host: config.redis.host,

3
src/queue/processors/db/delete-drive-files.ts
View file

@ -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);

5
src/queue/processors/db/export-blocking.ts
View file

@ -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);

5
src/queue/processors/db/export-following.ts
View file

@ -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);

5
src/queue/processors/db/export-mute.ts
View file

@ -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);

9
src/queue/processors/db/export-notes.ts
View file

@ -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);

5
src/queue/processors/db/export-user-lists.ts
View file

@ -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);

3
src/queue/processors/db/import-following.ts
View file

@ -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);

3
src/queue/processors/db/import-user-lists.ts
View file

@ -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);

7
src/queue/processors/db/index.ts
View file

@ -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);
}
}

3
src/queue/processors/deliver.ts
View file

@ -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);
// ブロックしてたら中断

4
src/queue/processors/inbox.ts
View file

@ -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

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

@ -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;

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

@ -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);

5
src/queue/processors/object-storage/index.ts
View file

@ -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);
}
}

9
src/queue/queues.ts
View file

@ -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<DeliverJobData>('deliver', config.deliverJobPerSec || 128);
export const inboxQueue = initializeQueue<InboxJobData>('inbox', config.inboxJobPerSec || 16);
export const dbQueue = initializeQueue<DbJobData>('db');
export const objectStorageQueue = initializeQueue<ObjectStorageJobData>('objectStorage');

39
src/queue/types.ts Normal file
View file

@ -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'];
};