1use std::{collections::HashMap, sync::Arc};
4
5use base64::{Engine, engine::general_purpose::STANDARD as BASE64};
6use kcode_kweb_db::{Node, NodeId};
7use kcode_kweb_manager::KwebManager;
8use serde_json::Value;
9use uuid::Uuid;
10
11use super::Config;
12use kcode_kennedy_sessions::ResolvedObject;
13
14const TELEGRAM_CAPTION_LIMIT_UTF16: usize = 1_024;
15
16#[derive(Clone)]
17pub struct LocalServices {
18 pub kmap: KwebManager,
19 pub intelligence: kcode_intelligence_router::Intelligence,
20 pub history: kcode_session_history::SessionHistory,
21 pub audio: kcode_audio_session_ingress::Coordinator,
22 pub directory: std::sync::Arc<kcode_telegram_identity::Directory>,
23 pub dev_tools: kcode_dev_tools::Service,
24 pub telegram: kcode_tg_kennedy_bot::Service,
25}
26
27#[derive(Debug, Clone)]
28pub struct ApiError {
29 pub code: String,
30 pub message: String,
31}
32
33impl std::fmt::Display for ApiError {
34 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
35 formatter.write_str(&self.message)
36 }
37}
38
39impl std::error::Error for ApiError {}
40
41#[derive(Clone)]
42pub struct Api {
43 services: Arc<LocalServices>,
44 user_root_node_id: String,
45 kennedy_root_node_id: String,
46 telegram_user_locks: Arc<tokio::sync::Mutex<HashMap<i64, Arc<tokio::sync::Mutex<()>>>>>,
47}
48
49impl Api {
50 pub fn new(config: &Config, services: LocalServices) -> Self {
51 Self {
52 services: Arc::new(services),
53 user_root_node_id: config.user_root_node_id.clone(),
54 kennedy_root_node_id: config.kennedy_root_node_id.clone(),
55 telegram_user_locks: Arc::new(tokio::sync::Mutex::new(HashMap::new())),
56 }
57 }
58
59 pub async fn telegram_user_lock(&self, telegram_user_id: i64) -> Arc<tokio::sync::Mutex<()>> {
60 self.telegram_user_locks
61 .lock()
62 .await
63 .entry(telegram_user_id)
64 .or_insert_with(|| Arc::new(tokio::sync::Mutex::new(())))
65 .clone()
66 }
67
68 fn telegram(&self) -> &kcode_tg_kennedy_bot::Service {
69 &self.services.telegram
70 }
71
72 pub fn kmap_node(&self, node_id: &str) -> Result<Node, ApiError> {
73 let node_id = node_id.parse::<NodeId>().map_err(local_api_error)?;
74 self.services.kmap.get_node(node_id).map_err(kmap_error)
75 }
76
77 pub fn user_root_node_id(&self) -> &str {
78 &self.user_root_node_id
79 }
80
81 pub fn kennedy_root_node_id(&self) -> &str {
82 &self.kennedy_root_node_id
83 }
84
85 pub fn cancel_intelligence(&self, operation_id: Uuid) -> Result<bool, ApiError> {
86 self.services
87 .intelligence
88 .cancel(operation_id)
89 .map_err(intelligence_error)
90 }
91
92 pub fn history_health(&self) -> Result<(), ApiError> {
93 self.services.history.health().map_err(history_error)
94 }
95
96 pub async fn history_list(
97 &self,
98 ) -> Result<Vec<kcode_session_history::SessionRecord>, ApiError> {
99 self.services.history.list().await.map_err(history_error)
100 }
101
102 pub async fn history_get_session(
103 &self,
104 id: &str,
105 ) -> Result<kcode_session_history::SessionRecord, ApiError> {
106 self.services.history.get(id).await.map_err(history_error)
107 }
108
109 pub async fn history_register(
110 &self,
111 input: kcode_session_history::RegisterSession,
112 ) -> Result<kcode_session_history::SessionRecord, ApiError> {
113 self.services
114 .history
115 .register(input)
116 .await
117 .map_err(history_error)
118 }
119
120 pub async fn history_command_heads(
121 &self,
122 ) -> Result<Vec<kcode_session_history::SessionCommand>, ApiError> {
123 self.services
124 .history
125 .command_heads()
126 .await
127 .map_err(history_error)
128 }
129
130 pub async fn history_claim_command(
131 &self,
132 id: &str,
133 ) -> Result<kcode_session_history::SessionCommand, ApiError> {
134 self.services
135 .history
136 .claim_command(id)
137 .await
138 .map_err(history_error)
139 }
140
141 pub async fn history_complete_command(
142 &self,
143 id: &str,
144 outcome: Value,
145 ) -> Result<kcode_session_history::SessionCommand, ApiError> {
146 self.services
147 .history
148 .complete_command(id, kcode_session_history::CommandOutcome { outcome })
149 .await
150 .map_err(history_error)
151 }
152
153 pub fn history_listen_for_stop(
154 &self,
155 id: &str,
156 ) -> Result<kcode_session_history::StopListener, ApiError> {
157 self.services
158 .history
159 .listen_for_stop(id)
160 .map_err(history_error)
161 }
162
163 pub async fn history_stop_heads(
164 &self,
165 ) -> Result<Vec<kcode_session_history::SessionStopRequest>, ApiError> {
166 self.services
167 .history
168 .stop_heads()
169 .await
170 .map_err(history_error)
171 }
172
173 pub async fn history_complete_stop(
174 &self,
175 id: &str,
176 outcome: Value,
177 ) -> Result<kcode_session_history::SessionStopRequest, ApiError> {
178 self.services
179 .history
180 .complete_stop(id, kcode_session_history::StopOutcome { outcome })
181 .await
182 .map_err(history_error)
183 }
184
185 pub async fn history_checkpoint(
186 &self,
187 id: &str,
188 input: kcode_session_history::Checkpoint,
189 ) -> Result<kcode_session_history::SessionRecord, ApiError> {
190 self.services
191 .history
192 .checkpoint(id, input)
193 .await
194 .map_err(history_error)
195 }
196
197 pub async fn history_request_ingress(
198 &self,
199 id: &str,
200 input: kcode_session_history::Checkpoint,
201 ) -> Result<kcode_session_history::SessionRecord, ApiError> {
202 self.services
203 .history
204 .request_ingress(id, input)
205 .await
206 .map_err(history_error)
207 }
208
209 pub async fn history_start_ingress(
210 &self,
211 id: &str,
212 input: kcode_session_history::StartIngress,
213 ) -> Result<kcode_session_history::SessionRecord, ApiError> {
214 self.services
215 .history
216 .start_ingress(id, input)
217 .await
218 .map_err(history_error)
219 }
220
221 pub async fn history_complete_ingress(
222 &self,
223 id: &str,
224 expected_version: i64,
225 ) -> Result<kcode_session_history::SessionRecord, ApiError> {
226 self.services
227 .history
228 .complete_ingress(
229 id,
230 kcode_session_history::ExpectedVersion { expected_version },
231 )
232 .await
233 .map_err(history_error)
234 }
235
236 pub async fn history_fail_ingress(
237 &self,
238 id: &str,
239 input: kcode_session_history::IngressFailure,
240 ) -> Result<kcode_session_history::SessionRecord, ApiError> {
241 self.services
242 .history
243 .fail_ingress(id, input)
244 .await
245 .map_err(history_error)
246 }
247
248 pub async fn history_complete(
249 &self,
250 id: &str,
251 input: kcode_session_history::Checkpoint,
252 ) -> Result<kcode_session_history::SessionRecord, ApiError> {
253 self.services
254 .history
255 .complete(id, input)
256 .await
257 .map_err(history_error)
258 }
259
260 pub async fn history_release_interrupted_ingress(&self) -> Result<Vec<String>, ApiError> {
261 self.services
262 .history
263 .release_interrupted_ingress()
264 .await
265 .map_err(history_error)
266 }
267
268 pub fn directory_user(
269 &self,
270 telegram_user_id: i64,
271 ) -> Result<kcode_telegram_identity::User, ApiError> {
272 self.services
273 .directory
274 .user(telegram_user_id)
275 .map_err(directory_error)
276 }
277
278 pub fn directory_group(
279 &self,
280 group_id: &str,
281 ) -> Result<kcode_telegram_identity::Group, ApiError> {
282 self.services
283 .directory
284 .group(group_id)
285 .map_err(directory_error)
286 }
287
288 pub async fn release_managed_sources(&self, session_id: &str) {
289 if let Err(error) = self.services.dev_tools.release(session_id.to_owned()).await {
290 tracing::warn!(error=%error.message, "Managed-source session release failed");
291 }
292 }
293
294 pub fn telegram_health(&self) {
295 let _ = self.telegram().status();
296 }
297
298 pub async fn telegram_private_sessions(
299 &self,
300 ) -> Result<Vec<kcode_tg_kennedy_bot::PrivateSession>, ApiError> {
301 self.telegram()
302 .list_private_sessions()
303 .await
304 .map_err(telegram_error)
305 }
306
307 pub async fn telegram_events(&self) -> Result<Value, ApiError> {
308 self.telegram().list_events().await.map_err(telegram_error)
309 }
310
311 pub async fn telegram_group_ingress(&self) -> Result<Value, ApiError> {
312 self.telegram()
313 .list_group_ingress()
314 .await
315 .map_err(telegram_error)
316 }
317
318 pub async fn telegram_complete_group_ingress(&self, batch_id: &str) -> Result<Value, ApiError> {
319 self.telegram()
320 .complete_group_ingress(batch_id.to_owned())
321 .await
322 .map_err(telegram_error)
323 }
324
325 pub async fn telegram_group_session_updates(&self) -> Result<Value, ApiError> {
326 self.telegram()
327 .list_group_session_updates()
328 .await
329 .map_err(telegram_error)
330 }
331
332 pub async fn telegram_complete_silent_group_reset(
333 &self,
334 conversation_id: &str,
335 ) -> Result<Value, ApiError> {
336 self.telegram()
337 .complete_silent_group_reset(conversation_id.to_owned())
338 .await
339 .map_err(telegram_error)
340 }
341
342 pub async fn telegram_acknowledge_group_context(
343 &self,
344 conversation_id: &str,
345 through_message_id: i64,
346 ) -> Result<Value, ApiError> {
347 self.telegram()
348 .acknowledge_group_session_context(conversation_id.to_owned(), through_message_id)
349 .await
350 .map_err(telegram_error)
351 }
352
353 pub async fn telegram_detach_group_session(
354 &self,
355 conversation_id: &str,
356 group_id: &str,
357 telegram_user_id: i64,
358 ) -> Result<Value, ApiError> {
359 self.telegram()
360 .detach_group_session(
361 conversation_id.to_owned(),
362 group_id.to_owned(),
363 telegram_user_id,
364 )
365 .await
366 .map_err(telegram_error)
367 }
368
369 pub async fn telegram_save_group_message_preparation(
370 &self,
371 chat_id: i64,
372 message_id: i64,
373 text: &str,
374 model: Option<&str>,
375 format: Option<&str>,
376 truncated: bool,
377 ) -> Result<Value, ApiError> {
378 self.telegram()
379 .save_group_message_preparation(
380 chat_id,
381 message_id,
382 text.to_owned(),
383 model.map(ToOwned::to_owned),
384 format.map(ToOwned::to_owned),
385 truncated,
386 )
387 .await
388 .map_err(telegram_error)
389 }
390
391 pub async fn telegram_bind_event(
392 &self,
393 event_id: &str,
394 conversation_id: &str,
395 expected_conversation_id: Option<&str>,
396 ) -> Result<Value, ApiError> {
397 self.telegram()
398 .bind_event(
399 event_id.to_owned(),
400 conversation_id.to_owned(),
401 expected_conversation_id.map(ToOwned::to_owned),
402 )
403 .await
404 .map_err(telegram_error)
405 }
406
407 pub async fn telegram_reply_event(
408 &self,
409 event_id: &str,
410 conversation_id: &str,
411 text: &str,
412 context_warning: Option<&str>,
413 ) -> Result<Value, ApiError> {
414 self.telegram()
415 .reply_event(
416 event_id.to_owned(),
417 conversation_id.to_owned(),
418 text.to_owned(),
419 context_warning.map(ToOwned::to_owned),
420 )
421 .await
422 .map_err(telegram_error)
423 }
424
425 pub async fn telegram_abort_event(
426 &self,
427 event_id: &str,
428 conversation_id: Option<&str>,
429 message: &str,
430 ) -> Result<Value, ApiError> {
431 self.telegram()
432 .abort_event(
433 event_id.to_owned(),
434 conversation_id.map(ToOwned::to_owned),
435 message.to_owned(),
436 )
437 .await
438 .map_err(telegram_error)
439 }
440
441 pub async fn telegram_interrupt_event(
442 &self,
443 event_id: &str,
444 conversation_id: &str,
445 ) -> Result<Value, ApiError> {
446 self.telegram()
447 .interrupt_event(event_id.to_owned(), conversation_id.to_owned())
448 .await
449 .map_err(telegram_error)
450 }
451
452 pub async fn telegram_complete_reset(
453 &self,
454 event_id: &str,
455 message: Option<&str>,
456 ) -> Result<Value, ApiError> {
457 self.telegram()
458 .complete_reset(event_id.to_owned(), message.map(ToOwned::to_owned))
459 .await
460 .map_err(telegram_error)
461 }
462
463 pub fn telegram_event_media(&self, event_id: &str) -> Result<(Vec<u8>, String), ApiError> {
464 self.telegram()
465 .event_media(event_id)
466 .map(|media| (media.bytes, media.media_type))
467 .map_err(telegram_error)
468 }
469
470 pub fn telegram_group_message_media(
471 &self,
472 chat_id: i64,
473 message_id: i64,
474 ) -> Result<(Vec<u8>, String), ApiError> {
475 self.telegram()
476 .group_message_media(chat_id, message_id)
477 .map(|media| (media.bytes, media.media_type))
478 .map_err(telegram_error)
479 }
480
481 pub fn telegram_group_message_media_metadata(
482 &self,
483 chat_id: i64,
484 message_id: i64,
485 ) -> Result<(u64, String), ApiError> {
486 self.telegram()
487 .group_message_media_metadata(chat_id, message_id)
488 .map(|media| (media.size_bytes, media.media_type))
489 .map_err(telegram_error)
490 }
491
492 pub async fn telegram_send_object(
493 &self,
494 event_id: &str,
495 conversation_id: &str,
496 file: &ResolvedObject,
497 caption: Option<&str>,
498 complete: bool,
499 ) -> Result<Value, ApiError> {
500 self.telegram()
501 .send_event_attachment(
502 event_id.to_owned(),
503 conversation_id.to_owned(),
504 telegram_attachment(file, caption),
505 complete,
506 )
507 .await
508 .map_err(telegram_error)
509 }
510
511 pub async fn extract_document(
512 &self,
513 bytes: Vec<u8>,
514 filename: String,
515 mime: &str,
516 ) -> Result<kcode_intelligence_router::DocumentExtraction, ApiError> {
517 self.services
518 .intelligence
519 .extract_document(kcode_intelligence_router::Document {
520 bytes,
521 file_name: filename,
522 content_type: mime.to_owned(),
523 })
524 .await
525 .map_err(intelligence_error)
526 }
527
528 pub async fn synchronize_audio_ingress(&self) -> Result<(), ApiError> {
529 self.services
530 .audio
531 .synchronize_completed_transcripts()
532 .await
533 .map_err(audio_error)
534 }
535}
536
537fn kmap_error(error: kcode_kweb_manager::Error) -> ApiError {
538 let (code, message) = match error.kind() {
539 kcode_kweb_manager::ErrorKind::InvalidInput => ("invalid_request", error.to_string()),
540 kcode_kweb_manager::ErrorKind::NotFound => ("not_found", error.to_string()),
541 kcode_kweb_manager::ErrorKind::Conflict => ("conflict", error.to_string()),
542 _ => (
543 "internal_error",
544 "An unexpected Kmap database error occurred.".into(),
545 ),
546 };
547 ApiError {
548 code: code.into(),
549 message,
550 }
551}
552
553fn intelligence_error(error: kcode_intelligence_router::Error) -> ApiError {
554 ApiError {
555 code: error.code().into(),
556 message: error.message().into(),
557 }
558}
559
560fn directory_error(error: kcode_telegram_identity::Error) -> ApiError {
561 let code = match error.kind() {
562 kcode_telegram_identity::ErrorKind::InvalidInput => "invalid_request",
563 kcode_telegram_identity::ErrorKind::NotFound => "not_found",
564 kcode_telegram_identity::ErrorKind::Conflict => "state_conflict",
565 kcode_telegram_identity::ErrorKind::Storage => "internal_error",
566 };
567 ApiError {
568 code: code.into(),
569 message: error.message().into(),
570 }
571}
572
573fn history_error(error: kcode_session_history::Error) -> ApiError {
574 ApiError {
575 code: error.kind.code().into(),
576 message: error.message,
577 }
578}
579
580fn audio_error(error: kcode_audio_session_ingress::Error) -> ApiError {
581 let internal = error.kind() == kcode_audio_session_ingress::ErrorKind::Internal;
582 ApiError {
583 code: match error.kind() {
584 kcode_audio_session_ingress::ErrorKind::InvalidInput => "invalid_request",
585 kcode_audio_session_ingress::ErrorKind::NotFound => "not_found",
586 kcode_audio_session_ingress::ErrorKind::Conflict => "state_conflict",
587 kcode_audio_session_ingress::ErrorKind::Internal => "internal_error",
588 }
589 .into(),
590 message: if internal {
591 "An unexpected Kennedy audio error occurred.".into()
592 } else {
593 error.message().into()
594 },
595 }
596}
597
598fn local_api_error(error: impl std::fmt::Display) -> ApiError {
599 ApiError {
600 code: "invalid_request".into(),
601 message: error.to_string(),
602 }
603}
604
605fn telegram_error(error: kcode_tg_kennedy_bot::Error) -> ApiError {
606 ApiError {
607 code: error.code().to_owned(),
608 message: error.message().to_owned(),
609 }
610}
611
612fn telegram_attachment(
613 file: &ResolvedObject,
614 caption: Option<&str>,
615) -> kcode_tg_kennedy_bot::Attachment {
616 kcode_tg_kennedy_bot::Attachment {
617 bytes: file.bytes.clone(),
618 file_name: Some(file.file_name.clone()),
619 media_type: Some(file.media_type.clone()),
620 kind: telegram_native_kind(&file.media_type, file.transport_kind.as_deref())
621 .map(ToOwned::to_owned),
622 caption: caption.map(ToOwned::to_owned),
623 }
624}
625
626pub fn telegram_caption_for<'a>(file: &ResolvedObject, text: &'a str) -> Option<&'a str> {
627 if text.is_empty() || text.encode_utf16().count() > TELEGRAM_CAPTION_LIMIT_UTF16 {
628 return None;
629 }
630 if matches!(
631 telegram_native_kind(&file.media_type, file.transport_kind.as_deref()),
632 Some("video_note" | "sticker")
633 ) {
634 return None;
635 }
636 Some(text)
637}
638
639pub fn data_url(mime: &str, bytes: &[u8]) -> String {
640 format!("data:{mime};base64,{}", BASE64.encode(bytes))
641}
642
643fn telegram_native_kind(media_type: &str, transport_kind: Option<&str>) -> Option<&'static str> {
644 match transport_kind {
645 Some("photo") => return Some("photo"),
646 Some("video") => return Some("video"),
647 Some("animation") => return Some("animation"),
648 Some("audio") => return Some("audio"),
649 Some("video_note") => return Some("video_note"),
650 Some("sticker") => return Some("sticker"),
651 _ => {}
652 }
653 if media_type == "image/gif" {
654 Some("animation")
655 } else if media_type.starts_with("image/") {
656 Some("photo")
657 } else if media_type.starts_with("video/") {
658 Some("video")
659 } else if media_type.starts_with("audio/") {
660 Some("audio")
661 } else {
662 None
663 }
664}
665
666#[cfg(test)]
667mod tests {
668 use super::*;
669
670 #[test]
671 fn telegram_captions_are_exact_and_fall_back_when_telegram_cannot_attach_them() {
672 let file = |media_type: &str, transport_kind: Option<&str>| ResolvedObject {
673 object_id: "object".into(),
674 bytes: vec![1],
675 file_name: "object.bin".into(),
676 media_type: media_type.into(),
677 transport_kind: transport_kind.map(ToOwned::to_owned),
678 };
679 let exact = " exact caption\n";
680 assert_eq!(
681 telegram_caption_for(&file("image/jpeg", Some("photo")), exact),
682 Some(exact)
683 );
684 assert_eq!(
685 telegram_caption_for(&file("application/pdf", Some("document")), exact),
686 Some(exact)
687 );
688 assert_eq!(
689 telegram_caption_for(&file("image/webp", Some("sticker")), exact),
690 None
691 );
692 assert_eq!(
693 telegram_caption_for(&file("video/mp4", Some("video_note")), exact),
694 None
695 );
696 assert_eq!(
697 telegram_caption_for(
698 &file("image/jpeg", Some("photo")),
699 &"x".repeat(TELEGRAM_CAPTION_LIMIT_UTF16 + 1),
700 ),
701 None
702 );
703 }
704}