use core::{future::Future, pin::Pin, task::Poll};
use std::{
    fs::FileType,
    path::{Path, PathBuf},
    sync::{Arc, Mutex},
};

use futures::{FutureExt, Stream, StreamExt};
use iref::IriBuf;
use taganak_core::{
    graphs::{
        bijection::{InvertedBijection, TermBijectionOverlay},
        AndFilter, Matches, MatchesFilter,
    },
    sources::parsers::turtle::TurtleParser,
};
use taganak_orm::{
    re::{BlankNode, BlankNodeInterface, GraphError, GraphView, SourceError, Triple, TripleFilter},
    GraphORM,
};
use tokio::fs::{self, ReadDir};
use tokio_stream::wrappers::ReadDirStream;

use crate::graphs::{
    local::persistent::local_file::{LocalFileSourceGraph, LocalFileSourceGraphConfig},
    GraphConfig,
};

use super::{FidibatsMetaData, FidibatsSourceGraphView, SkolemizeBijection};

type LocalFileBijection =
    TermBijectionOverlay<InvertedBijection<SkolemizeBijection>, LocalFileSourceGraph>;

pub enum FidibatsStream<F, M>
where
    M: Matches + Unpin,
    F: TripleFilter + Unpin,
{
    FullStream(FidibatsFullStream<PathBuf>, AndFilter<F, MatchesFilter<M>>),
    Describe(<<<LocalFileBijection as GraphView>::Matches<M> as GraphView>::Filter<F> as GraphView>::Stream),
    Empty,
}

impl<F, M> FidibatsStream<F, M>
where
    M: Matches + Unpin,
    F: TripleFilter + Unpin,
{
    pub async fn new(view: FidibatsSourceGraphView<F, M>) -> Result<Self, GraphError> {
        let matches = view.matches.get();
        if matches.is_empty() {
            return Ok(Self::Empty);
        }

        let meta = FidibatsMetaData::open(view.config.source_iri.path().as_str()).await?;

        if let Some(subj) = matches.subject {
            let path = view.config.subject_file_path(subj, &meta)?;
            let graph = TermBijectionOverlay::new(
                LocalFileSourceGraphConfig::new(
                    IriBuf::new(format!("file://{}", path.display())).unwrap(),
                    None,
                    None,
                    false,
                )
                .into_graph()?,
                InvertedBijection(SkolemizeBijection {
                    graph_uuid: meta.uuid(),
                }),
            )
            .matches(view.matches, view.limit)
            .await?
            .filter(view.filter)
            .await?
            .stream()
            .await?;
            return Ok(Self::Describe(graph));
        }

        Ok(Self::FullStream(
            FidibatsFullStream::new(PathBuf::from(view.config.source_iri.as_str().to_string()), meta.uuid()),
            AndFilter::new(view.filter, MatchesFilter(view.matches)),
        ))
    }
}

impl<F, M> Stream for FidibatsStream<F, M>
where
    M: Matches + Unpin,
    F: TripleFilter + Unpin,
{
    type Item = Result<Arc<Triple>, GraphError>;

    fn poll_next(
        mut self: Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
    ) -> Poll<Option<Self::Item>> {
        let s = &mut *self;

        loop {
            match s {
                FidibatsStream::FullStream(fidibats_full_stream, and_filter) => {
                    match fidibats_full_stream.poll_next_unpin(cx) {
                        Poll::Ready(Some(Ok(triple))) => {
                            if and_filter.filter(triple.as_triple_ref()) {
                                break Poll::Ready(Some(Ok(triple)));
                            }
                        }
                        Poll::Ready(Some(Err(triple))) => break Poll::Ready(Some(Err(triple))),
                        Poll::Ready(None) => break Poll::Ready(None),
                        Poll::Pending => break Poll::Pending,
                    }
                }
                FidibatsStream::Describe(stream) => match stream.poll_next_unpin(cx) {
                    Poll::Ready(r) => break Poll::Ready(r),
                    Poll::Pending => break Poll::Pending,
                },
                FidibatsStream::Empty => break Poll::Ready(None),
            }
        }
    }
}

