Skip to main content

max / balanced_breakfast

10.3 KB · 283 lines History Blame Raw
1 use super::{
2 CIRCUIT_BREAKER_THRESHOLD, CreateFeed, DbFeed, ErrorCategory, FeedId, SqlitePool,
3 StructuredError, TIMESTAMP_FMT, Utc,
4 };
5
6 #[derive(Clone)]
7 /// Repository for feed subscription CRUD and fetch tracking
8 pub struct FeedsRepository {
9 pool: SqlitePool,
10 }
11
12 impl FeedsRepository {
13 /// Create a new feeds repository backed by the given pool.
14 #[tracing::instrument(skip_all)]
15 pub fn new(pool: SqlitePool) -> Self {
16 Self { pool }
17 }
18
19 /// Insert a new feed and return the created row.
20 /// New feeds default to `enabled = 1` (auto-fetch active) and
21 /// `created_at = updated_at` (no separate "first modified" vs "created" notion).
22 #[tracing::instrument(skip_all)]
23 pub async fn create(&self, input: CreateFeed) -> Result<DbFeed, sqlx::Error> {
24 let id = FeedId::new();
25 let now = Utc::now().format(TIMESTAMP_FMT).to_string();
26 // serde_json::Value is always serializable; unwrap is safe here.
27 let config = serde_json::to_string(&input.config).unwrap_or_else(|_| "{}".to_string());
28
29 sqlx::query_as(
30 r"
31 INSERT INTO feeds (id, busser_id, name, config, enabled, created_at, updated_at)
32 VALUES (?1, ?2, ?3, ?4, 1, ?5, ?5)
33 RETURNING *
34 ",
35 )
36 .bind(id)
37 .bind(input.busser_id.as_str())
38 .bind(&input.name)
39 .bind(&config)
40 .bind(&now)
41 .fetch_one(&self.pool)
42 .await
43 }
44
45 /// Look up a single feed by its ID. Returns `None` if not found.
46 #[tracing::instrument(skip_all)]
47 pub async fn get(&self, id: FeedId) -> Result<Option<DbFeed>, sqlx::Error> {
48 sqlx::query_as("SELECT * FROM feeds WHERE id = ?1")
49 .bind(id)
50 .fetch_optional(&self.pool)
51 .await
52 }
53
54 /// List all feeds belonging to a given busser, ordered by name.
55 #[tracing::instrument(skip_all)]
56 pub async fn get_by_busser(&self, busser_id: &str) -> Result<Vec<DbFeed>, sqlx::Error> {
57 sqlx::query_as("SELECT * FROM feeds WHERE busser_id = ?1 ORDER BY name")
58 .bind(busser_id)
59 .fetch_all(&self.pool)
60 .await
61 }
62
63 /// List only enabled feeds that are not circuit-broken, ordered by name.
64 #[tracing::instrument(skip_all)]
65 pub async fn list_enabled(&self) -> Result<Vec<DbFeed>, sqlx::Error> {
66 sqlx::query_as("SELECT * FROM feeds WHERE enabled = 1 AND circuit_broken = 0 ORDER BY name")
67 .fetch_all(&self.pool)
68 .await
69 }
70
71 /// List every feed (enabled or disabled), ordered by name.
72 #[tracing::instrument(skip_all)]
73 pub async fn list_all(&self) -> Result<Vec<DbFeed>, sqlx::Error> {
74 sqlx::query_as("SELECT * FROM feeds ORDER BY name")
75 .fetch_all(&self.pool)
76 .await
77 }
78
79 /// Update the `last_fetch` and `updated_at` timestamps to now.
80 #[tracing::instrument(skip_all)]
81 pub async fn update_last_fetch(&self, id: FeedId) -> Result<(), sqlx::Error> {
82 let now = Utc::now().format(TIMESTAMP_FMT).to_string();
83 sqlx::query("UPDATE feeds SET last_fetch = ?1, updated_at = ?1 WHERE id = ?2")
84 .bind(&now)
85 .bind(id)
86 .execute(&self.pool)
87 .await?;
88 Ok(())
89 }
90
91 /// Enable or disable a feed.
92 #[tracing::instrument(skip_all)]
93 pub async fn set_enabled(&self, id: FeedId, enabled: bool) -> Result<(), sqlx::Error> {
94 let now = Utc::now().format(TIMESTAMP_FMT).to_string();
95 sqlx::query("UPDATE feeds SET enabled = ?1, updated_at = ?2 WHERE id = ?3")
96 .bind(enabled)
97 .bind(&now)
98 .bind(id)
99 .execute(&self.pool)
100 .await?;
101 Ok(())
102 }
103
104 /// Record a successful fetch: reset failure counter, clear error, update timestamps.
105 #[tracing::instrument(skip_all)]
106 pub async fn record_fetch_success(&self, id: FeedId) -> Result<(), sqlx::Error> {
107 let now = Utc::now().format(TIMESTAMP_FMT).to_string();
108 sqlx::query(
109 "UPDATE feeds SET consecutive_failures = 0, last_error = NULL, \
110 last_success_at = ?1, last_fetch = ?1, updated_at = ?1 WHERE id = ?2",
111 )
112 .bind(&now)
113 .bind(id)
114 .execute(&self.pool)
115 .await?;
116 Ok(())
117 }
118
119 /// Record a fetch failure: increment counter, store error, update timestamp.
120 ///
121 /// Returns `true` if the circuit breaker tripped (i.e. the feed just crossed
122 /// the [`CIRCUIT_BREAKER_THRESHOLD`] and was marked `circuit_broken = 1`).
123 #[tracing::instrument(skip_all)]
124 pub async fn record_fetch_failure(&self, id: FeedId, error: &str) -> Result<bool, sqlx::Error> {
125 let now = Utc::now().format(TIMESTAMP_FMT).to_string();
126 sqlx::query(
127 "UPDATE feeds SET consecutive_failures = consecutive_failures + 1, \
128 last_error = ?1, last_fetch = ?2, updated_at = ?2 WHERE id = ?3",
129 )
130 .bind(error)
131 .bind(&now)
132 .bind(id)
133 .execute(&self.pool)
134 .await?;
135
136 // Check if we just crossed the threshold
137 let feed = self.get(id).await?;
138 if let Some(feed) = feed
139 && !feed.circuit_broken
140 && feed.consecutive_failures >= CIRCUIT_BREAKER_THRESHOLD
141 {
142 self.set_circuit_broken(id, true).await?;
143 return Ok(true);
144 }
145 Ok(false)
146 }
147
148 /// Record a structured fetch failure with category-aware behavior:
149 ///
150 /// - `RateLimited`: store error JSON, do NOT increment `consecutive_failures`.
151 /// - `Auth` / `Config`: increment + immediately set `circuit_broken = 1`.
152 /// - `Transient` / `Parse` / `Unknown`: increment normally (existing behavior).
153 ///
154 /// Returns `true` if the circuit breaker tripped.
155 #[tracing::instrument(skip_all)]
156 pub async fn record_fetch_failure_structured(
157 &self,
158 id: FeedId,
159 error: &StructuredError,
160 ) -> Result<bool, sqlx::Error> {
161 let now = Utc::now().format(TIMESTAMP_FMT).to_string();
162 let error_json = error.to_json();
163
164 match error.category {
165 ErrorCategory::RateLimited => {
166 // Store error but don't increment failure counter
167 sqlx::query(
168 "UPDATE feeds SET last_error = ?1, last_fetch = ?2, updated_at = ?2 WHERE id = ?3",
169 )
170 .bind(&error_json)
171 .bind(&now)
172 .bind(id)
173 .execute(&self.pool)
174 .await?;
175 Ok(false)
176 }
177 ErrorCategory::Auth | ErrorCategory::Config => {
178 // Increment + immediate circuit break
179 sqlx::query(
180 "UPDATE feeds SET consecutive_failures = consecutive_failures + 1, \
181 last_error = ?1, last_fetch = ?2, updated_at = ?2 WHERE id = ?3",
182 )
183 .bind(&error_json)
184 .bind(&now)
185 .bind(id)
186 .execute(&self.pool)
187 .await?;
188 self.set_circuit_broken(id, true).await?;
189 Ok(true)
190 }
191 ErrorCategory::Transient | ErrorCategory::Parse | ErrorCategory::Unknown => {
192 // Normal behavior: increment and check threshold
193 sqlx::query(
194 "UPDATE feeds SET consecutive_failures = consecutive_failures + 1, \
195 last_error = ?1, last_fetch = ?2, updated_at = ?2 WHERE id = ?3",
196 )
197 .bind(&error_json)
198 .bind(&now)
199 .bind(id)
200 .execute(&self.pool)
201 .await?;
202
203 let feed = self.get(id).await?;
204 if let Some(feed) = feed
205 && !feed.circuit_broken
206 && feed.consecutive_failures >= CIRCUIT_BREAKER_THRESHOLD
207 {
208 self.set_circuit_broken(id, true).await?;
209 return Ok(true);
210 }
211 Ok(false)
212 }
213 }
214 }
215
216 /// Mark a feed as circuit-broken (or clear the circuit breaker).
217 #[tracing::instrument(skip_all)]
218 pub async fn set_circuit_broken(&self, id: FeedId, broken: bool) -> Result<(), sqlx::Error> {
219 let now = Utc::now().format(TIMESTAMP_FMT).to_string();
220 sqlx::query("UPDATE feeds SET circuit_broken = ?1, updated_at = ?2 WHERE id = ?3")
221 .bind(broken)
222 .bind(&now)
223 .bind(id)
224 .execute(&self.pool)
225 .await?;
226 Ok(())
227 }
228
229 /// Reset the circuit breaker on a feed: clear `circuit_broken`, reset
230 /// `consecutive_failures` to 0, and clear `last_error`.
231 ///
232 /// Called when a user manually triggers a fetch for a circuit-broken feed.
233 #[tracing::instrument(skip_all)]
234 pub async fn reset_circuit_breaker(&self, id: FeedId) -> Result<(), sqlx::Error> {
235 let now = Utc::now().format(TIMESTAMP_FMT).to_string();
236 sqlx::query(
237 "UPDATE feeds SET circuit_broken = 0, consecutive_failures = 0, \
238 last_error = NULL, updated_at = ?1 WHERE id = ?2",
239 )
240 .bind(&now)
241 .bind(id)
242 .execute(&self.pool)
243 .await?;
244 Ok(())
245 }
246
247 /// Update a feed's config JSON string.
248 #[tracing::instrument(skip_all)]
249 pub async fn update_config(&self, id: FeedId, config: &str) -> Result<(), sqlx::Error> {
250 let now = Utc::now().format(TIMESTAMP_FMT).to_string();
251 sqlx::query("UPDATE feeds SET config = ?1, updated_at = ?2 WHERE id = ?3")
252 .bind(config)
253 .bind(&now)
254 .bind(id)
255 .execute(&self.pool)
256 .await?;
257 Ok(())
258 }
259
260 /// Update a feed's display name.
261 #[tracing::instrument(skip_all)]
262 pub async fn update_name(&self, id: FeedId, name: &str) -> Result<(), sqlx::Error> {
263 let now = Utc::now().format(TIMESTAMP_FMT).to_string();
264 sqlx::query("UPDATE feeds SET name = ?1, updated_at = ?2 WHERE id = ?3")
265 .bind(name)
266 .bind(&now)
267 .bind(id)
268 .execute(&self.pool)
269 .await?;
270 Ok(())
271 }
272
273 /// Delete a feed by ID.
274 #[tracing::instrument(skip_all)]
275 pub async fn delete(&self, id: FeedId) -> Result<(), sqlx::Error> {
276 sqlx::query("DELETE FROM feeds WHERE id = ?1")
277 .bind(id)
278 .execute(&self.pool)
279 .await?;
280 Ok(())
281 }
282 }
283