spiral-sdk 0.1.10

//! Shared transport and authentication for resource clients.

use std::fmt::{self, Debug, Formatter};
use std::sync::Arc;
use std::time::Duration;

use reqwest::Url;
#[cfg(feature = "tables")]
use uuid::Uuid;

use crate::auth::{
    CredentialChain, OidcAuthProvider, Token, TokenSource, default_auth_url, token_endpoint,
};
#[cfg(feature = "private-key-auth")]
use crate::auth::{PrivateKeyCredentials, PrivateKeyJwtProvider};
use crate::http::{HttpClient, http_client_builder};
use crate::resources::applications::{
    ApplicationService, ApplicationsClient, ExecutionSettings, InstallationsClient,
    clients as application_clients,
};
use crate::resources::arbor::Client as ArborClient;
use crate::resources::catalog::Client as CatalogClient;
use crate::resources::connection::HttpExternalConnectionClient;
use crate::resources::location::HttpLocationClient;
use crate::resources::organization::{Organization, OrganizationAdminClient};
use crate::resources::project::{AuthorizationClient, ProjectClient};
use crate::resources::task::HttpTaskClient;
use crate::resources::vector_index::HttpVectorIndexClient;
use crate::resources::workload::WorkloadClient;
#[cfg(feature = "tables")]
use crate::scandal::Scandal;
#[cfg(feature = "tables")]
use crate::scandal_backend::ScandalBackend;
#[cfg(feature = "tables")]
use crate::table::{Table, TableIdentity};
use crate::{Config, Error};
#[cfg(feature = "tables")]
use crate::{Functions, RemoteScandal};
#[cfg(feature = "embedded")]
use crate::{NativeScandal, ResourceUriResolver, ScandalConfig};

pub(crate) struct Connection {
    http: HttpClient,
    projects: ProjectClient,
    organizations: Organization,
    workloads: WorkloadClient,
    authorization: AuthorizationClient,
    pub(crate) locations: HttpLocationClient,
    connections: HttpExternalConnectionClient,
    pub(crate) tasks: HttpTaskClient,
    pub(crate) vector_indexes: HttpVectorIndexClient,
    auth: Option<TokenSource>,
    applications: Arc<ApplicationService>,
    pub(crate) arbor: ArborClient,
    pub(crate) catalog: CatalogClient,
    #[cfg(feature = "tables")]
    tables: Option<Arc<ScandalBackend>>,
}

impl Connection {
    /// Table execution through the selected scandal.
    #[cfg(feature = "tables")]
    pub(crate) fn tables(&self) -> Result<&ScandalBackend, Error> {
        self.tables.as_deref().ok_or_else(|| {
            Error::invalid("no scandal selected; configure Client::builder().scandal(...)")
        })
    }
}

/// A shared Spiral connection with one transport and credential provider.
#[derive(Clone)]
pub struct Client {
    pub(crate) connection: Arc<Connection>,
}

impl Debug for Client {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        formatter.debug_struct("Client").finish_non_exhaustive()
    }
}

impl Client {
    /// Configure the control-plane endpoints and the scandal that runs Table operations.
    pub fn builder() -> ClientBuilder {
        ClientBuilder::default()
    }

    /// The default endpoints and the noninteractive default credential chain.
    pub async fn from_env() -> Result<Self, Error> {
        Self::from_config(&Config::load().await?)
    }

    /// Share a resolved configuration's HTTP pool and credential provider.
    pub fn from_config(config: &Config) -> Result<Self, Error> {
        Self::builder().config(config).build()
    }

    /// Shared authenticated transport for HTTP operations without a typed SDK client.
    pub fn http(&self) -> &HttpClient {
        &self.connection.http
    }

    /// Resolve the current credential without changing the selected identity.
    pub async fn access_token(&self) -> Result<Token, Error> {
        self.connection
            .auth
            .as_ref()
            .ok_or(Error::CredentialsNotFound)?
            .token()
            .await?
            .ok_or(Error::CredentialsNotFound)
    }

    /// Project lifecycle and shared presentation state.
    pub fn projects(&self) -> &ProjectClient {
        &self.connection.projects
    }

    /// Access grants and role management through the shared authenticated connection.
    pub fn authorization(&self) -> &AuthorizationClient {
        &self.connection.authorization
    }

    /// Organizations, memberships, users, invitations and settings.
    pub fn organization(&self) -> &Organization {
        &self.connection.organizations
    }

    /// Platform administration of organization synchronization and lifecycle.
    pub fn organization_admin(&self) -> &OrganizationAdminClient {
        &self.connection.organizations
    }

    /// Workload identities, public keys and trust policies.
    pub fn workloads(&self) -> &WorkloadClient {
        &self.connection.workloads
    }

