import dayjs from 'dayjs' import duration from 'dayjs/plugin/duration' import { getPipelineDisplayState, normalizePipelineStatusName } from '../Pipeline.utils' import { RetryPolicy, SlotLagMetrics, SlotWalStatus, TableState, } from './ReplicationPipelineStatus.types' import type { StateDotVariant } from '@/components/ui/StateDot' import { ReplicationPipelineStatusData } from '@/data/replication/pipeline-status-query' import { formatBytes } from '@/lib/helpers' import { PipelineStatusRequestStatus } from '@/state/replication-pipeline-request-status' dayjs.extend(duration) export const getStatusConfig = ( state: TableState['state'] ): { variant: StateDotVariant; label: string; description: string; isPulsing?: boolean } => { switch (state.name) { case 'queued': return { variant: 'default', label: 'Queued', description: 'Waiting to copy' } case 'copying_table': return { variant: 'default', label: 'Copying', description: 'Copying existing rows', isPulsing: true, } case 'copied_table': return { variant: 'default', label: 'Copied', description: 'Copy finished, about to start streaming', } case 'following_wal': return { variant: 'success', label: 'Live', description: 'Streaming changes as they happen' } case 'error': return { variant: 'destructive', label: 'Error', description: 'Stopped after an error' } default: return { variant: 'warning', label: 'Unknown', description: 'Table status is unavailable' } } } export const getDisabledStateConfig = ({ requestStatus, statusName, }: { requestStatus: PipelineStatusRequestStatus statusName?: ReplicationPipelineStatusData['status']['name'] }) => { const normalizedStatusName = normalizePipelineStatusName(statusName) const displayState = getPipelineDisplayState(requestStatus, normalizedStatusName) const { title, message } = displayState return { title, message } } export const isValidRetryPolicy = (policy: any): policy is RetryPolicy => { if (!policy || typeof policy !== 'object' || !policy.policy) return false switch (policy.policy) { case 'no_retry': case 'manual_retry': return true case 'timed_retry': return typeof policy.next_retry === 'string' default: return false } } const formatLagBytesValue = (value?: number) => { if (typeof value !== 'number' || Number.isNaN(value)) { return { display: 'n/a', detail: undefined } } // Scale to the most readable unit (e.g. "4 GB"). We intentionally don't surface the raw byte // count as a detail line, since it's unreadable at GB scale (e.g. "4,294,967,296 bytes"). const decimals = value < 1024 ? 0 : value < 1024 * 1024 ? 1 : 2 return { display: formatBytes(value, decimals), detail: undefined } } // Scale to a single readable unit (ms, s, min, h) based on size, with no precise sub-line. const formatLagDurationValue = (value?: number) => { if (typeof value !== 'number' || Number.isNaN(value)) { return { display: 'n/a', detail: undefined } } const sign = value < 0 ? '-' : '' const absMilliseconds = Math.abs(value) const duration = dayjs.duration(absMilliseconds, 'milliseconds') if (absMilliseconds < 1000) { return { display: `${value} ms`, detail: undefined } } const seconds = duration.asSeconds() if (seconds < 60) { return { display: `${sign}${seconds.toFixed(seconds >= 10 ? 1 : 2)} s`, detail: undefined } } const minutes = duration.asMinutes() if (minutes < 60) { return { display: `${sign}${minutes.toFixed(minutes >= 10 ? 1 : 2)} min`, detail: undefined } } const hours = duration.asHours() return { display: `${sign}${hours.toFixed(hours >= 10 ? 1 : 2)} h`, detail: undefined } } export const getFormattedLagValue = (type: 'bytes' | 'duration', value?: number) => type === 'bytes' ? formatLagBytesValue(value) : formatLagDurationValue(value) const COPYING_STATES: TableState['state']['name'][] = ['queued', 'copying_table', 'copied_table'] /** * How much of the initial copy is left. A pipeline can be caught up on its ongoing change stream * while tables are still copying, so several surfaces need to know this to stay honest. */ export const getInitialSyncProgress = ( tableStatuses: { state: { name: TableState['state']['name'] } }[] ) => { const count = (name: TableState['state']['name']) => tableStatuses.filter((table) => table.state.name === name).length return { // Everything not yet streaming, whatever stage of the initial sync it is at syncingCount: tableStatuses.filter((table) => COPYING_STATES.includes(table.state.name)).length, copyingCount: count('copying_table'), queuedCount: count('queued'), totalCount: tableStatuses.length, } } export type LagSeverity = 'normal' | 'warning' | 'critical' type SlotStatusBadgeVariant = 'success' | 'warning' | 'destructive' | 'default' interface WalStatusMeta { label: string variant: SlotStatusBadgeVariant severity: LagSeverity // Shown in the pipeline-level metrics panel. description: string // Shown in the per-table inline sync view where the slot belongs to a single table. tableDescription: string } // Plain-language meaning, color, and severity for each WAL status Postgres can report for a slot. // `variant` drives the badge color; `severity` drives whether the list view raises a warning icon // (e.g. "extended" is shown amber as a heads-up but isn't alarming on its own). export const WAL_STATUS_META: Record = { reserved: { label: 'Reserved', variant: 'success', severity: 'normal', description: 'Postgres will keep the WAL for every change until this pipeline sends it.', tableDescription: "Healthy. Your database is keeping the WAL files this table's replication slot needs, and they are within the normal WAL size limit.", }, extended: { label: 'Extended', variant: 'warning', severity: 'normal', description: 'The pipeline is behind. Postgres is retaining more WAL than usual, but nothing is discarded yet.', tableDescription: "Healthy, but growing. This table's replication slot is holding on to more WAL than usual, but your database is still keeping everything it needs.", }, unreserved: { label: 'Unreserved', variant: 'warning', severity: 'warning', description: 'Postgres may discard WAL this pipeline has not sent yet.', tableDescription: "At risk. Your database is no longer reserving all WAL files this table's replication slot needs. If the pipeline does not catch up soon, those files may be removed.", }, lost: { label: 'Lost', variant: 'destructive', severity: 'critical', description: 'Postgres already discarded WAL this pipeline needed. Replication cannot continue from here.', tableDescription: "Broken. Some WAL files this table's replication slot needs have already been removed. The pipeline can no longer continue from this slot. You can recreate a new pipeline, or set the invalidation behavior to recreate and restart the pipeline.", }, unknown: { label: 'Unknown', variant: 'default', severity: 'normal', description: 'Postgres did not report a recognized status for this pipeline’s slot.', tableDescription: "Unknown. Your database reported an unknown state for this table's replication slot.", }, } // Postgres reports no WAL status (restart_lsn is null) as "unknown" too, so fall back to it. export const getWalStatusMeta = (status?: SlotWalStatus | null): WalStatusMeta => WAL_STATUS_META[status ?? 'unknown'] // Legend entries from healthiest to most severe, ending with the unknown/unavailable case. export const WAL_STATUS_LEGEND: WalStatusMeta[] = [ WAL_STATUS_META.reserved, WAL_STATUS_META.extended, WAL_STATUS_META.unreserved, WAL_STATUS_META.lost, WAL_STATUS_META.unknown, ] export const getWalStatusSeverity = (status?: SlotWalStatus | null): LagSeverity => getWalStatusMeta(status).severity // Slot-loss risk from how much of the slot's WAL budget has been consumed, rather than fixed byte // thresholds: max_slot_wal_keep_size ≈ retained WAL (restart_lsn_bytes) + remaining headroom // (safe_wal_size_bytes), so the consumed fraction is how close the slot is to the "lost" state. // A null/absent safe_wal_size_bytes now means unlimited retention, so it carries no budget risk. export const SLOT_LOSS_WARNING_RATIO = 0.75 export const SLOT_LOSS_CRITICAL_RATIO = 0.9 export const getSlotBudgetSeverity = ( retainedBytes?: number, safeWalSizeBytes?: number | null ): LagSeverity => { if ( typeof retainedBytes !== 'number' || typeof safeWalSizeBytes !== 'number' || !Number.isFinite(retainedBytes) || !Number.isFinite(safeWalSizeBytes) ) { return 'normal' } const total = retainedBytes + safeWalSizeBytes // Nothing retained and no headroom left: nothing to flag (also avoids a 0/0 division). When the // headroom is 0 but WAL is still retained, the ratio is 1 and the slot is correctly critical. if (total <= 0) return 'normal' const consumedRatio = retainedBytes / total if (consumedRatio >= SLOT_LOSS_CRITICAL_RATIO) return 'critical' if (consumedRatio >= SLOT_LOSS_WARNING_RATIO) return 'warning' return 'normal' } const SEVERITY_RANK: Record = { normal: 0, warning: 1, critical: 2 } const maxSeverity = (a: LagSeverity, b: LagSeverity): LagSeverity => SEVERITY_RANK[a] >= SEVERITY_RANK[b] ? a : b // Overall slot health = the worse of the reported WAL status and how close the WAL budget is to // running out. Used to color/flag the lag value in the destinations list. export const getSlotHealthSeverity = (slot?: { restart_lsn_bytes?: number safe_wal_size_bytes?: number | null wal_status?: SlotWalStatus }): LagSeverity => { if (!slot) return 'normal' return maxSeverity( getWalStatusSeverity(slot.wal_status), getSlotBudgetSeverity(slot.restart_lsn_bytes, slot.safe_wal_size_bytes) ) } /** * A table's own replication slot, as a short list of phrases for one table cell. Skips anything * that carries no signal, such as a zero backlog or a reserved WAL status, so the line only ever * says what's worth reading. Connection is skipped on purpose: a copying table's slot is inactive * until the copy finishes, so flagging it would look like a fault. */ export const getTableSyncLagLabel = (metrics: SlotLagMetrics): string[] => { const parts: string[] = [] const pendingBytes = metrics.confirmed_flush_lsn_bytes if (typeof pendingBytes === 'number' && pendingBytes > 0) { parts.push(`${formatBytes(pendingBytes, pendingBytes < 1024 ? 0 : 1)} waiting to sync`) } const safeWalSizeBytes = metrics.safe_wal_size_bytes if (safeWalSizeBytes === null) { parts.push('Unlimited WAL retention') } else if ( typeof safeWalSizeBytes === 'number' && Number.isFinite(safeWalSizeBytes) && safeWalSizeBytes >= 0 ) { const formattedBytes = formatBytes(safeWalSizeBytes, safeWalSizeBytes < 1024 ? 0 : 1) parts.push(`${formattedBytes} WAL retention remaining`) } if (metrics.wal_status === 'unreserved') parts.push('Some changes at risk') if (metrics.wal_status === 'lost') parts.push('Some changes lost') const replyLag = metrics.reply_time_lag if (typeof replyLag === 'number' && replyLag > 0) { parts.push(`Last check-in ${getFormattedLagValue('duration', replyLag).display}`) } return parts }