refactor: console is now using tanstack query

frontend/src/api/modules/use-modules.ts

@@ -1,45 +1,45 @@
-import { ConsoleService } from '../../protos/xyz/block/ftl/v1/console/console_connect'
-import { GetModulesResponse } from '../../protos/xyz/block/ftl/v1/console/console_pb'
-
 import { Code, ConnectError } from '@connectrpc/connect'
+import { useQuery, useQueryClient } from '@tanstack/react-query'
+import { useEffect } from 'react'
 import { useClient } from '../../hooks/use-client'
+import { ConsoleService } from '../../protos/xyz/block/ftl/v1/console/console_connect'
+import { useSchema } from '../schema/use-schema'
 
-const fetchModules = async (client: ConsoleService, isVisible: boolean): Promise<GetModulesResponse> => {
-  if (!isVisible) {
-    throw new Error('Component is not visible')
-  }
+const useModulesKey = 'modules'
 
-  const abortController = new AbortController()
+export const useModules = () => {
+  const client = useClient(ConsoleService)
+  const queryClient = useQueryClient()
+  const { data: streamingData } = useSchema()
 
-  try {
-    const modules = await client.getModules({}, { signal: abortController.signal })
-    return modules ?? []
-  } catch (error) {
-    if (error instanceof ConnectError) {
-      if (error.code !== Code.Canceled) {
-        console.error('fetchModules - Connect error:', error)
+  useEffect(() => {
+    if (streamingData) {
+      queryClient.invalidateQueries({
+        queryKey: [useModulesKey],
+      })
+    }
+  }, [streamingData, queryClient])
+
+  const fetchModules = async (signal: AbortSignal) => {
+    try {
+      console.debug('fetching modules from FTL')
+      const modules = await client.getModules({}, { signal })
+      return modules ?? []
+    } catch (error) {
+      if (error instanceof ConnectError) {
+        if (error.code !== Code.Canceled) {
+          console.error('fetchModules - Connect error:', error)
+        }
+      } else {
+        console.error('fetchModules:', error)
       }
-    } else {
-      console.error('fetchModules:', error)
+      throw error
     }
-    throw error
-  } finally {
-    abortController.abort()
   }
-}
-
-export const useModules = () => {
-  const client = useClient(ConsoleService)
-  const isVisible = useVisibility()
-  const schema = useSchema()
 
-  return useQuery<GetModulesResponse>(
-    ['modules', schema, isVisible], // The query key, include schema and isVisible as dependencies
-    () => fetchModules(client, isVisible),
-    {
-      enabled: isVisible, // Only run the query when the component is visible
-      refetchOnWindowFocus: false, // Optional: Disable refetching on window focus
-      staleTime: 1000 * 60 * 5, // Optional: Cache data for 5 minutes
-    }
-  )
+  return useQuery({
+    queryKey: [useModulesKey],
+    queryFn: async ({ signal }) => fetchModules(signal),
+    enabled: !!streamingData,
+  })
 }

frontend/src/api/schema/use-schema.ts

@@ -1,65 +1,53 @@
 import { Code, ConnectError } from '@connectrpc/connect'
-import { useEffect, useState } from 'react'
+import { useQuery, useQueryClient } from '@tanstack/react-query'
 import { useClient } from '../../hooks/use-client.ts'
 import { useVisibility } from '../../hooks/use-visibility.ts'
 import { ControllerService } from '../../protos/xyz/block/ftl/v1/ftl_connect.ts'
 import { DeploymentChangeType, type PullSchemaResponse } from '../../protos/xyz/block/ftl/v1/ftl_pb.ts'
 
+const streamingSchemaKey = 'streamingSchema'
+
 export const useSchema = () => {
   const client = useClient(ControllerService)
-  const [schema, setSchema] = useState<PullSchemaResponse[]>([])
+  const queryClient = useQueryClient()
   const isVisible = useVisibility()
 
-  useEffect(() => {
-    const abortController = new AbortController()
-
-    const fetchSchema = async () => {
-      try {
-        if (!isVisible) {
-          abortController.abort()
-          return
+  const streamSchema = async (signal: AbortSignal) => {
+    try {
+      const schemaMap = new Map<string, PullSchemaResponse>()
+      for await (const response of client.pullSchema({}, { signal })) {
+        const moduleName = response.moduleName ?? ''
+        console.log(`schema changed: ${DeploymentChangeType[response.changeType]} ${moduleName}`)
+        switch (response.changeType) {
+          case DeploymentChangeType.DEPLOYMENT_ADDED:
+            schemaMap.set(moduleName, response)
+            break
+          case DeploymentChangeType.DEPLOYMENT_CHANGED:
+            schemaMap.set(moduleName, response)
+            break
+          case DeploymentChangeType.DEPLOYMENT_REMOVED:
+            schemaMap.delete(moduleName)
         }
 
-        const schemaMap = new Map<string, PullSchemaResponse>()
-        for await (const response of client.pullSchema(
-          {},
-          {
-            signal: abortController.signal,
-          },
-        )) {
-          const moduleName = response.moduleName ?? ''
-          console.log(`${response.changeType} ${moduleName}`)
-          switch (response.changeType) {
-            case DeploymentChangeType.DEPLOYMENT_ADDED:
-              schemaMap.set(moduleName, response)
-              break
-            case DeploymentChangeType.DEPLOYMENT_CHANGED:
-              schemaMap.set(moduleName, response)
-              break
-            case DeploymentChangeType.DEPLOYMENT_REMOVED:
-              schemaMap.delete(moduleName)
-          }
-
-          if (!response.more) {
-            setSchema(Array.from(schemaMap.values()).sort((a, b) => a.schema?.name?.localeCompare(b.schema?.name ?? '') ?? 0))
-          }
+        if (!response.more) {
+          const schema = Array.from(schemaMap.values()).sort((a, b) => a.schema?.name?.localeCompare(b.schema?.name ?? '') ?? 0)
+          queryClient.setQueryData([streamingSchemaKey], schema)
         }
-      } catch (error) {
-        if (error instanceof ConnectError) {
-          if (error.code !== Code.Canceled) {
-            console.error('Console service - streamEvents - Connect error:', error)
-          }
-        } else {
-          console.error('Console service - streamEvents:', error)
+      }
+    } catch (error) {
+      if (error instanceof ConnectError) {
+        if (error.code !== Code.Canceled) {
+          console.error('useSchema - streamSchema - Connect error:', error)
         }
+      } else {
+        console.error('useSchema - streamSchema:', error)
       }
     }
+  }
 
-    fetchSchema()
-    return () => {
-      abortController.abort()
-    }
-  }, [client, isVisible])
-
-  return schema
+  return useQuery({
+    queryKey: [streamingSchemaKey],
+    queryFn: async ({ signal }) => streamSchema(signal),
+    enabled: isVisible,
+  })
 }

frontend/src/api/timeline/index.ts

@@ -0,0 +1,5 @@
+export * from './stream-verb-calls'
+export * from './timeline-filters'
+export * from './use-request-calls'
+export * from './use-timeline-calls'
+export * from './use-timeline'

frontend/src/api/timeline/stream-verb-calls.ts

@@ -0,0 +1,6 @@
+import { callFilter } from './timeline-filters.ts'
+import { useTimelineCalls } from './use-timeline-calls.ts'
+
+export const useStreamVerbCalls = (moduleName?: string, verbName?: string, enabled = true) => {
+  return useTimelineCalls(true, [callFilter(moduleName || '', verbName)], enabled)
+}

frontend/src/services/console.service.ts → frontend/src/api/timeline/timeline-filters.ts

@@ -1,24 +1,16 @@
 import type { Timestamp } from '@bufbuild/protobuf'
-import { Code, ConnectError } from '@connectrpc/connect'
-import { createClient } from '../hooks/use-client'
-import { ConsoleService } from '../protos/xyz/block/ftl/v1/console/console_connect'
 import {
-  type CallEvent,
-  type Event,
-  EventType,
+  type EventType,
   EventsQuery_CallFilter,
   EventsQuery_DeploymentFilter,
   EventsQuery_EventTypeFilter,
   EventsQuery_Filter,
   EventsQuery_IDFilter,
   EventsQuery_LogLevelFilter,
-  EventsQuery_Order,
   EventsQuery_RequestFilter,
   EventsQuery_TimeFilter,
   type LogLevel,
-} from '../protos/xyz/block/ftl/v1/console/console_pb'
-
-const client = createClient(ConsoleService)
+} from '../../protos/xyz/block/ftl/v1/console/console_pb'
 
 export const requestKeysFilter = (requestKeys: string[]): EventsQuery_Filter => {
   const filter = new EventsQuery_Filter()
@@ -106,88 +98,3 @@ export const eventIdFilter = ({
   }
   return filter
 }
-
-export const getRequestCalls = async ({
-  abortControllerSignal,
-  requestKey,
-}: {
-  abortControllerSignal: AbortSignal
-  requestKey: string
-}): Promise<CallEvent[]> => {
-  const allEvents = await getEvents({
-    abortControllerSignal,
-    filters: [requestKeysFilter([requestKey]), eventTypesFilter([EventType.CALL])],
-  })
-  return allEvents.map((e) => e.entry.value) as CallEvent[]
-}
-
-export const getCalls = async ({
-  abortControllerSignal,
-  destModule,
-  destVerb,
-  sourceModule,
-}: {
-  abortControllerSignal: AbortSignal
-  destModule: string
-  destVerb?: string
-  sourceModule?: string
-}): Promise<CallEvent[]> => {
-  const allEvents = await getEvents({
-    abortControllerSignal,
-    filters: [callFilter(destModule, destVerb, sourceModule), eventTypesFilter([EventType.CALL])],
-  })
-  return allEvents.map((e) => e.entry.value) as CallEvent[]
-}
-
-export const getEvents = async ({
-  abortControllerSignal,
-  limit = 1000,
-  order = EventsQuery_Order.DESC,
-  filters = [],
-}: {
-  abortControllerSignal: AbortSignal
-  limit?: number
-  order?: EventsQuery_Order
-  filters?: EventsQuery_Filter[]
-}): Promise<Event[]> => {
-  try {
-    const response = await client.getEvents({ filters, limit, order }, { signal: abortControllerSignal })
-    return response.events
-  } catch (error) {
-    if (error instanceof ConnectError) {
-      if (error.code === Code.Canceled) {
-        return []
-      }
-    }
-    throw error
-  }
-}
-
-export const streamEvents = async ({
-  abortControllerSignal,
-  filters,
-  onEventsReceived,
-}: {
-  abortControllerSignal: AbortSignal
-  filters: EventsQuery_Filter[]
-  onEventsReceived: (events: Event[]) => void
-}) => {
-  try {
-    for await (const response of client.streamEvents(
-      { updateInterval: { seconds: BigInt(1) }, query: { limit: 200, filters, order: EventsQuery_Order.DESC } },
-      { signal: abortControllerSignal },
-    )) {
-      if (response.events) {
-        onEventsReceived(response.events)
-      }
-    }
-  } catch (error) {
-    if (error instanceof ConnectError) {
-      if (error.code !== Code.Canceled) {
-        console.error('Console service - streamEvents - Connect error:', error)
-      }
-    } else {
-      console.error('Console service - streamEvents:', error)
-    }
-  }
-}

frontend/src/api/timeline/use-request-calls.ts

@@ -0,0 +1,6 @@
+import { requestKeysFilter } from './timeline-filters'
+import { useTimelineCalls } from './use-timeline-calls'
+
+export const useRequestCalls = (requestKey?: string) => {
+  return useTimelineCalls(true, [requestKeysFilter([requestKey || ''])], !!requestKey)
+}

frontend/src/api/timeline/use-timeline-calls.ts

@@ -0,0 +1,16 @@
+import { type CallEvent, EventType, type EventsQuery_Filter } from '../../protos/xyz/block/ftl/v1/console/console_pb.ts'
+import { eventTypesFilter } from './timeline-filters.ts'
+import { useTimeline } from './use-timeline.ts'
+
+export const useTimelineCalls = (isStreaming: boolean, filters: EventsQuery_Filter[], enabled = true) => {
+  const allFilters = [...filters, eventTypesFilter([EventType.CALL])]
+  const timelineQuery = useTimeline(isStreaming, allFilters, enabled)
+
+  // Map the events to CallEvent for ease of use
+  const data = timelineQuery.data?.map((event) => event.entry.value as CallEvent) || []
+
+  return {
+    ...timelineQuery,
+    data,
+  }
+}

frontend/src/api/timeline/use-timeline.ts

@@ -0,0 +1,63 @@
+import { Code, ConnectError } from '@connectrpc/connect'
+import { useQuery, useQueryClient } from '@tanstack/react-query'
+import { useClient } from '../../hooks/use-client'
+import { useVisibility } from '../../hooks/use-visibility'
+import { ConsoleService } from '../../protos/xyz/block/ftl/v1/console/console_connect'
+import { type EventsQuery_Filter, EventsQuery_Order } from '../../protos/xyz/block/ftl/v1/console/console_pb'
+
+const timelineKey = 'timeline'
+const maxTimelineEntries = 1000
+
+export const useTimeline = (isStreaming: boolean, filters: EventsQuery_Filter[], enabled = true) => {
+  const client = useClient(ConsoleService)
+  const queryClient = useQueryClient()
+  const isVisible = useVisibility()
+
+  const order = EventsQuery_Order.DESC
+  const limit = isStreaming ? 200 : 1000
+
+  const queryKey = [timelineKey, isStreaming, filters, order, limit]
+
+  const fetchTimeline = async ({ signal }: { signal: AbortSignal }) => {
+    try {
+      console.log('fetching timeline')
+      const response = await client.getEvents({ filters, limit, order }, { signal })
+      return response.events
+    } catch (error) {
+      if (error instanceof ConnectError) {
+        if (error.code === Code.Canceled) {
+          return []
+        }
+      }
+      throw error
+    }
+  }
+
+  const streamTimeline = async ({ signal }: { signal: AbortSignal }) => {
+    try {
+      console.log('streaming timeline')
+      console.log('filters:', filters)
+      for await (const response of client.streamEvents({ updateInterval: { seconds: BigInt(1) }, query: { limit, filters, order } }, { signal })) {
+        if (response.events) {
+          const prev = queryClient.getQueryData<Event[]>(queryKey) ?? []
+          const allEvents = [...response.events, ...prev].slice(0, maxTimelineEntries)
+          queryClient.setQueryData(queryKey, allEvents)
+        }
+      }
+    } catch (error) {
+      if (error instanceof ConnectError) {
+        if (error.code !== Code.Canceled) {
+          console.error('Console service - streamEvents - Connect error:', error)
+        }
+      } else {
+        console.error('Console service - streamEvents:', error)
+      }
+    }
+  }
+
+  return useQuery({
+    queryKey: queryKey,
+    queryFn: async ({ signal }) => (isStreaming ? streamTimeline({ signal }) : fetchTimeline({ signal })),
+    enabled: enabled && isVisible,
+  })
+}