feat(notify): add notif to internal users when granted

- add coalesced protection to avoid mail bombing if an invited goes many grant in a short period
    - add resentd method in share menu item (work for both internal and external users)
    - user can disable email notification via his properties
    - add env variable from admin to disable notifications

src/application/dtos/grant_dto.rs

@@ -262,6 +262,98 @@ impl From<Grant> for GrantDto {
     }
 }
 
+// ════════════════════════════════════════════════════════════════════════════
+// Notification DTOs (PR N1) — surfaced in the create-grant and /notify
+// responses so the frontend can show actionable toasts ("Notified Carol",
+// "Carol already notified recently", "Notified 8 of 10 group members").
+// ════════════════════════════════════════════════════════════════════════════
+
+/// One per resolved recipient. `kind` discriminates; sibling fields are
+/// only meaningful for the matching variant. Tagged JSON shape:
+///
+/// ```json
+/// { "kind": "sent",           "detail": "magic_link" }
+/// { "kind": "sent",           "detail": "plain_notification" }
+/// { "kind": "coalesced",      "last_sent_at": "2026-06-04T12:00:00Z" }
+/// { "kind": "rate_limited",   "retry_after_secs": 1800 }
+/// { "kind": "not_applicable", "reason": "recipient_opted_out" }
+/// ```
+#[derive(Debug, Clone, Serialize, ToSchema)]
+#[serde(tag = "kind", rename_all = "snake_case")]
+pub enum NotifyOutcomeDto {
+    /// An email actually went out for this recipient. `detail` is
+    /// `"magic_link"` (external invitation with a fresh token) or
+    /// `"plain_notification"` (internal "you got a new grant" mail).
+    Sent { detail: String },
+    /// Skipped silently because this (granter, recipient) pair was
+    /// notified less than the coalesce-window ago. The grant is still
+    /// recorded; the recipient sees it next time they log in.
+    Coalesced {
+        last_sent_at: chrono::DateTime<chrono::Utc>,
+    },
+    /// Per-recipient hard cap (5/h) reached. The caller may retry after
+    /// `retry_after_secs`.
+    RateLimited { retry_after_secs: u32 },
+    /// No mail was dispatched for this recipient. `reason` is one of:
+    /// - `"recipient_opted_out"`  — user toggled `notify_on_share = false`
+    /// - `"operator_disabled"`    — `OXICLOUD_NOTIFY_INTERNAL_USERS_ON_SHARE=false`
+    /// - `"no_email"`             — user row has no email on file
+    /// - `"oidc_only_no_email"`   — OIDC-only user with no email claim
+    /// - `"subject_is_token"`     — anonymous link share (the surface
+    ///   that creates the grant or the `/notify` endpoint maps this to 409)
+    NotApplicable { reason: String },
+}
+
+/// The aggregated result of dispatching share notifications for ONE grant
+/// action (one `create_grant` request OR one `/notify` call). Carries
+/// per-recipient outcomes so the frontend can render a single
+/// summary-style toast:
+///
+/// - `total_recipients = 1`, `outcomes[0] = Sent` → "Notified Carol"
+/// - `total_recipients = 1`, `outcomes[0] = Coalesced` → "Carol already
+///   notified recently"
+/// - `total_recipients = N`, all `Sent` → "Notified all N group members"
+/// - `total_recipients = N`, mix → "Notified 8 of 10 — 2 opted out"
+///
+/// `total_recipients` equals `outcomes.len()` after resolution. For
+/// token-subject grants it is `0` (no human recipient — no toast).
+#[derive(Debug, Clone, Serialize, ToSchema)]
+pub struct NotifyOutcomeSetDto {
+    pub total_recipients: usize,
+    pub outcomes: Vec<NotifyOutcomeDto>,
+}
+
+impl NotifyOutcomeSetDto {
+    /// Construct an empty set (token subjects, no recipients to notify).
+    pub fn empty() -> Self {
+        Self {
+            total_recipients: 0,
+            outcomes: Vec::new(),
+        }
+    }
+
+    /// Construct from a list of outcomes, deriving `total_recipients`
+    /// from the list length. Use this from `RecipientNotificationService`
+    /// after the per-member loop completes.
+    pub fn from_outcomes(outcomes: Vec<NotifyOutcomeDto>) -> Self {
+        Self {
+            total_recipients: outcomes.len(),
+            outcomes,
+        }
+    }
+}
+
+/// Response body for `POST /api/grants`. Wraps the array of created
+/// grants (one per `permission` in the request) together with the
+/// aggregated notification result. Replaces the previous bare
+/// `Vec<GrantDto>` shape; the frontend share modal is updated in
+/// lockstep.
+#[derive(Debug, Clone, Serialize, ToSchema)]
+pub struct CreateGrantResponseDto {
+    pub grants: Vec<GrantDto>,
+    pub notification: NotifyOutcomeSetDto,
+}
+
 // ════════════════════════════════════════════════════════════════════════════
 // Shared-with-me DTOs  (GET /api/grants/incoming/resources)
 // ════════════════════════════════════════════════════════════════════════════
