/*
    This file is part of jellything (https://codeberg.org/metamuffin/jellything)
    which is licensed under the GNU Affero General Public License (version 3); see /COPYING.
    Copyright (C) 2023 metamuffin <metamuffin.org>
*/
use crate::{database::Database, federation::Federation, CONF};
use anyhow::{anyhow, bail, Context, Ok};
use async_recursion::async_recursion;
use futures::{stream::FuturesUnordered, StreamExt, TryFutureExt};
use jellybase::{cache_file, AssetLocationExt};
use jellyclient::Session;
use jellycommon::{AssetLocation, MediaSource, Node, NodePrivate, RemoteImportOptions};
use log::{debug, error, info};
use std::{ffi::OsStr, fs::File, os::unix::prelude::OsStrExt, path::PathBuf, sync::LazyLock};
use tokio::sync::Semaphore;

static IMPORT_SEM: LazyLock<Semaphore> = LazyLock::new(|| Semaphore::new(1));

pub async fn import(db: &Database, fed: &Federation) -> anyhow::Result<()> {
    info!("clearing node tree");
    let permit = IMPORT_SEM.try_acquire()?;
    db.node.clear()?;
    info!("importing...");
    let (_, errs) = import_path(CONF.library_path.clone(), db, fed, None)
        .await
        .context("indexing")?;
    info!("import completed");
    drop(permit);
    if errs == 0 {
        Ok(())
    } else {
        Err(anyhow!(
            "partial import, {errs} errors occured; see server log"
        ))
    }
}

#[async_recursion]
pub async fn import_path(
    path: PathBuf,
    db: &Database,
    fed: &Federation,
    parent: Option<String>,
) -> anyhow::Result<(Vec<String>, usize)> {
    if path.is_dir() {
        let mpath = path.join("directory.json");
        let children_paths = path.read_dir()?.map(Result::unwrap).filter_map(|e| {
            if e.path().extension() == Some(&OsStr::from_bytes(b"jelly"))
                || e.metadata().unwrap().is_dir()
            {
                Some(e.path())
            } else {
                None
            }
        });
        let identifier = if mpath.exists() {
            path.file_name().unwrap().to_str().unwrap().to_string()
        } else {
            parent
                .clone()
                .ok_or(anyhow!("non-root node requires parent"))?
        };

        let mut all: FuturesUnordered<_> = children_paths
            .into_iter()
            .map(|p| import_path(p.clone(), db, fed, Some(identifier.clone())).map_err(|e| (p, e)))
            .collect();

        let mut children_ids = Vec::new();
        let mut errs = 0;
        while let Some(k) = all.next().await {
            match k {
                core::result::Result::Ok((els, errs2)) => {
                    errs += errs2;
                    children_ids.extend(els)
                }
                Err((p, e)) => {
                    errs += 1;
                    error!("import of {p:?} failed: {e:?}")
                }
            }
        }
        if mpath.exists() {
            let mut data: Node =
                serde_json::from_reader(File::open(mpath).context("metadata missing")?)?;

            data.public.children = children_ids;
            data.public.parent = parent;
            info!("adding {identifier}");
            db.node.insert(&identifier, &data)?;
            Ok((vec![identifier], errs))
        } else {
            Ok((children_ids, errs))
        }
    } else if path.is_file() {
        info!("loading {path:?}");
        let datafile = File::open(path.clone()).context("cant load metadata")?;
        let mut data: Node = serde_json::from_reader(datafile).context("invalid metadata")?;
        let identifier = data.private.id.clone().unwrap_or_else(|| {
            path.file_name()
                .unwrap()
                .to_str()
                .unwrap()
                .strip_suffix(".jelly")
                .unwrap()
                .to_string()
        });

        let idents = if let Some(io) = data.private.import.take() {
            let session = fed
                .get_session(&io.host)
                .await
                .context("creating session")?;

            import_remote(io, db, &session, identifier.clone(), parent)
                .await
                .context("federated import")?
        } else {
            debug!("adding {identifier}");
            data.public.parent = parent;
            db.node.insert(&identifier, &data)?;
            vec![identifier]
        };
        Ok((idents, 0))
    } else {
        bail!("did somebody really put a fifo or socket in the library?!")
    }
}

static SEM_REMOTE_IMPORT: LazyLock<Semaphore> = LazyLock::new(|| Semaphore::new(16));

#[async_recursion]
async fn import_remote(
    mut opts: RemoteImportOptions,
    db: &Database,
    session: &Session,
    identifier: String,
    parent: Option<String>,
) -> anyhow::Result<Vec<String>> {
    let _permit = SEM_REMOTE_IMPORT.acquire().await.unwrap();
    info!("loading federated node {identifier:?}");

    let flatten = opts.flatten;
    opts.flatten = false;

    let node = session
        .node(&opts.id)
        .await
        .context("fetching remote node")?;

    let poster = cache_federation_asset(session, &opts.id, "poster").await?;
    let backdrop = cache_federation_asset(session, &opts.id, "backdrop").await?;

    drop(_permit);

    let child_parent = if flatten {
        parent
    } else {
        let mut node = Node {
            public: node.clone(),
            private: NodePrivate {
                backdrop: Some(backdrop),
                poster: Some(poster),
                import: None,
                id: None,
                source: Some(MediaSource::Remote {
                    host: opts.host.clone(),
                    remote_id: opts.id.clone(),
                }),
            },
        };
        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();

    let mut children_idents = Vec::new();
    while let Some(r) = children.next().await {
        children_idents.extend(r?);
    }
    Ok(if flatten {
        children_idents
    } else {
        vec![identifier]
    })
}

async fn cache_federation_asset(
    session: &Session,
    identifier: &String,
    role: &str,
) -> anyhow::Result<AssetLocation> {
    let poster = cache_file(&["federation-asset", role, identifier]);
    if !poster.path().exists() {
        session
            .node_asset(&identifier, role, File::create(&poster.path())?)
            .await?;
    }
    Ok(poster)
}