Blame for backend/settings.py

"""
Settings, in two scopes:

- Personal (per person, per org): the name shown in the console and which notices they want.
  Any member edits their own.
- Organisation: the org's name (owner or admin) and how its emails go out (operator or above):
  the sender name beside the address, a reply-to address, the mailbox the composer picks by
  default, and whether a copy of each email is filed in the mailbox's Sent folder.

Defaults live here, so a setting nobody has touched still reads as a full set of values.
"""

from __future__ import annotations

import re
from datetime import datetime, timezone

from backend import firestore_client as db
from backend import mailboxes, orgs

_EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$")
_MAX_NAME = 80

NOTIFICATION_KEYS = ("campaign_finished", "emails_turned_away", "mailbox_problem", "new_signup", "weekly_summary")
DEFAULT_NOTIFICATIONS = {
    "campaign_finished": False,
    "emails_turned_away": True,
    "mailbox_problem": True,
    "new_signup": False,
    "weekly_summary": False,
}
DEFAULT_EMAILS = {"from_name": "", "reply_to": "", "default_mailbox_id": "", "save_sent_copy": True, "pause_seconds": 2}
# Seconds M8ion waits between two emails of a campaign. Some mail servers accept a fast burst and
# then never deliver it, so campaigns are spaced out unless the organisation turns this down.
MAX_PAUSE_SECONDS = 30
ORG_PROFILE_FIELDS = ("website", "domain", "country", "state", "lga", "address", "postal_code", "timezone")
_DOMAIN_RE = re.compile(r"^(?=.{3,253}$)([a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)+[a-z]{2,}$")


def _now() -> str:
    return datetime.now(timezone.utc).isoformat()


def _personal_path(org_id: str, uid: str) -> str:
    return f"orgs/{org_id}/personal_settings/{uid}"


def _emails_path(org_id: str) -> str:
    return f"orgs/{org_id}/settings/emails"


def _clean_name(value: str, what: str) -> str:
    value = " ".join(str(value or "").split())
    if len(value) > _MAX_NAME:
        raise ValueError(f"Keep the {what} to {_MAX_NAME} characters or fewer")
    return value


def get_settings(org_id: str, uid: str) -> dict:
    """Everything the Settings pages show, with defaults filled in."""
    role = orgs.require_permission(org_id, uid)
    permissions = sorted(orgs.permissions_for(org_id, role))
    personal = db.get_doc(_personal_path(org_id, uid)) or {}
    return {
        "role": role,
        "permissions": permissions,
        "profile": profile_of(org_id, uid, personal),
        "organisation": organisation(org_id),
        "notifications": DEFAULT_NOTIFICATIONS | {k: bool(v) for k, v in (personal.get("notifications") or {}).items() if k in NOTIFICATION_KEYS},
        "emails": email_settings(org_id),
    }


def email_settings(org_id: str) -> dict:
    """How the org's emails go out. Read by the send itself, so it takes no actor."""
    return DEFAULT_EMAILS | {k: v for k, v in (db.get_doc(_emails_path(org_id)) or {}).items() if k in DEFAULT_EMAILS}


PROFILE_EXTRAS = ("job_title", "phone", "about")


def profile_of(org_id: str, uid: str, personal: dict | None = None) -> dict:
    """The person's own profile: what they chose to share, plus facts about their use of this organisation."""
    personal = personal if personal is not None else (db.get_doc(_personal_path(org_id, uid)) or {})
    member = db.get_doc(f"orgs/{org_id}/members/{uid}") or {}
    sends = [d for d in db.query_prefix(f"orgs/{org_id}/sends/").values() if d.get("created_by") == uid]
    return {
        "display_name": personal.get("display_name", ""),
        **{k: personal.get(k, "") for k in PROFILE_EXTRAS},
        "active_since": member.get("added_at"),
        "campaigns_sent": len(sends),
        "last_active": max((d.get("created_at") or "" for d in sends), default="") or None,
    }


def update_profile(org_id: str, uid: str, display_name: str, extras: dict | None = None) -> dict:
    orgs.require_permission(org_id, uid)
    name = _clean_name(display_name, "name")
    update = {"display_name": name}
    for key in PROFILE_EXTRAS:
        if extras and key in extras:
            limit = 500 if key == "about" else 80
            value = " ".join(str(extras[key] or "").split()) if key != "about" else str(extras[key] or "").strip()
            if len(value) > limit:
                raise ValueError(f"Keep this to {limit} characters or fewer")
            update[key] = value
    _merge(_personal_path(org_id, uid), update)
    return profile_of(org_id, uid)


