use std::{collections::HashMap, sync::Arc};
use base64::{Engine, engine::general_purpose::STANDARD as BASE64};
use kcode_kweb_db::{Node, NodeId};
use kcode_kweb_manager::KwebManager;
use serde_json::Value;
use uuid::Uuid;
use super::Config;
use kcode_kennedy_sessions::ResolvedObject;
const TELEGRAM_CAPTION_LIMIT_UTF16: usize = 1_024;
#[derive(Clone)]
pub struct LocalServices {
pub kmap: KwebManager,
pub intelligence: kcode_intelligence_router::Intelligence,
pub history: kcode_session_history::SessionHistory,
pub audio: kcode_audio_session_ingress::Coordinator,
pub directory: std::sync::Arc<kcode_telegram_identity::Directory>,
pub dev_tools: kcode_dev_tools::Service,
pub telegram: kcode_tg_kennedy_bot::Service,
}
#[derive(Debug, Clone)]
pub struct ApiError {
pub code: String,
pub message: String,
}
impl std::fmt::Display for ApiError {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter.write_str(&self.message)
}
}
impl std::error::Error for ApiError {}
#[derive(Clone)]
pub struct Api {
services: Arc<LocalServices>,
user_root_node_id: String,
kennedy_root_node_id: String,
telegram_user_locks: Arc<tokio::sync::Mutex<HashMap<i64, Arc<tokio::sync::Mutex<()>>>>>,
}
impl Api {
pub fn new(config: &Config, services: LocalServices) -> Self {
Self {
services: Arc::new(services),
user_root_node_id: config.user_root_node_id.clone(),
kennedy_root_node_id: config.kennedy_root_node_id.clone(),
telegram_user_locks: Arc::new(tokio::sync::Mutex::new(HashMap::new())),
}
}
pub async fn telegram_user_lock(&self, telegram_user_id: i64) -> Arc<tokio::sync::Mutex<()>> {
self.telegram_user_locks
.lock()
.await
.entry(telegram_user_id)
.or_insert_with(|| Arc::new(tokio::sync::Mutex::new(())))
.clone()
}
fn telegram(&self) -> &kcode_tg_kennedy_bot::Service {
&self.services.telegram
}
pub fn kmap_node(&self, node_id: &str) -> Result<Node, ApiError> {
let node_id = node_id.parse::<NodeId>().map_err(local_api_error)?;
self.services.kmap.get_node(node_id).map_err(kmap_error)
}
pub fn user_root_node_id(&self) -> &str {
&self.user_root_node_id
}
pub fn kennedy_root_node_id(&self) -> &str {
&self.kennedy_root_node_id
}
pub fn cancel_intelligence(&self, operation_id: Uuid) -> Result<bool, ApiError> {
self.services
.intelligence
.cancel(operation_id)
.map_err(intelligence_error)
}
pub fn history_health(&self) -> Result<(), ApiError> {
self.services.history.health().map_err(history_error)
}
pub async fn history_list(
&self,
) -> Result<Vec<kcode_session_history::SessionRecord>, ApiError> {
self.services.history.list().await.map_err(history_error)
}
pub async fn history_get_session(
&self,
id: &str,
) -> Result<kcode_session_history::SessionRecord, ApiError> {
self.services.history.get(id).await.map_err(history_error)
}
pub async fn history_register(
&self,
input: kcode_session_history::RegisterSession,
) -> Result<kcode_session_history::SessionRecord, ApiError> {
self.services
.history
.register(input)
.await
.map_err(history_error)
}
pub async fn history_command_heads(
&self,
) -> Result<Vec<kcode_session_history::SessionCommand>, ApiError> {
self.services
.history
.command_heads()
.await
.map_err(history_error)
}
pub async fn history_claim_command(
&self,
id: &str,
) -> Result<kcode_session_history::SessionCommand, ApiError> {
self.services
.history
.claim_command(id)
.await
.map_err(history_error)
}
pub async fn history_complete_command(
&self,
id: &str,
outcome: Value,
) -> Result<kcode_session_history::SessionCommand, ApiError> {
self.services
.history
.complete_command(id, kcode_session_history::CommandOutcome { outcome })
.await
.map_err(history_error)
}
pub fn history_listen_for_stop(
&self,
id: &str,
) -> Result<kcode_session_history::StopListener, ApiError> {
self.services
.history
.listen_for_stop(id)
.map_err(history_error)
}
pub async fn history_stop_heads(
&self,
) -> Result<Vec<kcode_session_history::SessionStopRequest>, ApiError> {
self.services
.history
.stop_heads()
.await
.map_err(history_error)
}
pub async fn history_complete_stop(
&self,
id: &str,
outcome: Value,
) -> Result<kcode_session_history::SessionStopRequest, ApiError> {
self.services
.history
.complete_stop(id, kcode_session_history::StopOutcome { outcome })
.await
.map_err(history_error)
}
pub async fn history_checkpoint(
&self,
id: &str,
input: kcode_session_history::Checkpoint,
) -> Result<kcode_session_history::SessionRecord, ApiError> {
self.services
.history
.checkpoint(id, input)
.await
.map_err(history_error)
}
pub async fn history_request_ingress(
&self,
id: &str,
input: kcode_session_history::Checkpoint,
) -> Result<kcode_session_history::SessionRecord, ApiError> {
self.services
.history
.request_ingress(id, input)
.await
.map_err(history_error)
}
pub async fn history_start_ingress(
&self,
id: &str,
input: kcode_session_history::StartIngress,
) -> Result<kcode_session_history::SessionRecord, ApiError> {
self.services
.history
.start_ingress(id, input)
.await
.map_err(history_error)
}
pub async fn history_complete_ingress(
&self,
id: &str,
expected_version: i64,
) -> Result<kcode_session_history::SessionRecord, ApiError> {
self.services
.history
.complete_ingress(
id,
kcode_session_history::ExpectedVersion { expected_version },
)
.await
.map_err(history_error)
}
pub async fn history_fail_ingress(
&self,
id: &str,
input: kcode_session_history::IngressFailure,
) -> Result<kcode_session_history::SessionRecord, ApiError> {
self.services
.history
.fail_ingress(id, input)
.await
.map_err(history_error)
}
pub async fn history_complete(
&self,
id: &str,
input: kcode_session_history::Checkpoint,
) -> Result<kcode_session_history::SessionRecord, ApiError> {
self.services
.history
.complete(id, input)
.await
.map_err(history_error)
}
pub async fn history_release_interrupted_ingress(&self) -> Result<Vec<String>, ApiError> {
self.services
.history
.release_interrupted_ingress()
.await
.map_err(history_error)
}
pub fn directory_user(
&self,
telegram_user_id: i64,
) -> Result<kcode_telegram_identity::User, ApiError> {
self.services
.directory
.user(telegram_user_id)
.map_err(directory_error)
}
pub fn directory_group(
&self,
group_id: &str,
) -> Result<kcode_telegram_identity::Group, ApiError> {
self.services
.directory
.group(group_id)
.map_err(directory_error)
}
pub async fn release_managed_sources(&self, session_id: &str) {
if let Err(error) = self.services.dev_tools.release(session_id.to_owned()).await {
tracing::warn!(error=%error.message, "Managed-source session release failed");
}
}
pub fn telegram_health(&self) {
let _ = self.telegram().status();
}
pub async fn telegram_private_sessions(
&self,
) -> Result<Vec<kcode_tg_kennedy_bot::PrivateSession>, ApiError> {
self.telegram()
.list_private_sessions()
.await
.map_err(telegram_error)
}
pub async fn telegram_events(&self) -> Result<Value, ApiError> {
self.telegram().list_events().await.map_err(telegram_error)
}
pub async fn telegram_group_ingress(&self) -> Result<Value, ApiError> {
self.telegram()
.list_group_ingress()
.await
.map_err(telegram_error)
}
pub async fn telegram_complete_group_ingress(&self, batch_id: &str) -> Result<Value, ApiError> {
self.telegram()
.complete_group_ingress(batch_id.to_owned())
.await
.map_err(telegram_error)
}
pub async fn telegram_group_session_updates(&self) -> Result<Value, ApiError> {
self.telegram()
.list_group_session_updates()
.await
.map_err(telegram_error)
}
pub async fn telegram_complete_silent_group_reset(
&self,
conversation_id: &str,
) -> Result<Value, ApiError> {
self.telegram()
.complete_silent_group_reset(conversation_id.to_owned())
.await
.map_err(telegram_error)
}
pub async fn telegram_acknowledge_group_context(
&self,
conversation_id: &str,
through_message_id: i64,
) -> Result<Value, ApiError> {
self.telegram()
.acknowledge_group_session_context(conversation_id.to_owned(), through_message_id)
.await
.map_err(telegram_error)
}
pub async fn telegram_detach_group_session(
&self,
conversation_id: &str,
group_id: &str,
telegram_user_id: i64,
) -> Result<Value, ApiError> {
self.telegram()
.detach_group_session(
conversation_id.to_owned(),
group_id.to_owned(),
telegram_user_id,
)
.await
.map_err(telegram_error)
}
pub async fn telegram_save_group_message_preparation(
&self,
chat_id: i64,
message_id: i64,
text: &str,
model: Option<&str>,
format: Option<&str>,
truncated: bool,
) -> Result<Value, ApiError> {
self.telegram()
.save_group_message_preparation(
chat_id,
message_id,
text.to_owned(),
model.map(ToOwned::to_owned),
format.map(ToOwned::to_owned),
truncated,
)
.await
.map_err(telegram_error)
}
pub async fn telegram_bind_event(
&self,
event_id: &str,
conversation_id: &str,
expected_conversation_id: Option<&str>,
) -> Result<Value, ApiError> {
self.telegram()
.bind_event(
event_id.to_owned(),
conversation_id.to_owned(),
expected_conversation_id.map(ToOwned::to_owned),
)
.await
.map_err(telegram_error)
}
pub async fn telegram_reply_event(
&self,
event_id: &str,
conversation_id: &str,
text: &str,
context_warning: Option<&str>,
) -> Result<Value, ApiError> {
self.telegram()
.reply_event(
event_id.to_owned(),
conversation_id.to_owned(),
text.to_owned(),
context_warning.map(ToOwned::to_owned),
)
.await
.map_err(telegram_error)
}
pub async fn telegram_abort_event(
&self,
event_id: &str,
conversation_id: Option<&str>,
message: &str,
) -> Result<Value, ApiError> {
self.telegram()
.abort_event(
event_id.to_owned(),
conversation_id.map(ToOwned::to_owned),
message.to_owned(),
)
.await
.map_err(telegram_error)
}
pub async fn telegram_interrupt_event(
&self,
event_id: &str,
conversation_id: &str,
) -> Result<Value, ApiError> {
self.telegram()
.interrupt_event(event_id.to_owned(), conversation_id.to_owned())
.await
.map_err(telegram_error)
}
pub async fn telegram_complete_reset(
&self,
event_id: &str,
message: Option<&str>,
) -> Result<Value, ApiError> {
self.telegram()
.complete_reset(event_id.to_owned(), message.map(ToOwned::to_owned))
.await
.map_err(telegram_error)
}
pub fn telegram_event_media(&self, event_id: &str) -> Result<(Vec<u8>, String), ApiError> {
self.telegram()
.event_media(event_id)
.map(|media| (media.bytes, media.media_type))
.map_err(telegram_error)
}
pub fn telegram_group_message_media(
&self,
chat_id: i64,
message_id: i64,
) -> Result<(Vec<u8>, String), ApiError> {
self.telegram()
.group_message_media(chat_id, message_id)
.map(|media| (media.bytes, media.media_type))
.map_err(telegram_error)
}
pub fn telegram_group_message_media_metadata(
&self,
chat_id: i64,
message_id: i64,
) -> Result<(u64, String), ApiError> {
self.telegram()
.group_message_media_metadata(chat_id, message_id)
.map(|media| (media.size_bytes, media.media_type))
.map_err(telegram_error)
}
pub async fn telegram_send_object(
&self,
event_id: &str,
conversation_id: &str,
file: &ResolvedObject,
caption: Option<&str>,
complete: bool,
) -> Result<Value, ApiError> {
self.telegram()
.send_event_attachment(
event_id.to_owned(),
conversation_id.to_owned(),
telegram_attachment(file, caption),
complete,
)
.await
.map_err(telegram_error)
}
pub async fn extract_document(
&self,
bytes: Vec<u8>,
filename: String,
mime: &str,
) -> Result<kcode_intelligence_router::DocumentExtraction, ApiError> {
self.services
.intelligence
.extract_document(kcode_intelligence_router::Document {
bytes,
file_name: filename,
content_type: mime.to_owned(),
})
.await
.map_err(intelligence_error)
}
pub async fn synchronize_audio_ingress(&self) -> Result<(), ApiError> {
self.services
.audio
.synchronize_completed_transcripts()
.await
.map_err(audio_error)
}
}
fn kmap_error(error: kcode_kweb_manager::Error) -> ApiError {
let (code, message) = match error.kind() {
kcode_kweb_manager::ErrorKind::InvalidInput => ("invalid_request", error.to_string()),
kcode_kweb_manager::ErrorKind::NotFound => ("not_found", error.to_string()),
kcode_kweb_manager::ErrorKind::Conflict => ("conflict", error.to_string()),
_ => (
"internal_error",
"An unexpected Kmap database error occurred.".into(),
),
};
ApiError {
code: code.into(),
message,
}
}
fn intelligence_error(error: kcode_intelligence_router::Error) -> ApiError {
ApiError {
code: error.code().into(),
message: error.message().into(),
}
}
fn directory_error(error: kcode_telegram_identity::Error) -> ApiError {
let code = match error.kind() {
kcode_telegram_identity::ErrorKind::InvalidInput => "invalid_request",
kcode_telegram_identity::ErrorKind::NotFound => "not_found",
kcode_telegram_identity::ErrorKind::Conflict => "state_conflict",
kcode_telegram_identity::ErrorKind::Storage => "internal_error",
};
ApiError {
code: code.into(),
message: error.message().into(),
}
}
fn history_error(error: kcode_session_history::Error) -> ApiError {
ApiError {
code: error.kind.code().into(),
message: error.message,
}
}
fn audio_error(error: kcode_audio_session_ingress::Error) -> ApiError {
let internal = error.kind() == kcode_audio_session_ingress::ErrorKind::Internal;
ApiError {
code: match error.kind() {
kcode_audio_session_ingress::ErrorKind::InvalidInput => "invalid_request",
kcode_audio_session_ingress::ErrorKind::NotFound => "not_found",
kcode_audio_session_ingress::ErrorKind::Conflict => "state_conflict",
kcode_audio_session_ingress::ErrorKind::Internal => "internal_error",
}
.into(),
message: if internal {
"An unexpected Kennedy audio error occurred.".into()
} else {
error.message().into()
},
}
}
fn local_api_error(error: impl std::fmt::Display) -> ApiError {
ApiError {
code: "invalid_request".into(),
message: error.to_string(),
}
}
fn telegram_error(error: kcode_tg_kennedy_bot::Error) -> ApiError {
ApiError {
code: error.code().to_owned(),
message: error.message().to_owned(),
}
}
fn telegram_attachment(
file: &ResolvedObject,
caption: Option<&str>,
) -> kcode_tg_kennedy_bot::Attachment {
kcode_tg_kennedy_bot::Attachment {
bytes: file.bytes.clone(),
file_name: Some(file.file_name.clone()),
media_type: Some(file.media_type.clone()),
kind: telegram_native_kind(&file.media_type, file.transport_kind.as_deref())
.map(ToOwned::to_owned),
caption: caption.map(ToOwned::to_owned),
}
}
pub fn telegram_caption_for<'a>(file: &ResolvedObject, text: &'a str) -> Option<&'a str> {
if text.is_empty() || text.encode_utf16().count() > TELEGRAM_CAPTION_LIMIT_UTF16 {
return None;
}
if matches!(
telegram_native_kind(&file.media_type, file.transport_kind.as_deref()),
Some("video_note" | "sticker")
) {
return None;
}
Some(text)
}
pub fn data_url(mime: &str, bytes: &[u8]) -> String {
format!("data:{mime};base64,{}", BASE64.encode(bytes))
}
fn telegram_native_kind(media_type: &str, transport_kind: Option<&str>) -> Option<&'static str> {
match transport_kind {
Some("photo") => return Some("photo"),
Some("video") => return Some("video"),
Some("animation") => return Some("animation"),
Some("audio") => return Some("audio"),
Some("video_note") => return Some("video_note"),
Some("sticker") => return Some("sticker"),
_ => {}
}
if media_type == "image/gif" {
Some("animation")
} else if media_type.starts_with("image/") {
Some("photo")
} else if media_type.starts_with("video/") {
Some("video")
} else if media_type.starts_with("audio/") {
Some("audio")
} else {
None
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn telegram_captions_are_exact_and_fall_back_when_telegram_cannot_attach_them() {
let file = |media_type: &str, transport_kind: Option<&str>| ResolvedObject {
object_id: "object".into(),
bytes: vec![1],
file_name: "object.bin".into(),
media_type: media_type.into(),
transport_kind: transport_kind.map(ToOwned::to_owned),
};
let exact = " exact caption\n";
assert_eq!(
telegram_caption_for(&file("image/jpeg", Some("photo")), exact),
Some(exact)
);
assert_eq!(
telegram_caption_for(&file("application/pdf", Some("document")), exact),
Some(exact)
);
assert_eq!(
telegram_caption_for(&file("image/webp", Some("sticker")), exact),
None
);
assert_eq!(
telegram_caption_for(&file("video/mp4", Some("video_note")), exact),
None
);
assert_eq!(
telegram_caption_for(
&file("image/jpeg", Some("photo")),
&"x".repeat(TELEGRAM_CAPTION_LIMIT_UTF16 + 1),
),
None
);
}
}