dapr 0.19.0-rc.0

use crate::dapr::proto::common::v1::JobFailurePolicyConstant;
use crate::dapr::proto::common::v1::job_failure_policy::Policy;
use crate::dapr::proto::{common::v1 as common_v1, runtime::v1 as dapr_v1};
use crate::error::Error;
#[cfg(feature = "workflow")]
use crate::workflow;
use async_trait::async_trait;
use futures::StreamExt;
use prost_types::Any;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::collections::HashMap;
use std::time::Duration;
use tokio::io::AsyncRead;
use tonic::codegen::tokio_stream;
use tonic::service::interceptor::InterceptedService;
use tonic::{Request, transport::Channel as TonicChannel};
use tonic::{Status, Streaming};

pub mod config;
pub mod interceptor;

pub use config::{
    API_TOKEN_METADATA_KEY, APP_API_TOKEN_ENV, ClientOptions, DAPR_API_TOKEN_ENV,
    DAPR_CLIENT_TIMEOUT_SECONDS_ENV, DAPR_GRPC_ENDPOINT_ENV, DAPR_GRPC_PORT_ENV,
    DEFAULT_CLIENT_TIMEOUT_SECONDS, DEFAULT_DAPR_GRPC_PORT, default_sidecar_address,
};
pub use interceptor::{ApiTokenInterceptor, AppApiTokenLayer, AppApiTokenService};

#[derive(Clone)]
pub struct Client<T>(T, String);

