"""Read/write logic for the multi-country public-holiday calendar.

Everything above this module — the HR console, the employee 公示 page, the
Word/Excel exports, the notice generator, the External API and MCP — reads
holidays through here, so the aggregation and country-resolution rules live in
exactly one place.

Two ideas carry most of the weight:

**Groups.** Storage is one row per day, which is what makes the month grid and
date-range queries trivial. But humans think in holidays, not days ("劳动节 5/1–5/5"),
and so do the Word export and the notice generator. ``group_key`` is the bridge:
every day of one holiday — plus the makeup workdays that offset it — shares a
key, and :func:`build_groups` folds rows back into :class:`HolidayGroup` objects
with contiguous date segments.

**Observed days.** The count HR publishes ("Australia Holidays - 8 days") is not
the number of stored rows: it is the number of holiday dates falling Mon–Fri, net
of makeup workdays. :func:`observed_day_count` computes it, which lets the console
cross-check HR's stored ``stated_days`` instead of silently overwriting it.
"""
from __future__ import annotations

from collections import defaultdict
from collections.abc import Iterable, Sequence
from dataclasses import dataclass, field
from datetime import date, datetime, timedelta
from typing import Any

from sqlalchemy import and_, delete, func, or_, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.exceptions import ConflictError, NotFoundError, ValidationError
from app.core.geo import normalize_country_code
from app.domains.hr.holiday_keys import group_base_key
from app.domains.hr.holiday_models import (
    DAY_TYPES,
    MAKEUP_WORKDAY,
    PUBLIC_HOLIDAY,
    SOURCE_MANUAL,
    SOURCE_SEED,
    SOURCES,
    HrPublicHoliday,
    HrPublicHolidayCalendar,
    name_key_for,
)
from app.domains.hr.holiday_seed_2026 import SEED_2026, SEED_YEAR
from app.identity.person_models import IdentityPerson

# A calendar year is the natural maintenance and display unit. Bound the
# accepted range so a typo can neither create a 1970 calendar nor let an
# unbounded ``year`` scan the whole table.
MIN_YEAR = 2000
MAX_YEAR = 2100

# Upper bound for one bulk write (Excel import, AI-parsed batch, group create).
# A single country-year is ~35 rows, so this is ~30 country-years of headroom
# while still capping the transaction size.
MAX_BULK_ROWS = 1000

# Guard against an unbounded date window on the read paths.
MAX_WINDOW_DAYS = 366 * 5

#: How far apart two same-named imported dates may be and still be one holiday.
#: Above the widest real span (China's National Day spans its 09-20 makeup
#: workday to its 10-10 one, 21 days) and far below the months separating two
#: genuinely distinct runs of one name (Japan's two "New Year Holidays").
_GROUP_GAP_DAYS = 31


# ──────────────────────────── aggregation ────────────────────────────


@dataclass
class HolidayGroup:
    """One holiday, folded back together from its per-day rows."""

    country: str
    group_key: str
    name_en: str
    name_local: str | None
    note: str | None
    # Paid days off, ascending.
    days: list[date] = field(default_factory=list)
    # 调休上班日 offsetting this holiday, ascending.
    makeup_days: list[date] = field(default_factory=list)
    row_ids: list[int] = field(default_factory=list)
    sources: set[str] = field(default_factory=set)

    @property
    def start(self) -> date:
        return self.days[0] if self.days else self.makeup_days[0]

    @property
    def end(self) -> date:
        return self.days[-1] if self.days else self.makeup_days[-1]

    @property
    def year(self) -> int:
        return self.start.year

    @property
    def total_days(self) -> int:
        """Calendar days off (weekends included)."""
        return len(self.days)

    @property
    def observed_days(self) -> int:
        """Days off that actually cost a workday, HR's published unit."""
        return observed_day_count(self.days, self.makeup_days)

    @property
    def segments(self) -> list[tuple[date, date]]:
        """Contiguous runs of days off, e.g. ``[(5/1, 5/5)]``.

        A group is normally one run. It can legitimately be several — HR may
        group two halves of a festival split by a working day — and the notice
        generator and Word export both render each run separately rather than
        printing a range that includes days people are working.
        """
        return contiguous_segments(self.days)

    @property
    def is_makeup_only(self) -> bool:
        return not self.days and bool(self.makeup_days)


def contiguous_segments(days: Sequence[date]) -> list[tuple[date, date]]:
    """Collapse a sorted date list into ``(start, end)`` runs of consecutive days."""
    out: list[tuple[date, date]] = []
    for day in sorted(days):
        if out and day == out[-1][1] + timedelta(days=1):
            out[-1] = (out[-1][0], day)
        elif not out or day != out[-1][1]:
            out.append((day, day))
    return out


def observed_day_count(
    days: Iterable[date], makeup_days: Iterable[date] = (),
) -> int:
    """Holiday dates falling Mon–Fri, net of makeup workdays.

    This reproduces the headline count in HR's source document for 21 of its 22
    countries. (Romania states 12 against 11 derivable, because 2026-06-01
    carries two holiday names and is counted twice there — which is exactly the
    kind of divergence the console surfaces rather than hides.)

    Never returns a negative number: more makeup days than weekday holidays is
    nonsense data, and reporting it as ``-1`` days off would be worse than
    reporting ``0``.
    """
    weekday_days = {d for d in days if d.weekday() < 5}
    return max(0, len(weekday_days) - len(set(makeup_days)))


def build_groups(rows: Iterable[HrPublicHoliday]) -> list[HolidayGroup]:
    """Fold per-day rows into holiday groups, ordered by start date."""
    by_key: dict[tuple[str, str], HolidayGroup] = {}
    for row in rows:
        key = (row.country, row.group_key)
        group = by_key.get(key)
        if group is None:
            group = HolidayGroup(
                country=row.country,
                group_key=row.group_key,
                name_en=row.name_en,
                name_local=row.name_local,
                note=row.note,
            )
            by_key[key] = group
        if row.day_type == MAKEUP_WORKDAY:
            group.makeup_days.append(row.day)
        else:
            group.days.append(row.day)
            # A makeup row's name is derived ("Makeup Workday (…)"), so the
            # group's display name must come from a real holiday row — the
            # EARLIEST one, so the result does not depend on row order (these
            # queries carry no ORDER BY guarantee for ties).
            if not group.row_ids or row.day == min(group.days):
                group.name_en = row.name_en
                group.name_local = row.name_local
                group.note = row.note or group.note
        group.row_ids.append(row.id)
        group.sources.add(row.source)

    for group in by_key.values():
        group.days.sort()
        group.makeup_days.sort()
    return sorted(by_key.values(), key=lambda g: (g.start, g.country, g.group_key))


# ──────────────────────────── validation ────────────────────────────


