alien-bindings 3.3.28

//! App-facing binding handles that keep minted credentials fresh.
//!
//! Each operation re-enters the configured [`BindingsProviderApi`]. Static or
//! fresh providers remain cheap cache hits; short-lived providers inside their
//! refresh window are rebuilt through their single-flight path. The resolved
//! provider is held for the full operation so credential rotation cannot swap
//! it midway through a request.
//!
//! Methods that return an owned stream or multipart-upload session refresh
//! before creating it. An already-returned opaque stream/session remains bound
//! to that provider; the `object_store` API offers no way to replace its
//! credentials midway through the operation.

use std::fmt;
use std::ops::Range;
use std::sync::Arc;
use std::time::Duration;

use alien_error::AlienError;
use async_trait::async_trait;
use bytes::Bytes;
use futures::stream::{self, BoxStream};
use futures::{StreamExt, TryStreamExt};
use object_store::path::Path;
use object_store::{
    GetOptions, GetResult, ListResult, MultipartUpload, ObjectMeta, ObjectStore,
    PutMultipartOptions, PutOptions as ObjectStorePutOptions, PutPayload, PutResult,
};
use url::Url;

use crate::error::{ErrorData, Result};
use crate::presigned::PresignedRequest;
#[cfg(feature = "platform-sdk")]
use crate::remote::RemoteStorage;
use crate::traits::{
    Binding, BindingsProviderApi, Key, Kv, KvEntry, MessagePayload, PutOptions as KvPutOptions,
    Queue, QueueMessage, ScanResult, Storage, Vault, Worker, WorkerInvokeRequest,
    WorkerInvokeResponse,
};

const OBJECT_STORE_NAME: &str = "Alien binding";

/// The smallest provider surface needed by a refreshable Storage handle.
///
/// Environment-backed providers implement the full bindings API and receive
/// this implementation automatically. Remote bindings implement only this
/// trait, so unsupported binding kinds cannot leak into their public surface.
#[async_trait]
pub(super) trait StorageProviderApi: Send + Sync + fmt::Debug {
    async fn load_storage(&self, binding_name: &str) -> Result<Arc<dyn Storage>>;
}

/// The smallest provider surface needed by a refreshable Key handle.
#[async_trait]
pub(super) trait KeyProviderApi: Send + Sync + fmt::Debug {
    async fn load_key(&self, binding_name: &str) -> Result<Arc<dyn Key>>;
}

/// The smallest provider surface needed by a refreshable KV handle.
#[async_trait]
pub(super) trait KvProviderApi: Send + Sync + fmt::Debug {
    async fn load_kv(&self, binding_name: &str) -> Result<Arc<dyn Kv>>;
}

/// The smallest provider surface needed by a refreshable Queue handle.
#[async_trait]
pub(super) trait QueueProviderApi: Send + Sync + fmt::Debug {
    async fn load_queue(&self, binding_name: &str) -> Result<Arc<dyn Queue>>;
}

#[async_trait]
impl<T> QueueProviderApi for T
where
    T: BindingsProviderApi + Send + Sync + fmt::Debug,
{
    async fn load_queue(&self, binding_name: &str) -> Result<Arc<dyn Queue>> {
        BindingsProviderApi::load_queue(self, binding_name).await
    }
}

#[async_trait]
impl<T> KvProviderApi for T
where
    T: BindingsProviderApi + Send + Sync + fmt::Debug,
{
    async fn load_kv(&self, binding_name: &str) -> Result<Arc<dyn Kv>> {
        BindingsProviderApi::load_kv(self, binding_name).await
    }
}

#[async_trait]
impl<T> KeyProviderApi for T
where
    T: BindingsProviderApi + Send + Sync + fmt::Debug,
{
    async fn load_key(&self, binding_name: &str) -> Result<Arc<dyn Key>> {
        BindingsProviderApi::load_key(self, binding_name).await
    }
}

#[async_trait]
impl<T> StorageProviderApi for T
where
    T: BindingsProviderApi + Send + Sync + fmt::Debug,
{
    async fn load_storage(&self, binding_name: &str) -> Result<Arc<dyn Storage>> {
        BindingsProviderApi::load_storage(self, binding_name).await
    }
}

#[derive(Debug, Clone)]
struct Resolver {
    provider: Arc<dyn BindingsProviderApi>,
    binding_name: String,
}

impl Resolver {
    fn new(provider: Arc<dyn BindingsProviderApi>, binding_name: String) -> Self {
        Self {
            provider,
            binding_name,
        }
    }

    async fn vault(&self) -> Result<Arc<dyn Vault>> {
        self.provider.load_vault(&self.binding_name).await
    }

