use std::path::{Path, PathBuf};
use std::sync::Arc;
use mongreldb_core::Database;
use mongreldb_query::ai_retrieval::RemoteAiEndpoint;
use mongreldb_query::distributed::RemoteFragmentEndpoint;
use crate::cluster_runtime::ClusterRuntimeHandle;
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct StorageRuntimeError {
message: String,
}
impl StorageRuntimeError {
pub fn new(message: impl Into<String>) -> Self {
Self {
message: message.into(),
}
}
pub fn cluster_refuses_standalone_bypass() -> Self {
Self::new(
"cluster mode refuses standalone AppState.db data-plane access; \
public data operations are owned by consensus/tablet state",
)
}
}
impl std::fmt::Display for StorageRuntimeError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(&self.message)
}
}
impl std::error::Error for StorageRuntimeError {}
impl From<StorageRuntimeError> for mongreldb_core::MongrelError {
fn from(error: StorageRuntimeError) -> Self {
mongreldb_core::MongrelError::Other(error.message)
}
}
#[derive(Clone)]
pub struct StandaloneRuntime {
db: Arc<Database>,
}
impl StandaloneRuntime {
pub fn new(db: Arc<Database>) -> Self {
Self { db }
}
pub fn db(&self) -> &Arc<Database> {
&self.db
}
pub fn root(&self) -> &Path {
self.db.root()
}
}
#[derive(Clone)]
pub struct ClusterGatewayRuntime {
handle: ClusterRuntimeHandle,
fragment_endpoint: Option<Arc<RemoteFragmentEndpoint>>,
ai_endpoint: Option<Arc<RemoteAiEndpoint>>,
}
impl ClusterGatewayRuntime {
pub fn new(handle: ClusterRuntimeHandle) -> Self {
Self {
handle,
fragment_endpoint: None,
ai_endpoint: None,
}
}
pub fn with_workers(
handle: ClusterRuntimeHandle,
fragment_endpoint: Arc<RemoteFragmentEndpoint>,
ai_endpoint: Arc<RemoteAiEndpoint>,
) -> Self {
Self {
handle,
fragment_endpoint: Some(fragment_endpoint),
ai_endpoint: Some(ai_endpoint),
}
}
pub fn handle(&self) -> &ClusterRuntimeHandle {
&self.handle
}
pub fn node_data(&self) -> &Path {
self.handle.node_data()
}
pub fn fragment_endpoint(&self) -> Option<&Arc<RemoteFragmentEndpoint>> {
self.fragment_endpoint.as_ref()
}
pub fn ai_endpoint(&self) -> Option<&Arc<RemoteAiEndpoint>> {
self.ai_endpoint.as_ref()
}
pub fn set_workers(
&mut self,
fragment_endpoint: Arc<RemoteFragmentEndpoint>,
ai_endpoint: Arc<RemoteAiEndpoint>,
) {
self.fragment_endpoint = Some(fragment_endpoint);
self.ai_endpoint = Some(ai_endpoint);
}
pub fn workers_installed(&self) -> bool {
self.fragment_endpoint.is_some() && self.ai_endpoint.is_some()
}
}
#[derive(Clone)]
pub enum ServerStorageRuntime {
Standalone(StandaloneRuntime),
Cluster(ClusterGatewayRuntime),
}
impl ServerStorageRuntime {
pub fn standalone(db: Arc<Database>) -> Self {
Self::Standalone(StandaloneRuntime::new(db))
}
pub fn cluster(handle: ClusterRuntimeHandle) -> Self {
Self::Cluster(ClusterGatewayRuntime::new(handle))
}
pub fn cluster_with_workers(
handle: ClusterRuntimeHandle,
fragment_endpoint: Arc<RemoteFragmentEndpoint>,
ai_endpoint: Arc<RemoteAiEndpoint>,
) -> Self {
Self::Cluster(ClusterGatewayRuntime::with_workers(
handle,
fragment_endpoint,
ai_endpoint,
))
}
pub fn is_cluster(&self) -> bool {
matches!(self, Self::Cluster(_))
}
pub fn is_standalone(&self) -> bool {
matches!(self, Self::Standalone(_))
}
pub fn standalone_db(&self) -> Option<&Arc<Database>> {
match self {
Self::Standalone(rt) => Some(rt.db()),
Self::Cluster(_) => None,
}
}
pub fn require_standalone_db(&self) -> Result<&Arc<Database>, StorageRuntimeError> {
self.standalone_db()
.ok_or_else(StorageRuntimeError::cluster_refuses_standalone_bypass)
}
pub fn cluster_handle(&self) -> Option<&ClusterRuntimeHandle> {
match self {
Self::Cluster(rt) => Some(rt.handle()),
Self::Standalone(_) => None,
}
}
pub fn cluster_gateway(&self) -> Option<&ClusterGatewayRuntime> {
match self {
Self::Cluster(rt) => Some(rt),
Self::Standalone(_) => None,
}
}
pub fn cluster_gateway_mut(&mut self) -> Option<&mut ClusterGatewayRuntime> {
match self {
Self::Cluster(rt) => Some(rt),
Self::Standalone(_) => None,
}
}
pub fn durable_path(&self) -> PathBuf {
match self {
Self::Standalone(rt) => rt.root().to_path_buf(),
Self::Cluster(rt) => rt.node_data().to_path_buf(),
}
}
pub fn mode_name(&self) -> &'static str {
match self {
Self::Standalone(_) => "standalone",
Self::Cluster(_) => "cluster",
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use tempfile::tempdir;
#[test]
fn standalone_exposes_db_cluster_refuses() {
let dir = tempdir().unwrap();
let db = Arc::new(Database::create(dir.path()).unwrap());
let standalone = ServerStorageRuntime::standalone(Arc::clone(&db));
assert!(standalone.is_standalone());
assert!(!standalone.is_cluster());
assert!(Arc::ptr_eq(
standalone.require_standalone_db().unwrap(),
&db
));
assert_eq!(standalone.mode_name(), "standalone");
}
#[test]
fn cluster_runtime_enum_exists_and_has_no_standalone_db() {
let dir = tempdir().unwrap();
let _ = dir;
let err = StorageRuntimeError::cluster_refuses_standalone_bypass();
assert!(err.to_string().contains("cluster mode refuses"));
let converted: mongreldb_core::MongrelError = err.into();
assert!(converted.to_string().contains("cluster mode refuses"));
}
}