def validate_year(year: int) -> int:
    if not MIN_YEAR <= year <= MAX_YEAR:
        raise ValidationError(
            f"year must be between {MIN_YEAR} and {MAX_YEAR}",
        )
    return year


def resolve_country(raw: str | None, *, field_name: str = "country") -> str:
    """Normalise HR-supplied country input to a real ISO-3166 alpha-2 code.

    Rejects rather than guesses: an unrecognised value would create a calendar
    nobody's ``residence_country`` can ever match, i.e. holidays no employee
    would see.
    """
    code = normalize_country_code(raw)
    if not code:
        raise ValidationError(
            f"{field_name} must be a recognised country/region "
            f"(ISO-3166 alpha-2, alpha-3 or English/Chinese name): {raw!r}",
        )
    return code


def validate_day_type(value: str | None) -> str:
    day_type = (value or PUBLIC_HOLIDAY).strip()
    if day_type not in DAY_TYPES:
        raise ValidationError(f"day_type must be one of {list(DAY_TYPES)}")
    return day_type


def validate_source(value: str | None) -> str:
    source = (value or SOURCE_MANUAL).strip()
    if source not in SOURCES:
        raise ValidationError(f"source must be one of {list(SOURCES)}")
    return source


def _escape_like(value: str) -> str:
    """Escape ``\\``, ``%`` and ``_`` so a search term is matched literally."""
    return value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")


def year_bounds(year: int) -> tuple[date, date]:
    validate_year(year)
    return date(year, 1, 1), date(year, 12, 31)


def _validate_window(start: date, end: date) -> None:
    if end < start:
        raise ValidationError("end date must not precede start date")
    if (end - start).days + 1 > MAX_WINDOW_DAYS:
        raise ValidationError(
            f"date window must not exceed {MAX_WINDOW_DAYS} days",
        )


# ──────────────────────────── reads ────────────────────────────


async def list_days(
    db: AsyncSession,
    *,
    countries: Sequence[str] | None = None,
    start: date | None = None,
    end: date | None = None,
    year: int | None = None,
    day_types: Sequence[str] | None = None,
    group_key: str | None = None,
) -> list[HrPublicHoliday]:
    """Fetch holiday rows for a country set and date window.

    ``year`` is a convenience for the (Jan 1, Dec 31) window; passing both a
    year and explicit bounds is an error rather than a silently ignored input.
    """
    if year is not None and (start is not None or end is not None):
        raise ValidationError("pass either 'year' or an explicit start/end window")
    if year is not None:
        start, end = year_bounds(year)
    if start is not None and end is not None:
        _validate_window(start, end)

    stmt = select(HrPublicHoliday)
    if countries:
        stmt = stmt.where(HrPublicHoliday.country.in_(list(countries)))
    if start is not None:
        stmt = stmt.where(HrPublicHoliday.day >= start)
    if end is not None:
        stmt = stmt.where(HrPublicHoliday.day <= end)
    if day_types:
        stmt = stmt.where(HrPublicHoliday.day_type.in_(list(day_types)))
    if group_key:
        stmt = stmt.where(HrPublicHoliday.group_key == group_key)
    stmt = stmt.order_by(
        HrPublicHoliday.day,
        HrPublicHoliday.country,
        HrPublicHoliday.name_en,
    )
    return list((await db.execute(stmt)).scalars().all())


async def list_calendars(
    db: AsyncSession,
    *,
    year: int | None = None,
    countries: Sequence[str] | None = None,
) -> list[HrPublicHolidayCalendar]:
    stmt = select(HrPublicHolidayCalendar)
    if year is not None:
        stmt = stmt.where(HrPublicHolidayCalendar.year == validate_year(year))
    if countries:
        stmt = stmt.where(HrPublicHolidayCalendar.country.in_(list(countries)))
    stmt = stmt.order_by(
        HrPublicHolidayCalendar.year.desc(),
        HrPublicHolidayCalendar.country,
    )
    return list((await db.execute(stmt)).scalars().all())


async def get_calendar(
    db: AsyncSession, *, country: str, year: int,
) -> HrPublicHolidayCalendar | None:
    return (
        await db.execute(
            select(HrPublicHolidayCalendar).where(
                HrPublicHolidayCalendar.country == country,
                HrPublicHolidayCalendar.year == year,
            )
        )
    ).scalar_one_or_none()


async def available_years(db: AsyncSession) -> list[int]:
    """Years that have either a calendar row or at least one holiday day."""
    cal_years = set(
        (await db.execute(select(HrPublicHolidayCalendar.year).distinct())).scalars().all()
    )
    day_years = set(
        (
            await db.execute(
                select(func.extract("year", HrPublicHoliday.day)).distinct()
            )
        ).scalars().all()
    )
    return sorted({int(y) for y in cal_years | day_years if y})


async def list_countries(
    db: AsyncSession, *, year: int | None = None,
) -> list[dict[str, Any]]:
    """Country registry for pickers: code, English label, and day counts.

    Driven by the calendar table (the maintained set) left-joined onto actual
    day counts, so a country HR has created but not yet filled in still shows
    up — with ``day_count`` 0 — instead of vanishing from the picker.
    """
    calendars = await list_calendars(db, year=year)
    counts: dict[tuple[str, int], int] = defaultdict(int)
    if calendars:
        # Aggregate in SQL rather than hydrating every day row: this feeds a
        # picker hint, and with ``year=None`` the old version read the whole
        # table as ORM objects on every console page load.
        stmt = (
            select(
                HrPublicHoliday.country,
                func.extract("year", HrPublicHoliday.day).label("y"),
                func.count().label("n"),
            )
            .where(
                HrPublicHoliday.country.in_([c.country for c in calendars]),
                HrPublicHoliday.day_type == PUBLIC_HOLIDAY,
            )
            .group_by(HrPublicHoliday.country, "y")
        )
        if year is not None:
            start, end = year_bounds(year)
            stmt = stmt.where(
                HrPublicHoliday.day >= start, HrPublicHoliday.day <= end,
            )
        for country, day_year, count in (await db.execute(stmt)).all():
            counts[(country, int(day_year))] = int(count)

    out: list[dict[str, Any]] = []
    for cal in calendars:
        out.append({
            "country": cal.country,
            "label": cal.label or cal.country,
            "year": cal.year,
            "day_count": counts.get((cal.country, cal.year), 0),
            "stated_days": cal.stated_days,
        })
    return out


