Tria/edr-platform
1
0
Fork 0
mirror of https://github.com/Tria-plc/edr-platform.git synced 2026-08-26 18:42:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

Added cron job that check duplicate seat and assign if free available

Browse Source
This commit is contained in:
Roba Boru
2026-07-16 23:22:20 +03:00
parent bf5f62a9ca
commit 969c524afe
4 changed files with 523 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files

109
apps/edr-passenger-api/src/modules/payments/payment-sync.service.ts Normal file
View File

@@ -0,0 +1,109 @@
import { Injectable, Logger } from '@nestjs/common';
import { Cron } from '@nestjs/schedule';
import { ModuleRef } from '@nestjs/core';
import { PrismaService } from '../../common/prisma.service';
import { PaymentClientService } from './payment-client.service';
import { PaymentsService } from './payments.service';
import { PaymentReferenceType, ProviderPaymentStatus } from '@edr/types';
// PaymentsService is request-scoped (AuditService.@Inject(REQUEST) bubbles up).
// This service keeps only singleton deps so its @Cron method registers correctly,
// then resolves PaymentsService per-tick via ModuleRef (same pattern as
// PaymentEventsConsumer).
@Injectable()
export class PaymentSyncService {
private readonly logger = new Logger(PaymentSyncService.name);
constructor(
private readonly prisma: PrismaService,
private readonly paymentClient: PaymentClientService,
private readonly moduleRef: ModuleRef,
) {}
// ─────────────────────────────────────────────────────────────────────────
// Every 1 min: poll the payment service for any PENDING_PAYMENT bookings
// whose payment intent has moved to SUCCEEDED on the gateway but whose
// confirmation event was never delivered (missed RabbitMQ message, network
// blip, etc.). finalizePaymentSuccess() is fully idempotent so re-running
// it for an already-confirmed booking is safe.
//
// Processes at most 50 bookings per cycle to avoid hammering the payment
// service; the next tick picks up the remainder.
// ─────────────────────────────────────────────────────────────────────────
@Cron('*/1 * * * *')
async syncPaymentStatuses() {
const BATCH_SIZE = 50;
const bookings = await this.prisma.booking.findMany({
where: {
status: 'PENDING_PAYMENT',
paymentIntent: { status: { in: ['REQUIRES_ACTION', 'PROCESSING'] } },
},
include: { paymentIntent: true },
take: BATCH_SIZE,
orderBy: { createdAt: 'asc' },
});
if (bookings.length === 0) return;
let confirmed = 0;
let failed = 0;
let errored = 0;
// resolve() (not get()) because PaymentsService is scoped — same pattern
// as PaymentEventsConsumer.
const paymentsService = await this.moduleRef.resolve(
PaymentsService,
undefined,
{ strict: false },
);
for (const booking of bookings) {
if (!booking.paymentIntent) continue;
try {
const snapshot = await this.paymentClient.getIntentByReference(
PaymentReferenceType.BOOKING,
booking.id,
);
if (!snapshot) continue;
if (snapshot.status === ProviderPaymentStatus.SUCCEEDED) {
const result = await paymentsService.finalizePaymentSuccess({
intentId: booking.paymentIntent.id,
providerTxnId: snapshot.providerTxnId,
paidAt: snapshot.paidAt ? new Date(snapshot.paidAt) : undefined,
});
if (!result.alreadyFinalized) {
this.logger.log(`Payment sync confirmed: ${booking.bookingRef}`);
confirmed++;
}
} else if (
snapshot.status === ProviderPaymentStatus.FAILED ||
snapshot.status === ProviderPaymentStatus.CANCELLED
) {
this.logger.warn(
`Payment sync: ${booking.bookingRef} intent is ${snapshot.status} — ` +
`booking will be auto-cancelled at payment deadline`,
);
failed++;
}
// REQUIRES_ACTION / PROCESSING → still pending, retry next cycle
} catch (err) {
this.logger.error(
`Payment sync error for ${booking.bookingRef}: ` +
`${err instanceof Error ? err.message : String(err)}`,
);
errored++;
}
}
if (confirmed > 0 || failed > 0 || errored > 0) {
this.logger.log(
`Payment sync run: ${bookings.length} checked, ` +
`${confirmed} confirmed, ${failed} failed/cancelled, ${errored} errors`,
);
}
}
}