impl<T: DaprInterface> Client<T> {
    /// Connect to a Dapr enabled app.
    ///
    /// # Deprecated
    ///
    /// Will be removed in `0.20.0`. Use [`Client::new`] for env-driven setup
    /// or [`Client::from_options`] for explicit configuration. This method
    /// requires `DAPR_GRPC_PORT` to be set and ignores `DAPR_GRPC_ENDPOINT`
    /// and `DAPR_API_TOKEN`.
    ///
    /// # Arguments
    ///
    /// * `addr` - Address of gRPC server to connect to.
    #[deprecated(
        since = "0.19.0",
        note = "Will be removed in 0.20.0. Use Client::new() or Client::from_options()."
    )]
    pub async fn connect(addr: String) -> Result<Self, Error> {
        // Get the Dapr port to create a connection
        let port: u16 = std::env::var("DAPR_GRPC_PORT")?.parse()?;
        let address = format!("{addr}:{port}");

        Ok(Client(T::connect(address.clone()).await?, address))
    }

    /// Connect to the Dapr sidecar with a specific port.
    ///
    /// # Deprecated
    ///
    /// Will be removed in `0.20.0`. Use [`Client::new`] or
    /// [`Client::from_options`] instead.
    ///
    /// # Arguments
    ///
    /// * `addr` - Address of gRPC server to connect to.
    /// * `port` - Port of the gRPC server to connect to.
    #[deprecated(
        since = "0.19.0",
        note = "Will be removed in 0.20.0. Use Client::new() or Client::from_options()."
    )]
    pub async fn connect_with_port(addr: String, port: String) -> Result<Self, Error> {
        // assert that port is between 1 and 65535
        let port: u16 = match port.parse::<u16>() {
            Ok(p) => p,
            Err(_) => {
                panic!("Port must be a number between 1 and 65535");
            }
        };

        let address = format!("{addr}:{port}");

        Ok(Client(T::connect(address.clone()).await?, address))
    }

    /// Create a workflow client connected to the same Dapr sidecar.
    #[cfg(feature = "workflow")]
    pub async fn new_workflow_client(&self) -> workflow::Result<workflow::WorkflowClient> {
        workflow::WorkflowClient::new_with_address(self.1.clone()).await
    }

    /// Invoke a method in a Dapr enabled app.
    ///
    /// # Arguments
    ///
    /// * `app_id` - Id of the application running.
    /// * `method_name` - Name of the method to invoke.
    /// * `data` - Required. Bytes value or data required to invoke service.
    pub async fn invoke_service<I, M>(
        &mut self,
        app_id: I,
        method_name: M,
        data: Option<Any>,
    ) -> Result<InvokeServiceResponse, Error>
    where
        I: Into<String>,
        M: Into<String>,
    {
        self.0
            .invoke_service(InvokeServiceRequest {
                id: app_id.into(),
                message: common_v1::InvokeRequest {
                    method: method_name.into(),
                    data,
                    ..Default::default()
                }
                .into(),
            })
            .await
    }

    /// Invoke a Dapr output binding.
    ///
    /// # Arguments
    ///
    /// * `name` - The name of the output binding to invoke.
    /// * `data` - The data which will be sent to the output binding.
    /// * `operation` - The operation name for the binding to invoke.
    /// * `metadata` - The metadata key-value pairs to be sent to the binding.
    pub async fn invoke_binding<S>(
        &mut self,
        name: S,
        data: Vec<u8>,
        operation: S,
        metadata: Option<HashMap<String, String>>,
    ) -> Result<InvokeBindingResponse, Error>
    where
        S: Into<String>,
    {
        self.0
            .invoke_binding(InvokeBindingRequest {
                name: name.into(),
                data,
                operation: operation.into(),
                metadata: metadata.unwrap_or_default(),
            })
            .await
    }

    /// Invoke a Dapr output binding without expecting a response payload.
    ///
    /// # Arguments
    ///
    /// * `name` - The name of the output binding to invoke.
    /// * `operation` - The operation name for the binding to invoke.
    pub async fn invoke_output_binding<S>(&mut self, name: S, operation: S) -> Result<(), Error>
    where
        S: Into<String>,
    {
        self.0
            .invoke_binding(InvokeBindingRequest {
                name: name.into(),
                operation: operation.into(),
                ..Default::default()
            })
            .await
            .map(|_| ())
    }

    /// Publish a payload to multiple consumers who are listening on a topic.
    ///
    /// Dapr guarantees at least once semantics for this endpoint.
    ///
    /// # Arguments
    ///
    /// * `pubsub_name` - Name of the pubsub component
    /// * `topic` - Pubsub topic.
    /// * `data` - The data which will be published to topic.
    pub async fn publish_event<S>(
        &mut self,
        pubsub_name: S,
        topic: S,
        data_content_type: S,
        data: Vec<u8>,
        metadata: Option<HashMap<String, String>>,
    ) -> Result<(), Error>
    where
        S: Into<String>,
    {
        let mut mdata = HashMap::<String, String>::new();
        if let Some(m) = metadata {
            mdata = m;
        }
        self.0
            .publish_event(PublishEventRequest {
                pubsub_name: pubsub_name.into(),
                topic: topic.into(),
                data_content_type: data_content_type.into(),
                data,
                metadata: mdata,
            })
            .await
    }

    /// Get the secret for a specific key.
    ///
    /// # Arguments
    ///
    /// * `store_name` - The name of secret store.
    /// * `key` - The name of secret key.
    pub async fn get_secret<S>(&mut self, store_name: S, key: S) -> Result<GetSecretResponse, Error>
    where
        S: Into<String>,
    {
        self.0
            .get_secret(GetSecretRequest {
                store_name: store_name.into(),
                key: key.into(),
                ..Default::default()
            })
            .await
    }

    /// Get all secrets for a given store
    ///
    /// # Arguments
    ///
    /// * `store_name` - The name of the secret store.
    pub async fn get_bulk_secret<S>(
        &mut self,
        store_name: S,
        metadata: Option<HashMap<String, String>>,
    ) -> Result<GetBulkSecretResponse, Error>
    where
        S: Into<String>,
    {
        self.0
            .get_bulk_secret(GetBulkSecretRequest {
                store_name: store_name.into(),
                metadata: metadata.unwrap_or_default(),
            })
            .await
    }

    /// Get the state for a specific key.
    ///
    /// # Arguments
    ///
    /// * `store_name` - The name of state store.
    /// * `key` - The key of the desired state.
    /// * `metadata` - Any metadata pairs to include in the request.
    pub async fn get_state<S>(
        &mut self,
        store_name: S,
        key: S,
        metadata: Option<HashMap<String, String>>,
    ) -> Result<GetStateResponse, Error>
    where
        S: Into<String>,
    {
        let mut mdata = HashMap::<String, String>::new();
        if let Some(m) = metadata {
            mdata = m;
        }

        self.0
            .get_state(GetStateRequest {
                store_name: store_name.into(),
                key: key.into(),
                metadata: mdata,
                ..Default::default()
            })
            .await
    }

    /// Save an array of state objects.
    ///
    /// This does not include any etag or metadata options.
    ///
    /// # Arguments
    ///
    /// * `store_name` - The name of state store.
    /// * `key` - The key for the value
    /// * `value` - The value to be saved for the key
    /// * `etag` - The etag identifier
    /// * `metadata` - Any metadata pairs to include in the request.
    /// * `options` - Any state option
    pub async fn save_state<S>(
        &mut self,
        store_name: S,
        key: S,
        value: Vec<u8>,
        etag: Option<Etag>,
        metadata: Option<HashMap<String, String>>,
        options: Option<StateOptions>,
    ) -> Result<(), Error>
    where
        S: Into<String>,
    {
        let states = vec![StateItem {
            key: key.into(),
            value,
            etag,
            metadata: metadata.unwrap_or_default(),
            options,
        }];

        self.save_bulk_states(store_name, states).await
    }

    /// Save an array of state objects.
    ///
    /// # Arguments
    ///
    /// * `store_name` - The name of state store.
    /// * `items` - The array of the state items.
    pub async fn save_bulk_states<S, I>(&mut self, store_name: S, items: I) -> Result<(), Error>
    where
        S: Into<String>,
        I: Into<Vec<StateItem>>,
    {
        self.0
            .save_state(SaveStateRequest {
                store_name: store_name.into(),
                states: items.into(),
            })
            .await
    }

    /// Query state objects based on specific query conditions
    ///
    /// # Arguments
    ///
    /// * `store_name` - The name of state store.
    /// * `query` - The query request (json)
    pub async fn query_state_alpha1<S>(
        &mut self,
        store_name: S,
        query: Value,
        metadata: Option<HashMap<String, String>>,
    ) -> Result<QueryStateResponse, Error>
    where
        S: Into<String>,
    {
        let mut mdata = HashMap::<String, String>::new();
        if let Some(m) = metadata {
            mdata = m;
        }

        self.0
            .query_state_alpha1(QueryStateRequest {
                store_name: store_name.into(),
                query: serde_json::to_string(&query).unwrap(),
                metadata: mdata,
            })
            .await
    }

    /// Delete an array of state objects.
    ///
    /// # Arguments
    ///
    /// * `store_name` - The name of state store.
    /// * `states` - The array of the state key values.
    pub async fn delete_bulk_state<I, K>(&mut self, store_name: K, states: I) -> Result<(), Error>
    where
        I: IntoIterator<Item = (K, Vec<u8>)>,
        K: Into<String>,
    {
        self.0
            .delete_bulk_state(DeleteBulkStateRequest {
                store_name: store_name.into(),
                states: states.into_iter().map(|pair| pair.into()).collect(),
            })
            .await
    }

    /// Delete the state for a specific key.
    ///
    /// # Arguments
    ///
    /// * `store_name` - The name of state store.
    /// * `key` - The key of the desired state.
    pub async fn delete_state<S>(
        &mut self,
        store_name: S,
        key: S,
        metadata: Option<HashMap<String, String>>,
    ) -> Result<(), Error>
    where
        S: Into<String>,
    {
        let mut mdata = HashMap::<String, String>::new();
        if let Some(m) = metadata {
            mdata = m;
        }

        self.0
            .delete_state(DeleteStateRequest {
                store_name: store_name.into(),
                key: key.into(),
                metadata: mdata,
                ..Default::default()
            })
            .await
    }

    /// Set sidecar Metadata
    ///
    /// # Arguments
    ///
    /// * `key` - The metadata key
    /// * `value` - The metadata value
    pub async fn set_metadata<S>(&mut self, key: S, value: S) -> Result<(), Error>
    where
        S: Into<String>,
    {
        self.0
            .set_metadata(SetMetadataRequest {
                key: key.into(),
                value: value.into(),
            })
            .await
    }

    /// Set sidecar Metadata
    ///
    pub async fn get_metadata(&mut self) -> Result<GetMetadataResponse, Error> {
        self.0.get_metadata().await
    }

    /// Invoke a method in a Dapr actor.
    ///
    /// # Arguments
    ///
    /// * `actor_type` - Type of the actor.
    /// * `actor_id` - Id of the actor.
    /// * `method_name` - Name of the method to invoke.
    /// * `input` - Required. Data required to invoke service, should be json serializable.
    pub async fn invoke_actor<I, M, TInput, TOutput>(
        &mut self,
        actor_type: I,
        actor_id: I,
        method_name: M,
        input: TInput,
        metadata: Option<HashMap<String, String>>,
    ) -> Result<TOutput, Error>
    where
        I: Into<String>,
        M: Into<String>,
        TInput: Serialize,
        TOutput: for<'a> Deserialize<'a>,
    {
        let mut mdata = HashMap::<String, String>::new();
        if let Some(m) = metadata {
            mdata = m;
        }

        mdata.insert("Content-Type".to_string(), "application/json".to_string());

        let data = match serde_json::to_vec(&input) {
            Ok(data) => data,
            Err(_e) => return Err(Error::SerializationError),
        };

        let res = self
            .0
            .invoke_actor(InvokeActorRequest {
                actor_type: actor_type.into(),
                actor_id: actor_id.into(),
                method: method_name.into(),
                data,
                metadata: mdata,
            })
            .await?;

        match serde_json::from_slice::<TOutput>(&res.data) {
            Ok(output) => Ok(output),
            Err(_e) => Err(Error::SerializationError),
        }
    }

    /// Get the configuration for a specific key
    /// ///
    /// # Arguments
    ///
    /// * `store_name` - The name of config store.
    /// * `keys` - The key of the desired configuration.
    pub async fn get_configuration<S, K>(
        &mut self,
        store_name: S,
        keys: Vec<K>,
        metadata: Option<HashMap<String, String>>,
    ) -> Result<GetConfigurationResponse, Error>
    where
        S: Into<String>,
        K: Into<String>,
    {
        let request = GetConfigurationRequest {
            store_name: store_name.into(),
            keys: keys.into_iter().map(|key| key.into()).collect(),
            metadata: metadata.unwrap_or_default(),
        };
        self.0.get_configuration(request).await
    }

    /// Subscribe to configuration changes
    pub async fn subscribe_configuration<S>(
        &mut self,
        store_name: S,
        keys: Vec<S>,
        metadata: Option<HashMap<String, String>>,
    ) -> Result<Streaming<SubscribeConfigurationResponse>, Error>
    where
        S: Into<String>,
    {
        let request = SubscribeConfigurationRequest {
            store_name: store_name.into(),
            keys: keys.into_iter().map(|key| key.into()).collect(),
            metadata: metadata.unwrap_or_default(),
        };
        self.0.subscribe_configuration(request).await
    }

    /// Unsubscribe from configuration changes
    pub async fn unsubscribe_configuration<S>(
        &mut self,
        store_name: S,
        id: S,
    ) -> Result<UnsubscribeConfigurationResponse, Error>
    where
        S: Into<String>,
    {
        let request = UnsubscribeConfigurationRequest {
            id: id.into(),
            store_name: store_name.into(),
        };
        self.0.unsubscribe_configuration(request).await
    }

    /// Encrypt binary data using Dapr. returns `Vec<StreamPayload>` to be used in decrypt method
    ///
    /// # Arguments
    ///
    /// * `payload` - ReaderStream to the data to encrypt
    /// * `request_option` - Encryption request options.
    pub async fn encrypt<R>(
        &mut self,
        payload: ReaderStream<R>,
        request_options: EncryptRequestOptions,
    ) -> Result<Vec<StreamPayload>, Status>
    where
        R: AsyncRead + Send,
    {
        // have to have it as a reference for the async move below
        let request_options = &Some(request_options);
        let requested_items: Vec<EncryptRequest> = payload
            .0
            .enumerate()
            .fold(vec![], |mut init, (i, bytes)| async move {
                let stream_payload = StreamPayload {
                    data: bytes.unwrap().to_vec(),
                    seq: 0,
                };
                if i == 0 {
                    init.push(EncryptRequest {
                        options: request_options.clone(),
                        payload: Some(stream_payload),
                    });
                } else {
                    init.push(EncryptRequest {
                        options: None,
                        payload: Some(stream_payload),
                    });
                }
                init
            })
            .await;
        self.0.encrypt(requested_items).await
    }

    /// Decrypt binary data using Dapr. returns `Vec<u8>`.
    ///
    /// # Arguments
    ///
    /// * `encrypted` - Encrypted data usually returned from encrypted, `Vec<StreamPayload>`
    /// * `options` - Decryption request options.
    pub async fn decrypt(
        &mut self,
        encrypted: Vec<StreamPayload>,
        options: DecryptRequestOptions,
    ) -> Result<Vec<u8>, Status> {
        let requested_items: Vec<DecryptRequest> = encrypted
            .iter()
            .enumerate()
            .map(|(i, item)| {
                if i == 0 {
                    DecryptRequest {
                        options: Some(options.clone()),
                        payload: Some(item.clone()),
                    }
                } else {
                    DecryptRequest {
                        options: None,
                        payload: Some(item.clone()),
                    }
                }
            })
            .collect();
        self.0.decrypt(requested_items).await
    }

    /// Schedules a job with the Dapr Distributed Scheduler
    ///
    /// # Arguments
    ///
    /// * job - The job to schedule
    /// * overwrite - Optional flag to overwrite an existing job with the same name
    pub async fn schedule_job_alpha1(
        &mut self,
        job: Job,
        overwrite: Option<bool>,
    ) -> Result<ScheduleJobResponse, Error> {
        let request = ScheduleJobRequest {
            job: Some(job.clone()),
            overwrite: overwrite.unwrap_or(false),
        };
        self.0.schedule_job_alpha1(request).await
    }

    /// Retrieves a scheduled job from the Dapr Distributed Scheduler
    ///
    /// # Arguments
    ///
    /// * name - The name of the job to retrieve
    pub async fn get_job_alpha1(&mut self, name: &str) -> Result<GetJobResponse, Error> {
        let request = GetJobRequest {
            name: name.to_string(),
        };
        self.0.get_job_alpha1(request).await
    }

    /// Deletes a scheduled job from the Dapr Distributed Scheduler
    ///
    /// # Arguments
    ///
    /// * name - The name of the job to delete
    pub async fn delete_job_alpha1(&mut self, name: &str) -> Result<DeleteJobResponse, Error> {
        let request = DeleteJobRequest {
            name: name.to_string(),
        };
        self.0.delete_job_alpha1(request).await
    }

    /// Converse with an LLM
    ///
    /// # Arguments
    ///
    /// * ConversationRequest - The request containing inputs to send to the LLM
    pub async fn converse_alpha1(
        &mut self,
        request: ConversationRequest,
    ) -> Result<ConversationResponse, Error> {
        self.0.converse_alpha1(request).await
    }

    /// Converse with an LLM using alpha2
    ///
    /// # Arguments
    ///
    /// * ConversationRequestAlpha2 - The request containing inputs to send to the LLM
    pub async fn converse_alpha2(
        &mut self,
        request: ConversationRequestAlpha2,
    ) -> Result<ConversationResponseAlpha2, Error> {
        self.0.converse_alpha2(request).await
    }
}

