From 4468e86e2d7d7e919a683bce4973904fb2a8f893 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?m=C3=BCde?= Date: Tue, 7 Jul 2026 09:10:24 +0200 Subject: [PATCH] refactor(hive-ag3nt): port forge_notify to forgejo-api --- hive-ag3nt/src/forge_notify.rs | 462 ++++++++++++++++++++------------- 1 file changed, 279 insertions(+), 183 deletions(-) diff --git a/hive-ag3nt/src/forge_notify.rs b/hive-ag3nt/src/forge_notify.rs index 99d14203..da6f6d89 100644 --- a/hive-ag3nt/src/forge_notify.rs +++ b/hive-ag3nt/src/forge_notify.rs @@ -24,9 +24,18 @@ use std::fmt::Write as _; use std::path::{Path, PathBuf}; use std::time::Duration; +use forgejo_api::structs::{ + NotificationThread, NotifyGetListQuery, NotifyReadThreadQuery, NotifySubjectType, +}; +use forgejo_api::{Auth, Forgejo, ForgejoError}; +use time::OffsetDateTime; +use time::format_description::well_known::Rfc3339; use tracing::{debug, info, warn}; const POLL_INTERVAL_SECS: u64 = 30; +/// Per-request cap applied to every forge call — natively on the reqwest +/// enrichment client, via `tokio::time::timeout` around the typed +/// `forgejo-api` client (which exposes no timeout knob of its own). const HTTP_TIMEOUT_SECS: u64 = 10; /// Page size of the unread-notifications fetch. This is also the hard /// bound on the persisted delivery-dedupe cursor: each poll prunes the @@ -97,6 +106,27 @@ pub async fn run(socket: PathBuf) { } }; + // Typed Forgejo client for the API calls with stable shapes (identity + // probe, notification list, mark-read). The plain reqwest client below + // stays for the best-effort enrichment fetches of `subject.url` / + // `latest_comment_url`: those follow server-provided URLs whose payload + // shape is heterogeneous (issue vs comment vs review), which the typed + // client cannot express (its `Endpoint` trait is sealed). + let base_url = match url::Url::parse(&forge_url) { + Ok(u) => u, + Err(e) => { + warn!("forge_notify: invalid HIVE_FORGE_URL {forge_url}: {e}"); + return; + } + }; + let forge = match Forgejo::new(Auth::Token(&token), base_url) { + Ok(f) => f, + Err(e) => { + warn!("forge_notify: failed to build forge client: {e}"); + return; + } + }; + let client = match reqwest::Client::builder() .timeout(Duration::from_secs(HTTP_TIMEOUT_SECS)) .build() @@ -111,13 +141,15 @@ pub async fn run(socket: PathBuf) { // Fetch own login once for self-notification filtering. Falls back // to empty string on failure — no filtering (safe degradation; see // `docs/forge.md::Self-notification filtering`). - let own_login = { - let url = format!("{forge_url}/api/v1/user"); - fetch_json(&client, &url, &token) - .await - .and_then(|v| v["login"].as_str().map(std::borrow::ToOwned::to_owned)) - .unwrap_or_default() - }; + let own_login = tokio::time::timeout( + Duration::from_secs(HTTP_TIMEOUT_SECS), + forge.user_get_current().send(), + ) + .await + .ok() + .and_then(Result::ok) + .and_then(|u| u.login) + .unwrap_or_default(); if own_login.is_empty() { warn!("forge_notify: could not resolve own login — self-notification filtering disabled"); } else { @@ -160,15 +192,7 @@ pub async fn run(socket: PathBuf) { loop { interval.tick().await; - poll_once( - &client, - &forge_url, - &token, - &socket, - &mut delivered, - &own_login, - ) - .await; + poll_once(&forge, &client, &token, &socket, &mut delivered, &own_login).await; } } @@ -188,14 +212,16 @@ async fn fetch_json(client: &reqwest::Client, url: &str, token: &str) -> Option< } /// Map a Forgejo notification `subject.type` to a human-readable label. -/// Known values: "Pull", "Issue", "Commit", "Repository". Any unknown -/// type is passed through as-is so new Forgejo types degrade gracefully -/// rather than silently collapsing into a generic label. -fn notif_type_label(t: &str) -> &str { +/// `Commit` / `Repository` keep their API names, matching the old raw +/// pass-through of types we don't relabel; a missing type degrades to +/// `?` like every other absent field. +fn notif_type_label(t: Option) -> &'static str { match t { - "Pull" => "PR", - "Issue" => "issue", - other => other, + Some(NotifySubjectType::Pull) => "PR", + Some(NotifySubjectType::Issue) => "issue", + Some(NotifySubjectType::Commit) => "Commit", + Some(NotifySubjectType::Repository) => "Repository", + None => "?", } } @@ -349,14 +375,15 @@ fn review_state_label(state: &str) -> Option<&str> { async fn format_notification( client: &reqwest::Client, token: &str, - notif: &serde_json::Value, + notif: &PolledNotification, own_login: &str, ) -> Option { - let title = notif["subject"]["title"].as_str().unwrap_or("?"); - let notif_type = notif["subject"]["type"].as_str().unwrap_or("?"); - let html_url = notif["subject"]["html_url"] - .as_str() - .unwrap_or_else(|| notif["subject"]["url"].as_str().unwrap_or("")); + let subj = notif.thread.subject.as_ref(); + let title = subj.and_then(|s| s.title.as_deref()).unwrap_or("?"); + let subject_type = subj.and_then(|s| s.r#type); + let html_url = subj + .and_then(|s| s.html_url.as_ref().or(s.url.as_ref())) + .map_or("", url::Url::as_str); // Extract issue/PR number from the html_url. URL ends with /issues/N or // /pulls/N (possibly followed by #anchor for comments). Best-effort. @@ -369,19 +396,24 @@ async fn format_notification( .unwrap_or_default(); // Repo slug for multi-repo disambiguation. Falls back gracefully when absent. - let repo = notif["repository"]["full_name"] - .as_str() + let repo = notif + .thread + .repository + .as_ref() + .and_then(|r| r.full_name.as_deref()) .map(|r| format!(" {r}")) .unwrap_or_default(); // API URLs for fetching content - let subject_api_url = notif["subject"]["url"].as_str().unwrap_or(""); - let comment_api_url = notif["subject"]["latest_comment_url"] - .as_str() - .unwrap_or(""); - let comment_html_url = notif["subject"]["latest_comment_html_url"] - .as_str() - .unwrap_or(""); + let subject_api_url = subj + .and_then(|s| s.url.as_ref()) + .map_or("", url::Url::as_str); + let comment_api_url = subj + .and_then(|s| s.latest_comment_url.as_ref()) + .map_or("", url::Url::as_str); + let comment_html_url = subj + .and_then(|s| s.latest_comment_html_url.as_ref()) + .map_or("", url::Url::as_str); // Always fetch subject detail for assignee/reviewer metadata so // the meta suffix can show current ownership without a follow-up @@ -394,7 +426,7 @@ async fn format_notification( // Forgejo's notification `subject.type` is "Pull" / "Issue", never // "Pull Request". - let is_pr = notif_type == "Pull"; + let is_pr = subject_type == Some(NotifySubjectType::Pull); let meta_suffix = build_meta_suffix(subject.as_ref(), is_pr); // Determine whether this notification was triggered by a comment/review or @@ -403,7 +435,7 @@ async fn format_notification( let meta = NotifMeta { title, - notif_type, + subject_type, html_url, num, repo, @@ -422,14 +454,14 @@ async fn format_notification( ) .await } else { - format_state_change_notification(notif, &meta, own_login) + format_state_change_notification(notif.thread.updated_at, ¬if.state, &meta, own_login) } } -/// Shared notification metadata extracted from the raw Forgejo JSON. +/// Shared notification metadata extracted from the polled notification. struct NotifMeta<'a> { title: &'a str, - notif_type: &'a str, + subject_type: Option, html_url: &'a str, num: String, repo: String, @@ -522,7 +554,7 @@ async fn format_comment_notification( }; let NotifMeta { title, - notif_type, + subject_type, num, repo, meta_suffix, @@ -547,7 +579,7 @@ async fn format_comment_notification( Some(out) } else { // Regular comment. - let kind = format!("comment on {}{num}{repo}", notif_type_label(notif_type)); + let kind = format!("comment on {}{num}{repo}", notif_type_label(*subject_type)); let mut out = format!( "[{kind}] {title}\nurl: {url}\n\n{author}: {body_for_embed}{truncated_mentions}" ); @@ -563,17 +595,17 @@ async fn format_comment_notification( /// path applies. Only creations are dropped: a later state change on the /// agent's own subject is driven by someone else and stays a wake. fn format_state_change_notification( - notif: &serde_json::Value, + event_time: Option, + notif_state: &str, meta: &NotifMeta<'_>, own_login: &str, ) -> Option { - // Classification uses notif["subject"]["state"] directly — Forgejo - // returns "open" / "closed" / "merged" here. We do NOT rely on - // fetching the PR/issue detail for `merged`: + // Classification uses the raw `subject.state` string extracted in + // `parse_notification` — Forgejo returns "open" / "closed" / "merged" + // there. We do NOT rely on fetching the PR/issue detail for `merged`: // - `subject.url` points to the *issues* endpoint, which returns // `pull_request.merged`, not top-level `merged`. // - Forgejo API type is "Pull" / "Issue", never "Pull Request". - let notif_state = notif["subject"]["state"].as_str().unwrap_or(""); // "New" = the subject is open (or state is absent). Used below for // the review-request override. @@ -581,7 +613,7 @@ fn format_state_change_notification( let NotifMeta { title, - notif_type, + subject_type, html_url, num, repo, @@ -589,7 +621,7 @@ fn format_state_change_notification( subject, is_pr, } = meta; - let label = notif_type_label(notif_type); + let label = notif_type_label(*subject_type); // Only claim "new" when the notification actually fired at creation // time. A review submitted with no body carries no // `latest_comment_url`, so it lands here instead of on the comment @@ -598,7 +630,7 @@ fn format_state_change_notification( // on"): agents dismiss it as a // duplicate of the original open notification. When we can't confirm // creation, fall back to a neutral "activity on" label. - let looks_new = notification_is_creation(notif, subject.as_ref()); + let looks_new = notification_is_creation(event_time, subject.as_ref()); // Self-authored creation filter: skip an agent being woken by its own // freshly-opened issue/PR. The subject payload is already fetched (for @@ -668,81 +700,66 @@ fn format_state_change_notification( /// new item behind the neutral fallback. See docs/forge.md, "new vs /// activity on". fn notification_is_creation( - notif: &serde_json::Value, + event: Option, subject: Option<&serde_json::Value>, ) -> bool { let created = subject .and_then(|s| s["created_at"].as_str()) - .and_then(parse_rfc3339_secs); - let event = notif["updated_at"].as_str().and_then(parse_rfc3339_secs); + .and_then(parse_rfc3339); match (created, event) { - (Some(c), Some(e)) => (e - c).abs() <= NEW_ITEM_TOLERANCE_SECS, + (Some(c), Some(e)) => (e - c).whole_seconds().abs() <= NEW_ITEM_TOLERANCE_SECS, _ => true, } } -/// Minimal dependency-free RFC 3339 / ISO 8601 parser → Unix epoch -/// seconds. Forgejo emits timestamps like `2026-06-13T11:18:42+02:00` -/// or `...Z`, optionally with fractional seconds. We only need -/// second-granularity comparison, so the fractional part is skipped. -/// Returns `None` on any shape we don't recognise so callers can fall -/// back gracefully. -fn parse_rfc3339_secs(s: &str) -> Option { - if s.len() < 19 { - return None; - } - let year: i64 = s.get(0..4)?.parse().ok()?; - let month: i64 = s.get(5..7)?.parse().ok()?; - let day: i64 = s.get(8..10)?.parse().ok()?; - let hour: i64 = s.get(11..13)?.parse().ok()?; - let minute: i64 = s.get(14..16)?.parse().ok()?; - let second: i64 = s.get(17..19)?.parse().ok()?; - - let mut epoch = - days_from_civil(year, month, day) * 86_400 + hour * 3_600 + minute * 60 + second; - - // Remainder after seconds: optional `.fff` fraction, then a zone. - let mut rest = &s[19..]; - if let Some(frac) = rest.strip_prefix('.') { - let end = frac - .find(|c: char| !c.is_ascii_digit()) - .unwrap_or(frac.len()); - rest = &frac[end..]; - } - // Zone: `Z`/empty = UTC; otherwise `±HH:MM`. Subtract the offset to - // normalise to UTC epoch seconds. - if !(rest.is_empty() || rest.starts_with('Z')) { - let sign = rest.as_bytes()[0]; - let off = &rest[1..]; - // Accept `HH:MM` (Forgejo's form) and bare `HHMM`; minutes optional. - // Both fields fail the same way — a present-but-unparseable component - // returns `None` rather than one silently defaulting. - let (hh, mm) = match off.split_once(':') { - Some((h, m)) => (h, m), - None => (off.get(0..2)?, off.get(2..4).unwrap_or("00")), - }; - let oh: i64 = hh.parse().ok()?; - let om: i64 = if mm.is_empty() { 0 } else { mm.parse().ok()? }; - let offset = oh * 3_600 + om * 60; - match sign { - b'+' => epoch -= offset, - b'-' => epoch += offset, - _ => return None, - } - } - Some(epoch) +/// Parse an RFC 3339 timestamp as Forgejo emits them +/// (`2026-06-13T11:18:42+02:00` or `...Z`, optionally with fractional +/// seconds). Returns `None` on any shape `time` doesn't recognise so +/// callers can fall back gracefully. +fn parse_rfc3339(s: &str) -> Option { + OffsetDateTime::parse(s, &Rfc3339).ok() } -/// Days since the Unix epoch for a proleptic-Gregorian `y-m-d` -/// (Howard Hinnant's `days_from_civil`). Valid for all dates Forgejo -/// can emit. -fn days_from_civil(y: i64, m: i64, d: i64) -> i64 { - let y = if m <= 2 { y - 1 } else { y }; - let era = (if y >= 0 { y } else { y - 399 }) / 400; - let yoe = y - era * 400; - let doy = (153 * (if m > 2 { m - 3 } else { m + 9 }) + 2) / 5 + d - 1; - let doe = yoe * 365 + yoe / 4 - yoe / 100 + doy; - era * 146_097 + doe - 719_468 +/// One notification from the poll page: the typed thread plus two raw +/// fields the typed structs can't carry faithfully. +struct PolledNotification { + thread: NotificationThread, + /// Raw `subject.state`. Forgejo reports `"merged"` for merged PRs + /// (`services/convert/notification.go`), which forgejo-api's + /// `StateType` (open/closed only) rejects at deserialization — so + /// the state is extracted verbatim before the typed parse and + /// matched as a string, exactly like the pre-typed code. + state: String, + /// Raw `updated_at` string, byte-identical to what Forgejo sent, so + /// the persisted delivery-dedupe cursor keeps its exact format + /// across the typed-client port (no spurious re-deliveries from a + /// reformatting round-trip). + updated_at: String, +} + +/// Parse one notification JSON object: pull out the raw `subject.state` +/// and `updated_at` (see [`PolledNotification`]), null the state so the +/// closed `StateType` enum can't reject it, and deserialize the rest +/// into the typed [`NotificationThread`]. Returns `None` (with a warn) +/// for an item the typed struct can't represent — the rest of the page +/// still delivers. +fn parse_notification(mut value: serde_json::Value) -> Option { + let state = value["subject"]["state"].as_str().unwrap_or("").to_owned(); + if let Some(s) = value.get_mut("subject").and_then(|s| s.get_mut("state")) { + *s = serde_json::Value::Null; + } + let updated_at = value["updated_at"].as_str().unwrap_or("").to_owned(); + match serde_json::from_value::(value) { + Ok(thread) => Some(PolledNotification { + thread, + state, + updated_at, + }), + Err(e) => { + warn!("forge_notify: notification parse error: {e}"); + None + } + } } #[allow( @@ -752,33 +769,45 @@ fn days_from_civil(y: i64, m: i64, d: i64) -> i64 { functions for state shared across all three phases" )] async fn poll_once( + forge: &Forgejo, client: &reqwest::Client, - forge_url: &str, token: &str, socket: &Path, delivered: &mut HashMap, own_login: &str, ) { - let url = format!("{forge_url}/api/v1/notifications?all=false&limit={UNREAD_FETCH_LIMIT}"); - let resp = match client - .get(&url) - .header("Authorization", format!("token {token}")) - .send() - .await - { - Ok(r) => r, - Err(e) => { - debug!("forge_notify: poll request failed: {e}"); - return; - } + // Fetch the page as raw JSON (`response_type::`) instead of the + // crate's `Vec`: one merged-PR notification + // (`subject.state = "merged"`, unrepresentable in `StateType`) would + // otherwise poison deserialization of the whole page. HTTP-level errors + // still surface as `ForgejoError` exactly like the fully-typed call; + // `parse_notification` below does the per-item typed parse. + let query = NotifyGetListQuery { + all: Some(false), + ..NotifyGetListQuery::default() }; + let request = forge + .notify_get_list(query) + .page_size(u32::try_from(UNREAD_FETCH_LIMIT).unwrap_or(u32::MAX)) + .response_type::(); + let raw = + match tokio::time::timeout(Duration::from_secs(HTTP_TIMEOUT_SECS), request.send()).await { + Ok(Ok(raw)) => raw, + Ok(Err(ForgejoError::UnexpectedStatusCode(status))) => { + debug!("forge_notify: poll status {status}"); + return; + } + Ok(Err(e)) => { + debug!("forge_notify: poll request failed: {e}"); + return; + } + Err(_) => { + debug!("forge_notify: poll request failed: timed out"); + return; + } + }; - if !resp.status().is_success() { - debug!("forge_notify: poll status {}", resp.status()); - return; - } - - let notifications: Vec = match resp.json().await { + let values: Vec = match serde_json::from_str(&raw) { Ok(v) => v, Err(e) => { warn!("forge_notify: response parse error: {e}"); @@ -786,22 +815,25 @@ async fn poll_once( } }; - if notifications.is_empty() { + if values.is_empty() { return; } debug!( - count = notifications.len(), + count = values.len(), "forge_notify: delivering notifications" ); + let notifications: Vec = + values.into_iter().filter_map(parse_notification).collect(); + // Tracks whether the dedupe cursor changed this poll (a new delivery // recorded, or the prune below dropped now-read threads) so we only // rewrite the on-disk cursor when there's something to persist. let mut cursor_dirty = false; for notif in ¬ifications { - let Some(id) = notif["id"].as_u64() else { + let Some(id) = notif.thread.id.and_then(|id| u64::try_from(id).ok()) else { continue; }; @@ -810,7 +842,7 @@ async fn poll_once( // Skip it silently unless its `updated_at` advanced since the // version we last delivered a wake for (i.e. genuinely new // activity). See the `delivered` cursor note in `run`. - let updated_at = notif["updated_at"].as_str().unwrap_or("").to_owned(); + let updated_at = notif.updated_at.clone(); if !should_deliver(delivered, id, &updated_at) { debug!(%id, "forge_notify: skipping (already delivered this version)"); continue; @@ -820,7 +852,7 @@ async fn poll_once( // None means self-echo — mark read silently, no delivery. let Some(body) = body_opt else { - mark_read(client, forge_url, token, id).await; + mark_read(forge, id).await; continue; }; @@ -864,7 +896,7 @@ async fn poll_once( // loud in tests/dev if a future pagination change silently breaks it. let current_ids: HashSet = notifications .iter() - .filter_map(|n| n["id"].as_u64()) + .filter_map(|n| n.thread.id.and_then(|id| u64::try_from(id).ok())) .collect(); let before_prune = delivered.len(); delivered.retain(|id, _| current_ids.contains(id)); @@ -899,21 +931,31 @@ fn should_deliver(delivered: &HashMap, id: u64, updated_at: &str) - /// deliberately left unread for the read-before-comment guard, and a failed /// delivery is left unread + out of the dedupe cursor so it resurfaces on the /// next poll tick. -async fn mark_read(client: &reqwest::Client, forge_url: &str, token: &str, id: u64) { - let mark_url = format!("{forge_url}/api/v1/notifications/threads/{id}"); - match client - .patch(&mark_url) - .header("Authorization", format!("token {token}")) - .send() - .await - { - Err(e) => { +async fn mark_read(forge: &Forgejo, id: u64) { + let Ok(thread_id) = i64::try_from(id) else { + // Thread ids originate from `i64` in the poll parse, so an + // unrepresentable value can't actually reach here. + return; + }; + // `to_status: None` → Forgejo's default transition (unread → read), + // matching the old bare PATCH. The 205 response body is the thread + // JSON, which can carry `subject.state = "merged"` — take it as an + // opaque `String` (see `poll_once`) so a merged PR doesn't turn a + // successful mark-read into a spurious parse error. + let request = forge + .notify_read_thread(thread_id, NotifyReadThreadQuery { to_status: None }) + .response_type::(); + match tokio::time::timeout(Duration::from_secs(HTTP_TIMEOUT_SECS), request.send()).await { + Err(_) => { + warn!(%id, "forge_notify: mark-read request failed — notification will resurface"); + } + Ok(Err(e @ ForgejoError::ReqwestError(_))) => { warn!(%id, error = ?e, "forge_notify: mark-read request failed — notification will resurface"); } - Ok(r) if !r.status().is_success() => { - warn!(%id, status = %r.status(), "forge_notify: mark-read returned non-2xx — notification will resurface"); + Ok(Err(e)) => { + warn!(%id, error = %e, "forge_notify: mark-read returned non-2xx — notification will resurface"); } - Ok(_) => { + Ok(Ok(_)) => { debug!(%id, "forge_notify: marked read"); } } @@ -1129,39 +1171,98 @@ mod tests { } #[test] - fn parse_rfc3339_secs_handles_offsets_and_z() { - // Same instant expressed three ways must parse equal. - let utc = parse_rfc3339_secs("2026-06-13T09:18:42Z").unwrap(); - let plus2 = parse_rfc3339_secs("2026-06-13T11:18:42+02:00").unwrap(); - let minus5 = parse_rfc3339_secs("2026-06-13T04:18:42-05:00").unwrap(); + fn parse_rfc3339_handles_offsets_and_z() { + // Same instant expressed three ways must parse equal + // (`OffsetDateTime` comparison is instant-based). + let utc = parse_rfc3339("2026-06-13T09:18:42Z").unwrap(); + let plus2 = parse_rfc3339("2026-06-13T11:18:42+02:00").unwrap(); + let minus5 = parse_rfc3339("2026-06-13T04:18:42-05:00").unwrap(); assert_eq!(utc, plus2); assert_eq!(utc, minus5); - // Fractional seconds are skipped (second granularity). - assert_eq!(parse_rfc3339_secs("2026-06-13T09:18:42.512Z").unwrap(), utc); + // Fractional seconds parse; second-granularity comparison holds. + assert_eq!( + parse_rfc3339("2026-06-13T09:18:42.512Z") + .unwrap() + .unix_timestamp(), + utc.unix_timestamp(), + ); } #[test] - fn parse_rfc3339_secs_rejects_garbage() { - assert!(parse_rfc3339_secs("").is_none()); - assert!(parse_rfc3339_secs("not-a-date").is_none()); - assert!(parse_rfc3339_secs("2026-06-13").is_none()); + fn parse_rfc3339_rejects_garbage() { + assert!(parse_rfc3339("").is_none()); + assert!(parse_rfc3339("not-a-date").is_none()); + assert!(parse_rfc3339("2026-06-13").is_none()); } #[test] fn notification_is_creation_flags_fresh_and_later_activity() { - // Fresh PR: notification event time == created_at → "new". - let fresh = serde_json::json!({ "updated_at": "2026-06-13T11:18:42+02:00" }); let subject = serde_json::json!({ "created_at": "2026-06-13T11:18:40+02:00" }); - assert!(notification_is_creation(&fresh, Some(&subject))); + + // Fresh PR: notification event time == created_at → "new". + let fresh = parse_rfc3339("2026-06-13T11:18:42+02:00"); + assert!(notification_is_creation(fresh, Some(&subject))); // Review hours later on the same PR → not a creation. - let later = serde_json::json!({ "updated_at": "2026-06-13T14:55:00+02:00" }); - assert!(!notification_is_creation(&later, Some(&subject))); + let later = parse_rfc3339("2026-06-13T14:55:00+02:00"); + assert!(!notification_is_creation(later, Some(&subject))); // Missing timestamps → default to creation (preserve prior behavior). - let empty = serde_json::json!({}); - assert!(notification_is_creation(&empty, None)); - assert!(notification_is_creation(&empty, Some(&subject))); + assert!(notification_is_creation(None, None)); + assert!(notification_is_creation(None, Some(&subject))); + assert!(notification_is_creation( + fresh, + Some(&serde_json::json!({})) + )); + } + + #[test] + fn parse_notification_tolerates_merged_state() { + // forgejo-api's `StateType` has no "merged" variant; the raw + // extraction must keep the item parseable AND preserve the + // string for the `PR merged` wrapper. + let polled = parse_notification(serde_json::json!({ + "id": 7, + "updated_at": "2026-06-13T11:18:42+02:00", + "url": "http://forge/api/v1/notifications/threads/7", + "subject": { + "title": "t", + "type": "Pull", + "state": "merged", + "html_url": "http://forge/o/r/pulls/5", + "latest_comment_html_url": "", + "latest_comment_url": "", + "url": "http://forge/api/v1/repos/o/r/issues/5", + }, + })) + .expect("merged-state notification must parse"); + assert_eq!(polled.state, "merged"); + // Cursor string is the raw `updated_at`, byte-identical. + assert_eq!(polled.updated_at, "2026-06-13T11:18:42+02:00"); + assert_eq!(polled.thread.id, Some(7)); + assert_eq!( + polled.thread.subject.as_ref().and_then(|s| s.r#type), + Some(NotifySubjectType::Pull), + ); + } + + #[test] + fn parse_notification_missing_subject_degrades() { + let polled = parse_notification(serde_json::json!({ + "id": 1, + "updated_at": "2026-06-22T16:00:00Z", + "url": "", + })) + .expect("subject-less notification must parse"); + assert_eq!(polled.state, ""); + assert!(polled.thread.subject.is_none()); + } + + #[test] + fn parse_notification_rejects_unrepresentable_item() { + // A non-object item can't become a `NotificationThread`; it is + // dropped alone instead of failing the page. + assert!(parse_notification(serde_json::json!("nonsense")).is_none()); } /// Build a `NotifMeta` for the state-change formatter tests. The `&str` @@ -1169,7 +1270,7 @@ mod tests { fn state_change_meta(subject: serde_json::Value) -> NotifMeta<'static> { NotifMeta { title: "subject title", - notif_type: "Issue", + subject_type: Some(NotifySubjectType::Issue), html_url: "http://forge/issues/1", num: " #1".to_owned(), repo: " [agents/x]".to_owned(), @@ -1183,15 +1284,13 @@ mod tests { fn state_change_drops_self_authored_creation() { // No timestamps ⇒ treated as a creation; poster login == own_login. let meta = state_change_meta(serde_json::json!({ "user": { "login": "damocles" } })); - let notif = serde_json::json!({ "subject": { "state": "open" } }); - assert!(format_state_change_notification(¬if, &meta, "damocles").is_none()); + assert!(format_state_change_notification(None, "open", &meta, "damocles").is_none()); } #[test] fn state_change_keeps_other_authored_creation() { let meta = state_change_meta(serde_json::json!({ "user": { "login": "someone-else" } })); - let notif = serde_json::json!({ "subject": { "state": "open" } }); - assert!(format_state_change_notification(¬if, &meta, "damocles").is_some()); + assert!(format_state_change_notification(None, "open", &meta, "damocles").is_some()); } #[test] @@ -1203,10 +1302,7 @@ mod tests { "user": { "login": "damocles" }, "created_at": "2020-01-01T00:00:00Z", })); - let notif = serde_json::json!({ - "subject": { "state": "closed" }, - "updated_at": "2026-06-22T16:00:00Z", - }); - assert!(format_state_change_notification(¬if, &meta, "damocles").is_some()); + let event = parse_rfc3339("2026-06-22T16:00:00Z"); + assert!(format_state_change_notification(event, "closed", &meta, "damocles").is_some()); } }