implement batch board notification system for schedule changes and updates

apps/edr-freight-api/src/modules/train-scheduling/booking-batch.service.ts

@@ -532,6 +532,7 @@ export class BookingBatchService implements OnModuleInit {
         `Wagon allocation issue for ${booking.reference ?? bookingId}: ${issue?.issue ?? issue?.status}`,
       );
     }
+    this.notifyBoardChanged(booking.trainScheduleId, "booking_paid_allocated");
   }
 
   /** Customer paid — delegate to ensurePaidBookingAllocated. */
@@ -766,6 +767,7 @@ export class BookingBatchService implements OnModuleInit {
     if (schedule && (await this.isTrainFull(schedule))) {
       await this.setWindow(scheduleId, 'FULL');
     }
+    this.notifyBoardChanged(scheduleId, 'export_booking_accepted');
   }
 
   /** Link PAID bookings that have no train_schedule_bookings row (cron backstop). */
@@ -778,6 +780,9 @@ export class BookingBatchService implements OnModuleInit {
         `Reconciled PAID booking ${booking.reference ?? booking.id} → schedule ${scheduleId}`,
       );
     }
+    if (unlinked.length > 0) {
+      this.notifyBoardChanged(scheduleId, "paid_reconciled");
+    }
   }
 
   // ---- legacy fill entry point ----------------------------------------------
