//! Integration tests against Stalwart Mail Server.
//!
//! Stalwart is a Rust-based server with full IMAP4rev2 (RFC 9051) support,
//! making it ideal for testing rev2-specific features that Dovecot and
//! GreenMail don't cover.
//!
//! # Setup
//!
//! ```sh
//! docker compose up -d
//! ./scripts/setup-stalwart.sh
//! cargo test --test stalwart -- --ignored --test-threads=1
//! docker compose down
//! ```

#![allow(clippy::unwrap_used, clippy::expect_used, clippy::doc_markdown)]

use std::collections::{BTreeMap, BTreeSet};
use std::sync::{Arc, Once, OnceLock};
use std::time::{Duration, Instant};

use daaki_imap::IdleEvent;
use tokio::sync::Mutex;
use tokio_util::sync::CancellationToken;

use daaki_imap::types::{
    Capability, FetchAttr, Flag, MailboxAttribute, QuotaResource, SequenceSet,
};
use daaki_imap::{ImapConnection, SessionState, TlsMode};

static TRACING_INIT: Once = Once::new();

fn init_tracing() {
    TRACING_INIT.call_once(|| {
        tracing_subscriber::fmt()
            .with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
            .with_test_writer()
            .try_init()
            .ok();
    });
}

const TIMEOUT: Duration = Duration::from_secs(10);

/// Minimal RFC 5322 message for APPEND tests.
const TEST_MESSAGE: &[u8] = b"\
From: sender@example.com\r\n\
To: recipient@example.com\r\n\
Subject: Stalwart integration test\r\n\
Date: Sun, 15 Mar 2026 00:00:00 +0000\r\n\
Message-ID: <stalwart-test-1@daaki>\r\n\
MIME-Version: 1.0\r\n\
Content-Type: text/plain; charset=utf-8\r\n\
\r\n\
Hello from daaki Stalwart integration tests.\r\n";

mod stalwart {
    pub const HOST: &str = "127.0.0.1";
    pub const IMAP_PORT: u16 = 12143;
    /// Stalwart's API-created users have auth issues, so we use the admin
    /// account which is auto-generated on first start. The password is
    /// extracted from the container logs by the setup script and written
    /// to a file that we read here.
    pub fn credentials() -> (String, String) {
        // Try env var first (CI), then file (local).
        if let Ok(pass) = std::env::var("STALWART_ADMIN_PASS") {
            return ("admin".to_owned(), pass);
        }
        // The setup script writes the password to this file.
        let mut path = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"));
        path.pop();
        path.pop();
        path.push(".stalwart-admin-pass");
        match std::fs::read_to_string(&path) {
            Ok(pass) => ("admin".to_owned(), pass.trim().to_owned()),
            Err(e) => {
                panic!("Stalwart admin password not found ({e}). Run ./scripts/setup-stalwart.sh first.");
            }
        }
    }
}

// ==========================================================================
// Helpers
// ==========================================================================

/// Build a TLS config that accepts any certificate (for Docker self-signed certs).
fn build_danger_tls_config() -> Arc<rustls::ClientConfig> {
    mod danger {
        #[derive(Debug)]
        pub struct AcceptAll;

        impl rustls::client::danger::ServerCertVerifier for AcceptAll {
            fn verify_server_cert(
                &self,
                _end_entity: &rustls_pki_types::CertificateDer<'_>,
                _intermediates: &[rustls_pki_types::CertificateDer<'_>],
                _server_name: &rustls_pki_types::ServerName<'_>,
                _ocsp_response: &[u8],
                _now: rustls_pki_types::UnixTime,
            ) -> Result<rustls::client::danger::ServerCertVerified, rustls::Error> {
                Ok(rustls::client::danger::ServerCertVerified::assertion())
            }

            fn verify_tls12_signature(
                &self,
                _message: &[u8],
                _cert: &rustls_pki_types::CertificateDer<'_>,
                _dss: &rustls::DigitallySignedStruct,
            ) -> Result<rustls::client::danger::HandshakeSignatureValid, rustls::Error>
            {
                Ok(rustls::client::danger::HandshakeSignatureValid::assertion())
            }

            fn verify_tls13_signature(
                &self,
                _message: &[u8],
                _cert: &rustls_pki_types::CertificateDer<'_>,
                _dss: &rustls::DigitallySignedStruct,
            ) -> Result<rustls::client::danger::HandshakeSignatureValid, rustls::Error>
            {
                Ok(rustls::client::danger::HandshakeSignatureValid::assertion())
            }

            fn supported_verify_schemes(&self) -> Vec<rustls::SignatureScheme> {
                vec![
                    rustls::SignatureScheme::RSA_PKCS1_SHA256,
                    rustls::SignatureScheme::RSA_PKCS1_SHA384,
                    rustls::SignatureScheme::RSA_PKCS1_SHA512,
                    rustls::SignatureScheme::ECDSA_NISTP256_SHA256,
                    rustls::SignatureScheme::ECDSA_NISTP384_SHA384,
                    rustls::SignatureScheme::RSA_PSS_SHA256,
                    rustls::SignatureScheme::RSA_PSS_SHA384,
                    rustls::SignatureScheme::RSA_PSS_SHA512,
                    rustls::SignatureScheme::ED25519,
                ]
            }
        }
    }

    let _ = rustls::crypto::ring::default_provider().install_default();

    let config = rustls::ClientConfig::builder()
        .dangerous()
        .with_custom_certificate_verifier(Arc::new(danger::AcceptAll))
        .with_no_client_auth();
    Arc::new(config)
}

/// Connect to Stalwart, negotiate STARTTLS, and return a session in the Not
/// Authenticated state.
///
/// Stalwart doesn't support the IMAP LOGIN command — only SASL mechanisms
/// (AUTH=PLAIN, AUTH=OAUTHBEARER, AUTH=XOAUTH2). Auth-specific tests use this
/// helper to begin from the RFC 3501 Section 3 Not Authenticated state.
async fn connect_stalwart_not_authenticated() -> ImapConnection {
    init_tracing();
    let tls_config = build_danger_tls_config();
    let conn = ImapConnection::connect_with_tls_config(
        stalwart::HOST,
        stalwart::IMAP_PORT,
        TlsMode::None,
        tls_config.clone(),
        TIMEOUT,
    )
    .await
    .unwrap();
    conn.starttls_with_config(tls_config, TIMEOUT)
        .await
        .unwrap();
    conn
}

/// Connect and authenticate against Stalwart via STARTTLS + AUTHENTICATE PLAIN.
///
/// Stalwart doesn't support the IMAP LOGIN command — only SASL mechanisms
/// (AUTH=PLAIN, AUTH=OAUTHBEARER, AUTH=XOAUTH2). So we use `authenticate_plain`.
async fn connect_stalwart() -> ImapConnection {
    let (user, pass) = stalwart::credentials();
    let conn = connect_stalwart_not_authenticated().await;
    conn.authenticate_plain(&user, &pass, TIMEOUT)
        .await
        .unwrap();
    conn
}

/// Helper: check if server advertises a specific capability.
fn has_capability(conn: &ImapConnection, cap: &Capability) -> bool {
    conn.capabilities().iter().any(|c| c == cap)
}

/// Serialize CREATE-SPECIAL-USE tests so they don't race on Stalwart's
/// one-mailbox-per-role constraint. RFC 6154 Section 2.
fn special_use_role_lock() -> &'static Mutex<()> {
    static LOCK: OnceLock<Mutex<()>> = OnceLock::new();
    LOCK.get_or_init(|| Mutex::new(()))
}

/// Pick an RFC 6154 special-use role that is not currently assigned so each
/// CREATE-SPECIAL-USE test remains independent from leftover server state.
async fn available_special_use_role(conn: &ImapConnection) -> MailboxAttribute {
    let listing = conn
        .list_extended("", &["*"], &["SPECIAL-USE"], &[], TIMEOUT)
        .await
        .unwrap();

    [
        MailboxAttribute::Archive,
        MailboxAttribute::Drafts,
        MailboxAttribute::Flagged,
        MailboxAttribute::Junk,
        MailboxAttribute::Sent,
        MailboxAttribute::Trash,
    ]
    .into_iter()
    .find(|candidate| {
        listing.iter().all(|mailbox| {
            mailbox
                .attributes
                .iter()
                .all(|attribute| attribute != candidate)
        })
    })
    .expect(
        "Stalwart test environment must leave at least one RFC 6154 special-use role unassigned",
    )
}

/// Discover the server's hierarchy delimiter via LIST (RFC 3501 Section 6.3.8).
async fn hierarchy_delimiter(conn: &ImapConnection) -> char {
    conn.list("", "INBOX", TIMEOUT)
        .await
        .unwrap()
        .into_iter()
        .find_map(|info| info.delimiter)
        .unwrap_or('/')
}

/// Return whether the server advertises the named quota resource.
fn has_quota_resource(conn: &ImapConnection, resource: &str) -> bool {
    conn.capabilities().iter().any(|cap| match cap {
        Capability::QuotaResource(name) => name.eq_ignore_ascii_case(resource),
        _ => false,
    })
}

/// Convert a QUOTA response into a comparable resource-limit map.
///
/// RFC 9208 Sections 4.1.1, 4.1.2, and 4.2.1 model QUOTA as named
/// `(usage, limit)` resource triplets for one quota root. Separate commands
/// can observe different live usage, so compare the stable resource names and
/// limits rather than transient usage counters.
fn quota_resource_limit_map(resources: &[QuotaResource]) -> BTreeMap<String, u64> {
    resources
        .iter()
        .map(|resource| (resource.name.clone(), resource.limit))
        .collect()
}

/// Assert that a selected mailbox remains selected and usable after a command.
///
/// RFC 3501 Section 6.4 requires commands valid in selected state to leave the
/// mailbox selected for follow-up commands.
async fn assert_selected_mailbox_still_usable(conn: &ImapConnection, command: &str) {
    assert_eq!(
        conn.session_state(),
        SessionState::Selected,
        "{command} must not leave selected state"
    );
    assert_eq!(
        conn.search("ALL", TIMEOUT).await.unwrap().ids,
        vec![1],
        "selected mailbox must remain usable after {command}"
    );
}

/// Build a minimal RFC 5322 message for APPEND-based integration tests.
///
/// RFC 5322 Section 2.2 defines header fields as individual lines, and
/// leading whitespace on a later line folds that line into the previous field.
/// RFC 5322 Section 3.6 defines the originator and identification fields used
/// here (`From`, `To`, `Subject`, `Date`, and `Message-ID`).
fn build_message(subject: &str, message_id: &str, body: &str) -> Vec<u8> {
    let mut message = String::new();
    message.push_str("From: sender@example.com\r\n");
    message.push_str("To: recipient@example.com\r\n");
    message.push_str("Subject: ");
    message.push_str(subject);
    message.push_str("\r\n");
    message.push_str("Date: Sun, 15 Mar 2026 00:00:00 +0000\r\n");
    message.push_str("Message-ID: <");
    message.push_str(message_id);
    message.push_str(">\r\n");
    message.push_str("MIME-Version: 1.0\r\n");
    message.push_str("Content-Type: text/plain; charset=utf-8\r\n");
    // RFC 5322 Section 2.2: a blank line terminates the header block.
    message.push_str("\r\n");
    message.push_str(body);
    message.push_str("\r\n");
    message.into_bytes()
}

/// Build a mailbox name unique across concurrent integration runs.
fn unique_name(prefix: &str) -> String {
    use std::sync::atomic::{AtomicU64, Ordering};

    static COUNTER: AtomicU64 = AtomicU64::new(0);

    let n = COUNTER.fetch_add(1, Ordering::SeqCst);
    let nanos = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_nanos())
        .unwrap_or(0);
    format!("{prefix}-{}-{nanos}-{n}", std::process::id())
}

/// APPEND a deterministic batch of messages to a mailbox.
///
/// RFC 3501 Section 6.3.11 defines APPEND as adding new RFC 5322 messages to
/// the target mailbox in arrival order.
async fn append_messages(
    conn: &ImapConnection,
    mailbox: &str,
    subject_prefix: &str,
    message_id_prefix: &str,
    count: usize,
) {
    for i in 1..=count {
        let message = build_message(
            &format!("{subject_prefix} {i}"),
            &format!("{message_id_prefix}-{i}-{}@daaki", std::process::id()),
            &format!("{subject_prefix} body {i}"),
        );
        conn.append(mailbox, &[], None, &message, TIMEOUT)
            .await
            .unwrap();
    }
}

/// SELECT a mailbox with CONDSTORE and return the resulting snapshot.
///
/// RFC 7162 Section 3.1.2.1 requires `HIGHESTMODSEQ` in successful `SELECT`
/// responses that request CONDSTORE.
async fn select_condstore(
    conn: &ImapConnection,
    mailbox: &str,
) -> daaki_imap::types::SelectedMailbox {
    conn.select_with(
        mailbox,
        &daaki_imap::types::SelectOptions::condstore(),
        TIMEOUT,
    )
    .await
    .unwrap()
}

/// ENABLE CONDSTORE and QRESYNC when the server advertises both extensions.
///
/// RFC 7162 Section 3.2.5 requires QRESYNC to be ENABLEd before use, and
/// RFC 7162 Section 3.1 makes QRESYNC depend on CONDSTORE support.
async fn enable_condstore_and_qresync(conn: &ImapConnection) -> bool {
    if !has_capability(conn, &Capability::Condstore) || !has_capability(conn, &Capability::QResync)
    {
        eprintln!("skipping: Stalwart does not advertise CONDSTORE/QRESYNC");
        return false;
    }

    let enabled = conn
        .enable(&["CONDSTORE", "QRESYNC"], TIMEOUT)
        .await
        .unwrap();
    assert!(
        enabled
            .iter()
            .any(|cap| cap.eq_ignore_ascii_case("QRESYNC")),
        "ENABLE must report QRESYNC before UID FETCH VANISHED: {enabled:?}"
    );
    true
}

/// Return the sorted UID set of the currently selected mailbox.
///
/// RFC 3501 Section 6.4.8 applies `UID` to SEARCH while preserving SEARCH's
/// selection semantics from RFC 3501 Section 6.4.4.
async fn sorted_uids(conn: &ImapConnection) -> Vec<u32> {
    let mut uids = conn.uid_search("ALL", TIMEOUT).await.unwrap().ids;
    uids.sort_unstable();
    uids
}

/// Append one seed message, delete it, and expunge it so later APPENDs receive
/// UIDs that no longer match the surviving messages' sequence numbers.
///
/// RFC 3501 Section 2.3.1.1 requires UIDs to increase monotonically, and
/// RFC 3501 Section 6.4.3 permanently removes the expunged message while
/// leaving later-assigned UIDs unchanged.
async fn create_uid_gap(conn: &ImapConnection, mailbox: &str) {
    let seed_message = build_message(
        "Stalwart UID gap seed",
        &format!("{mailbox}-uid-gap-seed-{}@daaki", std::process::id()),
        "Seed message for creating a UID gap",
    );
    conn.append(mailbox, &[], None, &seed_message, TIMEOUT)
        .await
        .unwrap();

    let seed_uid = conn
        .uid_fetch(
            &SequenceSet::new("1:*").unwrap(),
            &[FetchAttr::Uid],
            TIMEOUT,
        )
        .await
        .unwrap()
        .into_iter()
        .last()
        .and_then(|fetch| fetch.uid)
        .expect("UID gap seed fetch must include the appended message UID");

    conn.uid_store(
        &SequenceSet::new(seed_uid.to_string()).unwrap(),
        daaki_imap::types::StoreOperation::Add,
        &[Flag::Deleted],
        None,
        TIMEOUT,
    )
    .await
    .unwrap();
    conn.expunge(TIMEOUT).await.unwrap();
}

/// Flatten a THREAD response tree into the concrete IDs it contains.
///
/// RFC 5256 Section 4 models THREAD responses as nested parent/child
/// structures, so tests must walk the whole tree to compare the returned
/// identifiers with the mailbox contents.
fn collect_thread_ids(nodes: &[daaki_imap::types::ThreadNode], ids: &mut Vec<u32>) {
    for node in nodes {
        if let Some(id) = node.id {
            ids.push(id);
        }
        collect_thread_ids(&node.children, ids);
    }
}

/// Expand an ESEARCH `ALL` uid-set into concrete message numbers or UIDs.
///
/// RFC 4731 Section 3.1 returns `ALL` as a compact sequence-set, so tests must
/// expand the returned ranges before comparing them with concrete expectations.
fn expand_uid_ranges(ranges: &[daaki_imap::types::UidRange]) -> Vec<u32> {
    let mut expanded = Vec::new();
    for range in ranges {
        let end = range.end.unwrap_or(range.start);
        for value in range.start..=end {
            expanded.push(value);
        }
    }
    expanded
}

/// Extract the `MESSAGES` count from a STATUS response.
fn status_messages(items: &[daaki_imap::types::StatusItem]) -> Option<u32> {
    items.iter().find_map(|item| match item {
        daaki_imap::types::StatusItem::Messages(value) => Some(*value),
        _ => None,
    })
}

/// Extract the `DELETED-STORAGE` value from a STATUS response.
fn status_deleted_storage(items: &[daaki_imap::types::StatusItem]) -> Option<u64> {
    items.iter().find_map(|item| match item {
        daaki_imap::types::StatusItem::DeletedStorage(value) => Some(*value),
        _ => None,
    })
}

/// Extract the `SIZE` value from a STATUS response.
fn status_size(items: &[daaki_imap::types::StatusItem]) -> Option<u64> {
    items.iter().find_map(|item| match item {
        daaki_imap::types::StatusItem::Size(value) => Some(*value),
        _ => None,
    })
}

/// Extract the `UIDNEXT` value from a STATUS response (RFC 3501 Section 6.3.10).
fn status_uidnext(items: &[daaki_imap::types::StatusItem]) -> Option<u32> {
    items.iter().find_map(|item| match item {
        daaki_imap::types::StatusItem::UidNext(value) => Some(*value),
        _ => None,
    })
}

/// Remove a mailbox and any test messages it contains.
///
/// RFC 3501 Section 6.4.6 marks messages as `\Deleted`, RFC 3501 Section
/// 6.4.3 expunges them, and RFC 3501 Section 6.3.4 then deletes the empty
/// mailbox.
async fn delete_mailbox_and_contents(conn: &ImapConnection, mailbox: &str) {
    let selected = conn.select(mailbox, TIMEOUT).await.unwrap();
    if selected.exists > 0 {
        conn.store(
            &SequenceSet::new("1:*").unwrap(),
            daaki_imap::types::StoreOperation::Add,
            &[Flag::Deleted],
            None,
            TIMEOUT,
        )
        .await
        .unwrap();
        conn.expunge(TIMEOUT).await.unwrap();
    }
    conn.close(TIMEOUT).await.unwrap();
    conn.delete(mailbox, TIMEOUT).await.unwrap();
}

// ==========================================================================
// Connection & Authentication
// ==========================================================================

#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_connect_starttls() {
    let (user, pass) = stalwart::credentials();
    let tls_config = build_danger_tls_config();
    let conn = ImapConnection::connect_with_tls_config(
        stalwart::HOST,
        stalwart::IMAP_PORT,
        TlsMode::None,
        tls_config.clone(),
        TIMEOUT,
    )
    .await
    .unwrap();

    assert_eq!(conn.session_state(), SessionState::NotAuthenticated);

    conn.starttls_with_config(tls_config, TIMEOUT)
        .await
        .unwrap();

    // Stalwart only supports SASL mechanisms, not the LOGIN command.
    conn.authenticate_plain(&user, &pass, TIMEOUT)
        .await
        .unwrap();
    assert_eq!(conn.session_state(), SessionState::Authenticated);

    conn.logout().await.unwrap();
}

/// `TlsMode::StartTls` must complete the STARTTLS upgrade before returning.
///
/// RFC 3501 Section 6.2.1 / RFC 9051 Section 6.2.1 require the client to
/// negotiate TLS before using the protected session. The public
/// `TlsMode::StartTls` contract therefore means `connect_with_tls_config`
/// returns only after the channel has been upgraded and capabilities refreshed.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_connect_starttls_mode_auto_upgrades_before_return() {
    let (user, pass) = stalwart::credentials();
    let tls_config = build_danger_tls_config();
    let conn = ImapConnection::connect_with_tls_config(
        stalwart::HOST,
        stalwart::IMAP_PORT,
        TlsMode::StartTls,
        tls_config,
        TIMEOUT,
    )
    .await
    .unwrap();

    assert_eq!(
        conn.session_state(),
        SessionState::NotAuthenticated,
        "connect should return a pre-authenticated session after STARTTLS"
    );

    let caps = conn.capabilities();
    assert!(
        !caps.iter().any(|cap| matches!(cap, Capability::StartTls)),
        "post-STARTTLS capabilities must be refreshed and must not advertise STARTTLS: {caps:?}"
    );

    conn.authenticate_plain(&user, &pass, TIMEOUT)
        .await
        .unwrap();
    assert_eq!(conn.session_state(), SessionState::Authenticated);

    conn.logout().await.unwrap();
}

/// A connection created with `TlsMode::StartTls` must reject a second manual
/// STARTTLS upgrade attempt.
///
/// RFC 3501 Section 6.2.1 / RFC 9051 Section 6.2.1 allow STARTTLS only on the
/// cleartext session. Once `connect_with_tls_config(..., TlsMode::StartTls, ...)`
/// returns, the protected channel is already established.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_connect_starttls_mode_rejects_second_manual_upgrade() {
    let (user, pass) = stalwart::credentials();
    let tls_config = build_danger_tls_config();
    let conn = ImapConnection::connect_with_tls_config(
        stalwart::HOST,
        stalwart::IMAP_PORT,
        TlsMode::StartTls,
        tls_config.clone(),
        TIMEOUT,
    )
    .await
    .unwrap();

    let result = conn.starttls_with_config(tls_config, TIMEOUT).await;
    assert!(
        matches!(
            result,
            Err(daaki_imap::Error::StartTlsUnavailable | daaki_imap::Error::Protocol(_))
        ),
        "a connection already upgraded via TlsMode::StartTls must reject a second STARTTLS, got {result:?}"
    );

    conn.authenticate_plain(&user, &pass, TIMEOUT)
        .await
        .unwrap();
    conn.logout().await.unwrap();
}

/// A failed AUTHENTICATE PLAIN must leave the session Not Authenticated and
/// allow a later AUTHENTICATE PLAIN retry on the same TLS connection.
///
/// RFC 3501 Section 6.2.2 keeps the client in the Not Authenticated state
/// until AUTHENTICATE succeeds, so a tagged NO for bad credentials must not
/// poison the transport or consume the session.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_authenticate_plain_bad_credentials_preserves_not_authenticated_state_and_allows_retry(
) {
    let (user, pass) = stalwart::credentials();
    let conn = connect_stalwart_not_authenticated().await;

    let result = conn
        .authenticate_plain(&user, "wrong-password-retry", TIMEOUT)
        .await;
    assert!(
        matches!(result, Err(daaki_imap::Error::Auth { .. })),
        "expected Error::Auth for rejected AUTHENTICATE PLAIN, got {result:?}"
    );
    assert_eq!(
        conn.session_state(),
        SessionState::NotAuthenticated,
        "failed AUTHENTICATE PLAIN must leave the session in Not Authenticated state"
    );

    conn.authenticate_plain(&user, &pass, TIMEOUT)
        .await
        .unwrap();
    assert_eq!(
        conn.session_state(),
        SessionState::Authenticated,
        "successful retry AUTHENTICATE PLAIN must authenticate the existing connection"
    );
    conn.noop(TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

/// A failed AUTHENTICATE XOAUTH2 must leave the session Not Authenticated and
/// allow a later AUTHENTICATE PLAIN retry on the same TLS connection.
///
/// RFC 3501 Section 6.2.2 keeps the client in the Not Authenticated state
/// until AUTHENTICATE succeeds, so a tagged NO for a bad XOAUTH2 bearer token
/// must not poison the transport or consume the session.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_authenticate_xoauth2_bad_token_preserves_not_authenticated_state_and_allows_plain_retry(
) {
    let (user, pass) = stalwart::credentials();
    let conn = connect_stalwart_not_authenticated().await;

    assert!(
        has_capability(&conn, &Capability::Auth("XOAUTH2".to_owned())),
        "setup must expose AUTH=XOAUTH2 before this retry test can run"
    );

    let result = conn
        .authenticate_xoauth2(&user, "invalid-token", TIMEOUT)
        .await;
    assert!(
        matches!(result, Err(daaki_imap::Error::Auth { .. })),
        "expected Error::Auth for rejected AUTHENTICATE XOAUTH2, got {result:?}"
    );
    assert_eq!(
        conn.session_state(),
        SessionState::NotAuthenticated,
        "failed AUTHENTICATE XOAUTH2 must leave the session in Not Authenticated state"
    );

    conn.authenticate_plain(&user, &pass, TIMEOUT)
        .await
        .unwrap();
    assert_eq!(
        conn.session_state(),
        SessionState::Authenticated,
        "successful AUTHENTICATE PLAIN after failed XOAUTH2 must authenticate the existing connection"
    );
    conn.noop(TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

// ==========================================================================
// UNAUTHENTICATE (RFC 8437)
// ==========================================================================

/// `UNAUTHENTICATE` must return the session to Not Authenticated and leave
/// the connection ready to advertise a new authentication mechanism.
///
/// RFC 8437 Section 2 resets the session without closing the connection, and
/// once reset the client is free to issue a new AUTHENTICATE or LOGIN command
/// as permitted by the server's capabilities.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_unauthenticate_resets_state_and_allows_new_auth_mechanism() {
    let conn = connect_stalwart().await;

    assert!(
        has_capability(&conn, &Capability::Unauthenticate),
        "authenticated Stalwart session must advertise UNAUTHENTICATE"
    );
    assert_eq!(conn.session_state(), SessionState::Authenticated);

    conn.unauthenticate(TIMEOUT).await.unwrap();

    assert_eq!(
        conn.session_state(),
        SessionState::NotAuthenticated,
        "UNAUTHENTICATE must reset the session to Not Authenticated"
    );
    assert!(
        conn.capabilities()
            .iter()
            .any(|cap| matches!(cap, Capability::Auth(mech) if mech.eq_ignore_ascii_case("PLAIN"))),
        "unauthenticated capability list must still advertise AUTH=PLAIN: {:?}",
        conn.capabilities()
    );

    conn.logout().await.unwrap();
}

/// `UNAUTHENTICATE` from Selected must clear the selected mailbox and still
/// allow re-authentication on the same TLS connection.
///
/// RFC 8437 Section 2 returns the session to Not Authenticated and discards
/// per-user state, including any selected mailbox.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_unauthenticate_from_selected_allows_reauthentication() {
    let (user, pass) = stalwart::credentials();
    let conn = connect_stalwart().await;

    let mbox = unique_name("test-unauth-selected");
    conn.create(&mbox, TIMEOUT).await.unwrap();
    append_messages(
        &conn,
        &mbox,
        "Stalwart unauthenticate selected",
        "stalwart-unauth-selected",
        1,
    )
    .await;
    conn.select(&mbox, TIMEOUT).await.unwrap();
    assert_eq!(conn.session_state(), SessionState::Selected);

    conn.unauthenticate(TIMEOUT).await.unwrap();
    assert_eq!(conn.session_state(), SessionState::NotAuthenticated);

    conn.authenticate_plain(&user, &pass, TIMEOUT)
        .await
        .unwrap();
    assert_eq!(
        conn.session_state(),
        SessionState::Authenticated,
        "re-authentication should restore the authenticated state"
    );

    let selected = conn.select(&mbox, TIMEOUT).await.unwrap();
    assert_eq!(
        selected.exists, 1,
        "the same TLS connection should remain usable after UNAUTHENTICATE"
    );

    delete_mailbox_and_contents(&conn, &mbox).await;
    conn.logout().await.unwrap();
}

/// `UNAUTHENTICATE` must clear ENABLE'd QRESYNC state so later QRESYNC-only
/// commands are rejected until the extension is enabled again.
///
/// RFC 8437 Section 2 clears per-user state across unauthentication, and
/// RFC 7162 Section 3.2.6 requires QRESYNC to be ENABLEd before `UID FETCH`
/// may use the `VANISHED` modifier.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_unauthenticate_clears_enabled_qresync() {
    let (user, pass) = stalwart::credentials();
    let conn = connect_stalwart().await;

    if !has_capability(&conn, &Capability::Condstore)
        || !has_capability(&conn, &Capability::QResync)
        || !has_capability(&conn, &Capability::Unauthenticate)
    {
        eprintln!("skipping: Stalwart does not advertise CONDSTORE/QRESYNC/UNAUTHENTICATE");
        conn.logout().await.unwrap();
        return;
    }

    let mbox = unique_name("test-unauth-qresync");
    conn.create(&mbox, TIMEOUT).await.unwrap();
    append_messages(
        &conn,
        &mbox,
        "Stalwart unauthenticate qresync",
        "stalwart-unauth-qresync",
        1,
    )
    .await;

    let enabled = conn
        .enable(&["CONDSTORE", "QRESYNC"], TIMEOUT)
        .await
        .unwrap();
    assert!(
        enabled
            .iter()
            .any(|cap| cap.eq_ignore_ascii_case("QRESYNC")),
        "ENABLE must report QRESYNC before UNAUTHENTICATE: {enabled:?}"
    );

    conn.unauthenticate(TIMEOUT).await.unwrap();
    conn.authenticate_plain(&user, &pass, TIMEOUT)
        .await
        .unwrap();
    conn.select(&mbox, TIMEOUT).await.unwrap();

    let uid = sorted_uids(&conn)
        .await
        .into_iter()
        .next()
        .expect("mailbox seeded for QRESYNC reset test must contain one UID");
    let err = conn
        .uid_fetch_vanished(
            &SequenceSet::new(uid.to_string()).unwrap(),
            &[FetchAttr::Uid],
            1,
            TIMEOUT,
        )
        .await
        .unwrap_err();
    assert!(
        matches!(err, daaki_imap::Error::MissingCapability(ref cap) if cap.contains("QRESYNC")),
        "UID FETCH VANISHED after UNAUTHENTICATE must require QRESYNC to be ENABLEd again, got {err:?}"
    );

    delete_mailbox_and_contents(&conn, &mbox).await;
    conn.logout().await.unwrap();
}

