vstorage 0.6.0

// Copyright 2025 Hugo Osvaldo Barrera
//
// SPDX-License-Identifier: EUPL-1.2

//! A [`JmapStorage`] implements the Storage trait for JMAP servers.
//!
//! JMAP (JSON Meta Application Protocol) is defined in RFC 8620. This implementation
//! supports contact management via the JMAP Contacts extension (RFC 9553).
//!
//! This module is only available when the `jmap` feature is enabled.

use std::collections::HashMap;
use std::str::FromStr;
use std::sync::Arc;

use async_trait::async_trait;
use calcard::{icalendar::ICalendar, jscalendar::JSCalendar};
use http::{Request, Response};
use hyper::body::Incoming;
use libjmap::{ChangeStatus, ChangesResponse, JmapClient, calendar::Calendar};
use log::{debug, info, trace};
use serde_json::Value;
use tokio::sync::Mutex;
use tower::Service;

use super::cache::StateCache;
use crate::{
    CollectionId, Error, ErrorKind, Etag, Href, ItemKind, Result,
    base::{
        Collection, CollectionChanges, CreateItemOptions, FetchedItem, Item, ItemVersion, Property,
        Storage,
    },
    calendar::CalendarProperty,
    disco::{DiscoveredCollection, Discovery},
    jmap::cache::StateChanges,
};

/// Builder for [`JmapStorage`].
pub struct JmapStorageBuilder<C>
where
    C: Service<Request<String>, Response = Response<Incoming>> + Send + Sync + 'static,
    C::Error: std::error::Error + Send + Sync,
    C::Future: Send + Sync,
{
    client: JmapClient<C>,
}

impl<C> JmapStorageBuilder<C>
where
    C: Service<Request<String>, Response = Response<Incoming>> + Send + Sync + 'static,
    C::Error: std::error::Error + Send + Sync,
    C::Future: Send + Sync,
{
    /// Build the storage instance.
    #[must_use]
    pub fn build(self, item_kind: ItemKind) -> JmapStorage<C> {
        if item_kind == ItemKind::AddressBook {
            todo!("JMAP Contacts is not implemented");
        }
        JmapStorage {
            inner: Arc::new(Mutex::new((self.client, StateCache::new()))),
            item_kind,
        }
    }
}

/// Storage backed by a JMAP server.
///
/// A single storage represents a single JMAP server with a specific set of credentials.
/// This implementation focuses on calendar support using Icalendar format.
///
/// # Caveats
///
/// Write operations are serialized. This is necessary due to JMAP's concurrency model.
/// Operations use `ifInState` to ensure the server is in the expected state. If more than
/// one concurrent requests specify the same state (e.g., both with `ifInState=S1`), only
/// one can succeed, and the others will fail with `stateMismatch`. This means operations
/// MUST be serialized at the protocol level.
///
/// Additionally, JMAP states are global (server-wide) opaque strings. The cache tracks a
/// linear history of state transitions. Concurrent operations would create a branching
/// history that breaks cache invariants.
///
/// In future, we might have a function to apply multiple operations at once. This
/// requires changes to the [`Storage`]  trait, but would negate the negative implications
/// of this caveat.
pub struct JmapStorage<C>
where
    C: Service<Request<String>, Response = Response<Incoming>> + Send + Sync + 'static,
    C::Error: std::error::Error + Send + Sync,
    C::Future: Send + Sync,
{
    inner: Arc<Mutex<(JmapClient<C>, StateCache)>>,
    item_kind: ItemKind,
}

impl<C> JmapStorage<C>
where
    C: Service<Request<String>, Response = Response<Incoming>> + Send + Sync + 'static,
    C::Error: std::error::Error + Send + Sync,
    C::Future: Send + Sync,
{
    /// Create a new builder for this storage type.
    #[must_use]
    pub fn builder(client: JmapClient<C>) -> JmapStorageBuilder<C> {
        JmapStorageBuilder { client }
    }
}

fn href_for_item(collection_id: &str, object_id: &str) -> String {
    format!("{collection_id}/{object_id}")
}

fn parse_item_href(href: &str) -> Result<(&str, &str)> {
    href.split_once('/')
        .ok_or_else(|| Error::new(ErrorKind::InvalidInput, "Invalid href format"))
}

