Skip to main content

max / balanced_breakfast

16.2 KB · 468 lines History Blame Raw
1 use super::{
2 CreateFeedItem, DateTime, DbFeedItem, FeedId, ItemId, MAX_SEARCH_QUERY_LENGTH, SqlitePool,
3 TIMESTAMP_FMT, Utc, sanitize_fts_query,
4 };
5
6 #[derive(Clone)]
7 /// Repository for feed item CRUD, read/star toggling, and paginated listing
8 pub struct ItemsRepository {
9 pool: SqlitePool,
10 }
11
12 impl ItemsRepository {
13 /// Create a new items 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 feed item or update it if `external_id` already exists.
20 ///
21 /// The ON CONFLICT clause deliberately preserves user state (`is_read`,
22 /// `is_starred`), these are never overwritten by a re-fetch. Content fields
23 /// (author, text, body, etc.) are updated in case the source edited the post.
24 #[tracing::instrument(skip_all)]
25 pub async fn upsert(&self, input: CreateFeedItem) -> Result<DbFeedItem, sqlx::Error> {
26 let id = ItemId::new();
27 let now = Utc::now().format(TIMESTAMP_FMT).to_string();
28 let published = input.published_at.format(TIMESTAMP_FMT).to_string();
29 // Vec<String> is always serializable; unwrap is safe here.
30 let media = serde_json::to_string(&input.media).unwrap_or_else(|_| "[]".to_string());
31 let tags = serde_json::to_string(&input.tags).unwrap_or_else(|_| "[]".to_string());
32 let actions = serde_json::to_string(&input.actions).unwrap_or_else(|_| "[]".to_string());
33
34 sqlx::query_as(
35 r"
36 INSERT INTO feed_items (
37 id, external_id, feed_id, busser_id,
38 bite_author, bite_text, bite_secondary, bite_indicator,
39 title, body, url, media, actions,
40 published_at, fetched_at, source_name, score, tags,
41 is_read, is_starred, created_at, updated_at
42 )
43 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18, 0, 0, ?19, ?19)
44 ON CONFLICT (external_id) DO UPDATE SET
45 bite_author = EXCLUDED.bite_author,
46 bite_text = EXCLUDED.bite_text,
47 bite_secondary = EXCLUDED.bite_secondary,
48 bite_indicator = EXCLUDED.bite_indicator,
49 title = EXCLUDED.title,
50 body = EXCLUDED.body,
51 url = EXCLUDED.url,
52 media = EXCLUDED.media,
53 actions = EXCLUDED.actions,
54 score = EXCLUDED.score,
55 tags = EXCLUDED.tags,
56 updated_at = EXCLUDED.updated_at
57 RETURNING *
58 ",
59 )
60 .bind(id)
61 .bind(&input.external_id)
62 .bind(input.feed_id)
63 .bind(input.busser_id.as_str())
64 .bind(&input.bite_author)
65 .bind(&input.bite_text)
66 .bind(&input.bite_secondary)
67 .bind(&input.bite_indicator)
68 .bind(&input.title)
69 .bind(&input.body)
70 .bind(&input.url)
71 .bind(&media)
72 .bind(&actions)
73 .bind(&published)
74 .bind(&now)
75 .bind(&input.source_name)
76 .bind(input.score)
77 .bind(&tags)
78 .bind(&now)
79 .fetch_one(&self.pool)
80 .await
81 }
82
83 /// Look up a feed item by its internal ID.
84 #[tracing::instrument(skip_all)]
85 pub async fn get(&self, id: ItemId) -> Result<Option<DbFeedItem>, sqlx::Error> {
86 sqlx::query_as("SELECT * FROM feed_items WHERE id = ?1")
87 .bind(id)
88 .fetch_optional(&self.pool)
89 .await
90 }
91
92 /// Look up a feed item by the `external_id` produced by the busser.
93 #[tracing::instrument(skip_all)]
94 pub async fn get_by_external_id(
95 &self,
96 external_id: &str,
97 ) -> Result<Option<DbFeedItem>, sqlx::Error> {
98 sqlx::query_as("SELECT * FROM feed_items WHERE external_id = ?1")
99 .bind(external_id)
100 .fetch_optional(&self.pool)
101 .await
102 }
103
104 /// List all items ordered by `published_at` descending, with pagination.
105 /// Sorted by publish date (not fetch date) so items appear where the author
106 /// intended, even if fetched out of order during backfill.
107 #[tracing::instrument(skip_all)]
108 pub async fn list_all(&self, limit: i64, offset: i64) -> Result<Vec<DbFeedItem>, sqlx::Error> {
109 sqlx::query_as("SELECT * FROM feed_items ORDER BY published_at DESC LIMIT ?1 OFFSET ?2")
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(
155 &self,
156 limit: i64,
157 offset: i64,
158 ) -> Result<Vec<DbFeedItem>, sqlx::Error> {
159 sqlx::query_as(
160 "SELECT * FROM feed_items WHERE is_read = 0 ORDER BY published_at DESC LIMIT ?1 OFFSET ?2",
161 )
162 .bind(limit)
163 .bind(offset)
164 .fetch_all(&self.pool)
165 .await
166 }
167
168 /// List starred items only, newest first.
169 #[tracing::instrument(skip_all)]
170 pub async fn list_starred(
171 &self,
172 limit: i64,
173 offset: i64,
174 ) -> Result<Vec<DbFeedItem>, sqlx::Error> {
175 sqlx::query_as(
176 "SELECT * FROM feed_items WHERE is_starred = 1 ORDER BY published_at DESC LIMIT ?1 OFFSET ?2",
177 )
178 .bind(limit)
179 .bind(offset)
180 .fetch_all(&self.pool)
181 .await
182 }
183
184 /// Full-text search using FTS5 across title, body, and bite_text.
185 ///
186 /// Uses the `feed_items_fts` virtual table for fast ranked search.
187 /// Results are ordered by FTS5 relevance then by `published_at DESC`.
188 /// Additional boolean filters (source, unread, starred) are combined.
189 #[tracing::instrument(skip_all)]
190 pub async fn list_search(
191 &self,
192 query: &str,
193 source: Option<&str>,
194 unread_only: bool,
195 starred_only: bool,
196 limit: i64,
197 offset: i64,
198 ) -> Result<Vec<DbFeedItem>, sqlx::Error> {
199 // Reject excessively long queries before any processing.
200 if query.len() > MAX_SEARCH_QUERY_LENGTH {
201 return Ok(vec![]);
202 }
203
204 let fts_query = sanitize_fts_query(query);
205
206 // If sanitization stripped everything (e.g. query was just `*` or `^`),
207 // return early, an empty MATCH expression would be a SQL error.
208 if fts_query.is_empty() {
209 return Ok(vec![]);
210 }
211
212 let mut sql = String::from(
213 "SELECT fi.* FROM feed_items fi \
214 INNER JOIN feed_items_fts fts ON fi.rowid = fts.rowid \
215 WHERE feed_items_fts MATCH ?1",
216 );
217 if source.is_some() {
218 sql.push_str(" AND fi.busser_id = ?4");
219 }
220 if unread_only {
221 sql.push_str(" AND fi.is_read = 0");
222 }
223 if starred_only {
224 sql.push_str(" AND fi.is_starred = 1");
225 }
226 sql.push_str(" ORDER BY fts.rank, fi.published_at DESC LIMIT ?2 OFFSET ?3");
227
228 let mut q = sqlx::query_as::<_, DbFeedItem>(sqlx::AssertSqlSafe(sql.as_str()))
229 .bind(&fts_query) // ?1
230 .bind(limit) // ?2
231 .bind(offset); // ?3
232
233 if let Some(src) = source {
234 q = q.bind(src); // ?4
235 }
236
237 q.fetch_all(&self.pool).await
238 }
239
240 /// List feed items with any combination of filters pushed into SQL.
241 ///
242 /// Unifies `list_all`, `list_by_busser`, `list_unread`, `list_starred`,
243 /// and `list_search` into a single method so callers don't need an
244 /// if/else chain that drops filter combinations.
245 #[tracing::instrument(skip_all)]
246 pub async fn list_filtered(
247 &self,
248 search: Option<&str>,
249 source: Option<&str>,
250 unread_only: bool,
251 starred_only: bool,
252 limit: i64,
253 offset: i64,
254 ) -> Result<Vec<DbFeedItem>, sqlx::Error> {
255 // When a search query is present, use FTS5.
256 if let Some(query) = search {
257 return self
258 .list_search(query, source, unread_only, starred_only, limit, offset)
259 .await;
260 }
261
262 // Build a dynamic query with conditional WHERE clauses.
263 let mut sql = String::from("SELECT * FROM feed_items WHERE 1=1");
264 if source.is_some() {
265 sql.push_str(" AND busser_id = ?3");
266 }
267 if unread_only {
268 sql.push_str(" AND is_read = 0");
269 }
270 if starred_only {
271 sql.push_str(" AND is_starred = 1");
272 }
273 sql.push_str(" ORDER BY published_at DESC LIMIT ?1 OFFSET ?2");
274
275 let mut q = sqlx::query_as::<_, DbFeedItem>(sqlx::AssertSqlSafe(sql.as_str()))
276 .bind(limit) // ?1
277 .bind(offset); // ?2
278
279 if let Some(src) = source {
280 q = q.bind(src); // ?3
281 }
282
283 q.fetch_all(&self.pool).await
284 }
285
286 /// Set the read flag on a feed item.
287 #[tracing::instrument(skip_all)]
288 pub async fn mark_read(&self, id: ItemId, is_read: bool) -> Result<(), sqlx::Error> {
289 let now = Utc::now().format(TIMESTAMP_FMT).to_string();
290 sqlx::query("UPDATE feed_items SET is_read = ?1, updated_at = ?2 WHERE id = ?3")
291 .bind(is_read)
292 .bind(&now)
293 .bind(id)
294 .execute(&self.pool)
295 .await?;
296 Ok(())
297 }
298
299 /// Set the starred flag on a feed item.
300 #[tracing::instrument(skip_all)]
301 pub async fn mark_starred(&self, id: ItemId, is_starred: bool) -> Result<(), sqlx::Error> {
302 let now = Utc::now().format(TIMESTAMP_FMT).to_string();
303 sqlx::query("UPDATE feed_items SET is_starred = ?1, updated_at = ?2 WHERE id = ?3")
304 .bind(is_starred)
305 .bind(&now)
306 .bind(id)
307 .execute(&self.pool)
308 .await?;
309 Ok(())
310 }
311
312 /// Mark all unread items as read, optionally filtered to a specific source.
313 #[tracing::instrument(skip_all)]
314 pub async fn mark_all_read(&self, busser_id: Option<&str>) -> Result<u64, sqlx::Error> {
315 let now = Utc::now().format(TIMESTAMP_FMT).to_string();
316 let result = match busser_id {
317 Some(id) => {
318 sqlx::query(
319 "UPDATE feed_items SET is_read = 1, updated_at = ?1 WHERE is_read = 0 AND busser_id = ?2",
320 )
321 .bind(&now)
322 .bind(id)
323 .execute(&self.pool)
324 .await?
325 }
326 None => {
327 sqlx::query(
328 "UPDATE feed_items SET is_read = 1, updated_at = ?1 WHERE is_read = 0",
329 )
330 .bind(&now)
331 .execute(&self.pool)
332 .await?
333 }
334 };
335 Ok(result.rows_affected())
336 }
337
338 /// Count all feed items.
339 #[tracing::instrument(skip_all)]
340 pub async fn count_all(&self) -> Result<i64, sqlx::Error> {
341 let row: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM feed_items")
342 .fetch_one(&self.pool)
343 .await?;
344 Ok(row.0)
345 }
346
347 /// Count items from a specific busser source.
348 #[tracing::instrument(skip_all)]
349 pub async fn count_by_busser(&self, busser_id: &str) -> Result<i64, sqlx::Error> {
350 let row: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM feed_items WHERE busser_id = ?1")
351 .bind(busser_id)
352 .fetch_one(&self.pool)
353 .await?;
354 Ok(row.0)
355 }
356
357 /// Count unread items across all sources.
358 #[tracing::instrument(skip_all)]
359 pub async fn count_unread(&self) -> Result<i64, sqlx::Error> {
360 let row: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM feed_items WHERE is_read = 0")
361 .fetch_one(&self.pool)
362 .await?;
363 Ok(row.0)
364 }
365
366 /// Count starred items.
367 #[tracing::instrument(skip_all)]
368 pub async fn count_starred(&self) -> Result<i64, sqlx::Error> {
369 let row: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM feed_items WHERE is_starred = 1")
370 .fetch_one(&self.pool)
371 .await?;
372 Ok(row.0)
373 }
374
375 /// Count unread items from a specific busser source.
376 #[tracing::instrument(skip_all)]
377 pub async fn count_unread_by_busser(&self, busser_id: &str) -> Result<i64, sqlx::Error> {
378 let row: (i64,) =
379 sqlx::query_as("SELECT COUNT(*) FROM feed_items WHERE busser_id = ?1 AND is_read = 0")
380 .bind(busser_id)
381 .fetch_one(&self.pool)
382 .await?;
383 Ok(row.0)
384 }
385
386 /// Count items matching a combination of source, unread, and starred filters.
387 #[tracing::instrument(skip_all)]
388 pub async fn count_filtered(
389 &self,
390 source: Option<&str>,
391 unread_only: bool,
392 starred_only: bool,
393 ) -> Result<i64, sqlx::Error> {
394 let mut sql = "SELECT COUNT(*) FROM feed_items WHERE 1=1".to_string();
395 if source.is_some() {
396 sql.push_str(" AND busser_id = ?1");
397 }
398 if unread_only {
399 sql.push_str(" AND is_read = 0");
400 }
401 if starred_only {
402 sql.push_str(" AND is_starred = 1");
403 }
404 let mut query = sqlx::query_as::<_, (i64,)>(sqlx::AssertSqlSafe(sql.as_str()));
405 if let Some(s) = source {
406 query = query.bind(s);
407 }
408 let row = query.fetch_one(&self.pool).await?;
409 Ok(row.0)
410 }
411
412 /// Get total and unread counts for all busser sources in a single query.
413 ///
414 /// Returns `(busser_id, total_count, unread_count)` tuples. This avoids
415 /// issuing separate count queries per source (N+1 problem).
416 pub async fn counts_by_busser(&self) -> Result<Vec<(String, i64, i64)>, sqlx::Error> {
417 let rows: Vec<(String, i64, i64)> = sqlx::query_as(
418 r"
419 SELECT busser_id,
420 COUNT(*) AS total_count,
421 SUM(CASE WHEN is_read = 0 THEN 1 ELSE 0 END) AS unread_count
422 FROM feed_items
423 GROUP BY busser_id
424 ",
425 )
426 .fetch_all(&self.pool)
427 .await?;
428 Ok(rows)
429 }
430
431 /// Delete a single feed item by ID.
432 #[tracing::instrument(skip_all)]
433 pub async fn delete(&self, id: ItemId) -> Result<(), sqlx::Error> {
434 sqlx::query("DELETE FROM feed_items WHERE id = ?1")
435 .bind(id)
436 .execute(&self.pool)
437 .await?;
438 Ok(())
439 }
440
441 /// Delete all items belonging to a feed. Returns the number of rows removed.
442 #[tracing::instrument(skip_all)]
443 pub async fn delete_by_feed(&self, feed_id: FeedId) -> Result<u64, sqlx::Error> {
444 let result = sqlx::query("DELETE FROM feed_items WHERE feed_id = ?1")
445 .bind(feed_id)
446 .execute(&self.pool)
447 .await?;
448 Ok(result.rows_affected())
449 }
450
451 /// Delete old read items that are not starred. Returns the number of rows removed.
452 ///
453 /// Items published before `before` that have been read and are not starred
454 /// will be permanently deleted. Starred items are always preserved regardless
455 /// of age or read status.
456 #[tracing::instrument(skip_all)]
457 pub async fn delete_stale_read(&self, before: DateTime<Utc>) -> Result<u64, sqlx::Error> {
458 let cutoff = before.format(TIMESTAMP_FMT).to_string();
459 let result = sqlx::query(
460 "DELETE FROM feed_items WHERE is_read = 1 AND is_starred = 0 AND published_at < ?1",
461 )
462 .bind(&cutoff)
463 .execute(&self.pool)
464 .await?;
465 Ok(result.rows_affected())
466 }
467 }
468