refactor(email): send mail through KenOAuth gRPC and drop the mails service

Deleted app/mails/src/main.rs +0 −79

@@ -1,79 +0,0 @@

1use std::net::SocketAddr;

2use std::time::Duration;

3

4use anyhow::Context;

5use config::scoped::{MailsConfig, ScopedConfig};

6use email::runtime::{WorkerOptions, run_until_shutdown};

7use tokio::net::TcpListener;

8use tracing::info;

9

10#[tokio::main]

11async fn main() -> anyhow::Result<()> {

12 export::telemetry::init_tracing();

13

14 let cfg = MailsConfig::load().context("loading configuration")?;

15

16 let redis_url = cfg.redis.url().to_string();

17 let redis_host = match redis_url.rsplit_once('@') {

18 Some((_, host)) => host.to_string(),

19 None => redis_url

20 .split_once("://")

21 .map(|x| x.1)

22 .unwrap_or("?")

23 .to_string(),

24 };

25

26 info!(

27 smtp_enabled = cfg.email.enabled(),

28 queue_enabled = cfg.email_queue.enabled(),

29 redis = %redis_host,

30 shutdown_timeout_secs = 5,

31 "starting mails worker"

32 );

33

34 if let Err(error) = export::telemetry::init_meter_provider() {

35 tracing::warn!(%error, "failed to init meter provider; metrics disabled");

36 }

37

38 let opts = WorkerOptions {

39 template_dir: None,

40 shutdown_timeout: Some(Duration::from_secs(5)),

41 };

42

43 let (health_tx, health_rx) = tokio::sync::oneshot::channel::<()>();

44 let mut health_handle = None;

45 if cfg.health.enabled() {

46 let redis = export::AppRedis::connect(cfg.redis.clone())

47 .await

48 .context("connecting to redis")?;

49 let addr: SocketAddr = format!("{}:{}", cfg.health.bind_addr(), cfg.health.mails_port())

50 .parse()

51 .context("parsing health bind address")?;

52 let listener = TcpListener::bind(addr)

53 .await

54 .with_context(|| format!("binding health {addr}"))?;

55 info!(%addr, "mails health listening");

56 let router = export::health::router_with(None, redis, None);

57 health_handle = Some(tokio::spawn(async move {

58 let _ = axum::serve(listener, router)

59 .with_graceful_shutdown(async {

60 let _ = health_rx.await;

61 })

62 .await;

63 }));

64 }

65

66 let result =

67 run_until_shutdown(cfg.email.clone(), cfg.email_queue.clone(), redis_url, opts).await;

68

69 drop(health_tx);

70 if let Some(handle) = health_handle {

71 let _ = handle.await;

72 }

73

74 result.map_err(|e| anyhow::anyhow!(e.to_string()))?;

75

76 export::telemetry::shutdown_telemetry();

77 info!("mails worker exited cleanly");

78 Ok(())

79}

Deleted app/mails/Cargo.toml +0 −15

@@ -1,15 +0,0 @@

1[package]

2name = "mails"

3version.workspace = true

4edition.workspace = true

5publish = false

6

7[dependencies]

8email = { workspace = true }

9export = { workspace = true }

10config = { workspace = true }

11axum = { workspace = true }

12tokio = { workspace = true, features = ["signal", "net", "rt-multi-thread", "macros", "sync"] }

13tracing = { workspace = true }

14tracing-subscriber = { workspace = true }

15anyhow = { workspace = true }

No newline at end of file

Deleted charts/core/templates/mails.yaml +0 −61

@@ -1,61 +0,0 @@

1{{- if .Values.mails.enabled }}

2apiVersion: apps/v1

3kind: Deployment

4metadata:

5 name: {{ printf "%s-mails" (include "gitdata.fullname" .) }}

6 labels:

7 {{- include "gitdata.labels" . | nindent 4 }}

8 app.kubernetes.io/component: mails

9spec:

10 replicas: {{ .Values.mails.replicaCount }}

11 revisionHistoryLimit: 3

12 selector:

13 matchLabels:

14 {{- include "gitdata.selectorLabels" . | nindent 6 }}

15 app.kubernetes.io/component: mails

16 template:

17 metadata:

18 labels:

19 {{- include "gitdata.selectorLabels" . | nindent 8 }}

20 app.kubernetes.io/component: mails

21 {{- with .Values.podLabels }}

22 {{- toYaml . | nindent 8 }}

23 {{- end }}

24 annotations:

25 checksum/env: {{ toYaml .Values.config.env | sha256sum }}

26 {{- with .Values.podAnnotations }}

27 {{- toYaml . | nindent 8 }}

28 {{- end }}

29 spec:

30 serviceAccountName: {{ include "gitdata.serviceAccountName" . }}

31 {{- include "gitdata.imagePullSecrets" . | nindent 6 }}

32 securityContext:

33 {{- toYaml .Values.podSecurityContext | nindent 8 }}

34 containers:

35 - name: mails

36 image: {{ include "gitdata.image" . }}

37 imagePullPolicy: {{ .Values.image.pullPolicy }}

38 args: ["mails"]

39 {{- include "gitdata.envFrom" . | nindent 10 }}

40 env:

41 {{- include "gitdata.env" . | nindent 12 }}

42 {{- with .Values.mails.extraEnv }}

43 {{- toYaml . | nindent 12 }}

44 {{- end }}

45 ports:

46 - name: health

47 containerPort: {{ .Values.health.mailsPort }}

48 livenessProbe:

49 httpGet: { path: /healthz, port: health }

50 initialDelaySeconds: 10

51 periodSeconds: 15

52 readinessProbe:

53 httpGet: { path: /readyz, port: health }

54 initialDelaySeconds: 5

55 periodSeconds: 10

56 failureThreshold: 3

57 securityContext:

58 {{- toYaml .Values.securityContext | nindent 12 }}

59 resources:

60 {{- toYaml .Values.mails.resources | nindent 12 }}

61{{- end }}

Modified charts/core/Chart.yaml +1 −1 View charts/core/Chart.yaml

@@ -1,6 +1,6 @@

11apiVersion: v2

22name: gitdata

3description: GitData — Rust workspace (apihost / githost / synchost / mails / migrate)

3description: GitData — Rust workspace (apihost / githost / synchost / migrate)

44type: application

55version: 0.1.0

66appVersion: "1.0.0"

Modified charts/core/values.yaml +1 −1 View charts/core/values.yaml

@@ -28,7 +28,7 @@

2828 APP_S3__BUCKET: ""

2929 APP_S3__REGION: "us-east-1"

3030 APP_S3__PATH_STYLE: "true"

31 # 读写库/副本、邮件、配额等其余段均可在此覆盖，例如：

31 # 读写库/副本、配额等其余段均可在此覆盖，例如：

3232 # APP_DATABASE__REPLICAS: '[{url="postgres://readonly@pg:5432/gitdata"}]'

3333 APP_SEARCH__ENABLED: "false"

3434 APP_SEARCH__URL: "http://meilisearch:7700"

Modified charts/prod/values/prod.yaml +0 −2 View charts/prod/values/prod.yaml

@@ -29,8 +29,6 @@

2929 APP_OAUTH__COOKIE_SECURE: "true"

3030 APP_GIT__BASE_URL: "https://gitdata.ai"

3131 APP_LIVEKIT__RECORDING_ENABLED: "false"

32 APP_EMAIL__SMTP_USERNAME: "bot@mail.gitdata.ai"

33 APP_EMAIL__FROM_ADDRESS: "bot@mail.gitdata.ai"

3432 # CI 开关：runner 注册走 /api/v1/runner/register，该处理函数第一件事就是

3533 # ensure_enabled()（lib/ci/src/service.rs:46），关闭时直接返回 403 Forbidden。

3634 # 开启后必须同时提供 APP_CI__SECRET_ENCRYPTION_KEY（已在 gitdata-env Secret 中）。

Modified config/default.toml +3 −16 View config/default.toml

@@ -236,25 +236,12 @@

236236keepalive_max = 3

237237

238238[email]

239enabled = false

240smtp_host = "smtp.example.com"

241smtp_port = 587

242smtp_username = ""

243smtp_password = ""

244from_address = "[email protected]"

245from_name = "GitData"

246connection_timeout_ms = 5000

247response_timeout_ms = 30000

248max_attachment_bytes = 26214400

249239default_locale = "en"

250template_dir = "config/email_templates"

251240

252enabled = false

253channel_prefix = "email"

254subscriber_group_id = "email-worker"

255max_retries = 3

256retry_backoff_ms = 500

257241[email.rpc]

242endpoint = ""

243rpc_key = ""

244timeout_ms = 2000

258245

259246[apihost]

260247enabled = true

Modified docs/oauth.md +1 −1 View docs/oauth.md

@@ -119,7 +119,7 @@

119119 PostgreSQL(gitdata) + Redis + RustFS + Meilisearch

120120 ▲

121121 │

122 synchost / searchhost / im / webhook / aihost / mails / mcp

122 synchost / searchhost / im / webhook

