Press n or j to go to the next uncovered block, b, p or k for the previous block.
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 | 58x 58x 58x 58x 58x 58x 58x 58x 88x 82x 88x 83x 154x 20x 140x 137x 467x 137x 467x 467x 467x 88x 20x 20x 10x 58x 5x 72x | import { queueAsPromised, promise as queuePromise } from "fastq";
import { QueueError } from "shared/types/api/errors";
import {
Result,
makeErrorResult,
makeSuccessResult,
} from "shared/utils/result";
import { EmailSender } from "server/features/notifications/email";
import { emailNotificationWorker } from "server/features/notifications/emailNotificationWorker";
export enum NotificationTaskType {
SEND_EMAIL_ACCEPTED,
SEND_EMAIL_REJECTED,
SEND_EMAIL_PROGRAM_ITEM_CANCELLED,
SEND_EMAIL_PROGRAM_ITEM_DELETED,
SEND_EMAIL_PROGRAM_ITEM_NO_KONSTI_SIGNUP_ANYMORE,
SEND_EMAIL_PROGRAM_ITEM_NO_LOTTERY_ANYMORE,
SEND_EMAIL_PROGRAM_ITEM_TIME_CHANGED,
}
export interface NotificationTask {
type: NotificationTaskType;
username: string;
programItemId: string;
programItemStartTime: string;
programItemTitle?: string;
}
export interface NotificationQueueService {
addNotificationsBulk(
notifications: NotificationTask[],
): Result<boolean, QueueError>;
drain(): Promise<void>;
kill(): Promise<void>;
getItems(): NotificationTask[];
getQueue(): queueAsPromised<NotificationTask>;
getSender(): EmailSender;
}
export function createNotificationQueueService(
sender: EmailSender,
workerCount = 1,
stopOnStart = false,
): NotificationQueueService {
const queue: queueAsPromised<NotificationTask> = queuePromise(
(notification: NotificationTask) =>
emailNotificationWorker(sender, notification),
workerCount,
);
Eif (stopOnStart) {
queue.pause();
}
function addNotificationsBulk(
notifications: NotificationTask[],
): Result<boolean, QueueError> {
if (notifications.length === 0) {
return makeSuccessResult(true);
}
try {
for (const notification of notifications) {
addNotification(notification);
}
return makeSuccessResult(true);
} catch {
return makeErrorResult(QueueError.FAILED_TO_PUSH);
}
}
function addNotification(
notification: NotificationTask,
): Result<boolean, QueueError> {
try {
// Promise returned by push is fullfilled after task is completed.
void queue.push(notification);
return makeSuccessResult(true);
} catch {
return makeErrorResult(QueueError.FAILED_TO_PUSH);
}
}
return {
addNotificationsBulk,
drain: async () => {
await queue.drain();
},
kill: async () => {
await queue.kill();
},
getItems(): NotificationTask[] {
return queue.getQueue();
},
getQueue(): queueAsPromised<NotificationTask> {
return queue;
},
getSender(): EmailSender {
return sender;
},
};
}
let globalNotificationQueueService: NotificationQueueService | null = null;
export function setGlobalNotificationQueueService(
service: NotificationQueueService | null,
): void {
globalNotificationQueueService = service;
}
export function getGlobalNotificationQueueService(): NotificationQueueService | null {
return globalNotificationQueueService;
}
|