duners 0.1.0

//! Dune API client implementation.
//!
//! This module provides [`DuneClient`] for calling the [Dune Analytics API](https://dune.com/docs/api/).

use crate::error::{DuneError, DuneRequestError};
use crate::parameters::Parameter;
use crate::response::{
    CancellationResponse, CreateTableRequest, CreateTableResponse, DuneQuery, ExecutionResponse,
    ExecutionStatus, GetResultResponse, GetStatusResponse, InsertTableResponse, QueryBody,
    QueryResponse, SuccessResponse, UploadCsvRequest,
};
use dotenvy::dotenv;
use log::{debug, error, info, warn};
use reqwest::{Error, Response};
use serde::de::DeserializeOwned;
use serde_json::json;
use std::collections::HashMap;
use std::env;
use tokio::time::{sleep, Duration};

/// Base URL for the Dune API (v1).
const BASE_URL: &str = "https://api.dune.com/api/v1";

/// Client for the [Dune Analytics API](https://dune.com/docs/api/).
///
/// Create a client with [`DuneClient::new`] (pass the API key directly) or [`DuneClient::from_env`]
/// (reads `DUNE_API_KEY` from the environment, including from a `.env` file if present).
///
/// ## High-level usage
///
/// Use **[`refresh`](DuneClient::refresh)** to execute a query, wait until it finishes, and get
/// the result rows in one call. This is the easiest way to run a query.
///
/// ## Low-level usage
///
/// For more control (e.g. polling yourself or cancelling), use:
/// - **[`execute_query`](DuneClient::execute_query)** — Start a query, get an `execution_id`.
/// - **[`get_status`](DuneClient::get_status)** — Check whether the execution is still running.
/// - **[`get_results`](DuneClient::get_results)** — Fetch the result rows (only valid when complete).
/// - **[`cancel_execution`](DuneClient::cancel_execution)** — Cancel a running execution.
pub struct DuneClient {
    /// API key used for request authentication.
    api_key: String,
}

impl DuneClient {
    /// Creates a client with the given API key.
    ///
    /// Get your API key from [Dune → Settings → API](https://dune.com/settings/api).
    pub fn new(api_key: &str) -> DuneClient {
        DuneClient {
            api_key: api_key.to_string(),
        }
    }

    /// Creates a client using the `DUNE_API_KEY` environment variable.
    ///
    /// Loads `.env` from the current directory if present (via the `dotenvy` crate).
    /// Panics if `DUNE_API_KEY` is not set.
    pub fn from_env() -> DuneClient {
        dotenv().ok();
        DuneClient {
            api_key: env::var("DUNE_API_KEY").unwrap(),
        }
    }

    /// Internal POST request handler
    async fn _post(&self, route: &str, params: Option<Vec<Parameter>>) -> Result<Response, Error> {
        let params = params
            .unwrap_or_default()
            .into_iter()
            .map(|p| (p.key, p.value))
            .collect::<HashMap<_, _>>();
        let request_url = format!("{BASE_URL}/{route}");
        debug!("POST to {} with parameters {:?}", route, &params);
        let client = reqwest::Client::new();
        client
            .post(&request_url)
            .header("x-dune-api-key", &self.api_key)
            .json(&json!({ "query_parameters": params }))
            .send()
            .await
    }

    /// Internal POST request handler with arbitrary JSON body
    async fn _post_json(&self, route: &str, body: serde_json::Value) -> Result<Response, Error> {
        let request_url = format!("{BASE_URL}/{route}");
        debug!("POST to {} with body {:?}", route, &body);
        let client = reqwest::Client::new();
        client
            .post(&request_url)
            .header("x-dune-api-key", &self.api_key)
            .json(&body)
            .send()
            .await
    }

