fix bug
This commit is contained in:
@@ -74,18 +74,29 @@ export class NotificationQueueService {
|
|||||||
|
|
||||||
async addBulkSmsNotifications(jobs: SmsNotificationQueueJob[]): Promise<void> {
|
async addBulkSmsNotifications(jobs: SmsNotificationQueueJob[]): Promise<void> {
|
||||||
try {
|
try {
|
||||||
const queueJobs = jobs.map(job => ({
|
const queueJobs = jobs.map((job, index) => {
|
||||||
|
const recipientId =
|
||||||
|
'adminId' in job.recipient ? job.recipient.adminId : job.recipient.userId;
|
||||||
|
return {
|
||||||
name: 'sms-notification',
|
name: 'sms-notification',
|
||||||
data: job,
|
data: job,
|
||||||
opts: {
|
opts: {
|
||||||
jobId: `${job.templateId}-${this.generateRandomInt()}`,
|
jobId: `sms-${job.templateId}-${recipientId}-${index}-${Date.now()}`,
|
||||||
attempts: 3,
|
attempts: 3,
|
||||||
backoff: {
|
backoff: {
|
||||||
type: 'exponential',
|
type: 'exponential',
|
||||||
delay: 2000,
|
delay: 2000,
|
||||||
},
|
},
|
||||||
|
removeOnComplete: {
|
||||||
|
age: 24 * 3600,
|
||||||
|
count: 1000,
|
||||||
},
|
},
|
||||||
}));
|
removeOnFail: {
|
||||||
|
age: 7 * 24 * 3600,
|
||||||
|
},
|
||||||
|
},
|
||||||
|
};
|
||||||
|
});
|
||||||
|
|
||||||
await this.smsQueue.addBulk(queueJobs);
|
await this.smsQueue.addBulk(queueJobs);
|
||||||
this.logger.log(`Added ${jobs.length} SMS notification jobs to queue`);
|
this.logger.log(`Added ${jobs.length} SMS notification jobs to queue`);
|
||||||
|
|||||||
@@ -1,6 +1,7 @@
|
|||||||
import { Injectable, Logger } from '@nestjs/common';
|
import { Injectable, Logger } from '@nestjs/common';
|
||||||
import { Cron } from '@nestjs/schedule';
|
import { Cron } from '@nestjs/schedule';
|
||||||
import { EntityManager } from '@mikro-orm/postgresql';
|
import { EntityManager } from '@mikro-orm/postgresql';
|
||||||
|
import { DeadlockException } from '@mikro-orm/core';
|
||||||
import { EventEmitter2 } from '@nestjs/event-emitter';
|
import { EventEmitter2 } from '@nestjs/event-emitter';
|
||||||
import { Payment } from '../../payments/entities/payment.entity';
|
import { Payment } from '../../payments/entities/payment.entity';
|
||||||
import { PaymentMethodEnum, PaymentStatusEnum } from '../../payments/interface/payment';
|
import { PaymentMethodEnum, PaymentStatusEnum } from '../../payments/interface/payment';
|
||||||
@@ -8,6 +9,9 @@ import { OrderStatus } from '../interface/order.interface';
|
|||||||
import { Order } from '../entities/order.entity';
|
import { Order } from '../entities/order.entity';
|
||||||
import { OrderStatusChangedEvent } from '../events/order.events';
|
import { OrderStatusChangedEvent } from '../events/order.events';
|
||||||
|
|
||||||
|
const MAX_DEADLOCK_RETRIES = 3;
|
||||||
|
const DEADLOCK_RETRY_DELAY_MS = 100;
|
||||||
|
|
||||||
@Injectable()
|
@Injectable()
|
||||||
export class OrdersCrone {
|
export class OrdersCrone {
|
||||||
private readonly logger = new Logger(OrdersCrone.name);
|
private readonly logger = new Logger(OrdersCrone.name);
|
||||||
@@ -23,11 +27,43 @@ export class OrdersCrone {
|
|||||||
timeZone: 'UTC',
|
timeZone: 'UTC',
|
||||||
})
|
})
|
||||||
async cancelOldOnlinePendingOrders() {
|
async cancelOldOnlinePendingOrders() {
|
||||||
|
let lastError: Error | null = null;
|
||||||
|
for (let attempt = 1; attempt <= MAX_DEADLOCK_RETRIES; attempt++) {
|
||||||
try {
|
try {
|
||||||
|
await this.runCancelOldOnlinePendingOrders();
|
||||||
|
return;
|
||||||
|
} catch (err) {
|
||||||
|
lastError = err;
|
||||||
|
const isDeadlock =
|
||||||
|
err instanceof DeadlockException ||
|
||||||
|
(err as Error).message?.includes('deadlock detected');
|
||||||
|
if (!isDeadlock || attempt === MAX_DEADLOCK_RETRIES) {
|
||||||
|
this.logger.error(`OrdersCrone failed: ${(err as Error).message}`, (err as Error).stack);
|
||||||
|
throw err;
|
||||||
|
}
|
||||||
|
const delay = DEADLOCK_RETRY_DELAY_MS * Math.pow(2, attempt - 1);
|
||||||
|
this.logger.warn(
|
||||||
|
`OrdersCrone deadlock (attempt ${attempt}/${MAX_DEADLOCK_RETRIES}), retrying in ${delay}ms`,
|
||||||
|
);
|
||||||
|
await this.sleep(delay);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if (lastError) {
|
||||||
|
this.logger.error(`OrdersCrone failed after ${MAX_DEADLOCK_RETRIES} retries`, lastError.stack);
|
||||||
|
throw lastError;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private sleep(ms: number): Promise<void> {
|
||||||
|
return new Promise(resolve => setTimeout(resolve, ms));
|
||||||
|
}
|
||||||
|
|
||||||
|
private async runCancelOldOnlinePendingOrders(): Promise<void> {
|
||||||
const cutoff = new Date(Date.now() - 15 * 60 * 1000);
|
const cutoff = new Date(Date.now() - 15 * 60 * 1000);
|
||||||
|
|
||||||
this.logger.debug('Searching for pending online payments older than 15 minutes');
|
this.logger.debug('Searching for pending online payments older than 15 minutes');
|
||||||
|
|
||||||
|
// Minimal query: only payment ids, no joins – reduces lock contention and deadlock risk
|
||||||
const payments = await this.em.find(
|
const payments = await this.em.find(
|
||||||
Payment,
|
Payment,
|
||||||
{
|
{
|
||||||
@@ -35,7 +71,7 @@ export class OrdersCrone {
|
|||||||
status: PaymentStatusEnum.Pending,
|
status: PaymentStatusEnum.Pending,
|
||||||
createdAt: { $lte: cutoff },
|
createdAt: { $lte: cutoff },
|
||||||
},
|
},
|
||||||
{ populate: ['order', 'order.items', 'order.items.variant'] },
|
{ fields: ['id'] },
|
||||||
);
|
);
|
||||||
|
|
||||||
if (!payments || payments.length === 0) {
|
if (!payments || payments.length === 0) {
|
||||||
@@ -71,12 +107,9 @@ export class OrdersCrone {
|
|||||||
);
|
);
|
||||||
});
|
});
|
||||||
} catch (err) {
|
} catch (err) {
|
||||||
this.logger.error(`Error processing payment ${p.id}: ${err.message}`, err.stack);
|
this.logger.error(`Error processing payment ${p.id}: ${(err as Error).message}`, (err as Error).stack);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
} catch (err) {
|
|
||||||
this.logger.error(`OrdersCrone failed: ${err.message}`, err.stack);
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// run every 15 minutes to complete orders that have been in shipped/delivered statuses for more than 3 hours
|
// run every 15 minutes to complete orders that have been in shipped/delivered statuses for more than 3 hours
|
||||||
|
|||||||
Reference in New Issue
Block a user