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

core: implement get_attachment_fd event for dbus, message limit for get_messages

Browse Source
This commit is contained in:
James Magahern
2026-02-21 23:26:00 -08:00
parent f38702bc95
commit f0ec6b8cb4
5 changed files with 153 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
core/kordophoned/include/net.buzzert.kordophonecd.Server.xml
View File

@@ -128,6 +128,20 @@
"/>
</method>
<method name="OpenAttachmentFd">
<arg type="s" name="attachment_id" direction="in"/>
<arg type="b" name="preview" direction="in"/>
<arg type="h" name="fd" direction="out"/>
<annotation name="org.freedesktop.DBus.DocString"
value="Opens a read-only file descriptor for an attachment path.
Arguments:
attachment_id: the attachment GUID
preview: whether to open the preview path (true) or full path (false)
Returns:
fd: a Unix file descriptor to read attachment bytes"/>
</method>
<method name="DownloadAttachment">
<arg type="s" name="attachment_id" direction="in"/>
<arg type="b" name="preview" direction="in"/>

61
core/kordophoned/src/daemon/attachment_store.rs
View File

@@ -115,39 +115,57 @@ impl AttachmentStore {
base_path: base_path,
metadata: None,
mime_type: None,
cached_full_path: None,
cached_preview_path: None,
};
// Best-effort: if a file already exists, try to infer MIME type from extension
let kind = "full";
// Best-effort: if files already exist, cache their exact paths and infer MIME type.
let stem = attachment
.base_path
.file_name()
.map(|s| s.to_string_lossy().to_string())
.unwrap_or_default();
let legacy = attachment.base_path.with_extension(kind);
let existing_path = if legacy.exists() {
Some(legacy)
} else {
let prefix = format!("{}.{}.", stem, kind);
let legacy_full = attachment.base_path.with_extension("full");
if legacy_full.exists() {
attachment.cached_full_path = Some(legacy_full);
}
let legacy_preview = attachment.base_path.with_extension("preview");
if legacy_preview.exists() {
attachment.cached_preview_path = Some(legacy_preview);
}
if attachment.cached_full_path.is_none() || attachment.cached_preview_path.is_none() {
let full_prefix = format!("{}.full.", stem);
let preview_prefix = format!("{}.preview.", stem);
let parent = attachment
.base_path
.parent()
.unwrap_or_else(|| std::path::Path::new("."));
let mut found: Option<PathBuf> = None;
if let Ok(entries) = std::fs::read_dir(parent) {
for entry in entries.flatten() {
let name = entry.file_name().to_string_lossy().to_string();
if name.starts_with(&prefix) && !name.ends_with(".download") {
found = Some(entry.path());
break;
}
}
}
found
};
if let Some(existing) = existing_path {
if let Some(m) = mime_guess::from_path(&existing).first_raw() {
if !name.ends_with(".download") {
if attachment.cached_full_path.is_none() && name.starts_with(&full_prefix) {
attachment.cached_full_path = Some(entry.path());
continue;
}
if attachment.cached_preview_path.is_none()
&& name.starts_with(&preview_prefix)
{
attachment.cached_preview_path = Some(entry.path());
}
}
}
}
}
if let Some(existing_full) = &attachment.cached_full_path {
if let Some(m) = mime_guess::from_path(existing_full).first_raw() {
attachment.mime_type = Some(m.to_string());
}
}
@@ -342,6 +360,9 @@ impl AttachmentStore {
match kind {
AttachmentStoreError::AttachmentAlreadyDownloaded => {
log::debug!(target: target::ATTACHMENTS, "Attachment already downloaded: {}", &_guid);
let _ = daemon_event_sink
.send(DaemonEvent::AttachmentDownloaded(_guid.clone()))
.await;
}
AttachmentStoreError::DownloadAlreadyInProgress => {
// Already logged a warning where detected
@@ -360,6 +381,10 @@ impl AttachmentStore {
log::debug!(target: target::ATTACHMENTS, "Queued download for attachment: {}", &guid);
} else {
log::debug!(target: target::ATTACHMENTS, "Attachment already downloaded: {}", guid);
let _ = self
.daemon_event_sink
.send(DaemonEvent::AttachmentDownloaded(guid))
.await;
}
}

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

@@ -15,6 +15,7 @@ use std::collections::HashMap;
use std::error::Error;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Instant;
use thiserror::Error;
use tokio::sync::mpsc::{Receiver, Sender};
@@ -72,6 +73,8 @@ pub mod target {
pub static DAEMON: &str = "daemon";
}
const GET_MESSAGES_INITIAL_WINDOW: usize = 300;
pub struct Daemon {
pub event_sender: Sender<Event>,
event_receiver: Receiver<Event>,
@@ -314,7 +317,9 @@ impl Daemon {
Event::GetMessages(conversation_id, last_message_id, reply) => {
let messages = self.get_messages(conversation_id, last_message_id).await;
let _ = reply.send(messages);
if reply.send(messages).is_err() {
log::warn!(target: target::EVENT, "GetMessages reply receiver dropped before send");
}
}
Event::DeleteAllConversations(reply) => {
@@ -448,8 +453,9 @@ impl Daemon {
async fn get_messages(
&mut self,
conversation_id: String,
_last_message_id: Option<MessageID>,
last_message_id: Option<MessageID>,
) -> Vec<Message> {
let started = Instant::now();
// Get outgoing messages for this conversation.
let empty_vec: Vec<OutgoingMessage> = vec![];
let outgoing_messages: &Vec<OutgoingMessage> = self
@@ -488,6 +494,27 @@ impl Daemon {
result.push(om.into());
}
if let Some(last_id) = last_message_id {
if let Some(last_index) = result.iter().position(|message| message.id == last_id) {
result = result.split_off(last_index + 1);
}
} else if result.len() > GET_MESSAGES_INITIAL_WINDOW {
let dropped = result.len() - GET_MESSAGES_INITIAL_WINDOW;
result = result.split_off(dropped);
log::debug!(
target: target::EVENT,
"GetMessages initial window applied: dropped {} older messages",
dropped
);
}
log::debug!(
target: target::EVENT,
"GetMessages completed in {}ms: {} messages",
started.elapsed().as_millis(),
result.len()
);
result
}

62
core/kordophoned/src/daemon/models/attachment.rs
View File

@@ -1,4 +1,4 @@
use std::path::{Path, PathBuf};
use std::path::PathBuf;
#[derive(Debug, Clone)]
pub struct AttachmentMetadata {
@@ -17,6 +17,8 @@ pub struct Attachment {
pub base_path: PathBuf,
pub metadata: Option<AttachmentMetadata>,
pub mime_type: Option<String>,
pub cached_full_path: Option<PathBuf>,
pub cached_preview_path: Option<PathBuf>,
}
impl Attachment {
@@ -25,8 +27,7 @@ impl Attachment {
// Prefer common, user-friendly extensions over obscure ones
match normalized {
"image/jpeg" | "image/pjpeg" => Some("jpg"),
_ => mime_guess::get_mime_extensions_str(normalized)
.and_then(|list| {
_ => mime_guess::get_mime_extensions_str(normalized).and_then(|list| {
// If jpg is one of the candidates, prefer it
if list.iter().any(|e| *e == "jpg") {
Some("jpg")
@@ -45,17 +46,21 @@ impl Attachment {
}
pub fn get_path_for_preview_scratch(&self, preview: bool, scratch: bool) -> PathBuf {
if !scratch {
let cached = if preview {
self.cached_preview_path.as_ref()
} else {
self.cached_full_path.as_ref()
};
if let Some(path) = cached {
return path.clone();
}
}
// Determine whether this is a preview or full attachment.
let kind = if preview { "preview" } else { "full" };
// If not a scratch path, and a file already exists on disk with a concrete
// file extension (e.g., guid.full.jpg), return that existing path.
if !scratch {
if let Some(existing) = self.find_existing_path(preview) {
return existing;
}
}
// Fall back to constructing a path using known info. If we know the MIME type,
// prefer an extension guessed from it; otherwise keep legacy naming.
let ext_from_mime = self
@@ -77,44 +82,15 @@ impl Attachment {
}
pub fn is_downloaded(&self, preview: bool) -> bool {
std::fs::exists(&self.get_path_for_preview(preview)).expect(
let path = self.get_path_for_preview(preview);
std::fs::exists(&path).expect(
format!(
"Wasn't able to check for the existence of an attachment file path at {}",
&self.get_path_for_preview(preview).display()
path.display()
)
.as_str(),
)
}
fn find_existing_path(&self, preview: bool) -> Option<PathBuf> {
let kind = if preview { "preview" } else { "full" };
// First, check legacy path without a concrete file extension.
let legacy = self.base_path.with_extension(kind);
if legacy.exists() {
return Some(legacy);
}
// Next, search for a filename like: <guid>.<kind>.<ext>
let file_stem = self
.base_path
.file_name()
.map(|s| s.to_string_lossy().to_string())?;
let prefix = format!("{}.{}.", file_stem, kind);
let parent = self.base_path.parent().unwrap_or_else(|| Path::new("."));
if let Ok(dir) = std::fs::read_dir(parent) {
for entry in dir.flatten() {
let file_name = entry.file_name();
let name = file_name.to_string_lossy();
if name.starts_with(&prefix) && !name.ends_with(".download") {
return Some(entry.path());
}
}
}
None
}
}
impl From<kordophone::model::message::AttachmentMetadata> for AttachmentMetadata {

38
core/kordophoned/src/dbus/agent.rs
View File

@@ -1,6 +1,9 @@
use dbus::arg;
use dbus_tree::MethodErr;
use std::fs::OpenOptions;
use std::os::fd::{FromRawFd, IntoRawFd};
use std::sync::Arc;
use std::time::Instant;
use std::{future::Future, thread};
use tokio::sync::{mpsc, oneshot, Mutex};
@@ -277,6 +280,7 @@ impl DbusRepository for DBusAgent {
conversation_id: String,
last_message_id: String,
) -> Result<Vec<arg::PropMap>, MethodErr> {
let started = Instant::now();
let last_message_id_opt = if last_message_id.is_empty() {
None
} else {
@@ -286,6 +290,9 @@ impl DbusRepository for DBusAgent {
let messages =
self.send_event_sync(|r| Event::GetMessages(conversation_id, last_message_id_opt, r))?;
let mut attachment_count: usize = 0;
let mut text_bytes: usize = 0;
let mapped: Vec<arg::PropMap> = messages
.into_iter()
.map(|msg| {
@@ -294,6 +301,7 @@ impl DbusRepository for DBusAgent {
// Remove the attachment placeholder here.
let text = msg.text.replace("\u{FFFC}", "");
text_bytes += text.len();
map.insert("text".into(), arg::Variant(Box::new(text)));
map.insert(
@@ -305,10 +313,12 @@ impl DbusRepository for DBusAgent {
arg::Variant(Box::new(msg.sender.display_name())),
);
if !msg.attachments.is_empty() {
let attachments: Vec<arg::PropMap> = msg
.attachments
.into_iter()
.map(|attachment| {
attachment_count += 1;
let mut attachment_map = arg::PropMap::new();
attachment_map.insert(
"guid".into(),
@@ -351,17 +361,26 @@ impl DbusRepository for DBusAgent {
arg::Variant(Box::new(metadata_map)),
);
}
attachment_map
})
.collect();
map.insert("attachments".into(), arg::Variant(Box::new(attachments)));
}
map
})
.collect();
log::debug!(
target: "dbus",
"GetMessages mapped in {}ms: {} messages, {} attachments, {} text-bytes",
started.elapsed().as_millis(),
mapped.len(),
attachment_count,
text_bytes
);
Ok(mapped)
}
@@ -406,6 +425,23 @@ impl DbusRepository for DBusAgent {
self.send_event_sync(|r| Event::DownloadAttachment(attachment_id, preview, r))
}
fn open_attachment_fd(
&mut self,
attachment_id: String,
preview: bool,
) -> Result<arg::OwnedFd, MethodErr> {
let attachment = self.send_event_sync(|r| Event::GetAttachment(attachment_id, r))?;
let path = attachment.get_path_for_preview(preview);
let file = OpenOptions::new()
.read(true)
.open(&path)
.map_err(|e| MethodErr::failed(&format!("Failed to open attachment: {}", e)))?;
let fd = file.into_raw_fd();
Ok(unsafe { arg::OwnedFd::from_raw_fd(fd) })
}
fn upload_attachment(&mut self, path: String) -> Result<String, MethodErr> {
use std::path::PathBuf;
let path = PathBuf::from(path);