    /// Internal PATCH request handler with JSON body
    async fn _patch(&self, route: &str, body: serde_json::Value) -> Result<Response, Error> {
        let request_url = format!("{BASE_URL}/{route}");
        debug!("PATCH to {} with body {:?}", route, &body);
        let client = reqwest::Client::new();
        client
            .patch(&request_url)
            .header("x-dune-api-key", &self.api_key)
            .json(&body)
            .send()
            .await
    }

    /// Internal GET request handler for arbitrary routes
    async fn _get_url(&self, route: &str) -> Result<Response, Error> {
        let request_url = format!("{BASE_URL}/{route}");
        debug!("GET from {}", &request_url);
        let client = reqwest::Client::new();
        client
            .get(&request_url)
            .header("x-dune-api-key", &self.api_key)
            .send()
            .await
    }

    /// Internal GET request handler for execution endpoints
    async fn _get(&self, job_id: &str, command: &str) -> Result<Response, Error> {
        self._get_url(&format!("execution/{job_id}/{command}"))
            .await
    }

    /// Deserializes Responses into appropriate type.
    /// Some "invalid" requests return response JSON, which are parsed and returned as Errors.
    async fn _parse_response<T: DeserializeOwned>(resp: Response) -> Result<T, DuneRequestError> {
        if resp.status().is_success() {
            resp.json::<T>().await.map_err(DuneRequestError::from)
        } else {
            let err = resp
                .json::<DuneError>()
                .await
                .map_err(DuneRequestError::from)?;
            error!("request error {:?}", &err);
            Err(DuneRequestError::from(err))
        }
    }

    /// Internal DELETE request handler
    async fn _delete(&self, route: &str) -> Result<Response, Error> {
        let request_url = format!("{BASE_URL}/{route}");
        debug!("DELETE {}", &request_url);
        let client = reqwest::Client::new();
        client
            .delete(&request_url)
            .header("x-dune-api-key", &self.api_key)
            .send()
            .await
    }

    /// Internal POST request handler with raw body and custom content type
    async fn _post_raw(
        &self,
        route: &str,
        content_type: &str,
        body: String,
    ) -> Result<Response, Error> {
        let request_url = format!("{BASE_URL}/{route}");
        debug!("POST raw to {} ({} bytes)", route, body.len());
        let client = reqwest::Client::new();
        client
            .post(&request_url)
            .header("x-dune-api-key", &self.api_key)
            .header("content-type", content_type)
            .body(body)
            .send()
            .await
    }

    /// Parses response body as text (for CSV endpoints).
    async fn _parse_text_response(resp: Response) -> Result<String, DuneRequestError> {
        if resp.status().is_success() {
            resp.text().await.map_err(DuneRequestError::from)
        } else {
            let err = resp
                .json::<DuneError>()
                .await
                .map_err(DuneRequestError::from)?;
            error!("request error {:?}", &err);
            Err(DuneRequestError::from(err))
        }
    }

    /// Execute Query (with or without parameters)
    /// cf. [https://dune.com/docs/api/api-reference/execute-queries/execute-query-id/](https://dune.com/docs/api/api-reference/execute-queries/execute-query-id/)
    ///
    /// # Example
    ///
    /// ```no_run
    /// use duners::{DuneClient, DuneRequestError};
    ///
    /// # async fn run() -> Result<(), DuneRequestError> {
    /// let client = DuneClient::from_env();
    /// let exec = client.execute_query(971694, None).await?;
    /// println!("Execution ID: {}", exec.execution_id);
    /// # Ok(()) }
    /// ```
    pub async fn execute_query(
        &self,
        query_id: u32,
        params: Option<Vec<Parameter>>,
    ) -> Result<ExecutionResponse, DuneRequestError> {
        let response = self
            ._post(&format!("query/{query_id}/execute"), params)
            .await
            .map_err(DuneRequestError::from)?;
        DuneClient::_parse_response::<ExecutionResponse>(response).await
    }

