feat(server): refactor mail queue (#15204)

packages/backend/native/src/content_policy.rs

@@ -0,0 +1,333 @@
+use napi_derive::napi;
+use unicode_normalization::UnicodeNormalization;
+use unicode_skeleton::UnicodeSkeleton;
+
+const VERSION: u32 = 1;
+
+#[napi(object)]
+pub struct ContentPolicyScanInput {
+  pub value: String,
+  pub checks: Option<Vec<String>>,
+}
+
+#[napi(object)]
+pub struct ContentPolicyMatchSpan {
+  pub start: u32,
+  pub end: u32,
+}
+
+#[napi(object)]
+pub struct ContentPolicyMatch {
+  #[napi(js_name = "type")]
+  pub match_type: String,
+  pub reason: String,
+  pub value: Option<String>,
+  pub span: Option<ContentPolicyMatchSpan>,
+}
+
+#[napi(object)]
+pub struct ContentPolicyScanResult {
+  pub version: u32,
+  pub original: String,
+  pub normalized: String,
+  pub skeleton: String,
+  pub matched: bool,
+  pub matches: Vec<ContentPolicyMatch>,
+  pub flags: Vec<String>,
+}
+
+#[napi]
+pub fn scan_content_policy_v1(input: ContentPolicyScanInput) -> ContentPolicyScanResult {
+  let original = input.value;
+  let normalized = normalize_content(&original);
+  let skeleton = skeleton_content(&normalized);
+  let mut flags = Vec::new();
+  if normalized != original {
+    flags.push("unicode_normalized".to_string());
+  }
+  if skeleton != normalized {
+    flags.push("confusable_skeleton_changed".to_string());
+  }
+
+  let mut matches = Vec::new();
+  if should_run_check(input.checks.as_deref(), "url_or_domain") {
+    matches.extend(scan_url_or_domain(&skeleton));
+  }
+
+  ContentPolicyScanResult {
+    version: VERSION,
+    original,
+    normalized,
+    skeleton,
+    matched: !matches.is_empty(),
+    matches,
+    flags,
+  }
+}
+
+fn should_run_check(checks: Option<&[String]>, check: &str) -> bool {
+  checks
+    .map(|checks| checks.iter().any(|configured| configured == check))
+    .unwrap_or(true)
+}
+
+fn normalize_content(value: &str) -> String {
+  value.nfkc().collect::<String>()
+}
+
+fn skeleton_content(value: &str) -> String {
+  let mut output = String::with_capacity(value.len());
+  for ch in value.chars() {
+    push_skeleton_char(&mut output, ch);
+  }
+  output
+}
+
+fn push_skeleton_char(output: &mut String, ch: char) {
+  if is_default_ignorable(ch) {
+    return;
+  }
+  if let Some(replacement) = url_punctuation_ascii(ch).or_else(|| enclosed_ascii(ch)) {
+    output.push_str(replacement);
+    return;
+  }
+  if ch.is_ascii() {
+    output.extend(ch.to_lowercase());
+    return;
+  }
+
+  let source = ch.to_string();
+  let skeleton = source.skeleton_chars().collect::<String>();
+  if skeleton == source {
+    output.extend(ch.to_lowercase());
+    return;
+  }
+
+  for skeleton_ch in skeleton.chars() {
+    push_skeleton_char(output, skeleton_ch);
+  }
+}
+
+fn is_default_ignorable(ch: char) -> bool {
+  matches!(
+    ch,
+    '\u{00ad}'
+      | '\u{034f}'
+      | '\u{061c}'
+      | '\u{180e}'
+      | '\u{200b}'..='\u{200f}'
+      | '\u{202a}'..='\u{202e}'
+      | '\u{2060}'..='\u{206f}'
+      | '\u{fe00}'..='\u{fe0f}'
+      | '\u{feff}'
+      | '\u{e0100}'..='\u{e01ef}'
+  )
+}
+
+fn enclosed_ascii(ch: char) -> Option<&'static str> {
+  let code = ch as u32;
+  if (0x1f150..=0x1f169).contains(&code) {
+    return ASCII_LOWERCASE.get((code - 0x1f150) as usize).copied();
+  }
+  if ch == '⓿' || ch == '🄌' {
+    return Some("0");
+  }
+  for (start, digits) in [
+    (0x2776, ASCII_DIGITS_1_TO_9),
+    (0x2780, ASCII_DIGITS_1_TO_9),
+    (0x278a, ASCII_DIGITS_1_TO_9),
+  ] {
+    if (start..=start + 8).contains(&code) {
+      return digits.get((code - start) as usize).copied();
+    }
+  }
+  None
+}
+
+const ASCII_LOWERCASE: [&str; 26] = [
+  "a", "b", "c", "d", "e", "f", "g", "h", "i", "j", "k", "l", "m", "n", "o", "p", "q", "r", "s", "t", "u", "v", "w",
+  "x", "y", "z",
+];
+
+const ASCII_DIGITS_1_TO_9: [&str; 9] = ["1", "2", "3", "4", "5", "6", "7", "8", "9"];
+
+fn url_punctuation_ascii(ch: char) -> Option<&'static str> {
+  match ch {
+    // URL punctuation variants that are not always useful in generic skeletons.
+    '。' | '．' | '｡' | '․' | '‧' | '∙' | '●' | '•' => Some("."),
+    '⁄' | '∕' | '╱' | '／' => Some("/"),
+    '：' | '꞉' | 'ː' | '։' => Some(":"),
+    '‐' | '‑' | '‒' | '–' | '—' | '―' | '−' | '﹘' | '－' => Some("-"),
+    '_' | '＿' => Some("_"),
+    _ => None,
+  }
+}
+
+fn scan_url_or_domain(value: &str) -> Vec<ContentPolicyMatch> {
+  let mut matches = Vec::new();
+  let chars = value.char_indices().collect::<Vec<_>>();
+  let mut index = 0;
+  while index < chars.len() {
+    let start = chars[index].0;
+    let remaining = &value[start..];
+    if remaining.starts_with("http://") || remaining.starts_with("https://") || remaining.starts_with("www.") {
+      let end = consume_non_space(value, start);
+      matches.push(match_result(value, start, end));
+      index = char_index_after(&chars, end);
+      continue;
+    }
+
+    if is_domain_boundary_before(value, start) {
+      let end = consume_domain(value, start);
+      if end > start && is_domain_candidate(&value[start..end]) && is_domain_boundary_after(value, end) {
+        matches.push(match_result(value, start, end));
+        index = char_index_after(&chars, end);
+        continue;
+      }
+    }
+    index += 1;
+  }
+  matches
+}
+
+fn match_result(value: &str, start: usize, end: usize) -> ContentPolicyMatch {
+  ContentPolicyMatch {
+    match_type: "url_or_domain".to_string(),
+    reason: "contains_url_or_domain".to_string(),
+    value: Some(value[start..end].to_string()),
+    span: Some(ContentPolicyMatchSpan {
+      start: start as u32,
+      end: end as u32,
+    }),
+  }
+}
+
+fn consume_non_space(value: &str, start: usize) -> usize {
+  value[start..]
+    .char_indices()
+    .find_map(|(offset, ch)| ch.is_whitespace().then_some(start + offset))
+    .unwrap_or(value.len())
+}
+
+fn consume_domain(value: &str, start: usize) -> usize {
+  value[start..]
+    .char_indices()
+    .find_map(|(offset, ch)| (!is_domain_char(ch)).then_some(start + offset))
+    .unwrap_or(value.len())
+}
+
+fn is_domain_candidate(candidate: &str) -> bool {
+  let labels = candidate.split('.').collect::<Vec<_>>();
+  if labels.len() < 2 {
+    return false;
+  }
+  let Some(tld) = labels.last() else {
+    return false;
+  };
+  if !(2..=63).contains(&tld.len()) || !tld.chars().all(|ch| ch.is_ascii_alphabetic()) {
+    return false;
+  }
+  labels.iter().all(|label| {
+    !label.is_empty()
+      && label.len() <= 63
+      && !label.starts_with('-')
+      && !label.ends_with('-')
+      && label.chars().all(|ch| ch.is_ascii_alphanumeric() || ch == '-')
+  })
+}
+
+fn is_domain_char(ch: char) -> bool {
+  ch.is_ascii_alphanumeric() || ch == '-' || ch == '.'
+}
+
+fn is_domain_boundary_before(value: &str, start: usize) -> bool {
+  if start == 0 {
+    return true;
+  }
+  let Some(previous) = value[..start].chars().next_back() else {
+    return true;
+  };
+  !(previous.is_ascii_alphanumeric() || previous == '-' || previous == '_' || previous == '@')
+}
+
+fn is_domain_boundary_after(value: &str, end: usize) -> bool {
+  if end >= value.len() {
+    return true;
+  }
+  let Some(next) = value[end..].chars().next() else {
+    return true;
+  };
+  !(next.is_ascii_alphanumeric() || next == '-' || next == '_' || next == '.')
+}
+
+fn char_index_after(chars: &[(usize, char)], byte_index: usize) -> usize {
+  chars
+    .iter()
+    .position(|(offset, _)| *offset >= byte_index)
+    .unwrap_or(chars.len())
+}
+
+#[cfg(test)]
+mod tests {
+  use super::*;
+
+  fn scan(value: &str) -> ContentPolicyScanResult {
+    scan_content_policy_v1(ContentPolicyScanInput {
+      value: value.to_string(),
+      checks: None,
+    })
+  }
+
+  #[test]
+  fn scans_url_or_domain_cases() {
+    struct Case {
+      input: &'static str,
+      matched: bool,
+      skeleton: &'static str,
+      value: Option<&'static str>,
+    }
+
+    for case in [
+      Case {
+        input: "https://ｅxample．com",
+        matched: true,
+        skeleton: "https://example.com",
+        value: Some("https://example.com"),
+      },
+      Case {
+        input: "join 🅠⓿❶.example",
+        matched: true,
+        skeleton: "join q01.example",
+        value: Some("q01.example"),
+      },
+      Case {
+        input: "exa\u{200b}mple.com",
+        matched: true,
+        skeleton: "example.com",
+        value: Some("example.com"),
+      },
+      Case {
+        input: "hello workspace",
+        matched: false,
+        skeleton: "hello workspace",
+        value: None,
+      },
+      Case {
+        input: "user@example.com",
+        matched: false,
+        skeleton: "user@example.com",
+        value: None,
+      },
+    ] {
+      let result = scan(case.input);
+      assert_eq!(result.matched, case.matched, "{}", case.input);
+      assert_eq!(result.skeleton, case.skeleton, "{}", case.input);
+      assert_eq!(
+        result.matches.first().and_then(|matched| matched.value.as_deref()),
+        case.value,
+        "{}",
+        case.input
+      );
+    }
+  }
+}

packages/backend/native/src/lib.rs

@@ -2,6 +2,7 @@
 
 mod utils;
 
+pub mod content_policy;
 pub mod doc;
 pub mod doc_loader;
 pub mod entitlement;

packages/backend/native/src/runtime/backend_runtime/mod.rs

@@ -4,6 +4,7 @@ mod doc_compactor;
 mod doc_storage;
 mod gate;
 mod housekeeping;
+mod rolling_quota;
 mod runtime_state;
 #[cfg(test)]
 mod tests;
@@ -17,8 +18,9 @@ use tokio::sync::Mutex;
 
 use self::types::BackendRuntimeHealth;
 pub(crate) use super::types;