    /// Application registrations, scopes and organization app-access policy.
    pub fn applications(&self) -> ApplicationsClient {
        ApplicationsClient {
            service: self.connection.applications.clone(),
        }
    }

    /// Organization application installations and their approval lifecycle.
    pub fn installations(&self) -> InstallationsClient {
        InstallationsClient {
            service: self.connection.applications.clone(),
            connection: self.clone(),
        }
    }

    /// Low-level table resource client for tools and integrations.
    pub fn tables(&self) -> &ArborClient {
        &self.connection.arbor
    }

    /// Project catalog client.
    pub fn catalog(&self) -> &CatalogClient {
        &self.connection.catalog
    }

    /// Location resource client.
    pub fn locations(&self) -> &HttpLocationClient {
        &self.connection.locations
    }

    /// Connection service client for credentials, metadata and connectivity checks.
    pub fn connections(&self) -> &HttpExternalConnectionClient {
        &self.connection.connections
    }

    /// Task resource client.
    pub fn tasks(&self) -> &HttpTaskClient {
        &self.connection.tasks
    }

    /// Vector-index resource client.
    pub fn vector_indexes(&self) -> &HttpVectorIndexClient {
        &self.connection.vector_indexes
    }

    /// The SPQL functions of the scandal that runs this client's Table operations.
    #[cfg(feature = "tables")]
    pub async fn functions(&self) -> Result<Functions, Error> {
        Functions::new(self.connection.tables()?.scandal()).await
    }

    /// Open an exact table version by IDs for tools and integrations.
    #[cfg(feature = "tables")]
    pub fn table_by_id(&self, resource_id: Uuid, version_id: Uuid) -> Table {
        Table::new(
            Arc::clone(&self.connection),
            TableIdentity {
                resource_id,
                version_id,
            },
        )
    }
}

/// Settings shared by generated resource clients and SDK orchestration.
#[derive(Default)]
pub struct ClientBuilder {
    api_url: Option<String>,
    auth_url: Option<String>,
    scandal_url: Option<String>,
    authentication: Option<ClientAuthentication>,
    http: Option<reqwest::Client>,
    #[cfg(feature = "tables")]
    scandal: Option<Arc<dyn Scandal>>,
    #[cfg(feature = "embedded")]
    embedded: Option<EmbeddedScandals>,
}

enum ClientAuthentication {
    Provider(TokenSource),
    #[cfg(feature = "private-key-auth")]
    PrivateKey(PrivateKeyCredentials),
    TrustedIssuer(String, TokenSource),
}

impl ClientBuilder {
    /// Share resolved settings, endpoints, credentials and the HTTP pool.
    pub fn config(mut self, config: &Config) -> Self {
        self.api_url = Some(config.api_url.to_string());
        self.auth_url = Some(config.auth_url.to_string());
        self.scandal_url = config.scandal_url.as_ref().map(ToString::to_string);
        self.authentication = Some(ClientAuthentication::Provider(config.auth.clone()));
        self.http = Some(config.http.clone());
        self
    }

    /// Set the control-plane endpoint.
    pub fn api_url(mut self, url: impl Into<String>) -> Self {
        self.api_url = Some(url.into());
        self
    }

    /// Set the authorization server for app and installation token exchanges.
    pub fn auth_url(mut self, url: impl Into<String>) -> Self {
        self.auth_url = Some(url.into());
        self
    }

    /// Authenticate with a registered private JWK, renewing tokens automatically.
    #[cfg(feature = "private-key-auth")]
    pub fn private_key(mut self, client_id: impl Into<String>, jwk: &str) -> Result<Self, Error> {
        self.authentication = Some(ClientAuthentication::PrivateKey(
            PrivateKeyCredentials::from_jwk(client_id.into(), jwk)?,
        ));
        Ok(self)
    }

    /// Exchange fresh trusted-issuer assertions for this client's renewable tokens.
    pub fn trusted_issuer(mut self, client_id: impl Into<String>, source: TokenSource) -> Self {
        self.authentication = Some(ClientAuthentication::TrustedIssuer(
            client_id.into(),
            source,
        ));
        self
    }

    /// Execute table operations through remote scandal at this endpoint.
    ///
    /// With credentials, every request carries a scandal capability exchanged from this
    /// client's access token at the authorization server; each identity the client acts as,
    /// such as an installation, obtains its own.
    pub fn scandal_url(mut self, url: impl Into<String>) -> Self {
        self.scandal_url = Some(url.into());
        self
    }

    /// Use one redirect-disabled HTTP pool for all clients.
    /// Configure credentials through the builder independently of this pool.
    pub fn http_client(mut self, http: reqwest::Client) -> Self {
        self.http = Some(http);
        self
    }

