alemi/upub
alemi/upub
1
0
Fork 1
Code Issues 4 Pull requests Projects Releases Packages 1 Wiki Activity Actions

fix: db fix tasks would get stuck on sqlite

Browse source 
i think because the db would end up locked, added brackets to be sure to
drop the streaming reference. also it could crash when updating records
not found, just spit a warn
This commit is contained in:
əlemi 2024-04-29 21:23:12 +02:00
parent e4b92584db
commit a6e12468b7
Signed by: alemi
GPG key ID: A4895B84D311642C
1 changed files with 35 additions and 20 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

33
src/main.rs
View file

@ -239,46 +239,57 @@ async fn fix(db: sea_orm::DatabaseConnection, likes: bool, shares: bool, replies
if likes { if likes {
tracing::info!("fixing likes..."); tracing::info!("fixing likes...");
let mut store = std::collections::HashMap::new(); let mut store = std::collections::HashMap::new();
{
let mut stream = model::like::Entity::find().stream(&db).await?; let mut stream = model::like::Entity::find().stream(&db).await?;
while let Some(like) = stream.try_next().await? { while let Some(like) = stream.try_next().await? {
store.insert(like.likes.clone(), store.get(&like.likes).unwrap_or(&0) + 1); store.insert(like.likes.clone(), store.get(&like.likes).unwrap_or(&0) + 1);
} }
}
for (k, v) in store { for (k, v) in store {
let m = model::object::ActiveModel { let m = model::object::ActiveModel {
id: sea_orm::Set(k), id: sea_orm::Set(k.clone()),
likes: sea_orm::Set(v), likes: sea_orm::Set(v),
..Default::default() ..Default::default()
}; };
model::object::Entity::update(m) if let Err(e) = model::object::Entity::update(m)
.exec(&db) .exec(&db)
.await?; .await
{
tracing::warn!("record not updated ({k}): {e}");
}
} }
} }
if shares { if shares {
tracing::info!("fixing shares..."); tracing::info!("fixing shares...");
let mut store = std::collections::HashMap::new(); let mut store = std::collections::HashMap::new();
{
let mut stream = model::share::Entity::find().stream(&db).await?; let mut stream = model::share::Entity::find().stream(&db).await?;
while let Some(share) = stream.try_next().await? { while let Some(share) = stream.try_next().await? {
store.insert(share.shares.clone(), store.get(&share.shares).unwrap_or(&0) + 1); store.insert(share.shares.clone(), store.get(&share.shares).unwrap_or(&0) + 1);
} }
}
for (k, v) in store { for (k, v) in store {
let m = model::object::ActiveModel { let m = model::object::ActiveModel {
id: sea_orm::Set(k), id: sea_orm::Set(k.clone()),
shares: sea_orm::Set(v), shares: sea_orm::Set(v),
..Default::default() ..Default::default()
}; };
model::object::Entity::update(m) if let Err(e) = model::object::Entity::update(m)
.exec(&db) .exec(&db)
.await?; .await
{
tracing::warn!("record not updated ({k}): {e}");
}
} }
} }
if replies { if replies {
tracing::info!("fixing replies..."); tracing::info!("fixing replies...");
let mut store = std::collections::HashMap::new(); let mut store = std::collections::HashMap::new();
{
let mut stream = model::object::Entity::find().stream(&db).await?; let mut stream = model::object::Entity::find().stream(&db).await?;
while let Some(object) = stream.try_next().await? { while let Some(object) = stream.try_next().await? {
if let Some(reply) = object.in_reply_to { if let Some(reply) = object.in_reply_to {
@ -286,16 +297,20 @@ async fn fix(db: sea_orm::DatabaseConnection, likes: bool, shares: bool, replies
store.insert(reply, before + 1); store.insert(reply, before + 1);
} }
} }
}
for (k, v) in store { for (k, v) in store {
let m = model::object::ActiveModel { let m = model::object::ActiveModel {
id: sea_orm::Set(k), id: sea_orm::Set(k.clone()),
comments: sea_orm::Set(v), comments: sea_orm::Set(v),
..Default::default() ..Default::default()
}; };
model::object::Entity::update(m) if let Err(e) = model::object::Entity::update(m)
.exec(&db) .exec(&db)
.await?; .await
{
tracing::warn!("record not updated ({k}): {e}");
}
} }
} }