ticksupply 0.1.0

//! exports — Create historical data exports and download the resulting artifacts.
//!
//! # Examples
//!
//! ```no_run
//! # async fn example() -> ticksupply::Result<()> {
//! use chrono::{Utc, Duration};
//! let client = ticksupply::Client::new()?;
//! let job = client.exports()
//!     .create(12345, Utc::now() - Duration::hours(1), Utc::now())
//!     .idempotency_key("550e8400-e29b-41d4-a716-446655440000")
//!     .send().await?;
//! let download = client.exports().get_download(&job.id).await?;
//! for art in download.artifacts {
//!     println!("{} -> {}", art.filename, art.url);
//! }
//! # 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::timestamp::{IntoTimestamp, Nanos, Timestamp};

/// Lifecycle state of an export job.
#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
#[serde(rename_all = "snake_case")]
#[non_exhaustive]
pub enum ExportStatus {
    /// Queued, not yet running.
    Queued,
    /// Actively running.
    Running,
    /// Completed successfully.
    Succeeded,
    /// Failed.
    Failed,
    /// Canceled by the user.
    Canceled,
    /// Deleted.
    Deleted,
    /// Forward-compat: unknown status from the server.
    #[serde(other)]
    Unknown,
}

/// An export job.
#[derive(Debug, Clone, Deserialize)]
pub struct ExportJob {
    /// Unique export identifier (`exp_…`).
    pub id: String,
    /// Datastream being exported.
    pub datastream_id: i64,
    /// Inclusive start of the export window.
    pub start_time: Nanos,
    /// Exclusive end of the export window.
    pub end_time: Nanos,
    /// Output file format (e.g. `"csv"`, `"parquet"`, `"jsonl"`).
    pub format: String,
    /// Current lifecycle state.
    pub status: ExportStatus,
    /// Server-reported failure reason, when `status` is [`ExportStatus::Failed`].
    #[serde(default)]
    pub reason: Option<String>,
    /// When the job was created (RFC 3339 / ISO 8601).
    pub created_at: Timestamp,
    /// When processing started (RFC 3339 / ISO 8601), when known.
    #[serde(default)]
    pub started_at: Option<Timestamp>,
    /// When the job finished (RFC 3339 / ISO 8601), when applicable.
    #[serde(default)]
    pub finished_at: Option<Timestamp>,
}

/// A downloadable artifact from a completed export.
#[derive(Debug, Clone, Deserialize)]
pub struct ArtifactDownload {
    /// Artifact identifier (`art_…`).
    pub id: String,
    /// Pre-signed URL for downloading the artifact. Typically valid for a
    /// short window (a few minutes).
    pub url: String,
    /// Byte size.
    pub bytes: u64,
    /// Suggested filename, without directory.
    pub filename: String,
}

/// Download response for a succeeded export.
#[derive(Debug, Clone, Deserialize)]
pub struct ExportDownloadResponse {
    /// List of artifact files.
    pub artifacts: Vec<ArtifactDownload>,
    /// Number of artifacts (`artifacts.len()`).
    pub count: u32,
    /// Sum of artifact sizes in bytes.
    pub total_bytes: u64,
}

impl ExportDownloadResponse {
    /// Returns the sole artifact when this is a single-file export, or `None`
    /// for multi-file exports (in which case iterate [`Self::artifacts`]).
    ///
    /// Equivalent to Python's `ExportDownloadResponse.is_single_file` /
    /// `.url` pair: `resp.single().is_some()` replaces `is_single_file`,
    /// and `resp.single().map(|a| a.url.as_str())` replaces `url`.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # async fn example() -> ticksupply::Result<()> {
    /// let client = ticksupply::Client::new()?;
    /// let download = client.exports().get_download("exp_abc").await?;
    /// if let Some(artifact) = download.single() {
    ///     println!("single file: {} ({} bytes)", artifact.url, artifact.bytes);
    /// } else {
    ///     for artifact in &download.artifacts {
    ///         println!("{}: {}", artifact.filename, artifact.url);
    ///     }
    /// }
    /// # Ok(()) }
    /// ```
    pub fn single(&self) -> Option<&ArtifactDownload> {
        match self.artifacts.as_slice() {
            [only] => Some(only),
            _ => None,
        }
    }
}

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

impl<'a> ExportsResource<'a> {
    /// Retrieves an export job by its identifier.
    ///
    /// # Errors
    ///
    /// - [`crate::Error::NotFound`] if no export 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 job = client.exports().get("exp_abc").await?;
    /// println!("{}: {:?}", job.id, job.status);
    /// # Ok(()) }
    /// ```
    pub async fn get(&self, id: &str) -> Result<ExportJob> {
        let path = format!("/exports/{id}");
        send::<_, ()>(
            self.client,
            reqwest::Method::GET,
            &path,
            None,
            None,
            RequestOpts::default(),
        )
        .await
    }

