History

Sync tool inventory via worker watchers

Track workspace tool-source updates in Convex and have the worker refresh cached tool inventory so the web UI stays live without calling /api/tools for warmup.

Rhys Sullivan committed 9 months ago

Commit 9783c8d1 parent 1ced8666 files · +77 −29Browse files

Unified · Split

6 files
  1. executor/apps/server/src/database.ts +5 −0
  2. executor/apps/server/src/service.ts +8 −0
  3. executor/apps/server/src/worker.ts +41 −0
  4. executor/apps/web/src/components/tools-view.tsx +1 −12
  5. executor/apps/web/src/hooks/use-workspace-tools.ts +3 −17
  6. executor/convex/database.ts +19 −0
executor/apps/server/src/database.ts+5−0
@@ -45,6 +45,7 @@ type QueryName = · 44 unmodified lines
45 | "database:listCredentials"45 | "database:listCredentials"
46 | "database:resolveCredential"46 | "database:resolveCredential"
47 | "database:listToolSources"47 | "database:listToolSources"
48 | "database:listToolSourceWorkspaceUpdates"
48 | "database:listWorkspaceToolsForContext"49 | "database:listWorkspaceToolsForContext"
49 | "database:listTaskEvents";50 | "database:listTaskEvents";
50 51
@@ -210,6 +211,10 @@ export class ExecutorDatabase { · 159 unmodified lines
210 return await this.query("database:listToolSources", { workspaceId });211 return await this.query("database:listToolSources", { workspaceId });
211 }212 }
212 213
214 async listToolSourceWorkspaceUpdates(): Promise<Array<{ workspaceId: string; updatedAt: number }>> {
215 return await this.query("database:listToolSourceWorkspaceUpdates", {});
216 }
217
213 async syncWorkspaceTools(params: {218 async syncWorkspaceTools(params: {
214 workspaceId: string;219 workspaceId: string;
215 tools: ToolDescriptor[];220 tools: ToolDescriptor[];
executor/apps/server/src/service.ts+8−0
@@ -213,6 +213,14 @@ export class ExecutorService { · 212 unmodified lines
213 return await this.db.listToolSources(workspaceId);213 return await this.db.listToolSources(workspaceId);
214 }214 }
215 215
216 async listToolSourceWorkspaceUpdates(): Promise<Array<{ workspaceId: string; updatedAt: number }>> {
217 return await this.db.listToolSourceWorkspaceUpdates();
218 }
219
220 async refreshWorkspaceTools(workspaceId: string): Promise<void> {
221 await this.getWorkspaceTools(workspaceId);
222 }
223
216 async upsertToolSource(input: {224 async upsertToolSource(input: {
217 id?: string;225 id?: string;
218 workspaceId: string;226 workspaceId: string;
executor/apps/server/src/worker.ts+41−0
@@ -51,6 +51,7 @@ const convexClient = new ConvexClient(convexUrl, { · 50 unmodified lines
51});51});
52 52
53let draining = false;53let draining = false;
54let syncingToolInventories = false;
54 55
55async function drainQueue(trigger: string): Promise<void> {56async function drainQueue(trigger: string): Promise<void> {
56 if (draining) {57 if (draining) {
@@ -78,6 +79,30 @@ async function drainQueue(trigger: string): Promise<void> { · 21 unmodified lines
78 }79 }
79}80}
80 81
82async function syncWorkspaceToolInventories(trigger: string): Promise<void> {
83 if (syncingToolInventories) {
84 return;
85 }
86
87 syncingToolInventories = true;
88 try {
89 const updates = await service.listToolSourceWorkspaceUpdates();
90 if (updates.length === 0) {
91 return;
92 }
93
94 console.log(`[worker] ${trigger}: syncing tool inventories for ${updates.length} workspace(s)`);
95 for (const update of updates) {
96 await service.refreshWorkspaceTools(update.workspaceId);
97 }
98 } catch (error) {
99 const message = error instanceof Error ? error.message : String(error);
100 console.error(`[worker] tool inventory sync failed: ${message}`);
101 } finally {
102 syncingToolInventories = false;
103 }
104}
105
81const queueSubscription = convexClient.onUpdate(106const queueSubscription = convexClient.onUpdate(
82 api.database.listQueuedTaskIds,107 api.database.listQueuedTaskIds,
83 { limit: 1 },108 { limit: 1 },
@@ -91,15 +116,31 @@ const queueSubscription = convexClient.onUpdate( · 7 unmodified lines
91 },116 },
92);117);
93 118
119const toolSourceSubscription = convexClient.onUpdate(
120 api.database.listToolSourceWorkspaceUpdates,
121 {},
122 (updates) => {
123 if (updates.length > 0) {
124 void syncWorkspaceToolInventories("onUpdate");
125 }
126 },
127 (error) => {
128 console.warn(`[worker] tool source watcher error: ${error.message}`);
129 },
130);
131
94const interval = setInterval(() => {132const interval = setInterval(() => {
95 void drainQueue("interval");133 void drainQueue("interval");
134 void syncWorkspaceToolInventories("interval");
96}, pollMs);135}, pollMs);
97 136
98await drainQueue("startup");137await drainQueue("startup");
138await syncWorkspaceToolInventories("startup");
99 139
100function shutdown(signal: string): void {140function shutdown(signal: string): void {
101 console.log(`[worker] received ${signal}, shutting down...`);141 console.log(`[worker] received ${signal}, shutting down...`);
102 queueSubscription.unsubscribe();142 queueSubscription.unsubscribe();
143 toolSourceSubscription.unsubscribe();
103 clearInterval(interval);144 clearInterval(interval);
104 void convexClient.close();145 void convexClient.close();
105 process.exit(0);146 process.exit(0);
executor/apps/web/src/components/tools-view.tsx+1−12
@@ -5,7 +5,6 @@ import { · 4 unmodified lines
5 Wrench,5 Wrench,
6 Plus,6 Plus,
7 Trash2,7 Trash2,
8 RefreshCw,
9 ShieldCheck,8 ShieldCheck,
10 Globe,9 Globe,
11 Server,10 Server,
@@ -699,17 +698,7 @@ export function ToolsView() { · 687 unmodified lines
699 <PageHeader698 <PageHeader
700 title="Tools"699 title="Tools"
701 description="Manage tool sources and view available tools"700 description="Manage tool sources and view available tools"
702 >701 />
703 <Button
704 variant="outline"
705 size="sm"
706 className="h-8 text-xs"
707 onClick={() => void refreshInventory()}
708 >
709 <RefreshCw className="h-3.5 w-3.5 mr-1.5" />
710 Refresh
711 </Button>
712 </PageHeader>
713 702
714 <Tabs defaultValue="sources" className="w-full">703 <Tabs defaultValue="sources" className="w-full">
715 <TabsList className="bg-muted/50 h-9">704 <TabsList className="bg-muted/50 h-9">
executor/apps/web/src/hooks/use-workspace-tools.ts+3−17
@@ -1,9 +1,8 @@
1"use client";1"use client";
2 2
3import { useCallback, useEffect } from "react";3import { useCallback } from "react";
4import { useQuery } from "convex/react";4import { useQuery } from "convex/react";
5import { convexApi } from "../lib/convex-api";5import { convexApi } from "../lib/convex-api";
6import * as api from "../lib/api";
7 6
8interface WorkspaceContext {7interface WorkspaceContext {
9 workspaceId: string;8 workspaceId: string;
@@ -24,21 +23,8 @@ export function useWorkspaceTools(context: WorkspaceContext | null) { · 14 unmodified lines
24 );23 );
25 24
26 const refresh = useCallback(async () => {25 const refresh = useCallback(async () => {
27 if (!context) return;26 // Convex queries are live; manual refresh is not required.
28 try {27 }, []);
29 await api.listToolsForContext({
30 workspaceId: context.workspaceId,
31 actorId: context.actorId,
32 clientId: context.clientId,
33 });
34 } catch {
35 // best effort warm-up for server-side tool discovery sync
36 }
37 }, [context]);
38
39 useEffect(() => {
40 void refresh();
41 }, [refresh]);
42 28
43 return {29 return {
44 tools,30 tools,
executor/convex/database.ts+19−0
@@ -763,6 +763,25 @@ export const listToolSources = query({ · 762 unmodified lines
763 },763 },
764});764});
765 765
766export const listToolSourceWorkspaceUpdates = query({
767 args: {},
768 handler: async (ctx) => {
769 const docs = await ctx.db.query("toolSources").collect();
770 const byWorkspace = new Map<string, number>();
771
772 for (const doc of docs) {
773 const existing = byWorkspace.get(doc.workspaceId) ?? 0;
774 if (doc.updatedAt > existing) {
775 byWorkspace.set(doc.workspaceId, doc.updatedAt);
776 }
777 }
778
779 return [...byWorkspace.entries()]
780 .map(([workspaceId, updatedAt]) => ({ workspaceId, updatedAt }))
781 .sort((a, b) => a.workspaceId.localeCompare(b.workspaceId));
782 },
783});
784
766export const syncWorkspaceTools = mutation({785export const syncWorkspaceTools = mutation({
767 args: {786 args: {
768 workspaceId: v.string(),787 workspaceId: v.string(),