Merge branch 'master' of github.com:revoltchat/backend into webhooks

crates/quark/Cargo.toml

@@ -69,6 +69,7 @@ reqwest = "0.11.10"
 bitfield = "0.13.2"
 once_cell = "1.13.0"
 lazy_static = "1.4.0"
+async-lock = "2.6.0"
 
 lru = { version = "0.7.6", optional = true }
 dashmap = { version = "5.2.0", optional = true }

crates/quark/src/tasks/process_embeds.rs

@@ -1,11 +1,14 @@
-use crate::util::variables::delta::{JANUARY_URL, MAX_EMBED_COUNT};
+use crate::util::variables::delta::{JANUARY_URL, MAX_EMBED_COUNT, JANUARY_CONCURRENT_CONNECTIONS};
 use crate::{
     models::{message::AppendMessage, Message},
     types::january::Embed,
     Database,
 };
 
+use async_lock::Semaphore;
+use async_std::task::spawn;
 use deadqueue::limited::Queue;
+use std::sync::Arc;
 
 /// Task information
 #[derive(Debug)]
@@ -36,21 +39,30 @@ pub async fn queue(channel: String, id: String, content: String) {
 
 /// Start a new worker
 pub async fn worker(db: Database) {
+    let semaphore = Arc::new(Semaphore::new(*JANUARY_CONCURRENT_CONNECTIONS));
+
     loop {
         let task = Q.pop().await;
-        if let Ok(embeds) = Embed::generate(task.content, &JANUARY_URL, *MAX_EMBED_COUNT).await {
-            if let Err(err) = Message::append(
-                &db,
-                task.id,
-                task.channel,
-                AppendMessage {
-                    embeds: Some(embeds),
-                },
-            )
-            .await
-            {
-                error!("Encountered an error appending to message: {:?}", err);
+        let db = db.clone();
+        let semaphore = semaphore.clone();
+
+        spawn(async move {
+            let embeds = Embed::generate(task.content, &JANUARY_URL, *MAX_EMBED_COUNT, semaphore).await;
+
+            if let Ok(embeds) = embeds {
+                if let Err(err) = Message::append(
+                    &db,
+                    task.id,
+                    task.channel,
+                    AppendMessage {
+                        embeds: Some(embeds),
+                    },
+                )
+                .await
+                {
+                    error!("Encountered an error appending to message: {:?}", err);
+                }
             }
-        }
+        });
     }
 }

crates/quark/src/types/january.rs

@@ -1,7 +1,10 @@
+use async_lock::Semaphore;
+use async_std::task::spawn;
+use futures::future::join_all;
 use linkify::{LinkFinder, LinkKind};
 use regex::Regex;
 use serde::{Deserialize, Serialize};
-use std::collections::HashSet;
+use std::{collections::HashSet, sync::Arc};
 
 use crate::{models::attachment::File, Error, Result};
 
@@ -95,6 +98,8 @@ pub enum Special {
         content_type: BandcampType,
         id: String,
     },
+    /// Streamable Video
+    Streamable { id: String },
 }
 
 /// Website metadata
@@ -172,7 +177,12 @@ pub enum Embed {
 
 impl Embed {
     /// Generate embeds from given content
-    pub async fn generate(content: String, host: &str, max_embeds: usize) -> Result<Vec<Embed>> {
+    pub async fn generate(
+        content: String,
+        host: &str,
+        max_embeds: usize,
+        semaphore: Arc<Semaphore>,
+    ) -> Result<Vec<Embed>> {
         lazy_static! {
             static ref RE_CODE: Regex = Regex::new("```(?:.|\n)+?```|`(?:.|\n)+?`").unwrap();
             static ref RE_IGNORED: Regex = Regex::new("(<http.+>)").unwrap();
@@ -223,26 +233,37 @@ impl Embed {
         }
 
         // ! FIXME: batch request to january?
-        let mut embeds: Vec<Embed> = Vec::new();
         let client = reqwest::Client::new();
+
+        let mut tasks = Vec::new();
+        let url = format!("{host}/embed");
+
         for link in links {
-            let result = client
-                .get(&format!("{}/embed", host))
-                .query(&[("url", link)])
-                .send()
-                .await;
+            let client = client.clone();
+            let url = url.clone();
+            let semaphore = semaphore.clone();
 
-            if result.is_err() {
-                continue;
-            }
+            tasks.push(spawn(async move {
+                let guard = semaphore.acquire().await;
 
-            let response = result.unwrap();
-            if response.status().is_success() {
-                let res: Embed = response.json().await.map_err(|_| Error::InvalidOperation)?;
-                embeds.push(res);
-            }
+                let response = client.get(url).query(&[("url", link)]).send().await.ok()?;
+
+                drop(guard);
+
+                if response.status().is_success() {
+                    response.json::<Embed>().await.ok()
+                } else {
+                    None
+                }
+            }));
         }
 
+        let embeds = join_all(tasks)
+            .await
+            .into_iter()
+            .flatten()
+            .collect::<Vec<Embed>>();
+
         // Prevent database update when no embeds are found.
         if !embeds.is_empty() {
             Ok(embeds)

crates/quark/src/util/variables/delta.rs

@@ -13,6 +13,8 @@ lazy_static! {
         env::var("AUTUMN_PUBLIC_URL").unwrap_or_else(|_| "https://example.com".to_string());
     pub static ref JANUARY_URL: String =
         env::var("JANUARY_PUBLIC_URL").unwrap_or_else(|_| "https://example.com".to_string());
+    pub static ref JANUARY_CONCURRENT_CONNECTIONS: usize =
+        env::var("JANUARY_CONCURRENT_CONNECTIONS").map_or(50, |v| v.parse().unwrap());
     pub static ref VOSO_URL: String =
         env::var("VOSO_PUBLIC_URL").unwrap_or_else(|_| "https://example.com".to_string());
     pub static ref VOSO_WS_HOST: String =