diff --git a/docs/configuration.md b/docs/configuration.md index 9dbf8b0e4..5b5599574 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -180,8 +180,92 @@ fields as `[defaults]`, plus: |-----|------|---------|-------------| | `lists` | list | `[]` | Mailing list addresses that map to this subsystem. | | `patchwork.enabled` | bool | `false` | Enable Patchwork integration for this subsystem. | -| `patchwork.api_url` | string | -- | Patchwork API URL. | -| `patchwork.token` | string | -- | Patchwork API token. | +| `patchwork.api_url` | string | -- | Patchwork REST API URL (e.g. `https://patchwork.kernel.org/api/1.3`). Trailing slashes are stripped automatically. Invalid schemes are rejected with a warning. | +| `patchwork.token` | string | -- | Patchwork API token. Can also be set via `SASHIKO_PATCHWORK_TOKEN` env var (fills in where token is omitted in TOML). | +| `patchwork.email` | string | -- | Email address for email-based Patchwork notifications. | +| `patchwork.min_severity` | string | -- | Minimum finding severity to include in patchwork checks. Findings below this threshold are excluded. Accepts: `Low`, `Medium`, `High`, `Critical` (case-insensitive). Default: all findings included. | +| `patchwork.fail_severity` | string | `High` | Minimum severity of NEW findings that triggers the `fail` check state instead of `warning`. New findings at or above this threshold produce `fail`; below it produce `warning`. Pre-existing findings never affect the check state. | + +### Patchwork integration + +Sashiko can report review results as +[checks](https://patchwork.readthedocs.io/en/latest/usage/overview/#checks) +on a Patchwork instance. Two delivery modes are available and can be +enabled simultaneously for the same subsystem. + +**API mode** posts checks directly to the Patchwork REST API with +retry-queuing (3 attempts, exponential backoff). Requires a maintainer +API token. Note: Patchwork tokens grant full project-maintainer +permissions (state changes, delegation, etc.), not just check access. + +```toml +[subsystems.net.patchwork] +enabled = true +api_url = "https://patchwork.kernel.org/api/1.3" +token = "your-api-token" # or set SASHIKO_PATCHWORK_TOKEN env var +``` + +**Email mode** sends a structured notification email to a bot address. +A local script (such as +[pw_tools](https://github.com/mchehab/pw_tools)) parses the email and +posts the check. This avoids giving Sashiko a write token. + +```toml +[subsystems.linux-media.patchwork] +enabled = true +email = "pw-bot@lists.example.org" +``` + +#### Severity filtering and check state mapping + +By default, all findings are included in the patchwork check count. +Set `min_severity` to exclude findings below a threshold. When all +findings fall below the threshold, the check is posted as `success`. + +The check state depends only on **new** findings (not pre-existing): + +- `fail` -- new findings at or above `fail_severity` (default: `High`) +- `warning` -- new findings below `fail_severity` +- `success` -- no new findings (pre-existing findings are still + shown in the description but do not affect the state) + +The check description shows a per-severity breakdown with +pre-existing counts in parentheses, dropping zero-count severities. +For example: `Critical: 1 · High: 2 (1 pre-existing)`. + +```toml +[subsystems.net.patchwork] +enabled = true +api_url = "https://patchwork.kernel.org/api/1.3" +min_severity = "Medium" # exclude Low findings entirely +fail_severity = "High" # High+ new findings = fail (default) +``` + +Edge case behaviors: + +- Missing or null `preexisting` flag on a finding is treated as new +- When `min_severity` filters out all findings, the check is `success` + with "Sashiko AI review found no regressions" +- When only pre-existing findings remain after filtering, the check + is `success` but the description shows the pre-existing breakdown + +#### Email notification format + +When email mode is enabled, Sashiko sends a plain-text email with: + +- **To**: the configured `patchwork.email` address +- **Subject**: `[sashiko-check] {status} - {patch_subject}` +- **Body** (one key-value pair per line): + +``` +msgid: +status: success|warning +description: Sashiko AI review found N potential issue(s) +target_url: https://sashiko.dev/#/patchset/... +context: sashiko +``` + +Downstream tools can parse this format with simple line splitting. ## Environment variables @@ -196,5 +280,6 @@ fields as `[defaults]`, plus: | `CLOUD_ML_REGION` | GCP region for Vertex AI provider. | | `SASHIKO_SERVER` | Override daemon URL for CLI commands. | | `SASHIKO__*` | Override any Settings.toml value (e.g. `SASHIKO__AI__PROVIDER`). | +| `SASHIKO_PATCHWORK_TOKEN` | Patchwork API token. Fills in `patchwork.token` for enabled subsystems that have `api_url` set but no explicit token in TOML. | | `NO_COLOR` | Disable ANSI color output. | | `SASHIKO_LOG_PLAIN` | Use plain log format (no level/target/timestamp). | diff --git a/docs/examples/email_policy.toml b/docs/examples/email_policy.toml index 07a7b16d3..04a7ee8d0 100644 --- a/docs/examples/email_policy.toml +++ b/docs/examples/email_policy.toml @@ -28,3 +28,40 @@ ignored_emails = [] # If true, Sashiko will send a reply indicating that the review was positive. send_positive_review = false +# --- Patchwork integration examples --- +# +# API mode: direct REST API calls with retry-queuing (3 attempts). +# Requires a maintainer API token. The token can also be injected via +# the SASHIKO_PATCHWORK_TOKEN environment variable. +# +# [subsystems.net.patchwork] +# enabled = true +# api_url = "https://patchwork.kernel.org/api/1.3" +# token = "your-api-token" + +# Email mode: sends a structured notification email that a local script +# (e.g. pw_tools) can parse and use to post the check. +# +# [subsystems.linux-media.patchwork] +# enabled = true +# email = "pw-bot@lists.example.org" + +# Both modes simultaneously: +# +# [subsystems.dri-devel.patchwork] +# enabled = true +# api_url = "https://patchwork.freedesktop.org/api/1.3" +# token = "your-api-token" +# email = "backup-bot@lists.example.org" + +# Severity filtering and check state mapping: +# min_severity excludes findings below the threshold entirely. +# fail_severity controls which new findings trigger "fail" vs "warning". +# Pre-existing findings are shown in the description but never affect state. +# Default fail_severity is "High" (High+ new = fail, Medium/Low new = warning). +# +# [subsystems.net.patchwork] +# enabled = true +# api_url = "https://patchwork.kernel.org/api/1.3" +# min_severity = "Medium" +# fail_severity = "High" diff --git a/src/db.rs b/src/db.rs index 12c0bd58e..584be1dc2 100644 --- a/src/db.rs +++ b/src/db.rs @@ -147,7 +147,7 @@ pub struct Finding { pub struct EmailOutboxRow { pub id: i64, - pub patch_id: i64, + pub patch_id: Option, pub status: String, pub to_addresses: String, pub cc_addresses: String, @@ -160,6 +160,22 @@ pub struct EmailOutboxRow { pub created_at: i64, } +pub struct PatchworkOutboxRow { + pub id: i64, + pub patch_msg_id: String, + pub api_url: String, + pub check_state: String, + pub description: String, + pub target_url: String, + pub context: String, + pub status: String, + pub retry_count: i64, + pub next_retry_at: Option, + pub locked_at: Option, + pub error_log: Option, + pub created_at: i64, +} + impl Database { pub async fn get_oldest_message_timestamp(&self) -> Result> { let mut rows = self @@ -3766,7 +3782,7 @@ impl Database { if let Ok(Some(row)) = rows.next().await { let id: i64 = row.get(0)?; - let patch_id: i64 = row.get(1)?; + let patch_id: Option = row.get::(1).ok(); let status: String = row.get(2)?; let to_addresses: String = row.get(3)?; let cc_addresses: String = row.get(4)?; @@ -3820,6 +3836,168 @@ impl Database { ).await?; Ok(count) } + + // -- Patchwork outbox operations -- + + pub async fn insert_patchwork_outbox( + &self, + patch_msg_id: &str, + api_url: &str, + check_state: &str, + description: &str, + target_url: &str, + context: &str, + ) -> Result<()> { + let created_at = chrono::Utc::now().timestamp(); + self.conn + .execute( + "INSERT INTO patchwork_outbox (patch_msg_id, api_url, check_state, description, target_url, context, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?)", + libsql::params![ + patch_msg_id, + api_url, + check_state, + description, + target_url, + context, + created_at, + ], + ) + .await?; + Ok(()) + } + + pub async fn lock_pending_patchwork(&self) -> Result> { + let now = chrono::Utc::now().timestamp(); + let mut rows = self + .conn + .query( + "UPDATE patchwork_outbox + SET status = 'Sending', locked_at = ? + WHERE id = ( + SELECT id FROM patchwork_outbox + WHERE status = 'Pending' + AND (next_retry_at IS NULL OR next_retry_at <= ?) + LIMIT 1 + ) + RETURNING id, patch_msg_id, api_url, check_state, description, target_url, context, status, retry_count, next_retry_at, locked_at, error_log, created_at", + libsql::params![now, now], + ) + .await?; + + if let Ok(Some(row)) = rows.next().await { + let id: i64 = row.get(0)?; + let patch_msg_id: String = row.get(1)?; + let api_url: String = row.get(2)?; + let check_state: String = row.get(3)?; + let description: String = row.get(4)?; + let target_url: String = row.get(5)?; + let context: String = row.get(6)?; + let status: String = row.get(7)?; + let retry_count: i64 = row.get(8)?; + let next_retry_at: Option = row.get::(9).ok(); + let locked_at: Option = row.get::(10).ok(); + let error_log: Option = row.get::(11).ok(); + let created_at: i64 = row.get(12)?; + + Ok(Some(PatchworkOutboxRow { + id, + patch_msg_id, + api_url, + check_state, + description, + target_url, + context, + status, + retry_count, + next_retry_at, + locked_at, + error_log, + created_at, + })) + } else { + Ok(None) + } + } + + pub async fn mark_patchwork_sent(&self, id: i64) -> Result<()> { + self.conn + .execute( + "UPDATE patchwork_outbox SET status = 'Sent', locked_at = NULL WHERE id = ?", + libsql::params![id], + ) + .await?; + Ok(()) + } + + pub async fn mark_patchwork_failed(&self, id: i64, error_log: &str) -> Result<()> { + self.conn + .execute( + "UPDATE patchwork_outbox SET status = 'Failed', error_log = ?, locked_at = NULL WHERE id = ?", + libsql::params![error_log.to_string(), id], + ) + .await?; + Ok(()) + } + + /// Mark a patchwork outbox entry for retry at a future timestamp. + /// Increments retry_count, sets next_retry_at, and returns to + /// Pending status so the worker loop continues without blocking. + pub async fn set_patchwork_retry_at(&self, id: i64, next_retry_at: i64) -> Result<()> { + self.conn + .execute( + "UPDATE patchwork_outbox SET status = 'Pending', retry_count = retry_count + 1, next_retry_at = ?, locked_at = NULL WHERE id = ?", + libsql::params![next_retry_at, id], + ) + .await?; + Ok(()) + } + + pub async fn sweep_ghost_patchwork(&self) -> Result { + let ten_mins_ago = chrono::Utc::now().timestamp() - 600; + let count = self.conn + .execute( + "UPDATE patchwork_outbox SET status = 'Pending', locked_at = NULL WHERE status = 'Sending' AND locked_at < ?", + libsql::params![ten_mins_ago], + ) + .await?; + Ok(count) + } + + /// Insert a patchwork notification email into the email outbox. + /// + /// Uses patch_id = NULL to avoid colliding with the per-patch dedup + /// guard in insert_email_outbox(). The EmailWorker processes these + /// rows normally since it picks up any row with status = 'Pending'. + pub async fn insert_patchwork_notification( + &self, + status: &str, + to_address: &str, + subject: &str, + in_reply_to: &str, + references_hdr: &str, + body: &str, + ) -> Result<()> { + let created_at = chrono::Utc::now().timestamp(); + let to_json = serde_json::to_string(&[to_address]) + .map_err(|e| libsql::Error::Misuse(e.to_string()))?; + self.conn + .execute( + "INSERT INTO email_outbox (patch_id, status, to_addresses, cc_addresses, subject, in_reply_to, references_hdr, body, created_at) + VALUES (NULL, ?, ?, '[]', ?, ?, ?, ?, ?)", + libsql::params![ + status, + to_json, + subject, + in_reply_to, + references_hdr, + body, + created_at, + ], + ) + .await?; + Ok(()) + } } #[cfg(test)] diff --git a/src/email_policy.rs b/src/email_policy.rs index a9aa06b79..1f17d688e 100644 --- a/src/email_policy.rs +++ b/src/email_policy.rs @@ -3,12 +3,85 @@ use std::collections::HashMap; use std::fs; use std::path::Path; -#[derive(Deserialize, Debug, Clone, Default)] +#[derive(Deserialize, Debug, Clone)] pub struct PatchworkPolicy { #[serde(default)] pub enabled: bool, pub api_url: Option, pub token: Option, + pub email: Option, + /// Minimum finding severity to include in patchwork checks. + /// Findings below this threshold are excluded from the check + /// count and description. Accepts: "Low", "Medium", "High", + /// "Critical" (case-insensitive). Default: None (all findings). + pub min_severity: Option, + /// Minimum severity of NEW findings that triggers the "fail" + /// check state instead of "warning". Accepts: "Low", "Medium", + /// "High", "Critical" (case-insensitive). Default: "High". + /// New findings at or above this threshold produce "fail"; + /// below it produce "warning". Pre-existing findings never + /// affect the check state. + #[serde(default = "default_fail_severity")] + pub fail_severity: String, +} + +fn default_fail_severity() -> String { + "High".to_string() +} + +impl Default for PatchworkPolicy { + fn default() -> Self { + Self { + enabled: false, + api_url: None, + token: None, + email: None, + min_severity: None, + fail_severity: default_fail_severity(), + } + } +} + +impl PatchworkPolicy { + /// Normalize api_url: strip trailing slashes, validate scheme. + /// Invalid schemes produce a warning and clear api_url. + /// Non-localhost http:// URLs produce a security warning since + /// the API token would be sent in plaintext. + pub fn normalize(&mut self) { + if let Some(url) = &self.api_url { + let trimmed = url.trim_end_matches('/'); + if !trimmed.starts_with("https://") && !trimmed.starts_with("http://") { + tracing::warn!("Patchwork api_url has invalid scheme: {}", url); + self.api_url = None; + } else { + if trimmed.starts_with("http://") && !Self::is_localhost_url(trimmed) { + tracing::warn!( + "Patchwork api_url uses http:// for a non-localhost host. \ + The API token will be sent in plaintext: {}", + trimmed + ); + } + self.api_url = Some(trimmed.to_string()); + } + } + } + + /// Check whether a URL points to a localhost address. + fn is_localhost_url(url: &str) -> bool { + let after_scheme = url + .strip_prefix("http://") + .or_else(|| url.strip_prefix("https://")) + .unwrap_or(url); + // Extract host+port before first path separator + let host_port = after_scheme.split('/').next().unwrap_or(""); + // Handle IPv6 bracket notation: [::1]:8000 + if host_port.starts_with('[') { + host_port.starts_with("[::1]") + } else { + let host = host_port.split(':').next().unwrap_or(""); + matches!(host, "localhost" | "127.0.0.1") + } + } } #[derive(Deserialize, Debug, Clone, Default)] @@ -58,9 +131,44 @@ impl EmailPolicyConfig { } let content = fs::read_to_string(path)?; - let config: Self = toml::from_str(&content)?; + let mut config: Self = toml::from_str(&content)?; + + let env_token = std::env::var("SASHIKO_PATCHWORK_TOKEN").ok(); + config.apply_token_override(env_token.as_deref()); + config.normalize_patchwork_urls(); + Ok(config) } + + /// Apply a fallback patchwork token to any enabled patchwork policy + /// that has an api_url but no explicit token. Explicit TOML tokens + /// are never overwritten. + pub fn apply_token_override(&mut self, token: Option<&str>) { + let Some(token) = token else { return }; + + if self.defaults.patchwork.enabled + && self.defaults.patchwork.api_url.is_some() + && self.defaults.patchwork.token.is_none() + { + self.defaults.patchwork.token = Some(token.to_string()); + } + for sub in self.subsystems.values_mut() { + if sub.patchwork.enabled + && sub.patchwork.api_url.is_some() + && sub.patchwork.token.is_none() + { + sub.patchwork.token = Some(token.to_string()); + } + } + } + + /// Normalize patchwork URLs across all policies. + fn normalize_patchwork_urls(&mut self) { + self.defaults.patchwork.normalize(); + for sub in self.subsystems.values_mut() { + sub.patchwork.normalize(); + } + } } #[cfg(test)] @@ -143,4 +251,347 @@ mod tests { assert!(!config.defaults.reply_to_author); assert!(config.subsystems.is_empty()); } + + #[test] + fn test_patchwork_email_field() { + let toml_content = r#" + [defaults] + + [subsystems.media] + lists = ["linux-media@vger.kernel.org"] + + [subsystems.media.patchwork] + enabled = true + email = "pw-bot@lists.example.org" + "#; + + let mut file = NamedTempFile::new().unwrap(); + write!(file, "{}", toml_content).unwrap(); + + let config = EmailPolicyConfig::load(file.path()).expect("Failed to load policy"); + let media = config.subsystems.get("media").expect("media missing"); + assert!(media.patchwork.enabled); + assert_eq!( + media.patchwork.email.as_deref(), + Some("pw-bot@lists.example.org") + ); + assert!(media.patchwork.api_url.is_none()); + assert!(media.patchwork.token.is_none()); + } + + #[test] + fn test_patchwork_email_field_absent() { + let toml_content = r#" + [defaults] + + [subsystems.net.patchwork] + enabled = true + api_url = "https://patchwork.kernel.org/api/1.3" + "#; + + let mut file = NamedTempFile::new().unwrap(); + write!(file, "{}", toml_content).unwrap(); + + let config = EmailPolicyConfig::load(file.path()).expect("Failed to load policy"); + let net = config.subsystems.get("net").expect("net missing"); + assert!(net.patchwork.enabled); + assert!(net.patchwork.email.is_none()); + assert!(net.patchwork.api_url.is_some()); + } + + #[test] + fn test_url_normalization_trailing_slash() { + let toml_content = r#" + [defaults] + + [subsystems.net.patchwork] + enabled = true + api_url = "https://patchwork.kernel.org/api/1.3/" + "#; + + let mut file = NamedTempFile::new().unwrap(); + write!(file, "{}", toml_content).unwrap(); + + let config = EmailPolicyConfig::load(file.path()).expect("Failed to load policy"); + let net = config.subsystems.get("net").expect("net missing"); + assert_eq!( + net.patchwork.api_url.as_deref(), + Some("https://patchwork.kernel.org/api/1.3") + ); + } + + #[test] + fn test_url_normalization_multiple_trailing_slashes() { + let toml_content = r#" + [defaults] + + [subsystems.net.patchwork] + enabled = true + api_url = "https://patchwork.kernel.org/api/1.3///" + "#; + + let mut file = NamedTempFile::new().unwrap(); + write!(file, "{}", toml_content).unwrap(); + + let config = EmailPolicyConfig::load(file.path()).expect("Failed to load policy"); + let net = config.subsystems.get("net").expect("net missing"); + assert_eq!( + net.patchwork.api_url.as_deref(), + Some("https://patchwork.kernel.org/api/1.3") + ); + } + + #[test] + fn test_url_normalization_invalid_scheme() { + let toml_content = r#" + [defaults] + + [subsystems.net.patchwork] + enabled = true + api_url = "ftp://patchwork.kernel.org/api/1.3" + "#; + + let mut file = NamedTempFile::new().unwrap(); + write!(file, "{}", toml_content).unwrap(); + + let config = EmailPolicyConfig::load(file.path()).expect("Failed to load policy"); + let net = config.subsystems.get("net").expect("net missing"); + // Invalid scheme should be cleared + assert!(net.patchwork.api_url.is_none()); + } + + #[test] + fn test_url_normalization_valid_http() { + let toml_content = r#" + [defaults] + + [subsystems.net.patchwork] + enabled = true + api_url = "http://localhost:8000/api/1.3" + "#; + + let mut file = NamedTempFile::new().unwrap(); + write!(file, "{}", toml_content).unwrap(); + + let config = EmailPolicyConfig::load(file.path()).expect("Failed to load policy"); + let net = config.subsystems.get("net").expect("net missing"); + assert_eq!( + net.patchwork.api_url.as_deref(), + Some("http://localhost:8000/api/1.3") + ); + } + + fn make_config_with_patchwork( + enabled: bool, + api_url: Option<&str>, + token: Option<&str>, + ) -> EmailPolicyConfig { + let mut subsystems = HashMap::new(); + subsystems.insert( + "net".to_string(), + SubsystemPolicy { + patchwork: PatchworkPolicy { + enabled, + api_url: api_url.map(String::from), + token: token.map(String::from), + ..Default::default() + }, + ..Default::default() + }, + ); + EmailPolicyConfig { + defaults: SubsystemPolicy::default(), + subsystems, + } + } + + #[test] + fn test_token_override_fills_gap() { + let mut config = + make_config_with_patchwork(true, Some("https://patchwork.kernel.org/api/1.3"), None); + config.apply_token_override(Some("injected-token")); + + let net = config.subsystems.get("net").unwrap(); + assert_eq!(net.patchwork.token.as_deref(), Some("injected-token")); + } + + #[test] + fn test_token_override_no_overwrite_explicit() { + let mut config = make_config_with_patchwork( + true, + Some("https://patchwork.kernel.org/api/1.3"), + Some("toml-explicit-token"), + ); + config.apply_token_override(Some("injected-token")); + + let net = config.subsystems.get("net").unwrap(); + assert_eq!( + net.patchwork.token.as_deref(), + Some("toml-explicit-token"), + "explicit TOML token should not be overwritten" + ); + } + + #[test] + fn test_token_override_skips_disabled() { + let mut config = + make_config_with_patchwork(false, Some("https://patchwork.kernel.org/api/1.3"), None); + config.apply_token_override(Some("injected-token")); + + let net = config.subsystems.get("net").unwrap(); + assert!( + net.patchwork.token.is_none(), + "disabled patchwork should not get override token" + ); + } + + #[test] + fn test_token_override_skips_no_api_url() { + let mut config = make_config_with_patchwork(true, None, None); + config.apply_token_override(Some("injected-token")); + + let net = config.subsystems.get("net").unwrap(); + assert!( + net.patchwork.token.is_none(), + "patchwork without api_url should not get override token" + ); + } + + #[test] + fn test_token_override_none_is_noop() { + let mut config = + make_config_with_patchwork(true, Some("https://patchwork.kernel.org/api/1.3"), None); + config.apply_token_override(None); + + let net = config.subsystems.get("net").unwrap(); + assert!(net.patchwork.token.is_none()); + } + + #[test] + fn test_patchwork_normalize_direct() { + let mut policy = PatchworkPolicy { + enabled: true, + api_url: Some("https://example.org/api/1.3/".to_string()), + ..Default::default() + }; + policy.normalize(); + assert_eq!( + policy.api_url.as_deref(), + Some("https://example.org/api/1.3") + ); + + let mut bad = PatchworkPolicy { + enabled: true, + api_url: Some("ftp://example.org".to_string()), + ..Default::default() + }; + bad.normalize(); + assert!(bad.api_url.is_none()); + } + + #[test] + fn test_is_localhost_url() { + assert!(PatchworkPolicy::is_localhost_url( + "http://localhost:8000/api" + )); + assert!(PatchworkPolicy::is_localhost_url( + "http://127.0.0.1:8000/api" + )); + assert!(PatchworkPolicy::is_localhost_url("http://[::1]:8000/api")); + assert!(PatchworkPolicy::is_localhost_url("http://localhost/api")); + assert!(!PatchworkPolicy::is_localhost_url( + "http://patchwork.kernel.org/api" + )); + assert!(!PatchworkPolicy::is_localhost_url("http://10.0.0.1/api")); + } + + #[test] + fn test_normalize_http_non_localhost_still_accepted() { + // http:// for non-localhost is accepted (with a warning) not rejected + let mut policy = PatchworkPolicy { + enabled: true, + api_url: Some("http://patchwork.example.org/api/1.3".to_string()), + ..Default::default() + }; + policy.normalize(); + assert_eq!( + policy.api_url.as_deref(), + Some("http://patchwork.example.org/api/1.3"), + "http:// non-localhost should be accepted with warning, not rejected" + ); + } + + #[test] + fn test_min_severity_deserialization() { + let toml_content = r#" + [defaults] + + [subsystems.net.patchwork] + enabled = true + api_url = "https://patchwork.kernel.org/api/1.3" + min_severity = "Medium" + "#; + + let mut file = NamedTempFile::new().unwrap(); + write!(file, "{}", toml_content).unwrap(); + + let config = EmailPolicyConfig::load(file.path()).expect("Failed to load policy"); + let net = config.subsystems.get("net").expect("net missing"); + assert_eq!(net.patchwork.min_severity.as_deref(), Some("Medium")); + } + + #[test] + fn test_min_severity_absent_is_none() { + let toml_content = r#" + [defaults] + + [subsystems.net.patchwork] + enabled = true + api_url = "https://patchwork.kernel.org/api/1.3" + "#; + + let mut file = NamedTempFile::new().unwrap(); + write!(file, "{}", toml_content).unwrap(); + + let config = EmailPolicyConfig::load(file.path()).expect("Failed to load policy"); + let net = config.subsystems.get("net").expect("net missing"); + assert!(net.patchwork.min_severity.is_none()); + } + + #[test] + fn test_fail_severity_default_is_high() { + let toml_content = r#" + [defaults] + + [subsystems.net.patchwork] + enabled = true + api_url = "https://patchwork.kernel.org/api/1.3" + "#; + + let mut file = NamedTempFile::new().unwrap(); + write!(file, "{}", toml_content).unwrap(); + + let config = EmailPolicyConfig::load(file.path()).expect("Failed to load policy"); + let net = config.subsystems.get("net").expect("net missing"); + assert_eq!(net.patchwork.fail_severity, "High"); + } + + #[test] + fn test_fail_severity_custom() { + let toml_content = r#" + [defaults] + + [subsystems.net.patchwork] + enabled = true + api_url = "https://patchwork.kernel.org/api/1.3" + fail_severity = "Critical" + "#; + + let mut file = NamedTempFile::new().unwrap(); + write!(file, "{}", toml_content).unwrap(); + + let config = EmailPolicyConfig::load(file.path()).expect("Failed to load policy"); + let net = config.subsystems.get("net").expect("net missing"); + assert_eq!(net.patchwork.fail_severity, "Critical"); + } } diff --git a/src/main.rs b/src/main.rs index 24fb0e6cf..367b7f6c0 100644 --- a/src/main.rs +++ b/src/main.rs @@ -532,6 +532,20 @@ async fn main() -> Result<(), Box> { }); } + // Start Patchwork Worker (processes API check entries when they exist) + { + let pw_policy_path = settings.review.email_policy_path.clone(); + let pw_max_retries = settings.review.max_retries; + let patchwork_worker = sashiko::worker::patchwork::PatchworkWorker::new( + db.clone(), + pw_policy_path, + pw_max_retries, + ); + tokio::spawn(async move { + patchwork_worker.run().await; + }); + } + // Initialize custom remotes let repo_path = std::path::PathBuf::from(&settings.git.repository_path); diff --git a/src/patchwork.rs b/src/patchwork.rs index 40cf36401..d2f0dd627 100644 --- a/src/patchwork.rs +++ b/src/patchwork.rs @@ -1,7 +1,149 @@ +use crate::db::Severity; use crate::email_policy::PatchworkPolicy; use reqwest::{Client, header}; use serde::{Deserialize, Serialize}; -use tracing::{debug, error, info, warn}; +use serde_json::Value; +use tracing::{debug, info, warn}; + +/// Result of evaluating a patchwork check from a policy and findings. +/// Separates config (PatchworkPolicy) from computed output. +#[derive(Debug, PartialEq)] +pub struct PatchworkCheckResult { + /// Patchwork check state: "success", "warning", or "fail" + pub state: String, + /// Human-readable description with per-severity breakdown + pub description: String, +} + +/// Per-severity counts for new and pre-existing findings. +#[derive(Debug, Default)] +struct SeverityCounts { + new: usize, + preexisting: usize, +} + +impl PatchworkCheckResult { + /// Build a check result from a patchwork policy and raw findings. + /// + /// Applies min_severity filtering, splits by preexisting flag, + /// computes the check state from fail_severity threshold (new + /// findings only), and formats the description with per-severity + /// breakdown including pre-existing counts. + pub fn from_policy(policy: &PatchworkPolicy, findings: &[Value]) -> Self { + let min_threshold = policy + .min_severity + .as_ref() + .map(|s| Severity::from_str(s) as i32) + .unwrap_or(Severity::Low as i32); + + let fail_threshold = Severity::from_str(&policy.fail_severity) as i32; + + // Count findings per severity, split by new vs pre-existing + let mut critical = SeverityCounts::default(); + let mut high = SeverityCounts::default(); + let mut medium = SeverityCounts::default(); + let mut low = SeverityCounts::default(); + + for f in findings { + let sev = + Severity::from_str(f.get("severity").and_then(|v| v.as_str()).unwrap_or("Low")); + + // Apply min_severity filter + if (sev as i32) < min_threshold { + continue; + } + + let is_preexisting = f + .get("preexisting") + .and_then(|v| v.as_bool()) + .unwrap_or(false); + + let bucket = match sev { + Severity::Critical => &mut critical, + Severity::High => &mut high, + Severity::Medium => &mut medium, + Severity::Low => &mut low, + }; + + if is_preexisting { + bucket.preexisting += 1; + } else { + bucket.new += 1; + } + } + + let total_new = critical.new + high.new + medium.new + low.new; + let total_preexisting = + critical.preexisting + high.preexisting + medium.preexisting + low.preexisting; + + // Determine state from NEW findings only + let state = if total_new == 0 { + "success" + } else { + // Check if any new finding meets the fail threshold + let has_fail_level = [ + (Severity::Critical as i32, critical.new), + (Severity::High as i32, high.new), + (Severity::Medium as i32, medium.new), + (Severity::Low as i32, low.new), + ] + .iter() + .any(|(sev, count)| *count > 0 && *sev >= fail_threshold); + + if has_fail_level { "fail" } else { "warning" } + }; + + // Build description + let description = if total_new == 0 && total_preexisting == 0 { + "Sashiko AI review found no regressions".to_string() + } else { + Self::format_description(&critical, &high, &medium, &low) + }; + + Self { + state: state.to_string(), + description, + } + } + + /// Format per-severity description, dropping zero-count severities. + /// New counts shown bare, pre-existing in parentheses. + /// Example: "Critical: 1 · High: 2 (1 pre-existing)" + fn format_description( + critical: &SeverityCounts, + high: &SeverityCounts, + medium: &SeverityCounts, + low: &SeverityCounts, + ) -> String { + let mut parts = Vec::new(); + + for (label, counts) in [ + ("Critical", critical), + ("High", high), + ("Medium", medium), + ("Low", low), + ] { + if counts.new == 0 && counts.preexisting == 0 { + continue; + } + + let part = if counts.new > 0 && counts.preexisting > 0 { + format!( + "{}: {} ({} pre-existing)", + label, counts.new, counts.preexisting + ) + } else if counts.new > 0 { + format!("{}: {}", label, counts.new) + } else { + format!("{}: {} pre-existing", label, counts.preexisting) + }; + + parts.push(part); + } + + parts.join(" \u{00b7} ") // middle dot separator + } +} #[derive(Debug, Deserialize)] struct PatchworkListResponse { @@ -16,71 +158,66 @@ struct PatchworkCheckRequest { context: String, } +/// Post a check result to the Patchwork REST API for a given patch. +/// +/// Looks up the patch by message-ID, then POSTs the check. Returns Ok +/// on success or Err with a description on failure so the caller can +/// decide whether to retry. pub async fn post_patchwork_check( - policy: &PatchworkPolicy, + client: &Client, + api_url: &str, + token: Option<&str>, msgid: &str, status: &str, description: &str, target_url: &str, -) { - if !policy.enabled { - return; - } +) -> Result<(), String> { + let api_url = api_url.trim_end_matches('/'); - let api_url = match &policy.api_url { - Some(url) => url.trim_end_matches('/'), - None => { - warn!("Patchwork enabled but no api_url provided"); - return; - } - }; - - let client = Client::new(); - - // 1. Get the patch ID using the Message-ID - // Patchwork expects the msgid to be without the angle brackets, or URL encoded. - // The msgid from the database or headers might contain `<` and `>`. + // Strip angle brackets from the message-ID. let clean_msgid = msgid.trim_matches(|c| c == '<' || c == '>'); - let list_url = format!("{}/patches/?msgid={}", api_url, clean_msgid); - debug!("Fetching Patchwork ID from: {}", list_url); + // Build the lookup URL with proper URL-encoding for the msgid. + let base_url = format!("{}/patches/", api_url); + let patches_url = reqwest::Url::parse_with_params(&base_url, &[("msgid", clean_msgid)]) + .map_err(|e| format!("failed to build patchwork URL: {}", e))?; + debug!("Fetching Patchwork patch by msgid: {}", clean_msgid); - let mut get_req = client.get(&list_url); - if let Some(token) = &policy.token { + let mut get_req = client.get(patches_url); + if let Some(token) = token { get_req = get_req.header(header::AUTHORIZATION, format!("Token {}", token)); } - let resp = match get_req.send().await { - Ok(r) => r, - Err(e) => { - error!("Failed to fetch patchwork patch list: {}", e); - return; - } - }; + let resp = get_req + .send() + .await + .map_err(|e| format!("failed to fetch patchwork patch list: {}", e))?; if !resp.status().is_success() { - error!( - "Patchwork API returned {}: {:?}", - resp.status(), - resp.text().await.unwrap_or_default() - ); - return; + let status_code = resp.status(); + let body = resp.text().await.unwrap_or_default(); + return Err(format!("patchwork API returned {}: {}", status_code, body)); } - let patches: Vec = match resp.json().await { - Ok(p) => p, - Err(e) => { - error!("Failed to parse patchwork list response: {}", e); - return; - } - }; + let patches: Vec = resp + .json() + .await + .map_err(|e| format!("failed to parse patchwork list response: {}", e))?; if patches.is_empty() { debug!("Patchwork returned no patches for msgid {}", msgid); - return; + return Ok(()); + } + + if patches.len() > 1 { + warn!( + "Patchwork returned {} patches for msgid {}, using first (id={})", + patches.len(), + msgid, + patches[0].id + ); } - // We take the first match if there are multiple. let patch_id = patches[0].id; let check_url = format!("{}/patches/{}/checks/", api_url, patch_id); @@ -94,25 +231,293 @@ pub async fn post_patchwork_check( debug!("Posting check to Patchwork: {} {:?}", check_url, payload); let mut post_req = client.post(&check_url).json(&payload); - if let Some(token) = &policy.token { + if let Some(token) = token { post_req = post_req.header(header::AUTHORIZATION, format!("Token {}", token)); } - let post_resp = match post_req.send().await { - Ok(r) => r, - Err(e) => { - error!("Failed to post patchwork check: {}", e); - return; - } - }; + let post_resp = post_req + .send() + .await + .map_err(|e| format!("failed to post patchwork check: {}", e))?; if post_resp.status().is_success() { info!("Successfully posted check to Patchwork for msgid {}", msgid); + Ok(()) } else { - error!( - "Patchwork check post failed with status {}: {:?}", - post_resp.status(), - post_resp.text().await.unwrap_or_default() + let status_code = post_resp.status(); + let body = post_resp.text().await.unwrap_or_default(); + Err(format!( + "patchwork check post failed with status {}: {}", + status_code, body + )) + } +} + +/// Compose a structured patchwork notification email for email-based mode. +/// +/// Returns (subject, body). The body uses a simple key-value format +/// parseable by downstream tools such as pw_tools. +pub fn compose_patchwork_email( + msgid: &str, + status: &str, + description: &str, + target_url: &str, + patch_subject: &str, +) -> (String, String) { + let subject = format!("[sashiko-check] {} - {}", status, patch_subject); + let body = format!( + "msgid: {}\nstatus: {}\ndescription: {}\ntarget_url: {}\ncontext: sashiko\n", + msgid, status, description, target_url + ); + (subject, body) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::email_policy::PatchworkPolicy; + + fn finding(severity: &str, preexisting: bool) -> Value { + serde_json::json!({"severity": severity, "problem": "test", "preexisting": preexisting}) + } + + fn finding_new(severity: &str) -> Value { + finding(severity, false) + } + + fn finding_preexisting(severity: &str) -> Value { + finding(severity, true) + } + + fn default_policy() -> PatchworkPolicy { + PatchworkPolicy::default() + } + + // -- PatchworkCheckResult tests -- + + #[test] + fn test_no_findings_success() { + let result = PatchworkCheckResult::from_policy(&default_policy(), &[]); + assert_eq!(result.state, "success"); + assert_eq!(result.description, "Sashiko AI review found no regressions"); + } + + #[test] + fn test_new_high_produces_fail() { + let findings = vec![finding_new("High")]; + let result = PatchworkCheckResult::from_policy(&default_policy(), &findings); + assert_eq!(result.state, "fail"); + assert!(result.description.contains("High: 1")); + } + + #[test] + fn test_new_critical_produces_fail() { + let findings = vec![finding_new("Critical")]; + let result = PatchworkCheckResult::from_policy(&default_policy(), &findings); + assert_eq!(result.state, "fail"); + assert!(result.description.contains("Critical: 1")); + } + + #[test] + fn test_new_medium_produces_warning() { + let findings = vec![finding_new("Medium")]; + let result = PatchworkCheckResult::from_policy(&default_policy(), &findings); + assert_eq!(result.state, "warning"); + assert!(result.description.contains("Medium: 1")); + } + + #[test] + fn test_new_low_produces_warning() { + let findings = vec![finding_new("Low")]; + let result = PatchworkCheckResult::from_policy(&default_policy(), &findings); + assert_eq!(result.state, "warning"); + assert!(result.description.contains("Low: 1")); + } + + #[test] + fn test_only_preexisting_produces_success() { + let findings = vec![finding_preexisting("Critical"), finding_preexisting("High")]; + let result = PatchworkCheckResult::from_policy(&default_policy(), &findings); + assert_eq!(result.state, "success"); + assert!(result.description.contains("pre-existing")); + } + + #[test] + fn test_mixed_new_and_preexisting() { + let findings = vec![ + finding_new("Critical"), + finding_preexisting("High"), + finding_preexisting("High"), + ]; + let result = PatchworkCheckResult::from_policy(&default_policy(), &findings); + assert_eq!(result.state, "fail"); + assert!(result.description.contains("Critical: 1")); + assert!(result.description.contains("High: 2 pre-existing")); + } + + #[test] + fn test_mixed_new_and_preexisting_same_severity() { + let findings = vec![ + finding_new("High"), + finding_new("High"), + finding_new("High"), + finding_preexisting("High"), + ]; + let result = PatchworkCheckResult::from_policy(&default_policy(), &findings); + assert_eq!(result.state, "fail"); + assert!(result.description.contains("High: 3 (1 pre-existing)")); + } + + #[test] + fn test_zero_counts_dropped_from_description() { + let findings = vec![finding_new("Critical")]; + let result = PatchworkCheckResult::from_policy(&default_policy(), &findings); + assert!(!result.description.contains("High")); + assert!(!result.description.contains("Medium")); + assert!(!result.description.contains("Low")); + } + + #[test] + fn test_min_severity_filters_both_new_and_preexisting() { + let policy = PatchworkPolicy { + min_severity: Some("Medium".to_string()), + ..Default::default() + }; + let findings = vec![ + finding_new("Low"), + finding_preexisting("Low"), + finding_new("Medium"), + finding_preexisting("High"), + ]; + let result = PatchworkCheckResult::from_policy(&policy, &findings); + assert_eq!(result.state, "warning"); + assert!(result.description.contains("Medium: 1")); + assert!(result.description.contains("High: 1 pre-existing")); + assert!(!result.description.contains("Low")); + } + + #[test] + fn test_min_severity_filters_all_to_success() { + let policy = PatchworkPolicy { + min_severity: Some("Critical".to_string()), + ..Default::default() + }; + let findings = vec![ + finding_new("Low"), + finding_new("Medium"), + finding_new("High"), + ]; + let result = PatchworkCheckResult::from_policy(&policy, &findings); + assert_eq!(result.state, "success"); + assert_eq!(result.description, "Sashiko AI review found no regressions"); + } + + #[test] + fn test_custom_fail_severity_critical() { + let policy = PatchworkPolicy { + fail_severity: "Critical".to_string(), + ..Default::default() + }; + // High is below Critical threshold, so warning not fail + let findings = vec![finding_new("High")]; + let result = PatchworkCheckResult::from_policy(&policy, &findings); + assert_eq!(result.state, "warning"); + } + + #[test] + fn test_custom_fail_severity_low() { + let policy = PatchworkPolicy { + fail_severity: "Low".to_string(), + ..Default::default() + }; + // Any new finding triggers fail + let findings = vec![finding_new("Low")]; + let result = PatchworkCheckResult::from_policy(&policy, &findings); + assert_eq!(result.state, "fail"); + } + + #[test] + fn test_missing_preexisting_treated_as_new() { + // findings without preexisting field default to new + let findings = vec![serde_json::json!({"severity": "High", "problem": "test"})]; + let result = PatchworkCheckResult::from_policy(&default_policy(), &findings); + assert_eq!(result.state, "fail"); + assert!(result.description.contains("High: 1")); + assert!(!result.description.contains("pre-existing")); + } + + #[test] + fn test_null_preexisting_treated_as_new() { + let findings = + vec![serde_json::json!({"severity": "High", "problem": "test", "preexisting": null})]; + let result = PatchworkCheckResult::from_policy(&default_policy(), &findings); + assert_eq!(result.state, "fail"); + } + + #[test] + fn test_description_dot_separator() { + let findings = vec![finding_new("Critical"), finding_new("Medium")]; + let result = PatchworkCheckResult::from_policy(&default_policy(), &findings); + assert!(result.description.contains("\u{00b7}")); // middle dot + } + + // -- compose_patchwork_email tests -- + + #[test] + fn test_compose_patchwork_email_warning() { + let (subject, body) = compose_patchwork_email( + "<12345@kernel.org>", + "warning", + "Sashiko AI review found 2 potential issue(s)", + "https://sashiko.dev/#/patchset/abc?part=1", + "[PATCH] fix null deref in foo", + ); + + assert_eq!( + subject, + "[sashiko-check] warning - [PATCH] fix null deref in foo" + ); + assert!(body.contains("msgid: <12345@kernel.org>")); + assert!(body.contains("status: warning")); + assert!(body.contains("description: Sashiko AI review found 2 potential issue(s)")); + assert!(body.contains("target_url: https://sashiko.dev/#/patchset/abc?part=1")); + assert!(body.contains("context: sashiko")); + } + + #[test] + fn test_compose_patchwork_email_success() { + let (subject, body) = compose_patchwork_email( + "<99@kernel.org>", + "success", + "Sashiko AI review found no regressions", + "https://sashiko.dev/#/patchset/xyz?part=0", + "[PATCH v2] improve error handling", ); + + assert_eq!( + subject, + "[sashiko-check] success - [PATCH v2] improve error handling" + ); + assert!(body.contains("status: success")); + } + + #[test] + fn test_compose_patchwork_email_line_format() { + let (_, body) = compose_patchwork_email( + "", + "warning", + "desc", + "https://example.com", + "subj", + ); + + // Each key-value pair must be on its own line for downstream parsing + let lines: Vec<&str> = body.lines().collect(); + assert_eq!(lines.len(), 5); + assert!(lines[0].starts_with("msgid: ")); + assert!(lines[1].starts_with("status: ")); + assert!(lines[2].starts_with("description: ")); + assert!(lines[3].starts_with("target_url: ")); + assert!(lines[4].starts_with("context: ")); } } diff --git a/src/reviewer.rs b/src/reviewer.rs index 5cdd81f4f..aa9a3e78a 100644 --- a/src/reviewer.rs +++ b/src/reviewer.rs @@ -2146,35 +2146,56 @@ impl Reviewer { let patchwork_policies = crate::email_router::EmailRouter::resolve_patchwork(&policy, &to_list, &cc_list); - let patchwork_status = if findings_count > 0 { - "warning" - } else { - "success" - }; - let patchwork_desc = if findings_count > 0 { - format!( - "Sashiko AI review found {} potential issue(s)", - findings_count - ) - } else { - "Sashiko AI review found no regressions".to_string() - }; + let findings_slice = findings.map(|f| f.as_slice()).unwrap_or(&[]); - for pw_policy in patchwork_policies { - let msg_id_owned = msg_id.to_string(); - let status_owned = patchwork_status.to_string(); - let desc_owned = patchwork_desc.clone(); - let url_owned = target_url.clone(); - tokio::spawn(async move { - crate::patchwork::post_patchwork_check( - &pw_policy, - &msg_id_owned, - &status_owned, - &desc_owned, - &url_owned, - ) - .await; - }); + for pw_policy in &patchwork_policies { + let check_result = + crate::patchwork::PatchworkCheckResult::from_policy(pw_policy, findings_slice); + + // API mode: insert into patchwork_outbox for retry-queued delivery + if let Some(api_url) = &pw_policy.api_url { + ctx.db + .insert_patchwork_outbox( + msg_id, + api_url, + &check_result.state, + &check_result.description, + &target_url, + "sashiko", + ) + .await?; + } + + // Email mode: queue structured notification via email outbox + if let Some(email_addr) = &pw_policy.email { + let (pw_subject, pw_body) = crate::patchwork::compose_patchwork_email( + msg_id, + &check_result.state, + &check_result.description, + &target_url, + &patch_subject, + ); + let pw_email_status = match &ctx.settings.smtp { + None => "Disabled", + Some(s) if s.dry_run => "Dry-Run", + _ => "Pending", + }; + ctx.db + .insert_patchwork_notification( + pw_email_status, + email_addr, + &pw_subject, + msg_id.trim_matches(|c| c == '<' || c == '>'), + msg_id.trim_matches(|c| c == '<' || c == '>'), + &pw_body, + ) + .await?; + } + + // Neither mode configured but patchwork enabled + if pw_policy.api_url.is_none() && pw_policy.email.is_none() { + tracing::warn!("Patchwork enabled but no api_url or email provided"); + } } let action = EmailRouter::resolve_recipients( diff --git a/src/schema.sql b/src/schema.sql index 4aab80574..1d415d7ff 100644 --- a/src/schema.sql +++ b/src/schema.sql @@ -252,4 +252,21 @@ CREATE INDEX IF NOT EXISTS idx_reviews_patchset_status ON reviews(patchset_id, s CREATE INDEX IF NOT EXISTS idx_reviews_day ON reviews(strftime('%Y-%m-%d', created_at, 'unixepoch'), status); CREATE INDEX IF NOT EXISTS idx_email_outbox_patch_id ON email_outbox(patch_id); +CREATE TABLE IF NOT EXISTS patchwork_outbox ( + id INTEGER PRIMARY KEY, + patch_msg_id TEXT NOT NULL, + api_url TEXT NOT NULL, + check_state TEXT NOT NULL, + description TEXT NOT NULL, + target_url TEXT NOT NULL, + context TEXT NOT NULL DEFAULT 'sashiko', + status TEXT DEFAULT 'Pending', + retry_count INTEGER DEFAULT 0, + next_retry_at INTEGER, + locked_at INTEGER, + error_log TEXT, + created_at INTEGER NOT NULL +); +CREATE INDEX IF NOT EXISTS idx_patchwork_outbox_status ON patchwork_outbox(status); + diff --git a/src/worker/email.rs b/src/worker/email.rs index 93f9a1f4d..c8ee7b4d9 100644 --- a/src/worker/email.rs +++ b/src/worker/email.rs @@ -28,7 +28,7 @@ impl EmailWorker { match self.db.lock_pending_email().await { Ok(Some(email)) => { info!( - "Locked pending email ID {} for patch {}", + "Locked pending email ID {} for patch {:?}", email.id, email.patch_id ); match self.send_email(&email).await { diff --git a/src/worker/mod.rs b/src/worker/mod.rs index a168e3c8f..488c08224 100644 --- a/src/worker/mod.rs +++ b/src/worker/mod.rs @@ -13,6 +13,7 @@ // limitations under the License. pub mod email; +pub mod patchwork; pub mod prefetch; pub mod prompts; pub mod tools; diff --git a/src/worker/patchwork.rs b/src/worker/patchwork.rs new file mode 100644 index 000000000..7007f8fee --- /dev/null +++ b/src/worker/patchwork.rs @@ -0,0 +1,148 @@ +// Copyright 2026 The Sashiko Authors +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use crate::db::Database; +use crate::email_policy::EmailPolicyConfig; +use std::sync::Arc; +use std::time::Duration; +use tokio::time::sleep; +use tracing::{error, info, warn}; + +pub struct PatchworkWorker { + db: Arc, + email_policy_path: String, + max_retries: u32, +} + +impl PatchworkWorker { + pub fn new(db: Arc, email_policy_path: String, max_retries: u32) -> Self { + Self { + db, + email_policy_path, + max_retries, + } + } + + /// Resolve the patchwork API token for a given api_url by loading + /// the email policy config and matching against subsystem policies. + /// Tokens are never stored in the database -- they are resolved + /// from the config file (and SASHIKO_PATCHWORK_TOKEN env var) at + /// delivery time. + fn resolve_token(&self, api_url: &str) -> Option { + let config = match EmailPolicyConfig::load(&self.email_policy_path) { + Ok(c) => c, + Err(e) => { + warn!("Failed to load email policy for token resolution: {}", e); + return None; + } + }; + + // Check subsystem policies for a matching api_url + for sub in config.subsystems.values() { + if sub.patchwork.enabled && sub.patchwork.api_url.as_deref() == Some(api_url) { + return sub.patchwork.token.clone(); + } + } + + // Fall back to defaults + if config.defaults.patchwork.enabled + && config.defaults.patchwork.api_url.as_deref() == Some(api_url) + { + return config.defaults.patchwork.token.clone(); + } + + None + } + + /// Compute the backoff delay in seconds based on retry count. + fn backoff_seconds(retry_count: i64) -> i64 { + match retry_count { + 0 => 5, + 1 => 30, + _ => 180, + } + } + + pub async fn run(&self) { + info!("Starting Patchwork Worker..."); + let client = reqwest::Client::new(); + loop { + if let Err(e) = self.db.sweep_ghost_patchwork().await { + error!("Failed to sweep ghost patchwork entries: {}", e); + } + + match self.db.lock_pending_patchwork().await { + Ok(Some(entry)) => { + info!( + "Processing patchwork check ID {} for msgid {}", + entry.id, entry.patch_msg_id + ); + + // Resolve the token from config at delivery time, + // not from the database row. + let token = self.resolve_token(&entry.api_url); + + match crate::patchwork::post_patchwork_check( + &client, + &entry.api_url, + token.as_deref(), + &entry.patch_msg_id, + &entry.check_state, + &entry.description, + &entry.target_url, + ) + .await + { + Ok(()) => { + info!("Successfully posted patchwork check ID {}", entry.id); + if let Err(e) = self.db.mark_patchwork_sent(entry.id).await { + error!("Failed to mark patchwork {} as sent: {}", entry.id, e); + } + } + Err(e) => { + error!("Patchwork check failed for ID {}: {}", entry.id, e); + if entry.retry_count + 1 >= self.max_retries as i64 { + if let Err(db_err) = + self.db.mark_patchwork_failed(entry.id, &e).await + { + error!( + "Failed to mark patchwork {} as failed: {}", + entry.id, db_err + ); + } + } else { + // Schedule retry with a future timestamp + // instead of blocking the worker loop. + let delay = Self::backoff_seconds(entry.retry_count); + let retry_at = chrono::Utc::now().timestamp() + delay; + if let Err(db_err) = + self.db.set_patchwork_retry_at(entry.id, retry_at).await + { + error!("Failed to schedule retry for {}: {}", entry.id, db_err); + } + } + } + } + } + Ok(None) => { + sleep(Duration::from_secs(5)).await; + } + Err(e) => { + error!("Database error while locking patchwork entry: {}", e); + sleep(Duration::from_secs(10)).await; + } + } + } + } +}