From b0047cc5696fe83b9291685cb29da326e48cb90c Mon Sep 17 00:00:00 2001 From: jry Date: Fri, 30 Oct 2026 07:30:06 +0800 Subject: [PATCH 1/2] Persist and sync the message send-later queue across restarts and clients The conversation send-later queue was held purely in React state (useMessageQueue) with no persistence and no cross-client sync: a restart or a second window/web client started from an empty queue. - Add a message_queue_item SQLite table (SeaORM entity + migration), keyed by owner_id (the conversation id) with a client-generated item_uid per item. The draft is stored as JSON since PromptDraft is a frontend-only type. - Add message_queue_service (list / full-replace save in one transaction) shared by both run modes, a Tauri command and an axum web handler (/message_queue_list, /message_queue_save). - Broadcast message-queue://changed (owner_id, origin, items) on every save via the existing EventEmitter, so every Tauri window and web client converges to the same queue. Each client ignores its own echo. - The hook persists + subscribes only when an ownerId is present; unbound draft tabs stay in-memory and behave exactly as before, so the existing no-owner unit tests keep passing. Closes the queue-lost-on-restart / not-shared-across-clients gap. --- src-tauri/src/commands/message_queue.rs | 42 ++++ src-tauri/src/commands/mod.rs | 1 + .../src/db/entities/message_queue_item.rs | 28 +++ src-tauri/src/db/entities/mod.rs | 1 + src-tauri/src/db/entities/prelude.rs | 1 + .../m20261030_000001_message_queue.rs | 106 ++++++++++ src-tauri/src/db/migration/mod.rs | 2 + .../src/db/service/message_queue_service.rs | 81 ++++++++ src-tauri/src/db/service/mod.rs | 1 + src-tauri/src/lib.rs | 3 + src-tauri/src/models/message_queue.rs | 14 ++ src-tauri/src/models/mod.rs | 2 + src-tauri/src/web/event_bridge.rs | 20 ++ src-tauri/src/web/handlers/message_queue.rs | 53 +++++ src-tauri/src/web/handlers/mod.rs | 1 + src-tauri/src/web/router.rs | 9 + .../conversation-detail-panel.tsx | 4 +- src/hooks/use-message-queue.ts | 182 ++++++++++++++++-- src/lib/api.ts | 16 ++ src/lib/types.ts | 26 +++ 20 files changed, 572 insertions(+), 21 deletions(-) create mode 100644 src-tauri/src/commands/message_queue.rs create mode 100644 src-tauri/src/db/entities/message_queue_item.rs create mode 100644 src-tauri/src/db/migration/m20261030_000001_message_queue.rs create mode 100644 src-tauri/src/db/service/message_queue_service.rs create mode 100644 src-tauri/src/models/message_queue.rs create mode 100644 src-tauri/src/web/handlers/message_queue.rs diff --git a/src-tauri/src/commands/message_queue.rs b/src-tauri/src/commands/message_queue.rs new file mode 100644 index 0000000000..89faaa6b2c --- /dev/null +++ b/src-tauri/src/commands/message_queue.rs @@ -0,0 +1,42 @@ +#[cfg(feature = "tauri-runtime")] +use crate::db::AppDatabase; +#[cfg(feature = "tauri-runtime")] +use crate::db::error::DbError; +#[cfg(feature = "tauri-runtime")] +use crate::db::service::message_queue_service::{self, QueueItemInput}; +#[cfg(feature = "tauri-runtime")] +use crate::models::MessageQueueItem; +#[cfg(feature = "tauri-runtime")] +use crate::web::event_bridge::{emit_event, EventEmitter, MessageQueueChange, MESSAGE_QUEUE_CHANGED_EVENT}; + +#[cfg(feature = "tauri-runtime")] +#[cfg_attr(feature = "tauri-runtime", tauri::command)] +pub async fn message_queue_list( + db: tauri::State<'_, AppDatabase>, + owner_id: String, +) -> Result, DbError> { + message_queue_service::list(&db.conn, &owner_id).await +} + +#[cfg(feature = "tauri-runtime")] +#[cfg_attr(feature = "tauri-runtime", tauri::command)] +pub async fn message_queue_save( + db: tauri::State<'_, AppDatabase>, + app: tauri::AppHandle, + owner_id: String, + items: Vec, + origin: String, +) -> Result<(), DbError> { + let saved = message_queue_service::save(&db.conn, &owner_id, items).await?; + let emitter = EventEmitter::Tauri(app); + emit_event( + &emitter, + MESSAGE_QUEUE_CHANGED_EVENT, + MessageQueueChange { + owner_id, + origin, + items: saved, + }, + ); + Ok(()) +} diff --git a/src-tauri/src/commands/mod.rs b/src-tauri/src/commands/mod.rs index c3bdf78399..158091b876 100644 --- a/src-tauri/src/commands/mod.rs +++ b/src-tauri/src/commands/mod.rs @@ -68,6 +68,7 @@ pub mod notification; pub mod pet; pub mod project_boot; pub mod question; +pub mod message_queue; pub mod quick_messages; #[cfg(feature = "tauri-runtime")] pub mod remote_proxy; diff --git a/src-tauri/src/db/entities/message_queue_item.rs b/src-tauri/src/db/entities/message_queue_item.rs new file mode 100644 index 0000000000..a6022c4647 --- /dev/null +++ b/src-tauri/src/db/entities/message_queue_item.rs @@ -0,0 +1,28 @@ +use sea_orm::entity::prelude::*; + +/// One row per queued message in a conversation's "send later" queue. +/// +/// The queue is keyed by `owner_id` (a conversation id). `item_uid` is a +/// client-generated UUID that is stable across edits, so reordering and updates +/// can target the same logical item without a server round-trip for ids. The +/// `draft` is stored as JSON because `PromptDraft` is a frontend-only type. +#[derive(Clone, Debug, PartialEq, DeriveEntityModel)] +#[sea_orm(table_name = "message_queue_item")] +pub struct Model { + #[sea_orm(primary_key)] + pub id: i32, + pub owner_id: String, + pub item_uid: String, + #[sea_orm(column_type = "Text")] + pub draft_json: String, + pub mode_id: Option, + pub adopt_send_time_mode: bool, + pub sort_order: i32, + pub created_at: DateTimeUtc, + pub updated_at: DateTimeUtc, +} + +#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)] +pub enum Relation {} + +impl ActiveModelBehavior for ActiveModel {} diff --git a/src-tauri/src/db/entities/mod.rs b/src-tauri/src/db/entities/mod.rs index 34f2861ade..f213271d04 100644 --- a/src-tauri/src/db/entities/mod.rs +++ b/src-tauri/src/db/entities/mod.rs @@ -20,6 +20,7 @@ pub mod folder_link; pub mod model_provider; pub mod opened_tab; pub mod prelude; +pub mod message_queue_item; pub mod quick_message; pub mod remote_workspace_connection; pub mod token_usage_sync; diff --git a/src-tauri/src/db/entities/prelude.rs b/src-tauri/src/db/entities/prelude.rs index 3f6f338838..cda1c950f7 100644 --- a/src-tauri/src/db/entities/prelude.rs +++ b/src-tauri/src/db/entities/prelude.rs @@ -21,6 +21,7 @@ pub use super::folder_group::Entity as FolderGroup; pub use super::folder_link::Entity as FolderLink; pub use super::model_provider::Entity as ModelProvider; pub use super::opened_tab::Entity as OpenedTab; +pub use super::message_queue_item::Entity as MessageQueueItem; pub use super::quick_message::Entity as QuickMessage; pub use super::token_usage_sync::Entity as TokenUsageSync; pub use super::token_usage_turn::Entity as TokenUsageTurn; diff --git a/src-tauri/src/db/migration/m20261030_000001_message_queue.rs b/src-tauri/src/db/migration/m20261030_000001_message_queue.rs new file mode 100644 index 0000000000..a9e761fd04 --- /dev/null +++ b/src-tauri/src/db/migration/m20261030_000001_message_queue.rs @@ -0,0 +1,106 @@ +use sea_orm_migration::prelude::*; + +#[derive(DeriveMigrationName)] +pub struct Migration; + +#[async_trait::async_trait] +impl MigrationTrait for Migration { + async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> { + // message_queue_item: the per-conversation "send later" queue, persisted + // so it survives an app restart and — via the `message-queue://changed` + // broadcast — stays in sync across every open window / web client. + // + // One row per queued message. `item_uid` is a client-generated UUID that + // is stable across edits; `owner_id` is the conversation the queue belongs + // to. The `draft` is JSON because `PromptDraft` is a frontend-only type. + manager + .create_table( + Table::create() + .table(MessageQueueItem::Table) + .if_not_exists() + .col( + ColumnDef::new(MessageQueueItem::Id) + .integer() + .not_null() + .auto_increment() + .primary_key(), + ) + .col(ColumnDef::new(MessageQueueItem::OwnerId).string().not_null()) + .col(ColumnDef::new(MessageQueueItem::ItemUid).string().not_null()) + .col(ColumnDef::new(MessageQueueItem::DraftJson).text().not_null()) + .col(ColumnDef::new(MessageQueueItem::ModeId).string().null()) + .col( + ColumnDef::new(MessageQueueItem::AdoptSendTimeMode) + .boolean() + .not_null() + .default(false), + ) + .col( + ColumnDef::new(MessageQueueItem::SortOrder) + .integer() + .not_null() + .default(0), + ) + .col( + ColumnDef::new(MessageQueueItem::CreatedAt) + .timestamp_with_time_zone() + .not_null(), + ) + .col( + ColumnDef::new(MessageQueueItem::UpdatedAt) + .timestamp_with_time_zone() + .not_null(), + ) + .to_owned(), + ) + .await?; + + // "All items for a conversation", read on every queue load. + manager + .create_index( + Index::create() + .if_not_exists() + .name("idx_message_queue_item_owner") + .table(MessageQueueItem::Table) + .col(MessageQueueItem::OwnerId) + .to_owned(), + ) + .await?; + + // The upsert / delete conflict target. + manager + .create_index( + Index::create() + .if_not_exists() + .name("idx_message_queue_item_owner_uid") + .table(MessageQueueItem::Table) + .col(MessageQueueItem::OwnerId) + .col(MessageQueueItem::ItemUid) + .unique() + .to_owned(), + ) + .await?; + + Ok(()) + } + + async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> { + manager + .drop_table(Table::drop().table(MessageQueueItem::Table).to_owned()) + .await + } +} + +#[derive(DeriveIden)] +enum MessageQueueItem { + Table, + Id, + OwnerId, + ItemUid, + DraftJson, + ModeId, + AdoptSendTimeMode, + SortOrder, + CreatedAt, + UpdatedAt, +} diff --git a/src-tauri/src/db/migration/mod.rs b/src-tauri/src/db/migration/mod.rs index 4bae1e013f..3717667d38 100644 --- a/src-tauri/src/db/migration/mod.rs +++ b/src-tauri/src/db/migration/mod.rs @@ -56,6 +56,7 @@ mod m20261017_000002_conversation_source; mod m20261020_000001_folder_command_run_local; mod m20261022_000001_tool_approval; mod m20261022_000001_work_task_delivery; +mod m20261030_000001_message_queue; pub struct Migrator; #[async_trait::async_trait] @@ -118,6 +119,7 @@ impl MigratorTrait for Migrator { Box::new(m20261020_000001_folder_command_run_local::Migration), Box::new(m20261022_000001_tool_approval::Migration), Box::new(m20261022_000001_work_task_delivery::Migration), + Box::new(m20261030_000001_message_queue::Migration), ] } } diff --git a/src-tauri/src/db/service/message_queue_service.rs b/src-tauri/src/db/service/message_queue_service.rs new file mode 100644 index 0000000000..c7ef53d353 --- /dev/null +++ b/src-tauri/src/db/service/message_queue_service.rs @@ -0,0 +1,81 @@ +use chrono::Utc; +use sea_orm::{ + ActiveModelTrait, ActiveValue::NotSet, ColumnTrait, ConnectionTrait, DatabaseConnection, + EntityTrait, IntoActiveModel, QueryFilter, QueryOrder, Set, TransactionTrait, +}; +use serde::Deserialize; + +use crate::db::entities::message_queue_item; +use crate::db::error::DbError; +use crate::models::MessageQueueItem; + +/// A queue item as supplied by the frontend when saving. Field names are +/// camelCase on the wire (both the Tauri command and the axum web handler +/// expect it), so serde rename handles the mapping to the Rust fields. +#[derive(Debug, Clone, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct QueueItemInput { + pub id: String, + pub draft_json: String, + pub mode_id: Option, + pub adopt_send_time_mode: bool, +} + +fn to_item(m: message_queue_item::Model) -> MessageQueueItem { + MessageQueueItem { + id: m.item_uid, + draft_json: m.draft_json, + mode_id: m.mode_id, + adopt_send_time_mode: m.adopt_send_time_mode, + } +} + +pub async fn list( + conn: &DatabaseConnection, + owner_id: &str, +) -> Result, DbError> { + let rows = message_queue_item::Entity::find() + .filter(message_queue_item::Column::OwnerId.eq(owner_id)) + .order_by_asc(message_queue_item::Column::SortOrder) + .all(conn) + .await?; + Ok(rows.into_iter().map(to_item).collect()) +} + +/// Replace the entire queue for `owner_id` with `items` (in order). Simple and +/// idempotent: last-writer-wins, and the broadcast converges every client to the +/// same list. Multi-end sync is the goal, so the frontend is the source of +/// truth for ordering and the backend just persists the canonical list. The +/// whole replace runs in one transaction so a concurrent reader never sees a +/// half-written queue. +pub async fn save( + conn: &DatabaseConnection, + owner_id: &str, + items: Vec, +) -> Result, DbError> { + let txn = conn.begin().await?; + message_queue_item::Entity::delete_many() + .filter(message_queue_item::Column::OwnerId.eq(owner_id)) + .exec(&txn) + .await?; + + let mut saved = Vec::with_capacity(items.len()); + for (idx, item) in items.into_iter().enumerate() { + let now = Utc::now(); + let active = message_queue_item::ActiveModel { + id: NotSet, + owner_id: Set(owner_id.to_string()), + item_uid: Set(item.id), + draft_json: Set(item.draft_json), + mode_id: Set(item.mode_id), + adopt_send_time_mode: Set(item.adopt_send_time_mode), + sort_order: Set(idx as i32), + created_at: Set(now), + updated_at: Set(now), + }; + let m = active.insert(&txn).await?; + saved.push(to_item(m)); + } + txn.commit().await?; + Ok(saved) +} diff --git a/src-tauri/src/db/service/mod.rs b/src-tauri/src/db/service/mod.rs index d11b8a081e..09b75ae077 100644 --- a/src-tauri/src/db/service/mod.rs +++ b/src-tauri/src/db/service/mod.rs @@ -14,6 +14,7 @@ pub mod folder_link_service; pub mod folder_service; pub mod import_service; pub mod model_provider_service; +pub mod message_queue_service; pub mod quick_message_service; pub mod remote_workspace_connection_service; pub mod sender_context_service; diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 18d2fee42f..40fa28e6f2 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -117,6 +117,7 @@ mod tauri_app { folders, logging as logging_commands, mcp as mcp_commands, model_provider as model_provider_commands, notification, pet as pet_commands, project_boot, question as question_commands, quick_messages as quick_messages_commands, + message_queue as message_queue_commands, remote_proxy as remote_proxy_commands, remote_workspace as remote_workspace_commands, science as science_commands, session_info as session_info_commands, @@ -2282,6 +2283,8 @@ mod tauri_app { quick_messages_commands::quick_messages_update, quick_messages_commands::quick_messages_delete, quick_messages_commands::quick_messages_reorder, + message_queue_commands::message_queue_list, + message_queue_commands::message_queue_save, automation_commands::automation_list, automation_commands::automation_get, automation_commands::automation_runs, diff --git a/src-tauri/src/models/message_queue.rs b/src-tauri/src/models/message_queue.rs new file mode 100644 index 0000000000..d80ddc9203 --- /dev/null +++ b/src-tauri/src/models/message_queue.rs @@ -0,0 +1,14 @@ +use serde::Serialize; + +/// A single persisted message-queue item, returned to the frontend. The `draft` +/// travels as a JSON string (`draft_json`) because `PromptDraft` is a +/// frontend-only type; the client re-parses it. Field names are snake_case to +/// match how the rest of the codebase serializes command returns (no camelCase +/// conversion happens on the wire for event/return payloads). +#[derive(Debug, Clone, Serialize)] +pub struct MessageQueueItem { + pub id: String, + pub draft_json: String, + pub mode_id: Option, + pub adopt_send_time_mode: bool, +} diff --git a/src-tauri/src/models/mod.rs b/src-tauri/src/models/mod.rs index 43960b7cd4..15e3f0dc86 100644 --- a/src-tauri/src/models/mod.rs +++ b/src-tauri/src/models/mod.rs @@ -9,6 +9,7 @@ pub mod folder; pub mod message; pub mod model_provider; pub mod pet; +pub mod message_queue; pub mod quick_message; pub mod remote_workspace_connection; pub mod system; @@ -42,6 +43,7 @@ pub use message::{ AgentExecutionStats, AgentTimelineEntry, AgentToolCall, ContentBlock, ImageData, MessageRole, MessageTurn, TurnRole, TurnUsage, UnifiedMessage, }; +pub use message_queue::MessageQueueItem; pub use quick_message::QuickMessageInfo; pub use remote_workspace_connection::{ RemoteWorkspaceConnectionInfo, RemoteWorkspaceHeader, ToHeaderMap, diff --git a/src-tauri/src/web/event_bridge.rs b/src-tauri/src/web/event_bridge.rs index 689d3a4ca0..fa4b263937 100644 --- a/src-tauri/src/web/event_bridge.rs +++ b/src-tauri/src/web/event_bridge.rs @@ -394,6 +394,13 @@ pub struct ConversationsBulkChanged { /// webview and every WebSocket client. pub const TABS_CHANGED_EVENT: &str = "tabs://changed"; +/// Global side-channel announcing a message-queue change for a given owner +/// (conversation). Mirrors [`CONVERSATION_CHANGED_EVENT`]: a single +/// [`emit_event`] reaches the Tauri webview and every WebSocket client, so the +/// "send later" queue — which is persisted in the DB and edited from any window +/// or web client — converges across all of them. Payload: [`MessageQueueChange`]. +pub const MESSAGE_QUEUE_CHANGED_EVENT: &str = "message-queue://changed"; + /// Payload for the [`TABS_CHANGED_EVENT`] side-channel. Carries the full /// conversation-bound tab set (a snapshot, not a delta) so every client /// converges idempotently — matching the full-replacement save semantics. @@ -412,6 +419,19 @@ pub struct TabsChanged { pub tabs: Vec, } +/// Payload for [`MESSAGE_QUEUE_CHANGED_EVENT`]. Carries the full queue for the +/// owner so every client replaces its local copy idempotently (last-writer-wins). +/// +/// `origin` is the client id that made the change; that client ignores its own +/// echo so a concurrent local mutation is never clobbered by the broadcast it +/// triggered. +#[derive(Debug, Clone, Serialize)] +pub struct MessageQueueChange { + pub owner_id: String, + pub origin: String, + pub items: Vec, +} + /// Global side-channel for cross-client automation list + run sync. Mirrors /// [`CONVERSATION_CHANGED_EVENT`]: a single [`emit_event`] reaches the Tauri /// webview and every WebSocket client. The scheduler runs headless (no window), diff --git a/src-tauri/src/web/handlers/message_queue.rs b/src-tauri/src/web/handlers/message_queue.rs new file mode 100644 index 0000000000..f034bb6b4e --- /dev/null +++ b/src-tauri/src/web/handlers/message_queue.rs @@ -0,0 +1,53 @@ +use std::sync::Arc; + +use axum::{extract::Extension, Json}; +use serde::Deserialize; + +use crate::app_error::AppCommandError; +use crate::app_state::AppState; +use crate::db::service::message_queue_service::{self, QueueItemInput}; +use crate::models::MessageQueueItem; +use crate::web::event_bridge::{emit_event, MessageQueueChange, MESSAGE_QUEUE_CHANGED_EVENT}; + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct ListMessageQueueParams { + pub owner_id: String, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct SaveMessageQueueParams { + pub owner_id: String, + pub items: Vec, + pub origin: String, +} + +pub async fn message_queue_list( + Extension(state): Extension>, + Json(params): Json, +) -> Result>, AppCommandError> { + let result = message_queue_service::list(&state.db.conn, ¶ms.owner_id) + .await + .map_err(AppCommandError::from)?; + Ok(Json(result)) +} + +pub async fn message_queue_save( + Extension(state): Extension>, + Json(params): Json, +) -> Result, AppCommandError> { + let saved = message_queue_service::save(&state.db.conn, ¶ms.owner_id, params.items) + .await + .map_err(AppCommandError::from)?; + emit_event( + &state.emitter, + MESSAGE_QUEUE_CHANGED_EVENT, + MessageQueueChange { + owner_id: params.owner_id, + origin: params.origin, + items: saved, + }, + ); + Ok(Json(())) +} diff --git a/src-tauri/src/web/handlers/mod.rs b/src-tauri/src/web/handlers/mod.rs index b98d870696..1c14f7ef00 100644 --- a/src-tauri/src/web/handlers/mod.rs +++ b/src-tauri/src/web/handlers/mod.rs @@ -40,6 +40,7 @@ pub mod pet; pub mod project_boot; pub mod question; pub mod quick_messages; +pub mod message_queue; pub mod science; pub mod session_info; pub mod system_settings; diff --git a/src-tauri/src/web/router.rs b/src-tauri/src/web/router.rs index 5057faa9de..2afe82ee80 100644 --- a/src-tauri/src/web/router.rs +++ b/src-tauri/src/web/router.rs @@ -1701,6 +1701,15 @@ pub fn build_router_with_proxy_auth( "/quick_messages_reorder", post(handlers::quick_messages::quick_messages_reorder), ) + // ─── Message Queue ─── + .route( + "/message_queue_list", + post(handlers::message_queue::message_queue_list), + ) + .route( + "/message_queue_save", + post(handlers::message_queue::message_queue_save), + ) // ─── Automations ─── .route( "/automation_list", diff --git a/src/components/conversations/conversation-detail-panel.tsx b/src/components/conversations/conversation-detail-panel.tsx index b361fae7fd..316d0240f4 100644 --- a/src/components/conversations/conversation-detail-panel.tsx +++ b/src/components/conversations/conversation-detail-panel.tsx @@ -698,7 +698,9 @@ const ConversationTabView = memo(function ConversationTabView({ ), }) const { status: connStatus, sessionId: connSessionId } = conn - const messageQueue = useMessageQueue() + const messageQueue = useMessageQueue( + dbConversationId != null ? String(dbConversationId) : null + ) const { queue: msgQueue, enqueue: mqEnqueue, diff --git a/src/hooks/use-message-queue.ts b/src/hooks/use-message-queue.ts index 326aa8e7bf..ad4cad9a0c 100644 --- a/src/hooks/use-message-queue.ts +++ b/src/hooks/use-message-queue.ts @@ -1,8 +1,13 @@ "use client" -import { useCallback, useRef, useState } from "react" +import { useCallback, useEffect, useRef, useState } from "react" import type { PromptDraft } from "@/lib/types" +import type { MessageQueueChangedPayload, PersistedQueueItem, MessageQueueSaveItem } from "@/lib/types" import { randomUUID } from "@/lib/utils" +import { messageQueueList, messageQueueSave } from "@/lib/api" +import { subscribe } from "@/lib/platform" + +export const MESSAGE_QUEUE_CHANGED_EVENT = "message-queue://changed" export interface QueuedMessage { id: string @@ -61,7 +66,51 @@ export interface UseMessageQueueReturn { cancelEditing: () => void } -export function useMessageQueue(): UseMessageQueueReturn { +function serializeDraft(draft: PromptDraft): string { + return JSON.stringify(draft) +} + +function parseItem(item: PersistedQueueItem): QueuedMessage | null { + try { + const draft = JSON.parse(item.draft_json) as PromptDraft + return { + id: item.id, + draft, + modeId: item.mode_id, + adoptSendTimeMode: item.adopt_send_time_mode || undefined, + } + } catch { + // A corrupt persisted draft must not crash the queue; skip it. + return null + } +} + +function toSaveItem(q: QueuedMessage): MessageQueueSaveItem { + return { + id: q.id, + draftJson: serializeDraft(q.draft), + modeId: q.modeId, + adoptSendTimeMode: q.adoptSendTimeMode ?? false, + } +} + +/** + * Manages a conversation's "send later" message queue. + * + * Without an `ownerId` (e.g. an unbound draft tab) the queue is purely in-memory + * and behaves exactly as before — synchronous, no persistence, no network. This + * keeps the existing unit tests (which construct the hook with no owner) valid. + * + * With an `ownerId` (a conversation id) the queue is PERSISTED to the backend and + * SYNCHRONIZED across every window / web client: + * - on mount (and whenever the owner changes) the persisted queue is loaded; + * - every local mutation is persisted via `message_queue_save`; + * - a `message-queue://changed` broadcast (emitted by the backend on every save) + * lets other clients converge to the same queue. The originating client + * ignores its own echo via the `origin` tag so a concurrent local mutation is + * never clobbered by the broadcast it triggered. + */ +export function useMessageQueue(ownerId?: string | null): UseMessageQueueReturn { const [queue, setQueue] = useState([]) const [editingItemId, setEditingItemId] = useState(null) // Authoritative copy of the queue, updated SYNCHRONOUSLY by every mutation @@ -73,6 +122,25 @@ export function useMessageQueue(): UseMessageQueueReturn { // where a caller still sees an empty queue. const queueRef = useRef(queue) + // Latest owner / client id, read inside async callbacks without re-subscribing. + const ownerIdRef = useRef(ownerId) + ownerIdRef.current = ownerId + const clientIdRef = useRef(randomUUID()) + const editingItemIdRef = useRef(null) + editingItemIdRef.current = editingItemId + // Tracks the previously-bound owner so we can tell "bind" from "switch". + const prevOwnerRef = useRef(undefined) + + // Persist the current queue to the backend (if owned) and broadcast. + const persist = useCallback((next: QueuedMessage[]) => { + const owner = ownerIdRef.current + if (!owner) return + const items = next.map(toSaveItem) + // Fire-and-forget: a failed persist must not break the synchronous flush + // path. The broadcast converges other clients when it succeeds. + void messageQueueSave(owner, items, clientIdRef.current).catch(() => {}) + }, []) + // Update the authoritative ref first, then schedule the render. A plain value // (not a functional updater) is correct because `queueRef.current` is always // the latest committed value. @@ -81,13 +149,76 @@ export function useMessageQueue(): UseMessageQueueReturn { setQueue(next) }, []) + // Load the persisted queue when the owner appears or changes. + useEffect(() => { + if (!ownerId) { + prevOwnerRef.current = ownerId + return + } + const prev = prevOwnerRef.current + prevOwnerRef.current = ownerId + let cancelled = false + messageQueueList(ownerId) + .then((items) => { + if (cancelled) return + const parsed = items.map(parseItem).filter((x): x is QueuedMessage => x !== null) + const hadInMemory = queueRef.current.length > 0 + if (prev === null || prev === undefined) { + // Binding from an unbound tab: keep any parked prompts (they have no + // persisted row yet) and persist them so they survive from here on. + if (hadInMemory) { + persist(queueRef.current) + } else { + commit(parsed) + } + } else { + // Switched to a different conversation: load its persisted queue. + commit(parsed) + } + }) + .catch(() => {}) + return () => { + cancelled = true + } + }, [ownerId, commit, persist]) + + // Cross-client sync: apply broadcasts that originated elsewhere. + useEffect(() => { + if (!ownerId) return + let unsub: (() => void) | undefined + let cancelled = false + subscribe(MESSAGE_QUEUE_CHANGED_EVENT, (payload) => { + if (payload.owner_id !== ownerIdRef.current) return + if (payload.origin === clientIdRef.current) return + const parsed = payload.items + .map(parseItem) + .filter((x): x is QueuedMessage => x !== null) + // If the item being edited vanished on another client, drop edit mode. + if ( + editingItemIdRef.current !== null && + !parsed.some((p) => p.id === editingItemIdRef.current) + ) { + setEditingItemId(null) + } + commit(parsed) + }) + .then((u) => { + if (!cancelled) unsub = u + }) + .catch(() => {}) + return () => { + cancelled = true + unsub?.() + } + }, [ownerId, commit]) + const enqueue = useCallback( ( draft: PromptDraft, modeId: string | null, opts?: { adoptSendTimeMode?: boolean } ) => { - commit([ + const next = [ ...queueRef.current, { id: randomUUID(), @@ -95,33 +226,41 @@ export function useMessageQueue(): UseMessageQueueReturn { modeId, ...(opts?.adoptSendTimeMode ? { adoptSendTimeMode: true } : {}), }, - ]) + ] + commit(next) + persist(next) }, - [commit] + [commit, persist] ) const requeueFront = useCallback( (draft: PromptDraft, modeId: string | null) => { - commit([{ id: randomUUID(), draft, modeId }, ...queueRef.current]) + const next = [{ id: randomUUID(), draft, modeId }, ...queueRef.current] + commit(next) + persist(next) }, - [commit] + [commit, persist] ) const dequeue = useCallback((): QueuedMessage | undefined => { const current = queueRef.current if (current.length === 0) return undefined - commit(current.slice(1)) + const next = current.slice(1) + commit(next) + persist(next) return current[0] - }, [commit]) + }, [commit, persist]) const remove = useCallback( (id: string) => { - if (editingItemId === id) { + if (editingItemIdRef.current === id) { setEditingItemId(null) } - commit(queueRef.current.filter((item) => item.id !== id)) + const next = queueRef.current.filter((item) => item.id !== id) + commit(next) + persist(next) }, - [commit, editingItemId] + [commit, persist] ) const reorder = useCallback( @@ -146,20 +285,21 @@ export function useMessageQueue(): UseMessageQueueReturn { next.push(authoritative) } commit(next) + persist(next) }, - [commit] + [commit, persist] ) const updateItem = useCallback( (id: string, draft: PromptDraft) => { - commit( - queueRef.current.map((item) => - item.id === id ? { ...item, draft } : item - ) + const next = queueRef.current.map((item) => + item.id === id ? { ...item, draft } : item ) + commit(next) + persist(next) setEditingItemId(null) }, - [commit] + [commit, persist] ) const moveToFront = useCallback( @@ -167,9 +307,11 @@ export function useMessageQueue(): UseMessageQueueReturn { const current = queueRef.current const index = current.findIndex((item) => item.id === id) if (index <= 0) return - commit([current[index], ...current.filter((item) => item.id !== id)]) + const next = [current[index], ...current.filter((item) => item.id !== id)] + commit(next) + persist(next) }, - [commit] + [commit, persist] ) const getQueueLength = useCallback(() => queueRef.current.length, []) diff --git a/src/lib/api.ts b/src/lib/api.ts index 60a0858544..edd2db451a 100644 --- a/src/lib/api.ts +++ b/src/lib/api.ts @@ -3948,6 +3948,22 @@ export async function quickMessagesReorder(ids: number[]): Promise { return getTransport().call("quick_messages_reorder", { ids }) } +// Persisted, multi-client-synced message queue (the "send later" queue). + +export async function messageQueueList( + ownerId: string +): Promise { + return getTransport().call("message_queue_list", { ownerId }) +} + +export async function messageQueueSave( + ownerId: string, + items: MessageQueueSaveItem[], + origin: string +): Promise { + return getTransport().call("message_queue_save", { ownerId, items, origin }) +} + // Token usage dashboard export async function tokenUsageReport( diff --git a/src/lib/types.ts b/src/lib/types.ts index 0ec29db6a8..daf657276c 100644 --- a/src/lib/types.ts +++ b/src/lib/types.ts @@ -4745,6 +4745,32 @@ export interface QuickMessage { updated_at: string } +// Persisted message-queue (the "send later" queue). `draft_json` is the +// serialized PromptDraft; fields are snake_case because the Rust side serializes +// command returns and event payloads without camelCase conversion. +export interface PersistedQueueItem { + id: string + draft_json: string + mode_id: string | null + adopt_send_time_mode: boolean +} + +// Shape accepted by `message_queue_save`. camelCase because both the Tauri +// command (nested arg) and the axum web handler (JSON body) expect it. +export interface MessageQueueSaveItem { + id: string + draftJson: string + modeId: string | null + adoptSendTimeMode: boolean +} + +// Payload of the `message-queue://changed` broadcast (snake_case on the wire). +export interface MessageQueueChangedPayload { + owner_id: string + origin: string + items: PersistedQueueItem[] +} + export interface GitStatusEntry { status: string file: string From 6112ee78cad670b3a6c28352a9da89917db9b846 Mon Sep 17 00:00:00 2001 From: jry Date: Sun, 1 Nov 2026 00:23:11 +0800 Subject: [PATCH 2/2] =?UTF-8?q?fix(message-queue):=20satisfy=20CI=20?= =?UTF-8?q?=E2=80=94=20unused=20imports,=20refs=20in=20render,=20missing?= =?UTF-8?q?=20type=20imports?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - drop unused ConnectionTrait / IntoActiveModel imports: clippy -D warnings failed all 9 Rust jobs on them - sync ownerIdRef / editingItemIdRef in effects instead of during render: the react-hooks lint bans ref writes in render - import PersistedQueueItem / MessageQueueSaveItem in api.ts: the build's type-check (TS2304) was never reached in CI because lint failed first - run prettier over the hook (import wrapping, indentation) --- .../src/db/service/message_queue_service.rs | 4 +- src/hooks/use-message-queue.ts | 57 ++++++++++++------- src/lib/api.ts | 2 + 3 files changed, 42 insertions(+), 21 deletions(-) diff --git a/src-tauri/src/db/service/message_queue_service.rs b/src-tauri/src/db/service/message_queue_service.rs index c7ef53d353..98c2741c23 100644 --- a/src-tauri/src/db/service/message_queue_service.rs +++ b/src-tauri/src/db/service/message_queue_service.rs @@ -1,7 +1,7 @@ use chrono::Utc; use sea_orm::{ - ActiveModelTrait, ActiveValue::NotSet, ColumnTrait, ConnectionTrait, DatabaseConnection, - EntityTrait, IntoActiveModel, QueryFilter, QueryOrder, Set, TransactionTrait, + ActiveModelTrait, ActiveValue::NotSet, ColumnTrait, DatabaseConnection, EntityTrait, + QueryFilter, QueryOrder, Set, TransactionTrait, }; use serde::Deserialize; diff --git a/src/hooks/use-message-queue.ts b/src/hooks/use-message-queue.ts index ad4cad9a0c..cb9316d3b8 100644 --- a/src/hooks/use-message-queue.ts +++ b/src/hooks/use-message-queue.ts @@ -2,7 +2,11 @@ import { useCallback, useEffect, useRef, useState } from "react" import type { PromptDraft } from "@/lib/types" -import type { MessageQueueChangedPayload, PersistedQueueItem, MessageQueueSaveItem } from "@/lib/types" +import type { + MessageQueueChangedPayload, + PersistedQueueItem, + MessageQueueSaveItem, +} from "@/lib/types" import { randomUUID } from "@/lib/utils" import { messageQueueList, messageQueueSave } from "@/lib/api" import { subscribe } from "@/lib/platform" @@ -110,7 +114,9 @@ function toSaveItem(q: QueuedMessage): MessageQueueSaveItem { * ignores its own echo via the `origin` tag so a concurrent local mutation is * never clobbered by the broadcast it triggered. */ -export function useMessageQueue(ownerId?: string | null): UseMessageQueueReturn { +export function useMessageQueue( + ownerId?: string | null +): UseMessageQueueReturn { const [queue, setQueue] = useState([]) const [editingItemId, setEditingItemId] = useState(null) // Authoritative copy of the queue, updated SYNCHRONOUSLY by every mutation @@ -123,11 +129,19 @@ export function useMessageQueue(ownerId?: string | null): UseMessageQueueReturn const queueRef = useRef(queue) // Latest owner / client id, read inside async callbacks without re-subscribing. + // Synced in effects, not during render: the react-hooks lint bans ref writes + // in render. Discrete events flush passive effects first, so a handler can + // never observe a stale owner; a broadcast racing that flush window fails its + // owner check and is healed by the load effect below. const ownerIdRef = useRef(ownerId) - ownerIdRef.current = ownerId + useEffect(() => { + ownerIdRef.current = ownerId + }, [ownerId]) const clientIdRef = useRef(randomUUID()) const editingItemIdRef = useRef(null) - editingItemIdRef.current = editingItemId + useEffect(() => { + editingItemIdRef.current = editingItemId + }, [editingItemId]) // Tracks the previously-bound owner so we can tell "bind" from "switch". const prevOwnerRef = useRef(undefined) @@ -161,7 +175,9 @@ export function useMessageQueue(ownerId?: string | null): UseMessageQueueReturn messageQueueList(ownerId) .then((items) => { if (cancelled) return - const parsed = items.map(parseItem).filter((x): x is QueuedMessage => x !== null) + const parsed = items + .map(parseItem) + .filter((x): x is QueuedMessage => x !== null) const hadInMemory = queueRef.current.length > 0 if (prev === null || prev === undefined) { // Binding from an unbound tab: keep any parked prompts (they have no @@ -187,21 +203,24 @@ export function useMessageQueue(ownerId?: string | null): UseMessageQueueReturn if (!ownerId) return let unsub: (() => void) | undefined let cancelled = false - subscribe(MESSAGE_QUEUE_CHANGED_EVENT, (payload) => { - if (payload.owner_id !== ownerIdRef.current) return - if (payload.origin === clientIdRef.current) return - const parsed = payload.items - .map(parseItem) - .filter((x): x is QueuedMessage => x !== null) - // If the item being edited vanished on another client, drop edit mode. - if ( - editingItemIdRef.current !== null && - !parsed.some((p) => p.id === editingItemIdRef.current) - ) { - setEditingItemId(null) + subscribe( + MESSAGE_QUEUE_CHANGED_EVENT, + (payload) => { + if (payload.owner_id !== ownerIdRef.current) return + if (payload.origin === clientIdRef.current) return + const parsed = payload.items + .map(parseItem) + .filter((x): x is QueuedMessage => x !== null) + // If the item being edited vanished on another client, drop edit mode. + if ( + editingItemIdRef.current !== null && + !parsed.some((p) => p.id === editingItemIdRef.current) + ) { + setEditingItemId(null) + } + commit(parsed) } - commit(parsed) - }) + ) .then((u) => { if (!cancelled) unsub = u }) diff --git a/src/lib/api.ts b/src/lib/api.ts index edd2db451a..7f77779da9 100644 --- a/src/lib/api.ts +++ b/src/lib/api.ts @@ -154,6 +154,7 @@ import type { GeneratedCommitMessage, GitRemote, GitStashEntry, + PersistedQueueItem, PreflightResult, FolderCommand, TerminalInfo, @@ -200,6 +201,7 @@ import type { ChannelStatusInfo, ChatChannelMessageLog, WebhookConfig, + MessageQueueSaveItem, ModelProviderInfo, UpdateModelProviderResult, OpencodePluginList,