| 1 | """ |
| 2 | Service helpers for the case timeline / audit log (issue #792, Phase 4). |
| 3 | |
| 4 | The ``CaseEvent`` table is append-only. Every meaningful case mutation |
| 5 | (status change, alert link, assignment, escalation, comment, template |
| 6 | application, task add/status change) writes one row here. The timeline |
| 7 | view (``GET /case/{id}/timeline``) reads from this table. |
| 8 | |
| 9 | Design notes: |
| 10 | - Emits never raise — a failed audit write should not break the |
| 11 | underlying mutation. We log and swallow. |
| 12 | - Service-side emits are used where the same logic is invoked from |
| 13 | multiple call sites (e.g., ``apply_template_to_case`` runs from both |
| 14 | the create-from-alert hook and the manual apply endpoint, so its |
| 15 | emit lives in the service). |
| 16 | - Route-side emits are used where the actor is naturally available at |
| 17 | the route layer (most case mutation routes already resolve |
| 18 | ``current_user``). This keeps the service layer auth-agnostic. |
| 19 | """ |
| 20 | |
| 21 | from datetime import datetime |
| 22 | from typing import Any |
| 23 | from typing import Dict |
| 24 | from typing import List |
| 25 | from typing import Optional |
| 26 | |
| 27 | from loguru import logger |
| 28 | from sqlalchemy import select |
| 29 | from sqlalchemy.ext.asyncio import AsyncSession |
| 30 | |
| 31 | from app.incidents.models import CaseEvent |
| 32 | from app.incidents.schema.case_templates import CaseEventResponse |
| 33 | from app.incidents.schema.case_templates import CaseEventType |
| 34 | from app.incidents.schema.case_templates import CaseTimelineResponse |
| 35 | |
| 36 | |
| 37 | async def emit_case_event( |
| 38 | session: AsyncSession, |
| 39 | case_id: int, |
| 40 | event_type: CaseEventType, |
| 41 | actor: str, |
| 42 | payload: Optional[Dict[str, Any]] = None, |
| 43 | *, |
| 44 | commit: bool = False, |
| 45 | ) -> None: |
| 46 | """ |
| 47 | Append one CaseEvent row. |
| 48 | |
| 49 | Set ``commit=True`` for top-level emits (the mutation has already |
| 50 | persisted). Use ``commit=False`` when emitting from inside an |
| 51 | existing transaction so the audit row lands atomically with the |
| 52 | underlying mutation — the caller commits. |
| 53 | |
| 54 | Failures are logged but never raised: the audit log should not |
| 55 | cause user-facing 500s. |
| 56 | """ |
| 57 | try: |
| 58 | event = CaseEvent( |
| 59 | case_id=case_id, |
| 60 | event_type=event_type.value if isinstance(event_type, CaseEventType) else str(event_type), |
| 61 | actor=actor or "system", |
| 62 | timestamp=datetime.utcnow(), |
| 63 | payload=payload or None, |
| 64 | ) |
| 65 | session.add(event) |
| 66 | if commit: |
| 67 | await session.commit() |
| 68 | except Exception as e: |
| 69 | logger.warning( |
| 70 | f"Failed to emit CaseEvent (case_id={case_id}, type={event_type}, actor={actor}): {e}", |
| 71 | ) |
| 72 | |
| 73 | |
| 74 | async def list_case_events( |
| 75 | case_id: int, |
| 76 | session: AsyncSession, |
| 77 | *, |
| 78 | limit: int = 500, |
| 79 | offset: int = 0, |
| 80 | ) -> CaseTimelineResponse: |
| 81 | """ |
| 82 | Return the case timeline ordered most-recent-first. The ``limit`` |
| 83 | cap protects the client; cases with very long histories can paginate |
| 84 | via ``offset``. |
| 85 | """ |
| 86 | try: |
| 87 | stmt = ( |
| 88 | select(CaseEvent) |
| 89 | .where(CaseEvent.case_id == case_id) |
| 90 | .order_by(CaseEvent.timestamp.desc(), CaseEvent.id.desc()) |
| 91 | .limit(limit) |
| 92 | .offset(offset) |
| 93 | ) |
| 94 | result = await session.execute(stmt) |
| 95 | events = result.scalars().all() |
| 96 | |
| 97 | return CaseTimelineResponse( |
| 98 | case_id=case_id, |
| 99 | events=[ |
| 100 | CaseEventResponse( |
| 101 | id=e.id, |
| 102 | case_id=e.case_id, |
| 103 | event_type=e.event_type, |
| 104 | actor=e.actor, |
| 105 | timestamp=e.timestamp, |
| 106 | payload=e.payload, |
| 107 | ) |
| 108 | for e in events |
| 109 | ], |
| 110 | success=True, |
| 111 | message=f"Retrieved {len(events)} timeline event(s) for case id={case_id}", |
| 112 | ) |
| 113 | except Exception as e: |
| 114 | logger.error(f"Failed to load timeline for case id={case_id}: {e}") |
| 115 | return CaseTimelineResponse( |
| 116 | case_id=case_id, |
| 117 | events=[], |
| 118 | success=False, |
| 119 | message=f"Failed to load case timeline: {e}", |
| 120 | ) |
| 121 | |
| 122 | |
| 123 | # --------------------------------------------------------------------------- |
| 124 | # Convenience constructors for the typed payloads we emit. Keeps the |
| 125 | # event_type/payload shapes consistent across emit sites. |
| 126 | # --------------------------------------------------------------------------- |
| 127 | |
| 128 | |
| 129 | def payload_status_change(from_status: Optional[str], to_status: str, forced: bool = False) -> Dict[str, Any]: |
| 130 | return {"from": from_status, "to": to_status, "forced": forced} |
| 131 | |
| 132 | |
| 133 | def payload_assignment(from_assignee: Optional[str], to_assignee: Optional[str]) -> Dict[str, Any]: |
| 134 | return {"from": from_assignee, "to": to_assignee} |
| 135 | |
| 136 | |
| 137 | def payload_escalation(escalated: bool) -> Dict[str, Any]: |
| 138 | return {"escalated": escalated} |
| 139 | |
| 140 | |
| 141 | def payload_alert_link(alert_id: int, **extra: Any) -> Dict[str, Any]: |
| 142 | """Used for both ``alert_linked`` and ``alert_unlinked``. The unlink path |
| 143 | passes ``tasks_orphaned=<count>`` so the timeline can show how many |
| 144 | tasks survived as case-wide tasks after the alert was detached.""" |
| 145 | out: Dict[str, Any] = {"alert_id": alert_id} |
| 146 | out.update(extra) |
| 147 | return out |
| 148 | |
| 149 | |
| 150 | def payload_alert_links_bulk(alert_ids: List[int]) -> Dict[str, Any]: |
| 151 | return {"alert_ids": list(alert_ids), "count": len(alert_ids)} |
| 152 | |
| 153 | |
| 154 | def payload_comment(comment_id: int, snippet: Optional[str] = None) -> Dict[str, Any]: |
| 155 | """``snippet`` is a short preview (first ~140 chars) for the timeline UI.""" |
| 156 | out: Dict[str, Any] = {"comment_id": comment_id} |
| 157 | if snippet: |
| 158 | out["snippet"] = snippet[:140] |
| 159 | return out |
| 160 | |
| 161 | |
| 162 | def payload_template_applied(template_id: int, template_name: str, tasks_added: int) -> Dict[str, Any]: |
| 163 | return { |
| 164 | "template_id": template_id, |
| 165 | "template_name": template_name, |
| 166 | "tasks_added": tasks_added, |
| 167 | } |
| 168 | |
| 169 | |
| 170 | def payload_task(task_id: int, title: str, mandatory: bool, **extra: Any) -> Dict[str, Any]: |
| 171 | out: Dict[str, Any] = {"task_id": task_id, "title": title, "mandatory": mandatory} |
| 172 | out.update(extra) |
| 173 | return out |