hatchery-server 0.2.15

//! P6: the hatchery messenger. It is the mail4agent mailbox engine (participants, addressing,
//! rooms, delivery, acknowledgement; SQLite store) driven by the P2 provision ledger: a node is a
//! participant exactly while the ledger says it is active and holds the `mail` grant, and room
//! membership follows the ledger's rooms. The engine's rule is kept: a sender is never a field a
//! caller fills in, it is the authenticated node the call arrives for.
//!
//! No second protocol: this is the same engine and wire types the m4a server uses. Participant
//! secrets are generated by the engine and dropped here; nodes are identified by C2 (signature +
//! policy), not by a mailbox secret. Server identity keys stay in the vault (P2).

use crate::error::ServerError;
use crate::provision::{read_ledger, LedgerEntry, NodeStatus};
use mail4agent_api::{Address, MessageId, ParticipantId, RoomId, SendRequest};
use mail4agent_core::{MailboxEngine, ParticipantPermissions};
use mail4agent_store_sqlite::{migrations, Db, DbConfig, MigrationRunner, SqliteMailStore};
use serde::Serialize;
use std::collections::BTreeSet;
use std::path::{Path, PathBuf};

/// Reserved participant for the operator (HQ) side. Node ids may not use it.
pub const OPERATOR: &str = "hatchery.operator";

fn unix_ms() -> u64 {
    std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap_or_default().as_millis() as u64
}

#[derive(Debug, thiserror::Error)]
pub enum MessengerError {
    #[error("not permitted")]
    Forbidden,
    #[error("{0}")]
    Bad(String),
    #[error("{0}")]
    Internal(String),
}

#[derive(Serialize)]
pub struct MailView {
    pub message_id: String,
    pub from: String,
    pub to: String,
    pub subject: String,
    pub body: String,
    pub created_at_ms: u64,
}

pub struct Messenger {
    engine: MailboxEngine<SqliteMailStore>,
    data_dir: PathBuf,
    registered: BTreeSet<String>,
    /// Rooms each registered node currently belongs to.
    memberships: std::collections::BTreeMap<String, BTreeSet<String>>,
    stamp: Option<(std::time::SystemTime, u64)>,
}

fn pid(id: &str) -> Result<ParticipantId, MessengerError> {
    ParticipantId::new(id).map_err(|e| MessengerError::Bad(e.to_string()))
}

fn mail_allowed(entry: &LedgerEntry) -> bool {
    matches!(entry.status, NodeStatus::Member | NodeStatus::Trusted)
        && entry.grants.mailbox
        && entry.grants.capabilities.iter().any(|c| c == "mail")
}

impl Messenger {
    pub fn open(data_dir: &Path) -> Result<Self, ServerError> {
        let dir = data_dir.join("messenger");
        std::fs::create_dir_all(&dir)?;
        let db = Db::open(&DbConfig::new(dir.join("mail.db"))).map_err(|e| ServerError::sub("messenger", format!("open: {e}")))?;
        db.run_migrations_blocking(MigrationRunner::new(migrations()))
            .map_err(|e| ServerError::sub("messenger", format!("migrate: {e}")))?;
        let mut me = Self {
            engine: MailboxEngine::new(SqliteMailStore::new(db)),
            data_dir: data_dir.to_owned(),
            registered: BTreeSet::new(),
            memberships: Default::default(),
            stamp: None,
        };
        let operator = pid(OPERATOR).map_err(|e| ServerError::sub("messenger", e.to_string()))?;
        // Idempotent across restarts: already-registered is fine (the secret is never needed).
        let _ = me.engine.register_participant(operator, Some("operator".into()), ParticipantPermissions { may_send: true, may_read: true, operator: true });
        Ok(me)
    }

    /// Re-read the ledger when it changed and bring participants and rooms in line.
    pub fn sync(&mut self) -> Result<(), MessengerError> {
        let path = self.data_dir.join("provision").join("ledger.json");
        let stamp = std::fs::metadata(&path).ok().and_then(|m| m.modified().ok().map(|t| (t, m.len())));
        if stamp.is_some() && stamp == self.stamp {
            return Ok(());
        }
        let entries = read_ledger(&self.data_dir).map_err(|e| MessengerError::Internal(e.to_string()))?;
        let mut rooms_seen = BTreeSet::new();
        for entry in &entries {
            let allowed = mail_allowed(entry) && entry.node_id != OPERATOR;
            let id = pid(&entry.node_id)?;
            if allowed {
                if !self.registered.contains(&entry.node_id) {
                    // Already present after a restart is fine; otherwise register (secret dropped).
                    let _ = self.engine.register_participant(id.clone(), Some(entry.node_id.clone()), ParticipantPermissions { may_send: true, may_read: true, operator: false });
                    self.registered.insert(entry.node_id.clone());
                }
                let want: BTreeSet<String> = entry.grants.rooms.iter().cloned().collect();
                let have = self.memberships.entry(entry.node_id.clone()).or_default().clone();
                for room in want.difference(&have) {
                    let rid = RoomId::new(room).map_err(|e| MessengerError::Bad(e.to_string()))?;
                    let _ = self.engine.create_room(rid.clone(), unix_ms());
                    self.engine.add_room_member(&rid, id.clone()).map_err(|e| MessengerError::Internal(e.to_string()))?;
                }
                for room in have.difference(&want) {
                    let rid = RoomId::new(room).map_err(|e| MessengerError::Bad(e.to_string()))?;
                    let _ = self.engine.remove_room_member(&rid, &id);
                }
                rooms_seen.extend(want.iter().cloned());
                self.memberships.insert(entry.node_id.clone(), want);
            } else {
                // Suspended, revoked, neutral or without the grant: out of the mailbox entirely.
                let _ = self.engine.deregister_participant(&id);
                self.registered.remove(&entry.node_id);
                if let Some(rooms) = self.memberships.remove(&entry.node_id) {
                    for room in rooms {
                        if let Ok(rid) = RoomId::new(&room) {
                            let _ = self.engine.remove_room_member(&rid, &id);
                        }
                    }
                }
            }
        }
        self.stamp = stamp;
        Ok(())
    }

