chore: refactor, moved cli tasks under directory

This commit is contained in:
əlemi 2024-04-30 16:48:28 +02:00
parent b0ebeb9fff
commit f2dc3f69f2
Signed by: alemi
GPG key ID: A4895B84D311642C
9 changed files with 245 additions and 182 deletions

View file

@ -45,7 +45,6 @@ time = { version = "0.3", features = ["serde"], optional = true }
async-recursion = "1.1" async-recursion = "1.1"
[features] [features]
default = ["faker", "migrations", "mastodon"] default = ["migrations"]
faker = []
migrations = ["dep:sea-orm-migration"] migrations = ["dep:sea-orm-migration"]
mastodon = ["dep:mastodon-async-entities", "dep:time"] mastodon = ["dep:mastodon-async-entities", "dep:time"]

View file

@ -1,5 +1,4 @@
use crate::model::{addressing, config, credential}; use crate::model::{addressing, config, credential, activity, object, user, Audience};
use super::{activity, object, user, Audience};
use openssl::rsa::Rsa; use openssl::rsa::Rsa;
use sea_orm::IntoActiveModel; use sea_orm::IntoActiveModel;
@ -7,7 +6,7 @@ pub async fn faker(db: &sea_orm::DatabaseConnection, domain: String, count: u64)
use sea_orm::{EntityTrait, Set}; use sea_orm::{EntityTrait, Set};
let key = Rsa::generate(2048).unwrap(); let key = Rsa::generate(2048).unwrap();
let test_user = super::user::Model { let test_user = user::Model {
id: format!("{domain}/users/test"), id: format!("{domain}/users/test"),
name: Some("μpub".into()), name: Some("μpub".into()),
domain: clean_domain(&domain), domain: clean_domain(&domain),

42
src/cli/fetch.rs Normal file
View file

@ -0,0 +1,42 @@
use sea_orm::{EntityTrait, IntoActiveModel};
use crate::server::fetcher::Fetchable;
pub async fn fetch(db: sea_orm::DatabaseConnection, domain: String, uri: String, save: bool) -> crate::Result<()> {
use apb::Base;
let ctx = crate::server::Context::new(db, domain)
.await.expect("failed creating server context");
let mut node = apb::Node::link(uri.to_string());
node.fetch(&ctx).await?;
let obj = node.get().expect("node still empty after fetch?");
if save {
match obj.base_type() {
Some(apb::BaseType::Object(apb::ObjectType::Actor(_))) => {
crate::model::user::Entity::insert(
crate::model::user::Model::new(obj).unwrap().into_active_model()
).exec(ctx.db()).await.unwrap();
},
Some(apb::BaseType::Object(apb::ObjectType::Activity(_))) => {
crate::model::activity::Entity::insert(
crate::model::activity::Model::new(obj).unwrap().into_active_model()
).exec(ctx.db()).await.unwrap();
},
Some(apb::BaseType::Object(apb::ObjectType::Note)) => {
crate::model::object::Entity::insert(
crate::model::object::Model::new(obj).unwrap().into_active_model()
).exec(ctx.db()).await.unwrap();
},
Some(apb::BaseType::Object(t)) => tracing::warn!("not implemented: {:?}", t),
Some(apb::BaseType::Link(_)) => tracing::error!("fetched another link?"),
None => tracing::error!("no type on object"),
}
}
println!("{}", serde_json::to_string_pretty(&obj).unwrap());
Ok(())
}

87
src/cli/fix.rs Normal file
View file

@ -0,0 +1,87 @@
use sea_orm::EntityTrait;
pub async fn fix(db: sea_orm::DatabaseConnection, likes: bool, shares: bool, replies: bool) -> crate::Result<()> {
use futures::TryStreamExt;
if likes {
tracing::info!("fixing likes...");
let mut store = std::collections::HashMap::new();
{
let mut stream = crate::model::like::Entity::find().stream(&db).await?;
while let Some(like) = stream.try_next().await? {
store.insert(like.likes.clone(), store.get(&like.likes).unwrap_or(&0) + 1);
}
}
for (k, v) in store {
let m = crate::model::object::ActiveModel {
id: sea_orm::Set(k.clone()),
likes: sea_orm::Set(v),
..Default::default()
};
if let Err(e) = crate::model::object::Entity::update(m)
.exec(&db)
.await
{
tracing::warn!("record not updated ({k}): {e}");
}
}
}
if shares {
tracing::info!("fixing shares...");
let mut store = std::collections::HashMap::new();
{
let mut stream = crate::model::share::Entity::find().stream(&db).await?;
while let Some(share) = stream.try_next().await? {
store.insert(share.shares.clone(), store.get(&share.shares).unwrap_or(&0) + 1);
}
}
for (k, v) in store {
let m = crate::model::object::ActiveModel {
id: sea_orm::Set(k.clone()),
shares: sea_orm::Set(v),
..Default::default()
};
if let Err(e) = crate::model::object::Entity::update(m)
.exec(&db)
.await
{
tracing::warn!("record not updated ({k}): {e}");
}
}
}
if replies {
tracing::info!("fixing replies...");
let mut store = std::collections::HashMap::new();
{
let mut stream = crate::model::object::Entity::find().stream(&db).await?;
while let Some(object) = stream.try_next().await? {
if let Some(reply) = object.in_reply_to {
let before = store.get(&reply).unwrap_or(&0);
store.insert(reply, before + 1);
}
}
}
for (k, v) in store {
let m = crate::model::object::ActiveModel {
id: sea_orm::Set(k.clone()),
comments: sea_orm::Set(v),
..Default::default()
};
if let Err(e) = crate::model::object::Entity::update(m)
.exec(&db)
.await
{
tracing::warn!("record not updated ({k}): {e}");
}
}
}
tracing::info!("done running fix tasks");
Ok(())
}

14
src/cli/mod.rs Normal file
View file

@ -0,0 +1,14 @@
mod fix;
pub use fix::*;
mod fetch;
pub use fetch::*;
mod faker;
pub use faker::*;
mod relay;
pub use relay::*;
mod register;
pub use register::*;

44
src/cli/register.rs Normal file
View file

@ -0,0 +1,44 @@
use openssl::rsa::Rsa;
use sea_orm::{EntityTrait, IntoActiveModel};
pub async fn register(
db: sea_orm::DatabaseConnection,
domain: String,
) -> crate::Result<()> {
let key = Rsa::generate(2048).unwrap();
let test_user = crate::model::user::Model {
id: format!("{domain}/users/test"),
name: Some("μpub".into()),
domain: clean_domain(&domain),
preferred_username: "test".to_string(),
summary: Some("hello world! i'm manually generated but served dynamically from db! check progress at https://git.alemi.dev/upub.git".to_string()),
following: None,
following_count: 0,
followers: None,
followers_count: 0,
statuses_count: 0,
icon: Some("https://cdn.alemi.dev/social/circle-square.png".to_string()),
image: Some("https://cdn.alemi.dev/social/someriver-xs.jpg".to_string()),
inbox: None,
shared_inbox: None,
outbox: None,
actor_type: apb::ActorType::Person,
created: chrono::Utc::now(),
updated: chrono::Utc::now(),
private_key: Some(std::str::from_utf8(&key.private_key_to_pem().unwrap()).unwrap().to_string()),
// TODO generate a fresh one every time
public_key: std::str::from_utf8(&key.public_key_to_pem().unwrap()).unwrap().to_string(),
};
crate::model::user::Entity::insert(test_user.clone().into_active_model()).exec(&db).await?;
Ok(())
}
// TODO duplicated, make an util?? idk
fn clean_domain(domain: &str) -> String {
domain
.replace("http://", "")
.replace("https://", "")
.replace('/', "")
}

40
src/cli/relay.rs Normal file
View file

@ -0,0 +1,40 @@
use sea_orm::{ColumnTrait, EntityTrait, IntoActiveModel, QueryFilter, QueryOrder};
pub async fn relay(db: sea_orm::DatabaseConnection, domain: String, actor: String, accept: bool) -> crate::Result<()> {
let ctx = crate::server::Context::new(db, domain).await?;
let aid = ctx.aid(uuid::Uuid::new_v4().to_string());
let mut activity_model = crate::model::activity::Model {
id: aid.clone(),
activity_type: apb::ActivityType::Follow,
actor: ctx.base(),
object: Some(actor.clone()),
target: None,
published: chrono::Utc::now(),
to: crate::model::Audience(vec![actor.clone()]),
bto: crate::model::Audience::default(),
cc: crate::model::Audience(vec![apb::target::PUBLIC.to_string()]),
bcc: crate::model::Audience::default(),
};
if accept {
let follow_req = crate::model::activity::Entity::find()
.filter(crate::model::activity::Column::ActivityType.eq("Follow"))
.filter(crate::model::activity::Column::Actor.eq(&actor))
.filter(crate::model::activity::Column::Object.eq(ctx.base()))
.order_by_desc(crate::model::activity::Column::Published)
.one(ctx.db())
.await?
.expect("no follow request to accept");
activity_model.activity_type = apb::ActivityType::Accept(apb::AcceptType::Accept);
activity_model.object = Some(follow_req.id);
};
crate::model::activity::Entity::insert(activity_model.into_active_model())
.exec(ctx.db()).await?;
ctx.dispatch(&ctx.base(), vec![actor, apb::target::PUBLIC.to_string()], &aid, None).await?;
Ok(())
}

View file

@ -1,6 +1,7 @@
pub mod server; pub mod server;
pub mod model; pub mod model;
pub mod routes; pub mod routes;
pub mod cli;
pub mod errors; pub mod errors;
@ -11,13 +12,11 @@ mod migrations;
use sea_orm_migration::MigratorTrait; use sea_orm_migration::MigratorTrait;
use clap::{Parser, Subcommand}; use clap::{Parser, Subcommand};
use sea_orm::{ColumnTrait, ConnectOptions, Database, EntityTrait, IntoActiveModel, QueryFilter, QueryOrder}; use sea_orm::{ConnectOptions, Database};
pub use errors::UpubResult as Result; pub use errors::UpubResult as Result;
use tower_http::{cors::CorsLayer, trace::TraceLayer}; use tower_http::{cors::CorsLayer, trace::TraceLayer};
use crate::server::fetcher::Fetchable;
pub const VERSION: &str = env!("CARGO_PKG_VERSION"); pub const VERSION: &str = env!("CARGO_PKG_VERSION");
#[derive(Parser)] #[derive(Parser)]
@ -113,58 +112,25 @@ async fn main() {
match args.command { match args.command {
#[cfg(feature = "migrations")] #[cfg(feature = "migrations")]
CliCommand::Migrate => migrations::Migrator::up(&db, None) CliCommand::Migrate =>
.await.expect("error applying migrations"), migrations::Migrator::up(&db, None)
.await.expect("error applying migrations"),
#[cfg(feature = "faker")] #[cfg(feature = "faker")]
CliCommand::Faker { count } => model::faker::faker(&db, args.domain, count) CliCommand::Faker { count } =>
.await.expect("error creating fake entities"), cli::faker(&db, args.domain, count)
.await.expect("error creating fake entities"),
CliCommand::Fetch { uri, save } => fetch(db, args.domain, uri, save) CliCommand::Fetch { uri, save } =>
.await.expect("error fetching object"), cli::fetch(db, args.domain, uri, save)
.await.expect("error fetching object"),
CliCommand::Relay { actor, accept } => { CliCommand::Relay { actor, accept } =>
let ctx = server::Context::new(db, args.domain) cli::relay(db, args.domain, actor, accept)
.await.expect("failed creating server context"); .await.expect("error registering/accepting relay"),
let aid = ctx.aid(uuid::Uuid::new_v4().to_string());
let mut activity_model = model::activity::Model {
id: aid.clone(),
activity_type: apb::ActivityType::Follow,
actor: ctx.base(),
object: Some(actor.clone()),
target: None,
published: chrono::Utc::now(),
to: model::Audience(vec![actor.clone()]),
bto: model::Audience::default(),
cc: model::Audience(vec![apb::target::PUBLIC.to_string()]),
bcc: model::Audience::default(),
};
if accept {
let follow_req = model::activity::Entity::find()
.filter(model::activity::Column::ActivityType.eq("Follow"))
.filter(model::activity::Column::Actor.eq(&actor))
.filter(model::activity::Column::Object.eq(ctx.base()))
.order_by_desc(model::activity::Column::Published)
.one(ctx.db())
.await
.expect("failed querying db for relay follow req")
.expect("no follow request to accept");
activity_model.activity_type = apb::ActivityType::Accept(apb::AcceptType::Accept);
activity_model.object = Some(follow_req.id);
};
model::activity::Entity::insert(activity_model.into_active_model())
.exec(ctx.db()).await.expect("could not insert activity in db");
ctx.dispatch(&ctx.base(), vec![actor, apb::target::PUBLIC.to_string()], &aid, None).await
.expect("could not dispatch relay activity");
},
CliCommand::Fix { likes, shares, replies } => CliCommand::Fix { likes, shares, replies } =>
fix(db, likes, shares, replies) cli::fix(db, likes, shares, replies)
.await .await
.expect("failed running fix task"), .expect("failed running fix task"),
@ -192,128 +158,3 @@ async fn main() {
}, },
} }
} }
async fn fetch(db: sea_orm::DatabaseConnection, domain: String, uri: String, save: bool) -> crate::Result<()> {
use apb::Base;
let ctx = server::Context::new(db, domain)
.await.expect("failed creating server context");
let mut node = apb::Node::link(uri.to_string());
node.fetch(&ctx).await?;
let obj = node.get().expect("node still empty after fetch?");
if save {
match obj.base_type() {
Some(apb::BaseType::Object(apb::ObjectType::Actor(_))) => {
model::user::Entity::insert(
model::user::Model::new(obj).unwrap().into_active_model()
).exec(ctx.db()).await.unwrap();
},
Some(apb::BaseType::Object(apb::ObjectType::Activity(_))) => {
model::activity::Entity::insert(
model::activity::Model::new(obj).unwrap().into_active_model()
).exec(ctx.db()).await.unwrap();
},
Some(apb::BaseType::Object(apb::ObjectType::Note)) => {
model::object::Entity::insert(
model::object::Model::new(obj).unwrap().into_active_model()
).exec(ctx.db()).await.unwrap();
},
Some(apb::BaseType::Object(t)) => tracing::warn!("not implemented: {:?}", t),
Some(apb::BaseType::Link(_)) => tracing::error!("fetched another link?"),
None => tracing::error!("no type on object"),
}
}
println!("{}", serde_json::to_string_pretty(&obj).unwrap());
Ok(())
}
async fn fix(db: sea_orm::DatabaseConnection, likes: bool, shares: bool, replies: bool) -> crate::Result<()> {
use futures::TryStreamExt;
if likes {
tracing::info!("fixing likes...");
let mut store = std::collections::HashMap::new();
{
let mut stream = model::like::Entity::find().stream(&db).await?;
while let Some(like) = stream.try_next().await? {
store.insert(like.likes.clone(), store.get(&like.likes).unwrap_or(&0) + 1);
}
}
for (k, v) in store {
let m = model::object::ActiveModel {
id: sea_orm::Set(k.clone()),
likes: sea_orm::Set(v),
..Default::default()
};
if let Err(e) = model::object::Entity::update(m)
.exec(&db)
.await
{
tracing::warn!("record not updated ({k}): {e}");
}
}
}
if shares {
tracing::info!("fixing shares...");
let mut store = std::collections::HashMap::new();
{
let mut stream = model::share::Entity::find().stream(&db).await?;
while let Some(share) = stream.try_next().await? {
store.insert(share.shares.clone(), store.get(&share.shares).unwrap_or(&0) + 1);
}
}
for (k, v) in store {
let m = model::object::ActiveModel {
id: sea_orm::Set(k.clone()),
shares: sea_orm::Set(v),
..Default::default()
};
if let Err(e) = model::object::Entity::update(m)
.exec(&db)
.await
{
tracing::warn!("record not updated ({k}): {e}");
}
}
}
if replies {
tracing::info!("fixing replies...");
let mut store = std::collections::HashMap::new();
{
let mut stream = model::object::Entity::find().stream(&db).await?;
while let Some(object) = stream.try_next().await? {
if let Some(reply) = object.in_reply_to {
let before = store.get(&reply).unwrap_or(&0);
store.insert(reply, before + 1);
}
}
}
for (k, v) in store {
let m = model::object::ActiveModel {
id: sea_orm::Set(k.clone()),
comments: sea_orm::Set(v),
..Default::default()
};
if let Err(e) = model::object::Entity::update(m)
.exec(&db)
.await
{
tracing::warn!("record not updated ({k}): {e}");
}
}
}
tracing::info!("done running fix tasks");
Ok(())
}

View file

@ -14,9 +14,6 @@ pub mod delivery;
pub mod attachment; pub mod attachment;
pub mod application; pub mod application;
#[cfg(feature = "faker")]
pub mod faker;
#[derive(Debug, Clone, thiserror::Error)] #[derive(Debug, Clone, thiserror::Error)]
#[error("missing required field: '{0}'")] #[error("missing required field: '{0}'")]
pub struct FieldError(pub &'static str); pub struct FieldError(pub &'static str);