ticksupply 0.2.1

//! subscriptions — Manage ongoing data stream subscriptions.
//!
//! # Examples
//!
//! ```no_run
//! # async fn example() -> ticksupply::Result<()> {
//! let client = ticksupply::Client::new()?;
//! let sub = client.subscriptions().create(12345).send().await?;
//! client.subscriptions().pause(&sub.id).send().await?;
//! client.subscriptions().resume(&sub.id).send().await?;
//! client.subscriptions().delete(&sub.id).send().await?;
//! # Ok(()) }
//! ```

use futures::Stream;
use serde::{Deserialize, Serialize};

use crate::client::Client;
use crate::error::Result;
use crate::http::{send, send_empty, RequestOpts};
use crate::pagination::Page;
use crate::resources::catalog::DatastreamInfo;
use crate::timestamp::Timestamp;

/// Lifecycle state of a subscription.
#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
#[serde(rename_all = "snake_case")]
#[non_exhaustive]
pub enum SubscriptionStatus {
    /// Actively collecting data.
    Active,
    /// Paused by the user; no new data.
    Paused,
    /// Soft-deleted; scheduled for purge.
    Deleted,
    /// Forward-compat: unknown server-reported status.
    #[serde(other)]
    Unknown,
}

/// A data stream subscription (row shape returned by list endpoints).
#[derive(Debug, Clone, Deserialize)]
pub struct Subscription {
    /// Unique subscription identifier (`sub_…`).
    pub id: String,
    /// Datastream this subscription is bound to.
    pub datastream_id: i64,
    /// Current lifecycle status.
    pub status: SubscriptionStatus,
    /// Creation timestamp (RFC 3339 / ISO 8601).
    pub created_at: Timestamp,
    /// Optional expanded datastream info (populated only when the endpoint
    /// includes it).
    #[serde(default)]
    pub datastream: Option<DatastreamInfo>,
}

/// Full subscription detail returned by [`CreateSubscriptionRequest::send`] —
/// includes datastream info and activity spans.
///
/// The `spans` field is always empty in the create response; call
/// [`SubscriptionsResource::list_spans`] to retrieve the initial active span.
#[derive(Debug, Clone, Deserialize)]
pub struct SubscriptionResponse {
    /// Unique subscription identifier (`sub_…`).
    pub id: String,
    /// Current lifecycle status.
    pub status: SubscriptionStatus,
    /// Datastream this subscription is bound to.
    pub datastream: DatastreamInfo,
    /// Creation timestamp (RFC 3339 / ISO 8601).
    pub created_at: Timestamp,
    /// Activity spans associated with this subscription.
    #[serde(default)]
    pub spans: Vec<SubscriptionSpan>,
}

/// A single activity span for a subscription — a contiguous window during
/// which the subscription was active.
#[derive(Debug, Clone, Deserialize)]
pub struct SubscriptionSpan {
    /// Span identifier (`spn_…`).
    pub id: String,
    /// Start of the span (RFC 3339 / ISO 8601).
    pub started_at: Timestamp,
    /// End of the span (RFC 3339 / ISO 8601); `None` if the span is ongoing.
    #[serde(default)]
    pub ended_at: Option<Timestamp>,
}

/// Accessor for the `/subscriptions` endpoints.
pub struct SubscriptionsResource<'a> {
    pub(crate) client: &'a Client,
}

impl<'a> SubscriptionsResource<'a> {
    /// Retrieves a subscription by its identifier.
    ///
    /// Returns the flat [`Subscription`] shape — the same type yielded by
    /// [`ListSubscriptionsRequest::send`]. To retrieve activity spans, call
    /// [`SubscriptionsResource::list_spans`].
    ///
    /// # Errors
    ///
    /// - [`crate::Error::NotFound`] if no subscription has this ID.
    /// - [`crate::Error::Authentication`] on invalid credentials.
    /// - [`crate::Error::Network`] on transport failure.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # async fn example() -> ticksupply::Result<()> {
    /// let client = ticksupply::Client::new()?;
    /// let sub = client.subscriptions().get("sub_abc").await?;
    /// println!("{}: {:?}", sub.id, sub.status);
    /// # Ok(()) }
    /// ```
    pub async fn get(&self, id: &str) -> Result<Subscription> {
        let path = format!("/subscriptions/{id}");
        send::<_, ()>(
            self.client,
            reqwest::Method::GET,
            &path,
            None,
            None,
            RequestOpts::default(),
        )
        .await
    }

