"""Compliance incident CRUD + creation helpers. Automated sweeps use :func:`create_or_upsert_incident` to register a violation of the platform's data-freshness guarantee. The helper is idempotent on ``(incident_type, resource_type, resource_id)``: re- scanning the same problem doesn't create duplicate tickets, it refreshes ``detected_at`` on the existing open row so the admin can see "this has persisted for N hours". """ from __future__ import annotations from datetime import UTC, datetime from typing import Any from uuid import UUID from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from app.audit.service import AuditService from app.compliance.models import ( INCIDENT_ACCEPTED, INCIDENT_OPEN, INCIDENT_RESOLVED, SEVERITY_CRITICAL, ComplianceIncident, ) from app.core.exceptions import AppException, NotFoundError class ComplianceIncidentService: def __init__(self, db: AsyncSession) -> None: self.db = db self.audit = AuditService(db) async def create_or_upsert_incident( self, *, incident_type: str, title: str, severity: str = SEVERITY_CRITICAL, person_id: UUID | None = None, resource_type: str | None = None, resource_id: UUID | str | None = None, resource_label: str | None = None, detail: dict[str, Any] | None = None, ) -> tuple[ComplianceIncident, bool]: """Return ``(incident, created)``. If a non-terminal incident (``open`` OR ``accepted``) with the same ``(incident_type, resource_type, resource_id)`` tuple already exists, we return it and bump ``detected_at``. Otherwise a new row is created. ``accepted`` rows are matched on purpose: ``accepted`` means "known & expected exception" for a violation that will *keep* showing up on every sweep. If we only matched ``open`` rows, an accepted incident whose underlying resource still exists would spawn a fresh ``open`` ticket each cycle, defeating the acknowledgement. So an accepted match is refreshed in place and stays accepted (never silently re-opened). A ``resolved`` row is terminal and is NOT matched — a still-present violation re-opens correctly, and a fixed one simply stops appearing on sweeps. """ stmt = select(ComplianceIncident).where( ComplianceIncident.incident_type == incident_type, ComplianceIncident.status.in_([INCIDENT_OPEN, INCIDENT_ACCEPTED]), ) if resource_type is not None: stmt = stmt.where(ComplianceIncident.resource_type == resource_type) if resource_id is not None: stmt = stmt.where(ComplianceIncident.resource_id == resource_id) # Prefer an already-open row over an accepted one if legacy data # left both for the same key; otherwise take the most recent. stmt = stmt.order_by( (ComplianceIncident.status == INCIDENT_ACCEPTED), ComplianceIncident.detected_at.desc(), ).limit(1) existing = (await self.db.execute(stmt)).scalars().first() if existing is not None: existing.detected_at = datetime.now(UTC) # refresh detail so admin sees the latest context if detail is not None: merged = dict(existing.detail or {}) merged.update(detail) existing.detail = merged # Don't rewrite the title of an acknowledged exception; only # keep an open ticket's title fresh. if ( title and title != existing.title and existing.status == INCIDENT_OPEN ): existing.title = title await self.db.flush() return existing, False row = ComplianceIncident( incident_type=incident_type, severity=severity, status=INCIDENT_OPEN, person_id=person_id, resource_type=resource_type, resource_id=resource_id, resource_label=resource_label, title=title, detail=detail, detected_at=datetime.now(UTC), ) self.db.add(row) await self.db.flush() return row, True async def list_incidents( self, *, status: str | None = INCIDENT_OPEN, severity: str | None = None, q: str | None = None, page: int = 1, page_size: int = 50, ) -> tuple[list[ComplianceIncident], int]: stmt = select(ComplianceIncident) if status is not None: stmt = stmt.where(ComplianceIncident.status == status) if severity is not None: stmt = stmt.where(ComplianceIncident.severity == severity) if q and q.strip(): from sqlalchemy import or_ # Escape SQL LIKE wildcards in the user input so a literal # ``%`` / ``_`` in the query doesn't silently match every # row. ``escape='\\'`` only declares the escape character # — the substring itself must be pre-escaped. escaped = ( q.strip() .replace("\\", "\\\\") .replace("%", "\\%") .replace("_", "\\_") ) like = f"%{escaped}%" stmt = stmt.where(or_( ComplianceIncident.title.ilike(like, escape="\\"), ComplianceIncident.incident_type.ilike(like, escape="\\"), ComplianceIncident.resource_label.ilike(like, escape="\\"), )) count_stmt = select(func.count()).select_from(stmt.subquery()) total = (await self.db.execute(count_stmt)).scalar() or 0 stmt = ( stmt .order_by(ComplianceIncident.detected_at.desc()) .offset((page - 1) * page_size) .limit(page_size) ) rows = list((await self.db.execute(stmt)).scalars().all()) return rows, total async def get_incident(self, incident_id: UUID) -> ComplianceIncident: row = await self.db.get(ComplianceIncident, incident_id) if row is None: raise NotFoundError("ComplianceIncident", str(incident_id)) return row async def resolve_incident( self, incident_id: UUID, *, resolved_by_person_id: UUID | None, resolution_note: str | None, accepted: bool = False, trace_id: str = "", ) -> ComplianceIncident: """Close an incident. ``accepted=False`` means "fixed — the underlying resource is now properly owned/removed". ``accepted=True`` means "known and acknowledged, intentional exception" — the incident is closed but the distinction is kept in status for auditing. """ row = await self.get_incident(incident_id) # Idempotency guard: an incident is closed exactly once. Without # this, a double-submit (e.g. a retried resolve after a transient # client error) silently overwrites resolved_at / resolved_by / # note and can flip accepted↔resolved on an already-closed row. # Reject the second write so the close stays attributable. if row.status != INCIDENT_OPEN: raise AppException( code="INCIDENT_ALREADY_CLOSED", message=f"Incident is already {row.status!r}; cannot re-resolve.", status_code=409, ) row.status = INCIDENT_ACCEPTED if accepted else INCIDENT_RESOLVED row.resolved_at = datetime.now(UTC) row.resolved_by_person_id = resolved_by_person_id row.resolution_note = resolution_note await self.db.flush() await self.audit.log( trace_id=trace_id, actor_type="user" if resolved_by_person_id else "system", actor_user_id=resolved_by_person_id, action="compliance_incident.resolve", resource_type="compliance_incident", resource_id=str(row.id), status="success", detail={ "incident_type": row.incident_type, "severity": row.severity, "accepted": accepted, "resolution_note": resolution_note, }, ) return row async def count_open_by_severity(self) -> dict[str, int]: """Return ``{severity: count}`` for open incidents.""" rows = ( await self.db.execute( select( ComplianceIncident.severity, func.count(ComplianceIncident.id), ) .where(ComplianceIncident.status == INCIDENT_OPEN) .group_by(ComplianceIncident.severity) ) ).all() return {sev: int(cnt) for sev, cnt in rows} __all__ = ["ComplianceIncidentService"]