@@ -1159,7 +1164,12 @@ export class BookingBatchService implements OnModuleInit {
 
   /** Run wagon-level allocation for all eligible linked bookings on a schedule. */
   async runWagonAllocation(scheduleId: string) {
-    return this.trainSchedulingService.tryAutoWagonAllocation(scheduleId);
+    const result =
+      await this.trainSchedulingService.tryAutoWagonAllocation(scheduleId);
+    if (result.assignedBookingIds.length > 0) {
+      this.notifyBoardChanged(scheduleId, "wagon_allocation_run");
+    }
+    return result;
   }
 
   /**
@@ -1329,6 +1339,7 @@ export class BookingBatchService implements OnModuleInit {
     this.resortPoolByPriority(pool);
     const units = this.groupConsolidatedPool(pool);
     let armed = false;
+    let preempted = false;
     let reservedThisPass = 0;
     let commercialReserved = 0;
 
@@ -1367,6 +1378,7 @@ export class BookingBatchService implements OnModuleInit {
             budget,
             wagonDims,
           );
+          preempted = true;
           if (!freed) continue; // still doesn't fit even after preempt
         } else {
           // Doesn't fit whole. A split-eligible import booking is offered the part
@@ -1415,6 +1427,10 @@ export class BookingBatchService implements OnModuleInit {
     );
     if (budget.isExhausted(minPerWagon)) await this.setWindow(scheduleId, "FULL");
     if (armed) this.armSettle(scheduleId);
+    // One push per fill pass (never per booking) — only when rows changed.
+    if (reservedThisPass > 0 || armed || preempted) {
+      this.notifyBoardChanged(scheduleId, "batch_fill");
+    }
     void this.triggerWagonAllocation(scheduleId);
     return commercialReserved;
   }
@@ -1515,8 +1531,13 @@ export class BookingBatchService implements OnModuleInit {
 
     const wagonDims = await this.loadWagonDims();
 
-    // Live per-schedule corridor budget + arm flag, in departure order.
-    const trains: Array<{ id: string; budget: CorridorBudget; armed: boolean }> = [];
+    // Live per-schedule corridor budget + arm/changed flags, in departure order.
+    const trains: Array<{
+      id: string;
+      budget: CorridorBudget;
+      armed: boolean;
+      changed: boolean;
+    }> = [];
     for (const id of scheduleIds) {
       const schedule =
         await this.trainSchedulesRepository.findByIdWithFullGraph(id);
@@ -1530,7 +1551,7 @@ export class BookingBatchService implements OnModuleInit {
       const limits = await this.capacityLimits(locomotive);
       await this.syncScheduleMaxWagons(schedule, locomotive);
       const budget = await this.remainingBudget(schedule, limits, wagonDims);
-      trains.push({ id, budget, armed: false });
+      trains.push({ id, budget, armed: false, changed: false });
     }
     if (trains.length === 0) return { scheduleIds, commercialReserved: 0 };
 
@@ -1603,6 +1624,9 @@ export class BookingBatchService implements OnModuleInit {
             t.budget,
             wagonDims,
           );
+          // Preempt may have displaced (expired) victims even when the need
+          // still doesn't fit — the board must refresh either way.
+          t.changed = true;
           if (freed) {
             target = t;
             break;
@@ -1647,6 +1671,7 @@ export class BookingBatchService implements OnModuleInit {
           commercialReserved += 1;
         }
         target.budget.subtract(need, legOn(target)!);
+        target.changed = true;
         reservedThisPass += 1;
       } catch (err) {
         this.logger.error(
@@ -1664,6 +1689,10 @@ export class BookingBatchService implements OnModuleInit {
     for (const t of trains) {
       if (t.budget.isExhausted(minPerWagon)) await this.setWindow(t.id, "FULL");
       if (t.armed) this.armSettle(t.id);
+      // One push per touched train per pass (never per booking). `armed` covers
+      // commercial reserves + partial offers; `changed` covers gov allocations
+      // and preemption.
+      if (t.armed || t.changed) this.notifyBoardChanged(t.id, "batch_fill");
       void this.triggerWagonAllocation(t.id);
     }
 
@@ -1910,6 +1939,10 @@ export class BookingBatchService implements OnModuleInit {
           `— payment phase extended for them`,
       );
     }
+    // Emitted here (not in settleDueReservations/settleBatch, which both wrap
+    // this) so one settle produces one push, after every allocation/expiry/
+    // top-up extension for this schedule has been persisted.
+    this.notifyBoardChanged(scheduleId, "reservations_settled");
     return true;
   }
 
@@ -1962,6 +1995,21 @@ export class BookingBatchService implements OnModuleInit {
       );
   }
 
+  /**
+   * Announce that a schedule's batch-board data changed so open boards refetch.
+   * Called AFTER the state change is persisted; a push failure only logs — it
+   * must never break the business transaction that triggered it.
+   */
+  private notifyBoardChanged(scheduleId: string, reason: string): void {
+    try {
+      this.bookingWindowGateway.emitBatchChanged(scheduleId, reason);
+    } catch (err) {
+      this.logger.warn(
+        `Batch-board push (${reason}) failed for ${scheduleId}: ${(err as Error).message}`,
+      );
+    }
+  }
+
   // ---- staff override actions ----------------------------------------------
 
   /** Staff "mark paid" override → set PAID and allocate immediately (don't wait for settle). */
@@ -1987,6 +2035,7 @@ export class BookingBatchService implements OnModuleInit {
       await this.setWindow(booking.trainScheduleId, "FULL");
     }
     void this.triggerWagonAllocation(booking.trainScheduleId!);
+    this.notifyBoardChanged(booking.trainScheduleId, "booking_marked_paid");
   }
 
   /**
@@ -2021,6 +2070,7 @@ export class BookingBatchService implements OnModuleInit {
       );
     }
 
+    const sourceScheduleId = booking.trainScheduleId ?? null;
     await this.dataSource.transaction(async (manager) => {
       if (booking.trainScheduleId) {
         await this.trainScheduleBookingsRepository.deleteByScheduleAndBooking(
@@ -2043,6 +2093,11 @@ export class BookingBatchService implements OnModuleInit {
         selectedForBatchAt: null,
       } as never);
     });
+    // Both boards changed: the booking left the source train and joined the target.
+    if (sourceScheduleId && sourceScheduleId !== newScheduleId) {
+      this.notifyBoardChanged(sourceScheduleId, "booking_moved");
+    }
+    this.notifyBoardChanged(newScheduleId, "booking_moved");
   }
 
   /** Staff "expire" override → free a reservation now (booking becomes EXPIRED). */
@@ -2060,6 +2115,8 @@ export class BookingBatchService implements OnModuleInit {
       if (topUpReserved > 0) {
         await this.extendPaymentPhaseForTopUp(freedScheduleId);
       }
+      // After the top-up + phase extension so one push carries the final state.
+      this.notifyBoardChanged(freedScheduleId, "reservation_expired");
     }
   }
 
@@ -2099,10 +2156,12 @@ export class BookingBatchService implements OnModuleInit {
         .update(booking.id, { trainScheduleId: scheduleId });
       booking.trainScheduleId = scheduleId;
       await this.allocate(scheduleId, booking, 'gov');
+      this.notifyBoardChanged(scheduleId, 'intercity_accepted');
       return;
     }
     await this.reserve(booking, scheduleId);
     this.armSettle(scheduleId);