    /// Execute raw SQL directly without a saved query.
    ///
    /// The `performance` parameter controls the execution tier:
    /// `"medium"` (default), `"large"`, or `"community"`.
    ///
    /// # Example
    ///
    /// ```no_run
    /// use duners::{DuneClient, DuneRequestError};
    ///
    /// # async fn run() -> Result<(), DuneRequestError> {
    /// let client = DuneClient::from_env();
    /// let exec = client.execute_sql("SELECT 1 AS n", None).await?;
    /// println!("Execution ID: {}", exec.execution_id);
    /// # Ok(()) }
    /// ```
    pub async fn execute_sql(
        &self,
        sql: &str,
        performance: Option<&str>,
    ) -> Result<ExecutionResponse, DuneRequestError> {
        let mut body = json!({ "sql": sql });
        if let Some(perf) = performance {
            body["performance"] = json!(perf);
        }
        let response = self
            ._post_json("sql/execute", body)
            .await
            .map_err(DuneRequestError::from)?;
        DuneClient::_parse_response::<ExecutionResponse>(response).await
    }

    /// Cancel Query Execution by `job_id`
    /// cf. [https://dune.com/docs/api/api-reference/execute-queries/cancel-execution/](https://dune.com/docs/api/api-reference/execute-queries/cancel-execution/)
    pub async fn cancel_execution(
        &self,
        job_id: &str,
    ) -> Result<CancellationResponse, DuneRequestError> {
        let response = self
            ._post(&format!("execution/{job_id}/cancel"), None)
            .await
            .map_err(DuneRequestError::from)?;
        DuneClient::_parse_response::<CancellationResponse>(response).await
    }

    /// Get Query Execution Status (by `job_id`)
    /// cf. [https://dune.com/docs/api/api-reference/get-results/execution-status/](https://dune.com/docs/api/api-reference/get-results/execution-status/)
    pub async fn get_status(&self, job_id: &str) -> Result<GetStatusResponse, DuneRequestError> {
        let response = self
            ._get(job_id, "status")
            .await
            .map_err(DuneRequestError::from)?;
        DuneClient::_parse_response::<GetStatusResponse>(response).await
    }

    /// Get Query Execution Results (by `job_id`)
    /// cf. [https://dune.com/docs/api/api-reference/get-results/execution-results/](https://dune.com/docs/api/api-reference/get-results/execution-results/)
    ///
    /// # Example
    ///
    /// ```no_run
    /// use duners::{DuneClient, DuneRequestError};
    /// use serde::Deserialize;
    ///
    /// #[derive(Deserialize, Debug)]
    /// struct Row { symbol: String, max_price: f64 }
    ///
    /// # async fn run() -> Result<(), DuneRequestError> {
    /// let client = DuneClient::from_env();
    /// let results = client.get_results::<Row>("your-execution-id").await?;
    /// for row in results.get_rows() { println!("{:?}", row); }
    /// # Ok(()) }
    /// ```
    pub async fn get_results<T: DeserializeOwned>(
        &self,
        job_id: &str,
    ) -> Result<GetResultResponse<T>, DuneRequestError> {
        let response = self
            ._get(job_id, "results")
            .await
            .map_err(DuneRequestError::from)?;
        DuneClient::_parse_response::<GetResultResponse<T>>(response).await
    }

    /// Get the latest results for a query without triggering a new execution.
    ///
    /// Returns the most recent execution results for the given query ID.
    /// Does not consume credits (no re-execution).
    ///
    /// # Example
    ///
    /// ```no_run
    /// use duners::{DuneClient, DuneRequestError};
    /// use serde::Deserialize;
    ///
    /// #[derive(Deserialize, Debug)]
    /// struct Row { symbol: String, price: f64 }
    ///
    /// # async fn run() -> Result<(), DuneRequestError> {
    /// let client = DuneClient::from_env();
    /// let results = client.get_latest_results::<Row>(971694).await?;
    /// for row in results.get_rows() { println!("{:?}", row); }
    /// # Ok(()) }
    /// ```
    pub async fn get_latest_results<T: DeserializeOwned>(
        &self,
        query_id: u32,
    ) -> Result<GetResultResponse<T>, DuneRequestError> {
        let response = self
            ._get_url(&format!("query/{query_id}/results"))
            .await
            .map_err(DuneRequestError::from)?;
        DuneClient::_parse_response::<GetResultResponse<T>>(response).await
    }

