use std::collections::HashMap;
use std::pin;
use std::time::Duration;
use async_imap::types::NameAttribute;
use serde_json::Value;
use tokio::sync::mpsc;
use tokio::time::{MissedTickBehavior, interval, timeout};
use tokio_stream::StreamExt;
use crate::channels::{InboundEvent, MessagingTask};
use crate::types::MessagingSubscriptionError;
use super::connection::{ImapConnection, ImapSession, email_protocol, email_timeout};
use super::helpers::normalize_update;
const POLL_INTERVAL: Duration = Duration::from_secs(30);
const IO_TIMEOUT: Duration = Duration::from_secs(60);
#[derive(Clone, Copy)]
struct MailboxCursor {
validity: Option<u32>,
uid: u32,
}
pub(super) async fn listen(
connection: ImapConnection,
tx: mpsc::Sender<InboundEvent>,
) -> Result<MessagingTask, MessagingSubscriptionError> {
Ok(tokio::spawn(async move {
let mut cursors = HashMap::<String, MailboxCursor>::new();
let mut ticks = interval(POLL_INTERVAL);
ticks.set_missed_tick_behavior(MissedTickBehavior::Skip);
loop {
tokio::select! { _ = tx.closed() => return Ok(()), _ = ticks.tick() => {} }
let mut slot = connection.lock().await;
let session = slot
.as_mut()
.ok_or_else(|| email_protocol("IMAP connection closed".to_owned()))?;
let result = tokio::select! {
_ = tx.closed() => return Ok(()),
result = scan_mailboxes(session, &mut cursors, &tx) => result,
};
if result.is_err() {
*slot = None;
}
result?;
}
}))
}
async fn scan_mailboxes(
session: &mut ImapSession,
cursors: &mut HashMap<String, MailboxCursor>,
tx: &mpsc::Sender<InboundEvent>,
) -> Result<(), MessagingSubscriptionError> {
let folders = timeout(IO_TIMEOUT, selectable_folders(session))
.await
.map_err(|_| email_timeout("imap_list", IO_TIMEOUT))??;
cursors.retain(|folder, _| folders.contains(folder));
for folder in folders {
let mailbox = timeout(IO_TIMEOUT, session.examine(&folder))
.await
.map_err(|_| email_timeout("imap_examine", IO_TIMEOUT))?
.map_err(|error| email_protocol(format!("IMAP EXAMINE failed: {error}")))?;
let initial = MailboxCursor {
validity: mailbox.uid_validity,
uid: mailbox.uid_next.unwrap_or(1).saturating_sub(1),
};
let cursor = cursors.entry(folder.clone()).or_insert(initial);
if cursor.validity != mailbox.uid_validity {
*cursor = initial;
}
let Some(first) = cursor.uid.checked_add(1) else {
continue;
};
let Some(next) = mailbox.uid_next else {
continue;
};
if first >= next {
continue;
}
let previous_uid = cursor.uid;
let range = format!("{first}:{}", next - 1);
let mut fetches = timeout(IO_TIMEOUT, session.uid_fetch(range, "(UID BODY.PEEK[])"))
.await
.map_err(|_| email_timeout("imap_uid_fetch", IO_TIMEOUT))?
.map_err(|error| email_protocol(format!("IMAP UID FETCH failed: {error}")))?;
while let Some(fetch) = timeout(IO_TIMEOUT, fetches.next())
.await
.map_err(|_| email_timeout("imap_fetch", IO_TIMEOUT))?
{
let fetch = fetch.map_err(|error| email_protocol(format!("IMAP FETCH stream failed: {error}")))?;
let Some(uid) = fetch.uid.filter(|uid| *uid > previous_uid) else {
continue;
};
cursor.uid = cursor.uid.max(uid);
if let Some(raw) = fetch.body() {
let mut normalized = normalize_update(raw);
if let Some(message) = normalized.as_mut() {
message.metadata["channel_id"] = Value::String(folder.clone());
}
if tx
.send(InboundEvent {
raw_payload: raw.to_vec(),
normalized,
})
.await
.is_err()
{
return Ok(());
}
}
}
}
Ok(())
}
async fn selectable_folders(session: &mut ImapSession) -> Result<Vec<String>, MessagingSubscriptionError> {
let names = session
.list(Some(""), Some("*"))
.await
.map_err(|error| email_protocol(format!("IMAP LIST failed: {error}")))?;
let mut names = pin::pin!(names);
let mut folders = Vec::new();
while let Some(name) = names.next().await {
let name = name.map_err(|error| email_protocol(format!("IMAP LIST stream failed: {error}")))?;
if !name.attributes().contains(&NameAttribute::NoSelect) {
folders.push(name.name().to_owned());
}
}
Ok(folders)
}