PacketStation/server/packetstation/api/notifications.py

"""Notifications: the Admin's routing rules, test send and sent log, and each person's own view on Account.

Who gets what lives in the database (notification_rules, _recipients, _exclusions), never in the environment.
"""

from __future__ import annotations

import logging
import re
import uuid
from datetime import time
from typing import Any, Literal, cast

from fastapi import APIRouter, Depends, Query, Request
from psycopg.rows import dict_row
from pydantic import BaseModel, ConfigDict, Field

from .. import audit
from ..auth.ldap import LdapUnavailable
from ..notify import seed as seed_rules
from ..notify import send as notify_send
from ..notify import smtp
from ..notify.catalog import EVENT_TYPES, SECURITY_TYPES
from ..notify.routing import (
    DIRECTORY_DOWN,
    NotifyDirectory,
    Rule,
    address_ok,
    cn_of,
    dn_under,
    load_rules,
    supervisor_depts,
)
from ..settings import Settings
from ..text import clean_text
from .deps import Db, Principal, admin_principal, current_principal, directory_of, get_db, password_principal
from .errors import ApiError
from .packets import WRITE

log = logging.getLogger(__name__)

router = APIRouter(tags=["notifications"])

ADMIN = "/api/admin/notifications"
ACCOUNT = "/api/account/notifications"
RECIPIENTS_MAX = 50
EXCLUSIONS_MAX = 200
TIME_RE = re.compile(r"^([01][0-9]|2[0-3]):([0-5][0-9])$")
NAME_MAX = 256
DN_MAX = 1024


def _directory(request: Request) -> NotifyDirectory:
    return cast(NotifyDirectory, directory_of(request))


def _settings(request: Request) -> Settings:
    s: Settings = request.app.state.settings
    return s


def _known(event_type: str) -> None:
    if event_type not in EVENT_TYPES:
        raise ApiError(404, "not_found", "No such notification.")


# ---------------------------------------------------------------- views


def rule_view(conn: Any, rule: Rule) -> dict[str, Any]:
    et = EVENT_TYPES[rule.event_type]
    last = conn.execute(
        "SELECT at, result FROM notifications_sent WHERE event_type = %s ORDER BY id DESC LIMIT 1",
        (rule.event_type,),
    ).fetchone()
    return {
        "event_type": rule.event_type,
        "label": et.label,
        "description": et.description,
        "security": et.security,
        "dept_scoped": et.dept_scoped,
        "param": {"name": et.param, "value": rule.param_value, "default": et.param_default} if et.param else None,
        "enabled": rule.enabled,
        "delivery": rule.delivery,
        "digest_time": rule.digest_time.strftime("%H:%M"),
        "recipients": [
            {
                "id": r["id"],
                "kind": r["kind"],
                "group_dn": r["group_dn"],
                "group_name": r["group_name"],
                "user_id": str(r["user_id"]) if r["user_id"] else None,
                "display_name": r.get("display_name"),
                "address": r["address"],
            }
            for r in rule.recipients
        ],
        "exclusions": [
            {"user_id": str(x["user_id"]), "display_name": x.get("display_name"), "self": bool(x["self"])}
            for x in rule.exclusions
        ],
        "version": rule.version,
        "last_sent": {"at": last[0], "result": last[1]} if last else None,
    }


def _rules(conn: Any, settings: Settings) -> dict[str, Rule]:
    seed_rules(conn, settings)
    return load_rules(conn)


@router.get(ADMIN)
def admin_rules(
    request: Request,
    p: Principal = Depends(admin_principal),  # noqa: B008
    d: Db = Depends(get_db),  # noqa: B008
) -> dict[str, Any]:
    s = _settings(request)
    rules = _rules(d.conn, s)
    return {
        "dry_run": bool(s.notify_dry_run),
        "smtp": {"host": s.smtp_host, "port": s.smtp_port, "from": s.smtp_from, "starttls": s.smtp_starttls},
        "subject_prefix": smtp.subject_prefix(s),
        "rules": [rule_view(d.conn, rules[k]) for k in EVENT_TYPES if k in rules],
    }


# ---------------------------------------------------------------- update a rule


class RecipientIn(BaseModel):
    model_config = ConfigDict(extra="forbid")

    kind: Literal["group", "user", "mailbox", "dept_supervisor"]
    group_dn: str | None = Field(default=None, max_length=DN_MAX)
    group_name: str | None = Field(default=None, max_length=NAME_MAX)
    user_id: uuid.UUID | None = None
    address: str | None = Field(default=None, max_length=254)


