feat(api): align basin info with stream info (#338)

Align basin lifecycle metadata with stream info while keeping the deprecated basin `state` field as a compatibility-only JSON field.

- add `created_at` and `deleted_at` to basin info in the common, API, SDK, and lite layers
- derive basin deletion state from `deleted_at` and stop publishing `state` in OpenAPI
- update CLI and TUI basin presentation to follow the stream info patterns
- tolerate missing basin `created_at` during deserialize by defaulting it to `now_utc()`

Author: shikhar

api/src/v1/basin.rs

@@ -3,6 +3,7 @@ use s2_common::types::{
     basin::{BasinName, BasinNamePrefix, BasinNameStartAfter},
 };
 use serde::{Deserialize, Serialize};
+use time::OffsetDateTime;
 
 use super::config::{BasinConfig, BasinReconfiguration};
 
@@ -41,29 +42,92 @@ pub struct ListBasinsResponse {
 }
 
 #[rustfmt::skip]
-#[derive(Debug, Clone, Serialize, Deserialize)]
+#[derive(Debug, Clone, Serialize)]
 #[cfg_attr(feature = "utoipa", derive(utoipa::ToSchema))]
 pub struct BasinInfo {
     /// Basin name.
     pub name: BasinName,
     /// Basin scope.
     pub scope: Option<BasinScope>,
-    /// Basin state.
+    /// Creation time in RFC 3339 format.
+    #[serde(with = "time::serde::rfc3339")]
+    pub created_at: OffsetDateTime,
+    /// Deletion time in RFC 3339 format, if the basin is being deleted.
+    #[serde(default, with = "time::serde::rfc3339::option")]
+    pub deleted_at: Option<OffsetDateTime>,
+    /// Deprecated basin state inferred from `deleted_at`.
+    #[cfg_attr(feature = "utoipa", schema(ignore))]
     pub state: BasinState,
 }
 
 impl From<types::basin::BasinInfo> for BasinInfo {
     fn from(value: types::basin::BasinInfo) -> Self {
-        let types::basin::BasinInfo { name, scope, state } = value;
+        let types::basin::BasinInfo {
+            name,
+            scope,
+            created_at,
+            deleted_at,
+        } = value;
 
         Self {
             name,
             scope: scope.map(Into::into),
-            state: state.into(),
+            created_at,
+            deleted_at,
+            state: basin_state_for_deleted_at(deleted_at.as_ref()),
         }
     }
 }
 
+fn basin_state_for_deleted_at(deleted_at: Option<&OffsetDateTime>) -> BasinState {
+    if deleted_at.is_some() {
+        BasinState::Deleting
+    } else {
+        BasinState::Active
+    }
+}
+
+#[derive(Deserialize)]
+struct BasinInfoSerde {
+    name: BasinName,
+    scope: Option<BasinScope>,
+    #[serde(default, with = "time::serde::rfc3339::option")]
+    created_at: Option<OffsetDateTime>,
+    #[serde(default, with = "time::serde::rfc3339::option")]
+    deleted_at: Option<OffsetDateTime>,
+    state: Option<BasinState>,
+}
+
+impl<'de> Deserialize<'de> for BasinInfo {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        let BasinInfoSerde {
+            name,
+            scope,
+            created_at,
+            deleted_at,
+            state,
+        } = BasinInfoSerde::deserialize(deserializer)?;
+        let created_at = created_at.unwrap_or_else(OffsetDateTime::now_utc);
+        let deleted_at = match (deleted_at, state) {
+            (Some(deleted_at), _) => Some(deleted_at),
+            (None, Some(BasinState::Deleting)) => Some(OffsetDateTime::now_utc()),
+            (None, _) => None,
+        };
+        let state = basin_state_for_deleted_at(deleted_at.as_ref());
+
+        Ok(Self {
+            name,
+            scope,
+            created_at,
+            deleted_at,
+            state,
+        })
+    }
+}
+
 #[rustfmt::skip]
 #[derive(Debug, Clone, Copy, Serialize, Deserialize)]
 #[cfg_attr(feature = "utoipa", derive(utoipa::ToSchema))]
