mirror of
https://github.com/simstudioai/sim.git
synced 2026-04-06 03:00:16 -04:00
feat(mothership): server-persisted unread task indicators via SSE (#3549)
* feat(mothership): server-persisted unread task indicators via SSE Replace fragile client-side polling + timer-based green flash with server-persisted lastSeenAt semantics, real-time SSE push via Redis pub/sub, and dot overlay UI on the Blimp icon. - Add lastSeenAt column to copilotChats for server-persisted read state - Add Redis/local pub/sub singleton for task status events (started, completed, created, deleted, renamed) - Add SSE endpoint (GET /api/mothership/events) with heartbeat and workspace-scoped filtering - Add mark-read endpoint (POST /api/mothership/chats/read) - Publish SSE events from chat, rename, delete, and auto-title handlers - Add useTaskEvents hook for client-side SSE subscription - Add useMarkTaskRead mutation with optimistic update - Replace timer logic in sidebar with TaskStatus state machine (running/unread/idle) and dot overlay using brand color variables - Mark tasks read on mount and stream completion in home page - Fix security: add userId check to delete WHERE clause - Fix: bump updatedAt on stream completion - Fix: set lastSeenAt on rename to prevent false-positive unread Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * fix: address PR review feedback - Return 404 when delete finds no matching chat (was silent no-op) - Move log after ownership check so it only fires on actual deletion - Publish completed SSE event from stop route so sidebar dot clears on abort Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * fix: backfill last_seen_at in migration to prevent false unread dots Existing rows would have last_seen_at = NULL after migration, causing all past completed tasks to show as unread. Backfill sets last_seen_at to updated_at for all existing rows. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * fix: timestamp mismatch on task creation + wasSendingRef leak across navigation - Pass updatedAt explicitly alongside lastSeenAt on chat creation so both use the same JS timestamp (DB defaultNow() ran later, causing updatedAt > lastSeenAt → false unread) - Reset wasSendingRef when chatId changes to prevent a stale true from task A triggering a redundant markRead on task B Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * fix: mark-read fires for inline-created chats + encode workspaceId in SSE URL Expose resolvedChatId from useChat so home.tsx can mark-read even when chatId prop stays undefined after replaceState URL update. Also URL-encode workspaceId in EventSource URL as a defensive measure. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * fix: auto-focus home input on initial view + fix sidebar task click handling Auto-focus the textarea when the initial home view renders. Also fix sidebar task click to always call onMultiSelectClick so selection state stays consistent. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * fix: auto-title sets lastSeenAt + move started event inside DB guard Auto-title now sets both updatedAt and lastSeenAt (matching the rename route pattern) to prevent false-positive unread dots. Also move the 'started' SSE event inside the if(updated) guard so it only fires when the DB update actually matched a row. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * modified tasks multi select to be just like workflows * fix * refactor: extract generic pub/sub and SSE factories + fixes - Extract createPubSubChannel factory (lib/events/pubsub.ts) to eliminate duplicated Redis/EventEmitter boilerplate between task and MCP pub/sub - Extract createWorkspaceSSE factory (lib/events/sse-endpoint.ts) to share auth, heartbeat, and cleanup logic across SSE endpoints - Fix auto-title race suppressing unread status by removing updatedAt/lastSeenAt from title-only DB update - Fix wheel event listener leak in ResourceTabs (RefCallback cleanup was silently discarded) - Fix getFullSelection() missing taskIds (inconsistent with hasAnySelection) - Deduplicate SSE_RESPONSE_HEADERS to spread from shared SSE_HEADERS - Hoist isSttAvailable to module-level constant to avoid per-render IIFE Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Waleed
@@ -1,10 +1,11 @@
|
||||
import { db } from '@sim/db'
|
||||
import { copilotChats } from '@sim/db/schema'
|
||||
import { createLogger } from '@sim/logger'
|
||||
import { eq } from 'drizzle-orm'
|
||||
import { and, eq } from 'drizzle-orm'
|
||||
import { type NextRequest, NextResponse } from 'next/server'
|
||||
import { z } from 'zod'
|
||||
import { getSession } from '@/lib/auth'
|
||||
import { taskPubSub } from '@/lib/copilot/task-events'
|
||||
|
||||
const logger = createLogger('DeleteChatAPI')
|
||||
|
||||
@@ -22,11 +23,25 @@ export async function DELETE(request: NextRequest) {
|
||||
const body = await request.json()
|
||||
const parsed = DeleteChatSchema.parse(body)
|
||||
|
||||
// Delete the chat
|
||||
await db.delete(copilotChats).where(eq(copilotChats.id, parsed.chatId))
|
||||
const [deleted] = await db
|
||||
.delete(copilotChats)
|
||||
.where(and(eq(copilotChats.id, parsed.chatId), eq(copilotChats.userId, session.user.id)))
|
||||
.returning({ workspaceId: copilotChats.workspaceId })
|
||||
|
||||
if (!deleted) {
|
||||
return NextResponse.json({ success: false, error: 'Chat not found' }, { status: 404 })
|
||||
}
|
||||
|
||||
logger.info('Chat deleted', { chatId: parsed.chatId })
|
||||
|
||||
if (deleted.workspaceId) {
|
||||
taskPubSub?.publishStatusChanged({
|
||||
workspaceId: deleted.workspaceId,
|
||||
chatId: parsed.chatId,
|
||||
type: 'deleted',
|
||||
})
|
||||
}
|
||||
|
||||
return NextResponse.json({ success: true })
|
||||
} catch (error) {
|
||||
logger.error('Error deleting chat:', error)
|
||||
|
||||
@@ -5,6 +5,7 @@ import { and, eq } from 'drizzle-orm'
|
||||
import { type NextRequest, NextResponse } from 'next/server'
|
||||
import { z } from 'zod'
|
||||
import { getSession } from '@/lib/auth'
|
||||
import { taskPubSub } from '@/lib/copilot/task-events'
|
||||
|
||||
const logger = createLogger('RenameChatAPI')
|
||||
|
||||
@@ -23,11 +24,12 @@ export async function PATCH(request: NextRequest) {
|
||||
const body = await request.json()
|
||||
const { chatId, title } = RenameChatSchema.parse(body)
|
||||
|
||||
const now = new Date()
|
||||
const [updated] = await db
|
||||
.update(copilotChats)
|
||||
.set({ title, updatedAt: new Date() })
|
||||
.set({ title, updatedAt: now, lastSeenAt: now })
|
||||
.where(and(eq(copilotChats.id, chatId), eq(copilotChats.userId, session.user.id)))
|
||||
.returning({ id: copilotChats.id })
|
||||
.returning({ id: copilotChats.id, workspaceId: copilotChats.workspaceId })
|
||||
|
||||
if (!updated) {
|
||||
return NextResponse.json({ success: false, error: 'Chat not found' }, { status: 404 })
|
||||
@@ -35,6 +37,14 @@ export async function PATCH(request: NextRequest) {
|
||||
|
||||
logger.info('Chat renamed', { chatId, title })
|
||||
|
||||
if (updated.workspaceId) {
|
||||
taskPubSub?.publishStatusChanged({
|
||||
workspaceId: updated.workspaceId,
|
||||
chatId,
|
||||
type: 'renamed',
|
||||
})
|
||||
}
|
||||
|
||||
return NextResponse.json({ success: true })
|
||||
} catch (error) {
|
||||
if (error instanceof z.ZodError) {
|
||||
|
||||
@@ -8,70 +8,19 @@
|
||||
* Auth is handled via session cookies (EventSource sends cookies automatically).
|
||||
*/
|
||||
|
||||
import { createLogger } from '@sim/logger'
|
||||
import type { NextRequest } from 'next/server'
|
||||
import { getSession } from '@/lib/auth'
|
||||
import { SSE_HEADERS } from '@/lib/core/utils/sse'
|
||||
import { createWorkspaceSSE } from '@/lib/events/sse-endpoint'
|
||||
import { mcpConnectionManager } from '@/lib/mcp/connection-manager'
|
||||
import { mcpPubSub } from '@/lib/mcp/pubsub'
|
||||
import { decrementSSEConnections, incrementSSEConnections } from '@/lib/monitoring/sse-connections'
|
||||
import { getUserEntityPermissions } from '@/lib/workspaces/permissions/utils'
|
||||
|
||||
const logger = createLogger('McpEventsSSE')
|
||||
|
||||
export const dynamic = 'force-dynamic'
|
||||
|
||||
const HEARTBEAT_INTERVAL_MS = 30_000
|
||||
|
||||
export async function GET(request: NextRequest) {
|
||||
const session = await getSession()
|
||||
if (!session?.user?.id) {
|
||||
return new Response('Unauthorized', { status: 401 })
|
||||
}
|
||||
|
||||
const { searchParams } = new URL(request.url)
|
||||
const workspaceId = searchParams.get('workspaceId')
|
||||
if (!workspaceId) {
|
||||
return new Response('Missing workspaceId query parameter', { status: 400 })
|
||||
}
|
||||
|
||||
const permissions = await getUserEntityPermissions(session.user.id, 'workspace', workspaceId)
|
||||
if (!permissions) {
|
||||
return new Response('Access denied to workspace', { status: 403 })
|
||||
}
|
||||
|
||||
const encoder = new TextEncoder()
|
||||
const unsubscribers: Array<() => void> = []
|
||||
let cleaned = false
|
||||
|
||||
const cleanup = () => {
|
||||
if (cleaned) return
|
||||
cleaned = true
|
||||
for (const unsub of unsubscribers) {
|
||||
unsub()
|
||||
}
|
||||
decrementSSEConnections('mcp-events')
|
||||
logger.info(`SSE connection closed for workspace ${workspaceId}`)
|
||||
}
|
||||
|
||||
const stream = new ReadableStream({
|
||||
start(controller) {
|
||||
incrementSSEConnections('mcp-events')
|
||||
|
||||
const send = (eventName: string, data: Record<string, unknown>) => {
|
||||
if (cleaned) return
|
||||
try {
|
||||
controller.enqueue(
|
||||
encoder.encode(`event: ${eventName}\ndata: ${JSON.stringify(data)}\n\n`)
|
||||
)
|
||||
} catch {
|
||||
// Stream already closed
|
||||
}
|
||||
}
|
||||
|
||||
// Subscribe to external MCP server tool changes
|
||||
if (mcpConnectionManager) {
|
||||
const unsub = mcpConnectionManager.subscribe((event) => {
|
||||
export const GET = createWorkspaceSSE({
|
||||
label: 'mcp-events',
|
||||
subscriptions: [
|
||||
{
|
||||
subscribe: (workspaceId, send) => {
|
||||
if (!mcpConnectionManager) return () => {}
|
||||
return mcpConnectionManager.subscribe((event) => {
|
||||
if (event.workspaceId !== workspaceId) return
|
||||
send('tools_changed', {
|
||||
source: 'external',
|
||||
@@ -79,12 +28,12 @@ export async function GET(request: NextRequest) {
|
||||
timestamp: event.timestamp,
|
||||
})
|
||||
})
|
||||
unsubscribers.push(unsub)
|
||||
}
|
||||
|
||||
// Subscribe to workflow CRUD tool changes
|
||||
if (mcpPubSub) {
|
||||
const unsub = mcpPubSub.onWorkflowToolsChanged((event) => {
|
||||
},
|
||||
},
|
||||
{
|
||||
subscribe: (workspaceId, send) => {
|
||||
if (!mcpPubSub) return () => {}
|
||||
return mcpPubSub.onWorkflowToolsChanged((event) => {
|
||||
if (event.workspaceId !== workspaceId) return
|
||||
send('tools_changed', {
|
||||
source: 'workflow',
|
||||
@@ -92,43 +41,7 @@ export async function GET(request: NextRequest) {
|
||||
timestamp: Date.now(),
|
||||
})
|
||||
})
|
||||
unsubscribers.push(unsub)
|
||||
}
|
||||
|
||||
// Heartbeat to keep the connection alive
|
||||
const heartbeat = setInterval(() => {
|
||||
if (cleaned) {
|
||||
clearInterval(heartbeat)
|
||||
return
|
||||
}
|
||||
try {
|
||||
controller.enqueue(encoder.encode(': heartbeat\n\n'))
|
||||
} catch {
|
||||
clearInterval(heartbeat)
|
||||
}
|
||||
}, HEARTBEAT_INTERVAL_MS)
|
||||
unsubscribers.push(() => clearInterval(heartbeat))
|
||||
|
||||
// Cleanup when client disconnects
|
||||
request.signal.addEventListener(
|
||||
'abort',
|
||||
() => {
|
||||
cleanup()
|
||||
try {
|
||||
controller.close()
|
||||
} catch {
|
||||
// Already closed
|
||||
}
|
||||
},
|
||||
{ once: true }
|
||||
)
|
||||
|
||||
logger.info(`SSE connection opened for workspace ${workspaceId}`)
|
||||
},
|
||||
},
|
||||
cancel() {
|
||||
cleanup()
|
||||
},
|
||||
})
|
||||
|
||||
return new Response(stream, { headers: SSE_HEADERS })
|
||||
}
|
||||
],
|
||||
})
|
||||
|
||||
@@ -10,6 +10,7 @@ import { buildCopilotRequestPayload } from '@/lib/copilot/chat-payload'
|
||||
import { createSSEStream, SSE_RESPONSE_HEADERS } from '@/lib/copilot/chat-streaming'
|
||||
import type { OrchestratorResult } from '@/lib/copilot/orchestrator/types'
|
||||
import { createRequestTracker, createUnauthorizedResponse } from '@/lib/copilot/request-helpers'
|
||||
import { taskPubSub } from '@/lib/copilot/task-events'
|
||||
import { generateWorkspaceContext } from '@/lib/copilot/workspace-context'
|
||||
import { getUserEntityPermissions } from '@/lib/workspaces/permissions/utils'
|
||||
|
||||
@@ -146,6 +147,7 @@ export async function POST(req: NextRequest) {
|
||||
if (updated) {
|
||||
const freshMessages: any[] = Array.isArray(updated.messages) ? updated.messages : []
|
||||
conversationHistory = freshMessages.filter((m: any) => m.id !== userMessageId)
|
||||
taskPubSub?.publishStatusChanged({ workspaceId, chatId: actualChatId, type: 'started' })
|
||||
}
|
||||
}
|
||||
|
||||
@@ -182,6 +184,7 @@ export async function POST(req: NextRequest) {
|
||||
message,
|
||||
titleModel: 'claude-opus-4-5',
|
||||
requestId: tracker.requestId,
|
||||
workspaceId,
|
||||
orchestrateOptions: {
|
||||
userId: authenticatedUserId,
|
||||
workspaceId,
|
||||
@@ -243,8 +246,15 @@ export async function POST(req: NextRequest) {
|
||||
.set({
|
||||
messages: sql`${copilotChats.messages} || ${JSON.stringify([assistantMessage])}::jsonb`,
|
||||
conversationId: sql`CASE WHEN ${copilotChats.conversationId} = ${userMessageId} THEN NULL ELSE ${copilotChats.conversationId} END`,
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(eq(copilotChats.id, actualChatId))
|
||||
|
||||
taskPubSub?.publishStatusChanged({
|
||||
workspaceId,
|
||||
chatId: actualChatId,
|
||||
type: 'completed',
|
||||
})
|
||||
}
|
||||
} catch (error) {
|
||||
logger.error(`[${tracker.requestId}] Failed to persist chat messages`, {
|
||||
|
||||
@@ -5,6 +5,7 @@ import { and, eq, sql } from 'drizzle-orm'
|
||||
import { type NextRequest, NextResponse } from 'next/server'
|
||||
import { z } from 'zod'
|
||||
import { getSession } from '@/lib/auth'
|
||||
import { taskPubSub } from '@/lib/copilot/task-events'
|
||||
|
||||
const logger = createLogger('MothershipChatStopAPI')
|
||||
|
||||
@@ -78,7 +79,7 @@ export async function POST(req: NextRequest) {
|
||||
setClause.messages = sql`${copilotChats.messages} || ${JSON.stringify([assistantMessage])}::jsonb`
|
||||
}
|
||||
|
||||
await db
|
||||
const [updated] = await db
|
||||
.update(copilotChats)
|
||||
.set(setClause)
|
||||
.where(
|
||||
@@ -88,6 +89,15 @@ export async function POST(req: NextRequest) {
|
||||
eq(copilotChats.conversationId, streamId)
|
||||
)
|
||||
)
|
||||
.returning({ workspaceId: copilotChats.workspaceId })
|
||||
|
||||
if (updated?.workspaceId) {
|
||||
taskPubSub?.publishStatusChanged({
|
||||
workspaceId: updated.workspaceId,
|
||||
chatId,
|
||||
type: 'completed',
|
||||
})
|
||||
}
|
||||
|
||||
return NextResponse.json({ success: true })
|
||||
} catch (error) {
|
||||
|
||||
43
apps/sim/app/api/mothership/chats/read/route.ts
Normal file
43
apps/sim/app/api/mothership/chats/read/route.ts
Normal file
@@ -0,0 +1,43 @@
|
||||
import { db } from '@sim/db'
|
||||
import { copilotChats } from '@sim/db/schema'
|
||||
import { createLogger } from '@sim/logger'
|
||||
import { and, eq } from 'drizzle-orm'
|
||||
import { type NextRequest, NextResponse } from 'next/server'
|
||||
import { z } from 'zod'
|
||||
import {
|
||||
authenticateCopilotRequestSessionOnly,
|
||||
createBadRequestResponse,
|
||||
createInternalServerErrorResponse,
|
||||
createUnauthorizedResponse,
|
||||
} from '@/lib/copilot/request-helpers'
|
||||
|
||||
const logger = createLogger('MarkTaskReadAPI')
|
||||
|
||||
const MarkReadSchema = z.object({
|
||||
chatId: z.string().min(1),
|
||||
})
|
||||
|
||||
export async function POST(request: NextRequest) {
|
||||
try {
|
||||
const { userId, isAuthenticated } = await authenticateCopilotRequestSessionOnly()
|
||||
if (!isAuthenticated || !userId) {
|
||||
return createUnauthorizedResponse()
|
||||
}
|
||||
|
||||
const body = await request.json()
|
||||
const { chatId } = MarkReadSchema.parse(body)
|
||||
|
||||
await db
|
||||
.update(copilotChats)
|
||||
.set({ lastSeenAt: new Date() })
|
||||
.where(and(eq(copilotChats.id, chatId), eq(copilotChats.userId, userId)))
|
||||
|
||||
return NextResponse.json({ success: true })
|
||||
} catch (error) {
|
||||
if (error instanceof z.ZodError) {
|
||||
return createBadRequestResponse('chatId is required')
|
||||
}
|
||||
logger.error('Error marking task as read:', error)
|
||||
return createInternalServerErrorResponse('Failed to mark task as read')
|
||||
}
|
||||
}
|
||||
@@ -10,6 +10,7 @@ import {
|
||||
createInternalServerErrorResponse,
|
||||
createUnauthorizedResponse,
|
||||
} from '@/lib/copilot/request-helpers'
|
||||
import { taskPubSub } from '@/lib/copilot/task-events'
|
||||
|
||||
const logger = createLogger('MothershipChatsAPI')
|
||||
|
||||
@@ -34,6 +35,8 @@ export async function GET(request: NextRequest) {
|
||||
id: copilotChats.id,
|
||||
title: copilotChats.title,
|
||||
updatedAt: copilotChats.updatedAt,
|
||||
conversationId: copilotChats.conversationId,
|
||||
lastSeenAt: copilotChats.lastSeenAt,
|
||||
})
|
||||
.from(copilotChats)
|
||||
.where(
|
||||
@@ -70,6 +73,7 @@ export async function POST(request: NextRequest) {
|
||||
const body = await request.json()
|
||||
const { workspaceId } = CreateChatSchema.parse(body)
|
||||
|
||||
const now = new Date()
|
||||
const [chat] = await db
|
||||
.insert(copilotChats)
|
||||
.values({
|
||||
@@ -79,9 +83,13 @@ export async function POST(request: NextRequest) {
|
||||
title: null,
|
||||
model: 'claude-opus-4-5',
|
||||
messages: [],
|
||||
updatedAt: now,
|
||||
lastSeenAt: now,
|
||||
})
|
||||
.returning({ id: copilotChats.id })
|
||||
|
||||
taskPubSub?.publishStatusChanged({ workspaceId, chatId: chat.id, type: 'created' })
|
||||
|
||||
return NextResponse.json({ success: true, id: chat.id })
|
||||
} catch (error) {
|
||||
if (error instanceof z.ZodError) {
|
||||
|
||||
32
apps/sim/app/api/mothership/events/route.ts
Normal file
32
apps/sim/app/api/mothership/events/route.ts
Normal file
@@ -0,0 +1,32 @@
|
||||
/**
|
||||
* SSE endpoint for task status events.
|
||||
*
|
||||
* Pushes `task_status` events to the browser when tasks are
|
||||
* started, completed, created, deleted, or renamed.
|
||||
*
|
||||
* Auth is handled via session cookies (EventSource sends cookies automatically).
|
||||
*/
|
||||
|
||||
import { taskPubSub } from '@/lib/copilot/task-events'
|
||||
import { createWorkspaceSSE } from '@/lib/events/sse-endpoint'
|
||||
|
||||
export const dynamic = 'force-dynamic'
|
||||
|
||||
export const GET = createWorkspaceSSE({
|
||||
label: 'mothership-events',
|
||||
subscriptions: [
|
||||
{
|
||||
subscribe: (workspaceId, send) => {
|
||||
if (!taskPubSub) return () => {}
|
||||
return taskPubSub.onStatusChanged((event) => {
|
||||
if (event.workspaceId !== workspaceId) return
|
||||
send('task_status', {
|
||||
chatId: event.chatId,
|
||||
type: event.type,
|
||||
timestamp: Date.now(),
|
||||
})
|
||||
})
|
||||
},
|
||||
},
|
||||
],
|
||||
})
|
||||
@@ -16,6 +16,13 @@ const PLACEHOLDER_MOBILE = 'Enter a message'
|
||||
const PLACEHOLDER_DESKTOP = 'Enter a message or click the mic to speak'
|
||||
const MAX_TEXTAREA_HEIGHT = 120 // Max height in pixels (e.g., for about 3-4 lines)
|
||||
const MAX_TEXTAREA_HEIGHT_MOBILE = 100 // Smaller for mobile
|
||||
const IS_STT_AVAILABLE =
|
||||
typeof window !== 'undefined' &&
|
||||
!!(
|
||||
(window as Window & { SpeechRecognition?: unknown; webkitSpeechRecognition?: unknown })
|
||||
.SpeechRecognition ||
|
||||
(window as Window & { webkitSpeechRecognition?: unknown }).webkitSpeechRecognition
|
||||
)
|
||||
|
||||
interface AttachedFile {
|
||||
id: string
|
||||
@@ -43,13 +50,6 @@ export const ChatInput: React.FC<{
|
||||
const [dragCounter, setDragCounter] = useState(0)
|
||||
const isDragOver = dragCounter > 0
|
||||
|
||||
// Check if speech-to-text is available in the browser
|
||||
const isSttAvailable = (() => {
|
||||
if (typeof window === 'undefined') return false
|
||||
const w = window as Window & { SpeechRecognition?: unknown; webkitSpeechRecognition?: unknown }
|
||||
return !!(w.SpeechRecognition || w.webkitSpeechRecognition)
|
||||
})()
|
||||
|
||||
// Function to adjust textarea height
|
||||
const adjustTextareaHeight = () => {
|
||||
if (textareaRef.current) {
|
||||
@@ -199,7 +199,7 @@ export const ChatInput: React.FC<{
|
||||
<Tooltip.Provider>
|
||||
<div className='flex items-center justify-center'>
|
||||
{/* Voice Input Only */}
|
||||
{isSttAvailable && (
|
||||
{IS_STT_AVAILABLE && (
|
||||
<Tooltip.Root>
|
||||
<Tooltip.Trigger asChild>
|
||||
<div>
|
||||
@@ -410,7 +410,7 @@ export const ChatInput: React.FC<{
|
||||
</div>
|
||||
|
||||
{/* Voice Input */}
|
||||
{isSttAvailable && (
|
||||
{IS_STT_AVAILABLE && (
|
||||
<Tooltip.Root>
|
||||
<Tooltip.Trigger asChild>
|
||||
<div>
|
||||
|
||||
@@ -1,12 +1,6 @@
|
||||
'use client'
|
||||
|
||||
import {
|
||||
type ElementType,
|
||||
type ReactNode,
|
||||
type RefCallback,
|
||||
type SVGProps,
|
||||
useCallback,
|
||||
} from 'react'
|
||||
import { type ElementType, type ReactNode, type SVGProps, useEffect, useRef } from 'react'
|
||||
import { Button, Tooltip } from '@/components/emcn'
|
||||
import { BookOpen, PanelLeft, Table as TableIcon } from '@/components/emcn/icons'
|
||||
import { WorkflowIcon } from '@/components/icons'
|
||||
@@ -83,7 +77,10 @@ export function ResourceTabs({
|
||||
onCyclePreviewMode,
|
||||
actions,
|
||||
}: ResourceTabsProps) {
|
||||
const scrollRef = useCallback<RefCallback<HTMLDivElement>>((node) => {
|
||||
const scrollNodeRef = useRef<HTMLDivElement>(null)
|
||||
|
||||
useEffect(() => {
|
||||
const node = scrollNodeRef.current
|
||||
if (!node) return
|
||||
const handler = (e: WheelEvent) => {
|
||||
if (e.deltaY !== 0) {
|
||||
@@ -113,7 +110,7 @@ export function ResourceTabs({
|
||||
</Tooltip.Content>
|
||||
</Tooltip.Root>
|
||||
<div
|
||||
ref={scrollRef}
|
||||
ref={scrollNodeRef}
|
||||
className='mx-[2px] flex min-w-0 items-center gap-[6px] overflow-x-auto px-[6px] [scrollbar-width:none] [&::-webkit-scrollbar]:hidden'
|
||||
>
|
||||
{resources.map((resource) => {
|
||||
|
||||
@@ -144,6 +144,12 @@ export function UserInput({
|
||||
wasSendingRef.current = isSending
|
||||
}, [isSending])
|
||||
|
||||
useEffect(() => {
|
||||
if (isInitialView) {
|
||||
textareaRef.current?.focus()
|
||||
}
|
||||
}, [isInitialView])
|
||||
|
||||
const handleContainerClick = useCallback((e: React.MouseEvent<HTMLDivElement>) => {
|
||||
if ((e.target as HTMLElement).closest('button')) return
|
||||
textareaRef.current?.focus()
|
||||
|
||||
@@ -14,7 +14,7 @@ import {
|
||||
LandingWorkflowSeedStorage,
|
||||
} from '@/lib/core/utils/browser-storage'
|
||||
import { persistImportedWorkflow } from '@/lib/workflows/operations/import-export'
|
||||
import { useChatHistory } from '@/hooks/queries/tasks'
|
||||
import { useChatHistory, useMarkTaskRead } from '@/hooks/queries/tasks'
|
||||
import { MessageContent, MothershipView, TemplatePrompts, UserInput } from './components'
|
||||
import type { FileAttachmentForApi } from './components/user-input/user-input'
|
||||
import { useAutoScroll, useChat } from './hooks'
|
||||
@@ -158,19 +158,35 @@ export function Home({ chatId }: HomeProps = {}) {
|
||||
}
|
||||
}, [createWorkflowFromLandingSeed, workspaceId, router])
|
||||
|
||||
const wasSendingRef = useRef(false)
|
||||
|
||||
const { isLoading: isLoadingHistory } = useChatHistory(chatId)
|
||||
const { mutate: markRead } = useMarkTaskRead(workspaceId)
|
||||
|
||||
const {
|
||||
messages,
|
||||
isSending,
|
||||
sendMessage,
|
||||
stopGeneration,
|
||||
resolvedChatId,
|
||||
resources,
|
||||
isResourceCleanupSettled,
|
||||
activeResourceId,
|
||||
setActiveResourceId,
|
||||
} = useChat(workspaceId, chatId)
|
||||
|
||||
useEffect(() => {
|
||||
wasSendingRef.current = false
|
||||
if (resolvedChatId) markRead(resolvedChatId)
|
||||
}, [resolvedChatId, markRead])
|
||||
|
||||
useEffect(() => {
|
||||
if (wasSendingRef.current && !isSending && resolvedChatId) {
|
||||
markRead(resolvedChatId)
|
||||
}
|
||||
wasSendingRef.current = isSending
|
||||
}, [isSending, resolvedChatId, markRead])
|
||||
|
||||
const [isResourceCollapsed, setIsResourceCollapsed] = useState(false)
|
||||
const [showExpandButton, setShowExpandButton] = useState(false)
|
||||
const [isResourceAnimatingIn, setIsResourceAnimatingIn] = useState(false)
|
||||
|
||||
@@ -56,6 +56,7 @@ export interface UseChatReturn {
|
||||
messages: ChatMessage[]
|
||||
isSending: boolean
|
||||
error: string | null
|
||||
resolvedChatId: string | undefined
|
||||
sendMessage: (message: string, fileAttachments?: FileAttachmentForApi[]) => Promise<void>
|
||||
stopGeneration: () => Promise<void>
|
||||
resources: MothershipResource[]
|
||||
@@ -248,6 +249,7 @@ export function useChat(workspaceId: string, initialChatId?: string): UseChatRet
|
||||
const [messages, setMessages] = useState<ChatMessage[]>([])
|
||||
const [isSending, setIsSending] = useState(false)
|
||||
const [error, setError] = useState<string | null>(null)
|
||||
const [resolvedChatId, setResolvedChatId] = useState<string | undefined>(initialChatId)
|
||||
const [resources, setResources] = useState<MothershipResource[]>([])
|
||||
const [activeResourceId, setActiveResourceId] = useState<string | null>(null)
|
||||
const abortControllerRef = useRef<AbortController | null>(null)
|
||||
@@ -352,9 +354,11 @@ export function useChat(workspaceId: string, initialChatId?: string): UseChatRet
|
||||
useEffect(() => {
|
||||
if (sendingRef.current) {
|
||||
chatIdRef.current = initialChatId
|
||||
setResolvedChatId(initialChatId)
|
||||
return
|
||||
}
|
||||
chatIdRef.current = initialChatId
|
||||
setResolvedChatId(initialChatId)
|
||||
appliedChatIdRef.current = undefined
|
||||
setMessages([])
|
||||
setError(null)
|
||||
@@ -370,6 +374,7 @@ export function useChat(workspaceId: string, initialChatId?: string): UseChatRet
|
||||
if (!isHomePage || !chatIdRef.current) return
|
||||
streamGenRef.current++
|
||||
chatIdRef.current = undefined
|
||||
setResolvedChatId(undefined)
|
||||
appliedChatIdRef.current = undefined
|
||||
abortControllerRef.current?.abort()
|
||||
abortControllerRef.current = null
|
||||
@@ -506,6 +511,7 @@ export function useChat(workspaceId: string, initialChatId?: string): UseChatRet
|
||||
if (parsed.chatId) {
|
||||
const isNewChat = !chatIdRef.current
|
||||
chatIdRef.current = parsed.chatId
|
||||
setResolvedChatId(parsed.chatId)
|
||||
queryClient.invalidateQueries({ queryKey: taskKeys.list(workspaceId) })
|
||||
if (isNewChat) {
|
||||
const userMsg = pendingUserMsgRef.current
|
||||
@@ -1071,6 +1077,7 @@ export function useChat(workspaceId: string, initialChatId?: string): UseChatRet
|
||||
messages,
|
||||
isSending,
|
||||
error,
|
||||
resolvedChatId,
|
||||
sendMessage,
|
||||
stopGeneration,
|
||||
resources,
|
||||
|
||||
@@ -63,6 +63,7 @@ import {
|
||||
import { useDeleteTask, useDeleteTasks, useRenameTask, useTasks } from '@/hooks/queries/tasks'
|
||||
import { usePermissionConfig } from '@/hooks/use-permission-config'
|
||||
import { useSettingsNavigation } from '@/hooks/use-settings-navigation'
|
||||
import { useTaskEvents } from '@/hooks/use-task-events'
|
||||
import { SIDEBAR_WIDTH } from '@/stores/constants'
|
||||
import { useFolderStore } from '@/stores/folders/store'
|
||||
import { useSearchModalStore } from '@/stores/modals/search/store'
|
||||
@@ -70,6 +71,22 @@ import { useSidebarStore } from '@/stores/sidebar/store'
|
||||
|
||||
const logger = createLogger('Sidebar')
|
||||
|
||||
type TaskStatus = 'running' | 'unread' | 'idle'
|
||||
|
||||
function TaskStatusIcon({ status }: { status: TaskStatus }) {
|
||||
return (
|
||||
<div className='relative h-[16px] w-[16px] flex-shrink-0'>
|
||||
<Blimp className='h-[16px] w-[16px] text-[var(--text-icon)]' />
|
||||
{status === 'running' && (
|
||||
<span className='-bottom-[1px] -right-[1px] absolute h-[7px] w-[7px] animate-pulse rounded-full border border-[var(--surface-1)] bg-[var(--brand-tertiary-2)]' />
|
||||
)}
|
||||
{status === 'unread' && (
|
||||
<span className='-bottom-[1px] -right-[1px] absolute h-[7px] w-[7px] rounded-full border border-[var(--surface-1)] bg-[var(--brand-tertiary)]' />
|
||||
)}
|
||||
</div>
|
||||
)
|
||||
}
|
||||
|
||||
function SidebarItemSkeleton() {
|
||||
return (
|
||||
<div className='mx-[2px] flex h-[30px] items-center px-[8px]'>
|
||||
@@ -80,15 +97,17 @@ function SidebarItemSkeleton() {
|
||||
|
||||
const SidebarTaskItem = memo(function SidebarTaskItem({
|
||||
task,
|
||||
active,
|
||||
isCurrentRoute,
|
||||
isSelected,
|
||||
status,
|
||||
showCollapsedContent,
|
||||
onMultiSelectClick,
|
||||
onContextMenu,
|
||||
}: {
|
||||
task: { id: string; href: string; name: string }
|
||||
active: boolean
|
||||
isCurrentRoute: boolean
|
||||
isSelected: boolean
|
||||
status: TaskStatus
|
||||
showCollapsedContent: boolean
|
||||
onMultiSelectClick: (taskId: string, shiftKey: boolean, metaKey: boolean) => void
|
||||
onContextMenu: (e: React.MouseEvent, taskId: string) => void
|
||||
@@ -100,7 +119,7 @@ const SidebarTaskItem = memo(function SidebarTaskItem({
|
||||
href={task.href}
|
||||
className={cn(
|
||||
'mx-[2px] flex h-[30px] items-center gap-[8px] rounded-[8px] px-[8px] text-[14px] hover:bg-[var(--surface-active)]',
|
||||
(active || isSelected) && 'bg-[var(--surface-active)]'
|
||||
(isCurrentRoute || isSelected) && 'bg-[var(--surface-active)]'
|
||||
)}
|
||||
onClick={(e) => {
|
||||
if (task.id === 'new') return
|
||||
@@ -108,12 +127,15 @@ const SidebarTaskItem = memo(function SidebarTaskItem({
|
||||
e.preventDefault()
|
||||
onMultiSelectClick(task.id, e.shiftKey, e.metaKey || e.ctrlKey)
|
||||
} else {
|
||||
useFolderStore.getState().clearTaskSelection()
|
||||
useFolderStore.setState({
|
||||
selectedTasks: new Set<string>(),
|
||||
lastSelectedTaskId: task.id,
|
||||
})
|
||||
}
|
||||
}}
|
||||
onContextMenu={task.id !== 'new' ? (e) => onContextMenu(e, task.id) : undefined}
|
||||
>
|
||||
<Blimp className='h-[16px] w-[16px] flex-shrink-0 text-[var(--text-icon)]' />
|
||||
<TaskStatusIcon status={status} />
|
||||
<div className='min-w-0 truncate font-[var(--sidebar-font-weight)] text-[var(--text-body)]'>
|
||||
{task.name}
|
||||
</div>
|
||||
@@ -507,6 +529,8 @@ export const Sidebar = memo(function Sidebar() {
|
||||
|
||||
const { data: fetchedTasks = [], isLoading: tasksLoading } = useTasks(workspaceId)
|
||||
|
||||
useTaskEvents(workspaceId)
|
||||
|
||||
const tasks = useMemo(
|
||||
() =>
|
||||
fetchedTasks.length > 0
|
||||
@@ -514,7 +538,15 @@ export const Sidebar = memo(function Sidebar() {
|
||||
...t,
|
||||
href: `/workspace/${workspaceId}/task/${t.id}`,
|
||||
}))
|
||||
: [{ id: 'new', name: 'New task', href: `/workspace/${workspaceId}/home` }],
|
||||
: [
|
||||
{
|
||||
id: 'new',
|
||||
name: 'New task',
|
||||
href: `/workspace/${workspaceId}/home`,
|
||||
isActive: false,
|
||||
isUnread: false,
|
||||
},
|
||||
],
|
||||
[fetchedTasks, workspaceId]
|
||||
)
|
||||
|
||||
@@ -1018,9 +1050,16 @@ export const Sidebar = memo(function Sidebar() {
|
||||
) : (
|
||||
<>
|
||||
{tasks.slice(0, visibleTaskCount).map((task) => {
|
||||
const active = task.id !== 'new' && pathname === task.href
|
||||
const isCurrentRoute = task.id !== 'new' && pathname === task.href
|
||||
const isRenaming = renamingTaskId === task.id
|
||||
const isSelected = task.id !== 'new' && selectedTasks.has(task.id)
|
||||
const status: TaskStatus = isCurrentRoute
|
||||
? 'idle'
|
||||
: task.isActive
|
||||
? 'running'
|
||||
: task.isUnread
|
||||
? 'unread'
|
||||
: 'idle'
|
||||
|
||||
if (!isCollapsed && isRenaming) {
|
||||
return (
|
||||
@@ -1028,7 +1067,7 @@ export const Sidebar = memo(function Sidebar() {
|
||||
key={task.id}
|
||||
className='mx-[2px] flex h-[30px] items-center gap-[8px] rounded-[8px] bg-[var(--surface-active)] px-[8px] text-[14px]'
|
||||
>
|
||||
<Blimp className='h-[16px] w-[16px] flex-shrink-0 text-[var(--text-icon)]' />
|
||||
<TaskStatusIcon status={status} />
|
||||
<input
|
||||
ref={renameInputRef}
|
||||
value={renameValue}
|
||||
@@ -1045,8 +1084,9 @@ export const Sidebar = memo(function Sidebar() {
|
||||
<SidebarTaskItem
|
||||
key={task.id}
|
||||
task={task}
|
||||
active={active}
|
||||
isCurrentRoute={isCurrentRoute}
|
||||
isSelected={isSelected}
|
||||
status={status}
|
||||
showCollapsedContent={showCollapsedContent}
|
||||
onMultiSelectClick={handleTaskClick}
|
||||
onContextMenu={handleTaskContextMenu}
|
||||
|
||||
@@ -4,6 +4,8 @@ export interface TaskMetadata {
|
||||
id: string
|
||||
name: string
|
||||
updatedAt: Date
|
||||
isActive: boolean
|
||||
isUnread: boolean
|
||||
}
|
||||
|
||||
export interface TaskChatHistory {
|
||||
@@ -65,13 +67,20 @@ interface TaskResponse {
|
||||
id: string
|
||||
title: string | null
|
||||
updatedAt: string
|
||||
conversationId: string | null
|
||||
lastSeenAt: string | null
|
||||
}
|
||||
|
||||
function mapTask(chat: TaskResponse): TaskMetadata {
|
||||
const updatedAt = new Date(chat.updatedAt)
|
||||
return {
|
||||
id: chat.id,
|
||||
name: chat.title ?? 'New task',
|
||||
updatedAt: new Date(chat.updatedAt),
|
||||
updatedAt,
|
||||
isActive: chat.conversationId !== null,
|
||||
isUnread:
|
||||
chat.conversationId === null &&
|
||||
(chat.lastSeenAt === null || updatedAt > new Date(chat.lastSeenAt)),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -207,3 +216,43 @@ export function useRenameTask(workspaceId?: string) {
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
async function markTaskRead(chatId: string): Promise<void> {
|
||||
const response = await fetch('/api/mothership/chats/read', {
|
||||
method: 'POST',
|
||||
headers: { 'Content-Type': 'application/json' },
|
||||
body: JSON.stringify({ chatId }),
|
||||
})
|
||||
if (!response.ok) {
|
||||
throw new Error('Failed to mark task as read')
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Marks a task as read with optimistic update.
|
||||
*/
|
||||
export function useMarkTaskRead(workspaceId?: string) {
|
||||
const queryClient = useQueryClient()
|
||||
return useMutation({
|
||||
mutationFn: markTaskRead,
|
||||
onMutate: async (chatId) => {
|
||||
await queryClient.cancelQueries({ queryKey: taskKeys.list(workspaceId) })
|
||||
|
||||
const previousTasks = queryClient.getQueryData<TaskMetadata[]>(taskKeys.list(workspaceId))
|
||||
|
||||
queryClient.setQueryData<TaskMetadata[]>(taskKeys.list(workspaceId), (old) =>
|
||||
old?.map((task) => (task.id === chatId ? { ...task, isUnread: false } : task))
|
||||
)
|
||||
|
||||
return { previousTasks }
|
||||
},
|
||||
onError: (_err, _variables, context) => {
|
||||
if (context?.previousTasks) {
|
||||
queryClient.setQueryData(taskKeys.list(workspaceId), context.previousTasks)
|
||||
}
|
||||
},
|
||||
onSettled: () => {
|
||||
queryClient.invalidateQueries({ queryKey: taskKeys.list(workspaceId) })
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
33
apps/sim/hooks/use-task-events.ts
Normal file
33
apps/sim/hooks/use-task-events.ts
Normal file
@@ -0,0 +1,33 @@
|
||||
import { useEffect } from 'react'
|
||||
import { createLogger } from '@sim/logger'
|
||||
import { useQueryClient } from '@tanstack/react-query'
|
||||
import { taskKeys } from '@/hooks/queries/tasks'
|
||||
|
||||
const logger = createLogger('TaskEvents')
|
||||
|
||||
/**
|
||||
* Subscribes to task status SSE events and invalidates the task list on changes.
|
||||
*/
|
||||
export function useTaskEvents(workspaceId: string | undefined) {
|
||||
const queryClient = useQueryClient()
|
||||
|
||||
useEffect(() => {
|
||||
if (!workspaceId) return
|
||||
|
||||
const eventSource = new EventSource(
|
||||
`/api/mothership/events?workspaceId=${encodeURIComponent(workspaceId)}`
|
||||
)
|
||||
|
||||
eventSource.addEventListener('task_status', () => {
|
||||
queryClient.invalidateQueries({ queryKey: taskKeys.lists() })
|
||||
})
|
||||
|
||||
eventSource.onerror = () => {
|
||||
logger.warn(`SSE connection error for workspace ${workspaceId}`)
|
||||
}
|
||||
|
||||
return () => {
|
||||
eventSource.close()
|
||||
}
|
||||
}, [workspaceId, queryClient])
|
||||
}
|
||||
@@ -10,7 +10,9 @@ import {
|
||||
resetStreamBuffer,
|
||||
setStreamMeta,
|
||||
} from '@/lib/copilot/orchestrator/stream/buffer'
|
||||
import { taskPubSub } from '@/lib/copilot/task-events'
|
||||
import { env } from '@/lib/core/config/env'
|
||||
import { SSE_HEADERS } from '@/lib/core/utils/sse'
|
||||
|
||||
const logger = createLogger('CopilotChatStreaming')
|
||||
|
||||
@@ -85,6 +87,7 @@ export interface StreamingOrchestrationParams {
|
||||
titleModel: string
|
||||
titleProvider?: string
|
||||
requestId: string
|
||||
workspaceId?: string
|
||||
orchestrateOptions: Omit<OrchestrateStreamOptions, 'onEvent'>
|
||||
}
|
||||
|
||||
@@ -100,6 +103,7 @@ export function createSSEStream(params: StreamingOrchestrationParams): ReadableS
|
||||
titleModel,
|
||||
titleProvider,
|
||||
requestId,
|
||||
workspaceId,
|
||||
orchestrateOptions,
|
||||
} = params
|
||||
|
||||
@@ -157,6 +161,9 @@ export function createSSEStream(params: StreamingOrchestrationParams): ReadableS
|
||||
if (title) {
|
||||
await db.update(copilotChats).set({ title }).where(eq(copilotChats.id, chatId!))
|
||||
await pushEvent({ type: 'title_updated', title })
|
||||
if (workspaceId) {
|
||||
taskPubSub?.publishStatusChanged({ workspaceId, chatId: chatId!, type: 'renamed' })
|
||||
}
|
||||
}
|
||||
})
|
||||
.catch((error) => {
|
||||
@@ -220,9 +227,6 @@ export function createSSEStream(params: StreamingOrchestrationParams): ReadableS
|
||||
}
|
||||
|
||||
export const SSE_RESPONSE_HEADERS = {
|
||||
'Content-Type': 'text/event-stream',
|
||||
...SSE_HEADERS,
|
||||
'Content-Encoding': 'none',
|
||||
'Cache-Control': 'no-cache',
|
||||
Connection: 'keep-alive',
|
||||
'X-Accel-Buffering': 'no',
|
||||
} as const
|
||||
|
||||
29
apps/sim/lib/copilot/task-events.ts
Normal file
29
apps/sim/lib/copilot/task-events.ts
Normal file
@@ -0,0 +1,29 @@
|
||||
/**
|
||||
* Task Status Pub/Sub Adapter
|
||||
*
|
||||
* Broadcasts task status events across processes using Redis Pub/Sub.
|
||||
* Gracefully falls back to process-local EventEmitter when Redis is unavailable.
|
||||
*
|
||||
* Channel: `task:status_changed`
|
||||
*/
|
||||
|
||||
import { createPubSubChannel } from '@/lib/events/pubsub'
|
||||
|
||||
export interface TaskStatusEvent {
|
||||
workspaceId: string
|
||||
chatId: string
|
||||
type: 'started' | 'completed' | 'created' | 'deleted' | 'renamed'
|
||||
}
|
||||
|
||||
const channel =
|
||||
typeof window !== 'undefined'
|
||||
? null
|
||||
: createPubSubChannel<TaskStatusEvent>({ channel: 'task:status_changed', label: 'task' })
|
||||
|
||||
export const taskPubSub = channel
|
||||
? {
|
||||
publishStatusChanged: (event: TaskStatusEvent) => channel.publish(event),
|
||||
onStatusChanged: (handler: (event: TaskStatusEvent) => void) => channel.subscribe(handler),
|
||||
dispose: () => channel.dispose(),
|
||||
}
|
||||
: null
|
||||
154
apps/sim/lib/events/pubsub.ts
Normal file
154
apps/sim/lib/events/pubsub.ts
Normal file
@@ -0,0 +1,154 @@
|
||||
/**
|
||||
* Generic Pub/Sub Channel Factory
|
||||
*
|
||||
* Creates a single-channel pub/sub adapter backed by Redis (with EventEmitter fallback).
|
||||
* Each call creates its own Redis connections — use multiple instances for multiple channels.
|
||||
*/
|
||||
|
||||
import { EventEmitter } from 'events'
|
||||
import { createLogger } from '@sim/logger'
|
||||
import Redis from 'ioredis'
|
||||
import { env } from '@/lib/core/config/env'
|
||||
|
||||
const logger = createLogger('PubSub')
|
||||
|
||||
export interface PubSubChannel<T> {
|
||||
publish(event: T): void
|
||||
subscribe(handler: (event: T) => void): () => void
|
||||
dispose(): void
|
||||
}
|
||||
|
||||
interface PubSubChannelConfig {
|
||||
channel: string
|
||||
label: string
|
||||
}
|
||||
|
||||
class RedisPubSubChannel<T> implements PubSubChannel<T> {
|
||||
private pub: Redis
|
||||
private sub: Redis
|
||||
private handlers = new Set<(event: T) => void>()
|
||||
private disposed = false
|
||||
|
||||
constructor(
|
||||
redisUrl: string,
|
||||
private config: PubSubChannelConfig
|
||||
) {
|
||||
const commonOpts = {
|
||||
keepAlive: 1000,
|
||||
connectTimeout: 10000,
|
||||
maxRetriesPerRequest: null as unknown as number,
|
||||
enableOfflineQueue: true,
|
||||
retryStrategy: (times: number) => {
|
||||
if (times > 10) return 30000
|
||||
return Math.min(times * 500, 5000)
|
||||
},
|
||||
}
|
||||
|
||||
this.pub = new Redis(redisUrl, { ...commonOpts, connectionName: `${config.label}-pub` })
|
||||
this.sub = new Redis(redisUrl, { ...commonOpts, connectionName: `${config.label}-sub` })
|
||||
|
||||
this.pub.on('error', (err) =>
|
||||
logger.error(`${config.label} publish client error:`, err.message)
|
||||
)
|
||||
this.sub.on('error', (err) =>
|
||||
logger.error(`${config.label} subscribe client error:`, err.message)
|
||||
)
|
||||
this.pub.on('connect', () => logger.info(`${config.label} publish client connected`))
|
||||
this.sub.on('connect', () => logger.info(`${config.label} subscribe client connected`))
|
||||
|
||||
this.sub.subscribe(config.channel, (err) => {
|
||||
if (err) {
|
||||
logger.error(`Failed to subscribe to ${config.label} channel:`, err)
|
||||
} else {
|
||||
logger.info(`Subscribed to ${config.label} channel`)
|
||||
}
|
||||
})
|
||||
|
||||
this.sub.on('message', (channel: string, message: string) => {
|
||||
if (channel !== config.channel) return
|
||||
try {
|
||||
const parsed = JSON.parse(message) as T
|
||||
for (const handler of this.handlers) {
|
||||
try {
|
||||
handler(parsed)
|
||||
} catch (err) {
|
||||
logger.error(`Error in ${config.label} handler:`, err)
|
||||
}
|
||||
}
|
||||
} catch (err) {
|
||||
logger.error(`Failed to parse ${config.label} message:`, err)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
publish(event: T): void {
|
||||
if (this.disposed) return
|
||||
this.pub.publish(this.config.channel, JSON.stringify(event)).catch((err) => {
|
||||
logger.error(`Failed to publish to ${this.config.label}:`, err)
|
||||
})
|
||||
}
|
||||
|
||||
subscribe(handler: (event: T) => void): () => void {
|
||||
this.handlers.add(handler)
|
||||
return () => {
|
||||
this.handlers.delete(handler)
|
||||
}
|
||||
}
|
||||
|
||||
dispose(): void {
|
||||
this.disposed = true
|
||||
this.handlers.clear()
|
||||
|
||||
const noop = () => {}
|
||||
this.pub.removeAllListeners()
|
||||
this.sub.removeAllListeners()
|
||||
this.pub.on('error', noop)
|
||||
this.sub.on('error', noop)
|
||||
|
||||
this.sub.unsubscribe().catch(noop)
|
||||
this.pub.quit().catch(noop)
|
||||
this.sub.quit().catch(noop)
|
||||
logger.info(`${this.config.label} Redis pub/sub disposed`)
|
||||
}
|
||||
}
|
||||
|
||||
class LocalPubSubChannel<T> implements PubSubChannel<T> {
|
||||
private emitter = new EventEmitter()
|
||||
|
||||
constructor(private config: PubSubChannelConfig) {
|
||||
this.emitter.setMaxListeners(100)
|
||||
logger.info(`${config.label}: Using process-local EventEmitter (Redis not configured)`)
|
||||
}
|
||||
|
||||
publish(event: T): void {
|
||||
this.emitter.emit(this.config.channel, event)
|
||||
}
|
||||
|
||||
subscribe(handler: (event: T) => void): () => void {
|
||||
this.emitter.on(this.config.channel, handler)
|
||||
return () => {
|
||||
this.emitter.off(this.config.channel, handler)
|
||||
}
|
||||
}
|
||||
|
||||
dispose(): void {
|
||||
this.emitter.removeAllListeners()
|
||||
logger.info(`${this.config.label} local pub/sub disposed`)
|
||||
}
|
||||
}
|
||||
|
||||
export function createPubSubChannel<T>(config: PubSubChannelConfig): PubSubChannel<T> {
|
||||
const redisUrl = env.REDIS_URL
|
||||
|
||||
if (redisUrl) {
|
||||
try {
|
||||
logger.info(`${config.label}: Using Redis`)
|
||||
return new RedisPubSubChannel<T>(redisUrl, config)
|
||||
} catch (err) {
|
||||
logger.error(`Failed to create Redis ${config.label}, falling back to local:`, err)
|
||||
return new LocalPubSubChannel<T>(config)
|
||||
}
|
||||
}
|
||||
|
||||
return new LocalPubSubChannel<T>(config)
|
||||
}
|
||||
118
apps/sim/lib/events/sse-endpoint.ts
Normal file
118
apps/sim/lib/events/sse-endpoint.ts
Normal file
@@ -0,0 +1,118 @@
|
||||
/**
|
||||
* Generic Workspace SSE Endpoint Factory
|
||||
*
|
||||
* Creates a GET handler that authenticates the user, verifies workspace access,
|
||||
* and streams Server-Sent Events with heartbeats and cleanup.
|
||||
*/
|
||||
|
||||
import { createLogger } from '@sim/logger'
|
||||
import type { NextRequest } from 'next/server'
|
||||
import { getSession } from '@/lib/auth'
|
||||
import { SSE_HEADERS } from '@/lib/core/utils/sse'
|
||||
import { decrementSSEConnections, incrementSSEConnections } from '@/lib/monitoring/sse-connections'
|
||||
import { getUserEntityPermissions } from '@/lib/workspaces/permissions/utils'
|
||||
|
||||
export interface SSESubscription {
|
||||
subscribe(
|
||||
workspaceId: string,
|
||||
send: (eventName: string, data: Record<string, unknown>) => void
|
||||
): () => void
|
||||
}
|
||||
|
||||
interface WorkspaceSSEConfig {
|
||||
label: string
|
||||
subscriptions: SSESubscription[]
|
||||
}
|
||||
|
||||
const HEARTBEAT_INTERVAL_MS = 30_000
|
||||
|
||||
export function createWorkspaceSSE(config: WorkspaceSSEConfig) {
|
||||
const logger = createLogger(`${config.label}-SSE`)
|
||||
|
||||
return async function GET(request: NextRequest): Promise<Response> {
|
||||
const session = await getSession()
|
||||
if (!session?.user?.id) {
|
||||
return new Response('Unauthorized', { status: 401 })
|
||||
}
|
||||
|
||||
const { searchParams } = new URL(request.url)
|
||||
const workspaceId = searchParams.get('workspaceId')
|
||||
if (!workspaceId) {
|
||||
return new Response('Missing workspaceId query parameter', { status: 400 })
|
||||
}
|
||||
|
||||
const permissions = await getUserEntityPermissions(session.user.id, 'workspace', workspaceId)
|
||||
if (!permissions) {
|
||||
return new Response('Access denied to workspace', { status: 403 })
|
||||
}
|
||||
|
||||
const encoder = new TextEncoder()
|
||||
const unsubscribers: Array<() => void> = []
|
||||
let cleaned = false
|
||||
|
||||
const cleanup = () => {
|
||||
if (cleaned) return
|
||||
cleaned = true
|
||||
for (const unsub of unsubscribers) {
|
||||
unsub()
|
||||
}
|
||||
decrementSSEConnections(config.label)
|
||||
logger.info(`SSE connection closed for workspace ${workspaceId}`)
|
||||
}
|
||||
|
||||
const stream = new ReadableStream({
|
||||
start(controller) {
|
||||
incrementSSEConnections(config.label)
|
||||
|
||||
const send = (eventName: string, data: Record<string, unknown>) => {
|
||||
if (cleaned) return
|
||||
try {
|
||||
controller.enqueue(
|
||||
encoder.encode(`event: ${eventName}\ndata: ${JSON.stringify(data)}\n\n`)
|
||||
)
|
||||
} catch {
|
||||
// Stream already closed
|
||||
}
|
||||
}
|
||||
|
||||
for (const subscription of config.subscriptions) {
|
||||
const unsub = subscription.subscribe(workspaceId, send)
|
||||
unsubscribers.push(unsub)
|
||||
}
|
||||
|
||||
const heartbeat = setInterval(() => {
|
||||
if (cleaned) {
|
||||
clearInterval(heartbeat)
|
||||
return
|
||||
}
|
||||
try {
|
||||
controller.enqueue(encoder.encode(': heartbeat\n\n'))
|
||||
} catch {
|
||||
clearInterval(heartbeat)
|
||||
}
|
||||
}, HEARTBEAT_INTERVAL_MS)
|
||||
unsubscribers.push(() => clearInterval(heartbeat))
|
||||
|
||||
request.signal.addEventListener(
|
||||
'abort',
|
||||
() => {
|
||||
cleanup()
|
||||
try {
|
||||
controller.close()
|
||||
} catch {
|
||||
// Already closed
|
||||
}
|
||||
},
|
||||
{ once: true }
|
||||
)
|
||||
|
||||
logger.info(`SSE connection opened for workspace ${workspaceId}`)
|
||||
},
|
||||
cancel() {
|
||||
cleanup()
|
||||
},
|
||||
})
|
||||
|
||||
return new Response(stream, { headers: SSE_HEADERS })
|
||||
}
|
||||
}
|
||||
@@ -11,197 +11,43 @@
|
||||
* (published by serve route, consumed by serve route on other processes to push to local SSE clients)
|
||||
*/
|
||||
|
||||
import { EventEmitter } from 'events'
|
||||
import { createLogger } from '@sim/logger'
|
||||
import Redis from 'ioredis'
|
||||
import { env } from '@/lib/core/config/env'
|
||||
import { createPubSubChannel } from '@/lib/events/pubsub'
|
||||
import type { ToolsChangedEvent, WorkflowToolsChangedEvent } from '@/lib/mcp/types'
|
||||
|
||||
const logger = createLogger('McpPubSub')
|
||||
|
||||
const CHANNEL_TOOLS_CHANGED = 'mcp:tools_changed'
|
||||
const CHANNEL_WORKFLOW_TOOLS_CHANGED = 'mcp:workflow_tools_changed'
|
||||
|
||||
type ToolsChangedHandler = (event: ToolsChangedEvent) => void
|
||||
type WorkflowToolsChangedHandler = (event: WorkflowToolsChangedEvent) => void
|
||||
|
||||
interface McpPubSubAdapter {
|
||||
publishToolsChanged(event: ToolsChangedEvent): void
|
||||
publishWorkflowToolsChanged(event: WorkflowToolsChangedEvent): void
|
||||
onToolsChanged(handler: ToolsChangedHandler): () => void
|
||||
onWorkflowToolsChanged(handler: WorkflowToolsChangedHandler): () => void
|
||||
onToolsChanged(handler: (event: ToolsChangedEvent) => void): () => void
|
||||
onWorkflowToolsChanged(handler: (event: WorkflowToolsChangedEvent) => void): () => void
|
||||
dispose(): void
|
||||
}
|
||||
|
||||
/**
|
||||
* Redis-backed pub/sub adapter.
|
||||
* Uses dedicated pub and sub clients (ioredis requires separate connections for subscribers).
|
||||
*/
|
||||
class RedisMcpPubSub implements McpPubSubAdapter {
|
||||
private pub: Redis
|
||||
private sub: Redis
|
||||
private toolsChangedHandlers = new Set<ToolsChangedHandler>()
|
||||
private workflowToolsChangedHandlers = new Set<WorkflowToolsChangedHandler>()
|
||||
private disposed = false
|
||||
const toolsChannel =
|
||||
typeof window !== 'undefined'
|
||||
? null
|
||||
: createPubSubChannel<ToolsChangedEvent>({
|
||||
channel: 'mcp:tools_changed',
|
||||
label: 'mcp-tools',
|
||||
})
|
||||
|
||||
constructor(redisUrl: string) {
|
||||
const commonOpts = {
|
||||
keepAlive: 1000,
|
||||
connectTimeout: 10000,
|
||||
maxRetriesPerRequest: null as unknown as number,
|
||||
enableOfflineQueue: true,
|
||||
retryStrategy: (times: number) => {
|
||||
if (times > 10) return 30000
|
||||
return Math.min(times * 500, 5000)
|
||||
},
|
||||
}
|
||||
|
||||
this.pub = new Redis(redisUrl, { ...commonOpts, connectionName: 'mcp-pubsub-pub' })
|
||||
this.sub = new Redis(redisUrl, { ...commonOpts, connectionName: 'mcp-pubsub-sub' })
|
||||
|
||||
this.pub.on('error', (err) => logger.error('MCP pub/sub publish client error:', err.message))
|
||||
this.sub.on('error', (err) => logger.error('MCP pub/sub subscribe client error:', err.message))
|
||||
this.pub.on('connect', () => logger.info('MCP pub/sub publish client connected'))
|
||||
this.sub.on('connect', () => logger.info('MCP pub/sub subscribe client connected'))
|
||||
|
||||
this.sub.subscribe(CHANNEL_TOOLS_CHANGED, CHANNEL_WORKFLOW_TOOLS_CHANGED, (err) => {
|
||||
if (err) {
|
||||
logger.error('Failed to subscribe to MCP pub/sub channels:', err)
|
||||
} else {
|
||||
logger.info('Subscribed to MCP pub/sub channels')
|
||||
}
|
||||
})
|
||||
|
||||
this.sub.on('message', (channel: string, message: string) => {
|
||||
try {
|
||||
const parsed = JSON.parse(message)
|
||||
if (channel === CHANNEL_TOOLS_CHANGED) {
|
||||
for (const handler of this.toolsChangedHandlers) {
|
||||
try {
|
||||
handler(parsed as ToolsChangedEvent)
|
||||
} catch (err) {
|
||||
logger.error('Error in tools_changed handler:', err)
|
||||
}
|
||||
}
|
||||
} else if (channel === CHANNEL_WORKFLOW_TOOLS_CHANGED) {
|
||||
for (const handler of this.workflowToolsChangedHandlers) {
|
||||
try {
|
||||
handler(parsed as WorkflowToolsChangedEvent)
|
||||
} catch (err) {
|
||||
logger.error('Error in workflow_tools_changed handler:', err)
|
||||
}
|
||||
}
|
||||
}
|
||||
} catch (err) {
|
||||
logger.error('Failed to parse pub/sub message:', err)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
publishToolsChanged(event: ToolsChangedEvent): void {
|
||||
if (this.disposed) return
|
||||
this.pub.publish(CHANNEL_TOOLS_CHANGED, JSON.stringify(event)).catch((err) => {
|
||||
logger.error('Failed to publish tools_changed:', err)
|
||||
})
|
||||
}
|
||||
|
||||
publishWorkflowToolsChanged(event: WorkflowToolsChangedEvent): void {
|
||||
if (this.disposed) return
|
||||
this.pub.publish(CHANNEL_WORKFLOW_TOOLS_CHANGED, JSON.stringify(event)).catch((err) => {
|
||||
logger.error('Failed to publish workflow_tools_changed:', err)
|
||||
})
|
||||
}
|
||||
|
||||
onToolsChanged(handler: ToolsChangedHandler): () => void {
|
||||
this.toolsChangedHandlers.add(handler)
|
||||
return () => {
|
||||
this.toolsChangedHandlers.delete(handler)
|
||||
}
|
||||
}
|
||||
|
||||
onWorkflowToolsChanged(handler: WorkflowToolsChangedHandler): () => void {
|
||||
this.workflowToolsChangedHandlers.add(handler)
|
||||
return () => {
|
||||
this.workflowToolsChangedHandlers.delete(handler)
|
||||
}
|
||||
}
|
||||
|
||||
dispose(): void {
|
||||
this.disposed = true
|
||||
this.toolsChangedHandlers.clear()
|
||||
this.workflowToolsChangedHandlers.clear()
|
||||
|
||||
const noop = () => {}
|
||||
this.pub.removeAllListeners()
|
||||
this.sub.removeAllListeners()
|
||||
this.pub.on('error', noop)
|
||||
this.sub.on('error', noop)
|
||||
|
||||
this.sub.unsubscribe().catch(noop)
|
||||
this.pub.quit().catch(noop)
|
||||
this.sub.quit().catch(noop)
|
||||
logger.info('Redis MCP pub/sub disposed')
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Process-local fallback using EventEmitter.
|
||||
* Used when Redis is not configured — notifications only reach listeners in the same process.
|
||||
*/
|
||||
class LocalMcpPubSub implements McpPubSubAdapter {
|
||||
private emitter = new EventEmitter()
|
||||
|
||||
constructor() {
|
||||
this.emitter.setMaxListeners(100)
|
||||
logger.info('MCP pub/sub: Using process-local EventEmitter (Redis not configured)')
|
||||
}
|
||||
|
||||
publishToolsChanged(event: ToolsChangedEvent): void {
|
||||
this.emitter.emit(CHANNEL_TOOLS_CHANGED, event)
|
||||
}
|
||||
|
||||
publishWorkflowToolsChanged(event: WorkflowToolsChangedEvent): void {
|
||||
this.emitter.emit(CHANNEL_WORKFLOW_TOOLS_CHANGED, event)
|
||||
}
|
||||
|
||||
onToolsChanged(handler: ToolsChangedHandler): () => void {
|
||||
this.emitter.on(CHANNEL_TOOLS_CHANGED, handler)
|
||||
return () => {
|
||||
this.emitter.off(CHANNEL_TOOLS_CHANGED, handler)
|
||||
}
|
||||
}
|
||||
|
||||
onWorkflowToolsChanged(handler: WorkflowToolsChangedHandler): () => void {
|
||||
this.emitter.on(CHANNEL_WORKFLOW_TOOLS_CHANGED, handler)
|
||||
return () => {
|
||||
this.emitter.off(CHANNEL_WORKFLOW_TOOLS_CHANGED, handler)
|
||||
}
|
||||
}
|
||||
|
||||
dispose(): void {
|
||||
this.emitter.removeAllListeners()
|
||||
logger.info('Local MCP pub/sub disposed')
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Create the appropriate pub/sub adapter based on Redis availability.
|
||||
*/
|
||||
function createMcpPubSub(): McpPubSubAdapter {
|
||||
const redisUrl = env.REDIS_URL
|
||||
|
||||
if (redisUrl) {
|
||||
try {
|
||||
logger.info('MCP pub/sub: Using Redis')
|
||||
return new RedisMcpPubSub(redisUrl)
|
||||
} catch (err) {
|
||||
logger.error('Failed to create Redis pub/sub, falling back to local:', err)
|
||||
return new LocalMcpPubSub()
|
||||
}
|
||||
}
|
||||
|
||||
return new LocalMcpPubSub()
|
||||
}
|
||||
const workflowToolsChannel =
|
||||
typeof window !== 'undefined'
|
||||
? null
|
||||
: createPubSubChannel<WorkflowToolsChangedEvent>({
|
||||
channel: 'mcp:workflow_tools_changed',
|
||||
label: 'mcp-workflow-tools',
|
||||
})
|
||||
|
||||
export const mcpPubSub: McpPubSubAdapter =
|
||||
typeof window !== 'undefined' ? (null as unknown as McpPubSubAdapter) : createMcpPubSub()
|
||||
typeof window !== 'undefined' || !toolsChannel || !workflowToolsChannel
|
||||
? (null as unknown as McpPubSubAdapter)
|
||||
: {
|
||||
publishToolsChanged: (event) => toolsChannel.publish(event),
|
||||
publishWorkflowToolsChanged: (event) => workflowToolsChannel.publish(event),
|
||||
onToolsChanged: (handler) => toolsChannel.subscribe(handler),
|
||||
onWorkflowToolsChanged: (handler) => workflowToolsChannel.subscribe(handler),
|
||||
dispose: () => {
|
||||
toolsChannel.dispose()
|
||||
workflowToolsChannel.dispose()
|
||||
},
|
||||
}
|
||||
|
||||
@@ -44,7 +44,7 @@ interface FolderState {
|
||||
isTaskSelected: (taskId: string) => boolean
|
||||
|
||||
// Unified selection helpers
|
||||
getFullSelection: () => { workflowIds: string[]; folderIds: string[] }
|
||||
getFullSelection: () => { workflowIds: string[]; folderIds: string[]; taskIds: string[] }
|
||||
hasAnySelection: () => boolean
|
||||
isMixedSelection: () => boolean
|
||||
clearAllSelection: () => void
|
||||
@@ -294,6 +294,7 @@ export const useFolderStore = create<FolderState>()(
|
||||
getFullSelection: () => ({
|
||||
workflowIds: Array.from(get().selectedWorkflows),
|
||||
folderIds: Array.from(get().selectedFolders),
|
||||
taskIds: Array.from(get().selectedTasks),
|
||||
}),
|
||||
|
||||
hasAnySelection: () =>
|
||||
|
||||
2
packages/db/migrations/0171_yielding_venom.sql
Normal file
2
packages/db/migrations/0171_yielding_venom.sql
Normal file
@@ -0,0 +1,2 @@
|
||||
ALTER TABLE "copilot_chats" ADD COLUMN "last_seen_at" timestamp;
|
||||
UPDATE "copilot_chats" SET "last_seen_at" = "updated_at" WHERE "last_seen_at" IS NULL;
|
||||
13051
packages/db/migrations/meta/0171_snapshot.json
Normal file
13051
packages/db/migrations/meta/0171_snapshot.json
Normal file
File diff suppressed because it is too large
Load Diff
@@ -1191,6 +1191,13 @@
|
||||
"when": 1773209025121,
|
||||
"tag": "0170_careful_saracen",
|
||||
"breakpoints": true
|
||||
},
|
||||
{
|
||||
"idx": 171,
|
||||
"version": "7",
|
||||
"when": 1773353601402,
|
||||
"tag": "0171_yielding_venom",
|
||||
"breakpoints": true
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
@@ -1568,6 +1568,7 @@ export const copilotChats = pgTable(
|
||||
previewYaml: text('preview_yaml'),
|
||||
planArtifact: text('plan_artifact'),
|
||||
config: jsonb('config'),
|
||||
lastSeenAt: timestamp('last_seen_at'),
|
||||
createdAt: timestamp('created_at').notNull().defaultNow(),
|
||||
updatedAt: timestamp('updated_at').notNull().defaultNow(),
|
||||
},
|
||||
|
||||
Reference in New Issue
Block a user