    /// Get the latest results for a query as CSV text.
    pub async fn get_latest_results_csv(&self, query_id: u32) -> Result<String, DuneRequestError> {
        let response = self
            ._get_url(&format!("query/{query_id}/results/csv"))
            .await
            .map_err(DuneRequestError::from)?;
        DuneClient::_parse_text_response(response).await
    }

    /// Get execution results as CSV text (by `job_id`).
    pub async fn get_results_csv(&self, job_id: &str) -> Result<String, DuneRequestError> {
        let response = self
            ._get_url(&format!("execution/{job_id}/results/csv"))
            .await
            .map_err(DuneRequestError::from)?;
        DuneClient::_parse_text_response(response).await
    }

    /// Create a new Dune query.
    ///
    /// `body.name` and `body.query_sql` are required by the API.
    ///
    /// # Example
    ///
    /// ```no_run
    /// use duners::{DuneClient, DuneRequestError, QueryBody};
    ///
    /// # async fn run() -> Result<(), DuneRequestError> {
    /// let client = DuneClient::from_env();
    /// let resp = client.create_query(QueryBody {
    ///     name: Some("My query".into()),
    ///     query_sql: Some("SELECT 1 AS n".into()),
    ///     ..Default::default()
    /// }).await?;
    /// println!("Query ID: {}", resp.query_id);
    /// # Ok(()) }
    /// ```
    pub async fn create_query(&self, body: QueryBody) -> Result<QueryResponse, DuneRequestError> {
        let response = self
            ._post_json("query", serde_json::to_value(&body).unwrap())
            .await
            .map_err(DuneRequestError::from)?;
        DuneClient::_parse_response::<QueryResponse>(response).await
    }

    /// Read a query's metadata and SQL by ID.
    pub async fn get_query(&self, query_id: u32) -> Result<DuneQuery, DuneRequestError> {
        let response = self
            ._get_url(&format!("query/{query_id}"))
            .await
            .map_err(DuneRequestError::from)?;
        DuneClient::_parse_response::<DuneQuery>(response).await
    }

    /// Update a query's SQL, name, description, tags, or privacy.
    pub async fn update_query(
        &self,
        query_id: u32,
        body: QueryBody,
    ) -> Result<QueryResponse, DuneRequestError> {
        let response = self
            ._patch(
                &format!("query/{query_id}"),
                serde_json::to_value(&body).unwrap(),
            )
            .await
            .map_err(DuneRequestError::from)?;
        DuneClient::_parse_response::<QueryResponse>(response).await
    }

    /// Archive a query (prevents running or editing).
    pub async fn archive_query(&self, query_id: u32) -> Result<QueryResponse, DuneRequestError> {
        let response = self
            ._post_json(&format!("query/{query_id}/archive"), json!({}))
            .await
            .map_err(DuneRequestError::from)?;
        DuneClient::_parse_response::<QueryResponse>(response).await
    }

    /// Unarchive a previously archived query.
    pub async fn unarchive_query(&self, query_id: u32) -> Result<QueryResponse, DuneRequestError> {
        let response = self
            ._post_json(&format!("query/{query_id}/unarchive"), json!({}))
            .await
            .map_err(DuneRequestError::from)?;
        DuneClient::_parse_response::<QueryResponse>(response).await
    }

    /// Make a query private (owner-only access).
    pub async fn make_query_private(
        &self,
        query_id: u32,
    ) -> Result<QueryResponse, DuneRequestError> {
        let response = self
            ._post_json(&format!("query/{query_id}/private"), json!({}))
            .await
            .map_err(DuneRequestError::from)?;
        DuneClient::_parse_response::<QueryResponse>(response).await
    }

