Skip to main content

max / makenotwork

5.4 KB · 179 lines History Blame Raw
1 //! Alert history (for cooldown checks) and the pending-alert retry queue that
2 //! backs delivery to Postmark, WAM, and MNW.
3
4 use super::{Result, SqlitePool};
5 use tracing::instrument;
6
7 #[derive(Debug, sqlx::FromRow)]
8 pub struct AlertRow {
9 pub id: i64,
10 pub target: String,
11 pub alert_type: String,
12 pub from_status: Option<String>,
13 pub to_status: Option<String>,
14 pub sent_at: String,
15 pub error: Option<String>,
16 }
17
18 #[instrument(skip_all)]
19 pub async fn insert_alert(
20 pool: &SqlitePool,
21 target: &str,
22 alert_type: &str,
23 from_status: Option<&str>,
24 to_status: Option<&str>,
25 error: Option<&str>,
26 ) -> Result<i64> {
27 let now = chrono::Utc::now().to_rfc3339();
28 let result = sqlx::query(
29 "INSERT INTO alerts (target, alert_type, from_status, to_status, sent_at, error)
30 VALUES (?, ?, ?, ?, ?, ?)",
31 )
32 .bind(target)
33 .bind(alert_type)
34 .bind(from_status)
35 .bind(to_status)
36 .bind(&now)
37 .bind(error)
38 .execute(pool)
39 .await?;
40 Ok(result.last_insert_rowid())
41 }
42
43 #[instrument(skip_all)]
44 pub async fn get_latest_alert_for_target(
45 pool: &SqlitePool,
46 target: &str,
47 ) -> Result<Option<AlertRow>> {
48 Ok(sqlx::query_as::<_, AlertRow>(
49 "SELECT id, target, alert_type, from_status, to_status, sent_at, error
50 FROM alerts WHERE target = ? AND alert_type NOT LIKE '%recovery%'
51 ORDER BY id DESC LIMIT 1",
52 )
53 .bind(target)
54 .fetch_optional(pool)
55 .await?)
56 }
57
58 /// The most recent alert row for a target of a specific type-set. Unlike
59 /// [`get_latest_alert_for_target`] (which excludes recoveries), this can target
60 /// recovery rows so recoveries get their own cooldown, and so a recorded recovery
61 /// can clear a prior failure's cooldown (fail→recover→fail). `like` is a SQL
62 /// LIKE pattern matched against `alert_type` (e.g. `'%recovery%'`).
63 #[instrument(skip_all)]
64 pub async fn get_latest_alert_matching(
65 pool: &SqlitePool,
66 target: &str,
67 like: &str,
68 ) -> Result<Option<AlertRow>> {
69 Ok(sqlx::query_as::<_, AlertRow>(
70 "SELECT id, target, alert_type, from_status, to_status, sent_at, error
71 FROM alerts WHERE target = ? AND alert_type LIKE ?
72 ORDER BY id DESC LIMIT 1",
73 )
74 .bind(target)
75 .bind(like)
76 .fetch_optional(pool)
77 .await?)
78 }
79
80 /// A delivery that failed and must be retried, so a transient WAM/Postmark error
81 /// at a status transition can't silence the whole outage (the transition fires
82 /// once; this queue is the durable state that outlives it).
83 #[derive(Debug, Clone, sqlx::FromRow)]
84 pub struct PendingAlertRow {
85 pub id: i64,
86 pub alert_key: String,
87 pub category: String,
88 pub channel: String,
89 pub subject: String,
90 pub body: String,
91 pub priority: Option<String>,
92 pub source: Option<String>,
93 pub source_ref: Option<String>,
94 pub from_status: Option<String>,
95 pub to_status: Option<String>,
96 pub error: Option<String>,
97 pub attempts: i64,
98 }
99
100 /// Fields needed to enqueue an undelivered alert for retry.
101 #[derive(Debug, Clone)]
102 pub struct NewPendingAlert<'a> {
103 pub alert_key: &'a str,
104 pub category: &'a str,
105 pub channel: &'a str,
106 pub subject: &'a str,
107 pub body: &'a str,
108 pub priority: Option<&'a str>,
109 pub source: Option<&'a str>,
110 pub source_ref: Option<&'a str>,
111 pub from_status: Option<&'a str>,
112 pub to_status: Option<&'a str>,
113 pub error: Option<&'a str>,
114 }
115
116 #[instrument(skip_all)]
117 pub async fn enqueue_pending_alert(pool: &SqlitePool, a: &NewPendingAlert<'_>) -> Result<()> {
118 let now = chrono::Utc::now().to_rfc3339();
119 sqlx::query(
120 "INSERT INTO pending_alerts
121 (alert_key, category, channel, subject, body, priority, source, source_ref,
122 from_status, to_status, error, attempts, created_at, next_retry_at)
123 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, ?, ?)",
124 )
125 .bind(a.alert_key)
126 .bind(a.category)
127 .bind(a.channel)
128 .bind(a.subject)
129 .bind(a.body)
130 .bind(a.priority)
131 .bind(a.source)
132 .bind(a.source_ref)
133 .bind(a.from_status)
134 .bind(a.to_status)
135 .bind(a.error)
136 .bind(&now)
137 .bind(&now) // due immediately for the first retry
138 .execute(pool)
139 .await?;
140 Ok(())
141 }
142
143 /// Pending alerts whose `next_retry_at` is due (<= now), oldest first.
144 #[instrument(skip_all)]
145 pub async fn due_pending_alerts(pool: &SqlitePool, limit: i64) -> Result<Vec<PendingAlertRow>> {
146 let now = chrono::Utc::now().to_rfc3339();
147 Ok(sqlx::query_as::<_, PendingAlertRow>(
148 "SELECT id, alert_key, category, channel, subject, body, priority, source, source_ref,
149 from_status, to_status, error, attempts
150 FROM pending_alerts WHERE next_retry_at <= ? ORDER BY id ASC LIMIT ?",
151 )
152 .bind(&now)
153 .bind(limit)
154 .fetch_all(pool)
155 .await?)
156 }
157
158 #[instrument(skip_all)]
159 pub async fn delete_pending_alert(pool: &SqlitePool, id: i64) -> Result<()> {
160 sqlx::query("DELETE FROM pending_alerts WHERE id = ?")
161 .bind(id)
162 .execute(pool)
163 .await?;
164 Ok(())
165 }
166
167 /// Record a failed retry: bump the attempt count and push `next_retry_at` out.
168 #[instrument(skip_all)]
169 pub async fn bump_pending_alert(pool: &SqlitePool, id: i64, next_retry_at: &str) -> Result<()> {
170 sqlx::query(
171 "UPDATE pending_alerts SET attempts = attempts + 1, next_retry_at = ? WHERE id = ?",
172 )
173 .bind(next_retry_at)
174 .bind(id)
175 .execute(pool)
176 .await?;
177 Ok(())
178 }
179