From c0af36ec7e5eb07003fb5acafda5d0f9d161b8c3 Mon Sep 17 00:00:00 2001 From: Wes Date: Thu, 30 Apr 2026 09:19:16 -0700 Subject: [PATCH] feat: replace polling with WS subscriptions + wire workflow approvals (WF-08) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace REST polling with WebSocket subscriptions across the desktop app for real-time reactivity. Wire the workflow approval infrastructure so runs suspend at approval gates instead of failing. Desktop subscriptions: - Forum posts/threads: subscribeLive for kinds 45001/45003 (was 15s/10s poll) - Workflows: subscribeLive for kind 46010 (was 1s/10s poll) - Pulse timeline: subscribeLive for contact note events (was 30s poll) - Channels list: reduced backstop to 5min (live updates already wired) - Agents/personas/teams: removed polling, CRUD mutations invalidate cache - Home feed: kept at 30s poll (relay scoping prevents global subscriptions) Shared useReactiveSubscription hook extracts common subscribe/reconnect/ cleanup lifecycle across forum, workflow, and pulse hooks. WF-08 workflow approvals: - executor.rs: create_approval() called with token, expiry, approver spec - lib.rs: finalize_run sets WaitingApproval instead of Failed stub - Trace entry pushed for suspended steps (UI can render approval card) - Approver spec validated at creation (rejects unresolvable role specs) - e2e test: full round-trip trigger → waiting_approval → grant → completed Docs: added "Subscriptions, Not Polling" principle to AGENTS.md. Co-Authored-By: Claude Opus 4.6 --- AGENTS.md | 12 ++ .../sprout-test-client/tests/e2e_workflows.rs | 108 ++++++++++++++---- crates/sprout-workflow/src/executor.rs | 59 +++++++++- crates/sprout-workflow/src/lib.rs | 16 +-- desktop/src/features/agents/hooks.ts | 12 +- desktop/src/features/channels/hooks.ts | 4 +- desktop/src/features/forum/hooks.ts | 30 ++++- desktop/src/features/forum/ui/ForumView.tsx | 2 + desktop/src/features/home/hooks.ts | 4 +- desktop/src/features/pulse/hooks.ts | 43 ++++++- desktop/src/features/pulse/ui/PulseView.tsx | 4 + desktop/src/features/workflows/hooks.ts | 35 +++++- .../workflows/ui/WorkflowDetailPanel.tsx | 2 + .../shared/hooks/useReactiveSubscription.ts | 89 +++++++++++++++ 14 files changed, 373 insertions(+), 47 deletions(-) create mode 100644 desktop/src/shared/hooks/useReactiveSubscription.ts diff --git a/AGENTS.md b/AGENTS.md index 374eb3e62..603a33730 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -92,6 +92,18 @@ simple and testable. thread root events. Any code that inserts replies must update these counters — check existing reply handlers for the pattern. +### Subscriptions, Not Polling + +The desktop app uses WebSocket subscriptions (NIP-29 relay protocol) instead of polling for real-time data. When adding new data-fetching hooks: + +- **Subscribe to events** via `relayClient.subscribeLive()` or existing `subscribe*` methods +- **Update TanStack Query cache** directly with `setQueriesData` (targeted) or `invalidateQueries` (broad) +- **Handle reconnects** via `relayClient.subscribeToReconnects()` to recover missed events +- **Backstop polling** (60s+) is acceptable ONLY for data written by non-WS clients (e.g., ACP agents via REST) +- **CRUD-driven data** (agents, personas, teams) should invalidate cache in mutations, not poll + +Reference: `usePresenceSubscription()` in `features/presence/hooks.ts` + --- ## Testing diff --git a/crates/sprout-test-client/tests/e2e_workflows.rs b/crates/sprout-test-client/tests/e2e_workflows.rs index 4bc408dbd..6d15c46aa 100644 --- a/crates/sprout-test-client/tests/e2e_workflows.rs +++ b/crates/sprout-test-client/tests/e2e_workflows.rs @@ -614,17 +614,13 @@ async fn test_workflow_update_and_delete() { ); } -// ── Test 7: Approval gate (WF-08 stub) ──────────────────────────────────────── +// ── Test 7: Approval gate round-trip (WF-08) ────────────────────────────────── -/// Create a workflow with a `request_approval` step, trigger it, and verify -/// the run fails with the "approval gates not yet implemented" message. -/// -/// This test documents the current stub behavior. When WF-08 is implemented, -/// this test should be updated to verify the full approval round-trip: -/// create → trigger → poll for waiting_approval → grant → verify completed. +/// Full approval round-trip: create workflow with `request_approval` step, +/// trigger, poll for `waiting_approval`, fetch the approval token, grant it, +/// and verify the run completes. #[tokio::test] -#[ignore] -async fn test_approval_gate_stub_fails_gracefully() { +async fn test_approval_gate_round_trip() { let client = http_client(); let pubkey_hex: &str = SEEDED_PUBKEY; let base = relay_http_url(); @@ -683,10 +679,10 @@ steps: .expect("trigger response must include 'run_id'") .to_string(); - // ── Step 3: Poll until the run reaches a terminal status ────────────────── + // ── Step 3: Poll until the run reaches waiting_approval ────────────────── let runs_url = format!("{base}/api/workflows/{workflow_id}/runs"); - let mut final_run: Option = None; - for _ in 0..10 { + let mut waiting_run: Option = None; + for _ in 0..20 { tokio::time::sleep(std::time::Duration::from_millis(100)).await; let runs_resp = client .get(&runs_url) @@ -696,6 +692,78 @@ steps: .expect("GET runs failed"); assert_eq!(runs_resp.status(), 200, "GET runs must return 200"); let runs: Vec = runs_resp.json().await.expect("runs must be JSON array"); + if let Some(run) = runs.iter().find(|r| r["id"].as_str() == Some(&run_id)) { + let status = run["status"].as_str().unwrap_or(""); + if status == "waiting_approval" { + waiting_run = Some(run.clone()); + break; + } + if matches!(status, "completed" | "failed" | "cancelled") { + panic!("run reached terminal status '{status}' instead of waiting_approval"); + } + } + } + + let _run = waiting_run.expect("run must reach waiting_approval within 2 seconds"); + + // ── Step 4: Fetch the approval token ───────────────────────────────────── + let approvals_url = format!("{base}/api/workflows/{workflow_id}/runs/{run_id}/approvals"); + let approvals_resp = client + .get(&approvals_url) + .header("X-Pubkey", pubkey_hex) + .send() + .await + .expect("GET approvals failed"); + assert_eq!( + approvals_resp.status(), + 200, + "GET approvals must return 200" + ); + let approvals: Vec = approvals_resp + .json() + .await + .expect("approvals must be JSON array"); + assert!( + !approvals.is_empty(), + "there must be at least one pending approval" + ); + let approval = &approvals[0]; + assert_eq!( + approval["status"].as_str().unwrap_or(""), + "pending", + "approval must be pending" + ); + let approval_token_hash = approval["token"] + .as_str() + .expect("approval must have a token hash"); + + // ── Step 5: Grant the approval (using by-hash endpoint since the listing + // returns the stored hash, not the raw token) ───────────────────────────── + let grant_url = format!("{base}/api/approvals/by-hash/{approval_token_hash}/grant"); + let grant_resp = client + .post(&grant_url) + .header("X-Pubkey", pubkey_hex) + .send() + .await + .expect("POST grant failed"); + assert!( + grant_resp.status().is_success(), + "grant must succeed, got {}", + grant_resp.status() + ); + + // ── Step 6: Poll until the run completes ───────────────────────────────── + let mut final_run: Option = None; + for _ in 0..20 { + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + let runs_resp = client + .get(&runs_url) + .header("X-Pubkey", pubkey_hex) + .send() + .await + .expect("GET runs failed"); + assert_eq!(runs_resp.status(), 200); + let runs: Vec = runs_resp.json().await.expect("runs JSON"); if let Some(run) = runs.iter().find(|r| r["id"].as_str() == Some(&run_id)) { let status = run["status"].as_str().unwrap_or(""); if matches!(status, "completed" | "failed" | "cancelled") { @@ -705,22 +773,14 @@ steps: } } - // ── Step 4: Assert the run failed with the expected stub error ──────────── - let run = final_run.expect("run must reach a terminal status within 1 second"); - + let run = final_run.expect("run must reach terminal status after approval"); assert_eq!( run["status"].as_str().unwrap_or(""), - "failed", - "approval gate stub must cause the run to fail" + "completed", + "run must complete after approval grant" ); - let error_msg = run["error_message"].as_str().unwrap_or(""); - assert!( - error_msg.contains("approval gates not yet implemented"), - "run error must contain 'approval gates not yet implemented', got: {error_msg:?}" - ); - - // ── Step 5: Clean up ────────────────────────────────────────────────────── + // ── Step 7: Clean up ────────────────────────────────────────────────────── let del_status = delete_workflow(&client, &base, pubkey_hex, &workflow_id).await; assert_eq!(del_status, 204, "cleanup DELETE should return 204"); } diff --git a/crates/sprout-workflow/src/executor.rs b/crates/sprout-workflow/src/executor.rs index 09de4a4df..5cd691323 100644 --- a/crates/sprout-workflow/src/executor.rs +++ b/crates/sprout-workflow/src/executor.rs @@ -552,6 +552,7 @@ pub async fn dispatch_action( engine: &WorkflowEngine, run_id: Uuid, trigger_ctx: &TriggerContext, + step_index: usize, ) -> Result { use ActionDef::*; @@ -680,10 +681,58 @@ pub async fn dispatch_action( "RequestApproval from={from} timeout={timeout_str}: {message}" ); + // Validate the approver spec: only "any" or a 64-char hex pubkey + // are supported today. Role-based specs (e.g. "@engineering-lead") + // will need relay-side group role resolution — reject them early so + // workflow authors get a clear error at trigger time. + let is_valid_approver = + from == "any" || (from.len() == 64 && from.chars().all(|c| c.is_ascii_hexdigit())); + if !is_valid_approver { + return Err(WorkflowError::InvalidDefinition(format!( + "unsupported approver spec \"{from}\" — only \"any\" or a 64-char hex pubkey are supported" + ))); + } + let token = generate_approval_token(run_id, step_id); - // TODO (WF-08): create approval record in DB, emit kind:46010. - // For now, return Suspended with the token so the caller can persist state. + // Parse timeout duration and compute expiry. + // TODO: expires_at is stored but not yet enforced — there is no + // background sweep that auto-denies approvals past their deadline. + // Until that is added, expired approvals remain in "pending" state + // and the run stays in WaitingApproval indefinitely. + let timeout_secs = parse_duration_secs(timeout_str)?; + let expires_at = chrono::Utc::now() + chrono::Duration::seconds(timeout_secs as i64); + + // Look up workflow metadata for the approval record. + let wf_run = engine.db.get_workflow_run(run_id).await.map_err(|e| { + WorkflowError::WebhookError(format!( + "RequestApproval: failed to load workflow run {run_id}: {e}" + )) + })?; + + // Persist the approval record in the database. + engine + .db + .create_approval(sprout_db::workflow::CreateApprovalParams { + token: &token, + workflow_id: wf_run.workflow_id, + run_id, + step_id, + step_index: step_index as i32, + approver_spec: from, + expires_at, + }) + .await + .map_err(|e| { + WorkflowError::WebhookError(format!( + "RequestApproval: failed to create approval record: {e}" + )) + })?; + + info!( + run_id = %run_id, step = step_id, + "Approval record created — token persisted, awaiting approval" + ); Ok(StepResult::Suspended { approval_token: token, @@ -1162,7 +1211,7 @@ async fn execute_steps( .unwrap_or(engine.config.default_timeout_secs); let dispatch_result = tokio::time::timeout( std::time::Duration::from_secs(timeout_secs), - dispatch_action(&step.id, &resolved_action, engine, run_id, trigger_ctx), + dispatch_action(&step.id, &resolved_action, engine, run_id, trigger_ctx, i), ) .await; @@ -1205,6 +1254,10 @@ async fn execute_steps( run_id = %run_id, step = %step.id, "Step suspended — awaiting approval (token: )" ); + trace.push(serde_json::json!({ + "step_id": step.id, + "status": "waiting_approval", + })); // Return the token and current state so the caller can persist the // approval record and update the run's execution trace. return Ok(ExecutionResult { diff --git a/crates/sprout-workflow/src/lib.rs b/crates/sprout-workflow/src/lib.rs index fa66bf8ad..52a4a2bd3 100644 --- a/crates/sprout-workflow/src/lib.rs +++ b/crates/sprout-workflow/src/lib.rs @@ -156,27 +156,29 @@ impl WorkflowEngine { let step_count = result.step_index as i32; if result.approval_token.is_some() { - // Approval gates are not yet implemented (WF-08). - // Fail explicitly rather than creating unreachable WaitingApproval rows. - tracing::warn!( + // WF-08: Approval gate reached — suspend the run. + // The approval record was already created by the executor's + // dispatch_action. Mark the run as WaitingApproval so the + // resume path can pick up at step_index + 1 after grant. + tracing::info!( run_id = %run_id, step_index = result.step_index, - "Workflow hit approval gate — not yet implemented, marking as failed" + "Workflow suspended — waiting for approval" ); if let Err(e) = self .db .update_workflow_run( run_id, - RunStatus::Failed, + RunStatus::WaitingApproval, step_count, &trace_json, - Some("approval gates not yet implemented — see WF-08"), + None, ) .await { tracing::error!( run_id = %run_id, - "Failed to update run to Failed (approval gate): {e}" + "Failed to update run to WaitingApproval: {e}" ); } } else { diff --git a/desktop/src/features/agents/hooks.ts b/desktop/src/features/agents/hooks.ts index cf5aa9d7b..c4bf00882 100644 --- a/desktop/src/features/agents/hooks.ts +++ b/desktop/src/features/agents/hooks.ts @@ -119,7 +119,7 @@ export function usePersonasQuery() { queryKey: personasQueryKey, queryFn: listPersonas, staleTime: 30_000, - refetchInterval: 30_000, + // CRUD-driven: mutations invalidate cache directly, no polling needed. }); } @@ -150,7 +150,7 @@ export function useRelayAgentsQuery(options?: { enabled?: boolean }) { queryKey: relayAgentsQueryKey, queryFn: listRelayAgents, staleTime: 30_000, - refetchInterval: 30_000, + // CRUD-driven: mutations invalidate cache directly, no polling needed. enabled: options?.enabled, }); } @@ -166,9 +166,10 @@ export function useManagedAgentsQuery(options?: { enabled?: boolean }) { // Only local "running" agents need fast polling (process state can // change). "deployed" is static control-plane state — presence polling // handles the live signal for remote agents separately. + // CRUD-driven: mutations invalidate cache directly, no idle polling. return agents?.some((agent) => agent.status === "running") ? 5_000 - : 30_000; + : false; }, }); } @@ -484,7 +485,8 @@ export function useManagedAgentLogQuery( enabled: pubkey !== null, retry: false, staleTime: 3_000, - refetchInterval: pubkey ? 30_000 : false, + // Logs are local process state; keep polling while viewing. + refetchInterval: pubkey ? 5_000 : false, }); } @@ -493,7 +495,7 @@ export function useTeamsQuery() { queryKey: teamsQueryKey, queryFn: listTeams, staleTime: 30_000, - refetchInterval: 30_000, + // CRUD-driven: mutations invalidate cache directly, no polling needed. }); } diff --git a/desktop/src/features/channels/hooks.ts b/desktop/src/features/channels/hooks.ts index 20d874203..7e5ef2ad5 100644 --- a/desktop/src/features/channels/hooks.ts +++ b/desktop/src/features/channels/hooks.ts @@ -110,7 +110,9 @@ export function useChannelsQuery() { queryKey: channelsQueryKey, queryFn: async () => sortChannels(await getChannels()), staleTime: 60_000, - refetchInterval: 60_000, + // Live updates handled by useLiveChannelUpdates (subscribes to channel + // events and invalidates on reconnect). 5-minute backstop for edge cases. + refetchInterval: 5 * 60_000, refetchIntervalInBackground: false, }); } diff --git a/desktop/src/features/forum/hooks.ts b/desktop/src/features/forum/hooks.ts index d8480d429..2eb0c70a4 100644 --- a/desktop/src/features/forum/hooks.ts +++ b/desktop/src/features/forum/hooks.ts @@ -8,6 +8,7 @@ import type { ForumThreadResponse, } from "@/shared/api/types"; import { KIND_FORUM_COMMENT, KIND_FORUM_POST } from "@/shared/constants/kinds"; +import { useReactiveSubscription } from "@/shared/hooks/useReactiveSubscription"; export function forumPostsQueryKey(channelId: string) { return ["forum-posts", channelId] as const; @@ -25,7 +26,8 @@ export function useForumPostsQuery(channel: Channel | null) { queryKey: forumPostsQueryKey(channelId), queryFn: () => getForumPosts(channelId, 50), staleTime: 15_000, - refetchInterval: 15_000, + // Live updates via useForumSubscription; 60s backstop for edge cases. + refetchInterval: 60_000, }); } @@ -38,10 +40,34 @@ export function useForumThreadQuery( queryKey: forumThreadQueryKey(channelId ?? "", eventId ?? ""), queryFn: () => getForumThread(channelId ?? "", eventId ?? ""), staleTime: 10_000, - refetchInterval: 10_000, + // Live updates via useForumSubscription; 60s backstop for edge cases. + refetchInterval: 60_000, }); } +/** + * Subscribe to forum post and comment events for a channel. + * Invalidates forum queries on incoming events and reconnects. + */ +export function useForumSubscription(channelId: string | null) { + const queryClient = useQueryClient(); + + useReactiveSubscription( + channelId + ? { kinds: [KIND_FORUM_POST, KIND_FORUM_COMMENT], "#h": [channelId] } + : null, + () => { + void queryClient.invalidateQueries({ + predicate: (query) => + (query.queryKey[0] === "forum-posts" || + query.queryKey[0] === "forum-thread") && + query.queryKey[1] === channelId, + }); + }, + "forum", + ); +} + export function useCreateForumPostMutation(channel: Channel | null) { const queryClient = useQueryClient(); diff --git a/desktop/src/features/forum/ui/ForumView.tsx b/desktop/src/features/forum/ui/ForumView.tsx index e9aa44417..e8d8b6f96 100644 --- a/desktop/src/features/forum/ui/ForumView.tsx +++ b/desktop/src/features/forum/ui/ForumView.tsx @@ -12,6 +12,7 @@ import { useDeleteForumPostMutation, useDeleteForumReplyMutation, useForumPostsQuery, + useForumSubscription, useForumThreadQuery, } from "../hooks"; import { ForumComposer } from "./ForumComposer"; @@ -47,6 +48,7 @@ export function ForumView({ const [isComposerOpen, setIsComposerOpen] = React.useState(false); const profileQuery = useProfileQuery(); + useForumSubscription(channel.id); const postsQuery = useForumPostsQuery(channel); const threadQuery = useForumThreadQuery( selectedPostId ? channel.id : null, diff --git a/desktop/src/features/home/hooks.ts b/desktop/src/features/home/hooks.ts index 9b4f59378..1b51a643e 100644 --- a/desktop/src/features/home/hooks.ts +++ b/desktop/src/features/home/hooks.ts @@ -2,9 +2,11 @@ import { useQuery } from "@tanstack/react-query"; import { getHomeFeed } from "@/shared/api/tauri"; +const homeFeedQueryKey = ["home-feed"] as const; + export function useHomeFeedQuery() { return useQuery({ - queryKey: ["home-feed"], + queryKey: homeFeedQueryKey, queryFn: () => getHomeFeed({ limit: 12, diff --git a/desktop/src/features/pulse/hooks.ts b/desktop/src/features/pulse/hooks.ts index 8bfc2af2c..56748869b 100644 --- a/desktop/src/features/pulse/hooks.ts +++ b/desktop/src/features/pulse/hooks.ts @@ -1,5 +1,7 @@ +import * as React from "react"; import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query"; +import { useReactiveSubscription } from "@/shared/hooks/useReactiveSubscription"; import { getContactList, getNotesTimeline, @@ -46,7 +48,8 @@ export function useMyNotesQuery(pubkey?: string) { enabled: !!pubkey, staleTime: 15_000, gcTime: 5 * 60_000, - refetchInterval: 30_000, + // Invalidated by usePublishNoteMutation; 120s backstop. + refetchInterval: 120_000, }); } @@ -59,10 +62,46 @@ export function useTimelineQuery(contactPubkeys: string[], enabled: boolean) { enabled: enabled && contactPubkeys.length > 0, staleTime: 15_000, gcTime: 5 * 60_000, - refetchInterval: 30_000, + // Live updates via usePulseSubscription; 120s backstop. + refetchInterval: 120_000, }); } +/** + * Subscribe to note events (kind:1) from contacts. + * Invalidates timeline and my-notes queries on incoming events and reconnects. + */ +export function usePulseSubscription( + contactPubkeys: string[], + currentPubkey: string | undefined, +) { + const queryClient = useQueryClient(); + const normalizedPubkey = currentPubkey?.trim().toLowerCase() ?? ""; + + const authors = React.useMemo(() => { + const set = new Set(contactPubkeys); + if (normalizedPubkey.length > 0) { + set.add(normalizedPubkey); + } + return [...set].sort(); + }, [contactPubkeys, normalizedPubkey]); + + useReactiveSubscription( + authors.length > 0 ? { kinds: [1], authors } : null, + () => { + void queryClient.invalidateQueries({ + queryKey: pulseQueryKeys.allTimelines, + }); + if (normalizedPubkey) { + void queryClient.invalidateQueries({ + queryKey: pulseQueryKeys.myNotes(normalizedPubkey), + }); + } + }, + "pulse", + ); +} + // ── Publish note mutation ─────────────────────────────────────────────────── export function usePublishNoteMutation(currentPubkey?: string) { diff --git a/desktop/src/features/pulse/ui/PulseView.tsx b/desktop/src/features/pulse/ui/PulseView.tsx index 99f01a588..74492a607 100644 --- a/desktop/src/features/pulse/ui/PulseView.tsx +++ b/desktop/src/features/pulse/ui/PulseView.tsx @@ -8,6 +8,7 @@ import { useFollowMutation, useMyNotesQuery, usePublishNoteMutation, + usePulseSubscription, useTimelineQuery, useUnfollowMutation, } from "@/features/pulse/hooks"; @@ -182,6 +183,9 @@ export function PulseView({ currentPubkey }: PulseViewProps) { [peoplePubkeys, agentPubkeys], ); + // ── Live subscription for note events ─────────────────────────────── + usePulseSubscription(forYouPubkeys, currentPubkey); + // ── Queries per tab ──────────────────────────────────────────────────── const forYouQuery = useTimelineQuery(forYouPubkeys, activeTab === "foryou"); const peopleQuery = useTimelineQuery(peoplePubkeys, activeTab === "people"); diff --git a/desktop/src/features/workflows/hooks.ts b/desktop/src/features/workflows/hooks.ts index 61ab9643f..cb94579f4 100644 --- a/desktop/src/features/workflows/hooks.ts +++ b/desktop/src/features/workflows/hooks.ts @@ -1,6 +1,8 @@ import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query"; import type { WorkflowRun, WorkflowRunStatus } from "@/shared/api/types"; +import { KIND_APPROVAL_REQUEST } from "@/shared/constants/kinds"; +import { useReactiveSubscription } from "@/shared/hooks/useReactiveSubscription"; import { createWorkflow, deleteWorkflow, @@ -71,10 +73,12 @@ export function useWorkflowRunsQuery(workflowId: string | null) { getWorkflowRuns(resolvedWorkflowId), enabled: workflowId !== null, staleTime: 10_000, + // Live updates via useWorkflowSubscription (kind:46010). + // 10s backstop only when a workflow is actively running. refetchInterval: (query) => { const runs = query.state.data as WorkflowRun[] | undefined; return runs?.some((run) => isActiveWorkflowRunStatus(run.status)) - ? 1_000 + ? 10_000 : false; }, }); @@ -90,10 +94,37 @@ export function useRunApprovalsQuery( getRunApprovals(resolvedWorkflowId, resolvedRunId), enabled: workflowId !== null && runId !== null, staleTime: 10_000, - refetchInterval: 10_000, + // Live updates via useWorkflowSubscription (kind:46010); 30s backstop. + refetchInterval: 30_000, }); } +/** + * Subscribe to workflow status change events (kind:46010). + * Invalidates workflow run and approval queries on incoming events. + * + * NOTE: This subscription is forward-looking — the executor does not emit + * kind:46010 events yet. The subscription is wired up now so the UI will + * react automatically once the relay-side publish is implemented. Until then, + * refetchInterval backstops in the queries above drive updates. + */ +export function useWorkflowSubscription() { + const queryClient = useQueryClient(); + + useReactiveSubscription( + { kinds: [KIND_APPROVAL_REQUEST] }, + () => { + void queryClient.invalidateQueries({ + predicate: (query) => + query.queryKey[0] === "workflow-runs" || + query.queryKey[0] === "run-approvals" || + query.queryKey[0] === "workflow", + }); + }, + "workflow", + ); +} + export function useCreateWorkflowMutation(channelId: string) { const queryClient = useQueryClient(); diff --git a/desktop/src/features/workflows/ui/WorkflowDetailPanel.tsx b/desktop/src/features/workflows/ui/WorkflowDetailPanel.tsx index 62c02c27c..ab86cd452 100644 --- a/desktop/src/features/workflows/ui/WorkflowDetailPanel.tsx +++ b/desktop/src/features/workflows/ui/WorkflowDetailPanel.tsx @@ -6,6 +6,7 @@ import { useTriggerWorkflowMutation, useWorkflowQuery, useWorkflowRunsQuery, + useWorkflowSubscription, } from "@/features/workflows/hooks"; import { WorkflowRunTrace } from "@/features/workflows/ui/WorkflowRunTrace"; import type { Workflow } from "@/shared/api/types"; @@ -29,6 +30,7 @@ export function WorkflowDetailPanel({ onClose, onEdit, }: WorkflowDetailPanelProps) { + useWorkflowSubscription(); const workflowQuery = useWorkflowQuery(workflowId); const runsQuery = useWorkflowRunsQuery(workflowId); const triggerMutation = useTriggerWorkflowMutation(workflowId); diff --git a/desktop/src/shared/hooks/useReactiveSubscription.ts b/desktop/src/shared/hooks/useReactiveSubscription.ts new file mode 100644 index 000000000..2f3b5d4a6 --- /dev/null +++ b/desktop/src/shared/hooks/useReactiveSubscription.ts @@ -0,0 +1,89 @@ +import * as React from "react"; + +import { relayClient } from "@/shared/api/relayClient"; +import type { RelaySubscriptionFilter } from "@/shared/api/relayClientShared"; + +type SubscriptionFilter = Omit; + +/** + * Subscribe to live relay events matching `filter` and invalidate React Query + * cache via `onInvalidate` whenever a new event arrives or the WebSocket + * reconnects. + * + * Pass `null` as the filter to skip subscribing (e.g. when a required ID isn't + * available yet). The reconnect listener is still registered so cache is + * refreshed after connection recovery even when the subscription is skipped. + * + * @param filter Nostr subscription filter, or `null` to disable. + * @param onInvalidate Called on each incoming event and on reconnect. + * @param label Human-readable label for error logging (e.g. "forum"). + */ +export function useReactiveSubscription( + filter: SubscriptionFilter | null, + onInvalidate: () => void, + label: string, +) { + // Stabilise the invalidation callback so the effect doesn't re-run when + // the caller passes an inline arrow. + const onInvalidateRef = React.useRef(onInvalidate); + React.useLayoutEffect(() => { + onInvalidateRef.current = onInvalidate; + }); + + // Memoize the filter by its serialized content so callers can pass object + // literals without churning the subscription effect. + const filterKey = filter === null ? "null" : stableFilterKey(filter); + // biome-ignore lint/correctness/useExhaustiveDependencies: filterKey is the stable serialization of filter + const stableFilter = React.useMemo(() => filter, [filterKey]); + + React.useEffect(() => { + let isCancelled = false; + let cleanup: (() => Promise) | undefined; + + const invalidate = () => { + onInvalidateRef.current(); + }; + + const disposeReconnect = relayClient.subscribeToReconnects(invalidate); + + if (stableFilter === null) { + return () => { + disposeReconnect(); + }; + } + + relayClient + .subscribeLive( + { ...stableFilter, limit: 0, since: Math.floor(Date.now() / 1_000) }, + () => { + if (!isCancelled) { + invalidate(); + } + }, + ) + .then((dispose) => { + if (isCancelled) { + void dispose(); + return; + } + cleanup = dispose; + }) + .catch((error) => { + console.error(`Failed to subscribe to ${label} events`, error); + }); + + return () => { + isCancelled = true; + disposeReconnect(); + if (cleanup) void cleanup(); + }; + }, [label, stableFilter]); +} + +/** + * Produce a stable string key for a filter object so the effect only re-runs + * when the filter semantically changes. + */ +function stableFilterKey(filter: SubscriptionFilter): string { + return JSON.stringify(filter, Object.keys(filter).sort()); +}