flat import (untested) + parallel fed download

1 files changed, 51 insertions, 33 deletions

diff --git a/server/src/import.rs b/server/src/import.rs
index ccf5274..8908c00 100644
--- a/server/src/import.rs
+++ b/server/src/import.rs
@@ -112,7 +112,7 @@ pub async fn import_path(
                 .await
                 .context("creating session")?;
 
-            import_remote(io, db, &session, &identifier, parent)
+            import_remote(io, db, &session, identifier.clone(), parent)
                 .await
                 .context("federated import")?;
         } else {
@@ -126,14 +126,17 @@ pub async fn import_path(
     }
 }
 
+static SEM_REMOTE_IMPORT: LazyLock<Semaphore> = LazyLock::new(|| Semaphore::new(16));
+
 #[async_recursion]
 async fn import_remote(
     opts: RemoteImportOptions,
     db: &Database,
     session: &Session,
-    identifier: &String,
+    identifier: String,
     parent: Option<String>,
 ) -> anyhow::Result<()> {
+    let _permit = SEM_REMOTE_IMPORT.acquire().await.unwrap();
     info!("loading federated node {identifier:?}");
     let node = session
         .node(&opts.id)
@@ -143,40 +146,55 @@ async fn import_remote(
     let poster = cache_federation_asset(session, &opts.id, "poster").await?;
     let backdrop = cache_federation_asset(session, &opts.id, "backdrop").await?;
 
-    let mut node = Node {
-        public: node,
-        private: NodePrivate {
-            backdrop: Some(AssetLocation::Cache(backdrop)),
-            poster: Some(AssetLocation::Cache(poster)),
-            import: None,
-            id: None,
-            source: Some(MediaSource::Remote {
-                host: opts.host.clone(),
-                remote_id: opts.id.clone(),
-            }),
-        },
-    };
+    drop(_permit);
 
-    for c in &mut node.public.children {
-        let prefixed = format!("{}{c}", opts.prefix.clone().unwrap_or(String::new()));
-        import_remote(
-            RemoteImportOptions {
-                id: c.to_owned(),
-                ..opts.clone()
+    let child_parent = if opts.flatten {
+        parent
+    } else {
+        let mut node = Node {
+            public: node.clone(),
+            private: NodePrivate {
+                backdrop: Some(AssetLocation::Cache(backdrop)),
+                poster: Some(AssetLocation::Cache(poster)),
+                import: None,
+                id: None,
+                source: Some(MediaSource::Remote {
+                    host: opts.host.clone(),
+                    remote_id: opts.id.clone(),
+                }),
             },
-            db,
-            session,
-            &prefixed,
-            Some(opts.id.clone()),
-        )
-        .await?;
-        *c = prefixed;
-    }
+        };
+        node.public.parent = parent;
+        node.public.federated = Some(opts.host.clone());
+        node.public
+            .children
+            .iter_mut()
+            .for_each(|c| *c = format!("{}{c}", opts.prefix.clone().unwrap_or(String::new())));
+
+        debug!("adding {identifier}");
+        db.node.insert(&identifier, &node)?;
+        Some(opts.id.clone())
+    };
+
+    let mut children: FuturesUnordered<_> = node
+        .children
+        .iter()
+        .map(|c| {
+            let prefixed = format!("{}{c}", opts.prefix.clone().unwrap_or(String::new()));
+            import_remote(
+                RemoteImportOptions {
+                    id: c.to_owned(),
+                    ..opts.clone()
+                },
+                db,
+                session,
+                prefixed,
+                child_parent.clone(),
+            )
+        })
+        .collect();
 
-    node.public.parent = parent;
-    node.public.federated = Some(opts.host.clone());
-    debug!("adding {identifier}");
-    db.node.insert(identifier, &node)?;
+    while let Some(_) = children.next().await {}
 
     Ok(())
 }