Add iOS Live Activity webhook handlers 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,11 @@
 
 STORAGE_KEY = DOMAIN
 STORAGE_VERSION = 1
+STORAGE_VERSION_MINOR = 2
+STORAGE_SAVE_DELAY = 10
+
+# A Live Activity can be active for up to eight hours unless its app or a person ends it before this limit. https://developer.apple.com/documentation/activitykit/displaying-live-data-with-live-activities#Understand-constraints
+LIVE_ACTIVITY_TOKEN_TTL_SECONDS = 8 * 3600
 
 CONF_CLOUDHOOK_URL = "cloudhook_url"
 CONF_REMOTE_UI_URL = "remote_ui_url"
@@ -17,6 +22,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 +46,10 @@
 ATTR_PUSH_RATE_LIMITS_SUCCESSFUL = "successful"
 ATTR_SUPPORTS_ENCRYPTION = "supports_encryption"
 
+ATTR_LIVE_UPDATE = "live_update"
+ATTR_LIVE_ACTIVITY_TOKEN = "live_activity_token"
+ATTR_TAG = "tag"
+
 ATTR_EVENT_DATA = "event_data"
 ATTR_EVENT_TYPE = "event_type"
 
@@ -92,6 +102,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,64 @@
+"""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 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["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:
+    """Sweep expired tokens, keep the loop alive if any remain, save changes."""
+    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["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))

homeassistant/components/mobile_app/notify.py

@@ -37,6 +37,8 @@
     ATTR_APP_ID,
     ATTR_APP_VERSION,
     ATTR_DEVICE_NAME,
+    ATTR_LIVE_ACTIVITY_TOKEN,
+    ATTR_LIVE_UPDATE,
     ATTR_OS_VERSION,
     ATTR_PUSH_RATE_LIMITS,
     ATTR_PUSH_RATE_LIMITS_ERRORS,
@@ -45,8 +47,10 @@
     ATTR_PUSH_RATE_LIMITS_SUCCESSFUL,
     ATTR_PUSH_TOKEN,
     ATTR_PUSH_URL,
+    ATTR_TAG,
     ATTR_WEBHOOK_ID,
     DATA_CONFIG_ENTRIES,
+    DATA_LIVE_ACTIVITY_TOKENS,
     DATA_NOTIFY,
     DATA_PUSH_CHANNEL,
     DOMAIN,
@@ -234,12 +238,42 @@ async def async_send_message(self, message: str = "", **kwargs: Any) -> None:
                 " not connected to local push notifications"
             )
 
+    async def _get_live_activity_token(
+        self, entry: ConfigEntry, data: dict[str, Any]
+    ) -> str | None:
+        """Return the Live Activity APNs token for this notification, or None."""
+        notification_data = data.get(ATTR_DATA) or {}
+        if not notification_data.get(ATTR_LIVE_UPDATE):
+            return None
+
+        tag = notification_data.get(ATTR_TAG)
+        if not tag:
+            return None
+
+        webhook_id = entry.data[ATTR_WEBHOOK_ID]
+        live_activity_tokens = self.hass.data[DOMAIN][DATA_LIVE_ACTIVITY_TOKENS]
+        device_tokens = live_activity_tokens.get(webhook_id, {})
+        if (stored := device_tokens.get(tag)) and stored[
+            "expires_at"
+        ] > dt_util.utcnow().timestamp():
+            # The activity is already running on the device and the token is valid
+            return stored["token"]
+
+        # Start a new activity remotely
+        app_data = entry.data[ATTR_APP_DATA]
+        return app_data.get(ATTR_LIVE_ACTIVITY_TOKEN)
+
     async def _async_send_remote_message_target(
         self, entry: ConfigEntry, data: dict[str, Any]
-    ):
+    ) -> None:
         """Send a message to a target."""
         try:
-            await _send_message(async_get_clientsession(self.hass), entry, data)
+            await _send_message(
+                async_get_clientsession(self.hass),
+                entry,
+                data,
+                live_activity_token=await self._get_live_activity_token(entry, data),
+            )
         except HomeAssistantError as e:
             if e.translation_key == "rate_limit_exceeded_sending_notification":
                 _LOGGER.warning(str(e))
@@ -248,7 +282,11 @@ async def _async_send_remote_message_target(
 
 
 async def _send_message(
-    session: ClientSession, entry: ConfigEntry, data: dict[str, Any]
+    session: ClientSession,
+    entry: ConfigEntry,
+    data: dict[str, Any],
+    *,
+    live_activity_token: str | None = None,
 ) -> None:
     """Shared internal helper to send messages via cloud push notification services."""
     reg_info = {
@@ -259,15 +297,19 @@ async def _send_message(
     if ATTR_OS_VERSION in entry.data:
         reg_info[ATTR_OS_VERSION] = entry.data[ATTR_OS_VERSION]
 
+    payload: dict[str, Any] = {
+        **data,
+        ATTR_PUSH_TOKEN: entry.data[ATTR_APP_DATA][ATTR_PUSH_TOKEN],
+        "registration_info": reg_info,
+    }
+    if live_activity_token:
+        payload[ATTR_LIVE_ACTIVITY_TOKEN] = live_activity_token
+
     try:
         async with asyncio.timeout(10):
             response = await session.post(
                 entry.data[ATTR_APP_DATA][ATTR_PUSH_URL],
-                json={
-                    **data,
-                    ATTR_PUSH_TOKEN: entry.data[ATTR_APP_DATA][ATTR_PUSH_TOKEN],
-                    "registration_info": reg_info,
-                },
+                json=payload,
             )
             result: dict[str, Any] = await response.json()