From baa16a9687be71e82d645fd2c846bd1aed8d24c6 Mon Sep 17 00:00:00 2001 From: hitalin Date: Fri, 7 Aug 2026 15:31:53 +0900 Subject: [PATCH] =?UTF-8?q?feat(streaming):=20polling=20=E3=83=A2=E3=83=BC?= =?UTF-8?q?=E3=83=89=E3=81=A7=E3=83=81=E3=83=A3=E3=83=83=E3=83=88=E3=83=A1?= =?UTF-8?q?=E3=83=83=E3=82=BB=E3=83=BC=E3=82=B8=E3=82=92=E9=85=8D=E4=BF=A1?= =?UTF-8?q?=E3=81=99=E3=82=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ポーリングループの対象にチャット購読 (chatUser / chatRoom) が含まれず、 polling モードではチャットの新着が届かなかった (notedeck#1008)。チャット 購読を検出して user-timeline / room-timeline を定期取得し、WS と同じ ChatMessage イベントで配る。chat は subscription_id で 1:1 配送される ため購読 ID をそのまま載せる。DB キャッシュも WS 受信側と同じ経路。 初回サイクルはカーソル確立のみで emit しない (#1003 の通知 polling と 同じ方針)。既読/リアクション/削除の polling 追随はスコープ外。 Co-Authored-By: Claude Opus 4.8 --- src/streaming.rs | 222 +++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 222 insertions(+) diff --git a/src/streaming.rs b/src/streaming.rs index 90d1d02..a3b6126 100644 --- a/src/streaming.rs +++ b/src/streaming.rs @@ -1778,6 +1778,14 @@ struct MainPollState { mention_since_id: Option, } +/// チャット購読ごとの polling カーソル (notedeck#1008)。prime の意味は +/// MainPollState と同じ。 +#[derive(Default)] +struct ChatPollState { + primed: bool, + since_id: Option, +} + /// Top-level polling task. Periodically fetches notes for all active subscriptions. #[allow(clippy::too_many_arguments)] async fn polling_loop( @@ -1795,6 +1803,7 @@ async fn polling_loop( ) { let mut sub_states: HashMap = HashMap::new(); let mut main_state = MainPollState::default(); + let mut chat_states: HashMap = HashMap::new(); // Cached reaction counts for captured notes (for diff detection). let mut note_reaction_cache: HashMap> = HashMap::new(); let mut consecutive_failures: u64 = 0; @@ -2018,6 +2027,118 @@ async fn polling_loop( } } + // チャット購読 (chatUser / chatRoom) の polling (notedeck#1008)。 + // WS と同じ ChatMessage イベントで emit する。chat は subscription_id + // で 1:1 配送されるため購読 ID をそのまま載せる。メッセージ挿入のみ + // 対応 (既読/リアクション/削除の polling 追随はスコープ外)。 + let chat_snapshot: Vec<(String, SubscriptionTarget)> = { + let subs = subscriptions.read().await; + subs.iter() + .filter(|(_, info)| info.account_id == account_id && info.active) + .filter_map(|(id, info)| match &info.target { + t @ (SubscriptionTarget::ChatUser { .. } + | SubscriptionTarget::ChatRoom { .. }) => Some((id.clone(), t.clone())), + _ => None, + }) + .collect() + }; + // DB キャッシュに own user_id が要る (WS 受信側と同じ)。アカウントは + // 不変なのでサイクルごとに 1 回だけ引く + let own_user_id: Option = if chat_snapshot.is_empty() { + None + } else { + db.get_account(&account_id) + .ok() + .flatten() + .map(|a| a.user_id.clone()) + }; + for (sub_id, target) in &chat_snapshot { + let state = chat_states.entry(sub_id.clone()).or_default(); + let (limit, since) = if state.primed { + (30, state.since_id.clone()) + } else { + // 初回はカーソル確立のみで emit しない + (1, None) + }; + let result = match target { + SubscriptionTarget::ChatUser { other_id } => { + api_client + .get_chat_user_messages( + &host, + &token, + other_id, + limit, + since.as_deref(), + None, + ) + .await + } + SubscriptionTarget::ChatRoom { room_id } => { + api_client + .get_chat_room_messages( + &host, + &token, + room_id, + limit, + since.as_deref(), + None, + ) + .await + } + _ => unreachable!("chat_snapshot filters to chat targets"), + }; + match result { + Ok(messages) => { + if let Some(newest) = messages.first() { + state.since_id = Some(newest.id.clone()); + } + if state.primed { + for msg in messages.into_iter().rev() { + if let Some(user_id) = &own_user_id { + let db = db.clone(); + let msg_for_cache = msg.clone(); + let account_id_owned = account_id.clone(); + let host_owned = host.clone(); + let user_id = user_id.clone(); + tokio::task::spawn_blocking(move || { + if let Err(e) = db.cache_chat_message( + &msg_for_cache, + &account_id_owned, + &user_id, + &host_owned, + ) { + tracing::warn!(error = %e, "failed to cache polled chat message"); + } + }); + } + let payload = StreamChatMessageEvent { + account_id: account_id.clone(), + subscription_id: sub_id.clone(), + message: msg, + }; + emit_both( + emitter.as_ref(), + &event_bus, + StreamEvent::ChatMessage(Box::new(payload)), + ); + } + } else { + state.primed = true; + } + consecutive_failures = 0; + } + Err(e) => { + tracing::warn!( + account_id = %account_id, + subscription_id = %sub_id, + error = %e, + "chat polling failed" + ); + poll_failed = true; + } + } + } + // Note capture: poll every 2nd cycle (2x interval) poll_count += 1; if poll_count.is_multiple_of(2) { @@ -2775,6 +2896,107 @@ mod tests { manager.disconnect("acc-1").await; } + #[tokio::test] + async fn polling_delivers_chat_messages() { + // ポーリングモードではチャット購読 (chatUser / chatRoom) の取得経路が + // 存在しなかった (notedeck#1008)。polling_loop がチャット購読を検出して + // user-timeline / room-timeline を定期取得し、WS と同じ ChatMessage + // イベントで配ることを検証する。chat は subscription_id で 1:1 配送 + // されるため、購読 ID が正しく載ることも見る。 + use serde_json::json; + use wiremock::matchers::{body_partial_json, method, path}; + use wiremock::{Mock, MockServer, ResponseTemplate}; + + fn chat_message_json(id: &str) -> serde_json::Value { + json!({ + "id": id, + "createdAt": "2025-01-01T00:00:00.000Z", + "fromUserId": "u-other", + "fromUser": {"id": "u-other", "name": null, "username": "they", + "host": null, "avatarUrl": null}, + "toUserId": "u-self", + "toUser": {"id": "u-self", "name": null, "username": "me", + "host": null, "avatarUrl": null}, + "toRoomId": null, + "toRoom": null, + "text": "hi", + "fileId": null, + "file": null, + "isRead": false, + "reactions": [] + }) + } + + let server = MockServer::start().await; + // 初回 prime (limit 1): 既存分。emit されないこと + Mock::given(method("POST")) + .and(path("/api/chat/messages/user-timeline")) + .and(body_partial_json(json!({"limit": 1}))) + .respond_with( + ResponseTemplate::new(200).set_body_json(json!([chat_message_json("msg-old")])), + ) + .mount(&server) + .await; + // 2 回目以降: カーソル付きで新着 + Mock::given(method("POST")) + .and(path("/api/chat/messages/user-timeline")) + .and(body_partial_json(json!({"sinceId": "msg-old"}))) + .respond_with( + ResponseTemplate::new(200).set_body_json(json!([chat_message_json("msg-new")])), + ) + .mount(&server) + .await; + Mock::given(method("POST")) + .and(path("/api/chat/messages/user-timeline")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!([]))) + .mount(&server) + .await; + + let dir = tempfile::tempdir().unwrap(); + let db = Arc::new(crate::db::Database::open(&dir.path().join("test.db")).unwrap()); + let (tx, mut rx) = mpsc::unbounded_channel(); + let mut manager = + StreamingManager::new(Arc::new(ChannelEmitter(tx)), Arc::new(EventBus::new()), db); + manager.api_client = Arc::new(MisskeyClient::with_base_url(&server.uri())); + + manager + .set_mode("acc-1", "127.0.0.1:1", "token", "polling", Some(100)) + .await + .unwrap(); + let sub_id = manager + .subscribe_chat_user("acc-1", "u-other") + .await + .unwrap(); + + let mut message_ids = Vec::new(); + let mut sub_ids = Vec::new(); + let deadline = tokio::time::Instant::now() + Duration::from_secs(5); + while message_ids.is_empty() && tokio::time::Instant::now() < deadline { + match tokio::time::timeout(Duration::from_millis(200), rx.recv()).await { + Ok(Some(StreamEvent::ChatMessage(e))) => { + assert_eq!(e.account_id, "acc-1"); + message_ids.push(e.message.id.clone()); + sub_ids.push(e.subscription_id.clone()); + } + Ok(Some(_)) => {} + _ => {} + } + } + + assert_eq!( + message_ids, + vec!["msg-new"], + "チャットは新着のみ配信されること" + ); + assert_eq!( + sub_ids, + vec![sub_id], + "1:1 配送に使う subscription_id が購読 ID と一致すること" + ); + + manager.disconnect("acc-1").await; + } + #[tokio::test] async fn unsubscribe_drops_the_subscription_in_both_modes() { let (_dir, manager, _rx) = manager_with_dead_connection(&["acc-1"]).await;