/* * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC * * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ 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, types: Bitmap, tx: mpsc::Sender, }, Publish { notification: PushNotification, broadcast: bool, }, PushServerRegister { activate: Vec, expired: Vec, }, PushServerUpdate { account_id: u32, broadcast: bool, }, 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, pub uid: String, pub alert_id: String, } #[derive(Debug)] pub enum BroadcastEvent { PushNotification(PushNotification), PushServerUpdate(u32), RegistryChange(RegistryChange), CacheInvalidate(Vec), CacheInvalidateAll, CacheInvalidateNegative, MtaQueueStatus { is_running: bool }, } #[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), } #[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), Tls(Box), Stop, } #[derive(Debug)] pub struct DmarcEvent { pub domain: String, pub report_record: Record, pub dmarc_record: Arc, pub interval: AggregateFrequency, pub span_id: u64, } #[derive(Debug)] pub struct TlsEvent { pub domain: String, pub policy: PolicyType, pub failure: Option, pub tls_record: Arc, pub interval: AggregateFrequency, pub span_id: u64, } #[derive(Debug, Hash, PartialEq, Eq)] pub enum PolicyType { Tlsa(Option>), Sts(Option>), 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> { 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 for ReportingEvent { fn from(value: DmarcEvent) -> Self { ReportingEvent::Dmarc(Box::new(value)) } } impl From for ReportingEvent { fn from(value: TlsEvent) -> Self { ReportingEvent::Tls(Box::new(value)) } } impl From> for PolicyType { fn from(value: Arc) -> Self { PolicyType::Tlsa(Some(value)) } } impl From> for PolicyType { fn from(value: Arc) -> Self { PolicyType::Sts(Some(value)) } } impl From<&Arc> for PolicyType { fn from(value: &Arc) -> Self { PolicyType::Tlsa(Some(value.clone())) } } impl From<&Arc> for PolicyType { fn from(value: &Arc) -> Self { PolicyType::Sts(Some(value.clone())) } } impl From<(&Option>, &Option>)> for PolicyType { fn from(value: (&Option>, &Option>)) -> 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) -> Option { 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, ]), } } }