#[async_trait]
pub trait DaprInterface: Sized {
    async fn connect(addr: String) -> Result<Self, Error>;
    async fn publish_event(&mut self, request: PublishEventRequest) -> Result<(), Error>;
    async fn invoke_service(
        &mut self,
        request: InvokeServiceRequest,
    ) -> Result<InvokeServiceResponse, Error>;
    async fn invoke_binding(
        &mut self,
        request: InvokeBindingRequest,
    ) -> Result<InvokeBindingResponse, Error>;
    async fn get_secret(&mut self, request: GetSecretRequest) -> Result<GetSecretResponse, Error>;
    async fn get_bulk_secret(
        &mut self,
        request: GetBulkSecretRequest,
    ) -> Result<GetBulkSecretResponse, Error>;
    async fn get_state(&mut self, request: GetStateRequest) -> Result<GetStateResponse, Error>;
    async fn save_state(&mut self, request: SaveStateRequest) -> Result<(), Error>;
    async fn query_state_alpha1(
        &mut self,
        request: QueryStateRequest,
    ) -> Result<QueryStateResponse, Error>;
    async fn delete_state(&mut self, request: DeleteStateRequest) -> Result<(), Error>;
    async fn delete_bulk_state(&mut self, request: DeleteBulkStateRequest) -> Result<(), Error>;
    async fn set_metadata(&mut self, request: SetMetadataRequest) -> Result<(), Error>;
    async fn get_metadata(&mut self) -> Result<GetMetadataResponse, Error>;
    async fn invoke_actor(
        &mut self,
        request: InvokeActorRequest,
    ) -> Result<InvokeActorResponse, Error>;
    async fn get_configuration(
        &mut self,
        request: GetConfigurationRequest,
    ) -> Result<GetConfigurationResponse, Error>;
    async fn subscribe_configuration(
        &mut self,
        request: SubscribeConfigurationRequest,
    ) -> Result<Streaming<SubscribeConfigurationResponse>, Error>;
    async fn unsubscribe_configuration(
        &mut self,
        request: UnsubscribeConfigurationRequest,
    ) -> Result<UnsubscribeConfigurationResponse, Error>;

