lukasandClaude Opus 5.5
0ddc0cb001
Branding: the page titles and app name are now "Ultimo Email". Admin panel (now two columns with a divider): - Users: search by login or by a mail account's address or name (results only, nothing listed until something is typed). A user opens a window to change their quota, block/unblock them, or delete them; block and delete ask for confirmation, and an admin cannot do either to their own account. - Blocking (users.blocked_at, migration 0019) refuses sign-in with a clear message and revokes every session; a new user.signed_out event sends the user's open tabs to the sign-in screen at once. Any "not signed in" answer now does the same, not only /auth/me. Deleting removes mail accounts (sync stops), plugin rows (cleared inside the user's RLS context, since plugin tables have no FK to users), blobs, the search index, sessions and the user. - Outgoing system mail (migration 0020): SMTP server, port, security, login, password (vault-encrypted, never sent back), From address and the public address reset links point to, plus a "Send test" that reports the server's error. Password reset: - "Forgot password?" on the sign-in page asks for the user name and always answers "If this account exists you will receive an email to your recovery email address"; the mail goes out after the response, so neither the reply nor its timing reveals an account, and it is capped at 3 per hour. - The emailed link (token in the URL fragment, single use, 30 minutes) opens a page to set a new password. With 2-step verification on, the code is checked before anything changes; five wrong codes burn the link. - Settings > Profile: the recovery email can be added, changed or removed, confirmed with the current password, which must be typed — autofill and paste are refused (TypedPasswordInput). Also adds frontend/icons/ultimo-logo-bimi.svg, an SVG Tiny PS version of the logo for a BIMI record. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
97 lines
3.4 KiB
Python
97 lines
3.4 KiB
Python
# INTERFACE VERSION: 1
|
|
"""EventBus — Redis pub/sub fan-out + in-process durable consumers (SPEC §5.3, §8.3).
|
|
|
|
``publish`` fans out to the Redis channel ``evt:{event}`` (the WS hub relays these
|
|
to the owning user) and invokes any in-process subscribers. Payloads always
|
|
include ``user_id`` (SPEC §5.3). Only the event names in SPEC §8.3 are allowed.
|
|
|
|
NOTE(spec): the frozen contract also enqueues a durable arq job ``on_{event}`` for
|
|
out-of-process consumers (plugins). In M0 the only durable consumers (search index,
|
|
thread denormalization) run inline in the indexer per FR-30, so cross-process arq
|
|
fan-out is deferred to the plugin milestone (M3). The signatures here are frozen.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
from collections.abc import Awaitable, Callable
|
|
from typing import Protocol
|
|
|
|
from app.redis_client import get_redis
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
Handler = Callable[[dict[str, object]], Awaitable[None]]
|
|
|
|
# The complete, allowed event catalog (SPEC §8.3). Plugin events (plugin.{name}.*)
|
|
# are validated separately by the plugin loader (M3).
|
|
ALLOWED_EVENTS: frozenset[str] = frozenset(
|
|
{
|
|
"message.new",
|
|
"message.changed",
|
|
"message.deleted",
|
|
"message.sent",
|
|
"flags.changed",
|
|
"thread.updated",
|
|
"sync.progress",
|
|
"account.state_changed",
|
|
"draft.saved",
|
|
"outbox.state_changed",
|
|
# Sync v2 §3: published on login and every WS connect; drives tier
|
|
# promotion in the sync workers (§5). Payload: {user_id}.
|
|
"user.active",
|
|
# The user's sessions were revoked by an administrator (blocked or
|
|
# deleted): open tabs drop to the sign-in screen. Payload: {user_id}.
|
|
"user.signed_out",
|
|
}
|
|
)
|
|
|
|
|
|
def channel_for(event: str) -> str:
|
|
"""Return the Redis pub/sub channel name for an event."""
|
|
return f"evt:{event}"
|
|
|
|
|
|
class EventBusProtocol(Protocol):
|
|
async def publish(self, event: str, payload: dict[str, object]) -> None: ...
|
|
def subscribe(self, event: str, handler: Handler) -> None: ...
|
|
|
|
|
|
class EventBus:
|
|
"""Concrete EventBus backed by Redis pub/sub with in-process subscribers."""
|
|
|
|
def __init__(self) -> None:
|
|
self._handlers: dict[str, list[Handler]] = {}
|
|
|
|
def subscribe(self, event: str, handler: Handler) -> None:
|
|
"""Register an in-process durable consumer for ``event``."""
|
|
self._handlers.setdefault(event, []).append(handler)
|
|
|
|
async def publish(self, event: str, payload: dict[str, object]) -> None:
|
|
"""Fan out ``event`` to Redis subscribers and in-process handlers."""
|
|
if event not in ALLOWED_EVENTS and not event.startswith("plugin."):
|
|
raise ValueError(f"unknown event {event!r} (see SPEC §8.3)")
|
|
if "user_id" not in payload:
|
|
raise ValueError("event payload must include user_id (SPEC §5.3)")
|
|
|
|
message = json.dumps({"event": event, "payload": payload}, default=str)
|
|
await get_redis().publish(channel_for(event), message)
|
|
|
|
for handler in self._handlers.get(event, ()):
|
|
try:
|
|
await handler(payload)
|
|
except Exception: # a failing consumer must not break the publisher
|
|
logger.exception("event handler failed for %s", event)
|
|
|
|
|
|
_bus: EventBus | None = None
|
|
|
|
|
|
def get_event_bus() -> EventBus:
|
|
"""Return the process-wide EventBus singleton."""
|
|
global _bus
|
|
if _bus is None:
|
|
_bus = EventBus()
|
|
return _bus
|