async def upcoming_groups(
    db: AsyncSession,
    *,
    countries: Sequence[str],
    today: date,
    limit: int = 3,
    horizon_days: int = 400,
) -> list[HolidayGroup]:
    """The next few holiday groups starting on or after ``today``.

    A group already under way (started yesterday, ends tomorrow) still counts as
    upcoming: an employee on day 2 of a 5-day holiday should see it, not the one
    after it. Bounded by ``horizon_days`` so the scan stays on the
    ``(country, day)`` index.
    """
    if not countries:
        return []
    limit = max(1, min(limit, 20))
    rows = await list_days(
        db,
        countries=countries,
        start=today - timedelta(days=30),
        end=today + timedelta(days=min(horizon_days, MAX_WINDOW_DAYS)),
    )
    groups = [g for g in build_groups(rows) if not g.is_makeup_only and g.end >= today]
    return groups[:limit]


async def coverage(
    db: AsyncSession, *, countries: Sequence[str], year: int,
) -> dict[str, str]:
    """Per-country maintenance state: ``covered`` | ``no_data``.

    Mirrors the three-state contract of the leave-availability External API: a
    country with no calendar row is *unmaintained*, which is a different fact
    from "maintained and has no holidays that year". Collapsing the two into an
    empty list would let a consumer conclude people are working when nobody has
    told Helix otherwise.
    """
    validate_year(year)
    maintained = {
        c.country for c in await list_calendars(db, year=year, countries=countries)
    }
    return {
        code: ("covered" if code in maintained else "no_data") for code in countries
    }


async def last_updated_at(
    db: AsyncSession, *, countries: Sequence[str] | None = None,
) -> datetime | None:
    """Newest ``updated_at`` across days and calendars — the freshness stamp.

    Reported as ``meta.as_of`` on the External API so a consumer can tell how
    current the calendar is, per the platform's freshness convention.
    """
    day_stmt = select(func.max(HrPublicHoliday.updated_at))
    cal_stmt = select(func.max(HrPublicHolidayCalendar.updated_at))
    if countries:
        day_stmt = day_stmt.where(HrPublicHoliday.country.in_(list(countries)))
        cal_stmt = cal_stmt.where(
            HrPublicHolidayCalendar.country.in_(list(countries))
        )
    stamps = [
        (await db.execute(day_stmt)).scalar(),
        (await db.execute(cal_stmt)).scalar(),
    ]
    present = [s for s in stamps if s is not None]
    return max(present) if present else None  # type: ignore[no-any-return]


# ──────────────────────── employee country axis ────────────────────────


def person_country(person: IdentityPerson | None) -> str | None:
    """The country whose holiday calendar applies to this person.

    ``residence_country`` first — it is the *geographic* axis, derived from the
    SuccessFactors work-location code, and holidays follow where someone
    actually works. ``country`` (the legal-entity axis) is the fallback, and it
    needs normalising rather than slicing: the field is documented as alpha-2 but
    ``identity_sync`` stores alpha-3 in practice, so a naive ``[:2]`` turns
    ``CHN`` into ``CH`` and quietly serves Swiss holidays to Beijing.

    Returns ``None`` when neither field resolves, and the caller falls back to
    showing all countries rather than guessing one.
    """
    if person is None:
        return None
    for raw in (person.residence_country, person.country):
        code = normalize_country_code(raw)
        if code:
            return code
    return None


async def resolve_person_country(
    db: AsyncSession, *, person_id: str,
) -> tuple[str | None, IdentityPerson | None]:
    person = (
        await db.execute(
            select(IdentityPerson).where(
                IdentityPerson.id == person_id,
                IdentityPerson.deleted_at.is_(None),
            )
        )
    ).scalar_one_or_none()
    return person_country(person), person


# ──────────────────────────── writes ────────────────────────────


async def _existing_keys(
    db: AsyncSession, *, country: str, year: int | None = None,
) -> set[tuple[date, str]]:
    stmt = select(HrPublicHoliday.day, HrPublicHoliday.name_key).where(
        HrPublicHoliday.country == country
    )
    if year is not None:
        start, end = year_bounds(year)
        stmt = stmt.where(HrPublicHoliday.day >= start, HrPublicHoliday.day <= end)
    return {(row[0], row[1]) for row in (await db.execute(stmt)).all()}


async def allocate_group_key(
    db: AsyncSession, *, country: str, year: int, name_en: str,
) -> str:
    """Deterministic, collision-free group key for a new holiday group.

    Same shape as the seed (``cn-2026-labor-day-holiday``) so seeded and
    HR-created groups are indistinguishable downstream. A country that already
    has that name in that year gets a ``-2`` suffix rather than merging into the
    existing group: Japan really does observe two separate "New Year Holidays".
    """
    base = f"{country.lower()}-{year}-{name_key_for(name_en)}"[:80]
    taken = set(
        (
            await db.execute(
                select(HrPublicHoliday.group_key).where(
                    HrPublicHoliday.country == country,
                    HrPublicHoliday.group_key.like(f"{base}%"),
                )
            )
        ).scalars().all()
    )
    if base not in taken:
        return base
    n = 2
    while f"{base}-{n}" in taken:
        n += 1
    return f"{base}-{n}"


async def create_group(
    db: AsyncSession,
    *,
    country: str,
    name_en: str,
    name_local: str | None,
    days: Sequence[date],
    makeup_days: Sequence[date] = (),
    note: str | None = None,
    source: str = SOURCE_MANUAL,
    actor_id: str | None = None,
    group_key: str | None = None,
) -> tuple[str, list[HrPublicHoliday]]:
    """Create one holiday (possibly spanning several days) plus its makeup days.

    The primary HR write path: a holiday is created as a unit, not day by day.
    Returns the allocated ``group_key`` and the created rows.
    """
    country = resolve_country(country)
    name_en = (name_en or "").strip()
    if not name_en:
        raise ValidationError("name_en is required")
    day_list = sorted(set(days))
    makeup_list = sorted(set(makeup_days))
    if not day_list and not makeup_list:
        raise ValidationError("a holiday group needs at least one day")
    overlap = set(day_list) & set(makeup_list)
    if overlap:
        raise ValidationError(
            "a date cannot be both a day off and a makeup workday: "
            + ", ".join(d.isoformat() for d in sorted(overlap)),
        )
    if len(day_list) + len(makeup_list) > MAX_BULK_ROWS:
        raise ValidationError(f"a group may not exceed {MAX_BULK_ROWS} days")

    anchor_year = (day_list or makeup_list)[0].year
    # Validate every year the days touch, not just the first: a cross-year
    # holiday must not slip a 2101 date past the bound on its January half.
    for touched in {d.year for d in (*day_list, *makeup_list)}:
        validate_year(touched)
    source = validate_source(source)
    if group_key is None:
        group_key = await allocate_group_key(
            db, country=country, year=anchor_year, name_en=name_en,
        )

    taken = await _existing_keys(db, country=country)
    created: list[HrPublicHoliday] = []
    conflicts: list[str] = []

    def _add(day: date, display_name: str, local: str | None, day_type: str) -> None:
        key = name_key_for(display_name)
        if (day, key) in taken:
            conflicts.append(f"{day.isoformat()} {display_name}")
            return
        taken.add((day, key))
        row = HrPublicHoliday(
            country=country,
            day=day,
            group_key=group_key,
            name_en=display_name,
            name_local=local,
            name_key=key,
            day_type=day_type,
            note=note,
            source=source,
            created_by_id=actor_id,
            updated_by_id=actor_id,
        )
        db.add(row)
        created.append(row)

    for day in day_list:
        _add(day, name_en, name_local, PUBLIC_HOLIDAY)
    makeup_en = f"Makeup Workday ({name_en})"
    makeup_local = f"{name_local}调休上班" if name_local else None
    for day in makeup_list:
        _add(day, makeup_en, makeup_local, MAKEUP_WORKDAY)

    if conflicts:
        raise ConflictError(
            "these dates already carry the same holiday name: "
            + "; ".join(conflicts),
        )

    await db.flush()
    await _ensure_calendars_for(
        db, country=country, days=[*day_list, *makeup_list], actor_id=actor_id,
    )
    return group_key, created


