Technical Design TDD-0003 — Notification Service¶

1. Summary¶

The platform lacks a centralized mechanism for delivering notifications to users. Individual services (IAM, Grocery, Wallet, Issues) each have events that should trigger user-facing messages (welcome emails, activity alerts, security notices), but no service owns the delivery pipeline.

This document describes the design for loryth-notifications, a new microservice responsible for accepting notification requests (via RabbitMQ events or REST API), rendering templates, dispatching through channel adapters (email, in-app inbox), and tracking delivery lifecycle with retries and idempotency.

Expected impact: any service can trigger a notification with a single RabbitMQ message, without owning email/SMTP logic or delivery reliability concerns.

2. Context and Problem Statement¶

Currently, no service sends transactional emails or in-app messages. Events like account creation, password reset, or issue assignment have no user-facing notification path. Building email delivery into each service would duplicate SMTP configuration, retry logic, and template management across the codebase.

A dedicated notification service centralizes:

• Template rendering and channel dispatch
• Delivery tracking with retries and exponential backoff
• Idempotency to prevent duplicate sends
• An in-app inbox for realtime message delivery

2.1 Goals¶

Functional Requirements¶

• Accept notification requests from any service via RabbitMQ or REST API
• Render file-based templates with per-notification context
• Deliver via email (SMTP) and in-app inbox channels
• Track delivery status (pending, processing, sent, failed, cancelled) with full audit trail
• Deduplicate notifications using idempotency keys
• Retry failed deliveries with exponential backoff
• Provide a paginated inbox with read/unread/archive state
• Stream realtime inbox updates via SSE

Non-functional Requirements¶

• Throughput: <100 notifications/hour (initial target)
• Reliability: no message loss — outbox pattern ensures persistence before acknowledgement
• Observability: structured logging with notification_uuid for end-to-end tracing
• Latency: email delivery within 30 seconds of notification creation under normal conditions

2.2 Out of Scope¶

• Email attachments (adds MIME complexity, deferred to v1.1+)
• Multi-tenancy (matches existing single-tenant platform)
• DB-based template management API (deferred to v1.1)
• SES/SendGrid provider adapters (deferred to v1.1, when provider is chosen)
• Bounce/complaint webhook handling (deferred to v1.1)
• User notification preferences / opt-out (deferred to v1.1)
• Rate limiting (not needed at <100/hour throughput)
• Push notifications (mobile)

2.3 Assumptions and Constraints¶

2.4 Open Questions¶

1. Should the notification service own a copy of user email addresses, or always resolve them at send-time via IAM? (Assumed: resolve at send-time via event payload)
2. Will there be admin APIs to manage templates, channel configs, or retry policies? (Deferred to v1.1)
3. Should bounce/complaint webhooks from future providers (SES, SendGrid) update notification status? (Deferred to v1.1)

3. Design Overview¶

The service has two runtime components:

• API (FastAPI): REST endpoints for creating/querying notifications and inbox management
• Worker (Typer CLI): outbox consumer that picks up pending notifications, renders templates, and dispatches to channel adapters

Both share the domain layer and connect to the same PostgreSQL database.

