fix(realtime): booking writes share one use case on REST and MCP

This commit is contained in:
Maurice
2026-10-22 12:14:25 +02:00
parent 262dd97679
commit 052964f4ab
15 changed files with 948 additions and 398 deletions
@@ -0,0 +1,141 @@
import type { Reservation, User } from '../../types';
import { DomainError } from '../common/domain-error';
import { restTripWriter, type TripWriter } from '../common/trip-writer';
import { AirtrailLinkService } from '../integrations/airtrail-link.service';
import { RealtimeService } from '../realtime/realtime.service';
import {
ReservationsService,
type CostEvent,
type CreateReservationData,
type UpdateReservationData,
} from './reservations.service';
import { Injectable } from '@nestjs/common';
/** The linked cost a booking write takes, as ReservationsService.createWithCost reads it. */
type BudgetEntry = Parameters<ReservationsService['createWithCost']>[2];
/** The day plan order a reorder writes, as ReservationsService.updatePositions reads it. */
type Positions = Parameters<ReservationsService['updatePositions']>[1];
const bookingNotFound = (mcpMessage: string) => new DomainError(404, 'Reservation not found', { mcpMessage });
/**
* The booking writes REST and MCP share, one method per use case: the write,
* then every event it sends and the notice to fellow travellers, so both
* surfaces send the same things in the same order.
*
* Each runs after its caller checked the trip and the references (REST's
* TripAccessGuard with reservation_edit and the controller's foreign id check,
* the tool's own checks on MCP), writes in ReservationsService's transaction and
* sends once that committed. The day stop a hotel booking brings is announced by
* ReservationsService itself, before the events here, to every socket.
*
* MCP used to send these events from the tool and had drifted: no booking
* notice, no push of an edited AirTrail flight (the next pull reverted the
* edit), no `accommodation:deleted` when a transport still carried a stay, its
* own day key on `reservation:positions`, and the hotel link announced a stay it
* had not changed.
*/
@Injectable()
export class ReservationWritesService {
constructor(
private readonly reservations: ReservationsService,
private readonly airtrailLink: AirtrailLinkService,
private readonly realtime: RealtimeService,
) {}
/** The REST caller of a booking write: skips the sender's socket, as the routes always did. */
restWriter(tripId: string, user: User, socketId: string | undefined): TripWriter {
return restTripWriter(this.realtime, tripId, user, socketId);
}
/** A new booking with its linked cost, and the stay a hotel booking brings. */
async createBooking(tripId: string, data: CreateReservationData, entry: BudgetEntry, writer: TripWriter) {
const { reservation, accommodationCreated, costEvents } = await this.reservations.createWithCost(
tripId,
data,
entry,
);
if (accommodationCreated) writer.events.emit('accommodation:created', {});
this.sendCost(costEvents, writer);
writer.events.emit('reservation:created', { reservation });
await this.reservations.notifyBookingChange(tripId, writer.userId, data.title, data.type ?? '');
return { reservation };
}
/**
* An edit of a booking the caller already read (`current`), with the costs it
* re-files. A stay the edit wrote for the first time is `accommodation:created`,
* one it changed `accommodation:updated`. A linked AirTrail flight is pushed
* back to AirTrail, or the next pull would overwrite the edit.
*
* @txIndependent the booking write commits in one transaction; the AirTrail
* push after it is network I/O against another system and may detach the link
* on its own, so it never shares that transaction.
*/
async updateBooking(
id: string,
tripId: string,
data: UpdateReservationData,
current: Reservation,
entry: BudgetEntry,
writer: TripWriter,
) {
const { reservation, accommodationChanged, accommodationCreated, costEvents } =
await this.reservations.updateWithCost(id, tripId, data, current, entry);
if (accommodationCreated) writer.events.emit('accommodation:created', {});
else if (accommodationChanged) writer.events.emit('accommodation:updated', {});
this.sendCost(costEvents, writer);
writer.events.emit('reservation:updated', { reservation });
// Fire and forget, under the importer's credentials (#214).
if (reservation.external_source === 'airtrail' && reservation.sync_enabled) {
void this.airtrailLink.pushReservationToAirtrail(Number(reservation.id), Number(tripId)).catch(() => {});
}
await this.reservations.notifyBookingChange(
tripId,
writer.userId,
data.title || current.title,
data.type || current.type || '',
);
return { reservation };
}
/**
* A booking deleted with its stay and its linked costs. `mcpNotFound` is the
* wording the tool has always answered a missing booking with.
*/
async deleteBooking(
id: string | number,
tripId: string | number,
writer: TripWriter,
mcpNotFound = 'Reservation not found.',
) {
const { deleted, accommodationDeleted, deletedBudgetItemIds } = await this.reservations.remove(id, tripId);
if (!deleted) throw bookingNotFound(mcpNotFound);
if (accommodationDeleted)
writer.events.emit('accommodation:deleted', { accommodationId: deleted.accommodation_id });
for (const itemId of deletedBudgetItemIds) writer.events.emit('budget:deleted', { itemId });
writer.events.emit('reservation:deleted', { reservationId: Number(id) });
await this.reservations.notifyBookingChange(tripId, writer.userId, deleted.title, deleted.type || '');
return { success: true };
}
/** The order of the bookings in the day plan, of one day when `dayId` names it. */
async reorderBookings(
tripId: string | number,
positions: Positions,
dayId: number | string | null | undefined,
writer: TripWriter,
) {
await this.reservations.updatePositions(tripId, positions, dayId);
// REST's `day_id` is the canonical key. The MCP tool always named it `dayId`,
// so its event keeps that key next to it for a client that reads it.
writer.events.emit(
'reservation:positions',
writer.surface === 'mcp' ? { positions, day_id: dayId, dayId } : { positions, day_id: dayId },
);
}
private sendCost(events: readonly CostEvent[], writer: TripWriter): void {
for (const e of events) writer.events.emit(e.event, e.payload);
}
}
@@ -1,8 +1,8 @@
import type { User } from '../../types';
import { CurrentUser } from '../auth-core/current-user.decorator';
import { JwtAuthGuard } from '../auth-core/jwt-auth.guard';
import { AirtrailLinkService } from '../integrations/airtrail-link.service';
import { RequirePermission, TripAccessGuard } from '../permissions/trip-access.guard';
import { ReservationWritesService } from './reservation-writes.service';
import {
ReservationCreateDto,
ReservationUpdateDto,
@@ -25,6 +25,8 @@ type ReservationBody = Record<string, unknown> & {
* trip access (404), 'reservation_edit' permission (403), create 201 / rest 200,
* the bespoke 404 bodies, the accommodation + budget side effects, the booking
* notifications, and all WebSocket broadcasts with the forwarded X-Socket-Id.
* Create, edit, delete and reorder run the use cases of ReservationWritesService,
* which the MCP tools share, so both surfaces send the same events.
* Invalid bodies (missing title, non-array positions/user_ids) now 400 through
* the global ZodValidationPipe envelope instead of the bespoke strings.
* /positions is declared before /:id so it wins over the param.
@@ -37,8 +39,7 @@ type ReservationBody = Record<string, unknown> & {
export class ReservationsController {
constructor(
private readonly reservations: ReservationsService,
// Injected from AirtrailCoreModule — the split that retired airtrail.bridge.
private readonly airtrailLink: AirtrailLinkService,
private readonly writes: ReservationWritesService,
) {}
@Get()
@@ -60,18 +61,12 @@ export class ReservationsController {
// at a rate frozen now (#2525).
const budgetEntry = await this.reservations.withFrozenRate(tripId, body.create_budget_entry);
// The booking and its linked cost are one write.
const { reservation, accommodationCreated, costEvents } = await this.reservations.createWithCost(
return this.writes.createBooking(
tripId,
body as never,
budgetEntry,
this.writes.restWriter(tripId, user, socketId),
);
if (accommodationCreated) {
this.reservations.broadcast(tripId, 'accommodation:created', {}, socketId);
}
this.reservations.announceCost(tripId, costEvents, socketId);
this.reservations.broadcast(tripId, 'reservation:created', { reservation }, socketId);
await this.reservations.notifyBookingChange(tripId, user.id, body.title, body.type ?? '');
return { reservation };
}
@RequirePermission('reservation_edit')
@@ -84,16 +79,11 @@ export class ReservationsController {
) {
// The legacy signature declares day_plan_position required, but the wire
// contract tolerates absent values (bind NULL) — see the shared schema.
await this.reservations.updatePositions(
await this.writes.reorderBookings(
tripId,
body.positions as { id: number; day_plan_position: number }[],
body.day_id,
);
this.reservations.broadcast(
tripId,
'reservation:positions',
{ positions: body.positions, day_id: body.day_id },
socketId,
this.writes.restWriter(tripId, user, socketId),
);
return { success: true };
}
@@ -113,27 +103,15 @@ export class ReservationsController {
throw new HttpException({ error: 'Reservation not found' }, 404);
}
await this.rejectForeignReferences(tripId, body);
// The booking and its linked cost are one write.
const { reservation, accommodationChanged, costEvents } = await this.reservations.updateWithCost(
// The booking and its linked cost are one write; a linked AirTrail flight is pushed back after it.
return this.writes.updateBooking(
id,
tripId,
body as never,
current,
body.create_budget_entry,
this.writes.restWriter(tripId, user, socketId),
);
if (accommodationChanged) {
this.reservations.broadcast(tripId, 'accommodation:updated', {}, socketId);
}
const cur = current as { title: string; type?: string };
this.reservations.announceCost(tripId, costEvents, socketId);
this.reservations.broadcast(tripId, 'reservation:updated', { reservation }, socketId);
// Push a locally-edited AirTrail flight back to AirTrail (fire-and-forget,
// under the importer's credentials — see airtrailSync). #214
if ((reservation as any)?.external_source === 'airtrail' && (reservation as any)?.sync_enabled) {
void this.airtrailLink.pushReservationToAirtrail(Number((reservation as any).id), Number(tripId)).catch(() => {});
}
await this.reservations.notifyBookingChange(tripId, user.id, body.title || cur.title, body.type || cur.type || '');
return { reservation };
}
@RequirePermission('reservation_edit')
@@ -166,24 +144,8 @@ export class ReservationsController {
@Param('id') id: string,
@Headers('x-socket-id') socketId?: string,
) {
const { deleted, accommodationDeleted, deletedBudgetItemIds } = await this.reservations.remove(id, tripId);
if (!deleted) {
throw new HttpException({ error: 'Reservation not found' }, 404);
}
if (accommodationDeleted) {
this.reservations.broadcast(
tripId,
'accommodation:deleted',
{ accommodationId: deleted.accommodation_id },
socketId,
);
}
for (const itemId of deletedBudgetItemIds) {
this.reservations.broadcast(tripId, 'budget:deleted', { itemId }, socketId);
}
this.reservations.broadcast(tripId, 'reservation:deleted', { reservationId: Number(id) }, socketId);
await this.reservations.notifyBookingChange(tripId, user.id, deleted.title, deleted.type || '');
return { success: true };
// A missing booking answers { error: 'Reservation not found' } with 404.
return this.writes.deleteBooking(id, tripId, this.writes.restWriter(tripId, user, socketId));
}
/**
@@ -9,6 +9,7 @@ import { z } from 'zod';
import { BudgetService } from '../budget/budget.service';
import { noAccess, permissionDenied } from '../../mcp/tools/_shared';
import { ReservationsService } from './reservations.service';
import { ReservationWritesService } from './reservation-writes.service';
import { DaysService } from '../days/days.service';
import { findByIata } from '../airports/airports.data';
import type { EndpointInput } from './reservations.service';
@@ -330,6 +331,9 @@ export class ReservationsMcp {
// positionally.
private readonly assignments: AssignmentsService,
private readonly guards: McpToolGuardsService,
// The booking use cases the REST routes run too: the events, the booking
// notice and the AirTrail push leave from there, not from the tools.
private readonly writes: ReservationWritesService,
) {}
@Tool({
@@ -388,20 +392,13 @@ export class ReservationsMcp {
const metadata = price != null ? { price: String(price) } : undefined;
// The booking and its linked cost are one write, the same service path REST takes.
const { reservation, accommodationCreated, costEvents } = await this.reservations.createWithCost(String(tripId), {
// The booking and its linked cost are one write, the same use case REST runs.
const { reservation } = await this.writes.createBooking(String(tripId), {
title, type, reservation_time, reservation_end_time, url, location, confirmation_number,
notes, day_id, place_id, assignment_id,
create_accommodation: createAccommodation,
metadata,
}, linkedCost(price, budget_category || type));
if (accommodationCreated) {
this.guards.safeBroadcast(tripId, 'accommodation:created', {});
}
for (const { event, payload } of costEvents) this.guards.safeBroadcast(tripId, event, payload);
this.guards.safeBroadcast(tripId, 'reservation:created', { reservation });
}, linkedCost(price, budget_category || type), await this.guards.tripWriter(tripId, ctx.userId));
return ok({ reservation });
}
@@ -446,15 +443,12 @@ export class ReservationsMcp {
return errorResult('assignment_id does not belong to this trip.');
// The booking and the linked expenses a type change re-files are one write,
// the same service path REST and the plugin RPC take.
const { reservation, accommodationChanged, costEvents } = await this.reservations.updateWithCost(String(reservationId), String(tripId), {
// the same use case REST runs.
const { reservation } = await this.writes.updateBooking(String(reservationId), String(tripId), {
title, type, reservation_time, reservation_end_time, url, location, confirmation_number, notes, status,
place_id: place_id !== undefined ? place_id ?? undefined : undefined,
assignment_id: assignment_id !== undefined ? assignment_id ?? undefined : undefined,
}, existing, undefined);
if (accommodationChanged) this.guards.safeBroadcast(tripId, 'accommodation:updated', {});
for (const { event, payload } of costEvents) this.guards.safeBroadcast(tripId, event, payload);
this.guards.safeBroadcast(tripId, 'reservation:updated', { reservation });
}, existing, undefined, await this.guards.tripWriter(tripId, ctx.userId));
return ok({ reservation });
}
@@ -471,14 +465,7 @@ export class ReservationsMcp {
async deleteReservation({ tripId, reservationId }: { tripId: number; reservationId: number }, ctx: McpContext) {
if (!(await this.reservations.verifyTripAccess(tripId, ctx.userId))) return noAccess();
if (!(await this.guards.hasTripPermission('reservation_edit', tripId, ctx.userId))) return permissionDenied();
const { deleted, accommodationDeleted, deletedBudgetItemIds } = await this.reservations.remove(reservationId, tripId);
if (!deleted) return errorResult('Reservation not found.');
if (accommodationDeleted) {
this.guards.safeBroadcast(tripId, 'accommodation:deleted', { accommodationId: deleted.accommodation_id });
}
for (const itemId of deletedBudgetItemIds) this.guards.safeBroadcast(tripId, 'budget:deleted', { itemId });
this.guards.safeBroadcast(tripId, 'reservation:deleted', { reservationId });
return ok({ success: true });
return ok(await this.writes.deleteBooking(reservationId, tripId, await this.guards.tripWriter(tripId, ctx.userId)));
}
@Tool({
@@ -542,8 +529,7 @@ export class ReservationsMcp {
if (dayId && !(await this.days.getDay(dayId, tripId)))
return errorResult('dayId does not belong to this trip.');
await this.reservations.updatePositions(tripId, positions, dayId);
this.guards.safeBroadcast(tripId, 'reservation:positions', { positions, dayId });
await this.writes.reorderBookings(tripId, positions, dayId, await this.guards.tripWriter(tripId, ctx.userId));
return ok({ success: true });
}
@@ -582,16 +568,14 @@ export class ReservationsMcp {
if (!(await this.days.getDay(end_day_id, tripId)))
return errorResult('end_day_id does not belong to this trip.');
const isNewAccommodation = !current.accommodation_id;
const { reservation } = await this.reservations.update(reservationId, tripId, {
// The edit the booking form's PUT makes, through the same use case: a stay
// written for the first time is announced as created, a changed one as updated.
const { reservation } = await this.writes.updateBooking(String(reservationId), String(tripId), {
place_id,
type: current.type,
status: current.status as string,
create_accommodation: { place_id, start_day_id, end_day_id, check_in: check_in || undefined, check_out: check_out || undefined },
}, current);
this.guards.safeBroadcast(tripId, isNewAccommodation ? 'accommodation:created' : 'accommodation:updated', {});
this.guards.safeBroadcast(tripId, 'reservation:updated', { reservation });
}, current, undefined, await this.guards.tripWriter(tripId, ctx.userId));
return ok({ reservation, accommodation_id: reservation?.accommodation_id ?? null });
}
@@ -736,7 +720,7 @@ export class ReservationsMcp {
if (price != null) meta.price = String(price);
const { reservation, costEvents } = await this.reservations.createWithCost(String(tripId), {
const { reservation } = await this.writes.createBooking(String(tripId), {
title,
type,
reservation_time: departureTime,
@@ -751,10 +735,7 @@ export class ReservationsMcp {
metadata: Object.keys(meta).length > 0 ? meta : undefined,
endpoints: transportEndpoints,
needs_review,
}, linkedCost(price, budget_category || type));
for (const { event, payload } of costEvents) this.guards.safeBroadcast(tripId, event, payload);
this.guards.safeBroadcast(tripId, 'reservation:created', { reservation });
}, linkedCost(price, budget_category || type), await this.guards.tripWriter(tripId, ctx.userId));
return ok({ reservation });
}
@@ -856,8 +837,9 @@ export class ReservationsMcp {
if (applied.endpointsChanged) resolvedEndpoints = applied.endpoints;
}
// One write with the linked expenses a type change re-files, as on update_reservation.
const { reservation, accommodationChanged, costEvents } = await this.reservations.updateWithCost(String(reservationId), String(tripId), {
// One write with the linked expenses a type change re-files, as on update_reservation;
// a linked AirTrail flight is pushed back, as on REST.
const { reservation } = await this.writes.updateBooking(String(reservationId), String(tripId), {
title,
type,
reservation_time: departureTime,
@@ -871,10 +853,7 @@ export class ReservationsMcp {
metadata: nextMetadata,
endpoints: resolvedEndpoints,
needs_review,
}, existing, undefined);
if (accommodationChanged) this.guards.safeBroadcast(tripId, 'accommodation:updated', {});
for (const { event, payload } of costEvents) this.guards.safeBroadcast(tripId, event, payload);
this.guards.safeBroadcast(tripId, 'reservation:updated', { reservation });
}, existing, undefined, await this.guards.tripWriter(tripId, ctx.userId));
return ok({ reservation });
}
@@ -891,10 +870,9 @@ export class ReservationsMcp {
async deleteTransport({ tripId, reservationId }: { tripId: number; reservationId: number }, ctx: McpContext) {
if (!(await this.reservations.verifyTripAccess(tripId, ctx.userId))) return noAccess();
if (!(await this.guards.hasTripPermission('reservation_edit', tripId, ctx.userId))) return permissionDenied();
const { deleted, deletedBudgetItemIds } = await this.reservations.remove(reservationId, tripId);
if (!deleted) return errorResult('Transport not found.');
for (const itemId of deletedBudgetItemIds) this.guards.safeBroadcast(tripId, 'budget:deleted', { itemId });
this.guards.safeBroadcast(tripId, 'reservation:deleted', { reservationId });
return ok({ success: true });
// The same cascade as delete_reservation: a row a REST edit retyped from a
// hotel still takes its stay along, and says so with accommodation:deleted.
const writer = await this.guards.tripWriter(tripId, ctx.userId);
return ok(await this.writes.deleteBooking(reservationId, tripId, writer, 'Transport not found.'));
}
}
@@ -20,6 +20,7 @@ import { McpSharedModule } from '../mcp-shared/mcp-shared.module';
import { NotificationsModule } from '../notifications/notifications.module';
import { PermissionsModule } from '../permissions/permissions.module';
import { RealtimeModule } from '../realtime/realtime.module';
import { ReservationWritesService } from './reservation-writes.service';
import { ReservationsReadModule } from './reservations-read.module';
import { ReservationsController } from './reservations.controller';
import { ReservationsMcp } from './reservations.mcp';
@@ -79,7 +80,8 @@ import { Module } from '@nestjs/common';
]),
],
controllers: [ReservationsController, UpcomingReservationsController],
providers: [ReservationsService, ReservationsMcp, ReservationsRpc],
// ReservationWritesService: the booking use cases the controller and the MCP tools share.
providers: [ReservationsService, ReservationWritesService, ReservationsMcp, ReservationsRpc],
// For in-container consumers (ReservationsRpc, TripsService, BookingImportService).
exports: [ReservationsService],
})
@@ -68,14 +68,17 @@ export class ReservationsRpc {
if (!current) throw new ForbiddenResource(`no reservation ${reservationId} on trip ${tripId}`);
await this.requireOwnReferences(tripId, input);
const i = input as { title?: string; type?: string; create_budget_entry?: unknown };
const { reservation, accommodationChanged, costEvents } = await this.reservations.updateWithCost(
String(reservationId),
String(tripId),
input as never,
current,
i.create_budget_entry as never,
);
if (accommodationChanged) this.realtime.broadcast(tripId, 'accommodation:updated', {}, undefined);
const { reservation, accommodationChanged, accommodationCreated, costEvents } =
await this.reservations.updateWithCost(
String(reservationId),
String(tripId),
input as never,
current,
i.create_budget_entry as never,
);
// A stay this edit wrote for the first time is new, as the REST route announces it.
if (accommodationCreated) this.realtime.broadcast(tripId, 'accommodation:created', {}, undefined);
else if (accommodationChanged) this.realtime.broadcast(tripId, 'accommodation:updated', {}, undefined);
const cur = current as { title: string; type?: string };
this.reservations.announceCost(tripId, costEvents, undefined);
this.realtime.broadcast(tripId, 'reservation:updated', { reservation }, undefined);
@@ -43,7 +43,7 @@ import { decodeJson } from '../../utils/json-column';
type Trip = TripAccess;
type BudgetEntry = { total_price?: number; category?: string; currency?: string | null; exchange_rate?: number } | undefined;
/** A booking's cost as the service writes it: the entry, dated on the booking's day (bookingDate), never by the caller. */
/** A booking's cost as written: the entry, dated on the booking's day (bookingDate), never by the caller. */
type DatedBudgetEntry = (NonNullable<BudgetEntry> & { expense_date?: string | null }) | undefined;
/** What a booking write did to its linked costs, for the caller to send once the write has committed. */
@@ -842,7 +842,7 @@ export class ReservationsService {
return written;
}
private async updateInTx(id: string | number, tripId: string | number, data: UpdateReservationData, current: Reservation): Promise<{ reservation: ReservationRow; accommodationChanged: boolean; stayMirror: AccommodationMirror }> {
private async updateInTx(id: string | number, tripId: string | number, data: UpdateReservationData, current: Reservation) {
const {
title, reservation_time, reservation_end_time, location,
confirmation_number, notes, url, day_id, end_day_id, place_id, assignment_id,
@@ -891,6 +891,7 @@ export class ReservationsService {
const accExists = await this.dayAccommodationsRepo.existsInTrip(accIdForRead(resolvedAccId)!, this.rowIdNum(tripId));
if (!accExists) resolvedAccId = null;
}
const hadStay = Boolean(resolvedAccId);
if (type === 'hotel' && create_accommodation) {
const { place_id: accPlaceId, start_day_id, end_day_id, check_in, check_out, confirmation: accConf } = create_accommodation;
if (start_day_id && end_day_id) {
@@ -1027,7 +1028,7 @@ export class ReservationsService {
// The caller passed the pre-checked `current` row, so the re-select can't
// miss (legacy typed this any).
const reservation = (await this.getReservationWithJoins(id))!;
return { reservation, accommodationChanged, stayMirror };
return { reservation, accommodationChanged, accommodationCreated: !hadStay && Boolean(resolvedAccId), stayMirror };
}
/** The accommodation + budget-item + reservation deletes are one logical