"""Business logic for planned leave days (calendar planning).""" from dataclasses import dataclass, field from datetime import date, datetime from uuid import UUID, uuid4 from dateutil.relativedelta import relativedelta from app.config import settings from app.entities.planned_leave import PlannedLeaveDay from app.repositories.parental_leave_repository import ParentalLeaveRepository from app.repositories.planned_leave_repository import PlannedLeaveRepository from app.services.analytics_service import AnalyticsService from app.utils.errors import DubbeldagError, NotFoundError, ValidationError # Day types that can actually be persisted as a `planned_leave_day` row. A # proposal may also contain 'free' entries (automatic weekend/red-day # non-workdays, see allocation_service.ProposedDay) purely for explainability # — those never become rows here, see `apply_proposed_days`. _PERSISTABLE_DAY_TYPES = {"sjukpenning", "lagsta", "vacation"} # --------------------------------------------------------------------------- # POST /api/plan/apply — bulk accept of a (previously proposed) set of days. # --------------------------------------------------------------------------- @dataclass(frozen=True) class ProposedDayInput: """One day out of a batch handed to `apply_proposed_days`. Deliberately a plain domain dataclass (not a Pydantic model) so the service stays independent of the wire schema in `app.models.plan` — the router is responsible for that translation. Shape mirrors `allocation_service.ProposedDay` minus the explain-only `wish_id`/`reason` fields, which carry no persistence meaning. """ date: date parent_personal_data_uuid: UUID child_id: UUID day_type: str is_dubbeldag: bool = False @dataclass(frozen=True) class DayCollision: """A date already planned differently than what the batch proposes.""" date: date parent_personal_data_uuid: UUID child_id: UUID existing_day_type: str proposed_day_type: str @dataclass(frozen=True) class Rejection: """A single reason a batch (or part of it) was refused. §3.5: mirrors `tips_service.Tip`'s `message_key` + `data` shape rather than a pre-rendered string, so `PlanApplyResponse.rejections` can be rendered in the caller's active locale instead of leaking backend- generated (previously English-only, regardless of locale) text. """ message_key: str data: dict[str, object] @dataclass(frozen=True) class PlanApplyResult: accepted: bool created: list[PlannedLeaveDay] = field(default_factory=list) created_count: int = 0 skipped_identical_count: int = 0 skipped_free_count: int = 0 replaced_count: int = 0 collisions: list[DayCollision] = field(default_factory=list) rejections: list[Rejection] = field(default_factory=list) class PlanningService: def __init__( self, repo: PlannedLeaveRepository, analytics_service: AnalyticsService, parental_leave_repo: ParentalLeaveRepository, ) -> None: self._repo = repo self._analytics = analytics_service self._parental_leave_repo = parental_leave_repo async def add_planned_day( self, workspace_id: UUID, child_id: UUID, parent_personal_data_uuid: UUID, date: date, day_type: str, is_dubbeldag: bool = False, ) -> PlannedLeaveDay: """Add a planned leave day. Validates that the day is available (dubbeldagar rules, no double-booking). """ # Validation 1: Check dubbeldagar quota if this is a dubbeldag if is_dubbeldag: await self._analytics.validate_dubbeldagar(child_id, date) # Validation 2: Check no double-booking on the same date for the same parent existing = await self._repo.list_by_workspace_child( workspace_id=workspace_id, child_id=child_id, from_date=date, to_date=date, ) # Only reject if the same parent already has this date planned parent_existing = [ e for e in existing if e.parent_personal_data_uuid == parent_personal_data_uuid ] if parent_existing: raise ValidationError(f"Parent already has a planned day on {date}") now = datetime.now() entity = PlannedLeaveDay( id=uuid4(), workspace_id=workspace_id, child_id=child_id, parent_personal_data_uuid=parent_personal_data_uuid, date=date, day_type=day_type, is_dubbeldag=is_dubbeldag, created_at=now, updated_at=now, ) return await self._repo.create(entity) async def remove_planned_day(self, planned_leave_day_id: UUID) -> None: """Remove a planned leave day.""" entity = await self._repo.get_by_id(planned_leave_day_id) if entity is None: raise NotFoundError(f"PlannedLeaveDay {planned_leave_day_id} not found") await self._repo.delete(planned_leave_day_id) async def list_by_workspace(self, workspace_id: UUID) -> list[PlannedLeaveDay]: """Get all planned days for a workspace.""" return await self._repo.list(workspace_id=workspace_id) async def list_by_date_range( self, workspace_id: UUID, child_id: UUID, start_date: date, end_date: date, ) -> list[PlannedLeaveDay]: """Get planned days within a date range for a child.""" return await self._repo.list_by_workspace_child( workspace_id=workspace_id, child_id=child_id, from_date=start_date, to_date=end_date, ) async def get(self, planned_leave_day_id: UUID) -> PlannedLeaveDay: """Get a planned day by ID.""" entity = await self._repo.get_by_id(planned_leave_day_id) if entity is None: raise NotFoundError(f"PlannedLeaveDay {planned_leave_day_id} not found") return entity async def list_by_child(self, child_id: UUID) -> list[PlannedLeaveDay]: """Get all planned days for a child across workspaces.""" return await self._repo.list(child_id=child_id) async def update( self, planned_leave_day_id: UUID, day_type: str | None = None, is_dubbeldag: bool | None = None, ) -> PlannedLeaveDay: """Update a planned day (partial update).""" entity = await self._repo.get_by_id(planned_leave_day_id) if entity is None: raise NotFoundError(f"PlannedLeaveDay {planned_leave_day_id} not found") if day_type is not None: entity.day_type = day_type if is_dubbeldag is not None: entity.is_dubbeldag = is_dubbeldag return await self._repo.update(planned_leave_day_id, entity) # -- bulk apply (POST /api/plan/apply) -------------------------------- # # This intentionally does NOT reuse `add_planned_day` for the write path. # `add_planned_day`'s double-booking check # (`list_by_workspace_child(..., from_date=date, to_date=date)`) looks at # *any* parent planned on that child+date, not just the one being added, # so calling it once per incoming day would misfire on legitimate # multi-parent days (e.g. dubbeldagar, where both parents are # deliberately planned on the same date) and would abort a re-apply of an # already-identical plan instead of treating it as a no-op. A bulk apply # also needs to validate the *whole* batch before writing anything ("a # half-applied plan is far worse than a rejected one"), which a # one-day-at-a-time call cannot do. `validate_dubbeldagar` on # `AnalyticsService` *is* reused below, since that check (15-month window) # is legitimately per-day and stateless. async def apply_proposed_days( self, workspace_id: UUID, days: list[ProposedDayInput], replace: bool = False, ) -> PlanApplyResult: """Persist a batch of proposed days as `planned_leave_day` rows. Idempotent per (workspace, child, parent, date): re-applying an identical batch creates nothing and is reported as all `skipped_identical_count`. A date already planned *differently* is a collision — by default the whole batch is refused and nothing is written; pass `replace=True` to overwrite those specific collisions (every other collision-free day in the batch is written too). Validates the entire batch up front (day types, per-day dubbeldagar window, batch-wide dubbeldagar quota, child eligibility window, remaining sjukpenning/lagsta balance) and only writes if everything passes — SQLite gives us no cross-request transaction here, so this explicit validate-then-write split is what keeps a rejection from ever leaving partial writes behind. """ # -- stage 1: structural validation ------------------------------ structural_errors: list[Rejection] = [] for i, d in enumerate(days): if d.day_type not in _PERSISTABLE_DAY_TYPES and d.day_type != "free": structural_errors.append( Rejection( message_key="planApply.rejection.invalidDayType", data={"index": i, "date": d.date.isoformat(), "day_type": d.day_type}, ) ) persistable = [d for d in days if d.day_type != "free"] skipped_free_count = len(days) - len(persistable) deduped: dict[tuple[UUID, UUID, date], ProposedDayInput] = {} for d in persistable: key = (d.child_id, d.parent_personal_data_uuid, d.date) prior = deduped.get(key) if prior is not None and prior != d: structural_errors.append( Rejection( message_key="planApply.rejection.conflictingDuplicate", data={ "date": d.date.isoformat(), "child_id": str(d.child_id), "parent_personal_data_uuid": str(d.parent_personal_data_uuid), }, ) ) deduped[key] = d if structural_errors: return PlanApplyResult(accepted=False, rejections=structural_errors) unique_days = list(deduped.values()) if not unique_days: return PlanApplyResult(accepted=True, skipped_free_count=skipped_free_count) # -- stage 2: classify against existing rows (collision detection) -- by_child: dict[UUID, list[date]] = {} for d in unique_days: by_child.setdefault(d.child_id, []).append(d.date) existing_by_key: dict[tuple[UUID, UUID, date], PlannedLeaveDay] = {} for child_id, dates in by_child.items(): existing_rows = await self._repo.list_by_workspace_child( workspace_id=workspace_id, child_id=child_id, from_date=min(dates), to_date=max(dates), ) for row in existing_rows: existing_by_key[(row.child_id, row.parent_personal_data_uuid, row.date)] = row to_create: list[ProposedDayInput] = [] to_replace: list[tuple[PlannedLeaveDay, ProposedDayInput]] = [] collisions: list[DayCollision] = [] skipped_identical_count = 0 for d in unique_days: key = (d.child_id, d.parent_personal_data_uuid, d.date) existing = existing_by_key.get(key) if existing is None: to_create.append(d) continue if existing.day_type == d.day_type and existing.is_dubbeldag == d.is_dubbeldag: skipped_identical_count += 1 continue collisions.append( DayCollision( date=d.date, parent_personal_data_uuid=d.parent_personal_data_uuid, child_id=d.child_id, existing_day_type=existing.day_type, proposed_day_type=d.day_type, ) ) if replace: to_replace.append((existing, d)) if collisions and not replace: return PlanApplyResult( accepted=False, collisions=collisions, rejections=[ Rejection( message_key="planApply.rejection.collisionsNeedReplace", data={"count": len(collisions)}, ) ], ) pending = to_create + [new for _, new in to_replace] if not pending: # Nothing left to write — either an empty batch or a fully # idempotent re-apply. return PlanApplyResult( accepted=True, skipped_identical_count=skipped_identical_count, skipped_free_count=skipped_free_count, ) # -- stage 3: domain validation on the batch that would actually write domain_errors: list[str] = [] domain_errors.extend(await self._validate_dubbeldagar_batch(pending, to_replace)) domain_errors.extend(await self._validate_eligible_range(pending)) domain_errors.extend(await self._validate_remaining_balance(pending, to_replace)) if domain_errors: return PlanApplyResult(accepted=False, collisions=collisions, rejections=domain_errors) # -- stage 4: write -------------------------------------------------- for old, _ in to_replace: await self._repo.delete(old.id) now = datetime.now() created: list[PlannedLeaveDay] = [] for d in pending: entity = PlannedLeaveDay( id=uuid4(), workspace_id=workspace_id, child_id=d.child_id, parent_personal_data_uuid=d.parent_personal_data_uuid, date=d.date, day_type=d.day_type, is_dubbeldag=d.is_dubbeldag, created_at=now, updated_at=now, ) created.append(await self._repo.create(entity)) return PlanApplyResult( accepted=True, created=created, created_count=len(to_create), skipped_identical_count=skipped_identical_count, skipped_free_count=skipped_free_count, replaced_count=len(to_replace), ) async def _validate_dubbeldagar_batch( self, pending: list[ProposedDayInput], to_replace: list[tuple[PlannedLeaveDay, ProposedDayInput]], ) -> list[Rejection]: """Per-day window check (reused from AnalyticsService) + batch-wide quota. `AnalyticsService.validate_dubbeldagar` only sees rows already committed to the DB, so it cannot by itself catch a batch that introduces more than `max_dubbeldagar` new dubbeldagar in one go — that aggregate check is done here, on top of it. """ errors: list[Rejection] = [] dubbeldag_days = [d for d in pending if d.is_dubbeldag] if not dubbeldag_days: return errors for d in dubbeldag_days: try: await self._analytics.validate_dubbeldagar(d.child_id, d.date) except DubbeldagError as exc: message_key = exc.message_key or "planApply.rejection.dubbeldagInvalid" data: dict[str, object] = {**exc.data, "date": d.date.isoformat()} errors.append(Rejection(message_key=message_key, data=data)) replaced_out_dubbeldag_by_child: dict[UUID, int] = {} for old, _new in to_replace: if old.is_dubbeldag: replaced_out_dubbeldag_by_child[old.child_id] = ( replaced_out_dubbeldag_by_child.get(old.child_id, 0) + 1 ) by_child: dict[UUID, int] = {} for d in dubbeldag_days: by_child[d.child_id] = by_child.get(d.child_id, 0) + 1 for child_id, new_count in by_child.items(): already = await self._repo.count_dubbeldagar(child_id) replaced_out = replaced_out_dubbeldag_by_child.get(child_id, 0) projected_total = already - replaced_out + new_count if projected_total > settings.rules.max_dubbeldagar: errors.append( Rejection( message_key="planApply.rejection.dubbeldagBatchQuotaExceeded", data={ "child_id": str(child_id), "projected_total": projected_total, "max_dubbeldagar": settings.rules.max_dubbeldagar, }, ) ) return errors async def _validate_eligible_range(self, pending: list[ProposedDayInput]) -> list[Rejection]: """Every date must fall within the child's eligible window. Lower bound is the child's birth date; upper bound is the hard expiry (`LeaveRules.hard_expiry_age_years`, default 12) after which no FP days may be used at all, per Swedish law. """ errors: list[Rejection] = [] birth_dates: dict[UUID, date] = {} for d in pending: if d.child_id not in birth_dates: projection = await self._analytics.get_days_left_at_age_4( d.parent_personal_data_uuid, d.child_id ) birth_dates[d.child_id] = projection.birth_date birth_date = birth_dates[d.child_id] hard_expiry = birth_date + relativedelta(years=settings.rules.hard_expiry_age_years) if d.date < birth_date or d.date > hard_expiry: errors.append( Rejection( message_key="planApply.rejection.outsideEligibleWindow", data={ "date": d.date.isoformat(), "child_id": str(d.child_id), "birth_date": birth_date.isoformat(), "hard_expiry": hard_expiry.isoformat(), }, ) ) return errors async def _tier_pool_left(self, parent_uuid: UUID, child_id: UUID) -> tuple[int, int]: """Remaining (sjukpenning_left, lagsta_left), net of already-planned rows. Mirrors the pool computation in `allocation_service.AllocationInputLoader.load` (same FK-snapshot-or- rule-default, minus locally committed days) so this check agrees with what the planning engine itself would have allowed. """ rules = settings.rules latest = await self._parental_leave_repo.get_latest_by_parent(parent_uuid, child_id) if latest is not None: sjuk_total_left = latest.tier_data.sjukpenning_left lagsta_total_left = latest.tier_data.lagsta_niva_left else: sjuk_total_left = rules.sjukpenning_days lagsta_total_left = rules.lagsta_niva_days committed_sjuk = await self._repo.count_by_parent_type(parent_uuid, child_id, "sjukpenning") committed_lagsta = await self._repo.count_by_parent_type(parent_uuid, child_id, "lagsta") return ( max(sjuk_total_left - committed_sjuk, 0), max(lagsta_total_left - committed_lagsta, 0), ) async def _validate_remaining_balance( self, pending: list[ProposedDayInput], to_replace: list[tuple[PlannedLeaveDay, ProposedDayInput]], ) -> list[Rejection]: """The batch must not spend more sjukpenning/lagsta days than remain. Per (parent, child, tier): net_new = (new days of that tier the batch adds) - (days of that tier being replaced away, which free up a slot). Rejects if net_new exceeds what's left. """ errors: list[Rejection] = [] pairs = {(d.parent_personal_data_uuid, d.child_id) for d in pending} for parent_uuid, child_id in pairs: sjuk_left, lagsta_left = await self._tier_pool_left(parent_uuid, child_id) for tier, left in (("sjukpenning", sjuk_left), ("lagsta", lagsta_left)): added = sum( 1 for d in pending if d.parent_personal_data_uuid == parent_uuid and d.child_id == child_id and d.day_type == tier ) removed = sum( 1 for old, new in to_replace if new.parent_personal_data_uuid == parent_uuid and new.child_id == child_id and old.day_type == tier ) net_new = added - removed if net_new > left: errors.append( Rejection( message_key="planApply.rejection.insufficientBalance", data={ "parent_personal_data_uuid": str(parent_uuid), "child_id": str(child_id), "tier": tier, "needed": net_new, "remaining": left, }, ) ) return errors