s2-cli 0.40.1

use std::{str::FromStr, time::Duration};

use clap::{Args, Parser, ValueEnum};
use colored::Colorize;
use s2_sdk::{
    self as sdk,
    types::{
        AccessTokenId, AccessTokenIdPrefix, BasinName, BasinNamePrefix, EncryptionAlgorithm,
        StreamName, StreamNamePrefix, TimeseriesInterval,
    },
};
use serde::Serialize;

use crate::error::{OpGroupsParseError, S2UriParseError};

#[derive(Debug, Clone, PartialEq)]
struct S2Uri {
    basin: BasinName,
    stream: Option<String>,
}

impl FromStr for S2Uri {
    type Err = S2UriParseError;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        let (scheme, s) = s
            .split_once("://")
            .ok_or(S2UriParseError::MissingUriScheme)?;
        if scheme != "s2" {
            return Err(S2UriParseError::InvalidUriScheme(scheme.to_owned()));
        }

        let (basin, stream) = match s.split_once("/") {
            Some((basin, stream)) => (basin, (!stream.is_empty()).then(|| stream.to_owned())),
            None => (s, None),
        };

        Ok(S2Uri {
            basin: basin
                .parse()
                .map_err(|e| S2UriParseError::InvalidBasinName(format!("{e}")))?,
            stream,
        })
    }
}

#[derive(Debug, Clone, PartialEq)]
pub struct S2BasinUri(pub BasinName);

impl From<S2BasinUri> for BasinName {
    fn from(value: S2BasinUri) -> Self {
        value.0
    }
}

impl FromStr for S2BasinUri {
    type Err = S2UriParseError;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        match S2Uri::from_str(s) {
            Ok(S2Uri {
                basin,
                stream: None,
            }) => Ok(Self(basin)),
            Ok(S2Uri {
                basin: _,
                stream: Some(_),
            }) => Err(S2UriParseError::UnexpectedStreamName),
            Err(S2UriParseError::MissingUriScheme) => {
                Ok(Self(s.parse().map_err(|e| {
                    S2UriParseError::InvalidBasinName(format!("{e}"))
                })?))
            }
            Err(other) => Err(other),
        }
    }
}

#[derive(Debug, Clone, PartialEq)]
pub struct S2BasinAndMaybeStreamUri {
    pub basin: BasinName,
    pub stream: Option<StreamNamePrefix>,
}

impl FromStr for S2BasinAndMaybeStreamUri {
    type Err = S2UriParseError;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        match S2Uri::from_str(s) {
            Ok(S2Uri { basin, stream }) => {
                let stream = stream
                    .map(|s| {
                        s.parse()
                            .map_err(|e| S2UriParseError::InvalidStreamName(format!("{e}")))
                    })
                    .transpose()?;
                Ok(Self { basin, stream })
            }
            Err(S2UriParseError::MissingUriScheme) => Ok(Self {
                basin: s
                    .parse()
                    .map_err(|e| S2UriParseError::InvalidBasinName(format!("{e}")))?,
                stream: None,
            }),
            Err(other) => Err(other),
        }
    }
}

/// String Format: s2://{basin}/{stream}
#[derive(Debug, Clone, PartialEq)]
pub struct S2BasinAndStreamUri {
    pub basin: BasinName,
    pub stream: StreamName,
}

impl FromStr for S2BasinAndStreamUri {
    type Err = S2UriParseError;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        let S2Uri { basin, stream } = s.parse()?;
        let stream = stream.ok_or(S2UriParseError::MissingStreamName)?;
        let stream: StreamName = stream
            .parse()
            .map_err(|e| S2UriParseError::InvalidStreamName(format!("{e}")))?;
        Ok(Self { basin, stream })
    }
}

#[derive(Debug, Clone, PartialEq)]
pub enum DiffResource {
    Basin(BasinName),
    Stream(S2BasinAndStreamUri),
    AccessToken(AccessTokenId),
}

impl FromStr for DiffResource {
    type Err = String;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        if !s.contains("://") {
            return Err(
                "resource type cannot be inferred from a bare name; use `--resource`".to_owned(),
            );
        }

        let S2Uri { basin, stream } = s.parse().map_err(|e: S2UriParseError| e.to_string())?;
        match stream {
            Some(stream) => Ok(Self::Stream(S2BasinAndStreamUri {
                basin,
                stream: stream.parse().map_err(|e| format!("{e}"))?,
            })),
            None => Ok(Self::Basin(basin)),
        }
    }
}

#[derive(Parser, Debug, Clone, Serialize)]
pub struct BasinConfig {
    #[clap(flatten)]
    pub default_stream_config: StreamConfig,
    /// Encryption algorithm to apply to newly created streams in this basin.
    #[arg(long)]
    pub stream_cipher: Option<EncryptionAlgorithm>,
    /// Create stream on append with basin defaults if it doesn't exist.
    #[arg(long, default_value_t = false)]
    pub create_stream_on_append: bool,
    /// Create stream on read with basin defaults if it doesn't exist.
    #[arg(long, default_value_t = false)]
    pub create_stream_on_read: bool,
}