def update_notifications(org_id: str, uid: str, values: dict) -> dict:
    orgs.require_permission(org_id, uid)
    unknown = set(values) - set(NOTIFICATION_KEYS)
    if unknown:
        raise ValueError(f"Unknown notice: {', '.join(sorted(unknown))}")
    current = get_settings(org_id, uid)["notifications"] | {k: bool(v) for k, v in values.items()}
    _merge(_personal_path(org_id, uid), {"notifications": current})
    return current


def organisation(org_id: str) -> dict:
    org = orgs.get_org(org_id)
    profile = org.get("profile") or {}
    return {"name": org["name"], **{k: profile.get(k, "") for k in ORG_PROFILE_FIELDS}}


def rename_org(org_id: str, uid: str, name: str, profile: dict | None = None) -> dict:
    """Saves the org's name and its details: website, main domain, country, state, LGA, postal
    address and postal code (many countries' email laws ask for one in marketing emails) and time zone."""
    orgs.require_permission(org_id, uid, "manage_settings")
    name = _clean_name(name, "organisation name")
    if not name:
        raise ValueError("Give the organisation a name")
    update: dict = {"name": name}
    if profile is not None:
        unknown = set(profile) - set(ORG_PROFILE_FIELDS)
        if unknown:
            raise ValueError(f"Unknown organisation detail: {', '.join(sorted(unknown))}")
        clean = {k: " ".join(str(profile.get(k) or "").split()) for k in ORG_PROFILE_FIELDS}
        if len(clean["address"]) > 300:
            raise ValueError("Keep the address to 300 characters or fewer")
        if len(clean["postal_code"]) > 20:
            raise ValueError("Keep the postal code to 20 characters or fewer")
        for key in ("country", "state", "lga"):
            if len(clean[key]) > 100:
                raise ValueError(f"Keep the {key} to 100 characters or fewer")
        clean["domain"] = clean["domain"].lower().removeprefix("https://").removeprefix("http://").removeprefix("www.").rstrip("/")
        if clean["domain"] and not _DOMAIN_RE.match(clean["domain"]):
            raise ValueError("Enter a domain like yourcompany.com")
        if clean["website"] and not clean["website"].startswith(("https://", "http://")):
            clean["website"] = "https://" + clean["website"]
        if clean["timezone"]:
            try:
                from zoneinfo import ZoneInfo
                ZoneInfo(clean["timezone"])
            except Exception as exc:  # noqa: BLE001
                raise ValueError("Pick a time zone from the list") from exc
        update["profile"] = clean
    db.update_doc(f"orgs/{org_id}", update)
    return organisation(org_id)


def update_emails(org_id: str, uid: str, values: dict) -> dict:
    orgs.require_permission(org_id, uid, "manage_settings")
    unknown = set(values) - set(DEFAULT_EMAILS)
    if unknown:
        raise ValueError(f"Unknown email setting: {', '.join(sorted(unknown))}")
    clean: dict = {}
    if "from_name" in values:
        clean["from_name"] = _clean_name(values["from_name"], "sender name").replace('"', "")
    if "reply_to" in values:
        reply_to = str(values["reply_to"] or "").strip()
        if reply_to and not _EMAIL_RE.match(reply_to):
            raise ValueError("The reply-to address does not look like an email address")
        clean["reply_to"] = reply_to
    if "default_mailbox_id" in values:
        mailbox_id = str(values["default_mailbox_id"] or "")
        if mailbox_id:
            mailboxes.get_mailbox(org_id, mailbox_id)  # raises NotFoundError for another org's id
        clean["default_mailbox_id"] = mailbox_id
    if "save_sent_copy" in values:
        clean["save_sent_copy"] = bool(values["save_sent_copy"])
    if "pause_seconds" in values:
        try:
            pause = float(values["pause_seconds"] or 0)
        except (TypeError, ValueError) as exc:
            raise ValueError("The pause must be a number of seconds") from exc
        if not 0 <= pause <= MAX_PAUSE_SECONDS:
            raise ValueError(f"The pause must be between 0 and {MAX_PAUSE_SECONDS} seconds")
        clean["pause_seconds"] = pause
    _merge(_emails_path(org_id), clean)
    return email_settings(org_id)


def _merge(path: str, values: dict) -> None:
    doc = db.get_doc(path) or {}
    db.set_doc(path, doc | values | {"updated_at": _now()})