Add live activity token retention, notify usage and clean up cycle to mobile_app

homeassistant/components/mobile_app/__init__.py

@@ -46,16 +46,19 @@
     DATA_CONFIG_ENTRIES,
     DATA_DELETED_IDS,
     DATA_DEVICES,
+    DATA_LIVE_ACTIVITY_TOKENS,
     DATA_PENDING_UPDATES,
     DATA_PUSH_CHANNEL,
     DATA_STORE,
     DOMAIN,
     SENSOR_TYPES,
     STORAGE_KEY,
     STORAGE_VERSION,
+    STORAGE_VERSION_MINOR,
 )
 from .helpers import async_is_local_only_user, savable_state
 from .http_api import RegistrationsView
+from .live_activity import async_cleanup_expired_tokens
 from .timers import async_handle_timer_event
 from .util import async_create_cloud_hook, supports_push
 from .webhook import handle_webhook
@@ -72,24 +75,26 @@
 
 async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
     """Set up the mobile app component."""
-    store = Store[dict[str, Any]](hass, STORAGE_VERSION, STORAGE_KEY)
+    store = _MobileAppStore(
+        hass, STORAGE_VERSION, STORAGE_KEY, minor_version=STORAGE_VERSION_MINOR
+    )
     if (app_config := await store.async_load()) is None or not isinstance(
         app_config, dict
     ):
-        app_config = {
-            DATA_CONFIG_ENTRIES: {},
-            DATA_DELETED_IDS: [],
-        }
+        app_config = {DATA_DELETED_IDS: [], DATA_LIVE_ACTIVITY_TOKENS: {}}
 
     hass.data[DOMAIN] = {
         DATA_CONFIG_ENTRIES: {},
-        DATA_DELETED_IDS: app_config.get(DATA_DELETED_IDS, []),
+        DATA_DELETED_IDS: app_config[DATA_DELETED_IDS],
         DATA_DEVICES: {},
+        DATA_LIVE_ACTIVITY_TOKENS: app_config[DATA_LIVE_ACTIVITY_TOKENS],
         DATA_PUSH_CHANNEL: {},
         DATA_STORE: store,
         DATA_PENDING_UPDATES: {sensor_type: {} for sensor_type in SENSOR_TYPES},
     }
 
+    hass.async_create_task(async_cleanup_expired_tokens(hass))
+
     hass.http.register_view(RegistrationsView())
 
     for deleted_id in hass.data[DOMAIN][DATA_DELETED_IDS]:
@@ -243,10 +248,25 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
 
 async def async_remove_entry(hass: HomeAssistant, entry: ConfigEntry) -> None:
     """Cleanup when entry is removed."""
-    hass.data[DOMAIN][DATA_DELETED_IDS].append(entry.data[CONF_WEBHOOK_ID])
+    webhook_id = entry.data[CONF_WEBHOOK_ID]
+    hass.data[DOMAIN][DATA_DELETED_IDS].append(webhook_id)
+    hass.data[DOMAIN][DATA_LIVE_ACTIVITY_TOKENS].pop(webhook_id, None)
     store = hass.data[DOMAIN][DATA_STORE]
     await store.async_save(savable_state(hass))
 
     if CONF_CLOUDHOOK_URL in entry.data:
         with suppress(cloud.CloudNotAvailable, ValueError):
             await cloud.async_delete_cloudhook(hass, entry.data[CONF_WEBHOOK_ID])
+
+
+class _MobileAppStore(Store[dict[str, Any]]):
+    async def _async_migrate_func(
+        self,
+        old_major_version: int,
+        old_minor_version: int,
+        old_data: dict[str, Any],
+    ) -> dict[str, Any]:
+        """Migrate mobile_app storage to the current version."""
+        if old_major_version == 1 and old_minor_version < 2:
+            old_data.setdefault(DATA_LIVE_ACTIVITY_TOKENS, {})
+        return old_data

homeassistant/components/mobile_app/const.py

@@ -8,6 +8,8 @@
 
 STORAGE_KEY = DOMAIN
 STORAGE_VERSION = 1
+STORAGE_VERSION_MINOR = 2
+STORAGE_SAVE_DELAY_SECONDS = 10
 
 CONF_CLOUDHOOK_URL = "cloudhook_url"
 CONF_REMOTE_UI_URL = "remote_ui_url"
@@ -17,6 +19,7 @@
 DATA_CONFIG_ENTRIES = "config_entries"
 DATA_DELETED_IDS = "deleted_ids"
 DATA_DEVICES = "devices"
+DATA_LIVE_ACTIVITY_TOKENS = "live_activity_tokens"
 DATA_STORE = "store"
 DATA_NOTIFY = "notify"
 DATA_PUSH_CHANNEL = "push_channel"
@@ -40,6 +43,14 @@
 ATTR_PUSH_RATE_LIMITS_SUCCESSFUL = "successful"
 ATTR_SUPPORTS_ENCRYPTION = "supports_encryption"
 
