danak/dsc-api
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore: add notif service for new criticisms ticket and subscribtion

Browse Source
This commit is contained in:
mahyargdz
2025-05-04 14:48:39 +03:30
parent 9d50578cc2
commit 170bb867d0
14 changed files with 236 additions and 93 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/modules/invoices/constants/index.ts
+21 -12
View File
@@ -1,20 +1,29 @@
export const INVOICE = Object.freeze({
INVOICE_QUEUE_NAME: "invoice",
INVOICE_REMINDER_JOB_NAME: "reminderInvoice",
INVOICE_RECURRING_JOB_NAME: "recurringInvoice",
QUEUE_NAME: "invoice",
REMINDER_JOB_NAME: "reminderInvoice",
RECURRING_JOB_NAME: "recurringInvoice",
SUBSCRIPTION_ADMIN_NOTIFICATION_JOB_NAME: "subscriptionAdminNotification",
//
INVOICE_REMINDER_JOB_PRIORITY: 1, // high priority
INVOICE_REMINDER_JOB_ATTEMPTS: 3, // 3 times
INVOICE_REMINDER_JOB_BACKOFF: 5 * 1000, // ms
INVOICE_REMINDER_JOB_TIMEOUT: 10000, // timeout after 10 seconds
INVOICE_REMINDER_REPEAT_DELAY: 1 * 24 * 60 * 60 * 1000, // 1 day delay in milliseconds
REMINDER_JOB_PRIORITY: 1, // high priority
REMINDER_JOB_ATTEMPTS: 3, // 3 times
REMINDER_JOB_BACKOFF: 5 * 1000, // ms
REMINDER_JOB_TIMEOUT: 10000, // timeout after 10 seconds
REMINDER_REPEAT_DELAY: 1 * 24 * 60 * 60 * 1000, // 1 day delay in milliseconds
//
INVOICE_RECURRING_JOB_PRIORITY: 1, // high priority
INVOICE_RECURRING_JOB_ATTEMPTS: 3, // retry 3 times
INVOICE_RECURRING_JOB_BACKOFF: 5 * 1000, // retry after 1 second
INVOICE_RECURRING_JOB_TIMEOUT: 10000, // timeout after 10 seconds
RECURRING_JOB_PRIORITY: 1, // high priority
RECURRING_JOB_ATTEMPTS: 3, // retry 3 times
RECURRING_JOB_BACKOFF: 5 * 1000, // retry after 5 second
RECURRING_JOB_TIMEOUT: 10000, // timeout after 10 seconds
//
SUBSCRIPTION_ADMIN_NOTIFICATION_JOB_DELAY: 1000, // delay after 1 second
SUBSCRIPTION_ADMIN_NOTIFICATION_JOB_PRIORITY: 1, // high priority
SUBSCRIPTION_ADMIN_NOTIFICATION_JOB_ATTEMPTS: 3, // retry 3 times
SUBSCRIPTION_ADMIN_NOTIFICATION_JOB_BACKOFF: 5 * 1000, // retry after 5 second
SUBSCRIPTION_ADMIN_NOTIFICATION_JOB_TIMEOUT: 10000, // timeout after 10 seconds
//
FINE_PERCENTAGE: 0.01, // 1% of the total price
MAX_DAYS_AFTER_OVERDUE: 10,
DUEDATE: 7,
src/modules/invoices/invoices.module.ts
+1 -1
View File
@@ -23,7 +23,7 @@ import { WalletsModule } from "../wallets/wallets.module";
@Module({
imports: [
LoggerModule,
BullModule.registerQueue({ name: INVOICE.INVOICE_QUEUE_NAME }),
BullModule.registerQueue({ name: INVOICE.QUEUE_NAME }),
TypeOrmModule.forFeature([Invoice, InvoiceItem, Discount, UsageDiscount]),
UsersModule,
WalletsModule,
src/modules/invoices/providers/invoices.service.ts
+59 -40
View File
@@ -35,7 +35,7 @@ export class InvoicesService {
private readonly logger = new Logger(InvoicesService.name);
constructor(
@InjectQueue(INVOICE.INVOICE_QUEUE_NAME) private readonly invoiceQueue: Queue,
@InjectQueue(INVOICE.QUEUE_NAME) private readonly invoiceQueue: Queue,
private readonly notificationsService: NotificationsService,
private readonly invoiceRepository: InvoicesRepository,
private readonly invoiceItemsRepository: InvoiceItemsRepository,
@@ -380,12 +380,13 @@ export class InvoicesService {
);
await this.invoiceQueue.add(
INVOICE.INVOICE_REMINDER_JOB_NAME,
INVOICE.REMINDER_JOB_NAME,
{ invoiceId: invoice.id },
{
delay: dayjs(invoice.dueDate).subtract(INVOICE.DAYS_BEFORE_OVERDUE, "day").diff(dayjs()),
attempts: INVOICE.INVOICE_REMINDER_JOB_ATTEMPTS,
backoff: { type: "exponential", delay: INVOICE.INVOICE_REMINDER_JOB_BACKOFF },
attempts: INVOICE.REMINDER_JOB_ATTEMPTS,
backoff: { type: "exponential", delay: INVOICE.REMINDER_JOB_BACKOFF },
priority: INVOICE.REMINDER_JOB_PRIORITY,
},
);
@@ -470,9 +471,11 @@ export class InvoicesService {
async payInvoice(invoiceId: string, userId: string) {
const queryRunner = this.dataSource.createQueryRunner();
await queryRunner.connect();
await queryRunner.startTransaction();
try {
await queryRunner.connect();
await queryRunner.startTransaction();
const user = await this.usersService.findOneByIdWithQueryRunner(userId, queryRunner);
const invoice = await this.getPendingInvoiceByIdWithQueryRunner(invoiceId, user.id, queryRunner);
@@ -494,6 +497,7 @@ export class InvoicesService {
//
await queryRunner.manager.save(UserSubscription, userSubscription);
await this.walletsService.createSubscriptionTransaction(invoice.totalPrice, userWallet.id, queryRunner);
await this.addNotifyAdminForSubscriptionPaymentToQueue(invoice);
//
} else {
await this.walletsService.createInvoiceTransaction(invoice.totalPrice, userWallet.id, queryRunner);
@@ -603,40 +607,6 @@ export class InvoicesService {
return delayMs;
}
//*********************************** */
private async scheduleInvoiceJobs(invoice: Invoice, createDto?: CreateInvoiceDto) {
// Add to queue for billing reminder
await this.invoiceQueue.add(
INVOICE.INVOICE_REMINDER_JOB_NAME,
{ invoiceId: invoice.id },
{
delay: dayjs(invoice.dueDate).subtract(INVOICE.DAYS_BEFORE_OVERDUE, "day").diff(dayjs()),
attempts: INVOICE.INVOICE_REMINDER_JOB_ATTEMPTS,
backoff: { type: "exponential", delay: INVOICE.INVOICE_REMINDER_JOB_BACKOFF },
},
);
if (createDto?.isRecurring && !createDto?.recurringPeriod) throw new BadRequestException(InvoiceMessage.RECURRING_PERIOD_REQUIRED);
// Add to queue for recurring invoice if applicable
if (createDto?.isRecurring && createDto?.recurringPeriod) {
await this.invoiceQueue.add(
INVOICE.INVOICE_RECURRING_JOB_NAME,
{
invoiceId: invoice.id,
adminCreated: true,
},
{
delay: await this.calculateRecurringDelay(createDto.recurringPeriod),
attempts: INVOICE.INVOICE_RECURRING_JOB_ATTEMPTS,
backoff: { type: "exponential", delay: INVOICE.INVOICE_REMINDER_JOB_BACKOFF },
},
);
this.logger.log(`Scheduled recurring invoice for user ${createDto.userId} with interval ${createDto.recurringPeriod}`);
}
}
//*********************************** */
async applyDiscount(invoiceId: string, discountCode: string, userId: string) {
const queryRunner = this.dataSource.createQueryRunner();
@@ -721,4 +691,53 @@ export class InvoicesService {
if (!invoice) throw new BadRequestException(InvoiceMessage.NOT_FOUND_BY_ID_OR_NOT_BELONG_TO_USER);
return invoice;
}
//*********************************** */
private async scheduleInvoiceJobs(invoice: Invoice, createDto?: CreateInvoiceDto) {
// Add to queue for billing reminder
await this.invoiceQueue.add(
INVOICE.REMINDER_JOB_NAME,
{ invoiceId: invoice.id },
{
delay: dayjs(invoice.dueDate).subtract(INVOICE.DAYS_BEFORE_OVERDUE, "day").diff(dayjs()),
attempts: INVOICE.REMINDER_JOB_ATTEMPTS,
backoff: { type: "exponential", delay: INVOICE.REMINDER_JOB_BACKOFF },
priority: INVOICE.REMINDER_JOB_PRIORITY,
},
);
if (createDto?.isRecurring && !createDto?.recurringPeriod) throw new BadRequestException(InvoiceMessage.RECURRING_PERIOD_REQUIRED);
// Add to queue for recurring invoice if applicable
if (createDto?.isRecurring && createDto?.recurringPeriod) {
await this.invoiceQueue.add(
INVOICE.RECURRING_JOB_NAME,
{
invoiceId: invoice.id,
adminCreated: true,
},
{
delay: await this.calculateRecurringDelay(createDto.recurringPeriod),
attempts: INVOICE.RECURRING_JOB_ATTEMPTS,
backoff: { type: "exponential", delay: INVOICE.RECURRING_JOB_BACKOFF },
priority: INVOICE.RECURRING_JOB_PRIORITY,
},
);
this.logger.log(`Scheduled recurring invoice for user ${createDto.userId} with interval ${createDto.recurringPeriod}`);
}
}
//*********************************** */
private async addNotifyAdminForSubscriptionPaymentToQueue(invoice: Invoice) {
await this.invoiceQueue.add(
INVOICE.SUBSCRIPTION_ADMIN_NOTIFICATION_JOB_NAME,
{ invoiceId: invoice.id },
{
delay: INVOICE.SUBSCRIPTION_ADMIN_NOTIFICATION_JOB_DELAY,
attempts: INVOICE.SUBSCRIPTION_ADMIN_NOTIFICATION_JOB_ATTEMPTS,
backoff: { type: "exponential", delay: INVOICE.SUBSCRIPTION_ADMIN_NOTIFICATION_JOB_BACKOFF },
priority: INVOICE.SUBSCRIPTION_ADMIN_NOTIFICATION_JOB_PRIORITY,
},
);
}
}
src/modules/invoices/queue/invoice.processor.ts
+50 -9
View File
@@ -15,10 +15,10 @@ import { Invoice } from "../entities/invoice.entity";
import { InvoiceStatus } from "../enums/invoice-status.enum";
import { InvoicesService } from "../providers/invoices.service";
@Processor(INVOICE.INVOICE_QUEUE_NAME, { concurrency: 5 })
@Processor(INVOICE.QUEUE_NAME, { concurrency: 5 })
export class InvoiceProcessor extends WorkerProcessor {
constructor(
@InjectQueue(INVOICE.INVOICE_QUEUE_NAME) private readonly invoiceQueue: Queue,
@InjectQueue(INVOICE.QUEUE_NAME) private readonly invoiceQueue: Queue,
private readonly invoicesService: InvoicesService,
private readonly dataSource: DataSource,
private readonly notificationService: NotificationsService,
@@ -30,10 +30,12 @@ export class InvoiceProcessor extends WorkerProcessor {
async process(job: Job, token?: string) {
this.logger.log(job);
switch (job.name) {
case INVOICE.INVOICE_REMINDER_JOB_NAME:
case INVOICE.REMINDER_JOB_NAME:
return this.sendBillInvoiceReminder(job, token);
case INVOICE.INVOICE_RECURRING_JOB_NAME:
case INVOICE.RECURRING_JOB_NAME:
return this.createRecurringInvoice(job, token);
case INVOICE.SUBSCRIPTION_ADMIN_NOTIFICATION_JOB_NAME:
return this.notifyAdminForSubscriptionInvoice(job);
default:
this.logger.error(`Unknown job name: ${job.name}`);
return;
@@ -176,7 +178,7 @@ export class InvoiceProcessor extends WorkerProcessor {
private async fetchInvoice(invoiceId: string, queryRunner: QueryRunner) {
const invoice = await queryRunner.manager.findOne(Invoice, {
where: { id: invoiceId },
relations: { user: true, items: { subscriptionPlan: { plan: true } } },
relations: { user: true, items: { subscriptionPlan: { plan: { service: true } } } },
});
if (!invoice) throw new Error(`Invoice not found: ${invoiceId}`);
@@ -291,12 +293,12 @@ export class InvoiceProcessor extends WorkerProcessor {
private async scheduleNextReminder(invoiceId: string) {
this.logger.log(`Invoice ${invoiceId} is still unpaid. Scheduling a new reminder.`);
await this.invoiceQueue.add(
INVOICE.INVOICE_REMINDER_JOB_NAME,
INVOICE.REMINDER_JOB_NAME,
{ invoiceId },
{
delay: INVOICE.INVOICE_REMINDER_REPEAT_DELAY, // 1 day delay in milliseconds
attempts: INVOICE.INVOICE_REMINDER_JOB_ATTEMPTS,
backoff: { type: "exponential", delay: INVOICE.INVOICE_REMINDER_JOB_BACKOFF },
delay: INVOICE.REMINDER_REPEAT_DELAY, // 1 day delay in milliseconds
attempts: INVOICE.REMINDER_JOB_ATTEMPTS,
backoff: { type: "exponential", delay: INVOICE.REMINDER_JOB_BACKOFF },
},
);
}
@@ -326,4 +328,43 @@ export class InvoiceProcessor extends WorkerProcessor {
admins.map((admin) => this.notificationService.createRecurringInvoiceNotification(admin.id, notificationPayload, queryRunner)),
);
}
//********************************** */
private async notifyAdminForSubscriptionInvoice(job: Job<{ invoiceId: string }>) {
const queryRunner = this.dataSource.createQueryRunner();
try {
await queryRunner.connect();
await queryRunner.startTransaction();
const { invoiceId } = job.data;
const invoice = await this.fetchInvoice(invoiceId, queryRunner);
const superAdmins = await this.fetchAdmin(queryRunner);
for (const admin of superAdmins) {
await this.notificationService.createNewSubscriptionNotification(
admin.id,
{
date: invoice.dueDate,
userPhone: invoice.user.phone,
userEmail: invoice.user.email,
serviceName: invoice.items[0]?.subscriptionPlan?.plan.service.name ?? "",
fullName: invoice.user.firstName + " " + invoice.user.lastName,
},
queryRunner,
);
}
await queryRunner.commitTransaction();
return true;
} catch (error) {
this.logger.error(`Failed to notify admins for subscription invoice:`, error);
await queryRunner.rollbackTransaction();
throw error;
} finally {
await queryRunner.release();
}
}
}