flowchart TB
    subgraph "Producer Services"
        IAM["loryth-iam"]
        GROCERY["loryth-grocery"]
        WALLET["loryth-wallet"]
        ISSUES["loryth-issues"]
    end

    subgraph "RabbitMQ"
        EX_IAM["iam.events (topic)"]
        EX_GROCERY["grocery.events (topic)"]
        EX_NOTIFY["notifications.inbound (topic)"]
        Q_NOTIFY["notifications.process (queue)"]
        Q_DLQ["notifications.dlq (queue)"]
    end

    subgraph "loryth-notifications"
        subgraph "API (FastAPI)"
            API_CREATE["POST /notification"]
            API_LIST["GET /notification"]
            API_STATUS["GET /notification/{id}"]
            API_INBOX["GET /inbox/*"]
            API_SSE["GET /inbox/stream (SSE)"]
        end

        subgraph "Worker (Typer CLI)"
            CONSUMER["EventConsumer (outbox poller)"]
            HANDLER["NotificationHandler"]
            DISPATCHER["ChannelDispatcher"]
        end

        subgraph "Domain"
            SVC["NotificationService"]
            TMPL["TemplateRenderer"]
            REPO["NotificationRepository"]
            ATTEMPT_REPO["DeliveryAttemptRepository"]
            INBOX_REPO["InboxMessageRepository"]
        end

        subgraph "Channel Adapters"
            EMAIL_ADAPTER["SmtpEmailAdapter"]
            INBOX_ADAPTER["InboxChannelAdapter"]
        end

        DB[(PostgreSQL)]
    end

    subgraph "External"
        SMTP["SMTP Server (MailHog / Postfix / SES)"]
    end

    IAM -->|"AccountCreatedEvent"| EX_IAM
    GROCERY -->|"domain events"| EX_GROCERY
    EX_IAM --> Q_NOTIFY
    EX_GROCERY --> Q_NOTIFY
    EX_NOTIFY --> Q_NOTIFY
    Q_NOTIFY --> HANDLER
    Q_NOTIFY -.->|"reject after max retries"| Q_DLQ

    API_CREATE --> SVC
    API_LIST --> REPO
    API_STATUS --> REPO
    API_INBOX --> INBOX_REPO

    HANDLER --> SVC
    SVC --> TMPL
    SVC --> REPO
    SVC --> DISPATCHER
    DISPATCHER --> EMAIL_ADAPTER
    DISPATCHER --> INBOX_ADAPTER
    EMAIL_ADAPTER --> SMTP
    INBOX_ADAPTER --> INBOX_REPO
    INBOX_ADAPTER --> API_SSE

    SVC --> DB
    CONSUMER --> REPO
    ATTEMPT_REPO --> DB

3.1 Key Components¶

3.2 Architecture¶

Data Model¶

Notification (Aggregate Root)¶

class NotificationStatus(StrEnum):
    PENDING = "pending"
    PROCESSING = "processing"
    SENT = "sent"
    FAILED = "failed"
    CANCELLED = "cancelled"

class NotificationChannel(StrEnum):
    EMAIL = "email"
    IN_APP = "in_app"

class NotificationCategory(StrEnum):
    ACCOUNT = "account"
    SECURITY = "security"
    SYSTEM = "system"
    ACTIVITY = "activity"

class Notification(BaseAggregate):
    __tablename__ = "notification"

    recipient_id: Mapped[UUID] = mapped_uuid("recipient_id")
    recipient_email: Mapped[str | None] = mapped_column("recipient_email", String(255), nullable=True)
    channel: Mapped[NotificationChannel] = mapped_enum("channel", NotificationChannel)
    category: Mapped[NotificationCategory] = mapped_enum("category", NotificationCategory)
    status: Mapped[NotificationStatus] = mapped_enum("status", NotificationStatus)
    template_key: Mapped[str] = mapped_column("template_key", String(128), nullable=False)
    template_context: Mapped[dict[str, Any]] = mapped_column("template_context", JSONB(), nullable=False)
    idempotency_key: Mapped[str | None] = mapped_column("idempotency_key", String(255), nullable=True, unique=True)
    source_event_type: Mapped[str | None] = mapped_column("source_event_type", String(255), nullable=True)
    attempt_count: Mapped[int] = mapped_column("attempt_count", Integer, nullable=False, default=0)
    max_attempts: Mapped[int] = mapped_column("max_attempts", Integer, nullable=False, default=3)
    sent_at: Mapped[datetime | None] = mapped_column("sent_at", DateTime, nullable=True)
    failed_at: Mapped[datetime | None] = mapped_column("failed_at", DateTime, nullable=True)
    next_retry_at: Mapped[datetime | None] = mapped_column("next_retry_at", DateTime, nullable=True)

