mirror of
https://github.com/TheFunny/TelegramTwitterMediaBot.git
synced 2026-09-25 23:52:04 +00:00
refactor(db): use rusqlite query_row/optional instead of hand-rolled reads
pending_backlog, earliest_run_after, ChatStore::get and LinkCache::get each hand-rolled prepare + query + rows.next() for what is a single-row read. query_row + OptionalExtension::optional is the same statement and the same error mapping with less scaffolding; the backlog's NULL-on-empty MIN still goes through the count check, so a pending row with a NULL run_after is not misread. Also fixes the rustfmt drift from the previous commit.
This commit is contained in:
@@ -9,6 +9,7 @@
|
|||||||
//! by the periodic prune in `main`.
|
//! by the periodic prune in `main`.
|
||||||
|
|
||||||
use crate::db::now_f64;
|
use crate::db::now_f64;
|
||||||
|
use rusqlite::OptionalExtension;
|
||||||
use rusqlite::params;
|
use rusqlite::params;
|
||||||
use serde::{Deserialize, Serialize};
|
use serde::{Deserialize, Serialize};
|
||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
@@ -76,14 +77,16 @@ impl LinkCache {
|
|||||||
let result = self
|
let result = self
|
||||||
.pool
|
.pool
|
||||||
.with_conn(move |conn| {
|
.with_conn(move |conn| {
|
||||||
let mut stmt =
|
let Some((payload, created_at)) = conn
|
||||||
conn.prepare("SELECT payload, created_at FROM link_cache WHERE url = ?1")?;
|
.query_row(
|
||||||
let mut rows = stmt.query(params![key])?;
|
"SELECT payload, created_at FROM link_cache WHERE url = ?1",
|
||||||
let Some(row) = rows.next()? else {
|
params![key],
|
||||||
|
|row| Ok((row.get::<_, String>(0)?, row.get::<_, f64>(1)?)),
|
||||||
|
)
|
||||||
|
.optional()?
|
||||||
|
else {
|
||||||
return Ok(None);
|
return Ok(None);
|
||||||
};
|
};
|
||||||
let payload: String = row.get(0)?;
|
|
||||||
let created_at: f64 = row.get(1)?;
|
|
||||||
if now_f64() - created_at > ttl {
|
if now_f64() - created_at > ttl {
|
||||||
conn.execute("DELETE FROM link_cache WHERE url = ?1", params![key])?;
|
conn.execute("DELETE FROM link_cache WHERE url = ?1", params![key])?;
|
||||||
return Ok(None);
|
return Ok(None);
|
||||||
|
|||||||
@@ -294,19 +294,16 @@ impl PersistentTaskQueue {
|
|||||||
let result = self
|
let result = self
|
||||||
.pool
|
.pool
|
||||||
.with_conn(|conn| {
|
.with_conn(|conn| {
|
||||||
let mut stmt = conn
|
let (count, oldest) = conn.query_row(
|
||||||
.prepare("SELECT COUNT(*), MIN(run_after) FROM tasks WHERE status='pending'")?;
|
"SELECT COUNT(*), MIN(run_after) FROM tasks WHERE status='pending'",
|
||||||
let mut rows = stmt.query([])?;
|
[],
|
||||||
match rows.next()? {
|
|row| Ok((row.get::<_, i64>(0)?, row.get::<_, Option<f64>>(1)?)),
|
||||||
Some(row) => {
|
)?;
|
||||||
let count = row.get::<_, i64>(0)?;
|
// `MIN` over zero rows is NULL, so the count is what decides.
|
||||||
match row.get::<_, Option<f64>>(1)? {
|
Ok(match oldest {
|
||||||
Some(oldest) if count > 0 => Ok(Some((count, oldest))),
|
Some(oldest) if count > 0 => Some((count, oldest)),
|
||||||
_ => Ok(None),
|
_ => None,
|
||||||
}
|
})
|
||||||
}
|
|
||||||
None => Ok(None),
|
|
||||||
}
|
|
||||||
})
|
})
|
||||||
.await;
|
.await;
|
||||||
match result {
|
match result {
|
||||||
@@ -461,13 +458,11 @@ impl QueueWorker {
|
|||||||
let result = self
|
let result = self
|
||||||
.pool
|
.pool
|
||||||
.with_conn(|conn| {
|
.with_conn(|conn| {
|
||||||
let mut stmt =
|
conn.query_row(
|
||||||
conn.prepare("SELECT MIN(run_after) FROM tasks WHERE status='pending'")?;
|
"SELECT MIN(run_after) FROM tasks WHERE status='pending'",
|
||||||
let mut rows = stmt.query([])?;
|
[],
|
||||||
match rows.next()? {
|
|row| row.get::<_, Option<f64>>(0),
|
||||||
Some(row) => Ok(row.get::<_, Option<f64>>(0)?),
|
)
|
||||||
None => Ok(None),
|
|
||||||
}
|
|
||||||
})
|
})
|
||||||
.await;
|
.await;
|
||||||
match result {
|
match result {
|
||||||
|
|||||||
@@ -170,7 +170,7 @@ pub(super) async fn prepare_upload_item(
|
|||||||
let media_url = item_url(&item);
|
let media_url = item_url(&item);
|
||||||
if !media_url.starts_with("http://") && !media_url.starts_with("https://") {
|
if !media_url.starts_with("http://") && !media_url.starts_with("https://") {
|
||||||
let media = media_from_file(&item, std::path::PathBuf::from(media_url), caption)
|
let media = media_from_file(&item, std::path::PathBuf::from(media_url), caption)
|
||||||
.map_err(|message| FallbackError::Permanent { message })?;
|
.map_err(|message| FallbackError::Permanent { message })?;
|
||||||
return Ok(PreparedItem {
|
return Ok(PreparedItem {
|
||||||
index,
|
index,
|
||||||
media,
|
media,
|
||||||
|
|||||||
@@ -3,6 +3,7 @@
|
|||||||
|
|
||||||
use crate::db::unix_now;
|
use crate::db::unix_now;
|
||||||
use parking_lot::Mutex;
|
use parking_lot::Mutex;
|
||||||
|
use rusqlite::OptionalExtension;
|
||||||
use rusqlite::params;
|
use rusqlite::params;
|
||||||
use serde::{Deserialize, Serialize};
|
use serde::{Deserialize, Serialize};
|
||||||
use std::collections::HashMap;
|
use std::collections::HashMap;
|
||||||
@@ -64,12 +65,12 @@ impl ChatStore {
|
|||||||
// Concurrent handler tasks (batch-forwards) may write chat_state
|
// Concurrent handler tasks (batch-forwards) may write chat_state
|
||||||
// while this read runs; the shared busy timeout handles the
|
// while this read runs; the shared busy timeout handles the
|
||||||
// write-lock collision instead of failing the query.
|
// write-lock collision instead of failing the query.
|
||||||
let mut stmt = conn.prepare("SELECT payload FROM chat_state WHERE chat_id = ?1")?;
|
conn.query_row(
|
||||||
let mut rows = stmt.query(params![chat_key])?;
|
"SELECT payload FROM chat_state WHERE chat_id = ?1",
|
||||||
match rows.next()? {
|
params![chat_key],
|
||||||
Some(row) => Ok(Some(row.get::<_, String>(0)?)),
|
|row| row.get::<_, String>(0),
|
||||||
None => Ok(None),
|
)
|
||||||
}
|
.optional()
|
||||||
})
|
})
|
||||||
.await
|
.await
|
||||||
.unwrap_or_else(|e| {
|
.unwrap_or_else(|e| {
|
||||||
|
|||||||
Reference in New Issue
Block a user