    async fn encrypt(&mut self, payload: Vec<EncryptRequest>)
    -> Result<Vec<StreamPayload>, Status>;

    async fn decrypt(&mut self, payload: Vec<DecryptRequest>) -> Result<Vec<u8>, Status>;

    async fn schedule_job_alpha1(
        &mut self,
        request: ScheduleJobRequest,
    ) -> Result<ScheduleJobResponse, Error>;

    async fn get_job_alpha1(&mut self, request: GetJobRequest) -> Result<GetJobResponse, Error>;

    async fn delete_job_alpha1(
        &mut self,
        request: DeleteJobRequest,
    ) -> Result<DeleteJobResponse, Error>;

    async fn converse_alpha1(
        &mut self,
        request: ConversationRequest,
    ) -> Result<ConversationResponse, Error>;
    async fn converse_alpha2(
        &mut self,
        request: ConversationRequestAlpha2,
    ) -> Result<ConversationResponseAlpha2, Error>;
}

// Implement `DaprInterface` for both the plain and intercepted tonic clients.
// The actual RPC method bodies are identical (the generated tonic client has
// the same method names regardless of the inner transport), so they're
// generated by a macro. `connect` is supplied via a free function rather
// than a closure to avoid expression-position parser ambiguity in the macro.
async fn connect_plain(
    addr: String,
) -> Result<dapr_v1::dapr_client::DaprClient<TonicChannel>, Error> {
    Ok(dapr_v1::dapr_client::DaprClient::connect(addr).await?)
}

async fn connect_intercepted(
    addr: String,
) -> Result<
    dapr_v1::dapr_client::DaprClient<InterceptedService<TonicChannel, ApiTokenInterceptor>>,
    Error,
> {
    // The intercepted variant cannot be constructed from just an address
    // because the interceptor needs to be supplied. Use [`Client::new`] /
    // [`Client::from_options`] / [`Client::connect_with_address`] instead,
    // which build the channel and interceptor together.
    Err(Error::InvalidEndpoint(
        crate::error::sanitize_endpoint_for_diagnostics(&addr),
    ))
}

