inbuxa-server/crates/smtp/src/queue/journal.rs

/*
 * SPDX-FileCopyrightText: 2026 Coffey Labs
 *
 * SPDX-License-Identifier: AGPL-3.0-only
 */

//! inbuxa: journaling (journaling spec, JR-1 to JR-11): the copy taken as a
//! message is queued, after DLP and transport rules, so it has the envelope
//! the message actually leaves or arrives with; reports to outside archives,
//! and what happens when an archive doesn't take one.

use crate::queue::{
    FROM_AUTHENTICATED, FROM_AUTOGENERATED, FROM_DSN, FROM_REPORT, Message, MessageSource, Status,
    spool::{QueueParams, SmtpSpool},
};
use common::Server;
use inbuxa_features::{
    audit::{Action, Actor, Outcome, Record, Target},
    hold::Member,
    journal::{
        self, Direction,
        archive::{self, Pending},
        entries::{self, Entry},
        report::{self, Envelope, Recipient},
    },
    mailflow::held::HOLD_SECONDS,
};
use store::write::{BatchBuilder, BlobLink, BlobOp, now};
use types::blob_hash::BlobHash;

/// What mail flow rules decided about a message at DATA (JR-3, JR-10).
#[derive(Debug, Clone, Default)]
pub struct Hints {
    /// Journals a rule sent it to.
    pub marks: Vec<u32>,
    /// Recipients a rule added (lowercase), and the rule's name.
    pub added: Vec<(String, String)>,
}

/// Marks a journal report the server queued itself, so it's never
/// journaled (JR-2). Free in the message flags (the MAIL parameters use
/// the low bits, the sources bits 32 to 37).
pub const FROM_JOURNAL: u64 = 1 << 48;

/// Journals `message`, whose queued bytes are `raw`, into every enabled
/// journal that takes it. An error means it may not have been journaled,
/// and the caller must not queue it.
pub async fn capture(
    server: &Server,
    queue_id: u64,
    message: &Message,
    raw: &[u8],
    hints: &Hints,
) -> trc::Result<()> {
    if message.flags & (FROM_JOURNAL | FROM_REPORT) != 0 {
        return Ok(());
    }
    let journals = journal::enabled(server.store()).await?;
    if journals.is_empty() {
        return Ok(());
    }

    // Who's here on either side, and which way it goes
    let mut members: Vec<Member> = Vec::new();
    let mut sender_local = message.flags & FROM_AUTHENTICATED != 0
        || (message.return_path.is_empty() && message.flags & (FROM_DSN | FROM_AUTOGENERATED) != 0);
    if !message.return_path.is_empty()
        && let Some(id) = server
            .account_id_from_email(&message.return_path, false)
            .await?
    {
        sender_local = true;
        if let Some(member) = server.member_of(id).await {
            members.push(member);
        }
    }
    let (mut any_local, mut any_remote) = (false, false);
    for rcpt in &message.recipients {
        let address = rcpt.address.to_lowercase();
        let domain = address.rsplit_once('@').map_or("", |(_, d)| d);
        let local_domain = server.domain(domain).await.ok().flatten().is_some();
        match server.account_id_from_email(&address, false).await? {
            Some(id) => {
                any_local = true;
                if !members.iter().any(|m| m.account == id)
                    && let Some(member) = server.member_of(id).await
                {
                    members.push(member);
                }
            }
            None if local_domain => any_local = true,
            None => any_remote = true,
        }
    }
    let direction = Direction::of(sender_local, any_remote, any_local);
    // A journal takes it through its scope, or because a rule sent it there
    let taken: Vec<&journal::Journal> = journals
        .iter()
        .filter(|j| j.takes(direction, &members) || hints.marks.contains(&j.id))
        .collect();
    if taken.is_empty() {
        return Ok(());
    }

    // DLP holds a message by putting its release a century off
    let at = now();
    let held = !message.recipients.is_empty()
        && message
            .recipients
            .iter()
            .all(|rcpt| rcpt.retry.due >= at + HOLD_SECONDS / 2);
    let recipients: Vec<Recipient> = message
        .recipients
        .iter()
        .map(|rcpt| Recipient {
            address: rcpt.address.to_string(),
            orcpt: rcpt.orcpt.as_deref().map(Into::into),
            added_by: hints
                .added
                .iter()
                .find(|(address, _)| address.eq_ignore_ascii_case(&rcpt.address))
                .map(|(_, rule)| rule.clone()),
        })
        .collect();
    let envelope = Envelope {
        sender: &message.return_path,
        authenticated: message.flags & FROM_AUTHENTICATED != 0,
        recipients: &recipients,
        queue_id,
        received: message.created,
        direction,
        held,
    };
    let host = server.core.network.server_name.as_str();
    let (bytes, fields) = report::build(&envelope, raw, &format!("postmaster@{host}"), host);

    let mut tenants: Vec<u32> = members.iter().filter_map(|m| m.tenant).collect();
    tenants.sort_unstable();
    tenants.dedup();
    let hash = BlobHash::generate(&bytes);
    let entry_for = |journals: &[&journal::Journal]| {
        let retention_days = journals
            .iter()
            .map(|j| j.retention_days)
            .max()
            .unwrap_or_default();
        Entry {
            queue_id,
            at,
            direction,
            sender: message.return_path.to_string(),
            authenticated: envelope.authenticated,
            recipients: recipients.iter().map(|r| r.address.clone()).collect(),
            subject: fields.subject.clone(),
            message_id: fields.message_id.clone(),
            accounts: members.iter().map(|m| m.account).collect(),
            tenants: tenants.clone(),
            journals: journals.iter().map(|j| j.id).collect(),
            held,
            blob: entries::hex(hash.as_slice()),
            size: bytes.len() as u64,
            sha256: entries::sha256(&bytes),
            expires_at: at + u64::from(retention_days) * 86_400,
        }
    };

    // The built-in journal: one entry, however many journals keep it there
    let built_in: Vec<&journal::Journal> = taken.iter().copied().filter(|j| j.built_in).collect();
    if !built_in.is_empty() {
        // The report's blob, reserved until the entry links it
        let mut batch = BatchBuilder::new();
        batch.set(
            BlobOp::Link {
                hash: hash.clone(),
                to: BlobLink::Temporary { until: at + 120 },
            },
            vec![],
        );
        server.store().write(batch.build_all()).await?;
        server
            .blob_store()
            .put_blob(hash.as_slice(), &bytes, server.core.email.compression)
            .await?;
        entries::append(
            server.store(),
            server.core.network.node_id,
            &entry_for(&built_in),
        )
        .await?;
    }

    // Outside archives: one report per address (JR-4, JR-7)
    let mut addresses: Vec<(String, Vec<&journal::Journal>)> = Vec::new();
    for journal in &taken {
        if let Some(address) = &journal.archive_address {
            let address = address.to_lowercase();
            match addresses.iter_mut().find(|(a, _)| *a == address) {
                Some((_, journals)) => journals.push(journal),
                None => addresses.push((address, vec![journal])),
            }
        }
    }
    for (address, journals) in addresses {
        // From nobody: an archive's refusal comes back to no one, and the
        // queue's own record of it is what counts (settle, below)
        let mut report = server.new_message("", MessageSource::Autogenerated, 0);
        report.message.flags |= FROM_JOURNAL;
        report.add_expanded_recipient(&address, server).await;
        let pending = Pending {
            address,
            entry: entry_for(&journals),
        };
        archive::set_pending(server.store(), report.queue_id, &pending).await?;
        let report_id = report.queue_id;
        // Boxed: queueing the report comes back through this function
        let queued = Box::pin(report.queue(QueueParams::new(&bytes, 0, server))).await;
        if !queued {
            archive::clear_pending(server.store(), report_id).await?;
            return Err(trc::StoreEvent::UnexpectedError
                .into_err()
                .details("Failed to queue a journal report"));
        }
    }
    Ok(())
}

