35ef7e8
use actix_web::http::header;
use actix_web::web::Bytes;
use futures::stream::BoxStream;
use imaging::layout::{ImageVariant, PreviewState};
use imaging::preset::PreviewPreset;
use models::chat::attachments::{self, AttachmentStatus};
use sea_orm::sea_query::Expr;
use sea_orm::{
ActiveModelTrait, ColumnTrait, Condition, EntityTrait, PaginatorTrait, QueryFilter,
QuerySelect, Set,
};
use serde::{Deserialize, Serialize};
use time::OffsetDateTime;
use utoipa::ToSchema;
use uuid::Uuid;
use crate::errors::ChatError;
use crate::services::service::ChatService;
pub const ATTACHMENT_PREFIX: &str = "chat/attachments";
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct AttachmentDto {
pub id: Uuid,
pub user_id: i64,
pub room_id: Option<Uuid>,
pub message_id: Option<Uuid>,
pub filename: String,
pub description: Option<String>,
pub content_type: String,
pub size: i64,
pub status: AttachmentStatus,
pub width: Option<i32>,
pub height: Option<i32>,
pub duration_secs: Option<i32>,
pub created_at: OffsetDateTime,
}
impl From<attachments::Model> for AttachmentDto {
fn from(a: attachments::Model) -> Self {
Self {
id: a.id,
user_id: a.user_id,
room_id: a.room_id,
message_id: a.message_id,
filename: a.filename,
description: a.description,
content_type: a.content_type,
size: a.size,
status: a.status,
width: a.width,
height: a.height,
duration_secs: a.duration_secs,
created_at: a.created_at,
}
}
}
#[derive(Debug, Clone, Deserialize, ToSchema)]
pub struct CreateAttachmentCmd {
pub filename: String,
pub content_type: String,
pub size: i64,
#[serde(default)]
pub description: Option<String>,
}
#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct UploadIntentDto {
pub id: Uuid,
pub upload_url: String,
pub expires_secs: u64,
}
#[derive(Debug, Clone, Deserialize, ToSchema)]
pub struct FinalizeAttachmentCmd {
#[serde(default)]
pub width: Option<i32>,
#[serde(default)]
pub height: Option<i32>,
#[serde(default)]
pub duration_secs: Option<i32>,
}
fn sanitize_filename(name: &str) -> String {
let base = name.rsplit(['/', '\\']).next().unwrap_or("file");
let cleaned: String = base
.chars()
.map(|c| match c {
'a'..='z' | 'A'..='Z' | '0'..='9' | '-' | '_' | '.' => c,
_ => '_',
})
.collect();
if cleaned.is_empty() || cleaned.chars().all(|c| c == '.') {
"file".to_string()
} else {
cleaned
}
}
/// Content types that browsers may execute/script inline; never accepted or
/// served embedded.
const UNSAFE_INLINE_TYPES: &[&str] = &[
"text/html",
"application/xhtml+xml",
"image/svg+xml",
"text/xml",
"application/xml",
"text/javascript",
"application/javascript",
"application/x-javascript",
"application/ecmascript",
];
fn is_unsafe_inline(content_type: &str) -> bool {
let main = content_type.split(';').next().unwrap_or("").trim();
UNSAFE_INLINE_TYPES
.iter()
.any(|blocked| main.eq_ignore_ascii_case(blocked))
}
/// Whether the attachment is an inline image, i.e. something worth deriving a
/// preview from. Kept separate from [`is_inline_renderable`] so the preview
/// machinery never touches video/audio/PDF.
fn is_inline_image(content_type: &str) -> bool {
let main = content_type.split(';').next().unwrap_or("").trim();
main.to_ascii_lowercase().starts_with("image/")
}
/// Content type for a streamed original, mirroring the safety rule applied to
/// response headers: an unsafe inline type is served as a plain download.
fn attachment_stream_content_type(row: &attachments::Model) -> String {
if is_unsafe_inline(&row.content_type) {
"application/octet-stream".to_string()
} else {
row.content_type.clone()
}
}
fn state_ready_key(state: &PreviewState) -> String {
match state {
PreviewState::Ready(ready) => ready.key.clone(),
PreviewState::Skipped(_) => String::new(),
}
}
fn state_ready_content_type(state: &PreviewState) -> String {
match state {
PreviewState::Ready(ready) => ready.content_type.clone(),
PreviewState::Skipped(_) => "application/octet-stream".to_string(),
}
}
fn is_inline_renderable(content_type: &str) -> bool {
let main = content_type.split(';').next().unwrap_or("").trim();
main.starts_with("image/")
|| main.starts_with("video/")
|| main.starts_with("audio/")
|| main.eq_ignore_ascii_case("text/plain")
|| main.eq_ignore_ascii_case("text/markdown")
|| main.eq_ignore_ascii_case("application/pdf")
}
impl ChatService {
pub async fn create_attachment(
&self,
caller: i64,
cmd: CreateAttachmentCmd,
) -> Result<UploadIntentDto, ChatError> {
let max_size = self.config().chat.upload.max_size;
if cmd.size <= 0 || cmd.size > max_size {
return Err(ChatError::BadRequest(format!(
"file size must be within 1..{max_size} bytes"
)));
}
if cmd.filename.trim().is_empty() || cmd.filename.chars().count() > 255 {
return Err(ChatError::BadRequest("invalid filename".to_string()));
}
if cmd.content_type.trim().is_empty() || is_unsafe_inline(&cmd.content_type) {
return Err(ChatError::BadRequest(
"unsupported or unsafe content_type".to_string(),
));
}
let unattached = attachments::Entity::find()
.filter(attachments::Column::UserId.eq(caller))
.filter(attachments::Column::MessageId.is_null())
.count(self.db().read())
.await?;
if unattached >= self.config().chat.max_pending_attachments_per_user as u64 {
return Err(ChatError::RateLimited(format!(
"too many unused attachments (max {}), finalize or reference them in a message first",
self.config().chat.max_pending_attachments_per_user
)));
}
let now = OffsetDateTime::now_utc();
let id = Uuid::now_v7();
let filename = sanitize_filename(&cmd.filename);
let key = format!("{ATTACHMENT_PREFIX}/{caller}/{id}/{filename}");
let bucket = self.storage().bucket_name().to_string();
let row = attachments::ActiveModel {
id: Set(id),
user_id: Set(caller),
room_id: Set(None),
message_id: Set(None),
filename: Set(filename.clone()),
description: Set(cmd.description.clone()),
content_type: Set(cmd.content_type.clone()),
size: Set(cmd.size),
bucket: Set(bucket),
s3_key: Set(key.clone()),
status: Set(AttachmentStatus::Pending),
width: Set(None),
height: Set(None),
duration_secs: Set(None),
variants: Set(serde_json::json!([])),
created_at: Set(now),
updated_at: Set(now),
}
.insert(self.db().write())
.await?;
let ttl = self.config().chat.upload.presign_ttl_secs.min(604_800);
let mut put_headers = http::HeaderMap::new();
if let Ok(value) = cmd.content_type.parse() {
put_headers.insert(http::header::CONTENT_TYPE, value);
}
let upload_url = self
.storage()
.presign_put_with_headers(&key, ttl as u32, put_headers)
.map_err(ChatError::from)?;
let _ = row;
Ok(UploadIntentDto {
id,
upload_url,
expires_secs: ttl,
})
}
pub async fn finalize_attachment(
&self,
caller: i64,
attachment_id: Uuid,
cmd: FinalizeAttachmentCmd,
) -> Result<AttachmentDto, ChatError> {
let row = attachments::Entity::find_by_id(attachment_id)
.one(self.db().write())
.await?
.filter(|a| a.user_id == caller)
.ok_or(ChatError::NotFound)?;
if row.status != AttachmentStatus::Pending {
return Err(ChatError::Conflict(
"attachment already finalized".to_string(),
));
}
let head = self.storage().head(&row.s3_key).await?;
let Some((actual_size, actual_content_type)) = head else {
return Err(ChatError::BadRequest("object not uploaded yet".to_string()));
};
let max_size = self.config().chat.upload.max_size;
if actual_size == 0 || actual_size > max_size as u64 {
return Err(ChatError::BadRequest(format!(
"uploaded object size {actual_size} outside 1..{max_size} bytes"
)));
}
let stored_content_type = if actual_content_type.is_empty() {
row.content_type.clone()
} else {
actual_content_type.clone()
};
if is_unsafe_inline(&stored_content_type) {
return Err(ChatError::BadRequest(
"uploaded object has an unsafe content_type".to_string(),
));
}
let now = OffsetDateTime::now_utc();
let mut am: attachments::ActiveModel = row.into();
am.size = Set(actual_size as i64);
am.content_type = Set(stored_content_type.clone());
am.status = Set(AttachmentStatus::Uploaded);
am.width = Set(cmd.width);
am.height = Set(cmd.height);
am.duration_secs = Set(cmd.duration_secs);
am.updated_at = Set(now);
let updated = am.update(self.db().write()).await?;
// Build the list-view derivative while the bytes are still hot. A
// failure here must never fail the upload: the attachment is already
// valid, and the read path falls back to the original.
if is_inline_image(&stored_content_type)
&& let Err(error) = self.generate_attachment_preview(&updated).await
{
tracing::warn!(%error, attachment_id = %updated.id, "attachment preview generation failed");
}
Ok(AttachmentDto::from(updated))
}
/// Render and store the derivative for one uploaded attachment, recording
/// the outcome (including "skipped") in `variants` so a read never repeats
/// the decision.
async fn generate_attachment_preview(&self, row: &attachments::Model) -> Result<(), ChatError> {
let key_base = export::imaging::preview_key_base_of(&row.s3_key);
let preview = export::imaging::ensure_preview(
self.storage(),
&self.state.redis,
&key_base,
&row.s3_key,
&row.content_type,
PreviewPreset::Content,
None,
)
.await;
let Some(state) = preview else {
tracing::debug!(attachment_id = %row.id, "attachment preview unavailable; serving the original");
return Ok(());
};
self.record_attachment_preview(row.id, row.variants.clone(), &state)
.await
}
/// Persist which derivative (or skip reason) belongs to one attachment.
async fn record_attachment_preview(
&self,
attachment_id: Uuid,
mut variants: serde_json::Value,
state: &PreviewState,
) -> Result<(), ChatError> {
imaging::layout::preview_into_variants(&mut variants, state);
attachments::Entity::update_many()
.filter(attachments::Column::Id.eq(attachment_id))
.col_expr(attachments::Column::Variants, Expr::value(variants))
.exec(self.db().write())
.await?;
Ok(())
}
/// Remove both encodings of an attachment's derivative. Safe to call for any
/// key: a non-image attachment has nothing stored under these names.
async fn delete_attachment_previews(&self, original_key: &str) {
let key_base = export::imaging::preview_key_base_of(original_key);
for candidate in imaging::layout::preview_candidate_keys(&key_base) {
if let Err(error) = self.storage().delete(&candidate).await {
tracing::warn!(%error, key = %candidate, "failed to delete attachment preview");
}
}
}
/// Locator of the representation to stream: the derivative when it exists,
/// the original otherwise. Legacy rows (finalized before previews existed)
/// are healed here on first demand.
async fn attachment_variant_locator(
&self,
row: &attachments::Model,
variant: ImageVariant,
) -> (String, String) {
if variant == ImageVariant::Original || !is_inline_image(&row.content_type) {
return (row.s3_key.clone(), attachment_stream_content_type(row));
}
let key_base = export::imaging::preview_key_base_of(&row.s3_key);
let recorded = imaging::layout::preview_from_variants(&row.variants);
let state = export::imaging::ensure_preview(
self.storage(),
&self.state.redis,
&key_base,
&row.s3_key,
&row.content_type,
PreviewPreset::Content,
recorded.as_ref(),
)
.await;
match state {
Some(state @ PreviewState::Ready(_)) => {
if recorded.as_ref() != Some(&state)
&& let Err(error) = self
.record_attachment_preview(row.id, row.variants.clone(), &state)
.await
{
tracing::warn!(%error, attachment_id = %row.id, "failed to record attachment preview");
}
(state_ready_key(&state), state_ready_content_type(&state))
}
Some(PreviewState::Skipped(_)) | None => {
(row.s3_key.clone(), attachment_stream_content_type(row))
}
}
}
pub async fn cleanup_stale_attachments(&self) -> Result<u64, ChatError> {
let now = OffsetDateTime::now_utc();
let pending_cutoff = now - time::Duration::hours(24);
let uploaded_cutoff = now - time::Duration::days(7);
let stale = attachments::Entity::find()
.filter(
Condition::any()
.add(
Condition::all()
.add(attachments::Column::Status.eq(AttachmentStatus::Pending))
.add(attachments::Column::CreatedAt.lt(pending_cutoff)),
)
.add(
Condition::all()
.add(attachments::Column::Status.eq(AttachmentStatus::Uploaded))
.add(attachments::Column::MessageId.is_null())
.add(attachments::Column::CreatedAt.lt(uploaded_cutoff)),
),
)
.limit(200)
.all(self.db().write())
.await?;
let count = stale.len() as u64;
for row in stale {
let shared = attachments::Entity::find()
.filter(attachments::Column::S3Key.eq(row.s3_key.clone()))
.filter(attachments::Column::Id.ne(row.id))
.count(self.db().read())
.await
.unwrap_or(1);
let _ = attachments::Entity::delete_by_id(row.id)
.exec(self.db().write())
.await;
if shared == 0 {
let _ = self.storage().delete(&row.s3_key).await;
self.delete_attachment_previews(&row.s3_key).await;
}
}
Ok(count)
}
pub async fn get_attachment_meta(
&self,
caller: i64,
attachment_id: Uuid,
) -> Result<AttachmentDto, ChatError> {
let row = attachments::Entity::find_by_id(attachment_id)
.one(self.db().read())
.await?
.ok_or(ChatError::NotFound)?;
self.check_attachment_access(caller, &row).await?;
Ok(AttachmentDto::from(row))
}
/// Stream one representation of an attachment.
///
/// `ImageVariant::Original` is what the download link and the lightbox use;
/// `ImageVariant::Preview` is what the transcript grid loads.
pub async fn download_attachment(
&self,
caller: i64,
attachment_id: Uuid,
variant: ImageVariant,
) -> Result<
(
attachments::Model,
BoxStream<'static, Result<Bytes, std::io::Error>>,
String,
),
ChatError,
> {
let row = attachments::Entity::find_by_id(attachment_id)
.one(self.db().read())
.await?
.ok_or(ChatError::NotFound)?;
self.check_attachment_access(caller, &row).await?;
if row.status == AttachmentStatus::Pending {
return Err(ChatError::NotFound);
}
let (key, content_type) = self.attachment_variant_locator(&row, variant).await;
let stream = self.storage().get_stream(&key).await?;
let body = Box::pin(futures::StreamExt::map(stream, |item| {
item.map(Bytes::from)
.map_err(|e| std::io::Error::other(e.to_string()))
})) as BoxStream<'static, Result<Bytes, std::io::Error>>;
Ok((row, body, content_type))
}
async fn check_attachment_access(
&self,
caller: i64,
row: &attachments::Model,
) -> Result<(), ChatError> {
if row.user_id == caller {
return Ok(());
}
if let Some(room_id) = row.room_id {
// Mirror message visibility: anyone who can read the room can read
// its attachments (public rooms do not require a membership row).
if crate::share::lookup::require_room_readable(self, room_id, caller)
.await
.is_ok()
{
return Ok(());
}
}
Err(ChatError::Forbidden)
}
/// Response headers for one streamed representation. `content_type` is what
/// is actually being served (the derivative may be JPEG where the original
/// was PNG), so it is passed in rather than re-read from the row.
pub fn attachment_content_headers(
row: &attachments::Model,
content_type: String,
) -> Vec<(header::HeaderName, String)> {
let mut headers = Vec::new();
headers.push((header::CONTENT_TYPE, content_type));
headers.push((
header::HeaderName::from_static("x-content-type-options"),
"nosniff".to_string(),
));
let disposition = if is_inline_renderable(&row.content_type) {
format!("inline; filename=\"{}\"", row.filename.replace('"', "_"))
} else {
format!(
"attachment; filename=\"{}\"",
row.filename.replace('"', "_")
)
};
headers.push((header::CONTENT_DISPOSITION, disposition));
headers
}
}