Skip to main content

max / balanced_breakfast

Split bb-db repository god-module into per-entity files The single 2872-line repository.rs bundled seven independent repositories (Feeds, Items, Tags, State, Config, QueryFeeds, Bookmarks). Split into a repository/ directory: mod.rs holds the shared imports, sanitize_fts_query, and consts; one file per repository; tests move to repository/tests.rs. Public API preserved via `pub use <sub>::*`. Also collapse two pre-existing collapsible-if nits in the feeds repository. No behavior change. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-13 01:39 UTC
Commit: 5484fa0cdebb70972ac41caa0c8bcf1a17add74a
Parent: b117459
10 files changed, +1875 insertions, -500 deletions
D crates/bb-db/src/repository.rs -500
@@ -1,2872 +0,0 @@
1 - //! Database repositories for feeds, items, and busser state.
2 - //!
3 - //! Each repository wraps an SQLite connection pool and provides typed
4 - //! CRUD operations for a single table.
5 -
6 - use chrono::{DateTime, Utc};
7 - use sqlx::SqlitePool;
8 -
9 - use bb_interface::{ErrorCategory, StructuredError};
10 -
11 - use crate::id_types::{BookmarkId, BusserStateId, FeedId, ItemId, QueryFeedId};
12 - use crate::models::*;
13 - use crate::TIMESTAMP_FMT;
14 -
15 - /// Sanitize a user-provided search string for FTS5 MATCH syntax.
16 - ///
17 - /// Wraps each word in double quotes to prevent FTS5 syntax injection
18 - /// (e.g. a user typing `AND`, `OR`, `NOT`, `NEAR`, or `NEAR/N` won't be
19 - /// interpreted as FTS5 operators). Words are joined with spaces (implicit AND).
20 - ///
21 - /// Inside quoted strings FTS5 still honours two special characters:
22 - /// - `*` at the end of the last token triggers prefix matching
23 - /// - `^` at the start of the first token triggers beginning-of-column matching
24 - ///
25 - /// We strip those so user input like `^hello` or `world*` is treated literally.
26 - /// The `column:` prefix syntax (e.g. `title:word`) is neutralised by the quoting
27 - /// itself — colons inside double-quoted strings are treated as literal characters.
28 - fn sanitize_fts_query(query: &str) -> String {
29 - query
30 - .split_whitespace()
31 - .map(|word| {
32 - // Escape embedded double quotes (FTS5 uses "" to represent a literal ")
33 - let escaped = word.replace('"', "\"\"");
34 - // Strip `^` prefix and `*` suffix — both are special inside FTS5 quotes
35 - let escaped = escaped.trim_start_matches('^');
36 - let escaped = escaped.trim_end_matches('*');
37 - format!("\"{}\"", escaped)
38 - })
39 - // Drop tokens that became empty after stripping (e.g. bare `^`, `*`, `^*`)
40 - .filter(|token| token != "\"\"")
41 - .collect::<Vec<_>>()
42 - .join(" ")
43 - }
44 -
45 - /// Maximum allowed length for a search query string.
46 - ///
47 - /// Queries longer than this are rejected early (returning an empty result set)
48 - /// to prevent excessively large FTS5 MATCH expressions from consuming memory
49 - /// or CPU in SQLite.
50 - const MAX_SEARCH_QUERY_LENGTH: usize = 500;
51 -
52 - /// Number of consecutive failures before a feed is automatically disabled.
53 - ///
54 - /// Once a feed accumulates this many failures without a successful fetch,
55 - /// the circuit breaker trips: the feed is marked `circuit_broken = 1` and
56 - /// excluded from automatic fetch scheduling until manually reset.
57 - pub const CIRCUIT_BREAKER_THRESHOLD: i64 = 10;
58 -
59 - #[derive(Clone)]
60 - /// Repository for feed subscription CRUD and fetch tracking
61 - pub struct FeedsRepository {
62 - pool: SqlitePool,
63 - }
64 -
65 - impl FeedsRepository {
66 - /// Create a new feeds repository backed by the given pool.
67 - #[tracing::instrument(skip_all)]
68 - pub fn new(pool: SqlitePool) -> Self {
69 - Self { pool }
70 - }
71 -
72 - /// Insert a new feed and return the created row.
73 - /// New feeds default to `enabled = 1` (auto-fetch active) and
74 - /// `created_at = updated_at` (no separate "first modified" vs "created" notion).
75 - #[tracing::instrument(skip_all)]
76 - pub async fn create(&self, input: CreateFeed) -> Result<DbFeed, sqlx::Error> {
77 - let id = FeedId::new();
78 - let now = Utc::now().format(TIMESTAMP_FMT).to_string();
79 - // serde_json::Value is always serializable; unwrap is safe here.
80 - let config = serde_json::to_string(&input.config).unwrap_or_else(|_| "{}".to_string());
81 -
82 - sqlx::query_as(
83 - r#"
84 - INSERT INTO feeds (id, busser_id, name, config, enabled, created_at, updated_at)
85 - VALUES (?1, ?2, ?3, ?4, 1, ?5, ?5)
86 - RETURNING *
87 - "#,
88 - )
89 - .bind(id)
90 - .bind(input.busser_id.as_str())
91 - .bind(&input.name)
92 - .bind(&config)
93 - .bind(&now)
94 - .fetch_one(&self.pool)
95 - .await
96 - }
97 -
98 - /// Look up a single feed by its ID. Returns `None` if not found.
99 - #[tracing::instrument(skip_all)]
100 - pub async fn get(&self, id: FeedId) -> Result<Option<DbFeed>, sqlx::Error> {
101 - sqlx::query_as("SELECT * FROM feeds WHERE id = ?1")
102 - .bind(id)
103 - .fetch_optional(&self.pool)
104 - .await
105 - }
106 -
107 - /// List all feeds belonging to a given busser, ordered by name.
108 - #[tracing::instrument(skip_all)]
109 - pub async fn get_by_busser(&self, busser_id: &str) -> Result<Vec<DbFeed>, sqlx::Error> {
110 - sqlx::query_as("SELECT * FROM feeds WHERE busser_id = ?1 ORDER BY name")
111 - .bind(busser_id)
112 - .fetch_all(&self.pool)
113 - .await
114 - }
115 -
116 - /// List only enabled feeds that are not circuit-broken, ordered by name.
117 - #[tracing::instrument(skip_all)]
118 - pub async fn list_enabled(&self) -> Result<Vec<DbFeed>, sqlx::Error> {
119 - sqlx::query_as(
120 - "SELECT * FROM feeds WHERE enabled = 1 AND circuit_broken = 0 ORDER BY name",
121 - )
122 - .fetch_all(&self.pool)
123 - .await
124 - }
125 -
126 - /// List every feed (enabled or disabled), ordered by name.
127 - #[tracing::instrument(skip_all)]
128 - pub async fn list_all(&self) -> Result<Vec<DbFeed>, sqlx::Error> {
129 - sqlx::query_as("SELECT * FROM feeds ORDER BY name")
130 - .fetch_all(&self.pool)
131 - .await
132 - }
133 -
134 - /// Update the `last_fetch` and `updated_at` timestamps to now.
135 - #[tracing::instrument(skip_all)]
136 - pub async fn update_last_fetch(&self, id: FeedId) -> Result<(), sqlx::Error> {
137 - let now = Utc::now().format(TIMESTAMP_FMT).to_string();
138 - sqlx::query("UPDATE feeds SET last_fetch = ?1, updated_at = ?1 WHERE id = ?2")
139 - .bind(&now)
140 - .bind(id)
141 - .execute(&self.pool)
142 - .await?;
143 - Ok(())
144 - }
145 -
146 - /// Enable or disable a feed.
147 - #[tracing::instrument(skip_all)]
148 - pub async fn set_enabled(&self, id: FeedId, enabled: bool) -> Result<(), sqlx::Error> {
149 - let now = Utc::now().format(TIMESTAMP_FMT).to_string();
150 - sqlx::query("UPDATE feeds SET enabled = ?1, updated_at = ?2 WHERE id = ?3")
151 - .bind(enabled)
152 - .bind(&now)
153 - .bind(id)
154 - .execute(&self.pool)
155 - .await?;
156 - Ok(())
157 - }
158 -
159 - /// Record a successful fetch: reset failure counter, clear error, update timestamps.
160 - #[tracing::instrument(skip_all)]
161 - pub async fn record_fetch_success(&self, id: FeedId) -> Result<(), sqlx::Error> {
162 - let now = Utc::now().format(TIMESTAMP_FMT).to_string();
163 - sqlx::query(
164 - "UPDATE feeds SET consecutive_failures = 0, last_error = NULL, \
165 - last_success_at = ?1, last_fetch = ?1, updated_at = ?1 WHERE id = ?2",
166 - )
167 - .bind(&now)
168 - .bind(id)
169 - .execute(&self.pool)
170 - .await?;
171 - Ok(())
172 - }
173 -
174 - /// Record a fetch failure: increment counter, store error, update timestamp.
175 - ///
176 - /// Returns `true` if the circuit breaker tripped (i.e. the feed just crossed
177 - /// the [`CIRCUIT_BREAKER_THRESHOLD`] and was marked `circuit_broken = 1`).
178 - #[tracing::instrument(skip_all)]
179 - pub async fn record_fetch_failure(
180 - &self,
181 - id: FeedId,
182 - error: &str,
183 - ) -> Result<bool, sqlx::Error> {
184 - let now = Utc::now().format(TIMESTAMP_FMT).to_string();
185 - sqlx::query(
186 - "UPDATE feeds SET consecutive_failures = consecutive_failures + 1, \
187 - last_error = ?1, last_fetch = ?2, updated_at = ?2 WHERE id = ?3",
188 - )
189 - .bind(error)
190 - .bind(&now)
191 - .bind(id)
192 - .execute(&self.pool)
193 - .await?;
194 -
195 - // Check if we just crossed the threshold
196 - let feed = self.get(id).await?;
197 - if let Some(feed) = feed {
198 - if !feed.circuit_broken
199 - && feed.consecutive_failures >= CIRCUIT_BREAKER_THRESHOLD
200 - {
201 - self.set_circuit_broken(id, true).await?;
202 - return Ok(true);
203 - }
204 - }
205 - Ok(false)
206 - }
207 -
208 - /// Record a structured fetch failure with category-aware behavior:
209 - ///
210 - /// - `RateLimited` — store error JSON, do NOT increment `consecutive_failures`.
211 - /// - `Auth` / `Config` — increment + immediately set `circuit_broken = 1`.
212 - /// - `Transient` / `Parse` / `Unknown` — increment normally (existing behavior).
213 - ///
214 - /// Returns `true` if the circuit breaker tripped.
215 - #[tracing::instrument(skip_all)]
216 - pub async fn record_fetch_failure_structured(
217 - &self,
218 - id: FeedId,
219 - error: &StructuredError,
220 - ) -> Result<bool, sqlx::Error> {
221 - let now = Utc::now().format(TIMESTAMP_FMT).to_string();
222 - let error_json = error.to_json();
223 -
224 - match error.category {
225 - ErrorCategory::RateLimited => {
226 - // Store error but don't increment failure counter
227 - sqlx::query(
228 - "UPDATE feeds SET last_error = ?1, last_fetch = ?2, updated_at = ?2 WHERE id = ?3",
229 - )
230 - .bind(&error_json)
231 - .bind(&now)
232 - .bind(id)
233 - .execute(&self.pool)
234 - .await?;
235 - Ok(false)
236 - }
237 - ErrorCategory::Auth | ErrorCategory::Config => {
238 - // Increment + immediate circuit break
239 - sqlx::query(
240 - "UPDATE feeds SET consecutive_failures = consecutive_failures + 1, \
241 - last_error = ?1, last_fetch = ?2, updated_at = ?2 WHERE id = ?3",
242 - )
243 - .bind(&error_json)
244 - .bind(&now)
245 - .bind(id)
246 - .execute(&self.pool)
247 - .await?;
248 - self.set_circuit_broken(id, true).await?;
249 - Ok(true)
250 - }
251 - ErrorCategory::Transient | ErrorCategory::Parse | ErrorCategory::Unknown => {
252 - // Normal behavior: increment and check threshold
253 - sqlx::query(
254 - "UPDATE feeds SET consecutive_failures = consecutive_failures + 1, \
255 - last_error = ?1, last_fetch = ?2, updated_at = ?2 WHERE id = ?3",
256 - )
257 - .bind(&error_json)
258 - .bind(&now)
259 - .bind(id)
260 - .execute(&self.pool)
261 - .await?;
262 -
263 - let feed = self.get(id).await?;
264 - if let Some(feed) = feed {
265 - if !feed.circuit_broken
266 - && feed.consecutive_failures >= CIRCUIT_BREAKER_THRESHOLD
267 - {
268 - self.set_circuit_broken(id, true).await?;
269 - return Ok(true);
270 - }
271 - }
272 - Ok(false)
273 - }
274 - }
275 - }
276 -
277 - /// Mark a feed as circuit-broken (or clear the circuit breaker).
278 - #[tracing::instrument(skip_all)]
279 - pub async fn set_circuit_broken(
280 - &self,
281 - id: FeedId,
282 - broken: bool,
283 - ) -> Result<(), sqlx::Error> {
284 - let now = Utc::now().format(TIMESTAMP_FMT).to_string();
285 - sqlx::query(
286 - "UPDATE feeds SET circuit_broken = ?1, updated_at = ?2 WHERE id = ?3",
287 - )
288 - .bind(broken)
289 - .bind(&now)
290 - .bind(id)
291 - .execute(&self.pool)
292 - .await?;
293 - Ok(())
294 - }
295 -
296 - /// Reset the circuit breaker on a feed: clear `circuit_broken`, reset
297 - /// `consecutive_failures` to 0, and clear `last_error`.
298 - ///
299 - /// Called when a user manually triggers a fetch for a circuit-broken feed.
300 - #[tracing::instrument(skip_all)]
301 - pub async fn reset_circuit_breaker(&self, id: FeedId) -> Result<(), sqlx::Error> {
302 - let now = Utc::now().format(TIMESTAMP_FMT).to_string();
303 - sqlx::query(
304 - "UPDATE feeds SET circuit_broken = 0, consecutive_failures = 0, \
305 - last_error = NULL, updated_at = ?1 WHERE id = ?2",
306 - )
307 - .bind(&now)
308 - .bind(id)
309 - .execute(&self.pool)
310 - .await?;
311 - Ok(())
312 - }
313 -
314 - /// Update a feed's config JSON string.
315 - #[tracing::instrument(skip_all)]
316 - pub async fn update_config(&self, id: FeedId, config: &str) -> Result<(), sqlx::Error> {
317 - let now = Utc::now().format(TIMESTAMP_FMT).to_string();
318 - sqlx::query("UPDATE feeds SET config = ?1, updated_at = ?2 WHERE id = ?3")
319 - .bind(config)
320 - .bind(&now)
321 - .bind(id)
322 - .execute(&self.pool)
323 - .await?;
324 - Ok(())
325 - }
326 -
327 - /// Update a feed's display name.
328 - #[tracing::instrument(skip_all)]
329 - pub async fn update_name(&self, id: FeedId, name: &str) -> Result<(), sqlx::Error> {
330 - let now = Utc::now().format(TIMESTAMP_FMT).to_string();
331 - sqlx::query("UPDATE feeds SET name = ?1, updated_at = ?2 WHERE id = ?3")
332 - .bind(name)
333 - .bind(&now)
334 - .bind(id)
335 - .execute(&self.pool)
336 - .await?;
337 - Ok(())
338 - }
339 -
340 - /// Delete a feed by ID.
341 - #[tracing::instrument(skip_all)]
342 - pub async fn delete(&self, id: FeedId) -> Result<(), sqlx::Error> {
343 - sqlx::query("DELETE FROM feeds WHERE id = ?1")
344 - .bind(id)
345 - .execute(&self.pool)
346 - .await?;
347 - Ok(())
348 - }
349 - }
350 -
351 - #[derive(Clone)]
352 - /// Repository for feed item CRUD, read/star toggling, and paginated listing
353 - pub struct ItemsRepository {
354 - pool: SqlitePool,
355 - }
356 -
357 - impl ItemsRepository {
358 - /// Create a new items repository backed by the given pool.
359 - #[tracing::instrument(skip_all)]
360 - pub fn new(pool: SqlitePool) -> Self {
361 - Self { pool }
362 - }
363 -
364 - /// Insert a feed item or update it if `external_id` already exists.
365 - ///
366 - /// The ON CONFLICT clause deliberately preserves user state (`is_read`,
367 - /// `is_starred`) — these are never overwritten by a re-fetch. Content fields
368 - /// (author, text, body, etc.) are updated in case the source edited the post.
369 - #[tracing::instrument(skip_all)]
370 - pub async fn upsert(&self, input: CreateFeedItem) -> Result<DbFeedItem, sqlx::Error> {
371 - let id = ItemId::new();
372 - let now = Utc::now().format(TIMESTAMP_FMT).to_string();
373 - let published = input.published_at.format(TIMESTAMP_FMT).to_string();
374 - // Vec<String> is always serializable; unwrap is safe here.
375 - let media = serde_json::to_string(&input.media).unwrap_or_else(|_| "[]".to_string());
376 - let tags = serde_json::to_string(&input.tags).unwrap_or_else(|_| "[]".to_string());
377 - let actions =
378 - serde_json::to_string(&input.actions).unwrap_or_else(|_| "[]".to_string());
379 -
380 - sqlx::query_as(
381 - r#"
382 - INSERT INTO feed_items (
383 - id, external_id, feed_id, busser_id,
384 - bite_author, bite_text, bite_secondary, bite_indicator,
385 - title, body, url, media, actions,
386 - published_at, fetched_at, source_name, score, tags,
387 - is_read, is_starred, created_at, updated_at
388 - )
389 - VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18, 0, 0, ?19, ?19)
390 - ON CONFLICT (external_id) DO UPDATE SET
391 - bite_author = EXCLUDED.bite_author,
392 - bite_text = EXCLUDED.bite_text,
393 - bite_secondary = EXCLUDED.bite_secondary,
394 - bite_indicator = EXCLUDED.bite_indicator,
395 - title = EXCLUDED.title,
396 - body = EXCLUDED.body,
397 - url = EXCLUDED.url,
398 - media = EXCLUDED.media,
399 - actions = EXCLUDED.actions,
400 - score = EXCLUDED.score,
401 - tags = EXCLUDED.tags,
402 - updated_at = EXCLUDED.updated_at
403 - RETURNING *
404 - "#,
405 - )
406 - .bind(id)
407 - .bind(&input.external_id)
408 - .bind(input.feed_id)
409 - .bind(input.busser_id.as_str())
410 - .bind(&input.bite_author)
411 - .bind(&input.bite_text)
412 - .bind(&input.bite_secondary)
413 - .bind(&input.bite_indicator)
414 - .bind(&input.title)
415 - .bind(&input.body)
416 - .bind(&input.url)
417 - .bind(&media)
418 - .bind(&actions)
419 - .bind(&published)
420 - .bind(&now)
421 - .bind(&input.source_name)
422 - .bind(input.score)
423 - .bind(&tags)
424 - .bind(&now)
425 - .fetch_one(&self.pool)
426 - .await
427 - }
428 -
429 - /// Look up a feed item by its internal ID.
430 - #[tracing::instrument(skip_all)]
431 - pub async fn get(&self, id: ItemId) -> Result<Option<DbFeedItem>, sqlx::Error> {
432 - sqlx::query_as("SELECT * FROM feed_items WHERE id = ?1")
433 - .bind(id)
434 - .fetch_optional(&self.pool)
435 - .await
436 - }
437 -
438 - /// Look up a feed item by the `external_id` produced by the busser.
439 - #[tracing::instrument(skip_all)]
440 - pub async fn get_by_external_id(
441 - &self,
442 - external_id: &str,
443 - ) -> Result<Option<DbFeedItem>, sqlx::Error> {
444 - sqlx::query_as("SELECT * FROM feed_items WHERE external_id = ?1")
445 - .bind(external_id)
446 - .fetch_optional(&self.pool)
447 - .await
448 - }
449 -
450 - /// List all items ordered by `published_at` descending, with pagination.
451 - /// Sorted by publish date (not fetch date) so items appear where the author
452 - /// intended, even if fetched out of order during backfill.
453 - #[tracing::instrument(skip_all)]
454 - pub async fn list_all(&self, limit: i64, offset: i64) -> Result<Vec<DbFeedItem>, sqlx::Error> {
455 - sqlx::query_as(
456 - "SELECT * FROM feed_items ORDER BY published_at DESC LIMIT ?1 OFFSET ?2",
457 - )
458 - .bind(limit)
459 - .bind(offset)
460 - .fetch_all(&self.pool)
461 - .await
462 - }
463 -
464 - /// List items belonging to a specific feed, newest first.
465 - #[tracing::instrument(skip_all)]
466 - pub async fn list_by_feed(
467 - &self,
468 - feed_id: FeedId,
469 - limit: i64,
470 - offset: i64,
471 - ) -> Result<Vec<DbFeedItem>, sqlx::Error> {
472 - sqlx::query_as(
473 - "SELECT * FROM feed_items WHERE feed_id = ?1 ORDER BY published_at DESC LIMIT ?2 OFFSET ?3",
474 - )
475 - .bind(feed_id)
476 - .bind(limit)
477 - .bind(offset)
478 - .fetch_all(&self.pool)
479 - .await
480 - }
481 -
482 - /// List items from a specific busser source, newest first.
483 - #[tracing::instrument(skip_all)]
484 - pub async fn list_by_busser(
485 - &self,
486 - busser_id: &str,
487 - limit: i64,
488 - offset: i64,
489 - ) -> Result<Vec<DbFeedItem>, sqlx::Error> {
490 - sqlx::query_as(
491 - "SELECT * FROM feed_items WHERE busser_id = ?1 ORDER BY published_at DESC LIMIT ?2 OFFSET ?3",
492 - )
493 - .bind(busser_id)
494 - .bind(limit)
495 - .bind(offset)
496 - .fetch_all(&self.pool)
497 - .await
498 - }
499 -
500 - /// List unread items only, newest first.
Lines truncated
A crates/bb-db/src/repository/bookmarks.rs +206
@@ -0,0 +1,206 @@
1 + use super::*;
2 +
3 + #[derive(Clone)]
4 + /// Repository for bookmark (reading list) CRUD
5 + pub struct BookmarksRepository {
6 + pool: SqlitePool,
7 + }
8 +
9 + impl BookmarksRepository {
10 + pub fn new(pool: SqlitePool) -> Self {
11 + Self { pool }
12 + }
13 +
14 + /// Insert a new bookmark and return the created row.
15 + #[tracing::instrument(skip_all)]
16 + pub async fn create(&self, input: CreateBookmark) -> Result<DbBookmark, sqlx::Error> {
17 + let id = BookmarkId::new();
18 + let now = Utc::now().format(TIMESTAMP_FMT).to_string();
19 +
20 + let mut tx = self.pool.begin().await?;
21 +
22 + let bookmark: DbBookmark = sqlx::query_as(
23 + r#"
24 + INSERT INTO bookmarks (id, url, title, description, author, source_name,
25 + feed_item_id, notes, is_pinned, created_at, updated_at)
26 + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, 0, ?9, ?9)
27 + RETURNING *
28 + "#,
29 + )
30 + .bind(id)
31 + .bind(&input.url)
32 + .bind(&input.title)
33 + .bind(&input.description)
34 + .bind(&input.author)
35 + .bind(&input.source_name)
36 + .bind(&input.feed_item_id)
37 + .bind(&input.notes)
38 + .bind(&now)
39 + .fetch_one(&mut *tx)
40 + .await?;
41 +
42 + // Insert tags
43 + for tag in &input.tags {
44 + let tag = tag.trim();
45 + if tag.is_empty() {
46 + continue;
47 + }
48 + sqlx::query("INSERT OR IGNORE INTO bookmark_tags (bookmark_id, tag) VALUES (?1, ?2)")
49 + .bind(id)
50 + .bind(tag)
51 + .execute(&mut *tx)
52 + .await?;
53 + }
54 +
55 + tx.commit().await?;
56 + Ok(bookmark)
57 + }
58 +
59 + /// Look up a single bookmark by ID. Returns `None` if not found.
60 + #[tracing::instrument(skip_all)]
61 + pub async fn get(&self, id: BookmarkId) -> Result<Option<DbBookmark>, sqlx::Error> {
62 + sqlx::query_as("SELECT * FROM bookmarks WHERE id = ?1")
63 + .bind(id)
64 + .fetch_optional(&self.pool)
65 + .await
66 + }
67 +
68 + /// Look up a bookmark by URL. Returns `None` if not found.
69 + #[tracing::instrument(skip_all)]
70 + pub async fn get_by_url(&self, url: &str) -> Result<Option<DbBookmark>, sqlx::Error> {
71 + sqlx::query_as("SELECT * FROM bookmarks WHERE url = ?1")
72 + .bind(url)
73 + .fetch_optional(&self.pool)
74 + .await
75 + }
76 +
77 + /// Look up a bookmark by its linked feed item ID. Returns `None` if not found.
78 + #[tracing::instrument(skip_all)]
79 + pub async fn get_by_feed_item(&self, feed_item_id: &str) -> Result<Option<DbBookmark>, sqlx::Error> {
80 + sqlx::query_as("SELECT * FROM bookmarks WHERE feed_item_id = ?1")
81 + .bind(feed_item_id)
82 + .fetch_optional(&self.pool)
83 + .await
84 + }
85 +
86 + /// List bookmarks, optionally filtered by tag. Ordered by pinned first, then newest.
87 + #[tracing::instrument(skip_all)]
88 + pub async fn list(&self, tag: Option<&str>) -> Result<Vec<DbBookmark>, sqlx::Error> {
89 + match tag {
90 + Some(tag) => {
91 + sqlx::query_as(
92 + r#"
93 + SELECT b.* FROM bookmarks b
94 + INNER JOIN bookmark_tags bt ON bt.bookmark_id = b.id
95 + WHERE bt.tag = ?1
96 + ORDER BY b.is_pinned DESC, b.created_at DESC
97 + "#,
98 + )
99 + .bind(tag)
100 + .fetch_all(&self.pool)
101 + .await
102 + }
103 + None => {
104 + sqlx::query_as(
105 + "SELECT * FROM bookmarks ORDER BY is_pinned DESC, created_at DESC",
106 + )
107 + .fetch_all(&self.pool)
108 + .await
109 + }
110 + }
111 + }
112 +
113 + /// Update a bookmark's mutable fields.
114 + #[tracing::instrument(skip_all)]
115 + pub async fn update(&self, id: BookmarkId, input: UpdateBookmark) -> Result<(), sqlx::Error> {
116 + let now = Utc::now().format(TIMESTAMP_FMT).to_string();
117 +
118 + sqlx::query(
119 + r#"
120 + UPDATE bookmarks SET
121 + title = COALESCE(?2, title),
122 + description = COALESCE(?3, description),
123 + notes = COALESCE(?4, notes),
124 + is_pinned = COALESCE(?5, is_pinned),
125 + updated_at = ?1
126 + WHERE id = ?6
127 + "#,
128 + )
129 + .bind(&now)
130 + .bind(&input.title)
131 + .bind(&input.description)
132 + .bind(&input.notes)
133 + .bind(input.is_pinned.map(|b| b as i32))
134 + .bind(id)
135 + .execute(&self.pool)
136 + .await?;
137 + Ok(())
138 + }
139 +
140 + /// Delete a bookmark by ID (cascade deletes its tags).
141 + #[tracing::instrument(skip_all)]
142 + pub async fn delete(&self, id: BookmarkId) -> Result<(), sqlx::Error> {
143 + sqlx::query("DELETE FROM bookmarks WHERE id = ?1")
144 + .bind(id)
145 + .execute(&self.pool)
146 + .await?;
147 + Ok(())
148 + }
149 +
150 + /// Replace all tags for a bookmark (delete-all-then-insert).
151 + #[tracing::instrument(skip_all)]
152 + pub async fn set_tags(&self, id: BookmarkId, tags: &[String]) -> Result<(), sqlx::Error> {
153 + let mut tx = self.pool.begin().await?;
154 +
155 + sqlx::query("DELETE FROM bookmark_tags WHERE bookmark_id = ?1")
156 + .bind(id)
157 + .execute(&mut *tx)
158 + .await?;
159 +
160 + for tag in tags {
161 + let tag = tag.trim();
162 + if tag.is_empty() {
163 + continue;
164 + }
165 + sqlx::query("INSERT OR IGNORE INTO bookmark_tags (bookmark_id, tag) VALUES (?1, ?2)")
166 + .bind(id)
167 + .bind(tag)
168 + .execute(&mut *tx)
169 + .await?;
170 + }
171 +
172 + tx.commit().await?;
173 + Ok(())
174 + }
175 +
176 + /// Get all tags for a bookmark.
177 + #[tracing::instrument(skip_all)]
178 + pub async fn get_tags(&self, id: BookmarkId) -> Result<Vec<String>, sqlx::Error> {
179 + let rows: Vec<(String,)> =
180 + sqlx::query_as("SELECT tag FROM bookmark_tags WHERE bookmark_id = ?1 ORDER BY tag")
181 + .bind(id)
182 + .fetch_all(&self.pool)
183 + .await?;
184 + Ok(rows.into_iter().map(|r| r.0).collect())
185 + }
186 +
187 + /// Get all distinct tags across all bookmarks.
188 + #[tracing::instrument(skip_all)]
189 + pub async fn list_all_tags(&self) -> Result<Vec<String>, sqlx::Error> {
190 + let rows: Vec<(String,)> =
191 + sqlx::query_as("SELECT DISTINCT tag FROM bookmark_tags ORDER BY tag")
192 + .fetch_all(&self.pool)
193 + .await?;
194 + Ok(rows.into_iter().map(|r| r.0).collect())
195 + }
196 +
197 + /// Count total bookmarks.
198 + #[tracing::instrument(skip_all)]
199 + pub async fn count(&self) -> Result<i64, sqlx::Error> {
200 + let row: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM bookmarks")
201 + .fetch_one(&self.pool)
202 + .await?;
203 + Ok(row.0)
204 + }
205 + }
206 +
A crates/bb-db/src/repository/config.rs +53
@@ -0,0 +1,53 @@
1 + use super::*;
2 +
3 + #[derive(Clone)]
4 + /// Repository for user_config key-value pairs (theme, welcome flag, etc.)
5 + pub struct ConfigRepository {
6 + pool: SqlitePool,
7 + }
8 +
9 + impl ConfigRepository {
10 + #[tracing::instrument(skip_all)]
11 + pub fn new(pool: SqlitePool) -> Self {
12 + Self { pool }
13 + }
14 +
15 + /// Get a config value by key.
16 + #[tracing::instrument(skip_all)]
17 + pub async fn get(&self, key: &str) -> Result<Option<String>, sqlx::Error> {
18 + let row: Option<(String,)> =
19 + sqlx::query_as("SELECT value FROM user_config WHERE key = ?1")
20 + .bind(key)
21 + .fetch_optional(&self.pool)
22 + .await?;
23 + Ok(row.map(|(v,)| v))
24 + }
25 +
26 + /// Set a config value, inserting or updating on conflict.
27 + #[tracing::instrument(skip_all)]
28 + pub async fn set(&self, key: &str, value: &str) -> Result<(), sqlx::Error> {
29 + sqlx::query(
30 + r#"
31 + INSERT INTO user_config (key, value)
32 + VALUES (?1, ?2)
33 + ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value
34 + "#,
35 + )
36 + .bind(key)
37 + .bind(value)
38 + .execute(&self.pool)
39 + .await?;
40 + Ok(())
41 + }
42 +
43 + /// Delete a config entry by key.
44 + #[tracing::instrument(skip_all)]
45 + pub async fn delete(&self, key: &str) -> Result<(), sqlx::Error> {
46 + sqlx::query("DELETE FROM user_config WHERE key = ?1")
47 + .bind(key)
48 + .execute(&self.pool)
49 + .await?;
50 + Ok(())
51 + }
52 + }
53 +
A crates/bb-db/src/repository/feeds.rs +291
@@ -0,0 +1,291 @@
1 + use super::*;
2 +
3 + #[derive(Clone)]
4 + /// Repository for feed subscription CRUD and fetch tracking
5 + pub struct FeedsRepository {
6 + pool: SqlitePool,
7 + }
8 +
9 + impl FeedsRepository {
10 + /// Create a new feeds repository backed by the given pool.
11 + #[tracing::instrument(skip_all)]
12 + pub fn new(pool: SqlitePool) -> Self {
13 + Self { pool }
14 + }
15 +
16 + /// Insert a new feed and return the created row.
17 + /// New feeds default to `enabled = 1` (auto-fetch active) and
18 + /// `created_at = updated_at` (no separate "first modified" vs "created" notion).
19 + #[tracing::instrument(skip_all)]
20 + pub async fn create(&self, input: CreateFeed) -> Result<DbFeed, sqlx::Error> {
21 + let id = FeedId::new();
22 + let now = Utc::now().format(TIMESTAMP_FMT).to_string();
23 + // serde_json::Value is always serializable; unwrap is safe here.
24 + let config = serde_json::to_string(&input.config).unwrap_or_else(|_| "{}".to_string());
25 +
26 + sqlx::query_as(
27 + r#"
28 + INSERT INTO feeds (id, busser_id, name, config, enabled, created_at, updated_at)
29 + VALUES (?1, ?2, ?3, ?4, 1, ?5, ?5)
30 + RETURNING *
31 + "#,
32 + )
33 + .bind(id)
34 + .bind(input.busser_id.as_str())
35 + .bind(&input.name)
36 + .bind(&config)
37 + .bind(&now)
38 + .fetch_one(&self.pool)
39 + .await
40 + }
41 +
42 + /// Look up a single feed by its ID. Returns `None` if not found.
43 + #[tracing::instrument(skip_all)]
44 + pub async fn get(&self, id: FeedId) -> Result<Option<DbFeed>, sqlx::Error> {
45 + sqlx::query_as("SELECT * FROM feeds WHERE id = ?1")
46 + .bind(id)
47 + .fetch_optional(&self.pool)
48 + .await
49 + }
50 +
51 + /// List all feeds belonging to a given busser, ordered by name.
52 + #[tracing::instrument(skip_all)]
53 + pub async fn get_by_busser(&self, busser_id: &str) -> Result<Vec<DbFeed>, sqlx::Error> {
54 + sqlx::query_as("SELECT * FROM feeds WHERE busser_id = ?1 ORDER BY name")
55 + .bind(busser_id)
56 + .fetch_all(&self.pool)
57 + .await
58 + }
59 +
60 + /// List only enabled feeds that are not circuit-broken, ordered by name.
61 + #[tracing::instrument(skip_all)]
62 + pub async fn list_enabled(&self) -> Result<Vec<DbFeed>, sqlx::Error> {
63 + sqlx::query_as(
64 + "SELECT * FROM feeds WHERE enabled = 1 AND circuit_broken = 0 ORDER BY name",
65 + )
66 + .fetch_all(&self.pool)
67 + .await
68 + }
69 +
70 + /// List every feed (enabled or disabled), ordered by name.
71 + #[tracing::instrument(skip_all)]
72 + pub async fn list_all(&self) -> Result<Vec<DbFeed>, sqlx::Error> {
73 + sqlx::query_as("SELECT * FROM feeds ORDER BY name")
74 + .fetch_all(&self.pool)
75 + .await
76 + }
77 +
78 + /// Update the `last_fetch` and `updated_at` timestamps to now.
79 + #[tracing::instrument(skip_all)]
80 + pub async fn update_last_fetch(&self, id: FeedId) -> Result<(), sqlx::Error> {
81 + let now = Utc::now().format(TIMESTAMP_FMT).to_string();
82 + sqlx::query("UPDATE feeds SET last_fetch = ?1, updated_at = ?1 WHERE id = ?2")
83 + .bind(&now)
84 + .bind(id)
85 + .execute(&self.pool)
86 + .await?;
87 + Ok(())
88 + }
89 +
90 + /// Enable or disable a feed.
91 + #[tracing::instrument(skip_all)]
92 + pub async fn set_enabled(&self, id: FeedId, enabled: bool) -> Result<(), sqlx::Error> {
93 + let now = Utc::now().format(TIMESTAMP_FMT).to_string();
94 + sqlx::query("UPDATE feeds SET enabled = ?1, updated_at = ?2 WHERE id = ?3")
95 + .bind(enabled)
96 + .bind(&now)
97 + .bind(id)
98 + .execute(&self.pool)
99 + .await?;
100 + Ok(())
101 + }
102 +
103 + /// Record a successful fetch: reset failure counter, clear error, update timestamps.
104 + #[tracing::instrument(skip_all)]
105 + pub async fn record_fetch_success(&self, id: FeedId) -> Result<(), sqlx::Error> {
106 + let now = Utc::now().format(TIMESTAMP_FMT).to_string();
107 + sqlx::query(
108 + "UPDATE feeds SET consecutive_failures = 0, last_error = NULL, \
109 + last_success_at = ?1, last_fetch = ?1, updated_at = ?1 WHERE id = ?2",
110 + )
111 + .bind(&now)
112 + .bind(id)
113 + .execute(&self.pool)
114 + .await?;
115 + Ok(())
116 + }
117 +
118 + /// Record a fetch failure: increment counter, store error, update timestamp.
119 + ///
120 + /// Returns `true` if the circuit breaker tripped (i.e. the feed just crossed
121 + /// the [`CIRCUIT_BREAKER_THRESHOLD`] and was marked `circuit_broken = 1`).
122 + #[tracing::instrument(skip_all)]
123 + pub async fn record_fetch_failure(
124 + &self,
125 + id: FeedId,
126 + error: &str,
127 + ) -> Result<bool, sqlx::Error> {
128 + let now = Utc::now().format(TIMESTAMP_FMT).to_string();
129 + sqlx::query(
130 + "UPDATE feeds SET consecutive_failures = consecutive_failures + 1, \
131 + last_error = ?1, last_fetch = ?2, updated_at = ?2 WHERE id = ?3",
132 + )
133 + .bind(error)
134 + .bind(&now)
135 + .bind(id)
136 + .execute(&self.pool)
137 + .await?;
138 +
139 + // Check if we just crossed the threshold
140 + let feed = self.get(id).await?;
141 + if let Some(feed) = feed
142 + && !feed.circuit_broken
143 + && feed.consecutive_failures >= CIRCUIT_BREAKER_THRESHOLD
144 + {
145 + self.set_circuit_broken(id, true).await?;
146 + return Ok(true);
147 + }
148 + Ok(false)
149 + }
150 +
151 + /// Record a structured fetch failure with category-aware behavior:
152 + ///
153 + /// - `RateLimited` — store error JSON, do NOT increment `consecutive_failures`.
154 + /// - `Auth` / `Config` — increment + immediately set `circuit_broken = 1`.
155 + /// - `Transient` / `Parse` / `Unknown` — increment normally (existing behavior).
156 + ///
157 + /// Returns `true` if the circuit breaker tripped.
158 + #[tracing::instrument(skip_all)]
159 + pub async fn record_fetch_failure_structured(
160 + &self,
161 + id: FeedId,
162 + error: &StructuredError,
163 + ) -> Result<bool, sqlx::Error> {
164 + let now = Utc::now().format(TIMESTAMP_FMT).to_string();
165 + let error_json = error.to_json();
166 +
167 + match error.category {
168 + ErrorCategory::RateLimited => {
169 + // Store error but don't increment failure counter
170 + sqlx::query(
171 + "UPDATE feeds SET last_error = ?1, last_fetch = ?2, updated_at = ?2 WHERE id = ?3",
172 + )
173 + .bind(&error_json)
174 + .bind(&now)
175 + .bind(id)
176 + .execute(&self.pool)
177 + .await?;
178 + Ok(false)
179 + }
180 + ErrorCategory::Auth | ErrorCategory::Config => {
181 + // Increment + immediate circuit break
182 + sqlx::query(
183 + "UPDATE feeds SET consecutive_failures = consecutive_failures + 1, \
184 + last_error = ?1, last_fetch = ?2, updated_at = ?2 WHERE id = ?3",
185 + )
186 + .bind(&error_json)
187 + .bind(&now)
188 + .bind(id)
189 + .execute(&self.pool)
190 + .await?;
191 + self.set_circuit_broken(id, true).await?;
192 + Ok(true)
193 + }
194 + ErrorCategory::Transient | ErrorCategory::Parse | ErrorCategory::Unknown => {
195 + // Normal behavior: increment and check threshold
196 + sqlx::query(
197 + "UPDATE feeds SET consecutive_failures = consecutive_failures + 1, \
198 + last_error = ?1, last_fetch = ?2, updated_at = ?2 WHERE id = ?3",
199 + )
200 + .bind(&error_json)
201 + .bind(&now)
202 + .bind(id)
203 + .execute(&self.pool)
204 + .await?;
205 +
206 + let feed = self.get(id).await?;
207 + if let Some(feed) = feed
208 + && !feed.circuit_broken
209 + && feed.consecutive_failures >= CIRCUIT_BREAKER_THRESHOLD
210 + {
211 + self.set_circuit_broken(id, true).await?;
212 + return Ok(true);
213 + }
214 + Ok(false)
215 + }
216 + }
217 + }
218 +
219 + /// Mark a feed as circuit-broken (or clear the circuit breaker).
220 + #[tracing::instrument(skip_all)]
221 + pub async fn set_circuit_broken(
222 + &self,
223 + id: FeedId,
224 + broken: bool,
225 + ) -> Result<(), sqlx::Error> {
226 + let now = Utc::now().format(TIMESTAMP_FMT).to_string();
227 + sqlx::query(
228 + "UPDATE feeds SET circuit_broken = ?1, updated_at = ?2 WHERE id = ?3",
229 + )
230 + .bind(broken)
231 + .bind(&now)
232 + .bind(id)
233 + .execute(&self.pool)
234 + .await?;
235 + Ok(())
236 + }
237 +
238 + /// Reset the circuit breaker on a feed: clear `circuit_broken`, reset
239 + /// `consecutive_failures` to 0, and clear `last_error`.
240 + ///
241 + /// Called when a user manually triggers a fetch for a circuit-broken feed.
242 + #[tracing::instrument(skip_all)]
243 + pub async fn reset_circuit_breaker(&self, id: FeedId) -> Result<(), sqlx::Error> {
244 + let now = Utc::now().format(TIMESTAMP_FMT).to_string();
245 + sqlx::query(
246 + "UPDATE feeds SET circuit_broken = 0, consecutive_failures = 0, \
247 + last_error = NULL, updated_at = ?1 WHERE id = ?2",
248 + )
249 + .bind(&now)
250 + .bind(id)
251 + .execute(&self.pool)
252 + .await?;
253 + Ok(())
254 + }
255 +
256 + /// Update a feed's config JSON string.
257 + #[tracing::instrument(skip_all)]
258 + pub async fn update_config(&self, id: FeedId, config: &str) -> Result<(), sqlx::Error> {
259 + let now = Utc::now().format(TIMESTAMP_FMT).to_string();
260 + sqlx::query("UPDATE feeds SET config = ?1, updated_at = ?2 WHERE id = ?3")
261 + .bind(config)
262 + .bind(&now)
263 + .bind(id)
264 + .execute(&self.pool)
265 + .await?;
266 + Ok(())
267 + }
268 +
269 + /// Update a feed's display name.
270 + #[tracing::instrument(skip_all)]
271 + pub async fn update_name(&self, id: FeedId, name: &str) -> Result<(), sqlx::Error> {
272 + let now = Utc::now().format(TIMESTAMP_FMT).to_string();
273 + sqlx::query("UPDATE feeds SET name = ?1, updated_at = ?2 WHERE id = ?3")
274 + .bind(name)
275 + .bind(&now)
276 + .bind(id)
277 + .execute(&self.pool)
278 + .await?;
279 + Ok(())
280 + }
281 +
282 + /// Delete a feed by ID.
283 + #[tracing::instrument(skip_all)]
284 + pub async fn delete(&self, id: FeedId) -> Result<(), sqlx::Error> {
285 + sqlx::query("DELETE FROM feeds WHERE id = ?1")
286 + .bind(id)
287 + .execute(&self.pool)
288 + .await?;
289 + Ok(())
290 + }
291 + }
A crates/bb-db/src/repository/items.rs +464
@@ -0,0 +1,464 @@
1 + use super::*;
2 +
3 + #[derive(Clone)]
4 + /// Repository for feed item CRUD, read/star toggling, and paginated listing
5 + pub struct ItemsRepository {
6 + pool: SqlitePool,
7 + }
8 +
9 + impl ItemsRepository {
10 + /// Create a new items repository backed by the given pool.
11 + #[tracing::instrument(skip_all)]
12 + pub fn new(pool: SqlitePool) -> Self {
13 + Self { pool }
14 + }
15 +
16 + /// Insert a feed item or update it if `external_id` already exists.
17 + ///
18 + /// The ON CONFLICT clause deliberately preserves user state (`is_read`,
19 + /// `is_starred`) — these are never overwritten by a re-fetch. Content fields
20 + /// (author, text, body, etc.) are updated in case the source edited the post.
21 + #[tracing::instrument(skip_all)]
22 + pub async fn upsert(&self, input: CreateFeedItem) -> Result<DbFeedItem, sqlx::Error> {
23 + let id = ItemId::new();
24 + let now = Utc::now().format(TIMESTAMP_FMT).to_string();
25 + let published = input.published_at.format(TIMESTAMP_FMT).to_string();
26 + // Vec<String> is always serializable; unwrap is safe here.
27 + let media = serde_json::to_string(&input.media).unwrap_or_else(|_| "[]".to_string());
28 + let tags = serde_json::to_string(&input.tags).unwrap_or_else(|_| "[]".to_string());
29 + let actions =
30 + serde_json::to_string(&input.actions).unwrap_or_else(|_| "[]".to_string());
31 +
32 + sqlx::query_as(
33 + r#"
34 + INSERT INTO feed_items (
35 + id, external_id, feed_id, busser_id,
36 + bite_author, bite_text, bite_secondary, bite_indicator,
37 + title, body, url, media, actions,
38 + published_at, fetched_at, source_name, score, tags,
39 + is_read, is_starred, created_at, updated_at
40 + )
41 + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18, 0, 0, ?19, ?19)
42 + ON CONFLICT (external_id) DO UPDATE SET
43 + bite_author = EXCLUDED.bite_author,
44 + bite_text = EXCLUDED.bite_text,
45 + bite_secondary = EXCLUDED.bite_secondary,
46 + bite_indicator = EXCLUDED.bite_indicator,
47 + title = EXCLUDED.title,
48 + body = EXCLUDED.body,
49 + url = EXCLUDED.url,
50 + media = EXCLUDED.media,
51 + actions = EXCLUDED.actions,
52 + score = EXCLUDED.score,
53 + tags = EXCLUDED.tags,
54 + updated_at = EXCLUDED.updated_at
55 + RETURNING *
56 + "#,
57 + )
58 + .bind(id)
59 + .bind(&input.external_id)
60 + .bind(input.feed_id)
61 + .bind(input.busser_id.as_str())
62 + .bind(&input.bite_author)
63 + .bind(&input.bite_text)
64 + .bind(&input.bite_secondary)
65 + .bind(&input.bite_indicator)
66 + .bind(&input.title)
67 + .bind(&input.body)
68 + .bind(&input.url)
69 + .bind(&media)
70 + .bind(&actions)
71 + .bind(&published)
72 + .bind(&now)
73 + .bind(&input.source_name)
74 + .bind(input.score)
75 + .bind(&tags)
76 + .bind(&now)
77 + .fetch_one(&self.pool)
78 + .await
79 + }
80 +
81 + /// Look up a feed item by its internal ID.
82 + #[tracing::instrument(skip_all)]
83 + pub async fn get(&self, id: ItemId) -> Result<Option<DbFeedItem>, sqlx::Error> {
84 + sqlx::query_as("SELECT * FROM feed_items WHERE id = ?1")
85 + .bind(id)
86 + .fetch_optional(&self.pool)
87 + .await
88 + }
89 +
90 + /// Look up a feed item by the `external_id` produced by the busser.
91 + #[tracing::instrument(skip_all)]
92 + pub async fn get_by_external_id(
93 + &self,
94 + external_id: &str,
95 + ) -> Result<Option<DbFeedItem>, sqlx::Error> {
96 + sqlx::query_as("SELECT * FROM feed_items WHERE external_id = ?1")
97 + .bind(external_id)
98 + .fetch_optional(&self.pool)
99 + .await
100 + }
101 +
102 + /// List all items ordered by `published_at` descending, with pagination.
103 + /// Sorted by publish date (not fetch date) so items appear where the author
104 + /// intended, even if fetched out of order during backfill.
105 + #[tracing::instrument(skip_all)]
106 + pub async fn list_all(&self, limit: i64, offset: i64) -> Result<Vec<DbFeedItem>, sqlx::Error> {
107 + sqlx::query_as(
108 + "SELECT * FROM feed_items ORDER BY published_at DESC LIMIT ?1 OFFSET ?2",
109 + )
110 + .bind(limit)
111 + .bind(offset)
112 + .fetch_all(&self.pool)
113 + .await
114 + }
115 +
116 + /// List items belonging to a specific feed, newest first.
117 + #[tracing::instrument(skip_all)]
118 + pub async fn list_by_feed(
119 + &self,
120 + feed_id: FeedId,
121 + limit: i64,
122 + offset: i64,
123 + ) -> Result<Vec<DbFeedItem>, sqlx::Error> {
124 + sqlx::query_as(
125 + "SELECT * FROM feed_items WHERE feed_id = ?1 ORDER BY published_at DESC LIMIT ?2 OFFSET ?3",
126 + )
127 + .bind(feed_id)
128 + .bind(limit)
129 + .bind(offset)
130 + .fetch_all(&self.pool)
131 + .await
132 + }
133 +
134 + /// List items from a specific busser source, newest first.
135 + #[tracing::instrument(skip_all)]
136 + pub async fn list_by_busser(
137 + &self,
138 + busser_id: &str,
139 + limit: i64,
140 + offset: i64,
141 + ) -> Result<Vec<DbFeedItem>, sqlx::Error> {
142 + sqlx::query_as(
143 + "SELECT * FROM feed_items WHERE busser_id = ?1 ORDER BY published_at DESC LIMIT ?2 OFFSET ?3",
144 + )
145 + .bind(busser_id)
146 + .bind(limit)
147 + .bind(offset)
148 + .fetch_all(&self.pool)
149 + .await
150 + }
151 +
152 + /// List unread items only, newest first.
153 + #[tracing::instrument(skip_all)]
154 + pub async fn list_unread(&self, limit: i64, offset: i64) -> Result<Vec<DbFeedItem>, sqlx::Error> {
155 + sqlx::query_as(
156 + "SELECT * FROM feed_items WHERE is_read = 0 ORDER BY published_at DESC LIMIT ?1 OFFSET ?2",
157 + )
158 + .bind(limit)
159 + .bind(offset)
160 + .fetch_all(&self.pool)
161 + .await
162 + }
163 +
164 + /// List starred items only, newest first.
165 + #[tracing::instrument(skip_all)]
166 + pub async fn list_starred(
167 + &self,
168 + limit: i64,
169 + offset: i64,
170 + ) -> Result<Vec<DbFeedItem>, sqlx::Error> {
171 + sqlx::query_as(
172 + "SELECT * FROM feed_items WHERE is_starred = 1 ORDER BY published_at DESC LIMIT ?1 OFFSET ?2",
173 + )
174 + .bind(limit)
175 + .bind(offset)
176 + .fetch_all(&self.pool)
177 + .await
178 + }
179 +
180 + /// Full-text search using FTS5 across title, body, and bite_text.
181 + ///
182 + /// Uses the `feed_items_fts` virtual table for fast ranked search.
183 + /// Results are ordered by FTS5 relevance then by `published_at DESC`.
184 + /// Additional boolean filters (source, unread, starred) are combined.
185 + #[tracing::instrument(skip_all)]
186 + pub async fn list_search(
187 + &self,
188 + query: &str,
189 + source: Option<&str>,
190 + unread_only: bool,
191 + starred_only: bool,
192 + limit: i64,
193 + offset: i64,
194 + ) -> Result<Vec<DbFeedItem>, sqlx::Error> {
195 + // Reject excessively long queries before any processing.
196 + if query.len() > MAX_SEARCH_QUERY_LENGTH {
197 + return Ok(vec![]);
198 + }
199 +
200 + let fts_query = sanitize_fts_query(query);
201 +
202 + // If sanitization stripped everything (e.g. query was just `*` or `^`),
203 + // return early — an empty MATCH expression would be a SQL error.
204 + if fts_query.is_empty() {
205 + return Ok(vec![]);
206 + }
207 +
208 + let mut sql = String::from(
209 + "SELECT fi.* FROM feed_items fi \
210 + INNER JOIN feed_items_fts fts ON fi.rowid = fts.rowid \
211 + WHERE feed_items_fts MATCH ?1",
212 + );
213 + if source.is_some() {
214 + sql.push_str(" AND fi.busser_id = ?4");
215 + }
216 + if unread_only {
217 + sql.push_str(" AND fi.is_read = 0");
218 + }
219 + if starred_only {
220 + sql.push_str(" AND fi.is_starred = 1");
221 + }
222 + sql.push_str(" ORDER BY fts.rank, fi.published_at DESC LIMIT ?2 OFFSET ?3");
223 +
224 + let mut q = sqlx::query_as::<_, DbFeedItem>(&sql)
225 + .bind(&fts_query) // ?1
226 + .bind(limit) // ?2
227 + .bind(offset); // ?3
228 +
229 + if let Some(src) = source {
230 + q = q.bind(src); // ?4
231 + }
232 +
233 + q.fetch_all(&self.pool).await
234 + }
235 +
236 + /// List feed items with any combination of filters pushed into SQL.
237 + ///
238 + /// Unifies `list_all`, `list_by_busser`, `list_unread`, `list_starred`,
239 + /// and `list_search` into a single method so callers don't need an
240 + /// if/else chain that drops filter combinations.
241 + #[tracing::instrument(skip_all)]
242 + pub async fn list_filtered(
243 + &self,
244 + search: Option<&str>,
245 + source: Option<&str>,
246 + unread_only: bool,
247 + starred_only: bool,
248 + limit: i64,
249 + offset: i64,
250 + ) -> Result<Vec<DbFeedItem>, sqlx::Error> {
251 + // When a search query is present, use FTS5.
252 + if let Some(query) = search {
253 + return self
254 + .list_search(query, source, unread_only, starred_only, limit, offset)
255 + .await;
256 + }
257 +
258 + // Build a dynamic query with conditional WHERE clauses.
259 + let mut sql = String::from("SELECT * FROM feed_items WHERE 1=1");
260 + if source.is_some() {
261 + sql.push_str(" AND busser_id = ?3");
262 + }
263 + if unread_only {
264 + sql.push_str(" AND is_read = 0");
265 + }
266 + if starred_only {
267 + sql.push_str(" AND is_starred = 1");
268 + }
269 + sql.push_str(" ORDER BY published_at DESC LIMIT ?1 OFFSET ?2");
270 +
271 + let mut q = sqlx::query_as::<_, DbFeedItem>(&sql)
272 + .bind(limit) // ?1
273 + .bind(offset); // ?2
274 +
275 + if let Some(src) = source {
276 + q = q.bind(src); // ?3
277 + }
278 +
279 + q.fetch_all(&self.pool).await
280 + }
281 +
282 + /// Set the read flag on a feed item.
283 + #[tracing::instrument(skip_all)]
284 + pub async fn mark_read(&self, id: ItemId, is_read: bool) -> Result<(), sqlx::Error> {
285 + let now = Utc::now().format(TIMESTAMP_FMT).to_string();
286 + sqlx::query("UPDATE feed_items SET is_read = ?1, updated_at = ?2 WHERE id = ?3")
287 + .bind(is_read)
288 + .bind(&now)
289 + .bind(id)
290 + .execute(&self.pool)
291 + .await?;
292 + Ok(())
293 + }
294 +
295 + /// Set the starred flag on a feed item.
296 + #[tracing::instrument(skip_all)]
297 + pub async fn mark_starred(&self, id: ItemId, is_starred: bool) -> Result<(), sqlx::Error> {
298 + let now = Utc::now().format(TIMESTAMP_FMT).to_string();
299 + sqlx::query("UPDATE feed_items SET is_starred = ?1, updated_at = ?2 WHERE id = ?3")
300 + .bind(is_starred)
301 + .bind(&now)
302 + .bind(id)
303 + .execute(&self.pool)
304 + .await?;
305 + Ok(())
306 + }
307 +
308 + /// Mark all unread items as read, optionally filtered to a specific source.
309 + #[tracing::instrument(skip_all)]
310 + pub async fn mark_all_read(&self, busser_id: Option<&str>) -> Result<u64, sqlx::Error> {
311 + let now = Utc::now().format(TIMESTAMP_FMT).to_string();
312 + let result = match busser_id {
313 + Some(id) => {
314 + sqlx::query(
315 + "UPDATE feed_items SET is_read = 1, updated_at = ?1 WHERE is_read = 0 AND busser_id = ?2",
316 + )
317 + .bind(&now)
318 + .bind(id)
319 + .execute(&self.pool)
320 + .await?
321 + }
322 + None => {
323 + sqlx::query(
324 + "UPDATE feed_items SET is_read = 1, updated_at = ?1 WHERE is_read = 0",
325 + )
326 + .bind(&now)
327 + .execute(&self.pool)
328 + .await?
329 + }
330 + };
331 + Ok(result.rows_affected())
332 + }
333 +
334 + /// Count all feed items.
335 + #[tracing::instrument(skip_all)]
336 + pub async fn count_all(&self) -> Result<i64, sqlx::Error> {
337 + let row: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM feed_items")
338 + .fetch_one(&self.pool)
339 + .await?;
340 + Ok(row.0)
341 + }
342 +
343 + /// Count items from a specific busser source.
344 + #[tracing::instrument(skip_all)]
345 + pub async fn count_by_busser(&self, busser_id: &str) -> Result<i64, sqlx::Error> {
346 + let row: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM feed_items WHERE busser_id = ?1")
347 + .bind(busser_id)
348 + .fetch_one(&self.pool)
349 + .await?;
350 + Ok(row.0)
351 + }
352 +
353 + /// Count unread items across all sources.
354 + #[tracing::instrument(skip_all)]
355 + pub async fn count_unread(&self) -> Result<i64, sqlx::Error> {
356 + let row: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM feed_items WHERE is_read = 0")
357 + .fetch_one(&self.pool)
358 + .await?;
359 + Ok(row.0)
360 + }
361 +
362 + /// Count starred items.
363 + #[tracing::instrument(skip_all)]
364 + pub async fn count_starred(&self) -> Result<i64, sqlx::Error> {
365 + let row: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM feed_items WHERE is_starred = 1")
366 + .fetch_one(&self.pool)
367 + .await?;
368 + Ok(row.0)
369 + }
370 +
371 + /// Count unread items from a specific busser source.
372 + #[tracing::instrument(skip_all)]
373 + pub async fn count_unread_by_busser(&self, busser_id: &str) -> Result<i64, sqlx::Error> {
374 + let row: (i64,) =
375 + sqlx::query_as("SELECT COUNT(*) FROM feed_items WHERE busser_id = ?1 AND is_read = 0")
376 + .bind(busser_id)
377 + .fetch_one(&self.pool)
378 + .await?;
379 + Ok(row.0)
380 + }
381 +
382 + /// Count items matching a combination of source, unread, and starred filters.
383 + #[tracing::instrument(skip_all)]
384 + pub async fn count_filtered(
385 + &self,
386 + source: Option<&str>,
387 + unread_only: bool,
388 + starred_only: bool,
389 + ) -> Result<i64, sqlx::Error> {
390 + let mut sql = "SELECT COUNT(*) FROM feed_items WHERE 1=1".to_string();
391 + if source.is_some() {
392 + sql.push_str(" AND busser_id = ?1");
393 + }
394 + if unread_only {
395 + sql.push_str(" AND is_read = 0");
396 + }
397 + if starred_only {
398 + sql.push_str(" AND is_starred = 1");
399 + }
400 + let mut query = sqlx::query_as::<_, (i64,)>(&sql);
401 + if let Some(s) = source {
402 + query = query.bind(s);
403 + }
404 + let row = query.fetch_one(&self.pool).await?;
405 + Ok(row.0)
406 + }
407 +
408 + /// Get total and unread counts for all busser sources in a single query.
409 + ///
410 + /// Returns `(busser_id, total_count, unread_count)` tuples. This avoids
411 + /// issuing separate count queries per source (N+1 problem).
412 + pub async fn counts_by_busser(&self) -> Result<Vec<(String, i64, i64)>, sqlx::Error> {
413 + let rows: Vec<(String, i64, i64)> = sqlx::query_as(
414 + r#"
415 + SELECT busser_id,
416 + COUNT(*) AS total_count,
417 + SUM(CASE WHEN is_read = 0 THEN 1 ELSE 0 END) AS unread_count
418 + FROM feed_items
419 + GROUP BY busser_id
420 + "#,
421 + )
422 + .fetch_all(&self.pool)
423 + .await?;
424 + Ok(rows)
425 + }
426 +
427 + /// Delete a single feed item by ID.
428 + #[tracing::instrument(skip_all)]
429 + pub async fn delete(&self, id: ItemId) -> Result<(), sqlx::Error> {
430 + sqlx::query("DELETE FROM feed_items WHERE id = ?1")
431 + .bind(id)
432 + .execute(&self.pool)
433 + .await?;
434 + Ok(())
435 + }
436 +
437 + /// Delete all items belonging to a feed. Returns the number of rows removed.
438 + #[tracing::instrument(skip_all)]
439 + pub async fn delete_by_feed(&self, feed_id: FeedId) -> Result<u64, sqlx::Error> {
440 + let result = sqlx::query("DELETE FROM feed_items WHERE feed_id = ?1")
441 + .bind(feed_id)
442 + .execute(&self.pool)
443 + .await?;
444 + Ok(result.rows_affected())
445 + }
446 +
447 + /// Delete old read items that are not starred. Returns the number of rows removed.
448 + ///
449 + /// Items published before `before` that have been read and are not starred
450 + /// will be permanently deleted. Starred items are always preserved regardless
451 + /// of age or read status.
452 + #[tracing::instrument(skip_all)]
453 + pub async fn delete_stale_read(&self, before: DateTime<Utc>) -> Result<u64, sqlx::Error> {
454 + let cutoff = before.format(TIMESTAMP_FMT).to_string();
455 + let result = sqlx::query(
456 + "DELETE FROM feed_items WHERE is_read = 1 AND is_starred = 0 AND published_at < ?1",
457 + )
458 + .bind(&cutoff)
459 + .execute(&self.pool)
460 + .await?;
461 + Ok(result.rows_affected())
462 + }
463 + }
464 +
A crates/bb-db/src/repository/mod.rs +76
@@ -0,0 +1,76 @@
1 + //! Database repositories for feeds, items, and busser state.
2 + //!
3 + //! Each repository wraps an SQLite connection pool and provides typed
4 + //! CRUD operations for a single table.
5 +
6 + use chrono::{DateTime, Utc};
7 + use sqlx::SqlitePool;
8 +
9 + use bb_interface::{ErrorCategory, StructuredError};
10 +
11 + use crate::id_types::{BookmarkId, BusserStateId, FeedId, ItemId, QueryFeedId};
12 + use crate::models::*;
13 + use crate::TIMESTAMP_FMT;
14 +
15 + /// Sanitize a user-provided search string for FTS5 MATCH syntax.
16 + ///
17 + /// Wraps each word in double quotes to prevent FTS5 syntax injection
18 + /// (e.g. a user typing `AND`, `OR`, `NOT`, `NEAR`, or `NEAR/N` won't be
19 + /// interpreted as FTS5 operators). Words are joined with spaces (implicit AND).
20 + ///
21 + /// Inside quoted strings FTS5 still honours two special characters:
22 + /// - `*` at the end of the last token triggers prefix matching
23 + /// - `^` at the start of the first token triggers beginning-of-column matching
24 + ///
25 + /// We strip those so user input like `^hello` or `world*` is treated literally.
26 + /// The `column:` prefix syntax (e.g. `title:word`) is neutralised by the quoting
27 + /// itself — colons inside double-quoted strings are treated as literal characters.
28 + fn sanitize_fts_query(query: &str) -> String {
29 + query
30 + .split_whitespace()
31 + .map(|word| {
32 + // Escape embedded double quotes (FTS5 uses "" to represent a literal ")
33 + let escaped = word.replace('"', "\"\"");
34 + // Strip `^` prefix and `*` suffix — both are special inside FTS5 quotes
35 + let escaped = escaped.trim_start_matches('^');
36 + let escaped = escaped.trim_end_matches('*');
37 + format!("\"{}\"", escaped)
38 + })
39 + // Drop tokens that became empty after stripping (e.g. bare `^`, `*`, `^*`)
40 + .filter(|token| token != "\"\"")
41 + .collect::<Vec<_>>()
42 + .join(" ")
43 + }
44 +
45 + /// Maximum allowed length for a search query string.
46 + ///
47 + /// Queries longer than this are rejected early (returning an empty result set)
48 + /// to prevent excessively large FTS5 MATCH expressions from consuming memory
49 + /// or CPU in SQLite.
50 + const MAX_SEARCH_QUERY_LENGTH: usize = 500;
51 +
52 + /// Number of consecutive failures before a feed is automatically disabled.
53 + ///
54 + /// Once a feed accumulates this many failures without a successful fetch,
55 + /// the circuit breaker trips: the feed is marked `circuit_broken = 1` and
56 + /// excluded from automatic fetch scheduling until manually reset.
57 + pub const CIRCUIT_BREAKER_THRESHOLD: i64 = 10;
58 +
59 + mod bookmarks;
60 + mod config;
61 + mod feeds;
62 + mod items;
63 + mod query_feeds;
64 + mod state;
65 + mod tags;
66 +
67 + pub use bookmarks::*;
68 + pub use config::*;
69 + pub use feeds::*;
70 + pub use items::*;
71 + pub use query_feeds::*;
72 + pub use state::*;
73 + pub use tags::*;
74 +
75 + #[cfg(test)]
76 + mod tests;
A crates/bb-db/src/repository/query_feeds.rs +89
@@ -0,0 +1,89 @@
1 + use super::*;
2 +
3 + #[derive(Clone)]
4 + /// Repository for saved query feed (virtual source) CRUD
5 + pub struct QueryFeedsRepository {
6 + pool: SqlitePool,
7 + }
8 +
9 + impl QueryFeedsRepository {
10 + #[tracing::instrument(skip_all)]
11 + pub fn new(pool: SqlitePool) -> Self {
12 + Self { pool }
13 + }
14 +
15 + /// Insert a new query feed and return the created row.
16 + pub async fn create(&self, input: CreateQueryFeed) -> Result<DbQueryFeed, sqlx::Error> {
17 + let id = QueryFeedId::new();
18 + let now = Utc::now().format(TIMESTAMP_FMT).to_string();
19 + let rules_json =
20 + serde_json::to_string(&input.rules).unwrap_or_else(|_| "[]".to_string());
21 +
22 + sqlx::query_as(
23 + r#"
24 + INSERT INTO query_feeds (id, name, rules, created_at, updated_at)
25 + VALUES (?1, ?2, ?3, ?4, ?4)
26 + RETURNING *
27 + "#,
28 + )
29 + .bind(id)
30 + .bind(&input.name)
31 + .bind(&rules_json)
32 + .bind(&now)
33 + .fetch_one(&self.pool)
34 + .await
35 + }
36 +
37 + /// Look up a single query feed by ID. Returns `None` if not found.
38 + #[tracing::instrument(skip_all)]
39 + pub async fn get(&self, id: QueryFeedId) -> Result<Option<DbQueryFeed>, sqlx::Error> {
40 + sqlx::query_as("SELECT * FROM query_feeds WHERE id = ?1")
41 + .bind(id)
42 + .fetch_optional(&self.pool)
43 + .await
44 + }
45 +
46 + /// List all query feeds, ordered by name.
47 + #[tracing::instrument(skip_all)]
48 + pub async fn list_all(&self) -> Result<Vec<DbQueryFeed>, sqlx::Error> {
49 + sqlx::query_as("SELECT * FROM query_feeds ORDER BY name")
50 + .fetch_all(&self.pool)
51 + .await
52 + }
53 +
54 + /// Update a query feed's name and rules.
55 + #[tracing::instrument(skip_all)]
56 + pub async fn update(
57 + &self,
58 + id: QueryFeedId,
59 + name: &str,
60 + rules: &[QueryCondition],
61 + ) -> Result<(), sqlx::Error> {
62 + let now = Utc::now().format(TIMESTAMP_FMT).to_string();
63 + let rules_json = serde_json::to_string(rules).unwrap_or_else(|_| "[]".to_string());
64 +
65 + sqlx::query(
66 + "UPDATE query_feeds SET name = ?1, rules = ?2, updated_at = ?3 WHERE id = ?4",
67 + )
68 + .bind(name)
69 + .bind(&rules_json)
70 + .bind(&now)
71 + .bind(id)
72 + .execute(&self.pool)
73 + .await?;
74 + Ok(())
75 + }
76 +
77 + /// Delete a query feed by ID.
78 + #[tracing::instrument(skip_all)]
79 + pub async fn delete(&self, id: QueryFeedId) -> Result<(), sqlx::Error> {
80 + sqlx::query("DELETE FROM query_feeds WHERE id = ?1")
81 + .bind(id)
82 + .execute(&self.pool)
83 + .await?;
84 + Ok(())
85 + }
86 + }
87 +
88 + // ── BookmarksRepository ──────────────────────────────────────────
89 +
A crates/bb-db/src/repository/state.rs +82
@@ -0,0 +1,82 @@
1 + use super::*;
2 +
3 + #[derive(Clone)]
4 + /// Repository for busser key-value state (cursors, tokens, pagination markers)
5 + pub struct StateRepository {
6 + pool: SqlitePool,
7 + }
8 +
9 + impl StateRepository {
10 + /// Create a new state repository backed by the given pool.
11 + #[tracing::instrument(skip_all)]
12 + pub fn new(pool: SqlitePool) -> Self {
13 + Self { pool }
14 + }
15 +
16 + /// Get a single state value for a busser by key.
17 + #[tracing::instrument(skip_all)]
18 + pub async fn get(&self, busser_id: &str, key: &str) -> Result<Option<String>, sqlx::Error> {
19 + let row: Option<(String,)> =
20 + sqlx::query_as("SELECT value FROM busser_state WHERE busser_id = ?1 AND key = ?2")
21 + .bind(busser_id)
22 + .bind(key)
23 + .fetch_optional(&self.pool)
24 + .await?;
25 + Ok(row.map(|(v,)| v))
26 + }
27 +
28 + /// Set a state value, inserting or updating on conflict.
29 + /// Uses upsert on the `(busser_id, key)` composite unique constraint so
30 + /// callers don't need to check existence first.
31 + #[tracing::instrument(skip_all)]
32 + pub async fn set(&self, busser_id: &str, key: &str, value: &str) -> Result<(), sqlx::Error> {
33 + let id = BusserStateId::new();
34 + let now = Utc::now().format(TIMESTAMP_FMT).to_string();
35 + sqlx::query(
36 + r#"
37 + INSERT INTO busser_state (id, busser_id, key, value, created_at, updated_at)
38 + VALUES (?1, ?2, ?3, ?4, ?5, ?5)
39 + ON CONFLICT (busser_id, key) DO UPDATE SET value = EXCLUDED.value, updated_at = EXCLUDED.updated_at
40 + "#,
41 + )
42 + .bind(id)
43 + .bind(busser_id)
44 + .bind(key)
45 + .bind(value)
46 + .bind(&now)
47 + .execute(&self.pool)
48 + .await?;
49 + Ok(())
50 + }
51 +
52 + /// Delete a single state entry by busser ID and key.
53 + #[tracing::instrument(skip_all)]
54 + pub async fn delete(&self, busser_id: &str, key: &str) -> Result<(), sqlx::Error> {
55 + sqlx::query("DELETE FROM busser_state WHERE busser_id = ?1 AND key = ?2")
56 + .bind(busser_id)
57 + .bind(key)
58 + .execute(&self.pool)
59 + .await?;
60 + Ok(())
61 + }
62 +
63 + /// Delete all state entries for a busser. Returns the number of rows removed.
64 + #[tracing::instrument(skip_all)]
65 + pub async fn delete_all(&self, busser_id: &str) -> Result<u64, sqlx::Error> {
66 + let result = sqlx::query("DELETE FROM busser_state WHERE busser_id = ?1")
67 + .bind(busser_id)
68 + .execute(&self.pool)
69 + .await?;
70 + Ok(result.rows_affected())
71 + }
72 +
73 + /// List all state entries for a busser, ordered by key.
74 + #[tracing::instrument(skip_all)]
75 + pub async fn list(&self, busser_id: &str) -> Result<Vec<DbBusserState>, sqlx::Error> {
76 + sqlx::query_as("SELECT * FROM busser_state WHERE busser_id = ?1 ORDER BY key")
77 + .bind(busser_id)
78 + .fetch_all(&self.pool)
79 + .await
80 + }
81 + }
82 +
A crates/bb-db/src/repository/tags.rs +114
@@ -0,0 +1,114 @@
1 + use super::*;
2 +
3 + #[derive(Clone)]
4 + /// Repository for per-feed tag assignment and listing
5 + pub struct TagsRepository {
6 + pool: SqlitePool,
7 + }
8 +
9 + impl TagsRepository {
10 + /// Create a new tags repository backed by the given pool.
11 + #[tracing::instrument(skip_all)]
12 + pub fn new(pool: SqlitePool) -> Self {
13 + Self { pool }
14 + }
15 +
16 + /// Replace all tags on a feed. Deletes existing tags, then inserts the new set.
17 + ///
18 + /// Wrapped in a transaction so the delete + inserts are atomic — a failure
19 + /// mid-way won't leave the feed with zero tags.
20 + #[tracing::instrument(skip_all)]
21 + pub async fn set_tags(&self, feed_id: FeedId, tags: &[String]) -> Result<(), sqlx::Error> {
22 + let mut tx = self.pool.begin().await?;
23 +
24 + sqlx::query("DELETE FROM feed_tags WHERE feed_id = ?1")
25 + .bind(feed_id)
26 + .execute(&mut *tx)
27 + .await?;
28 +
29 + for tag in tags {
30 + if !tag.is_empty() {
31 + sqlx::query("INSERT OR IGNORE INTO feed_tags (feed_id, tag) VALUES (?1, ?2)")
32 + .bind(feed_id)
33 + .bind(tag)
34 + .execute(&mut *tx)
35 + .await?;
36 + }
37 + }
38 +
39 + tx.commit().await?;
40 + Ok(())
41 + }
42 +
43 + /// Add a single tag to a feed (idempotent).
44 + #[tracing::instrument(skip_all)]
45 + pub async fn add_tag(&self, feed_id: FeedId, tag: &str) -> Result<(), sqlx::Error> {
46 + sqlx::query("INSERT OR IGNORE INTO feed_tags (feed_id, tag) VALUES (?1, ?2)")
47 + .bind(feed_id)
48 + .bind(tag)
49 + .execute(&self.pool)
50 + .await?;
51 + Ok(())
52 + }
53 +
54 + /// Remove a single tag from a feed.
55 + #[tracing::instrument(skip_all)]
56 + pub async fn remove_tag(&self, feed_id: FeedId, tag: &str) -> Result<(), sqlx::Error> {
57 + sqlx::query("DELETE FROM feed_tags WHERE feed_id = ?1 AND tag = ?2")
58 + .bind(feed_id)
59 + .bind(tag)
60 + .execute(&self.pool)
61 + .await?;
62 + Ok(())
63 + }
64 +
65 + /// Get all tags for a feed, ordered alphabetically.
66 + #[tracing::instrument(skip_all)]
67 + pub async fn get_tags(&self, feed_id: FeedId) -> Result<Vec<String>, sqlx::Error> {
68 + let rows: Vec<(String,)> =
69 + sqlx::query_as("SELECT tag FROM feed_tags WHERE feed_id = ?1 ORDER BY tag")
70 + .bind(feed_id)
71 + .fetch_all(&self.pool)
72 + .await?;
73 + Ok(rows.into_iter().map(|(t,)| t).collect())
74 + }
75 +
76 + /// List all distinct tags across all feeds, ordered alphabetically.
77 + #[tracing::instrument(skip_all)]
78 + pub async fn list_all_tags(&self) -> Result<Vec<String>, sqlx::Error> {
79 + let rows: Vec<(String,)> =
80 + sqlx::query_as("SELECT DISTINCT tag FROM feed_tags ORDER BY tag")
81 + .fetch_all(&self.pool)
82 + .await?;
83 + Ok(rows.into_iter().map(|(t,)| t).collect())
84 + }
85 +
86 + /// Get all (feed_id, tag) pairs for bulk rendering.
87 + #[tracing::instrument(skip_all)]
88 + pub async fn all_feed_tags(&self) -> Result<Vec<(FeedId, String)>, sqlx::Error> {
89 + sqlx::query_as("SELECT feed_id, tag FROM feed_tags ORDER BY feed_id, tag")
90 + .fetch_all(&self.pool)
91 + .await
92 + }
93 +
94 + /// Get feed IDs that have any of the given tags.
95 + #[tracing::instrument(skip_all)]
96 + pub async fn feed_ids_with_tags(&self, tags: &[String]) -> Result<Vec<FeedId>, sqlx::Error> {
97 + if tags.is_empty() {
98 + return Ok(Vec::new());
99 + }
100 + // Build a query with placeholders for each tag
101 + let placeholders: Vec<String> = (1..=tags.len()).map(|i| format!("?{}", i)).collect();
102 + let sql = format!(
103 + "SELECT DISTINCT feed_id FROM feed_tags WHERE tag IN ({})",
104 + placeholders.join(", ")
105 + );
106 + let mut query = sqlx::query_as::<_, (FeedId,)>(&sql);
107 + for tag in tags {
108 + query = query.bind(tag);
109 + }
110 + let rows = query.fetch_all(&self.pool).await?;
111 + Ok(rows.into_iter().map(|(id,)| id).collect())
112 + }
113 + }
114 +
A crates/bb-db/src/repository/tests.rs +500
@@ -0,0 +1,1523 @@
1 + use super::*;
2 + use crate::id_types::BusserId;
3 + use chrono::{DateTime, Duration};
4 + use serde_json::json;
5 +
6 + /// Connect to in-memory SQLite and run all migrations.
7 + async fn test_db() -> SqlitePool {
8 + let pool = SqlitePool::connect("sqlite::memory:").await.unwrap();
9 + sqlx::migrate!("../../migrations/sqlite")
10 + .run(&pool)
11 + .await
12 + .unwrap();
13 + pool
14 + }
15 +
16 + /// Insert a feed with the given busser and name, return the created row.
17 + async fn make_feed(pool: &SqlitePool, busser_id: &str, name: &str) -> DbFeed {
18 + FeedsRepository::new(pool.clone())
19 + .create(CreateFeed {
20 + busser_id: BusserId::new(busser_id),
21 + name: name.to_string(),
22 + config: json!({}),
23 + })
24 + .await
25 + .unwrap()
26 + }
27 +
28 + /// Insert a feed item with minimal required fields, return the created row.
29 + async fn make_item(pool: &SqlitePool, feed: &DbFeed, external_id: &str) -> DbFeedItem {
30 + make_item_at(pool, feed, external_id, Utc::now()).await
31 + }
32 +
33 + /// Insert a feed item at a specific published_at time.
34 + async fn make_item_at(
35 + pool: &SqlitePool,
36 + feed: &DbFeed,
37 + external_id: &str,
38 + published_at: DateTime<Utc>,
39 + ) -> DbFeedItem {
40 + ItemsRepository::new(pool.clone())
41 + .upsert(CreateFeedItem {
42 + external_id: external_id.to_string(),
43 + feed_id: feed.id,
44 + busser_id: feed.busser_id.clone(),
45 + bite_author: "author".to_string(),
46 + bite_text: format!("Item {external_id}"),
47 + bite_secondary: None,
48 + bite_indicator: None,
49 + title: Some(format!("Title {external_id}")),
50 + body: None,
51 + url: None,
52 + media: vec![],
53 + published_at,
54 + source_name: "test".to_string(),
55 + score: None,
56 + tags: vec![],
57 + actions: vec![],
58 + })
59 + .await
60 + .unwrap()
61 + }
62 +
63 + // ── FeedsRepository ───────────────────────────────────────────
64 +
65 + #[tokio::test]
66 + async fn feeds_create_and_get() {
67 + let pool = test_db().await;
68 + let feed = make_feed(&pool, "rss", "My Feed").await;
69 +
70 + assert_eq!(feed.busser_id, "rss");
71 + assert_eq!(feed.name, "My Feed");
72 + assert!(feed.enabled);
73 + assert!(feed.last_fetch.is_none());
74 +
75 + let fetched = FeedsRepository::new(pool.clone())
76 + .get(feed.id)
77 + .await
78 + .unwrap()
79 + .expect("feed should exist");
80 + assert_eq!(fetched.id, feed.id);
81 + assert_eq!(fetched.name, "My Feed");
82 + }
83 +
84 + #[tokio::test]
85 + async fn feeds_get_nonexistent_returns_none() {
86 + let pool = test_db().await;
87 + let result = FeedsRepository::new(pool.clone())
88 + .get(FeedId::new())
89 + .await
90 + .unwrap();
91 + assert!(result.is_none());
92 + }
93 +
94 + #[tokio::test]
95 + async fn feeds_list_all_returns_created() {
96 + let pool = test_db().await;
97 + make_feed(&pool, "rss", "Beta Feed").await;
98 + make_feed(&pool, "rss", "Alpha Feed").await;
99 +
100 + let all = FeedsRepository::new(pool.clone())
101 + .list_all()
102 + .await
103 + .unwrap();
104 + assert_eq!(all.len(), 2);
105 + assert_eq!(all[0].name, "Alpha Feed");
106 + assert_eq!(all[1].name, "Beta Feed");
107 + }
108 +
109 + #[tokio::test]
110 + async fn feeds_get_by_busser_filters() {
111 + let pool = test_db().await;
112 + make_feed(&pool, "rss", "RSS Feed").await;
113 + make_feed(&pool, "hn", "HN Feed").await;
114 + make_feed(&pool, "rss", "RSS Feed 2").await;
115 +
116 + let rss = FeedsRepository::new(pool.clone())
117 + .get_by_busser("rss")
118 + .await
119 + .unwrap();
120 + assert_eq!(rss.len(), 2);
121 + for f in &rss {
122 + assert_eq!(f.busser_id, "rss");
123 + }
124 +
125 + let hn = FeedsRepository::new(pool.clone())
126 + .get_by_busser("hn")
127 + .await
128 + .unwrap();
129 + assert_eq!(hn.len(), 1);
130 + assert_eq!(hn[0].busser_id, "hn");
131 + }
132 +
133 + #[tokio::test]
134 + async fn feeds_list_enabled_excludes_disabled() {
135 + let pool = test_db().await;
136 + let feeds_repo = FeedsRepository::new(pool.clone());
137 + let feed = make_feed(&pool, "rss", "Disabled Feed").await;
138 + make_feed(&pool, "rss", "Enabled Feed").await;
139 +
140 + feeds_repo.set_enabled(feed.id, false).await.unwrap();
141 +
142 + let enabled = feeds_repo.list_enabled().await.unwrap();
143 + assert_eq!(enabled.len(), 1);
144 + assert_eq!(enabled[0].name, "Enabled Feed");
145 + }
146 +
147 + #[tokio::test]
148 + async fn feeds_set_enabled_toggle() {
149 + let pool = test_db().await;
150 + let feeds_repo = FeedsRepository::new(pool.clone());
151 + let feed = make_feed(&pool, "rss", "Toggle Feed").await;
152 + assert!(feed.enabled);
153 +
154 + feeds_repo.set_enabled(feed.id, false).await.unwrap();
155 + let updated = feeds_repo.get(feed.id).await.unwrap().unwrap();
156 + assert!(!updated.enabled);
157 +
158 + feeds_repo.set_enabled(feed.id, true).await.unwrap();
159 + let updated = feeds_repo.get(feed.id).await.unwrap().unwrap();
160 + assert!(updated.enabled);
161 + }
162 +
163 + #[tokio::test]
164 + async fn feeds_update_last_fetch_sets_timestamp() {
165 + let pool = test_db().await;
166 + let feeds_repo = FeedsRepository::new(pool.clone());
167 + let feed = make_feed(&pool, "rss", "Fetch Feed").await;
168 + assert!(feed.last_fetch.is_none());
169 +
170 + feeds_repo.update_last_fetch(feed.id).await.unwrap();
171 + let updated = feeds_repo.get(feed.id).await.unwrap().unwrap();
172 + assert!(updated.last_fetch.is_some());
173 + }
174 +
175 + #[tokio::test]
176 + async fn feeds_delete_removes_feed() {
177 + let pool = test_db().await;
178 + let feeds_repo = FeedsRepository::new(pool.clone());
179 + let feed = make_feed(&pool, "rss", "Doomed Feed").await;
180 +
181 + feeds_repo.delete(feed.id).await.unwrap();
182 + let result = feeds_repo.get(feed.id).await.unwrap();
183 + assert!(result.is_none());
184 + }
185 +
186 + // ── ItemsRepository ───────────────────────────────────────────
187 +
188 + #[tokio::test]
189 + async fn items_upsert_and_get() {
190 + let pool = test_db().await;
191 + let feed = make_feed(&pool, "rss", "Feed").await;
192 + let item = make_item(&pool, &feed, "rss:1").await;
193 +
194 + assert_eq!(item.external_id, "rss:1");
195 + assert_eq!(item.feed_id, feed.id);
196 + assert_eq!(item.busser_id, "rss");
197 + assert_eq!(item.bite_author, "author");
198 + assert!(!item.is_read);
199 + assert!(!item.is_starred);
200 +
201 + let fetched = ItemsRepository::new(pool.clone())
202 + .get(item.id)
203 + .await
204 + .unwrap()
205 + .expect("item should exist");
206 + assert_eq!(fetched.id, item.id);
207 + }
208 +
209 + #[tokio::test]
210 + async fn items_upsert_conflict_updates() {
211 + let pool = test_db().await;
212 + let feed = make_feed(&pool, "rss", "Feed").await;
213 + let items_repo = ItemsRepository::new(pool.clone());
214 +
215 + let first = make_item(&pool, &feed, "rss:dup").await;
216 +
217 + let second = items_repo
218 + .upsert(CreateFeedItem {
219 + external_id: "rss:dup".to_string(),
220 + feed_id: feed.id,
221 + busser_id: feed.busser_id.clone(),
222 + bite_author: "updated_author".to_string(),
223 + bite_text: "Updated text".to_string(),
224 + bite_secondary: None,
225 + bite_indicator: None,
226 + title: Some("Updated Title".to_string()),
227 + body: None,
228 + url: None,
229 + media: vec![],
230 + published_at: Utc::now(),
231 + source_name: "test".to_string(),
232 + score: None,
233 + tags: vec![],
234 + actions: vec![],
235 + })
236 + .await
237 + .unwrap();
238 +
239 + assert_eq!(first.id, second.id);
240 + assert_eq!(second.bite_author, "updated_author");
241 + assert_eq!(second.bite_text, "Updated text");
242 + assert_eq!(items_repo.count_all().await.unwrap(), 1);
243 + }
244 +
245 + #[tokio::test]
246 + async fn items_get_by_external_id() {
247 + let pool = test_db().await;
248 + let feed = make_feed(&pool, "rss", "Feed").await;
249 + let item = make_item(&pool, &feed, "rss:ext1").await;
250 +
251 + let found = ItemsRepository::new(pool.clone())
252 + .get_by_external_id("rss:ext1")
253 + .await
254 + .unwrap()
255 + .expect("should find by external_id");
256 + assert_eq!(found.id, item.id);
257 + }
258 +
259 + #[tokio::test]
260 + async fn items_list_all_pagination() {
261 + let pool = test_db().await;
262 + let feed = make_feed(&pool, "rss", "Feed").await;
263 + let now = Utc::now();
264 +
265 + make_item_at(&pool, &feed, "p:1", now - Duration::hours(3)).await;
266 + make_item_at(&pool, &feed, "p:2", now - Duration::hours(2)).await;
267 + make_item_at(&pool, &feed, "p:3", now - Duration::hours(1)).await;
268 +
269 + let items_repo = ItemsRepository::new(pool.clone());
270 +
271 + let page1 = items_repo.list_all(2, 0).await.unwrap();
272 + assert_eq!(page1.len(), 2);
273 + assert_eq!(page1[0].external_id, "p:3");
274 + assert_eq!(page1[1].external_id, "p:2");
275 +
276 + let page2 = items_repo.list_all(2, 2).await.unwrap();
277 + assert_eq!(page2.len(), 1);
278 + assert_eq!(page2[0].external_id, "p:1");
279 + }
280 +
281 + #[tokio::test]
282 + async fn items_list_by_feed_filters() {
283 + let pool = test_db().await;
284 + let feed_a = make_feed(&pool, "rss", "Feed A").await;
285 + let feed_b = make_feed(&pool, "rss", "Feed B").await;
286 +
287 + make_item(&pool, &feed_a, "a:1").await;
288 + make_item(&pool, &feed_a, "a:2").await;
289 + make_item(&pool, &feed_b, "b:1").await;
290 +
291 + let items_repo = ItemsRepository::new(pool.clone());
292 +
293 + let a_items = items_repo.list_by_feed(feed_a.id, 100, 0).await.unwrap();
294 + assert_eq!(a_items.len(), 2);
295 + for i in &a_items {
296 + assert_eq!(i.feed_id, feed_a.id);
297 + }
298 +
299 + let b_items = items_repo.list_by_feed(feed_b.id, 100, 0).await.unwrap();
300 + assert_eq!(b_items.len(), 1);
301 + assert_eq!(b_items[0].feed_id, feed_b.id);
302 + }
303 +
304 + #[tokio::test]
305 + async fn items_list_by_busser_filters() {
306 + let pool = test_db().await;
307 + let feed_rss = make_feed(&pool, "rss", "RSS Feed").await;
308 + let feed_hn = make_feed(&pool, "hn", "HN Feed").await;
309 +
310 + make_item(&pool, &feed_rss, "rss:1").await;
311 + make_item(&pool, &feed_hn, "hn:1").await;
312 + make_item(&pool, &feed_hn, "hn:2").await;
313 +
314 + let items_repo = ItemsRepository::new(pool.clone());
315 +
316 + let rss = items_repo.list_by_busser("rss", 100, 0).await.unwrap();
317 + assert_eq!(rss.len(), 1);
318 +
319 + let hn = items_repo.list_by_busser("hn", 100, 0).await.unwrap();
320 + assert_eq!(hn.len(), 2);
321 + }
322 +
323 + #[tokio::test]
324 + async fn items_list_unread_excludes_read() {
325 + let pool = test_db().await;
326 + let feed = make_feed(&pool, "rss", "Feed").await;
327 + let items_repo = ItemsRepository::new(pool.clone());
328 +
329 + let item1 = make_item(&pool, &feed, "u:1").await;
330 + make_item(&pool, &feed, "u:2").await;
331 +
332 + items_repo.mark_read(item1.id, true).await.unwrap();
333 +
334 + let unread = items_repo.list_unread(100, 0).await.unwrap();
335 + assert_eq!(unread.len(), 1);
336 + assert_eq!(unread[0].external_id, "u:2");
337 + }
338 +
339 + #[tokio::test]
340 + async fn items_list_starred_only() {
341 + let pool = test_db().await;
342 + let feed = make_feed(&pool, "rss", "Feed").await;
343 + let items_repo = ItemsRepository::new(pool.clone());
344 +
345 + make_item(&pool, &feed, "s:1").await;
346 + let item2 = make_item(&pool, &feed, "s:2").await;
347 +
348 + items_repo.mark_starred(item2.id, true).await.unwrap();
349 +
350 + let starred = items_repo.list_starred(100, 0).await.unwrap();
351 + assert_eq!(starred.len(), 1);
352 + assert_eq!(starred[0].external_id, "s:2");
353 + }
354 +
355 + #[tokio::test]
356 + async fn items_mark_read_and_unread() {
357 + let pool = test_db().await;
358 + let feed = make_feed(&pool, "rss", "Feed").await;
359 + let items_repo = ItemsRepository::new(pool.clone());
360 + let item = make_item(&pool, &feed, "r:1").await;
361 +
362 + assert!(!item.is_read);
363 +
364 + items_repo.mark_read(item.id, true).await.unwrap();
365 + let updated = items_repo.get(item.id).await.unwrap().unwrap();
366 + assert!(updated.is_read);
367 +
368 + items_repo.mark_read(item.id, false).await.unwrap();
369 + let updated = items_repo.get(item.id).await.unwrap().unwrap();
370 + assert!(!updated.is_read);
371 + }
372 +
373 + #[tokio::test]
374 + async fn items_mark_starred_and_unstarred() {
375 + let pool = test_db().await;
376 + let feed = make_feed(&pool, "rss", "Feed").await;
377 + let items_repo = ItemsRepository::new(pool.clone());
378 + let item = make_item(&pool, &feed, "st:1").await;
379 +
380 + assert!(!item.is_starred);
381 +
382 + items_repo.mark_starred(item.id, true).await.unwrap();
383 + let updated = items_repo.get(item.id).await.unwrap().unwrap();
384 + assert!(updated.is_starred);
385 +
386 + items_repo.mark_starred(item.id, false).await.unwrap();
387 + let updated = items_repo.get(item.id).await.unwrap().unwrap();
388 + assert!(!updated.is_starred);
389 + }
390 +
391 + #[tokio::test]
392 + async fn items_count_all_and_unread() {
393 + let pool = test_db().await;
394 + let feed = make_feed(&pool, "rss", "Feed").await;
395 + let items_repo = ItemsRepository::new(pool.clone());
396 +
397 + let item1 = make_item(&pool, &feed, "c:1").await;
398 + make_item(&pool, &feed, "c:2").await;
399 + make_item(&pool, &feed, "c:3").await;
400 +
401 + assert_eq!(items_repo.count_all().await.unwrap(), 3);
402 + assert_eq!(items_repo.count_unread().await.unwrap(), 3);
403 +
404 + items_repo.mark_read(item1.id, true).await.unwrap();
405 + assert_eq!(items_repo.count_all().await.unwrap(), 3);
406 + assert_eq!(items_repo.count_unread().await.unwrap(), 2);
407 + }
408 +
409 + #[tokio::test]
410 + async fn items_delete_by_feed_removes_items() {
411 + let pool = test_db().await;
412 + let feed = make_feed(&pool, "rss", "Feed").await;
413 + let items_repo = ItemsRepository::new(pool.clone());
414 +
415 + make_item(&pool, &feed, "d:1").await;
416 + make_item(&pool, &feed, "d:2").await;
417 + assert_eq!(items_repo.count_all().await.unwrap(), 2);
418 +
419 + let removed = items_repo.delete_by_feed(feed.id).await.unwrap();
420 + assert_eq!(removed, 2);
421 + assert_eq!(items_repo.count_all().await.unwrap(), 0);
422 + }
423 +
424 + #[tokio::test]
425 + async fn items_delete_stale_read() {
426 + let pool = test_db().await;
427 + let feed = make_feed(&pool, "rss", "Feed").await;
428 + let items_repo = ItemsRepository::new(pool.clone());
429 + let now = Utc::now();
430 +
431 + // Old read item (should be deleted)
432 + let old_read = make_item_at(&pool, &feed, "stale:1", now - Duration::days(60)).await;
433 + items_repo.mark_read(old_read.id, true).await.unwrap();
434 +
435 + // Old read + starred item (should be preserved)
436 + let old_starred = make_item_at(&pool, &feed, "stale:2", now - Duration::days(60)).await;
437 + items_repo.mark_read(old_starred.id, true).await.unwrap();
438 + items_repo.mark_starred(old_starred.id, true).await.unwrap();
439 +
440 + // Old unread item (should be preserved)
441 + make_item_at(&pool, &feed, "stale:3", now - Duration::days(60)).await;
442 +
443 + // Recent read item (should be preserved)
444 + let recent_read = make_item_at(&pool, &feed, "stale:4", now - Duration::days(5)).await;
445 + items_repo.mark_read(recent_read.id, true).await.unwrap();
446 +
447 + // Recent unread item (should be preserved)
448 + make_item_at(&pool, &feed, "stale:5", now - Duration::days(5)).await;
449 +
450 + assert_eq!(items_repo.count_all().await.unwrap(), 5);
451 +
452 + let cutoff = now - Duration::days(30);
453 + let deleted = items_repo.delete_stale_read(cutoff).await.unwrap();
454 + assert_eq!(deleted, 1); // only old_read
455 +
456 + assert_eq!(items_repo.count_all().await.unwrap(), 4);
457 +
458 + // Verify the right items remain
459 + assert!(items_repo.get(old_starred.id).await.unwrap().is_some());
460 + assert!(items_repo.get(recent_read.id).await.unwrap().is_some());
461 + assert!(items_repo.get(old_read.id).await.unwrap().is_none());
462 + }
463 +
464 + // ── Feed Health ──────────────────────────────────────────────
465 +
466 + #[tokio::test]
467 + async fn health_success_resets_counter() {
468 + let pool = test_db().await;
469 + let feeds_repo = FeedsRepository::new(pool.clone());
470 + let feed = make_feed(&pool, "rss", "Feed").await;
471 +
472 + // Simulate two failures
473 + feeds_repo.record_fetch_failure(feed.id, "timeout").await.unwrap();
474 + feeds_repo.record_fetch_failure(feed.id, "timeout").await.unwrap();
475 +
476 + let f = feeds_repo.get(feed.id).await.unwrap().unwrap();
477 + assert_eq!(f.consecutive_failures, 2);
478 + assert_eq!(f.last_error.as_deref(), Some("timeout"));
479 +
480 + // Then a success
481 + feeds_repo.record_fetch_success(feed.id).await.unwrap();
482 + let f = feeds_repo.get(feed.id).await.unwrap().unwrap();
483 + assert_eq!(f.consecutive_failures, 0);
484 + assert!(f.last_error.is_none());
485 + assert!(f.last_success_at.is_some());
486 + }
487 +
488 + #[tokio::test]
489 + async fn health_failure_increments() {
490 + let pool = test_db().await;
491 + let feeds_repo = FeedsRepository::new(pool.clone());
492 + let feed = make_feed(&pool, "rss", "Feed").await;
493 +
494 + feeds_repo.record_fetch_failure(feed.id, "dns error").await.unwrap();
495 + let f = feeds_repo.get(feed.id).await.unwrap().unwrap();
496 + assert_eq!(f.consecutive_failures, 1);
497 + assert_eq!(f.last_error.as_deref(), Some("dns error"));
498 +
499 + feeds_repo.record_fetch_failure(feed.id, "connection refused").await.unwrap();
500 + let f = feeds_repo.get(feed.id).await.unwrap().unwrap();
Lines truncated