rustcities 1.0.4

use std::{
    borrow::Cow, ffi::OsString, fmt::Display, future::Future, iter::once, num::NonZeroUsize,
    ops::ControlFlow, path::PathBuf, sync::Arc,
};

use log::{debug, error, warn};
use smallvec::SmallVec;
use tokio::{
    fs::ReadDir,
    sync::{
        mpsc::{self, error::SendError, unbounded_channel},
        Semaphore,
    },
    task::JoinHandle,
};

/// Translate an ascii character byte from 0123456789abcdefABCDEF -> a number from 0-16
pub const fn ascii_char_translator(c: u8) -> Option<u8> {
    Some(match c {
        b'0' => 0,
        b'1' => 1,
        b'2' => 2,
        b'3' => 3,
        b'4' => 4,
        b'5' => 5,
        b'6' => 6,
        b'7' => 7,
        b'8' => 8,
        b'9' => 9,
        b'a' | b'A' => 10,
        b'b' | b'B' => 11,
        b'c' | b'C' => 12,
        b'd' | b'D' => 13,
        b'e' | b'E' => 14,
        b'f' | b'F' => 15,
        _ => return None,
    })
}

#[derive(thiserror::Error, Debug)]
/// Error understanding the format of the neocities hash returned by the API
pub enum HashFormatErr {
    #[error("\"{0}\" is a non-ascii string, and hence not a valid SHA1 hash")]
    NonAscii(String),
    #[error("\"{0}\" is the wrong length for a SHA1 hash (expected 40 characters)")]
    WrongLength(String),
    #[error("Non-hex character in SHA1 string at {position}")]
    InvalidSHA1HexCharacter { position: usize },
}

/// Get a hash from the given 40 character hex string.
pub fn hash_from_string(v: &str) -> Result<[u8; 20], HashFormatErr> {
    if !v.is_ascii() {
        return Err(HashFormatErr::NonAscii(v.to_owned()));
    };
    let (raw_str_bytes, raw_str) = (v.as_bytes(), v);
    if raw_str_bytes.len() != 2 * 20 {
        return Err(HashFormatErr::WrongLength(raw_str.to_owned()));
    };
    // TODO: Make this core::array::try_from_fn when it stabilises.
    let mut v: [u8; 20] = Default::default();
    for (idx, item) in v.iter_mut().enumerate() {
        *item = (|idx| -> Result<u8, HashFormatErr> {
            // Length is pre-validated so unwrap.
            let lhs = ascii_char_translator(*raw_str_bytes.get(2 * idx).unwrap())
                .ok_or(HashFormatErr::InvalidSHA1HexCharacter { position: 2 * idx })?
                << 4;
            let rhs = ascii_char_translator(*raw_str_bytes.get(2 * idx + 1).unwrap()).ok_or(
                HashFormatErr::InvalidSHA1HexCharacter {
                    position: 2 * idx + 1,
                },
            )?;
            Ok(rhs | lhs)
        })(idx)?;
    }
    Ok(v)
}

#[cfg(test)]
pub mod test_hash_from_str {
    use crate::util::hash_from_string;

    #[test]
    pub fn basic() {
        // Ox3C
        const VALUE: u8 = 60;
        const FIRST_ASCII: u8 = b'3';
        const SECOND_ASCII: u8 = b'C';
        const ASCII_ENCODED_BYTE: [u8; 2] = [FIRST_ASCII, SECOND_ASCII];

        let bytestring: [u8; 40] = core::array::from_fn(|e| ASCII_ENCODED_BYTE[e % 2]);
        let rawstring = core::str::from_utf8(&bytestring).unwrap();

        // The parsed "hash" value.
        let actual_hash = [VALUE; 20];

        assert_eq!(
            actual_hash,
            hash_from_string(rawstring).expect("Should be valid parseable string")
        );
    }
}