/// JR-7: a journal report is leaving the queue. Delivered, its pending
/// record goes; not delivered (refused, expired, or deleted from the
/// queue), it goes into the built-in journal instead, and the journals
/// that sent it count a failure. An error means nothing was settled, and
/// the report must stay queued.
pub async fn settle(server: &Server, queue_id: u64, message: &Message) -> trc::Result<()> {
    let store = server.store();
    let Some(pending) = archive::pending(store, queue_id).await? else {
        return Ok(());
    };
    let delivered = !message.recipients.is_empty()
        && message
            .recipients
            .iter()
            .all(|rcpt| matches!(rcpt.status, Status::Completed(_)));
    if !delivered {
        let reason = if message
            .recipients
            .iter()
            .any(|rcpt| matches!(rcpt.status, Status::PermanentFailure(_)))
        {
            "the archive refused it"
        } else {
            "it wasn't delivered before leaving the queue"
        };
        entries::append(store, server.core.network.node_id, &pending.entry).await?;
        let at = now();
        archive::record_failure(store, &pending.entry.journals, at, reason).await?;
        server
            .audit_note(Record {
                at: at * 1000,
                actor: Actor::system("Journal"),
                via: None,
                remote_ip: None,
                action: Action::Create,
                target: Target {
                    kind: "inbuxa:JournalEntry".into(),
                    id: Some(format!("{:x}", pending.entry.queue_id)),
                    name: None,
                    account_id: None,
                    tenant_id: None,
                },
                changes: vec![],
                details: Some(format!(
                    "A journal report to {} wasn't delivered ({reason}); kept in the built-in journal",
                    pending.address
                )),
                reason: None,
                outcome: Outcome::success(),
            })
            .await;
    }
    archive::clear_pending(store, queue_id).await
}