#[derive(Parser, Debug, Clone, Serialize, Default)]
pub struct StreamConfig {
    #[arg(long)]
    /// Storage class for a stream.
    pub storage_class: Option<StorageClass>,
    #[arg(long, help("Example: 1d, 1w, 1y"))]
    /// Retention policy for a stream.
    pub retention_policy: Option<RetentionPolicy>,
    #[clap(flatten)]
    /// Timestamping configuration.
    pub timestamping: Option<TimestampingConfig>,
    #[clap(flatten)]
    /// Delete-on-empty configuration.
    pub delete_on_empty: Option<DeleteOnEmptyConfig>,
}

impl StreamConfig {
    pub fn is_empty(&self) -> bool {
        let Self {
            storage_class,
            retention_policy,
            timestamping,
            delete_on_empty,
        } = self;
        storage_class.is_none()
            && retention_policy.is_none()
            && timestamping.is_none()
            && delete_on_empty.is_none()
    }
}

pub use sdk::types::LocationName;

#[derive(ValueEnum, Debug, Clone, Serialize)]
#[serde(rename_all = "kebab-case")]
pub enum StorageClass {
    Standard,
    Express,
}

#[derive(ValueEnum, Debug, Clone, Serialize)]
#[serde(rename_all = "kebab-case")]
pub enum TimestampingMode {
    ClientPrefer,
    ClientRequire,
    Arrival,
}

#[derive(Parser, Debug, Clone, Serialize)]
pub struct TimestampingConfig {
    #[arg(long)]
    /// Timestamping mode.
    pub timestamping_mode: Option<TimestampingMode>,

    #[arg(long)]
    /// Uncapped timestamps.
    pub timestamping_uncapped: Option<bool>,
}

#[derive(Clone, Debug, Serialize)]
pub enum RetentionPolicy {
    Age(#[serde(serialize_with = "serialize_duration_humantime")] Duration),
    Infinite,
}

impl TryFrom<&str> for RetentionPolicy {
    type Error = &'static str;

    fn try_from(value: &str) -> Result<Self, Self::Error> {
        if value == "infinite" {
            return Ok(RetentionPolicy::Infinite);
        } else if let Ok(d) = humantime::parse_duration(value) {
            return Ok(RetentionPolicy::Age(d));
        }
        Err("invalid retention policy: expected a duration, or 'infinite'")
    }
}

impl FromStr for RetentionPolicy {
    type Err = &'static str;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        RetentionPolicy::try_from(s)
    }
}

#[derive(Args, Clone, Debug, Serialize)]
pub struct DeleteOnEmptyConfig {
    #[arg(long, value_parser = humantime::parse_duration, required = false)]
    #[serde(serialize_with = "serialize_duration_humantime")]
    /// Minimum age before an empty stream can be deleted.
    /// Example: 1d, 1w, 1y
    pub delete_on_empty_min_age: Duration,
}

impl From<DeleteOnEmptyConfig> for sdk::types::DeleteOnEmptyConfig {
    fn from(value: DeleteOnEmptyConfig) -> Self {
        sdk::types::DeleteOnEmptyConfig::new().with_min_age(value.delete_on_empty_min_age)
    }
}

impl From<DeleteOnEmptyConfig> for sdk::types::DeleteOnEmptyReconfiguration {
    fn from(value: DeleteOnEmptyConfig) -> Self {
        sdk::types::DeleteOnEmptyReconfiguration::new().with_min_age(value.delete_on_empty_min_age)
    }
}

impl From<sdk::types::DeleteOnEmptyConfig> for DeleteOnEmptyConfig {
    fn from(value: sdk::types::DeleteOnEmptyConfig) -> Self {
        Self {
            delete_on_empty_min_age: Duration::from_secs(value.min_age_secs),
        }
    }
}

impl From<BasinConfig> for sdk::types::BasinConfig {
    fn from(config: BasinConfig) -> Self {
        let mut basin_config = sdk::types::BasinConfig::new()
            .with_default_stream_config(config.default_stream_config.into());
        if let Some(algorithm) = config.stream_cipher {
            basin_config = basin_config.with_stream_cipher(algorithm);
        }
        basin_config
            .with_create_stream_on_append(config.create_stream_on_append)
            .with_create_stream_on_read(config.create_stream_on_read)
    }
}

impl From<StreamConfig> for sdk::types::StreamConfig {
    fn from(config: StreamConfig) -> Self {
        let mut stream_config = sdk::types::StreamConfig::new();
        if let Some(storage_class) = config.storage_class {
            stream_config = stream_config.with_storage_class(storage_class.into());
        }
        if let Some(retention_policy) = config.retention_policy {
            stream_config = stream_config.with_retention_policy(retention_policy.into());
        }
        if let Some(timestamping) = config.timestamping {
            stream_config = stream_config.with_timestamping(timestamping.into());
        }
        if let Some(delete_on_empty) = config.delete_on_empty {
            stream_config = stream_config.with_delete_on_empty(delete_on_empty.into());
        }
        stream_config
    }
}

impl From<StorageClass> for sdk::types::StorageClass {
    fn from(class: StorageClass) -> Self {
        match class {
            StorageClass::Standard => sdk::types::StorageClass::Standard,
            StorageClass::Express => sdk::types::StorageClass::Express,
        }
    }
}