/// Normalise a remote neocities path for the purposes of insertion or retrieval in a hashmap
/// (in case of insertion, you need a [`ToOwned::to_owned`] call of course unless you are retreiving.
///
/// This strips all slashes from the start and end. Now, it seems that the remote
/// api does this already, but it might change, so having a normalisation
/// capability is useful, and it also lets us meaningfully compare local and remote
/// paths for e.g. pruning of remote data.
pub fn normalise_remote_neocities_path(s: Cow<'_, str>) -> Cow<'_, str> {
    match s {
        Cow::Borrowed(v) => Cow::Borrowed(v.trim_matches('/')),
        Cow::Owned(v) => {
            let r = v.trim_matches('/');
            if r == v.as_str() {
                Cow::Owned(v)
            } else {
                Cow::Owned(r.to_owned())
            }
        }
    }
}

/// Simple path split into OsString component chunks.
pub type SplitPath = SmallVec<[OsString; 8]>;

#[derive(Clone, Copy, PartialEq, Eq, Debug)]
/// Type of an entry once it has actually been resolved through symlinks. This is sent to avoid
/// repeating a symlink traversal system call to check again in case an entry is a symlink.
pub enum SymlinkResolvedType {
    File,
    Directory,
}

/// Spawn a task that traverses the given local directory recursively and returns directory entries
/// into a channel along with their paths relative to the initial directory as a list of
/// OsString components, excluding the last part, which is embedded
///
/// This will not emit anything for the initial directory itself - because it doesn't have a
/// direntry. The sent list is the directory in which the entry is contained, the actual
/// entry contains it's own name.
///
/// In the case of io errors reading directory entries, you can either continue (with warning)
/// or halt entirely. This is controlled with `ignore_io_errors`
///
/// This puts bounds on how many symlinks to follow, to reduce the risk of loops breaking the
/// program (the current max symlink depth is 256, which should be more than enough to be
/// honest).
///
/// The full path to the entry is retrievable from the entry. This will leave symlinks
/// untraversed (and deliberately so, because it means you can see the actual strings used in
/// the path as-seen by the user). This will traverse those symlinks to obtain metadata though
/// (due to some limitations of the current stable [`std::io::ErrorKind`] enum variants, which
/// does mean an extra IO call.
///
/// This also sends a [`SymlinkResolvedType`] indicating whether the item is a file or directory,
/// after the resolution of any symlinks. This means that if the entry is a symlink, you don't
/// have to resolve it again yourself, when it's already been resolved inside this function.
pub async fn recursive_directory_traverser(
    initial_path: PathBuf,
    ignore_io_errors: bool,
) -> (
    JoinHandle<Result<(), anyhow::Error>>,
    mpsc::UnboundedReceiver<
        Result<(SplitPath, tokio::fs::DirEntry, SymlinkResolvedType), tokio::io::Error>,
    >,
) {
    let (sender, receiver) = mpsc::unbounded_channel();

    const MAX_SYMLINK_DEPTH_INCLUSIVE: usize = 256;

    (
        tokio::spawn(async move {
            let initial_directory = tokio::fs::read_dir(&initial_path).await.map_err(|err| {
                error!("Can't open directory to even upload");
                error!("{err}");
                err
            })?;
            // Stack of the current directories.
            //
            // Whenever a directory entry is itself a directory, it is sent off and then added to
            // the stack for further iteration. When it runs out of entries, it is removed and the
            // next entry down is continued on.
            //
            // We start with an arbitrary small constant for efficiency reasons. Most directory
            // trees are likely to  have minimal depth nya. The first item has no component because
            // the paths are relative to the first directory.
            //
            // The boolean indicates whether or not we were a symlink. This is used to count
            // symlinks and prevent an infinite symlink loop.
            let mut directory_stack = Vec::<(Option<OsString>, ReadDir, bool)>::with_capacity(8);
            directory_stack.push((None, initial_directory, false));
            let mut symlink_cnt: usize = 0;
            // Ease of closure over this, since sending only requires a ref anyhow nya
            let sender_by_ref = &sender;

            // Note here that we do not do let Some(a) = directory_stack.get(directory_stack.len()
            // -1), so that we can modify the directory stack inside the loop.
            'lp: while !directory_stack.is_empty() {
                let mut current_directory =
                    directory_stack.pop().expect("just checked for emptiness");
                let next_entry = current_directory.1.next_entry().await;
                match next_entry {
                    Err(e) => {
                        error!("Error reading directory entry");
                        error!("{e}");
                        sender_by_ref.send(Err(e))?;
                        // Place ourselves back on the stack.
                        directory_stack.push(current_directory);
                    }
                    Ok(Some(directory_entry)) => {
                        // Obtain metadata for future purposes.
                        let metadata = match directory_entry.metadata().await {
                            Ok(v) => v,
                            Err(err) => {
                                warn!("Error reading information about directory entry");
                                warn!("{err}");
                                if ignore_io_errors {
                                    // Add ourselves back to the stack before continuing, to
                                    // skip the entry.
                                    directory_stack.push(current_directory);
                                    continue 'lp;
                                } else {
                                    // Die entirely
                                    error!("Ceasing all directory traversal with failure");
                                    return Err(err.into());
                                }
                            }
                        };
                        let entry_filename = directory_entry.file_name();
                        let full_entry_path = directory_entry.path();

                        let directory_path_stack: SplitPath = directory_stack
                            .iter()
                            .map(|(name, _, _)| name)
                            .cloned()
                            .chain(once(current_directory.0.clone()))
                            .flatten()
                            .collect();

                        // Send off with the current stack entry and path, once the actual
                        // type of the target is resolved through symlinks.
                        let send_current_as = move |resolved_type: SymlinkResolvedType| {
                            sender_by_ref.send(Ok((
                                directory_path_stack,
                                directory_entry,
                                resolved_type,
                            )))
                        };

                        // If directory, then open. If symlink, check the amount of existing
                        // symlinks including the current directory and if it is <= the maximum
                        // then attempt to open nya.
                        //
                        // Increment the total symlinks when opening the inner symlink if it is
                        // a directory, else do not.

                        // Now we attempt
                        if metadata.is_dir() {
                            send_current_as(SymlinkResolvedType::Directory)?;
                            // Directly just dive in if possible.
                            let read_dir = match tokio::fs::read_dir(&full_entry_path).await {
                                Ok(read_dir) => read_dir,
                                Err(err) => {
                                    warn!(
                                        "Could not open {} for reading as directory",
                                        full_entry_path.display()
                                    );
                                    warn!("{err}");
                                    if ignore_io_errors {
                                        warn!("Skipping");
                                        // Add ourselves back
                                        directory_stack.push(current_directory);
                                        continue 'lp;
                                    } else {
                                        error!("Aborting traversal");
                                        return Err(err.into());
                                    }
                                }
                            };

                            // Now we add BOTH back on the stack nya.
                            directory_stack.extend(
                                [current_directory, (Some(entry_filename), read_dir, false)]
                                    .into_iter(),
                            );
                        } else if metadata.is_symlink()
                            && symlink_cnt <= MAX_SYMLINK_DEPTH_INCLUSIVE
                        {
                            // We want to try and open this up more. In the case that it fails
                            // because the symlink does not point to a directory, that's fine,
                            // we just skip it because it's a file instead.
                            //
                            //
                            // Ideally, what we would do is just try to `read_dir` and test if
                            // the error is of the `std::io::ErrorKind::NotADirectory` form.
                            // However - that error kind is NOT CURRENTLY STABILISED.
                            //
                            // This means that to avoid spurious error messages on every symbolic
                            // link, we must manually obtain metadata via a symlink-traversing
                            // function call, and *then* check if it is a directory, and *then*
                            // *unconditionally* forward the error because we know now it is
                            // not spurious.
                            //
                            // The *ideal* code template is commented below nya.
                            /*
                            let read_dir = match tokio::fs::read_dir(full_entry_path).await {
                                Ok(v) => v,
                                // Silently do nothing, this is perfectly ok because it just
                                // means that the symlink isn't  a directory after all.
                                //
                                // Note that currently -
                                Err(e) if e.kind() == std::io::ErrorKind::NotADirectory => {

                                },
                                Err(e) => {

                                }
                            };
                            */
                            let refined_metadata = match tokio::fs::metadata(&full_entry_path).await
                            {
                                Ok(v) => v,
                                Err(e) => {
                                    warn!("Could not obtain symlink metadata about {} to determine traversal properties", &full_entry_path.display());
                                    warn!("{e}");
                                    if ignore_io_errors {
                                        warn!("Skipping");
                                        directory_stack.push(current_directory);
                                        continue 'lp;
                                    } else {
                                        error!(
                                            "Could not obtain information about symlink, dieing"
                                        );
                                        return Err(e.into());
                                    }
                                }
                            };

                            if refined_metadata.is_dir() {
                                send_current_as(SymlinkResolvedType::Directory)?;

                                // Any errors here must be actual errors opening the directory
                                // rather than that the symlink doesn't point to a directory
                                // nya.
                                let read_dir = match tokio::fs::read_dir(&full_entry_path).await {
                                    Ok(v) => v,
                                    Err(e) => {
                                        warn!(
                                            "Could not open symlinked-directory {}",
                                            full_entry_path.display()
                                        );
                                        warn!("{e}");
                                        if ignore_io_errors {
                                            warn!("Skipping");
                                            directory_stack.push(current_directory);
                                            continue 'lp;
                                        } else {
                                            error!("Could not open symlinked directory, dieing!");
                                            return Err(e.into());
                                        }
                                    }
                                };

                                directory_stack.extend(
                                    [current_directory, (Some(entry_filename), read_dir, true)]
                                        .into_iter(),
                                );
                                // We ARE a symlink, so... nya
                                symlink_cnt += 1;
                            } else {
                                send_current_as(SymlinkResolvedType::File)?;
                                debug!(
                                    "Symlink {} is not a directory, not attempting to traverse",
                                    full_entry_path.display()
                                );
                                // Skip so just add ourselves back on.
                                directory_stack.push(current_directory);
                            }
                        } else if metadata.is_symlink() && symlink_cnt > MAX_SYMLINK_DEPTH_INCLUSIVE
                        {
                            warn!("Not traversing symlink at {} because it would overflow the limit of {MAX_SYMLINK_DEPTH_INCLUSIVE} to prevent infinite loops", full_entry_path.display());
                            directory_stack.push(current_directory);
                        } else {
                            // We are a file
                            send_current_as(SymlinkResolvedType::File)?;
                            // Just add ourselves back
                            directory_stack.push(current_directory);
                        }
                    }
                    // In this case, there are no more entries and we should not add ourselves
                    // back. Instead, just log and trace.
                    Ok(None) => {
                        // Symlink, so remove from count.
                        if current_directory.2 {
                            symlink_cnt -= 1;
                        }
                    }
                }
            }

            Ok(())
        }),
        receiver,
    )
}