macro_rules! impl_dapr_interface_for {
    ($type:ty, $connect_fn:path) => {
        #[async_trait]
        impl DaprInterface for $type {
            async fn connect(addr: String) -> Result<Self, Error> {
                $connect_fn(addr).await
            }

            async fn publish_event(&mut self, request: PublishEventRequest) -> Result<(), Error> {
                self.publish_event(Request::new(request))
                    .await?
                    .into_inner();
                Ok(())
            }

            async fn invoke_service(
                &mut self,
                request: InvokeServiceRequest,
            ) -> Result<InvokeServiceResponse, Error> {
                Ok(self
                    .invoke_service(Request::new(request))
                    .await?
                    .into_inner())
            }

            async fn invoke_binding(
                &mut self,
                request: InvokeBindingRequest,
            ) -> Result<InvokeBindingResponse, Error> {
                Ok(self
                    .invoke_binding(Request::new(request))
                    .await?
                    .into_inner())
            }

            async fn get_secret(
                &mut self,
                request: GetSecretRequest,
            ) -> Result<GetSecretResponse, Error> {
                Ok(self.get_secret(Request::new(request)).await?.into_inner())
            }

            async fn get_bulk_secret(
                &mut self,
                request: GetBulkSecretRequest,
            ) -> Result<GetBulkSecretResponse, Error> {
                Ok(self
                    .get_bulk_secret(Request::new(request))
                    .await?
                    .into_inner())
            }

            async fn get_state(
                &mut self,
                request: GetStateRequest,
            ) -> Result<GetStateResponse, Error> {
                Ok(self.get_state(Request::new(request)).await?.into_inner())
            }

            async fn save_state(&mut self, request: SaveStateRequest) -> Result<(), Error> {
                self.save_state(Request::new(request)).await?.into_inner();
                Ok(())
            }

            async fn query_state_alpha1(
                &mut self,
                request: QueryStateRequest,
            ) -> Result<QueryStateResponse, Error> {
                Ok(self
                    .query_state_alpha1(Request::new(request))
                    .await?
                    .into_inner())
            }

            async fn delete_state(&mut self, request: DeleteStateRequest) -> Result<(), Error> {
                self.delete_state(Request::new(request)).await?.into_inner();
                Ok(())
            }

            async fn delete_bulk_state(
                &mut self,
                request: DeleteBulkStateRequest,
            ) -> Result<(), Error> {
                self.delete_bulk_state(Request::new(request))
                    .await?
                    .into_inner();
                Ok(())
            }

            async fn set_metadata(&mut self, request: SetMetadataRequest) -> Result<(), Error> {
                self.set_metadata(Request::new(request)).await?.into_inner();
                Ok(())
            }

            async fn get_metadata(&mut self) -> Result<GetMetadataResponse, Error> {
                Ok(self.get_metadata(GetMetadataRequest {}).await?.into_inner())
            }

            async fn invoke_actor(
                &mut self,
                request: InvokeActorRequest,
            ) -> Result<InvokeActorResponse, Error> {
                Ok(self.invoke_actor(Request::new(request)).await?.into_inner())
            }

            async fn get_configuration(
                &mut self,
                request: GetConfigurationRequest,
            ) -> Result<GetConfigurationResponse, Error> {
                Ok(self
                    .get_configuration(Request::new(request))
                    .await?
                    .into_inner())
            }

            async fn subscribe_configuration(
                &mut self,
                request: SubscribeConfigurationRequest,
            ) -> Result<Streaming<SubscribeConfigurationResponse>, Error> {
                Ok(self
                    .subscribe_configuration(Request::new(request))
                    .await?
                    .into_inner())
            }

            async fn unsubscribe_configuration(
                &mut self,
                request: UnsubscribeConfigurationRequest,
            ) -> Result<UnsubscribeConfigurationResponse, Error> {
                Ok(self
                    .unsubscribe_configuration(Request::new(request))
                    .await?
                    .into_inner())
            }

            async fn encrypt(
                &mut self,
                request: Vec<EncryptRequest>,
            ) -> Result<Vec<StreamPayload>, Status> {
                let request = Request::new(tokio_stream::iter(request));
                let stream = self.encrypt_alpha1(request).await?;
                let mut stream = stream.into_inner();
                let mut return_data = vec![];
                while let Some(resp) = stream.next().await {
                    if let Ok(resp) = resp
                        && let Some(data) = resp.payload
                    {
                        return_data.push(data)
                    }
                }
                Ok(return_data)
            }

            async fn decrypt(&mut self, request: Vec<DecryptRequest>) -> Result<Vec<u8>, Status> {
                let request = Request::new(tokio_stream::iter(request));
                let stream = self.decrypt_alpha1(request).await?;
                let mut stream = stream.into_inner();
                let mut data = vec![];
                while let Some(resp) = stream.next().await {
                    if let Ok(resp) = resp
                        && let Some(mut payload) = resp.payload
                    {
                        data.append(payload.data.as_mut())
                    }
                }
                Ok(data)
            }

            async fn schedule_job_alpha1(
                &mut self,
                request: ScheduleJobRequest,
            ) -> Result<ScheduleJobResponse, Error> {
                Ok(self.schedule_job_alpha1(request).await?.into_inner())
            }

            async fn get_job_alpha1(
                &mut self,
                request: GetJobRequest,
            ) -> Result<GetJobResponse, Error> {
                Ok(self
                    .get_job_alpha1(Request::new(request))
                    .await?
                    .into_inner())
            }

            async fn delete_job_alpha1(
                &mut self,
                request: DeleteJobRequest,
            ) -> Result<DeleteJobResponse, Error> {
                Ok(self
                    .delete_job_alpha1(Request::new(request))
                    .await?
                    .into_inner())
            }

            async fn converse_alpha1(
                &mut self,
                request: ConversationRequest,
            ) -> Result<ConversationResponse, Error> {
                Ok(self
                    .converse_alpha1(Request::new(request))
                    .await?
                    .into_inner())
            }

            async fn converse_alpha2(
                &mut self,
                request: ConversationRequestAlpha2,
            ) -> Result<ConversationResponseAlpha2, Error> {
                Ok(self
                    .converse_alpha2(Request::new(request))
                    .await?
                    .into_inner())
            }
        }
    };
}

impl_dapr_interface_for!(
    dapr_v1::dapr_client::DaprClient<TonicChannel>,
    connect_plain
);

impl_dapr_interface_for!(
    dapr_v1::dapr_client::DaprClient<InterceptedService<TonicChannel, ApiTokenInterceptor>>,
    connect_intercepted
);

/// A request from invoking a service
pub type InvokeServiceRequest = dapr_v1::InvokeServiceRequest;

/// A response from invoking a service
pub type InvokeServiceResponse = common_v1::InvokeResponse;

/// A request from invoking a binding
pub type InvokeBindingRequest = dapr_v1::InvokeBindingRequest;

/// A reponse from invoking a binding
pub type InvokeBindingResponse = dapr_v1::InvokeBindingResponse;

/// A request for publishing event
pub type PublishEventRequest = dapr_v1::PublishEventRequest;

/// A request for getting state
pub type GetStateRequest = dapr_v1::GetStateRequest;

/// A response from getting state
pub type GetStateResponse = dapr_v1::GetStateResponse;

/// A request for saving state
pub type SaveStateRequest = dapr_v1::SaveStateRequest;

/// A state item
pub type StateItem = common_v1::StateItem;

/// State options
pub type StateOptions = common_v1::StateOptions;

/// Etag identifier
pub type Etag = common_v1::Etag;

/// A request for querying state
pub type QueryStateRequest = dapr_v1::QueryStateRequest;

