Skip to main content

max / goingson

10.0 KB · 300 lines History Blame Raw
1 //! The wam adapter: GoingsOn's first problem source.
2 //!
3 //! wam (Whack-a-Mole) is the distributed ticket manager under `MNW/wam`. It
4 //! keeps its own SQLite store and its own painhours model; this adapter reads
5 //! `GET /tickets` over its authenticated HTTP API and hands the results to the
6 //! Problems repository.
7 //!
8 //! The two systems already agree on the ranking, because both compute painhours
9 //! from the shared `painhours` crate. So a wam ticket and an `/audit` finding
10 //! land in one list ordered by the same rule, which is the whole reason that
11 //! crate was extracted.
12
13 use async_trait::async_trait;
14 use chrono::{DateTime, Utc};
15 use goingson_core::NewProblem;
16 use serde::Deserialize;
17 use tracing::{debug, warn};
18
19 use super::{ProblemSource, PullError};
20
21 /// The adapter's stable name. Half of every provenance key it produces, so
22 /// changing it would orphan every problem already mirrored from wam.
23 pub const SOURCE: &str = "wam";
24
25 /// How long to wait on wam before giving up.
26 ///
27 /// Short on purpose: wam is a tailnet-local service, and the inbox should tell
28 /// the user it is unreachable rather than hang on a view refresh.
29 const TIMEOUT_SECS: u64 = 10;
30
31 /// One ticket as wam serializes it.
32 ///
33 /// Only the fields this adapter maps are declared; wam's `node_id` and its own
34 /// `source`/`source_ref` (which describe where *wam* got the ticket) are
35 /// deliberately ignored, since re-exporting them would confuse wam's provenance
36 /// with GoingsOn's.
37 #[derive(Debug, Deserialize)]
38 struct Ticket {
39 id: String,
40 title: String,
41 #[serde(default)]
42 body: Option<String>,
43 pain: u8,
44 scale: u8,
45 /// `open` | `in-progress` | `resolved` | `closed`.
46 status: String,
47 /// `system` | `request` | `task`.
48 #[serde(default)]
49 channel: Option<String>,
50 created_at: DateTime<Utc>,
51 updated_at: DateTime<Utc>,
52 }
53
54 #[derive(Debug, Deserialize)]
55 struct ListResponse {
56 data: Vec<Ticket>,
57 }
58
59 impl Ticket {
60 /// Whether wam considers this ticket finished.
61 ///
62 /// Both terminal states settle the mirrored problem. GoingsOn does not
63 /// distinguish "resolved" from "closed": either way it is no longer a live
64 /// problem here, and the triage state a human already set wins regardless
65 /// (the repository's upsert guards that).
66 fn resolved_upstream(&self) -> bool {
67 matches!(self.status.as_str(), "resolved" | "closed")
68 }
69
70 fn into_new_problem(self) -> NewProblem {
71 // Read the terminal state before the struct is taken apart below.
72 let resolved_upstream = self.resolved_upstream();
73 // wam's channel is a useful lens once these sit next to audit findings:
74 // a `system` ticket is an automated alert, a `task` one is a human note.
75 let mut tags = vec![SOURCE.to_string()];
76 if let Some(channel) = &self.channel {
77 tags.push(format!("channel:{channel}"));
78 }
79
80 NewProblem {
81 source: SOURCE.to_string(),
82 // wam's ticket id is stable and unique within wam, which is exactly
83 // what the provenance key needs.
84 source_ref: self.id,
85 title: self.title,
86 body: self.body.unwrap_or_default(),
87 pain: self.pain,
88 scale: self.scale,
89 project_id: None,
90 tags,
91 // wam's timestamps, not ours. Age drives the score, so a ticket
92 // filed weeks ago must not read as new because we pulled it today.
93 created_at: self.created_at,
94 updated_at: self.updated_at,
95 resolved_upstream,
96 }
97 }
98 }
99
100 /// Reads open problems from a wam instance.
101 pub struct WamSource {
102 base_url: String,
103 token: Option<String>,
104 client: reqwest::Client,
105 }
106
107 impl WamSource {
108 /// Build an adapter for a wam instance.
109 ///
110 /// `base_url` is wam's HTTP root (e.g. `http://fw13:7890`); a trailing slash
111 /// is tolerated. `token` is the shared bearer wam enforces when it was
112 /// started with one. Passing `None` is allowed because wam also runs
113 /// unauthenticated, and it warns loudly on its own side when it does; a
114 /// mismatched token surfaces here as a 401.
115 pub fn new(base_url: impl Into<String>, token: Option<String>) -> Self {
116 let base_url = base_url.into().trim_end_matches('/').to_string();
117 let client = reqwest::Client::builder()
118 .timeout(std::time::Duration::from_secs(TIMEOUT_SECS))
119 .build()
120 .unwrap_or_default();
121 Self {
122 base_url,
123 token,
124 client,
125 }
126 }
127 }
128
129 #[async_trait]
130 impl ProblemSource for WamSource {
131 fn name(&self) -> &str {
132 SOURCE
133 }
134
135 async fn pull(&self) -> Result<Vec<NewProblem>, PullError> {
136 if self.base_url.is_empty() {
137 return Err(PullError::NotConfigured(SOURCE));
138 }
139
140 // No status filter: terminal tickets have to come back too, or a ticket
141 // fixed in wam would sit open in the inbox forever with nothing to
142 // settle it.
143 let url = format!("{}/tickets", self.base_url);
144 let mut request = self.client.get(&url);
145 if let Some(token) = &self.token {
146 request = request.bearer_auth(token);
147 }
148
149 let response = request.send().await.map_err(|e| PullError::Unreachable {
150 name: SOURCE,
151 message: e.to_string(),
152 })?;
153
154 let status = response.status();
155 if !status.is_success() {
156 return Err(PullError::Unreachable {
157 name: SOURCE,
158 message: if status == reqwest::StatusCode::UNAUTHORIZED {
159 "unauthorized (check the wam token)".to_string()
160 } else {
161 format!("HTTP {status}")
162 },
163 });
164 }
165
166 let body: ListResponse = response.json().await.map_err(|e| PullError::Malformed {
167 name: SOURCE,
168 message: e.to_string(),
169 })?;
170
171 debug!(count = body.data.len(), "pulled tickets from wam");
172
173 // A ticket whose factors are out of range still ranks (the model clamps
174 // on use), so nothing is dropped here. Warn instead, since it means the
175 // producer is writing something wam itself did not intend.
176 for t in &body.data {
177 if !(1..=5).contains(&t.pain) || !(1..=5).contains(&t.scale) {
178 warn!(
179 ticket = %t.id,
180 pain = t.pain,
181 scale = t.scale,
182 "wam ticket has out-of-range painhours factors; clamping"
183 );
184 }
185 }
186
187 Ok(body
188 .data
189 .into_iter()
190 .map(Ticket::into_new_problem)
191 .collect())
192 }
193 }
194
195 #[cfg(test)]
196 mod tests {
197 use super::*;
198
199 fn ticket(status: &str) -> Ticket {
200 Ticket {
201 id: "abc123".into(),
202 title: "sync drops attachments".into(),
203 body: Some("seen on two devices".into()),
204 pain: 4,
205 scale: 3,
206 status: status.into(),
207 channel: Some("system".into()),
208 created_at: Utc::now() - chrono::Duration::days(30),
209 updated_at: Utc::now(),
210 }
211 }
212
213 #[test]
214 fn both_terminal_states_settle_the_mirror() {
215 assert!(!ticket("open").resolved_upstream());
216 assert!(!ticket("in-progress").resolved_upstream());
217 assert!(ticket("resolved").resolved_upstream());
218 assert!(ticket("closed").resolved_upstream());
219 }
220
221 #[test]
222 fn mapping_preserves_the_source_timestamps_and_factors() {
223 let t = ticket("open");
224 let created = t.created_at;
225 let p = t.into_new_problem();
226
227 assert_eq!(p.source, "wam");
228 assert_eq!(p.source_ref, "abc123");
229 assert_eq!(p.pain, 4);
230 assert_eq!(p.scale, 3);
231 // The ticket's own age, not the pull time: a month-old ticket must
232 // arrive already a month old, or it would restart its climb on import.
233 assert_eq!(p.created_at, created);
234 assert!(p.tags.contains(&"wam".to_string()));
235 assert!(p.tags.contains(&"channel:system".to_string()));
236 }
237
238 #[test]
239 fn a_channelless_ticket_still_maps() {
240 let mut t = ticket("open");
241 t.channel = None;
242 let p = t.into_new_problem();
243 assert_eq!(p.tags, vec!["wam".to_string()]);
244 }
245
246 #[test]
247 fn base_url_tolerates_a_trailing_slash() {
248 super::super::init_crypto_for_tests();
249 let a = WamSource::new("http://fw13:7890/", None);
250 let b = WamSource::new("http://fw13:7890", None);
251 assert_eq!(a.base_url, b.base_url);
252 }
253
254 #[tokio::test]
255 async fn an_empty_url_is_unconfigured_not_an_error() {
256 super::super::init_crypto_for_tests();
257 let source = WamSource::new("", None);
258 match source.pull().await {
259 Err(PullError::NotConfigured("wam")) => {}
260 other => panic!("expected NotConfigured, got {other:?}"),
261 }
262 }
263
264 #[test]
265 fn list_response_parses_wam_json() {
266 // The exact shape wam's GET /tickets returns.
267 let raw = r#"{
268 "data": [{
269 "id": "t-1",
270 "title": "disk filling",
271 "body": null,
272 "pain": 5,
273 "scale": 5,
274 "status": "open",
275 "channel": "system",
276 "node_id": "node-a",
277 "source": "pom",
278 "source_ref": "alert-9",
279 "created_at": "2026-07-01T12:00:00Z",
280 "updated_at": "2026-07-02T12:00:00Z",
281 "resolved_at": null
282 }],
283 "count": 1
284 }"#;
285
286 let parsed: ListResponse = serde_json::from_str(raw).expect("wam JSON must parse");
287 assert_eq!(parsed.data.len(), 1);
288
289 let p = parsed.data.into_iter().next().unwrap().into_new_problem();
290 assert_eq!(p.source_ref, "t-1");
291 assert_eq!(
292 p.body, "",
293 "a null body maps to empty, not the string 'null'"
294 );
295 // wam's own source/source_ref describe where WAM got it; they must not
296 // leak into GoingsOn's provenance.
297 assert_eq!(p.source, "wam");
298 }
299 }
300