#[derive(thiserror::Error, Debug)]
/// Either an error sending down a channel, or an error causing something not to be sent at
/// all.
pub enum SendOrBreakError<ChannelItem, BreakError> {
    #[error(transparent)]
    SendErr(SendError<ChannelItem>),
    #[error(transparent)]
    BreakErr(BreakError),
}

/// Function that spawns a collection of tasks with some arbitrary maximum degree of
/// parallelism, emitting raw tasks to a channel for you to process their output.
/// The result is:
/// * A stream of spawned tasks to handle the result of
/// * A processor function that allows skipping or ceasing on a given task and extracting the
///   actual thing to be processed.
/// * A handle to the task actually being spawned, which returns a result with an error in the
///   case of some form of sending error - this should only happen if you close the task
///   receiver stream too early.
///
/// This is NOT implemented in terms of [`restricted_parallel_streaming_async`], because that
/// adds criteria to `O`
pub async fn restricted_parallel_async<
    O: Send + 'static,
    State: Send + Clone + 'static,
    BreakError: Send + 'static,
    IterGenerator: IntoIterator<Item = Item> + Send + 'static,
    Item: Send + 'static,
    ResolvedItem: Send + 'static,
    GeneratedTask: Future<Output = O> + Send + 'static,
>(
    state: State,
    task_generator: IterGenerator,
    mut resolver: impl 'static + Send + FnMut(Item) -> Result<ResolvedItem, ControlFlow<BreakError, ()>>,
    mut task_ctor: impl 'static + Send + FnMut(State, ResolvedItem) -> GeneratedTask,
    maximum_concurrency: NonZeroUsize,
) -> (
    mpsc::UnboundedReceiver<JoinHandle<O>>,
    JoinHandle<Result<(), SendOrBreakError<JoinHandle<O>, BreakError>>>,
)
where
    IterGenerator::Item: Send,
    IterGenerator::IntoIter: Send,
{
    /*
    // This is implemented in terms of [`restric`]
    let (iter_send, mut iter_recv) = unbounded_channel();

    let iterator_extractor = tokio::spawn(async move {
        for item in task_generator {
            iter_send
                .send(item)
                // item needs to implement Debug otherwise.
                .map_err(|e| ())
                .unwrap();
        }
    });

    let (task_recv, inner_spawn) =
        restricted_parallel_streaming_async(state, iter_recv, task_ctor, maximum_concurrency)
            .await;
    let spawner_task = tokio::spawn(async move {
        iterator_extractor
            .await
            // item needs to implement Debug otherwise.
            .map_err(|e| ())
            .unwrap();
        inner_spawn.await.unwrap()
    });
    */

    let semaphore = Arc::new(Semaphore::new(maximum_concurrency.get()));
    let mut iter = task_generator.into_iter();
    let (task_send, task_recv) = unbounded_channel();
    let spawner_task = tokio::spawn(async move {
        while let Ok(permit) = semaphore.clone().acquire_owned().await {
            if let Some(next_gen_item) = iter.next() {
                match resolver(next_gen_item) {
                    Ok(next_resolved_item) => {
                        let state = state.clone();
                        let task = task_ctor(state, next_resolved_item);
                        let running_handle = tokio::spawn(async move {
                            let task_result = task.await;
                            drop(permit);
                            task_result
                        });
                        // Send off
                        task_send
                            .send(running_handle)
                            .map_err(SendOrBreakError::SendErr)?;
                    }
                    Err(ControlFlow::Break(b)) => return Err(SendOrBreakError::BreakErr(b)),
                    Err(ControlFlow::Continue(_)) => continue,
                }
            } else {
                break;
            }
        }
        Ok(())
    });
    (task_recv, spawner_task)
}

