//! Resources — the read side of the MCP object surface.
//!
//! Where [`crate::tool`] exposes an app's *actions*, resources expose its live
//! *objects*: things a client reads and (in the push layer) subscribes to. deox
//! exposes editor buffers this way; a content-addressed store could expose its
//! samples.
//!
//! Unlike tools, resources are **dynamic** — a buffer opens and closes at
//! runtime — so [`ResourceRegistry`] is interior-mutable and cloneable:
//! [`register`](ResourceRegistry::register) and [`remove`](ResourceRegistry::remove)
//! take `&self` and any clone observes the change. Each resource carries an
//! async read handler the registry invokes on `resources/read`; the app owns the
//! underlying state (for deox, behind the main-loop channel).
//!
//! The subscription + `notifications/resources/updated` push layer builds on
//! this registry; see the server module.

use std::collections::HashMap;
use std::future::Future;
use std::pin::Pin;
use std::sync::{Arc, RwLock};

use serde::Serialize;

use crate::error::{Error, Result};

type BoxFuture<T> = Pin<Box<dyn Future<Output = T> + Send>>;
type ReadHandler = Arc<dyn Fn(String) -> BoxFuture<Result<ResourceContents>> + Send + Sync>;

/// Metadata for a single resource, as it appears in `resources/list`.
///
/// `uri` is the stable identifier the client passes back to `resources/read`
/// and `resources/subscribe`. Choose a scheme the app owns (deox uses
/// `buffer://<path>`); it is an addressing contract, not a filesystem path.
#[derive(Debug, Clone, Serialize)]
pub struct ResourceDescriptor {
    pub uri: String,
    pub name: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub description: Option<String>,
    #[serde(rename = "mimeType", skip_serializing_if = "Option::is_none")]
    pub mime_type: Option<String>,
}

impl ResourceDescriptor {
    pub fn new(uri: impl Into<String>, name: impl Into<String>) -> Self {
        Self {
            uri: uri.into(),
            name: name.into(),
            description: None,
            mime_type: None,
        }
    }

    #[must_use]
    pub fn with_description(mut self, description: impl Into<String>) -> Self {
        self.description = Some(description.into());
        self
    }

    #[must_use]
    pub fn with_mime_type(mut self, mime_type: impl Into<String>) -> Self {
        self.mime_type = Some(mime_type.into());
        self
    }
}

/// The contents of one resource, as returned inside a `resources/read` result.
///
/// Mirrors MCP's resource-contents shape: a `uri`, an optional `mimeType`, and
/// exactly one of `text` (UTF-8) or `blob` (base64). Use [`text`](Self::text)
/// for buffers and other textual resources; [`blob`](Self::blob) for binary.
#[derive(Debug, Clone, Serialize)]
pub struct ResourceContents {
    pub uri: String,
    #[serde(rename = "mimeType", skip_serializing_if = "Option::is_none")]
    pub mime_type: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub text: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub blob: Option<String>,
}

impl ResourceContents {
    /// Textual contents (the common case: a buffer, a note, a config file).
    pub fn text(uri: impl Into<String>, text: impl Into<String>) -> Self {
        Self {
            uri: uri.into(),
            mime_type: None,
            text: Some(text.into()),
            blob: None,
        }
    }

    /// Binary contents, already base64-encoded.
    pub fn blob(uri: impl Into<String>, blob_base64: impl Into<String>) -> Self {
        Self {
            uri: uri.into(),
            mime_type: None,
            text: None,
            blob: Some(blob_base64.into()),
        }
    }

    #[must_use]
    pub fn with_mime_type(mut self, mime_type: impl Into<String>) -> Self {
        self.mime_type = Some(mime_type.into());
        self
    }
}

struct Entry {
    descriptor: ResourceDescriptor,
    handler: ReadHandler,
}

/// Registry of resources a client can list and read.
///
/// Cheaply cloneable and interior-mutable: the map lives behind `Arc<RwLock>`,
/// so the app can [`register`](Self::register) and [`remove`](Self::remove)
/// resources at runtime while the server holds its own clone. This is the
/// deliberate contrast with [`crate::tool::ToolRegistry`], whose surface is
/// fixed before serving.
///
/// With the `server` feature, [`notify_updated`](Self::notify_updated) pushes a
/// change to subscribed clients over the SSE stream.
#[derive(Clone)]
pub struct ResourceRegistry {
    entries: Arc<RwLock<HashMap<String, Entry>>>,
    /// Fan-out of updated URIs to every connected SSE session, which filters by
    /// its own subscription set. Capacity bounds how far a slow session may lag
    /// before it drops updates (it then re-reads current state on next use).
    #[cfg(feature = "server")]
    updates: tokio::sync::broadcast::Sender<String>,
}

impl Default for ResourceRegistry {
    fn default() -> Self {
        Self::new()
    }
}

impl ResourceRegistry {
    pub fn new() -> Self {
        Self {
            entries: Arc::new(RwLock::new(HashMap::new())),
            #[cfg(feature = "server")]
            updates: tokio::sync::broadcast::channel(256).0,
        }
    }

