inbuxa-server/crates/common/src/ipc.rs

/*
 * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <[email protected]>
 *
 * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
 *
 * Modified by Coffey Labs in 2026 for INBUXA.
 */

use crate::config::smtp::{
    queue::QueueName,
    report::AggregateFrequency,
    resolver::{Policy, Tlsa},
};
use ahash::RandomState;
use mail_auth::{
    dmarc::Dmarc,
    mta_sts::TlsRpt,
    report::{Record, tlsrpt::FailureDetails},
};
use registry::{schema::prelude::ObjectType, types::id::ObjectId};
use std::sync::{
    Arc,
    atomic::{AtomicBool, Ordering},
};
use tokio::sync::{Semaphore, SemaphorePermit, mpsc};
use types::type_state::{DataType, StateChange};
use utils::map::bitmap::Bitmap;

#[derive(Debug)]
pub enum PushEvent {
    Subscribe {
        account_ids: Vec<u32>,
        types: Bitmap<DataType>,
        tx: mpsc::Sender<PushNotification>,
    },
    Publish {
        notification: PushNotification,
        broadcast: bool,
    },
    PushServerRegister {
        activate: Vec<u32>,
        expired: Vec<u32>,
    },
    PushServerUpdate {
        account_id: u32,
        broadcast: bool,
    },
    // inbuxa: SCIM-52: ends the push subscriptions the account itself holds
    // (IMAP IDLE, JMAP event streams and WebSockets) on this node
    Revoke {
        account_id: u32,
    },
    Stop,
}

#[derive(Debug, Clone)]
pub enum PushNotification {
    StateChange(StateChange),
    CalendarAlert(CalendarAlert),
    EmailPush(EmailPush),
}

#[derive(Debug, Clone)]
pub struct EmailPush {
    pub account_id: u32,
    pub email_id: u32,
    pub change_id: u64,
}

#[derive(Debug, Clone)]
pub struct CalendarAlert {
    pub account_id: u32,
    pub event_id: u32,
    pub recurrence_id: Option<i64>,
    pub uid: String,
    pub alert_id: String,
}

#[derive(Debug)]
pub enum BroadcastEvent {
    PushNotification(PushNotification),
    PushServerUpdate(u32),
    RegistryChange(RegistryChange),
    CacheInvalidate(Vec<CacheInvalidation>),
    CacheInvalidateAll,
    CacheInvalidateNegative,
    MtaQueueStatus { is_running: bool },
    QueueRefresh,
    // inbuxa: AL-3: end an account's open sessions on every node
    EndSessions(u32),
    // inbuxa: deliverability spec, DL-15: every node checks itself now
    DeliverabilityCheck,
}

#[derive(Debug, Clone, Copy)]
pub enum RegistryChange {
    Insert(ObjectId),
    Delete(ObjectId),
    Reload(ObjectType),
}

#[derive(Debug, Clone, Copy, Hash, PartialEq, Eq)]
pub enum CacheInvalidation {
    AccessToken(u32),
    DavResources(u32),
    Domain(u32),
    Account(u32),
    DkimSignature(u32),
    Tenant(u32),
    Role(u32),
    List(u32),
    DomainLogo(u32),
    TenantLogo(u32),
    EmailNegative {
        domain_id: u32,
        local_part_hash: u32,
    },
    DomainNegative,
}

#[derive(Debug)]
pub enum QueueEvent {
    Refresh,
    WorkerDone {
        queue_id: u64,
        queue_name: QueueName,
        status: QueueEventStatus,
    },
    Paused(bool),
    ReloadSettings,
    Stop,
}

#[derive(Debug)]
pub enum QueueEventStatus {
    Completed,
    Locked,
    Deferred,
}

#[derive(Debug)]
pub enum ReportingEvent {
    Dmarc(Box<DmarcEvent>),
    Tls(Box<TlsEvent>),
    Stop,
}

#[derive(Debug)]
pub struct DmarcEvent {
    pub domain: String,
    pub report_record: Record,
    pub dmarc_record: Arc<Dmarc>,
    pub interval: AggregateFrequency,
    pub span_id: u64,
}

#[derive(Debug)]
pub struct TlsEvent {
    pub domain: String,
    pub policy: PolicyType,
    pub failure: Option<FailureDetails>,
    pub tls_record: Arc<TlsRpt>,
    pub interval: AggregateFrequency,
    pub span_id: u64,
}

#[derive(Debug, Hash, PartialEq, Eq)]
pub enum PolicyType {
    Tlsa(Option<Arc<Tlsa>>),
    Sts(Option<Arc<Policy>>),
    None,
}

pub struct TrainTaskController {
    semaphore: Semaphore,
    stop_flag: AtomicBool,
}

impl Default for TrainTaskController {
    fn default() -> Self {
        Self {
            semaphore: Semaphore::new(1),
            stop_flag: AtomicBool::new(false),
        }
    }
}

impl TrainTaskController {
    pub fn try_run(&self) -> Option<SemaphorePermit<'_>> {
        let permit = self.semaphore.try_acquire().ok()?;

        self.stop_flag.store(false, Ordering::SeqCst);

        Some(permit)
    }

    pub fn is_running(&self) -> bool {
        self.semaphore.available_permits() == 0
    }

    pub fn stop(&self) {
        self.stop_flag.store(true, Ordering::SeqCst);
    }

    pub fn should_stop(&self) -> bool {
        self.stop_flag.load(Ordering::SeqCst)
    }
}

impl BroadcastEvent {
    pub fn reload(object: ObjectType) -> Self {
        BroadcastEvent::RegistryChange(RegistryChange::Reload(object))
    }
}

pub trait ToHash {
    fn to_hash(&self) -> u64;
}

