hatchery_server/
messenger.rs

1//! P6: the hatchery messenger. It is the mail4agent mailbox engine (participants, addressing,
2//! rooms, delivery, acknowledgement; SQLite store) driven by the P2 provision ledger: a node is a
3//! participant exactly while the ledger says it is active and holds the `mail` grant, and room
4//! membership follows the ledger's rooms. The engine's rule is kept: a sender is never a field a
5//! caller fills in, it is the authenticated node the call arrives for.
6//!
7//! No second protocol: this is the same engine and wire types the m4a server uses. Participant
8//! secrets are generated by the engine and dropped here; nodes are identified by C2 (signature +
9//! policy), not by a mailbox secret. Server identity keys stay in the vault (P2).
10
11use crate::error::ServerError;
12use crate::provision::{read_ledger, LedgerEntry, NodeStatus};
13use mail4agent_api::{Address, MessageId, ParticipantId, RoomId, SendRequest};
14use mail4agent_core::{MailboxEngine, ParticipantPermissions};
15use mail4agent_store_sqlite::{migrations, Db, DbConfig, MigrationRunner, SqliteMailStore};
16use serde::Serialize;
17use std::collections::BTreeSet;
18use std::path::{Path, PathBuf};
19
20/// Reserved participant for the operator (HQ) side. Node ids may not use it.
21pub const OPERATOR: &str = "hatchery.operator";
22
23fn unix_ms() -> u64 {
24    std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap_or_default().as_millis() as u64
25}
26
27#[derive(Debug, thiserror::Error)]
28pub enum MessengerError {
29    #[error("not permitted")]
30    Forbidden,
31    #[error("{0}")]
32    Bad(String),
33    #[error("{0}")]
34    Internal(String),
35}
36
37#[derive(Serialize)]
38pub struct MailView {
39    pub message_id: String,
40    pub from: String,
41    pub to: String,
42    pub subject: String,
43    pub body: String,
44    pub created_at_ms: u64,
45}
46
47pub struct Messenger {
48    engine: MailboxEngine<SqliteMailStore>,
49    data_dir: PathBuf,
50    registered: BTreeSet<String>,
51    /// Rooms each registered node currently belongs to.
52    memberships: std::collections::BTreeMap<String, BTreeSet<String>>,
53    stamp: Option<(std::time::SystemTime, u64)>,
54}
55
56fn pid(id: &str) -> Result<ParticipantId, MessengerError> {
57    ParticipantId::new(id).map_err(|e| MessengerError::Bad(e.to_string()))
58}
59
60fn mail_allowed(entry: &LedgerEntry) -> bool {
61    matches!(entry.status, NodeStatus::Member | NodeStatus::Trusted)
62        && entry.grants.mailbox
63        && entry.grants.capabilities.iter().any(|c| c == "mail")
64}
65
66impl Messenger {
67    pub fn open(data_dir: &Path) -> Result<Self, ServerError> {
68        let dir = data_dir.join("messenger");
69        std::fs::create_dir_all(&dir)?;
70        let db = Db::open(&DbConfig::new(dir.join("mail.db"))).map_err(|e| ServerError::sub("messenger", format!("open: {e}")))?;
71        db.run_migrations_blocking(MigrationRunner::new(migrations()))
72            .map_err(|e| ServerError::sub("messenger", format!("migrate: {e}")))?;
73        let mut me = Self {
74            engine: MailboxEngine::new(SqliteMailStore::new(db)),
75            data_dir: data_dir.to_owned(),
76            registered: BTreeSet::new(),
77            memberships: Default::default(),
78            stamp: None,
79        };
80        let operator = pid(OPERATOR).map_err(|e| ServerError::sub("messenger", e.to_string()))?;
81        // Idempotent across restarts: already-registered is fine (the secret is never needed).
82        let _ = me.engine.register_participant(operator, Some("operator".into()), ParticipantPermissions { may_send: true, may_read: true, operator: true });
83        Ok(me)
84    }
85
86    /// Re-read the ledger when it changed and bring participants and rooms in line.
87    pub fn sync(&mut self) -> Result<(), MessengerError> {
88        let path = self.data_dir.join("provision").join("ledger.json");
89        let stamp = std::fs::metadata(&path).ok().and_then(|m| m.modified().ok().map(|t| (t, m.len())));
90        if stamp.is_some() && stamp == self.stamp {
91            return Ok(());
92        }
93        let entries = read_ledger(&self.data_dir).map_err(|e| MessengerError::Internal(e.to_string()))?;
94        let mut rooms_seen = BTreeSet::new();
95        for entry in &entries {
96            let allowed = mail_allowed(entry) && entry.node_id != OPERATOR;
97            let id = pid(&entry.node_id)?;
98            if allowed {
99                if !self.registered.contains(&entry.node_id) {
100                    // Already present after a restart is fine; otherwise register (secret dropped).
101                    let _ = self.engine.register_participant(id.clone(), Some(entry.node_id.clone()), ParticipantPermissions { may_send: true, may_read: true, operator: false });
102                    self.registered.insert(entry.node_id.clone());
103                }
104                let want: BTreeSet<String> = entry.grants.rooms.iter().cloned().collect();
105                let have = self.memberships.entry(entry.node_id.clone()).or_default().clone();
106                for room in want.difference(&have) {
107                    let rid = RoomId::new(room).map_err(|e| MessengerError::Bad(e.to_string()))?;
108                    let _ = self.engine.create_room(rid.clone(), unix_ms());
109                    self.engine.add_room_member(&rid, id.clone()).map_err(|e| MessengerError::Internal(e.to_string()))?;
110                }
111                for room in have.difference(&want) {
112                    let rid = RoomId::new(room).map_err(|e| MessengerError::Bad(e.to_string()))?;
113                    let _ = self.engine.remove_room_member(&rid, &id);
114                }
115                rooms_seen.extend(want.iter().cloned());
116                self.memberships.insert(entry.node_id.clone(), want);
117            } else {
118                // Suspended, revoked, neutral or without the grant: out of the mailbox entirely.
119                let _ = self.engine.deregister_participant(&id);
120                self.registered.remove(&entry.node_id);
121                if let Some(rooms) = self.memberships.remove(&entry.node_id) {
122                    for room in rooms {
123                        if let Ok(rid) = RoomId::new(&room) {
124                            let _ = self.engine.remove_room_member(&rid, &id);
125                        }
126                    }
127                }
128            }
129        }
130        self.stamp = stamp;
131        Ok(())
132    }
133
134    fn address(&mut self, who: &str) -> Result<Address, MessengerError> {
135        self.sync()?;
136        if who != OPERATOR && !self.registered.contains(who) {
137            return Err(MessengerError::Forbidden);
138        }
139        Ok(Address::Direct { participant: pid(who)? })
140    }
141
142    /// `to` is a node id (or `hatchery.operator`) or `room:NAME`.
143    pub fn send(&mut self, from: &str, to: &str, subject: &str, body: &str, idempotency_key: Option<String>) -> Result<String, MessengerError> {
144        let sender = self.address(from)?;
145        let to = if let Some(room) = to.strip_prefix("room:") {
146            Address::Room { room: RoomId::new(room).map_err(|e| MessengerError::Bad(e.to_string()))? }
147        } else {
148            Address::Direct { participant: pid(to)? }
149        };
150        let request = SendRequest { to, subject: subject.to_owned(), body: body.to_owned(), reply_to: None, correlation: None, refs: Vec::new(), idempotency_key };
151        let response = self.engine.send(&sender, request, unix_ms()).map_err(map_mail)?;
152        Ok(response.message_id.as_str().to_owned())
153    }
154
155    pub fn inbox(&mut self, who: &str, since_ms: u64, limit: u16) -> Result<Vec<MailView>, MessengerError> {
156        let reader = self.address(who)?;
157        let page = self.engine.inbox(&reader, since_ms, limit.clamp(1, 256)).map_err(map_mail)?;
158        Ok(page
159            .messages
160            .into_iter()
161            .map(|m| MailView {
162                message_id: m.message_id.as_str().to_owned(),
163                from: addr(&m.from),
164                to: addr(&m.to),
165                subject: m.subject,
166                body: m.body,
167                created_at_ms: m.created_at_unix_ms,
168            })
169            .collect())
170    }
171
172    /// Conversation between the operator and one node, both directions, oldest first. Empty for a
173    /// node that is not a mailbox (neutral, suspended, revoked).
174    pub fn thread(&mut self, node: &str, limit: u16) -> Result<Vec<MailView>, MessengerError> {
175        self.sync()?;
176        if !self.registered.contains(node) {
177            return Ok(Vec::new());
178        }
179        let operator = Address::Direct { participant: pid(OPERATOR)? };
180        let target = Address::Direct { participant: pid(node)? };
181        let to_op = self.engine.inbox(&operator, 0, 256).map_err(map_mail)?;
182        let to_node = self.engine.inbox_of(&operator, &target, 0, 256).map_err(map_mail)?;
183        let mut all: Vec<MailView> = to_op
184            .messages
185            .into_iter()
186            .filter(|m| addr(&m.from) == node)
187            .chain(to_node.messages.into_iter().filter(|m| addr(&m.from) == OPERATOR))
188            .map(|m| MailView { message_id: m.message_id.as_str().to_owned(), from: addr(&m.from), to: addr(&m.to), subject: m.subject, body: m.body, created_at_ms: m.created_at_unix_ms })
189            .collect();
190        all.sort_by_key(|m| (m.created_at_ms, m.message_id.clone()));
191        let skip = all.len().saturating_sub(limit.clamp(1, 256) as usize);
192        Ok(all.split_off(skip))
193    }
194
195    pub fn ack(&mut self, who: &str, message_id: &str) -> Result<(), MessengerError> {
196        let reader = self.address(who)?;
197        let id = MessageId::new(message_id).map_err(|e| MessengerError::Bad(e.to_string()))?;
198        self.engine.ack(&reader, &id, unix_ms()).map(|_| ()).map_err(map_mail)
199    }
200
201    pub fn is_registered(&mut self, who: &str) -> bool {
202        self.sync().is_ok() && self.registered.contains(who)
203    }
204}
205
206fn addr(a: &Address) -> String {
207    match a {
208        Address::Direct { participant } => participant.as_str().to_owned(),
209        Address::Session { participant, session } => format!("{}/{}", participant.as_str(), session.as_str()),
210        Address::Room { room } => format!("room:{}", room.as_str()),
211    }
212}
213
214fn map_mail(error: mail4agent_api::MailError) -> MessengerError {
215    use mail4agent_api::MailError as E;
216    match error {
217        E::PermissionDenied { .. } | E::UnknownParticipant { .. } | E::UnknownRoom { .. } => MessengerError::Forbidden,
218        other => MessengerError::Bad(other.to_string()),
219    }
220}
hatchery_server/messenger.rs

hatchery_server/
messenger.rs
