@@ -44,7 +44,7 @@ use actix_web_prometheus::PrometheusMetrics;
4444use arrow_schema:: Schema ;
4545use async_trait:: async_trait;
4646use bytes:: Bytes ;
47- use chrono:: Local ;
47+ use chrono:: { DateTime , Local , Utc } ;
4848use datafusion:: { datasource:: listing:: ListingTableUrl , execution:: runtime_env:: RuntimeEnvBuilder } ;
4949use once_cell:: sync:: OnceCell ;
5050use relative_path:: RelativePath ;
@@ -217,6 +217,42 @@ pub trait ObjectStorage: Debug + Send + Sync + 'static {
217217 Ok ( ( ) )
218218 }
219219
220+ /// Updates the first event timestamp in the object store for the specified stream.
221+ ///
222+ /// This function retrieves the current object-store format for the given stream,
223+ /// updates the `first_event_at` field with the provided timestamp, and then
224+ /// stores the updated format back in the object store.
225+ ///
226+ /// # Arguments
227+ ///
228+ /// * `stream_name` - The name of the stream to update.
229+ /// * `first_event` - The timestamp of the first event to set.
230+ ///
231+ /// # Returns
232+ ///
233+ /// * `Result<(), ObjectStorageError>` - Returns `Ok(())` if the update is successful,
234+ /// or an `ObjectStorageError` if an error occurs.
235+ ///
236+ /// # Examples
237+ /// ```ignore
238+ /// ```rust
239+ /// let result = object_store.update_first_event_in_stream("my_stream", "2023-01-01T00:00:00Z").await;
240+ /// assert!(result.is_ok());
241+ /// ```
242+ async fn update_first_event_in_stream (
243+ & self ,
244+ stream_name : & str ,
245+ first_event : & str ,
246+ ) -> Result < ( ) , ObjectStorageError > {
247+ let mut format = self . get_object_store_format ( stream_name) . await ?;
248+ format. first_event_at = Some ( first_event. to_string ( ) ) ;
249+ let format_json = to_bytes ( & format) ;
250+ self . put_object ( & stream_json_path ( stream_name) , format_json)
251+ . await ?;
252+
253+ Ok ( ( ) )
254+ }
255+
220256 async fn put_alerts (
221257 & self ,
222258 stream_name : & str ,
@@ -623,6 +659,78 @@ pub trait ObjectStorage: Debug + Send + Sync + 'static {
623659 Ok ( ( ) )
624660 }
625661
662+ async fn get_stream_meta_from_storage (
663+ & self ,
664+ stream_name : & str ,
665+ ) -> Result < Vec < ObjectStoreFormat > , ObjectStorageError > {
666+ let mut stream_metas = vec ! [ ] ;
667+ let stream_meta_bytes = self
668+ . get_objects (
669+ Some ( & RelativePathBuf :: from_iter ( [
670+ stream_name,
671+ STREAM_ROOT_DIRECTORY ,
672+ ] ) ) ,
673+ Box :: new ( |file_name| file_name. ends_with ( "stream.json" ) ) ,
674+ )
675+ . await ;
676+ if let Ok ( stream_meta_bytes) = stream_meta_bytes {
677+ for stream_meta in stream_meta_bytes {
678+ let stream_meta_ob = serde_json:: from_slice :: < ObjectStoreFormat > ( & stream_meta) ?;
679+ stream_metas. push ( stream_meta_ob) ;
680+ }
681+ }
682+
683+ Ok ( stream_metas)
684+ }
685+
686+ /// Retrieves the earliest first-event-at from the storage for the specified stream.
687+ ///
688+ /// This function fetches the object-store format from all the stream.json files for the given stream from the storage,
689+ /// extracts the `first_event_at` timestamps, and returns the earliest `first_event_at`.
690+ ///
691+ /// # Arguments
692+ ///
693+ /// * `stream_name` - The name of the stream for which `first_event_at` is to be retrieved.
694+ ///
695+ /// # Returns
696+ ///
697+ /// * `Result<Option<String>, ObjectStorageError>` - Returns `Ok(Some(String))` with the earliest
698+ /// first event timestamp if found, `Ok(None)` if no timestamps are found, or an `ObjectStorageError`
699+ /// if an error occurs.
700+ ///
701+ /// # Examples
702+ /// ```ignore
703+ /// ```rust
704+ /// let result = get_first_event_from_storage("my_stream").await;
705+ /// match result {
706+ /// Ok(Some(first_event)) => println!("first-event-at: {}", first_event),
707+ /// Ok(None) => println!("first-event-at not found"),
708+ /// Err(err) => println!("Error: {:?}", err),
709+ /// }
710+ /// ```
711+ async fn get_first_event_from_storage (
712+ & self ,
713+ stream_name : & str ,
714+ ) -> Result < Option < String > , ObjectStorageError > {
715+ let mut all_first_events = vec ! [ ] ;
716+ let stream_metas = self . get_stream_meta_from_storage ( stream_name) . await ;
717+ if let Ok ( stream_metas) = stream_metas {
718+ for stream_meta in stream_metas. iter ( ) {
719+ if let Some ( first_event) = & stream_meta. first_event_at {
720+ let first_event = DateTime :: parse_from_rfc3339 ( first_event) . unwrap ( ) ;
721+ let first_event = first_event. with_timezone ( & Utc ) ;
722+ all_first_events. push ( first_event) ;
723+ }
724+ }
725+ }
726+
727+ if all_first_events. is_empty ( ) {
728+ return Ok ( None ) ;
729+ }
730+ let first_event_at = all_first_events. iter ( ) . min ( ) . unwrap ( ) . to_rfc3339 ( ) ;
731+ Ok ( Some ( first_event_at) )
732+ }
733+
626734 // pick a better name
627735 fn get_bucket_name ( & self ) -> String ;
628736}
0 commit comments