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

style: rustfmt the DbPool call sites from the connection-pool change

Browse Source 
Formatting-only; the pool commit landed before cargo fmt was run.
This commit is contained in:
YoursFunny
2026-08-13 22:25:13 +08:00
parent 1747d321d8
commit 2e2d1b3506
3 changed files with 99 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/xmedia-bot/src/link_cache.rs
+50 -40
View File
@@ -70,24 +70,26 @@ impl LinkCache {
pub async fn get(&self, key: &str, ttl: Duration) -> Option<CachedPost> { pub async fn get(&self, key: &str, ttl: Duration) -> Option<CachedPost> {
let key = key.to_string(); let key = key.to_string();
let ttl = ttl.as_secs_f64(); let ttl = ttl.as_secs_f64();
let result = self.pool.with_conn(move |conn| { let result = self
let mut stmt = .pool
conn.prepare("SELECT payload, created_at FROM link_cache WHERE url = ?1")?; .with_conn(move |conn| {
let mut rows = stmt.query(params![key])?; let mut stmt =
let Some(row) = rows.next()? else { conn.prepare("SELECT payload, created_at FROM link_cache WHERE url = ?1")?;
return Ok(None); let mut rows = stmt.query(params![key])?;
}; let Some(row) = rows.next()? else {
let payload: String = row.get(0)?; return Ok(None);
let created_at: f64 = row.get(1)?; };
if now_f64() - created_at > ttl { let payload: String = row.get(0)?;
conn.execute("DELETE FROM link_cache WHERE url = ?1", params![key])?; let created_at: f64 = row.get(1)?;
return Ok(None); if now_f64() - created_at > ttl {
} conn.execute("DELETE FROM link_cache WHERE url = ?1", params![key])?;
Ok(Some(serde_json::from_str::<CachedPost>(&payload).map_err( return Ok(None);
|e| rusqlite::Error::ToSqlConversionFailure(Box::new(e)), }
)?)) Ok(Some(serde_json::from_str::<CachedPost>(&payload).map_err(
}) |e| rusqlite::Error::ToSqlConversionFailure(Box::new(e)),
.await; )?))
})
.await;
match result { match result {
Ok(v) => v, Ok(v) => v,
Err(e) => { Err(e) => {
@@ -100,14 +102,16 @@ impl LinkCache {
pub async fn put(&self, key: &str, post: &CachedPost) { pub async fn put(&self, key: &str, post: &CachedPost) {
let key = key.to_string(); let key = key.to_string();
let payload = serde_json::to_string(post).expect("cached post serializes"); let payload = serde_json::to_string(post).expect("cached post serializes");
let result = self.pool.with_conn(move |conn| { let result = self
conn.execute( .pool
.with_conn(move |conn| {
conn.execute(
"INSERT OR REPLACE INTO link_cache (url, payload, created_at) VALUES (?1, ?2, ?3)", "INSERT OR REPLACE INTO link_cache (url, payload, created_at) VALUES (?1, ?2, ?3)",
params![key, payload, now_f64()], params![key, payload, now_f64()],
)?; )?;
Ok(()) Ok(())
}) })
.await; .await;
if let Err(e) = result { if let Err(e) = result {
log::error!("link cache write failed: {e}"); log::error!("link cache write failed: {e}");
} }
@@ -116,11 +120,13 @@ impl LinkCache {
/// Drops an entry (e.g. a cached file id that turned out invalid). /// Drops an entry (e.g. a cached file id that turned out invalid).
pub async fn remove(&self, key: &str) { pub async fn remove(&self, key: &str) {
let key = key.to_string(); let key = key.to_string();
let result = self.pool.with_conn(move |conn| { let result = self
conn.execute("DELETE FROM link_cache WHERE url = ?1", params![key])?; .pool
Ok(()) .with_conn(move |conn| {
}) conn.execute("DELETE FROM link_cache WHERE url = ?1", params![key])?;
.await; Ok(())
})
.await;
if let Err(e) = result { if let Err(e) = result {
log::error!("link cache delete failed: {e}"); log::error!("link cache delete failed: {e}");
} }
@@ -129,13 +135,15 @@ impl LinkCache {
/// Removes expired entries; returns how many were deleted. /// Removes expired entries; returns how many were deleted.
pub async fn prune(&self, ttl: Duration) -> usize { pub async fn prune(&self, ttl: Duration) -> usize {
let cutoff = now_f64() - ttl.as_secs_f64(); let cutoff = now_f64() - ttl.as_secs_f64();
let result = self.pool.with_conn(move |conn| { let result = self
conn.execute( .pool
"DELETE FROM link_cache WHERE created_at < ?1", .with_conn(move |conn| {
params![cutoff], conn.execute(
) "DELETE FROM link_cache WHERE created_at < ?1",
}) params![cutoff],
.await; )
})
.await;
match result { match result {
Ok(n) => n, Ok(n) => n,
Err(e) => { Err(e) => {
@@ -149,11 +157,13 @@ impl LinkCache {
/// `key` is `None`. Returns how many rows were removed. /// `key` is `None`. Returns how many rows were removed.
pub async fn clear(&self, key: Option<&str>) -> usize { pub async fn clear(&self, key: Option<&str>) -> usize {
let key = key.map(str::to_string); let key = key.map(str::to_string);
let result = self.pool.with_conn(move |conn| match &key { let result = self
Some(key) => conn.execute("DELETE FROM link_cache WHERE url = ?1", params![key]), .pool
None => conn.execute("DELETE FROM link_cache", []), .with_conn(move |conn| match &key {
}) Some(key) => conn.execute("DELETE FROM link_cache WHERE url = ?1", params![key]),
.await; None => conn.execute("DELETE FROM link_cache", []),
})
.await;
match result { match result {
Ok(n) => n, Ok(n) => n,
Err(e) => { Err(e) => {
crates/xmedia-bot/src/queue.rs
+20 -17
View File
@@ -160,8 +160,7 @@ impl PersistentTaskQueue {
if sweep_stop.load(Ordering::Relaxed) { if sweep_stop.load(Ordering::Relaxed) {
break; break;
} }
let result = let result = sweep_pool.with_conn(move |conn| recover_update(conn)).await;
sweep_pool.with_conn(move |conn| recover_update(conn)).await;
if let Err(e) = result { if let Err(e) = result {
log::error!("queue sweep failed: {e}"); log::error!("queue sweep failed: {e}");
} }
@@ -309,16 +308,18 @@ impl QueueWorker {
} }
async fn earliest_run_after(&self) -> Option<f64> { async fn earliest_run_after(&self) -> Option<f64> {
let result = self.pool.with_conn(|conn| { let result = self
let mut stmt = .pool
conn.prepare("SELECT MIN(run_after) FROM tasks WHERE status='pending'")?; .with_conn(|conn| {
let mut rows = stmt.query([])?; let mut stmt =
match rows.next()? { conn.prepare("SELECT MIN(run_after) FROM tasks WHERE status='pending'")?;
Some(row) => Ok(row.get::<_, Option<f64>>(0)?), let mut rows = stmt.query([])?;
None => Ok(None), match rows.next()? {
} Some(row) => Ok(row.get::<_, Option<f64>>(0)?),
}) None => Ok(None),
.await; }
})
.await;
match result { match result {
Ok(v) => v, Ok(v) => v,
Err(e) => { Err(e) => {
@@ -374,11 +375,13 @@ impl QueueWorker {
async fn delete_row(&self, id: &str) { async fn delete_row(&self, id: &str) {
let id = id.to_string(); let id = id.to_string();
let result = self.pool.with_conn(move |conn| { let result = self
conn.execute("DELETE FROM tasks WHERE id = ?1", params![id])?; .pool
Ok(()) .with_conn(move |conn| {
}) conn.execute("DELETE FROM tasks WHERE id = ?1", params![id])?;
.await; Ok(())
})
.await;
if let Err(e) = result { if let Err(e) = result {
log::error!("queue delete failed: {e}"); log::error!("queue delete failed: {e}");
} }
crates/xmedia-bot/src/state.rs
+29 -25
View File
@@ -76,23 +76,25 @@ impl ChatStore {
return data.clone(); return data.clone();
} }
let chat_key = chat_id.to_string(); let chat_key = chat_id.to_string();
let payload = self.pool.with_conn(move |conn| { let payload = self
// Concurrent handler tasks (batch-forwards) may write chat_state .pool
// while this read runs; the shared busy timeout handles the .with_conn(move |conn| {
// write-lock collision instead of failing the query. // Concurrent handler tasks (batch-forwards) may write chat_state
let mut stmt = conn.prepare("SELECT payload FROM chat_state WHERE chat_id = ?1")?; // while this read runs; the shared busy timeout handles the
let mut rows = stmt.query(params![chat_key])?; // write-lock collision instead of failing the query.
match rows.next()? { let mut stmt = conn.prepare("SELECT payload FROM chat_state WHERE chat_id = ?1")?;
Some(row) => Ok(Some(row.get::<_, String>(0)?)), let mut rows = stmt.query(params![chat_key])?;
None => Ok(None), match rows.next()? {
} Some(row) => Ok(Some(row.get::<_, String>(0)?)),
}) None => Ok(None),
.await }
.unwrap_or_else(|e| { })
log::error!("chat_state read failed: {e}"); .await
None .unwrap_or_else(|e| {
}) log::error!("chat_state read failed: {e}");
.unwrap_or_default(); None
})
.unwrap_or_default();
let data: ChatData = serde_json::from_str(&payload).unwrap_or_default(); let data: ChatData = serde_json::from_str(&payload).unwrap_or_default();
self.cache.lock().insert(chat_id, data.clone()); self.cache.lock().insert(chat_id, data.clone());
data data
@@ -103,14 +105,16 @@ impl ChatStore {
self.cache.lock().insert(chat_id, data.clone()); self.cache.lock().insert(chat_id, data.clone());
let payload = serde_json::to_string(data).expect("chat state serializes"); let payload = serde_json::to_string(data).expect("chat state serializes");
let chat_id = chat_id.to_string(); let chat_id = chat_id.to_string();
let result = self.pool.with_conn(move |conn| { let result = self
conn.execute( .pool
"INSERT OR REPLACE INTO chat_state (chat_id, payload) VALUES (?1, ?2)", .with_conn(move |conn| {
params![chat_id, payload], conn.execute(
)?; "INSERT OR REPLACE INTO chat_state (chat_id, payload) VALUES (?1, ?2)",
Ok(()) params![chat_id, payload],
}) )?;
.await; Ok(())
})
.await;
if let Err(e) = result { if let Err(e) = result {
log::error!("chat_state write failed: {e}"); log::error!("chat_state write failed: {e}");
} }