impl From<sdk::types::StorageClass> for StorageClass {
    fn from(class: sdk::types::StorageClass) -> Self {
        match class {
            sdk::types::StorageClass::Standard => StorageClass::Standard,
            sdk::types::StorageClass::Express => StorageClass::Express,
        }
    }
}

impl From<TimestampingMode> for sdk::types::TimestampingMode {
    fn from(mode: TimestampingMode) -> Self {
        match mode {
            TimestampingMode::ClientPrefer => sdk::types::TimestampingMode::ClientPrefer,
            TimestampingMode::ClientRequire => sdk::types::TimestampingMode::ClientRequire,
            TimestampingMode::Arrival => sdk::types::TimestampingMode::Arrival,
        }
    }
}

impl From<sdk::types::TimestampingMode> for TimestampingMode {
    fn from(mode: sdk::types::TimestampingMode) -> Self {
        match mode {
            sdk::types::TimestampingMode::ClientPrefer => TimestampingMode::ClientPrefer,
            sdk::types::TimestampingMode::ClientRequire => TimestampingMode::ClientRequire,
            sdk::types::TimestampingMode::Arrival => TimestampingMode::Arrival,
        }
    }
}

impl From<TimestampingConfig> for sdk::types::TimestampingConfig {
    fn from(config: TimestampingConfig) -> Self {
        let mut result = sdk::types::TimestampingConfig::new();
        if let Some(mode) = config.timestamping_mode {
            result = result.with_mode(mode.into());
        }
        if let Some(uncapped) = config.timestamping_uncapped {
            result = result.with_uncapped(uncapped);
        }
        result
    }
}

impl From<sdk::types::TimestampingConfig> for TimestampingConfig {
    fn from(config: sdk::types::TimestampingConfig) -> Self {
        TimestampingConfig {
            timestamping_mode: config.mode.map(Into::into),
            timestamping_uncapped: config.uncapped,
        }
    }
}

impl From<RetentionPolicy> for sdk::types::RetentionPolicy {
    fn from(policy: RetentionPolicy) -> Self {
        match policy {
            RetentionPolicy::Age(d) => sdk::types::RetentionPolicy::Age(d.as_secs()),
            RetentionPolicy::Infinite => sdk::types::RetentionPolicy::Infinite,
        }
    }
}

impl From<sdk::types::RetentionPolicy> for RetentionPolicy {
    fn from(policy: sdk::types::RetentionPolicy) -> Self {
        match policy {
            sdk::types::RetentionPolicy::Age(secs) => {
                RetentionPolicy::Age(Duration::from_secs(secs))
            }
            sdk::types::RetentionPolicy::Infinite => RetentionPolicy::Infinite,
        }
    }
}

impl From<sdk::types::BasinConfig> for BasinConfig {
    fn from(config: sdk::types::BasinConfig) -> Self {
        BasinConfig {
            default_stream_config: config
                .default_stream_config
                .map(Into::into)
                .unwrap_or_default(),
            stream_cipher: config.stream_cipher,
            create_stream_on_append: config.create_stream_on_append,
            create_stream_on_read: config.create_stream_on_read,
        }
    }
}

impl From<sdk::types::StreamConfig> for StreamConfig {
    fn from(config: sdk::types::StreamConfig) -> Self {
        StreamConfig {
            storage_class: config.storage_class.map(Into::into),
            retention_policy: config.retention_policy.map(Into::into),
            timestamping: config.timestamping.map(Into::into),
            delete_on_empty: config.delete_on_empty.map(Into::into),
        }
    }
}

impl From<StreamConfig> for sdk::types::StreamReconfiguration {
    fn from(config: StreamConfig) -> Self {
        let mut reconfig = sdk::types::StreamReconfiguration::new();
        if let Some(storage_class) = config.storage_class {
            reconfig = reconfig.with_storage_class(storage_class.into());
        }
        if let Some(retention_policy) = config.retention_policy {
            reconfig = reconfig.with_retention_policy(retention_policy.into());
        }
        if let Some(timestamping) = config.timestamping {
            let ts_reconfig = sdk::types::TimestampingReconfiguration::from(timestamping);
            reconfig = reconfig.with_timestamping(ts_reconfig);
        }
        if let Some(delete_on_empty) = config.delete_on_empty {
            reconfig = reconfig.with_delete_on_empty(delete_on_empty.into());
        }
        reconfig
    }
}

impl From<TimestampingConfig> for sdk::types::TimestampingReconfiguration {
    fn from(config: TimestampingConfig) -> Self {
        let mut result = sdk::types::TimestampingReconfiguration::new();
        if let Some(mode) = config.timestamping_mode {
            result = result.with_mode(mode.into());
        }
        if let Some(uncapped) = config.timestamping_uncapped {
            result = result.with_uncapped(uncapped);
        }
        result
    }
}

#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum BasinMatcher {
    #[serde(serialize_with = "serialize_display")]
    Exact(BasinName),
    #[serde(serialize_with = "serialize_display")]
    Prefix(BasinNamePrefix),
}

#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum StreamMatcher {
    #[serde(serialize_with = "serialize_display")]
    Exact(StreamName),
    #[serde(serialize_with = "serialize_display")]
    Prefix(StreamNamePrefix),
}