async def replace_group(
    db: AsyncSession,
    *,
    country: str,
    group_key: str,
    name_en: str,
    name_local: str | None,
    days: Sequence[date],
    makeup_days: Sequence[date] = (),
    note: str | None = None,
    actor_id: str | None = None,
) -> list[HrPublicHoliday]:
    """Set a group's name and exact day set, preserving untouched rows.

    Diff-based rather than delete-and-reinsert so ``created_at`` / ``created_by``
    survive an edit that only adds a day, and so the audit trail reflects what
    actually changed.
    """
    country = resolve_country(country)
    name_en = (name_en or "").strip()
    if not name_en:
        raise ValidationError("name_en is required")
    existing = await list_days(db, countries=[country], group_key=group_key)
    if not existing:
        raise NotFoundError(f"holiday group {group_key!r} not found for {country}")

    day_list = sorted(set(days))
    makeup_list = sorted(set(makeup_days))
    overlap = set(day_list) & set(makeup_list)
    if overlap:
        raise ValidationError(
            "a date cannot be both a day off and a makeup workday: "
            + ", ".join(d.isoformat() for d in sorted(overlap)),
        )
    if not day_list and not makeup_list:
        raise ValidationError(
            "a holiday group needs at least one day — delete the group instead",
        )
    if len(day_list) + len(makeup_list) > MAX_BULK_ROWS:
        raise ValidationError(f"a group may not exceed {MAX_BULK_ROWS} days")

    makeup_en = f"Makeup Workday ({name_en})"
    makeup_local = f"{name_local}调休上班" if name_local else None
    wanted: dict[date, tuple[str, str | None, str]] = {
        d: (name_en, name_local, PUBLIC_HOLIDAY) for d in day_list
    }
    wanted.update({d: (makeup_en, makeup_local, MAKEUP_WORKDAY) for d in makeup_list})

    by_day = {row.day: row for row in existing}
    # Rows belonging to OTHER groups in this country, which the new day set must
    # not collide with (the unique key is country+day+name).
    other_keys = {
        (row.day, row.name_key)
        for row in await list_days(db, countries=[country])
        if row.group_key != group_key
    }

    for day, (display_name, local, day_type) in wanted.items():
        key = name_key_for(display_name)
        row = by_day.get(day)
        if row is None:
            if (day, key) in other_keys:
                raise ConflictError(
                    f"{day.isoformat()} already carries a holiday named "
                    f"{display_name!r} in {country}",
                )
            db.add(HrPublicHoliday(
                country=country, day=day, group_key=group_key,
                name_en=display_name, name_local=local, name_key=key,
                day_type=day_type, note=note, source=SOURCE_MANUAL,
                created_by_id=actor_id, updated_by_id=actor_id,
            ))
            continue
        if (day, key) in other_keys and row.name_key != key:
            raise ConflictError(
                f"{day.isoformat()} already carries a holiday named "
                f"{display_name!r} in {country}",
            )
        row.name_en = display_name
        row.name_local = local
        row.name_key = key
        row.day_type = day_type
        row.note = note
        row.updated_by_id = actor_id

    for day, row in by_day.items():
        if day not in wanted:
            await db.delete(row)

    await db.flush()
    await _ensure_calendars_for(
        db, country=country, days=[*day_list, *makeup_list], actor_id=actor_id,
    )
    return await list_days(db, countries=[country], group_key=group_key)


async def delete_group(
    db: AsyncSession, *, country: str, group_key: str,
) -> int:
    """Delete every day of one holiday.

    The country-year's calendar row is deliberately left behind, unlike
    :func:`delete_calendar` which refuses to orphan days. The asymmetry is
    intentional: "HR maintains this country and it genuinely has no holidays
    left" is a real, reportable state (``coverage="covered"`` with an empty
    list), whereas days with no calendar row would report ``no_data`` while
    visibly serving holidays — a contradiction. Removing the country entirely is
    what ``delete_calendar`` is for.
    """
    country = resolve_country(country)
    rows = await list_days(db, countries=[country], group_key=group_key)
    if not rows:
        raise NotFoundError(f"holiday group {group_key!r} not found for {country}")
    await db.execute(
        delete(HrPublicHoliday).where(
            HrPublicHoliday.country == country,
            HrPublicHoliday.group_key == group_key,
        )
    )
    return len(rows)


async def get_day(db: AsyncSession, *, day_id: int) -> HrPublicHoliday:
    row = (
        await db.execute(
            select(HrPublicHoliday).where(HrPublicHoliday.id == day_id)
        )
    ).scalar_one_or_none()
    if row is None:
        raise NotFoundError(f"holiday day {day_id} not found")
    return row


async def update_day(
    db: AsyncSession,
    *,
    day_id: int,
    day: date | None = None,
    name_en: str | None = None,
    name_local: str | None = None,
    day_type: str | None = None,
    note: str | None = None,
    clear_local: bool = False,
    clear_note: bool = False,
    actor_id: str | None = None,
) -> HrPublicHoliday:
    """Edit one day in place. ``None`` means "leave unchanged"."""
    row = await get_day(db, day_id=day_id)
    new_day = day or row.day
    new_name = (name_en or row.name_en).strip()
    if not new_name:
        raise ValidationError("name_en must not be empty")
    new_key = name_key_for(new_name)

    if (new_day, new_key) != (row.day, row.name_key):
        clash = (
            await db.execute(
                select(HrPublicHoliday.id).where(
                    HrPublicHoliday.country == row.country,
                    HrPublicHoliday.day == new_day,
                    HrPublicHoliday.name_key == new_key,
                    HrPublicHoliday.id != row.id,
                )
            )
        ).scalar_one_or_none()
        if clash is not None:
            raise ConflictError(
                f"{new_day.isoformat()} already carries a holiday named "
                f"{new_name!r} in {row.country}",
            )
    validate_year(new_day.year)

    row.day = new_day
    row.name_en = new_name
    row.name_key = new_key
    if clear_local:
        row.name_local = None
    elif name_local is not None:
        row.name_local = name_local
    if day_type is not None:
        row.day_type = validate_day_type(day_type)
    if clear_note:
        row.note = None
    elif note is not None:
        row.note = note
    row.updated_by_id = actor_id
    await db.flush()
    await _ensure_calendar(
        db, country=row.country, year=new_day.year, actor_id=actor_id,
    )
    return row