    /// Authenticate with a fixed bearer credential.
    pub fn bearer_token(mut self, token: impl Into<String>) -> Self {
        self.authentication = Some(ClientAuthentication::Provider(TokenSource::fixed(
            token.into().into(),
        )));
        self
    }

    /// Resolve the current credential immediately before every request.
    ///
    /// The provider renews one identity's credential: table reads reuse the storage access that
    /// earlier reads resolved with it. Build another client to act as another identity.
    pub fn token_source(mut self, provider: TokenSource) -> Self {
        self.authentication = Some(ClientAuthentication::Provider(provider));
        self
    }

    /// Execute table operations through `scandal`, embedded or remote.
    ///
    /// The client resolves storage bindings with its own identity, so the same scandal serves
    /// an installation authenticated later. A [`RemoteScandal`] given here calls as whichever
    /// caller it was built with; [`scandal_url`](Self::scandal_url) gives each identity its own.
    #[cfg(feature = "tables")]
    pub fn scandal(mut self, scandal: Arc<dyn Scandal>) -> Self {
        self.scandal = Some(scandal);
        self
    }

    /// Execute table operations in an embedded [`NativeScandal`] with the storage transports
    /// `config` allows. It binds the `spiral://` addresses that rows name, such as a file under
    /// a Location, as this client's identity, and each identity the client acts as, such as an
    /// installation, gets its own. Build the client inside a Tokio runtime.
    #[cfg(feature = "embedded")]
    pub fn embedded_scandal(mut self, config: ScandalConfig) -> Self {
        // Only a caller of this method reaches the native library, so a binary that never
        // selects an embedded scandal links without it.
        self.embedded = Some(EmbeddedScandals(Arc::new(move |http, caller, api_url| {
            embedded(&config, http, caller, api_url)
        })));
        self
    }

    /// Execute table operations in embedded scandals that `scandals` builds.
    #[cfg(feature = "embedded")]
    pub(crate) fn embedded_scandals(mut self, scandals: EmbeddedScandals) -> Self {
        self.embedded = Some(scandals);
        self
    }

    /// Build all resource clients and the streaming table client.
    pub fn build(self) -> Result<Client, Error> {
        let api = endpoint(
            self.api_url
                .as_deref()
                .ok_or_else(|| Error::invalid("api_url is required"))?,
        )?;
        let http = self.http.unwrap_or_else(|| {
            http_client_builder()
                .redirect(reqwest::redirect::Policy::none())
                .timeout(Duration::from_secs(60))
                .build()
                .expect("valid HTTP defaults")
        });
        let auth_url = self
            .auth_url
            .as_deref()
            .map(endpoint)
            .transpose()?
            .unwrap_or_else(default_auth_url);
        let auth: Option<TokenSource> = match self.authentication {
            None => None,
            Some(ClientAuthentication::Provider(provider)) => Some(provider),
            #[cfg(feature = "private-key-auth")]
            Some(ClientAuthentication::PrivateKey(credentials)) => Some(
                CredentialChain::new(vec![
                    PrivateKeyJwtProvider::new(&auth_url)?
                        .with_credentials(credentials)
                        .into(),
                ])
                .into(),
            ),
            Some(ClientAuthentication::TrustedIssuer(id, source)) => Some(
                CredentialChain::new(vec![
                    OidcAuthProvider::new(token_endpoint(&auth_url)?, id)
                        .with_assertion_provider(source)
                        .into(),
                ])
                .into(),
            ),
        };

        let management = HttpClient::new(api.clone(), http.clone(), auth.clone());
        #[cfg(feature = "tables")]
        let mut selections = Vec::new();
        #[cfg(feature = "tables")]
        {
            selections.extend(self.scandal.map(ScandalSelection::Given));
            if let Some(url) = self.scandal_url {
                selections.push(ScandalSelection::Remote(endpoint(&url)?));
            }
        }
        #[cfg(feature = "embedded")]
        selections.extend(self.embedded.map(ScandalSelection::Embedded));
        #[cfg(feature = "tables")]
        let selection = match selections.len() {
            0 | 1 => selections.pop(),
            _ => {
                return Err(Error::invalid(
                    "select one of a scandal, a scandal endpoint or an embedded scandal",
                ));
            }
        };
        #[cfg(feature = "tables")]
        let scandal = selection
            .as_ref()
            .map(|selection| selection.scandal(&http, auth.as_ref(), &auth_url, &api))
            .transpose()?;
        let applications = application_clients(
            api.as_str(),
            auth_url,
            http,
            auth.clone(),
            ExecutionSettings {
                #[cfg(feature = "tables")]
                scandal: selection,
            },
        )?;

        let arbor = ArborClient::from_http(management.clone())?;
        let catalog = CatalogClient::from_http(management.clone())?;
        #[cfg(feature = "tables")]
        let tables = scandal.map(|scandal| Arc::new(ScandalBackend::new(scandal, arbor.clone())));
        Ok(Client {
            connection: Arc::new(Connection {
                projects: ProjectClient::from_http(management.clone())?,
                organizations: Organization::from_http(management.clone())?,
                workloads: WorkloadClient::from_http(management.clone())?,
                authorization: AuthorizationClient::from_http(management.clone())?,
                locations: HttpLocationClient::from_http(management.clone())?,
                connections: HttpExternalConnectionClient::from_http(management.clone())?,
                tasks: HttpTaskClient::from_http(management.clone())?,
                vector_indexes: HttpVectorIndexClient::from_http(management.clone())?,
                http: management,
                auth,
                applications,
                arbor,
                catalog,
                #[cfg(feature = "tables")]
                tables,
            }),
        })
    }
}