class RuleIn(BaseModel):
    model_config = ConfigDict(extra="forbid")

    enabled: bool
    delivery: Literal["immediate", "digest"]
    digest_time: str = Field(max_length=5)
    param: int | None = Field(default=None, ge=1, le=10000)
    recipients: list[RecipientIn] = Field(max_length=RECIPIENTS_MAX)
    exclusions: list[uuid.UUID] = Field(default_factory=list, max_length=EXCLUSIONS_MAX)
    version: int


def _invalid(text: str) -> ApiError:
    return ApiError(422, "invalid_request", text)


def _clean_recipients(conn: Any, s: Settings, event_type: str, items: list[RecipientIn]) -> list[dict[str, Any]]:
    et = EVENT_TYPES[event_type]
    base = s.ldap_base_dn.strip() or s.ldap_group_base_dn.strip()
    out: list[dict[str, Any]] = []
    seen: set[tuple[str, str]] = set()
    for r in items:
        row: dict[str, Any] = {"kind": r.kind, "group_dn": None, "group_name": None, "user_id": None, "address": None}
        if r.kind == "group":
            dn = clean_text(r.group_dn or "", DN_MAX)
            if not dn or not dn.lower().startswith("cn="):
                raise _invalid("A group needs its DN.")
            if base and not dn_under(dn, base):
                raise _invalid("The group must be in the domain (under LDAP_BASE_DN).")
            row["group_dn"] = dn
            row["group_name"] = clean_text(r.group_name or "", NAME_MAX) or cn_of(dn)
            key = ("group", dn.lower())
        elif r.kind == "user":
            if r.user_id is None:
                raise _invalid("A user recipient needs a user.")
            if conn.execute("SELECT 1 FROM users WHERE id = %s", (r.user_id,)).fetchone() is None:
                raise _invalid("No such user.")
            row["user_id"] = r.user_id
            key = ("user", str(r.user_id))
        elif r.kind == "mailbox":
            addr = (r.address or "").strip()
            if not address_ok(addr):
                raise _invalid(f"Not an email address: {addr[:80]}")
            row["address"] = addr
            key = ("mailbox", addr.lower())
        else:
            if not et.dept_scoped:
                raise _invalid("This notification has no department, so it has no Supervisor group.")
            key = ("dept_supervisor", "")
        if key in seen:
            continue
        seen.add(key)
        out.append(row)
    return out


@router.put(f"{ADMIN}/{{event_type}}", dependencies=WRITE)
def update_rule(
    event_type: str,
    body: RuleIn,
    request: Request,
    p: Principal = Depends(admin_principal),  # noqa: B008
    d: Db = Depends(get_db),  # noqa: B008
) -> dict[str, Any]:
    _known(event_type)
    s = _settings(request)
    conn = d.conn
    et = EVENT_TYPES[event_type]
    m = TIME_RE.match(body.digest_time)
    if m is None:
        raise _invalid("The digest time is HH:MM.")
    digest_at = time(int(m.group(1)), int(m.group(2)))
    if body.param is not None and et.param is None:
        raise _invalid("This notification takes no setting.")
    excl = list(dict.fromkeys(body.exclusions))
    if excl and event_type in SECURITY_TYPES:
        raise _invalid("Nobody can be left out of a security notification.")
    recipients = _clean_recipients(conn, s, event_type, body.recipients)
    if excl:
        found = {r[0] for r in conn.execute("SELECT id FROM users WHERE id = ANY(%s)", (excl,)).fetchall()}
        if len(found) != len(excl):
            raise _invalid("An excluded person is not a known user.")
    _rules(conn, s)
    with conn.transaction():
        before_rule = load_rules(conn, event_type).get(event_type)
        assert before_rule is not None
        before = rule_view(conn, before_rule)
        done = conn.execute(
            "UPDATE notification_rules SET enabled = %s, delivery = %s, digest_time = %s, param = %s,"
            " version = version + 1, updated_at = now(), updated_by = %s WHERE event_type = %s AND version = %s"
            " RETURNING version",
            (
                body.enabled,
                body.delivery,
                digest_at,
                (body.param or et.param_default) if et.param else None,
                p.user_id,
                event_type,
                body.version,
            ),
        ).fetchone()
        if done is None:
            raise ApiError(
                409,
                "version_conflict",
                "Someone else changed this notification. Here is the current one.",
                extra={"current": before},
            )
        conn.execute("DELETE FROM notification_recipients WHERE event_type = %s", (event_type,))
        for r in recipients:
            conn.execute(
                "INSERT INTO notification_recipients (event_type, kind, group_dn, group_name, user_id, address)"
                " VALUES (%s, %s, %s, %s, %s, %s)",
                (event_type, r["kind"], r["group_dn"], r["group_name"], r["user_id"], r["address"]),
            )
        conn.execute(
            "DELETE FROM notification_exclusions WHERE event_type = %s AND NOT self AND NOT (user_id = ANY(%s))",
            (event_type, excl),
        )
        for uid in excl:
            conn.execute(
                "INSERT INTO notification_exclusions (event_type, user_id, self, created_by) VALUES (%s, %s, false, %s)"
                " ON CONFLICT (event_type, user_id) DO NOTHING",
                (event_type, uid, p.user_id),
            )
        after_rule = load_rules(conn, event_type)[event_type]
        after = rule_view(conn, after_rule)
        audit.append(
            conn,
            action="admin.notification_rule_updated",
            user_id=p.user_id,
            session_id=p.session_id,
            role_snapshot=p.role_snapshot,
            subject_type="notification_rule",
            subject_id=event_type,
            detail={"event_type": event_type, "before": _audit_view(before), "after": _audit_view(after)},
            ip=p.ip,
        )
    return after


