diff --git a/apps/api/src/lib/redis.ts b/apps/api/src/lib/redis.ts index 309cdc3..e6b8e1a 100644 --- a/apps/api/src/lib/redis.ts +++ b/apps/api/src/lib/redis.ts @@ -102,6 +102,9 @@ export interface ClusterStatusCache { serverCount: number; version: string; lastCheck: string; + serverId?: string; + serverName?: string; + rtt?: string; } export async function setClusterStatus(clusterId: string, data: ClusterStatusCache): Promise { diff --git a/apps/api/src/modules/clusters/clusters.service.ts b/apps/api/src/modules/clusters/clusters.service.ts index 7f43491..8d2a6b4 100644 --- a/apps/api/src/modules/clusters/clusters.service.ts +++ b/apps/api/src/modules/clusters/clusters.service.ts @@ -254,9 +254,10 @@ export async function checkHealth( return { connected: cachedStatus.status === 'connected', status: cachedStatus.status as ClusterStatus, + rtt: cachedStatus.rtt ? parseInt(cachedStatus.rtt) : undefined, serverInfo: { - serverId: '', - serverName: '', + serverId: cachedStatus.serverId || '', + serverName: cachedStatus.serverName || '', version: cachedStatus.version, jetstream: true, }, @@ -308,6 +309,9 @@ export async function checkHealth( serverCount: 1, version, lastCheck: new Date().toISOString(), + serverId: health.serverInfo?.serverId || '', + serverName: health.serverInfo?.serverName || '', + rtt: health.rtt?.toString() || '', }); return { diff --git a/apps/api/src/modules/consumers/consumers.routes.ts b/apps/api/src/modules/consumers/consumers.routes.ts index 4b2600c..c18bcf0 100644 --- a/apps/api/src/modules/consumers/consumers.routes.ts +++ b/apps/api/src/modules/consumers/consumers.routes.ts @@ -7,6 +7,18 @@ export const consumerRoutes: FastifyPluginAsync = async (fastify) => { // All routes require authentication fastify.addHook('preHandler', authenticate); + // GET /clusters/:cid/consumers - List ALL consumers across all streams + fastify.get<{ Params: { cid: string } }>( + '/:cid/consumers', + async (request) => { + const consumers = await consumerService.listAllConsumers( + request.user!.orgId, + request.params.cid + ); + return { consumers }; + } + ); + // GET /clusters/:cid/streams/:sid/consumers - List consumers fastify.get<{ Params: { cid: string; sid: string } }>( '/:cid/streams/:sid/consumers', diff --git a/apps/api/src/modules/consumers/consumers.service.ts b/apps/api/src/modules/consumers/consumers.service.ts index 0f120ef..be9a593 100644 --- a/apps/api/src/modules/consumers/consumers.service.ts +++ b/apps/api/src/modules/consumers/consumers.service.ts @@ -7,10 +7,16 @@ import { deleteConsumer as natsDeleteConsumer, pauseConsumer as natsPauseConsumer, resumeConsumer as natsResumeConsumer, + listStreams as natsListStreams, } from '../../lib/nats'; import { NotFoundError } from '../../../../shared/src/index'; import type { ConsumerInfo, CreateConsumerInput, UpdateConsumerInput } from '../../../../shared/src/index'; +// Extended consumer info with stream name +export interface ConsumerInfoWithStream extends ConsumerInfo { + streamName: string; +} + // ==================== Consumer Operations ==================== export async function listConsumers( @@ -30,6 +36,41 @@ export async function listConsumers( return natsListConsumers(clusterId, streamName); } +export async function listAllConsumers( + orgId: string, + clusterId: string +): Promise { + // Verify cluster belongs to org + const cluster = await prisma.natsCluster.findFirst({ + where: { id: clusterId, orgId }, + }); + + if (!cluster) { + throw new NotFoundError('Cluster', clusterId); + } + + // Get all streams + const streams = await natsListStreams(clusterId); + + // Get consumers from each stream + const allConsumers: ConsumerInfoWithStream[] = []; + for (const stream of streams) { + try { + const consumers = await natsListConsumers(clusterId, stream.config.name); + for (const consumer of consumers) { + allConsumers.push({ + ...consumer, + streamName: stream.config.name, + }); + } + } catch { + // Skip streams that have issues + } + } + + return allConsumers; +} + export async function getConsumer( orgId: string, clusterId: string, diff --git a/apps/web/app/(dashboard)/analytics/page.tsx b/apps/web/app/(dashboard)/analytics/page.tsx index be65b52..7e987a4 100644 --- a/apps/web/app/(dashboard)/analytics/page.tsx +++ b/apps/web/app/(dashboard)/analytics/page.tsx @@ -1,15 +1,17 @@ 'use client'; -import { useState } from 'react'; +import { useState, useEffect } from 'react'; import { useQuery } from '@tanstack/react-query'; -import { BarChart3, TrendingUp, TrendingDown, Activity, ArrowUpRight, ArrowDownRight, Loader2 } from 'lucide-react'; +import { BarChart3, TrendingUp, TrendingDown, Activity, ArrowUpRight, ArrowDownRight, Loader2, WifiOff, RefreshCw } from 'lucide-react'; import { api } from '@/lib/api'; import { Card, CardContent, CardDescription, CardHeader, CardTitle } from '@/components/ui/card'; +import { Button } from '@/components/ui/button'; import { LineChart, BarChart, MultiLineChart } from '@/components/charts'; import { formatBytes, formatNumber } from '@nats-console/shared'; +import { useClusterStore } from '@/stores/cluster'; export default function AnalyticsPage() { - const [selectedCluster, setSelectedCluster] = useState(null); + const { selectedClusterId, setSelectedClusterId } = useClusterStore(); const [timeRange, setTimeRange] = useState('24h'); const { data: clustersData } = useQuery({ @@ -17,47 +19,68 @@ export default function AnalyticsPage() { queryFn: () => api.clusters.list(), }); + // Ensure cluster connection before fetching analytics + const { data: healthData, isLoading: isLoadingHealth, refetch: refetchHealth } = useQuery({ + queryKey: ['cluster-health', selectedClusterId], + queryFn: () => (selectedClusterId ? api.clusters.health(selectedClusterId) : null), + enabled: !!selectedClusterId, + staleTime: 30000, + retry: 1, + }); + + const isClusterConnected = healthData?.connected === true; + const isClusterDisconnected = selectedClusterId && healthData && !healthData.connected; + const { data: analyticsData, isLoading } = useQuery({ - queryKey: ['analytics', selectedCluster, timeRange], + queryKey: ['analytics', selectedClusterId, timeRange], queryFn: () => - selectedCluster - ? api.analytics.overview(selectedCluster, timeRange) + selectedClusterId + ? api.analytics.overview(selectedClusterId, timeRange) : null, - enabled: !!selectedCluster, + enabled: !!selectedClusterId && isClusterConnected, }); // Chart data queries const { data: throughputData, isLoading: throughputLoading } = useQuery({ - queryKey: ['analytics-throughput', selectedCluster, timeRange], + queryKey: ['analytics-throughput', selectedClusterId, timeRange], queryFn: () => - selectedCluster - ? api.analytics.chartThroughput(selectedCluster, timeRange) + selectedClusterId + ? api.analytics.chartThroughput(selectedClusterId, timeRange) : null, - enabled: !!selectedCluster, + enabled: !!selectedClusterId && isClusterConnected, }); const { data: consumerLagData, isLoading: consumerLagLoading } = useQuery({ - queryKey: ['analytics-consumer-lag', selectedCluster, timeRange], + queryKey: ['analytics-consumer-lag', selectedClusterId, timeRange], queryFn: () => - selectedCluster - ? api.analytics.chartConsumerLag(selectedCluster, timeRange) + selectedClusterId + ? api.analytics.chartConsumerLag(selectedClusterId, timeRange) : null, - enabled: !!selectedCluster, + enabled: !!selectedClusterId && isClusterConnected, }); const { data: streamActivityData, isLoading: streamActivityLoading } = useQuery({ - queryKey: ['analytics-stream-activity', selectedCluster, timeRange], + queryKey: ['analytics-stream-activity', selectedClusterId, timeRange], queryFn: () => - selectedCluster - ? api.analytics.chartStreamActivity(selectedCluster, timeRange) + selectedClusterId + ? api.analytics.chartStreamActivity(selectedClusterId, timeRange) : null, - enabled: !!selectedCluster, + enabled: !!selectedClusterId && isClusterConnected, }); - // Auto-select first cluster - if (clustersData?.clusters?.length && !selectedCluster) { - setSelectedCluster(clustersData.clusters[0].id); - } + // Auto-select saved cluster or first cluster + useEffect(() => { + if (clustersData?.clusters?.length) { + const savedClusterExists = clustersData.clusters.some( + (c: any) => c.id === selectedClusterId + ); + if (!savedClusterExists) { + setSelectedClusterId(clustersData.clusters[0].id); + } else if (!selectedClusterId) { + setSelectedClusterId(clustersData.clusters[0].id); + } + } + }, [clustersData?.clusters, selectedClusterId, setSelectedClusterId]); const stats = analyticsData || { totalMessages: 0, @@ -132,8 +155,8 @@ export default function AnalyticsPage() {
- {!selectedCluster && ( + {!selectedClusterId && ( @@ -165,13 +188,29 @@ export default function AnalyticsPage() { )} - {selectedCluster && isLoading && ( + {isClusterDisconnected && ( + + + +

Cluster Not Reachable

+

+ Unable to connect to the selected cluster. Please check if the cluster is running and accessible. +

+ +
+
+ )} + + {selectedClusterId && (isLoading || isLoadingHealth) && !isClusterDisconnected && (
)} - {selectedCluster && !isLoading && ( + {selectedClusterId && isClusterConnected && !isLoading && ( <>
{ + if (clusterId) { + setSelectedClusterId(clusterId); + } + }, [clusterId, setSelectedClusterId]); + const { data: clusterData, isLoading } = useQuery({ queryKey: ['cluster', clusterId], queryFn: () => api.clusters.get(clusterId), @@ -120,15 +129,17 @@ function ClusterDetailContent() { const getEnvironmentBadge = (env: string) => { const colors: Record = { - production: 'bg-red-100 text-red-700', - staging: 'bg-yellow-100 text-yellow-700', - development: 'bg-green-100 text-green-700', + production: 'bg-purple-100 text-purple-700', + staging: 'bg-amber-100 text-amber-700', + development: 'bg-sky-100 text-sky-700', }; return colors[env] || 'bg-gray-100 text-gray-700'; }; - const jetstream = infoData?.jetstream; + const jetstream = infoData?.jetstream || { streams: 0, consumers: 0, messages: 0, bytes: 0 }; const streams = streamsData?.streams || []; + // Get URL from primary connection + const clusterUrl = cluster?.connections?.[0]?.serverUrl || ''; return (
@@ -229,7 +240,7 @@ function ClusterDetailContent() {

URL

-

{cluster.url}

+

{clusterUrl || 'Not configured'}

Version

@@ -240,50 +251,48 @@ function ClusterDetailContent() { {/* JetStream Info */} - {jetstream && ( -
- - - Streams - - -
- {jetstream.streams || 0} -
-
-
- - - Consumers - - -
- {jetstream.consumers || 0} -
-
-
- - - Messages - - -
- {formatNumber(jetstream.messages || 0)} -
-
-
- - - Storage Used - - -
- {formatBytes(jetstream.bytes || 0)} -
-
-
-
- )} +
+ + + Streams + + +
+ {jetstream.streams || 0} +
+
+
+ + + Consumers + + +
+ {jetstream.consumers || 0} +
+
+
+ + + Messages + + +
+ {formatNumber(jetstream.messages || 0)} +
+
+
+ + + Storage Used + + +
+ {formatBytes(jetstream.bytes || 0)} +
+
+
+
{/* Server Info */} @@ -291,44 +300,117 @@ function ClusterDetailContent() { Server Information -
-
-
- Server ID - - {healthData?.serverInfo?.serverId || '-'} - + +
+
+
+ Server ID + {healthData?.serverInfo?.serverId ? ( + + + + {healthData.serverInfo.serverId} + + + +

{healthData.serverInfo.serverId}

+
+
+ ) : ( + - + )} +
+
+ Server Name + {healthData?.serverInfo?.serverName ? ( + + + + {healthData.serverInfo.serverName} + + + +

{healthData.serverInfo.serverName}

+
+
+ ) : ( + - + )} +
+
+ Version + {healthData?.serverInfo?.version || cluster.version || '-'} +
-
- Server Name - {healthData?.serverInfo?.serverName || '-'} -
-
- Version - {healthData?.serverInfo?.version || cluster.version || '-'} +
+
+ JetStream + + {healthData?.serverInfo?.jetstream ? 'Enabled' : 'Disabled'} + +
+
+ RTT + {healthData?.rtt ? `${healthData.rtt}ms` : '-'} +
+
+ Status + + {cluster.status} + +
-
-
- JetStream - - {healthData?.serverInfo?.jetstream ? 'Enabled' : 'Disabled'} - -
-
- RTT - {healthData?.rtt ? `${healthData.rtt}ms` : '-'} -
-
- Status - - {cluster.status} - -
-
-
+
+ + {/* Resource Usage Gauges */} +
+ + + Streams + + + + + + + + Consumers + + + + + + + + Storage Usage + + + + + +
)} @@ -412,7 +494,7 @@ function ClusterDetailContent() {
-

{cluster.url}

+

{clusterUrl || 'Not configured'}

@@ -428,133 +510,6 @@ function ClusterDetailContent() { )} - {/* Metrics Tab */} - {activeTab === 'metrics' && ( -
- {/* Resource Usage Gauges */} -
- - - Streams - - - - - - - - Consumers - - - - - - - - Storage Usage - - - - - -
- - {/* JetStream Stats */} - - - JetStream Overview - Current JetStream resource usage - - -
-
-

Streams

-

{formatNumber(jetstream?.streams || 0)}

-
-
-

Consumers

-

{formatNumber(jetstream?.consumers || 0)}

-
-
-

Messages

-

{formatNumber(jetstream?.messages || 0)}

-
-
-

Data Size

-

{formatBytes(jetstream?.bytes || 0)}

-
-
-
-
- - {/* Server Info */} - - - Server Information - Server runtime statistics - - -
-
-
- Server ID - - {healthData?.serverInfo?.serverId || '-'} - -
-
- Server Name - {healthData?.serverInfo?.serverName || '-'} -
-
- RTT - {healthData?.rtt ? `${healthData.rtt}ms` : '-'} -
-
-
-
- Server Version - {healthData?.serverInfo?.version || cluster.version || '-'} -
-
- JetStream - - {healthData?.serverInfo?.jetstream ? 'Enabled' : 'Disabled'} - -
-
- Status - - {cluster.status} - -
-
-
-
-
-
- )}
); } diff --git a/apps/web/app/(dashboard)/clusters/page.tsx b/apps/web/app/(dashboard)/clusters/page.tsx index 9739a29..736e0ab 100644 --- a/apps/web/app/(dashboard)/clusters/page.tsx +++ b/apps/web/app/(dashboard)/clusters/page.tsx @@ -1,22 +1,48 @@ 'use client'; -import { useState } from 'react'; +import { useState, useMemo } from 'react'; import { useQuery } from '@tanstack/react-query'; -import { Plus, Server, CheckCircle, XCircle, AlertCircle, RefreshCw } from 'lucide-react'; +import { Plus, Server, CheckCircle, XCircle, AlertCircle, RefreshCw, Search, Star } from 'lucide-react'; import Link from 'next/link'; import { api } from '@/lib/api'; import { Button } from '@/components/ui/button'; +import { Input } from '@/components/ui/input'; import { Card, CardContent, CardDescription, CardHeader, CardTitle } from '@/components/ui/card'; import { CreateClusterDialog } from '@/components/forms/create-cluster-dialog'; +import { useClusterStore } from '@/stores/cluster'; export default function ClustersPage() { const [showCreateDialog, setShowCreateDialog] = useState(false); + const [search, setSearch] = useState(''); + const { selectedClusterId, setSelectedClusterId } = useClusterStore(); const { data, isLoading, error, refetch } = useQuery({ queryKey: ['clusters'], queryFn: () => api.clusters.list(), }); + // Filter and sort clusters alphabetically + const filteredClusters = useMemo(() => { + if (!data?.clusters) return []; + + let clusters = [...data.clusters]; + + // Filter by search + if (search) { + clusters = clusters.filter((cluster: any) => + cluster.name.toLowerCase().includes(search.toLowerCase()) || + cluster.description?.toLowerCase().includes(search.toLowerCase()) + ); + } + + // Sort alphabetically by name + clusters.sort((a: any, b: any) => a.name.localeCompare(b.name)); + + return clusters; + }, [data?.clusters, search]); + + const showSearch = (data?.clusters?.length || 0) > 3; + const getStatusIcon = (status: string) => { switch (status) { case 'connected': @@ -32,9 +58,9 @@ export default function ClustersPage() { const getEnvironmentBadge = (env: string) => { const colors: Record = { - production: 'bg-red-100 text-red-700', - staging: 'bg-yellow-100 text-yellow-700', - development: 'bg-green-100 text-green-700', + production: 'bg-purple-100 text-purple-700', + staging: 'bg-amber-100 text-amber-700', + development: 'bg-sky-100 text-sky-700', }; return ( @@ -62,6 +88,19 @@ export default function ClustersPage() {
+ {/* Search - only show if more than 3 clusters */} + {showSearch && ( +
+ + setSearch(e.target.value)} + className="pl-9" + /> +
+ )} + {isLoading && (
@@ -92,34 +131,51 @@ export default function ClustersPage() { )} - {data?.clusters && data.clusters.length > 0 && ( + {filteredClusters.length > 0 && (
- {data.clusters.map((cluster: any) => ( - - - -
- {getStatusIcon(cluster.status)} -
- {cluster.name} - {cluster.description || 'No description'} + {filteredClusters.map((cluster: any) => { + const isSelected = cluster.id === selectedClusterId; + return ( + setSelectedClusterId(cluster.id)}> + + +
+ {getStatusIcon(cluster.status)} +
+ + {cluster.name} + {isSelected && } + + {cluster.description || 'No description'} +
-
- - -
- {getEnvironmentBadge(cluster.environment)} - - {cluster.version || 'Unknown version'} - -
-
- - - ))} + + +
+ {getEnvironmentBadge(cluster.environment)} + + {cluster.version || 'Unknown version'} + +
+
+ + + ); + })}
)} + {/* No results from search */} + {search && filteredClusters.length === 0 && data?.clusters && data.clusters.length > 0 && ( + + + +

No clusters found

+

No clusters match "{search}"

+
+
+ )} +
); } diff --git a/apps/web/app/(dashboard)/consumers/[clusterId]/[stream]/[name]/page.tsx b/apps/web/app/(dashboard)/consumers/[clusterId]/[stream]/[name]/page.tsx index 920d141..e480742 100644 --- a/apps/web/app/(dashboard)/consumers/[clusterId]/[stream]/[name]/page.tsx +++ b/apps/web/app/(dashboard)/consumers/[clusterId]/[stream]/[name]/page.tsx @@ -39,7 +39,6 @@ import { LineChart } from '@/components/charts'; const tabs: Tab[] = [ { id: 'overview', label: 'Overview', icon: Users }, { id: 'config', label: 'Configuration', icon: Settings }, - { id: 'metrics', label: 'Metrics', icon: BarChart3 }, ]; function ConsumerDetailContent() { @@ -103,7 +102,7 @@ function ConsumerDetailContent() { const { data: metricsData, isLoading: isLoadingMetrics } = useQuery({ queryKey: ['consumer-metrics', clusterId, streamName, consumerName, metricsTimeRange], queryFn: () => api.analytics.consumerLag(consumerName, getTimeRangeParams()), - enabled: activeTab === 'metrics', + enabled: activeTab === 'overview', }); // Transform metrics data for charts @@ -442,6 +441,108 @@ function ConsumerDetailContent() {
+ + {/* Consumer Lag Chart */} + + +
+ Consumer Lag Over Time + Message lag trend +
+ +
+ + {isLoadingMetrics ? ( +
+
+
+ ) : chartData.lag.length > 0 ? ( + + ) : ( +
+
+ +

No metrics data available

+
+
+ )} +
+
+ + {/* Pending Messages Chart */} + + + Pending Messages Over Time + Messages pending delivery trend + + + {isLoadingMetrics ? ( +
+
+
+ ) : chartData.pending.length > 0 ? ( + + ) : ( +
+
+ +

No metrics data available

+
+
+ )} +
+
+ + {/* Ack Rate Chart */} + + + Acknowledgment Rate + Message acknowledgment rate over time + + + {isLoadingMetrics ? ( +
+
+
+ ) : chartData.ackRate.length > 0 ? ( + + ) : ( +
+
+ +

No metrics data available

+
+
+ )} +
+
)} @@ -460,153 +561,6 @@ function ConsumerDetailContent() { )} - {/* Metrics Tab */} - {activeTab === 'metrics' && ( -
- {/* Time Range Selector */} -
- -
- - {/* Stats Cards - Moved to top */} -
- - - Pending - - -
{formatNumber(consumer.num_pending || 0)}
-
-
- - - Ack Pending - - -
{formatNumber(consumer.num_ack_pending || 0)}
-
-
- - - Redelivered - - -
{formatNumber(consumer.num_redelivered || 0)}
-
-
- - - Waiting - - -
{formatNumber(consumer.num_waiting || 0)}
-
-
-
- - {/* Consumer Lag Chart */} - - - Consumer Lag - Message lag over time - - - {isLoadingMetrics ? ( -
-
-
- ) : chartData.lag.length > 0 ? ( - - ) : ( -
-
- -

No metrics data available

-

Metrics will appear once the consumer has activity

-
-
- )} -
-
- - {/* Pending Messages Chart */} - - - Pending Messages - Messages pending delivery over time - - - {isLoadingMetrics ? ( -
-
-
- ) : chartData.pending.length > 0 ? ( - - ) : ( -
-
- -

No metrics data available

-

Metrics will appear once the consumer has activity

-
-
- )} -
-
- - {/* Ack Rate Chart */} - - - Acknowledgment Rate - Message acknowledgment rate over time - - - {isLoadingMetrics ? ( -
-
-
- ) : chartData.ackRate.length > 0 ? ( - - ) : ( -
-
- -

No metrics data available

-

Metrics will appear once the consumer has activity

-
-
- )} -
-
-
- )}
); } diff --git a/apps/web/app/(dashboard)/consumers/page.tsx b/apps/web/app/(dashboard)/consumers/page.tsx index feea800..2354770 100644 --- a/apps/web/app/(dashboard)/consumers/page.tsx +++ b/apps/web/app/(dashboard)/consumers/page.tsx @@ -1,8 +1,8 @@ 'use client'; -import { useState } from 'react'; +import { useState, useEffect } from 'react'; import { useQuery, useMutation, useQueryClient } from '@tanstack/react-query'; -import { Plus, Users, Search, AlertTriangle, CheckCircle, ChevronRight, RefreshCw, Trash2 } from 'lucide-react'; +import { Plus, Users, Search, AlertTriangle, CheckCircle, ChevronRight, RefreshCw, Trash2, WifiOff } from 'lucide-react'; import Link from 'next/link'; import { api } from '@/lib/api'; import { Button } from '@/components/ui/button'; @@ -20,53 +20,74 @@ import { } from '@/components/ui/alert-dialog'; import { formatNumber, formatDuration } from '@nats-console/shared'; import { CreateConsumerDialog } from '@/components/forms/create-consumer-dialog'; +import { useClusterStore } from '@/stores/cluster'; export default function ConsumersPage() { const queryClient = useQueryClient(); - const [selectedCluster, setSelectedCluster] = useState(null); - const [selectedStream, setSelectedStream] = useState(null); + const { selectedClusterId, setSelectedClusterId } = useClusterStore(); + const [selectedStream, setSelectedStream] = useState('__all__'); const [search, setSearch] = useState(''); const [showCreateDialog, setShowCreateDialog] = useState(false); - const [consumerToDelete, setConsumerToDelete] = useState(null); + const [consumerToDelete, setConsumerToDelete] = useState<{ name: string; streamName: string } | null>(null); const { data: clustersData } = useQuery({ queryKey: ['clusters'], queryFn: () => api.clusters.list(), }); - const { data: streamsData } = useQuery({ - queryKey: ['streams', selectedCluster], - queryFn: () => (selectedCluster ? api.streams.list(selectedCluster) : null), - enabled: !!selectedCluster, + // Ensure cluster connection before fetching data + const { data: healthData, isLoading: isLoadingHealth, refetch: refetchHealth } = useQuery({ + queryKey: ['cluster-health', selectedClusterId], + queryFn: () => (selectedClusterId ? api.clusters.health(selectedClusterId) : null), + enabled: !!selectedClusterId, + staleTime: 30000, // Cache for 30 seconds + retry: 1, }); + const isClusterConnected = healthData?.connected === true; + const isClusterDisconnected = selectedClusterId && healthData && !healthData.connected; + + const { data: streamsData } = useQuery({ + queryKey: ['streams', selectedClusterId], + queryFn: () => (selectedClusterId ? api.streams.list(selectedClusterId) : null), + enabled: !!selectedClusterId && isClusterConnected, + }); + + // Fetch all consumers when "All" is selected, or specific stream consumers const { data: consumersData, isLoading, refetch } = useQuery({ - queryKey: ['consumers', selectedCluster, selectedStream], - queryFn: () => - selectedCluster && selectedStream - ? api.consumers.list(selectedCluster, selectedStream) - : null, - enabled: !!selectedCluster && !!selectedStream, + queryKey: ['consumers', selectedClusterId, selectedStream], + queryFn: () => { + if (!selectedClusterId) return null; + if (selectedStream === '__all__') { + return api.consumers.listAll(selectedClusterId); + } + return api.consumers.list(selectedClusterId, selectedStream); + }, + enabled: !!selectedClusterId && isClusterConnected, }); const deleteMutation = useMutation({ - mutationFn: (consumerName: string) => - api.consumers.delete(selectedCluster!, selectedStream!, consumerName), + mutationFn: ({ name, streamName }: { name: string; streamName: string }) => + api.consumers.delete(selectedClusterId!, streamName, name), onSuccess: () => { - queryClient.invalidateQueries({ queryKey: ['consumers', selectedCluster, selectedStream] }); + queryClient.invalidateQueries({ queryKey: ['consumers', selectedClusterId, selectedStream] }); setConsumerToDelete(null); }, }); - // Auto-select first cluster - if (clustersData?.clusters?.length && !selectedCluster) { - setSelectedCluster(clustersData.clusters[0].id); - } - - // Auto-select first stream when cluster changes - if (streamsData?.streams?.length && !selectedStream) { - setSelectedStream(streamsData.streams[0].config.name); - } + // Auto-select saved cluster or first cluster + useEffect(() => { + if (clustersData?.clusters?.length) { + const savedClusterExists = clustersData.clusters.some( + (c: any) => c.id === selectedClusterId + ); + if (!savedClusterExists) { + setSelectedClusterId(clustersData.clusters[0].id); + } else if (!selectedClusterId) { + setSelectedClusterId(clustersData.clusters[0].id); + } + } + }, [clustersData?.clusters, selectedClusterId, setSelectedClusterId]); const filteredConsumers = consumersData?.consumers?.filter((consumer: any) => consumer.name.toLowerCase().includes(search.toLowerCase()) @@ -87,19 +108,19 @@ export default function ConsumersPage() {

Manage JetStream consumers

- -
- {selectedCluster && selectedStream && ( + {selectedClusterId && selectedStream && selectedStream !== '__all__' && ( )} @@ -108,10 +129,10 @@ export default function ConsumersPage() {
setMetricsTimeRange(e.target.value)} + > + + + + + + + + {isLoadingMetrics ? ( +
+
+
+ ) : chartData.messages.length > 0 ? ( + + ) : ( +
+
+ +

No metrics data available

+
+
+ )} +
+ + + + + Data Throughput + Bytes per second over time + + + {isLoadingMetrics ? ( +
+
+
+ ) : chartData.bytes.length > 0 ? ( + + ) : ( +
+
+ +

No metrics data available

+
+
+ )} +
+
)} @@ -1052,122 +1122,6 @@ function StreamDetailContent() { )} - {/* Metrics Tab */} - {activeTab === 'metrics' && ( -
- {/* Time Range Selector */} -
- -
- - {/* Quick Stats */} -
- - - Current Messages - - -
{formatNumber(stream.state?.messages || 0)}
-
-
- - - Current Size - - -
{formatBytes(stream.state?.bytes || 0)}
-
-
- - - First Sequence - - -
{formatNumber(stream.state?.first_seq || 0)}
-
-
- - - Last Sequence - - -
{formatNumber(stream.state?.last_seq || 0)}
-
-
-
- - {/* Message Throughput Chart */} - - - Message Throughput - Messages per second over time - - - {isLoadingMetrics ? ( -
-
-
- ) : chartData.messages.length > 0 ? ( - - ) : ( -
-
- -

No metrics data available

-

Metrics will appear once the stream has activity

-
-
- )} -
-
- - {/* Bytes Throughput Chart */} - - - Data Throughput - Bytes per second over time - - - {isLoadingMetrics ? ( -
-
-
- ) : chartData.bytes.length > 0 ? ( - - ) : ( -
-
- -

No metrics data available

-

Metrics will appear once the stream has activity

-
-
- )} -
-
-
- )} ); } diff --git a/apps/web/app/(dashboard)/streams/page.tsx b/apps/web/app/(dashboard)/streams/page.tsx index 407ca65..933bcc7 100644 --- a/apps/web/app/(dashboard)/streams/page.tsx +++ b/apps/web/app/(dashboard)/streams/page.tsx @@ -1,8 +1,8 @@ 'use client'; -import { useState } from 'react'; +import { useState, useEffect } from 'react'; import { useQuery } from '@tanstack/react-query'; -import { Plus, Database, Search, ChevronRight, RefreshCw } from 'lucide-react'; +import { Plus, Database, Search, ChevronRight, RefreshCw, WifiOff } from 'lucide-react'; import Link from 'next/link'; import { api } from '@/lib/api'; import { Button } from '@/components/ui/button'; @@ -10,9 +10,10 @@ import { Input } from '@/components/ui/input'; import { Card, CardContent, CardDescription, CardHeader, CardTitle } from '@/components/ui/card'; import { formatBytes, formatNumber } from '@nats-console/shared'; import { CreateStreamDialog } from '@/components/forms/create-stream-dialog'; +import { useClusterStore } from '@/stores/cluster'; export default function StreamsPage() { - const [selectedCluster, setSelectedCluster] = useState(null); + const { selectedClusterId, setSelectedClusterId } = useClusterStore(); const [search, setSearch] = useState(''); const [showCreateDialog, setShowCreateDialog] = useState(false); @@ -21,16 +22,38 @@ export default function StreamsPage() { queryFn: () => api.clusters.list(), }); - const { data: streamsData, isLoading, refetch } = useQuery({ - queryKey: ['streams', selectedCluster], - queryFn: () => (selectedCluster ? api.streams.list(selectedCluster) : null), - enabled: !!selectedCluster, + // Ensure cluster connection before fetching streams + const { data: healthData, isLoading: isLoadingHealth, refetch: refetchHealth } = useQuery({ + queryKey: ['cluster-health', selectedClusterId], + queryFn: () => (selectedClusterId ? api.clusters.health(selectedClusterId) : null), + enabled: !!selectedClusterId, + staleTime: 30000, // Cache for 30 seconds + retry: 1, }); - // Auto-select first cluster - if (clustersData?.clusters?.length && !selectedCluster) { - setSelectedCluster(clustersData.clusters[0].id); - } + const isClusterConnected = healthData?.connected === true; + const isClusterDisconnected = selectedClusterId && healthData && !healthData.connected; + + const { data: streamsData, isLoading, refetch } = useQuery({ + queryKey: ['streams', selectedClusterId], + queryFn: () => (selectedClusterId ? api.streams.list(selectedClusterId) : null), + enabled: !!selectedClusterId && isClusterConnected, + }); + + // Auto-select saved cluster or first cluster + useEffect(() => { + if (clustersData?.clusters?.length) { + // Check if saved cluster still exists + const savedClusterExists = clustersData.clusters.some( + (c: any) => c.id === selectedClusterId + ); + if (!savedClusterExists) { + setSelectedClusterId(clustersData.clusters[0].id); + } else if (!selectedClusterId) { + setSelectedClusterId(clustersData.clusters[0].id); + } + } + }, [clustersData?.clusters, selectedClusterId, setSelectedClusterId]); const filteredStreams = streamsData?.streams?.filter((stream: any) => stream.config.name.toLowerCase().includes(search.toLowerCase()) @@ -44,19 +67,19 @@ export default function StreamsPage() {

Manage JetStream streams

- -
- {selectedCluster && ( + {selectedClusterId && ( )} @@ -64,8 +87,8 @@ export default function StreamsPage() {