import re from datetime import datetime, timezone from typing import List, Optional from uuid import UUID from sqlalchemy import delete, select from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import selectinload from src.apps.memorials.models.memorial import Memorial from src.apps.memorials.models.timeline_event import MemorialTimelineEvent from src.apps.memorials.schemas.requests import ( CreateMemorialRequest, TimelineEventRequest, ) from src.apps.records.models.record import Record from src.apps.site_admin.services.audit_service import AuditService from src.core.exceptions import ConflictError, NotFoundError _DEFAULT_VISIBILITY_CONFIG = { "show_dob": True, "show_dod": True, "show_family_contact": False, "allow_tributes": True, } def _build_slug(name: str) -> str: """Convert an arbitrary display name to a URL-safe slug.""" return re.sub(r"[^a-z0-9]+", "-", name.lower()).strip("-") class MemorialService: @staticmethod async def create_for_record( db: AsyncSession, record_id: UUID, tenant_id: UUID, data: CreateMemorialRequest, current_user, request, ) -> Memorial: """Create a new Memorial for an existing Record. Raises: NotFoundError: if the Record does not exist or has been soft-deleted. ConflictError: if a Memorial already exists for this Record. """ # 1. Verify record exists and is not soft-deleted record_stmt = select(Record).where( Record.id == record_id, Record.tenant_id == tenant_id, Record.deleted_at.is_(None), ) record: Optional[Record] = (await db.execute(record_stmt)).scalar_one_or_none() if record is None: raise NotFoundError("Record not found") # 2. Check no memorial already exists for this record existing_stmt = select(Memorial).where( Memorial.record_id == record_id, Memorial.tenant_id == tenant_id, ) existing: Optional[Memorial] = (await db.execute(existing_stmt)).scalar_one_or_none() if existing is not None: raise ConflictError("A memorial already exists for this record") # 3. Auto-generate a unique slug within the tenant base_name = data.display_name or f"{record.first_name}-{record.last_name}" base_slug = _build_slug(base_name) slug = base_slug counter = 2 while True: collision_stmt = select(Memorial).where( Memorial.tenant_id == tenant_id, Memorial.slug == slug, ) collision = (await db.execute(collision_stmt)).scalar_one_or_none() if collision is None: break slug = f"{base_slug}-{counter}" counter += 1 # 4. Default visibility_config if not supplied visibility_config = data.visibility_config or dict(_DEFAULT_VISIBILITY_CONFIG) # 5. Resolve published_at published_at: Optional[datetime] = ( datetime.now(timezone.utc) if data.is_published else None ) # 6. Instantiate the Memorial memorial = Memorial( tenant_id=tenant_id, record_id=record_id, slug=slug, display_name=data.display_name, headstone_inscription=data.headstone_inscription, biography_text=data.biography_text, video_url=data.video_url, visibility_config=visibility_config, is_published=data.is_published, published_at=published_at, ) # 7. Persist and obtain generated PK db.add(memorial) await db.flush() # 8. Insert timeline events when provided if data.timeline_events: for idx, event in enumerate(data.timeline_events): timeline_event = MemorialTimelineEvent( tenant_id=tenant_id, memorial_id=memorial.id, event_date=datetime(event.year, 1, 1, tzinfo=timezone.utc), title=event.title, description=event.description, sort_order=idx, ) db.add(timeline_event) await db.flush() # 9. Audit trail await AuditService.log( db, "memorial", memorial.id, "created", current_user, request, old_value=None, new_value={ "display_name": memorial.display_name, "slug": memorial.slug, "is_published": memorial.is_published, }, ) # 10. Reload with all relationships eager-loaded so the caller gets a # fully-populated object without triggering lazy-load errors. reload_stmt = ( select(Memorial) .options( selectinload(Memorial.timeline_events), selectinload(Memorial.tributes), ) .where(Memorial.id == memorial.id) ) memorial = (await db.execute(reload_stmt)).scalar_one() return memorial @staticmethod async def replace_timeline( db: AsyncSession, memorial_id: UUID, tenant_id: UUID, events: List[TimelineEventRequest], current_user, request, ) -> List[MemorialTimelineEvent]: """Replace all timeline events for a Memorial atomically. Raises: NotFoundError: if the Memorial does not exist for the tenant. """ # 1. Verify memorial exists memorial_stmt = select(Memorial).where( Memorial.id == memorial_id, Memorial.tenant_id == tenant_id, ) memorial: Optional[Memorial] = (await db.execute(memorial_stmt)).scalar_one_or_none() if memorial is None: raise NotFoundError("Memorial not found") # 2. Capture old events for the audit log old_events_stmt = select(MemorialTimelineEvent).where( MemorialTimelineEvent.memorial_id == memorial_id, MemorialTimelineEvent.tenant_id == tenant_id, ) old_event_rows = (await db.execute(old_events_stmt)).scalars().all() old_events_snapshot = [ { "year": row.event_date.year if row.event_date else None, "title": row.title, "description": row.description, } for row in old_event_rows ] # 3. Delete all existing timeline events for this memorial delete_stmt = delete(MemorialTimelineEvent).where( MemorialTimelineEvent.memorial_id == memorial_id, MemorialTimelineEvent.tenant_id == tenant_id, ) await db.execute(delete_stmt) # 4. Insert the replacement events new_event_objects: List[MemorialTimelineEvent] = [] for idx, event in enumerate(events): new_event = MemorialTimelineEvent( tenant_id=tenant_id, memorial_id=memorial_id, event_date=datetime(event.year, 1, 1, tzinfo=timezone.utc), title=event.title, description=event.description, sort_order=idx, ) db.add(new_event) new_event_objects.append(new_event) if new_event_objects: await db.flush() # 5. Audit log new_events_snapshot = [ { "year": event.year, "title": event.title, "description": event.description, } for event in events ] await AuditService.log( db, "memorial", memorial_id, "timeline_updated", current_user, request, old_value={"events": old_events_snapshot}, new_value={"events": new_events_snapshot}, ) # 6. Return new events ordered by sort_order result_stmt = ( select(MemorialTimelineEvent) .where( MemorialTimelineEvent.memorial_id == memorial_id, MemorialTimelineEvent.tenant_id == tenant_id, ) .order_by(MemorialTimelineEvent.sort_order) ) new_events = (await db.execute(result_stmt)).scalars().all() return list(new_events)