saksham
1a4c80958f
FastAPI + Celery + Next.js + Postgres/Redis app with company monitoring, source collection, LLM-based change analysis, enrichment, and account security (Turnstile, escalating lockout, email verification).
203 lines
7.4 KiB
Python
203 lines
7.4 KiB
Python
"""Turns a DetectedChange that crosses the company's alert threshold into an
|
|
Alert, generates its summary via Task F (LLM), and dispatches it to every
|
|
enabled notification destination that also meets its own severity threshold
|
|
- recording one NotificationDelivery per attempt. Two independent
|
|
thresholds by design: MonitorConfiguration.severity_threshold gates whether
|
|
an Alert is created at all; NotificationDestination.minimum_severity then
|
|
gates whether *this* destination gets notified about it (spec section 6I/22).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from datetime import UTC, datetime
|
|
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.analysis.llm.base import LLMProvider
|
|
from app.core.config import Settings
|
|
from app.core.errors import NotFoundError
|
|
from app.models.alert import Alert
|
|
from app.models.company import Company
|
|
from app.models.detected_change import DetectedChange
|
|
from app.models.enums import (
|
|
SEVERITY_ORDER,
|
|
NotificationDeliveryStatus,
|
|
NotificationType,
|
|
SeverityLevel,
|
|
)
|
|
from app.models.notification_delivery import NotificationDelivery
|
|
from app.notifications.base import DeliveryResult, NotificationMessage
|
|
from app.notifications.factory import get_notification_provider
|
|
from app.notifications.message_builder import build_alert_message
|
|
from app.prompts.alert_summarization import summarize_alert
|
|
from app.repositories.alert_repository import AlertRepository
|
|
from app.repositories.notification_delivery_repository import NotificationDeliveryRepository
|
|
from app.repositories.notification_destination_repository import NotificationDestinationRepository
|
|
|
|
|
|
def _meets_threshold(severity: SeverityLevel, threshold: SeverityLevel) -> bool:
|
|
"""SEVERITY_ORDER[0] is most severe, so meeting a threshold means being
|
|
at least as severe - i.e. an index <= the threshold's index."""
|
|
return SEVERITY_ORDER.index(severity) <= SEVERITY_ORDER.index(threshold)
|
|
|
|
|
|
async def create_alert_for_change(
|
|
db: AsyncSession, settings: Settings, llm: LLMProvider, change: DetectedChange, company: Company
|
|
) -> Alert | None:
|
|
config = company.monitor_configuration
|
|
if config is None or not _meets_threshold(change.severity, config.severity_threshold):
|
|
return None
|
|
|
|
evidence_snippets = [
|
|
*change.raw_diff.get("text_added_lines", [])[:5],
|
|
*change.raw_diff.get("structured_added", [])[:5],
|
|
]
|
|
summary = await summarize_alert(
|
|
llm,
|
|
company_name=company.name,
|
|
change_type=change.change_type.value,
|
|
change_summary=change.summary,
|
|
severity=change.severity.value,
|
|
confidence=change.confidence_score,
|
|
evidence_snippets=evidence_snippets,
|
|
)
|
|
|
|
alert = Alert(
|
|
company_id=company.id,
|
|
detected_change_id=change.id,
|
|
user_id=company.user_id,
|
|
title=summary.title,
|
|
summary=summary.summary,
|
|
why_it_matters=summary.why_it_matters,
|
|
severity=change.severity,
|
|
confidence=change.confidence_score,
|
|
)
|
|
await AlertRepository(db).create(alert)
|
|
|
|
destinations = await NotificationDestinationRepository(db).list_for_company(company.id)
|
|
delivery_repo = NotificationDeliveryRepository(db)
|
|
for destination in destinations:
|
|
if not destination.enabled:
|
|
continue
|
|
if not _meets_threshold(change.severity, destination.minimum_severity):
|
|
continue
|
|
if destination.type == NotificationType.SMS and not settings.notification_sms_enabled:
|
|
continue
|
|
|
|
provider = get_notification_provider(destination.type, settings)
|
|
message = build_alert_message(
|
|
destination.type, destination.destination_value, company, alert, settings
|
|
)
|
|
delivery = NotificationDelivery(
|
|
alert_id=alert.id, destination_id=destination.id, provider=provider.provider_name
|
|
)
|
|
await delivery_repo.create(delivery)
|
|
|
|
result = await provider.send(message)
|
|
delivery.attempt_count = 1
|
|
delivery.last_attempt = datetime.now(UTC)
|
|
delivery.status = (
|
|
NotificationDeliveryStatus.SENT if result.success else NotificationDeliveryStatus.FAILED
|
|
)
|
|
delivery.external_message_id = result.external_message_id
|
|
delivery.error_message = result.error
|
|
|
|
await db.commit()
|
|
await db.refresh(alert)
|
|
return alert
|
|
|
|
|
|
async def list_alerts(
|
|
db: AsyncSession,
|
|
user_id: uuid.UUID,
|
|
*,
|
|
company_id: uuid.UUID | None = None,
|
|
severity: SeverityLevel | None = None,
|
|
read: bool | None = None,
|
|
resolved: bool | None = None,
|
|
) -> list[Alert]:
|
|
return await AlertRepository(db).list_for_user(
|
|
user_id, company_id=company_id, severity=severity, read=read, resolved=resolved
|
|
)
|
|
|
|
|
|
async def get_alert(db: AsyncSession, user_id: uuid.UUID, alert_id: uuid.UUID) -> Alert:
|
|
alert = await AlertRepository(db).get_for_user(alert_id, user_id)
|
|
if alert is None:
|
|
raise NotFoundError("Alert not found")
|
|
return alert
|
|
|
|
|
|
async def mark_read(db: AsyncSession, user_id: uuid.UUID, alert_id: uuid.UUID) -> Alert:
|
|
alert = await get_alert(db, user_id, alert_id)
|
|
alert.read = True
|
|
await db.commit()
|
|
await db.refresh(alert)
|
|
return alert
|
|
|
|
|
|
async def mark_resolved(db: AsyncSession, user_id: uuid.UUID, alert_id: uuid.UUID) -> Alert:
|
|
alert = await get_alert(db, user_id, alert_id)
|
|
alert.resolved = True
|
|
await db.commit()
|
|
await db.refresh(alert)
|
|
return alert
|
|
|
|
|
|
async def update_alert(
|
|
db: AsyncSession,
|
|
user_id: uuid.UUID,
|
|
alert_id: uuid.UUID,
|
|
*,
|
|
read: bool | None,
|
|
resolved: bool | None,
|
|
) -> Alert:
|
|
alert = await get_alert(db, user_id, alert_id)
|
|
if read is not None:
|
|
alert.read = read
|
|
if resolved is not None:
|
|
alert.resolved = resolved
|
|
await db.commit()
|
|
await db.refresh(alert)
|
|
return alert
|
|
|
|
|
|
async def get_alert_with_deliveries(
|
|
db: AsyncSession, user_id: uuid.UUID, alert_id: uuid.UUID
|
|
) -> tuple[Alert, list[NotificationDelivery]]:
|
|
alert = await get_alert(db, user_id, alert_id)
|
|
deliveries = await NotificationDeliveryRepository(db).list_for_alert(alert.id)
|
|
return alert, deliveries
|
|
|
|
|
|
async def send_test_notification(
|
|
db: AsyncSession, settings: Settings, user_id: uuid.UUID, destination_id: uuid.UUID
|
|
):
|
|
destination = await NotificationDestinationRepository(db).get_for_user(destination_id, user_id)
|
|
if destination is None:
|
|
raise NotFoundError("Notification destination not found")
|
|
|
|
if destination.type == NotificationType.SMS and not settings.notification_sms_enabled:
|
|
return DeliveryResult(
|
|
success=False,
|
|
error="SMS delivery is currently disabled (NOTIFICATION_SMS_ENABLED=false) - no "
|
|
"message was sent.",
|
|
)
|
|
|
|
provider = get_notification_provider(destination.type, settings)
|
|
if destination.type == NotificationType.SMS:
|
|
message = NotificationMessage(
|
|
destination_value=destination.destination_value,
|
|
subject="CI Agent test",
|
|
body_text="CI Agent test SMS: your notification destination is configured correctly.",
|
|
)
|
|
else:
|
|
message = NotificationMessage(
|
|
destination_value=destination.destination_value,
|
|
subject="CI Agent test notification",
|
|
body_text="This is a test notification from CI Agent. Your destination is configured correctly.",
|
|
body_html="<p>This is a test notification from <strong>CI Agent</strong>. Your destination is configured correctly.</p>",
|
|
)
|
|
return await provider.send(message)
|