fn jscalendar_to_icalendar(jscalendar: &Value) -> Result<Item> {
    let jscalendar_str = jscalendar.to_string();
    let jscalendar = JSCalendar::parse(&jscalendar_str).map_err(|e| {
        Error::new(
            ErrorKind::InvalidData,
            format!("Failed to parse JSCalendar: {e}"),
        )
    })?;
    let icalendar = jscalendar.into_icalendar().ok_or_else(|| {
        Error::new(
            ErrorKind::InvalidData,
            "Failed to convert JSCalendar to iCalendar",
        )
    })?;
    Ok(Item::from(icalendar.to_string()))
}

fn icalendar_to_jscalendar(icalendar_str: &str) -> Result<Value> {
    let icalendar = ICalendar::parse(icalendar_str).map_err(|e| {
        Error::new(
            ErrorKind::InvalidInput,
            format!("Failed to parse iCalendar: {e:?}"),
        )
    })?;
    let jscalendar = icalendar.into_jscalendar();
    serde_json::to_value(&jscalendar.0).map_err(|e| {
        Error::new(
            ErrorKind::Io,
            format!("Failed to serialize JSCalendar: {e}"),
        )
    })
}

impl<C> JmapStorage<C>
where
    C: Service<Request<String>, Response = Response<Incoming>> + Send + Sync + 'static,
    C::Error: std::error::Error + Send + Sync,
    C::Future: Send + Sync,
{
    /// Check if an item has changed since a given state.
    ///
    /// First check the state cache to see if we can determine the change status without
    /// making a network request. On cache miss, it falls back to querying the server for
    /// changes and populates the cache with all results.
    ///
    /// Returns whether the specified item has changed and the latest known state.
    async fn cached_changed_since(
        client: &JmapClient<C>,
        cache: &mut StateCache,
        object_id: &str,
        old_state: &str,
    ) -> Result<(ChangeStatus, String)> {
        if let Some(status) = cache.query_changes(old_state, object_id) {
            // Cache hit. Return status and current state.
            if let Some(latest_state) = cache.latest_state() {
                info!(
                    "State cache hit for object_id={object_id}, from_state={old_state}, status={status:?}"
                );
                return Ok((status, latest_state.to_string()));
            }
            // else: latest_state should always return Some(_).
        }

        // Cache miss.
        info!("State cache miss for object_id={object_id}, from_state={old_state}");

        let mut current_state = old_state.to_string();
        loop {
            let changes_response = client
                .changes::<Calendar>(&current_state, Some(500))
                .await
                .map_err(|err| Error::new(ErrorKind::Io, err))?;

            let ChangesResponse {
                new_state,
                has_more_changes,
                created,
                updated,
                destroyed,
                ..
            } = changes_response;

            // Build and store StateChanges from the response.
            let mut changes = StateChanges::new();
            changes.created.extend(created);
            changes.updated.extend(updated);
            changes.destroyed.extend(destroyed);
            cache.add_transition(current_state.clone(), new_state.clone(), changes);

            if !has_more_changes {
                let status = cache
                    .query_changes(old_state, object_id)
                    .expect("Cache should have the status after populating from changes");
                return Ok((status, new_state));
            }

            current_state = new_state;
        }
    }
}

