konnosDashboard
ReplicationPipelineStatus.tsx631 lines · main
1import { useParams } from 'common'
2import {
3 Activity,
4 ArrowUpCircle,
5 Ban,
6 ChevronDown,
7 ChevronLeft,
8 Info,
9 Pause,
10 Play,
11 RotateCcw,
12 Search,
13 WifiOff,
14 X,
15} from 'lucide-react'
16import Link from 'next/link'
17import { parseAsString, useQueryState } from 'nuqs'
18import { useEffect, useMemo, useState } from 'react'
19import { toast } from 'sonner'
20import {
21 Button,
22 Card,
23 CardContent,
24 DropdownMenu,
25 DropdownMenuContent,
26 DropdownMenuTrigger,
27 Table,
28 TableBody,
29 TableHead,
30 TableHeader,
31 TableRow,
32} from 'ui'
33import { GenericSkeletonLoader } from 'ui-patterns'
34import { Input } from 'ui-patterns/DataInputs/Input'
35
36import { BatchRestartDialog } from '../BatchRestartDialog'
37import { ErrorDetailsDialog } from '../ErrorDetailsDialog'
38import {
39 getPipelineDisplayState,
40 getStatusName,
41 PIPELINE_ACTIONABLE_STATES,
42} from '../Pipeline.utils'
43import { PipelineStatus } from '../PipelineStatus'
44import { PipelineStatusName, STATUS_REFRESH_FREQUENCY_MS } from '../Replication.constants'
45import { RestartTableDialog } from '../RestartTableDialog'
46import { UpdateVersionModal } from '../UpdateVersionModal'
47import { SlotLagMetrics } from './ReplicationPipelineStatus.types'
48import { getDisabledStateConfig } from './ReplicationPipelineStatus.utils'
49import { SlotLagMetricsInline, SlotLagMetricsList } from './SlotLagMetrics'
50import { TableReplicationRow } from './TableReplicationRow'
51import { AlertError } from '@/components/ui/AlertError'
52import { DropdownMenuItemTooltip } from '@/components/ui/DropdownMenuItemTooltip'
53import { useReplicationPipelineByIdQuery } from '@/data/replication/pipeline-by-id-query'
54import { useReplicationPipelineReplicationStatusQuery } from '@/data/replication/pipeline-replication-status-query'
55import { useReplicationPipelineStatusQuery } from '@/data/replication/pipeline-status-query'
56import { useReplicationPipelineVersionQuery } from '@/data/replication/pipeline-version-query'
57import { useRestartPipelineHelper } from '@/data/replication/restart-pipeline-helper'
58import { useStartPipelineMutation } from '@/data/replication/start-pipeline-mutation'
59import { useStopPipelineMutation } from '@/data/replication/stop-pipeline-mutation'
60import {
61 PipelineStatusRequestStatus,
62 usePipelineRequestStatus,
63} from '@/state/replication-pipeline-request-status'
64import { type ResponseError } from '@/types'
65
66/**
67 * Component for displaying replication pipeline status and table replication details.
68 * Supports both legacy 'error' state and new 'errored' state with retry policies.
69 */
70export const ReplicationPipelineStatus = () => {
71 const { ref: projectRef, pipelineId: _pipelineId } = useParams()
72 const [searchString, setSearchString] = useQueryState('search', parseAsString.withDefault(''))
73
74 const [showUpdateVersionModal, setShowUpdateVersionModal] = useState(false)
75 const [showErrorDialog, setShowErrorDialog] = useState(false)
76 const [selectedTableError, setSelectedTableError] = useState<{
77 tableName: string
78 reason: string
79 solution?: string
80 } | null>(null)
81 const [showRestartDialog, setShowRestartDialog] = useState(false)
82 const [selectedTableForRestart, setSelectedTableForRestart] = useState<{
83 tableId: number
84 tableName: string
85 } | null>(null)
86 const [showBatchRestartDialog, setShowBatchRestartDialog] = useState(false)
87 const [batchRestartMode, setBatchRestartMode] = useState<'all' | 'errored' | null>(null)
88 const [restartingTableIds, setRestartingTableIds] = useState<Set<number>>(new Set())
89
90 const pipelineId = Number(_pipelineId)
91 const { getRequestStatus, updatePipelineStatus, setRequestStatus } = usePipelineRequestStatus()
92 const requestStatus = getRequestStatus(pipelineId)
93
94 const {
95 data: pipeline,
96 error: pipelineError,
97 isPending: isPipelineLoading,
98 isError: isPipelineError,
99 } = useReplicationPipelineByIdQuery({
100 projectRef,
101 pipelineId,
102 })
103
104 const {
105 data: pipelineStatusData,
106 error: pipelineStatusError,
107 isLoading: isPipelineStatusLoading,
108 isError: isPipelineStatusError,
109 isSuccess: isPipelineStatusSuccess,
110 } = useReplicationPipelineStatusQuery(
111 { projectRef, pipelineId },
112 {
113 enabled: !!pipelineId,
114 refetchInterval: STATUS_REFRESH_FREQUENCY_MS,
115 }
116 )
117
118 const {
119 data: replicationStatusData,
120 isPending: isStatusLoading,
121 isError: isStatusError,
122 } = useReplicationPipelineReplicationStatusQuery(
123 { projectRef, pipelineId },
124 {
125 enabled: !!pipelineId,
126 refetchInterval: STATUS_REFRESH_FREQUENCY_MS,
127 }
128 )
129
130 const { data: versionData } = useReplicationPipelineVersionQuery({
131 projectRef,
132 pipelineId: pipeline?.id,
133 })
134 const hasUpdate = Boolean(versionData?.new_version)
135
136 const { mutateAsync: startPipeline, isPending: isStartingPipeline } = useStartPipelineMutation()
137 const { mutateAsync: stopPipeline, isPending: isStoppingPipeline } = useStopPipelineMutation()
138 const { restartPipeline } = useRestartPipelineHelper()
139
140 const destinationName = pipeline?.destination_name
141 const statusName = getStatusName(pipelineStatusData?.status)
142 const displayState = getPipelineDisplayState(requestStatus, statusName)
143 const config = getDisabledStateConfig({ requestStatus, statusName })
144
145 // Sort tables by name for consistent ordering (memoized)
146 const tableStatuses = useMemo(
147 () =>
148 (replicationStatusData?.table_statuses || []).sort((a, b) =>
149 a.table_name.localeCompare(b.table_name)
150 ),
151 [replicationStatusData?.table_statuses]
152 )
153
154 const applyLagMetrics = replicationStatusData?.apply_lag
155
156 // Filter tables based on search (memoized)
157 const filteredTableStatuses = useMemo(
158 () =>
159 searchString.length === 0
160 ? tableStatuses
161 : tableStatuses.filter((table) =>
162 table.table_name.toLowerCase().includes(searchString.toLowerCase())
163 ),
164 [tableStatuses, searchString]
165 )
166
167 const tablesWithLag = useMemo(
168 () => tableStatuses.filter((table) => Boolean(table.table_sync_lag)),
169 [tableStatuses]
170 )
171
172 const erroredTables = useMemo(
173 () =>
174 tableStatuses.filter(
175 (table) =>
176 table.state.name === 'error' &&
177 'retry_policy' in table.state &&
178 table.state.retry_policy?.policy === 'manual_retry'
179 ),
180 [tableStatuses]
181 )
182
183 const hasErroredTables = erroredTables.length > 0
184 const isAnyRestartInProgress = restartingTableIds.size > 0
185
186 const hasTableData = tableStatuses.length > 0
187 const isPipelineActionable =
188 statusName === PipelineStatusName.STARTED ||
189 statusName === PipelineStatusName.STOPPED ||
190 statusName === PipelineStatusName.FAILED
191 const isEnablingDisabling =
192 requestStatus === PipelineStatusRequestStatus.StartRequested ||
193 requestStatus === PipelineStatusRequestStatus.StopRequested ||
194 requestStatus === PipelineStatusRequestStatus.RestartRequested
195 const isPipelineBusy = isEnablingDisabling || isAnyRestartInProgress
196 const showDisabledState = isPipelineBusy || !isPipelineActionable
197 const lastKnownStateMessage =
198 statusName === PipelineStatusName.STOPPED
199 ? 'Showing the last known table state before the pipeline was stopped.'
200 : statusName === PipelineStatusName.FAILED
201 ? 'Showing the last reported table state before the pipeline failed.'
202 : null
203 const refreshIntervalLabel =
204 STATUS_REFRESH_FREQUENCY_MS >= 1000
205 ? `${Math.round(STATUS_REFRESH_FREQUENCY_MS / 1000)}s`
206 : `${STATUS_REFRESH_FREQUENCY_MS}ms`
207
208 const logsUrl = `/project/${projectRef}/logs/replication-logs${
209 pipelineId ? `?f=${encodeURIComponent(JSON.stringify({ pipeline_id: pipelineId }))}` : ''
210 }`
211
212 const label = isEnablingDisabling
213 ? displayState.label
214 : statusName === PipelineStatusName.STOPPED
215 ? 'Start'
216 : statusName === PipelineStatusName.STARTED
217 ? 'Stop'
218 : statusName === PipelineStatusName.FAILED
219 ? 'Restart'
220 : displayState.label
221
222 const icon =
223 statusName === PipelineStatusName.STOPPED ? (
224 <Play />
225 ) : statusName === PipelineStatusName.STARTED ? (
226 <Pause />
227 ) : statusName === PipelineStatusName.FAILED ? (
228 <RotateCcw />
229 ) : (
230 <Ban />
231 )
232
233 const onPrimaryAction = async () => {
234 if (!projectRef) return console.error('Project ref is required')
235 if (!pipeline) return toast.error('No pipeline found')
236
237 const action =
238 statusName === PipelineStatusName.STOPPED
239 ? 'start'
240 : statusName === PipelineStatusName.STARTED
241 ? 'stop'
242 : 'restart'
243 try {
244 if (statusName === PipelineStatusName.STOPPED) {
245 setRequestStatus(pipeline.id, PipelineStatusRequestStatus.StartRequested, statusName)
246 await startPipeline({ projectRef, pipelineId: pipeline.id })
247 } else if (statusName === PipelineStatusName.STARTED) {
248 setRequestStatus(pipeline.id, PipelineStatusRequestStatus.StopRequested, statusName)
249 await stopPipeline({ projectRef, pipelineId: pipeline.id })
250 } else if (statusName === PipelineStatusName.FAILED) {
251 setRequestStatus(pipeline.id, PipelineStatusRequestStatus.RestartRequested, statusName)
252 await restartPipeline({ projectRef, pipelineId: pipeline.id })
253 }
254 } catch (error) {
255 setRequestStatus(pipeline.id, PipelineStatusRequestStatus.None)
256 toast.error(`Failed to ${action} pipeline: ${(error as ResponseError).message}`)
257 }
258 }
259
260 useEffect(() => {
261 updatePipelineStatus(pipelineId, statusName)
262 }, [pipelineId, statusName, updatePipelineStatus])
263
264 return (
265 <>
266 <div className="flex flex-col gap-y-4">
267 <div className="flex items-center justify-between">
268 <div className="flex items-center gap-x-3">
269 <Button asChild type="outline" icon={<ChevronLeft />} style={{ padding: '5px' }}>
270 <Link href={`/project/${projectRef}/database/replication`} />
271 </Button>
272 <div className="flex items-center gap-x-3">
273 <h3 className="text-xl font-semibold">{destinationName || 'Pipeline'}</h3>
274 <PipelineStatus
275 pipelineStatus={pipelineStatusData?.status}
276 error={pipelineStatusError}
277 isLoading={isPipelineStatusLoading}
278 isError={isPipelineStatusError}
279 isSuccess={isPipelineStatusSuccess}
280 requestStatus={requestStatus}
281 pipelineId={pipelineId}
282 />
283 </div>
284 </div>
285
286 <div className="flex items-center gap-x-2">
287 {hasUpdate && (
288 <Button
289 type="primary"
290 icon={<ArrowUpCircle />}
291 onClick={() => setShowUpdateVersionModal(true)}
292 >
293 Update available
294 </Button>
295 )}
296
297 <Button asChild type="default">
298 <Link href={logsUrl}>View logs</Link>
299 </Button>
300
301 <Button
302 type={statusName === PipelineStatusName.STOPPED ? 'primary' : 'default'}
303 onClick={onPrimaryAction}
304 loading={
305 isPipelineError ||
306 displayState.type === 'loading' ||
307 isEnablingDisabling ||
308 isStartingPipeline ||
309 isStoppingPipeline ||
310 isAnyRestartInProgress
311 }
312 disabled={
313 isPipelineBusy ||
314 !PIPELINE_ACTIONABLE_STATES.includes(statusName as PipelineStatusName)
315 }
316 icon={icon}
317 className="capitalize"
318 >
319 {label}
320 </Button>
321 </div>
322 </div>
323 {isPipelineError && (
324 <AlertError error={pipelineError} subject="Failed to retrieve pipeline information" />
325 )}
326
327 {isStatusError && (
328 <div className="flex items-center gap-2 rounded-lg border border-warning-400 bg-warning-50 px-3 py-2 text-xs text-warning-800">
329 <WifiOff size={14} />
330 <span className="font-medium">Live updates paused</span>
331 <span className="text-warning-700">Retrying automatically</span>
332 </div>
333 )}
334
335 {(isPipelineLoading || isStatusLoading) && (
336 <div className="space-y-3">
337 <div className="flex items-center gap-x-3">
338 <div className="h-6 w-40 rounded-sm bg-surface-200" />
339 <div className="h-5 w-24 rounded-sm bg-surface-200" />
340 </div>
341 <GenericSkeletonLoader />
342 </div>
343 )}
344
345 {applyLagMetrics && (
346 <div className="border border-default rounded-lg bg-surface-100 px-4 py-4 space-y-3">
347 <div className="flex flex-wrap items-baseline justify-between gap-y-1">
348 <div>
349 <h4 className="text-sm font-semibold text-foreground">Replication lag</h4>
350 <p className="text-xs text-foreground-light">
351 Snapshot of how far this pipeline is trailing behind right now.
352 </p>
353 </div>
354 <p className="text-xs text-foreground-lighter">
355 Updates every {refreshIntervalLabel}
356 </p>
357 </div>
358
359 {isStatusError && (
360 <p className="text-xs text-warning-700">
361 Unable to refresh data. Showing the last values we received.
362 </p>
363 )}
364
365 <SlotLagMetricsList metrics={applyLagMetrics} />
366
367 {tablesWithLag.length > 0 && (
368 <>
369 <div className="border-t border-default/40" />
370 <div className="space-y-3 text-xs text-foreground">
371 <div className="flex items-start gap-2 rounded-md border border-default/50 bg-surface-200/60 px-3 py-2 text-foreground-light">
372 <Info size={14} className="mt-0.5" />
373 <span>
374 During initial sync, tables can copy and stream independently before
375 reconciling with the overall pipeline.
376 </span>
377 </div>
378 <div className="rounded-sm border border-default/50 bg-surface-200/40">
379 <ul className="divide-y divide-default/40">
380 {tablesWithLag.map((table) => (
381 <li key={`${table.table_id}-${table.table_name}`} className="px-3 py-2">
382 <SlotLagMetricsInline
383 tableName={table.table_name}
384 metrics={table.table_sync_lag as SlotLagMetrics}
385 />
386 </li>
387 ))}
388 </ul>
389 </div>
390 </div>
391 </>
392 )}
393 </div>
394 )}
395
396 {!isPipelineLoading && !isStatusLoading && hasTableData && (
397 <div className="flex flex-col gap-y-3">
398 <div className="flex items-center justify-between">
399 <Input
400 icon={<Search />}
401 size="tiny"
402 className="text-xs w-52"
403 placeholder="Search for tables"
404 value={searchString}
405 disabled={isPipelineError}
406 onChange={(e) => setSearchString(e.target.value)}
407 actions={
408 searchString.length > 0 && [
409 <X
410 key="close"
411 className="mx-2 cursor-pointer text-foreground"
412 size={14}
413 strokeWidth={1.5}
414 onClick={() => setSearchString('')}
415 />,
416 ]
417 }
418 />
419 <div className="flex items-center">
420 <Button
421 size="tiny"
422 type="default"
423 className="rounded-r-none hover:z-2"
424 icon={<RotateCcw />}
425 disabled={isAnyRestartInProgress || showDisabledState || isPipelineError}
426 loading={isAnyRestartInProgress}
427 onClick={() => {
428 setBatchRestartMode('all')
429 setShowBatchRestartDialog(true)
430 }}
431 >
432 Restart all tables
433 </Button>
434 <DropdownMenu>
435 <DropdownMenuTrigger asChild>
436 <Button
437 type="default"
438 icon={<ChevronDown />}
439 className="w-7 rounded-l-none -ml-px"
440 disabled={showDisabledState || isPipelineError}
441 />
442 </DropdownMenuTrigger>
443 <DropdownMenuContent align="end" className="w-44">
444 <DropdownMenuItemTooltip
445 disabled={!hasErroredTables || isAnyRestartInProgress || showDisabledState}
446 onClick={() => {
447 setBatchRestartMode('errored')
448 setShowBatchRestartDialog(true)
449 }}
450 tooltip={{
451 content: {
452 side: 'left',
453 text: !hasErroredTables ? 'No tables require manual retry' : undefined,
454 },
455 }}
456 >
457 Restart failed tables only
458 </DropdownMenuItemTooltip>
459 </DropdownMenuContent>
460 </DropdownMenu>
461 </div>
462 </div>
463
464 {lastKnownStateMessage !== null && !showDisabledState && (
465 <div className="flex items-start gap-2 rounded-md border border-default/50 bg-surface-200/60 px-3 py-2 text-xs text-foreground-light">
466 <Info size={14} className="mt-0.5" />
467 <span>{lastKnownStateMessage}</span>
468 </div>
469 )}
470
471 <Card>
472 <CardContent className="p-0">
473 <Table>
474 <TableHeader>
475 <TableRow>
476 <TableHead key="table">Table</TableHead>
477 <TableHead key="status">Status</TableHead>
478 <TableHead key="details">Details</TableHead>
479 <TableHead key="actions" />
480 </TableRow>
481 </TableHeader>
482 <TableBody>
483 {filteredTableStatuses.map((table) => {
484 const isRestarting = restartingTableIds.has(table.table_id)
485 const isErrorState = table.state.name === 'error'
486 const errorReason =
487 isErrorState && 'reason' in table.state ? table.state.reason : undefined
488 const errorSolution =
489 isErrorState && 'solution' in table.state ? table.state.solution : undefined
490 return (
491 <TableReplicationRow
492 key={table.table_id}
493 table={table}
494 isRestarting={isRestarting}
495 showDisabledState={showDisabledState}
496 disabledStateMessage={config.message}
497 isAnyRestartInProgress={isAnyRestartInProgress}
498 isPipelineStopped={statusName === PipelineStatusName.STOPPED}
499 onSelectRestart={() => {
500 setSelectedTableForRestart({
501 tableId: table.table_id,
502 tableName: table.table_name,
503 })
504 setShowRestartDialog(true)
505 }}
506 onSelectShowError={
507 isErrorState && errorReason
508 ? () => {
509 setSelectedTableError({
510 tableName: table.table_name,
511 reason: errorReason,
512 solution: errorSolution,
513 })
514 setShowErrorDialog(true)
515 }
516 : () => {}
517 }
518 />
519 )
520 })}
521 </TableBody>
522 </Table>
523 </CardContent>
524 </Card>
525 </div>
526 )}
527
528 {!isPipelineLoading && !isStatusLoading && tableStatuses.length === 0 && (
529 <div className="flex flex-col items-center justify-center py-16 px-4 border rounded-lg border-dashed">
530 <div className="w-full max-w-sm mx-auto text-center space-y-4">
531 <div className="w-16 h-16 bg-surface-200 rounded-full flex items-center justify-center mx-auto">
532 <Activity className="w-8 h-8 text-foreground-lighter" />
533 </div>
534 <div className="space-y-2">
535 <h4 className="text-lg font-semibold text-foreground">
536 {showDisabledState
537 ? config.title
538 : statusName === PipelineStatusName.STOPPED
539 ? 'Pipeline stopped'
540 : statusName === PipelineStatusName.FAILED
541 ? 'Pipeline failed'
542 : 'No table data yet'}
543 </h4>
544 <p className="text-sm text-foreground-light leading-relaxed">
545 {showDisabledState
546 ? config.message
547 : statusName === PipelineStatusName.STOPPED
548 ? 'Start the pipeline to begin replication.'
549 : statusName === PipelineStatusName.FAILED
550 ? 'The pipeline encountered an error. Restart it or reset your tables to recover.'
551 : 'Table status will appear here once replication begins.'}
552 </p>
553 </div>
554 {statusName !== PipelineStatusName.STOPPED && (
555 <p className="text-xs text-foreground-lighter">
556 Data refreshes every {refreshIntervalLabel}
557 </p>
558 )}
559 </div>
560 </div>
561 )}
562 </div>
563
564 <UpdateVersionModal
565 visible={showUpdateVersionModal}
566 pipeline={pipeline}
567 onClose={() => setShowUpdateVersionModal(false)}
568 confirmLabel={
569 statusName === PipelineStatusName.STARTED || statusName === PipelineStatusName.FAILED
570 ? 'Update and restart'
571 : 'Update version'
572 }
573 />
574
575 {/* Restart Table Confirmation Dialog */}
576 {selectedTableForRestart && (
577 <RestartTableDialog
578 open={showRestartDialog}
579 onOpenChange={setShowRestartDialog}
580 tableId={selectedTableForRestart.tableId}
581 tableName={selectedTableForRestart.tableName}
582 pipelineStatusName={statusName}
583 onRestartStart={() => {
584 setRestartingTableIds((prev) => new Set(prev).add(selectedTableForRestart.tableId))
585 }}
586 onRestartComplete={() => {
587 setRestartingTableIds((prev) => {
588 const next = new Set(prev)
589 next.delete(selectedTableForRestart.tableId)
590 return next
591 })
592 }}
593 />
594 )}
595
596 {/* Error Details Dialog */}
597 {selectedTableError && (
598 <ErrorDetailsDialog
599 open={showErrorDialog}
600 onOpenChange={setShowErrorDialog}
601 tableName={selectedTableError.tableName}
602 reason={selectedTableError.reason}
603 solution={selectedTableError.solution}
604 />
605 )}
606
607 {/* Batch Restart Dialog */}
608 {batchRestartMode && (
609 <BatchRestartDialog
610 open={showBatchRestartDialog}
611 onOpenChange={setShowBatchRestartDialog}
612 mode={batchRestartMode}
613 totalTables={tableStatuses.length}
614 erroredTablesCount={erroredTables.length}
615 tables={tableStatuses}
616 pipelineStatusName={statusName}
617 onRestartStart={(tableIds) => {
618 setRestartingTableIds((prev) => new Set([...prev, ...tableIds]))
619 }}
620 onRestartComplete={(tableIds) => {
621 setRestartingTableIds((prev) => {
622 const next = new Set(prev)
623 tableIds.forEach((id) => next.delete(id))
624 return next
625 })
626 }}
627 />
628 )}
629 </>
630 )
631}