"""Add day_assertion: the intent-vs-derived-output restructure's step 3. `planned_leave_day` used to do three jobs at once — allocator output cache, user override store (`is_pinned`, and ad-hoc rows with `wish_id IS NULL`), and resource ledger/historical record. That blur was the root cause of a string of defects (see `~/.claude/plans/please-make-a-plan-cheerful-beacon.md.restructure`, step 3). This migration adds `day_assertion` — one row per (workspace, child, parent, date) recording what the USER asserts about that day, independent of what the planner has derived: - a pinned `planned_leave_day` row (`is_pinned=1`) -> an assertion carrying that row's `day_type`/`is_dubbeldag`/`wish_id`. - an ad-hoc `planned_leave_day` row (`wish_id IS NULL`) -> an assertion the same way (a row can be both pinned AND ad-hoc; backfilled once, not twice). - every `planned_day_exclusion` row -> an assertion with `day_type='none'` ("the planner may not place leave here"). Backfilled AFTER the `planned_leave_day` pass and written with `INSERT OR REPLACE`, so an exclusion overwrites whatever assertion its (workspace, child, parent, date) already picked up — matching runtime behavior, where `PlannedDayExclusionService.exclude()` always deletes any existing `planned_leave_day` row at that key, so the two are never meant to coexist; this only matters for historical data where they still might. Step 3a is deliberately additive and reversible: `planned_leave_day.is_pinned` and the `planned_day_exclusion` table are NOT dropped here (`downgrade()` leaves them exactly as it found them, since it never touched them). They stay in place, unused by the allocator's new assertion-based input, so the frontend can keep calling the legacy `/api/planned-days` and `/api/planned-day-exclusions` endpoints (now a compatibility shim writing `day_assertion` rows underneath — see `PlanningService`/ `PlannedDayExclusionService`) until step 3b moves it over and drops both. Must stay byte-for-byte equivalent to the `day_assertion` table in `app/db/connection.py`'s `_FULL_SCHEMA` — `tests/integration/test_schema_parity.py` asserts exactly that. Revision ID: 009 Revises: 008 Create Date: 2026-09-20 """ from __future__ import annotations import uuid from datetime import datetime from alembic import op revision: str = "009" down_revision: str | None = "008" branch_labels: str | None = None depends_on: str | None = None _TABLE_DDL = """ CREATE TABLE IF NOT EXISTS day_assertion ( id TEXT PRIMARY KEY, workspace_id TEXT NOT NULL, child_id TEXT NOT NULL, parent_personal_data_uuid TEXT NOT NULL, date DATE NOT NULL, day_type TEXT NOT NULL CHECK (day_type IN ('sjukpenning','lagsta','vacation','none')), is_dubbeldag INTEGER NOT NULL DEFAULT 0, wish_id TEXT, created_at TIMESTAMP NOT NULL, updated_at TIMESTAMP NOT NULL, UNIQUE(workspace_id, child_id, parent_personal_data_uuid, date), FOREIGN KEY (workspace_id) REFERENCES workspace(id) ON DELETE CASCADE, FOREIGN KEY (child_id) REFERENCES child(id) ON DELETE CASCADE, FOREIGN KEY (parent_personal_data_uuid) REFERENCES personal_data(uuid) ON DELETE CASCADE, FOREIGN KEY (wish_id) REFERENCES leave_wish(id) ON DELETE SET NULL ) """ _INDEX_DDL = """ CREATE INDEX IF NOT EXISTS idx_day_assertion_workspace ON day_assertion (workspace_id) """ def upgrade() -> None: from sqlalchemy import text conn = op.get_bind() op.execute(_TABLE_DDL) op.execute(_INDEX_DDL) now = datetime.now().isoformat() # -- pinned / ad-hoc planned_leave_day rows -> concrete assertions ------- planned_rows = conn.execute( text( "SELECT workspace_id, child_id, parent_personal_data_uuid, date, " "day_type, is_dubbeldag, wish_id FROM planned_leave_day " "WHERE is_pinned = 1 OR wish_id IS NULL" ) ).fetchall() for row in planned_rows: conn.execute( text( "INSERT INTO day_assertion " "(id, workspace_id, child_id, parent_personal_data_uuid, date, " " day_type, is_dubbeldag, wish_id, created_at, updated_at) " "VALUES (:id, :workspace_id, :child_id, :parent_personal_data_uuid, :date, " " :day_type, :is_dubbeldag, :wish_id, :now, :now)" ), { "id": str(uuid.uuid4()), "workspace_id": row.workspace_id, "child_id": row.child_id, "parent_personal_data_uuid": row.parent_personal_data_uuid, "date": row.date, "day_type": row.day_type, "is_dubbeldag": row.is_dubbeldag, "wish_id": row.wish_id, "now": now, }, ) # -- planned_day_exclusion rows -> day_type='none' assertions ------------ # `INSERT OR REPLACE`: an exclusion overwrites whatever the pass above # wrote for the same (workspace, child, parent, date) — see module # docstring for why the two should never really coexist, and why # exclusion wins when historical data disagrees. exclusion_rows = conn.execute( text( "SELECT workspace_id, child_id, parent_personal_data_uuid, date " "FROM planned_day_exclusion" ) ).fetchall() for row in exclusion_rows: conn.execute( text( "INSERT OR REPLACE INTO day_assertion " "(id, workspace_id, child_id, parent_personal_data_uuid, date, " " day_type, is_dubbeldag, wish_id, created_at, updated_at) " "VALUES (:id, :workspace_id, :child_id, :parent_personal_data_uuid, :date, " " 'none', 0, NULL, :now, :now)" ), { "id": str(uuid.uuid4()), "workspace_id": row.workspace_id, "child_id": row.child_id, "parent_personal_data_uuid": row.parent_personal_data_uuid, "date": row.date, "now": now, }, ) def downgrade() -> None: """Drop `day_assertion` — the point of no return for anything it alone records. On a `downgrade` chain past 010, 010's `downgrade()` has already run and reconstructed `is_pinned`/`planned_day_exclusion` from this table (see its docstring). The one thing that reconstruction cannot produce is a concrete assertion with no matching `planned_leave_day` row — there is no pre-009 column to put that fact in. Once this migration drops the table, that fact is gone for good, so it is worth surfacing rather than dropping in silence: warn on stdout, listing each orphaned key, before the `DROP TABLE`. This does not block the downgrade — an operator rolling back a bad deploy needs the rollback to complete, not a new failure — it only makes sure they see what is about to become unrecoverable. """ from sqlalchemy import text conn = op.get_bind() orphaned = conn.execute( text( "SELECT da.workspace_id, da.child_id, da.parent_personal_data_uuid, " " da.date, da.day_type " "FROM day_assertion da " "WHERE da.day_type != 'none' AND NOT EXISTS (" " SELECT 1 FROM planned_leave_day pld " " WHERE pld.workspace_id = da.workspace_id " " AND pld.child_id = da.child_id " " AND pld.parent_personal_data_uuid = da.parent_personal_data_uuid " " AND pld.date = da.date" ")" ) ).fetchall() if orphaned: print( f"WARNING: dropping day_assertion discards {len(orphaned)} row(s) with no " "matching planned_leave_day row to carry the fact forward (the pre-009 " "schema has no field for it). Unrecoverable after this migration:" ) for row in orphaned: print( f" workspace={row.workspace_id} child={row.child_id} " f"parent={row.parent_personal_data_uuid} date={row.date} " f"day_type={row.day_type}" ) op.execute("DROP INDEX IF EXISTS idx_day_assertion_workspace") op.execute("DROP TABLE IF EXISTS day_assertion")