sneakerweb 1.0.1

use anyhow::{Context, Result, anyhow, bail};
use clap::Args;
use glob::glob;
use smol::{
    fs::{File, metadata, read_dir},
    future::zip,
    io::{AsyncBufReadExt, BufReader},
    stream::StreamExt,
};
use std::{
    collections::{BTreeSet, VecDeque},
    ffi::OsStr,
    path::{Path as FsPath, PathBuf},
};
use ufotofu::{
    channels::new_sssr, prelude::*, producer::compat::reader::reader_to_bulk_producer,
    queues::new_unbounded_elastic,
};
use willow25::{prelude::*, storage::*};

use crate::util::*;

#[derive(Args)]
pub struct PublishArgs {
    /// The path to the directory containing the sneakerwebsite to publish.
    pub src: PathBuf,
    /// The domain to which the site should be published.
    #[arg(short, long)]
    pub domain: Option<String>,
    /// The secret key authorising publishing to the specified domain.
    #[arg(short, long)]
    pub secret: Option<String>,
}

pub async fn publish(args: &PublishArgs) -> Result<()> {
    // Check for the keypair + write cap for the given domain, and if present and valid
    // iterate through the given fs directory, creating new entries for each file.
    let dir = &args
        .src
        .canonicalize()
        .context("could not resolve filesystem path to publish")?;

    let dir_announce = domain_style(
        &dir.file_name()
            .map(OsStr::to_string_lossy)
            .unwrap_or(dir.to_string_lossy()),
    );
    println!("Publishing {}...", dir_announce);

    let (subspace, domain) = get_domain(args.domain.as_deref(), "publish to")?;
    let secret = get_secret(args.secret.as_deref(), &domain)?;

    if secret.corresponding_subspace_id() != subspace {
        bail!("invalid secret for the requested domain, publishing cancelled");
    }

    let namespace = NamespaceId::from_bytes(&SNEAKERWEB_NAMESPACE_ID_BYTES);
    let write_cap = WriteCapability::new_communal(namespace.clone(), subspace.clone());

    let sneakerweb_fs_path = sneakerweb_dir().await.context(
        "could not retrieve the filesystem path to sneakerweb configuration and storage",
    )?;

    let mut store = PersistentStore::new(&sneakerweb_fs_path)
        .await
        .context("could not open sneakerweb storage")?;

    let now = Timestamp::now().context("could not establish a current timestamp")?;

    store
        .create_entry(
            &namespace,
            &subspace,
            &Path::new(),
            now,
            &mut [].into_producer(),
            0,
            &write_cap,
            &secret,
        )
        .await
        .map_err(|_| anyhow!("failed to clean the target domain"))?;
    yay("domain cleaned and ready for publication").await;

    let mut publish_filters = BTreeSet::new();
    let global_nopublish_path = sneakerweb_fs_path.join("nopublish");
    if global_nopublish_path.exists() {
        let global_nopublish = File::open(sneakerweb_fs_path.join("nopublish"))
            .await
            .context("a 'nopublish' file was found in the sneakerweb config directory, but it could not be opened")?;
        update_filters(&mut publish_filters, global_nopublish, dir)
                .await.context("failed to add global publishing filters defined in the 'nopublish' file in the sneakerweb config directory")?
    }

    let (sender, receiver) = new_sssr(new_unbounded_elastic());

    zip(
        gather(dir, sender, &mut publish_filters),
        ingest(
            dir, receiver, &mut store, &namespace, &subspace, &write_cap, &secret,
        ),
    )
    .await;

    store
        .flush()
        .await
        .map_err(|_| anyhow!("failed to flush updates to sneakerweb storage"))
}

