Add per-room opt-in email notifications on mention (#67)

Lets a member subscribe to email when they're @mentioned in a room
while offline, alongside #66's always-on DM email. Debounced the same
way #66 is (one email per unread burst, not one per mention), and
deliberately scoped to regular rooms only -- DMs already have #66's
automatic offline email with no separate toggle needed.

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

backend/alembic/versions/05b28b0a2261_room_email_notifications_opt_in.py

@@ -0,0 +1,32 @@
+"""room email notifications opt-in
+
+Revision ID: 05b28b0a2261
+Revises: e2fc4d65f93e
+Create Date: 2026-08-28 19:34:23.507200
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = '05b28b0a2261'
+down_revision: Union[str, Sequence[str], None] = 'e2fc4d65f93e'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column('room_memberships', sa.Column('email_notifications', sa.Boolean(), server_default='false', nullable=False))
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column('room_memberships', 'email_notifications')
+    # ### end Alembic commands ###

backend/app/models/membership.py

@@ -2,7 +2,7 @@ import enum
 import uuid
 from datetime import datetime
 
-from sqlalchemy import DateTime, Enum, ForeignKey, PrimaryKeyConstraint, func
+from sqlalchemy import Boolean, DateTime, Enum, ForeignKey, PrimaryKeyConstraint, func
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from app.models.base import Base
@@ -42,6 +42,13 @@ class RoomMembership(Base):
     # arrives in the room, or when find_or_create_dm resolves back to it --
     # both count as the conversation being active again.
     hidden_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    # #67: opt-in, per-member -- email when a message in this room mentions
+    # them while they're offline. Deliberately separate from #66's DM
+    # emails (always-on, no toggle) rather than a shared flag, since DMs
+    # are explicitly out of scope for this setting (see message_events.py).
+    email_notifications: Mapped[bool] = mapped_column(
+        Boolean, default=False, server_default="false", nullable=False
+    )
 
     room = relationship("Room", back_populates="memberships")
     user = relationship("User")

backend/app/routers/rooms.py

@@ -25,6 +25,7 @@ from app.schemas.room import (
     RoomMemberAdd,
     RoomMemberRead,
     RoomMemberRoleUpdate,
+    RoomNotificationSettingsUpdate,
     RoomRead,
     RoomUpdate,
     StartDmRequest,
@@ -72,6 +73,7 @@ from app.services.room_service import (
     list_room_members,
     mark_room_read,
     remove_member,
+    set_room_email_notifications,
     transfer_ownership,
     update_room,
 )
@@ -179,6 +181,7 @@ async def list_my_rooms_endpoint(
             role=role,
             has_unread=has_unread,
             has_mention=has_mention,
+            email_notifications=email_notifications,
             dm_partner=(
                 DmPartnerInfo(
                     user_id=partner.id,
@@ -191,7 +194,7 @@ async def list_my_rooms_endpoint(
                 else None
             ),
         )
-        for room, role, has_unread, has_mention, partner in rooms
+        for room, role, has_unread, has_mention, email_notifications, partner in rooms
     ]
 
 
@@ -293,6 +296,20 @@ async def mark_room_read_endpoint(
     await mark_room_read(db, room_id, current_user.id)
 
 
+@router.patch("/{room_id}/notifications", status_code=204)
+async def update_room_notifications_endpoint(
+    room_id: uuid.UUID,
+    data: RoomNotificationSettingsUpdate,
+    current_user: User = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db),
+):
+    await require_room_member(room_id, current_user, db)
+    try:
+        await set_room_email_notifications(db, room_id, current_user.id, data.email_notifications)
+    except CannotModifyDmError:
+        raise HTTPException(status_code=400, detail="Email notifications aren't available for DMs")
+
+
 def _member_status(user: User, online_ids: set[uuid.UUID]) -> str:
     # appear_offline always wins, regardless of actual connection -- that's
     # the whole point of the override (lurking in a room undetected).

backend/app/schemas/room.py

@@ -63,12 +63,20 @@ class MyRoomItem(RoomRead):
     # avatar, not this room's internal `name`) without a second fetch per
     # row. None for a regular room.
     dm_partner: DmPartnerInfo | None = None
+    # #67: this viewer's own opt-in for "email me when mentioned here while
+    # offline" -- always false for a DM (see message_events.py, deliberately
+    # out of scope; DMs already get #66's automatic offline email).
+    email_notifications: bool = False
 
 
 class StartDmRequest(BaseModel):
     other_user_id: uuid.UUID
 
 
+class RoomNotificationSettingsUpdate(BaseModel):
+    email_notifications: bool
+
+
 class RoomMemberRead(BaseModel):
     user_id: uuid.UUID
     username: str

backend/app/services/message_events.py

@@ -249,6 +249,79 @@ async def _maybe_email_dm_notification(
     )
 
 
+async def _maybe_email_room_mention_notifications(
+    db: AsyncSession,
+    global_presence: GlobalPresence,
+    base_url: str,
+    room_id: uuid.UUID,
+    sender: User,
+    message: Message,
+) -> None:
+    """#67: opt-in, per-room email on mention -- deliberately scoped to
+    regular rooms only (see set_room_email_notifications: DMs already get
+    #66's always-on offline email, no separate toggle). Debounced the same
+    shape as #66 -- only the first unread mention in this room emails, not
+    one per mention while the recipient is away.
+    """
+    room = await db.get(Room, room_id)
+    if room is None or room.is_dm:
+        return
+
+    result = await db.execute(
+        select(MessageMention.user_id).where(MessageMention.message_id == message.id)
+    )
+    mentioned_ids = {row[0] for row in result.all()} - {sender.id}
+    if not mentioned_ids:
+        return
+
+    for user_id in mentioned_ids:
+        membership_result = await db.execute(
+            select(RoomMembership).where(
+                RoomMembership.room_id == room_id, RoomMembership.user_id == user_id
+            )
+        )
+        membership = membership_result.scalar_one_or_none()
+        if membership is None or not membership.email_notifications:
+            continue
+
+        recipient = await db.get(User, user_id)
+        if recipient is None:
+            continue
+        # appear_offline always wins here too, same as _maybe_email_dm_notification.
+        if not recipient.appear_offline and await global_presence.is_online(recipient.id):
+            continue
+
+        already_unread_mention = await db.execute(
+            select(Message.id)
+            .join(MessageMention, MessageMention.message_id == Message.id)
+            .where(
+                MessageMention.user_id == user_id,
+                Message.room_id == room_id,
+                Message.id != message.id,
+                Message.created_at > membership.last_read_at,
+            )
+            .limit(1)
+        )
+        if already_unread_mention.scalar_one_or_none() is not None:
+            continue
+
+        body_line = (
+            f"{sender.username} mentioned you: {message.content[:200]}"
+            if message.content
+            else f"{sender.username} mentioned you"
+        )
+        link = f"{base_url.rstrip('/')}/rooms/{room_id}"
+        await send_email(
+            db,
+            recipient.email,
+            f"New mention in #{room.name}",
+            [body_line],
+            cta_label="Open room",
+            cta_url=link,
+            theme_user=recipient,
+        )
+
+
 async def broadcast_new_message(
     db: AsyncSession,
     broadcaster: Broadcaster,
@@ -288,6 +361,7 @@ async def broadcast_new_message(
         )
     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 _maybe_email_room_mention_notifications(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/services/room_service.py

@@ -182,7 +182,7 @@ async def list_open_rooms(db: AsyncSession, user_id: uuid.UUID) -> list[tuple[Ro
 
 async def list_member_rooms(
     db: AsyncSession, user_id: uuid.UUID
-) -> list[tuple[Room, RoomRole, bool, bool, User | None]]:
+) -> list[tuple[Room, RoomRole, bool, bool, bool, User | None]]:
     last_message_at = (
         select(func.max(Message.created_at))
         .where(Message.room_id == Room.id)
@@ -205,7 +205,12 @@ async def list_member_rooms(
     )
     result = await db.execute(
         select(
-            Room, RoomMembership.role, RoomMembership.last_read_at, last_message_at, has_unread_mention
+            Room,
+            RoomMembership.role,
+            RoomMembership.last_read_at,
+            last_message_at,
+            has_unread_mention,
+            RoomMembership.email_notifications,
         )
         .join(RoomMembership, RoomMembership.room_id == Room.id)
         .where(RoomMembership.user_id == user_id, RoomMembership.hidden_at.is_(None))
@@ -238,9 +243,10 @@ async def list_member_rooms(
             role,
             last_message_at is not None and last_message_at > last_read_at,
             has_mention,
+            email_notifications,
             partners_by_room.get(room.id),
         )
-        for room, role, last_read_at, last_message_at, has_mention in rows
+        for room, role, last_read_at, last_message_at, has_mention, email_notifications in rows
     ]
 
 
@@ -486,6 +492,22 @@ async def mark_room_read(db: AsyncSession, room_id: uuid.UUID, user_id: uuid.UUI
     await db.commit()
 
 
+async def set_room_email_notifications(
+    db: AsyncSession, room_id: uuid.UUID, user_id: uuid.UUID, enabled: bool
+) -> None:
+    """#67: DMs are deliberately excluded -- they already get #66's
+    automatic offline email with no opt-in needed, and this setting only
+    makes sense for a regular room's mention-based notifications."""
+    room = await db.get(Room, room_id)
+    if room is None:
+        raise RoomNotFoundError()
+    if room.is_dm:
+        raise CannotModifyDmError()
+    membership = await _get_membership(db, room_id, user_id)
+    membership.email_notifications = enabled
+    await db.commit()
+
+
 async def leave_room(db: AsyncSession, room_id: uuid.UUID, user_id: uuid.UUID) -> None:
     membership = await _get_membership(db, room_id, user_id)
     if membership.role == RoomRole.owner:

backend/tests/test_room_mention_email_notifications.py

@@ -0,0 +1,225 @@
+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)
+
+    # See test_dm_email_notifications.py's identical helper for why this
+    # monkeypatches get_smtp_settings directly instead of configuring a real
+    # row through the admin endpoint.
+    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:
+    """See test_dm_email_notifications.py's identical helper -- the
+    "message" ack alone proves nothing about whether the email step
+    (awaited afterward in the same handler) has finished; a second,
+    idempotent join's ack only arrives once the whole frame is done."""
+    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 _subscribe(ws_client, room_id: str, username: str, password: str = "password123") -> None:
+    login = ws_client.post(
+        "/api/auth/login", json={"username_or_email": username, "password": password}
+    )
+    assert login.status_code == 200, login.text
+    resp = ws_client.patch(f"/api/rooms/{room_id}/notifications", json={"email_notifications": True})
+    assert resp.status_code == 204, resp.text
+
+
+def test_room_mention_emails_offline_subscribed_member(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"))
+    room = instance1.post("/api/rooms", json={"name": _unique("general")}).json()
+    instance2.post(f"/api/rooms/{room['id']}/join")
+    _subscribe(instance2, room["id"], bob["username"])
+    # bob never connects via WS -- genuinely offline.
+
+    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"], f"hey @{bob['username']}, look at this")
+
+    assert len(calls) == 1
+    email = calls[0]["message"]
+    assert email["To"] == bob["email"]
+    assert f"New mention in #{room['name']}" in email["Subject"]
+    body = email.get_body(preferencelist=("plain",)).get_content()
+    assert alice["username"] in body
+    assert f"/rooms/{room['id']}" in body
+
+
+def test_room_message_without_mention_does_not_email_subscribed_member(ws_client_factory, monkeypatch):
+    calls = _fake_smtp(monkeypatch)
+    instance1 = ws_client_factory()
+    instance2 = ws_client_factory()
+
+    _register_ws(instance1, _unique("alice"))
+    bob = _register_ws(instance2, _unique("bob"))
+    room = instance1.post("/api/rooms", json={"name": _unique("general")}).json()
+    instance2.post(f"/api/rooms/{room['id']}/join")
+    _subscribe(instance2, room["id"], bob["username"])
+
+    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, no mention here")
+
+    assert calls == []
+
+
+def test_room_mention_does_not_email_unsubscribed_member(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"))
+    room = instance1.post("/api/rooms", json={"name": _unique("general")}).json()
+    instance2.post(f"/api/rooms/{room['id']}/join")
+    # bob never opts in -- email_notifications defaults to False.
+
+    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"], f"hey @{bob['username']}")
+
+    assert calls == []
+
+
+def test_room_mention_does_not_email_online_subscribed_member(ws_client_factory, monkeypatch):
+    calls = _fake_smtp(monkeypatch)
+    instance1 = ws_client_factory()
+    instance2 = ws_client_factory()
+
+    _register_ws(instance1, _unique("alice"))
+    bob = _register_ws(instance2, _unique("bob"))
+    room = instance1.post("/api/rooms", json={"name": _unique("general")}).json()
+    instance2.post(f"/api/rooms/{room['id']}/join")
+    _subscribe(instance2, room["id"], bob["username"])
+
+    with instance2.websocket_connect("/ws/chat"):
+        # bob has an open connection -- genuinely online -- even though he
+        # never joins this room's own channel.
+        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"], f"hey @{bob['username']}")
+
+    assert calls == []
+
+
+def test_room_mention_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()
+
+    _register_ws(instance1, _unique("alice"))
+    bob = _register_ws(instance2, _unique("bob"))
+    room = instance1.post("/api/rooms", json={"name": _unique("general")}).json()
+    instance2.post(f"/api/rooms/{room['id']}/join")
+    _subscribe(instance2, room["id"], bob["username"])
+
+    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"], f"@{bob['username']} first mention")
+        assert len(calls) == 1
+
+        # A second mention while bob still hasn't read the first -- no
+        # second email for the same burst.
+        _send_and_sync(alice_ws, room["id"], f"@{bob['username']} second mention")
+        assert len(calls) == 1
+
+    instance2.post(
+        "/api/auth/login", json={"username_or_email": bob["username"], "password": "password123"}
+    )
+    read_resp = instance2.post(f"/api/rooms/{room['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": room["id"]})
+        assert alice_ws.receive_json()["type"] == "joined"
+        _send_and_sync(alice_ws, room["id"], f"@{bob['username']} third mention")
+
+    assert len(calls) == 2
+
+
+def test_enabling_notifications_rejected_for_dm(ws_client_factory):
+    instance1 = ws_client_factory()
+    instance2 = ws_client_factory()
+
+    _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 = instance1.patch(f"/api/rooms/{dm['id']}/notifications", json={"email_notifications": True})
+    assert resp.status_code == 400
+
+
+def test_notifications_setting_reflected_in_my_rooms(ws_client_factory):
+    instance1 = ws_client_factory()
+    _register_ws(instance1, _unique("alice"))
+    room = instance1.post("/api/rooms", json={"name": _unique("general")}).json()
+
+    rooms = instance1.get("/api/rooms/mine").json()
+    mine = next(r for r in rooms if r["id"] == room["id"])
+    assert mine["email_notifications"] is False
+
+    resp = instance1.patch(f"/api/rooms/{room['id']}/notifications", json={"email_notifications": True})
+    assert resp.status_code == 204
+
+    rooms = instance1.get("/api/rooms/mine").json()
+    mine = next(r for r in rooms if r["id"] == room["id"])
+    assert mine["email_notifications"] is True