""" 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()})