vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

refactor(hive-ag3nt): port forge_notify to forgejo-api

Browse source
This commit is contained in:
müde 2026-07-07 09:10:24 +02:00
commit 4468e86e2d
1 changed files with 278 additions and 182 deletions
Download patch file Download diff file Expand all files Collapse all files

462
hive-ag3nt/src/forge_notify.rs
View file

@ -24,9 +24,18 @@ use std::fmt::Write as _;
use std::path::{Path, PathBuf}; use std::path::{Path, PathBuf};
use std::time::Duration; 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}; use tracing::{debug, info, warn};
const POLL_INTERVAL_SECS: u64 = 30; 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; const HTTP_TIMEOUT_SECS: u64 = 10;
/// Page size of the unread-notifications fetch. This is also the hard /// Page size of the unread-notifications fetch. This is also the hard
/// bound on the persisted delivery-dedupe cursor: each poll prunes the /// 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() let client = match reqwest::Client::builder()
.timeout(Duration::from_secs(HTTP_TIMEOUT_SECS)) .timeout(Duration::from_secs(HTTP_TIMEOUT_SECS))
.build() .build()
@ -111,13 +141,15 @@ pub async fn run(socket: PathBuf) {
// Fetch own login once for self-notification filtering. Falls back // Fetch own login once for self-notification filtering. Falls back
// to empty string on failure — no filtering (safe degradation; see // to empty string on failure — no filtering (safe degradation; see
// `docs/forge.md::Self-notification filtering`). // `docs/forge.md::Self-notification filtering`).
let own_login = { let own_login = tokio::time::timeout(
let url = format!("{forge_url}/api/v1/user"); Duration::from_secs(HTTP_TIMEOUT_SECS),
fetch_json(&client, &url, &token) forge.user_get_current().send(),
.await )
.and_then(|v| v["login"].as_str().map(std::borrow::ToOwned::to_owned)) .await
.unwrap_or_default() .ok()
}; .and_then(Result::ok)
.and_then(|u| u.login)
.unwrap_or_default();
if own_login.is_empty() { if own_login.is_empty() {
warn!("forge_notify: could not resolve own login — self-notification filtering disabled"); warn!("forge_notify: could not resolve own login — self-notification filtering disabled");
} else { } else {
@ -160,15 +192,7 @@ pub async fn run(socket: PathBuf) {
loop { loop {
interval.tick().await; interval.tick().await;
poll_once( poll_once(&forge, &client, &token, &socket, &mut delivered, &own_login).await;
&client,
&forge_url,
&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. /// Map a Forgejo notification `subject.type` to a human-readable label.
/// Known values: "Pull", "Issue", "Commit", "Repository". Any unknown /// `Commit` / `Repository` keep their API names, matching the old raw
/// type is passed through as-is so new Forgejo types degrade gracefully /// pass-through of types we don't relabel; a missing type degrades to
/// rather than silently collapsing into a generic label. /// `?` like every other absent field.
fn notif_type_label(t: &str) -> &str { fn notif_type_label(t: Option<NotifySubjectType>) -> &'static str {
match t { match t {
"Pull" => "PR", Some(NotifySubjectType::Pull) => "PR",
"Issue" => "issue", Some(NotifySubjectType::Issue) => "issue",
other => other, 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( async fn format_notification(
client: &reqwest::Client, client: &reqwest::Client,
token: &str, token: &str,
notif: &serde_json::Value, notif: &PolledNotification,
own_login: &str, own_login: &str,
) -> Option<String> { ) -> Option<String> {
let title = notif["subject"]["title"].as_str().unwrap_or("?"); let subj = notif.thread.subject.as_ref();
let notif_type = notif["subject"]["type"].as_str().unwrap_or("?"); let title = subj.and_then(|s| s.title.as_deref()).unwrap_or("?");
let html_url = notif["subject"]["html_url"] let subject_type = subj.and_then(|s| s.r#type);
.as_str() let html_url = subj
.unwrap_or_else(|| notif["subject"]["url"].as_str().unwrap_or("")); .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 // Extract issue/PR number from the html_url. URL ends with /issues/N or
// /pulls/N (possibly followed by #anchor for comments). Best-effort. // /pulls/N (possibly followed by #anchor for comments). Best-effort.
@ -369,19 +396,24 @@ async fn format_notification(
.unwrap_or_default(); .unwrap_or_default();
// Repo slug for multi-repo disambiguation. Falls back gracefully when absent. // Repo slug for multi-repo disambiguation. Falls back gracefully when absent.
let repo = notif["repository"]["full_name"] let repo = notif
.as_str() .thread
.repository
.as_ref()
.and_then(|r| r.full_name.as_deref())
.map(|r| format!(" {r}")) .map(|r| format!(" {r}"))
.unwrap_or_default(); .unwrap_or_default();
// API URLs for fetching content // API URLs for fetching content
let subject_api_url = notif["subject"]["url"].as_str().unwrap_or(""); let subject_api_url = subj
let comment_api_url = notif["subject"]["latest_comment_url"] .and_then(|s| s.url.as_ref())
.as_str() .map_or("", url::Url::as_str);
.unwrap_or(""); let comment_api_url = subj
let comment_html_url = notif["subject"]["latest_comment_html_url"] .and_then(|s| s.latest_comment_url.as_ref())
.as_str() .map_or("", url::Url::as_str);
.unwrap_or(""); 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 // Always fetch subject detail for assignee/reviewer metadata so
// the meta suffix can show current ownership without a follow-up // 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 // Forgejo's notification `subject.type` is "Pull" / "Issue", never
// "Pull Request". // "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); let meta_suffix = build_meta_suffix(subject.as_ref(), is_pr);
// Determine whether this notification was triggered by a comment/review or // Determine whether this notification was triggered by a comment/review or
@ -403,7 +435,7 @@ async fn format_notification(
let meta = NotifMeta { let meta = NotifMeta {
title, title,
notif_type, subject_type,
html_url, html_url,
num, num,
repo, repo,
@ -422,14 +454,14 @@ async fn format_notification(
) )
.await .await
} else { } else {
format_state_change_notification(notif, &meta, own_login) format_state_change_notification(notif.thread.updated_at, &notif.state, &meta, own_login)
} }
} }
/// Shared notification metadata extracted from the raw Forgejo JSON. /// Shared notification metadata extracted from the polled notification.
struct NotifMeta<'a> { struct NotifMeta<'a> {
title: &'a str, title: &'a str,
notif_type: &'a str, subject_type: Option<NotifySubjectType>,
html_url: &'a str, html_url: &'a str,
num: String, num: String,
repo: String, repo: String,
@ -522,7 +554,7 @@ async fn format_comment_notification(
}; };
let NotifMeta { let NotifMeta {
title, title,
notif_type, subject_type,
num, num,
repo, repo,
meta_suffix, meta_suffix,
@ -547,7 +579,7 @@ async fn format_comment_notification(
Some(out) Some(out)
} else { } else {
// Regular comment. // 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!( let mut out = format!(
"[{kind}] {title}\nurl: {url}\n\n{author}: {body_for_embed}{truncated_mentions}" "[{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 /// 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. /// agent's own subject is driven by someone else and stays a wake.
fn format_state_change_notification( fn format_state_change_notification(
notif: &serde_json::Value, event_time: Option<OffsetDateTime>,
notif_state: &str,
meta: &NotifMeta<'_>, meta: &NotifMeta<'_>,
own_login: &str, own_login: &str,
) -> Option<String> { ) -> Option<String> {
// Classification uses notif["subject"]["state"] directly — Forgejo // Classification uses the raw `subject.state` string extracted in
// returns "open" / "closed" / "merged" here. We do NOT rely on // `parse_notification` — Forgejo returns "open" / "closed" / "merged"
// fetching the PR/issue detail for `merged`: // there. We do NOT rely on fetching the PR/issue detail for `merged`:
// - `subject.url` points to the *issues* endpoint, which returns // - `subject.url` points to the *issues* endpoint, which returns
// `pull_request.merged`, not top-level `merged`. // `pull_request.merged`, not top-level `merged`.
// - Forgejo API type is "Pull" / "Issue", never "Pull Request". // - 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 // "New" = the subject is open (or state is absent). Used below for
// the review-request override. // the review-request override.
@ -581,7 +613,7 @@ fn format_state_change_notification(
let NotifMeta { let NotifMeta {
title, title,
notif_type, subject_type,
html_url, html_url,
num, num,
repo, repo,
@ -589,7 +621,7 @@ fn format_state_change_notification(
subject, subject,
is_pr, is_pr,
} = meta; } = 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 // Only claim "new" when the notification actually fired at creation
// time. A review submitted with no body carries no // time. A review submitted with no body carries no
// `latest_comment_url`, so it lands here instead of on the comment // `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 // on"): agents dismiss it as a
// duplicate of the original open notification. When we can't confirm // duplicate of the original open notification. When we can't confirm
// creation, fall back to a neutral "activity on" label. // 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 // Self-authored creation filter: skip an agent being woken by its own
// freshly-opened issue/PR. The subject payload is already fetched (for // 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 /// new item behind the neutral fallback. See docs/forge.md, "new vs
/// activity on". /// activity on".
fn notification_is_creation( fn notification_is_creation(
notif: &serde_json::Value, event: Option<OffsetDateTime>,
subject: Option<&serde_json::Value>, subject: Option<&serde_json::Value>,
) -> bool { ) -> bool {
let created = subject let created = subject
.and_then(|s| s["created_at"].as_str()) .and_then(|s| s["created_at"].as_str())
.and_then(parse_rfc3339_secs); .and_then(parse_rfc3339);
let event = notif["updated_at"].as_str().and_then(parse_rfc3339_secs);
match (created, event) { 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, _ => true,
} }
} }
/// Minimal dependency-free RFC 3339 / ISO 8601 parser → Unix epoch /// Parse an RFC 3339 timestamp as Forgejo emits them
/// seconds. Forgejo emits timestamps like `2026-06-13T11:18:42+02:00` /// (`2026-06-13T11:18:42+02:00` or `...Z`, optionally with fractional
/// or `...Z`, optionally with fractional seconds. We only need /// seconds). Returns `None` on any shape `time` doesn't recognise so
/// second-granularity comparison, so the fractional part is skipped. /// callers can fall back gracefully.
/// Returns `None` on any shape we don't recognise so callers can fall fn parse_rfc3339(s: &str) -> Option<OffsetDateTime> {
/// back gracefully. OffsetDateTime::parse(s, &Rfc3339).ok()
fn parse_rfc3339_secs(s: &str) -> Option<i64> {
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)
} }
/// Days since the Unix epoch for a proleptic-Gregorian `y-m-d` /// One notification from the poll page: the typed thread plus two raw
/// (Howard Hinnant's `days_from_civil`). Valid for all dates Forgejo /// fields the typed structs can't carry faithfully.
/// can emit. struct PolledNotification {
fn days_from_civil(y: i64, m: i64, d: i64) -> i64 { thread: NotificationThread,
let y = if m <= 2 { y - 1 } else { y }; /// Raw `subject.state`. Forgejo reports `"merged"` for merged PRs
let era = (if y >= 0 { y } else { y - 399 }) / 400; /// (`services/convert/notification.go`), which forgejo-api's
let yoe = y - era * 400; /// `StateType` (open/closed only) rejects at deserialization — so
let doy = (153 * (if m > 2 { m - 3 } else { m + 9 }) + 2) / 5 + d - 1; /// the state is extracted verbatim before the typed parse and
let doe = yoe * 365 + yoe / 4 - yoe / 100 + doy; /// matched as a string, exactly like the pre-typed code.
era * 146_097 + doe - 719_468 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<PolledNotification> {
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::<NotificationThread>(value) {
Ok(thread) => Some(PolledNotification {
thread,
state,
updated_at,
}),
Err(e) => {
warn!("forge_notify: notification parse error: {e}");
None
}
}
} }
#[allow( #[allow(
@ -752,33 +769,45 @@ fn days_from_civil(y: i64, m: i64, d: i64) -> i64 {
functions for state shared across all three phases" functions for state shared across all three phases"
)] )]
async fn poll_once( async fn poll_once(
forge: &Forgejo,
client: &reqwest::Client, client: &reqwest::Client,
forge_url: &str,
token: &str, token: &str,
socket: &Path, socket: &Path,
delivered: &mut HashMap<u64, String>, delivered: &mut HashMap<u64, String>,
own_login: &str, own_login: &str,
) { ) {
let url = format!("{forge_url}/api/v1/notifications?all=false&limit={UNREAD_FETCH_LIMIT}"); // Fetch the page as raw JSON (`response_type::<String>`) instead of the
let resp = match client // crate's `Vec<NotificationThread>`: one merged-PR notification
.get(&url) // (`subject.state = "merged"`, unrepresentable in `StateType`) would
.header("Authorization", format!("token {token}")) // otherwise poison deserialization of the whole page. HTTP-level errors
.send() // still surface as `ForgejoError` exactly like the fully-typed call;
.await // `parse_notification` below does the per-item typed parse.
{ let query = NotifyGetListQuery {
Ok(r) => r, all: Some(false),
Err(e) => { ..NotifyGetListQuery::default()
debug!("forge_notify: poll request failed: {e}");
return;
}
}; };
let request = forge
.notify_get_list(query)
.page_size(u32::try_from(UNREAD_FETCH_LIMIT).unwrap_or(u32::MAX))
.response_type::<String>();
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() { let values: Vec<serde_json::Value> = match serde_json::from_str(&raw) {
debug!("forge_notify: poll status {}", resp.status());
return;
}
let notifications: Vec<serde_json::Value> = match resp.json().await {
Ok(v) => v, Ok(v) => v,
Err(e) => { Err(e) => {
warn!("forge_notify: response parse error: {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; return;
} }
debug!( debug!(
count = notifications.len(), count = values.len(),
"forge_notify: delivering notifications" "forge_notify: delivering notifications"
); );
let notifications: Vec<PolledNotification> =
values.into_iter().filter_map(parse_notification).collect();
// Tracks whether the dedupe cursor changed this poll (a new delivery // Tracks whether the dedupe cursor changed this poll (a new delivery
// recorded, or the prune below dropped now-read threads) so we only // recorded, or the prune below dropped now-read threads) so we only
// rewrite the on-disk cursor when there's something to persist. // rewrite the on-disk cursor when there's something to persist.
let mut cursor_dirty = false; let mut cursor_dirty = false;
for notif in &notifications { for notif in &notifications {
let Some(id) = notif["id"].as_u64() else { let Some(id) = notif.thread.id.and_then(|id| u64::try_from(id).ok()) else {
continue; continue;
}; };
@ -810,7 +842,7 @@ async fn poll_once(
// Skip it silently unless its `updated_at` advanced since the // Skip it silently unless its `updated_at` advanced since the
// version we last delivered a wake for (i.e. genuinely new // version we last delivered a wake for (i.e. genuinely new
// activity). See the `delivered` cursor note in `run`. // 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) { if !should_deliver(delivered, id, &updated_at) {
debug!(%id, "forge_notify: skipping (already delivered this version)"); debug!(%id, "forge_notify: skipping (already delivered this version)");
continue; continue;
@ -820,7 +852,7 @@ async fn poll_once(
// None means self-echo — mark read silently, no delivery. // None means self-echo — mark read silently, no delivery.
let Some(body) = body_opt else { let Some(body) = body_opt else {
mark_read(client, forge_url, token, id).await; mark_read(forge, id).await;
continue; continue;
}; };
@ -864,7 +896,7 @@ async fn poll_once(
// loud in tests/dev if a future pagination change silently breaks it. // loud in tests/dev if a future pagination change silently breaks it.
let current_ids: HashSet<u64> = notifications let current_ids: HashSet<u64> = notifications
.iter() .iter()
.filter_map(|n| n["id"].as_u64()) .filter_map(|n| n.thread.id.and_then(|id| u64::try_from(id).ok()))
.collect(); .collect();
let before_prune = delivered.len(); let before_prune = delivered.len();
delivered.retain(|id, _| current_ids.contains(id)); delivered.retain(|id, _| current_ids.contains(id));
@ -899,21 +931,31 @@ fn should_deliver(delivered: &HashMap<u64, String>, id: u64, updated_at: &str) -
/// deliberately left unread for the read-before-comment guard, and a failed /// 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 /// delivery is left unread + out of the dedupe cursor so it resurfaces on the
/// next poll tick. /// next poll tick.
async fn mark_read(client: &reqwest::Client, forge_url: &str, token: &str, id: u64) { async fn mark_read(forge: &Forgejo, id: u64) {
let mark_url = format!("{forge_url}/api/v1/notifications/threads/{id}"); let Ok(thread_id) = i64::try_from(id) else {
match client // Thread ids originate from `i64` in the poll parse, so an
.patch(&mark_url) // unrepresentable value can't actually reach here.
.header("Authorization", format!("token {token}")) return;
.send() };
.await // `to_status: None` → Forgejo's default transition (unread → read),
{ // matching the old bare PATCH. The 205 response body is the thread
Err(e) => { // 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::<String>();
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"); warn!(%id, error = ?e, "forge_notify: mark-read request failed — notification will resurface");
} }
Ok(r) if !r.status().is_success() => { Ok(Err(e)) => {
warn!(%id, status = %r.status(), "forge_notify: mark-read returned non-2xx — notification will resurface"); warn!(%id, error = %e, "forge_notify: mark-read returned non-2xx — notification will resurface");
} }
Ok(_) => { Ok(Ok(_)) => {
debug!(%id, "forge_notify: marked read"); debug!(%id, "forge_notify: marked read");
} }
} }
@ -1129,39 +1171,98 @@ mod tests {
} }
#[test] #[test]
fn parse_rfc3339_secs_handles_offsets_and_z() { fn parse_rfc3339_handles_offsets_and_z() {
// Same instant expressed three ways must parse equal. // Same instant expressed three ways must parse equal
let utc = parse_rfc3339_secs("2026-06-13T09:18:42Z").unwrap(); // (`OffsetDateTime` comparison is instant-based).
let plus2 = parse_rfc3339_secs("2026-06-13T11:18:42+02:00").unwrap(); let utc = parse_rfc3339("2026-06-13T09:18:42Z").unwrap();
let minus5 = parse_rfc3339_secs("2026-06-13T04:18:42-05:00").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, plus2);
assert_eq!(utc, minus5); assert_eq!(utc, minus5);
// Fractional seconds are skipped (second granularity). // Fractional seconds parse; second-granularity comparison holds.
assert_eq!(parse_rfc3339_secs("2026-06-13T09:18:42.512Z").unwrap(), utc); assert_eq!(
parse_rfc3339("2026-06-13T09:18:42.512Z")
.unwrap()
.unix_timestamp(),
utc.unix_timestamp(),
);
} }
#[test] #[test]
fn parse_rfc3339_secs_rejects_garbage() { fn parse_rfc3339_rejects_garbage() {
assert!(parse_rfc3339_secs("").is_none()); assert!(parse_rfc3339("").is_none());
assert!(parse_rfc3339_secs("not-a-date").is_none()); assert!(parse_rfc3339("not-a-date").is_none());
assert!(parse_rfc3339_secs("2026-06-13").is_none()); assert!(parse_rfc3339("2026-06-13").is_none());
} }
#[test] #[test]
fn notification_is_creation_flags_fresh_and_later_activity() { 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" }); 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. // Review hours later on the same PR → not a creation.
let later = serde_json::json!({ "updated_at": "2026-06-13T14:55:00+02:00" }); let later = parse_rfc3339("2026-06-13T14:55:00+02:00");
assert!(!notification_is_creation(&later, Some(&subject))); assert!(!notification_is_creation(later, Some(&subject)));
// Missing timestamps → default to creation (preserve prior behavior). // Missing timestamps → default to creation (preserve prior behavior).
let empty = serde_json::json!({}); assert!(notification_is_creation(None, None));
assert!(notification_is_creation(&empty, None)); assert!(notification_is_creation(None, Some(&subject)));
assert!(notification_is_creation(&empty, 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` /// 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> { fn state_change_meta(subject: serde_json::Value) -> NotifMeta<'static> {
NotifMeta { NotifMeta {
title: "subject title", title: "subject title",
notif_type: "Issue", subject_type: Some(NotifySubjectType::Issue),
html_url: "http://forge/issues/1", html_url: "http://forge/issues/1",
num: " #1".to_owned(), num: " #1".to_owned(),
repo: " [agents/x]".to_owned(), repo: " [agents/x]".to_owned(),
@ -1183,15 +1284,13 @@ mod tests {
fn state_change_drops_self_authored_creation() { fn state_change_drops_self_authored_creation() {
// No timestamps ⇒ treated as a creation; poster login == own_login. // No timestamps ⇒ treated as a creation; poster login == own_login.
let meta = state_change_meta(serde_json::json!({ "user": { "login": "damocles" } })); let meta = state_change_meta(serde_json::json!({ "user": { "login": "damocles" } }));
let notif = serde_json::json!({ "subject": { "state": "open" } }); assert!(format_state_change_notification(None, "open", &meta, "damocles").is_none());
assert!(format_state_change_notification(&notif, &meta, "damocles").is_none());
} }
#[test] #[test]
fn state_change_keeps_other_authored_creation() { fn state_change_keeps_other_authored_creation() {
let meta = state_change_meta(serde_json::json!({ "user": { "login": "someone-else" } })); 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(None, "open", &meta, "damocles").is_some());
assert!(format_state_change_notification(&notif, &meta, "damocles").is_some());
} }
#[test] #[test]
@ -1203,10 +1302,7 @@ mod tests {
"user": { "login": "damocles" }, "user": { "login": "damocles" },
"created_at": "2020-01-01T00:00:00Z", "created_at": "2020-01-01T00:00:00Z",
})); }));
let notif = serde_json::json!({ let event = parse_rfc3339("2026-06-22T16:00:00Z");
"subject": { "state": "closed" }, assert!(format_state_change_notification(event, "closed", &meta, "damocles").is_some());
"updated_at": "2026-06-22T16:00:00Z",
});
assert!(format_state_change_notification(&notif, &meta, "damocles").is_some());
} }
} }