/// A response from querying state
pub type QueryStateResponse = dapr_v1::QueryStateResponse;

/// A request for deleting state
pub type DeleteStateRequest = dapr_v1::DeleteStateRequest;

/// A request for deleting bulk state
pub type DeleteBulkStateRequest = dapr_v1::DeleteBulkStateRequest;

/// A request for getting secret
pub type GetSecretRequest = dapr_v1::GetSecretRequest;

/// A response from getting secret
pub type GetSecretResponse = dapr_v1::GetSecretResponse;

/// A request for getting bulk secrets
pub type GetBulkSecretRequest = dapr_v1::GetBulkSecretRequest;

/// A response for getting bulk secrets
pub type GetBulkSecretResponse = dapr_v1::GetBulkSecretResponse;

/// A response from getting metadata
pub type GetMetadataResponse = dapr_v1::GetMetadataResponse;

/// A request for getting metadata
pub type GetMetadataRequest = dapr_v1::GetMetadataRequest;

/// A request for setting metadata
pub type SetMetadataRequest = dapr_v1::SetMetadataRequest;

/// A request for invoking an actor
pub type InvokeActorRequest = dapr_v1::InvokeActorRequest;

/// A response from invoking an actor
pub type InvokeActorResponse = dapr_v1::InvokeActorResponse;
/// A request for getting configuration
pub type GetConfigurationRequest = dapr_v1::GetConfigurationRequest;

/// A response from getting configuration
pub type GetConfigurationResponse = dapr_v1::GetConfigurationResponse;

/// A request for subscribing to configuration changes
pub type SubscribeConfigurationRequest = dapr_v1::SubscribeConfigurationRequest;

/// A response from subscribing tto configuration changes
pub type SubscribeConfigurationResponse = dapr_v1::SubscribeConfigurationResponse;

/// A request for unsubscribing from configuration changes
pub type UnsubscribeConfigurationRequest = dapr_v1::UnsubscribeConfigurationRequest;

/// A tonic based gRPC client without any interceptor wiring.
///
/// This is the type used by the (deprecated) [`Client::connect`] /
/// [`Client::connect_with_port`] constructors.
pub type TonicClient = dapr_v1::dapr_client::DaprClient<TonicChannel>;

/// A tonic based gRPC client wrapped with an [`ApiTokenInterceptor`] that
/// injects the configured `dapr-api-token` metadata on every outgoing call.
///
/// This is the type returned by [`Client::new`], [`Client::from_options`],
/// and [`Client::connect_with_address`]. When the configured token is empty,
/// the interceptor is a no-op, so it is safe to use this type even when no
/// token is required.
pub type TonicClientWithAuth =
    dapr_v1::dapr_client::DaprClient<InterceptedService<TonicChannel, ApiTokenInterceptor>>;

impl Client<TonicClientWithAuth> {
    /// Create a new Dapr client using configuration drawn entirely from the
    /// environment.
    ///
    /// Recognised environment variables:
    ///
    /// | Variable                          | Purpose                                            |
    /// |-----------------------------------|----------------------------------------------------|
    /// | [`DAPR_GRPC_ENDPOINT_ENV`]        | Full endpoint (preferred).                         |
    /// | [`DAPR_GRPC_PORT_ENV`]            | Port on `127.0.0.1`. Used when endpoint is unset.  |
    /// | [`DAPR_API_TOKEN_ENV`]            | Outbound `dapr-api-token` metadata value.          |
    /// | [`DAPR_CLIENT_TIMEOUT_SECONDS_ENV`] | gRPC connect timeout (default 5s).               |
    ///
    /// Default endpoint when nothing is configured: `http://127.0.0.1:50001`.
    ///
    /// # Errors
    ///
    /// Returns [`Error::InvalidEndpoint`] when the configured endpoint cannot
    /// be parsed, [`Error::ParseIntError`] when
    /// `DAPR_CLIENT_TIMEOUT_SECONDS` is malformed,
    /// [`Error::ConnectTimeout`] if the connection cannot be established
    /// within the configured timeout, or [`Error::TransportError`] for other
    /// transport failures.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # async fn run() -> Result<(), dapr::error::Error> {
    /// let mut client = dapr::Client::new().await?;
    /// # let _ = &mut client;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn new() -> Result<Self, Error> {
        let opts = ClientOptions::from_env()?;
        Self::from_options(opts).await
    }

    /// Create a new Dapr client from an explicit [`ClientOptions`].
    ///
    /// # Errors
    ///
    /// See [`Client::new`].
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # async fn run() -> Result<(), dapr::error::Error> {
    /// use std::time::Duration;
    /// use dapr::client::ClientOptions;
    ///
    /// let opts = ClientOptions::new()
    ///     .with_address("https://my-sidecar:443?tls=true")
    ///     .with_api_token("super-secret")
    ///     .with_timeout(Duration::from_secs(10));
    /// let mut client = dapr::Client::from_options(opts).await?;
    /// # let _ = &mut client;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn from_options(opts: ClientOptions) -> Result<Self, Error> {
        let address = opts.address().to_string();
        let interceptor = ApiTokenInterceptor::try_new(opts.api_token().map(|s| s.to_string()))?;

        let sanitized_address = crate::error::sanitize_endpoint_for_diagnostics(&address);
        let endpoint = tonic::transport::Endpoint::from_shared(address.clone())
            .map_err(|_| Error::InvalidEndpoint(sanitized_address.clone()))?
            .connect_timeout(opts.timeout());

        let channel = match tokio::time::timeout(opts.timeout(), endpoint.connect()).await {
            Ok(Ok(c)) => c,
            Ok(Err(e)) => return Err(Error::from(e)),
            Err(_) => return Err(Error::ConnectTimeout),
        };

        let grpc = dapr_v1::dapr_client::DaprClient::with_interceptor(channel, interceptor);
        Ok(Client(grpc, address))
    }

    /// Create a new Dapr client connected to an explicit address. All other
    /// settings are read from the environment.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # async fn run() -> Result<(), dapr::error::Error> {
    /// let mut client = dapr::Client::connect_with_address(
    ///     "https://my-sidecar:443?tls=true",
    /// ).await?;
    /// # let _ = &mut client;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn connect_with_address(address: impl Into<String>) -> Result<Self, Error> {
        let opts = ClientOptions::from_env()?.with_address(address);
        Self::from_options(opts).await
    }
}