    /// Register (or replace) a resource. `handler` is invoked on each
    /// `resources/read` for this uri and returns the resource's current
    /// contents — it should fetch live state, not a snapshot taken at
    /// registration. Registering an existing uri replaces its descriptor and
    /// handler.
    pub fn register<F, Fut>(&self, descriptor: ResourceDescriptor, handler: F)
    where
        F: Fn(String) -> Fut + Send + Sync + 'static,
        Fut: Future<Output = Result<ResourceContents>> + Send + 'static,
    {
        let handler: ReadHandler = Arc::new(move |uri| Box::pin(handler(uri)));
        let uri = descriptor.uri.clone();
        self.entries.write().unwrap().insert(
            uri,
            Entry {
                descriptor,
                handler,
            },
        );
    }

    /// Remove a resource. Returns whether it was present.
    pub fn remove(&self, uri: &str) -> bool {
        self.entries.write().unwrap().remove(uri).is_some()
    }

    pub fn contains(&self, uri: &str) -> bool {
        self.entries.read().unwrap().contains_key(uri)
    }

    /// Descriptors for every registered resource, for `resources/list`.
    pub fn list(&self) -> Vec<ResourceDescriptor> {
        self.entries
            .read()
            .unwrap()
            .values()
            .map(|e| e.descriptor.clone())
            .collect()
    }

    /// Read one resource's current contents by uri. The read handler runs
    /// without the registry lock held, so a handler may itself touch the
    /// registry.
    pub async fn read(&self, uri: &str) -> Result<ResourceContents> {
        let handler = {
            let map = self.entries.read().unwrap();
            map.get(uri).map(|e| e.handler.clone())
        };
        match handler {
            Some(handler) => handler(uri.to_string()).await,
            None => Err(Error::ResourceNotFound(uri.to_string())),
        }
    }
}

#[cfg(feature = "server")]
impl ResourceRegistry {
    /// Announce that a resource changed. Subscribed clients receive a
    /// `notifications/resources/updated` for `uri` over their SSE stream; they
    /// then re-`read` it to get the new contents (the notification carries the
    /// uri, not the body). The app calls this after mutating the underlying
    /// state — for deox, on a buffer or selection change. A no-op when no
    /// session is subscribed.
    pub fn notify_updated(&self, uri: impl Into<String>) {
        // `send` errors only when there are zero receivers; that is the normal
        // "nobody is streaming" case, so the error is deliberately ignored.
        let _ = self.updates.send(uri.into());
    }

    /// A receiver for the update fan-out, used by each SSE session.
    pub(crate) fn subscribe_updates(&self) -> tokio::sync::broadcast::Receiver<String> {
        self.updates.subscribe()
    }
}

/// `resources/read` param object: `{ "uri": "..." }`.
#[cfg(feature = "server")]
#[derive(Debug, serde::Deserialize)]
pub(crate) struct ReadResourceParams {
    pub uri: String,
}

/// Wrap a single [`ResourceContents`] into the `resources/read` result shape
/// (`{ "contents": [ ... ] }`).
#[cfg(feature = "server")]
// contents is consumed into the returned JSON value.
#[allow(clippy::needless_pass_by_value)]
pub(crate) fn read_result(contents: ResourceContents) -> serde_json::Value {
    serde_json::json!({ "contents": [contents] })
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;

    #[tokio::test]
    async fn register_list_read_roundtrip() {
        let reg = ResourceRegistry::new();
        reg.register(
            ResourceDescriptor::new("buffer://main.rs", "main.rs").with_mime_type("text/x-rust"),
            |uri| async move { Ok(ResourceContents::text(uri, "fn main() {}")) },
        );

        let list = reg.list();
        assert_eq!(list.len(), 1);
        assert_eq!(list[0].uri, "buffer://main.rs");

        let contents = reg.read("buffer://main.rs").await.unwrap();
        assert_eq!(contents.text.as_deref(), Some("fn main() {}"));

        // Serialized read result matches the MCP `contents` array shape.
        let v = read_result(contents);
        assert_eq!(v["contents"][0]["uri"], "buffer://main.rs");
        assert_eq!(v["contents"][0]["mimeType"], json!(null)); // not set on this content
        assert_eq!(v["contents"][0]["text"], "fn main() {}");
    }

    #[tokio::test]
    async fn read_unknown_uri_is_not_found() {
        let reg = ResourceRegistry::new();
        let err = reg.read("buffer://gone.rs").await.unwrap_err();
        assert!(matches!(err, Error::ResourceNotFound(_)));
    }

    #[tokio::test]
    async fn remove_is_observed_through_a_clone() {
        let reg = ResourceRegistry::new();
        let clone = reg.clone();
        reg.register(
            ResourceDescriptor::new("buffer://a", "a"),
            |uri| async move { Ok(ResourceContents::text(uri, "x")) },
        );
        assert!(clone.contains("buffer://a"));
        assert!(reg.remove("buffer://a"));
        assert!(!clone.contains("buffer://a"));
        assert!(!reg.remove("buffer://a")); // second remove reports absent
    }

    #[tokio::test]
    async fn register_replaces_existing_uri() {
        let reg = ResourceRegistry::new();
        reg.register(
            ResourceDescriptor::new("buffer://a", "a"),
            |uri| async move { Ok(ResourceContents::text(uri, "old")) },
        );
        reg.register(
            ResourceDescriptor::new("buffer://a", "a"),
            |uri| async move { Ok(ResourceContents::text(uri, "new")) },
        );
        assert_eq!(reg.list().len(), 1);
        assert_eq!(
            reg.read("buffer://a").await.unwrap().text.as_deref(),
            Some("new")
        );
    }
}