Add issue resolution confirmation workflow

backend/app/services/issue_resolution.py

@@ -0,0 +1,378 @@
+from __future__ import annotations
+
+import asyncio
+from datetime import datetime, timedelta, timezone
+from html import escape
+import json
+import logging
+from typing import Any, Dict, Optional
+
+from ..config import settings as env_settings
+from ..db import (
+    add_portal_item_activity,
+    get_portal_item,
+    get_user_by_username,
+    list_portal_items,
+    update_portal_item,
+)
+from ..runtime import get_runtime_settings
+from .invite_email import resolve_user_delivery_email, send_generic_email
+
+
+logger = logging.getLogger(__name__)
+_SYSTEM_USER = "Magent"
+
+
+def _now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def _parse_datetime(value: Any) -> Optional[datetime]:
+    if not isinstance(value, str) or not value.strip():
+        return None
+    try:
+        parsed = datetime.fromisoformat(value.strip().replace("Z", "+00:00"))
+    except ValueError:
+        return None
+    return parsed.replace(tzinfo=parsed.tzinfo or timezone.utc).astimezone(timezone.utc)
+
+
+def _metadata(item: Dict[str, Any]) -> Dict[str, Any]:
+    raw = item.get("metadata_json")
+    if not isinstance(raw, str) or not raw.strip():
+        return {}
+    try:
+        parsed = json.loads(raw)
+    except (TypeError, ValueError):
+        return {}
+    return parsed if isinstance(parsed, dict) else {}
+
+
+def issue_resolution_state(item: Dict[str, Any]) -> Dict[str, Any]:
+    state = _metadata(item).get("resolutionConfirmation")
+    return dict(state) if isinstance(state, dict) else {}
+
+
+def _metadata_with_resolution(item: Dict[str, Any], state: Dict[str, Any]) -> str:
+    metadata = _metadata(item)
+    metadata["resolutionConfirmation"] = state
+    return json.dumps(metadata, separators=(",", ":"), sort_keys=True)
+
+
+def _interval_delta(value: int, unit: str) -> timedelta:
+    safe_value = max(1, min(int(value), 365))
+    normalized_unit = str(unit or "days").strip().lower()
+    if normalized_unit == "weeks":
+        return timedelta(weeks=safe_value)
+    if normalized_unit == "months":
+        return timedelta(days=30 * safe_value)
+    return timedelta(days=safe_value)
+
+
+def _workflow_settings() -> tuple[int, int, str]:
+    runtime = get_runtime_settings()
+    attempts = max(0, min(int(runtime.issue_confirmation_contact_attempts or 0), 10))
+    interval_value = max(1, min(int(runtime.issue_confirmation_interval_value or 1), 365))
+    interval_unit = str(runtime.issue_confirmation_interval_unit or "days").strip().lower()
+    if interval_unit not in {"days", "weeks", "months"}:
+        interval_unit = "days"
+    return attempts, interval_value, interval_unit
+
+
+def _app_url() -> str:
+    runtime = get_runtime_settings()
+    for value in (runtime.magent_application_url, runtime.magent_proxy_base_url, env_settings.cors_allow_origin):
+        candidate = str(value or "").strip()
+        if candidate:
+            return candidate.rstrip("/")
+    return f"http://localhost:{int(runtime.magent_application_port or 3000)}"
+
+
+def _issue_url(item_id: int) -> str:
+    return f"{_app_url()}/portal/issues?item={item_id}"
+
+
+def _activity(
+    item_id: int,
+    event_type: str,
+    message: str,
+    *,
+    actor_username: str = _SYSTEM_USER,
+    actor_role: str = "system",
+    metadata: Optional[Dict[str, Any]] = None,
+) -> None:
+    add_portal_item_activity(
+        item_id,
+        event_type=event_type,
+        actor_username=actor_username,
+        actor_role=actor_role,
+        message=message,
+        metadata_json=json.dumps(metadata, separators=(",", ":"), sort_keys=True) if metadata else None,
+    )
+
+
+def _close_issue(
+    item: Dict[str, Any],
+    *,
+    reason: str,
+    confirmed: bool,
+    actor_username: str = _SYSTEM_USER,
+    actor_role: str = "system",
+) -> Dict[str, Any]:
+    now = _now().isoformat()
+    state = issue_resolution_state(item)
+    state.update(
+        {
+            "status": "confirmed" if confirmed else "auto_closed",
+            "confirmedAt": now if confirmed else state.get("confirmedAt"),
+            "closedAt": now,
+            "nextContactAt": None,
+            "closedReason": reason,
+        }
+    )
+    updated = update_portal_item(
+        int(item["id"]),
+        status="closed",
+        issue_resolved_at=now,
+        metadata_json=_metadata_with_resolution(item, state),
+    )
+    if not updated:
+        raise RuntimeError("Issue could not be closed")
+    _activity(
+        int(item["id"]),
+        "resolution_confirmed" if confirmed else "issue_auto_closed",
+        reason,
+        actor_username=actor_username,
+        actor_role=actor_role,
+    )
+    return updated
+
+
+async def _contact_reporter(item: Dict[str, Any]) -> Dict[str, Any]:
+    maximum, interval_value, interval_unit = _workflow_settings()
+    state = issue_resolution_state(item)
+    attempts = max(0, int(state.get("attemptsSent") or 0))
+    if maximum <= 0:
+        return _close_issue(
+            item,
+            reason="Issue closed automatically because reporter confirmation emails are disabled.",
+            confirmed=False,
+        )
+    if attempts >= maximum:
+        return _close_issue(
+            item,
+            reason=f"Issue closed automatically after {attempts} confirmation email attempt(s) without a response.",
+            confirmed=False,
+        )
+
+    attempt_number = attempts + 1
+    reporter = get_user_by_username(str(item.get("created_by_username") or ""))
+    recipient = resolve_user_delivery_email(reporter)
+    issue_url = _issue_url(int(item["id"]))
+    sent = False
+    delivery_error: Optional[str] = None
+    if recipient:
+        subject = f"Is your issue fixed? #{item['id']} {item.get('title') or ''}".strip()
+        body_text = (
+            f"We have marked issue #{item['id']} as fixed and need your confirmation.\n\n"
+            f"Issue: {item.get('title') or 'Untitled issue'}\n"
+            f"Confirmation request: {attempt_number} of {maximum}\n\n"
+            f"Open the issue and choose whether it is fixed or still happening:\n{issue_url}\n\n"
+            "If you do not respond, Magent will close the issue automatically after the configured confirmation period."
+        )
+        body_html = (
+            '<div style="font-family:Segoe UI,Arial,sans-serif;color:#132033;">'
+            '<h2 style="margin:0 0 12px;">Is your issue fixed?</h2>'
+            f'<p style="line-height:1.6;">We have marked issue <strong>#{int(item["id"])}</strong> as fixed and need your confirmation.</p>'
+            f'<p style="line-height:1.6;"><strong>{escape(str(item.get("title") or "Untitled issue"))}</strong><br>'
+            f'Confirmation request {attempt_number} of {maximum}</p>'
+            f'<a href="{escape(issue_url)}" style="display:inline-block;padding:11px 18px;border-radius:8px;background:#1c6bff;color:#fff;text-decoration:none;font-weight:700;">Confirm the outcome</a>'
+            '<p style="margin-top:18px;color:#64748b;line-height:1.6;">If you do not respond, Magent will close the issue automatically after the configured confirmation period.</p>'
+            '</div>'
+        )
+        try:
+            await send_generic_email(
+                recipient_email=recipient,
+                subject=subject,
+                body_text=body_text,
+                body_html=body_html,
+            )
+            sent = True
+        except Exception as exc:
+            delivery_error = str(exc)
+            logger.exception("issue confirmation email failed item_id=%s attempt=%s", item.get("id"), attempt_number)
+    else:
+        delivery_error = "No email address is stored for the reporter."
+
+    now = _now()
+    state.update(
+        {
+            "status": "awaiting_confirmation",
+            "attemptsSent": attempt_number,
+            "maximumAttempts": maximum,
+            "lastContactAt": now.isoformat(),
+            "nextContactAt": (now + _interval_delta(interval_value, interval_unit)).isoformat(),
+            "intervalValue": interval_value,
+            "intervalUnit": interval_unit,
+            "lastDeliverySucceeded": sent,
+            "lastDeliveryError": delivery_error,
+        }
+    )
+    updated = update_portal_item(
+        int(item["id"]),
+        metadata_json=_metadata_with_resolution(item, state),
+    )
+    if not updated:
+        raise RuntimeError("Issue confirmation schedule could not be saved")
+    if sent:
+        message = f"Confirmation email {attempt_number} of {maximum} was sent to the reporter."
+    else:
+        message = f"Confirmation email {attempt_number} of {maximum} could not be delivered."
+    _activity(
+        int(item["id"]),
+        "confirmation_email_sent" if sent else "confirmation_email_failed",
+        message,
+        metadata={
+            "attempt": attempt_number,
+            "maximum": maximum,
+            "nextContactAt": state["nextContactAt"],
+            "deliveryError": delivery_error,
+        },
+    )
+    return updated
+
+
+async def begin_issue_confirmation(
+    item_id: int,
+    *,
+    actor_username: str,
+    actor_role: str,
+) -> Dict[str, Any]:
+    item = get_portal_item(item_id)
+    if not item or str(item.get("kind") or "").lower() != "issue":
+        raise ValueError("Issue not found")
+    now = _now().isoformat()
+    maximum, interval_value, interval_unit = _workflow_settings()
+    state = {
+        "status": "awaiting_confirmation",
+        "startedAt": now,
+        "attemptsSent": 0,
+        "maximumAttempts": maximum,
+        "lastContactAt": None,
+        "nextContactAt": now,
+        "intervalValue": interval_value,
+        "intervalUnit": interval_unit,
+        "confirmedAt": None,
+        "closedAt": None,
+    }
+    updated = update_portal_item(
+        item_id,
+        status="awaiting_confirmation",
+        issue_resolved_at=None,
+        metadata_json=_metadata_with_resolution(item, state),
+    )
+    if not updated:
+        raise RuntimeError("Issue confirmation workflow could not be started")
+    _activity(
+        item_id,
+        "resolution_proposed",
+        "The issue was marked fixed and sent to the reporter for confirmation.",
+        actor_username=actor_username,
+        actor_role=actor_role,
+        metadata={"maximumAttempts": maximum, "intervalValue": interval_value, "intervalUnit": interval_unit},
+    )
+    return await _contact_reporter(updated)
+
+
+def respond_to_issue_confirmation(
+    item_id: int,
+    *,
+    resolved: bool,
+    actor_username: str,
+    actor_role: str,
+) -> Dict[str, Any]:
+    item = get_portal_item(item_id)
+    if not item or str(item.get("kind") or "").lower() != "issue":
+        raise ValueError("Issue not found")
+    if str(item.get("status") or "").lower() != "awaiting_confirmation":
+        raise ValueError("This issue is not waiting for resolution confirmation")
+    if resolved:
+        return _close_issue(
+            item,
+            reason="The reporter confirmed that the issue is fixed.",
+            confirmed=True,
+            actor_username=actor_username,
+            actor_role=actor_role,
+        )
+
+    now = _now().isoformat()
+    state = issue_resolution_state(item)
+    state.update(
+        {
+            "status": "reported_still_broken",
+            "reporterResponseAt": now,
+            "nextContactAt": None,
+            "closedAt": None,
+        }
+    )
+    updated = update_portal_item(
+        item_id,
+        status="in_progress",
+        issue_resolved_at=None,
+        metadata_json=_metadata_with_resolution(item, state),
+    )
+    if not updated:
+        raise RuntimeError("Issue could not be reopened")
+    _activity(
+        item_id,
+        "resolution_rejected",
+        "The reporter said the issue is still happening. The issue was returned to In progress.",
+        actor_username=actor_username,
+        actor_role=actor_role,
+    )
+    return updated
+
+
+async def process_due_issue_confirmations(now: Optional[datetime] = None) -> Dict[str, int]:
+    current = (now or _now()).astimezone(timezone.utc)
+    items = list_portal_items(kind="issue", status="awaiting_confirmation", limit=500)
+    result = {"checked": len(items), "contacted": 0, "closed": 0, "failed": 0}
+    maximum, _, _ = _workflow_settings()
+    for item in items:
+        state = issue_resolution_state(item)
+        due_at = _parse_datetime(state.get("nextContactAt"))
+        if due_at and due_at > current:
+            continue
+        try:
+            attempts = max(0, int(state.get("attemptsSent") or 0))
+            if maximum <= 0 or attempts >= maximum:
+                _close_issue(
+                    item,
+                    reason=(
+                        "Issue closed automatically because reporter confirmation emails are disabled."
+                        if maximum <= 0
+                        else f"Issue closed automatically after {attempts} confirmation email attempt(s) without a response."
+                    ),
+                    confirmed=False,
+                )
+                result["closed"] += 1
+            else:
+                await _contact_reporter(item)
+                result["contacted"] += 1
+        except Exception:
+            result["failed"] += 1
+            logger.exception("issue confirmation processing failed item_id=%s", item.get("id"))
+    return result
+
+
+async def run_issue_confirmation_loop() -> None:
+    while True:
+        try:
+            result = await process_due_issue_confirmations()
+            if result["contacted"] or result["closed"] or result["failed"]:
+                logger.info("issue confirmation sweep complete result=%s", result)
+        except asyncio.CancelledError:
+            raise
+        except Exception:
+            logger.exception("issue confirmation sweep failed")
+        await asyncio.sleep(15 * 60)