use chrono::Utc;
use serde_json::{Value, from_str, to_string};
use tokio::spawn;
use tokio::sync::oneshot;
use uuid::Uuid;
use aves::flow::{
DeleteFlowTriggerInput, ListFlowTriggersInput, ListFlowsInput, dispatch_delete_flow_trigger,
dispatch_list_flow_triggers, dispatch_list_flows,
};
use aves::utils::uuid::SpectaUuid;
use crate::channels::for_kind;
use crate::database::MessagingConnection;
use crate::messages::{ListenerType, MessagingRequest, MessagingRequestPayload, MessagingResponse, SubscriptionRow};
use crate::types::{
ConnectionRecord, ConnectionTransport, EditConnection, MessagingConfig, MessagingCredentials, MessagingError,
MessagingSenderConfig, NewConnection, NormalizedMessage, PersistencePolicy, RetentionDuration,
};
use super::{Actor, SendCompletion};
impl Actor {
pub(super) async fn handle_message(&mut self, request: MessagingRequest) {
let result = match request.payload {
MessagingRequestPayload::StartServer => self.start_server().await,
MessagingRequestPayload::StopServer => self.stop_server().await,
MessagingRequestPayload::SupplyCredential {
credential_id,
credentials,
} => {
self.credentials.insert(credential_id, credentials);
*self.credential_generation.entry(credential_id).or_default() += 1;
self.reconcile().await
}
MessagingRequestPayload::RemoveCredential { credential_id } => {
self.credentials.remove(&credential_id);
self.credential_generation.remove(&credential_id);
self.reconcile().await
}
MessagingRequestPayload::ListConnections => self.list_connections().await,
MessagingRequestPayload::CreateConnection { input } => self.create_connection(input).await,
MessagingRequestPayload::UpdateConnection { id, input } => self.update_connection(id, input).await,
MessagingRequestPayload::SetConnectionEnabled { connection_id, enabled } => {
self.set_connection_enabled(connection_id, enabled).await
}
MessagingRequestPayload::DeleteConnection { connection_id } => self.delete_connection(connection_id).await,
MessagingRequestPayload::List => Ok(self.list_subscriptions()),
MessagingRequestPayload::Send { sender, message } => {
self.start_send(request.reply_to, sender, message).await;
return;
}
MessagingRequestPayload::SendMetadataSchema { listener_type } => {
Ok(MessagingResponse::SendMetadataSchema {
schema: Value::from(for_kind(listener_type).send_metadata_schema()),
})
}
};
let _ = request.reply_to.send(result);
}
async fn list_connections(&mut self) -> Result<MessagingResponse, MessagingError> {
let connections = self.store().await?.list_connections().await?;
let connections = connections
.iter()
.map(Self::connection_to_record)
.collect::<Result<Vec<_>, _>>()?;
Ok(MessagingResponse::Connections { connections })
}
async fn create_connection(&mut self, input: NewConnection) -> Result<MessagingResponse, MessagingError> {
Self::validate_connection_provider(input.provider, &input.transport)?;
let record = ConnectionRecord {
id: SpectaUuid(Uuid::new_v4()),
display_name: input.display_name,
provider: input.provider,
credential_id: input.credential_id,
transport: input.transport,
enabled: input.enabled,
retention: input.retention,
};
self.store()
.await?
.create_connection(&Self::record_to_connection(&record))
.await?;
let _ = self.reconcile().await;
Ok(MessagingResponse::Connection { connection: record })
}
async fn update_connection(
&mut self,
id: SpectaUuid,
input: EditConnection,
) -> Result<MessagingResponse, MessagingError> {
Self::validate_connection_provider(input.provider, &input.transport)?;
let connection = MessagingConnection {
id: id.to_string(),
display_name: input.display_name,
provider_type: input.provider.to_string(),
credential_id: input.credential_id.to_string(),
transport_settings: to_string(&input.transport)?,
enabled: false,
retention_seconds: input.retention.seconds as i64,
};
self.store().await?.update_connection(&connection).await?;
let stored = self
.store()
.await?
.get_connection(&id.to_string())
.await?
.ok_or_else(|| MessagingError::ConnectionNotFound {
connection_id: id.to_string(),
})?;
let _ = self.reconcile().await;
Ok(MessagingResponse::Connection {
connection: Self::connection_to_record(&stored)?,
})
}
async fn set_connection_enabled(
&mut self,
connection_id: SpectaUuid,
enabled: bool,
) -> Result<MessagingResponse, MessagingError> {
self.store()
.await?
.set_enabled(&connection_id.to_string(), enabled)
.await?;
let _ = self.reconcile().await;
Ok(MessagingResponse::Success)
}
async fn delete_connection(&mut self, connection_id: SpectaUuid) -> Result<MessagingResponse, MessagingError> {
let id = connection_id.to_string();
self.remove_connection_triggers(&id).await;
self.store()
.await?
.mark_deleted_for_connection(&id, Utc::now().timestamp())
.await?;
self.store().await?.delete_connection(&id).await?;
let _ = self.reconcile().await;
Ok(MessagingResponse::Success)
}
async fn remove_connection_triggers(&self, connection_id: &str) {
match dispatch_list_flows(&self.knowledge_context, ListFlowsInput {}).await {
Ok(output) => {
for flow in output.flows {
match dispatch_list_flow_triggers(
&self.knowledge_context,
ListFlowTriggersInput { flow_id: flow.flow_id },
)
.await
{
Ok(output) => {
for trigger in output.triggers {
if trigger.adapter_kind == "subscription"
&& trigger
.adapter_config
.get("connection_id")
.and_then(Value::as_str)
== Some(connection_id)
&& let Err(error) = dispatch_delete_flow_trigger(
&self.knowledge_context,
DeleteFlowTriggerInput {
trigger_id: trigger.trigger_id,
},
)
.await
{
tracing::warn!(%error, trigger_id = %trigger.trigger_id, "Failed to remove subscription trigger for deleted connection");
}
}
}
Err(error) => {
tracing::warn!(%error, flow_id = %flow.flow_id, "Failed to list Flow triggers during connection deletion")
}
}
}
}
Err(error) => tracing::warn!(%error, "Failed to list Flows during connection deletion"),
}
}
fn list_subscriptions(&self) -> MessagingResponse {
let subscriptions = self
.listeners
.values()
.filter_map(|listener| {
let connection_id = listener.configuration.connection_id;
let inflow = self.inflows.get(&connection_id)?;
Some(SubscriptionRow {
id: SpectaUuid(listener.trigger.trigger_id),
listener_type: inflow.transport.provider(),
subscribed_channels: listener.configuration.subscribed_channels.clone(),
flow_id: SpectaUuid(listener.trigger.flow_id.0),
})
})
.collect();
MessagingResponse::Subscriptions { subscriptions }
}
async fn start_send(
&mut self,
reply_to: oneshot::Sender<Result<MessagingResponse, MessagingError>>,
sender: MessagingSenderConfig,
message: NormalizedMessage,
) {
let setup = match self.prepare_send(&sender).await {
Ok(setup) => setup,
Err(error) => {
let _ = reply_to.send(Err(error));
return;
}
};
let (connection_id, settings, credentials, retention_seconds, persistence) = setup;
let send_tx = self.send_tx.clone();
let body = message.body.clone();
let metadata = to_string(&message.metadata).ok();
spawn(async move {
let result = Actor::send_message(sender, message, settings, credentials).await;
if let Err(error) = send_tx
.send(SendCompletion {
reply_to,
connection_id,
body,
metadata,
retention_seconds,
result,
persistence,
})
.await
{
tracing::warn!(%error, "Could not deliver outbound send completion to actor");
}
});
}
async fn prepare_send(
&mut self,
sender: &MessagingSenderConfig,
) -> Result<
(
SpectaUuid,
MessagingConfig,
MessagingCredentials,
i64,
PersistencePolicy,
),
MessagingError,
> {
let (listener_type, connection_id, persistence) = sender.connection_info();
let connection = self
.store()
.await?
.get_connection(&connection_id.to_string())
.await?
.ok_or_else(|| MessagingError::ConnectionNotFound {
connection_id: connection_id.to_string(),
})?;
let credential_id = match Uuid::parse_str(&connection.credential_id) {
Ok(id) => SpectaUuid(id),
Err(error) => {
tracing::warn!(%error, connection_id = %connection.id, "Invalid messaging credential identifier");
return Err(Self::missing_send_credential(listener_type));
}
};
let credentials = self
.credentials
.get(&credential_id)
.cloned()
.ok_or_else(|| Self::missing_send_credential(listener_type))?;
let transport = match from_str::<ConnectionTransport>(&connection.transport_settings) {
Ok(transport) => transport,
Err(error) => {
tracing::warn!(%error, %connection_id, "Invalid messaging connection transport settings");
return Err(error.into());
}
};
Ok((
connection_id,
transport.messaging_config(),
credentials,
connection.retention_seconds,
persistence.clone(),
))
}
fn missing_send_credential(listener_type: ListenerType) -> MessagingError {
MessagingError::ChannelConnectionFailed {
listener_type,
message: "configured credential is not available".to_owned(),
}
}
fn connection_to_record(conn: &MessagingConnection) -> Result<ConnectionRecord, MessagingError> {
let provider =
ListenerType::from_storage(&conn.provider_type).ok_or_else(|| MessagingError::Serialization {
message: "invalid messaging provider type".to_owned(),
})?;
let transport = from_str(&conn.transport_settings)?;
let id = Uuid::parse_str(&conn.id)
.map(SpectaUuid)
.map_err(|_| MessagingError::ConnectionNotFound {
connection_id: conn.id.clone(),
})?;
let credential_id = Uuid::parse_str(&conn.credential_id)
.map(SpectaUuid)
.map_err(|_| MessagingError::ConnectionNotFound {
connection_id: conn.id.clone(),
})?;
let retention_seconds =
u64::try_from(conn.retention_seconds).map_err(|_| MessagingError::InvalidRetention {
reason: "stored duration is invalid".to_owned(),
})?;
Ok(ConnectionRecord {
id,
display_name: conn.display_name.clone(),
provider,
credential_id,
transport,
enabled: conn.enabled,
retention: RetentionDuration::new(retention_seconds)?,
})
}
fn record_to_connection(record: &ConnectionRecord) -> MessagingConnection {
MessagingConnection {
id: record.id.to_string(),
display_name: record.display_name.clone(),
provider_type: record.provider.to_string(),
credential_id: record.credential_id.to_string(),
transport_settings: to_string(&record.transport).expect("connection transport serializes"),
enabled: record.enabled,
retention_seconds: i64::try_from(record.retention.seconds).expect("validated retention fits storage"),
}
}
fn validate_connection_provider(
provider: ListenerType,
transport: &ConnectionTransport,
) -> Result<(), MessagingError> {
if provider == transport.provider() {
Ok(())
} else {
Err(MessagingError::InboundConfigurationInvalid {
provider,
reason: "provider does not match transport".to_owned(),
})
}
}
}