feat: file transfer system (avatar/icon download with cacache) (#40)
* docs(architecture): add file transfer design, research, and implementation plan * feat(cache): add chanora_cache crate with cacache-backed blob cache - New chanora_cache crate: content-addressed blob store wrapping cacache - BlobCache API: async put/get/remove/clear/total_size/evict - Key validation: av_ prefix (32 hex chars), ic_ prefix (decimal digits) - Cacache provides crash safety, SSRI integrity, content dedup - Mtime-based eviction via cacache::list_sync + sort by timestamp - 7 unit tests all passing - Added to workspace members * feat(protocol): add file download support for avatars and icons - Add Request::DownloadFile variant with oneshot reply - Add ProtocolClient::download_avatar(client_uid) and download_icon(icon_id) - Track pending file downloads by FiletransferHandle - Handle StreamItem::FileDownload: read bytes from TCP stream - Handle StreamItem::FiletransferFailed: map to ProtocolError - Add ProtocolError::FileTransfer(String) variant - Add path helper tests for avatar/icon download paths - No tsclientlib types leak across the adapter boundary * feat(core): add blob cache wiring and avatar download orchestration - Add chanora_cache dependency to Cargo.toml - Add blob_cache field to ChanoraSession (Arc<Mutex<Option<BlobCache>>>) - Add init_cache() method: creates BlobCache, runs eviction - Add get_avatar() method: cache-first, download on miss, store in cache - Add clear_cache() and cache_size() methods for cache management - Add CoreError::Cache variant for BlobCacheError conversion - Add avatar_cache integration test * feat(bridge): add init_cache, download_avatar, and cache management functions - Add init_cache(dir) bridge function - Add download_avatar(avatar_hash, client_uid) bridge function - Add clear_file_cache() and file_cache_size() bridge functions - Map CoreError::Cache and ProtocolError::FileTransfer in BridgeError * feat(flutter): add cache initialization wiring and avatar download shims - Add wireCache() to app_bootstrap using getApplicationCacheDirectory() - Call wireCache() after wireStorage() in main bootstrap flow - Add Dart-side initCache and downloadAvatar wrapper shims in api.dart - Update Cargo.lock for new chanora_cache dependency * feat(core): FileTransferService with coalescing, throttling, negative cache - New file_transfer module with FileTransferService struct - Semaphore(2) throttles concurrent downloads - In-flight HashMap coalesces duplicate avatar requests - 5-min negative cache short-circuits ServerRejected misses - ChanoraSession delegates get_avatar through the service - connect/disconnect update shared protocol handle - clear_cache/cache_size delegate to service - 2 new unit tests (cached hit, negative cache) * feat(core,bridge): add get_icon with coalescing and negative cache - FileTransferService::get_icon() mirrors get_avatar pattern - ChanoraSession::get_icon() delegates through FileTransferService - Bridge download_icon() exposed for Flutter - Dart downloadIcon() shim added - Uses PREFIX_ICON (ic_<crc32u>) cache key format - 1 new unit test (cached icon hit) * fix(core,protocol): simplify store_protocol and add download size cap - store_protocol: always write to shared Arc<Mutex<Option<ProtocolClient>>>; the FileTransferService holds the same Arc so it sees updates automatically - read_download_bytes: reject downloads exceeding 10 MB to prevent malicious servers from causing OOM
This commit is contained in:
Edison Jwa
@@ -1465,6 +1465,54 @@ pub async fn init_storage(dir: String) -> Result<(), BridgeError> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Configure the bridge blob cache root.
|
||||
pub async fn init_cache(dir: String) -> Result<(), BridgeError> {
|
||||
runtime()
|
||||
.spawn(async move { session().init_cache(&dir).await })
|
||||
.await
|
||||
.map_err(|e| task_join_error("init_cache", e))??;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Resolve avatar bytes through the bridge.
|
||||
pub async fn download_avatar(
|
||||
avatar_hash: String,
|
||||
client_uid: String,
|
||||
) -> Result<Option<Vec<u8>>, BridgeError> {
|
||||
runtime()
|
||||
.spawn(async move { session().get_avatar(&avatar_hash, &client_uid).await })
|
||||
.await
|
||||
.map_err(|e| task_join_error("download_avatar", e))?
|
||||
.map_err(BridgeError::from)
|
||||
}
|
||||
|
||||
/// Resolve icon bytes through the bridge.
|
||||
pub async fn download_icon(icon_id: u64) -> Result<Option<Vec<u8>>, BridgeError> {
|
||||
runtime()
|
||||
.spawn(async move { session().get_icon(icon_id).await })
|
||||
.await
|
||||
.map_err(|e| task_join_error("download_icon", e))?
|
||||
.map_err(BridgeError::from)
|
||||
}
|
||||
|
||||
/// Purge cached protocol-owned assets.
|
||||
pub async fn clear_file_cache() -> Result<(), BridgeError> {
|
||||
runtime()
|
||||
.spawn(async move { session().clear_cache().await })
|
||||
.await
|
||||
.map_err(|e| task_join_error("clear_file_cache", e))??;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Report the configured file-cache size.
|
||||
pub async fn file_cache_size() -> Result<u64, BridgeError> {
|
||||
runtime()
|
||||
.spawn(async move { session().cache_size().await })
|
||||
.await
|
||||
.map_err(|e| task_join_error("file_cache_size", e))?
|
||||
.map_err(BridgeError::from)
|
||||
}
|
||||
|
||||
/// Bookmark DTO mirroring [`chanora_core::Bookmark`].
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct BridgeBookmark {
|
||||
|
||||
@@ -110,9 +110,13 @@ impl From<chanora_core::CoreError> for BridgeError {
|
||||
code,
|
||||
message,
|
||||
}) => BridgeError::ServerRejected { code, message },
|
||||
chanora_core::CoreError::Protocol(chanora_core::ProtocolError::FileTransfer(p)) => {
|
||||
BridgeError::Connection(format!("file transfer: {p}"))
|
||||
}
|
||||
chanora_core::CoreError::Protocol(p) => BridgeError::Connection(format!("{p}")),
|
||||
chanora_core::CoreError::Audio(a) => BridgeError::Connection(format!("audio: {a}")),
|
||||
chanora_core::CoreError::Storage(s) => BridgeError::Connection(format!("storage: {s}")),
|
||||
chanora_core::CoreError::Cache(c) => BridgeError::Connection(format!("cache: {c}")),
|
||||
other => BridgeError::Unmapped(format!("{other}")),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,20 @@
|
||||
[package]
|
||||
name = "chanora_cache"
|
||||
description = "Chanora disposable content-addressed blob cache for avatars and icons"
|
||||
version.workspace = true
|
||||
edition.workspace = true
|
||||
rust-version.workspace = true
|
||||
authors.workspace = true
|
||||
license.workspace = true
|
||||
repository.workspace = true
|
||||
publish.workspace = true
|
||||
|
||||
[dependencies]
|
||||
cacache = "13"
|
||||
thiserror.workspace = true
|
||||
tokio = { version = "1", features = ["fs", "rt"] }
|
||||
tracing.workspace = true
|
||||
|
||||
[dev-dependencies]
|
||||
tempfile = "3"
|
||||
tokio = { version = "1", features = ["rt", "macros", "time"] }
|
||||
@@ -0,0 +1,359 @@
|
||||
//! Disposable content-addressed blob cache for avatar and icon files.
|
||||
//!
|
||||
//! Wraps [`cacache`] for production-tested crash safety and integrity
|
||||
//! verification. The on-disk layout is managed by cacache (content-v2,
|
||||
//! index-v2). Chanora maps protocol keys (`av_<md5>`, `ic_<crc32>`) to
|
||||
//! cacache string keys.
|
||||
//!
|
||||
//! This crate is intentionally separate from `chanora_storage`:
|
||||
//! storage owns persistent identity/bookmark data; cache owns
|
||||
//! reconstructible, disposable blob data with different durability
|
||||
//! and backup semantics.
|
||||
|
||||
#![forbid(unsafe_code)]
|
||||
#![warn(missing_docs)]
|
||||
|
||||
use std::path::{Path, PathBuf};
|
||||
|
||||
/// Errors raised by the blob cache.
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub enum BlobCacheError {
|
||||
/// Filesystem I/O error.
|
||||
#[error("io: {0}")]
|
||||
Io(String),
|
||||
/// Key validation error.
|
||||
#[error("invalid key: {0}")]
|
||||
InvalidKey(String),
|
||||
}
|
||||
|
||||
/// Content-addressed blob cache backed by cacache.
|
||||
pub struct BlobCache {
|
||||
cache_dir: PathBuf,
|
||||
/// Maximum total cache size in bytes. 0 = no limit.
|
||||
max_bytes: u64,
|
||||
}
|
||||
|
||||
/// Avatar blob prefix.
|
||||
pub const PREFIX_AVATAR: &str = "av_";
|
||||
/// Icon blob prefix.
|
||||
pub const PREFIX_ICON: &str = "ic_";
|
||||
|
||||
impl BlobCache {
|
||||
/// Create or open a [`BlobCache`] rooted at `cache_dir/chanora/`.
|
||||
///
|
||||
/// Creates the cacache directory. `max_bytes` sets the eviction
|
||||
/// threshold; 0 means no automatic eviction.
|
||||
pub fn new(cache_dir: impl AsRef<Path>, max_bytes: u64) -> Result<Self, BlobCacheError> {
|
||||
let cache_dir = cache_dir.as_ref().join("chanora").join("blobs");
|
||||
// cacache creates the directory on first write, but we create
|
||||
// it eagerly so total_size() works before any writes.
|
||||
std::fs::create_dir_all(&cache_dir)
|
||||
.map_err(|e| BlobCacheError::Io(format!("mkdir cache: {e}")))?;
|
||||
Ok(Self {
|
||||
cache_dir,
|
||||
max_bytes,
|
||||
})
|
||||
}
|
||||
|
||||
/// Store a blob. `prefix` is [`PREFIX_AVATAR`] or [`PREFIX_ICON`].
|
||||
/// `key` is the content hash (MD5 hex for avatars, unsigned
|
||||
/// decimal CRC32 for icons).
|
||||
///
|
||||
/// Cacache handles dedup and integrity internally.
|
||||
pub async fn put(
|
||||
&self,
|
||||
prefix: &str,
|
||||
key: &str,
|
||||
data: &[u8],
|
||||
) -> Result<(), BlobCacheError> {
|
||||
validate_key(prefix, key)?;
|
||||
let cache_key = format!("{prefix}{key}");
|
||||
cacache::write(&self.cache_dir, &cache_key, data)
|
||||
.await
|
||||
.map_err(|e| BlobCacheError::Io(format!("cacache write: {e}")))?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Read a blob. Returns `None` if not cached.
|
||||
///
|
||||
/// Cacache verifies SSRI integrity on every read.
|
||||
pub async fn get(&self, prefix: &str, key: &str) -> Result<Option<Vec<u8>>, BlobCacheError> {
|
||||
validate_key(prefix, key)?;
|
||||
let cache_key = format!("{prefix}{key}");
|
||||
match cacache::read(&self.cache_dir, &cache_key).await {
|
||||
Ok(data) => Ok(Some(data)),
|
||||
Err(cacache::Error::EntryNotFound(_, _)) => Ok(None),
|
||||
Err(e) => {
|
||||
// Integrity failure or I/O error — remove corrupt entry.
|
||||
tracing::warn!(
|
||||
target: "chanora_cache",
|
||||
key = %cache_key,
|
||||
error = %e,
|
||||
"cache read failed; removing entry"
|
||||
);
|
||||
let _ = cacache::remove(&self.cache_dir, &cache_key).await;
|
||||
Ok(None)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Delete a specific blob.
|
||||
pub async fn remove(&self, prefix: &str, key: &str) -> Result<(), BlobCacheError> {
|
||||
validate_key(prefix, key)?;
|
||||
let cache_key = format!("{prefix}{key}");
|
||||
cacache::remove(&self.cache_dir, &cache_key)
|
||||
.await
|
||||
.map_err(|e| BlobCacheError::Io(format!("cacache remove: {e}")))?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Delete all blobs.
|
||||
pub async fn clear(&self) -> Result<(), BlobCacheError> {
|
||||
let path = self.cache_dir.clone();
|
||||
tokio::task::spawn_blocking(move || {
|
||||
if path.exists() {
|
||||
std::fs::remove_dir_all(&path)
|
||||
.map_err(|e| BlobCacheError::Io(format!("clear cache: {e}")))?;
|
||||
std::fs::create_dir_all(&path)
|
||||
.map_err(|e| BlobCacheError::Io(format!("recreate cache dir: {e}")))?;
|
||||
}
|
||||
Ok(())
|
||||
})
|
||||
.await
|
||||
.map_err(|e| BlobCacheError::Io(format!("clear task: {e}")))?
|
||||
}
|
||||
|
||||
/// Return total bytes used by all blobs.
|
||||
///
|
||||
/// Walks cacache entries and sums sizes.
|
||||
pub async fn total_size(&self) -> Result<u64, BlobCacheError> {
|
||||
let cache_dir = self.cache_dir.clone();
|
||||
tokio::task::spawn_blocking(move || {
|
||||
let mut total: u64 = 0;
|
||||
for entry in cacache::list_sync(&cache_dir) {
|
||||
match entry {
|
||||
Ok(meta) => total += meta.size as u64,
|
||||
Err(e) => {
|
||||
tracing::warn!(
|
||||
target: "chanora_cache",
|
||||
error = %e,
|
||||
"skipping bad entry during size scan"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(total)
|
||||
})
|
||||
.await
|
||||
.map_err(|e| BlobCacheError::Io(format!("total_size task: {e}")))?
|
||||
}
|
||||
|
||||
/// Evict oldest entries by timestamp until total size is under
|
||||
/// `max_bytes`. Call on startup or periodically. No-op if
|
||||
/// `max_bytes` is 0.
|
||||
pub async fn evict(&self) -> Result<(), BlobCacheError> {
|
||||
if self.max_bytes == 0 {
|
||||
return Ok(());
|
||||
}
|
||||
let cache_dir = self.cache_dir.clone();
|
||||
let max_bytes = self.max_bytes;
|
||||
tokio::task::spawn_blocking(move || {
|
||||
let mut entries: Vec<(String, usize, u128)> = Vec::new();
|
||||
for entry in cacache::list_sync(&cache_dir) {
|
||||
match entry {
|
||||
Ok(meta) => {
|
||||
entries.push((meta.key, meta.size, meta.time));
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!(
|
||||
target: "chanora_cache",
|
||||
error = %e,
|
||||
"skipping bad entry during eviction scan"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
let total: usize = entries.iter().map(|(_, s, _)| *s).sum();
|
||||
if total as u64 <= max_bytes {
|
||||
return Ok(());
|
||||
}
|
||||
entries.sort_by_key(|(_, _, t)| *t);
|
||||
let mut freed: usize = 0;
|
||||
let target = total - max_bytes as usize;
|
||||
for (key, size, _) in entries {
|
||||
if freed >= target {
|
||||
break;
|
||||
}
|
||||
let _ = cacache::remove_sync(&cache_dir, &key);
|
||||
freed += size;
|
||||
}
|
||||
tracing::info!(
|
||||
target: "chanora_cache",
|
||||
freed_bytes = freed,
|
||||
"evicted oldest blobs"
|
||||
);
|
||||
Ok(())
|
||||
})
|
||||
.await
|
||||
.map_err(|e| BlobCacheError::Io(format!("evict task: {e}")))?
|
||||
}
|
||||
}
|
||||
|
||||
/// Validate key format to prevent malformed entries.
|
||||
fn validate_key(prefix: &str, key: &str) -> Result<(), BlobCacheError> {
|
||||
if !matches!(prefix, PREFIX_AVATAR | PREFIX_ICON) {
|
||||
return Err(BlobCacheError::InvalidKey(format!("bad prefix: {prefix}")));
|
||||
}
|
||||
match prefix {
|
||||
PREFIX_AVATAR => {
|
||||
// MD5 hex = exactly 32 hex chars.
|
||||
if key.len() != 32 || !key.chars().all(|c| c.is_ascii_hexdigit()) {
|
||||
return Err(BlobCacheError::InvalidKey(format!(
|
||||
"avatar key must be 32 hex chars, got: {key}"
|
||||
)));
|
||||
}
|
||||
}
|
||||
PREFIX_ICON => {
|
||||
// Unsigned CRC32 = decimal digits.
|
||||
if key.is_empty() || !key.chars().all(|c| c.is_ascii_digit()) {
|
||||
return Err(BlobCacheError::InvalidKey(format!(
|
||||
"icon key must be decimal digits, got: {key}"
|
||||
)));
|
||||
}
|
||||
}
|
||||
_ => unreachable!(),
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
fn tempdir() -> tempfile::TempDir {
|
||||
tempfile::Builder::new()
|
||||
.prefix("chanora_cache_test_")
|
||||
.tempdir()
|
||||
.unwrap()
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn put_get_roundtrip() {
|
||||
let tmp = tempdir();
|
||||
let cache = BlobCache::new(&tmp, 0).unwrap();
|
||||
assert!(cache
|
||||
.get(PREFIX_AVATAR, "a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6")
|
||||
.await
|
||||
.unwrap()
|
||||
.is_none());
|
||||
cache
|
||||
.put(PREFIX_AVATAR, "a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6", b"avatar-bytes")
|
||||
.await
|
||||
.unwrap();
|
||||
let data = cache
|
||||
.get(PREFIX_AVATAR, "a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6")
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(data.as_deref(), Some(b"avatar-bytes".as_slice()));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn get_missing_returns_none() {
|
||||
let tmp = tempdir();
|
||||
let cache = BlobCache::new(&tmp, 0).unwrap();
|
||||
assert!(cache
|
||||
.get(PREFIX_AVATAR, "00000000000000000000000000000000")
|
||||
.await
|
||||
.unwrap()
|
||||
.is_none());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn clear_removes_all() {
|
||||
let tmp = tempdir();
|
||||
let cache = BlobCache::new(&tmp, 0).unwrap();
|
||||
cache
|
||||
.put(PREFIX_AVATAR, "a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6", b"data")
|
||||
.await
|
||||
.unwrap();
|
||||
cache.put(PREFIX_ICON, "12345", b"icon").await.unwrap();
|
||||
cache.clear().await.unwrap();
|
||||
assert_eq!(cache.total_size().await.unwrap(), 0);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn total_size_accounts_for_all_entries() {
|
||||
let tmp = tempdir();
|
||||
let cache = BlobCache::new(&tmp, 0).unwrap();
|
||||
cache
|
||||
.put(PREFIX_AVATAR, "a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6", b"12345")
|
||||
.await
|
||||
.unwrap();
|
||||
cache.put(PREFIX_ICON, "99999", b"12").await.unwrap();
|
||||
assert_eq!(cache.total_size().await.unwrap(), 5 + 2);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn invalid_key_rejected() {
|
||||
let tmp = tempdir();
|
||||
let cache = BlobCache::new(&tmp, 0).unwrap();
|
||||
// Too short for MD5.
|
||||
assert!(cache.put(PREFIX_AVATAR, "abc", b"data").await.is_err());
|
||||
// Non-hex in MD5.
|
||||
assert!(cache
|
||||
.put(PREFIX_AVATAR, "g".repeat(32).as_str(), b"data")
|
||||
.await
|
||||
.is_err());
|
||||
// Non-digit in icon key.
|
||||
assert!(cache.put(PREFIX_ICON, "12a45", b"data").await.is_err());
|
||||
// Bad prefix.
|
||||
assert!(cache.put("xx_", "abc", b"data").await.is_err());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn evict_deletes_oldest_until_under_cap() {
|
||||
let tmp = tempdir();
|
||||
// 10 byte cap.
|
||||
let cache = BlobCache::new(&tmp, 10).unwrap();
|
||||
cache
|
||||
.put(
|
||||
PREFIX_AVATAR,
|
||||
"a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6",
|
||||
b"12345678",
|
||||
)
|
||||
.await
|
||||
.unwrap(); // 8 bytes
|
||||
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
|
||||
cache
|
||||
.put(PREFIX_ICON, "11111", b"12345")
|
||||
.await
|
||||
.unwrap(); // 5 bytes → total 13, over cap
|
||||
cache.evict().await.unwrap();
|
||||
// Oldest (avatar) should be evicted.
|
||||
assert!(cache
|
||||
.get(PREFIX_AVATAR, "a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6")
|
||||
.await
|
||||
.unwrap()
|
||||
.is_none());
|
||||
assert!(cache.get(PREFIX_ICON, "11111").await.unwrap().is_some());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn remove_deletes_entry() {
|
||||
let tmp = tempdir();
|
||||
let cache = BlobCache::new(&tmp, 0).unwrap();
|
||||
cache
|
||||
.put(PREFIX_AVATAR, "a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6", b"data")
|
||||
.await
|
||||
.unwrap();
|
||||
cache
|
||||
.remove(PREFIX_AVATAR, "a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6")
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(cache
|
||||
.get(PREFIX_AVATAR, "a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6")
|
||||
.await
|
||||
.unwrap()
|
||||
.is_none());
|
||||
}
|
||||
}
|
||||
@@ -24,6 +24,7 @@ use base64::prelude::*;
|
||||
use chanora_resolver::ChanoraResolver;
|
||||
use futures::prelude::*;
|
||||
use std::collections::HashMap;
|
||||
use tokio::io::AsyncReadExt;
|
||||
use tokio::sync::{mpsc, oneshot};
|
||||
use tracing::{info, warn};
|
||||
|
||||
@@ -32,7 +33,8 @@ use tsclientlib::messages::s2c::{InClientDbInfoPart, InMessage};
|
||||
use tsclientlib::prelude::*;
|
||||
use tsclientlib::{
|
||||
ChannelId as TsChannelId, ClientId as TsClientId, Connection, ConnectionStats,
|
||||
DisconnectOptions, Identity, MessageHandle, OutCommandExt, StreamItem, Version,
|
||||
DisconnectOptions, FileDownloadResult, FiletransferHandle, Identity, MessageHandle,
|
||||
OutCommandExt, StreamItem, Version,
|
||||
};
|
||||
use tsproto_packets::packets::{Direction, Flags, InAudioBuf, OutCommand, OutPacket, PacketType};
|
||||
use tsproto_types::ClientType;
|
||||
@@ -60,6 +62,9 @@ type PendingMoves = HashMap<
|
||||
),
|
||||
>;
|
||||
|
||||
type PendingDownloads =
|
||||
HashMap<FiletransferHandle, oneshot::Sender<Result<Vec<u8>, ProtocolError>>>;
|
||||
|
||||
struct EventChannels {
|
||||
voice_in: mpsc::Sender<InboundVoice>,
|
||||
chat: mpsc::Sender<ChatMessage>,
|
||||
@@ -207,6 +212,10 @@ enum Request {
|
||||
client_id: u64,
|
||||
reply: oneshot::Sender<Result<ClientProfile, ProtocolError>>,
|
||||
},
|
||||
DownloadFile {
|
||||
path: String,
|
||||
reply: oneshot::Sender<Result<Vec<u8>, ProtocolError>>,
|
||||
},
|
||||
}
|
||||
|
||||
/// Why a [`ProtocolClient`] task ended. Distinguishes a user-driven
|
||||
@@ -366,6 +375,26 @@ impl ProtocolClient {
|
||||
.map_err(|_| ProtocolError::Lost("client_profile reply dropped".to_string()))?
|
||||
}
|
||||
|
||||
async fn download_file(&self, path: String) -> Result<Vec<u8>, ProtocolError> {
|
||||
let (tx, rx) = oneshot::channel();
|
||||
self.tx
|
||||
.send(Request::DownloadFile { path, reply: tx })
|
||||
.await
|
||||
.map_err(|_| ProtocolError::Lost("connection task is gone".to_string()))?;
|
||||
rx.await
|
||||
.map_err(|_| ProtocolError::Lost("download_file reply dropped".to_string()))?
|
||||
}
|
||||
|
||||
/// Download the current avatar bytes for a TeamSpeak client UID.
|
||||
pub async fn download_avatar(&self, client_uid: &str) -> Result<Vec<u8>, ProtocolError> {
|
||||
self.download_file(avatar_download_path(client_uid)).await
|
||||
}
|
||||
|
||||
/// Download the current channel/server icon bytes for an icon id.
|
||||
pub async fn download_icon(&self, icon_id: u64) -> Result<Vec<u8>, ProtocolError> {
|
||||
self.download_file(icon_download_path(icon_id)).await
|
||||
}
|
||||
|
||||
/// Disconnect cleanly. Blocks until the task exits.
|
||||
pub async fn disconnect(self) {
|
||||
let (tx, rx) = oneshot::channel();
|
||||
@@ -713,6 +742,7 @@ async fn connection_task(
|
||||
// deadline so a server that never replies doesn't leak the
|
||||
// reply channel — at most 3 s of pending state per move.
|
||||
let mut pending_moves: PendingMoves = HashMap::new();
|
||||
let mut pending_downloads: PendingDownloads = HashMap::new();
|
||||
let mut voice_activity: HashMap<u64, Instant> = HashMap::new();
|
||||
let mut poke_limiter = PokeLimiter::new();
|
||||
|
||||
@@ -737,6 +767,12 @@ async fn connection_task(
|
||||
StreamItem::Audio(buf) => {
|
||||
handle_audio_stream_item(&channels.voice_in, &mut voice_activity, buf).await;
|
||||
}
|
||||
StreamItem::FileDownload(handle, result) => {
|
||||
handle_download_stream_item(&mut pending_downloads, handle, result).await;
|
||||
}
|
||||
StreamItem::FiletransferFailed(handle, error) => {
|
||||
handle_download_failure(&mut pending_downloads, handle, error);
|
||||
}
|
||||
other => handle_non_audio_stream_item(
|
||||
&con,
|
||||
other,
|
||||
@@ -843,12 +879,25 @@ async fn connection_task(
|
||||
client_id,
|
||||
&channels,
|
||||
&mut pending_moves,
|
||||
&mut pending_downloads,
|
||||
&mut voice_activity,
|
||||
&mut poke_limiter,
|
||||
)
|
||||
.await;
|
||||
let _ = reply.send(r);
|
||||
}
|
||||
Ok(Request::DownloadFile { path, reply }) => {
|
||||
match con.download_file(TsChannelId(0), &path, None, None) {
|
||||
Ok(handle) => {
|
||||
pending_downloads.insert(handle, reply);
|
||||
}
|
||||
Err(e) => {
|
||||
let _ = reply.send(Err(ProtocolError::FileTransfer(format!(
|
||||
"start download {path}: {e}"
|
||||
))));
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(Request::Disconnect(reply)) => {
|
||||
let _ = con.disconnect(DisconnectOptions::new());
|
||||
bounded_drain_stream(con.events(), DISCONNECT_EVENT_DRAIN_TIMEOUT).await;
|
||||
@@ -1024,6 +1073,47 @@ fn handle_non_audio_stream_item(
|
||||
}
|
||||
}
|
||||
|
||||
async fn handle_download_stream_item(
|
||||
pending_downloads: &mut PendingDownloads,
|
||||
handle: FiletransferHandle,
|
||||
result: FileDownloadResult,
|
||||
) {
|
||||
if let Some(reply) = pending_downloads.remove(&handle) {
|
||||
let _ = reply.send(read_download_bytes(result).await);
|
||||
}
|
||||
}
|
||||
|
||||
fn handle_download_failure(
|
||||
pending_downloads: &mut PendingDownloads,
|
||||
handle: FiletransferHandle,
|
||||
error: tsclientlib::Error,
|
||||
) {
|
||||
if let Some(reply) = pending_downloads.remove(&handle) {
|
||||
let _ = reply.send(Err(ProtocolError::FileTransfer(error.to_string())));
|
||||
}
|
||||
}
|
||||
|
||||
const MAX_DOWNLOAD_SIZE: u64 = 10 * 1024 * 1024;
|
||||
|
||||
async fn read_download_bytes(result: FileDownloadResult) -> Result<Vec<u8>, ProtocolError> {
|
||||
if result.size > MAX_DOWNLOAD_SIZE {
|
||||
return Err(ProtocolError::FileTransfer(format!(
|
||||
"download too large: {} bytes (max {})",
|
||||
result.size, MAX_DOWNLOAD_SIZE
|
||||
)));
|
||||
}
|
||||
let size = usize::try_from(result.size).map_err(|_| {
|
||||
ProtocolError::FileTransfer(format!("download too large to buffer: {} bytes", result.size))
|
||||
})?;
|
||||
let mut stream = result.stream;
|
||||
let mut bytes = vec![0_u8; size];
|
||||
stream
|
||||
.read_exact(&mut bytes)
|
||||
.await
|
||||
.map_err(|e| ProtocolError::FileTransfer(e.to_string()))?;
|
||||
Ok(bytes)
|
||||
}
|
||||
|
||||
async fn resolve_server_socket(address: &str) -> Result<SocketAddr, ProtocolError> {
|
||||
let resolver = ChanoraResolver::new().map_err(|err| ProtocolError::DnsFailed {
|
||||
host: address.to_string(),
|
||||
@@ -1174,6 +1264,7 @@ async fn fetch_client_profile(
|
||||
client_id: u64,
|
||||
channels: &EventChannels,
|
||||
pending_moves: &mut PendingMoves,
|
||||
pending_downloads: &mut PendingDownloads,
|
||||
voice_activity: &mut HashMap<u64, Instant>,
|
||||
poke_limiter: &mut PokeLimiter,
|
||||
) -> Result<ClientProfile, ProtocolError> {
|
||||
@@ -1220,6 +1311,7 @@ async fn fetch_client_profile(
|
||||
build_command("servergrouplist", &[], &[]),
|
||||
channels,
|
||||
pending_moves,
|
||||
pending_downloads,
|
||||
voice_activity,
|
||||
poke_limiter,
|
||||
)
|
||||
@@ -1231,6 +1323,7 @@ async fn fetch_client_profile(
|
||||
build_command("channelgrouplist", &[], &[]),
|
||||
channels,
|
||||
pending_moves,
|
||||
pending_downloads,
|
||||
voice_activity,
|
||||
poke_limiter,
|
||||
)
|
||||
@@ -1246,6 +1339,7 @@ async fn fetch_client_profile(
|
||||
),
|
||||
channels,
|
||||
pending_moves,
|
||||
pending_downloads,
|
||||
voice_activity,
|
||||
poke_limiter,
|
||||
)
|
||||
@@ -1265,6 +1359,7 @@ async fn fetch_client_profile(
|
||||
build_command("getconnectioninfo", &[("clid", client_id.to_string())], &[]),
|
||||
channels,
|
||||
pending_moves,
|
||||
pending_downloads,
|
||||
voice_activity,
|
||||
poke_limiter,
|
||||
)
|
||||
@@ -1285,6 +1380,7 @@ async fn fetch_client_profile(
|
||||
database_id,
|
||||
channels,
|
||||
pending_moves,
|
||||
pending_downloads,
|
||||
voice_activity,
|
||||
poke_limiter,
|
||||
)
|
||||
@@ -1448,6 +1544,7 @@ async fn request_messages(
|
||||
command: OutCommand,
|
||||
channels: &EventChannels,
|
||||
pending_moves: &mut PendingMoves,
|
||||
pending_downloads: &mut PendingDownloads,
|
||||
voice_activity: &mut HashMap<u64, Instant>,
|
||||
poke_limiter: &mut PokeLimiter,
|
||||
) -> Result<Vec<InMessage>, ProtocolError> {
|
||||
@@ -1485,6 +1582,12 @@ async fn request_messages(
|
||||
StreamItem::Audio(buf) => {
|
||||
handle_audio_stream_item(&channels.voice_in, voice_activity, buf).await;
|
||||
}
|
||||
StreamItem::FileDownload(handle, result) => {
|
||||
handle_download_stream_item(pending_downloads, handle, result).await;
|
||||
}
|
||||
StreamItem::FiletransferFailed(handle, error) => {
|
||||
handle_download_failure(pending_downloads, handle, error);
|
||||
}
|
||||
other => handle_non_audio_stream_item(
|
||||
con,
|
||||
other,
|
||||
@@ -1503,6 +1606,7 @@ async fn request_client_db_info(
|
||||
dbid: tsclientlib::ClientDbId,
|
||||
channels: &EventChannels,
|
||||
pending_moves: &mut PendingMoves,
|
||||
pending_downloads: &mut PendingDownloads,
|
||||
voice_activity: &mut HashMap<u64, Instant>,
|
||||
poke_limiter: &mut PokeLimiter,
|
||||
) -> Result<InClientDbInfoPart, ProtocolError> {
|
||||
@@ -1511,6 +1615,7 @@ async fn request_client_db_info(
|
||||
build_command("clientdbinfo", &[("cldbid", dbid.0.to_string())], &[]),
|
||||
channels,
|
||||
pending_moves,
|
||||
pending_downloads,
|
||||
voice_activity,
|
||||
poke_limiter,
|
||||
)
|
||||
@@ -1568,6 +1673,14 @@ fn uid_to_avatar_path(uid_b64: &str) -> String {
|
||||
rendered
|
||||
}
|
||||
|
||||
fn avatar_download_path(client_uid: &str) -> String {
|
||||
format!("/avatar_{}", uid_to_avatar_path(client_uid))
|
||||
}
|
||||
|
||||
fn icon_download_path(icon_id: u64) -> String {
|
||||
format!("/icon_{icon_id}")
|
||||
}
|
||||
|
||||
fn find_client_by_id<'a>(
|
||||
clients: impl IntoIterator<Item = &'a Client>,
|
||||
client_id: u64,
|
||||
@@ -1951,10 +2064,11 @@ const _: () = {
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::{
|
||||
bounded_drain_stream, client_profile_refresh_plan, drain_voice_packets_for_tick,
|
||||
is_server_query_client_type, send_with_timeout, server_socket_from_config,
|
||||
sort_channels_tree_by, std_duration_millis, ConnectConfig, ProtocolClient, Request,
|
||||
SendTimeoutError, DISCONNECT_REPLY_TIMEOUT,
|
||||
avatar_download_path, bounded_drain_stream, client_profile_refresh_plan,
|
||||
drain_voice_packets_for_tick, icon_download_path, is_server_query_client_type,
|
||||
send_with_timeout, server_socket_from_config, sort_channels_tree_by,
|
||||
std_duration_millis, ConnectConfig, ProtocolClient, Request, SendTimeoutError,
|
||||
DISCONNECT_REPLY_TIMEOUT,
|
||||
};
|
||||
use futures::stream;
|
||||
use std::time::Duration;
|
||||
@@ -2052,6 +2166,16 @@ mod tests {
|
||||
assert!(plan.needs_channel_groups);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn avatar_download_path_uses_uid_hex_encoding() {
|
||||
assert_eq!(avatar_download_path("AQID"), "/avatar_abacad");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn icon_download_path_uses_unsigned_icon_id() {
|
||||
assert_eq!(icon_download_path(42), "/icon_42");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn channel_sort_linked_list_under_one_parent() {
|
||||
// Server emits four root-level channels in arbitrary HashMap
|
||||
|
||||
@@ -115,4 +115,8 @@ pub enum ProtocolError {
|
||||
/// should never see this; if they do, it is a mapping bug here.
|
||||
#[error("protocol backend: {0}")]
|
||||
Backend(String),
|
||||
|
||||
/// A file transfer failed while downloading protocol-owned assets.
|
||||
#[error("file transfer failed: {0}")]
|
||||
FileTransfer(String),
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user