path: root/src/api/store.rs

//! /store and /stream handlers.

use std::io::Error as IoError;

use axum::{
    body::{Body, Bytes},
    extract::{Path, Query, State},
    http::{
        header::{ACCEPT, CONTENT_TYPE},
        HeaderMap, HeaderValue, StatusCode,
    },
    response::{IntoResponse, Response},
};
use serde::Deserialize;
use serde_json::Value as JsonValue;
use tokio::sync::broadcast;
use tracing::{debug, error, instrument};

use super::{parse_key_path, response_for_value, store_status};
use crate::state::AppState;
use crate::store::{self, StoreError};
use crate::sync::paths_related;

#[derive(Debug, Default, Deserialize)]
pub(crate) struct ReadQuery {
    #[serde(default)]
    conflicts: bool,
}

#[derive(Debug, Deserialize)]
pub(crate) struct DeleteQuery {
    #[serde(default)]
    prune: bool,
}

fn conflicts_response(
    value: JsonValue,
    conflicts: Vec<store::ConflictReport>,
) -> Result<Response, StatusCode> {
    let body = serde_json::json!({ "value": value, "conflicts": conflicts });
    response_for_value(&body, None) // always JSON
}

#[utoipa::path(
    get,
    path = "/store",
    tag = "Store",
    params(
        ("conflicts" = Option<bool>, Query, description = "Wrap the value and list every conflict resolved on the way"),
    ),
    responses(
        (status = 200, description = "Returns the complete store as JSON or text, depending on Accept"),
        (status = 503, description = "Node is not ready"),
        (status = 500, description = "Failed to read or serialize the document")
    )
)]
pub(crate) async fn get_store_root(
    State(state): State<AppState>,
    headers: HeaderMap,
    Query(query): Query<ReadQuery>,
) -> Result<Response, StatusCode> {
    let accept = headers.get(ACCEPT).cloned();
    let want_conflicts = query.conflicts;

    state
        .render(
            move |doc| {
                if want_conflicts {
                    store::read_path_with_conflicts(doc, &Vec::new())
                        .map(|found| found.map(|(value, conflicts)| (value, Some(conflicts))))
                } else {
                    store::read_path(doc, &Vec::new()).map(|found| found.map(|value| (value, None)))
                }
            },
            move |result| {
                let (value, conflicts) = result
                    .map_err(|e| store_status(e, "Failed to read store path"))?
                    .unwrap_or_else(|| (JsonValue::Object(Default::default()), Some(Vec::new())));
                match conflicts {
                    Some(conflicts) => conflicts_response(value, conflicts),
                    None => response_for_value(&value, accept.as_ref()),
                }
            },
        )
        .await
}

#[utoipa::path(
    get,
    path = "/store/{key}",
    tag = "Store",
    params(
        ("key" = String, Path, description = "Slash-separated store path, for example settings/theme"),
        ("conflicts" = Option<bool>, Query, description = "Wrap the value and list every conflict resolved on the way"),
    ),
    responses(
        (status = 200, description = "Returns the value as JSON or text, depending on Accept"),
        (status = 400, description = "Invalid store path"),
        (status = 404, description = "Value does not exist"),
        (status = 500, description = "Failed to read or serialize the value"),
        (status = 503, description = "Node is not ready")
    )
)]
pub(crate) async fn get_store_entry(
    State(state): State<AppState>,
    Path(key): Path<String>,
    headers: HeaderMap,
    Query(query): Query<ReadQuery>,
) -> Result<Response, StatusCode> {
    if !state.is_ready().await {
        return Err(StatusCode::SERVICE_UNAVAILABLE);
    }

    let path = parse_key_path(&key)?;

    let accept = headers.get(ACCEPT).cloned();
    let want_conflicts = query.conflicts;

    state
        .render(
            move |doc| {
                if want_conflicts {
                    store::read_path_with_conflicts(doc, &path)
                        .map(|found| found.map(|(value, conflicts)| (value, Some(conflicts))))
                } else {
                    store::read_path(doc, &path).map(|found| found.map(|value| (value, None)))
                }
            },
            move |result| {
                let (value, conflicts) = result
                    .map_err(|e| store_status(e, "Failed to read store path"))?
                    .ok_or(StatusCode::NOT_FOUND)?;
                match conflicts {
                    Some(conflicts) => conflicts_response(value, conflicts),
                    None => response_for_value(&value, accept.as_ref()),
                }
            },
        )
        .await
}

#[utoipa::path(
    put,
    path = "/store/{key}",
    tag = "Store",
    params(
        ("key" = String, Path, description = "Slash-separated store path")
    ),
    request_body(
        content = String,
        description = "Value to write. Use application/json for JSON values or text/plain for strings.",
        content_type = "application/json"
    ),
    responses(
        (status = 200, description = "Value written successfully"),
        (status = 400, description = "Invalid path, malformed JSON, invalid UTF-8, or unsupported JSON value"),
        (status = 409, description = "Path conflicts with an existing non-map value"),
        (status = 415, description = "Unsupported Content-Type"),
        (status = 500, description = "Internal error or synchronization notification failure"),
        (status = 503, description = "Node is not ready")
    )
)]
pub(crate) async fn put_store_entry(
    State(state): State<AppState>,
    Path(key): Path<String>,
    headers: HeaderMap,
    body: Bytes,
) -> Result<Response, StatusCode> {
    if !state.is_ready().await {
        return Err(StatusCode::SERVICE_UNAVAILABLE);
    }

    let path = parse_key_path(&key)?;

    let content_type = headers
        .get(CONTENT_TYPE)
        .and_then(|value| value.to_str().ok())
        .unwrap_or("application/json")
        .split(';')
        .next()
        .unwrap_or("")
        .trim();

    let value = match content_type {
        "application/json" => {
            serde_json::from_slice::<JsonValue>(&body).map_err(|_| StatusCode::BAD_REQUEST)?
        }
        "text/plain" => JsonValue::String(
            String::from_utf8(body.to_vec()).map_err(|_| StatusCode::BAD_REQUEST)?,
        ),
        _ => return Err(StatusCode::UNSUPPORTED_MEDIA_TYPE),
    };

    let written = value.clone();
    state
        .mutate_doc(move |doc| store::write_path(doc, &path, &written))
        .await
        .map_err(|e| store_status(e, "Failed to write store value"))?;

    response_for_value(&value, headers.get(ACCEPT))
}