DeliveryAttempt (Child Entity)¶

class DeliveryStatus(StrEnum):
    SUCCESS = "success"
    FAILED = "failed"

class DeliveryAttempt(AuditMixin, BaseEntity):
    __tablename__ = "delivery_attempt"

    id: Mapped[int] = mapped_primary_key("id")
    notification_id: Mapped[int] = mapped_column(
        "notification_id", BigInteger, ForeignKey("notification.id", ondelete="CASCADE"), nullable=False
    )
    attempt_number: Mapped[int] = mapped_column("attempt_number", Integer, nullable=False)
    status: Mapped[DeliveryStatus] = mapped_enum("status", DeliveryStatus)
    provider_response: Mapped[str | None] = mapped_column("provider_response", Text, nullable=True)
    error_message: Mapped[str | None] = mapped_column("error_message", Text, nullable=True)
    duration_ms: Mapped[int | None] = mapped_column("duration_ms", Integer, nullable=True)

InboxMessage (Aggregate Root — for in-app channel)¶

class InboxMessage(BaseAggregate):
    __tablename__ = "inbox_message"

    notification_id: Mapped[int] = mapped_column(
        "notification_id", BigInteger, ForeignKey("notification.id", ondelete="CASCADE"), nullable=False
    )
    recipient_id: Mapped[UUID] = mapped_uuid("recipient_id")
    title: Mapped[str] = mapped_column("title", String(255), nullable=False)
    body: Mapped[str] = mapped_column("body", Text, nullable=False)
    category: Mapped[str] = mapped_column("category", String(64), nullable=False)
    is_read: Mapped[bool] = mapped_column("is_read", Boolean, nullable=False, default=False)
    read_at: Mapped[datetime | None] = mapped_column("read_at", DateTime, nullable=True)
    archived_at: Mapped[datetime | None] = mapped_column("archived_at", DateTime, nullable=True)

State Machine¶

                    ┌─────────────┐
                    │   PENDING    │ ◄──── (created / retry scheduled)
                    └──────┬──────┘
                           │ worker picks up
                           ▼
                    ┌─────────────┐
                    │ PROCESSING  │
                    └──────┬──────┘
                      ┌────┴────┐
                      │         │
                 success    failure
                      │         │
                      ▼         ▼
               ┌──────────┐  ┌──────────────┐
               │   SENT   │  │ attempt < max │──► back to PENDING
               └──────────┘  │ attempt ≥ max │──► FAILED
                             └──────────────┘
                                               ┌──────────┐
                                               │ CANCELLED │ (admin action)
                                               └──────────┘

Valid transitions:

• PENDING → PROCESSING (worker claims it)
• PROCESSING → SENT (delivery succeeded)
• PROCESSING → PENDING (delivery failed, retries remain)
• PROCESSING → FAILED (delivery failed, max retries exhausted)
• PENDING → CANCELLED (admin cancellation)

3.3 Interface Impact (Before → After)¶

Not applicable — new service addition.

4. Implementation Details¶

4.1 Core Contracts and APIs¶

REST API¶

RabbitMQ Message Schemas¶

Inbound: SendNotificationTask (from any service)

{
  "message_name": "loryth.notifications.domain.notification.events.SendNotificationTask",
  "message_type": "task",
  "recipient_id": "550e8400-e29b-41d4-a716-446655440000",
  "recipient_email": "kostas@example.com",
  "channel": "email",
  "category": "account",
  "template_key": "welcome",
  "template_context": {
    "username": "kostas",
    "app_name": "Loryth"
  },
  "idempotency_key": "iam.account-created.550e8400-e29b-41d4-a716-446655440000"
}

Outbound: NotificationSentEvent

