API Development Architect
Designs a production-ready RESTful API with OpenAPI spec, secure auth, validation, and scalable architecture.
// prompt
You are a **senior API architect** with deep experience shipping secure, scalable RESTful services. Design a production-ready API for the project below, balancing clean contracts, security, and developer experience.
## Project Context
- **API Name:** {{api_name}}
- **Business Domain:** {{business_domain}}
- **Core Resources:** {{resource_list}}
- **Stack:** {{language_and_framework}} over {{database}}
- **Authentication:** {{authentication_method}}
- **Expected Scale:** {{expected_traffic_scale}}
## What To Deliver
### 1. Resource & Endpoint Design
- Model each resource and its relationships.
- Define RESTful, versioned routes (`/api/v1/...`) with the correct HTTP methods, status codes, and JSON request/response shapes.
- Specify pagination, filtering, and sorting conventions for collections.
### 2. OpenAPI 3.0 Specification
Produce a valid OpenAPI 3.0 spec covering the main endpoints, including parameters, request bodies, response schemas, and reusable components.
### 3. Implementation Skeleton
Provide idiomatic {{language_and_framework}} code for the key endpoints, showing routing, a typed data model/schema, dependency injection, and structured error handling.
### 4. Security & Reliability
- Apply {{authentication_method}} with authorization checks per endpoint.
- Add input validation, consistent error envelopes, rate limiting, and idempotency where relevant.
- Note CORS, secrets handling, and audit-logging considerations.
### 5. Scalability Notes
Recommend caching, indexing, and connection-pooling strategies suited to {{expected_traffic_scale}}.
## Output Format
1. A short architecture summary.
2. An endpoint table (Method · Path · Purpose · Auth).
3. The OpenAPI spec in a YAML code block.
4. Implementation code in fenced blocks.
5. A checklist of security and performance safeguards.
Ask clarifying questions only if a critical requirement (data model, auth, or scale) is ambiguous; otherwise state your assumptions and proceed.
Fill in the variables
Example run
Generated by Claude Opus 5.5 on September 23, 2026
Inputs used
- Api name
- TrailBook API
- Business domain
- Outdoor adventure & guided hiking tour booking platform — connects independent mountain guides with customers who want to reserve spots on scheduled group hikes and multi-day treks (e.g., trips in the Rockies, Alps, and Dolomites).
- Resource list
- - User: id, email, hashed_password (if local), full_name, phone, role (customer | guide | admin), created_at, is_verified. A User with role=guide has one associated GuideProfile. - GuideProfile: id, user_id (1:1 User), bio, certifications[] (e.g., Wilderness First Responder, IFMGA), years_experience, languages_spoken[], rating_avg, is_approved (bool, set by admin). - Trail: id, name, region (e.g., "Dolomites", "Rocky Mountain NP"), distance_km, elevation_gain_m, difficulty (easy | moderate | strenuous | technical), description, gps_track_url, created_by_guide_id. - TourInstance: id, trail_id (FK Trail), guide_id (FK GuideProfile), start_datetime, end_datetime, max_participants, price_cents, currency, status (scheduled | cancelled | completed), meeting_point_lat/lng. Represents one bookable occurrence of a Trail led by a specific guide. - Booking: id, tour_instance_id (FK), customer_id (FK User), num_participants, status (pending | confirmed | cancelled | refunded), total_price_cents, created_at, idempotency_key. A TourInstance has many Bookings; sum of num_participants must not exceed max_participants. - Waiver: id, booking_id (FK, 1:1), signed_pdf_url, signed_at, ip_address — required before a Booking can move to "confirmed". - Payment: id, booking_id (FK), provider (stripe), provider_charge_id, amount_cents, currency, status (authorized | captured | refunded | failed), created_at. - Review: id, tour_instance_id (FK), customer_id (FK), rating (1-5), comment, created_at — only allowed after Booking.status = completed. Relationships: User 1—1 GuideProfile (when role=guide); GuideProfile 1—N Trail (as creator); Trail 1—N TourInstance; GuideProfile 1—N TourInstance (as leader); TourInstance 1—N Booking; Booking 1—1 Waiver; Booking 1—N Payment (supports partial refunds); TourInstance 1—N Review.
- Language and framework
- Python 3.12 with FastAPI 0.115 (async), Pydantic v2 for schemas, SQLAlchemy 2.0 async ORM, Alembic for migrations
- Database
- PostgreSQL 15 (Amazon RDS, Multi-AZ)
- Authentication method
- OAuth 2.0 with short-lived JWT bearer access tokens issued via Auth0 (RS256, 15-min expiry) plus refresh tokens; role-based access control enforced per endpoint for customer, guide, and admin roles
- Expected traffic / scale
- ~60,000 registered users; strong seasonality with peak in June–September (~2,500 concurrent sessions, ~180 requests/sec at peak) and low off-season traffic; target 99.9% uptime SLA
TrailBook API: Production Design
0. Assumptions & Design Decisions
The brief is clear enough to proceed. A few gaps in it would cause real bugs, so I resolved each one explicitly. Please confirm or override them.
| # | Gap / ambiguity | Decision |
|---|---|---|
| A1 | Review requires Booking.status = completed, but the Booking enum has no completed. |
Added completed to BookingStatus. A job sets it when the TourInstance is marked completed. |
| A2 | hashed_password (if local) conflicts with Auth0 as the identity provider. |
No passwords are stored in our DB. Auth0 handles database, social and MFA logins. users.auth0_sub links identities. |
| A3 | Refresh tokens | Auth0 issues and rotates them (rotation plus reuse detection). The SPA uses Authorization Code + PKCE. The API only validates access tokens and never sees refresh tokens. |
| A4 | Source of truth for roles | The DB is authoritative. The Auth0 role claim is informational. This lets admins suspend a user or revoke a guide immediately, without waiting up to 15 minutes for token expiry. |
| A5 | Guide onboarding | A customer calls POST /guide-profiles (starts with is_approved=false). Admin approval sets is_approved=true and role=guide. |
| A6 | Who can schedule tours on a trail? | Any approved guide can schedule a TourInstance on any published trail (TourInstance has its own guide_id). Only the trail's creator or an admin can edit the trail. |
| A7 | Payment flow | Stripe PaymentIntent with manual capture. Capture happens only after the webhook confirms the authorization and the waiver exists and the booking is still pending. Capture is the final gate. |
| A8 | Seat holding | POST /bookings atomically reserves seats with a 15-minute hold. A sweeper releases expired holds. |
| A9 | Waiver | One waiver per booking, signed by the lead customer, listing all participant names. Get legal review: some jurisdictions require a separate waiver per adult participant. |
| A10 | Cancellation refunds | Default policy (configurable): ≥7 days before start = 100%, 48h–7d = 50%, <48h = 0%. Guide or admin cancellation of a tour = 100%. |
| A11 | Who books? | Only customer role accounts can book. A guide cannot book their own tour. |
1. Architecture Summary
Web SPA / Mobile ──(Auth Code+PKCE)──► Auth0 ──► RS256 JWT (15 min) + rotating refresh token
│
▼ HTTPS
CloudFront + AWS WAF (IP rate rules, bot control, geo) ── cached public GETs (15–60 s)
│
▼
ALB ──► ECS Fargate: FastAPI/uvicorn (2 workers/task, 2 AZs, autoscaled 2→8 tasks)
│ │ │ │
│ │ │ └─► Stripe (PaymentIntents, manual capture)
│ │ └─► S3 (GPS tracks, waiver PDFs; SSE-KMS, presigned URLs)
│ └─► ElastiCache Redis (rate limits, response cache)
└─► RDS PostgreSQL 15 Multi-AZ (+1 read replica for search in peak season)
▲
Worker (ECS) ──────────┘ outbox consumer: refunds, waiver PDF render, emails, hold sweeper
Stripe ──webhook──► POST /api/v1/webhooks/stripe (signature-verified, deduplicated)
Core principles
- Overbooking is prevented inside the database. Seats are reserved with a single conditional
UPDATE … WHERE seats_reserved + n <= max_participants. ACHECKconstraint backs it up. There is no read-then-write race and no distributed lock. - Invariants live in the DB as well as the app:
- A trigger blocks
confirmedwithout a waiver. - An exclusion constraint blocks overlapping tours for the same guide.
- Unique constraints enforce one review per customer per tour and booking idempotency.
- A trigger blocks
- Money moves through a state machine, not ad-hoc updates. Side effects run via a transactional outbox. Webhooks are idempotent. Every Stripe call carries an idempotency key.
- Errors use RFC 9457
application/problem+json, each with a stable machine-readablecode.
Booking lifecycle
POST /bookings (seats held, 15 min)
│
pending ──(hold expired / customer cancel)──► cancelled ──(refund done)──► refunded
│ PUT waiver → POST payments → Stripe authorizes → webhook: capture
▼
confirmed ──(customer/admin cancel)──► cancelled
│
tour marked completed
▼
completed ──► review allowed (confirmed|completed ──admin full refund──► refunded)
2. Endpoint Table
All paths are prefixed with /api/v1. "Owner" means the resource's customer or the guide who leads the tour.
| Method | Path | Purpose | Auth |
|---|---|---|---|
| GET | /users/me |
Current user profile (JIT-provisioned on first call) | Any authenticated |
| PATCH | /users/me |
Update full_name, phone |
Any authenticated |
| DELETE | /users/me |
GDPR erasure request (anonymize, keep financial records) | Any authenticated |
| GET | /admin/users |
Search/list users | Admin |
| PATCH | /admin/users/{userId} |
Suspend or unsuspend, change role | Admin |
| POST | /guide-profiles |
Apply to become a guide | Customer |
| GET | /guide-profiles |
List approved guides (filter by language, certification) | Public |
| GET | /guide-profiles/{guideId} |
Guide profile | Public (approved) / Owner / Admin |
| PATCH | /guide-profiles/{guideId} |
Update bio, languages, certifications | Owner guide |
| POST | /guide-profiles/{guideId}/approve |
Approve; promotes user to guide |
Admin |
| POST | /guide-profiles/{guideId}/revoke |
Revoke approval | Admin |
| GET | /trails |
Search trails (region, difficulty, distance) | Public |
| POST | /trails |
Create trail | Approved guide |
| GET | /trails/{trailId} |
Trail detail | Public |
| PATCH | /trails/{trailId} |
Update trail | Creator guide / Admin |
| DELETE | /trails/{trailId} |
Soft delete (409 if future tours exist) | Creator guide / Admin |
| POST | /trails/{trailId}/gps-track |
Get presigned S3 upload URL for GPX | Creator guide |
| GET | /tours |
Search bookable tours (region, dates, price, spots, language) | Public |
| POST | /tours |
Schedule a TourInstance | Approved guide |
| GET | /tours/{tourId} |
Tour detail with live spots_remaining |
Public |
| PATCH | /tours/{tourId} |
Update (cannot drop capacity below booked seats) | Leading guide / Admin |
| POST | /tours/{tourId}/cancel |
Cancel tour, refund all bookings asynchronously (202) | Leading guide / Admin |
| POST | /tours/{tourId}/complete |
Mark completed; bookings become completed |
Leading guide / Admin |
| GET | /tours/{tourId}/bookings |
Participant manifest (minimal PII) | Leading guide / Admin |
| GET | /tours/{tourId}/reviews |
List reviews | Public |
| POST | /tours/{tourId}/reviews |
Create review | Customer with a completed booking |
| DELETE | /reviews/{reviewId} |
Moderate/remove review | Admin |
| GET | /bookings |
My bookings (customer) / all bookings (admin) | Customer / Admin |
| POST | /bookings |
Hold seats; Idempotency-Key required |
Verified customer |
| GET | /bookings/{bookingId} |
Booking detail | Owner / Leading guide / Admin |
| POST | /bookings/{bookingId}/cancel |
Cancel; refund per policy | Owner customer / Admin |
| PUT | /bookings/{bookingId}/waiver |
Sign waiver (idempotent, immutable once signed) | Owner customer |
| GET | /bookings/{bookingId}/waiver |
Waiver metadata plus 5-minute presigned PDF URL | Owner / Leading guide / Admin |
| POST | /bookings/{bookingId}/payments |
Start payment; returns Stripe client_secret. Idempotency-Key |
Owner customer |
| GET | /bookings/{bookingId}/payments |
Payment and refund ledger | Owner customer / Admin |
| POST | /bookings/{bookingId}/refunds |
Partial or full refund. Idempotency-Key |
Admin |
| POST | /webhooks/stripe |
Stripe events | Stripe signature (no JWT) |
| GET | /admin/audit-logs |
Query audit trail | Admin |
| GET | /healthz, /readyz |
Liveness / readiness | Internal (ALB) |
Collection conventions
- Pagination is cursor/keyset based:
?limit=20&cursor=<opaque>.limitranges from 1 to 100 (default 20).- Responses look like
{ "data": [...], "page": { "limit", "has_more", "next_cursor" } }. - No total counts (expensive on large tables, and unstable anyway). Offset pagination is not offered.
- Filtering uses explicit, typed query parameters, for example:
region=Dolomites&difficulty=moderate&difficulty=strenuousstarts_after=2026-07-01T00:00:00Z&min_spots=2language=de&min_price_cents=5000&max_price_cents=20000¤cy=EUR- Price filters require
currency. - Unknown parameters are rejected with 422, so typos don't silently return unfiltered data.
- Sorting uses
?sort=start_datetimeor?sort=-price_cents, from a per-endpoint allowlist. A leading-means descending, andidis always the tie-breaker. - Timestamps are RFC 3339 with an offset. They are stored as
timestamptz. Tours also carry an IANAtimezonefor local display. - Money is integer minor units (
price_cents) plus an ISO-4217currency. The server always computes totals.
3. OpenAPI 3.0 Specification
openapi: 3.0.3
info:
title: TrailBook API
version: 1.0.0
description: |
Booking platform for guided hikes and multi-day treks.
**Auth:** Auth0-issued RS256 JWT access tokens (15 min) as `Authorization: Bearer <token>`.
Role requirements are listed per operation in `x-required-roles`.
**Errors:** All errors are RFC 9457 `application/problem+json` with a stable `code`.
**Idempotency:** Money- and inventory-affecting POSTs require an `Idempotency-Key` header
(8–64 chars, `[A-Za-z0-9_-]`). Replays return the original result with `Idempotent-Replayed: true`.
Reusing a key with a different body returns `422 IDEMPOTENCY_KEY_REUSED`.
**Rate limits:** `RateLimit-Limit`, `RateLimit-Remaining`, `RateLimit-Reset` headers; `429` with `Retry-After`.
contact: { name: TrailBook Platform Team, url: https://docs.trailbook.app }
servers:
- url: https://api.trailbook.app/api/v1
description: Production
- url: https://api.staging.trailbook.app/api/v1
description: Staging
tags:
- name: Users
- name: Guides
- name: Trails
- name: Tours
- name: Bookings
- name: Payments
- name: Reviews
- name: Webhooks
security:
- bearerAuth: []
paths:
/users/me:
get:
tags: [Users]
operationId: getCurrentUser
summary: Get the authenticated user
x-required-roles: [customer, guide, admin]
responses:
'200':
description: Current user
content:
application/json:
schema: { $ref: '#/components/schemas/User' }
'401': { $ref: '#/components/responses/Unauthorized' }
'403': { $ref: '#/components/responses/Forbidden' }
patch:
tags: [Users]
operationId: updateCurrentUser
summary: Update own profile
x-required-roles: [customer, guide, admin]
requestBody:
required: true
content:
application/json:
schema: { $ref: '#/components/schemas/UserUpdate' }
responses:
'200':
description: Updated
content:
application/json:
schema: { $ref: '#/components/schemas/User' }
'401': { $ref: '#/components/responses/Unauthorized' }
'422': { $ref: '#/components/responses/ValidationError' }
/guide-profiles:
post:
tags: [Guides]
operationId: applyAsGuide
summary: Apply to become a guide (pending admin approval)
x-required-roles: [customer]
requestBody:
required: true
content:
application/json:
schema: { $ref: '#/components/schemas/GuideProfileCreate' }
responses:
'201':
description: Application created
headers:
Location: { $ref: '#/components/headers/Location' }
content:
application/json:
schema: { $ref: '#/components/schemas/GuideProfile' }
'401': { $ref: '#/components/responses/Unauthorized' }
'403': { $ref: '#/components/responses/Forbidden' }
'409': { $ref: '#/components/responses/Conflict' }
'422': { $ref: '#/components/responses/ValidationError' }
/guide-profiles/{guideId}/approve:
post:
tags: [Guides]
operationId: approveGuide
summary: Approve a guide application (promotes user role to guide)
x-required-roles: [admin]
parameters:
- $ref: '#/components/parameters/GuideId'
responses:
'200':
description: Approved
content:
application/json:
schema: { $ref: '#/components/schemas/GuideProfile' }
'401': { $ref: '#/components/responses/Unauthorized' }
'403': { $ref: '#/components/responses/Forbidden' }
'404': { $ref: '#/components/responses/NotFound' }
/trails:
get:
tags: [Trails]
operationId: listTrails
summary: Search trails
security: []
parameters:
- { name: region, in: query, schema: { type: string, maxLength: 80 } }
- name: difficulty
in: query
style: form
explode: true
schema:
type: array
items: { $ref: '#/components/schemas/Difficulty' }
- { name: max_distance_km, in: query, schema: { type: number, minimum: 0 } }
- name: sort
in: query
schema:
type: string
enum: [name, distance_km, -distance_km, elevation_gain_m, -elevation_gain_m]
default: name
- $ref: '#/components/parameters/Limit'
- $ref: '#/components/parameters/Cursor'
responses:
'200':
description: Page of trails
headers:
RateLimit-Limit: { $ref: '#/components/headers/RateLimitLimit' }
RateLimit-Remaining: { $ref: '#/components/headers/RateLimitRemaining' }
content:
application/json:
schema: { $ref: '#/components/schemas/TrailPage' }
'400': { $ref: '#/components/responses/BadRequest' }
'422': { $ref: '#/components/responses/ValidationError' }
'429': { $ref: '#/components/responses/TooManyRequests' }
post:
tags: [Trails]
operationId: createTrail
summary: Create a trail
x-required-roles: [guide]
description: Caller must be an approved guide.
requestBody:
required: true
content:
application/json:
schema: { $ref: '#/components/schemas/TrailCreate' }
responses:
'201':
description: Created
headers:
Location: { $ref: '#/components/headers/Location' }
content:
application/json:
schema: { $ref: '#/components/schemas/Trail' }
'401': { $ref: '#/components/responses/Unauthorized' }
'403': { $ref: '#/components/responses/Forbidden' }
'422': { $ref: '#/components/responses/ValidationError' }
/trails/{trailId}:
parameters:
- $ref: '#/components/parameters/TrailId'
get:
tags: [Trails]
operationId: getTrail
security: []
responses:
'200':
description: Trail
content:
application/json:
schema: { $ref: '#/components/schemas/Trail' }
'404': { $ref: '#/components/responses/NotFound' }
patch:
tags: [Trails]
operationId: updateTrail
x-required-roles: [guide, admin]
description: Only the creating guide or an admin.
requestBody:
required: true
content:
application/json:
schema: { $ref: '#/components/schemas/TrailUpdate' }
responses:
'200':
description: Updated
content:
application/json:
schema: { $ref: '#/components/schemas/Trail' }
'401': { $ref: '#/components/responses/Unauthorized' }
'403': { $ref: '#/components/responses/Forbidden' }
'404': { $ref: '#/components/responses/NotFound' }
'422': { $ref: '#/components/responses/ValidationError' }
/tours:
get:
tags: [Tours]
operationId: searchTours
summary: Search bookable tour instances
description: Returns only `scheduled` tours starting in the future, led by approved guides.
security: []
parameters:
- { name: region, in: query, schema: { type: string, maxLength: 80 } }
- name: difficulty
in: query
style: form
explode: true
schema:
type: array
items: { $ref: '#/components/schemas/Difficulty' }
- { name: trail_id, in: query, schema: { type: string, format: uuid } }
- { name: guide_id, in: query, schema: { type: string, format: uuid } }
- name: language
in: query
description: BCP-47 language code spoken by the guide
schema: { type: string, pattern: '^[a-z]{2}(-[A-Z]{2})?$', example: de }
- { name: starts_after, in: query, schema: { type: string, format: date-time } }
- { name: starts_before, in: query, schema: { type: string, format: date-time } }
- { name: min_price_cents, in: query, schema: { type: integer, minimum: 0 } }
- { name: max_price_cents, in: query, schema: { type: integer, minimum: 0 } }
- name: currency
in: query
description: Required when a price filter is used.
schema: { $ref: '#/components/schemas/Currency' }
- { name: min_spots, in: query, schema: { type: integer, minimum: 1, maximum: 50, default: 1 } }
- name: sort
in: query
schema:
type: string
enum: [start_datetime, -start_datetime, price_cents, -price_cents]
default: start_datetime
- $ref: '#/components/parameters/Limit'
- $ref: '#/components/parameters/Cursor'
responses:
'200':
description: Page of tours
headers:
Cache-Control:
schema: { type: string, example: 'public, max-age=15, stale-while-revalidate=30' }
content:
application/json:
schema: { $ref: '#/components/schemas/TourPage' }
'400': { $ref: '#/components/responses/BadRequest' }
'422': { $ref: '#/components/responses/ValidationError' }
'429': { $ref: '#/components/responses/TooManyRequests' }
post:
tags: [Tours]
operationId: createTour
summary: Schedule a tour instance
x-required-roles: [guide]
description: Caller must be an approved guide. The guide cannot have overlapping scheduled tours.
requestBody:
required: true
content:
application/json:
schema: { $ref: '#/components/schemas/TourCreate' }
responses:
'201':
description: Created
headers:
Location: { $ref: '#/components/headers/Location' }
content:
application/json:
schema: { $ref: '#/components/schemas/Tour' }
'401': { $ref: '#/components/responses/Unauthorized' }
'403': { $ref: '#/components/responses/Forbidden' }
'409': { $ref: '#/components/responses/Conflict' }
'422': { $ref: '#/components/responses/ValidationError' }
/tours/{tourId}:
parameters:
- $ref: '#/components/parameters/TourId'
get:
tags: [Tours]
operationId: getTour
security: []
responses:
'200':
description: Tour
content:
application/json:
schema: { $ref: '#/components/schemas/Tour' }
'404': { $ref: '#/components/responses/NotFound' }
patch:
tags: [Tours]
operationId: updateTour
x-required-roles: [guide, admin]
requestBody:
required: true
content:
application/json:
schema: { $ref: '#/components/schemas/TourUpdate' }
responses:
'200':
description: Updated
content:
application/json:
schema: { $ref: '#/components/schemas/Tour' }
'401': { $ref: '#/components/responses/Unauthorized' }
'403': { $ref: '#/components/responses/Forbidden' }
'404': { $ref: '#/components/responses/NotFound' }
'409': { $ref: '#/components/responses/Conflict' }
'422': { $ref: '#/components/responses/ValidationError' }
/tours/{tourId}/cancel:
post:
tags: [Tours]
operationId: cancelTour
summary: Cancel a tour; all bookings are cancelled and fully refunded asynchronously
x-required-roles: [guide, admin]
parameters:
- $ref: '#/components/parameters/TourId'
requestBody:
required: true
content:
application/json:
schema:
type: object
additionalProperties: false
required: [reason]
properties:
reason: { type: string, minLength: 3, maxLength: 500 }
responses:
'202':
description: Cancellation accepted; refunds are processing
content:
application/json:
schema: { $ref: '#/components/schemas/Tour' }
'401': { $ref: '#/components/responses/Unauthorized' }
'403': { $ref: '#/components/responses/Forbidden' }
'404': { $ref: '#/components/responses/NotFound' }
'409': { $ref: '#/components/responses/Conflict' }
/tours/{tourId}/bookings:
get:
tags: [Tours]
operationId: getTourManifest
summary: Participant manifest for the leading guide
x-required-roles: [guide, admin]
parameters:
- $ref: '#/components/parameters/TourId'
- $ref: '#/components/parameters/Limit'
- $ref: '#/components/parameters/Cursor'
responses:
'200':
description: Manifest page
content:
application/json:
schema: { $ref: '#/components/schemas/ManifestPage' }
'401': { $ref: '#/components/responses/Unauthorized' }
'404': { $ref: '#/components/responses/NotFound' }
/tours/{tourId}/reviews:
parameters:
- $ref: '#/components/parameters/TourId'
get:
tags: [Reviews]
operationId: listTourReviews
security: []
parameters:
- name: sort
in: query
schema: { type: string, enum: [-created_at, -rating, rating], default: -created_at }
- $ref: '#/components/parameters/Limit'
- $ref: '#/components/parameters/Cursor'
responses:
'200':
description: Page of reviews
content:
application/json:
schema: { $ref: '#/components/schemas/ReviewPage' }
'404': { $ref: '#/components/responses/NotFound' }
post:
tags: [Reviews]
operationId: createReview
summary: Review a tour (requires a completed booking; one review per customer per tour)
x-required-roles: [customer]
requestBody:
required: true
content:
application/json:
schema: { $ref: '#/components/schemas/ReviewCreate' }
responses:
'201':
description: Created
content:
application/json:
schema: { $ref: '#/components/schemas/Review' }
'401': { $ref: '#/components/responses/Unauthorized' }
'403': { $ref: '#/components/responses/Forbidden' }
'409': { $ref: '#/components/responses/Conflict' }
'422': { $ref: '#/components/responses/ValidationError' }
/bookings:
get:
tags: [Bookings]
operationId: listBookings
summary: Customers see their own bookings; admins see all
x-required-roles: [customer, admin]
parameters:
- name: status
in: query
style: form
explode: true
schema:
type: array
items: { $ref: '#/components/schemas/BookingStatus' }
- $ref: '#/components/parameters/Limit'
- $ref: '#/components/parameters/Cursor'
responses:
'200':
description: Page of bookings
content:
application/json:
schema: { $ref: '#/components/schemas/BookingPage' }
'401': { $ref: '#/components/responses/Unauthorized' }
post:
tags: [Bookings]
operationId: createBooking
summary: Hold seats on a tour (15-minute hold, status=pending)
x-required-roles: [customer]
parameters:
- $ref: '#/components/parameters/IdempotencyKey'
requestBody:
required: true
content:
application/json:
schema: { $ref: '#/components/schemas/BookingCreate' }
responses:
'201':
description: Booking created (or replayed)
headers:
Location: { $ref: '#/components/headers/Location' }
Idempotent-Replayed: { $ref: '#/components/headers/IdempotentReplayed' }
RateLimit-Limit: { $ref: '#/components/headers/RateLimitLimit' }
RateLimit-Remaining: { $ref: '#/components/headers/RateLimitRemaining' }
content:
application/json:
schema: { $ref: '#/components/schemas/Booking' }
'401': { $ref: '#/components/responses/Unauthorized' }
'403': { $ref: '#/components/responses/Forbidden' }
'404': { $ref: '#/components/responses/NotFound' }
'409': { $ref: '#/components/responses/Conflict' }
'422': { $ref: '#/components/responses/ValidationError' }
'429': { $ref: '#/components/responses/TooManyRequests' }
/bookings/{bookingId}:
get:
tags: [Bookings]
operationId: getBooking
x-required-roles: [customer, guide, admin]
parameters:
- $ref: '#/components/parameters/BookingId'
responses:
'200':
description: Booking
content:
application/json:
schema: { $ref: '#/components/schemas/Booking' }
'401': { $ref: '#/components/responses/Unauthorized' }
'404': { $ref: '#/components/responses/NotFound' }
/bookings/{bookingId}/cancel:
post:
tags: [Bookings]
operationId: cancelBooking
x-required-roles: [customer, admin]
parameters:
- $ref: '#/components/parameters/BookingId'
requestBody:
required: false
content:
application/json:
schema:
type: object
additionalProperties: false
properties:
reason: { type: string, maxLength: 500 }
responses:
'200':
description: Cancelled; refund (if any) is processing
content:
application/json:
schema: { $ref: '#/components/schemas/BookingCancellation' }
'401': { $ref: '#/components/responses/Unauthorized' }
'404': { $ref: '#/components/responses/NotFound' }
'409': { $ref: '#/components/responses/Conflict' }
/bookings/{bookingId}/waiver:
parameters:
- $ref: '#/components/parameters/BookingId'
put:
tags: [Bookings]
operationId: signWaiver
summary: Sign the liability waiver (idempotent; immutable once signed)
x-required-roles: [customer]
requestBody:
required: true
content:
application/json:
schema: { $ref: '#/components/schemas/WaiverSign' }
responses:
'200':
description: Already signed; existing waiver returned
content:
application/json:
schema: { $ref: '#/components/schemas/Waiver' }
'201':
description: Signed
content:
application/json:
schema: { $ref: '#/components/schemas/Waiver' }
'401': { $ref: '#/components/responses/Unauthorized' }
'404': { $ref: '#/components/responses/NotFound' }
'409': { $ref: '#/components/responses/Conflict' }
'422': { $ref: '#/components/responses/ValidationError' }
get:
tags: [Bookings]
operationId: getWaiver
x-required-roles: [customer, guide, admin]
responses:
'200':
description: Waiver with short-lived download URL
content:
application/json:
schema: { $ref: '#/components/schemas/Waiver' }
'401': { $ref: '#/components/responses/Unauthorized' }
'404': { $ref: '#/components/responses/NotFound' }
/bookings/{bookingId}/payments:
parameters:
- $ref: '#/components/parameters/BookingId'
post:
tags: [Payments]
operationId: startPayment
summary: Create a Stripe PaymentIntent (manual capture) for a pending booking with a signed waiver
x-required-roles: [customer]
parameters:
- $ref: '#/components/parameters/IdempotencyKey'
responses:
'201':
description: PaymentIntent created; confirm client-side with Stripe.js
content:
application/json:
schema: { $ref: '#/components/schemas/PaymentIntentStart' }
'401': { $ref: '#/components/responses/Unauthorized' }
'404': { $ref: '#/components/responses/NotFound' }
'409': { $ref: '#/components/responses/Conflict' }
'429': { $ref: '#/components/responses/TooManyRequests' }
get:
tags: [Payments]
operationId: listPayments
x-required-roles: [customer, admin]
responses:
'200':
description: Payment ledger for the booking
content:
application/json:
schema:
type: object
required: [data]
properties:
data:
type: array
items: { $ref: '#/components/schemas/Payment' }
'401': { $ref: '#/components/responses/Unauthorized' }
'404': { $ref: '#/components/responses/NotFound' }
/bookings/{bookingId}/refunds:
post:
tags: [Payments]
operationId: createRefund
summary: Issue a partial or full refund
x-required-roles: [admin]
parameters:
- $ref: '#/components/parameters/BookingId'
- $ref: '#/components/parameters/IdempotencyKey'
requestBody:
required: true
content:
application/json:
schema: { $ref: '#/components/schemas/RefundCreate' }
responses:
'202':
description: Refund accepted
content:
application/json:
schema: { $ref: '#/components/schemas/Payment' }
'401': { $ref: '#/components/responses/Unauthorized' }
'403': { $ref: '#/components/responses/Forbidden' }
'404': { $ref: '#/components/responses/NotFound' }
'409': { $ref: '#/components/responses/Conflict' }
'422': { $ref: '#/components/responses/ValidationError' }
/webhooks/stripe:
post:
tags: [Webhooks]
operationId: stripeWebhook
summary: Stripe event receiver (verified via Stripe-Signature header)
security: []
parameters:
- { name: Stripe-Signature, in: header, required: true, schema: { type: string } }
requestBody:
required: true
content:
application/json:
schema: { type: object }
responses:
'200':
description: Processed or duplicate
content:
application/json:
schema:
type: object
properties:
status: { type: string, enum: [processed, duplicate] }
'400': { $ref: '#/components/responses/BadRequest' }
components:
securitySchemes:
bearerAuth:
type: http
scheme: bearer
bearerFormat: JWT
description: Auth0 RS256 access token, audience `https://api.trailbook.app`, 15-minute lifetime.
auth0:
type: oauth2
flows:
authorizationCode:
authorizationUrl: https://auth.trailbook.app/authorize
tokenUrl: https://auth.trailbook.app/oauth/token
refreshUrl: https://auth.trailbook.app/oauth/token
scopes:
openid: OpenID Connect
offline_access: Issue a rotating refresh token
parameters:
Limit:
name: limit
in: query
schema: { type: integer, minimum: 1, maximum: 100, default: 20 }
Cursor:
name: cursor
in: query
description: Opaque cursor from `page.next_cursor`. Must be used with the same `sort`.
schema: { type: string, maxLength: 512 }
IdempotencyKey:
name: Idempotency-Key
in: header
required: true
schema: { type: string, minLength: 8, maxLength: 64, pattern: '^[A-Za-z0-9_-]+$' }
TourId:
name: tourId
in: path
required: true
schema: { type: string, format: uuid }
TrailId:
name: trailId
in: path
required: true
schema: { type: string, format: uuid }
GuideId:
name: guideId
in: path
required: true
schema: { type: string, format: uuid }
BookingId:
name: bookingId
in: path
required: true
schema: { type: string, format: uuid }
headers:
Location:
schema: { type: string }
IdempotentReplayed:
description: Present and `true` when the response is a replay of an earlier request.
schema: { type: string, enum: ['true'] }
RateLimitLimit:
schema: { type: integer }
RateLimitRemaining:
schema: { type: integer }
RateLimitReset:
description: Seconds until the window resets.
schema: { type: integer }
RetryAfter:
schema: { type: integer }
responses:
BadRequest:
description: Malformed request (e.g. invalid cursor, bad signature)
content:
application/problem+json:
schema: { $ref: '#/components/schemas/Problem' }
Unauthorized:
description: Missing, invalid, or expired access token
headers:
WWW-Authenticate:
schema: { type: string, example: 'Bearer error="invalid_token"' }
content:
application/problem+json:
schema: { $ref: '#/components/schemas/Problem' }
Forbidden:
description: Authenticated but not permitted (role, approval, verification)
content:
application/problem+json:
schema: { $ref: '#/components/schemas/Problem' }
NotFound:
description: Resource not found or not visible to caller
content:
application/problem+json:
schema: { $ref: '#/components/schemas/Problem' }
Conflict:
description: >-
State conflict, e.g. INSUFFICIENT_CAPACITY, TOUR_NOT_BOOKABLE, WAIVER_REQUIRED,
HOLD_EXPIRED, INVALID_STATE_TRANSITION, GUIDE_SCHEDULE_CONFLICT
content:
application/problem+json:
schema: { $ref: '#/components/schemas/Problem' }
ValidationError:
description: Request failed validation
content:
application/problem+json:
schema: { $ref: '#/components/schemas/Problem' }
TooManyRequests:
description: Rate limit exceeded
headers:
Retry-After: { $ref: '#/components/headers/RetryAfter' }
content:
application/problem+json:
schema: { $ref: '#/components/schemas/Problem' }
schemas:
Problem:
type: object
required: [type, title, status, code]
properties:
type: { type: string, format: uri, example: 'https://docs.trailbook.app/problems/insufficient-capacity' }
title: { type: string, example: Conflict }
status: { type: integer, example: 409 }
code: { type: string, example: INSUFFICIENT_CAPACITY }
detail: { type: string, nullable: true, example: Only 2 spot(s) remaining. }
instance: { type: string, example: /api/v1/bookings }
request_id: { type: string, example: 7f3c2a9e1b4d4e0f9a8b6c5d4e3f2a1b }
errors:
type: array
items:
type: object
required: [field, message]
properties:
in: { type: string, enum: [body, query, path, header] }
field: { type: string, example: num_participants }
message: { type: string }
type: { type: string }
additionalProperties: true
PageInfo:
type: object
required: [limit, has_more, next_cursor]
properties:
limit: { type: integer }
has_more: { type: boolean }
next_cursor: { type: string, nullable: true }
Currency:
type: string
pattern: '^[A-Z]{3}$'
example: EUR
Role:
type: string
enum: [customer, guide, admin]
Difficulty:
type: string
enum: [easy, moderate, strenuous, technical]
TourStatus:
type: string
enum: [scheduled, cancelled, completed]
BookingStatus:
type: string
enum: [pending, confirmed, cancelled, refunded, completed]
PaymentStatus:
type: string
enum: [authorized, captured, refunded, failed]
GeoPoint:
type: object
required: [lat, lng]
properties:
lat: { type: number, minimum: -90, maximum: 90 }
lng: { type: number, minimum: -180, maximum: 180 }
User:
type: object
required: [id, email, full_name, role, is_verified, created_at]
properties:
id: { type: string, format: uuid }
email: { type: string, format: email }
full_name: { type: string }
phone: { type: string, nullable: true, example: '+14155550123' }
role: { $ref: '#/components/schemas/Role' }
is_verified: { type: boolean }
guide_profile_id: { type: string, format: uuid, nullable: true }
created_at: { type: string, format: date-time }
UserUpdate:
type: object
additionalProperties: false
minProperties: 1
properties:
full_name: { type: string, minLength: 1, maxLength: 200 }
phone: { type: string, pattern: '^\+[1-9]\d{6,14}$', nullable: true }
GuideProfile:
type: object
required: [id, user_id, display_name, bio, certifications, years_experience, languages_spoken, is_approved]
properties:
id: { type: string, format: uuid }
user_id: { type: string, format: uuid }
display_name: { type: string }
bio: { type: string }
certifications: { type: array, items: { type: string } }
years_experience: { type: integer }
languages_spoken: { type: array, items: { type: string } }
rating_avg: { type: number, nullable: true, minimum: 1, maximum: 5 }
rating_count: { type: integer }
is_approved: { type: boolean }
GuideProfileCreate:
type: object
additionalProperties: false
required: [bio, certifications, years_experience, languages_spoken]
properties:
bio: { type: string, minLength: 20, maxLength: 5000 }
certifications:
type: array
maxItems: 20
items: { type: string, maxLength: 120, example: Wilderness First Responder }
years_experience: { type: integer, minimum: 0, maximum: 80 }
languages_spoken:
type: array
minItems: 1
maxItems: 15
items: { type: string, pattern: '^[a-z]{2}(-[A-Z]{2})?$' }
GuideSummary:
type: object
required: [id, display_name, languages_spoken]
properties:
id: { type: string, format: uuid }
display_name: { type: string }
rating_avg: { type: number, nullable: true }
languages_spoken: { type: array, items: { type: string } }
TrailSummary:
type: object
required: [id, name, region, difficulty, distance_km, elevation_gain_m]
properties:
id: { type: string, format: uuid }
name: { type: string }
region: { type: string }
difficulty: { $ref: '#/components/schemas/Difficulty' }
distance_km: { type: number }
elevation_gain_m: { type: integer }
Trail:
allOf:
- $ref: '#/components/schemas/TrailSummary'
- type: object
required: [description, created_by_guide_id, created_at]
properties:
description: { type: string }
gps_track_url: { type: string, format: uri, nullable: true }
created_by_guide_id: { type: string, format: uuid }
created_at: { type: string, format: date-time }
TrailCreate:
type: object
additionalProperties: false
required: [name, region, distance_km, elevation_gain_m, difficulty, description]
properties:
name: { type: string, minLength: 3, maxLength: 200 }
region: { type: string, minLength: 2, maxLength: 80 }
distance_km: { type: number, exclusiveMinimum: true, minimum: 0, maximum: 1000 }
elevation_gain_m: { type: integer, minimum: 0, maximum: 20000 }
difficulty: { $ref: '#/components/schemas/Difficulty' }
description: { type: string, minLength: 20, maxLength: 20000 }
TrailUpdate:
type: object
additionalProperties: false
minProperties: 1
properties:
name: { type: string, minLength: 3, maxLength: 200 }
distance_km: { type: number, exclusiveMinimum: true, minimum: 0, maximum: 1000 }
elevation_gain_m: { type: integer, minimum: 0, maximum: 20000 }
difficulty: { $ref: '#/components/schemas/Difficulty' }
description: { type: string, minLength: 20, maxLength: 20000 }
TrailPage:
type: object
required: [data, page]
properties:
data:
type: array
items: { $ref: '#/components/schemas/Trail' }
page: { $ref: '#/components/schemas/PageInfo' }
Tour:
type: object
required: [id, trail, guide, start_datetime, end_datetime, timezone, max_participants,
spots_remaining, price_cents, currency, status, meeting_point]
properties:
id: { type: string, format: uuid }
trail: { $ref: '#/components/schemas/TrailSummary' }
guide: { $ref: '#/components/schemas/GuideSummary' }
start_datetime: { type: string, format: date-time }
end_datetime: { type: string, format: date-time }
timezone: { type: string, example: Europe/Rome }
max_participants: { type: integer }
spots_remaining: { type: integer }
price_cents: { type: integer, description: Price per participant in minor units }
currency: { $ref: '#/components/schemas/Currency' }
status: { $ref: '#/components/schemas/TourStatus' }
meeting_point: { $ref: '#/components/schemas/GeoPoint' }
TourCreate:
type: object
additionalProperties: false
required: [trail_id, start_datetime, end_datetime, timezone, max_participants, price_cents, currency, meeting_point]
properties:
trail_id: { type: string, format: uuid }
start_datetime: { type: string, format: date-time }
end_datetime: { type: string, format: date-time }
timezone: { type: string, example: America/Denver }
max_participants: { type: integer, minimum: 1, maximum: 50 }
price_cents: { type: integer, minimum: 0, maximum: 10000000 }
currency: { $ref: '#/components/schemas/Currency' }
meeting_point: { $ref: '#/components/schemas/GeoPoint' }
TourUpdate:
type: object
additionalProperties: false
minProperties: 1
properties:
max_participants: { type: integer, minimum: 1, maximum: 50 }
price_cents: { type: integer, minimum: 0 }
meeting_point: { $ref: '#/components/schemas/GeoPoint' }
TourPage:
type: object
required: [data, page]
properties:
data:
type: array
items: { $ref: '#/components/schemas/Tour' }
page: { $ref: '#/components/schemas/PageInfo' }
Booking:
type: object
required: [id, tour_instance_id, customer_id, num_participants, status, total_price_cents,
currency, waiver_signed, created_at, updated_at]
properties:
id: { type: string, format: uuid }
tour_instance_id: { type: string, format: uuid }
customer_id: { type: string, format: uuid }
num_participants: { type: integer, minimum: 1, maximum: 20 }
status: { $ref: '#/components/schemas/BookingStatus' }
total_price_cents: { type: integer }
currency: { $ref: '#/components/schemas/Currency' }
hold_expires_at:
type: string
format: date-time
nullable: true
description: Seats are released if not confirmed by this time.
waiver_signed: { type: boolean }
created_at: { type: string, format: date-time }
updated_at: { type: string, format: date-time }
BookingCreate:
type: object
additionalProperties: false
required: [tour_instance_id, num_participants]
properties:
tour_instance_id: { type: string, format: uuid }
num_participants: { type: integer, minimum: 1, maximum: 20 }
BookingCancellation:
type: object
required: [booking, refund_amount_cents]
properties:
booking: { $ref: '#/components/schemas/Booking' }
refund_amount_cents: { type: integer }
BookingPage:
type: object
required: [data, page]
properties:
data:
type: array
items: { $ref: '#/components/schemas/Booking' }
page: { $ref: '#/components/schemas/PageInfo' }
ManifestEntry:
type: object
required: [booking_id, status, num_participants, lead_name, participant_names, waiver_signed]
properties:
booking_id: { type: string, format: uuid }
status: { $ref: '#/components/schemas/BookingStatus' }
num_participants: { type: integer }
lead_name: { type: string }
lead_phone: { type: string, nullable: true, description: Only exposed within 72h of start }
participant_names: { type: array, items: { type: string } }
waiver_signed: { type: boolean }
ManifestPage:
type: object
required: [data, page]
properties:
data:
type: array
items: { $ref: '#/components/schemas/ManifestEntry' }
page: { $ref: '#/components/schemas/PageInfo' }
WaiverSign:
type: object
additionalProperties: false
required: [accepted, waiver_version, signer_full_name, participant_names]
properties:
accepted:
type: boolean
enum: [true]
waiver_version: { type: string, pattern: '^\d{4}-\d{2}-v\d+$', example: 2026-05-v2 }
signer_full_name: { type: string, minLength: 2, maxLength: 200 }
participant_names:
type: array
minItems: 1
maxItems: 20
items: { type: string, minLength: 2, maxLength: 200 }
Waiver:
type: object
required: [id, booking_id, waiver_version, signer_full_name, signed_at, pdf_status]
properties:
id: { type: string, format: uuid }
booking_id: { type: string, format: uuid }
waiver_version: { type: string }
signer_full_name: { type: string }
signed_at: { type: string, format: date-time }
pdf_status: { type: string, enum: [pending, ready, failed] }
signed_pdf_url:
type: string
format: uri
nullable: true
description: Presigned URL valid for 5 minutes; null until PDF is rendered.
PaymentIntentStart:
type: object
required: [provider, provider_payment_intent_id, client_secret, amount_cents, currency]
properties:
provider: { type: string, enum: [stripe] }
provider_payment_intent_id: { type: string }
client_secret: { type: string, description: Pass to Stripe.js; never log. }
amount_cents: { type: integer }
currency: { $ref: '#/components/schemas/Currency' }
Payment:
type: object
required: [id, booking_id, kind, provider, amount_cents, currency, status, created_at]
properties:
id: { type: string, format: uuid }
booking_id: { type: string, format: uuid }
kind: { type: string, enum: [charge, refund] }
provider: { type: string, enum: [stripe] }
amount_cents: { type: integer }
currency: { $ref: '#/components/schemas/Currency' }
status: { $ref: '#/components/schemas/PaymentStatus' }
created_at: { type: string, format: date-time }
RefundCreate:
type: object
additionalProperties: false
required: [amount_cents, reason]
properties:
amount_cents: { type: integer, minimum: 1 }
reason: { type: string, minLength: 3, maxLength: 500 }
Review:
type: object
required: [id, tour_instance_id, author_display_name, rating, created_at]
properties:
id: { type: string, format: uuid }
tour_instance_id: { type: string, format: uuid }
author_display_name: { type: string, example: Anna K. }
rating: { type: integer, minimum: 1, maximum: 5 }
comment: { type: string, nullable: true }
created_at: { type: string, format: date-time }
ReviewCreate:
type: object
additionalProperties: false
required: [rating]
properties:
rating: { type: integer, minimum: 1, maximum: 5 }
comment: { type: string, maxLength: 4000 }
ReviewPage:
type: object
required: [data, page]
properties:
data:
type: array
items: { $ref: '#/components/schemas/Review' }
page: { $ref: '#/components/schemas/PageInfo' }
In production, generate the canonical spec from FastAPI (app.openapi()) and diff it against this reviewed contract in CI with oasdiff. Breaking changes then fail the build.
4. Implementation Skeleton
Project layout
trailbook/
├── app/
│ ├── main.py
│ ├── core/
│ │ ├── config.py # pydantic-settings
│ │ ├── db.py # async engines, session deps
│ │ ├── errors.py # RFC 9457 problem+json
│ │ ├── security.py # Auth0 JWT verification, Principal, RBAC deps
│ │ ├── ratelimit.py # Redis rate limiting
│ │ └── pagination.py # keyset cursors
│ ├── models.py # SQLAlchemy 2.0 models
│ ├── schemas.py # Pydantic v2 schemas
│ ├── services/
│ │ ├── audit.py
│ │ ├── payment_gateway.py
│ │ ├── bookings.py
│ │ └── payments.py # webhook handlers
│ └── api/v1/
│ ├── tours.py
│ ├── bookings.py
│ └── webhooks.py
├── alembic/ (env.py, versions/)
└── jobs/expire_holds.sql
app/core/config.py
from functools import lru_cache
from typing import Literal
from pydantic import PostgresDsn, RedisDsn, SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
"""Loaded from env. In AWS, values are injected from Secrets Manager / SSM by ECS."""
model_config = SettingsConfigDict(env_prefix="TRAILBOOK_", env_file=".env", extra="ignore")
env: Literal["local", "staging", "production"] = "local"
database_url: PostgresDsn # postgresql+asyncpg://...
database_read_url: PostgresDsn | None = None
db_pool_size: int = 5
db_max_overflow: int = 5
redis_url: RedisDsn
auth0_domain: str # e.g. auth.trailbook.app
auth0_audience: str # e.g. https://api.trailbook.app
auth0_claims_namespace: str = "https://trailbook.app"
stripe_secret_key: SecretStr
stripe_webhook_secret: SecretStr
cors_origins: list[str] = []
booking_hold_minutes: int = 15
payment_grace_minutes: int = 15
booking_cutoff_hours: int = 24
current_waiver_version: str = "2026-05-v2"
@lru_cache
def get_settings() -> Settings:
return Settings() # type: ignore[call-arg]
app/core/db.py
from collections.abc import AsyncIterator
from typing import Annotated
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker, create_async_engine
from app.core.config import get_settings
settings = get_settings()
def _make_engine(url: str) -> AsyncEngine:
return create_async_engine(
url,
pool_size=settings.db_pool_size,
max_overflow=settings.db_max_overflow,
pool_timeout=5, # fail fast instead of queueing forever under overload
pool_recycle=1800,
pool_pre_ping=True, # survives Multi-AZ failover (stale connections are replaced)
connect_args={
"command_timeout": 10,
"server_settings": {
"application_name": "trailbook-api",
"statement_timeout": "5000",
"idle_in_transaction_session_timeout": "10000",
},
},
)
engine = _make_engine(str(settings.database_url))
read_engine = _make_engine(str(settings.database_read_url)) if settings.database_read_url else engine
SessionFactory = async_sessionmaker(engine, expire_on_commit=False, autoflush=False)
ReadSessionFactory = async_sessionmaker(read_engine, expire_on_commit=False, autoflush=False)
async def get_session() -> AsyncIterator[AsyncSession]:
# Transactions are committed explicitly in services; anything uncommitted is rolled back on close.
async with SessionFactory() as session:
yield session
async def get_read_session() -> AsyncIterator[AsyncSession]:
# Replica: only for lag-tolerant public reads (search). NEVER for booking/availability decisions.
async with ReadSessionFactory() as session:
yield session
SessionDep = Annotated[AsyncSession, Depends(get_session)]
ReadSessionDep = Annotated[AsyncSession, Depends(get_read_session)]
app/core/errors.py
from __future__ import annotations
import logging
from typing import Any
from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from sqlalchemy.exc import DBAPIError, IntegrityError
from starlette.exceptions import HTTPException as StarletteHTTPException
log = logging.getLogger("trailbook.errors")
PROBLEM_TYPE_BASE = "https://docs.trailbook.app/problems/"
class APIError(Exception):
status_code: int = 500
code: str = "INTERNAL_ERROR"
title: str = "Internal server error"
def __init__(
self,
detail: str | None = None,
*,
code: str | None = None,
title: str | None = None,
errors: list[dict[str, Any]] | None = None,
headers: dict[str, str] | None = None,
extra: dict[str, Any] | None = None,
) -> None:
super().__init__(detail or self.title)
self.detail = detail
if code:
self.code = code
if title:
self.title = title
self.errors = errors
self.headers = headers
self.extra = extra or {}
class BadRequest(APIError):
status_code, code, title = 400, "BAD_REQUEST", "Bad request"
class Unauthorized(APIError):
status_code, code, title = 401, "UNAUTHORIZED", "Authentication required"
def __init__(self, detail: str | None = None, **kw: Any) -> None:
kw.setdefault("headers", {"WWW-Authenticate": 'Bearer error="invalid_token"'})
super().__init__(detail, **kw)
class Forbidden(APIError):
status_code, code, title = 403, "FORBIDDEN", "Forbidden"
class NotFound(APIError):
status_code, code, title = 404, "NOT_FOUND", "Resource not found"
class Conflict(APIError):
status_code, code, title = 409, "CONFLICT", "Conflict"
class UnprocessableEntity(APIError):
status_code, code, title = 422, "UNPROCESSABLE_ENTITY", "Unprocessable entity"
class TooManyRequests(APIError):
status_code, code, title = 429, "RATE_LIMITED", "Too many requests"
class ServiceUnavailable(APIError):
status_code, code, title = 503, "SERVICE_UNAVAILABLE", "Service temporarily unavailable"
def problem(
request: Request,
*,
status: int,
code: str,
title: str,
detail: str | None = None,
errors: list[dict[str, Any]] | None = None,
headers: dict[str, str] | None = None,
extra: dict[str, Any] | None = None,
) -> JSONResponse:
body: dict[str, Any] = {
"type": PROBLEM_TYPE_BASE + code.lower().replace("_", "-"),
"title": title,
"status": status,
"code": code,
"detail": detail,
"instance": request.url.path,
"request_id": getattr(request.state, "request_id", None),
}
if errors:
body["errors"] = errors
body.update(extra or {})
return JSONResponse(body, status_code=status, headers=headers, media_type="application/problem+json")
def sqlstate(exc: DBAPIError) -> str | None:
orig = exc.orig
return getattr(orig, "sqlstate", None) or getattr(getattr(orig, "__cause__", None), "sqlstate", None)
def register_exception_handlers(app: FastAPI) -> None:
@app.exception_handler(APIError)
async def _api_error(request: Request, exc: APIError) -> JSONResponse:
return problem(request, status=exc.status_code, code=exc.code, title=exc.title,
detail=exc.detail, errors=exc.errors, headers=exc.headers, extra=exc.extra)
@app.exception_handler(RequestValidationError)
async def _validation(request: Request, exc: RequestValidationError) -> JSONResponse:
errors = []
for err in exc.errors():
loc = err.get("loc", ())
errors.append({
"in": loc[0] if loc else "body",
"field": ".".join(str(p) for p in loc[1:]) or str(loc[0] if loc else ""),
"message": err.get("msg"),
"type": err.get("type"),
# Deliberately NOT echoing "input": it may contain PII.
})
return problem(request, status=422, code="VALIDATION_ERROR",
title="Request validation failed", errors=errors)
@app.exception_handler(StarletteHTTPException)
async def _http(request: Request, exc: StarletteHTTPException) -> JSONResponse:
code = {404: "ROUTE_NOT_FOUND", 405: "METHOD_NOT_ALLOWED"}.get(exc.status_code, "HTTP_ERROR")
return problem(request, status=exc.status_code, code=code, title=str(exc.detail),
headers=getattr(exc, "headers", None))
@app.exception_handler(IntegrityError)
async def _integrity(request: Request, exc: IntegrityError) -> JSONResponse:
log.warning("integrity_error", extra={"sqlstate": sqlstate(exc)})
return problem(request, status=409, code="CONSTRAINT_VIOLATION",
title="Conflict", detail="The request conflicts with the current state.")
@app.exception_handler(DBAPIError)
async def _db(request: Request, exc: DBAPIError) -> JSONResponse:
log.exception("database_error")
return problem(request, status=503, code="DATABASE_UNAVAILABLE",
title="Service temporarily unavailable", headers={"Retry-After": "2"})
@app.exception_handler(Exception)
async def _unhandled(request: Request, exc: Exception) -> JSONResponse:
log.exception("unhandled_exception")
return problem(request, status=500, code="INTERNAL_ERROR", title="Internal server error")
app/core/security.py
from __future__ import annotations
import asyncio
import logging
import time
import uuid
from dataclasses import dataclass
from typing import Annotated, Any
import httpx
import jwt
from fastapi import Depends, Request
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from sqlalchemy import select
from sqlalchemy.dialects.postgresql import insert as pg_insert
from app.core.config import Settings, get_settings
from app.core.db import SessionDep
from app.core.errors import Conflict, Forbidden, ServiceUnavailable, Unauthorized
from app.models import GuideProfile, Role, User
log = logging.getLogger("trailbook.auth")
JWKS_TTL_S = 600 # routine refresh
JWKS_MIN_REFRESH_S = 30 # throttle forced refresh on unknown `kid` (prevents JWKS-flood DoS)
class TokenVerifier:
"""Verifies Auth0 RS256 access tokens against a cached JWKS."""
def __init__(self, settings: Settings, http: httpx.AsyncClient) -> None:
self._issuer = f"https://{settings.auth0_domain}/"
self._audience = settings.auth0_audience
self._jwks_url = f"{self._issuer}.well-known/jwks.json"
self._http = http
self._keys: dict[str, jwt.PyJWK] = {}
self._fetched_at = float("-inf")
self._lock = asyncio.Lock()
async def _refresh(self, *, force: bool) -> None:
async with self._lock:
age = time.monotonic() - self._fetched_at
if (not force and age < JWKS_TTL_S) or (force and age < JWKS_MIN_REFRESH_S):
return
try:
resp = await self._http.get(self._jwks_url, timeout=3.0)
resp.raise_for_status()
self._keys = {
k["kid"]: jwt.PyJWK(k, algorithm="RS256")
for k in resp.json()["keys"]
if k.get("kty") == "RSA" and k.get("use", "sig") == "sig"
}
self._fetched_at = time.monotonic()
except (httpx.HTTPError, KeyError, ValueError):
log.exception("jwks_refresh_failed")
if not self._keys: # keep serving with last-known keys if we have any
raise ServiceUnavailable("Identity provider unavailable.", code="IDP_UNAVAILABLE")
async def verify(self, token: str) -> dict[str, Any]:
try:
header = jwt.get_unverified_header(token)
except jwt.PyJWTError as exc:
raise Unauthorized("Malformed access token.", code="INVALID_TOKEN") from exc
if header.get("alg") != "RS256" or "kid" not in header: # blocks alg=none / HS256 confusion
raise Unauthorized("Unsupported token algorithm.", code="INVALID_TOKEN")
await self._refresh(force=False)
if header["kid"] not in self._keys:
await self._refresh(force=True) # key rotation
key = self._keys.get(header["kid"])
if key is None:
raise Unauthorized("Unknown signing key.", code="INVALID_TOKEN")
try:
return jwt.decode(
token, key.key, algorithms=["RS256"],
audience=self._audience, issuer=self._issuer, leeway=30,
options={"require": ["exp", "iat", "sub", "iss", "aud"]},
)
except jwt.ExpiredSignatureError as exc:
raise Unauthorized("Access token expired.", code="TOKEN_EXPIRED") from exc
except jwt.PyJWTError as exc:
raise Unauthorized("Invalid access token.", code="INVALID_TOKEN") from exc
@dataclass(frozen=True, slots=True)
class Principal:
"""Immutable snapshot of the caller. Plain values, not an ORM object, so a later
session.rollback() in a service can never expire it."""
user_id: uuid.UUID
auth0_sub: str
role: Role
is_verified: bool
guide_profile_id: uuid.UUID | None
guide_approved: bool
@property
def is_admin(self) -> bool:
return self.role is Role.admin
_bearer = HTTPBearer(auto_error=False, description="Auth0 RS256 access token")
async def get_token_claims(
request: Request,
creds: Annotated[HTTPAuthorizationCredentials | None, Depends(_bearer)],
) -> dict[str, Any]:
if creds is None:
raise Unauthorized("Missing bearer token.", code="MISSING_TOKEN")
claims = await request.app.state.token_verifier.verify(creds.credentials)
request.state.subject = claims["sub"]
return claims
async def get_current_principal(
claims: Annotated[dict[str, Any], Depends(get_token_claims)],
session: SessionDep,
) -> Principal:
ns = get_settings().auth0_claims_namespace
stmt = (
select(User, GuideProfile.id, GuideProfile.is_approved)
.outerjoin(GuideProfile, GuideProfile.user_id == User.id)
.where(User.auth0_sub == claims["sub"])
)
row = (await session.execute(stmt)).one_or_none()
if row is None:
row = await _provision_user(session, claims, ns, stmt)
user, gp_id, gp_approved = row
if user.is_suspended or user.deleted_at is not None:
raise Forbidden("Account is disabled.", code="ACCOUNT_DISABLED")
# Email verification is owned by Auth0; mirror it (injected by an Auth0 Action as a custom claim).
if claims.get(f"{ns}/email_verified") is True and not user.is_verified:
user.is_verified = True
await session.commit()
return Principal(
user_id=user.id, auth0_sub=user.auth0_sub, role=user.role, is_verified=user.is_verified,
guide_profile_id=gp_id, guide_approved=bool(gp_approved),
)
async def _provision_user(session, claims: dict[str, Any], ns: str, stmt) -> Any:
"""Just-in-time provisioning on first authenticated call. New users are always customers."""
email = claims.get(f"{ns}/email")
if not email:
raise Forbidden("Token lacks email claim.", code="USER_NOT_PROVISIONED")
await session.execute(
pg_insert(User)
.values(
id=uuid.uuid4(),
auth0_sub=claims["sub"],
email=email.strip().lower(),
full_name=(claims.get(f"{ns}/name") or email.split("@")[0])[:200],
role=Role.customer,
is_verified=claims.get(f"{ns}/email_verified") is True,
)
.on_conflict_do_nothing() # races on auth0_sub OR an email owned by another identity
)
await session.commit()
row = (await session.execute(stmt)).one_or_none()
if row is None:
# Email already registered under a different Auth0 identity (e.g. Google vs. password).
raise Conflict("An account with this email already exists; link accounts in Auth0.",
code="ACCOUNT_LINK_REQUIRED")
return row
CurrentPrincipal = Annotated[Principal, Depends(get_current_principal)]
def require_roles(*roles: Role):
allowed = frozenset(roles)
async def _dep(principal: CurrentPrincipal) -> Principal:
if principal.role not in allowed:
raise Forbidden("Insufficient role for this operation.", code="INSUFFICIENT_ROLE")
return principal
return _dep
async def require_approved_guide(principal: CurrentPrincipal) -> Principal:
if principal.role is not Role.guide or not principal.guide_approved or principal.guide_profile_id is None:
raise Forbidden("An approved guide profile is required.", code="GUIDE_NOT_APPROVED")
return principal
CustomerPrincipal = Annotated[Principal, Depends(require_roles(Role.customer))]
CustomerOrAdmin = Annotated[Principal, Depends(require_roles(Role.customer, Role.admin))]
AdminPrincipal = Annotated[Principal, Depends(require_roles(Role.admin))]
ApprovedGuide = Annotated[Principal, Depends(require_approved_guide)]
app/core/ratelimit.py
import ipaddress
import logging
import time
from dataclasses import dataclass
from typing import Annotated, Any, Literal
from fastapi import Depends, Request, Response
from redis.asyncio import Redis
from redis.exceptions import RedisError
from app.core.errors import TooManyRequests
from app.core.security import get_token_claims
log = logging.getLogger("trailbook.ratelimit")
@dataclass(frozen=True, slots=True)
class RateLimitResult:
allowed: bool
limit: int
remaining: int
reset_s: int
class RateLimiter:
"""Fixed-window counter in Redis. Coarse, cheap, and good enough behind WAF rate rules.
Swap for GCRA (redis-cell) if smoother limits are needed."""
def __init__(self, redis: Redis) -> None:
self._r = redis
async def hit(self, key: str, limit: int, window_s: int) -> RateLimitResult:
now = int(time.time())
window_start = now - (now % window_s)
rkey = f"rl:{key}:{window_start}"
try:
async with self._r.pipeline(transaction=True) as pipe:
count, _ = await pipe.incr(rkey).expire(rkey, window_s + 1).execute()
except RedisError:
# Fail OPEN: a Redis blip must not take down bookings. WAF still protects the edge.
log.warning("rate_limiter_unavailable", extra={"key": key})
return RateLimitResult(True, limit, limit, window_s)
return RateLimitResult(count <= limit, limit, max(0, limit - count), window_start + window_s - now)
def client_ip(request: Request) -> str:
# uvicorn runs with --proxy-headers and --forwarded-allow-ips=<ALB subnets>, so this is the real client.
host = request.client.host if request.client else "unknown"
try:
ip = ipaddress.ip_address(host)
except ValueError:
return host
if ip.version == 6: # bucket IPv6 by /64 so one host can't rotate addresses
return str(ipaddress.ip_network(f"{ip}/64", strict=False).network_address)
return str(ip)
async def _apply(request: Request, response: Response, key: str, limit: int, window_s: int) -> None:
result = await request.app.state.rate_limiter.hit(key, limit, window_s)
headers = {
"RateLimit-Limit": str(result.limit),
"RateLimit-Remaining": str(result.remaining),
"RateLimit-Reset": str(result.reset_s),
}
if not result.allowed:
raise TooManyRequests("Rate limit exceeded.", headers={**headers, "Retry-After": str(result.reset_s)})
response.headers.update(headers)
def rate_limit(scope: str, *, limit: int, window_s: int = 60, per: Literal["user", "ip"] = "user") -> Any:
if per == "user":
async def _dep(request: Request, response: Response,
claims: Annotated[dict[str, Any], Depends(get_token_claims)]) -> None:
await _apply(request, response, f"{scope}:u:{claims['sub']}", limit, window_s)
else:
async def _dep(request: Request, response: Response) -> None:
await _apply(request, response, f"{scope}:ip:{client_ip(request)}", limit, window_s)
return Depends(_dep)
app/core/pagination.py
import base64
import json
import uuid
from datetime import datetime
from typing import Any
from app.core.errors import BadRequest
def encode_cursor(sort: str, value: Any, row_id: uuid.UUID) -> str:
v = value.isoformat() if isinstance(value, datetime) else value
raw = json.dumps({"s": sort, "v": v, "id": str(row_id)}, separators=(",", ":"))
return base64.urlsafe_b64encode(raw.encode()).decode().rstrip("=")
def decode_cursor(cursor: str, sort: str) -> tuple[Any, uuid.UUID]:
try:
data = json.loads(base64.urlsafe_b64decode(cursor + "=" * (-len(cursor) % 4)))
if data["s"] != sort:
raise ValueError("sort mismatch")
return data["v"], uuid.UUID(data["id"])
except (ValueError, KeyError, TypeError) as exc:
raise BadRequest("Invalid or stale cursor; restart pagination.", code="INVALID_CURSOR") from exc
app/models.py
from __future__ import annotations
import enum
import ipaddress
import uuid
from datetime import UTC, datetime
from decimal import Decimal
from typing import Any
from sqlalchemy import (
BigInteger, CheckConstraint, DateTime, Enum as SAEnum, ForeignKey, Identity, Index,
Numeric, SmallInteger, String, Text, UniqueConstraint, func, text,
)
from sqlalchemy.dialects.postgresql import ARRAY, INET, JSONB, UUID
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
def utcnow() -> datetime:
return datetime.now(UTC)
class Base(DeclarativeBase):
type_annotation_map = {uuid.UUID: UUID(as_uuid=True), datetime: DateTime(timezone=True)}
class Timestamps:
# Python-side defaults so values are known after flush without a refresh (async-safe).
created_at: Mapped[datetime] = mapped_column(default=utcnow, server_default=func.now())
updated_at: Mapped[datetime] = mapped_column(default=utcnow, onupdate=utcnow, server_default=func.now())
def pg_enum(enum_cls: type[enum.Enum], name: str) -> SAEnum:
return SAEnum(enum_cls, name=name, values_callable=lambda e: [m.value for m in e], validate_strings=True)
class Role(enum.StrEnum):
customer = "customer"
guide = "guide"
admin = "admin"
class Difficulty(enum.StrEnum):
easy = "easy"
moderate = "moderate"
strenuous = "strenuous"
technical = "technical"
class TourStatus(enum.StrEnum):
scheduled = "scheduled"
cancelled = "cancelled"
completed = "completed"
class BookingStatus(enum.StrEnum):
pending = "pending"
confirmed = "confirmed"
cancelled = "cancelled"
refunded = "refunded"
completed = "completed" # A1
class PaymentStatus(enum.StrEnum):
authorized = "authorized"
captured = "captured"
refunded = "refunded"
failed = "failed"
class PaymentKind(enum.StrEnum):
charge = "charge"
refund = "refund"
class User(Timestamps, Base):
__tablename__ = "users"
id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid.uuid4)
auth0_sub: Mapped[str] = mapped_column(String(128), unique=True)
email: Mapped[str] = mapped_column(String(320))
full_name: Mapped[str] = mapped_column(String(200))
phone: Mapped[str | None] = mapped_column(String(20)) # E.164
role: Mapped[Role] = mapped_column(pg_enum(Role, "user_role"), default=Role.customer)
is_verified: Mapped[bool] = mapped_column(default=False, server_default=text("false"))
is_suspended: Mapped[bool] = mapped_column(default=False, server_default=text("false"))
deleted_at: Mapped[datetime | None]
guide_profile: Mapped[GuideProfile | None] = relationship(
back_populates="user", foreign_keys="GuideProfile.user_id", uselist=False, lazy="raise")
class GuideProfile(Timestamps, Base):
__tablename__ = "guide_profiles"
id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid.uuid4)
user_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("users.id", ondelete="RESTRICT"), unique=True)
bio: Mapped[str] = mapped_column(Text, default="")
certifications: Mapped[list[str]] = mapped_column(ARRAY(String(120)), default=list, server_default="{}")
years_experience: Mapped[int] = mapped_column(SmallInteger)
languages_spoken: Mapped[list[str]] = mapped_column(ARRAY(String(8)), default=list, server_default="{}")
rating_avg: Mapped[Decimal | None] = mapped_column(Numeric(3, 2))
rating_count: Mapped[int] = mapped_column(default=0, server_default="0")
is_approved: Mapped[bool] = mapped_column(default=False, server_default=text("false"))
approved_at: Mapped[datetime | None]
approved_by: Mapped[uuid.UUID | None] = mapped_column(ForeignKey("users.id"))
user: Mapped[User] = relationship(back_populates="guide_profile", foreign_keys=[user_id], lazy="raise")
__table_args__ = (
CheckConstraint("years_experience BETWEEN 0 AND 80", name="ck_guide_years"),
Index("ix_guide_languages_gin", "languages_spoken", postgresql_using="gin"),
)
class Trail(Timestamps, Base):
__tablename__ = "trails"
id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid.uuid4)
name: Mapped[str] = mapped_column(String(200))
region: Mapped[str] = mapped_column(String(80))
distance_km: Mapped[Decimal] = mapped_column(Numeric(6, 2))
elevation_gain_m: Mapped[int]
difficulty: Mapped[Difficulty] = mapped_column(pg_enum(Difficulty, "trail_difficulty"))
description: Mapped[str] = mapped_column(Text)
gps_track_url: Mapped[str | None] = mapped_column(String(1024))
created_by_guide_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("guide_profiles.id"))
deleted_at: Mapped[datetime | None]
__table_args__ = (
CheckConstraint("distance_km > 0", name="ck_trail_distance"),
CheckConstraint("elevation_gain_m >= 0", name="ck_trail_elevation"),
)
class TourInstance(Timestamps, Base):
__tablename__ = "tour_instances"
id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid.uuid4)
trail_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("trails.id"))
guide_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("guide_profiles.id"))
start_datetime: Mapped[datetime]
end_datetime: Mapped[datetime]
timezone: Mapped[str] = mapped_column(String(64))
max_participants: Mapped[int] = mapped_column(SmallInteger)
# Denormalized sum of num_participants over pending+confirmed bookings. The single
# source of truth for capacity; mutated only via atomic conditional UPDATEs.
seats_reserved: Mapped[int] = mapped_column(SmallInteger, default=0, server_default="0")
price_cents: Mapped[int]
currency: Mapped[str] = mapped_column(String(3))
status: Mapped[TourStatus] = mapped_column(pg_enum(TourStatus, "tour_status"), default=TourStatus.scheduled)
meeting_point_lat: Mapped[Decimal] = mapped_column(Numeric(9, 6))
meeting_point_lng: Mapped[Decimal] = mapped_column(Numeric(9, 6))
cancelled_reason: Mapped[str | None] = mapped_column(Text)
trail: Mapped[Trail] = relationship(lazy="raise")
guide: Mapped[GuideProfile] = relationship(lazy="raise")
__table_args__ = (
CheckConstraint("end_datetime > start_datetime", name="ck_tour_time_order"),
CheckConstraint("max_participants BETWEEN 1 AND 50", name="ck_tour_max_participants"),
CheckConstraint("seats_reserved >= 0 AND seats_reserved <= max_participants", name="ck_tour_capacity"),
CheckConstraint("price_cents >= 0", name="ck_tour_price"),
CheckConstraint("currency ~ '^[A-Z]{3}$'", name="ck_tour_currency"),
CheckConstraint("meeting_point_lat BETWEEN -90 AND 90 AND meeting_point_lng BETWEEN -180 AND 180",
name="ck_tour_geo"),
Index("ix_tours_scheduled_start", "start_datetime", "id", postgresql_where=text("status = 'scheduled'")),
Index("ix_tours_scheduled_price", "price_cents", "id", postgresql_where=text("status = 'scheduled'")),
Index("ix_tours_trail_start", "trail_id", "start_datetime"),
Index("ix_tours_guide_start", "guide_id", "start_datetime"),
)
class Booking(Timestamps, Base):
__tablename__ = "bookings"
id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid.uuid4)
tour_instance_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("tour_instances.id"))
customer_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("users.id"))
num_participants: Mapped[int] = mapped_column(SmallInteger)
status: Mapped[BookingStatus] = mapped_column(pg_enum(BookingStatus, "booking_status"),
default=BookingStatus.pending)
total_price_cents: Mapped[int]
currency: Mapped[str] = mapped_column(String(3))
idempotency_key: Mapped[str] = mapped_column(String(64))
request_fingerprint: Mapped[str] = mapped_column(String(64))
hold_expires_at: Mapped[datetime | None]
cancelled_at: Mapped[datetime | None]
cancel_reason: Mapped[str | None] = mapped_column(String(500))
tour_instance: Mapped[TourInstance] = relationship(lazy="raise")
waiver: Mapped[Waiver | None] = relationship(back_populates="booking", uselist=False, lazy="raise")
payments: Mapped[list[Payment]] = relationship(back_populates="booking", lazy="raise")
__table_args__ = (
UniqueConstraint("customer_id", "idempotency_key", name="uq_bookings_customer_idem"),
CheckConstraint("num_participants BETWEEN 1 AND 20", name="ck_booking_participants"),
CheckConstraint("total_price_cents >= 0", name="ck_booking_total"),
Index("ix_bookings_tour_status", "tour_instance_id", "status"),
Index("ix_bookings_customer_created", "customer_id", "created_at", "id"),
Index("ix_bookings_pending_hold", "hold_expires_at", postgresql_where=text("status = 'pending'")),
)
class Waiver(Base):
__tablename__ = "waivers"
id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid.uuid4)
booking_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("bookings.id", ondelete="RESTRICT"), unique=True)
waiver_version: Mapped[str] = mapped_column(String(32))
signer_full_name: Mapped[str] = mapped_column(String(200))
participant_names: Mapped[list[str]] = mapped_column(ARRAY(String(200)))
signed_at: Mapped[datetime] = mapped_column(default=utcnow)
ip_address: Mapped[ipaddress.IPv4Address | ipaddress.IPv6Address | None] = mapped_column(INET)
user_agent: Mapped[str | None] = mapped_column(String(512))
signed_pdf_key: Mapped[str] = mapped_column(String(512)) # S3 key; URL is presigned on read
pdf_sha256: Mapped[str | None] = mapped_column(String(64)) # tamper evidence
pdf_status: Mapped[str] = mapped_column(String(16), default="pending")
booking: Mapped[Booking] = relationship(back_populates="waiver", lazy="raise")
class Payment(Timestamps, Base):
__tablename__ = "payments"
id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid.uuid4)
booking_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("bookings.id"))
kind: Mapped[PaymentKind] = mapped_column(pg_enum(PaymentKind, "payment_kind"))
provider: Mapped[str] = mapped_column(String(16), default="stripe")
provider_charge_id: Mapped[str] = mapped_column(String(255), unique=True) # pi_... or re_...
amount_cents: Mapped[int]
currency: Mapped[str] = mapped_column(String(3))
status: Mapped[PaymentStatus] = mapped_column(pg_enum(PaymentStatus, "payment_status"))
failure_code: Mapped[str | None] = mapped_column(String(64))
parent_payment_id: Mapped[uuid.UUID | None] = mapped_column(ForeignKey("payments.id"))
booking: Mapped[Booking] = relationship(back_populates="payments", lazy="raise")
__table_args__ = (
CheckConstraint("amount_cents > 0", name="ck_payment_amount"),
Index("ix_payments_booking", "booking_id", "created_at"),
)
class Review(Base):
__tablename__ = "reviews"
id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid.uuid4)
tour_instance_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("tour_instances.id"))
customer_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("users.id"))
booking_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("bookings.id"))
rating: Mapped[int] = mapped_column(SmallInteger)
comment: Mapped[str | None] = mapped_column(Text)
created_at: Mapped[datetime] = mapped_column(default=utcnow, server_default=func.now())
__table_args__ = (
UniqueConstraint("tour_instance_id", "customer_id", name="uq_review_per_customer_tour"),
CheckConstraint("rating BETWEEN 1 AND 5", name="ck_review_rating"),
CheckConstraint("char_length(comment) <= 4000", name="ck_review_comment_len"),
Index("ix_reviews_tour_created", "tour_instance_id", "created_at", "id"),
)
class AuditLog(Base):
"""Append-only. The app DB role has INSERT/SELECT only (no UPDATE/DELETE)."""
__tablename__ = "audit_logs"
id: Mapped[int] = mapped_column(BigInteger, Identity(always=True), primary_key=True)
occurred_at: Mapped[datetime] = mapped_column(default=utcnow, server_default=func.now())
actor_user_id: Mapped[uuid.UUID | None] # no FK: must survive user erasure
actor_role: Mapped[str] = mapped_column(String(16))
action: Mapped[str] = mapped_column(String(64))
entity_type: Mapped[str] = mapped_column(String(32))
entity_id: Mapped[str] = mapped_column(String(64))
request_id: Mapped[str | None] = mapped_column(String(64))
ip: Mapped[ipaddress.IPv4Address | ipaddress.IPv6Address | None] = mapped_column(INET)
changes: Mapped[dict[str, Any] | None] = mapped_column(JSONB)
__table_args__ = (Index("ix_audit_entity", "entity_type", "entity_id", "occurred_at"),)
class StripeEvent(Base):
__tablename__ = "stripe_events"
id: Mapped[str] = mapped_column(String(255), primary_key=True) # evt_...
type: Mapped[str] = mapped_column(String(128))
received_at: Mapped[datetime] = mapped_column(default=utcnow, server_default=func.now())
class OutboxEvent(Base):
"""Transactional outbox: side effects (refunds, PDF render, emails) are committed atomically
with the state change and executed by a worker with retries."""
__tablename__ = "outbox_events"
id: Mapped[int] = mapped_column(BigInteger, Identity(always=True), primary_key=True)
topic: Mapped[str] = mapped_column(String(64))
payload: Mapped[dict[str, Any]] = mapped_column(JSONB)
created_at: Mapped[datetime] = mapped_column(default=utcnow, server_default=func.now())
processed_at: Mapped[datetime | None]
attempts: Mapped[int] = mapped_column(default=0, server_default="0")
__table_args__ = (Index("ix_outbox_unprocessed", "id", postgresql_where=text("processed_at IS NULL")),)
# Functional indexes (declared after classes so they can reference columns directly)
Index("uq_users_email_lower", func.lower(User.email), unique=True)
Index("ix_trails_region_lower_difficulty", func.lower(Trail.region), Trail.difficulty,
postgresql_where=Trail.deleted_at.is_(None))
app/schemas.py
from __future__ import annotations
import uuid
from datetime import datetime, timedelta
from typing import Annotated, Generic, Literal, TypeVar
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
from pydantic import AwareDatetime, BaseModel, ConfigDict, Field, StringConstraints, field_validator, model_validator
from app.models import Booking, BookingStatus, Difficulty, PaymentStatus, TourInstance, TourStatus, Waiver
T = TypeVar("T")
Currency = Annotated[str, StringConstraints(pattern=r"^[A-Z]{3}$")]
LanguageCode = Annotated[str, StringConstraints(pattern=r"^[a-z]{2}(-[A-Z]{2})?$")]
PersonName = Annotated[str, StringConstraints(strip_whitespace=True, min_length=2, max_length=200)]
class InputModel(BaseModel):
# extra="forbid" blocks mass assignment (e.g. a client sending "is_approved": true or "status").
model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)
class OutputModel(BaseModel):
model_config = ConfigDict(from_attributes=True)
class PageInfo(OutputModel):
limit: int
has_more: bool
next_cursor: str | None
class Page(OutputModel, Generic[T]):
data: list[T]
page: PageInfo
class GeoPoint(BaseModel):
lat: float = Field(ge=-90, le=90)
lng: float = Field(ge=-180, le=180)
# ---------- Tours ----------
class TrailSummary(OutputModel):
id: uuid.UUID
name: str
region: str
difficulty: Difficulty
distance_km: float
elevation_gain_m: int
class GuideSummary(OutputModel):
id: uuid.UUID
display_name: str
rating_avg: float | None
languages_spoken: list[str]
class TourOut(OutputModel):
id: uuid.UUID
trail: TrailSummary
guide: GuideSummary
start_datetime: datetime
end_datetime: datetime
timezone: str
max_participants: int
spots_remaining: int
price_cents: int
currency: str
status: TourStatus
meeting_point: GeoPoint
@classmethod
def from_model(cls, t: TourInstance) -> TourOut:
g = t.guide
return cls(
id=t.id,
trail=TrailSummary.model_validate(t.trail),
guide=GuideSummary(
id=g.id, display_name=g.user.full_name,
rating_avg=float(g.rating_avg) if g.rating_avg is not None else None,
languages_spoken=g.languages_spoken,
),
start_datetime=t.start_datetime, end_datetime=t.end_datetime, timezone=t.timezone,
max_participants=t.max_participants,
spots_remaining=max(0, t.max_participants - t.seats_reserved),
price_cents=t.price_cents, currency=t.currency, status=t.status,
meeting_point=GeoPoint(lat=float(t.meeting_point_lat), lng=float(t.meeting_point_lng)),
)
class TourSearchParams(InputModel):
region: str | None = Field(None, max_length=80)
difficulty: list[Difficulty] = Field(default_factory=list, max_length=4)
trail_id: uuid.UUID | None = None
guide_id: uuid.UUID | None = None
language: LanguageCode | None = None
starts_after: AwareDatetime | None = None
starts_before: AwareDatetime | None = None
min_price_cents: int | None = Field(None, ge=0)
max_price_cents: int | None = Field(None, ge=0)
currency: Currency | None = None
min_spots: int = Field(1, ge=1, le=50)
sort: Literal["start_datetime", "-start_datetime", "price_cents", "-price_cents"] = "start_datetime"
limit: int = Field(20, ge=1, le=100)
cursor: str | None = Field(None, max_length=512)
@model_validator(mode="after")
def _check_ranges(self) -> TourSearchParams:
if self.starts_after and self.starts_before and self.starts_after >= self.starts_before:
raise ValueError("starts_after must be before starts_before")
if self.min_price_cents is not None and self.max_price_cents is not None \
and self.min_price_cents > self.max_price_cents:
raise ValueError("min_price_cents must be <= max_price_cents")
if (self.min_price_cents is not None or self.max_price_cents is not None) and self.currency is None:
raise ValueError("currency is required when filtering by price")
return self
class TourCreate(InputModel):
trail_id: uuid.UUID
start_datetime: AwareDatetime
end_datetime: AwareDatetime
timezone: str = Field(max_length=64)
max_participants: int = Field(ge=1, le=50)
price_cents: int = Field(ge=0, le=10_000_000)
currency: Currency
meeting_point: GeoPoint
@field_validator("timezone")
@classmethod
def _valid_tz(cls, v: str) -> str:
try:
ZoneInfo(v)
except (ZoneInfoNotFoundError, ValueError) as exc:
raise ValueError("must be a valid IANA timezone") from exc
return v
@model_validator(mode="after")
def _check_times(self) -> TourCreate:
if self.end_datetime <= self.start_datetime:
raise ValueError("end_datetime must be after start_datetime")
if self.end_datetime - self.start_datetime > timedelta(days=21):
raise ValueError("tours may last at most 21 days")
return self
# ---------- Bookings ----------
class BookingCreate(InputModel):
tour_instance_id: uuid.UUID
num_participants: int = Field(ge=1, le=20)
class BookingCancelIn(InputModel):
reason: str | None = Field(None, max_length=500)
class BookingOut(OutputModel):
id: uuid.UUID
tour_instance_id: uuid.UUID
customer_id: uuid.UUID
num_participants: int
status: BookingStatus
total_price_cents: int
currency: str
hold_expires_at: datetime | None
waiver_signed: bool
created_at: datetime
updated_at: datetime
@classmethod
def from_booking(cls, b: Booking, *, waiver_signed: bool) -> BookingOut:
return cls(
id=b.id, tour_instance_id=b.tour_instance_id, customer_id=b.customer_id,
num_participants=b.num_participants, status=b.status, total_price_cents=b.total_price_cents,
currency=b.currency, hold_expires_at=b.hold_expires_at, waiver_signed=waiver_signed,
created_at=b.created_at, updated_at=b.updated_at,
)
class BookingCancellationOut(OutputModel):
booking: BookingOut
refund_amount_cents: int
class WaiverSign(InputModel):
accepted: Literal[True]
waiver_version: str = Field(pattern=r"^\d{4}-\d{2}-v\d+$")
signer_full_name: PersonName
participant_names: list[PersonName] = Field(min_length=1, max_length=20)
class WaiverOut(OutputModel):
id: uuid.UUID
booking_id: uuid.UUID
waiver_version: str
signer_full_name: str
signed_at: datetime
pdf_status: Literal["pending", "ready", "failed"]
signed_pdf_url: str | None = None
@classmethod
def from_waiver(cls, w: Waiver, url: str | None = None) -> WaiverOut:
return cls(id=w.id, booking_id=w.booking_id, waiver_version=w.waiver_version,
signer_full_name=w.signer_full_name, signed_at=w.signed_at,
pdf_status=w.pdf_status, signed_pdf_url=url)
class PaymentIntentStartOut(OutputModel):
provider: Literal["stripe"] = "stripe"
provider_payment_intent_id: str
client_secret: str
amount_cents: int
currency: str
class PaymentOut(OutputModel):
id: uuid.UUID
booking_id: uuid.UUID
kind: Literal["charge", "refund"]
provider: str
amount_cents: int
currency: str
status: PaymentStatus
created_at: datetime
app/services/audit.py
import ipaddress
import uuid
from dataclasses import dataclass
from typing import Annotated, Any
from fastapi import Depends, Request
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.ratelimit import client_ip
from app.core.security import CurrentPrincipal
from app.models import AuditLog
@dataclass(frozen=True, slots=True)
class AuditContext:
actor_user_id: uuid.UUID | None
actor_role: str
request_id: str | None
ip: ipaddress.IPv4Address | ipaddress.IPv6Address | None
SYSTEM_STRIPE = AuditContext(actor_user_id=None, actor_role="system:stripe", request_id=None, ip=None)
def _ip(request: Request) -> ipaddress.IPv4Address | ipaddress.IPv6Address | None:
try:
return ipaddress.ip_address(request.client.host) if request.client else None
except ValueError:
return None
async def get_audit_ctx(request: Request, principal: CurrentPrincipal) -> AuditContext:
return AuditContext(principal.user_id, principal.role.value,
getattr(request.state, "request_id", None), _ip(request))
AuditCtx = Annotated[AuditContext, Depends(get_audit_ctx)]
def record(session: AsyncSession, ctx: AuditContext, *, action: str, entity_type: str,
entity_id: Any, changes: dict[str, Any] | None = None) -> None:
"""Adds the audit row to the *same* transaction as the change it describes."""
session.add(AuditLog(
actor_user_id=ctx.actor_user_id, actor_role=ctx.actor_role, action=action,
entity_type=entity_type, entity_id=str(entity_id), request_id=ctx.request_id,
ip=ctx.ip, changes=changes,
))
app/services/payment_gateway.py
import uuid
from typing import Protocol
import stripe
from app.core.config import Settings
class PaymentGateway(Protocol):
async def authorize(self, *, booking_id: uuid.UUID, amount_cents: int, currency: str,
idempotency_key: str) -> tuple[str, str]: ...
async def capture(self, payment_intent_id: str, *, idempotency_key: str) -> None: ...
async def cancel(self, payment_intent_id: str, *, idempotency_key: str) -> None: ...
async def refund(self, payment_intent_id: str, amount_cents: int, *, idempotency_key: str) -> str: ...
class StripeGateway:
"""Thin async adapter; every mutating call carries a deterministic idempotency key."""
def __init__(self, settings: Settings) -> None:
self._c = stripe.StripeClient(
settings.stripe_secret_key.get_secret_value(),
http_client=stripe.HTTPXClient(),
max_network_retries=2,
)
async def authorize(self, *, booking_id: uuid.UUID, amount_cents: int, currency: str,
idempotency_key: str) -> tuple[str, str]:
pi = await self._c.payment_intents.create_async(
params={
"amount": amount_cents,
"currency": currency.lower(),
"capture_method": "manual", # capture only after server-side checks
"automatic_payment_methods": {"enabled": True},
"metadata": {"booking_id": str(booking_id)},
},
options={"idempotency_key": idempotency_key},
)
return pi.id, pi.client_secret
async def capture(self, payment_intent_id: str, *, idempotency_key: str) -> None:
await self._c.payment_intents.capture_async(payment_intent_id,
options={"idempotency_key": idempotency_key})
async def cancel(self, payment_intent_id: str, *, idempotency_key: str) -> None:
await self._c.payment_intents.cancel_async(payment_intent_id,
options={"idempotency_key": idempotency_key})
async def refund(self, payment_intent_id: str, amount_cents: int, *, idempotency_key: str) -> str:
r = await self._c.refunds.create_async(
params={"payment_intent": payment_intent_id, "amount": amount_cents},
options={"idempotency_key": idempotency_key},
)
return r.id
app/services/bookings.py: core business logic
from __future__ import annotations
import hashlib
import uuid
from datetime import datetime, timedelta
from sqlalchemy import exists, func, select, update
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.config import get_settings
from app.core.errors import Conflict, NotFound, UnprocessableEntity, APIError
from app.core.security import Principal
from app.models import Booking, BookingStatus, OutboxEvent, Role, TourInstance, TourStatus, Waiver, utcnow
from app.schemas import BookingCreate, PaymentIntentStartOut, WaiverSign
from app.services import audit
from app.services.audit import AuditContext
from app.services.payment_gateway import PaymentGateway
settings = get_settings()
HOLD = timedelta(minutes=settings.booking_hold_minutes)
PAYMENT_GRACE = timedelta(minutes=settings.payment_grace_minutes)
CUTOFF = timedelta(hours=settings.booking_cutoff_hours)
ALLOWED_TRANSITIONS: dict[BookingStatus, frozenset[BookingStatus]] = {
BookingStatus.pending: frozenset({BookingStatus.confirmed, BookingStatus.cancelled}),
BookingStatus.confirmed: frozenset({BookingStatus.cancelled, BookingStatus.completed, BookingStatus.refunded}),
BookingStatus.cancelled: frozenset({BookingStatus.refunded}),
BookingStatus.completed: frozenset({BookingStatus.refunded}),
BookingStatus.refunded: frozenset(),
}
def transition(booking: Booking, new: BookingStatus) -> None:
if new not in ALLOWED_TRANSITIONS[booking.status]:
raise Conflict(f"Cannot move booking from {booking.status} to {new}.", code="INVALID_STATE_TRANSITION")
booking.status = new
def _fingerprint(payload: BookingCreate) -> str:
return hashlib.sha256(payload.model_dump_json().encode()).hexdigest()
async def _by_idempotency_key(session: AsyncSession, customer_id: uuid.UUID, key: str) -> Booking | None:
return await session.scalar(
select(Booking).where(Booking.customer_id == customer_id, Booking.idempotency_key == key))
def _replay(existing: Booking, fp: str) -> Booking:
if existing.request_fingerprint != fp:
raise UnprocessableEntity("Idempotency-Key was already used with a different request body.",
code="IDEMPOTENCY_KEY_REUSED")
return existing
async def has_waiver(session: AsyncSession, booking_id: uuid.UUID) -> bool:
return bool(await session.scalar(select(exists().where(Waiver.booking_id == booking_id))))
async def create_booking(
session: AsyncSession, *, customer_id: uuid.UUID, payload: BookingCreate,
idempotency_key: str, ctx: AuditContext,
) -> tuple[Booking, bool]:
"""Returns (booking, created). Safe under concurrency and client retries."""
fp = _fingerprint(payload)
if (existing := await _by_idempotency_key(session, customer_id, idempotency_key)) is not None:
return _replay(existing, fp), False
# Atomic seat reservation: one conditional UPDATE. Concurrent requests serialize on the
# tour row lock; each re-evaluates the WHERE clause against the committed value, so
# overselling is impossible. The CHECK constraint ck_tour_capacity is a second line of defense.
reserve = (
update(TourInstance)
.where(
TourInstance.id == payload.tour_instance_id,
TourInstance.status == TourStatus.scheduled,
TourInstance.start_datetime > func.now() + CUTOFF,
TourInstance.seats_reserved + payload.num_participants <= TourInstance.max_participants,
)
.values(seats_reserved=TourInstance.seats_reserved + payload.num_participants, updated_at=func.now())
.returning(TourInstance.price_cents, TourInstance.currency)
.execution_options(synchronize_session=False)
)
row = (await session.execute(reserve)).one_or_none()
if row is None:
await session.rollback()
raise await _explain_unbookable(session, payload)
booking = Booking(
tour_instance_id=payload.tour_instance_id,
customer_id=customer_id,
num_participants=payload.num_participants,
status=BookingStatus.pending,
total_price_cents=row.price_cents * payload.num_participants, # server-computed price snapshot
currency=row.currency,
idempotency_key=idempotency_key,
request_fingerprint=fp,
hold_expires_at=utcnow() + HOLD,
)
session.add(booking)
try:
await session.flush()
except IntegrityError:
# Concurrent duplicate with the same Idempotency-Key. Rollback also undoes our seat increment.
await session.rollback()
if (existing := await _by_idempotency_key(session, customer_id, idempotency_key)) is not None:
return _replay(existing, fp), False
raise
audit.record(session, ctx, action="booking.created", entity_type="booking", entity_id=booking.id,
changes={"tour_instance_id": str(booking.tour_instance_id),
"num_participants": booking.num_participants,
"total_price_cents": booking.total_price_cents})
await session.commit()
return booking, True
async def _explain_unbookable(session: AsyncSession, payload: BookingCreate) -> APIError:
tour = await session.get(TourInstance, payload.tour_instance_id)
if tour is None:
return NotFound("Tour not found.", code="TOUR_NOT_FOUND")
if tour.status is not TourStatus.scheduled:
return Conflict(f"Tour is {tour.status}.", code="TOUR_NOT_BOOKABLE")
if tour.start_datetime <= utcnow() + CUTOFF:
return Conflict("Booking window for this tour has closed.", code="BOOKING_WINDOW_CLOSED")
remaining = max(0, tour.max_participants - tour.seats_reserved)
return Conflict(f"Only {remaining} spot(s) remaining.", code="INSUFFICIENT_CAPACITY",
extra={"spots_remaining": remaining})
async def get_visible_booking(session: AsyncSession, booking_id: uuid.UUID, principal: Principal) -> Booking:
row = (await session.execute(
select(Booking, TourInstance.guide_id)
.join(TourInstance, TourInstance.id == Booking.tour_instance_id)
.where(Booking.id == booking_id)
)).one_or_none()
if row is None:
raise NotFound("Booking not found.", code="BOOKING_NOT_FOUND")
booking, guide_id = row
visible = (
principal.is_admin
or (principal.role is Role.customer and booking.customer_id == principal.user_id)
or (principal.role is Role.guide and principal.guide_profile_id == guide_id)
)
if not visible:
# 404 rather than 403: don't reveal that the ID exists (BOLA/enumeration hardening).
raise NotFound("Booking not found.", code="BOOKING_NOT_FOUND")
return booking
def refund_fraction(start: datetime, now: datetime) -> float:
lead = start - now
if lead >= timedelta(days=7):
return 1.0
if lead >= timedelta(hours=48):
return 0.5
return 0.0
async def cancel_booking(
session: AsyncSession, *, booking_id: uuid.UUID, principal: Principal,
reason: str | None, ctx: AuditContext,
) -> tuple[Booking, int]:
booking = await session.scalar(select(Booking).where(Booking.id == booking_id).with_for_update())
if booking is None or not (principal.is_admin or booking.customer_id == principal.user_id):
raise NotFound("Booking not found.", code="BOOKING_NOT_FOUND")
if booking.status not in (BookingStatus.pending, BookingStatus.confirmed):
raise Conflict(f"Booking is {booking.status}.", code="INVALID_STATE_TRANSITION")
tour = await session.get(TourInstance, booking.tour_instance_id)
assert tour is not None
previous = booking.status
transition(booking, BookingStatus.cancelled)
booking.cancelled_at = utcnow()
booking.cancel_reason = (reason or "customer_request")[:500]
booking.hold_expires_at = None
await session.execute(
update(TourInstance)
.where(TourInstance.id == booking.tour_instance_id)
.values(seats_reserved=TourInstance.seats_reserved - booking.num_participants)
.execution_options(synchronize_session=False)
)
refund_cents = 0
if previous is BookingStatus.confirmed:
refund_cents = int(booking.total_price_cents * refund_fraction(tour.start_datetime, utcnow()))
if refund_cents > 0:
session.add(OutboxEvent(topic="payments.refund", payload={
"booking_id": str(booking.id), "amount_cents": refund_cents,
"idempotency_key": f"refund:{booking.id}:cancel",
}))
else:
# Pending: release any in-flight authorization.
session.add(OutboxEvent(topic="payments.void_authorization", payload={"booking_id": str(booking.id)}))
audit.record(session, ctx, action="booking.cancelled", entity_type="booking", entity_id=booking.id,
changes={"from": previous.value, "refund_cents": refund_cents, "reason": booking.cancel_reason})
await session.commit()
return booking, refund_cents
async def sign_waiver(
session: AsyncSession, *, booking_id: uuid.UUID, principal: Principal, payload: WaiverSign,
ip, user_agent: str | None, ctx: AuditContext,
) -> tuple[Waiver, bool]:
booking = await session.scalar(select(Booking).where(Booking.id == booking_id).with_for_update())
if booking is None or booking.customer_id != principal.user_id:
raise NotFound("Booking not found.", code="BOOKING_NOT_FOUND")
existing = await session.scalar(select(Waiver).where(Waiver.booking_id == booking_id))
if existing is not None:
return existing, False # immutable; idempotent PUT
if booking.status is not BookingStatus.pending:
raise Conflict("Waiver can only be signed for pending bookings.", code="INVALID_STATE_TRANSITION")
if payload.waiver_version != settings.current_waiver_version:
raise Conflict("Waiver text has changed; reload and sign the current version.",
code="WAIVER_VERSION_OUTDATED", extra={"current_version": settings.current_waiver_version})
if len(payload.participant_names) != booking.num_participants:
raise UnprocessableEntity(
"participant_names must list every participant.", code="PARTICIPANT_COUNT_MISMATCH",
errors=[{"in": "body", "field": "participant_names",
"message": f"expected {booking.num_participants} names"}])
waiver = Waiver(
booking_id=booking.id, waiver_version=payload.waiver_version,
signer_full_name=payload.signer_full_name, participant_names=payload.participant_names,
ip_address=ip, user_agent=(user_agent or "")[:512] or None,
signed_pdf_key=f"waivers/{booking.id}/{payload.waiver_version}.pdf", pdf_status="pending",
)
session.add(waiver)
session.add(OutboxEvent(topic="waiver.render_pdf", payload={"booking_id": str(booking.id)}))
audit.record(session, ctx, action="waiver.signed", entity_type="booking", entity_id=booking.id,
changes={"waiver_version": payload.waiver_version})
await session.commit()
return waiver, True
async def start_payment(
session: AsyncSession, *, booking_id: uuid.UUID, customer_id: uuid.UUID, gateway: PaymentGateway,
) -> PaymentIntentStartOut:
booking = await session.scalar(select(Booking).where(Booking.id == booking_id).with_for_update())
if booking is None or booking.customer_id != customer_id:
raise NotFound("Booking not found.", code="BOOKING_NOT_FOUND")
if booking.status is not BookingStatus.pending:
raise Conflict(f"Booking is {booking.status}.", code="BOOKING_NOT_PAYABLE")
now = utcnow()
if booking.hold_expires_at is None or booking.hold_expires_at <= now:
raise Conflict("Seat hold has expired; please book again.", code="HOLD_EXPIRED")
if not await has_waiver(session, booking.id):
raise Conflict("The waiver must be signed before payment.", code="WAIVER_REQUIRED")
booking.hold_expires_at = max(booking.hold_expires_at, now + PAYMENT_GRACE) # time for 3-D Secure
amount, currency = booking.total_price_cents, booking.currency
await session.commit() # release the row lock BEFORE the network call
# One PaymentIntent per booking (deterministic key). Retries return the same intent,
# and the client reuses its client_secret after a decline.
pi_id, client_secret = await gateway.authorize(
booking_id=booking_id, amount_cents=amount, currency=currency,
idempotency_key=f"authorize:{booking_id}",
)
return PaymentIntentStartOut(provider_payment_intent_id=pi_id, client_secret=client_secret,
amount_cents=amount, currency=currency)
app/services/payments.py: Stripe webhook handlers
import logging
import uuid
from collections.abc import Awaitable, Callable
from typing import Any
from sqlalchemy import func, select
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload
from app.models import Booking, BookingStatus, Payment, PaymentKind, PaymentStatus
from app.services import audit
from app.services.bookings import transition
from app.services.payment_gateway import PaymentGateway
log = logging.getLogger("trailbook.payments")
async def _upsert_charge(session: AsyncSession, booking_id: uuid.UUID, pi: Any, status: PaymentStatus) -> Payment:
stmt = (
pg_insert(Payment)
.values(id=uuid.uuid4(), booking_id=booking_id, kind=PaymentKind.charge, provider="stripe",
provider_charge_id=pi["id"], amount_cents=pi["amount"], currency=pi["currency"].upper(),
status=status)
.on_conflict_do_update(index_elements=[Payment.provider_charge_id],
set_={"status": status, "updated_at": func.now()})
.returning(Payment)
)
return (await session.scalars(stmt, execution_options={"populate_existing": True})).one()
async def on_amount_capturable_updated(session: AsyncSession, gateway: PaymentGateway, pi: Any) -> None:
"""Authorization succeeded. This is the final gate: capture only if the booking is still
pending, has a waiver, and the authorized amount matches the server-computed total.
The booking row lock is held across the Stripe call deliberately: if anything fails, the
whole transaction (including the event-dedupe row) rolls back and Stripe retries. The
capture itself is idempotent."""
booking_id = uuid.UUID(pi["metadata"]["booking_id"])
booking = await session.scalar(
select(Booking).where(Booking.id == booking_id)
.options(selectinload(Booking.waiver)).with_for_update(of=Booking))
if booking is None:
log.error("authorized_pi_for_unknown_booking", extra={"pi": pi["id"]})
await gateway.cancel(pi["id"], idempotency_key=f"cancel:{pi['id']}")
return
payment = await _upsert_charge(session, booking_id, pi, PaymentStatus.authorized)
confirmable = (
booking.status is BookingStatus.pending
and booking.waiver is not None
and pi["amount_capturable"] == booking.total_price_cents
and pi["currency"].upper() == booking.currency
)
if not confirmable:
await gateway.cancel(pi["id"], idempotency_key=f"cancel:{pi['id']}") # release the customer's funds
payment.status = PaymentStatus.failed
payment.failure_code = "booking_not_confirmable"
return
await gateway.capture(pi["id"], idempotency_key=f"capture:{pi['id']}")
payment.status = PaymentStatus.captured
transition(booking, BookingStatus.confirmed) # DB trigger also enforces waiver presence
booking.hold_expires_at = None
audit.record(session, audit.SYSTEM_STRIPE, action="booking.confirmed", entity_type="booking",
entity_id=booking.id, changes={"payment_intent": pi["id"]})
session.add_all([]) # outbox: confirmation email would be enqueued here
async def on_payment_failed(session: AsyncSession, gateway: PaymentGateway, pi: Any) -> None:
# The booking stays pending; the customer may retry until the hold expires.
booking_id = uuid.UUID(pi["metadata"]["booking_id"])
await _upsert_charge(session, booking_id, pi, PaymentStatus.failed)
WebhookHandler = Callable[[AsyncSession, PaymentGateway, Any], Awaitable[None]]
HANDLERS: dict[str, WebhookHandler] = {
"payment_intent.amount_capturable_updated": on_amount_capturable_updated,
"payment_intent.payment_failed": on_payment_failed,
# "charge.refunded", "charge.dispute.created" → mark refunds / flag for admin review
}
app/api/v1/tours.py
from datetime import datetime
from typing import Annotated
from uuid import UUID
from fastapi import APIRouter, Query, Response
from sqlalchemy import Select, func, select, tuple_
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import contains_eager
from app.core.db import ReadSessionDep, SessionDep
from app.core.errors import Conflict, NotFound, UnprocessableEntity, sqlstate
from app.core.pagination import decode_cursor, encode_cursor
from app.core.ratelimit import rate_limit
from app.core.security import ApprovedGuide
from app.models import GuideProfile, TourInstance, TourStatus, Trail
from app.schemas import Page, PageInfo, TourCreate, TourOut, TourSearchParams
from app.services import audit
from app.services.audit import AuditCtx
router = APIRouter(prefix="/tours", tags=["Tours"])
SORT_COLUMNS = {"start_datetime": TourInstance.start_datetime, "price_cents": TourInstance.price_cents}
SORT_PARSERS = {"start_datetime": datetime.fromisoformat, "price_cents": int}
def _tour_query() -> Select[tuple[TourInstance]]:
return (
select(TourInstance)
.join(TourInstance.trail)
.join(TourInstance.guide)
.join(GuideProfile.user)
.options(contains_eager(TourInstance.trail),
contains_eager(TourInstance.guide).contains_eager(GuideProfile.user))
)
@router.get("", response_model=Page[TourOut], summary="Search bookable tours",
dependencies=[rate_limit("tours:search", limit=120, per="ip")])
async def search_tours(
params: Annotated[TourSearchParams, Query()],
session: ReadSessionDep,
response: Response,
) -> Page[TourOut]:
field = params.sort.lstrip("-")
desc = params.sort.startswith("-")
col = SORT_COLUMNS[field]
spots = TourInstance.max_participants - TourInstance.seats_reserved
stmt = _tour_query().where(
TourInstance.status == TourStatus.scheduled,
TourInstance.start_datetime > func.now(),
spots >= params.min_spots,
Trail.deleted_at.is_(None),
GuideProfile.is_approved.is_(True),
)
if params.region:
stmt = stmt.where(func.lower(Trail.region) == params.region.lower())
if params.difficulty:
stmt = stmt.where(Trail.difficulty.in_(params.difficulty))
if params.trail_id:
stmt = stmt.where(TourInstance.trail_id == params.trail_id)
if params.guide_id:
stmt = stmt.where(TourInstance.guide_id == params.guide_id)
if params.language:
stmt = stmt.where(GuideProfile.languages_spoken.contains([params.language])) # GIN @>
if params.starts_after:
stmt = stmt.where(TourInstance.start_datetime >= params.starts_after)
if params.starts_before:
stmt = stmt.where(TourInstance.start_datetime < params.starts_before)
if params.currency:
stmt = stmt.where(TourInstance.currency == params.currency)
if params.min_price_cents is not None:
stmt = stmt.where(TourInstance.price_cents >= params.min_price_cents)
if params.max_price_cents is not None:
stmt = stmt.where(TourInstance.price_cents <= params.max_price_cents)
if params.cursor:
raw_value, last_id = decode_cursor(params.cursor, params.sort)
try:
value = SORT_PARSERS[field](raw_value)
except (TypeError, ValueError):
from app.core.errors import BadRequest
raise BadRequest("Invalid cursor.", code="INVALID_CURSOR") from None
key, bound = tuple_(col, TourInstance.id), tuple_(value, last_id)
stmt = stmt.where(key < bound if desc else key > bound) # keyset: O(log n) at any depth
order = (col.desc(), TourInstance.id.desc()) if desc else (col.asc(), TourInstance.id.asc())
rows = list((await session.scalars(stmt.order_by(*order).limit(params.limit + 1))).all())
has_more = len(rows) > params.limit
rows = rows[: params.limit]
next_cursor = encode_cursor(params.sort, getattr(rows[-1], field), rows[-1].id) if has_more else None
# Short public TTL: spots_remaining is advisory here; the booking UPDATE is authoritative.
response.headers["Cache-Control"] = "public, max-age=15, stale-while-revalidate=30"
return Page[TourOut](data=[TourOut.from_model(t) for t in rows],
page=PageInfo(limit=params.limit, has_more=has_more, next_cursor=next_cursor))
@router.get("/{tour_id}", response_model=TourOut)
async def get_tour(tour_id: UUID, session: SessionDep) -> TourOut:
tour = await session.scalar(_tour_query().where(TourInstance.id == tour_id))
if tour is None:
raise NotFound("Tour not found.", code="TOUR_NOT_FOUND")
return TourOut.from_model(tour)
@router.post("", status_code=201, response_model=TourOut,
dependencies=[rate_limit("tours:create", limit=30)])
async def create_tour(payload: TourCreate, session: SessionDep, guide: ApprovedGuide,
ctx: AuditCtx, response: Response) -> TourOut:
trail = await session.get(Trail, payload.trail_id)
if trail is None or trail.deleted_at is not None:
raise UnprocessableEntity("Trail does not exist.", code="TRAIL_NOT_FOUND",
errors=[{"in": "body", "field": "trail_id", "message": "unknown trail"}])
tour = TourInstance(
trail_id=payload.trail_id, guide_id=guide.guide_profile_id, # never from the request body
start_datetime=payload.start_datetime, end_datetime=payload.end_datetime, timezone=payload.timezone,
max_participants=payload.max_participants, price_cents=payload.price_cents, currency=payload.currency,
meeting_point_lat=payload.meeting_point.lat, meeting_point_lng=payload.meeting_point.lng,
)
session.add(tour)
try:
await session.flush()
except IntegrityError as exc:
await session.rollback()
if sqlstate(exc) == "23P01": # exclusion_violation: ex_tour_guide_overlap
raise Conflict("You already lead a tour overlapping this time.",
code="GUIDE_SCHEDULE_CONFLICT") from exc
raise
tour_id = tour.id
audit.record(session, ctx, action="tour.created", entity_type="tour", entity_id=tour_id,
changes=payload.model_dump(mode="json"))
await session.commit()
created = await session.scalar(_tour_query().where(TourInstance.id == tour_id)
.execution_options(populate_existing=True))
response.headers["Location"] = f"/api/v1/tours/{tour_id}"
return TourOut.from_model(created)
app/api/v1/bookings.py
from typing import Annotated
from uuid import UUID
from fastapi import APIRouter, Header, Request, Response
from app.core.db import SessionDep
from app.core.errors import Forbidden
from app.core.ratelimit import rate_limit
from app.core.security import CurrentPrincipal, CustomerOrAdmin, CustomerPrincipal
from app.schemas import (BookingCancelIn, BookingCancellationOut, BookingCreate, BookingOut,
PaymentIntentStartOut, WaiverOut, WaiverSign)
from app.services import bookings as svc
from app.services.audit import AuditCtx
router = APIRouter(prefix="/bookings", tags=["Bookings"])
IdempotencyKey = Annotated[str, Header(alias="Idempotency-Key", min_length=8, max_length=64,
pattern=r"^[A-Za-z0-9_\-]+$")]
@router.post("", status_code=201, response_model=BookingOut, summary="Hold seats on a tour",
dependencies=[rate_limit("bookings:create", limit=10)])
async def create_booking(
payload: BookingCreate, response: Response, session: SessionDep,
principal: CustomerPrincipal, ctx: AuditCtx, idempotency_key: IdempotencyKey,
) -> BookingOut:
if not principal.is_verified:
raise Forbidden("Verify your email address before booking.", code="EMAIL_NOT_VERIFIED")
booking, created = await svc.create_booking(
session, customer_id=principal.user_id, payload=payload,
idempotency_key=idempotency_key, ctx=ctx,
)
response.headers["Location"] = f"/api/v1/bookings/{booking.id}"
if not created:
response.headers["Idempotent-Replayed"] = "true"
waiver_signed = False if created else await svc.has_waiver(session, booking.id)
return BookingOut.from_booking(booking, waiver_signed=waiver_signed)
@router.get("/{booking_id}", response_model=BookingOut)
async def get_booking(booking_id: UUID, session: SessionDep, principal: CurrentPrincipal) -> BookingOut:
booking = await svc.get_visible_booking(session, booking_id, principal)
return BookingOut.from_booking(booking, waiver_signed=await svc.has_waiver(session, booking.id))
@router.post("/{booking_id}/cancel", response_model=BookingCancellationOut,
dependencies=[rate_limit("bookings:cancel", limit=10)])
async def cancel_booking(
booking_id: UUID, session: SessionDep, principal: CustomerOrAdmin, ctx: AuditCtx,
body: BookingCancelIn | None = None,
) -> BookingCancellationOut:
booking, refund = await svc.cancel_booking(
session, booking_id=booking_id, principal=principal,
reason=body.reason if body else None, ctx=ctx,
)
return BookingCancellationOut(
booking=BookingOut.from_booking(booking, waiver_signed=await svc.has_waiver(session, booking.id)),
refund_amount_cents=refund,
)
@router.put("/{booking_id}/waiver", response_model=WaiverOut,
responses={200: {"description": "Already signed"}, 201: {"description": "Signed"}},
dependencies=[rate_limit("waivers:sign", limit=10)])
async def sign_waiver(
booking_id: UUID, payload: WaiverSign, request: Request, response: Response,
session: SessionDep, principal: CustomerPrincipal, ctx: AuditCtx,
) -> WaiverOut:
waiver, created = await svc.sign_waiver(
session, booking_id=booking_id, principal=principal, payload=payload,
ip=ctx.ip, user_agent=request.headers.get("user-agent"), ctx=ctx,
)
response.status_code = 201 if created else 200
return WaiverOut.from_waiver(waiver)
@router.post("/{booking_id}/payments", status_code=201, response_model=PaymentIntentStartOut,
dependencies=[rate_limit("payments:create", limit=10)])
async def start_payment(
booking_id: UUID, request: Request, session: SessionDep,
principal: CustomerPrincipal, _idempotency_key: IdempotencyKey,
) -> PaymentIntentStartOut:
# Stripe-side idempotency is derived from booking_id; the header keeps the contract uniform.
return await svc.start_payment(session, booking_id=booking_id, customer_id=principal.user_id,
gateway=request.app.state.payments)
app/api/v1/webhooks.py
import stripe
from fastapi import APIRouter, Request
from sqlalchemy.dialects.postgresql import insert as pg_insert
from app.core.config import get_settings
from app.core.db import SessionDep
from app.core.errors import BadRequest
from app.models import StripeEvent
from app.services.payments import HANDLERS
router = APIRouter(prefix="/webhooks", tags=["Webhooks"])
@router.post("/stripe")
async def stripe_webhook(request: Request, session: SessionDep) -> dict[str, str]:
payload = await request.body() # raw bytes: required for signature verification
signature = request.headers.get("stripe-signature")
if not signature:
raise BadRequest("Missing Stripe-Signature header.", code="MISSING_SIGNATURE")
try:
event = stripe.Webhook.construct_event(
payload, signature, get_settings().stripe_webhook_secret.get_secret_value(), tolerance=300)
except (ValueError, stripe.SignatureVerificationError) as exc:
raise BadRequest("Invalid webhook signature.", code="INVALID_SIGNATURE") from exc
# Exactly-once processing: the dedupe row commits in the same transaction as the handler's effects.
first_seen = await session.scalar(
pg_insert(StripeEvent).values(id=event.id, type=event.type)
.on_conflict_do_nothing(index_elements=[StripeEvent.id]).returning(StripeEvent.id))
if first_seen is None:
return {"status": "duplicate"}
if (handler := HANDLERS.get(event.type)) is not None:
await handler(session, request.app.state.payments, event.data.object)
await session.commit()
return {"status": "processed"}
app/main.py
import logging
import re
import time
import uuid
from contextlib import asynccontextmanager
import httpx
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware
from fastapi.responses import JSONResponse
from redis.asyncio import Redis
from sqlalchemy import text
from app.api.v1 import bookings, tours, webhooks
from app.core.config import get_settings
from app.core.db import engine, read_engine
from app.core.errors import register_exception_handlers
from app.core.ratelimit import RateLimiter
from app.core.security import TokenVerifier
from app.services.payment_gateway import StripeGateway
settings = get_settings()
log = logging.getLogger("trailbook.access")
_REQUEST_ID_RE = re.compile(r"^[A-Za-z0-9\-]{8,64}$")
@asynccontextmanager
async def lifespan(app: FastAPI):
app.state.http = httpx.AsyncClient(timeout=httpx.Timeout(5.0))
app.state.redis = Redis.from_url(str(settings.redis_url), decode_responses=True,
socket_timeout=0.25, socket_connect_timeout=0.5, health_check_interval=30)
app.state.token_verifier = TokenVerifier(settings, app.state.http)
app.state.rate_limiter = RateLimiter(app.state.redis)
app.state.payments = StripeGateway(settings)
yield
await app.state.http.aclose()
await app.state.redis.aclose()
await engine.dispose()
if read_engine is not engine:
await read_engine.dispose()
app = FastAPI(
title="TrailBook API",
version="1.0.0",
lifespan=lifespan,
openapi_url="/api/v1/openapi.json" if settings.env != "production" else None,
docs_url="/api/v1/docs" if settings.env != "production" else None,
redoc_url=None,
)
register_exception_handlers(app)
app.add_middleware(GZipMiddleware, minimum_size=1024)
app.add_middleware(
CORSMiddleware,
allow_origins=settings.cors_origins, # explicit allowlist; never "*"
allow_credentials=False, # bearer tokens, not cookies
allow_methods=["GET", "POST", "PATCH", "PUT", "DELETE"],
allow_headers=["Authorization", "Content-Type", "Idempotency-Key", "X-Request-ID"],
expose_headers=["Location", "X-Request-ID", "Idempotent-Replayed", "Retry-After",
"RateLimit-Limit", "RateLimit-Remaining", "RateLimit-Reset"],
max_age=600,
)
@app.middleware("http")
async def request_context(request: Request, call_next):
incoming = request.headers.get("x-request-id", "")
request_id = incoming if _REQUEST_ID_RE.match(incoming) else uuid.uuid4().hex
request.state.request_id = request_id
start = time.perf_counter()
response = await call_next(request)
response.headers["X-Request-ID"] = request_id
response.headers["X-Content-Type-Options"] = "nosniff"
response.headers["Strict-Transport-Security"] = "max-age=63072000; includeSubDomains"
log.info("request", extra={
"request_id": request_id, "method": request.method, "path": request.url.path,
"status": response.status_code, "duration_ms": round((time.perf_counter() - start) * 1000, 1),
"subject": getattr(request.state, "subject", None), # Auth0 sub, not email/PII
})
return response
API_PREFIX = "/api/v1"
app.include_router(tours.router, prefix=API_PREFIX)
app.include_router(bookings.router, prefix=API_PREFIX)
app.include_router(webhooks.router, prefix=API_PREFIX)
# users, guide_profiles, trails, reviews, admin routers follow the same patterns.
@app.get("/healthz", include_in_schema=False)
async def healthz() -> dict[str, str]:
return {"status": "ok"} # liveness: no dependency checks (avoid restart storms)
@app.get("/readyz", include_in_schema=False)
async def readyz(request: Request):
try:
async with engine.connect() as conn:
await conn.execute(text("SELECT 1"))
except Exception:
return JSONResponse({"status": "db_unavailable"}, status_code=503)
return {"status": "ready"} # Redis deliberately excluded: the app degrades without it
Run with:
uvicorn app.main:app --workers 2 --proxy-headers --forwarded-allow-ips="10.0.0.0/16" \
--timeout-graceful-shutdown 20 --no-server-header --h11-max-incomplete-event-size 65536
Also cap request bodies at the ALB/WAF, for example at 64 KB for JSON routes.
Alembic: async env.py (essentials) and key migration
# alembic/env.py
import asyncio
from logging.config import fileConfig
from alembic import context
from sqlalchemy.ext.asyncio import create_async_engine
from app.core.config import get_settings
from app.models import Base
config = context.config
if config.config_file_name:
fileConfig(config.config_file_name)
target_metadata = Base.metadata
def do_run_migrations(connection) -> None:
context.configure(connection=connection, target_metadata=target_metadata, compare_type=True)
with context.begin_transaction():
# Never block production traffic behind a migration lock.
context.execute("SET lock_timeout = '3s'")
context.execute("SET statement_timeout = '60s'")
context.run_migrations()
async def run_async_migrations() -> None:
engine = create_async_engine(str(get_settings().database_url)) # migration role, not app role
async with engine.connect() as conn:
await conn.run_sync(do_run_migrations)
await engine.dispose()
asyncio.run(run_async_migrations())
# alembic/versions/0003_tours_bookings_invariants.py (abridged: shows the non-obvious DB invariants)
"""tour/booking invariants: guide overlap exclusion, waiver-before-confirm trigger, audit hardening"""
from alembic import op
revision = "0003_tours_bookings_invariants"
down_revision = "0002_tours_bookings"
def upgrade() -> None:
op.execute("CREATE EXTENSION IF NOT EXISTS btree_gist")
# A guide can't lead two overlapping scheduled tours.
op.execute("""
ALTER TABLE tour_instances
ADD CONSTRAINT ex_tour_guide_overlap
EXCLUDE USING gist (guide_id WITH =, tstzrange(start_datetime, end_datetime, '[)') WITH &&)
WHERE (status = 'scheduled')
""")
# Defense in depth: a booking can never become confirmed without a waiver.
op.execute("""
CREATE FUNCTION enforce_waiver_before_confirm() RETURNS trigger
LANGUAGE plpgsql AS $$
BEGIN
IF NEW.status = 'confirmed' AND OLD.status <> 'confirmed'
AND NOT EXISTS (SELECT 1 FROM waivers w WHERE w.booking_id = NEW.id) THEN
RAISE EXCEPTION 'booking % cannot be confirmed without a signed waiver', NEW.id
USING ERRCODE = 'check_violation';
END IF;
RETURN NEW;
END $$;
""")
op.execute("""
CREATE TRIGGER trg_bookings_waiver_required
BEFORE UPDATE OF status ON bookings
FOR EACH ROW EXECUTE FUNCTION enforce_waiver_before_confirm()
""")
# Audit log is append-only for the application role.
op.execute("REVOKE UPDATE, DELETE, TRUNCATE ON audit_logs FROM trailbook_app")
op.execute("GRANT INSERT, SELECT ON audit_logs TO trailbook_app")
# Bookings can't be inserted as confirmed.
op.execute("ALTER TABLE bookings ADD CONSTRAINT ck_booking_insert_state "
"CHECK (status <> 'confirmed' OR created_at < updated_at OR true) NOT VALID") # placeholder; enforced in app
def downgrade() -> None:
op.execute("DROP TRIGGER IF EXISTS trg_bookings_waiver_required ON bookings")
op.execute("DROP FUNCTION IF EXISTS enforce_waiver_before_confirm()")
op.execute("ALTER TABLE tour_instances DROP CONSTRAINT IF EXISTS ex_tour_guide_overlap")
op.execute("ALTER TABLE bookings DROP CONSTRAINT IF EXISTS ck_booking_insert_state")
Correction: delete the
ck_booking_insert_stateplaceholder before using this migration. It is a no-op. Bookings are only ever inserted aspendingbycreate_booking, and the trigger above guards every later transition.
Migrations follow an expand → migrate → contract pattern. Every migration must be backward-compatible with the currently deployed version, so blue/green deploys stay safe. CREATE INDEX CONCURRENTLY runs in a separate autocommit_block().
Hold-expiry sweeper (jobs/expire_holds.sql, run every minute by the worker)
WITH expired AS (
UPDATE bookings b
SET status = 'cancelled', cancel_reason = 'hold_expired',
cancelled_at = now(), updated_at = now(), hold_expires_at = NULL
WHERE b.id IN (
SELECT id FROM bookings
WHERE status = 'pending' AND hold_expires_at < now()
AND NOT EXISTS (SELECT 1 FROM payments p -- capture in flight: leave it alone
WHERE p.booking_id = bookings.id AND p.status = 'authorized')
ORDER BY hold_expires_at
LIMIT 500
FOR UPDATE SKIP LOCKED) -- safe with multiple workers
AND b.status = 'pending'
RETURNING b.tour_instance_id, b.num_participants
), per_tour AS (
SELECT tour_instance_id, sum(num_participants) AS n FROM expired GROUP BY tour_instance_id
)
UPDATE tour_instances t
SET seats_reserved = t.seats_reserved - per_tour.n, updated_at = now()
FROM per_tour
WHERE t.id = per_tour.tour_instance_id;
A nightly reconciliation job does two things:
- Asserts that
seats_reserved = SUM(num_participants) WHERE status IN ('pending','confirmed')for each future tour, and alerts on any drift. - Reconciles
paymentsagainst Stripe balance transactions.
Must-have test: fire 50 concurrent POST /bookings requests at a 10-seat tour. Assert exactly 10 seats are sold, seats_reserved = 10, and every other request gets 409 INSUFFICIENT_CAPACITY.
5. Scalability Notes
Sizing reality check. 180 req/s at peak is modest. With ~10–20 ms of DB time per request, you need about 2–4 concurrent DB connections on average. The risks are tail latency, failover, and bursty contention on popular tours, not raw throughput.
| Area | Recommendation |
|---|---|
| Compute | ECS Fargate across 2 AZs, 2 uvicorn workers per task (1 vCPU / 2 GB). Minimum 2 tasks off-season (needed for 99.9%); scheduled scale-up to 6 before June 1; target-tracking on CPU 50% and ALB p95 latency, max 10. Load-test to 3× peak (~550 req/s) every May. |
| Connection pooling | Per worker: pool_size=5, max_overflow=5, so at most 10 per worker. At 10 tasks × 2 workers that is ≤200, well under max_connections for a db.r6g.large (≈1,700). pool_timeout=5s fails fast. pool_pre_ping plus pool_recycle=1800 covers Multi-AZ failover (~60–120 s). Skip RDS Proxy initially. If you add it for faster failover, pass prepared_statement_cache_size=0 to avoid asyncpg pinning. |
| Database | Start with db.r6g.large Multi-AZ on gp3. Add one read replica for June–September and route only GET /tours, /trails, and /reviews to it. Booking paths always use the primary. Enable Performance Insights and pg_stat_statements. Tune autovacuum_vacuum_scale_factor=0.05 on bookings and tour_instances (hot updates of seats_reserved); consider fillfactor=85 on tour_instances for HOT updates. |
| Indexes | See the table below. |
| Caching | Four layers. (1) CloudFront for anonymous GET /tours, /trails/{id}, /guide-profiles/{id}, /reviews with 15–60 s TTL and stale-while-revalidate; the cache key includes the normalized query string. (2) Redis caches trail and guide profile payloads (5 min, invalidated on write) and aggregate review stats. (3) In-process caching of the JWKS (10 min), so Auth0 is off the hot path. (4) Never cache capacity for decisions: the atomic UPDATE is the arbiter, and spots_remaining in search is advisory. |
| Hot-tour contention | A popular Dolomites trek opening for sale serializes on one row lock. Each UPDATE holds it for about a millisecond, so ~500 bookings/s per tour is fine. Keep that transaction short: no network calls while holding it. The payment flow already commits before calling Stripe. |
| Static assets | GPX tracks and waiver PDFs go through S3 and CloudFront with presigned URLs. The API never proxies file bytes. |
| Async work | Use the outbox with an SQS worker for refunds, PDF rendering, emails, and tour cancellation fan-out. The API stays fast when Stripe or SES is slow. |
| 99.9% SLA (≈43 min/month) | Multi-AZ everything (RDS, ElastiCache with a replica, ≥2 tasks). ALB health checks on /readyz. Blue/green deploys via CodeDeploy with automatic rollback on 5xx or latency alarms. PITR backups with 14-day retention, plus quarterly restore drills. Degradation rules: if Redis is down, rate limiting fails open and caching is bypassed; if Stripe is down, bookings still hold seats and payment returns 503 with a retry. |
| Observability | OpenTelemetry traces (API, DB, Stripe). JSON logs with request_id. SLOs: availability 99.9%, p95 < 300 ms for search and < 500 ms for booking creation. Alerts on DB connection saturation > 70%, webhook failures, hold-sweeper lag, and reconciliation drift. |
| Off-season | Scale to 2 tasks, drop the read replica, and consider downsizing RDS one class in November. This roughly halves infra cost with no code changes. |
Index plan (beyond PK and unique constraints)
| Table | Index | Serves |
|---|---|---|
tour_instances |
(start_datetime, id) WHERE status='scheduled' |
Default search sort plus keyset |
tour_instances |
(price_cents, id) WHERE status='scheduled' |
Price sort |
tour_instances |
(trail_id, start_datetime), (guide_id, start_datetime) |
Trail and guide pages, guide dashboard |
tour_instances |
GiST exclusion (guide_id, tstzrange) |
Overlap prevention |
trails |
(lower(region), difficulty) WHERE deleted_at IS NULL |
Region and difficulty filters |
guide_profiles |
GIN (languages_spoken) |
language= filter |
bookings |
(customer_id, created_at, id) |
"My bookings" |
bookings |
(tour_instance_id, status) |
Manifest, cancellation fan-out |
bookings |
(hold_expires_at) WHERE status='pending' |
Sweeper |
bookings |
UNIQUE (customer_id, idempotency_key) |
Idempotency |
payments |
UNIQUE (provider_charge_id), (booking_id, created_at) |
Webhook upsert, ledger |
reviews |
UNIQUE (tour_instance_id, customer_id), (tour_instance_id, created_at, id) |
One review rule, listing |
audit_logs |
(entity_type, entity_id, occurred_at) |
Investigations. Partition monthly after it reaches ~50M rows. |
6. Security & Performance Checklist
Authentication & authorization
- RS256 only.
algis pinned, which preventsnoneand HS256 key-confusion attacks.iss,aud,exp, andiatare required, with 30 s leeway. - JWKS is cached for 10 min. Forced refresh on an unknown
kidis throttled to once per 30 s, and the last-known keys are kept if Auth0 is unreachable. - 15-minute access tokens. Refresh tokens stay with Auth0 (rotation, reuse detection, absolute lifetime). SPAs use Auth Code + PKCE and keep access tokens in memory, not in localStorage.
- The DB is the role source of truth. Suspension and guide revocation take effect on the next request, not the next token.
- RBAC runs through typed dependencies (
CustomerPrincipal,ApprovedGuide,AdminPrincipal). Object-level checks run in services, and callers get 404 on resources they don't own (BOLA/IDOR protection). - Ownership fields (
guide_id,customer_id,total_price_cents,status,is_approved,rating_avg) are never accepted from clients.extra="forbid"rejects them. - Booking requires a verified email. Admin accounts require MFA, enforced by an Auth0 rule.
Input, errors, idempotency
- Pydantic v2 strict models: bounded lengths and ranges, ISO-4217 and BCP-47 patterns, IANA timezone validation, aware datetimes only.
- Unknown query parameters are rejected. Sort fields come from an allowlist. Cursors are opaque and validated against the sort.
- RFC 9457 errors with a stable
codeandrequest_id. No stack traces, SQL, or echoed input (PII) in responses. -
Idempotency-Keyis required on booking creation, payment start, and refunds. A fingerprint mismatch returns 422. Concurrent duplicates are resolved by a unique constraint with rollback. - Every Stripe mutation uses a deterministic idempotency key. Webhooks are deduplicated by event ID in the same transaction as their effects.
Rate limiting & abuse
-
AWS WAF at the edge: managed rule groups, IP rate rules (e.g. 2,000/5 min), bot control, body size limits.
-
App-level Redis limits:
Scope Limit Search 120/min per IP (IPv6 bucketed by /64) Booking creation 10/min per user Payment start 10/min per user Waiver signing 10/min per user Reviews 5/hour per user Standard
RateLimit-*headers are returned, plusRetry-Afteron 429. -
Seat-hold abuse is bounded by the 15-minute hold, the per-user booking limit, and verified-email accounts. Consider a cap on concurrent pending holds per user (e.g. 3).
Payments & data integrity
- Manual capture. Capture happens only after server-side checks: still pending, waiver present, amount and currency match the server total. Otherwise the authorization is voided.
- Webhook signatures are verified on raw bytes with a 300 s tolerance. The endpoint is excluded from JWT auth and CORS.
- DB-enforced invariants: capacity
CHECK, waiver-before-confirm trigger, guide overlap exclusion, one review per customer per tour, positive amounts. - Transactional outbox for refunds and emails, so there are no lost side effects and no double refunds.
- Nightly reconciliation of seat counts and Stripe settlements.
Secrets, transport, CORS
- Secrets (DB credentials, Stripe keys, webhook secret) live in AWS Secrets Manager and are injected via ECS task definitions. Rotation is automated; RDS uses Secrets Manager rotation or IAM auth. Nothing sits in images, repos, or logs.
SecretStrprevents accidental printing. - Separate DB roles:
trailbook_migrator(DDL) andtrailbook_app(DML only, no DDL, no UPDATE/DELETE onaudit_logs). - TLS 1.2+ at the ALB and CloudFront, with HSTS. RDS enforces TLS (
rds.force_ssl=1). Encryption at rest with KMS for RDS, S3, and ElastiCache. - CORS uses an explicit origin allowlist per environment and
allow_credentials=False, since bearer tokens make cookies unnecessary.Idempotency-KeyandX-Request-IDare allowed. Rate-limit and replay headers are exposed. - OpenAPI docs are disabled in production, or served behind auth.
Privacy & audit
- Append-only
audit_logswritten in the same transaction as the change. Covered actions include booking and tour changes, waiver signing, guide approval, role and suspension changes, refunds, and admin reads of PII. Rows are streamed to S3 with Object Lock (WORM) for ≥7-year retention, to support waiver and liability disputes. - Waiver PDFs live in a private S3 bucket (SSE-KMS) with a SHA-256 digest stored in the DB and 5-minute presigned download URLs. Signer IP and user agent are captured server-side from the trusted proxy chain.
- Manifests expose minimal PII: the lead customer's phone is shown to the guide only within 72 h of the start.
- GDPR/CCPA erasure anonymizes users while retaining financial, waiver, and audit records under legal hold.
- Logs carry the Auth0
subandrequest_idonly. Emails, phone numbers, client secrets, and card data never appear; Stripe Elements keeps card data out of PCI scope (SAQ-A).
Performance & reliability
- Keyset pagination only, with composite and partial indexes matching every sort.
- Short transactions. No network calls while holding locks, except the deliberate webhook capture path.
-
statement_timeout=5s,idle_in_transaction_session_timeout=10s, andpool_timeout=5sfail fast instead of cascading. - The read replica serves lag-tolerant public reads only. Availability decisions always hit the primary.
- CDN and Redis caching for public catalog data. Capacity is never cached for decisions.
- Scheduled seasonal scaling, a pre-season 3× load test, blue/green deploys with auto-rollback, and lock-safe migrations (
lock_timeout=3s, expand/contract). - Liveness and readiness are separate. Redis is excluded from readiness, so its failure degrades the service instead of failing it.
Open items to confirm: A1 (the completed booking status), A9 (per-participant waivers for legal compliance), and A10 (refund policy tiers). None of these change the API shape materially; they are business rules isolated in services/bookings.py.
Related prompts
Programming & Development
Algorithm Design Expert
Designs, analyzes, and optimizes algorithms with multiple approaches, Big-O complexity analysis, production code, and tests.
Programming & Development
Advanced Debugging Assistant
Diagnoses bugs in your code, pinpoints root causes, and delivers fixed code with prevention tips.
Programming & Development
Code Review Expert
Performs a rigorous senior-level code review covering correctness, design, performance, and security with prioritized, actionable fixes.
Programming & Development
SQL Query Optimizer
Diagnoses a slow SQL query and returns an optimized rewrite, index plan, and execution-plan reasoning.