mostly ignore errors when importing

4 files changed, 112 insertions, 54 deletions

diff --git a/import/src/db.rs b/import/src/db.rs
index e6174e4..87350ac 100644
--- a/import/src/db.rs
+++ b/import/src/db.rs
@@ -1,6 +1,6 @@
 use anyhow::{anyhow, Context};
 use jellybase::database::{
-    redb::ReadableTable,
+    redb::{ReadableTable, ReadableTableMetadata},
     tantivy::{doc, DateTime},
     DataAcid, Ser, T_NODE, T_NODE_EXTENDED, T_NODE_IMPORT,
 };
@@ -21,6 +21,7 @@ pub(crate) trait ImportStorage: Sync {
     fn get_partial_parts(&self, id: &str) -> anyhow::Result<Vec<(Vec<usize>, Node)>>;
     fn insert_complete_node(&self, id: &str, node: Node) -> anyhow::Result<()>;
 
+    fn pre_clean(&self) -> anyhow::Result<()>;
     fn remove_prev_nodes(&self) -> anyhow::Result<()>;
     fn finish(&self) -> anyhow::Result<()>;
 }
@@ -34,6 +35,17 @@ impl<'a> DatabaseStorage<'a> {
     }
 }
 impl ImportStorage for DatabaseStorage<'_> {
+    fn pre_clean(&self) -> anyhow::Result<()> {
+        let txn = self.db.begin_write()?;
+        let mut table = txn.open_table(T_NODE_IMPORT)?;
+        if !table.is_empty()? {
+            info!("clearing temporary node tree from an aborted last import...");
+            table.retain(|_, _| false)?;
+        }
+        drop(table);
+        txn.commit()?;
+        Ok(())
+    }
     fn remove_prev_nodes(&self) -> anyhow::Result<()> {
         info!("removing old nodes...");
         let txn = self.db.inner.begin_write()?;
@@ -107,6 +119,9 @@ impl<'a> MemoryStorage<'a> {
     }
 }
 impl ImportStorage for MemoryStorage<'_> {