#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum AccessTokenMatcher {
    #[serde(serialize_with = "serialize_display")]
    Exact(AccessTokenId),
    #[serde(serialize_with = "serialize_display")]
    Prefix(AccessTokenIdPrefix),
}

fn serialize_display<T, S>(value: &T, serializer: S) -> Result<S::Ok, S::Error>
where
    T: std::fmt::Display,
    S: serde::Serializer,
{
    serializer.serialize_str(&value.to_string())
}

fn serialize_duration_humantime<S>(value: &Duration, serializer: S) -> Result<S::Ok, S::Error>
where
    S: serde::Serializer,
{
    serializer.serialize_str(&humantime::format_duration(*value).to_string())
}

impl From<sdk::types::BasinMatcher> for BasinMatcher {
    fn from(matcher: sdk::types::BasinMatcher) -> Self {
        match matcher {
            sdk::types::BasinMatcher::Exact(v) => BasinMatcher::Exact(v),
            sdk::types::BasinMatcher::Prefix(v) => BasinMatcher::Prefix(v),
            sdk::types::BasinMatcher::None => BasinMatcher::Prefix(Default::default()),
        }
    }
}

impl From<sdk::types::StreamMatcher> for StreamMatcher {
    fn from(matcher: sdk::types::StreamMatcher) -> Self {
        match matcher {
            sdk::types::StreamMatcher::Exact(v) => StreamMatcher::Exact(v),
            sdk::types::StreamMatcher::Prefix(v) => StreamMatcher::Prefix(v),
            sdk::types::StreamMatcher::None => StreamMatcher::Prefix(Default::default()),
        }
    }
}

impl From<sdk::types::AccessTokenMatcher> for AccessTokenMatcher {
    fn from(matcher: sdk::types::AccessTokenMatcher) -> Self {
        match matcher {
            sdk::types::AccessTokenMatcher::Exact(v) => AccessTokenMatcher::Exact(v),
            sdk::types::AccessTokenMatcher::Prefix(v) => AccessTokenMatcher::Prefix(v),
            sdk::types::AccessTokenMatcher::None => AccessTokenMatcher::Prefix(Default::default()),
        }
    }
}

/// Compact human summary for a resource matcher, used in `s2 list-access-tokens`:
///
/// - `*` matches any resource (an empty prefix).
/// - `<prefix>*` matches names beginning with `<prefix>`.
/// - `<name>` matches exactly that name.
macro_rules! impl_matcher_summary {
    ($($matcher:ty),+ $(,)?) => {$(
        impl $matcher {
            fn summary(&self) -> String {
                match self {
                    Self::Exact(name) => name.to_string(),
                    Self::Prefix(prefix) => format!("{prefix}*"),
                }
            }
        }
    )+};
}
impl_matcher_summary!(BasinMatcher, StreamMatcher, AccessTokenMatcher);

/// Renders an optional matcher, using `∅` when the scope grants no access to that resource.
fn matcher_summary(matcher: Option<String>) -> String {
    matcher.unwrap_or_else(|| "∅".to_owned())
}

#[derive(Debug, Clone, Serialize, PartialEq)]
pub struct PermittedOperationGroups {
    pub account: Option<ReadWritePermissions>,
    pub basin: Option<ReadWritePermissions>,
    pub stream: Option<ReadWritePermissions>,
}

impl PermittedOperationGroups {
    /// Compact grant summary like `a:rw b:r s:rw`, labelling account/basin/stream. Groups with
    /// no permission are omitted; a token with no group permissions renders as `none`.
    fn summary(&self) -> String {
        let rendered: Vec<String> = [
            ("a", self.account.as_ref()),
            ("b", self.basin.as_ref()),
            ("s", self.stream.as_ref()),
        ]
        .into_iter()
        .filter_map(|(label, perm)| {
            let perm = perm?;
            let mut rw = String::new();
            if perm.read {
                rw.push('r');
            }
            if perm.write {
                rw.push('w');
            }
            (!rw.is_empty()).then(|| format!("{label}:{rw}"))
        })
        .collect();

        if rendered.is_empty() {
            "none".to_owned()
        } else {
            rendered.join(" ")
        }
    }
}

impl From<PermittedOperationGroups> for sdk::types::OperationGroupPermissions {
    fn from(groups: PermittedOperationGroups) -> Self {
        let mut result = sdk::types::OperationGroupPermissions::new();
        if let Some(account) = groups.account {
            result = result.with_account(account.into());
        }
        if let Some(basin) = groups.basin {
            result = result.with_basin(basin.into());
        }
        if let Some(stream) = groups.stream {
            result = result.with_stream(stream.into());
        }
        result
    }
}

impl From<sdk::types::OperationGroupPermissions> for PermittedOperationGroups {
    fn from(groups: sdk::types::OperationGroupPermissions) -> Self {
        PermittedOperationGroups {
            account: groups.account.map(Into::into),
            basin: groups.basin.map(Into::into),
            stream: groups.stream.map(Into::into),
        }
    }
}

impl FromStr for PermittedOperationGroups {
    type Err = OpGroupsParseError;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        let mut account = None;
        let mut basin = None;
        let mut stream = None;

        if s.is_empty() {
            return Ok(PermittedOperationGroups {
                account,
                basin,
                stream,
            });
        }

