Initial Matrix implementation

CHANGELOG.md

@@ -1 +1,3 @@
-- Return API error response on missing query parameter
+- Initial Matrix implementation
+    - Allows bots to know, when a new meme was uploaded
+    - First step towards decentralization

src/cdn/error.rs

@@ -7,14 +7,14 @@ use axum::{
 use reqwest::StatusCode;
 use thiserror::Error;
 
-use crate::ipfs::error::IPFSError;
+use crate::error::ServiceError;
 
 #[derive(Error, Debug)]
 pub enum CDNError {
     #[error("SQL error: {0}")]
     Sql(#[from] sqlx::Error),
-    #[error("IPFS error: {0}")]
+    #[error("JMService error: {0}")]
-    Ipfs(#[from] IPFSError),
+    Service(#[from] ServiceError),
     #[error("Decode error: {0}")]
     Decode(#[from] FromUtf8Error),
     #[error("Internal server error")]

src/config.rs

@@ -10,6 +10,9 @@ pub struct Config {
     pub database: String,
     pub cdn: String,
     pub ipfs_api: Url,
+    pub matrix_url: Url,
+    pub matrix_token: String,
+    pub matrix_domain: String,
 }
 
 impl Config {
@@ -19,6 +22,9 @@ impl Config {
             client,
             ipfs_url: self.ipfs_api.clone(),
             cdn_url: self.cdn.clone(),
+            matrix_url: self.matrix_url.clone(),
+            matrix_token: self.matrix_token.clone(),
+            matrix_domain: self.matrix_domain.clone(),
         }))
     }
 }

src/error.rs

@@ -1,4 +1,6 @@
+use hyper::StatusCode;
 use thiserror::Error;
+use url::ParseError;
 
 #[derive(Error, Debug)]
 pub enum JMError {
@@ -13,3 +15,13 @@ pub enum JMError {
     #[error("Reqwest error: {0}")]
     Reqwest(#[from] reqwest::Error),
 }
+
+#[derive(Error, Debug)]
+pub enum ServiceError {
+    #[error("Reqwest error: {0}")]
+    Reqwest(#[from] reqwest::Error),
+    #[error("URL parse error: {0}")]
+    Url(#[from] ParseError),
+    #[error("Invalid response code: {0}")]
+    InvalidResponse(StatusCode),
+}

src/ipfs/error.rs

@@ -1,10 +0,0 @@
-use thiserror::Error;
-use url::ParseError;
-
-#[derive(Error, Debug)]
-pub enum IPFSError {
-    #[error("Reqwest error: {0}")]
-    Reqwest(#[from] reqwest::Error),
-    #[error("URL parse error: {0}")]
-    Url(#[from] ParseError),
-}

src/ipfs/mod.rs

@@ -7,11 +7,7 @@ use reqwest::{
 };
 use serde::{Deserialize, Serialize};
 
-use crate::JMServiceInner;
+use crate::{error::ServiceError, JMServiceInner};
-
-use self::error::IPFSError;
-
-pub(crate) mod error;
 
 #[derive(Deserialize)]
 pub struct IPFSFile {
@@ -39,7 +35,7 @@ pub struct PinQuery {
 }
 
 impl JMServiceInner {
-    pub async fn cat(&self, cid: String) -> Result<Response, IPFSError> {
+    pub async fn cat(&self, cid: String) -> Result<Response, ServiceError> {
         let request = self
             .client
             .post(self.ipfs_url.join("/api/v0/cat")?)
@@ -47,7 +43,7 @@ impl JMServiceInner {
         Ok(request.send().await?)
     }
 
-    pub async fn add(&self, file: Bytes, filename: String) -> Result<IPFSFile, IPFSError> {
+    pub async fn add(&self, file: Bytes, filename: String) -> Result<IPFSFile, ServiceError> {
         let request = self
             .client
             .post(self.ipfs_url.join("/api/v0/add")?)
@@ -58,7 +54,7 @@ impl JMServiceInner {
         Ok(res)
     }
 
-    pub async fn pin(&self, cid: String) -> Result<(), IPFSError> {
+    pub async fn pin(&self, cid: String) -> Result<(), ServiceError> {
         let request = self
             .client
             .post(self.ipfs_url.join("/api/v0/pin/add")?)

src/main.rs

@@ -16,6 +16,7 @@ mod config;
 mod error;
 mod ipfs;
 mod lib;
+mod matrix;
 mod models;
 mod sql;
 mod v1;
@@ -35,6 +36,9 @@ pub struct JMServiceInner {
     client: Client,
     ipfs_url: Url,
     cdn_url: String,
+    matrix_url: Url,
+    matrix_token: String,
+    matrix_domain: String,
 }
 
 pub type JMService = Arc<JMServiceInner>;

src/matrix/mod.rs

@@ -0,0 +1,158 @@
+use serde::{Deserialize, Serialize};
+
+use crate::{error::ServiceError, JMServiceInner};
+
+#[derive(Serialize)]
+pub struct Meme {
+    pub category: String,
+    pub filename: String,
+    pub cid: String,
+}
+
+#[derive(Serialize, Deserialize)]
+pub struct UserID {
+    pub user_id: String,
+}
+
+#[derive(Deserialize)]
+pub struct RoomID {
+    pub room_id: String,
+}
+
+#[derive(Deserialize, Serialize)]
+pub struct EventID {
+    pub event_id: String,
+}
+
+#[derive(Serialize)]
+pub struct RegisterRequest {
+    #[serde(rename = "type")]
+    pub reg_type: String,
+    pub username: String,
+}
+
+impl JMServiceInner {
+    pub async fn add_meme(
+        &self,
+        category: String,
+        filename: String,
+        cid: String,
+        user: String,
+        id: u64,
+    ) -> Result<(), ServiceError> {
+        let meme = Meme {
+            category,
+            filename,
+            cid,
+        };
+        let txid = meme.calc_txid(user.clone());
+        let usr = self.check_user(user).await?;
+        let room_id = self.join_room(&usr).await?;
+        let path = format!(
+            "/_matrix/client/r0/rooms/{}/send/es.jensmem.meme/{}",
+            &room_id, txid
+        );
+        let url = self.matrix_url.join(path.as_str())?;
+        let req = self
+            .client
+            .put(url)
+            .bearer_auth(self.matrix_token.clone())
+            .query(&usr)
+            .json(&meme);
+        let res = req.send().await?;
+        if res.status().is_success() {
+            let event: EventID = res.json().await?;
+            let path = format!(
+                "/_matrix/client/r0/rooms/{}/state/es.jensmem.index/{}",
+                &room_id, id
+            );
+            let req = self
+                .client
+                .put(self.matrix_url.join(path.as_str())?)
+                .bearer_auth(self.matrix_token.clone())
+                .json(&event);
+            let res = req.send().await?;
+            if res.status().is_success() {
+                Ok(())
+            } else {
+                Err(ServiceError::InvalidResponse(res.status()))
+            }
+        } else {
+            Err(ServiceError::InvalidResponse(res.status()))
+        }
+    }
+
+    async fn check_user(&self, user: String) -> Result<UserID, ServiceError> {
+        let username = format!("jm_{}", user);
+        let user = self.get_mxid(username.clone());
+        let req = self
+            .client
+            .get(self.matrix_url.join("/_matrix/client/r0/account/whoami")?)
+            .bearer_auth(self.matrix_token.clone())
+            .query(&user);
+        let res = req.send().await?;
+        if res.status().is_success() {
+            let mxid: UserID = res.json().await?;
+            Ok(mxid)
+        } else {
+            let mxid = self.register_user(username).await?;
+            Ok(mxid)
+        }
+    }
+
+    async fn register_user(&self, username: String) -> Result<UserID, ServiceError> {
+        let req = self
+            .client
+            .post(self.matrix_url.join("/_matrix/client/r0/register")?)
+            .bearer_auth(self.matrix_token.clone())
+            .json(&RegisterRequest::new(username));
+        let res = req.send().await?;
+        if res.status().is_success() {
+            let user: UserID = res.json().await?;
+            Ok(user)
+        } else {
+            Err(ServiceError::InvalidResponse(res.status()))
+        }
+    }
+
+    async fn join_room(&self, user: &UserID) -> Result<String, ServiceError> {
+        let req = self
+            .client
+            .post(
+                self.matrix_url
+                    .join("/_matrix/client/r0/join/%23memes%3Atilera.org")?,
+            )
+            .bearer_auth(self.matrix_token.clone())
+            .query(user);
+
+        let res = req.send().await?;
+        if res.status().is_success() {
+            let room: RoomID = res.json().await?;
+            Ok(room.room_id)
+        } else {
+            Err(ServiceError::InvalidResponse(res.status()))
+        }
+    }
+
+    fn get_mxid(&self, username: String) -> UserID {
+        UserID {
+            user_id: format!("@{}:{}", username, self.matrix_domain.clone()),
+        }
+    }
+}
+
+impl RegisterRequest {
+    pub fn new(username: String) -> Self {
+        Self {
+            reg_type: "m.login.application_service".to_string(),
+            username,
+        }
+    }
+}
+
+impl Meme {
+    pub fn calc_txid(&self, user: String) -> String {
+        let txid = format!("{}/{}/{}/{}", user, self.category, self.filename, self.cid);
+        urlencoding::encode(txid.as_str()).into_owned()
+    }
+}

src/v1/error.rs

@@ -10,7 +10,7 @@ use reqwest::StatusCode;
 use thiserror::Error;
 
 use super::models::ErrorResponse;
-use crate::ipfs::error::IPFSError;
+use crate::error::ServiceError;
 
 #[derive(Error, Debug)]
 pub enum APIError {
@@ -28,8 +28,8 @@ pub enum APIError {
     NotFound(String),
     #[error("{0}")]
     Internal(String),
-    #[error("IPFS error: {0}")]
+    #[error("JMService error: {0}")]
-    Ipfs(#[from] IPFSError),
+    Service(#[from] ServiceError),
     #[error("Query rejection: {0}")]
     Query(#[from] QueryRejection),
 }
@@ -60,10 +60,8 @@ impl IntoResponse for APIError {
             APIError::Unauthorized(err) => ErrorResponse::new(StatusCode::UNAUTHORIZED, Some(err)),
             APIError::Forbidden(err) => ErrorResponse::new(StatusCode::FORBIDDEN, Some(err)),
             APIError::NotFound(err) => ErrorResponse::new(StatusCode::NOT_FOUND, Some(err)),
-            APIError::Internal(err) => {
+            APIError::Internal(err) => ErrorResponse::new(StatusCode::INTERNAL_SERVER_ERROR, Some(err)),
-                ErrorResponse::new(StatusCode::INTERNAL_SERVER_ERROR, Some(err))
+            APIError::Service(_) => ErrorResponse::new(StatusCode::INTERNAL_SERVER_ERROR, None),
-            }
-            APIError::Ipfs(_) => ErrorResponse::new(StatusCode::INTERNAL_SERVER_ERROR, None),
             APIError::Query(_) => ErrorResponse::new(StatusCode::BAD_REQUEST, None),
         };
         let status = res.status;

src/v1/routes.rs

@@ -166,7 +166,15 @@ async fn upload(
         if res == 0 {
             return Err(APIError::Internal("Database insertion error".to_string()));
         }
-
+        service
+            .add_meme(
+                cat.id.clone(),
+                f.name.clone(),
+                f.hash.clone(),
+                user.id.clone(),
+                res,
+            )
+            .await?;
         service.pin(f.hash).await?;
         links.push(format!(
             "{}/{}/{}",