/// Just like [`restricted_parallel_async`], except instead of using iterators for input, it
/// uses a receiver stream.
pub async fn restricted_parallel_streaming_async<
    O: Send + 'static,
    BreakError: Send + 'static,
    State: Send + Clone + 'static,
    Item: Send + 'static,
    ResolvedItem: Send + 'static,
    GeneratedTask: Future<Output = O> + Send + 'static,
>(
    state: State,
    mut istream: mpsc::UnboundedReceiver<Item>,
    mut resolver: impl 'static + Send + FnMut(Item) -> Result<ResolvedItem, ControlFlow<BreakError, ()>>,
    mut task_ctor: impl 'static + Send + FnMut(State, ResolvedItem) -> GeneratedTask,
    maximum_concurrency: NonZeroUsize,
) -> (
    mpsc::UnboundedReceiver<JoinHandle<O>>,
    JoinHandle<Result<(), SendOrBreakError<JoinHandle<O>, BreakError>>>,
) {
    let semaphore = Arc::new(Semaphore::new(maximum_concurrency.get()));
    let (task_send, task_recv) = unbounded_channel();

    let spawner_task = tokio::spawn(async move {
        while let Ok(permit) = semaphore.clone().acquire_owned().await {
            if let Some(next_gen_item) = istream.recv().await {
                match resolver(next_gen_item) {
                    Ok(next_gen_item) => {
                        let state = state.clone();
                        let task = task_ctor(state, next_gen_item);
                        let running_handle = tokio::spawn(async move {
                            let task_result = task.await;
                            drop(permit);
                            task_result
                        });
                        // Send off
                        task_send
                            .send(running_handle)
                            .map_err(SendOrBreakError::SendErr)?;
                    }
                    Err(ControlFlow::Break(b)) => return Err(SendOrBreakError::BreakErr(b)),
                    Err(ControlFlow::Continue(_)) => continue,
                }
            } else {
                break;
            }
        }
        Ok(())
    });
    (task_recv, spawner_task)
}

/// In the case of the value being an error, this lets you wrap it in a control flow according to a
/// boolean indicating whether to ignore the error (wrapping it in [`ControlFlow::Continue`]), or
/// whether to break out of some operation with the error.
pub fn wrap_err_as_control_flow<T, E>(v: Result<T, E>, skip: bool) -> Result<T, ControlFlow<E>> {
    v.map_err(|e| {
        if skip {
            ControlFlow::Continue(())
        } else {
            ControlFlow::Break(e)
        }
    })
}

/// Uninhabited error type (like ! when it stabilises)
#[derive(Debug)]
pub enum ImpossibleErr {}
impl Display for ImpossibleErr {
    fn fmt(&self, _f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        unreachable!("Uninhabited type");
    }
}
impl std::error::Error for ImpossibleErr {}

// rustcities
// Copyright (C) 2022  sapient_cogbag <sapient_cogbag at protonmail dot com>
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.