use async_trait::async_trait;
use serde::{Deserialize, Serialize};
use std::fmt;
use std::time::Duration;
use vfs::engine::{
    BlockKey, ChunkEdit, ChunkRange, ChunkRef, CreateInodeAttrs, DentryStat, InodeMeta, InodePatch,
    MetadataStore, SnapshotId, VfsError, VfsResult,
};

pub const VFS_METADATA_EXT_NAMESPACE: &str = "secure-exec.vfs.metadata.v1";
const CALLBACK_METADATA_TIMEOUT: Duration = Duration::from_secs(30);

pub trait CallbackMetadataClient: Clone + Send + Sync + 'static {
    type Ownership: Clone + Send + Sync + 'static;
    type Error: fmt::Display;

    fn invoke_metadata_callback(
        &self,
        ownership: Self::Ownership,
        namespace: &str,
        payload: Vec<u8>,
        timeout: Duration,
    ) -> Result<(String, Vec<u8>), Self::Error>;
}

#[derive(Clone)]
pub struct CallbackMetadataStore<C>
where
    C: CallbackMetadataClient,
{
    requests: C,
    ownership: C::Ownership,
    mount_id: String,
}

impl<C> CallbackMetadataStore<C>
where
    C: CallbackMetadataClient,
{
    pub fn new(requests: C, ownership: C::Ownership, mount_id: String) -> Self {
        Self {
            requests,
            ownership,
            mount_id,
        }
    }

    fn invoke(&self, method: MetadataCallbackMethod) -> VfsResult<MetadataCallbackResponse> {
        let request = MetadataCallbackRequest {
            mount_id: self.mount_id.clone(),
            method,
        };
        let payload = serde_json::to_vec(&request).map_err(|error| {
            VfsError::eio(format!(
                "encode vfs metadata callback request for mount '{}': {error}",
                self.mount_id
            ))
        })?;
        let (namespace, payload) = self
            .requests
            .invoke_metadata_callback(
                self.ownership.clone(),
                VFS_METADATA_EXT_NAMESPACE,
                payload,
                CALLBACK_METADATA_TIMEOUT,
            )
            .map_err(Self::sidecar_error_to_vfs)?;
        if namespace != VFS_METADATA_EXT_NAMESPACE {
            return Err(VfsError::eio(format!(
                "unexpected vfs metadata callback namespace '{namespace}'"
            )));
        }
        let response: MetadataCallbackResponse =
            serde_json::from_slice(&payload).map_err(|error| {
                VfsError::eio(format!(
                    "decode vfs metadata callback response for mount '{}': {error}",
                    self.mount_id
                ))
            })?;
        if let MetadataCallbackResponse::Err { code, message } = &response {
            return Err(VfsError::new(code_from_string(code), message.clone()));
        }
        Ok(response)
    }

    fn sidecar_error_to_vfs(error: C::Error) -> VfsError {
        VfsError::eio(error.to_string())
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct MetadataCallbackRequest {
    pub mount_id: String,
    pub method: MetadataCallbackMethod,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "camelCase")]
pub enum MetadataCallbackMethod {
    Resolve {
        path: String,
    },
    ResolveParent {
        path: String,
    },
    Lstat {
        path: String,
    },
    ListDir {
        ino: u64,
    },
    Create {
        parent: u64,
        name: String,
        attrs: CreateInodeAttrs,
    },
    Link {
        parent: u64,
        name: String,
        target: u64,
    },
    Remove {
        parent: u64,
        name: String,
    },
    Rename {
        src_parent: u64,
        src: String,
        dst_parent: u64,
        dst: String,
    },
    SetAttr {
        ino: u64,
        patch: InodePatch,
    },
    CommitWrite {
        ino: u64,
        edits: Vec<ChunkEdit>,
        new_size: u64,
    },
    GetChunks {
        ino: u64,
        range: ChunkRange,
    },
    Snapshot {
        root: u64,
    },
    Fork {
        snap: SnapshotId,
    },
    Gc,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "camelCase")]
