feat(immich): per-album scheduled/memory dispatch + template tooling

Dispatch: honor {kind}_collection_mode on TrackingConfig — "per_collection"
fans out one event per album; "combined" pools assets as before. Extract
build_immich_dispatch_events shared by cron and test paths.

Assets: collect_scheduled_assets attaches album_name/album_url/album_public_url
to each asset so combined-mode templates can attribute rows to their source
album. Default scheduled_assets templates render a multi-album header with
inline album list and per-row album link; memory_mode follows the same pattern.

UI: "Reset to default" buttons on notification and command template slots
(per-slot and whole-template), backed by new GET /*-template-configs/defaults
endpoints. tracking-configs "Preview template" now opens an inline preview
modal with locale tabs instead of navigating away; Edit button deep-links
with ?edit_slot=<name> so the destination auto-opens the config and scrolls
to the slot. Reset confirmations use ConfirmModal instead of window.confirm.

Fixes:
* NotificationDispatcher._session_ctx infinite recursion when no shared
  aiohttp.ClientSession was passed — broke test dispatch for periodic/
  scheduled/memory (cron path was unaffected).
* telegram-bots /chats/{id}/test now resolves chat.language_override /
  language_code instead of using the raw ?locale query param, matching
  the resolution the tracker-target test endpoint already used.
* scheduled_assets default template no longer emits a blank line between
  header and the first asset when the multi-album branch is taken.

packages/server/src/notify_bridge_server/services/scheduled_dispatch.py

@@ -46,12 +46,62 @@ from .dispatch_helpers import (
     get_app_timezone,
     load_link_data,
 )
-from .manual_dispatch import _build_immich_event, _build_immich_periodic_event
+from .manual_dispatch import build_immich_dispatch_events
 
 _LOGGER = logging.getLogger(__name__)
 
 ScheduledKind = Literal["scheduled", "periodic", "memory"]
 
+# Reasons a scheduled cron fire can end up producing no notification. We write
+# these to EventLog.details.skip_reason so users can see *why* a 09:00 memory
+# didn't arrive, rather than silently treating the fire as if it never happened.
+_SKIP_REASON_TRACKER_DISABLED = "tracker_disabled"
+_SKIP_REASON_NOT_IMMICH = "not_immich_provider"
+_SKIP_REASON_KIND_DISABLED = "kind_disabled_on_default_config"
+_SKIP_REASON_NO_LINKS = "no_enabled_links"
+_SKIP_REASON_NO_EVENT = "provider_returned_no_event"
+_SKIP_REASON_EMPTY_PAYLOAD = "zero_assets_matched"
+_SKIP_REASON_NO_TARGETS = "no_targets_after_filtering"
+
+
+async def _log_skip(
+    tracker_id: int,
+    kind: ScheduledKind,
+    reason: str,
+    *,
+    tracker_user_id: int | None = None,
+    tracker_name: str = "",
+    provider_id: int | None = None,
+    provider_name: str = "",
+) -> None:
+    """Persist an EventLog row for a skipped scheduled fire.
+
+    Separate from the success-path log (which records targets dispatched) so
+    operators and users can filter "why didn't this fire" from "what was sent".
+    ``event_type`` mirrors the success path's value; the skip is disambiguated
+    by ``details.status == "skipped"``.
+    """
+    engine = get_engine()
+    async with AsyncSession(engine) as session:
+        session.add(EventLog(
+            user_id=tracker_user_id,
+            tracker_id=tracker_id,
+            tracker_name=tracker_name,
+            provider_id=provider_id,
+            provider_name=provider_name,
+            event_type=EventType.SCHEDULED_MESSAGE.value,
+            collection_id="",
+            collection_name="",
+            assets_count=0,
+            details={
+                "kind": kind,
+                "trigger": "cron",
+                "status": "skipped",
+                "skip_reason": reason,
+            },
+        ))
+        await session.commit()
+
 # Maps the dispatch kind to the DB slot name that holds its template.
 # The dispatcher keys templates by ``event.event_type.value`` (always
 # ``scheduled_message`` here), so we read the right ``TemplateSlot`` row and
@@ -76,9 +126,23 @@ async def dispatch_scheduled_for_tracker(
     async with AsyncSession(engine) as session:
         tracker = await session.get(NotificationTracker, tracker_id)
         if not tracker or not tracker.enabled:
+            # No user context available (tracker missing/disabled); still log so
+            # operators can correlate cron fires that went nowhere.
+            await _log_skip(
+                tracker_id, kind, _SKIP_REASON_TRACKER_DISABLED,
+                tracker_user_id=(tracker.user_id if tracker else None),
+                tracker_name=(tracker.name if tracker else ""),
+            )
             return
         provider = await session.get(ServiceProvider, tracker.provider_id)
         if not provider or provider.type != "immich":
+            await _log_skip(
+                tracker_id, kind, _SKIP_REASON_NOT_IMMICH,
+                tracker_user_id=tracker.user_id,
+                tracker_name=tracker.name or "",
+                provider_id=(provider.id if provider else None),
+                provider_name=(provider.name if provider else ""),
+            )
             return
 
         default_tc: TrackingConfig | None = None
@@ -94,6 +158,13 @@ async def dispatch_scheduled_for_tracker(
                 "Scheduled %s skipped for tracker %d: kind disabled on default config",
                 kind, tracker_id,
             )
+            await _log_skip(
+                tracker_id, kind, _SKIP_REASON_KIND_DISABLED,
+                tracker_user_id=tracker.user_id,
+                tracker_name=tracker.name or "",
+                provider_id=provider.id,
+                provider_name=provider.name or provider.type,
+            )
             return
 
         # Snapshot every field we need outside the session — after the
@@ -115,90 +186,54 @@ async def dispatch_scheduled_for_tracker(
             "Scheduled %s for tracker %d: no enabled links, skipping",
             kind, tracker_id,
         )
-        return
-
-    if kind == "periodic":
-        event = await _build_immich_periodic_event(
-            provider_config=provider_config,
-            provider_name=provider_name,
+        await _log_skip(
+            tracker_id, kind, _SKIP_REASON_NO_LINKS,
+            tracker_user_id=tracker_user_id,
             tracker_name=tracker_name,
-            collection_ids=collection_ids,
-        )
-    else:
-        event = await _build_immich_event(
-            provider_config=provider_config,
+            provider_id=provider_id,
             provider_name=provider_name,
-            tracker_name=tracker_name,
-            collection_ids=collection_ids,
-            test_type=kind,
-            tracking_config=default_tc,
-        )
-    if event is None:
-        _LOGGER.warning(
-            "Scheduled %s for tracker %d: provider returned no event",
-            kind, tracker_id,
         )
         return
 
-    # Skip empty payloads for asset-bearing kinds — sending the bare
-    # "On this day:" / "Scheduled delivery —" header with no items below
-    # spams chats with title-only messages every day. ``periodic`` is
-    # different: it's a stats summary that's still meaningful with zero
-    # assets, so we let it through.
-    if kind in ("scheduled", "memory") and not event.added_assets:
+    # Resolve mode + build events via the shared helper (same decision logic
+    # the test-dispatch path uses). "per_collection" fans out one event per
+    # album; "combined" pools assets into a single event. ``collection_mode``
+    # is threaded through to EventLog.details so operators can see *which*
+    # mode a fire used when auditing behaviour.
+    collection_mode = (
+        "combined" if kind == "periodic"
+        else getattr(default_tc, f"{kind}_collection_mode", "combined") or "combined"
+    )
+    events = await build_immich_dispatch_events(
+        provider_config=provider_config,
+        provider_name=provider_name,
+        tracker_name=tracker_name,
+        collection_ids=collection_ids,
+        kind=kind,
+        tracking_config=default_tc,
+    )
+
+    if not events:
+        # All albums yielded 0 matching assets (per_collection), or the single
+        # combined build produced nothing. Log the same skip reason used for
+        # the legacy single-event path so operators see a consistent signal.
+        reason = (
+            _SKIP_REASON_NO_EVENT if kind == "periodic" else _SKIP_REASON_EMPTY_PAYLOAD
+        )
         _LOGGER.info(
-            "Scheduled %s for tracker %d: 0 assets matched, skipping dispatch",
-            kind, tracker_id,
+            "Scheduled %s for tracker %d: no events to dispatch (mode=%s)",
+            kind, tracker_id, collection_mode,
+        )
+        await _log_skip(
+            tracker_id, kind, reason,
+            tracker_user_id=tracker_user_id,
+            tracker_name=tracker_name,
+            provider_id=provider_id,
+            provider_name=provider_name,
         )
         return
 
     slot_name = _SLOT_MAP[kind]
-    target_configs: list[TargetConfig] = []
-    async with AsyncSession(engine) as session:
-        for ld in link_data:
-            tc = ld["tracking_config"] or default_tc
-            tmpl = ld["template_config"]
-            if tc is not None:
-                # Per-link override may disable this kind even when the
-                # default has it on — honour that here.
-                if not getattr(tc, f"{kind}_enabled", True):
-                    continue
-                if not event_allowed_by_config(event, tc, app_tz):
-                    continue
-            if tmpl is None:
-                continue
-
-            slot_rows = (await session.exec(
-                select(TemplateSlot).where(
-                    TemplateSlot.config_id == tmpl.id,
-                    TemplateSlot.slot_name == slot_name,
-                )
-            )).all()
-            if not slot_rows:
-                continue
-            locale_map = {s.locale: s.template for s in slot_rows}
-            template_slots = {EventType.SCHEDULED_MESSAGE.value: locale_map}
-
-            target_configs.append(TargetConfig(
-                type=ld["target_type"],
-                config=ld["target_config"],
-                template_slots=template_slots,
-                date_format=tmpl.date_format,
-                date_only_format=(
-                    tmpl.date_only_format or "%d.%m.%Y"
-                ),
-                provider_api_key=provider_config.get("api_key"),
-                provider_internal_url=provider_config.get("url", ""),
-                provider_external_url=provider_config.get("external_domain", ""),
-                receivers=ld["receivers"],
-            ))
-
-    if not target_configs:
-        _LOGGER.info(
-            "Scheduled %s for tracker %d: no targets after filtering",
-            kind, tracker_id,
-        )
-        return
 
     # Lazy import to break the watcher↔scheduler↔scheduled_dispatch cycle.
     from .watcher import _get_telegram_caches
@@ -209,34 +244,96 @@ async def dispatch_scheduled_for_tracker(
     dispatcher = NotificationDispatcher(
         url_cache=url_cache, asset_cache=asset_cache, session=http_session,
     )
-    _LOGGER.info(
-        "Dispatching scheduled %s for tracker %d to %d link(s)",
-        kind, tracker_id, len(target_configs),
-    )
-    results = await dispatcher.dispatch(event, target_configs)
 
-    # Mirror the watcher's audit trail: surface scheduled fires in EventLog so
-    # the dashboard shows *why* a notification arrived (otherwise these would
-    # be invisible to the activity feed).
-    successes = sum(1 for r in results if isinstance(r, dict) and r.get("success"))
-    async with AsyncSession(engine) as session:
-        session.add(EventLog(
-            user_id=tracker_user_id,
-            tracker_id=tracker_id,
+    any_sent = False
+    for event in events:
+        # Target config assembly depends on the event for quiet-hours /
+        # event_allowed_by_config, which inspects event timestamp. Per-event
+        # rebuilding also lets a per-link override disable one kind while
+        # keeping others live.
+        target_configs: list[TargetConfig] = []
+        async with AsyncSession(engine) as session:
+            for ld in link_data:
+                tc = ld["tracking_config"] or default_tc
+                tmpl = ld["template_config"]
+                if tc is not None:
+                    if not getattr(tc, f"{kind}_enabled", True):
+                        continue
+                    if not event_allowed_by_config(event, tc, app_tz):
+                        continue
+                if tmpl is None:
+                    continue
+
+                slot_rows = (await session.exec(
+                    select(TemplateSlot).where(
+                        TemplateSlot.config_id == tmpl.id,
+                        TemplateSlot.slot_name == slot_name,
+                    )
+                )).all()
+                if not slot_rows:
+                    continue
+                locale_map = {s.locale: s.template for s in slot_rows}
+                template_slots = {EventType.SCHEDULED_MESSAGE.value: locale_map}
+
+                target_configs.append(TargetConfig(
+                    type=ld["target_type"],
+                    config=ld["target_config"],
+                    template_slots=template_slots,
+                    date_format=tmpl.date_format,
+                    date_only_format=(
+                        tmpl.date_only_format or "%d.%m.%Y"
+                    ),
+                    provider_api_key=provider_config.get("api_key"),
+                    provider_internal_url=provider_config.get("url", ""),
+                    provider_external_url=provider_config.get("external_domain", ""),
+                    receivers=ld["receivers"],
+                ))
+
+        if not target_configs:
+            _LOGGER.info(
+                "Scheduled %s for tracker %d (collection=%r): no targets after filtering",
+                kind, tracker_id, event.collection_name,
+            )
+            continue
+
+        _LOGGER.info(
+            "Dispatching scheduled %s for tracker %d (collection=%r) to %d link(s)",
+            kind, tracker_id, event.collection_name, len(target_configs),
+        )
+        results = await dispatcher.dispatch(event, target_configs)
+        any_sent = True
+
+        successes = sum(1 for r in results if isinstance(r, dict) and r.get("success"))
+        async with AsyncSession(engine) as session:
+            session.add(EventLog(
+                user_id=tracker_user_id,
+                tracker_id=tracker_id,
+                tracker_name=tracker_name,
+                provider_id=provider_id,
+                provider_name=provider_name,
+                event_type=event.event_type.value,
+                collection_id=event.collection_id,
+                collection_name=event.collection_name,
+                assets_count=event.added_count or 0,
+                details={
+                    "kind": kind,
+                    "slot": slot_name,
+                    "trigger": "cron",
+                    "timezone": app_tz,
+                    "collection_mode": collection_mode,
+                    "status": "sent",
+                    "targets_dispatched": len(target_configs),
+                    "targets_succeeded": successes,
+                },
+            ))
+            await session.commit()
+
+    if not any_sent:
+        # All events produced zero targets after filtering (quiet hours, etc.).
+        await _log_skip(
+            tracker_id, kind, _SKIP_REASON_NO_TARGETS,
+            tracker_user_id=tracker_user_id,
             tracker_name=tracker_name,
             provider_id=provider_id,
             provider_name=provider_name,
-            event_type=event.event_type.value,
-            collection_id=event.collection_id,
-            collection_name=event.collection_name,
-            assets_count=event.added_count or 0,
-            details={
-                "kind": kind,
-                "slot": slot_name,
-                "trigger": "cron",
-                "timezone": app_tz,
-                "targets_dispatched": len(target_configs),
-                "targets_succeeded": successes,
-            },
-        ))
-        await session.commit()
+        )