import asyncio
import base64
from collections import deque
from contextlib import asynccontextmanager
import json
import os
import random
import re
import secrets
import time
import uuid
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any
import resend
from fastapi import FastAPI, HTTPException, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.trustedhost import TrustedHostMiddleware
from fastapi.responses import JSONResponse, Response
from starlette.types import ASGIApp, Receive, Scope, Send
import db as admin_db
from mail_api import myob
from mail_api.config import (
ALLOWED_EMAILS_FILE as _ALLOWED_EMAILS_FILE,
APP_VERSION,
AUTH_CODE_MAX_ATTEMPTS,
AUTH_CODE_REQUESTS_PER_HOUR,
AUTH_CODE_TTL_SECONDS,
AUTH_IP_BLOCK_DURATION,
AUTH_IP_FAILURE_WINDOW,
AUTH_IP_MAX_FAILURES,
AUTH_SESSION_TTL_SECONDS,
BIRTHDAY_CHECK_INTERVAL_SECONDS,
CLIENT_BCC,
CLIENT_PROFILES_FILE as _CLIENT_PROFILES_FILE,
CORS_ALLOWED_ORIGINS,
CP_ADMIN_EMAILS,
DEPLOY_SMOKE_SECRET,
DEV_MODE,
DRAFTS_FILE as _DRAFTS_FILE,
EMAIL_SEND_TIMEOUT_SECONDS,
ENABLE_GENERAL_ENQUIRIES,
FORM_MAX_SECONDS,
FORM_MIN_SECONDS,
FROM_EMAIL,
LEGACY_SEED_FILE as _LEGACY_SEED_FILE,
LOGO_URL,
MAX_REQUEST_BODY_BYTES,
MAX_SEND_ATTEMPTS,
OWNER_BCC,
OWNER_EMAIL,
RATE_LIMIT_MAX_PER_EMAIL,
RATE_LIMIT_MAX_PER_IP,
RATE_LIMIT_MIN_INTERVAL_SECONDS,
RATE_LIMIT_WINDOW_SECONDS,
REPLY_TO,
SCHEDULED_CHECK_INTERVAL_SECONDS,
SCHEDULED_EMAILS_FILE as _SCHEDULED_EMAILS_FILE,
STARTUP_TEST_RECIPIENT,
TRUSTED_HOSTS,
logger,
)
from mail_api.models import (
BaseSubmission,
BirthdayAutoSendRequest,
BirthdayEmailRequest,
ClientProfileUpdate,
BookingSubmission,
ClientStatusUpdate,
ContractSubmission,
NewClientRequest,
OnboardingSubmission,
RenderMessageRequest,
ResetOnboardingRequest,
ScheduledEmailCancelRequest,
ScheduledEmailRescheduleRequest,
SendMessageRequest,
WelcomePackEmailRequest,
)
@asynccontextmanager
async def _lifespan(app: FastAPI):
await _startup_mail_check()
try:
yield
finally:
await _shutdown_background_tasks()
app = FastAPI(title="GoodWalk Mail API", lifespan=_lifespan)
# ── Auth state ───────────────────────────────────────────────────────────────
def _write_pii_json(path: Path, payload: object) -> None:
"""Atomically write a JSON file and chmod it owner-only (0600).
The chmod is best-effort: it is a no-op on Windows, but on the Linux
Docker host it ensures the file with PII is unreadable by other users.
"""
path.parent.mkdir(parents=True, exist_ok=True)
tmp = path.with_suffix(path.suffix + ".tmp")
tmp.write_text(json.dumps(payload, indent=2), encoding="utf-8")
try:
os.chmod(tmp, 0o600)
except OSError:
pass
os.replace(tmp, path)
def _load_allowed_emails_from_file() -> set[str]:
seed = {e.strip().lower() for e in os.environ.get("ALLOWED_EMAILS", "").split(",") if e.strip()}
try:
if _ALLOWED_EMAILS_FILE.exists():
data = json.loads(_ALLOWED_EMAILS_FILE.read_text(encoding="utf-8"))
seed.update(e.lower() for e in data.get("emails", []) if isinstance(e, str))
except Exception as exc:
logger.warning("Could not load allowed_emails file: %s", exc)
return seed
def _save_allowed_emails_file(emails: set[str]) -> None:
try:
_write_pii_json(_ALLOWED_EMAILS_FILE, {"emails": sorted(emails)})
except Exception as exc:
logger.warning("Could not save allowed_emails file: %s", exc)
def _load_client_profiles_from_file() -> dict[str, dict]:
try:
if _CLIENT_PROFILES_FILE.exists():
return json.loads(_CLIENT_PROFILES_FILE.read_text(encoding="utf-8"))
except Exception as exc:
logger.warning("Could not load client_profiles file: %s", exc)
return {}
def _save_client_profiles_file(profiles: dict) -> None:
try:
_write_pii_json(_CLIENT_PROFILES_FILE, profiles)
except Exception as exc:
logger.warning("Could not save client_profiles file: %s", exc)
def _load_drafts_from_file() -> dict:
try:
if _DRAFTS_FILE.exists():
return json.loads(_DRAFTS_FILE.read_text(encoding="utf-8"))
except Exception as exc:
logger.warning("Could not load drafts file: %s", exc)
return {}
def _save_drafts_file(drafts: dict) -> None:
try:
_write_pii_json(_DRAFTS_FILE, drafts)
except Exception as exc:
logger.warning("Could not save drafts file: %s", exc)
def _load_scheduled_emails_from_file() -> dict[str, dict]:
try:
if _SCHEDULED_EMAILS_FILE.exists():
data = json.loads(_SCHEDULED_EMAILS_FILE.read_text(encoding="utf-8"))
if isinstance(data, dict):
return data
except Exception as exc:
logger.warning("Could not load scheduled_emails file: %s", exc)
return {}
def _save_scheduled_emails_file(scheduled: dict) -> None:
try:
_write_pii_json(_SCHEDULED_EMAILS_FILE, scheduled)
except Exception as exc:
logger.warning("Could not save scheduled_emails file: %s", exc)
async def _save_active_sessions_async() -> None:
"""Persist live sessions to admin_kv so they survive container restarts.
Snapshot filters out expired entries before writing. Best-effort —
failure is logged but does not block the auth flow (memory remains
authoritative for the current process).
"""
now = time.time()
snapshot = {tok: s for tok, s in _active_sessions.items() if s.get("expires_at", 0) > now}
try:
await admin_db.set_kv("active_sessions", snapshot)
except Exception as exc:
logger.warning("Could not persist active_sessions: %s", exc)
async def _load_active_sessions_async() -> dict[str, dict]:
if not admin_db.is_enabled():
return {}
try:
data = await admin_db.get_kv("active_sessions")
if not isinstance(data, dict):
return {}
now = time.time()
return {
tok: s
for tok, s in data.items()
if isinstance(s, dict) and isinstance(s.get("expires_at"), (int, float)) and s["expires_at"] > now
}
except Exception as exc:
logger.warning("Could not load active_sessions from admin_kv: %s", exc)
return {}
async def _persist_admin_state(key: str, value: Any) -> None:
"""Write a single admin_kv blob to postgres when the database is available."""
try:
await admin_db.set_kv(key, value)
except Exception as exc:
logger.warning("Postgres persist (%s) failed; JSON copy is still authoritative: %s", key, exc)
async def _seed_admin_state_from_json_if_needed() -> None:
"""Seed admin_kv from the JSON files on disk.
Controlled by ADMIN_DATA_SEED_FROM_JSON:
- "never": do nothing
- "auto": seed only when admin_kv has no rows yet (default, safe on every boot)
- "force": overwrite postgres with whatever the JSON files currently hold
The deployer exposes -SeedAdminData which sets this to "force" for one boot.
"""
mode = (os.environ.get("ADMIN_DATA_SEED_FROM_JSON", "auto") or "auto").strip().lower()
if mode == "never":
return
if not admin_db.is_enabled():
return
try:
if mode == "auto" and await admin_db.has_any_value():
return
seed_clients = _load_client_profiles_from_file()
seed_emails = sorted(_load_allowed_emails_from_file())
seed_drafts = _load_drafts_from_file()
if not seed_clients and not seed_emails and not seed_drafts:
return
if seed_clients:
await admin_db.set_kv("client_profiles", seed_clients)
if seed_emails:
await admin_db.set_kv("allowed_emails", {"emails": seed_emails})
if seed_drafts:
await admin_db.set_kv("drafts", seed_drafts)
logger.info(
"Seeded admin_kv from JSON (mode=%s): clients=%d emails=%d drafts=%d",
mode, len(seed_clients), len(seed_emails), len(seed_drafts),
)
except Exception as exc:
logger.warning("Admin seed from JSON failed: %s", exc)
async def _merge_legacy_seed_if_present() -> None:
"""Merge the shipped legacy-clients-seed.json into _client_profiles.
Add-only: never overwrites an email that already exists in the live data.
Idempotent: re-running on every boot is a no-op once the entries are in.
Writes the updated profiles back to the JSON file + admin_kv so the merged
state survives container restarts.
"""
global _client_profiles
seed_path = _LEGACY_SEED_FILE
if not seed_path.exists():
return
try:
seed = json.loads(seed_path.read_text(encoding="utf-8"))
except Exception as exc:
logger.warning("Legacy seed file unreadable (%s): %s", seed_path, exc)
return
if not isinstance(seed, dict) or not seed:
return
added: list[str] = []
skipped_existing = 0
for raw_email, profile in seed.items():
if not isinstance(raw_email, str) or not isinstance(profile, dict):
continue
email = raw_email.strip().lower()
if not email:
continue
if email == OWNER_EMAIL.strip().lower():
continue
if email in _client_profiles:
skipped_existing += 1
continue
_client_profiles[email] = profile
added.append(email)
if not added:
logger.info(
"Legacy seed already merged (existing=%d, candidates=%d).",
skipped_existing, len(seed),
)
return
snapshot = dict(_client_profiles)
try:
await asyncio.to_thread(_save_client_profiles_file, snapshot)
except Exception as exc:
logger.warning("Could not save client_profiles after legacy merge: %s", exc)
try:
await _persist_admin_state("client_profiles", snapshot)
except Exception as exc:
logger.warning("Could not persist client_profiles to postgres after legacy merge: %s", exc)
logger.info(
"Legacy seed merged: added=%d skipped_existing=%d total_after=%d",
len(added), skipped_existing, len(_client_profiles),
)
async def _load_allowed_emails_async() -> set[str]:
if admin_db.is_enabled():
data = await admin_db.get_kv("allowed_emails")
if isinstance(data, dict):
emails = data.get("emails", [])
if isinstance(emails, list):
seed = {e.strip().lower() for e in os.environ.get("ALLOWED_EMAILS", "").split(",") if e.strip()}
seed.update(e.lower() for e in emails if isinstance(e, str))
return seed
return _load_allowed_emails_from_file()
async def _load_client_profiles_async() -> dict[str, dict]:
if admin_db.is_enabled():
data = await admin_db.get_kv("client_profiles")
if isinstance(data, dict):
return data
return _load_client_profiles_from_file()
async def _load_drafts_async() -> dict:
if admin_db.is_enabled():
data = await admin_db.get_kv("drafts")
if isinstance(data, dict):
return data
return _load_drafts_from_file()
async def _load_scheduled_emails_async() -> dict[str, dict]:
if admin_db.is_enabled():
data = await admin_db.get_kv("scheduled_emails")
if isinstance(data, dict):
return data
return _load_scheduled_emails_from_file()
_allowed_emails: set[str] = _load_allowed_emails_from_file()
if OWNER_EMAIL:
_allowed_emails.add(OWNER_EMAIL.strip().lower())
_pending_codes: dict[str, dict] = {} # email -> {code, expires_at, attempts}
_active_sessions: dict[str, dict] = {} # token -> {email, expires_at}
_code_requests: dict[str, deque] = {} # email -> deque of monotonic timestamps
_client_profiles: dict[str, dict] = _load_client_profiles_from_file()
_drafts: dict[str, dict] = _load_drafts_from_file() # email -> {onboarding: {...}, contract: {...}}
_auth_failures_by_ip: dict[str, deque] = {} # ip -> deque of failure timestamps
_blocked_ips: dict[str, float] = {} # ip -> unblock_at (monotonic)
_auth_lock = asyncio.Lock()
_birthday_auto_task: asyncio.Task | None = None
# Durable queue of owner-scheduled emails, keyed by id. Each entry is a dict
# (see _enqueue_scheduled_welcome). Persisted to admin_kv / JSON via
# _persist_scheduled_emails so queued sends survive a container restart.
_scheduled_emails: dict[str, dict] = _load_scheduled_emails_from_file()
_scheduled_task: asyncio.Task | None = None
_scheduled_lock = asyncio.Lock()
logger.info("Auth: loaded %d allowed email(s)", len(_allowed_emails))
async def _require_session_email(request: Request) -> str:
auth_header = request.headers.get("Authorization", "")
token = auth_header.removeprefix("Bearer ").strip()
if not token:
raise HTTPException(status_code=401, detail="No token provided.")
async with _auth_lock:
session = _active_sessions.get(token)
if not session:
raise HTTPException(status_code=401, detail="Invalid session.")
if time.time() > session["expires_at"]:
_active_sessions.pop(token, None)
raise HTTPException(status_code=401, detail="Session expired. Please sign in again.")
return session["email"]
async def _require_owner_email(request: Request) -> str:
email = await _require_session_email(request)
if email not in CP_ADMIN_EMAILS:
raise HTTPException(status_code=403, detail="Owner access required.")
return email
async def _register_email(email: str) -> None:
normalized = email.strip().lower()
if not normalized:
return
async with _auth_lock:
if normalized not in _allowed_emails:
_allowed_emails.add(normalized)
snapshot = sorted(_allowed_emails)
await asyncio.to_thread(_save_allowed_emails_file, set(_allowed_emails))
await _persist_admin_state("allowed_emails", {"emails": snapshot})
logger.info("Auth: registered new allowed email: %s", normalized)
def _client_is_reachable(profile: dict) -> bool:
"""True if outreach (welcome pack, birthday email, etc.) should still target
this client. Excludes lifecycle states that mean the relationship has ended.
"""
lifecycle = profile.get("lifecycle")
if not isinstance(lifecycle, dict):
return True
return lifecycle.get("status") not in {"cancelled", "archived"}
async def _store_client_profile(email: str, profile: dict) -> None:
normalized = email.strip().lower()
if not normalized:
return
async with _auth_lock:
existing = _client_profiles.get(normalized, {})
merged = {
k: v
for k, v in {**existing, **profile}.items()
if v is not None and not (isinstance(v, str) and v == "")
}
if merged != existing:
_client_profiles[normalized] = merged
snapshot = dict(_client_profiles)
await asyncio.to_thread(_save_client_profiles_file, snapshot)
await _persist_admin_state("client_profiles", snapshot)
async def _persist_scheduled_emails() -> None:
"""Write the scheduled-email queue to the JSON file + postgres. Callers
must hold _scheduled_lock so the snapshot is internally consistent."""
snapshot = dict(_scheduled_emails)
await asyncio.to_thread(_save_scheduled_emails_file, snapshot)
await _persist_admin_state("scheduled_emails", snapshot)
def _parse_schedule_datetime(value: str) -> datetime:
"""Parse an ISO 8601 datetime from the owner UI (datetime-local sends
'YYYY-MM-DDTHH:MM'). Returns a naive local datetime to match the rest of
the app, which compares against datetime.now(). Raises HTTPException(400)
on anything unparseable."""
raw = (value or "").strip()
if not raw:
raise HTTPException(status_code=400, detail="Please choose a date and time to schedule the email.")
try:
parsed = datetime.fromisoformat(raw)
except ValueError:
raise HTTPException(status_code=400, detail="That schedule time could not be understood.")
# Drop any timezone so comparisons against datetime.now() (naive) are valid.
if parsed.tzinfo is not None:
parsed = parsed.astimezone().replace(tzinfo=None)
return parsed
async def _update_client_profile(
email: str,
next_email: str,
profile_updates: dict[str, Any],
) -> dict[str, Any]:
normalized = email.strip().lower()
next_normalized = next_email.strip().lower()
if not normalized or not next_normalized:
raise HTTPException(status_code=400, detail="A valid email is required.")
async with _auth_lock:
existing = _client_profiles.get(normalized)
if not isinstance(existing, dict):
raise HTTPException(status_code=404, detail="Client not found.")
target_existing = _client_profiles.get(next_normalized)
if next_normalized != normalized and isinstance(target_existing, dict):
raise HTTPException(status_code=409, detail="Another client already uses that email address.")
merged = {
k: v
for k, v in {**existing, **profile_updates}.items()
if v is not None and not (isinstance(v, str) and v == "")
}
if next_normalized != normalized:
_client_profiles.pop(normalized, None)
_client_profiles[next_normalized] = merged
draft = _drafts.pop(normalized, None)
if draft is not None:
_drafts[next_normalized] = draft
pending = _pending_codes.pop(normalized, None)
if pending is not None:
_pending_codes[next_normalized] = pending
code_requests = _code_requests.pop(normalized, None)
if code_requests is not None:
_code_requests[next_normalized] = code_requests
for session in _active_sessions.values():
if isinstance(session, dict) and session.get("email") == normalized:
session["email"] = next_normalized
if normalized != OWNER_EMAIL.strip().lower():
_allowed_emails.discard(normalized)
_allowed_emails.add(next_normalized)
else:
_client_profiles[normalized] = merged
profiles_snapshot = dict(_client_profiles)
drafts_snapshot = dict(_drafts)
allowed_emails_snapshot = sorted(_allowed_emails)
await asyncio.to_thread(_save_client_profiles_file, profiles_snapshot)
await asyncio.to_thread(_save_drafts_file, drafts_snapshot)
await asyncio.to_thread(_save_allowed_emails_file, set(allowed_emails_snapshot))
await _persist_admin_state("client_profiles", profiles_snapshot)
await _persist_admin_state("drafts", drafts_snapshot)
await _persist_admin_state("allowed_emails", {"emails": allowed_emails_snapshot})
await _save_active_sessions_async()
return merged
async def _reset_client_onboarding(email: str) -> dict[str, Any]:
"""Mark a client's onboarding as incomplete while keeping every saved detail.
Used for clients imported from the legacy Gravity Forms data: they keep their
contact/dog details (and their previous submission is archived for reference),
but they drop back into the pending list so they can sign in with their email
and complete the new onboarding form.
"""
normalized = email.strip().lower()
if not normalized:
raise HTTPException(status_code=400, detail="A valid email is required.")
async with _auth_lock:
existing = _client_profiles.get(normalized)
if not isinstance(existing, dict):
raise HTTPException(status_code=404, detail="Client not found.")
updated = dict(existing)
updated["onboardingCompleted"] = False
# Archive the prior completion so nothing is lost, then clear the live
# markers that keep them out of the pending / onboarding views.
previous_submitted = updated.pop("onboardingSubmittedAt", "")
previous_submission = updated.pop("onboardingSubmission", None)
if previous_submitted:
updated["previousOnboardingSubmittedAt"] = previous_submitted
if previous_submission is not None:
updated["previousOnboardingSubmission"] = previous_submission
updated["onboardingResetAt"] = datetime.now().isoformat(timespec="seconds")
_client_profiles[normalized] = updated
snapshot = dict(_client_profiles)
await asyncio.to_thread(_save_client_profiles_file, snapshot)
await _persist_admin_state("client_profiles", snapshot)
return updated
def _check_ip_blocked(ip: str, request_id: str) -> None:
now = time.monotonic()
unblock_at = _blocked_ips.get(ip)
if unblock_at is not None:
if now < unblock_at:
remaining = int(unblock_at - now)
logger.warning("[%s] auth: blocked ip=%s (%ds remaining)", request_id, ip, remaining)
raise HTTPException(
status_code=429,
detail=f"Too many failed attempts. Try again in {remaining // 60 + 1} minute(s).",
headers={"Retry-After": str(remaining)},
)
else:
del _blocked_ips[ip]
def _record_auth_failure(ip: str, request_id: str, reason: str) -> None:
now = time.monotonic()
failures = _auth_failures_by_ip.setdefault(ip, deque())
while failures and now - failures[0] > AUTH_IP_FAILURE_WINDOW:
failures.popleft()
failures.append(now)
logger.warning("[%s] auth: failure ip=%s reason=%r total_in_window=%d", request_id, ip, reason, len(failures))
if len(failures) >= AUTH_IP_MAX_FAILURES:
_blocked_ips[ip] = now + AUTH_IP_BLOCK_DURATION
logger.warning(
"[%s] auth: ip=%s BLOCKED for %ds after %d failures",
request_id, ip, AUTH_IP_BLOCK_DURATION, len(failures),
)
class _BodySizeLimitMiddleware:
"""Reject requests whose Content-Length exceeds MAX_REQUEST_BODY_BYTES.
Defence-in-depth alongside nginx ``client_max_body_size``. Streaming
requests without a Content-Length header are tracked byte-by-byte and
short-circuited if they overflow the cap.
"""
def __init__(self, app: ASGIApp, max_bytes: int) -> None:
self.app = app
self.max_bytes = max_bytes
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
if scope["type"] != "http":
await self.app(scope, receive, send)
return
headers = {k.decode("latin-1").lower(): v.decode("latin-1") for k, v in scope.get("headers", [])}
declared = headers.get("content-length")
if declared is not None:
try:
if int(declared) > self.max_bytes:
await _send_413(send)
return
except ValueError:
pass
received = 0
overflowed = False
async def _wrapped_receive():
nonlocal received, overflowed
message = await receive()
if message["type"] == "http.request":
received += len(message.get("body", b""))
if received > self.max_bytes:
overflowed = True
return {"type": "http.disconnect"}
return message
if overflowed:
await _send_413(send)
return
await self.app(scope, _wrapped_receive, send)
async def _send_413(send: Send) -> None:
await send({
"type": "http.response.start",
"status": 413,
"headers": [(b"content-type", b"application/json")],
})
await send({
"type": "http.response.body",
"body": b'{"detail":"Request body too large."}',
})
app.add_middleware(_BodySizeLimitMiddleware, max_bytes=MAX_REQUEST_BODY_BYTES)
app.add_middleware(TrustedHostMiddleware, allowed_hosts=list(TRUSTED_HOSTS))
app.add_middleware(
CORSMiddleware,
allow_origins=list(CORS_ALLOWED_ORIGINS),
allow_methods=["POST", "GET"],
allow_headers=["Authorization", "Content-Type", "X-Requested-With"],
allow_credentials=False,
max_age=600,
)
@app.middleware("http")
async def _request_logging_middleware(request: Request, call_next):
request_id = uuid.uuid4().hex[:8]
request.state.request_id = request_id
started = time.monotonic()
try:
response = await call_next(request)
except Exception:
elapsed_ms = (time.monotonic() - started) * 1000
logger.exception(
"[%s] %s %s crashed after %.0fms",
request_id, request.method, request.url.path, elapsed_ms,
)
raise
elapsed_ms = (time.monotonic() - started) * 1000
logger.info(
"[%s] %s %s → %d (%.0fms)",
request_id, request.method, request.url.path, response.status_code, elapsed_ms,
)
response.headers["X-Request-ID"] = request_id
return response
# ── Helpers ──────────────────────────────────────────────────────────────────
def _get_ip(request: Request) -> str:
forwarded = request.headers.get("x-forwarded-for")
if forwarded:
return forwarded.split(",")[0].strip()
return request.client.host if request.client else "unknown"
def _is_deploy_smoke(request: Request) -> bool:
"""True when the request carries a matching X-Deploy-Smoke header.
Used by the deploy script to verify the form endpoints are reachable and
parse a valid payload, without producing a real submission. Disabled
entirely when DEPLOY_SMOKE_SECRET is unset.
"""
if not DEPLOY_SMOKE_SECRET:
return False
presented = request.headers.get("x-deploy-smoke") or ""
if not presented:
return False
return secrets.compare_digest(presented, DEPLOY_SMOKE_SECRET)
_submit_attempts_by_ip: dict[str, deque[float]] = {}
_submit_attempts_by_email: dict[str, deque[float]] = {}
_submit_rate_limit_lock = asyncio.Lock()
def _trimmed(value: str) -> str:
return value.strip()
def _prune_attempts(attempts: deque[float], now: float, window_seconds: int) -> None:
while attempts and now - attempts[0] > window_seconds:
attempts.popleft()
def _seconds_until_allowed(last_attempt_at: float, now: float, min_interval_seconds: int) -> int:
retry_after = max(1, int(min_interval_seconds - (now - last_attempt_at)))
return retry_after
async def _enforce_submit_rate_limits(request_id: str, ip: str, email: str) -> None:
now = time.monotonic()
normalized_email = email.strip().lower()
async with _submit_rate_limit_lock:
ip_attempts = _submit_attempts_by_ip.setdefault(ip, deque())
email_attempts = _submit_attempts_by_email.setdefault(normalized_email, deque())
_prune_attempts(ip_attempts, now, RATE_LIMIT_WINDOW_SECONDS)
_prune_attempts(email_attempts, now, RATE_LIMIT_WINDOW_SECONDS)
if ip_attempts and now - ip_attempts[-1] < RATE_LIMIT_MIN_INTERVAL_SECONDS:
retry_after = _seconds_until_allowed(ip_attempts[-1], now, RATE_LIMIT_MIN_INTERVAL_SECONDS)
logger.warning(
"[%s] rate limited: ip=%s submitted again after %.1fs (minimum %ss)",
request_id,
ip,
now - ip_attempts[-1],
RATE_LIMIT_MIN_INTERVAL_SECONDS,
)
raise HTTPException(
status_code=429,
detail=f"Please wait about {retry_after} seconds before trying again.",
)
if len(ip_attempts) >= RATE_LIMIT_MAX_PER_IP:
logger.warning(
"[%s] rate limited: ip=%s exceeded %d submissions in %ss",
request_id,
ip,
RATE_LIMIT_MAX_PER_IP,
RATE_LIMIT_WINDOW_SECONDS,
)
raise HTTPException(
status_code=429,
detail="Too many enquiries from this connection. Please try again a little later.",
)
if len(email_attempts) >= RATE_LIMIT_MAX_PER_EMAIL:
logger.warning(
"[%s] rate limited: email=%s exceeded %d submissions in %ss",
request_id,
normalized_email,
RATE_LIMIT_MAX_PER_EMAIL,
RATE_LIMIT_WINDOW_SECONDS,
)
raise HTTPException(
status_code=429,
detail="That email address has reached the enquiry limit for now. Please try again later.",
)
ip_attempts.append(now)
email_attempts.append(now)
def _enforce_form_timing(request_id: str, data: BaseSubmission) -> None:
if data.formStartedAt is None or data.formStartedAt <= 0:
logger.warning("[%s] rejected: missing or invalid formStartedAt", request_id)
raise HTTPException(
status_code=400,
detail="Please refresh the page and try again.",
)
elapsed_seconds = (time.time() * 1000 - data.formStartedAt) / 1000
if elapsed_seconds < FORM_MIN_SECONDS:
logger.warning(
"[%s] rejected: form submitted too quickly (%.2fs < %ss)",
request_id,
elapsed_seconds,
FORM_MIN_SECONDS,
)
raise HTTPException(
status_code=400,
detail="Please take a moment to fill in the form before sending it.",
)
if elapsed_seconds > FORM_MAX_SECONDS:
logger.warning(
"[%s] rejected: stale form submission (%.0fs > %ss)",
request_id,
elapsed_seconds,
FORM_MAX_SECONDS,
)
raise HTTPException(
status_code=400,
detail="This form has been open for too long. Please refresh the page and try again.",
)
def _is_honeypot_triggered(data: BaseSubmission) -> bool:
return bool(_trimmed(data.website))
def _is_general_enquiry(data: BookingSubmission) -> bool:
return _trimmed(data.enquiryType).lower() == "general"
def _enquiry_type_label(data: BookingSubmission) -> str:
return "General enquiry" if _is_general_enquiry(data) else "Booking enquiry"
def _validate_submission(request_id: str, data: BookingSubmission) -> None:
enquiry_type = _trimmed(data.enquiryType).lower()
if enquiry_type not in {"booking", "general"}:
logger.warning("[%s] rejected: invalid enquiryType=%r", request_id, data.enquiryType)
raise HTTPException(
status_code=400,
detail="Please choose a valid enquiry type and try again.",
)
if not _trimmed(data.fullName):
logger.warning("[%s] rejected: missing full name", request_id)
raise HTTPException(
status_code=400,
detail="Please enter your full name.",
)
if not _trimmed(data.phone):
logger.warning("[%s] rejected: missing phone number", request_id)
raise HTTPException(
status_code=400,
detail="Please enter your contact number.",
)
if _is_general_enquiry(data):
if not ENABLE_GENERAL_ENQUIRIES:
logger.warning("[%s] rejected: general enquiries are disabled", request_id)
raise HTTPException(
status_code=403,
detail="General enquiries are currently unavailable through this form.",
)
if not _trimmed(data.message):
logger.warning("[%s] rejected: missing general enquiry message", request_id)
raise HTTPException(
status_code=400,
detail="Please tell us how we can help.",
)
return
if not _trimmed(data.petName):
logger.warning("[%s] rejected: missing pet name", request_id)
raise HTTPException(
status_code=400,
detail="Please enter your dog's name.",
)
if not _trimmed(data.location):
logger.warning("[%s] rejected: missing location", request_id)
raise HTTPException(
status_code=400,
detail="Please enter your location.",
)
def _normalize_submission(data: BookingSubmission) -> None:
data.enquiryType = "general" if _is_general_enquiry(data) else "booking"
data.fullName = _trimmed(data.fullName)
data.phone = _trimmed(data.phone)
data.petName = _trimmed(data.petName)
data.location = _trimmed(data.location)
data.message = _trimmed(data.message)
data.referrer = _trimmed(data.referrer)
data.page = _trimmed(data.page)
data.services = [_trimmed(service) for service in data.services if _trimmed(service)]
data.journey = [_trimmed(step) for step in data.journey if _trimmed(step)][:12]
data.stepChanges = max(0, data.stepChanges)
for field_name in ("visitStartedAt", "pageEnteredAt", "firstInteractionAt", "sendClickedAt"):
value = getattr(data, field_name)
if value is None or value <= 0:
setattr(data, field_name, None)
if _is_general_enquiry(data):
data.petName = ""
data.location = ""
data.services = []
def _validate_onboarding_submission(request_id: str, data: OnboardingSubmission) -> None:
if not _trimmed(data.fullName):
logger.warning("[%s] onboarding rejected: missing full name", request_id)
raise HTTPException(status_code=400, detail="Please enter your full name.")
if not _trimmed(data.phone):
logger.warning("[%s] onboarding rejected: missing phone", request_id)
raise HTTPException(status_code=400, detail="Please enter your phone number.")
required_fields = {
"address": "Please enter your address.",
"dogName": "Please enter your dog's name.",
"dogBreed": "Please enter your dog's breed.",
"dogAge": "Please enter your dog's date of birth.",
"vetName": "Please enter your vet clinic name.",
"vetAddress": "Please enter your vet address.",
"vetPhone": "Please enter your vet phone number.",
"emergencyContactName": "Please enter an emergency contact name.",
"emergencyContactPhone": "Please enter an emergency contact phone number.",
}
for field_name, message in required_fields.items():
if not _trimmed(getattr(data, field_name)):
logger.warning("[%s] onboarding rejected: missing %s", request_id, field_name)
raise HTTPException(status_code=400, detail=message)
if not data.servicesNeeded:
logger.warning("[%s] onboarding rejected: missing services", request_id)
raise HTTPException(status_code=400, detail="Please choose at least one service.")
if data.regularFleaTickTreatment not in {"yes", "no"}:
raise HTTPException(status_code=400, detail="Please confirm whether your dog gets regular flea and tick treatment.")
if data.petInsurance not in {"yes", "no"}:
raise HTTPException(status_code=400, detail="Please confirm whether your dog has pet insurance.")
if data.petInsurance == "no" and not data.petInsuranceOwnerExpenseAccepted:
raise HTTPException(status_code=400, detail="Please confirm the owner-expense acknowledgement for dogs without pet insurance.")
# Council registration and vaccination are no longer hard blockers: the owner
# reviews these on the submission and follows up with the client separately.
# The actual yes/no answers are preserved on the snapshot and surfaced below.
if not data.emergencyVetConsent:
raise HTTPException(status_code=400, detail="Please confirm emergency veterinary consent.")
if not data.termsAccepted:
raise HTTPException(status_code=400, detail="Please confirm the onboarding declaration.")
signature = _trimmed(data.signatureDataUrl)
if not signature.startswith("data:image/png;base64,") or len(signature) < 128:
logger.warning("[%s] onboarding rejected: invalid signature payload", request_id)
raise HTTPException(status_code=400, detail="Please add your signature before sending.")
def _normalize_onboarding_submission(data: OnboardingSubmission) -> None:
data.fullName = _trimmed(data.fullName)
data.phone = _trimmed(data.phone)
data.address = _trimmed(data.address)
data.dogName = _trimmed(data.dogName)
data.dogBreed = _trimmed(data.dogBreed)
data.dogAge = _trimmed(data.dogAge)
data.temperament = _trimmed(data.temperament)
data.medicalNotes = _trimmed(data.medicalNotes)
data.accessInstructions = _trimmed(data.accessInstructions)
data.vetName = _trimmed(data.vetName)
data.vetAddress = _trimmed(data.vetAddress)
data.vetPhone = _trimmed(data.vetPhone)
data.emergencyContactName = _trimmed(data.emergencyContactName)
data.emergencyContactPhone = _trimmed(data.emergencyContactPhone)
data.regularFleaTickTreatment = _trimmed(data.regularFleaTickTreatment).lower()
data.petInsurance = _trimmed(data.petInsurance).lower()
data.referrer = _trimmed(data.referrer)
data.page = _trimmed(data.page)
data.servicesNeeded = [_trimmed(service) for service in data.servicesNeeded if _trimmed(service)][:8]
for field_name in ("visitStartedAt", "pageEnteredAt", "firstInteractionAt", "sendClickedAt"):
value = getattr(data, field_name)
if value is None or value <= 0:
setattr(data, field_name, None)
def _parse_ua(ua: str) -> str:
if not ua:
return "Unknown"
browsers = [("Edg/", "Edge"), ("OPR/", "Opera"), ("Chrome/", "Chrome"),
("Firefox/", "Firefox"), ("Safari/", "Safari")]
systems = [("Windows NT 10", "Windows 10/11"), ("Windows NT 6", "Windows 8"),
("Mac OS X", "macOS"), ("iPhone", "iPhone"), ("iPad", "iPad"),
("Android", "Android"), ("Linux", "Linux")]
browser = next((n for p, n in browsers if p in ua), "Unknown browser")
system = next((n for p, n in systems if p in ua), "Unknown OS")
return f"{browser} on {system}"
def _detail_row(label: str, value: str) -> str:
if not value:
return ""
return f"""
{label}
{value}
"""
def _meta_row(label: str, value: str) -> str:
if not value:
return ""
return f"""
{label}
{value}
"""
def _format_duration_ms(duration_ms: int | None) -> str:
if duration_ms is None or duration_ms < 0:
return ""
total_seconds = int(round(duration_ms / 1000))
minutes, seconds = divmod(total_seconds, 60)
hours, minutes = divmod(minutes, 60)
if hours > 0:
return f"{hours}h {minutes}m"
if minutes > 0:
return f"{minutes}m {seconds}s"
return f"{seconds}s"
def _duration_between(start_ms: int | None, end_ms: int | None) -> str:
if start_ms is None or end_ms is None or end_ms < start_ms:
return ""
return _format_duration_ms(end_ms - start_ms)
def _journey_text(journey: list[str]) -> str:
if not journey:
return ""
return " -> ".join(journey)
# ── Email templates ──────────────────────────────────────────────────────────
def _logo_header(badge_html: str = "", subtitle: str = "") -> str:
badge = f'
{badge_html}
' if badge_html else ""
sub = f"""
{subtitle}
""" if subtitle else ""
return f"""
{sub}
{badge}
"""
def client_email(data: BookingSubmission) -> str:
is_general = _is_general_enquiry(data)
services_text = ", ".join(data.services) if data.services else "Not specified"
enquiry_summary_rows = [
_detail_row("Your name", data.fullName),
_detail_row("Email", str(data.email)),
_detail_row("Phone", data.phone),
_detail_row("Type", _enquiry_type_label(data)),
]
if is_general:
if data.message:
enquiry_summary_rows.append(_detail_row("Message", data.message))
intro_html = (
"We’ve received your message and we will be in touch shortly."
)
next_steps_html = (
"We will review your message and reply within 1 business day."
)
logo_subtitle = "General enquiries and dog walking support"
else:
enquiry_summary_rows.extend(
[
_detail_row("Dog’s name", data.petName),
_detail_row("Location", data.location),
_detail_row("Services", services_text),
]
)
if data.message:
enquiry_summary_rows.append(_detail_row("About the dog", data.message))
intro_html = (
"We’ve received your enquiry and we will be in touch shortly to arrange "
"a Meet & Greet with you and "
f"{data.petName}."
)
next_steps_html = (
"We will review your details and reach out within 1 business day "
"to schedule a free Meet & Greet. No commitment required — just a "
f"chance for {data.petName} to make a new best friend."
)
logo_subtitle = "Professional dog walking services"
return f"""
We received your enquiry
{_logo_header(subtitle=logo_subtitle)}
Thanks, {data.fullName.split()[0]}! 🐾
{intro_html}
Your enquiry summary
{"".join(enquiry_summary_rows)}
What happens next?
{next_steps_html}
Questions? Just reply to this email or reach us at 022 642 1011.
"""
def _render_pdf_sync(html: str) -> bytes:
from weasyprint import HTML # imported lazily so unit tests don't require the native libs
return HTML(string=html).write_pdf()
# Feature flags — flip to True to attach a PDF copy of the signed form to the owner email.
# Kept as in-code booleans (not env vars) so the contract path stays off until explicitly enabled.
CONTRACT_PDF_ATTACHMENT_ENABLED = False
ONBOARDING_PDF_ATTACHMENT_ENABLED = True
async def _signed_form_pdf_attachment(html: str, full_name: str, kind: str, request_id: str) -> dict | None:
safe_name = re.sub(r"[^a-z0-9]+", "-", _trimmed(full_name).lower()).strip("-") or "client"
try:
pdf_bytes = await asyncio.to_thread(_render_pdf_sync, html)
except Exception as exc:
logger.error("[%s] %s PDF generation failed: %s", request_id, kind, exc, exc_info=True)
return None
logger.info("[%s] %s PDF generated: %d bytes", request_id, kind, len(pdf_bytes))
return {
"filename": f"goodwalk-{kind}-{safe_name}.pdf",
"content": base64.b64encode(pdf_bytes).decode("ascii"),
}
# ── Sending with retries ─────────────────────────────────────────────────────
def _client_bcc_list() -> list[str]:
"""BCC recipients for any real email sent home to a client (welcome pack,
birthday, enquiry reply, onboarding confirmation, ...). The business owner is
always copied so they can see exactly what each client received; CLIENT_BCC
adds an optional extra inbox when configured. The OWNER_BCC placeholder is
treated as unset."""
out: list[str] = []
for addr in (OWNER_EMAIL, CLIENT_BCC):
a = (addr or "").strip()
if not a or a.lower() == "example@example.com":
continue
if a.lower() not in {x.lower() for x in out}:
out.append(a)
return out
async def _send_email(payload: dict, label: str, request_id: str) -> dict:
if DEV_MODE:
to = payload.get("to", [])
subject = payload.get("subject", "(no subject)")
logger.warning("[DEV] skipping email send — label=%s to=%s subject=%r", label, to, subject)
return {"id": "dev-mode"}
last_exc: Exception | None = None
for attempt in range(1, MAX_SEND_ATTEMPTS + 1):
started = time.monotonic()
try:
result = await asyncio.wait_for(
asyncio.to_thread(resend.Emails.send, payload),
timeout=EMAIL_SEND_TIMEOUT_SECONDS,
)
elapsed_ms = (time.monotonic() - started) * 1000
email_id = result.get("id") if isinstance(result, dict) else None
logger.info(
"[%s] %s sent to %s (attempt %d/%d, %.0fms, id=%s)",
request_id, label, payload.get("to"), attempt, MAX_SEND_ATTEMPTS,
elapsed_ms, email_id or "n/a",
)
return result or {}
except Exception as exc:
last_exc = exc
elapsed_ms = (time.monotonic() - started) * 1000
status = getattr(exc, "status_code", None) or getattr(exc, "code", None)
non_retryable = (
isinstance(status, int) and 400 <= status < 500 and status != 429
)
logger.warning(
"[%s] %s send failed (attempt %d/%d, %.0fms): %s: %s (status=%s)",
request_id, label, attempt, MAX_SEND_ATTEMPTS, elapsed_ms,
type(exc).__name__, exc, status,
exc_info=True,
)
if non_retryable:
logger.info(
"[%s] %s: non-retryable status %s, aborting retries",
request_id, label, status,
)
break
if attempt == MAX_SEND_ATTEMPTS:
break
backoff = (2 ** (attempt - 1)) + random.uniform(0, 0.4)
logger.info("[%s] retrying %s in %.2fs", request_id, label, backoff)
await asyncio.sleep(backoff)
assert last_exc is not None
raise last_exc
def _build_startup_test_submission() -> BookingSubmission:
now_ms = int(time.time() * 1000)
sample = BookingSubmission(
enquiryType="booking",
fullName="Sarah Thompson",
email="sarah.thompson@example.com",
phone="021 555 0142",
petName="Milo",
location="Grey Lynn",
message=(
"Milo is a 2-year-old cavoodle with good recall and a friendly nature. "
"He loves other dogs, is comfortable off lead in safe areas, and we are "
"looking for regular weekday pack walks while we are at work."
),
services=["Pack Walks", "Puppy Visits"],
formStartedAt=now_ms - (6 * 60 * 1000 + 35 * 1000),
visitStartedAt=now_ms - (14 * 60 * 1000 + 10 * 1000),
pageEnteredAt=now_ms - (7 * 60 * 1000 + 5 * 1000),
firstInteractionAt=now_ms - (5 * 60 * 1000 + 20 * 1000),
sendClickedAt=now_ms,
stepChanges=3,
journey=["/", "/pack-walks", "/our-pricing", "/book"],
referrer="https://www.google.com/search?q=goodwalk+auckland+dog+walking",
page="https://www.goodwalk.co.nz/book?service=pack-walks",
)
_normalize_submission(sample)
return sample
async def _send_startup_test_email() -> None:
if not STARTUP_TEST_RECIPIENT:
logger.info("Startup test email skipped: OWNER_BCC is not set to a real address")
return
request_id = "startup-test"
sample = _build_startup_test_submission()
payload = {
"from": FROM_EMAIL,
"to": [STARTUP_TEST_RECIPIENT],
"reply_to": str(sample.email),
"subject": f"Startup preview — New GoodWalk lead — {sample.fullName} ({sample.petName})",
"html": owner_email(sample, "127.0.0.1", f"Startup Preview ({APP_VERSION})"),
}
await _send_email(payload, label="startup_test_email", request_id=request_id)
# ── Routes ───────────────────────────────────────────────────────────────────
async def _startup_smoke_pdf() -> None:
"""Import WeasyPrint and run a trivial render to surface native-lib issues
(libpango/cairo/etc.) at boot rather than on the first PDF request."""
try:
await asyncio.to_thread(_render_pdf_sync, "ok")
logger.info("Startup smoke: WeasyPrint OK — PDF attachments available")
except Exception as exc:
logger.error("Startup smoke: WeasyPrint UNAVAILABLE — PDF attachments will be skipped (%s)", exc)
async def _startup_verify_schema() -> None:
"""Force schema creation at boot and verify the new tables exist so the
activity log isn't silently empty if CREATE permission is missing."""
if not admin_db.is_enabled():
logger.warning("Startup smoke: postgres disabled — activity/submissions will NOT be recorded")
return
try:
pool = await admin_db.get_pool()
if pool is None:
logger.warning("Startup smoke: postgres pool unavailable — activity/submissions will NOT be recorded")
return
await admin_db._ensure_schema() # idempotent
async with pool.acquire() as conn:
row = await conn.fetchrow(
"select to_regclass('public.events') as ev, to_regclass('public.submissions') as sub"
)
if row and row["ev"] and row["sub"]:
logger.info("Startup smoke: pg tables OK — events + submissions ready")
else:
logger.error("Startup smoke: pg tables MISSING (events=%s submissions=%s) — check CREATE perms",
row["ev"] if row else None, row["sub"] if row else None)
except Exception as exc:
logger.error("Startup smoke: pg schema verify FAILED (%s)", exc)
async def _startup_mail_check() -> None:
global _birthday_auto_task, _scheduled_task, _allowed_emails, _client_profiles, _drafts, _scheduled_emails
# 0. Boot-time smoke tests so silent failures surface immediately.
await _startup_smoke_pdf()
await _startup_verify_schema()
# 1. Seed postgres from JSON if admin_kv is empty (one-time migration).
await _seed_admin_state_from_json_if_needed()
# 2. Refresh the in-memory caches from postgres so the app reads the
# canonical dataset even after restarts.
if admin_db.is_enabled():
try:
db_clients = await _load_client_profiles_async()
if isinstance(db_clients, dict):
_client_profiles = db_clients
db_emails = await _load_allowed_emails_async()
if isinstance(db_emails, set):
_allowed_emails = db_emails
if OWNER_EMAIL:
_allowed_emails.add(OWNER_EMAIL.strip().lower())
db_drafts = await _load_drafts_async()
if isinstance(db_drafts, dict):
_drafts = db_drafts
db_sessions = await _load_active_sessions_async()
if db_sessions:
_active_sessions.update(db_sessions)
db_scheduled = await _load_scheduled_emails_async()
if isinstance(db_scheduled, dict):
_scheduled_emails = db_scheduled
logger.info(
"Admin state refreshed from postgres: clients=%d emails=%d drafts=%d sessions=%d scheduled=%d",
len(_client_profiles), len(_allowed_emails), len(_drafts), len(_active_sessions), len(_scheduled_emails),
)
except Exception:
logger.exception("Admin state refresh from postgres failed; using JSON snapshot")
# 3. Merge any shipped legacy seed (add-only — never clobbers live entries).
await _merge_legacy_seed_if_present()
try:
await _send_startup_test_email()
except Exception:
logger.exception("Startup test email failed")
if _birthday_auto_task is None or _birthday_auto_task.done():
_birthday_auto_task = asyncio.create_task(_birthday_auto_sender_loop())
if _scheduled_task is None or _scheduled_task.done():
_scheduled_task = asyncio.create_task(_scheduled_sender_loop())
async def _shutdown_background_tasks() -> None:
global _birthday_auto_task, _scheduled_task
for task_name in ("_birthday_auto_task", "_scheduled_task"):
task = globals().get(task_name)
if task is not None:
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
globals()[task_name] = None
@app.get("/health")
async def health() -> dict:
return {"status": "ok"}
def _auth_code_email(email: str, code: str) -> str:
return f"""
Your Goodwalk login code
Your login code
{code}
Enter this code on the Goodwalk onboarding page.
This code expires in {AUTH_CODE_TTL_SECONDS // 60} minutes. If you didn’t request this, you can safely ignore it.
Goodwalk · Auckland, New Zealand
"""
def _format_date_label(value: str) -> str:
raw = _trimmed(value)
if not raw:
return "To be confirmed"
try:
parsed = datetime.fromisoformat(raw)
return f"{parsed.day} {parsed.strftime('%b %Y')}"
except ValueError:
return raw
def _welcome_pack_email_html(client_name: str, dog_name: str, service_type: str, price_details: str, start_date: str) -> str:
first_name = client_name.split()[0] if client_name.strip() else "there"
dog_line = f" for {dog_name}" if dog_name.strip() else ""
formatted_start_date = _format_date_label(start_date)
return f"""
Welcome to the pack
Welcome to the pack
Hi {first_name}, we’d love to get {dog_name or 'your dog'} started with Goodwalk.
We’ve set aside the details below{dog_line}. When you’re ready, complete your onboarding form and we’ll take it from there.
Use the same email address you originally used with Goodwalk. We’ll send you a one-time code when you sign in.
"""
def _onboarding_confirmation_email_html(data: OnboardingSubmission) -> str:
first_name = data.fullName.split()[0] if data.fullName.strip() else "there"
dog_name = _trimmed(data.dogName)
service_names = [service.strip() for service in data.servicesNeeded if isinstance(service, str) and service.strip()]
service_summary = ", ".join(service_names[:2]) if service_names else "your selected service"
if len(service_names) > 2:
service_summary += f" + {len(service_names) - 2} more"
onboarding_url = "https://clients.goodwalk.co.nz/"
badge_html = (
'
Submitted
"
)
return f"""
Your onboarding has been submitted
{_logo_header(badge_html=badge_html, subtitle="Your onboarding details are safely with us")}
Thanks, {first_name}. Your onboarding is complete.
We’ve received your details{f" for {dog_name}" if dog_name else ""} and they’re now on file with Goodwalk.
You can sign back in any time to review what you submitted.
Snapshot
Owner
{data.fullName}
Dog
{dog_name or 'Details submitted'}
Services
{service_summary}
What happens next?
We’ll review your submission and come back to you if we need anything clarified.
If you need to check your details again, use the button below to sign back in with a one-time code.