        for part in s.split(',') {
            let part = part.trim();
            if part.is_empty() {
                continue;
            }
            let (key, value) =
                part.split_once('=')
                    .ok_or_else(|| OpGroupsParseError::InvalidFormat {
                        value: part.to_owned(),
                    })?;
            let perms = value.parse::<ReadWritePermissions>()?;
            match key {
                "account" => account = Some(perms),
                "basin" => basin = Some(perms),
                "stream" => stream = Some(perms),
                _ => {
                    return Err(OpGroupsParseError::InvalidKey {
                        key: key.to_owned(),
                    });
                }
            }
        }

        Ok(PermittedOperationGroups {
            account,
            basin,
            stream,
        })
    }
}

#[derive(Debug, Clone, Serialize, PartialEq)]
pub struct ReadWritePermissions {
    pub read: bool,
    pub write: bool,
}

impl FromStr for ReadWritePermissions {
    type Err = OpGroupsParseError;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        let mut read = false;
        let mut write = false;
        for c in s.chars() {
            match c {
                'r' => read = true,
                'w' => write = true,
                _ => return Err(OpGroupsParseError::InvalidPermissionChar(c)),
            }
        }
        if !read && !write {
            return Err(OpGroupsParseError::MissingPermission);
        }
        Ok(ReadWritePermissions { read, write })
    }
}

impl From<ReadWritePermissions> for sdk::types::ReadWritePermissions {
    fn from(permissions: ReadWritePermissions) -> Self {
        match (permissions.read, permissions.write) {
            (true, true) => sdk::types::ReadWritePermissions::read_write(),
            (true, false) => sdk::types::ReadWritePermissions::read_only(),
            (false, true) => sdk::types::ReadWritePermissions::write_only(),
            (false, false) => sdk::types::ReadWritePermissions::new(),
        }
    }
}

impl From<sdk::types::ReadWritePermissions> for ReadWritePermissions {
    fn from(permissions: sdk::types::ReadWritePermissions) -> Self {
        ReadWritePermissions {
            read: permissions.read,
            write: permissions.write,
        }
    }
}

#[derive(Debug, Serialize)]
pub struct AccessTokenInfo {
    pub id: String,
    pub expires_at: Option<String>,
    pub auto_prefix_streams: bool,
    pub scope: AccessTokenScope,
}

impl From<sdk::types::AccessTokenInfo> for AccessTokenInfo {
    fn from(info: sdk::types::AccessTokenInfo) -> Self {
        AccessTokenInfo {
            id: info.id.to_string(),
            expires_at: info.expires_at.map(|expires_at| expires_at.to_string()),
            auto_prefix_streams: info.auto_prefix_streams,
            scope: info.scope.into(),
        }
    }
}

impl AccessTokenInfo {
    /// Two-line compact summary for `s2 list-access-tokens`.
    ///
    /// - Header: the token id (padded to `id_width` for column alignment across the listing)
    ///   followed by its expiry, or `never` when the token does not expire.
    /// - Detail: a single indented line summarizing the scope — the basin/stream/token matchers,
    ///   the operation-group permissions, and the count of explicitly granted operations.
    pub fn summary_block(&self, id_width: usize) -> String {
        let expires = match &self.expires_at {
            Some(at) => at.green().to_string(),
            None => "never".dimmed().to_string(),
        };
        let padding = " ".repeat(id_width.saturating_sub(self.id.len()));
        let header = format!("{}{padding}  expires {expires}", self.id.bold());

        let scope = &self.scope;
        let detail = format!(
            "  basins={}  streams={}  tokens={}  perms={}  ops={}",
            matcher_summary(scope.basins.as_ref().map(BasinMatcher::summary)),
            matcher_summary(scope.streams.as_ref().map(StreamMatcher::summary)),
            matcher_summary(
                scope
                    .access_tokens
                    .as_ref()
                    .map(AccessTokenMatcher::summary)
            ),
            scope
                .op_group_perms
                .as_ref()
                .map_or_else(|| "none".to_owned(), PermittedOperationGroups::summary),
            scope.ops.len(),
        );

        format!("{header}\n{detail}")
    }
}

#[derive(Debug, Serialize)]
pub struct AccessTokenScope {
    pub basins: Option<BasinMatcher>,
    pub streams: Option<StreamMatcher>,
    pub access_tokens: Option<AccessTokenMatcher>,
    pub op_group_perms: Option<PermittedOperationGroups>,
    pub ops: Vec<Operation>,
}

impl From<sdk::types::AccessTokenScope> for AccessTokenScope {
    fn from(scope: sdk::types::AccessTokenScope) -> Self {
        AccessTokenScope {
            basins: scope.basins.map(Into::into),
            streams: scope.streams.map(Into::into),
            access_tokens: scope.access_tokens.map(Into::into),
            op_group_perms: scope.op_group_perms.map(Into::into),
            ops: scope.ops.into_iter().map(Operation::from).collect(),
        }
    }
}