    async fn worker(&self) -> Result<Arc<dyn Worker>> {
        self.provider.load_worker(&self.binding_name).await
    }
}

fn object_store_error(source: AlienError<ErrorData>) -> object_store::Error {
    object_store::Error::Generic {
        store: OBJECT_STORE_NAME,
        source: Box::new(source),
    }
}

/// Storage handle that resolves a fresh-enough provider for every operation.
#[derive(Debug, Clone)]
pub(super) struct RefreshingStorage {
    provider: Arc<dyn StorageProviderApi>,
    binding_name: String,
    /// Storage topology does not change when credentials rotate. Capture it
    /// from the initially validated handle for the trait's synchronous calls,
    /// without retaining that handle's eventually stale credential client.
    base_dir: Path,
    url: Url,
}

impl RefreshingStorage {
    pub(super) fn new(
        provider: Arc<dyn StorageProviderApi>,
        binding_name: String,
        initial: Arc<dyn Storage>,
    ) -> Self {
        let base_dir = initial.get_base_dir();
        let url = initial.get_url();
        Self {
            provider,
            binding_name,
            base_dir,
            url,
        }
    }

    async fn current(&self) -> object_store::Result<Arc<dyn Storage>> {
        self.provider
            .load_storage(&self.binding_name)
            .await
            .map_err(object_store_error)
    }
}

impl fmt::Display for RefreshingStorage {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "Alien storage binding '{}'", self.binding_name)
    }
}

impl Binding for RefreshingStorage {}

#[async_trait]
impl Storage for RefreshingStorage {
    fn get_base_dir(&self) -> Path {
        self.base_dir.clone()
    }

    fn get_url(&self) -> Url {
        self.url.clone()
    }

    async fn presigned_put(&self, path: &Path, expires_in: Duration) -> Result<PresignedRequest> {
        self.provider
            .load_storage(&self.binding_name)
            .await?
            .presigned_put(path, expires_in)
            .await
    }

    async fn presigned_get(&self, path: &Path, expires_in: Duration) -> Result<PresignedRequest> {
        self.provider
            .load_storage(&self.binding_name)
            .await?
            .presigned_get(path, expires_in)
            .await
    }

    async fn presigned_delete(
        &self,
        path: &Path,
        expires_in: Duration,
    ) -> Result<PresignedRequest> {
        self.provider
            .load_storage(&self.binding_name)
            .await?
            .presigned_delete(path, expires_in)
            .await
    }
}

#[async_trait]
impl ObjectStore for RefreshingStorage {
    async fn put(&self, location: &Path, payload: PutPayload) -> object_store::Result<PutResult> {
        self.current().await?.put(location, payload).await
    }

    async fn put_opts(
        &self,
        location: &Path,
        payload: PutPayload,
        options: ObjectStorePutOptions,
    ) -> object_store::Result<PutResult> {
        self.current()
            .await?
            .put_opts(location, payload, options)
            .await
    }

    async fn put_multipart(
        &self,
        location: &Path,
    ) -> object_store::Result<Box<dyn MultipartUpload>> {
        self.current().await?.put_multipart(location).await
    }

    async fn put_multipart_opts(
        &self,
        location: &Path,
        options: PutMultipartOptions,
    ) -> object_store::Result<Box<dyn MultipartUpload>> {
        self.current()
            .await?
            .put_multipart_opts(location, options)
            .await
    }

    async fn get(&self, location: &Path) -> object_store::Result<GetResult> {
        self.current().await?.get(location).await
    }

    async fn get_opts(
        &self,
        location: &Path,
        options: GetOptions,
    ) -> object_store::Result<GetResult> {
        self.current().await?.get_opts(location, options).await
    }

    async fn get_range(&self, location: &Path, range: Range<u64>) -> object_store::Result<Bytes> {
        self.current().await?.get_range(location, range).await
    }

    async fn get_ranges(
        &self,
        location: &Path,
        ranges: &[Range<u64>],
    ) -> object_store::Result<Vec<Bytes>> {
        self.current().await?.get_ranges(location, ranges).await
    }

    async fn head(&self, location: &Path) -> object_store::Result<ObjectMeta> {
        self.current().await?.head(location).await
    }

    async fn delete(&self, location: &Path) -> object_store::Result<()> {
        self.current().await?.delete(location).await
    }

