diff --git a/apps/edr-passenger-api/src/modules/payments/payment-sync.service.ts b/apps/edr-passenger-api/src/modules/payments/payment-sync.service.ts new file mode 100644 index 000000000..7be85714a --- /dev/null +++ b/apps/edr-passenger-api/src/modules/payments/payment-sync.service.ts @@ -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`, + ); + } + } +} diff --git a/apps/edr-passenger-api/src/modules/payments/payments.module.ts b/apps/edr-passenger-api/src/modules/payments/payments.module.ts index 2a3906398..07f452e43 100644 --- a/apps/edr-passenger-api/src/modules/payments/payments.module.ts +++ b/apps/edr-passenger-api/src/modules/payments/payments.module.ts @@ -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], diff --git a/apps/edr-passenger-api/src/modules/tasks/tasks.module.ts b/apps/edr-passenger-api/src/modules/tasks/tasks.module.ts index 335d11f81..fb8a29a6e 100644 --- a/apps/edr-passenger-api/src/modules/tasks/tasks.module.ts +++ b/apps/edr-passenger-api/src/modules/tasks/tasks.module.ts @@ -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 {} diff --git a/apps/edr-passenger-api/src/modules/tasks/tasks.service.ts b/apps/edr-passenger-api/src/modules/tasks/tasks.service.ts index 7a9208cae..ab275ee09 100644 --- a/apps/edr-passenger-api/src/modules/tasks/tasks.service.ts +++ b/apps/edr-passenger-api/src/modules/tasks/tasks.service.ts @@ -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(); + 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>(); // scheduleId → Set + 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, + ) { + 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(); + 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>(); + 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(); + + 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)`, + ); } // ─────────────────────────────────────────────────────────────────────────