1 files changed, 77 insertions, 1 deletions

diff --git a/packages/meshbay-hub/src/meshbay_hub/mail.py b/packages/meshbay-hub/src/meshbay_hub/mail.py
index 661a1f3..aa6b74e 100644
--- a/packages/meshbay-hub/src/meshbay_hub/mail.py
+++ b/packages/meshbay-hub/src/meshbay_hub/mail.py
@@ -105,6 +105,67 @@ async def _take(db, key: str, window: timedelta, ceiling: int,
     row.last_sent = now
 
 
+# The two global ceilings, and what each one means to somebody waiting.
+EXHAUSTED_GENERAL = "general"      # sign-ups and address changes refused
+EXHAUSTED_ALL = "all"              # even a passphrase reset is refused
+
+
+async def _announce_exhaustion(scope: str) -> None:
+    """Tell the administrators, once per hour, that the hub stopped sending.
+
+    In its own session, and never raising: this runs while a request is being
+    refused, and an alert that fails must not turn a refusal into a 500. Same
+    reason `_mark_hosted` opens its own session next door.
+
+    Once per window, keyed on a row rather than a flag in memory, because the
+    thing being reported is exactly the kind of event a restart would erase —
+    and a hub that is refusing mail is a hub somebody is probably restarting.
+    """
+    from sqlalchemy import select
+
+    from meshbay_hub.api.deps import user_is_admin
+    from meshbay_hub.api.notifications import create_notification
+    from meshbay_hub.db.engine import get_session_factory
+    from meshbay_hub.db.models import MailQuota, User
+
+    try:
+        async with get_session_factory()() as db:
+            key = f"alert:{scope}"
+            now = datetime.now(timezone.utc)
+            row = await db.get(MailQuota, key)
+            if row is not None and row.last_sent is not None:
+                last = row.last_sent
+                if last.tzinfo is None:
+                    last = last.replace(tzinfo=timezone.utc)
+                if now - last < timedelta(hours=1):
+                    return
+            if row is None:
+                row = MailQuota(key=key, window_start=now, count=0)
+                db.add(row)
+            row.last_sent = now
+            row.count += 1
+
+            admins = [u for u in (await db.execute(select(User).where(
+                User.status == "active"))).scalars().all() if user_is_admin(u)]
+            for admin in admins:
+                await create_notification(
+                    db, admin.id, f"mail_budget_{scope}",
+                    "This hub has stopped sending mail for this hour"
+                    if scope == EXHAUSTED_ALL
+                    else "This hub has stopped sending sign-up mail for this hour",
+                    detail=("Passphrase resets and invitations are refused too."
+                            if scope == EXHAUSTED_ALL
+                            else "Passphrase resets and invitations still go out."),
+                    link="#/admin",
+                    aggregate=False,
+                )
+            await db.commit()
+            log.warning("Mail budget exhausted (%s) — %d administrator(s) told",
+                        scope, len(admins))
+    except Exception as e:
+        log.warning("Could not announce the mail budget: %s", e)
+
+
 async def reserve(db, purpose: str, address: str) -> None:
     """Charge one send, or raise MailRefused. The caller owns the commit."""
     from meshbay_hub import hub_settings
@@ -120,7 +181,15 @@ async def reserve(db, purpose: str, address: str) -> None:
     budget = limits["hourly_budget"]
     if purpose not in RECOVERY_PURPOSES:
         budget = max(0, budget - limits["hourly_reserved_for_recovery"])
-    await _take(db, "hour", timedelta(hours=1), budget, None)
+    try:
+        await _take(db, "hour", timedelta(hours=1), budget, None)
+    except MailRefused:
+        # An operator finds out here or not at all: a refusal is otherwise a
+        # line in the journal, and a hub that has stopped sending sign-up
+        # codes looks exactly like one nobody is signing up to.
+        await _announce_exhaustion(
+            EXHAUSTED_ALL if purpose in RECOVERY_PURPOSES else EXHAUSTED_GENERAL)
+        raise
 
     # Then the recipient: across every purpose, account and endpoint. This is
     # what a person being mail-bombed actually experiences, and the only bound
@@ -153,7 +222,14 @@ async def status(db) -> dict:
         .where(MailQuota.key.like("dest:%"))) or 0
 
     general = max(0, limits["hourly_budget"] - limits["hourly_reserved_for_recovery"])
+    # Named rather than left for the reader to compute from two numbers: the
+    # panel draws a warning off these, and "is it still sending" is the
+    # question an operator opens this page to answer.
+    general_exhausted = used >= general
+    all_exhausted = used >= limits["hourly_budget"]
     return {
+        "general_exhausted":  general_exhausted,
+        "all_exhausted":      all_exhausted,
         "hourly_budget":      limits["hourly_budget"],
         "hourly_used":        used,
         "hour_started_at":    window_start,