Email a DM's recipient when they're genuinely offline (#66)

Scoped to direct messages only, and deliberately narrower than the
existing push/desktop "offline" (not connected to this room's channel
right now, which fires on every message) -- email uses GlobalPresence
instead (no open connection anywhere, or appear_offline), since the
other participant could easily just be active in a different room.
Debounced to the first unread message in the conversation rather than
firing on every message in a burst, resetting once they mark it read.
Reuses the existing SMTP/send_email infrastructure from the invite
feature, so it silently no-ops if SMTP isn't configured, same as
everywhere else that already uses it.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

backend/app/routers/webhooks.py

@@ -28,4 +28,15 @@ async def incoming_webhook_endpoint(
     broadcaster = request.app.state.broadcaster
     presence = request.app.state.presence
     focus_presence = request.app.state.focus_presence
-    await broadcast_new_message(db, broadcaster, presence, focus_presence, room.id, message, sender)
+    global_presence = request.app.state.global_presence
+    await broadcast_new_message(
+        db,
+        broadcaster,
+        presence,
+        focus_presence,
+        global_presence,
+        str(request.base_url),
+        room.id,
+        message,
+        sender,
+    )

backend/app/services/message_events.py

@@ -6,12 +6,14 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.models import Message, MessageFile, MessageMention, Room, RoomMembership, User
 from app.schemas.message import ReactionSummary
+from app.services.email_service import send_email
 from app.services.link_preview_service import fetch_and_broadcast_link_preview
 from app.services.push_service import send_push_to_user
 from app.services.room_service import list_dm_partner_ids
 from app.services.webhook_service import dispatch_event
 from app.ws.broadcaster import Broadcaster
 from app.ws.focus_presence import FocusPresence
+from app.ws.global_presence import GlobalPresence
 from app.ws.presence import Presence
 
 
@@ -148,11 +150,82 @@ def _maybe_fetch_link_preview(broadcaster: Broadcaster, room_id: uuid.UUID, mess
         )
 
 
+async def _maybe_email_dm_notification(
+    db: AsyncSession,
+    global_presence: GlobalPresence,
+    base_url: str,
+    room_id: uuid.UUID,
+    sender: User,
+    message: Message,
+) -> None:
+    """#66: DMs only, deliberately narrower than _notify_offline_members'
+    own "offline" -- that one means "not connected to this room's channel
+    right now," which fires on every message and is fine for a lightweight
+    channel (push/desktop). Email is heavier-weight and a DM's other
+    participant could easily be actively using the app in a different room,
+    so this uses GlobalPresence (genuinely no open connection anywhere)
+    instead -- the same "is this user actually offline" logic as
+    rooms.py's private _member_status (not importable from here), just
+    re-derived.
+    """
+    room = await db.get(Room, room_id)
+    if room is None or not room.is_dm:
+        return
+
+    result = await db.execute(
+        select(RoomMembership).where(
+            RoomMembership.room_id == room_id, RoomMembership.user_id != sender.id
+        )
+    )
+    membership = result.scalar_one_or_none()
+    if membership is None:
+        return
+    recipient = await db.get(User, membership.user_id)
+    if recipient is None:
+        return
+    # appear_offline is a manual "always look offline" override -- treated
+    # the same as genuinely offline here, same as everywhere else it's
+    # checked in this codebase.
+    if not recipient.appear_offline and await global_presence.is_online(recipient.id):
+        return
+
+    # Debounced to the first unread message in this conversation, not
+    # every single one -- a burst of DMs while someone's asleep should be
+    # one email, not one per message.
+    already_unread = await db.execute(
+        select(Message.id)
+        .where(
+            Message.room_id == room_id,
+            Message.id != message.id,
+            Message.created_at > membership.last_read_at,
+        )
+        .limit(1)
+    )
+    if already_unread.scalar_one_or_none() is not None:
+        return
+
+    if message.content:
+        body_line = f"{sender.username}: {message.content[:200]}"
+    elif message.file_id:
+        body_line = f"{sender.username} sent a file"
+    else:
+        body_line = f"{sender.username} sent an image"
+    link = f"{base_url.rstrip('/')}/rooms/{room_id}"
+    await send_email(
+        db,
+        recipient.email,
+        f"New message from {sender.username}",
+        f"{body_line}\n\nView it here:\n{link}",
+    )
+
+
 async def broadcast_new_message(
     db: AsyncSession,
     broadcaster: Broadcaster,
     presence: Presence,
     focus_presence: FocusPresence,
+    global_presence: GlobalPresence,
+    base_url: str,
     room_id: uuid.UUID,
     message: Message,
     sender: User,
@@ -184,6 +257,7 @@ async def broadcast_new_message(
             unhidden_user_id, {"type": "room_added", "room_id": str(room_id)}
         )
     await _notify_offline_members(db, broadcaster, presence, focus_presence, room_id, sender, message)
+    await _maybe_email_dm_notification(db, global_presence, base_url, room_id, sender, message)
     await dispatch_event(db, "message.created", room_id, payload)
     _maybe_fetch_link_preview(broadcaster, room_id, message)
 

backend/app/ws/chat.py

@@ -230,7 +230,15 @@ async def chat_endpoint(websocket: WebSocket, db: AsyncSession = Depends(get_db)
                     # attributed sender may not actually be watching.
                     await mark_room_read(db, envelope.room_id, user.id)
                     await broadcast_new_message(
-                        db, broadcaster, presence, focus_presence, envelope.room_id, message, user
+                        db,
+                        broadcaster,
+                        presence,
+                        focus_presence,
+                        global_presence,
+                        str(websocket.base_url),
+                        envelope.room_id,
+                        message,
+                        user,
                     )
 
                 elif envelope.type == "edit":

backend/tests/test_dm_email_notifications.py

@@ -0,0 +1,204 @@
+import uuid
+
+from app.models import SmtpSettings
+from app.schemas.user import UserCreate
+from app.services.auth_service import register_user
+
+
+def _unique(prefix: str) -> str:
+    return f"{prefix}-{uuid.uuid4().hex[:8]}"
+
+
+def _fake_smtp(monkeypatch):
+    calls = []
+
+    async def fake_send(message, **kwargs):
+        calls.append({"message": message, **kwargs})
+
+    monkeypatch.setattr("app.services.email_service.aiosmtplib.send", fake_send)
+
+    # Monkeypatches get_smtp_settings directly rather than configuring it
+    # for real through the admin endpoint (like test_smtp_settings.py's own
+    # _configure_smtp does) -- ws_client_factory-based tests commit for
+    # real, no rollback, and SmtpSettings is a genuine single global row.
+    # Configuring it for real here previously leaked into every later test
+    # in the same run, breaking test_smtp_settings.py's "starts
+    # unconfigured" assumption. This never touches the DB at all.
+    fake_settings = SmtpSettings(
+        host="smtp.example.com",
+        port=587,
+        username="bot",
+        password_encrypted=None,
+        from_address="noreply@example.com",
+        use_tls=True,
+    )
+
+    async def fake_get_smtp_settings(db):
+        return fake_settings
+
+    monkeypatch.setattr("app.services.email_service.get_smtp_settings", fake_get_smtp_settings)
+    return calls
+
+
+def _register_ws(ws_client, username: str) -> dict:
+    async def _seed():
+        async with ws_client.session_factory() as session:
+            await register_user(
+                session,
+                UserCreate(username=username, email=f"{username}@example.com", password="password123"),
+            )
+
+    ws_client.portal.call(_seed)
+    resp = ws_client.post(
+        "/api/auth/login", json={"username_or_email": username, "password": "password123"}
+    )
+    assert resp.status_code == 200, resp.text
+    return resp.json()
+
+
+def _send_and_sync(ws, room_id: str, content: str) -> dict:
+    """Sync barrier -- see test_mentions.py's identical helper. The
+    "message" ack fires the instant broadcaster.publish() runs, the very
+    first line of broadcast_new_message -- it proves nothing about whether
+    _maybe_email_dm_notification (awaited afterward, in the same handler)
+    has finished. A second, idempotent join's own ack only arrives once
+    the whole prior frame's handling -- including the email step -- is
+    done, since one connection processes frames strictly sequentially."""
+    ws.send_json({"type": "message", "room_id": room_id, "content": content})
+    message = ws.receive_json()
+    ws.send_json({"type": "join", "room_id": room_id})
+    assert ws.receive_json()["type"] == "joined"
+    return message
+
+
+def test_dm_message_emails_globally_offline_recipient(ws_client_factory, monkeypatch):
+    calls = _fake_smtp(monkeypatch)
+    instance1 = ws_client_factory()
+    instance2 = ws_client_factory()
+
+    alice = _register_ws(instance1, _unique("alice"))
+    bob = _register_ws(instance2, _unique("bob"))
+
+    dm = instance1.post("/api/rooms/dm", json={"other_user_id": bob["id"]}).json()
+
+    # bob never connects via WS at all -- genuinely offline, not just
+    # absent from this room's own channel.
+    with instance1.websocket_connect("/ws/chat") as alice_ws:
+        alice_ws.send_json({"type": "join", "room_id": dm["id"]})
+        assert alice_ws.receive_json()["type"] == "joined"
+        _send_and_sync(alice_ws, dm["id"], "hey, you there?")
+
+    assert len(calls) == 1
+    email = calls[0]["message"]
+    assert email["To"] == bob["email"]
+    assert f"New message from {alice['username']}" in email["Subject"]
+    body = email.get_content()
+    assert f"{alice['username']}: hey, you there?" in body
+    assert f"/rooms/{dm['id']}" in body
+
+
+def test_dm_message_does_not_email_online_recipient(ws_client_factory, monkeypatch):
+    calls = _fake_smtp(monkeypatch)
+    instance1 = ws_client_factory()
+    instance2 = ws_client_factory()
+
+    alice = _register_ws(instance1, _unique("alice"))
+    bob = _register_ws(instance2, _unique("bob"))
+
+    dm = instance1.post("/api/rooms/dm", json={"other_user_id": bob["id"]}).json()
+
+    with instance2.websocket_connect("/ws/chat"):
+        # bob has an open connection -- genuinely online -- even though he
+        # never joins the DM's own room channel.
+        with instance1.websocket_connect("/ws/chat") as alice_ws:
+            alice_ws.send_json({"type": "join", "room_id": dm["id"]})
+            assert alice_ws.receive_json()["type"] == "joined"
+            _send_and_sync(alice_ws, dm["id"], "hey")
+
+    assert calls == []
+
+
+def test_dm_message_emails_appear_offline_recipient_even_when_connected(ws_client_factory, monkeypatch):
+    calls = _fake_smtp(monkeypatch)
+    instance1 = ws_client_factory()
+    instance2 = ws_client_factory()
+
+    alice = _register_ws(instance1, _unique("alice"))
+    bob = _register_ws(instance2, _unique("bob"))
+
+    dm = instance1.post("/api/rooms/dm", json={"other_user_id": bob["id"]}).json()
+
+    resp = instance2.patch("/api/auth/me", json={"appear_offline": True})
+    assert resp.status_code == 200
+
+    with instance2.websocket_connect("/ws/chat"):
+        # bob is connected (genuinely online) but lurking -- appear_offline
+        # should still count as "email me," matching how it already
+        # overrides the presence dot everywhere else.
+        with instance1.websocket_connect("/ws/chat") as alice_ws:
+            alice_ws.send_json({"type": "join", "room_id": dm["id"]})
+            assert alice_ws.receive_json()["type"] == "joined"
+            _send_and_sync(alice_ws, dm["id"], "hey")
+
+    assert len(calls) == 1
+    assert calls[0]["message"]["To"] == bob["email"]
+
+
+def test_regular_room_message_does_not_email_offline_member(ws_client_factory, monkeypatch):
+    calls = _fake_smtp(monkeypatch)
+    instance1 = ws_client_factory()
+    instance2 = ws_client_factory()
+
+    _register_ws(instance1, _unique("alice"))
+    room = instance1.post("/api/rooms", json={"name": _unique("general")}).json()
+
+    _register_ws(instance2, _unique("bob"))
+    instance2.post(f"/api/rooms/{room['id']}/join")
+    # bob never connects -- genuinely offline, same as the DM case -- but
+    # this isn't a DM, so #66's email notification is out of scope here.
+
+    with instance1.websocket_connect("/ws/chat") as alice_ws:
+        alice_ws.send_json({"type": "join", "room_id": room["id"]})
+        assert alice_ws.receive_json()["type"] == "joined"
+        _send_and_sync(alice_ws, room["id"], "hello room")
+
+    assert calls == []
+
+
+def test_dm_email_debounced_to_first_unread_then_resets_after_read(ws_client_factory, monkeypatch):
+    calls = _fake_smtp(monkeypatch)
+    instance1 = ws_client_factory()
+    instance2 = ws_client_factory()
+
+    alice = _register_ws(instance1, _unique("alice"))
+    bob = _register_ws(instance2, _unique("bob"))
+
+    dm = instance1.post("/api/rooms/dm", json={"other_user_id": bob["id"]}).json()
+
+    with instance1.websocket_connect("/ws/chat") as alice_ws:
+        alice_ws.send_json({"type": "join", "room_id": dm["id"]})
+        assert alice_ws.receive_json()["type"] == "joined"
+
+        _send_and_sync(alice_ws, dm["id"], "message one")
+        assert len(calls) == 1
+
+        # A second message while bob still hasn't read the first -- no
+        # second email for the same burst.
+        _send_and_sync(alice_ws, dm["id"], "message two")
+        assert len(calls) == 1
+
+    # bob "reads" the conversation via REST -- he never has to have been
+    # connected via WS for this to be meaningful, mark-read is independent
+    # of live connection state.
+    instance2.post(
+        "/api/auth/login", json={"username_or_email": bob["username"], "password": "password123"}
+    )
+    read_resp = instance2.post(f"/api/rooms/{dm['id']}/read")
+    assert read_resp.status_code == 204
+
+    with instance1.websocket_connect("/ws/chat") as alice_ws:
+        alice_ws.send_json({"type": "join", "room_id": dm["id"]})
+        assert alice_ws.receive_json()["type"] == "joined"
+        _send_and_sync(alice_ws, dm["id"], "message three")
+
+    assert len(calls) == 2