"""HTTP API v2 handler — Cognito-authenticated employee/admin shift API.""" from __future__ import annotations import json import logging from decimal import Decimal from typing import Any from urllib.parse import unquote import shared.sentry_init # noqa: F401 from shared.cognito import CognitoVerificationUnavailable, verify_cognito_id_token from shared.portal_ops import ActionError, snapshot from shared.schedule import ShiftSchedule from shared import portal_ops as ops logger = logging.getLogger() logger.setLevel(logging.INFO) CORS_ORIGINS = { "https://internal.seahaven.com", "https://internal.dev.seahaven.com", "http://localhost:5173", "http://localhost:4173", } class DecimalEncoder(json.JSONEncoder): def default(self, o): if isinstance(o, Decimal): return float(o) return super().default(o) def _cors_headers(event: dict) -> dict[str, str]: headers = event.get("headers") or {} origin = "" for key, value in headers.items(): if str(key).lower() == "origin": origin = "" if value is None else str(value) break allowed = origin if origin in CORS_ORIGINS else "" out = { "Content-Type": "application/json", "Vary": "Origin", } if allowed: out["Access-Control-Allow-Origin"] = allowed out["Access-Control-Allow-Headers"] = "Authorization,Content-Type" out["Access-Control-Allow-Methods"] = "GET,POST,DELETE,OPTIONS" return out def _response(event: dict, status: int, body: dict[str, Any] | None = None) -> dict: return { "statusCode": status, "headers": _cors_headers(event), "body": json.dumps({} if body is None else body, cls=DecimalEncoder, separators=(",", ":")), } def _header(event: dict, name: str) -> str: headers = event.get("headers") or {} if not isinstance(headers, dict): return "" target = name.lower() for key, value in headers.items(): if str(key).lower() == target: return "" if value is None else str(value) return "" def _route(event: dict) -> tuple[str, str]: request_context = event.get("requestContext") or {} http = request_context.get("http") or {} method = str(http.get("method") or event.get("httpMethod") or "").upper() path = str(event.get("rawPath") or http.get("path") or event.get("path") or "") route_key = event.get("routeKey") if isinstance(route_key, str) and " " in route_key: rk_method, rk_path = route_key.split(" ", 1) method = method or rk_method.upper() path = path or rk_path return method, path.rstrip("/") or "/" def _json_body(event: dict) -> dict: raw = event.get("body") if raw in (None, ""): return {} if event.get("isBase64Encoded"): import base64 raw = base64.b64decode(raw).decode("utf-8") if isinstance(raw, bytes): raw = raw.decode("utf-8") try: parsed = json.loads(raw) except json.JSONDecodeError as exc: raise ActionError(400, "INVALID_JSON", "Invalid JSON body.") from exc if not isinstance(parsed, dict): raise ActionError(400, "INVALID_JSON", "JSON body must be an object.") return parsed def _identity(event: dict) -> dict: presented = _header(event, "authorization") parts = presented.split(None, 1) if len(parts) != 2 or parts[0].lower() != "bearer" or not parts[1].strip(): raise ActionError(401, "UNAUTHORIZED", "Sign in to continue.") try: identity = verify_cognito_id_token(parts[1].strip()) except CognitoVerificationUnavailable as exc: raise ActionError(503, "AUTH_UNAVAILABLE", "Sign-in verification is unavailable.") from exc if not identity: raise ActionError(401, "UNAUTHORIZED", "Sign in to continue.") return identity def _employee(schedule: ShiftSchedule, identity: dict) -> dict: employee = schedule.get_employee_by_email(identity["email"]) if not employee: raise ActionError( 404, "UNLINKED", "Your Google account is not on the after-hours roster yet.", ) return employee def handler(event, context): try: method, path = _route(event) if method == "OPTIONS": return _response(event, 204, {}) identity = _identity(event) schedule = ShiftSchedule() if method == "GET" and path == "/api/shifts": employee = schedule.get_employee_by_email(identity["email"]) if not employee: return _response( event, 200, { "linked": False, "email": identity["email"], "isAdmin": False, }, ) params = event.get("queryStringParameters") or {} week = (params.get("week") if isinstance(params, dict) else None) or "this" return _response(event, 200, snapshot(schedule, employee, week)) employee = _employee(schedule, identity) body = _json_body(event) if method in {"POST", "PUT", "PATCH"} else {} return _dispatch(event, method, path, schedule, employee, body) except ActionError as exc: return _response(event, exc.status, {"error": {"code": exc.code, "message": exc.message}}) except Exception: logger.exception("portal api unexpected failure") return _response(event, 500, {"error": {"code": "INTERNAL", "message": "Internal error"}}) def _dispatch(event, method, path, schedule, employee, body): parts = [p for p in path.split("/") if p] if method == "POST" and path == "/api/shifts/pick": return _response( event, 200, ops.pick(schedule, employee, body.get("date", ""), body.get("shiftType")), ) if method == "POST" and path == "/api/shifts/drop": return _response( event, 200, ops.drop(schedule, employee, body.get("date", ""), body.get("shiftType")), ) if method == "POST" and path == "/api/shifts/swap": return _response( event, 200, ops.swap( schedule, employee, body.get("date", ""), body.get("targetExtension", ""), body.get("shiftType"), ), ) if method == "POST" and len(parts) == 6 and parts[:3] == ["api", "shifts", "swaps"] and parts[5] in ("accept", "decline"): return _response( event, 200, ops.respond_swap( schedule, employee, unquote(parts[3]), unquote(parts[4]), accept=parts[5] == "accept", ), ) if method == "POST" and path == "/api/shifts/admin/override": return _response( event, 200, ops.admin_override( schedule, employee, body.get("date", ""), body.get("extension", ""), body.get("shiftType"), ), ) if method == "POST" and path == "/api/shifts/admin/open": return _response( event, 200, ops.admin_open(schedule, employee, body.get("date", ""), body.get("shiftType")), ) if method == "POST" and path == "/api/shifts/admin/clear": return _response( event, 200, ops.admin_clear(schedule, employee, body.get("date", ""), body.get("shiftType")), ) if method == "POST" and path == "/api/shifts/admin/holidays": return _response( event, 200, ops.admin_holiday_add( schedule, employee, body.get("date", ""), body.get("slots"), body.get("label", ""), body.get("multiplier"), ), ) if method == "DELETE" and len(parts) == 5 and parts[:4] == ["api", "shifts", "admin", "holidays"]: return _response( event, 200, ops.admin_holiday_remove(schedule, employee, unquote(parts[4])), ) if ( method == "POST" and len(parts) == 8 and parts[:4] == ["api", "shifts", "admin", "pickups"] and parts[7] in ("approve", "deny") ): return _response( event, 200, ops.admin_pickup( schedule, employee, unquote(parts[4]), unquote(parts[5]), unquote(parts[6]), approve=parts[7] == "approve", ), ) return _response(event, 405, {"error": {"code": "METHOD", "message": "Method not allowed"}})