Files
stackpilot/backend/services/notify_service.py
menzeljandClaude Opus 4.8 8d19b09abd Phase 4: backups w/ volumes, notifications, settings & users, audit page (0.4.0)
- Backup/restore: per-stack tar.gz incl. named-volume snapshots (helper
  container), upload restore with rename/overwrite/conflict detection.
- Notifications: ntfy/Discord/Slack/Gotify/generic webhooks, per-event
  subscriptions; wired into the update checker and stack lifecycle.
- Settings page: update-check interval, webhook CRUD + test, user management
  (with last-admin safeguards).
- Audit log page (searchable, paginated).
- Mobile-responsive sidebar/layout.

Multi-host agents and remote backup destinations (SFTP/S3) deferred.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-07 20:58:05 +00:00

134 lines
4.4 KiB
Python

"""Outbound notification webhooks.
Webhooks are configured two ways:
* DB-managed (the ``Webhook`` table) — per-webhook type + event subscriptions,
editable from the Settings page.
* Env ``NOTIFY_WEBHOOKS`` — a comma-separated list of generic JSON endpoints
that receive every event (kept for backward compatibility / GitOps setups).
Supported types: ntfy, discord, slack, gotify, generic (JSON POST).
All delivery is best-effort: failures are logged, never raised to the caller.
"""
from __future__ import annotations
import logging
from typing import Optional
import httpx
from sqlmodel import Session, select
from config import settings as env_settings
from database import engine
from models.setting import Webhook
logger = logging.getLogger("stackpilot.notify")
_TIMEOUT = 10.0
# --------------------------------------------------------------------------- #
# Payload formatting per webhook type
# --------------------------------------------------------------------------- #
def _build_request(wtype: str, url: str, event: str, title: str, message: str):
"""Return (method-kwargs) for httpx.post for the given webhook type."""
if wtype == "ntfy":
return {
"url": url,
"content": message.encode("utf-8"),
"headers": {"Title": title, "Tags": _ntfy_tag(event)},
}
if wtype == "discord":
return {"url": url, "json": {"content": f"**{title}**\n{message}"}}
if wtype == "slack":
return {"url": url, "json": {"text": f"*{title}*\n{message}"}}
if wtype == "gotify":
priority = 8 if event in ("stack_error", "pull_failed") else 5
return {
"url": url,
"json": {"title": title, "message": message, "priority": priority},
}
# generic
return {
"url": url,
"json": {"event": event, "title": title, "message": message},
}
def _ntfy_tag(event: str) -> str:
return {
"update_available": "arrow_up",
"stack_start": "white_check_mark",
"stack_stop": "stop_button",
"stack_error": "rotating_light",
"pull_failed": "warning",
}.get(event, "bell")
async def _deliver(client: httpx.AsyncClient, wtype: str, url: str, event: str, title: str, message: str) -> bool:
kwargs = _build_request(wtype, url, event, title, message)
target = kwargs.pop("url")
try:
resp = await client.post(target, timeout=_TIMEOUT, **kwargs)
resp.raise_for_status()
return True
except httpx.HTTPError as exc:
logger.warning("Webhook delivery failed (%s): %s", wtype, exc)
return False
# --------------------------------------------------------------------------- #
# Public API
# --------------------------------------------------------------------------- #
def _targets_for_event(session: Session, event: str) -> list[tuple[str, str]]:
"""Return [(type, url)] of all destinations subscribed to ``event``."""
targets: list[tuple[str, str]] = []
for wh in session.exec(select(Webhook)).all():
if not wh.enabled:
continue
subscribed = [e.strip() for e in (wh.events or "").split(",") if e.strip()]
if event in subscribed:
targets.append((wh.type, wh.url))
# Env-configured generic endpoints receive everything.
for url in env_settings.NOTIFY_WEBHOOKS:
targets.append(("generic", url))
return targets
async def notify(
event: str,
title: str,
message: str,
session: Optional[Session] = None,
) -> int:
"""Fan out ``event`` to all subscribed webhooks. Returns delivered count."""
if session is None:
with Session(engine) as own:
return await notify(event, title, message, own)
targets = _targets_for_event(session, event)
if not targets:
return 0
delivered = 0
async with httpx.AsyncClient(follow_redirects=True) as client:
for wtype, url in targets:
if await _deliver(client, wtype, url, event, title, message):
delivered += 1
return delivered
async def test_webhook(wtype: str, url: str) -> bool:
"""Send a one-off test notification to a single destination."""
async with httpx.AsyncClient(follow_redirects=True) as client:
return await _deliver(
client,
wtype,
url,
"update_available",
"StackPilot test notification",
"If you can read this, your webhook is configured correctly. 🚀",
)