Journaling: outside archives, and Journal it in mail flow rules

Phase 3 of the journaling spec.

- A journal's destination: builtIn (true for journals stored before) and
  archiveAddress, at least one. Reports to an archive are queued from the
  empty sender, one per address, flagged so they're never journaled.
- A pending record per report. When the queue lets go of one without
  delivering it (refused, expired, deleted), it becomes its own entry in
  the built-in journal under the sending journals' retention, the
  journal's archiveFailures (count, last time, reason) goes up, and the
  audit log records it; if that can't be written it stays queued.
- Journal it: a rule action naming a journal, on mail flow rules and
  beside a DLP rule's block, warn or hold. A journal whose scope chooses
  nobody takes only what rules send it.
- The report lists recipients a rule added or redirected to under
  "Added by rule", by rule name.
- A rule's route is cleared between messages in one SMTP session, with the
  new journal marks; a second message used to keep the first one's route.

tests/src/system/journal.rs: destination validation, a rule-only journal
fed by a rule that also adds a recipient, an unreachable archive's report
kept in the built-in journal with the failure counted, a report delivered
to an archive here and not journaled itself.

crates/smtp/src/queue/journal.rs

@@ -4,16 +4,22 @@
  * SPDX-License-Identifier: AGPL-3.0-only
  */
 
-//! inbuxa: journaling (journaling spec, JR-1 to JR-5, 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.
+//! 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};
+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},
     },
@@ -22,6 +28,15 @@ use inbuxa_features::{
 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).
@@ -35,6 +50,7 @@ pub async fn capture(
     queue_id: u64,
     message: &Message,
     raw: &[u8],
+    hints: &Hints,
 ) -> trc::Result<()> {
     if message.flags & (FROM_JOURNAL | FROM_REPORT) != 0 {
         return Ok(());
@@ -77,9 +93,10 @@ pub async fn capture(
         }
     }
     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))
+        .filter(|j| j.takes(direction, &members) || hints.marks.contains(&j.id))
         .collect();
     if taken.is_empty() {
         return Ok(());
@@ -98,6 +115,11 @@ pub async fn capture(
         .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 {
@@ -112,48 +134,147 @@ pub async fn capture(
     let host = server.core.network.server_name.as_str();
     let (bytes, fields) = report::build(&envelope, raw, &format!("postmaster@{host}"), host);
 
-    // The report's blob, reserved until the entry links it
-    let hash = BlobHash::generate(&bytes);
-    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?;
-
-    let retention_days = taken
-        .iter()
-        .map(|j| j.retention_days)
-        .max()
-        .unwrap_or_default();
     let mut tenants: Vec<u32> = members.iter().filter_map(|m| m.tenant).collect();
     tenants.sort_unstable();
     tenants.dedup();
-    let entry = 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,
-        message_id: fields.message_id,
-        accounts: members.iter().map(|m| m.account).collect(),
-        tenants,
-        journals: taken.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,
+    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,
+        }
     };
-    entries::append(server.store(), server.core.network.node_id, &entry).await?;
+
+    // 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
+}