aitzol/Plume
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

WIP: use the activitystreams crate

Browse Source
This commit is contained in:
Bat
2018-05-16 19:20:44 +01:00
parent 1870f52332
commit c779b1c58a
14 changed files with 365 additions and 177 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/activity_pub/inbox.rs
+114 -76
View File
@@ -1,100 +1,138 @@
use activitystreams_types::{
actor::Person,
activity::{Create, Follow, Like, Undo},
object::{Article, Note}
};
use activitystreams::activity::Activity;
use diesel::PgConnection;
use failure::Error;
use serde_json;
use activity_pub::activity;
use activity_pub::actor::Actor;
use activity_pub::outbox::broadcast;
// use activity_pub::broadcast;
use activity_pub::actor::Actor as APActor;
use activity_pub::sign::*;
use models::blogs::Blog;
use models::comments::*;
use models::follows::*;
use models::likes::*;
use models::likes;
use models::posts::*;
use models::users::User;
#[derive(Fail, Debug)]
enum InboxError {
#[fail(display = "The `type` property is required, but was not present")]
NoType,
#[fail(display = "Invalid activity type")]
InvalidType,
#[fail(display = "Couldn't undo activity")]
CantUndo
}
pub trait Inbox {
fn received(&self, conn: &PgConnection, act: serde_json::Value);
fn save(&self, conn: &PgConnection, act: serde_json::Value) {
match act["type"].as_str().unwrap() {
"Create" => {
match act["object"]["type"].as_str().unwrap() {
"Article" => {
Post::insert(conn, NewPost {
blog_id: 0, // TODO
slug: String::from(""), // TODO
title: String::from(""), // TODO
content: act["object"]["content"].as_str().unwrap().to_string(),
published: true,
license: String::from("CC-0"),
ap_url: act["object"]["url"].as_str().unwrap().to_string()
});
},
"Note" => {
let previous_comment = Comment::find_by_ap_url(conn, act["object"]["inReplyTo"].as_str().unwrap().to_string());
Comment::insert(conn, NewComment {
content: act["object"]["content"].as_str().unwrap().to_string(),
spoiler_text: act["object"]["summary"].as_str().unwrap_or("").to_string(),
ap_url: Some(act["object"]["id"].as_str().unwrap().to_string()),
in_response_to_id: previous_comment.clone().map(|c| c.id),
post_id: previous_comment
.map(|c| c.post_id)
.unwrap_or_else(|| Post::find_by_ap_url(conn, act["object"]["inReplyTo"].as_str().unwrap().to_string()).unwrap().id),
author_id: User::from_url(conn, act["actor"].as_str().unwrap().to_string()).unwrap().id,
sensitive: act["object"]["sensitive"].as_bool().unwrap_or(false)
});
}
x => println!("Received a new {}, but didn't saved it", x)
}
},
"Follow" => {
let follow_act = activity::Follow::deserialize(act.clone());
let from = User::from_url(conn, act["actor"].as_str().unwrap().to_string()).unwrap();
match User::from_url(conn, act["object"].as_str().unwrap().to_string()) {
Some(u) => self.accept_follow(conn, &from, &u, &follow_act, from.id, u.id),
None => {
let blog = Blog::from_url(conn, follow_act.get_target_id()).unwrap();
self.accept_follow(conn, &from, &blog, &follow_act, from.id, blog.id)
}
};
fn new_article(&self, conn: &PgConnection, article: Article) -> Result<(), Error> {
Post::insert(conn, NewPost {
blog_id: 0, // TODO
slug: String::from(""), // TODO
title: article.object_props.name_string().unwrap(),
content: article.object_props.content_string().unwrap(),
published: true,
license: String::from("CC-0"),
ap_url: article.object_props.url_string()?
});
Ok(())
}
fn new_comment(&self, conn: &PgConnection, note: Note, actor_id: String) -> Result<(), Error> {
let previous_url = note.object_props.in_reply_to.clone().unwrap().as_str().unwrap().to_string();
let previous_comment = Comment::find_by_ap_url(conn, previous_url.clone());
Comment::insert(conn, NewComment {
content: note.object_props.content_string().unwrap(),
spoiler_text: note.object_props.summary_string().unwrap(),
ap_url: note.object_props.id_string().ok(),
in_response_to_id: previous_comment.clone().map(|c| c.id),
post_id: previous_comment
.map(|c| c.post_id)
.unwrap_or_else(|| Post::find_by_ap_url(conn, previous_url).unwrap().id),
author_id: User::from_url(conn, actor_id).unwrap().id,
sensitive: false // "sensitive" is not a standard property, we need to think about how to support it with the activitystreams crate
});
Ok(())
}
fn follow(&self, conn: &PgConnection, follow: Follow) -> Result<(), Error> {
let from = User::from_url(conn, follow.actor.as_str().unwrap().to_string()).unwrap();
match User::from_url(conn, follow.object.as_str().unwrap().to_string()) {
Some(u) => self.accept_follow(conn, &from, &u, &follow, from.id, u.id),
None => {
let blog = Blog::from_url(conn, follow.object.as_str().unwrap().to_string()).unwrap();
self.accept_follow(conn, &from, &blog, &follow, from.id, blog.id)
}
"Like" => {
let liker = User::from_url(conn, act["actor"].as_str().unwrap().to_string());
let post = Post::find_by_ap_url(conn, act["object"].as_str().unwrap().to_string());
Like::insert(conn, NewLike {
post_id: post.unwrap().id,
user_id: liker.unwrap().id,
ap_url: act["id"].as_str().unwrap().to_string()
});
},
"Undo" => {
match act["object"]["type"].as_str().unwrap() {
"Like" => {
let like = Like::find_by_ap_url(conn, act["object"]["id"].as_str().unwrap().to_string()).unwrap();
like.delete(conn);
};
Ok(())
}
fn like(&self, conn: &PgConnection, like: Like) -> Result<(), Error> {
let liker = User::from_url(conn, like.actor.as_str().unwrap().to_string());
let post = Post::find_by_ap_url(conn, like.object.as_str().unwrap().to_string());
likes::Like::insert(conn, likes::NewLike {
post_id: post.unwrap().id,
user_id: liker.unwrap().id,
ap_url: like.object_props.id_string()?
});
Ok(())
}
fn unlike(&self, conn: &PgConnection, undo: Undo) -> Result<(), Error> {
let like = likes::Like::find_by_ap_url(conn, undo.object_object::<Like>()?.object_props.id_string()?).unwrap();
like.delete(conn);
Ok(())
}
fn save(&self, conn: &PgConnection, act: serde_json::Value) -> Result<(), Error> {
match act["type"].as_str() {
Some(t) => {
match t {
"Create" => {
let act: Create = serde_json::from_value(act.clone())?;
match act.object["type"].as_str().unwrap() {
"Article" => self.new_article(conn, act.object_object().unwrap()),
"Note" => self.new_comment(conn, act.object_object().unwrap(), act.actor_object::<Person>()?.object_props.id_string()?),
_ => Err(InboxError::InvalidType)?
}
},
"Follow" => self.follow(conn, serde_json::from_value(act.clone())?),
"Like" => self.like(conn, serde_json::from_value(act.clone())?),
"Undo" => {
let act: Undo = serde_json::from_value(act.clone())?;
match act.object["type"].as_str().unwrap() {
"Like" => self.unlike(conn, act),
_ => Err(InboxError::CantUndo)?
}
}
x => println!("Wanted to Undo a {}, but it is not supported yet", x)
_ => Err(InboxError::InvalidType)?
}
},
x => println!("Received unknow activity type: {}", x)
None => Err(InboxError::NoType)?
}
}
fn accept_follow<A: Actor + Signer, B: Actor + Clone, T: activity::Activity>(
fn accept_follow<A: Signer, B: Clone, T: Activity>(
&self,
conn: &PgConnection,
from: &A,
target: &B,
follow: &T,
from_id: i32,
target_id: i32
_conn: &PgConnection,
_from: &A,
_target: &B,
_follow: &T,
_from_id: i32,
_target_id: i32
) {
Follow::insert(conn, NewFollow {
follower_id: from_id,
following_id: target_id
});
// TODO
//Follow::insert(conn, NewFollow {
// follower_id: from_id,
// following_id: target_id
//});
let accept = activity::Accept::new(target, follow, conn);
broadcast(conn, from, accept, vec![target.clone()]);
//let accept = activity::Accept::new(target, follow, conn);
//broadcast(conn, from, accept, vec![target.clone()]);
}
}
src/activity_pub/mod.rs
+67 -4
View File
@@ -1,13 +1,20 @@
use rocket::http::ContentType;
use rocket::response::Content;
use activitystreams_traits::{Activity, Object, Link};
use array_tool::vec::Uniq;
use diesel::PgConnection;
use reqwest::Client;
use rocket::http::{ContentType, Status};
use rocket::response::{Response, Responder, Content};
use rocket::request::Request;
use rocket_contrib::Json;
use serde_json;
pub mod activity;
use self::sign::Signable;
// pub mod activity;
pub mod actor;
pub mod inbox;
pub mod object;
pub mod outbox;
// pub mod outbox;
pub mod request;
pub mod sign;
pub mod webfinger;
@@ -54,3 +61,59 @@ pub fn context() -> serde_json::Value {
pub fn activity_pub(json: serde_json::Value) -> ActivityPub {
Content(ContentType::new("application", "activity+json"), Json(json))
}
pub struct ActivityStream<T> (T);
impl<T> ActivityStream<T> {
pub fn new(t: T) -> ActivityStream<T> {
ActivityStream(t)
}
}
impl<'r, O: Object> Responder<'r> for ActivityStream<O> {
fn respond_to(self, request: &Request) -> Result<Response<'r>, Status> {
serde_json::to_string(&self.0).respond_to(request)
}
}
pub fn broadcast<A: Activity + Clone, S: actor::Actor + sign::Signer, T: actor::Actor>(conn: &PgConnection, sender: &S, act: A, to: Vec<T>) {
let boxes = to.into_iter()
.map(|u| u.get_shared_inbox_url().unwrap_or(u.get_inbox_url()))
.collect::<Vec<String>>()
.unique();
for inbox in boxes {
// TODO: run it in Sidekiq or something like that
let mut act = serde_json::to_value(act.clone()).unwrap();
act["@context"] = context();
let signed = act.sign(sender, conn);
let res = Client::new()
.post(&inbox[..])
.headers(request::headers())
.header(request::signature(sender, request::headers(), conn))
.header(request::digest(signed.to_string()))
.body(signed.to_string())
.send();
match res {
Ok(mut r) => println!("Successfully sent activity to inbox ({})\n\n{:?}", inbox, r.text().unwrap()),
Err(e) => println!("Error while sending to inbox ({:?})", e)
}
}
}
#[derive(Serialize, Deserialize)]
pub struct Id {
#[serde(flatten)]
id: String
}
impl Id {
pub fn new<T: Into<String>>(id: T) -> Id {
Id {
id: id.into()
}
}
}
impl Link for Id {}
src/activity_pub/outbox.rs
+1 -57
View File
@@ -1,3 +1,4 @@
use activitystreams_traits::{Activity, Object};
use array_tool::vec::Uniq;
use diesel::PgConnection;
use reqwest::Client;
@@ -8,64 +9,7 @@ use serde_json;
use std::sync::Arc;
use activity_pub::{activity_pub, ActivityPub, context};
use activity_pub::activity::Activity;
use activity_pub::actor::Actor;
use activity_pub::request;
use activity_pub::sign::*;
pub struct Outbox {
id: String,
items: Vec<Arc<Activity>>
}
impl Outbox {
pub fn new(id: String, items: Vec<Arc<Activity>>) -> Outbox {
Outbox {
id: id,
items: items
}
}
fn serialize(&self) -> ActivityPub {
let items = self.items.clone().into_iter().map(|i| i.serialize()).collect::<Vec<serde_json::Value>>();
activity_pub(json!({
"@context": context(),
"type": "OrderedCollection",
"id": self.id,
"totalItems": items.len(),
"orderedItems": items
}))
}
}
impl<'r> Responder<'r> for Outbox {
fn respond_to(self, request: &Request) -> Result<Response<'r>, Status> {
self.serialize().respond_to(request)
}
}
pub fn broadcast<A: Activity + Clone, S: Actor + Signer, T: Actor>(conn: &PgConnection, sender: &S, act: A, to: Vec<T>) {
let boxes = to.into_iter()
.map(|u| u.get_shared_inbox_url().unwrap_or(u.get_inbox_url()))
.collect::<Vec<String>>()
.unique();
for inbox in boxes {
// TODO: run it in Sidekiq or something like that
let mut act = act.serialize();
act["@context"] = context();
let signed = act.sign(sender, conn);
let res = Client::new()
.post(&inbox[..])
.headers(request::headers())
.header(request::signature(sender, request::headers(), conn))
.header(request::digest(signed.to_string()))
.body(signed.to_string())
.send();
match res {
Ok(mut r) => println!("Successfully sent activity to inbox ({})\n\n{:?}", inbox, r.text().unwrap()),
Err(e) => println!("Error while sending to inbox ({:?})", e)
}
}
}
src/main.rs
+6
View File
@@ -1,10 +1,16 @@
#![feature(plugin, custom_derive, iterator_find_map)]
#![plugin(rocket_codegen)]
extern crate activitystreams;
extern crate activitystreams_traits;
extern crate activitystreams_types;
extern crate array_tool;
extern crate base64;
extern crate bcrypt;
extern crate chrono;
extern crate failure;
#[macro_use]
extern crate failure_derive;
extern crate heck;
extern crate hex;
#[macro_use]
src/models/blogs.rs
+8 -6
View File
@@ -1,3 +1,4 @@
use activitystreams_types::collection::OrderedCollection;
use reqwest::Client;
use reqwest::header::{Accept, qitem};
use reqwest::mime::Mime;
@@ -9,11 +10,9 @@ use openssl::hash::MessageDigest;
use openssl::pkey::{PKey, Private};
use openssl::rsa::Rsa;
use openssl::sign::Signer;
use std::sync::Arc;
use activity_pub::activity::Activity;
use activity_pub::ActivityStream;
use activity_pub::actor::{Actor, ActorType};
use activity_pub::outbox::Outbox;
use activity_pub::sign;
use activity_pub::webfinger::*;
use models::instance::Instance;
@@ -158,11 +157,14 @@ impl Blog {
}
}
pub fn outbox(&self, conn: &PgConnection) -> Outbox {
Outbox::new(self.compute_outbox(conn), self.get_activities(conn))
pub fn outbox(&self, conn: &PgConnection) -> ActivityStream<OrderedCollection> {
let mut coll = OrderedCollection::default();
coll.collection_props.items = serde_json::to_value(self.get_activities(conn)).unwrap();
coll.collection_props.set_total_items_u64(self.get_activities(conn).len() as u64).unwrap();
ActivityStream::new(coll)
}
fn get_activities(&self, _conn: &PgConnection) -> Vec<Arc<Activity>> {
fn get_activities(&self, _conn: &PgConnection) -> Vec<serde_json::Value> {
vec![]
}
src/models/instance.rs
+1 -1
View File
@@ -86,7 +86,7 @@ impl Instance {
impl Inbox for Instance {
fn received(&self, conn: &PgConnection, act: serde_json::Value) {
self.save(conn, act.clone());
self.save(conn, act.clone()).unwrap();
// TODO: add to stream, or whatever needs to be done
}
src/models/users.rs
+16 -9
View File
@@ -1,3 +1,7 @@
use activitystreams_types::{
activity::Create,
collection::OrderedCollection
};
use bcrypt;
use chrono::NaiveDateTime;
use diesel::{self, QueryDsl, RunQueryDsl, ExpressionMethods, BelongingToDsl, PgConnection};
@@ -12,15 +16,12 @@ use reqwest::mime::Mime;
use rocket::request::{self, FromRequest, Request};
use rocket::outcome::IntoOutcome;
use serde_json;
use std::sync::Arc;
use url::Url;
use BASE_URL;
use activity_pub::ap_url;
use activity_pub::activity::{Create, Activity};
use activity_pub::{ap_url, ActivityStream};
use activity_pub::actor::{ActorType, Actor};
use activity_pub::inbox::Inbox;
use activity_pub::outbox::Outbox;
use activity_pub::sign::{Signer, gen_keypair};
use activity_pub::webfinger::{Webfinger, resolve};
use db_conn::DbConn;
@@ -224,16 +225,22 @@ impl User {
}
}
pub fn outbox(&self, conn: &PgConnection) -> Outbox {
Outbox::new(self.compute_outbox(conn), self.get_activities(conn))
pub fn outbox(&self, conn: &PgConnection) -> ActivityStream<OrderedCollection> {
let mut coll = OrderedCollection::default(); // TODO
coll.collection_props.items = serde_json::to_value(self.get_activities(conn)).unwrap();
ActivityStream::new(coll)
}
fn get_activities(&self, conn: &PgConnection) -> Vec<Arc<Activity>> {
fn get_activities(&self, conn: &PgConnection) -> Vec<serde_json::Value> {
use schema::posts;
use schema::post_authors;
let posts_by_self = PostAuthor::belonging_to(self).select(post_authors::post_id);
let posts = posts::table.filter(posts::id.eq(any(posts_by_self))).load::<Post>(conn).unwrap();
posts.into_iter().map(|p| Arc::new(Create::new(self, &p, conn)) as Arc<Activity>).collect::<Vec<Arc<Activity>>>()
posts.into_iter().map(|_| {
// TODO Create::new(self, &p, conn)
// TODO: add a method to convert Post -> Create
serde_json::to_value(Create::default()).unwrap()
}).collect::<Vec<serde_json::Value>>()
}
pub fn get_fqn(&self, conn: &PgConnection) -> String {
@@ -352,7 +359,7 @@ impl Actor for User {
impl Inbox for User {
fn received(&self, conn: &PgConnection, act: serde_json::Value) {
self.save(conn, act.clone());
self.save(conn, act.clone()).unwrap();
// Notifications
match act["type"].as_str().unwrap() {
src/routes/blogs.rs
+3 -3
View File
@@ -1,11 +1,11 @@
use activitystreams_types::collection::OrderedCollection;
use rocket::request::Form;
use rocket::response::Redirect;
use rocket_contrib::Template;
use serde_json;
use activity_pub::ActivityPub;
use activity_pub::{ActivityStream, ActivityPub};
use activity_pub::actor::Actor;
use activity_pub::outbox::Outbox;
use db_conn::DbConn;
use models::blog_authors::*;
use models::blogs::*;
@@ -78,7 +78,7 @@ fn create(conn: DbConn, data: Form<NewBlogForm>, user: User) -> Redirect {
}
#[get("/~/<name>/outbox")]
fn outbox(name: String, conn: DbConn) -> Outbox {
fn outbox(name: String, conn: DbConn) -> ActivityStream<OrderedCollection> {
let blog = Blog::find_local(&*conn, name).unwrap();
blog.outbox(&*conn)
}
src/routes/comments.rs
+4 -4
View File
@@ -1,9 +1,9 @@
use activitystreams_types::activity::Create;
use rocket::request::Form;
use rocket::response::Redirect;
use rocket_contrib::Template;
use activity_pub::activity::Create;
use activity_pub::outbox::broadcast;
use activity_pub::broadcast;
use db_conn::DbConn;
use models::comments::*;
use models::posts::Post;
@@ -41,8 +41,8 @@ fn create(blog: String, slug: String, query: CommentQuery, data: Form<NewComment
sensitive: false,
spoiler_text: "".to_string()
});
let act = Create::new(&user, &comment, &*conn);
broadcast(&*conn, &user, act, user.get_followers(&*conn));
// TODO: let act = Create::new(&user, &comment, &*conn);
// broadcast(&*conn, &user, act, user.get_followers(&*conn));
Redirect::to(format!("/~/{}/{}/#comment-{}", blog, slug, comment.id).as_ref())
}
src/routes/likes.rs
+6 -6
View File
@@ -1,7 +1,7 @@
use activitystreams_types::activity::{Like, Undo};
use rocket::response::Redirect;
use activity_pub::activity::{Like, Undo};
use activity_pub::outbox::broadcast;
use activity_pub::broadcast;
use db_conn::DbConn;
use models::likes;
use models::posts::Post;
@@ -18,12 +18,12 @@ fn create(blog: String, slug: String, user: User, conn: DbConn) -> Redirect {
ap_url: "".to_string()
});
like.update_ap_url(&*conn);
let act = Like::new(&user, &post, &*conn);
broadcast(&*conn, &user, act, user.get_followers(&*conn));
// TODO: let act = Like::new(&user, &post, &*conn);
// TODO: broadcast(&*conn, &user, act, user.get_followers(&*conn));
} else {
let like = likes::Like::find_by_user_on_post(&*conn, &user, &post).unwrap();
like.delete(&*conn);
broadcast(&*conn, &user, Undo::new(&user, &like, &*conn), user.get_followers(&*conn));
// TODO: like.delete(&*conn);
// TODO: broadcast(&*conn, &user, Undo::new(&user, &like, &*conn), user.get_followers(&*conn));
}
Redirect::to(format!("/~/{}/{}/", blog, slug).as_ref())
src/routes/posts.rs
+8 -5
View File
@@ -1,13 +1,12 @@
use activitystreams_types::activity::Create;
use heck::KebabCase;
use rocket::request::Form;
use rocket::response::Redirect;
use rocket_contrib::Template;
use serde_json;
use activity_pub::{context, activity_pub, ActivityPub};
use activity_pub::activity::Create;
use activity_pub::{broadcast, context, activity_pub, ActivityPub, Id};
use activity_pub::object::Object;
use activity_pub::outbox::broadcast;
use db_conn::DbConn;
use models::blogs::*;
use models::comments::Comment;
@@ -86,8 +85,12 @@ fn create(blog_name: String, data: Form<NewPostForm>, user: User, conn: DbConn)
author_id: user.id
});
let act = Create::new(&user, &post, &*conn);
broadcast(&*conn, &user, act, user.get_followers(&*conn));
// TODO: use Post -> Create conversion
// let act = Create::default();
// act.object_props.set_id_string(format!("{}/activity", post.compute_id(&*conn)));
// act.set_actor_link(Id::new(user.ap_url));
// act.set_object_object();
// broadcast(&*conn, &user, act, user.get_followers(&*conn));
Redirect::to(format!("/~/{}/{}", blog_name, slug).as_str())
}
src/routes/user.rs
+11 -6
View File
@@ -1,14 +1,17 @@
use activitystreams_types::{
activity::Follow,
collection::OrderedCollection
};
use rocket::request::Form;
use rocket::response::Redirect;
use rocket_contrib::Template;
use serde_json;
use activity_pub::{activity, activity_pub, ActivityPub, context};
use activity_pub::{activity_pub, ActivityPub, ActivityStream, context, broadcast};
use activity_pub::actor::Actor;
use activity_pub::inbox::Inbox;
use activity_pub::outbox::{broadcast, Outbox};
use db_conn::DbConn;
use models::follows::*;
use models::follows;
use models::instance::Instance;
use models::posts::Post;
use models::users::*;
@@ -49,11 +52,13 @@ fn details(name: String, conn: DbConn, account: Option<User>) -> Template {
#[get("/@/<name>/follow")]
fn follow(name: String, conn: DbConn, user: User) -> Redirect {
let target = User::find_by_fqn(&*conn, name.clone()).unwrap();
Follow::insert(&*conn, NewFollow {
follows::Follow::insert(&*conn, follows::NewFollow {
follower_id: user.id,
following_id: target.id
});
broadcast(&*conn, &user, activity::Follow::new(&user, &target, &*conn), vec![target]);
let act = Follow::default();
// TODO
broadcast(&*conn, &user, act, vec![target]);
Redirect::to(format!("/@/{}", name).as_ref())
}
@@ -145,7 +150,7 @@ fn create(conn: DbConn, data: Form<NewUserForm>) -> Redirect {
}
#[get("/@/<name>/outbox")]
fn outbox(name: String, conn: DbConn) -> Outbox {
fn outbox(name: String, conn: DbConn) -> ActivityStream<OrderedCollection> {
let user = User::find_local(&*conn, name).unwrap();
user.outbox(&*conn)
}