async def delete_day(db: AsyncSession, *, day_id: int) -> HrPublicHoliday:
    row = await get_day(db, day_id=day_id)
    await db.delete(row)
    await db.flush()
    return row


@dataclass
class BulkRow:
    """One inbound row from an Excel import, AI parse, or batch API call."""

    country: str
    day: date
    name_en: str
    name_local: str | None = None
    day_type: str = PUBLIC_HOLIDAY
    note: str | None = None
    group_key: str | None = None


@dataclass
class BulkResult:
    created: int = 0
    updated: int = 0
    skipped: int = 0
    calendars_created: int = 0
    # Human-readable, row-addressed notes for the console's result panel.
    details: list[str] = field(default_factory=list)


async def bulk_upsert(
    db: AsyncSession,
    *,
    rows: Sequence[BulkRow],
    source: str = SOURCE_MANUAL,
    actor_id: str | None = None,
    overwrite: bool = False,
) -> BulkResult:
    """Insert (or optionally update) many rows across countries.

    ``overwrite=False`` — the default and what the import preview promises —
    leaves an existing (country, day, name) row exactly as it is and counts it as
    skipped, so re-importing a spreadsheet can never quietly revert an HR edit.
    ``overwrite=True`` refreshes the local name, note and day type of matched
    rows.
    """
    if not rows:
        return BulkResult()
    if len(rows) > MAX_BULK_ROWS:
        raise ValidationError(
            f"a bulk write may not exceed {MAX_BULK_ROWS} rows (got {len(rows)})",
        )
    source = validate_source(source)
    result = BulkResult()

    # Resolve and validate everything before touching the DB, so a bad row on
    # line 300 fails the request instead of half-applying it.
    prepared: list[tuple[str, date, str, str | None, str, str | None, str | None]] = []
    for idx, raw in enumerate(rows, start=1):
        country = resolve_country(raw.country, field_name=f"row {idx} country")
        name_en = (raw.name_en or "").strip()
        if not name_en:
            raise ValidationError(f"row {idx}: name_en is required")
        validate_year(raw.day.year)
        prepared.append((
            country, raw.day, name_en, (raw.name_local or None),
            validate_day_type(raw.day_type), (raw.note or None), raw.group_key,
        ))

    countries = {p[0] for p in prepared}
    existing: dict[tuple[str, date, str], HrPublicHoliday] = {}
    for country in countries:
        for row in await list_days(db, countries=[country]):
            existing[(country, row.day, row.name_key)] = row

    # Group keys for rows that did not bring one.
    #
    # Grouping by (country, year, name) alone was wrong: Japan observes two
    # separate "New Year Holidays" runs in one year (Jan 1-2 and Dec 29-31), and
    # an imported year merged them into a single group spanning January to
    # December — which the Word export then rendered as one holiday and
    # ``shortRange`` printed as "1/1 – 12/31". The seed is unaffected because it
    # carries explicit keys, but import and AI parse are the primary paths.
    #
    # So same-named rows are walked in date order and split on a gap wider than
    # ``_GROUP_GAP_DAYS``. That window has to stay comfortably above the widest
    # real span — China's National Day reaches from its 09-20 makeup workday to
    # its 10-10 one, 21 days — while staying far below the months that separate
    # two genuinely distinct runs of the same name.
    def _base_key(name_en: str, day_type: str) -> str:
        return group_base_key(name_en, is_makeup=day_type == MAKEUP_WORKDAY)

    runs: dict[tuple[str, str], list[list[date]]] = defaultdict(list)
    keyless = [row for row in prepared if not row[6]]
    for country, day, name_en, _local, day_type, _note, _gk in sorted(
        keyless, key=lambda r: (r[0], _base_key(r[2], r[4]), r[1]),
    ):
        bucket = runs[(country, _base_key(name_en, day_type))]
        if bucket and (day - bucket[-1][-1]).days <= _GROUP_GAP_DAYS:
            bucket[-1].append(day)
        else:
            bucket.append([day])

    # Reuse an existing group only when it is near this run — the same name far
    # away in the calendar is a different holiday, not a continuation.
    #
    # Indexed rather than scanned: the obvious `for r in existing.values()` inside
    # the run loop recomputes ``_base_key`` (three regex substitutions) for every
    # stored row on every run, which measured 2.9s of blocking CPU at 1000 runs ×
    # 3000 rows — and it sits between the read and the flush, so that is a
    # multi-second write-transaction hold on a single-threaded event loop. The
    # same shape as the outbox FK-lock and TiDB-1205 incidents.
    by_name: dict[tuple[str, str], list[tuple[date, str]]] = defaultdict(list)
    for row in existing.values():
        by_name[(row.country, _base_key(row.name_en, row.day_type))].append(
            (row.day, row.group_key),
        )

    allocated: dict[tuple[str, str, date], str] = {}
    for (country, base_key), buckets in runs.items():
        candidates = by_name.get((country, base_key), [])
        for run_days in buckets:
            match = next(
                (
                    existing_key
                    for existing_day, existing_key in candidates
                    if min(abs((existing_day - d).days) for d in run_days)
                    <= _GROUP_GAP_DAYS
                ),
                None,
            )
            run_group_key = match or await allocate_group_key(
                db, country=country, year=run_days[0].year,
                name_en=base_key,
            )
            for run_day in run_days:
                allocated[(country, base_key, run_day)] = run_group_key

    touched_years: set[tuple[str, int]] = set()
    for country, day, name_en, local, day_type, note, group_key in prepared:
        key = (country, day, name_key_for(name_en))
        touched_years.add((country, day.year))
        current = existing.get(key)
        if current is not None:
            if not overwrite:
                result.skipped += 1
                result.details.append(
                    f"{country} {day.isoformat()} {name_en}: already present, kept"
                )
                continue
            current.name_local = local
            current.day_type = day_type
            current.note = note
            current.updated_by_id = actor_id
            result.updated += 1
            continue
        if not group_key:
            group_key = allocated[(country, _base_key(name_en, day_type), day)]
        row = HrPublicHoliday(
            country=country, day=day, group_key=group_key,
            name_en=name_en, name_local=local, name_key=key[2],
            day_type=day_type, note=note, source=source,
            created_by_id=actor_id, updated_by_id=actor_id,
        )
        db.add(row)
        existing[key] = row
        result.created += 1

    await db.flush()
    for country, year in sorted(touched_years):
        if await _ensure_calendar(
            db, country=country, year=year, actor_id=actor_id,
        ):
            result.calendars_created += 1
    return result


