Add the nightly missed-quest sweep

Fires one nestquest_quest_missed event per past-due open instance at
the configured rollover time and at startup, watermark-guarded in the
new nestquest_meta_state table (migration 7, strictly additive).
Read-only over the domain tables; never reports an instance that was
ever completed.
This commit is contained in:
2026-09-19 14:15:21 -04:00
parent 25005cda77
commit 3de9f4b970
9 changed files with 473 additions and 8 deletions
+17 -5
View File
@@ -35,6 +35,7 @@ from .migrations import apply_migrations
from .services import async_deregister_services, async_register_services
from .store import async_get_db_path
from .admin_allowlist import seed_setup_admin
from .sweep import run_missed_sweep
_LOGGER = LOGGER
@@ -327,18 +328,23 @@ def _register_day_rollover_listener(
database: NestQuestDatabase,
options: dict[str, Any],
) -> Callable[[], Any]:
"""Register the daily materialization listener and return its remover.
"""Register the daily rollover listener and return its remover.
Fires once per day at the configured local ``day_rollover_time`` and
materializes the rolling horizon ``[today, today + horizon_days]``
through the shared :func:`_run_horizon_materialization` path, using the
entry's configured (validated) ``horizon_days``.
Fires once per day at the configured local ``day_rollover_time``:
first re-materializes the rolling horizon
``[today, today + horizon_days]`` through the shared
:func:`_run_horizon_materialization` path (the entry's configured,
validated ``horizon_days``), then runs the Feature 11 missed sweep
— announcing yesterday's still-open quests as missed AFTER today's
instances exist, watermark-guarded so a re-fire of the listener the
same night announces nothing twice.
"""
hour, minute = _day_rollover_hour_minute(options)
horizon_days = _configured_horizon_days(options)
async def _run_materialization(_now: datetime.datetime) -> None:
await _run_horizon_materialization(hass, database, horizon_days)
await run_missed_sweep(hass, database)
return async_track_time_change(
hass, _run_materialization, hour=hour, minute=minute
@@ -498,6 +504,12 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
await _run_horizon_materialization(
hass, database, _configured_horizon_days(dict(entry.options))
)
# The missed sweep runs at startup too (Feature 11): if HA was
# down at the rollover time, the previous night's still-open
# quests are announced as missed now instead of waiting for
# the next midnight. Watermark-guarded, so this is a no-op
# when the night's sweep already ran.
await run_missed_sweep(hass, database)
_register_services(hass)
# The shared Feature 10 coordinator: one refresh cycle every
# entity reads from (CONF_UPDATE_INTERVAL seconds, default
@@ -493,6 +493,43 @@ class QuestInstancesDao:
)
return [_instance_from_row(row) for row in rows]
async def list_missed_unswept(
self, today_iso: str, watermark: str | None
) -> list[QuestInstanceRecord]:
"""Return past-due open instances the sweep has not announced.
Feature 11's missed-sweep query: ``due_date`` strictly before
``today_iso`` (past due), at or after ``watermark`` when one is
recorded, and with NO completion event ever (the Feature 08
append-only contract: an instance that was ever completed is
never reported missed, even when later reversed — history keeps
both rows and the sweep never mutates any). The watermark is
the HA-local date the sweep last ran, and the window is
``[watermark, today)``: the nightly run entering day D+1 sweeps
exactly day D's instances, so every day's quests are announced
once — at the midnight after their due date — and a same-night
rerun (watermark == today) sees an empty window. Read-only;
ordered deterministically by (due_date, id).
"""
_validate_date(today_iso, "today_iso")
if watermark is not None:
_validate_date(watermark, "watermark")
sql = (
f"SELECT {_INSTANCE_COLUMNS} FROM quest_instances "
"WHERE due_date < ? "
"AND NOT EXISTS ("
" SELECT 1 FROM completion_events "
" WHERE instance_id = quest_instances.id"
")"
)
parameters: tuple = (today_iso,)
if watermark is not None:
sql += " AND due_date >= ?"
parameters += (watermark,)
sql += " ORDER BY due_date, id"
rows = await self._database.fetch_all(sql, parameters)
return [_instance_from_row(row) for row in rows]
async def delete_future_uncompleted(
self, definition_id: int, cutoff_date: str, *, today: datetime.date | None = None
) -> int:
+44
View File
@@ -0,0 +1,44 @@
"""Typed DAO for the integration-owned ``nestquest_meta_state`` table.
Generic persistent key/value bookkeeping that belongs to the
integration itself rather than any domain table — currently the
Feature 11 missed-sweep watermark. Values are plain strings; callers
interpret them. All SQL for this table lives here per the feature
guardrails, and every method runs through
:class:`~.db.NestQuestDatabase` so nothing blocks the event loop.
"""
from __future__ import annotations
from .dao_children import _connection_lock
from .db import NestQuestDatabase
_TABLE = "nestquest_meta_state"
class MetaStateDao:
"""Typed async access to the ``nestquest_meta_state`` table."""
def __init__(self, database: NestQuestDatabase) -> None:
self._database = database
async def get(self, key: str) -> str | None:
"""Return the stored value for ``key``, or None when absent."""
row = await self._database.fetch_one(
f"SELECT value FROM {_TABLE} WHERE key = ?", (key,)
)
return row[0] if row is not None else None
async def set(self, key: str, value: str) -> None:
"""Upsert one key/value row atomically.
The write runs inside one transaction under the shared
connection lock, so a racing reader either sees the old value
or the fully-written new one, never a half-applied state.
"""
async with _connection_lock(self._database):
async with self._database.transaction():
await self._database.execute(
f"INSERT INTO {_TABLE} (key, value) VALUES (?, ?) "
"ON CONFLICT (key) DO UPDATE SET value = excluded.value",
(key, value),
)
+10 -1
View File
@@ -28,7 +28,10 @@ copying each definition's assignee across. Migration 4 adds the
windows table, migration 5 rebuilds ``quest_instances`` onto the
widened (definition_id, child_id, due_date, window) key, and migration
6 rebuilds ``completion_events`` with the D-008 ``actor_child_id``
column. Each callable migration runs inside its own transaction
column. Migration 7 adds the generic ``nestquest_meta_state``
key/value table for integration-owned bookkeeping (Feature 11's
missed-sweep watermark). Each callable migration runs inside its own
transaction
together with its version stamp, so a crash mid-step rolls the
statements and the stamp back together.
@@ -80,6 +83,7 @@ from .schema import (
SCHEMA_V1_STATEMENTS,
SCHEMA_V1_QUEST_DEFINITION_ASSIGNEES_DDL,
SCHEMA_V1_QUEST_DEFINITION_WINDOWS_DDL,
SCHEMA_V7_META_STATE_DDL,
)
LOGGER = logging.getLogger(__name__)
@@ -414,6 +418,11 @@ MIGRATIONS: Sequence[MigrationStep] = [
# actor_child_id column and its actor-pair coherence CHECK,
# backfilling legacy panel rows to their own child_id.
_add_actor_child_id,
# Migration 7 (Feature 11): the generic meta_state key/value table
# — a plain new table, so CREATE IF NOT EXISTS suffices for both
# fresh databases (which already made it in migration 1) and
# pre-v7 databases. Strictly additive; no existing table changes.
list(SCHEMA_V7_META_STATE_DDL),
]
+16
View File
@@ -384,6 +384,21 @@ SCHEMA_V1_COMPLETION_EVENTS_DDL: list[str] = [
""",
]
#: Generic integration-owned key/value state (schema version 7,
#: Feature 11): persistent bookkeeping that belongs to the integration
#: rather than any domain table — currently the missed-sweep
#: watermark. ``value`` is a plain string; callers interpret it.
META_STATE_TABLE_SQL = """(
key TEXT NOT NULL PRIMARY KEY,
value TEXT NOT NULL
)"""
SCHEMA_V7_META_STATE_DDL: list[str] = [
f"""
CREATE TABLE IF NOT EXISTS nestquest_meta_state {META_STATE_TABLE_SQL}
""",
]
SCHEMA_V1_STATEMENTS: list[str] = [
*SCHEMA_V1_CHILDREN_DDL,
*SCHEMA_V1_ADMIN_USERS_DDL,
@@ -395,6 +410,7 @@ SCHEMA_V1_STATEMENTS: list[str] = [
*SCHEMA_V1_PRESENCE_OVERRIDES_DDL,
*SCHEMA_V1_QUEST_INSTANCES_DDL,
*SCHEMA_V1_COMPLETION_EVENTS_DDL,
*SCHEMA_V7_META_STATE_DDL,
]
+103
View File
@@ -0,0 +1,103 @@
"""The nightly missed-quest sweep (Feature 11).
At the configured day-rollover time (the same HA-local listener that
re-materializes the rolling horizon) this job finds every instance
whose ``due_date`` is before the HA-local date, that has NO completion
event, and that falls after the persisted sweep watermark — and fires
ONE ``nestquest_quest_missed`` event per instance with the documented
payload. The sweep is read-only over the domain tables: it never
mutates an instance and never writes a completion event (``missed``
is derived state, never an event_type — Feature 08's contract).
Idempotency is the watermark: after a run, ``nestquest_meta_state``
records the HA-local date the sweep ran, and the next run only
sweeps instances due in ``[watermark, today)``. A second run the
same night therefore fires nothing new, and a run after days of
downtime sweeps exactly the accumulated window once — no
per-instance journal, no re-announcing old misses. An instance swept
as missed that is later completed simply gains its completion event;
history keeps both rows (append-only), and the sweep's later runs
skip it via the no-completion-event rule.
"""
from __future__ import annotations
import datetime
import logging
from zoneinfo import ZoneInfo
from .const import EVENT_QUEST_MISSED
from .dao_children import ChildrenDao
from .dao_instances import QuestInstancesDao
from .dao_meta import MetaStateDao
from .dao_rules import QuestDefinitionsDao
from .db import NestQuestDatabase
LOGGER = logging.getLogger(__name__)
#: meta_state key holding the HA-local date the sweep last ran.
SWEEP_WATERMARK_KEY = "missed_sweep_last_run_date"
#: UTC ISO-8601 timestamp policy shared with the DAO layers.
_UTC_TIMESTAMP_FORMAT = "%Y-%m-%dT%H:%M:%S+00:00"
def _now_stamp() -> str:
return datetime.datetime.now(datetime.timezone.utc).strftime(
_UTC_TIMESTAMP_FORMAT
)
async def run_missed_sweep(
hass,
database: NestQuestDatabase,
*,
today: datetime.date | None = None,
) -> int:
"""Fire one missed event per unswept past-due open instance.
``today`` optionally pins the HA-local date (threaded like the
materialization and completion layers); when omitted it is read
from ``hass.config.time_zone``. Returns the number of events
fired. Fires nothing and returns 0 when the watermark already
covers ``today`` (a same-night rerun).
"""
today_date = (
today
if today is not None
else datetime.datetime.now(ZoneInfo(hass.config.time_zone)).date()
)
today_iso = today_date.isoformat()
dao = MetaStateDao(database)
watermark = await dao.get(SWEEP_WATERMARK_KEY)
if watermark == today_iso:
LOGGER.debug("NestQuest missed sweep already ran for %s", today_iso)
return 0
instances_dao = QuestInstancesDao(database)
rows = await instances_dao.list_missed_unswept(today_iso, watermark)
children_dao = ChildrenDao(database)
definitions_dao = QuestDefinitionsDao(database)
now_stamp = _now_stamp()
fired = 0
for instance in rows:
child = await children_dao.get(instance.child_id)
definition = await definitions_dao.get(instance.definition_id)
hass.bus.async_fire(
EVENT_QUEST_MISSED,
{
"child_id": instance.child_id,
"child_name": child.display_name if child else None,
"instance_id": instance.id,
"quest_title": definition.title if definition else None,
"window": instance.window,
"due_date": instance.due_date,
"due_time": instance.due_time,
"occurred_at": now_stamp,
},
)
fired += 1
await dao.set(SWEEP_WATERMARK_KEY, today_iso)
LOGGER.info(
"NestQuest missed sweep for %s fired %d event(s)", today_iso, fired
)
return fired