2
apps/edr-passenger-api/src/modules/payments/payments.module.ts
View File

@@ -16,6 +16,7 @@ import { SupplementaryChargesService } from "./supplementary-charges.service";
import { InternalPaymentsController } from "./internal-payments.controller";
import { PaymentClientService } from "./payment-client.service";
import { PaymentEventsConsumer } from "./payment-events.consumer";
import { PaymentSyncService } from "./payment-sync.service";
import { ServiceAuthGuard } from "../../common/guards/service-auth.guard";
import { SeatsModule } from "../seats/seats.module";
import { TicketsModule } from "../tickets/tickets.module";
@@ -70,6 +71,7 @@ function rabbitMQImport(): DynamicModule[] {
SupplementaryChargesService,
PaymentClientService,
PaymentEventsConsumer,
PaymentSyncService,
ServiceAuthGuard,
],
exports: [PaymentClientService, PaymentsService],

3
apps/edr-passenger-api/src/modules/tasks/tasks.module.ts
View File

@@ -2,11 +2,10 @@ import { Module } from '@nestjs/common';
import { PrismaModule } from '../../common/prisma.module';
import { NotificationsModule } from '../notifications/notifications.module';
import { CurrencyModule } from '../currency/currency.module';
import { PaymentsModule } from '../payments/payments.module';
import { TasksService } from './tasks.service';
@Module({
imports: [PrismaModule, NotificationsModule, CurrencyModule, PaymentsModule],
imports: [PrismaModule, NotificationsModule, CurrencyModule],
providers: [TasksService],
})
export class TasksModule {}

475
apps/edr-passenger-api/src/modules/tasks/tasks.service.ts
View File