    /// Returns a builder for pausing a subscription.
    ///
    /// Idempotent server-side; this is a no-op if the subscription is
    /// already paused. Supply an `idempotency_key` to enable transport-level
    /// retries on transient failures.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # async fn example() -> ticksupply::Result<()> {
    /// let client = ticksupply::Client::new()?;
    /// client.subscriptions().pause("sub_abc").send().await?;
    /// # Ok(()) }
    /// ```
    pub fn pause(&self, id: impl Into<String>) -> PauseSubscriptionRequest<'a> {
        PauseSubscriptionRequest {
            client: self.client,
            id: id.into(),
            idempotency_key: None,
        }
    }

    /// Returns a builder for resuming a paused subscription.
    ///
    /// Idempotent server-side; this is a no-op if the subscription is
    /// already active. Supply an `idempotency_key` to enable transport-level
    /// retries on transient failures.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # async fn example() -> ticksupply::Result<()> {
    /// let client = ticksupply::Client::new()?;
    /// client.subscriptions().resume("sub_abc").send().await?;
    /// # Ok(()) }
    /// ```
    pub fn resume(&self, id: impl Into<String>) -> ResumeSubscriptionRequest<'a> {
        ResumeSubscriptionRequest {
            client: self.client,
            id: id.into(),
            idempotency_key: None,
        }
    }

    /// Returns a builder for soft-deleting a subscription.
    ///
    /// Supply an `idempotency_key` to enable transport-level retries on
    /// transient failures.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # async fn example() -> ticksupply::Result<()> {
    /// let client = ticksupply::Client::new()?;
    /// client.subscriptions().delete("sub_abc").send().await?;
    /// # Ok(()) }
    /// ```
    pub fn delete(&self, id: impl Into<String>) -> DeleteSubscriptionRequest<'a> {
        DeleteSubscriptionRequest {
            client: self.client,
            id: id.into(),
            idempotency_key: None,
        }
    }

    /// Lists activity spans for a subscription.
    ///
    /// Each span is a contiguous window during which the subscription was
    /// actively collecting data.
    ///
    /// # Errors
    ///
    /// - [`crate::Error::NotFound`] if no subscription has this ID.
    /// - [`crate::Error::Authentication`] on invalid credentials.
    /// - [`crate::Error::Network`] on transport failure.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # async fn example() -> ticksupply::Result<()> {
    /// let client = ticksupply::Client::new()?;
    /// let spans = client.subscriptions().list_spans("sub_abc").await?;
    /// for s in spans {
    ///     println!("{}: {} -> {:?}", s.id, s.started_at, s.ended_at);
    /// }
    /// # Ok(()) }
    /// ```
    pub async fn list_spans(&self, id: &str) -> Result<Vec<SubscriptionSpan>> {
        let path = format!("/subscriptions/{id}/spans");
        send::<_, ()>(
            self.client,
            reqwest::Method::GET,
            &path,
            None,
            None,
            RequestOpts::default(),
        )
        .await
    }

    /// Returns a builder for creating a new subscription.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # async fn example() -> ticksupply::Result<()> {
    /// let client = ticksupply::Client::new()?;
    /// let sub = client.subscriptions()
    ///     .create(12345)
    ///     .idempotency_key("550e8400-e29b-41d4-a716-446655440000")
    ///     .send().await?;
    /// # let _ = sub;
    /// # Ok(()) }
    /// ```
    pub fn create(&self, datastream_id: i64) -> CreateSubscriptionRequest<'a> {
        CreateSubscriptionRequest {
            client: self.client,
            datastream_id,
            idempotency_key: None,
        }
    }

    /// Returns a builder for listing subscriptions.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # async fn example() -> ticksupply::Result<()> {
    /// let client = ticksupply::Client::new()?;
    /// let page = client.subscriptions().list().limit(25).send().await?;
    /// # let _ = page;
    /// # Ok(()) }
    /// ```
    pub fn list(&self) -> ListSubscriptionsRequest<'a> {
        ListSubscriptionsRequest {
            client: self.client,
            limit: None,
            page_token: None,
        }
    }
}

/// Builder for `POST /v1/subscriptions`.
pub struct CreateSubscriptionRequest<'a> {
    client: &'a Client,
    datastream_id: i64,
    idempotency_key: Option<String>,
}

