o2o-site-AEO/solution/backend/services/alert_service.py

"""장애 알림 — 영구 저장 + 재시도 + 중복 억제.

★ 왜 이 모양인가
  잡 큐 소진(JobStatus.DEAD) · BUILD 잡의 업무 실패(게이트 반려가 아닌 렌더·인프라 실패) ·
  노래 같은 곁가지의 부분 실패 · 잡 큐 정체를 Teams 로 알린다. 알림을 만드는 자리(worker/runner.py ·
  build_service.py · scheduler)는 이 모듈의 send_alert() 하나만 부르면 된다 — 언제 실제로
  보낼지, 같은 사유를 몇 번이나 다시 보낼지는 전부 여기서 정한다.

★ 재시도마다 중복 스팸을 내지 않는다 (dedupe)
  같은 dedupe_key 로 "아직 안 풀린" 알림이 있으면 새로 만들지 않는다 — 잡이 몇 번을 실패하며
  재큐되든 사람에게는 처음 한 통만 간다. 문제가 사라지면(resolve_alert) 그 dedupe_key 는
  다시 "풀린" 상태가 되고, 다음에 같은 사유가 또 터지면 새로 알린다.

★ 영구 저장 + 재시도 (outbox)
  webhook 전송이 그 자리에서 실패해도(네트워크 순단 등) 알림 자체를 잃지 않는다 — DB 에
  PENDING 으로 남기고 process_outbox() 가 백오프를 두고 다시 시도한다. 워커·API 프로세스가
  재시작돼도 이 표만 보면 뭐가 안 나갔는지 안다.

★ 비밀·개인정보를 남기지 않는다 (scrub)
  detail 은 저장 **전에** 한 번 걸러진다 — 외부 API 예외 메시지가 쿼리스트링에 키를 실어
  보내는 경우가 있다(TourAPI·Suno 등). 전화번호·API 키·bearer 토큰·이메일을 마스킹한다.

★ webhook 미설정이면 조용히 아무 일도 안 한다(teams_webhook.is_configured). 서버는 그대로 뜬다.
"""
import os
import re
from datetime import timedelta

from common.database.db_session_manager import DB_SESSION_MNG
from common.enums import DBType
from common.logger import LOG
from common.utils.gtime import GTime
from crud import alert_crud
from crud.job_crud import compute_backoff
from services import teams_webhook

# 중복 억제 창(분). 이 시간 안에 같은 dedupe_key 로 또 send_alert 가 불리면 새로 만들지 않는다.
DEDUPE_WINDOW_MIN_ENV = "ALERT_DEDUPE_WINDOW_MIN"
DEFAULT_DEDUPE_WINDOW_MIN = 60
# 재시도 상한. 소진되면 AlertStatus.FAILED — 더 자동으로는 안 보낸다.
MAX_ATTEMPTS = 5

_DETAIL_MAX_LEN = 2000

# ── 비밀·개인정보 마스킹 ──────────────────────────────────────────────────
_RE_QUERY_SECRET = re.compile(
    r"(?i)([?&](?:key|token|api[_-]?key|secret|access[_-]?token|auth)=)[^\s&]+"
)
_RE_BEARER = re.compile(r"(?i)\bBearer\s+[A-Za-z0-9\-_.]{8,}")
_RE_KV_SECRET = re.compile(r"(?i)\b(password|passwd|pwd|secret|api[_-]?key)\s*[:=]\s*\S+")
_RE_EMAIL = re.compile(r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}")


def _scrub(text: str) -> str:
    """저장 전에 반드시 한 번 거친다. 순서가 중요하다 — 쿼리스트링을 먼저 지워야
    그 값이 이메일 형태여도 뒤의 이메일 마스킹이 이중으로 손대지 않는다."""
    if not text:
        return ""
    out = _RE_QUERY_SECRET.sub(r"\1***", text)
    out = _RE_BEARER.sub("Bearer ***", out)
    out = _RE_KV_SECRET.sub(lambda m: f"{m.group(1)}=***", out)
    out = _RE_EMAIL.sub(lambda m: m.group(0)[:2] + "***@***", out)
    return out[:_DETAIL_MAX_LEN]


def _dedupe_window_min() -> int:
    try:
        return int(os.environ.get(DEDUPE_WINDOW_MIN_ENV) or DEFAULT_DEDUPE_WINDOW_MIN)
    except ValueError:
        return DEFAULT_DEDUPE_WINDOW_MIN