@@ -1,11 +1,9 @@
import { Injectable, Logger } from '@nestjs/common';
import { Cron } from '@nestjs/schedule';
import { SeatStatus } from '@prisma/client';
import { PrismaService } from '../../common/prisma.service';
import { SmsClientService } from '../notifications/sms-client.service';
import { CurrencyService } from '../currency/currency.service';
import { PaymentsService } from '../payments/payments.service';
import { PaymentClientService } from '../payments/payment-client.service';
import { PaymentReferenceType, ProviderPaymentStatus } from '@edr/types';
import { MAX_PAYMENT_HOURS, CUTOFF_MINUTES, computePaymentDeadline } from '../../common/utils/payment-deadline.utils';
// Retention windows
@@ -31,8 +29,6 @@ export class TasksService {
private readonly prisma: PrismaService,
private readonly sms: SmsClientService,
private readonly currencyService: CurrencyService,
private readonly paymentsService: PaymentsService,
private readonly paymentClient: PaymentClientService,
) {}
// ─────────────────────────────────────────────────────────────────────────
@@ -259,84 +255,435 @@ export class TasksService {
}
// ─────────────────────────────────────────────────────────────────────────
// Every 1 min: poll the payment service for any PENDING_PAYMENT bookings
// whose payment intent has moved to SUCCEEDED on the gateway but whose
// confirmation event was never delivered (missed RabbitMQ message, network
// blip, etc.). finalizePaymentSuccess() is fully idempotent so re-running
// it for an already-confirmed booking is safe.
// Every 1 min: detect and resolve duplicate seat assignments.
//
// Processes at most 50 bookings per cycle to avoid hammering the payment
// service; the next tick picks up the remainder.
// Root cause: a stale RabbitMQ message, delivered after system recovery,
// re-confirmed a cancelled booking whose seat had already been assigned to
// a new booking — leaving two CONFIRMED bookings holding the same seat on
// the same schedule.
//
// Resolution (FCFS):
// • Earliest confirmed booking keeps the original seat.
// • All later duplicates are reassigned to the next free seat within the
// SAME coach type (same coach preferred; any coach of same type as
// fallback).
// • If no seat is available in that coach type the booking is flagged for
// manual intervention and logged as unresolved.
//
// Idempotent: after reassignment the BookingSeat/JourneySegment rows no
// longer share the same (seatId, scheduleId) key, so the next tick finds
// nothing to do for the same pair.
//
// Scope: only schedules departing in the last 24 h or in the future, to
// keep the per-tick DB scan bounded.
// ─────────────────────────────────────────────────────────────────────────
@Cron('*/1 * * * *')
async syncPaymentStatuses() {
const BATCH_SIZE = 50;
async resolveDuplicateSeatAssignments() {
const BATCH_SIZE = 20;
const since = new Date(Date.now() - 24 * 60 * 60 * 1000);
const bookings = await this.prisma.booking.findMany({
// Fetch all BookingSeat rows for CONFIRMED bookings on upcoming/recent schedules.
const confirmedSeats = await this.prisma.bookingSeat.findMany({
where: {
status: 'PENDING_PAYMENT',
paymentIntent: { status: { in: ['REQUIRES_ACTION', 'PROCESSING'] } },
booking: {
status: 'CONFIRMED',
schedule: { departureAt: { gte: since } },
},
},
include: {
booking: {
select: {
id: true,
bookingRef: true,
scheduleId: true,
createdAt: true,
contactPhone: true,
schedule: {
include: {
originStation: { select: { name: true } },
destinationStation: { select: { name: true } },
},
},
},
},
seat: {
include: {
coach: {
include: { coachType: { select: { id: true, name: true } } },
},
},
},
},
include: { paymentIntent: true },
take: BATCH_SIZE,
orderBy: { createdAt: 'asc' },
});
if (bookings.length === 0) return;
// Group by (seatId, scheduleId). BookingSeat.scheduleId is per-leg for
// round-trips; fall back to Booking.scheduleId for single-leg bookings.
const groups = new Map<string, typeof confirmedSeats>();
for (const bs of confirmedSeats) {
if (!bs.seatId) continue;
const scheduleId = bs.scheduleId ?? bs.booking.scheduleId;
if (!scheduleId) continue;
const key = `${bs.seatId}:${scheduleId}`;
if (!groups.has(key)) groups.set(key, []);
groups.get(key)!.push(bs);
}
let confirmed = 0;
let failed = 0;
let errored = 0;
const duplicateGroups = [...groups.values()]
.filter(g => g.length > 1)
.slice(0, BATCH_SIZE);
for (const booking of bookings) {
if (!booking.paymentIntent) continue;
if (duplicateGroups.length === 0) return;
try {
const snapshot = await this.paymentClient.getIntentByReference(
PaymentReferenceType.BOOKING,
booking.id,
);
this.logger.warn(`Seat dedup: ${duplicateGroups.length} duplicate seat group(s) detected`);
if (!snapshot) continue;
// Track seats newly assigned within this run to prevent double-assignment.
const newlyAssigned = new Map<string, Set<string>>(); // scheduleId → Set<seatId>
let resolved = 0;
let unresolved = 0;
if (snapshot.status === ProviderPaymentStatus.SUCCEEDED) {
const result = await this.paymentsService.finalizePaymentSuccess({
intentId: booking.paymentIntent.id,
providerTxnId: snapshot.providerTxnId,
paidAt: snapshot.paidAt ? new Date(snapshot.paidAt) : undefined,
});
if (!result.alreadyFinalized) {
this.logger.log(`Payment sync confirmed: ${booking.bookingRef}`);
confirmed++;
}
} else if (
snapshot.status === ProviderPaymentStatus.FAILED ||
snapshot.status === ProviderPaymentStatus.CANCELLED
) {
// The payment deadline enforcer will cancel the booking when its
// window expires; log now so operations can see failed intents early.
this.logger.warn(
`Payment sync: ${booking.bookingRef} intent is ${snapshot.status} — ` +
`booking will be auto-cancelled at payment deadline`,
for (const group of duplicateGroups) {
// FCFS: earliest confirmed booking keeps the seat.
const sorted = [...group].sort(
(a, b) =>
new Date(a.booking.createdAt as Date).getTime() -
new Date(b.booking.createdAt as Date).getTime(),
);
const [keeper, ...duplicates] = sorted;
for (const dup of duplicates) {
const scheduleId = (dup.scheduleId ?? dup.booking.scheduleId)!;
const coachTypeId = dup.seat?.coach?.coachTypeId;
const oldCoachId = dup.seat?.coachId;
if (!coachTypeId) {
this.logger.error(
`Seat dedup: missing coachTypeId for BookingSeat ${dup.id}, booking ${dup.booking.bookingRef}`,
);
failed++;
unresolved++;
continue;
}
if (!newlyAssigned.has(scheduleId)) newlyAssigned.set(scheduleId, new Set());
const takenThisRun = newlyAssigned.get(scheduleId)!;
// All seats already taken: confirmed bookings + those assigned this tick.
const occupiedIds = new Set([
...confirmedSeats
.filter(bs => (bs.scheduleId ?? bs.booking.scheduleId) === scheduleId && bs.seatId)
.map(bs => bs.seatId as string),
...takenThisRun,
]);
try {
const newSeat = await this.findReplacementSeat(scheduleId, coachTypeId, oldCoachId, occupiedIds);
if (!newSeat) {
this.logger.warn(
`Seat dedup: no available seat for booking ${dup.booking.bookingRef} ` +
`(schedule ${scheduleId}, coachType ${coachTypeId}) — manual intervention required`,
);
unresolved++;
continue;
}
await this.prisma.$transaction(async (tx) => {
// 1. Update BookingSeat to the new seat.
await tx.bookingSeat.update({
where: { id: dup.id },
data: { seatId: newSeat.id, seatLabelSnapshot: newSeat.seatNumber },
});
// 2. Update JourneySegment — look up journeyId first to avoid a
// nested-relation filter in updateMany (not supported in all Prisma versions).
const journey = await tx.journey.findUnique({
where: { bookingId: dup.booking.id } as any,
select: { id: true },
});
if (journey) {
await tx.journeySegment.updateMany({
where: { journeyId: journey.id, seatId: dup.seatId!, scheduleId },
data: { seatId: newSeat.id, coachId: newSeat.coachId },
});
}
// 3. Update Ticket seat reference (QR payload regeneration is out of scope
// here; the backoffice can trigger that separately if required).
await tx.ticket.updateMany({
where: { bookingId: dup.booking.id, seatId: dup.seatId! },
data: { seatId: newSeat.id },
});
});
takenThisRun.add(newSeat.id);
const oldLabel = dup.seat?.seatNumber ?? dup.seatId ?? '?';
const newCoach = (newSeat as any).coach;
const coachTypeName = newCoach?.coachType?.name ?? '';
const coachNumber = newCoach?.number ?? '';
const origin = dup.booking.schedule?.originStation?.name ?? '';
const dest = dup.booking.schedule?.destinationStation?.name ?? '';
if (dup.booking.contactPhone) {
const message =
`EDR: Your booking ${dup.booking.bookingRef} (${origin} → ${dest}): ` +
`your seat has been changed from ${oldLabel} to seat ${newSeat.seatNumber} ` +
`in coach ${coachNumber} (${coachTypeName}). ` +
`We apologize for the inconvenience.`;
await this.sms.sendSms({ to: dup.booking.contactPhone, message }).catch(() => null);
}
this.logger.log(
`Seat dedup resolved: booking ${dup.booking.bookingRef} ` +
`seat ${oldLabel} → ${newSeat.seatNumber} (coach ${coachNumber}, ${coachTypeName}), ` +
`keeper: ${keeper.booking.bookingRef}`,
);
resolved++;
} catch (err) {
this.logger.error(
`Seat dedup error for booking ${dup.booking.bookingRef}: ` +
`${err instanceof Error ? err.message : String(err)}`,
);
unresolved++;
}
// REQUIRES_ACTION / PROCESSING → still pending, retry next cycle
} catch (err) {
this.logger.error(
`Payment sync error for ${booking.bookingRef}: ` +
`${err instanceof Error ? err.message : String(err)}`,
);
errored++;
}
}
if (confirmed > 0 || failed > 0 || errored > 0) {
this.logger.log(
`Payment sync run: ${bookings.length} checked, ` +
`${confirmed} confirmed, ${failed} failed/cancelled, ${errored} errors`,
);
this.logger.log(`Seat dedup run: ${resolved} resolved, ${unresolved} unresolved`);
}
private async findReplacementSeat(
scheduleId: string,
coachTypeId: string,
preferredCoachId: string | undefined,
occupiedIds: Set<string>,
) {
const includeCoach = {
coach: { include: { coachType: { select: { id: true, name: true } } } },
};
const baseWhere = (coachId?: string) => ({
...(coachId ? { coachId } : {}),
seatNumber: { not: '' },
id: { notIn: [...occupiedIds] },
coach: { coachTypeId, assignments: { some: { scheduleId } } },
NOT: [
{ seatNumber: { startsWith: '-' } },
{ status: SeatStatus.BLOCKED },
],
});
// 1. Prefer the exact same coach.
if (preferredCoachId) {
const seat = await this.prisma.seat.findFirst({
where: baseWhere(preferredCoachId),
include: includeCoach,
orderBy: [{ row: 'asc' }, { col: 'asc' }],
});
if (seat) return seat;
}
// 2. Any coach of the same coach type assigned to this schedule.
return this.prisma.seat.findFirst({
where: baseWhere(),
include: includeCoach,
orderBy: [{ coach: { number: 'asc' } }, { row: 'asc' }, { col: 'asc' }],
});
}
// ─────────────────────────────────────────────────────────────────────────
// Every 1 min: detect and resolve duplicate seat assignments caused by
// RabbitMQ-recovered events re-confirming already-cancelled bookings.
//
// Detection: group confirmed BookingSeat rows by (scheduleId, seatId, leg).
// Any group with >1 row means multiple bookings share the same physical seat.
//
// Resolution (FCFS): the booking created first keeps the seat; all later
// bookings are reassigned to an available seat in:
// 1. Same coach + same coach type (preferred)
// 2. Same coach type, any coach (fallback)
// 3. No seat available → logged, needs manual intervention
//
// Idempotency: once a duplicate's BookingSeat is updated to a new seatId it
// no longer appears in the duplicate group on the next tick — naturally safe
// to re-run without any extra flag.
// ─────────────────────────────────────────────────────────────────────────
@Cron('*/1 * * * *')
async deduplicateSeatAssignments() {
this.logger.log('Seat dedup cron started');
// Scan at most 500 confirmed seat rows per run to stay lightweight.
const confirmedSeats = await this.prisma.bookingSeat.findMany({
where: { booking: { status: { in: ['CONFIRMED', 'BOARDED'] } } },
select: {
id: true,
seatId: true,
scheduleId: true,
leg: true,
passengerName: true,
booking: {
select: {
id: true,
bookingRef: true,
scheduleId: true,
createdAt: true,
contactPhone: true,
},
},
seat: {
select: {
id: true,
seatNumber: true,
coachId: true,
coach: {
select: {
id: true,
number: true,
coachTypeId: true,
coachType: { select: { id: true, name: true } },
},
},
},
},
},
take: 500,
});
// Group by (effectiveScheduleId :: seatId :: leg)
type BsRow = (typeof confirmedSeats)[number];
const groups = new Map<string, BsRow[]>();
for (const bs of confirmedSeats) {
const schedId = bs.scheduleId ?? bs.booking.scheduleId;
if (!schedId) continue;
const key = `${schedId}::${bs.seatId}::${bs.leg}`;
if (!groups.has(key)) groups.set(key, []);
groups.get(key)!.push(bs);
}
const duplicateGroups = [...groups.values()].filter(g => g.length > 1);
if (duplicateGroups.length === 0) return;
this.logger.warn(`Seat dedup: ${duplicateGroups.length} conflict(s) detected`);
// Build taken-seat sets keyed by (scheduleId::leg) — used when finding
// a replacement seat so we don't assign an already-occupied seat.
const takenByScheduleLeg = new Map<string, Set<string>>();
for (const bs of confirmedSeats) {
const schedId = bs.scheduleId ?? bs.booking.scheduleId;
if (!schedId) continue;
const key = `${schedId}::${bs.leg}`;
if (!takenByScheduleLeg.has(key)) takenByScheduleLeg.set(key, new Set());
takenByScheduleLeg.get(key)!.add(bs.seatId);
}
let resolved = 0;
let unresolved = 0;
for (const group of duplicateGroups) {
// FCFS: earliest booking keeps the seat
group.sort((a, b) =>
new Date(a.booking.createdAt).getTime() - new Date(b.booking.createdAt).getTime(),
);
const [winner, ...duplicates] = group;
const schedId = winner.scheduleId ?? winner.booking.scheduleId;
const coachTypeId = winner.seat.coach.coachTypeId;
const origCoachId = winner.seat.coachId;
const taken = takenByScheduleLeg.get(`${schedId}::${winner.leg}`) ?? new Set<string>();
for (const dup of duplicates) {
try {
// 1st choice: same coach + same coach type
const newSeat =
(await this.prisma.seat.findFirst({
where: {
id: { notIn: [...taken] },
status: { not: SeatStatus.BLOCKED },
coachId: origCoachId,
coach: {
coachTypeId,
assignments: { some: { scheduleId: schedId } },
},
},
select: {
id: true, seatNumber: true, coachId: true,
coach: { select: { number: true, coachType: { select: { name: true } } } },
},
})) ??
// 2nd choice: any coach within same coach type
(await this.prisma.seat.findFirst({
where: {
id: { notIn: [...taken] },
status: { not: SeatStatus.BLOCKED },
coach: {
coachTypeId,
assignments: { some: { scheduleId: schedId } },
},
},
select: {
id: true, seatNumber: true, coachId: true,
coach: { select: { number: true, coachType: { select: { name: true } } } },
},
}));
if (!newSeat) {
this.logger.warn(
`Seat dedup: no available seat in coach type for ` +
`booking ${dup.booking.bookingRef} (${dup.passengerName}) — manual intervention required`,
);
unresolved++;
continue;
}
// Atomically update BookingSeat + Ticket + JourneySegment
await this.prisma.$transaction(async (tx) => {
await tx.bookingSeat.update({
where: { id: dup.id },
data: { seatId: newSeat!.id, seatLabelSnapshot: newSeat!.seatNumber },
});
await tx.ticket.updateMany({
where: { bookingId: dup.booking.id, seatId: dup.seatId, leg: dup.leg },
data: { seatId: newSeat!.id },
});
await tx.journeySegment.updateMany({
where: {
journey: { bookingId: dup.booking.id },
seatId: dup.seatId,
scheduleId: schedId,
},
data: { seatId: newSeat!.id, coachId: newSeat!.coachId },
});
});
// Claim the new seat so subsequent duplicates in this run don't use it
taken.add(newSeat.id);
const message =
`EDR: Your seat for booking ${dup.booking.bookingRef} has been updated ` +
`due to a system correction. ` +
`New seat: ${newSeat.seatNumber}, Coach: ${newSeat.coach.number} ` +
`(${newSeat.coach.coachType.name}). We apologize for the inconvenience.`;
if (dup.booking.contactPhone) {
await this.sms.sendSms({ to: dup.booking.contactPhone, message }).catch(() => null);
}
this.logger.log(
`Seat dedup: booking ${dup.booking.bookingRef} (${dup.passengerName}) ` +
`seat ${dup.seat.seatNumber} → ${newSeat.seatNumber} (coach ${newSeat.coach.number})`,
);
resolved++;
} catch (err) {
this.logger.error(
`Seat dedup error for ${dup.booking.bookingRef}: ` +
`${err instanceof Error ? err.message : String(err)}`,
);
unresolved++;
}
}
}
this.logger.log(
`Seat dedup complete: ${resolved} reassigned, ${unresolved} unresolved ` +
`across ${duplicateGroups.length} conflict(s)`,
);
}
// ─────────────────────────────────────────────────────────────────────────