add Prism mode

Author: nikhilsinhaparseable

src/catalog/mod.rs

@@ -343,10 +343,10 @@ pub async fn remove_manifest_from_snapshot(
             Ok(get_first_event(storage.clone(), stream_name, Vec::new()).await?)
         }
         Mode::Query => Ok(get_first_event(storage, stream_name, dates).await?),
-        Mode::Index => Err(ObjectStorageError::UnhandledError(Box::new(
+        Mode::Index | Mode::Prism => Err(ObjectStorageError::UnhandledError(Box::new(
             std::io::Error::new(
                 std::io::ErrorKind::Unsupported,
-                "Can't remove manifest from within Index server",
+                "Can't remove manifest from within Index or Prism server",
             ),
         ))),
     }
@@ -359,7 +359,7 @@ pub async fn get_first_event(
 ) -> Result<Option<String>, ObjectStorageError> {
     let mut first_event_at: String = String::default();
     match PARSEABLE.options.mode {
-        Mode::Index => unimplemented!(),
+        Mode::Index | Mode::Prism => unimplemented!(),
         Mode::All | Mode::Ingest => {
             // get current snapshot
             let stream_first_event = PARSEABLE.get_stream(stream_name)?.get_first_event();

src/handlers/http/middleware.rs

@@ -358,7 +358,7 @@ where
                 })
             }
 
-            Mode::Index => {
+            Mode::Index | Mode::Prism => {
                 let fut = self.service.call(req);
 
                 Box::pin(async move {

src/handlers/http/modal/mod.rs

@@ -230,11 +230,7 @@ impl NodeType {
 
 impl fmt::Display for NodeType {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        match self {
-            NodeType::Ingestor => write!(f, "ingestor"),
-            NodeType::Indexer => write!(f, "indexer"),
-            NodeType::Querier => write!(f, "querier"),
-        }
+        write!(f, "{}", self.as_str())
     }
 }
 

src/main.rs

@@ -43,6 +43,10 @@ async fn main() -> anyhow::Result<()> {
             println!("Indexing is an enterprise feature. Check out https://www.parseable.com/pricing to know more!");
             exit(0)
         }
+        Mode::Prism => {
+            println!("Prism is an enterprise feature. Check out https://www.parseable.com/pricing to know more!");
+            exit(0)
+        }
         Mode::All => Box::new(Server),
     };
 

src/option.rs

@@ -23,6 +23,7 @@ pub enum Mode {
     Query,
     Ingest,
     Index,
+    Prism,
     #[default]
     All,
 }

src/parseable/mod.rs

@@ -282,6 +282,7 @@ impl Parseable {
             Mode::Query => "Distributed (Query)",
             Mode::Ingest => "Distributed (Ingest)",
             Mode::Index => "Distributed (Index)",
+            Mode::Prism => "Distributed (Prism)",
             Mode::All => "Standalone",
         }
     }

src/storage/object_storage.rs

@@ -895,40 +895,36 @@ pub fn to_bytes(any: &(impl ?Sized + serde::Serialize)) -> Bytes {
 }
 
 pub fn schema_path(stream_name: &str) -> RelativePathBuf {
-    match &PARSEABLE.options.mode {
-        Mode::Ingest => {
-            let id = PARSEABLE
-                .ingestor_metadata
-                .as_ref()
-                .expect(INGESTOR_EXPECT)
-                .get_node_id();
-            let file_name = format!(".ingestor.{id}{SCHEMA_FILE_NAME}");
-
-            RelativePathBuf::from_iter([stream_name, STREAM_ROOT_DIRECTORY, &file_name])
-        }
-        Mode::All | Mode::Query | Mode::Index => {
-            RelativePathBuf::from_iter([stream_name, STREAM_ROOT_DIRECTORY, SCHEMA_FILE_NAME])
-        }
+    if PARSEABLE.options.mode == Mode::Ingest {
+        let id = PARSEABLE
+            .ingestor_metadata
+            .as_ref()
+            .expect(INGESTOR_EXPECT)
+            .get_node_id();
+        let file_name = format!(".ingestor.{id}{SCHEMA_FILE_NAME}");
+
+        RelativePathBuf::from_iter([stream_name, STREAM_ROOT_DIRECTORY, &file_name])
+    } else {
+        RelativePathBuf::from_iter([stream_name, STREAM_ROOT_DIRECTORY, SCHEMA_FILE_NAME])
     }
 }
 
 #[inline(always)]
 pub fn stream_json_path(stream_name: &str) -> RelativePathBuf {
-    match &PARSEABLE.options.mode {
-        Mode::Ingest => {
-            let id = PARSEABLE
-                .ingestor_metadata
-                .as_ref()
-                .expect(INGESTOR_EXPECT)
-                .get_node_id();
-            let file_name = format!(".ingestor.{id}{STREAM_METADATA_FILE_NAME}",);
-            RelativePathBuf::from_iter([stream_name, STREAM_ROOT_DIRECTORY, &file_name])
-        }
-        Mode::Query | Mode::All | Mode::Index => RelativePathBuf::from_iter([
+    if PARSEABLE.options.mode == Mode::Ingest {
+        let id = PARSEABLE
+            .ingestor_metadata
+            .as_ref()
+            .expect(INGESTOR_EXPECT)
+            .get_node_id();
+        let file_name = format!(".ingestor.{id}{STREAM_METADATA_FILE_NAME}",);
+        RelativePathBuf::from_iter([stream_name, STREAM_ROOT_DIRECTORY, &file_name])
+    } else {
+        RelativePathBuf::from_iter([
             stream_name,
             STREAM_ROOT_DIRECTORY,
             STREAM_METADATA_FILE_NAME,
-        ]),
+        ])
     }
 }
 

src/storage/store_metadata.rs

@@ -159,7 +159,7 @@ pub async fn resolve_parseable_metadata(
                         metadata.server_mode = PARSEABLE.options.mode;
                         metadata.staging = PARSEABLE.options.staging_dir().to_path_buf();
                       },
-                    Mode::Index => {
+                    Mode::Index | Mode::Prism => {
                         // if index server is started fetch the metadata from remote
                         // update the server mode for local metadata
                         metadata.server_mode = PARSEABLE.options.mode;