ASA Go: Send FCM push notifications when fire behaviour advisories are issued

.vscode/bruno-collections/Predictive Services API/FCM/Get Notification Settings.bru

@@ -0,0 +1,19 @@
+meta {
+  name: Get Notification Settings
+  type: http
+  seq: 1
+}
+
+get {
+  url: {{API_BASE_URL}}/device/notification-settings?device_id=9242842270074025
+  body: none
+  auth: bearer
+}
+
+params:query {
+  device_id: 9242842270074025
+}
+
+auth:bearer {
+  token: {{AUTH_TOKEN}}
+}

.vscode/bruno-collections/Predictive Services API/FCM/Update Notification Settings.bru

@@ -0,0 +1,22 @@
+meta {
+  name: Update Notification Settings
+  type: http
+  seq: 2
+}
+
+post {
+  url: {{API_BASE_URL}}/device/notification-settings
+  body: json
+  auth: bearer
+}
+
+auth:bearer {
+  token: {{AUTH_TOKEN}}
+}
+
+body:json {
+  {
+    "device_id": "9242842270074025",
+    "fire_zone_source_ids": ["24"]
+  }
+}

.vscode/bruno-collections/Predictive Services API/environments/dev.bru

@@ -0,0 +1,6 @@
+vars {
+  API_BASE_URL: https://wps-pr-5246-e1e498-dev.apps.silver.devops.gov.bc.ca/api
+}
+vars:secret [
+  AUTH_TOKEN
+]

backend/packages/wps-api/pyproject.toml

@@ -49,6 +49,7 @@ dependencies = [
     "gdal==3.12.3",
     "wps-wf1",
     "earthaccess>=0.15.1",
+    "firebase-admin>=7.3.0",
 ]
 
 [project.optional-dependencies]

backend/packages/wps-api/src/app/auto_spatial_advisory/nats_consumer.py

@@ -4,25 +4,28 @@
 
 import asyncio
 import json
-from datetime import datetime
 import logging
+from datetime import datetime
 from typing import List
-from starlette.background import BackgroundTasks
+
 import nats
-from nats.js.api import StreamConfig, RetentionPolicy
 from nats.aio.msg import Msg
+from nats.js.api import RetentionPolicy, StreamConfig
+from starlette.background import BackgroundTasks
+from wps_shared import config
+from wps_shared.utils.time import get_utc_datetime
+from wps_shared.wps_logging import configure_logging
+
 from app.auto_spatial_advisory.nats_config import (
+    hfi_classify_durable_group,
     server,
-    stream_name,
     sfms_file_subject,
+    stream_name,
     subjects,
-    hfi_classify_durable_group,
 )
 from app.auto_spatial_advisory.process_hfi import RunType
 from app.auto_spatial_advisory.process_stats import process_sfms_hfi_stats
 from app.nats_publish import publish
-from wps_shared.wps_logging import configure_logging
-from wps_shared.utils.time import get_utc_datetime
 
 logger = logging.getLogger(__name__)
 
@@ -102,6 +105,13 @@ async def closed_cb():
 
 if __name__ == "__main__":
     configure_logging()
+    creds_json = config.get("FCM_CREDS")
+    if creds_json:
+        from firebase_admin import credentials, initialize_app
+
+        initialize_app(credentials.Certificate(json.loads(creds_json)))
+    else:
+        raise ValueError("FCM_CREDS is not set — Firebase cannot be initialized.")
     loop = asyncio.new_event_loop()
     asyncio.set_event_loop(loop)
     asyncio.run(run())

backend/packages/wps-api/src/app/auto_spatial_advisory/process_stats.py

@@ -1,5 +1,10 @@
+import logging
 from datetime import date, datetime
 
+from wps_shared.db.crud.auto_spatial_advisory import mark_run_parameter_complete
+from wps_shared.db.database import get_async_write_session_scope
+from wps_shared.db.models.auto_spatial_advisory import RunTypeEnum
+
 from app.auto_spatial_advisory.critical_hours import calculate_critical_hours
 from app.auto_spatial_advisory.hfi_minimum_wind_speed import process_hfi_min_wind_speed
 from app.auto_spatial_advisory.hfi_percent_conifer import process_hfi_percent_conifer