123123```

124124

125125### 3.2 令牌流转

Modified lib/auth/src/services/mod.rs +1 −15 View lib/auth/src/services/mod.rs

@@ -55,20 +55,6 @@

5555 }

5656 }

5757

58 pub async fn send_email(

59 &self,

60 to: &str,

61 subject: &str,

62 text: &str,

63 html: &str,

64 ) -> Result<(), AuthError> {

65 self.state

66 .email

67 .send_mail(to, subject, text, html)

68 .await

69 .map_err(|error| AuthError::Internal(format!("email send failed: {error}")))

70 }

71

7258 pub async fn verify_access_token(

7359 &self,

7460 token: &str,

@@ -91,4 +77,4 @@

9177 },

9278 }))

9379 }

94}

No newline at end of file

80}

Modified lib/auth/src/templates/mod.rs +0 −38 View lib/auth/src/templates/mod.rs

@@ -1,42 +1,4 @@

11mod common;

22mod page;

33

4use axum::response::Html;

5pub(crate) use common::page_shell;

6

7/// Thin wrappers over the `email` templates: the mail crate owns the markup and

8/// the catalogue, this crate owns the data the request happened to have.

9pub fn mail_verify_code(

10 code: &str,

11 minutes: u64,

12 purpose: &str,

13 link: Option<&str>,

14 locale: i18n::UiLocale,

15) -> Html<String> {

16 Html(email::templates::mail_verify_code(code, minutes, purpose, link, locale).into_string())

17}

18

19pub fn mail_reset_link(

20 email: &str,

21 token: &str,

22 temp_id: &str,

23 base_url: &str,

24 locale: i18n::UiLocale,

25) -> Html<String> {

26 Html(email::templates::mail_reset_link(email, token, temp_id, base_url, locale).into_string())

27}

28

29pub fn mail_email_change_confirm(

30 new_email: &str,

31 ticket_id: &str,

32 code: &str,

33 base_url: &str,

34 locale: i18n::UiLocale,

35) -> Html<String> {

36 Html(

37 email::templates::mail_email_change_confirm(new_email, ticket_id, code, base_url, locale)

38 .into_string(),

39 )

40}

41

424pub use page::{page_agreement, page_message, page_verify_result, render_markdown};

Modified lib/auth/Cargo.toml +0 −1 View lib/auth/Cargo.toml

@@ -27,7 +27,6 @@

2727aes-gcm = { workspace = true }

2828rand = { workspace = true }

2929tracing = { workspace = true }

30email = { workspace = true }

3130kenauth = { workspace = true }

3231reqwest = { workspace = true }

3332url = { workspace = true }

Modified lib/chat/src/services/service.rs +5 −2 View lib/chat/src/services/service.rs

@@ -95,8 +95,11 @@

9595 text: &str,

9696 html: &str,

9797 ) -> Result<(), ChatError> {

98 self.state

99 .email

98 let Some(client) = self.state.email_rpc.as_ref() else {

99 tracing::warn!(to, "email rpc disabled; skipping chat mail");

100 return Ok(());

101 };

102 client

100103 .send_mail(to, subject, text, html)

101104 .await

102105 .map_err(|error| ChatError::Internal(format!("email send failed: {error}")))

Modified lib/chat/Cargo.toml +1 −0 View lib/chat/Cargo.toml

@@ -13,6 +13,7 @@

1313search = { workspace = true }

1414social = { workspace = true }

1515jwt = { workspace = true }

16kenauth = { workspace = true }

1617space = { workspace = true }

1718storage = { workspace = true }

1819axum = { workspace = true }

Modified lib/ci/examples/actions_fixture.rs +6 −8 View lib/ci/examples/actions_fixture.rs

@@ -58,8 +58,6 @@

5858 config.s3.access_key_id = Some("fixture-access".into());

5959 config.s3.secret_access_key = Some("fixture-secret-password".into());

6060 config.s3.path_style = Some(true);

61 config.email.enabled = Some(false);

62 config.email.from_address = Some("fixture@localhost".into());

6361 config.ci.enabled = true;

6462 config.ci.secret_encryption_key = "QUFBQUFBQUFBQUFBQUFBQUFBQUFBQUFBQUFBQUFBQUE=".into();

6563 // Built by the git-data-actions repository now; override with ACTIONS_ENGINE_BINARY.

@@ -225,7 +223,7 @@

225223 service: &CiService,

226224 repo: &repos::Model,

227225 space: &spaces::Model,

228 owner: Uuid,

226 owner: i64,

229227 source: &str,

230228) -> anyhow::Result<Uuid> {

231229 let actions = service

@@ -288,13 +286,13 @@

288286 anyhow::bail!("fixture run did not reach a terminal state")

289287}

290288

291async fn seed(service: &CiService) -> anyhow::Result<(repos::Model, spaces::Model, Uuid, Uuid)> {

292 let owner = Uuid::now_v7();

293 let reviewer = Uuid::now_v7();

289async fn seed(service: &CiService) -> anyhow::Result<(repos::Model, spaces::Model, i64, i64)> {

290 let owner = 1001i64;

291 let reviewer = 1002i64;

294292 let now = OffsetDateTime::now_utc();

295293 for (id, name) in [

296 (owner, format!("fixture-{}", owner.simple())),

297 (reviewer, format!("fixture-{}", reviewer.simple())),

294 (owner, format!("fixture-{owner}")),

295 (reviewer, format!("fixture-{reviewer}")),

298296 ] {

299297 models::users::user::ActiveModel {

300298 uid: Set(id),

Modified lib/config/src/email.rs +16 −86 View lib/config/src/email.rs

@@ -1,102 +1,32 @@

1use std::path::PathBuf;

2

31use serde::{Deserialize, Serialize};

42

53#[derive(Debug, Clone, Default, Serialize, Deserialize)]

64#[serde(default)]

7pub struct EmailConfig {

8 pub enabled: Option<bool>,

9 pub smtp_host: Option<String>,

10 pub smtp_port: Option<u16>,

11 pub smtp_username: Option<String>,

12 pub smtp_password: Option<String>,

13 pub from_address: Option<String>,

14 pub from_name: Option<String>,

15 pub connection_timeout_ms: Option<u64>,

16 pub response_timeout_ms: Option<u64>,

17 pub max_attachment_bytes: Option<usize>,

18 pub default_locale: Option<String>,

19 pub template_dir: Option<PathBuf>,

5pub struct EmailRpcConfig {

6 pub endpoint: String,

7 pub rpc_key: String,

8 pub timeout_ms: Option<u64>,

209}

2110

22impl EmailConfig {

11impl EmailRpcConfig {

2312 pub fn enabled(&self) -> bool {

24 self.enabled.unwrap_or(true)

25 }

26

27 pub fn smtp_host(&self) -> &str {

28 self.smtp_host.as_deref().unwrap_or("")

29 }

30

31 pub fn smtp_port(&self) -> u16 {

32 self.smtp_port.unwrap_or(587)

33 }

34

35 pub fn smtp_username(&self) -> Option<&str> {

36 self.smtp_username.as_deref().filter(|s| !s.is_empty())

13 !self.endpoint.is_empty()

3714 }

3815

39 pub fn smtp_password(&self) -> Option<&str> {

40 self.smtp_password.as_deref().filter(|s| !s.is_empty())

16 pub fn timeout_ms(&self) -> u64 {

17 self.timeout_ms.unwrap_or(2000)

4118 }

19}

4220

43 pub fn from_address(&self) -> Option<&str> {

44 self.from_address.as_deref().filter(|s| !s.is_empty())

45 }

21#[derive(Debug, Clone, Default, Serialize, Deserialize)]

22#[serde(default)]

23pub struct EmailConfig {

24 pub default_locale: Option<String>,

25 pub rpc: EmailRpcConfig,

26}

4627

47 pub fn from_name(&self) -> Option<&str> {

48 self.from_name.as_deref().filter(|s| !s.is_empty())

49 }

50

51 pub fn connection_timeout_ms(&self) -> u64 {

52 self.connection_timeout_ms.unwrap_or(5_000)

53 }

54

55 pub fn response_timeout_ms(&self) -> u64 {

56 self.response_timeout_ms.unwrap_or(30_000)

57 }

58

59 pub fn max_attachment_bytes(&self) -> usize {

60 self.max_attachment_bytes.unwrap_or(26_214_400)

61 }

62

28impl EmailConfig {

6329 pub fn default_locale(&self) -> &str {

6430 self.default_locale.as_deref().unwrap_or("en")

65 }

66

67 pub fn template_dir(&self) -> Option<&PathBuf> {

68 self.template_dir.as_ref()

69 }

70

71 pub fn is_valid(&self) -> Result<(), String> {

72 if !self.enabled() {

73 return Ok(());

74 }

75 let host = self.smtp_host();

76 if host.is_empty() {

77 return Err("email.smtp_host must be set when enabled".to_string());

78 }

79 let port = self.smtp_port();

80 if port == 0 {

81 return Err("email.smtp_port must be greater than 0".to_string());

82 }

83 match self.from_address() {

84 None => return Err("email.from_address must be set when enabled".to_string()),

85 Some(addr) => {

86 if !addr.contains('@') {

87 return Err(format!("email.from_address is invalid: {addr:?}"));

88 }

89 }

90 }

91 if self.from_name.as_deref().is_some_and(|s| s.is_empty()) {

92 return Err("email.from_name must not be empty".to_string());

93 }

94 if matches!(self.connection_timeout_ms, Some(0)) {

95 return Err("email.connection_timeout_ms must be greater than 0".to_string());

96 }

97 if matches!(self.response_timeout_ms, Some(0)) {

98 return Err("email.response_timeout_ms must be greater than 0".to_string());

99 }

100 Ok(())

10131 }

10232}

Deleted lib/config/src/email_queue.rs +0 −66

@@ -1,66 +0,0 @@

1use std::time::Duration;

2

3use serde::{Deserialize, Serialize};

4

5#[derive(Debug, Clone, Default, Serialize, Deserialize)]

6#[serde(default)]

7pub struct EmailQueueConfig {

8 pub enabled: Option<bool>,

9 pub channel_prefix: Option<String>,

10 pub subscriber_group_id: Option<String>,

11 pub max_retries: Option<u32>,

12 pub retry_backoff_ms: Option<u64>,

13}

14

15impl EmailQueueConfig {

16 pub fn enabled(&self) -> bool {

17 self.enabled.unwrap_or(true)

18 }

19

20 pub fn channel_prefix(&self) -> &str {

21 self.channel_prefix.as_deref().unwrap_or("email")

22 }

23

24 pub fn subscriber_group_id(&self) -> &str {

25 self.subscriber_group_id

26 .as_deref()

27 .unwrap_or("email-worker")

28 }

29

30 pub fn max_retries(&self) -> u32 {

31 self.max_retries.unwrap_or(3)

32 }

33

34 pub fn retry_backoff(&self) -> Duration {

35 Duration::from_millis(self.retry_backoff_ms.unwrap_or(500))

36 }

37

38 pub fn channel(&self, name: &str) -> String {

39 format!("{}.{}", self.channel_prefix(), name)

40 }

41

42 pub fn is_valid(&self) -> Result<(), String> {

43 if !self.enabled() {

44 return Ok(());

45 }

46 if self.channel_prefix.as_deref().is_some_and(|s| s.is_empty()) {

47 return Err("email_queue.channel_prefix must not be empty".to_string());

48 }

49 if self

50 .subscriber_group_id

51 .as_deref()

52 .is_some_and(|s| s.is_empty())

53 {

54 return Err(

55 "email_queue.subscriber_group_id must not be empty when enabled".to_string(),

56 );

57 }

58 if matches!(self.max_retries, Some(0)) {

59 return Err("email_queue.max_retries must be greater than 0".to_string());

60 }

61 if matches!(self.retry_backoff_ms, Some(0)) {

62 return Err("email_queue.retry_backoff_ms must be greater than 0".to_string());

63 }

64 Ok(())

65 }

66}

Modified lib/config/src/health.rs +1 −9 View lib/config/src/health.rs

@@ -9,7 +9,7 @@

99}

1010

1111/// Health endpoints for binaries without a business HTTP listener

12/// (synchost, mails). Apps that already serve HTTP (apihost, githost, im)

12/// (synchost). Apps that already serve HTTP (apihost, githost, im)

1313/// expose /healthz + /readyz on their own port instead.

1414#[derive(Debug, Clone, Default, Serialize, Deserialize)]

1515#[serde(default)]

@@ -17,7 +17,6 @@

1717 pub enabled: Option<bool>,

1818 pub bind_addr: Option<String>,

1919 pub synchost: HealthServiceConfig,

20 pub mails: HealthServiceConfig,

2120}

2221

2322impl HealthConfig {

@@ -33,10 +32,6 @@

3332 self.synchost.port.unwrap_or(3002)

3433 }

3534

36 pub fn mails_port(&self) -> u16 {

37 self.mails.port.unwrap_or(3003)

38 }

39

4035 pub fn is_valid(&self) -> Result<(), String> {

4136 if !self.enabled() {

4237 return Ok(());

@@ -47,9 +42,6 @@

4742 }

4843 if self.synchost.port == Some(0) {

4944 return Err("health.synchost.port must be greater than 0".to_string());

50 }

51 if self.mails.port == Some(0) {

52 return Err("health.mails.port must be greater than 0".to_string());

5345 }

5446 Ok(())

5547 }

Modified lib/config/src/lib.rs +0 −7 View lib/config/src/lib.rs

@@ -4,7 +4,6 @@

44pub mod ci;

55pub mod database;

66pub mod email;

7pub mod email_queue;

87pub mod git;

98pub mod githost;

109pub mod health;

@@ -38,7 +37,6 @@

3837use crate::ci::CiConfig;

3938use crate::database::DatabaseConfig;

4039use crate::email::EmailConfig;

41use crate::email_queue::EmailQueueConfig;

4240use crate::git::GitConfig;

4341use crate::githost::GithostConfig;

4442use crate::health::HealthConfig;

@@ -80,7 +78,6 @@

8078 pub oauth: OAuthConfig,

8179 pub git: GitConfig,

8280 pub email: EmailConfig,

83 pub email_queue: EmailQueueConfig,

8481 pub apihost: ApihostConfig,

8582 pub githost: GithostConfig,

8683 pub im: ImServerConfig,

@@ -103,10 +100,6 @@

103100 self.pow.validate().map_err(ConfigError::Validation)?;

104101 self.auth.validate().map_err(ConfigError::Validation)?;

105102 self.chat.is_valid().map_err(ConfigError::Validation)?;

106 self.email.is_valid().map_err(ConfigError::Validation)?;

107 self.email_queue

108 .is_valid()

109 .map_err(ConfigError::Validation)?;

110103 self.search.validate().map_err(ConfigError::Validation)?;

111104 self.import.validate().map_err(ConfigError::Validation)?;

112105 self.apihost.is_valid().map_err(ConfigError::Validation)?;

Modified lib/config/src/scoped.rs +15 −36 View lib/config/src/scoped.rs

@@ -7,7 +7,6 @@

77use crate::ci::CiConfig;

88use crate::database::DatabaseConfig;

99use crate::email::EmailConfig;

10use crate::email_queue::EmailQueueConfig;

1110use crate::error::ConfigError;

1211use crate::git::GitConfig;

1312use crate::githost::GithostConfig as GithostSection;

@@ -16,6 +15,7 @@

1615use crate::import::ImportConfig;

1716use crate::issues::IssueConfig;

1817use crate::livekit::LiveKitConfig;

18use crate::oauth::OAuthConfig;

1919use crate::pow::PowConfig;

2020use crate::push::PushConfig;

2121use crate::quota::QuotaConfig;

@@ -47,24 +47,6 @@

4747impl ScopedConfig for MigrateConfig {

4848 fn validate_scoped(&self) -> Result<(), String> {

4949 self.database.validate()

50 }

51}

52

53#[derive(Clone, Debug, Default, Deserialize)]

54#[serde(default)]

55pub struct MailsConfig {

56 pub redis: RedisConfig,

57 pub email: EmailConfig,

58 pub email_queue: EmailQueueConfig,

59 pub health: HealthConfig,

60}

61

62impl ScopedConfig for MailsConfig {

63 fn validate_scoped(&self) -> Result<(), String> {

64 self.redis.validate()?;

65 self.email.is_valid()?;

66 self.email_queue.is_valid()?;

67 self.health.is_valid()

6850 }

6951}

7052

@@ -75,7 +57,7 @@

7557 pub redis: RedisConfig,

7658 pub s3: S3Config,

7759 pub email: EmailConfig,

78 pub email_queue: EmailQueueConfig,

60 pub oauth: OAuthConfig,

7961 pub web: WebConfig,

8062 pub git: GitConfig,

8163 pub ci: CiConfig,

@@ -90,7 +72,7 @@

9072 redis: self.redis,

9173 s3: self.s3,

9274 email: self.email,

93 email_queue: self.email_queue,

75 oauth: self.oauth,

9476 web: self.web,

9577 git: self.git,

9678 ci: self.ci,

@@ -105,10 +87,9 @@

10587 fn validate_scoped(&self) -> Result<(), String> {

10688 self.database.validate()?;

10789 self.redis.validate()?;

108 self.email.is_valid()?;

109 self.email_queue.is_valid()?;

11090 self.ci.validate()?;

11191 self.webhook.is_valid()?;

92 self.oauth.is_valid()?;

11293 self.health.is_valid()

11394 }

11495}

@@ -141,7 +122,7 @@

141122 pub redis: RedisConfig,

142123 pub s3: S3Config,

143124 pub email: EmailConfig,

144 pub email_queue: EmailQueueConfig,

125 pub oauth: OAuthConfig,

145126 pub web: WebConfig,

146127 pub git: GitConfig,

147128 pub githost: GithostSection,

@@ -155,7 +136,7 @@

155136 redis: self.redis,

156137 s3: self.s3,

157138 email: self.email,

158 email_queue: self.email_queue,

139 oauth: self.oauth,

159140 web: self.web,

160141 git: self.git,

161142 githost: self.githost,

@@ -169,9 +150,9 @@

169150 fn validate_scoped(&self) -> Result<(), String> {

170151 self.database.validate()?;

171152 self.redis.validate()?;

172 self.email.is_valid()?;

173 self.email_queue.is_valid()?;

174 self.githost.is_valid()

153 self.githost.is_valid()?;

154 self.oauth.is_valid()?;

155 Ok(())

175156 }

176157}

177158

@@ -193,7 +174,7 @@

193174 pub issues: IssueConfig,

194175 pub git: GitConfig,

195176 pub email: EmailConfig,

196 pub email_queue: EmailQueueConfig,

177 pub oauth: OAuthConfig,

197178 pub apihost: ApihostSection,

198179 pub search: SearchConfig,

199180 pub import: ImportConfig,

@@ -219,7 +200,7 @@

219200 issues: self.issues,

220201 git: self.git,

221202 email: self.email,

222 email_queue: self.email_queue,

203 oauth: self.oauth,

223204 apihost: self.apihost,

224205 search: self.search,

225206 import: self.import,

@@ -238,12 +219,11 @@

238219 self.pow.validate()?;

239220 self.auth.validate()?;

240221 self.chat.is_valid()?;

241 self.email.is_valid()?;

242 self.email_queue.is_valid()?;

243222 self.search.validate()?;

244223 self.import.validate()?;

245224 self.apihost.is_valid()?;

246225 self.livekit.is_valid()?;

226 self.oauth.is_valid()?;

247227 self.push.is_valid()

248228 }

249229}

@@ -255,7 +235,7 @@

255235 pub redis: RedisConfig,

256236 pub s3: S3Config,

257237 pub email: EmailConfig,

258 pub email_queue: EmailQueueConfig,

238 pub oauth: OAuthConfig,

259239 pub chat: ChatConfig,

260240 pub web: WebConfig,

261241 pub auth: AuthConfig,

@@ -269,7 +249,7 @@

269249 redis: self.redis,

270250 s3: self.s3,

271251 email: self.email,

272 email_queue: self.email_queue,

252 oauth: self.oauth,

273253 chat: self.chat,

274254 web: self.web,

275255 auth: self.auth,

@@ -285,8 +265,7 @@

285265 self.redis.validate()?;

286266 self.auth.validate()?;

287267 self.chat.is_valid()?;

288 self.email.is_valid()?;

289 self.email_queue.is_valid()?;

268 self.oauth.is_valid()?;

290269 self.im.is_valid()

291270 }

292271}

Deleted lib/email/src/templates/email_change_confirm.rs +0 −57

@@ -1,57 +0,0 @@

1use i18n::{UiLocale, catalog};

2use maud::Markup;

3use maud::html;

4

5use super::{append_query, mail_shell, url_join};

6

7pub fn mail_email_change_confirm(

8 new_email: &str,

9 ticket_id: &str,

10 code: &str,

11 base_url: &str,

12 locale: UiLocale,

13) -> Markup {

14 let confirm_link = append_query(

15 &url_join(base_url, "/email/change/confirm"),

16 &[("ticket_id", ticket_id), ("code", code)],

17 );

18 let cancel_link = append_query(

19 &url_join(base_url, "/email/change/cancel"),

20 &[("ticket_id", ticket_id)],

21 );

22 let body = html! {

23 h1 style="font-size:24px;font-weight:700;margin:0 0 8px;" { (catalog::email_change_heading(locale)) }

24 p style="color:#374151;margin:0 0 24px;" {

25 (catalog::email_change_body(locale, new_email))

26 }

27 table role="presentation" cellpadding="0" cellspacing="0" {

28 tr {

29 td align="center" {

30 a href=(confirm_link) class="btn" { (catalog::email_change_button(locale)) }

31 }

32 }

33 }

34 p class="muted" style="margin:16px 0 0;" {

35 (catalog::email_change_or_code(locale))

36 strong style="color:#111827;" { (code) }

37 }

38 p class="muted" style="margin:16px 0 0;" {

39 (catalog::email_change_not_requested(locale))

40 a href=(cancel_link) style="color:#2563eb;" { (catalog::email_change_cancel(locale)) }

41 }

42 };

43 mail_shell(catalog::email_change_subject(locale), body, locale)

44}

45

46pub fn render(vars: &serde_json::Value) -> Markup {

47 let new_email = vars.get("new_email").and_then(|v| v.as_str()).unwrap_or("");

48 let ticket_id = vars.get("ticket_id").and_then(|v| v.as_str()).unwrap_or("");

49 let code = vars.get("code").and_then(|v| v.as_str()).unwrap_or("");

50 let base_url = vars.get("base_url").and_then(|v| v.as_str()).unwrap_or("");

51 let locale = vars

52 .get("locale")

53 .and_then(|v| v.as_str())

54 .and_then(UiLocale::parse)

55 .unwrap_or(UiLocale::En);

56 mail_email_change_confirm(new_email, ticket_id, code, base_url, locale)

57}

Deleted lib/email/src/templates/mod.rs +0 −93

@@ -1,93 +0,0 @@

1use maud::{DOCTYPE, Markup, html};

2

3pub mod email_change_confirm;

4pub mod reset_link;

5pub mod verify_code;

6

7pub use email_change_confirm::mail_email_change_confirm;

8pub use reset_link::mail_reset_link;

9pub use verify_code::mail_verify_code;

10

11pub(crate) fn mail_shell(title: &str, body: Markup, locale: i18n::UiLocale) -> Markup {

12 html! {

13 (DOCTYPE)

14 html lang=(locale.tag()) xmlns="http://www.w3.org/1999/xhtml" {

15 head {

16 meta charset="utf-8";

17 meta name="viewport" content="width=device-width, initial-scale=1";

18 meta http-equiv="X-UA-Compatible" content="IE=edge";

19 title { (title) }

20 style type="text/css" {

21 "body,table,td,a{-webkit-text-size-adjust:100%;-ms-text-size-adjust:100%;}"

22 "table,td{mso-table-lspace:0pt;mso-table-rspace:0pt;}"

23 "img{border:0;height:auto;line-height:100%;outline:none;text-decoration:none;}"

24 "table{border-collapse:collapse;}"

25 "body{height:100%;margin:0;padding:0;width:100%;}"

26 "a[x-apple-data-detectors]{color:inherit;text-decoration:none;}"

27 ".btn{display:inline-block;padding:12px 28px;font-size:16px;font-weight:600;"

28 "text-decoration:none;border-radius:8px;color:#ffffff;background-color:#2563eb;"

29 "font-family:-apple-system,BlinkMacSystemFont,'Segoe UI',Roboto,sans-serif;}"

30 ".btn:hover{background-color:#1d4ed8;}"

31 ".container{max-width:600px;margin:0 auto;padding:24px;"

32 "font-family:-apple-system,BlinkMacSystemFont,'Segoe UI',Roboto,sans-serif;}"

33 ".code{font-size:32px;font-weight:700;letter-spacing:8px;color:#111827;"

34 "padding:16px 24px;background:#f3f4f6;border-radius:8px;text-align:center;margin:16px 0;}"

35 ".muted{color:#6b7280;font-size:14px;}"

36 }

37 }

38 body style="background-color:#f9fafb;margin:0;padding:0;" {

39 table role="presentation" width="100%" cellpadding="0" cellspacing="0" {

40 tr {

41 td align="center" style="padding:24px;" {

42 table role="presentation" class="container" width="600" cellpadding="0" cellspacing="0" style="background:#ffffff;border-radius:12px;box-shadow:0 1px 3px rgba(0,0,0,0.1);" {

43 tr {

44 td style="padding:32px 40px;" {

45 (body)

46 }

47 }

48 }

49 }

50 }

51 }

52 }

53 }

54 }

55}

56

57pub(crate) fn url_join(base: &str, path: &str) -> String {

58 let base = base.trim_end_matches('/');

59 let path = path.trim_start_matches('/');

60 format!("{base}/{path}")

61}

62

63pub(crate) fn append_query(url: &str, pairs: &[(&str, &str)]) -> String {

64 if pairs.is_empty() {

65 return url.to_string();

66 }

67 let mut s = format!("{url}?");

68 for (i, (k, v)) in pairs.iter().enumerate() {

69 if i > 0 {

70 s.push('&');

71 }

72 s.push_str(&percent_encode(k));

73 s.push('=');

74 s.push_str(&percent_encode(v));

75 }

76 s

77}

78

79fn percent_encode(input: &str) -> String {

80 let mut out = String::with_capacity(input.len());

81 for &b in input.as_bytes() {

82 match b {

83 b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => {

84 out.push(b as char);

85 }

86 _ => {

87 out.push('%');

88 out.push_str(&format!("{b:02X}"));

89 }

90 }

91 }

92 out

93}

Deleted lib/email/src/templates/reset_link.rs +0 −46

@@ -1,46 +0,0 @@

1use i18n::{UiLocale, catalog};

2use maud::Markup;

3use maud::html;

4

5use super::{append_query, mail_shell, url_join};

6

7pub fn mail_reset_link(

8 email: &str,

9 token: &str,

10 temp_id: &str,

11 base_url: &str,

12 locale: UiLocale,

13) -> Markup {

14 let link = append_query(

15 &url_join(base_url, "/auth/reset-password"),

16 &[("email", email), ("token", token), ("temp_id", temp_id)],

17 );

18 let body = html! {

19 h1 style="font-size:24px;font-weight:700;margin:0 0 8px;" { (catalog::reset_heading(locale)) }

20 p style="color:#374151;margin:0 0 24px;" { (catalog::reset_body(locale)) }

21 table role="presentation" cellpadding="0" cellspacing="0" {

22 tr {

23 td align="center" {

24 a href=(link) class="btn" { (catalog::reset_button(locale)) }

25 }

26 }

27 }

28 p class="muted" style="margin:24px 0 0;" { (catalog::link_fallback(locale)) }

29 p style="word-break:break-all;font-size:13px;color:#2563eb;margin:4px 0 0;" { (link) }

30 p class="muted" style="margin:24px 0 0;" { (catalog::reset_ignore(locale)) }

31 };

32 mail_shell(catalog::reset_subject(locale), body, locale)

33}

34

35pub fn render(vars: &serde_json::Value) -> Markup {

36 let email = vars.get("email").and_then(|v| v.as_str()).unwrap_or("");

37 let token = vars.get("token").and_then(|v| v.as_str()).unwrap_or("");

38 let temp_id = vars.get("temp_id").and_then(|v| v.as_str()).unwrap_or("");

39 let base_url = vars.get("base_url").and_then(|v| v.as_str()).unwrap_or("");

40 let locale = vars

41 .get("locale")

42 .and_then(|v| v.as_str())

43 .and_then(UiLocale::parse)

44 .unwrap_or(UiLocale::En);

45 mail_reset_link(email, token, temp_id, base_url, locale)

46}

Deleted lib/email/src/templates/verify_code.rs +0 −69

@@ -1,69 +0,0 @@

1use i18n::{UiLocale, catalog};

2use maud::Markup;

3use maud::html;

4

5use super::{append_query, mail_shell, url_join};

6

7pub fn mail_verify_code(

8 code: &str,

9 minutes: u64,

10 purpose: &str,

11 link: Option<&str>,

12 locale: UiLocale,

13) -> Markup {

14 let body = html! {

15 h1 style="font-size:24px;font-weight:700;margin:0 0 8px;" { (purpose) }

16 p style="color:#374151;margin:0 0 8px;" { (catalog::verify_body(locale, minutes)) }

17 div class="code" { (code) }

18 @if let Some(url) = link {

19 p style="margin:24px 0 0;" {

20 a href=(url) class="btn" { (catalog::verify_verify_button(locale)) }

21 }

22 p class="muted" style="margin:16px 0 0;" {

23 (catalog::link_fallback(locale))

24 }

25 p style="word-break:break-all;font-size:13px;color:#2563eb;margin:4px 0 0;" { (url) }

26 }

27 p class="muted" style="margin:16px 0 0;" {

28 (catalog::ignore_if_unrequested(locale))

29 }

30 };

31 mail_shell(purpose, body, locale)

32}

33

34pub fn render(vars: &serde_json::Value) -> Markup {

35 let code = vars.get("code").and_then(|v| v.as_str()).unwrap_or("");

36 let minutes = vars.get("minutes").and_then(|v| v.as_u64()).unwrap_or(0);

37 let purpose = vars

38 .get("purpose")

39 .and_then(|v| v.as_str())

40 .unwrap_or("Verification");

41 let link = build_link(vars);

42 // Templates registered through `render_and_send` receive the already

43 // localized `purpose`; the locale only drives the surrounding chrome.

44 let locale = vars

45 .get("locale")

46 .and_then(|v| v.as_str())

47 .and_then(UiLocale::parse)

48 .unwrap_or(UiLocale::En);

49 mail_verify_code(code, minutes, purpose, link.as_deref(), locale)

50}

51

52fn build_link(vars: &serde_json::Value) -> Option<String> {

53 let base_url = vars.get("base_url").and_then(|v| v.as_str())?;

54 let email = vars.get("email").and_then(|v| v.as_str())?;

55 let temp_id = vars.get("temp_id").and_then(|v| v.as_str())?;

56 let code = vars.get("code").and_then(|v| v.as_str()).unwrap_or("");

57 let mut query = vec![("email", email), ("temp_id", temp_id), ("code", code)];

58 // Optional slug (signup / password_reset / ...); the human-readable

59 // `purpose` var is a heading and must not leak into the URL.

60 let purpose_slug;

61 if let Some(slug) = vars.get("purpose_slug").and_then(|v| v.as_str()) {

62 purpose_slug = slug;

63 query.push(("purpose", purpose_slug));

64 }

65 Some(append_query(

66 &url_join(base_url, "/auth/verify-email"),

67 &query,

68 ))

69}

Deleted lib/email/src/error.rs +0 −107

@@ -1,107 +0,0 @@

1use std::fmt;

2

3use lettre::address::AddressError;

4use lettre::error::Error as LettreError;

5use lettre::transport::smtp::Error as SmtpError;

6

7#[derive(Debug)]

8pub enum EmailError {

9 Config(String),

10 InvalidConfig(String),

11 Disabled,

12 Queue(redis::RedisError),

13 Serde(serde_json::Error),

14 Smtp(SmtpError),

15 Address(AddressError),

16 Lettre(LettreError),

17 TemplateNotFound(String),

18 TemplateRender(String),

19 Base64(base64::DecodeError),

20 AttachmentTooLarge { actual: usize, max: usize },

21 Missing(String),

22 Io(std::io::Error),

23 Cancelled,

24 Other(String),

25}

26

27impl fmt::Display for EmailError {

28 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {

29 match self {

30 Self::Config(msg) => write!(f, "email config error: {msg}"),

31 Self::InvalidConfig(msg) => write!(f, "email invalid config: {msg}"),

32 Self::Disabled => write!(f, "email service is disabled"),

33 Self::Queue(err) => write!(f, "email queue error: {err}"),

34 Self::Serde(err) => write!(f, "email serde error: {err}"),

35 Self::Smtp(err) => write!(f, "email smtp error: {err}"),

36 Self::Address(err) => write!(f, "email address error: {err}"),

37 Self::Lettre(err) => write!(f, "email lettre error: {err}"),

38 Self::TemplateNotFound(name) => write!(f, "email template not found: {name}"),

39 Self::TemplateRender(msg) => write!(f, "email template render error: {msg}"),

40 Self::Base64(err) => write!(f, "email base64 decode error: {err}"),

41 Self::AttachmentTooLarge { actual, max } => {

42 write!(f, "attachment too large: {actual} bytes (max {max})")

43 }

44 Self::Missing(msg) => write!(f, "missing: {msg}"),

45 Self::Io(err) => write!(f, "email io error: {err}"),

46 Self::Cancelled => write!(f, "email operation cancelled"),

47 Self::Other(msg) => write!(f, "email error: {msg}"),

48 }

49 }

50}

51

52impl std::error::Error for EmailError {

53 fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {

54 match self {

55 Self::Queue(err) => Some(err),

56 Self::Serde(err) => Some(err),

57 Self::Smtp(err) => Some(err),

58 Self::Address(err) => Some(err),

59 Self::Lettre(err) => Some(err),

60 Self::Base64(err) => Some(err),

61 Self::Io(err) => Some(err),

62 _ => None,

63 }

64 }

65}

66

67impl From<redis::RedisError> for EmailError {

68 fn from(err: redis::RedisError) -> Self {

69 Self::Queue(err)

70 }

71}

72

73impl From<serde_json::Error> for EmailError {

74 fn from(err: serde_json::Error) -> Self {

75 Self::Serde(err)

76 }

77}

78

79impl From<SmtpError> for EmailError {

80 fn from(err: SmtpError) -> Self {

81 Self::Smtp(err)

82 }

83}

84

85impl From<AddressError> for EmailError {

86 fn from(err: AddressError) -> Self {

87 Self::Address(err)

88 }

89}

90

91impl From<LettreError> for EmailError {

92 fn from(err: LettreError) -> Self {

93 Self::Lettre(err)

94 }

95}

96

97impl From<base64::DecodeError> for EmailError {

98 fn from(err: base64::DecodeError) -> Self {

99 Self::Base64(err)

100 }

101}

102

103impl From<std::io::Error> for EmailError {

104 fn from(err: std::io::Error) -> Self {

105 Self::Io(err)

106 }

107}

Deleted lib/email/src/events.rs +0 −128

@@ -1,128 +0,0 @@

1use std::collections::HashMap;

2

3use serde::{Deserialize, Serialize};

4use uuid::Uuid;

5

6pub const CHANNEL_SEND: &str = "send";

7pub const CHANNEL_BATCH: &str = "batch";

8pub const CHANNEL_RENDER: &str = "render";

9pub const CHANNEL_SENT: &str = "sent";

10pub const CHANNEL_FAILED: &str = "failed";

11

12#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]

13#[serde(default)]

14pub struct EmailAddress {

15 pub address: String,

16 pub name: Option<String>,

17}

18

19impl EmailAddress {

20 pub fn new(address: impl Into<String>) -> Self {

21 Self {

22 address: address.into(),

23 name: None,

24 }

25 }

26

27 pub fn with_name(address: impl Into<String>, name: impl Into<String>) -> Self {

28 Self {

29 address: address.into(),

30 name: Some(name.into()),

31 }

32 }

33}

34

35#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]

36#[serde(default)]

37pub struct Attachment {

38 pub filename: String,

39 pub content_type: String,

40 pub data_b64: String,

41 pub content_id: Option<String>,

42}

43

44#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]

45#[serde(default)]

46pub struct SendEmailRequest {

47 pub request_id: Uuid,

48 pub to: Vec<EmailAddress>,

49 pub cc: Vec<EmailAddress>,

50 pub bcc: Vec<EmailAddress>,

51 pub subject: String,

52 pub text: Option<String>,

53 pub html: Option<String>,

54 pub reply_to: Option<EmailAddress>,

55 pub attachments: Vec<Attachment>,

56 pub metadata: HashMap<String, String>,

57 pub queued_at_ms: i64,

58 pub trace: Option<String>,

59}

60

61#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]

62#[serde(default)]

63pub struct BatchRecipient {

64 pub request_id: Uuid,

65 pub to: EmailAddress,

66 pub subject: String,

67 pub text: Option<String>,

68 pub html: Option<String>,

69 pub metadata: HashMap<String, String>,

70}

71

72#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]

73#[serde(default)]

74pub struct SendBatchEmailRequest {

75 pub request_id: Uuid,

76 pub default_subject: String,

77 pub default_text: Option<String>,

78 pub default_html: Option<String>,

79 pub recipients: Vec<BatchRecipient>,

80 pub metadata: HashMap<String, String>,

81 pub queued_at_ms: i64,

82 pub trace: Option<String>,

83}

84

85#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]

86#[serde(default)]

87pub struct RenderAndSendEmailRequest {

88 pub request_id: Uuid,

89 pub template: String,

90 pub vars: serde_json::Value,

91 pub to: Vec<EmailAddress>,

92 pub cc: Vec<EmailAddress>,

93 pub bcc: Vec<EmailAddress>,

94 pub subject: String,

95 pub text: Option<String>,

96 pub locale: Option<String>,

97 pub metadata: HashMap<String, String>,

98 pub queued_at_ms: i64,

99 pub trace: Option<String>,

100}

101

102#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]

103#[serde(default)]

104pub struct EmailSent {

105 pub request_id: Uuid,

106 pub recipient_count: usize,

107 pub smtp_message_id: String,

108 pub sent_at_ms: i64,

109 pub metadata: HashMap<String, String>,

110}

111

112#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]

113#[serde(default)]

114pub struct EmailFailed {

115 pub request_id: Uuid,

116 pub error: String,

117 pub recipient_count: usize,

118 pub failed_at_ms: i64,

119 pub metadata: HashMap<String, String>,

120}

121

122pub fn now_ms() -> i64 {

123 use std::time::{SystemTime, UNIX_EPOCH};

124 let dur = SystemTime::now()

125 .duration_since(UNIX_EPOCH)

126 .unwrap_or_default();

127 dur.as_millis() as i64

128}

Deleted lib/email/src/lib.rs +0 −24

@@ -1,24 +0,0 @@

1pub mod error;

2pub mod events;

3pub mod metrics;

4pub mod queue;

5pub mod service;

6pub mod subscriber;

7pub mod template;

8pub mod templates;

9pub mod trace;

10pub mod transport;

11

12pub mod runtime;

13

14pub use error::EmailError;

15pub use events::{

16 Attachment, BatchRecipient, CHANNEL_BATCH, CHANNEL_FAILED, CHANNEL_RENDER, CHANNEL_SEND,

17 CHANNEL_SENT, EmailAddress, EmailFailed, EmailSent, RenderAndSendEmailRequest,

18 SendBatchEmailRequest, SendEmailRequest,

19};

20pub use queue::{EmailQueue, QueueSettings, SubscribeHandle, TypedMessageMeta};

21pub use service::{EmailService, EmailServiceBuilder};

22pub use subscriber::{EmailSubscribers, start_subscribers};

23pub use template::{TemplateFn, TemplateRegistry};

24pub use transport::{EmailTransport, EmailTransportBuilder, build_message};

Deleted lib/email/src/metrics.rs +0 −16

@@ -1,16 +0,0 @@

1use std::sync::LazyLock;

2

3use opentelemetry::KeyValue;

4use opentelemetry::global;

5use opentelemetry::metrics::Counter;

6

7static EMAIL_SENT: LazyLock<Counter<u64>> = LazyLock::new(|| {

8 global::meter("gitdata")

9 .u64_counter("email.sent")

10 .with_description("Email sends by outcome")

11 .build()

12});

13

14pub fn email_sent(outcome: &'static str) {

15 EMAIL_SENT.add(1, &[KeyValue::new("outcome", outcome)]);

16}

Deleted lib/email/src/queue.rs +0 −310

@@ -1,310 +0,0 @@

1use std::future::Future;

2use std::marker::PhantomData;

3use std::sync::Arc;

4use std::time::Duration;

5

6use redis::aio::ConnectionManager;

7use redis::cluster_async::ClusterConnection;

8use serde::Serialize;

9use serde::de::DeserializeOwned;

10use tokio::task::JoinHandle;

11use tokio_util::sync::CancellationToken;

12use tracing::{debug, error, warn};

13

14use crate::error::EmailError;

15

16#[derive(Clone)]

17pub enum QueueConnection {

18 Standalone(ConnectionManager),

19 Cluster(ClusterConnection),

20}

21

22impl QueueConnection {

23 pub fn is_cluster(&self) -> bool {

24 matches!(self, QueueConnection::Cluster(_))

25 }

26}

27

28macro_rules! dispatch_conn {

29 ($conn:expr, |$c:ident| $body:expr) => {

30 match &mut $conn {

31 QueueConnection::Standalone($c) => $body,

32 QueueConnection::Cluster($c) => $body,

33 }

34 };

35}

36

37pub struct EmailQueue {

38 connection: QueueConnection,

39 config: QueueSettings,

40 pubsub_client: redis::Client,

41}

42

43#[derive(Clone)]

44pub struct QueueSettings {

45 pub channel_prefix: String,

46 pub max_retries: u32,

47 pub retry_backoff: Duration,

48 pub enabled: bool,

49}

50

51impl EmailQueue {

52 pub async fn connect(url: &str, config: QueueSettings) -> Result<Self, EmailError> {

53 let connection = open_connection(url).await?;

54 let pubsub_client = open_pubsub_client(url)?;

55 Ok(Self {

56 connection,

57 config,

58 pubsub_client,

59 })

60 }

61

62 pub fn enabled(&self) -> bool {

63 self.config.enabled

64 }

65

66 pub fn channel(&self, name: &str) -> String {

67 format!("{}.{}", self.config.channel_prefix, name)

68 }

69

70 pub fn config(&self) -> &QueueSettings {

71 &self.config

72 }

73

74 pub async fn publish<T: Serialize + ?Sized>(

75 &self,

76 channel: &str,

77 msg: &T,

78 ) -> Result<(), EmailError> {

79 let payload = serde_json::to_vec(msg)?;

80 let mut conn = self.connection.clone();

81 let receiver_count: i64 = dispatch_conn!(conn, |c| {

82 Box::pin(

83 redis::cmd("PUBLISH")

84 .arg(channel)

85 .arg(payload.as_slice())

86 .query_async(c),

87 )

88 .await

89 })?;

90 debug!(

91 channel = %channel,

92 receivers = receiver_count,

93 bytes = payload.len(),

94 "published email message"

95 );

96 Ok(())

97 }

98

99 pub async fn publish_to<T: Serialize + ?Sized>(

100 &self,

101 logical: &str,

102 msg: &T,

103 ) -> Result<(), EmailError> {

104 let channel = self.channel(logical);

105 self.publish(&channel, msg).await

106 }

107

108 pub async fn subscribe<T, F, Fut>(

109 &self,

110 logical: &str,

111 group_id: &str,

112 handler: F,

113 ) -> Result<SubscribeHandle<T>, EmailError>

114 where

115 T: DeserializeOwned + Send + Clone + 'static,

116 F: Fn(T, TypedMessageMeta) -> Fut + Send + Sync + 'static,

117 Fut: Future<Output = Result<(), EmailError>> + Send + 'static,

118 {

119 let _ = group_id;

120 let channel = self.channel(logical);

121 let max_retries = self.config.max_retries;

122 let retry_backoff = self.config.retry_backoff;

123 let cancel = CancellationToken::new();

124 let cancel_child = cancel.clone();

125 let channel_for_log = channel.clone();

126 let mut pubsub = self.pubsub_client.get_async_pubsub().await?;

127 pubsub.subscribe(&channel).await?;

128

129 let join: JoinHandle<Result<(), EmailError>> = tokio::spawn(async move {

130 let mut stream = pubsub.on_message();

131 loop {

132 tokio::select! {

133 _ = cancel_child.cancelled() => {

134 debug!(channel = %channel_for_log, "subscriber cancelled");

135 return Ok(());

136 }

137 msg = futures::StreamExt::next(&mut stream) => {

138 let Some(msg) = msg else {

139 warn!(channel = %channel_for_log, "pubsub stream ended");

140 return Ok(());

141 };

142 let raw: String = match msg.get_payload() {

143 Ok(s) => s,

144 Err(err) => {

145 warn!(channel = %channel_for_log, error = %err, "invalid pubsub payload type");

146 continue;

147 }

148 };

149 let meta = TypedMessageMeta {

150 channel: channel_for_log.clone(),

151 raw: raw.clone(),

152 };

153 let value: T = match serde_json::from_str(&raw) {

154 Ok(v) => v,

155 Err(err) => {

156 warn!(channel = %channel_for_log, error = %err, "message deserialize failed, dropping");

157 continue;

158 }

159 };

160 run_with_retries(

161 &channel_for_log,

162 value,

163 meta,

164 &handler,

165 max_retries,

166 retry_backoff,

167 &cancel_child,

168 )

169 .await;

170 }

171 }

172 }

173 });

174

175 Ok(SubscribeHandle {

176 cancel,

177 join: Some(Arc::new(tokio::sync::Mutex::new(join))),

178 _marker: PhantomData,

179 })

180 }

181}

182

183#[derive(Clone)]

184pub struct TypedMessageMeta {

185 pub channel: String,

186 pub raw: String,

187}

188

189/// The join handle of a running subscriber loop, shared so `shutdown` can await

190/// it from outside the task.

191type SharedJoinHandle = Arc<tokio::sync::Mutex<JoinHandle<Result<(), EmailError>>>>;

192

193pub struct SubscribeHandle<T> {

194 cancel: CancellationToken,

195 join: Option<SharedJoinHandle>,

196 _marker: PhantomData<T>,

197}

198

199impl<T> SubscribeHandle<T> {

200 pub async fn shutdown(mut self, timeout: Duration) -> Result<(), EmailError> {

201 self.cancel.cancel();

202 if let Some(handle) = self.join.take() {

203 let join = {

204 let mut guard = handle.lock().await;

205 std::mem::replace(&mut *guard, tokio::spawn(async { Ok(()) }))

206 };

207 match tokio::time::timeout(timeout, join).await {

208 Ok(Ok(res)) => res,

209 Ok(Err(join_err)) => {

210 error!(error = %join_err, "subscriber task join failed");

211 Err(EmailError::Other(format!(

212 "subscriber join failed: {join_err}"

213 )))

214 }

215 Err(_) => {

216 warn!(

217 timeout_secs = timeout.as_secs(),

218 "subscriber shutdown timed out"

219 );

220 Err(EmailError::Cancelled)

221 }

222 }

223 } else {

224 Ok(())

225 }

226 }

227}

228

229impl<T> Drop for SubscribeHandle<T> {

230 fn drop(&mut self) {

231 self.cancel.cancel();

232 }

233}

234

235async fn open_connection(url: &str) -> Result<QueueConnection, EmailError> {

236 if url.split(',').count() > 1 {

237 let nodes: Vec<String> = url

238 .split(',')

239 .map(str::trim)

240 .filter(|s| !s.is_empty())

241 .map(String::from)

242 .collect();

243 let client = redis::cluster::ClusterClient::new(nodes)?;

244 let conn = client.get_async_connection().await?;

245 Ok(QueueConnection::Cluster(conn))

246 } else {

247 let client = redis::Client::open(url)?;

248 let manager = ConnectionManager::new(client).await?;

249 Ok(QueueConnection::Standalone(manager))

250 }

251}

252

253fn open_pubsub_client(url: &str) -> Result<redis::Client, EmailError> {

254 if url.split(',').count() > 1 {

255 let first = url

256 .split(',')

257 .map(str::trim)

258 .find(|s| !s.is_empty())

259 .ok_or_else(|| EmailError::Config("empty redis url".into()))?;

260 Ok(redis::Client::open(first)?)

261 } else {

262 Ok(redis::Client::open(url)?)

263 }

264}

265

266async fn run_with_retries<T, F, Fut>(

267 channel: &str,

268 value: T,

269 meta: TypedMessageMeta,

270 handler: &F,

271 max_retries: u32,

272 retry_backoff: Duration,

273 cancel: &CancellationToken,

274) where

275 T: Send + Clone,

276 F: Fn(T, TypedMessageMeta) -> Fut + Send + Sync,

277 Fut: Future<Output = Result<(), EmailError>> + Send,

278{

279 let mut attempt: u32 = 0;

280 loop {

281 attempt += 1;

282 match handler(value.clone(), meta.clone()).await {

283 Ok(()) => {

284 debug!(channel = %channel, attempt, "message processed");

285 return;

286 }

287 Err(err) if attempt >= max_retries => {

288 error!(

289 channel = %channel,

290 attempt,

291 error = %err,

292 "handler exhausted retries, dropping message"

293 );

294 return;

295 }

296 Err(err) => {

297 warn!(

298 channel = %channel,

299 attempt,

300 error = %err,

301 "handler failed, will retry"

302 );

303 tokio::select! {

304 _ = cancel.cancelled() => return,

305 _ = tokio::time::sleep(retry_backoff) => {}

306 }

307 }

308 }

309 }

310}

Deleted lib/email/src/runtime.rs +0 −104

@@ -1,104 +0,0 @@

1use std::path::PathBuf;

2use std::time::Duration;

3

4use tracing::{debug, info, warn};

5

6use crate::error::EmailError;

7use crate::service::{EmailService, EmailServiceBuilder};

8use crate::subscriber::EmailSubscribers;

9

10#[derive(Default)]

11pub struct WorkerOptions {

12 pub template_dir: Option<PathBuf>,

13 pub shutdown_timeout: Option<Duration>,

14}

15

16pub struct RunningWorker {

17 pub service: EmailService,

18 pub subscribers: EmailSubscribers,

19}

20

21pub async fn start_worker(

22 email_cfg: config::email::EmailConfig,

23 queue_cfg: config::email_queue::EmailQueueConfig,

24 redis_url: String,

25 opts: WorkerOptions,

26) -> Result<RunningWorker, EmailError> {

27 let _ = opts.template_dir;

28 let enabled = email_cfg.enabled();

29 info!(

30 smtp_enabled = enabled,

31 queue_enabled = queue_cfg.enabled(),

32 "building email service"

33 );

34

35 let builder = EmailServiceBuilder::new(email_cfg)

36 .with_queue(redis_url, queue_cfg)

37 .register_default_templates()

38 .await;

39 let service = builder.build().await?;

40 let subscribers = service.start_subscribers().await?;

41

42 info!(

43 smtp_enabled = service.enabled(),

44 subscribers_started = service.queue().is_some_and(|q| q.enabled()),

45 "email worker ready"

46 );

47

48 Ok(RunningWorker {

49 service,

50 subscribers,

51 })

52}

53

54pub async fn run_until_shutdown(

55 email_cfg: config::email::EmailConfig,

56 queue_cfg: config::email_queue::EmailQueueConfig,

57 redis_url: String,

58 opts: WorkerOptions,

59) -> Result<(), EmailError> {

60 let worker = start_worker(email_cfg, queue_cfg, redis_url, opts).await?;

61 if !worker.service.enabled() {

62 warn!("email service is disabled at startup; exiting immediately");

63 return Ok(());

64 }

65 info!("email worker started; waiting for shutdown signal");

66 debug!("entering wait_for_shutdown");

67 wait_for_shutdown().await;

68 let timeout = Duration::from_secs(5);

69 debug!(

70 timeout_secs = timeout.as_secs(),

71 "calling subscribers.shutdown"

72 );

73 worker.subscribers.shutdown(timeout).await?;

74 info!("email worker shut down");

75 Ok(())

76}

77

78pub async fn wait_for_shutdown() {

79 #[cfg(unix)]

80 {

81 use tokio::signal::unix::{SignalKind, signal};

82 let mut sigterm = match signal(SignalKind::terminate()) {

83 Ok(s) => s,

84 Err(err) => {

85 warn!(error = %err, "failed to install SIGTERM handler");

86 let _ = tokio::signal::ctrl_c().await;

87 return;

88 }

89 };

90 tokio::select! {

91 _ = tokio::signal::ctrl_c() => {

92 info!("received SIGINT");

93 }

94 _ = sigterm.recv() => {

95 info!("received SIGTERM");

96 }

97 }

98 }

99 #[cfg(not(unix))]

100 {

101 let _ = tokio::signal::ctrl_c().await;

102 info!("received ctrl-c");

103 }

104}

Deleted lib/email/src/service.rs +0 −451

@@ -1,451 +0,0 @@

1use std::sync::Arc;

2

3use lettre::address::Address;

4use lettre::message::Mailbox;

5use tracing::{debug, info, trace, warn};

6use uuid::Uuid;

7

8use crate::error::EmailError;

9use crate::events::{

10 CHANNEL_BATCH, CHANNEL_FAILED, CHANNEL_RENDER, CHANNEL_SEND, CHANNEL_SENT, EmailAddress,

11 EmailFailed, EmailSent, RenderAndSendEmailRequest, SendBatchEmailRequest, SendEmailRequest,

12 now_ms,

13};

14use crate::queue::EmailQueue;

15use crate::queue::QueueSettings;

16use crate::template::TemplateRegistry;

17use crate::templates;

18use crate::transport::{EmailTransport, EmailTransportBuilder, build_message};

19

20#[derive(Clone)]

21pub struct EmailService {

22 inner: Arc<EmailServiceInner>,

23}

24

25struct EmailServiceInner {

26 config: Arc<config::email::EmailConfig>,

27 transport: Option<EmailTransport>,

28 from: Option<Mailbox>,

29 queue: Option<EmailQueue>,

30 templates: TemplateRegistry,

31 max_attachment_bytes: usize,

32}

33

34impl EmailService {

35 pub fn enabled(&self) -> bool {

36 self.inner.transport.is_some()

37 }

38

39 pub fn config(&self) -> &config::email::EmailConfig {

40 &self.inner.config

41 }

42

43 pub fn templates(&self) -> &TemplateRegistry {

44 &self.inner.templates

45 }

46

47 pub fn queue(&self) -> Option<&EmailQueue> {

48 self.inner.queue.as_ref()

49 }

50

51 pub async fn render(&self, name: &str, vars: &serde_json::Value) -> Result<String, EmailError> {

52 self.inner.templates.render(name, vars).await

53 }

54

55 pub async fn send(&self, mut req: SendEmailRequest) -> Result<EmailSent, EmailError> {

56 let transport = self.inner.transport.as_ref().ok_or(EmailError::Disabled)?;

57 let from = self

58 .inner

59 .from

60 .clone()

61 .ok_or_else(|| EmailError::Config("from address not configured".into()))?;

62

63 if req.request_id.is_nil() {

64 req.request_id = Uuid::now_v7();

65 }

66 let recipient_count = req.to.len() + req.cc.len() + req.bcc.len();

67 if recipient_count == 0 {

68 return Err(EmailError::Missing("email has no recipients".into()));

69 }

70 debug!(

71 request_id = %req.request_id,

72 recipients = recipient_count,

73 subject_len = req.subject.len(),

74 attachments = req.attachments.len(),

75 "service::send dispatch"

76 );

77

78 let message = build_message(&req, from, self.inner.max_attachment_bytes)?;

79 let smtp_message_id = match transport.send(message).await {

80 Ok(message_id) => {

81 crate::metrics::email_sent("ok");

82 message_id

83 }

84 Err(error) => {

85 crate::metrics::email_sent("error");

86 return Err(error);

87 }

88 };

89 let sent = EmailSent {

90 request_id: req.request_id,

91 recipient_count,

92 smtp_message_id,

93 sent_at_ms: now_ms(),

94 metadata: req.metadata,

95 };

96

97 if let Some(q) = &self.inner.queue

98 && let Err(err) = q.publish_to(CHANNEL_SENT, &sent).await

99 {

100 warn!(error = %err, "failed to publish EmailSent event");

101 }

102 info!(

103 request_id = %sent.request_id,

104 recipients = sent.recipient_count,

105 "email sent"

106 );

107 Ok(sent)

108 }

109

110 pub async fn send_mail(

111 &self,

112 to: &str,

113 subject: &str,

114 text: &str,

115 html: &str,

116 ) -> Result<(), EmailError> {

117 trace!(to, subject_len = subject.len(), "service::send_mail");

118 let req = SendEmailRequest {

119 request_id: Uuid::now_v7(),

120 to: vec![EmailAddress::new(to)],

121 cc: vec![],

122 bcc: vec![],

123 subject: subject.to_string(),

124 text: if text.is_empty() {

125 None

126 } else {

127 Some(text.to_string())

128 },

129 html: if html.is_empty() {

130 None

131 } else {

132 Some(html.to_string())

133 },

134 reply_to: None,

135 attachments: vec![],

136 metadata: Default::default(),

137 queued_at_ms: now_ms(),

138 trace: crate::trace::capture(),

139 };

140 let _ = self.send(req).await?;

141 Ok(())

142 }

143

144 pub async fn send_batch(

145 &self,

146 req: SendBatchEmailRequest,

147 ) -> Result<Vec<EmailSent>, EmailError> {

148 let mut sent_list = Vec::with_capacity(req.recipients.len());

149 let default_meta = req.metadata.clone();

150 let total = req.recipients.len();

151 let batch_id = req.request_id;

152 debug!(batch = %batch_id, total, "service::send_batch dispatch");

153

154 for (idx, recipient) in req.recipients.into_iter().enumerate() {

155 if recipient.to.address.is_empty() {

156 warn!(batch = %batch_id, index = idx, "skipping batch recipient without address");

157 continue;

158 }

159 debug!(batch = %batch_id, index = idx, address = %recipient.to.address, "batch recipient processing");

160 let mut meta = default_meta.clone();

161 meta.insert("batch_request_id".into(), batch_id.to_string());

162 meta.insert("batch_index".into(), idx.to_string());

163 for (k, v) in &recipient.metadata {

164 meta.insert(k.clone(), v.clone());

165 }

166

167 let send_req = SendEmailRequest {

168 request_id: if recipient.request_id.is_nil() {

169 Uuid::now_v7()

170 } else {

171 recipient.request_id

172 },

173 to: vec![recipient.to.clone()],

174 cc: vec![],

175 bcc: vec![],

176 subject: if recipient.subject.is_empty() {

177 req.default_subject.clone()

178 } else {

179 recipient.subject

180 },

181 text: recipient.text.or_else(|| req.default_text.clone()),

182 html: recipient.html.or_else(|| req.default_html.clone()),

183 reply_to: None,

184 attachments: vec![],

185 metadata: meta,

186 queued_at_ms: now_ms(),

187 trace: crate::trace::capture(),

188 };

189 match self.send(send_req).await {

190 Ok(sent) => sent_list.push(sent),

191 Err(err) => {

192 warn!(

193 batch = %batch_id,

194 index = idx,

195 error = %err,

196 "batch recipient failed"

197 );

198 let failed = EmailFailed {

199 request_id: Uuid::nil(),

200 error: err.to_string(),

201 recipient_count: 1,

202 failed_at_ms: now_ms(),

203 metadata: Default::default(),

204 };

205 if let Some(q) = &self.inner.queue

206 && let Err(err) = q.publish_to(CHANNEL_FAILED, &failed).await

207 {

208 warn!(error = %err, "failed to publish EmailFailed event");

209 }

210 }

211 }

212 }

213 debug!(batch = %batch_id, sent = sent_list.len(), total, "batch processed");

214 Ok(sent_list)

215 }

216

217 pub async fn render_and_send(

218 &self,

219 req: RenderAndSendEmailRequest,

220 ) -> Result<EmailSent, EmailError> {

221 let recipient_count = req.to.len() + req.cc.len() + req.bcc.len();

222 debug!(

223 request_id = %req.request_id,

224 template = %req.template,

225 recipients = recipient_count,

226 "service::render_and_send dispatch"

227 );

228 // The locale travels inside the template vars: registered templates are

229 // `Fn(&Value) -> Markup`, and every one of them reads `locale` to pick

230 // its catalogue. Injecting here keeps the `render_and_send` contract

231 // (and the queue payload) unchanged for callers.

232 let mut vars = req.vars.clone();

233 if let Some(object) = vars.as_object_mut()

234 && !object.contains_key("locale")

235 {

236 let tag = req

237 .locale

238 .as_deref()

239 .and_then(i18n::UiLocale::parse)

240 .unwrap_or(i18n::UiLocale::En);

241 object.insert(

242 "locale".to_string(),

243 serde_json::Value::String(tag.tag().to_string()),

244 );

245 }

246 let html = self.inner.templates.render(&req.template, &vars).await?;

247 debug!(

248 request_id = %req.request_id,

249 template = %req.template,

250 bytes = html.len(),

251 "template rendered"

252 );

253 let send_req = SendEmailRequest {

254 request_id: req.request_id,

255 to: req.to,

256 cc: req.cc,

257 bcc: req.bcc,

258 subject: req.subject,

259 text: req.text,

260 html: Some(html),

261 reply_to: None,

262 attachments: vec![],

263 metadata: req.metadata,

264 queued_at_ms: now_ms(),

265 trace: crate::trace::capture(),

266 };

267 self.send(send_req).await

268 }

269

270 pub async fn publish_send(&self, req: &SendEmailRequest) -> Result<(), EmailError> {

271 let q = self.inner.queue.as_ref().ok_or(EmailError::Disabled)?;

272 q.publish_to(CHANNEL_SEND, req).await

273 }

274

275 pub async fn publish_batch(&self, req: &SendBatchEmailRequest) -> Result<(), EmailError> {

276 let q = self.inner.queue.as_ref().ok_or(EmailError::Disabled)?;

277 q.publish_to(CHANNEL_BATCH, req).await

278 }

279

280 pub async fn publish_render(&self, req: &RenderAndSendEmailRequest) -> Result<(), EmailError> {

281 let q = self.inner.queue.as_ref().ok_or(EmailError::Disabled)?;

282 q.publish_to(CHANNEL_RENDER, req).await

283 }

284

285 pub async fn publish_sent(&self, sent: &EmailSent) -> Result<(), EmailError> {

286 let q = self.inner.queue.as_ref().ok_or(EmailError::Disabled)?;

287 q.publish_to(CHANNEL_SENT, sent).await

288 }

289

290 pub async fn publish_failed(&self, failed: &EmailFailed) -> Result<(), EmailError> {

291 let q = self.inner.queue.as_ref().ok_or(EmailError::Disabled)?;

292 q.publish_to(CHANNEL_FAILED, failed).await

293 }

294

295 pub async fn start_subscribers(

296 &self,

297 ) -> Result<crate::subscriber::EmailSubscribers, EmailError> {

298 crate::subscriber::start_subscribers(self.clone()).await

299 }

300}

301

302pub struct EmailServiceBuilder {

303 email_config: config::email::EmailConfig,

304 queue_config: Option<config::email_queue::EmailQueueConfig>,

305 redis_url: Option<String>,

306 templates: TemplateRegistry,

307}

308

309impl EmailServiceBuilder {

310 pub fn new(email_config: config::email::EmailConfig) -> Self {

311 Self {

312 email_config,

313 queue_config: None,

314 redis_url: None,

315 templates: TemplateRegistry::new(),

316 }

317 }

318

319 pub fn with_queue(

320 mut self,

321 redis_url: impl Into<String>,

322 queue_config: config::email_queue::EmailQueueConfig,

323 ) -> Self {

324 self.redis_url = Some(redis_url.into());

325 self.queue_config = Some(queue_config);

326 self

327 }

328

329 pub fn with_templates(mut self, templates: TemplateRegistry) -> Self {

330 self.templates = templates;

331 self

332 }

333

334 pub async fn register_default_templates(self) -> Self {

335 self.templates

336 .register("verify_code", templates::verify_code::render)

337 .await;

338 self.templates

339 .register("reset_link", templates::reset_link::render)

340 .await;

341 self.templates

342 .register(

343 "email_change_confirm",

344 templates::email_change_confirm::render,

345 )

346 .await;

347 self

348 }

349

350 pub async fn build(self) -> Result<EmailService, EmailError> {

351 if !self.email_config.enabled() {

352 info!("email disabled at config; transport=None, queue=None");

353 return Ok(EmailService {

354 inner: Arc::new(EmailServiceInner {

355 config: Arc::new(self.email_config),

356 transport: None,

357 from: None,

358 queue: None,

359 templates: self.templates,

360 max_attachment_bytes: 0,

361 }),

362 });

363 }

364

365 self.email_config

366 .is_valid()

367 .map_err(EmailError::InvalidConfig)?;

368

369 let from_addr = self

370 .email_config

371 .from_address()

372 .ok_or_else(|| EmailError::Config("email.from_address missing".into()))?;

373 let parsed: Address = from_addr.parse()?;

374 let from = match self.email_config.from_name() {

375 Some(name) => Mailbox::new(Some(name.to_string()), parsed),

376 None => Mailbox::new(None, parsed),

377 };

378

379 let smtp_host = self.email_config.smtp_host().to_string();

380 let smtp_port = self.email_config.smtp_port();

381 let tls_mode = match smtp_port {

382 465 => "implicit-tls",

383 587 => "starttls",

384 _ => "opportunistic",

385 };

386 let transport = EmailTransportBuilder::new(

387 smtp_host.clone(),

388 smtp_port,

389 self.email_config.smtp_username().map(|s| s.to_string()),

390 self.email_config.smtp_password().map(|s| s.to_string()),

391 )

392 .connection_timeout(std::time::Duration::from_millis(

393 self.email_config.connection_timeout_ms(),

394 ))

395 .build()?;

396 info!(host = %smtp_host, port = smtp_port, tls = tls_mode, "smtp transport ready");

397

398 let queue = match (&self.redis_url, &self.queue_config) {

399 (Some(url), Some(qcfg)) if qcfg.enabled() => {

400 let queue = EmailQueue::connect(

401 url,

402 QueueSettings {

403 channel_prefix: qcfg.channel_prefix().to_string(),

404 max_retries: qcfg.max_retries(),

405 retry_backoff: qcfg.retry_backoff(),

406 enabled: true,

407 },

408 )

409 .await?;

410 info!(

411 prefix = %qcfg.channel_prefix(),

412 max_retries = qcfg.max_retries(),

413 "redis queue connected"

414 );

415 Some(queue)

416 }

417 _ => {

418 info!("redis queue disabled; pubsub subscribers will not start");

419 None

420 }

421 };

422

423 let max_attachment_bytes = self.email_config.max_attachment_bytes();

424

425 Ok(EmailService {

426 inner: Arc::new(EmailServiceInner {

427 config: Arc::new(self.email_config),

428 transport: Some(transport),

429 from: Some(from),

430 queue,

431 templates: self.templates,

432 max_attachment_bytes,

433 }),

434 })

435 }

436}

437

438impl Default for EmailService {

439 fn default() -> Self {

440 EmailService {

441 inner: Arc::new(EmailServiceInner {

442 config: Arc::new(config::email::EmailConfig::default()),

443 transport: None,

444 from: None,

445 queue: None,

446 templates: TemplateRegistry::new(),

447 max_attachment_bytes: 0,

448 }),

449 }

450 }

451}

Deleted lib/email/src/subscriber.rs +0 −230

@@ -1,230 +0,0 @@

1use std::time::Duration;

2

3use tracing::{Instrument as _, debug, info};

4

5use crate::error::EmailError;

6use crate::events::{

7 CHANNEL_BATCH, CHANNEL_RENDER, CHANNEL_SEND, EmailFailed, RenderAndSendEmailRequest,

8 SendBatchEmailRequest, SendEmailRequest, now_ms,

9};

10use crate::queue::{SubscribeHandle, TypedMessageMeta};

11use crate::service::EmailService;

12

13pub struct EmailSubscribers {

14 single: Option<SubscribeHandle<SendEmailRequest>>,

15 batch: Option<SubscribeHandle<SendBatchEmailRequest>>,

16 render: Option<SubscribeHandle<RenderAndSendEmailRequest>>,

17}

18

19impl EmailSubscribers {

20 pub fn empty() -> Self {

21 Self {

22 single: None,

23 batch: None,

24 render: None,

25 }

26 }

27

28 pub async fn shutdown(self, timeout: Duration) -> Result<(), EmailError> {

29 let s = self.single;

30 let b = self.batch;

31 let r = self.render;

32 let mut last_err: Option<EmailError> = None;

33 if let Some(h) = s

34 && let Err(e) = h.shutdown(timeout).await

35 {

36 last_err = Some(e);

37 }

38 if let Some(h) = b

39 && let Err(e) = h.shutdown(timeout).await

40 {

41 last_err = Some(e);

42 }

43 if let Some(h) = r

44 && let Err(e) = h.shutdown(timeout).await

45 {

46 last_err = Some(e);

47 }

48 match last_err {

49 Some(e) => Err(e),

50 None => Ok(()),

51 }

52 }

53}

54

55pub async fn start_subscribers(service: EmailService) -> Result<EmailSubscribers, EmailError> {

56 let queue = match service.queue() {

57 Some(q) if q.enabled() => q,

58 _ => {

59 info!("email queue disabled; skipping pubsub subscribers");

60 return Ok(EmailSubscribers::empty());

61 }

62 };

63 let group = service.config().smtp_host().to_string();

64 info!(

65 group = %group,

66 prefix = %queue.config().channel_prefix,

67 "subscribing to email channels"

68 );

69

70 let single = queue

71 .subscribe(CHANNEL_SEND, &group, {

72 let svc = service.clone();

73 move |req: SendEmailRequest, _meta: TypedMessageMeta| {

74 let svc = svc.clone();

75 let span = tracing::info_span!("email.send", request_id = %req.request_id);

76 crate::trace::adopt(&span, req.trace.as_deref());

77 async move {

78 let recipient_count = req.to.len() + req.cc.len() + req.bcc.len();

79 let request_id = req.request_id;

80 let metadata = req.metadata.clone();

81 debug!(

82 request_id = %request_id,

83 recipients = recipient_count,

84 channel = CHANNEL_SEND,

85 "handler invoked"

86 );

87 match svc.send(req).await {

88 Ok(sent) => {

89 debug!(

90 request_id = %sent.request_id,

91 smtp_message_id = %sent.smtp_message_id,

92 "smtp send ok"

93 );

94 if let Err(err) = svc.publish_sent(&sent).await {

95 tracing::warn!(error = %err, "publish EmailSent failed");

96 }

97 }

98 Err(err) => {

99 let failed = EmailFailed {

100 request_id,

101 error: err.to_string(),

102 recipient_count,

103 failed_at_ms: now_ms(),

104 metadata,

105 };

106 tracing::error!(

107 request_id = %request_id,

108 error = %err,

109 "email send failed"

110 );

111 if let Err(pub_err) = svc.publish_failed(&failed).await {

112 tracing::warn!(error = %pub_err, "publish EmailFailed failed");

113 }

114 }

115 }

116 Ok(())

117 }

118 .instrument(span)

119 }

120 })

121 .await?;

122 info!(channel = CHANNEL_SEND, "subscribed channel");

123

124 let batch = queue

125 .subscribe(CHANNEL_BATCH, &group, {

126 let svc = service.clone();

127 move |req: SendBatchEmailRequest, _meta: TypedMessageMeta| {

128 let svc = svc.clone();

129 let span = tracing::info_span!("email.batch", request_id = %req.request_id);

130 crate::trace::adopt(&span, req.trace.as_deref());

131 async move {

132 let total = req.recipients.len();

133 let request_id = req.request_id;

134 debug!(

135 request_id = %request_id,

136 total,

137 channel = CHANNEL_BATCH,

138 "batch handler invoked"

139 );

140 match svc.send_batch(req).await {

141 Ok(_) => {

142 tracing::debug!(request_id = %request_id, total, "batch completed");

143 }

144 Err(err) => {

145 let failed = EmailFailed {

146 request_id,

147 error: err.to_string(),

148 recipient_count: total,

149 failed_at_ms: now_ms(),

150 metadata: Default::default(),

151 };

152 tracing::error!(

153 request_id = %request_id,

154 error = %err,

155 "batch failed"

156 );

157 if let Err(pub_err) = svc.publish_failed(&failed).await {

158 tracing::warn!(error = %pub_err, "publish EmailFailed failed");

159 }

160 }

161 }

162 Ok(())

163 }

164 .instrument(span)

165 }

166 })

167 .await?;

168 info!(channel = CHANNEL_BATCH, "subscribed channel");

169

170 let render = queue

171 .subscribe(CHANNEL_RENDER, &group, {

172 let svc = service.clone();

173 move |req: RenderAndSendEmailRequest, _meta: TypedMessageMeta| {

174 let svc = svc.clone();

175 let span = tracing::info_span!("email.render", request_id = %req.request_id);

176 crate::trace::adopt(&span, req.trace.as_deref());

177 async move {

178 let request_id = req.request_id;

179 let recipient_count = req.to.len() + req.cc.len() + req.bcc.len();

180 let metadata = req.metadata.clone();

181 debug!(

182 request_id = %request_id,

183 template = %req.template,

184 recipients = recipient_count,

185 channel = CHANNEL_RENDER,

186 "render handler invoked"

187 );

188 match svc.render_and_send(req).await {

189 Ok(sent) => {

190 debug!(

191 request_id = %sent.request_id,

192 smtp_message_id = %sent.smtp_message_id,

193 "render+send ok"

194 );

195 if let Err(err) = svc.publish_sent(&sent).await {

196 tracing::warn!(error = %err, "publish EmailSent failed");

197 }

198 }

199 Err(err) => {

200 let failed = EmailFailed {

201 request_id,

202 error: err.to_string(),

203 recipient_count,

204 failed_at_ms: now_ms(),

205 metadata,

206 };

207 tracing::error!(

208 request_id = %request_id,

209 error = %err,

210 "render+send failed"

211 );

212 if let Err(pub_err) = svc.publish_failed(&failed).await {

213 tracing::warn!(error = %pub_err, "publish EmailFailed failed");

214 }

215 }

216 }

217 Ok(())

218 }

219 .instrument(span)

220 }

221 })

222 .await?;

223 info!(channel = CHANNEL_RENDER, "subscribed channel");

224

225 Ok(EmailSubscribers {

226 single: Some(single),

227 batch: Some(batch),

228 render: Some(render),

229 })

230}

Deleted lib/email/src/template.rs +0 −57

@@ -1,57 +0,0 @@

1use std::collections::HashMap;

2use std::sync::Arc;

3

4use maud::Markup;

5use tokio::sync::RwLock;

6use tracing::debug;

7

8use crate::error::EmailError;

9

10pub type TemplateFn = Arc<dyn Fn(&serde_json::Value) -> Markup + Send + Sync>;

11

12#[derive(Default)]

13pub struct TemplateRegistry {

14 inner: RwLock<HashMap<String, TemplateFn>>,

15}

16

17impl TemplateRegistry {

18 pub fn new() -> Self {

19 Self::default()

20 }

21

22 pub async fn register<F>(&self, name: &str, f: F)

23 where

24 F: Fn(&serde_json::Value) -> Markup + Send + Sync + 'static,

25 {

26 let mut guard = self.inner.write().await;

27 let replacing = guard.contains_key(name);

28 guard.insert(name.to_string(), Arc::new(f));

29 debug!(template = %name, replacing, "template registered");

30 }

31

32 pub async fn render(&self, name: &str, vars: &serde_json::Value) -> Result<String, EmailError> {

33 let guard = self.inner.read().await;

34 let Some(f) = guard.get(name) else {

35 debug!(template = %name, "template not found");

36 return Err(EmailError::TemplateNotFound(name.to_string()));

37 };

38 let markup = f(vars).into_string();

39 debug!(

40 template = %name,

41 bytes = markup.len(),

42 vars_bytes = vars.to_string().len(),

43 "template rendered"

44 );

45 Ok(markup)

46 }

47

48 pub async fn contains(&self, name: &str) -> bool {

49 let guard = self.inner.read().await;

50 guard.contains_key(name)

51 }

52

53 pub async fn names(&self) -> Vec<String> {

54 let guard = self.inner.read().await;

55 guard.keys().cloned().collect()

56 }

57}

Deleted lib/email/src/trace.rs +0 −48

@@ -1,48 +0,0 @@

1use std::collections::HashMap;

2

3use opentelemetry::Context;

4use opentelemetry::global;

5use opentelemetry::propagation::{Extractor, Injector};

6use tracing::Span;

7use tracing_opentelemetry::OpenTelemetrySpanExt;

8

9#[derive(Default)]

10struct Carrier(HashMap<String, String>);

11

12impl Injector for Carrier {

13 fn set(&mut self, key: &str, value: String) {

14 self.0.insert(key.to_string(), value);

15 }

16}

17

18impl Extractor for Carrier {

19 fn get(&self, key: &str) -> Option<&str> {

20 self.0.get(key).map(String::as_str)

21 }

22

23 fn keys(&self) -> Vec<&str> {

24 self.0.keys().map(String::as_str).collect()

25 }

26}

27

28pub fn capture() -> Option<String> {

29 let mut carrier = Carrier::default();

30 global::get_text_map_propagator(|propagator| {

31 propagator.inject_context(&Span::current().context(), &mut carrier);

32 });

33 carrier.0.get("traceparent").cloned()

34}

35

36pub fn adopt(span: &Span, traceparent: Option<&str>) {

37 let Some(traceparent) = traceparent else {

38 return;

39 };

40 let mut carrier = Carrier::default();

41 carrier

42 .0

43 .insert("traceparent".to_string(), traceparent.to_string());

44 let parent = global::get_text_map_propagator(|propagator| {

45 propagator.extract_with_context(&Context::new(), &carrier)

46 });

47 let _ = span.set_parent(parent);

48}

Deleted lib/email/src/transport.rs +0 −213

@@ -1,213 +0,0 @@

1use std::time::{Duration, Instant};

2

3use base64::Engine;

4use lettre::AsyncTransport;

5use lettre::Tokio1Executor;

6use lettre::address::Address;

7use lettre::message::header::ContentType;

8use lettre::message::{

9 Attachment as LettreAttachment, Mailbox, Message, MessageBuilder, MultiPart, SinglePart,

10};

11use lettre::transport::smtp::AsyncSmtpTransport;

12use lettre::transport::smtp::authentication::{Credentials, Mechanism};

13use tracing::debug;

14

15use crate::error::EmailError;

16use crate::events::{EmailAddress, SendEmailRequest};

17

18pub struct EmailTransport {

19 inner: AsyncSmtpTransport<Tokio1Executor>,

20}

21

22impl EmailTransport {

23 pub async fn send(&self, msg: Message) -> Result<String, EmailError> {

24 let started = Instant::now();

25 let response = self.inner.send(msg).await?;

26 let id = response.message().next().unwrap_or_default().to_string();

27 let id_short: String = id.chars().take(16).collect();

28 debug!(

29 elapsed_ms = started.elapsed().as_millis() as u64,

30 message_id = %id_short,

31 "smtp response received"

32 );

33 Ok(id)

34 }

35}

36

37impl Clone for EmailTransport {

38 fn clone(&self) -> Self {

39 Self {

40 inner: self.inner.clone(),

41 }

42 }

43}

44

45pub struct EmailTransportBuilder {

46 host: String,

47 port: u16,

48 username: Option<String>,

49 password: Option<String>,

50 connection_timeout: Option<Duration>,

51}

52

53impl EmailTransportBuilder {

54 pub fn new(

55 host: impl Into<String>,

56 port: u16,

57 username: Option<String>,

58 password: Option<String>,

59 ) -> Self {

60 Self {

61 host: host.into(),

62 port,

63 username,

64 password,

65 connection_timeout: None,

66 }

67 }

68

69 pub fn connection_timeout(mut self, dur: Duration) -> Self {

70 self.connection_timeout = Some(dur);

71 self

72 }

73

74 pub fn build(self) -> Result<EmailTransport, EmailError> {

75 if self.host.is_empty() {

76 return Err(EmailError::Config("smtp host is empty".into()));

77 }

78 let mut builder = if self.port == 465 {

79 AsyncSmtpTransport::<Tokio1Executor>::relay(&self.host)?

80 } else if self.port == 587 {

81 AsyncSmtpTransport::<Tokio1Executor>::starttls_relay(&self.host)?

82 } else {

83 AsyncSmtpTransport::<Tokio1Executor>::builder_dangerous(self.host.clone())

84 };

85 builder = builder.port(self.port);

86 if let Some(ct) = self.connection_timeout {

87 builder = builder.timeout(Some(ct));

88 }

89 if let (Some(u), Some(p)) = (&self.username, &self.password) {

90 builder = builder

91 .credentials(Credentials::new(u.clone(), p.clone()))

92 .authentication(vec![Mechanism::Plain]);

93 }

94 debug!(

95 host = %self.host,

96 port = self.port,

97 has_credentials = self.username.is_some() && self.password.is_some(),

98 timeout_ms = self.connection_timeout.map(|d| d.as_millis() as u64).unwrap_or(0),

99 "smtp transport builder configured"

100 );

101 Ok(EmailTransport {

102 inner: builder.build(),

103 })

104 }

105}

106

107fn to_mailbox(addr: &EmailAddress) -> Result<Mailbox, EmailError> {

108 let parsed: Address = addr.address.parse()?;

109 let mb = match &addr.name {

110 Some(name) if !name.is_empty() => Mailbox::new(Some(name.clone()), parsed),

111 _ => Mailbox::new(None, parsed),

112 };

113 Ok(mb)

114}

115

116pub fn build_message(

117 req: &SendEmailRequest,

118 from: Mailbox,

119 max_attachment_bytes: usize,

120) -> Result<Message, EmailError> {

121 let has_text = req.text.as_deref().is_some_and(|s| !s.is_empty());

122 let has_html = req.html.as_deref().is_some_and(|s| !s.is_empty());

123 let recipient_count = req.to.len() + req.cc.len() + req.bcc.len();

124 debug!(

125 subject_len = req.subject.len(),

126 recipients = recipient_count,

127 attachments = req.attachments.len(),

128 has_text,

129 has_html,

130 "building email message"

131 );

132 if !has_text && !has_html {

133 return Err(EmailError::Missing(

134 "email body requires text or html".into(),

135 ));

136 }

137

138 let mut builder = MessageBuilder::new()

139 .from(from)

140 .subject(req.subject.clone());

141

142 for addr in &req.to {

143 let mb = to_mailbox(addr)?;

144 builder = builder.to(mb);

145 }

146 for addr in &req.cc {

147 let mb = to_mailbox(addr)?;

148 builder = builder.cc(mb);

149 }

150 for addr in &req.bcc {

151 let mb = to_mailbox(addr)?;

152 builder = builder.bcc(mb);

153 }

154 if let Some(reply) = &req.reply_to {

155 let mb = to_mailbox(reply)?;

156 builder = builder.reply_to(mb);

157 }

158

159 let body_part: MultiPart = match (has_text, has_html) {

160 (true, true) => {

161 let text = SinglePart::plain(req.text.as_deref().unwrap_or_default().to_string());

162 let html = SinglePart::html(req.html.as_deref().unwrap_or_default().to_string());

163 MultiPart::alternative().singlepart(text).singlepart(html)

164 }

165 (true, false) => MultiPart::alternative().singlepart(SinglePart::plain(

166 req.text.as_deref().unwrap_or_default().to_string(),

167 )),

168 (false, true) => MultiPart::alternative().singlepart(SinglePart::html(

169 req.html.as_deref().unwrap_or_default().to_string(),

170 )),

171 (false, false) => unreachable!(),

172 };

173

174 let message = if req.attachments.is_empty() {

175 builder.multipart(body_part)?

176 } else {

177 let mut mixed = MultiPart::mixed().multipart(body_part);

178 for att in &req.attachments {

179 let bytes =

180 base64::engine::general_purpose::STANDARD.decode(att.data_b64.as_bytes())?;

181 if bytes.len() > max_attachment_bytes {

182 return Err(EmailError::AttachmentTooLarge {

183 actual: bytes.len(),

184 max: max_attachment_bytes,

185 });

186 }

187 let ct_str = if att.content_type.is_empty() {

188 "application/octet-stream".to_string()

189 } else {

190 att.content_type.clone()

191 };

192 let content_type: ContentType =

193 ct_str

194 .parse()

195 .map_err(|e: lettre::message::header::ContentTypeErr| {

196 EmailError::Other(format!("invalid content_type {ct_str:?}: {e}"))

197 })?;

198 if let Some(cid) = &att.content_id {

199 mixed = mixed.singlepart(

200 LettreAttachment::new_inline_with_name(cid.clone(), att.filename.clone())

201 .body(bytes, content_type),

202 );

203 } else {

204 mixed = mixed.singlepart(

205 LettreAttachment::new(att.filename.clone()).body(bytes, content_type),

206 );

207 }

208 }

209 builder.multipart(mixed)?

210 };

211

212 Ok(message)

213}

Deleted lib/email/Cargo.toml +0 −24

@@ -1,24 +0,0 @@

1[package]

2name = "email"

3version.workspace = true

4edition.workspace = true

5

6[dependencies]

7config = { workspace = true }

8i18n = { workspace = true }

9lettre = { workspace = true }

10maud = { workspace = true }

11opentelemetry = { workspace = true }

12redis = { workspace = true }

13serde = { workspace = true, features = ["derive"] }

14serde_json = { workspace = true }

15tokio = { workspace = true, features = ["signal", "rt-multi-thread", "macros"] }

16tokio-util = { workspace = true }

17tracing = { workspace = true }

18tracing-opentelemetry = { workspace = true }

19futures = { workspace = true }

20async-trait = { workspace = true }

21uuid = { workspace = true, features = ["v7", "serde"] }

22base64 = { workspace = true }

23bytes = { workspace = true }

24time = { workspace = true }

No newline at end of file

Modified lib/export/src/error.rs +1 −12 View lib/export/src/error.rs

@@ -7,10 +7,8 @@

77 Database(DbErr),

88 Redis(redis::RedisError),

99 Storage(storage::StorageError),

10 Email(email::EmailError),

1110 InvalidConfig(String),

1211 Connection(String),

13 Disabled,

1412 Unsupported(String),

1513}

1614

@@ -20,11 +18,9 @@

2018 ExportError::Database(err) => write!(f, "database error: {err}"),

2119 ExportError::Redis(err) => write!(f, "redis error: {err}"),

2220 ExportError::Storage(err) => write!(f, "storage error: {err}"),

23 ExportError::Email(err) => write!(f, "email error: {err}"),

2421 ExportError::InvalidConfig(msg) => write!(f, "invalid config: {msg}"),

2522 ExportError::Connection(msg) => write!(f, "connection error: {msg}"),

26 ExportError::Disabled => write!(f, "email service is disabled"),

27 ExportError::Unsupported(msg) => write!(f, "unsupported email operation: {msg}"),

23 ExportError::Unsupported(msg) => write!(f, "unsupported operation: {msg}"),

2824 }

2925 }

3026}

@@ -35,7 +31,6 @@

3531 ExportError::Database(err) => Some(err),

3632 ExportError::Redis(err) => Some(err),

3733 ExportError::Storage(err) => Some(err),

38 ExportError::Email(err) => Some(err),

3934 _ => None,

4035 }

4136 }

@@ -56,11 +51,5 @@

5651impl From<storage::StorageError> for ExportError {

5752 fn from(err: storage::StorageError) -> Self {

5853 ExportError::Storage(err)

59 }

60}

61

62impl From<email::EmailError> for ExportError {

63 fn from(err: email::EmailError) -> Self {

64 ExportError::Email(err)

6554 }

6655}

Modified lib/export/src/lib.rs +15 −10 View lib/export/src/lib.rs

@@ -27,16 +27,14 @@

2727pub use storage::AppStorage;

2828pub use urls::{resolve_space_image_url, resolve_user_avatar_url};

2929

30pub use email::EmailService;

31

3230#[derive(Clone)]

3331pub struct AppState {

3432 pub database: Arc<AppDatabase>,

3533 pub redis: AppRedis,

3634 pub storage: Arc<AppStorage>,

37 pub email: EmailService,

3835 pub config: AppConfig,

3936 pub kenauth: Option<Arc<kenauth::KenAuthClient>>,

37 pub email_rpc: Option<Arc<kenauth::KenAuthClient>>,

4038}

4139

4240impl AppState {

@@ -53,12 +51,6 @@

5351 // `export::push::enqueue` 能自己判断该不该产生任务。

5452 redis.set_push_enabled(config.push.enabled);

5553 let storage = Arc::new(AppStorage::connect(config.s3.clone())?);

56 let redis_url = redis.config().url().to_string();

57 let builder = email::EmailServiceBuilder::new(config.email.clone())

58 .with_queue(redis_url, config.email_queue.clone())

59 .register_default_templates()

60 .await;

61 let email = builder.build().await?;

6254

6355 let kenauth = if config.oauth.enabled {

6456 let client = Arc::new(

@@ -71,13 +63,26 @@

7163 None

7264 };

7365

66 let email_rpc = if config.email.rpc.enabled() {

67 Some(Arc::new(

68 kenauth::KenAuthClient::connect_rpc(

69 &config.email.rpc.endpoint,

70 &config.email.rpc.rpc_key,

71 config.email.rpc.timeout_ms(),

72 )

73 .map_err(|err| ExportError::InvalidConfig(err.to_string()))?,

74 ))

75 } else {

76 kenauth.clone()

77 };

78

7479 Ok(Self {

7580 database,

7681 redis,

7782 storage,

78 email,

7983 config,

8084 kenauth,

85 email_rpc,

8186 })

8287 }

8388}

Modified lib/export/src/metrics.rs +0 −11 View lib/export/src/metrics.rs

@@ -61,13 +61,6 @@

6161 .build()

6262});

6363

64static EMAIL_SENT: LazyLock<Counter<u64>> = LazyLock::new(|| {

65 meter()

66 .u64_counter("email.sent")

67 .with_description("Email sends by outcome")

68 .build()

69});

70

7164static PUSH_DELIVERIES: LazyLock<Counter<u64>> = LazyLock::new(|| {

7265 meter()

7366 .u64_counter("push.deliveries")

@@ -150,10 +143,6 @@

150143

151144pub fn ci_webhook_delivery(outcome: &'static str) {

152145 CI_WEBHOOK_DELIVERIES.add(1, &[KeyValue::new("outcome", outcome)]);

153}

154

155pub fn email_sent(outcome: &'static str) {

156 EMAIL_SENT.add(1, &[KeyValue::new("outcome", outcome)]);

157146}

158147

159148pub fn push_delivery(outcome: &'static str) {

Modified lib/export/Cargo.toml +0 −1 View lib/export/Cargo.toml

@@ -9,7 +9,6 @@

99sea-orm = { workspace = true }

1010redis = { workspace = true }

1111tokio = { workspace = true, features = ["macros", "rt-multi-thread", "signal"] }

12email = { workspace = true }

1312kenauth = { workspace = true }

1413imaging = { workspace = true }

1514axum = { workspace = true }

Modified lib/issues/src/services/service.rs +5 −7 View lib/issues/src/services/service.rs

@@ -2,7 +2,6 @@

22use std::sync::Arc;

33

44use config::AppConfig;

5use email::EmailService;

65use export::AppState;

76use export::database::AppDatabase;

87use i18n;

@@ -59,10 +58,6 @@

5958 pub async fn queue_search_issue(&self, issue_id: Uuid) -> Result<(), IssueError> {

6059 search::outbox::enqueue_issue(self.db().write(), issue_id).await?;

6160 Ok(())

62 }

63

64 pub fn email(&self) -> &EmailService {

65 &self.state.email

6661 }

6762

6863 pub async fn load_user_briefs(

@@ -260,8 +255,11 @@

260255 text: &str,

261256 html: &str,

262257 ) -> Result<(), IssueError> {

263 self.state

264 .email

258 let Some(client) = self.state.email_rpc.as_ref() else {

259 tracing::warn!(to, "email rpc disabled; skipping issue mail");

260 return Ok(());

261 };

262 client

265263 .send_mail(to, subject, text, html)

266264 .await

267265 .map_err(|error| IssueError::Internal(format!("email send failed: {error}")))

Modified lib/issues/Cargo.toml +1 −1 View lib/issues/Cargo.toml

@@ -8,9 +8,9 @@

88config = { workspace = true }

99export = { workspace = true }

1010i18n = { workspace = true }

11email = { workspace = true }

1211models = { workspace = true }

1312jwt = { workspace = true }

13kenauth = { workspace = true }

1414space = { workspace = true }

1515axum = { workspace = true }

1616http = { workspace = true }

Added lib/kenauth/proto/internal/v1/email.proto +120 −0 View lib/kenauth/proto/internal/v1/email.proto

@@ -0,0 +1,120 @@

1syntax = "proto3";

2

3// 邮件发送入口（内部面）。

4//

5// 调用方：api、notify 以及任何需要发信的内部服务，通过 `x-rpc-key` 头鉴权。

6// 服务端只把请求放进进程内有界队列并立即返回，SMTP 由内嵌 worker 异步完成；

7// 队列满时按反压阻塞，超过 `email.enqueue_timeout_ms` 才返回 RESOURCE_EXHAUSTED。

8package auth.internal.v1;

9

10message EmailAddress {

11 // 邮箱地址，非空。

12 string address = 1;

13 // 显示名，可为空。

14 string name = 2;

15}

16

17// 附件内容用 base64 传输（proto 的 bytes 字段在本仓库不参与 JSON 编码）。

18message EmailAttachment {

19 string filename = 1;

20 string content_type = 2;

21 string data_b64 = 3;

22 // 内联附件的 Content-ID，普通附件留空。

23 string content_id = 4;

24}

25

26message SendEmailRequest {

27 // 留空时服务端生成 UUIDv7，并在响应里回传。

28 string request_id = 1;

29 // 收件人，至少一个。

30 repeated EmailAddress to = 2;

31 repeated EmailAddress cc = 3;

32 repeated EmailAddress bcc = 4;

33 // 必填。

34 string subject = 5;

35 // 纯文本正文；与 html 至少给一个。

36 string text = 6;

37 string html = 7;

38 EmailAddress reply_to = 8;

39 repeated EmailAttachment attachments = 9;

40 // 透传到投递事件里的业务标签。

41 map<string, string> metadata = 10;

42 // 链路追踪 ID，只用于日志。

43 string trace_id = 11;

44}

45

46message SendEmailResponse {

47 string request_id = 1;

48 // 已入队即为 true；入队失败由 gRPC 状态码表达。

49 bool accepted = 2;

50 // 入队后的积压条数。

51 uint32 queue_depth = 3;

52}

53

54message BatchRecipient {

55 string request_id = 1;

56 EmailAddress to = 2;

57 // 留空时回落到批量的 default_subject。

58 string subject = 3;

59 string text = 4;

60 string html = 5;

61 map<string, string> metadata = 6;

62}

63

64message SendBatchEmailRequest {

65 string request_id = 1;

66 string default_subject = 2;

67 string default_text = 3;

68 string default_html = 4;

69 // 逐个收件人单独发信，单个失败不影响其余。

70 repeated BatchRecipient recipients = 5;

71 map<string, string> metadata = 6;

72 string trace_id = 7;

73}

74

75message SendBatchEmailResponse {

76 string request_id = 1;

77 uint32 accepted = 2;

78 uint32 queue_depth = 3;

79}

80

81message RenderAndSendEmailRequest {

82 string request_id = 1;

83 // 模板名，如 verify_code / reset_link / email_change_confirm。

84 string template = 2;

85 // 模板变量（JSON 对象字符串）；locale 缺省时按服务端默认语言渲染。

86 string vars_json = 3;

87 repeated EmailAddress to = 4;

88 repeated EmailAddress cc = 5;

89 repeated EmailAddress bcc = 6;

90 string subject = 7;

91 string text = 8;

92 string locale = 9;

93 map<string, string> metadata = 10;

94 string trace_id = 11;

95}

96

97message EmailQueueStatus {

98 // 当前积压条数。

99 uint32 depth = 1;

100 // 队列容量。

101 uint32 capacity = 2;

102 // SMTP 是否已配置；false 时所有发送请求都会被拒绝。

103 bool smtp_enabled = 3;

104 // 成功投递计数。

105 uint64 sent = 4;

106 // 失败计数。

107 uint64 failed = 5;

108}

109

110message EmailQueueStatusRequest {}

111

112service EmailService {

113 // 入队一封邮件，立即返回。

114 rpc SendEmail(SendEmailRequest) returns (SendEmailResponse);

115 rpc SendBatchEmail(SendBatchEmailRequest) returns (SendBatchEmailResponse);

116 // 用模板渲染 HTML 后入队。

117 rpc RenderAndSendEmail(RenderAndSendEmailRequest) returns (SendEmailResponse);

118 // 队列观测：积压、容量、累计成功/失败。

119 rpc GetQueueStatus(EmailQueueStatusRequest) returns (EmailQueueStatus);

120}

Modified lib/kenauth/src/client.rs +60 −6 View lib/kenauth/src/client.rs

@@ -1,3 +1,4 @@

1use std::collections::HashMap;

12use std::time::Duration;

23

34use config::oauth::OAuthConfig;

@@ -23,7 +24,7 @@

2324 } else {

2425 Some(

2526 MetadataValue::try_from(rpc_key)

26 .map_err(|_| KenAuthError::Config("oauth.rpc_key is not ASCII".into()))?,

27 .map_err(|_| KenAuthError::Config("rpc_key is not ASCII".into()))?,

2728 )

2829 };

2930 Ok(Self { key })

@@ -61,18 +62,26 @@

6162 if !config.enabled {

6263 return Err(KenAuthError::Config("oauth.enabled is false".into()));

6364 }

64 if config.grpc_endpoint.is_empty() {

65 return Err(KenAuthError::Config("oauth.grpc_endpoint is empty".into()));

65 Self::connect_rpc(&config.grpc_endpoint, &config.rpc_key, config.http_timeout_ms)

66 }

67

68 pub fn connect_rpc(

69 grpc_endpoint: &str,

70 rpc_key: &str,

71 http_timeout_ms: u64,

72 ) -> Result<Self, KenAuthError> {

73 if grpc_endpoint.is_empty() {

74 return Err(KenAuthError::Config("grpc endpoint is empty".into()));

6675 }

67 let timeout = Duration::from_millis(config.http_timeout_ms.max(200));

68 let endpoint = Endpoint::from_shared(config.grpc_endpoint.clone())

76 let timeout = Duration::from_millis(http_timeout_ms.max(200));

77 let endpoint = Endpoint::from_shared(grpc_endpoint.to_string())

6978 .map_err(|err| KenAuthError::Config(err.to_string()))?

7079 .connect_timeout(timeout)

7180 .timeout(timeout);

7281 let channel = endpoint.connect_lazy();

7382 Ok(Self {

7483 channel,

75 interceptor: RpcInterceptor::new(&config.rpc_key)?,

84 interceptor: RpcInterceptor::new(rpc_key)?,

7685 read_timeout: Duration::from_millis(300),

7786 write_timeout: timeout.max(Duration::from_secs(2)),

7887 })

@@ -114,6 +123,15 @@

114123 )

115124 }

116125

126 pub fn email(

127 &self,

128 ) -> proto::internal::v1::email_service_client::EmailServiceClient<Intercepted> {

129 proto::internal::v1::email_service_client::EmailServiceClient::with_interceptor(

130 self.channel.clone(),

131 self.interceptor.clone(),

132 )

133 }

134

117135 pub fn session(&self) -> proto::v1::session_service_client::SessionServiceClient<Intercepted> {

118136 proto::v1::session_service_client::SessionServiceClient::with_interceptor(

119137 self.channel.clone(),

@@ -190,6 +208,42 @@

190208 request

191209 }

192210

211 /// Enqueue one plain text / HTML mail on `kenoauth`.

212 ///

213 /// `kenoauth` owns SMTP, retries and backpressure; a `Ok` here means the

214 /// message was accepted into its queue, while delivery outcome is visible

215 /// through `GetQueueStatus` / the delivery events on that side.

216 pub async fn send_mail(

217 &self,

218 to: &str,

219 subject: &str,

220 text: &str,

221 html: &str,

222 ) -> Result<(), KenAuthError> {

223 let mut client = self.email();

224 let request = proto::internal::v1::SendEmailRequest {

225 request_id: Uuid::now_v7().to_string(),

226 to: vec![proto::internal::v1::EmailAddress {

227 address: to.to_string(),

228 name: String::new(),

229 }],

230 cc: Vec::new(),

231 bcc: Vec::new(),

232 subject: subject.to_string(),

233 text: text.to_string(),

234 html: html.to_string(),

235 reply_to: None,

236 attachments: Vec::new(),

237 metadata: HashMap::new(),

238 trace_id: String::new(),

239 };

240 client

241 .send_email(self.write_request(request))

242 .await

243 .map_err(KenAuthError::from_status)?;

244 Ok(())

245 }

246

193247 pub async fn get_session(

194248 &self,

195249 user_id: i64,

Modified lib/kenauth/build.rs +1 −0 View lib/kenauth/build.rs

@@ -16,6 +16,7 @@

1616 "proto/v1/ssh_key.proto",

1717 "proto/v1/token.proto",

1818 "proto/internal/v1/authority.proto",

19 "proto/internal/v1/email.proto",

1920 ];

2021

2122 tonic_prost_build::configure()

Modified lib/space/src/services/service.rs +5 −8 View lib/space/src/services/service.rs

@@ -1,7 +1,6 @@

11use std::sync::Arc;

22

33use config::AppConfig;

4use email::EmailService;

54use export::{AppDatabase, AppState, AppStorage};

65

76use uuid::Uuid;

@@ -30,10 +29,6 @@

3029 &self.state.storage

3130 }

3231

33 pub fn email(&self) -> &EmailService {

34 &self.state.email

35 }

36

3732 /// Language for one recipient's space mail.

3833 pub(crate) async fn recipient_locale(&self, uid: Option<i64>) -> i18n::UiLocale {

3934 let fallback = i18n::config_fallback(self.config().email.default_locale());

@@ -50,11 +45,13 @@

5045 text: &str,

5146 html: &str,

5247 ) -> Result<(), SpaceError> {

53 self.state

54 .email

48 let Some(client) = self.state.email_rpc.as_ref() else {

49 tracing::warn!(to, "email rpc disabled; skipping space mail");

50 return Ok(());

51 };

52 client

5553 .send_mail(to, subject, text, html)

5654 .await

5755 .map_err(|error| SpaceError::Internal(format!("email send failed: {error}")))

5856 }

5957}

60

Modified lib/space/Cargo.toml +0 −1 View lib/space/Cargo.toml

@@ -8,7 +8,6 @@

88async-trait = { workspace = true }

99config = { workspace = true }

1010export = { workspace = true }

11email = { workspace = true }

1211i18n = { workspace = true }

1312imaging = { workspace = true }

1413models = { workspace = true }

Modified lib/user/Cargo.toml +0 −1 View lib/user/Cargo.toml

@@ -10,7 +10,6 @@

1010models = { workspace = true }

1111i18n = { workspace = true }

1212imaging = { workspace = true }

13email = { workspace = true }

1413jwt = { workspace = true }

1514storage = { workspace = true }

1615axum = { workspace = true }

Modified scripts/dev.ps1 +3 −6 View scripts/dev.ps1

@@ -19,18 +19,16 @@

1919 im = @{ Exe = 'im'; Port = 4082; Health = $true }

2020 synchost = @{ Exe = 'synchost'; Port = 4083; Health = $true }

2121 searchhost = @{ Exe = 'searchhost'; Port = 0; Health = $false }

22 mails = @{ Exe = 'mails'; Port = 4084; Health = $true }

2322 webhook = @{ Exe = 'webhook'; Port = 3005; Health = $true }

2423 web = @{ Bun = 'web'; Port = 5173; Health = $false }

2524}

26$Order = @('migrate', 'apihost', 'githost', 'im', 'synchost', 'searchhost', 'mails', 'webhook', 'web')

27$Backends = @('migrate', 'apihost', 'githost', 'im', 'synchost', 'searchhost', 'mails', 'webhook')

25$Order = @('migrate', 'apihost', 'githost', 'im', 'synchost', 'searchhost', 'webhook', 'web')

26$Backends = @('migrate', 'apihost', 'githost', 'im', 'synchost', 'searchhost', 'webhook')

2827$Aliases = @{

2928 api = 'apihost'

3029 git = 'githost'

3130 chat = 'im'

3231 sync = 'synchost'

33 mail = 'mails'

3432 hook = 'webhook'

3533 ui = 'web'

3634}

@@ -40,7 +38,6 @@

4038 githost = 'http://127.0.0.1:9418 ssh://git@127.0.0.1:4022'

4139 im = 'ws://127.0.0.1:4082'

4240 synchost = 'http://127.0.0.1:4083/healthz'

43 mails = 'http://127.0.0.1:4084/healthz'

4441 webhook = 'http://127.0.0.1:3005'

4542 web = 'http://127.0.0.1:5173'

4643}

@@ -283,7 +280,7 @@

283280 continue

284281 }

285282 if (-not $Services.Contains($key)) {

286 throw "未知服务: $part；可用: $($Order -join ', ')（别名: api/git/chat/sync/mail/hook/ui/backends/frontends）"

283 throw "未知服务: $part；可用: $($Order -join ', ')（别名: api/git/chat/sync/hook/ui/backends/frontends）"

287284 }

288285 if (-not $names.Contains($key)) { $names.Add($key) }

289286 }

Modified scripts/dev.sh +3 −6 View scripts/dev.sh

@@ -12,8 +12,8 @@

1212DEV_HOST_KEY="$DEV_DIR/ssh_host_ed25519"

1313COMPOSE_FILE="$ROOT/docker-compose.dev.yml"

1414

15ORDER="apihost githost im synchost mails webhook web"

16BACKENDS="apihost githost im synchost mails webhook"

15ORDER="apihost githost im synchost webhook web"

16BACKENDS="apihost githost im synchost webhook"

1717INFRA="postgresql:5432 redis:6379 s3:9000 meilisearch:7700"

1818

1919if [ -t 1 ]; then

@@ -41,7 +41,6 @@

4141 githost) echo 9418 ;;

4242 im) echo 4082 ;;

4343 synchost) echo 4083 ;;

44 mails) echo 4084 ;;

4544 webhook) echo 3005 ;;

4645 web) echo 5173 ;;

4746 *) echo 0 ;;

@@ -61,7 +60,6 @@

6160 git) echo githost ;;

6261 chat) echo im ;;

6362 sync) echo synchost ;;

64 mail) echo mails ;;

6563 hook) echo webhook ;;

6664 ui) echo web ;;

6765 *) echo "$1" ;;

@@ -74,7 +72,6 @@

7472 githost) echo "http://127.0.0.1:9418 ssh://git@127.0.0.1:4022" ;;

7573 im) echo "ws://127.0.0.1:4082" ;;

7674 synchost) echo "http://127.0.0.1:4083/healthz" ;;

77 mails) echo "http://127.0.0.1:4084/healthz" ;;

7875 webhook) echo "http://127.0.0.1:3005" ;;

7976 web) echo "http://127.0.0.1:5173" ;;

8077 esac

@@ -588,7 +585,7 @@

588585 if ! contains "$key" $ORDER; then

589586 fail "未知服务: $part"

590587 info "可用: $ORDER"

591 info "别名: api git chat sync mail hook ui / backends / frontends"

588 info "别名: api git chat sync hook ui / backends / frontends"

592589 return 1

593590 fi

594591 out="$out $key"

Modified Cargo.lock +2 −397 View Cargo.lock

@@ -97,55 +97,6 @@

9797]

9898

9999[[package]]

100name = "agent"

101version = "1.0.0"

102dependencies = [

103 "ci-protocol",

104 "config",

105 "export",

106 "futures",

107 "git",

108 "i18n",

109 "kenauth",

110 "models",

111 "redis",

112 "reqwest 0.12.28",

113 "sea-orm",

114 "serde",

115 "serde_json",

116 "thiserror 2.0.20",

117 "time 0.3.55",

118 "tokio",

119 "tracing",

120 "utoipa",

121 "uuid",

122]

123

124[[package]]

125name = "agents"

126version = "1.0.0"

127dependencies = [

128 "agent",

129 "async-trait",

130 "axum",

131 "config",

132 "export",

133 "git",

134 "jwt",

135 "mcp-api",

136 "models",

137 "sandbox-rpc",

138 "sea-orm",

139 "serde",

140 "serde_json",

141 "space",

142 "time 0.3.55",

143 "tracing",

144 "utoipa",

145 "uuid",

146]

147

148[[package]]

149100name = "ahash"

150101version = "0.7.8"

151102source = "registry+https://github.com/rust-lang/crates.io-index"

@@ -180,25 +131,6 @@

180131]

181132

182133[[package]]

183name = "aihost"

184version = "1.0.0"

185dependencies = [

186 "agent",

187 "anyhow",

188 "axum",

189 "config",

190 "export",

191 "git",

192 "models",

193 "sea-orm",

194 "time 0.3.55",

195 "tokio",

196 "tokio-util",

197 "tracing",

198 "uuid",

199]

200

201[[package]]

202134name = "aliasable"

203135version = "0.1.3"

204136source = "registry+https://github.com/rust-lang/crates.io-index"

@@ -681,7 +613,6 @@

681613 "axum",

682614 "base64 0.23.1",

683615 "config",

684 "email",

685616 "export",

686617 "hkdf 0.13.0",

687618 "hmac 0.13.0",

@@ -1236,6 +1167,7 @@

12361167 "i18n",

12371168 "imaging",

12381169 "jwt",

1170 "kenauth",

12391171 "livekit-api",

12401172 "livekit-protocol",

12411173 "models",

@@ -1749,16 +1681,6 @@

17491681]

17501682

17511683[[package]]

1752name = "darling"

1753version = "0.24.1"

1754source = "registry+https://github.com/rust-lang/crates.io-index"

1755checksum = "ed17f5901b6630b993ca003def43f2f8ef4014fc13b047b57aad617ff32bc2ec"

1756dependencies = [

1757 "darling_core 0.24.1",

1758 "darling_macro 0.24.1",

1759]

1760

1761[[package]]

17621684name = "darling_core"

17631685version = "0.20.11"

17641686source = "registry+https://github.com/rust-lang/crates.io-index"

@@ -1786,19 +1708,6 @@

17861708]

17871709

17881710[[package]]

1789name = "darling_core"

1790version = "0.24.1"

1791source = "registry+https://github.com/rust-lang/crates.io-index"

1792checksum = "6837e2cf7485aaae18f86181d2f0e9a7ed297a025e220aeabf63fdebd3a2ddff"

1793dependencies = [

1794 "ident_case",

1795 "proc-macro2",

1796 "quote",

1797 "strsim",

1798 "syn 3.0.3",

1799]

1800

1801[[package]]

18021711name = "darling_macro"

18031712version = "0.20.11"

18041713source = "registry+https://github.com/rust-lang/crates.io-index"

@@ -1821,17 +1730,6 @@

18211730]

18221731

18231732[[package]]

1824name = "darling_macro"

1825version = "0.24.1"

1826source = "registry+https://github.com/rust-lang/crates.io-index"

1827checksum = "2ac7135c3ef02b2f7833bbeb1be5ba7f966dcde8a87c6b87f65a778d71a02785"

1828dependencies = [

1829 "darling_core 0.24.1",

1830 "quote",

1831 "syn 3.0.3",

1832]

1833

1834[[package]]

18351733name = "dashmap"

18361734version = "6.2.1"

18371735source = "registry+https://github.com/rust-lang/crates.io-index"

@@ -2108,12 +2006,6 @@

21082006checksum = "92773504d58c093f6de2459af4af33faa518c13451eb8f2b5698ed3d36e7c813"

21092007

21102008[[package]]

2111name = "dyn-clone"

2112version = "1.0.20"

2113source = "registry+https://github.com/rust-lang/crates.io-index"

2114checksum = "d0881ea181b1df73ff77ffaaf9c7544ecc11e82fba9b5f27b262a3c73a332555"

2115

2116[[package]]

21172009name = "ecdsa"

21182010version = "0.16.9"

21192011source = "registry+https://github.com/rust-lang/crates.io-index"

@@ -2242,46 +2134,6 @@

22422134]

22432135

22442136[[package]]

2245name = "email"

2246version = "1.0.0"

2247dependencies = [

2248 "async-trait",

2249 "base64 0.23.1",

2250 "bytes",

2251 "config",

2252 "futures",

2253 "i18n",

2254 "lettre",

2255 "maud",

2256 "opentelemetry",

2257 "redis",

2258 "serde",

2259 "serde_json",

2260 "time 0.3.55",

2261 "tokio",

2262 "tokio-util",

2263 "tracing",

2264 "tracing-opentelemetry",

2265 "uuid",

2266]

2267

2268[[package]]

2269name = "email-encoding"

2270version = "0.4.2"

2271source = "registry+https://github.com/rust-lang/crates.io-index"

2272checksum = "420b9da095f052ea597503e39073b5b3c522f7db933fbac202d91d24492693fd"

2273dependencies = [

2274 "base64 0.23.1",

2275 "memchr",

2276]

2277

2278[[package]]

2279name = "email_address"

2280version = "0.2.9"

2281source = "registry+https://github.com/rust-lang/crates.io-index"

2282checksum = "e079f19b08ca6239f47f8ba8509c11cf3ea30095831f7fed61441475edd8c449"

2283

2284[[package]]

22852137name = "encoding_rs"

22862138version = "0.8.35"

22872139source = "registry+https://github.com/rust-lang/crates.io-index"

@@ -2365,7 +2217,6 @@

23652217 "axum-otel-metrics",

23662218 "bytes",

23672219 "config",

2368 "email",

23692220 "http-body-util",

23702221 "imaging",

23712222 "kenauth",

@@ -4843,11 +4694,11 @@

48434694 "async-trait",

48444695 "axum",

48454696 "config",

4846 "email",

48474697 "export",

48484698 "http 0.2.12",

48494699 "i18n",

48504700 "jwt",

4701 "kenauth",

48514702 "models",

48524703 "sea-orm",

48534704 "search",

@@ -5172,33 +5023,6 @@

51725023]

51735024

51745025[[package]]

5175name = "lettre"

5176version = "0.11.23"

5177source = "registry+https://github.com/rust-lang/crates.io-index"

5178checksum = "f2c646bd5cc763b1087b15493e29a64be6147ba8f19342004fa52048ee596eae"

5179dependencies = [

5180 "async-trait",

5181 "base64 0.23.1",

5182 "email-encoding",

5183 "email_address",

5184 "fastrand",

5185 "futures-io",

5186 "futures-util",

5187 "httpdate",

5188 "idna 1.1.0",

5189 "mime 0.3.17",

5190 "nom",

5191 "percent-encoding 2.3.2",

5192 "quoted_printable",

5193 "rustls",

5194 "rustls-native-certs",

5195 "socket2 0.6.5",

5196 "tokio",

5197 "tokio-rustls",

5198 "url 2.5.8",

5199]

5200

5201[[package]]

52025026name = "lexical-core"

52035027version = "1.0.6"

52045028source = "registry+https://github.com/rust-lang/crates.io-index"

@@ -5473,20 +5297,6 @@

54735297]

54745298

54755299[[package]]

5476name = "mails"

5477version = "1.0.0"

5478dependencies = [

5479 "anyhow",

5480 "axum",

5481 "config",

5482 "email",

5483 "export",

5484 "tokio",

5485 "tracing",

5486 "tracing-subscriber",

5487]

5488

5489[[package]]

54905300name = "matchers"

54915301version = "0.2.0"

54925302source = "registry+https://github.com/rust-lang/crates.io-index"

@@ -5532,53 +5342,6 @@

55325342]

55335343

55345344[[package]]

5535name = "mcp"

5536version = "1.0.0"

5537dependencies = [

5538 "anyhow",

5539 "axum",

5540 "config",

5541 "export",

5542 "mcp-api",

5543 "tokio",

5544 "tower-http 0.7.0",

5545 "tracing",

5546]

5547

5548[[package]]

5549name = "mcp-api"

5550version = "1.0.0"

5551dependencies = [

5552 "agent",

5553 "auth",

5554 "axum",

5555 "base64 0.23.1",

5556 "cache",

5557 "config",

5558 "export",

5559 "git",

5560 "gitbaby",

5561 "ipnet",

5562 "issues",

5563 "models",

5564 "pulls",

5565 "rand 0.8.7",

5566 "rmcp",

5567 "schemars",

5568 "sea-orm",

5569 "serde",

5570 "serde_json",

5571 "sha2 0.11.0",

5572 "space",

5573 "thiserror 2.0.20",

5574 "time 0.3.55",

5575 "tokio",

5576 "tracing",

5577 "user",

5578 "uuid",

5579]

5580

5581[[package]]

55825345name = "md-5"

55835346version = "0.11.0"

55845347source = "registry+https://github.com/rust-lang/crates.io-index"

@@ -5842,15 +5605,6 @@

58425605]

58435606

58445607[[package]]

5845name = "nom"

5846version = "8.0.0"

5847source = "registry+https://github.com/rust-lang/crates.io-index"

5848checksum = "df9761775871bdef83bee530e60050f7e54b1105350d6884eb0fb4f46c2f9405"

5849dependencies = [

5850 "memchr",

5851]

5852

5853[[package]]

58545608name = "nonempty"

58555609version = "0.12.0"

58565610source = "registry+https://github.com/rust-lang/crates.io-index"

@@ -6469,12 +6223,6 @@

64696223checksum = "57c0d7b74b563b49d38dae00a0c37d4d6de9b432382b2892f0574ddcae73fd0a"

64706224

64716225[[package]]

6472name = "pastey"

6473version = "0.2.3"

6474source = "registry+https://github.com/rust-lang/crates.io-index"

6475checksum = "2ee67f1008b1ba2321834326597b8e186293b049a023cdef258527550b9935b4"

6476

6477[[package]]

64786226name = "pbjson"

64796227version = "0.6.0"

64806228source = "registry+https://github.com/rust-lang/crates.io-index"

@@ -7191,7 +6939,6 @@

71916939name = "pulls"

71926940version = "1.0.0"

71936941dependencies = [

7194 "agent",

71956942 "axum",

71966943 "ci",

71976944 "ci-protocol",

@@ -7320,12 +7067,6 @@

73207067dependencies = [

73217068 "proc-macro2",

73227069]

7323

7324[[package]]

7325name = "quoted_printable"

7326version = "0.5.2"

7327source = "registry+https://github.com/rust-lang/crates.io-index"

7328checksum = "478e0585659a122aa407eb7e3c0e1fa51b1d8a870038bd29f0cf4a8551eea972"

73297070

73307071[[package]]

73317072name = "r-efi"

@@ -7492,26 +7233,6 @@

74927233]

74937234

74947235[[package]]

7495name = "ref-cast"

7496version = "1.0.27"

7497source = "registry+https://github.com/rust-lang/crates.io-index"

7498checksum = "7e440fb4e4b4147295338efb76001ab9e4efc0e5839df2c47fc5ac2381d365c3"

7499dependencies = [

7500 "ref-cast-impl",

7501]

7502

7503[[package]]

7504name = "ref-cast-impl"

7505version = "1.0.27"

7506source = "registry+https://github.com/rust-lang/crates.io-index"

7507checksum = "92ecd8964f8453721699a1ed72037b0db49ce2f5a5138486ee89bed6f67cdf3a"

7508dependencies = [

7509 "proc-macro2",

7510 "quote",

7511 "syn 3.0.3",

7512]

7513

7514[[package]]

75157236name = "regex"

75167237version = "1.13.1"

75177238source = "registry+https://github.com/rust-lang/crates.io-index"

@@ -7688,51 +7409,6 @@

76887409]

76897410

76907411[[package]]

7691name = "rmcp"

7692version = "3.5.0"

7693source = "registry+https://github.com/rust-lang/crates.io-index"

7694checksum = "fae7019994ae0fe4ada40b732f798f3ff26f0f04facb1477f1bf37eb4f18a2d3"

7695dependencies = [

7696 "async-trait",

7697 "base64 0.23.1",

7698 "bytes",

7699 "chrono",

7700 "futures",

7701 "http 1.5.0",

7702 "http-body 1.1.0",

7703 "http-body-util",

7704 "indexmap",

7705 "pastey",

7706 "pin-project-lite",

7707 "rand 0.10.2",

7708 "rmcp-macros",

7709 "schemars",

7710 "serde",

7711 "serde_json",

7712 "sse-stream",

7713 "thiserror 2.0.20",

7714 "tokio",

7715 "tokio-stream",

7716 "tokio-util",

7717 "tower-service",

7718 "tracing",

7719 "uuid",

7720]

7721

7722[[package]]

7723name = "rmcp-macros"

7724version = "3.5.0"

7725source = "registry+https://github.com/rust-lang/crates.io-index"

7726checksum = "c7e66877136e969d5a2d561ee5841b17717a0619aebd64abb47a2a2f46e34683"

7727dependencies = [

7728 "darling 0.24.1",

7729 "proc-macro2",

7730 "quote",

7731 "serde_json",

7732 "syn 3.0.3",

7733]

7734

7735[[package]]

77367412name = "rsa"

77377413version = "0.9.10"

77387414source = "registry+https://github.com/rust-lang/crates.io-index"

@@ -8024,7 +7700,6 @@

80247700checksum = "0283386ce02abc0151e1761d08802dfe86c173b0b494af5cbc086574e453da06"

80257701dependencies = [

80267702 "aws-lc-rs",

8027 "log 0.4.33",

80287703 "once_cell",

80297704 "ring",

80307705 "rustls-pki-types",

@@ -8149,22 +7824,6 @@

81497824]

81507825

81517826[[package]]

8152name = "sandbox-rpc"

8153version = "1.0.0"

8154dependencies = [

8155 "config",

8156 "export",

8157 "opentelemetry",

8158 "prost 0.14.4",

8159 "thiserror 2.0.20",

8160 "tonic",

8161 "tonic-prost",

8162 "tonic-prost-build",

8163 "tracing",

8164 "tracing-opentelemetry",

8165]

8166

8167[[package]]

81687827name = "schannel"

81697828version = "0.1.29"

81707829source = "registry+https://github.com/rust-lang/crates.io-index"

@@ -8174,33 +7833,6 @@

81747833]

81757834

81767835[[package]]

8177name = "schemars"

8178version = "1.2.2"

8179source = "registry+https://github.com/rust-lang/crates.io-index"

8180checksum = "687274d293b6cdc6e73e0fee520bf2049650090d7164f87672d212a3c530cf4a"

8181dependencies = [

8182 "chrono",

8183 "dyn-clone",

8184 "ref-cast",

8185 "schemars_derive",

8186 "serde",

8187 "serde_json",

8188 "uuid",

8189]

8190

8191[[package]]

8192name = "schemars_derive"

8193version = "1.2.2"

8194source = "registry+https://github.com/rust-lang/crates.io-index"

8195checksum = "d98c67716b46af2f0b8cf752abc930f6f9aecfbf671ecfb531db8a31dbe4e2ba"

8196dependencies = [

8197 "proc-macro2",

8198 "quote",

8199 "serde_derive_internals",

8200 "syn 3.0.3",

8201]

8202

8203[[package]]

82047836name = "scopeguard"

82057837version = "1.2.0"

82067838source = "registry+https://github.com/rust-lang/crates.io-index"

@@ -8495,17 +8127,6 @@

84958127]

84968128

84978129[[package]]

8498name = "serde_derive_internals"

8499version = "0.30.0"

8500source = "registry+https://github.com/rust-lang/crates.io-index"

8501checksum = "f852137cce035d6a4df67ccce505ff6b3e9fd3a10e3e52b24dc71e650bb1a9bd"

8502dependencies = [

8503 "proc-macro2",

8504 "quote",

8505 "syn 3.0.3",

8506]

8507

8508[[package]]

85098130name = "serde_json"

85108131version = "1.0.151"

85118132source = "registry+https://github.com/rust-lang/crates.io-index"

@@ -8838,7 +8459,6 @@

88388459 "async-trait",

88398460 "axum",

88408461 "config",

8841 "email",

88428462 "export",

88438463 "i18n",

88448464 "imaging",

@@ -9081,19 +8701,6 @@

90818701]

90828702

90838703[[package]]

9084name = "sse-stream"

9085version = "0.2.6"

9086source = "registry+https://github.com/rust-lang/crates.io-index"

9087checksum = "c25ac7aff0abd1dbc474536e40416e1102c7dd9bfba0b9861c6d357f835dcfb4"

9088dependencies = [

9089 "bytes",

9090 "futures-util",

9091 "http-body 1.1.0",

9092 "http-body-util",

9093 "pin-project-lite",

9094]

9095

9096[[package]]

90978704name = "ssh-cipher"

90988705version = "0.3.0"

90998706source = "registry+https://github.com/rust-lang/crates.io-index"

@@ -10173,7 +9780,6 @@

101739780 "axum",

101749781 "base64 0.23.1",

101759782 "config",

10176 "email",

101779783 "export",

101789784 "futures",

101799785 "i18n",

@@ -10476,7 +10082,6 @@

1047610082name = "webapi"

1047710083version = "1.0.0"

1047810084dependencies = [

10479 "agents",

1048010085 "auth",

1048110086 "axum",

1048210087 "chat",

Modified Cargo.toml +0 −2 View Cargo.toml

@@ -20,7 +20,6 @@

2020ci-protocol = { path = "lib/ci-protocol" }

2121ci = { path = "lib/ci" }

2222config = { path = "lib/config" }

23email = { path = "lib/email" }

2423export = { path = "lib/export" }

2524models = { path = "lib/models" }

2625push = { path = "lib/push" }

@@ -93,7 +92,6 @@

9392hayro = { version = "0.7.1", features = ["embed-fonts", "embed-cmaps"] }

9493maud = { version = "0.27", features = ["axum"] }

9594pulldown-cmark = { version = "0.13", default-features = false, features = ["html"] }

96lettre = { version = "0.11", default-features = false, features = ["smtp-transport", "tokio1-rustls", "ring", "rustls-native-certs", "builder"] }

9795reqwest = { version = "0.12", default-features = false, features = ["rustls-tls", "json", "gzip", "brotli", "deflate"] }

9896livekit-api = { version = "0.7.1", default-features = false, features = ["services", "webhooks", "rustls-tls-webpki-roots"] }

9997livekit-protocol = "0.7.13"

Modified Dockerfile +2 −2 View Dockerfile

@@ -43,7 +43,7 @@

4343 /build/target/release/deps/web-* 2>/dev/null; \

4444 cargo build --release --workspace --bins && \

4545 mkdir -p /out && \

46 for b in apihost githost synchost searchhost mails migrate openapi im webhook runner; do \

46 for b in apihost githost synchost searchhost migrate openapi im webhook runner; do \

4747 cp "target/release/$b" /out/; \

4848 done

4949

@@ -79,7 +79,7 @@

7979WORKDIR /app

8080

8181COPY --chown=gitdata:gitdata config/default.toml config/default.toml

82COPY --from=builder /out/apihost /out/githost /out/synchost /out/searchhost /out/mails /out/migrate /out/openapi /out/im /out/webhook /out/runner /usr/local/bin/

82COPY --from=builder /out/apihost /out/githost /out/synchost /out/searchhost /out/migrate /out/openapi /out/im /out/webhook /out/runner /usr/local/bin/

8383

8484COPY --from=builder /out/docker /usr/local/bin/docker

8585

Modified justfile +1 −3 View justfile

@@ -60,7 +60,7 @@

6060up *svc='':

6161 @{{ invoke }}"{{ dev }}" up {{ svc }}

6262

63# 仅启动后端服务（apihost/githost/im/synchost/mails/webhook/aihost）

63# 仅启动后端服务（apihost/githost/im/synchost/webhook）

6464up-backend:

6565 @{{ invoke }}"{{ dev }}" up backends

6666

@@ -99,8 +99,6 @@

9999 @echo " im ws://127.0.0.1:4082"

100100 @echo " synchost http://127.0.0.1:4083/healthz"

101101 @echo " webhook http://127.0.0.1:3005"

102 @echo " aihost http://127.0.0.1:3006/healthz"

103 @echo " mcp http://127.0.0.1:3007/v1 (healthz: /healthz)"

104102 @echo " web http://127.0.0.1:5173"

105103

106104# 重新生成 openapi.json