    fn list(&self, prefix: Option<&Path>) -> BoxStream<'static, object_store::Result<ObjectMeta>> {
        let provider = self.provider.clone();
        let binding_name = self.binding_name.clone();
        let prefix = prefix.cloned();
        stream::once(async move {
            let storage = provider
                .load_storage(&binding_name)
                .await
                .map_err(object_store_error)?;
            Ok::<BoxStream<'static, object_store::Result<ObjectMeta>>, object_store::Error>(
                storage.list(prefix.as_ref()),
            )
        })
        .try_flatten()
        .boxed()
    }

    fn list_with_offset(
        &self,
        prefix: Option<&Path>,
        offset: &Path,
    ) -> BoxStream<'static, object_store::Result<ObjectMeta>> {
        let provider = self.provider.clone();
        let binding_name = self.binding_name.clone();
        let prefix = prefix.cloned();
        let offset = offset.clone();
        stream::once(async move {
            let storage = provider
                .load_storage(&binding_name)
                .await
                .map_err(object_store_error)?;
            Ok::<BoxStream<'static, object_store::Result<ObjectMeta>>, object_store::Error>(
                storage.list_with_offset(prefix.as_ref(), &offset),
            )
        })
        .try_flatten()
        .boxed()
    }

    async fn list_with_delimiter(&self, prefix: Option<&Path>) -> object_store::Result<ListResult> {
        self.current().await?.list_with_delimiter(prefix).await
    }

    async fn copy(&self, from: &Path, to: &Path) -> object_store::Result<()> {
        self.current().await?.copy(from, to).await
    }

    async fn rename(&self, from: &Path, to: &Path) -> object_store::Result<()> {
        self.current().await?.rename(from, to).await
    }

    async fn copy_if_not_exists(&self, from: &Path, to: &Path) -> object_store::Result<()> {
        self.current().await?.copy_if_not_exists(from, to).await
    }

    async fn rename_if_not_exists(&self, from: &Path, to: &Path) -> object_store::Result<()> {
        self.current().await?.rename_if_not_exists(from, to).await
    }
}

#[async_trait]
#[cfg(feature = "platform-sdk")]
impl RemoteStorage for RefreshingStorage {
    async fn get(&self, path: &Path) -> object_store::Result<GetResult> {
        ObjectStore::get(self, path).await
    }

    async fn get_opts(&self, path: &Path, options: GetOptions) -> object_store::Result<GetResult> {
        ObjectStore::get_opts(self, path, options).await
    }

    async fn put(&self, path: &Path, payload: PutPayload) -> object_store::Result<PutResult> {
        ObjectStore::put(self, path, payload).await
    }

    async fn put_opts(
        &self,
        path: &Path,
        payload: PutPayload,
        options: ObjectStorePutOptions,
    ) -> object_store::Result<PutResult> {
        ObjectStore::put_opts(self, path, payload, options).await
    }

    async fn head(&self, path: &Path) -> object_store::Result<ObjectMeta> {
        ObjectStore::head(self, path).await
    }

    async fn delete(&self, path: &Path) -> object_store::Result<()> {
        ObjectStore::delete(self, path).await
    }

    fn list(&self, prefix: Option<&Path>) -> BoxStream<'static, object_store::Result<ObjectMeta>> {
        ObjectStore::list(self, prefix)
    }
}

/// Key-value handle that resolves a fresh-enough provider for every operation.
#[derive(Debug)]
pub(super) struct RefreshingKv {
    provider: Arc<dyn KvProviderApi>,
    binding_name: String,
}

impl RefreshingKv {
    pub(super) fn new(provider: Arc<dyn KvProviderApi>, binding_name: String) -> Self {
        Self {
            provider,
            binding_name,
        }
    }

    async fn current(&self) -> Result<Arc<dyn Kv>> {
        self.provider.load_kv(&self.binding_name).await
    }
}

impl Binding for RefreshingKv {}

#[async_trait]
impl Kv for RefreshingKv {
    async fn get(&self, key: &str) -> Result<Option<KvEntry>> {
        self.current().await?.get(key).await
    }

    async fn put(&self, key: &str, value: Vec<u8>, options: Option<KvPutOptions>) -> Result<bool> {
        self.current().await?.put(key, value, options).await
    }

    async fn delete(&self, key: &str, if_version: Option<&str>) -> Result<bool> {
        self.current().await?.delete(key, if_version).await
    }

    async fn exists(&self, key: &str) -> Result<bool> {
        self.current().await?.exists(key).await
    }

    async fn scan_prefix(
        &self,
        prefix: &str,
        limit: Option<usize>,
        cursor: Option<String>,
    ) -> Result<ScanResult> {
        self.current()
            .await?
            .scan_prefix(prefix, limit, cursor)
            .await
    }
}

/// Queue handle that resolves a fresh-enough provider for every operation.
#[derive(Debug)]
pub(super) struct RefreshingQueue {
    provider: Arc<dyn QueueProviderApi>,
    binding_name: String,
}