+ATTR_LIVE_UPDATE = "live_update"
+ATTR_LIVE_ACTIVITY_TOKEN = "live_activity_token"
+ATTR_EXPIRES_AT = "expires_at"
+ATTR_TOKEN = "token"
+ATTR_TAG = "tag"
+
+CLEAR_NOTIFICATION = "clear_notification"
+
 ATTR_EVENT_DATA = "event_data"
 ATTR_EVENT_TYPE = "event_type"
 
@@ -92,6 +103,7 @@
         # will connect via websocket channel to receive
         # push notifications.
         vol.Optional(ATTR_PUSH_WEBSOCKET_CHANNEL): cv.boolean,
+        vol.Optional(ATTR_LIVE_ACTIVITY_TOKEN): cv.string,
     },
     extra=vol.ALLOW_EXTRA,
 )

homeassistant/components/mobile_app/helpers.py

@@ -32,6 +32,7 @@
     CONF_SECRET,
     CONF_USER_ID,
     DATA_DELETED_IDS,
+    DATA_LIVE_ACTIVITY_TOKENS,
     DOMAIN,
 )
 
@@ -170,10 +171,11 @@ def safe_registration(registration: dict) -> dict:
 
 def savable_state(hass: HomeAssistant) -> dict:
     """Return a clean object containing things that should be saved."""
+    # pylint: disable-next=home-assistant-use-runtime-data
+    domain_data = hass.data[DOMAIN]
     return {
-        # Uses legacy hass.data[DOMAIN] pattern
-        # pylint: disable-next=home-assistant-use-runtime-data
-        DATA_DELETED_IDS: hass.data[DOMAIN][DATA_DELETED_IDS],
+        DATA_DELETED_IDS: domain_data[DATA_DELETED_IDS],
+        DATA_LIVE_ACTIVITY_TOKENS: domain_data[DATA_LIVE_ACTIVITY_TOKENS],
     }
 
 

homeassistant/components/mobile_app/live_activity.py

@@ -0,0 +1,71 @@
+"""Live Activity push token lifecycle: expiry-driven cleanup loop."""
+# pylint: disable=home-assistant-use-runtime-data  # Uses legacy hass.data[DOMAIN] pattern
+
+from datetime import datetime
+
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.helpers.event import async_call_later
+from homeassistant.util import dt as dt_util
+
+from .const import ATTR_EXPIRES_AT, DATA_LIVE_ACTIVITY_TOKENS, DATA_STORE, DOMAIN
+from .helpers import savable_state
+
+
+@callback
+def async_schedule_next_cleanup(hass: HomeAssistant) -> None:
+    """Schedule a sweep for the earliest token expiry.
+
+    Only call when no sweep is already in flight. The invariant is: tokens
+    non-empty ⟹ sweep scheduled. So the two safe call sites are (1) the
+    webhook that added the first token (tokens was empty beforehand) and
+    (2) the tail of a sweep that found surviving tokens (its own timer
+    just fired).
+    """
+    tokens = hass.data[DOMAIN][DATA_LIVE_ACTIVITY_TOKENS]
+    earliest_expires_at = min(
+        (
+            token[ATTR_EXPIRES_AT]
+            for device_tokens in tokens.values()
+            for token in device_tokens.values()
+        ),
+        default=None,
+    )
+    if earliest_expires_at is None:
+        return
+
+    delay = earliest_expires_at - dt_util.utcnow().timestamp()
+
+    async def run_cleanup(_now: datetime) -> None:
+        await async_cleanup_expired_tokens(hass)
+
+    async_call_later(hass, delay, run_cleanup)
+
+
+async def async_cleanup_expired_tokens(hass: HomeAssistant) -> None:
+    """Remove expired tokens and reschedule the next sweep at the earliest expiry.
+
+    Runs as a one-shot callback scheduled by ``async_schedule_next_cleanup``. After
+    sweeping, if any tokens remain it calls ``async_schedule_next_cleanup`` again to
+    queue the next sweep — this self-rescheduling chain is what "the loop" refers to.
+    When tokens are empty no further sweep is scheduled; the chain restarts the next
+    time the webhook stores a token into an empty store.
+    """
+    now = dt_util.utcnow().timestamp()
+    tokens = hass.data[DOMAIN][DATA_LIVE_ACTIVITY_TOKENS]
+    changed = False
+
+    for webhook_id in list(tokens):
+        device_tokens = tokens[webhook_id]
+        for tag, data in list(device_tokens.items()):
+            if data[ATTR_EXPIRES_AT] <= now:
+                del device_tokens[tag]
+                changed = True
+        if not device_tokens:
+            del tokens[webhook_id]
+            changed = True
+
+    if tokens:
+        async_schedule_next_cleanup(hass)
+
+    if changed:
+        await hass.data[DOMAIN][DATA_STORE].async_save(savable_state(hass))