    fn address(&mut self, who: &str) -> Result<Address, MessengerError> {
        self.sync()?;
        if who != OPERATOR && !self.registered.contains(who) {
            return Err(MessengerError::Forbidden);
        }
        Ok(Address::Direct { participant: pid(who)? })
    }

    /// `to` is a node id (or `hatchery.operator`) or `room:NAME`.
    pub fn send(&mut self, from: &str, to: &str, subject: &str, body: &str, idempotency_key: Option<String>) -> Result<String, MessengerError> {
        let sender = self.address(from)?;
        let to = if let Some(room) = to.strip_prefix("room:") {
            Address::Room { room: RoomId::new(room).map_err(|e| MessengerError::Bad(e.to_string()))? }
        } else {
            Address::Direct { participant: pid(to)? }
        };
        let request = SendRequest { to, subject: subject.to_owned(), body: body.to_owned(), reply_to: None, correlation: None, refs: Vec::new(), idempotency_key };
        let response = self.engine.send(&sender, request, unix_ms()).map_err(map_mail)?;
        Ok(response.message_id.as_str().to_owned())
    }

    pub fn inbox(&mut self, who: &str, since_ms: u64, limit: u16) -> Result<Vec<MailView>, MessengerError> {
        let reader = self.address(who)?;
        let page = self.engine.inbox(&reader, since_ms, limit.clamp(1, 256)).map_err(map_mail)?;
        Ok(page
            .messages
            .into_iter()
            .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,
            })
            .collect())
    }

    /// Conversation between the operator and one node, both directions, oldest first. Empty for a
    /// node that is not a mailbox (neutral, suspended, revoked).
    pub fn thread(&mut self, node: &str, limit: u16) -> Result<Vec<MailView>, MessengerError> {
        self.sync()?;
        if !self.registered.contains(node) {
            return Ok(Vec::new());
        }
        let operator = Address::Direct { participant: pid(OPERATOR)? };
        let target = Address::Direct { participant: pid(node)? };
        let to_op = self.engine.inbox(&operator, 0, 256).map_err(map_mail)?;
        let to_node = self.engine.inbox_of(&operator, &target, 0, 256).map_err(map_mail)?;
        let mut all: Vec<MailView> = to_op
            .messages
            .into_iter()
            .filter(|m| addr(&m.from) == node)
            .chain(to_node.messages.into_iter().filter(|m| addr(&m.from) == OPERATOR))
            .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 })
            .collect();
        all.sort_by_key(|m| (m.created_at_ms, m.message_id.clone()));
        let skip = all.len().saturating_sub(limit.clamp(1, 256) as usize);
        Ok(all.split_off(skip))
    }

    pub fn ack(&mut self, who: &str, message_id: &str) -> Result<(), MessengerError> {
        let reader = self.address(who)?;
        let id = MessageId::new(message_id).map_err(|e| MessengerError::Bad(e.to_string()))?;
        self.engine.ack(&reader, &id, unix_ms()).map(|_| ()).map_err(map_mail)
    }

    /// A session identity issued by the m4a signer becomes its own mailbox participant.
    pub fn ensure_session_participant(&mut self, nick: &str) -> Result<(), MessengerError> {
        if self.registered.contains(nick) {
            return Ok(());
        }
        let id = pid(nick)?;
        let _ = self.engine.register_participant(id, Some(nick.to_owned()), ParticipantPermissions { may_send: true, may_read: true, operator: false });
        self.registered.insert(nick.to_owned());
        Ok(())
    }

    pub fn is_registered(&mut self, who: &str) -> bool {
        self.sync().is_ok() && self.registered.contains(who)
    }
}

fn addr(a: &Address) -> String {
    match a {
        Address::Direct { participant } => participant.as_str().to_owned(),
        Address::Session { participant, session } => format!("{}/{}", participant.as_str(), session.as_str()),
        Address::Room { room } => format!("room:{}", room.as_str()),
    }
}

fn map_mail(error: mail4agent_api::MailError) -> MessengerError {
    use mail4agent_api::MailError as E;
    match error {
        E::PermissionDenied { .. } | E::UnknownParticipant { .. } | E::UnknownRoom { .. } => MessengerError::Forbidden,
        other => MessengerError::Bad(other.to_string()),
    }
}