impl RefreshingQueue {
    pub(super) fn new(provider: Arc<dyn QueueProviderApi>, binding_name: String) -> Self {
        Self {
            provider,
            binding_name,
        }
    }
}

impl RefreshingQueue {
    async fn current(&self) -> Result<Arc<dyn Queue>> {
        self.provider.load_queue(&self.binding_name).await
    }
}

impl Binding for RefreshingQueue {}

#[async_trait]
impl Queue for RefreshingQueue {
    async fn send(&self, queue: &str, message: MessagePayload) -> Result<()> {
        self.current().await?.send(queue, message).await
    }

    async fn send_batch(
        &self,
        queue: &str,
        messages: Vec<MessagePayload>,
    ) -> Result<Vec<crate::traits::QueueSendResult>> {
        self.current().await?.send_batch(queue, messages).await
    }

    async fn receive(&self, queue: &str, max_messages: usize) -> Result<Vec<QueueMessage>> {
        self.current().await?.receive(queue, max_messages).await
    }

    async fn ack(&self, queue: &str, receipt_handle: &str) -> Result<()> {
        self.current().await?.ack(queue, receipt_handle).await
    }

    async fn nack(&self, queue: &str, receipt_handle: &str) -> Result<()> {
        self.current().await?.nack(queue, receipt_handle).await
    }

    async fn purge(&self, queue: &str) -> Result<()> {
        self.current().await?.purge(queue).await
    }
}

/// Key handle that resolves a fresh-enough provider for every operation.
#[derive(Debug)]
pub(super) struct RefreshingKey {
    provider: Arc<dyn KeyProviderApi>,
    binding_name: String,
}

impl RefreshingKey {
    pub(super) fn new(provider: Arc<dyn KeyProviderApi>, binding_name: String) -> Self {
        Self {
            provider,
            binding_name,
        }
    }
}

impl Binding for RefreshingKey {}

#[async_trait]
impl Key for RefreshingKey {
    async fn encrypt(
        &self,
        plaintext: &[u8],
        context: Option<&std::collections::BTreeMap<String, String>>,
    ) -> Result<Vec<u8>> {
        self.provider
            .load_key(&self.binding_name)
            .await?
            .encrypt(plaintext, context)
            .await
    }

    async fn decrypt(
        &self,
        ciphertext: &[u8],
        context: Option<&std::collections::BTreeMap<String, String>>,
    ) -> Result<Vec<u8>> {
        self.provider
            .load_key(&self.binding_name)
            .await?
            .decrypt(ciphertext, context)
            .await
    }
}

/// Vault handle that resolves a fresh-enough provider for every operation.
#[derive(Debug)]
pub(super) struct RefreshingVault {
    resolver: Resolver,
}

impl RefreshingVault {
    pub(super) fn new(provider: Arc<dyn BindingsProviderApi>, binding_name: String) -> Self {
        Self {
            resolver: Resolver::new(provider, binding_name),
        }
    }
}

impl Binding for RefreshingVault {}

#[async_trait]
impl Vault for RefreshingVault {
    async fn get_secret(&self, secret_name: &str) -> Result<String> {
        self.resolver.vault().await?.get_secret(secret_name).await
    }

    async fn secret_presence(&self, secret_name: &str) -> Result<crate::traits::SecretPresence> {
        self.resolver
            .vault()
            .await?
            .secret_presence(secret_name)
            .await
    }

    async fn set_secret(&self, secret_name: &str, value: &str) -> Result<()> {
        self.resolver
            .vault()
            .await?
            .set_secret(secret_name, value)
            .await
    }

    async fn delete_secret(&self, secret_name: &str) -> Result<()> {
        self.resolver
            .vault()
            .await?
            .delete_secret(secret_name)
            .await
    }

    async fn list_secrets(&self) -> Result<Vec<String>> {
        self.resolver.vault().await?.list_secrets().await
    }
}

/// Worker handle that resolves a fresh-enough provider for every operation.
#[derive(Debug)]
pub(super) struct RefreshingWorker {
    resolver: Resolver,
}

impl RefreshingWorker {
    pub(super) fn new(provider: Arc<dyn BindingsProviderApi>, binding_name: String) -> Self {
        Self {
            resolver: Resolver::new(provider, binding_name),
        }
    }
}

impl Binding for RefreshingWorker {}

#[async_trait]
impl Worker for RefreshingWorker {
    async fn invoke(&self, request: WorkerInvokeRequest) -> Result<WorkerInvokeResponse> {
        self.resolver.worker().await?.invoke(request).await
    }

    async fn get_worker_url(&self) -> Result<Option<String>> {
        self.resolver.worker().await?.get_worker_url().await
    }

    fn as_any(&self) -> &dyn std::any::Any {
        self
    }
}