import json import logging import os import re import sys import time import urllib.error import urllib.request import datetime as _dt from datetime import timedelta from decimal import Decimal, ROUND_HALF_UP from zoneinfo import ZoneInfo import boto3 from shared.cognito import looks_like_cognito_token, verify_cognito_id_token from shared.db import ( current_week, delete_order, get_form_status, get_menu, get_order, get_orders, get_roster, get_settings, get_summary, list_weeks, put_menu, put_order, ) from shared.secrets import get_parameter logger = logging.getLogger(__name__) logger.setLevel(logging.INFO) if not logger.handlers: logger.addHandler(logging.StreamHandler(sys.stderr)) EASTERN = ZoneInfo("America/New_York") CACHE_TTL_SECONDS = 300 # 5-minute TTL for cached config values PRESIGNED_URL_TTL_SECONDS = 300 # summary-PDF presigned URL lifetime ALLOWED_DOMAINS = {"seahavenind.com", "seahaven.com"} def _eastern_now() -> _dt.datetime: """Wall-clock 'now' in Eastern time (patch target for form-status tests).""" return _dt.datetime.now(EASTERN) _settings = None _settings_ts = 0.0 _google_client_id = None _google_client_id_ts = 0.0 _lambda = boto3.client("lambda") _s3 = boto3.client("s3") def _get_discount_settings() -> tuple[Decimal, Decimal]: global _settings, _settings_ts now = time.monotonic() if _settings is None or (now - _settings_ts) > CACHE_TTL_SECONDS: s = get_settings() _settings = ( Decimal(str(s.get("bulk_discount_percent", 0))), Decimal(str(s.get("company_subsidy_percent", 0))), ) _settings_ts = now return _settings def _get_admin_emails() -> set[str]: return {email.lower() for email in get_settings().get("admin_emails", [])} def _google_auth_configured() -> bool: return bool(os.environ.get("GOOGLE_CLIENT_ID_PARAM", "")) def _is_parameter_not_found(exc: Exception) -> bool: response_data = getattr(exc, "response", {}) if not isinstance(response_data, dict): return False return response_data.get("Error", {}).get("Code") == "ParameterNotFound" def _official_menu_retail_by_name(week: str) -> dict[str, Decimal]: """Map meal name -> retail price from Dynamo menu (authoritative for billing).""" row = get_menu(week) meals = (row or {}).get("meals") or [] out: dict[str, Decimal] = {} for m in meals: name = (m.get("name") or "").strip() if not name or m.get("price") is None: continue out[name] = Decimal(str(m["price"])) return out def _get_google_client_id() -> str: global _google_client_id, _google_client_id_ts now = time.monotonic() if _google_client_id is None or (now - _google_client_id_ts) > CACHE_TTL_SECONDS: param = os.environ.get("GOOGLE_CLIENT_ID_PARAM", "") if param: try: _google_client_id = get_parameter(param, decrypt=False) or "" except Exception as exc: if not _is_parameter_not_found(exc): raise _google_client_id = "" else: _google_client_id = "" _google_client_id_ts = now return _google_client_id def _verify_google_token(token: str) -> tuple[dict | None, str]: """Verify a Google ID token via the tokeninfo endpoint. Returns a tuple of (user_info, error_kind) where: - ({"name": ..., "email": ...}, "ok") on success - (None, "invalid") for bad/expired tokens or wrong audience/domain - (None, "unavailable") when the Google verification service is unreachable NOTE: The token is passed as a query parameter to Google's tokeninfo endpoint. This is acceptable because ID tokens are short-lived (typically ~1 hour) and this is Google's own documented verification method, but be aware that the token will appear in Google's server access logs. """ client_id = _get_google_client_id() if not client_id: return None, "invalid" try: req = urllib.request.Request( f"https://oauth2.googleapis.com/tokeninfo?id_token={token}" ) with urllib.request.urlopen(req, timeout=5) as resp: data = json.loads(resp.read()) if data.get("aud") != client_id: logger.warning("Google token audience mismatch: got %s", data.get("aud")) return None, "invalid" if data.get("hd") not in ALLOWED_DOMAINS: logger.warning("Google token domain mismatch: got %s", data.get("hd")) return None, "invalid" return {"name": data.get("name", ""), "email": data.get("email", "")}, "ok" except urllib.error.HTTPError as exc: logger.warning("Google token rejected (HTTP %s): %s", exc.code, exc) return None, "invalid" except (urllib.error.URLError, TimeoutError, OSError) as exc: logger.error("Google token verification service unavailable: %s", exc) return None, "unavailable" except Exception as exc: logger.error("Google token verification failed (bad token data): %s", exc) return None, "invalid" def _authentication_service_unavailable() -> dict: return response(503, {"error": "Authentication service temporarily unavailable"}) def _extract_bearer_token(event) -> str: headers = event.get("headers") or {} raw = headers.get("authorization") or headers.get("Authorization") or "" if raw.lower().startswith("bearer "): return raw[7:].strip() return "" def _verify_portal_token(token: str) -> dict | None: return verify_cognito_id_token( token, os.environ.get("PORTAL_COGNITO_ISSUER_PARAM", ""), os.environ.get("PORTAL_COGNITO_AUDIENCE_PARAM", ""), ) def _verify_admin(event) -> tuple[dict | None, dict | None]: """Verify portal or Google auth and admin access.""" token = _extract_bearer_token(event) if not token: return None, response(403, {"error": "Authentication required"}) if looks_like_cognito_token(token): user_info = _verify_portal_token(token) if user_info is None: return None, response( 403, {"error": "Invalid or unauthorized portal account"} ) else: if not _google_auth_configured(): return None, response(403, {"error": "Authentication not configured"}) user_info, status = _verify_google_token(token) if status == "unavailable": return None, _authentication_service_unavailable() if user_info is None: return None, response( 403, {"error": "Invalid or unauthorized Google account"} ) if user_info["email"].lower() not in _get_admin_emails(): return None, response(403, {"error": "Admin access required"}) return user_info, None def lambda_handler(event, context): method = event.get("requestContext", {}).get("http", {}).get("method", "GET") path = event.get("rawPath", "") if "/api/menu/" in path and method == "GET": return handle_menu(event) if path == "/api/publish/settings" and method == "GET": return handle_publish_settings() if path == "/api/publish/menu" and method == "POST": return handle_publish_menu(event) if "/admin/orders" in path: if method == "DELETE": return handle_admin_delete(event) if method == "PUT": return handle_admin_update(event) return handle_admin_orders(event) if "/admin/summary-pdf" in path: return handle_admin_summary_pdf(event) if "/form-status/" in path: return handle_form_status(event) if "/roster" in path: return handle_roster() if method == "POST": return handle_submit(event) return response(405, {"error": "Method not allowed"}) def handle_menu(event): """Return the published menu in the portal's public MenuPayload shape.""" requested_week = (event.get("pathParameters") or {}).get("week", "") if requested_week == "current": week = current_week() elif re.fullmatch(r"\d{4}-W\d{2}", requested_week): week = requested_week else: return response(400, {"error": "week path param must be current or YYYY-WNN"}) menu = get_menu(week) if menu is None: return response(404, {"error": "No menu available for this week"}) meals = menu.get("meals") or [] settings = get_settings() return response( 200, { "week": week, "form_status": menu.get("form_status", "closed"), "scraped_at": menu.get("scraped_at"), "menu_url": menu.get("menu_url"), "meal_count": int(menu.get("meal_count", len(meals))), "meals": meals, "bulk_discount_percent": float(settings.get("bulk_discount_percent", 0)), "company_subsidy_percent": float( settings.get("company_subsidy_percent", 0) ), "order_deadline": settings.get("order_deadline") or "Thursday at 11:59 PM", }, ) def handle_publish_settings(): """Return only the pricing fields needed by the weekly-menu workflow.""" settings = get_settings() return response( 200, { "bulk_discount_percent": float(settings.get("bulk_discount_percent", 0)), "company_subsidy_percent": float( settings.get("company_subsidy_percent", 0) ), }, ) def handle_publish_menu(event): """Persist a scraped menu for the current Eastern-time week.""" try: body = json.loads(event.get("body", "{}")) except json.JSONDecodeError: return response(400, {"error": "Invalid JSON"}) if not isinstance(body, dict): return response(400, {"error": "Request body must be a JSON object"}) meals = body.get("meals") if not isinstance(meals, list) or not meals: return response(400, {"error": "At least one meal is required"}) for meal in meals: price = meal.get("price") if isinstance(meal, dict) else None if ( not isinstance(meal, dict) or not str(meal.get("name", "")).strip() or isinstance(price, bool) or not isinstance(price, (int, float)) or price < 0 ): return response( 400, {"error": "Each meal requires a name and non-negative price"} ) week = current_week() menu = { "scraped_at": body.get("scraped_at"), "menu_url": body.get("menu_url"), "meal_count": len(meals), "meals": meals, } put_menu(week, menu) logger.info("Published %s meals for %s", len(meals), week) return response( 200, {"status": "published", "week": week, "meal_count": len(meals)}, ) def handle_admin_orders(event): user, err = _verify_admin(event) if err: return err qs = event.get("queryStringParameters") or {} week = qs.get("week") if not week: return response(200, {"weeks": list_weeks()}) orders = get_orders(week) order_list = [] for order in orders: items = [] for item in order.get("items", []): items.append( { "name": item.get("name", ""), "quantity": int(item.get("quantity", 0)), "retail_price": float(item.get("retail_price", 0)), "price": float(item.get("price", 0)), "subtotal": float(item.get("subtotal", 0)), } ) order_list.append( { "employee_name": order.get("employee_name", ""), "employee_email": order.get("employee_email", ""), "items": items, "total": float(order.get("total", 0)), "submitted_at": order.get("submitted_at", ""), } ) order_list.sort(key=lambda o: o["employee_name"]) return response( 200, { "week": week, "orders": order_list, "total_employees": len(order_list), "grand_total": round(sum(o["total"] for o in order_list), 2), }, ) def handle_admin_summary_pdf(event): """Return a short-lived presigned URL for the week's summary PDF.""" user, err = _verify_admin(event) if err: return err qs = event.get("queryStringParameters") or {} week = qs.get("week", "") if not re.match(r"^\d{4}-W\d{2}$", week): return response(400, {"error": "week query param must be YYYY-WNN"}) summary = get_summary(week) if summary is None: return response(404, {"error": "No summary PDF: week not found or not closed"}) pdf_key = summary.get("weekly_summary_pdf_s3_key", "") # The key comes from a DynamoDB record; only presign keys matching the # shape aggregate_orders writes, so a tampered record can't expose other # report files (e.g. payroll CSVs). if not re.match( r"^reports/\d{4}-W\d{2}/weekly-summary-\d{4}-W\d{2}\.pdf$", pdf_key ): if pdf_key: logger.error("Unexpected summary PDF key shape for %s: %s", week, pdf_key) return response(500, {"error": "Internal error"}) return response(404, {"error": "No summary PDF available for this week"}) url = _s3.generate_presigned_url( "get_object", Params={"Bucket": os.environ["REPORTS_BUCKET"], "Key": pdf_key}, ExpiresIn=PRESIGNED_URL_TTL_SECONDS, ) logger.info("Admin %s requested summary PDF for %s", user["email"], week) return response(200, {"week": week, "url": url}) def handle_admin_delete(event): user, err = _verify_admin(event) if err: return err qs = event.get("queryStringParameters") or {} week = qs.get("week", "") email = qs.get("email", "") if not week or not email: return response(400, {"error": "week and email query params are required"}) slug = email.lower() existing = get_order(week, slug) if not existing: return response(404, {"error": "Order not found"}) delete_order(week, slug) logger.info("Admin %s deleted order for %s in %s", user["email"], email, week) return response(200, {"status": "deleted", "week": week, "email": email}) def handle_admin_update(event): user, err = _verify_admin(event) if err: return err try: body = json.loads(event.get("body", "{}")) except json.JSONDecodeError: return response(400, {"error": "Invalid JSON"}) week = body.get("week", "") email = body.get("email", "") new_items = body.get("items", []) if not week or not email: return response(400, {"error": "week and email are required"}) slug = email.lower() existing = get_order(week, slug) if not existing: return response(404, {"error": "Order not found"}) filtered = [i for i in new_items if i.get("quantity", 0) > 0] if not filtered: return response( 400, {"error": "At least one item with quantity > 0 is required"} ) official_retail = _official_menu_retail_by_name(week) if not official_retail: return response(503, {"error": "Menu temporarily unavailable"}) TWO_PLACES = Decimal("0.01") bulk_pct, subsidy_pct = _get_discount_settings() bulk_mult = Decimal("1") - ( max(Decimal("0"), min(Decimal("100"), bulk_pct)) / Decimal("100") ) subsidy_mult = Decimal("1") - ( max(Decimal("0"), min(Decimal("100"), subsidy_pct)) / Decimal("100") ) for item in filtered: meal_name = (item.get("name") or "").strip() if meal_name not in official_retail: return response(400, {"error": f"'{meal_name}' not on this week's menu"}) retail = official_retail[meal_name] qty = Decimal(str(item["quantity"])) bulk_price = (retail * bulk_mult).quantize(TWO_PLACES, rounding=ROUND_HALF_UP) emp_price = (bulk_price * subsidy_mult).quantize( TWO_PLACES, rounding=ROUND_HALF_UP ) subtotal = (emp_price * qty).quantize(TWO_PLACES, rounding=ROUND_HALF_UP) item["retail_price"] = float(retail) item["bulk_price"] = float(bulk_price) item["price"] = float(emp_price) item["subtotal"] = float(subtotal) total = float( sum(Decimal(str(i["subtotal"])) for i in filtered).quantize( TWO_PLACES, rounding=ROUND_HALF_UP ) ) order_data = { "employee_name": existing.get("employee_name", ""), "employee_email": existing.get("employee_email", ""), "submitted_at": existing.get("submitted_at", ""), "items": filtered, "total": total, } put_order(week, slug, order_data) logger.info("Admin %s updated order for %s in %s", user["email"], email, week) return response( 200, {"status": "updated", "week": week, "email": email, "total": total} ) def handle_form_status(event): week = event.get("pathParameters", {}).get("week", current_week()) status = get_form_status(week) result = {"week": week, "status": status} if status == "closed": # Next Monday 8:00 AM Eastern: use calendar date math + combine() so reopen_at # stays on the correct civil Monday across DST (timedelta(days=n) is always 24n hours). now_et = _eastern_now() today = now_et.date() weekday = today.weekday() # Monday=0 ... Sunday=6 days_until_monday = (7 - weekday) % 7 if days_until_monday == 0 and now_et.hour >= 8: # If today is Monday past 8am, next Monday is 7 days away days_until_monday = 7 reopen_date = today + timedelta(days=days_until_monday) next_monday = _dt.datetime.combine(reopen_date, _dt.time(8, 0), tzinfo=EASTERN) result["reopen_at"] = int(next_monday.timestamp()) return response(200, result) def handle_roster(): roster = get_roster() employees = [{"name": e["name"], "email": e["email"]} for e in roster] return response(200, {"employees": employees}) def handle_submit(event): try: body = json.loads(event.get("body", "{}")) except json.JSONDecodeError: return response(400, {"error": "Invalid JSON"}) portal_token = _extract_bearer_token(event) if portal_token: user_info = _verify_portal_token(portal_token) if user_info is None: return response(403, {"error": "Invalid or unauthorized portal account"}) else: # The Lambda is the cloud submission path, so one of the two verified # identity providers is required. Local/manual submissions remain in Flask. google_token = body.get("google_id_token") try: client_id = _get_google_client_id() except Exception as exc: logger.error("SSM fetch failed for Google client ID: %s", exc) return _authentication_service_unavailable() if not client_id: logger.error("Google client ID is unavailable") return _authentication_service_unavailable() if not google_token: return response(403, {"error": "Google authentication is required"}) user_info, verify_status = _verify_google_token(google_token) if verify_status == "unavailable": return _authentication_service_unavailable() if user_info is None: return response(403, {"error": "Invalid or unauthorized Google account"}) name = user_info["name"] email = user_info["email"] items = body.get("items", []) if not name: return response(400, {"error": "Employee name is required"}) if not email: return response(400, {"error": "Employee email is required"}) if not items or not any(i.get("quantity", 0) > 0 for i in items): return response(400, {"error": "Please select at least one meal"}) week = current_week() status = get_form_status(week) is_admin_user = email.lower() in _get_admin_emails() if status == "closed" and not is_admin_user: return response(410, {"error": "Orders are closed for this week"}) if status == "not_found": return response(404, {"error": "No menu available for this week"}) filtered_items = [i for i in items if i.get("quantity", 0) > 0] official_retail = _official_menu_retail_by_name(week) if not official_retail: logger.error("Week %s: menu has no priced meals; refusing order", week) return response(503, {"error": "Menu temporarily unavailable"}) for item in filtered_items: meal_name = (item.get("name") or "").strip() if meal_name not in official_retail: return response( 400, {"error": "One or more meals are not on this week's menu"}, ) # --- Price calculation using Decimal for financial precision --- # Rounding approach (two-step intermediate rounding): # 1. bulk_price = retail * bulk_mult, rounded to 2 decimal places # 2. emp_price = bulk_price * subsidy_mult, rounded to 2 decimal places # The frontend should match this two-step rounding to avoid discrepancies. TWO_PLACES = Decimal("0.01") bulk_pct, subsidy_pct = _get_discount_settings() bulk_pct = max(Decimal("0"), min(Decimal("100"), bulk_pct)) subsidy_pct = max(Decimal("0"), min(Decimal("100"), subsidy_pct)) bulk_mult = Decimal("1") - (bulk_pct / Decimal("100")) subsidy_mult = Decimal("1") - (subsidy_pct / Decimal("100")) for item in filtered_items: meal_name = (item.get("name") or "").strip() retail = official_retail[meal_name] qty = Decimal(str(item.get("quantity", 0))) # Step 1: apply bulk discount and round bulk_price = (retail * bulk_mult).quantize(TWO_PLACES, rounding=ROUND_HALF_UP) # Step 2: apply company subsidy and round emp_price = (bulk_price * subsidy_mult).quantize( TWO_PLACES, rounding=ROUND_HALF_UP ) subtotal = (emp_price * qty).quantize(TWO_PLACES, rounding=ROUND_HALF_UP) # Convert back to float for JSON serialization item["retail_price"] = float(retail) item["bulk_price"] = float(bulk_price) item["price"] = float(emp_price) item["subtotal"] = float(subtotal) total = float( sum(Decimal(str(i["subtotal"])) for i in filtered_items).quantize( TWO_PLACES, rounding=ROUND_HALF_UP ) ) slug = email.lower() order_data = { "employee_name": name, "employee_email": email, "submitted_at": _eastern_now().isoformat(), "items": filtered_items, "total": total, } put_order(week, slug, order_data) # Slack notification is best-effort — order is already persisted above, # so we return success to the user even if this invocation fails. try: _lambda.invoke( FunctionName=os.environ["SLACK_NOTIFIER_ARN"], InvocationType="Event", Payload=json.dumps( { "event": "order_confirmed", "employee_name": name, "employee_email": email, "items": filtered_items, "total": order_data["total"], "week": week, }, default=float, ), ) except Exception as exc: logger.error("Slack notifier invocation failed (order already saved): %s", exc) return response( 200, { "status": "ok", "message": f"Order saved for {name}", "total": order_data["total"], }, ) def response(status_code: int, body: dict) -> dict: return { "statusCode": status_code, "headers": {"Content-Type": "application/json"}, "body": json.dumps(body, default=float), }