def _audit_view(v: dict[str, Any]) -> dict[str, Any]:
    keep = ("enabled", "delivery", "digest_time", "param", "version")
    out = {k: v[k] for k in keep}
    out["recipients"] = [
        {k: r[k] for k in ("kind", "group_dn", "group_name", "user_id", "address") if r.get(k)} for r in v["recipients"]
    ]
    out["exclusions"] = [{"user_id": x["user_id"], "self": x["self"]} for x in v["exclusions"]]
    return out


# ---------------------------------------------------------------- group search, test send, sent log


@router.get(f"{ADMIN}/groups")
def search_groups(
    request: Request,
    p: Principal = Depends(admin_principal),  # noqa: B008
    q: str = Query(..., min_length=2, max_length=100),
) -> list[dict[str, Any]]:
    try:
        found = _directory(request).search_groups(q.strip(), 25)
    except LdapUnavailable as e:
        log.warning("group search: AD unavailable: %s", e)
        raise ApiError(503, "ldap_unavailable", "The directory is not available. Try again in a minute.") from None
    return [{"dn": g.get("dn"), "name": g.get("name"), "description": g.get("description")} for g in found]


class TestIn(BaseModel):
    model_config = ConfigDict(extra="forbid")

    to: str | None = Field(default=None, max_length=254)


@router.post(f"{ADMIN}/{{event_type}}/test", dependencies=WRITE)
def test_send(
    event_type: str,
    body: TestIn,
    request: Request,
    p: Principal = Depends(admin_principal),  # noqa: B008
    d: Db = Depends(get_db),  # noqa: B008
) -> dict[str, Any]:
    _known(event_type)
    s = _settings(request)
    to = (body.to or "").strip() or None
    if to is not None and not address_ok(to):
        raise _invalid("Not an email address.")
    conn = d.conn
    _rules(conn, s)
    out = notify_send.send_test(conn, s, _directory(request), event_type, to=to, requested_by=p.user_id)
    audit.append(
        conn,
        action="admin.notification_test",
        user_id=p.user_id,
        session_id=p.session_id,
        role_snapshot=p.role_snapshot,
        subject_type="notification_rule",
        subject_id=event_type,
        detail={
            "event_type": event_type,
            "to": to,
            "result": out["result"],
            "recipients": len(out["recipients"]),
            "sent_id": out["sent_id"],
        },
        ip=p.ip,
    )
    return out


@router.get(f"{ADMIN}/sent")
def sent_log(
    p: Principal = Depends(admin_principal),  # noqa: B008
    d: Db = Depends(get_db),  # noqa: B008
    event_type: str | None = Query(None, max_length=64),
    limit: int = Query(50, ge=1, le=500),
    before: int | None = Query(None, ge=1),
) -> dict[str, Any]:
    sql = (
        "SELECT id, at, event_type, digest, test, subject, recipients, result, error FROM notifications_sent"
        " WHERE (%s::text IS NULL OR event_type = %s) AND (%s::bigint IS NULL OR id < %s) ORDER BY id DESC LIMIT %s"
    )
    args = (event_type or None, event_type or None, before, before)
    with d.conn.cursor(row_factory=dict_row) as cur:
        rows = [dict(r) for r in cur.execute(sql, (*args, limit)).fetchall()]
    return {"rows": rows, "next_before": rows[-1]["id"] if len(rows) == limit else None}


# ---------------------------------------------------------------- account


def _my_groups(directory: NotifyDirectory, user_id: uuid.UUID) -> tuple[Any, set[str]]:
    """(the AdUser or None, lower-case names of the user's groups in the PacketStation OU, nested)."""
    u = directory.get_user_by_guid(user_id)
    if u is None:
        return None, set()
    return u, {g.lower() for g in directory.groups_of(u.dn)}


