fix object store listing

Author: nikhilsinhaparseable

src/storage/azure_blob.rs

@@ -710,10 +710,15 @@ impl ObjectStorage for BlobStore {
             .common_prefixes
             .iter()
             .filter_map(|path| {
-                path.as_ref()
-                    .strip_prefix(&format!("{}/{}/", stream_name, date))
-                    .and_then(|s| s.strip_suffix('/'))
-                    .map(String::from)
+                let path_str = path.as_ref();
+                if let Some(stripped) = path_str.strip_prefix(&format!("{}/{}/", stream_name, date))
+                {
+                    // Remove trailing slash if present, otherwise use as is
+                    let clean_path = stripped.strip_suffix('/').unwrap_or(stripped);
+                    Some(clean_path.to_string())
+                } else {
+                    None
+                }
             })
             .filter(|dir| dir.starts_with("hour="))
             .collect();
@@ -734,10 +739,16 @@ impl ObjectStorage for BlobStore {
             .common_prefixes
             .iter()
             .filter_map(|path| {
-                path.as_ref()
-                    .strip_prefix(&format!("{}/{}/{}/", stream_name, date, hour))
-                    .and_then(|s| s.strip_suffix('/'))
-                    .map(String::from)
+                let path_str = path.as_ref();
+                if let Some(stripped) =
+                    path_str.strip_prefix(&format!("{}/{}/{}/", stream_name, date, hour))
+                {
+                    // Remove trailing slash if present, otherwise use as is
+                    let clean_path = stripped.strip_suffix('/').unwrap_or(stripped);
+                    Some(clean_path.to_string())
+                } else {
+                    None
+                }
             })
             .filter(|dir| dir.starts_with("minute="))
             .collect();

src/storage/gcs.rs

@@ -617,10 +617,15 @@ impl ObjectStorage for Gcs {
             .common_prefixes
             .iter()
             .filter_map(|path| {
-                path.as_ref()
-                    .strip_prefix(&format!("{}/{}/", stream_name, date))
-                    .and_then(|s| s.strip_suffix('/'))
-                    .map(String::from)
+                let path_str = path.as_ref();
+                if let Some(stripped) = path_str.strip_prefix(&format!("{}/{}/", stream_name, date))
+                {
+                    // Remove trailing slash if present, otherwise use as is
+                    let clean_path = stripped.strip_suffix('/').unwrap_or(stripped);
+                    Some(clean_path.to_string())
+                } else {
+                    None
+                }
             })
             .filter(|dir| dir.starts_with("hour="))
             .collect();
@@ -641,10 +646,16 @@ impl ObjectStorage for Gcs {
             .common_prefixes
             .iter()
             .filter_map(|path| {
-                path.as_ref()
-                    .strip_prefix(&format!("{}/{}/{}/", stream_name, date, hour))
-                    .and_then(|s| s.strip_suffix('/'))
-                    .map(String::from)
+                let path_str = path.as_ref();
+                if let Some(stripped) =
+                    path_str.strip_prefix(&format!("{}/{}/{}/", stream_name, date, hour))
+                {
+                    // Remove trailing slash if present, otherwise use as is
+                    let clean_path = stripped.strip_suffix('/').unwrap_or(stripped);
+                    Some(clean_path.to_string())
+                } else {
+                    None
+                }
             })
             .filter(|dir| dir.starts_with("minute="))
             .collect();

src/storage/s3.rs

@@ -789,14 +789,19 @@ impl ObjectStorage for S3 {
         let pre = object_store::path::Path::from(format!("{}/{}/", stream_name, date));
         let resp = self.client.list_with_delimiter(Some(&pre)).await?;
 
-        let hours = resp
+        let hours: Vec<String> = resp
             .common_prefixes
             .iter()
             .filter_map(|path| {
-                path.as_ref()
-                    .strip_prefix(&format!("{}/{}/", stream_name, date))
-                    .and_then(|s| s.strip_suffix('/'))
-                    .map(String::from)
+                let path_str = path.as_ref();
+                if let Some(stripped) = path_str.strip_prefix(&format!("{}/{}/", stream_name, date))
+                {
+                    // Remove trailing slash if present, otherwise use as is
+                    let clean_path = stripped.strip_suffix('/').unwrap_or(stripped);
+                    Some(clean_path.to_string())
+                } else {
+                    None
+                }
             })
             .filter(|dir| dir.starts_with("hour="))
             .collect();
@@ -813,14 +818,20 @@ impl ObjectStorage for S3 {
         let pre = object_store::path::Path::from(format!("{}/{}/{}/", stream_name, date, hour));
         let resp = self.client.list_with_delimiter(Some(&pre)).await?;
 
-        let minutes = resp
+        let minutes: Vec<String> = resp
             .common_prefixes
             .iter()
             .filter_map(|path| {
-                path.as_ref()
-                    .strip_prefix(&format!("{}/{}/{}/", stream_name, date, hour))
-                    .and_then(|s| s.strip_suffix('/'))
-                    .map(String::from)
+                let path_str = path.as_ref();
+                if let Some(stripped) =
+                    path_str.strip_prefix(&format!("{}/{}/{}/", stream_name, date, hour))
+                {
+                    // Remove trailing slash if present, otherwise use as is
+                    let clean_path = stripped.strip_suffix('/').unwrap_or(stripped);
+                    Some(clean_path.to_string())
+                } else {
+                    None
+                }
             })
             .filter(|dir| dir.starts_with("minute="))
             .collect();