YoursFunny/TelegramTwitterMediaBot
1
0
Fork 0
mirror of https://github.com/TheFunny/TelegramTwitterMediaBot.git synced 2026-09-26 23:52:05 +00:00
Code Releases Activity

refactor(db): add DbPool::with_conn_or for the default-and-log tails

Browse Source 
Ten call sites across link_cache, state and queue repeated the same
match/if-let over a with_conn result with their own log line and default.
with_conn_or takes the level, the operation name and the default; each site
keeps its exact message and the same Ok/Err behaviour.
This commit is contained in:
YoursFunny
2026-09-21 17:24:51 +08:00
parent f21570f579
commit d9f3ee99c1
4 changed files with 176 additions and 167 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/xmedia-bot/src/db.rs
+20
View File
@@ -82,6 +82,26 @@ impl DbPool {
pub fn path(&self) -> &str {
&self.inner.path
}
/// [`with_conn`] for the many callers that answer a failed statement with
/// a default plus one log line: `what` names the operation and `level`
/// says how bad it is (`Error` when the failure loses work the caller
/// expected, `Warn` when the user is still served).
///
/// [`with_conn`]: DbPool::with_conn
pub async fn with_conn_or<T, F>(&self, level: log::Level, what: &str, default: T, f: F) -> T
where
T: Send + 'static,
F: FnOnce(&mut Connection) -> rusqlite::Result<T> + Send + 'static,
{
match self.with_conn(f).await {
Ok(value) => value,
Err(e) => {
log::log!(level, "{what}: {e}");
default
}
}
}
}
impl PoolInner {
crates/xmedia-bot/src/link_cache.rs
+79 -84
View File
@@ -74,115 +74,110 @@ impl LinkCache {
pub async fn get(&self, key: &str, ttl: Duration) -> Option<CachedPost> {
let key = key.to_string();
let ttl = ttl.as_secs_f64();
let result = self
.pool
.with_conn(move |conn| {
let Some((payload, created_at)) = conn
.query_row(
"SELECT payload, created_at FROM link_cache WHERE url = ?1",
params![key],
|row| Ok((row.get::<_, String>(0)?, row.get::<_, f64>(1)?)),
)
.optional()?
else {
return Ok(None);
};
if now_f64() - created_at > ttl {
conn.execute("DELETE FROM link_cache WHERE url = ?1", params![key])?;
return Ok(None);
}
match serde_json::from_str::<CachedPost>(&payload) {
Ok(post) => Ok(Some(post)),
Err(e) => {
// Unreadable payload (e.g. an older schema): drop it
// instead of re-failing the parse on every later hit.
self.pool
.with_conn_or(
log::Level::Warn,
"link cache read failed",
None,
move |conn| {
let Some((payload, created_at)) = conn
.query_row(
"SELECT payload, created_at FROM link_cache WHERE url = ?1",
params![key],
|row| Ok((row.get::<_, String>(0)?, row.get::<_, f64>(1)?)),
)
.optional()?
else {
return Ok(None);
};
if now_f64() - created_at > ttl {
conn.execute("DELETE FROM link_cache WHERE url = ?1", params![key])?;
Err(rusqlite::Error::ToSqlConversionFailure(Box::new(e)))
return Ok(None);
}
}
})
.await;
match result {
Ok(v) => v,
Err(e) => {
log::warn!("link cache read failed: {e}");
None
}
}
match serde_json::from_str::<CachedPost>(&payload) {
Ok(post) => Ok(Some(post)),
Err(e) => {
// Unreadable payload (e.g. an older schema): drop it
// instead of re-failing the parse on every later hit.
conn.execute("DELETE FROM link_cache WHERE url = ?1", params![key])?;
Err(rusqlite::Error::ToSqlConversionFailure(Box::new(e)))
}
}
},
)
.await
}
pub async fn put(&self, key: &str, post: &CachedPost) {
let key = key.to_string();
let payload = serde_json::to_string(post).expect("cached post serializes");
let result = self
.pool
.with_conn(move |conn| {
conn.execute(
"INSERT OR REPLACE INTO link_cache (url, payload, created_at) VALUES (?1, ?2, ?3)",
params![key, payload, now_f64()],
)?;
Ok(())
})
self.pool
.with_conn_or(
log::Level::Warn,
"link cache write failed",
(),
move |conn| {
conn.execute(
"INSERT OR REPLACE INTO link_cache (url, payload, created_at) VALUES (?1, ?2, ?3)",
params![key, payload, now_f64()],
)?;
Ok(())
},
)
.await;
if let Err(e) = result {
log::warn!("link cache write failed: {e}");
}
}
/// Drops an entry (e.g. a cached file id that turned out invalid).
pub async fn remove(&self, key: &str) {
let key = key.to_string();
let result = self
.pool
.with_conn(move |conn| {
conn.execute("DELETE FROM link_cache WHERE url = ?1", params![key])?;
Ok(())
})
self.pool
.with_conn_or(
log::Level::Warn,
"link cache delete failed",
(),
move |conn| {
conn.execute("DELETE FROM link_cache WHERE url = ?1", params![key])?;
Ok(())
},
)
.await;
if let Err(e) = result {
log::warn!("link cache delete failed: {e}");
}
}
/// Removes expired entries; returns how many were deleted.
pub async fn prune(&self, ttl: Duration) -> usize {
let cutoff = now_f64() - ttl.as_secs_f64();
let result = self
.pool
.with_conn(move |conn| {
conn.execute(
"DELETE FROM link_cache WHERE created_at < ?1",
params![cutoff],
)
})
.await;
match result {
Ok(n) => n,
Err(e) => {
log::warn!("link cache prune failed: {e}");
0
}
}
self.pool
.with_conn_or(
log::Level::Warn,
"link cache prune failed",
0,
move |conn| {
conn.execute(
"DELETE FROM link_cache WHERE created_at < ?1",
params![cutoff],
)
},
)
.await
}
/// Deletes one entry (by normalized cache key) or the whole cache when
/// `key` is `None`. Returns how many rows were removed.
pub async fn clear(&self, key: Option<&str>) -> usize {
let key = key.map(str::to_string);
let result = self
.pool
.with_conn(move |conn| match &key {
Some(key) => conn.execute("DELETE FROM link_cache WHERE url = ?1", params![key]),
None => conn.execute("DELETE FROM link_cache", []),
})
.await;
match result {
Ok(n) => n,
Err(e) => {
log::warn!("link cache clear failed: {e}");
0
}
}
self.pool
.with_conn_or(
log::Level::Warn,
"link cache clear failed",
0,
move |conn| match &key {
Some(key) => {
conn.execute("DELETE FROM link_cache WHERE url = ?1", params![key])
}
None => conn.execute("DELETE FROM link_cache", []),
},
)
.await
}
}
crates/xmedia-bot/src/queue.rs
+47 -56
View File
@@ -239,23 +239,20 @@ impl PersistentTaskQueue {
/// `in_progress`). The startup repair reads these before the workers start:
/// with no worker running, no row can be leased while it writes.
pub async fn runnable_rows(&self) -> Vec<(String, String)> {
let result = self
.pool
.with_conn(|conn| {
let mut stmt = conn.prepare(
"SELECT id, payload FROM tasks WHERE status IN ('pending', 'in_progress') ORDER BY run_after",
)?;
let rows = stmt.query_map([], |row| Ok((row.get(0)?, row.get(1)?)))?;
rows.collect::<rusqlite::Result<Vec<(String, String)>>>()
})
.await;
match result {
Ok(rows) => rows,
Err(e) => {
log::error!("queue row scan failed: {e}");
Vec::new()
}
}
self.pool
.with_conn_or(
log::Level::Error,
"queue row scan failed",
Vec::new(),
|conn| {
let mut stmt = conn.prepare(
"SELECT id, payload FROM tasks WHERE status IN ('pending', 'in_progress') ORDER BY run_after",
)?;
let rows = stmt.query_map([], |row| Ok((row.get(0)?, row.get(1)?)))?;
rows.collect::<rusqlite::Result<Vec<(String, String)>>>()
},
)
.await
}
/// Replaces a runnable row's payload and restarts its attempt budget: the
@@ -291,28 +288,25 @@ impl PersistentTaskQueue {
}
pub async fn pending_backlog(&self) -> Option<(i64, f64)> {
let result = self
.pool
.with_conn(|conn| {
let (count, oldest) = conn.query_row(
"SELECT COUNT(*), MIN(run_after) FROM tasks WHERE status='pending'",
[],
|row| Ok((row.get::<_, i64>(0)?, row.get::<_, Option<f64>>(1)?)),
)?;
// `MIN` over zero rows is NULL, so the count is what decides.
Ok(match oldest {
Some(oldest) if count > 0 => Some((count, oldest)),
_ => None,
})
})
.await;
match result {
Ok(v) => v,
Err(e) => {
log::error!("queue backlog query failed: {e}");
None
}
}
self.pool
.with_conn_or(
log::Level::Error,
"queue backlog query failed",
None,
|conn| {
let (count, oldest) = conn.query_row(
"SELECT COUNT(*), MIN(run_after) FROM tasks WHERE status='pending'",
[],
|row| Ok((row.get::<_, i64>(0)?, row.get::<_, Option<f64>>(1)?)),
)?;
// `MIN` over zero rows is NULL, so the count is what decides.
Ok(match oldest {
Some(oldest) if count > 0 => Some((count, oldest)),
_ => None,
})
},
)
.await
}
}
@@ -455,23 +449,20 @@ impl QueueWorker {
}
async fn earliest_run_after(&self) -> Option<f64> {
let result = self
.pool
.with_conn(|conn| {
conn.query_row(
"SELECT MIN(run_after) FROM tasks WHERE status='pending'",
[],
|row| row.get::<_, Option<f64>>(0),
)
})
.await;
match result {
Ok(v) => v,
Err(e) => {
log::error!("queue timing query failed: {e}");
None
}
}
self.pool
.with_conn_or(
log::Level::Error,
"queue timing query failed",
None,
|conn| {
conn.query_row(
"SELECT MIN(run_after) FROM tasks WHERE status='pending'",
[],
|row| row.get::<_, Option<f64>>(0),
)
},
)
.await
}
/// Processes one leased row, keeping the lease alive while the handler
crates/xmedia-bot/src/state.rs
+30 -27
View File
@@ -61,22 +61,24 @@ impl ChatStore {
let chat_key = chat_id.to_string();
let payload = self
.pool
.with_conn(move |conn| {
// Concurrent handler tasks (batch-forwards) may write chat_state
// while this read runs; the shared busy timeout handles the
// write-lock collision instead of failing the query.
conn.query_row(
"SELECT payload FROM chat_state WHERE chat_id = ?1",
params![chat_key],
|row| row.get::<_, String>(0),
)
.optional()
})
.with_conn_or(
log::Level::Warn,
"chat_state read failed",
None,
move |conn| {
// Concurrent handler tasks (batch-forwards) may write
// chat_state while this read runs; the shared busy timeout
// handles the write-lock collision instead of failing the
// query.
conn.query_row(
"SELECT payload FROM chat_state WHERE chat_id = ?1",
params![chat_key],
|row| row.get::<_, String>(0),
)
.optional()
},
)
.await
.unwrap_or_else(|e| {
log::warn!("chat_state read failed: {e}");
None
})
.unwrap_or_default();
let data: ChatData = serde_json::from_str(&payload).unwrap_or_default();
self.cache.lock().insert(chat_id, data.clone());
@@ -88,19 +90,20 @@ impl ChatStore {
self.cache.lock().insert(chat_id, data.clone());
let payload = serde_json::to_string(data).expect("chat state serializes");
let chat_id = chat_id.to_string();
let result = self
.pool
.with_conn(move |conn| {
conn.execute(
"INSERT OR REPLACE INTO chat_state (chat_id, payload) VALUES (?1, ?2)",
params![chat_id, payload],
)?;
Ok(())
})
self.pool
.with_conn_or(
log::Level::Warn,
"chat_state write failed",
(),
move |conn| {
conn.execute(
"INSERT OR REPLACE INTO chat_state (chat_id, payload) VALUES (?1, ?2)",
params![chat_id, payload],
)?;
Ok(())
},
)
.await;
if let Err(e) = result {
log::warn!("chat_state write failed: {e}");
}
}
/// The per-chat async lock serializing get→mutate→set cycles.