+    fn pre_clean(&self) -> anyhow::Result<()> {
+        Ok(())
+    }
     fn remove_prev_nodes(&self) -> anyhow::Result<()> {
         info!("removing old nodes...");
         let txn = self.db.inner.begin_write()?;
diff --git a/import/src/lib.rs b/import/src/lib.rs
index 3adfd85..49f709a 100644
--- a/import/src/lib.rs
+++ b/import/src/lib.rs
@@ -9,7 +9,7 @@ pub mod infojson;
 pub mod tmdb;
 pub mod trakt;
 
-use anyhow::{anyhow, bail, Context, Ok};
+use anyhow::{anyhow, bail, Context, Error, Ok};
 use async_recursion::async_recursion;
 use base64::Engine;
 use db::{DatabaseStorage, ImportStorage, MemoryStorage};
@@ -17,7 +17,7 @@ use futures::{stream::FuturesUnordered, StreamExt};
 use jellybase::{
     assetfed::AssetInner,
     cache::{async_cache_file, cache_memory},
-    database::{redb::ReadableTableMetadata, DataAcid, T_NODE_IMPORT},
+    database::DataAcid,
     federation::Federation,
     CONF, SECRETS,
 };
@@ -43,11 +43,17 @@ use std::{
     sync::{Arc, LazyLock},
 };
 use tmdb::{parse_release_date, Tmdb};
-use tokio::{io::AsyncWriteExt, sync::Semaphore, task::spawn_blocking};
+use tokio::{
+    io::AsyncWriteExt,
+    sync::{RwLock, Semaphore},
+    task::spawn_blocking,
+};
 use trakt::Trakt;
 
 static IMPORT_SEM: LazyLock<Semaphore> = LazyLock::new(|| Semaphore::new(1));
 
+pub static IMPORT_ERRORS: RwLock<Vec<String>> = RwLock::const_new(Vec::new());
+
 struct Apis {
     trakt: Option<Trakt>,
     tmdb: Option<Tmdb>,
@@ -59,27 +65,22 @@ pub fn is_importing() -> bool {
 
 pub async fn import(db: &DataAcid, fed: &Federation) -> anyhow::Result<()> {
     let permit = IMPORT_SEM.try_acquire()?;
-    {
-        let txn = db.inner.begin_write()?;
-        let mut table = txn.open_table(T_NODE_IMPORT)?;
-        if !table.is_empty()? {
-            info!("clearing temporary node tree from an aborted last import...");
-            table.retain(|_, _| false)?;
-        }
-        drop(table);
-        txn.commit()?;
-    }
 
     let ap = Apis {
         trakt: SECRETS.api.trakt.as_ref().map(|key| Trakt::new(key)),
         tmdb: SECRETS.api.tmdb.as_ref().map(|key| Tmdb::new(key)),
     };
 
-    if CONF.use_in_memory_import_storage {
-        import_inner(&MemoryStorage::new(db), fed, &ap).await?;
+    let e = if CONF.use_in_memory_import_storage {
+        import_inner(&MemoryStorage::new(db), fed, &ap).await
     } else {
-        import_inner(&DatabaseStorage::new(db), fed, &ap).await?;
-    }
+        import_inner(&DatabaseStorage::new(db), fed, &ap).await
+    };
+    let e = match e {
+        Result::Ok(e) => e,
+        Result::Err(e) => vec![e],
+    };
+    *IMPORT_ERRORS.write().await = e.into_iter().map(|e| format!("{e:?}")).collect();
 
     drop(permit);
     Ok(())
@@ -89,27 +90,37 @@ pub(crate) async fn import_inner(
     db: &impl ImportStorage,
     fed: &Federation,
     ap: &Apis,
-) -> anyhow::Result<()> {
+) -> anyhow::Result<Vec<anyhow::Error>> {
+    db.pre_clean()?;
     info!("loading sources...");
-    import_path(CONF.library_path.clone(), vec![], db, fed, ap)
+    let mut errors = Vec::new();
+    match import_path(CONF.library_path.clone(), vec![], db, fed, ap)
         .await
-        .context("indexing")?;
+        .context("indexing")
+    {
+        Result::Ok(o) => errors.extend(o),
+        Result::Err(e) => errors.push(e),
+    };
     db.remove_prev_nodes()?;
     info!("merging nodes...");
-    generate_node_paths(db).context("merging nodes")?;
+    match generate_node_paths(db).context("merging nodes") {
+        Result::Ok(o) => errors.extend(o),
+        Result::Err(e) => errors.push(e),
+    }
     db.finish()?;
     info!("import completed");
-    Ok(())
+    Ok(errors)
 }
 
-fn generate_node_paths(db: &impl ImportStorage) -> anyhow::Result<()> {
+fn generate_node_paths(db: &impl ImportStorage) -> anyhow::Result<Vec<Error>> {
     // TODO mark nodes done to allow recursion
     fn traverse(
         db: &impl ImportStorage,
         id: String,
         mut path: Vec<String>,
         parent_title: &str,
-    ) -> anyhow::Result<()> {
+    ) -> anyhow::Result<Vec<Error>> {
+        let mut errors = Vec::new();
         let node = {
             let mut parts = db
                 .get_partial_parts(&id)
@@ -166,12 +177,14 @@ fn generate_node_paths(db: &impl ImportStorage) -> anyhow::Result<()> {
         path.push(id);
         let ps = node.public.title.unwrap_or_default();
         for c in node.public.children {
-            traverse(db, c, path.clone(), &ps)?;
+            match traverse(db, c, path.clone(), &ps) {
+                Result::Ok(o) => errors.extend(o),
+                Result::Err(e) => errors.push(e),
+            }
         }
-        Ok(())
+        Ok(errors)
     }
-    traverse(db, "library".to_string(), vec![], "Root")?;
-    Ok(())
+    traverse(db, "library".to_string(), vec![], "Root")
 }
 
 fn compare_index_path(x: &[usize], y: &[usize]) -> Ordering {
@@ -194,7 +207,8 @@ async fn import_path(
     db: &impl ImportStorage,
     fed: &Federation,
     ap: &Apis,
-) -> anyhow::Result<()> {
+) -> anyhow::Result<Vec<anyhow::Error>> {
+    let mut errors = Vec::new();
     if path.is_dir() {
         let mut children_paths = path
             .read_dir()?
@@ -232,7 +246,10 @@ async fn import_path(
             .collect();
 
         while let Some(k) = children.next().await {
-            k?
+            match k {
+                Result::Ok(o) => errors.extend(o),
+                Result::Err(e) => errors.push(e),
+            }
         }
     } else {
         info!("reading {path:?}");
@@ -245,13 +262,16 @@ async fn import_path(
 
         for (i, s) in opts.sources.into_iter().enumerate() {
             index_path.push(i);
-            process_source(opts.id.clone(), s, &path, &index_path, db, fed, ap)
+            if let Err(e) = process_source(opts.id.clone(), s, &path, &index_path, db, fed, ap)
                 .await
-                .context(anyhow!("processing source in {path:?}"))?;
+                .context(anyhow!("processing source in {path:?}"))
+            {
+                errors.push(e)
+            }
             index_path.pop();
         }
     }
-    Ok(())
+    Ok(errors)
 }
 
 static SEM_IMPORT: Semaphore = Semaphore::const_new(2);
@@ -265,7 +285,8 @@ async fn process_source(
     db: &impl ImportStorage,
     fed: &Federation,
     ap: &Apis,
-) -> anyhow::Result<()> {
+) -> anyhow::Result<Vec<anyhow::Error>> {
+    let mut errors = vec![];
     match s {
         ImportSource::Override(mut n) => {
             if let Some(backdrop) = n.private.backdrop.clone() {
@@ -351,7 +372,7 @@ async fn process_source(
                 } {
                     let mut index_path = index_path.to_vec();
                     index_path.push(1);
-                    process_source(
+                    match process_source(
                         id,
                         ImportSource::Tmdb { id: tid, kind },
                         path,
@@ -360,7 +381,11 @@ async fn process_source(
                         fed,
                         ap,
                     )
-                    .await?;
+                    .await
+                    {
+                        Result::Ok(o) => errors.extend(o),
+                        Result::Err(e) => errors.push(e),
+                    }
                 }
             }
         }
@@ -420,7 +445,8 @@ async fn process_source(
                     {
                         let inf_id =
                             infer_id_from_path(&child_path).context("inferring child id")?;
-                        process_source(
+
+                        match process_source(
                             inf_id.clone(),
                             ImportSource::Media {
                                 path: mpath.join(f.file_name()),
@@ -435,7 +461,11 @@ async fn process_source(
                             ap,
                         )
                         .await
-                        .context(anyhow!("recursive media import: {:?}", f.path()))?;
+                        .context(anyhow!("recursive media import: {:?}", f.path()))
+                        {
+                            Result::Ok(o) => errors.extend(o),
+                            Result::Err(e) => errors.push(e),
+                        };
                         node.public.children.push(inf_id);
                     }
                 }
@@ -581,7 +611,7 @@ async fn process_source(
             )?;
         }
     }
-    Ok(())
+    Ok(errors)
 }
 
 const RE_YOUTUBE_ID: LazyLock<Regex> =
diff --git a/server/src/routes/ui/admin/mod.rs b/server/src/routes/ui/admin/mod.rs
index 00a68ac..540ac72 100644
--- a/server/src/routes/ui/admin/mod.rs
+++ b/server/src/routes/ui/admin/mod.rs
@@ -31,7 +31,7 @@ use jellybase::{
     federation::Federation,
     CONF,
 };
-use jellyimport::{import, is_importing};
+use jellyimport::{import, is_importing, IMPORT_ERRORS};
 use markup::DynRender;
 use rand::Rng;
 use rocket::{form::Form, get, post, FromForm, State};
@@ -40,14 +40,14 @@ use tokio::sync::Semaphore;
 use user::rocket_uri_macro_r_admin_users;
 
 #[get("/admin/dashboard")]
-pub fn r_admin_dashboard(
+pub async fn r_admin_dashboard(
     _session: AdminSession,
     database: &State<DataAcid>,
 ) -> MyResult<DynLayoutPage<'static>> {
-    admin_dashboard(database, None)
+    admin_dashboard(database, None).await
 }
 
-pub fn admin_dashboard<'a>(
+pub async fn admin_dashboard<'a>(
     database: &DataAcid,
     flash: Option<MyResult<String>>,
 ) -> MyResult<DynLayoutPage<'a>> {
@@ -66,12 +66,22 @@ pub fn admin_dashboard<'a>(
     };
     let flash = flash.map(|f| f.map_err(|e| format!("{e:?}")));
 
+    let last_import_err = IMPORT_ERRORS.read().await.to_owned();
+
     let database = database.to_owned();
     Ok(LayoutPage {
         title: "Admin Dashboard".to_string(),
         content: markup::new! {
             h1 { "Admin Panel" }
             @FlashDisplay { flash: flash.clone() }
+            @if !last_import_err.is_empty() {
+                section.message.error {
+                    p.error {"The last import resulted in at least one error:"}
+                    ol { @for e in &last_import_err {
+                        li.error { pre.error { @e } }
+                    }}
+                }
+            }
             ul {
                 li{a[href=uri!(r_admin_log(true))] { "Server Log (Warnings only)" }}
                 li{a[href=uri!(r_admin_log(false))] { "Server Log (Full) " }}
@@ -119,14 +129,14 @@ pub fn admin_dashboard<'a>(
 }
 
 #[post("/admin/generate_invite")]
-pub fn r_admin_invite(
+pub async fn r_admin_invite(
     _session: AdminSession,
     database: &State<DataAcid>,
 ) -> MyResult<DynLayoutPage<'static>> {
     let i = format!("{}", rand::thread_rng().gen::<u128>());
     T_INVITE.insert(&database, &*i, ())?;
 
-    admin_dashboard(database, Some(Ok(format!("Invite: {}", i))))
+    admin_dashboard(database, Some(Ok(format!("Invite: {}", i)))).await
 }
 
 #[derive(FromForm)]
@@ -135,7 +145,7 @@ pub struct DeleteInvite {
 }
 
 #[post("/admin/remove_invite", data = "<form>")]
-pub fn r_admin_remove_invite(
+pub async fn r_admin_remove_invite(
     session: AdminSession,
     database: &State<DataAcid>,
     form: Form<DeleteInvite>,
@@ -145,7 +155,7 @@ pub fn r_admin_remove_invite(
         .remove(&database, form.invite.as_str())?
         .ok_or(anyhow!("invite did not exist"))?;
 
-    admin_dashboard(database, Some(Ok("Invite invalidated".into())))
+    admin_dashboard(database, Some(Ok("Invite invalidated".into()))).await
 }
 
 #[post("/admin/import")]
@@ -164,6 +174,7 @@ pub async fn r_admin_import(
                 .map(|_| format!("Import successful; took {:?}", t.elapsed())),
         ),
     )
+    .await
 }
 
 #[post("/admin/delete_cache")]
@@ -182,6 +193,7 @@ pub async fn r_admin_delete_cache(
                 .map(|_| format!("Cache deleted; took {:?}", t.elapsed())),
         ),
     )
+    .await
 }
 
 static SEM_TRANSCODING: Semaphore = Semaphore::const_new(1);
@@ -227,6 +239,7 @@ pub async fn r_admin_transcode_posters(
             t.elapsed()
         ))),
     )
+    .await
 }
 
 fn db_stats(db: &DataAcid) -> anyhow::Result<DynRender> {
@@ -244,7 +257,7 @@ fn db_stats(db: &DataAcid) -> anyhow::Result<DynRender> {
 
     Ok(markup::new! {
         h3 { "Key-Value-Store Statistics" }
-        table {
+        table.border {
             tbody {
                 tr {
                     th { "table name" }
diff --git a/web/style/layout.css b/web/style/layout.css
index b18b42e..fb3da7c 100644
--- a/web/style/layout.css
+++ b/web/style/layout.css
@@ -74,18 +74,18 @@ section.message {
     border-radius: 8px;
 }
 .error {
-    padding: 1em;
     color: var(--c-error);
     font-family: monospace;
 }
 .warn {
-    padding: 1em;
     color: var(--c-warn);
 }
 .success {
-    padding: 1em;
     color: var(--c-success);
 }
+.message p {
+    padding: 1em;
+}
 
 footer {
     padding: 0.1em;
@@ -158,7 +158,7 @@ summary h3 {
     width: max(10em, 40%);
 }
 
-td,
-th {
+table.border td,
+table.border th {
     border: 1px solid gray;
 }