def account_view(conn: Any, settings: Settings, directory: NotifyDirectory, p: Principal) -> list[dict[str, Any]]:
    rules = _rules(conn, settings)
    email = str(p.user.get("email") or "").strip().lower()
    me = str(p.user_id)
    sup = supervisor_depts(p.roles)
    groups: tuple[Any, set[str]] | None = None
    down = False
    member_cache: dict[str, bool] = {}
    out: list[dict[str, Any]] = []
    for key, et in EVENT_TYPES.items():
        rule = rules.get(key)
        if rule is None or not rule.enabled:
            continue
        why: list[str] = []
        for r in rule.recipients:
            kind = r["kind"]
            if kind == "user" and str(r["user_id"]) == me:
                why.append("user")
            elif kind == "mailbox" and email and str(r["address"]).strip().lower() == email:
                why.append("mailbox")
            elif kind == "dept_supervisor" and et.dept_scoped and sup:
                why.extend(f"Supervisor group of {code}" for code in sup)
            elif kind == "group" and not down:
                dn = str(r["group_dn"])
                name = str(r["group_name"] or cn_of(dn))
                try:
                    if dn_under(dn, settings.ldap_group_base_dn):
                        if groups is None:
                            groups = _my_groups(directory, p.user_id)
                        hit = name.lower() in groups[1]
                    else:
                        if dn.lower() not in member_cache:
                            member_cache[dn.lower()] = any(
                                u.guid == p.user_id for u in directory.members_of_dn(dn) if not u.disabled
                            )
                        hit = member_cache[dn.lower()]
                except Exception as e:
                    log.warning("account notifications: directory: %s", type(e).__name__)
                    down = True
                    continue
                if hit:
                    why.append(f"group {name}")
        excl = next((x for x in rule.exclusions if str(x["user_id"]) == me), None)
        opted_out = bool(excl and excl["self"])
        if excl and not excl["self"]:
            why.append("left out by an Admin")
        if down and any(r["kind"] == "group" for r in rule.recipients):
            why.append(DIRECTORY_DOWN)
        reaches = excl is None and any(w not in (DIRECTORY_DOWN, "left out by an Admin") for w in why)
        out.append(
            {
                "event_type": key,
                "label": et.label,
                "security": et.security,
                "delivery": rule.delivery,
                "reaches": reaches,
                "why": list(dict.fromkeys(why)),
                "opted_out": opted_out,
                "can_opt_out": not et.security,
            }
        )
    return out


@router.get(ACCOUNT)
def my_notifications(
    request: Request,
    p: Principal = Depends(current_principal),  # noqa: B008
    d: Db = Depends(get_db),  # noqa: B008
) -> list[dict[str, Any]]:
    return account_view(d.conn, _settings(request), _directory(request), p)


class OptOutIn(BaseModel):
    model_config = ConfigDict(extra="forbid")

    opt_out: bool


@router.put(f"{ACCOUNT}/{{event_type}}", dependencies=WRITE)
def opt_out(
    event_type: str,
    body: OptOutIn,
    request: Request,
    p: Principal = Depends(password_principal),  # noqa: B008
    d: Db = Depends(get_db),  # noqa: B008
) -> dict[str, Any]:
    _known(event_type)
    if event_type in SECURITY_TYPES:
        raise ApiError(403, "forbidden", "Security notifications cannot be turned off.")
    conn = d.conn
    s = _settings(request)
    _rules(conn, s)
    with conn.transaction():
        if body.opt_out:
            changed = conn.execute(
                "INSERT INTO notification_exclusions (event_type, user_id, self, created_by) VALUES (%s, %s, true, %s)"
                " ON CONFLICT (event_type, user_id) DO NOTHING",
                (event_type, p.user_id, p.user_id),
            ).rowcount
        else:
            changed = conn.execute(
                "DELETE FROM notification_exclusions WHERE event_type = %s AND user_id = %s AND self",
                (event_type, p.user_id),
            ).rowcount
        if changed:
            audit.append(
                conn,
                action="account.notification_opt_out",
                user_id=p.user_id,
                session_id=p.session_id,
                role_snapshot=p.role_snapshot,
                subject_type="notification_rule",
                subject_id=event_type,
                detail={"event_type": event_type, "opt_out": body.opt_out},
                ip=p.ip,
            )
    item = next((x for x in account_view(conn, s, _directory(request), p) if x["event_type"] == event_type), None)
    if item is None:
        # The rule is off: nothing reaches anyone, but the choice is kept
        et = EVENT_TYPES[event_type]
        item = {
            "event_type": event_type,
            "label": et.label,
            "security": False,
            "delivery": None,
            "reaches": False,
            "why": [],
            "opted_out": body.opt_out,
            "can_opt_out": True,
        }
    return item