// ==========================================================================
// IMAP4rev2 Capability (RFC 9051)
// ==========================================================================

#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_capabilities_imap4rev2() {
    let conn = connect_stalwart().await;
    let caps = conn.capabilities();

    // Stalwart should advertise IMAP4rev2.
    assert!(
        caps.iter().any(|c| matches!(c, Capability::Imap4Rev2)),
        "expected IMAP4rev2 capability, got: {caps:?}"
    );

    // It should also still advertise rev1 for backward compatibility.
    assert!(
        caps.iter().any(|c| matches!(c, Capability::Imap4Rev1)),
        "expected IMAP4rev1 capability for backward compat, got: {caps:?}"
    );

    conn.logout().await.unwrap();
}

/// STARTTLS must refetch capabilities on the protected channel, and the
/// refreshed capability set must no longer advertise `STARTTLS`.
///
/// RFC 3501 Section 6.2.1 / RFC 9051 Section 6.2.1 require the client to
/// discard cached capabilities after a successful STARTTLS command and
/// refresh them on the encrypted channel.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_starttls_refetches_capabilities_without_starttls() {
    let (user, pass) = stalwart::credentials();
    let tls_config = build_danger_tls_config();
    let conn = ImapConnection::connect_with_tls_config(
        stalwart::HOST,
        stalwart::IMAP_PORT,
        TlsMode::None,
        tls_config.clone(),
        TIMEOUT,
    )
    .await
    .unwrap();

    let caps_before = conn.capabilities();
    assert!(
        caps_before
            .iter()
            .any(|cap| matches!(cap, Capability::StartTls)),
        "plaintext capability list must advertise STARTTLS before upgrade: {caps_before:?}"
    );

    conn.starttls_with_config(tls_config, TIMEOUT)
        .await
        .unwrap();

    let caps_after = conn.capabilities();
    assert!(
        !caps_after
            .iter()
            .any(|cap| matches!(cap, Capability::StartTls)),
        "post-STARTTLS capabilities must be refreshed and must not advertise STARTTLS: {caps_after:?}"
    );

    conn.authenticate_plain(&user, &pass, TIMEOUT)
        .await
        .unwrap();
    conn.logout().await.unwrap();
}

/// Verify that Stalwart advertises rev2 base features as capabilities.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_rev2_base_capabilities() {
    let conn = connect_stalwart().await;

    // RFC 9051 incorporates these extensions into the base protocol.
    // Stalwart should advertise them.
    let expected = [
        ("MOVE", Capability::Move),
        ("UNSELECT", Capability::Unselect),
        ("NAMESPACE", Capability::Namespace),
        ("ENABLE", Capability::Enable),
        ("UIDPLUS", Capability::UidPlus),
        ("ESEARCH", Capability::Esearch),
        ("IDLE", Capability::Idle),
    ];

    for (name, cap) in &expected {
        assert!(
            has_capability(&conn, cap),
            "expected {name} capability (base in IMAP4rev2)"
        );
    }

    conn.logout().await.unwrap();
}

// ==========================================================================
// LIST mailboxes
// ==========================================================================

#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_list_mailboxes() {
    let conn = connect_stalwart().await;
    let mailboxes = conn.list("", "*", TIMEOUT).await.unwrap();

    assert!(
        mailboxes
            .iter()
            .any(|m| m.name.as_str().eq_ignore_ascii_case("INBOX")),
        "expected INBOX in LIST response: {:?}",
        mailboxes.iter().map(|m| &m.name).collect::<Vec<_>>()
    );

    conn.logout().await.unwrap();
}

