kcode_kennedy_session_objects/

lib.rs

//! Object resolution and media staging for Kennedy sessions.

#![forbid(unsafe_code)]

use anyhow::Context as _;
use kcode_kweb_db::ObjectId;
use kcode_server_object_envelopes::{StoredFile, sanitize_file_name};
use kcode_session_history::{
    Session as HistorySession,
    chatend::{ObjectMetadata, PendingId},
};
use kcode_telegram_session_coordinator::{Attachment, AttachmentRequest};
use serde_json::{Value, json};

/// Exact bytes and authoritative delivery metadata for a session object.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ResolvedObject {
    pub object_id: String,
    pub bytes: Vec<u8>,
    pub file_name: String,
    pub media_type: String,
    pub transport_kind: Option<String>,
}

/// Authoritative facts for an object staged in the session journal.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct StagedDescriptor {
    pub pending_id: String,
    pub file_name: String,
    pub media_type: String,
    pub size_bytes: u64,
    pub transport_kind: Option<String>,
}

impl StagedDescriptor {
    /// Replaces caller-supplied descriptor facts with authoritative values.
    pub fn apply_to(&self, descriptor: &mut Value) {
        if !descriptor.is_object() {
            *descriptor = json!({});
        }
        descriptor["pendingId"] = json!(self.pending_id);
        descriptor["fileName"] = json!(self.file_name);
        descriptor["extension"] = json!(file_name_extension(&self.file_name));
        descriptor["mimeType"] = json!(self.media_type);
        descriptor["sizeBytes"] = json!(self.size_bytes);
    }
}

/// The result of staging or reusing one Telegram group media object.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct StagedTelegramMedia {
    pub descriptor: StagedDescriptor,
    pub kind: String,
    pub reused: bool,
}

/// Correlation, limits, metadata, and timestamp for one Telegram staging call.
#[derive(Clone, Debug, PartialEq)]
pub struct TelegramStageRequest {
    pub chat_id: i64,
    pub message_id: i64,
    pub maximum_bytes: u64,
    pub transport_metadata: Value,
    pub recorded_at: String,
}

/// Resolves a pending journal object or a canonical object through a caller
/// supplied read boundary.
pub fn resolve_object(
    journal: &mut HistorySession,
    object_id: &str,
    read_canonical: impl FnOnce(&str) -> anyhow::Result<StoredFile>,
) -> anyhow::Result<ResolvedObject> {
    if object_id.starts_with("pending:") {
        let pending_id = PendingId::parse(object_id.to_owned())?;
        let location = journal
            .objects()
            .get(&pending_id)
            .cloned()
            .with_context(|| {
                format!("staged object {pending_id} does not exist in this session")
            })?;
        let transport_kind = staged_object_transport_kind(journal, &pending_id);
        let bytes = journal.read_object(&pending_id)?;
        anyhow::ensure!(
            bytes.len() as u64 == location.payload_len,
            "staged object {pending_id} declared {} bytes but resolved to {}",
            location.payload_len,
            bytes.len()
        );
        let fallback = format!("object-{}.bin", pending_id.number());
        return Ok(ResolvedObject {
            object_id: pending_id.to_string(),
            bytes,
            file_name: sanitize_file_name(
                location.metadata.file_name.as_deref().unwrap_or_default(),
                &fallback,
            ),
            media_type: location.metadata.media_type,
            transport_kind,
        });
    }

    let canonical_id = object_id
        .parse::<ObjectId>()
        .with_context(|| format!("{object_id:?} is not an object ID"))?;
    let file = read_canonical(object_id)?;
    anyhow::ensure!(
        file.object_id == canonical_id,
        "object store returned {} while resolving {canonical_id}",
        file.object_id
    );
    Ok(ResolvedObject {
        object_id: canonical_id.to_string(),
        bytes: file.bytes,
        file_name: file.file_name,
        media_type: file.media_type,
        transport_kind: file.transport_kind,
    })
}

