feat(replication): Add ability to copy pipeline id and filter logs by pipeline id (#38407)

apps/studio/components/interfaces/Database/Replication/DestinationPanel.tsx

@@ -74,6 +74,7 @@ interface DestinationPanelProps {
     destinationId: number
     pipelineId?: number
     enabled: boolean
+    statusName?: string
   }
 }
 
@@ -178,7 +179,8 @@ export const DestinationPanel = ({
           sourceId,
         })
         // Set request status only right before starting, then fire and close
-        const snapshot = existingDestination.enabled ? 'started' : 'stopped'
+        const snapshot =
+          existingDestination.statusName ?? (existingDestination.enabled ? 'started' : 'stopped')
         if (existingDestination.enabled) {
           setRequestStatus(
             existingDestination.pipelineId,
@@ -244,6 +246,13 @@ export const DestinationPanel = ({
     }
   }, [destinationData, pipelineData, editMode, defaultValues, form])
 
+  // Ensure the form always reflects the freshest data whenever the panel opens
+  useEffect(() => {
+    if (visible) {
+      form.reset(defaultValues)
+    }
+  }, [visible, defaultValues, form])
+
   return sourceId ? (
     <>
       <Sheet open={visible} onOpenChange={onClose}>

apps/studio/components/interfaces/Database/Replication/DestinationRow.tsx

@@ -125,7 +125,20 @@ export const DestinationRow = ({
       )}
       {isPipelineSuccess && (
         <Table.tr>
-          <Table.td>{isPipelineLoading ? <ShimmeringLoader /> : destinationName}</Table.td>
+          <Table.td>
+            {isPipelineLoading ? (
+              <ShimmeringLoader />
+            ) : pipeline?.id ? (
+              <Tooltip>
+                <TooltipTrigger>
+                  <span className="cursor-default">{destinationName}</span>
+                </TooltipTrigger>
+                <TooltipContent side="bottom">Pipeline ID: {pipeline.id}</TooltipContent>
+              </Tooltip>
+            ) : (
+              destinationName
+            )}
+          </Table.td>
           <Table.td>{isPipelineLoading ? <ShimmeringLoader /> : type}</Table.td>
           <Table.td>
             {isPipelineLoading || !pipeline ? (
@@ -138,6 +151,7 @@ export const DestinationRow = ({
                 isError={isPipelineStatusError}
                 isSuccess={isPipelineStatusSuccess}
                 requestStatus={requestStatus}
+                pipelineId={pipeline?.id}
               />
             )}
           </Table.td>
@@ -195,7 +209,9 @@ export const DestinationRow = ({
           sourceId,
           destinationId: destinationId,
           pipelineId: pipeline?.id,
-          enabled: statusName === PipelineStatusName.STARTED,
+          enabled:
+            statusName === PipelineStatusName.STARTED || statusName === PipelineStatusName.FAILED,
+          statusName,
         }}
       />
     </>

apps/studio/components/interfaces/Database/Replication/Pipeline.utils.ts

@@ -1,5 +1,6 @@
 import { ReplicationPipelineStatusData } from 'data/replication/pipeline-status-query'
 import { PipelineStatusRequestStatus } from 'state/replication-pipeline-request-status'
+import { PipelineStatusName } from './PipelineStatus'
 
 export const PIPELINE_ERROR_MESSAGES = {
   RETRIEVE_PIPELINE: 'Failed to retrieve pipeline information',
@@ -24,7 +25,7 @@ export const getStatusName = (
 
 export const PIPELINE_ENABLE_ALLOWED_FROM = ['stopped'] as const
 export const PIPELINE_DISABLE_ALLOWED_FROM = ['started', 'failed'] as const
-export const PIPELINE_ACTIONABLE_STATES = ['failed', 'started', 'stopped'] as const
+export const PIPELINE_ACTIONABLE_STATES = ['failed', 'started', 'stopped'] as PipelineStatusName[]
 
 const PIPELINE_STATE_MESSAGES = {
   enabling: {

apps/studio/components/interfaces/Database/Replication/PipelineStatus.tsx

@@ -1,7 +1,8 @@
+import { AlertTriangle, Loader2 } from 'lucide-react'
+
 import { useParams } from 'common'
 import { InlineLink } from 'components/ui/InlineLink'
 import { ReplicationPipelineStatusData } from 'data/replication/pipeline-status-query'
-import { AlertTriangle, Loader2 } from 'lucide-react'
 import { PipelineStatusRequestStatus } from 'state/replication-pipeline-request-status'
 import { ResponseError } from 'types'
 import { cn, Tooltip, TooltipContent, TooltipTrigger, WarningIcon } from 'ui'
@@ -23,6 +24,7 @@ interface PipelineStatusProps {
   isError: boolean
   isSuccess: boolean
   requestStatus?: PipelineStatusRequestStatus
+  pipelineId?: number
 }
 
 export const PipelineStatus = ({
@@ -32,6 +34,7 @@ export const PipelineStatus = ({
   isError,
   isSuccess,
   requestStatus,
+  pipelineId,
 }: PipelineStatusProps) => {
   const { ref } = useParams()
 
@@ -129,6 +132,12 @@ export const PipelineStatus = ({
 
   const statusConfig = getStatusConfig()
 
+  const pipelineLogsUrl = pipelineId
+    ? `/project/${ref}/logs/etl-replication-logs?f=${encodeURIComponent(
+        JSON.stringify({ pipeline_id: pipelineId })
+      )}`
+    : `/project/${ref}/logs/etl-replication-logs`
+
   return (
     <>
       {isLoading && <ShimmeringLoader />}
@@ -155,9 +164,7 @@ export const PipelineStatus = ({
             {['unknown', 'failed'].includes(pipelineStatus?.name ?? '') && (
               <>
                 {' '}
-                Check the{' '}
-                <InlineLink href={`/project/${ref}/logs/etl-replication-logs`}>logs</InlineLink> for
-                more information.
+                Check the <InlineLink href={pipelineLogsUrl}>logs</InlineLink> for more information.
               </>
             )}
           </TooltipContent>

apps/studio/components/interfaces/Database/Replication/ReplicationPipelineStatus.tsx

@@ -1,4 +1,14 @@
-import { Activity, ChevronLeft, ExternalLink, Search, X } from 'lucide-react'
+import {
+  Activity,
+  Ban,
+  ChevronLeft,
+  ExternalLink,
+  Pause,
+  Play,
+  RotateCcw,
+  Search,
+  X,
+} from 'lucide-react'
 import Link from 'next/link'
 import { useEffect, useState } from 'react'
 import { toast } from 'sonner'
@@ -22,12 +32,10 @@ import { Input } from 'ui-patterns/DataInputs/Input'
 import { ErroredTableDetails } from './ErroredTableDetails'
 import {
   PIPELINE_ACTIONABLE_STATES,
-  PIPELINE_DISABLE_ALLOWED_FROM,
-  PIPELINE_ENABLE_ALLOWED_FROM,
   PIPELINE_ERROR_MESSAGES,
   getStatusName,
 } from './Pipeline.utils'
-import { PipelineStatus } from './PipelineStatus'
+import { PipelineStatus, PipelineStatusName } from './PipelineStatus'
 import { STATUS_REFRESH_FREQUENCY_MS } from './Replication.constants'
 import { TableState } from './ReplicationPipelineStatus.types'
 import { getDisabledStateConfig, getStatusConfig } from './ReplicationPipelineStatus.utils'
@@ -104,21 +112,44 @@ export const ReplicationPipelineStatus = () => {
     requestStatus === PipelineStatusRequestStatus.RestartRequested
   const showDisabledState = !isPipelineRunning || isEnablingDisabling
 
-  const onTogglePipeline = async () => {
-    if (!projectRef) {
-      return console.error('Project ref is required')
-    }
-    if (!pipeline) {
-      return toast.error(PIPELINE_ERROR_MESSAGES.NO_PIPELINE_FOUND)
-    }
+  const logsUrl = `/project/${projectRef}/logs/etl-replication-logs${
+    pipelineId ? `?f=${encodeURIComponent(JSON.stringify({ pipeline_id: pipelineId }))}` : ''
+  }`
+
+  const label =
+    statusName === 'stopped'
+      ? 'Start'
+      : statusName === 'started'
+        ? 'Stop'
+        : statusName === 'failed'
+          ? 'Restart'
+          : 'Action unavailable'
+
+  const icon =
+    statusName === 'stopped' ? (
+      <Play />
+    ) : statusName === 'started' ? (
+      <Pause />
+    ) : statusName === 'failed' ? (
+      <RotateCcw />
+    ) : (
+      <Ban />
+    )
+
+  const onPrimaryAction = async () => {
+    if (!projectRef) return console.error('Project ref is required')
+    if (!pipeline) return toast.error(PIPELINE_ERROR_MESSAGES.NO_PIPELINE_FOUND)
 
     try {
-      if (PIPELINE_ENABLE_ALLOWED_FROM.includes(statusName as any)) {
+      if (statusName === 'stopped') {
         setRequestStatus(pipeline.id, PipelineStatusRequestStatus.StartRequested, statusName)
         await startPipeline({ projectRef, pipelineId: pipeline.id })
-      } else if (PIPELINE_DISABLE_ALLOWED_FROM.includes(statusName as any)) {
+      } else if (statusName === 'started') {
         setRequestStatus(pipeline.id, PipelineStatusRequestStatus.StopRequested, statusName)
         await stopPipeline({ projectRef, pipelineId: pipeline.id })
+      } else if (statusName === 'failed') {
+        setRequestStatus(pipeline.id, PipelineStatusRequestStatus.RestartRequested, statusName)
+        await startPipeline({ projectRef, pipelineId: pipeline.id })
       }
     } catch (error) {
       toast.error(PIPELINE_ERROR_MESSAGES.ENABLE_DESTINATION)
@@ -136,57 +167,57 @@ export const ReplicationPipelineStatus = () => {
           <Button asChild type="outline" icon={<ChevronLeft />} style={{ padding: '5px' }}>
             <Link href={`/project/${projectRef}/database/replication`} />
           </Button>
-          <div>
-            <div className="flex items-center gap-x-3">
-              <h3 className="text-xl font-semibold">{destinationName || 'Pipeline'}</h3>
-              <PipelineStatus
-                pipelineStatus={pipelineStatusData?.status}
-                error={pipelineStatusError}
-                isLoading={isPipelineStatusLoading}
-                isError={isPipelineStatusError}
-                isSuccess={isPipelineStatusSuccess}
-                requestStatus={requestStatus}
-              />
-            </div>
+          <div className="flex items-center gap-x-3">
+            <h3 className="text-xl font-semibold">{destinationName || 'Pipeline'}</h3>
+            <PipelineStatus
+              pipelineStatus={pipelineStatusData?.status}
+              error={pipelineStatusError}
+              isLoading={isPipelineStatusLoading}
+              isError={isPipelineStatusError}
+              isSuccess={isPipelineStatusSuccess}
+              requestStatus={requestStatus}
+              pipelineId={pipelineId}
+            />
           </div>
         </div>
         <div className="flex items-center gap-x-2">
-          <div className="relative">
-            <Search
-              className="absolute left-2 top-1/2 transform -translate-y-1/2 text-foreground-lighter"
-              size={14}
-            />
-            <Input
-              className="pl-7 h-[26px] text-xs"
-              placeholder="Search for tables"
-              value={filterString}
-              disabled={isPipelineError}
-              onChange={(e) => setFilterString(e.target.value)}
-              actions={
-                filterString.length > 0
-                  ? [
-                      <X
-                        key="close"
-                        className="mx-2 cursor-pointer text-foreground"
-                        size={14}
-                        strokeWidth={2}
-                        onClick={() => setFilterString('')}
-                      />,
-                    ]
-                  : undefined
-              }
-            />
-          </div>
+          <Input
+            icon={<Search size={12} />}
+            className="pl-7 h-[26px] text-xs"
+            placeholder="Search for tables"
+            value={filterString}
+            disabled={isPipelineError}
+            onChange={(e) => setFilterString(e.target.value)}
+            actions={
+              filterString.length > 0
+                ? [
+                    <X
+                      key="close"
+                      className="mx-2 cursor-pointer text-foreground"
+                      size={14}
+                      strokeWidth={2}
+                      onClick={() => setFilterString('')}
+                    />,
+                  ]
+                : undefined
+            }
+          />
+
           <Button asChild type="default">
-            <Link href={`/project/${projectRef}/logs/etl-replication-logs`}>View logs</Link>
+            <Link href={logsUrl}>View logs</Link>
           </Button>
+
           <Button
             type={statusName === 'stopped' ? 'primary' : 'default'}
-            onClick={() => onTogglePipeline()}
+            onClick={onPrimaryAction}
             loading={isPipelineError || isStartingPipeline || isStoppingPipeline}
-            disabled={!PIPELINE_ACTIONABLE_STATES.includes((statusName ?? '') as any)}
+            disabled={
+              isEnablingDisabling ||
+              !PIPELINE_ACTIONABLE_STATES.includes(statusName as PipelineStatusName)
+            }
+            icon={icon}
           >
-            {statusName === 'stopped' ? 'Start' : 'Stop'} pipeline
+            {label}
           </Button>
         </div>
       </div>
@@ -231,6 +262,7 @@ export const ReplicationPipelineStatus = () => {
           )}
 
           <div className="w-full overflow-hidden overflow-x-auto">
+            {/* [Joshen] Should update to use new Table components next time */}
             <Table
               head={[
                 <Table.th key="table">Table</Table.th>,

apps/studio/components/interfaces/Database/Replication/RowMenu.tsx

@@ -1,8 +1,9 @@
-import { Edit, MoreVertical, Pause, Play, Trash } from 'lucide-react'
+import { Edit, MoreVertical, Pause, Play, RotateCcw, Trash } from 'lucide-react'
 import { toast } from 'sonner'
 
 import { useParams } from 'common'
 import AlertError from 'components/ui/AlertError'
+import { ReplicationPipelineStatusData } from 'data/replication/pipeline-status-query'
 import { Pipeline } from 'data/replication/pipelines-query'
 import { useStartPipelineMutation } from 'data/replication/start-pipeline-mutation'
 import { useStopPipelineMutation } from 'data/replication/stop-pipeline-mutation'
@@ -21,6 +22,7 @@ import {
 } from 'ui'
 import ShimmeringLoader from 'ui-patterns/ShimmeringLoader'
 import {
+  PIPELINE_ACTIONABLE_STATES,
   PIPELINE_DISABLE_ALLOWED_FROM,
   PIPELINE_ENABLE_ALLOWED_FROM,
   PIPELINE_ERROR_MESSAGES,
@@ -30,7 +32,7 @@ import { PipelineStatusName } from './PipelineStatus'
 
 interface RowMenuProps {
   pipeline: Pipeline | undefined
-  pipelineStatus: any
+  pipelineStatus?: ReplicationPipelineStatusData['status']
   error: ResponseError | null
   isLoading: boolean
   isError: boolean
@@ -48,21 +50,42 @@ export const RowMenu = ({
   onDeleteClick,
 }: RowMenuProps) => {
   const { ref: projectRef } = useParams()
-
   const statusName = getStatusName(pipelineStatus)
-  const pipelineEnabled = statusName !== PipelineStatusName.STOPPED
 
   const { mutateAsync: startPipeline } = useStartPipelineMutation()
   const { mutateAsync: stopPipeline } = useStopPipelineMutation()
-  const { setRequestStatus: setGlobalRequestStatus } = usePipelineRequestStatus()
+  const { getRequestStatus, setRequestStatus: setGlobalRequestStatus } = usePipelineRequestStatus()
+  const requestStatus = pipeline?.id
+    ? getRequestStatus(pipeline.id)
+    : PipelineStatusRequestStatus.None
+
+  const hasPipelineAction =
+    requestStatus === PipelineStatusRequestStatus.None &&
+    [PipelineStatusName.STOPPED, PipelineStatusName.STARTED, PipelineStatusName.FAILED].includes(
+      statusName as PipelineStatusName
+    )
+
+  const pipelineActionIcon =
+    statusName === PipelineStatusName.STOPPED ? (
+      <Play size={14} />
+    ) : statusName === PipelineStatusName.STARTED ? (
+      <Pause size={14} />
+    ) : statusName === PipelineStatusName.FAILED ? (
+      <RotateCcw size={14} />
+    ) : null
+
+  const pipelineActionLabel =
+    statusName === PipelineStatusName.STOPPED
+      ? 'Start pipeline'
+      : statusName === PipelineStatusName.STARTED
+        ? 'Stop pipeline'
+        : statusName === PipelineStatusName.FAILED
+          ? 'Restart pipeline'
+          : null
 
   const onEnablePipeline = async () => {
-    if (!projectRef) {
-      return console.error('Project ref is required')
-    }
-    if (!pipeline) {
-      return toast.error(PIPELINE_ERROR_MESSAGES.NO_PIPELINE_FOUND)
-    }
+    if (!projectRef) return console.error('Project ref is required')
+    if (!pipeline) return toast.error(PIPELINE_ERROR_MESSAGES.NO_PIPELINE_FOUND)
 
     try {
       // Only show 'enabling' when transitioning from allowed states
@@ -77,14 +100,8 @@ export const RowMenu = ({
   }
 
   const onDisablePipeline = async () => {
-    if (!projectRef) {
-      console.error('Project ref is required')
-      return
-    }
-    if (!pipeline) {
-      toast.error(PIPELINE_ERROR_MESSAGES.NO_PIPELINE_FOUND)
-      return
-    }
+    if (!projectRef) return console.error('Project ref is required')
+    if (!pipeline) return toast.error(PIPELINE_ERROR_MESSAGES.NO_PIPELINE_FOUND)
 
     try {
       // Only show 'disabling' when transitioning from allowed states
@@ -98,9 +115,23 @@ export const RowMenu = ({
     }
   }
 
+  const onRestartPipeline = async () => {
+    if (!projectRef) return console.error('Project ref is required')
+    if (!pipeline) return toast.error(PIPELINE_ERROR_MESSAGES.NO_PIPELINE_FOUND)
+
+    try {
+      setGlobalRequestStatus(pipeline.id, PipelineStatusRequestStatus.RestartRequested, statusName)
+      await startPipeline({ projectRef, pipelineId: pipeline.id })
+    } catch (error) {
+      setGlobalRequestStatus(pipeline.id, PipelineStatusRequestStatus.None)
+      toast.error(PIPELINE_ERROR_MESSAGES.ENABLE_DESTINATION)
+    }
+  }
+
   return (
     <div className="flex justify-end items-center space-x-2">
       {isLoading && <ShimmeringLoader />}
+
       {isError && (
         <AlertError error={error} subject={PIPELINE_ERROR_MESSAGES.RETRIEVE_PIPELINE_STATUS} />
       )}
@@ -109,19 +140,30 @@ export const RowMenu = ({
         <DropdownMenuTrigger asChild>
           <Button type="default" className="px-1.5" icon={<MoreVertical />} />
         </DropdownMenuTrigger>
+
         <DropdownMenuContent side="bottom" align="end" className="w-52">
-          {pipelineEnabled ? (
-            <DropdownMenuItem className="space-x-2" onClick={onDisablePipeline}>
-              <Pause size={14} />
-              <p>Stop pipeline</p>
-            </DropdownMenuItem>
-          ) : (
-            <DropdownMenuItem className="space-x-2" onClick={onEnablePipeline}>
-              <Play size={14} />
-              <p>Start pipeline</p>
-            </DropdownMenuItem>
+          {hasPipelineAction && (
+            <>
+              <DropdownMenuItem
+                className="space-x-2"
+                disabled={!PIPELINE_ACTIONABLE_STATES.includes((statusName ?? '') as any)}
+                onClick={() => {
+                  if (statusName === PipelineStatusName.STOPPED) {
+                    onEnablePipeline()
+                  } else if (statusName === PipelineStatusName.STARTED) {
+                    onDisablePipeline()
+                  } else if (statusName === PipelineStatusName.FAILED) {
+                    onRestartPipeline()
+                  }
+                }}
+              >
+                {pipelineActionIcon}
+                <p>{pipelineActionLabel}</p>
+              </DropdownMenuItem>
+              <DropdownMenuSeparator />
+            </>
           )}
-          <DropdownMenuSeparator />
+
           <DropdownMenuItem className="space-x-2" onClick={onEditClick}>
             <Edit size={14} />
             <p>Edit destination</p>

apps/studio/components/interfaces/Settings/Logs/Logs.constants.ts

@@ -357,6 +357,10 @@ export const SQL_FILTER_TEMPLATES: any = {
   pg_cron_logs: {
     ..._SQL_FILTER_COMMON,
   },
+  etl_replication_logs: {
+    ..._SQL_FILTER_COMMON,
+    pipeline_id: (value: string | number) => `pipeline_id = ${value}`,
+  },
 }
 
 export enum LogsTableName {

apps/studio/data/replication/update-destination-pipeline-mutation.ts

@@ -97,11 +97,15 @@ export const useUpdateDestinationPipelineMutation = ({
     (vars) => updateDestinationPipeline(vars),
     {
       async onSuccess(data, variables, context) {
-        const { projectRef } = variables
+        const { projectRef, destinationId, pipelineId } = variables
 
         await Promise.all([
+          // Invalidate lists
           queryClient.invalidateQueries(replicationKeys.destinations(projectRef)),
           queryClient.invalidateQueries(replicationKeys.pipelines(projectRef)),
+          // Invalidate item-level caches used by the editor panel
+          queryClient.invalidateQueries(replicationKeys.destinationById(projectRef, destinationId)),
+          queryClient.invalidateQueries(replicationKeys.pipelineById(projectRef, pipelineId)),
         ])
 
         await onSuccess?.(data, variables, context)