#[derive(
    Debug, Clone, PartialEq, Eq, Serialize, clap::ValueEnum, strum::Display, strum::EnumString,
)]
#[serde(rename_all = "snake_case")]
#[clap(rename_all = "snake_case")]
#[strum(serialize_all = "snake_case")]
pub enum Operation {
    ListBasins,
    CreateBasin,
    DeleteBasin,
    GetBasinConfig,
    ReconfigureBasin,
    ListAccessTokens,
    IssueAccessToken,
    RevokeAccessToken,
    GetAccountMetrics,
    GetBasinMetrics,
    GetStreamMetrics,
    ListStreams,
    CreateStream,
    DeleteStream,
    GetStreamConfig,
    ReconfigureStream,
    CheckTail,
    Trim,
    Fence,
    Append,
    Read,
    ListLocations,
    GetDefaultLocation,
    SetDefaultLocation,
}

impl From<Operation> for sdk::types::Operation {
    fn from(op: Operation) -> Self {
        match op {
            Operation::ListBasins => sdk::types::Operation::ListBasins,
            Operation::CreateBasin => sdk::types::Operation::CreateBasin,
            Operation::DeleteBasin => sdk::types::Operation::DeleteBasin,
            Operation::GetBasinConfig => sdk::types::Operation::GetBasinConfig,
            Operation::ReconfigureBasin => sdk::types::Operation::ReconfigureBasin,
            Operation::ListAccessTokens => sdk::types::Operation::ListAccessTokens,
            Operation::IssueAccessToken => sdk::types::Operation::IssueAccessToken,
            Operation::RevokeAccessToken => sdk::types::Operation::RevokeAccessToken,
            Operation::GetAccountMetrics => sdk::types::Operation::GetAccountMetrics,
            Operation::GetBasinMetrics => sdk::types::Operation::GetBasinMetrics,
            Operation::GetStreamMetrics => sdk::types::Operation::GetStreamMetrics,
            Operation::ListLocations => sdk::types::Operation::ListLocations,
            Operation::GetDefaultLocation => sdk::types::Operation::GetDefaultLocation,
            Operation::SetDefaultLocation => sdk::types::Operation::SetDefaultLocation,
            Operation::ListStreams => sdk::types::Operation::ListStreams,
            Operation::CreateStream => sdk::types::Operation::CreateStream,
            Operation::DeleteStream => sdk::types::Operation::DeleteStream,
            Operation::GetStreamConfig => sdk::types::Operation::GetStreamConfig,
            Operation::ReconfigureStream => sdk::types::Operation::ReconfigureStream,
            Operation::CheckTail => sdk::types::Operation::CheckTail,
            Operation::Trim => sdk::types::Operation::Trim,
            Operation::Fence => sdk::types::Operation::Fence,
            Operation::Append => sdk::types::Operation::Append,
            Operation::Read => sdk::types::Operation::Read,
        }
    }
}

impl From<sdk::types::Operation> for Operation {
    fn from(op: sdk::types::Operation) -> Self {
        match op {
            sdk::types::Operation::ListBasins => Operation::ListBasins,
            sdk::types::Operation::CreateBasin => Operation::CreateBasin,
            sdk::types::Operation::DeleteBasin => Operation::DeleteBasin,
            sdk::types::Operation::GetBasinConfig => Operation::GetBasinConfig,
            sdk::types::Operation::ReconfigureBasin => Operation::ReconfigureBasin,
            sdk::types::Operation::ListAccessTokens => Operation::ListAccessTokens,
            sdk::types::Operation::IssueAccessToken => Operation::IssueAccessToken,
            sdk::types::Operation::RevokeAccessToken => Operation::RevokeAccessToken,
            sdk::types::Operation::GetAccountMetrics => Operation::GetAccountMetrics,
            sdk::types::Operation::GetBasinMetrics => Operation::GetBasinMetrics,
            sdk::types::Operation::GetStreamMetrics => Operation::GetStreamMetrics,
            sdk::types::Operation::ListLocations => Operation::ListLocations,
            sdk::types::Operation::GetDefaultLocation => Operation::GetDefaultLocation,
            sdk::types::Operation::SetDefaultLocation => Operation::SetDefaultLocation,
            sdk::types::Operation::ListStreams => Operation::ListStreams,
            sdk::types::Operation::CreateStream => Operation::CreateStream,
            sdk::types::Operation::DeleteStream => Operation::DeleteStream,
            sdk::types::Operation::GetStreamConfig => Operation::GetStreamConfig,
            sdk::types::Operation::ReconfigureStream => Operation::ReconfigureStream,
            sdk::types::Operation::CheckTail => Operation::CheckTail,
            sdk::types::Operation::Trim => Operation::Trim,
            sdk::types::Operation::Fence => Operation::Fence,
            sdk::types::Operation::Append => Operation::Append,
            sdk::types::Operation::Read => Operation::Read,
        }
    }
}

#[derive(ValueEnum, Debug, Clone, Copy)]
pub enum Interval {
    /// Per-minute intervals.
    Minute,
    /// Per-hour intervals.
    Hour,
    /// Per-day intervals.
    Day,
}

impl From<Interval> for TimeseriesInterval {
    fn from(value: Interval) -> Self {
        match value {
            Interval::Minute => TimeseriesInterval::Minute,
            Interval::Hour => TimeseriesInterval::Hour,
            Interval::Day => TimeseriesInterval::Day,
        }
    }
}

