F-r-ldraledigplannerare/backend/migrations/versions/009_day_assertion.py

"""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")