feat: added users update task

src/cli/mod.rs

@@ -12,3 +12,6 @@ pub use relay::*;
 
 mod register;
 pub use register::*;
+
+mod update;
+pub use update::*;

src/cli/update.rs

@@ -0,0 +1,38 @@
+use futures::TryStreamExt;
+use sea_orm::{ColumnTrait, EntityTrait, IntoActiveModel, QueryFilter};
+
+use crate::server::{fetcher::Fetcher, Context};
+
+pub async fn update_users(db: sea_orm::DatabaseConnection, domain: String, days: i64) -> crate::Result<()> {
+	let ctx = Context::new(db, domain).await?;
+	let mut count = 0;
+	let mut insertions = Vec::new();
+
+	{
+		let mut stream = crate::model::user::Entity::find()
+			.filter(crate::model::user::Column::Updated.lt(chrono::Utc::now() - chrono::Duration::days(days)))
+			.stream(ctx.db())
+			.await?;
+
+
+		while let Some(user) = stream.try_next().await? {
+			match ctx.pull_user(&user.id).await {
+				Err(e) => tracing::warn!("could not update user {}: {e}", user.id),
+				Ok(u) => {
+					insertions.push(u);
+					count += 1;
+				},
+			}
+		}
+	}
+
+	for u in insertions {
+		tracing::info!("updating user {}", u.id);
+		crate::model::user::Entity::delete_by_id(&u.id).exec(ctx.db()).await?;
+		crate::model::user::Entity::insert(u.into_active_model()).exec(ctx.db()).await?;
+	}
+
+	tracing::info!("updated {count} users");
+
+	Ok(())
+}

src/main.rs

@@ -88,6 +88,13 @@ enum CliCommand {
 		#[arg(long, default_value_t = false)]
 		/// fix replies counts for posts
 		replies: bool,
+	},
+
+	/// update remote users
+	Update {
+		#[arg(long, short, default_value_t = 7)]
+		/// number of days after which users should get updated
+		days: i64,
 	}
 }
 
@@ -131,8 +138,11 @@ async fn main() {
 
 		CliCommand::Fix { likes, shares, replies } =>
 			cli::fix(db, likes, shares, replies)
-				.await
+				.await.expect("failed running fix task"),
-				.expect("failed running fix task"),
+
+		CliCommand::Update { days } =>
+			cli::update_users(db, args.domain, days)
+				.await.expect("error updating users"),
 
 		CliCommand::Serve => {
 			let ctx = server::Context::new(db, args.domain)