impl<'a> CreateSubscriptionRequest<'a> {
    /// Sets the idempotency key for this request.
    ///
    /// Required for the underlying retry loop to safely replay this mutation
    /// on transient failures. Accepts any UUID format, up to 128 chars.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # async fn example() -> ticksupply::Result<()> {
    /// let client = ticksupply::Client::new()?;
    /// let sub = client.subscriptions()
    ///     .create(12345)
    ///     .idempotency_key("550e8400-e29b-41d4-a716-446655440000")
    ///     .send().await?;
    /// # let _ = sub;
    /// # Ok(()) }
    /// ```
    pub fn idempotency_key(mut self, k: impl Into<String>) -> Self {
        self.idempotency_key = Some(k.into());
        self
    }

    /// Sends the create request and returns the resulting subscription.
    ///
    /// # Errors
    ///
    /// - [`crate::Error::Validation`] if the request is malformed or the
    ///   datastream does not exist.
    /// - [`crate::Error::Authentication`] on invalid credentials.
    /// - [`crate::Error::RateLimited`] if the account is over its rate limit.
    /// - [`crate::Error::Network`] on transport failure.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # async fn example() -> ticksupply::Result<()> {
    /// let client = ticksupply::Client::new()?;
    /// let sub = client.subscriptions().create(12345).send().await?;
    /// # let _ = sub;
    /// # Ok(()) }
    /// ```
    pub async fn send(self) -> Result<SubscriptionResponse> {
        #[derive(Serialize)]
        struct Body {
            datastream_id: i64,
        }
        let body = Body {
            datastream_id: self.datastream_id,
        };
        send(
            self.client,
            reqwest::Method::POST,
            "/subscriptions",
            None,
            Some(&body),
            RequestOpts {
                idempotency_key: self.idempotency_key.as_deref(),
            },
        )
        .await
    }
}

/// Builder for `GET /v1/subscriptions`.
pub struct ListSubscriptionsRequest<'a> {
    client: &'a Client,
    limit: Option<u32>,
    page_token: Option<String>,
}

impl<'a> ListSubscriptionsRequest<'a> {
    /// Sets the page size (default 50, max 100).
    pub fn limit(mut self, n: u32) -> Self {
        self.limit = Some(n);
        self
    }
    /// Sets the page token returned by a prior response.
    pub fn page_token(mut self, t: impl Into<String>) -> Self {
        self.page_token = Some(t.into());
        self
    }

    fn query(&self) -> Vec<(&'static str, String)> {
        let mut q = Vec::new();
        if let Some(n) = self.limit {
            q.push(("limit", n.to_string()));
        }
        if let Some(t) = &self.page_token {
            q.push(("page_token", t.clone()));
        }
        q
    }

    /// Fetches a single page of results.
    ///
    /// # Errors
    ///
    /// - [`crate::Error::Authentication`] on invalid credentials.
    /// - [`crate::Error::Validation`] if `limit` is out of range.
    /// - [`crate::Error::Network`] on transport failure.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # async fn example() -> ticksupply::Result<()> {
    /// let client = ticksupply::Client::new()?;
    /// let page = client.subscriptions().list().limit(25).send().await?;
    /// println!("{} of {:?}", page.items.len(), page.total);
    /// # Ok(()) }
    /// ```
    pub async fn send(self) -> Result<Page<Subscription>> {
        let q = self.query();
        send::<_, ()>(
            self.client,
            reqwest::Method::GET,
            "/subscriptions",
            Some(q.as_slice()),
            None,
            RequestOpts::default(),
        )
        .await
    }

    /// Auto-paginates across all pages, yielding each subscription.
    ///
    /// Each yielded item is a [`Result`] that surfaces the same errors as
    /// [`Self::send`] if a page fetch fails.
    ///
    /// Streaming always starts from the first page; any `page_token`
    /// previously set on the builder is ignored. `limit` is preserved
    /// across pages.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # async fn example() -> ticksupply::Result<()> {
    /// use futures::StreamExt;
    /// let client = ticksupply::Client::new()?;
    /// let mut stream = Box::pin(client.subscriptions().list().stream());
    /// while let Some(sub) = stream.next().await {
    ///     let sub = sub?;
    ///     println!("{}: {:?}", sub.id, sub.status);
    /// }
    /// # Ok(()) }
    /// ```
    pub fn stream(self) -> impl Stream<Item = Result<Subscription>> + 'a {
        let Self { client, limit, .. } = self;
        async_stream::try_stream! {
            let mut page_token: Option<String> = None;
            loop {
                let req = ListSubscriptionsRequest {
                    client,
                    limit,
                    page_token: page_token.clone(),
                };
                let page = req.send().await?;
                for item in page.items { yield item; }
                match page.next_page_token {
                    Some(t) => page_token = Some(t),
                    None => break,
                }
            }
        }
    }
}

