"""Service Scheduling Phase 2 (INDL-34) — business logic. Security posture (see PRD §Security): * Every query filters `tenant_id == current tenant` AND (services/docs) `deleted_at IS NULL`. Cross-tenant / missing → NotFoundError (404), never 403. * A cross-tenant probe emits a structured `cross_tenant_access_attempt` event. * Status FSM: awaiting_family_confirm -> confirmed ONLY (PATCH). Idempotent. * Documents: MIME sniffed via python-magic, 25 MB cap, path-traversal guard, server-built S3 key `scheduling/{tenant}/{service}/{uuid4}.{ext}`. * Presigned URLs generated on-demand, ExpiresIn=900, attachment disposition; never stored, never logged. * Activity log metadata built from a trusted allow-list only, inside the same transaction as the state change. """ from __future__ import annotations import asyncio import logging import re import uuid from datetime import date, datetime, time as dt_time, timedelta, timezone from typing import Any, List, Optional from uuid import UUID import boto3 from arq.jobs import Job as ArqJob from sqlalchemy import and_, delete, func, select from sqlalchemy.ext.asyncio import AsyncSession from src.apps.auth.models.user import User from src.apps.crew_members.models.crew_member import CrewMember from src.apps.memorials.models.memorial import Memorial from src.apps.scheduling.models.scheduling_activity_log import SchedulingActivityLog from src.apps.scheduling.models.scheduling_document import SchedulingDocument from src.apps.scheduling.models.scheduling_run_sheet_item import SchedulingRunSheetItem from src.apps.scheduling.models.service_crew_assignment import ServiceCrewAssignment from src.apps.scheduling.models.service_event import ServiceEvent from src.apps.scheduling.schemas.scheduling import ServiceCreate, ServiceUpdate from src.apps.scheduling.services.scheduling_pdf import build_service_detail_pdf_bytes from src.core.config import settings from src.core.constants import ROLE_HIERARCHY, UserRole from src.core.exceptions import ( ForbiddenError, NotFoundError, PayloadTooLargeError, UnprocessableError, UnsupportedMediaTypeError, ValidationError, ) logger = logging.getLogger(__name__) # ── File upload policy (S-04..S-06 / SAC-04/05) ──────────────────────────────── MAX_FILE_SIZE_BYTES = 25 * 1024 * 1024 # 25 MB # Sniffed MIME allow-list -> canonical extension used for the S3 key. MIME_TO_EXT = { "application/pdf": "pdf", "image/jpeg": "jpg", "image/png": "png", "image/tiff": "tiff", "application/vnd.openxmlformats-officedocument.wordprocessingml.document": "docx", } ALLOWED_MIME_TYPES = set(MIME_TO_EXT.keys()) # DOCX is a ZIP container — libmagic commonly reports one of these for it. _DOCX_SNIFF_ALIASES = { "application/vnd.openxmlformats-officedocument.wordprocessingml.document", "application/zip", "application/octet-stream", } _PRESIGN_TTL_SECONDS = 900 # 15 min (SAC-08 / S-10) # Filename rejection: path separators, traversal, null byte, url-encoded traversal. _BAD_FILENAME_RE = re.compile(r"(\.\.|/|\\|%2e%2e|%2f|%5c|\x00)", re.IGNORECASE) # ── Status FSM ───────────────────────────────────────────────────────────────── _STATUS_TRANSITION_MSG = ( "Status can only be updated from 'awaiting_family_confirm' to 'confirmed'." ) def _make_s3_client(): """Create a boto3 S3 client per-request (no credential caching).""" return boto3.client( "s3", region_name=settings.S3_DOCUMENTS_REGION, aws_access_key_id=settings.AWS_ACCESS_KEY_ID, aws_secret_access_key=settings.AWS_SECRET_ACCESS_KEY, ) def _sniff_mime(content: bytes) -> Optional[str]: """Return the sniffed MIME type via python-magic, or None if unavailable.""" try: import magic # type: ignore except Exception: # pragma: no cover - libmagic not installed logger.warning("python-magic/libmagic unavailable; cannot sniff MIME") return None try: return magic.from_buffer(content, mime=True) except Exception: # pragma: no cover return None def _sanitize_log_value(value: Any) -> Any: """Coerce a value into a short, safe scalar for the activity-log metadata.""" if value is None or isinstance(value, (bool, int, float)): return value text = str(value) if len(text) > 120: text = text[:117] + "..." return text class SchedulingService: def __init__(self, db: AsyncSession): self.db = db # ── SVC-10: internal audit writer ────────────────────────────────────────── async def log_event( self, *, service_id: UUID, tenant_id: UUID, actor_id: Optional[UUID], event_type: str, description: str, from_status: Optional[str] = None, to_status: Optional[str] = None, metadata: Optional[dict] = None, ) -> SchedulingActivityLog: """Append a row to scheduling_activity_log (same txn as the caller). `metadata` is expected to already contain only trusted, allow-listed keys with sanitised scalar values (built by the calling method). """ entry = SchedulingActivityLog( tenant_id=tenant_id, service_id=service_id, actor_id=actor_id, event_type=event_type, from_status=from_status, to_status=to_status, description=description, log_metadata=metadata, ) self.db.add(entry) await self.db.flush() return entry # ── internal fetch helpers ────────────────────────────────────────────────── async def _fetch_service( self, tenant_id: UUID, service_id: UUID, actor_id: UUID ) -> ServiceEvent: """Fetch a non-deleted service scoped to tenant, else 404. On a UUID that exists but belongs to another tenant, emit a structured cross-tenant probe event before raising 404 (A09 / SAC-11). """ stmt = select(ServiceEvent).where( ServiceEvent.id == service_id, ServiceEvent.tenant_id == tenant_id, ServiceEvent.deleted_at.is_(None), ) event = (await self.db.execute(stmt)).scalar_one_or_none() if event is None: await self._maybe_log_cross_tenant(tenant_id, service_id, actor_id) raise NotFoundError("Service not found") return event async def _maybe_log_cross_tenant( self, tenant_id: UUID, service_id: UUID, actor_id: UUID ) -> None: exists = ( await self.db.execute( select(ServiceEvent.tenant_id).where(ServiceEvent.id == service_id) ) ).scalar_one_or_none() if exists is not None and str(exists) != str(tenant_id): logger.warning( "security_event", extra={ "event": "cross_tenant_access_attempt", "user_id": str(actor_id), "target_service_id": str(service_id), "request_tenant_id": str(tenant_id), "actual_tenant_id": str(exists), }, ) async def _validate_crew_ids( self, tenant_id: UUID, crew_ids: List[UUID] ) -> None: """Every crew_member_id must belong to this tenant and be non-deleted. Unknown/foreign IDs → 422 invalid_crew_member (API3 / SAC-07).""" if not crew_ids: return unique_ids = list(set(crew_ids)) rows = ( await self.db.execute( select(CrewMember.id).where( CrewMember.id.in_(unique_ids), CrewMember.tenant_id == tenant_id, CrewMember.deleted_at.is_(None), ) ) ).scalars().all() found = {str(r) for r in rows} missing = [str(cid) for cid in unique_ids if str(cid) not in found] if missing: raise ValidationError( message="One or more crew members are invalid for this tenant.", errors=[{"reason": "invalid_crew_member", "ids": missing}], ) async def _replace_crew( self, tenant_id: UUID, service_id: UUID, crew_ids: List[UUID] ) -> None: await self.db.execute( delete(ServiceCrewAssignment).where( ServiceCrewAssignment.service_id == service_id ) ) for cid in set(crew_ids): self.db.add( ServiceCrewAssignment( tenant_id=tenant_id, service_id=service_id, crew_member_id=cid ) ) async def _replace_run_sheet( self, tenant_id: UUID, service_id: UUID, items ) -> None: await self.db.execute( delete(SchedulingRunSheetItem).where( SchedulingRunSheetItem.service_id == service_id ) ) for item in items: self.db.add( SchedulingRunSheetItem( tenant_id=tenant_id, service_id=service_id, sort_order=item.sort_order, time_label=item.time_label, details=item.details, ) ) # ── INDL-38: family notification helpers ──────────────────────────────────── # Job payloads carry ONLY service_id + tenant_id (A04/API2) — the ARQ job # re-fetches the row and re-validates tenant ownership itself (SEC-AC-01). @staticmethod def _service_datetime(event: ServiceEvent) -> Optional[datetime]: if not event.scheduled_date: return None return datetime.combine( event.scheduled_date, event.scheduled_time or dt_time(0, 0), tzinfo=timezone.utc ) async def _enqueue_confirmation(self, arq_redis, tenant_id: UUID, service_id: UUID) -> None: if arq_redis is None: return try: await arq_redis.enqueue_job( "send_scheduling_confirmation", service_id=str(service_id), tenant_id=str(tenant_id), ) except Exception: logger.warning("scheduling.confirmation_enqueue_failed service=%s", service_id) async def _schedule_reminder( self, arq_redis, tenant_id: UUID, service_id: UUID, service_dt: datetime ) -> Optional[str]: if arq_redis is None: return None reminder_at = service_dt - timedelta(hours=24) try: job = await arq_redis.enqueue_job( "send_scheduling_reminder", service_id=str(service_id), tenant_id=str(tenant_id), _defer_until=reminder_at, ) return job.job_id if job else None except Exception: logger.warning("scheduling.reminder_enqueue_failed service=%s", service_id) return None async def _cancel_reminder(self, arq_redis, job_id: Optional[str]) -> None: if arq_redis is None or not job_id: return try: await ArqJob(job_id, arq_redis).abort() except Exception: logger.warning("scheduling.reminder_cancel_failed job_id=%s", job_id) async def _apply_create_notifications( self, arq_redis, tenant_id: UUID, event: ServiceEvent, data: ServiceCreate ) -> None: """AC-01/AC-03/AC-09/AC-10: only for a newly-scheduled (non-draft) service.""" if data.status != "awaiting_family_confirm": return if data.notify_family_confirmation: await self._enqueue_confirmation(arq_redis, tenant_id, event.id) if data.notify_family_reminder_24h: service_dt = self._service_datetime(event) if service_dt is not None: job_id = await self._schedule_reminder(arq_redis, tenant_id, event.id, service_dt) if job_id: event.reminder_job_id = job_id await self.db.flush() async def _apply_update_notifications( self, arq_redis, tenant_id: UUID, event: ServiceEvent, *, old_date, old_time, old_reminder_job_id: Optional[str], ) -> None: """AC-05: cancel + re-enqueue the reminder if date/time changed. Draft services never notify (AC-09). If the reminder flag was turned off, the stale job is cancelled proactively (the job also re-checks the flag at fire time as a safety net). If the flag was turned on after creation and no job exists yet, schedule one now. """ if event.status == "draft": return datetime_changed = (event.scheduled_date != old_date) or ( event.scheduled_time != old_time ) if not event.notify_family_reminder_24h: if old_reminder_job_id: await self._cancel_reminder(arq_redis, old_reminder_job_id) event.reminder_job_id = None await self.db.flush() return if datetime_changed and old_reminder_job_id: await self._cancel_reminder(arq_redis, old_reminder_job_id) event.reminder_job_id = None if event.reminder_job_id is None: service_dt = self._service_datetime(event) if service_dt is not None: job_id = await self._schedule_reminder(arq_redis, tenant_id, event.id, service_dt) if job_id: event.reminder_job_id = job_id if event.reminder_job_id != old_reminder_job_id: await self.db.flush() # ── SVC-01: list / kanban ──────────────────────────────────────────────────── async def list( self, tenant_id: UUID, *, status: Optional[str] = None, service_type: Optional[str] = None, date_from=None, date_to=None, page: int = 1, page_size: int = 20, kanban: bool = False, ): """Return either (items, total) for list view, or a grouped dict for kanban.""" base = [ ServiceEvent.tenant_id == tenant_id, ServiceEvent.deleted_at.is_(None), ] if service_type: base.append(ServiceEvent.service_type == service_type) if date_from is not None: base.append(ServiceEvent.scheduled_date >= date_from) if date_to is not None: base.append(ServiceEvent.scheduled_date <= date_to) if kanban: buckets = { "draft": [], "awaiting_family_confirm": [], "confirmed": [], } all_rows: List[ServiceEvent] = [] bucket_rows: dict = {} for key in buckets: rows = ( await self.db.execute( select(ServiceEvent) .where(and_(*base, ServiceEvent.status == key)) .order_by( ServiceEvent.scheduled_date, ServiceEvent.scheduled_time ) .limit(200) # API4 cap per bucket ) ).scalars().all() bucket_rows[key] = rows all_rows.extend(rows) # C2: batch-load crew names for every returned service in ONE query. crew_map = await self._crew_names_map(tenant_id, [r.id for r in all_rows]) for key, rows in bucket_rows.items(): buckets[key] = [self._list_item(r, crew_map) for r in rows] return buckets conditions = list(base) if status: conditions.append(ServiceEvent.status == status) total = ( await self.db.execute( select(func.count(ServiceEvent.id)).where(and_(*conditions)) ) ).scalar_one() offset = (page - 1) * page_size rows = ( await self.db.execute( select(ServiceEvent) .where(and_(*conditions)) .order_by(ServiceEvent.scheduled_date, ServiceEvent.scheduled_time) .offset(offset) .limit(page_size) ) ).scalars().all() # C2: batch-load crew names for every returned service in ONE query. crew_map = await self._crew_names_map(tenant_id, [r.id for r in rows]) items = [self._list_item(r, crew_map) for r in rows] return items, total async def _crew_names_map( self, tenant_id: UUID, service_ids: List[UUID] ) -> dict: """{service_id: [crew names]} for all given services in one grouped query.""" if not service_ids: return {} rows = ( await self.db.execute( select(ServiceCrewAssignment.service_id, CrewMember.name) .join( CrewMember, CrewMember.id == ServiceCrewAssignment.crew_member_id, ) .where( ServiceCrewAssignment.service_id.in_(list(set(service_ids))), ServiceCrewAssignment.tenant_id == tenant_id, CrewMember.deleted_at.is_(None), ) .order_by(CrewMember.name.asc()) ) ).all() result: dict = {} for service_id, name in rows: result.setdefault(service_id, []).append(name) return result def _list_item(self, event: ServiceEvent, crew_map: dict) -> dict: return { "id": event.id, "service_type": event.service_type, "decedent_name": event.decedent_name, "record_id": event.record_id, "date": event.scheduled_date, "start_time": event.scheduled_time, "status": event.status, "family_contact_name": event.family_contact_name, "plot_location": event.plot_location, "crew": crew_map.get(event.id, []), "officiant": event.officiant, } # ── SVC-11: weekly calendar (INDL-36) ──────────────────────────────────────── @staticmethod def _normalize_service_type(service_type: Optional[str]) -> Optional[str]: """Map the legacy ``memorial`` value onto the canonical ``memorial_service``.""" if service_type == "memorial": return "memorial_service" return service_type async def _week_events( self, tenant_id: UUID, week_start: date, event_type: str ) -> tuple[List[ServiceEvent], dict]: """Fetch non-draft services in [week_start, week_start+7d) + their crew map. Fail-closed: tenant_id must be present (never a global query).""" assert tenant_id is not None, "tenant_id is required for a week query" week_end = week_start + timedelta(days=7) conditions = [ ServiceEvent.tenant_id == tenant_id, ServiceEvent.deleted_at.is_(None), ServiceEvent.status != "draft", ServiceEvent.scheduled_date >= week_start, ServiceEvent.scheduled_date < week_end, ] if event_type and event_type != "all": # ``memorial_service`` is the canonical value, but legacy rows may # still carry the raw ``memorial`` value (see _normalize_service_type), # so match both when filtering by memorial_service. if event_type == "memorial_service": conditions.append( ServiceEvent.service_type.in_(["memorial_service", "memorial"]) ) else: conditions.append(ServiceEvent.service_type == event_type) rows = ( await self.db.execute( select(ServiceEvent) .where(and_(*conditions)) .order_by(ServiceEvent.scheduled_date, ServiceEvent.scheduled_time) .limit(500) # API4 cap — a week can't reasonably exceed this ) ).scalars().all() crew_map = await self._crew_names_map(tenant_id, [r.id for r in rows]) return list(rows), crew_map async def list_week( self, tenant_id: UUID, week_start: date, event_type: str = "all" ) -> List[dict]: """PII-minimized weekly calendar rows matching ``ServiceCalendarItem``. Only non-draft services with a scheduled date inside the 7-day window are returned; ``family_contact_name`` and other PII are deliberately omitted. """ rows, crew_map = await self._week_events(tenant_id, week_start, event_type) return [self._calendar_item(r, crew_map) for r in rows] def _calendar_item(self, event: ServiceEvent, crew_map: dict) -> dict: return { "id": event.id, "service_type": self._normalize_service_type(event.service_type), "decedent_name": event.decedent_name, "plot_location": event.plot_location, "section": event.section, "date": event.scheduled_date, "start_time": event.scheduled_time, "duration_minutes": event.duration_minutes, "status": event.status, "crew_count": len(crew_map.get(event.id, [])), } async def fetch_week_for_pdf( self, tenant_id: UUID, week_start: date, event_type: str = "all" ) -> List[dict]: """Fuller week fetch for the run-sheet PDF (includes officiant/family/crew). Same tenant + deleted_at + non-draft scoping as ``list_week``.""" rows, crew_map = await self._week_events(tenant_id, week_start, event_type) return [ { "id": r.id, "service_type": self._normalize_service_type(r.service_type), "decedent_name": r.decedent_name, "family_contact_name": r.family_contact_name, "plot_location": r.plot_location, "section": r.section, "officiant": r.officiant, "date": r.scheduled_date, "start_time": r.scheduled_time, "duration_minutes": r.duration_minutes, "status": r.status, "crew": crew_map.get(r.id, []), "crew_count": len(crew_map.get(r.id, [])), } for r in rows ] def log_week_pdf_export( self, tenant_id: UUID, actor_id: UUID, week_start: date, service_count: int ) -> None: """Audit the weekly PDF export. A weekly export is not scoped to a single service, and ``scheduling_activity_log.service_id`` is NOT NULL, so this event cannot be written via ``log_event()``. It is emitted through the structured application logger instead — the same channel used for the other non-service-scoped audit events in this module (e.g. ``cross_tenant_access_attempt``). No PII beyond counts/week is recorded. """ logger.info( "audit_event", extra={ "event": "week_pdf_exported", "tenant_id": str(tenant_id), "actor_id": str(actor_id), "week_start": week_start.isoformat(), "service_count": service_count, }, ) # ── SVC-02: full detail ────────────────────────────────────────────────────── async def get_by_id( self, tenant_id: UUID, service_id: UUID, actor_id: UUID ) -> dict: event = await self._fetch_service(tenant_id, service_id, actor_id) crew = ( await self.db.execute( select(CrewMember) .join( ServiceCrewAssignment, ServiceCrewAssignment.crew_member_id == CrewMember.id, ) .where(ServiceCrewAssignment.service_id == service_id) .order_by(CrewMember.name.asc()) ) ).scalars().all() run_sheet = ( await self.db.execute( select(SchedulingRunSheetItem) .where(SchedulingRunSheetItem.service_id == service_id) .order_by(SchedulingRunSheetItem.sort_order.asc()) ) ).scalars().all() docs = ( await self.db.execute( select(SchedulingDocument) .where( SchedulingDocument.service_id == service_id, SchedulingDocument.tenant_id == tenant_id, SchedulingDocument.deleted_at.is_(None), ) .order_by(SchedulingDocument.created_at.desc()) ) ).scalars().all() activity = await self._activity_entries(tenant_id, service_id) memorial_id: Optional[UUID] = None if event.record_id is not None: memorial_id = ( await self.db.execute( select(Memorial.id).where( Memorial.record_id == event.record_id, Memorial.tenant_id == tenant_id, ) ) ).scalar_one_or_none() document_items = [] for d in docs: document_items.append( { "id": d.id, "file_name": d.file_name, "file_size": d.file_size, "mime_type": d.mime_type, "download_url": await self._presign(d.s3_key, d.file_name), "created_at": d.created_at, "uploaded_by": d.uploaded_by, } ) return { "id": event.id, "service_type": event.service_type, "decedent_name": event.decedent_name, "record_id": event.record_id, "plot_id": event.plot_id, "memorial_id": memorial_id, "date": event.scheduled_date, "start_time": event.scheduled_time, "duration_minutes": event.duration_minutes, "status": event.status, "plot_location": event.plot_location, "section": event.section, "officiant": event.officiant, "family_contact_name": event.family_contact_name, "family_contact_phone": event.family_contact_phone, "family_contact_email": event.family_contact_email, "expected_attendees": event.expected_attendees, "event_note": event.event_note, "equipment_needed": event.equipment_needed, "notes_for_crew": event.notes, "notify_family_confirmation": event.notify_family_confirmation, "notify_family_reminder_24h": event.notify_family_reminder_24h, "crew": [ { "id": c.id, "name": c.name, "email": c.email, "phone": c.phone, "is_available": c.is_available, } for c in crew ], "run_sheet_items": [ { "sort_order": r.sort_order, "time_label": r.time_label, "details": r.details, } for r in run_sheet ], "documents": document_items, "activity_log": activity, "created_by": event.created_by, "created_at": event.created_at, "updated_at": event.updated_at, } async def _presign(self, s3_key: str, file_name: str) -> Optional[str]: """On-demand presigned GET, 15-min TTL, forced attachment. Never logged.""" try: s3 = _make_s3_client() url: str = await asyncio.to_thread( lambda: s3.generate_presigned_url( "get_object", Params={ "Bucket": settings.S3_DOCUMENTS_BUCKET, "Key": s3_key, "ResponseContentDisposition": ( f'attachment; filename="{file_name}"' ), }, ExpiresIn=_PRESIGN_TTL_SECONDS, ) ) return url except Exception: # pragma: no cover - S3 misconfig must not 500 the read logger.warning("presign failed for a scheduling document") return None # ── SVC-11: single-service PDF export (INDL-37) ────────────────────────────── async def generate_pdf( self, tenant_id: UUID, service_id: UUID, current_user: User ) -> tuple[bytes, dict]: """Fetch full service detail, build the ReportLab PDF, audit, return bytes + detail.""" detail = await self.get_by_id(tenant_id, service_id, current_user.id) pdf_bytes = build_service_detail_pdf_bytes(detail) await self.log_event( service_id=service_id, tenant_id=tenant_id, actor_id=current_user.id, event_type="pdf_exported", description="Service PDF exported", ) await self.db.flush() logger.info("scheduling.pdf_exported id=%s tenant=%s", service_id, tenant_id) return pdf_bytes, detail # ── SVC-03: create ─────────────────────────────────────────────────────────── async def create( self, tenant_id: UUID, data: ServiceCreate, current_user: User, arq_redis=None ) -> dict: await self._validate_crew_ids(tenant_id, data.crew_member_ids) event = ServiceEvent( tenant_id=tenant_id, created_by=current_user.id, service_type=data.service_type, decedent_name=data.decedent_name, record_id=data.record_id, scheduled_date=data.date, scheduled_time=data.start_time, duration_minutes=data.duration_minutes, officiant=data.officiant, family_contact_name=data.family_contact_name, family_contact_phone=data.family_contact_phone, family_contact_email=data.family_contact_email, expected_attendees=data.expected_attendees, event_note=data.event_note, plot_location=data.plot_location, section=data.section, equipment_needed=data.equipment_needed, notes=data.notes_for_crew, notify_family_confirmation=data.notify_family_confirmation, notify_family_reminder_24h=data.notify_family_reminder_24h, status=data.status, ) self.db.add(event) await self.db.flush() # generate event.id await self._replace_crew(tenant_id, event.id, data.crew_member_ids) await self._replace_run_sheet(tenant_id, event.id, data.run_sheet_items) event_type = "scheduled" if data.status == "awaiting_family_confirm" else "draft_saved" await self.log_event( service_id=event.id, tenant_id=tenant_id, actor_id=current_user.id, event_type=event_type, description=f"Service created ({data.status}).", to_status=data.status, metadata={"created": True, "status": data.status}, ) await self.db.flush() logger.info("scheduling.created id=%s tenant=%s", event.id, tenant_id) await self._apply_create_notifications(arq_redis, tenant_id, event, data) return await self.get_by_id(tenant_id, event.id, current_user.id) # ── SVC-04: update (full replace, status untouched) ────────────────────────── async def update( self, tenant_id: UUID, service_id: UUID, data: ServiceUpdate, current_user: User, arq_redis=None, ) -> dict: event = await self._fetch_service(tenant_id, service_id, current_user.id) await self._validate_crew_ids(tenant_id, data.crew_member_ids) old_date, old_time, old_reminder_job_id = ( event.scheduled_date, event.scheduled_time, event.reminder_job_id, ) # Diff for the audit trail — allow-listed field NAMES + sanitized values only. field_map = { "service_type": event.service_type, "decedent_name": event.decedent_name, "record_id": event.record_id, "date": event.scheduled_date, "start_time": event.scheduled_time, "duration_minutes": event.duration_minutes, "officiant": event.officiant, "family_contact_name": event.family_contact_name, "family_contact_phone": event.family_contact_phone, "family_contact_email": event.family_contact_email, "expected_attendees": event.expected_attendees, "event_note": event.event_note, "plot_location": event.plot_location, "section": event.section, "equipment_needed": event.equipment_needed, "notes_for_crew": event.notes, "notify_family_confirmation": event.notify_family_confirmation, "notify_family_reminder_24h": event.notify_family_reminder_24h, } new_values = { "service_type": data.service_type, "decedent_name": data.decedent_name, "record_id": data.record_id, "date": data.date, "start_time": data.start_time, "duration_minutes": data.duration_minutes, "officiant": data.officiant, "family_contact_name": data.family_contact_name, "family_contact_phone": data.family_contact_phone, "family_contact_email": data.family_contact_email, "expected_attendees": data.expected_attendees, "event_note": data.event_note, "plot_location": data.plot_location, "section": data.section, "equipment_needed": data.equipment_needed, "notes_for_crew": data.notes_for_crew, "notify_family_confirmation": data.notify_family_confirmation, "notify_family_reminder_24h": data.notify_family_reminder_24h, } changed = [k for k, old in field_map.items() if new_values[k] != old] # Status is intentionally NOT touched here (PATCH /status only, A04). event.service_type = data.service_type event.decedent_name = data.decedent_name event.record_id = data.record_id event.scheduled_date = data.date event.scheduled_time = data.start_time event.duration_minutes = data.duration_minutes event.officiant = data.officiant event.family_contact_name = data.family_contact_name event.family_contact_phone = data.family_contact_phone event.family_contact_email = data.family_contact_email event.expected_attendees = data.expected_attendees event.event_note = data.event_note event.plot_location = data.plot_location event.section = data.section event.equipment_needed = data.equipment_needed event.notes = data.notes_for_crew event.notify_family_confirmation = data.notify_family_confirmation event.notify_family_reminder_24h = data.notify_family_reminder_24h await self._replace_crew(tenant_id, service_id, data.crew_member_ids) await self._replace_run_sheet(tenant_id, service_id, data.run_sheet_items) # Never log PII values; log changed field NAMES only (V7.1.3 / A03). await self.log_event( service_id=service_id, tenant_id=tenant_id, actor_id=current_user.id, event_type="updated", description=f"Service updated ({len(changed)} field(s) changed).", metadata={"changed_fields": [_sanitize_log_value(f) for f in changed]}, ) await self.db.flush() logger.info("scheduling.updated id=%s tenant=%s", service_id, tenant_id) await self._apply_update_notifications( arq_redis, tenant_id, event, old_date=old_date, old_time=old_time, old_reminder_job_id=old_reminder_job_id, ) return await self.get_by_id(tenant_id, service_id, current_user.id) # ── SVC-05: status transition ──────────────────────────────────────────────── async def update_status( self, tenant_id: UUID, service_id: UUID, new_status: str, current_user: User ) -> dict: event = await self._fetch_service(tenant_id, service_id, current_user.id) # C4 defense-in-depth: the target must be 'confirmed' regardless of the # schema validator — assert the invariant BEFORE the idempotent branch. if new_status != "confirmed": raise UnprocessableError(_STATUS_TRANSITION_MSG) # Idempotent: already confirmed → no new log, no email, 200. if event.status == "confirmed": return await self.get_by_id(tenant_id, service_id, current_user.id) if event.status != "awaiting_family_confirm": raise UnprocessableError(_STATUS_TRANSITION_MSG) prev = event.status event.status = "confirmed" await self.log_event( service_id=service_id, tenant_id=tenant_id, actor_id=current_user.id, event_type="confirmed", description="Service confirmed.", from_status=prev, to_status="confirmed", metadata={"transition": "awaiting_family_confirm->confirmed"}, ) await self.db.flush() logger.info("scheduling.confirmed id=%s tenant=%s", service_id, tenant_id) return await self.get_by_id(tenant_id, service_id, current_user.id) # ── SVC-06: soft delete ────────────────────────────────────────────────────── async def soft_delete( self, tenant_id: UUID, service_id: UUID, current_user: User ) -> None: event = await self._fetch_service(tenant_id, service_id, current_user.id) event.deleted_at = datetime.now(timezone.utc) await self.log_event( service_id=service_id, tenant_id=tenant_id, actor_id=current_user.id, event_type="deleted", description="Service deleted.", metadata={"deleted": True}, ) await self.db.flush() logger.info("scheduling.deleted id=%s tenant=%s", service_id, tenant_id) # ── SVC-07: upload document ────────────────────────────────────────────────── async def upload_document( self, tenant_id: UUID, service_id: UUID, *, file_name: str, declared_content_type: Optional[str], content: bytes, current_user: User, ) -> dict: # Ownership / tenant scope first (404 on mismatch). await self._fetch_service(tenant_id, service_id, current_user.id) # Filename hygiene (S-04) — reject traversal/separators/null bytes. if not file_name or _BAD_FILENAME_RE.search(file_name): self._log_upload_rejected("invalid_filename", declared_content_type, len(content)) raise ValidationError(message="Invalid file name.") # Size cap enforced BEFORE any S3 call (S-06 / V12.1.1). size = len(content) if size > MAX_FILE_SIZE_BYTES: self._log_upload_rejected("file_too_large", declared_content_type, size) raise PayloadTooLargeError("File exceeds the 25 MB upload limit.") if size == 0: self._log_upload_rejected("empty_file", declared_content_type, size) raise ValidationError(message="Empty file.") # Magic-byte MIME validation (S-05 / V12.2.1). sniffed = _sniff_mime(content) if sniffed is None: self._log_upload_rejected("mime_unverifiable", declared_content_type, size) raise UnsupportedMediaTypeError("Could not verify file type.") resolved = self._resolve_mime(sniffed, file_name, declared_content_type) if resolved is None: self._log_upload_rejected( "invalid_mime_type", declared_content_type, size ) raise UnsupportedMediaTypeError( "File type not allowed or does not match its declared type." ) ext = MIME_TO_EXT[resolved] s3_key = f"scheduling/{tenant_id}/{service_id}/{uuid.uuid4()}.{ext}" s3 = _make_s3_client() await asyncio.to_thread( s3.put_object, Bucket=settings.S3_DOCUMENTS_BUCKET, Key=s3_key, Body=content, ContentType=resolved, ) doc = SchedulingDocument( tenant_id=tenant_id, service_id=service_id, file_name=file_name, s3_key=s3_key, file_size=size, mime_type=resolved, uploaded_by=current_user.id, ) self.db.add(doc) await self.db.flush() await self.log_event( service_id=service_id, tenant_id=tenant_id, actor_id=current_user.id, event_type="document_uploaded", description=f"Document uploaded: {_sanitize_log_value(file_name)}.", metadata={"document_id": str(doc.id), "mime_type": resolved, "size": size}, ) await self.db.flush() logger.info("scheduling.doc_uploaded id=%s service=%s", doc.id, service_id) return { "id": doc.id, "file_name": doc.file_name, "file_size": doc.file_size, "mime_type": doc.mime_type, "download_url": await self._presign(doc.s3_key, doc.file_name), "created_at": doc.created_at, "uploaded_by": doc.uploaded_by, } def _resolve_mime( self, sniffed: str, file_name: str, declared: Optional[str] ) -> Optional[str]: """Map a sniffed MIME to a canonical allowed type, or None if disallowed. DOCX is a ZIP container, so libmagic may report application/zip or application/octet-stream — accept those only when the filename ends in .docx AND the declared type matches DOCX. """ if sniffed in ALLOWED_MIME_TYPES: # Declared type must not contradict the sniffed type (S-05). if declared and declared != sniffed and declared not in _DOCX_SNIFF_ALIASES: # Allow common browser variants for jpeg. if not (sniffed == "image/jpeg" and declared == "image/jpg"): return None return sniffed if sniffed in _DOCX_SNIFF_ALIASES and file_name.lower().endswith(".docx"): docx = "application/vnd.openxmlformats-officedocument.wordprocessingml.document" if declared in (None, docx) or declared in _DOCX_SNIFF_ALIASES: return docx return None def _log_upload_rejected( self, reason: str, mime_claimed: Optional[str], file_size: int ) -> None: logger.warning( "security_event", extra={ "event": "upload_rejected", "reason": reason, "mime_type_claimed": mime_claimed, "file_size": file_size, }, ) # ── SVC-08: delete document ────────────────────────────────────────────────── async def delete_document( self, tenant_id: UUID, service_id: UUID, doc_id: UUID, current_user: User ) -> None: # Single joined lookup: doc -> service, both tenant-scoped & non-deleted (API1/S-02). stmt = ( select(SchedulingDocument) .join( ServiceEvent, ServiceEvent.id == SchedulingDocument.service_id, ) .where( SchedulingDocument.id == doc_id, SchedulingDocument.service_id == service_id, SchedulingDocument.tenant_id == tenant_id, SchedulingDocument.deleted_at.is_(None), ServiceEvent.tenant_id == tenant_id, ServiceEvent.deleted_at.is_(None), ) ) doc = (await self.db.execute(stmt)).scalar_one_or_none() if doc is None: await self._maybe_log_cross_tenant(tenant_id, service_id, current_user.id) raise NotFoundError("Document not found") # Ownership check (API5): uploader OR >= manager. user_level = ROLE_HIERARCHY.get(UserRole(current_user.role), 0) manager_level = ROLE_HIERARCHY[UserRole.MANAGER] if str(doc.uploaded_by) != str(current_user.id) and user_level < manager_level: raise ForbiddenError("You can only delete documents you uploaded.") doc.soft_delete() await self.log_event( service_id=service_id, tenant_id=tenant_id, actor_id=current_user.id, event_type="document_deleted", description=f"Document deleted: {_sanitize_log_value(doc.file_name)}.", metadata={"document_id": str(doc.id)}, ) await self.db.flush() logger.info("scheduling.doc_deleted id=%s service=%s", doc_id, service_id) # ── SVC-09: activity log ───────────────────────────────────────────────────── async def get_activity_log( self, tenant_id: UUID, service_id: UUID, current_user: User ) -> List[dict]: # Enforce tenant scope on the parent service first (404 on mismatch). await self._fetch_service(tenant_id, service_id, current_user.id) return await self._activity_entries(tenant_id, service_id) async def _activity_entries(self, tenant_id: UUID, service_id: UUID) -> List[dict]: rows = ( await self.db.execute( select(SchedulingActivityLog, User.first_name, User.last_name) .outerjoin(User, User.id == SchedulingActivityLog.actor_id) .where( SchedulingActivityLog.service_id == service_id, SchedulingActivityLog.tenant_id == tenant_id, ) .order_by(SchedulingActivityLog.created_at.desc()) ) ).all() entries = [] for entry, first, last in rows: actor_name = None if first or last: actor_name = f"{first or ''} {last or ''}".strip() # NOTE: metadata JSONB is deliberately NOT surfaced (API8 / S-09). entries.append( { "id": entry.id, "event_type": entry.event_type, "from_status": entry.from_status, "to_status": entry.to_status, "description": entry.description, "actor_name": actor_name, "created_at": entry.created_at, } ) return entries