{
  "message_name": "loryth.notifications.domain.notification.events.NotificationSentEvent",
  "message_type": "event",
  "notification_uuid": "7c9e6679-7425-40de-944b-e07fc1f90ae7",
  "recipient_id": "550e8400-e29b-41d4-a716-446655440000",
  "channel": "email",
  "sent_at": "2026-02-09T14:30:00Z"
}

Outbound: NotificationFailedEvent

{
  "message_name": "loryth.notifications.domain.notification.events.NotificationFailedEvent",
  "message_type": "event",
  "notification_uuid": "7c9e6679-7425-40de-944b-e07fc1f90ae7",
  "recipient_id": "550e8400-e29b-41d4-a716-446655440000",
  "reason": "Max retries exceeded: Connection refused",
  "total_attempts": 3
}

Message Definitions (Python)¶

class NotificationCreatedEvent(BaseEvent):
    notification_uuid: UUID
    recipient_id: UUID
    channel: NotificationChannel

class NotificationSentEvent(BaseEvent):
    notification_uuid: UUID
    recipient_id: UUID
    channel: NotificationChannel

class NotificationFailedEvent(BaseEvent):
    notification_uuid: UUID
    recipient_id: UUID
    reason: str

class SendNotificationTask(BaseTask):
    __loryth_task_name__ = "loryth.notifications.send-notification"
    recipient_id: UUID
    recipient_email: str | None = None
    channel: str  # "email" | "in_app"
    category: str  # "account" | "security" | "system" | "activity"
    template_key: str
    template_context: dict[str, Any]
    idempotency_key: str | None = None

4.2 Component Architecture¶

Email Channel¶

Provider Adapter Interface¶

@dataclass(frozen=True)
class EmailMessage:
    to: str
    subject: str
    body_html: str
    body_text: str | None = None
    from_address: str | None = None  # falls back to config default
    reply_to: str | None = None

@dataclass(frozen=True)
class EmailResult:
    success: bool
    provider_message_id: str | None = None
    error: str | None = None

class EmailProvider(Protocol):
    def send(self, message: EmailMessage) -> EmailResult: ...

SMTP Configuration¶

@dataclass
class SmtpConfig:
    host: str = "localhost"
    port: int = 1025
    username: str = ""
    password: str = ""
    use_tls: bool = False
    from_address: str = "noreply@loryth.app"

    @staticmethod
    def from_env() -> "SmtpConfig":
        return SmtpConfig(
            host=os.environ.get("SMTP_HOST", "localhost"),
            port=int(os.environ.get("SMTP_PORT", "1025")),
            username=os.environ.get("SMTP_USERNAME", ""),
            password=os.environ.get("SMTP_PASSWORD", ""),
            use_tls=os.environ.get("SMTP_USE_TLS", "false").lower() == "true",
            from_address=os.environ.get("SMTP_FROM_ADDRESS", "noreply@loryth.app"),
        )

MailHog for Development¶

Add to docker-compose.override.yml:

mailhog:
  image: mailhog/mailhog:v1.0.1
  ports:
    - "1025:1025"   # SMTP
    - "8025:8025"   # Web UI

Service environment:

loryth-notifications-api:
  environment:
    SMTP_HOST: mailhog
    SMTP_PORT: 1025
    SMTP_USE_TLS: "false"
    SMTP_FROM_ADDRESS: "noreply@loryth.local"

Template System¶

File-based templates stored in the service package:

loryth/notifications/templates/
  welcome/
    subject.txt          → "Welcome to $app_name, $username!"
    body.html            → <html>...</html>
    body.txt             → Plain text version
  password_reset/
    subject.txt
    body.html
    body.txt
  email_verification/
    subject.txt
    body.html
    body.txt

@dataclass(frozen=True)
class RenderedMessage:
    subject: str
    body_html: str
    body_text: str | None

