import {ArrowRight, RefreshCw} from 'lucide-react' import {useMutation, useQuery, useQueryClient} from '@tanstack/react-query' import {type ReactNode, useEffect, useState} from 'react' import {Link} from 'react-router-dom' import {getCloudResource, sendSqsMessage, startCloudResource, stopCloudResource} from '@/api/cloudProxyClient' import {cloudQueryKeys} from '@/api/queries/cloudQueries' import {useAccountId} from '@/lib/accountStore' import type {CloudResource} from '@/types/resource' type Props = { resource?: CloudResource runtimeReachable: boolean } /** Where each ARN's service lives in TheCloud, so a step links to the resource behind it. */ const CONSOLE_SERVICE: Record = { sqs: 'messaging', lambda: 'serverless', sns: 'notifications', events: 'events', states: 'workflows', kinesis: 'kinesis', dynamodb: 'nosql', } const SAMPLE_MESSAGE = '{"kind": "order", "id": 1}' export function PipePanel({resource, runtimeReachable}: Props) { const accountId = useAccountId() const queryClient = useQueryClient() const [message, setMessage] = useState(SAMPLE_MESSAGE) const [notice, setNotice] = useState('') const pipeKey = ['pipe', accountId, resource?.id] // The table row is a ListPipes entry; the filter, template and role come from DescribePipe. const pipeQuery = useQuery({ queryKey: pipeKey, queryFn: ({signal}) => getCloudResource('aws', 'pipes', resource?.id ?? '', signal), enabled: Boolean(resource) && runtimeReachable, }) useEffect(() => { setMessage(SAMPLE_MESSAGE) setNotice('') }, [resource?.id]) const refresh = () => Promise.all([ queryClient.invalidateQueries({queryKey: pipeKey}), queryClient.invalidateQueries({queryKey: cloudQueryKeys.resources('aws', 'pipes')}), ]) const stateMutation = useMutation({ mutationFn: (running: boolean) => running ? startCloudResource('aws', 'pipes', resource?.id ?? '') : stopCloudResource('aws', 'pipes', resource?.id ?? ''), onSuccess: (_result, running) => { setNotice(running ? 'Pipe started. It reads the source again.' : 'Pipe stopped. Records wait in the source until it starts again.') return refresh() }, }) const sendMutation = useMutation({ mutationFn: (queue: string) => sendSqsMessage('aws', queue, {messageBody: message}), onSuccess: (sent) => setNotice(`Sent message ${sent.messageId} to the source queue. The pipe picks it up within a few seconds.`), }) if (!resource) { return

Select a pipe

Choose a pipe to see where its events flow, start or stop it, or send it a test message.

} const pipe = pipeQuery.data ?? resource const metadata = pipe.metadata const state = pipe.status ?? 'UNKNOWN' const running = state === 'RUNNING' const filters = Array.isArray(metadata.filterPatterns) ? metadata.filterPatterns as string[] : [] const source = text(metadata.source) const sourceQueue = source.split(':')[2] === 'sqs' ? source.split(':')[5] : '' const pending = stateMutation.isPending || sendMutation.isPending const disabled = !runtimeReachable || pending const error = pipeQuery.error ?? stateMutation.error ?? sendMutation.error return

Pipe

{state}{metadata.desiredState && metadata.desiredState !== state ? ` · going to ${String(metadata.desiredState)}` : ''}{metadata.description ? ` · ${String(metadata.description)}` : ''}
{typeof metadata.stateReason === 'string' && metadata.stateReason &&

{metadata.stateReason}

}
    {filters.length > 0 &&
  1. Filter{filters.map((pattern) => {pattern})}
  2. } {text(metadata.enrichment) && } input template {text(metadata.inputTemplate)} : undefined}/>
{text(metadata.roleArn) &&

Runs as {text(metadata.roleArn)}

} {sourceQueue && <>

Try it

Send a message to the source queue{filters.length > 0 ? '. Only messages matching the filter reach the target' : ''}.
{ event.preventDefault(); setNotice(''); sendMutation.mutate(sourceQueue) }}>