From 21e91765d010bca40d8958adb82cd264415a4f65 Mon Sep 17 00:00:00 2001 From: anonsaber Date: Mon, 12 Oct 2026 22:08:51 +0800 Subject: [PATCH] feat(reconcile): filter initial baseline to 7-day unread emails MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit New mailbox onboarding no longer floods Telegram with the entire mailbox history. The initial baseline walk (ReconcileCursor::Walk, any position) now passes an EmailQueryFilter to list_emails_page that bounds Email/query to messages received within the last 7 days and not yet seen ($seen keyword, RFC 8621 §4.1.2). - EmailQueryFilter { received_after: Option, unread_only: bool } in src/domain/jmap.rs (backend-agnostic domain type). - StalwartJmap::fetch_email_page builds a compound Filter::and from the filter conditions (after + not_keyword "$seen"). - walk_baseline computes the 7-day cutoff once and passes it to every page fetch, so resumed walks stay filtered. - Replay(state) path returns early and never calls list_emails_page, so /changes incremental is unaffected. - const INITIAL_BASELINE_WINDOW_DAYS: u64 = 7. - New test first_reconcile_filters_the_initial_baseline_to_recent_unread pins the filter on every Walk page and the cursor transition to Replay. - 3 doc anchors re-anchored (worker.rs:227→241, client.rs:485→513 ×2). GATE-P0: 104 passed / 0 failed / 4 ignored. --- docs/design.zh-CN.md | 2 +- docs/opengaps.md | 2 +- docs/opengaps.zh-CN.md | 2 +- src/domain/jmap.rs | 28 +++++- src/domain/jmap/client.rs | 36 +++++++- src/worker.rs | 183 +++++++++++++++++++++++++++++++++++++- 6 files changed, 242 insertions(+), 11 deletions(-) diff --git a/docs/design.zh-CN.md b/docs/design.zh-CN.md index b12e82d..a137905 100644 --- a/docs/design.zh-CN.md +++ b/docs/design.zh-CN.md @@ -279,7 +279,7 @@ TG /read 3 → handler 取会话里的 folder+page 游标 当前**没有 FSM**。命令路由由 `src/worker.rs` 的 `parse_intent` 解析为 `Intent`；AI 授权态是 Redis 里的一个布尔加过期时间（键与 TTL 见 `docs/reference.md` 的 AI 授权态一节），不是多步确认态。 -仍成立的不变量：AI 摘要必须先有用户显式授权，**没有授权态不得调用 LLM**。授权一旦生效，摘要会拉取邮件正文全文送 LLM（`read_email` 请求 `TextBody` + `BodyValues` 并 `fetch_text_body_values(true)`，见 `src/domain/jmap/client.rs`），但正文**不回传给 Telegram**——出站只发摘要，或失败时回退为前 300 字符（`worker.rs:227` 注释：body text never reaches Telegram）。 +仍成立的不变量：AI 摘要必须先有用户显式授权，**没有授权态不得调用 LLM**。授权一旦生效，摘要会拉取邮件正文全文送 LLM（`read_email` 请求 `TextBody` + `BodyValues` 并 `fetch_text_body_values(true)`，见 `src/domain/jmap/client.rs`），但正文**不回传给 Telegram**——出站只发摘要，或失败时回退为前 300 字符（`worker.rs:241` 注释：body text never reaches Telegram）。 曾设计过的 5 态 FSM（`Idle` / `AwaitClarify` / `AwaitConfirm` / `Analyzing` / `AwaitFallback`）连同状态转移表与渠道中立说明，见 `docs/retired.md`。 diff --git a/docs/opengaps.md b/docs/opengaps.md index 0ec48bc..2b0caa1 100644 --- a/docs/opengaps.md +++ b/docs/opengaps.md @@ -107,7 +107,7 @@ Single accounts are fixed by requirement (`REQ-SINGLE-ACCOUNT`) and never enter - `state:jmap:since` is a **single global cursor**, no account dimension (read `src/state.rs:801`, write `src/state.rs:809`) - both accounts will overwrite each other's cursors, which is the first place to move. - `lock:reconcile` is a **single global solo lock** (Redis implementation `src/state.rs:1139`) - the reconciliation of two accounts will be serialized by the same lock. - `dedup:jmap:{account_id}:{email_id}` **Already has the account segment** (`docs/reference.md` §1), the key shape does not need to be changed. -- The `ACCOUNT_ID` environment variable already exists and has been verified (leave it blank to take the session main account, and an error will be reported if the out-of-bounds value is `src/domain/jmap/client.rs:485`). The business configuration layer can also override `account_id` (`unwrap_or_else` of `src/config.rs:465` falls back). The configuration interface is ready, all that is missing is the sub-dimension status within the instance. +- The `ACCOUNT_ID` environment variable already exists and has been verified (leave it blank to take the session main account, and an error will be reported if the out-of-bounds value is `src/domain/jmap/client.rs:513`). The business configuration layer can also override `account_id` (`unwrap_or_else` of `src/config.rs:465` falls back). The configuration interface is ready, all that is missing is the sub-dimension status within the instance. This item is not skipped, but explicitly rejected: the deployment unit saved by single instance and multiple accounts cannot be exchanged for the routing and cursor complexity it introduces. diff --git a/docs/opengaps.zh-CN.md b/docs/opengaps.zh-CN.md index ffd393d..749174c 100644 --- a/docs/opengaps.zh-CN.md +++ b/docs/opengaps.zh-CN.md @@ -107,7 +107,7 @@ TTL 那条真机绿灯已经用你给的 Upstash URL 跑过了（见 §1）， - `state:jmap:since` 是**单一全局游标**，无账户维度（读 `src/state.rs:801`、写 `src/state.rs:809`）——两个账户会互相覆盖游标，这是第一个要动的地方。 - `lock:reconcile` 是**单一全局单飞锁**（Redis 实现 `src/state.rs:1139`）——两个账户的对账会被同一把锁串行化。 - `dedup:jmap:{account_id}:{email_id}` **已经带账户段**（`docs/reference.md` §1），键形无需改。 -- `ACCOUNT_ID` 环境变量已存在且经校验（留空取 session 主账户，越界值在 `src/domain/jmap/client.rs:485` 报错），业务配置层也能覆盖 `account_id`（`src/config.rs:465` 的 `unwrap_or_else` 回落）。配置面已就绪，缺的只是实例内的分维度状态。 +- `ACCOUNT_ID` 环境变量已存在且经校验（留空取 session 主账户，越界值在 `src/domain/jmap/client.rs:513` 报错），业务配置层也能覆盖 `account_id`（`src/config.rs:465` 的 `unwrap_or_else` 回落）。配置面已就绪，缺的只是实例内的分维度状态。 这条不是被跳过，是被明确拒绝：单实例多账户省下的一个部署单元，换不来它引入的路由与游标复杂度。 diff --git a/src/domain/jmap.rs b/src/domain/jmap.rs index dccc868..cd3a76f 100644 --- a/src/domain/jmap.rs +++ b/src/domain/jmap.rs @@ -97,6 +97,21 @@ pub struct SearchResult { pub snippets: Vec, } +/// Optional server-side constraints for `Email/query`. +/// +/// The initial baseline walk uses this so a freshly-connected mailbox does not +/// enqueue its entire history: it narrows to recent unread mail. Incremental +/// `/changes` walks never call `list_emails_page`, so they are unaffected. +/// +/// `received_after` is epoch **seconds** (matching the JMAP `after` filter, which +/// the server compares against `receivedAt`). `unread_only` maps to +/// `notKeyword "$seen"` (RFC 8621 §4.1.2: `$seen` is the read keyword). +#[derive(Debug, Clone, Default)] +pub struct EmailQueryFilter { + pub received_after: Option, + pub unread_only: bool, +} + #[async_trait] pub trait JmapBackend: Send + Sync { async fn create_push_subscription(&self, callback_url: &str) -> Result; @@ -144,13 +159,23 @@ pub trait JmapBackend: Send + Sync { /// Fetch one bounded page without requiring the caller to materialize the /// entire mailbox. Implementations with a native JMAP position query should /// override this; the fallback preserves test backends' existing contract. + /// + /// `filter` narrows the server-side query (see [`EmailQueryFilter`]). It + /// carries the baseline's recent-unread window, so `position` paginates + /// within the *filtered* set: it must be held constant across every page of + /// one walk, or a later page would paginate the unfiltered mailbox. async fn list_emails_page( &self, account_id: &str, folder_id: Option<&str>, position: usize, limit: usize, + filter: Option, ) -> Result, JmapError> { + // The fallback has no server-side query to push the filter into; it is + // honored only where `Email/query` exists (`JmapClientBackend`), so a + // test backend that ignores it keeps its existing contract. + let _ = filter; let all = self .list_emails(account_id, folder_id, position.saturating_add(limit)) .await?; @@ -277,12 +302,13 @@ impl JmapService { folder_id: Option<&str>, position: usize, limit: usize, + filter: Option, ) -> Result, JmapError> { if limit == 0 { return Err(JmapError::InvalidRequest("limit")); } self.backend - .list_emails_page(&self.account_id, folder_id, position, limit) + .list_emails_page(&self.account_id, folder_id, position, limit, filter) .await } diff --git a/src/domain/jmap/client.rs b/src/domain/jmap/client.rs index e777d89..43dda44 100644 --- a/src/domain/jmap/client.rs +++ b/src/domain/jmap/client.rs @@ -2,13 +2,17 @@ //! Concrete read-only adapter (MOD-JMAP-CLIENT, GATE-G1-JMAP-READONLY). use super::{ - EmailChanges, EmailContent, EmailMetadata, Folder, JmapBackend, JmapError, SearchResult, - SearchSnippet, + EmailChanges, EmailContent, EmailMetadata, EmailQueryFilter, Folder, JmapBackend, JmapError, + SearchResult, SearchSnippet, }; use crate::state::{runtime_provider, OutboundConfig, RuntimeConfigProvider}; use async_trait::async_trait; use jmap_client::{email, mailbox, DataType, Get}; +/// RFC 8621 §4.1.2: the read keyword. `notKeyword "$seen"` narrows `Email/query` +/// to unread mail for the initial-baseline filter. +const KEYWORD_SEEN: &str = "$seen"; + pub struct JmapClientBackend { client: jmap_client::client::Client, account_id: String, @@ -270,8 +274,9 @@ impl JmapBackend for JmapClientBackend { folder_id: Option<&str>, position: usize, limit: usize, + filter: Option, ) -> Result, JmapError> { - self.fetch_email_page(account_id, folder_id, position, limit) + self.fetch_email_page(account_id, folder_id, position, limit, filter) .await } @@ -369,14 +374,37 @@ impl JmapClientBackend { folder_id: Option<&str>, position: usize, limit: usize, + filter: Option, ) -> Result, JmapError> { if account_id != self.account_id || limit == 0 { return Err(JmapError::InvalidRequest("email query")); } let mut request = self.client.build(); let query = request.query_email(); + // Compose the folder condition (if any) with the baseline filter into one + // `Email/query` filter. A single condition is sent as-is; two or more are + // AND'd. Keeping `position` paginating the *filtered* set is what stops a + // multi-page baseline from flooding page 2+ with the unfiltered mailbox. + let mut conditions: Vec = Vec::new(); if let Some(folder_id) = folder_id { - query.filter(email::query::Filter::in_mailbox(folder_id)); + conditions.push(email::query::Filter::in_mailbox(folder_id)); + } + if let Some(filter) = filter { + if let Some(after) = filter.received_after { + conditions.push(email::query::Filter::after(after)); + } + if filter.unread_only { + conditions.push(email::query::Filter::not_keyword(KEYWORD_SEEN.to_string())); + } + } + match conditions.len() { + 0 => {} + 1 => { + query.filter(conditions.swap_remove(0)); + } + _ => { + query.filter(jmap_client::core::query::Filter::and(conditions)); + } } query.position(position as i32).limit(limit.min(100)); let mut response = tokio::time::timeout(self.timeout(), request.send_query_email()) diff --git a/src/worker.rs b/src/worker.rs index 75c2d4a..7b4494d 100644 --- a/src/worker.rs +++ b/src/worker.rs @@ -3,7 +3,7 @@ use crate::ai::LlmClient; use crate::channel::telegram::TelegramClient; use crate::config::BusinessConfig; -use crate::domain::jmap::{JmapBackend, JmapService, SearchResult}; +use crate::domain::jmap::{EmailQueryFilter, JmapBackend, JmapService, SearchResult}; use crate::domain::Notification; use crate::state::ttl; use crate::state::ReliableState; @@ -12,7 +12,7 @@ use std::{ future::Future, pin::Pin, sync::{Arc, RwLock}, - time::Duration, + time::{Duration, SystemTime, UNIX_EPOCH}, }; #[async_trait] @@ -180,6 +180,11 @@ const CHANGE_WINDOW_CAP: usize = 4_096; const RECONCILE_MAX_PAGES: usize = 100; /// Wall-clock ceiling for the whole invocation, walk phase included. const RECONCILE_BUDGET: Duration = Duration::from_secs(20); +/// The initial-baseline window: the walk only enqueues mail received within +/// the last N days and still unread, so a freshly-connected mailbox does not +/// flood notifications with its entire archive. `/changes` replay is +/// unaffected (it never calls `list_emails_page`). +const INITIAL_BASELINE_WINDOW_DAYS: u64 = 7; /// `reconcile:state` is bimodal, and the two modes need different handling. /// @@ -401,6 +406,18 @@ impl MetadataWorker { let mut pages = 0_usize; let mut emails = 0_usize; + // The baseline walk is the only caller of `list_emails_page`, and it must + // not enqueue a freshly-connected mailbox's entire history. Narrow every + // page to recent unread mail so the position walk paginates the *filtered* + // set consistently — filtering only page 0 would flood page 2+ from the + // unfiltered set at position 100. `/changes` replay never reaches here. + let baseline_filter = Some(EmailQueryFilter { + received_after: SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|now| now.as_secs() as i64 - (INITIAL_BASELINE_WINDOW_DAYS as i64) * 86_400) + .ok(), + unread_only: true, + }); loop { if pages >= BASELINE_MAX_PAGES || emails >= BASELINE_MAX_EMAILS @@ -417,7 +434,7 @@ impl MetadataWorker { } let page = self .jmap - .list_emails_page(None, position, BASELINE_PAGE_SIZE) + .list_emails_page(None, position, BASELINE_PAGE_SIZE, baseline_filter.clone()) .await .map_err(|_| "jmap_list_failed")?; if page.is_empty() { @@ -1347,6 +1364,7 @@ mod reconcile_rebaseline_tests { _folder_id: Option<&str>, _position: usize, _limit: usize, + _filter: Option, ) -> Result, JmapError> { Ok(vec![]) } @@ -1462,6 +1480,7 @@ mod reconcile_rebaseline_tests { _folder_id: Option<&str>, _position: usize, _limit: usize, + _filter: Option, ) -> Result, JmapError> { Ok(vec![EmailMetadata { id: "m1".to_string(), @@ -1527,4 +1546,162 @@ mod reconcile_rebaseline_tests { "a truncated walk must not replay /changes from a half-listed state" ); } + + /// Records every `EmailQueryFilter` the baseline walk hands to + /// `list_emails_page`. Page 0 returns a short filtered-looking set so the + /// walk advances; every later page returns empty so the walk exhausts after + /// one page. The recorded filters let the test pin acceptance #1 (only + /// recent unread) and the whole-Walk-phase deviation (every page carries + /// the filter, not just position 0). + struct FilteredBaselineBackend { + filters: Arc>>>, + } + + #[async_trait::async_trait] + impl JmapBackend for FilteredBaselineBackend { + async fn create_push_subscription(&self, _callback_url: &str) -> Result { + unreachable!("baseline walk never creates push subscriptions") + } + async fn destroy_push_subscription(&self, _subscription_id: &str) -> Result<(), JmapError> { + unreachable!("baseline walk never destroys push subscriptions") + } + async fn verify_push_subscription( + &self, + _subscription_id: &str, + _verification_code: &str, + ) -> Result<(), JmapError> { + unreachable!("baseline walk never verifies push subscriptions") + } + async fn session_state(&self) -> Result { + Ok("state-fresh".to_string()) + } + async fn email_changes( + &self, + _account_id: &str, + _since_state: &str, + _max_changes: usize, + ) -> Result { + Ok(EmailChanges { + new_state: "state-listed".to_string(), + created: vec![], + updated: vec![], + has_more_changes: false, + }) + } + async fn list_folders(&self, _account_id: &str) -> Result, JmapError> { + unreachable!("baseline walk never lists folders") + } + async fn list_emails( + &self, + _account_id: &str, + _folder_id: Option<&str>, + _limit: usize, + ) -> Result, JmapError> { + unreachable!("baseline walk uses list_emails_page, not list_emails") + } + async fn list_emails_page( + &self, + _account_id: &str, + _folder_id: Option<&str>, + position: usize, + _limit: usize, + filter: Option, + ) -> Result, JmapError> { + self.filters + .lock() + .expect("filter recorder mutex") + .push(filter); + if position == 0 { + Ok(vec![ + EmailMetadata { + id: "m-recent-unread-1".to_string(), + subject: Some("recent".to_string()), + sender: Some("a@example.org".to_string()), + received_at: Some(1), + preview: None, + size: None, + has_attachment: false, + }, + EmailMetadata { + id: "m-recent-unread-2".to_string(), + subject: Some("recent".to_string()), + sender: Some("b@example.org".to_string()), + received_at: Some(2), + preview: None, + size: None, + has_attachment: false, + }, + ]) + } else { + Ok(vec![]) + } + } + async fn read_email( + &self, + _account_id: &str, + _email_id: &str, + ) -> Result, JmapError> { + unreachable!("baseline walk never reads individual emails") + } + async fn search_emails( + &self, + _account_id: &str, + _query: &str, + _limit: usize, + ) -> Result { + unreachable!("baseline walk never searches") + } + } + + #[tokio::test] + async fn first_reconcile_filters_the_initial_baseline_to_recent_unread() { + let filters: Arc>>> = + Arc::new(std::sync::Mutex::new(Vec::new())); + let backend = FilteredBaselineBackend { + filters: filters.clone(), + }; + let jmap = JmapService::new(backend, "account-x").expect("account id"); + let telegram = TelegramClient::with_endpoint( + SecretString::from("token".to_string()), + runtime_provider(OutboundConfig::default()), + "http://127.0.0.1:1/".to_string(), + ); + let state: Arc = Arc::new(MemoryState::enabled_for_tests()); + let worker = MetadataWorker::new(jmap, telegram, 42, state, None, 0); + + let cursor = worker + .reconcile(None, 100) + .await + .expect("a filtered baseline walk must complete, not error"); + + assert!( + matches!( + ReconcileCursor::from_persisted(&cursor), + ReconcileCursor::Replay(_) + ), + "the walk must exhaust the filtered set and move to the /changes phase" + ); + + let recorded = filters.lock().expect("filter recorder mutex"); + assert!( + !recorded.is_empty(), + "the baseline walk must call list_emails_page at least once" + ); + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .expect("post-epoch clock") + .as_secs() as i64; + let expected_after = now - (INITIAL_BASELINE_WINDOW_DAYS as i64) * 86_400; + for f in recorded.iter() { + let f = f.as_ref().expect("every baseline page must carry a filter"); + assert!(f.unread_only, "baseline filter must request unread only"); + let after = f + .received_after + .expect("baseline filter must bound received_at to the 7-day window"); + assert!( + after >= expected_after - 60 && after <= expected_after + 60, + "received_after must be ~now - 7d (got {after}, expected ~{expected_after})" + ); + } + } }