    /// Returns a builder for canceling a queued or running export job.
    ///
    /// Idempotent when the job is already canceled; returns a validation
    /// error when the job has reached any other terminal state. 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.exports().cancel("exp_abc").send().await?;
    /// # Ok(()) }
    /// ```
    pub fn cancel(&self, id: impl Into<String>) -> CancelExportRequest<'a> {
        CancelExportRequest {
            client: self.client,
            id: id.into(),
            idempotency_key: None,
        }
    }

    /// Returns a builder for deleting an export job record.
    ///
    /// 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.exports().delete("exp_abc").send().await?;
    /// # Ok(()) }
    /// ```
    pub fn delete(&self, id: impl Into<String>) -> DeleteExportRequest<'a> {
        DeleteExportRequest {
            client: self.client,
            id: id.into(),
            idempotency_key: None,
        }
    }

    /// Retrieves the download response (artifact URLs) for a succeeded export.
    ///
    /// Artifact URLs are short-lived pre-signed URLs — typically valid for a
    /// few minutes. Re-call this method to get fresh URLs.
    ///
    /// # Errors
    ///
    /// - [`crate::Error::NotFound`] if no export has this ID.
    /// - [`crate::Error::Validation`] if the export has not succeeded.
    /// - [`crate::Error::PaymentRequired`] if billing denies access.
    /// - [`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 download = client.exports().get_download("exp_abc").await?;
    /// for art in download.artifacts {
    ///     println!("{} -> {}", art.filename, art.url);
    /// }
    /// # Ok(()) }
    /// ```
    pub async fn get_download(&self, id: &str) -> Result<ExportDownloadResponse> {
        let path = format!("/exports/{id}/download");
        send::<_, ()>(
            self.client,
            reqwest::Method::GET,
            &path,
            None,
            None,
            RequestOpts::default(),
        )
        .await
    }

    /// Returns a builder for creating a new export job.
    ///
    /// The caller must already have an active or paused subscription to
    /// `datastream_id`; the export only covers periods when the subscription
    /// was active.
    ///
    /// # Arguments
    ///
    /// * `datastream_id` — datastream to export.
    /// * `start_time` — inclusive start of the export window.
    /// * `end_time` — exclusive end of the export window.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # async fn example() -> ticksupply::Result<()> {
    /// let client = ticksupply::Client::new()?;
    /// let start = 1_704_067_200_000_000_000_i64;
    /// let end   = 1_704_070_800_000_000_000_i64;
    /// let job = client.exports()
    ///     .create(12345, start, end)
    ///     .idempotency_key("550e8400-e29b-41d4-a716-446655440000")
    ///     .send().await?;
    /// # let _ = job;
    /// # Ok(()) }
    /// ```
    pub fn create(
        &self,
        datastream_id: i64,
        start_time: impl IntoTimestamp,
        end_time: impl IntoTimestamp,
    ) -> CreateExportRequest<'a> {
        CreateExportRequest {
            client: self.client,
            datastream_id,
            start_time: start_time.into_timestamp_string(),
            end_time: end_time.into_timestamp_string(),
            idempotency_key: None,
            schema: None,
        }
    }

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

/// Builder for `POST /v1/exports`.
pub struct CreateExportRequest<'a> {
    client: &'a Client,
    datastream_id: i64,
    start_time: String,
    end_time: String,
    idempotency_key: Option<String>,
    schema: Option<SchemaArg>,
}

enum SchemaArg {
    Ref(String),
    Inline(serde_json::Value),
}

