use serde_json::Value;
use crate::internal::auth::session::{AsyncSessionProvider, SessionProvider};
use crate::internal::transport::{
AsyncAttachmentObjectBody, AsyncAttachmentObjectTransport, AsyncAuthenticatedRpcTransport,
AttachmentObjectTransport, AuthenticatedRpcTransport,
};
const MESSAGE_RPC_ENDPOINT: &str = crate::internal::message_runtime::direct::MESSAGE_RPC_ENDPOINT;
pub(crate) struct AttachmentUploadRuntime<'a, P, T> {
client: &'a crate::core::ImClient,
session_provider: P,
transport: T,
}
pub(crate) struct AttachmentSendInput {
pub target: crate::messages::MessageTarget,
pub request: crate::attachments::AttachmentSendRequest,
pub resolved_target_did: Option<String>,
pub client_message_id: Option<crate::ids::MessageId>,
pub credentials: Option<AttachmentUploadCredentials>,
}
pub(crate) struct AttachmentPrepareObjectInput {
pub target: crate::messages::MessageTarget,
pub request: crate::attachments::AttachmentSendRequest,
pub resolved_target_did: Option<String>,
pub message_security_profile: &'static str,
}
#[derive(Debug, Clone, PartialEq)]
pub(crate) struct AttachmentUploadCredentials {
pub identity_name: String,
pub did_document: Option<Value>,
pub key1_private_pem: String,
}
#[derive(Debug, Clone, PartialEq)]
pub struct AttachmentUploadResult {
pub sdk_result: crate::messages::SendMessageResult,
pub target_kind: &'static str,
pub target_did: String,
pub prepared: crate::attachments::manifest::PreparedAttachment,
pub slot: crate::internal::wire::attachment::AttachmentCreateSlotResult,
pub manifest: Value,
pub raw: Value,
}
#[derive(Debug, Clone, PartialEq)]
pub(crate) struct PreparedCommittedAttachment {
pub target_kind: &'static str,
pub target_did: String,
pub prepared: crate::attachments::manifest::PreparedAttachment,
pub slot: crate::internal::wire::attachment::AttachmentCreateSlotResult,
pub descriptor: crate::attachments::manifest::AttachmentDescriptor,
pub redacted_manifest: Value,
pub full_manifest: Value,
pub grant_ref: Value,
}
struct PreparedAttachmentUpload {
prepared: crate::attachments::manifest::PreparedAttachment,
caption: String,
mention_payload: Option<Value>,
body: PreparedAttachmentBody,
object_e2ee_secrets: Option<crate::attachments::manifest::ObjectE2eeAttachmentSecrets>,
}
struct ManifestSendResult {
raw: Value,
meta: Value,
}
enum PreparedAttachmentBody {
Bytes(Vec<u8>),
File(std::path::PathBuf),
}
impl PreparedAttachmentBody {
fn clone_for_sync(&self) -> Vec<u8> {
match self {
Self::Bytes(bytes) => bytes.clone(),
Self::File(_) => Vec::new(),
}
}
}
impl<'a, P, T> AttachmentUploadRuntime<'a, P, T>
where
P: SessionProvider,
T: AuthenticatedRpcTransport + AttachmentObjectTransport,
{
pub(crate) fn new(
client: &'a crate::core::ImClient,
session_provider: P,
transport: T,
) -> Self {
Self {
client,
session_provider,
transport,
}
}
pub(crate) fn send(
mut self,
input: AttachmentSendInput,
) -> crate::ImResult<AttachmentUploadResult> {
let target = send_target(&input.target, input.resolved_target_did)?;
let service_did = message_service_did(self.client)?;
self.session_provider.ensure_session(auth_scope(&target))?;
let operation_id = input.request.delivery.idempotency_key.clone();
let upload = prepare_request(input.request)?;
let prepared = upload.prepared;
let slot = self.create_slot(&target, &service_did, &prepared)?;
self.transport.put_attachment_object(
slot.upload_uri.as_str(),
upload_headers(&slot.upload_headers),
upload.body.clone_for_sync(),
)?;
self.commit_object(&service_did, &prepared, &slot)?;
let descriptor = crate::attachments::manifest::AttachmentDescriptor::from_prepared(
&prepared,
slot.attachment_id.clone(),
slot.object_uri.clone(),
);
let manifest =
crate::attachments::manifest::build_attachment_manifest_with_mention_payload(
&descriptor,
&upload.caption,
upload.mention_payload.as_ref(),
)?;
let credentials = match input.credentials {
Some(credentials) => credentials,
None => load_credentials(self.client)?,
};
let send_result = self.send_manifest(
&target,
manifest.clone(),
input.client_message_id.as_ref(),
operation_id.as_deref(),
credentials,
)?;
let sdk_result = sdk_result_from_raw(
send_result.raw.clone(),
&send_result.meta,
self.client.did().clone(),
&target,
&manifest,
)?;
Ok(AttachmentUploadResult {
sdk_result,
target_kind: target.kind(),
target_did: target.did().to_string(),
prepared,
slot,
manifest,
raw: send_result.raw,
})
}
pub(crate) fn prepare_and_commit_object(
mut self,
input: AttachmentPrepareObjectInput,
) -> crate::ImResult<PreparedCommittedAttachment> {
let target = send_target(&input.target, input.resolved_target_did)?;
let service_did = message_service_did(self.client)?;
self.session_provider.ensure_session(auth_scope(&target))?;
let upload = prepare_request_object_e2ee(input.request)?;
let prepared = upload.prepared;
let slot = self.create_slot_with_security_profile(
&target,
&service_did,
input.message_security_profile,
&prepared,
)?;
self.transport.put_attachment_object(
slot.upload_uri.as_str(),
upload_headers(&slot.upload_headers),
upload.body.clone_for_sync(),
)?;
self.commit_object(&service_did, &prepared, &slot)?;
let descriptor = crate::attachments::manifest::AttachmentDescriptor::from_prepared(
&prepared,
slot.attachment_id.clone(),
slot.object_uri.clone(),
);
let redacted_manifest =
crate::attachments::manifest::build_attachment_manifest_with_mention_payload(
&descriptor,
&upload.caption,
upload.mention_payload.as_ref(),
)?;
let secrets =
upload
.object_e2ee_secrets
.as_ref()
.ok_or_else(|| crate::ImError::Internal {
message: "object-e2ee attachment secrets missing after encryption".to_owned(),
})?;
let full_manifest = crate::attachments::manifest::build_attachment_manifest_with_object_e2ee_secrets_and_mention_payload(
&descriptor,
&upload.caption,
secrets,
upload.mention_payload.as_ref(),
)?;
let grant_ref = crate::attachments::manifest::build_attachment_grant_ref(&descriptor)?;
Ok(PreparedCommittedAttachment {
target_kind: target.kind(),
target_did: target.did().to_string(),
prepared,
slot,
descriptor,
redacted_manifest,
full_manifest,
grant_ref,
})
}
fn create_slot(
&mut self,
target: &ResolvedAttachmentTarget,
service_did: &str,
prepared: &crate::attachments::manifest::PreparedAttachment,
) -> crate::ImResult<crate::internal::wire::attachment::AttachmentCreateSlotResult> {
self.create_slot_with_security_profile(target, service_did, "transport-protected", prepared)
}
fn create_slot_with_security_profile(
&mut self,
target: &ResolvedAttachmentTarget,
service_did: &str,
message_security_profile: &str,
prepared: &crate::attachments::manifest::PreparedAttachment,
) -> crate::ImResult<crate::internal::wire::attachment::AttachmentCreateSlotResult> {
let params =
crate::internal::wire::attachment::build_attachment_create_slot_rpc_params_with_security_profile(
self.client.did().as_str(),
service_did,
target.kind(),
target.did(),
message_security_profile,
prepared,
)?;
let raw = self.transport.authenticated_rpc(
MESSAGE_RPC_ENDPOINT,
"attachment.create_slot",
params,
)?;
let mut slot: crate::internal::wire::attachment::AttachmentCreateSlotResult =
serde_json::from_value(raw).map_err(|err| crate::ImError::Serialization {
detail: err.to_string(),
})?;
slot.request_service_did = service_did.to_string();
Ok(slot)
}
fn commit_object(
&mut self,
service_did: &str,
prepared: &crate::attachments::manifest::PreparedAttachment,
slot: &crate::internal::wire::attachment::AttachmentCreateSlotResult,
) -> crate::ImResult<()> {
let params = crate::internal::wire::attachment::build_attachment_commit_object_rpc_params(
self.client.did().as_str(),
service_did,
prepared,
slot,
)?;
let _: crate::internal::wire::attachment::AttachmentCommitObjectResult =
serde_json::from_value(self.transport.authenticated_rpc(
MESSAGE_RPC_ENDPOINT,
"attachment.commit_object",
params,
)?)
.map_err(|err| crate::ImError::Serialization {
detail: err.to_string(),
})?;
Ok(())
}
fn send_manifest(
&mut self,
target: &ResolvedAttachmentTarget,
manifest: Value,
client_message_id: Option<&crate::ids::MessageId>,
operation_id: Option<&str>,
credentials: AttachmentUploadCredentials,
) -> crate::ImResult<ManifestSendResult> {
let identity = crate::internal::wire::attachment::AttachmentSigningIdentity {
identity_name: credentials.identity_name,
did: self.client.did().as_str().to_string(),
did_document: credentials.did_document,
key1_private_pem: credentials.key1_private_pem,
};
let (method, params) = match target {
ResolvedAttachmentTarget::Direct { target_did, .. } => (
"direct.send",
crate::internal::wire::attachment::build_direct_attachment_send_rpc_params(
&identity,
target_did,
manifest,
client_message_id,
operation_id,
)?,
),
ResolvedAttachmentTarget::Group { group } => (
"group.send",
crate::internal::wire::attachment::build_group_attachment_send_rpc_params(
&identity,
group.as_str(),
manifest,
client_message_id,
operation_id,
)?,
),
};
let meta = params.get("meta").cloned().unwrap_or(Value::Null);
let raw = self
.transport
.authenticated_rpc(MESSAGE_RPC_ENDPOINT, method, params)?;
Ok(ManifestSendResult { raw, meta })
}
}
impl<'a, P, T> AttachmentUploadRuntime<'a, P, T>
where
P: AsyncSessionProvider,
T: AsyncAuthenticatedRpcTransport + AsyncAttachmentObjectTransport,
{
pub(crate) async fn send_async(
mut self,
input: AttachmentSendInput,
) -> crate::ImResult<AttachmentUploadResult> {
let target = send_target(&input.target, input.resolved_target_did)?;
let service_did = message_service_did(self.client)?;
self.session_provider
.ensure_session(auth_scope(&target))
.await?;
let operation_id = input.request.delivery.idempotency_key.clone();
let upload = prepare_request_async(input.request).await?;
let prepared = upload.prepared;
let slot = self
.create_slot_async(&target, &service_did, &prepared)
.await?;
let body = async_object_body(&prepared, upload.body)?;
self.transport
.put_attachment_object_stream(
slot.upload_uri.as_str(),
upload_headers(&slot.upload_headers),
body,
)
.await?;
self.commit_object_async(&service_did, &prepared, &slot)
.await?;
let descriptor = crate::attachments::manifest::AttachmentDescriptor::from_prepared(
&prepared,
slot.attachment_id.clone(),
slot.object_uri.clone(),
);
let manifest =
crate::attachments::manifest::build_attachment_manifest_with_mention_payload(
&descriptor,
&upload.caption,
upload.mention_payload.as_ref(),
)?;
let credentials = match input.credentials {
Some(credentials) => credentials,
None => load_credentials_async(self.client).await?,
};
let send_result = self
.send_manifest_async(
&target,
manifest.clone(),
input.client_message_id.as_ref(),
operation_id.as_deref(),
credentials,
)
.await?;
let sdk_result = sdk_result_from_raw(
send_result.raw.clone(),
&send_result.meta,
self.client.did().clone(),
&target,
&manifest,
)?;
Ok(AttachmentUploadResult {
sdk_result,
target_kind: target.kind(),
target_did: target.did().to_string(),
prepared,
slot,
manifest,
raw: send_result.raw,
})
}
pub(crate) async fn prepare_and_commit_object_async(
mut self,
input: AttachmentPrepareObjectInput,
) -> crate::ImResult<PreparedCommittedAttachment> {
let target = send_target(&input.target, input.resolved_target_did)?;
let service_did = message_service_did(self.client)?;
self.session_provider
.ensure_session(auth_scope(&target))
.await?;
let upload = prepare_request_object_e2ee_async(input.request).await?;
let prepared = upload.prepared;
let slot = self
.create_slot_with_security_profile_async(
&target,
&service_did,
input.message_security_profile,
&prepared,
)
.await?;
let body = async_object_body(&prepared, upload.body)?;
self.transport
.put_attachment_object_stream(
slot.upload_uri.as_str(),
upload_headers(&slot.upload_headers),
body,
)
.await?;
self.commit_object_async(&service_did, &prepared, &slot)
.await?;
let descriptor = crate::attachments::manifest::AttachmentDescriptor::from_prepared(
&prepared,
slot.attachment_id.clone(),
slot.object_uri.clone(),
);
let redacted_manifest =
crate::attachments::manifest::build_attachment_manifest_with_mention_payload(
&descriptor,
&upload.caption,
upload.mention_payload.as_ref(),
)?;
let secrets =
upload
.object_e2ee_secrets
.as_ref()
.ok_or_else(|| crate::ImError::Internal {
message: "object-e2ee attachment secrets missing after encryption".to_owned(),
})?;
let full_manifest = crate::attachments::manifest::build_attachment_manifest_with_object_e2ee_secrets_and_mention_payload(
&descriptor,
&upload.caption,
secrets,
upload.mention_payload.as_ref(),
)?;
let grant_ref = crate::attachments::manifest::build_attachment_grant_ref(&descriptor)?;
Ok(PreparedCommittedAttachment {
target_kind: target.kind(),
target_did: target.did().to_string(),
prepared,
slot,
descriptor,
redacted_manifest,
full_manifest,
grant_ref,
})
}
async fn create_slot_async(
&mut self,
target: &ResolvedAttachmentTarget,
service_did: &str,
prepared: &crate::attachments::manifest::PreparedAttachment,
) -> crate::ImResult<crate::internal::wire::attachment::AttachmentCreateSlotResult> {
self.create_slot_with_security_profile_async(
target,
service_did,
"transport-protected",
prepared,
)
.await
}
async fn create_slot_with_security_profile_async(
&mut self,
target: &ResolvedAttachmentTarget,
service_did: &str,
message_security_profile: &str,
prepared: &crate::attachments::manifest::PreparedAttachment,
) -> crate::ImResult<crate::internal::wire::attachment::AttachmentCreateSlotResult> {
let params =
crate::internal::wire::attachment::build_attachment_create_slot_rpc_params_with_security_profile(
self.client.did().as_str(),
service_did,
target.kind(),
target.did(),
message_security_profile,
prepared,
)?;
let raw = self
.transport
.authenticated_rpc(MESSAGE_RPC_ENDPOINT, "attachment.create_slot", params)
.await?;
let mut slot: crate::internal::wire::attachment::AttachmentCreateSlotResult =
serde_json::from_value(raw).map_err(|err| crate::ImError::Serialization {
detail: err.to_string(),
})?;
slot.request_service_did = service_did.to_string();
Ok(slot)
}
async fn commit_object_async(
&mut self,
service_did: &str,
prepared: &crate::attachments::manifest::PreparedAttachment,
slot: &crate::internal::wire::attachment::AttachmentCreateSlotResult,
) -> crate::ImResult<()> {
let params = crate::internal::wire::attachment::build_attachment_commit_object_rpc_params(
self.client.did().as_str(),
service_did,
prepared,
slot,
)?;
let _: crate::internal::wire::attachment::AttachmentCommitObjectResult =
serde_json::from_value(
self.transport
.authenticated_rpc(MESSAGE_RPC_ENDPOINT, "attachment.commit_object", params)
.await?,
)
.map_err(|err| crate::ImError::Serialization {
detail: err.to_string(),
})?;
Ok(())
}
async fn send_manifest_async(
&mut self,
target: &ResolvedAttachmentTarget,
manifest: Value,
client_message_id: Option<&crate::ids::MessageId>,
operation_id: Option<&str>,
credentials: AttachmentUploadCredentials,
) -> crate::ImResult<ManifestSendResult> {
let identity = crate::internal::wire::attachment::AttachmentSigningIdentity {
identity_name: credentials.identity_name,
did: self.client.did().as_str().to_string(),
did_document: credentials.did_document,
key1_private_pem: credentials.key1_private_pem,
};
let (method, params) = match target {
ResolvedAttachmentTarget::Direct { target_did, .. } => (
"direct.send",
crate::internal::wire::attachment::build_direct_attachment_send_rpc_params(
&identity,
target_did,
manifest,
client_message_id,
operation_id,
)?,
),
ResolvedAttachmentTarget::Group { group } => (
"group.send",
crate::internal::wire::attachment::build_group_attachment_send_rpc_params(
&identity,
group.as_str(),
manifest,
client_message_id,
operation_id,
)?,
),
};
let meta = params.get("meta").cloned().unwrap_or(Value::Null);
let raw = self
.transport
.authenticated_rpc(MESSAGE_RPC_ENDPOINT, method, params)
.await?;
Ok(ManifestSendResult { raw, meta })
}
}
fn send_target(
target: &crate::messages::MessageTarget,
resolved_target_did: Option<String>,
) -> crate::ImResult<ResolvedAttachmentTarget> {
match target {
crate::messages::MessageTarget::Direct(peer) => {
let resolved = resolved_target_did
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())
.unwrap_or_else(|| peer.as_str().trim());
if !resolved.starts_with("did:") {
return Err(crate::ImError::PeerNotFound {
peer: peer.as_str().to_string(),
});
}
Ok(ResolvedAttachmentTarget::Direct {
peer: peer.clone(),
target_did: resolved.to_string(),
})
}
crate::messages::MessageTarget::Group(group) => {
if group.as_str().trim().is_empty() {
return Err(crate::ImError::invalid_input(
Some("group".to_string()),
"group target is required",
));
}
Ok(ResolvedAttachmentTarget::Group {
group: group.clone(),
})
}
}
}
fn auth_scope(target: &ResolvedAttachmentTarget) -> crate::auth::AuthScope {
match target {
ResolvedAttachmentTarget::Direct { .. } => crate::auth::AuthScope::Messaging,
ResolvedAttachmentTarget::Group { .. } => crate::auth::AuthScope::GroupMessaging,
}
}
fn prepare_request(
request: crate::attachments::AttachmentSendRequest,
) -> crate::ImResult<PreparedAttachmentUpload> {
let caption = request.caption.unwrap_or_default();
let mention_payload = validated_mention_payload(request.mention_payload)?;
let request_filename = request.filename;
let request_mime = request.mime_type;
let source = crate::internal::blob::source::attachment_input_to_blob_source(request.input)?;
let filename = request_filename
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned)
.or_else(|| {
source
.filename
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned)
});
let mime_type = request_mime
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned)
.or_else(|| {
source
.mime_type
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned)
})
.unwrap_or_default();
let prepared = if let Some(filename) = filename {
crate::attachments::manifest::prepare_attachment_payload(
&filename,
&mime_type,
source.bytes,
)?
} else if let Some(path) = source.path.as_deref() {
crate::attachments::manifest::prepare_attachment_payload_from_path(
path,
&mime_type,
source.bytes,
)?
} else {
return Err(crate::ImError::invalid_input(
Some("filename".to_string()),
"attachment filename is required",
));
};
let body = PreparedAttachmentBody::Bytes(prepared.payload.clone());
Ok(PreparedAttachmentUpload {
prepared,
caption,
mention_payload,
body,
object_e2ee_secrets: None,
})
}
fn prepare_request_object_e2ee(
request: crate::attachments::AttachmentSendRequest,
) -> crate::ImResult<PreparedAttachmentUpload> {
let caption = request.caption.unwrap_or_default();
let mention_payload = validated_mention_payload(request.mention_payload)?;
let request_filename = request.filename;
let request_mime = request.mime_type;
let source = crate::internal::blob::source::attachment_input_to_blob_source(request.input)?;
let filename = request_filename
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned)
.or_else(|| {
source
.filename
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned)
});
let mime_type = request_mime
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned)
.or_else(|| {
source
.mime_type
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned)
})
.unwrap_or_default();
let e2ee = if let Some(filename) = filename {
crate::attachments::manifest::prepare_object_e2ee_attachment_payload(
&filename,
&mime_type,
source.bytes,
)?
} else if let Some(path) = source.path.as_deref() {
let prepared = crate::attachments::manifest::prepare_attachment_payload_from_path(
path,
&mime_type,
source.bytes,
)?;
crate::attachments::manifest::prepare_object_e2ee_attachment_payload(
&prepared.filename,
&prepared.mime_type,
prepared.payload,
)?
} else {
return Err(crate::ImError::invalid_input(
Some("filename".to_string()),
"attachment filename is required",
));
};
let body = PreparedAttachmentBody::Bytes(e2ee.prepared.payload.clone());
Ok(PreparedAttachmentUpload {
prepared: e2ee.prepared,
caption,
mention_payload,
body,
object_e2ee_secrets: Some(e2ee.secrets),
})
}
async fn prepare_request_async(
request: crate::attachments::AttachmentSendRequest,
) -> crate::ImResult<PreparedAttachmentUpload> {
let caption = request.caption.unwrap_or_default();
let mention_payload = validated_mention_payload(request.mention_payload)?;
let request_filename = request.filename;
let request_mime = request.mime_type;
let source =
crate::internal::blob::source::attachment_input_to_async_blob_source(request.input)?;
let prepared = match source {
crate::internal::blob::source::AsyncBlobSource::Memory {
filename,
mime_type,
bytes,
} => {
let filename = request_filename
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned)
.or_else(|| {
filename
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned)
});
let mime_type = request_mime
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned)
.or_else(|| {
mime_type
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned)
})
.unwrap_or_default();
let Some(filename) = filename else {
return Err(crate::ImError::invalid_input(
Some("filename".to_string()),
"attachment filename is required",
));
};
let prepared = crate::attachments::manifest::prepare_attachment_payload(
&filename, &mime_type, bytes,
)?;
let body = PreparedAttachmentBody::Bytes(prepared.payload.clone());
PreparedAttachmentUpload {
prepared,
caption,
mention_payload: mention_payload.clone(),
body,
object_e2ee_secrets: None,
}
}
crate::internal::blob::source::AsyncBlobSource::LocalFile { path } => {
let filename = request_filename
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned);
let mime_type = request_mime
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned)
.unwrap_or_default();
let mut prepared = crate::attachments::manifest::prepare_attachment_metadata_from_path(
&path, &mime_type,
)
.await?;
if let Some(filename) = filename {
prepared.filename = filename;
}
PreparedAttachmentUpload {
prepared,
caption,
mention_payload: mention_payload.clone(),
body: PreparedAttachmentBody::File(path),
object_e2ee_secrets: None,
}
}
};
Ok(prepared)
}
async fn prepare_request_object_e2ee_async(
request: crate::attachments::AttachmentSendRequest,
) -> crate::ImResult<PreparedAttachmentUpload> {
let caption = request.caption.unwrap_or_default();
let mention_payload = validated_mention_payload(request.mention_payload)?;
let request_filename = request.filename;
let request_mime = request.mime_type;
let source =
crate::internal::blob::source::attachment_input_to_async_blob_source(request.input)?;
let (filename, mime_type, plaintext) = match source {
crate::internal::blob::source::AsyncBlobSource::Memory {
filename,
mime_type,
bytes,
} => {
let filename = request_filename
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned)
.or_else(|| {
filename
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned)
})
.ok_or_else(|| {
crate::ImError::invalid_input(
Some("filename".to_string()),
"attachment filename is required",
)
})?;
let mime_type = request_mime
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned)
.or_else(|| {
mime_type
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned)
})
.unwrap_or_default();
(filename, mime_type, bytes)
}
crate::internal::blob::source::AsyncBlobSource::LocalFile { path } => {
let mut prepared = crate::attachments::manifest::prepare_attachment_metadata_from_path(
&path,
request_mime.as_deref().unwrap_or_default(),
)
.await?;
if let Some(filename) = request_filename
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned)
{
prepared.filename = filename;
}
let plaintext = tokio::fs::read(&path)
.await
.map_err(|err| crate::ImError::Io {
detail: format!("read attachment file {}: {err}", path.display()),
})?;
(prepared.filename, prepared.mime_type, plaintext)
}
};
let e2ee = crate::attachments::manifest::prepare_object_e2ee_attachment_payload(
&filename, &mime_type, plaintext,
)?;
let body = PreparedAttachmentBody::Bytes(e2ee.prepared.payload.clone());
Ok(PreparedAttachmentUpload {
prepared: e2ee.prepared,
caption,
mention_payload,
body,
object_e2ee_secrets: Some(e2ee.secrets),
})
}
fn validated_mention_payload(payload: Option<Value>) -> crate::ImResult<Option<Value>> {
if let Some(payload) = payload.as_ref() {
crate::messages::validate_message_mention_payload(payload)?;
}
Ok(payload)
}
fn async_object_body(
prepared: &crate::attachments::manifest::PreparedAttachment,
body: PreparedAttachmentBody,
) -> crate::ImResult<AsyncAttachmentObjectBody> {
match body {
PreparedAttachmentBody::Bytes(bytes) => Ok(AsyncAttachmentObjectBody::Bytes(bytes)),
PreparedAttachmentBody::File(path) => Ok(AsyncAttachmentObjectBody::File {
path,
len: prepared.size_bytes,
content_type: Some(prepared.mime_type.clone()),
}),
}
}
fn message_service_did(client: &crate::core::ImClient) -> crate::ImResult<String> {
client
.core_inner()
.sdk_config()
.anp_service_did
.as_ref()
.map(|did| did.as_str().to_string())
.filter(|did| !did.trim().is_empty())
.ok_or_else(|| {
crate::ImError::invalid_input(
Some("service_did".to_string()),
"message service did is required",
)
})
}
fn upload_headers(
headers: &serde_json::Map<String, Value>,
) -> std::collections::BTreeMap<String, String> {
headers
.iter()
.filter_map(|(key, value)| value.as_str().map(|value| (key.clone(), value.to_string())))
.collect()
}
fn load_credentials(
client: &crate::core::ImClient,
) -> crate::ImResult<AttachmentUploadCredentials> {
let runtime = client.runtime();
let did_document = runtime.key_provider.optional_did_document()?;
let key1_private_pem = runtime.key_provider.default_signing_private_pem()?;
Ok(AttachmentUploadCredentials {
identity_name: runtime.owner.identity_id.as_str().to_string(),
did_document,
key1_private_pem,
})
}
async fn load_credentials_async(
client: &crate::core::ImClient,
) -> crate::ImResult<AttachmentUploadCredentials> {
let runtime = client.runtime();
let did_document = runtime.key_provider.optional_did_document()?;
let key1_private_pem = runtime.key_provider.default_signing_private_pem()?;
Ok(AttachmentUploadCredentials {
identity_name: runtime.owner.identity_id.as_str().to_string(),
did_document,
key1_private_pem,
})
}
fn sdk_result_from_raw(
raw: Value,
meta: &Value,
sender: crate::ids::Did,
target: &ResolvedAttachmentTarget,
manifest: &Value,
) -> crate::ImResult<crate::messages::SendMessageResult> {
match target {
ResolvedAttachmentTarget::Direct { peer, target_did } => {
let mut result: DirectAttachmentRpcResult =
serde_json::from_value(raw).map_err(|err| crate::ImError::Serialization {
detail: err.to_string(),
})?;
fill_direct_result_defaults(&mut result, meta, target_did);
sdk_result_from_direct_result(&result, sender, peer.clone(), manifest)
}
ResolvedAttachmentTarget::Group { group } => {
let mut result: GroupAttachmentRpcResult =
serde_json::from_value(raw).map_err(|err| crate::ImError::Serialization {
detail: err.to_string(),
})?;
fill_group_result_defaults(&mut result, meta, group.as_str());
sdk_result_from_group_result(&result, sender, group.clone(), manifest)
}
}
}
fn fill_direct_result_defaults(
result: &mut DirectAttachmentRpcResult,
meta: &Value,
target_did: &str,
) {
if result.message_id.trim().is_empty() {
result.message_id = string_value(meta.get("message_id")).unwrap_or_else(|| {
format!(
"msg-{}",
crate::internal::wire::common::generate_operation_id()
)
});
}
if result.operation_id.trim().is_empty() {
result.operation_id = string_value(meta.get("operation_id")).unwrap_or_else(|| {
format!(
"op-{}",
crate::internal::wire::common::generate_operation_id()
)
});
}
if result.target_did.trim().is_empty() {
result.target_did = target_did.to_string();
}
}
fn fill_group_result_defaults(
result: &mut GroupAttachmentRpcResult,
meta: &Value,
group_did: &str,
) {
if result.group_did.trim().is_empty() {
result.group_did = group_did.to_string();
}
if result.message_id.trim().is_empty() && result.group_event_seq.trim().is_empty() {
result.message_id = string_value(meta.get("message_id")).unwrap_or_else(|| {
format!(
"msg-{}",
crate::internal::wire::common::generate_operation_id()
)
});
}
if result.operation_id.trim().is_empty() {
result.operation_id = string_value(meta.get("operation_id")).unwrap_or_else(|| {
format!(
"op-{}",
crate::internal::wire::common::generate_operation_id()
)
});
}
}
fn string_value(value: Option<&Value>) -> Option<String> {
value
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned)
}
fn sdk_result_from_direct_result(
result: &DirectAttachmentRpcResult,
sender: crate::ids::Did,
peer: crate::ids::PeerRef,
manifest: &Value,
) -> crate::ImResult<crate::messages::SendMessageResult> {
let message_id = crate::ids::MessageId::parse(&result.message_id)?;
let delivery = direct_delivery_state(result);
let (send_state, retry_plan) =
crate::internal::message_runtime::state::send_state_from_delivery(
&delivery,
Some(result.operation_id.clone()).filter(|value| !value.trim().is_empty()),
Some(message_id.clone()),
Some(result.accepted_at.clone()).filter(|value| !value.trim().is_empty()),
None,
);
let delivery_state = Some(result.delivery_state.clone())
.filter(|value| !value.trim().is_empty())
.unwrap_or_else(|| {
crate::internal::message_runtime::state::send_state_label(&send_state.state).to_string()
});
Ok(crate::messages::SendMessageResult {
message: crate::messages::Message {
id: message_id,
thread: crate::messages::ThreadRef::Direct(peer.clone()),
direction: crate::messages::MessageDirection::Outgoing,
sender: crate::ids::PeerRef::parse(sender.as_str(), "")?,
receiver: Some(peer),
group: None,
body: crate::messages::MessageBodyView::Unsupported {
content_type: Some(
crate::attachments::manifest::attachment_manifest_content_type().to_string(),
),
},
sent_at: Some(result.accepted_at.clone()).filter(|value| !value.trim().is_empty()),
received_at: None,
metadata: attachment_metadata(
Some(result.operation_id.clone()),
Some(delivery_state),
Some(send_state),
retry_plan,
None,
manifest,
),
},
delivery,
warnings: Vec::new(),
})
}
fn sdk_result_from_group_result(
result: &GroupAttachmentRpcResult,
sender: crate::ids::Did,
group: crate::ids::GroupRef,
manifest: &Value,
) -> crate::ImResult<crate::messages::SendMessageResult> {
let message_id = group_message_id(group.as_str(), result)?;
let delivery = group_delivery_state(result);
let (send_state, retry_plan) =
crate::internal::message_runtime::state::send_state_from_delivery(
&delivery,
Some(result.operation_id.clone()).filter(|value| !value.trim().is_empty()),
Some(message_id.clone()),
Some(result.accepted_at.clone()).filter(|value| !value.trim().is_empty()),
None,
);
Ok(crate::messages::SendMessageResult {
message: crate::messages::Message {
id: message_id,
thread: crate::messages::ThreadRef::Group(group.clone()),
direction: crate::messages::MessageDirection::Outgoing,
sender: crate::ids::PeerRef::parse(sender.as_str(), "")?,
receiver: None,
group: Some(group),
body: crate::messages::MessageBodyView::Unsupported {
content_type: Some(
crate::attachments::manifest::attachment_manifest_content_type().to_string(),
),
},
sent_at: Some(result.accepted_at.clone()).filter(|value| !value.trim().is_empty()),
received_at: None,
metadata: attachment_metadata(
Some(result.operation_id.clone()),
Some(
crate::internal::message_runtime::state::send_state_label(&send_state.state)
.to_string(),
),
Some(send_state),
retry_plan,
result.group_event_seq.trim().parse().ok(),
manifest,
)
.with_attributes(group_metadata_attributes(result)),
},
delivery,
warnings: Vec::new(),
})
}
fn attachment_metadata(
operation_id: Option<String>,
delivery_state: Option<String>,
send_state: Option<crate::messages::MessageSendState>,
retry_plan: Option<crate::messages::MessageRetryPlan>,
server_sequence: Option<i64>,
manifest: &Value,
) -> crate::messages::MessageMetadata {
let attributes = vec![crate::messages::MessageMetadataAttribute {
key: "attachment_manifest".to_string(),
value: crate::attachments::manifest::manifest_content_string(manifest),
}];
crate::messages::MessageMetadata {
operation_id: operation_id.filter(|value| !value.trim().is_empty()),
delivery_state: delivery_state.filter(|value| !value.trim().is_empty()),
send_state,
retry_plan,
server_sequence,
content_type: Some(
crate::attachments::manifest::attachment_manifest_content_type().to_string(),
),
conversation_identity: None,
attributes,
}
}
trait WithAttributes {
fn with_attributes(self, attributes: Vec<crate::messages::MessageMetadataAttribute>) -> Self;
}
impl WithAttributes for crate::messages::MessageMetadata {
fn with_attributes(
mut self,
attributes: Vec<crate::messages::MessageMetadataAttribute>,
) -> Self {
self.attributes.extend(attributes);
self
}
}
fn group_message_id(
group_did: &str,
result: &GroupAttachmentRpcResult,
) -> crate::ImResult<crate::ids::MessageId> {
if !result.group_did.trim().is_empty() && !result.group_event_seq.trim().is_empty() {
return crate::ids::MessageId::parse(format!(
"{}:{}",
result.group_did.trim(),
result.group_event_seq.trim()
));
}
if !result.group_event_seq.trim().is_empty() {
return crate::ids::MessageId::parse(format!(
"{}:{}",
group_did.trim(),
result.group_event_seq.trim()
));
}
if !result.message_id.trim().is_empty() {
return crate::ids::MessageId::parse(&result.message_id);
}
crate::ids::MessageId::parse(format!(
"msg-{}",
crate::internal::wire::common::generate_operation_id()
))
}
fn group_metadata_attributes(
result: &GroupAttachmentRpcResult,
) -> Vec<crate::messages::MessageMetadataAttribute> {
let mut attributes = Vec::new();
if !result.message_id.trim().is_empty() {
attributes.push(crate::messages::MessageMetadataAttribute {
key: "raw_message_id".to_string(),
value: result.message_id.clone(),
});
}
if !result.group_event_seq.trim().is_empty() {
attributes.push(crate::messages::MessageMetadataAttribute {
key: "group_event_seq".to_string(),
value: result.group_event_seq.clone(),
});
}
if !result.group_state_version.trim().is_empty() {
attributes.push(crate::messages::MessageMetadataAttribute {
key: "group_state_version".to_string(),
value: result.group_state_version.clone(),
});
}
attributes
}
fn direct_delivery_state(result: &DirectAttachmentRpcResult) -> crate::messages::DeliveryState {
if !result.delivery_state.trim().is_empty() {
match result.delivery_state.trim().to_ascii_lowercase().as_str() {
"sent" => crate::messages::DeliveryState::Sent,
"stored_locally" | "stored-locally" => crate::messages::DeliveryState::StoredLocally,
"failed" => crate::messages::DeliveryState::Failed {
reason: result.delivery_state.clone(),
},
_ => crate::messages::DeliveryState::Accepted,
}
} else if result.accepted || result.final_acceptance {
crate::messages::DeliveryState::Accepted
} else {
crate::messages::DeliveryState::Failed {
reason: "not accepted".to_string(),
}
}
}
fn group_delivery_state(result: &GroupAttachmentRpcResult) -> crate::messages::DeliveryState {
if result.accepted || result.final_acceptance {
crate::messages::DeliveryState::Accepted
} else {
crate::messages::DeliveryState::Failed {
reason: "not accepted".to_string(),
}
}
}
#[derive(Debug, Clone, PartialEq)]
enum ResolvedAttachmentTarget {
Direct {
peer: crate::ids::PeerRef,
target_did: String,
},
Group {
group: crate::ids::GroupRef,
},
}
impl ResolvedAttachmentTarget {
fn kind(&self) -> &'static str {
match self {
Self::Direct { .. } => "agent",
Self::Group { .. } => "group",
}
}
fn did(&self) -> &str {
match self {
Self::Direct { target_did, .. } => target_did,
Self::Group { group } => group.as_str(),
}
}
}
#[derive(Debug, Clone, Default, serde::Deserialize, serde::Serialize, PartialEq)]
struct DirectAttachmentRpcResult {
#[serde(default)]
accepted: bool,
#[serde(default)]
message_id: String,
#[serde(default)]
operation_id: String,
#[serde(default)]
target_did: String,
#[serde(default)]
accepted_at: String,
#[serde(default)]
final_acceptance: bool,
#[serde(default)]
delivery_state: String,
}
#[derive(Debug, Clone, Default, serde::Deserialize, serde::Serialize, PartialEq)]
struct GroupAttachmentRpcResult {
#[serde(default)]
accepted: bool,
#[serde(default)]
final_acceptance: bool,
#[serde(default)]
group_did: String,
#[serde(default)]
message_id: String,
#[serde(default)]
operation_id: String,
#[serde(default)]
group_event_seq: String,
#[serde(default)]
group_state_version: String,
#[serde(default)]
accepted_at: String,
}
#[cfg(test)]
mod tests;