From 2ee2077ead4e4d752538199dd58357c8385d792a Mon Sep 17 00:00:00 2001 From: arkanoider Date: Tue, 7 Jul 2026 16:23:42 +0200 Subject: [PATCH 1/3] feat(chat): replace 2s chat poll with shared-key subscription router 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. --- docs/ADMIN_DISPUTES.md | 2 +- docs/MESSAGE_FLOW_AND_PROTOCOL.md | 20 +- docs/STARTUP_AND_CONFIG.md | 24 +- docs/TUI_INTERFACE.md | 2 +- src/main.rs | 77 ++-- src/startup.rs | 38 +- src/ui/helpers/mod.rs | 1 + src/ui/helpers/startup.rs | 78 +++- src/ui/key_handler/async_tasks.rs | 52 ++- src/ui/key_handler/mod.rs | 5 +- src/util/chat_listener.rs | 366 +++++++++++++++++++ src/util/dm_utils/mod.rs | 10 + src/util/mod.rs | 5 + src/util/order_utils/execute_take_dispute.rs | 14 + src/util/order_utils/fetch_scheduler.rs | 56 --- src/util/order_utils/mod.rs | 3 +- 16 files changed, 634 insertions(+), 119 deletions(-) create mode 100644 src/util/chat_listener.rs diff --git a/docs/ADMIN_DISPUTES.md b/docs/ADMIN_DISPUTES.md index 7c69a5e..5b79c8f 100644 --- a/docs/ADMIN_DISPUTES.md +++ b/docs/ADMIN_DISPUTES.md @@ -681,7 +681,7 @@ Buyers and sellers can send encrypted file or image attachments in dispute chat. - Rebuilds `admin_dispute_chats` so existing disputes immediately show their chat history in the UI. - Computes per‑party max timestamps and updates `AppState.admin_chat_last_seen`. - These timestamps are also stored in the `admin_disputes` table as `buyer_chat_last_seen` and `seller_chat_last_seen`. - - The background listener uses these DB fields as cursors for `fetch_admin_chat_updates`, so only newer NIP‑59 events are fetched after restart. A single-flight guard ensures only one admin chat fetch runs at a time (see `src/util/order_utils/fetch_scheduler.rs`). + - The shared-key chat subscription router (`listen_for_chat_messages` in `src/util/chat_listener.rs`) uses these DB fields as cursors to hydrate history once per key on track (`fetch_gift_wraps_for_shared_key`), then receives newer NIP‑59 events live over one batched `kind: 1059` subscription. Disputes are tracked via `track_dispute_chat` when taken and re-tracked by `track_startup_chats` at startup/reconnect. - This hybrid approach keeps the protocol stateless while giving admins a smooth, restart-safe chat experience across application restarts. #### Keyboard Shortcuts diff --git a/docs/MESSAGE_FLOW_AND_PROTOCOL.md b/docs/MESSAGE_FLOW_AND_PROTOCOL.md index 13d9cbf..863be71 100644 --- a/docs/MESSAGE_FLOW_AND_PROTOCOL.md +++ b/docs/MESSAGE_FLOW_AND_PROTOCOL.md @@ -321,7 +321,7 @@ In addition to relay-driven trade DMs, Mostrix keeps a lightweight local transcr - **Path**: `~/.mostrix/orders_chat/.txt` - **Startup restore**: `load_user_order_chats_at_startup` restores cached chat into `AppState.order_chats`, seeds `order_chat_last_seen`, then runs an initial `fetch_user_order_chat_updates` relay backfill. -- **Periodic relay sync (User role)**: on the shared **`admin_chat_interval`** timer (every **2 seconds** in `src/main.rs`), the User-role branch calls `spawn_user_order_chat_fetch`, which polls active orders via `fetch_user_order_chat_updates`. Uses the same single-flight `CHAT_MESSAGES_SEMAPHORE` as admin chat so overlapping fetches are skipped. Shared keys come from persisted `order_chat_shared_key_hex` when set, otherwise ECDH from local `trade_keys` + `counterparty_pubkey` (`src/util/chat_utils.rs`). +- **Live relay sync (User role)**: the **shared-key chat subscription router** (`listen_for_chat_messages` in `src/util/chat_listener.rs`) maintains one batched `kind: 1059` subscription over all active order shared pubkeys and routes incoming gift wraps by `p` tag. `track_startup_chats` seeds the active-order set at startup; the DM router tracks/untracks orders as their shared key becomes resolvable or they reach a terminal status. Shared keys come from persisted `order_chat_shared_key_hex` when set, otherwise ECDH from local `trade_keys` + `counterparty_pubkey` (`src/util/chat_utils.rs`). - **Incremental merge**: `apply_user_order_chat_updates` in `src/ui/helpers/startup.rs`: - **Skip own relay echoes**: each `OrderChatUpdate` carries `local_trade_pubkey`; messages whose decrypted `sender_pubkey` matches are ignored (same rule as admin chat and Mostro Mobile — avoids showing your send on both **You** and **Peer** after the optimistic local append on Enter). - **Dedup**: relay self-echoes are skipped by `sender_pubkey == local_trade_pubkey`; peer dedup matches only existing **Peer** rows at the same `(timestamp, content)` (or same attachment / legacy placeholder) so an optimistic **You** line cannot hide a real counterparty message in the same second. @@ -588,26 +588,20 @@ User-facing strings for `Payload::CantDo(Some(reason))` come from [`get_cant_do_ Trade DMs carrying `CantDo` are not upserted into the Messages list ([DM_LISTENER_FLOW.md](DM_LISTENER_FLOW.md)); they surface via waiters / `OperationResult` popups. -## Admin Chat Fetch (Single-Flight, Shared-Key Based) +## Admin Chat (Shared-Key Subscription Router) -When the user is in **Admin** mode, the main event loop runs a periodic admin chat sync so the "Disputes in Progress" tab stays up to date with NIP‑59 gift-wrap messages exchanged over **per‑dispute shared keys**. +When the user is in **Admin** mode, the shared-key chat subscription router keeps the "Disputes in Progress" tab up to date with NIP‑59 gift-wrap messages exchanged over **per‑dispute shared keys** — live via a relay subscription, not a timer. -- **Trigger**: Every **2 seconds** on the shared **`admin_chat_interval`** timer in `src/main.rs` when `app.user_role == UserRole::Admin` (the same timer’s User-role branch calls `spawn_user_order_chat_fetch` instead). -- **Shared keys**: For each `AdminDispute` in `InProgress` state, the database may hold `buyer_shared_key_hex` / `seller_shared_key_hex`. At runtime these are converted back to `Keys` via `keys_from_shared_hex` in `src/util/chat_utils.rs`. -- **Entry point**: `spawn_admin_chat_fetch` in `src/util/order_utils/fetch_scheduler.rs` is called with the Nostr client, the current disputes, `admin_chat_last_seen`, and the channel to send results. -- **Single-flight guard**: A shared `AtomicBool` (`CHAT_MESSAGES_SEMAPHORE`) ensures that only one admin chat fetch runs concurrently. If a previous fetch is still running, subsequent ticks are skipped until the flag is cleared. -- **Fetch work**: The spawned task calls `fetch_admin_chat_updates`, which, for every dispute+party that has a stored shared key: - - Rebuilds the shared `Keys` from hex. - - Fetches `Kind::GiftWrap` events **addressed to the shared key’s public key** over a 7‑day rolling window. - - Decrypts each event with the shared key (first trying standard NIP‑59 `from_gift_wrap`, then falling back to the simplified Mostro‑chat format). - - Applies per‑(dispute, party) `last_seen_timestamp` filtering so only newer messages are returned. +- **Trigger**: `execute_take_dispute` calls `track_dispute_chat` for the buyer and seller when a dispute is taken; `track_startup_chats` re-tracks all `InProgress` disputes at startup and on reconnect. Untracked when the dispute leaves `InProgress`. +- **Shared keys**: For each `AdminDispute` in `InProgress` state, the database holds `buyer_shared_key_hex` / `seller_shared_key_hex`, converted back to `Keys` via `keys_from_shared_hex` in `src/util/chat_utils.rs`. +- **Router**: `listen_for_chat_messages` (`src/util/chat_listener.rs`) subscribes one batched `kind: 1059` filter over all tracked shared pubkeys, hydrates history once per key on track (`fetch_gift_wraps_for_shared_key`, 7‑day window, filtered by `last_seen`), and routes live gift wraps by `p` tag. - **Application**: The main loop receives results on `admin_chat_updates_rx` and applies them via `apply_admin_chat_updates`, which: - Appends new `DisputeChatMessage` items into `AppState.admin_dispute_chats`. - Updates in‑memory `admin_chat_last_seen` entries. - Persists cursors to the `admin_disputes` table (`buyer_chat_last_seen`, `seller_chat_last_seen`) via `update_chat_last_seen_by_dispute_id`. - **Attachments**: Attachment messages (Mostro Mobile Encrypted File Messaging: `image_encrypted` / `file_encrypted`) are parsed into structured attachment entries. From the dispute chat, the admin presses **Ctrl+S** to open a **Save attachment** popup listing all attachments for the current dispute/party; they select one with ↑/↓ and press Enter to download from Blossom (`blossom://` → `https://`), optionally decrypt with ChaCha20‑Poly1305 (nonce + ciphertext + tag), and save to `~/.mostrix/downloads/_`. See `src/util/blossom.rs` and the "Receiving and saving file attachments" section in [ADMIN_DISPUTES.md](ADMIN_DISPUTES.md). -This avoids overlapping relay queries and duplicate work when the 2‑second tick fires before a previous fetch has finished, while ensuring admin chat is driven entirely by the per‑dispute shared keys stored in the database. +Admin chat is driven entirely by the per‑dispute shared keys stored in the database, delivered live over one batched subscription (same model as Mostro Mobile's `SubscriptionManager`). ### Database Errors Database operations (saving orders, updating trade indices) log errors but don't necessarily fail the entire operation, allowing the user to continue using the client. diff --git a/docs/STARTUP_AND_CONFIG.md b/docs/STARTUP_AND_CONFIG.md index 1018c40..25accb1 100644 --- a/docs/STARTUP_AND_CONFIG.md +++ b/docs/STARTUP_AND_CONFIG.md @@ -169,18 +169,16 @@ Several background tasks are spawned to keep the UI and data in sync: - On `Offline`, startup overlay text indicates automatic retry. - On `Online`, `main.rs` triggers `reload_runtime_session_after_reconnect(...)` to reconnect and reload runtime background tasks. -5. **Shared chat relay poll** (`admin_chat_interval`, 2 seconds in `src/main.rs`): - - **Admin role**: triggers `spawn_admin_chat_fetch` → `fetch_admin_chat_updates` (see `src/util/order_utils/fetch_scheduler.rs`). - - For each in-progress dispute, rebuilds per-party shared `Keys` from `buyer_shared_key_hex` / `seller_shared_key_hex` stored in the `admin_disputes` table. - - Fetches NIP‑59 `GiftWrap` events addressed to each shared key's public key (ECDH-derived, same model as `mostro-chat`). - - Uses per‑party `last_seen_timestamp` values to request only new events. - - Delegates application of updates to `ui::helpers::apply_admin_chat_updates` (implemented in `src/ui/helpers/startup.rs`), which: - - Appends new `DisputeChatMessage` items into `AppState.admin_dispute_chats`. - - Persists updated buyer/seller chat cursors in the `admin_disputes` table (`buyer_chat_last_seen`, `seller_chat_last_seen`). - - **User role**: triggers `spawn_user_order_chat_fetch` → `fetch_user_order_chat_updates` on the same timer (shared keys from `order_chat_shared_key_hex` or `trade_keys` + `counterparty_pubkey`; applied via `apply_user_order_chat_updates`). - - A **single-flight guard** (`CHAT_MESSAGES_SEMAPHORE`: `AtomicBool`) ensures only one shared-key chat fetch runs at a time; overlapping ticks skip spawning a new fetch until the current one completes. - -**Source**: `src/main.rs` (background task setup), `src/util/order_utils/fetch_scheduler.rs` (admin chat scheduler), `src/ui/helpers/startup.rs` (`apply_admin_chat_updates`) +5. **Shared-key chat subscription router** (`listen_for_chat_messages` in `src/util/chat_listener.rs`): + - A **single long-lived task** (spawned in `src/startup.rs`, respawned on reconnect/key reload) maintains **one** relay subscription whose filter batches **all** active shared-key pubkeys (`kind: 1059`, `#p: [shared pubkeys]`) — the same model as Mostro Mobile's `SubscriptionManager`. No timed polling. + - **Track/untrack** via the global command channel (`ChatRouterCmd`, published by `set_chat_router_cmd_tx`). Helpers: `track_order_chat` / `untrack_order_chat` / `track_dispute_chat` / `untrack_dispute_chat`. + - On `TrackChatKey`, the router hydrates history once (`fetch_gift_wraps_for_shared_key`, filtered by the last-seen cursor), then rebuilds the batched subscription (idempotent: re-tracking an already-tracked key is a no-op). + - Live `kind: 1059` gift wraps are routed by the event's `p` tag to the owning shared key, decrypted with `unwrap_giftwrap_with_shared_key`, and emitted on the existing `admin_chat_updates` / `user_order_chat_updates` channels. + - **Application is unchanged**: `ui::helpers::apply_admin_chat_updates` / `apply_user_order_chat_updates` (in `src/ui/helpers/startup.rs`) still merge updates, persist cursors (`buyer_chat_last_seen` / `seller_chat_last_seen`, order transcripts), and dedupe. + - **Startup track set (option B)**: `track_startup_chats` emits tracks for `Order::get_startup_active_orders` rows (active + `success`) with a resolvable shared key (User), and each `InProgress` dispute's buyer/seller shared key (Admin). + - **Untrack** when an order reaches `TERMINAL_DM_STATUSES` or its row is removed/reverted (DM router hooks), or a dispute leaves `InProgress`. Chat is **not** untracked on `success`; on-disk transcripts preserve history for untracked terminal orders. + +**Source**: `src/util/chat_listener.rs` (router), `src/startup.rs` + `src/ui/helpers/startup.rs` (`track_startup_chats`), `src/util/dm_utils/mod.rs` (track/untrack hooks), `src/ui/helpers/startup.rs` (`apply_admin_chat_updates`) 6. **DM Router Wiring (trade messages)**: - App channel creation includes `dm_subscription_tx` / `dm_subscription_rx`. @@ -215,7 +213,7 @@ For **User** role, Mostrix restores peer-to-peer order chat alongside trade DMs: - Cached transcripts live under `~/.mostrix/orders_chat/.txt` and are loaded into `AppState.order_chats` by `load_user_order_chats_at_startup`. - **Attachment rows in transcripts** are stored as **JSON** (`image_encrypted` / `file_encrypted` via `serialize_attachment_for_transcript`) so **Ctrl+S** and file counts work immediately after restart; legacy `[Image: … - Ctrl+S to save]` lines are hydrated in memory when relay returns the same attachment at the same timestamp. -- An immediate relay fetch (`fetch_user_order_chat_updates`) merges any newer gift-wrap messages; subsequent polls run every **2 seconds** on the shared `admin_chat_interval` timer via `spawn_user_order_chat_fetch` in `src/util/order_utils/fetch_scheduler.rs`. +- An immediate relay fetch (`fetch_user_order_chat_updates`) merges any newer gift-wrap messages; live updates then arrive via the **shared-key chat subscription router** (`listen_for_chat_messages`), which `track_startup_chats` seeds with the active-order set — no timed polling. - `apply_user_order_chat_updates` skips relay echoes of the local trade pubkey; peer dedup is scoped to existing **Peer** rows so optimistic **You** sends are not mirrored as **Peer** and do not suppress unrelated peer text at the same timestamp. See [MESSAGE_FLOW_AND_PROTOCOL.md](MESSAGE_FLOW_AND_PROTOCOL.md) — "User order chat local cache". ## Main Event Loop diff --git a/docs/TUI_INTERFACE.md b/docs/TUI_INTERFACE.md index 9c5c6df..bfd03d3 100644 --- a/docs/TUI_INTERFACE.md +++ b/docs/TUI_INTERFACE.md @@ -428,7 +428,7 @@ The key handler processes input in this order: - Published to relays without blocking the main UI thread. - **Receiving messages**: - - The main loop calls `spawn_admin_chat_fetch` every **2 seconds** on the shared `admin_chat_interval` timer when in Admin mode (User mode uses the same timer for `spawn_user_order_chat_fetch`). Each spawn runs `fetch_admin_chat_updates` in a one-off task. A single-flight guard (`CHAT_MESSAGES_SEMAPHORE`) ensures only one shared-key chat fetch runs at a time; overlapping interval ticks skip spawning until the current fetch completes. + - The shared-key chat subscription router (`listen_for_chat_messages`) delivers messages live over one batched `kind: 1059` subscription; disputes are tracked via `track_dispute_chat` when taken and re-tracked by `track_startup_chats` at startup/reconnect. History is hydrated once per key on track. - For each in-progress dispute, the fetch: - Rebuilds buyer/seller shared `Keys` from the stored hex. - Fetches `GiftWrap` events addressed to each shared key's public key (7-day rolling window). diff --git a/src/main.rs b/src/main.rs index da8de73..ead056d 100644 --- a/src/main.rs +++ b/src/main.rs @@ -17,13 +17,12 @@ use crate::ui::helpers::{ use crate::ui::key_handler::{ apply_pending_runtime_reloads, create_app_channels, handle_key_event, handle_mouse_invoice_paste_fallback, reload_runtime_session_after_reconnect, - respawn_trade_dm_listener, AppChannels, RuntimeReconnectContext, + respawn_chat_listener, respawn_trade_dm_listener, AppChannels, RuntimeReconnectContext, }; use crate::ui::{LnAddressVerifyResult, MostroInfoFetchResult, OperationResult}; use crate::util::{ blossom_servers_from_settings, handle_message_notification, handle_operation_result, - install_background_panic_hook, - order_utils::{spawn_admin_chat_fetch, spawn_user_order_chat_fetch, validate_range_amount}, + install_background_panic_hook, order_utils::validate_range_amount, set_chat_router_cmd_tx, set_dm_router_cmd_tx, set_fatal_error_tx, set_order_result_tx, spawn_save_attachment, spawn_send_order_chat_attachment, }; @@ -273,6 +272,8 @@ async fn main() -> Result<(), anyhow::Error> { mut mostro_info_rx, mut dm_subscription_tx, dm_subscription_rx, + mut chat_router_cmd_tx, + chat_router_cmd_rx, network_status_tx, mut network_status_rx, fatal_error_tx, @@ -289,6 +290,10 @@ async fn main() -> Result<(), anyhow::Error> { set_dm_router_cmd_tx(dm_subscription_tx.clone()).map_err(|msg| { anyhow::anyhow!("{msg}: DM router sender was not registered; restart the application.") })?; + // Set chat router tx (shared-key order + dispute chat subscriptions). + set_chat_router_cmd_tx(chat_router_cmd_tx.clone()).map_err(|msg| { + anyhow::anyhow!("{msg}: chat router sender was not registered; restart the application.") + })?; set_order_result_tx(order_result_tx.clone()).map_err(|msg| anyhow::anyhow!(msg))?; let configured_relays: Vec = settings @@ -311,6 +316,7 @@ async fn main() -> Result<(), anyhow::Error> { mut dispute_task, mut app, mut message_listener_handle, + mut chat_listener_handle, .. } = startup::run_startup_with_splash( &mut terminal, @@ -321,9 +327,12 @@ async fn main() -> Result<(), anyhow::Error> { user_role, configured_relays, dm_subscription_rx, + chat_router_cmd_rx, mostro_info_tx: mostro_info_tx.clone(), network_status_tx, message_notification_tx: message_notification_tx.clone(), + admin_chat_updates_tx: admin_chat_updates_tx.clone(), + user_order_chat_updates_tx: user_order_chat_updates_tx.clone(), }, ) .await?; @@ -331,7 +340,6 @@ async fn main() -> Result<(), anyhow::Error> { // Event handling: keyboard input and periodic UI refresh. let mut events = EventStream::new(); let mut refresh_interval = interval(Duration::from_millis(150)); - let mut admin_chat_interval = interval(Duration::from_secs(2)); loop { tokio::select! { @@ -341,6 +349,7 @@ async fn main() -> Result<(), anyhow::Error> { order_task.abort(); dispute_task.abort(); message_listener_handle.abort(); + chat_listener_handle.abort(); app.fatal_exit_on_close = true; app.mode = UiMode::operation_result(OperationResult::Error(msg)); } @@ -378,6 +387,22 @@ async fn main() -> Result<(), anyhow::Error> { { Ok(()) => { app.offline_overlay_message = None; + // Reconnect ran `unsubscribe_all`; rebuild the chat subscription. + if let Err(e) = respawn_chat_listener( + &app, + &client, + &pool, + &mut chat_listener_handle, + &mut chat_router_cmd_tx, + &admin_chat_updates_tx, + &user_order_chat_updates_tx, + ) + .await + { + log::error!( + "Failed to respawn chat listener after reconnect: {e}" + ); + } } Err(e) => { app.offline_overlay_message = Some(format!( @@ -578,7 +603,9 @@ async fn main() -> Result<(), anyhow::Error> { &dm_subscription_tx, ) { Some(true) => { - if app.pending_key_reload || app.pending_fetch_scheduler_reload { + let needs_chat_respawn = + app.pending_key_reload || app.pending_fetch_scheduler_reload; + if needs_chat_respawn { apply_pending_runtime_reloads( &mut app, &mut client, @@ -595,6 +622,25 @@ async fn main() -> Result<(), anyhow::Error> { settings, ) .await; + // Reloads replace the client / run `unsubscribe_all`, dropping the + // chat subscription; respawn once the reload actually completed. + if !app.pending_key_reload && !app.pending_fetch_scheduler_reload { + if let Err(e) = respawn_chat_listener( + &app, + &client, + &pool, + &mut chat_listener_handle, + &mut chat_router_cmd_tx, + &admin_chat_updates_tx, + &user_order_chat_updates_tx, + ) + .await + { + log::error!( + "Failed to respawn chat listener after reload: {e}" + ); + } + } } if app.pending_admin_disputes_reload { app.pending_admin_disputes_reload = false; @@ -612,25 +658,6 @@ async fn main() -> Result<(), anyhow::Error> { _ = refresh_interval.tick() => { // Refresh the UI even if there is no input. } - _ = admin_chat_interval.tick() => { - if app.user_role == UserRole::Admin { - if app.admin_keys.is_some() { - spawn_admin_chat_fetch( - client.clone(), - app.admin_disputes_in_progress.clone(), - app.admin_chat_last_seen.clone(), - admin_chat_updates_tx.clone(), - ); - } - } else if app.user_role == UserRole::User { - spawn_user_order_chat_fetch( - client.clone(), - pool.clone(), - app.order_chat_last_seen.clone(), - user_order_chat_updates_tx.clone(), - ); - } - } } // Ensure the selected index is valid when orders list changes. @@ -644,6 +671,7 @@ async fn main() -> Result<(), anyhow::Error> { order_task.abort(); dispute_task.abort(); message_listener_handle.abort(); + chat_listener_handle.abort(); app.fatal_exit_on_close = true; app.mode = UiMode::operation_result(OperationResult::Error(msg)); 0 @@ -668,6 +696,7 @@ async fn main() -> Result<(), anyhow::Error> { order_task.abort(); dispute_task.abort(); message_listener_handle.abort(); + chat_listener_handle.abort(); app.fatal_exit_on_close = true; app.mode = UiMode::operation_result(OperationResult::Error(msg)); continue; diff --git a/src/startup.rs b/src/startup.rs index 77c1dcc..0f994ff 100644 --- a/src/startup.rs +++ b/src/startup.rs @@ -19,7 +19,7 @@ use crate::models::User; use crate::settings::Settings; use crate::ui::helpers::{ hydrate_app_admin_keys_from_privkey, load_admin_disputes_at_startup, - load_user_order_chats_at_startup, + load_user_order_chats_at_startup, track_startup_chats, }; use crate::ui::network_status::spawn_network_status_monitor; use crate::ui::startup_splash::{ @@ -28,7 +28,8 @@ use crate::ui::startup_splash::{ use crate::ui::{AppState, OperationResult, UiMode, UserRole}; use crate::util::{ any_relay_reachable, catch_unwind_request_fatal_restart, connect_client_safely, - fetch_mostro_instance_info, hydrate_startup_active_order_dm_state, listen_for_order_messages, + fetch_mostro_instance_info, hydrate_startup_active_order_dm_state, listen_for_chat_messages, + listen_for_order_messages, order_utils::{ run_relay_order_db_reconcile_once, run_targeted_relay_order_db_reconcile_tick, start_fetch_scheduler, FetchSchedulerResult, @@ -44,9 +45,17 @@ pub struct PostTerminalStartupInput<'a> { pub configured_relays: Vec, pub dm_subscription_rx: tokio::sync::mpsc::UnboundedReceiver, + /// Receiver for the shared-key chat subscription router (track/untrack commands). + pub chat_router_cmd_rx: tokio::sync::mpsc::UnboundedReceiver, pub mostro_info_tx: tokio::sync::mpsc::UnboundedSender, pub network_status_tx: tokio::sync::mpsc::UnboundedSender, pub message_notification_tx: tokio::sync::mpsc::UnboundedSender, + /// Sender used by the chat router to publish decoded admin dispute chat updates. + pub admin_chat_updates_tx: + tokio::sync::mpsc::UnboundedSender, anyhow::Error>>, + /// Sender used by the chat router to publish decoded user order chat updates. + pub user_order_chat_updates_tx: + tokio::sync::mpsc::UnboundedSender, anyhow::Error>>, } pub struct StartupBootstrap { @@ -59,6 +68,8 @@ pub struct StartupBootstrap { pub dispute_task: JoinHandle<()>, pub app: AppState, pub message_listener_handle: JoinHandle<()>, + /// Shared-key chat subscription router task (user order + admin dispute chat). + pub chat_listener_handle: JoinHandle<()>, pub relays_reachable: bool, } @@ -162,6 +173,9 @@ pub async fn run_post_terminal_startup( } } load_user_order_chats_at_startup(&client, input.pool, &mut app).await; + // Emit initial chat-router track commands for the active set (option B). Buffered on the + // router's channel until the chat listener task (spawned below) starts consuming them. + track_startup_chats(input.pool, &app).await; set_startup_phase(phase_tx, "Almost ready…"); let startup_dm_hydration = match hydrate_startup_active_order_dm_state(input.pool).await { @@ -234,6 +248,25 @@ pub async fn run_post_terminal_startup( .await; }); + // Single shared-key chat subscription router (user order chat + admin dispute chat). + // Track/untrack commands arrive via the global sender registered in `main.rs`. + let client_for_chat = client.clone(); + let admin_chat_updates_tx_for_chat = input.admin_chat_updates_tx.clone(); + let user_order_chat_updates_tx_for_chat = input.user_order_chat_updates_tx.clone(); + let chat_router_cmd_rx = input.chat_router_cmd_rx; + let chat_listener_handle = tokio::spawn(async move { + catch_unwind_request_fatal_restart("chat subscription router", async move { + listen_for_chat_messages( + client_for_chat, + admin_chat_updates_tx_for_chat, + user_order_chat_updates_tx_for_chat, + chat_router_cmd_rx, + ) + .await; + }) + .await; + }); + Ok(StartupBootstrap { client, mostro_pubkey, @@ -244,6 +277,7 @@ pub async fn run_post_terminal_startup( dispute_task, app, message_listener_handle, + chat_listener_handle, relays_reachable, }) } diff --git a/src/ui/helpers/mod.rs b/src/ui/helpers/mod.rs index d4bc321..1538ff3 100644 --- a/src/ui/helpers/mod.rs +++ b/src/ui/helpers/mod.rs @@ -39,4 +39,5 @@ pub use startup::{ hydrate_app_admin_keys_from_privkey, load_admin_disputes_at_startup, load_user_order_chats_at_startup, recover_admin_chat_from_files, refresh_my_trades_maker_book_cache, sync_user_order_history_messages_from_db, + track_startup_chats, }; diff --git a/src/ui/helpers/startup.rs b/src/ui/helpers/startup.rs index 3436a35..868d68d 100644 --- a/src/ui/helpers/startup.rs +++ b/src/ui/helpers/startup.rs @@ -1,7 +1,9 @@ use std::collections::HashMap; use std::str::FromStr; -use mostro_core::prelude::{Action, Kind as OrderKind, Message, Payload, SmallOrder, Status}; +use mostro_core::prelude::{ + Action, DisputeStatus, Kind as OrderKind, Message, Payload, SmallOrder, Status, +}; use nostr_sdk::prelude::{Client, Keys, PublicKey}; use sqlx::SqlitePool; use uuid::Uuid; @@ -13,7 +15,11 @@ use crate::ui::{ OrderChatLastSeen, OrderChatStaticHeader, OrderMessage, UserChatSender, UserOrderChatMessage, UserRole, }; -use crate::util::{chat_utils::fetch_user_order_chat_updates, seed_admin_chat_last_seen}; +use crate::util::{ + chat_listener::{track_dispute_chat, track_order_chat}, + chat_utils::{derive_shared_key_hex, fetch_user_order_chat_updates}, + seed_admin_chat_last_seen, +}; use super::attachments::{ build_attachment_toast, legacy_placeholder_matches_filename, try_parse_attachment_message, @@ -119,6 +125,74 @@ pub async fn load_admin_disputes_at_startup(pool: &SqlitePool, app: &mut AppStat } } +/// Emit initial chat-router track commands for the active set (option B). +/// +/// - **User**: every [`Order::get_startup_active_orders`] row (active states + `success`; +/// excludes [`crate::models::TERMINAL_DM_STATUSES`]) with a resolvable shared key — +/// persisted `order_chat_shared_key_hex`, else ECDH from `trade_keys` + `counterparty_pubkey`. +/// - **Admin**: each InProgress dispute's buyer/seller shared key. +/// +/// Commands are buffered on the router's channel until the task starts consuming them, so this +/// is safe to call before the chat router task is spawned. History for each key is hydrated by +/// the router on `TrackChatKey` using the passed `since` (last-seen) cursor. +pub async fn track_startup_chats(pool: &SqlitePool, app: &AppState) { + match app.user_role { + UserRole::User => { + let Ok(rows) = Order::get_startup_active_orders(pool).await else { + return; + }; + for row in rows { + let Ok(order) = Order::get_by_id(pool, &row.id).await else { + continue; + }; + let Some(trade_keys_hex) = order.trade_keys.as_deref().filter(|v| !v.is_empty()) + else { + continue; + }; + let Ok(trade_keys) = Keys::parse(trade_keys_hex) else { + continue; + }; + let shared_hex = order.order_chat_shared_key_hex.clone().or_else(|| { + derive_shared_key_hex(Some(&trade_keys), order.counterparty_pubkey.as_deref()) + }); + let Some(shared_hex) = shared_hex else { + continue; + }; + let since = app + .order_chat_last_seen + .get(&row.id) + .and_then(|s| s.last_seen_timestamp); + track_order_chat(row.id.clone(), shared_hex, trade_keys.public_key(), since); + } + } + UserRole::Admin => { + for dispute in &app.admin_disputes_in_progress { + let is_in_progress = dispute + .status + .as_deref() + .and_then(|s| DisputeStatus::from_str(s).ok()) + == Some(DisputeStatus::InProgress); + if !is_in_progress { + continue; + } + for (party, hex) in [ + (ChatParty::Buyer, dispute.buyer_shared_key_hex.as_deref()), + (ChatParty::Seller, dispute.seller_shared_key_hex.as_deref()), + ] { + let Some(hex) = hex else { + continue; + }; + let since = app + .admin_chat_last_seen + .get(&(dispute.dispute_id.clone(), party)) + .and_then(|s| s.last_seen_timestamp); + track_dispute_chat(dispute.dispute_id.clone(), party, hex.to_string(), since); + } + } + } + } +} + /// Load user order chat at startup. pub async fn load_user_order_chats_at_startup( client: &Client, diff --git a/src/ui/key_handler/async_tasks.rs b/src/ui/key_handler/async_tasks.rs index 739de12..f9e52c6 100644 --- a/src/ui/key_handler/async_tasks.rs +++ b/src/ui/key_handler/async_tasks.rs @@ -1,7 +1,7 @@ use crate::models::{Order, User}; use crate::settings::load_settings_from_disk; use crate::settings::Settings; -use crate::ui::helpers::hydrate_app_admin_keys_from_privkey; +use crate::ui::helpers::{hydrate_app_admin_keys_from_privkey, track_startup_chats}; use crate::ui::key_handler::EnterKeyContext; use crate::ui::FormState; use crate::ui::{ @@ -14,8 +14,9 @@ use crate::util::listen_for_order_messages; use crate::util::order_utils::spawn_fetch_scheduler_loops; use crate::util::{ any_relay_reachable, catch_unwind_request_fatal_restart, connect_client_safely, - hydrate_startup_active_order_dm_state, set_dm_router_cmd_tx, - unsubscribe_dm_listener_subscriptions, OrderDmSubscriptionCmd, StartupDmHydration, + hydrate_startup_active_order_dm_state, listen_for_chat_messages, set_chat_router_cmd_tx, + set_dm_router_cmd_tx, unsubscribe_dm_listener_subscriptions, ChatRouterCmd, + OrderDmSubscriptionCmd, StartupDmHydration, }; use mostro_core::prelude::{Dispute, SmallOrder, Transport}; use nostr_sdk::prelude::{Client, Keys, PublicKey}; @@ -717,6 +718,45 @@ pub async fn reload_runtime_session_after_reconnect( } } +/// Abort and respawn the shared-key chat subscription router on a new/renewed client. +/// +/// Needed after key reload (client replaced) or reconnect / fetch-scheduler reload +/// (`client.unsubscribe_all()` drops the chat subscription). Rebuilds the router with a fresh +/// command channel, re-registers the global sender, and re-emits the active track set +/// ([`track_startup_chats`], option B) so every chat resubscribes on the new client/session. +pub async fn respawn_chat_listener( + app: &AppState, + client: &Client, + pool: &SqlitePool, + chat_listener_handle: &mut JoinHandle<()>, + chat_router_cmd_tx: &mut UnboundedSender, + admin_chat_updates_tx: &UnboundedSender, anyhow::Error>>, + user_order_chat_updates_tx: &UnboundedSender, anyhow::Error>>, +) -> Result<(), String> { + chat_listener_handle.abort(); + // Await the old task so it releases its subscription/state before the new one starts. + let old = std::mem::replace(chat_listener_handle, tokio::spawn(async {})); + let _ = old.await; + + let (new_tx, new_rx) = tokio::sync::mpsc::unbounded_channel::(); + *chat_router_cmd_tx = new_tx; + set_chat_router_cmd_tx(chat_router_cmd_tx.clone()).map_err(|msg| msg.to_string())?; + + let client_for_chat = client.clone(); + let admin_tx = admin_chat_updates_tx.clone(); + let user_tx = user_order_chat_updates_tx.clone(); + *chat_listener_handle = tokio::spawn(async move { + catch_unwind_request_fatal_restart("chat subscription router", async move { + listen_for_chat_messages(client_for_chat, admin_tx, user_tx, new_rx).await; + }) + .await; + }); + + // Re-emit the active track set so all chats resubscribe on the new client/session. + track_startup_chats(pool, app).await; + Ok(()) +} + pub struct AppChannels { pub order_result_tx: UnboundedSender, pub order_result_rx: UnboundedReceiver, @@ -738,6 +778,8 @@ pub struct AppChannels { pub mostro_info_rx: UnboundedReceiver, pub dm_subscription_tx: UnboundedSender, pub dm_subscription_rx: UnboundedReceiver, + pub chat_router_cmd_tx: UnboundedSender, + pub chat_router_cmd_rx: UnboundedReceiver, pub network_status_tx: UnboundedSender, pub network_status_rx: UnboundedReceiver, pub fatal_error_tx: UnboundedSender, @@ -767,6 +809,8 @@ pub fn create_app_channels() -> AppChannels { tokio::sync::mpsc::unbounded_channel::(); let (dm_subscription_tx, dm_subscription_rx) = tokio::sync::mpsc::unbounded_channel::(); + let (chat_router_cmd_tx, chat_router_cmd_rx) = + tokio::sync::mpsc::unbounded_channel::(); let (network_status_tx, network_status_rx) = tokio::sync::mpsc::unbounded_channel::(); let (fatal_error_tx, fatal_error_rx) = tokio::sync::mpsc::unbounded_channel::(); @@ -794,6 +838,8 @@ pub fn create_app_channels() -> AppChannels { mostro_info_rx, dm_subscription_tx, dm_subscription_rx, + chat_router_cmd_tx, + chat_router_cmd_rx, network_status_tx, network_status_rx, fatal_error_tx, diff --git a/src/ui/key_handler/mod.rs b/src/ui/key_handler/mod.rs index e2ef0bb..4ed6a59 100644 --- a/src/ui/key_handler/mod.rs +++ b/src/ui/key_handler/mod.rs @@ -76,8 +76,9 @@ fn is_terminal_order_status(status: Option) -> bool { // Re-export public functions pub use async_tasks::{ apply_pending_fetch_scheduler_reload, apply_pending_key_reload, apply_pending_runtime_reloads, - create_app_channels, reload_runtime_session_after_reconnect, respawn_trade_dm_listener, - spawn_refresh_mostro_info_task, AppChannels, RuntimeReconnectContext, + create_app_channels, reload_runtime_session_after_reconnect, respawn_chat_listener, + respawn_trade_dm_listener, spawn_refresh_mostro_info_task, AppChannels, + RuntimeReconnectContext, }; pub use confirmation::{handle_cancel_key, handle_confirm_key}; pub use enter_handlers::handle_enter_key; diff --git a/src/util/chat_listener.rs b/src/util/chat_listener.rs new file mode 100644 index 0000000..70d40b2 --- /dev/null +++ b/src/util/chat_listener.rs @@ -0,0 +1,366 @@ +//! Shared-key chat subscription router (P2P order chat + admin dispute chat). +//! +//! Replaces the previous 2-second timed `fetch_events` polling with a single +//! long-lived subscription, mirroring Mostro Mobile's `SubscriptionManager`: one +//! relay subscription whose filter batches **all** active shared-key pubkeys +//! (`kind: 1059`, `#p: [shared pubkeys]`). Incoming gift wraps are routed to the +//! owning chat by matching the event's `p` tag against the tracked pubkey set, +//! decrypted with the per-channel shared `Keys`, and emitted on the existing +//! `admin_chat_updates` / `user_order_chat_updates` channels so the +//! `apply_*_chat_updates` handlers stay unchanged. +//! +//! Lifecycle (see also `docs/DM_LISTENER_FLOW.md`): the task is spawned once at +//! startup and respawned on client reload/reconnect, exactly like the trade DM +//! listener. Chat keys are tracked/untracked via the global command channel +//! published by [`set_chat_router_cmd_tx`]. + +use std::collections::HashMap; +use std::sync::Mutex; + +use nostr_sdk::prelude::*; +use tokio::sync::mpsc::{self, UnboundedSender}; +use uuid::Uuid; + +use crate::models::Order; +use crate::ui::{AdminChatUpdate, ChatParty, OrderChatUpdate}; +use crate::util::chat_utils::{ + derive_shared_key_hex, fetch_gift_wraps_for_shared_key, keys_from_shared_hex, + unwrap_giftwrap_with_shared_key, +}; + +/// Identifies which chat a shared key belongs to. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum ChatKeyId { + /// User P2P order chat, keyed by order id (UUID string). + Order(String), + /// Admin dispute chat, keyed by dispute id + party (buyer/seller). + Dispute(String, ChatParty), +} + +/// Commands consumed by [`listen_for_chat_messages`]. +pub enum ChatRouterCmd { + /// Start tracking a shared-key chat: hydrate history once, then include the + /// key's pubkey in the batched live subscription. + TrackChatKey { + key_id: ChatKeyId, + /// Hex of the ECDH shared secret (round-trips via [`keys_from_shared_hex`]). + shared_key_hex: String, + /// Local trade pubkey for order chats (used to skip relay echoes of our own sends). + local_trade_pubkey: Option, + /// Only emit history messages at/after this unix timestamp (last-seen cursor). + since: Option, + }, + /// Stop tracking a shared-key chat (order row deleted / dispute finalized, etc.). + UntrackChatKey { key_id: ChatKeyId }, +} + +/// Per-tracked-key routing metadata. +struct ChatTarget { + key_id: ChatKeyId, + shared_keys: Keys, + /// Order chat only; enables echo-skip in `apply_user_order_chat_updates`. + local_trade_pubkey: Option, +} + +/// Global sender published for track/untrack helpers, mirroring the DM router's +/// `DM_ROUTER_CMD_TX`. Set at startup and on every chat-router respawn. +static CHAT_ROUTER_CMD_TX: Mutex>> = Mutex::new(None); + +/// Publishes the sender consumed by [`listen_for_chat_messages`]. +/// +/// Returns `Err` if the mutex is poisoned (the sender was **not** updated). +pub fn set_chat_router_cmd_tx( + tx: mpsc::UnboundedSender, +) -> Result<(), &'static str> { + match CHAT_ROUTER_CMD_TX.lock() { + Ok(mut guard) => { + *guard = Some(tx); + Ok(()) + } + Err(_) => { + crate::util::request_fatal_restart( + "Mostrix encountered an internal error (poisoned chat router lock). Please restart the app." + .to_string(), + ); + Err("CHAT_ROUTER_CMD_TX mutex poisoned") + } + } +} + +fn send_chat_router_cmd(cmd: ChatRouterCmd) { + if let Ok(guard) = CHAT_ROUTER_CMD_TX.lock() { + if let Some(tx) = guard.as_ref() { + let _ = tx.send(cmd); + } + } +} + +/// Track a user P2P order chat by its shared key. +pub fn track_order_chat( + order_id: String, + shared_key_hex: String, + local_trade_pubkey: PublicKey, + since: Option, +) { + send_chat_router_cmd(ChatRouterCmd::TrackChatKey { + key_id: ChatKeyId::Order(order_id), + shared_key_hex, + local_trade_pubkey: Some(local_trade_pubkey), + since, + }); +} + +/// Stop tracking a user P2P order chat (order row removed / terminal cancel). +pub fn untrack_order_chat(order_id: String) { + send_chat_router_cmd(ChatRouterCmd::UntrackChatKey { + key_id: ChatKeyId::Order(order_id), + }); +} + +/// Track an admin dispute chat party by its shared key. +pub fn track_dispute_chat( + dispute_id: String, + party: ChatParty, + shared_key_hex: String, + since: Option, +) { + send_chat_router_cmd(ChatRouterCmd::TrackChatKey { + key_id: ChatKeyId::Dispute(dispute_id, party), + shared_key_hex, + local_trade_pubkey: None, + since, + }); +} + +/// Stop tracking an admin dispute chat party (dispute finalized / no longer InProgress). +pub fn untrack_dispute_chat(dispute_id: String, party: ChatParty) { + send_chat_router_cmd(ChatRouterCmd::UntrackChatKey { + key_id: ChatKeyId::Dispute(dispute_id, party), + }); +} + +/// Track a user P2P order chat once its shared key is resolvable (DM router hook). +/// +/// Loads the order, resolves the shared key (persisted `order_chat_shared_key_hex`, else ECDH +/// from `trade_keys` + `counterparty_pubkey`), and emits a track command. Idempotent at the +/// router level: re-tracking an already-tracked key is a cheap no-op, so this is safe to call +/// on every trade DM. +pub async fn maybe_track_order_chat(pool: &sqlx::SqlitePool, order_id: Uuid, trade_keys: &Keys) { + let order = match Order::get_by_id(pool, &order_id.to_string()).await { + Ok(o) => o, + Err(_) => return, + }; + let shared_hex = order + .order_chat_shared_key_hex + .clone() + .or_else(|| derive_shared_key_hex(Some(trade_keys), order.counterparty_pubkey.as_deref())); + if let Some(hex) = shared_hex { + track_order_chat(order_id.to_string(), hex, trade_keys.public_key(), None); + } +} + +/// Emit decrypted chat messages on the appropriate update channel. +/// +/// Reuses the same `AdminChatUpdate` / `OrderChatUpdate` shapes as the old +/// polling path so `apply_admin_chat_updates` / `apply_user_order_chat_updates` +/// (which dedupe by timestamp/last-seen) are unchanged. +fn emit_messages( + target: &ChatTarget, + messages: Vec<(String, i64, PublicKey)>, + admin_tx: &UnboundedSender, anyhow::Error>>, + user_tx: &UnboundedSender, anyhow::Error>>, +) { + if messages.is_empty() { + return; + } + match &target.key_id { + ChatKeyId::Order(order_id) => { + let Some(local_trade_pubkey) = target.local_trade_pubkey else { + log::warn!("Order chat {order_id} missing local trade pubkey; skipping chat emit"); + return; + }; + let _ = user_tx.send(Ok(vec![OrderChatUpdate { + order_id: order_id.clone(), + local_trade_pubkey, + messages, + }])); + } + ChatKeyId::Dispute(dispute_id, party) => { + let _ = admin_tx.send(Ok(vec![AdminChatUpdate { + dispute_id: dispute_id.clone(), + party: *party, + messages, + }])); + } + } +} + +/// Rebuild the single batched live subscription from the current tracked set. +/// +/// Unsubscribes the previous subscription and, if any keys remain, subscribes to +/// `kind: 1059` gift wraps addressed to all tracked shared pubkeys. Uses +/// `.limit(0)` (live-only, same as the DM listener's `LiveOnly` mode); startup +/// history is hydrated separately per key on [`ChatRouterCmd::TrackChatKey`]. +async fn resubscribe( + client: &Client, + targets: &HashMap, + current_sub: &mut Option, +) { + if let Some(id) = current_sub.take() { + client.unsubscribe(&id).await; + } + if targets.is_empty() { + return; + } + let pubkeys: Vec = 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}"), + } +} + +/// Single background router for all shared-key chats (user order + admin dispute). +/// +/// Spawned once at startup and respawned on client reload/reconnect (mirrors +/// `listen_for_order_messages`). Consumes [`ChatRouterCmd`] for track/untrack and +/// routes live `kind: 1059` gift wraps by `p` tag to the owning chat. +pub async fn listen_for_chat_messages( + client: Client, + admin_chat_updates_tx: UnboundedSender, anyhow::Error>>, + user_order_chat_updates_tx: UnboundedSender, anyhow::Error>>, + mut cmd_rx: mpsc::UnboundedReceiver, +) { + // Create the notification receiver BEFORE subscribing so no live event is missed. + let mut notifications = client.notifications(); + let mut targets: HashMap = HashMap::new(); + let mut current_sub: Option = None; + + loop { + tokio::select! { + cmd = cmd_rx.recv() => { + let Some(cmd) = cmd else { + // Sender dropped (respawn/shutdown): unsubscribe and exit. + if let Some(id) = current_sub.take() { + client.unsubscribe(&id).await; + } + break; + }; + match cmd { + ChatRouterCmd::TrackChatKey { key_id, shared_key_hex, local_trade_pubkey, since } => { + let Some(shared_keys) = keys_from_shared_hex(&shared_key_hex) else { + log::warn!("[chat_live] invalid shared key hex for {key_id:?}; not tracking"); + continue; + }; + let target_pubkey = shared_keys.public_key(); + // Idempotent: skip redundant history fetch + resubscribe if already tracked. + if targets.get(&target_pubkey).is_some_and(|t| t.key_id == key_id) { + continue; + } + let target = ChatTarget { + key_id: key_id.clone(), + shared_keys: shared_keys.clone(), + local_trade_pubkey, + }; + + // 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; + } + ChatRouterCmd::UntrackChatKey { key_id } => { + let before = targets.len(); + targets.retain(|_, t| t.key_id != key_id); + if targets.len() != before { + resubscribe(&client, &targets, &mut current_sub).await; + } + } + } + } + notification = notifications.recv() => { + let event = match notification { + Ok(RelayPoolNotification::Event { event, .. }) => *event, + Ok(_) => continue, + // Lagged/closed broadcast: log and keep going (not fatal), like the order/dispute loops. + Err(e) => { + log::debug!("[chat_live] notification channel: {e}"); + continue; + } + }; + if event.kind != Kind::GiftWrap { + continue; + } + // Gift wrap recipient is the shared-key pubkey in the `p` tag. + let Some(target_pubkey) = event.tags.public_keys().next().copied() else { + continue; + }; + let Some(target) = targets.get(&target_pubkey) else { + continue; + }; + match unwrap_giftwrap_with_shared_key(&target.shared_keys, &event).await { + Ok((content, ts, sender)) => { + emit_messages( + target, + vec![(content, ts, sender)], + &admin_chat_updates_tx, + &user_order_chat_updates_tx, + ); + } + Err(e) => log::warn!( + "[chat_live] failed to unwrap chat gift wrap {}: {e}", + event.id + ), + } + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// `ChatKeyId` equality backs both the untrack retention filter (`t.key_id != key_id`) + /// and the track idempotency check, so it must distinguish order vs dispute and party. + #[test] + fn chat_key_id_equality_distinguishes_targets() { + assert_eq!( + ChatKeyId::Order("a".to_string()), + ChatKeyId::Order("a".to_string()) + ); + assert_ne!( + ChatKeyId::Order("a".to_string()), + ChatKeyId::Order("b".to_string()) + ); + assert_eq!( + ChatKeyId::Dispute("d".to_string(), ChatParty::Buyer), + ChatKeyId::Dispute("d".to_string(), ChatParty::Buyer) + ); + assert_ne!( + ChatKeyId::Dispute("d".to_string(), ChatParty::Buyer), + ChatKeyId::Dispute("d".to_string(), ChatParty::Seller) + ); + assert_ne!( + ChatKeyId::Order("d".to_string()), + ChatKeyId::Dispute("d".to_string(), ChatParty::Buyer) + ); + } +} diff --git a/src/util/dm_utils/mod.rs b/src/util/dm_utils/mod.rs index 196b7fb..6e5cd37 100644 --- a/src/util/dm_utils/mod.rs +++ b/src/util/dm_utils/mod.rs @@ -29,6 +29,7 @@ use uuid::Uuid; use crate::models::{Order, User}; use crate::ui::order_message_to_notification; use crate::ui::{MessageNotification, OrderMessage}; +use crate::util::chat_listener::{maybe_track_order_chat, untrack_order_chat}; use crate::util::db_utils::{delete_order_by_id, save_order, update_order_status}; use crate::util::filters::filter_protocol_dm_from_mostro; use crate::util::mostro_info::{ @@ -413,6 +414,8 @@ async fn drop_pre_active_taker_take( ); } remove_order_from_messages(messages, order_id); + // Row deleted: stop the P2P order chat subscription for this order. + untrack_order_chat(order_id.to_string()); } /// Refreshes the local `orders` row from embedded order data on trade DMs that carry a full @@ -526,6 +529,8 @@ async fn revert_maker_to_pending_on_book_republish( } remove_order_from_messages(messages, order_id); + // Back on the book as a pending maker listing (no counterparty): stop order chat subscription. + untrack_order_chat(order_id.to_string()); try_notify_my_trades_maker_book_changed(); log::info!( @@ -714,6 +719,9 @@ async fn handle_trade_dm_for_order( ) .await; + // Keep the P2P order chat subscription live once the shared key is resolvable (idempotent). + maybe_track_order_chat(pool, order_id, trade_keys).await; + // Extract invoice and sat_amount from payload based on action type. // For `PayBondInvoice` mostrod populates the bond satoshis in the third // `Option` field of `Payload::PaymentRequest` (the SmallOrder is @@ -1068,6 +1076,8 @@ async fn dispatch_giftwrap_batch( } if has_terminal_status { + // Order reached a terminal status: stop its P2P order chat subscription. + untrack_order_chat(order_id.to_string()); match terminal_policy { GiftWrapTerminalPolicy::TrackedSubscription(subscription_id) => { log::info!( diff --git a/src/util/mod.rs b/src/util/mod.rs index 93bf4de..fa60f21 100644 --- a/src/util/mod.rs +++ b/src/util/mod.rs @@ -1,4 +1,5 @@ pub mod blossom; +pub mod chat_listener; pub mod chat_utils; pub mod db_utils; pub mod dm_utils; @@ -18,6 +19,10 @@ pub use blossom::{ blossom_url_to_https, decrypt_blob, encrypt_blob, fetch_blob, save_attachment_to_disk, spawn_save_attachment, upload_blob_with_retry, BLOSSOM_MAX_BLOB_SIZE, DEFAULT_BLOSSOM_SERVERS, }; +pub use chat_listener::{ + listen_for_chat_messages, set_chat_router_cmd_tx, track_dispute_chat, track_order_chat, + untrack_dispute_chat, untrack_order_chat, ChatKeyId, ChatRouterCmd, +}; pub use chat_utils::send_admin_chat_message_via_shared_key; pub use db_utils::save_order; pub use dm_utils::{ diff --git a/src/util/order_utils/execute_take_dispute.rs b/src/util/order_utils/execute_take_dispute.rs index 4ef7172..36e43d5 100644 --- a/src/util/order_utils/execute_take_dispute.rs +++ b/src/util/order_utils/execute_take_dispute.rs @@ -6,6 +6,9 @@ use sqlx::SqlitePool; use uuid::Uuid; use crate::models::AdminDispute; +use crate::ui::ChatParty; +use crate::util::chat_listener::track_dispute_chat; +use crate::util::chat_utils::derive_shared_key_hex; use crate::util::dm_utils::{parse_dm_events, send_dm, wait_for_dm, FETCH_EVENTS_TIMEOUT}; use crate::util::mostro_info::MostroInstanceInfo; use crate::util::order_utils::helper::fetch_order_fiat_from_relay; @@ -137,6 +140,17 @@ pub async fn execute_take_dispute( "✅ Dispute {} taken successfully and saved to database with InProgress status!", dispute_info.id ); + + // Start live shared-key chat subscriptions for both parties of this dispute. + for (party, cp_pubkey) in [ + (ChatParty::Buyer, dispute_info.buyer_pubkey.as_deref()), + (ChatParty::Seller, dispute_info.seller_pubkey.as_deref()), + ] { + if let Some(hex) = derive_shared_key_hex(Some(admin_keys), cp_pubkey) { + track_dispute_chat(dispute_id.to_string(), party, hex, None); + } + } + Ok(()) } else { Err(anyhow::anyhow!( diff --git a/src/util/order_utils/fetch_scheduler.rs b/src/util/order_utils/fetch_scheduler.rs index 8234421..7194758 100644 --- a/src/util/order_utils/fetch_scheduler.rs +++ b/src/util/order_utils/fetch_scheduler.rs @@ -1,20 +1,12 @@ -use std::collections::HashMap; -use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex}; use mostro_core::prelude::*; use nostr_sdk::prelude::*; -use tokio::sync::mpsc::UnboundedSender; use tokio::task::JoinHandle; use tokio::time::{interval_at, Duration, Instant}; -use crate::models::AdminDispute; use crate::settings::Settings; -use crate::ui::{ - AdminChatLastSeen, AdminChatUpdate, ChatParty, OrderChatLastSeen, OrderChatUpdate, -}; use crate::util::catch_unwind_request_fatal_restart; -use crate::util::chat_utils::{fetch_admin_chat_updates, fetch_user_order_chat_updates}; use sqlx::SqlitePool; use super::get_disputes; @@ -39,7 +31,6 @@ pub struct FetchSchedulerResult { } // Semaphore to prevent multiple chat messages from being processed at the same time -pub static CHAT_MESSAGES_SEMAPHORE: AtomicBool = AtomicBool::new(false); const RECONCILIATION_INTERVAL_SECS: u64 = 30; fn apply_live_order_update(orders: &Arc>>, order: SmallOrder) { @@ -418,50 +409,3 @@ pub fn spawn_fetch_scheduler_loops( (order_task, dispute_task) } - -/// Spawns a one-off background task to fetch admin chat updates and send the result on the given channel. -pub fn spawn_admin_chat_fetch( - client: Client, - disputes: Vec, - admin_chat_last_seen: HashMap<(String, ChatParty), AdminChatLastSeen>, - tx: UnboundedSender, anyhow::Error>>, -) { - // If the semaphore is already true, return - if CHAT_MESSAGES_SEMAPHORE - .compare_exchange(false, true, Ordering::Relaxed, Ordering::Relaxed) - .is_err() - { - return; - } - tokio::spawn(async move { - catch_unwind_request_fatal_restart("admin chat fetch", async move { - let result = fetch_admin_chat_updates(&client, &disputes, &admin_chat_last_seen).await; - CHAT_MESSAGES_SEMAPHORE.store(false, Ordering::Relaxed); - let _ = tx.send(result); - }) - .await; - }); -} - -/// Spawns a one-off background task to fetch user order chat updates. -pub fn spawn_user_order_chat_fetch( - client: Client, - pool: sqlx::SqlitePool, - order_chat_last_seen: HashMap, - tx: UnboundedSender, anyhow::Error>>, -) { - if CHAT_MESSAGES_SEMAPHORE - .compare_exchange(false, true, Ordering::Relaxed, Ordering::Relaxed) - .is_err() - { - return; - } - tokio::spawn(async move { - catch_unwind_request_fatal_restart("user order chat fetch", async move { - let result = fetch_user_order_chat_updates(&client, &pool, &order_chat_last_seen).await; - CHAT_MESSAGES_SEMAPHORE.store(false, Ordering::Relaxed); - let _ = tx.send(result); - }) - .await; - }); -} diff --git a/src/util/order_utils/mod.rs b/src/util/order_utils/mod.rs index 14abfa7..6696510 100644 --- a/src/util/order_utils/mod.rs +++ b/src/util/order_utils/mod.rs @@ -23,8 +23,7 @@ pub use execute_finalize_dispute::execute_finalize_dispute; pub use execute_send_msg::{execute_rate_user, execute_send_msg}; pub use execute_take_dispute::execute_take_dispute; pub use fetch_scheduler::{ - spawn_admin_chat_fetch, spawn_fetch_scheduler_loops, spawn_user_order_chat_fetch, - start_fetch_scheduler, FetchSchedulerResult, + spawn_fetch_scheduler_loops, start_fetch_scheduler, FetchSchedulerResult, }; pub use helper::{ aggregate_latest_orders_by_id, dispute_from_tags, fetch_events_list, fetch_mostro_order_events, From 0d3544f4c4d32092f02fce84d11b15c4b685ed3b Mon Sep 17 00:00:00 2001 From: arkanoider Date: Tue, 7 Jul 2026 21:49:55 +0200 Subject: [PATCH 2/3] fix: untrack dispute chat subscriptions on finalize Co-authored-by: Cursor --- src/util/chat_listener.rs | 6 ++++++ src/util/mod.rs | 3 ++- src/util/order_utils/execute_finalize_dispute.rs | 4 ++++ 3 files changed, 12 insertions(+), 1 deletion(-) diff --git a/src/util/chat_listener.rs b/src/util/chat_listener.rs index 70d40b2..ed709e2 100644 --- a/src/util/chat_listener.rs +++ b/src/util/chat_listener.rs @@ -139,6 +139,12 @@ pub fn untrack_dispute_chat(dispute_id: String, party: ChatParty) { }); } +/// Stop live shared-key chat subscriptions for both buyer and seller parties. +pub fn untrack_dispute_chat_parties(dispute_id: &str) { + untrack_dispute_chat(dispute_id.to_string(), ChatParty::Buyer); + untrack_dispute_chat(dispute_id.to_string(), ChatParty::Seller); +} + /// Track a user P2P order chat once its shared key is resolvable (DM router hook). /// /// Loads the order, resolves the shared key (persisted `order_chat_shared_key_hex`, else ECDH diff --git a/src/util/mod.rs b/src/util/mod.rs index fa60f21..61a630f 100644 --- a/src/util/mod.rs +++ b/src/util/mod.rs @@ -21,7 +21,8 @@ pub use blossom::{ }; pub use chat_listener::{ listen_for_chat_messages, set_chat_router_cmd_tx, track_dispute_chat, track_order_chat, - untrack_dispute_chat, untrack_order_chat, ChatKeyId, ChatRouterCmd, + untrack_dispute_chat, untrack_dispute_chat_parties, untrack_order_chat, ChatKeyId, + ChatRouterCmd, }; pub use chat_utils::send_admin_chat_message_via_shared_key; pub use db_utils::save_order; diff --git a/src/util/order_utils/execute_finalize_dispute.rs b/src/util/order_utils/execute_finalize_dispute.rs index 43c86b4..eecc2b4 100644 --- a/src/util/order_utils/execute_finalize_dispute.rs +++ b/src/util/order_utils/execute_finalize_dispute.rs @@ -5,6 +5,7 @@ use sqlx::SqlitePool; use uuid::Uuid; use crate::models::AdminDispute; +use crate::util::chat_listener::untrack_dispute_chat_parties; use crate::util::mostro_info::MostroInstanceInfo; use super::{execute_admin_cancel, execute_admin_settle, BondSlashChoice}; @@ -122,6 +123,9 @@ pub async fn execute_finalize_dispute( AdminDispute::set_status_seller_refunded(pool, &dispute.id).await?; } + // Dispute left InProgress: drop buyer/seller shared-key chat subscriptions. + untrack_dispute_chat_parties(&dispute_id_str); + let action_name = if is_settle { "settled (buyer paid)" } else { From 737e98cbe35f01e7510a58561b3d1d3fd741b3db Mon Sep 17 00:00:00 2001 From: arkanoider Date: Tue, 7 Jul 2026 21:55:56 +0200 Subject: [PATCH 3/3] fix: keep order chat subscribed through success status Co-authored-by: Cursor --- docs/MESSAGE_FLOW_AND_PROTOCOL.md | 2 +- src/util/dm_utils/mod.rs | 74 ++++++++++++++++++++++++++++++- 2 files changed, 73 insertions(+), 3 deletions(-) diff --git a/docs/MESSAGE_FLOW_AND_PROTOCOL.md b/docs/MESSAGE_FLOW_AND_PROTOCOL.md index 863be71..c42b2d5 100644 --- a/docs/MESSAGE_FLOW_AND_PROTOCOL.md +++ b/docs/MESSAGE_FLOW_AND_PROTOCOL.md @@ -321,7 +321,7 @@ In addition to relay-driven trade DMs, Mostrix keeps a lightweight local transcr - **Path**: `~/.mostrix/orders_chat/.txt` - **Startup restore**: `load_user_order_chats_at_startup` restores cached chat into `AppState.order_chats`, seeds `order_chat_last_seen`, then runs an initial `fetch_user_order_chat_updates` relay backfill. -- **Live relay sync (User role)**: the **shared-key chat subscription router** (`listen_for_chat_messages` in `src/util/chat_listener.rs`) maintains one batched `kind: 1059` subscription over all active order shared pubkeys and routes incoming gift wraps by `p` tag. `track_startup_chats` seeds the active-order set at startup; the DM router tracks/untracks orders as their shared key becomes resolvable or they reach a terminal status. Shared keys come from persisted `order_chat_shared_key_hex` when set, otherwise ECDH from local `trade_keys` + `counterparty_pubkey` (`src/util/chat_utils.rs`). +- **Live relay sync (User role)**: the **shared-key chat subscription router** (`listen_for_chat_messages` in `src/util/chat_listener.rs`) maintains one batched `kind: 1059` subscription over all active order shared pubkeys and routes incoming gift wraps by `p` tag. `track_startup_chats` seeds the active-order set at startup; the DM router tracks/untracks orders when the shared key becomes resolvable or the order hits a chat-terminal status ([`TERMINAL_DM_STATUSES`](../src/models.rs) — **`success` keeps chat live**). Shared keys come from persisted `order_chat_shared_key_hex` when set, otherwise ECDH from local `trade_keys` + `counterparty_pubkey` (`src/util/chat_utils.rs`). - **Incremental merge**: `apply_user_order_chat_updates` in `src/ui/helpers/startup.rs`: - **Skip own relay echoes**: each `OrderChatUpdate` carries `local_trade_pubkey`; messages whose decrypted `sender_pubkey` matches are ignored (same rule as admin chat and Mostro Mobile — avoids showing your send on both **You** and **Peer** after the optimistic local append on Enter). - **Dedup**: relay self-echoes are skipped by `sender_pubkey == local_trade_pubkey`; peer dedup matches only existing **Peer** rows at the same `(timestamp, content)` (or same attachment / legacy placeholder) so an optimistic **You** line cannot hide a real counterparty message in the same second. diff --git a/src/util/dm_utils/mod.rs b/src/util/dm_utils/mod.rs index 6e5cd37..71d8c2f 100644 --- a/src/util/dm_utils/mod.rs +++ b/src/util/dm_utils/mod.rs @@ -156,6 +156,40 @@ fn is_terminal_order_status(status: Status) -> bool { ) } +/// P2P chat untrack set: aligned with [`crate::models::TERMINAL_DM_STATUSES`] (omits `success`). +fn order_status_untracks_chat(status: Status) -> bool { + matches!( + status, + Status::Canceled + | Status::CanceledByAdmin + | Status::SettledByAdmin + | Status::CompletedByAdmin + | Status::Expired + | Status::CooperativelyCanceled + ) +} + +/// Whether a trade DM should drop the shared-key order chat subscription. +/// +/// [`trade_message_is_terminal`] still includes [`Status::Success`] so the trade DM listener can +/// tear down its per-trade-key subscription; chat stays tracked through the post-success window. +fn trade_message_should_untrack_order_chat(message: &Message) -> bool { + let kind = message.get_inner_message_kind(); + if matches!( + &kind.action, + Action::AdminCanceled | Action::Canceled | Action::CooperativeCancelAccepted + ) { + return true; + } + kind.payload + .as_ref() + .and_then(|payload| match payload { + Payload::Order(order) => order.status, + _ => None, + }) + .is_some_and(order_status_untracks_chat) +} + /// Loads active-order rows for DM bootstrap; status filter is [`crate::models::TERMINAL_DM_STATUSES`]. pub async fn hydrate_startup_active_order_dm_state( pool: &sqlx::sqlite::SqlitePool, @@ -1037,6 +1071,7 @@ async fn dispatch_giftwrap_batch( for (message, timestamp, sender) in parsed_messages { let has_terminal_status = trade_message_is_terminal(&message); + let should_untrack_chat = trade_message_should_untrack_order_chat(&message); log::info!( "order id: {} has_terminal_status: {:?}", order_id, @@ -1075,9 +1110,11 @@ async fn dispatch_giftwrap_batch( ); } - if has_terminal_status { - // Order reached a terminal status: stop its P2P order chat subscription. + if should_untrack_chat { untrack_order_chat(order_id.to_string()); + } + + if has_terminal_status { match terminal_policy { GiftWrapTerminalPolicy::TrackedSubscription(subscription_id) => { log::info!( @@ -1859,6 +1896,7 @@ mod tests { default_dm_expiration, effective_is_mine_for_trade_dm_message, is_pre_active_maker_listing, is_pre_active_taker_take, new_order_would_regress_messages_row, small_order_pending_from_new_order_payload, trade_message_is_terminal, + trade_message_should_untrack_order_chat, }; use crate::models::Order; use mostro_core::prelude::{Action, Message, Payload, SmallOrder, Status}; @@ -1868,6 +1906,38 @@ mod tests { fn action_only_canceled_is_terminal() { let message = Message::new_order(None, None, None, Action::Canceled, None); assert!(trade_message_is_terminal(&message)); + assert!(trade_message_should_untrack_order_chat(&message)); + } + + #[test] + fn success_order_payload_is_terminal_for_dm_but_keeps_chat_tracked() { + let message = Message::new_order( + None, + None, + None, + Action::Released, + Some(Payload::Order(SmallOrder { + status: Some(Status::Success), + ..Default::default() + })), + ); + assert!(trade_message_is_terminal(&message)); + assert!(!trade_message_should_untrack_order_chat(&message)); + } + + #[test] + fn canceled_order_payload_untracks_chat() { + let message = Message::new_order( + None, + None, + None, + Action::Canceled, + Some(Payload::Order(SmallOrder { + status: Some(Status::Canceled), + ..Default::default() + })), + ); + assert!(trade_message_should_untrack_order_chat(&message)); } #[test]