def _makeup_base_name(name_en: str) -> str:
    """``"Makeup Workday (Labor Day)"`` → ``"Labor Day"``.

    Lets an imported makeup row join the same group as the holiday it offsets
    without the spreadsheet having to carry group keys.
    """
    text = (name_en or "").strip()
    lowered = text.lower()
    if lowered.startswith("makeup workday (") and text.endswith(")"):
        return text[len("makeup workday ("):-1].strip() or text
    return text


async def _ensure_calendars_for(
    db: AsyncSession,
    *,
    country: str,
    days: Iterable[date],
    actor_id: str | None,
) -> int:
    """Ensure a coverage row for EVERY year the given days touch.

    A holiday may cross a year boundary (2026-12-31 to 2027-01-02 is a real shape
    HR enters). Ensuring only the anchor year left the January days stored but
    *invisible on the External API*: with no 2027 calendar row that year reports
    ``coverage="no_data"``, and the contract says a ``no_data`` year returns an
    empty ``holidays`` array — so a scheduler would never see them.
    """
    created = 0
    for year in sorted({d.year for d in days}):
        if await _ensure_calendar(
            db, country=country, year=year, actor_id=actor_id,
        ):
            created += 1
    return created


async def _ensure_calendar(
    db: AsyncSession, *, country: str, year: int, actor_id: str | None,
) -> bool:
    """Create the (country, year) calendar row if the days imply one.

    Days and calendar metadata are separate tables but not independent facts: a
    country-year with holidays and no calendar row would report ``no_data`` on
    the External API while visibly having holidays. Any write that lands a day
    guarantees the coverage row exists.
    """
    existing = await get_calendar(db, country=country, year=year)
    if existing is not None:
        return False
    label = SEED_2026.get(country, {}).get("label") if year == SEED_YEAR else None
    db.add(HrPublicHolidayCalendar(
        country=country,
        year=year,
        label=label or country,
        source=SOURCE_MANUAL,
        created_by_id=actor_id,
        updated_by_id=actor_id,
    ))
    await db.flush()
    return True


async def upsert_calendar(
    db: AsyncSession,
    *,
    country: str,
    year: int,
    label: str | None = None,
    stated_days: int | None = None,
    gcal_url: str | None = None,
    region_note: str | None = None,
    clear_gcal: bool = False,
    clear_note: bool = False,
    clear_stated: bool = False,
    clear_label: bool = False,
    actor_id: str | None = None,
) -> HrPublicHolidayCalendar:
    country = resolve_country(country)
    validate_year(year)
    if gcal_url:
        gcal_url = _validate_gcal_url(gcal_url)
    if stated_days is not None and not 0 <= stated_days <= 366:
        raise ValidationError("stated_days must be between 0 and 366")

    row = await get_calendar(db, country=country, year=year)
    if row is None:
        row = HrPublicHolidayCalendar(
            country=country, year=year, source=SOURCE_MANUAL,
            created_by_id=actor_id,
        )
        db.add(row)
    # ``clear_label`` exists for the same reason as its three siblings: with only
    # ``label=None`` meaning "unchanged", clearing a label was a silent no-op
    # that still reported success.
    # Any HR edit takes the row out of restore-defaults' fill scope for good.
    row.source = SOURCE_MANUAL
    if clear_label:
        row.label = None
    elif label is not None:
        row.label = label.strip() or None
    if clear_stated:
        row.stated_days = None
    elif stated_days is not None:
        row.stated_days = stated_days
    if clear_gcal:
        row.gcal_url = None
    elif gcal_url is not None:
        row.gcal_url = gcal_url
    if clear_note:
        row.region_note = None
    elif region_note is not None:
        row.region_note = region_note.strip() or None
    row.updated_by_id = actor_id
    await db.flush()
    return row


# Subscription links are rendered as clickable buttons on a page every employee
# sees, so the stored value is restricted to the one host we actually publish.
#
# ``www.google.com`` was here initially and had to go: it is Google's own open
# redirector, so ``https://www.google.com/url?q=http://evil.com`` passed the
# check and the employee-facing button would have sent every reader to an
# attacker-chosen host — defeating the entire point of the allow-list. All 22
# seeded links are on ``calendar.google.com``, so nothing legitimate used it.
_ALLOWED_GCAL_HOSTS = frozenset({"calendar.google.com"})


def _validate_gcal_url(url: str) -> str:
    from urllib.parse import urlsplit, urlunsplit

    value = url.strip()
    if len(value) > 512:
        raise ValidationError("gcal_url must not exceed 512 characters")
    if any(ch in value for ch in "\r\n\t"):
        raise ValidationError("gcal_url must not contain control characters")
    try:
        parsed = urlsplit(value)
        host = parsed.hostname
    except ValueError as exc:
        # ``urlsplit`` raises on shapes like ``https://[::1/``. Left to escape,
        # the generic handler turns that into a 500 instead of a field error.
        raise ValidationError(f"gcal_url is not a valid URL: {exc}") from exc
    if parsed.scheme != "https" or host not in _ALLOWED_GCAL_HOSTS:
        raise ValidationError(
            "gcal_url must be an https link on "
            + ", ".join(sorted(_ALLOWED_GCAL_HOSTS)),
        )
    if parsed.username or parsed.password:
        # Credentials in the URL end up in the Word export's hyperlink
        # relationship, which is a document HR mails around.
        raise ValidationError("gcal_url must not carry credentials")
    # Store the normalised form: the host lower-cased and the default port
    # dropped, so the value in the DB is the value browsers will resolve.
    netloc = host if parsed.port in (None, 443) else f"{host}:{parsed.port}"
    return urlunsplit(("https", netloc, parsed.path, parsed.query, parsed.fragment))


async def delete_calendar(
    db: AsyncSession, *, country: str, year: int, cascade: bool = False,
) -> dict[str, int]:
    """Remove a country-year calendar, optionally with its days.

    Refuses to orphan days: without ``cascade`` a calendar that still has
    holidays cannot be deleted, because dropping only the metadata row would
    leave the country reporting ``no_data`` while still serving holidays.
    """
    country = resolve_country(country)
    validate_year(year)
    row = await get_calendar(db, country=country, year=year)
    if row is None:
        raise NotFoundError(f"no {year} calendar for {country}")
    start, end = year_bounds(year)
    day_rows = await list_days(db, countries=[country], year=year)
    if day_rows and not cascade:
        raise ConflictError(
            f"{country} {year} still has {len(day_rows)} day(s); "
            "delete them first or pass cascade=true",
        )
    if day_rows:
        await db.execute(
            delete(HrPublicHoliday).where(
                HrPublicHoliday.country == country,
                HrPublicHoliday.day >= start,
                HrPublicHoliday.day <= end,
            )
        )
    await db.delete(row)
    await db.flush()
    return {"days_deleted": len(day_rows)}