/// Explores from directory `root` downwards, respecting discovered `.nopublish` files, and forwards discovered files to the `filepath_consumer`.
async fn gather<C: Consumer<Item = PathBuf, Final = ()>>(
    root: &PathBuf,
    mut filepath_consumer: C,
    filters: &mut BTreeSet<PathBuf>,
) {
    let dir_str = root.to_string_lossy();

    let mut to_read = VecDeque::new();
    to_read.push_back(root.clone());

    while let Some(this_path) = to_read.pop_front() {
        if filters.contains(&this_path) {
            continue;
        }

        // Before trying to read anything, check whether we are being asked to ignore certain files or directories.
        let nopublish = this_path.join(".nopublish");
        if nopublish.exists() {
            let Ok(file) = File::open(&nopublish).await else {
                oh_no(&format!(
                    "found a '.nopublish' file at {}, but could not open it. Skipping this directory for security.",
                    emph_style(&dir_str)
                ))
                .await;
                continue;
            };
            let Ok(()) = update_filters(filters, file, &this_path).await else {
                oh_no(&format!(
                    "failed to update publication filters from '.nopublish' file at {}. Skipping this directory for security.",
                    emph_style(&dir_str)
                ))
                .await;
                continue;
            };
            filters.insert(nopublish);
        }

        let relative_path = this_path
            .strip_prefix(root)
            .expect("dir is a prefix of path")
            .to_path_buf();

        let Some(path_str) = relative_path.to_str() else {
            oh_no(&format!(
                "path {} is not supported (only unicode paths are supported)",
                relative_path.to_string_lossy()
            ))
            .await;
            continue;
        };

        let Ok(metadata) = metadata(&this_path).await else {
            oh_no(&format!(
                "could not read metadata for path {}",
                emph_style(path_str)
            ))
            .await;
            continue;
        };

        if metadata.is_dir() || metadata.is_symlink() {
            let Ok(mut entries) = read_dir(&this_path).await else {
                oh_no(&format!(
                    "could not read from directory {}",
                    emph_style(path_str)
                ))
                .await;
                continue;
            };

            while let Some(possible_entry) = entries.next().await {
                match possible_entry {
                    Ok(entry) => to_read.push_back(entry.path()),
                    Err(err) => oh_no(&format!("{}", err)).await,
                }
            }
        }

        if metadata.is_file() {
            let Ok(()) = filepath_consumer.consume_item(this_path).await else {
                oh_no(&format!(
                    "encountered an error telling sneakerweb storage about {}",
                    emph_style(path_str)
                ))
                .await;
                return;
            };
        }
    }

    if filepath_consumer.consume_final(()).await.is_err() {
        oh_no("encountered an error ending the file discovery process").await;
    }
}

/// Receives a stream of filepaths from the `filepath_producer`, converts these into [`paths`](Path), and writes corresponding entries into the `store`.
async fn ingest<P: Producer<Item = PathBuf>>(
    root: &PathBuf,
    mut filepath_producer: P,
    store: &mut PersistentStore,
    namespace: &NamespaceId,
    subspace: &SubspaceId,
    write_capability: &WriteCapability,
    secret: &SubspaceSecret,
) {
    while let Ok(filepath) = filepath_producer.produce_item().await {
        let relative_path = filepath
            .strip_prefix(root)
            .expect("root is a prefix of path since path was found from root");

        let Some(path_str) = relative_path.to_str() else {
            oh_no(&format!(
                "could not interpret path {} (only unicode paths are supported)",
                emph_style(&relative_path.to_string_lossy())
            ))
            .await;
            continue;
        };

        let Ok(entry_metadata) = metadata(&filepath).await else {
            oh_no(&format!(
                "could not read metadata for file {}",
                emph_style(path_str)
            ))
            .await;
            continue;
        };

        let Ok(file) = File::open(&filepath).await else {
            oh_no(&format!("could not open file {}", emph_style(path_str))).await;
            continue;
        };

        let mut file_producer = reader_to_bulk_producer(file, new_unbounded_elastic());

        let path_slices: Vec<&[u8]> = relative_path
            .iter()
            .map(|component| component.to_str().map(|str| str.as_bytes()))
            .collect::<Option<Vec<&[u8]>>>()
            .expect("already verified unicode path");

        let Ok(store_path) = Path::from_slices(&path_slices) else {
            oh_no(&format!(
                "filesystem path {} is not a valid sneakerweb storage path",
                path_str
            ))
            .await;
            continue;
        };

        let Ok(now) = Timestamp::now() else {
            oh_no("could not establish a current timestamp").await;
            continue;
        };

        // TODO: Some kind of progress indicator, for creating entries from large files?
        match store
            .create_entry(
                namespace,
                subspace,
                &store_path,
                now,
                &mut file_producer,
                entry_metadata.len(),
                write_capability,
                secret,
            )
            .await
        {
            Ok(Some(_)) => {
                yay(&format!("created an entry for {}", emph_style(path_str))).await;
            }
            Ok(None) => {
                hmm(&format!(
                    "entry for {} was overwritten by a more recent entry",
                    emph_style(path_str)
                ))
                .await
            }
            Err(err) => match err {
                // TODO: Should these be considered fatal?
                CreateEntryError::StoreError(_) => {
                    oh_no(&format!(
                        "sneakerweb storage could not accept {}",
                        emph_style(path_str)
                    ))
                    .await;
                }
                CreateEntryError::AuthorisationTokenError => unreachable!(
                    "already checked that the provided secret authorises the provided domain"
                ),
            },
        }
    }
}

/// Reads glob patterns from a file, and adds to the set of `filters` any paths relative to `from` matching those patterns.
async fn update_filters(filters: &mut BTreeSet<PathBuf>, file: File, from: &FsPath) -> Result<()> {
    let mut lines = BufReader::new(file).lines();
    while let Some(line) = lines.try_next().await? {
        let filter = from.join(line);
        for excluded in glob(filter.to_str().context(format!(
            "could not interpret path {} (only unicode paths are supported)",
            filter.to_string_lossy()
        ))?)
        .unwrap()
        .filter(Result::is_ok)
        .flatten()
        {
            filters.insert(excluded);
        }
    }
    Ok(())
}