@service()
class TemplateRenderer:
    def render(self, template_key: str, context: dict[str, Any]) -> RenderedMessage:
        base_path = importlib.resources.files("loryth.notifications.templates") / template_key
        subject = Template((base_path / "subject.txt").read_text()).safe_substitute(context)
        body_html = Template((base_path / "body.html").read_text()).safe_substitute(context)
        body_txt_path = base_path / "body.txt"
        body_text = Template(body_txt_path.read_text()).safe_substitute(context) if body_txt_path.is_file() else None
        return RenderedMessage(subject=subject, body_html=body_html, body_text=body_text)

Uses Python stdlib string.Template ($variable / ${variable} syntax) — intentionally simple, no external dependency. Upgrade to Jinja2 if needed later.

In-App Inbox Channel¶

• In-app is a delivery channel — the dispatch pipeline treats it like email
• inbox_message table stores the materialized, user-facing message with read/unread/archive state
• Read/unread tracking via is_read boolean + read_at timestamp
• Realtime delivery via SSE (Server-Sent Events) — simpler than WebSockets, built-in reconnection

SSE Design¶

• In-process InboxEventBroker maintains dict[UUID, set[asyncio.Queue]] for fan-out to multiple tabs/devices
• 30-second heartbeat to keep connections alive
• Last-Event-ID support for reconnection catch-up
• SSE event types: new_message, message_read, messages_read_bulk, message_archived, unread_count

Client Sync Strategy¶

1. Initial load: GET /inbox + GET /inbox/unread-count in parallel
2. Realtime: open SSE connection immediately after initial load
3. Reconnection: EventSource auto-reconnects; server replays missed events via Last-Event-ID
4. Visibility change: fetch authoritative unread count when tab becomes visible
5. Badge count: server pushes unread_count event every 60s as drift correction

4.3 Integration Points¶

Idempotency Strategy¶

Database-level deduplication via unique constraint on idempotency_key:

@transactional
def create(self, task: SendNotificationTask) -> NotificationDto:
    if task.idempotency_key:
        existing = self._repository.find_by_idempotency_key(task.idempotency_key)
        if existing is not None:
            logger.info(f"Duplicate notification suppressed: {task.idempotency_key}")
            return NotificationDto.map(existing)

    notification = Notification(
        recipient_id=task.recipient_id,
        recipient_email=task.recipient_email,
        channel=NotificationChannel(task.channel),
        category=NotificationCategory(task.category),
        template_key=task.template_key,
        template_context=task.template_context,
        idempotency_key=task.idempotency_key,
        source_event_type=task.message_name,
    )
    self._repository.insert(notification)
    return NotificationDto.map(notification)

Idempotency key convention: {service}.{event_type}.{entity_uuid} — e.g., iam.account-created.550e8400-...

Processing Algorithm¶

1. Worker receives message from RabbitMQ (or outbox consumer picks up task)
2. Deserialize to SendNotificationTask
3. Check idempotency_key → if duplicate, ACK and skip
4. Create Notification entity (status=PENDING)
5. Persist to database (within @transactional)
6. ACK the RabbitMQ message
7. Delivery loop (same or next worker cycle): a. Query: SELECT * FROM notification WHERE status='pending' AND (next_retry_at IS NULL OR next_retry_at <= NOW()) b. For each notification:
   1. Set status = PROCESSING
   2. Render template (template_key + template_context)
   3. Dispatch to channel adapter (email → SmtpEmailAdapter, in_app → InboxChannelAdapter)
   4. On SUCCESS: create DeliveryAttempt(SUCCESS), notification.record_attempt_success()
   5. On FAILURE: create DeliveryAttempt(FAILED, error), notification.record_attempt_failure()
   6. Flush changes (within transaction)

Cross-Service Integration (Phase 1.2)¶

• IAM worker publishes SendNotificationTask on AccountCreatedEvent
• RabbitMQ bindings: iam.events → notifications.process queue
• Welcome email and email verification templates

4.4 Error Handling and Edge Cases¶

