Skip to content
Merged
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 6 additions & 1 deletion app/api/admin/ama/bookings/[bookingId]/route.ts
Original file line number Diff line number Diff line change
Expand Up @@ -3,17 +3,22 @@ import {
getAmaAdminServices,
ownerRequestAuthenticator,
} from '~/lib/ama/admin/server'
import { kickAmaOperations } from '~/lib/ama/booking/server'

export const maxDuration = 60

export async function POST(
request: Request,
{ params }: { params: Promise<{ bookingId: string }> },
) {
const { bookingId } = await params
const { bookingAdmin, security, baseUrl } = getAmaAdminServices()
return createAdminBookingActionHandler({
const response = await createAdminBookingActionHandler({
authenticator: ownerRequestAuthenticator,
service: bookingAdmin,
security,
baseUrl,
})(request, bookingId)
if (response.ok) kickAmaOperations()
return response
}
7 changes: 6 additions & 1 deletion app/api/admin/ama/operations/[operationId]/route.ts
Original file line number Diff line number Diff line change
Expand Up @@ -3,17 +3,22 @@ import {
getAmaAdminServices,
ownerRequestAuthenticator,
} from '~/lib/ama/admin/server'
import { kickAmaOperations } from '~/lib/ama/booking/server'

export const maxDuration = 60

export async function POST(
request: Request,
{ params }: { params: Promise<{ operationId: string }> },
) {
const { operationId } = await params
const { bookingAdmin, security, baseUrl } = getAmaAdminServices()
return createAdminOperationActionHandler({
const response = await createAdminOperationActionHandler({
authenticator: ownerRequestAuthenticator,
service: bookingAdmin,
security,
baseUrl,
})(request, operationId)
if (response.ok) kickAmaOperations()
return response
}
14 changes: 12 additions & 2 deletions app/api/ama/manage/[token]/cancel/route.ts
Original file line number Diff line number Diff line change
@@ -1,7 +1,12 @@
import { createManageCancelHandler } from '~/lib/ama/booking/http'
import { getAmaBookingServices } from '~/lib/ama/booking/server'
import {
getAmaBookingServices,
kickAmaOperations,
} from '~/lib/ama/booking/server'
import { protectAmaLaunchBoundary } from '~/lib/ama/security/launch-boundary-server'

export const maxDuration = 60

export async function POST(
request: Request,
{ params }: { params: Promise<{ token: string }> },
Expand All @@ -13,5 +18,10 @@ export async function POST(
if (blocked) return blocked
const { token } = await params
const { manage, guard } = getAmaBookingServices()
return createManageCancelHandler({ manage, guard })(request, token)
const response = await createManageCancelHandler({ manage, guard })(
request,
token,
)
if (response.ok) kickAmaOperations()
return response
}
14 changes: 12 additions & 2 deletions app/api/ama/manage/[token]/reschedule/route.ts
Original file line number Diff line number Diff line change
@@ -1,7 +1,12 @@
import { createManageRescheduleHandler } from '~/lib/ama/booking/http'
import { getAmaBookingServices } from '~/lib/ama/booking/server'
import {
getAmaBookingServices,
kickAmaOperations,
} from '~/lib/ama/booking/server'
import { protectAmaLaunchBoundary } from '~/lib/ama/security/launch-boundary-server'

export const maxDuration = 60

export async function POST(
request: Request,
{ params }: { params: Promise<{ token: string }> },
Expand All @@ -13,5 +18,10 @@ export async function POST(
if (blocked) return blocked
const { token } = await params
const { manage, guard } = getAmaBookingServices()
return createManageRescheduleHandler({ manage, guard })(request, token)
const response = await createManageRescheduleHandler({ manage, guard })(
request,
token,
)
if (response.ok) kickAmaOperations()
return response
}
12 changes: 11 additions & 1 deletion app/api/ama/stripe/webhook/route.ts
Original file line number Diff line number Diff line change
@@ -1,7 +1,12 @@
import { createStripeWebhookHandler, json } from '~/lib/ama/booking/http'
import { getAmaBookingServices } from '~/lib/ama/booking/server'
import {
getAmaBookingServices,
kickAmaOperations,
} from '~/lib/ama/booking/server'
import { protectAmaLaunchBoundary } from '~/lib/ama/security/launch-boundary-server'

export const maxDuration = 60

export async function POST(request: Request) {
const blocked = protectAmaLaunchBoundary(request, ['payments'])
if (blocked) return blocked
Expand All @@ -10,5 +15,10 @@ export async function POST(request: Request) {
return createStripeWebhookHandler({
service: booking,
signingSecret: stripeWebhookSecret,
// Only booking creation enqueues durable work; duplicate, ignored,
// orphaned, and hold-release deliveries return 200 without any.
onOutcome: (outcome) => {
if (outcome === 'booking_created') kickAmaOperations()
},
})(request)
}
9 changes: 8 additions & 1 deletion lib/ama/booking/http.ts
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@ import {
import type { SecurityRateLimiter } from '../security/service'
import { verifyStripeWebhook } from '../stripe/webhook'
import type { ManageService } from './manage'
import type { BookingService } from './service'
import type { BookingService, WebhookOutcome } from './service'

const MAX_JSON_BODY_BYTES = 32 * 1024

Expand Down Expand Up @@ -253,12 +253,18 @@ type WebhookDependencies = {
service: Pick<BookingService, 'processWebhookEvent'>
signingSecret: string
clock?: { now(): Date }
/**
* Observes the processed outcome so callers can react to the ones that
* enqueued durable work without re-parsing the response body.
*/
onOutcome?: (outcome: WebhookOutcome) => void
}

export function createStripeWebhookHandler({
service,
signingSecret,
clock = { now: () => new Date() },
onOutcome,
}: WebhookDependencies) {
return async function POST(request: Request) {
let payload: string
Expand All @@ -276,6 +282,7 @@ export function createStripeWebhookHandler({
if (!event) return json(400, { error: 'invalid_signature' })
try {
const outcome = await service.processWebhookEvent(event)
onOutcome?.(outcome)
return json(200, { received: true, outcome })
} catch {
// Signal Stripe to redeliver; the persisted provider event makes the
Expand Down
36 changes: 36 additions & 0 deletions lib/ama/booking/server.ts
Original file line number Diff line number Diff line change
@@ -1,5 +1,7 @@
import 'server-only'

import { waitUntil } from '@vercel/functions'

import { createRateLimiter } from '~/lib/rate-limit/server'

import { availabilityRepository } from '../availability/repository'
Expand Down Expand Up @@ -30,6 +32,10 @@ import { bookingRepository } from './repository'
import { createBookingService } from './service'

const PROVIDER_REQUEST_TIMEOUT_MS = 8_000
const OPERATIONS_BATCH_SIZE = 10
// Leaves headroom under the mutating routes' maxDuration for the drain pass
// already in flight when the budget runs out.
const INLINE_DRAIN_BUDGET_MS = 30_000

let services: ReturnType<typeof createServices> | undefined

Expand Down Expand Up @@ -186,6 +192,7 @@ function createServices() {
clock,
})
const runner = createOperationsRunner({
batchSize: OPERATIONS_BATCH_SIZE,
operations: durableOperationsRepository,
handler: createOperationHandlers({
repository: bookingRepository,
Expand Down Expand Up @@ -227,3 +234,32 @@ export function getAmaBookingServices() {
services ??= createServices()
return services
}

/**
* Starts a background drain of due durable operations inside the current
* invocation, so work enqueued by a mutation (booking emails, Finalizing
* Booking recovery, refunds) begins immediately instead of waiting for the
* next scheduled sweep. The scheduled endpoint remains the sole driver for
* reminders, retry backoff, and expired Slot Hold release.
*/
export function kickAmaOperations() {
const { runner } = getAmaBookingServices()
waitUntil(
(async () => {
// A full batch means older due work may have crowded out the operation
// this mutation enqueued, so keep draining until a partial batch shows
// the due queue is empty or the inline budget is spent. Anything left
// over stays with the scheduled sweep.
const startedAtMs = Date.now()
let result = await runner.run()
while (
result.claimed >= OPERATIONS_BATCH_SIZE &&
Date.now() - startedAtMs < INLINE_DRAIN_BUDGET_MS
) {
result = await runner.run()
Comment thread
CaliCastle marked this conversation as resolved.
}
})().catch((error) => {
console.error('ama: inline operations drain failed', error)
}),
)
}
Comment thread
greptile-apps[bot] marked this conversation as resolved.
4 changes: 2 additions & 2 deletions vercel.json
Original file line number Diff line number Diff line change
Expand Up @@ -6,11 +6,11 @@
"crons": [
{
"path": "/api/internal/media/reconcile",
"schedule": "*/15 * * * *"
"schedule": "0 * * * *"
},
{
"path": "/api/internal/ama/work",
"schedule": "*/5 * * * *"
"schedule": "*/30 * * * *"
}
]
}