async def send_alert(kind: str, title: str, detail: str = "", dedupe_key: str | None = None) -> None:
    """알림을 큐에 넣는다(즉시 보내지 않는다 — process_outbox 가 보낸다).

    ★ 즉시 안 보내는 이유: 이 함수는 워커의 실패 처리 경로(예외 발생 지점)에서 불린다.
      여기서 동기적으로 webhook 을 때리면 그 지연·재시도가 잡 처리 자체를 늦춘다. 큐에
      적재만 하고 별도 스윕(scheduler)이 실제 전송을 맡는다 — 알림 발송 실패가 발행
      파이프라인에 영향을 주지 않는다(파일 머리주석의 관심사 분리)."""
    try:
        async def _op(session):
            if dedupe_key:
                existing = await alert_crud.latest_unresolved(session, dedupe_key)
                if existing is not None:
                    return  # 이미 이 사유로 풀리지 않은 알림이 있다 — 또 만들지 않는다.
            await alert_crud.insert(session, {
                "kind": kind[:50],
                "dedupe_key": dedupe_key[:200] if dedupe_key else None,
                "title": title[:200],
                "detail": _scrub(detail),
            })

        await DB_SESSION_MNG.execute_lambda_write(DBType.MAIN.value, _op)
    except Exception as ex:  # noqa: BLE001 — 알림 적재 실패가 원래 하던 일(잡 처리)을 죽이면 안 된다
        LOG.w(f"[alert] 적재 실패(무시하고 계속): {type(ex).__name__}: {ex}")


async def resolve_alert(dedupe_key: str, title: str, detail: str = "") -> None:
    """이 dedupe_key 로 안 풀린 알림이 있으면 "복구됨" 을 한 번 알리고 풀린 것으로 남긴다.

    ★ 안 풀린 알림이 없으면(애초에 문제가 없었다) 아무것도 하지 않는다 — 정상 상태마다
      "복구됨" 을 보내면 그게 새로운 스팸이 된다."""
    try:
        async def _op(session):
            existing = await alert_crud.latest_unresolved(session, dedupe_key)
            if existing is None:
                return
            await alert_crud.mark_resolved(session, existing.alert_id)
            await alert_crud.insert(session, {
                "kind": "recovery",
                "dedupe_key": None,  # 복구 알림 자신은 dedupe 대상이 아니다 — 매번 보낸다.
                "title": title[:200],
                "detail": _scrub(detail),
            })

        await DB_SESSION_MNG.execute_lambda_write(DBType.MAIN.value, _op)
    except Exception as ex:  # noqa: BLE001
        LOG.w(f"[alert] 복구 알림 적재 실패(무시하고 계속): {type(ex).__name__}: {ex}")


async def process_outbox(limit: int = 20) -> dict:
    """PENDING 알림을 실제로 보낸다. 스케줄러가 주기적으로 부른다(scheduler/jobs.py).

    ★ 잡 큐의 백오프·소진 규칙(crud/job_crud.compute_backoff)을 그대로 재사용한다 —
      "몇 번 실패하면 얼마나 쉬고 언제 포기하나" 를 두 번 설계하지 않는다."""
    sent = failed = 0
    try:
        async def _load(session):
            return await alert_crud.due_pending(session, limit)

        due = await DB_SESSION_MNG.execute_lambda_write(DBType.MAIN.value, _load)
    except Exception as ex:  # noqa: BLE001
        LOG.w(f"[alert] outbox 조회 실패: {type(ex).__name__}: {ex}")
        return {"sent": 0, "failed": 0}

    for row in due:
        ok = await teams_webhook.send(row.title, row.detail or "")

        async def _update(session, row=row, ok=ok):
            if ok:
                await alert_crud.mark_sent(session, row.alert_id)
            else:
                attempts = row.attempts + 1
                if attempts >= MAX_ATTEMPTS:
                    await alert_crud.mark_exhausted(session, row.alert_id, attempts)
                else:
                    next_at = GTime.UTC() + timedelta(seconds=compute_backoff(attempts))
                    await alert_crud.mark_retry(session, row.alert_id, attempts, next_at)

        try:
            await DB_SESSION_MNG.execute_lambda_write(DBType.MAIN.value, _update)
        except Exception as ex:  # noqa: BLE001
            LOG.w(f"[alert] outbox 갱신 실패 {row.alert_id}: {type(ex).__name__}: {ex}")
            continue
        if ok:
            sent += 1
        else:
            failed += 1

    if sent or failed:
        LOG.i(f"[alert] outbox 스윕 — 전송 {sent}건 · 재시도/소진 {failed}건")
    return {"sent": sent, "failed": failed}