feat(search): filter and sort messages by server-side timestamps

POST /api/v1/search-messages previously filtered and sorted only on the
sender-controlled Date: header. Add support for two server-controlled
timestamps that are already indexed as FAST i64 fields:

- internal_date (IMAP INTERNALDATE)
- ingest_at (Bichon's archival time)

EmailSearchFilter gains internal_date_since/before and ingest_since/before
range bounds, mirroring the existing `since`/`before` Date: handling.
SortBy gains InternalDate and IngestAt variants (wire values INTERNAL_DATE
and INGEST_AT), mirroring the existing DATE/SIZE sort handling.

The envelope and attachment Tantivy schemas already declare these fields
as INDEXED | STORED | FAST, so no re-index or migration is required.
Attachments carry no IMAP INTERNALDATE, so the attachment search maps the
InternalDate sort to the attachment's own date field as a defined fallback.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

crates/core/src/message/search.rs

@@ -45,6 +45,14 @@ pub struct EmailSearchFilter {
     pub bcc: Option<String>,
     pub since: Option<i64>,
     pub before: Option<i64>,
+    /// Lower bound (inclusive) on the IMAP server INTERNALDATE timestamp.
+    pub internal_date_since: Option<i64>,
+    /// Upper bound (inclusive) on the IMAP server INTERNALDATE timestamp.
+    pub internal_date_before: Option<i64>,
+    /// Lower bound (inclusive) on Bichon's archival (ingest) timestamp.
+    pub ingest_since: Option<i64>,
+    /// Upper bound (inclusive) on Bichon's archival (ingest) timestamp.
+    pub ingest_before: Option<i64>,
     pub account_ids: Option<HashSet<u64>>,
     pub mailbox_ids: Option<HashSet<u64>>,
     pub min_size: Option<u64>,
@@ -64,6 +72,12 @@ pub enum SortBy {
     #[default]
     DATE,
     SIZE,
+    /// Sort by the IMAP server INTERNALDATE timestamp.
+    #[serde(rename = "INTERNAL_DATE")]
+    InternalDate,
+    /// Sort by Bichon's archival (ingest) timestamp.
+    #[serde(rename = "INGEST_AT")]
+    IngestAt,
 }
 
 #[derive(Debug, Clone, Default, Eq, PartialEq, Serialize, Deserialize)]

crates/core/src/store/tantivy/attachment.rs

@@ -38,8 +38,8 @@ use crate::{
     store::tantivy::{
         fatal_commit,
         fields::{
-            F_ATTACHMENT_CATEGORY, F_ATTACHMENT_CONTENT_TYPE, F_ATTACHMENT_EXT, F_DATE, F_SIZE,
-            F_TAGS,
+            F_ATTACHMENT_CATEGORY, F_ATTACHMENT_CONTENT_TYPE, F_ATTACHMENT_EXT, F_DATE,
+            F_INGEST_AT, F_SIZE, F_TAGS,
         },
         model::{extract_senders, AttachmentModel},
         schema::SchemaTools,
@@ -864,6 +864,30 @@ impl IndexManager {
                     .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?;
                 attachment_docs = size_docs.into_iter().map(|(_, addr)| addr).collect();
             }
+            // Attachments carry no IMAP INTERNALDATE; fall back to the
+            // attachment's own date field so the sort remains well defined.
+            SortBy::InternalDate => {
+                let date_docs: Vec<(Option<i64>, DocAddress)> = searcher
+                    .search(
+                        &query,
+                        &TopDocs::with_limit(page_size as usize)
+                            .and_offset(offset as usize)
+                            .order_by_fast_field(F_DATE, order),
+                    )
+                    .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?;
+                attachment_docs = date_docs.into_iter().map(|(_, addr)| addr).collect();
+            }
+            SortBy::IngestAt => {
+                let ingest_at_docs: Vec<(Option<i64>, DocAddress)> = searcher
+                    .search(
+                        &query,
+                        &TopDocs::with_limit(page_size as usize)
+                            .and_offset(offset as usize)
+                            .order_by_fast_field(F_INGEST_AT, order),
+                    )
+                    .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?;
+                attachment_docs = ingest_at_docs.into_iter().map(|(_, addr)| addr).collect();
+            }
         }
 
         let mut result = Vec::new();

crates/core/src/store/tantivy/envelope.rs

@@ -42,8 +42,8 @@ use crate::{
             attachment::ATTACHMENT_MANAGER,
             fatal_commit,
             fields::{
-                F_ACCOUNT_ID, F_DATE, F_FROM, F_ID, F_REGULAR_ATTACHMENT_COUNT, F_SIZE, F_TAGS,
-                F_THREAD_ID, F_UID,
+                F_ACCOUNT_ID, F_DATE, F_FROM, F_ID, F_INGEST_AT, F_INTERNAL_DATE,
+                F_REGULAR_ATTACHMENT_COUNT, F_SIZE, F_TAGS, F_THREAD_ID, F_UID,
             },
             model::{extract_contacts, EnvelopeWithAttachments},
             schema::SchemaTools,
@@ -457,6 +457,40 @@ impl IndexManager {
             subqueries.push((Occur::Must, Box::new(q)));
         }
 
+        let start_bound = if let Some(from) = filter.internal_date_since {
+            Bound::Included(Term::from_field_i64(f.f_internal_date, from))
+        } else {
+            Bound::Unbounded
+        };
+
+        let end_bound = if let Some(to) = filter.internal_date_before {
+            Bound::Included(Term::from_field_i64(f.f_internal_date, to))
+        } else {
+            Bound::Unbounded
+        };
+
+        if start_bound != Bound::Unbounded || end_bound != Bound::Unbounded {
+            let q = RangeQuery::new(start_bound, end_bound);
+            subqueries.push((Occur::Must, Box::new(q)));
+        }
+
+        let start_bound = if let Some(from) = filter.ingest_since {
+            Bound::Included(Term::from_field_i64(f.f_ingest_at, from))
+        } else {
+            Bound::Unbounded
+        };
+
+        let end_bound = if let Some(to) = filter.ingest_before {
+            Bound::Included(Term::from_field_i64(f.f_ingest_at, to))
+        } else {
+            Bound::Unbounded
+        };
+
+        if start_bound != Bound::Unbounded || end_bound != Bound::Unbounded {
+            let q = RangeQuery::new(start_bound, end_bound);
+            subqueries.push((Occur::Must, Box::new(q)));
+        }
+
         if let Some(account_ids) = filter.account_ids {
             let mut should_queries: Vec<(Occur, Box<dyn Query>)> = Vec::new();
             for id in account_ids {
@@ -1141,6 +1175,31 @@ impl IndexManager {
                     .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?;
                 mailbox_docs = size_docs.into_iter().map(|(_, addr)| addr).collect();
             }
+            SortBy::InternalDate => {
+                let internal_date_docs: Vec<(Option<i64>, DocAddress)> = searcher
+                    .search(
+                        &query,
+                        &TopDocs::with_limit(page_size as usize)
+                            .and_offset(offset as usize)
+                            .order_by_fast_field(F_INTERNAL_DATE, order),
+                    )
+                    .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?;
+                mailbox_docs = internal_date_docs
+                    .into_iter()
+                    .map(|(_, addr)| addr)
+                    .collect();
+            }
+            SortBy::IngestAt => {
+                let ingest_at_docs: Vec<(Option<i64>, DocAddress)> = searcher
+                    .search(
+                        &query,
+                        &TopDocs::with_limit(page_size as usize)
+                            .and_offset(offset as usize)
+                            .order_by_fast_field(F_INGEST_AT, order),
+                    )
+                    .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?;
+                mailbox_docs = ingest_at_docs.into_iter().map(|(_, addr)| addr).collect();
+            }
         }
 
         let mut result = Vec::new();