cortexfs-channels 0.1.0

Runtime-neutral multi-channel messaging abstractions for agents.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82

use serde_json::json;

use super::{ChannelCodec, OutboundRequest, object, participant, scalar, text, timestamp_ms};
use crate::{
    ChannelError, ChannelId, ConversationId, InboundMessage, MessageTarget, OutboundMessage,
};

/// Telegram Bot API update codec. Polling and HTTP transport remain host-owned.
#[derive(Clone, Copy, Debug, Default)]
pub struct TelegramCodec;

impl ChannelCodec for TelegramCodec {
    fn channel(&self) -> ChannelId {
        ChannelId::from_static("telegram")
    }

    fn decode(&self, payload: &str) -> Result<Option<InboundMessage>, ChannelError> {
        let root = object(payload)?;
        if root.get("ok").and_then(serde_json::Value::as_bool) == Some(false) {
            return Ok(None);
        }
        let message = root.get("message").or_else(|| root.get("channel_post"));
        let Some(message) = message else {
            return Ok(None);
        };
        let id = scalar(message.get("message_id"), "message_id")?;
        let chat = message
            .get("chat")
            .ok_or_else(|| ChannelError::Protocol("telegram chat is missing".to_owned()))?;
        let conversation = ConversationId::new(scalar(chat.get("id"), "chat.id")?)?;
        let sender_id = message
            .get("from")
            .map(|from| scalar(from.get("id"), "from.id"))
            .transpose()?
            .unwrap_or_else(|| conversation.to_string());
        Ok(Some(InboundMessage {
            id,
            target: MessageTarget {
                channel: self.channel(),
                conversation,
                thread: message
                    .get("message_thread_id")
                    .map(|value| scalar(Some(value), "message_thread_id"))
                    .transpose()?,
                reply_to: message
                    .get("reply_to_message")
                    .and_then(|reply| reply.get("message_id"))
                    .map(|value| scalar(Some(value), "reply_to_message.message_id"))
                    .transpose()?,
            },
            sender: participant(message.get("from"), sender_id),
            body: text(message.get("text").or_else(|| message.get("caption")))?,
            timestamp_ms: timestamp_ms(message.get("date")),
            metadata: std::collections::BTreeMap::new(),
        }))
    }

    fn encode(&self, message: &OutboundMessage) -> Result<OutboundRequest, ChannelError> {
        message.body.validate()?;
        if !message.body.attachments.is_empty() {
            return Err(ChannelError::Unsupported(
                "telegram media attachments".to_owned(),
            ));
        }
        let mut fields = serde_json::Map::new();
        fields.insert(
            "chat_id".to_owned(),
            json!(message.target.conversation.as_str()),
        );
        fields.insert("text".to_owned(), json!(message.body.text));
        if let Some(reply) = message.target.reply_to.as_deref() {
            fields.insert("reply_to_message_id".to_owned(), json!(reply));
        }
        Ok(OutboundRequest {
            method: "POST".to_owned(),
            path: "sendMessage".to_owned(),
            content_type: "application/json".to_owned(),
            body: serde_json::Value::Object(fields).to_string(),
            headers: std::collections::BTreeMap::new(),
        })
    }
}