fix(server): delay send mail if retry many times (#13225)
fix AF-2748 <!-- This is an auto-generated comment: release notes by coderabbit.ai --> ## Summary by CodeRabbit * **New Features** * Improved mail sending job with adaptive retry delays based on elapsed time, enhancing reliability of email delivery. * **Chores** * Updated job payload to include a start time for better retry management. * Added an internal delay utility to support asynchronous pause in processes. <!-- end of auto-generated comment: release notes by coderabbit.ai -->
This commit is contained in:
@@ -1,3 +1,5 @@
|
|||||||
|
import { setTimeout } from 'node:timers/promises';
|
||||||
|
|
||||||
import { defer as rxjsDefer, retry } from 'rxjs';
|
import { defer as rxjsDefer, retry } from 'rxjs';
|
||||||
|
|
||||||
export class RetryablePromise<T> extends Promise<T> {
|
export class RetryablePromise<T> extends Promise<T> {
|
||||||
@@ -48,3 +50,7 @@ export function defer(dispose: () => Promise<void>) {
|
|||||||
[Symbol.asyncDispose]: dispose,
|
[Symbol.asyncDispose]: dispose,
|
||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
|
export function sleep(ms: number): Promise<void> {
|
||||||
|
return setTimeout(ms);
|
||||||
|
}
|
||||||
|
|||||||
@@ -1,7 +1,7 @@
|
|||||||
import { Injectable } from '@nestjs/common';
|
import { Injectable } from '@nestjs/common';
|
||||||
import { getStreamAsBuffer } from 'get-stream';
|
import { getStreamAsBuffer } from 'get-stream';
|
||||||
|
|
||||||
import { JOB_SIGNAL, OnJob } from '../../base';
|
import { JOB_SIGNAL, OnJob, sleep } from '../../base';
|
||||||
import { type MailName, MailProps, Renderers } from '../../mails';
|
import { type MailName, MailProps, Renderers } from '../../mails';
|
||||||
import { UserProps, WorkspaceProps } from '../../mails/components';
|
import { UserProps, WorkspaceProps } from '../../mails/components';
|
||||||
import { Models } from '../../models';
|
import { Models } from '../../models';
|
||||||
@@ -34,7 +34,7 @@ type SendMailJob<Mail extends MailName = MailName, Props = MailProps<Mail>> = {
|
|||||||
|
|
||||||
declare global {
|
declare global {
|
||||||
interface Jobs {
|
interface Jobs {
|
||||||
'notification.sendMail': {
|
'notification.sendMail': { startTime: number } & {
|
||||||
[K in MailName]: SendMailJob<K>;
|
[K in MailName]: SendMailJob<K>;
|
||||||
}[MailName];
|
}[MailName];
|
||||||
}
|
}
|
||||||
@@ -50,7 +50,12 @@ export class MailJob {
|
|||||||
) {}
|
) {}
|
||||||
|
|
||||||
@OnJob('notification.sendMail')
|
@OnJob('notification.sendMail')
|
||||||
async sendMail({ name, to, props }: Jobs['notification.sendMail']) {
|
async sendMail({
|
||||||
|
startTime,
|
||||||
|
name,
|
||||||
|
to,
|
||||||
|
props,
|
||||||
|
}: Jobs['notification.sendMail']) {
|
||||||
let options: Partial<SendOptions> = {};
|
let options: Partial<SendOptions> = {};
|
||||||
|
|
||||||
for (const key in props) {
|
for (const key in props) {
|
||||||
@@ -100,8 +105,15 @@ export class MailJob {
|
|||||||
)),
|
)),
|
||||||
...options,
|
...options,
|
||||||
});
|
});
|
||||||
|
if (result === false) {
|
||||||
|
// wait for a while before retrying
|
||||||
|
const elapsed = Date.now() - startTime;
|
||||||
|
const retryDelay = Math.min(30 * 1000, Math.round(elapsed / 2000) * 1000);
|
||||||
|
await sleep(retryDelay);
|
||||||
|
return JOB_SIGNAL.Retry;
|
||||||
|
}
|
||||||
|
|
||||||
return result === false ? JOB_SIGNAL.Retry : undefined;
|
return undefined;
|
||||||
}
|
}
|
||||||
|
|
||||||
private async fetchWorkspaceProps(workspaceId: string) {
|
private async fetchWorkspaceProps(workspaceId: string) {
|
||||||
|
|||||||
@@ -15,11 +15,14 @@ export class Mailer {
|
|||||||
*
|
*
|
||||||
* @note never throw
|
* @note never throw
|
||||||
*/
|
*/
|
||||||
async trySend(command: Jobs['notification.sendMail']) {
|
async trySend(command: Omit<Jobs['notification.sendMail'], 'startTime'>) {
|
||||||
return this.send(command, true);
|
return this.send(command, true);
|
||||||
}
|
}
|
||||||
|
|
||||||
async send(command: Jobs['notification.sendMail'], suppressError = false) {
|
async send(
|
||||||
|
command: Omit<Jobs['notification.sendMail'], 'startTime'>,
|
||||||
|
suppressError = false
|
||||||
|
) {
|
||||||
if (!this.sender.configured) {
|
if (!this.sender.configured) {
|
||||||
if (suppressError) {
|
if (suppressError) {
|
||||||
return false;
|
return false;
|
||||||
@@ -28,7 +31,12 @@ export class Mailer {
|
|||||||
}
|
}
|
||||||
|
|
||||||
try {
|
try {
|
||||||
await this.queue.add('notification.sendMail', command);
|
await this.queue.add(
|
||||||
|
'notification.sendMail',
|
||||||
|
Object.assign({}, command, {
|
||||||
|
startTime: Date.now(),
|
||||||
|
}) as Jobs['notification.sendMail']
|
||||||
|
);
|
||||||
return true;
|
return true;
|
||||||
} catch {
|
} catch {
|
||||||
return false;
|
return false;
|
||||||
|
|||||||
Reference in New Issue
Block a user