use std::collections::BTreeMap;
use std::collections::hash_map::Entry;
use std::str::FromStr;
use chrono::Utc;
use serde_json::{Value, from_str, to_string, to_value};
use thiserror::Error;
use tokio::sync::{mpsc, oneshot};
use tokio::task::spawn;
use uuid::Uuid;
use aves::flow::{BoundOutputs, LifecycleCorrelation, NodeLifecycleDispatchError, NodeLifecycleResult};
use aves::utils::uuid::SpectaUuid;
use crate::channels::for_kind;
use crate::database::{PermissionRequest, StoreError};
use crate::messages::ListenerType;
use crate::registrations::NodeCommand;
use crate::types::{
ConnectionTransport, InteractionAction, MessageNodeConfig, MessagingConfig, MessagingCredentials, MessagingError,
MessagingInteraction, MessagingSendReceipt, MessagingSenderConfig, NormalizedMessage, PermissionDuration,
PermissionNodeConfig,
};
use super::{Actor, ActorInteraction, EditCompleted, NodePending, PermSendCompleted};
#[derive(Debug, Error)]
enum NodeSetupError {
#[error(transparent)]
Messaging(#[from] MessagingError),
#[error(transparent)]
Json(#[from] serde_json::Error),
#[error(transparent)]
Uuid(#[from] uuid::Error),
#[error(transparent)]
Store(#[from] StoreError),
#[error("unknown connection")]
UnknownConnection,
#[error("credential unavailable")]
CredentialUnavailable,
#[error("invalid messaging provider type")]
InvalidProvider,
#[error("provider does not match connection")]
ProviderMismatch,
#[error("PermissionNode does not support Email")]
EmailNotSupported,
}
struct PreparedMessageNode {
connection_id: SpectaUuid,
retention_seconds: i64,
sender: MessagingSenderConfig,
message: NormalizedMessage,
metadata: Option<String>,
transport: ConnectionTransport,
credentials: MessagingCredentials,
}
struct PreparedPermissionNode {
connection_id: SpectaUuid,
retention_seconds: i64,
request_id: Uuid,
request: PermissionRequest,
sender: MessagingSenderConfig,
provider: ListenerType,
settings: MessagingConfig,
credentials: MessagingCredentials,
body: String,
metadata: Option<String>,
}
impl Actor {
pub(super) async fn handle_worker_tick(&mut self) {
let now = Utc::now().timestamp();
self.expire_permission_requests(now).await;
self.retry_permission_edits().await;
self.prune_messaging_history(now).await;
}
async fn expire_permission_requests(&mut self, now: i64) {
let rows = match self.store().await {
Ok(store) => match store.scan_unfinished(now).await {
Ok(rows) => rows,
Err(error) => {
tracing::warn!(%error, "Could not scan unfinished PermissionNode requests");
return;
}
},
Err(error) => {
tracing::warn!(%error, "Could not open store to scan unfinished PermissionNode requests");
return;
}
};
for row in rows {
if !matches!(row.status.as_str(), "prepared" | "pending") || now < row.expires_at {
continue;
}
let finalized = match self.store().await {
Ok(store) => match store
.finalize_atomic(&row.request_id, "expired", now, true)
.await
{
Ok(finalized) => finalized,
Err(error) => {
tracing::warn!(%error, request_id = %row.request_id, "Could not expire PermissionNode request");
false
}
},
Err(error) => {
tracing::warn!(%error, request_id = %row.request_id, "Could not open store to expire PermissionNode request");
false
}
};
if finalized {
self.edit_finalized(row, "Expired").await;
}
}
}
async fn retry_permission_edits(&mut self) {
let rows = match self.store().await {
Ok(store) => match store.scan_outstanding_edits().await {
Ok(rows) => rows,
Err(error) => {
tracing::warn!(%error, "Could not scan outstanding PermissionNode edits");
return;
}
},
Err(error) => {
tracing::warn!(%error, "Could not open store to scan outstanding PermissionNode edits");
return;
}
};
for row in rows {
let label = match row.status.as_str() {
"approved" => "Approved",
"denied" => "Denied",
"expired" => "Expired",
"cancelled" => "Cancelled",
"deleted" => "Deleted",
_ => "Expired",
};
self.edit_finalized(row, label).await;
}
}
async fn prune_messaging_history(&mut self, now: i64) {
match self.store().await {
Ok(store) => {
if let Err(error) = store.prune_expired(now).await {
tracing::warn!(%error, "Could not prune expired messaging records");
}
}
Err(error) => tracing::warn!(%error, "Could not open store to prune expired messaging records"),
}
}
pub(super) async fn handle_interaction(&mut self, event: ActorInteraction) {
let ActorInteraction {
connection_id,
interaction,
} = event;
let now = Utc::now().timestamp();
let request_id = interaction.request_id.to_string();
let request = self.load_callback_request(&request_id).await;
let (callback_connection_id, retention_seconds) = self
.callback_retention(request.as_ref(), &connection_id)
.await;
self.persist_permission_callback(
&interaction,
&request_id,
&callback_connection_id,
retention_seconds,
now,
)
.await;
let Some(request) = request else { return };
if !matches!(request.status.as_str(), "prepared" | "pending") {
return;
}
let (status, label) = if now > request.expires_at {
("expired", "Expired")
} else {
match interaction.action {
InteractionAction::Approve => ("approved", "Approved"),
InteractionAction::Deny => ("denied", "Denied"),
InteractionAction::Unknown => return,
}
};
match self.store().await {
Ok(store) => match store
.finalize_atomic(&request.request_id, status, now, true)
.await
{
Ok(true) => self.edit_finalized(request, label).await,
Ok(false) => {}
Err(error) => {
tracing::warn!(%error, request_id = %request.request_id, "Could not finalize permission callback decision")
}
},
Err(error) => {
tracing::warn!(%error, request_id = %request.request_id, "Could not open store to finalize permission callback decision")
}
}
}
async fn load_callback_request(&mut self, request_id: &str) -> Option<PermissionRequest> {
match self.store().await {
Ok(store) => match store.get_permission_request(request_id).await {
Ok(request) => request,
Err(error) => {
tracing::warn!(%error, %request_id, "Could not load PermissionNode request for callback arbitration");
None
}
},
Err(error) => {
tracing::warn!(%error, %request_id, "Could not open store for callback arbitration");
None
}
}
}
async fn callback_retention(
&mut self,
request: Option<&PermissionRequest>,
ingress_connection: &SpectaUuid,
) -> (Option<String>, i64) {
let connection_id = request
.map(|request| request.connection_id.clone())
.unwrap_or_else(|| ingress_connection.to_string());
let warnings = match request {
Some(_) => (
"Permission callback connection is missing; using one-hour retention",
"Could not load Permission callback retention; using one-hour retention",
"Could not open store for callback retention; using one-hour retention",
),
None => (
"Unknown callback connection; using one-hour retention",
"Could not load callback retention; using one-hour retention",
"Could not open store for callback retention; using one-hour retention",
),
};
let retention = match self.store().await {
Ok(store) => match store.get_connection(&connection_id).await {
Ok(Some(connection)) => connection.retention_seconds,
Ok(None) => {
tracing::warn!(%connection_id, "{}", warnings.0);
3_600
}
Err(error) => {
tracing::warn!(%error, %connection_id, "{}", warnings.1);
3_600
}
},
Err(error) => {
tracing::warn!(%error, %connection_id, "{}", warnings.2);
3_600
}
};
(Some(connection_id), retention)
}
async fn persist_permission_callback(
&mut self,
interaction: &MessagingInteraction,
request_id: &str,
connection_id: &Option<String>,
retention_seconds: i64,
now: i64,
) {
match self.store().await {
Ok(store) => {
if let Err(error) = store
.insert_callback(
&interaction.raw_payload,
Some(request_id),
Some(interaction.action.as_str()),
Some(now),
connection_id.as_deref(),
now,
retention_seconds,
now.saturating_add(retention_seconds),
)
.await
{
tracing::warn!(%error, %request_id, "Could not persist permission callback; continuing arbitration");
}
}
Err(error) => {
tracing::warn!(%error, %request_id, "Could not open store to persist permission callback; continuing arbitration")
}
}
}
pub(super) async fn handle_permission_send_completion(&mut self, completion: PermSendCompleted) {
let PermSendCompleted {
request_id,
connection_id,
retention_seconds,
body,
metadata,
result,
} = completion;
match result {
Ok(provider_message_id) => {
self.attach_permission_receipt(&request_id, &provider_message_id)
.await;
self.persist_permission_acceptance(
&connection_id,
&body,
metadata.as_deref(),
retention_seconds,
&provider_message_id,
)
.await;
}
Err(error) => self.fail_permission_send(&request_id, error).await,
}
}
async fn attach_permission_receipt(&mut self, request_id: &str, provider_message_id: &str) {
let marked = match self.store().await {
Ok(store) => match store
.mark_pending(request_id, provider_message_id, provider_message_id)
.await
{
Ok(marked) => marked,
Err(error) => {
tracing::warn!(%error, %request_id, "Could not mark PermissionNode request pending");
false
}
},
Err(error) => {
tracing::warn!(%error, %request_id, "Could not open store to mark PermissionNode pending");
false
}
};
if marked {
return;
}
match self.store().await {
Ok(store) => {
if let Err(error) = store
.attach_provider_message(request_id, provider_message_id, provider_message_id)
.await
{
tracing::warn!(%error, %request_id, "Could not attach PermissionNode provider receipt");
}
match store.get_permission_request(request_id).await {
Ok(Some(row)) if row.outstanding_edit => {
let label = match row.status.as_str() {
"approved" => Some("Approved"),
"denied" => Some("Denied"),
"expired" => Some("Expired"),
"cancelled" => Some("Cancelled"),
_ => None,
};
if let Some(label) = label {
self.edit_finalized(row, label).await;
}
}
Ok(_) => {}
Err(error) => {
tracing::warn!(%error, %request_id, "Could not reload PermissionNode request after receipt")
}
}
}
Err(error) => tracing::warn!(%error, %request_id, "Could not open store to attach PermissionNode receipt"),
}
}
async fn persist_permission_acceptance(
&mut self,
connection_id: &SpectaUuid,
body: &str,
metadata: Option<&str>,
retention_seconds: i64,
provider_message_id: &str,
) {
let now = Utc::now().timestamp();
let expires_at = now.saturating_add(retention_seconds);
match self.store().await {
Ok(store) => {
if let Err(error) = store
.insert_accepted_send(
body,
metadata,
provider_message_id,
&connection_id.to_string(),
now,
retention_seconds,
expires_at,
)
.await
{
tracing::warn!(%error, %connection_id, "Could not persist accepted PermissionNode send; continuing");
}
}
Err(error) => {
tracing::warn!(%error, %connection_id, "Could not open store for accepted PermissionNode send")
}
}
}
async fn fail_permission_send(&mut self, request_id: &str, error: MessagingError) {
tracing::warn!(%error, %request_id, "PermissionNode interactive send failed");
let now = Utc::now().timestamp();
match self.store().await {
Ok(store) => {
if let Err(error) = store
.finalize_atomic(request_id, "failed", now, false)
.await
{
tracing::warn!(%error, %request_id, "Could not persist failed PermissionNode request");
}
}
Err(error) => {
tracing::warn!(%error, %request_id, "Could not open store to persist failed PermissionNode request")
}
}
}
pub(super) async fn handle_edit_completion(&mut self, completion: EditCompleted) {
let EditCompleted { request_id, result } = completion;
match result {
Ok(()) => match self.store().await {
Ok(store) => {
if let Err(error) = store.clear_outstanding_edit(&request_id).await {
tracing::warn!(%error, %request_id, "Could not clear PermissionNode outstanding edit flag");
}
}
Err(error) => {
tracing::warn!(%error, %request_id, "Could not open store to clear PermissionNode edit flag")
}
},
Err(error) => {
tracing::warn!(%error, %request_id, "PermissionNode provider edit failed; retry remains pending")
}
}
}
pub(super) async fn handle_node_command(&mut self, command: NodeCommand) {
match command {
NodeCommand::MessageStart {
correlation,
config,
reply,
} => self.start_message_node(correlation, config, reply).await,
NodeCommand::MessageTick { correlation, reply } => self.tick_message_node(correlation, reply).await,
NodeCommand::MessageCancel { correlation, reply } => self.cancel_message_node(correlation, reply),
NodeCommand::PermissionStart {
correlation,
config,
reply,
} => self.start_permission_node(correlation, config, reply).await,
NodeCommand::PermissionTick { correlation, reply } => self.tick_permission_node(correlation, reply).await,
NodeCommand::PermissionCancel { correlation, reply } => {
self.cancel_permission_node(correlation, reply).await
}
}
}
async fn start_message_node(
&mut self,
correlation: LifecycleCorrelation,
config: MessageNodeConfig,
reply: oneshot::Sender<Result<NodeLifecycleResult, NodeLifecycleDispatchError>>,
) {
let prepared = match self.prepare_message_node(&config).await {
Ok(prepared) => prepared,
Err(error) => return Self::reply_failed(reply, error.to_string()),
};
let PreparedMessageNode {
connection_id,
retention_seconds,
sender,
message,
metadata,
transport,
credentials,
} = prepared;
self.node_pending.insert(
correlation.clone(),
NodePending {
connection_id,
retention_seconds,
body: config.body,
metadata,
result: None,
},
);
let done = self.node_done_tx.clone();
spawn(async move {
let result = Self::send_message(sender, message, transport.messaging_config(), credentials).await;
let _ = done.send((correlation, result)).await;
});
let _ = reply.send(Ok(NodeLifecycleResult::Running));
}
async fn prepare_message_node(
&mut self,
config: &MessageNodeConfig,
) -> Result<PreparedMessageNode, NodeSetupError> {
let connection = self
.store()
.await?
.get_connection(&config.connection_id.to_string())
.await
.map_err(MessagingError::from)?
.ok_or(NodeSetupError::UnknownConnection)?;
let transport = from_str::<ConnectionTransport>(&connection.transport_settings)?;
let sender =
MessagingSenderConfig::from_destination(config.provider, config.connection_id, &config.destination);
let credential_id = SpectaUuid::from_str(&connection.credential_id)?;
let credentials = self
.credentials
.get(&credential_id)
.cloned()
.ok_or(NodeSetupError::CredentialUnavailable)?;
let message = NormalizedMessage {
body: config.body.clone(),
metadata: config.metadata.clone(),
};
let metadata = Some(to_string(&config.metadata)?);
Ok(PreparedMessageNode {
connection_id: config.connection_id,
retention_seconds: connection.retention_seconds,
sender,
message,
metadata,
transport,
credentials,
})
}
fn reply_failed(
reply: oneshot::Sender<Result<NodeLifecycleResult, NodeLifecycleDispatchError>>,
message: impl Into<String>,
) {
let _ = reply.send(Ok(NodeLifecycleResult::Failed {
message: message.into(),
state: Value::Null,
}));
}
async fn tick_message_node(
&mut self,
correlation: LifecycleCorrelation,
reply: oneshot::Sender<Result<NodeLifecycleResult, NodeLifecycleDispatchError>>,
) {
let pending = match self.node_pending.entry(correlation) {
Entry::Occupied(entry) if entry.get().result.is_some() => entry.remove(),
Entry::Occupied(_) | Entry::Vacant(_) => {
let _ = reply.send(Ok(NodeLifecycleResult::Running));
return;
}
};
let NodePending {
connection_id,
retention_seconds,
body,
metadata,
result,
} = pending;
match result.expect("terminal pending result was checked") {
Ok(receipt) => {
self.finish_message_node(connection_id, retention_seconds, body, metadata, receipt, reply)
.await
}
Err(error) => {
let _ = reply.send(Ok(NodeLifecycleResult::Failed {
message: error.to_string(),
state: Value::Null,
}));
}
}
}
async fn finish_message_node(
&mut self,
connection_id: SpectaUuid,
retention_seconds: i64,
body: String,
metadata: Option<String>,
receipt: MessagingSendReceipt,
reply: oneshot::Sender<Result<NodeLifecycleResult, NodeLifecycleDispatchError>>,
) {
self.persist_accepted_send(
&connection_id,
&body,
metadata.as_deref(),
&receipt,
retention_seconds,
[
"Could not serialize accepted Flow message receipt",
"Could not persist accepted Flow message; continuing delivery",
"Could not open message store for accepted Flow message",
],
)
.await;
let outputs = match to_value(&receipt) {
Ok(value) => BoundOutputs::new(BTreeMap::from([("receipt".to_owned(), value)])),
Err(error) => {
let _ = reply.send(Ok(NodeLifecycleResult::Failed {
message: error.to_string(),
state: Value::Null,
}));
return;
}
};
let _ = reply.send(Ok(NodeLifecycleResult::Succeeded {
outputs,
state: Value::Null,
}));
}
fn cancel_message_node(
&mut self,
correlation: LifecycleCorrelation,
reply: oneshot::Sender<Result<NodeLifecycleResult, NodeLifecycleDispatchError>>,
) {
self.node_pending.remove(&correlation);
let _ = reply.send(Ok(NodeLifecycleResult::Cancelled));
}
async fn start_permission_node(
&mut self,
correlation: LifecycleCorrelation,
config: PermissionNodeConfig,
reply: oneshot::Sender<Result<NodeLifecycleResult, NodeLifecycleDispatchError>>,
) {
let prepared = match self.prepare_permission_node(&correlation, &config).await {
Ok(prepared) => prepared,
Err(error) => return Self::reply_failed(reply, error.to_string()),
};
let PreparedPermissionNode {
connection_id,
retention_seconds,
request_id,
request,
sender,
provider,
settings,
credentials,
body,
metadata,
} = prepared;
let write = match self.store().await {
Ok(store) => store
.insert_prepared(&request)
.await
.map_err(NodeSetupError::Store),
Err(error) => Err(NodeSetupError::Messaging(error)),
};
if let Err(error) = write {
return Self::reply_failed(reply, error.to_string());
}
spawn(Self::run_interactive_send(
sender,
provider,
settings,
credentials,
SpectaUuid(request_id),
body.clone(),
connection_id,
retention_seconds,
body,
metadata,
self.perm_tx.clone(),
));
let _ = reply.send(Ok(NodeLifecycleResult::Running));
}
async fn prepare_permission_node(
&mut self,
correlation: &LifecycleCorrelation,
config: &PermissionNodeConfig,
) -> Result<PreparedPermissionNode, NodeSetupError> {
if config.provider == ListenerType::Email {
return Err(NodeSetupError::EmailNotSupported);
}
let connection = self
.store()
.await?
.get_connection(&config.connection_id.to_string())
.await?;
let connection = connection.ok_or(NodeSetupError::UnknownConnection)?;
let now = Utc::now().timestamp();
let expires_at = PermissionDuration::new(config.duration_seconds)?.expires_at(now)?;
let transport = from_str::<ConnectionTransport>(&connection.transport_settings)?;
let provider = ListenerType::from_storage(&connection.provider_type).ok_or(NodeSetupError::InvalidProvider)?;
if provider != config.provider {
return Err(NodeSetupError::ProviderMismatch);
}
let connection_id = SpectaUuid(Uuid::parse_str(&connection.id)?);
let sender = MessagingSenderConfig::from_destination(provider, connection_id, &config.destination);
let credential_id = SpectaUuid::from_str(&connection.credential_id)?;
let credentials = self
.credentials
.get(&credential_id)
.cloned()
.ok_or(NodeSetupError::CredentialUnavailable)?;
let request_id = Uuid::new_v4();
let metadata = to_string(&config.metadata).ok();
let request = PermissionRequest {
request_id: request_id.to_string(),
connection_id: connection.id,
provider_message_id: None,
destination: config.destination.to_string(),
request_text: config.body.clone(),
expires_at,
flow_execution_id: correlation.flow_execution_id.to_string(),
flow_node_id: correlation.flow_node_id.to_string(),
status: "prepared".to_owned(),
provider_receipt: None,
decided_at: None,
outstanding_edit: false,
};
Ok(PreparedPermissionNode {
connection_id,
retention_seconds: connection.retention_seconds,
request_id,
request,
sender,
provider,
settings: transport.messaging_config(),
credentials,
body: config.body.clone(),
metadata,
})
}
async fn tick_permission_node(
&mut self,
correlation: LifecycleCorrelation,
reply: oneshot::Sender<Result<NodeLifecycleResult, NodeLifecycleDispatchError>>,
) {
let row = match self.store().await {
Ok(store) => {
store
.get_by_correlation(
&correlation.flow_execution_id.to_string(),
&correlation.flow_node_id.to_string(),
)
.await
}
Err(error) => {
Self::reply_failed(reply, error.to_string());
return;
}
};
match row {
Ok(None) => Self::reply_failed(reply, "permission request not found"),
Err(error) => Self::reply_failed(reply, error.to_string()),
Ok(Some(row)) => match row.status.as_str() {
"approved" => {
let outputs = BoundOutputs::new(BTreeMap::from([
("approved".to_owned(), Value::Bool(true)),
(
"provider_receipt".to_owned(),
to_value(row.provider_receipt).unwrap_or(Value::Null),
),
("decided_at".to_owned(), to_value(row.decided_at).unwrap_or(Value::Null)),
]));
let _ = reply.send(Ok(NodeLifecycleResult::Succeeded {
outputs,
state: Value::Null,
}));
}
"denied" | "expired" | "cancelled" | "deleted" | "failed" => {
Self::reply_failed(reply, format!("permission {}", row.status));
}
_ => {
let _ = reply.send(Ok(NodeLifecycleResult::Running));
}
},
}
}
async fn cancel_permission_node(
&mut self,
correlation: LifecycleCorrelation,
reply: oneshot::Sender<Result<NodeLifecycleResult, NodeLifecycleDispatchError>>,
) {
let row = match self.store().await {
Ok(store) => {
store
.get_by_correlation(
&correlation.flow_execution_id.to_string(),
&correlation.flow_node_id.to_string(),
)
.await
}
Err(error) => {
tracing::warn!(%error, "Could not load PermissionNode for cancellation");
let _ = reply.send(Ok(NodeLifecycleResult::Cancelled));
return;
}
};
if let Ok(Some(row)) = row
&& matches!(row.status.as_str(), "prepared" | "pending")
{
self.finalize_permission_cancel(&row).await;
if let Some(provider_message_id) = row.provider_message_id.clone() {
self.launch_cancel_edit(row, provider_message_id).await;
}
}
let _ = reply.send(Ok(NodeLifecycleResult::Cancelled));
}
async fn finalize_permission_cancel(&mut self, row: &PermissionRequest) {
let now = Utc::now().timestamp();
match self.store().await {
Ok(store) => {
if let Err(error) = store
.finalize_atomic(&row.request_id, "cancelled", now, true)
.await
{
tracing::warn!(%error, request_id = %row.request_id, "Could not finalize PermissionNode cancellation");
}
}
Err(error) => {
tracing::warn!(%error, request_id = %row.request_id, "Could not open store to finalize PermissionNode cancellation")
}
}
}
async fn launch_cancel_edit(&mut self, row: PermissionRequest, provider_message_id: String) {
match self.prepare_cancel_edit(&row).await {
Ok((sender, provider, settings, credentials)) => {
spawn(Self::run_provider_edit(
sender,
provider,
settings,
credentials,
provider_message_id,
row.request_text,
"Cancelled".to_owned(),
row.request_id,
self.edit_tx.clone(),
));
}
Err(error) => tracing::warn!(%error, "Could not prepare PermissionNode provider edit"),
}
}
async fn edit_finalized(&mut self, row: PermissionRequest, outcome_label: &str) {
let Some(provider_message_id) = row.provider_message_id.clone() else {
tracing::warn!(request_id = %row.request_id, "Finalized PermissionNode request has no provider message to edit");
return;
};
match self.prepare_provider_edit(&row).await {
Ok((sender, provider, settings, credentials)) => {
spawn(Self::run_provider_edit(
sender,
provider,
settings,
credentials,
provider_message_id,
row.request_text,
outcome_label.to_owned(),
row.request_id,
self.edit_tx.clone(),
));
}
Err(error) => {
tracing::warn!(%error, request_id = %row.request_id, "Could not prepare finalized PermissionNode provider edit")
}
}
}
async fn prepare_provider_edit(
&mut self,
row: &PermissionRequest,
) -> Result<
(
MessagingSenderConfig,
ListenerType,
MessagingConfig,
MessagingCredentials,
),
MessagingError,
> {
let conn = self
.store()
.await?
.get_connection(&row.connection_id)
.await?
.ok_or_else(|| MessagingError::ConnectionNotFound {
connection_id: row.connection_id.clone(),
})?;
let provider =
ListenerType::from_storage(&conn.provider_type).ok_or_else(|| MessagingError::Serialization {
message: "invalid messaging provider type".to_owned(),
})?;
let transport: ConnectionTransport = from_str(&conn.transport_settings)?;
let destination: Value = from_str(&row.destination)?;
let connection_id = Self::connection_uuid(&row.connection_id)?;
let sender = MessagingSenderConfig::from_destination(provider, connection_id, &destination);
let credential_id =
SpectaUuid::from_str(&conn.credential_id).map_err(|error| MessagingError::ConnectionNotFound {
connection_id: error.to_string(),
})?;
let credentials = self
.credentials
.get(&credential_id)
.cloned()
.ok_or_else(|| MessagingError::ChannelConnectionFailed {
listener_type: provider,
message: "configured credential is not available".to_owned(),
})?;
Ok((sender, provider, transport.messaging_config(), credentials))
}
async fn prepare_cancel_edit(
&mut self,
row: &PermissionRequest,
) -> Result<
(
MessagingSenderConfig,
ListenerType,
MessagingConfig,
MessagingCredentials,
),
MessagingError,
> {
let conn = self
.store()
.await?
.get_connection(&row.connection_id)
.await?
.ok_or_else(|| MessagingError::ConnectionNotFound {
connection_id: row.connection_id.clone(),
})?;
let provider =
ListenerType::from_storage(&conn.provider_type).ok_or_else(|| MessagingError::Serialization {
message: "invalid messaging provider type".to_owned(),
})?;
let transport: ConnectionTransport = from_str(&conn.transport_settings)?;
let credential_id =
SpectaUuid::from_str(&conn.credential_id).map_err(|error| MessagingError::ConnectionNotFound {
connection_id: error.to_string(),
})?;
let credentials = self
.credentials
.get(&credential_id)
.cloned()
.ok_or_else(|| MessagingError::ChannelConnectionFailed {
listener_type: provider,
message: "configured credential is not available".to_owned(),
})?;
let destination: Value = from_str(&row.destination)?;
let connection_id = Self::connection_uuid(&conn.id)?;
let sender = MessagingSenderConfig::from_destination(provider, connection_id, &destination);
Ok((sender, provider, transport.messaging_config(), credentials))
}
#[allow(clippy::too_many_arguments)]
async fn run_interactive_send(
sender: MessagingSenderConfig,
provider: ListenerType,
settings: MessagingConfig,
credentials: MessagingCredentials,
request_id: SpectaUuid,
request_text: String,
connection_id: SpectaUuid,
retention_seconds: i64,
body: String,
metadata: Option<String>,
done: mpsc::Sender<PermSendCompleted>,
) {
let mut plugin = for_kind(provider);
let result = match plugin.connect(&settings, &credentials).await {
Ok(()) => {
plugin
.send_interactive(&sender, request_id, &request_text)
.await
}
Err(error) => Err(error),
};
if let Err(error) = done
.send(PermSendCompleted {
request_id: request_id.to_string(),
connection_id,
retention_seconds,
body,
metadata,
result,
})
.await
{
tracing::warn!(%error, "Could not deliver PermissionNode send completion to actor");
}
}
#[allow(clippy::too_many_arguments)]
async fn run_provider_edit(
sender: MessagingSenderConfig,
provider: ListenerType,
settings: MessagingConfig,
credentials: MessagingCredentials,
provider_message_id: String,
request_text: String,
outcome_label: String,
request_id: String,
done: mpsc::Sender<EditCompleted>,
) {
let mut plugin = for_kind(provider);
let result = match plugin.connect(&settings, &credentials).await {
Ok(()) => {
plugin
.edit_interactive(&sender, &provider_message_id, &request_text, &outcome_label)
.await
}
Err(error) => Err(error),
};
if let Err(error) = done.send(EditCompleted { request_id, result }).await {
tracing::warn!(%error, "Could not deliver PermissionNode edit completion to actor");
}
}
}