12b50f4868
CI / checks (pull_request) Successful in 27m2s
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.
136 lines
7.6 KiB
TypeScript
136 lines
7.6 KiB
TypeScript
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 : ''
|
|
}
|