wip
This commit is contained in:
parent
9dba657ffa
commit
bf2cb2d0b7
|
@ -16,6 +16,7 @@ import { IPoll } from '@/models/Poll.js';
|
|||
import { IdentifiableError } from '@/misc/identifiable-error.js';
|
||||
import { isRenote, isQuote } from '@/misc/is-renote.js';
|
||||
import { NoteEntityService } from '@/core/entities/NoteEntityService.js';
|
||||
import { QueueService } from '@/core/QueueService.js';
|
||||
|
||||
export type NoteDraftOptions = {
|
||||
replyId?: MiNote['id'] | null;
|
||||
|
@ -56,6 +57,7 @@ export class NoteDraftService {
|
|||
private roleService: RoleService,
|
||||
private idService: IdService,
|
||||
private noteEntityService: NoteEntityService,
|
||||
private queueService: QueueService,
|
||||
) {
|
||||
}
|
||||
|
||||
|
@ -311,4 +313,22 @@ export class NoteDraftService {
|
|||
|
||||
return appliedDraft;
|
||||
}
|
||||
|
||||
@bindThis
|
||||
public async schedule(draft: MiNoteDraft, scheduledAt: Date): Promise<void> {
|
||||
const delay = scheduledAt.getTime() - Date.now();
|
||||
this.queueService.deleteUserMutingQueue.add(draft.id, {
|
||||
noteDraftId: draft.id,
|
||||
}, {
|
||||
delay,
|
||||
removeOnComplete: {
|
||||
age: 3600 * 24 * 7, // keep up to 7 days
|
||||
count: 30,
|
||||
},
|
||||
removeOnFail: {
|
||||
age: 3600 * 24 * 7, // keep up to 7 days
|
||||
count: 100,
|
||||
},
|
||||
});
|
||||
}
|
||||
}
|
||||
|
|
|
@ -16,11 +16,13 @@ import {
|
|||
RelationshipJobData,
|
||||
UserWebhookDeliverJobData,
|
||||
SystemWebhookDeliverJobData,
|
||||
PostScheduledNoteJobData,
|
||||
} from '../queue/types.js';
|
||||
import type { Provider } from '@nestjs/common';
|
||||
|
||||
export type SystemQueue = Bull.Queue<Record<string, unknown>>;
|
||||
export type EndedPollNotificationQueue = Bull.Queue<EndedPollNotificationJobData>;
|
||||
export type PostScheduledNoteQueue = Bull.Queue<PostScheduledNoteJobData>;
|
||||
export type DeliverQueue = Bull.Queue<DeliverJobData>;
|
||||
export type InboxQueue = Bull.Queue<InboxJobData>;
|
||||
export type DbQueue = Bull.Queue;
|
||||
|
@ -41,6 +43,12 @@ const $endedPollNotification: Provider = {
|
|||
inject: [DI.config],
|
||||
};
|
||||
|
||||
const $postScheduledNote: Provider = {
|
||||
provide: 'queue:postScheduledNote',
|
||||
useFactory: (config: Config) => new Bull.Queue(QUEUE.POST_SCHEDULED_NOTE, baseQueueOptions(config, QUEUE.POST_SCHEDULED_NOTE)),
|
||||
inject: [DI.config],
|
||||
};
|
||||
|
||||
const $deliver: Provider = {
|
||||
provide: 'queue:deliver',
|
||||
useFactory: (config: Config) => new Bull.Queue(QUEUE.DELIVER, baseQueueOptions(config, QUEUE.DELIVER)),
|
||||
|
@ -89,6 +97,7 @@ const $systemWebhookDeliver: Provider = {
|
|||
providers: [
|
||||
$system,
|
||||
$endedPollNotification,
|
||||
$postScheduledNote,
|
||||
$deliver,
|
||||
$inbox,
|
||||
$db,
|
||||
|
@ -100,6 +109,7 @@ const $systemWebhookDeliver: Provider = {
|
|||
exports: [
|
||||
$system,
|
||||
$endedPollNotification,
|
||||
$postScheduledNote,
|
||||
$deliver,
|
||||
$inbox,
|
||||
$db,
|
||||
|
@ -113,6 +123,7 @@ export class QueueModule implements OnApplicationShutdown {
|
|||
constructor(
|
||||
@Inject('queue:system') public systemQueue: SystemQueue,
|
||||
@Inject('queue:endedPollNotification') public endedPollNotificationQueue: EndedPollNotificationQueue,
|
||||
@Inject('queue:postScheduledNote') public postScheduledNoteQueue: PostScheduledNoteQueue,
|
||||
@Inject('queue:deliver') public deliverQueue: DeliverQueue,
|
||||
@Inject('queue:inbox') public inboxQueue: InboxQueue,
|
||||
@Inject('queue:db') public dbQueue: DbQueue,
|
||||
|
@ -129,6 +140,7 @@ export class QueueModule implements OnApplicationShutdown {
|
|||
await Promise.all([
|
||||
this.systemQueue.close(),
|
||||
this.endedPollNotificationQueue.close(),
|
||||
this.postScheduledNoteQueue.close(),
|
||||
this.deliverQueue.close(),
|
||||
this.inboxQueue.close(),
|
||||
this.dbQueue.close(),
|
||||
|
|
|
@ -31,6 +31,7 @@ import type {
|
|||
DbQueue,
|
||||
DeliverQueue,
|
||||
EndedPollNotificationQueue,
|
||||
PostScheduledNoteQueue,
|
||||
InboxQueue,
|
||||
ObjectStorageQueue,
|
||||
RelationshipQueue,
|
||||
|
@ -44,6 +45,7 @@ import type * as Bull from 'bullmq';
|
|||
export const QUEUE_TYPES = [
|
||||
'system',
|
||||
'endedPollNotification',
|
||||
'postScheduledNote',
|
||||
'deliver',
|
||||
'inbox',
|
||||
'db',
|
||||
|
@ -92,6 +94,7 @@ export class QueueService {
|
|||
|
||||
@Inject('queue:system') public systemQueue: SystemQueue,
|
||||
@Inject('queue:endedPollNotification') public endedPollNotificationQueue: EndedPollNotificationQueue,
|
||||
@Inject('queue:postScheduledNote') public postScheduledNoteQueue: PostScheduledNoteQueue,
|
||||
@Inject('queue:deliver') public deliverQueue: DeliverQueue,
|
||||
@Inject('queue:inbox') public inboxQueue: InboxQueue,
|
||||
@Inject('queue:db') public dbQueue: DbQueue,
|
||||
|
@ -717,6 +720,7 @@ export class QueueService {
|
|||
switch (type) {
|
||||
case 'system': return this.systemQueue;
|
||||
case 'endedPollNotification': return this.endedPollNotificationQueue;
|
||||
case 'postScheduledNote': return this.postScheduledNoteQueue;
|
||||
case 'deliver': return this.deliverQueue;
|
||||
case 'inbox': return this.inboxQueue;
|
||||
case 'db': return this.dbQueue;
|
||||
|
|
|
@ -10,6 +10,7 @@ import { QueueLoggerService } from './QueueLoggerService.js';
|
|||
import { QueueProcessorService } from './QueueProcessorService.js';
|
||||
import { DeliverProcessorService } from './processors/DeliverProcessorService.js';
|
||||
import { EndedPollNotificationProcessorService } from './processors/EndedPollNotificationProcessorService.js';
|
||||
import { PostScheduledNoteProcessorService } from './processors/PostScheduledNoteProcessorService.js';
|
||||
import { InboxProcessorService } from './processors/InboxProcessorService.js';
|
||||
import { UserWebhookDeliverProcessorService } from './processors/UserWebhookDeliverProcessorService.js';
|
||||
import { SystemWebhookDeliverProcessorService } from './processors/SystemWebhookDeliverProcessorService.js';
|
||||
|
@ -79,6 +80,7 @@ import { RelationshipProcessorService } from './processors/RelationshipProcessor
|
|||
UserWebhookDeliverProcessorService,
|
||||
SystemWebhookDeliverProcessorService,
|
||||
EndedPollNotificationProcessorService,
|
||||
PostScheduledNoteProcessorService,
|
||||
DeliverProcessorService,
|
||||
InboxProcessorService,
|
||||
AggregateRetentionProcessorService,
|
||||
|
|
|
@ -14,6 +14,7 @@ import { CheckModeratorsActivityProcessorService } from '@/queue/processors/Chec
|
|||
import { UserWebhookDeliverProcessorService } from './processors/UserWebhookDeliverProcessorService.js';
|
||||
import { SystemWebhookDeliverProcessorService } from './processors/SystemWebhookDeliverProcessorService.js';
|
||||
import { EndedPollNotificationProcessorService } from './processors/EndedPollNotificationProcessorService.js';
|
||||
import { PostScheduledNoteProcessorService } from './processors/PostScheduledNoteProcessorService.js';
|
||||
import { DeliverProcessorService } from './processors/DeliverProcessorService.js';
|
||||
import { InboxProcessorService } from './processors/InboxProcessorService.js';
|
||||
import { DeleteDriveFilesProcessorService } from './processors/DeleteDriveFilesProcessorService.js';
|
||||
|
@ -85,6 +86,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
|
|||
private relationshipQueueWorker: Bull.Worker;
|
||||
private objectStorageQueueWorker: Bull.Worker;
|
||||
private endedPollNotificationQueueWorker: Bull.Worker;
|
||||
private postScheduledNoteQueueWorker: Bull.Worker;
|
||||
|
||||
constructor(
|
||||
@Inject(DI.config)
|
||||
|
@ -94,6 +96,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
|
|||
private userWebhookDeliverProcessorService: UserWebhookDeliverProcessorService,
|
||||
private systemWebhookDeliverProcessorService: SystemWebhookDeliverProcessorService,
|
||||
private endedPollNotificationProcessorService: EndedPollNotificationProcessorService,
|
||||
private postScheduledNoteProcessorService: PostScheduledNoteProcessorService,
|
||||
private deliverProcessorService: DeliverProcessorService,
|
||||
private inboxProcessorService: InboxProcessorService,
|
||||
private deleteDriveFilesProcessorService: DeleteDriveFilesProcessorService,
|
||||
|
@ -520,6 +523,21 @@ export class QueueProcessorService implements OnApplicationShutdown {
|
|||
});
|
||||
}
|
||||
//#endregion
|
||||
|
||||
//#region post scheduled note
|
||||
{
|
||||
this.postScheduledNoteQueueWorker = new Bull.Worker(QUEUE.POST_SCHEDULED_NOTE, async (job) => {
|
||||
if (this.config.sentryForBackend) {
|
||||
return Sentry.startSpan({ name: 'Queue: PostScheduledNote' }, () => this.postScheduledNoteProcessorService.process(job));
|
||||
} else {
|
||||
return this.postScheduledNoteProcessorService.process(job);
|
||||
}
|
||||
}, {
|
||||
...baseWorkerOptions(this.config, QUEUE.POST_SCHEDULED_NOTE),
|
||||
autorun: false,
|
||||
});
|
||||
}
|
||||
//#endregion
|
||||
}
|
||||
|
||||
@bindThis
|
||||
|
@ -534,6 +552,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
|
|||
this.relationshipQueueWorker.run(),
|
||||
this.objectStorageQueueWorker.run(),
|
||||
this.endedPollNotificationQueueWorker.run(),
|
||||
this.postScheduledNoteQueueWorker.run(),
|
||||
]);
|
||||
}
|
||||
|
||||
|
@ -549,6 +568,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
|
|||
this.relationshipQueueWorker.close(),
|
||||
this.objectStorageQueueWorker.close(),
|
||||
this.endedPollNotificationQueueWorker.close(),
|
||||
this.postScheduledNoteQueueWorker.close(),
|
||||
]);
|
||||
}
|
||||
|
||||
|
|
|
@ -12,6 +12,7 @@ export const QUEUE = {
|
|||
INBOX: 'inbox',
|
||||
SYSTEM: 'system',
|
||||
ENDED_POLL_NOTIFICATION: 'endedPollNotification',
|
||||
POST_SCHEDULED_NOTE: 'postScheduledNote',
|
||||
DB: 'db',
|
||||
RELATIONSHIP: 'relationship',
|
||||
OBJECT_STORAGE: 'objectStorage',
|
||||
|
|
|
@ -0,0 +1,41 @@
|
|||
/*
|
||||
* SPDX-FileCopyrightText: syuilo and misskey-project
|
||||
* SPDX-License-Identifier: AGPL-3.0-only
|
||||
*/
|
||||
|
||||
import { Inject, Injectable } from '@nestjs/common';
|
||||
import { DI } from '@/di-symbols.js';
|
||||
import type { NoteDraftsRepository } from '@/models/_.js';
|
||||
import type Logger from '@/logger.js';
|
||||
import { NotificationService } from '@/core/NotificationService.js';
|
||||
import { bindThis } from '@/decorators.js';
|
||||
import { QueueLoggerService } from '../QueueLoggerService.js';
|
||||
import type * as Bull from 'bullmq';
|
||||
import type { PostScheduledNoteJobData } from '../types.js';
|
||||
|
||||
@Injectable()
|
||||
export class PostScheduledNoteProcessorService {
|
||||
private logger: Logger;
|
||||
|
||||
constructor(
|
||||
@Inject(DI.noteDraftsRepository)
|
||||
private noteDraftsRepository: NoteDraftsRepository,
|
||||
|
||||
private notificationService: NotificationService,
|
||||
private queueLoggerService: QueueLoggerService,
|
||||
) {
|
||||
this.logger = this.queueLoggerService.logger.createSubLogger('post-scheduled-note');
|
||||
}
|
||||
|
||||
@bindThis
|
||||
public async process(job: Bull.Job<PostScheduledNoteJobData>): Promise<void> {
|
||||
const draft = await this.noteDraftsRepository.findOneBy({ id: job.data.noteDraftId });
|
||||
if (draft == null) {
|
||||
return;
|
||||
}
|
||||
|
||||
this.notificationService.createNotification(draft.userId, 'scheduledNotePosted', {
|
||||
noteId: note.id,
|
||||
});
|
||||
}
|
||||
}
|
|
@ -109,6 +109,10 @@ export type EndedPollNotificationJobData = {
|
|||
noteId: MiNote['id'];
|
||||
};
|
||||
|
||||
export type PostScheduledNoteJobData = {
|
||||
noteDraftId: string;
|
||||
};
|
||||
|
||||
export type SystemWebhookDeliverJobData<T extends SystemWebhookEventType = SystemWebhookEventType> = {
|
||||
type: T;
|
||||
content: SystemWebhookPayload<T>;
|
||||
|
|
|
@ -5,7 +5,7 @@
|
|||
|
||||
import { Inject, Injectable } from '@nestjs/common';
|
||||
import { Endpoint } from '@/server/api/endpoint-base.js';
|
||||
import type { DbQueue, DeliverQueue, EndedPollNotificationQueue, InboxQueue, ObjectStorageQueue, SystemQueue, UserWebhookDeliverQueue, SystemWebhookDeliverQueue } from '@/core/QueueModule.js';
|
||||
import type { DbQueue, DeliverQueue, EndedPollNotificationQueue, PostScheduledNoteQueue, InboxQueue, ObjectStorageQueue, SystemQueue, UserWebhookDeliverQueue, SystemWebhookDeliverQueue } from '@/core/QueueModule.js';
|
||||
|
||||
export const meta = {
|
||||
tags: ['admin'],
|
||||
|
@ -49,6 +49,7 @@ export default class extends Endpoint<typeof meta, typeof paramDef> { // eslint-
|
|||
constructor(
|
||||
@Inject('queue:system') public systemQueue: SystemQueue,
|
||||
@Inject('queue:endedPollNotification') public endedPollNotificationQueue: EndedPollNotificationQueue,
|
||||
@Inject('postScheduledNote') public postScheduledNoteQueue: PostScheduledNoteQueue,
|
||||
@Inject('queue:deliver') public deliverQueue: DeliverQueue,
|
||||
@Inject('queue:inbox') public inboxQueue: InboxQueue,
|
||||
@Inject('queue:db') public dbQueue: DbQueue,
|
||||
|
|
|
@ -233,6 +233,7 @@ export const rolePolicies = [
|
|||
export const queueTypes = [
|
||||
'system',
|
||||
'endedPollNotification',
|
||||
'postScheduledNote',
|
||||
'deliver',
|
||||
'inbox',
|
||||
'db',
|
||||
|
|
Loading…
Reference in New Issue