#[utoipa::path(
    delete,
    path = "/store/{key}",
    tag = "Store",
    params(
        ("key" = String, Path, description = "Slash-separated store path"),
        ("prune" = Option<bool>, Query, description = "Remove empty ancestor maps after deletion")
    ),
    responses(
        (status = 204, description = "Value deleted successfully"),
        (status = 400, description = "Invalid store path"),
        (status = 404, description = "Value does not exist"),
        (status = 500, description = "Deletion or synchronization notification failed"),
        (status = 503, description = "Node is not ready")
    )
)]
#[instrument(skip(state), fields(path, prune = query.prune))]
pub(crate) async fn delete_store_entry(
    State(state): State<AppState>,
    Path(key): Path<String>,
    Query(query): Query<DeleteQuery>,
) -> Result<StatusCode, StatusCode> {
    let path = parse_key_path(&key)?;

    tracing::Span::current().record("path", path.join("/").as_str());

    if !state.is_ready().await {
        return Err(StatusCode::SERVICE_UNAVAILABLE);
    }

    if path.is_empty() {
        return Err(StatusCode::BAD_REQUEST);
    }

    let prune = query.prune;
    state
        .mutate_doc(move |doc| store::delete_path(doc, &path, prune))
        .await
        .map_err(|e| store_status(e, "Failed to delete store value"))?;

    Ok(StatusCode::NO_CONTENT)
}

// --- streams -----------------------------------------------------------------

async fn render_traced(
    state: &AppState,
    path: Vec<String>,
) -> Result<(Option<JsonValue>, bool), StoreError> {
    state
        .render(move |doc| store::read_path_traced(doc, &path), |result| result)
        .await
}

pub(crate) async fn stream_store_entry(
    State(state): State<AppState>,
    Path(key): Path<String>,
) -> Result<Response, StatusCode> {
    if !state.is_ready().await {
        return Err(StatusCode::SERVICE_UNAVAILABLE);
    }

    let path = parse_key_path(&key)?;

    // Subscribe first, so nothing between the initial read and the loop is missed.
    let mut changes = state.change_events.subscribe();

    let initial = render_traced(&state, path.clone())
        .await
        .map_err(|e| store_status(e, "Failed to read initial stream value"))?;

    let stream_state = state.clone();
    let stream_key = key.clone();

    let output = async_stream::stream! {
        let (mut last_value, mut conflicted) = initial;

        match stream_event_line(&stream_key, last_value.as_ref()) {
            Ok(line) => yield Ok::<Bytes, IoError>(line),
            Err(error) => {
                yield Err(error);
                return;
            }
        }

        loop {
            let relevant = match changes.recv().await {
                // Conflicted paths can change without a patch on their path; re-read always.
                Ok(set) => conflicted || set.paths.iter().any(|changed| paths_related(changed, &path)),
                Err(broadcast::error::RecvError::Lagged(skipped)) => {
                    debug!(skipped, key = %stream_key, "Stream lagged; re-reading");
                    true
                }
                Err(broadcast::error::RecvError::Closed) => break,
            };

            if !relevant {
                continue;
            }

            let (next_value, next_conflicted) = match render_traced(&stream_state, path.clone()).await {
                Ok(result) => result,
                Err(error) => {
                    error!(error = %format!("{error:#}"), key = %stream_key, "Failed to read updated stream value");
                    yield Err(IoError::other("Failed to read updated store value"));
                    return;
                }
            };
            conflicted = next_conflicted;

            // A related change can still leave the resolved value unchanged.
            if next_value == last_value {
                continue;
            }

            match stream_event_line(&stream_key, next_value.as_ref()) {
                Ok(line) => yield Ok(line),
                Err(error) => {
                    yield Err(error);
                    return;
                }
            }
            last_value = next_value;
        }
    };

    let mut response = Body::from_stream(output).into_response();
    response.headers_mut().insert(
        CONTENT_TYPE,
        HeaderValue::from_static("application/x-ndjson"),
    );
    response.headers_mut().insert(
        axum::http::header::CACHE_CONTROL,
        HeaderValue::from_static("no-cache, no-transform"),
    );
    Ok(response)
}

fn stream_event_line(key: &str, value: Option<&JsonValue>) -> std::io::Result<Bytes> {
    let mut event = serde_json::Map::new();

    event.insert("key".to_owned(), JsonValue::String(key.to_owned()));

    match value {
        Some(value) => {
            event.insert("value".to_owned(), value.clone());
        }
        None => {
            event.insert("deleted".to_owned(), JsonValue::Bool(true));
        }
    }

    let mut line = serde_json::to_vec(&JsonValue::Object(event)).map_err(IoError::other)?;

    line.push(b'\n');
    Ok(Bytes::from(line))
}