konnosDashboard
ReplicationPipelineStatus.utils.tsx139 lines · main
1import dayjs from 'dayjs'
2import { Badge } from 'ui'
3
4import { getPipelineDisplayState, normalizePipelineStatusName } from '../Pipeline.utils'
5import { RetryPolicy, TableState } from './ReplicationPipelineStatus.types'
6import { ReplicationPipelineStatusData } from '@/data/replication/pipeline-status-query'
7import { formatBytes } from '@/lib/helpers'
8import { PipelineStatusRequestStatus } from '@/state/replication-pipeline-request-status'
9
10const numberFormatter = new Intl.NumberFormat()
11
12export const getStatusConfig = (state: TableState['state']) => {
13 switch (state.name) {
14 case 'queued':
15 return {
16 badge: <Badge variant="warning">Queued</Badge>,
17 description: 'Table is waiting for ETL to pick it up for replication.',
18 tooltip: 'Table is waiting for ETL to pick it up for replication.',
19 color: 'text-warning',
20 }
21 case 'copying_table':
22 return {
23 badge: <Badge variant="success">Copying</Badge>,
24 description: "Table's existing rows are being copied before live streaming begins.",
25 tooltip: "Table's existing rows are being copied before live streaming begins.",
26 color: 'text-brand-600',
27 }
28 case 'copied_table':
29 return {
30 badge: <Badge variant="success">Copied</Badge>,
31 description: "Table copy is complete and it's preparing to follow WAL changes.",
32 tooltip: "Table copy is complete and it's preparing to follow WAL changes.",
33 color: 'text-success-600',
34 }
35 case 'following_wal':
36 return {
37 badge: <Badge variant="success">Live</Badge>,
38 description: 'Table is streaming new changes in real time from the WAL.',
39 tooltip: 'Table is streaming new changes in real time from the WAL.',
40 color: 'text-success-600',
41 }
42 case 'error':
43 return {
44 badge: <Badge variant="destructive">Error</Badge>,
45 description: 'Replication is paused because the table encountered an error.',
46 tooltip: 'Replication is paused because the table encountered an error.',
47 color: 'text-destructive-600',
48 }
49 default:
50 return {
51 badge: <Badge variant="warning">Unknown</Badge>,
52 description: 'Table status is unavailable.',
53 tooltip: 'Table status is unavailable.',
54 color: 'text-warning',
55 }
56 }
57}
58
59export const getDisabledStateConfig = ({
60 requestStatus,
61 statusName,
62}: {
63 requestStatus: PipelineStatusRequestStatus
64 statusName?: ReplicationPipelineStatusData['status']['name']
65}) => {
66 const normalizedStatusName = normalizePipelineStatusName(statusName)
67 const displayState = getPipelineDisplayState(requestStatus, normalizedStatusName)
68 const { title, message } = displayState
69
70 return { title, message }
71}
72
73export const isValidRetryPolicy = (policy: any): policy is RetryPolicy => {
74 if (!policy || typeof policy !== 'object' || !policy.policy) return false
75
76 switch (policy.policy) {
77 case 'no_retry':
78 case 'manual_retry':
79 return true
80 case 'timed_retry':
81 return typeof policy.next_retry === 'string'
82 default:
83 return false
84 }
85}
86
87const formatLagBytesValue = (value?: number) => {
88 if (typeof value !== 'number' || Number.isNaN(value)) {
89 return { display: '—', detail: undefined }
90 }
91
92 const decimals = value < 1024 ? 0 : value < 1024 * 1024 ? 1 : 2
93 const display = formatBytes(value, decimals)
94 const detail = `${numberFormatter.format(value)} bytes`
95
96 return { display, detail }
97}
98
99const formatLagDurationValue = (value?: number) => {
100 if (typeof value !== 'number' || Number.isNaN(value)) {
101 return { display: '—', detail: undefined }
102 }
103
104 const sign = value < 0 ? '-' : ''
105 const absMilliseconds = Math.abs(value)
106 const duration = dayjs.duration(absMilliseconds, 'milliseconds')
107
108 if (absMilliseconds < 1000) {
109 return { display: `${value} ms`, detail: undefined }
110 }
111
112 const seconds = duration.asSeconds()
113 if (seconds < 60) {
114 const decimals = seconds >= 10 ? 1 : 2
115 return {
116 display: `${sign}${seconds.toFixed(decimals)} s`,
117 detail: `${numberFormatter.format(value)} ms`,
118 }
119 }
120
121 const minutes = duration.asMinutes()
122 if (minutes < 60) {
123 const roundedSeconds = Math.round(seconds)
124 return {
125 display: `${sign}${minutes.toFixed(minutes >= 10 ? 1 : 2)} min`,
126 detail: `${numberFormatter.format(roundedSeconds)} s`,
127 }
128 }
129
130 const hours = duration.asHours()
131 const roundedMinutes = Math.round(minutes)
132 return {
133 display: `${sign}${hours.toFixed(hours >= 10 ? 1 : 2)} h`,
134 detail: `${numberFormatter.format(roundedMinutes)} min`,
135 }
136}
137
138export const getFormattedLagValue = (type: 'bytes' | 'duration', value?: number) =>
139 type === 'bytes' ? formatLagBytesValue(value) : formatLagDurationValue(value)