nutomic/ibis
1
0
Fork 0
mirror of https://github.com/Nutomic/ibis.git synced 2024-11-22 07:11:08 +00:00
Code Projects Releases Packages Activity

user working and tests passing

Browse source
This commit is contained in:
Felix Ableitner 2023-12-14 17:06:44 +01:00
parent e0bd8d2791
commit 69d08b824e
12 changed files with 250 additions and 81 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
migrations/2023-11-28-150402_fediwiki_setup/up.sql
View file

@ -29,7 +29,7 @@ create table local_user (
create table instance_follow (
id serial primary key,
instance_id int REFERENCES instance ON UPDATE CASCADE ON DELETE CASCADE NOT NULL,
follower_id int REFERENCES instance ON UPDATE CASCADE ON DELETE CASCADE NOT NULL,
follower_id int REFERENCES person ON UPDATE CASCADE ON DELETE CASCADE NOT NULL,
pending boolean not null,
unique(instance_id, follower_id)
);

8
src/api/instance.rs
View file

@ -1,8 +1,10 @@
use crate::database::instance::{DbInstance, InstanceView};
use crate::database::user::LocalUserView;
use crate::database::MyDataHandle;
use crate::error::MyResult;
use crate::federation::activities::follow::Follow;
use activitypub_federation::config::Data;
use axum::Extension;
use axum::{Form, Json};
use axum_macros::debug_handler;
use serde::{Deserialize, Serialize};
@ -25,14 +27,14 @@ pub struct FollowInstance {
/// updated articles.
#[debug_handler]
pub(in crate::api) async fn follow_instance(
Extension(user): Extension<LocalUserView>,
data: Data<MyDataHandle>,
Form(query): Form<FollowInstance>,
) -> MyResult<()> {
let local_instance = DbInstance::read_local_instance(&data.db_connection)?;
let target = DbInstance::read(query.id, &data.db_connection)?;
let pending = !target.local;
DbInstance::follow(local_instance.id, target.id, pending, &data)?;
DbInstance::follow(&user.person, &target, pending, &data)?;
let instance = DbInstance::read(query.id, &data.db_connection)?;
Follow::send(local_instance, instance, &data).await?;
Follow::send(user.person, instance, &data).await?;
Ok(())
}

22
src/database/instance.rs
View file

@ -1,4 +1,5 @@
use crate::database::schema::{instance, instance_follow};
use crate::database::user::DbPerson;
use crate::database::MyDataHandle;
use crate::error::MyResult;
use crate::federation::objects::articles_collection::DbArticleCollection;
@ -48,7 +49,7 @@ pub struct DbInstanceForm {
#[diesel(table_name = article, check_for_backend(diesel::pg::Pg))]
pub struct InstanceView {
pub instance: DbInstance,
pub followers: Vec<DbInstance>,
pub followers: Vec<DbPerson>,
pub following: Vec<DbInstance>,
}
@ -98,35 +99,36 @@ impl DbInstance {
}
pub fn follow(
follower_id_: i32,
instance_id_: i32,
follower: &DbPerson,
instance: &DbInstance,
pending_: bool,
data: &Data<MyDataHandle>,
) -> MyResult<()> {
debug_assert_ne!(follower_id_, instance_id_);
use instance_follow::dsl::{follower_id, instance_id, pending};
let mut conn = data.db_connection.lock().unwrap();
let form = (
instance_id.eq(instance_id_),
follower_id.eq(follower_id_),
instance_id.eq(instance.id),
follower_id.eq(follower.id),
pending.eq(pending_),
);
insert_into(instance_follow::table)
let rows = insert_into(instance_follow::table)
.values(form)
.on_conflict((instance_id, follower_id))
.do_update()
.set(form)
.execute(conn.deref_mut())?;
assert_eq!(1, rows);
Ok(())
}
pub fn read_followers(id_: i32, conn: &Mutex<PgConnection>) -> MyResult<Vec<Self>> {
pub fn read_followers(id_: i32, conn: &Mutex<PgConnection>) -> MyResult<Vec<DbPerson>> {
use crate::database::schema::person;
use instance_follow::dsl::{follower_id, instance_id};
let mut conn = conn.lock().unwrap();
Ok(instance_follow::table
.inner_join(instance::table.on(follower_id.eq(instance::dsl::id)))
.inner_join(person::table.on(follower_id.eq(person::dsl::id)))
.filter(instance_id.eq(id_))
.select(instance::all_columns)
.select(person::all_columns)
.get_results(conn.deref_mut())?)
}

33
src/federation/activities/accept.rs
View file

@ -1,7 +1,9 @@
use crate::database::instance::DbInstance;
use crate::error::MyResult;
use crate::federation::send_activity;
use crate::utils::generate_activity_id;
use crate::{database::MyDataHandle, federation::activities::follow::Follow};
use activitypub_federation::traits::Actor;
use activitypub_federation::{
config::Data, fetch::object_id::ObjectId, kinds::activity::AcceptType, traits::ActivityHandler,
};
@ -19,14 +21,28 @@ pub struct Accept {
}
impl Accept {
pub fn new(actor: ObjectId<DbInstance>, object: Follow) -> MyResult<Accept> {
let id = generate_activity_id(actor.inner())?;
Ok(Accept {
actor,
pub async fn send(
local_instance: DbInstance,
object: Follow,
data: &Data<MyDataHandle>,
) -> MyResult<()> {
let id = generate_activity_id(local_instance.ap_id.inner())?;
let follower = object.actor.dereference(data).await?;
let accept = Accept {
actor: local_instance.ap_id.clone(),
object,
kind: Default::default(),
id,
})
};
dbg!(&accept);
send_activity(
&local_instance,
accept,
vec![follower.shared_inbox_or_inbox()],
data,
)
.await?;
Ok(())
}
}
@ -48,10 +64,11 @@ impl ActivityHandler for Accept {
}
async fn receive(self, data: &Data<Self::DataType>) -> Result<(), Self::Error> {
dbg!(&self);
// add to follows
let local_instance = DbInstance::read_local_instance(&data.db_connection)?;
let actor = self.actor.dereference(data).await?;
DbInstance::follow(local_instance.id, actor.id, false, data)?;
let person = self.object.actor.dereference_local(data).await?;
let instance = self.actor.dereference(data).await?;
DbInstance::follow(&person, &instance, false, data)?;
Ok(())
}
}

29
src/federation/activities/follow.rs
View file

@ -1,5 +1,7 @@
use crate::database::instance::DbInstance;
use crate::database::user::DbPerson;
use crate::error::MyResult;
use crate::federation::send_activity;
use crate::{database::MyDataHandle, federation::activities::accept::Accept, generate_activity_id};
use activitypub_federation::{
config::Data,
@ -13,7 +15,7 @@ use url::Url;
#[derive(Deserialize, Serialize, Clone, Debug)]
#[serde(rename_all = "camelCase")]
pub struct Follow {
pub actor: ObjectId<DbInstance>,
pub actor: ObjectId<DbPerson>,
pub object: ObjectId<DbInstance>,
#[serde(rename = "type")]
kind: FollowType,
@ -21,21 +23,16 @@ pub struct Follow {
}
impl Follow {
pub async fn send(
local_instance: DbInstance,
to: DbInstance,
data: &Data<MyDataHandle>,
) -> MyResult<()> {
let id = generate_activity_id(local_instance.ap_id.inner())?;
pub async fn send(actor: DbPerson, to: DbInstance, data: &Data<MyDataHandle>) -> MyResult<()> {
let id = generate_activity_id(actor.ap_id.inner())?;
let follow = Follow {
actor: local_instance.ap_id.clone(),
actor: actor.ap_id.clone(),
object: to.ap_id.clone(),
kind: Default::default(),
id,
};
local_instance
.send(follow, vec![to.shared_inbox_or_inbox()], data)
.await?;
send_activity(&actor, follow, vec![to.shared_inbox_or_inbox()], data).await?;
Ok(())
}
}
@ -58,16 +55,14 @@ impl ActivityHandler for Follow {
}
async fn receive(self, data: &Data<Self::DataType>) -> Result<(), Self::Error> {
dbg!(&self);
let actor = self.actor.dereference(data).await?;
let local_instance = DbInstance::read_local_instance(&data.db_connection)?;
DbInstance::follow(actor.id, local_instance.id, false, data)?;
dbg!(&actor.ap_id, &local_instance.ap_id);
DbInstance::follow(&actor, &local_instance, false, data)?;
// send back an accept
let follower = self.actor.dereference(data).await?;
let accept = Accept::new(local_instance.ap_id.clone(), self)?;
local_instance
.send(accept, vec![follower.shared_inbox_or_inbox()], data)
.await?;
Accept::send(local_instance, self, data).await?;
Ok(())
}
}

11
src/federation/activities/reject.rs
View file

@ -11,6 +11,7 @@ use activitypub_federation::{
traits::ActivityHandler,
};
use crate::federation::send_activity;
use serde::{Deserialize, Serialize};
use url::Url;
@ -41,9 +42,13 @@ impl RejectEdit {
kind: Default::default(),
id,
};
local_instance
.send(reject, vec![Url::parse(&user_instance.inbox_url)?], data)
.await?;
send_activity(
&local_instance,
reject,
vec![Url::parse(&user_instance.inbox_url)?],
data,
)
.await?;
Ok(())
}
}

11
src/federation/activities/update_remote_article.rs
View file

@ -7,6 +7,7 @@ use crate::database::instance::DbInstance;
use crate::federation::activities::reject::RejectEdit;
use crate::federation::activities::update_local_article::UpdateLocalArticle;
use crate::federation::objects::edit::ApubEdit;
use crate::federation::send_activity;
use crate::utils::generate_activity_id;
use activitypub_federation::kinds::activity::UpdateType;
use activitypub_federation::{
@ -47,9 +48,13 @@ impl UpdateRemoteArticle {
kind: Default::default(),
id,
};
local_instance
.send(update, vec![Url::parse(&article_instance.inbox_url)?], data)
.await?;
send_activity(
&local_instance,
update,
vec![Url::parse(&article_instance.inbox_url)?],
data,
)
.await?;
Ok(())
}
}

31
src/federation/mod.rs
View file

@ -1,3 +1,34 @@
use crate::database::MyDataHandle;
use activitypub_federation::activity_sending::SendActivityTask;
use activitypub_federation::config::Data;
use activitypub_federation::protocol::context::WithContext;
use activitypub_federation::traits::{ActivityHandler, Actor};
use serde::Serialize;
use std::fmt::Debug;
use tracing::log::warn;
use url::Url;
pub mod activities;
pub mod objects;
pub mod routes;
pub async fn send_activity<Activity, ActorType: Actor>(
actor: &ActorType,
activity: Activity,
recipients: Vec<Url>,
data: &Data<MyDataHandle>,
) -> Result<(), <Activity as ActivityHandler>::Error>
where
Activity: ActivityHandler + Serialize + Debug + Send + Sync,
<Activity as ActivityHandler>::Error: From<activitypub_federation::error::Error>,
{
let activity = WithContext::new_default(activity);
let sends = SendActivityTask::prepare(&activity, actor, recipients, data).await?;
for send in sends {
let send = send.sign_and_send(data).await;
if let Err(e) = send {
warn!("Failed to send activity {:?}: {e}", activity);
}
}
Ok(())
}

25
src/federation/objects/instance.rs
View file

@ -2,6 +2,7 @@ use crate::database::instance::{DbInstance, DbInstanceForm};
use crate::database::MyDataHandle;
use crate::error::{Error, MyResult};
use crate::federation::objects::articles_collection::DbArticleCollection;
use crate::federation::send_activity;
use activitypub_federation::activity_sending::SendActivityTask;
use activitypub_federation::fetch::collection_id::CollectionId;
use activitypub_federation::kinds::actor::ServiceType;
@ -62,29 +63,7 @@ impl DbInstance {
.into_iter()
.map(|i| Url::parse(&i.inbox_url).unwrap()),
);
self.send(activity, inboxes, data).await?;
Ok(())
}
// TODO: move to user?
pub async fn send<Activity>(
&self,
activity: Activity,
recipients: Vec<Url>,
data: &Data<MyDataHandle>,
) -> Result<(), <Activity as ActivityHandler>::Error>
where
Activity: ActivityHandler + Serialize + Debug + Send + Sync,
<Activity as ActivityHandler>::Error: From<activitypub_federation::error::Error>,
{
let activity = WithContext::new_default(activity);
let sends = SendActivityTask::prepare(&activity, self, recipients, data).await?;
for send in sends {
let send = send.sign_and_send(data).await;
if let Err(e) = send {
warn!("Failed to send activity {:?}: {e}", activity);
}
}
send_activity(self, activity, inboxes, data).await?;
Ok(())
}
}

134
src/federation/routes.rs
View file

@ -1,7 +1,8 @@
use crate::database::article::DbArticle;
use crate::database::instance::DbInstance;
use crate::database::user::DbPerson;
use crate::database::MyDataHandle;
use crate::error::MyResult;
use crate::error::{Error, MyResult};
use crate::federation::activities::accept::Accept;
use crate::federation::activities::create_article::CreateArticle;
use crate::federation::activities::follow::Follow;
@ -12,10 +13,12 @@ use crate::federation::objects::article::ApubArticle;
use crate::federation::objects::articles_collection::{ArticleCollection, DbArticleCollection};
use crate::federation::objects::edits_collection::{ApubEditCollection, DbEditCollection};
use crate::federation::objects::instance::ApubInstance;
use crate::federation::objects::user::ApubUser;
use activitypub_federation::axum::inbox::{receive_activity, ActivityData};
use activitypub_federation::axum::json::FederationJson;
use activitypub_federation::config::Data;
use activitypub_federation::protocol::context::WithContext;
use activitypub_federation::traits::Actor;
use activitypub_federation::traits::Object;
use activitypub_federation::traits::{ActivityHandler, Collection};
use axum::extract::Path;
@ -23,12 +26,14 @@ use axum::response::IntoResponse;
use axum::routing::{get, post};
use axum::Router;
use axum_macros::debug_handler;
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use url::Url;
pub fn federation_routes() -> Router {
Router::new()
.route("/", get(http_get_instance))
.route("/user/:name", get(http_get_person))
.route("/all_articles", get(http_get_all_articles))
.route("/article/:title", get(http_get_article))
.route("/article/:title/edits", get(http_get_article_edits))
@ -44,6 +49,16 @@ async fn http_get_instance(
Ok(FederationJson(WithContext::new_default(json_instance)))
}
#[debug_handler]
async fn http_get_person(
Path(name): Path<String>,
data: Data<MyDataHandle>,
) -> MyResult<FederationJson<WithContext<ApubUser>>> {
let person = DbPerson::read_local_from_name(&name, &data)?.person;
let json_person = person.into_json(&data).await?;
Ok(FederationJson(WithContext::new_default(json_person)))
}
#[debug_handler]
async fn http_get_all_articles(
data: Data<MyDataHandle>,
@ -91,6 +106,119 @@ pub async fn http_post_inbox(
data: Data<MyDataHandle>,
activity_data: ActivityData,
) -> impl IntoResponse {
receive_activity::<WithContext<InboxActivities>, DbInstance, MyDataHandle>(activity_data, &data)
.await
receive_activity::<WithContext<InboxActivities>, UserOrInstance, MyDataHandle>(
activity_data,
&data,
)
.await
}
#[derive(Clone, Debug)]
pub enum UserOrInstance {
User(DbPerson),
Instance(DbInstance),
}
#[derive(Serialize, Deserialize, Clone, Debug)]
#[serde(untagged)]
pub enum PersonOrInstance {
Person(ApubUser),
Instance(ApubInstance),
}
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
pub enum PersonOrInstanceType {
Person,
Group,
}
#[async_trait::async_trait]
impl Object for UserOrInstance {
type DataType = MyDataHandle;
type Kind = PersonOrInstance;
type Error = Error;
fn last_refreshed_at(&self) -> Option<DateTime<Utc>> {
Some(match self {
UserOrInstance::User(p) => p.last_refreshed_at,
UserOrInstance::Instance(p) => p.last_refreshed_at,
})
}
#[tracing::instrument(skip_all)]
async fn read_from_id(
object_id: Url,
data: &Data<Self::DataType>,
) -> Result<Option<Self>, Error> {
let person = DbPerson::read_from_id(object_id.clone(), data).await?;
Ok(match person {
Some(o) => Some(UserOrInstance::User(o)),
None => DbInstance::read_from_id(object_id, data)
.await?
.map(UserOrInstance::Instance),
})
}
#[tracing::instrument(skip_all)]
async fn delete(self, data: &Data<Self::DataType>) -> Result<(), Error> {
match self {
UserOrInstance::User(p) => p.delete(data).await,
UserOrInstance::Instance(p) => p.delete(data).await,
}
}
async fn into_json(self, _data: &Data<Self::DataType>) -> Result<Self::Kind, Error> {
unimplemented!()
}
#[tracing::instrument(skip_all)]
async fn verify(
apub: &Self::Kind,
expected_domain: &Url,
data: &Data<Self::DataType>,
) -> Result<(), Error> {
match apub {
PersonOrInstance::Person(a) => DbPerson::verify(a, expected_domain, data).await,
PersonOrInstance::Instance(a) => DbInstance::verify(a, expected_domain, data).await,
}
}
#[tracing::instrument(skip_all)]
async fn from_json(apub: Self::Kind, data: &Data<Self::DataType>) -> Result<Self, Error> {
Ok(match apub {
PersonOrInstance::Person(p) => {
UserOrInstance::User(DbPerson::from_json(p, data).await?)
}
PersonOrInstance::Instance(p) => {
UserOrInstance::Instance(DbInstance::from_json(p, data).await?)
}
})
}
}
impl Actor for UserOrInstance {
fn id(&self) -> Url {
match self {
UserOrInstance::User(u) => u.id(),
UserOrInstance::Instance(c) => c.id(),
}
}
fn public_key_pem(&self) -> &str {
match self {
UserOrInstance::User(p) => p.public_key_pem(),
UserOrInstance::Instance(p) => p.public_key_pem(),
}
}
fn private_key_pem(&self) -> Option<String> {
match self {
UserOrInstance::User(p) => p.private_key_pem(),
UserOrInstance::Instance(p) => p.private_key_pem(),
}
}
fn inbox(&self) -> Url {
unimplemented!()
}
}

10
tests/common.rs
View file

@ -235,13 +235,13 @@ where
}
}
pub async fn follow_instance(api_instance: &str, follow_instance: &str) -> MyResult<()> {
pub async fn follow_instance(instance: &FediwikiInstance, follow_instance: &str) -> MyResult<()> {
// fetch beta instance on alpha
let resolve_form = ResolveObject {
id: Url::parse(&format!("http://{}", follow_instance))?,
};
let instance_resolved: DbInstance =
get_query(api_instance, "resolve_instance", Some(resolve_form)).await?;
get_query(&instance.hostname, "resolve_instance", Some(resolve_form)).await?;
// send follow
let follow_form = FollowInstance {
@ -249,8 +249,12 @@ pub async fn follow_instance(api_instance: &str, follow_instance: &str) -> MyRes
};
// cant use post helper because follow doesnt return json
let res = CLIENT
.post(format!("http://{}/api/v1/instance/follow", api_instance))
.post(format!(
"http://{}/api/v1/instance/follow",
instance.hostname
))
.form(&follow_form)
.bearer_auth(&instance.jwt)
.send()
.await?;
if res.status() == StatusCode::OK {

15
tests/test.rs
View file

@ -88,7 +88,7 @@ async fn test_follow_instance() -> MyResult<()> {
assert_eq!(0, beta_instance.followers.len());
assert_eq!(0, beta_instance.following.len());
follow_instance(&data.alpha.hostname, &data.beta.hostname).await?;
follow_instance(&data.alpha, &data.beta.hostname).await?;
// check that follow was federated
let alpha_instance: InstanceView = get(&data.alpha.hostname, "instance").await?;
@ -102,9 +102,10 @@ async fn test_follow_instance() -> MyResult<()> {
let beta_instance: InstanceView = get(&data.beta.hostname, "instance").await?;
assert_eq!(0, beta_instance.following.len());
assert_eq!(1, beta_instance.followers.len());
// TODO: compare full ap_id of alpha user, but its not available through api yet
assert_eq!(
alpha_instance.instance.ap_id,
beta_instance.followers[0].ap_id
alpha_instance.instance.ap_id.inner().domain(),
beta_instance.followers[0].ap_id.inner().domain()
);
data.stop()
@ -160,7 +161,7 @@ async fn test_synchronize_articles() -> MyResult<()> {
async fn test_edit_local_article() -> MyResult<()> {
let data = TestData::start().await;
follow_instance(&data.alpha.hostname, &data.beta.hostname).await?;
follow_instance(&data.alpha, &data.beta.hostname).await?;
// create new article
let title = "Manu_Chao".to_string();
@ -203,8 +204,8 @@ async fn test_edit_local_article() -> MyResult<()> {
async fn test_edit_remote_article() -> MyResult<()> {
let data = TestData::start().await;
follow_instance(&data.alpha.hostname, &data.beta.hostname).await?;
follow_instance(&data.gamma.hostname, &data.beta.hostname).await?;
follow_instance(&data.alpha, &data.beta.hostname).await?;
follow_instance(&data.gamma, &data.beta.hostname).await?;
// create new article
let title = "Manu_Chao".to_string();
@ -309,7 +310,7 @@ async fn test_local_edit_conflict() -> MyResult<()> {
async fn test_federated_edit_conflict() -> MyResult<()> {
let data = TestData::start().await;
follow_instance(&data.alpha.hostname, &data.beta.hostname).await?;
follow_instance(&data.alpha, &data.beta.hostname).await?;
// create new article
let title = "Manu_Chao".to_string();