#[derive(Debug, Clone)]
pub struct LatencyStats {
    pub min: std::time::Duration,
    pub p50: std::time::Duration,
    pub p90: std::time::Duration,
    pub p99: std::time::Duration,
    pub max: std::time::Duration,
}

impl LatencyStats {
    pub fn into_vec(self) -> Vec<(String, std::time::Duration)> {
        vec![
            ("min".to_owned(), self.min),
            ("p50".to_owned(), self.p50),
            ("p90".to_owned(), self.p90),
            ("p99".to_owned(), self.p99),
            ("max".to_owned(), self.max),
        ]
    }
}

#[cfg(test)]
mod tests {
    use rstest::rstest;

    use super::{
        AccessTokenInfo, AccessTokenMatcher, AccessTokenScope, BasinMatcher, OpGroupsParseError,
        PermittedOperationGroups, ReadWritePermissions, S2BasinAndMaybeStreamUri,
        S2BasinAndStreamUri, S2BasinUri, S2Uri, StreamMatcher, matcher_summary,
    };
    use crate::error::S2UriParseError;

    fn rw(read: bool, write: bool) -> ReadWritePermissions {
        ReadWritePermissions { read, write }
    }

    #[test]
    fn matcher_summaries_render_any_prefix_and_exact() {
        assert_eq!(
            BasinMatcher::Prefix("".parse().unwrap()).summary(),
            "*",
            "an empty prefix matches any resource"
        );
        assert_eq!(
            BasinMatcher::Prefix("ls-".parse().unwrap()).summary(),
            "ls-*"
        );
        assert_eq!(
            BasinMatcher::Exact("exact-basin".parse().unwrap()).summary(),
            "exact-basin"
        );
        assert_eq!(
            matcher_summary(None),
            "∅",
            "an unset matcher grants nothing"
        );
    }

    #[test]
    fn op_group_perms_summary_omits_empty_groups() {
        assert_eq!(
            PermittedOperationGroups {
                account: Some(rw(true, true)),
                basin: Some(rw(true, true)),
                stream: Some(rw(true, true)),
            }
            .summary(),
            "a:rw b:rw s:rw"
        );
        assert_eq!(
            PermittedOperationGroups {
                account: None,
                basin: Some(rw(true, false)),
                stream: Some(rw(false, true)),
            }
            .summary(),
            "b:r s:w"
        );
        assert_eq!(
            PermittedOperationGroups {
                account: None,
                basin: None,
                stream: None,
            }
            .summary(),
            "none"
        );
    }

    #[test]
    fn summary_block_renders_header_and_scope() {
        colored::control::set_override(false);

        let info = AccessTokenInfo {
            id: "tok".to_owned(),
            expires_at: None,
            auto_prefix_streams: false,
            scope: AccessTokenScope {
                basins: Some(BasinMatcher::Prefix("".parse().unwrap())),
                streams: Some(StreamMatcher::Prefix("tenant/".parse().unwrap())),
                access_tokens: None,
                op_group_perms: Some(PermittedOperationGroups {
                    account: Some(rw(true, false)),
                    basin: None,
                    stream: None,
                }),
                ops: vec![],
            },
        };

        // id padded to width 8, unset expiry renders `never`, unset matcher renders `∅`.
        assert_eq!(
            info.summary_block(8),
            "tok       expires never\n\
             \x20 basins=*  streams=tenant/*  tokens=∅  perms=a:r  ops=0"
        );
    }

    #[test]
    fn summary_block_renders_expiry_and_exact_token_matcher() {
        colored::control::set_override(false);

        let info = AccessTokenInfo {
            id: "prod".to_owned(),
            expires_at: Some("2030-01-01T00:00:00Z".to_owned()),
            auto_prefix_streams: false,
            scope: AccessTokenScope {
                basins: None,
                streams: None,
                access_tokens: Some(AccessTokenMatcher::Exact("root".parse().unwrap())),
                op_group_perms: None,
                ops: vec![super::Operation::Read, super::Operation::Append],
            },
        };

        assert_eq!(
            info.summary_block(4),
            "prod  expires 2030-01-01T00:00:00Z\n\
             \x20 basins=∅  streams=∅  tokens=root  perms=none  ops=2"
        );
    }