Retry/Backoff + DLQ Policy¶

Failure Modes¶

Recovery sweep (runs on worker startup):

@transactional
def recover_stale_processing(self, stale_threshold_minutes: int = 5) -> int:
    cutoff = datetime.now(timezone.utc) - timedelta(minutes=stale_threshold_minutes)
    stale = self._repository.find_stale_processing(cutoff)
    for notification in stale:
        notification.status = NotificationStatus.PENDING
        logger.warning(f"Recovered stale notification: {notification.uuid}")
    return len(stale)

Gotchas¶

1. Duplicate sends: The biggest risk. Mitigated by idempotency keys + unique constraint, but operators must ensure all producers include meaningful keys.
2. Partial failure: Notification created in DB but email fails → safe because retry picks it up. Email sent but DB commit fails → rare with local PG, but the idempotency key prevents re-creation on retry.
3. SMTP timeouts: Default 30s timeout on smtplib.SMTP. If the provider is slow, this blocks the worker thread. At <100/hour throughput this is acceptable; for higher throughput, use async SMTP (aiosmtplib).
4. Template injection: string.Template.safe_substitute does not execute code, but template context should still be sanitized at the API boundary (Pydantic handles this).
5. MailHog in production: Ensure SMTP_HOST environment variable points to a real SMTP server in production. The dev defaults (localhost:1025) will silently fail if not overridden.
6. Provider rate limits: SES has a default 1 email/second for sandbox accounts. The SMTP adapter should respect rate limits — at <100/hour this is not a concern, but add rate limiting before scaling up.

4.5 Observability¶

Structured Logging¶

All log lines include notification context:

logger.info(
    "Notification created",
    extra={
        "notification_uuid": str(notification.uuid),
        "recipient_id": str(notification.recipient_id),
        "channel": notification.channel,
        "template_key": notification.template_key,
        "idempotency_key": notification.idempotency_key,
    },
)

Key log events:

Metrics (Prometheus-style names)¶

Tracing¶

• RequestIdMiddleware already provides per-request UUID
• Add notification_uuid to all log/metric contexts for end-to-end tracing
• Answer "what happened to notification X?" via: GET /notification/{uuid} → shows status, all delivery attempts with timestamps and errors

Dashboard Suggestions¶

1. Notification Throughput: rate(notifications_created_total[5m]) by channel
2. Delivery Success Rate: notifications_sent_total / notifications_created_total
3. Delivery Latency: p50/p95/p99 of notifications_delivery_duration_seconds
4. Failure Rate: rate(notifications_failed_total[5m]) — alert if > 0 sustained for 5 min
5. Pending Queue Depth: notifications_pending_count — alert if > 50 sustained for 10 min
6. SMTP Errors: rate(notifications_delivery_attempts_total{status="failed"}[5m]) — alert if > 5/min

5. Semantics¶

Notification Lifecycle¶

• A notification is immutable once created — only status, attempt count, and timestamps change
• Template rendering happens at delivery time, not creation time — this allows template fixes to apply to pending retries
• Idempotency keys are globally unique and permanent — they are never recycled or expired
• In-app messages are independent records linked to their parent notification — archiving an inbox message does not affect the notification

Domain Events¶

• NotificationCreatedEvent is recorded on entity creation (for outbox dispatch)
• NotificationSentEvent is recorded on successful delivery (published to notifications.events exchange)
• NotificationFailedEvent is recorded when max retries are exhausted

7. Test Matrix¶

7.1 Unit Coverage (8 tests)¶

7.2 Integration Coverage (10 tests)¶

7.3 E2E Coverage (4 tests)¶

Test Infrastructure¶

• PostgreSQL: testcontainers[postgres] (session-scoped), migrations run once
• MailHog: custom MailHogContainer (session-scoped), API polling for verification
• RabbitMQ: testcontainers[rabbitmq] (session-scoped), per-test exclusive queues
• Cleanup: function-scoped cleanup_db deletes all rows; _clear_mailhog deletes messages