/// A response from unsubscribing from configuration changes
pub type UnsubscribeConfigurationResponse = dapr_v1::UnsubscribeConfigurationResponse;

/// Encryption gRPC request
pub type EncryptRequest = crate::dapr::proto::runtime::v1::EncryptRequest;

/// Decrypt gRPC request
pub type DecryptRequest = crate::dapr::proto::runtime::v1::DecryptRequest;

/// Encryption request options
pub type EncryptRequestOptions = crate::dapr::proto::runtime::v1::EncryptRequestOptions;

/// Decryption request options
pub type DecryptRequestOptions = crate::dapr::proto::runtime::v1::DecryptRequestOptions;

/// The basic job structure
pub type Job = crate::dapr::proto::runtime::v1::Job;

/// A failure policy for a job
pub type JobFailurePolicy = crate::dapr::proto::common::v1::JobFailurePolicy;

/// A request to schedule a job
pub type ScheduleJobRequest = crate::dapr::proto::runtime::v1::ScheduleJobRequest;

/// A response from a schedule job request
pub type ScheduleJobResponse = crate::dapr::proto::runtime::v1::ScheduleJobResponse;

/// A request to get a job
pub type GetJobRequest = crate::dapr::proto::runtime::v1::GetJobRequest;

/// A response from a get job request
pub type GetJobResponse = crate::dapr::proto::runtime::v1::GetJobResponse;

/// A request to delete a job
pub type DeleteJobRequest = crate::dapr::proto::runtime::v1::DeleteJobRequest;

/// A response from a delete job request
pub type DeleteJobResponse = crate::dapr::proto::runtime::v1::DeleteJobResponse;

/// A request to conversate with an LLM
pub type ConversationRequest = crate::dapr::proto::runtime::v1::ConversationRequest;

/// A response from conversating with an LLM
pub type ConversationResponse = crate::dapr::proto::runtime::v1::ConversationResponse;

/// A result from an interacting with a LLM
pub type ConversationResult = crate::dapr::proto::runtime::v1::ConversationResult;

/// An input to the conversation
pub type ConversationInput = crate::dapr::proto::runtime::v1::ConversationInput;

/// A request to converse with an LLM (alpha2)
pub type ConversationRequestAlpha2 = crate::dapr::proto::runtime::v1::ConversationRequestAlpha2;

/// A response from conversing with an LLM (alpha2)
pub type ConversationResponseAlpha2 = crate::dapr::proto::runtime::v1::ConversationResponseAlpha2;

/// An input to the conversation (alpha2)
pub type ConversationInputAlpha2 = crate::dapr::proto::runtime::v1::ConversationInputAlpha2;

type StreamPayload = crate::dapr::proto::common::v1::StreamPayload;
impl<K> From<(K, Vec<u8>)> for common_v1::StateItem
where
    K: Into<String>,
{
    fn from((key, value): (K, Vec<u8>)) -> Self {
        common_v1::StateItem {
            key: key.into(),
            value,
            ..Default::default()
        }
    }
}

pub struct ReaderStream<T>(tokio_util::io::ReaderStream<T>);

impl<T: AsyncRead> ReaderStream<T> {
    pub fn new(data: T) -> Self {
        ReaderStream(tokio_util::io::ReaderStream::new(data))
    }
}

pub struct JobBuilder {
    schedule: Option<String>,
    data: Option<Any>,
    name: String,
    ttl: Option<String>,
    repeats: Option<u32>,
    due_time: Option<String>,
    failure_policy: Option<JobFailurePolicy>,
}

impl JobBuilder {
    /// Create a new Job to be scheduled
    pub fn new(name: &str) -> Self {
        JobBuilder {
            schedule: None,
            data: None,
            name: name.to_string(),
            ttl: None,
            repeats: None,
            due_time: None,
            failure_policy: None,
        }
    }

    pub fn with_schedule(mut self, schedule: &str) -> Self {
        self.schedule = Some(schedule.into());
        self
    }

    pub fn with_data(mut self, data: Any) -> Self {
        self.data = Some(data);
        self
    }

    pub fn with_ttl(mut self, ttl: &str) -> Self {
        self.ttl = Some(ttl.into());
        self
    }

    pub fn with_repeats(mut self, repeats: u32) -> Self {
        self.repeats = Some(repeats);
        self
    }

    pub fn with_due_time(mut self, due_time: &str) -> Self {
        self.due_time = Some(due_time.into());
        self
    }

    pub fn with_failure_policy(mut self, policy: JobFailurePolicy) -> Self {
        self.failure_policy = Some(policy);
        self
    }

    pub fn build(self) -> Job {
        Job {
            schedule: self.schedule,
            data: self.data,
            name: self.name,
            ttl: self.ttl,
            repeats: self.repeats,
            due_time: self.due_time,
            failure_policy: self.failure_policy,
        }
    }
}

// Enum for a job failure policy
pub enum JobFailurePolicyType {
    Drop {},
    Constant {},
}

pub struct JobFailurePolicyBuilder {
    policy: JobFailurePolicyType,
    pub retry_interval: Option<Duration>,
    pub max_retries: Option<u32>,
}

impl JobFailurePolicyBuilder {
    pub fn new(policy: JobFailurePolicyType) -> Self {
        JobFailurePolicyBuilder {
            policy,
            retry_interval: None,
            max_retries: None,
        }
    }

    pub fn with_retry_interval(mut self, interval: Duration) -> Self {
        // Convert interval string (e.g., "5s") to ProstDuration
        self.retry_interval = Some(interval);
        self
    }

    pub fn with_max_retries(mut self, max_retries: u32) -> Self {
        self.max_retries = Some(max_retries);
        self
    }

    pub fn build(self) -> common_v1::JobFailurePolicy {
        match self.policy {
            JobFailurePolicyType::Drop {} => common_v1::JobFailurePolicy {
                policy: Some(Policy::Drop(Default::default())),
            },
            JobFailurePolicyType::Constant {} => JobFailurePolicy {
                policy: Some(Policy::Constant(JobFailurePolicyConstant {
                    interval: self.retry_interval.map(|interval| {
                        prost_types::Duration::try_from(interval)
                            .expect("Failed to convert Duration")
                    }),
                    max_retries: self.max_retries,
                })),
            },
        }
    }
}