/// LIST must resolve a relative mailbox name against the reference name.
///
/// RFC 3501 Section 6.3.8 defines `reference name` as the context against
/// which the mailbox pattern is interpreted.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_list_reference_resolves_relative_mailbox_name() {
    let conn = connect_stalwart().await;

    let delimiter = hierarchy_delimiter(&conn).await;
    let parent = unique_name("test-list-reference-parent");
    let child = format!("{parent}{delimiter}child");
    let reference = format!("{parent}{delimiter}");

    conn.create(&parent, TIMEOUT).await.unwrap();
    conn.create(&child, TIMEOUT).await.unwrap();

    let listing = conn.list(&reference, "child", TIMEOUT).await.unwrap();
    assert_eq!(
        listing.len(),
        1,
        "LIST with a relative mailbox name should resolve to one canonical match: {listing:?}"
    );
    assert_eq!(listing[0].name.as_str(), child.as_str());

    conn.delete(&child, TIMEOUT).await.unwrap();
    conn.delete(&parent, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

/// LIST with "%" and a non-empty reference name must match only the next
/// hierarchy level below that reference.
///
/// RFC 3501 Section 6.3.8 defines "%" as matching only one hierarchy level,
/// with the reference name supplying the context for relative patterns.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_list_reference_percent_matches_only_direct_children() {
    let conn = connect_stalwart().await;

    let delimiter = hierarchy_delimiter(&conn).await;
    let parent = unique_name("test-list-reference-percent-parent");
    let child = format!("{parent}{delimiter}child");
    let grandchild = format!("{child}{delimiter}grandchild");
    let reference = format!("{parent}{delimiter}");

    conn.create(&parent, TIMEOUT).await.unwrap();
    conn.create(&child, TIMEOUT).await.unwrap();
    conn.create(&grandchild, TIMEOUT).await.unwrap();

    let listing = conn.list(&reference, "%", TIMEOUT).await.unwrap();
    let mut names: Vec<_> = listing
        .iter()
        .map(|info| info.name.as_str().to_owned())
        .collect();
    names.sort();
    assert_eq!(
        names,
        vec![child.clone()],
        "LIST % under a reference must return only direct children: {listing:?}"
    );

    conn.delete(&grandchild, TIMEOUT).await.unwrap();
    conn.delete(&child, TIMEOUT).await.unwrap();
    conn.delete(&parent, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

/// When a server accepts the deprecated LSUB command, a non-empty reference
/// name must resolve a relative mailbox name against that reference and return
/// the canonical subscribed mailbox name.
///
/// RFC 3501 Sections 6.3.8 and 6.3.9 define the reference-name resolution for
/// mailbox patterns and the subscribed mailbox listing returned by LSUB. RFC
/// 9051 Appendix E deprecates LSUB, so IMAP4rev2 servers may reject it.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_lsub_reference_resolves_relative_mailbox_name() {
    let conn = connect_stalwart().await;

    let delimiter = hierarchy_delimiter(&conn).await;
    let parent = unique_name("test-lsub-reference-parent");
    let child = format!("{parent}{delimiter}child");
    let reference = format!("{parent}{delimiter}");

    conn.create(&parent, TIMEOUT).await.unwrap();
    conn.create(&child, TIMEOUT).await.unwrap();
    conn.subscribe(&child, TIMEOUT).await.unwrap();

    match conn.lsub(&reference, "child", TIMEOUT).await {
        Ok(subscribed) => {
            assert_eq!(
                subscribed.len(),
                1,
                "LSUB with a relative mailbox name should resolve to one canonical subscribed match: {subscribed:?}"
            );
            assert_eq!(subscribed[0].name.as_str(), child.as_str());
        }
        Err(daaki_imap::Error::Bad { .. }) => {
            // RFC 9051 Appendix E deprecates LSUB, so rev2 servers may reject it.
            eprintln!("Stalwart rejected LSUB reference lookup (strict rev2 behavior) — OK");
        }
        Err(e) => panic!("unexpected error from LSUB: {e:?}"),
    }

    conn.unsubscribe(&child, TIMEOUT).await.unwrap();
    conn.delete(&child, TIMEOUT).await.unwrap();
    conn.delete(&parent, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

/// LIST-EXTENDED with multiple relative patterns must resolve each pattern
/// against the reference name and return canonical mailbox names.
///
/// RFC 5258 Section 3 extends LIST to support multiple patterns, while RFC
/// 3501 Section 6.3.8 defines how those relative patterns are resolved.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_rev2_list_extended_reference_resolves_multiple_patterns() {
    let conn = connect_stalwart().await;

    let delimiter = hierarchy_delimiter(&conn).await;
    let parent = unique_name("test-list-extended-reference-parent");
    let child_one = format!("{parent}{delimiter}child-one");
    let child_two = format!("{parent}{delimiter}child-two");
    let reference = format!("{parent}{delimiter}");

    conn.create(&parent, TIMEOUT).await.unwrap();
    conn.create(&child_one, TIMEOUT).await.unwrap();
    conn.create(&child_two, TIMEOUT).await.unwrap();

    let listing = conn
        .list_extended(&reference, &["child-one", "child-two"], &[], &[], TIMEOUT)
        .await
        .unwrap();
    let mut names: Vec<_> = listing
        .iter()
        .map(|info| info.name.as_str().to_owned())
        .collect();
    names.sort();
    assert_eq!(
        names,
        vec![child_one.clone(), child_two.clone()],
        "LIST-EXTENDED should resolve every relative pattern against the reference: {listing:?}"
    );

    conn.delete(&child_one, TIMEOUT).await.unwrap();
    conn.delete(&child_two, TIMEOUT).await.unwrap();
    conn.delete(&parent, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

/// LIST-EXTENDED with "%" and a non-empty reference name must still match
/// only direct children below that reference.
///
/// RFC 5258 Sections 3 and 4 preserve RFC 3501 Section 6.3.8 wildcard
/// semantics when LIST-EXTENDED adds return options such as `CHILDREN`.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_rev2_list_extended_reference_percent_matches_only_direct_children() {
    let conn = connect_stalwart().await;

    let delimiter = hierarchy_delimiter(&conn).await;
    let parent = unique_name("test-list-extended-percent-parent");
    let child = format!("{parent}{delimiter}child");
    let grandchild = format!("{child}{delimiter}grandchild");
    let reference = format!("{parent}{delimiter}");

    conn.create(&parent, TIMEOUT).await.unwrap();
    conn.create(&child, TIMEOUT).await.unwrap();
    conn.create(&grandchild, TIMEOUT).await.unwrap();

    let listing = conn
        .list_extended(&reference, &["%"], &[], &["CHILDREN"], TIMEOUT)
        .await
        .unwrap();
    assert_eq!(
        listing.len(),
        1,
        "LIST-EXTENDED % under a reference must return only direct children: {listing:?}"
    );
    assert_eq!(listing[0].name.as_str(), child.as_str());
    assert!(
        listing[0]
            .attributes
            .iter()
            .any(|attr| matches!(attr, MailboxAttribute::HasChildren)),
        "direct child with a grandchild must report \\HasChildren: {:?}",
        listing[0].attributes
    );

    conn.delete(&grandchild, TIMEOUT).await.unwrap();
    conn.delete(&child, TIMEOUT).await.unwrap();
    conn.delete(&parent, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

/// LIST RETURN (CHILDREN) must distinguish parent and leaf mailboxes.
///
/// RFC 5258 Section 4 defines the `CHILDREN` return option and the
/// `\\HasChildren` / `\\HasNoChildren` attributes surfaced by LIST
/// responses.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_rev2_list_extended_return_children_distinguishes_parent_and_leaf() {
    let conn = connect_stalwart().await;

    let parent = format!("test-list-children-parent-{}", std::process::id());
    let child = format!("{parent}/child");
    conn.create(&parent, TIMEOUT).await.unwrap();
    conn.create(&child, TIMEOUT).await.unwrap();

    let listing = conn
        .list_extended("", &[&parent, &child], &[], &["CHILDREN"], TIMEOUT)
        .await
        .unwrap();

    let parent_entry = listing
        .iter()
        .find(|info| info.name.as_str() == parent)
        .unwrap_or_else(|| panic!("expected parent mailbox in LIST results: {listing:?}"));
    assert!(
        parent_entry
            .attributes
            .iter()
            .any(|attr| matches!(attr, MailboxAttribute::HasChildren)),
        "parent mailbox must report \\HasChildren: {:?}",
        parent_entry.attributes
    );

    let child_entry = listing
        .iter()
        .find(|info| info.name.as_str() == child)
        .unwrap_or_else(|| panic!("expected child mailbox in LIST results: {listing:?}"));
    assert!(
        child_entry
            .attributes
            .iter()
            .any(|attr| matches!(attr, MailboxAttribute::HasNoChildren)),
        "leaf mailbox must report \\HasNoChildren: {:?}",
        child_entry.attributes
    );

    conn.delete(&child, TIMEOUT).await.unwrap();
    conn.delete(&parent, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

// ==========================================================================
// SELECT / EXAMINE INBOX
// ==========================================================================

#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_select_inbox() {
    let conn = connect_stalwart().await;

    let selected = conn.select("INBOX", TIMEOUT).await.unwrap();
    assert!(
        selected.uid_validity.unwrap_or(0) > 0,
        "UIDVALIDITY must be > 0"
    );
    assert_eq!(conn.session_state(), SessionState::Selected);

    conn.logout().await.unwrap();
}

#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_examine_inbox() {
    let conn = connect_stalwart().await;

    let selected = conn.examine("INBOX", TIMEOUT).await.unwrap();
    assert!(selected.uid_validity.unwrap_or(0) > 0);

    conn.logout().await.unwrap();
}

// ==========================================================================
// APPEND + FETCH (verifies basic message flow)
// ==========================================================================

#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_append_and_fetch() {
    let conn = connect_stalwart().await;

    let mbox = &format!("test-append-{}", std::process::id());
    conn.create(mbox, TIMEOUT).await.unwrap();
    conn.select(mbox, TIMEOUT).await.unwrap();

    conn.append(mbox, &[Flag::Seen], None, TEST_MESSAGE, TIMEOUT)
        .await
        .unwrap();

    // Re-select to see the new message.
    let selected = conn.select(mbox, TIMEOUT).await.unwrap();
    assert!(
        selected.exists >= 1,
        "expected at least 1 message after APPEND"
    );

    let fetches = conn
        .uid_fetch(
            &SequenceSet::new("*").unwrap(),
            &[FetchAttr::Uid, FetchAttr::Flags, FetchAttr::Envelope],
            TIMEOUT,
        )
        .await
        .unwrap();
    assert!(!fetches.is_empty());

    let msg = &fetches[0];
    assert!(msg.uid.is_some());
    assert!(msg.envelope.is_some());

    // Clean up.
    conn.delete(mbox, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

// ==========================================================================
// MOVE (RFC 6851 / RFC 9051 base)
// ==========================================================================

#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_move_message() {
    let conn = connect_stalwart().await;

    let src = &format!("test-move-src-{}", std::process::id());
    let dst = &format!("test-move-dst-{}", std::process::id());
    conn.create(src, TIMEOUT).await.unwrap();
    conn.create(dst, TIMEOUT).await.unwrap();

    // Append a message to src.
    conn.append(src, &[], None, TEST_MESSAGE, TIMEOUT)
        .await
        .unwrap();
    conn.select(src, TIMEOUT).await.unwrap();

    // UID MOVE to dst (RFC 6851, base in rev2 per RFC 9051 Appendix E item 2).
    let fetches = conn
        .uid_fetch(
            &SequenceSet::new("1:*").unwrap(),
            &[FetchAttr::Uid],
            TIMEOUT,
        )
        .await
        .unwrap();
    assert!(!fetches.is_empty(), "expected at least one message in src");
    let uid_set = fetches[0].uid.unwrap().to_string();

    let result = conn
        .uid_move_messages(&SequenceSet::new(&uid_set).unwrap(), dst, TIMEOUT)
        .await
        .unwrap();
    let _ = result; // MoveResult with COPYUID and expunged info

    // Verify src is empty.
    let selected = conn.select(src, TIMEOUT).await.unwrap();
    assert_eq!(selected.exists, 0, "src should be empty after MOVE");

    // Verify dst has the message.
    let selected = conn.select(dst, TIMEOUT).await.unwrap();
    assert!(selected.exists >= 1, "dst should have the moved message");

    // Clean up.
    conn.delete(src, TIMEOUT).await.unwrap();
    conn.delete(dst, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

// ==========================================================================
// CLOSE (RFC 3501 Section 6.4.2)
// ==========================================================================

/// `CLOSE` must expunge `\\Deleted` messages and return the connection to the
/// authenticated state.
///
/// RFC 3501 Section 6.4.2 defines CLOSE as closing the selected mailbox,
/// removing all messages with the `\\Deleted` flag, and leaving the session in
/// Authenticated state.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_close_expunge_removes_deleted_messages_and_returns_authenticated() {
    let conn = connect_stalwart().await;

    let mbox = unique_name("test-close-expunge");
    conn.create(&mbox, TIMEOUT).await.unwrap();
    conn.append(
        &mbox,
        &[],
        None,
        &build_message(
            "Stalwart CLOSE deleted",
            &format!("{mbox}-close-deleted@daaki"),
            "Deleted by CLOSE",
        ),
        TIMEOUT,
    )
    .await
    .unwrap();
    conn.append(
        &mbox,
        &[],
        None,
        &build_message(
            "Stalwart CLOSE survivor",
            &format!("{mbox}-close-survivor@daaki"),
            "Survives CLOSE",
        ),
        TIMEOUT,
    )
    .await
    .unwrap();

    conn.select(&mbox, TIMEOUT).await.unwrap();
    conn.store(
        &SequenceSet::new("1").unwrap(),
        daaki_imap::types::StoreOperation::Add,
        &[Flag::Deleted],
        None,
        TIMEOUT,
    )
    .await
    .unwrap();

    conn.close(TIMEOUT).await.unwrap();
    assert_eq!(
        conn.session_state(),
        SessionState::Authenticated,
        "CLOSE must leave the connection in Authenticated state"
    );

    let selected = conn.select(&mbox, TIMEOUT).await.unwrap();
    assert_eq!(
        selected.exists, 1,
        "CLOSE must expunge the deleted message from the mailbox"
    );

    let remaining = conn
        .fetch(
            &SequenceSet::new("1:*").unwrap(),
            &[FetchAttr::Envelope],
            TIMEOUT,
        )
        .await
        .unwrap();
    assert_eq!(remaining.len(), 1, "exactly one message must remain");
    assert_eq!(
        remaining[0]
            .envelope
            .as_ref()
            .and_then(|envelope| envelope.subject.clone()),
        Some("Stalwart CLOSE survivor".to_owned()),
        "CLOSE must preserve the non-deleted message"
    );

    delete_mailbox_and_contents(&conn, &mbox).await;
    conn.logout().await.unwrap();
}

// ==========================================================================
// UNSELECT (RFC 3691 / RFC 9051 base)
// ==========================================================================

#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_unselect() {
    let conn = connect_stalwart().await;

    conn.select("INBOX", TIMEOUT).await.unwrap();
    assert_eq!(conn.session_state(), SessionState::Selected);

    conn.unselect(TIMEOUT).await.unwrap();
    assert_eq!(conn.session_state(), SessionState::Authenticated);

    conn.logout().await.unwrap();
}

// ==========================================================================
// ENABLE (RFC 5161 / RFC 9051 base)
// ==========================================================================

#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_enable_condstore() {
    let conn = connect_stalwart().await;

    let enabled = conn.enable(&["CONDSTORE"], TIMEOUT).await.unwrap();
    assert!(
        enabled.iter().any(|s| s.eq_ignore_ascii_case("CONDSTORE")),
        "expected CONDSTORE to be enabled: {enabled:?}"
    );

    conn.logout().await.unwrap();
}

#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_enable_qresync() {
    let conn = connect_stalwart().await;

    // QRESYNC requires CONDSTORE to be enabled first (RFC 7162 Section 3.2.5).
    let _ = conn.enable(&["CONDSTORE"], TIMEOUT).await.unwrap();

    match conn.enable(&["QRESYNC"], TIMEOUT).await {
        Ok(enabled) => {
            assert!(
                enabled.iter().any(|s| s.eq_ignore_ascii_case("QRESYNC")),
                "expected QRESYNC to be enabled: {enabled:?}"
            );
        }
        Err(daaki_imap::Error::MissingCapability(_)) => {
            eprintln!("skipping: Stalwart does not advertise QRESYNC");
        }
        Err(e) => panic!("unexpected error from ENABLE QRESYNC: {e:?}"),
    }

    conn.logout().await.unwrap();
}

// ==========================================================================
// NAMESPACE (RFC 2342 / RFC 9051 base)
// ==========================================================================

#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_namespace() {
    let conn = connect_stalwart().await;

    let ns = conn.namespace(TIMEOUT).await.unwrap();

    assert!(
        !ns.personal.is_empty(),
        "expected at least one personal namespace"
    );

    conn.logout().await.unwrap();
}

/// NAMESPACE on a selected mailbox must not deselect the mailbox.
///
/// RFC 2342 Section 5 defines NAMESPACE as a mailbox namespace query, and RFC
/// 3501 Section 6.4 requires selected-state commands to leave the selected
/// mailbox intact.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_rev2_namespace_selected_preserves_selected_mailbox() {
    let conn = connect_stalwart().await;

    let mailbox = unique_name("test-stalwart-namespace-selected");
    conn.create(&mailbox, TIMEOUT).await.unwrap();
    conn.append(&mailbox, &[], None, TEST_MESSAGE, TIMEOUT)
        .await
        .unwrap();
    conn.select(&mailbox, TIMEOUT).await.unwrap();

    let ns = conn.namespace(TIMEOUT).await.unwrap();
    assert!(
        !ns.personal.is_empty(),
        "NAMESPACE must report at least one personal namespace"
    );
    assert_selected_mailbox_still_usable(&conn, "NAMESPACE").await;

    conn.close(TIMEOUT).await.unwrap();
    conn.delete(&mailbox, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

// ==========================================================================
// ESEARCH (RFC 4731 / RFC 9051 base)
// ==========================================================================

#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_esearch() {
    let conn = connect_stalwart().await;

    let mbox = &format!("test-esearch-{}", std::process::id());
    conn.create(mbox, TIMEOUT).await.unwrap();

    // Append a few messages so we have something to search.
    for i in 0..3 {
        let msg = format!(
            "From: sender@example.com\r\n\
             To: recipient@example.com\r\n\
             Subject: ESEARCH test {i}\r\n\
             Date: Sun, 15 Mar 2026 00:00:0{i} +0000\r\n\
             Message-ID: <esearch-{i}-{}@daaki>\r\n\
             MIME-Version: 1.0\r\n\
             Content-Type: text/plain\r\n\
             \r\n\
             Body {i}\r\n",
            std::process::id()
        );
        conn.append(mbox, &[], None, msg.as_bytes(), TIMEOUT)
            .await
            .unwrap();
    }

    conn.select(mbox, TIMEOUT).await.unwrap();

    // ESEARCH with RETURN (COUNT) — RFC 4731 / RFC 9051 Section 6.4.4.
    // Note: Stalwart's admin account has search restrictions with SEARCH RETURN.
    // Use a simple sequence-number FETCH to verify messages exist, then test
    // ESEARCH on that set.
    match conn.search_esearch("1:*", &["COUNT"], TIMEOUT).await {
        Ok(esearch) => {
            assert_eq!(esearch.count, Some(3), "expected COUNT=3, got: {esearch:?}");

            // UID ESEARCH variant.
            let esearch = conn
                .uid_search_esearch("1:*", &["COUNT"], TIMEOUT)
                .await
                .unwrap();
            assert_eq!(esearch.count, Some(3), "expected COUNT=3 from UID ESEARCH");
        }
        Err(daaki_imap::Error::No { .. }) => {
            // Stalwart admin account search restriction — skip ESEARCH assertions.
            eprintln!("skipping ESEARCH assertions: Stalwart admin search restriction");
        }
        Err(e) => panic!("unexpected error from ESEARCH: {e:?}"),
    }

    // Clean up.
    conn.delete(mbox, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

/// Sequence-number ESEARCH on an empty mailbox with `MIN`, `MAX`, `COUNT`, and
/// `ALL` requested must report `COUNT=0`, omit `MIN`/`MAX`, and return an empty
/// `ALL` set.
///
/// RFC 4731 Section 3.1 defines `MIN`, `MAX`, and `ALL` over the current match
/// set, so an empty result must not fabricate message numbers.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_search_esearch_empty_mailbox_all_min_max_absent() {
    let conn = connect_stalwart().await;

    let mbox = unique_name("test-stalwart-esearch-empty");
    conn.create(&mbox, TIMEOUT).await.unwrap();
    conn.select(&mbox, TIMEOUT).await.unwrap();

    let esearch = conn
        .search_esearch("1:*", &["MIN", "MAX", "COUNT", "ALL"], TIMEOUT)
        .await
        .unwrap_or_else(|e| panic!("unexpected error from ESEARCH: {e:?}"));
    assert!(
        !esearch.uid,
        "plain SEARCH RETURN must not set the UID marker"
    );
    assert_eq!(esearch.count, Some(0), "empty mailbox must report COUNT=0");
    assert_eq!(esearch.min, None, "empty mailbox must not fabricate MIN");
    assert_eq!(esearch.max, None, "empty mailbox must not fabricate MAX");
    assert!(
        esearch.all.is_empty(),
        "empty mailbox must return an empty ALL set: {esearch:?}"
    );

    conn.close(TIMEOUT).await.unwrap();
    conn.delete(&mbox, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

/// `SEARCH RETURN (ALL ...)` must report sequence numbers, not immutable UIDs,
/// after earlier messages have been expunged.
///
/// RFC 4731 Section 3.1 defines `ALL` for `SEARCH RETURN` as the set of
/// matching message numbers, while RFC 3501 Section 2.3.1.2 keeps sequence
/// numbers distinct from UIDs.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_search_esearch_all_uses_sequence_numbers_after_uid_gap() {
    let conn = connect_stalwart().await;

    let mbox = unique_name("test-stalwart-esearch-seq-gap");
    conn.create(&mbox, TIMEOUT).await.unwrap();
    conn.select(&mbox, TIMEOUT).await.unwrap();
    create_uid_gap(&conn, &mbox).await;

    for label in ["stalwart-esearch-seq-gap-a", "stalwart-esearch-seq-gap-b"] {
        let message = build_message(
            label,
            &format!("{label}-{}@daaki", std::process::id()),
            label,
        );
        conn.append(&mbox, &[], None, &message, TIMEOUT)
            .await
            .unwrap();
    }

    let live_uids: Vec<u32> = conn
        .uid_fetch(
            &SequenceSet::new("1:*").unwrap(),
            &[FetchAttr::Uid],
            TIMEOUT,
        )
        .await
        .unwrap()
        .iter()
        .map(|fetch| fetch.uid.expect("FETCH must include UID"))
        .collect();
    assert_eq!(
        live_uids.len(),
        2,
        "expected exactly two live messages after creating the UID gap"
    );

    let esearch = conn
        .search_esearch("1:*", &["MIN", "MAX", "COUNT", "ALL"], TIMEOUT)
        .await
        .unwrap_or_else(|e| panic!("unexpected error from ESEARCH: {e:?}"));
    assert!(
        !esearch.uid,
        "SEARCH RETURN must remain sequence-number based"
    );
    assert_eq!(esearch.count, Some(2), "COUNT must be 2");
    assert_eq!(
        esearch.min,
        Some(1),
        "MIN must be the first sequence number"
    );
    assert_eq!(esearch.max, Some(2), "MAX must be the last sequence number");
    assert_eq!(
        expand_uid_ranges(&esearch.all),
        vec![1, 2],
        "SEARCH RETURN ALL must report sequence numbers"
    );
    assert_ne!(
        live_uids,
        vec![1, 2],
        "setup must create a divergence between sequence numbers and UIDs"
    );

    delete_mailbox_and_contents(&conn, &mbox).await;
    conn.logout().await.unwrap();
}

/// `UID SEARCH RETURN (ALL ...)` must report immutable UIDs, not current
/// sequence numbers, after earlier messages have been expunged.
///
/// RFC 4731 Section 3.1 defines `ALL` for `UID SEARCH RETURN` as the matching
/// UID set, and RFC 3501 Section 2.3.1.1 requires those UIDs to remain stable.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_uid_search_esearch_all_uses_uids_after_uid_gap() {
    let conn = connect_stalwart().await;

    let mbox = unique_name("test-stalwart-esearch-uid-gap");
    conn.create(&mbox, TIMEOUT).await.unwrap();
    conn.select(&mbox, TIMEOUT).await.unwrap();
    create_uid_gap(&conn, &mbox).await;

    for label in ["stalwart-esearch-uid-gap-a", "stalwart-esearch-uid-gap-b"] {
        let message = build_message(
            label,
            &format!("{label}-{}@daaki", std::process::id()),
            label,
        );
        conn.append(&mbox, &[], None, &message, TIMEOUT)
            .await
            .unwrap();
    }

    let live_uids: Vec<u32> = conn
        .uid_fetch(
            &SequenceSet::new("1:*").unwrap(),
            &[FetchAttr::Uid],
            TIMEOUT,
        )
        .await
        .unwrap()
        .iter()
        .map(|fetch| fetch.uid.expect("FETCH must include UID"))
        .collect();
    assert_eq!(
        live_uids.len(),
        2,
        "expected exactly two live messages after creating the UID gap"
    );

    let esearch = conn
        .uid_search_esearch("1:*", &["MIN", "MAX", "COUNT", "ALL"], TIMEOUT)
        .await
        .unwrap_or_else(|e| panic!("unexpected error from UID ESEARCH: {e:?}"));
    assert!(esearch.uid, "UID SEARCH RETURN must set the UID marker");
    assert_eq!(esearch.count, Some(2), "COUNT must be 2");
    assert_eq!(
        esearch.min,
        Some(*live_uids.first().expect("must have at least one live UID")),
        "MIN must be the first live UID"
    );
    assert_eq!(
        esearch.max,
        Some(*live_uids.last().expect("must have at least one live UID")),
        "MAX must be the last live UID"
    );
    assert_eq!(
        expand_uid_ranges(&esearch.all),
        live_uids,
        "UID SEARCH RETURN ALL must report UIDs"
    );

    delete_mailbox_and_contents(&conn, &mbox).await;
    conn.logout().await.unwrap();
}

/// SEARCH RETURN (SAVE) must let `$` drive a subsequent FETCH by sequence
/// number.
///
/// RFC 5182 Section 2 defines SEARCH RETURN (SAVE) and the special `$`
/// marker used by later sequence-number commands.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_search_save_fetches_saved_sequence_results() {
    let conn = connect_stalwart().await;

    if !has_capability(&conn, &Capability::SearchRes) {
        eprintln!("skipping: SEARCHRES not advertised");
        conn.logout().await.unwrap();
        return;
    }

    let mbox = format!("test-search-save-seq-{}", std::process::id());
    conn.create(&mbox, TIMEOUT).await.unwrap();
    conn.select(&mbox, TIMEOUT).await.unwrap();

    let seen_message = build_message(
        "Stalwart SEARCH SAVE seen",
        &format!("stalwart-search-save-seen-{}@daaki", std::process::id()),
        "Seen body",
    );
    let unseen_message = build_message(
        "Stalwart SEARCH SAVE unseen",
        &format!("stalwart-search-save-unseen-{}@daaki", std::process::id()),
        "Unseen body",
    );
    conn.append(&mbox, &[Flag::Seen], None, &seen_message, TIMEOUT)
        .await
        .unwrap();
    conn.append(&mbox, &[], None, &unseen_message, TIMEOUT)
        .await
        .unwrap();

    conn.select(&mbox, TIMEOUT).await.unwrap();
    conn.search_save("2", TIMEOUT)
        .await
        .expect("SEARCH RETURN (SAVE) with a sequence-set criterion must succeed");

    let fetches = conn
        .fetch(
            &SequenceSet::new("$").unwrap(),
            &[FetchAttr::Uid, FetchAttr::Flags],
            TIMEOUT,
        )
        .await
        .unwrap();
    assert_eq!(
        fetches.len(),
        1,
        "saved SEARCH result should contain one message"
    );
    assert_eq!(
        fetches[0].seq, 2,
        "UNSEEN message should be sequence number 2"
    );
    assert!(
        !fetches[0]
            .flags
            .as_ref()
            .is_some_and(|flags| flags.iter().any(|flag| matches!(flag, Flag::Seen))),
        "FETCH $ should return the unseen message, not the seen one"
    );

    conn.close(TIMEOUT).await.unwrap();
    conn.delete(&mbox, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

/// UID SEARCH RETURN (SAVE) must let `$` drive a subsequent UID FETCH.
///
/// RFC 5182 Section 2 defines UID SEARCH RETURN (SAVE) and requires `$` to
/// reference the saved UID result in later UID commands.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_uid_search_save_fetches_saved_uid_results() {
    let conn = connect_stalwart().await;

    if !has_capability(&conn, &Capability::SearchRes) {
        eprintln!("skipping: SEARCHRES not advertised");
        conn.logout().await.unwrap();
        return;
    }

    let mbox = format!("test-search-save-uid-{}", std::process::id());
    conn.create(&mbox, TIMEOUT).await.unwrap();
    conn.select(&mbox, TIMEOUT).await.unwrap();

    let seen_message = build_message(
        "Stalwart UID SEARCH SAVE seen",
        &format!("stalwart-uid-search-save-seen-{}@daaki", std::process::id()),
        "Seen body",
    );
    let unseen_message = build_message(
        "Stalwart UID SEARCH SAVE unseen",
        &format!(
            "stalwart-uid-search-save-unseen-{}@daaki",
            std::process::id()
        ),
        "Unseen body",
    );
    conn.append(&mbox, &[Flag::Seen], None, &seen_message, TIMEOUT)
        .await
        .unwrap();
    conn.append(&mbox, &[], None, &unseen_message, TIMEOUT)
        .await
        .unwrap();

    conn.select(&mbox, TIMEOUT).await.unwrap();
    let baseline = conn
        .uid_fetch(
            &SequenceSet::new("1:*").unwrap(),
            &[FetchAttr::Uid, FetchAttr::Flags],
            TIMEOUT,
        )
        .await
        .unwrap();
    assert_eq!(baseline.len(), 2, "expected 2 messages in test mailbox");
    let unseen_uid = baseline
        .iter()
        .find_map(|fetch| {
            let is_seen = fetch
                .flags
                .as_ref()
                .is_some_and(|flags| flags.iter().any(|flag| matches!(flag, Flag::Seen)));
            if is_seen {
                None
            } else {
                fetch.uid
            }
        })
        .expect("expected one unseen message with a UID");

    conn.uid_search_save(unseen_uid.to_string(), TIMEOUT)
        .await
        .expect("UID SEARCH RETURN (SAVE) with a UID criterion must succeed");

    let fetches = conn
        .uid_fetch(
            &SequenceSet::new("$").unwrap(),
            &[FetchAttr::Uid, FetchAttr::Flags],
            TIMEOUT,
        )
        .await
        .unwrap();
    assert_eq!(
        fetches.len(),
        1,
        "saved UID SEARCH result should contain one message"
    );
    assert_eq!(fetches[0].uid, Some(unseen_uid));
    assert!(
        !fetches[0]
            .flags
            .as_ref()
            .is_some_and(|flags| flags.iter().any(|flag| matches!(flag, Flag::Seen))),
        "UID FETCH $ should return the unseen message, not the seen one"
    );

    conn.close(TIMEOUT).await.unwrap();
    conn.delete(&mbox, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

// ==========================================================================
// SELECT with CONDSTORE (RFC 7162)
// ==========================================================================

#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_select_condstore() {
    let conn = connect_stalwart().await;

    match conn
        .select_with(
            "INBOX",
            &daaki_imap::types::SelectOptions::condstore(),
            TIMEOUT,
        )
        .await
    {
        Ok(selected) => {
            assert!(
                selected.uid_validity.unwrap_or(0) > 0,
                "UIDVALIDITY must be > 0"
            );
            assert!(
                selected.highest_mod_seq.is_some(),
                "expected HIGHESTMODSEQ in CONDSTORE SELECT response"
            );
        }
        Err(daaki_imap::Error::MissingCapability(_)) => {
            eprintln!("skipping: Stalwart does not advertise CONDSTORE");
        }
        Err(e) => panic!("unexpected error from SELECT CONDSTORE: {e:?}"),
    }

    conn.logout().await.unwrap();
}

// ==========================================================================
// EXAMINE / FETCH CHANGEDSINCE (RFC 7162)
// ==========================================================================

/// `examine_with(SelectOptions::condstore())` must surface `HIGHESTMODSEQ`
/// while keeping the mailbox read-only.
///
/// RFC 7162 Section 3.1.2.1 requires `HIGHESTMODSEQ` in successful
/// `EXAMINE CONDSTORE` responses, and RFC 3501 Section 6.3.2 defines EXAMINE
/// as the read-only variant of SELECT.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_examine_with_condstore_reports_highestmodseq_and_read_only() {
    let conn = connect_stalwart().await;

    if !has_capability(&conn, &Capability::Condstore) {
        eprintln!("skipping: Stalwart does not advertise CONDSTORE");
        conn.logout().await.unwrap();
        return;
    }

    let mbox = format!("test-stalwart-examine-condstore-{}", std::process::id());
    conn.create(&mbox, TIMEOUT).await.unwrap();
    append_messages(
        &conn,
        &mbox,
        "Stalwart EXAMINE CONDSTORE",
        "stalwart-examine-condstore",
        1,
    )
    .await;

    let examined = conn
        .examine_with(
            &mbox,
            &daaki_imap::types::SelectOptions::condstore(),
            TIMEOUT,
        )
        .await
        .unwrap();

    assert!(
        examined.read_only,
        "EXAMINE must keep the mailbox read-only"
    );
    assert_eq!(
        examined.exists, 1,
        "EXAMINE should report the appended message"
    );
    assert!(
        examined.highest_mod_seq.is_some(),
        "EXAMINE CONDSTORE must include HIGHESTMODSEQ"
    );
    assert!(
        !examined.no_mod_seq,
        "EXAMINE CONDSTORE must not report NOMODSEQ when CONDSTORE is active"
    );

    conn.close(TIMEOUT).await.unwrap();
    conn.delete(&mbox, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

/// `fetch_changed_since` must return only messages whose mod-sequence advanced
/// after the supplied baseline.
///
/// RFC 7162 Section 3.1.4 defines `CHANGEDSINCE` as a filter over the
/// requested FETCH set, returning only messages with a greater mod-sequence.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_fetch_changed_since_by_sequence() {
    let conn = connect_stalwart().await;

    if !has_capability(&conn, &Capability::Condstore) {
        eprintln!("skipping: Stalwart does not advertise CONDSTORE");
        conn.logout().await.unwrap();
        return;
    }

    let mbox = format!(
        "test-stalwart-fetch-changed-since-seq-{}",
        std::process::id()
    );
    conn.create(&mbox, TIMEOUT).await.unwrap();
    append_messages(
        &conn,
        &mbox,
        "Stalwart FETCH CHANGEDSINCE",
        "stalwart-fetch-changed-since-seq",
        3,
    )
    .await;

    let baseline = select_condstore(&conn, &mbox)
        .await
        .highest_mod_seq
        .expect("SELECT CONDSTORE must return HIGHESTMODSEQ");

    let unchanged = conn
        .fetch_changed_since(
            &SequenceSet::new("1:*").unwrap(),
            &[FetchAttr::Uid, FetchAttr::Flags, FetchAttr::ModSeq],
            baseline,
            TIMEOUT,
        )
        .await
        .unwrap();
    assert!(
        unchanged.is_empty(),
        "FETCH CHANGEDSINCE at the current HIGHESTMODSEQ must be empty: {unchanged:?}"
    );

    conn.store(
        &SequenceSet::new("2").unwrap(),
        daaki_imap::types::StoreOperation::Add,
        &[Flag::Flagged],
        None,
        TIMEOUT,
    )
    .await
    .unwrap();

    let changed = conn
        .fetch_changed_since(
            &SequenceSet::new("1:*").unwrap(),
            &[FetchAttr::Uid, FetchAttr::Flags, FetchAttr::ModSeq],
            baseline,
            TIMEOUT,
        )
        .await
        .unwrap();

    assert_eq!(
        changed.len(),
        1,
        "FETCH CHANGEDSINCE must return only the modified message: {changed:?}"
    );
    assert_eq!(changed[0].seq, 2);
    assert!(
        changed[0]
            .flags
            .as_ref()
            .is_some_and(|flags| flags.contains(&Flag::Flagged)),
        "FETCH CHANGEDSINCE must surface the updated flags: {:?}",
        changed[0].flags
    );
    assert!(
        changed[0].mod_seq.is_some_and(|mod_seq| mod_seq > baseline),
        "FETCH CHANGEDSINCE must return a MODSEQ greater than the baseline: {:?}",
        changed[0].mod_seq
    );

    conn.close(TIMEOUT).await.unwrap();
    conn.delete(&mbox, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

/// `fetch_changed_since` must not leak changes outside the requested
/// sequence-number set.
///
/// RFC 7162 Section 3.1.4 scopes `CHANGEDSINCE` to the supplied FETCH
/// sequence set, so only changed messages inside that set may be returned.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_fetch_changed_since_limits_results_to_requested_sequence_set() {
    let conn = connect_stalwart().await;

    if !has_capability(&conn, &Capability::Condstore) {
        eprintln!("skipping: Stalwart does not advertise CONDSTORE");
        conn.logout().await.unwrap();
        return;
    }

    let mbox = format!(
        "test-stalwart-fetch-changed-since-subset-{}",
        std::process::id()
    );
    conn.create(&mbox, TIMEOUT).await.unwrap();
    append_messages(
        &conn,
        &mbox,
        "Stalwart FETCH CHANGEDSINCE subset",
        "stalwart-fetch-changed-since-subset",
        3,
    )
    .await;

    let baseline = select_condstore(&conn, &mbox)
        .await
        .highest_mod_seq
        .expect("SELECT CONDSTORE must return HIGHESTMODSEQ");

    conn.store(
        &SequenceSet::new("1").unwrap(),
        daaki_imap::types::StoreOperation::Add,
        &[Flag::Flagged],
        None,
        TIMEOUT,
    )
    .await
    .unwrap();
    conn.store(
        &SequenceSet::new("3").unwrap(),
        daaki_imap::types::StoreOperation::Add,
        &[Flag::Seen],
        None,
        TIMEOUT,
    )
    .await
    .unwrap();

    let changed = conn
        .fetch_changed_since(
            &SequenceSet::new("2:3").unwrap(),
            &[FetchAttr::Uid, FetchAttr::Flags, FetchAttr::ModSeq],
            baseline,
            TIMEOUT,
        )
        .await
        .unwrap();

    assert_eq!(
        changed.len(),
        1,
        "FETCH CHANGEDSINCE must return only changed messages inside 2:3: {changed:?}"
    );
    assert_eq!(changed[0].seq, 3);
    assert!(
        changed[0]
            .flags
            .as_ref()
            .is_some_and(|flags| flags.contains(&Flag::Seen)),
        "FETCH CHANGEDSINCE must include the in-range flag update: {:?}",
        changed[0].flags
    );
    assert!(
        changed[0].mod_seq.is_some_and(|mod_seq| mod_seq > baseline),
        "FETCH CHANGEDSINCE must report an updated MODSEQ for the in-range message: {:?}",
        changed[0].mod_seq
    );

    conn.close(TIMEOUT).await.unwrap();
    conn.delete(&mbox, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

/// `fetch_changed_since` must return no rows when only out-of-set sequence
/// numbers changed after the baseline.
///
/// RFC 7162 Section 3.1.4 limits `CHANGEDSINCE` results to the requested FETCH
/// sequence set, so unrelated changed messages must not be returned.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_fetch_changed_since_returns_empty_when_only_out_of_set_messages_changed() {
    let conn = connect_stalwart().await;

    if !has_capability(&conn, &Capability::Condstore) {
        eprintln!("skipping: Stalwart does not advertise CONDSTORE");
        conn.logout().await.unwrap();
        return;
    }

    let mbox = format!(
        "test-stalwart-fetch-changed-since-out-of-set-{}",
        std::process::id()
    );
    conn.create(&mbox, TIMEOUT).await.unwrap();
    append_messages(
        &conn,
        &mbox,
        "Stalwart FETCH CHANGEDSINCE out-of-set",
        "stalwart-fetch-changed-since-out-of-set",
        3,
    )
    .await;

    let baseline = select_condstore(&conn, &mbox)
        .await
        .highest_mod_seq
        .expect("SELECT CONDSTORE must return HIGHESTMODSEQ");

    conn.store(
        &SequenceSet::new("3").unwrap(),
        daaki_imap::types::StoreOperation::Add,
        &[Flag::Flagged],
        None,
        TIMEOUT,
    )
    .await
    .unwrap();

    let changed = conn
        .fetch_changed_since(
            &SequenceSet::new("1:2").unwrap(),
            &[FetchAttr::Uid, FetchAttr::Flags],
            baseline,
            TIMEOUT,
        )
        .await
        .unwrap();

    assert!(
        changed.is_empty(),
        "FETCH CHANGEDSINCE must not return changes outside 1:2: {changed:?}"
    );

    conn.close(TIMEOUT).await.unwrap();
    conn.delete(&mbox, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

/// `uid_fetch_changed_since` must return only UIDs whose mod-sequence
/// advanced after the supplied baseline.
///
/// RFC 7162 Section 3.1.4 defines `CHANGEDSINCE` for FETCH, and RFC 3501
/// Section 6.4.8 applies `UID` to FETCH while preserving that behavior.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_uid_fetch_changed_since_filters_only_modified_messages() {
    let conn = connect_stalwart().await;

    if !has_capability(&conn, &Capability::Condstore) {
        eprintln!("skipping: Stalwart does not advertise CONDSTORE");
        conn.logout().await.unwrap();
        return;
    }

    let mbox = format!(
        "test-stalwart-uid-fetch-changed-since-{}",
        std::process::id()
    );
    conn.create(&mbox, TIMEOUT).await.unwrap();
    append_messages(
        &conn,
        &mbox,
        "Stalwart UID FETCH CHANGEDSINCE",
        "stalwart-uid-fetch-changed-since",
        3,
    )
    .await;

    let baseline = select_condstore(&conn, &mbox)
        .await
        .highest_mod_seq
        .expect("SELECT CONDSTORE must return HIGHESTMODSEQ");
    let uids = sorted_uids(&conn).await;
    let changed_uid = uids[1];
    let all_uid_set = SequenceSet::new(format!(
        "{}:{}",
        uids.first().copied().unwrap_or(1),
        uids.last().copied().unwrap_or(1)
    ))
    .unwrap();

    conn.uid_store(
        &SequenceSet::new(changed_uid.to_string()).unwrap(),
        daaki_imap::types::StoreOperation::Add,
        &[Flag::Flagged],
        None,
        TIMEOUT,
    )
    .await
    .unwrap();

    let changed = conn
        .uid_fetch_changed_since(
            &all_uid_set,
            &[FetchAttr::Uid, FetchAttr::Flags, FetchAttr::ModSeq],
            baseline,
            TIMEOUT,
        )
        .await
        .unwrap();

    assert_eq!(
        changed.len(),
        1,
        "UID FETCH CHANGEDSINCE must return only the modified UID: {changed:?}"
    );
    assert_eq!(changed[0].uid, Some(changed_uid));
    assert!(
        changed[0]
            .flags
            .as_ref()
            .is_some_and(|flags| flags.contains(&Flag::Flagged)),
        "UID FETCH CHANGEDSINCE must surface the updated flags: {:?}",
        changed[0].flags
    );
    assert!(
        changed[0].mod_seq.is_some_and(|mod_seq| mod_seq > baseline),
        "UID FETCH CHANGEDSINCE must report a MODSEQ greater than the baseline: {:?}",
        changed[0].mod_seq
    );

    conn.close(TIMEOUT).await.unwrap();
    conn.delete(&mbox, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

/// `uid_fetch_changed_since` must not return changed UIDs outside the
/// requested UID set.
///
/// RFC 7162 Section 3.1.4 filters FETCH results to the supplied set, and
/// RFC 3501 Section 6.4.8 applies the same rule to UID FETCH.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_uid_fetch_changed_since_limits_results_to_requested_uid_set() {
    let conn = connect_stalwart().await;

    if !has_capability(&conn, &Capability::Condstore) {
        eprintln!("skipping: Stalwart does not advertise CONDSTORE");
        conn.logout().await.unwrap();
        return;
    }

    let mbox = format!(
        "test-stalwart-uid-fetch-changed-since-subset-{}",
        std::process::id()
    );
    conn.create(&mbox, TIMEOUT).await.unwrap();
    append_messages(
        &conn,
        &mbox,
        "Stalwart UID FETCH CHANGEDSINCE subset",
        "stalwart-uid-fetch-changed-since-subset",
        3,
    )
    .await;

    let baseline = select_condstore(&conn, &mbox)
        .await
        .highest_mod_seq
        .expect("SELECT CONDSTORE must return HIGHESTMODSEQ");
    let uids = sorted_uids(&conn).await;

    conn.uid_store(
        &SequenceSet::new(uids[0].to_string()).unwrap(),
        daaki_imap::types::StoreOperation::Add,
        &[Flag::Flagged],
        None,
        TIMEOUT,
    )
    .await
    .unwrap();
    conn.uid_store(
        &SequenceSet::new(uids[2].to_string()).unwrap(),
        daaki_imap::types::StoreOperation::Add,
        &[Flag::Seen],
        None,
        TIMEOUT,
    )
    .await
    .unwrap();

    let requested_uid_set = SequenceSet::new(format!("{},{}", uids[1], uids[2])).unwrap();
    let changed = conn
        .uid_fetch_changed_since(
            &requested_uid_set,
            &[FetchAttr::Uid, FetchAttr::Flags, FetchAttr::ModSeq],
            baseline,
            TIMEOUT,
        )
        .await
        .unwrap();

    assert_eq!(
        changed.len(),
        1,
        "UID FETCH CHANGEDSINCE must return only changed UIDs inside the requested set: {changed:?}"
    );
    assert_eq!(changed[0].uid, Some(uids[2]));
    assert!(
        changed[0]
            .flags
            .as_ref()
            .is_some_and(|flags| flags.contains(&Flag::Seen)),
        "UID FETCH CHANGEDSINCE must include the in-range flag update: {:?}",
        changed[0].flags
    );
    assert!(
        changed[0].mod_seq.is_some_and(|mod_seq| mod_seq > baseline),
        "UID FETCH CHANGEDSINCE must report an updated MODSEQ for the in-range UID: {:?}",
        changed[0].mod_seq
    );

    conn.close(TIMEOUT).await.unwrap();
    conn.delete(&mbox, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

/// `uid_fetch_changed_since` must return no rows when only out-of-set UIDs
/// changed after the baseline.
///
/// RFC 7162 Section 3.1.4 narrows FETCH results to the supplied set, and
/// RFC 3501 Section 6.4.8 preserves that behavior for UID FETCH.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_uid_fetch_changed_since_returns_empty_when_only_out_of_set_messages_changed() {
    let conn = connect_stalwart().await;

    if !has_capability(&conn, &Capability::Condstore) {
        eprintln!("skipping: Stalwart does not advertise CONDSTORE");
        conn.logout().await.unwrap();
        return;
    }

    let mbox = format!(
        "test-stalwart-uid-fetch-changed-since-out-of-set-{}",
        std::process::id()
    );
    conn.create(&mbox, TIMEOUT).await.unwrap();
    append_messages(
        &conn,
        &mbox,
        "Stalwart UID FETCH CHANGEDSINCE out-of-set",
        "stalwart-uid-fetch-changed-since-out-of-set",
        3,
    )
    .await;

    let baseline = select_condstore(&conn, &mbox)
        .await
        .highest_mod_seq
        .expect("SELECT CONDSTORE must return HIGHESTMODSEQ");
    let uids = sorted_uids(&conn).await;

    conn.uid_store(
        &SequenceSet::new(uids[2].to_string()).unwrap(),
        daaki_imap::types::StoreOperation::Add,
        &[Flag::Flagged],
        None,
        TIMEOUT,
    )
    .await
    .unwrap();

    let requested_uid_set = SequenceSet::new(format!("{},{}", uids[0], uids[1])).unwrap();
    let changed = conn
        .uid_fetch_changed_since(
            &requested_uid_set,
            &[FetchAttr::Uid, FetchAttr::Flags],
            baseline,
            TIMEOUT,
        )
        .await
        .unwrap();

    assert!(
        changed.is_empty(),
        "UID FETCH CHANGEDSINCE must not return changes outside the requested UID set: {changed:?}"
    );

    conn.close(TIMEOUT).await.unwrap();
    conn.delete(&mbox, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

// ==========================================================================
// UID FETCH VANISHED (RFC 7162 Section 3.2.6)
// ==========================================================================

/// `uid_fetch_vanished` must return no FETCH or VANISHED data when called at
/// the current mod-sequence with no intervening changes.
///
/// RFC 7162 Section 3.2.6 limits `UID FETCH ... VANISHED` to expunges and flag
/// changes since the supplied mod-sequence, so the current `HIGHESTMODSEQ`
/// baseline must produce an empty delta.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_uid_fetch_vanished_at_current_modseq_returns_no_changes() {
    let conn = connect_stalwart().await;

    if !enable_condstore_and_qresync(&conn).await {
        conn.logout().await.unwrap();
        return;
    }

    let mbox = unique_name("test-stalwart-uid-fetch-vanished-baseline");
    conn.create(&mbox, TIMEOUT).await.unwrap();
    append_messages(
        &conn,
        &mbox,
        "Stalwart UID FETCH VANISHED baseline",
        "stalwart-uid-fetch-vanished-baseline",
        3,
    )
    .await;

    let baseline = select_condstore(&conn, &mbox)
        .await
        .highest_mod_seq
        .expect("SELECT CONDSTORE must return HIGHESTMODSEQ");
    let uids = sorted_uids(&conn).await;
    let requested = SequenceSet::new(format!(
        "{}:{}",
        uids.first().copied().unwrap_or(1),
        uids.last().copied().unwrap_or(1)
    ))
    .unwrap();

    let (fetches, vanished) = conn
        .uid_fetch_vanished(
            &requested,
            &[FetchAttr::Uid, FetchAttr::Flags],
            baseline,
            TIMEOUT,
        )
        .await
        .unwrap();

    assert!(
        fetches.is_empty(),
        "UID FETCH VANISHED at the current mod-sequence must not return FETCH rows: {fetches:?}"
    );
    assert!(
        vanished.is_empty(),
        "UID FETCH VANISHED at the current mod-sequence must not report VANISHED UIDs: {vanished:?}"
    );

    delete_mailbox_and_contents(&conn, &mbox).await;
    conn.logout().await.unwrap();
}

/// `uid_fetch_vanished` must surface expunged UIDs from the requested set.
///
/// RFC 7162 Section 3.2.6 requires `UID FETCH ... VANISHED` to return
/// `VANISHED (EARLIER)` data for requested UIDs that were expunged after the
/// supplied mod-sequence.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_uid_fetch_vanished_after_expunge() {
    let conn = connect_stalwart().await;

    if !enable_condstore_and_qresync(&conn).await {
        conn.logout().await.unwrap();
        return;
    }

    let mbox = unique_name("test-stalwart-uid-fetch-vanished-expunge");
    conn.create(&mbox, TIMEOUT).await.unwrap();
    append_messages(
        &conn,
        &mbox,
        "Stalwart UID FETCH VANISHED expunge",
        "stalwart-uid-fetch-vanished-expunge",
        3,
    )
    .await;

    let baseline = select_condstore(&conn, &mbox)
        .await
        .highest_mod_seq
        .expect("SELECT CONDSTORE must return HIGHESTMODSEQ");
    let uids = sorted_uids(&conn).await;
    let expunged_uid = uids[1];

    conn.uid_store(
        &SequenceSet::new(expunged_uid.to_string()).unwrap(),
        daaki_imap::types::StoreOperation::Add,
        &[Flag::Deleted],
        None,
        TIMEOUT,
    )
    .await
    .unwrap();
    conn.expunge(TIMEOUT).await.unwrap();

    let requested = SequenceSet::new(format!(
        "{}:{}",
        uids.first().copied().unwrap_or(1),
        uids.last().copied().unwrap_or(1)
    ))
    .unwrap();
    let (fetches, vanished) = conn
        .uid_fetch_vanished(
            &requested,
            &[FetchAttr::Uid, FetchAttr::Flags],
            baseline,
            TIMEOUT,
        )
        .await
        .unwrap();
    let mut vanished_uids = expand_uid_ranges(&vanished);
    vanished_uids.sort_unstable();

    assert!(
        fetches.is_empty(),
        "expunging one message must not synthesize unrelated FETCH rows: {fetches:?}"
    );
    assert_eq!(
        vanished_uids,
        vec![expunged_uid],
        "UID FETCH VANISHED must report the expunged UID"
    );

    delete_mailbox_and_contents(&conn, &mbox).await;
    conn.logout().await.unwrap();
}

/// `uid_fetch_vanished` must not report expunges outside the requested UID
/// set.
///
/// RFC 7162 Section 3.2.6 scopes `VANISHED (EARLIER)` to the requested UID
/// set, so expunged UIDs outside that set must be omitted.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_uid_fetch_vanished_limits_results_to_requested_uid_set() {
    let conn = connect_stalwart().await;

    if !enable_condstore_and_qresync(&conn).await {
        conn.logout().await.unwrap();
        return;
    }

    let mbox = unique_name("test-stalwart-uid-fetch-vanished-subset");
    conn.create(&mbox, TIMEOUT).await.unwrap();
    append_messages(
        &conn,
        &mbox,
        "Stalwart UID FETCH VANISHED subset",
        "stalwart-uid-fetch-vanished-subset",
        4,
    )
    .await;

    let baseline = select_condstore(&conn, &mbox)
        .await
        .highest_mod_seq
        .expect("SELECT CONDSTORE must return HIGHESTMODSEQ");
    let uids = sorted_uids(&conn).await;

    conn.uid_store(
        &SequenceSet::new(format!("{},{}", uids[0], uids[3])).unwrap(),
        daaki_imap::types::StoreOperation::Add,
        &[Flag::Deleted],
        None,
        TIMEOUT,
    )
    .await
    .unwrap();
    conn.expunge(TIMEOUT).await.unwrap();

    let requested = SequenceSet::new(format!("{},{}", uids[0], uids[1])).unwrap();
    let (fetches, vanished) = conn
        .uid_fetch_vanished(
            &requested,
            &[FetchAttr::Uid, FetchAttr::Flags],
            baseline,
            TIMEOUT,
        )
        .await
        .unwrap();
    let mut vanished_uids = expand_uid_ranges(&vanished);
    vanished_uids.sort_unstable();

    assert!(
        fetches.is_empty(),
        "subset query with only expunges must not return FETCH rows: {fetches:?}"
    );
    assert_eq!(
        vanished_uids,
        vec![uids[0]],
        "UID FETCH VANISHED must report only expunges inside the requested UID set"
    );

    delete_mailbox_and_contents(&conn, &mbox).await;
    conn.logout().await.unwrap();
}

/// `uid_fetch_vanished` must return no VANISHED data when only out-of-set UIDs
/// were expunged after the baseline.
///
/// RFC 7162 Section 3.2.6 narrows `VANISHED (EARLIER)` to the requested UID
/// set, so unrelated expunges must not leak into the result.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_uid_fetch_vanished_returns_empty_when_only_out_of_set_uids_were_expunged() {
    let conn = connect_stalwart().await;

    if !enable_condstore_and_qresync(&conn).await {
        conn.logout().await.unwrap();
        return;
    }

    let mbox = unique_name("test-stalwart-uid-fetch-vanished-out-of-set");
    conn.create(&mbox, TIMEOUT).await.unwrap();
    append_messages(
        &conn,
        &mbox,
        "Stalwart UID FETCH VANISHED out-of-set",
        "stalwart-uid-fetch-vanished-out-of-set",
        3,
    )
    .await;

    let baseline = select_condstore(&conn, &mbox)
        .await
        .highest_mod_seq
        .expect("SELECT CONDSTORE must return HIGHESTMODSEQ");
    let uids = sorted_uids(&conn).await;

    conn.uid_store(
        &SequenceSet::new(uids[2].to_string()).unwrap(),
        daaki_imap::types::StoreOperation::Add,
        &[Flag::Deleted],
        None,
        TIMEOUT,
    )
    .await
    .unwrap();
    conn.expunge(TIMEOUT).await.unwrap();

    let requested = SequenceSet::new(format!("{},{}", uids[0], uids[1])).unwrap();
    let (fetches, vanished) = conn
        .uid_fetch_vanished(
            &requested,
            &[FetchAttr::Uid, FetchAttr::Flags],
            baseline,
            TIMEOUT,
        )
        .await
        .unwrap();

    assert!(
        fetches.is_empty(),
        "out-of-set expunges must not produce FETCH rows for the requested UID set: {fetches:?}"
    );
    assert!(
        vanished.is_empty(),
        "out-of-set expunges must not be reported in VANISHED data: {vanished:?}"
    );

    delete_mailbox_and_contents(&conn, &mbox).await;
    conn.logout().await.unwrap();
}

/// `uid_fetch_vanished` must return flag-change FETCH rows and VANISHED UIDs
/// together when both happened after the baseline.
///
/// RFC 7162 Section 3.2.6 requires the command to surface both categories of
/// delta: FETCH data for changed surviving messages and VANISHED data for
/// expunged requested UIDs.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_uid_fetch_vanished_reports_fetches_and_vanished_uids_together() {
    let conn = connect_stalwart().await;

    if !enable_condstore_and_qresync(&conn).await {
        conn.logout().await.unwrap();
        return;
    }

    let mbox = unique_name("test-stalwart-uid-fetch-vanished-mixed");
    conn.create(&mbox, TIMEOUT).await.unwrap();
    append_messages(
        &conn,
        &mbox,
        "Stalwart UID FETCH VANISHED mixed",
        "stalwart-uid-fetch-vanished-mixed",
        3,
    )
    .await;

    let baseline = select_condstore(&conn, &mbox)
        .await
        .highest_mod_seq
        .expect("SELECT CONDSTORE must return HIGHESTMODSEQ");
    let uids = sorted_uids(&conn).await;

    conn.uid_store(
        &SequenceSet::new(uids[0].to_string()).unwrap(),
        daaki_imap::types::StoreOperation::Add,
        &[Flag::Flagged],
        None,
        TIMEOUT,
    )
    .await
    .unwrap();
    conn.uid_store(
        &SequenceSet::new(uids[2].to_string()).unwrap(),
        daaki_imap::types::StoreOperation::Add,
        &[Flag::Deleted],
        None,
        TIMEOUT,
    )
    .await
    .unwrap();
    conn.expunge(TIMEOUT).await.unwrap();

    let requested = SequenceSet::new(format!(
        "{}:{}",
        uids.first().copied().unwrap_or(1),
        uids.last().copied().unwrap_or(1)
    ))
    .unwrap();
    let (fetches, vanished) = conn
        .uid_fetch_vanished(
            &requested,
            &[FetchAttr::Uid, FetchAttr::Flags, FetchAttr::ModSeq],
            baseline,
            TIMEOUT,
        )
        .await
        .unwrap();
    let mut vanished_uids = expand_uid_ranges(&vanished);
    vanished_uids.sort_unstable();

    assert_eq!(
        fetches.len(),
        1,
        "UID FETCH VANISHED must return exactly one surviving changed message: {fetches:?}"
    );
    assert_eq!(fetches[0].uid, Some(uids[0]));
    assert!(
        fetches[0]
            .flags
            .as_ref()
            .is_some_and(|flags| flags.contains(&Flag::Flagged)),
        "surviving changed message must include the updated \\Flagged flag: {:?}",
        fetches[0].flags
    );
    assert!(
        fetches[0].mod_seq.is_some_and(|mod_seq| mod_seq > baseline),
        "surviving changed message must report a mod-sequence greater than the baseline: {:?}",
        fetches[0].mod_seq
    );
    assert_eq!(
        vanished_uids,
        vec![uids[2]],
        "UID FETCH VANISHED must also report the expunged UID"
    );

    delete_mailbox_and_contents(&conn, &mbox).await;
    conn.logout().await.unwrap();
}

// ==========================================================================
// ID (RFC 2971)
// ==========================================================================

#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_id() {
    let conn = connect_stalwart().await;

    match conn
        .id(
            &[("name", Some("daaki-test")), ("version", Some("0.1.0"))],
            TIMEOUT,
        )
        .await
    {
        Ok(server_id) => {
            assert!(
                !server_id.is_empty(),
                "expected server to return ID information"
            );
        }
        Err(daaki_imap::Error::MissingCapability(_)) => {
            eprintln!("skipping: Stalwart does not advertise ID capability");
        }
        Err(e) => panic!("unexpected error from ID: {e:?}"),
    }

    conn.logout().await.unwrap();
}

/// ID on a selected mailbox must not deselect the mailbox.
///
/// RFC 2971 Section 3.1 allows ID in any state, and RFC 3501 Section 6.4
/// requires selected-state commands to leave the selected mailbox intact.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_id_selected_preserves_selected_mailbox() {
    let conn = connect_stalwart().await;

    if !has_capability(&conn, &Capability::Id) {
        eprintln!("skipping: Stalwart does not advertise ID capability");
        conn.logout().await.unwrap();
        return;
    }

    let mailbox = unique_name("test-stalwart-id-selected");
    conn.create(&mailbox, TIMEOUT).await.unwrap();
    conn.append(&mailbox, &[], None, TEST_MESSAGE, TIMEOUT)
        .await
        .unwrap();
    conn.select(&mailbox, TIMEOUT).await.unwrap();

    conn.id(
        &[("name", Some("daaki-test")), ("version", Some("0.1.0"))],
        TIMEOUT,
    )
    .await
    .unwrap();
    assert_selected_mailbox_still_usable(&conn, "ID").await;

    conn.close(TIMEOUT).await.unwrap();
    conn.delete(&mailbox, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

// ==========================================================================
// IDLE (RFC 2177 / RFC 9051 base)
// ==========================================================================

/// Cancelling IDLE must eventually surface `IdleEvent::Cancelled`.
///
/// RFC 2177 Section 3 allows the server to send unrelated untagged responses
/// while the client is idling, so this shared-account integration test keeps
/// retrying through stray notifications from other parallel tests until the
/// cancellation token is observed.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_idle_cancel() {
    let conn = connect_stalwart().await;
    conn.select("INBOX", TIMEOUT).await.unwrap();

    // Flush server-side pending notifications (NOOP round-trip) then
    // drain the event queue. Stalwart delivers unsolicited LIST
    // responses for mailbox changes from other tests asynchronously —
    // the NOOP forces them into the queue, and drain_events clears them.
    conn.noop(TIMEOUT).await.unwrap();
    conn.drain_events().await;

    let cancel = CancellationToken::new();
    let cancel_clone = cancel.clone();

    tokio::spawn(async move {
        tokio::time::sleep(Duration::from_millis(500)).await;
        cancel_clone.cancel();
    });

    let deadline = Instant::now() + Duration::from_secs(5);
    loop {
        let event = conn.idle(TIMEOUT, cancel.clone()).await.unwrap();
        if event == IdleEvent::Cancelled {
            break;
        }
        assert!(
            Instant::now() < deadline,
            "expected IDLE cancellation, got only unsolicited events: {event:?}"
        );
    }

    conn.logout().await.unwrap();
}

/// With no relevant server response, IDLE must eventually return
/// `IdleEvent::Timeout`.
///
/// RFC 2177 Section 3 allows servers to send untagged responses while the
/// client is idling, so this test tolerates stray unsolicited events from
/// other parallel tests and keeps retrying until an actual timeout occurs.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_idle_timeout() {
    let conn = connect_stalwart().await;
    conn.select("INBOX", TIMEOUT).await.unwrap();

    // Flush server-side pending notifications (NOOP round-trip) then
    // drain the event queue. Stalwart delivers unsolicited LIST
    // responses for mailbox changes from other tests asynchronously —
    // the NOOP forces them into the queue, and drain_events clears them.
    conn.noop(TIMEOUT).await.unwrap();
    conn.drain_events().await;

    let cancel = CancellationToken::new();
    let deadline = Instant::now() + Duration::from_secs(12);
    loop {
        let event = conn
            .idle(Duration::from_secs(2), cancel.clone())
            .await
            .unwrap();
        if event == IdleEvent::Timeout {
            break;
        }
        assert!(
            Instant::now() < deadline,
            "expected IDLE timeout, got only unsolicited events: {event:?}"
        );
    }

    conn.logout().await.unwrap();
}

// ==========================================================================
// COMPRESS=DEFLATE (RFC 4978)
// ==========================================================================

#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_compress() {
    let conn = connect_stalwart().await;

    match conn.compress(TIMEOUT).await {
        Ok(()) => {
            // Verify operations work over compressed connection.
            let mailboxes = conn.list("", "*", TIMEOUT).await.unwrap();
            assert!(
                mailboxes
                    .iter()
                    .any(|m| m.name.as_str().eq_ignore_ascii_case("INBOX")),
                "LIST over compressed connection should still find INBOX"
            );
            conn.noop(TIMEOUT).await.unwrap();
        }
        Err(daaki_imap::Error::MissingCapability(_)) => {
            eprintln!("skipping: Stalwart does not advertise COMPRESS=DEFLATE");
        }
        Err(e) => panic!("unexpected error from COMPRESS: {e:?}"),
    }

    conn.logout().await.unwrap();
}

// ==========================================================================
// STATUS (RFC 3501 Section 6.3.10 / RFC 9051 Section 6.3.11)
// ==========================================================================

#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_status() {
    let conn = connect_stalwart().await;

    let items = conn
        .status("INBOX", "(MESSAGES UNSEEN UIDNEXT UIDVALIDITY)", TIMEOUT)
        .await
        .unwrap()
        .items;

    assert!(!items.is_empty(), "STATUS should return at least one item");

    conn.logout().await.unwrap();
}

/// STATUS with `DELETED-STORAGE` must report positive octets once a message is
/// marked `\Deleted` (RFC 9208 Section 3).
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_rev2_status_deleted_storage_reports_deleted_octets() {
    let conn = connect_stalwart().await;

    if !has_quota_resource(&conn, "STORAGE") {
        eprintln!("skipping: QUOTA=RES-STORAGE not advertised");
        conn.logout().await.unwrap();
        return;
    }

    let mbox = unique_name("test-status-deleted-storage");
    let message = build_message(
        "Stalwart deleted storage",
        &format!("{mbox}@daaki"),
        &"x".repeat(4096),
    );

    conn.create(&mbox, TIMEOUT).await.unwrap();
    conn.append(&mbox, &[], None, &message, TIMEOUT)
        .await
        .unwrap();
    conn.select(&mbox, TIMEOUT).await.unwrap();
    conn.store(
        &SequenceSet::new("1").unwrap(),
        daaki_imap::types::StoreOperation::Add,
        &[Flag::Deleted],
        None,
        TIMEOUT,
    )
    .await
    .unwrap();

    let status = conn
        .status(&mbox, "(DELETED-STORAGE)", TIMEOUT)
        .await
        .unwrap();
    let deleted_storage = status_deleted_storage(&status.items)
        .expect("STATUS DELETED-STORAGE must include the requested item");
    assert!(
        deleted_storage > 0,
        "deleted message octets must be positive, got {deleted_storage}: {:?}",
        status.items
    );

    delete_mailbox_and_contents(&conn, &mbox).await;
    conn.logout().await.unwrap();
}

/// STATUS with `DELETED-STORAGE` must return to zero after the deleted
/// message has been expunged (RFC 9208 Section 3 / RFC 3501 Section 6.4.3).
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_rev2_status_deleted_storage_clears_after_expunge() {
    let conn = connect_stalwart().await;

    if !has_quota_resource(&conn, "STORAGE") {
        eprintln!("skipping: QUOTA=RES-STORAGE not advertised");
        conn.logout().await.unwrap();
        return;
    }

    let mbox = unique_name("test-status-deleted-storage-expunge");
    let message = build_message(
        "Stalwart deleted storage expunge",
        &format!("{mbox}@daaki"),
        &"z".repeat(4096),
    );

    conn.create(&mbox, TIMEOUT).await.unwrap();
    conn.append(&mbox, &[], None, &message, TIMEOUT)
        .await
        .unwrap();
    conn.select(&mbox, TIMEOUT).await.unwrap();
    conn.store(
        &SequenceSet::new("1").unwrap(),
        daaki_imap::types::StoreOperation::Add,
        &[Flag::Deleted],
        None,
        TIMEOUT,
    )
    .await
    .unwrap();

    let before = conn
        .status(&mbox, "(MESSAGES DELETED-STORAGE)", TIMEOUT)
        .await
        .unwrap();
    let deleted_storage_before =
        status_deleted_storage(&before.items).expect("STATUS DELETED-STORAGE must be present");
    assert!(
        deleted_storage_before > 0,
        "deleted message octets must be positive before EXPUNGE: {:?}",
        before.items
    );

    conn.expunge(TIMEOUT).await.unwrap();

    let after = conn
        .status(&mbox, "(MESSAGES DELETED-STORAGE)", TIMEOUT)
        .await
        .unwrap();
    assert_eq!(
        status_messages(&after.items),
        Some(0),
        "EXPUNGE must remove the only message from the mailbox: {:?}",
        after.items
    );
    assert_eq!(
        status_deleted_storage(&after.items),
        Some(0),
        "DELETED-STORAGE must drop to zero after EXPUNGE: {:?}",
        after.items
    );

    conn.close(TIMEOUT).await.unwrap();
    conn.delete(&mbox, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

/// STATUS with `DELETED-STORAGE` must report zero octets when no message in
/// the mailbox is marked `\Deleted` (RFC 9208 Section 3).
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_rev2_status_deleted_storage_zero_without_deleted_messages() {
    let conn = connect_stalwart().await;

    if !has_quota_resource(&conn, "STORAGE") {
        eprintln!("skipping: QUOTA=RES-STORAGE not advertised");
        conn.logout().await.unwrap();
        return;
    }

    let mbox = unique_name("test-status-deleted-storage-zero");
    conn.create(&mbox, TIMEOUT).await.unwrap();
    conn.append(&mbox, &[], None, TEST_MESSAGE, TIMEOUT)
        .await
        .unwrap();

    let status = conn
        .status(&mbox, "(DELETED-STORAGE)", TIMEOUT)
        .await
        .unwrap();
    assert_eq!(
        status_deleted_storage(&status.items),
        Some(0),
        "mailboxes without deleted messages must report DELETED-STORAGE 0: {:?}",
        status.items
    );

    delete_mailbox_and_contents(&conn, &mbox).await;
    conn.logout().await.unwrap();
}

/// STATUS with `SIZE` must increase when additional message octets are
/// appended to the mailbox (RFC 9051 Section 6.3.11 / RFC 8438 Section 3).
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_rev2_status_size_grows_after_append() {
    let conn = connect_stalwart().await;

    let mbox = unique_name("test-status-size-growth");
    let large_message = build_message(
        "Stalwart status size growth",
        &format!("{mbox}-large@daaki"),
        &"y".repeat(8192),
    );

    conn.create(&mbox, TIMEOUT).await.unwrap();
    conn.append(&mbox, &[], None, TEST_MESSAGE, TIMEOUT)
        .await
        .unwrap();

    let before = conn
        .status(&mbox, "(MESSAGES SIZE)", TIMEOUT)
        .await
        .unwrap();
    let size_before = status_size(&before.items).expect("STATUS SIZE must be present");
    assert_eq!(status_messages(&before.items), Some(1));

    conn.append(&mbox, &[], None, &large_message, TIMEOUT)
        .await
        .unwrap();

    let after = conn
        .status(&mbox, "(MESSAGES SIZE)", TIMEOUT)
        .await
        .unwrap();
    let size_after = status_size(&after.items).expect("STATUS SIZE must be present");
    assert_eq!(status_messages(&after.items), Some(2));
    assert!(
        size_after > size_before,
        "mailbox SIZE must grow after appending more octets ({size_before} -> {size_after})"
    );

    delete_mailbox_and_contents(&conn, &mbox).await;
    conn.logout().await.unwrap();
}

/// STATUS on a selected mailbox must not clear the selected state, and
/// without NOTIFY STATUS delivery the result must remain unambiguous
/// (RFC 3501 Section 6.3.10 / RFC 5465 Section 4).
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_status_on_selected_mailbox_preserves_selected_state() {
    let conn = connect_stalwart().await;

    let mbox = unique_name("test-status-selected");
    conn.create(&mbox, TIMEOUT).await.unwrap();
    append_messages(&conn, &mbox, "status selected", "status-selected", 2).await;
    conn.select(&mbox, TIMEOUT).await.unwrap();

    let status = conn
        .status(&mbox, "(MESSAGES UIDNEXT)", TIMEOUT)
        .await
        .unwrap();
    assert_eq!(status_messages(&status.items), Some(2));
    assert!(
        status.ambiguous.is_empty(),
        "STATUS without NOTIFY STATUS delivery must be unambiguous: {:?}",
        status.ambiguous
    );

    let fetches = conn
        .fetch(
            &SequenceSet::new("1:*").unwrap(),
            &[FetchAttr::Uid],
            TIMEOUT,
        )
        .await
        .unwrap();
    assert_eq!(
        fetches.len(),
        2,
        "STATUS on a selected mailbox must not clear the selected state"
    );
    assert!(
        fetches.iter().all(|fetch| fetch.uid.is_some()),
        "subsequent FETCH after STATUS should still return UID data: {fetches:?}"
    );

    delete_mailbox_and_contents(&conn, &mbox).await;
    conn.logout().await.unwrap();
}

// ==========================================================================
// NOOP
// ==========================================================================

#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_noop() {
    let conn = connect_stalwart().await;
    conn.noop(TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

// ==========================================================================
// CREATE / DELETE / RENAME mailbox
// ==========================================================================

#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_create_delete_mailbox() {
    let conn = connect_stalwart().await;

    let name = &format!("test-cd-{}", std::process::id());
    conn.create(name, TIMEOUT).await.unwrap();

    let mailboxes = conn.list("", name, TIMEOUT).await.unwrap();
    assert!(!mailboxes.is_empty(), "mailbox should exist after CREATE");

    conn.delete(name, TIMEOUT).await.unwrap();

    let mailboxes = conn.list("", name, TIMEOUT).await.unwrap();
    assert!(mailboxes.is_empty(), "mailbox should be gone after DELETE");

    conn.logout().await.unwrap();
}

/// CREATE with MAILBOXID must return the same identifier later exposed by
/// STATUS on an OBJECTID-capable server.
///
/// RFC 8474 Section 4.1 defines the `MAILBOXID` response code on CREATE, and
/// RFC 8474 Section 4.3 defines STATUS `MAILBOXID`.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_rev2_create_with_mailbox_id_matches_status_mailboxid() {
    let conn = connect_stalwart().await;

    if !has_capability(&conn, &Capability::ObjectId) {
        eprintln!("skipping: OBJECTID not advertised");
        conn.logout().await.unwrap();
        return;
    }

    let mbox = format!("test-create-mailboxid-{}", std::process::id());
    let created_mailbox_id = conn
        .create_with_mailbox_id(&mbox, TIMEOUT)
        .await
        .unwrap()
        .expect("OBJECTID-capable server must return MAILBOXID on CREATE");
    assert!(
        !created_mailbox_id.is_empty(),
        "MAILBOXID must not be empty"
    );

    let status_items = conn
        .status(&mbox, "(MAILBOXID)", TIMEOUT)
        .await
        .unwrap()
        .items;
    let status_mailbox_id = status_items.iter().find_map(|item| {
        if let daaki_imap::types::StatusItem::MailboxId(id) = item {
            Some(id.clone())
        } else {
            None
        }
    });

    assert_eq!(
        status_mailbox_id.as_deref(),
        Some(created_mailbox_id.as_str()),
        "CREATE MAILBOXID must match STATUS MAILBOXID"
    );

    conn.delete(&mbox, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

/// CREATE USE must assign the requested special-use role to the mailbox and
/// surface it through LIST.
///
/// RFC 6154 Section 3 defines CREATE's `USE` parameter, and RFC 6154 Section 2
/// defines the special-use mailbox attributes returned by LIST.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_rev2_create_special_use_mailbox() {
    let conn = connect_stalwart().await;

    if !has_capability(&conn, &Capability::CreateSpecialUse)
        || !has_capability(&conn, &Capability::SpecialUse)
    {
        eprintln!("skipping: CREATE-SPECIAL-USE or SPECIAL-USE not advertised");
        conn.logout().await.unwrap();
        return;
    }

    let _special_use_role_guard = special_use_role_lock().lock().await;
    let mbox = unique_name("test-create-special-use");
    let special_use = available_special_use_role(&conn).await;
    conn.create_special_use(&mbox, std::slice::from_ref(&special_use), TIMEOUT)
        .await
        .unwrap_or_else(|e| {
            panic!("unexpected error creating special-use mailbox with role {special_use:?}: {e}")
        });

    let listing = conn
        .list_extended("", &[&mbox], &[], &["SPECIAL-USE"], TIMEOUT)
        .await
        .unwrap();
    assert_eq!(listing.len(), 1, "created mailbox must appear in LIST");
    assert!(
        listing[0]
            .attributes
            .iter()
            .any(|attr| attr == &special_use),
        "created mailbox must advertise the requested special-use role {:?}: {:?}",
        special_use,
        listing[0].attributes
    );

    conn.delete(&mbox, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

/// `create_special_use_with_mailbox_id` must apply the requested special-use
/// role and return the same `MAILBOXID` later exposed by STATUS when OBJECTID
/// is advertised.
///
/// RFC 6154 Section 3 defines CREATE's `USE` parameter, and RFC 8474 Sections
/// 4.1 and 4.3 define `MAILBOXID` on CREATE and STATUS.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_rev2_create_special_use_with_mailbox_id_matches_status_mailboxid() {
    let conn = connect_stalwart().await;

    if !has_capability(&conn, &Capability::CreateSpecialUse)
        || !has_capability(&conn, &Capability::SpecialUse)
    {
        eprintln!("skipping: CREATE-SPECIAL-USE or SPECIAL-USE not advertised");
        conn.logout().await.unwrap();
        return;
    }

    let _special_use_role_guard = special_use_role_lock().lock().await;
    let mbox = unique_name("test-create-special-use-mailboxid");
    let requested_role = available_special_use_role(&conn).await;
    let mailbox_id = conn
        .create_special_use_with_mailbox_id(&mbox, std::slice::from_ref(&requested_role), TIMEOUT)
        .await
        .unwrap_or_else(|e| {
            panic!(
                "unexpected error creating special-use mailbox with MAILBOXID and role {requested_role:?}: {e}"
            )
        });

    if has_capability(&conn, &Capability::ObjectId) {
        let created_mailbox_id =
            mailbox_id.expect("OBJECTID-capable server must return MAILBOXID on CREATE");
        assert!(
            !created_mailbox_id.is_empty(),
            "MAILBOXID returned by CREATE must not be empty"
        );

        let status_items = conn
            .status(&mbox, "(MAILBOXID)", TIMEOUT)
            .await
            .unwrap()
            .items;
        let status_mailbox_id = status_items.iter().find_map(|item| {
            if let daaki_imap::types::StatusItem::MailboxId(id) = item {
                Some(id.as_str())
            } else {
                None
            }
        });
        assert_eq!(
            status_mailbox_id,
            Some(created_mailbox_id.as_str()),
            "CREATE MAILBOXID must match STATUS MAILBOXID"
        );
    } else {
        assert!(
            mailbox_id.is_none(),
            "without OBJECTID, MAILBOXID should be absent"
        );
    }

    let listing = conn
        .list_extended("", &[&mbox], &[], &["SPECIAL-USE"], TIMEOUT)
        .await
        .unwrap();
    assert_eq!(listing.len(), 1, "created mailbox must appear in LIST");
    assert!(
        listing[0]
            .attributes
            .iter()
            .any(|attr| attr == &requested_role),
        "created mailbox must advertise the requested {} role: {:?}",
        requested_role.as_imap_str(),
        listing[0].attributes
    );

    conn.delete(&mbox, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_rename_mailbox() {
    let conn = connect_stalwart().await;

    let old_name = &format!("test-rename-old-{}", std::process::id());
    let new_name = &format!("test-rename-new-{}", std::process::id());
    conn.create(old_name, TIMEOUT).await.unwrap();

    conn.rename(old_name, new_name, TIMEOUT).await.unwrap();

    let old_list = conn.list("", old_name, TIMEOUT).await.unwrap();
    assert!(old_list.is_empty(), "old name should be gone after RENAME");

    let new_list = conn.list("", new_name, TIMEOUT).await.unwrap();
    assert!(!new_list.is_empty(), "new name should exist after RENAME");

    conn.delete(new_name, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

/// RENAME must rename inferior hierarchical names along with the parent.
///
/// RFC 3501 Section 6.3.5: if a mailbox name has inferior hierarchical names,
/// those inferiors MUST also be renamed.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_rename_parent_renames_inferiors() {
    let conn = connect_stalwart().await;

    let delimiter = hierarchy_delimiter(&conn).await;
    let old_parent = unique_name("test-rename-tree-old");
    let old_child = format!("{old_parent}{delimiter}child");
    let new_parent = unique_name("test-rename-tree-new");
    let new_child = format!("{new_parent}{delimiter}child");

    conn.create(&old_parent, TIMEOUT).await.unwrap();
    conn.create(&old_child, TIMEOUT).await.unwrap();

    conn.rename(&old_parent, &new_parent, TIMEOUT)
        .await
        .unwrap();

    let mailboxes = conn.list("", "*", TIMEOUT).await.unwrap();
    assert!(
        mailboxes.iter().any(|m| m.name.as_str() == new_parent),
        "renamed parent mailbox must exist"
    );
    assert!(
        mailboxes.iter().any(|m| m.name.as_str() == new_child),
        "inferior mailbox must be renamed together with its parent"
    );
    assert!(
        !mailboxes.iter().any(|m| m.name.as_str() == old_parent),
        "old parent mailbox name must not remain after rename"
    );
    assert!(
        !mailboxes.iter().any(|m| m.name.as_str() == old_child),
        "old inferior mailbox name must not remain after rename"
    );

    conn.delete(&new_child, TIMEOUT).await.unwrap();
    conn.delete(&new_parent, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

// ==========================================================================
// STORE + EXPUNGE
// ==========================================================================

#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_store_and_expunge() {
    let conn = connect_stalwart().await;

    let mbox = &format!("test-expunge-{}", std::process::id());
    conn.create(mbox, TIMEOUT).await.unwrap();

    conn.append(mbox, &[], None, TEST_MESSAGE, TIMEOUT)
        .await
        .unwrap();
    conn.select(mbox, TIMEOUT).await.unwrap();

    // STORE \Deleted flag.
    conn.uid_store(
        &SequenceSet::new("*").unwrap(),
        daaki_imap::types::StoreOperation::Add,
        &[Flag::Deleted],
        None,
        TIMEOUT,
    )
    .await
    .unwrap();

    // EXPUNGE.
    conn.expunge(TIMEOUT).await.unwrap();

    let selected = conn.select(mbox, TIMEOUT).await.unwrap();
    assert_eq!(selected.exists, 0, "mailbox should be empty after EXPUNGE");

    conn.delete(mbox, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

// ==========================================================================
// UID EXPUNGE (RFC 4315 / RFC 9051 base)
// ==========================================================================

#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_uid_expunge() {
    let conn = connect_stalwart().await;

    let mbox = &format!("test-uid-expunge-{}", std::process::id());
    conn.create(mbox, TIMEOUT).await.unwrap();

    // Append two messages.
    conn.append(mbox, &[], None, TEST_MESSAGE, TIMEOUT)
        .await
        .unwrap();
    conn.append(mbox, &[], None, TEST_MESSAGE, TIMEOUT)
        .await
        .unwrap();
    conn.select(mbox, TIMEOUT).await.unwrap();

    let fetches = conn
        .uid_fetch(
            &SequenceSet::new("1:*").unwrap(),
            &[FetchAttr::Uid],
            TIMEOUT,
        )
        .await
        .unwrap();
    assert_eq!(fetches.len(), 2, "expected 2 messages");
    let uids: Vec<u32> = fetches.iter().map(|f| f.uid.unwrap()).collect();

    // Mark only the first message as deleted.
    let first_uid = uids[0].to_string();
    conn.uid_store(
        &SequenceSet::new(&first_uid).unwrap(),
        daaki_imap::types::StoreOperation::Add,
        &[Flag::Deleted],
        None,
        TIMEOUT,
    )
    .await
    .unwrap();

    // UID EXPUNGE only the first message (RFC 4315 / RFC 9051).
    conn.uid_expunge(&SequenceSet::new(&first_uid).unwrap(), TIMEOUT)
        .await
        .unwrap();

    // Only the second message should remain.
    let remaining = conn
        .uid_fetch(
            &SequenceSet::new("1:*").unwrap(),
            &[FetchAttr::Uid],
            TIMEOUT,
        )
        .await
        .unwrap();
    assert_eq!(remaining.len(), 1, "expected 1 message after UID EXPUNGE");
    assert_eq!(
        remaining[0].uid.unwrap(),
        uids[1],
        "the second message should survive"
    );

    conn.delete(mbox, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

/// UID EXPUNGE must ignore matching UIDs that are not marked `\\Deleted`.
///
/// RFC 4315 Section 2 removes only messages that both have the `\\Deleted`
/// flag and whose UID is in the requested set.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_uid_expunge_ignores_non_deleted_uids() {
    let conn = connect_stalwart().await;

    let mailbox = unique_name("test-uid-expunge-ignore-nondeleted");
    conn.create(&mailbox, TIMEOUT).await.unwrap();
    conn.select(&mailbox, TIMEOUT).await.unwrap();

    conn.append(&mailbox, &[], None, TEST_MESSAGE, TIMEOUT)
        .await
        .unwrap();
    conn.append(&mailbox, &[], None, TEST_MESSAGE, TIMEOUT)
        .await
        .unwrap();

    let fetches = conn
        .uid_fetch(
            &SequenceSet::new("1:*").unwrap(),
            &[FetchAttr::Uid],
            TIMEOUT,
        )
        .await
        .unwrap();
    let first_uid = fetches[0].uid.expect("first message must have UID");
    let second_uid = fetches[1].uid.expect("second message must have UID");

    conn.uid_store(
        &SequenceSet::new(first_uid.to_string()).unwrap(),
        daaki_imap::types::StoreOperation::Add,
        &[Flag::Deleted],
        None,
        TIMEOUT,
    )
    .await
    .unwrap();

    conn.uid_expunge(
        &SequenceSet::new(format!("{first_uid},{second_uid}")).unwrap(),
        TIMEOUT,
    )
    .await
    .unwrap();

    let remaining = conn
        .uid_fetch(
            &SequenceSet::new("1:*").unwrap(),
            &[FetchAttr::Uid],
            TIMEOUT,
        )
        .await
        .unwrap();
    assert_eq!(
        remaining.len(),
        1,
        "UID EXPUNGE must not remove non-deleted messages even when their UID is requested"
    );
    assert_eq!(remaining[0].uid, Some(second_uid));

    delete_mailbox_and_contents(&conn, &mailbox).await;
    conn.logout().await.unwrap();
}

/// UID EXPUNGE must leave deleted messages untouched when their UID is not in
/// the requested set.
///
/// RFC 4315 Section 2 scopes removal to the requested UID set, so a deleted
/// message outside that set must remain in the mailbox.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_uid_expunge_limits_to_requested_deleted_uid_set() {
    let conn = connect_stalwart().await;

    let mailbox = unique_name("test-uid-expunge-scoped");
    conn.create(&mailbox, TIMEOUT).await.unwrap();
    conn.select(&mailbox, TIMEOUT).await.unwrap();

    conn.append(&mailbox, &[], None, TEST_MESSAGE, TIMEOUT)
        .await
        .unwrap();
    conn.append(&mailbox, &[], None, TEST_MESSAGE, TIMEOUT)
        .await
        .unwrap();

    let fetches = conn
        .uid_fetch(
            &SequenceSet::new("1:*").unwrap(),
            &[FetchAttr::Uid],
            TIMEOUT,
        )
        .await
        .unwrap();
    let first_uid = fetches[0].uid.expect("first message must have UID");
    let second_uid = fetches[1].uid.expect("second message must have UID");

    conn.uid_store(
        &SequenceSet::new(format!("{first_uid},{second_uid}")).unwrap(),
        daaki_imap::types::StoreOperation::Add,
        &[Flag::Deleted],
        None,
        TIMEOUT,
    )
    .await
    .unwrap();

    conn.uid_expunge(&SequenceSet::new(first_uid.to_string()).unwrap(), TIMEOUT)
        .await
        .unwrap();

    let remaining = conn
        .uid_fetch(
            &SequenceSet::new("1:*").unwrap(),
            &[FetchAttr::Uid, FetchAttr::Flags],
            TIMEOUT,
        )
        .await
        .unwrap();
    assert_eq!(
        remaining.len(),
        1,
        "UID EXPUNGE must remove only the requested deleted message"
    );
    assert_eq!(remaining[0].uid, Some(second_uid));
    assert!(
        remaining[0]
            .flags
            .as_ref()
            .expect("remaining message must include FLAGS")
            .contains(&Flag::Deleted),
        "the unrequested deleted message must remain deleted after UID EXPUNGE"
    );

    delete_mailbox_and_contents(&conn, &mailbox).await;
    conn.logout().await.unwrap();
}

/// UID EXPUNGE with a nonexistent UID must be a no-op, even if other messages
/// in the mailbox are marked `\\Deleted`.
///
/// RFC 4315 Section 2 limits removal to the requested UIDs, so a nonexistent
/// UID cannot expunge some other deleted message.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_uid_expunge_nonexistent_uid_is_noop() {
    let conn = connect_stalwart().await;

    let mailbox = unique_name("test-uid-expunge-missing");
    conn.create(&mailbox, TIMEOUT).await.unwrap();
    conn.select(&mailbox, TIMEOUT).await.unwrap();

    conn.append(&mailbox, &[], None, TEST_MESSAGE, TIMEOUT)
        .await
        .unwrap();

    let existing_uid = conn
        .uid_fetch(
            &SequenceSet::new("1:*").unwrap(),
            &[FetchAttr::Uid],
            TIMEOUT,
        )
        .await
        .unwrap()[0]
        .uid
        .expect("appended message must have UID");
    let missing_uid = existing_uid + 1000;

    conn.uid_store(
        &SequenceSet::new(existing_uid.to_string()).unwrap(),
        daaki_imap::types::StoreOperation::Add,
        &[Flag::Deleted],
        None,
        TIMEOUT,
    )
    .await
    .unwrap();

    conn.uid_expunge(&SequenceSet::new(missing_uid.to_string()).unwrap(), TIMEOUT)
        .await
        .unwrap();

    let remaining = conn
        .uid_fetch(
            &SequenceSet::new(existing_uid.to_string()).unwrap(),
            &[FetchAttr::Uid, FetchAttr::Flags],
            TIMEOUT,
        )
        .await
        .unwrap();
    assert_eq!(
        remaining.len(),
        1,
        "UID EXPUNGE with a nonexistent UID must not expunge other deleted messages"
    );
    assert_eq!(remaining[0].uid, Some(existing_uid));
    assert!(
        remaining[0]
            .flags
            .as_ref()
            .expect("remaining message must include FLAGS")
            .contains(&Flag::Deleted),
        "the existing deleted message must remain after a nonexistent UID EXPUNGE"
    );

    delete_mailbox_and_contents(&conn, &mailbox).await;
    conn.logout().await.unwrap();
}

// ==========================================================================
// COPY (RFC 3501 Section 6.4.7)
// ==========================================================================

#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_uid_copy() {
    let conn = connect_stalwart().await;

    let src = &format!("test-copy-src-{}", std::process::id());
    let dst = &format!("test-copy-dst-{}", std::process::id());
    conn.create(src, TIMEOUT).await.unwrap();
    conn.create(dst, TIMEOUT).await.unwrap();

    conn.append(src, &[], None, TEST_MESSAGE, TIMEOUT)
        .await
        .unwrap();
    conn.select(src, TIMEOUT).await.unwrap();

    let fetches = conn
        .uid_fetch(
            &SequenceSet::new("1:*").unwrap(),
            &[FetchAttr::Uid],
            TIMEOUT,
        )
        .await
        .unwrap();
    let uid_set = fetches[0].uid.unwrap().to_string();

    conn.uid_copy(&SequenceSet::new(&uid_set).unwrap(), dst, TIMEOUT)
        .await
        .unwrap();

    // Verify dst has a copy.
    let selected = conn.select(dst, TIMEOUT).await.unwrap();
    assert!(selected.exists >= 1, "dst should have the copied message");

    // src still has the original.
    let selected = conn.select(src, TIMEOUT).await.unwrap();
    assert!(selected.exists >= 1, "src should still have the original");

    conn.delete(src, TIMEOUT).await.unwrap();
    conn.delete(dst, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

/// UID COPY should preserve message flags in the destination mailbox.
///
/// RFC 3501 Section 6.4.7 says COPY copies messages to the end of the
/// destination mailbox and SHOULD preserve message flags.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_uid_copy_preserves_flags_in_destination() {
    let conn = connect_stalwart().await;

    let src = unique_name("test-uid-copy-flags-src");
    let dst = unique_name("test-uid-copy-flags-dst");
    conn.create(&src, TIMEOUT).await.unwrap();
    conn.create(&dst, TIMEOUT).await.unwrap();
    conn.select(&src, TIMEOUT).await.unwrap();

    conn.append(
        &src,
        &[Flag::Seen, Flag::Flagged],
        None,
        TEST_MESSAGE,
        TIMEOUT,
    )
    .await
    .unwrap();
    let source = conn
        .uid_fetch(
            &SequenceSet::new("1:*").unwrap(),
            &[FetchAttr::Uid],
            TIMEOUT,
        )
        .await
        .unwrap();
    assert_eq!(
        source.len(),
        1,
        "test setup must create exactly one source message before UID COPY"
    );
    let source_uid = source[0].uid.expect("source message must have UID");

    conn.uid_copy(
        &SequenceSet::new(source_uid.to_string()).unwrap(),
        &dst,
        TIMEOUT,
    )
    .await
    .unwrap();

    conn.select(&dst, TIMEOUT).await.unwrap();
    let copied = conn
        .uid_fetch(
            &SequenceSet::new("1:*").unwrap(),
            &[FetchAttr::Uid, FetchAttr::Flags],
            TIMEOUT,
        )
        .await
        .unwrap();
    assert_eq!(
        copied.len(),
        1,
        "destination mailbox must contain exactly one copied message"
    );
    let copied_flags = copied[0]
        .flags
        .as_ref()
        .expect("FETCH FLAGS requested by the test must be present");
    assert!(
        copied_flags.contains(&Flag::Seen),
        "UID COPY must preserve \\Seen in the destination mailbox"
    );
    assert!(
        copied_flags.contains(&Flag::Flagged),
        "UID COPY must preserve \\Flagged in the destination mailbox"
    );

    delete_mailbox_and_contents(&conn, &src).await;
    delete_mailbox_and_contents(&conn, &dst).await;
    conn.logout().await.unwrap();
}

/// COPY by sequence number should preserve message flags in the destination
/// mailbox.
///
/// RFC 3501 Section 6.4.7 says COPY appends copied messages to the end of the
/// destination mailbox and SHOULD preserve message flags.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_copy_by_sequence_preserves_flags_in_destination() {
    let conn = connect_stalwart().await;

    let src = unique_name("test-copy-seq-flags-src");
    let dst = unique_name("test-copy-seq-flags-dst");
    conn.create(&src, TIMEOUT).await.unwrap();
    conn.create(&dst, TIMEOUT).await.unwrap();
    conn.select(&src, TIMEOUT).await.unwrap();

    conn.append(
        &src,
        &[Flag::Seen, Flag::Flagged],
        None,
        TEST_MESSAGE,
        TIMEOUT,
    )
    .await
    .unwrap();
    conn.copy(&SequenceSet::new("1").unwrap(), &dst, TIMEOUT)
        .await
        .unwrap();

    conn.select(&dst, TIMEOUT).await.unwrap();
    let copied = conn
        .uid_fetch(
            &SequenceSet::new("1:*").unwrap(),
            &[FetchAttr::Uid, FetchAttr::Flags],
            TIMEOUT,
        )
        .await
        .unwrap();
    assert_eq!(
        copied.len(),
        1,
        "destination mailbox must contain exactly one copied message"
    );
    let copied_flags = copied[0]
        .flags
        .as_ref()
        .expect("FETCH FLAGS requested by the test must be present");
    assert!(
        copied_flags.contains(&Flag::Seen),
        "COPY must preserve \\Seen in the destination mailbox"
    );
    assert!(
        copied_flags.contains(&Flag::Flagged),
        "COPY must preserve \\Flagged in the destination mailbox"
    );

    delete_mailbox_and_contents(&conn, &src).await;
    delete_mailbox_and_contents(&conn, &dst).await;
    conn.logout().await.unwrap();
}

// ==========================================================================
// SEARCH variants
// ==========================================================================

#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_search_and_uid_search() {
    let conn = connect_stalwart().await;
    conn.select("INBOX", TIMEOUT).await.unwrap();

    // Stalwart admin account requires sequence set based search.
    let _ = conn.search("1:*", TIMEOUT).await.unwrap();
    let _ = conn.uid_search("1:*", TIMEOUT).await.unwrap();
    let _ = conn.search("UNSEEN", TIMEOUT).await.unwrap();

    conn.logout().await.unwrap();
}

// ==========================================================================
// LSUB (RFC 3501 Section 6.3.9 — deprecated in rev2 but should still work)
// ==========================================================================

#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_lsub() {
    let conn = connect_stalwart().await;

    // LSUB is deprecated in rev2 (RFC 9051 Appendix F) but servers should
    // still accept it for backward compatibility.
    match conn.lsub("", "*", TIMEOUT).await {
        Ok(subscribed) => {
            // At minimum, INBOX should be subscribed.
            assert!(
                subscribed
                    .iter()
                    .any(|m| m.name.as_str().eq_ignore_ascii_case("INBOX")),
                "expected INBOX in LSUB response"
            );
        }
        Err(daaki_imap::Error::Bad { .. }) => {
            // Some strict rev2 servers might reject LSUB — acceptable.
            eprintln!("Stalwart rejected LSUB (strict rev2 behavior) — OK");
        }
        Err(e) => panic!("unexpected error from LSUB: {e:?}"),
    }

    conn.logout().await.unwrap();
}

// ==========================================================================
// SORT (RFC 5256)
// ==========================================================================

#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_sort() {
    let conn = connect_stalwart().await;
    conn.select("INBOX", TIMEOUT).await.unwrap();

    match conn.sort("DATE", "UTF-8", "ALL", TIMEOUT).await {
        Ok(result) => {
            let _ = result;
        }
        Err(daaki_imap::Error::MissingCapability(_)) => {
            eprintln!("skipping: Stalwart does not advertise SORT capability");
        }
        Err(e) => panic!("unexpected error from SORT: {e:?}"),
    }

    conn.logout().await.unwrap();
}

/// `SORT` must return current sequence numbers in the requested sort order,
/// even when those sequence numbers no longer line up with immutable UIDs.
///
/// RFC 5256 Section 2 defines SORT in sequence-number space, while RFC 3501
/// Section 2.3.1.2 allows sequence numbers to diverge from UIDs after expunge.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_sort_returns_sequence_numbers_in_date_order() {
    let conn = connect_stalwart().await;

    if !has_capability(&conn, &Capability::Sort) {
        eprintln!("skipping: Stalwart does not advertise SORT");
        conn.logout().await.unwrap();
        return;
    }

    let mbox = unique_name("test-stalwart-sort-seq");
    conn.create(&mbox, TIMEOUT).await.unwrap();
    conn.select(&mbox, TIMEOUT).await.unwrap();
    create_uid_gap(&conn, &mbox).await;

    let dated_messages = [
        ("stalwart-sort-mar", "15 Mar 2026 00:00:00 +0000"),
        ("stalwart-sort-jan", "10 Jan 2026 00:00:00 +0000"),
        ("stalwart-sort-jun", "20 Jun 2026 00:00:00 +0000"),
    ];
    for (label, date) in dated_messages {
        let message = format!(
            "From: sender@example.com\r\n\
             To: recipient@example.com\r\n\
             Subject: {label}\r\n\
             Date: {date}\r\n\
             Message-ID: <{label}-{}@daaki>\r\n\
             MIME-Version: 1.0\r\n\
             Content-Type: text/plain; charset=utf-8\r\n\
             \r\n\
             {label}\r\n",
            std::process::id()
        );
        conn.append(&mbox, &[], None, message.as_bytes(), TIMEOUT)
            .await
            .unwrap();
    }

    let live_uids = sorted_uids(&conn).await;
    assert_ne!(
        live_uids,
        vec![1, 2, 3],
        "setup must create a UID/sequence divergence before SORT"
    );

    let sort_result = conn.sort("DATE", "UTF-8", "ALL", TIMEOUT).await.unwrap();
    assert_eq!(
        sort_result.ids,
        vec![2, 1, 3],
        "SORT DATE must return sequence numbers in ascending date order"
    );
    assert!(
        !sort_result.truncated,
        "complete SORT results must not be marked truncated"
    );

    delete_mailbox_and_contents(&conn, &mbox).await;
    conn.logout().await.unwrap();
}

// ==========================================================================
// THREAD (RFC 5256)
// ==========================================================================

#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_thread() {
    let conn = connect_stalwart().await;
    conn.select("INBOX", TIMEOUT).await.unwrap();

    match conn.thread("REFERENCES", "UTF-8", "ALL", TIMEOUT).await {
        Ok(threads) => {
            let _ = threads;
        }
        Err(daaki_imap::Error::MissingCapability(_)) => {
            eprintln!("skipping: Stalwart does not advertise THREAD capability");
        }
        Err(e) => panic!("unexpected error from THREAD: {e:?}"),
    }

    conn.logout().await.unwrap();
}

/// `THREAD REFERENCES` must return current sequence numbers, not immutable
/// UIDs, in the thread tree for the selected mailbox.
///
/// RFC 5256 Section 3 defines THREAD in sequence-number space unless the UID
/// variant is explicitly requested.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_thread_references_returns_sequence_numbers() {
    let conn = connect_stalwart().await;

    let has_thread_refs = conn.capabilities().iter().any(
        |cap| matches!(cap, Capability::Thread(algo) if algo.eq_ignore_ascii_case("REFERENCES")),
    );
    if !has_thread_refs {
        eprintln!("skipping: Stalwart does not advertise THREAD=REFERENCES");
        conn.logout().await.unwrap();
        return;
    }

    let mbox = unique_name("test-stalwart-thread-seq");
    conn.create(&mbox, TIMEOUT).await.unwrap();
    conn.select(&mbox, TIMEOUT).await.unwrap();
    create_uid_gap(&conn, &mbox).await;

    let pid = std::process::id();
    let original = format!(
        "From: alice@example.com\r\n\
         To: bob@example.com\r\n\
         Subject: Stalwart thread test\r\n\
         Date: 10 Jan 2026 00:00:00 +0000\r\n\
         Message-ID: <stalwart-thread-orig-{pid}@daaki>\r\n\
         MIME-Version: 1.0\r\n\
         Content-Type: text/plain; charset=utf-8\r\n\
         \r\n\
         Original message\r\n"
    );
    let reply1 = format!(
        "From: bob@example.com\r\n\
         To: alice@example.com\r\n\
         Subject: Re: Stalwart thread test\r\n\
         Date: 11 Jan 2026 00:00:00 +0000\r\n\
         Message-ID: <stalwart-thread-reply1-{pid}@daaki>\r\n\
         In-Reply-To: <stalwart-thread-orig-{pid}@daaki>\r\n\
         References: <stalwart-thread-orig-{pid}@daaki>\r\n\
         MIME-Version: 1.0\r\n\
         Content-Type: text/plain; charset=utf-8\r\n\
         \r\n\
         Reply 1\r\n"
    );
    let reply2 = format!(
        "From: charlie@example.com\r\n\
         To: alice@example.com\r\n\
         Subject: Re: Stalwart thread test\r\n\
         Date: 12 Jan 2026 00:00:00 +0000\r\n\
         Message-ID: <stalwart-thread-reply2-{pid}@daaki>\r\n\
         In-Reply-To: <stalwart-thread-orig-{pid}@daaki>\r\n\
         References: <stalwart-thread-orig-{pid}@daaki>\r\n\
         MIME-Version: 1.0\r\n\
         Content-Type: text/plain; charset=utf-8\r\n\
         \r\n\
         Reply 2\r\n"
    );
    for message in [&original, &reply1, &reply2] {
        conn.append(&mbox, &[], None, message.as_bytes(), TIMEOUT)
            .await
            .unwrap();
    }

    let expected_sequences: Vec<u32> = conn
        .fetch(
            &SequenceSet::new("1:*").unwrap(),
            &[FetchAttr::Flags],
            TIMEOUT,
        )
        .await
        .unwrap()
        .into_iter()
        .map(|fetch| fetch.seq)
        .collect();
    assert_eq!(
        expected_sequences,
        vec![1, 2, 3],
        "expected the three live messages to occupy sequence numbers 1..=3"
    );
    assert_ne!(
        sorted_uids(&conn).await,
        expected_sequences,
        "setup must create a UID/sequence divergence before THREAD"
    );

    let threads = conn
        .thread("REFERENCES", "UTF-8", "ALL", TIMEOUT)
        .await
        .unwrap();
    assert!(
        !threads.is_empty(),
        "THREAD REFERENCES must return at least one thread"
    );

    let mut actual_sequences = Vec::new();
    collect_thread_ids(&threads, &mut actual_sequences);
    actual_sequences.sort_unstable();

    assert_eq!(
        actual_sequences, expected_sequences,
        "THREAD must surface sequence numbers, not UIDs"
    );

    delete_mailbox_and_contents(&conn, &mbox).await;
    conn.logout().await.unwrap();
}

/// `UID SORT` must return UIDs in the requested sort order, not current
/// sequence numbers.
///
/// RFC 5256 Section 2 defines UID SORT as the UID-space variant of SORT, and
/// RFC 3501 Section 2.3.1.1 keeps UIDs stable even when sequence numbers shift.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_uid_sort_returns_uids_in_date_order() {
    let conn = connect_stalwart().await;

    if !has_capability(&conn, &Capability::Sort) {
        eprintln!("skipping: Stalwart does not advertise SORT");
        conn.logout().await.unwrap();
        return;
    }

    let mbox = unique_name("test-stalwart-uid-sort");
    conn.create(&mbox, TIMEOUT).await.unwrap();
    conn.select(&mbox, TIMEOUT).await.unwrap();
    create_uid_gap(&conn, &mbox).await;

    let dated_messages = [
        ("stalwart-uid-sort-mar", "15 Mar 2026 00:00:00 +0000"),
        ("stalwart-uid-sort-jan", "10 Jan 2026 00:00:00 +0000"),
        ("stalwart-uid-sort-jun", "20 Jun 2026 00:00:00 +0000"),
    ];
    for (label, date) in dated_messages {
        let msg = format!(
            "From: sender@example.com\r\n\
             To: recipient@example.com\r\n\
             Subject: {label}\r\n\
             Date: {date}\r\n\
             Message-ID: <{label}-{}@daaki>\r\n\
             MIME-Version: 1.0\r\n\
             Content-Type: text/plain; charset=utf-8\r\n\
             \r\n\
             {label}\r\n",
            std::process::id()
        );
        conn.append(&mbox, &[], None, msg.as_bytes(), TIMEOUT)
            .await
            .unwrap();
    }

    let fetches = conn
        .uid_fetch(
            &SequenceSet::new("1:*").unwrap(),
            &[FetchAttr::Uid],
            TIMEOUT,
        )
        .await
        .unwrap();
    assert_eq!(fetches.len(), 3, "expected 3 sortable messages");

    let march_uid = fetches[0]
        .uid
        .expect("UID SORT seed fetch must include UID");
    let january_uid = fetches[1]
        .uid
        .expect("UID SORT seed fetch must include UID");
    let june_uid = fetches[2]
        .uid
        .expect("UID SORT seed fetch must include UID");

    let sort_result = conn
        .uid_sort("DATE", "UTF-8", "ALL", TIMEOUT)
        .await
        .unwrap();
    assert_eq!(
        sort_result.ids,
        vec![january_uid, march_uid, june_uid],
        "UID SORT DATE must return UIDs in ascending date order"
    );
    assert!(
        !sort_result.truncated,
        "complete UID SORT results must not be marked truncated"
    );

    conn.close(TIMEOUT).await.unwrap();
    conn.delete(&mbox, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

/// LIST-STATUS on a selected mailbox must not deselect the mailbox.
///
/// RFC 5819 Section 2 defines LIST with STATUS return data, and RFC 3501
/// Section 6.4 requires selected-state commands to leave the selected mailbox
/// intact.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_rev2_list_status_selected_preserves_selected_state() {
    let conn = connect_stalwart().await;

    let mailbox = unique_name("test-stalwart-list-status-selected");
    conn.create(&mailbox, TIMEOUT).await.unwrap();
    conn.append(&mailbox, &[], None, TEST_MESSAGE, TIMEOUT)
        .await
        .unwrap();
    conn.select(&mailbox, TIMEOUT).await.unwrap();

    let listing = conn
        .list_status("", &mailbox, "MESSAGES UIDNEXT", TIMEOUT)
        .await
        .unwrap();
    assert_eq!(
        listing.len(),
        1,
        "LIST-STATUS must return one exact mailbox"
    );
    assert_eq!(
        status_messages(&listing[0].1),
        Some(1),
        "LIST-STATUS must report the selected mailbox's message count"
    );
    assert_selected_mailbox_still_usable(&conn, "LIST-STATUS").await;

    conn.close(TIMEOUT).await.unwrap();
    conn.delete(&mailbox, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

/// `UID THREAD` must return message UIDs in the thread tree, not the current
/// sequence numbers of the selected mailbox.
///
/// RFC 5256 Section 3 defines UID THREAD as the UID-space variant of THREAD.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_uid_thread_references_returns_uids() {
    let conn = connect_stalwart().await;

    let has_thread_refs = conn.capabilities().iter().any(
        |cap| matches!(cap, Capability::Thread(algo) if algo.eq_ignore_ascii_case("REFERENCES")),
    );
    if !has_thread_refs {
        eprintln!("skipping: Stalwart does not advertise THREAD=REFERENCES");
        conn.logout().await.unwrap();
        return;
    }

    let mbox = unique_name("test-stalwart-uid-thread");
    conn.create(&mbox, TIMEOUT).await.unwrap();
    conn.select(&mbox, TIMEOUT).await.unwrap();
    create_uid_gap(&conn, &mbox).await;

    let pid = std::process::id();
    let original = format!(
        "From: alice@example.com\r\n\
         To: bob@example.com\r\n\
         Subject: Stalwart UID thread test\r\n\
         Date: 10 Jan 2026 00:00:00 +0000\r\n\
         Message-ID: <stalwart-uid-thread-orig-{pid}@daaki>\r\n\
         MIME-Version: 1.0\r\n\
         Content-Type: text/plain; charset=utf-8\r\n\
         \r\n\
         Original message\r\n"
    );
    let reply1 = format!(
        "From: bob@example.com\r\n\
         To: alice@example.com\r\n\
         Subject: Re: Stalwart UID thread test\r\n\
         Date: 11 Jan 2026 00:00:00 +0000\r\n\
         Message-ID: <stalwart-uid-thread-reply1-{pid}@daaki>\r\n\
         In-Reply-To: <stalwart-uid-thread-orig-{pid}@daaki>\r\n\
         References: <stalwart-uid-thread-orig-{pid}@daaki>\r\n\
         MIME-Version: 1.0\r\n\
         Content-Type: text/plain; charset=utf-8\r\n\
         \r\n\
         Reply 1\r\n"
    );
    let reply2 = format!(
        "From: charlie@example.com\r\n\
         To: alice@example.com\r\n\
         Subject: Re: Stalwart UID thread test\r\n\
         Date: 12 Jan 2026 00:00:00 +0000\r\n\
         Message-ID: <stalwart-uid-thread-reply2-{pid}@daaki>\r\n\
         In-Reply-To: <stalwart-uid-thread-orig-{pid}@daaki>\r\n\
         References: <stalwart-uid-thread-orig-{pid}@daaki>\r\n\
         MIME-Version: 1.0\r\n\
         Content-Type: text/plain; charset=utf-8\r\n\
         \r\n\
         Reply 2\r\n"
    );
    for msg in [&original, &reply1, &reply2] {
        conn.append(&mbox, &[], None, msg.as_bytes(), TIMEOUT)
            .await
            .unwrap();
    }

    let fetches = conn
        .uid_fetch(
            &SequenceSet::new("1:*").unwrap(),
            &[FetchAttr::Uid],
            TIMEOUT,
        )
        .await
        .unwrap();
    assert_eq!(fetches.len(), 3, "expected 3 threaded messages");

    let mut expected_uids: Vec<u32> = fetches
        .into_iter()
        .map(|fetch| fetch.uid.expect("UID THREAD seed fetch must include UID"))
        .collect();
    expected_uids.sort_unstable();

    let threads = conn
        .uid_thread("REFERENCES", "UTF-8", "ALL", TIMEOUT)
        .await
        .unwrap();
    assert!(
        !threads.is_empty(),
        "UID THREAD REFERENCES must return at least one thread"
    );

    let mut actual_uids = Vec::new();
    collect_thread_ids(&threads, &mut actual_uids);
    actual_uids.sort_unstable();

    assert_eq!(
        actual_uids, expected_uids,
        "UID THREAD must surface message UIDs, not sequence numbers"
    );

    conn.close(TIMEOUT).await.unwrap();
    conn.delete(&mbox, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

// ==========================================================================
// FETCH BODYSTRUCTURE
// ==========================================================================

#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_fetch_bodystructure() {
    let conn = connect_stalwart().await;

    let mbox = &format!("test-bs-{}", std::process::id());
    conn.create(mbox, TIMEOUT).await.unwrap();
    conn.append(mbox, &[], None, TEST_MESSAGE, TIMEOUT)
        .await
        .unwrap();
    conn.select(mbox, TIMEOUT).await.unwrap();

    let fetches = conn
        .fetch(
            &SequenceSet::new("1").unwrap(),
            &[FetchAttr::Uid, FetchAttr::BodyStructure],
            TIMEOUT,
        )
        .await
        .unwrap();

    assert!(!fetches.is_empty());
    assert!(
        fetches[0].body_structure.is_some(),
        "BODYSTRUCTURE should be present"
    );

    conn.delete(mbox, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

/// BODY.PEEK[] MUST NOT set the `\\Seen` flag; BODY[] MUST set it.
///
/// RFC 3501 Section 6.4.5 defines `BODY.PEEK[<section>]` as the FETCH variant
/// that does not implicitly set `\\Seen`, while `BODY[<section>]` does.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_fetch_body_peek_preserves_unseen() {
    let conn = connect_stalwart().await;

    let mbox = unique_name("test-peek-seen");
    conn.create(&mbox, TIMEOUT).await.unwrap();
    conn.select(&mbox, TIMEOUT).await.unwrap();
    conn.append(&mbox, &[], None, TEST_MESSAGE, TIMEOUT)
        .await
        .unwrap();
    conn.select(&mbox, TIMEOUT).await.unwrap();

    let fetches = conn
        .uid_fetch(
            &SequenceSet::new("1:*").unwrap(),
            &[FetchAttr::Uid],
            TIMEOUT,
        )
        .await
        .unwrap();
    assert!(!fetches.is_empty(), "expected at least one message");
    let uid = fetches[0].uid.expect("appended message must have UID");

    let fetches = conn
        .uid_fetch(
            &SequenceSet::new(uid.to_string()).unwrap(),
            &[
                FetchAttr::Uid,
                FetchAttr::BodySection {
                    peek: true,
                    section: None,
                    partial: None,
                },
            ],
            TIMEOUT,
        )
        .await
        .unwrap();
    let data = fetches[0].body_sections[0]
        .data
        .as_ref()
        .expect("expected body data from BODY.PEEK[]");
    assert!(!data.is_empty(), "BODY.PEEK[] should return non-empty data");

    let fetches = conn
        .uid_fetch(
            &SequenceSet::new(uid.to_string()).unwrap(),
            &[FetchAttr::Uid, FetchAttr::Flags],
            TIMEOUT,
        )
        .await
        .unwrap();
    let flags = fetches[0].flags.as_deref().unwrap_or_default();
    assert!(
        !flags.contains(&Flag::Seen),
        "BODY.PEEK[] must not set \\Seen, got {flags:?}"
    );

    conn.uid_fetch(
        &SequenceSet::new(uid.to_string()).unwrap(),
        &[
            FetchAttr::Uid,
            FetchAttr::BodySection {
                peek: false,
                section: None,
                partial: None,
            },
        ],
        TIMEOUT,
    )
    .await
    .unwrap();

    let fetches = conn
        .uid_fetch(
            &SequenceSet::new(uid.to_string()).unwrap(),
            &[FetchAttr::Uid, FetchAttr::Flags],
            TIMEOUT,
        )
        .await
        .unwrap();
    let flags = fetches[0].flags.as_deref().unwrap_or_default();
    assert!(
        flags.contains(&Flag::Seen),
        "BODY[] must set \\Seen, got {flags:?}"
    );

    delete_mailbox_and_contents(&conn, &mbox).await;
    conn.logout().await.unwrap();
}

/// `fetch_streaming` must leave the selected mailbox usable for a later
/// regular FETCH.
///
/// RFC 3501 Section 6.4.5 defines FETCH without changing selected-mailbox
/// state, so a completed streaming fetch must not poison the connection.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_fetch_streaming_leaves_connection_usable() {
    let conn = connect_stalwart().await;

    let mbox = format!("test-fetch-stream-reuse-{}", std::process::id());
    conn.create(&mbox, TIMEOUT).await.unwrap();
    conn.select(&mbox, TIMEOUT).await.unwrap();

    for i in 0..2 {
        let message = build_message(
            &format!("Stalwart FETCH streaming reuse {i}"),
            &format!(
                "stalwart-fetch-stream-reuse-{i}-{}@daaki",
                std::process::id()
            ),
            &format!("Body {i}"),
        );
        conn.append(&mbox, &[], None, &message, TIMEOUT)
            .await
            .unwrap();
    }

    conn.select(&mbox, TIMEOUT).await.unwrap();

    let (tx, mut rx) = tokio::sync::mpsc::channel(16);
    let seq = SequenceSet::new("1:*").unwrap();
    conn.fetch_streaming(&seq, &[FetchAttr::Uid, FetchAttr::Flags], tx, TIMEOUT)
        .await
        .unwrap();

    let mut streamed = Vec::new();
    while let Some(item) = rx.recv().await {
        streamed.push(item.unwrap());
    }
    assert_eq!(streamed.len(), 2, "expected 2 streamed FETCH responses");

    let fetched = conn
        .fetch(&seq, &[FetchAttr::Uid, FetchAttr::Flags], TIMEOUT)
        .await
        .unwrap();
    assert_eq!(
        fetched.len(),
        2,
        "regular FETCH must still work after fetch_streaming"
    );

    conn.close(TIMEOUT).await.unwrap();
    conn.delete(&mbox, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

/// `uid_fetch_streaming` must leave the selected mailbox usable for a later
/// regular UID FETCH.
///
/// RFC 3501 Section 6.4.8 applies UID to FETCH while preserving FETCH's
/// selected-mailbox semantics from RFC 3501 Section 6.4.5.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_uid_fetch_streaming_leaves_connection_usable() {
    let conn = connect_stalwart().await;

    let mbox = format!("test-uid-fetch-stream-reuse-{}", std::process::id());
    conn.create(&mbox, TIMEOUT).await.unwrap();
    conn.select(&mbox, TIMEOUT).await.unwrap();

    for i in 0..2 {
        let message = build_message(
            &format!("Stalwart UID FETCH streaming reuse {i}"),
            &format!(
                "stalwart-uid-fetch-stream-reuse-{i}-{}@daaki",
                std::process::id()
            ),
            &format!("Body {i}"),
        );
        conn.append(&mbox, &[], None, &message, TIMEOUT)
            .await
            .unwrap();
    }

    conn.select(&mbox, TIMEOUT).await.unwrap();

    let (tx, mut rx) = tokio::sync::mpsc::channel(16);
    let uid_set = SequenceSet::new("1:*").unwrap();
    conn.uid_fetch_streaming(&uid_set, &[FetchAttr::Uid, FetchAttr::Flags], tx, TIMEOUT)
        .await
        .unwrap();

    let mut streamed = Vec::new();
    while let Some(item) = rx.recv().await {
        streamed.push(item.unwrap());
    }
    assert_eq!(streamed.len(), 2, "expected 2 streamed UID FETCH responses");

    let fetched = conn
        .uid_fetch(&uid_set, &[FetchAttr::Uid, FetchAttr::Flags], TIMEOUT)
        .await
        .unwrap();
    assert_eq!(
        fetched.len(),
        2,
        "regular UID FETCH must still work after uid_fetch_streaming"
    );

    conn.close(TIMEOUT).await.unwrap();
    conn.delete(&mbox, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

/// Streaming FETCH over a sequence subset must yield only the requested
/// message sequence numbers.
///
/// RFC 3501 Section 6.4.5 / RFC 9051 Section 6.4.5 define FETCH over a
/// sequence-set, so `2:3` must return only messages whose current sequence
/// numbers are 2 and 3.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_fetch_streaming_sequence_subset_only_returns_requested_messages() {
    let conn = connect_stalwart().await;

    let mbox = format!("test-fetch-stream-subset-{}", std::process::id());
    conn.create(&mbox, TIMEOUT).await.unwrap();
    conn.select(&mbox, TIMEOUT).await.unwrap();

    for i in 1..=3 {
        let message = build_message(
            &format!("Stalwart FETCH streaming subset {i}"),
            &format!(
                "stalwart-fetch-stream-subset-{i}-{}@daaki",
                std::process::id()
            ),
            &format!("Body {i}"),
        );
        conn.append(&mbox, &[], None, &message, TIMEOUT)
            .await
            .unwrap();
    }

    conn.select(&mbox, TIMEOUT).await.unwrap();

    let (tx, mut rx) = tokio::sync::mpsc::channel(16);
    let seq = SequenceSet::new("2:3").unwrap();
    conn.fetch_streaming(&seq, &[FetchAttr::Uid, FetchAttr::Envelope], tx, TIMEOUT)
        .await
        .unwrap();

    let mut streamed = Vec::new();
    while let Some(item) = rx.recv().await {
        streamed.push(item.unwrap());
    }

    assert_eq!(streamed.len(), 2, "expected 2 streamed FETCH responses");
    assert_eq!(
        streamed.iter().map(|resp| resp.seq).collect::<Vec<_>>(),
        vec![2, 3],
        "FETCH 2:3 must return only sequence numbers 2 and 3"
    );
    assert_eq!(
        streamed
            .iter()
            .map(|resp| resp.envelope.as_ref().and_then(|env| env.subject.clone()))
            .collect::<Vec<_>>(),
        vec![
            Some("Stalwart FETCH streaming subset 2".to_owned()),
            Some("Stalwart FETCH streaming subset 3".to_owned()),
        ],
        "FETCH 2:3 must return the requested messages in sequence order"
    );

    conn.close(TIMEOUT).await.unwrap();
    conn.delete(&mbox, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

/// Streaming UID FETCH against an empty mailbox must yield no responses.
///
/// RFC 3501 Section 6.4.8 applies UID to FETCH while preserving FETCH's
/// semantics from RFC 3501 Section 6.4.5, so a UID set that matches no
/// messages must return an empty result rather than an error.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_uid_fetch_streaming_empty_mailbox_returns_no_items() {
    let conn = connect_stalwart().await;

    let mbox = format!("test-uid-fetch-stream-empty-{}", std::process::id());
    conn.create(&mbox, TIMEOUT).await.unwrap();
    conn.select(&mbox, TIMEOUT).await.unwrap();

    let (tx, mut rx) = tokio::sync::mpsc::channel(16);
    let uid_set = SequenceSet::new("1:*").unwrap();
    conn.uid_fetch_streaming(&uid_set, &[FetchAttr::Uid, FetchAttr::Flags], tx, TIMEOUT)
        .await
        .unwrap();

    let mut streamed = Vec::new();
    while let Some(item) = rx.recv().await {
        streamed.push(item.unwrap());
    }

    assert!(
        streamed.is_empty(),
        "UID FETCH 1:* in an empty mailbox must return no items, got: {streamed:?}"
    );

    conn.close(TIMEOUT).await.unwrap();
    conn.delete(&mbox, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

/// Streaming UID FETCH for a nonexistent UID must yield no responses.
///
/// RFC 3501 Section 6.4.8 applies UID to FETCH while preserving FETCH's
/// matching behavior from RFC 3501 Section 6.4.5, so a nonexistent UID in the
/// requested set must be ignored rather than causing command failure.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_uid_fetch_streaming_nonexistent_uid_returns_no_items() {
    let conn = connect_stalwart().await;

    let mbox = format!("test-uid-fetch-stream-missing-{}", std::process::id());
    conn.create(&mbox, TIMEOUT).await.unwrap();
    conn.select(&mbox, TIMEOUT).await.unwrap();

    for i in 0..2 {
        let message = build_message(
            &format!("Stalwart UID FETCH streaming missing {i}"),
            &format!(
                "stalwart-uid-fetch-stream-missing-{i}-{}@daaki",
                std::process::id()
            ),
            &format!("Body {i}"),
        );
        conn.append(&mbox, &[], None, &message, TIMEOUT)
            .await
            .unwrap();
    }

    conn.select(&mbox, TIMEOUT).await.unwrap();

    let existing = conn.uid_search("ALL", TIMEOUT).await.unwrap();
    let missing_uid = existing.ids.iter().copied().max().unwrap_or(0) + 1000;

    let (tx, mut rx) = tokio::sync::mpsc::channel(16);
    let uid_set = SequenceSet::new(missing_uid.to_string()).unwrap();
    conn.uid_fetch_streaming(&uid_set, &[FetchAttr::Uid, FetchAttr::Flags], tx, TIMEOUT)
        .await
        .unwrap();

    let mut streamed = Vec::new();
    while let Some(item) = rx.recv().await {
        streamed.push(item.unwrap());
    }

    assert!(
        streamed.is_empty(),
        "UID FETCH for nonexistent UID {missing_uid} must return no items, got: {streamed:?}"
    );

    conn.close(TIMEOUT).await.unwrap();
    conn.delete(&mbox, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

/// Streaming UID FETCH over a mixed set must return the existing UIDs and skip
/// the missing ones.
///
/// RFC 3501 Section 6.4.8 applies UID to FETCH while preserving RFC 3501
/// Section 6.4.5 matching rules, so servers must ignore nonexistent UIDs in a
/// mixed UID set and still return the matching messages.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_uid_fetch_streaming_ignores_missing_uids_in_mixed_set() {
    let conn = connect_stalwart().await;

    let mbox = format!("test-uid-fetch-stream-mixed-{}", std::process::id());
    conn.create(&mbox, TIMEOUT).await.unwrap();
    conn.select(&mbox, TIMEOUT).await.unwrap();

    for i in 0..2 {
        let message = build_message(
            &format!("Stalwart UID FETCH streaming mixed {i}"),
            &format!(
                "stalwart-uid-fetch-stream-mixed-{i}-{}@daaki",
                std::process::id()
            ),
            &format!("Body {i}"),
        );
        conn.append(&mbox, &[], None, &message, TIMEOUT)
            .await
            .unwrap();
    }

    conn.select(&mbox, TIMEOUT).await.unwrap();

    let mut expected_uids = conn.uid_search("ALL", TIMEOUT).await.unwrap().ids;
    expected_uids.sort_unstable();
    let missing_uid = expected_uids.iter().copied().max().unwrap_or(0) + 1000;

    let (tx, mut rx) = tokio::sync::mpsc::channel(16);
    let uid_set = SequenceSet::new(format!(
        "{},{},{}",
        expected_uids[0], missing_uid, expected_uids[1]
    ))
    .unwrap();
    conn.uid_fetch_streaming(&uid_set, &[FetchAttr::Uid, FetchAttr::Flags], tx, TIMEOUT)
        .await
        .unwrap();

    let mut returned_uids = Vec::new();
    while let Some(item) = rx.recv().await {
        returned_uids.push(
            item.unwrap()
                .uid
                .expect("UID FETCH response must include UID"),
        );
    }
    returned_uids.sort_unstable();

    assert_eq!(
        returned_uids, expected_uids,
        "UID FETCH over a mixed set must skip missing UID {missing_uid} and return only the existing UIDs"
    );

    conn.close(TIMEOUT).await.unwrap();
    conn.delete(&mbox, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

// ==========================================================================
// MULTIAPPEND (RFC 3502)
// ==========================================================================

#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_multiappend() {
    let conn = connect_stalwart().await;

    let mbox = &format!("test-multiappend-{}", std::process::id());
    conn.create(mbox, TIMEOUT).await.unwrap();

    let messages: Vec<daaki_imap::types::AppendMessage> = (0..3)
        .map(|i| {
            let msg = format!(
                "From: sender@example.com\r\n\
                 To: recipient@example.com\r\n\
                 Subject: Multi {i}\r\n\
                 Date: Sun, 15 Mar 2026 00:00:0{i} +0000\r\n\
                 Message-ID: <multi-{i}-{}@daaki>\r\n\
                 MIME-Version: 1.0\r\n\
                 Content-Type: text/plain\r\n\
                 \r\n\
                 Body {i}\r\n",
                std::process::id()
            );
            daaki_imap::types::AppendMessage::new(msg.into_bytes())
        })
        .collect();

    match conn.multi_append(mbox, &messages, TIMEOUT).await {
        Ok(append_uids) => {
            let _ = append_uids;
        }
        Err(daaki_imap::Error::MissingCapability(_)) => {
            // Fall back to individual APPENDs.
            eprintln!("skipping MULTIAPPEND: not supported");
        }
        Err(e) => panic!("unexpected error from MULTIAPPEND: {e:?}"),
    }

    // Verify via STATUS.
    let items = conn
        .status(mbox, "(MESSAGES)", TIMEOUT)
        .await
        .unwrap()
        .items;
    let msg_count = items.iter().find_map(|i| {
        if let daaki_imap::types::StatusItem::Messages(n) = i {
            Some(*n)
        } else {
            None
        }
    });
    assert!(
        msg_count.unwrap_or(0) >= 3,
        "expected at least 3 messages after MULTIAPPEND: {items:?}"
    );

    conn.delete(mbox, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

// ##########################################################################
// IMAP4rev2-SPECIFIC TESTS (RFC 9051)
//
// These tests exercise protocol behaviors that are unique to rev2 or that
// changed between rev1 and rev2. They only pass against a server that
// advertises IMAP4rev2 (like Stalwart).
// ##########################################################################

// ==========================================================================
// ENABLE IMAP4rev2 (RFC 9051 Section 6.3.1)
// ==========================================================================

/// On a dual-mode server (rev1 + rev2), the client must ENABLE IMAP4rev2
/// to activate rev2 behavior. Verify this negotiation works.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_rev2_enable_imap4rev2() {
    let conn = connect_stalwart().await;

    let caps = conn.capabilities();
    let is_dual = caps.iter().any(|c| matches!(c, Capability::Imap4Rev1))
        && caps.iter().any(|c| matches!(c, Capability::Imap4Rev2));

    if !is_dual {
        eprintln!("skipping: Stalwart is not a dual-mode rev1+rev2 server");
        conn.logout().await.unwrap();
        return;
    }

    let enabled = conn.enable(&["IMAP4rev2"], TIMEOUT).await.unwrap();
    assert!(
        enabled.iter().any(|s| s.eq_ignore_ascii_case("IMAP4rev2")),
        "expected IMAP4rev2 to be enabled: {enabled:?}"
    );

    conn.logout().await.unwrap();
}

// ==========================================================================
// CHECK rejected in rev2 (RFC 9051 Appendix F item 17)
// ==========================================================================

/// CHECK was removed in IMAP4rev2. Our client should refuse to send it
/// when the server is in rev2 mode.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_rev2_check_rejected() {
    let conn = connect_stalwart().await;

    let _ = conn.enable(&["IMAP4rev2"], TIMEOUT).await;
    conn.select("INBOX", TIMEOUT).await.unwrap();

    let result = conn.check(TIMEOUT).await;
    assert!(
        matches!(result, Err(daaki_imap::Error::Protocol(_))),
        "expected Protocol error for CHECK in rev2 mode, got: {result:?}"
    );

    conn.logout().await.unwrap();
}

// ==========================================================================
// STATUS=SIZE (RFC 8438, base in rev2)
// ==========================================================================

/// STATUS with SIZE — total mailbox size in octets (RFC 9051 Section 6.3.11).
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_rev2_status_size() {
    let conn = connect_stalwart().await;

    let mbox = &format!("test-status-size-{}", std::process::id());
    conn.create(mbox, TIMEOUT).await.unwrap();
    conn.append(mbox, &[], None, TEST_MESSAGE, TIMEOUT)
        .await
        .unwrap();

    let items = conn
        .status(mbox, "(MESSAGES SIZE)", TIMEOUT)
        .await
        .unwrap()
        .items;

    let size = items.iter().find_map(|i| {
        if let daaki_imap::types::StatusItem::Size(s) = i {
            Some(*s)
        } else {
            None
        }
    });
    assert!(
        size.is_some() && size.unwrap() > 0,
        "expected SIZE > 0 in STATUS response: {items:?}"
    );

    conn.delete(mbox, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

// ==========================================================================
// OBJECTID — EMAILID and THREADID (RFC 8474, base in rev2)
// ==========================================================================

/// FETCH with EMAILID and THREADID — unique object identifiers (RFC 8474).
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_rev2_objectid() {
    let conn = connect_stalwart().await;

    if !conn
        .capabilities()
        .iter()
        .any(|c| matches!(c, Capability::ObjectId))
    {
        eprintln!("skipping: OBJECTID not advertised");
        conn.logout().await.unwrap();
        return;
    }

    let mbox = &format!("test-objectid-{}", std::process::id());
    conn.create(mbox, TIMEOUT).await.unwrap();
    conn.append(mbox, &[], None, TEST_MESSAGE, TIMEOUT)
        .await
        .unwrap();
    conn.select(mbox, TIMEOUT).await.unwrap();

    let fetches = conn
        .fetch(
            &SequenceSet::new("1").unwrap(),
            &[FetchAttr::Uid, FetchAttr::EmailId, FetchAttr::ThreadId],
            TIMEOUT,
        )
        .await
        .unwrap();

    assert!(!fetches.is_empty());
    let msg = &fetches[0];
    assert!(msg.email_id.is_some(), "EMAILID should be present");
    assert!(msg.thread_id.is_some(), "THREADID should be present");

    conn.delete(mbox, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

/// COPY must preserve both EMAILID and THREADID for the destination message.
///
/// RFC 8474 Section 5.1: the server MUST return the same EMAILID as the
/// source message for the matching destination message after COPY.
/// RFC 8474 Section 5.2: messages with the same EMAILID MUST have the same
/// THREADID.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_rev2_copy_preserves_emailid_and_threadid() {
    let conn = connect_stalwart().await;

    if !has_capability(&conn, &Capability::ObjectId) {
        eprintln!("skipping: OBJECTID not advertised");
        conn.logout().await.unwrap();
        return;
    }

    let src = &format!("test-objectid-copy-src-{}", std::process::id());
    let dst = &format!("test-objectid-copy-dst-{}", std::process::id());
    let message = build_message(
        "Stalwart OBJECTID COPY",
        &format!("stalwart-objectid-copy-{}@daaki", std::process::id()),
        "OBJECTID COPY body",
    );
    conn.create(src, TIMEOUT).await.unwrap();
    conn.create(dst, TIMEOUT).await.unwrap();
    conn.append(src, &[], None, &message, TIMEOUT)
        .await
        .unwrap();
    conn.select(src, TIMEOUT).await.unwrap();

    let source = conn
        .uid_fetch(
            &SequenceSet::new("1:*").unwrap(),
            &[FetchAttr::Uid, FetchAttr::EmailId, FetchAttr::ThreadId],
            TIMEOUT,
        )
        .await
        .unwrap();
    assert_eq!(source.len(), 1, "expected 1 source message");

    let uid = source[0].uid.unwrap();
    let email_id = source[0]
        .email_id
        .clone()
        .expect("OBJECTID-capable server must return EMAILID");
    let thread_id = source[0]
        .thread_id
        .clone()
        .expect("Stalwart should return THREADID");

    conn.uid_copy(&SequenceSet::new(uid.to_string()).unwrap(), dst, TIMEOUT)
        .await
        .unwrap();

    conn.select(dst, TIMEOUT).await.unwrap();
    let copied = conn
        .uid_fetch(
            &SequenceSet::new("1:*").unwrap(),
            &[FetchAttr::Uid, FetchAttr::EmailId, FetchAttr::ThreadId],
            TIMEOUT,
        )
        .await
        .unwrap();
    assert_eq!(copied.len(), 1, "expected 1 copied message");
    assert_eq!(
        copied[0].email_id.as_deref(),
        Some(email_id.as_str()),
        "COPY must preserve EMAILID"
    );
    assert_eq!(
        copied[0].thread_id.as_deref(),
        Some(thread_id.as_str()),
        "COPY must preserve THREADID for the copied message"
    );

    conn.close(TIMEOUT).await.unwrap();
    conn.delete(src, TIMEOUT).await.unwrap();
    conn.delete(dst, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

/// MOVE must preserve both EMAILID and THREADID for the destination message.
///
/// RFC 8474 Section 5.1: the server MUST return the same EMAILID as the
/// source message for the matching destination message after MOVE.
/// RFC 8474 Section 5.2: messages with the same EMAILID MUST have the same
/// THREADID.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_rev2_move_preserves_emailid_and_threadid() {
    let conn = connect_stalwart().await;

    if !has_capability(&conn, &Capability::ObjectId) {
        eprintln!("skipping: OBJECTID not advertised");
        conn.logout().await.unwrap();
        return;
    }

    let src = &format!("test-objectid-move-src-{}", std::process::id());
    let dst = &format!("test-objectid-move-dst-{}", std::process::id());
    let message = build_message(
        "Stalwart OBJECTID MOVE",
        &format!("stalwart-objectid-move-{}@daaki", std::process::id()),
        "OBJECTID MOVE body",
    );
    conn.create(src, TIMEOUT).await.unwrap();
    conn.create(dst, TIMEOUT).await.unwrap();
    conn.append(src, &[], None, &message, TIMEOUT)
        .await
        .unwrap();
    conn.select(src, TIMEOUT).await.unwrap();

    let source = conn
        .uid_fetch(
            &SequenceSet::new("1:*").unwrap(),
            &[FetchAttr::Uid, FetchAttr::EmailId, FetchAttr::ThreadId],
            TIMEOUT,
        )
        .await
        .unwrap();
    assert_eq!(source.len(), 1, "expected 1 source message");

    let uid = source[0].uid.unwrap();
    let email_id = source[0]
        .email_id
        .clone()
        .expect("OBJECTID-capable server must return EMAILID");
    let thread_id = source[0]
        .thread_id
        .clone()
        .expect("Stalwart should return THREADID");

    conn.uid_move_messages(&SequenceSet::new(uid.to_string()).unwrap(), dst, TIMEOUT)
        .await
        .unwrap();

    conn.select(dst, TIMEOUT).await.unwrap();
    let moved = conn
        .uid_fetch(
            &SequenceSet::new("1:*").unwrap(),
            &[FetchAttr::Uid, FetchAttr::EmailId, FetchAttr::ThreadId],
            TIMEOUT,
        )
        .await
        .unwrap();
    assert_eq!(moved.len(), 1, "expected 1 moved message");
    assert_eq!(
        moved[0].email_id.as_deref(),
        Some(email_id.as_str()),
        "MOVE must preserve EMAILID"
    );
    assert_eq!(
        moved[0].thread_id.as_deref(),
        Some(thread_id.as_str()),
        "MOVE must preserve THREADID for the moved message"
    );

    conn.close(TIMEOUT).await.unwrap();
    conn.delete(src, TIMEOUT).await.unwrap();
    conn.delete(dst, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

/// RENAME must preserve MAILBOXID for the renamed mailbox.
///
/// RFC 8474 Section 4: the server MUST keep the same MAILBOXID for the source
/// and destination when renaming a mailbox while preserving the same messages.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_rev2_rename_preserves_mailboxid() {
    let conn = connect_stalwart().await;

    if !has_capability(&conn, &Capability::ObjectId) {
        eprintln!("skipping: OBJECTID not advertised");
        conn.logout().await.unwrap();
        return;
    }

    let src = &format!("test-mailboxid-rename-src-{}", std::process::id());
    let dst = &format!("test-mailboxid-rename-dst-{}", std::process::id());
    conn.create(src, TIMEOUT).await.unwrap();

    let before = conn.status(src, "(MAILBOXID)", TIMEOUT).await.unwrap();
    let mailbox_id_before = before.items.iter().find_map(|item| {
        if let daaki_imap::types::StatusItem::MailboxId(id) = item {
            Some(id.clone())
        } else {
            None
        }
    });
    let mailbox_id_before =
        mailbox_id_before.expect("OBJECTID-capable server must return MAILBOXID");

    conn.rename(src, dst, TIMEOUT).await.unwrap();

    let after = conn.status(dst, "(MAILBOXID)", TIMEOUT).await.unwrap();
    let mailbox_id_after = after.items.iter().find_map(|item| {
        if let daaki_imap::types::StatusItem::MailboxId(id) = item {
            Some(id.clone())
        } else {
            None
        }
    });
    let mailbox_id_after = mailbox_id_after.expect("renamed mailbox must still report MAILBOXID");

    assert_eq!(
        mailbox_id_after, mailbox_id_before,
        "RENAME must preserve MAILBOXID"
    );

    conn.delete(dst, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

// ==========================================================================
// SAVEDATE (RFC 8514, base in rev2)
// ==========================================================================

/// FETCH with SAVEDATE — when the message was saved (RFC 8514).
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_rev2_savedate() {
    let conn = connect_stalwart().await;

    if !conn
        .capabilities()
        .iter()
        .any(|c| matches!(c, Capability::SaveDate))
    {
        eprintln!("skipping: SAVEDATE not advertised");
        conn.logout().await.unwrap();
        return;
    }

    let mbox = &format!("test-savedate-{}", std::process::id());
    conn.create(mbox, TIMEOUT).await.unwrap();
    conn.append(mbox, &[], None, TEST_MESSAGE, TIMEOUT)
        .await
        .unwrap();
    conn.select(mbox, TIMEOUT).await.unwrap();

    let fetches = conn
        .fetch(
            &SequenceSet::new("1").unwrap(),
            &[FetchAttr::Uid, FetchAttr::SaveDate],
            TIMEOUT,
        )
        .await
        .unwrap();

    assert!(!fetches.is_empty());
    assert!(fetches[0].save_date.is_some(), "SAVEDATE should be present");

    conn.delete(mbox, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

/// `UID SEARCH SAVEDSINCE` must match a freshly appended message.
///
/// RFC 8514 Section 4.3: `SAVEDSINCE <date>` matches messages whose save date
/// is within or later than the specified date.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_rev2_uid_search_savedsince_matches_recent_append() {
    let conn = connect_stalwart().await;

    if !has_capability(&conn, &Capability::SaveDate) {
        eprintln!("skipping: SAVEDATE not advertised");
        conn.logout().await.unwrap();
        return;
    }

    let mbox = &format!("test-savedsince-{}", std::process::id());
    let message = build_message(
        "Stalwart SAVEDSINCE search",
        &format!("stalwart-savedsince-{}@daaki", std::process::id()),
        "SAVEDSINCE search body",
    );
    conn.create(mbox, TIMEOUT).await.unwrap();
    conn.append(mbox, &[], None, &message, TIMEOUT)
        .await
        .unwrap();
    conn.select(mbox, TIMEOUT).await.unwrap();

    let fetches = conn
        .uid_fetch(
            &SequenceSet::new("1:*").unwrap(),
            &[FetchAttr::Uid, FetchAttr::SaveDate],
            TIMEOUT,
        )
        .await
        .unwrap();
    assert_eq!(fetches.len(), 1, "expected 1 message in test mailbox");
    let uid = fetches[0].uid.unwrap();
    assert!(
        fetches[0].save_date.is_some(),
        "SAVEDSINCE test requires a server-reported SAVEDATE"
    );

    let result = conn
        .uid_search("SAVEDSINCE 1-Jan-2000", TIMEOUT)
        .await
        .unwrap();
    assert_eq!(
        result.ids,
        vec![uid],
        "UID SEARCH SAVEDSINCE must match the freshly appended message"
    );

    conn.close(TIMEOUT).await.unwrap();
    conn.delete(mbox, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

/// `UID SEARCH SAVEDATESUPPORTED` must match all messages in a mailbox whose
/// storage reports save dates.
///
/// RFC 8514 Section 4.3: `SAVEDATESUPPORTED` matches all messages when the
/// underlying mailbox storage supports the save date attribute.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_rev2_uid_search_savedatesupported_matches_message() {
    let conn = connect_stalwart().await;

    if !has_capability(&conn, &Capability::SaveDate) {
        eprintln!("skipping: SAVEDATE not advertised");
        conn.logout().await.unwrap();
        return;
    }

    let mbox = &format!("test-savedatesupported-{}", std::process::id());
    let message = build_message(
        "Stalwart SAVEDATESUPPORTED search",
        &format!("stalwart-savedatesupported-{}@daaki", std::process::id()),
        "SAVEDATESUPPORTED search body",
    );
    conn.create(mbox, TIMEOUT).await.unwrap();
    conn.append(mbox, &[], None, &message, TIMEOUT)
        .await
        .unwrap();
    conn.select(mbox, TIMEOUT).await.unwrap();

    let fetches = conn
        .uid_fetch(
            &SequenceSet::new("1:*").unwrap(),
            &[FetchAttr::Uid, FetchAttr::SaveDate],
            TIMEOUT,
        )
        .await
        .unwrap();
    assert_eq!(fetches.len(), 1, "expected 1 message in test mailbox");
    let uid = fetches[0].uid.unwrap();
    assert!(
        fetches[0].save_date.is_some(),
        "SAVEDATESUPPORTED test requires a server-reported SAVEDATE"
    );

    let result = conn.uid_search("SAVEDATESUPPORTED", TIMEOUT).await.unwrap();
    assert_eq!(
        result.ids,
        vec![uid],
        "UID SEARCH SAVEDATESUPPORTED must match all messages in a SAVEDATE-capable mailbox"
    );

    conn.close(TIMEOUT).await.unwrap();
    conn.delete(mbox, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

/// `UID SEARCH SAVEDBEFORE` must not match a freshly appended message when the
/// cutoff date is far in the past.
///
/// RFC 8514 Section 4.3: `SAVEDBEFORE <date>` matches messages whose save date
/// is earlier than the specified date.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_rev2_uid_search_savedbefore_excludes_recent_append() {
    let conn = connect_stalwart().await;

    if !has_capability(&conn, &Capability::SaveDate) {
        eprintln!("skipping: SAVEDATE not advertised");
        conn.logout().await.unwrap();
        return;
    }

    let mbox = &format!("test-savedbefore-{}", std::process::id());
    let message = build_message(
        "Stalwart SAVEDBEFORE search",
        &format!("stalwart-savedbefore-{}@daaki", std::process::id()),
        "SAVEDBEFORE search body",
    );
    conn.create(mbox, TIMEOUT).await.unwrap();
    conn.append(mbox, &[], None, &message, TIMEOUT)
        .await
        .unwrap();
    conn.select(mbox, TIMEOUT).await.unwrap();

    let result = conn
        .uid_search("SAVEDBEFORE 1-Jan-2000", TIMEOUT)
        .await
        .unwrap();
    assert!(
        result.ids.is_empty(),
        "UID SEARCH SAVEDBEFORE must exclude a freshly appended message: {:?}",
        result.ids
    );

    conn.close(TIMEOUT).await.unwrap();
    conn.delete(mbox, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

// ==========================================================================
// STATUS MAILBOXID (RFC 8474, OBJECTID base in rev2)
// ==========================================================================

/// STATUS with MAILBOXID — unique mailbox identifier (RFC 8474 Section 5.1).
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_rev2_status_mailboxid() {
    let conn = connect_stalwart().await;

    if !conn
        .capabilities()
        .iter()
        .any(|c| matches!(c, Capability::ObjectId))
    {
        eprintln!("skipping: OBJECTID not advertised");
        conn.logout().await.unwrap();
        return;
    }

    let items = conn
        .status("INBOX", "(MAILBOXID)", TIMEOUT)
        .await
        .unwrap()
        .items;

    let mailbox_id = items.iter().find_map(|i| {
        if let daaki_imap::types::StatusItem::MailboxId(id) = i {
            Some(id.clone())
        } else {
            None
        }
    });

    assert!(mailbox_id.is_some(), "expected MAILBOXID: {items:?}");
    assert!(
        !mailbox_id.unwrap().is_empty(),
        "MAILBOXID should not be empty"
    );

    conn.logout().await.unwrap();
}

// ==========================================================================
// LIST-STATUS (RFC 5819, base in rev2)
// ==========================================================================

/// LIST-STATUS — LIST and STATUS combined (RFC 9051 Section 6.3.9).
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_rev2_list_status() {
    let conn = connect_stalwart().await;

    match conn
        .list_status("", "*", "(MESSAGES UNSEEN)", TIMEOUT)
        .await
    {
        Ok(results) => {
            assert!(
                results
                    .iter()
                    .any(|(m, _)| m.name.as_str().eq_ignore_ascii_case("INBOX")),
                "expected INBOX in LIST-STATUS response"
            );
            for (mbox, status_items) in &results {
                assert!(
                    !status_items.is_empty(),
                    "expected STATUS items for '{}': {status_items:?}",
                    mbox.name
                );
            }
        }
        Err(daaki_imap::Error::MissingCapability(_)) => {
            eprintln!("skipping: LIST-STATUS not advertised");
        }
        Err(daaki_imap::Error::Bad { .. }) => {
            // Stalwart may reject certain STATUS item formats in LIST-STATUS.
            eprintln!("skipping: Stalwart rejected LIST-STATUS syntax");
        }
        Err(e) => panic!("unexpected error from LIST-STATUS: {e:?}"),
    }

    conn.logout().await.unwrap();
}

/// LIST-STATUS must resolve a relative mailbox name against the reference name
/// and correlate the returned STATUS data with that mailbox.
///
/// RFC 5819 Section 2 extends LIST with STATUS data, and RFC 3501 Section
/// 6.3.8 defines how the reference name scopes relative mailbox names.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_rev2_list_status_reference_resolves_relative_mailbox_name() {
    let conn = connect_stalwart().await;

    let delimiter = hierarchy_delimiter(&conn).await;
    let parent = unique_name("test-list-status-reference-parent");
    let child = format!("{parent}{delimiter}child");
    let reference = format!("{parent}{delimiter}");

    conn.create(&parent, TIMEOUT).await.unwrap();
    conn.create(&child, TIMEOUT).await.unwrap();
    conn.append(&child, &[], None, TEST_MESSAGE, TIMEOUT)
        .await
        .unwrap();
    conn.append(&child, &[], None, TEST_MESSAGE, TIMEOUT)
        .await
        .unwrap();

    let results = conn
        .list_status(&reference, "child", "(MESSAGES UIDNEXT)", TIMEOUT)
        .await
        .unwrap();
    assert_eq!(
        results.len(),
        1,
        "LIST-STATUS with a relative mailbox name should resolve to one canonical match: {results:?}"
    );

    let (info, items) = &results[0];
    assert_eq!(info.name.as_str(), child.as_str());
    assert_eq!(
        status_messages(items),
        Some(2),
        "expected two appended messages: {items:?}"
    );
    assert!(
        status_uidnext(items).is_some_and(|uid_next| uid_next >= 3),
        "UIDNEXT should advance past the appended messages: {items:?}"
    );

    delete_mailbox_and_contents(&conn, &child).await;
    conn.delete(&parent, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

/// LIST-STATUS with "%" and a non-empty reference name must correlate STATUS
/// data only for the direct children below that reference.
///
/// RFC 5819 Section 2 extends LIST with STATUS data, while RFC 3501 Section
/// 6.3.8 restricts "%" to a single hierarchy level under the reference.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_rev2_list_status_reference_percent_matches_only_direct_children() {
    let conn = connect_stalwart().await;

    let delimiter = hierarchy_delimiter(&conn).await;
    let parent = unique_name("test-list-status-percent-parent");
    let child_one = format!("{parent}{delimiter}child-one");
    let child_two = format!("{parent}{delimiter}child-two");
    let grandchild = format!("{child_one}{delimiter}grandchild");
    let reference = format!("{parent}{delimiter}");

    conn.create(&parent, TIMEOUT).await.unwrap();
    conn.create(&child_one, TIMEOUT).await.unwrap();
    conn.create(&child_two, TIMEOUT).await.unwrap();
    conn.create(&grandchild, TIMEOUT).await.unwrap();
    conn.append(&child_one, &[], None, TEST_MESSAGE, TIMEOUT)
        .await
        .unwrap();
    conn.append(&child_two, &[], None, TEST_MESSAGE, TIMEOUT)
        .await
        .unwrap();
    conn.append(&child_two, &[], None, TEST_MESSAGE, TIMEOUT)
        .await
        .unwrap();

    let results = conn
        .list_status(&reference, "%", "MESSAGES UIDNEXT", TIMEOUT)
        .await
        .unwrap();
    let mut names: Vec<_> = results
        .iter()
        .map(|(info, _)| info.name.as_str().to_owned())
        .collect();
    names.sort();
    assert_eq!(
        names,
        vec![child_one.clone(), child_two.clone()],
        "LIST-STATUS % under a reference must return only direct children: {results:?}"
    );

    let child_one_items = results
        .iter()
        .find_map(|(info, items)| (info.name.as_str() == child_one).then_some(items))
        .expect("expected STATUS items for first direct child");
    assert_eq!(
        status_messages(child_one_items),
        Some(1),
        "expected one message for first direct child: {child_one_items:?}"
    );

    let child_two_items = results
        .iter()
        .find_map(|(info, items)| (info.name.as_str() == child_two).then_some(items))
        .expect("expected STATUS items for second direct child");
    assert_eq!(
        status_messages(child_two_items),
        Some(2),
        "expected two messages for second direct child: {child_two_items:?}"
    );

    delete_mailbox_and_contents(&conn, &grandchild).await;
    delete_mailbox_and_contents(&conn, &child_one).await;
    delete_mailbox_and_contents(&conn, &child_two).await;
    conn.delete(&parent, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

/// LIST-STATUS for an exact mailbox must return that mailbox's status data
/// without selecting it.
///
/// RFC 5819 Section 2 extends LIST with STATUS data, and RFC 3501 Section
/// 6.3.10 defines `MESSAGES` and `UIDNEXT` for the requested mailbox.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_rev2_list_status_exact_mailbox_reports_message_count() {
    let conn = connect_stalwart().await;

    let mailbox = unique_name("test-stalwart-list-status-exact");
    conn.create(&mailbox, TIMEOUT).await.unwrap();
    conn.append(&mailbox, &[], None, TEST_MESSAGE, TIMEOUT)
        .await
        .unwrap();
    conn.append(&mailbox, &[], None, TEST_MESSAGE, TIMEOUT)
        .await
        .unwrap();

    let results = conn
        .list_status("", &mailbox, "MESSAGES UIDNEXT", TIMEOUT)
        .await
        .unwrap();
    assert_eq!(
        results.len(),
        1,
        "LIST-STATUS should return exactly one mailbox: {results:?}"
    );
    assert_eq!(results[0].0.name.as_str(), mailbox.as_str());
    assert_eq!(
        status_messages(&results[0].1),
        Some(2),
        "LIST-STATUS must report the appended message count: {:?}",
        results[0].1
    );
    assert!(
        status_uidnext(&results[0].1).is_some_and(|uid_next| uid_next >= 3),
        "LIST-STATUS must report UIDNEXT after the appended messages: {:?}",
        results[0].1
    );

    delete_mailbox_and_contents(&conn, &mailbox).await;
    conn.logout().await.unwrap();
}

/// LIST-STATUS with a wildcard pattern must correlate each returned mailbox
/// with its own status data.
///
/// RFC 5819 Section 2 returns interleaved LIST and STATUS responses, so the
/// client must pair each mailbox with the matching STATUS item list.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_rev2_list_status_wildcard_correlates_each_mailbox() {
    let conn = connect_stalwart().await;

    let prefix = unique_name("test-stalwart-list-status-wildcard");
    let mailbox_one = format!("{prefix}-one");
    let mailbox_two = format!("{prefix}-two");
    conn.create(&mailbox_one, TIMEOUT).await.unwrap();
    conn.create(&mailbox_two, TIMEOUT).await.unwrap();
    conn.append(&mailbox_one, &[], None, TEST_MESSAGE, TIMEOUT)
        .await
        .unwrap();
    conn.append(&mailbox_two, &[], None, TEST_MESSAGE, TIMEOUT)
        .await
        .unwrap();
    conn.append(&mailbox_two, &[], None, TEST_MESSAGE, TIMEOUT)
        .await
        .unwrap();

    let results = conn
        .list_status("", &format!("{prefix}*"), "MESSAGES", TIMEOUT)
        .await
        .unwrap();
    assert_eq!(
        results.len(),
        2,
        "LIST-STATUS wildcard should return exactly the two matching mailboxes: {results:?}"
    );

    let first = results
        .iter()
        .find(|(info, _)| info.name.as_str() == mailbox_one)
        .unwrap_or_else(|| panic!("missing LIST-STATUS entry for {mailbox_one}: {results:?}"));
    let second = results
        .iter()
        .find(|(info, _)| info.name.as_str() == mailbox_two)
        .unwrap_or_else(|| panic!("missing LIST-STATUS entry for {mailbox_two}: {results:?}"));

    assert_eq!(status_messages(&first.1), Some(1));
    assert_eq!(status_messages(&second.1), Some(2));

    delete_mailbox_and_contents(&conn, &mailbox_one).await;
    delete_mailbox_and_contents(&conn, &mailbox_two).await;
    conn.logout().await.unwrap();
}

// ==========================================================================
// SPECIAL-USE (RFC 6154, base in rev2)
// ==========================================================================

/// LIST with the `SPECIAL-USE` selection option must return only special-use
/// mailboxes.
///
/// RFC 6154 Section 2 defines the `SPECIAL-USE` selection option for LIST and
/// requires the server to limit the response to mailboxes carrying a
/// special-use attribute.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_rev2_special_use() {
    let conn = connect_stalwart().await;

    if !has_capability(&conn, &Capability::CreateSpecialUse)
        || !has_capability(&conn, &Capability::SpecialUse)
    {
        eprintln!("skipping: CREATE-SPECIAL-USE or SPECIAL-USE not advertised");
        conn.logout().await.unwrap();
        return;
    }

    let _special_use_role_guard = special_use_role_lock().lock().await;
    let special_mbox = unique_name("test-special-use-selected");
    let regular_mbox = unique_name("test-regular-selected");
    let special_use = available_special_use_role(&conn).await;

    conn.create_special_use(&special_mbox, std::slice::from_ref(&special_use), TIMEOUT)
        .await
        .unwrap_or_else(|e| {
            panic!("unexpected error creating special-use mailbox with role {special_use:?}: {e}")
        });
    conn.create(&regular_mbox, TIMEOUT).await.unwrap();

    let listing = conn
        .list_extended("", &["*"], &["SPECIAL-USE"], &[], TIMEOUT)
        .await
        .unwrap();

    assert!(
        listing
            .iter()
            .any(|mailbox| mailbox.name.as_str() == special_mbox),
        "LIST SPECIAL-USE must include the mailbox created with USE={special_use:?}: {listing:?}"
    );
    assert!(
        listing
            .iter()
            .all(|mailbox| mailbox.name.as_str() != regular_mbox),
        "LIST SPECIAL-USE must exclude ordinary mailboxes: {listing:?}"
    );

    conn.delete(&regular_mbox, TIMEOUT).await.unwrap();
    conn.delete(&special_mbox, TIMEOUT).await.unwrap();

    conn.logout().await.unwrap();
}

/// `LIST (SUBSCRIBED)` must return only subscribed mailboxes that match the
/// requested pattern and surface `\\Subscribed` on those entries.
///
/// RFC 5258 Section 3.4 defines the `SUBSCRIBED` selection option and requires
/// LIST to limit the result set to subscribed mailboxes.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_rev2_list_extended_subscribed_filter() {
    let conn = connect_stalwart().await;

    let prefix = unique_name("test-stalwart-subscribed-filter");
    let subscribed_mailbox = format!("{prefix}-yes");
    let unsubscribed_mailbox = format!("{prefix}-no");
    let pattern = format!("{prefix}*");

    conn.create(&subscribed_mailbox, TIMEOUT).await.unwrap();
    conn.create(&unsubscribed_mailbox, TIMEOUT).await.unwrap();
    conn.subscribe(&subscribed_mailbox, TIMEOUT).await.unwrap();

    let listing = conn
        .list_extended("", &[&pattern], &["SUBSCRIBED"], &[], TIMEOUT)
        .await
        .unwrap();

    let subscribed_entry = listing
        .iter()
        .find(|mailbox| mailbox.name.as_str() == subscribed_mailbox)
        .unwrap_or_else(|| {
            panic!(
                "subscribed mailbox {subscribed_mailbox} must appear in LIST (SUBSCRIBED): {listing:?}"
            )
        });
    assert!(
        subscribed_entry
            .attributes
            .iter()
            .any(|attr| matches!(attr, MailboxAttribute::Subscribed)),
        "subscribed mailbox must carry the \\Subscribed attribute: {:?}",
        subscribed_entry.attributes
    );
    assert!(
        !listing
            .iter()
            .any(|mailbox| mailbox.name.as_str() == unsubscribed_mailbox),
        "unsubscribed mailbox must not appear in LIST (SUBSCRIBED): {listing:?}"
    );

    conn.unsubscribe(&subscribed_mailbox, TIMEOUT)
        .await
        .unwrap();
    conn.delete(&subscribed_mailbox, TIMEOUT).await.unwrap();
    conn.delete(&unsubscribed_mailbox, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

// ==========================================================================
// ACL / MYRIGHTS (RFC 4314)
// ==========================================================================

/// LISTRIGHTS must not repeat any right across the required and optional
/// groups.
///
/// RFC 4314 Section 3.7: "The same right MUST NOT be listed more than once in
/// the LISTRIGHTS command."
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_rev2_list_rights_does_not_repeat_any_right() {
    let conn = connect_stalwart().await;

    if !has_capability(&conn, &Capability::Acl) {
        eprintln!("skipping: ACL not advertised");
        conn.logout().await.unwrap();
        return;
    }

    let mailbox = unique_name("test-stalwart-list-rights-unique");
    conn.create(&mailbox, TIMEOUT).await.unwrap();

    let rights = conn.list_rights(&mailbox, "anyone", TIMEOUT).await.unwrap();
    let mut seen = BTreeSet::new();
    for ch in rights
        .required
        .chars()
        .chain(rights.optional.iter().flat_map(|group| group.chars()))
    {
        assert!(
            seen.insert(ch),
            "LISTRIGHTS must not repeat right {ch:?}; got {rights:?}"
        );
    }

    conn.delete(&mailbox, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

/// GETACL on a selected mailbox must leave the mailbox selected.
///
/// RFC 4314 Section 3.3 defines GETACL as returning the identifier/rights
/// pairs for a mailbox. RFC 3501 Section 6.4 says commands valid in selected
/// state must not implicitly deselect the current mailbox.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_rev2_get_acl_selected_mailbox_preserves_selected_state() {
    let conn = connect_stalwart().await;

    if !has_capability(&conn, &Capability::Acl) {
        eprintln!("skipping: ACL not advertised");
        conn.logout().await.unwrap();
        return;
    }

    let mailbox = unique_name("test-stalwart-acl-get-selected");
    conn.create(&mailbox, TIMEOUT).await.unwrap();

    conn.append(&mailbox, &[], None, TEST_MESSAGE, TIMEOUT)
        .await
        .unwrap();
    conn.select(&mailbox, TIMEOUT).await.unwrap();

    conn.get_acl(&mailbox, TIMEOUT).await.unwrap();
    assert_selected_mailbox_still_usable(&conn, "GETACL").await;

    conn.close(TIMEOUT).await.unwrap();
    conn.delete(&mailbox, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

/// LISTRIGHTS on a selected mailbox must not deselect the mailbox.
///
/// RFC 4314 Section 3.4 defines LISTRIGHTS as a mailbox/identifier query and
/// RFC 3501 Section 6.4 allows such commands to execute while a mailbox
/// remains selected.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_rev2_list_rights_selected_mailbox_preserves_selected_state() {
    let conn = connect_stalwart().await;

    if !has_capability(&conn, &Capability::Acl) {
        eprintln!("skipping: ACL not advertised");
        conn.logout().await.unwrap();
        return;
    }

    let mailbox = unique_name("test-stalwart-acl-list-rights-selected");
    conn.create(&mailbox, TIMEOUT).await.unwrap();
    conn.append(&mailbox, &[], None, TEST_MESSAGE, TIMEOUT)
        .await
        .unwrap();
    conn.select(&mailbox, TIMEOUT).await.unwrap();

    conn.list_rights(&mailbox, "anyone", TIMEOUT).await.unwrap();

    assert_selected_mailbox_still_usable(&conn, "LISTRIGHTS").await;

    conn.close(TIMEOUT).await.unwrap();
    conn.delete(&mailbox, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

/// MYRIGHTS on INBOX should report the authenticated user's rights.
///
/// RFC 4314 Section 3.5 defines MYRIGHTS as returning the current user's
/// rights on the named mailbox.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_rev2_my_rights() {
    let conn = connect_stalwart().await;

    match conn.my_rights("INBOX", TIMEOUT).await {
        Ok(rights) => {
            assert!(!rights.is_empty(), "expected non-empty rights on INBOX");
            assert!(
                rights.contains('l') || rights.contains('r'),
                "expected lookup or read rights, got: '{rights}'"
            );
        }
        Err(daaki_imap::Error::MissingCapability(_)) => {
            eprintln!("skipping: ACL not advertised");
        }
        Err(e) => panic!("unexpected error from MYRIGHTS: {e:?}"),
    }

    conn.logout().await.unwrap();
}

/// MYRIGHTS on a selected mailbox must report the current user's rights
/// without deselecting the mailbox.
///
/// RFC 4314 Section 3.5 returns the rights available to the logged-in user on
/// the mailbox. RFC 3501 Section 6.4 requires selected-state commands to leave
/// the selected mailbox intact.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_rev2_my_rights_selected_mailbox_preserves_selected_state() {
    let conn = connect_stalwart().await;

    if !has_capability(&conn, &Capability::Acl) {
        eprintln!("skipping: ACL not advertised");
        conn.logout().await.unwrap();
        return;
    }

    let mailbox = unique_name("test-stalwart-acl-my-rights-selected");
    conn.create(&mailbox, TIMEOUT).await.unwrap();
    conn.append(&mailbox, &[], None, TEST_MESSAGE, TIMEOUT)
        .await
        .unwrap();
    conn.select(&mailbox, TIMEOUT).await.unwrap();

    let rights = conn.my_rights(&mailbox, TIMEOUT).await.unwrap();
    assert!(
        !rights.is_empty(),
        "MYRIGHTS must report at least one right for the mailbox owner"
    );
    assert_selected_mailbox_still_usable(&conn, "MYRIGHTS").await;

    conn.close(TIMEOUT).await.unwrap();
    conn.delete(&mailbox, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

// ==========================================================================
// QUOTA (RFC 9208, base in rev2)
// ==========================================================================

/// GETQUOTAROOT should succeed when QUOTA is configured for the account.
///
/// RFC 9208 Section 4.1.2 defines GETQUOTAROOT as returning the quota roots
/// and any associated QUOTA data for the named mailbox.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_rev2_quota() {
    let conn = connect_stalwart().await;

    match conn.get_quota_root("INBOX", TIMEOUT).await {
        Ok(result) => {
            let _ = result;
        }
        Err(
            daaki_imap::Error::No { .. }
            | daaki_imap::Error::Bad { .. }
            | daaki_imap::Error::MissingCapability(_),
        ) => {
            eprintln!("skipping: QUOTA not supported or not configured");
        }
        Err(e) => panic!("unexpected error from GETQUOTAROOT: {e:?}"),
    }

    conn.logout().await.unwrap();
}

/// GETQUOTAROOT must return exactly one QUOTA block for each listed root.
///
/// RFC 9208 Section 4.1.2 requires GETQUOTAROOT to return the quota roots for
/// the mailbox, and RFC 9208 Section 4.2.1 defines each QUOTA response as
/// describing one quota root.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_rev2_get_quota_root_reports_resources_for_each_listed_root() {
    let conn = connect_stalwart().await;

    if !has_capability(&conn, &Capability::Quota) {
        eprintln!("skipping: QUOTA not advertised");
        conn.logout().await.unwrap();
        return;
    }

    let quota_root = conn.get_quota_root("INBOX", TIMEOUT).await.unwrap();
    let roots: BTreeSet<_> = quota_root.roots.iter().cloned().collect();
    let resource_roots: BTreeSet<_> = quota_root
        .resources
        .iter()
        .map(|(root, _)| root.clone())
        .collect();

    assert_eq!(
        roots.len(),
        quota_root.roots.len(),
        "GETQUOTAROOT must not repeat quota roots: {:?}",
        quota_root.roots
    );
    assert_eq!(
        resource_roots.len(),
        quota_root.resources.len(),
        "GETQUOTAROOT must not return duplicate QUOTA blocks: {:?}",
        quota_root.resources
    );
    assert_eq!(
        resource_roots, roots,
        "GETQUOTAROOT must return one QUOTA block for each listed root"
    );

    conn.logout().await.unwrap();
}

/// GETQUOTA and GETQUOTAROOT must agree on resource names and limits.
///
/// RFC 9208 Section 4.1.1 and Section 4.1.2 both return QUOTA data for a
/// named quota root. Because these are separate server snapshots, usage may
/// advance between commands, but the resource set and configured limits for
/// the same root must match.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_rev2_get_quota_matches_get_quota_root_resources() {
    let conn = connect_stalwart().await;

    if !has_capability(&conn, &Capability::Quota) {
        eprintln!("skipping: QUOTA not advertised");
        conn.logout().await.unwrap();
        return;
    }

    let quota_root = conn.get_quota_root("INBOX", TIMEOUT).await.unwrap();
    if quota_root.roots.is_empty() {
        eprintln!("skipping: Stalwart reported no quota roots for INBOX");
        conn.logout().await.unwrap();
        return;
    }

    for root in &quota_root.roots {
        let direct = conn.get_quota(root, TIMEOUT).await.unwrap();
        let from_quota_root = quota_root
            .resources
            .iter()
            .find(|(name, _)| name == root)
            .expect("GETQUOTAROOT must include resources for each listed root");
        let direct_limits = quota_resource_limit_map(&direct);
        let root_limits = quota_resource_limit_map(&from_quota_root.1);

        assert_eq!(
            direct.len(),
            direct_limits.len(),
            "GETQUOTA must not repeat resource names for root {root:?}: {direct:?}"
        );
        assert_eq!(
            from_quota_root.1.len(),
            root_limits.len(),
            "GETQUOTAROOT must not repeat resource names for root {root:?}: {:?}",
            from_quota_root.1
        );
        assert_eq!(
            direct_limits, root_limits,
            "GETQUOTA and GETQUOTAROOT must report the same resource names and limits for root {root:?}"
        );
    }

    conn.logout().await.unwrap();
}

/// GETQUOTAROOT on a selected mailbox must not deselect the mailbox.
///
/// RFC 9208 Section 4.1.2 defines GETQUOTAROOT as a mailbox query, and RFC
/// 3501 Section 6.4 requires selected-state commands to leave the current
/// mailbox intact.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_rev2_get_quota_root_selected_preserves_selected_state() {
    let conn = connect_stalwart().await;

    if !has_capability(&conn, &Capability::Quota) {
        eprintln!("skipping: QUOTA not advertised");
        conn.logout().await.unwrap();
        return;
    }

    let mailbox = unique_name("test-stalwart-get-quota-root-selected");
    conn.create(&mailbox, TIMEOUT).await.unwrap();
    conn.append(&mailbox, &[], None, TEST_MESSAGE, TIMEOUT)
        .await
        .unwrap();
    conn.select(&mailbox, TIMEOUT).await.unwrap();

    match conn.get_quota_root(&mailbox, TIMEOUT).await {
        Ok(_) | Err(daaki_imap::Error::No { .. } | daaki_imap::Error::Bad { .. }) => {}
        Err(err) => panic!("unexpected GETQUOTAROOT failure: {err:?}"),
    }

    assert_selected_mailbox_still_usable(&conn, "GETQUOTAROOT").await;

    conn.close(TIMEOUT).await.unwrap();
    conn.delete(&mailbox, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

// ==========================================================================
// ENABLE UTF8=ACCEPT (RFC 6855, base in rev2)
// ==========================================================================

/// UTF8=ACCEPT negotiation, then verify operations still work.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_rev2_enable_utf8() {
    let conn = connect_stalwart().await;

    if !conn
        .capabilities()
        .iter()
        .any(|c| matches!(c, Capability::Utf8Accept))
    {
        eprintln!("skipping: UTF8=ACCEPT not advertised");
        conn.logout().await.unwrap();
        return;
    }

    let accepted = conn.enable_utf8(TIMEOUT).await.unwrap();
    assert!(accepted, "expected UTF8=ACCEPT to be confirmed");

    // Operations should still work after enabling UTF-8.
    let mailboxes = conn.list("", "INBOX", TIMEOUT).await.unwrap();
    assert!(
        mailboxes
            .iter()
            .any(|m| m.name.as_str().eq_ignore_ascii_case("INBOX")),
        "LIST should work after ENABLE UTF8=ACCEPT"
    );

    conn.logout().await.unwrap();
}

// ==========================================================================
// Pipeline: STATUS + NAMESPACE (first Stalwart pipeline test)
// ==========================================================================

/// Pipeline STATUS and NAMESPACE on a rev2 server.
///
/// Verifies that the pipeline API works correctly on Stalwart (IMAP4rev2).
/// RFC 9051 Section 6.3.11 (STATUS) and RFC 9051 Section 6.3.10 (NAMESPACE).
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_rev2_pipeline_status_and_namespace() {
    let conn = connect_stalwart().await;

    let (status_result, ns_result) = conn
        .pipeline()
        .status(
            daaki_imap::MailboxName::new("INBOX".to_owned()).unwrap(),
            "(MESSAGES UIDNEXT)".to_owned(),
        )
        .namespace()
        .execute()
        .await
        .unwrap();

    // STATUS should return items.
    let status = status_result.unwrap();
    assert!(
        !status.items.is_empty(),
        "STATUS INBOX should return at least one item"
    );

    // NAMESPACE should return a personal namespace.
    let ns = ns_result.unwrap();
    assert!(
        !ns.personal.is_empty(),
        "RFC 2342 Section 5: server MUST return at least one personal namespace"
    );

    conn.logout().await.unwrap();
}

// ==========================================================================
// UID COPY returns COPYUID (RFC 4315 Section 3 / RFC 9051 Appendix E)
// ==========================================================================

/// UID COPY on a rev2 server MUST return a COPYUID response code with
/// valid uid-validity, source-uids, and dest-uids.
///
/// RFC 4315 Section 3 (incorporated into RFC 9051 as UIDPLUS): The server
/// SHOULD respond with a `[COPYUID uid-validity source-uids dest-uids]`
/// response code in the tagged OK.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_rev2_copy_returns_copyuid() {
    let conn = connect_stalwart().await;

    let src = &format!("test-copyuid-src-{}", std::process::id());
    let dst = &format!("test-copyuid-dst-{}", std::process::id());
    conn.create(src, TIMEOUT).await.unwrap();
    conn.create(dst, TIMEOUT).await.unwrap();
    conn.select(src, TIMEOUT).await.unwrap();

    // Append a message.
    conn.append(src, &[], None, TEST_MESSAGE, TIMEOUT)
        .await
        .unwrap();

    // Get source UID.
    let fetches = conn
        .uid_fetch(
            &SequenceSet::new("1:*").unwrap(),
            &[FetchAttr::Uid],
            TIMEOUT,
        )
        .await
        .unwrap();
    assert!(!fetches.is_empty());
    let src_uid = fetches[0].uid.unwrap();

    // UID COPY.
    let copy_result = conn
        .uid_copy(
            &SequenceSet::new(src_uid.to_string()).unwrap(),
            dst,
            TIMEOUT,
        )
        .await
        .unwrap();

    // Verify COPYUID response code.
    match copy_result.code {
        Some(daaki_imap::types::ResponseCode::CopyUid {
            uid_validity,
            source_uids,
            dest_uids,
        }) => {
            assert!(uid_validity > 0, "destination UIDVALIDITY must be positive");
            assert!(!source_uids.is_empty(), "source_uids must not be empty");
            assert!(!dest_uids.is_empty(), "dest_uids must not be empty");
        }
        other => panic!("expected COPYUID response code from rev2 server, got: {other:?}"),
    }

    conn.close(TIMEOUT).await.unwrap();
    conn.delete(src, TIMEOUT).await.unwrap();
    conn.delete(dst, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

// ==========================================================================
// SEARCH with FROM/SUBJECT criteria on rev2
// ==========================================================================

/// Complex SEARCH on an IMAP4rev2 server with FROM and SUBJECT criteria.
///
/// RFC 9051 Section 6.4.4: SEARCH criteria are specified as in rev1.
/// This validates that Stalwart handles multi-criterion searches correctly.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_rev2_search_from_subject() {
    let conn = connect_stalwart().await;

    let mbox = &format!("test-search-crit-{}", std::process::id());
    conn.create(mbox, TIMEOUT).await.unwrap();
    conn.select(mbox, TIMEOUT).await.unwrap();

    // Append messages with different senders/subjects.
    let msg_alice = b"\
From: alice@example.com\r\n\
To: test@example.com\r\n\
Subject: Project update\r\n\
Date: Mon, 10 Mar 2026 10:00:00 +0000\r\n\
Message-ID: <stalwart-alice@daaki>\r\n\
MIME-Version: 1.0\r\n\
Content-Type: text/plain\r\n\
\r\n\
Alice project update.\r\n";

    let msg_bob = b"\
From: bob@example.com\r\n\
To: test@example.com\r\n\
Subject: Budget review\r\n\
Date: Tue, 11 Mar 2026 14:00:00 +0000\r\n\
Message-ID: <stalwart-bob@daaki>\r\n\
MIME-Version: 1.0\r\n\
Content-Type: text/plain\r\n\
\r\n\
Bob budget review.\r\n";

    conn.append(mbox, &[], None, msg_alice, TIMEOUT)
        .await
        .unwrap();
    conn.append(mbox, &[], None, msg_bob, TIMEOUT)
        .await
        .unwrap();

    // Re-select to see messages.
    conn.select(mbox, TIMEOUT).await.unwrap();

    // SEARCH FROM alice — should match 1.
    let result = conn
        .uid_search("FROM alice@example.com", TIMEOUT)
        .await
        .unwrap();
    assert_eq!(
        result.ids.len(),
        1,
        "FROM alice should match 1 message, got: {:?}",
        result.ids
    );

    // SEARCH SUBJECT Budget — should match 1.
    let result = conn.uid_search("SUBJECT Budget", TIMEOUT).await.unwrap();
    assert_eq!(
        result.ids.len(),
        1,
        "SUBJECT Budget should match 1 message, got: {:?}",
        result.ids
    );

    // SEARCH ALL — should match 2.
    let result = conn.uid_search("ALL", TIMEOUT).await.unwrap();
    assert_eq!(
        result.ids.len(),
        2,
        "ALL should match 2 messages, got: {:?}",
        result.ids
    );

    conn.close(TIMEOUT).await.unwrap();
    conn.delete(mbox, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

// ==========================================================================
// CONDSTORE STORE UNCHANGEDSINCE on rev2 (RFC 7162 Section 3.1.3)
// ==========================================================================

/// STORE with a stale UNCHANGEDSINCE value on a rev2 server.
///
/// RFC 7162 Section 3.1.3: When UNCHANGEDSINCE is specified, the server
/// MUST check the message's MODSEQ against the given value. If the message
/// has been modified since that MODSEQ, the server MUST NOT perform the
/// store and SHOULD include a MODIFIED response code.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_rev2_condstore_store_unchangedsince() {
    let conn = connect_stalwart().await;

    let mbox = &format!("test-cs-store-{}", std::process::id());
    conn.create(mbox, TIMEOUT).await.unwrap();
    conn.append(mbox, &[], None, TEST_MESSAGE, TIMEOUT)
        .await
        .unwrap();

    // SELECT with (CONDSTORE) to activate mod-sequence tracking.
    let selected = conn
        .select_with(
            mbox,
            &daaki_imap::types::SelectOptions::condstore(),
            TIMEOUT,
        )
        .await;
    match selected {
        Err(daaki_imap::Error::MissingCapability(_)) => {
            eprintln!("skipping: Stalwart does not support CONDSTORE");
            conn.delete(mbox, TIMEOUT).await.unwrap();
            conn.logout().await.unwrap();
            return;
        }
        Err(e) => panic!("unexpected error from SELECT CONDSTORE: {e:?}"),
        Ok(_) => {}
    }

    // Get initial MODSEQ.
    let fetches = conn
        .uid_fetch(
            &SequenceSet::new("1:*").unwrap(),
            &[FetchAttr::Uid, FetchAttr::Flags, FetchAttr::ModSeq],
            TIMEOUT,
        )
        .await
        .unwrap();
    assert!(!fetches.is_empty());
    let uid = fetches[0].uid.unwrap();
    let old_modseq = fetches[0].mod_seq.unwrap();

    // First STORE to advance the MODSEQ.
    conn.uid_store(
        &SequenceSet::new(uid.to_string()).unwrap(),
        daaki_imap::types::StoreOperation::Add,
        &[Flag::Seen],
        None,
        TIMEOUT,
    )
    .await
    .unwrap();

    // Now try STORE with the OLD MODSEQ — should either:
    // 1. Return MODIFIED response code (the flag change is rejected), or
    // 2. Succeed if server considers it a no-op or is lenient.
    let result = conn
        .uid_store(
            &SequenceSet::new(uid.to_string()).unwrap(),
            daaki_imap::types::StoreOperation::Add,
            &[Flag::Flagged],
            Some(old_modseq),
            TIMEOUT,
        )
        .await;

    // Either Ok or No is acceptable — the key thing is no panic, no protocol error.
    match result {
        Ok(_) | Err(daaki_imap::Error::No { .. }) => {}
        Err(e) => panic!("unexpected error from stale UNCHANGEDSINCE: {e:?}"),
    }

    conn.close(TIMEOUT).await.unwrap();
    conn.delete(mbox, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

// ==========================================================================
// MULTIAPPEND with different flags (RFC 3502 / RFC 9051)
// ==========================================================================

/// Multi-append messages with different per-message flags.
///
/// RFC 3502 Section 3: Each message in a MULTIAPPEND may have its own
/// flag list. Verify that per-message flags are preserved.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_rev2_multi_append_different_flags() {
    let conn = connect_stalwart().await;

    if !has_capability(&conn, &Capability::MultiAppend) {
        eprintln!("skipping: MULTIAPPEND not advertised");
        conn.logout().await.unwrap();
        return;
    }

    let mbox = &format!("test-multi-flags-{}", std::process::id());
    conn.create(mbox, TIMEOUT).await.unwrap();

    let mut msg1 = daaki_imap::types::AppendMessage::new(
        b"From: a@example.com\r\n\
          To: b@example.com\r\n\
          Subject: Msg 1\r\n\
          Date: Sun, 15 Mar 2026 00:00:00 +0000\r\n\
          Message-ID: <multi-flag-1@daaki>\r\n\
          MIME-Version: 1.0\r\n\
          Content-Type: text/plain\r\n\
          \r\n\
          Body 1\r\n"
            .to_vec(),
    );
    msg1.flags = vec![Flag::Seen, Flag::Flagged];

    let mut msg2 = daaki_imap::types::AppendMessage::new(
        b"From: c@example.com\r\n\
          To: d@example.com\r\n\
          Subject: Msg 2\r\n\
          Date: Sun, 15 Mar 2026 00:00:01 +0000\r\n\
          Message-ID: <multi-flag-2@daaki>\r\n\
          MIME-Version: 1.0\r\n\
          Content-Type: text/plain\r\n\
          \r\n\
          Body 2\r\n"
            .to_vec(),
    );
    msg2.flags = vec![Flag::Answered];

    conn.multi_append(mbox, &[msg1, msg2], TIMEOUT)
        .await
        .unwrap();

    // Verify flags were preserved.
    conn.select(mbox, TIMEOUT).await.unwrap();
    let fetches = conn
        .uid_fetch(
            &SequenceSet::new("1:*").unwrap(),
            &[FetchAttr::Uid, FetchAttr::Flags],
            TIMEOUT,
        )
        .await
        .unwrap();
    assert_eq!(fetches.len(), 2, "expected 2 messages after MULTIAPPEND");

    // Message 1 should have \Seen and \Flagged.
    let f1 = fetches[0].flags.as_deref().unwrap_or_default();
    assert!(
        f1.contains(&Flag::Seen) && f1.contains(&Flag::Flagged),
        "msg1 should have \\Seen + \\Flagged, got: {f1:?}"
    );

    // Message 2 should have \Answered.
    let f2 = fetches[1].flags.as_deref().unwrap_or_default();
    assert!(
        f2.contains(&Flag::Answered),
        "msg2 should have \\Answered, got: {f2:?}"
    );

    conn.close(TIMEOUT).await.unwrap();
    conn.delete(mbox, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}

// ==========================================================================
// FETCH PREVIEW (RFC 8970)
// ==========================================================================

/// PREVIEW fetch returns a short plaintext snippet of the message.
///
/// RFC 8970 Section 3: The PREVIEW fetch data item returns a short
/// text preview of the message body. The server generates the preview
/// from the message content.
#[tokio::test]
#[ignore = "requires Docker: docker compose up -d && ./scripts/setup-stalwart.sh"]
async fn stalwart_rev2_fetch_preview() {
    let conn = connect_stalwart().await;

    if !has_capability(&conn, &Capability::Preview) {
        eprintln!("skipping: PREVIEW not advertised");
        conn.logout().await.unwrap();
        return;
    }

    let mbox = &format!("test-preview-{}", std::process::id());
    conn.create(mbox, TIMEOUT).await.unwrap();
    conn.select(mbox, TIMEOUT).await.unwrap();

    conn.append(mbox, &[], None, TEST_MESSAGE, TIMEOUT)
        .await
        .unwrap();

    // Re-select to see the message.
    conn.select(mbox, TIMEOUT).await.unwrap();

    let fetches = conn
        .uid_fetch(
            &SequenceSet::new("1:*").unwrap(),
            &[FetchAttr::Uid, FetchAttr::Preview],
            TIMEOUT,
        )
        .await
        .unwrap();
    assert!(!fetches.is_empty());

    // PREVIEW should return some text (may be NIL for empty body, but
    // our test message has content).
    let msg = &fetches[0];
    assert!(
        msg.preview.is_some(),
        "expected PREVIEW data for a message with body content, got: {msg:?}"
    );
    let preview = msg.preview.as_ref().unwrap();
    assert!(
        !preview.is_empty(),
        "PREVIEW text should be non-empty for a message with body"
    );

    conn.close(TIMEOUT).await.unwrap();
    conn.delete(mbox, TIMEOUT).await.unwrap();
    conn.logout().await.unwrap();
}