impl<'a> CreateExportRequest<'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
    }

    /// Sets an optional schema reference.
    ///
    /// Accepts a built-in name (`"raw"`, `"normalized"`) or a stored schema
    /// id (`"sch_…"`). Defaults server-side to `"raw"` when omitted. Replaces
    /// any prior [`Self::schema`] or [`Self::inline_schema`] call on this
    /// builder.
    pub fn schema(mut self, s: impl Into<String>) -> Self {
        self.schema = Some(SchemaArg::Ref(s.into()));
        self
    }

    /// Sets an inline, ad-hoc column schema.
    ///
    /// Use this to prototype a custom export without creating a stored
    /// schema first. The value is serialized as-is under the `schema` key
    /// and must match the API's inline shape — at minimum a `columns`
    /// array, optionally with `unfold` and `derive` maps (see
    /// [the API reference](https://docs.ticksupply.com/api/exports/create)).
    /// Replaces any prior [`Self::schema`] or [`Self::inline_schema`]
    /// call on this builder.
    ///
    /// For stored or built-in schemas, prefer [`Self::schema`].
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # async fn example() -> ticksupply::Result<()> {
    /// use serde_json::json;
    /// let client = ticksupply::Client::new()?;
    /// let job = client.exports()
    ///     .create(12345, 1_704_067_200_000_000_000_i64, 1_704_070_800_000_000_000_i64)
    ///     .inline_schema(json!({
    ///         "columns": [
    ///             {"name": "ts", "source": {"kind": "collection_timestamp_ns"}}
    ///         ]
    ///     }))
    ///     .send().await?;
    /// # let _ = job;
    /// # Ok(()) }
    /// ```
    pub fn inline_schema(mut self, v: serde_json::Value) -> Self {
        self.schema = Some(SchemaArg::Inline(v));
        self
    }

    /// Sends the create request.
    ///
    /// The server responds with `202 Accepted`; the returned [`ExportJob`]
    /// reflects the initial queued state. Poll [`ExportsResource::get`] to
    /// watch its status transition.
    ///
    /// # Errors
    ///
    /// - [`crate::Error::Validation`] if the datastream identifier is invalid
    ///   or the time window is rejected.
    /// - [`crate::Error::PermissionDenied`] if the caller has no active or
    ///   paused subscription for the datastream.
    /// - [`crate::Error::PaymentRequired`] if billing denies the request.
    /// - [`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 job = client.exports()
    ///     .create(12345, 1_704_067_200_000_000_000_i64, 1_704_070_800_000_000_000_i64)
    ///     .idempotency_key("550e8400-e29b-41d4-a716-446655440000")
    ///     .send().await?;
    /// # let _ = job;
    /// # Ok(()) }
    /// ```
    pub async fn send(self) -> Result<ExportJob> {
        #[derive(Serialize)]
        #[serde(untagged)]
        enum SchemaBody<'b> {
            Ref(&'b str),
            Inline(&'b serde_json::Value),
        }
        #[derive(Serialize)]
        struct Body<'b> {
            datastream_id: i64,
            start_time: &'b str,
            end_time: &'b str,
            #[serde(skip_serializing_if = "Option::is_none")]
            schema: Option<SchemaBody<'b>>,
        }
        let schema = self.schema.as_ref().map(|s| match s {
            SchemaArg::Ref(r) => SchemaBody::Ref(r),
            SchemaArg::Inline(v) => SchemaBody::Inline(v),
        });
        let body = Body {
            datastream_id: self.datastream_id,
            start_time: &self.start_time,
            end_time: &self.end_time,
            schema,
        };
        send(
            self.client,
            reqwest::Method::POST,
            "/exports",
            None,
            Some(&body),
            RequestOpts {
                idempotency_key: self.idempotency_key.as_deref(),
            },
        )
        .await
    }
}

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

impl<'a> ListExportsRequest<'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.exports().list().limit(25).send().await?;
    /// # let _ = page;
    /// # Ok(()) }
    /// ```
    pub async fn send(self) -> Result<Page<ExportJob>> {
        let q = self.query();
        send::<_, ()>(
            self.client,
            reqwest::Method::GET,
            "/exports",
            Some(q.as_slice()),
            None,
            RequestOpts::default(),
        )
        .await
    }

    /// Auto-paginates across all pages, yielding each export job.
    ///
    /// 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.exports().list().stream());
    /// while let Some(job) = stream.next().await {
    ///     let job = job?;
    ///     println!("{}: {:?}", job.id, job.status);
    /// }
    /// # Ok(()) }
    /// ```
    pub fn stream(self) -> impl Stream<Item = Result<ExportJob>> + 'a {
        let Self { client, limit, .. } = self;
        async_stream::try_stream! {
            let mut page_token: Option<String> = None;
            loop {
                let req = ListExportsRequest {
                    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/exports/{id}/cancel`.
pub struct CancelExportRequest<'a> {
    client: &'a Client,
    id: String,
    idempotency_key: Option<String>,
}

impl<'a> CancelExportRequest<'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 cancel request.
    ///
    /// # Errors
    ///
    /// - [`crate::Error::NotFound`] if no export has this ID.
    /// - [`crate::Error::Validation`] if the job has already succeeded, failed,
    ///   or been deleted.
    /// - [`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.exports().cancel("exp_abc").send().await?;
    /// # Ok(()) }
    /// ```
    pub async fn send(self) -> Result<()> {
        let path = format!("/exports/{}/cancel", 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/exports/{id}`.
pub struct DeleteExportRequest<'a> {
    client: &'a Client,
    id: String,
    idempotency_key: Option<String>,
}

impl<'a> DeleteExportRequest<'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 export 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.exports().delete("exp_abc").send().await?;
    /// # Ok(()) }
    /// ```
    pub async fn send(self) -> Result<()> {
        let path = format!("/exports/{}", self.id);
        send_empty::<()>(
            self.client,
            reqwest::Method::DELETE,
            &path,
            None,
            None,
            RequestOpts {
                idempotency_key: self.idempotency_key.as_deref(),
            },
        )
        .await
    }
}