/// Builder for `POST /v1/subscriptions/{id}/pause`.
pub struct PauseSubscriptionRequest<'a> {
    client: &'a Client,
    id: String,
    idempotency_key: Option<String>,
}

impl<'a> PauseSubscriptionRequest<'a> {
    /// Sets the idempotency key for this request (any UUID format, ≤128 chars).
    pub fn idempotency_key(mut self, k: impl Into<String>) -> Self {
        self.idempotency_key = Some(k.into());
        self
    }

    /// Sends the pause request.
    ///
    /// # Errors
    ///
    /// - [`crate::Error::NotFound`] if no subscription has this ID.
    /// - [`crate::Error::Validation`] if the subscription is deleted.
    /// - [`crate::Error::Network`] on transport failure.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # async fn example() -> ticksupply::Result<()> {
    /// let client = ticksupply::Client::new()?;
    /// client.subscriptions().pause("sub_abc").send().await?;
    /// # Ok(()) }
    /// ```
    pub async fn send(self) -> Result<()> {
        let path = format!("/subscriptions/{}/pause", self.id);
        send_empty::<()>(
            self.client,
            reqwest::Method::POST,
            &path,
            None,
            None,
            RequestOpts {
                idempotency_key: self.idempotency_key.as_deref(),
            },
        )
        .await
    }
}

/// Builder for `POST /v1/subscriptions/{id}/resume`.
pub struct ResumeSubscriptionRequest<'a> {
    client: &'a Client,
    id: String,
    idempotency_key: Option<String>,
}

impl<'a> ResumeSubscriptionRequest<'a> {
    /// Sets the idempotency key for this request (any UUID format, ≤128 chars).
    pub fn idempotency_key(mut self, k: impl Into<String>) -> Self {
        self.idempotency_key = Some(k.into());
        self
    }

    /// Sends the resume request.
    ///
    /// # Errors
    ///
    /// - [`crate::Error::NotFound`] if no subscription has this ID.
    /// - [`crate::Error::Validation`] if the subscription is deleted.
    /// - [`crate::Error::Network`] on transport failure.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # async fn example() -> ticksupply::Result<()> {
    /// let client = ticksupply::Client::new()?;
    /// client.subscriptions().resume("sub_abc").send().await?;
    /// # Ok(()) }
    /// ```
    pub async fn send(self) -> Result<()> {
        let path = format!("/subscriptions/{}/resume", self.id);
        send_empty::<()>(
            self.client,
            reqwest::Method::POST,
            &path,
            None,
            None,
            RequestOpts {
                idempotency_key: self.idempotency_key.as_deref(),
            },
        )
        .await
    }
}

/// Builder for `DELETE /v1/subscriptions/{id}`.
pub struct DeleteSubscriptionRequest<'a> {
    client: &'a Client,
    id: String,
    idempotency_key: Option<String>,
}

impl<'a> DeleteSubscriptionRequest<'a> {
    /// Sets the idempotency key for this request (any UUID format, ≤128 chars).
    pub fn idempotency_key(mut self, k: impl Into<String>) -> Self {
        self.idempotency_key = Some(k.into());
        self
    }

    /// Sends the delete request.
    ///
    /// # Errors
    ///
    /// - [`crate::Error::NotFound`] if no subscription has this ID.
    /// - [`crate::Error::Authentication`] on invalid credentials.
    /// - [`crate::Error::Network`] on transport failure.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # async fn example() -> ticksupply::Result<()> {
    /// let client = ticksupply::Client::new()?;
    /// client.subscriptions().delete("sub_abc").send().await?;
    /// # Ok(()) }
    /// ```
    pub async fn send(self) -> Result<()> {
        let path = format!("/subscriptions/{}", self.id);
        send_empty::<()>(
            self.client,
            reqwest::Method::DELETE,
            &path,
            None,
            None,
            RequestOpts {
                idempotency_key: self.idempotency_key.as_deref(),
            },
        )
        .await
    }
}