pub enum MetadataCallbackResponse {
    InodeMeta { meta: InodeMeta },
    ResolveParent { parent: InodeMeta, name: String },
    DentryStats { entries: Vec<DentryStat> },
    Unit,
    BlockKeys { keys: Vec<BlockKey> },
    ChunkRefs { chunks: Vec<ChunkRef> },
    Snapshot { id: SnapshotId },
    Inode { ino: u64 },
    Err { code: String, message: String },
}

#[async_trait]
impl<C> MetadataStore for CallbackMetadataStore<C>
where
    C: CallbackMetadataClient,
{
    async fn resolve(&self, path: &str) -> VfsResult<InodeMeta> {
        match self.invoke(MetadataCallbackMethod::Resolve {
            path: path.to_owned(),
        })? {
            MetadataCallbackResponse::InodeMeta { meta } => Ok(meta),
            other => Err(unexpected_response("resolve", other)),
        }
    }

    async fn resolve_parent(&self, path: &str) -> VfsResult<(InodeMeta, String)> {
        match self.invoke(MetadataCallbackMethod::ResolveParent {
            path: path.to_owned(),
        })? {
            MetadataCallbackResponse::ResolveParent { parent, name } => Ok((parent, name)),
            other => Err(unexpected_response("resolveParent", other)),
        }
    }

    async fn lstat(&self, path: &str) -> VfsResult<InodeMeta> {
        match self.invoke(MetadataCallbackMethod::Lstat {
            path: path.to_owned(),
        })? {
            MetadataCallbackResponse::InodeMeta { meta } => Ok(meta),
            other => Err(unexpected_response("lstat", other)),
        }
    }

    async fn list_dir(&self, ino: u64) -> VfsResult<Vec<DentryStat>> {
        match self.invoke(MetadataCallbackMethod::ListDir { ino })? {
            MetadataCallbackResponse::DentryStats { entries } => Ok(entries),
            other => Err(unexpected_response("listDir", other)),
        }
    }

    async fn create(
        &self,
        parent: u64,
        name: &str,
        attrs: CreateInodeAttrs,
    ) -> VfsResult<InodeMeta> {
        match self.invoke(MetadataCallbackMethod::Create {
            parent,
            name: name.to_owned(),
            attrs,
        })? {
            MetadataCallbackResponse::InodeMeta { meta } => Ok(meta),
            other => Err(unexpected_response("create", other)),
        }
    }

    async fn link(&self, parent: u64, name: &str, target: u64) -> VfsResult<()> {
        match self.invoke(MetadataCallbackMethod::Link {
            parent,
            name: name.to_owned(),
            target,
        })? {
            MetadataCallbackResponse::Unit => Ok(()),
            other => Err(unexpected_response("link", other)),
        }
    }

    async fn remove(&self, parent: u64, name: &str) -> VfsResult<Vec<BlockKey>> {
        match self.invoke(MetadataCallbackMethod::Remove {
            parent,
            name: name.to_owned(),
        })? {
            MetadataCallbackResponse::BlockKeys { keys } => Ok(keys),
            other => Err(unexpected_response("remove", other)),
        }
    }

    async fn rename(
        &self,
        src_parent: u64,
        src: &str,
        dst_parent: u64,
        dst: &str,
    ) -> VfsResult<Vec<BlockKey>> {
        match self.invoke(MetadataCallbackMethod::Rename {
            src_parent,
            src: src.to_owned(),
            dst_parent,
            dst: dst.to_owned(),
        })? {
            MetadataCallbackResponse::BlockKeys { keys } => Ok(keys),
            other => Err(unexpected_response("rename", other)),
        }
    }

    async fn set_attr(&self, ino: u64, patch: InodePatch) -> VfsResult<Vec<BlockKey>> {
        match self.invoke(MetadataCallbackMethod::SetAttr { ino, patch })? {
            MetadataCallbackResponse::BlockKeys { keys } => Ok(keys),
            other => Err(unexpected_response("setAttr", other)),
        }
    }

    async fn commit_write(
        &self,
        ino: u64,
        edits: Vec<ChunkEdit>,
        new_size: u64,
    ) -> VfsResult<Vec<BlockKey>> {
        match self.invoke(MetadataCallbackMethod::CommitWrite {
            ino,
            edits,
            new_size,
        })? {
            MetadataCallbackResponse::BlockKeys { keys } => Ok(keys),
            other => Err(unexpected_response("commitWrite", other)),
        }
    }

    async fn get_chunks(&self, ino: u64, range: ChunkRange) -> VfsResult<Vec<ChunkRef>> {
        match self.invoke(MetadataCallbackMethod::GetChunks { ino, range })? {
            MetadataCallbackResponse::ChunkRefs { chunks } => Ok(chunks),
            other => Err(unexpected_response("getChunks", other)),
        }
    }

    async fn snapshot(&self, root: u64) -> VfsResult<SnapshotId> {
        match self.invoke(MetadataCallbackMethod::Snapshot { root })? {
            MetadataCallbackResponse::Snapshot { id } => Ok(id),
            other => Err(unexpected_response("snapshot", other)),
        }
    }

    async fn fork(&self, snap: SnapshotId) -> VfsResult<u64> {
        match self.invoke(MetadataCallbackMethod::Fork { snap })? {
            MetadataCallbackResponse::Inode { ino } => Ok(ino),
            other => Err(unexpected_response("fork", other)),
        }
    }

    async fn gc(&self) -> VfsResult<Vec<BlockKey>> {
        match self.invoke(MetadataCallbackMethod::Gc)? {
            MetadataCallbackResponse::BlockKeys { keys } => Ok(keys),
            other => Err(unexpected_response("gc", other)),
        }
    }
}