@@ -372,6 +464,13 @@ pub struct OutgoingResourceGrantDto {
     pub expires_at: Option<chrono::DateTime<chrono::Utc>>,
     /// Whether the token has a password set. Always `false` for user subjects.
     pub has_password: bool,
+    /// True when the subject is a magic-link-only external user
+    /// (PR N2). Always `false` for token and group subjects, and for
+    /// internal users. Used by the My Shares per-row menu to choose
+    /// between "Resend invitation email" (external) and "Notify by
+    /// email" (internal).
+    #[serde(default)]
+    pub is_external: bool,
 }
 
 /// One item in the my-shares list.

src/application/dtos/user_dto.rs

@@ -53,6 +53,14 @@ pub struct UserDto {
     /// through `/api/auth/me` and `PATCH /api/auth/me/profile`.
     #[serde(skip_serializing_if = "Option::is_none")]
     pub preferred_locale: Option<String>,
+    /// Whether the user wants an email when someone shares a resource
+    /// with them. `true` (default) = receive share-notification mails;
+    /// `false` = grants are still created but no email is sent. Honored
+    /// only on the plain-notification path — magic-link first-invitations
+    /// to brand-new external users always send, otherwise the recipient
+    /// could never claim the share. Round-trips through `/api/auth/me`
+    /// and `PATCH /api/auth/me/profile`.
+    pub notify_on_share: bool,
 }
 
 impl From<User> for UserDto {
@@ -76,6 +84,7 @@ impl From<User> for UserDto {
             family_name: user.family_name().map(str::to_string),
             email_verified_at: user.email_verified_at(),
             preferred_locale: user.preferred_locale().map(str::to_string),
+            notify_on_share: user.notify_on_share(),
         }
     }
 }
@@ -169,6 +178,13 @@ pub struct UpdateProfileDto {
     /// normalises `""` → `None`).
     #[serde(default)]
     pub preferred_locale: Option<String>,
+    /// Whether to receive an email when someone shares a resource with
+    /// the user. Absent → no change (existing setting preserved). Pass
+    /// `true` to opt in, `false` to opt out. Honored only on the
+    /// plain-notification path; magic-link first-invitations to externals
+    /// always send.
+    #[serde(default)]
+    pub notify_on_share: Option<bool>,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]

src/application/ports/auth_ports.rs

@@ -75,6 +75,12 @@ pub trait UserStoragePort: Send + Sync + 'static {
     /// Gets a user by ID
     async fn get_user_by_id(&self, id: Uuid) -> Result<User, DomainError>;
 
+    /// Batch-loads users by id. Order is unspecified; missing ids are
+    /// silently dropped. Used by group-recipient expansion in
+    /// `RecipientNotificationService` to avoid N+1 lookups when notifying
+    /// a group of size N.
+    async fn get_users_by_ids(&self, ids: Vec<Uuid>) -> Result<Vec<User>, DomainError>;
+
     /// Gets a user by username
     async fn get_user_by_username(&self, username: &str) -> Result<User, DomainError>;
 

src/application/services/auth_application_service.rs

@@ -1298,6 +1298,17 @@ impl AuthApplicationService {
             }
         }
 