pub struct ConversationInputBuilder {
    content: String,
    role: Option<String>,
    scrub_pii: Option<bool>,
}

impl ConversationInputBuilder {
    pub fn new(message: &str) -> Self {
        ConversationInputBuilder {
            content: message.to_string(),
            role: None,
            scrub_pii: None,
        }
    }

    pub fn build(self) -> ConversationInput {
        ConversationInput {
            content: self.content,
            role: self.role,
            scrub_pii: self.scrub_pii,
        }
    }
}

pub struct ConversationRequestBuilder {
    name: String,
    context_id: Option<String>,
    inputs: Vec<ConversationInput>,
    parameters: HashMap<String, Any>,
    metadata: HashMap<String, String>,
    scrub_pii: Option<bool>,
    temperature: Option<f64>,
}
impl ConversationRequestBuilder {
    pub fn new(name: &str, inputs: Vec<ConversationInput>) -> Self {
        ConversationRequestBuilder {
            name: name.to_string(),
            context_id: None,
            inputs,
            parameters: Default::default(),
            metadata: Default::default(),
            scrub_pii: None,
            temperature: None,
        }
    }

    pub fn build(self) -> ConversationRequest {
        ConversationRequest {
            name: self.name,
            context_id: self.context_id,
            inputs: self.inputs,
            parameters: self.parameters,
            metadata: self.metadata,
            scrub_pii: self.scrub_pii,
            temperature: self.temperature,
        }
    }
}

pub struct ConversationInputAlpha2Builder {
    messages: Vec<crate::dapr::proto::runtime::v1::ConversationMessage>,
    scrub_pii: Option<bool>,
}
impl ConversationInputAlpha2Builder {
    pub fn new(messages: Vec<crate::dapr::proto::runtime::v1::ConversationMessage>) -> Self {
        ConversationInputAlpha2Builder {
            messages,
            scrub_pii: None,
        }
    }
    pub fn with_scrub_pii(mut self, scrub: bool) -> Self {
        self.scrub_pii = Some(scrub);
        self
    }
    pub fn build(self) -> ConversationInputAlpha2 {
        ConversationInputAlpha2 {
            messages: self.messages,
            scrub_pii: self.scrub_pii,
        }
    }
}
pub struct ConversationRequestAlpha2Builder {
    name: String,
    inputs: Vec<ConversationInputAlpha2>,
    metadata: HashMap<String, String>,
    scrub_pii: Option<bool>,
    temperature: Option<f64>,
    parameters: HashMap<String, prost_types::Any>,
    tools: Vec<crate::dapr::proto::runtime::v1::ConversationTools>,
    response_format: Option<prost_types::Struct>,
    prompt_cache_retention: Option<prost_types::Duration>,
}
impl ConversationRequestAlpha2Builder {
    pub fn new(name: &str, inputs: Vec<ConversationInputAlpha2>) -> Self {
        ConversationRequestAlpha2Builder {
            name: name.to_string(),
            inputs,
            metadata: Default::default(),
            scrub_pii: None,
            temperature: None,
            parameters: Default::default(),
            tools: Default::default(),
            response_format: None,
            prompt_cache_retention: None,
        }
    }
    pub fn with_metadata(mut self, metadata: HashMap<String, String>) -> Self {
        self.metadata = metadata;
        self
    }
    pub fn with_scrub_pii(mut self, scrub: bool) -> Self {
        self.scrub_pii = Some(scrub);
        self
    }
    pub fn with_temperature(mut self, temp: f64) -> Self {
        self.temperature = Some(temp);
        self
    }
    pub fn with_parameters(mut self, params: HashMap<String, prost_types::Any>) -> Self {
        self.parameters = params;
        self
    }
    pub fn with_tools(
        mut self,
        tools: Vec<crate::dapr::proto::runtime::v1::ConversationTools>,
    ) -> Self {
        self.tools = tools;
        self
    }
    pub fn with_response_format(mut self, format: prost_types::Struct) -> Self {
        self.response_format = Some(format);
        self
    }
    pub fn with_prompt_cache_retention(mut self, retention: prost_types::Duration) -> Self {
        self.prompt_cache_retention = Some(retention);
        self
    }
    pub fn build(self) -> ConversationRequestAlpha2 {
        ConversationRequestAlpha2 {
            name: self.name,
            inputs: self.inputs,
            metadata: self.metadata,
            scrub_pii: self.scrub_pii,
            temperature: self.temperature,
            parameters: self.parameters,
            tools: self.tools,
            response_format: self.response_format,
            prompt_cache_retention: self.prompt_cache_retention,
            ..Default::default()
        }
    }
}

pub use crate::dapr::proto::runtime::v1::{
    ConversationMessage, ConversationMessageContent, ConversationMessageOfAssistant,
    ConversationMessageOfDeveloper, ConversationMessageOfSystem, ConversationMessageOfTool,
    ConversationMessageOfUser, ConversationToolCalls, ConversationToolCallsOfFunction,
    ConversationTools, conversation_message, conversation_tool_calls,
};

impl From<ConversationMessageOfUser> for ConversationMessage {
    fn from(msg: ConversationMessageOfUser) -> Self {
        ConversationMessage {
            message_types: Some(conversation_message::MessageTypes::OfUser(msg)),
        }
    }
}

impl From<ConversationMessageOfSystem> for ConversationMessage {
    fn from(msg: ConversationMessageOfSystem) -> Self {
        ConversationMessage {
            message_types: Some(conversation_message::MessageTypes::OfSystem(msg)),
        }
    }
}

impl From<ConversationMessageOfDeveloper> for ConversationMessage {
    fn from(msg: ConversationMessageOfDeveloper) -> Self {
        ConversationMessage {
            message_types: Some(conversation_message::MessageTypes::OfDeveloper(msg)),
        }
    }
}

impl From<ConversationMessageOfAssistant> for ConversationMessage {
    fn from(msg: ConversationMessageOfAssistant) -> Self {
        ConversationMessage {
            message_types: Some(conversation_message::MessageTypes::OfAssistant(msg)),
        }
    }
}

impl From<ConversationMessageOfTool> for ConversationMessage {
    fn from(msg: ConversationMessageOfTool) -> Self {
        ConversationMessage {
            message_types: Some(conversation_message::MessageTypes::OfTool(msg)),
        }
    }
}