feat(chat): replace 2s chat poll with shared-key subscription router#94
Open
arkanoider wants to merge 3 commits into
Open
feat(chat): replace 2s chat poll with shared-key subscription router#94arkanoider wants to merge 3 commits into
arkanoider wants to merge 3 commits into
Conversation
Introduce a single long-lived chat subscription router (listen_for_chat_messages) that maintains one batched kind:1059 #p filter over all active shared-key pubkeys and routes incoming gift wraps by p tag, mirroring Mostro Mobile's SubscriptionManager. Replaces the 2-second admin_chat_interval poll (spawn_admin_chat_fetch / spawn_user_order_chat_fetch / CHAT_MESSAGES_SEMAPHORE). - Track/untrack via global ChatRouterCmd channel; startup track set (option B) via track_startup_chats (active orders + success, InProgress disputes). - Dynamic hooks: track after order upsert and on dispute take; untrack on terminal status, pre-active drop, and book-republish revert. - Respawn chat router after reconnect and key/fetch reloads. Chat kept live after success; on-disk transcripts preserve untracked terminal history.
Contributor
|
Warning Review limit reached
Next review available in: 54 minutes Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available. How can I continue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews. How do review limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please refer docs for additional details. Review details⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (2)
WalkthroughReplaces periodic polling-based admin/user chat fetch with a persistent shared-key subscription router that batches relay subscriptions, hydrates history once per key, routes live gift-wraps by ChangesShared-key chat subscription router
Estimated code review effort: 4 (Complex) | ~60 minutes Possibly related PRs
Suggested reviewers: Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/util/chat_listener.rs`:
- Around line 209-226: The chat subscription swap in chat_listener::subscribe
flow drops the active subscription too early by calling current_sub.take() and
unsubscribe before client.subscribe succeeds. Keep the existing subscription
alive until the new subscribe call in the match on client.subscribe(filter,
None).await returns Ok, then replace current_sub with the new subscription ID
and unsubscribe the old one only after that succeeds. If subscribe fails, leave
current_sub unchanged so live chat remains active.
- Around line 273-287: The chat listener currently hydrates history before the
new key is added to targets and before resubscribing, which creates a gap where
messages can be missed. In chat_listener.rs, update the flow around the one-shot
history hydration and resubscribe logic so the target is inserted into targets
and the live subscription is rebuilt before fetching/processing history, using
the existing emit_messages, fetch_gift_wraps_for_shared_key, and resubscribe
paths. Keep the history cutoff filtering intact, but ensure the subscription is
active first so any messages published during hydration are covered.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 9322f2ac-ab5b-4d9e-aac1-ecc325216293
📒 Files selected for processing (16)
docs/ADMIN_DISPUTES.mddocs/MESSAGE_FLOW_AND_PROTOCOL.mddocs/STARTUP_AND_CONFIG.mddocs/TUI_INTERFACE.mdsrc/main.rssrc/startup.rssrc/ui/helpers/mod.rssrc/ui/helpers/startup.rssrc/ui/key_handler/async_tasks.rssrc/ui/key_handler/mod.rssrc/util/chat_listener.rssrc/util/dm_utils/mod.rssrc/util/mod.rssrc/util/order_utils/execute_take_dispute.rssrc/util/order_utils/fetch_scheduler.rssrc/util/order_utils/mod.rs
💤 Files with no reviewable changes (1)
- src/util/order_utils/fetch_scheduler.rs
Comment on lines
+209
to
+226
| if let Some(id) = current_sub.take() { | ||
| client.unsubscribe(&id).await; | ||
| } | ||
| if targets.is_empty() { | ||
| return; | ||
| } | ||
| let pubkeys: Vec<PublicKey> = targets.keys().copied().collect(); | ||
| let filter = Filter::new().kind(Kind::GiftWrap).pubkeys(pubkeys).limit(0); | ||
| match client.subscribe(filter, None).await { | ||
| Ok(output) => { | ||
| log::debug!( | ||
| "[chat_live] subscribed to {} shared-key chat(s) subscription_id={}", | ||
| targets.len(), | ||
| output.val | ||
| ); | ||
| *current_sub = Some(output.val); | ||
| } | ||
| Err(e) => log::warn!("[chat_live] failed to subscribe shared-key chats: {e}"), |
Contributor
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Keep the old chat subscription until the replacement succeeds.
Line 209 unsubscribes the current live subscription before client.subscribe succeeds. If the new subscribe fails, targets still contains chats but no live subscription remains until another command or respawn.
Proposed fix
async fn resubscribe(
client: &Client,
targets: &HashMap<PublicKey, ChatTarget>,
current_sub: &mut Option<SubscriptionId>,
) {
- if let Some(id) = current_sub.take() {
- client.unsubscribe(&id).await;
- }
if targets.is_empty() {
+ if let Some(id) = current_sub.take() {
+ client.unsubscribe(&id).await;
+ }
return;
}
let pubkeys: Vec<PublicKey> = targets.keys().copied().collect();
let filter = Filter::new().kind(Kind::GiftWrap).pubkeys(pubkeys).limit(0);
match client.subscribe(filter, None).await {
@@
- *current_sub = Some(output.val);
+ let previous_sub = current_sub.replace(output.val);
+ if let Some(id) = previous_sub {
+ client.unsubscribe(&id).await;
+ }
}
Err(e) => log::warn!("[chat_live] failed to subscribe shared-key chats: {e}"),
}
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| if let Some(id) = current_sub.take() { | |
| client.unsubscribe(&id).await; | |
| } | |
| if targets.is_empty() { | |
| return; | |
| } | |
| let pubkeys: Vec<PublicKey> = targets.keys().copied().collect(); | |
| let filter = Filter::new().kind(Kind::GiftWrap).pubkeys(pubkeys).limit(0); | |
| match client.subscribe(filter, None).await { | |
| Ok(output) => { | |
| log::debug!( | |
| "[chat_live] subscribed to {} shared-key chat(s) subscription_id={}", | |
| targets.len(), | |
| output.val | |
| ); | |
| *current_sub = Some(output.val); | |
| } | |
| Err(e) => log::warn!("[chat_live] failed to subscribe shared-key chats: {e}"), | |
| if targets.is_empty() { | |
| if let Some(id) = current_sub.take() { | |
| client.unsubscribe(&id).await; | |
| } | |
| return; | |
| } | |
| let pubkeys: Vec<PublicKey> = targets.keys().copied().collect(); | |
| let filter = Filter::new().kind(Kind::GiftWrap).pubkeys(pubkeys).limit(0); | |
| match client.subscribe(filter, None).await { | |
| Ok(output) => { | |
| log::debug!( | |
| "[chat_live] subscribed to {} shared-key chat(s) subscription_id={}", | |
| targets.len(), | |
| output.val | |
| ); | |
| let previous_sub = current_sub.replace(output.val); | |
| if let Some(id) = previous_sub { | |
| client.unsubscribe(&id).await; | |
| } | |
| } | |
| Err(e) => log::warn!("[chat_live] failed to subscribe shared-key chats: {e}"), |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/util/chat_listener.rs` around lines 209 - 226, The chat subscription swap
in chat_listener::subscribe flow drops the active subscription too early by
calling current_sub.take() and unsubscribe before client.subscribe succeeds.
Keep the existing subscription alive until the new subscribe call in the match
on client.subscribe(filter, None).await returns Ok, then replace current_sub
with the new subscription ID and unsubscribe the old one only after that
succeeds. If subscribe fails, leave current_sub unchanged so live chat remains
active.
Comment on lines
+273
to
+287
| // One-shot history hydration (relay subscriptions alone don't replay history). | ||
| match fetch_gift_wraps_for_shared_key(&client, &shared_keys).await { | ||
| Ok(messages) => { | ||
| let cutoff = since.unwrap_or(0); | ||
| let history: Vec<(String, i64, PublicKey)> = messages | ||
| .into_iter() | ||
| .filter(|(_, ts, _)| *ts >= cutoff) | ||
| .collect(); | ||
| emit_messages(&target, history, &admin_chat_updates_tx, &user_order_chat_updates_tx); | ||
| } | ||
| Err(e) => log::warn!("[chat_live] history fetch failed for {key_id:?}: {e}"), | ||
| } | ||
|
|
||
| targets.insert(target_pubkey, target); | ||
| resubscribe(&client, &targets, &mut current_sub).await; |
Contributor
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Subscribe the key before history hydration.
The router fetches history before adding the key to targets and rebuilding the live subscription. Any message published after the fetch snapshot but before Line 287 subscribes can be missed.
Proposed fix
let target = ChatTarget {
key_id: key_id.clone(),
shared_keys: shared_keys.clone(),
local_trade_pubkey,
};
+ targets.insert(target_pubkey, target);
+ resubscribe(&client, &targets, &mut current_sub).await;
+
// One-shot history hydration (relay subscriptions alone don't replay history).
match fetch_gift_wraps_for_shared_key(&client, &shared_keys).await {
Ok(messages) => {
let cutoff = since.unwrap_or(0);
let history: Vec<(String, i64, PublicKey)> = messages
.into_iter()
.filter(|(_, ts, _)| *ts >= cutoff)
.collect();
- emit_messages(&target, history, &admin_chat_updates_tx, &user_order_chat_updates_tx);
+ if let Some(target) = targets.get(&target_pubkey) {
+ emit_messages(target, history, &admin_chat_updates_tx, &user_order_chat_updates_tx);
+ }
}
Err(e) => log::warn!("[chat_live] history fetch failed for {key_id:?}: {e}"),
}
-
- targets.insert(target_pubkey, target);
- resubscribe(&client, &targets, &mut current_sub).await;📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| // One-shot history hydration (relay subscriptions alone don't replay history). | |
| match fetch_gift_wraps_for_shared_key(&client, &shared_keys).await { | |
| Ok(messages) => { | |
| let cutoff = since.unwrap_or(0); | |
| let history: Vec<(String, i64, PublicKey)> = messages | |
| .into_iter() | |
| .filter(|(_, ts, _)| *ts >= cutoff) | |
| .collect(); | |
| emit_messages(&target, history, &admin_chat_updates_tx, &user_order_chat_updates_tx); | |
| } | |
| Err(e) => log::warn!("[chat_live] history fetch failed for {key_id:?}: {e}"), | |
| } | |
| targets.insert(target_pubkey, target); | |
| resubscribe(&client, &targets, &mut current_sub).await; | |
| let target = ChatTarget { | |
| key_id: key_id.clone(), | |
| shared_keys: shared_keys.clone(), | |
| local_trade_pubkey, | |
| }; | |
| targets.insert(target_pubkey, target); | |
| resubscribe(&client, &targets, &mut current_sub).await; | |
| // One-shot history hydration (relay subscriptions alone don't replay history). | |
| match fetch_gift_wraps_for_shared_key(&client, &shared_keys).await { | |
| Ok(messages) => { | |
| let cutoff = since.unwrap_or(0); | |
| let history: Vec<(String, i64, PublicKey)> = messages | |
| .into_iter() | |
| .filter(|(_, ts, _)| *ts >= cutoff) | |
| .collect(); | |
| if let Some(target) = targets.get(&target_pubkey) { | |
| emit_messages(target, history, &admin_chat_updates_tx, &user_order_chat_updates_tx); | |
| } | |
| } | |
| Err(e) => log::warn!("[chat_live] history fetch failed for {key_id:?}: {e}"), | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/util/chat_listener.rs` around lines 273 - 287, The chat listener
currently hydrates history before the new key is added to targets and before
resubscribing, which creates a gap where messages can be missed. In
chat_listener.rs, update the flow around the one-shot history hydration and
resubscribe logic so the target is inserted into targets and the live
subscription is rebuilt before fetching/processing history, using the existing
emit_messages, fetch_gift_wraps_for_shared_key, and resubscribe paths. Keep the
history cutoff filtering intact, but ensure the subscription is active first so
any messages published during hydration are covered.
Co-authored-by: Cursor <cursoragent@cursor.com>
Co-authored-by: Cursor <cursoragent@cursor.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Introduce a single long-lived chat subscription router (listen_for_chat_messages) that maintains one batched kind:1059 #p filter over all active shared-key pubkeys and routes incoming gift wraps by p tag, mirroring Mostro Mobile's SubscriptionManager. Replaces the 2-second admin_chat_interval poll (spawn_admin_chat_fetch / spawn_user_order_chat_fetch / CHAT_MESSAGES_SEMAPHORE).
Track/untrack via global ChatRouterCmd channel; startup track set (option B) via track_startup_chats (active orders + success, InProgress disputes).
Dynamic hooks: track after order upsert and on dispute take; untrack on terminal status, pre-active drop, and book-republish revert.
Respawn chat router after reconnect and key/fetch reloads. Chat kept live after success; on-disk transcripts preserve untracked terminal history.
Summary by CodeRabbit