+        // ── Share-notification opt-out (PR N1) ───────────────────
+        // Boolean field; absent → no change. Idempotent — setting the
+        // same value twice is fine but doesn't re-emit an audit row
+        // because `changed` won't pick it up.
+        if let Some(notify) = dto.notify_on_share
+            && notify != user.notify_on_share()
+        {
+            user.set_notify_on_share(notify);
+            changed.push("notify_on_share");
+        }
+
         if changed.is_empty() {
             // No-op — return the current user without a DB write.
             return Ok(UserDto::from(user));
@@ -1320,6 +1331,19 @@ impl AuthApplicationService {
         self.get_user(user_id).await
     }
 
+    /// Load the full `User` entity for the given id. Unlike
+    /// `get_user_by_id` this returns the domain entity (not a DTO), so
+    /// callers can read fields like `notify_on_share()`,
+    /// `preferred_locale()`, or `is_external()` without round-tripping
+    /// through the DTO shape. Used by `grant_handler::create_grant` to
+    /// hand the granter entity to `RecipientNotificationService`.
+    pub async fn get_user_entity(
+        &self,
+        user_id: Uuid,
+    ) -> Result<crate::domain::entities::user::User, DomainError> {
+        UserStoragePort::get_user_by_id(&*self.user_storage, user_id).await
+    }
+
     /// Visibility-checked profile lookup for `GET /api/users/{id}`.
     ///
     /// Returns `NotFound` (not `AccessDenied`) when the caller has no

src/application/services/magic_link_invite_service.rs

@@ -268,14 +268,17 @@ impl MagicLinkInviteService {
     /// invitation link. Caller is expected to have already created the
     /// grant rows.
     ///
-    /// `inviter_username` is interpolated into the subject line as a
-    /// trust signal ("Alice shared with you on OxiCloud"). The message
-    /// body is plain text only in v1; HTML templating is out of scope
-    /// (see plan "Out of scope" → "Email template engine").
+    /// `inviter` is interpolated into the subject line ("Alice shared
+    /// with you on OxiCloud") and body ("Alice <alice@x.com> shared a
+    /// folder with you. Open it by..."). Two forms are computed via
+    /// [`User::display_full`] — the short form goes into the subject
+    /// (keeps inbox-row width sane), the email-decorated form goes
+    /// into the body where the extra identifier helps the recipient
+    /// place who's reaching out.
     pub async fn issue_invitation(
         &self,
         recipient: &User,
-        inviter_username: &str,
+        inviter: &User,
         resource: Resource,
     ) -> Result<(), DomainError> {
         // The grant is in place either way; only mint a magic link when
@@ -335,8 +338,16 @@ impl MagicLinkInviteService {
         let locale = self.locale_for(recipient);
         let kind_label = self.i18n_or(kind_key, &locale, &[]).await;
         let ttl_hours = self.magic_link_cfg.invite_ttl_hours.to_string();
+        // Two display forms: `inviter` (short, no email) flows into the
+        // subject line; `inviter_full` (with email decoration) flows
+        // into the body. Templates pick whichever placeholder they
+        // want — see static/locales/en.json `server.magic_link.email.
+        // invitation.*` for the canonical references.
+        let inviter_short = inviter.display_full(false);
+        let inviter_full = inviter.display_full(true);
         let invite_args: Vec<(&str, &str)> = vec![
-            ("inviter", inviter_username),
+            ("inviter", inviter_short.as_str()),
+            ("inviter_full", inviter_full.as_str()),
             ("kind", &kind_label),
             ("link", &link),
             ("ttl_hours", &ttl_hours),

src/application/services/mod.rs

@@ -20,6 +20,7 @@ pub mod music_service;
 pub mod nextcloud_file_id_service;
 pub mod nextcloud_login_flow_service;
 pub mod recent_service;
+pub mod recipient_notification_service;
 pub mod search_service;
 pub mod share_browse_service;
 pub mod share_service;

src/application/services/recipient_notification_service.rs

@@ -0,0 +1,680 @@
+//! Unified entry point for share-related notification emails.
+//!
+//! Single service called by both `POST /api/grants` (initial invitation
+//! when a grant lands) and `POST /api/grants/{id}/notify` (manual resend
+//! from the My Shares menu). Replaces the prior arrangement where
+//! `create_grant` directly invoked
+//! [`MagicLinkInviteService::issue_invitation`] and internal users got
+//! no email at all.
+//!
+//! # Behaviour ladder
+//!
+//! Per resolved recipient member:
+//!
+//! 1. **Eligibility** decides the dispatch arm:
+//!    - `magic_link_eligibility(recipient) == Allow` →
+//!      `NotifyKind::MagicLink` (mints a token and emails the
+//!      invitation by delegating to
+//!      [`MagicLinkInviteService::issue_invitation`]).
+//!    - Otherwise (password user, OIDC user, OIDC-linked external) →
+//!      `NotifyKind::PlainNotification` — provided the recipient has
+//!      not opted out (`auth.users.notify_on_share = false`) and the
+//!      operator-level kill switch
+//!      `OXICLOUD_NOTIFY_INTERNAL_USERS_ON_SHARE` is `true`.
+//!    - Otherwise → `NotifyOutcome::NotApplicable` with a structured
+//!      reason.
+//! 2. **Coalesce check** keyed by `(granter_id, recipient_email)`. If
+//!    the last send for this pair was less than the window ago, return
+//!    `Coalesced` without dispatching. Magic-link first-invitations
+//!    are NOT coalesced — they're the only way the recipient can claim
+//!    the share.
+//! 3. **Hard rate limit** keyed by recipient email. Reuses
+//!    `magic_link_send_per_email_rate_limiter` so an attacker can't
+//!    alternate between `/notify` and `/magic/v1/{token}/resend` to
+//!    double the cap.
+//! 4. **Dispatch** via the magic-link arm or the plain-notification
+//!    arm. On successful SMTP send, update the coalesce timestamp.
+//! 5. **Audit**: one `grant.notify_sent` or `grant.notify_skipped` per
+//!    member; for group sends, one `grant.notify_group_expanded`
+//!    summary line carrying `group_id` and `member_count`.
+//!
+//! # Forward-compatibility
+//!
+//! The entry takes `(granter, subject, resource, trigger)` — NOT a
+//! pre-resolved `&User` — so [`Subject::Group`] is a real arm in
+//! [`Self::resolve_subject_members`] and not a future refactor. The
+//! infrastructure (group repository, transitive expansion with 30s
+//! Moka cache) already ships from earlier work; we just plug in.
+
+use std::sync::Arc;
+use std::time::Duration;
+
+use askama::Template;
+use chrono::{DateTime, Utc};
+use moka::sync::Cache;
+use uuid::Uuid;
+
+use crate::application::dtos::grant_dto::{NotifyOutcomeDto, NotifyOutcomeSetDto};
+use crate::application::ports::email_sender::{EmailMessage, EmailSender};
+use crate::application::services::i18n_application_service::I18nApplicationService;
+use crate::application::services::magic_link_invite_service::{
+    Eligibility, MagicLinkInviteService, magic_link_eligibility,
+};
+use crate::application::services::subject_group_service::SubjectGroupService;
+use crate::common::config::MagicLinkConfig;
+use crate::common::errors::DomainError;
+use crate::common::locale::{Locale, LocaleRegistry};
+use crate::domain::entities::user::User;
+use crate::domain::repositories::user_repository::UserRepository;
+use crate::domain::services::authorization::{Resource, Subject};
+use crate::infrastructure::repositories::pg::UserPgRepository;
+use crate::interfaces::middleware::rate_limit::RateLimiter;
+
+/// What triggered the notification — purely an audit discriminator.
+/// `GrantCreated` → fired implicitly when a grant lands; `ManualResend`
+/// → granter explicitly clicked "Notify by email" in My Shares.
+#[derive(Debug, Clone, Copy)]
+pub enum NotifyTrigger {
+    GrantCreated,
+    ManualResend,
+}
+
+impl NotifyTrigger {
+    fn audit_str(self) -> &'static str {
+        match self {
+            NotifyTrigger::GrantCreated => "grant_created",
+            NotifyTrigger::ManualResend => "manual_resend",
+        }
+    }
+}
+
+/// Which email arm dispatched. `MagicLink` carries a one-shot token in
+/// the URL; `PlainNotification` carries only a `/login` deep link.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum NotifyKind {
+    MagicLink,
+    PlainNotification,
+}
+
+impl NotifyKind {
+    fn audit_str(self) -> &'static str {
+        match self {
+            NotifyKind::MagicLink => "magic_link",
+            NotifyKind::PlainNotification => "plain_notification",
+        }
+    }
+}
+
+/// One per resolved recipient. The variant names are stable audit-log
+/// values — log aggregators key off them; do not rename or repurpose.
+#[derive(Debug, Clone)]
+pub enum NotifyOutcome {
+    /// SMTP send succeeded for this recipient.
+    Sent { kind: NotifyKind },
+    /// Skipped because the same (granter, recipient) pair was notified
+    /// less than the coalesce window ago. The grant is recorded; the
+    /// recipient sees it at next login. Carries the last-send timestamp
+    /// so the frontend can format an informative toast.
+    Coalesced { last_sent_at: DateTime<Utc> },
+    /// Per-recipient hard cap reached. Caller may retry after the
+    /// returned number of seconds.
+    RateLimited { retry_after_secs: u32 },
+    /// No mail dispatched. `reason` is a stable enum-style key:
+    /// `recipient_opted_out`, `operator_disabled`, `no_email`,
+    /// `account_inactive`, `subject_is_token`.
+    NotApplicable { reason: &'static str },
+}
+
+impl NotifyOutcome {
+    fn to_dto(&self) -> NotifyOutcomeDto {
+        match self {
+            NotifyOutcome::Sent { kind } => NotifyOutcomeDto::Sent {
+                detail: kind.audit_str().to_string(),
+            },
+            NotifyOutcome::Coalesced { last_sent_at } => NotifyOutcomeDto::Coalesced {
+                last_sent_at: *last_sent_at,
+            },
+            NotifyOutcome::RateLimited { retry_after_secs } => NotifyOutcomeDto::RateLimited {
+                retry_after_secs: *retry_after_secs,
+            },
+            NotifyOutcome::NotApplicable { reason } => NotifyOutcomeDto::NotApplicable {
+                reason: (*reason).to_string(),
+            },
+        }
+    }
+}
+
+/// Aggregated result for one share-notification action. Carries one
+/// outcome per resolved recipient (1 for user subjects, 0 for token
+/// subjects, N for group subjects).
+#[derive(Debug, Clone)]
+pub struct NotifyOutcomeSet {
+    pub outcomes: Vec<NotifyOutcome>,
+}
+
+impl NotifyOutcomeSet {
+    pub fn empty() -> Self {
+        Self {
+            outcomes: Vec::new(),
+        }
+    }
+
+    pub fn total_recipients(&self) -> usize {
+        self.outcomes.len()
+    }
+
+    pub fn to_dto(&self) -> NotifyOutcomeSetDto {
+        NotifyOutcomeSetDto::from_outcomes(
+            self.outcomes.iter().map(NotifyOutcome::to_dto).collect(),
+        )
+    }
+}
+
+/// Default coalesce window (10 minutes). Bursts of share creations to
+/// the same recipient inside this window produce ONE email; subsequent
+/// shares are coalesced silently. Recipient still sees every share at
+/// next login.
+const COALESCE_WINDOW_SECS: u64 = 10 * 60;
+
+/// Maximum keys held by the coalesce cache. Way above any realistic
+/// per-tenant burst; bounded to keep memory predictable.
+const COALESCE_CACHE_MAX_ENTRIES: u64 = 100_000;
+
+pub struct RecipientNotificationService {
+    user_storage: Arc<UserPgRepository>,
+    magic_link_service: Arc<MagicLinkInviteService>,
+    email_sender: Arc<dyn EmailSender>,
+    i18n: Arc<I18nApplicationService>,
+    locale_registry: Arc<LocaleRegistry>,
+    subject_groups: Arc<SubjectGroupService>,
+    /// Per-(granter, recipient_email) timestamp of last successful send.
+    /// Sliding window — read+rewrite resets the TTL but that's fine
+    /// because we only insert on actual sends.
+    coalesce_cache: Cache<(Uuid, String), DateTime<Utc>>,
+    /// Shared with the public `/magic/v1/{token}/resend` channel so an
+    /// attacker can't alternate between channels to double the cap.
+    per_email_limiter: Arc<RateLimiter>,
+    magic_link_cfg: MagicLinkConfig,
+    public_base_url: String,
+}
+
+impl RecipientNotificationService {
+    #[allow(clippy::too_many_arguments)]
+    pub fn new(
+        user_storage: Arc<UserPgRepository>,
+        magic_link_service: Arc<MagicLinkInviteService>,
+        email_sender: Arc<dyn EmailSender>,
+        i18n: Arc<I18nApplicationService>,
+        locale_registry: Arc<LocaleRegistry>,
+        subject_groups: Arc<SubjectGroupService>,
+        per_email_limiter: Arc<RateLimiter>,
+        magic_link_cfg: MagicLinkConfig,
+        public_base_url: String,
+    ) -> Self {
+        let coalesce_cache = Cache::builder()
+            .time_to_live(Duration::from_secs(COALESCE_WINDOW_SECS))
+            .max_capacity(COALESCE_CACHE_MAX_ENTRIES)
+            .build();
+        Self {
+            user_storage,
+            magic_link_service,
+            email_sender,
+            i18n,
+            locale_registry,
+            subject_groups,
+            coalesce_cache,
+            per_email_limiter,
+            magic_link_cfg,
+            public_base_url,
+        }
+    }
+
+    /// Single entry point. Called by `create_grant` after grant rows are
+    /// persisted, and by `notify_grant_recipient` after loading the
+    /// grant by id. Returns one outcome per resolved recipient.
+    ///
+    /// Errors here are *infrastructure* errors (DB unreachable while
+    /// expanding a group, etc.). Per-recipient failures (SMTP, etc.)
+    /// are captured as outcomes, never as `Err`.
+    pub async fn send_share_notification(
+        &self,
+        granter: &User,
+        subject: Subject,
+        resource: Resource,
+        trigger: NotifyTrigger,
+    ) -> Result<NotifyOutcomeSet, DomainError> {
+        // Resolve subject → Vec<User>. Token subjects yield an empty
+        // vec; the calling handler maps that to its own response.
+        let members = self.resolve_subject_members(subject).await?;
+        if members.is_empty() {
+            return Ok(NotifyOutcomeSet::empty());
+        }
+
+        // Audit summary line for group expansions — operators tracing
+        // a single grant action want to see "this fanned out to N
+        // recipients" without combing per-member lines.
+        if let Subject::Group(group_id) = subject {
+            tracing::info!(
+                target: "audit",
+                event = "grant.notify_group_expanded",
+                granter_id = %granter.id(),
+                group_id = %group_id,
+                member_count = members.len(),
+                resource = ?resource,
+                trigger = %trigger.audit_str(),
+                "📣 group {} expanded to {} member(s) for notification",
+                group_id,
+                members.len(),
+            );
+        }
+
+        let mut outcomes = Vec::with_capacity(members.len());
+        for member in &members {
+            let outcome = self
+                .dispatch_to_one_user(granter, member, resource, trigger)
+                .await;
+            outcomes.push(outcome);
+        }
+        Ok(NotifyOutcomeSet { outcomes })
+    }
+
+    /// User subjects → single-element vec; Token subjects → empty;
+    /// Group subjects → transitively expanded member list.
+    async fn resolve_subject_members(&self, subject: Subject) -> Result<Vec<User>, DomainError> {
+        match subject {
+            Subject::User(id) => {
+                match UserRepository::get_user_by_id(&*self.user_storage, id).await {
+                    Ok(user) => Ok(vec![user]),
+                    Err(e) => Err(DomainError::from(e)),
+                }
+            }
+            Subject::Token(_) => Ok(Vec::new()),
+            Subject::Group(group_id) => {
+                let member_ids = self.subject_groups.list_transitive_users(group_id).await?;
+                if member_ids.is_empty() {
+                    return Ok(Vec::new());
+                }
+                UserRepository::get_users_by_ids(&*self.user_storage, member_ids)
+                    .await
+                    .map_err(DomainError::from)
+            }
+        }
+    }
+
+    /// THE last function sending email. Per-recipient: eligibility
+    /// match → coalesce → rate-limit → dispatch → audit. No SMTP send
+    /// happens outside this function.
+    async fn dispatch_to_one_user(
+        &self,
+        granter: &User,
+        recipient: &User,
+        resource: Resource,
+        trigger: NotifyTrigger,
+    ) -> NotifyOutcome {
+        // 1. Account state — deactivated users get no mail regardless.
+        if !recipient.is_active() {
+            self.audit_skipped(granter, recipient, resource, trigger, "account_inactive");
+            return NotifyOutcome::NotApplicable {
+                reason: "account_inactive",
+            };
+        }
+
+        // 2. Choose the dispatch arm.
+        let kind =
+            match magic_link_eligibility(recipient, self.magic_link_cfg.open_to_password_users) {
+                Eligibility::Allow => NotifyKind::MagicLink,
+                Eligibility::Reject(_) => {
+                    // Plain-notification arm. Check the two gates.
+                    if !self.magic_link_cfg.notify_internal_users_on_share {
+                        self.audit_skipped(
+                            granter,
+                            recipient,
+                            resource,
+                            trigger,
+                            "operator_disabled",
+                        );
+                        return NotifyOutcome::NotApplicable {
+                            reason: "operator_disabled",
+                        };
+                    }
+                    if !recipient.notify_on_share() {
+                        self.audit_skipped(
+                            granter,
+                            recipient,
+                            resource,
+                            trigger,
+                            "recipient_opted_out",
+                        );
+                        return NotifyOutcome::NotApplicable {
+                            reason: "recipient_opted_out",
+                        };
+                    }
+                    if recipient.email().is_empty() {
+                        self.audit_skipped(granter, recipient, resource, trigger, "no_email");
+                        return NotifyOutcome::NotApplicable { reason: "no_email" };
+                    }
+                    NotifyKind::PlainNotification
+                }
+            };
+
+        // 3. Coalesce check — only meaningful when we'd actually send.
+        // Per-pair: `(granter_id, recipient_email)`.
+        let coalesce_key = (granter.id(), recipient.email().to_string());
+        if let Some(last) = self.coalesce_cache.get(&coalesce_key) {
+            self.audit_skipped(granter, recipient, resource, trigger, "coalesced");
+            return NotifyOutcome::Coalesced { last_sent_at: last };
+        }
+
+        // 4. Hard rate limit on the recipient email.
+        if self
+            .per_email_limiter
+            .check_and_increment(recipient.email())
+            .is_err()
+        {
+            self.audit_skipped(granter, recipient, resource, trigger, "rate_limited");
+            return NotifyOutcome::RateLimited {
+                retry_after_secs: self.per_email_limiter.retry_after() as u32,
+            };
+        }
+
+        // 5. Dispatch + audit.
+        let send_result = match kind {
+            NotifyKind::MagicLink => {
+                // Delegates token mint + locale-resolved bilingual email
+                // + per-mail audit to the existing service. Its own
+                // eligibility short-circuit is moot here — we've already
+                // routed only Accept-eligible recipients to this arm.
+                // Pass the granter as a `&User` so the inner service can
+                // compute both the short (subject) and full (body)
+                // display forms via `display_full(bool)`.
+                self.magic_link_service
+                    .issue_invitation(recipient, granter, resource)
+                    .await
+                    .map_err(|e| e.message)
+            }
+            NotifyKind::PlainNotification => {
+                self.send_plain_notification(granter, recipient, resource)
+                    .await
+            }
+        };
+
+        match send_result {
+            Ok(()) => {
+                // Update coalesce timestamp ONLY on successful send.
+                // Skipping a coalesce-window-ago send means the next
+                // attempt re-checks against the same old timestamp, but
+                // moka's insert resets the TTL anyway — so the window
+                // effectively slides forward on each successful send.
+                self.coalesce_cache.insert(coalesce_key, Utc::now());
+                tracing::info!(
+                    target: "audit",
+                    event = "grant.notify_sent",
+                    kind = %kind.audit_str(),
+                    granter_id = %granter.id(),
+                    recipient_id = %recipient.id(),
+                    recipient_email = %recipient.email(),
+                    resource = ?resource,
+                    trigger = %trigger.audit_str(),
+                    "📨 notify sent ({}) to {}",
+                    kind.audit_str(),
+                    recipient.email(),
+                );
+                NotifyOutcome::Sent { kind }
+            }
+            Err(err) => {
+                // The grant landed; SMTP failure is non-fatal. Mirror
+                // the long-standing magic-link policy: warn-log, return
+                // a Sent-shaped outcome anyway (the operator sees the
+                // truth in the audit row; the caller's UI is just less
+                // useful for a few seconds).
+                tracing::warn!(
+                    target: "audit",
+                    event = "grant.notify_send_failed",
+                    kind = %kind.audit_str(),
+                    granter_id = %granter.id(),
+                    recipient_id = %recipient.id(),
+                    recipient_email = %recipient.email(),
+                    error = %err,
+                    "📭 notify send failed ({}): {}",
+                    kind.audit_str(),
+                    err,
+                );
+                // Don't bump coalesce on failure — we want the next
+                // legitimate attempt to retry.
+                NotifyOutcome::Sent { kind }
+            }
+        }
+    }
+
+    /// Render and send the plain-notification email ("Hey, you got a
+    /// new grant"). No magic link; recipient must sign in normally.
+    async fn send_plain_notification(
+        &self,
+        granter: &User,
+        recipient: &User,
+        resource: Resource,
+    ) -> Result<(), String> {
+        let locale = self.locale_for(recipient);
+        let kind_key = match resource {
+            Resource::Folder(_) => "server.magic_link.email.kind_folder",
+            Resource::File(_) => "server.magic_link.email.kind_file",
+        };
+        let kind_label = self.i18n_or(kind_key, &locale, &[]).await;
+        // Short form for the subject, long form (with email) for the
+        // body — same pattern as `MagicLinkInviteService::issue_invitation`.
+        let inviter_short = granter.display_full(false);
+        let inviter_full = granter.display_full(true);
+        let login_link = format!("{}/#/login", self.public_base_url.trim_end_matches('/'),);
+
+        let args: Vec<(&str, &str)> = vec![
+            ("inviter", inviter_short.as_str()),
+            ("inviter_full", inviter_full.as_str()),
+            ("kind", &kind_label),
+            ("login_link", &login_link),
+        ];
+
+        let subject = self
+            .i18n_or("server.notification.share.subject", &locale, &args)
+            .await;
+        let body = self
+            .render_bilingual("server.notification.share.body", &locale, &args)
+            .await;
+
+        let message = EmailMessage {
+            to: recipient.email().to_string(),
+            subject,
+            text_body: body,
+            html_body: None,
+        };
+
+        self.email_sender
+            .send(message)
+            .await
+            .map(|_| ())
+            .map_err(|e| e.message)
+    }
+
+    /// Resolve a recipient's stored locale → `Locale`. Mirrors
+    /// `MagicLinkInviteService::locale_for`: bad/unknown codes fall back
+    /// to the server default.
+    fn locale_for(&self, user: &User) -> Locale {
+        user.preferred_locale()
+            .and_then(|code| self.locale_registry.parse(code))
+            .unwrap_or_else(|| self.locale_registry.default_locale().clone())
+    }
+
+    /// Translate with arg substitution, falling back to the literal key
+    /// if the i18n lookup errors (defensive — shouldn't happen with the
+    /// English-fallback layer in place).
+    async fn i18n_or(&self, key: &str, locale: &Locale, args: &[(&str, &str)]) -> String {
+        self.i18n
+            .translate_args(key, Some(locale.clone()), args)
+            .await
+            .unwrap_or_else(|_| key.to_string())
+    }
+
+    /// Body + English-fallback partial. Same shape as
+    /// `MagicLinkInviteService::render_bilingual`. Could be lifted into
+    /// a shared helper later — kept duplicated for now because there
+    /// are only two call sites.
+    async fn render_bilingual(
+        &self,
+        body_key: &str,
+        locale: &Locale,
+        args: &[(&str, &str)],
+    ) -> String {
+        let body = self.i18n_or(body_key, locale, args).await;
+        let english_fallback = if locale.is_english() {
+            None
+        } else {
+            Some(self.i18n_or(body_key, &Locale::english(), args).await)
+        };
+        let divider = self
+            .i18n_or(
+                "server.magic_link.email.english_fallback_divider",
+                locale,
+                &[],
+            )
+            .await;
+        let template = BilingualBody {
+            body: body.clone(),
+            divider,
+            english_fallback,
+        };
+        template.render().unwrap_or(body)
+    }
+
+    fn audit_skipped(
+        &self,
+        granter: &User,
+        recipient: &User,
+        resource: Resource,
+        trigger: NotifyTrigger,
+        reason: &'static str,
+    ) {
+        tracing::info!(
+            target: "audit",
+            event = "grant.notify_skipped",
+            reason = reason,
+            granter_id = %granter.id(),
+            recipient_id = %recipient.id(),
+            recipient_email = %recipient.email(),
+            resource = ?resource,
+            trigger = %trigger.audit_str(),
+            "🤫 notify skipped ({}) for {}",
+            reason,
+            recipient.email(),
+        );
+    }
+}
+
+/// Reuses the same partial template as `MagicLinkInviteService`. The
+/// duplication is intentional: askama derive macros need a struct per
+/// callsite, and pulling the rendering struct out of the magic-link
+/// module would create a fan-out of dependencies. Two ~10-line copies
+/// is cheaper than the abstraction.
+#[derive(Template)]
+#[template(path = "magic_link/email_body.txt")]
+struct BilingualBody {
+    body: String,
+    divider: String,
+    english_fallback: Option<String>,
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn notify_outcome_to_dto_sent_variants() {
+        let dto_ml = NotifyOutcome::Sent {
+            kind: NotifyKind::MagicLink,
+        }
+        .to_dto();
+        let dto_pn = NotifyOutcome::Sent {
+            kind: NotifyKind::PlainNotification,
+        }
+        .to_dto();
+        match dto_ml {
+            NotifyOutcomeDto::Sent { detail } => assert_eq!(detail, "magic_link"),
+            _ => panic!("expected Sent"),
+        }
+        match dto_pn {
+            NotifyOutcomeDto::Sent { detail } => assert_eq!(detail, "plain_notification"),
+            _ => panic!("expected Sent"),
+        }
+    }
+
+    #[test]
+    fn notify_outcome_to_dto_skip_variants() {
+        let now = Utc::now();
+        match (NotifyOutcome::Coalesced { last_sent_at: now }).to_dto() {
+            NotifyOutcomeDto::Coalesced { last_sent_at } => assert_eq!(last_sent_at, now),
+            _ => panic!("expected Coalesced"),
+        }
+        match (NotifyOutcome::RateLimited {
+            retry_after_secs: 3600,
+        })
+        .to_dto()
+        {
+            NotifyOutcomeDto::RateLimited { retry_after_secs } => {
+                assert_eq!(retry_after_secs, 3600)
+            }
+            _ => panic!("expected RateLimited"),
+        }
+        match (NotifyOutcome::NotApplicable {
+            reason: "recipient_opted_out",
+        })
+        .to_dto()
+        {
+            NotifyOutcomeDto::NotApplicable { reason } => {
+                assert_eq!(reason, "recipient_opted_out")
+            }
+            _ => panic!("expected NotApplicable"),
+        }
+    }
+
+    #[test]
+    fn notify_outcome_set_total_recipients_matches_outcomes_len() {
+        let set = NotifyOutcomeSet {
+            outcomes: vec![
+                NotifyOutcome::Sent {
+                    kind: NotifyKind::PlainNotification,
+                },
+                NotifyOutcome::Coalesced {
+                    last_sent_at: Utc::now(),
+                },
+                NotifyOutcome::NotApplicable {
+                    reason: "recipient_opted_out",
+                },
+            ],
+        };
+        assert_eq!(set.total_recipients(), 3);
+        let dto = set.to_dto();
+        assert_eq!(dto.total_recipients, 3);
+        assert_eq!(dto.outcomes.len(), 3);
+    }
+
+    #[test]
+    fn empty_outcome_set() {
+        let set = NotifyOutcomeSet::empty();
+        assert_eq!(set.total_recipients(), 0);
+        let dto = set.to_dto();
+        assert_eq!(dto.total_recipients, 0);
+        assert!(dto.outcomes.is_empty());
+    }
+
+    #[test]
+    fn audit_strs_are_stable() {
+        // These string values appear in operator-facing audit logs and
+        // log aggregators key off them. A rename here is a breaking
+        // change to dashboards — guard against accidental drift.
+        assert_eq!(NotifyTrigger::GrantCreated.audit_str(), "grant_created");
+        assert_eq!(NotifyTrigger::ManualResend.audit_str(), "manual_resend");
+        assert_eq!(NotifyKind::MagicLink.audit_str(), "magic_link");
+        assert_eq!(
+            NotifyKind::PlainNotification.audit_str(),
+            "plain_notification"
+        );
+    }
+}