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 | 11x 11x 11x 11x 11x 11x | import { Processor, WorkerHost } from '@nestjs/bullmq';
import { Job } from 'bullmq';
import { Injectable, Logger } from '@nestjs/common';
import { PrismaService } from '@app/modules/prisma/prisma.service';
import { InjectQueue } from '@nestjs/bullmq';
import { Queue } from 'bullmq';
interface DunningPayload { agencyId?: number }
@Processor('billing:dunning')
@Injectable()
export class DunningProcessor extends WorkerHost {
private readonly logger = new Logger(DunningProcessor.name);
constructor(
private readonly prisma: PrismaService,
@InjectQueue('notifications') private readonly notificationsQueue: Queue
) { super(); }
async process(job: Job<DunningPayload>): Promise<void> {
const { agencyId } = job.data;
// Find overdue subscriptions
const overdueSubscriptions = await this.prisma.client.subscription.findMany({
where: {
status: 'PAST_DUE',
gracePeriodEndsAt: {
lt: new Date()
},
...(agencyId ? { agencyId } : {})
},
include: {
agency: {
include: {
owner: true
}
}
}
});
for (const subscription of overdueSubscriptions) {
try {
// Archive all listings for this agency
await this.prisma.client.listing.updateMany({
where: {
user: {
agencyId: subscription.agencyId
}
},
data: {
status: 'ARCHIVED'
}
});
// Notify agency owner
await this.notificationsQueue.add('send', {
type: 'SUBSCRIPTION_SUSPENDED',
payload: {
agencyId: subscription.agencyId,
subscriptionId: subscription.id,
reason: 'Payment overdue - grace period expired'
}
});
this.logger.log(`Suspended agency ${subscription.agencyId} due to overdue subscription ${subscription.id}`);
} catch (error) {
this.logger.error(`Failed to process dunning for agency ${subscription.agencyId}:`, error);
}
}
}
}
|