""" SLA Service — breach detection and automatic due-date enforcement. Responsibilities ---------------- 1. check_sla_breaches(app) Called every 30 minutes by APScheduler. Finds all open/in-progress tickets whose due_date has passed and whose SLA breach has not yet been notified. Sends in-app notifications to the assignee (if any) and to every active IT admin, then stamps the ticket so repeat notifications are suppressed until the ticket is updated. 2. set_due_date(ticket) Convenience helper called by the ticket-creation routes so due dates are always derived from the same single source of truth. Notification suppression ------------------------ A dedicated SystemSetting key sla_notified_tickets stores a comma-separated list of ticket IDs that have already received a breach notification. When a ticket is resolved or closed the ID is removed from the list so the suppression does not persist across re-opens (edge case: ticket re-opened after resolution — unlikely but handled). Design notes ------------ - Runs inside the gunicorn worker process (no separate process needed). - Uses app.app_context() so SQLAlchemy sessions are properly scoped. - All DB writes commit independently from the main request cycle. - Errors are logged but never raised — a scheduler failure must not bring down the web process. """ import logging from datetime import datetime, timedelta logger = logging.getLogger(__name__) # ── Priority → SLA hours mapping (fallback if SystemSetting is absent) ──────── _DEFAULT_SLA_HOURS = { 'critical': 4, 'high': 8, 'medium': 48, 'low': 120, } _SETTING_KEYS = { 'critical': 'sla_critical_hours', 'high': 'sla_high_hours', 'medium': 'sla_medium_hours', 'low': 'sla_low_hours', } def _get_sla_hours(priority: str, app) -> int: """Return SLA hours for *priority*, reading from SystemSetting first.""" from app.models import SystemSetting key = _SETTING_KEYS.get(priority, 'sla_medium_hours') config_key = f'SLA_{priority.upper()}_HOURS' default = app.config.get(config_key, _DEFAULT_SLA_HOURS.get(priority, 48)) raw = SystemSetting.get(key) try: return int(raw) if raw is not None else int(default) except (ValueError, TypeError): return int(default) def set_due_date(ticket, app): """Set ticket.due_date from SLA config if not already set. Callers are responsible for committing after calling this function. """ if ticket.due_date: return # already set — respect manual override hours = _get_sla_hours(ticket.priority, app) ticket.due_date = datetime.utcnow() + timedelta(hours=hours) # ── SLA breach notification email template ──────────────────────────────────── _SLA_BREACH_EMAIL = """

⏰ SLA Breach — Action Required

The following ticket has exceeded its SLA response target and requires immediate attention.