@@ -100,15 +164,6 @@ pub enum BasinState {
     Deleting,
 }
 
-impl From<types::basin::BasinState> for BasinState {
-    fn from(value: types::basin::BasinState) -> Self {
-        match value {
-            types::basin::BasinState::Active => Self::Active,
-            types::basin::BasinState::Deleting => Self::Deleting,
-        }
-    }
-}
-
 #[rustfmt::skip]
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[cfg_attr(feature = "utoipa", derive(utoipa::ToSchema))]

api/src/v1/stream/sse.rs

@@ -1,4 +1,4 @@
-use std::{str::FromStr, time::Duration};
+use std::str::FromStr;
 
 use s2_common::{http::ParseableHeader, types};
 use serde::Serialize;
@@ -126,7 +126,8 @@ pub enum ReadEvent {
     },
 }
 
-fn elapsed_since_epoch() -> Duration {
+#[cfg(feature = "axum")]
+fn elapsed_since_epoch() -> std::time::Duration {
     std::time::SystemTime::now()
         .duration_since(std::time::SystemTime::UNIX_EPOCH)
         .expect("healthy clock")

cli/src/main.rs

@@ -32,9 +32,9 @@ use record_format::{
 use s2_sdk::{
     S2,
     types::{
-        AppendRetryPolicy, BasinState, CreateStreamInput, DeleteOnEmptyConfig, DeleteStreamInput,
-        MeteredBytes, Metric, RetentionPolicy, RetryConfig, StreamConfig as SdkStreamConfig,
-        StreamName, TimestampingConfig, TimestampingMode,
+        AppendRetryPolicy, CreateStreamInput, DeleteOnEmptyConfig, DeleteStreamInput, MeteredBytes,
+        Metric, RetentionPolicy, RetryConfig, StreamConfig as SdkStreamConfig, StreamName,
+        TimestampingConfig, TimestampingMode,
     },
 };
 use strum::VariantNames;
@@ -196,11 +196,10 @@ async fn run() -> Result<(), CliError> {
 
                 let (streams, _) = ops::list_streams(&s2, list_streams_args).await?;
                 for stream_info in streams {
-                    println!(
-                        "s2://{}/{} {}",
-                        basin,
-                        stream_info.name,
-                        stream_info.created_at.to_string().green(),
+                    print_listing_with_created_at(
+                        format!("s2://{}/{}", basin, stream_info.name),
+                        stream_info.created_at.to_string(),
+                        stream_info.deleted_at.is_some(),
                     );
                 }
             } else {
@@ -224,34 +223,21 @@ async fn run() -> Result<(), CliError> {
 
                 let (basins, _) = ops::list_basins(&s2, list_basins_args).await?;
                 for basin_info in basins {
-                    println!(
-                        "{} {}",
-                        basin_info.name,
-                        format_basin_state(basin_info.state)
-                    );
+                    print_listing_uri(basin_info.name.to_string(), basin_info.deleted_at.is_some());
                 }
             }
         }
 
         Command::ListBasins(args) => {
             let (basins, _) = ops::list_basins(&s2, args).await?;
             for basin_info in basins {
-                println!(
-                    "{} {}",
-                    basin_info.name,
-                    format_basin_state(basin_info.state)
-                );
+                print_listing_uri(basin_info.name.to_string(), basin_info.deleted_at.is_some());
             }
         }
 
         Command::CreateBasin(args) => {
-            let info = ops::create_basin(&s2, args).await?;
-
-            let message = match info.state {
-                BasinState::Active => "✓ Basin created".green().bold(),
-                BasinState::Deleting => "Basin is being deleted".red().bold(),
-            };
-            eprintln!("{message}");
+            let _info = ops::create_basin(&s2, args).await?;
+            eprintln!("{}", "✓ Basin created".green().bold());
         }
 
         Command::DeleteBasin { basin } => {
@@ -311,7 +297,10 @@ async fn run() -> Result<(), CliError> {
             let basin_name = args.uri.basin.clone();
             let (streams, _) = ops::list_streams(&s2, args).await?;
             for stream_info in streams {
-                println!("s2://{}/{}", basin_name, stream_info.name);
+                print_listing_uri(
+                    format!("s2://{}/{}", basin_name, stream_info.name),
+                    stream_info.deleted_at.is_some(),
+                );
             }
         }
 
@@ -623,15 +612,40 @@ async fn run() -> Result<(), CliError> {
     result.map_err(|err| err.with_token_source(token_source))
 }
 
-fn format_basin_state(state: BasinState) -> colored::ColoredString {
-    match state {
-        BasinState::Active => "active".green(),
-        BasinState::Deleting => "deleting".red(),
+fn format_position(seq_num: u64, timestamp: u64) -> String {
+    format!("{seq_num} @ {timestamp}")
+}
+
+fn print_listing_uri(uri: String, is_deleting: bool) {
+    let uri = format_listing_uri(uri, is_deleting);
+    if is_deleting {
+        println!("{} {}", uri, deletion_marker());
+    } else {
+        println!("{uri}");
     }
 }
 
-fn format_position(seq_num: u64, timestamp: u64) -> String {
-    format!("{seq_num} @ {timestamp}")
+fn print_listing_with_created_at(uri: String, created_at: String, is_deleting: bool) {
+    let uri = format_listing_uri(uri, is_deleting);
+    let created_at = if is_deleting {
+        created_at.red()
+    } else {
+        created_at.green()
+    };
+
+    if is_deleting {
+        println!("{} {} {}", uri, created_at, deletion_marker());
+    } else {
+        println!("{} {}", uri, created_at);
+    }
+}
+
+fn format_listing_uri(uri: String, is_deleting: bool) -> colored::ColoredString {
+    if is_deleting { uri.red() } else { uri.normal() }
+}
+
+fn deletion_marker() -> colored::ColoredString {
+    "[deleting]".red().bold()
 }
 
 async fn write_record(

cli/src/ops.rs

@@ -60,10 +60,9 @@ pub async fn list_basins(
             .list_basins(input)
             .await
             .map_err(|e| CliError::op(OpKind::ListBasins, e))?;
-
         Ok((page.values, page.has_more))
     } else {
-        let mut input = ListAllBasinsInput::new();
+        let mut input = ListAllBasinsInput::new().with_include_deleted(true);
         if let Some(p) = prefix {
             input = input.with_prefix(p);
         }
@@ -329,10 +328,9 @@ pub async fn list_streams(
             .list_streams(input)
             .await
             .map_err(|e| CliError::op(OpKind::ListStreams, e))?;
-
         Ok((page.values, page.has_more))
     } else {
-        let mut input = ListAllStreamsInput::new();
+        let mut input = ListAllStreamsInput::new().with_include_deleted(true);
         if let Some(p) = prefix {
             input = input.with_prefix(p);
         }

cli/src/tui/app.rs

@@ -12,7 +12,7 @@ use chrono::{Datelike, NaiveDate};
 use crossterm::event::{self, Event as CrosstermEvent, KeyCode, KeyEvent, KeyModifiers};
 use ratatui::{Terminal, prelude::Backend};
 use s2_sdk::types::{
-    AccessTokenId, AccessTokenInfo, BasinInfo, BasinMetricSet, BasinName, BasinState, StreamInfo,
+    AccessTokenId, AccessTokenInfo, BasinInfo, BasinMetricSet, BasinName, StreamInfo,
     StreamMetricSet, StreamName, StreamPosition, TimeRange,
 };
 use tokio::sync::mpsc;
@@ -3739,7 +3739,7 @@ impl App {
                     })
                     .collect();
                 if let Some(basin) = filtered.get(state.selected) {
-                    if basin.state == BasinState::Deleting {
+                    if basin.deleted_at.is_some() {
                         self.message = Some(StatusMessage {
                             text: "Basin is already being deleted".to_string(),
                             level: MessageLevel::Info,

cli/src/tui/ui.rs

@@ -2468,11 +2468,11 @@ fn draw_basins(f: &mut Frame, area: Rect, state: &BasinsState) {
         let max_name_len = name_col.saturating_sub(2);
         let display_name = truncate_str(&name, max_name_len, "…");
 
-        let (state_text, state_bg) = match basin.state {
-            s2_sdk::types::BasinState::Active => ("Active", BADGE_ACTIVE),
-            s2_sdk::types::BasinState::Deleting => ("Deleting", BADGE_DANGER),
+        let (state_text, state_bg) = if basin.deleted_at.is_some() {
+            ("Deleting", BADGE_DANGER)
+        } else {
+            ("Active", BADGE_ACTIVE)
         };
-
         let scope = basin
             .scope
             .as_ref()

common/src/types/basin.rs

@@ -1,6 +1,7 @@
 use std::{marker::PhantomData, ops::Deref, str::FromStr};
 
 use compact_str::{CompactString, ToCompactString};
+use time::OffsetDateTime;
 
 use super::{
     ValidationError,
@@ -199,12 +200,6 @@ impl crate::http::ParseableHeader for BasinName {
 
 pub type ListBasinsRequest = ListItemsRequest<BasinNamePrefix, BasinNameStartAfter>;
 
-#[derive(Debug, Clone, Copy)]
-pub enum BasinState {
-    Active,
-    Deleting,
-}
-
 #[derive(Debug, strum::Display, Clone, Copy, PartialEq, Eq)]
 pub enum BasinScope {
     #[strum(serialize = "aws:us-east-1")]
@@ -215,7 +210,8 @@ pub enum BasinScope {
 pub struct BasinInfo {
     pub name: BasinName,
     pub scope: Option<BasinScope>,
-    pub state: BasinState,
+    pub created_at: OffsetDateTime,
+    pub deleted_at: Option<OffsetDateTime>,
 }
 
 #[cfg(test)]

lite/src/backend/basins.rs

@@ -1,7 +1,7 @@
 use s2_common::{
     bash::Bash,
     types::{
-        basin::{BasinInfo, BasinName, BasinState, ListBasinsRequest},
+        basin::{BasinInfo, BasinName, ListBasinsRequest},
         config::{BasinConfig, BasinReconfiguration},
         resources::{CreateMode, ListItemsRequestParts, Page, RequestToken},
         stream::StreamNameStartAfter,
@@ -58,15 +58,11 @@ impl Backend {
                 break;
             }
             let meta = kv::basin_meta::deser_value(kv.value)?;
-            let state = if meta.deleted_at.is_some() {
-                BasinState::Deleting
-            } else {
-                BasinState::Active
-            };
             basins.push(BasinInfo {
                 name: basin,
                 scope: None,
-                state,
+                created_at: meta.created_at,
+                deleted_at: meta.deleted_at,
             });
         }
         Ok(Page::new(basins, has_more))
@@ -106,7 +102,8 @@ impl Backend {
                         Ok(CreatedOrReconfigured::Created(BasinInfo {
                             name: basin,
                             scope: None,
-                            state: BasinState::Active,
+                            created_at: existing_meta.created_at,
+                            deleted_at: None,
                         }))
                     } else {
                         Err(BasinAlreadyExistsError { basin }.into())
@@ -149,7 +146,8 @@ impl Backend {
         let info = BasinInfo {
             name: basin,
             scope: None,
-            state: BasinState::Active,
+            created_at,
+            deleted_at: None,
         };
 
         Ok(if is_reconfigure {