CI Pipeline¶

Stage 1 (parallel):  ruff format --check . | ruff check . | mypy .
Stage 2:             pytest tests/unit/ -v
Stage 3 (Docker):    pytest tests/integration/ tests/e2e/ -v --durations=0

8. Acceptance Criteria¶

• All SQL migrations apply cleanly on fresh database
• ruff check . passes with zero errors
• ruff format . produces no changes
• mypy . passes with zero errors
• All 22 test cases pass (pytest -v)
• Service starts cleanly with fastapi run
• Worker starts cleanly with python -m typer loryth.notifications.worker.cli run
• POST /notification creates notification and triggers email delivery
• Email arrives in MailHog with correct subject/body/recipient
• Duplicate idempotency key returns existing notification (no re-send)
• Failed delivery is retried (verify via delivery_attempt records)
• GET /notification/{id} shows full lifecycle (status, attempts, timestamps)
• Health check returns 200
• Structured logs include notification_uuid for traceability

9. Rollout and Migration¶

9.1 Phased Plan¶

Phase 0: Scaffolding (Small)

• Create loryth-notifications service skeleton (pyproject.toml, config/, domain/, api/, worker/)
• Set up DI container, infrastructure config (InfrastructureConfig)
• Add SQL migrations (eventlog/outbox + notification + delivery_attempt)
• Add Dockerfile, docker-compose entries, MailHog container
• Verify: service starts, health check passes, migrations run

Phase 1: Core Notification Pipeline — v1.0 (Medium)

• Notification entity, repository, DTO, service
• DeliveryAttempt entity and repository
• Template renderer (file-based, string.Template)
• Email provider interface + SMTP adapter
• Channel dispatcher (routes notification to email adapter)
• Worker: outbox consumer + notification handler + delivery loop
• REST API: POST /notification, GET /notification, GET /notification/{id}
• Idempotency (unique constraint + service-level check)
• Retry logic with exponential backoff
• Recovery sweep for stale PROCESSING notifications
• Structured logging throughout
• Tests: all 8 unit + 10 integration + 4 E2E tests
• Verify: end-to-end flow from API/event to email in MailHog

Phase 1.1: In-App Message Center (Medium)

• InboxMessage entity, repository, service
• Inbox channel adapter (creates InboxMessage on dispatch)
• REST API: GET/PUT/DELETE /inbox endpoints
• SSE endpoint with InboxEventBroker
• Additional tests for inbox functionality
• Verify: notifications appear in inbox, realtime SSE works

Phase 1.2: Cross-Service Integration (Small)

• IAM worker: publish SendNotificationTask on AccountCreatedEvent
• Wire RabbitMQ bindings: iam.events → notifications.process queue
• Add welcome email template
• Add email verification template
• E2E test: create account in IAM → notification → email arrives

Phase 2: Production Hardening — v1.1 (Medium, future)

• DLQ monitoring + admin retry API
• Prometheus metrics integration
• SES/SendGrid adapter (when provider is chosen)
• Bounce/complaint webhook handling
• DB-based template management API
• User notification preferences (opt-out per channel/category)
• Rate limiting

9.2 Migration Strategy¶

SQL Migrations¶

Three migration files applied via loryth-cli upgrade:

• v00-setup.sql — Standard eventlog/outbox infrastructure (reused from other services)
• v01-create-notification.sql — notification and delivery_attempt tables with enums and indexes
• v02-create-inbox-message.sql — inbox_message table with indexes

All tables use the shared loryth database (not iam).

10. Risks and Mitigations¶

11. Progress¶

• Phase 0: Scaffolding
• Phase 1: Core Notification Pipeline
• Phase 1.1: In-App Message Center
• Phase 1.2: Cross-Service Integration
• Phase 2: Production Hardening

12. Changelog¶

13. Glossary¶