async def restore_seed(
    db: AsyncSession, *, actor_id: str | None = None,
) -> dict[str, int]:
    """Re-apply the shipped 2026 baseline without disturbing HR's edits.

    Adds back anything missing and fills in blank calendar metadata. Rows HR has
    since edited or added keep their values: this is a "restore what shipped",
    not a "revert to what shipped" (same contract as the finance close
    calendar's restore-defaults).
    """
    added_days = 0
    added_calendars = 0
    filled_calendars = 0
    for country, spec in SEED_2026.items():
        cal = await get_calendar(db, country=country, year=SEED_YEAR)
        if cal is None:
            db.add(HrPublicHolidayCalendar(
                country=country, year=SEED_YEAR, label=spec["label"],
                stated_days=spec["stated_days"], gcal_url=spec["gcal_url"],
                region_note=spec["region_note"], source=SOURCE_SEED,
                created_by_id=actor_id, updated_by_id=actor_id,
            ))
            added_calendars += 1
        elif cal.source == SOURCE_SEED:
            # Only fill a row still marked ``seed``. A blank field cannot tell
            # "never set" from "HR deliberately cleared it", so filling blanks on
            # an HR-touched row silently undid every ``clear_*`` on the next
            # restore. Any edit through ``upsert_calendar`` stamps the row
            # ``manual``, which takes it out of scope here for good.
            changed = False
            for attr, value in (
                ("label", spec["label"]),
                ("stated_days", spec["stated_days"]),
                ("gcal_url", spec["gcal_url"]),
                ("region_note", spec["region_note"]),
            ):
                if getattr(cal, attr) in (None, "") and value is not None:
                    setattr(cal, attr, value)
                    changed = True
            if changed:
                cal.updated_by_id = actor_id
                filled_calendars += 1

        taken = await _existing_keys(db, country=country, year=SEED_YEAR)
        # Group keys already present, so a seeded holiday HR has since MOVED or
        # RENAMED is not resurrected alongside its edited self. Keyed on
        # (day, name_key) alone, moving Ching Ming from 04-04 to 04-06 freed the
        # seed's key and the next restore re-inserted the original — leaving two
        # rows for one holiday, double-counted in ``total_days``.
        seeded_groups = set(
            (
                await db.execute(
                    select(HrPublicHoliday.group_key).where(
                        HrPublicHoliday.country == country,
                    ).distinct()
                )
            ).scalars().all()
        )
        for day_str, group_key, name_en, name_local, day_type in spec["days"]:
            day = date.fromisoformat(day_str)
            key = name_key_for(name_en)
            if (day, key) in taken or group_key in seeded_groups:
                continue
            taken.add((day, key))
            db.add(HrPublicHoliday(
                country=country, day=day, group_key=group_key,
                name_en=name_en, name_local=name_local, name_key=key,
                day_type=day_type, source=SOURCE_SEED,
                created_by_id=actor_id, updated_by_id=actor_id,
            ))
            added_days += 1
    await db.flush()
    return {
        "days_added": added_days,
        "calendars_added": added_calendars,
        "calendars_filled": filled_calendars,
        "year": SEED_YEAR,
    }


async def roll_forward_year(
    db: AsyncSession,
    *,
    from_year: int,
    to_year: int,
    countries: Sequence[str] | None = None,
    actor_id: str | None = None,
) -> dict[str, object]:
    """Open a new year by copying the country roster — and only the roster.

    Every December HR has to build next year's calendar for 22 countries. The
    tedious, unchanging part is the roster itself: which countries exist, their
    display labels, their regional-holiday advisories. That is what this copies.

    **Dates are deliberately NOT copied.** Almost every holiday moves year to
    year (Chinese New Year, Easter, every "(observed)" shift), so a copied date
    is a *wrong* date that looks authoritative — the worst possible default for a
    calendar employees plan around. HR fills the dates in per country via the
    Excel import or the AI parse, both of which are built for exactly that.

    ``gcal_url`` is not copied either: a subscription link points at a specific
    year's Google calendar, so carrying it over would silently subscribe
    employees to last year's holidays. ``stated_days`` starts blank for the same
    reason — it is a published figure, and there is nothing published yet.

    Idempotent: a country that already has a ``to_year`` row is left untouched
    and counted as skipped, so running it twice is safe.
    """
    validate_year(from_year)
    validate_year(to_year)
    if from_year == to_year:
        raise ValidationError("from_year and to_year must differ")

    source = await list_calendars(db, year=from_year, countries=countries)
    if not source:
        raise NotFoundError(
            f"no {from_year} calendar to copy from"
            + (f" for {', '.join(countries)}" if countries else ""),
        )
    existing = {
        cal.country for cal in await list_calendars(db, year=to_year)
    }

    created: list[str] = []
    skipped: list[str] = []
    for cal in source:
        if cal.country in existing:
            skipped.append(cal.country)
            continue
        db.add(HrPublicHolidayCalendar(
            country=cal.country,
            year=to_year,
            label=cal.label,
            # stated_days / gcal_url intentionally left unset — see the docstring.
            region_note=cal.region_note,
            source=SOURCE_MANUAL,
            created_by_id=actor_id,
            updated_by_id=actor_id,
        ))
        created.append(cal.country)
    await db.flush()
    return {
        "from_year": from_year,
        "to_year": to_year,
        "countries_created": sorted(created),
        "countries_skipped": sorted(skipped),
        "created": len(created),
        "skipped": len(skipped),
    }


async def duplicate_keys(
    db: AsyncSession,
    *,
    years: Sequence[int] | None = None,
    countries: Sequence[str] | None = None,
) -> set[tuple[str, str, str]]:
    """``(country, iso day, name_key)`` triples, for marking duplicate imports.

    Projects the three columns instead of hydrating ORM rows: the callers only
    need the key set, and an earlier version read every row of every year as full
    objects to build it — small today at ~330 rows a year, unbounded over time.

    Defaults to the current year and its neighbours. An import targets a specific
    year; the window keeps a stale row from a decade ago out of the scan without
    ever hiding a duplicate in the year being imported.
    """
    stmt = select(
        HrPublicHoliday.country, HrPublicHoliday.day, HrPublicHoliday.name_key,
    )
    if countries:
        stmt = stmt.where(HrPublicHoliday.country.in_(list(countries)))
    scope = list(years) if years else None
    if scope is None:
        current = date.today().year
        scope = [current - 1, current, current + 1, current + 2]
    lo, hi = min(scope), max(scope)
    stmt = stmt.where(
        HrPublicHoliday.day >= date(validate_year(lo), 1, 1),
        HrPublicHoliday.day <= date(validate_year(hi), 12, 31),
    )
    return {
        (row[0], row[1].isoformat(), row[2])
        for row in (await db.execute(stmt)).all()
    }