@@ -8,8 +13,9 @@
 from app.auto_spatial_advisory.process_hfi import RunType, process_hfi
 from app.auto_spatial_advisory.process_high_hfi_area import process_high_hfi_area
 from app.auto_spatial_advisory.process_zone_status import process_zone_statuses
-from wps_shared.db.crud.auto_spatial_advisory import mark_run_parameter_complete
-from wps_shared.db.database import get_async_write_session_scope
+from app.fcm.notifications import trigger_notifications
+
+logger = logging.getLogger(__name__)
 
 
 async def process_sfms_hfi_stats(run_type: RunType, run_datetime: datetime, for_date: date):
@@ -24,3 +30,16 @@ async def process_sfms_hfi_stats(run_type: RunType, run_datetime: datetime, for_
 
     async with get_async_write_session_scope() as session:
         await mark_run_parameter_complete(session, run_type, run_datetime, for_date)
+
+    try:
+        async with get_async_write_session_scope() as session:
+            await trigger_notifications(
+                session, RunTypeEnum(run_type.value), run_datetime, for_date
+            )
+    except Exception:
+        logger.exception(
+            "Failed to send FCM notifications for run_type=%s run_datetime=%s for_date=%s.",
+            run_type,
+            run_datetime,
+            for_date,
+        )

backend/packages/wps-api/src/app/fcm/notifications.py

@@ -0,0 +1,145 @@
+import asyncio
+import logging
+from datetime import date, datetime, timedelta, timezone
+
+from firebase_admin import exceptions as firebase_exceptions
+from firebase_admin import messaging
+from sqlalchemy.ext.asyncio import AsyncSession
+from wps_shared import config
+from wps_shared.db.crud.auto_spatial_advisory import ZoneAdvisoryStatus, get_zones_with_advisories
+from wps_shared.db.crud.fcm import get_device_tokens_for_zone, update_device_tokens_are_active
+from wps_shared.db.models.auto_spatial_advisory import RunTypeEnum
+from wps_shared.utils.time import get_vancouver_now
+
+logger = logging.getLogger(__name__)
+
+FCM_BATCH_SIZE = 500
+
+
+def build_notification_content(zone_with_advisory: ZoneAdvisoryStatus, for_date: date):
+    return f"{for_date.strftime('%a, %B %-d')} - An advisory has been issued for {zone_with_advisory.placename_label}"
+
+
+def build_notification_title(zone_with_advisory: ZoneAdvisoryStatus):
+    zone = zone_with_advisory.placename_label.split("-")[0]
+    return f"Fire Behaviour Advisory, {zone}"
+
+
+def build_fcm_message(
+    for_date: date, zone_with_advisory: ZoneAdvisoryStatus, device_tokens: list[str]
+):
+    title = build_notification_title(zone_with_advisory)
+    content = build_notification_content(zone_with_advisory, for_date)
+    tag = f"advisory-{zone_with_advisory.source_identifier}"
+    ttl = timedelta(days=2)
+    apns_expiration = str(int((datetime.now(timezone.utc) + ttl).timestamp()))
+    message = messaging.MulticastMessage(
+        notification=messaging.Notification(title=title, body=content),
+        android=messaging.AndroidConfig(
+            ttl=ttl, notification=messaging.AndroidNotification(tag=tag)
+        ),
+        apns=messaging.APNSConfig(
+            headers={"apns-expiration": apns_expiration},
+            payload=messaging.APNSPayload(aps=messaging.Aps(thread_id=tag)),
+        ),
+        tokens=device_tokens,
+    )
+
+    return message
+
+
+async def trigger_notifications(
+    session: AsyncSession, run_type: RunTypeEnum, run_datetime: datetime, for_date: date
+) -> None:
+    if run_type == RunTypeEnum.actual:
+        return
+
+    vancouver_now = get_vancouver_now()
+
+    if for_date != vancouver_now.date():
+        logger.info("Skipping FCM notifications: for_date=%s is not today", for_date)
+        return
+
+    if config.get("ENVIRONMENT") == "production":
+        if vancouver_now.hour >= 12:
+            logger.info(
+                "Skipping FCM notifications: current Vancouver time hour=%d is at or after noon",
+                vancouver_now.hour,
+            )
+            return
+
+    logger.info("Checking for warnings/advisories to send FCM notifications for")
+    zones_with_advisories = await get_zones_with_advisories(
+        session, run_type, run_datetime, for_date
+    )
+    logger.info(
+        f"{len(zones_with_advisories)} have warnings/advisories, checking for devices to notify"
+    )
+    for zone_with_advisory in zones_with_advisories:
+        if not zone_with_advisory.placename_label:
+            logger.error(
+                "Skipping FCM notification: missing placename_label for zone source_identifier=%s",
+                zone_with_advisory.source_identifier,
+            )
+            continue
+        device_tokens = await get_device_tokens_for_zone(
+            session, zone_with_advisory.source_identifier
+        )
+        if len(device_tokens) == 0:
+            logger.info(f"No devices subscribed to {zone_with_advisory.placename_label}")
+            continue
+        logger.info(
+            f"{len(device_tokens)} are subscribed to {zone_with_advisory.placename_label} about to notify"
+        )
+        for i in range(0, len(device_tokens), FCM_BATCH_SIZE):
+            batch = device_tokens[i : i + FCM_BATCH_SIZE]
+            message = build_fcm_message(for_date, zone_with_advisory, batch)
+            try:
+                logger.info(f"Notifiying {len(batch)} devices")
+                # messaging.send_each_for_multicast is a synchronous blocking call
+                response = await asyncio.to_thread(messaging.send_each_for_multicast, message)
+            except firebase_exceptions.FirebaseError:
+                logger.exception(
+                    "FCM send failed for zone=%s date=%s token_count=%d",
+                    zone_with_advisory.placename_label,
+                    for_date,
+                    len(batch),
+                )
+                continue
+            await handle_fcm_response(
+                session, for_date, zone_with_advisory.placename_label, batch, response
+            )
+
+
+async def handle_fcm_response(
+    session: AsyncSession,
+    for_date: date,
+    placename_label: str,
+    device_tokens: list[str],
+    response: messaging.BatchResponse,
+):
+    logger.info(
+        f"Received FCM response with successful notifications sent: {response.success_count}"
+    )
+    # Only deactivate permanently invalid tokens (UnregisteredError — token is no longer registered).
+    # Transient failures (quota, server errors) are not deactivated; the token remains valid.
+    # Deactivated tokens are re-activated when the app re-registers on open.
+    permanently_failed = [
+        device_tokens[idx]
+        for idx, resp in enumerate(response.responses)
+        if not resp.success and isinstance(resp.exception, messaging.UnregisteredError)
+    ]
+    transient_failed_count = response.failure_count - len(permanently_failed)
+
+    if response.failure_count > 0:
+        logger.warning(
+            "FCM send for zone=%s date=%s: %d permanent failures, %d transient failures out of %d tokens",
+            placename_label,
+            for_date,
+            len(permanently_failed),
+            transient_failed_count,
+            len(device_tokens),
+        )
+
+    if permanently_failed:
+        await update_device_tokens_are_active(session, permanently_failed, False)

backend/packages/wps-api/src/app/main.py

@@ -88,7 +88,6 @@
         # We recommend adjusting this value in production.
         profiles_sample_rate=0.5,
     )
-
 # This is the api app.
 api = FastAPI(title="Predictive Services API", description=API_INFO, version="0.0.0")
 

backend/packages/wps-api/src/app/routers/fcm.py

@@ -96,6 +96,7 @@ async def update_notification_settings(
         if not found:
             logger.error("Notification settings update for unknown device_id: %s", request.device_id)
             raise HTTPException(status_code=404, detail=f"Device not found: {request.device_id}")
+        await session.flush()
         fire_zone_source_ids = await get_notification_settings_for_device(
             session, request.device_id
         )