refactor: move set_metadata

Author: Devdutt Shenoi

src/migration/mod.rs

@@ -138,7 +138,7 @@ pub async fn run_migration(config: &Parseable) -> anyhow::Result<()> {
         let Some(metadata) = migration_stream(&stream_name, &*storage).await? else {
             continue;
         };
-        PARSEABLE.set_stream_meta(&stream_name, metadata).await;
+        config.get_or_create_stream(&stream_name).set_metadata(metadata).await;
     }
 
     Ok(())

src/parseable/mod.rs

@@ -48,7 +48,7 @@ use crate::{
         object_storage::parseable_json_path, ObjectStorageError, ObjectStorageProvider,
         ObjectStoreFormat, Owner, Permisssion, StreamType,
     },
-    validator, LOCK_EXPECT,
+    validator,
 };
 
 mod staging;
@@ -175,32 +175,6 @@ impl Parseable {
         }
     }
 
-    /// Stores the provided stream metadata in memory mapping
-    pub async fn set_stream_meta(&self, stream_name: &str, updated_metadata: LogStreamMetadata) {
-        let map = self.streams.read().expect(LOCK_EXPECT);
-
-        match map.get(stream_name) {
-            Some(stream) => {
-                let mut metadata = stream.metadata.write().expect(LOCK_EXPECT);
-                *metadata = updated_metadata;
-            }
-            None => {
-                drop(map);
-                self.streams.write().expect(LOCK_EXPECT).insert(
-                    stream_name.to_owned(),
-                    Stream::new(
-                        self.options.clone(),
-                        stream_name,
-                        updated_metadata,
-                        self.ingestor_metadata
-                            .as_ref()
-                            .map(|meta| meta.get_ingestor_id()),
-                    ),
-                );
-            }
-        }
-    }
-
     // validate the storage, if the proper path for staging directory is provided
     // if the proper data directory is provided, or s3 bucket is provided etc
     pub async fn validate_storage(&self) -> Result<Option<Bytes>, ObjectStorageError> {

src/parseable/streams.rs

@@ -458,6 +458,11 @@ impl Stream {
         Schema::try_merge(vec![schema, current_schema]).unwrap()
     }
 
+    /// Stores the provided stream metadata in memory mapping
+    pub async fn set_metadata(&self, updated_metadata: LogStreamMetadata) {
+        *self.metadata.write().expect(LOCK_EXPECT) = updated_metadata;
+    }
+
     pub fn get_first_event(&self) -> Option<String> {
         self.metadata
             .read()