pub struct FidibatsFullStream<P> {
    graph_uuid: uuid::Uuid,
    base_dir: P,
    state: FidibatsFullStreamState,
}

impl<P> FidibatsFullStream<P>
where
    P: AsRef<Path>,
{
    pub fn new(path: P, uuid: uuid::Uuid) -> Self {
        FidibatsFullStream {
            graph_uuid: uuid,
            base_dir: path,
            state: FidibatsFullStreamState::Start,
        }
    }
}

enum FidibatsFullStreamState {
    Start,
    ReadingNamedNodes {
        stream: ReadImmediateSubDirs,
        state: ReadPrefixSchemeState,
    },
    ReadingBlankNodes(ReadTerms),
}

enum ReadPrefixSchemeState {
    Start,
    ReadPrefixAuthority {
        stream: ReadImmediateSubDirs,
        state: ReadPrefixAuthorityState,
    },
}

enum ReadPrefixAuthorityState {
    Start,
    ReadTerms(ReadTerms),
}

impl<P> Stream for FidibatsFullStream<P>
where
    P: AsRef<Path> + Unpin,
{
    type Item = Result<Arc<Triple>, GraphError>;

    fn poll_next(
        mut self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<Option<Self::Item>> {
        let s = &mut *self;
        loop {
            match &mut s.state {
                FidibatsFullStreamState::Start => {
                    tracing::trace!("start reading fidibats data");
                    let search_dir: PathBuf = [s.base_dir.as_ref(), "named_nodes".as_ref()]
                        .into_iter()
                        .collect();
                    s.state = FidibatsFullStreamState::ReadingNamedNodes {
                        stream: ReadImmediateSubDirs::new(search_dir),
                        state: ReadPrefixSchemeState::Start,
                    }
                }
                FidibatsFullStreamState::ReadingNamedNodes { state, stream } => match state {
                    ReadPrefixSchemeState::Start => match stream.poll_next_unpin(cx) {
                        Poll::Ready(Some(Ok(entry))) => {
                            let path = entry.path();
                            let scheme = path.file_name().expect(
                                "we are literally extracting the names of things in directories",
                            );
                            tracing::trace!(?scheme, "found iri scheme");
                            *state = ReadPrefixSchemeState::ReadPrefixAuthority {
                                stream: ReadImmediateSubDirs::new(path),
                                state: ReadPrefixAuthorityState::Start,
                            };
                        }
                        Poll::Ready(Some(Err(e))) => {
                            return Poll::Ready(Some(Err(GraphError::Source(SourceError::Io(e)))))
                        }
                        Poll::Ready(None) => {
                            s.state = FidibatsFullStreamState::ReadingBlankNodes(ReadTerms {
                                graph_uuid: s.graph_uuid,
                                stream: ReadImmediateSubDirs::new(
                                    [s.base_dir.as_ref(), "blank_nodes".as_ref()]
                                        .into_iter()
                                        .collect(),
                                ),
                                state: ReadTermState::Start,
                            })
                        }
                        Poll::Pending => return Poll::Pending,
                    },
                    ReadPrefixSchemeState::ReadPrefixAuthority {
                        state: auth_state,
                        stream,
                    } => match auth_state {
                        ReadPrefixAuthorityState::Start => match stream.poll_next_unpin(cx) {
                            Poll::Ready(Some(Ok(entry))) => {
                                let path = entry.path();
                                let prefix_athority = path.file_name().expect(
                                "we are literally extracting the names of things in directories",
                                );
                                tracing::trace!(?prefix_athority, "found prefix authority");
                                *auth_state = ReadPrefixAuthorityState::ReadTerms(ReadTerms {
                                    graph_uuid: s.graph_uuid,
                                    stream: ReadImmediateSubDirs::new(path),
                                    state: ReadTermState::Start,
                                })
                            }
                            Poll::Ready(Some(Err(e))) => {
                                return Poll::Ready(Some(Err(GraphError::Source(SourceError::Io(
                                    e,
                                )))))
                            }
                            Poll::Ready(None) => *state = ReadPrefixSchemeState::Start,
                            Poll::Pending => return Poll::Pending,
                        },
                        ReadPrefixAuthorityState::ReadTerms(terms) => {
                            match terms.poll_next_unpin(cx) {
                                ret @ Poll::Ready(Some(_)) => return ret,
                                Poll::Ready(None) => *auth_state = ReadPrefixAuthorityState::Start,
                                Poll::Pending => return Poll::Pending,
                            }
                        }
                    },
                },
                FidibatsFullStreamState::ReadingBlankNodes(terms) => {
                    match terms.poll_next_unpin(cx) {
                        Poll::Ready(Some(Err(GraphError::Source(SourceError::Io(e)))))
                            if e.kind() == std::io::ErrorKind::NotFound =>
                        {
                            tracing::trace!("blank_nodes directory in top level of fidibats source does not exist, skipping");
                            return Poll::Ready(None);
                        }
                        ret @ Poll::Ready(_) => return ret,
                        Poll::Pending => return Poll::Pending,
                    }
                }
            }
        }
    }
}

struct ReadTerms {
    graph_uuid: uuid::Uuid,
    stream: ReadImmediateSubDirs,
    state: ReadTermState,
}

enum ReadTermState {
    Start,
    ReadHash1 {
        stream: ReadImmediateSubDirs,
        state: ReadHash1State,
    },
}

enum ReadHash1State {
    Start,
    ReadFile {
        stream: ReadDirContents,
        state: ReadFileState,
    },
}

enum ReadFileState {
    Start,
    OpenTermFile(Pin<Box<dyn Send + Future<Output = std::io::Result<fs::File>>>>),
    ReadTermFile(
        TurtleParser<
            async_utf8_decoder::Utf8Decoder<
                tokio_util::compat::Compat<tokio::io::BufReader<tokio::fs::File>>,
            >,
        >,
    ),
}

impl Stream for ReadTerms {
    type Item = Result<Arc<Triple>, GraphError>;

    fn poll_next(
        mut self: Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
    ) -> Poll<Option<Self::Item>> {
        let s = &mut *self;
        loop {
            match &mut s.state {
                ReadTermState::Start => match s.stream.poll_next_unpin(cx) {
                    Poll::Ready(Some(Ok(entry))) => {
                        let path = entry.path();
                        let hash_0 = path.file_name().expect(
                            "we are literally extracting the names of things in directories",
                        );
                        tracing::trace!(?hash_0, "found first hash character");
                        s.state = ReadTermState::ReadHash1 {
                            stream: ReadImmediateSubDirs::new(path),
                            state: ReadHash1State::Start,
                        }
                    }
                    Poll::Ready(Some(Err(e))) => {
                        return Poll::Ready(Some(Err(GraphError::Source(SourceError::Io(e)))))
                    }
                    Poll::Ready(None) => return Poll::Ready(None),
                    Poll::Pending => return Poll::Pending,
                },
                ReadTermState::ReadHash1 { stream, state: st } => match st {
                    ReadHash1State::Start => match stream.poll_next_unpin(cx) {
                        Poll::Ready(Some(Ok(entry))) => {
                            let path = entry.path();
                            let hash_1 = path.file_name().expect(
                                "we are literally extracting the names of things in directories",
                            );
                            tracing::trace!(?hash_1, "found second hash character");
                            *st = ReadHash1State::ReadFile {
                                stream: ReadDirContents::new(path),
                                state: ReadFileState::Start,
                            };
                        }
                        Poll::Ready(Some(Err(e))) => {
                            return Poll::Ready(Some(Err(GraphError::Source(SourceError::Io(e)))))
                        }
                        Poll::Ready(None) => s.state = ReadTermState::Start,
                        Poll::Pending => return Poll::Pending,
                    },
                    ReadHash1State::ReadFile { stream, state } => match state {
                        ReadFileState::Start => match stream.poll_next_unpin(cx) {
                            Poll::Ready(Some(Ok((entry, filetype))))
                                if filetype.is_file()
                                    && entry.path().extension() == Some("ttl".as_ref()) =>
                            {
                                let path = entry.path();
                                let hash_1 = path.file_name().expect(
                                "we are literally extracting the names of things in directories",
                            );
                                tracing::trace!(?hash_1, "found first hash character");
                                *state = ReadFileState::OpenTermFile(Box::pin(fs::File::open(path)))
                            }
                            Poll::Ready(Some(Ok((entry, _)))) => {
                                tracing::trace!(?entry, "ignoring entry as it is not a file");
                            }
                            Poll::Ready(Some(Err(e))) => {
                                return Poll::Ready(Some(Err(GraphError::Source(SourceError::Io(
                                    e,
                                )))))
                            }
                            Poll::Ready(None) => *st = ReadHash1State::Start,
                            Poll::Pending => return Poll::Pending,
                        },
                        ReadFileState::OpenTermFile(fut) => match fut.poll_unpin(cx) {
                            Poll::Ready(Ok(file)) => {
                                *state = ReadFileState::ReadTermFile(TurtleParser::new(
                                    tokio::io::BufReader::new(file),
                                    Arc::new(Mutex::new(Default::default())),
                                ))
                            }
                            Poll::Ready(Err(e)) => {
                                return Poll::Ready(Some(Err(GraphError::Source(SourceError::Io(
                                    e,
                                )))))
                            }
                            Poll::Pending => return Poll::Pending,
                        },
                        ReadFileState::ReadTermFile(parser) => match parser.poll_next_unpin(cx) {
                            Poll::Ready(None) => *state = ReadFileState::Start,
                            Poll::Ready(Some(Ok(triple))) => {
                                return Poll::Ready(Some(Ok(SkolemizeBijection {
                                    graph_uuid: self.graph_uuid,
                                }
                                .inv_biject_triple(triple))))
                            }
                            ret @ Poll::Ready(_) => {
                                return ret.map_err(GraphError::Source);
                            }
                            Poll::Pending => return Poll::Pending,
                        },
                    },
                },
            }
        }
    }
}

struct ReadImmediateSubDirs(ReadDirContents);
impl ReadImmediateSubDirs {
    pub fn new(path: PathBuf) -> Self {
        Self(ReadDirContents::new(path))
    }
}

impl Stream for ReadImmediateSubDirs {
    type Item = std::io::Result<fs::DirEntry>;

    fn poll_next(
        mut self: Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
    ) -> Poll<Option<Self::Item>> {
        let s = &mut *self;
        loop {
            match s.0.poll_next_unpin(cx) {
                Poll::Ready(Some(Ok((entry, ft)))) if ft.is_dir() => {
                    return Poll::Ready(Some(Ok(entry)))
                }
                Poll::Ready(Some(Ok((entry, _)))) => {
                    tracing::trace!(?entry, "ignoring entry as it is not a directory");
                }
                ret => return ret.map(|o| o.map(|r| r.map(|(e, _)| e))),
            }
        }
    }
}

enum ReadDirContents {
    Start(PathBuf),
    CreateReadDir(Pin<Box<dyn Send + Future<Output = tokio::io::Result<ReadDir>>>>),
    ReadDirContents {
        stream: ReadDirStream,
        state: ReadDirEntryState,
    },
    End,
}
enum ReadDirEntryState {
    Start,
    GetFileType {
        entry: SendPtr<fs::DirEntry>,
        fut: Pin<Box<dyn Send + Future<Output = tokio::io::Result<FileType>>>>,
    },
}

impl ReadDirContents {
    fn new(path: PathBuf) -> Self {
        ReadDirContents::Start(path)
    }
}

impl Stream for ReadDirContents {
    type Item = std::io::Result<(fs::DirEntry, FileType)>;

    fn poll_next(
        mut self: Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
    ) -> Poll<Option<Self::Item>> {
        let s = &mut *self;
        loop {
            match s {
                ReadDirContents::Start(path) => {
                    tracing::trace!(
                        ?path,
                        "constructing stream for reading immediate directory contents"
                    );
                    *s = ReadDirContents::CreateReadDir(Box::pin(fs::read_dir(path.to_owned())))
                }
                ReadDirContents::CreateReadDir(ref mut pin) => match pin.poll_unpin(cx) {
                    Poll::Ready(Ok(stream)) => {
                        tracing::trace!("ready to read directory contents");
                        *s = ReadDirContents::ReadDirContents {
                            stream: ReadDirStream::new(stream),
                            state: ReadDirEntryState::Start,
                        }
                    }
                    Poll::Ready(Err(e)) => {
                        tracing::trace!(%e, "encountered error when constructing stream");
                        *s = ReadDirContents::End;
                        return Poll::Ready(Some(Err(e)));
                    }
                    Poll::Pending => return Poll::Pending,
                },
                ReadDirContents::ReadDirContents {
                    ref mut stream,
                    ref mut state,
                } => match state {
                    ReadDirEntryState::Start => match stream.poll_next_unpin(cx) {
                        Poll::Ready(Some(Ok(entry))) => {
                            tracing::trace!(?entry, "found immediate entry, determining filetype");
                            let entry = Box::new(entry);
                            let entry = Box::leak(entry);
                            let filetype = Box::pin(entry.file_type());
                            *state = ReadDirEntryState::GetFileType {
                                fut: filetype,
                                entry: SendPtr::new(entry as *const _ as *mut _),
                            };
                        }
                        Poll::Ready(Some(Err(e))) => {
                            tracing::trace!(%e, "encountered error while reading contents, continuing");
                            return Poll::Ready(Some(Err(e)));
                        }
                        Poll::Ready(None) => {
                            tracing::trace!("reached end of dir-content-stream terminating");
                            *s = ReadDirContents::End
                        }
                        Poll::Pending => return Poll::Pending,
                    },
                    ReadDirEntryState::GetFileType { entry, fut } => {
                        match fut.poll_unpin(cx) {
                            Poll::Ready(res) => {
                                // SECURITY: was created previously and the future it belonged to
                                // will be dropped dropped due to the following state transition thus the
                                // raw pointer will also not be accessible anymore not accessible anymore.
                                //
                                // Since the future will be dropped with the state transition and
                                // before not be invoked, this can be assumed safe as dropping a
                                // &entry does not do anything to the memory the ref points to.
                                let entry = unsafe { Box::from_raw(entry.get()) };
                                *state = ReadDirEntryState::Start;
                                match res {
                                    Ok(filetype) => {
                                        tracing::trace!(
                                            ?entry,
                                            ?filetype,
                                            "determined filetype of entry"
                                        );
                                        return Poll::Ready(Some(Ok((*entry, filetype))));
                                    }
                                    Err(e) => {
                                        tracing::trace!(%e, ?entry, "encountered error while determining filetype for entry");
                                        return Poll::Ready(Some(Err(e)));
                                    }
                                }
                            }
                            Poll::Pending => return Poll::Pending,
                        }
                    }
                },
                ReadDirContents::End => {
                    tracing::trace!("finished reading contents or encountered fatal error");
                    return Poll::Ready(None);
                }
            }
        }
    }
}

#[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
struct SendPtr<T> {
    ptr: *mut T,
}
// Safety: !Send for raw pointers is not for safety, just a lint
unsafe impl<T> Send for SendPtr<T> where T: Send {}
// Safety: !Sync for raw pointers is not for safety, just a lint
unsafe impl<T> Sync for SendPtr<T> where T: Send {}

impl<T> SendPtr<T> {
    /// Create a new [`SendPtr`] from a `*mut T`.
    pub fn new(ptr: *mut T) -> Self {
        Self { ptr }
    }

    /// Get the underlying raw pointer.
    pub fn get(self) -> *mut T {
        self.ptr
    }
}

impl<T> Clone for SendPtr<T> {
    fn clone(&self) -> Self {
        *self
    }
}

impl<T> Copy for SendPtr<T> {}
