buzzert/Kordophone
Private
Public Access
1
0
Fork 0
Code Packages Activity

new message: initial commit

Browse Source
This commit is contained in:
James Magahern
2026-04-01 15:29:37 -07:00
parent 45285892de
commit c2a697f2c1
23 changed files with 674 additions and 262 deletions
Download Patch File Download Diff File Expand all files Collapse all files

69
core/kordophone/src/api/http_client.rs
View File

@@ -24,7 +24,7 @@ use tokio_tungstenite::{MaybeTlsStream, WebSocketStream};
use crate::{
model::{
Conversation, ConversationID, Event, JwtToken, Message, MessageID, OutgoingMessage,
UpdateItem,
OutgoingMessageTarget, ResolveHandleResponse, SendMessageResponse, UpdateItem,
},
APIInterface,
};
@@ -312,16 +312,26 @@ impl<K: AuthenticationStore + Send + Sync> APIInterface for HTTPAPIClient<K> {
async fn send_message(
&mut self,
outgoing_message: &OutgoingMessage,
) -> Result<Message, Self::Error> {
let message: Message = self
) -> Result<SendMessageResponse, Self::Error> {
let message: SendMessageResponse = self
.deserialized_response_with_body("sendMessage", Method::POST, || {
serde_json::to_string(&outgoing_message).unwrap().into()
Self::send_message_request_body(outgoing_message)
})
.await?;
Ok(message)
}
async fn resolve_handle(
&mut self,
handle_id: &str,
) -> Result<ResolveHandleResponse, Self::Error> {
let endpoint = format!("resolveHandle?id={}", urlencoding::encode(handle_id));
let response: ResolveHandleResponse =
self.deserialized_response(&endpoint, Method::GET).await?;
Ok(response)
}
async fn fetch_attachment_data(
&mut self,
guid: &str,
@@ -394,8 +404,7 @@ impl<K: AuthenticationStore + Send + Sync> APIInterface for HTTPAPIClient<K> {
None => "updates".to_string(),
};
let uri = self
.uri_for_endpoint(&endpoint, Some(self.websocket_scheme()))?;
let uri = self.uri_for_endpoint(&endpoint, Some(self.websocket_scheme()))?;
loop {
log::debug!("Connecting to websocket: {:?}", uri);
@@ -426,18 +435,20 @@ impl<K: AuthenticationStore + Send + Sync> APIInterface for HTTPAPIClient<K> {
log::debug!("Websocket request: {:?}", request);
let mut should_retry = true; // retry once after authenticating.
let should_retry = true; // retry once after authenticating.
match connect_async(request).await.map_err(Error::from) {
Ok((socket, response)) => {
log::debug!("Websocket connected: {:?}", response.status());
break Ok(WebsocketEventSocket::new(socket))
break Ok(WebsocketEventSocket::new(socket));
}
Err(e) => match &e {
Error::ClientError(ce) => match ce.as_str() {
"HTTP error: 401 Unauthorized" | "Unauthorized" => {
// Try to authenticate
if let Some(credentials) = &self.auth_store.get_credentials().await {
log::warn!("Websocket connection failed, attempting to authenticate");
log::warn!(
"Websocket connection failed, attempting to authenticate"
);
let new_token = self.authenticate(credentials.clone()).await?;
self.auth_store.set_token(new_token.to_string()).await;
@@ -473,16 +484,44 @@ impl<K: AuthenticationStore + Send + Sync> HTTPAPIClient<K> {
.build();
let client = Client::builder().build::<_, Body>(https);
HTTPAPIClient { base_url, auth_store, client }
HTTPAPIClient {
base_url,
auth_store,
client,
}
}
fn send_message_request_body(outgoing_message: &OutgoingMessage) -> Body {
#[derive(Serialize)]
struct SendMessageRequest<'a> {
#[serde(rename = "body")]
text: &'a str,
#[serde(rename = "guid", skip_serializing_if = "Option::is_none")]
conversation_id: Option<&'a ConversationID>,
#[serde(rename = "handleIDs", skip_serializing_if = "Option::is_none")]
handle_ids: Option<&'a [String]>,
#[serde(rename = "fileTransferGUIDs")]
file_transfer_guids: &'a Vec<String>,
}
let (conversation_id, handle_ids) = match &outgoing_message.target {
OutgoingMessageTarget::Conversation(conversation_id) => (Some(conversation_id), None),
OutgoingMessageTarget::Handles(handle_ids) => (None, Some(handle_ids.as_slice())),
};
serde_json::to_string(&SendMessageRequest {
text: &outgoing_message.text,
conversation_id,
handle_ids,
file_transfer_guids: &outgoing_message.file_transfer_guids,
})
.unwrap()
.into()
}
fn uri_for_endpoint(&self, endpoint: &str, scheme: Option<&str>) -> Result<Uri, Error> {
let mut parts = self.base_url.clone().into_parts();
let root_path: PathBuf = parts
.path_and_query
.ok_or(Error::URLError)?
.path()
.into();
let root_path: PathBuf = parts.path_and_query.ok_or(Error::URLError)?.path().into();
let path = root_path.join(endpoint);
let path_str = path.to_str().ok_or(Error::URLError)?;

13
core/kordophone/src/api/mod.rs
View File

@@ -1,4 +1,7 @@
pub use crate::model::{Conversation, ConversationID, Message, MessageID, OutgoingMessage};
pub use crate::model::{
Conversation, ConversationID, Message, MessageID, OutgoingMessage, ResolveHandleResponse,
SendMessageResponse,
};
use async_trait::async_trait;
use bytes::Bytes;
@@ -42,7 +45,13 @@ pub trait APIInterface {
async fn send_message(
&mut self,
outgoing_message: &OutgoingMessage,
) -> Result<Message, Self::Error>;
) -> Result<SendMessageResponse, Self::Error>;
// (GET) /resolveHandle
async fn resolve_handle(
&mut self,
handle_id: &str,
) -> Result<ResolveHandleResponse, Self::Error>;
// (GET) /attachment
async fn fetch_attachment_data(

28
core/kordophone/src/model/handle.rs Normal file
View File

@@ -0,0 +1,28 @@
use serde::{Deserialize, Serialize};
use super::conversation::ConversationID;
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct ResolvedHandle {
pub id: String,
pub name: Option<String>,
}
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "lowercase")]
pub enum HandleResolutionStatus {
Valid,
Invalid,
Unknown,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct ResolveHandleResponse {
#[serde(rename = "resolvedHandle")]
pub resolved_handle: ResolvedHandle,
pub status: HandleResolutionStatus,
#[serde(rename = "existingChat")]
pub existing_chat: Option<ConversationID>,
}

9
core/kordophone/src/model/mod.rs
View File

@@ -1,7 +1,9 @@
pub mod conversation;
pub mod event;
pub mod handle;
pub mod message;
pub mod outgoing_message;
pub mod send_message_response;
pub mod update;
pub use conversation::Conversation;
@@ -10,8 +12,15 @@ pub use conversation::ConversationID;
pub use message::Message;
pub use message::MessageID;
pub use handle::HandleResolutionStatus;
pub use handle::ResolveHandleResponse;
pub use handle::ResolvedHandle;
pub use outgoing_message::OutgoingMessage;
pub use outgoing_message::OutgoingMessageBuilder;
pub use outgoing_message::OutgoingMessageTarget;
pub use send_message_response::SendMessageResponse;
pub use update::UpdateItem;

46
core/kordophone/src/model/outgoing_message.rs
View File

@@ -1,23 +1,23 @@
use super::conversation::ConversationID;
use chrono::NaiveDateTime;
use serde::Serialize;
use uuid::Uuid;
#[derive(Debug, Clone, Serialize)]
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum OutgoingMessageTarget {
Conversation(ConversationID),
Handles(Vec<String>),
}
#[derive(Debug, Clone)]
pub struct OutgoingMessage {
#[serde(skip)]
pub guid: Uuid,
#[serde(skip)]
pub date: NaiveDateTime,
#[serde(rename = "body")]
pub text: String,
#[serde(rename = "guid")]
pub conversation_id: ConversationID,
pub target: OutgoingMessageTarget,
#[serde(rename = "fileTransferGUIDs")]
pub file_transfer_guids: Vec<String>,
}
@@ -25,13 +25,27 @@ impl OutgoingMessage {
pub fn builder() -> OutgoingMessageBuilder {
OutgoingMessageBuilder::new()
}
pub fn conversation_id(&self) -> Option<&ConversationID> {
match &self.target {
OutgoingMessageTarget::Conversation(conversation_id) => Some(conversation_id),
OutgoingMessageTarget::Handles(_) => None,
}
}
pub fn handle_ids(&self) -> Option<&[String]> {
match &self.target {
OutgoingMessageTarget::Conversation(_) => None,
OutgoingMessageTarget::Handles(handle_ids) => Some(handle_ids.as_slice()),
}
}
}
#[derive(Default)]
pub struct OutgoingMessageBuilder {
guid: Option<Uuid>,
text: Option<String>,
conversation_id: Option<ConversationID>,
target: Option<OutgoingMessageTarget>,
file_transfer_guids: Option<Vec<String>>,
}
@@ -50,8 +64,18 @@ impl OutgoingMessageBuilder {
self
}
pub fn target(mut self, target: OutgoingMessageTarget) -> Self {
self.target = Some(target);
self
}
pub fn conversation_id(mut self, conversation_id: ConversationID) -> Self {
self.conversation_id = Some(conversation_id);
self.target = Some(OutgoingMessageTarget::Conversation(conversation_id));
self
}
pub fn handle_ids(mut self, handle_ids: Vec<String>) -> Self {
self.target = Some(OutgoingMessageTarget::Handles(handle_ids));
self
}
@@ -64,7 +88,7 @@ impl OutgoingMessageBuilder {
OutgoingMessage {
guid: self.guid.unwrap_or_else(Uuid::new_v4),
text: self.text.unwrap(),
conversation_id: self.conversation_id.unwrap(),
target: self.target.unwrap(),
file_transfer_guids: self.file_transfer_guids.unwrap_or_default(),
date: chrono::Utc::now().naive_utc(),
}

12
core/kordophone/src/model/send_message_response.rs Normal file
View File

@@ -0,0 +1,12 @@
use serde::Deserialize;
use super::{conversation::ConversationID, message::Message};
#[derive(Debug, Clone, Deserialize)]
pub struct SendMessageResponse {
#[serde(flatten)]
pub message: Message,
#[serde(rename = "conversationGUID")]
pub conversation_id: Option<ConversationID>,
}

26
core/kordophone/src/tests/mod.rs
View File

@@ -3,7 +3,7 @@ use self::test_client::TestClient;
use crate::APIInterface;
pub mod api_interface {
use crate::model::Conversation;
use crate::model::{Conversation, HandleResolutionStatus, OutgoingMessage};
use super::*;
@@ -28,4 +28,28 @@ pub mod api_interface {
assert_eq!(conversations.len(), 1);
assert_eq!(conversations[0].display_name, test_convo.display_name);
}
#[tokio::test]
async fn test_resolve_handle() {
let mut client = TestClient::new();
let resolved = client.resolve_handle("user@example.com").await.unwrap();
assert_eq!(resolved.resolved_handle.id, "user@example.com");
assert_eq!(resolved.status, HandleResolutionStatus::Valid);
assert_eq!(resolved.existing_chat, None);
}
#[tokio::test]
async fn test_send_message_with_handles() {
let mut client = TestClient::new();
let outgoing_message = OutgoingMessage::builder()
.text("hello".to_string())
.handle_ids(vec!["user@example.com".to_string()])
.build();
let sent = client.send_message(&outgoing_message).await.unwrap();
assert_eq!(sent.message.text, "hello");
assert_eq!(sent.conversation_id, None);
}
}

66
core/kordophone/src/tests/test_client.rs
View File

@@ -9,14 +9,17 @@ use crate::{
api::event_socket::{EventSocket, SinkMessage, SocketEvent, SocketUpdate},
api::http_client::Credentials,
model::{
Conversation, ConversationID, Event, JwtToken, Message, MessageID, OutgoingMessage,
UpdateItem,
Conversation, ConversationID, Event, HandleResolutionStatus, JwtToken, Message, MessageID,
OutgoingMessage, OutgoingMessageTarget, ResolveHandleResponse, ResolvedHandle,
SendMessageResponse,
},
};
use bytes::Bytes;
use futures_util::sink::drain;
use futures_util::stream::BoxStream;
use futures_util::Sink;
use futures_util::SinkExt;
use futures_util::StreamExt;
pub struct TestClient {
@@ -63,13 +66,18 @@ impl EventSocket for TestEventSocket {
impl Sink<SinkMessage, Error = Self::Error>,
) {
(
futures_util::stream::iter(self.events.into_iter().map(Ok)).boxed(),
futures_util::sink::sink(),
futures_util::stream::iter(
self.events
.into_iter()
.map(|event| Ok(SocketEvent::Update(event))),
)
.boxed(),
drain().sink_map_err(|err| match err {}),
)
}
async fn raw_updates(self) -> Self::UpdateStream {
let results: Vec<Result<Vec<UpdateItem>, TestError>> = vec![];
let results: Vec<Result<SocketUpdate, TestError>> = vec![];
futures_util::stream::iter(results.into_iter()).boxed()
}
}
@@ -94,9 +102,9 @@ impl APIInterface for TestClient {
async fn get_messages(
&mut self,
conversation_id: &ConversationID,
limit: Option<u32>,
before: Option<MessageID>,
after: Option<MessageID>,
_limit: Option<u32>,
_before: Option<MessageID>,
_after: Option<MessageID>,
) -> Result<Vec<Message>, Self::Error> {
if let Some(messages) = self.messages.get(conversation_id) {
return Ok(messages.clone());
@@ -108,18 +116,42 @@ impl APIInterface for TestClient {
async fn send_message(
&mut self,
outgoing_message: &OutgoingMessage,
) -> Result<Message, Self::Error> {
) -> Result<SendMessageResponse, Self::Error> {
let message = Message::builder()
.guid(Uuid::new_v4().to_string())
.text(outgoing_message.text.clone())
.date(OffsetDateTime::now_utc())
.build();
let conversation_id = match &outgoing_message.target {
OutgoingMessageTarget::Conversation(conversation_id) => {
self.messages
.entry(outgoing_message.conversation_id.clone())
.entry(conversation_id.clone())
.or_insert(vec![])
.push(message.clone());
Ok(message)
None
}
OutgoingMessageTarget::Handles(_) => None,
};
Ok(SendMessageResponse {
message,
conversation_id,
})
}
async fn resolve_handle(
&mut self,
handle_id: &str,
) -> Result<ResolveHandleResponse, Self::Error> {
Ok(ResolveHandleResponse {
resolved_handle: ResolvedHandle {
id: handle_id.to_string(),
name: None,
},
status: HandleResolutionStatus::Valid,
existing_chat: None,
})
}
async fn open_event_socket(
@@ -131,17 +163,17 @@ impl APIInterface for TestClient {
async fn fetch_attachment_data(
&mut self,
guid: &str,
preview: bool,
_guid: &str,
_preview: bool,
) -> Result<Self::ResponseStream, Self::Error> {
Ok(futures_util::stream::iter(vec![Ok(Bytes::from_static(b"test"))]).boxed())
}
async fn upload_attachment<R>(
&mut self,
data: tokio::io::BufReader<R>,
filename: &str,
size: u64,
_data: tokio::io::BufReader<R>,
_filename: &str,
_size: u64,
) -> Result<String, Self::Error>
where
R: tokio::io::AsyncRead + Unpin + Send + Sync + 'static,
@@ -151,7 +183,7 @@ impl APIInterface for TestClient {
async fn mark_conversation_as_read(
&mut self,
conversation_id: &ConversationID,
_conversation_id: &ConversationID,
) -> Result<(), Self::Error> {
Ok(())
}

1
core/kordophoned-client/build.rs
View File

@@ -23,4 +23,3 @@ fn main() {
println!("cargo:rerun-if-changed={}", KORDOPHONE_XML);
}

1
core/kordophoned-client/src/lib.rs
View File

@@ -2,4 +2,3 @@ mod platform;
mod worker;
pub use worker::{spawn_worker, ChatMessage, ConversationSummary, Event, Request};

1
core/kordophoned-client/src/platform/linux.rs
View File

@@ -186,4 +186,3 @@ impl DaemonClient for DBusClient {
Ok(())
}
}

24
core/kordophoned-client/src/platform/macos.rs
View File

@@ -95,8 +95,14 @@ impl XpcClient {
impl DaemonClient for XpcClient {
fn get_conversations(&mut self, limit: i32, offset: i32) -> Result<Vec<ConversationSummary>> {
let mut args = HashMap::new();
args.insert(Self::key("limit"), Message::String(Self::key(&limit.to_string())));
args.insert(Self::key("offset"), Message::String(Self::key(&offset.to_string())));
args.insert(
Self::key("limit"),
Message::String(Self::key(&limit.to_string())),
);
args.insert(
Self::key("offset"),
Message::String(Self::key(&offset.to_string())),
);
let reply = self
.transport
@@ -112,7 +118,9 @@ impl DaemonClient for XpcClient {
let mut conversations = Vec::new();
for item in items {
let Message::Dictionary(conv) = item else { continue };
let Message::Dictionary(conv) = item else {
continue;
};
let id = Self::get_string(conv, "guid").unwrap_or_default();
let display_name = Self::get_string(conv, "display_name").unwrap_or_default();
let preview = Self::get_string(conv, "last_message_preview").unwrap_or_default();
@@ -162,7 +170,10 @@ impl DaemonClient for XpcClient {
Message::String(Self::key(&conversation_id)),
);
if let Some(last) = last_message_id {
args.insert(Self::key("last_message_id"), Message::String(Self::key(&last)));
args.insert(
Self::key("last_message_id"),
Message::String(Self::key(&last)),
);
}
let reply = self
@@ -178,7 +189,9 @@ impl DaemonClient for XpcClient {
let mut messages = Vec::new();
for item in items {
let Message::Dictionary(msg) = item else { continue };
let Message::Dictionary(msg) = item else {
continue;
};
messages.push(ChatMessage {
sender: Self::get_string(msg, "sender").unwrap_or_default(),
text: Self::get_string(msg, "text").unwrap_or_default(),
@@ -230,4 +243,3 @@ impl DaemonClient for XpcClient {
Ok(())
}
}

1
core/kordophoned-client/src/platform/mod.rs
View File

@@ -21,4 +21,3 @@ pub(crate) fn new_daemon_client() -> Result<Box<dyn DaemonClient>> {
anyhow::bail!("Unsupported platform")
}
}

47
core/kordophoned-client/src/worker.rs
View File

@@ -21,10 +21,19 @@ pub struct ChatMessage {
pub enum Request {
RefreshConversations,
RefreshMessages { conversation_id: String },
SendMessage { conversation_id: String, text: String },
MarkRead { conversation_id: String },
SyncConversation { conversation_id: String },
RefreshMessages {
conversation_id: String,
},
SendMessage {
conversation_id: String,
text: String,
},
MarkRead {
conversation_id: String,
},
SyncConversation {
conversation_id: String,
},
}
pub enum Event {
@@ -38,9 +47,13 @@ pub enum Event {
outgoing_id: Option<String>,
},
MarkedRead,
ConversationSyncTriggered { conversation_id: String },
ConversationSyncTriggered {
conversation_id: String,
},
ConversationsUpdated,
MessagesUpdated { conversation_id: String },
MessagesUpdated {
conversation_id: String,
},
UpdateStreamReconnected,
Error(String),
}
@@ -59,16 +72,19 @@ pub fn spawn_worker(
};
if let Err(e) = client.install_signal_handlers(event_tx.clone()) {
let _ = event_tx.send(Event::Error(format!("Failed to install daemon signals: {e}")));
let _ = event_tx.send(Event::Error(format!(
"Failed to install daemon signals: {e}"
)));
}
loop {
match request_rx.recv_timeout(Duration::from_millis(100)) {
Ok(req) => {
let res = match req {
Request::RefreshConversations => client
.get_conversations(200, 0)
.map(Event::Conversations),
let res =
match req {
Request::RefreshConversations => {
client.get_conversations(200, 0).map(Event::Conversations)
}
Request::RefreshMessages { conversation_id } => client
.get_messages(conversation_id.clone(), None)
.map(|messages| Event::Messages {
@@ -78,12 +94,12 @@ pub fn spawn_worker(
Request::SendMessage {
conversation_id,
text,
} => client
.send_message(conversation_id.clone(), text)
.map(|outgoing_id| Event::MessageSent {
} => client.send_message(conversation_id.clone(), text).map(
|outgoing_id| Event::MessageSent {
conversation_id,
outgoing_id,
}),
},
),
Request::MarkRead { conversation_id } => client
.mark_conversation_as_read(conversation_id.clone())
.map(|_| Event::MarkedRead),
@@ -130,4 +146,3 @@ pub(crate) trait DaemonClient {
Ok(())
}
}

1
core/kordophoned/build.rs
View File

@@ -27,4 +27,3 @@ fn main() {
println!("cargo:rerun-if-changed={}", KORDOPHONE_XML);
}

5
core/kordophoned/src/daemon/mod.rs
View File

@@ -477,9 +477,8 @@ impl Daemon {
.await;
// Convert DB messages to daemon model, substituting local_id when an alias exists.
let mut result: Vec<Message> = Vec::with_capacity(
db_messages.len() + outgoing_messages.len(),
);
let mut result: Vec<Message> =
Vec::with_capacity(db_messages.len() + outgoing_messages.len());
for m in db_messages.into_iter() {
let server_id = m.id.clone();
let mut dm: Message = m.into();

26
core/kordophoned/src/daemon/post_office.rs
View File

@@ -8,6 +8,7 @@ use tokio_condvar::Condvar;
use crate::daemon::events::Event as DaemonEvent;
use kordophone::api::APIInterface;
use kordophone::model::outgoing_message::OutgoingMessage;
use kordophone::model::OutgoingMessageTarget;
use anyhow::Result;
@@ -102,10 +103,29 @@ impl<C: APIInterface, F: AsyncFnMut() -> Result<C>> PostOffice<C, F> {
Ok(sent_message) => {
log::info!(target: target::POST_OFFICE, "Message sent successfully: {}", message.guid);
let conversation_id = message.conversation_id.clone();
let event =
DaemonEvent::MessageSent(sent_message.into(), message, conversation_id);
let conversation_id = sent_message.conversation_id.clone().or_else(|| {
match &message.target {
OutgoingMessageTarget::Conversation(conversation_id) => {
Some(conversation_id.clone())
}
OutgoingMessageTarget::Handles(_) => None,
}
});
if let Some(conversation_id) = conversation_id {
let event = DaemonEvent::MessageSent(
sent_message.message.into(),
message,
conversation_id,
);
event_sink.send(event).await.unwrap();
} else {
log::error!(
target: target::POST_OFFICE,
"Message sent but no conversation id was available for {}",
message.guid
);
}
}
Err(e) => {

10
core/kordophoned/src/xpc/mod.rs
View File

@@ -15,10 +15,16 @@ pub struct DispatchResult {
impl DispatchResult {
pub fn new(message: Message) -> Self {
Self { message, cleanup: None }
Self {
message,
cleanup: None,
}
}
pub fn with_cleanup<T: Any + Send + 'static>(message: Message, cleanup: T) -> Self {
Self { message, cleanup: Some(Box::new(cleanup)) }
Self {
message,
cleanup: Some(Box::new(cleanup)),
}
}
}

135
core/kordophoned/src/xpc/rpc.rs
View File

@@ -105,7 +105,12 @@ pub async fn dispatch(
.and_then(|m| dict_get_str(m, "conversation_id"))
{
Some(id) => id,
None => return DispatchResult::new(make_error_reply("InvalidRequest", "Missing conversation_id")),
None => {
return DispatchResult::new(make_error_reply(
"InvalidRequest",
"Missing conversation_id",
))
}
};
match agent
.send_event(|r| Event::SyncConversation(conversation_id, r))
@@ -122,7 +127,12 @@ pub async fn dispatch(
.and_then(|m| dict_get_str(m, "conversation_id"))
{
Some(id) => id,
None => return DispatchResult::new(make_error_reply("InvalidRequest", "Missing conversation_id")),
None => {
return DispatchResult::new(make_error_reply(
"InvalidRequest",
"Missing conversation_id",
))
}
};
match agent
.send_event(|r| Event::MarkConversationAsRead(conversation_id, r))
@@ -137,11 +147,21 @@ pub async fn dispatch(
"GetMessages" => {
let args = match get_dictionary_field(root, "arguments") {
Some(a) => a,
None => return DispatchResult::new(make_error_reply("InvalidRequest", "Missing arguments")),
None => {
return DispatchResult::new(make_error_reply(
"InvalidRequest",
"Missing arguments",
))
}
};
let conversation_id = match dict_get_str(args, "conversation_id") {
Some(id) => id,
None => return DispatchResult::new(make_error_reply("InvalidRequest", "Missing conversation_id")),
None => {
return DispatchResult::new(make_error_reply(
"InvalidRequest",
"Missing conversation_id",
))
}
};
let last_message_id = dict_get_str(args, "last_message_id");
match agent
@@ -158,11 +178,8 @@ pub async fn dispatch(
dict_put_str(&mut m, "sender", &msg.sender.display_name());
// Include attachment GUIDs for the client to resolve/download
let attachment_guids: Vec<String> = msg
.attachments
.iter()
.map(|a| a.guid.clone())
.collect();
let attachment_guids: Vec<String> =
msg.attachments.iter().map(|a| a.guid.clone()).collect();
m.insert(cstr("attachment_guids"), array_from_strs(attachment_guids));
// Full attachments array with metadata (mirrors DBus fields)
@@ -193,12 +210,23 @@ pub async fn dispatch(
if let Some(attribution_info) = &metadata.attribution_info {
let mut attribution_map: XpcMap = HashMap::new();
if let Some(width) = attribution_info.width {
dict_put_i64_as_str(&mut attribution_map, "width", width as i64);
dict_put_i64_as_str(
&mut attribution_map,
"width",
width as i64,
);
}
if let Some(height) = attribution_info.height {
dict_put_i64_as_str(&mut attribution_map, "height", height as i64);
dict_put_i64_as_str(
&mut attribution_map,
"height",
height as i64,
);
}
metadata_map.insert(cstr("attribution_info"), Message::Dictionary(attribution_map));
metadata_map.insert(
cstr("attribution_info"),
Message::Dictionary(attribution_map),
);
}
if !metadata_map.is_empty() {
a.insert(cstr("metadata"), Message::Dictionary(metadata_map));
@@ -230,11 +258,21 @@ pub async fn dispatch(
"SendMessage" => {
let args = match get_dictionary_field(root, "arguments") {
Some(a) => a,
None => return DispatchResult::new(make_error_reply("InvalidRequest", "Missing arguments")),
None => {
return DispatchResult::new(make_error_reply(
"InvalidRequest",
"Missing arguments",
))
}
};
let conversation_id = match dict_get_str(args, "conversation_id") {
Some(v) => v,
None => return DispatchResult::new(make_error_reply("InvalidRequest", "Missing conversation_id")),
None => {
return DispatchResult::new(make_error_reply(
"InvalidRequest",
"Missing conversation_id",
))
}
};
let text = dict_get_str(args, "text").unwrap_or_default();
let attachment_guids: Vec<String> = match args.get(&cstr("attachment_guids")) {
@@ -265,11 +303,21 @@ pub async fn dispatch(
"GetAttachmentInfo" => {
let args = match get_dictionary_field(root, "arguments") {
Some(a) => a,
None => return DispatchResult::new(make_error_reply("InvalidRequest", "Missing arguments")),
None => {
return DispatchResult::new(make_error_reply(
"InvalidRequest",
"Missing arguments",
))
}
};
let attachment_id = match dict_get_str(args, "attachment_id") {
Some(v) => v,
None => return DispatchResult::new(make_error_reply("InvalidRequest", "Missing attachment_id")),
None => {
return DispatchResult::new(make_error_reply(
"InvalidRequest",
"Missing attachment_id",
))
}
};
match agent
.send_event(|r| Event::GetAttachment(attachment_id, r))
@@ -308,11 +356,21 @@ pub async fn dispatch(
"OpenAttachmentFd" => {
let args = match get_dictionary_field(root, "arguments") {
Some(a) => a,
None => return DispatchResult::new(make_error_reply("InvalidRequest", "Missing arguments")),
None => {
return DispatchResult::new(make_error_reply(
"InvalidRequest",
"Missing arguments",
))
}
};
let attachment_id = match dict_get_str(args, "attachment_id") {
Some(v) => v,
None => return DispatchResult::new(make_error_reply("InvalidRequest", "Missing attachment_id")),
None => {
return DispatchResult::new(make_error_reply(
"InvalidRequest",
"Missing attachment_id",
))
}
};
let preview = dict_get_str(args, "preview")
.map(|s| s == "true")
@@ -335,9 +393,14 @@ pub async fn dispatch(
dict_put_str(&mut reply, "type", "OpenAttachmentFdResponse");
reply.insert(cstr("fd"), Message::Fd(fd));
DispatchResult { message: Message::Dictionary(reply), cleanup: Some(Box::new(file)) }
DispatchResult {
message: Message::Dictionary(reply),
cleanup: Some(Box::new(file)),
}
}
Err(e) => {
DispatchResult::new(make_error_reply("OpenFailed", &format!("{}", e)))
}
Err(e) => DispatchResult::new(make_error_reply("OpenFailed", &format!("{}", e))),
}
}
Err(e) => DispatchResult::new(make_error_reply("DaemonError", &format!("{}", e))),
@@ -348,11 +411,21 @@ pub async fn dispatch(
"DownloadAttachment" => {
let args = match get_dictionary_field(root, "arguments") {
Some(a) => a,
None => return DispatchResult::new(make_error_reply("InvalidRequest", "Missing arguments")),
None => {
return DispatchResult::new(make_error_reply(
"InvalidRequest",
"Missing arguments",
))
}
};
let attachment_id = match dict_get_str(args, "attachment_id") {
Some(v) => v,
None => return DispatchResult::new(make_error_reply("InvalidRequest", "Missing attachment_id")),
None => {
return DispatchResult::new(make_error_reply(
"InvalidRequest",
"Missing attachment_id",
))
}
};
let preview = dict_get_str(args, "preview")
.map(|s| s == "true")
@@ -371,11 +444,18 @@ pub async fn dispatch(
use std::path::PathBuf;
let args = match get_dictionary_field(root, "arguments") {
Some(a) => a,
None => return DispatchResult::new(make_error_reply("InvalidRequest", "Missing arguments")),
None => {
return DispatchResult::new(make_error_reply(
"InvalidRequest",
"Missing arguments",
))
}
};
let path = match dict_get_str(args, "path") {
Some(v) => v,
None => return DispatchResult::new(make_error_reply("InvalidRequest", "Missing path")),
None => {
return DispatchResult::new(make_error_reply("InvalidRequest", "Missing path"))
}
};
match agent
.send_event(|r| Event::UploadAttachment(PathBuf::from(path), r))
@@ -413,7 +493,12 @@ pub async fn dispatch(
"UpdateSettings" => {
let args = match get_dictionary_field(root, "arguments") {
Some(a) => a,
None => return DispatchResult::new(make_error_reply("InvalidRequest", "Missing arguments")),
None => {
return DispatchResult::new(make_error_reply(
"InvalidRequest",
"Missing arguments",
))
}
};
let server_url = dict_get_str(args, "server_url");
let username = dict_get_str(args, "username");

8
core/kpcli/src/client/mod.rs
View File

@@ -146,8 +146,7 @@ impl ClientCli {
loop {
match stream.next().await.unwrap() {
Ok(update) => {
match update {
Ok(update) => match update {
SocketUpdate::Update(updates) => {
for update in updates {
println!("Got update: {:?}", update);
@@ -156,7 +155,6 @@ impl ClientCli {
SocketUpdate::Pong => {
println!("Pong");
}
}
},
Err(e) => {
@@ -175,8 +173,8 @@ impl ClientCli {
.text(message)
.build();
let message = self.api.send_message(&outgoing_message).await?;
println!("Message sent: {}", message.guid);
let response = self.api.send_message(&outgoing_message).await?;
println!("Message sent: {}", response.message.guid);
Ok(())
}

52
core/utilities/src/bin/snoozer.rs
View File

@@ -1,13 +1,13 @@
use std::env;
use std::process;
use kordophone::{
api::{HTTPAPIClient, InMemoryAuthenticationStore, EventSocket},
model::{ConversationID, event::EventData},
APIInterface,
};
use kordophone::api::http_client::Credentials;
use kordophone::api::AuthenticationStore;
use kordophone::api::http_client::Credentials;
use kordophone::{
APIInterface,
api::{EventSocket, HTTPAPIClient, InMemoryAuthenticationStore},
model::{ConversationID, event::EventData},
};
use futures_util::StreamExt;
use hyper::Uri;
@@ -18,7 +18,10 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
let args: Vec<String> = env::args().collect();
if args.len() < 2 {
eprintln!("Usage: {} <conversation_id1> [conversation_id2] [conversation_id3] ...", args[0]);
eprintln!(
"Usage: {} <conversation_id1> [conversation_id2] [conversation_id3] ...",
args[0]
);
eprintln!("Environment variables required:");
eprintln!(" KORDOPHONE_API_URL - Server URL");
eprintln!(" KORDOPHONE_USERNAME - Username for authentication");
@@ -40,12 +43,14 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
let credentials = Credentials { username, password };
// Collect all conversation IDs from command line arguments
let target_conversation_ids: Vec<ConversationID> = args[1..].iter()
.map(|id| id.clone())
.collect();
let target_conversation_ids: Vec<ConversationID> =
args[1..].iter().map(|id| id.clone()).collect();
println!("Monitoring {} conversation(s) for updates: {:?}",
target_conversation_ids.len(), target_conversation_ids);
println!(
"Monitoring {} conversation(s) for updates: {:?}",
target_conversation_ids.len(),
target_conversation_ids
);
let auth_store = InMemoryAuthenticationStore::new(Some(credentials.clone()));
let mut client = HTTPAPIClient::new(server_url, auth_store);
@@ -62,26 +67,33 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
match event_result {
Ok(socket_event) => {
match socket_event {
kordophone::api::event_socket::SocketEvent::Update(event) => {
match event.data {
kordophone::api::event_socket::SocketEvent::Update(event) => match event.data {
EventData::MessageReceived(conversation, _message) => {
if target_conversation_ids.contains(&conversation.guid) {
println!("Message update detected for conversation {}, marking as read...", conversation.guid);
println!(
"Message update detected for conversation {}, marking as read...",
conversation.guid
);
match client.mark_conversation_as_read(&conversation.guid).await {
Ok(_) => println!("Successfully marked conversation {} as read", conversation.guid),
Err(e) => eprintln!("Failed to mark conversation {} as read: {:?}", conversation.guid, e),
Ok(_) => println!(
"Successfully marked conversation {} as read",
conversation.guid
),
Err(e) => eprintln!(
"Failed to mark conversation {} as read: {:?}",
conversation.guid, e
),
}
}
}
},
_ => {}
}
},
kordophone::api::event_socket::SocketEvent::Pong => {
// Ignore pong messages
}
}
},
}
Err(e) => {
eprintln!("Error receiving event: {:?}", e);
break;

159
server/kordophone/Bridge/Operations/MBIMSendMessageOperation.m
View File

@@ -10,6 +10,7 @@
#import "IMCore_ClassDump.h"
#import "IMMessageItem+Encoded.h"
#import "MBIMErrorResponse.h"
@implementation MBIMSendMessageOperation
@@ -20,16 +21,64 @@
return @"sendMessage";
}
- (IMMessage *)_sendMessage:(NSString *)messageBody toChatWithGUID:(NSString *)chatGUID attachmentGUIDs:(NSArray<NSString *> *)guids
- (nullable IMChat *)_chatForHandleIDs:(NSArray<NSString *> *)handleIDs registry:(IMChatRegistry *)registry
{
__block IMMessage *result = nil;
dispatch_sync([[self class] sharedIMAccessQueue], ^{
IMChat *chat = [[IMChatRegistry sharedInstance] existingChatWithGUID:chatGUID];
// TODO: chat might not be an iMessage chat!
IMAccount *iMessageAccount = [[IMAccountController sharedInstance] bestAccountForService:[IMServiceImpl iMessageService]];
IMHandle *senderHandle = [iMessageAccount loginIMHandle];
if (!iMessageAccount) {
MBIMLogError(@"Unable to find an iMessage account for message send.");
return nil;
}
NSMutableArray<IMHandle *> *handles = [NSMutableArray arrayWithCapacity:[handleIDs count]];
for (NSString *handleID in handleIDs) {
IMHandle *handle = [iMessageAccount imHandleWithID:handleID];
if (!handle) {
MBIMLogError(@"Couldn't resolve IMHandle for id %@", handleID);
return nil;
}
[handles addObject:handle];
}
if ([handles count] == 1) {
IMHandle *handle = [handles firstObject];
IMChat *chat = [registry existingChatWithHandle:handle allowAlternativeService:NO];
if (!chat) {
chat = [registry chatWithHandle:handle];
}
return chat;
}
IMChat *chat = [registry existingChatWithHandles:handles
allowAlternativeService:NO
groupID:nil
displayName:nil
joinedChatsOnly:YES];
if (!chat) {
chat = [registry chatWithHandles:handles displayName:nil joinedChatsOnly:YES];
}
return chat;
}
- (nullable NSDictionary *)_sendMessage:(NSString *)messageBody toChat:(IMChat *)chat attachmentGUIDs:(NSArray<NSString *> *)guids includeConversationGUID:(BOOL)includeConversationGUID
{
if (!chat) {
return nil;
}
IMAccount *sendingAccount = [chat account];
if (!sendingAccount) {
sendingAccount = [[IMAccountController sharedInstance] bestAccountForService:[IMServiceImpl iMessageService]];
}
IMHandle *senderHandle = [sendingAccount loginIMHandle];
if (!senderHandle) {
MBIMLogError(@"Unable to determine sender handle for message send.");
return nil;
}
NSAttributedString *replyAttrString = [[NSAttributedString alloc] initWithString:messageBody];
NSAttributedString *attrStringWithFileTransfers = IMCreateSuperFormatStringWithAppendedFileTransfers(replyAttrString, guids);
@@ -40,19 +89,24 @@
flags:(kIMMessageFinished | kIMMessageIsFromMe)];
for (NSString *guid in [reply fileTransferGUIDs]) {
[[IMFileTransferCenter sharedInstance] assignTransfer:guid toHandle:chat.recipient];
[[IMFileTransferCenter sharedInstance] assignTransfer:guid toMessage:reply account:sendingAccount];
}
if (!chat) {
MBIMLogInfo(@"Chat does not exist: %@", chatGUID);
} else {
result = reply;
NSMutableDictionary *result = [[reply mbim_dictionaryRepresentation] mutableCopy];
if (includeConversationGUID) {
NSString *conversationGUID = [chat guid];
if (!conversationGUID) {
conversationGUID = [[[IMChatRegistry sharedInstance] allGUIDsForChat:chat] firstObject];
}
if (conversationGUID) {
result[@"conversationGUID"] = conversationGUID;
}
}
dispatch_async(dispatch_get_main_queue(), ^{
[chat sendMessage:reply];
});
}
});
return result;
}
@@ -79,7 +133,7 @@
- (void)main
{
NSObject<HTTPResponse> *response = [[HTTPErrorResponse alloc] initWithErrorCode:500];
__block NSObject<HTTPResponse> *response = [[HTTPErrorResponse alloc] initWithErrorCode:500];
NSError *error = nil;
NSDictionary *args = [NSJSONSerialization JSONObjectWithData:self.requestBodyData options:0 error:&error];
@@ -90,31 +144,70 @@
NSString *guid = [args objectForKey:@"guid"];
NSString *messageBody = [args objectForKey:@"body"];
if (!guid || !messageBody) {
NSArray *rawHandleIDs = [args objectForKey:@"handleIDs"];
BOOL hasGUID = [guid isKindOfClass:[NSString class]] && [guid length] > 0;
BOOL hasHandleIDs = [rawHandleIDs isKindOfClass:[NSArray class]] && [rawHandleIDs count] > 0;
if (![messageBody isKindOfClass:[NSString class]] || (!hasGUID && !hasHandleIDs) || (hasGUID && hasHandleIDs)) {
response = [[MBIMErrorResponse alloc] initWithErrorCode:500 message:@"sendMessage requires body and exactly one of guid or handleIDs."];
self.serverCompletionBlock(response);
return;
}
// tapbacks
#if 0
IMMessage *acknowledgment = [IMMessage instantMessageWithAssociatedMessageContent: /* [NSString stringWithFormat:@"%@ \"%%@\"", tapbackAction] */
flags:0
associatedMessageGUID:guid
associatedMessageType:IMAssociatedMessageTypeAcknowledgmentHeart
associatedMessageRange:[imMessage messagePartRange]
messageSummaryInfo:[self adjustMessageSummaryInfoForSending:message]
threadIdentifier:[imMessage threadIdentifier]];
#endif
NSArray *transferGUIDs = [args objectForKey:@"fileTransferGUIDs"];
if (!transferGUIDs) {
transferGUIDs = @[];
NSMutableArray<NSString *> *handleIDs = [NSMutableArray array];
if (hasHandleIDs) {
for (id handleID in rawHandleIDs) {
if ([handleID isKindOfClass:[NSString class]] && [handleID length] > 0) {
[handleIDs addObject:handleID];
}
}
IMMessage *result = [self _sendMessage:messageBody toChatWithGUID:guid attachmentGUIDs:transferGUIDs];
handleIDs = [[[NSOrderedSet orderedSetWithArray:handleIDs] array] mutableCopy];
if ([handleIDs count] == 0) {
response = [[MBIMErrorResponse alloc] initWithErrorCode:500 message:@"No valid handle IDs provided."];
self.serverCompletionBlock(response);
return;
}
}
NSArray *rawTransferGUIDs = [args objectForKey:@"fileTransferGUIDs"];
NSMutableArray<NSString *> *transferGUIDs = [NSMutableArray array];
if ([rawTransferGUIDs isKindOfClass:[NSArray class]]) {
for (id transferGUID in rawTransferGUIDs) {
if ([transferGUID isKindOfClass:[NSString class]] && [transferGUID length] > 0) {
[transferGUIDs addObject:transferGUID];
}
}
}
dispatch_sync([[self class] sharedIMAccessQueue], ^{
IMChatRegistry *registry = [IMChatRegistry sharedInstance];
IMChat *chat = nil;
BOOL includeConversationGUID = NO;
if (hasGUID) {
chat = [registry existingChatWithGUID:guid];
if (!chat) {
response = [[MBIMErrorResponse alloc] initWithErrorCode:500 message:@"Chat does not exist for the provided guid."];
return;
}
} else {
chat = [self _chatForHandleIDs:handleIDs registry:registry];
includeConversationGUID = YES;
if (!chat) {
response = [[MBIMErrorResponse alloc] initWithErrorCode:500 message:@"Unable to create or locate a chat for the provided handles."];
return;
}
}
NSDictionary *result = [self _sendMessage:messageBody
toChat:chat
attachmentGUIDs:transferGUIDs
includeConversationGUID:includeConversationGUID];
if (result) {
response = [MBIMJSONDataResponse responseWithJSONObject:[result mbim_dictionaryRepresentation]];
response = [MBIMJSONDataResponse responseWithJSONObject:result];
}
});
self.serverCompletionBlock(response);
}