fn unexpected_response(method: &str, response: MetadataCallbackResponse) -> VfsError {
    VfsError::eio(format!(
        "unexpected vfs metadata callback response for {method}: {response:?}"
    ))
}

fn code_from_string(code: &str) -> &'static str {
    match code {
        "ENOENT" => "ENOENT",
        "EEXIST" => "EEXIST",
        "ENOTDIR" => "ENOTDIR",
        "EISDIR" => "EISDIR",
        "ELOOP" => "ELOOP",
        "ENAMETOOLONG" => "ENAMETOOLONG",
        "ENOTEMPTY" => "ENOTEMPTY",
        "EOPNOTSUPP" => "EOPNOTSUPP",
        "EROFS" => "EROFS",
        "EINVAL" => "EINVAL",
        "EACCES" => "EACCES",
        "EPERM" => "EPERM",
        "ENOSYS" => "ENOSYS",
        _ => "EIO",
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::{Arc, Mutex};
    use vfs::engine::engines::{ChunkedFs, ChunkedFsOptions};
    use vfs::engine::mem::{InMemoryMetadataStore, MemoryBlockStore};
    use vfs::engine::{InodeType, MetadataStore, Storage, Timespec, VirtualFileSystem};

    #[derive(Default)]
    struct RecordingMetadataTransport {
        requests: Mutex<Vec<MetadataCallbackRequest>>,
    }

    impl CallbackMetadataClient for Arc<RecordingMetadataTransport> {
        type Ownership = String;
        type Error = String;

        fn invoke_metadata_callback(
            &self,
            _ownership: Self::Ownership,
            namespace: &str,
            payload: Vec<u8>,
            _timeout: Duration,
        ) -> Result<(String, Vec<u8>), Self::Error> {
            assert_eq!(namespace, VFS_METADATA_EXT_NAMESPACE);
            let callback_request: MetadataCallbackRequest =
                serde_json::from_slice(&payload).expect("decode metadata callback");
            self.requests
                .lock()
                .expect("lock request log")
                .push(callback_request);

            let now = Timespec { sec: 1, nsec: 2 };
            let response = MetadataCallbackResponse::InodeMeta {
                meta: InodeMeta {
                    ino: 7,
                    kind: InodeType::File,
                    mode: 0o644,
                    uid: 1000,
                    gid: 1000,
                    size: 5,
                    nlink: 1,
                    atime: now,
                    mtime: now,
                    ctime: now,
                    birthtime: now,
                    storage: Storage::Inline(b"hello".to_vec()),
                    symlink_target: None,
                },
            };
            let payload = serde_json::to_vec(&response).expect("encode metadata callback response");
            Ok((VFS_METADATA_EXT_NAMESPACE.to_string(), payload))
        }
    }

    #[tokio::test]
    async fn callback_metadata_store_sends_typed_ext_requests() {
        let transport = Arc::new(RecordingMetadataTransport::default());
        let store = CallbackMetadataStore::new(
            transport.clone(),
            "vm-owner".to_string(),
            "mount-a".to_string(),
        );

        let meta = store
            .resolve("/file.txt")
            .await
            .expect("resolve via callback");
        assert_eq!(meta.ino, 7);
        assert_eq!(meta.storage, Storage::Inline(b"hello".to_vec()));

        let requests = transport.requests.lock().expect("lock request log");
        assert_eq!(requests.len(), 1);
        assert_eq!(requests[0].mount_id, "mount-a");
        match &requests[0].method {
            MetadataCallbackMethod::Resolve { path } => assert_eq!(path, "/file.txt"),
            other => panic!("unexpected method: {other:?}"),
        }
    }

    #[derive(Default)]
    struct DelegatingMetadataTransport {
        inner: InMemoryMetadataStore,
        methods: Mutex<Vec<&'static str>>,
    }

    impl CallbackMetadataClient for Arc<DelegatingMetadataTransport> {
        type Ownership = String;
        type Error = String;

        fn invoke_metadata_callback(
            &self,
            _ownership: Self::Ownership,
            namespace: &str,
            payload: Vec<u8>,
            _timeout: Duration,
        ) -> Result<(String, Vec<u8>), Self::Error> {
            assert_eq!(namespace, VFS_METADATA_EXT_NAMESPACE);
            let request: MetadataCallbackRequest =
                serde_json::from_slice(&payload).map_err(|err| err.to_string())?;
            let inner = self.inner.clone();
            let response = std::thread::spawn(move || {
                let runtime = tokio::runtime::Builder::new_current_thread()
                    .enable_all()
                    .build()
                    .map_err(|err| err.to_string())?;
                runtime.block_on(handle_metadata_callback(inner, request.method))
            })
            .join()
            .map_err(|_| "metadata callback thread panicked".to_string())??;
            self.methods
                .lock()
                .expect("lock method log")
                .push(method_name(&response));
            let payload = serde_json::to_vec(&response).map_err(|err| err.to_string())?;
            Ok((VFS_METADATA_EXT_NAMESPACE.to_string(), payload))
        }
    }

    async fn handle_metadata_callback(
        inner: InMemoryMetadataStore,
        method: MetadataCallbackMethod,
    ) -> Result<MetadataCallbackResponse, String> {
        let result = match method {
            MetadataCallbackMethod::Resolve { path } => inner
                .resolve(&path)
                .await
                .map(|meta| MetadataCallbackResponse::InodeMeta { meta }),
            MetadataCallbackMethod::ResolveParent { path } => inner
                .resolve_parent(&path)
                .await
                .map(|(parent, name)| MetadataCallbackResponse::ResolveParent { parent, name }),
            MetadataCallbackMethod::Lstat { path } => inner
                .lstat(&path)
                .await
                .map(|meta| MetadataCallbackResponse::InodeMeta { meta }),
            MetadataCallbackMethod::ListDir { ino } => inner
                .list_dir(ino)
                .await
                .map(|entries| MetadataCallbackResponse::DentryStats { entries }),
            MetadataCallbackMethod::Create {
                parent,
                name,
                attrs,
            } => inner
                .create(parent, &name, attrs)
                .await
                .map(|meta| MetadataCallbackResponse::InodeMeta { meta }),
            MetadataCallbackMethod::Link {
                parent,
                name,
                target,
            } => inner
                .link(parent, &name, target)
                .await
                .map(|()| MetadataCallbackResponse::Unit),
            MetadataCallbackMethod::Remove { parent, name } => inner
                .remove(parent, &name)
                .await
                .map(|keys| MetadataCallbackResponse::BlockKeys { keys }),
            MetadataCallbackMethod::Rename {
                src_parent,
                src,
                dst_parent,
                dst,
            } => inner
                .rename(src_parent, &src, dst_parent, &dst)
                .await
                .map(|keys| MetadataCallbackResponse::BlockKeys { keys }),
            MetadataCallbackMethod::SetAttr { ino, patch } => inner
                .set_attr(ino, patch)
                .await
                .map(|keys| MetadataCallbackResponse::BlockKeys { keys }),
            MetadataCallbackMethod::CommitWrite {
                ino,
                edits,
                new_size,
            } => inner
                .commit_write(ino, edits, new_size)
                .await
                .map(|keys| MetadataCallbackResponse::BlockKeys { keys }),
            MetadataCallbackMethod::GetChunks { ino, range } => inner
                .get_chunks(ino, range)
                .await
                .map(|chunks| MetadataCallbackResponse::ChunkRefs { chunks }),
            MetadataCallbackMethod::Snapshot { root } => inner
                .snapshot(root)
                .await
                .map(|id| MetadataCallbackResponse::Snapshot { id }),
            MetadataCallbackMethod::Fork { snap } => inner
                .fork(snap)
                .await
                .map(|ino| MetadataCallbackResponse::Inode { ino }),
            MetadataCallbackMethod::Gc => inner
                .gc()
                .await
                .map(|keys| MetadataCallbackResponse::BlockKeys { keys }),
        };
        Ok(
            result.unwrap_or_else(|error| MetadataCallbackResponse::Err {
                code: error.code().to_string(),
                message: error.message().to_string(),
            }),
        )
    }

    fn method_name(response: &MetadataCallbackResponse) -> &'static str {
        match response {
            MetadataCallbackResponse::InodeMeta { .. } => "inodeMeta",
            MetadataCallbackResponse::ResolveParent { .. } => "resolveParent",
            MetadataCallbackResponse::DentryStats { .. } => "dentryStats",
            MetadataCallbackResponse::Unit => "unit",
            MetadataCallbackResponse::BlockKeys { .. } => "blockKeys",
            MetadataCallbackResponse::ChunkRefs { .. } => "chunkRefs",
            MetadataCallbackResponse::Snapshot { .. } => "snapshot",
            MetadataCallbackResponse::Inode { .. } => "inode",
            MetadataCallbackResponse::Err { .. } => "err",
        }
    }

    #[tokio::test]
    async fn callback_metadata_store_drives_chunked_filesystem_round_trip() {
        let transport = Arc::new(DelegatingMetadataTransport::default());
        let metadata = CallbackMetadataStore::new(
            transport.clone(),
            "vm-owner".to_string(),
            "mount-a".to_string(),
        );
        let fs = ChunkedFs::with_options(
            metadata,
            MemoryBlockStore::new(),
            ChunkedFsOptions {
                inline_threshold: 1,
                chunk_size: 4,
                ..ChunkedFsOptions::default()
            },
        );

        fs.write_file("/file.txt", b"abcdefgh").await.unwrap();
        fs.pwrite("/file.txt", b"YY", 2).await.unwrap();
        assert_eq!(fs.read_file("/file.txt").await.unwrap(), b"abYYefgh");
        fs.truncate("/file.txt", 5).await.unwrap();
        assert_eq!(fs.read_file("/file.txt").await.unwrap(), b"abYYe");

        let methods = transport.methods.lock().expect("lock method log");
        assert!(methods.contains(&"blockKeys"));
        assert!(methods.contains(&"chunkRefs"));
    }
}