    #[rstest]
    #[case("", Ok(PermittedOperationGroups {
        account: None,
        basin: None,
        stream: None,
    }))]
    #[case("account=r", Ok(PermittedOperationGroups {
        account: Some(ReadWritePermissions {
            read: true,
            write: false,
        }),
        basin: None,
        stream: None,
    }))]
    #[case("account=w", Ok(PermittedOperationGroups {
        account: Some(ReadWritePermissions {
            read: false,
            write: true,
        }),
        basin: None,
        stream: None,
    }))]
    #[case("account=rw", Ok(PermittedOperationGroups {
        account: Some(ReadWritePermissions {
            read: true,
            write: true,
        }),
        basin: None,
        stream: None,
    }))]
    #[case("basin=r,stream=w", Ok(PermittedOperationGroups {
        account: None,
        basin: Some(ReadWritePermissions {
            read: true,
            write: false,
        }),
        stream: Some(ReadWritePermissions {
            read: false,
            write: true,
        }),
    }))]
    #[case("account=rw,basin=rw,stream=rw", Ok(PermittedOperationGroups {
        account: Some(ReadWritePermissions {
            read: true,
            write: true,
        }),
        basin: Some(ReadWritePermissions {
            read: true,
            write: true,
        }),
        stream: Some(ReadWritePermissions {
            read: true,
            write: true,
        }),
    }))]
    #[case("invalid", Err(OpGroupsParseError::InvalidFormat { value: "invalid".to_owned() }))]
    #[case("unknown=rw", Err(OpGroupsParseError::InvalidKey { key: "unknown".to_owned() }))]
    #[case("account=", Err(OpGroupsParseError::MissingPermission))]
    #[case("account=x", Err(OpGroupsParseError::InvalidPermissionChar('x')))]
    fn test_parse_op_groups(
        #[case] input: &str,
        #[case] expected: Result<PermittedOperationGroups, OpGroupsParseError>,
    ) {
        assert_eq!(
            input.parse::<PermittedOperationGroups>(),
            expected,
            "Testing input: {input}"
        );
    }

    #[test]
    fn test_s2_uri_parse() {
        let test_cases = vec![
            (
                "valid-basin",
                Err(S2UriParseError::MissingUriScheme),
                Ok(S2BasinUri("valid-basin".parse().unwrap())),
                Err(S2UriParseError::MissingUriScheme),
                Ok(S2BasinAndMaybeStreamUri {
                    basin: "valid-basin".parse().unwrap(),
                    stream: None,
                }),
            ),
            (
                "s2://valid-basin",
                Ok(S2Uri {
                    basin: "valid-basin".parse().unwrap(),
                    stream: None,
                }),
                Ok(S2BasinUri("valid-basin".parse().unwrap())),
                Err(S2UriParseError::MissingStreamName),
                Ok(S2BasinAndMaybeStreamUri {
                    basin: "valid-basin".parse().unwrap(),
                    stream: None,
                }),
            ),
            (
                "s2://valid-basin/",
                Ok(S2Uri {
                    basin: "valid-basin".parse().unwrap(),
                    stream: None,
                }),
                Ok(S2BasinUri("valid-basin".parse().unwrap())),
                Err(S2UriParseError::MissingStreamName),
                Ok(S2BasinAndMaybeStreamUri {
                    basin: "valid-basin".parse().unwrap(),
                    stream: None,
                }),
            ),
            (
                "s2://valid-basin/stream/name",
                Ok(S2Uri {
                    basin: "valid-basin".parse().unwrap(),
                    stream: Some("stream/name".to_owned()),
                }),
                Err(S2UriParseError::UnexpectedStreamName),
                Ok(S2BasinAndStreamUri {
                    basin: "valid-basin".parse().unwrap(),
                    stream: "stream/name".parse().unwrap(),
                }),
                Ok(S2BasinAndMaybeStreamUri {
                    basin: "valid-basin".parse().unwrap(),
                    stream: Some("stream/name".parse().unwrap()),
                }),
            ),
            (
                "-invalid-basin",
                Err(S2UriParseError::MissingUriScheme),
                Err(S2UriParseError::InvalidBasinName("".to_owned())),
                Err(S2UriParseError::MissingUriScheme),
                Err(S2UriParseError::InvalidBasinName("".to_owned())),
            ),
            (
                "http://valid-basin",
                Err(S2UriParseError::InvalidUriScheme("http".to_owned())),
                Err(S2UriParseError::InvalidUriScheme("http".to_owned())),
                Err(S2UriParseError::InvalidUriScheme("http".to_owned())),
                Err(S2UriParseError::InvalidUriScheme("http".to_owned())),
            ),
            (
                "s2://-invalid-basin",
                Err(S2UriParseError::InvalidBasinName("".to_owned())),
                Err(S2UriParseError::InvalidBasinName("".to_owned())),
                Err(S2UriParseError::InvalidBasinName("".to_owned())),
                Err(S2UriParseError::InvalidBasinName("".to_owned())),
            ),
            (
                "s2:///stream/name",
                Err(S2UriParseError::InvalidBasinName("".to_owned())),
                Err(S2UriParseError::InvalidBasinName("".to_owned())),
                Err(S2UriParseError::InvalidBasinName("".to_owned())),
                Err(S2UriParseError::InvalidBasinName("".to_owned())),
            ),
            (
                "random:::string",
                Err(S2UriParseError::MissingUriScheme),
                Err(S2UriParseError::InvalidBasinName("".to_owned())),
                Err(S2UriParseError::MissingUriScheme),
                Err(S2UriParseError::InvalidBasinName("".to_owned())),
            ),
        ];

        for (
            s,
            expected_uri,
            expected_basin_uri,
            expected_basin_and_stream_uri,
            expected_basin_and_maybe_stream_uri,
        ) in test_cases
        {
            assert_eq!(s.parse(), expected_uri, "S2Uri: {s}");
            assert_eq!(s.parse(), expected_basin_uri, "S2BasinUri: {s}");
            assert_eq!(
                s.parse(),
                expected_basin_and_stream_uri,
                "S2BasinAndStreamUri: {s}"
            );
            assert_eq!(
                s.parse(),
                expected_basin_and_maybe_stream_uri,
                "S2BasinAndMaybeStreamUri: {s}"
            );
        }
    }
}