#[async_trait]
impl<C> Storage for JmapStorage<C>
where
    C: Service<Request<String>, Response = Response<Incoming>> + Send + Sync + 'static,
    C::Error: std::error::Error + Send + Sync,
    C::Future: Send + Sync,
{
    fn item_kind(&self) -> ItemKind {
        self.item_kind
    }

    async fn check(&self) -> Result<()> {
        // Check connectivity and required capabilities by fetching session resource.
        let (client, _cache) = &*self.inner.lock().await;
        let session = client.get_session_resource().await.map_err(|e| {
            Error::new(
                ErrorKind::Unavailable,
                format!("Failed to fetch JMAP session: {e}"),
            )
        })?;

        // TODO: if item_kind is address_book, check the other capability.
        if !session.supports_calendars() {
            return Err(Error::new(
                ErrorKind::Unsupported,
                "JMAP server does not support calendars",
            ));
        }

        Ok(())
    }

    async fn discover_collections(&self) -> Result<Discovery> {
        let (client, _cache) = &*self.inner.lock().await;
        let calendars = client
            .get_collections::<Calendar>()
            .await
            .map_err(|err| Error::new(ErrorKind::Io, err))?; // TODO: check ErrorKind
        debug!("Discovered {} collections.", calendars.len());

        let collections = calendars
            .into_iter()
            .map(|ab| {
                let id = CollectionId::from_str(&ab.id).map_err(|e| {
                    Error::new(
                        ErrorKind::InvalidInput, // TODO: check ErrorKind
                        format!("Invalid collection id '{}': {}", ab.id, e),
                    )
                })?;
                Ok(DiscoveredCollection::new(ab.id.clone(), id))
            })
            .collect::<Result<Vec<_>>>()?;

        Discovery::try_from(collections).map_err(|e| Error::new(ErrorKind::InvalidData, e))
    }

    async fn create_collection(&self, _href: &str) -> Result<Collection> {
        Err(Error::new(
            ErrorKind::Unsupported,
            "JMAP does not support creating collections with specified hrefs",
        ))
    }

    async fn delete_collection(&self, href: &str) -> Result<()> {
        // First check if the collection is empty.
        let items = self.list_items(href).await?;
        if !items.is_empty() {
            return Err(ErrorKind::CollectionNotEmpty.into()); // check ErrorKind
        }

        // TODO: should use Etag to avoid races.
        // (this is a documented caveat anyway; other storages have the same limitation).
        let (client, _cache) = &*self.inner.lock().await;
        client
            .delete_collection::<Calendar>(href, None)
            .await
            .map_err(|err| Error::new(ErrorKind::Io, err))?; // check ErrorKind

        Ok(())
    }

    async fn list_items(&self, collection_href: &str) -> Result<Vec<ItemVersion>> {
        let (client, _cache) = &*self.inner.lock().await;
        let records = client
            .get_records::<Calendar>(collection_href)
            .await
            .map_err(|err| Error::new(ErrorKind::Io, err))?; // check ErrorKind

        Ok(records
            .into_iter()
            .map(|record| {
                ItemVersion::new(
                    href_for_item(collection_href, &record.id),
                    Etag::from(record.state),
                )
            })
            .collect())
    }

    async fn changed_since(
        &self,
        collection: &str,
        since_state: Option<&str>,
    ) -> Result<CollectionChanges> {
        let (client, cache) = &mut *self.inner.lock().await;

        match since_state {
            None => {
                let records = client
                    .get_records::<Calendar>(collection)
                    .await
                    .map_err(|err| Error::new(ErrorKind::Io, err))?;

                // All records share the same collection state.
                let new_state = records
                    .first()
                    .map(|r| Some(r.state.clone()))
                    .unwrap_or_default();

                Ok(CollectionChanges {
                    new_state,
                    changed: records
                        .into_iter()
                        .map(|r| href_for_item(collection, &r.id))
                        .collect(),
                    deleted: Vec::new(),
                })
            }
            Some(since_state) => {
                let mut current_state = since_state.to_string();
                let mut all_changed = Vec::new();
                let mut all_deleted = Vec::new();

                let new_state = loop {
                    let changes_response = client
                        .changes::<Calendar>(&current_state, None)
                        .await
                        .map_err(|err| Error::new(ErrorKind::Io, err))?;

                    let ChangesResponse {
                        new_state,
                        has_more_changes,
                        created,
                        updated,
                        destroyed,
                        ..
                    } = changes_response;

                    // Record changes in state transition cache.
                    let mut changes = StateChanges::new();
                    changes.created.extend(created.iter().cloned());
                    changes.updated.extend(updated.iter().cloned());
                    changes.destroyed.extend(destroyed.iter().cloned());
                    cache.add_transition(current_state.clone(), new_state.clone(), changes);

                    // Accumulate changes to be returned.
                    all_changed.extend(created);
                    all_changed.extend(updated);
                    all_deleted.extend(destroyed);

                    if !has_more_changes {
                        break Some(new_state);
                    }
                    current_state = new_state;
                };

                Ok(CollectionChanges {
                    new_state,
                    changed: all_changed
                        .into_iter()
                        .map(|id| href_for_item(collection, &id))
                        .collect(),
                    deleted: all_deleted
                        .into_iter()
                        .map(|id| href_for_item(collection, &id))
                        .collect(),
                })
            }
        }
    }

    async fn get_item(&self, href: &str) -> Result<(Item, Etag)> {
        let (collection_id, object_id) = parse_item_href(href)?;

        trace!("Getting JMAP records for collection {collection_id}.");
        let (client, _cache) = &*self.inner.lock().await;
        let records = client
            .get_records::<Calendar>(collection_id)
            .await
            .map_err(|err| Error::new(ErrorKind::Io, err))?; // TODO: check ErrorKind

        let record = records
            .into_iter()
            .find(|r| r.id == object_id)
            .ok_or_else(|| Error::new(ErrorKind::DoesNotExist, "Item not found"))?;

        let item = jscalendar_to_icalendar(&record.data)?;
        Ok((item, Etag::from(record.state)))
    }

    async fn get_many_items(&self, hrefs: &[&str]) -> Result<Vec<FetchedItem>> {
        if hrefs.is_empty() {
            return Ok(Vec::new());
        }

        // Group hrefs by collection.
        let mut collections: HashMap<&str, Vec<&str>> = HashMap::new();
        for href in hrefs {
            let (collection_id, object_id) = parse_item_href(href)?;
            collections
                .entry(collection_id)
                .or_default()
                .push(object_id);
        }

        let mut fetched_items = Vec::new();
        let (client, _cache) = &*self.inner.lock().await;

        // TODO: get_records should be usable without a collection_id.
        //       this function is feasible with a single network call.
        //       (with some improvements to libjmap).
        for (collection_id, object_ids) in collections {
            let records = client
                .get_records::<Calendar>(collection_id)
                .await
                .map_err(|err| Error::new(ErrorKind::Io, err))?; // TODO: check ErrorKind

            for record in records {
                if object_ids.contains(&record.id.as_str()) {
                    let item = jscalendar_to_icalendar(&record.data)?;
                    fetched_items.push(FetchedItem {
                        href: href_for_item(collection_id, &record.id),
                        item,
                        etag: Etag::from(record.state),
                    });
                }
            }
        }

        Ok(fetched_items)
    }

    async fn create_item(
        &self,
        collection: &str,
        item: &Item,
        _opts: CreateItemOptions,
    ) -> Result<ItemVersion> {
        let (client, cache) = &mut *self.inner.lock().await;

        // Get current state before creating (if available in cache).
        let old_state = cache.latest_state().map(String::from);

        let json_jscalendar = icalendar_to_jscalendar(item.as_str())?;
        let record = client
            .create_record::<Calendar>(collection, &json_jscalendar, old_state.as_deref())
            .await
            .map_err(|err| Error::new(ErrorKind::Io, err))?; // TODO: check ErrorKind
        trace!(
            "Created JSCalendar collection={}, id={}.",
            collection, &record.id
        );

        if let Some(old_state) = old_state {
            cache.add_transition(
                old_state,
                record.state.clone(),
                StateChanges::created(record.id.clone()),
            );
        } else {
            // First operation. Initialize cache with this state.
            cache.add_transition(
                record.state.clone(), // Use new state as both old and new // TODO: review
                record.state.clone(),
                StateChanges::created(record.id.clone()),
            );
        }

        Ok(ItemVersion::new(
            href_for_item(collection, &record.id),
            Etag::from(record.state),
        ))
    }

    async fn update_item(&self, href: &str, etag: &Etag, item: &Item) -> Result<Etag> {
        let (collection_id, object_id) = parse_item_href(href)?;
        let json_jscalendar = icalendar_to_jscalendar(item.as_str())?;

        let (client, cache) = &mut *self.inner.lock().await;

        let result = client
            .update_record::<Calendar>(object_id, collection_id, &json_jscalendar, Some(&etag.0))
            .await;

        match result {
            Ok(record) => {
                cache.add_transition(
                    etag.0.clone(),
                    record.state.clone(),
                    StateChanges::updated(object_id.to_string()),
                );

                Ok(Etag::from(record.state))
            }
            Err(err) => {
                // Check if this is a stateMismatch error
                if let libjmap::error::Error::ServerError { ref error_type, .. } = err {
                    if error_type == "stateMismatch" {
                        let (change_status, new_state) =
                            Self::cached_changed_since(client, cache, object_id, &etag.0).await?;

                        match change_status {
                            ChangeStatus::NotChanged => {
                                // Item hasn't changed, retry with the new state.
                                let record = client
                                    .update_record::<Calendar>(
                                        object_id,
                                        collection_id,
                                        &json_jscalendar,
                                        Some(&new_state),
                                    )
                                    .await
                                    .map_err(|err| Error::new(ErrorKind::Io, err))?;

                                // Record the state transition
                                cache.add_transition(
                                    new_state,
                                    record.state.clone(),
                                    StateChanges::updated(object_id.to_string()),
                                );

                                return Ok(Etag::from(record.state));
                            }
                            ChangeStatus::Changed | ChangeStatus::Deleted => {
                                return Err(Error::new(
                                    ErrorKind::InvalidData,
                                    "Etag does not match; item has been modified",
                                ));
                            }
                        }
                    }
                }

                Err(Error::new(ErrorKind::Io, err))
            }
        }
    }

    async fn delete_item(&self, href: &str, etag: &Etag) -> Result<()> {
        let (_collection_id, object_id) = parse_item_href(href)?;

        let (client, cache) = &mut *self.inner.lock().await;

        let result = client
            .delete_record::<Calendar>(object_id, Some(&etag.0))
            .await;

        match result {
            Ok(new_state) => {
                // Record the state transition in cache
                cache.add_transition(
                    etag.0.clone(),
                    new_state,
                    StateChanges::destroyed(object_id.to_string()),
                );

                Ok(())
            }
            Err(err) => {
                if let libjmap::error::Error::ServerError { ref error_type, .. } = err {
                    if error_type == "stateMismatch" {
                        let (change_status, new_state) =
                            Self::cached_changed_since(client, cache, object_id, &etag.0).await?;

                        match change_status {
                            ChangeStatus::NotChanged => {
                                // Item hasn't changed, retry with the new state
                                let final_state = client
                                    .delete_record::<Calendar>(object_id, Some(&new_state))
                                    .await
                                    .map_err(|err| Error::new(ErrorKind::Io, err))?;

                                // Record the state transition
                                cache.add_transition(
                                    new_state,
                                    final_state,
                                    StateChanges::destroyed(object_id.to_string()),
                                );

                                return Ok(());
                            }
                            ChangeStatus::Changed => {
                                return Err(Error::new(
                                    ErrorKind::InvalidData,
                                    "Etag does not match; item has been modified",
                                ));
                            }
                            ChangeStatus::Deleted => {
                                return Ok(()); // Item was already deleted 🤷.
                            }
                        }
                    }
                }

                Err(Error::new(ErrorKind::Io, err))
            }
        }
    }

    async fn get_property(&self, href: &str, property: Property) -> Result<Option<String>> {
        let (client, _cache) = &*self.inner.lock().await;

        match property {
            Property::Calendar(CalendarProperty::DisplayName) => {
                let calendars = client
                    .get_collections::<Calendar>()
                    .await
                    .map_err(|err| Error::new(ErrorKind::Io, err))?;

                Ok(calendars
                    .into_iter()
                    .find(|cal| cal.id == href)
                    .map(|cal| cal.name))
            }
            Property::Calendar(CalendarProperty::Description) => {
                let calendars = client
                    .get_collections::<Calendar>()
                    .await
                    .map_err(|err| Error::new(ErrorKind::Io, err))?;

                Ok(calendars
                    .into_iter()
                    .find(|cal| cal.id == href)
                    .and_then(|cal| cal.description))
            }
            Property::Calendar(CalendarProperty::Colour) => {
                let calendars = client
                    .get_collections::<Calendar>()
                    .await
                    .map_err(|err| Error::new(ErrorKind::Io, err))?;

                Ok(calendars
                    .into_iter()
                    .find(|cal| cal.id == href)
                    .and_then(|cal| cal.color))
            }
            Property::Calendar(CalendarProperty::Order) => {
                let calendars = client
                    .get_collections::<Calendar>()
                    .await
                    .map_err(|err| Error::new(ErrorKind::Io, err))?;

                Ok(calendars
                    .into_iter()
                    .find(|cal| cal.id == href)
                    .map(|cal| cal.sort_order.to_string()))
            }
            Property::AddressBook(_) => Err(Error::new(
                ErrorKind::InvalidInput,
                "AddressBook properties not supported by Calendar storage",
            )),
        }
    }

    async fn set_property(&self, _href: &str, _property: Property, _value: &str) -> Result<()> {
        Err(Error::new(ErrorKind::Unsupported, "Not yet implemented"))
    }

    async fn unset_property(&self, _href: &str, _property: Property) -> Result<()> {
        Err(Error::new(ErrorKind::Unsupported, "Not yet implemented"))
    }

    fn href_for_collection_id(&self, id: &CollectionId) -> Result<Href> {
        Ok(id.to_string())
    }
}