fix: optimise first and latest event fetch (#1410)

Author: nikhilsinhaparseable

src/storage/azure_blob.rs

@@ -698,6 +698,64 @@ impl ObjectStorage for BlobStore {
         Ok(streams)
     }
 
+    async fn list_hours(
+        &self,
+        stream_name: &str,
+        date: &str,
+    ) -> Result<Vec<String>, ObjectStorageError> {
+        let pre = object_store::path::Path::from(format!("{}/{}/", stream_name, date));
+        let resp = self.client.list_with_delimiter(Some(&pre)).await?;
+
+        let hours = resp
+            .common_prefixes
+            .iter()
+            .filter_map(|path| {
+                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();
+
+        Ok(hours)
+    }
+
+    async fn list_minutes(
+        &self,
+        stream_name: &str,
+        date: &str,
+        hour: &str,
+    ) -> Result<Vec<String>, ObjectStorageError> {
+        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
+            .common_prefixes
+            .iter()
+            .filter_map(|path| {
+                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();
+
+        Ok(minutes)
+    }
+
     async fn list_manifest_files(
         &self,
         stream_name: &str,

src/storage/gcs.rs

@@ -605,6 +605,64 @@ impl ObjectStorage for Gcs {
         Ok(streams)
     }
 
+    async fn list_hours(
+        &self,
+        stream_name: &str,
+        date: &str,
+    ) -> Result<Vec<String>, ObjectStorageError> {
+        let pre = object_store::path::Path::from(format!("{}/{}/", stream_name, date));
+        let resp = self.client.list_with_delimiter(Some(&pre)).await?;
+
+        let hours = resp
+            .common_prefixes
+            .iter()
+            .filter_map(|path| {
+                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();
+
+        Ok(hours)
+    }
+
+    async fn list_minutes(
+        &self,
+        stream_name: &str,
+        date: &str,
+        hour: &str,
+    ) -> Result<Vec<String>, ObjectStorageError> {
+        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
+            .common_prefixes
+            .iter()
+            .filter_map(|path| {
+                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();
+
+        Ok(minutes)
+    }
+
     async fn list_manifest_files(
         &self,
         stream_name: &str,

src/storage/localfs.rs

@@ -422,6 +422,43 @@ impl ObjectStorage for LocalFS {
         Ok(dates.into_iter().flatten().collect())
     }
 
+    async fn list_hours(
+        &self,
+        stream_name: &str,
+        date: &str,
+    ) -> Result<Vec<String>, ObjectStorageError> {
+        let path = self.root.join(stream_name).join(date);
+        let directories = ReadDirStream::new(fs::read_dir(&path).await?);
+        let entries: Vec<DirEntry> = directories.try_collect().await?;
+        let entries = entries.into_iter().map(dir_name);
+        let hours: Vec<_> = FuturesUnordered::from_iter(entries).try_collect().await?;
+        Ok(hours
+            .into_iter()
+            .flatten()
+            .filter(|dir| dir.starts_with("hour="))
+            .collect())
+    }
+
+    async fn list_minutes(
+        &self,
+        stream_name: &str,
+        date: &str,
+        hour: &str,
+    ) -> Result<Vec<String>, ObjectStorageError> {
+        let path = self.root.join(stream_name).join(date).join(hour);
+        // Propagate any read_dir errors instead of swallowing them
+        let directories = ReadDirStream::new(fs::read_dir(&path).await?);
+        let entries: Vec<DirEntry> = directories.try_collect().await?;
+        let entries = entries.into_iter().map(dir_name);
+        let minutes: Vec<_> = FuturesUnordered::from_iter(entries).try_collect().await?;
+        // Filter down to only the "minute=" prefixed directories
+        Ok(minutes
+            .into_iter()
+            .flatten()
+            .filter(|dir| dir.starts_with("minute="))
+            .collect())
+    }
+
     async fn list_manifest_files(
         &self,
         _stream_name: &str,