/// Resolves and normalizes media while enforcing the current nonempty and size
/// requirements.
pub fn resolve_media_object(
    journal: &mut HistorySession,
    object_id: &str,
    maximum_bytes: u64,
    read_canonical: impl FnOnce(&str) -> anyhow::Result<StoredFile>,
) -> anyhow::Result<ResolvedObject> {
    let mut resolved = resolve_object(journal, object_id, read_canonical)?;
    resolved.media_type = normalize_media_type(&resolved.media_type);
    anyhow::ensure!(
        !resolved.bytes.is_empty(),
        "media object {} is empty",
        resolved.object_id
    );
    anyhow::ensure!(
        resolved.bytes.len() as u64 <= maximum_bytes,
        "media object {} is {} bytes, over the {}-byte enrichment limit",
        resolved.object_id,
        resolved.bytes.len(),
        maximum_bytes
    );
    Ok(resolved)
}

/// Returns authoritative staged metadata, using the same safe fallback filename
/// as object resolution.
pub fn staged_descriptor(
    journal: &HistorySession,
    pending_id: &PendingId,
) -> anyhow::Result<StagedDescriptor> {
    let location = journal
        .objects()
        .get(pending_id)
        .with_context(|| format!("user-provided object {pending_id} is not staged"))?;
    let fallback = format!("object-{}.bin", pending_id.number());
    Ok(StagedDescriptor {
        pending_id: pending_id.to_string(),
        file_name: sanitize_file_name(
            location.metadata.file_name.as_deref().unwrap_or_default(),
            &fallback,
        ),
        media_type: normalize_media_type(&location.metadata.media_type),
        size_bytes: location.payload_len,
        transport_kind: staged_object_transport_kind(journal, pending_id),
    })
}

/// Reuses media with matching Telegram correlation metadata or downloads and
/// stages it once. The closures keep transport lookup and naming outside this
/// crate.
pub fn stage_telegram_group_media(
    journal: &mut HistorySession,
    request: TelegramStageRequest,
    download: impl FnOnce() -> anyhow::Result<(Vec<u8>, String)>,
    file_name_for_media_type: impl FnOnce(&str) -> String,
) -> anyhow::Result<StagedTelegramMedia> {
    if let Some((pending_id, metadata, size_bytes)) =
        find_staged_telegram_group_media(journal, request.chat_id, request.message_id)
    {
        return telegram_stage_result(journal, pending_id, &metadata, size_bytes, true);
    }

    let (bytes, downloaded_media_type) = download()?;
    anyhow::ensure!(
        !bytes.is_empty(),
        "Telegram group media message {} is empty",
        request.message_id
    );
    anyhow::ensure!(
        bytes.len() as u64 <= request.maximum_bytes,
        "Telegram group media message {} is {} bytes, over the {}-byte enrichment limit",
        request.message_id,
        bytes.len(),
        request.maximum_bytes
    );
    let media_type = normalize_media_type(&downloaded_media_type);
    let file_name = file_name_for_media_type(&media_type);
    let pending_id = journal.stage_object(
        request.recorded_at,
        media_type,
        Some(file_name),
        request.transport_metadata,
        &bytes,
    )?;
    let metadata = journal
        .objects()
        .get(&pending_id)
        .context("newly staged Telegram group media is missing")?
        .metadata
        .clone();
    telegram_stage_result(journal, pending_id, &metadata, bytes.len() as u64, false)
}

/// Resolves delivery requests and constructs coordinator attachments without
/// invoking a transport.
pub fn delivery_attachments(
    journal: &mut HistorySession,
    requests: Vec<AttachmentRequest>,
    mut read_canonical: impl FnMut(&str) -> anyhow::Result<StoredFile>,
) -> anyhow::Result<Vec<Attachment>> {
    requests
        .into_iter()
        .map(|request| {
            let object = resolve_object(journal, &request.object_id, |id| read_canonical(id))?;
            let file_name = request
                .file_name
                .unwrap_or_else(|| object.file_name.clone());
            Ok(Attachment {
                object_id: object.object_id,
                bytes: object.bytes,
                file_name,
                media_type: object.media_type,
                transport_kind: object.transport_kind,
            })
        })
        .collect()
}