+    this.notifyBoardChanged(scheduleId, 'intercity_accepted');
   }
 
   // ---- mutations ------------------------------------------------------------
@@ -2365,7 +2424,12 @@ export class BookingBatchService implements OnModuleInit {
           day,
         );
     const leftovers = pool.filter((b) => !b.isGovernment);
+    // Capture pinned schedules BEFORE expire() clears trainScheduleId, so each
+    // touched board gets exactly one push at the end of the sweep.
+    const touchedScheduleIds = new Set<string>();
+    if (leftovers.length) touchedScheduleIds.add(scheduleId);
     for (const booking of leftovers) {
+      if (booking.trainScheduleId) touchedScheduleIds.add(booking.trainScheduleId);
       await this.expire(booking, "no-capacity");
     }
     if (leftovers.length) {
@@ -2374,6 +2438,9 @@ export class BookingBatchService implements OnModuleInit {
           `expired ${leftovers.length} waiting booking(s)`,
       );
     }
+    for (const id of touchedScheduleIds) {
+      this.notifyBoardChanged(id, "day_pool_expired");
+    }
     return leftovers.length;
   }
 
@@ -2436,7 +2503,12 @@ export class BookingBatchService implements OnModuleInit {
           `on ${group.originYardId}->${group.destinationYardId} ${group.day}`,
       );
     }
+    // Only bookings pinned to a train show on a board — collect their schedules
+    // and push once per schedule after the sweep (most unaccepted rows are
+    // unpinned under day-level pooling, so this usually emits nothing).
+    const touchedScheduleIds = new Set<string>();
     for (const booking of unaccepted) {
+      if (booking.trainScheduleId) touchedScheduleIds.add(booking.trainScheduleId);
       await this.bookingsRepository.update(booking.id, {
         status: "EXPIRED",
         schedulingStatus: "ELIGIBLE",
@@ -2453,6 +2525,9 @@ export class BookingBatchService implements OnModuleInit {
         `[BATCH] EXPIRED (unaccepted) ${booking.reference}:${booking.id} at doc-review end`,
       );
     }
+    for (const id of touchedScheduleIds) {
+      this.notifyBoardChanged(id, "unaccepted_expired");
+    }
   }
 
   /**

apps/edr-freight-api/src/modules/train-scheduling/booking-window.gateway.ts

@@ -1,6 +1,7 @@
 import {
   BOOKING_WINDOW_WS_EVENTS,
   BOOKING_WINDOW_WS_NAMESPACE,
+  type BatchBoardChangedEvent,
   type BookingWindowPhaseEvent,
 } from '@edr/types';
 import { Logger } from '@nestjs/common';
@@ -65,6 +66,20 @@ export class BookingWindowGateway implements OnGatewayConnection {
     this.server.emit(BOOKING_WINDOW_WS_EVENTS.PHASE, payload);
   }
 
+  /**
+   * Announce that a schedule's batch-board data changed (payment, allocation,
+   * fill, expiry, …). Broadcast namespace-wide like PHASE — the payload carries
+   * no board data, only the scheduleId; clients holding that board refetch.
+   */
+  emitBatchChanged(scheduleId: string, reason: string): void {
+    const payload: BatchBoardChangedEvent = {
+      scheduleId,
+      reason,
+      at: new Date().toISOString(),
+    };
+    this.server.emit(BOOKING_WINDOW_WS_EVENTS.BATCH_CHANGED, payload);
+  }
+
   private extractToken(socket: Socket): string | undefined {
     const authToken = socket.handshake.auth?.token as string | undefined;
     if (authToken) return authToken;

apps/edr-freight-api/src/modules/train-scheduling/booking-window.service.ts

@@ -153,6 +153,16 @@ export class BookingWindowService implements OnModuleInit {
         .update(s.id, { docReviewCompletedAt: now });
       s.docReviewCompletedAt = now;
       await this.advanceSchedule(s, effectiveWindowConfig(s, liveCfg), now);
+      // The batch fill emits only for trains it actually reserved onto — this
+      // covers the empty-pool case so every open board still refetches. Push
+      // failures only log; the doc-review completion itself already persisted.
+      try {
+        this.gateway.emitBatchChanged(s.id, 'doc_review_completed');
+      } catch (err) {
+        this.logger.warn(
+          `Batch-board push failed for ${s.id}: ${(err as Error).message}`,
+        );
+      }
     }
     const fresh = await this.trainSchedulesRepository.findById(scheduleId);
     return fresh ?? schedule;