feat(api): add EventBridge Pipes to the cloud explorer

Create pipes from SQS, Kinesis, DynamoDB streams or MSK to Lambda, SQS, SNS,
EventBridge buses or Step Functions, with an optional filter, enrichment and
input template; start and stop them, see the flow with links to each resource,
and send a test message to an SQS source. Sources and targets the runtime's
poller cannot deliver between are refused on create.

packages/frontend/src/components/DynamicResourceView.tsx

@@ -69,6 +69,7 @@ import { CognitoUsersPanel } from "@/components/CognitoUsersPanel";
 import { Route53RecordsPanel } from "@/components/Route53RecordsPanel";
 import { CloudFrontPanel } from "@/components/CloudFrontPanel";
 import { CertificatePanel } from "@/components/CertificatePanel";
+import { PipePanel } from "@/components/PipePanel";
 import { CloudWatchAlarmPanel } from "@/components/CloudWatchAlarmPanel";
 import { CloudWatchMetricsExplorer } from "@/components/CloudWatchMetricsExplorer";
 import { WorkflowExecutionDebugger } from "@/components/WorkflowExecutionDebugger";
@@ -991,6 +992,9 @@ export function DynamicResourceView({
       {service === "certificates" && cloud === "aws" && (
         <CertificatePanel resource={activeSelected} runtimeReachable={canUseRuntime} />
       )}
+      {service === "pipes" && cloud === "aws" && (
+        <PipePanel resource={activeSelected} runtimeReachable={canUseRuntime} />
+      )}
       {service === "cdn" && cloud === "aws" && (
         <CloudFrontPanel resource={activeSelected} runtimeReachable={canUseRuntime} />
       )}

packages/frontend/src/components/PipePanel.test.tsx

@@ -0,0 +1,81 @@
+import {QueryClient, QueryClientProvider} from '@tanstack/react-query'
+import {render, screen, within} from '@testing-library/react'
+import userEvent from '@testing-library/user-event'
+import {MemoryRouter} from 'react-router-dom'
+import {beforeEach, expect, test, vi} from 'vitest'
+import {PipePanel} from './PipePanel'
+import type {CloudResource} from '@/types/resource'
+
+const api = vi.hoisted(() => ({
+    getCloudResource: vi.fn(),
+    startCloudResource: vi.fn(),
+    stopCloudResource: vi.fn(),
+    sendSqsMessage: vi.fn(),
+}))
+
+vi.mock('@/api/cloudProxyClient', () => api)
+vi.mock('@/lib/accountStore', () => ({useAccountId: () => '000000000000'}))
+
+function pipe(status = 'RUNNING', metadata: Record<string, unknown> = {}): CloudResource {
+    return {
+        id: 'orders-pipe', name: 'orders-pipe', cloud: 'aws', service: 'pipes', type: 'pipe', region: 'us-east-1', createdAt: null, status,
+        metadata: {
+            desiredState: status, source: 'arn:aws:sqs:us-east-1:000000000000:pipe-src', sourceName: 'pipe-src (SQS queue)',
+            target: 'arn:aws:lambda:us-east-1:000000000000:function:worker', targetName: 'worker (Lambda function)',
+            filterPatterns: ['{"body":{"kind":["order"]}}'], batchSize: 5, inputTemplate: '{"id": <$.messageId>}',
+            roleArn: 'arn:aws:iam::000000000000:role/pipe-role',
+            ...metadata,
+        },
+    }
+}
+
+function renderPanel(resource: CloudResource) {
+    api.getCloudResource.mockResolvedValue(resource)
+    const client = new QueryClient({defaultOptions: {queries: {retry: false}}})
+    return render(<MemoryRouter><QueryClientProvider client={client}><PipePanel resource={resource} runtimeReachable/></QueryClientProvider></MemoryRouter>)
+}
+
+beforeEach(() => {
+    vi.clearAllMocks()
+    api.startCloudResource.mockResolvedValue(undefined)
+    api.stopCloudResource.mockResolvedValue(undefined)
+    api.sendSqsMessage.mockResolvedValue({messageId: 'm-1'})
+})
+
+test('shows the flow from source through the filter to the target, linked to each service', async () => {
+    renderPanel(pipe())
+
+    const flow = await screen.findByRole('list', {name: 'Event flow'})
+    expect(within(flow).getByRole('link', {name: 'pipe-src (SQS queue)'})).toHaveAttribute('href', '/cloud-explorer/aws/messaging')
+    expect(within(flow).getByRole('link', {name: 'worker (Lambda function)'})).toHaveAttribute('href', '/cloud-explorer/aws/serverless')
+    expect(within(flow).getByText('{"body":{"kind":["order"]}}')).toBeInTheDocument()
+    expect(within(flow).getByText('batches of 5')).toBeInTheDocument()
+    expect(within(flow).getByText('{"id": <$.messageId>}')).toBeInTheDocument()
+})
+
+test('stops a running pipe', async () => {
+    const user = userEvent.setup()
+    renderPanel(pipe())
+
+    await user.click(await screen.findByRole('button', {name: 'Stop'}))
+    expect(api.stopCloudResource).toHaveBeenCalledWith('aws', 'pipes', 'orders-pipe')
+    expect(await screen.findByText(/Pipe stopped/)).toBeInTheDocument()
+})
+
+test('sends a test message to an SQS source queue', async () => {
+    const user = userEvent.setup()
+    renderPanel(pipe())
+
+    const form = await screen.findByRole('form', {name: 'Send test message'})
+    await user.click(within(form).getByRole('button', {name: 'Send to pipe-src'}))
+    expect(api.sendSqsMessage).toHaveBeenCalledWith('aws', 'pipe-src', {messageBody: '{"kind": "order", "id": 1}'})
+    expect(await screen.findByText(/Sent message m-1/)).toBeInTheDocument()
+})
+
+test('offers no test message for a stream source, and cannot toggle a pipe mid-change', async () => {
+    renderPanel(pipe('STARTING', {source: 'arn:aws:kinesis:us-east-1:000000000000:stream/clicks', sourceName: 'clicks (Kinesis stream)', desiredState: 'RUNNING'}))
+
+    expect(await screen.findByRole('link', {name: 'clicks (Kinesis stream)'})).toBeInTheDocument()
+    expect(screen.queryByRole('form', {name: 'Send test message'})).not.toBeInTheDocument()
+    expect(screen.getByRole('button', {name: 'Start'})).toBeDisabled()
+})

packages/frontend/src/components/PipePanel.tsx

@@ -0,0 +1,135 @@
+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<string, string> = {
+    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 <section className="table-panel"><div className="empty compact"><h3>Select a pipe</h3><p>Choose a pipe to see where its events flow, start or stop it, or send it a test message.</p></div></section>
+    }
+
+    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 <section className="table-panel" style={{marginTop: 16}} aria-label={`Pipe ${pipe.name}`}>
+        <div className="input-row resource-table-bar">
+            <div><p className="eyebrow">Pipe</p><span className="muted">{state}{metadata.desiredState && metadata.desiredState !== state ? ` · going to ${String(metadata.desiredState)}` : ''}{metadata.description ? ` · ${String(metadata.description)}` : ''}</span></div>
+            <div className="cdn-actions">
+                <button className="button" type="button" disabled={disabled || !['RUNNING', 'STOPPED'].includes(state)} onClick={() => { setNotice(''); stateMutation.mutate(!running) }}>{running ? 'Stop' : 'Start'}</button>
+                <button className="button" type="button" disabled={!runtimeReachable || pipeQuery.isFetching} onClick={() => { void refresh() }}><RefreshCw size={14}/>{pipeQuery.isFetching ? 'Loading' : 'Refresh'}</button>
+            </div>
+        </div>
+        {typeof metadata.stateReason === 'string' && metadata.stateReason && <p className="error-text compact-text pipe-reason">{metadata.stateReason}</p>}
+
+        <ol className="pipe-flow" aria-label="Event flow">
+            <PipeStep label="Source" arn={source} name={metadata.sourceName} detail={sourceDetail(metadata)}/>
+            {filters.length > 0 && <li className="pipe-step"><span className="eyebrow">Filter</span>{filters.map((pattern) => <code key={pattern} className="pipe-code">{pattern}</code>)}<ArrowRight size={14} className="pipe-arrow" aria-hidden/></li>}
+            {text(metadata.enrichment) && <PipeStep label="Enrichment" arn={text(metadata.enrichment)} name={null}/>}
+            <PipeStep label="Target" arn={text(metadata.target)} name={metadata.targetName} detail={text(metadata.inputTemplate) ? <>input template <code className="pipe-code">{text(metadata.inputTemplate)}</code></> : undefined}/>
+        </ol>
+        {text(metadata.roleArn) && <p className="muted compact-text pipe-role">Runs as <code>{text(metadata.roleArn)}</code></p>}
+
+        {sourceQueue && <>
+            <div className="input-row resource-table-bar"><div><p className="eyebrow">Try it</p><span className="muted">Send a message to the source queue{filters.length > 0 ? '. Only messages matching the filter reach the target' : ''}.</span></div></div>
+            <form className="dynamic-form" aria-label="Send test message" onSubmit={(event) => { event.preventDefault(); setNotice(''); sendMutation.mutate(sourceQueue) }}>
+                <label className="dynamic-field dynamic-field--span"><span>Message body</span>
+                    <textarea className="textarea code-textarea small" rows={3} value={message} onChange={(event) => setMessage(event.target.value)} disabled={disabled} required/>
+                </label>
+                <div><button className="button primary" type="submit" disabled={disabled || !message.trim()}>{sendMutation.isPending ? 'Sending' : `Send to ${sourceQueue}`}</button></div>
+            </form>
+        </>}
+
+        {notice && <p className="cognito-outcome cognito-outcome--ok" role="status">{notice}</p>}
+        {error instanceof Error && <p className="error-text compact-text" role="alert">{error.message}</p>}
+    </section>
+}
+
+function PipeStep({label, arn, name, detail}: {label: string; arn: string; name: unknown; detail?: ReactNode}) {
+    const service = CONSOLE_SERVICE[arn.split(':')[2] ?? '']
+    const title = typeof name === 'string' && name ? name : arn
+    return <li className="pipe-step">
+        <span className="eyebrow">{label}</span>
+        {service ? <Link to={`/cloud-explorer/aws/${service}`} className="mono">{title}</Link> : <span className="mono">{title}</span>}
+        <small className="muted mono pipe-arn">{arn}</small>
+        {detail && <small className="muted">{detail}</small>}
+        <ArrowRight size={14} className="pipe-arrow" aria-hidden/>
+    </li>
+}
+
+function sourceDetail(metadata: Record<string, unknown>): ReactNode {
+    const parts = [
+        metadata.batchSize ? `batches of ${String(metadata.batchSize)}` : '',
+        metadata.startingPosition ? `from ${String(metadata.startingPosition)}` : '',
+        metadata.topicName ? `topic ${String(metadata.topicName)}` : '',
+    ].filter(Boolean)
+    return parts.length > 0 ? parts.join(' · ') : undefined
+}
+
+function text(value: unknown): string {
+    return typeof value === 'string' ? value : ''
+}

packages/frontend/src/components/serviceIcons.tsx

@@ -89,6 +89,7 @@ const SERVICE_ICONS: Record<string, LucideIcon> = {
     iac: Layers,
     messaging: MessageSquare,
     events: Radio,
+    pipes: Workflow,
     email: MessageSquare,
     queue: MessageSquare,
     logs: ScrollText,
@@ -124,6 +125,7 @@ const AWS_SERVICE_ICON_ASSETS: Record<string, string> = {
     ecr: ecrIcon,
     email: sesIcon,
     events: eventBridgeIcon,
+    pipes: eventBridgeIcon,
     iac: cloudFormationIcon,
     iam: iamIcon,
     kms: kmsIcon,

packages/frontend/src/index.css

@@ -6616,3 +6616,12 @@ tr[aria-current="true"] td { background: var(--hover-md); }
 .certificate-pem { display: grid; gap: 12px; padding: 0 16px 16px; }
 .certificate-pem-block { display: grid; gap: 6px; min-width: 0; }
 .certificate-pem-head { display: flex; flex-wrap: wrap; align-items: center; gap: 8px; }
+
+/* EventBridge Pipes: source, filter, enrichment and target as one flow. */
+.pipe-flow { display: flex; flex-wrap: wrap; gap: 8px 28px; margin: 0; padding: 4px 16px 12px; list-style: none; }
+.pipe-step { position: relative; display: grid; gap: 4px; align-content: start; min-width: 0; max-width: 100%; padding: 10px 12px; border: 1px solid var(--border); border-radius: 4px; background: var(--raised); }
+.pipe-step .pipe-arrow { position: absolute; top: 50%; right: -22px; transform: translateY(-50%); color: var(--text-3); }
+.pipe-step:last-child .pipe-arrow { display: none; }
+.pipe-arn, .pipe-code { overflow-wrap: anywhere; }
+.pipe-code { font-family: var(--mono); font-size: 11px; white-space: pre-wrap; }
+.pipe-role, .pipe-reason { padding: 0 16px 12px; margin: 0; overflow-wrap: anywhere; }