async def day_is_holiday(
    db: AsyncSession, *, country: str, day: date,
) -> list[HrPublicHoliday]:
    """Holiday rows on one date for one country (``[]`` when it is a workday)."""
    stmt = select(HrPublicHoliday).where(
        HrPublicHoliday.country == country,
        HrPublicHoliday.day == day,
    ).order_by(HrPublicHoliday.name_en)
    return list((await db.execute(stmt)).scalars().all())


async def country_year_summary(
    db: AsyncSession, *, year: int, countries: Sequence[str] | None = None,
) -> list[dict[str, Any]]:
    """Per-country roll-up for the HR console's overview table.

    Reports HR's ``stated_days`` next to the derived observed count and an
    explicit ``stated_days_matches_derived`` flag, so a divergence is visible
    without the console having to re-derive the rule.
    """
    validate_year(year)
    calendars = await list_calendars(db, year=year, countries=countries)
    codes = [c.country for c in calendars]
    rows = await list_days(db, year=year, countries=codes or None) if codes else []
    by_country: dict[str, list[HrPublicHoliday]] = defaultdict(list)
    for row in rows:
        by_country[row.country].append(row)

    out: list[dict[str, Any]] = []
    for cal in calendars:
        country_rows = by_country.get(cal.country, [])
        holiday_days = [r.day for r in country_rows if r.day_type == PUBLIC_HOLIDAY]
        makeup_days = [r.day for r in country_rows if r.day_type == MAKEUP_WORKDAY]
        derived = observed_day_count(holiday_days, makeup_days)
        groups = build_groups(country_rows)
        out.append({
            "country": cal.country,
            "label": cal.label or cal.country,
            "year": cal.year,
            "stated_days": cal.stated_days,
            "observed_days": derived,
            "stated_days_matches_derived": (
                cal.stated_days is None or cal.stated_days == derived
            ),
            "calendar_days": len(set(holiday_days)),
            "makeup_days": len(set(makeup_days)),
            "group_count": len([g for g in groups if not g.is_makeup_only]),
            "gcal_url": cal.gcal_url,
            "region_note": cal.region_note,
            "source": cal.source,
        })
    return out


def month_matrix(
    rows: Sequence[HrPublicHoliday], *, year: int, month: int,
) -> dict[str, list[dict[str, Any]]]:
    """Index rows by ISO date for the month grid: ``{"2026-05-01": [...]}``.

    Filtering here rather than in SQL lets one year-wide fetch drive twelve
    months of grid without a query per month.
    """
    out: dict[str, list[dict[str, Any]]] = defaultdict(list)
    for row in rows:
        if row.day.year != year or row.day.month != month:
            continue
        out[row.day.isoformat()].append({
            "id": row.id,
            "country": row.country,
            "group_key": row.group_key,
            "name_en": row.name_en,
            "name_local": row.name_local,
            "day_type": row.day_type,
            "note": row.note,
        })
    return dict(out)


async def complete_groups(
    db: AsyncSession, rows: Sequence[HrPublicHoliday],
) -> list[HolidayGroup]:
    """Fold rows into groups, widening any group the row window clipped.

    A year-scoped fetch cuts a cross-year holiday in half, and folding only the
    visible half produced a group that *looks* whole but is not: a VN holiday
    running 2026-12-31 to 2027-01-02 reported ``total_days = 1`` on the employee
    page for 2026 while the notice HR sends — which fetches by ``group_key`` and
    is therefore unclipped — said 3 days. Two surfaces, same holiday, different
    number.

    So any group whose rows touch a window edge is refetched by its key. The
    common case (no group at an edge) costs nothing.
    """
    if not rows:
        return []
    partial = {(r.country, r.group_key) for r in rows}
    days = [r.day for r in rows]
    lo, hi = min(days), max(days)
    at_edge = {
        (r.country, r.group_key) for r in rows if r.day == lo or r.day == hi
    }
    if not at_edge:
        return build_groups(rows)

    # Widen by a month either side: no real holiday, makeup days included, spans
    # further than that (China's National Day reaches 21 days).
    widened = await list_days(
        db,
        countries=sorted({c for c, _ in at_edge}),
        start=lo - timedelta(days=_GROUP_GAP_DAYS),
        end=hi + timedelta(days=_GROUP_GAP_DAYS),
    )
    merged = {(r.country, r.group_key, r.day): r for r in rows}
    for row in widened:
        if (row.country, row.group_key) in at_edge:
            merged[(row.country, row.group_key, row.day)] = row
    return [
        g for g in build_groups(merged.values())
        if (g.country, g.group_key) in partial
    ]


async def search_groups(
    db: AsyncSession,
    *,
    year: int,
    countries: Sequence[str] | None = None,
    query: str | None = None,
) -> list[HolidayGroup]:
    """Holiday groups for a year, optionally name-filtered.

    Name matching runs in SQL over both the English and local names so a
    Chinese-only search term ("春节") still finds its group.
    """
    validate_year(year)
    start, end = year_bounds(year)
    stmt = select(HrPublicHoliday).where(
        HrPublicHoliday.day >= start, HrPublicHoliday.day <= end,
    )
    if countries:
        stmt = stmt.where(HrPublicHoliday.country.in_(list(countries)))
    if query:
        # Escape the LIKE metacharacters. Unescaped, a search for "%" matched
        # every holiday and "_" matched any single character — the needle is a
        # bind parameter so this was never injection, but it was a wrong answer,
        # and a pattern of many "%" is a needless full scan.
        needle = f"%{_escape_like(query.strip())}%"
        stmt = stmt.where(
            or_(
                HrPublicHoliday.name_en.like(needle, escape="\\"),
                and_(
                    HrPublicHoliday.name_local.is_not(None),
                    HrPublicHoliday.name_local.like(needle, escape="\\"),
                ),
            )
        )
    rows = list((await db.execute(stmt)).scalars().all())
    if not query:
        return await complete_groups(db, rows)
    # A name filter can match only some days of a group; refetch the matched
    # groups whole so the console never renders a truncated holiday.
    keys = {(r.country, r.group_key) for r in rows}
    if not keys:
        return []
    full = await list_days(
        db, countries=sorted({c for c, _ in keys}), start=start, end=end,
    )
    return [g for g in build_groups(full) if (g.country, g.group_key) in keys]