    /// Make a private query public.
    pub async fn make_query_public(
        &self,
        query_id: u32,
    ) -> Result<QueryResponse, DuneRequestError> {
        let response = self
            ._post_json(&format!("query/{query_id}/unprivate"), json!({}))
            .await
            .map_err(DuneRequestError::from)?;
        DuneClient::_parse_response::<QueryResponse>(response).await
    }

    /// Create an empty table with an explicit schema.
    pub async fn create_table(
        &self,
        request: CreateTableRequest,
    ) -> Result<CreateTableResponse, DuneRequestError> {
        let response = self
            ._post_json("uploads", serde_json::to_value(&request).unwrap())
            .await
            .map_err(DuneRequestError::from)?;
        DuneClient::_parse_response::<CreateTableResponse>(response).await
    }

    /// Upload CSV data to create or replace a table.
    ///
    /// # Example
    ///
    /// ```no_run
    /// use duners::{DuneClient, DuneRequestError, UploadCsvRequest};
    ///
    /// # async fn run() -> Result<(), DuneRequestError> {
    /// let client = DuneClient::from_env();
    /// let resp = client.upload_csv(UploadCsvRequest {
    ///     data: "name,age\nAlice,30\nBob,25".into(),
    ///     table_name: "my_table".into(),
    ///     description: None,
    ///     is_private: Some(true),
    /// }).await?;
    /// println!("Table: {}", resp.full_name);
    /// # Ok(()) }
    /// ```
    pub async fn upload_csv(
        &self,
        request: UploadCsvRequest,
    ) -> Result<CreateTableResponse, DuneRequestError> {
        let response = self
            ._post_json("uploads/csv", serde_json::to_value(&request).unwrap())
            .await
            .map_err(DuneRequestError::from)?;
        DuneClient::_parse_response::<CreateTableResponse>(response).await
    }

    /// Insert rows into an existing table.
    ///
    /// `content_type` should be `"text/csv"` or `"application/x-ndjson"`.
    pub async fn insert_table_rows(
        &self,
        namespace: &str,
        table_name: &str,
        content_type: &str,
        data: String,
    ) -> Result<InsertTableResponse, DuneRequestError> {
        let response = self
            ._post_raw(
                &format!("uploads/{namespace}/{table_name}/insert"),
                content_type,
                data,
            )
            .await
            .map_err(DuneRequestError::from)?;
        DuneClient::_parse_response::<InsertTableResponse>(response).await
    }

    /// Remove all data from a table (preserves schema).
    pub async fn clear_table(
        &self,
        namespace: &str,
        table_name: &str,
    ) -> Result<SuccessResponse, DuneRequestError> {
        let response = self
            ._post_json(
                &format!("uploads/{namespace}/{table_name}/clear"),
                json!({}),
            )
            .await
            .map_err(DuneRequestError::from)?;
        DuneClient::_parse_response::<SuccessResponse>(response).await
    }

    /// Permanently delete a table and all its data.
    pub async fn delete_table(
        &self,
        namespace: &str,
        table_name: &str,
    ) -> Result<SuccessResponse, DuneRequestError> {
        let response = self
            ._delete(&format!("uploads/{namespace}/{table_name}"))
            .await
            .map_err(DuneRequestError::from)?;
        DuneClient::_parse_response::<SuccessResponse>(response).await
    }