Ticket # {{ ticket_number }}
Title {{ title }}
Priority {{ priority }}
Status {{ status }}
Assigned To {{ assigned_to }}
Due Date {{ due_date }} (overdue)
View & Action Ticket
IT Helpdesk System • This is an automated SLA alert.
""" def _priority_color(priority: str) -> str: return { 'low': '#28a745', 'medium': '#ffc107', 'high': '#dc3545', 'critical': '#7f1d1d', }.get(priority, '#6c757d') def check_sla_breaches(app): """Scheduled job: find overdue tickets and notify responsible parties. Safe to call repeatedly — already-notified tickets are suppressed via the sla_notified_tickets SystemSetting key. The suppression list is cleared for a ticket when it transitions to resolved/closed (handled by the update_ticket route clearing it on status change) or when the ticket is re-opened, ensuring fresh notifications if the issue resurfaces. """ with app.app_context(): from app.services.license_service import feature_enabled if not feature_enabled('sla'): logger.debug('[SLA] Skipped — SLA feature not enabled on current license tier.') return try: _run_sla_check(app) except Exception as exc: logger.error(f'[SLA] Unhandled error in check_sla_breaches: {exc}', exc_info=True) def _run_sla_check(app): from app import db from app.models import ( Ticket, TicketStatus, User, UserRole, NotificationType, SystemSetting, ) from app.services.notification_service import create_notification, send_email from flask import render_template_string now = datetime.utcnow() # ── Load suppression list ────────────────────────────────────────────────── raw = SystemSetting.get('sla_notified_tickets', '') already_notified = set(int(x) for x in raw.split(',') if x.strip().isdigit()) # ── Query overdue open tickets ───────────────────────────────────────────── overdue = Ticket.query.filter( Ticket.status.in_([TicketStatus.OPEN, TicketStatus.IN_PROGRESS]), Ticket.due_date.isnot(None), Ticket.due_date < now, ).all() if not overdue: logger.info(f'[SLA] Check complete — no overdue tickets at {now.strftime("%Y-%m-%d %H:%M")}') return base_url = app.config.get('APP_BASE_URL', '') it_dept_email = app.config.get('IT_DEPT_EMAIL', '') newly_notified = [] for ticket in overdue: if ticket.id in already_notified: continue # already sent — skip ticket_url = f'{base_url}/tickets/{ticket.id}' overdue_mins = int((now - ticket.due_date).total_seconds() / 60) overdue_label = ( f'{overdue_mins // 60}h {overdue_mins % 60}m' if overdue_mins >= 60 else f'{overdue_mins}m' ) logger.warning( f'[SLA BREACH] ticket_id={ticket.id} number={ticket.ticket_number} ' f'priority={ticket.priority} overdue_by={overdue_label} ' f'due={ticket.due_date.strftime("%Y-%m-%d %H:%M")} ' f'assigned_to={ticket.assigned_to_id}' ) # Build the email html = render_template_string( _SLA_BREACH_EMAIL, ticket_number = ticket.ticket_number, title = ticket.title, priority = ticket.priority.upper(), priority_color = _priority_color(ticket.priority), status = ticket.status.replace('_', ' ').title(), assigned_to = ticket.assignee.full_name if ticket.assignee else 'Unassigned', due_date = ticket.due_date.strftime('%b %d, %Y %H:%M UTC'), ticket_url = ticket_url, ) subject = ( f'[SLA BREACH] {ticket.ticket_number} — {ticket.priority.upper()} ' f'ticket overdue by {overdue_label}' ) notif_title = f'⏰ SLA Breach: {ticket.ticket_number}' notif_message = ( f'{ticket.priority.upper()} priority ticket "{ticket.title}" ' f'is overdue by {overdue_label}.' ) notified_user_ids = set() # Notify assignee (in-app + email) if ticket.assigned_to_id: create_notification( user_id = ticket.assigned_to_id, notif_type= NotificationType.TICKET_UPDATED, title = notif_title, message = notif_message, ticket_id = ticket.id, link = f'/tickets/{ticket.id}', ) if ticket.assignee and ticket.assignee.email_notif: send_email(subject, [ticket.assignee.email], html) notified_user_ids.add(ticket.assigned_to_id) # Notify all active IT admins (in-app + IT dept email) admins = User.query.filter( User.role == UserRole.ADMIN, User.is_active == True, ).all() for admin in admins: if admin.id not in notified_user_ids: create_notification( user_id = admin.id, notif_type= NotificationType.TICKET_UPDATED, title = notif_title, message = notif_message, ticket_id = ticket.id, link = f'/tickets/{ticket.id}', ) notified_user_ids.add(admin.id) # One email to the IT department inbox if it_dept_email: send_email(subject, [it_dept_email], html) newly_notified.append(ticket.id) db.session.commit() # ── Update suppression list ──────────────────────────────────────────────── if newly_notified: updated = already_notified | set(newly_notified) SystemSetting.set( 'sla_notified_tickets', ','.join(str(i) for i in sorted(updated)), 'Comma-separated ticket IDs that have received SLA breach notifications', ) db.session.commit() logger.info( f'[SLA] Notified {len(newly_notified)} breach(es): ' f'{[str(i) for i in newly_notified]}' ) def clear_sla_notification(ticket_id: int): """Remove a ticket from the SLA suppression list. Call this when a ticket is resolved, closed, or re-opened so that subsequent breaches (if the ticket re-opens) trigger fresh alerts. Callers are responsible for committing after calling this function. """ from app.models import SystemSetting raw = SystemSetting.get('sla_notified_tickets', '') current = set(int(x) for x in raw.split(',') if x.strip().isdigit()) current.discard(ticket_id) SystemSetting.set( 'sla_notified_tickets', ','.join(str(i) for i in sorted(current)), )