impl ToHash for Dmarc {
    fn to_hash(&self) -> u64 {
        RandomState::with_seeds(1, 9, 7, 9).hash_one(self)
    }
}

impl ToHash for PolicyType {
    fn to_hash(&self) -> u64 {
        RandomState::with_seeds(1, 9, 7, 9).hash_one(self)
    }
}

impl From<DmarcEvent> for ReportingEvent {
    fn from(value: DmarcEvent) -> Self {
        ReportingEvent::Dmarc(Box::new(value))
    }
}

impl From<TlsEvent> for ReportingEvent {
    fn from(value: TlsEvent) -> Self {
        ReportingEvent::Tls(Box::new(value))
    }
}

impl From<Arc<Tlsa>> for PolicyType {
    fn from(value: Arc<Tlsa>) -> Self {
        PolicyType::Tlsa(Some(value))
    }
}

impl From<Arc<Policy>> for PolicyType {
    fn from(value: Arc<Policy>) -> Self {
        PolicyType::Sts(Some(value))
    }
}

impl From<&Arc<Tlsa>> for PolicyType {
    fn from(value: &Arc<Tlsa>) -> Self {
        PolicyType::Tlsa(Some(value.clone()))
    }
}

impl From<&Arc<Policy>> for PolicyType {
    fn from(value: &Arc<Policy>) -> Self {
        PolicyType::Sts(Some(value.clone()))
    }
}

impl From<(&Option<Arc<Policy>>, &Option<Arc<Tlsa>>)> for PolicyType {
    fn from(value: (&Option<Arc<Policy>>, &Option<Arc<Tlsa>>)) -> Self {
        match value {
            (Some(value), _) => PolicyType::Sts(Some(value.clone())),
            (_, Some(value)) => PolicyType::Tlsa(Some(value.clone())),
            _ => PolicyType::None,
        }
    }
}

impl PushNotification {
    pub fn account_id(&self) -> u32 {
        match self {
            PushNotification::StateChange(state_change) => state_change.account_id,
            PushNotification::CalendarAlert(calendar_alert) => calendar_alert.account_id,
            PushNotification::EmailPush(email_push) => email_push.account_id,
        }
    }

    pub fn filter_types(&self, types: &Bitmap<DataType>) -> Option<PushNotification> {
        match self {
            PushNotification::StateChange(state_change) => {
                let mut filtered_types = state_change.types;
                filtered_types.intersection(types);
                if !filtered_types.is_empty() {
                    Some(PushNotification::StateChange(StateChange {
                        account_id: state_change.account_id,
                        change_id: state_change.change_id,
                        types: filtered_types,
                    }))
                } else {
                    None
                }
            }
            PushNotification::CalendarAlert(_) => {
                if types.contains(DataType::CalendarAlert) {
                    Some(self.clone())
                } else {
                    None
                }
            }
            PushNotification::EmailPush(_) => {
                if types.contains_any(
                    [
                        DataType::EmailDelivery,
                        DataType::Email,
                        DataType::Mailbox,
                        DataType::Thread,
                    ]
                    .into_iter(),
                ) {
                    Some(self.clone())
                } else {
                    None
                }
            }
        }
    }
}

impl EmailPush {
    pub fn to_state_change(&self) -> StateChange {
        StateChange {
            account_id: self.account_id,
            change_id: self.change_id,
            types: Bitmap::from_iter([
                DataType::EmailDelivery,
                DataType::Email,
                DataType::Mailbox,
                DataType::Thread,
            ]),
        }
    }
}

/// inbuxa: the task locks this node holds, so a graceful stop can hand them
/// back instead of leaving the tasks blocked until the locks expire.
pub struct TaskLocks {
    held: parking_lot::Mutex<ahash::AHashSet<u64>>,
    stopping: AtomicBool,
    expiry: std::sync::atomic::AtomicU64,
}

impl TaskLocks {
    /// How long a task lock lasts, in seconds, unless it is released first
    /// or renewed. inbuxa: upstream held a lock for an hour, so a killed
    /// node's tasks waited that long; the lock is now a five-minute lease
    /// that the task manager renews every third of it while the task runs
    /// (renew_task_locks), so a dead node's tasks run elsewhere within
    /// minutes.
    pub const DEFAULT_EXPIRY: u64 = 5 * 60;

    pub fn is_stopping(&self) -> bool {
        self.stopping.load(Ordering::Acquire)
    }

    /// Stops new claims and returns the ids of every lock still held.
    pub fn stop(&self) -> Vec<u64> {
        self.stopping.store(true, Ordering::Release);
        self.held.lock().drain().collect()
    }

    pub fn insert(&self, id: u64) {
        self.held.lock().insert(id);
    }

    pub fn remove(&self, id: u64) {
        self.held.lock().remove(&id);
    }

    pub fn held(&self) -> usize {
        self.held.lock().len()
    }

    /// inbuxa: the tasks this node holds, to renew their locks.
    pub fn held_ids(&self) -> Vec<u64> {
        self.held.lock().iter().copied().collect()
    }

    /// inbuxa: whether this node holds (and is running) the task.
    pub fn is_held(&self, id: u64) -> bool {
        self.held.lock().contains(&id)
    }

    pub fn expiry(&self) -> u64 {
        self.expiry.load(Ordering::Relaxed)
    }

    /// Changes the lock lifetime; the tests shorten it.
    pub fn set_expiry(&self, seconds: u64) {
        self.expiry.store(seconds.max(1), Ordering::Relaxed);
    }
}

impl Default for TaskLocks {
    fn default() -> Self {
        Self {
            held: Default::default(),
            stopping: AtomicBool::new(false),
            expiry: std::sync::atomic::AtomicU64::new(Self::DEFAULT_EXPIRY),
        }
    }
}