fn staged_object_transport_kind(
    journal: &HistorySession,
    pending_id: &PendingId,
) -> Option<String> {
    let pending_id_text = pending_id.to_string();
    for state in journal.state().boxes.values() {
        let Some(index) = state
            .canonical
            .content
            .objects
            .iter()
            .position(|object_id| object_id == &pending_id_text)
        else {
            continue;
        };
        let metadata = &state.canonical.content.metadata;
        let descriptor = metadata
            .get("attachments")
            .and_then(Value::as_array)
            .and_then(|attachments| {
                attachments
                    .iter()
                    .find(|attachment| {
                        attachment.get("pendingId").and_then(Value::as_str)
                            == Some(pending_id_text.as_str())
                    })
                    .or_else(|| attachments.get(index))
            })
            .or_else(|| metadata.get("media").filter(|value| value.is_object()));
        if let Some(kind) = descriptor
            .and_then(|descriptor| descriptor.get("kind"))
            .and_then(Value::as_str)
            .filter(|kind| !kind.trim().is_empty())
        {
            return Some(kind.to_owned());
        }
    }
    journal
        .objects()
        .get(pending_id)
        .and_then(|location| location.metadata.transport.get("kind"))
        .and_then(Value::as_str)
        .filter(|kind| !kind.trim().is_empty())
        .map(str::to_owned)
}

fn find_staged_telegram_group_media(
    journal: &HistorySession,
    chat_id: i64,
    message_id: i64,
) -> Option<(PendingId, ObjectMetadata, u64)> {
    journal.objects().iter().find_map(|(pending_id, location)| {
        let transport = &location.metadata.transport;
        (transport.get("source").and_then(Value::as_str) == Some("telegram-group")
            && transport.get("chatId").and_then(Value::as_i64) == Some(chat_id)
            && transport.get("messageId").and_then(Value::as_i64) == Some(message_id))
        .then(|| {
            (
                pending_id.clone(),
                location.metadata.clone(),
                location.payload_len,
            )
        })
    })
}

fn telegram_stage_result(
    journal: &HistorySession,
    pending_id: PendingId,
    metadata: &ObjectMetadata,
    size_bytes: u64,
    reused: bool,
) -> anyhow::Result<StagedTelegramMedia> {
    let file_name = metadata
        .file_name
        .as_deref()
        .filter(|value| !value.trim().is_empty())
        .with_context(|| {
            format!(
                "staged object {} has no authoritative filename",
                metadata.pending_id
            )
        })?
        .to_owned();
    Ok(StagedTelegramMedia {
        descriptor: StagedDescriptor {
            pending_id: pending_id.to_string(),
            file_name,
            media_type: normalize_media_type(&metadata.media_type),
            size_bytes,
            transport_kind: staged_object_transport_kind(journal, &pending_id),
        },
        kind: metadata
            .transport
            .get("kind")
            .and_then(Value::as_str)
            .unwrap_or("media")
            .to_owned(),
        reused,
    })
}

fn normalize_media_type(value: &str) -> String {
    value
        .split(';')
        .next()
        .unwrap_or(value)
        .trim()
        .to_ascii_lowercase()
}

fn file_name_extension(file_name: &str) -> String {
    file_name
        .rsplit_once('.')
        .and_then(|(stem, extension)| {
            (!stem.is_empty() && !extension.is_empty()).then_some(extension)
        })
        .map(|extension| format!(".{extension}"))
        .unwrap_or_else(|| "(none)".into())
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn descriptor_replaces_untrusted_file_facts() {
        let descriptor = StagedDescriptor {
            pending_id: "pending:7".into(),
            file_name: "voice.OGG".into(),
            media_type: "audio/ogg".into(),
            size_bytes: 42,
            transport_kind: Some("voice".into()),
        };
        let mut value = json!({
            "pendingId":"pending:wrong",
            "fileName":"../../wrong",
            "mimeType":"text/plain",
            "dataUrl":"retained only when caller has not sanitized it"
        });
        descriptor.apply_to(&mut value);
        assert_eq!(value["pendingId"], "pending:7");
        assert_eq!(value["fileName"], "voice.OGG");
        assert_eq!(value["extension"], ".OGG");
        assert_eq!(value["mimeType"], "audio/ogg");
        assert_eq!(value["sizeBytes"], 42);
    }

    #[test]
    fn normalization_is_parameter_insensitive() {
        assert_eq!(
            normalize_media_type(" Audio/OGG ; codecs=opus"),
            "audio/ogg"
        );
        assert_eq!(file_name_extension(".hidden"), "(none)");
        assert_eq!(file_name_extension("report.pdf"), ".pdf");
    }
}