/// The scandal a client was configured with, which each identity it acts as calls.
#[cfg(feature = "tables")]
#[derive(Clone)]
pub(crate) enum ScandalSelection {
    /// A scandal the application built, shared by every identity as it is.
    Given(Arc<dyn Scandal>),
    /// Remote scandal at this endpoint, which each identity calls with its own capability.
    Remote(Url),
    /// An embedded scandal for each identity, which resolves addresses as that identity.
    #[cfg(feature = "embedded")]
    Embedded(EmbeddedScandals),
}

/// Builds an embedded scandal for one identity: from the HTTP pool, the identity's credentials
/// when it has any, and the API its addresses resolve through.
#[cfg(feature = "embedded")]
type EmbeddedScandalFn = dyn Fn(&reqwest::Client, Option<&TokenSource>, &Url) -> Result<Arc<dyn Scandal>, Error>
    + Send
    + Sync;

/// The embedded scandals a client builds, one for each identity it acts as.
#[cfg(feature = "embedded")]
#[derive(Clone)]
pub(crate) struct EmbeddedScandals(Arc<EmbeddedScandalFn>);

#[cfg(feature = "tables")]
impl ScandalSelection {
    /// The scandal for `caller`'s requests.
    fn scandal(
        &self,
        http: &reqwest::Client,
        caller: Option<&TokenSource>,
        auth_url: &Url,
        api_url: &Url,
    ) -> Result<Arc<dyn Scandal>, Error> {
        let endpoint = match self {
            Self::Given(scandal) => return Ok(Arc::clone(scandal)),
            Self::Remote(endpoint) => endpoint,
            #[cfg(feature = "embedded")]
            Self::Embedded(scandals) => return (scandals.0)(http, caller, api_url),
        };
        let remote = RemoteScandal::new(endpoint.clone())?.with_http(http.clone());
        Ok(Arc::new(match caller {
            Some(caller) => remote.with_caller(caller.clone(), auth_url, api_url)?,
            None => remote,
        }))
    }
}

/// An embedded scandal that resolves the addresses rows name as `caller`, when there is one.
#[cfg(feature = "embedded")]
fn embedded(
    config: &ScandalConfig,
    http: &reqwest::Client,
    caller: Option<&TokenSource>,
    api_url: &Url,
) -> Result<Arc<dyn Scandal>, Error> {
    Ok(Arc::new(match caller {
        Some(caller) => NativeScandal::new_with_uri_resolver(
            config,
            Arc::new(ResourceUriResolver::new(
                api_url.clone(),
                caller.clone(),
                http.clone(),
            )),
        )?,
        None => NativeScandal::new(config)?,
    }))
}

pub(crate) fn endpoint(value: &str) -> Result<Url, Error> {
    let mut url = Url::parse(value).map_err(Error::invalid)?;
    if !matches!(url.scheme(), "http" | "https")
        || url.host_str().is_none()
        || !url.username().is_empty()
        || url.password().is_some()
        || url.query().is_some()
        || url.fragment().is_some()
    {
        return Err(Error::invalid(
            "service endpoint must be an HTTP(S) URL without credentials, query, or fragment",
        ));
    }
    if !url.path().ends_with('/') {
        url.set_path(&format!("{}/", url.path()));
    }
    Ok(url)
}

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

    #[test]
    fn builder_rejects_non_http_and_credentialed_endpoints() {
        for endpoint in [
            "file:///tmp/spiral",
            "https://user@example.com",
            "https://example.com?token=secret",
        ] {
            assert!(Client::builder().api_url(endpoint).build().is_err());
        }
    }

    #[test]
    fn builder_constructs_every_resource_client_without_native_code() {
        let client = Client::builder()
            .api_url("https://api.example.test/base")
            .scandal_url("https://scandal.example.test")
            .bearer_token("test-token")
            .build()
            .expect("valid SDK settings");
        let _ = client.catalog();
        let _ = client.locations();
        let _ = client.tasks();
        let _ = client.vector_indexes();
    }
}