    /// Convenience method for users to
    /// 1. execute,
    /// 2. wait for execution to complete,
    /// 3. fetch and return query results.
    /// # Arguments
    /// * `query_id` - an integer representing query ID
    ///   (found at the end of a Dune Query URL: [https://dune.com/queries/971694](https://dune.com/queries/971694))
    /// * `parameters` - an optional list of query `Parameter`
    ///   (cf. [https://dune.xyz/queries/3238619](https://dune.xyz/queries/3238619))
    /// * `ping_frequency` - how frequently (in seconds) should the loop check execution status.
    ///   Default is 5 seconds. Too frequently could result in rate limiting
    ///   (i.e. Too Many Requests) especially when executing multiple queries in parallel.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use duners::{DuneClient, DuneRequestError};
    /// use duners::parse_utils::{datetime_from_str, f64_from_str};
    /// use serde::Deserialize;
    /// use chrono::{DateTime, Utc};
    ///
    /// #[derive(Deserialize, Debug)]
    /// struct ResultStruct {
    ///     text_field: String,
    ///     #[serde(deserialize_with = "f64_from_str")]
    ///     number_field: f64,
    ///     #[serde(deserialize_with = "datetime_from_str")]
    ///     date_field: DateTime<Utc>,
    ///     list_field: String,
    /// }
    ///
    /// #[tokio::main]
    /// async fn main() -> Result<(), DuneRequestError> {
    ///     let client = DuneClient::from_env();
    ///     let result = client.refresh::<ResultStruct>(1215383, None, None).await?;
    ///     println!("{:?}", result.get_rows());
    ///     Ok(())
    /// }
    /// ```
    pub async fn refresh<T: DeserializeOwned>(
        &self,
        query_id: u32,
        parameters: Option<Vec<Parameter>>,
        ping_frequency: Option<u64>,
    ) -> Result<GetResultResponse<T>, DuneRequestError> {
        let job_id = self.execute_query(query_id, parameters).await?.execution_id;
        info!("Refreshing {} Execution ID {}", query_id, job_id);
        let mut status = self.get_status(&job_id).await?;
        while !status.state.is_terminal() {
            info!(
                "waiting for query execution {job_id} to complete: {:?}",
                status.state
            );
            sleep(Duration::from_secs(ping_frequency.unwrap_or(5))).await;
            status = self.get_status(&job_id).await?
        }
        let full_response = self.get_results::<T>(&job_id).await;
        if status.state == ExecutionStatus::Failed {
            warn!(
                "{:?} Perhaps your query took too long to run!",
                status.state
            );
        }
        full_response
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::parse_utils::{date_parse, datetime_from_str, f64_from_str};
    use crate::response::ExecutionStatus;
    use chrono::{DateTime, Utc};
    use serde::Deserialize;

    const QUERY_ID: u32 = 971694;
    const JOB_ID: &str = "01KHDCT5QFS1QPE9T2QEWPEAGG";

    #[tokio::test]
    async fn invalid_api_key() {
        let dune = DuneClient::new("Baloney");
        let error = dune.execute_query(QUERY_ID, None).await.unwrap_err();
        assert_eq!(
            error,
            DuneRequestError::Dune(String::from("invalid API Key"))
        )
    }

    #[tokio::test]
    async fn invalid_query_id() {
        let dune = DuneClient::from_env();
        let error = dune.execute_query(u32::MAX, None).await.unwrap_err();
        assert_eq!(
            error,
            DuneRequestError::Dune(String::from("An internal error occurred"))
        )
    }

    #[tokio::test]
    async fn invalid_job_id() {
        let dune = DuneClient::from_env();
        let error = dune
            .get_results::<DuneError>("wonky job ID")
            .await
            .unwrap_err();
        assert_eq!(
            error,
            DuneRequestError::Dune(String::from(
                "The requested execution ID (ID: wonky job ID) is invalid."
            ))
        )
    }

    #[tokio::test]
    async fn execute_query() {
        let dune = DuneClient::from_env();
        let exec = dune.execute_query(QUERY_ID, None).await.unwrap();
        // Also testing cancellation!
        let cancellation = dune.cancel_execution(&exec.execution_id).await.unwrap();
        assert!(cancellation.success);
    }

    #[tokio::test]
    async fn execute_query_with_params() {
        let dune = DuneClient::from_env();
        let all_parameter_types = vec![
            Parameter::date("DateField", date_parse("2022-05-04T00:00:00.0Z").unwrap()),
            Parameter::number("NumberField", "3.1415926535"),
            Parameter::text("TextField", "Plain Text"),
            Parameter::list("ListField", "Option 1"),
        ];
        let exec_result = dune.execute_query(1215383, Some(all_parameter_types)).await;
        assert!(exec_result.is_ok())
    }

    #[tokio::test]
    async fn get_status() {
        let dune = DuneClient::from_env();
        let status = dune.get_status(JOB_ID).await.unwrap();
        assert_eq!(status.state, ExecutionStatus::Complete)
    }

    #[tokio::test]
    async fn get_results() {
        let dune = DuneClient::from_env();

        #[derive(Deserialize, Debug)]
        struct ExpectedResults {
            token: String,
            symbol: String,
            max_price: f64,
        }

        let results = dune.get_results::<ExpectedResults>(JOB_ID).await.unwrap();
        // Query is for the max ETH price (should only have 1 result)
        let rows = results.result.rows;
        assert_eq!(1, rows.len());
        assert_eq!(rows[0].symbol, "WETH");
        assert_eq!(rows[0].token, "0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2");
        assert!(rows[0].max_price > 4148.0)
    }

    #[tokio::test]
    async fn refresh() {
        let dune = DuneClient::from_env();

        #[derive(Deserialize, Debug, PartialEq)]
        struct ResultStruct {
            text_field: String,
            #[serde(deserialize_with = "f64_from_str")]
            number_field: f64,
            #[serde(deserialize_with = "datetime_from_str")]
            date_field: DateTime<Utc>,
            list_field: String,
        }
        let results = dune
            .refresh::<ResultStruct>(
                3238619,
                Some(vec![Parameter::number("NumberField", "3.141592653589793")]),
                None,
            )
            .await
            .unwrap();
        assert_eq!(
            ResultStruct {
                text_field: "Plain Text".to_string(),
                number_field: std::f64::consts::PI,
                date_field: date_parse("2022-05-04T00:00:00.0Z").unwrap(),
                list_field: "Option 1".to_string(),
            },
            results.get_rows()[0]
        )
    }

    #[tokio::test]
    async fn table_lifecycle() {
        use crate::response::{ColumnDef, CreateTableRequest};

        let dune = DuneClient::from_env();
        let namespace = env::var("DUNE_NAMESPACE").unwrap_or_else(|_| "bh2smith".to_string());
        let table_name = "duners_test_table";

        // Create table with schema
        let created = dune
            .create_table(CreateTableRequest {
                namespace: namespace.clone(),
                table_name: table_name.to_string(),
                schema: vec![
                    ColumnDef {
                        name: "name".to_string(),
                        column_type: "varchar".to_string(),
                        nullable: None,
                    },
                    ColumnDef {
                        name: "age".to_string(),
                        column_type: "integer".to_string(),
                        nullable: None,
                    },
                ],
                description: None,
                is_private: Some(true),
            })
            .await
            .unwrap();
        assert!(!created.full_name.is_empty());

        // Insert rows via CSV
        let inserted = dune
            .insert_table_rows(
                &namespace,
                table_name,
                "text/csv",
                "name,age\nAlice,30\nBob,25".to_string(),
            )
            .await
            .unwrap();
        assert_eq!(inserted.rows_written, 2);

        // Clear table
        let cleared = dune.clear_table(&namespace, table_name).await.unwrap();
        assert!(cleared.message.is_some());

        // Delete table
        let deleted = dune.delete_table(&namespace, table_name).await.unwrap();
        assert!(deleted.message.is_some());
    }

    #[tokio::test]
    async fn upload_csv_lifecycle() {
        use crate::response::UploadCsvRequest;

        let dune = DuneClient::from_env();
        let namespace = env::var("DUNE_NAMESPACE").unwrap_or_else(|_| "bh2smith".to_string());

        // Upload CSV (creates the table)
        let upload = dune
            .upload_csv(UploadCsvRequest {
                data: "name,age\nAlice,30\nBob,25".to_string(),
                table_name: "duners_csv_test".to_string(),
                description: None,
                is_private: Some(true),
            })
            .await
            .unwrap();
        assert!(!upload.full_name.is_empty());

        // Clean up
        let actual_table = upload.table_name.as_deref().unwrap_or("duners_csv_test");
        dune.delete_table(&namespace, actual_table).await.unwrap();
    }

    #[tokio::test]
    async fn query_crud_lifecycle() {
        use crate::response::QueryBody;

        let dune = DuneClient::from_env();

        // Create
        let created = dune
            .create_query(QueryBody {
                name: Some("duners test query".to_string()),
                query_sql: Some("SELECT 1 AS n".to_string()),
                ..Default::default()
            })
            .await
            .unwrap();
        let qid = created.query_id;
        assert!(qid > 0);

        // Read
        let query = dune.get_query(qid).await.unwrap();
        assert_eq!(query.name, "duners test query");
        assert_eq!(query.query_sql, "SELECT 1 AS n");

        // Update
        let updated = dune
            .update_query(
                qid,
                QueryBody {
                    name: Some("duners test query updated".to_string()),
                    ..Default::default()
                },
            )
            .await
            .unwrap();
        assert_eq!(updated.query_id, qid);

        // Make private
        dune.make_query_private(qid).await.unwrap();
        let query = dune.get_query(qid).await.unwrap();
        assert!(query.is_private);

        // Make public
        dune.make_query_public(qid).await.unwrap();
        let query = dune.get_query(qid).await.unwrap();
        assert!(!query.is_private);

        // Archive
        dune.archive_query(qid).await.unwrap();
        let query = dune.get_query(qid).await.unwrap();
        assert!(query.is_archived);

        // Unarchive
        dune.unarchive_query(qid).await.unwrap();
        let query = dune.get_query(qid).await.unwrap();
        assert!(!query.is_archived);

        // Clean up: archive again
        dune.archive_query(qid).await.unwrap();
    }

    #[tokio::test]
    async fn execute_sql() {
        let dune = DuneClient::from_env();
        let exec = dune.execute_sql("SELECT 1 AS n", None).await.unwrap();
        assert!(!exec.execution_id.is_empty());
        let cancellation = dune.cancel_execution(&exec.execution_id).await.unwrap();
        assert!(cancellation.success);
    }

    #[tokio::test]
    async fn get_latest_results() {
        let dune = DuneClient::from_env();

        let results = dune
            .get_latest_results::<HashMap<String, serde_json::Value>>(QUERY_ID)
            .await
            .unwrap();
        let rows = results.result.rows;
        assert_eq!(1, rows.len());
        assert_eq!(rows[0]["symbol"], "WETH");
    }

    #[tokio::test]
    async fn get_latest_results_csv() {
        let dune = DuneClient::from_env();
        let csv = dune.get_latest_results_csv(QUERY_ID).await.unwrap();
        assert!(csv.contains("token"));
        assert!(csv.contains("WETH"));
    }

    #[tokio::test]
    async fn get_results_csv() {
        let dune = DuneClient::from_env();
        let csv = dune.get_results_csv(JOB_ID).await.unwrap();
        assert!(csv.contains("token"));
        assert!(csv.contains("WETH"));
    }

    #[tokio::test]
    #[ignore]
    async fn long_running_query() {
        let dune = DuneClient::from_env();
        let results = dune
            .refresh::<HashMap<String, f64>>(1229120, None, None)
            .await
            .unwrap();
        println!("Job ID {:?}", results.execution_id);
        assert_eq!(results.state, ExecutionStatus::Complete);
    }
}