-use super::{
-  BackendRuntimeConfig, RuntimeError, RuntimeResult, migrations::migrate_runtime_tables, napi_error, to_napi_error,
+pub(super) use super::{
+  BackendRuntimeConfig, InviteQuotaConfig, RuntimeError, RuntimeResult, migrations::migrate_runtime_tables, napi_error,
+  to_napi_error,
 };
 
 pub(super) fn token_hash(token: &str) -> String {

packages/backend/native/src/runtime/backend_runtime/rolling_quota/invite_abuse_actions.rs

@@ -0,0 +1,261 @@
+use napi::Result;
+use sqlx::{PgPool, Row};
+
+use super::{
+  BackendRuntime, RuntimeError, RuntimeInviteAbuseClaimedAction, RuntimeResult, napi_error, workspace_subject_key,
+};
+
+async fn invite_abuse_user_quarantined_or_banned(pool: &PgPool, user_id: &str) -> RuntimeResult<bool> {
+  let row: Option<i32> = sqlx::query_scalar(
+    r#"
+    SELECT 1
+    FROM runtime_invite_abuse_subjects
+    WHERE user_id = $1
+      AND status IN ('quarantined', 'banned')
+    LIMIT 1
+    "#,
+  )
+  .bind(user_id)
+  .fetch_optional(pool)
+  .await
+  .map_err(|err| RuntimeError::database("failed to load invite abuse subject by user", err))?;
+  Ok(row.is_some())
+}
+
+async fn invite_abuse_workspace_quarantined(pool: &PgPool, workspace_id: &str) -> RuntimeResult<bool> {
+  let row: Option<i32> = sqlx::query_scalar(
+    r#"
+    SELECT 1
+    FROM runtime_invite_abuse_subjects
+    WHERE subject_key = $1
+      AND status = 'quarantined'
+    LIMIT 1
+    "#,
+  )
+  .bind(workspace_subject_key(workspace_id))
+  .fetch_optional(pool)
+  .await
+  .map_err(|err| RuntimeError::database("failed to load invite abuse workspace subject", err))?;
+  Ok(row.is_some())
+}
+
+async fn claim_invite_abuse_action(pool: &PgPool, action_id: &str, worker_id: &str) -> RuntimeResult<bool> {
+  let result = sqlx::query(
+    r#"
+    UPDATE runtime_invite_abuse_actions action
+    SET status = 'running',
+        attempts = attempts + 1,
+        locked_by = $2,
+        locked_until = now() + interval '5 minutes',
+        last_error = NULL,
+        updated_at = now()
+    WHERE action.id = $1::bigint
+      AND action.action IN ('ban_actor', 'quarantine_actor', 'quarantine_workspace', 'quarantine_source_cohort')
+      AND (
+        (
+          action.status IN ('pending', 'retry_wait')
+          AND (action.next_attempt_at IS NULL OR action.next_attempt_at <= now())
+        )
+        OR (
+          action.status = 'running'
+          AND action.locked_until IS NOT NULL
+          AND action.locked_until <= now()
+        )
+      )
+    "#,
+  )
+  .bind(action_id)
+  .bind(worker_id)
+  .execute(pool)
+  .await
+  .map_err(|err| RuntimeError::database("failed to claim invite abuse action", err))?;
+  Ok(result.rows_affected() > 0)
+}
+
+async fn claim_retryable_invite_abuse_actions(
+  pool: &PgPool,
+  worker_id: &str,
+  limit: i64,
+) -> RuntimeResult<Vec<RuntimeInviteAbuseClaimedAction>> {
+  let rows = sqlx::query(
+    r#"
+    WITH candidates AS (
+      SELECT action.id
+      FROM runtime_invite_abuse_actions action
+      JOIN runtime_invite_abuse_evidence evidence
+        ON evidence.id = action.evidence_id
+      WHERE (
+          (
+            action.status IN ('pending', 'retry_wait')
+            AND (action.next_attempt_at IS NULL OR action.next_attempt_at <= now())
+          )
+          OR (
+            action.status = 'running'
+            AND action.locked_until IS NOT NULL
+            AND action.locked_until <= now()
+          )
+        )
+        AND evidence.user_id IS NOT NULL
+        AND evidence.workspace_id IS NOT NULL
+        AND action.action IN ('ban_actor', 'quarantine_actor', 'quarantine_workspace', 'quarantine_source_cohort')
+      ORDER BY COALESCE(action.next_attempt_at, action.created_at), action.id
+      LIMIT $2
+      FOR UPDATE SKIP LOCKED
+    ),
+    claimed AS (
+      UPDATE runtime_invite_abuse_actions action
+      SET status = 'running',
+          attempts = attempts + 1,
+          locked_by = $1,
+          locked_until = now() + interval '5 minutes',
+          last_error = NULL,
+          updated_at = now()
+      FROM candidates
+      WHERE action.id = candidates.id
+      RETURNING
+        action.id,
+        action.subject_key,
+        action.evidence_id,
+        action.action
+    )
+    SELECT
+      claimed.action,
+      claimed.subject_key,
+      claimed.evidence_id::text AS evidence_id,
+      claimed.id::text AS action_id,
+      evidence.user_id AS actor_user_id,
+      evidence.workspace_id
+    FROM claimed
+    JOIN runtime_invite_abuse_evidence evidence
+      ON evidence.id = claimed.evidence_id
+    "#,
+  )
+  .bind(worker_id)
+  .bind(limit)
+  .fetch_all(pool)
+  .await
+  .map_err(|err| RuntimeError::database("failed to claim retryable invite abuse actions", err))?;
+
+  Ok(
+    rows
+      .into_iter()
+      .map(|row| RuntimeInviteAbuseClaimedAction {
+        action: row.get("action"),
+        subject_key: row.get("subject_key"),
+        evidence_id: row.get("evidence_id"),
+        action_id: row.get("action_id"),
+        actor_user_id: row.get("actor_user_id"),
+        workspace_id: row.get("workspace_id"),
+      })
+      .collect(),
+  )
+}
+
+async fn mark_invite_abuse_action(
+  pool: &PgPool,
+  action_id: &str,
+  worker_id: &str,
+  status: &str,
+  error: Option<String>,
+) -> RuntimeResult<bool> {
+  let result = match status {
+    "succeeded" => sqlx::query(
+      r#"
+        UPDATE runtime_invite_abuse_actions
+        SET status = 'succeeded',
+            next_attempt_at = NULL,
+            locked_by = NULL,
+            locked_until = NULL,
+            last_error = $3,
+            updated_at = now()
+        WHERE id = $1::bigint
+          AND status = 'running'
+          AND locked_by = $2
+        "#,
+    )
+    .bind(action_id)
+    .bind(worker_id)
+    .bind(error)
+    .execute(pool)
+    .await
+    .map_err(|err| RuntimeError::database("failed to mark invite abuse action succeeded", err))?,
+    "failed" => sqlx::query(
+      r#"
+        UPDATE runtime_invite_abuse_actions
+        SET status = CASE WHEN attempts >= 5 THEN 'failed' ELSE 'retry_wait' END,
+            next_attempt_at = CASE WHEN attempts >= 5 THEN next_attempt_at ELSE now() + interval '5 minutes' END,
+            locked_by = NULL,
+            locked_until = NULL,
+            last_error = $3,
+            updated_at = now()
+        WHERE id = $1::bigint
+          AND status = 'running'
+          AND locked_by = $2
+        "#,
+    )
+    .bind(action_id)
+    .bind(worker_id)
+    .bind(error)
+    .execute(pool)
+    .await
+    .map_err(|err| RuntimeError::database("failed to mark invite abuse action failed", err))?,
+    _ => return Err(RuntimeError::invalid_input("invalid invite abuse action status")),
+  };
+  Ok(result.rows_affected() > 0)
+}
+
+#[napi_derive::napi]
+impl BackendRuntime {
+  #[napi]
+  pub async fn is_invite_abuse_user_quarantined_or_banned(&self, user_id: String) -> Result<bool> {
+    let pool = self.pool().await?;
+    invite_abuse_user_quarantined_or_banned(&pool, &user_id)
+      .await
+      .map_err(Into::into)
+  }
+
+  #[napi]
+  pub async fn is_invite_abuse_workspace_quarantined(&self, workspace_id: String) -> Result<bool> {
+    let pool = self.pool().await?;
+    invite_abuse_workspace_quarantined(&pool, &workspace_id)
+      .await
+      .map_err(Into::into)
+  }
+
+  #[napi]
+  pub async fn claim_invite_abuse_action(&self, action_id: String, worker_id: String) -> Result<bool> {
+    let pool = self.pool().await?;
+    claim_invite_abuse_action(&pool, &action_id, &worker_id)
+      .await
+      .map_err(Into::into)
+  }
+
+  #[napi]
+  pub async fn claim_retryable_invite_abuse_actions(
+    &self,
+    worker_id: String,
+    limit: i64,
+  ) -> Result<Vec<RuntimeInviteAbuseClaimedAction>> {
+    if limit <= 0 {
+      return Err(napi_error("invite abuse action claim limit must be positive"));
+    }
+    let pool = self.pool().await?;
+    claim_retryable_invite_abuse_actions(&pool, &worker_id, limit)
+      .await
+      .map_err(Into::into)
+  }
+
+  #[napi]
+  pub async fn mark_invite_abuse_action(
+    &self,
+    action_id: String,
+    worker_id: String,
+    status: String,
+    error: Option<String>,
+  ) -> Result<bool> {
+    let pool = self.pool().await?;
+    mark_invite_abuse_action(&pool, &action_id, &worker_id, &status, error)
+      .await
+      .map_err(Into::into)
+  }
+}

packages/backend/native/src/runtime/backend_runtime/rolling_quota/mail_delivery.rs

@@ -0,0 +1,172 @@
+use super::{
+  InviteQuotaConfig, QuotaViolation, RuntimeMailDeliveryQuotaDecision, RuntimeMailDeliveryQuotaInput, ScopeLimit,
+  high_risk_domain, normalize_domain, scope, short_hash, source_prefix,
+};
+#[cfg(test)]
+use super::{RuntimeMailDeliveryQuotaMetadataInput, RuntimeMailDeliveryQuotaRecipientInput, RuntimeQuotaSourceInput};
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(super) enum MailClass {
+  Auth,
+  WorkspaceInvitation,
+  CollaborationNotice,
+  WorkspaceLifecycle,
+  BillingLicense,
+}
+
+impl MailClass {
+  pub(super) fn as_str(self) -> &'static str {
+    match self {
+      Self::Auth => "auth",
+      Self::WorkspaceInvitation => "workspace_invitation",
+      Self::CollaborationNotice => "collaboration_notice",
+      Self::WorkspaceLifecycle => "workspace_lifecycle",
+      Self::BillingLicense => "billing_license",
+    }
+  }
+}
+
+pub(super) fn mail_class(mail_name: &str, config: &InviteQuotaConfig) -> Option<MailClass> {
+  match config.mail_class_mapping.get(mail_name).map(String::as_str) {
+    Some("auth") => Some(MailClass::Auth),
+    Some("workspace_invitation") => Some(MailClass::WorkspaceInvitation),
+    Some("collaboration_notice") => Some(MailClass::CollaborationNotice),
+    Some("workspace_lifecycle") => Some(MailClass::WorkspaceLifecycle),
+    Some("billing_license") => Some(MailClass::BillingLicense),
+    _ => None,
+  }
+}
+
+pub(super) fn build_mail_scopes(
+  input: &RuntimeMailDeliveryQuotaInput,
+  class: MailClass,
+  config: &InviteQuotaConfig,
+) -> Vec<ScopeLimit> {
+  let recipient_hash = short_hash(&input.recipient.email.trim().to_ascii_lowercase());
+  let domain = normalize_domain(&input.recipient.domain);
+  let class_name = class.as_str();
+  let mut scopes = vec![
+    scope(
+      format!("mail:recipient:{recipient_hash}:class:{class_name}"),
+      3600,
+      20,
+      1,
+    ),
+    scope(
+      format!("mail:recipient_domain:{domain}:class:{class_name}"),
+      3600,
+      250,
+      1,
+    ),
+    scope("mail:provider_global:default".to_string(), 60, 500, 1),
+  ];
+
+  match class {
+    MailClass::Auth => {
+      if let Some(prefix) = source_prefix(input.source.as_ref()) {
+        scopes.push(scope(format!("mail:source_prefix:{prefix}:class:auth"), 3600, 50, 1));
+      }
+    }
+    MailClass::WorkspaceInvitation => {
+      if let Some(prefix) = source_prefix(input.source.as_ref()) {
+        scopes.push(scope(
+          format!("mail:source_prefix_domain:{prefix}:{domain}:class:{class_name}"),
+          3600,
+          if high_risk_domain(&domain, config) { 10 } else { 50 },
+          1,
+        ));
+      }
+      if let Some(subject) = input.metadata.abuse_subject_key.as_deref() {
+        scopes.push(scope(
+          format!("mail:abuse_subject:{subject}:class:{class_name}"),
+          86_400,
+          0,
+          1,
+        ));
+      }
+    }
+    MailClass::CollaborationNotice | MailClass::WorkspaceLifecycle => {
+      if let Some(actor) = input.metadata.actor_user_id.as_deref() {
+        scopes.push(scope(format!("mail:actor:{actor}:class:{class_name}"), 3600, 200, 1));
+      }
+      if let Some(workspace) = input.metadata.workspace_id.as_deref() {
+        scopes.push(scope(
+          format!("mail:workspace:{workspace}:class:{class_name}"),
+          3600,
+          1000,
+          1,
+        ));
+      }
+    }
+    MailClass::BillingLicense => {}
+  }
+
+  scopes
+}
+
+pub(super) fn decision_from_violation(
+  violation: QuotaViolation,
+  class: MailClass,
+  reason: &str,
+) -> RuntimeMailDeliveryQuotaDecision {
+  RuntimeMailDeliveryQuotaDecision {
+    allowed: false,
+    reservation_id: None,
+    mail_class: class.as_str().to_string(),
+    retry_after_seconds: Some(60),
+    reason: Some(reason.to_string()),
+    scope_key: Some(violation.scope_key),
+    window_seconds: Some(violation.window_seconds),
+    limit: Some(violation.limit),
+    current: Some(violation.current),
+    requested: Some(violation.requested),
+  }
+}
+
+#[cfg(test)]
+mod tests {
+  use super::*;
+
+  fn invite_config() -> InviteQuotaConfig {
+    InviteQuotaConfig::default()
+  }
+
+  #[test]
+  fn mail_class_mapping_covers_known_renderers_and_fails_closed() {
+    let config = invite_config();
+    for (name, class) in [
+      ("SignIn", MailClass::Auth),
+      ("MemberInvitation", MailClass::WorkspaceInvitation),
+      ("CommentMention", MailClass::CollaborationNotice),
+      ("TeamWorkspaceExpired", MailClass::WorkspaceLifecycle),
+      ("TeamLicense", MailClass::BillingLicense),
+    ] {
+      assert_eq!(mail_class(name, &config), Some(class), "{name}");
+    }
+    assert_eq!(mail_class("TestMail", &config), None);
+    assert_eq!(mail_class("UnexpectedMail", &config), None);
+  }
+
+  #[test]
+  fn auth_mail_does_not_expand_actor_or_workspace_scopes() {
+    let input = RuntimeMailDeliveryQuotaInput {
+      request_id: None,
+      mail_name: "SignIn".to_string(),
+      recipient: RuntimeMailDeliveryQuotaRecipientInput {
+        email: "user@example.com".to_string(),
+        domain: "example.com".to_string(),
+        user_id: Some("u1".to_string()),
+      },
+      metadata: RuntimeMailDeliveryQuotaMetadataInput {
+        actor_user_id: Some("actor".to_string()),
+        workspace_id: Some("workspace".to_string()),
+        notification_id: None,
+        abuse_subject_key: None,
+      },
+      source: None::<RuntimeQuotaSourceInput>,
+    };
+    let scopes = build_mail_scopes(&input, MailClass::Auth, &invite_config());
+    assert!(scopes.iter().all(|scope| !scope.scope_key.contains("actor")));
+    assert!(scopes.iter().all(|scope| !scope.scope_key.contains("workspace")));
+  }
+}

packages/backend/native/src/runtime/backend_runtime/rolling_quota/mod.rs

@@ -0,0 +1,175 @@
+use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
+
+mod invite_abuse_actions;
+mod mail_delivery;
+mod reservation;
+mod workspace_invite;
+mod workspace_invite_policy;
+
+use mail_delivery::{build_mail_scopes, decision_from_violation as mail_decision_from_violation, mail_class};
+use napi::Result;
+use reservation::{
+  QuotaViolation, ScopeLimit, bucket_seconds, cleanup_expired, commit_reservation, release_reservation, reserve_scopes,
+  scope,
+};
+use sha2::{Digest, Sha256};
+use workspace_invite_policy::{
+  ActorFacts, InviteAbuseDecision, InviteActivityFacts, QuotaFacts, WorkspaceFacts, build_invite_scopes,
+  evaluate_projection, high_confidence_invite_abuse, invite_commit_usage_for_scope, source_cohort_subject_key,
+  subject_hash, sum_domains,
+};
+
+#[cfg(test)]
+pub(super) use super::types::{RuntimeMailDeliveryQuotaMetadataInput, RuntimeMailDeliveryQuotaRecipientInput};
+pub(super) use super::{
+  BackendRuntime, InviteQuotaConfig, RuntimeError, RuntimeResult, napi_error,
+  types::{
+    RuntimeInviteAbuseActionRequired, RuntimeInviteAbuseClaimedAction, RuntimeMailDeliveryQuotaDecision,
+    RuntimeMailDeliveryQuotaInput, RuntimeQuotaSourceInput, RuntimeQuotaTargetDomainInput,
+    RuntimeWorkspaceInviteQuotaDecision, RuntimeWorkspaceInviteQuotaInput, RuntimeWorkspaceInviteQuotaUsage,
+  },
+};
+
+pub(super) fn normalize_domain(domain: &str) -> String {
+  domain.trim().trim_end_matches('.').to_ascii_lowercase()
+}
+
+pub(super) fn high_risk_domain(domain: &str, config: &InviteQuotaConfig) -> bool {
+  let domain = normalize_domain(domain);
+  config
+    .high_risk_target_domains
+    .iter()
+    .any(|configured| normalize_domain(configured) == domain)
+}
+
+pub(super) fn workspace_subject_key(workspace_id: &str) -> String {
+  format!("workspace:v1:{}", short_hash(workspace_id))
+}
+
+pub(super) fn short_hash(value: &str) -> String {
+  hex::encode(Sha256::digest(value.as_bytes()))[..24].to_string()
+}
+
+pub(super) fn source_prefix(source: Option<&RuntimeQuotaSourceInput>) -> Option<String> {
+  let source = source?;
+  if !source.trusted {
+    return None;
+  }
+  let ip = source.ip.as_ref()?.parse::<IpAddr>().ok()?;
+  match ip {
+    IpAddr::V4(ip) => {
+      let octets = ip.octets();
+      Some(Ipv4Addr::new(octets[0], octets[1], octets[2], 0).to_string() + "/24")
+    }
+    IpAddr::V6(ip) => {
+      let segments = ip.segments();
+      Some(Ipv6Addr::new(segments[0], segments[1], segments[2], 0, 0, 0, 0, 0).to_string() + "/48")
+    }
+  }
+}
+
+#[napi_derive::napi]
+impl BackendRuntime {
+  #[napi]
+  pub async fn assert_mail_delivery_quota_v1(
+    &self,
+    input: RuntimeMailDeliveryQuotaInput,
+  ) -> Result<RuntimeMailDeliveryQuotaDecision> {
+    let config = self.config()?.invite_quota;
+    let Some(class) = mail_class(&input.mail_name, &config) else {
+      return Ok(RuntimeMailDeliveryQuotaDecision {
+        allowed: false,
+        reservation_id: None,
+        mail_class: "unmapped".to_string(),
+        retry_after_seconds: None,
+        reason: Some("unmapped_mail_name".to_string()),
+        scope_key: Some(format!("mail:name:{}", input.mail_name)),
+        window_seconds: None,
+        limit: None,
+        current: None,
+        requested: Some(1),
+      });
+    };
+    let pool = self.pool().await?;
+    let scopes = build_mail_scopes(&input, class, &config);
+    match reserve_scopes(&pool, "mail_delivery", input.request_id.as_deref(), scopes).await? {
+      Ok(reservation) => Ok(RuntimeMailDeliveryQuotaDecision {
+        allowed: true,
+        reservation_id: Some(reservation.reservation_id),
+        mail_class: class.as_str().to_string(),
+        retry_after_seconds: None,
+        reason: None,
+        scope_key: None,
+        window_seconds: None,
+        limit: None,
+        current: None,
+        requested: Some(1),
+      }),
+      Err(violation) => Ok(mail_decision_from_violation(violation, class, "mail_class")),
+    }
+  }
+
+  #[napi]
+  pub async fn commit_mail_delivery_quota_v1(&self, reservation_id: String) -> Result<bool> {
+    let pool = self.pool().await?;
+    commit_reservation(&pool, &reservation_id, 1, |_, reserved_count| reserved_count.min(1))
+      .await
+      .map_err(Into::into)
+  }
+
+  #[napi]
+  pub async fn release_mail_delivery_quota_v1(&self, reservation_id: String) -> Result<bool> {
+    let pool = self.pool().await?;
+    release_reservation(&pool, &reservation_id).await.map_err(Into::into)
+  }
+
+  #[napi]
+  pub async fn cleanup_expired_rolling_quota(&self, limit: i64) -> Result<i64> {
+    if limit <= 0 {
+      return Err(napi_error("rolling quota cleanup limit must be positive"));
+    }
+    let pool = self.pool().await?;
+    cleanup_expired(&pool, limit).await.map_err(Into::into)
+  }
+}
+
+#[cfg(test)]
+mod tests {
+  use super::*;
+
+  #[test]
+  fn derives_source_prefix_only_from_trusted_raw_ip() {
+    assert_eq!(
+      source_prefix(Some(&RuntimeQuotaSourceInput {
+        trusted: true,
+        ip: Some("192.168.12.34".to_string()),
+        country: None,
+        asn: None,
+        ray_id: None,
+      }))
+      .as_deref(),
+      Some("192.168.12.0/24")
+    );
+    assert_eq!(
+      source_prefix(Some(&RuntimeQuotaSourceInput {
+        trusted: true,
+        ip: Some("2001:db8:abcd:1234::1".to_string()),
+        country: None,
+        asn: None,
+        ray_id: None,
+      }))
+      .as_deref(),
+      Some("2001:db8:abcd::/48")
+    );
+    assert!(
+      source_prefix(Some(&RuntimeQuotaSourceInput {
+        trusted: false,
+        ip: Some("192.168.12.34".to_string()),
+        country: None,
+        asn: None,
+        ray_id: None,
+      }))
+      .is_none()
+    );
+  }
+}

packages/backend/native/src/runtime/backend_runtime/rolling_quota/reservation.rs

@@ -0,0 +1,344 @@
+use chrono::{DateTime, Duration, TimeZone, Utc};
+use sha2::{Digest, Sha256};
+use sqlx::{PgPool, Postgres, Row, Transaction};
+use uuid::Uuid;
+
+use super::{RuntimeError, RuntimeResult};
+
+const RESERVATION_TTL_SECONDS: i64 = 120;
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub(super) struct ScopeLimit {
+  pub(super) scope_key: String,
+  pub(super) window_seconds: i32,
+  pub(super) bucket_seconds: i64,
+  pub(super) limit: i32,
+  pub(super) requested: i32,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub(super) struct QuotaViolation {
+  pub(super) scope_key: String,
+  pub(super) window_seconds: i32,
+  pub(super) limit: i32,
+  pub(super) current: i32,
+  pub(super) requested: i32,
+}
+
+#[derive(Clone, Debug)]
+pub(super) struct ReservationDecision {
+  pub(super) reservation_id: String,
+}
+
+pub(super) fn bucket_seconds(window_seconds: i32) -> i64 {
+  match window_seconds {
+    60 => 10,
+    3600 => 5 * 60,
+    86_400 => 60 * 60,
+    604_800 => 6 * 60 * 60,
+    _ => 60,
+  }
+}
+
+fn bucket_start(now: DateTime<Utc>, bucket_seconds: i64) -> DateTime<Utc> {
+  let timestamp = now.timestamp();
+  Utc
+    .timestamp_opt(timestamp - timestamp.rem_euclid(bucket_seconds), 0)
+    .single()
+    .expect("valid unix timestamp bucket")
+}
+
+pub(super) fn scope(scope_key: String, window_seconds: i32, limit: i32, requested: i32) -> ScopeLimit {
+  ScopeLimit {
+    scope_key,
+    window_seconds,
+    bucket_seconds: bucket_seconds(window_seconds),
+    limit,
+    requested,
+  }
+}
+
+pub(super) async fn reserve_scopes(
+  pool: &PgPool,
+  purpose: &str,
+  request_id: Option<&str>,
+  scopes: Vec<ScopeLimit>,
+) -> RuntimeResult<std::result::Result<ReservationDecision, QuotaViolation>> {
+  let now: DateTime<Utc> = sqlx::query_scalar("SELECT clock_timestamp()")
+    .fetch_one(pool)
+    .await
+    .map_err(|err| RuntimeError::database("failed to read database clock", err))?;
+  let mut tx = pool
+    .begin()
+    .await
+    .map_err(|err| RuntimeError::database("failed to begin quota reservation transaction", err))?;
+
+  let mut sorted_scope_keys = std::collections::BTreeSet::new();
+  for scope in &scopes {
+    sorted_scope_keys.insert(scope.scope_key.clone());
+  }
+  for scope_key in sorted_scope_keys {
+    advisory_lock(&mut tx, &scope_key).await?;
+  }
+
+  let reservation_id = Uuid::new_v4();
+  for scope in &scopes {
+    let bucket_start = bucket_start(now, scope.bucket_seconds);
+    let window_start = now - Duration::seconds(scope.window_seconds as i64);
+    let committed: i64 = sqlx::query_scalar(
+      r#"
+      SELECT COALESCE(SUM(count), 0)::bigint
+      FROM runtime_rolling_quota_counters
+      WHERE scope_key = $1
+        AND window_seconds = $2
+        AND bucket_start >= $3
+      "#,
+    )
+    .bind(&scope.scope_key)
+    .bind(scope.window_seconds)
+    .bind(window_start)
+    .fetch_one(&mut *tx)
+    .await
+    .map_err(|err| RuntimeError::database("failed to read rolling quota counter", err))?;
+    let reserved: i64 = sqlx::query_scalar(
+      r#"
+      SELECT COALESCE(SUM(count), 0)::bigint
+      FROM runtime_rolling_quota_reservations
+      WHERE scope_key = $1
+        AND window_seconds = $2
+        AND bucket_start >= $3
+        AND status = 'reserved'
+        AND expires_at > $4
+      "#,
+    )
+    .bind(&scope.scope_key)
+    .bind(scope.window_seconds)
+    .bind(window_start)
+    .bind(now)
+    .fetch_one(&mut *tx)
+    .await
+    .map_err(|err| RuntimeError::database("failed to read rolling quota reservations", err))?;
+    let current = committed.saturating_add(reserved) as i32;
+    if current.saturating_add(scope.requested) > scope.limit {
+      tx.rollback()
+        .await
+        .map_err(|err| RuntimeError::database("failed to rollback rejected quota reservation", err))?;
+      return Ok(Err(QuotaViolation {
+        scope_key: scope.scope_key.clone(),
+        window_seconds: scope.window_seconds,
+        limit: scope.limit,
+        current,
+        requested: scope.requested,
+      }));
+    }
+    sqlx::query(
+      r#"
+      INSERT INTO runtime_rolling_quota_reservations (
+        id, scope_key, window_seconds, bucket_start, count, status, purpose, request_id, expires_at
+      )
+      VALUES ($1::uuid, $2, $3, $4, $5, 'reserved', $6, $7, $8)
+      "#,
+    )
+    .bind(reservation_id.to_string())
+    .bind(&scope.scope_key)
+    .bind(scope.window_seconds)
+    .bind(bucket_start)
+    .bind(scope.requested)
+    .bind(purpose)
+    .bind(request_id)
+    .bind(now + Duration::seconds(RESERVATION_TTL_SECONDS))
+    .execute(&mut *tx)
+    .await
+    .map_err(|err| RuntimeError::database("failed to create rolling quota reservation", err))?;
+  }
+
+  tx.commit()
+    .await
+    .map_err(|err| RuntimeError::database("failed to commit quota reservation", err))?;
+  Ok(Ok(ReservationDecision {
+    reservation_id: reservation_id.to_string(),
+  }))
+}
+
+async fn advisory_lock(tx: &mut Transaction<'_, Postgres>, scope_key: &str) -> RuntimeResult<()> {
+  let hash = Sha256::digest(scope_key.as_bytes());
+  let mut bytes = [0_u8; 8];
+  bytes.copy_from_slice(&hash[..8]);
+  let lock_key = i64::from_be_bytes(bytes);
+  sqlx::query("SELECT pg_advisory_xact_lock($1)")
+    .bind(lock_key)
+    .execute(&mut **tx)
+    .await
+    .map_err(|err| RuntimeError::database("failed to acquire rolling quota advisory lock", err))?;
+  Ok(())
+}
+
+pub(super) async fn commit_reservation<F>(
+  pool: &PgPool,
+  reservation_id: &str,
+  settle_usage: i32,
+  actual_usage_for_scope: F,
+) -> RuntimeResult<bool>
+where
+  F: Fn(&str, i32) -> i32,
+{
+  let reservation_id =
+    Uuid::parse_str(reservation_id).map_err(|_| RuntimeError::invalid_input("invalid reservation id"))?;
+  let reservation_id = reservation_id.to_string();
+  let now: DateTime<Utc> = sqlx::query_scalar("SELECT clock_timestamp()")
+    .fetch_one(pool)
+    .await
+    .map_err(|err| RuntimeError::database("failed to read database clock", err))?;
+  let mut tx = pool
+    .begin()
+    .await
+    .map_err(|err| RuntimeError::database("failed to begin quota commit transaction", err))?;
+  let rows = sqlx::query(
+    r#"
+    SELECT scope_key, window_seconds, bucket_start, count
+    FROM runtime_rolling_quota_reservations
+    WHERE id = $1::uuid AND status = 'reserved'
+    FOR UPDATE
+    "#,
+  )
+  .bind(&reservation_id)
+  .fetch_all(&mut *tx)
+  .await
+  .map_err(|err| RuntimeError::database("failed to load quota reservation", err))?;
+  if rows.is_empty() {
+    tx.rollback()
+      .await
+      .map_err(|err| RuntimeError::database("failed to rollback empty quota commit", err))?;
+    return Ok(false);
+  }
+  for row in &rows {
+    advisory_lock(&mut tx, row.get("scope_key")).await?;
+  }
+  for row in &rows {
+    let reserved_count: i32 = row.get("count");
+    let scope_key: String = row.get("scope_key");
+    let count = actual_usage_for_scope(&scope_key, reserved_count)
+      .min(reserved_count)
+      .max(0);
+    if count > 0 {
+      let window_seconds: i32 = row.get("window_seconds");
+      sqlx::query(
+        r#"
+        INSERT INTO runtime_rolling_quota_counters (scope_key, window_seconds, bucket_start, count, expires_at, updated_at)
+        VALUES ($1, $2, $3, $4, $5, $6)
+        ON CONFLICT (scope_key, window_seconds, bucket_start)
+        DO UPDATE SET count = runtime_rolling_quota_counters.count + EXCLUDED.count,
+                      expires_at = GREATEST(runtime_rolling_quota_counters.expires_at, EXCLUDED.expires_at),
+                      updated_at = EXCLUDED.updated_at
+        "#,
+      )
+      .bind(scope_key)
+      .bind(window_seconds)
+      .bind(row.get::<DateTime<Utc>, _>("bucket_start"))
+      .bind(count)
+      .bind(now + Duration::seconds(window_seconds as i64 * 2))
+      .bind(now)
+      .execute(&mut *tx)
+      .await
+      .map_err(|err| RuntimeError::database("failed to commit rolling quota counter", err))?;
+    }
+  }
+  sqlx::query(
+    r#"
+    UPDATE runtime_rolling_quota_reservations
+    SET status = CASE WHEN $2 > 0 THEN 'committed' ELSE 'released' END,
+        committed_at = CASE WHEN $2 > 0 THEN $3 ELSE committed_at END,
+        released_at = CASE WHEN $2 <= 0 THEN $3 ELSE released_at END,
+        updated_at = $3
+    WHERE id = $1::uuid AND status = 'reserved'
+    "#,
+  )
+  .bind(&reservation_id)
+  .bind(settle_usage)
+  .bind(now)
+  .execute(&mut *tx)
+  .await
+  .map_err(|err| RuntimeError::database("failed to settle quota reservation", err))?;
+  tx.commit()
+    .await
+    .map_err(|err| RuntimeError::database("failed to commit quota commit transaction", err))?;
+  Ok(true)
+}
+
+pub(super) async fn release_reservation(pool: &PgPool, reservation_id: &str) -> RuntimeResult<bool> {
+  let reservation_id =
+    Uuid::parse_str(reservation_id).map_err(|_| RuntimeError::invalid_input("invalid reservation id"))?;
+  let reservation_id = reservation_id.to_string();
+  let result = sqlx::query(
+    r#"
+    UPDATE runtime_rolling_quota_reservations
+    SET status = 'released',
+        released_at = clock_timestamp(),
+        updated_at = clock_timestamp()
+    WHERE id = $1::uuid AND status = 'reserved'
+    "#,
+  )
+  .bind(reservation_id)
+  .execute(pool)
+  .await
+  .map_err(|err| RuntimeError::database("failed to release quota reservation", err))?;
+  Ok(result.rows_affected() > 0)
+}
+
+pub(super) async fn cleanup_expired(pool: &PgPool, limit: i64) -> RuntimeResult<i64> {
+  let released = sqlx::query(
+    r#"
+    UPDATE runtime_rolling_quota_reservations
+    SET status = 'expired',
+        updated_at = clock_timestamp()
+    WHERE (id, scope_key, window_seconds, bucket_start) IN (
+      SELECT id, scope_key, window_seconds, bucket_start
+      FROM runtime_rolling_quota_reservations
+      WHERE status = 'reserved' AND expires_at <= clock_timestamp()
+      LIMIT $1
+    )
+    "#,
+  )
+  .bind(limit)
+  .execute(pool)
+  .await
+  .map_err(|err| RuntimeError::database("failed to expire rolling quota reservations", err))?
+  .rows_affected() as i64;
+
+  let counters = sqlx::query(
+    r#"
+    DELETE FROM runtime_rolling_quota_counters
+    WHERE (scope_key, window_seconds, bucket_start) IN (
+      SELECT scope_key, window_seconds, bucket_start
+      FROM runtime_rolling_quota_counters
+      WHERE expires_at <= clock_timestamp()
+      LIMIT $1
+    )
+    "#,
+  )
+  .bind(limit)
+  .execute(pool)
+  .await
+  .map_err(|err| RuntimeError::database("failed to delete expired rolling quota counters", err))?
+  .rows_affected() as i64;
+
+  Ok(released + counters)
+}
+
+#[cfg(test)]
+mod tests {
+  use super::*;
+
+  #[test]
+  fn bucket_boundaries_match_policy_windows() {
+    let now = Utc.with_ymd_and_hms(2026, 7, 6, 1, 2, 3).single().unwrap();
+    assert_eq!(bucket_seconds(60), 10);
+    assert_eq!(bucket_seconds(3600), 300);
+    assert_eq!(bucket_seconds(86_400), 3600);
+    assert_eq!(bucket_seconds(604_800), 21_600);
+    assert_eq!(
+      bucket_start(now, 300),
+      Utc.with_ymd_and_hms(2026, 7, 6, 1, 0, 0).single().unwrap()
+    );
+  }
+}

packages/backend/native/src/runtime/backend_runtime/rolling_quota/workspace_invite.rs

@@ -0,0 +1,452 @@
+use chrono::{DateTime, Utc};
+use napi::Result;
+use serde_json::json;
+use sqlx::{PgPool, Row};
+
+use super::{
+  ActorFacts, BackendRuntime, InviteAbuseDecision, InviteActivityFacts, InviteQuotaConfig, QuotaFacts, QuotaViolation,
+  RuntimeError, RuntimeInviteAbuseActionRequired, RuntimeResult, RuntimeWorkspaceInviteQuotaDecision,
+  RuntimeWorkspaceInviteQuotaInput, RuntimeWorkspaceInviteQuotaUsage, WorkspaceFacts, build_invite_scopes,
+  commit_reservation, evaluate_projection, high_confidence_invite_abuse, invite_commit_usage_for_scope, napi_error,
+  normalize_domain, release_reservation, reserve_scopes, short_hash, source_cohort_subject_key, source_prefix,
+  subject_hash, sum_domains, workspace_subject_key,
+};
+
+async fn load_actor(pool: &PgPool, user_id: &str) -> RuntimeResult<ActorFacts> {
+  let row = sqlx::query(
+    r#"
+    SELECT email, created_at, registered, email_verified IS NOT NULL AS email_verified, disabled
+    FROM users
+    WHERE id = $1
+    "#,
+  )
+  .bind(user_id)
+  .fetch_optional(pool)
+  .await
+  .map_err(|err| RuntimeError::database("failed to load invite actor", err))?
+  .ok_or_else(|| RuntimeError::invalid_input("invite actor not found"))?;
+
+  Ok(ActorFacts {
+    email: row.get("email"),
+    created_at: row.get("created_at"),
+    registered: row.get("registered"),
+    email_verified: row.get("email_verified"),
+    disabled: row.get("disabled"),
+  })
+}
+
+async fn load_workspace(pool: &PgPool, workspace_id: &str) -> RuntimeResult<WorkspaceFacts> {
+  let row = sqlx::query("SELECT created_at FROM workspaces WHERE id = $1")
+    .bind(workspace_id)
+    .fetch_optional(pool)
+    .await
+    .map_err(|err| RuntimeError::database("failed to load workspace", err))?
+    .ok_or_else(|| RuntimeError::invalid_input("workspace not found"))?;
+
+  Ok(WorkspaceFacts {
+    created_at: row.get("created_at"),
+  })
+}
+
+async fn load_invite_activity(
+  pool: &PgPool,
+  actor_user_id: &str,
+  workspace_id: &str,
+) -> RuntimeResult<InviteActivityFacts> {
+  let row = sqlx::query(
+    r#"
+    SELECT
+      COUNT(*) FILTER (
+        WHERE inviter_user_id = $1 AND created_at >= clock_timestamp() - interval '7 days'
+      )::int AS actor_created_7d,
+      COUNT(*) FILTER (
+        WHERE inviter_user_id = $1 AND accepted_at >= clock_timestamp() - interval '7 days'
+      )::int AS actor_accepted_7d,
+      COUNT(*) FILTER (
+        WHERE workspace_id = $2 AND status IN ('pending', 'waiting_review', 'waiting_seat')
+      )::int AS workspace_pending,
+      COUNT(*) FILTER (
+        WHERE workspace_id = $2 AND created_at >= clock_timestamp() - interval '7 days'
+      )::int AS workspace_created_7d,
+      COUNT(*) FILTER (
+        WHERE workspace_id = $2 AND accepted_at >= clock_timestamp() - interval '7 days'
+      )::int AS workspace_accepted_7d
+    FROM workspace_invitations
+    WHERE inviter_user_id = $1 OR workspace_id = $2
+    "#,
+  )
+  .bind(actor_user_id)
+  .bind(workspace_id)
+  .fetch_one(pool)
+  .await
+  .map_err(|err| RuntimeError::database("failed to load invite activity facts", err))?;
+
+  Ok(InviteActivityFacts {
+    actor_created_7d: row.get("actor_created_7d"),
+    actor_accepted_7d: row.get("actor_accepted_7d"),
+    workspace_pending: row.get("workspace_pending"),
+    workspace_created_7d: row.get("workspace_created_7d"),
+    workspace_accepted_7d: row.get("workspace_accepted_7d"),
+  })
+}
+
+async fn load_quota(pool: &PgPool, workspace_id: &str) -> RuntimeResult<Option<QuotaFacts>> {
+  let row = sqlx::query(
+    r#"
+    SELECT plan, owner_user_id, uses_owner_quota, seat_limit, member_count, known, stale, stale_after
+    FROM effective_workspace_quota_states
+    WHERE workspace_id = $1
+    "#,
+  )
+  .bind(workspace_id)
+  .fetch_optional(pool)
+  .await
+  .map_err(|err| RuntimeError::database("failed to load workspace quota state", err))?;
+
+  Ok(row.map(|row| QuotaFacts {
+    plan: row.get("plan"),
+    owner_user_id: row.get("owner_user_id"),
+    uses_owner_quota: row.get("uses_owner_quota"),
+    seat_limit: row.get("seat_limit"),
+    member_count: row.get("member_count"),
+    known: row.get("known"),
+    stale: row.get("stale"),
+    stale_after: row.get("stale_after"),
+  }))
+}
+
+async fn active_subject_status(pool: &PgPool, subject_key: &str) -> RuntimeResult<Option<String>> {
+  let row = sqlx::query("SELECT status FROM runtime_invite_abuse_subjects WHERE subject_key = $1")
+    .bind(subject_key)
+    .fetch_optional(pool)
+    .await
+    .map_err(|err| RuntimeError::database("failed to load invite abuse subject", err))?;
+  Ok(row.map(|row| row.get("status")))
+}
+
+async fn record_invite_abuse_action(
+  pool: &PgPool,
+  input: &RuntimeWorkspaceInviteQuotaInput,
+  actor: &ActorFacts,
+  decision: InviteAbuseDecision,
+  config: &InviteQuotaConfig,
+) -> RuntimeResult<RuntimeInviteAbuseActionRequired> {
+  let action = decision.action;
+  let status = if action == "ban_actor" { "banned" } else { "quarantined" };
+  let source_prefix_hash = source_prefix(input.source.as_ref()).map(|prefix| short_hash(&prefix));
+  let target_domains = json!(
+    input
+      .target_domains
+      .iter()
+      .map(|target| json!({
+        "domain": normalize_domain(&target.domain),
+        "count": target.count,
+      }))
+      .collect::<Vec<_>>()
+  );
+  let counters = json!({
+    "requested": input.target_count,
+  });
+  let actor_email_hash = subject_hash(&actor.email, config);
+  let actor_domain = actor.email.split('@').next_back().map(normalize_domain);
+  let source_asn = input.source.as_ref().and_then(|source| source.asn).map(i64::from);
+  let subject_user_id = if decision.subject_kind == "actor_email" {
+    Some(input.actor_user_id.as_str())
+  } else {
+    None
+  };
+
+  let mut tx = pool
+    .begin()
+    .await
+    .map_err(|err| RuntimeError::database("failed to start invite abuse transaction", err))?;
+
+  sqlx::query(
+    r#"
+    INSERT INTO runtime_invite_abuse_subjects (
+      subject_key,
+      kind,
+      user_id,
+      actor_email_hash,
+      email_domain,
+      first_seen_at,
+      last_seen_at,
+      status,
+      action,
+      action_reason,
+      action_at
+    )
+    VALUES ($1, $2, $3, $4, $5, now(), now(), $6, $7, $8, now())
+    ON CONFLICT (subject_key)
+    DO UPDATE SET
+      user_id = EXCLUDED.user_id,
+      actor_email_hash = EXCLUDED.actor_email_hash,
+      email_domain = EXCLUDED.email_domain,
+      last_seen_at = now(),
+      status = EXCLUDED.status,
+      action = EXCLUDED.action,
+      action_reason = EXCLUDED.action_reason,
+      action_at = now(),
+      updated_at = now()
+    "#,
+  )
+  .bind(&decision.subject_key)
+  .bind(decision.subject_kind)
+  .bind(subject_user_id)
+  .bind(actor_email_hash)
+  .bind(actor_domain)
+  .bind(status)
+  .bind(action)
+  .bind(decision.reason)
+  .execute(&mut *tx)
+  .await
+  .map_err(|err| RuntimeError::database("failed to upsert invite abuse subject", err))?;
+
+  let evidence_id: i64 = sqlx::query_scalar(
+    r#"
+    INSERT INTO runtime_invite_abuse_evidence (
+      subject_key,
+      request_id,
+      workspace_id,
+      user_id,
+      actor_email_hash,
+      source_prefix_hash,
+      source_asn,
+      target_domains,
+      counters,
+      decision,
+      reason
+    )
+    VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
+    RETURNING id
+    "#,
+  )
+  .bind(&decision.subject_key)
+  .bind(&input.request_id)
+  .bind(&input.workspace_id)
+  .bind(&input.actor_user_id)
+  .bind(subject_hash(&actor.email, config))
+  .bind(source_prefix_hash)
+  .bind(source_asn)
+  .bind(target_domains)
+  .bind(counters)
+  .bind(action)
+  .bind(decision.reason)
+  .fetch_one(&mut *tx)
+  .await
+  .map_err(|err| RuntimeError::database("failed to insert invite abuse evidence", err))?;
+
+  let action_id: i64 = sqlx::query_scalar(
+    r#"
+    INSERT INTO runtime_invite_abuse_actions (
+      subject_key,
+      evidence_id,
+      action,
+      status,
+      next_attempt_at
+    )
+    VALUES ($1, $2, $3, 'pending', now())
+    RETURNING id
+    "#,
+  )
+  .bind(&decision.subject_key)
+  .bind(evidence_id)
+  .bind(action)
+  .fetch_one(&mut *tx)
+  .await
+  .map_err(|err| RuntimeError::database("failed to insert invite abuse action", err))?;
+
+  tx.commit()
+    .await
+    .map_err(|err| RuntimeError::database("failed to commit invite abuse transaction", err))?;
+
+  Ok(RuntimeInviteAbuseActionRequired {
+    action: action.to_string(),
+    subject_key: decision.subject_key,
+    evidence_id: evidence_id.to_string(),
+    action_id: action_id.to_string(),
+  })
+}
+
+fn decision_from_violation(violation: QuotaViolation, reason: &str) -> RuntimeWorkspaceInviteQuotaDecision {
+  RuntimeWorkspaceInviteQuotaDecision {
+    allowed: false,
+    reservation_id: None,
+    retry_after_seconds: Some(60),
+    reason: Some(reason.to_string()),
+    scope_key: Some(violation.scope_key),
+    window_seconds: Some(violation.window_seconds),
+    limit: Some(violation.limit),
+    current: Some(violation.current),
+    requested: Some(violation.requested),
+    action_required: None,
+  }
+}
+
+#[napi_derive::napi]
+impl BackendRuntime {
+  #[napi]
+  pub async fn assert_workspace_invite_quota_v1(
+    &self,
+    input: RuntimeWorkspaceInviteQuotaInput,
+  ) -> Result<RuntimeWorkspaceInviteQuotaDecision> {
+    if input.target_count <= 0 {
+      return Err(napi_error("target_count must be positive"));
+    }
+    let config = self.config()?.invite_quota;
+    let pool = self.pool().await?;
+    let now: DateTime<Utc> = sqlx::query_scalar("SELECT clock_timestamp()")
+      .fetch_one(&pool)
+      .await
+      .map_err(|err| RuntimeError::database("failed to read database clock", err))?;
+    let actor = load_actor(&pool, &input.actor_user_id).await?;
+    let actor_subject = subject_hash(&actor.email, &config);
+    if let Some(status) = active_subject_status(&pool, &actor_subject).await?
+      && matches!(status.as_str(), "banned" | "quarantined")
+    {
+      return Ok(RuntimeWorkspaceInviteQuotaDecision {
+        allowed: false,
+        reservation_id: None,
+        retry_after_seconds: None,
+        reason: Some("abuse_subject".to_string()),
+        scope_key: Some(format!("invite:actor_subject:{actor_subject}")),
+        window_seconds: None,
+        limit: None,
+        current: None,
+        requested: Some(input.target_count),
+        action_required: None,
+      });
+    }
+    let workspace_subject = workspace_subject_key(&input.workspace_id);
+    if let Some(status) = active_subject_status(&pool, &workspace_subject).await?
+      && status == "quarantined"
+    {
+      return Ok(RuntimeWorkspaceInviteQuotaDecision {
+        allowed: false,
+        reservation_id: None,
+        retry_after_seconds: None,
+        reason: Some("abuse_workspace".to_string()),
+        scope_key: Some(format!("invite:workspace_subject:{workspace_subject}")),
+        window_seconds: None,
+        limit: None,
+        current: None,
+        requested: Some(input.target_count),
+        action_required: None,
+      });
+    }
+    if let Some(prefix) = source_prefix(input.source.as_ref()) {
+      for target in &input.target_domains {
+        let source_subject = source_cohort_subject_key(&prefix, &target.domain);
+        if let Some(status) = active_subject_status(&pool, &source_subject).await?
+          && status == "quarantined"
+        {
+          return Ok(RuntimeWorkspaceInviteQuotaDecision {
+            allowed: false,
+            reservation_id: None,
+            retry_after_seconds: None,
+            reason: Some("abuse_source_cohort".to_string()),
+            scope_key: Some(format!("invite:source_cohort_subject:{source_subject}")),
+            window_seconds: None,
+            limit: None,
+            current: None,
+            requested: Some(input.target_count),
+            action_required: None,
+          });
+        }
+      }
+    }
+
+    let quota = match load_quota(&pool, &input.workspace_id).await? {
+      Some(quota) => quota,
+      None => {
+        return Ok(RuntimeWorkspaceInviteQuotaDecision {
+          allowed: false,
+          reservation_id: None,
+          retry_after_seconds: None,
+          reason: Some("quota_state_unavailable".to_string()),
+          scope_key: None,
+          window_seconds: None,
+          limit: None,
+          current: None,
+          requested: Some(input.target_count),
+          action_required: None,
+        });
+      }
+    };
+    if let Some(reason) = evaluate_projection(&quota, now) {
+      return Ok(RuntimeWorkspaceInviteQuotaDecision {
+        allowed: false,
+        reservation_id: None,
+        retry_after_seconds: None,
+        reason: Some(reason.to_string()),
+        scope_key: None,
+        window_seconds: None,
+        limit: None,
+        current: None,
+        requested: Some(input.target_count),
+        action_required: None,
+      });
+    }
+    if let Some(abuse_decision) = high_confidence_invite_abuse(&input, &actor, &config) {
+      let reason = abuse_decision.reason;
+      let scope_key = match abuse_decision.subject_kind {
+        "workspace" => format!("invite:workspace_subject:{}", abuse_decision.subject_key),
+        "source_prefix_domain" => format!("invite:source_cohort_subject:{}", abuse_decision.subject_key),
+        _ => format!("invite:actor_subject:{}", abuse_decision.subject_key),
+      };
+      let action_required = record_invite_abuse_action(&pool, &input, &actor, abuse_decision, &config).await?;
+      return Ok(RuntimeWorkspaceInviteQuotaDecision {
+        allowed: false,
+        reservation_id: None,
+        retry_after_seconds: None,
+        reason: Some(reason.to_string()),
+        scope_key: Some(scope_key),
+        window_seconds: None,
+        limit: None,
+        current: None,
+        requested: Some(input.target_count),
+        action_required: Some(action_required),
+      });
+    }
+
+    let workspace = load_workspace(&pool, &input.workspace_id).await?;
+    let activity = load_invite_activity(&pool, &input.actor_user_id, &input.workspace_id).await?;
+    let scopes = build_invite_scopes(&input, &actor, &workspace, &quota, &activity, &config, now)?;
+    match reserve_scopes(&pool, "workspace_invite", input.request_id.as_deref(), scopes).await? {
+      Ok(reservation) => Ok(RuntimeWorkspaceInviteQuotaDecision {
+        allowed: true,
+        reservation_id: Some(reservation.reservation_id),
+        retry_after_seconds: None,
+        reason: None,
+        scope_key: None,
+        window_seconds: None,
+        limit: None,
+        current: None,
+        requested: Some(input.target_count),
+        action_required: None,
+      }),
+      Err(violation) => Ok(decision_from_violation(violation, "quota_subject")),
+    }
+  }
+
+  #[napi]
+  pub async fn commit_workspace_invite_quota_v1(
+    &self,
+    reservation_id: String,
+    usage: RuntimeWorkspaceInviteQuotaUsage,
+  ) -> Result<bool> {
+    let domain_usage = sum_domains(&usage.target_domains);
+    let pool = self.pool().await?;
+    commit_reservation(&pool, &reservation_id, usage.target_count, |scope_key, _| {
+      invite_commit_usage_for_scope(scope_key, usage.target_count, &domain_usage)
+    })
+    .await
+    .map_err(Into::into)
+  }
+
+  #[napi]
+  pub async fn release_workspace_invite_quota_v1(&self, reservation_id: String) -> Result<bool> {
+    let pool = self.pool().await?;
+    release_reservation(&pool, &reservation_id).await.map_err(Into::into)
+  }
+}

packages/backend/native/src/runtime/backend_runtime/rolling_quota/workspace_invite_policy.rs

@@ -0,0 +1,654 @@
+use std::collections::BTreeMap;
+
+use chrono::{DateTime, Duration, Utc};
+use napi::Result;
+use sha2::{Digest, Sha256};
+
+#[cfg(test)]
+use super::RuntimeQuotaSourceInput;
+use super::{
+  InviteQuotaConfig, RuntimeQuotaTargetDomainInput, RuntimeWorkspaceInviteQuotaInput, ScopeLimit, bucket_seconds,
+  high_risk_domain, napi_error, normalize_domain, scope, short_hash, source_prefix, workspace_subject_key,
+};
+
+#[derive(Clone, Debug)]
+pub(super) struct ActorFacts {
+  pub(super) email: String,
+  pub(super) created_at: DateTime<Utc>,
+  pub(super) registered: bool,
+  pub(super) email_verified: bool,
+  pub(super) disabled: bool,
+}
+
+#[derive(Clone, Debug)]
+pub(super) struct QuotaFacts {
+  pub(super) plan: String,
+  pub(super) owner_user_id: Option<String>,
+  pub(super) uses_owner_quota: bool,
+  pub(super) seat_limit: i32,
+  pub(super) member_count: i32,
+  pub(super) known: bool,
+  pub(super) stale: bool,
+  pub(super) stale_after: Option<DateTime<Utc>>,
+}
+
+#[derive(Clone, Debug)]
+pub(super) struct WorkspaceFacts {
+  pub(super) created_at: DateTime<Utc>,
+}
+
+#[derive(Clone, Debug, Default)]
+pub(super) struct InviteActivityFacts {
+  pub(super) actor_created_7d: i32,
+  pub(super) actor_accepted_7d: i32,
+  pub(super) workspace_pending: i32,
+  pub(super) workspace_created_7d: i32,
+  pub(super) workspace_accepted_7d: i32,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub(super) struct InviteAbuseDecision {
+  pub(super) reason: &'static str,
+  pub(super) action: &'static str,
+  pub(super) subject_kind: &'static str,
+  pub(super) subject_key: String,
+}
+
+pub(super) fn subject_hash(email: &str, config: &InviteQuotaConfig) -> String {
+  let normalized = email.trim().to_ascii_lowercase();
+  let hash = Sha256::digest(format!("{}:{normalized}", config.subject_hash_salt).as_bytes());
+  format!("actor_email_sha256:v1:{}", hex::encode(hash))
+}
+
+pub(super) fn source_cohort_subject_key(prefix: &str, domain: &str) -> String {
+  format!(
+    "source_prefix_domain_sha256:v1:{}",
+    short_hash(&format!("{}:{}", prefix, normalize_domain(domain)))
+  )
+}
+
+fn quota_subject(workspace_id: &str, quota: &QuotaFacts) -> (String, i32) {
+  if quota.uses_owner_quota || !quota.plan.to_ascii_lowercase().contains("team") {
+    (
+      format!("owner:{}", quota.owner_user_id.as_deref().unwrap_or(workspace_id)),
+      quota.seat_limit,
+    )
+  } else {
+    (format!("workspace:{workspace_id}"), quota.seat_limit)
+  }
+}
+
+fn plan_ceiling_7d(plan: &str, seat_limit: i32) -> i32 {
+  let normalized = plan.to_ascii_lowercase();
+  if normalized.contains("enterprise") || normalized.contains("team") && !normalized.contains("trial") {
+    seat_limit.saturating_mul(2)
+  } else if normalized.contains("trial") {
+    50
+  } else if normalized.contains("pro") {
+    20
+  } else {
+    10
+  }
+}
+
+fn base_invite_limits(
+  now: DateTime<Utc>,
+  actor: &ActorFacts,
+  workspace: &WorkspaceFacts,
+  quota: &QuotaFacts,
+  activity: &InviteActivityFacts,
+) -> (i32, i32, i32, i32) {
+  if actor.disabled || !actor.registered {
+    return (0, 0, 0, 0);
+  }
+  let account_age = now - actor.created_at;
+  let workspace_age = now - workspace.created_at;
+  let mut single = 5;
+  let mut per_hour = 5;
+  let mut per_day = 15;
+  let mut per_week = 30;
+
+  if account_age < Duration::hours(24) {
+    return (0, 0, 0, 0);
+  }
+  if !actor.email_verified {
+    single = 1;
+    per_hour = 1;
+    per_day = 3;
+    per_week = 5;
+  } else if account_age < Duration::days(7) {
+    single = 3;
+    per_hour = 3;
+    per_day = 5;
+    per_week = 10;
+  } else if account_age < Duration::days(30) {
+    single = 5;
+    per_hour = 5;
+    per_day = 15;
+    per_week = 30;
+  } else {
+    let plan = quota.plan.to_ascii_lowercase();
+    if plan.contains("team") && !plan.contains("trial") {
+      single = quota.seat_limit.min(20);
+      per_hour = 20;
+      per_day = 50;
+    } else if plan.contains("trial") {
+      single = 10;
+      per_hour = 10;
+      per_day = 30;
+      per_week = 80;
+    } else if plan.contains("pro") {
+      single = 8;
+      per_hour = 8;
+      per_day = 20;
+      per_week = 50;
+    }
+  }
+
+  if workspace_age < Duration::hours(24) || quota.member_count <= 1 {
+    single = single.min(3);
+    per_day = per_day.min(10);
+  } else if workspace_age < Duration::days(7) || quota.member_count <= 5 {
+    single = single.min((single as f32 * 0.5).ceil() as i32).max(1);
+    per_hour = per_hour.min((per_hour as f32 * 0.5).ceil() as i32).max(1);
+    per_day = per_day.min((per_day as f32 * 0.5).ceil() as i32).max(1);
+  }
+
+  if activity.workspace_pending > quota.seat_limit.saturating_mul(2).max(6) {
+    single = single.min(2);
+    per_day = per_day.min(5);
+  }
+  if activity.actor_created_7d >= 5 && activity.actor_accepted_7d.saturating_mul(10) < activity.actor_created_7d {
+    single = single.min(2);
+    per_hour = per_hour.min(2);
+    per_day = per_day.min(5);
+    per_week = per_week.min(10);
+  }
+  if activity.workspace_created_7d >= 10
+    && activity.workspace_accepted_7d.saturating_mul(10) < activity.workspace_created_7d
+  {
+    per_day = per_day.min(10);
+    per_week = per_week.min(20);
+  }
+
+  let seat_7d = plan_ceiling_7d(&quota.plan, quota.seat_limit).min(quota.seat_limit.saturating_mul(2));
+  (single, per_hour, per_day, per_week.min(seat_7d))
+}
+
+pub(super) fn evaluate_projection(quota: &QuotaFacts, now: DateTime<Utc>) -> Option<&'static str> {
+  if !quota.known {
+    return Some("quota_state_unavailable");
+  }
+  if quota.stale || quota.stale_after.is_some_and(|stale_after| stale_after <= now) {
+    return Some("quota_projection_stale");
+  }
+  None
+}
+
+pub(super) fn build_invite_scopes(
+  input: &RuntimeWorkspaceInviteQuotaInput,
+  actor: &ActorFacts,
+  workspace: &WorkspaceFacts,
+  quota: &QuotaFacts,
+  activity: &InviteActivityFacts,
+  config: &InviteQuotaConfig,
+  now: DateTime<Utc>,
+) -> Result<Vec<ScopeLimit>> {
+  if input.target_count <= 0 {
+    return Err(napi_error("target_count must be positive"));
+  }
+  let (single, per_hour, per_day, per_week) = base_invite_limits(now, actor, workspace, quota, activity);
+  if input.target_count > single {
+    return Ok(vec![ScopeLimit {
+      scope_key: format!("invite:single_request:{}", input.actor_user_id),
+      window_seconds: 60,
+      bucket_seconds: bucket_seconds(60),
+      limit: single,
+      requested: input.target_count,
+    }]);
+  }
+
+  let actor_subject = subject_hash(&actor.email, config);
+  let (quota_subject, seat_limit) = quota_subject(&input.workspace_id, quota);
+  let quota_subject_7d = plan_ceiling_7d(&quota.plan, seat_limit).min(seat_limit.saturating_mul(2));
+  let mut scopes = vec![
+    scope(
+      format!("invite:user:{}", input.actor_user_id),
+      3600,
+      per_hour,
+      input.target_count,
+    ),
+    scope(
+      format!("invite:user:{}", input.actor_user_id),
+      86_400,
+      per_day,
+      input.target_count,
+    ),
+    scope(
+      format!("invite:user:{}", input.actor_user_id),
+      604_800,
+      per_week,
+      input.target_count,
+    ),
+    scope(
+      format!("invite:actor_subject:{actor_subject}"),
+      604_800,
+      per_week,
+      input.target_count,
+    ),
+    scope(
+      format!("invite:workspace:{}", input.workspace_id),
+      3600,
+      per_hour.max(5),
+      input.target_count,
+    ),
+    scope(
+      format!("invite:workspace:{}", input.workspace_id),
+      86_400,
+      per_day.max(10),
+      input.target_count,
+    ),
+    scope(
+      format!("invite:quota_subject:{quota_subject}"),
+      604_800,
+      quota_subject_7d,
+      input.target_count,
+    ),
+  ];
+
+  for target in &input.target_domains {
+    let domain = normalize_domain(&target.domain);
+    let limit_7d = if high_risk_domain(&domain, config) {
+      quota_subject_7d.min((seat_limit / 2).max(2))
+    } else {
+      quota_subject_7d
+    };
+    scopes.push(scope(
+      format!("invite:user_domain:{}:{domain}", input.actor_user_id),
+      86_400,
+      per_day.min(limit_7d).max(1),
+      target.count,
+    ));
+    scopes.push(scope(
+      format!("invite:workspace_domain:{}:{domain}", input.workspace_id),
+      86_400,
+      per_day.max(10).min(limit_7d).max(1),
+      target.count,
+    ));
+    scopes.push(scope(
+      format!("invite:quota_subject_domain:{quota_subject}:{domain}"),
+      604_800,
+      limit_7d,
+      target.count,
+    ));
+    scopes.push(scope(
+      format!("invite:target_domain_global:{domain}"),
+      60,
+      if high_risk_domain(&domain, config) { 30 } else { 100 },
+      target.count,
+    ));
+    scopes.push(scope(
+      format!("invite:target_domain_global:{domain}"),
+      86_400,
+      if high_risk_domain(&domain, config) { 500 } else { 2000 },
+      target.count,
+    ));
+  }
+
+  if let Some(prefix) = source_prefix(input.source.as_ref()) {
+    scopes.push(scope(
+      format!("invite:source_prefix:{prefix}"),
+      3600,
+      30,
+      input.target_count,
+    ));
+    scopes.push(scope(
+      format!("invite:source_prefix:{prefix}"),
+      86_400,
+      100,
+      input.target_count,
+    ));
+    for target in &input.target_domains {
+      let domain = normalize_domain(&target.domain);
+      scopes.push(scope(
+        format!("invite:source_prefix_domain:{prefix}:{domain}"),
+        3600,
+        if high_risk_domain(&domain, config) { 5 } else { 15 },
+        target.count,
+      ));
+      scopes.push(scope(
+        format!("invite:source_prefix_domain:{prefix}:{domain}"),
+        86_400,
+        if high_risk_domain(&domain, config) { 15 } else { 50 },
+        target.count,
+      ));
+    }
+  }
+
+  if input.source.as_ref().is_some_and(|source| source.trusted)
+    && let Some(asn) = input.source.as_ref().and_then(|source| source.asn)
+  {
+    for target in &input.target_domains {
+      let domain = normalize_domain(&target.domain);
+      scopes.push(scope(
+        format!("invite:source_asn_domain:{asn}:{domain}"),
+        3600,
+        if high_risk_domain(&domain, config) { 50 } else { 150 },
+        target.count,
+      ));
+      scopes.push(scope(
+        format!("invite:source_asn_domain:{asn}:{domain}"),
+        86_400,
+        if high_risk_domain(&domain, config) { 150 } else { 500 },
+        target.count,
+      ));
+    }
+  }
+
+  Ok(scopes)
+}
+
+pub(super) fn high_confidence_invite_abuse(
+  input: &RuntimeWorkspaceInviteQuotaInput,
+  actor: &ActorFacts,
+  config: &InviteQuotaConfig,
+) -> Option<InviteAbuseDecision> {
+  let high_risk_domain_counts: Vec<(String, i32)> = input
+    .target_domains
+    .iter()
+    .filter(|target| high_risk_domain(&target.domain, config))
+    .map(|target| (normalize_domain(&target.domain), target.count.max(0)))
+    .collect();
+  let high_risk_targets: i32 = high_risk_domain_counts.iter().map(|(_, count)| *count).sum();
+
+  if !actor.email_verified && high_risk_targets >= 3 {
+    return Some(InviteAbuseDecision {
+      reason: "unverified_high_risk_domain_burst",
+      action: "quarantine_actor",
+      subject_kind: "actor_email",
+      subject_key: subject_hash(&actor.email, config),
+    });
+  }
+
+  if input.target_count >= 30 && high_risk_targets * 2 >= input.target_count {
+    return Some(InviteAbuseDecision {
+      reason: "workspace_high_risk_domain_burst",
+      action: "quarantine_workspace",
+      subject_kind: "workspace",
+      subject_key: workspace_subject_key(&input.workspace_id),
+    });
+  }
+
+  if let Some(prefix) = source_prefix(input.source.as_ref())
+    && high_risk_targets >= 10
+    && high_risk_targets * 2 >= input.target_count
+    && let Some((domain, _)) = high_risk_domain_counts.iter().max_by_key(|(_, count)| *count)
+  {
+    return Some(InviteAbuseDecision {
+      reason: "source_high_risk_domain_burst",
+      action: "quarantine_source_cohort",
+      subject_kind: "source_prefix_domain",
+      subject_key: source_cohort_subject_key(&prefix, domain),
+    });
+  }
+
+  if input.target_count >= 10 && high_risk_targets * 2 >= input.target_count {
+    return Some(InviteAbuseDecision {
+      reason: "high_risk_domain_burst",
+      action: if input.target_count >= 20 {
+        "ban_actor"
+      } else {
+        "quarantine_actor"
+      },
+      subject_kind: "actor_email",
+      subject_key: subject_hash(&actor.email, config),
+    });
+  }
+
+  None
+}
+
+pub(super) fn sum_domains(target_domains: &[RuntimeQuotaTargetDomainInput]) -> BTreeMap<String, i32> {
+  let mut domains = BTreeMap::new();
+  for domain in target_domains {
+    *domains.entry(normalize_domain(&domain.domain)).or_insert(0) += domain.count.max(0);
+  }
+  domains
+}
+
+pub(super) fn invite_commit_usage_for_scope(
+  scope_key: &str,
+  target_count: i32,
+  domain_usage: &BTreeMap<String, i32>,
+) -> i32 {
+  for (domain, count) in domain_usage {
+    if scope_key.ends_with(&format!(":{domain}")) {
+      return *count;
+    }
+  }
+  target_count
+}
+
+#[cfg(test)]
+mod tests {
+  use chrono::TimeZone;
+
+  use super::*;
+
+  fn user(created_at: DateTime<Utc>) -> ActorFacts {
+    ActorFacts {
+      email: "actor@example.com".to_string(),
+      created_at,
+      registered: true,
+      email_verified: true,
+      disabled: false,
+    }
+  }
+
+  fn workspace(created_at: DateTime<Utc>) -> WorkspaceFacts {
+    WorkspaceFacts { created_at }
+  }
+
+  fn quota(plan: &str, seat_limit: i32) -> QuotaFacts {
+    QuotaFacts {
+      plan: plan.to_string(),
+      owner_user_id: Some("owner-1".to_string()),
+      uses_owner_quota: false,
+      seat_limit,
+      member_count: 10,
+      known: true,
+      stale: false,
+      stale_after: None,
+    }
+  }
+
+  fn invite_config() -> InviteQuotaConfig {
+    InviteQuotaConfig::default()
+  }
+
+  #[test]
+  fn seat_based_weekly_limit_binds_paid_team_and_high_risk_domain() {
+    let now = Utc.with_ymd_and_hms(2026, 7, 6, 0, 0, 0).single().unwrap();
+    let input = RuntimeWorkspaceInviteQuotaInput {
+      actor_user_id: "u1".to_string(),
+      workspace_id: "w1".to_string(),
+      request_id: None,
+      target_count: 5,
+      target_domains: vec![RuntimeQuotaTargetDomainInput {
+        domain: "qq.com".to_string(),
+        count: 5,
+      }],
+      source: None,
+    };
+    let scopes = build_invite_scopes(
+      &input,
+      &user(now - Duration::days(60)),
+      &workspace(now - Duration::days(60)),
+      &quota("paid_team", 10),
+      &InviteActivityFacts::default(),
+      &invite_config(),
+      now,
+    )
+    .unwrap();
+    let quota_subject = scopes
+      .iter()
+      .find(|scope| scope.scope_key == "invite:quota_subject:workspace:w1" && scope.window_seconds == 604_800)
+      .unwrap();
+    assert_eq!(quota_subject.limit, 20);
+    let high_risk = scopes
+      .iter()
+      .find(|scope| scope.scope_key == "invite:quota_subject_domain:workspace:w1:qq.com")
+      .unwrap();
+    assert_eq!(high_risk.limit, 5);
+  }
+
+  #[test]
+  fn owner_quota_subject_does_not_scale_by_workspace_count() {
+    let facts = QuotaFacts {
+      uses_owner_quota: true,
+      owner_user_id: Some("owner-a".to_string()),
+      ..quota("free", 3)
+    };
+    assert_eq!(quota_subject("w1", &facts), ("owner:owner-a".to_string(), 3));
+    assert_eq!(plan_ceiling_7d("free", 3).min(3 * 2), 6);
+  }
+
+  #[test]
+  fn low_acceptance_activity_reduces_invite_limits() {
+    let now = Utc.with_ymd_and_hms(2026, 7, 6, 0, 0, 0).single().unwrap();
+    let input = RuntimeWorkspaceInviteQuotaInput {
+      actor_user_id: "u1".to_string(),
+      workspace_id: "w1".to_string(),
+      request_id: None,
+      target_count: 2,
+      target_domains: vec![RuntimeQuotaTargetDomainInput {
+        domain: "example.com".to_string(),
+        count: 2,
+      }],
+      source: None,
+    };
+    let activity = InviteActivityFacts {
+      actor_created_7d: 10,
+      actor_accepted_7d: 0,
+      ..Default::default()
+    };
+
+    let scopes = build_invite_scopes(
+      &input,
+      &user(now - Duration::days(60)),
+      &workspace(now - Duration::days(60)),
+      &quota("paid_team", 10),
+      &activity,
+      &invite_config(),
+      now,
+    )
+    .unwrap();
+
+    let day = scopes
+      .iter()
+      .find(|scope| scope.scope_key == "invite:user:u1" && scope.window_seconds == 86_400)
+      .unwrap();
+    assert_eq!(day.limit, 5);
+  }
+
+  #[test]
+  fn invite_commit_usage_uses_domain_counts_for_domain_scopes() {
+    let domain_usage = sum_domains(&[
+      RuntimeQuotaTargetDomainInput {
+        domain: "Example.com".to_string(),
+        count: 2,
+      },
+      RuntimeQuotaTargetDomainInput {
+        domain: "qq.com".to_string(),
+        count: 1,
+      },
+    ]);
+
+    assert_eq!(
+      invite_commit_usage_for_scope("invite:user_domain:u1:example.com", 3, &domain_usage),
+      2
+    );
+    assert_eq!(
+      invite_commit_usage_for_scope("invite:quota_subject:workspace:w1", 3, &domain_usage),
+      3
+    );
+  }
+
+  #[test]
+  fn high_confidence_abuse_selects_workspace_and_source_subjects() {
+    let config = invite_config();
+    let mut input = RuntimeWorkspaceInviteQuotaInput {
+      actor_user_id: "u1".to_string(),
+      workspace_id: "w1".to_string(),
+      request_id: None,
+      target_count: 30,
+      target_domains: vec![RuntimeQuotaTargetDomainInput {
+        domain: "qq.com".to_string(),
+        count: 30,
+      }],
+      source: Some(RuntimeQuotaSourceInput {
+        trusted: true,
+        ip: Some("192.168.12.34".to_string()),
+        country: Some("US".to_string()),
+        asn: Some(13335),
+        ray_id: Some("ray".to_string()),
+      }),
+    };
+
+    let workspace_decision = high_confidence_invite_abuse(
+      &input,
+      &user(Utc.with_ymd_and_hms(2026, 7, 1, 0, 0, 0).single().unwrap()),
+      &config,
+    )
+    .unwrap();
+    assert_eq!(workspace_decision.action, "quarantine_workspace");
+    assert_eq!(workspace_decision.subject_kind, "workspace");
+
+    input.target_count = 12;
+    input.target_domains[0].count = 12;
+    let source_decision = high_confidence_invite_abuse(
+      &input,
+      &user(Utc.with_ymd_and_hms(2026, 7, 1, 0, 0, 0).single().unwrap()),
+      &config,
+    )
+    .unwrap();
+    assert_eq!(source_decision.action, "quarantine_source_cohort");
+    assert_eq!(source_decision.subject_kind, "source_prefix_domain");
+    assert!(
+      source_decision
+        .subject_key
+        .starts_with("source_prefix_domain_sha256:v1:")
+    );
+  }
+
+  #[test]
+  fn source_cohort_action_requires_trusted_source() {
+    let config = invite_config();
+    let input = RuntimeWorkspaceInviteQuotaInput {
+      actor_user_id: "u1".to_string(),
+      workspace_id: "w1".to_string(),
+      request_id: None,
+      target_count: 12,
+      target_domains: vec![RuntimeQuotaTargetDomainInput {
+        domain: "qq.com".to_string(),
+        count: 12,
+      }],
+      source: Some(RuntimeQuotaSourceInput {
+        trusted: false,
+        ip: Some("192.168.12.34".to_string()),
+        country: Some("US".to_string()),
+        asn: Some(13335),
+        ray_id: Some("ray".to_string()),
+      }),
+    };
+
+    let decision = high_confidence_invite_abuse(
+      &input,
+      &user(Utc.with_ymd_and_hms(2026, 7, 1, 0, 0, 0).single().unwrap()),
+      &config,
+    )
+    .unwrap();
+    assert_eq!(decision.action, "quarantine_actor");
+    assert_eq!(decision.subject_kind, "actor_email");
+  }
+}

packages/backend/native/src/runtime/backend_runtime/tests.rs

@@ -72,13 +72,129 @@ async fn runtime_from_database_url() -> AnyResult<Option<BackendRuntime>> {
     .execute(&pool)
     .await
     .context("cleanup runtime_leases for backend runtime tests")?;
+  sqlx::query("DELETE FROM runtime_rolling_quota_reservations WHERE request_id LIKE 'rust-test:%'")
+    .execute(&pool)
+    .await
+    .context("cleanup rolling quota reservations for backend runtime tests")?;
+  sqlx::query("DELETE FROM runtime_rolling_quota_counters WHERE scope_key LIKE 'invite:%rust-test%'")
+    .execute(&pool)
+    .await
+    .context("cleanup rolling quota counters for backend runtime tests")?;
+  sqlx::query("DELETE FROM runtime_invite_abuse_actions WHERE subject_key LIKE 'rust-test:%'")
+    .execute(&pool)
+    .await
+    .context("cleanup invite abuse actions for backend runtime tests")?;
+  sqlx::query("DELETE FROM runtime_invite_abuse_evidence WHERE subject_key LIKE 'rust-test:%'")
+    .execute(&pool)
+    .await
+    .context("cleanup invite abuse evidence for backend runtime tests")?;
+  sqlx::query(
+    "DELETE FROM runtime_invite_abuse_subjects WHERE subject_key LIKE 'rust-test:%' OR user_id LIKE 'rust-test:%'",
+  )
+  .execute(&pool)
+  .await
+  .context("cleanup invite abuse subjects for backend runtime tests")?;
 
   Ok(Some(BackendRuntime {
-    config: std::sync::RwLock::new(BackendRuntimeConfig { database_url }),
+    config: std::sync::RwLock::new(BackendRuntimeConfig {
+      database_url,
+      invite_quota: Default::default(),
+    }),
     pool: Mutex::new(Some(pool)),
   }))
 }
 
+async fn insert_invite_quota_fixture(
+  runtime: &BackendRuntime,
+  suffix: &str,
+  stale: bool,
+) -> AnyResult<(String, String)> {
+  let pool = runtime.pool().await.map_err(|err| anyhow!(err.to_string()))?;
+  let user_id = format!("rust-test:quota:user:{suffix}");
+  let workspace_id = format!("rust-test:quota:workspace:{suffix}");
+  let email = format!("rust-test-quota-{suffix}@example.com");
+
+  sqlx::query("DELETE FROM effective_workspace_quota_states WHERE workspace_id = $1")
+    .bind(&workspace_id)
+    .execute(&pool)
+    .await?;
+  sqlx::query("DELETE FROM workspaces WHERE id = $1")
+    .bind(&workspace_id)
+    .execute(&pool)
+    .await?;
+  sqlx::query("DELETE FROM users WHERE id = $1")
+    .bind(&user_id)
+    .execute(&pool)
+    .await?;
+  sqlx::query(
+    r#"
+    INSERT INTO users (id, name, email, registered, email_verified, disabled, created_at)
+    VALUES ($1, 'Rust Quota Actor', $2, true, clock_timestamp(), false, clock_timestamp() - interval '60 days')
+    "#,
+  )
+  .bind(&user_id)
+  .bind(email)
+  .execute(&pool)
+  .await?;
+  sqlx::query(
+    "INSERT INTO workspaces (id, public, created_at) VALUES ($1, false, clock_timestamp() - interval '60 days')",
+  )
+  .bind(&workspace_id)
+  .execute(&pool)
+  .await?;
+  sqlx::query(
+    r#"
+    INSERT INTO effective_workspace_quota_states (
+      workspace_id,
+      plan,
+      owner_user_id,
+      uses_owner_quota,
+      seat_limit,
+      member_count,
+      overcapacity_member_count,
+      blob_limit,
+      storage_quota,
+      used_storage_quota,
+      history_period_seconds,
+      readonly,
+      readonly_reasons,
+      flags,
+      known,
+      stale,
+      last_reconciled_at,
+      stale_after
+    )
+    VALUES ($1, 'paid_team', $2, false, 10, 3, 0, 1, 1, 0, 1, false, ARRAY[]::TEXT[], '{}'::jsonb, true, $3, clock_timestamp(), clock_timestamp() + interval '1 day')
+    "#,
+  )
+  .bind(&workspace_id)
+  .bind(&user_id)
+  .bind(stale)
+  .execute(&pool)
+  .await?;
+
+  Ok((user_id, workspace_id))
+}
+
+fn invite_quota_input(
+  user_id: &str,
+  workspace_id: &str,
+  request_id: &str,
+  count: i32,
+) -> types::RuntimeWorkspaceInviteQuotaInput {
+  types::RuntimeWorkspaceInviteQuotaInput {
+    actor_user_id: user_id.to_string(),
+    workspace_id: workspace_id.to_string(),
+    request_id: Some(request_id.to_string()),
+    target_count: count,
+    target_domains: vec![types::RuntimeQuotaTargetDomainInput {
+      domain: "example.com".to_string(),
+      count,
+    }],
+    source: None,
+  }
+}
+
 #[tokio::test]
 async fn runtime_gate_sql_semantics_are_atomic_and_ttl_bound() {
   let _guard = pg_test_lock().lock().await;
@@ -160,6 +276,278 @@ async fn runtime_gate_sql_semantics_are_atomic_and_ttl_bound() {
   assert_eq!(runtime.cleanup_expired_runtime_gates(100).await.unwrap(), 0);
 }
 
+#[tokio::test]
+async fn rolling_quota_sql_state_machine_commits_releases_and_expires() {
+  let _guard = pg_test_lock().lock().await;
+  let Some(runtime) = runtime_from_database_url().await.unwrap() else {
+    eprintln!("skipping postgres integration test: DATABASE_URL is not set");
+    return;
+  };
+  let (user_id, workspace_id) = insert_invite_quota_fixture(&runtime, "state-machine", false)
+    .await
+    .unwrap();
+  let pool = runtime.pool().await.unwrap();
+
+  let decision = runtime
+    .assert_workspace_invite_quota_v1(invite_quota_input(&user_id, &workspace_id, "rust-test:quota:commit", 2))
+    .await
+    .unwrap();
+  assert!(decision.allowed);
+  let reservation_id = decision.reservation_id.unwrap();
+  assert!(
+    runtime
+      .commit_workspace_invite_quota_v1(
+        reservation_id,
+        types::RuntimeWorkspaceInviteQuotaUsage {
+          target_count: 1,
+          target_domains: vec![types::RuntimeQuotaTargetDomainInput {
+            domain: "example.com".to_string(),
+            count: 1,
+          }],
+        },
+      )
+      .await
+      .unwrap()
+  );
+  let committed: i64 = sqlx::query_scalar(
+    r#"
+    SELECT COALESCE(SUM(count), 0)::bigint
+    FROM runtime_rolling_quota_counters
+    WHERE scope_key = $1
+    "#,
+  )
+  .bind(format!("invite:user_domain:{user_id}:example.com"))
+  .fetch_one(&pool)
+  .await
+  .unwrap();
+  assert_eq!(committed, 1);
+
+  let decision = runtime
+    .assert_workspace_invite_quota_v1(invite_quota_input(
+      &user_id,
+      &workspace_id,
+      "rust-test:quota:release",
+      1,
+    ))
+    .await
+    .unwrap();
+  let reservation_id = decision.reservation_id.unwrap();
+  assert!(
+    runtime
+      .release_workspace_invite_quota_v1(reservation_id.clone())
+      .await
+      .unwrap()
+  );
+  let released: String =
+    sqlx::query_scalar("SELECT status FROM runtime_rolling_quota_reservations WHERE id = $1::uuid LIMIT 1")
+      .bind(&reservation_id)
+      .fetch_one(&pool)
+      .await
+      .unwrap();
+  assert_eq!(released, "released");
+
+  let decision = runtime
+    .assert_workspace_invite_quota_v1(invite_quota_input(&user_id, &workspace_id, "rust-test:quota:expire", 1))
+    .await
+    .unwrap();
+  let reservation_id = decision.reservation_id.unwrap();
+  sqlx::query(
+    "UPDATE runtime_rolling_quota_reservations SET expires_at = clock_timestamp() - interval '1 second' WHERE id = \
+     $1::uuid",
+  )
+  .bind(&reservation_id)
+  .execute(&pool)
+  .await
+  .unwrap();
+  assert!(runtime.cleanup_expired_rolling_quota(100).await.unwrap() > 0);
+  let expired: String =
+    sqlx::query_scalar("SELECT status FROM runtime_rolling_quota_reservations WHERE id = $1::uuid LIMIT 1")
+      .bind(&reservation_id)
+      .fetch_one(&pool)
+      .await
+      .unwrap();
+  assert_eq!(expired, "expired");
+}
+
+#[tokio::test]
+async fn rolling_quota_projection_stale_fails_closed() {
+  let _guard = pg_test_lock().lock().await;
+  let Some(runtime) = runtime_from_database_url().await.unwrap() else {
+    eprintln!("skipping postgres integration test: DATABASE_URL is not set");
+    return;
+  };
+  let (user_id, workspace_id) = insert_invite_quota_fixture(&runtime, "stale-projection", true)
+    .await
+    .unwrap();
+
+  let decision = runtime
+    .assert_workspace_invite_quota_v1(invite_quota_input(&user_id, &workspace_id, "rust-test:quota:stale", 1))
+    .await
+    .unwrap();
+
+  assert!(!decision.allowed);
+  assert_eq!(decision.reason.as_deref(), Some("quota_projection_stale"));
+  assert!(decision.reservation_id.is_none());
+}
+
+#[tokio::test]
+async fn invite_abuse_action_sql_state_machine_retries_and_fences_workers() {
+  let _guard = pg_test_lock().lock().await;
+  let Some(runtime) = runtime_from_database_url().await.unwrap() else {
+    eprintln!("skipping postgres integration test: DATABASE_URL is not set");
+    return;
+  };
+  let pool = runtime.pool().await.unwrap();
+  let actor_id = "rust-test:invite-abuse-action:user";
+  let workspace_id = "rust-test:invite-abuse-action:workspace";
+  let subject_key = "rust-test:invite-abuse-action:subject";
+  let action_id: i64 = sqlx::query_scalar(
+    r#"
+    WITH subject AS (
+      INSERT INTO runtime_invite_abuse_subjects (
+        subject_key,
+        kind,
+        user_id,
+        actor_email_hash,
+        status,
+        first_seen_at,
+        last_seen_at
+      )
+      VALUES ($1, 'actor_email', $2, 'hash', 'quarantined', now(), now())
+      RETURNING subject_key
+    ),
+    evidence AS (
+      INSERT INTO runtime_invite_abuse_evidence (
+        subject_key,
+        workspace_id,
+        user_id,
+        actor_email_hash,
+        decision,
+        reason
+      )
+      VALUES ($1, $3, $2, 'hash', 'quarantine_actor', 'test')
+      RETURNING id
+    )
+    INSERT INTO runtime_invite_abuse_actions (
+      subject_key,
+      evidence_id,
+      action,
+      status
+    )
+    SELECT $1, evidence.id, 'quarantine_actor', 'pending'
+    FROM evidence
+    RETURNING id
+    "#,
+  )
+  .bind(subject_key)
+  .bind(actor_id)
+  .bind(workspace_id)
+  .fetch_one(&pool)
+  .await
+  .unwrap();
+
+  assert!(
+    runtime
+      .claim_invite_abuse_action(action_id.to_string(), "rust-test:inline-worker".to_string())
+      .await
+      .unwrap()
+  );
+  assert!(
+    runtime
+      .mark_invite_abuse_action(
+        action_id.to_string(),
+        "rust-test:inline-worker".to_string(),
+        "failed".to_string(),
+        Some("transient cleanup failure".to_string())
+      )
+      .await
+      .unwrap()
+  );
+
+  let waiting = sqlx::query(
+    r#"
+    SELECT status, attempts, last_error
+    FROM runtime_invite_abuse_actions
+    WHERE id = $1
+    "#,
+  )
+  .bind(action_id)
+  .fetch_one(&pool)
+  .await
+  .unwrap();
+  assert_eq!(waiting.get::<String, _>("status"), "retry_wait");
+  assert_eq!(waiting.get::<i32, _>("attempts"), 1);
+  assert_eq!(
+    waiting.get::<Option<String>, _>("last_error").as_deref(),
+    Some("transient cleanup failure")
+  );
+
+  sqlx::query(
+    r#"
+    UPDATE runtime_invite_abuse_actions
+    SET next_attempt_at = now() - interval '1 second'
+    WHERE id = $1
+    "#,
+  )
+  .bind(action_id)
+  .execute(&pool)
+  .await
+  .unwrap();
+
+  let claimed = runtime
+    .claim_retryable_invite_abuse_actions("rust-test:worker".to_string(), 10)
+    .await
+    .unwrap();
+  let current = claimed
+    .iter()
+    .find(|action| action.action_id == action_id.to_string())
+    .expect("retryable action should be claimed");
+  assert_eq!(current.action, "quarantine_actor");
+  assert_eq!(current.subject_key, subject_key);
+  assert_eq!(current.actor_user_id, actor_id);
+  assert_eq!(current.workspace_id, workspace_id);
+
+  assert!(
+    !runtime
+      .mark_invite_abuse_action(
+        action_id.to_string(),
+        "rust-test:inline-worker".to_string(),
+        "succeeded".to_string(),
+        None
+      )
+      .await
+      .unwrap()
+  );
+  let still_claimed = sqlx::query(
+    r#"
+    SELECT status, locked_by
+    FROM runtime_invite_abuse_actions
+    WHERE id = $1
+    "#,
+  )
+  .bind(action_id)
+  .fetch_one(&pool)
+  .await
+  .unwrap();
+  assert_eq!(still_claimed.get::<String, _>("status"), "running");
+  assert_eq!(
+    still_claimed.get::<Option<String>, _>("locked_by").as_deref(),
+    Some("rust-test:worker")
+  );
+
+  assert!(
+    runtime
+      .mark_invite_abuse_action(
+        action_id.to_string(),
+        "rust-test:worker".to_string(),
+        "succeeded".to_string(),
+        None
+      )
+      .await
+      .unwrap()
+  );
+}
+
 #[tokio::test]
 async fn coordination_lease_sql_semantics_are_fenced_and_ttl_bound() {
   let _guard = pg_test_lock().lock().await;

packages/backend/native/src/runtime/config.rs

@@ -1,4 +1,5 @@
 use std::{
+  collections::BTreeMap,
   env, fs,
   path::{Path, PathBuf},
 };
@@ -12,6 +13,35 @@ use super::{RuntimeError, RuntimeResult};
 #[derive(Clone, Debug)]
 pub(crate) struct BackendRuntimeConfig {
   pub(crate) database_url: String,
+  pub(crate) invite_quota: InviteQuotaConfig,
+}
+
+#[derive(Clone, Debug)]
+pub(crate) struct InviteQuotaConfig {
+  pub(crate) high_risk_target_domains: Vec<String>,
+  pub(crate) subject_hash_salt: String,
+  pub(crate) mail_class_mapping: BTreeMap<String, String>,
+}
+
+impl Default for InviteQuotaConfig {
+  fn default() -> Self {
+    Self {
+      high_risk_target_domains: [
+        "qq.com",
+        "proton.me",
+        "protonmail.com",
+        "163.com",
+        "126.com",
+        "outlook.com",
+        "hotmail.com",
+      ]
+      .into_iter()
+      .map(str::to_string)
+      .collect(),
+      subject_hash_salt: "affine-runtime-invite-quota-v1-local".to_string(),
+      mail_class_mapping: default_mail_class_mapping(),
+    }
+  }
 }
 
 impl BackendRuntimeConfig {
@@ -20,7 +50,10 @@ impl BackendRuntimeConfig {
     let database_url = database_url_from_env()
       .or(app_config.database_url())
       .unwrap_or_else(|| "postgresql://localhost:5432/affine".to_string());
-    Ok(Self { database_url })
+    Ok(Self {
+      database_url,
+      invite_quota: app_config.invite_quota_config(),
+    })
   }
 
   pub(crate) async fn with_db_overrides(&self, pool: &PgPool) -> RuntimeResult<Self> {
@@ -30,6 +63,7 @@ impl BackendRuntimeConfig {
       // The DB override is loaded after this connection already exists, so it
       // must not rewrite the active datasource URL.
       database_url: self.database_url.clone(),
+      invite_quota: app_config.invite_quota_config(),
     })
   }
 }
@@ -53,6 +87,10 @@ impl AppConfigFile {
       .and_then(|db| db.datasource_url.clone())
       .and_then(non_empty_string)
   }
+
+  fn invite_quota_config(&self) -> InviteQuotaConfig {
+    InviteQuotaConfig::default()
+  }
 }
 
 fn database_url_from_env() -> Option<String> {
@@ -86,6 +124,43 @@ impl AppConfigFile {
   }
 }
 
+fn default_mail_class_mapping() -> BTreeMap<String, String> {
+  [
+    ("SignIn", "auth"),
+    ("SignUp", "auth"),
+    ("SetPassword", "auth"),
+    ("ChangePassword", "auth"),
+    ("VerifyEmail", "auth"),
+    ("ChangeEmail", "auth"),
+    ("VerifyChangeEmail", "auth"),
+    ("EmailChanged", "auth"),
+    ("MemberInvitation", "workspace_invitation"),
+    ("Mention", "collaboration_notice"),
+    ("Comment", "collaboration_notice"),
+    ("CommentMention", "collaboration_notice"),
+    ("MemberAccepted", "collaboration_notice"),
+    ("LinkInvitationReviewRequest", "collaboration_notice"),
+    ("LinkInvitationApprove", "collaboration_notice"),
+    ("LinkInvitationDecline", "collaboration_notice"),
+    ("MemberLeave", "workspace_lifecycle"),
+    ("MemberRemoved", "workspace_lifecycle"),
+    ("OwnershipTransferred", "workspace_lifecycle"),
+    ("OwnershipReceived", "workspace_lifecycle"),
+    ("TeamWorkspaceUpgraded", "workspace_lifecycle"),
+    ("TeamBecomeAdmin", "workspace_lifecycle"),
+    ("TeamBecomeCollaborator", "workspace_lifecycle"),
+    ("TeamDeleteIn24Hours", "workspace_lifecycle"),
+    ("TeamDeleteInOneMonth", "workspace_lifecycle"),
+    ("TeamWorkspaceDeleted", "workspace_lifecycle"),
+    ("TeamWorkspaceExpireSoon", "workspace_lifecycle"),
+    ("TeamWorkspaceExpired", "workspace_lifecycle"),
+    ("TeamLicense", "billing_license"),
+  ]
+  .into_iter()
+  .map(|(mail_name, class)| (mail_name.to_string(), class.to_string()))
+  .collect()
+}
+
 async fn load_app_config_overrides_from_db(pool: &PgPool) -> RuntimeResult<AppConfigFile> {
   let rows = match sqlx::query("SELECT id, value FROM app_configs").fetch_all(pool).await {
     Ok(rows) => rows,
@@ -197,4 +272,21 @@ mod tests {
       Some("postgresql://example/runtime")
     );
   }
+
+  #[test]
+  fn invite_quota_policy_is_internal_not_app_configurable() {
+    let app_config = app_config_from_flat_overrides([
+      ("auth.untrustedPolicyOverride", serde_json::json!("runtime-salt-v2")),
+      ("auth.untrustedDomainList", serde_json::json!(["Example.COM."])),
+    ])
+    .unwrap();
+
+    let config = app_config.invite_quota_config();
+    assert!(!config.high_risk_target_domains.contains(&"example.com".to_string()));
+    assert_ne!(config.subject_hash_salt, "runtime-salt-v2");
+    assert_eq!(
+      config.mail_class_mapping.get("MemberInvitation").map(String::as_str),
+      Some("workspace_invitation")
+    );
+  }
 }

packages/backend/native/src/runtime/mod.rs

@@ -6,5 +6,5 @@ pub(crate) mod error;
 pub(crate) mod migrations;
 pub(crate) mod types;
 
-pub(crate) use config::BackendRuntimeConfig;
+pub(crate) use config::{BackendRuntimeConfig, InviteQuotaConfig};
 pub(crate) use error::{RuntimeError, RuntimeResult, napi_error, to_napi_error};

packages/backend/native/src/runtime/sql/runtime_migrations.sql

@@ -110,3 +110,90 @@ CREATE TABLE IF NOT EXISTS blob_cleanup_candidates (
 
 CREATE INDEX IF NOT EXISTS blob_cleanup_candidates_run_idx
   ON blob_cleanup_candidates (run_id, status);
+
+CREATE TABLE IF NOT EXISTS runtime_rolling_quota_counters (
+  scope_key TEXT NOT NULL,
+  window_seconds INTEGER NOT NULL,
+  bucket_start TIMESTAMPTZ(3) NOT NULL,
+  count INTEGER NOT NULL CHECK (count >= 0),
+  expires_at TIMESTAMPTZ(3) NOT NULL,
+  updated_at TIMESTAMPTZ(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  PRIMARY KEY (scope_key, window_seconds, bucket_start)
+);
+
+CREATE INDEX IF NOT EXISTS runtime_rolling_quota_counters_expires_at_idx
+  ON runtime_rolling_quota_counters (expires_at);
+
+CREATE TABLE IF NOT EXISTS runtime_rolling_quota_reservations (
+  id UUID NOT NULL DEFAULT gen_random_uuid(),
+  scope_key TEXT NOT NULL,
+  window_seconds INTEGER NOT NULL,
+  bucket_start TIMESTAMPTZ(3) NOT NULL,
+  count INTEGER NOT NULL CHECK (count >= 0),
+  status TEXT NOT NULL CHECK (status IN ('reserved', 'committed', 'released', 'expired')),
+  purpose TEXT NOT NULL,
+  request_id TEXT,
+  expires_at TIMESTAMPTZ(3) NOT NULL,
+  committed_at TIMESTAMPTZ(3),
+  released_at TIMESTAMPTZ(3),
+  created_at TIMESTAMPTZ(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  updated_at TIMESTAMPTZ(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  PRIMARY KEY (id, scope_key, window_seconds, bucket_start)
+);
+
+CREATE INDEX IF NOT EXISTS runtime_rolling_quota_reservations_status_expires_at_idx
+  ON runtime_rolling_quota_reservations (status, expires_at);
+
+CREATE INDEX IF NOT EXISTS runtime_rolling_quota_reservations_scope_idx
+  ON runtime_rolling_quota_reservations (scope_key, window_seconds, bucket_start)
+  WHERE status = 'reserved';
+
+CREATE TABLE IF NOT EXISTS runtime_invite_abuse_subjects (
+  subject_key TEXT PRIMARY KEY,
+  kind TEXT NOT NULL,
+  user_id TEXT,
+  actor_email_hash TEXT,
+  email_domain TEXT,
+  first_seen_at TIMESTAMPTZ(3) NOT NULL,
+  last_seen_at TIMESTAMPTZ(3) NOT NULL,
+  status TEXT NOT NULL DEFAULT 'active',
+  action TEXT,
+  action_reason TEXT,
+  action_at TIMESTAMPTZ(3),
+  created_at TIMESTAMPTZ(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  updated_at TIMESTAMPTZ(3) NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+
+CREATE TABLE IF NOT EXISTS runtime_invite_abuse_evidence (
+  id BIGSERIAL PRIMARY KEY,
+  subject_key TEXT NOT NULL REFERENCES runtime_invite_abuse_subjects(subject_key),
+  request_id TEXT,
+  workspace_id TEXT,
+  user_id TEXT,
+  actor_email_hash TEXT,
+  source_prefix_hash TEXT,
+  source_asn BIGINT,
+  target_domains JSONB NOT NULL DEFAULT '[]'::jsonb,
+  counters JSONB NOT NULL DEFAULT '{}'::jsonb,
+  decision TEXT NOT NULL,
+  reason TEXT NOT NULL,
+  created_at TIMESTAMPTZ(3) NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+
+CREATE TABLE IF NOT EXISTS runtime_invite_abuse_actions (
+  id BIGSERIAL PRIMARY KEY,
+  subject_key TEXT NOT NULL REFERENCES runtime_invite_abuse_subjects(subject_key),
+  evidence_id BIGINT NOT NULL REFERENCES runtime_invite_abuse_evidence(id),
+  action TEXT NOT NULL CHECK (action IN ('ban_actor', 'quarantine_actor', 'quarantine_workspace', 'quarantine_source_cohort')),
+  status TEXT NOT NULL CHECK (status IN ('pending', 'running', 'succeeded', 'retry_wait', 'failed')),
+  attempts INTEGER NOT NULL DEFAULT 0 CHECK (attempts >= 0),
+  next_attempt_at TIMESTAMPTZ(3),
+  locked_by TEXT,
+  locked_until TIMESTAMPTZ(3),
+  last_error TEXT,
+  created_at TIMESTAMPTZ(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  updated_at TIMESTAMPTZ(3) NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+
+CREATE INDEX IF NOT EXISTS runtime_invite_abuse_actions_status_next_attempt_idx
+  ON runtime_invite_abuse_actions (status, next_attempt_at);

packages/backend/native/src/runtime/types.rs

@@ -14,6 +14,107 @@ pub struct BackendRuntimeHealth {
   pub database_connected: bool,
 }
 
+#[napi_derive::napi(object)]
+pub struct RuntimeQuotaTargetDomainInput {
+  pub domain: String,
+  pub count: i32,
+}
+
+#[napi_derive::napi(object)]
+pub struct RuntimeQuotaSourceInput {
+  pub trusted: bool,
+  pub ip: Option<String>,
+  pub country: Option<String>,
+  pub asn: Option<u32>,
+  pub ray_id: Option<String>,
+}
+
+#[napi_derive::napi(object)]
+pub struct RuntimeWorkspaceInviteQuotaInput {
+  pub actor_user_id: String,
+  pub workspace_id: String,
+  pub request_id: Option<String>,
+  pub target_count: i32,
+  pub target_domains: Vec<RuntimeQuotaTargetDomainInput>,
+  pub source: Option<RuntimeQuotaSourceInput>,
+}
+
+#[napi_derive::napi(object)]
+pub struct RuntimeWorkspaceInviteQuotaUsage {
+  pub target_count: i32,
+  pub target_domains: Vec<RuntimeQuotaTargetDomainInput>,
+}
+
+#[napi_derive::napi(object)]
+pub struct RuntimeInviteAbuseActionRequired {
+  pub action: String,
+  pub subject_key: String,
+  pub evidence_id: String,
+  pub action_id: String,
+}
+
+#[napi_derive::napi(object)]
+pub struct RuntimeInviteAbuseClaimedAction {
+  pub action: String,
+  pub subject_key: String,
+  pub evidence_id: String,
+  pub action_id: String,
+  pub actor_user_id: String,
+  pub workspace_id: String,
+}
+
+#[napi_derive::napi(object)]
+pub struct RuntimeWorkspaceInviteQuotaDecision {
+  pub allowed: bool,
+  pub reservation_id: Option<String>,
+  pub retry_after_seconds: Option<i32>,
+  pub reason: Option<String>,
+  pub scope_key: Option<String>,
+  pub window_seconds: Option<i32>,
+  pub limit: Option<i32>,
+  pub current: Option<i32>,
+  pub requested: Option<i32>,
+  pub action_required: Option<RuntimeInviteAbuseActionRequired>,
+}
+
+#[napi_derive::napi(object)]
+pub struct RuntimeMailDeliveryQuotaMetadataInput {
+  pub actor_user_id: Option<String>,
+  pub workspace_id: Option<String>,
+  pub notification_id: Option<String>,
+  pub abuse_subject_key: Option<String>,
+}
+
+#[napi_derive::napi(object)]
+pub struct RuntimeMailDeliveryQuotaRecipientInput {
+  pub email: String,
+  pub domain: String,
+  pub user_id: Option<String>,
+}
+
+#[napi_derive::napi(object)]
+pub struct RuntimeMailDeliveryQuotaInput {
+  pub request_id: Option<String>,
+  pub mail_name: String,
+  pub recipient: RuntimeMailDeliveryQuotaRecipientInput,
+  pub metadata: RuntimeMailDeliveryQuotaMetadataInput,
+  pub source: Option<RuntimeQuotaSourceInput>,
+}
+
+#[napi_derive::napi(object)]
+pub struct RuntimeMailDeliveryQuotaDecision {
+  pub allowed: bool,
+  pub reservation_id: Option<String>,
+  pub mail_class: String,
+  pub retry_after_seconds: Option<i32>,
+  pub reason: Option<String>,
+  pub scope_key: Option<String>,
+  pub window_seconds: Option<i32>,
+  pub limit: Option<i32>,
+  pub current: Option<i32>,
+  pub requested: Option<i32>,
+}
+
 #[napi_derive::napi(object)]
 pub struct CoordinationLeaseGrant {
   pub key: String,