diff --git a/apps/api/src/domains/bookingPublicationCommandRepository.integration.test.ts b/apps/api/src/domains/bookingPublicationCommandRepository.integration.test.ts index 81d0eb3583..02c3101922 100644 --- a/apps/api/src/domains/bookingPublicationCommandRepository.integration.test.ts +++ b/apps/api/src/domains/bookingPublicationCommandRepository.integration.test.ts @@ -1,4 +1,6 @@ -import { createProductReadinessResult } from "@vayada/domain-hotels"; +import { createProductReadinessResult, type ReadinessProviderFailure } from "@vayada/domain-hotels"; +import type { BookingPublicContent } from "@vayada/domain-distribution/booking-publication"; +import type { BookingPublicationBuilderPort } from "@vayada/domain-distribution/booking-publication-builder"; import type { ReadyBookingPublicationEvidence, RequestBookingPublicationCommand, @@ -28,6 +30,48 @@ const secondOrganizationId = "74747474-7474-4747-8747-747474747406"; const propertyId = "74747474-7474-4747-8747-747474747403"; const activeRevisionId = "74747474-7474-4747-8747-747474747404"; const resultRevisionId = "74747474-7474-4747-8747-747474747405"; +// The builder contract is tested in domain-distribution; this fixture isolates persistence and CAS. +const publicContent = { + contractVersion: "booking-public-content.v1", + profile: { hotel: { propertyId }, branding: undefined }, + rooms: [{ roomTypeId: "room-deluxe" }], + calendar: { sourceRevision: "calendar-r1" }, + payments: { readyMethods: ["pay_at_property"] }, +} as unknown as BookingPublicContent; +type BuilderMode = + | "built" + | "owner_snapshot_unavailable" + | "source_manifest_mismatch" + | "public_content_incomplete" + | "wrong_hash" + | "wrong_readiness_hash"; +let builderMode: BuilderMode = "built"; +let readinessProviderAvailable = true; +const builder: BookingPublicationBuilderPort = { + async build({ readiness }) { + if ( + builderMode !== "built" && + builderMode !== "wrong_hash" && + builderMode !== "wrong_readiness_hash" + ) { + return { outcome: "rejected", code: builderMode }; + } + return { + outcome: "built", + build: { + sourceManifestHash: + builderMode === "wrong_hash" + ? (`sha256:${"0".repeat(64)}` as const) + : readiness.sourceManifestHash, + readinessHash: + builderMode === "wrong_readiness_hash" + ? (`sha256:${"1".repeat(64)}` as const) + : readiness.readinessHash, + publicContent, + }, + }; + }, +}; describe.skipIf(!TEST_DATABASE_URL)("PostgreSQL Booking publication command safety", () => { let currentReadinessRevision = "booking-settings:4"; @@ -43,8 +87,12 @@ describe.skipIf(!TEST_DATABASE_URL)("PostgreSQL Booking publication command safe projection: distributionPublication, attempts: attemptStatus, readiness: { - getBookingReadiness: async () => readinessEvidence(currentReadinessRevision), + getBookingReadiness: async () => + readinessProviderAvailable + ? readinessEvidence(currentReadinessRevision) + : readinessProviderFailure(), }, + builder, now: () => new Date("2026-08-02T13:01:00.000Z"), }); const repository = createPgBookingPublicationCommandRepository({ @@ -61,6 +109,8 @@ describe.skipIf(!TEST_DATABASE_URL)("PostgreSQL Booking publication command safe beforeEach(async () => { currentReadinessRevision = "booking-settings:4"; + builderMode = "built"; + readinessProviderAvailable = true; await cleanup(); await seedAuthorizedScope(); }); @@ -330,7 +380,6 @@ describe.skipIf(!TEST_DATABASE_URL)("PostgreSQL Booking publication command safe }); it("projects the required outbox intent and reports success only after CAS activation", async () => { - await seedPublicBookabilityProfile(); const accepted = await repository.requestPublication(await command("project-success")); if (!accepted.ok) throw new Error("Expected accepted publication request"); @@ -374,8 +423,10 @@ describe.skipIf(!TEST_DATABASE_URL)("PostgreSQL Booking publication command safe outboxStatus: "published", publicContent: { contractVersion: "booking-public-content.v1", - profile: { publicId: "publication-hotel" }, - roomOffers: [], + profile: { hotel: { propertyId } }, + rooms: [{ roomTypeId: "room-deluxe" }], + calendar: { sourceRevision: "calendar-r1" }, + payments: { readyMethods: ["pay_at_property"] }, }, }, ], @@ -417,7 +468,6 @@ describe.skipIf(!TEST_DATABASE_URL)("PostgreSQL Booking publication command safe }); it("rejects owner-source drift that occurs after command acceptance", async () => { - await seedPublicBookabilityProfile(); const accepted = await repository.requestPublication(await command("projector-source-drift")); if (!accepted.ok) throw new Error("Expected accepted publication request"); currentReadinessRevision = "booking-settings:5"; @@ -443,8 +493,59 @@ describe.skipIf(!TEST_DATABASE_URL)("PostgreSQL Booking publication command safe }); }); + it("retries a readiness provider failure and publishes after recovery", async () => { + const accepted = await repository.requestPublication(await command("readiness-recovery")); + if (!accepted.ok) throw new Error("Expected accepted publication request"); + readinessProviderAvailable = false; + + await expect(projector.projectPending({ propertyId })).resolves.toEqual({ + processed: 1, + succeeded: 0, + failed: 1, + exhausted: 0, + }); + await expect(distributionPublication.getActive(propertyId)).resolves.toBeNull(); + + readinessProviderAvailable = true; + await expect(projector.projectPending({ propertyId })).resolves.toEqual({ + processed: 1, + succeeded: 1, + failed: 0, + exhausted: 0, + }); + await expect(distributionPublication.getActive(propertyId)).resolves.toMatchObject({ + revisionId: accepted.operation.operationId, + }); + }); + + it.each([ + ["source_manifest_mismatch", "source_content_changed"], + ["public_content_incomplete", "projection_failed"], + ["wrong_hash", "source_content_changed"], + ["wrong_readiness_hash", "source_content_changed"], + ] as const)("fails safely when the builder returns %s", async (mode, failureCode) => { + builderMode = mode; + const accepted = await repository.requestPublication(await command(`builder-${mode}`)); + if (!accepted.ok) throw new Error("Expected accepted publication request"); + + await expect(projector.projectPending({ propertyId })).resolves.toEqual({ + processed: 1, + succeeded: 0, + failed: 1, + exhausted: 0, + }); + await expect(distributionPublication.getActive(propertyId)).resolves.toBeNull(); + await expect( + repository.getPublicationStatus({ + actorUserId, + organizationId, + propertyId, + operationId: accepted.operation.operationId, + }), + ).resolves.toMatchObject({ status: "failed", failureCode }); + }); + it("retries status terminalization without failing an already-active revision", async () => { - await seedPublicBookabilityProfile(); const accepted = await repository.requestPublication(await command("status-write-retry")); if (!accepted.ok) throw new Error("Expected accepted publication request"); const statusFailure = createBookingPublicationProjector({ @@ -452,6 +553,7 @@ describe.skipIf(!TEST_DATABASE_URL)("PostgreSQL Booking publication command safe projection: distributionPublication, attempts: failSucceeded(attemptStatus), readiness: { getBookingReadiness: async () => readinessEvidence() }, + builder, now: () => new Date("2026-08-02T13:01:00.000Z"), }); try { @@ -497,7 +599,6 @@ describe.skipIf(!TEST_DATABASE_URL)("PostgreSQL Booking publication command safe }); it("reconciles an active revision before recovering an expired final lease", async () => { - await seedPublicBookabilityProfile(); const request = await command("expired-active-lease"); const accepted = await repository.requestPublication(request); if (!accepted.ok) throw new Error("Expected accepted publication request"); @@ -531,6 +632,7 @@ describe.skipIf(!TEST_DATABASE_URL)("PostgreSQL Booking publication command safe expectedPropertyLifecycleRevision: 1, requestedByUserId: actorUserId, readiness: request.readiness, + publicContent, projectedAt: new Date("2026-08-02T13:00:30.000Z"), }); await admin.query( @@ -640,8 +742,55 @@ describe.skipIf(!TEST_DATABASE_URL)("PostgreSQL Booking publication command safe } }); + it("refuses to reuse an operation for different immutable public content", async () => { + const request = await command("immutable-content-retry"); + const accepted = await repository.requestPublication(request); + if (!accepted.ok) throw new Error("Expected accepted publication request"); + await admin.query( + `UPDATE platform.outbox_events + SET status = 'leased', leased_until = '2026-08-02T13:01:30.000Z'::timestamptz, + outbox_metadata = jsonb_set( + outbox_metadata, + '{bookingPublicationProjection}', + jsonb_build_object('leaseToken', 'immutable-content'), + true + ) + WHERE resource_id = $1 + AND event_type = 'booking.publication.requested'`, + [accepted.operation.operationId], + ); + const input = { + operationId: accepted.operation.operationId, + outboxEventId: await outboxEventId(accepted.operation.operationId), + outboxLeaseToken: "immutable-content", + propertyId, + expectedActiveRevisionId: null, + expectedPropertyLifecycleRevision: 1, + requestedByUserId: actorUserId, + readiness: request.readiness, + publicContent, + projectedAt: new Date("2026-08-02T13:01:00.000Z"), + }; + await distributionPublication.projectPublication(input); + await expect(distributionPublication.projectPublication(input)).resolves.toMatchObject({ + revisionId: accepted.operation.operationId, + }); + + await expect( + distributionPublication.projectPublication({ + ...input, + publicContent: { + ...publicContent, + rooms: [{ roomTypeId: "different-room" }], + } as unknown as BookingPublicContent, + }), + ).rejects.toThrow("Stored Booking publication revision does not match its operation"); + await expect(distributionPublication.getActive(propertyId)).resolves.toMatchObject({ + revisionId: accepted.operation.operationId, + }); + }); + it("fences a paused activation against lease recovery and source drift", async () => { - await seedPublicBookabilityProfile(); const accepted = await repository.requestPublication(await command("lease-fence-race")); if (!accepted.ok) throw new Error("Expected accepted publication request"); const projectionPool = new pg.Pool({ connectionString: TEST_DATABASE_URL!, max: 1 }); @@ -658,6 +807,7 @@ describe.skipIf(!TEST_DATABASE_URL)("PostgreSQL Booking publication command safe projection: pausedProjection, attempts: failSucceeded(attemptStatus), readiness: { getBookingReadiness: async () => readinessEvidence() }, + builder, leaseDurationMs: 1, now: () => new Date("2026-08-02T13:01:00.000Z"), }); @@ -669,6 +819,7 @@ describe.skipIf(!TEST_DATABASE_URL)("PostgreSQL Booking publication command safe readiness: { getBookingReadiness: async () => readinessEvidence(currentReadinessRevision), }, + builder, now: () => new Date("2026-08-02T13:01:00.002Z"), }); try { @@ -715,6 +866,7 @@ describe.skipIf(!TEST_DATABASE_URL)("PostgreSQL Booking publication command safe }); it("rolls exhausted outbox and dead-letter state back when attempt failure cannot commit", async () => { + builderMode = "owner_snapshot_unavailable"; const accepted = await repository.requestPublication(await command("atomic-exhaustion")); if (!accepted.ok) throw new Error("Expected accepted publication request"); await admin.query( @@ -729,6 +881,7 @@ describe.skipIf(!TEST_DATABASE_URL)("PostgreSQL Booking publication command safe projection: distributionPublication, attempts: failTransactionFailure(attemptStatus), readiness: { getBookingReadiness: async () => readinessEvidence() }, + builder, now: () => new Date("2026-08-02T13:01:00.000Z"), }); try { @@ -754,6 +907,7 @@ describe.skipIf(!TEST_DATABASE_URL)("PostgreSQL Booking publication command safe }); it("exhausts unavailable public projection safely without reporting success", async () => { + builderMode = "owner_snapshot_unavailable"; const accepted = await repository.requestPublication(await command("projector-unavailable")); if (!accepted.ok) throw new Error("Expected accepted publication request"); await admin.query( @@ -1104,6 +1258,24 @@ async function readinessEvidence( return readiness as ReadyBookingPublicationEvidence; } +function readinessProviderFailure(): ReadinessProviderFailure { + return { + outcome: "provider_failure", + contractVersion: "onboarding-product-readiness.v1", + propertyId, + product: "booking", + status: "error", + error: { + kind: "system_error", + errorSource: "provider", + code: "readiness_unavailable", + message: "Readiness could not be evaluated.", + retryable: true, + }, + evaluatedAt: "2026-08-02T13:01:00.000Z", + }; +} + function failOutboxPool(pool: pg.Pool): BookingPublicationCommandPool { return { async connect() { diff --git a/apps/api/src/domains/bookingPublicationProjector.ts b/apps/api/src/domains/bookingPublicationProjector.ts index da3f7c01ca..53025686c6 100644 --- a/apps/api/src/domains/bookingPublicationProjector.ts +++ b/apps/api/src/domains/bookingPublicationProjector.ts @@ -1,11 +1,14 @@ import { randomUUID } from "node:crypto"; import { + createReadyProductReadinessEvidence, hashSourceManifest, type ProductReadinessResult, + type ReadyProductReadinessEvidence, type ReadinessProviderFailure, type SourceManifest, } from "@vayada/domain-hotels"; +import type { BookingPublicationBuilderPort } from "@vayada/domain-distribution/booking-publication-builder"; import pg, { type QueryResult, type QueryResultRow } from "pg"; import type { BookingPublicationAttemptStatusPort } from "./bookingPublicationAttemptStatusRepository.js"; @@ -44,7 +47,7 @@ type ProjectionClaim = ClaimedRow & { workerId: string; }; -type ParsedPublication = DistributionBookingPublicationInput & { +type ParsedPublication = Omit & { organizationId: string; }; @@ -82,6 +85,7 @@ export function createBookingPublicationProjector(config: { projection: DistributionBookingPublicationProjectionPort; attempts: BookingPublicationAttemptStatusPort; readiness: BookingPublicationProjectionReadinessPort; + builder: BookingPublicationBuilderPort; max?: number; pool?: BookingPublicationProjectorPool; now?: () => Date; @@ -145,7 +149,39 @@ export function createBookingPublicationProjector(config: { projectedAt, ); } - if (!(await hasCurrentReadiness(config.readiness, publication))) { + const readiness = await currentReadiness(config.readiness, publication); + if (!readiness) { + return await terminalFailure( + pool, + config.projection, + config.attempts, + claim, + "source_content_changed", + projectedAt, + ); + } + const built = await config.builder.build({ + organizationId: publication.organizationId, + readiness, + generatedAt: projectedAt.toISOString(), + }); + if (built.outcome === "rejected") { + if (built.code === "owner_snapshot_unavailable") throw new Error(built.code); + return await terminalFailure( + pool, + config.projection, + config.attempts, + claim, + built.code === "source_manifest_mismatch" + ? "source_content_changed" + : "projection_failed", + projectedAt, + ); + } + if ( + built.build.sourceManifestHash !== publication.readiness.sourceManifestHash || + built.build.readinessHash !== publication.readiness.readinessHash + ) { return await terminalFailure( pool, config.projection, @@ -156,7 +192,10 @@ export function createBookingPublicationProjector(config: { ); } try { - const active = await config.projection.projectPublication(publication); + const active = await config.projection.projectPublication({ + ...publication, + publicContent: built.build.publicContent, + }); try { await config.attempts.markSucceeded({ operationId: publication.operationId, @@ -737,22 +776,30 @@ async function parsePublicationInput( }; } -async function hasCurrentReadiness( +async function currentReadiness( readiness: BookingPublicationProjectionReadinessPort, publication: ParsedPublication, -): Promise { +): Promise | null> { const current = await readiness.getBookingReadiness({ organizationId: publication.organizationId, propertyId: publication.propertyId, }); - return ( - current.outcome === "evaluated" && + if (current.outcome === "provider_failure") { + throw new Error(current.error.code); + } + if ( current.product === "booking" && current.status === "ready" && current.propertyId === publication.propertyId && current.sourceManifestHash === publication.readiness.sourceManifestHash && current.readinessHash === publication.readiness.readinessHash - ); + ) { + return createReadyProductReadinessEvidence(current, { + propertyId: publication.propertyId, + product: "booking", + }); + } + return null; } function positiveInteger(value: number | undefined, fallback: number): number { diff --git a/apps/api/src/domains/distributionBookingPublicationProjection.ts b/apps/api/src/domains/distributionBookingPublicationProjection.ts index c4c899496b..d92a56b357 100644 --- a/apps/api/src/domains/distributionBookingPublicationProjection.ts +++ b/apps/api/src/domains/distributionBookingPublicationProjection.ts @@ -1,4 +1,5 @@ import { randomUUID } from "node:crypto"; +import { isDeepStrictEqual } from "node:util"; import type { BookingActiveContentPointer, @@ -8,6 +9,7 @@ import type { OnboardingLifecycleJsonObject, ReadyProductReadinessEvidence, } from "@vayada/domain-hotels"; +import type { BookingPublicContent } from "@vayada/domain-distribution/booking-publication"; import pg, { type QueryResult, type QueryResultRow } from "pg"; export type DistributionBookingPublicationTransaction = { @@ -42,6 +44,7 @@ export type DistributionBookingPublicationInput = { sourceManifestHash: string; readinessHash: string; }; + publicContent: BookingPublicContent; projectedAt: Date; }; @@ -68,13 +71,6 @@ export class BookingPublicationActiveRevisionConflictError extends Error { } } -export class BookingPublicationPublicContentUnavailableError extends Error { - constructor() { - super("A fresh public Booking projection is not available"); - this.name = "BookingPublicationPublicContentUnavailableError"; - } -} - export class BookingPublicationLeaseLostError extends Error { constructor() { super("The Booking publication outbox lease is no longer current"); @@ -108,8 +104,6 @@ type RevisionRow = { builtAt: Date | string; }; -type PublicContentRow = { publicContent: OnboardingLifecycleJsonObject }; - /** * Distribution-owned implementation of the Booking lifecycle boundary. * Projector retries use the operation ID as the immutable revision ID, so a @@ -233,13 +227,11 @@ export function createPgDistributionBookingPublicationProjection(config: { if ((current?.revisionId ?? null) !== input.expectedActiveRevisionId) { throw new BookingPublicationActiveRevisionConflictError(current?.revisionId ?? null); } - const publicContent = await loadPublicContent(client, input.propertyId); - if (!publicContent) throw new BookingPublicationPublicContentUnavailableError(); const revision = await insertRevision(client, { revisionId: input.operationId, propertyId: input.propertyId, readiness: input.readiness as ReadyProductReadinessEvidence<"booking">, - publicContent, + publicContent: input.publicContent, builtByUserId: input.requestedByUserId, builtAt: input.projectedAt.toISOString(), }); @@ -356,72 +348,13 @@ async function selectRevision( return result.rows[0] ?? null; } -async function loadPublicContent( - client: ProjectionClient, - propertyId: string, -): Promise { - const result = await client.query( - `SELECT jsonb_build_object( - 'contractVersion', 'booking-public-content.v1', - 'profile', jsonb_build_object( - 'publicId', profile.public_id, - 'canonicalSlug', profile.canonical_slug, - 'canonicalUrl', profile.canonical_url, - 'bookingBaseUrl', profile.booking_base_url, - 'customDomainUrl', profile.custom_domain_url, - 'timezone', profile.timezone, - 'defaultLocale', profile.default_locale, - 'supportedLocales', to_jsonb(profile.supported_locales), - 'defaultCurrency', profile.default_currency, - 'supportedCurrencies', to_jsonb(profile.supported_currencies), - 'publicIdentity', profile.public_identity, - 'location', profile.location, - 'media', profile.media, - 'amenities', profile.amenities, - 'policies', profile.policies, - 'capabilities', profile.capabilities, - 'supportedQuoteParameters', profile.supported_quote_parameters - ), - 'roomOffers', COALESCE(( - SELECT jsonb_agg( - jsonb_build_object( - 'publicOfferKey', offer.public_offer_key, - 'stayDate', offer.stay_date, - 'availabilityStatus', offer.availability_status, - 'sellablePublicly', offer.sellable_publicly, - 'availableRooms', offer.available_rooms, - 'basePriceAmount', offer.base_price_amount, - 'taxesAndFeesAmount', offer.taxes_and_fees_amount, - 'discountsAmount', offer.discounts_amount, - 'currency', offer.currency, - 'occupancy', offer.occupancy, - 'roomSummary', offer.room_summary, - 'rateSummary', offer.rate_summary, - 'paymentOptions', to_jsonb(offer.payment_options), - 'publicPolicy', offer.public_policy, - 'unavailableReasons', to_jsonb(offer.unavailable_reasons) - ) ORDER BY offer.stay_date, offer.public_offer_key - ) - FROM distribution.public_room_offer_snapshots offer - WHERE offer.property_id = profile.property_id - ), '[]'::jsonb) - ) AS "publicContent" - FROM distribution.public_hotel_bookability_profiles profile - WHERE profile.property_id = $1::uuid - AND profile.profile_status = 'public' - AND profile.freshness_status = 'fresh'`, - [propertyId], - ); - return result.rows[0]?.publicContent ?? null; -} - async function insertRevision( client: ProjectionClient, input: { revisionId: string; propertyId: string; readiness: ReadyProductReadinessEvidence<"booking">; - publicContent: OnboardingLifecycleJsonObject; + publicContent: BookingPublicContent | OnboardingLifecycleJsonObject; builtByUserId: string; builtAt: string; }, @@ -497,12 +430,17 @@ function assertMatchingRetry( existing.propertyId !== input.propertyId || existing.sourceManifestHash !== input.readiness.sourceManifestHash || existing.readinessHash !== input.readiness.readinessHash || - existing.builtByUserId !== input.requestedByUserId + existing.builtByUserId !== input.requestedByUserId || + !isDeepStrictEqual(existing.publicContent, normalizedJson(input.publicContent)) ) { throw new Error("Stored Booking publication revision does not match its operation"); } } +function normalizedJson(value: BookingPublicContent): unknown { + return JSON.parse(JSON.stringify(value)) as unknown; +} + function revisionProjection(row: RevisionRow): BookingContentRevision { return Object.freeze({ revisionId: row.revisionId as BookingContentRevisionId,